Struct async_tasks_recorder::AsyncTasksRecoder

pub struct AsyncTasksRecoder<T>
where
    T: Eq + Hash + Clone + Send + 'static,
{ /* private fields */ }

Arc was used internally, so after clone, the same TaskManager was used, which means you can share AsyncTasksRecoder by clone.

Usage

Launch a task with a unique task_id and a Future by launch.

Query the state of the task with its task_id by query_task_state or query_task_state_quick.

Further Propositions & Proofs

P01

A task (or tasks with the same task_id) wouldn’t be executed again after first success.

When a task fail, it wouldn’t break anything. Failure just means the task could be launched again, so if this proposition (P01) is true, there is almost nothing to worry about. For further discussion, please refer to P02.

From now on, only consider the situation of success.

working_tasks play the role of lock, which allow tasks with the same task_id to execute remaining codes (after insert & before remove) only once. And before remove from working_tasks, the succeeded task_id has been in success_tasks.

An equivalent pseudocode can be obtained.

• working_tasks become a mutex (maybe a spin lock) for one task_id.
• success_tasks become an atomic boolean, which can only change from false to true.
• An execution of a task becomes adding on an atomic int (count).

Therefore, if the count is never greater than 1, it means that the task will only be called once.

let working_task_id = mutex::new();
let success_task_id = atomic(false);
let count = atomic(0);
launch_multi_thread {
    working_task_id.lock();
    if success_task_id.get() {
        exit();
    }
    count.add(1);
    success_task_id.set(true);
    working_task_id.unlock();
}
assert_eq!(count.get(), 1);

Obviously, success_tasks.set(true) can only be executed once. After that, exit() is always called. This results in count.add(1) being called only once, too. Q.E.D.

P02

Task failure is not harmful for recorder.

Considering the situation of failure, the pseudocode becomes like this:

let working_task_id = mutex::new();
let success_task_id = atomic(false);
let failed_task_id = atomic(false); // Initially not in `failed_tasks`, but not important
let count = atomic(0);
launch_multi_thread {
    working_task_id.lock();
    if success_task_id.get() {
        exit();
    }
    // Here should be `real_working`
    failed_task_id.set(false); // So it shouldn't be `Failed`, just remove from `failed_tasks`
    count.add(1);
    if real_success {
        success_task_id.set(true);
    } else {
        // `real_failed`
        failed_task_id.set(true); // become `Failed`
    }
    working_task_id.unlock();
}
assert_eq!(count.get(), 1);

In a launch (critical section by working_tasks), the initial value of failed is ignored. Therefore, it’s not important whether failed_tasks changes are atomic for launches.

From the perspective of query_task_state, failed_tasks is only meaningful when task_id is in it.

task_id is in failed_tasks only when it become real_failed and before redo (next real_working). Very good.

P03

No state would turn back to Not found.

From the pseudocode in P02:

// entry `working_task_id`
if real_success {
    success_task_id.set(true);
} else {
    // `real_failed`
    failed_task_id.set(true); // Become `Failed`
}
// leave `working_task_id`

It can be found that as long as the task has entered working_tasks once, when exiting working_tasks, the task must already be in one of the failed_tasks or success_tasks options.

So after first Working, the task must be in one of tasks, then it won’t be Not found again. Q.E.D.

Implementations

impl<T> AsyncTasksRecoder<T>

where T: Eq + Hash + Clone + Send + Sync + 'static,

pub fn new() -> Self

Create a completely new AsyncTasksRecoder.

pub fn new_with_task_manager(task_manager: TaskManager<T>) -> Self

Create by TaskManager

pub fn new_with_task_manager_arc(task_manager: Arc<TaskManager<T>>) -> Self

Create by Arc of TaskManager

pub async fn launch<Fut, R, E>(&self, task_id: T, task: Fut)

where Fut: Future<Output = Result<R, E>> + Send + 'static, R: Send, E: Send,

Launch task that returns Result.

The return value of task is ignored, so please use other methods to handle the return value, such as channel or shared variable.

• task_id: Uniquely mark a task. Different Future with the same task_id is considered the same task.
• task: A Future to be executed automatically.

pub async fn query_task_state(&self, task_id: &T) -> TaskState

Query the state of a task by task_id.

Query priority of containers : success_tasks -> failed_tasks -> working_tasks.

NOTE: working_tasks usually has more contention.

If not found in all tasks, be NotFound. Only occurs before the launch or in a very short period of time after the first launch.

Note, if T is String, then parameter task_id would be &String instead of &str.

pub async fn query_task_state_quick(&self, task_id: &T) -> TaskState

Return Working if not in either success_tasks or failed_tasks.

No query in working_tasks, so less contention.

Even when the task_id’s launch have not occurred, return Working.

Use this if you are certain that the task’s launch must occur at some point in the past or future, and don’t care about when the launch occurs (because first launch always turns into Working at some point).

pub fn get_task_manager_arc(&self) -> Arc<TaskManager<T>>

Get a cloned Arc of task_manager. Then you can do anything you want (Every containers are public).

pub fn get_success_tasks_ref(&self) -> &HashSet<T>

Get a reference of success_tasks.

pub fn get_working_tasks_ref(&self) -> &HashSet<T>

Get a reference of working_tasks.

pub fn get_failed_tasks_ref(&self) -> &HashSet<T>

Get a reference of failed_tasks.

Trait Implementations

impl<T> Clone for AsyncTasksRecoder<T>

where T: Eq + Hash + Clone + Send + 'static + Clone,

fn clone(&self) -> AsyncTasksRecoder<T>

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<T> Debug for AsyncTasksRecoder<T>

where T: Eq + Hash + Clone + Send + 'static + Debug,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<T> Default for AsyncTasksRecoder<T>

where T: Eq + Hash + Clone + Send + 'static + Default,

fn default() -> AsyncTasksRecoder<T>

Returns the “default value” for a type.