title: Single Producer / Consumer Ring Buffer in Shared Memory
body: <p>Recently I've been playing about with using shared memory for IPC. One thing I've been trying to implement is a simple ring buffer with 1 process producing and 1 process consuming. Each process has its own sequence number to track its position. These sequence numbers are updated using atomic ops to ensure the correct values are visible to the other process. The producer will block once the ring buffer is full. The code is lock free in that no semaphores or mutexes are used.</p>
  
  <p>Performance wise I'm getting roughly 20 million messages per second on my rather modest VM - Pretty happy with that :)</p>
  
  <p>What I'm curious about how 'correct' my code is. Can anyone spot any inherent issues / race conditions? Here's my code. Thanks in advance for any comments.</p>
  
  <pre><code>#include &lt;stdlib.h&gt;
  #include &lt;stdio.h&gt;
  #include &lt;fcntl.h&gt;
  #include &lt;sys/mman.h&gt;
  #include &lt;sys/stat.h&gt;
  #include &lt;time.h&gt;
  #include &lt;unistd.h&gt;
  #include &lt;string.h&gt;
  
  #define SHM_ID "/mmap-test"
  #define BUFFER_SIZE 4096
  #define SLEEP_NANOS 1000   // 1 micro
  
  struct Message
  {
      long _id;
      char _data[128];
  };
  
  struct RingBuffer
  {
      size_t _rseq;
      char _pad1[64];
  
      size_t _wseq;
      char _pad2[64];
  
      Message _buffer[BUFFER_SIZE];
  };
  
  void
  producerLoop()
  {
      int size = sizeof( RingBuffer );
      int fd = shm_open( SHM_ID, O_RDWR | O_CREAT, 0600 );
      ftruncate( fd, size+1 );
  
      // create shared memory area
      RingBuffer* rb = (RingBuffer*)mmap( 0, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0 );
      close( fd );
  
      // initialize our sequence numbers in the ring buffer
      rb-&gt;_wseq = rb-&gt;_rseq = 0;
      int i = 0;
  
      timespec tss;
      tss.tv_sec = 0;
      tss.tv_nsec = SLEEP_NANOS;
  
      while( 1 )
      {
          // as long as the consumer isn't running behind keep producing
          while( (rb-&gt;_wseq+1)%BUFFER_SIZE != rb-&gt;_rseq%BUFFER_SIZE )
          {
              // write the next entry and atomically update the write sequence number
              Message* msg = &amp;rb-&gt;_buffer[rb-&gt;_wseq%BUFFER_SIZE];
              msg-&gt;_id = i++;
              __sync_fetch_and_add( &amp;rb-&gt;_wseq, 1 );
          }
  
          // give consumer some time to catch up
          nanosleep( &amp;tss, 0 );
      }
  }
  
  void
  consumerLoop()
  {
      int size = sizeof( RingBuffer );
      int fd = shm_open( SHM_ID, O_RDWR, 0600 );
      if( fd == -1 ) {
          perror( "argh!!!" ); return;
      }
  
      // lookup producers shared memory area
      RingBuffer* rb = (RingBuffer*)mmap( 0, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0 );
  
      // initialize our sequence numbers in the ring buffer
      size_t seq =  0;
      size_t pid = -1;
  
      timespec tss;
      tss.tv_sec = 0;
      tss.tv_nsec = SLEEP_NANOS;
  
      while( 1 )
      {
          // while there is data to consume
          while( seq%BUFFER_SIZE != rb-&gt;_wseq%BUFFER_SIZE )
          {
              // get the next message and validate the id
              // id should only ever increase by 1
              // quit immediately if not
              Message msg = rb-&gt;_buffer[seq%BUFFER_SIZE];
              if( msg._id != pid+1 ) {
                  printf( "error: %d %d\n", msg._id, pid ); return;
              }
              pid = msg._id;
              ++seq;
          }
  
          // atomically update the read sequence in the ring buffer
          // making it visible to the producer
          __sync_lock_test_and_set( &amp;rb-&gt;_rseq, seq );
  
          // wait for more data
          nanosleep( &amp;tss, 0 );
      }
  }
  
  int
  main( int argc, char** argv )
  {
      if( argc != 2 ) {
          printf( "please supply args (producer/consumer)\n" ); return -1;
      } else if( strcmp( argv[1], "consumer" ) == 0 ) {
          consumerLoop();
      } else if( strcmp( argv[1], "producer" ) == 0 ) {
          producerLoop();
      } else {
          printf( "invalid arg: %s\n", argv[1] ); return -1;
      }
  }
  </code></pre>