Pools of threads

Another thing that QNX Neutrino has added is the concept of thread pools. You'll often notice in your programs that you want to be able to run a certain number of threads, but you also want to be able to control the behavior of those threads within certain limits.

For example, in a server you may decide that initially just one thread should be blocked, waiting for a message from a client. When that thread gets a message and is off servicing a request, you may decide that it would be a good idea to create another thread, so that it could be blocked waiting in case another request arrived. This second thread would then be available to handle that request. And so on. After a while, when the requests had been serviced, you would now have a lot of threads sitting around, waiting for further requests. In order to conserve resources, you may decide to kill off some of those “extra” threads.

This is in fact a common operation, and QNX Neutrino provides a library to help with this. We'll see the thread pool functions again in the Resource Managers chapter.

It's important for the discussions that follow to realize there are really two distinct operations that threads that are used in thread pools perform:

• a blocking (waiting operation)
• a processing operation

The blocking operation doesn't generally consume CPU. In a typical server, this is where the thread is waiting for a message to arrive. Contrast that with the processing operation, where the thread may or may not be consuming CPU (depending on how the process is structured). When we look at thread pool functions later, you'll see that we have the ability to control the number of threads in the blocking operation as well as the number of threads in the processing operations.

QNX Neutrino provides the following functions to deal with thread pools:

#include <sys/dispatch.h>

thread_pool_t *
thread_pool_create (thread_pool_attr_t *attr,
                    unsigned flags);

int
thread_pool_destroy (thread_pool_t *pool);

int
thread_pool_start (void *pool);

int
thread_pool_limits (thread_pool_t *pool,
                    int lowater,
                    int hiwater,
                    int maximum,
                    int increment,
                    unsigned flags);

int
thread_pool_control (thread_pool_t *pool,
                     thread_pool_attr_t *attr,
                     uint16_t lower,
                     uint16_t upper,
                     unsigned flags);

As you can see from the functions provided, you first create a thread pool definition using thread_pool_create(), and then start the thread pool via thread_pool_start(). When you're done with the thread pool, you can use thread_pool_destroy() to clean up after yourself. Note that you might never call thread_pool_destroy(), as in the case where the program is a server that runs “forever.” The thread_pool_limits() function lets you modify thread pool behavior by adjusting the attributes of the thread pool, and the thread_pool_control() function is a convenience wrapper for the thread_pool_limits() function.

So, the first function to look at is thread_pool_create(). It takes two parameters, attr and flags. The attr is an attributes structure that defines the operating characteristics of the thread pool (from <sys/dispatch.h>):

typedef struct _thread_pool_attr {
    // thread pool functions and handle
    THREAD_POOL_HANDLE_T    *handle;

    THREAD_POOL_PARAM_T
        *(*block_func)(THREAD_POOL_PARAM_T *ctp);

    void
        (*unblock_func)(THREAD_POOL_PARAM_T *ctp);

    int
        (*handler_func)(THREAD_POOL_PARAM_T *ctp);

    THREAD_POOL_PARAM_T
        *(*context_alloc)(THREAD_POOL_HANDLE_T *handle);

    void
        (*context_free)(THREAD_POOL_PARAM_T *ctp);

    // thread pool parameters
    pthread_attr_t          *attr;
    unsigned short          lo_water;
    unsigned short          increment;
    unsigned short          hi_water;
    unsigned short          maximum;
    const char              *tid_name;
    void                    (*error_func)
                                (unsigned flags, int err_value);
} thread_pool_attr_t;