bdlmt_threadmultiplexor.h

Go to the documentation of this file.

/// @file bdlmt_threadmultiplexor.h
///
/// The content of this file has been pre-processed for Doxygen.
///
 
 
// bdlmt_threadmultiplexor.h                                          -*-C++-*-
#ifndef INCLUDED_BDLMT_THREADMULTIPLEXOR
#define INCLUDED_BDLMT_THREADMULTIPLEXOR
 
#include <bsls_ident.h>
BSLS_IDENT("$Id: $")
 
/// @defgroup bdlmt_threadmultiplexor bdlmt_threadmultiplexor
/// @brief  Provide a mechanism for partitioning a collection of threads.
/// @addtogroup bdl
/// @{
/// @addtogroup bdlmt
/// @{
/// @addtogroup bdlmt_threadmultiplexor
/// @{
///
/// <h1> Outline </h1>
/// * <a href="#bdlmt_threadmultiplexor-purpose"> Purpose</a>
/// * <a href="#bdlmt_threadmultiplexor-classes"> Classes </a>
/// * <a href="#bdlmt_threadmultiplexor-description"> Description </a>
///   * <a href="#bdlmt_threadmultiplexor-thread-safety"> Thread Safety </a>
///   * <a href="#bdlmt_threadmultiplexor-order-of-execution"> Order of Execution </a>
///   * <a href="#bdlmt_threadmultiplexor-usage"> Usage </a>
///     * <a href="#bdlmt_threadmultiplexor-example-1-multiple-work-queues"> Example 1: Multiple Work Queues </a>
///
/// # Purpose {#bdlmt_threadmultiplexor-purpose}
/// Provide a mechanism for partitioning a collection of threads.
///
/// # Classes {#bdlmt_threadmultiplexor-classes}
///
/// - bdlmt::ThreadMultiplexor: mechanism to partition multi-threaded processing
///
/// @see  bdlmt_threadpool, bdlmt_fixedthreadpool
///
/// # Description {#bdlmt_threadmultiplexor-description}
/// This component provides a mechanism for partitioning a
/// collection of threads, so that a single collection of threads (e.g., a
/// thread pool) can be used to process various types of user-defined functions
/// ("jobs") while sharing the thread resources equitably between them.  A
/// typical example where this type of partitioning is desired is an application
/// that performs both I/O and CPU-intensive processing.  The traditional
/// approach is to create two thread pools--one for I/O, and one for
/// processing--and pass control (in the form of a callback) from one thread
/// pool to the other.  However, there are several problems with this approach.
/// Firstly, the process incurs the overhead of context switching between
/// threads, which must necessarily occur because there are two different thread
/// pools.  Secondly, the process may not be able to adapt well to imbalances
/// between one type of processing versus the other if the number of threads in
/// each thread pool is bounded.  In this case, a large number of jobs may be
/// enqueued while some portion of threads allocated to the process go unused.
/// On the other hand, simply sharing a single thread pool without a provision
/// for partitioning the use of threads may result in one type of processing
/// starving the other.
///
/// The `bdlmt::ThreadMultiplexor` provides an API, `processJob`, to process
/// user-specified jobs.  A multiplexor instance is configured with a maximum
/// number of "processors", i.e., the maximum number of threads that may process
/// jobs at any particular time.  Additional threads enqueue jobs to a pending
/// job queue, which is processed by the next available processing thread.
///
/// Typically, a `bdlmt::ThreadMultiplexor` instance is used in conjunction with
/// a thread pool (e.g., `bdlmt::FixedThreadPool`), where each thread pool
/// thread calls the multiplexor `processJob` method to perform some work.  The
/// multiplexor guarantees that no more that the configured number of threads
/// will process jobs concurrently.  This guarantee allows a single thread pool
/// to be used in a variety of situations that require partitioning thread
/// resources.
///
/// ## Thread Safety {#bdlmt_threadmultiplexor-thread-safety}
///
///
/// The `bdlmt::ThreadMultiplexor` class is both **fully thread-safe** (i.e.,
/// all non-creator methods can correctly execute concurrently), and is
/// **thread-enabled** (i.e., the class does not function correctly in a
/// non-multi-threading environment).  See @ref bsldoc_glossary  for complete
/// definitions of **fully thread-safe** and **thread-enabled**.
///
/// ## Order of Execution {#bdlmt_threadmultiplexor-order-of-execution}
///
///
/// 'bdlmt::ThreadMultiplexor' does not guarantee that jobs in the pending job
/// queue will be processed in the order in which they were enqueued.
///
/// ## Usage {#bdlmt_threadmultiplexor-usage}
///
///
/// This section illustrates intended use of this component.
///
/// ### Example 1: Multiple Work Queues {#bdlmt_threadmultiplexor-example-1-multiple-work-queues}
///
///
/// The following usage example illustrates how the `bdlmt::ThreadMultiplexor`
/// can be used to share thread resources between three separate work queues.
/// Assume that there are three classes of jobs: jobs that are important, jobs
/// that are urgent, and jobs that are critical.  We would like to execute each
/// class of jobs in a single thread pool, but we want ensure that all types of
/// jobs can be executed at any time.
///
/// We begin by defining a class that encapsulates the notion of a job queue.
/// Our `JobQueue` class holds a reference to a `bdlmt::FixedThreadPool`, used
/// to instantiate the job queue, and owns an instance of
/// `bdlmt::ThreadMultiplexor`, used to process jobs.
/// @code
/// /// This class defines a generic processor for user-defined functions
/// /// ("jobs").  Jobs specified to the `processJob` method are executed
/// /// in the thread pool specified at construction.
/// class JobQueue {
///
///   public:
///     // PUBLIC TYPES
///
///     /// A callback of this type my be specified to the `processJob` method.
///     typedef bdlmt::ThreadMultiplexor::Job Job;
///
///   private:
///     // DATA
///     bdlmt::FixedThreadPool   *d_threadPool_p;  // (held, not owned)
///     bdlmt::ThreadMultiplexor  d_multiplexor;   // used to partition threads
///
///   private:
///     // NOT IMPLEMENTED
///     JobQueue(const JobQueue&);
///     JobQueue& operator=(const JobQueue&);
///
///   public:
///     // CREATORS
///
///     /// Create a job queue that executes jobs in the specified
///     /// 'threadPool' using no more than the specified 'maxProcessors'.
///     /// Optionally specify a 'basicAllocator' used to supply memory.  If
///     /// 'basicAllocator' is 0, the currently installed default allocator
///     /// is used.
///     JobQueue(int                     maxProcessors,
///              bdlmt::FixedThreadPool *threadPool,
///              bslma::Allocator       *basicAllocator = 0);
///
///     /// Destroy this object.
///     ~JobQueue();
///
///     // MANIPULATORS
///
///     /// Process the specified `job` in the thread pool specified at
///     /// construction.  Return 0 on success, and a non-zero value otherwise.
///     int processJob(const Job& job);
/// };
/// @endcode
/// The maximum number of processors for the multiplexor instance owned by each
/// `JobQueue` is configured using the following formula, for
/// T = number of threads and M = number of multiplexors > 1:
/// @code
///  maxProc = ceil(T / (M-1))-1
/// @endcode
/// This allows multiple `JobQueue` instances to share the same threadpool
/// without starving each other when the thread pool has more than one thread.
/// For this usage example, we assume M (number of multiplexors) = 3, and T
/// (number of threads) = 5, so maxProc = 2.  It is important to note that every
/// call to `processJob` enqueues a job to the thread pool, so the length of the
/// thread pool queue determines the maximum number of jobs that can be accepted
/// by the JobQueue.  (Multiple JobQueues share the same maximum *together*, so
/// not all will be able to reach their individual maximum at the same time).
/// @code
///
///    JobQueue::JobQueue(int                     maxProcessors,
///                       bdlmt::FixedThreadPool *threadPool,
///                       bslma::Allocator       *basicAllocator)
///    : d_threadPool_p(threadPool)
///    , d_multiplexor (maxProcessors,
///                     threadPool->queueCapacity(),
///                     basicAllocator)
///   {
///   }
///
///   JobQueue::~JobQueue()
///   {
///   }
/// @endcode
/// The `processJob` method enqueues a secondary callback into the thread pool
/// that executes the user-specified `job` through the multiplexor.
/// @code
/// int JobQueue::processJob(const JobQueue::Job& job)
/// {
///     return d_threadPool_p->tryEnqueueJob(bdlf::BindUtil::bind(
///                                 &bdlmt::ThreadMultiplexor::processJob<Job>,
///                                 &d_multiplexor,
///                                 job));
/// }
/// @endcode
/// The following program uses three instances of `JobQueue` to process
/// important, urgent, and critical jobs using a single collection of threads.
/// @code
/// int main(void)
/// {
///     enum {
///         NUM_THREADS   = 5,   // total number of threads
///         NUM_QUEUES    = 3,   // total number of JobQueue objects
///         MAX_QUEUESIZE = 20   // total number of pending jobs
///     };
///
///     int maxProc = bsl::max(1,
///                            ceil(double(NUM_THREADS) / (NUM_QUEUES-1))-1);
///
///     bdlmt::FixedThreadPool tp(NUM_THREADS, MAX_QUEUESIZE);
///     JobQueue             importantQueue(maxProc, &tp);
///     JobQueue             urgentQueue(maxProc, &tp);
///     JobQueue             criticalQueue(maxProc, &tp);
///
///     if (0 != tp.start()) {
///        ASSERT(0 == "Could not start thread pool!");
///        return -1;
///     }
///
///     JobQueue::Job ijob =
///        bdlf::BindUtil::bind(&bsls::AtomicInt::add, &iCheck, 1);
///
///     JobQueue::Job ujob = bdlf::BindUtil::bind(
///        bdlf::BindUtil::bind(&bsls::AtomicInt::add, &uCheck, 1);
///
///     JobQueue::Job cjob = bdlf::BindUtil::bind(
///        bdlf::BindUtil::bind(&bsls::AtomicInt::add, &cCheck, 1);
///
///     importantQueue.processJob(ijob);
///     importantQueue.processJob(ijob);
///     importantQueue.processJob(ijob);
///     importantQueue.processJob(ijob);
///     importantQueue.processJob(ijob);
///     importantQueue.processJob(ijob);
///
///     urgentQueue.processJob(ujob);
///     urgentQueue.processJob(ujob);
///     urgentQueue.processJob(ujob);
///     urgentQueue.processJob(ujob);
///
///     criticalQueue.processJob(cjob);
///     criticalQueue.processJob(cjob);
///
///     tp.stop();
///     ASSERT(6 == iCheck);
///     ASSERT(4 == uCheck);
///     ASSERT(2 == cCheck);
///     return 0;
/// }
/// @endcode
/// @}
/** @} */
/** @} */
 
/** @addtogroup bdl
 * @{
 */
/** @addtogroup bdlmt
 * @{
 */
/** @addtogroup bdlmt_threadmultiplexor
 * @{
 */
 
#include <bdlscm_version.h>
 
#include <bdlcc_fixedqueue.h>
 
#include <bsls_atomic.h>
 
#include <bslma_allocator.h>
#include <bslma_usesbslmaallocator.h>
 
#include <bslmf_nestedtraitdeclaration.h>
 
#include <bsl_functional.h>
 
#ifndef BDE_DONT_ALLOW_TRANSITIVE_INCLUDES
#include <bslalg_typetraits.h>
#endif // BDE_DONT_ALLOW_TRANSITIVE_INCLUDES
 
 
 
namespace bdlmt {
                          // =======================
                          // class ThreadMultiplexor
                          // =======================
 
/// This class provides a mechanism for facilitating the use of multiple
/// threads to perform various user-defined functions ("jobs") when some
/// degree of collaboration between threads is required.  The thread
/// multiplexor is configured with a total number of "processors",
/// representing the number of threads that may process jobs at any
/// particular time.  Additional threads enqueue jobs to a pending job
/// queue, which is processed by the next available processing thread.
///
/// See @ref bdlmt_threadmultiplexor 
class ThreadMultiplexor {
 
  public:
    // PUBLIC TYPES
 
    /// A callback of this type may be passed to the `processJob` method.
    typedef bsl::function<void()> Job;
 
  private:
    // DATA
    bslma::Allocator       *d_allocator_p;      // memory allocator (held)
    bdlcc::FixedQueue<Job> *d_jobQueue_p;       // pending job queue (owned)
    bsls::AtomicInt         d_numProcessors;    // current number of processors
    int                     d_maxProcessors;    // maximum number of processors
 
  private:
    // PRIVATE MANIPULATORS
 
    /// Process the pending job queue.  Execute each functor obtained from
    /// the queue in the calling thread if the current number of processors
    /// is less than the maximum number of processors.  Otherwise, enqueue
    /// the job back to the pending queue.  Return 0 on success, and a
    /// non-zero value otherwise.
    int processJobQueue();
 
  private:
    // NOT IMPLEMENTED
    ThreadMultiplexor(const ThreadMultiplexor&);
    ThreadMultiplexor& operator=(const ThreadMultiplexor&);
 
  public:
    // TRAITS
    BSLMF_NESTED_TRAIT_DECLARATION(ThreadMultiplexor,
                                   bslma::UsesBslmaAllocator);
 
    // CREATORS
 
    /// Create a thread multiplexor which uses, at most, the specified
    /// `maxProcessors` number of threads to process user-specified jobs,
    /// identified as callbacks of type `Job`.  Jobs that cannot be processed
    /// immediately are placed on a queue having the specified `maxQueueSize`
    /// to be processed by the next free thread.  Optionally specify a
    /// `basicAllocator` used to supply memory.  If `basicAllocator` is 0, the
    /// currently installed default allocator is used.  
    /// 
    /// \pre The behavior is undefined   unless `0 < maxProcessors` and `0 < maxQueueSize`.
    ThreadMultiplexor(int               maxProcessors,
                      int               maxQueueSize,
                      bslma::Allocator *basicAllocator = 0);
 
    /// Destroy this thread multiplexor object.
    ~ThreadMultiplexor();
 
    // MANIPULATORS
 
    /// Process the specified `job` functor in the calling thread if the
    /// current number of processors is less than the maximum number of
    /// processors.  Otherwise, enqueue `job` to the pending job queue.  Return 0 on success, and a non-zero value otherwise.  
    /// 
    /// \note Note that   the only
    /// requirements on `t_JOBTYPE` are that it defines `operator()`, having a
    /// `void` return type, and that it defines a copy constructor.
    template <class t_JOBTYPE>
    int processJob(const t_JOBTYPE& job);
 
    // ACCESSORS
 
    /// Return the maximum number of active processors.
    int maxProcessors() const;
 
    /// Return the current number of active processors.
    int numProcessors() const;
};
 
// ============================================================================
//                            INLINE DEFINITIONS
// ============================================================================
 
// MANIPULATORS
template <class t_JOBTYPE>
inline
int ThreadMultiplexor::processJob(const t_JOBTYPE& job)
{
    // Execute 'job' in the calling thread if the current number of processors
    // is less than the maximum number of processors.  Otherwise, enqueue 'job'
    // to the pending job queue.  In all cases, check the pending job queue at
    // the end of the loop, as the number of processors may have changed,
    // allowing the execution of a job in the current thread.
 
    int previousNumProcessors = d_numProcessors;
    if (previousNumProcessors < d_maxProcessors &&
        previousNumProcessors ==
                      d_numProcessors.testAndSwap(previousNumProcessors,
                                                  previousNumProcessors + 1)) {
        // Process the job
        job();
        --d_numProcessors;
    }
    else {
        int rc = d_jobQueue_p->tryPushBack(job);
        if (0 != rc) {
            return rc;                                                // RETURN
        }
    }
 
    return processJobQueue();
}
 
// ACCESSORS
inline
int ThreadMultiplexor::maxProcessors() const
{
    return d_maxProcessors;
}
 
inline
int ThreadMultiplexor::numProcessors() const
{
    return d_numProcessors;
}
 
}  // close package namespace
 
#endif
 
// ----------------------------------------------------------------------------
// Copyright 2015 Bloomberg Finance L.P.
//
// 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.
// ----------------------------- END-OF-FILE ----------------------------------
 
 
/** @} */
/** @} */
/** @} */