cgaebel / pipe Goto Github PK

A simple thread-safe FIFO in C.

C 97.55% Shell 0.20% Makefile 2.25%

pipe's Introduction

pipe.c
========

C has historically had minimal support for multithreading, and even less
for concurrency. pthreads was the first step to bringing threading
constructs to C, and it has served us well. However, threads and mutexes
alone aren't the only concurrency paradigm available to us. Very few
paradigms map nicely onto heavyweight threads and locking. For example,
the Actor model has no explicit locking at all.

Let's imagine a classic problem. We want to write a simple download
manager. First, we need to get the data from a high-latency internet
connection, then dump that data to a high-latency hard disk. In
classic C, waiting for one resource to finish transferring data is time
you can't spend on the other one! This is unacceptable since you're
dealing with two very slow connections. When the disk starts spinning
up, you don't want your download to slow down!

To solve this problem, we introduce a new data structure called a pipe.
We create two threads: one for receiving the data from the network,
and one for dumping the data to disk.

Whenever data is received, it will be pushed into the pipe instead of
directly to the disk. In the disk thread, it will read from the pipe,
dumping any data onto the disk *without blocking the networking thread*.
Since the two processes are decoupled by a fast in-memory pipe (which
is implemented with simple memcpy's), when one thread lags behind, the
other one is free to continue its operation. If the producer is lagging,
the consumer will block until there is data. If the consumer is lagging,
requests will just pile up in the queue to be handled "eventually". If
memory usage becomes a problem, we can limit the size of the pipe to
block the producer when the pipe gets too full.

Although this sample seems a bit trivial, the pipe is a lot more powerful
than it lets on. No matter how many producers or consumers you have, pipe
will behave predictably (although not necessarily fairly). Therefore, you
can use pipe for serialization, pipelining tasks, parallel processing,
futures, multiplexing, and so much more. It does this extremely quickly,
to the point where anything except trivial code will be the bottleneck,
not the pipe. It can do all this and more in under 1000 lines of C!

My true hope is that this library will spur development of concurrent
systems in all languages, and the creation of new ideas in the field.
Since C is so pervasive, I hope to see these concepts leaking into other
languages and libraries, hopefully being expanded upon and becoming a
building block of the next generation of concurrent programming.

Integration
------------

1. Copy pipe.c and pipe.h into your project's directory.
2. Get your compiler to build it (pipe.c needs -std=c99)
3. Use it!

Concurrency Specifications
---------------------------

For more details on this list, see:
http://www.1024cores.net/home/lock-free-algorithms/queues

Type of queue: Multi-producer/Multi-consumer
Underlying data structure: Array-based
Maximum size: Bounded/Unbounded can be selected at runtime
Overflow behavior: Block until there is room
Requirement for garbage collection: None
Support for priorities: No
Ordering guarantees: per-producer FIFO
Behavior on empty queue: Block until elements arrive

Compatibility
--------------

pipe.c has been tested on:

 * GNU/Linux i686
 * GNU/Linux x86-64
 * Windows 7 x64 (Cross-compiled from GNU/Linux with mingw64)
 * Windows 7 (Compiled with mingw)
 * Windows 7 i686 (Cygwin)
 * OSX 10.6

However, there's no reason it shouldn't work on any of the following
platforms. If you have successfully tested it on any of these, feel free
to let me know by dropping me an email <[email protected]>.

 * Any platform which supports pthreads (GNU/Linux, BSD, Mac)
 * Microsoft Windows NT -> Microsoft Windows 7

If you _must_ use Windows, try to only support windows Vista and
onwards. By using a recent operating system, you use the more robust
built-in condition variables instead of my hacky hand-rolled ones.

Supported Compilers:

Any compiler with C99 support, however, pipe.c has been tested with...

gcc 4.2.1
gcc 4.3.4 (Cygwin)
gcc 4.5.2
llvm-gcc 4.2.1
clang 2.8
clang 1.5 (Darwin)
icc 12.0
mingw 4.5.2
mingw 4.6.0

Unsupported Compilers:

msvc (any version) - No C99 support. The header will work, but the .c
                     itself will never compile on current versions.

The headers are supported by any compiler with ANSI C support and don't
do anything fancy. Therefore, if you want to use pipe.c with a compiler
such as msvc, you can compile the .c file with mingw and the rest of
your program with msvc, and link them all together without a hitch.

License
--------

The MIT License
Copyright (c) 2011 Clark Gaebel <[email protected]>

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.

Speed Tweaking
---------------

There are a few things you can do to get the most speed out of the pipe.

 1 Use a vectorizing compiler, and an SSE-enabled memcpy. It's a lot
   faster, trust me. I recommend icc.
 2 Tweak `MUTEX_SPINS` in pipe.c. Set it to a low value if you tend to
   push large quantities of data at once, and a high value if you tend to
   push small quantities of data. Feel free to set it to zero to disable
   spin locking altogether and just use the native lock implementation.
 3 Turn on your compiler's link-time optimization.
 4 Do profile-guided optimization.
 5 Large buffers are preferable to small ones, unless your use case
   dictates otherwise. Emptying the pipe frequently tends to be faster
   and use less memory.
 6 Avoid limiting the size of the pipe unless you get serious memory
   issues.
 7 If you are always pushing predictable amounts of data into the pipe
   (such as 1000 elements at a time), you should consider using
   `pipe_reserve` to keep the pipe from needlessly growing and
   shrinking.
 8 If you can combine a whole bunch of elements into a single struct, it
   will be faster to push and pop one large struct at a time rather than
   a push/pop of many small structs.
 9 Read through the source code, and find bottlenecks! Don't forget to
   submit changes upstream, so the rest of the world can be in awe of
   your speed-hackery. I have also left a couple optimization hints
   in-source as a reward for actually reading it.

API Overview
-------------

The API is fully documented in pipe.h

Contributing
-------------

This is github! Just fork away and send a pull request. I'll take a look
as soon as I can. You are also always free to drop me an email at
[email protected]. I won't ignore you. Promise!

Things I'd love to see are speed improvements (especially those which
reduce lock contention), support for more compilers/platforms, a better
makefile (the current one is pitiful), and any algorithmic improvements.

Essentially, any patches are welcome which fulfill the goals of pipe:
speed, cross-platform-ness, a simple API, and beautiful code.

If you have successfully built and tested the pipe on a
compiler/architecture not previously mentioned, it would be nice if you
could drop me an email with your success/failure story so I can update
this README accordingly.


Branch information
-------------------

Currently, there are two branches of the code.

master contains the fastest code. However, it is not necessarily the most
readable. It uses two mutexes: one for pushing, and one for popping. This
allows for excellent single-writer single-reader performance. It is "done"
in terms of the only new code going into it is bugfixes and performance
tweaks. If you want to use pipe in your code, you want to checkout this
branch.

single_mutex is the simplest version, and the easiest to read. It is "done"
in terms of the only new code going into it is bugfixes. It implements
the queue with a single god-mutex which is simple in practice, but is not
the most concurrent design decision. If you want to read over an initial
proof-of-concept (and very elegant C), you want to checkout this branch.

pipe's People

Contributors

Stargazers

Watchers

Forkers

andyzweb cgiogkarakis liaowang11 blackball lilrain lkunemail agiz wfrisby rpoisel pkarc cbeighley ideaorg abremges haimingwey mattli001 biyou weiai soon nushio3 pjcrosbie chixsh caodan4linux 01btc10 ifzz bugrepairman daanpape juniperdiego luckychen rharrison a320321wb abhatikar mikaelnousiainen coelias case2111 rcorrieri hermixy frankyyyt ewmailing falconray0704 changsoonkim ilaidlaw simon-p-r twoants vmolsa zhujie358 shangzuoyan baiyfcu victorsow youranli di3online vliuchao heibao111728 wgjobj kamingli1st vdedyukhin marquisrobb amlan-007 fcccode dreamchenhao pioneeralone strongerw bobdeng1974 wwkkww1983 wayyoung leonumix mgood7123 jiongyu embedded-solutions liming2013 curioustauseef diachenkoigor captaintsao puzoliang franander peilix ganjouille huangfcn asdlei99 steveng3 mfkiwl git-blame arielce joysaaaa franzbischoff chengkang123 disconnect3d haifenghuang meqyj

pipe's Issues

cond_t on [windows < vista] is racy

There's a race condition in the hand-rolled condition variables. It's only a little racy though, since it seemed to pass my test suite on windows. However, it is still incorrect and in need of a full rewrite.

If anyone else has more experience than I do on this, I'd love a pull request. I'm really scared of getting it wrong again. I've never worked on this level of concurrency primitives before.

The offending code is from here: https://github.com/wowus/pipe/blob/1d862b7938e2e28d58031831593e63f1810c692f/pipe.c#L112
to here: https://github.com/wowus/pipe/blob/1d862b7938e2e28d58031831593e63f1810c692f/pipe.c#L186

The only API the new condition variables need to support are cond_init, cond_signal, cond_broadcast, cond_wait, and cond_destroy. There is no timed wait, and I'm okay with cond_signal being defined as cond_broadcast if it simplifies the code. It's less efficient, but still correct.

pipe.h code example segfaults

The code example in pipe.h seqfaults:

#include <stdlib.h>
#include <stdio.h>
#include "pipe.h"

#define THREADS 4

int main()
{
  pipe_t* p = pipe_new(sizeof(int), 0);
  pipe_producer_t* pros[THREADS] = { pipe_producer_new(p) };
  pipe_consumer_t* cons[THREADS] = { pipe_consumer_new(p) };
  pipe_free(p);
  // At this point, you can freely use the producer and consumer handles.
  // Then clean them up when you're done!
  for(int i = 0; i < THREADS; ++i) {
    pipe_producer_free(pros[i]);
    pipe_consumer_free(cons[i]);
  }
}

on a mac (10.9) compiled with:

    gcc -Wall -Wextra -std=c99 -O2 -o ptest -I pipe/ ptest.c pipe/pipe.o

Am I doing something stupid?

pipe_push blocking

Any way of blocking pipe_push if the pipe is full?

memory leaks

please fix your memory leaks

YOUR Master branch

valgrind --leak-check=full --show-leak-kinds=all -s --track-origins=yes ./pipe_debug
==325637== Memcheck, a memory error detector
==325637== Copyright (C) 2002-2017, and GNU GPL'd, by Julian Seward et al.
==325637== Using Valgrind-3.15.0 and LibVEX; rerun with -h for copyright info
==325637== Command: ./pipe_debug
==325637== 
basic_storage -> [  OK  ]
pipeline_multiplier -> [  OK  ]
parallel_multiplier -> [  OK  ]
issue_4 -> [  OK  ]
issue_5 -> [  OK  ]
==325637== 
==325637== HEAP SUMMARY:
==325637==     in use at exit: 3,264 bytes in 12 blocks
==325637==   total heap usage: 874 allocs, 862 frees, 162,456,980 bytes allocated
==325637== 
==325637== 272 bytes in 1 blocks are possibly lost in loss record 1 of 8
==325637==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==325637==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==325637==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==325637==    by 0x4876322: allocate_stack (allocatestack.c:622)
==325637==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==325637==    by 0x10BF4D: thread_create (pipe_util.c:49)
==325637==    by 0x10C16B: pipe_connect (pipe_util.c:110)
==325637==    by 0x10C3B5: va_pipe_pipeline (pipe_util.c:158)
==325637==    by 0x10C4D3: pipe_pipeline (pipe_util.c:173)
==325637==    by 0x10B879: test_pipeline_multiplier (pipe_test.c:153)
==325637==    by 0x10BE7C: pipe_run_test_suite (pipe_test.c:306)
==325637==    by 0x10C524: main (main.c:5)
==325637== 
==325637== 272 bytes in 1 blocks are possibly lost in loss record 2 of 8
==325637==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==325637==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==325637==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==325637==    by 0x4876322: allocate_stack (allocatestack.c:622)
==325637==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==325637==    by 0x10BF4D: thread_create (pipe_util.c:49)
==325637==    by 0x10C16B: pipe_connect (pipe_util.c:110)
==325637==    by 0x10C3B5: va_pipe_pipeline (pipe_util.c:158)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C4D3: pipe_pipeline (pipe_util.c:173)
==325637==    by 0x10B879: test_pipeline_multiplier (pipe_test.c:153)
==325637==    by 0x10BE7C: pipe_run_test_suite (pipe_test.c:306)
==325637==    by 0x10C524: main (main.c:5)
==325637== 
==325637== 272 bytes in 1 blocks are possibly lost in loss record 3 of 8
==325637==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==325637==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==325637==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==325637==    by 0x4876322: allocate_stack (allocatestack.c:622)
==325637==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==325637==    by 0x10BF4D: thread_create (pipe_util.c:49)
==325637==    by 0x10C16B: pipe_connect (pipe_util.c:110)
==325637==    by 0x10C3B5: va_pipe_pipeline (pipe_util.c:158)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C4D3: pipe_pipeline (pipe_util.c:173)
==325637==    by 0x10B879: test_pipeline_multiplier (pipe_test.c:153)
==325637==    by 0x10BE7C: pipe_run_test_suite (pipe_test.c:306)
==325637==    by 0x10C524: main (main.c:5)
==325637== 
==325637== 272 bytes in 1 blocks are possibly lost in loss record 4 of 8
==325637==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==325637==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==325637==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==325637==    by 0x4876322: allocate_stack (allocatestack.c:622)
==325637==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==325637==    by 0x10BF4D: thread_create (pipe_util.c:49)
==325637==    by 0x10C16B: pipe_connect (pipe_util.c:110)
==325637==    by 0x10C3B5: va_pipe_pipeline (pipe_util.c:158)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C4D3: pipe_pipeline (pipe_util.c:173)
==325637==    by 0x10B879: test_pipeline_multiplier (pipe_test.c:153)
==325637==    by 0x10BE7C: pipe_run_test_suite (pipe_test.c:306)
==325637== 
==325637== 272 bytes in 1 blocks are possibly lost in loss record 5 of 8
==325637==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==325637==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==325637==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==325637==    by 0x4876322: allocate_stack (allocatestack.c:622)
==325637==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==325637==    by 0x10BF4D: thread_create (pipe_util.c:49)
==325637==    by 0x10C16B: pipe_connect (pipe_util.c:110)
==325637==    by 0x10C3B5: va_pipe_pipeline (pipe_util.c:158)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C4D3: pipe_pipeline (pipe_util.c:173)
==325637==    by 0x10B879: test_pipeline_multiplier (pipe_test.c:153)
==325637== 
==325637== 272 bytes in 1 blocks are possibly lost in loss record 6 of 8
==325637==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==325637==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==325637==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==325637==    by 0x4876322: allocate_stack (allocatestack.c:622)
==325637==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==325637==    by 0x10BF4D: thread_create (pipe_util.c:49)
==325637==    by 0x10C16B: pipe_connect (pipe_util.c:110)
==325637==    by 0x10C3B5: va_pipe_pipeline (pipe_util.c:158)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C4D3: pipe_pipeline (pipe_util.c:173)
==325637== 
==325637== 544 bytes in 2 blocks are possibly lost in loss record 7 of 8
==325637==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==325637==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==325637==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==325637==    by 0x4876322: allocate_stack (allocatestack.c:622)
==325637==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==325637==    by 0x10BF4D: thread_create (pipe_util.c:49)
==325637==    by 0x10C16B: pipe_connect (pipe_util.c:110)
==325637==    by 0x10C3B5: va_pipe_pipeline (pipe_util.c:158)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637==    by 0x10C3E8: va_pipe_pipeline (pipe_util.c:163)
==325637== 
==325637== 1,088 bytes in 4 blocks are possibly lost in loss record 8 of 8
==325637==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==325637==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==325637==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==325637==    by 0x4876322: allocate_stack (allocatestack.c:622)
==325637==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==325637==    by 0x10BF4D: thread_create (pipe_util.c:49)
==325637==    by 0x10C16B: pipe_connect (pipe_util.c:110)
==325637==    by 0x10C1EC: pipe_parallel (pipe_util.c:123)
==325637==    by 0x10B93D: test_parallel_multiplier (pipe_test.c:175)
==325637==    by 0x10BEAA: pipe_run_test_suite (pipe_test.c:307)
==325637==    by 0x10C524: main (main.c:5)
==325637== 
==325637== LEAK SUMMARY:
==325637==    definitely lost: 0 bytes in 0 blocks
==325637==    indirectly lost: 0 bytes in 0 blocks
==325637==      possibly lost: 3,264 bytes in 12 blocks
==325637==    still reachable: 0 bytes in 0 blocks
==325637==         suppressed: 0 bytes in 0 blocks
==325637== 
==325637== ERROR SUMMARY: 8 errors from 8 contexts (suppressed: 0 from 0)
valgrind --leak-check=full --show-leak-kinds=all -s --track-origins=yes ./pipe_release
==325699== Memcheck, a memory error detector
==325699== Copyright (C) 2002-2017, and GNU GPL'd, by Julian Seward et al.
==325699== Using Valgrind-3.15.0 and LibVEX; rerun with -h for copyright info
==325699== Command: ./pipe_release
==325699== 
basic_storage -> [  OK  ]
pipeline_multiplier -> [  OK  ]
parallel_multiplier -> [  OK  ]
issue_4 -> [  OK  ]
issue_5 -> [  OK  ]
==325699== 
==325699== HEAP SUMMARY:
==325699==     in use at exit: 3,264 bytes in 12 blocks
==325699==   total heap usage: 1,642 allocs, 1,630 frees, 164,161,428 bytes allocated
==325699== 
==325699== 544 bytes in 2 blocks are possibly lost in loss record 1 of 3
==325699==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==325699==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==325699==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==325699==    by 0x4876322: allocate_stack (allocatestack.c:622)
==325699==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==325699==    by 0x10B782: pipe_parallel (in /home/smallville7123/AndroidCompositor/app/src/main/jni/GLIS/examples/pipe2/pipe_release)
==325699==    by 0x10AD5C: pipe_run_test_suite (in /home/smallville7123/AndroidCompositor/app/src/main/jni/GLIS/examples/pipe2/pipe_release)
==325699==    by 0x10924C: main (in /home/smallville7123/AndroidCompositor/app/src/main/jni/GLIS/examples/pipe2/pipe_release)
==325699== 
==325699== 544 bytes in 2 blocks are possibly lost in loss record 2 of 3
==325699==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==325699==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==325699==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==325699==    by 0x4876322: allocate_stack (allocatestack.c:622)
==325699==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==325699==    by 0x10B7D1: pipe_parallel (in /home/smallville7123/AndroidCompositor/app/src/main/jni/GLIS/examples/pipe2/pipe_release)
==325699==    by 0x10AD5C: pipe_run_test_suite (in /home/smallville7123/AndroidCompositor/app/src/main/jni/GLIS/examples/pipe2/pipe_release)
==325699==    by 0x10924C: main (in /home/smallville7123/AndroidCompositor/app/src/main/jni/GLIS/examples/pipe2/pipe_release)
==325699== 
==325699== 2,176 bytes in 8 blocks are possibly lost in loss record 3 of 3
==325699==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==325699==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==325699==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==325699==    by 0x4876322: allocate_stack (allocatestack.c:622)
==325699==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==325699==    by 0x10B958: pipe_pipeline (in /home/smallville7123/AndroidCompositor/app/src/main/jni/GLIS/examples/pipe2/pipe_release)
==325699==    by 0x10AB81: pipe_run_test_suite (in /home/smallville7123/AndroidCompositor/app/src/main/jni/GLIS/examples/pipe2/pipe_release)
==325699==    by 0x10924C: main (in /home/smallville7123/AndroidCompositor/app/src/main/jni/GLIS/examples/pipe2/pipe_release)
==325699== 
==325699== LEAK SUMMARY:
==325699==    definitely lost: 0 bytes in 0 blocks
==325699==    indirectly lost: 0 bytes in 0 blocks
==325699==      possibly lost: 3,264 bytes in 12 blocks
==325699==    still reachable: 0 bytes in 0 blocks
==325699==         suppressed: 0 bytes in 0 blocks
==325699== 
==325699== ERROR SUMMARY: 3 errors from 3 contexts (suppressed: 0 from 0)
# do not know equivilant flags for these, attempting to specify generates unknown option
valgrind --tool=callgrind --dump-instr=yes --trace-jump=yes ./pipe_release
==325742== Callgrind, a call-graph generating cache profiler
==325742== Copyright (C) 2002-2017, and GNU GPL'd, by Josef Weidendorfer et al.
==325742== Using Valgrind-3.15.0 and LibVEX; rerun with -h for copyright info
==325742== Command: ./pipe_release
==325742== 
==325742== For interactive control, run 'callgrind_control -h'.
basic_storage -> [  OK  ]
pipeline_multiplier -> [  OK  ]
parallel_multiplier -> [  OK  ]
issue_4 -> [  OK  ]
issue_5 -> [  OK  ]
==325742== 
==325742== Events    : Ir
==325742== Collected : 506741867
==325742== 
==325742== I   refs:      506,741,867
valgrind --tool=cachegrind ./pipe_release
==325758== Cachegrind, a cache and branch-prediction profiler
==325758== Copyright (C) 2002-2017, and GNU GPL'd, by Nicholas Nethercote et al.
==325758== Using Valgrind-3.15.0 and LibVEX; rerun with -h for copyright info
==325758== Command: ./pipe_release
==325758== 
--325758-- warning: L3 cache found, using its data for the LL simulation.
basic_storage -> [  OK  ]
pipeline_multiplier -> [  OK  ]
parallel_multiplier -> [  OK  ]
issue_4 -> [  OK  ]
issue_5 -> [  OK  ]
==325758== 
==325758== I   refs:      525,712,605
==325758== I1  misses:          1,863
==325758== LLi misses:          1,818
==325758== I1  miss rate:        0.00%
==325758== LLi miss rate:        0.00%
==325758== 
==325758== D   refs:      224,821,010  (137,149,616 rd   + 87,671,394 wr)
==325758== D1  misses:      4,132,523  (  2,030,666 rd   +  2,101,857 wr)
==325758== LLd misses:      2,799,043  (    939,656 rd   +  1,859,387 wr)
==325758== D1  miss rate:         1.8% (        1.5%     +        2.4%  )
==325758== LLd miss rate:         1.2% (        0.7%     +        2.1%  )
==325758== 
==325758== LL refs:         4,134,386  (  2,032,529 rd   +  2,101,857 wr)
==325758== LL misses:       2,800,861  (    941,474 rd   +  1,859,387 wr)
==325758== LL miss rate:          0.4% (        0.1%     +        2.1%  )
valgrind --tool=massif ./pipe_release
==325771== Massif, a heap profiler
==325771== Copyright (C) 2003-2017, and GNU GPL'd, by Nicholas Nethercote
==325771== Using Valgrind-3.15.0 and LibVEX; rerun with -h for copyright info
==325771== Command: ./pipe_release
==325771== 
basic_storage -> [  OK  ]
pipeline_multiplier -> [  OK  ]
parallel_multiplier -> [  OK  ]
issue_4 -> [  OK  ]
issue_5 -> [  OK  ]
==325771==

my https://github.com/mgood7123/pipe that has the fix from issue 6 applied

valgrind --leak-check=full --show-leak-kinds=all -s --track-origins=yes ./pipe_debug
==324672== Memcheck, a memory error detector
==324672== Copyright (C) 2002-2017, and GNU GPL'd, by Julian Seward et al.
==324672== Using Valgrind-3.15.0 and LibVEX; rerun with -h for copyright info
==324672== Command: ./pipe_debug
==324672== 
basic_storage -> [  OK  ]
pipeline_multiplier -> [  OK  ]
parallel_multiplier -> [  OK  ]
issue_4 -> [  OK  ]
issue_5 -> [  OK  ]
issue_6_a -> [  OK  ]
issue_6_b -> [  OK  ]
issue_6_c -> [  OK  ]
==324672== 
==324672== HEAP SUMMARY:
==324672==     in use at exit: 3,536 bytes in 13 blocks
==324672==   total heap usage: 2,159 allocs, 2,146 frees, 168,726,704 bytes allocated
==324672== 
==324672== 272 bytes in 1 blocks are possibly lost in loss record 1 of 9
==324672==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==324672==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==324672==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==324672==    by 0x4876322: allocate_stack (allocatestack.c:622)
==324672==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==324672==    by 0x10CB9A: thread_create (pipe_util.c:49)
==324672==    by 0x10CDB8: pipe_connect (pipe_util.c:110)
==324672==    by 0x10D002: va_pipe_pipeline (pipe_util.c:158)
==324672==    by 0x10D120: pipe_pipeline (pipe_util.c:173)
==324672==    by 0x10B8EF: test_pipeline_multiplier (pipe_test.c:157)
==324672==    by 0x10CA3F: pipe_run_test_suite (pipe_test.c:449)
==324672==    by 0x10D171: main (main.c:5)
==324672== 
==324672== 272 bytes in 1 blocks are possibly lost in loss record 2 of 9
==324672==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==324672==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==324672==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==324672==    by 0x4876322: allocate_stack (allocatestack.c:622)
==324672==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==324672==    by 0x10CB9A: thread_create (pipe_util.c:49)
==324672==    by 0x10CDB8: pipe_connect (pipe_util.c:110)
==324672==    by 0x10D002: va_pipe_pipeline (pipe_util.c:158)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D120: pipe_pipeline (pipe_util.c:173)
==324672==    by 0x10B8EF: test_pipeline_multiplier (pipe_test.c:157)
==324672==    by 0x10CA3F: pipe_run_test_suite (pipe_test.c:449)
==324672==    by 0x10D171: main (main.c:5)
==324672== 
==324672== 272 bytes in 1 blocks are possibly lost in loss record 3 of 9
==324672==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==324672==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==324672==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==324672==    by 0x4876322: allocate_stack (allocatestack.c:622)
==324672==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==324672==    by 0x10CB9A: thread_create (pipe_util.c:49)
==324672==    by 0x10CDB8: pipe_connect (pipe_util.c:110)
==324672==    by 0x10D002: va_pipe_pipeline (pipe_util.c:158)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D120: pipe_pipeline (pipe_util.c:173)
==324672==    by 0x10B8EF: test_pipeline_multiplier (pipe_test.c:157)
==324672==    by 0x10CA3F: pipe_run_test_suite (pipe_test.c:449)
==324672==    by 0x10D171: main (main.c:5)
==324672== 
==324672== 272 bytes in 1 blocks are possibly lost in loss record 4 of 9
==324672==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==324672==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==324672==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==324672==    by 0x4876322: allocate_stack (allocatestack.c:622)
==324672==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==324672==    by 0x10CB9A: thread_create (pipe_util.c:49)
==324672==    by 0x10CDB8: pipe_connect (pipe_util.c:110)
==324672==    by 0x10D002: va_pipe_pipeline (pipe_util.c:158)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D120: pipe_pipeline (pipe_util.c:173)
==324672==    by 0x10B8EF: test_pipeline_multiplier (pipe_test.c:157)
==324672==    by 0x10CA3F: pipe_run_test_suite (pipe_test.c:449)
==324672== 
==324672== 272 bytes in 1 blocks are possibly lost in loss record 5 of 9
==324672==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==324672==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==324672==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==324672==    by 0x4876322: allocate_stack (allocatestack.c:622)
==324672==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==324672==    by 0x10CB9A: thread_create (pipe_util.c:49)
==324672==    by 0x10CDB8: pipe_connect (pipe_util.c:110)
==324672==    by 0x10D002: va_pipe_pipeline (pipe_util.c:158)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D120: pipe_pipeline (pipe_util.c:173)
==324672==    by 0x10B8EF: test_pipeline_multiplier (pipe_test.c:157)
==324672== 
==324672== 272 bytes in 1 blocks are possibly lost in loss record 6 of 9
==324672==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==324672==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==324672==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==324672==    by 0x4876322: allocate_stack (allocatestack.c:622)
==324672==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==324672==    by 0x10CB9A: thread_create (pipe_util.c:49)
==324672==    by 0x10CDB8: pipe_connect (pipe_util.c:110)
==324672==    by 0x10D002: va_pipe_pipeline (pipe_util.c:158)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D120: pipe_pipeline (pipe_util.c:173)
==324672== 
==324672== 272 bytes in 1 blocks are possibly lost in loss record 7 of 9
==324672==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==324672==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==324672==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==324672==    by 0x4876322: allocate_stack (allocatestack.c:622)
==324672==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==324672==    by 0x10C562: thread_create (pipe_test.c:323)
==324672==    by 0x10C832: test_issue_6_c (pipe_test.c:372)
==324672==    by 0x10CB53: pipe_run_test_suite (pipe_test.c:455)
==324672==    by 0x10D171: main (main.c:5)
==324672== 
==324672== 544 bytes in 2 blocks are possibly lost in loss record 8 of 9
==324672==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==324672==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==324672==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==324672==    by 0x4876322: allocate_stack (allocatestack.c:622)
==324672==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==324672==    by 0x10CB9A: thread_create (pipe_util.c:49)
==324672==    by 0x10CDB8: pipe_connect (pipe_util.c:110)
==324672==    by 0x10D002: va_pipe_pipeline (pipe_util.c:158)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672==    by 0x10D035: va_pipe_pipeline (pipe_util.c:163)
==324672== 
==324672== 1,088 bytes in 4 blocks are possibly lost in loss record 9 of 9
==324672==    at 0x483DD99: calloc (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==324672==    by 0x40149CA: allocate_dtv (dl-tls.c:286)
==324672==    by 0x40149CA: _dl_allocate_tls (dl-tls.c:532)
==324672==    by 0x4876322: allocate_stack (allocatestack.c:622)
==324672==    by 0x4876322: pthread_create@@GLIBC_2.2.5 (pthread_create.c:660)
==324672==    by 0x10CB9A: thread_create (pipe_util.c:49)
==324672==    by 0x10CDB8: pipe_connect (pipe_util.c:110)
==324672==    by 0x10CE39: pipe_parallel (pipe_util.c:123)
==324672==    by 0x10B9B3: test_parallel_multiplier (pipe_test.c:179)
==324672==    by 0x10CA6D: pipe_run_test_suite (pipe_test.c:450)
==324672==    by 0x10D171: main (main.c:5)
==324672== 
==324672== LEAK SUMMARY:
==324672==    definitely lost: 0 bytes in 0 blocks
==324672==    indirectly lost: 0 bytes in 0 blocks
==324672==      possibly lost: 3,536 bytes in 13 blocks
==324672==    still reachable: 0 bytes in 0 blocks
==324672==         suppressed: 0 bytes in 0 blocks
==324672== 
==324672== ERROR SUMMARY: 9 errors from 9 contexts (suppressed: 0 from 0)

Number of elements in pipe

Hi,
I'm trying to use pipe.c as the underlying mechanism for a Java InputStream in a JNI project.

All good - but I need a way to get the number of elements in the pipe at a specified time.

Is this possible?

Windows Compile

Without the restrict keywords, the code seems to compile and run ok using Visual Studio 2015. How critical is the use of the restrict keyword?

check_invariants: Assertion `((s.begin) >= (s.buffer) && (s.begin) <= (s.bufend))' failed

There is a situation, when s.begin becomes invalid. Let assume the following scenario:

s.buffer = 0x360e0, s.bufend = 0x362ec, elem_size = 0xc
In the pop_without_locking function, first_bytes_to_copy is 8, so, bytes_to_copy > 0.
s.begin = 0x362e0
After s.begin += elem_size; it becomes 0x362ec (s.bufend)
After s.begin = wrap_ptr_if_necessary(s.buffer, s.begin, s.bufend); it becomes 0x360e0 (s.buffer)
After s.begin += bytes_to_copy; it becomes 0x360e4
After s.begin -= elem_size; it becomes 0x360d8 and goes out of s.buffer bound, because it was wrapped before.

Steps to reproduce:

Invoke wrap situation
Read some (not zero!) bytes from the end
Wait until it asserts

I am sorry for the issue without testcase, but I will write it (and a solution, if lucky) as soon as possible.

Could you, please, add a fix? Thanks!

A problem of resize_buffer() function

It seems that resize_buffer() function does not work. See my sample code and output below. As we can see, if I use pipe_push() function to push an int array of size 32, the pipe_pop() function returns 0.

When I debug it, I think the following lines in pipe.c may incur problems but I'm not sure

747  if(unlikely(new_size >= max_cap))
748        new_size = max_cap;

test.c

#include "pipe.h"
#include <stdio.h>
#include <stdlib.h>

//when NUM is 32, 32*4=128bytes, the program does not work and seems it cannot resize_buffer().
//I use GDB to trace the problem. I find that the pip internal buffer size cannot be enlarged.
//I guess the bug comes from: line 748 of pipe.c, but it also does not work when I simply comment lines 747-748.
#define NUM 32
//#define NUM 31, when I change NUM to 31 it works!

int main()
{    
    pipe_t* pipe = pipe_new(sizeof(int), 0);
    pipe_producer_t* p = pipe_producer_new(pipe);
    pipe_consumer_t* c = pipe_consumer_new(pipe);
    pipe_free(pipe);

    int data[NUM];
    for(int i=0; i < NUM; ++i)
    {
        data[i] = i;
    }
    pipe_push(p, data, NUM);
    pipe_producer_free(p);


    int buf[NUM];
    size_t ret = pipe_pop(c, buf, NUM);
    printf("ret=%d\n", ret);
    for(int i=0; i < NUM; ++i)
    {
        printf("value=%d\n", buf[i]);
    }

    pipe_consumer_free(c);

    return 0;
}

The output becomes

cic10 $ gcc -std=c99 test.c pipe.c
cic10 $ ./a.out
ret=0
value=-163754450
value=0
value=-828334326
value=63
value=0
value=0
value=1874632056
value=32731
value=1
value=0
value=0
value=0
value=1
value=0
value=-826142328
value=63
value=73835688
value=32767
value=118
value=0
value=9
value=0
value=73835710
value=32767
value=0
value=0
value=-826141472
value=63
value=73835728
value=32767
value=-823589977
value=63
cic10 $ gcc -v
Using built-in specs.
Target: x86_64-redhat-linux
Configured with: ../configure --prefix=/usr --mandir=/usr/share/man --infodir=/usr/share/info --with-bugurl=http://bugzilla.redhat.com/bugzilla --enable-bootstrap --enable-shared --enable-threads=posix --enable-checking=release --with-system-zlib --enable-__cxa_atexit --disable-libunwind-exceptions --enable-gnu-unique-object --enable-languages=c,c++,objc,obj-c++,java,fortran,ada --enable-java-awt=gtk --disable-dssi --with-java-home=/usr/lib/jvm/java-1.5.0-gcj-1.5.0.0/jre --enable-libgcj-multifile --enable-java-maintainer-mode --with-ecj-jar=/usr/share/java/eclipse-ecj.jar --disable-libjava-multilib --with-ppl --with-cloog --with-tune=generic --with-arch_32=i686 --build=x86_64-redhat-linux
Thread model: posix
gcc version 4.4.7 20120313 (Red Hat 4.4.7-11) (GCC)

Consumers segfaults if processing data too fast

It seems if producers/consumers are writing/reading (i.e., changing content of buffer) very quickly, we get a situation where the pipe (data) is "clobbered" somehow resulting in crash.

This doesn't seem to occur in the regular code because of #6, the pipe buffer continually grows. However, if you apply the patch from the issue, you may run into this problem.

Recommend Projects

React

A declarative, efficient, and flexible JavaScript library for building user interfaces.
Vue.js

🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
Typescript

TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
TensorFlow

An Open Source Machine Learning Framework for Everyone
Django

The Web framework for perfectionists with deadlines.
Laravel

A PHP framework for web artisans
D3

Bring data to life with SVG, Canvas and HTML. 📊📈🎉

Recommend Topics

javascript

JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
web

Some thing interesting about web. New door for the world.
server

A server is a program made to process requests and deliver data to clients.
Machine learning

Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Visualization

Some thing interesting about visualization, use data art
Game

Some thing interesting about game, make everyone happy.

Recommend Org

Facebook

We are working to build community through open source technology. NB: members must have two-factor auth.
Microsoft

Open source projects and samples from Microsoft.
Google

Google ❤️ Open Source for everyone.
Alibaba

Alibaba Open Source for everyone
D3

Data-Driven Documents codes.
Tencent

China tencent open source team.