Added threading_test.py

Author: ReinierNL

Demos/Demo23/threading_test.py

@@ -0,0 +1,96 @@
+from threading import *
+from collections import deque
+from time import sleep
+
+def _test():
+
+ class BoundedQueue():
+
+ def __init__(self, limit):
+ self.mon = RLock()
+ self.rc = Condition(self.mon)
+ self.wc = Condition(self.mon)
+ self.limit = limit
+ self.queue = deque()
+
+ def put(self, item):
+ self.mon.acquire()
+ while len(self.queue) >= self.limit:
+ self._note("put(%s): queue full", item)
+ self.wc.wait()
+ self.queue.append(item)
+ self._note("put(%s): appended, length now %d",
+ item, len(self.queue))
+ self.rc.notify()
+ self.mon.release()
+
+ def get(self):
+ self.mon.acquire()
+ while not self.queue:
+ self._note("get(): queue empty")
+ self.rc.wait()
+ item = self.queue.popleft()
+ self._note("get(): got %s, %d left", item, len(self.queue))
+ self.wc.notify()
+ self.mon.release()
+ return item
+
+ def _note(self, format, *args):
+ format = format % args
+ ident = get_ident()
+ try:
+ name = current_thread().name
+ except KeyError:
+ name = "<OS thread %d>" % ident
+ format = "%s: %s" % (name, format)
+ print(format)
+
+ class ProducerThread(Thread):
+
+ def __init__(self, queue, quota):
+ Thread.__init__(self, name="Producer")
+ self.queue = queue
+ self.quota = quota
+
+ def run(self):
+ from random import random
+ counter = 0
+ while counter < self.quota:
+ counter = counter + 1
+ self.queue.put("%s.%d" % (self.name, counter))
+ sleep(random() * 0.00001)
+
+
+ class ConsumerThread(Thread):
+
+ def __init__(self, queue, count):
+ Thread.__init__(self, name="Consumer")
+ self.queue = queue
+ self.count = count
+
+ def run(self):
+ while self.count > 0:
+ item = self.queue.get()
+ print(item)
+ self.count = self.count - 1
+
+ NP = 3
+ QL = 4
+ NI = 5
+
+ Q = BoundedQueue(QL)
+ P = []
+ for i in range(NP):
+ t = ProducerThread(Q, NI)
+ t.name = ("Producer-%d" % (i+1))
+ P.append(t)
+ C = ConsumerThread(Q, NI*NP)
+ for t in P:
+ t.start()
+ sleep(0.000001)
+ C.start()
+ for t in P:
+ t.join()
+ C.join()
+
+_test()