tauthon/tauthon2.7
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Import Upstream version 2.7.18

Browse Source
This commit is contained in:
geos_one
2025-08-15 16:28:06 +02:00
commit ba1f69ab39
4521 changed files with 1778434 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

159
Demo/threads/Coroutine.py Normal file
View File

@@ -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

92
Demo/threads/Generator.py Normal file
View File

@@ -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()

11
Demo/threads/README Normal file
View File

@@ -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

64
Demo/threads/fcmp.py Normal file
View File

@@ -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

155
Demo/threads/find.py Normal file
View File

@@ -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()

105
Demo/threads/squasher.py Normal file
View File

@@ -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

603
Demo/threads/sync.py Normal file
View File

@@ -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

114
Demo/threads/telnet.py Normal file
View File

@@ -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()