pypy / pypy / rlib /

import sys
from pypy.rlib import _rffi_stacklet as _c
from pypy.rlib import jit
from pypy.rlib.objectmodel import we_are_translated
from pypy.rpython.lltypesystem import lltype, llmemory

DEBUG = False

class StackletThread(object):

    def __init__(self, config):
        self._gcrootfinder = _getgcrootfinder(config, we_are_translated())
        self._thrd = _c.newthread()
        if not self._thrd:
            raise MemoryError
        self._thrd_deleter = StackletThreadDeleter(self._thrd)
        if DEBUG:
            assert debug.sthread is None, "multithread debug support missing"
            debug.sthread = self

    def new(self, callback, arg=llmemory.NULL):
        if DEBUG:
            callback = _debug_wrapper(callback)
        h =, callback, arg)
        if DEBUG:
        return h
    new._annspecialcase_ = 'specialize:arg(1)'

    def switch(self, stacklet):
        if DEBUG:
        h = self._gcrootfinder.switch(self, stacklet)
        if DEBUG:
        return h

    def destroy(self, stacklet):
        if DEBUG:
        self._gcrootfinder.destroy(self, stacklet)

    def is_empty_handle(self, stacklet):
        # note that "being an empty handle" and being equal to
        # "get_null_handle()" may be the same, or not; don't rely on it
        return self._gcrootfinder.is_empty_handle(stacklet)

    def get_null_handle(self):
        return self._gcrootfinder.get_null_handle()

class StackletThreadDeleter(object):
    # quick hack: the __del__ is on another object, so that
    # if the main StackletThread ends up in random circular
    # references, on pypy deletethread() is only called
    # when all that circular reference mess is gone.
    def __init__(self, thrd):
        self._thrd = thrd
    def __del__(self):
        thrd = self._thrd
        if thrd:
            self._thrd = lltype.nullptr(_c.thread_handle.TO)

# ____________________________________________________________

def _getgcrootfinder(config, translated):
    if translated:
        assert config is not None, ("you have to pass a valid config, "
                                    "e.g. from 'driver.config'")
    elif '__pypy__' in sys.builtin_module_names:
        import py
        py.test.skip("cannot run the stacklet tests on top of pypy: "
                     "calling directly the C function stacklet_switch() "
                     "will crash, depending on details of your config")
    if config is not None:
        assert config.translation.continuation, (
            "stacklet: you have to translate with --continuation")
    if (config is None or
        config.translation.gc in ('ref', 'boehm', 'none')):   # for tests
        gcrootfinder = 'n/a'
        gcrootfinder = config.translation.gcrootfinder
    gcrootfinder = gcrootfinder.replace('/', '_')
    module = __import__('pypy.rlib._stacklet_%s' % gcrootfinder,
                        None, None, ['__doc__'])
    return module.gcrootfinder
_getgcrootfinder._annspecialcase_ = 'specialize:memo'

class StackletDebugError(Exception):

class Debug(object):
    def __init__(self):
        self.sthread = None = []
    def _cleanup_(self):
    def add(self, h):
        if not self.sthread.is_empty_handle(h):
            if h == self.sthread.get_null_handle():
                raise StackletDebugError("unexpected null handle")
    def remove(self, h):
            i =
        except ValueError:
            if self.sthread.is_empty_handle(h):
                msg = "empty stacklet handle"
            elif h == self.sthread.get_null_handle():
                msg = "unexpected null handle"
                msg = "double usage of handle %r" % (h,)
            raise StackletDebugError(msg)
debug = Debug()

def _debug_wrapper(callback):
    def wrapper(h, arg):
        h = callback(h, arg)
        return h
    return wrapper
_debug_wrapper._annspecialcase_ = 'specialize:memo'