Stream background job results to MPSC channel not ending

[PainOchoco]

Hello,

I've recently discovered this cool library and I wanted to use it to make parallel and scheduled calculations.
I want to get a stream output out of the jobs for real time data, I'm trying to use a MPSC channel where i give each job a sender and when the job is complete it will use that sender to send the data to the receiver.

The issue is that the Monitor never ends, and I don't know why. I think this is related to the original sender not being closed/dropped but I can't figure out a way to drop it without breaking something.

Here is a minimal example of what I described:

use std::thread::sleep;
use std::time::Duration;

use apalis::prelude::*;
use apalis::{
    prelude::{Data, Job, WorkerBuilder},
    redis::RedisStorage,
};
use futures::channel::mpsc::{channel, Sender};
use futures::StreamExt;
use serde::{Deserialize, Serialize};

#[derive(Clone)]
struct JobContext {
    sender: Sender<i32>,
}

#[derive(Debug, Clone, Deserialize, Serialize)]
struct SomeJob {
    id: i32,
}

impl Job for SomeJob {
    const NAME: &'static str = "apalis::SomeJob";
}

async fn send_id(job: SomeJob, ctx: Data<JobContext>) {
    let mut cloned_sender = ctx.sender.clone();
    sleep(Duration::from_secs(1));
    cloned_sender.try_send(job.id).unwrap();
    println!("Sent {}", job.id);
}

async fn produce_jobs(mut storage: RedisStorage<SomeJob>) {
    for i in 0..12 {
        let job = SomeJob { id: i };
        storage.push(job).await.unwrap();
    }
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = channel::<i32>(1024);

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

    produce_jobs(storage.clone()).await;

    tokio::task::spawn(async move {
        while let Some(x) = rx.next().await {
            println!("Received {:?}", x);
        }
        println!("This is never reached, loop will not finish");
    });

    let job_ctx = JobContext { sender: tx };
    let worker = WorkerBuilder::new("testing")
        .data(job_ctx)
        .with_storage(storage.clone())
        .build_fn(send_id);

    Monitor::<TokioExecutor>::new()
        .register_with_count(3, worker)
        .run()
        .await
        .unwrap();

    println!("This is never reached either, monitor is still waiting for workers to finish");
}

Please tell me if there's another way of doing what i'm trying to do!
And if you find a way to close the MPSC channel gracefully let me know!

Thanks

[geofmureithi]

Hello! Welcome to our discussions. We will walk together and hopefully help you out 🙂.

Your approach should work, you may consider sending an enum something like

enum Control {
   Continue(TaskId),
   Break
}

This can be combined with Monitor::run_with_signal where the break is sent when the shutdown is called.

That said apalis should provide this functionality via a tower layer. A tower layer is preferable because it's reusable and doesn't need that much code.

If you could specify what you want to do with the result I can write you a basic starter layer.

[PainOchoco]

Yes that's right, the job's result being a stream.

[PainOchoco]

Here's a quick example of a subscription:

use entity::async_graphql::{self, Context, Subscription};
use futures::{channel::mpsc, stream::Stream};
#[derive(Default)]
pub struct SomeSubscription;

#[Subscription]
impl SomeSubscription {
    async fn iterate(&self, _ctx: &Context<'_>) -> impl Stream<Item = i32> {
        let (tx, rx) = mpsc::channel::<i32>(1000);

        // Dispatch job to worker with a tx
        // Start Monitor

        rx
    }
}

[geofmureithi]

Ok, this should be doable, let me get you a working example later in the day.

[PainOchoco]

Bump 🙏

[geofmureithi]

The first thing to note is you cant run monitor in the subscription, unless you want to create a new monitor for every request.

[geofmureithi]

Step 1: Setup Both your monitor and async-graphql in the main
Take a look at how to run both monitor and asyncgraphql in parallel here:
https://github.com/IgnisDa/ryot/blob/6d0382113acc91e323b2b69ba7da8d34741b629d/apps/backend/src/main.rs#L250

Step 2: Create a channel in the tokio::main fn.

let (tx,  rx) = channel::<i32>(1024);

Step 3: Pass in the receiver to async graphql .data method eg in
https://github.com/async-graphql/examples/blob/7eee23ad4e005ac180df8111e6940cebb7c1f52f/axum/starwars/src/main.rs#L25

.data(rx)

Step 4: Create a layer that takes in the sender (this assumes your result is a String for simplicity)

pub struct ChannelResultLayer(Sender<String>);

impl<S> Layer<S> for ChannelResultLayer {
    type Service = ChannelResultService<S>;

    fn layer(&self, service: S) -> Self::Service {
        ChannelResultService {
            service,
            sender: self.sender.clone()
        }
    }
}

pub struct ChannelResultService<S> {
   service: S,
   sender: Sender<String>
}

impl<S, J> Service<Request<J>> for SaveResultService<S>
where
    S: Service<Request<J>>,
    Request: std::fmt::Debug,
    J: Job
{
   type Response = S::Response;
   type Error = S::Error;
   type Future = S::Future; 

   fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
       self.service.poll_ready(cx)
   }

   fn call(&mut self, request: Request<J>) -> Self::Future {
       let id = request.id();
       let fut = self.service.call(request);
      let sender = self.sender.clone();
       Box::pin(async move {
        let res = fut.await;
        sender.send(res.to_string());
        res
       })
   }
}

Step 5: Pass in our layer to the worker:

let worker = WorkerBuilder::new("testing")
        .layer(ChannelResultLayer(tx))
        .with_storage(storage.clone())
        .build_fn(send_email);

Step 6: Pass in the Receiver to the subscription:

#[Subscription]
impl SomeSubscription {
    async fn iterate(&self, ctx: &Context<'_>) -> impl Stream<Item = String> {
       let rx = ctx.data::<Receiver<String>>();
        rx
    }
}

This should demonstrate what you need to do. Let me know if it helps.

[PainOchoco]

Thanks for this example, it helps me a lot!
But what if I want to have each request their own independent stream? It seems that right now if multiple user subscribe in the same time frame, they will all get a mix/conflict of data.
The Subscription I'm making is not a global feed but rather a one-one feed with the client.

As such should I reconsider the choice of subscription over queries or should I create a unique channel for each of the subscription and try to adapt your example?

[geofmureithi]

Also please clarify what you mean by each request their own independent stream
A request in the context of tower and apalis is a single job, but from my understanding here its like you mean all jobs of the same type. Right?

[geofmureithi]

I think the current approach could be modified with 1. Make the sender send a tuple eg ‘Res(Id, String)’ or something similar. 2. Graphql subscriptions allow you to specify an input See https://graphql.org/blog/subscriptions-in-graphql-and-relay/ You could let the user pass the id they want to subscribe to. 3. Leverage stream functions to filter/filter_map the stream to provide each subscription with specific data. See https://docs.rs/futures/latest/futures/stream/trait.StreamExt.html#method.filter Let me if that works.

[PainOchoco]

That should work, I just can't get the rx from the subscription.

let rx = ctx.data::<Receiver<i32>>(); // not a Receiver<i32> but a &Receiver<i32>

And I can't do anything with a receiver reference. I've opened an issue over on the async-graphql repo to ask that question.
async-graphql/async-graphql#1495

[geofmureithi]

You might need to use Arc<Mutex<T>>. Apalis has Notify btw which implements Stream

[PainOchoco]

That was a good idea but I can't get that to work. It seems that an Arc<Mutex<Receiver>> cannot get filtered after lock() and unwrap().

let receiver = ctx.data::<Arc<Mutex<Receiver<i32>>>>().unwrap();
receiver
    .lock()
    .unwrap()
    .filter(|x| future::ready(true /* filter with id */));

std::sync::MutexGuard<'_, futures::futures_channel::mpsc::Receiver<i32>> is not an iterator

Also, this approach could result into a bottleneck since the receiver might need to get accessed by multiple threads as the same time, which the mutex prevents and might slow down the process.

The new idea i'm trying to implement is this:

1. Client request a subscription
2. Server create an MPSC channel for that unique request
3. It then produce jobs for the workers, the job must specify the sender* to send back the result (a stream in this case)
4. Some workers work on the jobs and send the result back

• I don't know how to do that with Apalis
  Do you have an idea on how could I tell the workers to send the data back via the sender i provided?

[geofmureithi]

You may need to deref the Mutex guard

[PainOchoco]

It's not possible

cannot move out of dereference of `std::sync::MutexGuard<'_, futures::futures_channel::mpsc::Receiver<i32>>`
move occurs because value has type `futures::futures_channel::mpsc::Receiver<i32>`, which does not implement the `Copy` trait

Do you think my idea that I explained above would be more suitable (one channel created by request)?

[geofmureithi]

Please check out https://github.com/geofmureithi/apalis/blob/master/packages/apalis-core/src/notify.rs It has the implementation you are looking for. Regarding your approach, its possible but you may need to create a type for each subscription. If the only issue remaining is the stream part I suggest you look at the code I have shared with you. Else you give me until Tuesday as I am on vacation.

[geofmureithi]

Let me know if this helped.

[PainOchoco]

What I've finally done is create a global hashmap that links job ids to senders.

When a client makes a request, the server :

• Creates a MPSC channel (tx, rx)

• Creates a Job

• Add tx to the global hashmap with the job id

• Add job to storage

• Workers receives the job

• Get the tx sender from the global hashmap and clone it

• Work on the job and send the results with the cloned tx

It seems to work pretty well, tho i've not tested it at big scale, I wonder if the hashmap mutex will bottleneck when trying to lock it in multiple threads at the same time.

[PainOchoco]

I'll mark that as the final answer since this works like I wanted too.
It's not perfect because the dashmap lives into the main process hence we can't scale that horizontally on multiple machines connecting to the same Redis database. If horizontal scale is needed I think that we need to use a message broker like amqp (apalis-amqp might be useful in this case).

Thanks a lot!

[geofmureithi]

You might want to consider dashmap or something else concurrent.

[PainOchoco]

Cool, I'll use that!