Utilities

grain.util.set_numpy_oneline_repr()

Change the default Numpy array __repr__ to a compact one. This is useful for keeping the log clean when the job functions involve large Numpy arrays as args.

async with grain.util.QueueLimiter(conc) as ql

Rate limit for the submission of delayed functions. Grain by default enqueues delayed functions eagerly (i.e. as soon as it is called). Sometimes if we have a lot of functions that can be run in parallel, we don’t want to overwhelm the queue, so we could set a rate limit check prior submission. Note that the context scope blocks until all functions submitted through it are done.

Parameters:

conc (int) – maximum number of concurrently running functions

e.g.:

@delayed
async def dfn(x):
    await trio.sleep(1)
    return x+1
r_ = 0
async with QueueLimiter(10) as ql:
    for i in range(30):
        r_ += await ql(dfn)(i)
        #     ^^^^^wait for submission, not for result
r = await r_

async with grain.util.open_nursery_with_capacity(conc) as nursery

A patched Trio.Nursery with child task capacity limit. Its start_once_acquired blocks when the number of running child tasks started by it exceeds conc.

e.g.:

async with open_nursery_with_capacity(10) as nursery:
    for _ in range(30):
        await nursery.start_once_acquired(trio.sleep, 1)