tpi_tnycthrd.c

/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
/*                                                                           */
/*                  This file is part of the program and library             */
/*         SCIP --- Solving Constraint Integer Programs                      */
/*                                                                           */
/*  Copyright (c) 2002-2024 Zuse Institute Berlin (ZIB)                      */
/*                                                                           */
/*  Licensed under the Apache License, Version 2.0 (the "License");          */
/*  you may not use this file except in compliance with the License.         */
/*  You may obtain a copy of the License at                                  */
/*                                                                           */
/*      http://www.apache.org/licenses/LICENSE-2.0                           */
/*                                                                           */
/*  Unless required by applicable law or agreed to in writing, software      */
/*  distributed under the License is distributed on an "AS IS" BASIS,        */
/*  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. */
/*  See the License for the specific language governing permissions and      */
/*  limitations under the License.                                           */
/*                                                                           */
/*  You should have received a copy of the Apache-2.0 license                */
/*  along with SCIP; see the file LICENSE. If not visit scipopt.org.         */
/*                                                                           */
/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */

/**@file   tpi_tnycthrd.c
 * @ingroup TASKINTERFACE
 * @brief  a TPI implementation using tinycthreads
 * @author Stephen J. Maher
 * @author Leona Gottwald
 * @author Marc Pfetsch
 */

/*---+----1----+----2----+----3----+----4----+----5----+----6----+----7----+----8----+----9----+----0----+----1----+----2*/

#include "tpi/tpi.h"
#include "blockmemshell/memory.h"
#include "tinycthread/tinycthread.h"
#include "scip/pub_message.h"

/* macros for direct access */

/* lock */
#define SCIPtnyInitLock(lock)                 ( mtx_init((lock), mtx_plain) == thrd_success ? SCIP_OKAY : SCIP_ERROR )
#define SCIPtnyDestroyLock(lock)              ( mtx_destroy(lock) )
#define SCIPtnyAcquireLock(lock)              ( mtx_lock(lock) == thrd_success ? SCIP_OKAY : SCIP_ERROR )
#define SCIPtnyReleaseLock(lock)              ( mtx_unlock(lock) == thrd_success ? SCIP_OKAY : SCIP_ERROR )

/* condition */
#define SCIPtnyInitCondition(condition)       ( cnd_init(condition) == thrd_success ? SCIP_OKAY : SCIP_ERROR )
#define SCIPtnyDestroyCondition(condition)    ( cnd_destroy(condition) )
#define SCIPtnySignalCondition(condition)     ( cnd_signal(condition) == thrd_success ? SCIP_OKAY : SCIP_ERROR )
#define SCIPtnyBroadcastCondition(condition)  ( cnd_broadcast(condition) == thrd_success ? SCIP_OKAY : SCIP_ERROR )
#define SCIPtnyWaitCondition(condition, lock) ( cnd_wait((condition), (lock)) == thrd_success ? SCIP_OKAY: SCIP_ERROR )

/** struct containing lock */
struct SCIP_Lock
{
   mtx_t                 lock;
};

/** struct containing condition */
struct SCIP_Condition
{
   cnd_t                 condition;
};


typedef struct SCIP_ThreadPool SCIP_THREADPOOL;
static SCIP_THREADPOOL* _threadpool = NULL;
_Thread_local int _threadnumber; /*lint !e129*/

/** A job added to the queue */
struct SCIP_Job
{
   int                   jobid;              /**< id to identify jobs from a common process */
   struct                SCIP_Job* nextjob;  /**< pointer to the next job in the queue */
   SCIP_RETCODE          (*jobfunc)(void* args);/**< pointer to the job function */
   void*                 args;               /**< pointer to the function arguments */
   SCIP_RETCODE          retcode;            /**< return code of the job */
};

/** the thread pool job queue */
struct SCIP_JobQueue
{
   SCIP_JOB*             firstjob;           /**< pointer to the first job in the queue */
   SCIP_JOB*             lastjob;            /**< pointer to the last job in the queue */
   int                   njobs;              /**< number of jobs in the queue */
};
typedef struct SCIP_JobQueue SCIP_JOBQUEUE;

/** The thread pool */
struct SCIP_ThreadPool
{
   /* Pool Characteristics */
   int                   nthreads;           /**< number of threads in the pool */
   int                   queuesize;          /**< the total number of items to enter the queue */

   /* Current pool state */
   thrd_t*               threads;            /**< the threads included in the pool */
   SCIP_JOBQUEUE*        jobqueue;           /**< the job queue */
   SCIP_JOBQUEUE*        currentjobs;        /**< the jobs currently being processed on a thread;
                                              *   only a single job is allowed per thread. */
   SCIP_JOBQUEUE*        finishedjobs;       /**< finished jobs that are not yet collected */
   int                   currworkingthreads; /**< the threads currently processing jobs */
   SCIP_Bool             blockwhenfull;      /**< indicates that the queue can only be as large as nthreads */
   int                   currentid;          /**< current job id */

   /* Control indicators */
   SCIP_Bool             shutdown;           /**< indicates whether the pool needs to be shut down */
   SCIP_Bool             queueopen;          /**< indicates whether the queue is open */

   /* mutex and locks for the thread pool */
   mtx_t                 poollock;           /**< mutex to allow read and write of the pool features */
   cnd_t                 queuenotempty;      /**< condition to broadcast the queue has jobs */
   cnd_t                 queuenotfull;       /**< condition to broadcast the queue is not full */
   cnd_t                 queueempty;         /**< condition to broadcast that the queue is empty */
   cnd_t                 jobfinished;        /**< condition to broadcast that a job has been finished */
};

/** this function controls the execution of each of the threads */
static
SCIP_RETCODE threadPoolThreadRetcode(
   void*                 threadnum           /**< thread number is passed in as argument stored inside a void pointer */
   )
{
   SCIP_JOB* newjob;
   SCIP_JOB* prevjob;
   SCIP_JOB* currjob;

   _threadnumber = (int)(uintptr_t) threadnum;

   /* Increase the number of active threads */
   SCIP_CALL( SCIPtnyAcquireLock(&(_threadpool->poollock)) );
   _threadpool->currworkingthreads += 1;
   SCIP_CALL( SCIPtnyReleaseLock(&(_threadpool->poollock)) );

   /* this is an endless loop that runs until the thrd_exit function is called */
   while( TRUE ) /*lint !e716*/
   {
      SCIP_CALL( SCIPtnyAcquireLock(&(_threadpool->poollock)) );

      /* the queue is empty but the shutdown command has not been given */
      while( _threadpool->jobqueue->njobs == 0 && !_threadpool->shutdown )
      {
         SCIP_CALL( SCIPtnyWaitCondition(&(_threadpool->queuenotempty), &(_threadpool->poollock)) );
      }

      /* if the shutdown command has been given, then exit the thread */
      if( _threadpool->shutdown )
      {
         /* Decrease the thread count when execution of job queue has completed */
         _threadpool->currworkingthreads -= 1;
         SCIP_CALL( SCIPtnyReleaseLock(&(_threadpool->poollock)) );

         thrd_exit((int)SCIP_OKAY);
      }

      /* getting the next job in the queue */
      newjob = _threadpool->jobqueue->firstjob;
      _threadpool->jobqueue->njobs--;            /* decreasing the number of jobs in the queue */

      if( _threadpool->jobqueue->njobs == 0 )
      {
         _threadpool->jobqueue->firstjob = NULL;
         _threadpool->jobqueue->lastjob = NULL;
      }
      else
         _threadpool->jobqueue->firstjob = newjob->nextjob;     /* updating the queue */

      /* if we want to wait when the queue is full, then we broadcast that the queue can now take new jobs */
      if( _threadpool->blockwhenfull &&
          _threadpool->jobqueue->njobs == _threadpool->queuesize - 1 )
      {
         SCIP_CALL( SCIPtnyBroadcastCondition(&(_threadpool->queuenotfull)) );
      }

      /* indicating that the queue is empty */
      if( _threadpool->jobqueue->njobs == 0 )
      {
         SCIP_CALL( SCIPtnyBroadcastCondition(&(_threadpool->queueempty)) );
      }

      /* updating the current job list */
      if( _threadpool->currentjobs->njobs == 0 )
      {
         _threadpool->currentjobs->firstjob = newjob;
         _threadpool->currentjobs->lastjob = newjob;
      }
      else
      {
         _threadpool->currentjobs->lastjob->nextjob = newjob;
         _threadpool->currentjobs->lastjob = newjob;
      }

      _threadpool->currentjobs->njobs++;

      SCIP_CALL( SCIPtnyReleaseLock(&(_threadpool->poollock)) );

      /* setting the job to run on this thread */
      newjob->retcode = (*(newjob->jobfunc))(newjob->args);

      /* setting the current job on this thread to NULL */
      SCIP_CALL( SCIPtnyAcquireLock(&(_threadpool->poollock)) );

      /* finding the location of the processed job in the currentjobs queue */
      currjob = _threadpool->currentjobs->firstjob;
      prevjob = NULL;

      while( currjob != newjob )
      {
         prevjob = currjob;
         currjob = prevjob->nextjob;
      }

      /* removing the processed job from current jobs list */
      if( currjob == _threadpool->currentjobs->firstjob )
         _threadpool->currentjobs->firstjob = currjob->nextjob;
      else
         prevjob->nextjob = currjob->nextjob; /*lint !e794*/

      if( currjob == _threadpool->currentjobs->lastjob )
         _threadpool->currentjobs->lastjob = prevjob;

      _threadpool->currentjobs->njobs--;

      /* updating the finished job list */
      if( _threadpool->finishedjobs->njobs == 0 )
      {
         _threadpool->finishedjobs->firstjob = newjob;
         _threadpool->finishedjobs->lastjob = newjob;
      }
      else
      {
         _threadpool->finishedjobs->lastjob->nextjob = newjob;
         _threadpool->finishedjobs->lastjob = newjob;
      }

      _threadpool->finishedjobs->njobs++;

      /* signalling that a job has been finished */
      SCIP_CALL( SCIPtnyBroadcastCondition(&(_threadpool)->jobfinished) );

      SCIP_CALL( SCIPtnyReleaseLock(&(_threadpool->poollock)) );
   }
}

/** this function controls the execution of each of the threads */
static
int threadPoolThread(
   void*                 threadnum           /**< thread number is passed in as argument stored inside a void pointer */
   )
{
   return (int) threadPoolThreadRetcode(threadnum);
}

/** creates a threadpool */
static
SCIP_RETCODE createThreadPool(
   SCIP_THREADPOOL**     thrdpool,           /**< pointer to store threadpool */
   int                   nthreads,           /**< number of threads in the threadpool */
   int                   qsize,              /**< maximum size of the jobqueue */
   SCIP_Bool             blockwhenfull       /**< should the jobqueue block if it is full */
   )
{
   uintptr_t i;

   assert(nthreads >= 0);
   assert(qsize >= 0);

   /* @todo think about the correct memory here */
   SCIP_ALLOC( BMSallocMemory(thrdpool) );
   (*thrdpool)->currentid = 0;
   (*thrdpool)->queuesize = qsize;
   (*thrdpool)->nthreads = nthreads;
   (*thrdpool)->blockwhenfull = blockwhenfull;
   (*thrdpool)->shutdown = FALSE;
   (*thrdpool)->queueopen = TRUE;

   /* allocating memory for the job queue */
   SCIP_ALLOC( BMSallocMemory(&(*thrdpool)->jobqueue) );
   (*thrdpool)->jobqueue->firstjob = NULL;
   (*thrdpool)->jobqueue->lastjob = NULL;
   (*thrdpool)->jobqueue->njobs = 0;

   /* allocating memory for the job queue */
   SCIP_ALLOC( BMSallocMemory(&(*thrdpool)->currentjobs) );
   (*thrdpool)->currentjobs->firstjob = NULL;
   (*thrdpool)->currentjobs->lastjob = NULL;
   (*thrdpool)->currentjobs->njobs = 0;

   /* allocating memory for the job queue */
   SCIP_ALLOC( BMSallocMemory(&(*thrdpool)->finishedjobs) );
   (*thrdpool)->finishedjobs->firstjob = NULL;
   (*thrdpool)->finishedjobs->lastjob = NULL;
   (*thrdpool)->finishedjobs->njobs = 0;

   /* initialising the mutex */
   SCIP_CALL( SCIPtnyInitLock(&(*thrdpool)->poollock) ); /*lint !e2482*/

   /* initialising the conditions */
   SCIP_CALL( SCIPtnyInitCondition(&(*thrdpool)->queuenotempty) );
   SCIP_CALL( SCIPtnyInitCondition(&(*thrdpool)->queuenotfull) );
   SCIP_CALL( SCIPtnyInitCondition(&(*thrdpool)->queueempty) );
   SCIP_CALL( SCIPtnyInitCondition(&(*thrdpool)->jobfinished) );

   /* creating the threads */
   (*thrdpool)->currworkingthreads = 0;

   /* allocating memory for the threads */
   SCIP_ALLOC( BMSallocMemoryArray(&((*thrdpool)->threads), nthreads) );

   /* create the threads */
   for( i = 0; i < (unsigned)nthreads; i++ )
   {
      if( thrd_create(&((*thrdpool)->threads[i]), threadPoolThread, (void*)i) != thrd_success )
         return SCIP_ERROR;
   }

   _threadnumber = nthreads;
   /* halt while all threads are not active TODO: is synchronization required here ? */
   /*TODO: this caused a deadlock, is it important to wait for all threads to start?
    * while( (*thrdpool)->currworkingthreads != nthreads )
   {}*/

   return SCIP_OKAY;
}

/** adding a job to the job queue.
 *
 *  This gives some more flexibility in the handling of new jobs.
 *  This function needs to be called from within a mutex.
 */
static
void jobQueueAddJob(
   SCIP_THREADPOOL*      threadpool,         /**< pointer to store threadpool */
   SCIP_JOB*             newjob              /**< pointer to new job */
   )
{
   /* @todo we want to work out what to do with a full job queue. Is there a problem if the limit is hit? */
   /* @note it is important to have a queuesize. This will stop the code submitting infinitely many jobs. */
   assert(threadpool->jobqueue->njobs < threadpool->queuesize);

   newjob->nextjob = NULL;

   /* checking the status of the job queue */
   if( threadpool->jobqueue->njobs == 0 )
   {
      threadpool->jobqueue->firstjob = newjob;
      threadpool->jobqueue->lastjob = newjob;
   }
   else /* it is assumed that the jobqueue is not full */
   {
      threadpool->jobqueue->lastjob->nextjob = newjob;
      threadpool->jobqueue->lastjob = newjob;
   }

   /* signalling to all threads that the queue has jobs using the signal instead of broadcast because only one thread
    * should be awakened */
   SCIP_CALL_ABORT( SCIPtnySignalCondition(&(threadpool->queuenotempty)) );

   threadpool->jobqueue->njobs++;
}

/** adds a job to the threadpool */
static
SCIP_RETCODE threadPoolAddWork(
   SCIP_JOB*             newjob,             /**< job to add to threadpool */
   SCIP_SUBMITSTATUS*    status              /**< pointer to store the job's submit status */
   )
{
   assert(newjob != NULL);
   assert(_threadpool != NULL);

   SCIP_CALL( SCIPtnyAcquireLock(&(_threadpool->poollock)) );

   /* if the queue is full and we are blocking, then return an error. */
   if( _threadpool->jobqueue->njobs == _threadpool->queuesize && _threadpool->blockwhenfull )
   {
      SCIP_CALL( SCIPtnyReleaseLock(&(_threadpool->poollock)) );
      *status = SCIP_SUBMIT_QUEUEFULL;
      return SCIP_OKAY;
   }

   /* Wait until the job queue is not full. If the queue is closed or the thread pool is shut down, then stop waiting. */
   /* @todo this needs to be checked. It is possible that a job can be submitted and then the queue is closed or the
    * thread pool is shut down. Need to work out the best way to handle this. */
   while( _threadpool->jobqueue->njobs == _threadpool->queuesize && !(_threadpool->shutdown || !_threadpool->queueopen) )
   {
      SCIP_CALL( SCIPtnyWaitCondition(&(_threadpool->queuenotfull), &(_threadpool->poollock)) );
   }

   /* if the thread pool is shut down or the queue is closed, then we need to leave the job submission */
   if( !_threadpool->queueopen )
   {
      SCIP_CALL( SCIPtnyReleaseLock(&(_threadpool->poollock)) );
      *status = SCIP_SUBMIT_QUEUECLOSED;
      return SCIP_OKAY;
   }
   else if( _threadpool->shutdown )
   {
      SCIP_CALL( SCIPtnyReleaseLock(&(_threadpool->poollock)) );
      *status =  SCIP_SUBMIT_SHUTDOWN;
      return SCIP_OKAY;
   }

   /* creating the job for submission */
   newjob->nextjob = NULL;

   /* adding the job to the queue */
   /* this can only happen if the queue is not full */
   assert(_threadpool->jobqueue->njobs != _threadpool->queuesize);
   jobQueueAddJob(_threadpool, newjob);

   SCIP_CALL( SCIPtnyReleaseLock(&(_threadpool->poollock)) );

   *status = SCIP_SUBMIT_SUCCESS;

   return SCIP_OKAY;
}

/** frees the jobqueue of the threadpool */
static
void freeJobQueue(
   SCIP_THREADPOOL*      thrdpool            /**< pointer to thread pool */
   )
{
   SCIP_JOB* currjob;

   assert(!thrdpool->queueopen);
   assert(thrdpool->shutdown);

   /* iterating through all jobs until all have been freed */
   while( thrdpool->jobqueue->firstjob != NULL )
   {
      currjob = thrdpool->jobqueue->firstjob->nextjob;
      thrdpool->jobqueue->firstjob = thrdpool->jobqueue->firstjob->nextjob;
      BMSfreeMemory(&currjob);
   }

   assert(thrdpool->jobqueue->firstjob == NULL);
   assert(thrdpool->jobqueue->lastjob == NULL);

   BMSfreeMemory(&thrdpool->jobqueue);
}

/** free the thread pool */
static
SCIP_RETCODE freeThreadPool(
   SCIP_THREADPOOL**     thrdpool,           /**< pointer to thread pool */
   SCIP_Bool             finishjobs,         /**< currently unused */
   SCIP_Bool             completequeue       /**< Wait until the queue has complete? */
   )
{
   int          i;
   SCIP_RETCODE retcode;

   /*TODO remove argument? */
   SCIP_UNUSED( finishjobs );

   SCIP_CALL( SCIPtnyAcquireLock(&((*thrdpool)->poollock)) );

   /* if the shutdown is already in progress, then we don't need to complete this function */
   if( !(*thrdpool)->queueopen || (*thrdpool)->shutdown )
   {
      SCIP_CALL( SCIPtnyReleaseLock(&((*thrdpool)->poollock)) );

      return SCIP_OKAY;
   }

   /* indicating that the job queue is now closed for new jobs */
   (*thrdpool)->queueopen = FALSE;

   /* if the jobs in the queue should be completed, then we wait until the queueempty condition is set */
   if( completequeue )
   {
      while( (*thrdpool)->jobqueue->njobs > 0 )
      {
         SCIP_CALL( SCIPtnyWaitCondition(&((*thrdpool)->queueempty), &((*thrdpool)->poollock)) );
      }
   }

   /* indicating that the tpi has commenced the shutdown process */
   (*thrdpool)->shutdown = TRUE;

   SCIP_CALL( SCIPtnyReleaseLock(&((*thrdpool)->poollock)) );

   /* waking up all threads so that they can check the shutdown condition;
    * this requires that the conditions queuenotempty and queuenotfull is broadcast
    */
   SCIP_CALL( SCIPtnyBroadcastCondition(&((*thrdpool)->queuenotempty)) );
   SCIP_CALL( SCIPtnyBroadcastCondition(&((*thrdpool)->queuenotfull)) );

   retcode = SCIP_OKAY;

   /* calling a join to ensure that all worker finish before the thread pool is closed */
   for( i = 0; i < (*thrdpool)->nthreads; i++ )
   {
      int thrdretcode;

      if( thrd_join((*thrdpool)->threads[i], &thrdretcode) != thrd_success )
         retcode = (SCIP_RETCODE) MIN((int)SCIP_ERROR, (int)retcode);
      else
         retcode = (SCIP_RETCODE) MIN(thrdretcode, (int)retcode);
   }

   /* freeing memory and data structures */
   BMSfreeMemoryArray(&(*thrdpool)->threads);

   /* Freeing the current jobs list. This assumes that all jobs complete before the tpi is closed. */
   assert((*thrdpool)->currentjobs->njobs == 0);
   BMSfreeMemory(&(*thrdpool)->currentjobs);
   assert((*thrdpool)->finishedjobs->njobs == 0);
   BMSfreeMemory(&(*thrdpool)->finishedjobs);

   freeJobQueue(*thrdpool);

   /* destroying the conditions */
   SCIPtnyDestroyCondition(&(*thrdpool)->jobfinished);
   SCIPtnyDestroyCondition(&(*thrdpool)->queueempty);
   SCIPtnyDestroyCondition(&(*thrdpool)->queuenotfull);
   SCIPtnyDestroyCondition(&(*thrdpool)->queuenotempty);

   /* destroying the mutex */
   SCIPtnyDestroyLock(&(*thrdpool)->poollock);

   BMSfreeMemory(thrdpool);

   return retcode;
}


/* checking a job queue */
static
SCIP_JOBSTATUS checkJobQueue(
   SCIP_JOBQUEUE*        jobqueue,           /**< pointer to the job queue */
   int                   jobid               /**< id of job to check */
   )
{
   SCIP_JOB* currjob = jobqueue->firstjob;

   /* checking the job ids */
   if( currjob != NULL )
   {
      while( currjob != jobqueue->lastjob )
      {
         if( currjob->jobid == jobid )
            return SCIP_JOB_INQUEUE;

         currjob = currjob->nextjob;
      }

      if( currjob->jobid == jobid )
         return SCIP_JOB_INQUEUE;
   }

   return SCIP_JOB_DOESNOTEXIST;
}

/** returns whether the job id is running */
static
SCIP_Bool isJobRunning(
   SCIP_JOBQUEUE*        currentjobs,        /**< queue of current jobs */
   int                   jobid               /**< id of job to check */
   )
{
   if( checkJobQueue(currentjobs, jobid) == SCIP_JOB_INQUEUE )
      return TRUE;
   else
      return FALSE;
}

/** returns the number of threads */
int SCIPtpiGetNumThreads(
   void
   )
{
   return _threadpool->nthreads;
}

/** initializes tpi */
SCIP_RETCODE SCIPtpiInit(
   int                   nthreads,           /**< the number of threads to be used */
   int                   queuesize,          /**< the size of the queue */
   SCIP_Bool             blockwhenfull       /**< should the queue block when full */
   )
{
   assert(_threadpool == NULL);
   SCIP_CALL( createThreadPool(&_threadpool, nthreads, queuesize, blockwhenfull) );
   return SCIP_OKAY;
}

/** deinitializes tpi */
SCIP_RETCODE SCIPtpiExit(
   void
   )
{
   assert(_threadpool != NULL);

   SCIP_CALL( freeThreadPool(&_threadpool, TRUE, TRUE) );

   return SCIP_OKAY;
}

/** creates a job for parallel processing */
SCIP_RETCODE SCIPtpiCreateJob(
   SCIP_JOB**            job,                /**< pointer to the job that will be created */
   int                   jobid,              /**< the id for the current job */
   SCIP_RETCODE          (*jobfunc)(void* args),/**< pointer to the job function */
   void*                 jobarg              /**< the job's argument */
   )
{
   SCIP_ALLOC( BMSallocMemory(job) );

   (*job)->jobid = jobid;
   (*job)->jobfunc = jobfunc;
   (*job)->args = jobarg;
   (*job)->nextjob = NULL;

   return SCIP_OKAY;
}

/** get a new job id for the new set of submitted jobs */
int SCIPtpiGetNewJobID(
   void
   )
{
   int id;
   assert(_threadpool != NULL);

   SCIP_CALL_ABORT( SCIPtnyAcquireLock(&_threadpool->poollock) );
   id = ++_threadpool->currentid;
   SCIP_CALL_ABORT( SCIPtnyReleaseLock(&_threadpool->poollock) );

   return id;
}

/** submit a job for parallel processing; the return value is a globally defined status */
SCIP_RETCODE SCIPtpiSubmitJob(
   SCIP_JOB*             job,                /**< pointer to the job to be submitted */
   SCIP_SUBMITSTATUS*    status              /**< pointer to store the job's submit status */
   )
{
   assert(job != NULL);

   /* the job id must be set before submitting the job. The submitter controls whether a new id is required. */
   assert(job->jobid == _threadpool->currentid);
   SCIP_CALL( threadPoolAddWork(job, status) );

   return SCIP_OKAY;
}

/** blocks until all jobs of the given jobid have finished
 *  and then returns the smallest SCIP_RETCODE of all the jobs
 */
SCIP_RETCODE SCIPtpiCollectJobs(
   int                   jobid               /**< the jobid of the jobs to wait for */
   )
{
   SCIP_RETCODE retcode;
   SCIP_JOB* currjob;
   SCIP_JOB* prevjob;

   SCIP_CALL( SCIPtnyAcquireLock(&(_threadpool->poollock)) );

   while( isJobRunning(_threadpool->currentjobs, jobid) || isJobRunning(_threadpool->jobqueue, jobid) )
   {
      SCIP_CALL( SCIPtnyWaitCondition(&_threadpool->jobfinished, &_threadpool->poollock) );
   }

   /* finding the location of the processed job in the currentjobs queue */
   retcode = SCIP_OKAY;
   currjob = _threadpool->finishedjobs->firstjob;
   prevjob = NULL;

   while( currjob )
   {
      if( currjob->jobid == jobid )
      {
         SCIP_JOB* nextjob;

         /* if the job has the right jobid collect its retcode,
          * remove it from the finished job list, and free it
          */
         retcode = MIN(retcode, currjob->retcode);

         /* removing the finished job from finished jobs list */
         if( currjob == _threadpool->finishedjobs->firstjob )
         {
            _threadpool->finishedjobs->firstjob = currjob->nextjob;
         }
         else
         {
            assert(prevjob != NULL);
            prevjob->nextjob = currjob->nextjob;
         }

         if( currjob == _threadpool->finishedjobs->lastjob )
            _threadpool->finishedjobs->lastjob = prevjob;

         _threadpool->finishedjobs->njobs--;

         /* update currjob and free finished job; prevjob stays the same */
         nextjob = currjob->nextjob;
         BMSfreeMemory(&currjob);
         currjob = nextjob;
      }
      else
      {
         /* otherwise leave job untouched */
         prevjob = currjob;
         currjob = prevjob->nextjob;
      }
   }

   SCIP_CALL( SCIPtnyReleaseLock(&_threadpool->poollock) );

   return retcode;
}


/*
 * locks
 */

/** initializes the given lock */
SCIP_RETCODE SCIPtpiInitLock(
   SCIP_LOCK**           lock                /**< the lock */
   )
{
   assert(lock != NULL);

   SCIP_ALLOC( BMSallocMemory(lock) );

   if( mtx_init(&(*lock)->lock, mtx_plain) == thrd_success )
      return SCIP_OKAY;
   else
   {
      BMSfreeMemory(lock);
      return SCIP_ERROR;
   }
}

/** destroys the given lock */
void SCIPtpiDestroyLock(
   SCIP_LOCK**           lock                /**< the lock */
   )
{
   assert(lock != NULL);

   mtx_destroy(&(*lock)->lock);
   BMSfreeMemory(lock);
}

/** acquires the given lock */
SCIP_RETCODE SCIPtpiAcquireLock(
   SCIP_LOCK*            lock                /**< the lock */
   )
{
   if( mtx_lock(&lock->lock) == thrd_success )
      return SCIP_OKAY;
   return SCIP_ERROR;
}

/** releases the given lock */
SCIP_RETCODE SCIPtpiReleaseLock(
   SCIP_LOCK*            lock                /**< the lock */
   )
{
   if( mtx_unlock(&lock->lock) == thrd_success )
      return SCIP_OKAY;
   return SCIP_ERROR;
}


/*
 * conditions
 */

/** initializes the given condition variable */
SCIP_RETCODE SCIPtpiInitCondition(
   SCIP_CONDITION**      condition           /**< condition to be created and initialized */
   )
{
   assert(condition != NULL);

   SCIP_ALLOC( BMSallocMemory(condition) );

   if( cnd_init(&(*condition)->condition) == thrd_success )
      return SCIP_OKAY;
   return SCIP_ERROR;
}

/** destroys the given condition variable */
void SCIPtpiDestroyCondition(
   SCIP_CONDITION**      condition           /**< condition to be destroyed and freed */
   )
{
   cnd_destroy(&(*condition)->condition);
   BMSfreeMemory(condition);
}

/** signals one waiting thread */
SCIP_RETCODE SCIPtpiSignalCondition(
   SCIP_CONDITION*       condition           /**< the condition variable to signal */
   )
{
   if( cnd_signal(&condition->condition) == thrd_success )
      return SCIP_OKAY;
   return SCIP_ERROR;
}

/** signals all waiting threads */
SCIP_EXPORT
SCIP_RETCODE SCIPtpiBroadcastCondition(
   SCIP_CONDITION*       condition           /**< the condition variable to broadcast */
   )
{
   if( cnd_broadcast(&condition->condition) == thrd_success )
      return SCIP_OKAY;
   return SCIP_ERROR;
}

/** waits on a condition variable. The given lock must be held by the caller and will
 *  be held when this function returns.
 */
SCIP_RETCODE SCIPtpiWaitCondition(
   SCIP_CONDITION*       condition,          /**< the condition variable to wait on */
   SCIP_LOCK*            lock                /**< the lock that is held by the caller */
   )
{
   if( cnd_wait(&condition->condition, &lock->lock) == thrd_success )
      return SCIP_OKAY;
   return SCIP_ERROR;
}

/** returns the thread number */
int SCIPtpiGetThreadNum(
   void
   )
{
   return _threadnumber;
}