Concurrency Limit apalis 0.5.0

[AzHicham]

Hello

Since version 0.5.0, we lost the concurrency limit feature introduced by this PR.
In fact using GlobalConcurrencyLimitLayer or ConcurrencyLimitLayer will allow to run at most 1 job per worker, without fetching jobs from Redis allowing other application (I have the same app scaled on a K8s cluster) to pull jobs which is good and not causing starvation !!!!

But what I do not understand is what's the point of having this function fn register_with_count + config.buffer_size = 1 if the worker will still run many async jobs ?
IMO with a config like the one below, we should not have to add a ConcurrencyLimitLayer because:

• The storage is configured to fetch 1 job at a time
• The ONLY worker present should handle a job then notify the monitor that he is available for more work allowing the monitor to fetch again in Redis for more jobs

let redis_connection = make_redis_connection(&settings.broker).await?;
let config = {
        let mut config = Config::default();
        config.set_buffer_size(1);
        config.set_fetch_interval(Duration::from_millis(100));
        config.set_max_retries(3);
        config.set_keep_alive(Duration::from_secs(120));
        config
    };
let storage = RedisStorage::new_with_config(redis_connection, config);
let worker_msintuit =
        WorkerBuilder::new(format!("worker-{}", rng.gen::<u16>()))
            .data(app_state.clone())
            .with_storage(storage)
            .build_fn(handler);

Thank you :)

[geofmureithi]

0.5 brought clarity between concurrency and parallelism.

Internally here is what happens:

1. The worker checks if service is ready. Meaning if you have layers like ConcurrencyLayer will be checked.
2. The worker will ask the poller for a job when service is ready.
3. The worker will get the next job and it will call spawn, meaning it will not wait for the job to complete and it will just request the next. (Parallelism)

Register with count offers concurrency. The number of workers will all follow the same process as above.
If you only need one worker, have you considered just using register?

The problem you are facing is the change that the service is now clone and each worker gets a clone of the service. The config provided handles interaction between the Backend and the Storage. The tower layers and register control the worker. These are separate.

[AzHicham]

I fact you're right using ConcurrencyLimitLayer works just fine if I set juste 1 worker.
I'll stick to that then :)

[AzHicham]

Also I'm already using register_with_count(1, worker) which should be the same as register

[geofmureithi]

See also
#222 (comment)

[geofmureithi]

You may not want to have a buffer_size of 1 anymore, This just means more requests to Redis, you should prefer the layer approach, to control workers.

let config = {
        let mut config = Config::default();
        config.set_buffer_size(1);
        config.set_fetch_interval(Duration::from_millis(100));
        config.set_max_retries(3);
        config.set_keep_alive(Duration::from_secs(120));
        config
    };
let storage = RedisStorage::new_with_config(redis_connection, config);

This does not affect workers, it just tells the backed to fetch one job every 100ms.

[AzHicham]

Maybe I'm wrong but the buffer_size still matters.
Let's say I have a Redis on a Server A, and my application running on server B and C.
I push to Redis 5 jobs, let's say my app on server B fill be the first to fetch jobs from Redis, it will pull buffer_size jobs.
If buffer_size >= 5 then my app running on server C will do nothing all jobs have been fetched.

Since my jobs are long running job (>= 20 min) I'll waste server C ressources.
On the other hand If I set buffer_size to 1 server B & C will process jobs in parallel which is what I want.

[geofmureithi]

Noted, at the end config should handle this.

[derekleverenz]

I'm working on controlling concurrency in my application too, and I want to check if my understanding is correct. It seems like there is no source of back pressure for a worker unless you configure a concurrency limit with tower's ConcurrencyLImitLayer or similarly use a rate limit layer, or other layer that would stop the service from being ready, so without that a single worker would consume anything on the queue as fast as it could and spawn tasks for it. Is that correct? I am using a concurrency limit layer but I'm trying to understand if there is some other source of backpressure I am missing.

[derekleverenz]

Actually i just saw #302 and I think that answers my question (that is to say, there isnt a limit unless i use a concurrency layer or other limit)