/usr/lib/python2.7/dist-packages/pygame_sdl2/threads/__init__.py is in python-pygame-sdl2 6.99.8-3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 | """
* Experimental *
Like the map function, but can use a pool of threads.
Really easy to use threads. eg. tmap(f, alist)
If you know how to use the map function, you can use threads.
"""
__author__ = "Rene Dudfield"
__version__ = "0.3.0"
__license__ = 'Python license'
import traceback, sys
from pygame.compat import geterror
if sys.version_info[0] == 3:
from queue import Queue
from queue import Empty
elif (sys.version_info[0] == 2 and sys.version_info[1] < 5):
from Py25Queue import Queue
from Py25Queue import Empty
else:
# use up to date version
from Queue import Queue
from Queue import Empty
import threading
Thread = threading.Thread
STOP = object()
FINISH = object()
# DONE_ONE = object()
# DONE_TWO = object()
# a default worker queue.
_wq = None
# if we are using threads or not. This is the number of workers.
_use_workers = 0
# Set this to the maximum for the amount of Cores/CPUs
# Note, that the tests early out.
# So it should only test the best number of workers +2
MAX_WORKERS_TO_TEST = 64
def init(number_of_workers = 0):
""" Does a little test to see if threading is worth it.
Sets up a global worker queue if it's worth it.
Calling init() is not required, but is generally better to do.
"""
global _wq, _use_workers
if number_of_workers:
_use_workers = number_of_workers
else:
_use_workers = benchmark_workers()
# if it is best to use zero workers, then use that.
_wq = WorkerQueue(_use_workers)
def quit():
""" cleans up everything.
"""
global _wq, _use_workers
_wq.stop()
_wq = None
_use_workers = False
def benchmark_workers(a_bench_func = None, the_data = None):
""" does a little test to see if workers are at all faster.
Returns the number of workers which works best.
Takes a little bit of time to run, so you should only really call
it once.
You can pass in benchmark data, and functions if you want.
a_bench_func - f(data)
the_data - data to work on.
"""
global _use_workers
#TODO: try and make this scale better with slower/faster cpus.
# first find some variables so that using 0 workers takes about 1.0 seconds.
# then go from there.
# note, this will only work with pygame 1.8rc3+
# replace the doit() and the_data with something that releases the GIL
import pygame
import pygame.transform
import time
if not a_bench_func:
def doit(x):
return pygame.transform.scale(x, (544, 576))
else:
doit = a_bench_func
if not the_data:
thedata = []
for x in range(10):
thedata.append(pygame.Surface((155,155), 0, 32))
else:
thedata = the_data
best = time.time() + 100000000
best_number = 0
last_best = -1
for num_workers in range(0, MAX_WORKERS_TO_TEST):
wq = WorkerQueue(num_workers)
t1 = time.time()
for xx in range(20):
print ("active count:%s" % threading.activeCount())
results = tmap(doit, thedata, worker_queue = wq)
t2 = time.time()
wq.stop()
total_time = t2 - t1
print ("total time num_workers:%s: time:%s:" % (num_workers, total_time))
if total_time < best:
last_best = best_number
best_number =num_workers
best = total_time
if num_workers - best_number > 1:
# We tried to add more, but it didn't like it.
# so we stop with testing at this number.
break
return best_number
class WorkerQueue(object):
def __init__(self, num_workers = 20):
self.queue = Queue()
self.pool = []
self._setup_workers(num_workers)
def _setup_workers(self, num_workers):
""" Sets up the worker threads
NOTE: undefined behaviour if you call this again.
"""
self.pool = []
for _ in range(num_workers):
self.pool.append(Thread(target=self.threadloop))
for a_thread in self.pool:
a_thread.setDaemon(True)
a_thread.start()
def do(self, f, *args, **kwArgs):
""" puts a function on a queue for running later.
"""
self.queue.put((f, args, kwArgs))
def stop(self):
""" Stops the WorkerQueue, waits for all of the threads to finish up.
"""
self.queue.put(STOP)
for thread in self.pool:
thread.join()
def threadloop(self): #, finish = False):
""" Loops until all of the tasks are finished.
"""
while True:
args = self.queue.get()
if args is STOP:
self.queue.put(STOP)
self.queue.task_done()
break
else:
try:
args[0](*args[1], **args[2])
finally:
# clean up the queue, raise the exception.
self.queue.task_done()
#raise
def wait(self):
""" waits until all tasks are complete.
"""
self.queue.join()
class FuncResult:
""" Used for wrapping up a function call so that the results are stored
inside the instances result attribute.
"""
def __init__(self, f, callback = None, errback = None):
""" f - is the function we that we call
callback(result) - this is called when the function(f) returns
errback(exception) - this is called when the function(f) raises
an exception.
"""
self.f = f
self.exception = None
self.callback = callback
self.errback = errback
def __call__(self, *args, **kwargs):
#we try to call the function here. If it fails we store the exception.
try:
self.result = self.f(*args, **kwargs)
if self.callback:
self.callback(self.result)
except Exception:
self.exception = geterror()
if self.errback:
self.errback(self.exception)
def tmap(f, seq_args, num_workers = 20, worker_queue = None, wait = True, stop_on_error = True):
""" like map, but uses a thread pool to execute.
num_workers - the number of worker threads that will be used. If pool
is passed in, then the num_workers arg is ignored.
worker_queue - you can optionally pass in an existing WorkerQueue.
wait - True means that the results are returned when everything is finished.
False means that we return the [worker_queue, results] right away instead.
results, is returned as a list of FuncResult instances.
stop_on_error -
"""
if worker_queue:
wq = worker_queue
else:
# see if we have a global queue to work with.
if _wq:
wq = _wq
else:
if num_workers == 0:
return map(f, seq_args)
wq = WorkerQueue(num_workers)
# we short cut it here if the number of workers is 0.
# normal map should be faster in this case.
if len(wq.pool) == 0:
return map(f, seq_args)
#print ("queue size:%s" % wq.queue.qsize())
#TODO: divide the data (seq_args) into even chunks and
# then pass each thread a map(f, equal_part(seq_args))
# That way there should be less locking, and overhead.
results = []
for sa in seq_args:
results.append(FuncResult(f))
wq.do(results[-1], sa)
#wq.stop()
if wait:
#print ("wait")
wq.wait()
#print ("after wait")
#print ("queue size:%s" % wq.queue.qsize())
if wq.queue.qsize():
raise Exception("buggy threadmap")
# if we created a worker queue, we need to stop it.
if not worker_queue and not _wq:
#print ("stoping")
wq.stop()
if wq.queue.qsize():
um = wq.queue.get()
if not um is STOP:
raise Exception("buggy threadmap")
# see if there were any errors. If so raise the first one. This matches map behaviour.
# TODO: the traceback doesn't show up nicely.
# NOTE: TODO: we might want to return the results anyway? This should be an option.
if stop_on_error:
error_ones = list(filter(lambda x:x.exception, results))
if error_ones:
raise error_ones[0].exception
return map(lambda x:x.result, results)
else:
return [wq, results]
|