summaryrefslogtreecommitdiffstats
path: root/Demo/threads
diff options
context:
space:
mode:
Diffstat (limited to 'Demo/threads')
-rw-r--r--Demo/threads/Coroutine.py159
-rw-r--r--Demo/threads/Generator.py92
-rw-r--r--Demo/threads/README11
-rw-r--r--Demo/threads/fcmp.py64
-rw-r--r--Demo/threads/find.py155
-rw-r--r--Demo/threads/squasher.py105
-rw-r--r--Demo/threads/sync.py603
-rw-r--r--Demo/threads/telnet.py114
8 files changed, 1303 insertions, 0 deletions
diff --git a/Demo/threads/Coroutine.py b/Demo/threads/Coroutine.py
new file mode 100644
index 0000000..5de2b62
--- /dev/null
+++ b/Demo/threads/Coroutine.py
@@ -0,0 +1,159 @@
+# Coroutine implementation using Python threads.
+#
+# Combines ideas from Guido's Generator module, and from the coroutine
+# features of Icon and Simula 67.
+#
+# To run a collection of functions as coroutines, you need to create
+# a Coroutine object to control them:
+# co = Coroutine()
+# and then 'create' a subsidiary object for each function in the
+# collection:
+# cof1 = co.create(f1 [, arg1, arg2, ...]) # [] means optional,
+# cof2 = co.create(f2 [, arg1, arg2, ...]) #... not list
+# cof3 = co.create(f3 [, arg1, arg2, ...])
+# etc. The functions need not be distinct; 'create'ing the same
+# function multiple times gives you independent instances of the
+# function.
+#
+# To start the coroutines running, use co.tran on one of the create'd
+# functions; e.g., co.tran(cof2). The routine that first executes
+# co.tran is called the "main coroutine". It's special in several
+# respects: it existed before you created the Coroutine object; if any of
+# the create'd coroutines exits (does a return, or suffers an unhandled
+# exception), EarlyExit error is raised in the main coroutine; and the
+# co.detach() method transfers control directly to the main coroutine
+# (you can't use co.tran() for this because the main coroutine doesn't
+# have a name ...).
+#
+# Coroutine objects support these methods:
+#
+# handle = .create(func [, arg1, arg2, ...])
+# Creates a coroutine for an invocation of func(arg1, arg2, ...),
+# and returns a handle ("name") for the coroutine so created. The
+# handle can be used as the target in a subsequent .tran().
+#
+# .tran(target, data=None)
+# Transfer control to the create'd coroutine "target", optionally
+# passing it an arbitrary piece of data. To the coroutine A that does
+# the .tran, .tran acts like an ordinary function call: another
+# coroutine B can .tran back to it later, and if it does A's .tran
+# returns the 'data' argument passed to B's tran. E.g.,
+#
+# in coroutine coA in coroutine coC in coroutine coB
+# x = co.tran(coC) co.tran(coB) co.tran(coA,12)
+# print x # 12
+#
+# The data-passing feature is taken from Icon, and greatly cuts
+# the need to use global variables for inter-coroutine communication.
+#
+# .back( data=None )
+# The same as .tran(invoker, data=None), where 'invoker' is the
+# coroutine that most recently .tran'ed control to the coroutine
+# doing the .back. This is akin to Icon's "&source".
+#
+# .detach( data=None )
+# The same as .tran(main, data=None), where 'main' is the
+# (unnameable!) coroutine that started it all. 'main' has all the
+# rights of any other coroutine: upon receiving control, it can
+# .tran to an arbitrary coroutine of its choosing, go .back to
+# the .detach'er, or .kill the whole thing.
+#
+# .kill()
+# Destroy all the coroutines, and return control to the main
+# coroutine. None of the create'ed coroutines can be resumed after a
+# .kill(). An EarlyExit exception does a .kill() automatically. It's
+# a good idea to .kill() coroutines you're done with, since the
+# current implementation consumes a thread for each coroutine that
+# may be resumed.
+
+import thread
+import sync
+
+class _CoEvent:
+ def __init__(self, func):
+ self.f = func
+ self.e = sync.event()
+
+ def __repr__(self):
+ if self.f is None:
+ return 'main coroutine'
+ else:
+ return 'coroutine for func ' + self.f.func_name
+
+ def __hash__(self):
+ return id(self)
+
+ def __cmp__(x,y):
+ return cmp(id(x), id(y))
+
+ def resume(self):
+ self.e.post()
+
+ def wait(self):
+ self.e.wait()
+ self.e.clear()
+
+class Killed(Exception): pass
+class EarlyExit(Exception): pass
+
+class Coroutine:
+ def __init__(self):
+ self.active = self.main = _CoEvent(None)
+ self.invokedby = {self.main: None}
+ self.killed = 0
+ self.value = None
+ self.terminated_by = None
+
+ def create(self, func, *args):
+ me = _CoEvent(func)
+ self.invokedby[me] = None
+ thread.start_new_thread(self._start, (me,) + args)
+ return me
+
+ def _start(self, me, *args):
+ me.wait()
+ if not self.killed:
+ try:
+ try:
+ apply(me.f, args)
+ except Killed:
+ pass
+ finally:
+ if not self.killed:
+ self.terminated_by = me
+ self.kill()
+
+ def kill(self):
+ if self.killed:
+ raise TypeError, 'kill() called on dead coroutines'
+ self.killed = 1
+ for coroutine in self.invokedby.keys():
+ coroutine.resume()
+
+ def back(self, data=None):
+ return self.tran( self.invokedby[self.active], data )
+
+ def detach(self, data=None):
+ return self.tran( self.main, data )
+
+ def tran(self, target, data=None):
+ if not self.invokedby.has_key(target):
+ raise TypeError, '.tran target %r is not an active coroutine' % (target,)
+ if self.killed:
+ raise TypeError, '.tran target %r is killed' % (target,)
+ self.value = data
+ me = self.active
+ self.invokedby[target] = me
+ self.active = target
+ target.resume()
+
+ me.wait()
+ if self.killed:
+ if self.main is not me:
+ raise Killed
+ if self.terminated_by is not None:
+ raise EarlyExit, '%r terminated early' % (self.terminated_by,)
+
+ return self.value
+
+# end of module
diff --git a/Demo/threads/Generator.py b/Demo/threads/Generator.py
new file mode 100644
index 0000000..2e814a0
--- /dev/null
+++ b/Demo/threads/Generator.py
@@ -0,0 +1,92 @@
+# Generator implementation using threads
+
+import sys
+import thread
+
+class Killed(Exception):
+ pass
+
+class Generator:
+ # Constructor
+ def __init__(self, func, args):
+ self.getlock = thread.allocate_lock()
+ self.putlock = thread.allocate_lock()
+ self.getlock.acquire()
+ self.putlock.acquire()
+ self.func = func
+ self.args = args
+ self.done = 0
+ self.killed = 0
+ thread.start_new_thread(self._start, ())
+
+ # Internal routine
+ def _start(self):
+ try:
+ self.putlock.acquire()
+ if not self.killed:
+ try:
+ apply(self.func, (self,) + self.args)
+ except Killed:
+ pass
+ finally:
+ if not self.killed:
+ self.done = 1
+ self.getlock.release()
+
+ # Called by producer for each value; raise Killed if no more needed
+ def put(self, value):
+ if self.killed:
+ raise TypeError, 'put() called on killed generator'
+ self.value = value
+ self.getlock.release() # Resume consumer thread
+ self.putlock.acquire() # Wait for next get() call
+ if self.killed:
+ raise Killed
+
+ # Called by producer to get next value; raise EOFError if no more
+ def get(self):
+ if self.killed:
+ raise TypeError, 'get() called on killed generator'
+ self.putlock.release() # Resume producer thread
+ self.getlock.acquire() # Wait for value to appear
+ if self.done:
+ raise EOFError # Say there are no more values
+ return self.value
+
+ # Called by consumer if no more values wanted
+ def kill(self):
+ if self.killed:
+ raise TypeError, 'kill() called on killed generator'
+ self.killed = 1
+ self.putlock.release()
+
+ # Clone constructor
+ def clone(self):
+ return Generator(self.func, self.args)
+
+def pi(g):
+ k, a, b, a1, b1 = 2L, 4L, 1L, 12L, 4L
+ while 1:
+ # Next approximation
+ p, q, k = k*k, 2L*k+1L, k+1L
+ a, b, a1, b1 = a1, b1, p*a+q*a1, p*b+q*b1
+ # Print common digits
+ d, d1 = a//b, a1//b1
+ while d == d1:
+ g.put(int(d))
+ a, a1 = 10L*(a%b), 10L*(a1%b1)
+ d, d1 = a//b, a1//b1
+
+def test():
+ g = Generator(pi, ())
+ g.kill()
+ g = Generator(pi, ())
+ for i in range(10): print g.get(),
+ print
+ h = g.clone()
+ g.kill()
+ while 1:
+ print h.get(),
+ sys.stdout.flush()
+
+test()
diff --git a/Demo/threads/README b/Demo/threads/README
new file mode 100644
index 0000000..a379521
--- /dev/null
+++ b/Demo/threads/README
@@ -0,0 +1,11 @@
+This directory contains some demonstrations of the thread module.
+
+These are mostly "proof of concept" type applications:
+
+Generator.py Generator class implemented with threads.
+sync.py Condition variables primitives by Tim Peters.
+telnet.py Version of ../sockets/telnet.py using threads.
+
+Coroutine.py Coroutines using threads, by Tim Peters (22 May 94)
+fcmp.py Example of above, by Tim
+squasher.py Another example of above, also by Tim
diff --git a/Demo/threads/fcmp.py b/Demo/threads/fcmp.py
new file mode 100644
index 0000000..27af76d
--- /dev/null
+++ b/Demo/threads/fcmp.py
@@ -0,0 +1,64 @@
+# Coroutine example: controlling multiple instances of a single function
+
+from Coroutine import *
+
+# fringe visits a nested list in inorder, and detaches for each non-list
+# element; raises EarlyExit after the list is exhausted
+def fringe(co, list):
+ for x in list:
+ if type(x) is type([]):
+ fringe(co, x)
+ else:
+ co.back(x)
+
+def printinorder(list):
+ co = Coroutine()
+ f = co.create(fringe, co, list)
+ try:
+ while 1:
+ print co.tran(f),
+ except EarlyExit:
+ pass
+ print
+
+printinorder([1,2,3]) # 1 2 3
+printinorder([[[[1,[2]]],3]]) # ditto
+x = [0, 1, [2, [3]], [4,5], [[[6]]] ]
+printinorder(x) # 0 1 2 3 4 5 6
+
+# fcmp lexicographically compares the fringes of two nested lists
+def fcmp(l1, l2):
+ co1 = Coroutine(); f1 = co1.create(fringe, co1, l1)
+ co2 = Coroutine(); f2 = co2.create(fringe, co2, l2)
+ while 1:
+ try:
+ v1 = co1.tran(f1)
+ except EarlyExit:
+ try:
+ v2 = co2.tran(f2)
+ except EarlyExit:
+ return 0
+ co2.kill()
+ return -1
+ try:
+ v2 = co2.tran(f2)
+ except EarlyExit:
+ co1.kill()
+ return 1
+ if v1 != v2:
+ co1.kill(); co2.kill()
+ return cmp(v1,v2)
+
+print fcmp(range(7), x) # 0; fringes are equal
+print fcmp(range(6), x) # -1; 1st list ends early
+print fcmp(x, range(6)) # 1; 2nd list ends early
+print fcmp(range(8), x) # 1; 2nd list ends early
+print fcmp(x, range(8)) # -1; 1st list ends early
+print fcmp([1,[[2],8]],
+ [[[1],2],8]) # 0
+print fcmp([1,[[3],8]],
+ [[[1],2],8]) # 1
+print fcmp([1,[[2],8]],
+ [[[1],2],9]) # -1
+
+# end of example
diff --git a/Demo/threads/find.py b/Demo/threads/find.py
new file mode 100644
index 0000000..7d5edc1
--- /dev/null
+++ b/Demo/threads/find.py
@@ -0,0 +1,155 @@
+# A parallelized "find(1)" using the thread module.
+
+# This demonstrates the use of a work queue and worker threads.
+# It really does do more stats/sec when using multiple threads,
+# although the improvement is only about 20-30 percent.
+# (That was 8 years ago. In 2002, on Linux, I can't measure
+# a speedup. :-( )
+
+# I'm too lazy to write a command line parser for the full find(1)
+# command line syntax, so the predicate it searches for is wired-in,
+# see function selector() below. (It currently searches for files with
+# world write permission.)
+
+# Usage: parfind.py [-w nworkers] [directory] ...
+# Default nworkers is 4
+
+
+import sys
+import getopt
+import string
+import time
+import os
+from stat import *
+import thread
+
+
+# Work queue class. Usage:
+# wq = WorkQ()
+# wq.addwork(func, (arg1, arg2, ...)) # one or more calls
+# wq.run(nworkers)
+# The work is done when wq.run() completes.
+# The function calls executed by the workers may add more work.
+# Don't use keyboard interrupts!
+
+class WorkQ:
+
+ # Invariants:
+
+ # - busy and work are only modified when mutex is locked
+ # - len(work) is the number of jobs ready to be taken
+ # - busy is the number of jobs being done
+ # - todo is locked iff there is no work and somebody is busy
+
+ def __init__(self):
+ self.mutex = thread.allocate()
+ self.todo = thread.allocate()
+ self.todo.acquire()
+ self.work = []
+ self.busy = 0
+
+ def addwork(self, func, args):
+ job = (func, args)
+ self.mutex.acquire()
+ self.work.append(job)
+ self.mutex.release()
+ if len(self.work) == 1:
+ self.todo.release()
+
+ def _getwork(self):
+ self.todo.acquire()
+ self.mutex.acquire()
+ if self.busy == 0 and len(self.work) == 0:
+ self.mutex.release()
+ self.todo.release()
+ return None
+ job = self.work[0]
+ del self.work[0]
+ self.busy = self.busy + 1
+ self.mutex.release()
+ if len(self.work) > 0:
+ self.todo.release()
+ return job
+
+ def _donework(self):
+ self.mutex.acquire()
+ self.busy = self.busy - 1
+ if self.busy == 0 and len(self.work) == 0:
+ self.todo.release()
+ self.mutex.release()
+
+ def _worker(self):
+ time.sleep(0.00001) # Let other threads run
+ while 1:
+ job = self._getwork()
+ if not job:
+ break
+ func, args = job
+ apply(func, args)
+ self._donework()
+
+ def run(self, nworkers):
+ if not self.work:
+ return # Nothing to do
+ for i in range(nworkers-1):
+ thread.start_new(self._worker, ())
+ self._worker()
+ self.todo.acquire()
+
+
+# Main program
+
+def main():
+ nworkers = 4
+ opts, args = getopt.getopt(sys.argv[1:], '-w:')
+ for opt, arg in opts:
+ if opt == '-w':
+ nworkers = string.atoi(arg)
+ if not args:
+ args = [os.curdir]
+
+ wq = WorkQ()
+ for dir in args:
+ wq.addwork(find, (dir, selector, wq))
+
+ t1 = time.time()
+ wq.run(nworkers)
+ t2 = time.time()
+
+ sys.stderr.write('Total time %r sec.\n' % (t2-t1))
+
+
+# The predicate -- defines what files we look for.
+# Feel free to change this to suit your purpose
+
+def selector(dir, name, fullname, stat):
+ # Look for world writable files that are not symlinks
+ return (stat[ST_MODE] & 0002) != 0 and not S_ISLNK(stat[ST_MODE])
+
+
+# The find procedure -- calls wq.addwork() for subdirectories
+
+def find(dir, pred, wq):
+ try:
+ names = os.listdir(dir)
+ except os.error, msg:
+ print repr(dir), ':', msg
+ return
+ for name in names:
+ if name not in (os.curdir, os.pardir):
+ fullname = os.path.join(dir, name)
+ try:
+ stat = os.lstat(fullname)
+ except os.error, msg:
+ print repr(fullname), ':', msg
+ continue
+ if pred(dir, name, fullname, stat):
+ print fullname
+ if S_ISDIR(stat[ST_MODE]):
+ if not os.path.ismount(fullname):
+ wq.addwork(find, (fullname, pred, wq))
+
+
+# Call the main program
+
+main()
diff --git a/Demo/threads/squasher.py b/Demo/threads/squasher.py
new file mode 100644
index 0000000..0d59cb8
--- /dev/null
+++ b/Demo/threads/squasher.py
@@ -0,0 +1,105 @@
+# Coroutine example: general coroutine transfers
+#
+# The program is a variation of a Simula 67 program due to Dahl & Hoare,
+# (Dahl/Dijkstra/Hoare, Structured Programming; Academic Press, 1972)
+# who in turn credit the original example to Conway.
+#
+# We have a number of input lines, terminated by a 0 byte. The problem
+# is to squash them together into output lines containing 72 characters
+# each. A semicolon must be added between input lines. Runs of blanks
+# and tabs in input lines must be squashed into single blanks.
+# Occurrences of "**" in input lines must be replaced by "^".
+#
+# Here's a test case:
+
+test = """\
+ d = sqrt(b**2 - 4*a*c)
+twoa = 2*a
+ L = -b/twoa
+ R = d/twoa
+ A1 = L + R
+ A2 = L - R\0
+"""
+
+# The program should print:
+
+# d = sqrt(b^2 - 4*a*c);twoa = 2*a; L = -b/twoa; R = d/twoa; A1 = L + R;
+#A2 = L - R
+#done
+
+# getline: delivers the next input line to its invoker
+# disassembler: grabs input lines from getline, and delivers them one
+# character at a time to squasher, also inserting a semicolon into
+# the stream between lines
+# squasher: grabs characters from disassembler and passes them on to
+# assembler, first replacing "**" with "^" and squashing runs of
+# whitespace
+# assembler: grabs characters from squasher and packs them into lines
+# with 72 character each, delivering each such line to putline;
+# when it sees a null byte, passes the last line to putline and
+# then kills all the coroutines
+# putline: grabs lines from assembler, and just prints them
+
+from Coroutine import *
+
+def getline(text):
+ for line in string.splitfields(text, '\n'):
+ co.tran(codisassembler, line)
+
+def disassembler():
+ while 1:
+ card = co.tran(cogetline)
+ for i in range(len(card)):
+ co.tran(cosquasher, card[i])
+ co.tran(cosquasher, ';')
+
+def squasher():
+ while 1:
+ ch = co.tran(codisassembler)
+ if ch == '*':
+ ch2 = co.tran(codisassembler)
+ if ch2 == '*':
+ ch = '^'
+ else:
+ co.tran(coassembler, ch)
+ ch = ch2
+ if ch in ' \t':
+ while 1:
+ ch2 = co.tran(codisassembler)
+ if ch2 not in ' \t':
+ break
+ co.tran(coassembler, ' ')
+ ch = ch2
+ co.tran(coassembler, ch)
+
+def assembler():
+ line = ''
+ while 1:
+ ch = co.tran(cosquasher)
+ if ch == '\0':
+ break
+ if len(line) == 72:
+ co.tran(coputline, line)
+ line = ''
+ line = line + ch
+ line = line + ' ' * (72 - len(line))
+ co.tran(coputline, line)
+ co.kill()
+
+def putline():
+ while 1:
+ line = co.tran(coassembler)
+ print line
+
+import string
+co = Coroutine()
+cogetline = co.create(getline, test)
+coputline = co.create(putline)
+coassembler = co.create(assembler)
+codisassembler = co.create(disassembler)
+cosquasher = co.create(squasher)
+
+co.tran(coputline)
+print 'done'
+
+# end of example
diff --git a/Demo/threads/sync.py b/Demo/threads/sync.py
new file mode 100644
index 0000000..a344abe
--- /dev/null
+++ b/Demo/threads/sync.py
@@ -0,0 +1,603 @@
+# Defines classes that provide synchronization objects. Note that use of
+# this module requires that your Python support threads.
+#
+# condition(lock=None) # a POSIX-like condition-variable object
+# barrier(n) # an n-thread barrier
+# event() # an event object
+# semaphore(n=1) # a semaphore object, with initial count n
+# mrsw() # a multiple-reader single-writer lock
+#
+# CONDITIONS
+#
+# A condition object is created via
+# import this_module
+# your_condition_object = this_module.condition(lock=None)
+#
+# As explained below, a condition object has a lock associated with it,
+# used in the protocol to protect condition data. You can specify a
+# lock to use in the constructor, else the constructor will allocate
+# an anonymous lock for you. Specifying a lock explicitly can be useful
+# when more than one condition keys off the same set of shared data.
+#
+# Methods:
+# .acquire()
+# acquire the lock associated with the condition
+# .release()
+# release the lock associated with the condition
+# .wait()
+# block the thread until such time as some other thread does a
+# .signal or .broadcast on the same condition, and release the
+# lock associated with the condition. The lock associated with
+# the condition MUST be in the acquired state at the time
+# .wait is invoked.
+# .signal()
+# wake up exactly one thread (if any) that previously did a .wait
+# on the condition; that thread will awaken with the lock associated
+# with the condition in the acquired state. If no threads are
+# .wait'ing, this is a nop. If more than one thread is .wait'ing on
+# the condition, any of them may be awakened.
+# .broadcast()
+# wake up all threads (if any) that are .wait'ing on the condition;
+# the threads are woken up serially, each with the lock in the
+# acquired state, so should .release() as soon as possible. If no
+# threads are .wait'ing, this is a nop.
+#
+# Note that if a thread does a .wait *while* a signal/broadcast is
+# in progress, it's guaranteeed to block until a subsequent
+# signal/broadcast.
+#
+# Secret feature: `broadcast' actually takes an integer argument,
+# and will wake up exactly that many waiting threads (or the total
+# number waiting, if that's less). Use of this is dubious, though,
+# and probably won't be supported if this form of condition is
+# reimplemented in C.
+#
+# DIFFERENCES FROM POSIX
+#
+# + A separate mutex is not needed to guard condition data. Instead, a
+# condition object can (must) be .acquire'ed and .release'ed directly.
+# This eliminates a common error in using POSIX conditions.
+#
+# + Because of implementation difficulties, a POSIX `signal' wakes up
+# _at least_ one .wait'ing thread. Race conditions make it difficult
+# to stop that. This implementation guarantees to wake up only one,
+# but you probably shouldn't rely on that.
+#
+# PROTOCOL
+#
+# Condition objects are used to block threads until "some condition" is
+# true. E.g., a thread may wish to wait until a producer pumps out data
+# for it to consume, or a server may wish to wait until someone requests
+# its services, or perhaps a whole bunch of threads want to wait until a
+# preceding pass over the data is complete. Early models for conditions
+# relied on some other thread figuring out when a blocked thread's
+# condition was true, and made the other thread responsible both for
+# waking up the blocked thread and guaranteeing that it woke up with all
+# data in a correct state. This proved to be very delicate in practice,
+# and gave conditions a bad name in some circles.
+#
+# The POSIX model addresses these problems by making a thread responsible
+# for ensuring that its own state is correct when it wakes, and relies
+# on a rigid protocol to make this easy; so long as you stick to the
+# protocol, POSIX conditions are easy to "get right":
+#
+# A) The thread that's waiting for some arbitrarily-complex condition
+# (ACC) to become true does:
+#
+# condition.acquire()
+# while not (code to evaluate the ACC):
+# condition.wait()
+# # That blocks the thread, *and* releases the lock. When a
+# # condition.signal() happens, it will wake up some thread that
+# # did a .wait, *and* acquire the lock again before .wait
+# # returns.
+# #
+# # Because the lock is acquired at this point, the state used
+# # in evaluating the ACC is frozen, so it's safe to go back &
+# # reevaluate the ACC.
+#
+# # At this point, ACC is true, and the thread has the condition
+# # locked.
+# # So code here can safely muck with the shared state that
+# # went into evaluating the ACC -- if it wants to.
+# # When done mucking with the shared state, do
+# condition.release()
+#
+# B) Threads that are mucking with shared state that may affect the
+# ACC do:
+#
+# condition.acquire()
+# # muck with shared state
+# condition.release()
+# if it's possible that ACC is true now:
+# condition.signal() # or .broadcast()
+#
+# Note: You may prefer to put the "if" clause before the release().
+# That's fine, but do note that anyone waiting on the signal will
+# stay blocked until the release() is done (since acquiring the
+# condition is part of what .wait() does before it returns).
+#
+# TRICK OF THE TRADE
+#
+# With simpler forms of conditions, it can be impossible to know when
+# a thread that's supposed to do a .wait has actually done it. But
+# because this form of condition releases a lock as _part_ of doing a
+# wait, the state of that lock can be used to guarantee it.
+#
+# E.g., suppose thread A spawns thread B and later wants to wait for B to
+# complete:
+#
+# In A: In B:
+#
+# B_done = condition() ... do work ...
+# B_done.acquire() B_done.acquire(); B_done.release()
+# spawn B B_done.signal()
+# ... some time later ... ... and B exits ...
+# B_done.wait()
+#
+# Because B_done was in the acquire'd state at the time B was spawned,
+# B's attempt to acquire B_done can't succeed until A has done its
+# B_done.wait() (which releases B_done). So B's B_done.signal() is
+# guaranteed to be seen by the .wait(). Without the lock trick, B
+# may signal before A .waits, and then A would wait forever.
+#
+# BARRIERS
+#
+# A barrier object is created via
+# import this_module
+# your_barrier = this_module.barrier(num_threads)
+#
+# Methods:
+# .enter()
+# the thread blocks until num_threads threads in all have done
+# .enter(). Then the num_threads threads that .enter'ed resume,
+# and the barrier resets to capture the next num_threads threads
+# that .enter it.
+#
+# EVENTS
+#
+# An event object is created via
+# import this_module
+# your_event = this_module.event()
+#
+# An event has two states, `posted' and `cleared'. An event is
+# created in the cleared state.
+#
+# Methods:
+#
+# .post()
+# Put the event in the posted state, and resume all threads
+# .wait'ing on the event (if any).
+#
+# .clear()
+# Put the event in the cleared state.
+#
+# .is_posted()
+# Returns 0 if the event is in the cleared state, or 1 if the event
+# is in the posted state.
+#
+# .wait()
+# If the event is in the posted state, returns immediately.
+# If the event is in the cleared state, blocks the calling thread
+# until the event is .post'ed by another thread.
+#
+# Note that an event, once posted, remains posted until explicitly
+# cleared. Relative to conditions, this is both the strength & weakness
+# of events. It's a strength because the .post'ing thread doesn't have to
+# worry about whether the threads it's trying to communicate with have
+# already done a .wait (a condition .signal is seen only by threads that
+# do a .wait _prior_ to the .signal; a .signal does not persist). But
+# it's a weakness because .clear'ing an event is error-prone: it's easy
+# to mistakenly .clear an event before all the threads you intended to
+# see the event get around to .wait'ing on it. But so long as you don't
+# need to .clear an event, events are easy to use safely.
+#
+# SEMAPHORES
+#
+# A semaphore object is created via
+# import this_module
+# your_semaphore = this_module.semaphore(count=1)
+#
+# A semaphore has an integer count associated with it. The initial value
+# of the count is specified by the optional argument (which defaults to
+# 1) passed to the semaphore constructor.
+#
+# Methods:
+#
+# .p()
+# If the semaphore's count is greater than 0, decrements the count
+# by 1 and returns.
+# Else if the semaphore's count is 0, blocks the calling thread
+# until a subsequent .v() increases the count. When that happens,
+# the count will be decremented by 1 and the calling thread resumed.
+#
+# .v()
+# Increments the semaphore's count by 1, and wakes up a thread (if
+# any) blocked by a .p(). It's an (detected) error for a .v() to
+# increase the semaphore's count to a value larger than the initial
+# count.
+#
+# MULTIPLE-READER SINGLE-WRITER LOCKS
+#
+# A mrsw lock is created via
+# import this_module
+# your_mrsw_lock = this_module.mrsw()
+#
+# This kind of lock is often useful with complex shared data structures.
+# The object lets any number of "readers" proceed, so long as no thread
+# wishes to "write". When a (one or more) thread declares its intention
+# to "write" (e.g., to update a shared structure), all current readers
+# are allowed to finish, and then a writer gets exclusive access; all
+# other readers & writers are blocked until the current writer completes.
+# Finally, if some thread is waiting to write and another is waiting to
+# read, the writer takes precedence.
+#
+# Methods:
+#
+# .read_in()
+# If no thread is writing or waiting to write, returns immediately.
+# Else blocks until no thread is writing or waiting to write. So
+# long as some thread has completed a .read_in but not a .read_out,
+# writers are blocked.
+#
+# .read_out()
+# Use sometime after a .read_in to declare that the thread is done
+# reading. When all threads complete reading, a writer can proceed.
+#
+# .write_in()
+# If no thread is writing (has completed a .write_in, but hasn't yet
+# done a .write_out) or reading (similarly), returns immediately.
+# Else blocks the calling thread, and threads waiting to read, until
+# the current writer completes writing or all the current readers
+# complete reading; if then more than one thread is waiting to
+# write, one of them is allowed to proceed, but which one is not
+# specified.
+#
+# .write_out()
+# Use sometime after a .write_in to declare that the thread is done
+# writing. Then if some other thread is waiting to write, it's
+# allowed to proceed. Else all threads (if any) waiting to read are
+# allowed to proceed.
+#
+# .write_to_read()
+# Use instead of a .write_in to declare that the thread is done
+# writing but wants to continue reading without other writers
+# intervening. If there are other threads waiting to write, they
+# are allowed to proceed only if the current thread calls
+# .read_out; threads waiting to read are only allowed to proceed
+# if there are no threads waiting to write. (This is a
+# weakness of the interface!)
+
+import thread
+
+class condition:
+ def __init__(self, lock=None):
+ # the lock actually used by .acquire() and .release()
+ if lock is None:
+ self.mutex = thread.allocate_lock()
+ else:
+ if hasattr(lock, 'acquire') and \
+ hasattr(lock, 'release'):
+ self.mutex = lock
+ else:
+ raise TypeError, 'condition constructor requires ' \
+ 'a lock argument'
+
+ # lock used to block threads until a signal
+ self.checkout = thread.allocate_lock()
+ self.checkout.acquire()
+
+ # internal critical-section lock, & the data it protects
+ self.idlock = thread.allocate_lock()
+ self.id = 0
+ self.waiting = 0 # num waiters subject to current release
+ self.pending = 0 # num waiters awaiting next signal
+ self.torelease = 0 # num waiters to release
+ self.releasing = 0 # 1 iff release is in progress
+
+ def acquire(self):
+ self.mutex.acquire()
+
+ def release(self):
+ self.mutex.release()
+
+ def wait(self):
+ mutex, checkout, idlock = self.mutex, self.checkout, self.idlock
+ if not mutex.locked():
+ raise ValueError, \
+ "condition must be .acquire'd when .wait() invoked"
+
+ idlock.acquire()
+ myid = self.id
+ self.pending = self.pending + 1
+ idlock.release()
+
+ mutex.release()
+
+ while 1:
+ checkout.acquire(); idlock.acquire()
+ if myid < self.id:
+ break
+ checkout.release(); idlock.release()
+
+ self.waiting = self.waiting - 1
+ self.torelease = self.torelease - 1
+ if self.torelease:
+ checkout.release()
+ else:
+ self.releasing = 0
+ if self.waiting == self.pending == 0:
+ self.id = 0
+ idlock.release()
+ mutex.acquire()
+
+ def signal(self):
+ self.broadcast(1)
+
+ def broadcast(self, num = -1):
+ if num < -1:
+ raise ValueError, '.broadcast called with num %r' % (num,)
+ if num == 0:
+ return
+ self.idlock.acquire()
+ if self.pending:
+ self.waiting = self.waiting + self.pending
+ self.pending = 0
+ self.id = self.id + 1
+ if num == -1:
+ self.torelease = self.waiting
+ else:
+ self.torelease = min( self.waiting,
+ self.torelease + num )
+ if self.torelease and not self.releasing:
+ self.releasing = 1
+ self.checkout.release()
+ self.idlock.release()
+
+class barrier:
+ def __init__(self, n):
+ self.n = n
+ self.togo = n
+ self.full = condition()
+
+ def enter(self):
+ full = self.full
+ full.acquire()
+ self.togo = self.togo - 1
+ if self.togo:
+ full.wait()
+ else:
+ self.togo = self.n
+ full.broadcast()
+ full.release()
+
+class event:
+ def __init__(self):
+ self.state = 0
+ self.posted = condition()
+
+ def post(self):
+ self.posted.acquire()
+ self.state = 1
+ self.posted.broadcast()
+ self.posted.release()
+
+ def clear(self):
+ self.posted.acquire()
+ self.state = 0
+ self.posted.release()
+
+ def is_posted(self):
+ self.posted.acquire()
+ answer = self.state
+ self.posted.release()
+ return answer
+
+ def wait(self):
+ self.posted.acquire()
+ if not self.state:
+ self.posted.wait()
+ self.posted.release()
+
+class semaphore:
+ def __init__(self, count=1):
+ if count <= 0:
+ raise ValueError, 'semaphore count %d; must be >= 1' % count
+ self.count = count
+ self.maxcount = count
+ self.nonzero = condition()
+
+ def p(self):
+ self.nonzero.acquire()
+ while self.count == 0:
+ self.nonzero.wait()
+ self.count = self.count - 1
+ self.nonzero.release()
+
+ def v(self):
+ self.nonzero.acquire()
+ if self.count == self.maxcount:
+ raise ValueError, '.v() tried to raise semaphore count above ' \
+ 'initial value %r' % self.maxcount
+ self.count = self.count + 1
+ self.nonzero.signal()
+ self.nonzero.release()
+
+class mrsw:
+ def __init__(self):
+ # critical-section lock & the data it protects
+ self.rwOK = thread.allocate_lock()
+ self.nr = 0 # number readers actively reading (not just waiting)
+ self.nw = 0 # number writers either waiting to write or writing
+ self.writing = 0 # 1 iff some thread is writing
+
+ # conditions
+ self.readOK = condition(self.rwOK) # OK to unblock readers
+ self.writeOK = condition(self.rwOK) # OK to unblock writers
+
+ def read_in(self):
+ self.rwOK.acquire()
+ while self.nw:
+ self.readOK.wait()
+ self.nr = self.nr + 1
+ self.rwOK.release()
+
+ def read_out(self):
+ self.rwOK.acquire()
+ if self.nr <= 0:
+ raise ValueError, \
+ '.read_out() invoked without an active reader'
+ self.nr = self.nr - 1
+ if self.nr == 0:
+ self.writeOK.signal()
+ self.rwOK.release()
+
+ def write_in(self):
+ self.rwOK.acquire()
+ self.nw = self.nw + 1
+ while self.writing or self.nr:
+ self.writeOK.wait()
+ self.writing = 1
+ self.rwOK.release()
+
+ def write_out(self):
+ self.rwOK.acquire()
+ if not self.writing:
+ raise ValueError, \
+ '.write_out() invoked without an active writer'
+ self.writing = 0
+ self.nw = self.nw - 1
+ if self.nw:
+ self.writeOK.signal()
+ else:
+ self.readOK.broadcast()
+ self.rwOK.release()
+
+ def write_to_read(self):
+ self.rwOK.acquire()
+ if not self.writing:
+ raise ValueError, \
+ '.write_to_read() invoked without an active writer'
+ self.writing = 0
+ self.nw = self.nw - 1
+ self.nr = self.nr + 1
+ if not self.nw:
+ self.readOK.broadcast()
+ self.rwOK.release()
+
+# The rest of the file is a test case, that runs a number of parallelized
+# quicksorts in parallel. If it works, you'll get about 600 lines of
+# tracing output, with a line like
+# test passed! 209 threads created in all
+# as the last line. The content and order of preceding lines will
+# vary across runs.
+
+def _new_thread(func, *args):
+ global TID
+ tid.acquire(); id = TID = TID+1; tid.release()
+ io.acquire(); alive.append(id); \
+ print 'starting thread', id, '--', len(alive), 'alive'; \
+ io.release()
+ thread.start_new_thread( func, (id,) + args )
+
+def _qsort(tid, a, l, r, finished):
+ # sort a[l:r]; post finished when done
+ io.acquire(); print 'thread', tid, 'qsort', l, r; io.release()
+ if r-l > 1:
+ pivot = a[l]
+ j = l+1 # make a[l:j] <= pivot, and a[j:r] > pivot
+ for i in range(j, r):
+ if a[i] <= pivot:
+ a[j], a[i] = a[i], a[j]
+ j = j + 1
+ a[l], a[j-1] = a[j-1], pivot
+
+ l_subarray_sorted = event()
+ r_subarray_sorted = event()
+ _new_thread(_qsort, a, l, j-1, l_subarray_sorted)
+ _new_thread(_qsort, a, j, r, r_subarray_sorted)
+ l_subarray_sorted.wait()
+ r_subarray_sorted.wait()
+
+ io.acquire(); print 'thread', tid, 'qsort done'; \
+ alive.remove(tid); io.release()
+ finished.post()
+
+def _randarray(tid, a, finished):
+ io.acquire(); print 'thread', tid, 'randomizing array'; \
+ io.release()
+ for i in range(1, len(a)):
+ wh.acquire(); j = randint(0,i); wh.release()
+ a[i], a[j] = a[j], a[i]
+ io.acquire(); print 'thread', tid, 'randomizing done'; \
+ alive.remove(tid); io.release()
+ finished.post()
+
+def _check_sort(a):
+ if a != range(len(a)):
+ raise ValueError, ('a not sorted', a)
+
+def _run_one_sort(tid, a, bar, done):
+ # randomize a, and quicksort it
+ # for variety, all the threads running this enter a barrier
+ # at the end, and post `done' after the barrier exits
+ io.acquire(); print 'thread', tid, 'randomizing', a; \
+ io.release()
+ finished = event()
+ _new_thread(_randarray, a, finished)
+ finished.wait()
+
+ io.acquire(); print 'thread', tid, 'sorting', a; io.release()
+ finished.clear()
+ _new_thread(_qsort, a, 0, len(a), finished)
+ finished.wait()
+ _check_sort(a)
+
+ io.acquire(); print 'thread', tid, 'entering barrier'; \
+ io.release()
+ bar.enter()
+ io.acquire(); print 'thread', tid, 'leaving barrier'; \
+ io.release()
+ io.acquire(); alive.remove(tid); io.release()
+ bar.enter() # make sure they've all removed themselves from alive
+ ## before 'done' is posted
+ bar.enter() # just to be cruel
+ done.post()
+
+def test():
+ global TID, tid, io, wh, randint, alive
+ import random
+ randint = random.randint
+
+ TID = 0 # thread ID (1, 2, ...)
+ tid = thread.allocate_lock() # for changing TID
+ io = thread.allocate_lock() # for printing, and 'alive'
+ wh = thread.allocate_lock() # for calls to random
+ alive = [] # IDs of active threads
+
+ NSORTS = 5
+ arrays = []
+ for i in range(NSORTS):
+ arrays.append( range( (i+1)*10 ) )
+
+ bar = barrier(NSORTS)
+ finished = event()
+ for i in range(NSORTS):
+ _new_thread(_run_one_sort, arrays[i], bar, finished)
+ finished.wait()
+
+ print 'all threads done, and checking results ...'
+ if alive:
+ raise ValueError, ('threads still alive at end', alive)
+ for i in range(NSORTS):
+ a = arrays[i]
+ if len(a) != (i+1)*10:
+ raise ValueError, ('length of array', i, 'screwed up')
+ _check_sort(a)
+
+ print 'test passed!', TID, 'threads created in all'
+
+if __name__ == '__main__':
+ test()
+
+# end of module
diff --git a/Demo/threads/telnet.py b/Demo/threads/telnet.py
new file mode 100644
index 0000000..707a353
--- /dev/null
+++ b/Demo/threads/telnet.py
@@ -0,0 +1,114 @@
+# Minimal interface to the Internet telnet protocol.
+#
+# *** modified to use threads ***
+#
+# It refuses all telnet options and does not recognize any of the other
+# telnet commands, but can still be used to connect in line-by-line mode.
+# It's also useful to play with a number of other services,
+# like time, finger, smtp and even ftp.
+#
+# Usage: telnet host [port]
+#
+# The port may be a service name or a decimal port number;
+# it defaults to 'telnet'.
+
+
+import sys, os, time
+from socket import *
+import thread
+
+BUFSIZE = 8*1024
+
+# Telnet protocol characters
+
+IAC = chr(255) # Interpret as command
+DONT = chr(254)
+DO = chr(253)
+WONT = chr(252)
+WILL = chr(251)
+
+def main():
+ if len(sys.argv) < 2:
+ sys.stderr.write('usage: telnet hostname [port]\n')
+ sys.exit(2)
+ host = sys.argv[1]
+ try:
+ hostaddr = gethostbyname(host)
+ except error:
+ sys.stderr.write(sys.argv[1] + ': bad host name\n')
+ sys.exit(2)
+ #
+ if len(sys.argv) > 2:
+ servname = sys.argv[2]
+ else:
+ servname = 'telnet'
+ #
+ if '0' <= servname[:1] <= '9':
+ port = eval(servname)
+ else:
+ try:
+ port = getservbyname(servname, 'tcp')
+ except error:
+ sys.stderr.write(servname + ': bad tcp service name\n')
+ sys.exit(2)
+ #
+ s = socket(AF_INET, SOCK_STREAM)
+ #
+ try:
+ s.connect((host, port))
+ except error, msg:
+ sys.stderr.write('connect failed: %r\n' % (msg,))
+ sys.exit(1)
+ #
+ thread.start_new(child, (s,))
+ parent(s)
+
+def parent(s):
+ # read socket, write stdout
+ iac = 0 # Interpret next char as command
+ opt = '' # Interpret next char as option
+ while 1:
+ data, dummy = s.recvfrom(BUFSIZE)
+ if not data:
+ # EOF -- exit
+ sys.stderr.write( '(Closed by remote host)\n')
+ sys.exit(1)
+ cleandata = ''
+ for c in data:
+ if opt:
+ print ord(c)
+## print '(replying: %r)' % (opt+c,)
+ s.send(opt + c)
+ opt = ''
+ elif iac:
+ iac = 0
+ if c == IAC:
+ cleandata = cleandata + c
+ elif c in (DO, DONT):
+ if c == DO: print '(DO)',
+ else: print '(DONT)',
+ opt = IAC + WONT
+ elif c in (WILL, WONT):
+ if c == WILL: print '(WILL)',
+ else: print '(WONT)',
+ opt = IAC + DONT
+ else:
+ print '(command)', ord(c)
+ elif c == IAC:
+ iac = 1
+ print '(IAC)',
+ else:
+ cleandata = cleandata + c
+ sys.stdout.write(cleandata)
+ sys.stdout.flush()
+## print 'Out:', repr(cleandata)
+
+def child(s):
+ # read stdin, write socket
+ while 1:
+ line = sys.stdin.readline()
+## print 'Got:', repr(line)
+ if not line: break
+ s.send(line)
+
+main()
generated by cgit v0.12 at 2025-09-16 03:57:47 (GMT)