Simple Producer-consumer implementation in Python

Question

I implemented a simple solution to the Producer–consumer problem that I'd love for you to take a look at.

The producer simply adds random numbers to a queue and the consumer (from a separate thread) pops numbers off the queue and prints them. I'd specifically like feedback on the concurrency aspects of this implementation. Thanks!

from collections import deque
import random
import threading


TIMES = 100


class SafeQueue:
  def __init__(self, capacity):
    self.capacity = capacity
    self.queue = deque([])
    self.remaining_space = threading.Semaphore(capacity)
    self.fill_count = threading.Semaphore(0)

  def append(self, item):
    self.remaining_space.acquire()
    self.queue.append(item)
    self.fill_count.release()

  def consume(self):
    self.fill_count.acquire()
    item = self.queue.popleft()
    self.remaining_space.release()
    return item


class Producer:
  def __init__(self, queue, times=TIMES):
    self.queue = queue
    self.times = times

  def run(self):
    for _ in range(self.times):
      self.queue.append(random.randint(0, 100))


class Consumer:
  def __init__(self, queue, times=TIMES):
    self.queue = queue
    self.times = times

  def run(self):
    for _ in range(self.times):
      print(self.queue.consume())


def main():
  queue = SafeQueue(10)

  producer = Producer(queue)
  producer_thread = threading.Thread(target=producer.run)
  consumer = Consumer(queue)
  consumer_thread = threading.Thread(target=consumer.run)

  producer_thread.start()
  consumer_thread.start()

  producer_thread.join()
  consumer_thread.join()


if __name__ == "__main__":
  main()

Answer 1

Actually, the initial producer-consumer scheme is over-complicated and has the following issues:

Using 2 separate threading.Semaphore objects:
One of them is used for producer (remaining_space), another - for consumer (fill_count).
Though they won't be convenient here (as I'll describe below), but for your future potential cases, a better naming would be self.producer_counter and self.consumer_counter.
What those 2 semaphors (as internal counters) do is just emulating limitation of the capacity of a resource (the resource is deque in our case)
How it starts and flows:

On running the producer starts appending random items in a loop of 100 iterations:

    ...
    for _ in range(self.times):
        self.queue.append(random.randint(0, 100))

on each iteration the producer thread will acquire self.remaining_space.acquire() decrementing the internal counter. When producer's counter is zero (0) the producer thread will block until the consumer thread calls self.remaining_space.release() thus increasing the producer's counter by 1 and letting the producer thread to proceed.
Such a "dragging" of acquire/release in multiple places can be just replaced with well-known queue.Queue, which is FIFO queue that already uses deque under the hood and already provides the needed locking mechanism and timeout logic.

The queue module implements multi-producer, multi-consumer queues. It is especially useful in threaded programming when information must be exchanged safely between multiple threads. The Queue class in this module implements all the required locking semantics.

...

class queue.Queue(maxsize=0)

Constructor for a FIFO queue. maxsize is an integer that sets the upperbound limit on the number of items that can be placed in the queue. Insertion will block once this size has been reached, until queue items are consumed. If maxsize is less than or equal to zero, the queue size is infinite.

So maxsize already gives us a limited capacity.

Namings:
A better names for a numeric constants in our new scheme would be:

MAX_QSIZE = 10  # max queue size
BUF_SIZE = 100  # total number of iterations/items to process

"Tricky" passing of TIMES constant:
It's not good to pass total number of processed items to Consumer constructor:

class Consumer:
    def __init__(self, queue, times=TIMES):
    ...

that allows compromising and specifying improper number of items to process.
Consider the following case:

TIMES = 100
...
    producer = Producer(queue)
    producer_thread = threading.Thread(target=producer.run)
    consumer = Consumer(queue, times=50)
    consumer_thread = threading.Thread(target=consumer.run)

this will point the consumer to consume less items than the queue contains, eventually having the queue with dangled items.
If you run that case you'll have the pipeline blocked/hanged.
To avoid that we need to ensure the consumer has gotten and processed all items in the queue:

class Consumer:
    def __init__(self, queue):
        self.queue = queue

    def run(self):
        while not self.queue.empty():
            item = self.queue.get()
            self.queue.task_done()
            print(item)

To enrich the queue communication we can incorporate queue.task_done() and queue.join() (the final version will incorporate them).

The final concise version:

from queue import Queue
import random
import threading

MAX_QSIZE = 10  # max queue size
BUF_SIZE = 100  # total number of iterations/items to process


class Producer:
    def __init__(self, queue, buf_size=BUF_SIZE):
        self.queue = queue
        self.buf_size = buf_size

    def run(self):
        for _ in range(self.buf_size):
            self.queue.put(random.randint(0, 100))


class Consumer:
    def __init__(self, queue):
        self.queue = queue

    def run(self):
        while not self.queue.empty():
            item = self.queue.get()
            self.queue.task_done()
            print(item)


def main():
    q = Queue(maxsize=MAX_QSIZE)

    producer = Producer(q)
    producer_thread = threading.Thread(target=producer.run)

    consumer = Consumer(q)
    consumer_thread = threading.Thread(target=consumer.run)

    producer_thread.start()
    consumer_thread.start()

    producer_thread.join()
    consumer_thread.join()
    q.join()


if __name__ == "__main__":
    main()