0

I have an async method, as shown below.

I pass in lists of 1000 numbers, where the method will pass in each number to a helper function which will return something from a website.

I have a global variable called count, which i surround with locks to make sure it doesnt get changed by anything else

I use add_done_callback with the task to make this method async.

The goal is to keep sending a number in the list of 1000 numbers to the server, and only when the server returns data (can take anywhere from 0.1 to 2 seconds), to pause, write the data to a sql database, and then continue

The code works as expected without locks, or without making the callback function, (which is named 'function' below) asyncrounous. But adding locks gives me an error: RuntimeWarning: coroutine 'function' was never awaited self._context.run(self._callback, *self._args) RuntimeWarning: Enable tracemalloc to get the object allocation traceback

I am super new to async in python so any help/advice is greatly appriciated

My code is shown below. It is just a simple draft:

import time
import random
import asyncio
# from helper import get_message_from_server

async def get(number):
    # get_message_from_server(number), which takes somewhere between 0.1 to 2 seconds
    await asyncio.sleep(random.uniform(0.1, 2)) 
    s = 'Done with number ' + number
    return s

async def function(future, lock):
    global count
    print(future.result())
    # write future.result() to db
    acquired = await lock.acquire()
    count -= 1 if (count > 1) else 0
    lock.release()

async def main(numbers, lock):
    global count 
    
    count = 0
    
    for i, number in enumerate(numbers):
        print('number:', number, 'count:', count)
        acquired = await lock.acquire()
        count += 1
        lock.release()
        task = asyncio.create_task(get(number))
        
        task.add_done_callback(
            lambda x: function(x, lock)
        )
        if (count == 50):
            print('Reached 50')
            await task

            acquired = await lock.acquire()
            count = 0
            lock.release()

        if (i == len(numbers) - 1):
            await task

def make_numbers():
    count = []
    for i in range(1001):
        count.append(str(i))
    return count

if __name__ == '__main__':
    numbers = make_numbers()
    loop = asyncio.get_event_loop()
    lock = asyncio.Lock()
    try:
        loop.run_until_complete(main(numbers, lock))
    except Exception as e:
        pass
    finally:
        loop.run_until_complete(loop.shutdown_asyncgens())
        loop.stop()
Sharhad
  • 63
  • 8
  • Try `functools.partial(function, lock=lock)` instead of `lambda`. However, I'd advice against `add_done_callback`, even docs mention it with comment "This method should only be used in low-level callback-based code". You can simply define new `async def` function and use it to `create_task`, like in [this answer](https://stackoverflow.com/questions/53701841/what-is-the-use-case-for-future-add-done-callback). – STerliakov May 23 '22 at 19:25

1 Answers1

0

The above comment helped a lot

This is what the final working code looks like:

import time
import random
import asyncio
from functools import partial
# from helper import get_message_from_server

async def get(number):
    # get_message_from_server(number), which takes somewhere between 0.1 to 2 seconds
    await asyncio.sleep(random.uniform(0.1, 2)) 
    s = 'Done with number ' + number
    return s

def function(result, lock):
    
    print(result.result())

    async def count_decrement(lock):
        global count
        print('in count decrement')
        acquired = await lock.acquire()
        count -= 1 if (count > 1) else 0
        lock.release()

    asyncio.create_task(count_decrement(lock))

async def main(numbers, lock):
    global count 
    count = 0
    for i, number in enumerate(numbers):
        print('number:', number, 'count:', count)
        acquired = await lock.acquire()
        count += 1
        lock.release()
        task = asyncio.create_task(get(number))
        
        task.add_done_callback(partial(function, lock = lock))

        if (count == 50):
            print('Reached 50')
            await task

            acquired = await lock.acquire()
            count = 0
            lock.release()

        if (i == len(numbers) - 1):
            await task

def make_numbers():
    count = []
    for i in range(1001):
        count.append(str(i))
    return count

if __name__ == '__main__':
    numbers = make_numbers()
    loop = asyncio.get_event_loop()
    lock = asyncio.Lock()
    try:
        loop.run_until_complete(main(numbers, lock))
    except Exception as e:
        pass
    finally:
        loop.run_until_complete(loop.shutdown_asyncgens())
        loop.stop()
Sharhad
  • 63
  • 8