tpi_openmp.c

Go to the documentation of this file.

/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
/*                                                                           */
/*                  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_openmp.c
 * @ingroup TASKINTERFACE
 * @brief  the interface functions for openmp
 * @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 "scip/pub_message.h"
#include <omp.h>
 
/* macros for direct access */
 
/* lock */
#define SCIPompInitLock(lock)     (omp_init_lock(lock), SCIP_OKAY)
#define SCIPompDestroyLock(lock)  (omp_destroy_lock(lock))
#define SCIPompAcquireLock(lock)  (omp_set_lock(lock), SCIP_OKAY)
#define SCIPompReleaseLock(lock)  (omp_unset_lock(lock), SCIP_OKAY)
 
/* condition */
#define SCIPompInitCondition(condition)    ( omp_init_lock(&(condition)->_lock), \
                                             (condition)->_waiters = 0, (condition)->_waitnum = 0, (condition)->_signals = 0, SCIP_OKAY )
#define SCIPompDestroyCondition(condition) do { assert((condition)->_waiters == 0); assert((condition)->_waitnum == 0); assert((condition)->_signals == 0); omp_destroy_lock(&(condition)->_lock); } while(0)
 
 
/** struct containing lock */
struct SCIP_Lock
{
   omp_lock_t            lock;
};
 
/** struct for condition */
struct SCIP_Condition
{
   omp_lock_t            _lock;
   int                   _waiters;
   int                   _waitnum;
   int                   _signals;
};
 
 
/** 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;
 
struct SCIP_JobQueues
{
   SCIP_JOBQUEUE         jobqueue;           /**< queue of unprocessed jobs */
   SCIP_JOB**            currentjobs;        /**< array with slot for each thread to store the currently running job */
   int                   ncurrentjobs;       /**< number of currently running jobs */
   int                   nthreads;           /**< number of threads */
   SCIP_JOBQUEUE         finishedjobs;       /**< jobqueue containing the finished jobs */
   omp_lock_t            lock;               /**< lock to protect this stucture from concurrent access */
   SCIP_CONDITION        jobfinished;        /**< condition to signal if a job was finished */
};
typedef struct SCIP_JobQueues SCIP_JOBQUEUES;
 
static SCIP_JOBQUEUES* _jobqueues = NULL;
 
 
/** create job queue */
static
SCIP_RETCODE createJobQueue(
   int                   nthreads,           /**< the number of threads */
   int                   qsize,              /**< the queue size */
   SCIP_Bool             blockwhenfull       /**< should the queue be blocked from new jobs when full */
   )
{
   int i;
 
   assert(nthreads >= 0);
   assert(qsize >= 0);
   SCIP_UNUSED( blockwhenfull );
 
   /* allocting memory for the job queue */
   SCIP_ALLOC( BMSallocMemory(&_jobqueues) );
   _jobqueues->jobqueue.firstjob = NULL;
   _jobqueues->jobqueue.lastjob = NULL;
   _jobqueues->jobqueue.njobs = 0;
   _jobqueues->finishedjobs.firstjob = NULL;
   _jobqueues->finishedjobs.lastjob = NULL;
   _jobqueues->finishedjobs.njobs = 0;
   _jobqueues->ncurrentjobs = 0;
 
   _jobqueues->nthreads = nthreads;
   SCIP_ALLOC( BMSallocMemoryArray(&_jobqueues->currentjobs, nthreads) );
 
   for( i = 0; i < nthreads; ++i )
      _jobqueues->currentjobs[i] = NULL;
 
   SCIP_CALL( SCIPompInitLock(&_jobqueues->lock) );
   SCIP_CALL( SCIPompInitCondition(&_jobqueues->jobfinished) );
 
   return SCIP_OKAY;
}
 
 
/** free job queue */
static
SCIP_RETCODE freeJobQueue(
   void
   )
{
   assert(_jobqueues != NULL);
 
   SCIPompDestroyLock(&_jobqueues->lock);
   SCIPompDestroyCondition(&_jobqueues->jobfinished);
   BMSfreeMemoryArray(&_jobqueues->currentjobs);
 
   BMSfreeMemory(&_jobqueues);
 
   return SCIP_OKAY;
}
 
 
/** execute job */
static
void executeJob(
   SCIP_JOB*             job                 /**< the job to be executed in parallel */
   )
{
   int threadnum;
 
   threadnum = SCIPtpiGetThreadNum();
 
   SCIP_CALL_ABORT( SCIPompAcquireLock(&_jobqueues->lock) );
   _jobqueues->currentjobs[threadnum] = job;
   SCIP_CALL_ABORT( SCIPompReleaseLock(&_jobqueues->lock) );
 
   job->retcode = (*(job->jobfunc))(job->args);
 
   SCIP_CALL_ABORT( SCIPompAcquireLock(&_jobqueues->lock) );
   _jobqueues->ncurrentjobs--;
   _jobqueues->currentjobs[threadnum] = NULL;
 
   /* insert job into finished jobs */
   if( _jobqueues->finishedjobs.njobs == 0 )
   {
      _jobqueues->finishedjobs.firstjob = job;
      _jobqueues->finishedjobs.lastjob = job;
   }
   else
   {
      _jobqueues->finishedjobs.lastjob->nextjob = job;
      _jobqueues->finishedjobs.lastjob = job;
   }
 
   ++_jobqueues->finishedjobs.njobs;
 
   SCIP_CALL_ABORT( SCIPtpiBroadcastCondition(&_jobqueues->jobfinished) );
 
   SCIP_CALL_ABORT( SCIPompReleaseLock(&_jobqueues->lock) );
}
 
/** wait for a condition */
SCIP_RETCODE SCIPtpiWaitCondition(
   SCIP_CONDITION*       condition,          /**< condition to wait for */
   SCIP_LOCK*            lock                /**< corresponding lock */
   )
{
   int waitnum;
 
   SCIP_CALL( SCIPtpiReleaseLock(lock) );
 
   SCIP_CALL( SCIPompAcquireLock(&condition->_lock) );
   waitnum = ++condition->_waitnum;
 
   ++condition->_waiters;
 
   do
   {
      SCIP_CALL( SCIPompReleaseLock(&condition->_lock) );
      #pragma omp taskyield
      SCIP_CALL( SCIPompAcquireLock(&condition->_lock) );
   }
   while( condition->_signals < waitnum );
 
   --condition->_waiters;
 
   if( condition->_waiters == 0 )
   {
      condition->_signals = 0;
      condition->_waitnum = 0;
   }
 
   SCIP_CALL( SCIPompReleaseLock(&condition->_lock) );
 
   SCIP_CALL( SCIPtpiAcquireLock(lock) );
 
   return SCIP_OKAY;
}
 
/** wait for a condition (direct access to lock) */
static
SCIP_RETCODE SCIPompWaitCondition(
   SCIP_CONDITION*       condition,          /**< condition to wait for */
   omp_lock_t*           lock                /**< corresponding lock */
   )
{
   int waitnum;
 
   SCIP_CALL( SCIPompReleaseLock(lock) );
 
   SCIP_CALL( SCIPompAcquireLock(&condition->_lock) );
   waitnum = ++condition->_waitnum;
 
   ++condition->_waiters;
 
   do
   {
      SCIP_CALL( SCIPompReleaseLock(&condition->_lock) );
      #pragma omp taskyield
      SCIP_CALL( SCIPompAcquireLock(&condition->_lock) );
   }
   while( condition->_signals < waitnum );
 
   --condition->_waiters;
 
   if( condition->_waiters == 0 )
   {
      condition->_signals = 0;
      condition->_waitnum = 0;
   }
 
   SCIP_CALL( SCIPompReleaseLock(&condition->_lock) );
 
   SCIP_CALL( SCIPompAcquireLock(lock) );
 
   return SCIP_OKAY;
}
 
 
/** process jobs from job queue
 *
 * The job will only be added when the number of active jobs is equal to the number of threads.
 * As such, there will always be number of threads + 1 tasks available for the scheduler to run.
 */
static
void jobQueueProcessJob(
   void
   )
{
   SCIP_JOB* job;
 
   SCIP_CALL_ABORT( SCIPompAcquireLock(&_jobqueues->lock) );
 
   while( _jobqueues->ncurrentjobs == SCIPtpiGetNumThreads() )
   {
      SCIP_CALL_ABORT( SCIPompWaitCondition(&_jobqueues->jobfinished, &_jobqueues->lock) );
   }
 
   if( _jobqueues->jobqueue.njobs == 1 )
   {
      job = _jobqueues->jobqueue.firstjob;
      _jobqueues->jobqueue.firstjob = NULL;
      _jobqueues->jobqueue.lastjob = NULL;
      --(_jobqueues->jobqueue.njobs);
   }
   else if( _jobqueues->jobqueue.njobs > 1 )
   {
      job = _jobqueues->jobqueue.firstjob;
      _jobqueues->jobqueue.firstjob = job->nextjob;
      --_jobqueues->jobqueue.njobs;
   }
   else
   {
      job = NULL;
   }
 
   ++(_jobqueues->ncurrentjobs);
   SCIP_CALL_ABORT( SCIPompReleaseLock(&_jobqueues->lock) );
 
   if( job )
   {
      executeJob(job);
   }
}
 
 
/** adding a job to the job queue
 *
 * This gives some more flexibility in the handling of new jobs.
 * IMPORTANT: This function MUST be called from within a mutex.
 */
static
SCIP_RETCODE jobQueueAddJob(
   SCIP_JOB*             newjob
   )
{
   /* @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(newjob != NULL);
 
   newjob->nextjob = NULL;
 
   /* This function queries the current job list. This could change by other threads writing to the list. So a lock is
    * required to ensure that the current joblist remains static. */
   SCIP_CALL( SCIPompAcquireLock(&_jobqueues->lock) );
 
   /* checking the status of the job queue */
   if( _jobqueues->ncurrentjobs == SCIPtpiGetNumThreads() )
   {
      if( _jobqueues->jobqueue.njobs == 0 )
      {
         _jobqueues->jobqueue.firstjob = newjob;
         _jobqueues->jobqueue.lastjob = newjob;
      }
      else /* it is assumed that the jobqueue is not full */
      {
         _jobqueues->jobqueue.lastjob->nextjob = newjob;
         _jobqueues->jobqueue.lastjob = newjob;
      }
 
      _jobqueues->jobqueue.njobs++;
 
      SCIP_CALL( SCIPompReleaseLock(&_jobqueues->lock) );
 
      #pragma omp task
      jobQueueProcessJob();
   }
   else
   {
      assert(_jobqueues->ncurrentjobs < SCIPtpiGetNumThreads());
 
      _jobqueues->ncurrentjobs++;
 
      SCIP_CALL( SCIPompReleaseLock(&_jobqueues->lock) );
      /* running the new job */
      #pragma omp task firstprivate(newjob)
      executeJob(newjob);
   }
 
   return SCIP_OKAY;
}
 
 
/** signal a condition */
SCIP_RETCODE SCIPtpiSignalCondition(
   SCIP_CONDITION*       condition           /**< condition to signal */
   )
{
   assert( condition != NULL );
 
   SCIP_CALL( SCIPompAcquireLock(&condition->_lock) );
 
   if( condition->_waitnum > condition->_signals )
      ++condition->_signals;
 
   SCIP_CALL( SCIPompReleaseLock(&condition->_lock) );
 
   return SCIP_OKAY;
}
 
 
/** broadcase a condition */
SCIP_RETCODE SCIPtpiBroadcastCondition(
   SCIP_CONDITION*       condition           /**< broadcast a condition */
   )
{
   assert( condition != NULL );
 
   SCIP_CALL( SCIPompAcquireLock(&condition->_lock) );
   condition->_signals = condition->_waitnum;
   SCIP_CALL( SCIPompReleaseLock(&condition->_lock) );
 
   return SCIP_OKAY;
}
 
 
 
/** returns the number of threads */
int SCIPtpiGetNumThreads(
   )
{
   return omp_get_num_threads();
}
 
/** returns the thread number */
int SCIPtpiGetThreadNum(
   )
{
   return omp_get_thread_num();
}
 
/** 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
   )
{
   static int currentjobid = 0;
   int jobid;
 
   #pragma omp atomic capture
   jobid = ++currentjobid;
 
   return jobid;
}
 
/** 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 submit status */
   )
{
   assert(_jobqueues != NULL);
 
   *status = SCIP_SUBMIT_SUCCESS;
   SCIP_CALL( jobQueueAddJob(job) );
 
   return SCIP_OKAY;
}
 
 
/** check whether a job is running */
static
SCIP_Bool isJobRunning(
   int                   jobid               /**< job id to check */
   )
{
   int i;
 
   if( _jobqueues->ncurrentjobs > 0 )
   {
      for( i = 0; i < _jobqueues->nthreads; ++i )
      {
         if( _jobqueues->currentjobs[i] != NULL && _jobqueues->currentjobs[i]->jobid == jobid )
            return TRUE;
      }
   }
 
   return FALSE;
}
 
 
/** check whether a job is waiting */
static
SCIP_Bool isJobWaiting(
   int                   jobid               /**< job id to check */
   )
{
   if( _jobqueues->jobqueue.njobs > 0 )
   {
      SCIP_JOB* currjob;
      currjob = _jobqueues->jobqueue.firstjob;
 
      do
      {
         if( currjob->jobid == jobid )
            return TRUE;
 
         if( currjob == _jobqueues->jobqueue.lastjob )
            break;
 
         currjob = currjob->nextjob;
      }
      while( TRUE ); /*lint !e506*/
   }
 
   return FALSE;
}
 
 
/** 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;
 
   retcode = SCIP_OKAY;
   SCIP_CALL( SCIPompAcquireLock(&_jobqueues->lock) );
 
   while( isJobRunning(jobid) || isJobWaiting(jobid) )
   {
      SCIP_CALL( SCIPompWaitCondition(&_jobqueues->jobfinished, &_jobqueues->lock) );
   }
 
   if( _jobqueues->finishedjobs.njobs > 0 )
   {
      SCIP_JOB* currjob = _jobqueues->finishedjobs.firstjob;
      SCIP_JOB* prevjob = NULL;
 
      /* finding the location of the processed job in the currentjobs queue */
      do
      {
         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 == _jobqueues->finishedjobs.firstjob )
               _jobqueues->finishedjobs.firstjob = currjob->nextjob;
            else
            {
               if( prevjob != NULL )
                  prevjob->nextjob = currjob->nextjob; /*lint !e613*/
            }
 
            if( currjob == _jobqueues->finishedjobs.lastjob )
               _jobqueues->finishedjobs.lastjob = prevjob;
 
            _jobqueues->finishedjobs.njobs--;
 
            /* update currjob and free finished job; prevjob stays the same */
            nextjob = currjob->nextjob;
            BMSfreeMemory(&currjob);
            currjob = nextjob;
         }
         else
         {
            prevjob = currjob;
            currjob = prevjob->nextjob;
         }
      }
      while( prevjob != _jobqueues->finishedjobs.lastjob );
   }
   else
   {
      /* given jobid was not submitted */
      printf("err1");
      retcode = SCIP_ERROR;
   }
 
   SCIP_CALL_ABORT( SCIPompReleaseLock(&_jobqueues->lock) );
 
   return retcode;
}
 
/** 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 */
   )
{
   omp_set_num_threads(nthreads);
   assert(_jobqueues == NULL);
 
   SCIP_CALL( createJobQueue(nthreads, queuesize, blockwhenfull) );
 
   return SCIP_OKAY;
}
 
/** deinitializes tpi */
SCIP_RETCODE SCIPtpiExit(
   void
   )
{
   assert(_jobqueues != NULL);
   assert(_jobqueues->finishedjobs.njobs == 0);
   assert(_jobqueues->jobqueue.njobs == 0);
   assert(_jobqueues->ncurrentjobs == 0);
 
   SCIP_CALL( freeJobQueue() );
 
   return SCIP_OKAY;
}
 
 
/*
 * locks
 */
 
/** initializes the given lock */
SCIP_RETCODE SCIPtpiInitLock(
   SCIP_LOCK**           lock                /**< the lock */
   )
{
   assert(lock != NULL);
 
   SCIP_ALLOC( BMSallocMemory(lock) );
   omp_init_lock(&(*lock)->lock);
   return SCIP_OKAY;
}
 
/** destroys the given lock */
void SCIPtpiDestroyLock(
   SCIP_LOCK**           lock                /**< the lock */
   )
{
   assert(lock != NULL);
 
   omp_destroy_lock(&(*lock)->lock);
   BMSfreeMemory(lock);
}
 
/** acquires the given lock */
SCIP_RETCODE SCIPtpiAcquireLock(
   SCIP_LOCK*            lock                /**< the lock */
   )
{
   omp_set_lock(&lock->lock);
   return SCIP_OKAY;
}
 
/** releases the given lock */
SCIP_RETCODE SCIPtpiReleaseLock(
   SCIP_LOCK*            lock                /**< the lock */
   )
{
   omp_unset_lock(&lock->lock);
   return SCIP_OKAY;
}
 
 
/*
 * 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) );
 
   omp_init_lock(&(*condition)->_lock);
   (*condition)->_waiters = 0;
   (*condition)->_waitnum = 0;
   (*condition)->_signals = 0;
 
   return SCIP_OKAY;
}
 
/** destroys the given condition variable */
void SCIPtpiDestroyCondition(
   SCIP_CONDITION**      condition           /**< condition to be destroyed and freed */
   )
{
   assert((*condition)->_waiters == 0);
   assert((*condition)->_waitnum == 0);
   assert((*condition)->_signals == 0);
 
   omp_destroy_lock(&(*condition)->_lock);
 
   BMSfreeMemory(condition);
}