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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
|
"""Module with utilities related to async operations"""
from threading import (
Lock,
current_thread,
_allocate_lock,
_Condition,
_sleep,
_time,
)
from Queue import (
Queue,
Empty,
)
from collections import deque
import sys
import os
#{ Routines
def cpu_count():
""":return:number of CPUs in the system
:note: inspired by multiprocessing"""
num = 0
try:
if sys.platform == 'win32':
num = int(os.environ['NUMBER_OF_PROCESSORS'])
elif 'bsd' in sys.platform or sys.platform == 'darwin':
num = int(os.popen('sysctl -n hw.ncpu').read())
else:
num = os.sysconf('SC_NPROCESSORS_ONLN')
except (ValueError, KeyError, OSError, AttributeError):
pass
# END exception handling
if num == 0:
raise NotImplementedError('cannot determine number of cpus')
return num
#} END routines
class DummyLock(object):
"""An object providing a do-nothing lock interface for use in sync mode"""
__slots__ = tuple()
def acquire(self):
pass
def release(self):
pass
class SyncQueue(deque):
"""Adapter to allow using a deque like a queue, without locking"""
def get(self, block=True, timeout=None):
try:
return self.popleft()
except IndexError:
raise Empty
# END raise empty
def empty(self):
return len(self) == 0
put = deque.append
class HSCondition(object):
"""An attempt to make conditions less blocking, which gains performance
in return by sleeping less"""
# __slots__ = ("acquire", "release", "_lock", '_waiters')
__slots__ = ("_lock", '_waiters')
delay = 0.00002 # reduces wait times, but increases overhead
def __init__(self, lock=None):
if lock is None:
lock = Lock()
self._lock = lock
#self.acquire = lock.acquire
#self.release = lock.release
self._waiters = list()
def release(self):
return self._lock.release()
def acquire(self, block=None):
if block is None:
self._lock.acquire()
else:
return self._lock.acquire(block)
def wait(self, timeout=None):
waiter = _allocate_lock()
waiter.acquire() # get it the first time, no blocking
self._waiters.append(waiter)
# in the momemnt we release our lock, someone else might actually resume
self.release()
try: # restore state no matter what (e.g., KeyboardInterrupt)
# now we block, as we hold the lock already
if timeout is None:
waiter.acquire()
else:
# Balancing act: We can't afford a pure busy loop, because of the
# GIL, so we have to sleep
# We try to sleep only tiny amounts of time though to be very responsive
endtime = _time() + timeout
delay = self.delay
acquire = waiter.acquire
while True:
gotit = acquire(0)
if gotit:
break
remaining = endtime - _time()
if remaining <= 0:
break
# this makes 4 threads working as good as two, but of course
# it causes more frequent micro-sleeping
#delay = min(delay * 2, remaining, .05)
_sleep(delay)
# END endless loop
if not gotit:
try:
self._waiters.remove(waiter)
except ValueError:
pass
# END didn't ever get it
finally:
# reacquire the lock
self.acquire()
def notify(self, n=1):
if not self._waiters:
return
waiters = self._waiters
if n == 1:
waiters[0].release()
try:
waiters.pop(0)
except IndexError:
pass
else:
for waiter in waiters[:n]:
waiter.release()
try:
waiters.remove(waiter)
except ValueError:
pass
# END handle n = 1 case faster
def notify_all(self):
self.notify(len(self._waiters))
class _AsyncQueue(Queue):
"""A queue using different condition objects to gain multithreading performance"""
def __init__(self, maxsize=0):
Queue.__init__(self, maxsize)
self.not_empty = HSCondition(self.mutex)
self.not_full = HSCondition(self.mutex)
self.all_tasks_done = HSCondition(self.mutex)
class AsyncQueue(Queue):
"""A queue using different condition objects to gain multithreading performance"""
__slots__ = ('mutex', 'not_empty', 'queue')
def __init__(self, maxsize=0):
self.queue = deque()
self.mutex = Lock()
self.not_empty = HSCondition(self.mutex)
def qsize(self):
self.mutex.acquire()
try:
return len(self.queue)
finally:
self.mutex.release()
def empty(self):
self.mutex.acquire()
try:
return not len(self.queue)
finally:
self.mutex.release()
def put(self, item, block=True, timeout=None):
self.mutex.acquire()
self.queue.append(item)
self.mutex.release()
self.not_empty.notify()
def get(self, block=True, timeout=None):
self.not_empty.acquire() # == self.mutex.acquire in that case
q = self.queue
try:
if not block:
if not len(q):
raise Empty
elif timeout is None:
while not len(q):
self.not_empty.wait()
else:
print "with timeout", timeout
import traceback
traceback.print_stack()
endtime = _time() + timeout
while not len(q):
remaining = endtime - _time()
if remaining <= 0.0:
raise Empty
self.not_empty.wait(remaining)
# END handle block
# can happen if someone else woke us up
try:
return q.popleft()
except IndexError:
raise Empty
# END handle unblocking reason
finally:
self.not_empty.release()
#} END utilities
|
