thread-pool.md 10.7 KB
Newer Older
laurent's avatar
laurent committed
1 2
# Thread pool

3
The **thread pool** is a working server, made of a set of worker threads that can be mapped on CPU cores.  Each thread pool has an **input queue** ("**FIFO**"), from which its workers pick **jobs** (FIFO element) to execute.  When a job is done, the worker sends a message to an output queue, if it has been defined.
laurent's avatar
laurent committed
4

5
All the thread pool functions are thread safe. The functions executed by worker threads are provided by the thread pool client, so the client has to handle the concurrency/parallel execution of his functions.
laurent's avatar
laurent committed
6 7

## license
8 9 10
    Author:
      Laurent Thomas, Open cells project
      The owner share this piece code to Openairsoftware alliance as per OSA license terms
11
      With contributions from Robert Schmidt, OSA
laurent's avatar
laurent committed
12

13
# Jobs
laurent's avatar
laurent committed
14

15
A job is a message of type `notifiedFIFO_elt_t`:
Laurent's avatar
Laurent committed
16

17 18 19 20 21 22
* `next`: internal FIFO chain, do not set it
* `key`: a `long int` that the client can use to identify a job or a group of messages
* `ResponseFifo`: if the client defines a response FIFO, the job will be posted back after processing
* `processingFunc`: any function (of type `void processingFunc(void *)`) that a worker will process
* `msgData`: the data passed to `processingFunc`. It can be added automatically, or you can set it to a buffer you are managing
* `malloced`: a boolean that enables internal free in the case of no return FIFO or abort feature
Laurent's avatar
Laurent committed
23

24
The job messages can be created and destroyed with `newNotifiedFIFO_elt()` and `delNotifiedFIFO_elt()`, or managed by the client:
laurent's avatar
laurent committed
25

26 27 28
* `newNotifiedFIFO_elt()`: Creates a job, that will later be used in queues/FIFO
* `delNotifiedFIFO_elt()`: Deletes a job
* `NotifiedFifoData()`: gives a pointer to the beginning of free usage memory in a job (you can put any data there, up to 'size' parameter you passed to `newNotifiedFIFO_elt()`)
laurent's avatar
laurent committed
29

30
These 3 calls are not mandatory, you can also use your own memory to save the `malloc()`/`free()` that are behind these calls.
Laurent's avatar
Laurent committed
31

32 33 34 35 36 37 38 39 40 41 42
## API details

### `notifiedFIFO_elt_t *newNotifiedFIFO_elt(int size, uint64_t key, notifiedFIFO_t *reponseFifo, void (*processingFunc)(void *))`

Creates a new job. The data part of the job will have size `size`, the job can be identified via `key`. `reponseFifo` is a FIFO queue to which the job should be pushed after being processed by a thread pool (see below for details) using function `processingFunc()`.

The function returns a pointer to an allocated job.

### `void delNotifiedFIFO_elt(notifiedFIFO_elt_t *elt)`

Free the memory allocated for a job `elt`.
Laurent's avatar
Laurent committed
43

44
### `void *NotifiedFifoData(notifiedFIFO_elt_t *elt)`
Laurent's avatar
Laurent committed
45

46
Returns a pointer (`void *`) to the data segment of an allocated job.
Laurent's avatar
Laurent committed
47

48
# Queues of jobs (`FIFO`)
Laurent's avatar
Laurent committed
49

50 51 52 53 54 55 56 57 58 59 60 61 62 63
Queues can be used to enqueue messages/jobs, of type `notifiedFIFO_t`.

* `initNotifiedFIFO()`: Create a queue
* No delete function is required, the creator has only to free the data of type `notifiedFIFO_t`
* `pushNotifiedFIFO()`: Add a job to a queue
* `pullNotifiedFIFO()`: Pull a job from a queue. This call is blocking until a job arrived.
* `pollNotifiedFIFO()`: Pull a job from a queue. This call is not blocking, so it returns always very shortly
* `abortNotifiedFIFO()`: Aborts a FIFO, such that it will always return `NULL`

Note that in 99.9% of cases, `pull()` is better than `poll()`.

The above push/pull/poll functions are built on not-thread-safe versions of these functions that are described below.

## Low level: fast and simple, but not thread-safe
Laurent's avatar
Laurent committed
64

65 66 67
* `initNotifiedFIFO_nothreadSafe()`: Create a queue
* `pushNotifiedFIFO_nothreadSafe()`: Add a element in a queue
* `pullNotifiedFIFO_nothreadSafe()`: Pull a element from a queue
Laurent's avatar
Laurent committed
68

69
As these queues are not thread safe, there is NO blocking mechanism, neither `pull()` versus `poll()` calls
Laurent's avatar
Laurent committed
70

71
There is no delete for a message queue: you only have to abandon the memory you allocated to call `initNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf)`
Laurent's avatar
Laurent committed
72

73 74 75
If you malloced the memory under `nf` parameter, you have to free it.

## FIFO queue abort
Laurent's avatar
Laurent committed
76

77
It is possible to *abort* the FIFO queue. In this case, any pulling/polling of jobs in the queue will return `NULL`. This can be used when stopping a program, to release multiple queues that might depend on each other (a proper design would try to avoid such dependencies, though).
Laurent's avatar
Laurent committed
78

79
## API details
Laurent's avatar
Laurent committed
80

81 82
### `void initNotifiedFIFO(notifiedFIFO_t *nf)`
### `void initNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf)`
Laurent's avatar
Laurent committed
83

84
Initializes a FIFO queue pointed to by `nf` ("notified FIFO").
Laurent's avatar
Laurent committed
85

86 87
### `void pushNotifiedFIFO(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg)`
### `void pushNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg)`
Laurent's avatar
Laurent committed
88

89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108
Pushes a job `msg` to the FIFO queue `nf`.

### `notifiedFIFO_elt_t *pullNotifiedFIFO(notifiedFIFO_t *nf)`

Pulls a job from a FIFO queue `nf`. The returned pointer points to the job.  This call is blocking, i.e., a caller will be blocked until there is a message.

If the FIFO has been aborted, the returned pointer will be `NULL`. In other words, if the call to this functions returns `NULL`, the FIFO has been deactivated.

### `notifiedFIFO_elt_t *pollNotifiedFIFO(notifiedFIFO_t *nf)`
### `notifiedFIFO_elt_t *pullNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf)`

Like `pullNotifiedFIFO()`, but non-blocking: they check if the queue `nf` contains a job, and return it. Otherwise, they return `NULL`.  Note the slight difference in naming: the thread-safe function is called *poll*, not *pull*.

Note that unlike for `pullNotifiedFIFO()`, returning `NULL` does not inform whether the queue has been aborted; the caller should manually check the `abortFIFO` flag of `nf` in this case.

### `void abortNotifiedFIFO(notifiedFIFO_t *nf)`

Aborts the entire FIFO queue `nf`: all jobs will be dropped, and the FIFO is marked as aborted, such that a call to `pullNotifiedFIFO()` returns `NULL`.

Returns the total number of jobs that were waiting or under execution.
laurent's avatar
laurent committed
109 110 111

# Thread pools

112
## Initialization
113

114
The clients can create one or more thread pools with
Laurent's avatar
Laurent committed
115

116 117
* `initTpool(char *params, tpool_t *pool, bool performanceMeas)` or
* `initNamedTpool(char *params,tpool_t *pool, bool performanceMeas , char *name)`
laurent's avatar
laurent committed
118

119
The threads are governed by the Linux real time scheduler, and their name is set automatically to `Tpool<thread index>_<core id>` if `initTpool()` is used or to `<name><thread index>_<core id>` when `initNamedTpool()` is used.
laurent's avatar
laurent committed
120

121
## Adding jobs
laurent's avatar
laurent committed
122

123
The client create their jobs messages as a `notifiedFIFO_elt_t`. They then push it with `pushTpool()` which internally calls `push_notifiedFIFO()`.
laurent's avatar
laurent committed
124

125
If they need a return value (e.g., result of a computation), they have to create response queues with `init_notifiedFIFO()` and set this FIFO pointer in the `notifiedFIFO_elt_t` before pushing the job. This way, multiple result queues can be used in one thread pool.
laurent's avatar
laurent committed
126

127
## Abort
laurent's avatar
laurent committed
128

129
`abortTpool()` kills all jobs in the Tpool, and terminates the pool.
laurent's avatar
laurent committed
130

Laurent's avatar
Laurent committed
131
## API details
132

133
Each thread pool (there can be several in the same process) should be initialized once using one of the two API's:
Laurent's avatar
Laurent committed
134

135 136 137
### `void initNamedTpool(char *params, tpool_t *pool, bool performanceMeas, char *name)`

### `void initTpool(char *params, tpool_t *pool, bool performanceMeas)`
Laurent's avatar
Laurent committed
138

139

140
The `params` string describes a list of cores, separated by "," that run a worker thread.  If the core exists on the CPU, the thread pool initialization sets the affinity between this thread and the related core. The following options are available:
Laurent's avatar
Laurent committed
141

142 143 144
* `N`: if an `N` is in the parameter list, no threads are created, and the threadpool will be disabled (jobs will be treated by each calling thread independently).
* A number that represent a valid CPU core on the target CPU, i.e., if there are `M` cores, a number within `[0,M-1]`: A thread is created and pinned to the core (with set affinity)
* `-1`: The thread will not be mapped onto a specific core (a "floating" thread)
Laurent's avatar
Laurent committed
145

146 147
Example: `-1,-1,-1`: the thread pool is made of 3 floating threads.
Be careful with a fixed allocation: it is hard to be more clever than Linux kernel!
Laurent's avatar
Laurent committed
148

149
`pool` is a pointer to the thread pool object.
150

151
`performanceMeas` is a flag to enable measurements (see also below).
Laurent's avatar
Laurent committed
152

153
`name` is used to build the thread names. 
Laurent's avatar
Laurent committed
154

155 156 157 158 159 160
### `void pushTpool(tpool_t *t, notifiedFIFO_elt_t *msg)`

Adds a job for processing in the thread pool.

The job data you can set are, inside `msg`:

161
* `key`: an arbitrary key to find a job in a response queue.
162 163 164 165
* `reponseFifo`: if non-`NULL`, the message will be sent back on this queue when the job is done. If `NULL`, the thread pool automatically frees the job when it is done.
* `processingFunc`: the function to execute for this job.

The `processingFunc` prototype is `void <func>(void *memory)`. The data part of the job (the pointer returned by `NotifiedFifoData(msg)`) is passed to the function. The job data will be updated by `processingFunc`, and the job will be pushed into return queue (the parameter `reponseFifo`).
laurent's avatar
laurent committed
166

167
### `notifiedFIFO_elt_t *pullTpool(notifiedFIFO_t *responseFifo, tpool_t *t)`
Laurent's avatar
Laurent committed
168

169
Collects a job result in a return queue that has been defined in `msg` when calling `pushTpool()`, and that has been updated by `processingFunc()`. Returns the corresponding job pointer of type `notifiedFIFO_elt_t *`.
Laurent's avatar
Laurent committed
170

171
Multiple return queues might be useful. Consider the following example in the eNB: I created one single thread pool (because it depends mainly on CPU hardware), but i use two return queues: one for turbo encode, one for turbo decode. Thus, when performing Turbo encoding/decoding, jobs are pushed into a single thread pool, but will end up in different result queues, such that DL/UL processing can be separated more easily.
Laurent's avatar
Laurent committed
172

173
### `notifiedFIFO_elt_t *tryPullTpool(notifiedFIFO_t *responseFifo, tpool_t *t)`
Laurent's avatar
Laurent committed
174

175
The same as `pullTpool()` in a non-blocking fashion (an alternative name would have been `pollTpool()`).
Laurent's avatar
Laurent committed
176

177
### `int abortTpool(tpool_t *t)`
Laurent's avatar
Laurent committed
178

179
Aborts the complete Tpool: cancel every work in the input queue, marks to drop existing jobs in processing, and terminates all worker threads. It is afterwards still possible to call functions such as `pushTpool()`, but each calling thread will execute the job itself.
Laurent's avatar
Laurent committed
180

181
Returns the total number of jobs that were aborted, i.e., waiting for execution or being executed.
Laurent's avatar
Laurent committed
182 183

## Performance measurements
184

185
A performance measurement is integrated: the pool will automacillay fill timestamps if you set the environement variable `threadPoolMeasurements` to a valid file name.  The following measurements will be written to Linux pipe on a per-job basis:
laurent's avatar
laurent committed
186

187 188 189 190
* `creationTime`: time the request is push to the pool;
* `startProcessingTime`: time a worker start to run on the job
* `endProcessingTime`: time the worker finished the job
* `returnTime`: time the client reads the result
laurent's avatar
laurent committed
191

192
The `measurement_display` tool to read the Linux pipe and display it in ASCII is provided.
193 194 195 196 197 198
In the cmake build directory, type `make/ninja measurement_display`. Use as
follows:
```
sudo threadPoolMeasurements=tpool.meas ./nr-softmodem ...
./measurement_display tpool.meas
```