How to guarantee only one job working at a time?

[makorne]

Hi!

I tried to use register_with_count(1 with long running jobs and added several jobs.
But sometimes storage.list_jobs(&JobState::Running shows more then 1 job.

[geofmureithi]

Hey, You should be able to handle this with ConcurrencyLimitLayer for one job per worker or GlobalConcurrencyLimitLayer if you want only one job for all workers.
Just do:

.layer(ConcurrencyLimitLayer::new(1))
....

You can read more on the tower docs.
The reason you are experiencing this is that jobs are pulled by a thread pool under the hood. So a worker may be running multiple jobs at the same time. Tower has solutions for this.
Let me know this answers your question.

[makorne]

I tried this: .layer(ConcurrencyLimitLayer::new(1))

And received jobs.len() = 4. or jobs.len() = 6 anyway.

[geofmureithi]

Could you share a reproducible example? What storage are you using?

[makorne]

async fn test_redis_worker() {
    async fn produce_jobs(mut storage: RedisStorage<EmailReminder>) ->  Result<JobResult, JobError> {
        let workers = storage.list_workers().await.unwrap();  
        dbg!(&workers.len());
        let jobs_pending = storage.list_jobs(&JobState::Pending, 1).await.unwrap();  
        dbg!(&jobs_pending.len());
        let jobs_running = storage.list_jobs(&JobState::Running, 1).await.unwrap();
        dbg!(&jobs_running.len());
        for _i in 0..10 {
            storage.push(EmailReminder { }).await?;
        };
        println!("ADDed 10 jobs");
        Ok(JobResult::Success)
    }

    #[derive(Debug, Deserialize, Serialize, Default)]
    struct EmailReminder;

    impl Job for EmailReminder {
        const NAME: &'static str = "reminder::Email";
    }

    async fn send_reminder( _job: EmailReminder, _ctx: JobContext, ) -> Result<JobResult, JobError> {
        sleep_until(Instant::now() + Duration::from_secs(10)).await;
        Ok(JobResult::Success)
    }

    let mut storage = RedisStorage::connect("redis://127.0.0.1/").await.unwrap();

    //This can be in another part of the program
    let producer_storage = storage.clone();
    tokio::spawn(async move {
        loop {
            produce_jobs(producer_storage.clone()).await.unwrap();
            sleep_until(Instant::now() + Duration::from_secs(10)).await;
        }
    });

    Monitor::new().register_with_count(1, move |_| {
        WorkerBuilder::new(storage.clone())
            .layer(ConcurrencyLimitLayer::new(1))
            .build_fn(send_reminder.clone())
    }).run().await.unwrap();
}

[src/main.rs:36] &workers.len() = 3
[src/main.rs:38] &jobs_pending.len() = 11
[src/main.rs:40] &jobs_running.len() = 3
ADDed 10 jobs
[src/main.rs:36] &workers.len() = 4
[src/main.rs:38] &jobs_pending.len() = 11
[src/main.rs:40] &jobs_running.len() = 4
ADDed 10 jobs

[geofmureithi]

The problem here is most likely that you are interacting with old data.
Even before you are adding the first job, you already have some stale data possibly a previous run.
Can you run redis-cli > FLUSHDB before running?

[makorne]

Yes.
But after first restart - again:

[src/main.rs:36] &workers.len() = 1
[src/main.rs:38] &jobs_pending.len() = 11
[src/main.rs:40] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] &workers.len() = 2
[src/main.rs:38] &jobs_pending.len() = 11
[src/main.rs:40] &jobs_running.len() = 2
ADDed 10 jobs 

No programs what never restart.
How to guarantee only one job working at a time?

[geofmureithi]

This issue is because of how data is fetched from redis. I have added an issue #42 and it should fix the worker part. For the case of jobs running, it is because we dont use transactions in redis. And hence, jobs_running.len() may be out of date.
Consider a scenario where a job is complete but the worker is updating redis in the background a new job can start while this is happening(hence only one job is running, but redis can show two are). This will be fixed in 0.4 when we remove actor-worker based job processing and move to entirely layer based, processing.
Nonetheless, I have one more extra check I would like you to run to confirm that only one job is running. Please incorporate this code:

// 1. Create an atomic counter
let running = Arc::new(AtomicI32::new(0)); 

// 2. clone it and pass it to job via Extension Layer
....
.layer(ConcurrencyLimitLayer::new(1))
.layer(Extension(running.clone()))

// 3. In the job fn request it and interact with it
async fn send_reminder( _job: EmailReminder, ctx: JobContext, ) -> Result<JobResult, JobError> {
    let running: Arc<AtomicI32> = ctx.data_opt().unwrap();
    running.fetch_add(1, Ordering::SeqCst);
    sleep_until(Instant::now() + Duration::from_secs(10)).await;
    running.fetch_sub(1, Ordering::SeqCst);
    Ok(JobResult::Success)
}

// 4. in your produce jobs, pass in running as fn argument just like storage and check
async fn produce_jobs(mut storage: RedisStorage<EmailReminder>, running:  Arc<AtomicI32>) {
    dbg!(running.load(Ordering::SeqCst));
    ....
}

This should give us a better idea of how many jobs are running.
Can you try this and get back
You can refer to this answer if need: https://stackoverflow.com/a/54598041/1418750

[makorne]

And hence, jobs_running.len() may be out of date. Consider a scenario where a job is complete but the worker is updating redis in the background a new job can start while this is happening(hence only one job is running, but redis can show two are).

For exclude this scenario I just changed executing time to 100 seconds:

    async fn send_reminder( _job: EmailReminder, _ctx: JobContext, ) -> Result<JobResult, JobError> {
        sleep_until(Instant::now() + Duration::from_secs(100)).await;
        Ok(JobResult::Success)
    } 

[src/main.rs:36] &workers.len() = 2
[src/main.rs:38] &jobs_pending.len() = 11
[src/main.rs:40] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] &workers.len() = 3
[src/main.rs:38] &jobs_pending.len() = 11
[src/main.rs:40] &jobs_running.len() = 3
ADDed 10 jobs

[geofmureithi]

Could you please do the check as requested, this does not help as there are bugs in fetching from redis already.

[makorne]

[src/main.rs:36] running.load(Ordering::SeqCst) = 0
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs

After restart:

[src/main.rs:36] running.load(Ordering::SeqCst) = 0
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 3
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 3
ADDed 10 jobs

[geofmureithi]

Can you let it run for more than just two times?

[makorne]

[src/main.rs:36] running.load(Ordering::SeqCst) = 0
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 3
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 3
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 3
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 3
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 3
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 3
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 3
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 3
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 3
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 3
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 3
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 3
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 3
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 3
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 3
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 3
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 3
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 3
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs

[geofmureithi]

Clearly there was some stale data from previous runs. I hope this solves your problem. 🙂

[makorne]

After FLUSHDB :

[src/main.rs:36] running.load(Ordering::SeqCst) = 0
[src/main.rs:38] &workers.len() = 0
[src/main.rs:40] &jobs_pending.len() = 0
[src/main.rs:42] &jobs_running.len() = 0
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 9
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs

Then after first restart:

[src/main.rs:36] running.load(Ordering::SeqCst) = 0
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs

After 2 restart:

[src/main.rs:36] running.load(Ordering::SeqCst) = 0
[src/main.rs:38] &workers.len() = 1
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 1
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs
[src/main.rs:36] running.load(Ordering::SeqCst) = 1
[src/main.rs:38] &workers.len() = 2
[src/main.rs:40] &jobs_pending.len() = 11
[src/main.rs:42] &jobs_running.len() = 2
ADDed 10 jobs

I think this is a bug.

[geofmureithi]

Sorry, this is empty

[makorne]

Do you mean that the jobs_running.len() = 3 is show wrong, and only one send_reminder really working at time?

[geofmureithi]

Yes, you can see only one job is running via running.load(Ordering::SeqCst) = 1. I will document the bug, but its most likely a during stopping, a job is left in the state of running, but its worker is dead. That job will stay in that state. This bug will be fixed by #42