Synchronization 3: Producer/Consumer Problem

in order to better appreciate the importance of synchronization we'll be looking at a very common synchronization problem in operating systems known as the producer consumer problem the idea here is that we have one or more processes or threads that are producing data and storing it in a buffer meanwhile one or more consumer threads are consuming that data we have to make sure that only one thread or process is accessing the buffer at a time and that we don't produce more data in the buffer can hold nor do we consume data from the buffer when it is empty so the idea is that we have some fixed size buffer which is simply an array of length n so there are n entries and we're using zero indexing so we have this buffer and we are going to gradually add items to it that's what the producers will do and while the producers are adding more and more data to this the consumers will extract data from the n now eventually the producers will reach the end of this array and what we're going to do is wrap it around and treat this as a circular buffer or circular array so if the producer has already placed an item in this last position and then wants to add a new item it will be added at the front now this can only be done if the item that was at the front earlier has been consumed by one of the consumer threads so we have to make sure that the consumption stays ahead of production but we also have to make sure that we don't start consuming things that haven't been produced yet so overflow would be when we produce so much data that we overwrite some data that hasn't been consumed yet underflow would be if we can zooom something that hasn't been produced yet we want to avoid both of these problems some of the hassle of managing this data structure can be abstracted away by simply treating it as a queue however in the low-level context where we often need to solve this problem namely in operating systems we generally don't want to have too much abstraction because we want to make sure everything functions as quickly and efficiently as possible but even if we have that level of abstraction we still need to provide synchronization and so I'll be providing solutions to this problem using both semaphores and monitors with condition variables this is a solution to the producer-consumer problem using semaphores now the code is broken up into three sections but that's only done to fit all the code on the screen all of this code belongs in a single source file from top to bottom this line followed by these lines followed by the main now of course this is pseudocode so you'd have to change it to make it work anyway the way it works is that we have a buffer size which is some fixed size that's the size of our circular buffer and then we have three semaphores now the semaphore s starts at one and all it really does is control access to the buffer it's really only providing mutual exclusion now the semaphore is in an e are providing synchronization they help the producer and consumer threads communicate with each other in is going to prevent underflow because it indicates the number of items in the buffer that are available for consumption and as it goes negative that will indicate the number of threads that are waiting to consume things but are blocked in contrast this semaphore e is the amount of free space left in the buffer so it will prevent overflow because as items are added to the buffer this number will go down notice that it starts off at the buffer size and when it goes below zero that indicates the number of threads that would like to produce something and add it to the buffer but are blocked because it's already full so here is our producer process or threat and we have a while loop and we're going to loop as long as we're able to produce so this is some condition that will make sense in the context using it in you could simply have while true here but more generally we're looping as long as there's something that can be produced and so what we'll do is some command produce will produce and return a value and save it in a variable X and then because we want to add this to the buffer we're going to wait on the semaphore II now II starts off at buffer size and waiting decrement sit so as long as there is room left in the buffer decrementing it won't cause any problems it won't cause any blocks will simply decrement that to indicate that we need to use up a slot and we'll continue on then we do a weight on s now remember that s controls overall access to the buffer and provides mutual exclusion in fact these calls to the semaphore s to both the weight and the signal here could easily easily be replaced by a mutex lock the weight is essentially a lot command in the signal here is essentially an unlock because with a value of one waiting sets it to zero meaning that if any other process tries to execute some weight on s then that process will block so we can view this here as being a very tightly associated critical section where we lock and then unlock access to the buffer now after locking access to the buffer we append that newly produced value X to the end of it and so a pend is once again some command that is defined elsewhere that takes care of the nitty-gritty details of managing the buffer then once we have appended a new value to the buffer and we've unlocked access to the buffer by signalling on s we then do a seminal on in now remember that n prevents underflow n indicates how many items are available for consumption so it starts off at zero because the buffer is initially empty but when we signal on n we're indicating we're going to increment this variable thus incrementing that something is now available to be consumed so this signal command is a way of communicating to the consumer thread which we'll discuss in a moment that it can consume something so going over to the consumer thread right now we have a similar structure where we loop while there is still something that can be consumed so this loop keeps going over and over and over again as long as they're something to consume and the first thing it does is wait on the semaphore n so you can imagine that if these threads are running at the same time if the consumer reaches this command first before anything has been produced then the consumer will block because with an initial value of zero this sim weight command will decrement in the negative one and by the rules of semaphores that causes the thread or process to block so it's forced to wait until something is actually produced now if this doesn't block it means that something was previously produced and we can continue and so we first wait on the semaphore s thus gaining exclusive mutually exclusive acts says to the buffer so we have once again this sort of structure here where we're going to lock the buffer and then this same signal s will unlock it and while this process or thread has mutually exclusive access to the buffer it will take out the next value and store it in a variable X so in terms of the data structure we're manipulating this append is like an NQ command on a q and the take is like a DQ command on the q so we take out X and then we unlock access to the buffer and then by taking out X we've created a new space in the buffer for something to be produced something to be added so we will then send signal on e so remember E is tracking how much empty space there is and so we've created a new empty space with this take command and so we now signal that so that a producer can go forward and produce more items in fact if we had filled the buffer let's say that the produce loop here ran as many times as the buffer size then eventually it would get to a point where this some weight command would block because we would consume or rather we would decrement this variable enough times that it went to zero then we would decrement it one more time it would be negative and the thing that would increment it and enable more items to be produced would be this same signal command here now after we've signaled that room has been made inside the buffer we then consume the value X that we already produced and consumed depends on the specifics of your particular problem so this same signal command is meant to communicate to this some weight so if a consumer had blocked because it was waiting at this point the same signal would tell that blocked process that it continued that it can continue from this point forward similarly this signal goes back here telling process that became blocked when it tried to produce something but it can now continue producing things so that is the back and forth communication of these synchronization operations and then we have the main function here and so we're doing a parallel launch of one producer in one consumer that is the bare minimum for this code to work however you could have multiple producers and multiple consumers and this exact same code would work fine regardless of how many producers and consumers you have because they provide mutually exclusive access to the buffer the control with a semaphore s and this SEM signal could communicate with any of the SEM weight commands in any number of different consumer threads likewise this SEM signal in the consumer could communicate with any number of assembly commands and any number of producer threads now there's one more important takeaway here and that is that although we have this produced command in this consume command outside of critical sections of the code you could conceivably want to move this into the critical section it essentially depends on exactly how you're producing things and how you're consuming things but you'll get more efficiency if you're able to avoid keeping this in the critical section similarly with the consume so it depends on particular problem and what you're producing how you're producing it and how you're consuming it so this is a semaphore based solution let's now look at a solution using monitors and variables now start by recalling that a monitor is a separate bit of code removed from the actual main body of the code the monitor is an encapsulated structure in a programming language that has some functions or methods associated with it to which access is mutually exclusive so in this solution we're actually going to define at least in part the append and take commands that were simply assumed to exist in the previous semaphore solution so in this monitor class or entity we have the buffer size defined and we have to conditioned variables one named not full in one name not empty we're also going to maintain a count which is the number of items inside the buffer now the append command that is similar to what was used in the semaphore solution but a bit more complicated we're going to do some synchronization now keep in mind that simply by virtue of being inside of a monitor access to this append command is already mutually exclusive so simply by calling append it's very similar in the previous solution to doing some weight on s in same signal honest so we enter append we want to append a particular object X to the end of the queue to the you know the end of the circular buffer and so what we do is we loop as long as the count the number of objects in the buffer is equal to the buffer size in other words if the buffer is full then we need to loop and what do we do in that loop we wait so we are waiting for the buffer to not be full so not full is one of our condition variables and so we're waiting until the buffer is not full now when this code starts as you'll see in a second the buffer will of course be empty so this will be false we'll skip right through it I've left out details of the implementation but some code would be here that adds an item to the end of the buffer and after adding that item we also need to increment our count because it's important for tracking the synchronization and then we use C notify on this not empty condition variable so where we are telling any other processes that need to know that because we have just added something to the buffer it is not empty now it may already have not been empty but we're just indicating that it's definitely not empty now so we either went from being empty to being not empty or we went from being not empty to still being not empty but we've sent this notification which is something that could be listened to or waited for by this take command so this take method is once again similar to what we had in the semaphore solution but a bit more complicated because it's inside the monitor and the tape command starts off with a while loop it says while the count is equal to zero so in other words as long as the buffer is empty we're going to wait for it to not be empty so this is our second condition variable and as long as the buffer is empty we are waiting but once C notify has been executed once this will send a signal that can be caught here at which point any blocked thread or process would become unblocked and the reason for this while loop is that you may have multiple different producer/consumer threads that all want to grab access to the buffer as soon as it becomes available and so even though you've just been told that the buffer is not empty and you can consume something the while loop versus the threader process to check one more time to make sure that the count is still not equal to zero before moving forward because in the tiny interval when these different processes are released from their blocked queue another like you don't know which one of those processes will swoop in and actually be able to execute so they all need to do one last check to make sure they're allowed to continue but one of them should be able to get through and then we'll execute some code I've left it out but essentially we're going to take the next item from the buffer we're going to store it in a variable called X we're going to decrement the count because there is now 1 for your item in the buffer which means there's space to add more things we will notify other threads of this fact so we do see notify on not full which can be caught here so if the buffer had reached a point where it was full then this see notify not full would allow any block threads here to go forward and then we return X now this code is only for the monitor we have to combine it with some actual main program code but the main program code is fairly simple now because we have a monitor m and then this is our producer thread that's our consumer thread we have our while loop that says as long as we can still produce keep looping and all we do is we produce a value X and then we do m dot append of X so this calls this method here and then in the consumer thread we have a while loop and we start off with take command and so n is our monitor so we take a value and that is this code here and then once we take the value we can see it in the loop continues and then the main function is the same we do a parallel launch of a producer and a consumer but as I said before there could be multiple producers and multiple consumers and that wouldn't cause any problems so although overall this code in combination is a bit longer the actual main body code here is far simpler because of the abstraction of having a monitor now of course even if you don't have a monitor you can simulate this by using mutexes as we saw previously in a different video but let me comment briefly on the similarities between this solution and the semaphore solution so this C wait on the not full condition variable is very similar to the SEM wait on the semaphore e in the semaphore solution similarly the notification on the not empty condition variable is very much like the signal on the semaphore n in a semaphore solution and in the consumer we have a weight in a signal command as well just like we have a weight and a notify in the take command of the monitor so the not full condition variable is behaving like the semaphore e for the sake of synchronization and the not empty condition variable is behaving like the semaphore in in this solution for synchronization and as I mentioned briefly before the reason the monitor not have any sort of construct or condition variable corresponding to this semaphore s is that all the semaphore s is doing is providing mutual exclusion so that doesn't have to be explicitly represented in the monitor because it's simply a given that by virtue of being in a monitor access to the append and tank methods is usually exclusive this is if this were Java code this would be indicated by the presence of the synchronized keyword so it's as if I said that this method and this method are both synchronized so this is not Java code it's a pseudo code but if if you want to think of it in those terms because these methods are synchronized there is an implicit locking and unlocking of some shared mutex simply by calling these methods and so that is the nice thing about the monitor is that it implicitly provides mutual exclusion which allows this code here the main code to be very very simple if you didn't have that implicit locking then you would need to have sort of a mutex lock here and a mutex unlock here and then the same thing would have to happen at these two points with a mutex lock and mu techs unlock so that is how you can solve the producer-consumer problem using either semaphores or a monitor with condition variables

Channel: Jacob Schrum

Views: 31,302

Rating: 4.9296875 out of 5

Keywords: producer, consumer, producer/consumer, monitor, mutex, lock, semaphore, semWait, semSignal, cwait, cnotify, synchronized, buffer, queue, circular array, circular buffer, synchronization, mutual exclusion

Id: nxw2y27z0V4

Length: 24min 25sec (1465 seconds)

Published: Wed Aug 03 2016