Programming with POSIX® Threads

47 return 0;
48
49 pthread_mutex_lock (&tail->mutex);
50 while (!tail->data_ready)
51 pthread_cond_wait (&tail->avail, &tail->mutex);
52 *result = tail->data;
53 tail->data_ready = 0;
54 pthread_cond_signal (&tail->ready);
55 pthread_mutex_unlock (&tail->mutex);
56 return 1;
57 }

Part 6 shows the main program that drives the pipeline. It creates a pipeline, and then loops reading lines from stdin. If the line is a single '=' character, it pulls a result from the pipeline and prints it. Otherwise, it converts the line to an integer value, which it feeds into the pipeline.

¦ pipe.c part 6 main

1 /*
2 * The main program to 'drive' the pipeline...
3 */
4 int main (int argc, char *argv[])
5 {
6 pipe_t my_pipe;
7 long value, result;
8 int status;
9 char line[128];
10
11 pipe_create (&my_pipe, 10);
12 printf ('Enter integer values, or '=' for next result
');
13
14 while (1) {
15 	printf ('Data> ');
16 	if (fgets (line, sizeof (line), stdin) == NULL) exit (0);
17 	if (strlen (line) <= 1) continue;
18 	if (strlen (line) <= 2 && line[0] == '=') {
19 		if (pipe_result (&my_pipe, &result))
20 			printf ('Result is %ld
', result);
21 		else
22 			printf ('Pipe is empty
');
23 	} else {
24 	if (sscanf (line, '%ld', &value) < 1)
25 		fprintf (stderr, 'Enter an integer value
');
26 	else
27 		pipe_start (&my_pipe, value);
28 	}
29 }
30 }

4.2 Work crew

The twelve jurors were all writing very busily on slates. 'What are they doing?'Alice whispered to the Gryphon.

'They ca'n't have anything to put down yet, before the trial's begun.' 'They're putting down their names,' the Gryphon whispered in reply, 'for fear they should forget them before the end of the trial.'

— Lewis Carroll, Alice's Adventures in Wonderland

In a work crew, data is processed independently by a set of threads (Figure 4.2). A 'parallel decomposition' of a loop generally falls into this category. A set of threads may be created, for example, each directed to process some set of rows or columns of an array. A single set of data is split between the threads, and the result is a single (filtered) set of data. Because all the threads in the work crew, in this model, are performing the same operation on different data, it is often known as SIMD parallel processing, for 'single instruction, multiple data.' The original use of SIMD was in an entirely different form of parallelism, and doesn't literally apply to threads — but the concept is similar.

FIGURE 4.2 Work crew

The threads in a work crew don't have to use a SIMD model, though. They may perform entirely different operations on different data. The members of our work crew, for example, each remove work requests from a shared queue, and do whatever is required by that request. Each queued request packet could describe a variety of operations—but the common queue and 'mission statement' (to process that queue) make them a 'crew' rather than independent worker threads. This model can be compared to the original definition of MIMD parallel processing, 'multiple instruction, multiple data.'

Section 7.2, by the way, shows the development of a more robust and general (and more complicated) 'work queue manager' package. A 'work crew' and a 'work queue' are related in much the same way as 'invariants' and 'critical sections' — it depends on how you look at what's happening. A work crew is the set of threads that independently processes data, whereas a work queue is a mechanism by which your code may request that data be processed by anonymous and independent 'agents.' So in this section, we develop a 'work crew,' whereas in Section 7.2 we develop a more sophisticated 'work queue.' The focus differs, but the principle is the same.

The following program, called crew.c, shows a simple work crew. Run the program with two arguments, a string, and a file path. The program will queue the file path to the work crew. A crew member will determine whether the file path is a file or a directory—if a file, it will search the file for the string; if a directory, it will use readdir_r to find all directories and regular files within the directory, and queue each entry as new work. Each file containing the search string will be reported on stdout.

Part 1 shows the header files and definitions used by the program. 7 The symbol CREW_SIZE determines how many threads are created for each work crew.

13-17 Each item of work is described by a work_t structure. This structure has a pointer to the next work item (set to NULL to indicate the end of the List), a pointer to the file path described by the work item, and a pointer to the string for which the program is searching. As currently constructed, all work items point to the same search string.

23-27 Each member of a work crew has a worker_t structure. This structure contains the index of the crew member in the crew vector, the thread identifier of the crew member (thread), and a pointer to the crew_t structure (crew).

33-41 The crew_t structure describes the work crew state. It records the number of members in the work crew (crew_size) and an array of worker_t structures (crew). It also has a counter of how many work items remain to be processed (work_count) and a list of outstanding work items (first points to the earliest item, and last to the latest). Finally, it contains the various Pthreads synchronization objects: a mutex to control access, a condition variable (done) to wait for the work crew to finish a task, and a condition variable on