code. compile. run. debug. share.
Source Code
Language
import threading
import queue
import time
import random
class BoundedBuffer:
def __init__(self, capacity):
self.buffer = queue.Queue(capacity)
self.lock = threading.Lock()
self.not_full = threading.Condition(self.lock)
self.not_empty = threading.Condition(self.lock)
def put(self, item):
with self.lock:
while self.buffer.full():
self.not_full.wait()
self.buffer.put(item)
self.not_empty.notify()
def get(self):
with self.lock:
while self.buffer.empty():
self.not_empty.wait()
item = self.buffer.get()
self.not_full.notify()
return item
def producer(buffer, id):
for i in range(10):
message = f"Message {i} from Producer {id}"
buffer.put(message)
time.sleep(random.random())
buffer.put("Bye")
def consumer(buffer, id):
while True:
message = buffer.get()
print(f"Consumer {id}: {message}")
if message == "Bye":
break
time.sleep(random.random() * 0.1)
def main():
BUFFER_SIZE = 32
NUM_PRODUCERS = 2
NUM_CONSUMERS = 2
buffer = BoundedBuffer(BUFFER_SIZE)
producers = [
threading.Thread(target=producer, args=(buffer, i))
for i in range(NUM_PRODUCERS)
]
consumers = [
threading.Thread(target=consumer, args=(buffer, i))
for i in range(NUM_CONSUMERS)
]
for t in producers + consumers:
t.start()
for t in producers + consumers:
t.join()
if __name__ == "__main__":
main()
Compiling Program...
Standard Input:
Interactive Console
Text
Program is not being debugged. Click "Debug" button to start program in debug mode.
| # | Function | File:Line |
|---|
| Variable | Value |
|---|
| Register | Value |
|---|
| Expression | Value |
|---|
| # | Description |
|---|