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;








0















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.










share|improve this question






























    0















    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.










    share|improve this question


























      0












      0








      0








      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.










      share|improve this question
















      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






      share|improve this question















      share|improve this question













      share|improve this question




      share|improve this question








      edited Mar 8 at 6:14







      Galaxy

















      asked Mar 8 at 4:34









      GalaxyGalaxy

      6251620




      6251620






















          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
          );



          );













          draft saved

          draft discarded


















          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















          draft saved

          draft discarded
















































          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.




          draft saved


          draft discarded














          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





















































          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







          Popular posts from this blog

          1928 у кіно

          Захаров Федір Захарович

          Ель Греко