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 to implement a feedback to keep the DC gain at zero for this conceptual passive filter?
Argument list too long when zipping large list of certain files in a folder
What does chmod -u do?
Is there a working SACD iso player for Ubuntu?
Is a bound state a stationary state?
Should I stop contributing to retirement accounts?
How should I respond when I lied about my education and the company finds out through background check?
How do I color the graph in datavisualization?
Not using 's' for he/she/it
Is this toilet slogan correct usage of the English language?
Aragorn's "guise" in the Orthanc Stone
Creature in Shazam mid-credits scene?
What is Cash Advance APR?
Is it possible to have a strip of cold climate in the middle of a planet?
Biological Blimps: Propulsion
How much character growth crosses the line into breaking the character
Is there a name for this algorithm to calculate the concentration of a mixture of two solutions containing the same solute?
Count the occurrence of each unique word in the file
How to explain what's wrong with this application of the chain rule?
Electoral considerations aside, what are potential benefits, for the US, of policy changes proposed by the tweet recognizing Golan annexation?
Approximating irrational number to rational number
How can Trident be so inexpensive? Will it orbit Triton or just do a (slow) flyby?
Did arcade monitors have same pixel aspect ratio as TV sets?
It grows, but water kills it
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)
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