completion_stage

Push-based futures for Rust similar to Java's CompletionStage.

Simple Example

use std::thread;
use std::time::Duration;
use completion_stage::CompletionStage;

fn main() {
    let result : String = CompletionStage::new_async::<thread::Thread>(|| {
        // Executed in a new virgin thread via thread::spawn, 
        // you can provide your own executor via the generic instead of 'thread::Thread' 
        // by implementing a simple trait for your executor.
        //
        // Do some background task here
        thread::sleep(Duration::from_secs(5));
        // 
        // eventually return the result.
        return 12345;
    }).and_then_apply(|intermediate| {
        // Executed in the same thread as above, 
        // or the main thread if the thread above is already finished, 
        // which is unlikely for this example
        return format!("The result is {intermediate}");
    }).unwrap();

    println!("{}", result);
}

Motivation

In the rust language 'futures' are poll-based and require use of 'async/await' which hides the callback-based execution flow from the programmer as well as requiring complex execution engines and frameworks that are unsuitable for a lot of situations.

This crate aims to provide futures that require no async execution engine, nor do they attempt to hide callback-based control flow from you.

What does push-based mean?

This example here illustrates what is meant by push-based, functionally it's completely identical to the example above.

use std::thread;
use std::time::Duration;
use completion_stage::CompletionStage;

fn main() {
    let first_stage : CompletionStage<i32> = CompletionStage::new();
    {
        //The CompletionStage is reference counted so it can be sent/shared with any thread.
        let first_stage = first_stage.clone();
        thread::spawn(move || {
            thread::sleep(Duration::from_secs(5));

            //'Push' the value into the future and execute all further child stages right here!
            first_stage.complete_with_value(12345i32)
        });
    }

    let second_stage : CompletionStage<String> = first_stage.and_then_apply(|intermediate : i32| {
        // Executed in the same thread as above, 
        // or the main thread if the thread above is already finished, 
        // which is unlikely for this example
        return format!("The result is {intermediate}");
    });

    println!("{}", second_stage.unwrap());
}

Panics

This crate does not catch any panic using panic::catch_unwind. If any user code panic's then a "drop guard" ensures that all dependent stages will be completed and in turn have their user code invoked during unwinding with Completion::Panic. This means that if user code panics again during this unwinding, then the application will abort as it normally does on a panic during unwind.

The panic value itself is not caught and therefore not available to dependent stages, as rust panic's should not be used in lieu of exceptions.

If you compile your code with panic=abort then none of this matters to you.

Example of panic

use std::thread;
use std::time::Duration;
use completion_stage::{Completion, CompletionStage};

fn some_func() {
    let stage1 = CompletionStage::new();
    let stage1_ref = stage1.clone();
    let join_handle = thread::spawn(move || {
        thread::sleep(Duration::from_secs(1));
        stage1_ref.complete_with_value(123); //This function will panic
    });

    let stage2 = stage1.and_then_apply(|value| {
        if value < 1000 {
            //This if is always true for this example.
            panic!("oops, number too small"); //Panic from here
        }
        value.to_string()
    });

    let stage3: CompletionStage<()> = stage2.and_then_apply(|_value| {
        std::process::abort(); //We never get here, this never gets called.
    });

    //So the join handle will hold the panic.
    assert!(join_handle.join().is_err());
    // Stage1 completed normally, but had its value taken into the closure where it was dropped.
    assert_eq!(stage1.get(), Completion::Taken);
    // Stage2 is completed as panicked
    assert_eq!(stage2.get(), Completion::Panic);
    // Stage3 will be completed as panicked because its parent stage, stage2 panicked.
    assert_eq!(stage3.get(), Completion::Panic);
    println!("Done!");
}

Deadlocks

The implementation automatically detects simple deadlocks. All deadlocks that are caused by a thread waiting on itself are detected. For example, borrowing the value of a stage while trying to take the value from the stage is detected. This generally causes the taking operation to fail. Some taking operation's panic in this case, which is made explicit in their documentation.

Example where Deadlock is detected

use completion_stage::{CompletionStage, Completion};

fn some_func() {
    let stage = CompletionStage::new_completed_value(123);
    let guard = stage.borrow().unwrap();
    let value = stage.get();
    assert_eq!(*guard, 123);
    assert_eq!(value, Completion::DeadLock);
    drop(guard);
    
    //Now we no longer deadlock, because we no longer borrow the value.
    let value = stage.get();
    assert_eq!(value, Completion::Value(123));
}

Use-Case

I intend to use this crate to "return" results to an opengl ui-thread. The opengl ui "thread" will start background tasks when, for example, a "button" is pressed but should not be blocked (as that would freeze the UI).

Currently, I have been using a lot of janky code that uses mpsc Channels and try_recv for this purpose, but I intend to replace all such code with this crate.