strawmelonjuice.com
Forking what is left of ZeroNet and hopefully adding an AT Proto Frontend/Proxy
1import gevent.pool
2
3
4class Pooled(object):
5 def __init__(self, size=100):
6 self.pool = gevent.pool.Pool(size)
7 self.pooler_running = False
8 self.queue = []
9 self.func = None
10
11 def waiter(self, evt, args, kwargs):
12 res = self.func(*args, **kwargs)
13 if type(res) == gevent.event.AsyncResult:
14 evt.set(res.get())
15 else:
16 evt.set(res)
17
18 def pooler(self):
19 while self.queue:
20 evt, args, kwargs = self.queue.pop(0)
21 self.pool.spawn(self.waiter, evt, args, kwargs)
22 self.pooler_running = False
23
24 def __call__(self, func):
25 def wrapper(*args, **kwargs):
26 evt = gevent.event.AsyncResult()
27 self.queue.append((evt, args, kwargs))
28 if not self.pooler_running:
29 self.pooler_running = True
30 gevent.spawn(self.pooler)
31 return evt
32 wrapper.__name__ = func.__name__
33 self.func = func
34
35 return wrapper
36
37if __name__ == "__main__":
38 import gevent
39 import gevent.pool
40 import gevent.queue
41 import gevent.event
42 import gevent.monkey
43 import time
44
45 gevent.monkey.patch_all()
46
47 def addTask(inner_path):
48 evt = gevent.event.AsyncResult()
49 gevent.spawn_later(1, lambda: evt.set(True))
50 return evt
51
52 def needFile(inner_path):
53 return addTask(inner_path)
54
55 @Pooled(10)
56 def pooledNeedFile(inner_path):
57 return needFile(inner_path)
58
59 threads = []
60 for i in range(100):
61 threads.append(pooledNeedFile(i))
62
63 s = time.time()
64 gevent.joinall(threads) # Should take 10 second
65 print(time.time() - s)