example of registering multiple job in one worker

[ardevik]

Hi,

Is there any example for how registering multiple job in one worker ?

My use case is, i want queue multiple job in one worker so only one job can run at a time in one machine, and if possible i want separate the one worker and each job in different project which is have his own virtual environment that contain different version of library, i plan run the job on subprocess for isolated virtual environment

[janbjorge]

Hi,

I'm having a hard time fully understanding what you mean, but it sounds a bit like you would want to create a QueueManager (QM) per virtual environment. Make sure that the entry point names are unique per QM, as a QM will only dequeue jobs for the entry points it knows of.

Here's an updated approach that could align more closely with your requirements:

1. QueueManager per Virtual Environment: Create a QueueManager for each virtual environment, ensuring that each has a unique entry point name. This will allow the individual QMs to manage job execution without conflict, especially when dealing with different library dependencies across projects.

For example:

from pgqueuer import QueueManager

async def qm1_factory(...):
    qm = QueueManager(...)
    
    @qm.entrypoint("qm1_entrypoint")
    def qm1_entrypoint(...):
        ...
    return qm

async def qm2_factory(...):
    qm = QueueManager(...)
    
    @qm.entrypoint("qm2_entrypoint")
    def qm2_entrypoint(...):
        ...
    return qm

Now you can create two or more virtual environments, each with its own version of the required libraries, and invoke them using:

python -m pgqueuer run <path-to-factory-function>

This approach lets you isolate different projects by using unique QueueManagers and virtual environments, achieving both separation and the ability to run one job at a time per machine.

Let me know if this makes sense or if you'd like more details about the setup!

[ardevik]

from your example you register in different entrypoint in one consumer, if i separate the 2 task into 2 consumer file and run each consumer. can it still only run 1 task at a time, is i need add concurrency_limit = 1 ?

and

I just found this

In this example, jobs registered under the entrypoint process_shared_resource will be processed strictly one at a time, ensuring no other jobs of the same type are processed concurrently.

is this implementation correct ?

consumer_1.py

from pgqueuer import QueueManager

from app import task_1

qm = QueueManager(...)

@qm.entrypoint("shared_job",  serialized_dispatch=True)
task_1()

consumer_2.py

from pgqueuer import QueueManager

from app import task_2

qm = QueueManager(...)

@qm.entrypoint("shared_job",  serialized_dispatch=True)
task_2()

but i still confused for how enqueue task_1 and task_2 simultaneously

trigger by different user
user 1 want trigger task_1, user 2 want trigger task_2
producer.py

from pgqueuer import QueueManager

qm = QueueManager(...)

qm.queries.enqueue( ? )

and what different between concurrency_limit and serialized_dispatch ?

[janbjorge]

To run one job at a time on a machine across multiple consumers (consumer_1.py, consumer_2.py), use serialized_dispatch=True in your entry points. This ensures only one job globally with the same entry point name runs concurrently. If you run consumer_1 and consumer_2 at the same time, only one of them will pick a job and run it until it's completed. The other will remain idle for the duration of the processing. This is because serialized_dispatch=True enforces strict serialization of job execution. However, this does not guarantee that jobs will be round-robin distributed to the consumers.

but i still confused for how enqueue task_1 and task_2

In your 'producer.py', you can use batch-insert to enqueue multiple jobs at once:

import asyncio

import asyncpg

from pgqueuer.db import AsyncpgDriver
from pgqueuer.queries import Queries

# Enqueue multiple jobs using the enqueue function
async def enqueue_jobs():
    connection = await asyncpg.connect()
    driver = AsyncpgDriver(connection)
    queries = Queries(driver)
    await queries.enqueue(
        entrypoint=["shared_job", "shared_job"],
        payload=[b"task_1_payload", b"task_2_payload"],
        priority=[0, 0]
    )

# Run the enqueue function
asyncio.run(enqueue_jobs())

what different between concurrency_limit and serialized_dispatch ?

Local vs Global Limits

• concurrency_limit: Limits the number of concurrent jobs a specific consumer can run at any given point (local limit). Note: There is no need to set concurrency_limit if serialized_dispatch=True, as serialized_dispatch is stricter.

• serialized_dispatch=True: Ensures that only one job globally for a given entry point can run at any time (global limit).

[janbjorge]

if i look in example above payload is an argument we can pass to job, i don't need round robin, what i want is enqueue specific job in same entrypoint with serialized_dispatch, so only 1 job we choose can run and other job we choose later get queued

Sounds to me that you want serialized_dispatch=true then. Just note that if your entrypoint functions is slow, you might end up with a really heavy backlog at times.

i understand your first approach now, but if i add more job or modify existing job, i need to modify the single consumer script and re-launch every instance that already run in different virtual environment, and it not make sense to me as you register in different entrypoint how you limit it only one job can run at a time as serialized_dispatch need same entrypoint right?

A consumer will only pick items from the queue which it got an entrypoint to route the job/item to. If you create a new consumer, it only needs to implement the new entrypoint as it will only pick up jobs for the given entrypoint (assuming the entrypoints names are unique).

also we run it in different virtual environment, that mean multiple instance of consumer ?

Yes, consumers share close to zero state (except a few counters to estimate requests per second)

Or can you advice other approach ?

I can, but i still feel like i do not fully understand your requirements. If your code is open source; Then im willing to take a quick peak at it.

[ardevik]

A consumer will only pick items from the queue which it got an entrypoint to route the job/item to. If you create a new consumer, it only needs to implement the new entrypoint as it will only pick up jobs for the given entrypoint (assuming the entrypoints names are unique).

Yes, consumers share close to zero state (except a few counters to estimate requests per second)

If new consumer only needs to implement new entrypoint and consumers share close to zero state, can i ask feature request to add flag to grouping multiple consumers and limit how many consumers at that group can run at a time ?

[janbjorge]

You can ask, but at this point of time. I still need to understand the requirements more. Your ofc. welcome to fork and submit a feature PR.

When you say grouping multiple consumers, what would identify an consumer as belonging to a group? To me, this sounds a bit like an infrastructure issue, that i typically would argue is something that should be solved in kubernetes or similar?

[ardevik]

my requirements:

• my machine is VM no container or orchestrator
• i have multiple jobs
• i need only 1 job can run per machine because the jobs is long running and take up resource like RAM and Hard Disk, my job is ETL script using polars library and need avoid in using multithread, the option is to use multiprocessing or subprocess
• can queue the job from fastapi
• each job need to be independent, i don't want restart existing consumer or job that already deployed when i deploy new one, each job own venv and run from that venv to accommodate different version of polars library :
  option1 - 1 consumer only 1 job can be registered, running on its own venv, only 1 consumer can run across multiple consumer, job executor run in multiprocessing
  option2 - 1 consumer multiple job can be registered, because i dont want restart existing running consumer is there a way to register new job to consumer without update consumer script or restart the consumer as long as the job script is callable from consumer?, only 1 job can run per consumer, job executor run in subprocess because the job script need to be run in different venv

[janbjorge]

my machine is VM no container or orchestrator

How will you rollout new consumers? Will you start a new VM for the new entrypoint? Say you got file-with-ETL-script-1 below, and have that running in VM as;

# Its own venv with pgqueuer installed.
pgq run file-with-ETL-script-1

from __future__ import annotations

from datetime import datetime

import asyncpg

from pgqueuer import PgQueuer
from pgqueuer.db import AsyncpgDriver
from pgqueuer.models import Job, Schedule


async def main() -> PgQueuer:
    connection = await asyncpg.connect()
    driver = AsyncpgDriver(connection)
    pgq = PgQueuer(driver)

    @pgq.entrypoint("run-etl-script-1", serialized_dispatch=True)
    async def process_message(job: Job) -> None:
        print(f"Processed message: {job!r}")

    return pgq

After some time, you want to add a consumer (its own file file; file-with-ETL-script-2)

# Its own venv with pgqueuer installed.
pgq run file-with-ETL-script-2

from __future__ import annotations

from datetime import datetime

import asyncpg

from pgqueuer import PgQueuer
from pgqueuer.db import AsyncpgDriver
from pgqueuer.models import Job, Schedule


async def main() -> PgQueuer:
    connection = await asyncpg.connect()
    driver = AsyncpgDriver(connection)
    pgq = PgQueuer(driver)

    @pgq.entrypoint("run-etl-script-2", serialized_dispatch=True)
    async def process_message(job: Job) -> None:
        print(f"Processed message: {job!r}")

    return pgq

In this case, these two 'runners' does not know about the other consumers. As long as you only ammend new consumers, it will work fine since consumers only pick job for entrypoints they know they have a consumer function for. They also have their own venv.

i have multiple jobs

Should be okey, just add them to the queue, and they will be picked up once a consumer is ready consume it.

i need only 1 job can run per machine because the jobs is long running and take up resource like RAM and Hard Disk, my job is ETL script using polars library and need avoid in using multithread, the option is to use multiprocessing or subprocess

I my ex. above i use serialized_dispatch=True, its very strict in terms of when its allowed to start on new jobs. For me, its sounds like what you want.

can queue the job from fastapi;

can you take a look at my fast api example / recipe? https://github.com/janbjorge/pgqueuer/blob/main/recipes/fastapi_recipe.py.

each job need to be independent, i don't want restart existing consumer or job that already deployed when i deploy new one, each job own venv and run from that venv to accommodate different version of polars library; option1 - 1 consumer only 1 job can be registered, running on its own venv, only 1 consumer can run across multiple consumer, job executor run in multiprocessing option2 - 1 consumer multiple job can be registered, because i dont want restart existing running consumer is there a way to register new job to consumer without update consumer script or restart the consumer as long as the job script is callable from consumer?, only 1 job can run per consumer, job executor run in subprocess because the job script need to be run in different venv

In my above ex. the consumers are independent, and can be extented by adding new entrypoints to own 'scripts' that you spin up as you go.

Hot swapping code in python is dueabole, but not something that i will add support for. Ive seen people use SIGHUP to force moudles reload using https://docs.python.org/3/library/importlib.html . To me, this sounds like a flaky path that i think should be avoided.