Created
August 29, 2022 17:06
-
-
Save dot1mav/87f3b5796061239b828b2342aff746df to your computer and use it in GitHub Desktop.
use sync variable in threading
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from threading import Lock, Event | |
from functools import wraps | |
from copy import deepcopy | |
from typing import Callable, Optional, List, Any | |
class Variable: | |
_data: Any | |
_lock: Lock | |
_events: List[Event] | |
def __init__(self, variable: Any) -> None: | |
self._lock = Lock() | |
self._events = [] | |
self._data = variable | |
@property | |
def data(self) -> Any: | |
return deepcopy(self._data) | |
def __is_lock(self, event: Optional[Event] = None) -> None: | |
if self._lock.locked() or self._events: | |
if event is None: | |
event = Event() | |
self._events.append(event) | |
event.wait() | |
def __free_event(self) -> None: | |
if self._events: | |
event = self._events.pop(0) | |
event.set() | |
def __call__(self, func: Callable[[Any], Any]) -> Callable[[Any], Any]: | |
@wraps(func) | |
def wraps_(*args, **kwargs): | |
self.__is_lock() | |
with self._lock: | |
wraps_.__setattr__("data", self._data) | |
result: Any = func(*args, **kwargs) | |
wraps_.__delattr__("data") | |
self.__free_event() | |
return result | |
return wraps_ | |
def __enter__(self) -> Any: | |
self.__is_lock() | |
self._lock.acquire() | |
return self._data | |
def __exit__(self, *exc) -> None: | |
self._lock.release() | |
self.__free_event() | |
def __str__(self) -> str: | |
return str(self._data) | |
def __repr__(self) -> str: | |
return f"<Variable, data = {self._data}, is_use = {self._lock.locked()}, number_waiting = {len(self._events)} >" |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment