sorcio

from itertools import count
import signal

from async_generator import asynccontextmanager
import trio


class ShutdownHandler:
    def __init__(self):
        self.event = trio.Event()

sorcio

from contextlib import contextmanager
from functools import partial
import threading
import types
from queue import Queue
import sys

import outcome
import trio

sorcio

END_OF_QUEUE = object()
class QueueTerminated(Exception): pass


class ObjectSender:

    def __init__(self, queue):
        self.queue = queue
        self.senders = 0

sorcio

import timeit

import trio


class Stop(Exception):
    pass


async def benchmark1(qsize):

sorcio

@pytest.mark.trio
async def test_worker_thread_autojump(nursery, autojump_clock):
    def slowish_task():
        for i in range(1000000):
            pass
        return 'foo'

    async def async_run_task(queue):
        result = await trio.run_sync_in_worker_thread(slowish_task)
        await queue.put(result)

sorcio

import random
import sre_parse


MAX_ALLOWED_REPEATS = 100


def randre(re_text):
    pattern = sre_parse.parse(re_text)
    return Generator().gen_pattern(pattern)

sorcio

Keybase proof

I hereby claim:

• I am sorcio on github.
• I am davider (https://keybase.io/davider) on keybase.
• I have a public key whose fingerprint is 0021 FE8B 13A5 5D6D 8218 B06C 8E3B 5D19 94BC D692

To claim this, I am signing this object:

sorcio

class PythonProperty(object):
    """
    A property-like descriptor written in Python.

    Simplifications from built-in property():
    - it doesn't implement the del operation
    - decorator syntax can only be used for read-only properties
    """

    def __init__(self, getter, setter=None):