Merge pull request #36 from realpython/jima/threading

Jima/threading

Author: dbader

intro-to-threading/daemon_thread.py

@@ -0,0 +1,23 @@
+#!/usr/bin/env python3
+import logging
+import threading
+import time
+
+
+def thread_function(name):
+    logging.info("Thread %s: starting", name)
+    time.sleep(2)
+    logging.info("Thread %s: finishing", name)
+
+
+if __name__ == "__main__":
+    format = "%(asctime)s: %(message)s"
+    logging.basicConfig(format=format, level=logging.INFO, datefmt="%H:%M:%S")
+
+    logging.info("Main    : before creating thread")
+    x = threading.Thread(target=thread_function, args=(1,), daemon=True)
+    logging.info("Main    : before running thread")
+    x.start()
+    logging.info("Main    : wait for the thread to finish")
+    x.join()
+    logging.info("Main    : all done")

intro-to-threading/executor.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python3
+import concurrent.futures
+import logging
+import time
+
+
+def thread_function(name):
+    logging.info("Thread %s: starting", name)
+    time.sleep(2)
+    logging.info("Thread %s: finishing", name)
+
+
+if __name__ == "__main__":
+    format = "%(asctime)s: %(message)s"
+    logging.basicConfig(format=format, level=logging.INFO, datefmt="%H:%M:%S")
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
+        executor.map(thread_function, range(3))

intro-to-threading/fixrace.py

@@ -0,0 +1,39 @@
+#!/usr/bin/env python3
+import concurrent.futures
+import logging
+import threading
+import time
+
+
+class FakeDatabase:
+    def __init__(self):
+        self.value = 0
+        self._lock = threading.Lock()
+
+    def locked_update(self, name):
+        logging.info("Thread %s: starting update", name)
+        logging.debug("Thread %s about to lock", name)
+        with self._lock:
+            logging.debug("Thread %s has lock", name)
+            local_copy = self.value
+            local_copy += 1
+            time.sleep(0.1)
+            self.value = local_copy
+            logging.debug("Thread %s about to release lock", name)
+        logging.debug("Thread %s after release", name)
+        logging.info("Thread %s: finishing update", name)
+
+
+if __name__ == "__main__":
+    format = "%(asctime)s: %(message)s"
+    logging.basicConfig(format=format, level=logging.INFO, datefmt="%H:%M:%S")
+    # logging.getLogger().setLevel(logging.DEBUG)
+
+    database = FakeDatabase()
+    logging.info(
+        "Testing locked update. Starting value is %d.", database.value
+    )
+    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
+        for index in range(2):
+            executor.submit(database.locked_update, index)
+    logging.info("Testing locked update. Ending value is %d.", database.value)

intro-to-threading/multiple_threads.py

@@ -0,0 +1,27 @@
+#!/usr/bin/env python3
+import logging
+import threading
+import time
+
+
+def thread_function(name):
+    logging.info("Thread %s: starting", name)
+    time.sleep(2)
+    logging.info("Thread %s: finishing", name)
+
+
+if __name__ == "__main__":
+    format = "%(asctime)s: %(message)s"
+    logging.basicConfig(format=format, level=logging.INFO, datefmt="%H:%M:%S")
+
+    threads = list()
+    for index in range(3):
+        logging.info("Main    : create and start thread %d.", index)
+        x = threading.Thread(target=thread_function, args=(index,))
+        threads.append(x)
+        x.start()
+
+    for index, thread in enumerate(threads):
+        logging.info("Main    : before joining thread %d.", index)
+        thread.join()
+        logging.info("Main    : thread %d done", index)

intro-to-threading/prodcom_better_queue.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+import concurrent.futures
+import logging
+import queue
+import random
+import threading
+import time
+
+
+def producer(queue, event):
+    """Pretend we're getting a number from the network."""
+    while not event.is_set():
+        message = random.randint(1, 101)
+        logging.info("Producer got message: %s", message)
+        queue.put(message)
+
+    logging.info("Producer received event. Exiting")
+
+
+def consumer(queue, event):
+    """ Pretend we're saving a number in the database. """
+    while not event.is_set() or not pipeline.empty():
+        message = queue.get()
+        logging.info(
+            "Consumer storing message: %s (size=%d)", message, queue.qsize()
+        )
+
+    logging.info("Consumer received event. Exiting")
+
+
+if __name__ == "__main__":
+    format = "%(asctime)s: %(message)s"
+    logging.basicConfig(format=format, level=logging.INFO, datefmt="%H:%M:%S")
+
+    pipeline = queue.Queue(maxsize=10)
+    event = threading.Event()
+    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
+        executor.submit(producer, pipeline, event)
+        executor.submit(consumer, pipeline, event)
+
+        time.sleep(0.1)
+        logging.info("Main: about to set event")
+        event.set()

intro-to-threading/prodcom_event.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+import concurrent.futures
+import logging
+import random
+import threading
+import time
+
+
+class Pipeline:
+    """Class to allow a single element pipeline between producer and consumer.
+    """
+
+    def __init__(self):
+        self.value = 0
+        self._set_lock = threading.Lock()
+        self._get_lock = threading.Lock()
+        self._get_lock.acquire()
+
+    def get_value(self, name):
+        logging.debug("%s:about to acquire getlock", name)
+        self._get_lock.acquire()
+        logging.debug("%s:have getlock", name)
+        value = self.value
+        logging.debug("%s:about to release setlock", name)
+        self._set_lock.release()
+        logging.debug("%s:setlock released", name)
+        return value
+
+    def set_value(self, value, name):
+        logging.debug("%s:about to acquire setlock", name)
+        self._set_lock.acquire()
+        logging.debug("%s:have setlock", name)
+        self.value = value
+        logging.debug("%s:about to release getlock", name)
+        self._get_lock.release()
+        logging.debug("%s:getlock released", name)
+
+
+def producer(pipeline, event):
+    """Pretend we're getting a number from the network."""
+    while not event.is_set():
+        new_datapoint = random.randint(1, 101)
+        # Sleep to simulate waiting for data from network
+        # time.sleep(float(new_datapoint)/100)
+        logging.info("Producer got data %d", new_datapoint)
+        pipeline.set_value(new_datapoint, "Producer")
+        if event.is_set():
+            logging.info("Producer received internal event. Exiting")
+
+        # don't put sleep here as this will cause the system to stall when the
+        # consumer is active.  That will result in deadlock.
+        # time.sleep(float(new_datapoint)/100)
+
+    logging.info("Producer received event. Exiting")
+
+
+def consumer(pipeline, event):
+    """ Pretend we're saving a number in the database. """
+    datapoint = 0
+    while not event.is_set():
+        datapoint = pipeline.get_value("Consumer")
+        logging.info("Consumer storing data: %d", datapoint)
+
+    logging.info("Consumer received event. Exiting")
+
+
+if __name__ == "__main__":
+    format = "%(asctime)s: %(message)s"
+    logging.basicConfig(format=format, level=logging.INFO, datefmt="%H:%M:%S")
+    # logging.getLogger().setLevel(logging.DEBUG)
+
+    pipeline = Pipeline()
+    event = threading.Event()
+    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
+        executor.submit(producer, pipeline, event)
+        executor.submit(consumer, pipeline, event)
+
+        time.sleep(0.1)
+        logging.info("Main: about to set event")
+        event.set()

intro-to-threading/prodcom_lock.py

@@ -0,0 +1,68 @@
+#!/usr/bin/env python3
+import concurrent.futures
+import logging
+import random
+import threading
+
+SENTINEL = -1
+
+
+class Pipeline:
+    """Class to allow a single element pipeline between producer and consumer.
+    """
+
+    def __init__(self):
+        self.message = 0
+        self.producer_lock = threading.Lock()
+        self.consumer_lock = threading.Lock()
+        self.consumer_lock.acquire()
+
+    def get_message(self, name):
+        logging.debug("%s:about to acquire getlock", name)
+        self.consumer_lock.acquire()
+        logging.debug("%s:have getlock", name)
+        message = self.message
+        logging.debug("%s:about to release setlock", name)
+        self.producer_lock.release()
+        logging.debug("%s:setlock released", name)
+        return message
+
+    def set_message(self, message, name):
+        logging.debug("%s:about to acquire setlock", name)
+        self.producer_lock.acquire()
+        logging.debug("%s:have setlock", name)
+        self.message = message
+        logging.debug("%s:about to release getlock", name)
+        self.consumer_lock.release()
+        logging.debug("%s:getlock released", name)
+
+
+def producer(pipeline):
+    """Pretend we're getting a message from the network."""
+    for index in range(10):
+        message = random.randint(1, 101)
+        logging.info("Producer got message: %s", message)
+        pipeline.set_message(message, "Producer")
+
+    # Send a sentinel message to tell consumer we're done
+    pipeline.set_message(SENTINEL, "Producer")
+
+
+def consumer(pipeline):
+    """ Pretend we're saving a number in the database. """
+    message = 0
+    while message != SENTINEL:
+        message = pipeline.get_message("Consumer")
+        if message != SENTINEL:
+            logging.info("Consumer storing message: %s", message)
+
+
+if __name__ == "__main__":
+    format = "%(asctime)s: %(message)s"
+    logging.basicConfig(format=format, level=logging.INFO, datefmt="%H:%M:%S")
+    # logging.getLogger().setLevel(logging.DEBUG)
+
+    pipeline = Pipeline()
+    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
+        executor.submit(producer, pipeline)
+        executor.submit(consumer, pipeline)

intro-to-threading/prodcom_queue.py

@@ -0,0 +1,62 @@
+#!/usr/bin/env python3
+import concurrent.futures
+import logging
+import queue
+import random
+import threading
+import time
+
+
+class Pipeline(queue.Queue):
+    def __init__(self):
+        super().__init__(maxsize=10)
+
+    def get_message(self, name):
+        logging.debug("%s:about to get from queue", name)
+        value = self.get()
+        logging.debug("%s:got %d from queue", name, value)
+        return value
+
+    def set_message(self, value, name):
+        logging.debug("%s:about to add %d to queue", name, value)
+        self.put(value)
+        logging.debug("%s:added %d to queue", name, value)
+
+
+def producer(pipeline, event):
+    """Pretend we're getting a number from the network."""
+    while not event.is_set():
+        message = random.randint(1, 101)
+        logging.info("Producer got message: %s", message)
+        pipeline.set_message(message, "Producer")
+
+    logging.info("Producer received EXIT event. Exiting")
+
+
+def consumer(pipeline, event):
+    """ Pretend we're saving a number in the database. """
+    while not event.is_set() or not pipeline.empty():
+        message = pipeline.get_message("Consumer")
+        logging.info(
+            "Consumer storing message: %s  (queue size=%s)",
+            message,
+            pipeline.qsize(),
+        )
+
+    logging.info("Consumer received EXIT event. Exiting")
+
+
+if __name__ == "__main__":
+    format = "%(asctime)s: %(message)s"
+    logging.basicConfig(format=format, level=logging.INFO, datefmt="%H:%M:%S")
+    # logging.getLogger().setLevel(logging.DEBUG)
+
+    pipeline = Pipeline()
+    event = threading.Event()
+    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
+        executor.submit(producer, pipeline, event)
+        executor.submit(consumer, pipeline, event)
+
+        time.sleep(0.1)
+        logging.info("Main: about to set event")
+        event.set()

intro-to-threading/racecond.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python3
+import concurrent.futures
+import logging
+import time
+
+
+class FakeDatabase:
+    def __init__(self):
+        self.value = 0
+
+    def update(self, name):
+        logging.info("Thread %s: starting update", name)
+        local_copy = self.value
+        local_copy += 1
+        time.sleep(0.1)
+        self.value = local_copy
+        logging.info("Thread %s: finishing update", name)
+
+
+if __name__ == "__main__":
+    format = "%(asctime)s: %(message)s"
+    logging.basicConfig(format=format, level=logging.INFO, datefmt="%H:%M:%S")
+
+    database = FakeDatabase()
+    logging.info("Testing update. Starting value is %d.", database.value)
+    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
+        for index in range(2):
+            executor.submit(database.update, index)
+    logging.info("Testing update. Ending value is %d.", database.value)