1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
# This file is part of Pebble.
# Copyright (c) 2013-2025, Matteo Cafasso
# Pebble is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License
# as published by the Free Software Foundation,
# either version 3 of the License, or (at your option) any later version.
# Pebble is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
# You should have received a copy of the GNU Lesser General Public License
# along with Pebble. If not, see <http://www.gnu.org/licenses/>.
import threading
from time import time
from types import MethodType
from typing import Callable, Optional
def waitforqueues(queues: list, timeout: float = None) -> filter:
"""Waits for one or more *Queue* to be ready or until *timeout* expires.
*queues* is a list containing one or more *Queue.Queue* objects.
If *timeout* is not None the function will block
for the specified amount of seconds.
The function returns a list containing the ready *Queues*.
"""
lock = threading.Condition(threading.Lock())
prepare_queues(queues, lock)
try:
wait_queues(queues, lock, timeout)
finally:
reset_queues(queues)
return filter(lambda q: not q.empty(), queues)
def prepare_queues(queues: list, lock: threading.Condition):
"""Replaces queue._put() method in order to notify the waiting Condition."""
for queue in queues:
queue._pebble_lock = lock
with queue.mutex:
queue._pebble_old_method = queue._put
queue._put = MethodType(new_method, queue)
def wait_queues(queues: list,
lock: threading.Condition,
timeout: Optional[float]):
with lock:
if not any(map(lambda q: not q.empty(), queues)):
lock.wait(timeout)
def reset_queues(queues: list):
"""Resets original queue._put() method."""
for queue in queues:
with queue.mutex:
queue._put = queue._pebble_old_method
delattr(queue, '_pebble_old_method')
delattr(queue, '_pebble_lock')
def waitforthreads(threads: list, timeout: float = None) -> filter:
"""Waits for one or more *Thread* to exit or until *timeout* expires.
.. note::
Expired *Threads* are not joined by *waitforthreads*.
*threads* is a list containing one or more *threading.Thread* objects.
If *timeout* is not None the function will block
for the specified amount of seconds.
The function returns a list containing the ready *Threads*.
"""
old_get_ident = None
event = threading.Event()
def new_get_ident(*args) -> int:
retval = old_get_ident(*args)
event.set()
return retval
old_get_ident = prepare_threads(new_get_ident)
try:
wait_threads(threads, event, timeout)
finally:
reset_threads(old_get_ident)
return filter(lambda t: not t.is_alive(), threads)
def prepare_threads(new_get_ident: Callable) -> Callable:
"""Replaces threading.get_ident() function in order to notify
the waiting Condition."""
with threading._active_limbo_lock:
old_get_ident = threading.get_ident
threading.get_ident = new_get_ident
return old_get_ident
def wait_threads(threads: list,
event: threading.Event,
timeout: Optional[float]):
timestamp = time()
while not any(map(lambda t: not t.is_alive(), threads)):
if timeout is None:
event.wait()
elif timeout - (time() - timestamp) > 0:
event.wait(timeout - (time() - timestamp))
else:
return
def reset_threads(old_function: Callable):
"""Resets original threading.get_ident() function."""
with threading._active_limbo_lock:
threading.get_ident = old_function
def new_method(self, *args):
self._pebble_old_method(*args)
with self._pebble_lock:
self._pebble_lock.notify_all()
|
