CON29-J. Do not execute interdependent tasks in a thread pool

There are certain problems associated with the incorrect use of the Executor interface. For instance, tasks that depend on other tasks should not execute in the same thread pool. A task that submits another task to a single threaded Executor remains blocked until the results are received whereas the second task may have dependencies on the first task. This constitutes a deadlock.

Noncompliant Code Example

This noncompliant code example shows a thread starvation deadlock. This situation not only occurs in single threaded Executors, but also in those with large Thread Pools. This can happen when all the threads executing in the pool are blocked on tasks that are waiting on the queue. A blocking operation within a subtask can also lead to unbounded queue growth. [[Goetz 06]]

// Field password is defined in class InitialHandshake
class NetworkServer extends InitialHandshake implements Runnable {
  private final ServerSocket serverSocket;
  private final ExecutorService pool;

  public NetworkServer(int port, int poolSize) throws IOException {
    serverSocket = new ServerSocket(port);
    pool = Executors.newFixedThreadPool(poolSize);
  }
 
  public void run() {
    try { 
      // Interdependent tasks
      pool.submit(new SanitizeInput(password));  // password is defined in class InitialHandshake 
      pool.submit(new CustomHandshake(password));  // for e.g. client puzzles 
      pool.execute(new Handle(serverSocket.accept()));  // Handle connection
    } catch (IOException ex) { 
      pool.shutdown();
    }	 
  }
}

In this noncompliant code example, the SanitizeInput task depends upon the CustomHandshake task for the value of password whereas the latter depends on the former to return a password that has been correctly sanitized.

Compliant Solution

This compliant solution recommends executing the interdependent tasks as a single task within the Executor. In other cases, where the subtasks do not require concurrency safeguards, the subtasks can be moved outside the threaded region that is required to be executed by the Executor.

class NetworkServer extends InitialHandshake implements Runnable {
  private final ServerSocket serverSocket;
  private final ExecutorService pool;

  public NetworkServer(int port, int poolSize) throws IOException {
    serverSocket = new ServerSocket(port);
    pool = Executors.newFixedThreadPool(poolSize);
  }
 
  public void run() {
    try {
      // Execute interdependent subtasks as a single combined task within this block
      // Tasks SanitizeInput() and CustomHandshake() are performed together in Handle()
      pool.execute(new Handle(serverSocket.accept())); // Handle connection
    } catch (IOException ex) { 
      pool.shutdown();
    }	 
  }
}

Always try to submit independent tasks to the Executor. Thread starvation issues can be mitigated by choosing a large pool size. Note that operations that have further constraints, such as the total number of database connections or total ResultSets open at a particular time, impose an upper bound on the thread pool size as each thread continues to block until the resource becomes available. The other rules of fair concurrency, such as not running time consuming tasks, also apply. When this is not possible, expecting to obtain real time result guarantees from the execution of tasks is conceivably, an unreasonable target.

Sometimes, a private static ThreadLocal variable is used per thread to maintain local state. When using thread pools, ThreadLocal variable should be used only if their lifetime is shorter than that of the corresponding task [[Goetz 06]]. Moreover, such variables should not be used as a communication mechanism between tasks.

Noncompliant Code Example

This noncompliant code example (based on [[Gafter 06]]) shows a BrowserManager class that has several methods that use a fork-join mechanism, that is, they start threads and wait for them to finish. The methods are called in the sequence perUser(), perProfile and perTab(). The method methodInvoker spawns several instances of the specified runnable depending on the value of the variable numberOfTimes. A fixed sized thread pool is used to execute the enumerated threads generated at different levels.

public class BrowserManager {
  private final ExecutorService pool = Executors.newFixedThreadPool(10);
  int numberOfTimes;
  private static volatile int count = 0;
  
  public BrowserManager(int n) {
    this.numberOfTimes = n;
  }

  public void perUser() {	
    methodInvoker(numberOfTimes, "perProfile"); 
    pool.shutdown();
  }

  public void perProfile() {
    methodInvoker(numberOfTimes, "perTab");	 
  }

  public void perTab() {	
    methodInvoker(numberOfTimes, "doSomething");
  }

  public void doSomething() {
    System.out.println(++count);
  }

  public void methodInvoker(int n, final String method) {
    final BrowserManager fm = this;
    Runnable run = new Runnable() {
      public void run() {
	try {
	  Method meth = fm.getClass().getMethod(method);
	  meth.invoke(fm);			
        } catch (Throwable t) {
	  // Forward to exception reporter
        }		  
      }
    };	
   
    Collection<Callable<Object>> c = new ArrayList<Callable<Object>>();
    for(int i=0;i < n; i++) {
      c.add(Executors.callable(run));
    }
	
    Collection<Future<Object>> futures = null;
    try {
      futures = pool.invokeAll(c);
    } catch (InterruptedException e) {     
      // Forward to handler	
    }
    // ... 
  }

  public static void main(String[] args) {
    BrowserManager manager = new BrowserManager(5);
    manager.perUser();
  }	
}

Contrary to what is expected, this program does not print the total count, that is, the number of times doSomething() is invoked. This is because it is susceptible to a thread starvation deadlock because the size of the thread pool (10) does not allow either thread from perTab() to invoke the doSomething() method. The output of the program varies for different values of numberOfTimes and the thread pool size. Note that different threads are allowed to invoke doSomething() in different order; we are concerned only with the maximum value of count to determine how many times the method executed.

Compliant Solution

This compliant solution selects tasks for scheduling and avoids the thread starvation deadlock. Every level (worker) must have a double ended queue where all sub-tasks are queued. Each level removes the most recently generated sub-task from the queue so that it can process it. When there are no more threads left to process, the current level runs the least-recently created sub-task of another level by picking and removing it from that level's queue.

This compliant solution sets the CallerRuns policy on a ThreadPoolExecutor, and uses a synchronous queue [[Gafter 06]].

public class BrowserManager {
  static ThreadPoolExecutor pool =
    new ThreadPoolExecutor(0, 10, 60L, TimeUnit.SECONDS,
                          new SynchronousQueue<Runnable>());
  int numberOfTimes;
  private static volatile int count = 0;

  static {
    pool.setRejectedExecutionHandler(
    new ThreadPoolExecutor.CallerRunsPolicy());
  }

  // ... 
}	

According to the Java API, class java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy documentation:

A handler for rejected tasks that runs the rejected task directly in the calling thread of the execute method, unless the executor has been shut down, in which case the task is discarded.

This compliant solution is subject to the vagaries of the thread scheduler which may not optimally schedule the tasks, however, it avoids the thread starvation deadlock.

Risk Assessment

Executing interdependent tasks in a thread pool can lead to denial of service.

Automated Detection

TODO

Related Vulnerabilities

Search for vulnerabilities resulting from the violation of this rule on the CERT website.

References

[[API 06]]
[[Gafter 06]] A Thread Pool Puzzler

FIO36-J. Do not create multiple buffered wrappers on an InputStream      09. Input Output (FIO)      09. Input Output (FIO)