224. Bounded Queues in Multithreading

224. Bounded Queues in Multithreading

🔹 1. Creating a Bounded Queue with queue.Queue

import queue

# Create a bounded queue with a maximum size of 5
q = queue.Queue(maxsize=5)

# Put items in the queue
for i in range(5):
    q.put(i)

# Check if the queue is full
print(f"Is the queue full? {q.full()}")

✅ Fix: queue.Queue() creates a bounded queue with a specified maxsize.

---

🔹 2. Using queue.Queue in a Producer-Consumer Example

import queue
import threading
import time

# Create a bounded queue
q = queue.Queue(maxsize=5)

def producer():
    for i in range(10):
        q.put(i)
        print(f"Produced: {i}")
        time.sleep(0.5)

def consumer():
    while True:
        item = q.get()
        print(f"Consumed: {item}")
        time.sleep(1)

# Create threads
producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

producer_thread.join()
consumer_thread.join()

✅ Fix: The producer adds items to the queue, and the consumer takes items out. The queue's bounded size prevents the producer from overwhelming the consumer.

---

🔹 3. Blocking Behavior on Full Queue

import queue
import threading
import time

q = queue.Queue(maxsize=3)

def producer():
    for i in range(5):
        print(f"Attempting to produce: {i}")
        q.put(i)
        print(f"Produced: {i}")
        time.sleep(1)

def consumer():
    while True:
        item = q.get()
        print(f"Consumed: {item}")
        time.sleep(2)

producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

producer_thread.join()
consumer_thread.join()

✅ Fix: When the queue reaches its max size, the producer will be blocked until the consumer processes an item.

---

🔹 4. Using queue.Queue for Task Scheduling

import queue
import threading

q = queue.Queue(maxsize=10)

def task_worker():
    while True:
        task = q.get()
        print(f"Task {task} is being processed")
        q.task_done()

# Start multiple worker threads
for i in range(3):
    t = threading.Thread(target=task_worker)
    t.daemon = True
    t.start()

# Add tasks to the queue
for task in range(20):
    q.put(task)

q.join()  # Wait for all tasks to be processed

✅ Fix: Tasks are queued and processed by worker threads. The queue's bounded nature ensures limited concurrent processing.

---

🔹 5. Handling Timeouts with Bounded Queues

import queue
import threading
import time

q = queue.Queue(maxsize=2)

def producer():
    try:
        q.put(1, timeout=1)  # Timeout if the queue is full
        print("Produced: 1")
    except queue.Full:
        print("Queue is full, producer is waiting...")

producer_thread = threading.Thread(target=producer)
producer_thread.start()
producer_thread.join()

✅ Fix: The timeout parameter allows the producer to timeout and handle cases when the queue is full.

---

🔹 6. Thread-Safe Queue with queue.Queue

import queue
import threading

q = queue.Queue(maxsize=5)

def thread_function():
    for i in range(5):
        q.put(i)
        print(f"Thread {threading.current_thread().name} produced: {i}")

# Create multiple threads
threads = [threading.Thread(target=thread_function) for _ in range(3)]
for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

# Check the queue's final state
print(f"Queue size: {q.qsize()}")

✅ Fix: Multiple threads produce items safely with a thread-safe queue.

---

🔹 7. Using queue.PriorityQueue for Task Prioritization

import queue
import threading
import time

pq = queue.PriorityQueue(maxsize=5)

def producer():
    for i in range(5, 0, -1):
        pq.put((i, f"task-{i}"))
        print(f"Produced: task-{i}")
        time.sleep(0.5)

def consumer():
    while True:
        priority, task = pq.get()
        print(f"Consumed: {task} with priority {priority}")
        pq.task_done()
        time.sleep(1)

producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

producer_thread.join()
consumer_thread.join()

✅ Fix: PriorityQueue allows tasks to be prioritized by a tuple of priority and task data.

---

🔹 8. Bounded Queue with queue.LifoQueue (Stack-like Behavior)

import queue
import threading
import time

q = queue.LifoQueue(maxsize=3)

def producer():
    for i in range(5):
        q.put(i)
        print(f"Produced: {i}")
        time.sleep(0.5)

def consumer():
    while True:
        item = q.get()
        print(f"Consumed: {item}")
        time.sleep(1)

producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

producer_thread.join()
consumer_thread.join()

✅ Fix: LifoQueue implements a stack (Last In, First Out) for the queue behavior.

---

🔹 9. Gracefully Closing Threads with queue.Queue

import queue
import threading
import time

q = queue.Queue(maxsize=5)

def producer():
    for i in range(5):
        q.put(i)
        print(f"Produced: {i}")
        time.sleep(1)

def consumer():
    while True:
        item = q.get()
        if item is None:  # Signal to stop
            break
        print(f"Consumed: {item}")
        time.sleep(1)

producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

producer_thread.join()
q.put(None)  # Signal the consumer to stop
consumer_thread.join()

✅ Fix: Use None as a sentinel value to signal the consumer to stop processing.

---

🔹 10. Using queue.Queue for Rate Limiting

import queue
import threading
import time

q = queue.Queue(maxsize=3)

def producer():
    for i in range(6):
        if q.full():
            print(f"Rate limit reached, producer waiting...")
        q.put(i)
        print(f"Produced: {i}")
        time.sleep(1)

producer_thread = threading.Thread(target=producer)
producer_thread.start()
producer_thread.join()

✅ Fix: A rate limit is imposed by the bounded queue's size, limiting the number of items produced in a fixed amount of time.

---

🚀 Summary: Why Use Bounded Queues?

Feature	Bounded Queue Benefits
Flow Control	✅ Prevents overwhelming consumers or resources.
Concurrency	✅ Enables safe access from multiple threads.
Rate Limiting	✅ Controls the rate of task production.
Thread Safety	✅ Ensures thread-safe interactions.

---

🚀 When to Use Bounded Queues?

Scenario	Use Bounded Queue?
Task Scheduling in a Multithreaded Environment	✅ Yes
Preventing Resource Exhaustion	✅ Yes
Rate Limiting or Throttling	✅ Yes
Ensuring Proper Synchronization Between Threads	✅ Yes