Skip to content

Instantly share code, notes, and snippets.

@richardsheridan
richardsheridan / universal_queue.py
Last active November 7, 2022 04:09 — forked from njsmith/universal-trio-queue.py
Universal cross-thread unbuffered queue for trio, asyncio, and threads
# Rough draft of a Queue object that can be used simultaneously from
# sync threads + *multiple* trio and asyncio threads, all at once.
#
# If you don't have multiple threads each doing their own separate calls to Xio.run,
# then don't use this; there are simpler solutions. This was mostly an exercise to
# figure out if and how this could be done.
#
# Currently, the test will provide 94% coverage given sufficient timeout. The
# remaining are (apparently rare) races and the durable aio shielding.
import random
@richardsheridan
richardsheridan / map_concurrently_in_subthread_trio.py
Last active January 2, 2023 22:14
map_concurrently_in_subthread_trio
import queue
import random
from functools import partial
from time import sleep, perf_counter
import trio
CONCURRENCY_LIMIT = 8
limiter = trio.CapacityLimiter(CONCURRENCY_LIMIT)
@richardsheridan
richardsheridan / Nd-testfuncs-python.md
Created March 9, 2021 23:14 — forked from denis-bz/Nd-testfuncs-python.md
N-dimensional test functions for optimization, in Python