BDE 4.39.x Production Release
Loading...
Searching...
No Matches
bdlmt_fixedthreadpool.h
Go to the documentation of this file.
1/// @file bdlmt_fixedthreadpool.h
2///
3/// The content of this file has been pre-processed for Doxygen.
4///
5
6
7// bdlmt_fixedthreadpool.h -*-C++-*-
8#ifndef INCLUDED_BDLMT_FIXEDTHREADPOOL
9#define INCLUDED_BDLMT_FIXEDTHREADPOOL
10
11#include <bsls_ident.h>
12BSLS_IDENT("$Id: $")
13
14/// @defgroup bdlmt_fixedthreadpool bdlmt_fixedthreadpool
15/// @brief Provide portable implementation for a fixed-size pool of threads.
16/// @addtogroup bdl
17/// @{
18/// @addtogroup bdlmt
19/// @{
20/// @addtogroup bdlmt_fixedthreadpool
21/// @{
22///
23/// <h1> Outline </h1>
24/// * <a href="#bdlmt_fixedthreadpool-purpose"> Purpose</a>
25/// * <a href="#bdlmt_fixedthreadpool-classes"> Classes </a>
26/// * <a href="#bdlmt_fixedthreadpool-metrics"> Metrics </a>
27/// * <a href="#bdlmt_fixedthreadpool-description"> Description </a>
28/// * <a href="#bdlmt_fixedthreadpool-thread-safety"> Thread Safety </a>
29/// * <a href="#bdlmt_fixedthreadpool-synchronous-signals-on-unix"> Synchronous Signals on Unix </a>
30/// * <a href="#bdlmt_fixedthreadpool-thread-names-for-sub-threads"> Thread Names for Sub-Threads </a>
31/// * <a href="#bdlmt_fixedthreadpool-usage"> Usage </a>
32/// * <a href="#bdlmt_fixedthreadpool-setting-fixedthreadpool-attributes"> Setting FixedThreadPool Attributes </a>
33/// * <a href="#bdlmt_fixedthreadpool-the-void-functionvoid-pointer-interface"> The "void functionvoid pointer" Interface </a>
34/// * <a href="#bdlmt_fixedthreadpool-the-functor-interface"> The Functor Interface </a>
35///
36/// # Purpose {#bdlmt_fixedthreadpool-purpose}
37/// Provide portable implementation for a fixed-size pool of threads.
38///
39/// # Classes {#bdlmt_fixedthreadpool-classes}
40///
41/// - bdlmt::FixedThreadPool: portable fixed-size thread pool
42///
43/// # Metrics {#bdlmt_fixedthreadpool-metrics}
44///
45///
46/// * `bde.backlog`
47/// > number of pending jobs minus number of "idle" threads in the thread pool
48/// > (may be negative)
49///
50/// * `bde.usedcapacity`
51/// > number of pending jobs divided by queue capacity
52///
53/// Associated Metric Attributes:
54/// * object type name: "bdlmt.fixedthreadpool"
55/// * object type abbreviation: "ftp"
56///
57/// @see bdlmt_threadpool
58///
59/// # Description {#bdlmt_fixedthreadpool-description}
60/// This component defines a portable and efficient implementation
61/// of a thread pool, `bdlmt::FixedThreadPool`, that can be used to distribute
62/// various user-defined functions ("jobs") to a separate threads to execute the
63/// jobs concurrently. Each thread pool object manages a fixed number of
64/// processing threads and can hold up to a fixed maximum number of pending
65/// jobs.
66///
67/// `bdlmt::FixedThreadPool` implements a queuing mechanism that distributes
68/// work among the threads. Jobs are queued for execution as they arrive, and
69/// each queued job is processed by the next available thread. If each of the
70/// concurrent threads is busy processing a job, new jobs will remain enqueued
71/// until a thread becomes available. If the queue capacity is reached,
72/// enqueuing jobs will block until threads consume more jobs from the queue,
73/// causing its length to drop below its capacity. Both the queue's capacity
74/// and number of threads are specified at construction and cannot be changed.
75///
76/// The thread pool provides two interfaces for specifying jobs: the commonly
77/// used "void function/void pointer" interface and the more versatile functor
78/// based interface. The void function/void pointer interface allows callers to
79/// use a C-style function to be executed as a job. The application need only
80/// specify the address of the function, and a single void pointer argument, to
81/// be passed to the function. The specified function will be invoked with the
82/// specified argument by the processing thread. The functor based interface
83/// allows for more flexible job execution such as the invocation of member
84/// functions or the passing of multiple user-defined arguments. See the `bdef`
85/// package-level documentation for more on functors and their usage.
86///
87/// Unlike a `bdlmt::ThreadPool`, an application can not tune a
88/// `bdlmt::FixedThreadPool` once it is created with a specified number of
89/// threads and queue capacity, hence the name "fixed" thread pool. An
90/// application can, however, specify the attributes of the threads in the pool
91/// (e.g., thread priority or stack size), by providing a
92/// `bslmt::ThreadAttributes` object with the desired values set. See
93/// @ref bslmt_threadutil package documentation for a description of
94/// `bslmt::ThreadAttributes`.
95///
96/// Thread pools are ideal for developing multi-threaded server applications. A
97/// server need only package client requests to execute as jobs, and
98/// `bdlmt::FixedThreadPool` will handle the queue management, thread
99/// management, and request dispatching. Thread pools are also well suited for
100/// parallelizing certain types of application logic. Without any complex or
101/// redundant thread management code, an application can easily create a thread
102/// pool, enqueue a series of jobs to be executed, and wait until all the jobs
103/// have executed.
104///
105/// ## Thread Safety {#bdlmt_fixedthreadpool-thread-safety}
106///
107///
108/// The `bdlmt::FixedThreadPool` class is both *fully thread-safe* (i.e., all
109/// non-creator methods can correctly execute concurrently), and is
110/// *thread-enabled* (i.e., the classes does not function correctly in a
111/// non-multi-threading environment). See @ref bsldoc_glossary for complete
112/// definitions of *fully thread-safe* and *thread-enabled*.
113///
114/// ## Synchronous Signals on Unix {#bdlmt_fixedthreadpool-synchronous-signals-on-unix}
115///
116///
117/// A thread pool ensures that, on unix platforms, all the threads in the pool
118/// block all asynchronous signals. Specifically all the signals, except the
119/// following synchronous signals are blocked:
120/// @code
121/// SIGBUS
122/// SIGFPE
123/// SIGILL
124/// SIGSEGV
125/// SIGSYS
126/// SIGABRT
127/// SIGTRAP
128/// SIGIOT
129/// @endcode
130///
131/// ## Thread Names for Sub-Threads {#bdlmt_fixedthreadpool-thread-names-for-sub-threads}
132///
133///
134/// To facilitate debugging, users can provide a thread name as the `threadName`
135/// attribute of the `bslmt::ThreadAttributes` argument passed to the
136/// constructor, that will be used for all the sub-threads. The thread name
137/// should not be used programmatically, but will appear in debugging tools on
138/// platforms that support naming threads to help users identify the source and
139/// purpose of a thread. If no `ThreadAttributes` object is passed, or if the
140/// `threadName` attribute is not set, the default value "bdl.FixedPool" will be
141/// used.
142///
143/// ## Usage {#bdlmt_fixedthreadpool-usage}
144///
145///
146/// This example demonstrates the use of a `bdlmt::FixedThreadPool` to
147/// parallelize a segment of program logic. The example implements a
148/// multi-threaded file search utility. The utility searches multiple files for
149/// a string, similar to the Unix command `fgrep`; the use of a
150/// `bdlmt::FixedThreadPool` allows the utility to search multiple files
151/// concurrently.
152///
153/// The example program will take as input a string and a list of files to
154/// search. The program creates a `bdlmt::FixedThreadPool`, and then enqueues a
155/// single "job" for each file to be searched. Each thread in the pool will
156/// take a job from the queue, open the file, and search for the string. If a
157/// match is found, the job adds the filename to an array of matching filenames.
158/// Because this array of filenames is shared across multiple jobs and across
159/// multiple threads, access to the array is controlled via a `bslmt::Mutex`.
160///
161/// ### Setting FixedThreadPool Attributes {#bdlmt_fixedthreadpool-setting-fixedthreadpool-attributes}
162///
163///
164/// To get started, we declare thread attributes, to be used in constructing the
165/// thread pool. In this example, our choices for number of threads and queue
166/// capacity are arbitrary.
167/// @code
168/// #define SEARCH_THREADS 10
169/// #define SEARCH_QUEUE_CAPACITY 50
170/// @endcode
171/// Below is the structure that will be used to pass arguments to the file
172/// search function. Since each job will be searching a separate file, a
173/// distinct instance of the structure will be used for each job.
174/// @code
175/// struct my_FastSearchJobInfo {
176/// const bsl::string *d_word; // word to search for
177/// const bsl::string *d_path; // path of the file to search
178/// bslmt::Mutex *d_mutex; // mutex to control access to the
179/// // result file list
180/// bsl::vector<bsl::string> *d_outList; // list of matching files
181/// };
182/// @endcode
183///
184/// ### The "void functionvoid pointer" Interface {#bdlmt_fixedthreadpool-the-void-functionvoid-pointer-interface}
185///
186///
187/// `myFastSearchJob` is the search function to be executed as a job by threads
188/// in the thread pool, matching the "void function/void pointer" interface.
189/// The single `void *` argument is received and cast to point to a
190/// `struct my_FastSearchJobInfo`, which then points to the search string and a
191/// single file to be searched. Note that different `my_FastSearchJobInfo`
192/// structures for the same search request will differ only in the attribute
193/// `d_path`, which points to a specific filename among the set of files to be
194/// searched; other fields will be identical across all structures for a given
195/// Fast Search.
196///
197/// See the following section for an illustration of the functor interface.
198/// @code
199/// static void myFastSearchJob(void *arg)
200/// {
201/// my_FastSearchJobInfo *job = (my_FastSearchJobInfo*)arg;
202/// FILE *file;
203///
204/// file = fopen(job->d_path->c_str(), "r");
205///
206/// if (file) {
207/// char buffer[1024];
208/// size_t nread;
209/// int wordLen = job->d_word->length();
210/// const char *word = job->d_word->c_str();
211///
212/// nread = fread(buffer, 1, sizeof(buffer) - 1, file);
213/// while (nread >= wordLen) {
214/// buffer[nread] = 0;
215/// if (strstr(buffer, word)) {
216/// @endcode
217/// If we find a match, we add the file to the result list and return. Since
218/// the result list is shared among multiple processing threads, we use a mutex
219/// lock to regulate access to the list. We use a `bslmt::LockGuard` to manage
220/// access to the mutex lock. This template object acquires a mutex lock on
221/// `job->d_mutex` at construction, releases that lock on destruction. Thus,
222/// the mutex will be locked within the scope of the `if` block, and released
223/// when the program exits that scope.
224///
225/// See @ref bslmt_threadutil for information about the `bslmt::Mutex` class, and
226/// component @ref bslmt_lockguard for information about the `bslmt::LockGuard`
227/// template class.
228/// @code
229/// bslmt::LockGuard<bslmt::Mutex> lock(job->d_mutex);
230/// job->d_outList->push_back(*job->d_path);
231/// break; // bslmt::LockGuard destructor unlocks mutex.
232/// }
233/// memcpy(buffer, &buffer[nread - wordLen - 1], wordLen - 1);
234/// nread = fread(buffer + wordLen - 1, 1, sizeof(buffer) - wordLen,
235/// file);
236/// }
237/// fclose(file);
238/// }
239/// }
240/// @endcode
241/// Routine `myFastSearch` is the main driving routine, taking three arguments:
242/// a single string to search for (`word`), a list of files to search, and an
243/// output list of files. When the function completes, the file list will
244/// contain the names of files where a match was found.
245/// @code
246/// void myFastSearch(const bsl::string& word,
247/// const bsl::vector<bsl::string>& fileList,
248/// bsl::vector<bsl::string>& outFileList)
249/// {
250/// bslmt::Mutex mutex;
251/// bslmt::ThreadAttributes defaultAttributes;
252/// @endcode
253/// We initialize the thread pool using default thread attributes. We then
254/// start the pool so that the threads can begin while we prepare the jobs.
255/// @code
256/// bdlmt::FixedThreadPool pool(defaultAttributes,
257/// SEARCH_THREADS,
258/// SEARCH_QUEUE_CAPACITY);
259///
260/// if (0 != pool.start()) {
261/// bsl::cerr << "Thread start() failed. Thread quota exceeded?"
262/// << bsl::endl;
263/// exit(1);
264/// }
265/// @endcode
266/// For each file to be searched, we create the job info structure that will be
267/// passed to the search function and add the job to the pool.
268///
269/// As noted above, all jobs will share a single mutex to guard the output file
270/// list. Function `myFastSearchJob` uses a `bslmt::LockGuard` on this mutex to
271/// serialize access to the list.
272/// @code
273/// int count = fileList.size();
274/// my_FastSearchJobInfo *jobInfoArray = new my_FastSearchJobInfo[count];
275///
276/// for (int i = 0; i < count; ++i) {
277/// my_FastSearchJobInfo &job = jobInfoArray[i];
278/// job.d_word = &word;
279/// job.d_path = &fileList[i];
280/// job.d_mutex = &mutex;
281/// job.d_outList = &outFileList;
282/// pool.enqueueJob(myFastSearchJob, &job);
283/// }
284/// @endcode
285/// Now we simply wait for all the jobs in the queue to complete. Any matched
286/// files should have been added to the output file list.
287/// @code
288/// pool.drain();
289/// delete[] jobInfoArray;
290/// }
291/// @endcode
292///
293/// ### The Functor Interface {#bdlmt_fixedthreadpool-the-functor-interface}
294///
295///
296/// The "void function/void pointer" convention is idiomatic for C programs.
297/// The `void` pointer argument provides a generic way of passing in user data,
298/// without regard to the data type. Clients who prefer better or more explicit
299/// type safety may wish to use the Functor Interface instead. This interface
300/// uses `bsl::function` to provide type-safe wrappers that can match argument
301/// number and type for a C++ free function or member function.
302///
303/// To illustrate the Functor Interface, we will make two small changes to the
304/// usage example above. First, we change the signature of the function that
305/// executes a single job, so that it uses a `my_FastSearchJobInfo` pointer
306/// rather than a `void` pointer. With this change, we can remove the first
307/// executable statement, which casts the `void *` pointer to
308/// `my_FastSearchJobInfo *`.
309/// @code
310/// static void myFastFunctorSearchJob(my_FastSearchJobInfo *job)
311/// {
312/// FILE *file;
313///
314/// file = fopen(job->d_path->c_str(), "r");
315/// // the rest of the function is unchanged.
316/// @endcode
317/// Next, we make a change to the loop that enqueues the jobs in `myFastSearch`.
318/// We create a functor - a C++ object that acts as a function. The thread pool
319/// will "execute" this functor (by calling its `operator()` member function) on
320/// a thread when one becomes available.
321/// @code
322/// for (int i = 0; i < count; ++i) {
323/// my_FastSearchJobInfo &job = jobInfoArray[i];
324/// job.d_word = &word;
325/// job.d_path = &fileList[i];
326/// job.d_mutex = &mutex;
327/// job.d_outList = &outFileList;
328///
329/// bsl::function<void()> jobHandle =
330/// bdlf::BindUtil::bind(&myFastFunctorSearchJob, &job);
331/// pool.enqueueJob(jobHandle);
332/// }
333/// @endcode
334/// Use of `bsl::function` and `bdlf::BindUtil` is described in the `bdef`
335/// package documentation. For this example, it is important to note that
336/// `jobHandle` is a functor object, and that `bdlf::BindUtil::bind` populates
337/// that functor object with a function pointer (to the `void` function
338/// `myFastFunctorSearchJob`) and user data (`&job`). When the functor is
339/// executed via `operator()`, it will in turn execute the
340/// `myFastFunctorSearchJob` function with the supplied data as its argument.
341///
342/// Note also that the functor is created locally and handed to the thread pool.
343/// The thread pool copies the functor onto its internal queue, and takes
344/// responsibility for the copied functor until execution is complete.
345///
346/// The function is completed exactly as it was in the previous example.
347/// @code
348/// pool.drain();
349/// delete[] jobInfoArray;
350/// }
351/// @endcode
352/// @}
353/** @} */
354/** @} */
355
356/** @addtogroup bdl
357 * @{
358 */
359/** @addtogroup bdlmt
360 * @{
361 */
362/** @addtogroup bdlmt_fixedthreadpool
363 * @{
364 */
365
366#include <bdlscm_version.h>
367
368#include <bdlcc_boundedqueue.h>
369
370#include <bdlf_bind.h>
371
372#include <bdlm_metricsregistry.h>
373
374#include <bsla_deprecated.h>
375
376#include <bslma_allocator.h>
377
378#include <bslmf_movableref.h>
379
380#include <bslmt_barrier.h>
381#include <bslmt_lockguard.h>
382#include <bslmt_mutex.h>
384#include <bslmt_threadutil.h>
385#include <bslmt_threadgroup.h>
386
387#include <bsls_assert.h>
388#include <bsls_atomic.h>
389#include <bsls_platform.h>
390
391#include <bsl_cstdlib.h>
392#include <bsl_functional.h>
393#include <bsl_string.h>
394
395#ifndef BDE_DONT_ALLOW_TRANSITIVE_INCLUDES
396
397#include <bdlcc_fixedqueue.h>
398
399#include <bslmt_condition.h>
400#include <bslmt_semaphore.h>
401
402#include <bsl_algorithm.h>
403
404#endif // BDE_DONT_ALLOW_TRANSITIVE_INCLUDES
405
406
407
408namespace bdlmt {
409
410/// This type declares the prototype for functions that are suitable to be
411/// specified `bdlmt::FixedThreadPool::enqueueJob`.
412extern "C" typedef void (*FixedThreadPoolJobFunc)(void *);
413
414 // =====================
415 // class FixedThreadPool
416 // =====================
417
418/// This class implements a thread pool used for concurrently executing
419/// multiple user-defined functions ("jobs").
420///
421/// See @ref bdlmt_fixedthreadpool
423
424 public:
425 // TYPES
426 typedef bsl::function<void()> Job;
428
429 // PUBLIC CONSTANTS
430 enum {
435 };
436
437 enum {
438 e_STOP
442#ifndef BDE_OMIT_INTERNAL_DEPRECATED
445 , BCEP_SUSPEND BSLA_DEPRECATED = e_SUSPEND
446 , BCEP_DRAIN BSLA_DEPRECATED = e_DRAIN
447#endif // BDE_OMIT_INTERNAL_DEPRECATED
448 };
449
450 private:
451 // PRIVATE CLASS DATA
452 static const char s_defaultThreadName[16]; // Thread name to use
453 // when none is
454 // specified.
455
456 // PRIVATE DATA
457 Queue d_queue; // underlying queue
458
459 bsls::AtomicInt d_numActiveThreads; // number of threads
460 // processing jobs
461
462 bsls::AtomicBool d_drainFlag; // set when draining
463
464 bslmt::Barrier d_barrier; // barrier to sync threads
465 // during `start` and `drain`
466
467 bslmt::Mutex d_metaMutex; // mutex to ensure that there
468 // is only one controlling
469 // thread at any time
470
471 bslmt::ThreadGroup d_threadGroup; // threads used by this pool
472
473 bslmt::ThreadAttributes d_threadAttributes; // thread attributes to be
474 // used when constructing
475 // processing threads
476
477 const int d_numThreads; // number of configured
478 // processing threads.
479
480#if defined(BSLS_PLATFORM_OS_UNIX)
481 sigset_t d_blockSet; // set of signals to be
482 // blocked in managed threads
483#endif
484
486 d_backlogHandle; // backlog metric handle
487
488
490 d_usedCapacityHandle; // used capacity metric
491 // handle
492
493 // PRIVATE MANIPULATORS
494
495 /// Initialize this thread pool using the stored attributes and the
496 /// specified `metricsRegistry` and `threadPoolName`. If
497 /// `metricsRegistry` is 0, `bdlm::MetricsRegistry::singleton()` is
498 /// used.
499 void initialize(bdlm::MetricsRegistry *metricsRegistry,
500 const bsl::string_view& threadPoolName);
501
502 /// The main function executed by each worker thread.
503 void workerThread();
504
505 /// Internal method to spawn a new processing thread and increment the current count.
506 ///
507 /// \note Note that this method must be called with
508 /// `d_metaMutex` locked.
509 int startNewThread();
510
511 private:
512 // NOT IMPLEMENTED
514 FixedThreadPool& operator=(const FixedThreadPool&);
515
516 public:
517 // CREATORS
518
519 /// Construct a thread pool with the specified `numThreads` number of
520 /// threads and a job queue of capacity sufficient to enqueue the
521 /// specified `maxNumPendingJobs` without blocking. Optionally specify
522 /// a `basicAllocator` used to supply memory. If `basicAllocator` is 0,
523 /// the currently installed default allocator is used. The name used for created threads is "bdl.FixedPool".
524 ///
525 /// \pre The behavior is undefined unless `1 <= numThreads`.
526 ///
527 /// \note Note that the actual job queue capacity can be
528 /// obtained using `queueCapacity`, and may be more than
529 /// `maxNumPendingJobs`.
531 int maxNumPendingJobs,
532 bslma::Allocator *basicAllocator = 0);
533
534 /// Construct a thread pool with the specified `numThreads` number of
535 /// threads, a job queue of capacity sufficient to enqueue the specified
536 /// `maxNumPendingJobs` without blocking, the specified
537 /// `threadPoolName` to be used to identify this thread pool, and the
538 /// specified `metricsRegistry` to be used for reporting metrics. If
539 /// `metricsRegistry` is 0, `bdlm::MetricsRegistry::singleton()` is
540 /// used. Optionally specify a'basicAllocator' used to supply memory.
541 /// If `basicAllocator` is 0, the currently installed default allocator
542 /// is used. The name used for created threads is `threadPoolName` if not empty, otherwise "bdl.FixedPool".
543 ///
544 /// \pre The behavior is undefined unless `1 <= numThreads`.
545 ///
546 /// \note Note that the actual job queue capacity can be
547 /// obtained using `queueCapacity`, and may be more than
548 /// `maxNumPendingJobs`.
550 int maxNumPendingJobs,
551 const bsl::string_view& threadPoolName,
552 bdlm::MetricsRegistry *metricsRegistry,
553 bslma::Allocator *basicAllocator = 0);
554
555 /// Construct a thread pool with the specified `threadAttributes`,
556 /// `numThreads` number of threads, and a job queue with capacity
557 /// sufficient to enqueue the specified `maxNumPendingJobs` without
558 /// blocking. Optionally specify a `basicAllocator` used to supply
559 /// memory. If `basicAllocator` is 0, the currently installed default
560 /// allocator is used. The name used for created threads is
561 /// `threadAttributes.threadName()` if not empty, otherwise
562 /// "bdl.FixedPool". The detached state of `threadAttributes` is ignored,
563 /// and `e_CREATE_JOINABLE` is used in all cases.
564 ///
565 /// \pre The behavior is undefined unless `1 <= numThreads`.
566 /// \note Note that the actual job queue
567 /// capacity can be obtained using `queueCapacity`, and may be more than
568 /// `maxNumPendingJobs`.
570 int numThreads,
571 int maxNumPendingJobs,
572 bslma::Allocator *basicAllocator = 0);
573
574 /// Construct a thread pool with the specified `threadAttributes`,
575 /// `numThreads` number of threads, a job queue with capacity sufficient
576 /// to enqueue the specified `maxNumPendingJobs` without blocking, the
577 /// specified `threadPoolName` to be used to identify this thread
578 /// pool, and the specified `metricsRegistry` to be used for reporting
579 /// metrics. If `metricsRegistry` is 0,
580 /// `bdlm::MetricsRegistry::singleton()` is used. Optionally specify a
581 /// `basicAllocator` used to supply memory. If `basicAllocator` is 0,
582 /// the currently installed default allocator is used. The name used for
583 /// created threads is `threadAttributes.threadName()` if not empty,
584 /// otherwise `threadPoolName` if not empty, otherwise "bdl.FixedPool".
585 /// The detached state of `threadAttributes` is ignored, and
586 /// `e_CREATE_JOINABLE` is used in all cases.
587 ///
588 /// \pre The behavior is undefined unless `1 <= numThreads`.
589 /// \note Note that the actual job queue capacity can
590 /// be obtained using `queueCapacity`, and may be more than
591 /// `maxNumPendingJobs`.
593 int numThreads,
594 int maxNumPendingJobs,
595 const bsl::string_view& threadPoolName,
596 bdlm::MetricsRegistry *metricsRegistry,
597 bslma::Allocator *basicAllocator = 0);
598
599 /// Remove all pending jobs from the queue without executing them, block
600 /// until all currently running jobs complete, and then destroy this
601 /// thread pool.
603
604 // MANIPULATORS
605
606 /// Disable enqueueing into this pool. All subsequent invocations of
607 /// `enqueueJob` or `tryEnqueueJob` will fail immediately. All blocked
608 /// invocations of `enqueueJob` will fail immediately. If the pool is already enqueue disabled, this method has no effect.
609 ///
610 /// \note Note that this
611 /// method has no effect on jobs currently in the pool.
612 void disable();
613
614 /// Enable queuing into this pool. If the queue is not enqueue
615 /// disabled, this call has no effect.
616 void enable();
617
618 /// Enqueue the specified `functor` to be executed by the next available
619 /// thread. Return 0 on success, and a non-zero value otherwise.
620 /// Specifically, return `e_SUCCESS` on success, `e_DISABLED` if
621 /// `!isEnabled()`, and `e_FAILED` if an error occurs. This operation
622 /// will block if there is not sufficient capacity in the underlying
623 /// queue until there is free capacity to successfully enqueue this job.
624 /// Threads blocked (on enqueue methods) due to the underlying queue
625 /// being full will unblock and return `e_DISABLED` if `disable` is invoked (on another thread).
626 ///
627 /// \pre The behavior is undefined unless
628 /// `functor` is not null.
629 int enqueueJob(const Job& functor);
631
632 /// Enqueue the specified `function` to be executed by the next
633 /// available thread. The specified `userData` pointer will be passed
634 /// to the function by the processing thread. Return 0 on success, and
635 /// a non-zero value otherwise. Specifically, return `e_SUCCESS` on
636 /// success, `e_DISABLED` if `!isEnabled()`, and `e_FAILED` if an error
637 /// occurs. This operation will block if there is not sufficient
638 /// capacity in the underlying queue until there is free capacity to
639 /// successfully enqueue this job. Threads blocked (on enqueue methods)
640 /// due to the underlying queue being full will unblock and return
641 /// `e_DISABLED` if `disable` is invoked (on another thread).
642 ///
643 /// \pre The behavior is undefined unless `function` is not null.
644 int enqueueJob(FixedThreadPoolJobFunc function, void *userData);
645
646 /// Enqueue the specified `functor` to be executed by the next available
647 /// thread. Return 0 on success, and a non-zero value otherwise.
648 /// Specifically, return `e_SUCCESS` on success, `e_DISABLED` if
649 /// `!isEnabled()`, `e_FULL` if `isEnabled()` and the underlying queue
650 /// was full, and `e_FAILED` if an error occurs.
651 ///
652 /// \pre The behavior is undefined unless `functor` is not null.
653 int tryEnqueueJob(const Job& functor);
655
656 /// Enqueue the specified `function` to be executed by the next
657 /// available thread. The specified `userData` pointer will be passed
658 /// to the function by the processing thread. Return 0 on success, and
659 /// a non-zero value otherwise. Specifically, return `e_SUCCESS` on
660 /// success, `e_DISABLED` if `!isEnabled()`, `e_FULL` if `isEnabled()`
661 /// and the underlying queue was full, and `e_FAILED` if an error occurs.
662 ///
663 /// \pre The behavior is undefined unless `function` is not null.
664 int tryEnqueueJob(FixedThreadPoolJobFunc function, void *userData);
665
666 /// Wait until the underlying queue is empty without disabling this pool
667 /// (and may thus wait indefinitely), and then wait until all executing
668 /// jobs complete. If the thread pool was not already started (`isStarted()` is `false`), this method has no effect.
669 ///
670 /// \note Note that if
671 /// any jobs are submitted concurrently with this method, this method
672 /// may or may not wait until they have also completed.
673 void drain();
674
675 /// Disable enqueuing jobs on this thread pool, cancel all pending jobs,
676 /// wait until all active jobs complete, and join all processing
677 /// threads. If the thread pool was not already started (`isStarted()`
678 /// is `false`), this method has no effect. At the completion of this
679 /// method, `false == isStarted()`.
680 void shutdown();
681
682 /// Spawn threads until there are `numThreads()` processing threads. On
683 /// success, enable enqueuing and return 0. Otherwise, join all threads
684 /// (ensuring `false == isStarted()`) and return -1. If the thread pool
685 /// was already started (`isStarted()` is `true`), this method has no
686 /// effect.
687 int start();
688
689 /// Disable enqueuing jobs on this thread pool, wait until all active
690 /// and pending jobs complete, and join all processing threads. If the
691 /// thread pool was not already started (`isStarted()` is `false`), this
692 /// method has no effect. At the completion of this method,
693 /// `false == isStarted()`.
694 void stop();
695
696 // ACCESSORS
697
698 /// Return `true` if enqueuing jobs is enabled on this thread pool, and
699 /// `false` otherwise.
700 bool isEnabled() const;
701
702 /// Return `true` if `numThreads()` are started on this threadpool and
703 /// `false` otherwise (indicating that 0 threads are started on this
704 /// thread pool.)
705 bool isStarted() const;
706
707 /// Return a snapshot of the number of threads that are currently
708 /// processing a job for this threadpool.
709 int numActiveThreads() const;
710
711 /// Return a snapshot of the number of jobs currently enqueued to be
712 /// processed by thread pool.
713 int numPendingJobs() const;
714
715 /// Return the number of threads passed to this thread pool at
716 /// construction.
717 int numThreads() const;
718
719 /// Return a snapshot of the number of threads currently started by this
720 /// thread pool.
721 int numThreadsStarted() const;
722
723 /// Return the capacity of the queue used to enqueue jobs by this thread
724 /// pool.
725 int queueCapacity() const;
726};
727
728// ============================================================================
729// INLINE DEFINITIONS
730// ============================================================================
731
732 // ---------------------
733 // class FixedThreadPool
734 // ---------------------
735
736// MANIPULATORS
737inline
739{
740 d_queue.disablePushBack();
741}
742
743inline
745{
746 d_queue.enablePushBack();
747}
748
749inline
751{
752 BSLS_ASSERT(functor);
753
754 return d_queue.pushBack(functor);
755}
756
757inline
764
765inline
767 void *userData)
768{
769 BSLS_ASSERT(0 != function);
770
771 return enqueueJob(bdlf::BindUtil::bindR<void>(function, userData));
772}
773
774inline
776{
777 BSLS_ASSERT(functor);
778
779 return d_queue.tryPushBack(functor);
780}
781
782inline
789
790inline
792 void *userData)
793{
794 BSLS_ASSERT(0 != function);
795
796 return tryEnqueueJob(bdlf::BindUtil::bindR<void>(function, userData));
797}
798
799inline
801{
802 bslmt::LockGuard<bslmt::Mutex> lock(&d_metaMutex);
803
804 if (isStarted()) {
805 d_queue.waitUntilEmpty();
806
807 d_drainFlag = true;
808 d_queue.disablePopFront();
809 d_barrier.wait();
810
811 d_drainFlag = false;
812 d_queue.enablePopFront();
813 d_barrier.wait();
814 }
815}
816
817inline
819{
820 bslmt::LockGuard<bslmt::Mutex> lock(&d_metaMutex);
821
822 if (isStarted()) {
823 d_queue.disablePushBack();
824 d_queue.disablePopFront();
825 d_threadGroup.joinAll();
826 d_queue.removeAll();
827 }
828}
829
830inline
832{
833 bslmt::LockGuard<bslmt::Mutex> lock(&d_metaMutex);
834
835 if (isStarted()) {
836 d_queue.disablePushBack();
837 d_queue.waitUntilEmpty();
838 d_queue.disablePopFront();
839 d_threadGroup.joinAll();
840 }
841}
842
843// ACCESSORS
844inline
846{
847 return !d_queue.isPushBackDisabled();
848}
849
850inline
852{
853 return d_numThreads == d_threadGroup.numThreads();
854}
855
856inline
858{
859 return d_numActiveThreads.loadAcquire();
860}
861
862inline
864{
865 return static_cast<int>(d_queue.numElements());
866}
867
868inline
870{
871 return d_numThreads;
872}
873
874inline
876{
877 return d_threadGroup.numThreads();
878}
879
880inline
882{
883 return static_cast<int>(d_queue.capacity());
884}
885
886} // close package namespace
887
888
889#endif
890
891// ----------------------------------------------------------------------------
892// Copyright 2024 Bloomberg Finance L.P.
893//
894// Licensed under the Apache License, Version 2.0 (the "License");
895// you may not use this file except in compliance with the License.
896// You may obtain a copy of the License at
897//
898// http://www.apache.org/licenses/LICENSE-2.0
899//
900// Unless required by applicable law or agreed to in writing, software
901// distributed under the License is distributed on an "AS IS" BASIS,
902// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
903// See the License for the specific language governing permissions and
904// limitations under the License.
905// ----------------------------- END-OF-FILE ----------------------------------
906
907/** @} */
908/** @} */
909/** @} */
Definition bdlcc_boundedqueue.h:418
bsl::size_t capacity() const
Definition bdlcc_boundedqueue.h:1398
int waitUntilEmpty() const
Definition bdlcc_boundedqueue.h:1439
void enablePushBack()
Definition bdlcc_boundedqueue.h:1390
void removeAll()
Definition bdlcc_boundedqueue.h:1196
void enablePopFront()
Definition bdlcc_boundedqueue.h:1383
@ e_FAILED
Definition bdlcc_boundedqueue.h:630
@ e_DISABLED
Definition bdlcc_boundedqueue.h:629
@ e_SUCCESS
Definition bdlcc_boundedqueue.h:626
@ e_FULL
Definition bdlcc_boundedqueue.h:628
void disablePopFront()
Definition bdlcc_boundedqueue.h:1364
int pushBack(const TYPE &value)
Definition bdlcc_boundedqueue.h:1123
bsl::size_t numElements() const
Definition bdlcc_boundedqueue.h:1433
bool isPushBackDisabled() const
Definition bdlcc_boundedqueue.h:1426
int tryPushBack(const TYPE &value)
Definition bdlcc_boundedqueue.h:1282
void disablePushBack()
Definition bdlcc_boundedqueue.h:1376
Definition bdlm_metricsregistry.h:306
Definition bdlm_metricsregistry.h:197
Definition bdlmt_fixedthreadpool.h:422
int numActiveThreads() const
Definition bdlmt_fixedthreadpool.h:857
@ e_SUSPEND
Definition bdlmt_fixedthreadpool.h:440
@ e_DRAIN
Definition bdlmt_fixedthreadpool.h:441
@ BSLA_DEPRECATED
Definition bdlmt_fixedthreadpool.h:443
@ e_STOP
Definition bdlmt_fixedthreadpool.h:438
@ e_RUN
Definition bdlmt_fixedthreadpool.h:439
void shutdown()
Definition bdlmt_fixedthreadpool.h:818
int tryEnqueueJob(const Job &functor)
Definition bdlmt_fixedthreadpool.h:775
bdlcc::BoundedQueue< Job > Queue
Definition bdlmt_fixedthreadpool.h:427
void disable()
Definition bdlmt_fixedthreadpool.h:738
@ e_FULL
Definition bdlmt_fixedthreadpool.h:432
@ e_FAILED
Definition bdlmt_fixedthreadpool.h:434
@ e_SUCCESS
Definition bdlmt_fixedthreadpool.h:431
@ e_DISABLED
Definition bdlmt_fixedthreadpool.h:433
bsl::function< void()> Job
Definition bdlmt_fixedthreadpool.h:426
FixedThreadPool(const bslmt::ThreadAttributes &threadAttributes, int numThreads, int maxNumPendingJobs, const bsl::string_view &threadPoolName, bdlm::MetricsRegistry *metricsRegistry, bslma::Allocator *basicAllocator=0)
FixedThreadPool(int numThreads, int maxNumPendingJobs, const bsl::string_view &threadPoolName, bdlm::MetricsRegistry *metricsRegistry, bslma::Allocator *basicAllocator=0)
int numThreads() const
Definition bdlmt_fixedthreadpool.h:869
int queueCapacity() const
Definition bdlmt_fixedthreadpool.h:881
bool isStarted() const
Definition bdlmt_fixedthreadpool.h:851
void enable()
Definition bdlmt_fixedthreadpool.h:744
bool isEnabled() const
Definition bdlmt_fixedthreadpool.h:845
int numThreadsStarted() const
Definition bdlmt_fixedthreadpool.h:875
void drain()
Definition bdlmt_fixedthreadpool.h:800
int enqueueJob(const Job &functor)
Definition bdlmt_fixedthreadpool.h:750
int numPendingJobs() const
Definition bdlmt_fixedthreadpool.h:863
void stop()
Definition bdlmt_fixedthreadpool.h:831
FixedThreadPool(int numThreads, int maxNumPendingJobs, bslma::Allocator *basicAllocator=0)
FixedThreadPool(const bslmt::ThreadAttributes &threadAttributes, int numThreads, int maxNumPendingJobs, bslma::Allocator *basicAllocator=0)
Definition bslstl_stringview.h:471
Forward declaration.
Definition bslstl_function.h:946
Definition bslma_allocator.h:545
Definition bslmf_movableref.h:752
Definition bslmt_barrier.h:353
Definition bslmt_lockguard.h:234
Definition bslmt_mutex.h:317
Definition bslmt_threadattributes.h:361
Definition bslmt_threadgroup.h:156
int numThreads() const
Definition bslmt_threadgroup.h:301
Definition bsls_atomic.h:1490
Definition bsls_atomic.h:744
int loadAcquire() const
Definition bsls_atomic.h:1753
#define BSLS_ASSERT(X)
Definition bsls_assert.h:1976
#define BSLS_IDENT(str)
BSLS_IDENT() - insert string into .comment binary segment (if supported)
Definition bsls_ident.h:238
Definition bdlmt_eventscheduler.h:550
void(* FixedThreadPoolJobFunc)(void *)
Definition bdlmt_fixedthreadpool.h:412
static MovableRef< t_TYPE > move(t_TYPE &reference) BSLS_KEYWORD_NOEXCEPT
Definition bslmf_movableref.h:1067
static t_TYPE & access(t_TYPE &ref) BSLS_KEYWORD_NOEXCEPT
Definition bslmf_movableref.h:1039