Laurent Pautet (pautet@telecom-paris.fr)
Objectives
The objective of this lab is to implement in C a blocking queue service of fixed size. Thus, you will be able to experiment in detail some POSIX concurrency tools.
A Blocking Queue is a thread-safe queue designed for concurrent programming, where threads can wait for the queue to become non-empty when dequeuing or non-full when enqueuing. It is commonly used in producer-consumer scenarios.
Key Functions in Terms of Concurrency:
See specifications of Java ArrayBlockingQueue service.
Course and Sources
Course material on threads can be found here. You can see the complete documentation of POSIX functions related to threads by following this link.
You will find all the sources in this compressed archive. The next lab requires the implementation of a blocking queue of fixed size with conditional variables (4 first questions of this lab).
How to submit your work
Create a git repository on gitlab for SE205 with a directory lab3 in it. Give access to Laurent Pautet and send him the url.
How To Debug
To debug, we strongly recommend using gdb or lldb. Do not fill your program with printf. If you have a memory problem (SIGSEGV, ...), do:
gdb ./main_blocking_queue (gdb) run test-00.txt
In case of a problem, the program stops on the incorrect operation. To understand the issue, use gdb commands:
MacOS
For MacOS users, use lldb, not gdb.
We consider the main_blocking_queue.c file . This program creates producers and consumers which exchange data using a shared buffer. We focus first on the creation of producers and consumers.
Producers run main_producer procedure and output data (integers) in a blocking_queue that is a bounded buffer that we protect against concurrent access. Consumers execute the main procedure main_consumer and consume the produced data in the same way.
The accesses of producers and consumers can be made according to several semantics. They can be blocking, immediate, or timed operations, that is to say that if they block, they do not block at all or block no more than a given amount of time.
Producers and consumers adopt a periodic behavior. All periods from a common start time start_time, they produce or consume data then wait for the next period to start over. For example, the producers reactivate every start_time + n * producer_period where n represents the number of activations.
Input data
The scenarios are provided in the test-<XY>.txt files and are described as follows.
The number of producers (n_producers), that of consumers (n_consumers), the size of the bounded buffer (buffer_size), the semantics of access for consumers (sem_consumers) and for producers (sem_producers) (0 for blocking, 1 for non- blocking, 2 for timed blocking) or the periods of producers (producer_period) and consumers (consumer_period) are configured using a file which name is passed on the command line. Note that the consumer operations can be blocking while the producer ones can be immediate. The value 0 of the sem_impl parameter indicates that we are using the implementation with variables conditional and the value 1 that we use the implementation with semaphores. In the example below, the consumers use the immediate semantic (1).
#sem_consumers 1
Output data
The output lines consist of several fields. The first one indicates the value of the clock in second. The second one the name of the task and the following one the number of the task. The last two fields indicate the operation and the input or output data. In the example below, on the date 000, the producer attached to thread 2 performs the put operation of data 200 in blocking mode (B). The value 200 corresponds to producer 2 (hundreds value) and to (first) production 0 of this server (value of units).
000 [producer 02] put (B) - data=200
It should also be noted that when we have P producers and C consumers, the producers produce C times, and consumers consume P times. Thus, under normal circumstances, all data produced is consumed.
|
Complete the main_blocking_queue.c file to create as many producers and consumers than requested and then wait for the termination of producers and consumers. Save the identifiers of consumers and producers in the tasks table. |
You can test your code by doing:
> ./main_blocking_queue test-00.txt
You have to verify that you are creating the correct number of consumers and producers. By consulting the 3rd field of outputs below, we can actually note that there are 2 consumers and 5 producers as specified in test-00.txt. The other information is not relevant at this stage.
... start consumer 0 start consumer 1 start producer 2 start producer 3 start producer 4 start producer 5 start producer 6 ...
We implement a fixed-size blocking queue, ie a bounded buffer protected against concurrent access. There are two possible implementations, one based on condition variables and one based on sempahores.
We start with an implementation based on condition variables provided in the file cond_blocking_queue.c. To do this, we can use the implementation of a non-protected bounded buffer provided in the files bounded_buffer.c and bounded_buffer.h. Check available functions in file bounded_buffer.h.
|
Complete the blocking_queue_t structure. It only contains an non-protected bounded buffer currently. You have to add POSIX synchronization tools. |
|
Complete the cond_blocking_queue_init, cond_blocking_queue_get and cond_blocking_queue_put so that it offers the following services:
|
Test the correct functioning of the blocking queue using the main program of the file main_blocking_queue.c. You can use the test case test-00.txt. Note the instants when consumers and producers are supposed to block. Normally, with this semantics, all data produced must be consumed. Typically, in the test-00.txt example, we expect an output like:
... 000 [producer 02] put (B) - data=200 000 [consumer 00] get (B) - data=200 000 [producer 03] put (B) - data=300 000 [consumer 01] get (B) - data=300 000 [producer 04] put (B) - data=400 005 [consumer 00] get (B) - data=400 005 [producer 05] put (B) - data=500 005 [consumer 01] get (B) - data=500 ...
We will implement in the cond_blocking_queue.c file concurrent accesses to the blocking queue similar to the previous ones. However, the semantics of these accesses will this time be immediate. The producer will return with a boolean return code if it succeeded in immediately adding an element in the buffer. The consumer will indicate by a null pointer or not on a data if it has succeeded in immediately removing an item from the buffer. However, the buffer should still be protected against concurrent access.
|
Complete the cond_blocking_queue_remove and cond_blocking_queue_add procedures so that they offer the services indicated. |
Test the correct functioning of the blocking queue using the main program of the file main_blocking_queue.c.
You can use the scenarii, test-01.txt and test-02.txt. Verify that consumers and producers do not get blocked and execute as expected. The test-01.txt scenario has more than producers than consumers, the size 1 buffer will soon be full and some producers will fail to deposit their data.
In the example below, on date 000, consumer 00 is trying to consume while the buffer is empty (data = NULL). Then, the 5 producers produce data but the size of the buffer being 1, only the data produced by the first producer 01 will be stored in the buffer. This is confirmed on the date 005, when the consumer consumes and removes the data 100 stored first and not the data 500 last stored. Moreover, at date 010 , data 101 is consumed again and not data 200 or 501. The failure of the producer is indicated by a produced NULL data. Note the indication (I)) to signify immediate semantics.
... 000 [consumer 00] remove (I)) - data=NULL 000 [producer 01] add (I)) - data=100 000 [producer 02] add (I)) - data=NULL 000 [producer 03] add (I)) - data=NULL 000 [producer 04] add (I)) - data=NULL 000 [producer 05] add (I)) - data=NULL 005 [consumer 00] remove (I)) - data=100 005 [producer 01] add (I)) - data=101 005 [producer 02] add (I)) - data=NULL 005 [producer 03] add (I)) - data=NULL 005 [producer 04] add (I)) - data=NULL 005 [producer 05] add (I)) - data=NULL 010 [consumer 00] remove (I)) - data=101 ...
In the test-02.txt scenario, on the opposite, it is the consumers who fail to extract data.
... 000 [consumer 00] remove (I)) - data=NULL 000 [consumer 01] remove (I)) - data=NULL 000 [consumer 02] remove (I)) - data=NULL 000 [consumer 03] remove (I)) - data=NULL 000 [consumer 04] remove (I)) - data=NULL 000 [producer 05] add (I)) - data=500 005 [consumer 00] remove (I)) - data=500 005 [consumer 01] remove (I)) - data=NULL 005 [consumer 02] remove (I)) - data=NULL 005 [consumer 03] remove (I)) - data=NULL ...
We implement in the cond_blocking_queue.c file concurrent accesses to the blocking queue similar to the previous ones. However, the semantics of these accesses is this time timed blocking. If the operation cannot be performed immediately, it waits until it can, but will not wait beyond a given deadline (which corresponds to a date ie an absolute time). Note that if a consumer or a producer is unblocked, and its operation is not successful despite this notification, it must resume its wait operation until it succeeds or the deadline occurs. In this case, it returns a special value as before (NULL).
|
Complete the cond_blocking_queue_poll and cond_blocking_queue_offer procedures so that it offers the indicated services. We will use the POSIX tools in their timed version. |
Test the correct functioning of the blocking queue using the main program of the file main_blocking_queue.c. You can use the test cases, test-03.txt and test-04.txt. You will check that consumers and producers remain blocked for as long as possible and activate themselves as expected.
For example, in the test-04.txt scenario, the producer 05 puts a data consumed data immediately by the first consumer. Then the next 4 consumers fail to consume a data and are released on the following period ie on date 05.
... 000 [producer 05] offer (T) - data=500 000 [consumer 00] poll (T) - data=500 005 [consumer 04] poll (T) - data=NULL 005 [consumer 02] poll (T) - data=NULL 005 [consumer 01] poll (T) - data=NULL 005 [consumer 03] poll (T) - data=NULL 005 [producer 05] offer (T) - data=501 005 [consumer 00] poll (T) - data=501 010 [consumer 02] poll (T) - data=NULL 010 [consumer 01] poll (T) - data=NULL 010 [consumer 04] poll (T) - data=NULL 010 [consumer 03] poll (T) - data=NULL 010 [producer 05] offer (T) - data=502 ...
Same scenario for producers in test-03.txt,
... 000 [producer 01] offer (T) - data=100 000 [consumer 00] poll (T) - data=100 000 [producer 02] offer (T) - data=200 005 [producer 04] offer (T) - data=NULL 005 [producer 03] offer (T) - data=NULL 005 [producer 05] offer (T) - data=NULL 005 [consumer 00] poll (T) - data=200 005 [producer 01] offer (T) - data=101 010 [producer 03] offer (T) - data=NULL 010 [producer 05] offer (T) - data=NULL 010 [producer 02] offer (T) - data=NULL 010 [producer 04] offer (T) - data=NULL 010 [consumer 00] poll (T) - data=101 ...
At last, test that your implementation is correct when the consumer operations are timed blocking and the producer operations are immediate (test-05.txt) ...
000 [producer 01] add (I)) - data=100 000 [consumer 00] poll (T) - data=100 000 [producer 02] add (I)) - data=200 000 [producer 03] add (I)) - data=NULL 000 [producer 04] add (I)) - data=NULL 000 [producer 05] add (I)) - data=NULL 005 [consumer 00] poll (T) - data=200 005 [producer 01] add (I)) - data=101 005 [producer 02] add (I)) - data=NULL 005 [producer 03] add (I)) - data=NULL 005 [producer 04] add (I)) - data=NULL 005 [producer 05] add (I)) - data=NULL 010 [consumer 00] poll (T) - data=101 020 [consumer 00] poll (T) - data=NULL ... 050 [consumer 00] poll (T) - data=NULL ...
... and when the consumer operations are immediate and the producer ones timed blocking (test-06.txt).
... 000 [consumer 00] remove (I)) - data=NULL 000 [consumer 01] remove (I)) - data=NULL 000 [consumer 02] remove (I)) - data=NULL 000 [consumer 03] remove (I)) - data=NULL 000 [consumer 04] remove (I)) - data=NULL 000 [producer 05] offer (T) - data=500 005 [consumer 00] remove (I)) - data=500 005 [consumer 01] remove (I)) - data=NULL 005 [consumer 02] remove (I)) - data=NULL 005 [consumer 03] remove (I)) - data=NULL 005 [consumer 04] remove (I)) - data=NULL 005 [producer 05] offer (T) - data=501 015 [producer 05] offer (T) - data=NULL ... 050 [producer 05] offer (T) - data=NULL ...
In the code for producer main_producer and consumer main_consumer, you use the delay_until function to suspend the thread until the next deadline (deadline).
|
Analyze the delay_until procedure from file utils.c. Explain how this procedure can make the task wait well beyond the deadline if the execution of gettimeofday is not immediately followed by the execution of nanosleep. By using the POSIX functions, especially those from the previous question, you will provide an alternative implementation of delay_until which will not present the drawback noted in the previous paragraph. |
You will verify that the previous scenarios work normally.
We implement again the semantics of previous concurrent accesses, but this time we are going to use semaphores. Instead of using conditional variables to block execution, you will be using semaphores. You will notice that, unlike the version with the conditional variables, you will necessarily use two queues. You will modify the files sem_blocking_queue.h and sem_blocking_queue.c.
|
The requested work and the expected result are the same as in the section Buffer protected against concurrent access (var cond / blocking). However, you must use the scenario file test-10.txt in order to test your work. |
We implement again the semantics of previous concurrent accesses, but this time we are going to use semaphores. Instead of using conditional variables to block execution, you will be using semaphores. You will modify the files sem_blocking_queue.h and sem_blocking_queue.c.
|
The requested work and the expected result are the same as in the section Buffer protected against concurrent access (var cond / immediate). However, you must use the scenario files test-11.txt and test-12.txt in order to test your work. |
We implement again the semantics of previous concurrent accesses, but this time we are going to use semaphores. Instead of using conditional variables to block execution, you will be using semaphores. You will modify the files sem_blocking_queue.h and sem_blocking_queue.c.
|
The requested work and the expected result are the same as in the section Buffer protected against concurrent access (var cond / timed). However, you must use the scenario files test-13.txt and test-14.txt in order to test your work. |
At last, test that your implementation is correct when the consumer operations are timed blocking and the producer ones are immediate (test-15.txt) and when the consumer operations are immediate and the producer ones are timed blocking (test-16.txt).