neclib.core.security.busy_impl#

Control executions of conflict-unsafe methods.

class busy(obj, flagname)[source]#

Bases: object

Manages the busy state of the object.

This context manager blocks and stops the execution of its contents until the object confirmed to be not busy. When the context manager exits, it frees the busy flag. This functionality would be useful to avoid conflicting operations to be executed at the same time.

Parameters:

obj (object) – The object to manage the busy state of.
flagname (str) – The name of the attribute to use to store the busy state.

Examples

>>> class Foo:
...     def task_a(self):
...         with neclib.core.logic.busy(self, "busy"):
...             print("Task A")
...             time.sleep(1)
...             print("Task A done")
...     def task_b(self):
...         with neclib.core.logic.busy(self, "busy"):
...             print("Task B")
...             time.sleep(0.5)
...             print("Task B done")
>>> with concurrent.futures.ThreadPoolExecutor() as executor:
...     foo = Foo()
...     future1 = executor.submit(foo.task_a)
...     future2 = executor.submit(foo.task_b)
...     concurrent.futures.wait([future1, future2])

The above example attempts to execute two tasks concurrently. However, the busy context prevents the second task from starting until the first task finishes.

Attention

The flagname argument must be the same across all busy contexts which blocks each other. To manage multiple busy states, use different flagname-s to distinguish the task groups.

property busy: bool#: The busy state of the object.