Re: Avoiding deadlock with mutex trylock in multithreaded app.

[Pierre Mallard <mallard dot pierre at gmail dot com>]

Thanks for your help Rémi

>Generally speaking you should avoid holding locks while entering a callback.
>Copy the callback parameters to the stack, release your lock, invoke the
>callback, and acquire your lock again. Alternatively, use a recursive lock.
>Either way, be careful that the lock-protected state might have changed in the
>mean time.

Following your advice, I found a solution that please me :
In A unregister does not tell A thread the current request is to be
removed. Instead it returns to caller telling it should wait callback
end.
In this case the release function set a boolean for the callback to
stops and cond_wait for the callback to say it stops.
This is OK for me because the first thing I do in the callback is to
know whether I should stop or not. In such a case I cond_signal and
get a new request in A.
I posted the new code at the end.

Still a question in this code. I saw that when my callback was
sleeping to simulate processing, i can not get the lock on B's mutex
in release. I have added a sched_yield after each call to callback in
A but still I have some callback calls that keep locking. The only
things that help was to add a sleep instead of sched_yield. Is there
anyway to better handle this problem ?.

Here is a little program that show the problem :
*******************************************************************************************************
#include <pthread.h>
#include <unistd.h>
#include <stdio.h>

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
volatile int stop = 0;
pthread_t thread;
void * task(void * dummy)
{
  while (!stop){
    pthread_mutex_lock(&mutex);
    fprintf(stderr,"%s %s %d\n",__FILE__,__FUNCTION__,__LINE__);
    /* I keep locking if I sleep there */
    sleep(1);
    fprintf(stderr,"%s %s %d\n",__FILE__,__FUNCTION__,__LINE__);
    pthread_mutex_unlock(&mutex);
  }
  pthread_exit(0);
}

int main(){

  pthread_create(&thread,NULL,&task,NULL);
  /*   sleep(1); */
  struct sched_param param; int policy;
  pthread_getschedparam(pthread_self(),&policy,&param);
  sleep(1);
  fprintf(stderr,"%s %s %d\n",__FILE__,__FUNCTION__,__LINE__);
  pthread_mutex_lock(&mutex);
  fprintf(stderr,"%s %s %d\n",__FILE__,__FUNCTION__,__LINE__);
  stop = 1;
  pthread_mutex_unlock(&mutex);
  return 0;
}
*************************************************************************************************

>By the way, your code is buggy. If thread cancellation is acted upon from
>within fprintf() or from within the callback, the lock will stale.

I am not really familiar with pthread_cancel (It was my first time,...
!), I red somewhere that glibc does not yet conform to POSIX on this
point (i.e. none glibc function are cancelation point). Anyway I will
be more carefull next time (I have revert to my old boolean stop in A
context ;) )
I noticed also a wrong copy/paste in main for B's initialisation : the
first one is called on an unitialised table index. It is removed in
new code.

Regards,
-- 
Pierre Mallard

#include <pthread.h>
#include <unistd.h>
#include <stdio.h>
#include <sched.h>
#include <string.h>

typedef int (*callback_t)(void *);

#define NB_B 10
#define A_QUEUE_SIZE (NB_B + 1) //+1 for circular
#define CALLBACK_RUNNING 0x00000001

/* request structure of Module A */
struct a_req{
  callback_t func;
  void * param;
};

/* Module A */
struct a {
  pthread_t task;
  pthread_mutex_t mutex;

  /* The queue has been simplified to a circular buffer, 
     has no unregistration takes place while request is in queue for now */
  struct a_req queue[A_QUEUE_SIZE];
  int rd;
  int wr;

  /* current request treated by A, 
     it allows to check whether it is the request 
     we want to unregister or not */
  struct a_req * cur_req;

  /* terminate thread */
  int stop;
};

int a_needs_stop(struct a * this){
  int ret = 0;
  pthread_mutex_lock(&this->mutex);
  if (this->stop){
    ret = 1;
  }
  pthread_mutex_unlock(&this->mutex);
  return ret;
}

/* Thread A task*/
void * a_task(void * param)
{
  struct a * this = (struct a *) param;
  struct a_req cur_req;

  while (!a_needs_stop(this)){
    //get a request from queue
    pthread_mutex_lock(&this->mutex);
    if (this->rd != this->wr){
      this->cur_req = &this->queue[this->rd];
      memcpy(&cur_req,&this->queue[this->rd],sizeof(struct a_req));
      this->rd = (this->rd ++)%A_QUEUE_SIZE;
    }
    else {
      pthread_mutex_unlock(&this->mutex);
      continue;
    }
    fprintf(stderr,"New B : %p\n",this->cur_req->param);
    pthread_mutex_unlock(&this->mutex);

    //Call callback till unregister
    while (1){
      // Needed to give a chance for b_release to get the lock
      sleep(1);

      if (cur_req.func(cur_req.param) == -1){
	break;
      }
    }
  }
  pthread_exit(NULL);
}

int a_init(struct a * this)
{
  this->rd = this->wr;
  pthread_mutex_init(&this->mutex,NULL);
  this->cur_req = NULL;
  this->stop=0;
  if (pthread_create(&this->task,NULL,&a_task,this) < 0){
    return -1;
  }
  return 0;
}

void a_release(struct a * this){

  pthread_mutex_lock(&this->mutex);
  this->stop = 1;
  pthread_mutex_unlock(&this->mutex);

  pthread_join(this->task,NULL);

  pthread_mutex_destroy(&this->mutex);

  return;
}

void * a_register(struct a * this,callback_t func, void * param){
  void * ret;
  pthread_mutex_lock(&this->mutex);
  if (this->wr == (this->rd-1) % A_QUEUE_SIZE){
    pthread_mutex_unlock(&this->mutex);
    return NULL;
  }
  this->queue[this->wr].func = func;
  this->queue[this->wr].param = param;
  ret = &this->queue[this->wr];
  this->wr = (this->wr + 1) % A_QUEUE_SIZE;
  pthread_mutex_unlock(&this->mutex);
  return ret;
}

int a_unregister(struct a * this,void * req){
  pthread_mutex_lock(&this->mutex);
  if (req != this->cur_req){ // we do not handle still pending request ...
    pthread_mutex_unlock(&this->mutex);
    return -1;
  }
  pthread_mutex_unlock(&this->mutex);
  return CALLBACK_RUNNING;
}

struct b{
  pthread_mutex_t mutex;
  pthread_cond_t stop_cb_cond;
  int stop_pending;
  void * req;
  struct a * a;
};

int b_init(struct b * this,struct a * a)
{
  this->req = NULL;
  this->a = a;
  this->stop_pending = 0;
  pthread_cond_init(&this->stop_cb_cond,NULL);
  pthread_mutex_init(&this->mutex,NULL);
  return 0;
}

int b_callback(void * vthis){
  struct b * this = (struct b *) vthis;

  pthread_mutex_lock(&this->mutex);

  if (this->stop_pending){
    /* tells b_release we are stops */
    pthread_cond_signal(&this->stop_cb_cond);
    pthread_mutex_unlock(&this->mutex);
    return -1;
  }

  fprintf(stderr,"CB : %p\n",this);
  sleep(1);

  // if ever process finish :
  // this->req = NULL; // I'm then unregistered in A

  pthread_mutex_unlock(&this->mutex);
  return 0;
}

int b_start(struct b * this){
  pthread_mutex_lock(&this->mutex);
  if ((this->req = a_register(this->a,&b_callback,this)) == NULL){
    pthread_mutex_unlock(&this->mutex);
    return -1;
  }
  pthread_mutex_unlock(&this->mutex);
  return 0;
}

void b_release(struct b * this){
  pthread_mutex_lock(&this->mutex);

  //stop callback...
  this->stop_pending = 1;

  // Unregister callback...
  if (this->req != NULL){
    if (a_unregister(this->a,this->req) == CALLBACK_RUNNING){
      /* wait callback to stops */
      pthread_cond_wait(&this->stop_cb_cond,&this->mutex);
    }
  }

  //do sthg...
  sleep(1);

  pthread_mutex_unlock(&this->mutex);

  /* release all structure rest*/
  pthread_mutex_destroy(&this->mutex);
  pthread_cond_destroy(&this->stop_cb_cond);
}

int main(void){
  struct a a;
  struct b b[NB_B];
  int i;

  if (a_init(&a) < 0){
    return -1;
  }

  for (i = 0 ; i < NB_B; i++){
    b_init(&b[i],&a);
    b_start(&b[i]);
  }

  for (i = 0 ; i < NB_B; i++){
    /* do sthg ...*/
    sleep(5);
    fprintf(stderr,"Releasing %p\n",&b[i]);
    b_release(&b[i]); /* b_stop is performed by b_release */
  }

  a_release(&a);
  
  return 0;
}