Concurrency in Android

As explained before, in order to achieve a scalable application in a multicore device environment, the Android developer should be capable of creating concurrent lines of execution that combine and aggregate data from multiple resources.

The Android SDK, as it is based on a subset of Java SDK, derived from the Apache Harmony project, provides access to low-level concurrency constructs such as java.lang.Thread, java.lang.Runnable, and the synchronized and volatile keywords.

These constructs are the most basic building blocks to achieve concurrency and parallelism, and all the high-level asynchronous constructs are created around these building blocks.

The most basic one, java.lang.Thread, is the class that is mostly used and is the construct that creates a new independent line of execution in a Java program:

public class MyThread extends Thread {
    public void run() {
        Log.d("Generic", "My Android Thread is running ...");
    }
}

In the preceding code, we subclassed java.lang.Thread to create our own independent line of execution. When Thread is started, the run method will be called automatically and it will print the message on the Android log:

MyThread myThread = new MyThread();
myTread.start();

At this time, we will create an instance of our MyThread, and when we start it in the second line, the system creates a thread inside the process and executes the run() method.

Other helpful thread-related methods include the following:

• Thread.currentThread(): This retrieves the current running instance of the thread
• Thread.sleep(time): This pauses the current thread from execution for the given period of time
• Thread.getName() and Thread.getId(): These get the name and TID, respectively so that they can be useful for debugging purposes
• Thread.isAlive(): This checks whether the thread is currently running or it has already finished its job
• Thread.join(): This blocks the current thread and waits until the accessed thread finishes its execution or dies

The Runnable interface, which is another building block that comes from the Java API, is an interface defined to specify and encapsulate code that is intended to be executed by a Java thread instance or any other class that handles this Runnable:

package java.lang;

public interface Runnable {   
    public abstract void run();
}

In the following code, we basically created the Runnable subclass so that it implements the run() method and can be passed and executed by a thread:

public class MyRunnable implements Runnable {

    public void run(){
        Log.d("Generic","Running in the Thread " +
                        Thread.currentThread().getId());
	// Do your work here
	...
    }
}

Now our Runnable subclass can be passed to Thread and is executed independently in the concurrent line of execution:

Thread thread = new Thread(new MyRunnable());
thread.start();

While starting new threads is easy, concurrency is actually a very difficult thing to do. Concurrent software faces many issues that fall into two broad categories: correctness (producing consistent and correct results) and liveness (making progress towards completion). Thread creation could also cause some performance overhead, and too many threads can reduce the performance, as the OS will have switch between these lines of execution.

Correctness issues in concurrent programs

A common example of a correctness problem occurs when two threads need to modify the value of the same variable based on its current value. Let's consider that we have a myInt integer variable with the current value of 2.

In order to increment myInt, we first need to read its current value and then add 1 to it. In a single-threaded world, the two increments would happen in a strict sequence—we will read the initial value 2, add 1 to it, set the new value back to the variable, and then repeat the sequence. After the two increments, myInt holds the value 4.

In a multithreaded environment, we will run into potential timing issues. It is possible that two threads trying to increment the variable would both read the same initial value 2, add 1 to it, and set the result (in both cases, 3) back to the variable:

int myInt = 2;
...
public class MyThread extends Thread {

    public void run() {
         super.run();
         myInt++;
   }
}
...
Thread t1 = new MyThread();
Thread t2 = new MyThread();
t1.start();
t2.start();

Both threads behaved correctly in their localized view of the world, but in terms of the overall program, we will clearly have a correctness problem; 2 + 2 should not equal 3! This kind of timing issue is known as a race condition.

A common solution to correctness problems, such as race conditions, is mutual exclusion—preventing multiple threads from accessing certain resources at the same time. Typically, this is achieved by ensuring that threads acquire an exclusive lock before reading or updating shared data.

To achieve this correctness, we can make use of the synchronized construct to solve the correctness issue on the following piece of code:

Object lock = new Object();
public class MyThread extends Thread {
    public void run() {
        super.run();
        synchronized(lock) {
            myInt++;
        }
    }
}

In the preceding code, we used the intrinsic lock available in each Java object to create a mutually exclusive scope of code that will enforce that the increment sentence will work properly and will not suffer from correctness issues as explained previously. When one of the threads gets access to the protected scope, it is said that the thread acquired the lock, and after the thread gets out of the protected scope, it releases the lock that could be acquired by another thread.

Another way to create mutually exclusive scopes is to create a method with a synchronized method:

int myInt = 2;
synchronized void increment(){
    myInt++;
}
...
public class IncrementThread extends Thread {
    public void run() {
        super.run();
        increment();
    }
}

The synchronized method will use the object-intrinsic lock, where myInt is defined to create a mutually exclusive zone so IncrementThread, incrementing myInt through the increment(), will prevent any thread interference and memory consistency errors.

Liveness issues in concurrent programs

Liveness can be thought of as the ability of the application to do useful work and make progress towards goals. Liveness problems tend to be an unfortunate side effect of the solution to the correctness problems.

Both properties should be achieved in a proper concurrent program, notwithstanding the correctness is concerned with making progress in a program preventing a deadlock, livelock, or starvation from happening, and the correctness is concerned with making consistent and correct results.

Note

Deadlock is a situation where two or more threads are unable to proceed because each is waiting for the others to do something. Livelock is a situation where two or more threads continuously change their states in response to the changes in the other threads without doing any useful work.

By locking access to data or system resources, it is possible to create bottlenecks where many threads are contending to access a single lock, leading to potentially significant delays.

Worse, where multiple locks are used, it is possible to create a situation where no thread can make progress because each requires exclusive access to a lock that another thread currently owns—a situation known as a deadlock.

Thread coordination

Thread coordination is an important topic in concurrent programming, especially when we want to perform the following tasks:

• Synchronize access of threads to shared resources or shared memory:
  • Shared database, files, system services, instance/class variables, or queues
• Coordinate work and execution within a group of threads:
  • Parallel execution, pipeline executions, inter-dependent tasks, and so on

When we want to coordinate thread efforts to achieve a goal, we should try to avoid waiting or polling mechanisms that keep the CPU busy while we wait for an event in another thread.

The following example shows us a small loop where we will continuously occupy the CPU while we wait for a certain state change to happen:

while(!readyToProcess) {
  // do nothing .. busy waiting wastes processor time.
}

To overcome the coordination issue, and to implement our own constructs, we should use some low-level signals or messaging mechanisms to communicate between threads and coordinate the interaction.

In Java, every object has the wait(), notify(), and notifyAll()methods that provide low-level mechanisms to send thread signals between a group of threads and put a thread in a waiting state until a condition is met.

This mechanism, also known as monitor or guard, is a design pattern commonly used in another languages and it ensures that only one thread can enter a given section of code at any given time with an ability to wait until a condition happens.

This design pattern, in comparison with our previous example, delivers a better and efficient CPU-cycle management while waiting for any particular situation to happen on another thread, and is generally used in situations where we need to coordinate work between different lines of execution.

In the following code example, we are going to explain how to use this construct to create a basic multithreaded Logger with 10 threads that will wait in the monitor section until a message is pushed (condition) by any other thread in the application.

The Logger, which is responsible for logging on to the output, has a queue with a maximum of 20 positions to store the new logging text messages:

public class Logger {
    LinkedList<String> queue = new LinkedList<String>();
    private final int MAX_QUEUE_SIZE = 20;
    private final int MAX_THREAD_COUNT = 10;

In the next code, we will create a Runnable unit of work that runs indefinitely and retrieves a message from the queue to print the message on the Android log.

After that, we will create and start 10 threads that are going to execute the Runnable unit of work task:

public void start() {
    // Creates the Loop as a Runnable
    Runnable task = new Runnable() {
        @Override
        public void run() {
            while(true) {
                String message = pullMessage();
                Log.d(Thread.currentThread().
                         getName(),message);
		     // Do another processing
             }
         }
     };
    // Create a Group of Threads for processing
    for(int i=0; i< MAX_THREAD_COUNT; i++){
         new Thread(task).start();
    }
 }

The pullMessage(), which is a synchorized method, runs a mutual exclusion and puts the thread in the waiting state when it reaches the wait() method. All the created threads will stay in this state until another thread calls notifyAll():

// Pulls a message from the queue
// Only returns when a new message is retrieves
// from the queue.
private synchronized String pullMessage(){
    while (queue.isEmpty()) {
        try {
            wait();
        } catch (InterruptedException e) { ... }
    }
    return queue.pop();
}
// Push a new message to the tail of the queue if
// the queue has available positions
public synchronized void pushMessage(String logMsg) {
    if ( queue.size()< MAX_QUEUE_SIZE ) {
        queue.push(logMsg);      
        notifyAll();
    }
}

When any thread is in the waiting state, it releases the lock temporarily and gives a chance to another thread to enter the mutual exclusion to push new messages or enter into the wait state.

In the following snippet, we will first create the Logger instance and then we will call the start method to start the working threads and we will push 10 messages into a queue of work to be processed.

When the pushMessage()method is invoked, a new logging message is inserted at the end of the queue and notifiyAll() is invoked to notify all the available threads.

As the pullMessage() method runs in a mutual-exclusion (synchronized) zone, only one thread will wake up and return from the pull method. Once pullMessage() returns, the logging message is printed:

Logger logger =new Logger();
logger.start();
for ( int i=0; i< 10 ; i++) {
    ...
    logger.pushMessage(date+" : "+"Log Message #"+i);
}

In the following console output, we have an example of the output that this code will generate and the logging messages are processed by any available threads in an ordered manner:

D/Thread-108(23915): <Date>: Log Message #0
D/Thread-109(23915): ...: Log Message #1
D/Thread-110(23915): ...: Log Message #2
D/Thread-111(23915): ...: Log Message #3

This kind of low-level construct can also be used to control shared resources (polling) to manage background execution (parallelism) and control thread pools.

Concurrent package constructs

Other Java concurrent constructs provided by java.util.concurrent, which are also available on Android SDK are as follows:

• Lock objects (java.util.concurrent): They implement locking behaviors with a higher level idiom.
• Executors: These are high-level APIs to launch and manage a group of thread executions (ThreadPool, and so on).
• Concurrent collections: These are the collections where the methods that change the collection are protected from synchronization issues.
• Synchronizers: These are high-level constructs that coordinate and control thread execution (Semaphore, Cyclic Barrier, and so on).
• Atomic variables (java.util.concurrent.atomic): These are classes that provide thread-safe operations on single variables. One example of it is AtomicInteger that could be used in our example to solve the correctness issue.

Some Android-specific constructs use these classes as basic building blocks to implement their concurrent behavior, although they could be used by a developer to build custom concurrent constructs to solve a specific use case.

Executor framework

The Executor framework is another framework available on java.util.concurrent that provides an interface to submit Runnable tasks, decoupling the task submission from the way the task will run:

public interface Executor {
  void execute(Runnable command);
}

Each Executor, which implements the interface that we defined earlier, can manage the asynchronous resources, such as thread creation destruction and caching, and task queueing in a variety of ways to achieve the perfect behavior to a specific use case.

The java.util.concurrent comes with a group of implementations available out of the box that cover most generic use cases, as follows:

• Executors.newCachedThreadPool(): This is a thread poll that could grow and reuse previously created threads
• Executors.newFixedThreadPool (nThreads): This is a thread pool with a fixed number of threads and a message queue for store work
• Executors.newSingleThreadPool(): This is similar to newFixedThreadPool, but with only one working thread

To run a task on Executor, the developer has to invoke execute() by passing Runnable as an argument:

public class MyRunnable implements Runnable {
    public void run() {
        Log.d("Generic", "Running From Thread " +
              Thread.currentThread().getId());   
	 // Your Long Running Computation Task
    }
}
public void startWorking(){
    Executor executor = Executors.newFixedThreadPool(5);
    for ( int i=0; i < 20; i++ ) {
        executor.execute(new MyRunnable());
    }
}

In the preceding code, we created ThreadPool over the factory methods with a fixed number of five threads ready to process work.

After the ExecutorService instance creation, new Runnable tasks are posted for asynchronous processing.

When a new unit of work is submitted, a thread that is free to work is chosen to handle the task; but when all the threads are occupied, Runnable will wait in a local queue until a thread is ready to work.