Sharing my experience of using APScheduler==4.0 with Sanic

[adk23333]

import os
from asyncio import sleep
from datetime import datetime
from inspect import isawaitable
from typing import Callable, Any

from apscheduler import AsyncScheduler, Job
from apscheduler.abc import JobExecutor
from apscheduler.datastores.sqlalchemy import SQLAlchemyDataStore
from apscheduler.eventbrokers.asyncpg import AsyncpgEventBroker
from apscheduler.triggers.interval import IntervalTrigger
from sanic import Sanic, text
from sanic.logging.loggers import logger
from sqlalchemy.ext.asyncio import create_async_engine

app = Sanic('TestApp')


class TaskJobExecutor(JobExecutor):

    async def run_job(self, func: Callable[..., Any], job: Job) -> Any:
        retval = func(*job.args, **job.kwargs)
        if isawaitable(retval):
            retval = await app.add_task(retval, name=job.schedule_id)
        return retval


async def start_scheduler():
    async with app.ctx.scheduler:
        await app.ctx.scheduler.run_until_stopped()


async def tick():
    print("Hello, the time is", datetime.now())


@app.before_server_start
async def start(_):
    engine = create_async_engine("postgresql...")
    data_store = SQLAlchemyDataStore(engine)
    event_broker = AsyncpgEventBroker.from_async_sqla_engine(engine)
    app.ctx.scheduler = AsyncScheduler(
        job_executors={'task': TaskJobExecutor()},
        data_store=data_store,
        event_broker=event_broker,
        logger=logger,
    )
    _ = app.add_task(start_scheduler())


@app.after_server_start
async def after(_):
    await app.ctx.scheduler.add_schedule(tick, IntervalTrigger(seconds=1), id="tick")


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=3000, workers=4)

And log

Main  2024-08-12 00:37:07 +0800 INFO: Sanic v24.6.0
Main  2024-08-12 00:37:07 +0800 INFO: Goin' Fast @ http://0.0.0.0:3000
Main  2024-08-12 00:37:07 +0800 INFO: app: TestApp
Main  2024-08-12 00:37:07 +0800 INFO: mode: production, w/ 4 workers
Main  2024-08-12 00:37:07 +0800 INFO: server: sanic, HTTP/1.1
Main  2024-08-12 00:37:07 +0800 INFO: python: 3.11.4
Main  2024-08-12 00:37:07 +0800 INFO: platform: Windows-10-10.0.22631-SP0
Main  2024-08-12 00:37:07 +0800 INFO: packages: sanic-routing==23.12.0, sanic-ext==23.12.0
Srv 0 2024-08-12 00:37:09 +0800 INFO: Sanic Extensions:
Srv 0 2024-08-12 00:37:09 +0800 INFO:   > injection [0 dependencies; 0 constants]
Srv 0 2024-08-12 00:37:09 +0800 INFO:   > openapi [http://0.0.0.0:3000/docs]
Srv 0 2024-08-12 00:37:09 +0800 INFO:   > http 
Srv 2 2024-08-12 00:37:09 +0800 INFO: Sanic Extensions:
Srv 2 2024-08-12 00:37:09 +0800 INFO:   > injection [0 dependencies; 0 constants]
Srv 2 2024-08-12 00:37:09 +0800 INFO:   > openapi [http://0.0.0.0:3000/docs]
Srv 2 2024-08-12 00:37:09 +0800 INFO:   > http 
Srv 3 2024-08-12 00:37:09 +0800 INFO: Sanic Extensions:
Srv 3 2024-08-12 00:37:09 +0800 INFO:   > injection [0 dependencies; 0 constants]
Srv 3 2024-08-12 00:37:09 +0800 INFO:   > openapi [http://0.0.0.0:3000/docs]
Srv 3 2024-08-12 00:37:09 +0800 INFO:   > http 
Srv 1 2024-08-12 00:37:09 +0800 INFO: Sanic Extensions:
Srv 1 2024-08-12 00:37:09 +0800 INFO:   > injection [0 dependencies; 0 constants]
Srv 1 2024-08-12 00:37:09 +0800 INFO:   > openapi [http://0.0.0.0:3000/docs]
Srv 1 2024-08-12 00:37:09 +0800 INFO:   > http 
Srv 0 2024-08-12 00:37:09 +0800 INFO: Added new schedule (task='__mp_main__:tick', trigger=IntervalTrigger(seconds=1, start_time='2024-08-12 00:37:09.451754+08:00')); next run time at 2024-08-12 00:37:09.451754+08:00
Srv 0 2024-08-12 00:37:09 +0800 INFO: Starting worker [26460]
Srv 2 2024-08-12 00:37:09 +0800 INFO: Added new schedule (task='__mp_main__:tick', trigger=IntervalTrigger(seconds=1, start_time='2024-08-12 00:37:09.452756+08:00')); next run time at 2024-08-12 00:37:09.452756+08:00
Srv 2 2024-08-12 00:37:09 +0800 INFO: Starting worker [36492]
Srv 0 2024-08-12 00:37:09 +0800 INFO: Scheduler started
Srv 2 2024-08-12 00:37:09 +0800 INFO: Scheduler started
Srv 0 2024-08-12 00:37:09 +0800 INFO: Cleaned up expired job results and finished schedules
Srv 1 2024-08-12 00:37:09 +0800 INFO: Added new schedule (task='__mp_main__:tick', trigger=IntervalTrigger(seconds=1, start_time='2024-08-12 00:37:09.511953+08:00')); next run time at 2024-08-12 00:37:09.511953+08:00
Srv 1 2024-08-12 00:37:09 +0800 INFO: Starting worker [7332]
Srv 2 2024-08-12 00:37:09 +0800 INFO: Cleaned up expired job results and finished schedules
Srv 3 2024-08-12 00:37:09 +0800 INFO: Scheduler started
Srv 3 2024-08-12 00:37:09 +0800 INFO: Cleaned up expired job results and finished schedules
Srv 3 2024-08-12 00:37:09 +0800 INFO: Added new schedule (task='__mp_main__:tick', trigger=IntervalTrigger(seconds=1, start_time='2024-08-12 00:37:09.507953+08:00')); next run time at 2024-08-12 00:37:09.507953+08:00
Srv 3 2024-08-12 00:37:09 +0800 INFO: Starting worker [29652]
Hello, the time is 2024-08-12 00:37:09.585752
Srv 3 2024-08-12 00:37:09 +0800 INFO: Job 959e4f24-44d0-48e2-9dc9-aa2f94c33e85 completed successfully
Srv 1 2024-08-12 00:37:09 +0800 INFO: Scheduler started
Srv 1 2024-08-12 00:37:09 +0800 INFO: Cleaned up expired job results and finished schedules
Hello, the time is 2024-08-12 00:37:09.712927
Srv 0 2024-08-12 00:37:09 +0800 INFO: Job 959e4f24-44d0-48e2-9dc9-aa2f94c33e85 completed successfully
Hello, the time is 2024-08-12 00:37:09.742460
Srv 0 2024-08-12 00:37:09 +0800 INFO: Job af4e6e29-8b4b-4b2a-bd6d-81c7ef526aff completed successfully
Hello, the time is 2024-08-12 00:37:10.695703
Srv 1 2024-08-12 00:37:10 +0800 INFO: Job 4ea58325-5625-4a44-a3ad-3b813079b986 completed successfully
Hello, the time is 2024-08-12 00:37:11.661717
Srv 0 2024-08-12 00:37:11 +0800 INFO: Job cb49c58e-fdd5-4637-8ca8-2094d523d2d9 completed successfully
Hello, the time is 2024-08-12 00:37:12.656477
Srv 1 2024-08-12 00:37:12 +0800 INFO: Job abd77d3e-fe3f-4a76-b8bf-c9409fd5c037 completed successfully
Hello, the time is 2024-08-12 00:37:13.697547
Srv 2 2024-08-12 00:37:13 +0800 INFO: Job e9d32181-22c3-403d-b3d5-490200af2f7e completed successfully
Hello, the time is 2024-08-12 00:37:14.687171
Srv 0 2024-08-12 00:37:14 +0800 INFO: Job b056653f-5409-481c-8c04-813d906f096d completed successfully
Hello, the time is 2024-08-12 00:37:15.745309
Srv 2 2024-08-12 00:37:15 +0800 INFO: Job 7a808c01-518e-473b-addf-75485eb17acb completed successfully
Hello, the time is 2024-08-12 00:37:16.648569
Srv 3 2024-08-12 00:37:16 +0800 INFO: Job 801a899b-f689-44d0-8f68-c6c78dbb5085 completed successfully
Hello, the time is 2024-08-12 00:37:17.707701
Srv 0 2024-08-12 00:37:17 +0800 INFO: Job 65888e2f-87e0-4943-826d-75953c3362c8 completed successfully
Hello, the time is 2024-08-12 00:37:18.731820
Srv 3 2024-08-12 00:37:18 +0800 INFO: Job a8ffc28c-121f-458e-87e1-50d92b74614d completed successfully
Hello, the time is 2024-08-12 00:37:19.707290
Srv 0 2024-08-12 00:37:19 +0800 INFO: Job 6c28fb26-0027-414c-b6b5-849bb2915801 completed successfully
Hello, the time is 2024-08-12 00:37:20.674947
Srv 0 2024-08-12 00:37:20 +0800 INFO: Job 9ab2a1eb-19a1-4ab2-bf77-ef3174c5766c completed successfully
Main  2024-08-12 00:37:21 +0800 INFO: Received signal SIGINT. Shutting down.
Main  2024-08-12 00:37:21 +0800 INFO: Server Stopped

[agronholm]

Everything okay then?

[adk23333]

I'm currently having trouble understanding why it would repeatedly execute tasks multiple times upon startup. The number of repetitions, based on my tests, is not related to the number of processes.

Srv 0 20:06:34 DEBUG: Process ack: Sanic-Server-0-0 [6060]
Srv 0 20:06:34 INFO: Starting worker [6060]
Srv 1 20:06:34 DEBUG: Process ack: Sanic-Server-1-0 [34916]
Srv 0 20:06:34 DEBUG: Started internal cleanup loop with interval: 0:15:00
Srv 1 20:06:34 INFO: Starting worker [34916]
Srv 0 20:06:34 INFO: Scheduler started
Srv 0 20:06:34 INFO: Cleaned up expired job results and finished schedules
Srv 1 20:06:34 DEBUG: Started internal cleanup loop with interval: 0:15:00
Srv 1 20:06:34 INFO: Scheduler started
Srv 1 20:06:34 INFO: Cleaned up expired job results and finished schedules
Srv 1 20:06:34 DEBUG: Sleeping -67.747 seconds until the next fire time (2024-08-12 20:05:27.059907+08:00)
Srv 1 20:06:34 DEBUG: Sleeping -67.828 seconds until the next fire time (2024-08-12 20:05:27.059907+08:00)
Srv 0 20:06:34 DEBUG: Detected a ScheduleUpdated event – waking up the scheduler to process schedules
Srv 0 20:06:34 DEBUG: Sleeping 2.153 seconds until the next fire time (2024-08-12 20:06:37.059907+08:00)
Srv 1 20:06:34 DEBUG: Sleeping 2.141 seconds until the next fire time (2024-08-12 20:06:37.059907+08:00)
Srv 0 20:06:34 DEBUG: Sleeping 2.109 seconds until the next fire time (2024-08-12 20:06:37.059907+08:00)
Srv 1 20:06:37 DEBUG: Sleeping -0.053 seconds until the next fire time (2024-08-12 20:06:37.059907+08:00)
Srv 0 20:06:37 DEBUG: Sleeping 4.918 seconds until the next fire time (2024-08-12 20:06:42.059907+08:00)
Srv 1 20:06:37 DEBUG: Sleeping 4.884 seconds until the next fire time (2024-08-12 20:06:42.059907+08:00)
Hello, the time is 2024-08-12 20:06:37.791666  pid:  6060
Srv 0 20:06:37 INFO: Job 761d7c70-2d1b-4022-826c-c57188d1c6e8 completed successfully
Hello, the time is 2024-08-12 20:06:37.853539  pid:  34916
Srv 1 20:06:37 INFO: Job 761d7c70-2d1b-4022-826c-c57188d1c6e8 completed successfully
Hello, the time is 2024-08-12 20:06:37.869554  pid:  6060
Srv 0 20:06:37 INFO: Job 2ea59d5a-b1aa-4749-8fc5-50f26f265d7b completed successfully
Hello, the time is 2024-08-12 20:06:40.172784  pid:  6060
Srv 0 20:06:40 INFO: Job b735152d-9ee1-47a0-b348-1cb0e3a7fe23 completed successfully
Srv 1 20:06:42 DEBUG: Sleeping -0.043 seconds until the next fire time (2024-08-12 20:06:42.059907+08:00)
Srv 0 20:06:42 DEBUG: Sleeping 4.929 seconds until the next fire time (2024-08-12 20:06:47.059907+08:00)
Srv 1 20:06:42 DEBUG: Sleeping 4.923 seconds until the next fire time (2024-08-12 20:06:47.059907+08:00)
Hello, the time is 2024-08-12 20:06:45.161052  pid:  6060
Srv 0 20:06:45 INFO: Job ecca34d3-4b78-459d-8b76-41e52edcbf40 completed successfully
Srv 1 20:06:47 DEBUG: Sleeping -0.053 seconds until the next fire time (2024-08-12 20:06:47.059907+08:00)
Srv 0 20:06:47 DEBUG: Sleeping 4.923 seconds until the next fire time (2024-08-12 20:06:52.059907+08:00)
Srv 1 20:06:47 DEBUG: Sleeping 4.887 seconds until the next fire time (2024-08-12 20:06:52.059907+08:00)
Hello, the time is 2024-08-12 20:06:50.213290  pid:  6060

I made a few modifications to the test code.

async def tick():
    await sleep(3)
    print("Hello, the time is", datetime.now(), " pid: ", os.getpid())

and
# await app.ctx.scheduler.add_schedule(tick, IntervalTrigger(seconds=5), id="tick")

[agronholm]

What kind of setup do you have for the scheduler?

[adk23333]

What kind of setup do you have for the scheduler?

The settings are like before, and the logs show the same problem.

[agronholm]

It looks like you're sharing a data store but without a shared event broker. This is not a working setup. If you're sharing a data store among two processes, you also need to share an event broker, or else the schedulers will get out of sync.

[adk23333]

also need to share an event broker

Indeed, thank you for your help.