Repository: MaineKuehn/usim

Issue:
Thread safety/initialisation of __LOOP_STATE__
**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``.
```pyrthon3
>>> 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.


Discussion:

Patch:
diff --git a/usim/_basics/pipe.py b/usim/_basics/pipe.py
index ed51b15..8729b17 100644
--- a/usim/_basics/pipe.py
+++ b/usim/_basics/pipe.py
@@ -1,7 +1,7 @@
 from typing import Optional, Dict
 
 from .._primitives.notification import Notification, suspend, postpone
-from .._core.loop import __LOOP_STATE__
+from .._primitives.timing import time
 
 
 class Pipe:
@@ -67,7 +67,7 @@ class Pipe:
         throughput = throughput if throughput is not None else self.throughput
         self._add_subscriber(identifier, throughput)
         while transferred < total:
-            window_start = __LOOP_STATE__.LOOP.time
+            window_start = time.now
             window_throughput = throughput * self._throughput_scale
             # Try to delay until we have transferred everything.
             # Be prepared to get interrupted if throughput changes.
@@ -80,7 +80,7 @@ class Pipe:
                 # At this point, we have been suspended for as long as calculated.
                 # Barring float *imprecision* we have transferred the desired volume.
                 transferred = total
-            window_end = __LOOP_STATE__.LOOP.time
+            window_end = time.now
             transferred += (window_end - window_start) * window_throughput
         self._del_subscriber(identifier)
 
diff --git a/usim/_basics/resource.py b/usim/_basics/resource.py
index 7ec42e6..a97f0c8 100644
--- a/usim/_basics/resource.py
+++ b/usim/_basics/resource.py
@@ -1,6 +1,6 @@
 from typing import TypeVar, Generic, Optional, Type, Union, Dict
 
-from .._core.loop import __LOOP_STATE__
+from .._core.handler import __USIM_STATE__
 from ._resource_level import __specialise__, ResourceLevels
 from .tracked import Tracked
 
@@ -134,10 +134,10 @@ class BorrowedResources(BaseResources[T]):
         if exc_type is GeneratorExit:
             # we are killed forcefully and cannot perform async operations
             # dispatch a new activity to release our resources eventually
-            __LOOP_STATE__.LOOP.schedule(
+            __USIM_STATE__.loop.schedule(
                 self.__remove_resources__(self._debits)
             )
-            __LOOP_STATE__.LOOP.schedule(
+            __USIM_STATE__.loop.schedule(
                 self._resources.__insert_resources__(self._debits)
             )
         else:
diff --git a/usim/_core/handler.py b/usim/_core/handler.py
new file mode 100644
index 0000000..00d0a15
--- /dev/null
+++ b/usim/_core/handler.py
@@ -0,0 +1,89 @@
+r"""
+Generic loop effect handler
+
+This is the core implementation allowing simulation objects to
+have "effects" – e.g. trigger activities or read the time.
+"""
+from typing_extensions import Protocol, Coroutine
+import threading
+import contextlib
+
+
+class AbstractLoop(Protocol):
+    #: The currently running activity
+    activity: Coroutine
+    #: The current time of the simulation
+    time: float
+
+
+class MissingLoop:
+    r"""
+    Placeholder for the :py:class:`~.Loop` expected in an active simulation
+
+    This class exists to provide helpful error messages if usim is used
+    without starting its event loop. If you encounter this class unexpectedly,
+    see 'https://usim.readthedocs.io' for usage details.
+    """
+    __slots__ = ("_entry_point",)
+    activity: Coroutine
+    time: float
+
+    def __init__(self, entry_point: str = "usim.run"):
+        self._entry_point = entry_point
+
+    def __getattr__(self, field: str):
+        raise RuntimeError(
+            (
+                "field '%s' can only be accessed with an active usim event loop\n\n"
+                "You have attempted to use an async feature of usim outside of\n"
+                "a usim simulation. This likely means that you used a different\n"
+                "async framework. You must run usim's async features as part of\n"
+                "an active usim simulation.\n\n"
+                "Use '%s' to start an appropriate simulation."
+            )
+            % (field, self._entry_point)
+        )
+
+    def __repr__(self):
+        return "{self.__class__.__name__}(entry_point={self._entry_point})".format(
+            self=self,
+        )
+
+
+class StateHandler(threading.local):
+    """
+    State of the current simulation
+
+    This class implements a basic effect handler for interactions
+    with the event loop: A single instance provides access to the
+    "current" event loop and simulation state from inside activities.
+
+    This class is thread-aware and always represents the simulation
+    active in the current thread, if any. Multiple simulations may
+    be nested.
+    """
+
+    loop: AbstractLoop
+
+    @property
+    def is_active(self) -> bool:
+        """Whether there is an active simulation going on"""
+        return type(self.loop) is not MissingLoop
+
+    def __init__(self):
+        self.loop = MissingLoop()
+
+    @contextlib.contextmanager
+    def assign(self, loop: AbstractLoop):
+        """Temporarily set a ``loop`` as the "current" loop of this thread"""
+        outer_loop, self.loop = self.loop, loop
+        try:
+            yield
+        finally:
+            self.loop = outer_loop
+
+
+#: Current state of the usim simulation
+#: This is a global, thread-aware object representing the state
+#: of all threads using usim.
+__USIM_STATE__ = StateHandler()
diff --git a/usim/_core/loop.py b/usim/_core/loop.py
index 2cee0b7..99fb1ea 100644
--- a/usim/_core/loop.py
+++ b/usim/_core/loop.py
@@ -1,5 +1,5 @@
 r"""
-event handling core facilities
+Single-threaded event loop
 
 This is the core scheduling/synchronization implementation.
 It defines a basic event loop with simulated time, as well as the signals
@@ -8,9 +8,9 @@ The entirety of this module is for internal use only, and should not be used
 by simulation code directly.
 """
 import collections
-import threading
-from typing import Coroutine, TypeVar, Optional, Union, Generator
+from typing import Coroutine, TypeVar, Optional, Generator
 
+from .handler import __USIM_STATE__ as __LOOP_STATE__
 from .waitq import WaitQueue
 
 
@@ -32,39 +32,6 @@ class ActivityLeak(RuntimeError):
 RT = TypeVar('RT')
 
 
-class MissingLoop:
-    r"""
-    Placeholder for the :py:class:`~.Loop` expected in an active simulation
-
-    This class exists to provide helpful error messages if usim is used
-    without starting its event loop. If you encounter this class unexpectedly,
-    see 'https://usim.readthedocs.io' for details.
-    """
-    __slots__ = ('_entry_point',)
-
-    def __init__(self, entry_point: str = 'usim.run'):
-        self._entry_point = entry_point
-
-    def __getattr__(self, field: str):
-        raise RuntimeError((
-            "field '%s' can only be accessed with an active usim event loop\n\n"
-            "You have attempted to use an async feature of usim outside of\n"
-            "a usim simulation. This likely means that you used a different\n"
-            "async framework. You must run usim's async features as part of\n"
-            "an active usim simulation.\n\n"
-            "Use '%s' to start an appropriate simulation."
-        ) % (field, self._entry_point))
-
-    def __repr__(self):
-        return '{self.__class__.__name__}(entry_point={self._entry_point})'.format(
-            self=self,
-        )
-
-
-__LOOP_STATE__ = threading.local()
-__LOOP_STATE__.LOOP = MissingLoop()  # type: Union[MissingLoop, Loop]
-
-
 # Commands
 class Hibernate(object):
     r"""
@@ -181,12 +148,8 @@ class Loop:
 
     def run(self):
         r"""Run the event loop in the current thread"""
-        outer_loop = __LOOP_STATE__.LOOP
-        __LOOP_STATE__.LOOP = self
-        try:
+        with __LOOP_STATE__.assign(self):
             self._run_events()
-        finally:
-            __LOOP_STATE__.LOOP = outer_loop
 
     def _run_events(self):
         r"""event loop core, processing all scheduled coroutines"""
diff --git a/usim/_primitives/condition.py b/usim/_primitives/condition.py
index ad2cb0d..158acd9 100644
--- a/usim/_primitives/condition.py
+++ b/usim/_primitives/condition.py
@@ -5,7 +5,7 @@ from typing import Coroutine, Generator, Any as AnyT
 from .._core.loop import Hibernate, Interrupt as CoreInterrupt
 
 from .notification import Notification, postpone
-from .._core.loop import __LOOP_STATE__
+from .._core.handler import __USIM_STATE__
 
 
 class Condition(Notification):
@@ -88,7 +88,7 @@ class Condition(Notification):
         # triggering will interrupt on the next async operation
         if self:
             interrupt.scheduled = True
-            __LOOP_STATE__.LOOP.schedule(waiter, signal=interrupt)
+            __USIM_STATE__.loop.schedule(waiter, signal=interrupt)
         else:
             super().__subscribe__(waiter, interrupt)
 
diff --git a/usim/_primitives/context.py b/usim/_primitives/context.py
index ae9a742..c6a40c5 100644
--- a/usim/_primitives/context.py
+++ b/usim/_primitives/context.py
@@ -1,6 +1,7 @@
 from typing import Coroutine, List, TypeVar, Any, Optional, Tuple
 
-from .._core.loop import __LOOP_STATE__, Interrupt as CoreInterrupt
+from .._core.loop import Interrupt as CoreInterrupt
+from .._core.handler import __USIM_STATE__
 from .notification import Notification
 from .flag import Flag
 from .task import Task, TaskClosed, TaskCancelled, try_close
@@ -179,14 +180,14 @@ class Scope:
         # resolve "now" to what the event loop expects
         if after == 0:
             after = None
-        elif at == __LOOP_STATE__.LOOP.time:
+        elif at == __USIM_STATE__.loop.time:
             at = None
         assert after is None or after > 0,\
             "start date must not be in the past"
-        assert at is None or at > __LOOP_STATE__.LOOP.time,\
+        assert at is None or at > __USIM_STATE__.loop.time,\
             "start date must not be in the past"
         child_task = Task(payload, self, delay=after, at=at, volatile=volatile)
-        __LOOP_STATE__.LOOP.schedule(
+        __USIM_STATE__.loop.schedule(
             child_task.__runner__,
         )
         if not volatile:
@@ -198,7 +199,7 @@ class Scope:
     def __cancel__(self):
         """Cancel this scope"""
         if self._interruptable:
-            __LOOP_STATE__.LOOP.schedule(self._activity, self._cancel_self)
+            __USIM_STATE__.loop.schedule(self._activity, self._cancel_self)
 
     def __child_finished__(self, child: Task, failed: bool):
         assert child.parent is self
@@ -234,7 +235,7 @@ class Scope:
     async def __aenter__(self):
         if self._activity is not None:
             raise RuntimeError('%r is not re-entrant' % self.__class__.__name__)
-        self._activity = __LOOP_STATE__.LOOP.activity
+        self._activity = __USIM_STATE__.loop.activity
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb) -> bool:
diff --git a/usim/_primitives/locks.py b/usim/_primitives/locks.py
index ea79a75..4bba3c3 100644
--- a/usim/_primitives/locks.py
+++ b/usim/_primitives/locks.py
@@ -1,6 +1,6 @@
 from typing import Coroutine
 
-from .._core.loop import __LOOP_STATE__
+from .._core.handler import __USIM_STATE__
 from .notification import Notification, NoSubscribers
 
 
@@ -53,10 +53,10 @@ class Lock:
         if self._owner is None:
             return True
         else:
-            return self._owner is __LOOP_STATE__.LOOP.activity
+            return self._owner is __USIM_STATE__.loop.activity
 
     async def __aenter__(self):
-        current_activity = __LOOP_STATE__.LOOP.activity
+        current_activity = __USIM_STATE__.loop.activity
         if self._owner is None:
             self._owner = current_activity
         elif self._owner is not current_activity:
@@ -71,7 +71,7 @@ class Lock:
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
-        assert exc_type is GeneratorExit or self._owner == __LOOP_STATE__.LOOP.activity
+        assert exc_type is GeneratorExit or self._owner == __USIM_STATE__.loop.activity
         self._depth -= 1
         if self._depth == 0:
             self.__release__()
diff --git a/usim/_primitives/notification.py b/usim/_primitives/notification.py
index b4c6218..a790b29 100644
--- a/usim/_primitives/notification.py
+++ b/usim/_primitives/notification.py
@@ -1,7 +1,8 @@
 from typing import List, Tuple, Coroutine, Optional
 from contextlib import contextmanager
 
-from .._core.loop import Interrupt, __LOOP_STATE__, __HIBERNATE__
+from .._core.loop import Interrupt, __HIBERNATE__
+from .._core.handler import __USIM_STATE__
 
 
 # TODO: add protocol for destroying a notification
@@ -18,15 +19,16 @@ async def postpone():
     This will safely requeue the current task,
     allowing other tasks to run and interrupts to occur.
     """
-    task = __LOOP_STATE__.LOOP.activity
+    loop = __USIM_STATE__.loop
+    task = loop.activity
     wake_up = Interrupt('postpone', task)
-    __LOOP_STATE__.LOOP.schedule(task, signal=wake_up)
+    loop.schedule(task, signal=wake_up)
     try:
         await __HIBERNATE__
     except Interrupt as err:
         if err is not wake_up:
             assert (
-                task is __LOOP_STATE__.LOOP.activity
+                task is loop.activity
             ), 'Break points cannot be passed to other coroutines'
             raise
     finally:
@@ -42,15 +44,16 @@ async def suspend(*, delay: Optional[float], until: Optional[float]):
     Time will pass as if ``time == until`` or ``time + delay``
     were used, but there is no ``Condition`` interface on top.
     """
-    task = __LOOP_STATE__.LOOP.activity
+    loop = __USIM_STATE__.loop
+    task = loop.activity
     wake_up = Interrupt('postpone', task)
-    __LOOP_STATE__.LOOP.schedule(task, signal=wake_up, delay=delay, at=until)
+    loop.schedule(task, signal=wake_up, delay=delay, at=until)
     try:
         await __HIBERNATE__
     except Interrupt as err:
         if err is not wake_up:
             assert (
-                task is __LOOP_STATE__.LOOP.activity
+                task is loop.activity
             ), 'Break points cannot be passed to other coroutines'
             raise
     finally:
@@ -84,7 +87,7 @@ class Notification:
         except IndexError:
             raise NoSubscribers
         else:
-            __LOOP_STATE__.LOOP.schedule(waiter, signal=interrupt)
+            __USIM_STATE__.loop.schedule(waiter, signal=interrupt)
             return waiter, interrupt
 
     def __awake_all__(self) -> List[Tuple[Coroutine, Interrupt]]:
@@ -92,7 +95,7 @@ class Notification:
         awoken = self._waiting.copy()
         self._waiting.clear()
         for waiter, interrupt in awoken:
-            __LOOP_STATE__.LOOP.schedule(waiter, signal=interrupt)
+            __USIM_STATE__.loop.schedule(waiter, signal=interrupt)
         return awoken
 
     # Subscribe/Unsubscribe
@@ -109,7 +112,8 @@ class Notification:
 
     @contextmanager
     def __subscription__(self):
-        task = __LOOP_STATE__.LOOP.activity
+        loop = __USIM_STATE__.loop
+        task = loop.activity
         wake_up = Interrupt(self, task)
         self.__subscribe__(task, wake_up)
         try:
@@ -117,7 +121,7 @@ class Notification:
         except Interrupt as err:
             if err is not wake_up:
                 assert (
-                    task is __LOOP_STATE__.LOOP.activity
+                    task is loop.activity
                 ), 'Break points cannot be passed to other coroutines'
                 raise
         finally:
diff --git a/usim/_primitives/task.py b/usim/_primitives/task.py
index c0fdecf..63bfa5f 100644
--- a/usim/_primitives/task.py
+++ b/usim/_primitives/task.py
@@ -4,7 +4,8 @@ import enum
 from typing import Coroutine, TypeVar, Awaitable, Optional, Tuple, Any, List,\
     TYPE_CHECKING
 
-from .._core.loop import __LOOP_STATE__, Interrupt
+from .._core.loop import Interrupt
+from .._core.handler import __USIM_STATE__
 from .notification import suspend
 from .condition import Condition
 if TYPE_CHECKING:
@@ -253,7 +254,7 @@ class Task(Awaitable[RT]):
                 cancellation = CancelTask(self, *token)
                 self._cancellations.append(cancellation)
                 cancellation.scheduled = True
-                __LOOP_STATE__.LOOP.schedule(self.__runner__, signal=cancellation)
+                __USIM_STATE__.loop.schedule(self.__runner__, signal=cancellation)
 
     @reprlib.recursive_repr()
     def __repr__(self):
diff --git a/usim/_primitives/timing.py b/usim/_primitives/timing.py
index f3a3d2d..75973f2 100644
--- a/usim/_primitives/timing.py
+++ b/usim/_primitives/timing.py
@@ -14,7 +14,8 @@ time can represent more than 285 million years of time accurately.
 """
 from typing import Coroutine, Generator, Any, AsyncIterable, Union
 
-from .._core.loop import __LOOP_STATE__, __HIBERNATE__, Interrupt as CoreInterrupt
+from .._core.loop import __HIBERNATE__, Interrupt as CoreInterrupt
+from .._core.handler import __USIM_STATE__
 from .notification import postpone, suspend, Notification
 from .condition import Condition
 
@@ -41,7 +42,7 @@ class After(Condition):
         self._scheduled = None
 
     def __bool__(self):
-        return __LOOP_STATE__.LOOP.time >= self.date
+        return __USIM_STATE__.loop.time >= self.date
 
     def __invert__(self):
         return Before(self.date)
@@ -49,7 +50,7 @@ class After(Condition):
     def _ensure_trigger(self):
         if not self._scheduled:
             self._scheduled = True
-            __LOOP_STATE__.LOOP.schedule(self._async_trigger(), at=self.date)
+            __USIM_STATE__.loop.schedule(self._async_trigger(), at=self.date)
 
     # we cannot schedule __trigger__ directly, since it is not async
     async def _async_trigger(self):
@@ -98,7 +99,7 @@ class Before(Condition):
         self.date = date
 
     def __bool__(self):
-        return __LOOP_STATE__.LOOP.time < self.date
+        return __USIM_STATE__.loop.time < self.date
 
     def __invert__(self):
         return After(self.date)
@@ -144,7 +145,7 @@ class Moment(Condition):
         self._transition = After(date)
 
     def __bool__(self):
-        return __LOOP_STATE__.LOOP.time == self.date
+        return __USIM_STATE__.loop.time == self.date
 
     def __invert__(self):
         raise NotImplementedError(
@@ -158,7 +159,7 @@ class Moment(Condition):
         # we will *never* wake up once the target has passed
         # either we wake up in the same time frame,
         # or just hibernate indefinitely
-        if __LOOP_STATE__.LOOP.time == self.date:
+        if __USIM_STATE__.loop.time == self.date:
             yield from postpone().__await__()
         elif not self._transition:
             yield from self._transition.__await__()
@@ -277,7 +278,7 @@ class Delay(Notification):
 
     def __subscribe__(self, waiter: Coroutine, interrupt: CoreInterrupt):
         interrupt.scheduled = True
-        __LOOP_STATE__.LOOP.schedule(waiter, interrupt, delay=self.duration)
+        __USIM_STATE__.loop.schedule(waiter, interrupt, delay=self.duration)
 
     def __repr__(self):
         return f'{self.__class__.__name__}(duration={self.duration})'
@@ -385,7 +386,7 @@ class Time:
     @property
     def now(self) -> float:
         """The current simulation time"""
-        return __LOOP_STATE__.LOOP.time
+        return __USIM_STATE__.loop.time
 
     def __add__(self, other: float) -> Union[Delay, Instant]:
         assert other >= 0, "delay must point at the future"
@@ -491,8 +492,7 @@ async def interval(period) -> AsyncIterable[float]:
     """
     if period < 0:
         raise ValueError('period must not be negative')
-    loop = __LOOP_STATE__.LOOP
-    last_time = loop.time
+    last_time = time.now
     while True:
         remaining_delay = last_time + period - time.now
         if remaining_delay < 0:
@@ -501,7 +501,7 @@ async def interval(period) -> AsyncIterable[float]:
             await suspend(delay=remaining_delay, until=None)
         else:
             await postpone()
-        last_time = loop.time
+        last_time = time.now
         yield last_time
 
 
@@ -530,12 +530,11 @@ async def delay(period) -> AsyncIterable[float]:
     """
     if period < 0:
         raise ValueError('period must not be negative')
-    loop = __LOOP_STATE__.LOOP
     if period > 0:
         while True:
             await suspend(delay=period, until=None)
-            yield loop.time
+            yield time.now
     else:
         while True:
             await postpone()
-            yield loop.time
+            yield time.now
diff --git a/usim/py/core.py b/usim/py/core.py
index 0656c35..ccaa2a5 100644
--- a/usim/py/core.py
+++ b/usim/py/core.py
@@ -9,7 +9,7 @@ The latter allows interactions between the μSim and SimPy components.
 """
 from typing import Optional, List, Tuple, Coroutine, Generator, TypeVar, Iterable,\
     Union
-from .._core.loop import __LOOP_STATE__, Loop
+from .._core.handler import __USIM_STATE__, AbstractLoop
 from .. import time, run as usim_run, Concurrent
 from .. import Scope
 
@@ -24,16 +24,6 @@ class EnvironmentScope(Scope):
     PROMOTE_CONCURRENT = Scope.PROMOTE_CONCURRENT + (StopSimulation,)
 
 
-def _inside_usim():
-    """Whether the current stack is run by ``usim``"""
-    try:
-        __LOOP_STATE__.LOOP.time
-    except RuntimeError:
-        return False
-    else:
-        return True
-
-
 V = TypeVar('V')
 
 
@@ -111,7 +101,7 @@ class Environment:
     def __init__(self, initial_time=0):
         self._initial_time = initial_time
         self._startup = []  # type: List[Tuple[Coroutine, float]]
-        self._loop = None  # type: Optional[Loop]
+        self._loop = None  # type: Optional[AbstractLoop]
         self._scope = EnvironmentScope()
         #: The currently active process
         self.active_process = None  # type: Optional[Process]
@@ -119,7 +109,7 @@ class Environment:
     async def __aenter__(self):
         if self._loop is not None:
             raise RuntimeError('%r is not re-entrant' % self.__class__.__name__)
-        self._loop = __LOOP_STATE__.LOOP
+        self._loop = __USIM_STATE__.loop
         await self._scope.__aenter__()
         if self._loop.time < self._initial_time:
             await (time == self._initial_time)
@@ -166,7 +156,7 @@ class Environment:
         otherwise
             The sub-simulation lasts until simulation time equals ``until``.
         """
-        if not _inside_usim():
+        if not __USIM_STATE__.is_active:
             usim_run(self.until(until))
             if isinstance(until, Event):
                 if until.triggered:


Test patch:
diff --git a/usim_pytest/test_core.py b/usim_pytest/test_core.py
index 8803169..9b472d8 100644
--- a/usim_pytest/test_core.py
+++ b/usim_pytest/test_core.py
@@ -1,7 +1,10 @@
+from concurrent.futures import ThreadPoolExecutor
+
 import pytest
 
 from usim import time, run
-from usim._core.loop import __LOOP_STATE__, Loop, ActivityLeak
+from usim._core.loop import Loop, ActivityLeak
+from usim._core.handler import __USIM_STATE__
 
 
 class TestCore:
@@ -10,7 +13,7 @@ class TestCore:
             time.now
         for field in Loop.__slots__:
             with pytest.raises(RuntimeError):
-                getattr(__LOOP_STATE__.LOOP, field)
+                getattr(__USIM_STATE__.loop, field)
 
     def test_after_sim(self):
         run()
@@ -23,3 +26,14 @@ class TestCore:
         with pytest.raises(ActivityLeak) as exc_info:
             run(returning(1138))
         assert exc_info.value.result == 1138
+
+    def test_threaded(self):
+        """Test that behaviour is consistent in threads"""
+        with ThreadPoolExecutor() as executor:
+            for test_case in (
+                self.test_no_sim,
+                self.test_after_sim,
+                self.test_exception,
+            ):
+                threaded_test = executor.submit(test_case)
+                threaded_test.result()
diff --git a/usim_pytest/test_tests.py b/usim_pytest/test_tests.py
index cf5e9e3..2528c38 100644
--- a/usim_pytest/test_tests.py
+++ b/usim_pytest/test_tests.py
@@ -1,3 +1,5 @@
+from concurrent.futures import ThreadPoolExecutor
+
 import pytest
 
 from usim import Scope, eternity, Concurrent
@@ -39,3 +41,12 @@ class TestTests:
     @via_usim
     async def test_hanging(self):
         await eternity
+
+    @pytest.mark.xfail(raises=AssertionError, strict=True)
+    def test_threaded(self):
+        def fail_assert():
+            assert False
+
+        with ThreadPoolExecutor() as executor:
+            fail_thread = executor.submit(fail_assert)
+            fail_thread.result()
diff --git a/usim_pytest/utility.py b/usim_pytest/utility.py
index 63277e9..a5efd6e 100644
--- a/usim_pytest/utility.py
+++ b/usim_pytest/utility.py
@@ -5,7 +5,7 @@ from collections import namedtuple
 import pytest
 
 from usim import run
-from usim._core.loop import __LOOP_STATE__
+from usim._core.handler import __USIM_STATE__
 
 
 RT = TypeVar('RT')
@@ -30,7 +30,7 @@ def noop(*args, **kwargs):
 
 def turnstamp() -> Turnstamp:
     """Get the precise progress as ``time, turn``"""
-    loop = __LOOP_STATE__.LOOP
+    loop = __USIM_STATE__.loop
     return Turnstamp(loop.time, loop.turn)