Skip to content.

logo Stackless Python

Personal tools


last edited 6 years ago by paul

You can copy the following sample code snippets directly into a Python interpreter session or run them as scripts.

Tasklets and channels

The following does only create a tasklet from a callable, in this case a function. The tasklet also gets bound to arguments for the callable. Then it is run.

Example: Creating a tasklet from a callable:

from stackless import tasklet

def func(): 
    print 42

t = tasklet(func) # bind tasklet to a callable
t = t() # bind tasklet to callable's arguments

# should print: 42

Example: Creating a channel:

import stackless
ch =

Simple communication

Sample 1

The following sample creates a channel with one sending and one receiving tasklet on it. The sending tasklet sends the value 42 which is received and printed by the other tasklet.

Example: Simple channel send between two explicit tasklets:

import stackless

def recv(chan): 
    print chan.receive()

def send(chan, val): 

ch =
t_recv = stackless.tasklet(recv)(ch)
t_send = stackless.tasklet(send)(ch, 42)

# should print: 42


Example: Pickling and unpickling simple tasklets:

from stackless import run, schedule, tasklet
import pickle

def aCallable(name):
    print "  aCallable<%s>: Before schedule()" % (name,)
    print "  aCallable<%s>: After schedule()" % (name,)

tasks = []
for name in "ABCDE":

print "Schedule 1:"

print "Pickling..."
pickledTasks = pickle.dumps(tasks)

print "Schedule 2:"

unpickledTasks = pickle.loads(pickledTasks)
for task in unpickledTasks:
print "Schedule Unpickled Tasks:"

Atomic Execution

I hacked this up tonight (untested) after reading the docs on atomic execution. With it, you should be able to do

Syntax for atomic execution using Python 2.5's with statement:

with (AtomicExecution?):

The AtomicExecution? contextmanager will handle all the calls to set_atomic() for you. Since this is just a direct translation of the code on , it should work correctly.

Here's the code:

Code for AtomicExecution? contextmanager:

from __future__ import with_statement
import stackless

class AtomicExecution?:
    def __enter__(self):
        self.tasklet = stackless.getcurrent()
        self.atomic = self.tasklet.set_atomic (True)

    def __exit__(exc_type, exc_val, exc_tb):
        self.tasklet.set_atomic (self.atomic)



Powered by Plone