Execute fire and forget for long running process

[DevEnol]

Hi!

I need to implement a method that would trigger fire and forget behavior. The code I wrote are something like this:

public Uni<String> startProcess() {
    longRunningProcess().subcribe().with(logger::info(message))
    return Uni.createFrom().item("Processing");
}

public Uni<String> longRunningProcess() {
    //Perform long running task then return with String: "Finished"
}

This code seems to work as my expected behavior. Is this code functioning as intended, or is there room for improvement?
And are there any better ways to implement this?

[cescoffier]

In general, when I need to do such kind of things, I use the following pattern:

public Uni<String> longRunningProcess() { ... }

// ...

var csJustInCaseINeedToCheck = longRunningProcess()
     .runSubscriptionOn(Infrastructure.getDefaultWorkerPool() // If it's blocking, make sure it does not block the current thread
    .subscribeAsCompletionStage(); // Trigger the operation and return a completion stage so I can check if it completed or failed

[DevEnol]

@cescoffier thanks for the reply.

[starksm64]

@cescoffier , your example does not behave the way I would expect for a "fire and forget" as I would not want the main thread to wait on the long running task Uni to finish. To see the behavior I would expect one needs to execute the the long running task in another thread and immediately return the Uni result. I have implemented an example in the current mutiny-project github repo, and run1 is your example, run2 is what is needed to get the behavior I would expect:

package _01_basics;

import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.infrastructure.Infrastructure;

import java.util.concurrent.TimeUnit;

public class _09a_Uni_From_CompletionStage {

    static long start = System.currentTimeMillis();
    public static void main(String[] args) throws Exception {
        log("⚡️ Uni from CompletionStage, fire and forget?");
        run1();
        boolean finished = Infrastructure.getDefaultWorkerPool().awaitTermination(5, TimeUnit.SECONDS);
        log("⚡️ awaitTermination called, "+finished);
    }
    static void run1() throws Exception {
        var cs = longRunningProcess()
                .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
                .subscribeAsCompletionStage();
        log("run1 begin sleep, longRunningProcess.isDone: "+cs.isDone());
        Thread.sleep(2000);
        log("run1 end sleep, longRunningProcess.isDone: "+cs.isDone());
        while (!cs.isDone()) {
            Thread.sleep(1000);
        }
        log("run1 done waiting, longRunningProcess.result: "+cs.get());
    }

    static void run2() throws Exception {
        var cs = longRunningProcess2()
                .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
                .subscribeAsCompletionStage();
        log("run2 begin sleep, longRunningProcess2.isDone: "+cs.isDone());
        Thread.sleep(2000);
        log("run2 end sleep, longRunningProcess2.isDone: "+cs.isDone());
        while (!cs.isDone()) {
            Thread.sleep(1000);
        }
        log("run2 done waiting, longRunningProcess2.result: "+cs.get());
    }

    static Uni<String> longRunningProcess() {
        Uni<String> result = Uni.createFrom().item("Finished.longRunningProcess");
        log("begin longRunningProcess");
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
        }
        log("end longRunningProcess");
        return  result;
    }

    static Uni<String> longRunningProcess2() {
        Uni<String> result = Uni.createFrom().item("Finished.longRunningProcess2");
        Infrastructure.getDefaultWorkerPool().submit(() -> {
            log("begin longRunningProcess2");
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
            }
            log("end longRunningProcess2");
        });
        return  result;
    }

    static void log(String message) {
        long timestamp = System.currentTimeMillis();
        long elapsed = timestamp - start;
        System.out.printf("[%6d](%s): %s\n", elapsed, Thread.currentThread().getName(), message);
    }

}

Example with run1():

[     0](main): ⚡️ Uni from CompletionStage, fire and forget?
[    14](main): begin longRunningProcess
[  5019](main): end longRunningProcess
[  5029](main): run1 begin sleep, longRunningProcess.isDone: false
[  7035](main): run1 end sleep, longRunningProcess.isDone: true
[  7036](main): run1 done waiting, longRunningProcess.result: Finished.longRunningProcess
[ 12039](main): ⚡️ awaitTermination called, false

Example with run2():

[     0](main): ⚡️ Uni from CompletionStage, fire and forget?
[    17](pool-1-thread-1): begin longRunningProcess2
[    23](main): run2 begin sleep, longRunningProcess2.isDone: false
[  2029](main): run2 end sleep, longRunningProcess2.isDone: true
[  2030](main): run2 done waiting, longRunningProcess2.result: Finished.longRunningProcess2
[  5023](pool-1-thread-1): end longRunningProcess2
[  7037](main): ⚡️ awaitTermination called, false

Of course in the run2 case, the CompletionStage.isDone value has no meaning with respect to the real background process.

[jponge]

I'm not sure what is the problem here, except that in your first case you have a thread sleep on the thread that creates the pipeline.

[starksm64]

Ok, looping back to this. What I was missing was having the first longRunningProcess impl do the work inside of its Uni.createFrom().item. With that change, the longRunningProcess does execute on the Infrastructure.getDefaultWorkerPool(), but the result is still not what I would want for a true fire and forget behavior like a CDI fireAsync() event would do. Here are the traces and updated code:

Example with run1():

[     0](main): ⚡️ Uni from CompletionStage, fire and forget?
[    19](main): run1 begin sleep, longRunningProcess.isDone: false (1)
[    20](pool-1-thread-1): begin longRunningProcess (2)
[  2025](main): run1 end sleep, longRunningProcess.isDone: false (3)
[  5025](pool-1-thread-1): end longRunningProcess (4)
[  5026](pool-1-thread-1): After longRunningProcess item (5)
[  5038](main): run1 done waiting, longRunningProcess.result: Finished.longRunningProcess (6)
[ 10047](main): ⚡️ awaitTermination called, false

(1): main thread is sleeping while longRunningProcess should be running in worker pool due to subscription
(2): longRunningProcess has started running in worker pool
(3): main thread is past sleep, checking on CompletionStage status and finds it is not done.
(4): longRunningProcess completes running in worker pool
(5): the invoke after the longRunningProcess is called in worker pool
(6): main thread waits until CompletionStage is done and then prints result.

Example with run2():

[     0](main): ⚡️ Uni from CompletionStage, fire and forget?
[    16](pool-1-thread-1): begin longRunningProcess2 (1)
[    22](main): run2 begin sleep, longRunningProcess2.isDone: false (2)
[    25](pool-1-thread-2): After longRunningProcess item (3)
[  2029](main): run2 end sleep, longRunningProcess2.isDone: true (4)
[  2030](main): run2 done waiting, longRunningProcess2.result: Work not part of Uni chain (5)
[  5021](pool-1-thread-1): end longRunningProcess2 (6)
[  7040](main): ⚡️ awaitTermination called, false

(1): longRunningProcess2 has started running in worker pool due to its own submit to worker pool
(2): main thread is sleeping while longRunningProcess should be running in worker pool
(3): the invoke after the longRunningProcess is called in worker pool
(4): main thread is past sleep, checking on CompletionStage status and finds it IS done because no work was done in the item
(5): main thread waits until CompletionStage is done and then prints result.
(6): longRunningProcess2 completes running in worker pool

The run1 behavior is what I would call 'fire in background thread', but you can access a result, and downstream elements wait for the longRunningProcess to complete before being invoked to act on the result.

The run2 behavior is what I would call 'fire and forget' as the longRunningProcess2 is run in the background (by its own scheduling), the item result is meaningless in that it has nothing to do with the running process, and downstream elements move forward with the meaningless result in parallel.

Code:

package _01_basics;

import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.infrastructure.Infrastructure;

import java.util.concurrent.TimeUnit;

public class _09a_Uni_From_CompletionStage {

    static long start = System.currentTimeMillis();
    public static void main(String[] args) throws Exception {
        log("⚡️ Uni from CompletionStage, fire and forget?");
        run2();
        boolean finished = Infrastructure.getDefaultWorkerPool().awaitTermination(5, TimeUnit.SECONDS);
        log("⚡️ awaitTermination called, "+finished);
        System.exit(0);
    }
    static void run1() throws Exception {
        var cs = longRunningProcess()
                .invoke(() -> log("After longRunningProcess item"))
                .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
                .subscribeAsCompletionStage();
        // longRunningProcess Uni should be running in background for 5s
        log("run1 begin sleep, longRunningProcess.isDone: "+cs.isDone());
        Thread.sleep(2000);
        // Should still be running, cs.isDone() == false
        log("run1 end sleep, longRunningProcess.isDone: "+cs.isDone());
        // Wait for CS to finish
        while (!cs.isDone()) {
            Thread.sleep(1000);
        }
        // Print Uni result
        log("run1 done waiting, longRunningProcess.result: "+cs.get());
    }

    static void run2() throws Exception {
        var cs = longRunningProcess2()
                .invoke(() -> log("After longRunningProcess item"))
                .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
                .subscribeAsCompletionStage();
        // longRunningProcess Uni should be running in background for 5s
        log("run2 begin sleep, longRunningProcess2.isDone: "+cs.isDone());
        Thread.sleep(2000);
        // Should still be running, but now cs.isDone() == true
        log("run2 end sleep, longRunningProcess2.isDone: "+cs.isDone());
        if (!cs.isDone()) {
            log("run2, assertion failure, subscription should have completed");
        }
        // Print Uni result
        log("run2 done waiting, longRunningProcess2.result: "+cs.get());
    }

    static Uni<String> longRunningProcess() {
        return Uni.createFrom()
        .item(() -> {
            log("begin longRunningProcess");
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
            }
            log("end longRunningProcess");
            return "Finished.longRunningProcess";
        });
    }

    static Uni<String> longRunningProcess2() {
        Uni<String> result = Uni.createFrom()
                .item("Work not part of Uni chain");

        Infrastructure.getDefaultWorkerPool().submit(() -> {
            log("begin longRunningProcess2");
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
            }
            log("end longRunningProcess2");
        });
        return  result;
    }

    static void log(String message) {
        long timestamp = System.currentTimeMillis();
        long elapsed = timestamp - start;
        System.out.printf("[%6d](%s): %s\n", elapsed, Thread.currentThread().getName(), message);
    }

}