mainekuehn / usim Goto Github PK

The InterruptScope is currently available as usim.until:

async with until(time==20):
    ...

This gives the appearance that it works like a loop (while vs. until) and does not reflect the interrupt aspect much.

Possible alternatives:

• Speaking with out or with draw:

async with out(time==20):
    ...
async with draw(time==20):
    ...

Visually, a leading _ could reflect the connection to the with:

async with _out(time==20):
    ...

However, leading _ has reserved meaning in Python.

usim can cover/emulate most (not all) of simpy functionality. Since __await__ internally uses generators, we can translate from the coroutines to generators.

We could offer a compatibility layer that is a drop-in replacement for simpy.

The usim namespace currently consists of usim, usim.basics and usim.typing. Since there are no name conflicts, there really is no point requiring people to remember which is which (I forgot already...). The only valid distinction seems to be "things to use" and "things not to use".

I propose the following structure:

• usim everything that people are supposed to use,
• usim.typing everything only useful for annotations.

Special packages, such as usim.py, are not affected by this.

Description of the bug
When a loop ends and leaves behind objects, collecting them after a new loop starts transfers state between the two loops. This is because usim objects interact with the "current" event loop; delayed cleanup means the current loop is the wrong one.

How To Reproduce
No repro yet. Encountered this in lapis when cleaning an async generator is cleaned up.

Expected behavior
Objects should not leak. We might have to register objects and explicitly clean them up. Keep in mind that not registering was an explicit design decision to avoid overhead; if needed, restrict it to objects that need it.

• Consider sys.set_asyncgen_hooks to link async generators (the most likely culprit) to the event loop.
• Consider gc.collect whenever a loop finishes and/or starts.

I am under the impression that after the last update the simulation finishes after when the job queue is empty and all jobs are scheduled to drones but does not await the execution of all jobs in the drones job queue.

You can reproduce this by running lapis with the attached input files.

The simulation stops with the last scheduling cycle at 180 instead of processing the second job with a walltime of 1000.

I would expect the simulation to finish running all jobs.

Cluster input

TotalSlotCPUs TotalSlotDisk TotalSlotMemory Count sitename
1 44624348.0 8000 1 site1

Job Input

[
{
"QDate": 1567155456,
"RequestCpus": 1,
"RequestWalltime": 60,
"RequestMemory": 2000,
"RequestDisk": 6000000,
"RemoteWallClockTime": 100.0,
"'Number of Allocated Processors'": 1,
"MemoryUsage": 2867,
"DiskUsage_RAW": 41898,
"RemoteSysCpu": 10.0,
"RemoteUserCpu": 40.0,
"Inputfiles": {
"a.root": {
"filesize": 25,
"usedsize": 20
},
"b.root": {
"filesize": 25,
"usedsize": 20
},
"c.root": {
"filesize": 25,
"usedsize": 20
}
}
},
{
"QDate": 1567155456,
"RequestCpus": 1,
"RequestWalltime": 60,
"RequestMemory": 2000,
"RequestDisk": 6000000,
"RemoteWallClockTime": 77.0,
"'Number of Allocated Processors'": 1,
"MemoryUsage": 1207,
"DiskUsage_RAW": 45562,
"RemoteSysCpu": 7.0,
"RemoteUserCpu": 50.0,
"Inputfiles": {
"a.root": {
"filesize": 25,
"usedsize": 20
},
"b.root": {
"filesize": 25,
"usedsize": 20
},
"c.root": {
"filesize": 25,
"usedsize": 20
}
}
},
{
"QDate": 1567155456,
"RequestCpus": 1,
"RequestWalltime": 60,
"RequestMemory": 2000,
"RequestDisk": 6000000,
"RemoteWallClockTime": 63.0,
"'Number of Allocated Processors'": 1,
"MemoryUsage": 1207,
"DiskUsage_RAW": 45562,
"RemoteSysCpu": 3.0,
"RemoteUserCpu": 40.0,
"Inputfiles": {
"a.root": {
"filesize": 25,
"usedsize": 20
},
"b.root": {
"filesize": 25,
"usedsize": 20
},
"c.root": {
"filesize": 25,
"usedsize": 20
}
}
},
{
"QDate": 1567155456,
"RequestCpus": 1,
"RequestWalltime": 60,
"RequestMemory": 2000,
"RequestDisk": 6000000,
"RemoteWallClockTime": 1000.0,
"'Number of Allocated Processors'": 1,
"MemoryUsage": 1207,
"DiskUsage_RAW": 45562,
"RemoteSysCpu": 7.0,
"RemoteUserCpu": 40.3,
"Inputfiles": {
"a.root": {
"filesize": 25,
"usedsize": 20
},
"b.root": {
"filesize": 25,
"usedsize": 20
},
"c.root": {
"filesize": 25,
"usedsize": 20
}
}
}
]

Currently the CancelTask Interruption can only be imported through protected member via

from usim._primitives.task import CancelTask

which raises a warning. So I would like to propose to include it into the general API of usim.

Short description of the feature
Resources inherently lend themselves to await a specific level. A shorthand notation to derive conditions of the underlying Tracked value is desirable.

Describe the solution you'd like
Resources should directly support the comparison operators of Tracked values. The other should be either an appropriate resource.resource_type or a compatible dict.

resource == resource.resource_type(desired=value)
resource == dict(desired=value)
resource == {'desired': value}

Describe alternatives and the context you have considered
Manually deriving conditions requires access to a hidden field and a helper:

resource._available == resource.resource_type(desired=value)

In #44 there was the question how code that is repeatedly used in documentation files could be reused. See metronome example in about (

Description of the bug
A Concurrent Exception may not contain another Concurrent Exception. If one Scope failure cascades into another, this breaks error handling.

How To Reproduce

from usim import Scope, run

async def outer():
    async with Scope() as scope:
        scope.do(inner())

async def inner():
    async with Scope() as scope:
        scope.do(async_raise(KeyError))

async def async_raise(e):
    raise e

run(outer())

This fails with AssertionError: 'Concurrent' may only be specialised by Exception subclasses.

Expected behavior
Cascading Concurrent Exceptions should be well-defined. Either allow Concurrent to contain itself, or flatten contained exceptions.

SimPy came up with version 4 around 2 months ago. There were some changes, most importantly they now only work with Python 3.6+. Further, the following changes are mentioned in the documentation:

• BaseEnvironment has been dropped, Environment should be used instead
• Returning from Process generators now by return keyword: usage of Environment.exit(value) and raise StopProcess(value) should be replaced by return

Source: https://simpy.readthedocs.io/en/latest/topical_guides/porting_from_simpy3.html

We should double check if everything works as expected.

Semmle/LGTM has found various issues in the usim repository:

• The result of 'run' is used even though it is always None.

Description of the bug
An usim.interval loop that takes longer in the body than the interval duration aborts silently.

How To Reproduce
The following loop runs each print once, then aborts the loop silently.

from usim import run, interval, time


async def foo():
    async for _ in interval(1):
        print('before', time.now)
        await (time + 2)
        print('after', time.now)

run(foo())

Expected behavior
The loop should not abort silently -- either it aborts with an error, or proceeds at the next appropriate interval.

Description of the bug
If an activity receives an exception during scope shutdown (__aexit__) it is suppressed. The cause is that Scope.__aexit__ -> False is assumed to re-raise the current exception, but it only re-raises the exception with which the scope was entered.

How To Reproduce
In the following example, suppressing should be killed by a CancelTask. The usim.instant causes this to be received during Scope shutdown.

async def suppressing():
    async with usim.Scope() as scope:
        ...

async def setup():
    async with usim.Scope() as scope:
        task = scope.do(suppressing())
        await usim.instant  # advance suppressing Scope to shutdown
        task.cancel()
    await task

usim.run(setup())

Expected behaviour
The Scope should raise the most recent exception.

As discussed on #27, it would be sensible for the user interface to allow delays of 0. This seems prudent when calculating delays and for usability.

The event loop currently special-cases None to mean "now" and so disallows other ways of expressing this. Ideally, we handle this in the user interface implementations - these are already doing much more.

There are currently two user-facing cases:

• time + 0
• scope.do(after=0)

Various places use assert for optional user sanity checks, since these can be erased completely. The related __debug__ flag can similarly be used to add more helpful error messages:

This means there might be different behaviour depending on whether Python is run in debug mode (default) or optimised mode (-O flag). Testing should cover both cases:

Currently, some tests cover the optional sanity checks, thereby failing in optimised mode. This should be fixed alongside enabling both test modes.

The internal subscribe/unsubscribe API of notifications is partially refactored. Notification.__await_notification__ has a useless parameter and the subscribe context is not used yet. Probably some more unused/incomplete code.

Short description of the feature
Allow to pause the simulation and later on resume it.

Describe the solution you'd like
Simulation code should be able to pause at any time. This freezes the current simulation state.

async def some_activity():
     if time == 20:
        await usim.resume  # new call to pause simulation until resumed

Each simulation is represented by an object. usim.run returns this object when the simulation is frozen. The object allows unfreezing:

simulation = usim.run(some_activity())
print("Current state:", simulation)
simulation.resume()

Describe alternatives and the context you have considered
There is currently no way to freeze a simulation.

Following #27, Tasks of a Scope are only cleaned up once at the end of the scope. This means that the Task object may remain alive longer than needed. In addition, Tasks created during graceful shutdown of a Sope may have ill-defined semantics.

usim currently knows two kinds of Awaitable objects:

1. bare Awaitable such as coroutines or objects with an __await__ method
2. wrapped Awaitable produced by Scope.do

Awaitable is a technical term that also applies to other frameworks (e.g. asyncio). The usim documentation thus calls "an Awaitable made for a usim simulation" an activity. This makes both 1 and 2 an activity.

However, 2 is represented by the class Activity. This leads to the rather confusing case that "an Activity is an activity wrapping an activity". The use of "rich" and "bare" activity is not helping.

At least one of 1 or 2 should be renamed to make more sense overall.

See also #2 .

The usim.each has now repeatedly led to confusion. It does not integrate well with async for wording (for each also means different things in e.g. C++) and having to use keywords makes the each word redundant. We could split each into its two use cases and name them directly:

# old: specify interval *or* delay
async for delta in each(interval=10):
    ...

# new
async for delta in interval(10):
    ...

Currently the usim.basics.Capacities only supports getting the current levels of underlying resources. However, it makes sense to also query for the maximum limit of resources available.

The Delay (time + delay) is the only time expression that produces a Notification instead of a Condition. It should provide rich __debug__ exceptions in case of accidental misuse as a Condition.

When trying to borrow resources with two competing activities at the same time where one already books all resources, the other does not recognise that resources are not available until doing the actual borrowing of resources.

• Write a unit test

Tests are passing in principle. CI failures are due to flake8 complaining about pre-existing code.

Originally posted by @maxfischer2781 in #93 (comment)

Description of the bug
If Queue.__aiter__ is closed via garbage collection, it still attempts to await. Side-effects of this are implementation dependent.

How To Reproduce
No idea how to reproduce this. It seems that CPython silently swallows the await, probably by replying None. PyPy fails to close the iterator gracefully.

Exception ignored in: finalizer of <async generator object Queue.__aiter__ at 0x0000000107103b48>
Traceback (most recent call last):
  File "/usr/local/Cellar/pypy3/7.1.1_1/libexec/site-packages/usim/_basics/streams.py", line 168, in _await_message
    await self._notification
GeneratorExit

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/Cellar/pypy3/7.1.1_1/libexec/site-packages/usim/_basics/streams.py", line 176, in __aiter__
    yield await self
  File "/usr/local/Cellar/pypy3/7.1.1_1/libexec/site-packages/usim/_basics/streams.py", line 158, in __await__
    return (yield from self._await_message().__await__())  # noqa: B901
  File "/usr/local/Cellar/pypy3/7.1.1_1/libexec/site-packages/usim/_basics/streams.py", line 171, in _await_message
    return self._buffer.popleft()
RuntimeError: PyPy limitation: cannot use 'yield from' or 'await' in a generator/coroutine that has been partially deallocated already, typically seen via __del__

Expected behavior
Deterministic cleanup on all platforms.

Exceptions currently are less than ideal for debugging because much relevant information is obscured. This is especially critical because async programming is already complicated.

• Concurrent errors do not contain tracebacks to their actual source.
• ActivityError needlessly blows up the trace with internals.

Short description of the feature
Provide a context that postpones exactly once, even when nested. This would allow combining multiple high-level calls as a quasi-atomic async operation.

This is intended as a helper for creating high-level operations. It is required to build new abstractions; users of existing abstractions do not need it.

Describe the solution you'd like
Add an async with context that opens a postponement scope or resumes an outer one. The outermost context will postpone if it is exited at the same time and turn as it was entered. Nested scopes delegate the responsibility to outer scopes.

Postponement is dynamically scoped, meaning that any postponement in an __await__, __aenter__/__aexit__ or __aiter__/__anext__ is considered nested in any outer postponements.

async with postponed():  # open scope
    async with postponed():  # resume scope
        await something()  # resume scope?

Enforcing only postponement, not suspension, is optional.

Describe alternatives and the context you have considered
This has originated from #64. In such a specific case, a special cased implementation could be made. However, these special cases must be carefully constructed to avoid duplicate postponement. Combining several high-level calls is very difficult.

Short description of the feature
Provide helpers/functions for common async patterns.

Describe the solution you'd like
A collection of efficient, reusable primitives to compose async operations:

• async versions of common Python helpers such as map, reduce, filter, ...
• concurrency specific helpers, such as collect(*:Awaitable) to concurrently evaluate several activities

async def collect(*activities: Awaitable):
    async with Scope() as scope:
        tasks = [scope.do(activity) for activity in activities]
    return [await task for task in tasks]

Describe alternatives and the context you have considered
There are basically no separate async helpers in the standard library. Looking at existing libraries shows that pretty much all helpers are tied to specific async libraries.

Since usim doesn't need any fancy setup.py magic, we could probably switch to flit easily.

We now have a Zenodo DOI. Should be added to the docs.

The documentation does not seem to show inherited methods from super classes. This should be improved as one cannot work with the documentation itself but requires the actual code to do the implementations.

The time iterators IntervalIter and DurationIter, and by extension each, treat the first time interval differently. In an async for now in each(...)::

• IntervalIter starts at the time at which each was entered
• DurationIter starts after each has enter and one delay has passed

start_time = time.now
async for now in each(...):
    assert now == start_time  # True for IntervalIter, False for DurationIter

The Concurrent exception introduced in #27 only implements the except Concurrent[E]: matching, not any tools to conveniently work with the error: errors must be manually extracted from err.exceptions.

For example, trio allows to fetch individual, concrete exceptions by type and to split exception groups to handle them partially.

It seems that Scope does not properly catch-reraise exceptions in children. It still misses the facilities to detect errors in children and abort itself.

The Connective used in connecting Conditions (a & b & c | d) does not properly unpack on chaining. The issue is the second set in Connective.__init__, which does not unpack:

        self._children = tuple(
            set(condition for condition in conditions if not isinstance(condition, self.__class__)) |
            set(condition._children for condition in conditions if isinstance(condition, self.__class__))
        )

The _children are "unpacked" to ((a, b), c) instead of (a, b, c).

Short description of the feature
The usim documentation introduced in #58 is organised into topics, but mostly only includes the raw, automatic object documentation. Manual documentation on how objects can be used still needs to be added.

Describe the solution you'd like
We should manually improve each subtopic with additional information. Since topics are distinct, it would be fine to have separate PRs for each topic.

See also issue #36, which can be solved along the way.

Short description of the feature
Cancelling a task currently a) allows the task to delay/ignore the cancellation and b) does not wait for completion. This means the cancelling task has to manually account for shutdown (e.g. MatterMiners/lapis#80).

Instead, we should offer an API that cancels a task and waits for completion. Notably, our internal, non-async API (Task.__close__) relies on prompt and successful termination.

Describe the solution you'd like
A public, asynchronous method for immediate shutdown. Roughly equivalent to:

class Task:
    ...

    async def terminate(self):
        self.__cancel__()
        await self.done

I'm still not 100% sure if we should use Coroutine.close(via .__close__) or add a cancellation (like .cancel currently does). close has guaranteed semantics, but allows no async action by the Task. A cancellation may be easier to handle in the activity, but requires a check for promptness.

Description of the bug
The asynchronous Resource.claim operation does not postpone when claiming fails. This violates usim's guarantee to always postpone on an asynchronous operation.

This issue may apply to other operations as well.

How To Reproduce
Performing an impossible claim does not allow other activities to run.

from usim import instant, run, Resources, ResourcesUnavailable

async def fail_postponed():
    res = Resources(capacity=1)
    try:
        async with res.claim(capacity=2):
            pass
    except ResourcesUnavailable:
        print("Postponed?")

async def postponed():
    print("Before postponement...")
    await instant
    print("After postponement...")

run(fail_postponed(), postponed())

Expected behavior
Any asynchronous operation should postpone at least once before raising a non-failure Exception.

Description of the bug
The loop state threading.local is not initialised properly -- only the main thread initialises it.

How To Reproduce
Using any feature that accesses the loop from a thread outside of a loop triggers AttributeError instead of the prepared RuntimeError.

>>> import usim, threading
>>> def fail():
...     print(usim.time.now)
...
>>> fail()
...
RuntimeError: field 'time' can only be accessed with an active usim event loop
...
>>> t = threading.Thread(target=fail)
>>> t.start()
...
AttributeError: '_thread._local' object has no attribute 'LOOP'

Expected behavior
Behaviour should be consistent from main and child threads.

Porting SimPy resources (#51 and #50) made it pretty obvious that uSim's Resources type is a bit awkward when there really is just one type of resources.

resources = Resources(capacity=8)
await resource.increase(capacity=8)
await resource.decrease(capacity=8)
async with resource.borrow(capacity=8):
    ...

Other ways to emulate a Semaphore are Tracked and Queue, which come with other unwanted baggage. IMO the Resources interface is good, but we may want a "just one type" version:

resources = Resource(8)
await resource.increase(8)
await resource.decrease(8)
async with resource.borrow(8):
    ...

The use-case of a raw Semaphore would be easily handled by having proper defaults:

resources = Resource()  # default level of 0
await resource.increase()  # default in/decrease of 1
await resource.decrease()
async with resource.borrow():  # default borrow/claim of 1
    ...

Most objects are missing useful representations. For example, failing an assert time == 20 tests produces the following message:

E       assert <usim._primitives.timing.Time object at 0x10aebe890> == 20

Short description of the feature
The documentation for ResourceLevels is lacking, and the eval source is extremely hard to understand. Provide full description and examples.

Describe the solution you'd like
Document automatic features. Provide examples for frequently encountered misuses.

# antipattern
dict(self._usedstorage.levels)["usedsize"]
# direct access
self._usedstorage.levels.usedsize