Skip to content

Commit 8d58dd0

Browse files
mbuferlimbuferli
committed
fix broken loop.
read more here python/asyncio#312 and here python/asyncio#332
1 parent 57e922d commit 8d58dd0

2 files changed

Lines changed: 39 additions & 34 deletions

File tree

‎data/one_producer_many_consumers.json‎

Lines changed: 28 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -1,72 +1,77 @@
11
[
22
{
3-
"expired_after": 1,
43
"id": 1,
4+
"expired_after": 0,
55
"status": "QUEUED"
66
},
77
{
8-
"expired_after": 10,
98
"id": 2,
9+
"expired_after": 0,
1010
"status": "QUEUED"
1111
},
1212
{
13-
"expired_after": 7,
1413
"id": 3,
14+
"expired_after": 0,
1515
"status": "QUEUED"
1616
},
1717
{
18-
"expired_after": 0,
1918
"id": 4,
19+
"expired_after": 0,
2020
"status": "QUEUED"
2121
},
2222
{
23-
"expired_after": 18,
2423
"id": 5,
24+
"expired_after": 0,
2525
"status": "QUEUED"
2626
},
2727
{
28-
"expired_after": 3,
2928
"id": 6,
30-
"status": "WAITING"
29+
"expired_after": 0,
30+
"status": "QUEUED"
3131
},
3232
{
33-
"expired_after": 6,
3433
"id": 7,
35-
"status": "WAITING"
34+
"expired_after": 0,
35+
"status": "QUEUED"
3636
},
3737
{
38-
"expired_after": 7,
3938
"id": 8,
40-
"status": "WAITING"
39+
"expired_after": 0,
40+
"status": "QUEUED"
4141
},
4242
{
43-
"expired_after": 5,
4443
"id": 9,
45-
"status": "WAITING"
44+
"expired_after": 0,
45+
"status": "QUEUED"
4646
},
4747
{
48-
"expired_after": 2,
4948
"id": 10,
50-
"status": "WAITING"
49+
"expired_after": 0,
50+
"status": "QUEUED"
5151
},
5252
{
53-
"expired_after": 12,
5453
"id": 11,
55-
"status": "WAITING"
54+
"expired_after": 0,
55+
"status": "QUEUED"
5656
},
5757
{
58-
"expired_after": 13,
5958
"id": 12,
60-
"status": "WAITING"
59+
"expired_after": 0,
60+
"status": "QUEUED"
6161
},
6262
{
63-
"expired_after": 4,
6463
"id": 13,
65-
"status": "WAITING"
64+
"expired_after": 0,
65+
"status": "QUEUED"
6666
},
6767
{
68-
"expired_after": 3,
6968
"id": 14,
70-
"status": "WAITING"
69+
"expired_after": 0,
70+
"status": "QUEUED"
71+
},
72+
{
73+
"id": 15,
74+
"expired_after": 0,
75+
"status": "QUEUED"
7176
}
7277
]

‎one_producer_many_consumers.py‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
import asyncio
22
from apscheduler.schedulers.asyncio import AsyncIOScheduler
3-
import time
43
import random
54
from driver import *
65
VALID_IDS = [1, 2, 3, 4]
@@ -9,16 +8,16 @@
98
MAX_FETCHED_COUNT = 5
109

1110

12-
async def monitor(queues):
11+
async def monitor(queues, loop):
1312
while True:
1413
print("> [MONITOR] Turn ...")
1514
for id in VALID_IDS:
1615
print("> [MONITOR] Q%s[%s]" % (id, queues[id].qsize()))
1716
print("> [MONITOR] Sleep.")
18-
await asyncio.sleep(MONITOR_SLEEP)
17+
await asyncio.sleep(MONITOR_SLEEP, loop=loop)
1918

2019

21-
async def producer(queues):
20+
async def producer(queues, loop):
2221
while True:
2322
print("> [PRODUCER] Turn ...")
2423
queued_count = 0
@@ -28,16 +27,16 @@ async def producer(queues):
2827
set_match_as_queued(match_id) # mark as queued in DB
2928
queued_count = queued_count + 1
3029
print("> [PRODUCER] Added new %s. Sleep." % queued_count)
31-
await asyncio.sleep(PRODUCER_SLEEP)
30+
await asyncio.sleep(PRODUCER_SLEEP, loop=loop)
3231

3332

34-
async def consumer(q):
33+
async def consumer(q, loop):
3534
while True:
3635
print("\n> [CONSUMER] Turn ...")
3736
match_id = await q.get()
3837
print("> [CONSUMER] Process <%s> ..." % match_id)
3938
proc_count = set_match_processed(match_id) # fake processing, just decrement a counter
40-
time.sleep(random.randint(1, 5)) # add delay
39+
await asyncio.sleep(random.randint(1, 5), loop=loop) # add delay
4140
if proc_count > 0:
4241
q.put_nowait(match_id)
4342
print("> [CONSUMER] Re-QUEUE <%s>" % match_id)
@@ -54,6 +53,8 @@ async def consumer(q):
5453
scheduler = AsyncIOScheduler()
5554
reset_data()
5655

56+
loop = asyncio.get_event_loop()
57+
5758
# init queues
5859
QUEUES = {
5960
1: asyncio.Queue(), # (match_id % 4 + 1) = 1
@@ -62,11 +63,10 @@ async def consumer(q):
6263
4: asyncio.Queue(), # (match_id % 4 + 1) = 4
6364
}
6465

65-
loop = asyncio.get_event_loop()
66-
loop.create_task(producer(QUEUES))
67-
loop.create_task(monitor(QUEUES))
66+
loop.create_task(producer(QUEUES, loop))
67+
loop.create_task(monitor(QUEUES, loop))
6868
for id in VALID_IDS:
69-
loop.create_task(consumer(QUEUES[id])) # pass to each consumer, the relative queues
69+
loop.create_task(consumer(QUEUES[id], loop)) # pass to each consumer, the relative queues
7070

7171
try:
7272
loop.run_forever()

0 commit comments

Comments
 (0)