Simple solution for calling a function only on one thread, queuing waiting calls?

Question

Let’s assume there is a function EventHandler that is called on multiple threads at different points in time.

EventHandler has to call SomeOtherFunction, but these calls shall only happen on one thread at a time.

That means: If a call to SomeOtherFunction is currently active on thread A, and EventHandler gets called again (on some other thread B), then SomeOtherFunction must not be called on thread B, but SomeOtherFunction shall be called when it returns on thread A.

Something like this (pseudo-code):

function EventHandler()
{
  numberPendingCallsSemaphore.Release();    // increase
  if (mayCallSemaphore.Acquire())           // (i)
  {
    while (numberPendingCallsSemaphore.Acquire())  // decrease
    {
      SomeOtherFunction();
    }
    // (ii)
    mayCallSemaphore.Release();
  }
}

Of course, this code does not work because of a race: One thread is at position (ii) with mayCallSemaphore still being taken, and another at position (i) not being able to take mayCallSemaphore.

I wrote working code using a lock, but the code is quite long and ugly. On the other, this problem looks quite simple at first sight, thus:

Is there a simple, potentially straightforward solution for this I was not able to find or currently don’t see?

Answer 1

Your proposed solution has two control objects. mayCallSemaphore looks like a Mutex based on how you are using it, and numberPendingCallsSemaphore appears to serve as a Condition Variable. You have it almost correct, but as you surmised, having two control structures that operate independently leads to race conditions, deadlock or both depending on the topology and situation. Lost wakeups and other things are also possible.

If you take a close look at this GitHub documentation about how to construct a Counting Semaphore using both a Mutex and a Condition Variable, you will see that the CV actually uses (cooperates with) the Mutex internally, so there can be no races or deadlocks. I will copy and paste the example solution right here for convenience:

typedef struct sem_t {
  int count; 
  pthread_mutex_t m;
  pthread_condition_t cv;
} sem_t;

int sem_init(sem_t *s, int pshared, int value) {
  if (pshared) { errno = ENOSYS /* 'Not implemented'*/; return -1;}

  s->count = value;
  pthread_mutex_init(&s->m, NULL);
  pthread_cond_init(&s->cv, NULL);
  return 0;
}

sem_post(sem_t *s) {
  pthread_mutex_lock(&s->m);
  s->count++;
  pthread_cond_signal(&s->cv); /* See note */
  /* A woken thread must acquire the lock, so it will also have to wait until we call unlock*/

  pthread_mutex_unlock(&s->m);
}

sem_wait(sem_t *s) {
  pthread_mutex_lock(&s->m);
  while (s->count == 0) {
      pthread_cond_wait(&s->cv, &s->m); /*unlock mutex, wait, relock mutex*/
  }
  s->count--;
  pthread_mutex_unlock(&s->m);
}

Simply protect your call to SomeOtherFunction() by calling sem_wait() before and then calling sem_post() after. The internals of the methods will take care of all of the details, causing concurrent accesses to block, allowing only one execution of your code's critical section at a time. This is all based on the Posix primitives which are part of the OS and cannot go wrong. In your case, the count should be restricted to one, so you can either simplify this code, or initialize the semaphore with 1 and make use of the same code for another situation later.

I have used this same approach - taken from the POSIX manual pages (about 20 years ago) - to create a multi-threaded, multi-process database transaction server before the Apache web server was multi-threaded. I had one main thread that accepts incoming TCP connections, and creates up to 'N' work threads at a time (transient, not a pool). It worked for a long time in a very large site with many users. Unix gives you the tools to construct proper solutions, you just need to put them together.