16_Producers_and_Consumers

1 .Lecture 16 – Producers and Consumers Describe the producer-consumer pattern for distributing work between threads Solve the producer-consumer problem using semaphores and/or condition variables Implement various types of thread-safe queues and identify when each should be applied Describe the generator pattern and when it applies Given a problem description, identify and justify the optimal approach for distributing work between threads/processes Learning Goals Producers and Consumers 1 © Paul Davies, C. Antonio Sanchez. Not to be copied, used, or revised without explicit written permission from the copyright owner.

2 .Concurrent Work Distribution What is the optimal way to distribute work between threads? Considerations: More threads potentially lead to more opportunity for concurrency Creating/managing each thread adds overhead Uneven distribution of work between threads can lead to cores sitting idle Producers and Consumers 1 – 1,000,000 1,000,001 – 2,000,000 e.g. Computing primes on two cores: idle 2

3 .Concurrent Work Distribution What if instead of manually deciding how to split up workload, we let the threads themselves decide ? Put everything that needs to be done into common memory If a thread is ready to do some work, it can grab the next task No worker threads will sit idle until all work is complete Producers and Consumers Worker Worker Worker ... To Do: 3

4 .Producer-Consumer Pattern Producers: add work to a shared thread-safe queue Consumers: remove work from a shared thread-safe queue Producers and Consumers ... Producer Consumer Producer Producer Consumer Consumer Queue All synchronization is handled internally by the queue. Queue supports two operations: push(…) adds an item pop(…) removes an item 4

5 .Producer-Consumer Problem Given a fixed-size memory block that has one or more producers populating the block with data, and one or more consumers removing data from the block, design a synchronization mechanism . Producers and Consumers Producer Consumer Memory Two issues to resolve: Concurrent access: we don’t want any two threads to read/modify the contents at the same time ( mutual exclusion ) Update notifications: the consumer needs to be notified when there is new data to be read, otherwise it will re-read old data ( synchronization ) 5

6 .Producer-Consumer Problem Producers and Consumers How would you do it if the memory stored a single element? Producer Consumer Producer Producer Consumer Consumer Element 6

7 .Producer-Consumer Problem Solution #1: Semaphores Create a producer semaphore psem(1) (one element free) Create a consumer semaphore csem(0) (no elements to read) Producer waits on psem , pushes new element, notifies consumer of new data available Consumer waits on csem , pops element, notifies producer of free slot Producers and Consumers Produce: psem.wait() push(data) csem.notify() Consume: csem.wait() data = pop() psem.notify() 7

8 .wait() notify() notify() wait() Producer-Consumer Problem Solution #1: Semaphores Producers and Consumers Producer Consumer Empty Full data data 0 1 0 1 psem csem Produce: psem.wait() push(data) csem.notify() Consume: csem.wait() data = pop() psem.notify() 8

9 .Producer-Consumer Problem Solution 2: One Condition Variable Boolean flag to indicate if slot is full , initialized to false Single condition variable to wait for changes of flag status, cv Producer waits for flag to be false, pushes data into slot, sets full to true, notifies cv Consumer waits for flag to be true, pops data out of slot, sets full to false, notifies cv Producers and Consumers Produce: cv.wait() until full == false push(data) flag = true cv.notify_all() Consume: cv.wait() until full == true pop(data) flag = false; cv.notify_all() 9

10 .wait(false) notify_all() notify_all() wait(true) Producer-Consumer Problem Solution #2: One Condition Variable Producers and Consumers Producer Consumer Empty Full data data true false cv Produce: cv.wait() until full == false push(data) flag = true cv.notify_all() Consume: cv.wait() until full == true pop(data) flag = false; cv.notify_all() Advantage: simpler design Disadvantage: all producers/consumers must be woken up every time 10

11 .Producer-Consumer Problem Solution 2: Two Condition Variables Boolean flag to indicate if slot is full , initialized to false Two condition variables to wait for changes of flag status, pcv and ccv Producer waits on pcv for flag to be false, pushes data, sets full to true, notifies ccv Consumer waits on ccv for flag to be true, pops data, sets full to false, notifies pcv Producers and Consumers Produce: pcv.wait() until full == false push(data) flag = true ccv.notify_one() Consume: ccv.wait() until full == true data = pop() flag = false; pcv.notify_one() 11

12 .false true ccv wait(false) notify_one() wait(true) Producer-Consumer Problem Solution #2: Two Condition Variables Producers and Consumers Consumer Empty Full data data true false pcv Produce: pcv.wait() until full == false push(data) flag = true ccv.notify_one() Consume: ccv.wait() until full == true data = pop() flag = false; pcv.notify_one() notify_one() Producer false 12

13 .Producer-Consumer Problem What if two consumers try to remove an item at the same time? Producers and Consumers 1 csem Since there is only one slot, only one consumer can “consume” at a time, others need to wait until next item is produced. wait() Consumer wait() Consumer 13

14 .Producer-Consumer Problem What if two producers try to add an item at the same time? Producers and Consumers wait() Producer 1 psem wait() Producer Since there is only one slot, only one producer can “produce” at a time, others need to wait until item is consumed. 14

15 .Producer-Consumer Problem Producers and Consumers Producers are usually faster than consumers, since consumers are the ones actually processing data . We may not want our producers to block, so they can continue gathering more data in their own threads. How can we allow more than one producer to produce at a time? Producer Consumer Producer Producer Consumer Consumer Queue 15

16 .Producer-Consumer Problem Producers and Consumers How would you solve the producer-consumer problem using a fixed-sized buffer ? Producer Consumer Producer Producer Consumer Consumer Queue Treat it as a thread-safe circular buffer, use mutexes and semaphores to control access. 16

17 .Circular Buffer Producers and Consumers Buffer Top Buffer Bottom Pointer to Reading Position Pointer to Writing Position Materials: index for read offset cidx(0) index for write offset pidx(0) mutex for read offset cmut mutex for write offset pmut semaphore for reading csem(0) semaphore for writing psem(N) Methods: Produce: psem.wait() // wait for empty slot pmut.lock() // grab index under lock idx = pidx; pidx = (pidx+1)%N push(data) // produce pmut.unlock() csem.notify() // one more item avail. Consume: csem.wait() // wait for item cmut.lock() // grab index under lock idx = cidx; cidx = (cidx+1)%N data = pop() // consume cmut.unlock() psem.notify() // one more empty slot 17

18 .Circular Buffer: Producer Producers and Consumers Buffer Top Buffer Bottom pidx Producer A Producer B queue queue.push(A) queue.push(B) Producer A safely gets idx = pidx = 0 Pushes data into queue idx = 0 Producer B safely gets idx = pidx = 1 Pushes data into queue idx = 1 A B 18

19 .Circular Buffer: Consumer Producers and Consumers Buffer Top Buffer Bottom cidx Consumer A Consumer B queue next = queue.pop() next = queue.pop() Consumer A safely gets idx = cidx = 2 Pops data out of queue idx = 2 Consumer B safely gets idx = cidx = 3 Pops data out of queue idx = 3 C D 19

20 .Producers and Consumers Producer-Consumer Problem As long as producers and consumers are operating at roughly the same rate , and if the buffer is large enough to account for any small deviations, the circular buffer solution is optimal . Q1: How big should the buffer be? Q2: What if producers are faster than consumers? Q3: What can we do to guarantee that a producer will never block? 20 The queue should be sufficiently large to minimize wait time for producers, while balancing memory restrictions of the queue. To guarantee a producer will never block, disregard memory restrictions and make a dynamically-sized queue .

21 .Dynamic Producer-Consumer Queue Producers and Consumers 21 ... Producer Consumer Producer Producer Consumer Consumer Queue Internal dynamic queue , protected by mutual exclusion One condition variable for consumers to wait while queue is empty Produce: mutex.lock() queue.push(data) mutex.unlock() cv.notify_one() Consume: mutex.lock() cv.wait( !queue.empty() ) data = queue.pop() mutex.unlock()

22 .Three Thread-Safe Queues Single element: Smallest memory footprint, least synchronization overhead Ideal when produced tasks are infrequent Producers and Consumers 22 Circular buffer: Fixed memory footprint, slightly more synchronization overhead Ideal when memory is restricted, or produce/consume rate roughy equal Dynamic buffer: Unbounded memory footprint, most synchronization overhead Ideal when we do not want producers to ever block, should have some sense of worst-case memory requirements

23 .Producer-Consumer Pattern Producers and Consumers 23 You want to find all prime numbers from 1 to 2,000,000,000 using “trial division”. Your machine has 2 cores. How many threads should you create? a) 2 b) 20 c) 2,000 d) 20,000,000,000 A: Two consumer threads using the producer-consumer pattern Q: What type of thread-safe queue should you use? Circular buffer with a size of about 4 to minimize consumer wait time. (we want to leave some extra room for producers to add to the queue before a consumer is ready)

24 .Stopping Conditions As written, when the work is complete, our consumers will wait forever for the next item in the queue which will never come . How do we tell consumers they are done? Producers and Consumers 24 Check for a “done” flag before trying to get next item? Could lead to race condition e.g. one item left to process, two consumers concurrently check flag We need a way to wake up waiting consumers from within pop() , inform them that they are done and so should shut down (or die ).

25 .Poison Pill A poison pill is a meaningful item that tells the consumer to ignore it and quit . MUST be unique enough that will never actually appear in data. If using pointers, could be nullptr or known constant pointer Otherwise, some known constant invalid entry Producers and Consumers 25 E.g. for prime numbers, could use POISON = 0. Consumer: while ( (data = queue.pop()) != POISON ) { // process data … }

26 .Generator Pattern In cases where inputs are predictable (e.g. compute all prime numbers…), do we really need producer threads to populate a queue? Instead, we can remove the producer side, and make the “queue” itself generate the next item. Producers and Consumers 26 ... Consumer Consumer Consumer Generator generator.yield() or next() or emit() or get() or pop()

27 .Generator Pattern Thread-safe generator: Producers and Consumers 27 class OddIntegerGenerator { int next_; int max_; std::mutex mutex_; public : const int POISON = 0 ; OddIntegerGenerator( int start, int max) : next_(start), max_(max), mutex_() {} int yield() { int out; { std::lock_guard< decltype (mutex_)> lock; out = next_; if (next_ > max) { next_ = POISON; } else { next_ += 2 ; } } return out; } }; void consumer( OddIntegerGenerator& oig ) { int val; while ((val = oig.yield()) != oig.POISON) { // ... process number } }

28 .Examples How would you compute all primes in the range [2, 2,000,000,000]? How would you process Google search requests? Processing of data already gathered and stored in a file? Live streams of weather data? Customer service calls? Producers and Consumers 28

29 .Homework Use Case Diagrams 29 Go through the course library examples in: examples/q14_monitor examples/q15_producer_consumer Quiz 6 next week on synchronization Finalize teams for course project , get started on overall system design UML DIAGRAMS!!!!