Using asyncio.Queue for parallel producer-consumer flow in Python3What is the difference between concurrency and parallelism?Why does comparing strings using either '==' or 'is' sometimes produce a different result?How to iterate through two lists in parallel?What is the difference between concurrent programming and parallel programming?StringIO in Python3“Large data” work flows using pandasShould I always use a parallel stream when possible?Custom thread pool in Java 8 parallel streamasyncio queue.get() gets stuckTypeError: _request() got an unexpected keyword argument 'cookies' (aiohttp)
How does one intimidate enemies without having the capacity for violence?
What are these boxed doors outside store fronts in New York?
What is the offset in a seaplane's hull?
Is it legal for company to use my work email to pretend I still work there?
Approximately how much travel time was saved by the opening of the Suez Canal in 1869?
Show that if two triangles built on parallel lines, with equal bases have the same perimeter only if they are congruent.
Can I make popcorn with any corn?
Why don't electron-positron collisions release infinite energy?
How can bays and straits be determined in a procedurally generated map?
Why doesn't H₄O²⁺ exist?
Why do I get two different answers for this counting problem?
To string or not to string
How old can references or sources in a thesis be?
What is the word for reserving something for yourself before others do?
The Two and the One
Accidentally leaked the solution to an assignment, what to do now? (I'm the prof)
Service Entrance Breakers Rain Shield
Why are electrically insulating heatsinks so rare? Is it just cost?
Has the BBC provided arguments for saying Brexit being cancelled is unlikely?
Maximum likelihood parameters deviate from posterior distributions
Theorems that impeded progress
How to say job offer in Mandarin/Cantonese?
Why can't I see bouncing of a switch on an oscilloscope?
What's the point of deactivating Num Lock on login screens?
Using asyncio.Queue for parallel producer-consumer flow in Python3
What is the difference between concurrency and parallelism?Why does comparing strings using either '==' or 'is' sometimes produce a different result?How to iterate through two lists in parallel?What is the difference between concurrent programming and parallel programming?StringIO in Python3“Large data” work flows using pandasShould I always use a parallel stream when possible?Custom thread pool in Java 8 parallel streamasyncio queue.get() gets stuckTypeError: _request() got an unexpected keyword argument 'cookies' (aiohttp)
.everyoneloves__top-leaderboard:empty,.everyoneloves__mid-leaderboard:empty,.everyoneloves__bot-mid-leaderboard:empty height:90px;width:728px;box-sizing:border-box;
I want to do some calculation on lines a huge file, and I find reading the lines take less than 5% of the time, calculation each lines takes 95%.
Then I find asyncio.Queue
seems a good abstract for my need.
However, it needs to use concurrent.futures.ThreadPoolExecutor
or ProcessPoolExecutor
to achieve parallel. Without them, asyncio
is single threaded.
I cannot find an example to know how to write it.
Most examples only prints in consumer, without return any values.
Would you please write an example for me ?
For demonstration, the producer put many lines of 10 numbers to queue, the 5 consumers split the line and calculate sum and average for a single line, and the main() adds up the sums and the averages to print two results in the end.
The default example is single threded. Just change await asyncio.sleep(sleep_for)
to for i in range(10000000):pass
.
And then, only "worker-0" shown up. The CPU% is always less than 100%.
import asyncio
import random
import time
async def worker(name, queue):
while True:
# Get a "work item" out of the queue.
sleep_for = await queue.get()
print(f'name has Begin')
# Sleep for the "sleep_for" seconds.
#await asyncio.sleep(sleep_for)
for i in range(10000000):pass
# Notify the queue that the "work item" has been processed.
queue.task_done()
print(f'name has slept for sleep_for:.2f seconds')
async def main():
# Create a queue that we will use to store our "workload".
queue = asyncio.Queue()
# Generate random timings and put them into the queue.
total_sleep_time = 0
for _ in range(20):
sleep_for = random.uniform(0.05, 1.0)
total_sleep_time += sleep_for
queue.put_nowait(sleep_for)
# Create three worker tasks to process the queue concurrently.
tasks = []
for i in range(3):
task = asyncio.create_task(worker(f'worker-i', queue))
tasks.append(task)
# Wait until the queue is fully processed.
started_at = time.monotonic()
await queue.join()
total_slept_for = time.monotonic() - started_at
# Cancel our worker tasks.
for task in tasks:
task.cancel()
# Wait until all worker tasks are cancelled.
await asyncio.gather(*tasks, return_exceptions=True)
print('====')
print(f'3 workers slept in parallel for total_slept_for:.2f seconds')
print(f'total expected sleep time: total_sleep_time:.2f seconds')
asyncio.run(main())
Also, I hope the code is backward compatible with Python 3.5.3, since PyPy seems faster.
python parallel-processing python-asyncio publish-subscribe
add a comment |
I want to do some calculation on lines a huge file, and I find reading the lines take less than 5% of the time, calculation each lines takes 95%.
Then I find asyncio.Queue
seems a good abstract for my need.
However, it needs to use concurrent.futures.ThreadPoolExecutor
or ProcessPoolExecutor
to achieve parallel. Without them, asyncio
is single threaded.
I cannot find an example to know how to write it.
Most examples only prints in consumer, without return any values.
Would you please write an example for me ?
For demonstration, the producer put many lines of 10 numbers to queue, the 5 consumers split the line and calculate sum and average for a single line, and the main() adds up the sums and the averages to print two results in the end.
The default example is single threded. Just change await asyncio.sleep(sleep_for)
to for i in range(10000000):pass
.
And then, only "worker-0" shown up. The CPU% is always less than 100%.
import asyncio
import random
import time
async def worker(name, queue):
while True:
# Get a "work item" out of the queue.
sleep_for = await queue.get()
print(f'name has Begin')
# Sleep for the "sleep_for" seconds.
#await asyncio.sleep(sleep_for)
for i in range(10000000):pass
# Notify the queue that the "work item" has been processed.
queue.task_done()
print(f'name has slept for sleep_for:.2f seconds')
async def main():
# Create a queue that we will use to store our "workload".
queue = asyncio.Queue()
# Generate random timings and put them into the queue.
total_sleep_time = 0
for _ in range(20):
sleep_for = random.uniform(0.05, 1.0)
total_sleep_time += sleep_for
queue.put_nowait(sleep_for)
# Create three worker tasks to process the queue concurrently.
tasks = []
for i in range(3):
task = asyncio.create_task(worker(f'worker-i', queue))
tasks.append(task)
# Wait until the queue is fully processed.
started_at = time.monotonic()
await queue.join()
total_slept_for = time.monotonic() - started_at
# Cancel our worker tasks.
for task in tasks:
task.cancel()
# Wait until all worker tasks are cancelled.
await asyncio.gather(*tasks, return_exceptions=True)
print('====')
print(f'3 workers slept in parallel for total_slept_for:.2f seconds')
print(f'total expected sleep time: total_sleep_time:.2f seconds')
asyncio.run(main())
Also, I hope the code is backward compatible with Python 3.5.3, since PyPy seems faster.
python parallel-processing python-asyncio publish-subscribe
add a comment |
I want to do some calculation on lines a huge file, and I find reading the lines take less than 5% of the time, calculation each lines takes 95%.
Then I find asyncio.Queue
seems a good abstract for my need.
However, it needs to use concurrent.futures.ThreadPoolExecutor
or ProcessPoolExecutor
to achieve parallel. Without them, asyncio
is single threaded.
I cannot find an example to know how to write it.
Most examples only prints in consumer, without return any values.
Would you please write an example for me ?
For demonstration, the producer put many lines of 10 numbers to queue, the 5 consumers split the line and calculate sum and average for a single line, and the main() adds up the sums and the averages to print two results in the end.
The default example is single threded. Just change await asyncio.sleep(sleep_for)
to for i in range(10000000):pass
.
And then, only "worker-0" shown up. The CPU% is always less than 100%.
import asyncio
import random
import time
async def worker(name, queue):
while True:
# Get a "work item" out of the queue.
sleep_for = await queue.get()
print(f'name has Begin')
# Sleep for the "sleep_for" seconds.
#await asyncio.sleep(sleep_for)
for i in range(10000000):pass
# Notify the queue that the "work item" has been processed.
queue.task_done()
print(f'name has slept for sleep_for:.2f seconds')
async def main():
# Create a queue that we will use to store our "workload".
queue = asyncio.Queue()
# Generate random timings and put them into the queue.
total_sleep_time = 0
for _ in range(20):
sleep_for = random.uniform(0.05, 1.0)
total_sleep_time += sleep_for
queue.put_nowait(sleep_for)
# Create three worker tasks to process the queue concurrently.
tasks = []
for i in range(3):
task = asyncio.create_task(worker(f'worker-i', queue))
tasks.append(task)
# Wait until the queue is fully processed.
started_at = time.monotonic()
await queue.join()
total_slept_for = time.monotonic() - started_at
# Cancel our worker tasks.
for task in tasks:
task.cancel()
# Wait until all worker tasks are cancelled.
await asyncio.gather(*tasks, return_exceptions=True)
print('====')
print(f'3 workers slept in parallel for total_slept_for:.2f seconds')
print(f'total expected sleep time: total_sleep_time:.2f seconds')
asyncio.run(main())
Also, I hope the code is backward compatible with Python 3.5.3, since PyPy seems faster.
python parallel-processing python-asyncio publish-subscribe
I want to do some calculation on lines a huge file, and I find reading the lines take less than 5% of the time, calculation each lines takes 95%.
Then I find asyncio.Queue
seems a good abstract for my need.
However, it needs to use concurrent.futures.ThreadPoolExecutor
or ProcessPoolExecutor
to achieve parallel. Without them, asyncio
is single threaded.
I cannot find an example to know how to write it.
Most examples only prints in consumer, without return any values.
Would you please write an example for me ?
For demonstration, the producer put many lines of 10 numbers to queue, the 5 consumers split the line and calculate sum and average for a single line, and the main() adds up the sums and the averages to print two results in the end.
The default example is single threded. Just change await asyncio.sleep(sleep_for)
to for i in range(10000000):pass
.
And then, only "worker-0" shown up. The CPU% is always less than 100%.
import asyncio
import random
import time
async def worker(name, queue):
while True:
# Get a "work item" out of the queue.
sleep_for = await queue.get()
print(f'name has Begin')
# Sleep for the "sleep_for" seconds.
#await asyncio.sleep(sleep_for)
for i in range(10000000):pass
# Notify the queue that the "work item" has been processed.
queue.task_done()
print(f'name has slept for sleep_for:.2f seconds')
async def main():
# Create a queue that we will use to store our "workload".
queue = asyncio.Queue()
# Generate random timings and put them into the queue.
total_sleep_time = 0
for _ in range(20):
sleep_for = random.uniform(0.05, 1.0)
total_sleep_time += sleep_for
queue.put_nowait(sleep_for)
# Create three worker tasks to process the queue concurrently.
tasks = []
for i in range(3):
task = asyncio.create_task(worker(f'worker-i', queue))
tasks.append(task)
# Wait until the queue is fully processed.
started_at = time.monotonic()
await queue.join()
total_slept_for = time.monotonic() - started_at
# Cancel our worker tasks.
for task in tasks:
task.cancel()
# Wait until all worker tasks are cancelled.
await asyncio.gather(*tasks, return_exceptions=True)
print('====')
print(f'3 workers slept in parallel for total_slept_for:.2f seconds')
print(f'total expected sleep time: total_sleep_time:.2f seconds')
asyncio.run(main())
Also, I hope the code is backward compatible with Python 3.5.3, since PyPy seems faster.
python parallel-processing python-asyncio publish-subscribe
python parallel-processing python-asyncio publish-subscribe
edited Mar 8 at 6:14
Galaxy
asked Mar 8 at 4:34
GalaxyGalaxy
6251620
6251620
add a comment |
add a comment |
0
active
oldest
votes
Your Answer
StackExchange.ifUsing("editor", function ()
StackExchange.using("externalEditor", function ()
StackExchange.using("snippets", function ()
StackExchange.snippets.init();
);
);
, "code-snippets");
StackExchange.ready(function()
var channelOptions =
tags: "".split(" "),
id: "1"
;
initTagRenderer("".split(" "), "".split(" "), channelOptions);
StackExchange.using("externalEditor", function()
// Have to fire editor after snippets, if snippets enabled
if (StackExchange.settings.snippets.snippetsEnabled)
StackExchange.using("snippets", function()
createEditor();
);
else
createEditor();
);
function createEditor()
StackExchange.prepareEditor(
heartbeatType: 'answer',
autoActivateHeartbeat: false,
convertImagesToLinks: true,
noModals: true,
showLowRepImageUploadWarning: true,
reputationToPostImages: 10,
bindNavPrevention: true,
postfix: "",
imageUploader:
brandingHtml: "Powered by u003ca class="icon-imgur-white" href="https://imgur.com/"u003eu003c/au003e",
contentPolicyHtml: "User contributions licensed under u003ca href="https://creativecommons.org/licenses/by-sa/3.0/"u003ecc by-sa 3.0 with attribution requiredu003c/au003e u003ca href="https://stackoverflow.com/legal/content-policy"u003e(content policy)u003c/au003e",
allowUrls: true
,
onDemand: true,
discardSelector: ".discard-answer"
,immediatelyShowMarkdownHelp:true
);
);
Sign up or log in
StackExchange.ready(function ()
StackExchange.helpers.onClickDraftSave('#login-link');
);
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
StackExchange.ready(
function ()
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f55056775%2fusing-asyncio-queue-for-parallel-producer-consumer-flow-in-python3%23new-answer', 'question_page');
);
Post as a guest
Required, but never shown
0
active
oldest
votes
0
active
oldest
votes
active
oldest
votes
active
oldest
votes
Thanks for contributing an answer to Stack Overflow!
- Please be sure to answer the question. Provide details and share your research!
But avoid …
- Asking for help, clarification, or responding to other answers.
- Making statements based on opinion; back them up with references or personal experience.
To learn more, see our tips on writing great answers.
Sign up or log in
StackExchange.ready(function ()
StackExchange.helpers.onClickDraftSave('#login-link');
);
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
StackExchange.ready(
function ()
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f55056775%2fusing-asyncio-queue-for-parallel-producer-consumer-flow-in-python3%23new-answer', 'question_page');
);
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function ()
StackExchange.helpers.onClickDraftSave('#login-link');
);
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function ()
StackExchange.helpers.onClickDraftSave('#login-link');
);
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function ()
StackExchange.helpers.onClickDraftSave('#login-link');
);
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown