Multiple instance of a @tasks.loop

[Marc-AntoineGelinas]

I'm developping a bot that scrape posts on reddit. Now I created a command so that I can fetch x amount of posts every y amount of time with @tasks.loop(time=y). Built the command in a cog, loaded it into the bot and it works as expected.

Now my problem is that as far as I understand only one instance of this command can run at once. i.e if I run the command in a server it will work fine, but I will not be able to run the command into another server until I .cancel() the running instance.

I was wondering the correct way to run multiple instances of tasks. Since as of right now I get the 'Task is already launched and is not completed' error when I try to start a second instance.

I ommited some parameters in the code for simplicity's sake, but here is the code in question

def launch_client():
bot = Bot()
reddit = client.Client()

bot.add_cog(ScheduleCog(bot, reddit))

class ScheduleCog(commands.Cog):
    time_unit = {'seconds': 1, 'minutes': 60, 'hours': 3600, 'days': 86400, 'weeks': 604800}
    def __init__(self, bot, reddit):
        self.bot = bot
        self.reddit = reddit

    def cog_unload(self):
        self.schedule_loop.cancel()

    @tasks.loop(seconds=1)
    async def schedule_loop(self, ctx, sub, sort, limit, time):
        posts = await self.reddit.get_posts(sub, sort, limit, time)

        if posts is not None:
            for post in posts:
                await ctx.channel.send(post.url)
        else:
            await ctx.channel.send('Error in your request. Please validate fields.')

    @schedule_loop.before_loop
    async def before_schedule_loop(self):
        await self.bot.wait_until_ready()

    @commands.command()
    async def subschedule(self, ctx, interval: int, interval_unit: str, sub: str, sort: str, limit: int,
                          time: str = 'hot'):
        unit: int = self.time_unit[interval_unit]
        self.schedule_loop.change_interval(seconds=interval * unit)
        self.schedule_loop.start(ctx, sub, sort, limit, time)

    @commands.command()
    async def subschedulestop(self, ctx):
        self.schedule_loop.cancel()

So I've written this accord to the official discord.ext.tasks documentation.

Now everything works as intended for the loop. I launch the subschedule() command and it send the posts every 'x' interval of 'y' unit of time.

Now as mentionned in the original question, is that it works in a single instance and I don't know how to have multiple instances. I want to be able to run multiple subschedule() instances in different guilds or even multiple in the same one, but as it works right now when I call the command again I get the Runtime error 'Task is already launched and is not completed'.

I simply want to know the correct method to be able to run multiple tasks in parallel.

I'm aware of this issue that was closed : #2294.

I tried to read the commit but I didn't understand how to implement it in my code.

Thanks.

[Rapptz]

The issue you linked is a different matter -- that's storing multiple instances of objects and making them not point to the same underlying task object. This was a bug and was fixed. The design of the task extension however does not allow multiple tasks per object, which is what you're trying to do here.

If you want to do some sort of scheduler you'll have to refactor the system so it works with a single task. That's not inherently difficult, though it depends on how you're storing this information in the first place. Your example code does not have any underlying storage, so as a result it doesn't really work once you restart the bot anyway, something to keep in mind for later.

That being said, the design for the task extension is not meant for schedulers since those are better off written by hand which have more control over the interval. My bot for example, implements a scheduler using a regular task + loop.

Essentially in your case you want to do something like the following:

class Cog(commands.Cog):
    def __init__(self, bot):
        # This is the bot object, necessary for the loop termination condition
        self.bot = bot
        # This actually starts the task in the bot's event loop
        # We maintain a reference to it so we can restart it later
        self.task = bot.loop.create_task(self.scheduler())
        # This is where we actually store our schedules.
        # Note that the way you obtain this data is up to you
        # For the sake of example, I'm going to assume it's a list of
        # datetime.datetime objects. In a real program, it's probably 
        # a list of objects with a datetime.datetime attribute.
        # Either way, it should be easy to adapt.
        self.schedules = []
        # This is an event that we use in case the scheduler has no work to do
        # We want to wait until we do have work to do so we use this.
        self.has_schedule = asyncio.Event()
        # This tells us what date we're currently waiting for
        self.current_schedule = None

    async def scheduler(self):
        # Let's get to the meat of our scheduler first.
        # The first thing we do is wait until the bot is ready to function
        await self.bot.wait_until_ready()

        # This loops until the bot is closed
        while not self.bot.is_closed():
            # Get the current schedule that we're waiting for (explained below)
            self.current_schedule = await self.get_oldest_schedule()
            # Sleep until we're needed for that schedule to get triggered
            # Note, in a real program current_schedule's datetime attribute might be
            # something else. In this example we're returning just the datetime object.
            await discord.utils.sleep_until(self.current_schedule)
            
            # Once the schedule is over, we can trigger what we want to do here...
            # some_function(...)

            # Then we remove it from our scheduler to clean up
            self.remove_schedule(self.current_schedule)

    async def get_oldest_schedule(self):
        # If we have nothing scheduled...
        if len(self.schedules) == 0:
            # Wait until we do...
            await self.has_schedule.wait()
        
        # When we have something scheduled, get the oldest
        # entry (i.e. the one that will be "triggered" soonest)
        # In real code it's probably something like
        # return min(self.schedules, key=lambda o: o.expires_at)
        return min(self.schedules)

    def schedule_item(self, item):
        # This is the function where we actually schedule an item.
        if len(self.schedules) == 0:
            # If we have no items scheduled, then we should add the schedule
            # and then wait our worker that's waiting for something to do
            self.schedules.append(item)
            self.has_schedule.set()
            return

        # If we have an item scheduled before, we could just add it as-is..
        self.schedules.append(item)

        # but there might be a chance that the current item is older than
        # the one we're currently waiting for
        if self.current_schedule is not None and item < self.current_schedule:
            # If this is the case then we want to restart the internal task
            # with the new data
            self.task.cancel()
            self.task = self.bot.loop.create_task(self.scheduler())
    
    def remove_schedule(self, item):
        # This function removes a scheduled item.
        # The only invariant to keep track of here is the event keeping track
        # of our schedule's length
        # Note this implementation will differ depending on the backing storage.
        try:
            self.schedules.remove(item)
        except ValueError:
            pass
        else:
            if len(self.schedules) == 0:
                self.has_schedule.clear()

The machinery is a bit involved, but hopefully the comments help understand the general flow of an approach of doing it.

[Marc-AntoineGelinas]

I'm new to this and just implementing things as I learn along the way. So thanks for the suggestion about storing tasks. I had noticed it was an issue, but had not put more thought into it.

As for the answer to the original question, thanks for the very thorough code answer. I'll look into it with more attention, but I think I get the concept and how I will have to refactor my code. Thanks a lot

[mystique09]

Nice, but will this block the other cogs from running?