APScheduler with Rpyc & FastAPI attempting to achieve single instance

[PREPONDERANCE]

Background info: I have a FastAPI project utilizing APScheduler for cron jobs. As the title suggests, since FastAPI will initialize multiple workers on start leading to multiple instances of APScheduler, I'm working on implementing a single-instance APScheduler.

After re-referencing to the FAQ and attempting to replicate the method provided by the author, I encounter some errors.

I create a SchedulerManager class encapsulating the SchedulerService class that provides a start API which is responsible for calling scheduler.start() and server.start(). But when I add this to lifespan function and start my local server with uvicorn app.main:app --reload (--reload simply for test purposes), FastAPI simply stop reloading and cannot address any incoming requests. I suspect that when I start this SchedulerManager, it will somehow block the current event loop that FastAPI is running on.

Then I tried start the FastAPI server with uvicorn app.main:app --workers 4 to initialize multiple workers. This time it shows 'address already in use'.

So why is this happening? Any workarounds to address this issue?

(I'm all aware 4.0 version will probably solve this issue for once and for all. But my current project require this feature very badly, so I need the solution desperately)

The code I mentioned is displayed below.

class SchedulerService(rpyc.Service):
    def exposed_add_job(self, func, *args, **kwargs):
        return scheduler.add_job(func, *args, **kwargs)

    def exposed_modify_job(self, job_id, jobstore=None, **changes):
        return scheduler.modify_job(job_id, jobstore, **changes)

    def exposed_reschedule_job(self, job_id, jobstore=None, trigger=None, **trigger_args):
        return scheduler.reschedule_job(job_id, jobstore, trigger, **trigger_args)

    def exposed_pause_job(self, job_id, jobstore=None):
        return scheduler.pause_job(job_id, jobstore)

    def exposed_resume_job(self, job_id, jobstore=None):
        return scheduler.resume_job(job_id, jobstore)

    def exposed_remove_job(self, job_id, jobstore=None):
        scheduler.remove_job(job_id, jobstore)

    def exposed_get_job(self, job_id):
        return scheduler.get_job(job_id)

    def exposed_get_jobs(self, jobstore=None):
        return scheduler.get_jobs(jobstore)


class SchedulerManager:
    def __init__(self):
        self._initialize_scheduler()

    def _initialize_scheduler(self):
        jobstores = {
            "default": RedisJobStore(
                host=REDIS_HOST, port=REDIS_PORT, password=REDIS_PASSWORD, db=REDIS_DB
            ),
        }
        protocol_config = {'allow_public_attrs': True}

        self.scheduler = AsyncIOScheduler(jobstores=jobstores)
        self.server = ThreadedServer(SchedulerService, port=8000, protocol_config=protocol_config)
        self._configure_jobs()

    def _configure_jobs(self):
        self.scheduler.add_job(_send_class_notice_job, "cron", hour=7, minute=40) # The _send_class_notice_job is a utility function
        self.scheduler.add_job(_send_class_notice_job, "cron", hour=9, minute=30)
        self.scheduler.add_job(_send_class_notice_job, "cron", hour=13, minute=10)
        self.scheduler.add_job(_send_class_notice_job, "cron", hour=15, minute=0)
        self.scheduler.add_job(_send_class_notice_job, "cron", hour=18, minute=10)

    def start(self):
        self.scheduler.start()
        self.server.start()

    def shutdown(self):
        self.scheduler.shutdown()
        self.server.close()

@asynccontextmanager
async def lifespan(app: FastAPI):
     scheduler = SchedulerManager()
     scheduler.start()
     yield
     scheduler.shutdown()

[PREPONDERANCE]

I tried using a file lock without rpyc and it worked fine with 4 workers but not desired when --reload is activated

@asynccontextmanager
async def lifespan(app: FastAPI):

    f = open("scheduler.lock", "wb")
    global_scheduler = None
    try:
        fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
        scheduler_manager = SchedulerManager()
        global_scheduler = scheduler_manager
        scheduler_manager.start()
    except Exception:
        pass

    yield

    global_scheduler.shutdown()
    fcntl.flock(f, fcntl.LOCK_UN)
    f.close()

[PREPONDERANCE]

I understand the viability of separating apscheduler into a separate process but that would require extra effort to manage the service.
And yes, my web app does share intersection with the scheduler (Does the word 'interface' mean 'interfere' ? At least that's how I approach it)

[agronholm]

No, I didn't misspell "interface". What it means here is whether your web app needs to add or modify jobs, or retrieve job information from the scheduler.

[PREPONDERANCE]

No that's not required. Thanks for the clarification

[agronholm]

Ok, then you don't need RPyC for anything. If I understand correctly, you just want to make sure that the scheduler is always running alongside your web app. How do you deploy your application? Maybe I can suggest an alternative approach.

[PREPONDERANCE]

Ok, then you don't need RPyC for anything. If I understand correctly, you just want to make sure that the scheduler is always running alongside your web app. How do you deploy your application? Maybe I can suggest an alternative approach.

That's exactly correct and currently I'm using Gunicorn as a process manager to deploy the FastAPI project.