title: Process-shared condition variable : how to recover after one process dies?
body: <p>I'm working on a simple FIFO queue to synchronize multiple instances of a server process.</p>
  
  <p>This is very similar to
  <a href="https://stackoverflow.com/questions/3050083/linux-synchronization-with-fifo-waiting-queue">Linux synchronization with FIFO waiting queue</a>, except dealing with multiple processes instead of threads. I adapted <a href="https://stackoverflow.com/a/3050871">caf's ticket lock</a> to use process-shared mutex and condition variable from a shared memory segment. It also handles timeouts in case one process dies while processing a request:</p>
  
  <pre><code>#include &lt;assert.h&gt;
  #include &lt;stdlib.h&gt;
  #include &lt;stdio.h&gt;
  #include &lt;string.h&gt;
  #include &lt;unistd.h&gt;
  #include &lt;sys/types.h&gt;
  #include &lt;sys/stat.h&gt;
  #include &lt;fcntl.h&gt;
  #include &lt;sys/mman.h&gt;
  #include &lt;pthread.h&gt;
  #include &lt;signal.h&gt;
  #include &lt;errno.h&gt;
  
  static inline void fail(char *str)
  {
      perror(str);
      exit(1);
  }
  
  /***************************************************************************************************/
  
  /* Simple ticket lock queue with pthreads
   * https://stackoverflow.com/questions/3050083/linux-synchronization-with-fifo-waiting-queue
   */
  
  typedef struct ticket_lock {
      pthread_mutex_t mutex;
      pthread_cond_t  cond;
      int queue_head, queue_tail;
  } ticket_lock_t;
  
  static void
  ticket_init(ticket_lock_t *t)
  {
      pthread_mutexattr_t mattr;
      pthread_mutexattr_init(&amp;mattr);
      pthread_mutexattr_setpshared(&amp;mattr, PTHREAD_PROCESS_SHARED);
      pthread_mutexattr_setrobust(&amp;mattr, PTHREAD_MUTEX_ROBUST);
      pthread_mutex_init(&amp;t-&gt;mutex, &amp;mattr);
      pthread_mutexattr_destroy(&amp;mattr);
  
      pthread_condattr_t cattr;
      pthread_condattr_init(&amp;cattr);
      pthread_condattr_setpshared(&amp;cattr, PTHREAD_PROCESS_SHARED);
      pthread_cond_init(&amp;t-&gt;cond, &amp;cattr);
      pthread_condattr_destroy(&amp;cattr);
  
      t-&gt;queue_head = t-&gt;queue_tail = 0;
  }
  
  static void
  ticket_broadcast(ticket_lock_t *ticket)
  {
      pthread_cond_broadcast(&amp;ticket-&gt;cond);
  }
  
  static int
  ticket_lock(ticket_lock_t *ticket)
  {
      pthread_mutex_lock(&amp;ticket-&gt;mutex);
      int queue_me = ticket-&gt;queue_tail++;
  
      while (queue_me &gt; ticket-&gt;queue_head) {
          time_t sec = time(NULL) + 5;  /* 5s timeout */
          struct timespec ts = { .tv_sec = sec, .tv_nsec = 0 };
  
          fprintf(stderr, "%i: waiting, current: %i  me: %i\n", getpid(), ticket-&gt;queue_head, queue_me);
  
          if (pthread_cond_timedwait(&amp;ticket-&gt;cond, &amp;ticket-&gt;mutex, &amp;ts) == 0)
              continue;
  
          if (errno != ETIMEDOUT) fail("pthread_cond_timedwait");
  
          /* Timeout, kick current user... */
          fprintf(stderr, "kicking stale ticket %i\n", ticket-&gt;queue_head);
          ticket-&gt;queue_head++;
          ticket_broadcast(ticket);
      }
  
      pthread_mutex_unlock(&amp;ticket-&gt;mutex);
      return queue_me;
  }
  
  static void
  ticket_unlock(ticket_lock_t *ticket, int me)
  {
      pthread_mutex_lock(&amp;ticket-&gt;mutex);
      if (ticket-&gt;queue_head == me) {  /* Normal case: we haven't timed out. */
          ticket-&gt;queue_head++;
          ticket_broadcast(ticket);
      }
      pthread_mutex_unlock(&amp;ticket-&gt;mutex);
  }
  
  
  /***************************************************************************************************/
  /* Shared memory */
  
  #define SHM_NAME    "fifo_sched"
  #define SHM_MAGIC   0xdeadbeef
  
  struct sched_shm {
      int  size;
      int  magic;
      int  ready;
  
      /* sched stuff */
      ticket_lock_t queue;
  };
  
  static unsigned int shm_size = 256;
  static struct sched_shm *shm = 0;
  
  /* Create new shared memory segment */
  static void
  create_shm()
  {
         int fd = shm_open(SHM_NAME, O_RDWR | O_CREAT | O_TRUNC, 0644);
         assert(fd != -1);
         int r = ftruncate(fd, shm_size);  assert(r == 0);
         void *pt = mmap(0, shm_size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
         assert(pt != MAP_FAILED);
         fprintf(stderr, "Created shared memory.\n");
  
         shm = pt;
         memset(shm, 0, sizeof(*shm));
         shm-&gt;size = shm_size;
         shm-&gt;magic = SHM_MAGIC;
         shm-&gt;ready = 0;
  
         ticket_init(&amp;shm-&gt;queue);
  
         shm-&gt;ready = 1;
  }
  
  /* Attach existing shared memory segment */
  static int
  attach_shm()
  {
      int fd = shm_open(SHM_NAME, O_RDWR, 0);
      if (fd == -1)  return 0;  /* Doesn't exist yet... */
  
      shm = mmap(0, shm_size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
      if (shm == MAP_FAILED)  fail("mmap");
      fprintf(stderr, "Mapped shared memory.\n");
  
      assert(shm-&gt;magic == SHM_MAGIC);
      assert(shm-&gt;ready);
      return 1;
  }
  
  static void
  shm_init()
  {  
      fprintf(stderr, "shm_init()\n");
      assert(shm_size &gt;= sizeof(struct sched_shm));
      if (!attach_shm())
          create_shm();
  }
  
  
  /***************************************************************************************************/
  
  int main()
  {
      shm_init();
  
      while (1) {
          int ticket = ticket_lock(&amp;shm-&gt;queue);
          printf("%i: start %i\n", getpid(), ticket);
          printf("%i: done  %i\n", getpid(), ticket);
          ticket_unlock(&amp;shm-&gt;queue, ticket);
      }
      return 0;
  }
  </code></pre>
  
  <p>This works well standalone and while adding extra processes:</p>
  
  <pre><code>$ gcc -g -Wall -std=gnu99 -o foo foo.c -lpthread -lrt
  $ ./foo
  $ ./foo     # (in other term)
  ...
  26370: waiting, current: 134803  me: 134804
  26370: start 134804
  26370: done  134804
  26370: waiting, current: 134805  me: 134806
  26370: start 134806
  26370: done  134806
  26370: waiting, current: 134807  me: 134808
  </code></pre>
  
  <p>However killing the 2nd instance breaks pthread_cond_timedwait() in the 1st:</p>
  
  <pre><code>pthread_cond_timedwait: No such file or directory
  </code></pre>
  
  <p>Which makes sense in a way, the condition variable was tracking this process and it's not there anymore.</p>
  
  <p>Surely there must be a way to recover from this ?</p>