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
137
138
|
# 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
_waitforthreads_lock = threading.Lock()
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_function = None
lock = threading.Condition(threading.Lock())
def new_function(*args):
old_function(*args)
with lock:
lock.notify_all()
old_function = prepare_threads(new_function)
try:
wait_threads(threads, lock, timeout)
finally:
reset_threads(old_function)
return filter(lambda t: not t.is_alive(), threads)
def prepare_threads(new_function: Callable) -> Callable:
"""Replaces threading._get_ident() function in order to notify
the waiting Condition."""
with _waitforthreads_lock:
old_function = threading.get_ident
threading.get_ident = new_function
return old_function
def wait_threads(threads: list,
lock: threading.Condition,
timeout: Optional[float]):
timestamp = time()
with lock:
while not any(map(lambda t: not t.is_alive(), threads)):
if timeout is None:
lock.wait()
elif timeout - (time() - timestamp) > 0:
lock.wait(timeout - (time() - timestamp))
else:
return
def reset_threads(old_function: Callable):
"""Resets original threading.get_ident() function."""
with _waitforthreads_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()
|
