2011/9/2 Armin Rigo <[email protected]>
>
> * stackless support could do with some volunteer work now --- in
> particular, lib_pypy/stackless.py could be refactored to use directly
> continulets, in a similar way to lib_pypy/greenlet.py. I am ready to
> explain a bit more on irc how continulets work, but I don't have
> enough motivation to do the complete refactoring myself. It's also a
> nice entry-level task for someone who wants to.
>
>
>
hi guys,
I made a first version of stackless with continulets. Attached the patch and
the complete stackless.py file for review. Someone have time for look the
patch?
I have attached too a file with a helper and examples for users to solve
"RuntimeError: maximum recursion depth exceeded" based on pypy stackless
online doc. I'm not sure if my version of stackless with continulets is
much faster than stackless with greenlets, in factorial.py file, in my
machine for input 2000, the new version runs on 20 seconds and the old
version 30.
best regards,
Rodrigo Araújo
diff --git a/lib_pypy/stackless.py b/lib_pypy/stackless.py
--- a/lib_pypy/stackless.py
+++ b/lib_pypy/stackless.py
@@ -4,121 +4,124 @@
Please refer to their documentation.
"""
-DEBUG = True
-
-def dprint(*args):
- for arg in args:
- print arg,
- print
import traceback
-import sys
+import _continuation
+from functools import partial
+
+class TaskletExit(Exception):
+ pass
+
+CoroutineExit = TaskletExit
+
+class GWrap(_continuation.continulet):
+ """This is just a wrapper around continulet to allow
+ to stick additional attributes to a continulet.
+ To be more concrete, we need a backreference to
+ the coroutine object"""
+
+
+class coroutine(object):
+ "we can't have continulet as a base, because continulets can't be rebound"
+
+ def __init__(self):
+ self._frame = None
+ self.is_zombie = False
+
+ def __getattr__(self, attr):
+ return getattr(self._frame, attr)
+
+ def __del__(self):
+ self.is_zombie = True
+ del self._frame
+ self._frame = None
+
+ def bind(self, func, *argl, **argd):
+ """coro.bind(f, *argl, **argd) -> None.
+ binds function f to coro. f will be called with
+ arguments *argl, **argd
+ """
+ if self._frame is None or not self._frame.is_pending():
+
+ def _func(c, *args, **kwargs):
+ return func(*args, **kwargs)
+
+ run = partial(_func, *argl, **argd)
+ self._frame = frame = GWrap(run)
+ else:
+ raise ValueError("cannot bind a bound coroutine")
+
+ def switch(self):
+ """coro.switch() -> returnvalue
+ switches to coroutine coro. If the bound function
+ f finishes, the returnvalue is that of f, otherwise
+ None is returned
+ """
+ current = _getcurrent()
+ current._jump_to(self)
+
+ def _jump_to(self, coroutine):
+ _tls.current_coroutine = coroutine
+ self._frame.switch(to=coroutine._frame)
+
+ def kill(self):
+ """coro.kill() : kill coroutine coro"""
+ _tls.current_coroutine = self
+ self._frame.throw(CoroutineExit)
+
+ def _is_alive(self):
+ if self._frame is None:
+ return False
+ return not self._frame.is_pending()
+ is_alive = property(_is_alive)
+ del _is_alive
+
+ def getcurrent():
+ """coroutine.getcurrent() -> the currently running coroutine"""
+ try:
+ return _getcurrent()
+ except AttributeError:
+ return _maincoro
+ getcurrent = staticmethod(getcurrent)
+
+ def __reduce__(self):
+ raise TypeError, 'pickling is not possible based upon continulets'
+
+
+def _getcurrent():
+ "Returns the current coroutine (i.e. the one which called this function)."
+ try:
+ return _tls.current_coroutine
+ except AttributeError:
+ # first call in this thread: current == main
+ _coroutine_create_main()
+ return _tls.current_coroutine
+
try:
- # If _stackless can be imported then TaskletExit and CoroutineExit are
- # automatically added to the builtins.
- from _stackless import coroutine, greenlet
-except ImportError: # we are running from CPython
- from greenlet import greenlet, GreenletExit
- TaskletExit = CoroutineExit = GreenletExit
- del GreenletExit
- try:
- from functools import partial
- except ImportError: # we are not running python 2.5
- class partial(object):
- # just enough of 'partial' to be usefull
- def __init__(self, func, *argl, **argd):
- self.func = func
- self.argl = argl
- self.argd = argd
+ from thread import _local
+except ImportError:
+ class _local(object): # assume no threads
+ pass
- def __call__(self):
- return self.func(*self.argl, **self.argd)
+_tls = _local()
- class GWrap(greenlet):
- """This is just a wrapper around greenlets to allow
- to stick additional attributes to a greenlet.
- To be more concrete, we need a backreference to
- the coroutine object"""
+def _coroutine_create_main():
+ # create the main coroutine for this thread
+ _tls.current_coroutine = None
+ main_coroutine = coroutine()
+ main_coroutine.bind(lambda x:x)
+ _tls.main_coroutine = main_coroutine
+ _tls.current_coroutine = main_coroutine
+ return main_coroutine
- class MWrap(object):
- def __init__(self,something):
- self.something = something
- def __getattr__(self, attr):
- return getattr(self.something, attr)
+_maincoro = _coroutine_create_main()
- class coroutine(object):
- "we can't have greenlet as a base, because greenlets can't be rebound"
-
- def __init__(self):
- self._frame = None
- self.is_zombie = False
-
- def __getattr__(self, attr):
- return getattr(self._frame, attr)
-
- def __del__(self):
- self.is_zombie = True
- del self._frame
- self._frame = None
-
- def bind(self, func, *argl, **argd):
- """coro.bind(f, *argl, **argd) -> None.
- binds function f to coro. f will be called with
- arguments *argl, **argd
- """
- if self._frame is None or self._frame.dead:
- self._frame = frame = GWrap()
- frame.coro = self
- if hasattr(self._frame, 'run') and self._frame.run:
- raise ValueError("cannot bind a bound coroutine")
- self._frame.run = partial(func, *argl, **argd)
-
- def switch(self):
- """coro.switch() -> returnvalue
- switches to coroutine coro. If the bound function
- f finishes, the returnvalue is that of f, otherwise
- None is returned
- """
- try:
- return greenlet.switch(self._frame)
- except TypeError, exp: # self._frame is the main coroutine
- return greenlet.switch(self._frame.something)
-
- def kill(self):
- """coro.kill() : kill coroutine coro"""
- self._frame.throw()
-
- def _is_alive(self):
- if self._frame is None:
- return False
- return not self._frame.dead
- is_alive = property(_is_alive)
- del _is_alive
-
- def getcurrent():
- """coroutine.getcurrent() -> the currently running coroutine"""
- try:
- return greenlet.getcurrent().coro
- except AttributeError:
- return _maincoro
- getcurrent = staticmethod(getcurrent)
-
- def __reduce__(self):
- raise TypeError, 'pickling is not possible based upon greenlets'
-
- _maincoro = coroutine()
- maingreenlet = greenlet.getcurrent()
- _maincoro._frame = frame = MWrap(maingreenlet)
- frame.coro = _maincoro
- del frame
- del maingreenlet
from collections import deque
import operator
-__all__ = 'run getcurrent getmain schedule tasklet channel coroutine \
- greenlet'.split()
+__all__ = 'run getcurrent getmain schedule tasklet channel coroutine'.split()
_global_task_id = 0
_squeue = None
@@ -131,7 +134,8 @@
def _scheduler_remove(value):
try:
del _squeue[operator.indexOf(_squeue, value)]
- except ValueError:pass
+ except ValueError:
+ pass
def _scheduler_append(value, normal=True):
if normal:
"""
The Stackless module allows you to do multitasking without using threads.
The essential objects are tasklets and channels.
Please refer to their documentation.
"""
import traceback
import _continuation
from functools import partial
class TaskletExit(Exception):
pass
CoroutineExit = TaskletExit
class GWrap(_continuation.continulet):
"""This is just a wrapper around continulet to allow
to stick additional attributes to a continulet.
To be more concrete, we need a backreference to
the coroutine object"""
class coroutine(object):
"we can't have continulet as a base, because continulets can't be rebound"
def __init__(self):
self._frame = None
self.is_zombie = False
def __getattr__(self, attr):
return getattr(self._frame, attr)
def __del__(self):
self.is_zombie = True
del self._frame
self._frame = None
def bind(self, func, *argl, **argd):
"""coro.bind(f, *argl, **argd) -> None.
binds function f to coro. f will be called with
arguments *argl, **argd
"""
if self._frame is None or not self._frame.is_pending():
def _func(c, *args, **kwargs):
return func(*args, **kwargs)
run = partial(_func, *argl, **argd)
self._frame = frame = GWrap(run)
else:
raise ValueError("cannot bind a bound coroutine")
def switch(self):
"""coro.switch() -> returnvalue
switches to coroutine coro. If the bound function
f finishes, the returnvalue is that of f, otherwise
None is returned
"""
current = _getcurrent()
current._jump_to(self)
def _jump_to(self, coroutine):
_tls.current_coroutine = coroutine
self._frame.switch(to=coroutine._frame)
def kill(self):
"""coro.kill() : kill coroutine coro"""
_tls.current_coroutine = self
self._frame.throw(CoroutineExit)
def _is_alive(self):
if self._frame is None:
return False
return not self._frame.is_pending()
is_alive = property(_is_alive)
del _is_alive
def getcurrent():
"""coroutine.getcurrent() -> the currently running coroutine"""
try:
return _getcurrent()
except AttributeError:
return _maincoro
getcurrent = staticmethod(getcurrent)
def __reduce__(self):
raise TypeError, 'pickling is not possible based upon continulets'
def _getcurrent():
"Returns the current coroutine (i.e. the one which called this function)."
try:
return _tls.current_coroutine
except AttributeError:
# first call in this thread: current == main
_coroutine_create_main()
return _tls.current_coroutine
try:
from thread import _local
except ImportError:
class _local(object): # assume no threads
pass
_tls = _local()
def _coroutine_create_main():
# create the main coroutine for this thread
_tls.current_coroutine = None
main_coroutine = coroutine()
main_coroutine.bind(lambda x:x)
_tls.main_coroutine = main_coroutine
_tls.current_coroutine = main_coroutine
return main_coroutine
_maincoro = _coroutine_create_main()
from collections import deque
import operator
__all__ = 'run getcurrent getmain schedule tasklet channel coroutine'.split()
_global_task_id = 0
_squeue = None
_main_tasklet = None
_main_coroutine = None
_last_task = None
_channel_callback = None
_schedule_callback = None
def _scheduler_remove(value):
try:
del _squeue[operator.indexOf(_squeue, value)]
except ValueError:
pass
def _scheduler_append(value, normal=True):
if normal:
_squeue.append(value)
else:
_squeue.rotate(-1)
_squeue.appendleft(value)
_squeue.rotate(1)
def _scheduler_contains(value):
try:
operator.indexOf(_squeue, value)
return True
except ValueError:
return False
def _scheduler_switch(current, next):
global _last_task
prev = _last_task
if (_schedule_callback is not None and
prev is not next):
_schedule_callback(prev, next)
_last_task = next
assert not next.blocked
if next is not current:
try:
next.switch()
except CoroutineExit:
raise TaskletExit
return current
def set_schedule_callback(callback):
global _schedule_callback
_schedule_callback = callback
def set_channel_callback(callback):
global _channel_callback
_channel_callback = callback
def getruncount():
return len(_squeue)
class bomb(object):
def __init__(self, exp_type=None, exp_value=None, exp_traceback=None):
self.type = exp_type
self.value = exp_value
self.traceback = exp_traceback
def raise_(self):
raise self.type, self.value, self.traceback
#
# helpers for pickling
#
_stackless_primitive_registry = {}
def register_stackless_primitive(thang, retval_expr='None'):
import types
func = thang
if isinstance(thang, types.MethodType):
func = thang.im_func
code = func.func_code
_stackless_primitive_registry[code] = retval_expr
# It is not too nice to attach info via the code object, but
# I can't think of a better solution without a real transform.
def rewrite_stackless_primitive(coro_state, alive, tempval):
flags, frame, thunk, parent = coro_state
while frame is not None:
retval_expr = _stackless_primitive_registry.get(frame.f_code)
if retval_expr:
# this tasklet needs to stop pickling here and return its value.
tempval = eval(retval_expr, globals(), frame.f_locals)
coro_state = flags, frame, thunk, parent
break
frame = frame.f_back
return coro_state, alive, tempval
#
#
class channel(object):
"""
A channel object is used for communication between tasklets.
By sending on a channel, a tasklet that is waiting to receive
is resumed. If there is no waiting receiver, the sender is suspended.
By receiving from a channel, a tasklet that is waiting to send
is resumed. If there is no waiting sender, the receiver is suspended.
Attributes:
preference
----------
-1: prefer receiver
0: don't prefer anything
1: prefer sender
Pseudocode that shows in what situation a schedule happens:
def send(arg):
if !receiver:
schedule()
elif schedule_all:
schedule()
else:
if (prefer receiver):
schedule()
else (don't prefer anything, prefer sender):
pass
NOW THE INTERESTING STUFF HAPPENS
def receive():
if !sender:
schedule()
elif schedule_all:
schedule()
else:
if (prefer sender):
schedule()
else (don't prefer anything, prefer receiver):
pass
NOW THE INTERESTING STUFF HAPPENS
schedule_all
------------
True: overwrite preference. This means that the current tasklet always
schedules before returning from send/receive (it always blocks).
(see Stackless/module/channelobject.c)
"""
def __init__(self, label=''):
self.balance = 0
self.closing = False
self.queue = deque()
self.label = label
self.preference = -1
self.schedule_all = False
def __str__(self):
return 'channel[%s](%s,%s)' % (self.label, self.balance, self.queue)
def close(self):
"""
channel.close() -- stops the channel from enlarging its queue.
If the channel is not empty, the flag 'closing' becomes true.
If the channel is empty, the flag 'closed' becomes true.
"""
self.closing = True
@property
def closed(self):
return self.closing and not self.queue
def open(self):
"""
channel.open() -- reopen a channel. See channel.close.
"""
self.closing = False
def _channel_action(self, arg, d):
"""
d == -1 : receive
d == 1 : send
the original CStackless has an argument 'stackl' which is not used
here.
'target' is the peer tasklet to the current one
"""
do_schedule=False
assert abs(d) == 1
source = getcurrent()
source.tempval = arg
if d > 0:
cando = self.balance < 0
dir = d
else:
cando = self.balance > 0
dir = 0
if _channel_callback is not None:
_channel_callback(self, source, dir, not cando)
self.balance += d
if cando:
# communication 1): there is somebody waiting
target = self.queue.popleft()
source.tempval, target.tempval = target.tempval, source.tempval
target.blocked = 0
if self.schedule_all:
# always schedule
_scheduler_append(target)
do_schedule = True
elif self.preference == -d:
_scheduler_append(target, False)
do_schedule = True
else:
_scheduler_append(target)
else:
# communication 2): there is nobody waiting
# if source.block_trap:
# raise RuntimeError("this tasklet does not like to be blocked")
# if self.closing:
# raise StopIteration()
source.blocked = d
self.queue.append(source)
_scheduler_remove(getcurrent())
do_schedule = True
if do_schedule:
schedule()
retval = source.tempval
if isinstance(retval, bomb):
retval.raise_()
return retval
def receive(self):
"""
channel.receive() -- receive a value over the channel.
If no other tasklet is already sending on the channel,
the receiver will be blocked. Otherwise, the receiver will
continue immediately, and the sender is put at the end of
the runnables list.
The above policy can be changed by setting channel flags.
"""
return self._channel_action(None, -1)
register_stackless_primitive(receive, retval_expr='receiver.tempval')
def send_exception(self, exp_type, msg):
self.send(bomb(exp_type, exp_type(msg)))
def send_sequence(self, iterable):
for item in iterable:
self.send(item)
def send(self, msg):
"""
channel.send(value) -- send a value over the channel.
If no other tasklet is already receiving on the channel,
the sender will be blocked. Otherwise, the receiver will
be activated immediately, and the sender is put at the end of
the runnables list.
"""
return self._channel_action(msg, 1)
register_stackless_primitive(send)
class tasklet(coroutine):
"""
A tasklet object represents a tiny task in a Python thread.
At program start, there is always one running main tasklet.
New tasklets can be created with methods from the stackless
module.
"""
tempval = None
def __new__(cls, func=None, label=''):
res = coroutine.__new__(cls)
res.label = label
res._task_id = None
return res
def __init__(self, func=None, label=''):
coroutine.__init__(self)
self._init(func, label)
def _init(self, func=None, label=''):
global _global_task_id
self.func = func
self.alive = False
self.blocked = False
self._task_id = _global_task_id
self.label = label
_global_task_id += 1
def __str__(self):
return '<tasklet[%s, %s]>' % (self.label,self._task_id)
__repr__ = __str__
def __call__(self, *argl, **argd):
return self.setup(*argl, **argd)
def bind(self, func):
"""
Binding a tasklet to a callable object.
The callable is usually passed in to the constructor.
In some cases, it makes sense to be able to re-bind a tasklet,
after it has been run, in order to keep its identity.
Note that a tasklet can only be bound when it doesn't have a frame.
"""
if not callable(func):
raise TypeError('tasklet function must be a callable')
self.func = func
def kill(self):
"""
tasklet.kill -- raise a TaskletExit exception for the tasklet.
Note that this is a regular exception that can be caught.
The tasklet is immediately activated.
If the exception passes the toplevel frame of the tasklet,
the tasklet will silently die.
"""
if not self.is_zombie:
# Killing the tasklet by throwing TaskletExit exception.
coroutine.kill(self)
_scheduler_remove(self)
self.alive = False
def setup(self, *argl, **argd):
"""
supply the parameters for the callable
"""
if self.func is None:
raise TypeError('cframe function must be callable')
func = self.func
def _func():
try:
try:
func(*argl, **argd)
except TaskletExit:
pass
finally:
_scheduler_remove(self)
self.alive = False
self.func = None
coroutine.bind(self, _func)
self.alive = True
_scheduler_append(self)
return self
def run(self):
self.insert()
_scheduler_switch(getcurrent(), self)
def insert(self):
if self.blocked:
raise RuntimeError, "You cannot run a blocked tasklet"
if not self.alive:
raise RuntimeError, "You cannot run an unbound(dead) tasklet"
_scheduler_append(self)
def remove(self):
if self.blocked:
raise RuntimeError, "You cannot remove a blocked tasklet."
if self is getcurrent():
raise RuntimeError, "The current tasklet cannot be removed."
# not sure if I will revive this " Use t=tasklet().capture()"
_scheduler_remove(self)
def __reduce__(self):
one, two, coro_state = coroutine.__reduce__(self)
assert one is coroutine
assert two == ()
# we want to get rid of the parent thing.
# for now, we just drop it
a, frame, c, d = coro_state
# Removing all frames related to stackless.py.
# They point to stuff we don't want to be pickled.
pickleframe = frame
while frame is not None:
if frame.f_code == schedule.func_code:
# Removing everything including and after the
# call to stackless.schedule()
pickleframe = frame.f_back
break
frame = frame.f_back
if d:
assert isinstance(d, coroutine)
coro_state = a, pickleframe, c, None
coro_state, alive, tempval = rewrite_stackless_primitive(coro_state, self.alive, self.tempval)
inst_dict = self.__dict__.copy()
inst_dict.pop('tempval', None)
return self.__class__, (), (coro_state, alive, tempval, inst_dict)
def __setstate__(self, (coro_state, alive, tempval, inst_dict)):
coroutine.__setstate__(self, coro_state)
self.__dict__.update(inst_dict)
self.alive = alive
self.tempval = tempval
def getmain():
"""
getmain() -- return the main tasklet.
"""
return _main_tasklet
def getcurrent():
"""
getcurrent() -- return the currently executing tasklet.
"""
curr = coroutine.getcurrent()
if curr is _main_coroutine:
return _main_tasklet
else:
return curr
_run_calls = []
def run():
"""
run_watchdog(timeout) -- run tasklets until they are all
done, or timeout instructions have passed. Tasklets must
provide cooperative schedule() calls.
If the timeout is met, the function returns.
The calling tasklet is put aside while the tasklets are running.
It is inserted back after the function stops, right before the
tasklet that caused a timeout, if any.
If an exception occours, it will be passed to the main tasklet.
Please note that the 'timeout' feature is not yet implemented
"""
curr = getcurrent()
_run_calls.append(curr)
_scheduler_remove(curr)
try:
schedule()
assert not _squeue
finally:
_scheduler_append(curr)
def schedule_remove(retval=None):
"""
schedule(retval=stackless.current) -- switch to the next runnable tasklet.
The return value for this call is retval, with the current
tasklet as default.
schedule_remove(retval=stackless.current) -- ditto, and remove self.
"""
_scheduler_remove(getcurrent())
r = schedule(retval)
return r
def schedule(retval=None):
"""
schedule(retval=stackless.current) -- switch to the next runnable tasklet.
The return value for this call is retval, with the current
tasklet as default.
schedule_remove(retval=stackless.current) -- ditto, and remove self.
"""
mtask = getmain()
curr = getcurrent()
if retval is None:
retval = curr
while True:
if _squeue:
if _squeue[0] is curr:
# If the current is at the head, skip it.
_squeue.rotate(-1)
task = _squeue[0]
#_squeue.rotate(-1)
elif _run_calls:
task = _run_calls.pop()
else:
raise RuntimeError('No runnable tasklets left.')
_scheduler_switch(curr, task)
if curr is _last_task:
# We are in the tasklet we want to resume at this point.
return retval
def _init():
global _main_tasklet
global _global_task_id
global _squeue
global _last_task
_global_task_id = 0
_main_tasklet = coroutine.getcurrent()
try:
_main_tasklet.__class__ = tasklet
except TypeError: # we are running pypy-c
class TaskletProxy(object):
"""TaskletProxy is needed to give the _main_coroutine tasklet behaviour"""
def __init__(self, coro):
self._coro = coro
def __getattr__(self,attr):
return getattr(self._coro,attr)
def __str__(self):
return '<tasklet %s a:%s>' % (self._task_id, self.is_alive)
def __reduce__(self):
return getmain, ()
__repr__ = __str__
global _main_coroutine
_main_coroutine = _main_tasklet
_main_tasklet = TaskletProxy(_main_tasklet)
assert _main_tasklet.is_alive and not _main_tasklet.is_zombie
_last_task = _main_tasklet
tasklet._init.im_func(_main_tasklet, label='main')
_squeue = deque()
_scheduler_append(_main_tasklet)
_init()
from _continuation import continulet
import sys
def _invoke(_, callable, args, kwargs):
return callable(*args, **kwargs)
class recursive_continulet(object):
def __init__(self, callable):
self.callable = callable
def bootstrap(c):
# this loop runs forever, at a very low recursion depth
args,kwargs = c.switch()
while True:
# start a new continulet from here, and switch to
# it using an "exchange", i.e. a switch with to=.
to = continulet(_invoke, callable, args,kwargs)
args,kwargs = c.switch(to=to)
self.continulet = continulet(bootstrap)
self.continulet.switch()
def __call__(self, *args, **kwargs):
return self.continulet.switch((args,kwargs))
if __name__ == "main":
@recursive_continulet
def recursive(n):
if n == 0:
return ("ok", n)
if n % 200 == 0:
prev = recursive((n - 1))
else:
prev = recursive(n - 1)
return (prev[0], prev[1] + 1)
print recursive(int(sys.argv[1]))
@recursive_continulet
def factorial(n):
if n <= 1:
return 1
return n * factorial(n-1)
print factorial(int(sys.argv[1]))
@recursive_continulet
def fibonacci(n):
if n == 1 or n == 0:
return 1
else:
return fibonacci(n-1) + fibonacci(n-2)
print fibonacci(int(sys.argv[1]))
#
# Factorial example using stackless to break Python's recustion limit of 1000
#
# by Andrew Dalke
#
# If you have any questions related to this example:
#
# - If they are related to how Twisted works, please contact the
# Twisted mailing list:
#
# http://twistedmatrix.com/cgi-bin/mailman/listinfo/twisted-python
#
# - If they are related to how Stackless works, please contact
# the Stackless mailing list:
#
# http://www.tismer.com/mailman/listinfo/stackless
#
import stackless
import time
import recursion_helper
def call_wrapper(f, args, kwargs, result_ch):
result_ch.send(f(*args, **kwargs))
def call(f, *args, **kwargs):
result_ch= stackless.channel()
stackless.tasklet(call_wrapper)(f, args, kwargs, result_ch)
return result_ch.receive()
@recursion_helper.recursive_continulet
def factorial(n):
if n <= 1:
return 1
return n * call(factorial, n-1)
st = time.time()
print factorial(2000)
print time.time() - st
#
# Almost the same Producer/Consumer example from PyQt but without
# the graphical interface, the queue status is printed in console.
#
# by Carlos Eduardo de Paula <[email protected]>
#
# If you have any questions related to this example:
#
# - If they are related to how Twisted works, please contact the
# Twisted mailing list:
#
# http://twistedmatrix.com/cgi-bin/mailman/listinfo/twisted-python
#
# - If they are related to how Stackless works, please contact
# the Stackless mailing list:
#
# http://www.tismer.com/mailman/listinfo/stackless
#
import stackless
import time
import random
# Nice way to put the tasklet to sleep - from stackless.com wiki/Idioms
##########################################################
sleepingTasklets = []
def Sleep(secondsToWait):
channel = stackless.channel()
endTime = time.time() + secondsToWait
sleepingTasklets.append((endTime, channel))
sleepingTasklets.sort()
# Block until we get sent an awakening notification.
channel.receive()
def ManageSleepingTasklets():
while True:
if len(sleepingTasklets):
endTime = sleepingTasklets[0][0]
if endTime <= time.time():
channel = sleepingTasklets[0][1]
del sleepingTasklets[0]
# We have to send something, but it doesn't matter what as it is not used.
channel.send(None)
elif stackless.getruncount() == 1:
#print "sleep real"
# We are the only tasklet running, the rest are blocked on channels sleeping.
# We can call time.sleep until the first awakens to avoid a busy wait.
delay = endTime - time.time()
#print "wait delay", delay
time.sleep(max(delay,0))
stackless.schedule()
stackless.tasklet(ManageSleepingTasklets)()
##########################################################
def printStatus(reporter):
print reporter + " " * (3 - len(reporter)) + "[" + "#" * len(queue) + " " * (q_size-len(queue)) + "] Qty:" , len(queue) , "\r",
time.sleep(0.05) # so we have time to see the displayed data
# keep in mind that this call to time.sleep will stall
# all tasklets stop until time.sleep returns
c_p = 0
def producer(who,sleeptime):
global full_queue
global c_p
while True:
c_p += 1
if (len(queue) < q_size):
queue.append("#")
p_counter[int(who)] += 1
printStatus('P'+who)
if len(queue) < q_size/4:
Sleep(sleeptime)
else:
Sleep(sleeptime*1.5)
else:
full_queue += 1
stackless.schedule()
c_c = 0
def consumer(who,sleeptime):
global zero_queue
global c_c
while True:
c_c += 1
if (len(queue) >= 1):
queue.pop()
c_counter[int(who)] += 1
printStatus('C'+who)
if len(queue) < q_size/4:
Sleep(sleeptime*1.5)
else:
Sleep(sleeptime)
else:
zero_queue += 1
stackless.schedule()
'''
def watch():
while True:
if len(sleepingTasklets) <= 0:
stackless.schedule()
stackless.tasklet(watch)()
'''
def launch_p (ind,sleeptime): # Launches and initializes the producers lists
producers.append(int(ind))
p_counter.append(0)
producers[int(ind)] = stackless.tasklet(producer)(ind,sleeptime)
def launch_c (ind,sleeptime): # Launches and initializes the consumers lists
consumers.append(int(ind))
c_counter.append(0)
consumers[int(ind)] = stackless.tasklet(consumer)(ind,sleeptime)
#-------------------- Configuration --------------------------------
q_size = 60 # Defines the queue size
queue = ["#"] * 0 # Defines the queue start size
producers = [] # List to reference the producers
consumers = [] # List to reference the consumers
p_counter = [] # Counter to hold how much units each producer inserted in queue
c_counter = [] # Counter to hold how much units each consumer removed from queue
num_prod = 5 # Number of starting producers
num_cons = 5 # Number of starting consumers
zero_queue = 0
full_queue = 0
for p in range(num_prod):
sl = random.random()
launch_p(repr(p),sl)
for c in range(num_cons):
sl = random.random()
launch_c(repr(c),sl)
try:
stackless.run()
# Handle the keyboard interruption and prints the production report
except KeyboardInterrupt:
print ""
print "** Detected ctrl-c in the console"
exit
total_p = 0
total_c = 0
print
for p in range(0,len(producers)):
print "Producer", p , "produced: ", p_counter[p]
total_p += p_counter[p]
for c in range(0,len(consumers)):
print "Consumer", c , "consumed: ", c_counter[c]
total_c += c_counter[c]
print
print "Produced units: ", total_p
print "Consumed units: ", total_c
print "Produced calls: ", c_p
print "Consumed call: ", c_c
print "Left in queue: ", len(queue)
print
print "Queue became zero ", zero_queue, " times."
print "Queue became full ", full_queue, " times."
print "Press enter to finish" , raw_input()
_______________________________________________
pypy-dev mailing list
[email protected]
http://mail.python.org/mailman/listinfo/pypy-dev
