Android Bolts - Easier thread scheduling and task management

Usain St Leo Bolt is a Jamaican sprinter, world record holder for the men's 100m, men's 200m and men's 400m relay, and winner of three consecutive Olympic gold medals in the above three events.

[[210857]]

Bolts can be used to split a complete operation into multiple subtasks. These subtasks can be freely split, combined and replaced. Each task can run in a specified thread as a link in the entire task chain. At the same time, it can obtain task results from upstream tasks and publish the results of the current task to downstream tasks without considering the interaction between threads.

Bolts-Android Bolts implementation in Android Bolts-ObjC Bolts implementation in OC Bolts-Swift Bolts implementation in Swift

Preface

A simple requirement for thread scheduling. The child thread downloads a picture from the network and returns the downloaded picture. The main thread uses the picture to update the UI and returns the current UI status json. The child thread saves the json data to a local file and pops up a prompt on the main thread after completion. This involves 4 thread switches, and the subsequent tasks require the return value of the previous task as a parameter.

Using Thread + Handler implementation, thread scheduling is very inflexible, the code is not readable, not beautiful, has poor scalability, and error handling is very troublesome.

 String url = "http://www.baidu.com" ;
 Handler handler = new Handler(Looper.getMainLooper());
 new Thread(() -> {
    // download
    Bitmap bitmap = downloadBitmap(url);
 handler.post(() -> {
 // Update the UI
        String json = updateUI(bitmap);
 new Thread(() -> {
 // Write UI state to storage
            saveUIState(json);
 // After saving successfully, prompt
            handler.post(() -> toastMsg( "save finish." ));
 }).start();
 });
 }).start();

Using RxJava, thread scheduling is very flexible, chain calling, clear code, good scalability, and a unified exception handling mechanism. However, Rx is a very powerful library. If it is only used for thread scheduling, Rx is a bit too heavy.

 Observable.just(URL)
        // download
        .map(this::downloadBitmap)
        .subscribeOn(Schedulers.newThread())
 // Update the UI
        .observeOn(AndroidSchedulers.mainThread())
 .map(this::updateUI)
 // Store UI state
        .observeOn(Schedulers.io())
        .map(this::saveUIState)
 // Display prompt
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(rst -> toastMsg( "save to " + rst),
                //handle error
                Throwable::printStackTrace);

It is implemented using bolts, with flexible thread scheduling, chain calls, clear code, good scalability, and a unified exception handling mechanism. Although it does not have as many operators as Rx, it wins in that the class library is very, very small, only 38 KB.

 Task
 .forResult(URL)
        // download
        .onSuccess(task -> downloadBitmap(task.getResult()), Task.BACKGROUND_EXECUTOR)
 // Update the UI
        .onSuccess(task -> updateUI(task.getResult()), Task.UI_THREAD_EXECUTOR)
 // Store UI state
        .onSuccess(task -> saveUIState(task.getResult()), Task.BACKGROUND_EXECUTOR)
        // hint
        .onSuccess(task -> toastMsg( "save to " + task.getResult()), Task.UI_THREAD_EXECUT
 // handle error
        .continueWith(task -> {
            if (task.isFaulted()) {
                task.getError().printStackTrace();
 return   false ;
 }
 return   true ;
 });

Thread Scheduler

There are 4 types of execution threads, which distribute tasks to specified threads for execution, namely

1. backgroud - Background thread pool that can execute tasks concurrently.
2. scheduled - Single thread pool, with only one thread, mainly used to perform delay operations.
3. immediate - immediate thread. If the thread call stack is less than 15, it is executed in the current thread, otherwise it is delegated to the background.
4. uiThread - Designed for Android, uses Handler to send to the main thread for execution.

background

Mainly used to perform multiple tasks concurrently in the background

 public   static final ExecutorService BACKGROUND_EXECUTOR = BoltsExecutors.background();

On the Android platform, a thread pool is created based on the number of CPU cores. In other cases, a cache thread pool is created.

 background = !isAndroidRuntime()
    ? java.util.concurrent.Executors.newCachedThreadPool()
    : AndroidExecutors.newCachedThreadPool();

scheduled

It is mainly used to delay operations between tasks and does not actually execute tasks.

 scheduled = Executors.newSingleThreadScheduledExecutor();

immediate

It is mainly used to simplify methods that do not specify the running thread. By default, the task is executed in the current thread. ThreadLocal is used to save the depth of each thread call stack. If the depth does not exceed 15, it is executed in the current thread, otherwise it is delegated to the background for execution.

 private static final Executor IMMEDIATE_EXECUTOR = BoltsExecutors.immediate(); 
 
 //Key methods
 @Override
 public void execute (Runnable command) {
 int depth = incrementDepth();
 try {
 if (depth <= MAX_DEPTH) {
 command.run();
 } else {
      BoltsExecutors.background(). execute (command)
 }
 finally
 decrementDepth();
 }
 }

uiThread

Designed specifically for Android, it executes tasks on the main thread.

 public   static final Executor UI_THREAD_EXECUTOR = AndroidExecutors.uiThread();

 private static class UIThreadExecutor implements Executor {
 @Override
 public void execute (Runnable command) {
    new Handler(Looper.getMainLooper()).post(command);
 }
 }

Core Classes

Task is the core class. Each subtask is a Task, which is responsible for the tasks it needs to perform. Each Task has three states: Result, Error and Cancel, representing success, exception and cancellation respectively.

Continuation is an interface that acts like a lock that links each link of the subtask, linking independent tasks together.

A complete task chain is formed in the form of Task - Continuation - Task - Continuation ... and executed sequentially in each thread.

Create Task

According to the three states of Task, create a simple Task and reuse the existing task object

 public   static <TResult> Task<TResult> forResult(TResult value)  
 public   static <TResult> Task<TResult> forError(Exception error)  
 public   static <TResult> Task<TResult> canceled()

Use the delay method to delay execution and create a Task

 public   static Task delay(long delay)  
 public   static Task delay(long delay, CancellationToken cancellationToken)

Use the whenAny method to execute multiple tasks and save the result when any task returns a result

 public   static Task> whenAnyResult(Collection> tasks)  
 public   static Task> whenAny(Collection> tasks)

Use the whenAll method to execute multiple tasks and return the results when all tasks are completed.

 public   static Task whenAll(Collection> tasks)  
 public   static Task> whenAllResult(final Collection> tasks)

Use the call method to execute a task and create a Task at the same time

 public   static Task call(final Callable callable, Executor executor,  
 final CancellationToken ct)

Linking subtasks

Use the continueWith method to link a subtask. If the previous task has been completed, the current task will be executed immediately. Otherwise, it will be added to the queue and wait.

 public <TContinuationResult> Task<TContinuationResult> continueWith(
      final Continuation<TResult, TContinuationResult> continuation, final Executor executor,
      final CancellationToken ct)

Use the continueWithTask method to link another task chain after the current task. This approach is to meet the scenario of combining some tasks together and separating them as public tasks. It accepts another completely independent task chain to be appended after the currently executed task.

 public <TContinuationResult> Task<TContinuationResult> continueWithTask(
      final Continuation<TResult, Task<TContinuationResult>> continuation, final Executor executor,
      final CancellationToken ct)

Use the continueWhile method to chain subtasks. The difference from continueWith is that it has a predicate expression. Only when the expression is true will the subtask be appended. This allows an interception operation to be performed before executing the task, and also to avoid destroying the overall style of chain calls.

 public Task<Void> continueWhile(final Callable<Boolean> predicate,
      final Continuation<Void, Task<Void>> continuation, final Executor executor,
      final CancellationToken ct)

Use onSuccess and onSuccessTask to link a single task to a task chain. The difference from continueWith is that in the onSuccess method, if the previous task fails, the subsequent task will also fail directly and will not be executed again. However, there is no connection between the subtasks of continueWith. Even if the previous task fails, the subsequent task will be executed.

 public <TContinuationResult> Task<TContinuationResult> onSuccess(
      final Continuation<TResult, TContinuationResult> continuation, Executor executor,
      final CancellationToken ct)

Cancel a task

 CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
 CancellationToken token = cancellationTokenSource.getToken();
 Task.call((Callable<String>) () -> null ,
        Task.BACKGROUND_EXECUTOR,
 token);
 // Cancel the task
 cancellationTokenSource.cancel();

Exception handling

Regarding exception handling, in the entire mechanism, each task is treated as an independent unit, and exceptions will be captured uniformly, so there is no need to handle the methods in the task separately.

If continueWith is used to link tasks, the exception information of the current task will be saved in the current Task and processed in the downstream task. The downstream task can also not handle the exception and directly execute the task. In this case, the exception stops here and will not be passed down. In other words, only the downstream task knows the result of the current task, whether it is success or exception.

Of course, if there is a relationship between tasks, since the abnormality of the upstream task is very likely to cause the abnormality of the current task, the information of the abnormality of the current task will be passed downward, but the abnormality of the upstream task will stop here.

If you use methods such as onSuccess, if the upstream task is abnormal, the downstream task will not be executed at all, but the exception will be passed directly downward until it is handled.

Separation and combination of tasks

We can break down a complete operation into multiple tasks. Each task follows the principle of single responsibility and is as simple as possible. In this way, new tasks can be inserted between tasks, or some tasks can be separated and combined together.

Scalability

We can add a new operation between the two subdivided tasks without affecting the upstream and downstream tasks, such as saving the Bitmap locally before updating the UI in the requirements at the beginning of the article.

 Task
 .forResult(URL)
        // download
        .onSuccess(task -> downloadBitmap(task.getResult()), Task.BACKGROUND_EXECUTOR)
 // Save locally
        .onSuccess(task -> saveBitmapToFile(task.getResult()),Task.BACKGROUND_EXECUTOR)
 // Update the UI
        .onSuccess(task -> updateUI(task.getResult()), Task.UI_THREAD_EXECUTOR)
 ...

Reusability

Some common operations can be separated into new tasks. When similar operations are needed, these functions can be reused. For example, the two functions of downloading pictures and updating the UI and saving status and popping up prompts can be separated as common tasks.

 // Download image -> Update UI
 public Continuation<String, Task<String>> downloadImageAndUpdateUI() {
 return task ->
            Task.call(() -> downloadBitmap(task.getResult()), Task.BACKGROUND_EXECUTOR)
                    .continueWith(taskWithBitmap -> updateUI(taskWithBitmap.getResult()), Task.UI_THREAD_EXECUTOR);
 } 
 
 // Save status->prompt information
 public Continuation<String, Task<Boolean>> saveStateAndToast() {
 return task ->
            Task.call(() -> saveUIState(task.getResult()), Task.BACKGROUND_EXECUTOR)
                    .continueWith(taskWithPath -> toastMsg( "save to " + taskWithPath.getResult()));
 }

Using separate tasks

 Task
 .forResult(URL)
        .continueWithTask(downloadImageAndUpdateUI())
        .continueWithTask(saveStateAndToast())
 ...

Summarize

In Task, there is a continuation, which is a list of tasks appended to the current task. When the current task succeeds, fails, or is canceled, the subsequent tasks in the list will be executed.

Normally, we use chain calls to build a task chain, and the result is a task chain without branches.

When adding tasks: Each time a Continuation is added, a Task is generated and added to the continuations list of the upstream task, waiting to be executed, and the current Task is returned so that the subsequent tasks can be linked to the current task.

When executing tasks: After the current task is executed, there may be 3 kinds of results, all of which will be saved in the current Task, and then the subsequent tasks in the continuations list will be checked, and the current Task will be passed as a parameter to the subsequent linked tasks to let the subsequent tasks know the results of the upstream tasks.