6
\$\begingroup\$

I occasionally find myself needing to run code on some arbitrary thread or pass a Runnable to some arbitrary consumer, and missing the full functionality of java.util.concurrent -- for instance, writing Swing and JavaFX applications, posting UI updates from a background thread to the main UI thread with SwingUtilities.invokeLater() or Platform.runLater(). (I'm not an Android developer but I assume Android would be similar, with Activity.runOnUiThread().) It's easy enough to wrap these in an Executor:

class AppThreadExecutor implements Executor {
  @Override
  public void execute(Runnable command) {
    if (Platform.isFxApplicationThread()) {
      command.run();
    } else {
      Platform.runLater(command);
    }
  }
}

However, that doesn't really get you much; in particular, it doesn't get you Futures, or any interaction with helpful APIs like those in Guava or Vavr.

It seems like there ought to be a library class out there that lets you bootstrap a simple Executor to a full ExecutorService—that this isn't the only situation where you might have some random thread you need to run things on and still want the full concurrency API. But I haven't been able to find it.

So here's what I cobbled together:

/**
 * A basic ExecutorService that wraps any Executor.
 */
public class ExecutorExecutorService extends AbstractExecutorService {

  // ------------------------------------------------------------
  // Final fields

  /** Main lock. */
  private final Object mutex = new Object();

  /** Underlying executor on which tasks are actualy run. */
  private final Executor exec;

  // ------------------------------------------------------------
  // Mutable fields

  /** Pending tasks in submission order. */
  private final Set<Runnable> pending = new LinkedHashSet<>();

  /** Currently running tasks. */
  private final Set<Runnable> running = new HashSet<>();

  /**
   * Current run state.
   * <pre>
   *   2 - running (shutdown() not yet called)
   *   1 - shut down, but not terminated (some tasks still running)
   *   0 - shut down & terminated (no running tasks)
   * </pre>
   */
  private final CountDownLatch runState = new CountDownLatch(2);

  /**
   * Hard shutdown: prevent execution of any pending tasks.
   */
  private boolean hardShutDown = false;

  // ------------------------------------------------------------
  // Constructor

  public ExecutorExecutorService(Executor exec) {
    this.exec = exec;
  }

  // ------------------------------------------------------------
  // ExecutorService

  /**
   * Schedules the specified command for execution.
   * @param command The command
   *
   * @see Executor#execute(Runnable)
   */
  @Override
  public void execute(Runnable command) {
    synchronized (mutex) {
      if (isShutdown()) {
        throw new RejectedExecutionException();
      }
      pending.add(command);
      Task task = new Task(command);
      exec.execute(task);
    }
  }

  /**
   * @return true if this ExecutorService has been shut down; false otherwise
   *
   * @see ExecutorService#isShutdown()
   */
  @Override
  public boolean isShutdown() {
    return runState.getCount() <= 1;
  }

  /**
   * @return true if this ExecutorService has been shut down and no tasks are
   * running; false otherwise
   *
   * @see ExecutorService#isTerminated()
   */
  @Override
  public boolean isTerminated() {
    return runState.getCount() == 0;
  }

  /**
   * Prevents submission of new tasks, but allows pending tasks to run.
   *
   * @see ExecutorService#shutdown()
   */
  @Override
  public void shutdown() {
    synchronized (mutex) {
      if (!isShutdown()) {
        runState.countDown();
        if (pending.isEmpty()) {
          maybeTerminate();
        }
      }
    }
  }

  /**
   * Prevents submission of new tasks and prevents pending tasks from starting.
   * <p>
   * Note that the "best-effort attempt to stop processing" required for
   * {@link ExecutorService#shutdownNow()} is, in this case, no effort, since
   * we really don't know anything about the underlying executor and how to stop
   * it. The only hard guarantee is that after this method is called, no pending
   * tasks will be run.
   *
   * @see ExecutorService#shutdownNow()
   */
  @Override
  public List<Runnable> shutdownNow() {
    synchronized (mutex) {
      hardShutDown = true;
      shutdown();
      maybeTerminate();
      Runnable[] pendingArray = pending.toArray(new Runnable[pending.size()]);
      List<Runnable> pendingList = Arrays.asList(pendingArray);
      return Collections.unmodifiableList(pendingList);
    }
  }

  /**
   * Blocks until this ExecutorService has been shut down and all running tasks have
   * completed, or until timed out or interrupted.
   *
   * @see ExecutorService#awaitTermination(long, TimeUnit)
   */
  @Override
  public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
    return runState.await(timeout, unit);
  }

  // ------------------------------------------------------------
  // Helper methods

  /**
   * Checks whether it's OK to run this command, and if so, moves it from the
   * pending queue to the running set
   *
   * @param command the command to prepare
   * @return true if it's OK to run the command, false if this ExecutorService
   * has been hard-shutdown with {@link #shutdownNow()}
   */
  private boolean prepare(Runnable command) {
    synchronized (mutex) {
      if (hardShutDown) {
        return false;
      }
      pending.remove(command);
      running.add(command);
      return true;
    }
  }

  /**
   * Removes this command from the running set after completion, and if we've
   * been shut down and this is the last task, marks us as terminated
   * @param command the command
   */
  private void cleanUp(Runnable command) {
    synchronized (mutex) {
      running.remove(command);
      maybeTerminate();
    }
  }

  /** If we're shut down & have no running tasks, mark us as terminated */
  private void maybeTerminate() {
    if (running.isEmpty() && isShutdown()) {
      runState.countDown();
    }
  }

  // ------------------------------------------------------------
  // Helper classes

  private class Task implements Runnable {
    private final Runnable command;

    private Task(Runnable command) {
      this.command = command;
    }

    @Override
    public void run() {
      boolean canRun = prepare(command);
      if (canRun) {
        try {
          command.run();
        } finally {
          cleanUp(command);
        }
      }
    }
  }

}

My questions are:

  1. Does this look correct? Have I correctly understood the ExecutorService API?
  2. Is it thread-safe? Are there any deadlocks or race conditions I haven't spotted?
    • Am I correct in thinking isShutdown() and isTerminated() don't need synchronization?
  3. Is it too thread-safe, i.e. am I synchronizing in places I don't have to?
  4. Does synchronized do the job here, or is there a reason to use a more sophisticated API?
    • For instance, I looked at using ReadWriteLock, but I found that I didn't have any pure reads, unless you count isShutdown() and isTerminated().
  5. Is there a better (= more legible, little or no more code) alternative to CountDownLatch to support awaitTermination()?

N.b. This isn't really a question about Swing or JavaFX; those are just examples of the kind of situation that might require bootstrapping from an arbitrary thread or an Executor-like task runner to a full ExecutorService. If you have specific advice about how this is the wrong approach for those frameworks, I appreciate that, but please make it a comment rather than an answer.

\$\endgroup\$
  • \$\begingroup\$ If you're using Swing and only occasionally need to run background code, look at SwingWorker. What you're doing seems like massive overkill and re-inventing the wheel. Just use the existing classes. \$\endgroup\$ – markspace Dec 7 '17 at 21:55
1
\$\begingroup\$
  1. Does this look correct? Have I correctly understood the ExecutorService API?

I guess, so.

  1. Is it thread-safe? Are there any deadlocks or race conditions I haven't spotted?

It's probably fine. All your public methods either operate on thread-safe objects like runState or synchronize on mutex.

  1. Is it too thread-safe, i.e. am I synchronizing in places I don't have to?

All but one methods deal with shutdown, so synchronizing too much in them won't be a big deal. Anyway, I think, it's all necessary. You might be eliminate it by using concurrent sets.

Synchronizing inside of execute feels somehow wrong as it might become a bottleneck in general (for things like Swing, it's probably fine). However, I can't see how to eliminate it.

  1. Does synchronized do the job here, or is there a reason to use a more sophisticated API?

I guess, it's fine as the only read operation need no lock (so ReadWriteLock won't help) and you need no tryLock either.

  1. Is there a better (= more legible, little or no more code) alternative to CountDownLatch to support awaitTermination()

I wondered why do you use it, but it seems to fit exactly.


This looks unnecessary complicated:

final Runnable[] pendingArray = pending.toArray(new Runnable[pending.size()]);
final List<Runnable> pendingList = Arrays.asList(pendingArray);
return Collections.unmodifiableList(pendingList);

There's no double copying, but there should be a simpler way, even when you want to avoid dependency on Guava

return ImmutableList.copyOf(pending);

Maybe just

return Collections.unmodifiableList(new ArrayList<>(pending));

Summary

It looks good, but I'm afraid, I'm not sufficiently familiar with the executor stuff. I'm surprised that there's nothing like you did in either JDK nor Guava. You're upgrading an Executor to an ExecutorService just like Guava MoreExecutors.listeningDecorator upgrades an ExecutorService to a ListeningExecutorService.

\$\endgroup\$
  • \$\begingroup\$ hi, I have a similar question related to thread safety issues here. Wanted to see if you can help me out? \$\endgroup\$ – david Dec 24 '17 at 17:48
0
\$\begingroup\$

I'm not entirely sure what exactly you're trying to achieve here.

The whole idea behind the swing thread (actually java EDT) and javaFX thread is that you have 1 single thread to handle all GUI updates. They are designed to apply each GUI change in the order they are requested and assume all those changes can be handled fast.

This design ensures that you don't get into trouble with 2 updates happening at the same time causing weird behaviour.

A typical example taken from this site (who took it from the book "Filthy rich clients") looks like this:

public void actionPerformed(ActionEvent e) {
  new Thread(new Runnable() {
    final String text = readHugeFile();
    SwingUtilities.invokeLater(new Runnable() {
      public void run() {
        textArea.setText(text);
      }
    });
  }).start();
}

What's happening here is that you have a really computation heavy method readHugeFile(). You really don't want this to be handled on the GUI thread because as long as this method doesn't return, the entire application will become unresponsive.

So instead this method is being executed on a different java thread.

But we do need to make sure that when the file is read and we want to show the resulting text in the designated textArea that this happens on our specific GUI thread.

The nice thing about SwingUtilities.invokeLater is that it's asynchronous. You're literally saying: "I'm done with the heavy work. Here's the text you need to display".

This textArea-update is then put into the GUI update queue and once all the small requests before it are done, the text is put into the `textArea´.

Do you really need to have any other concurrency mechanism for a use case like this?

\$\endgroup\$
  • \$\begingroup\$ I'm well aware how SwingUtilities.invokeLater() is normally used, thank you. For your use case it's sufficient. For other cases it's not. Off the top of my head (all real-world cases): reading a model value from outside the EDT (reads also need to happen on the EDT to be consistent); invoking a callback after a UI operation completes; blocking a background operation till after a UI operation completes (the last you can do with SwingUtilities.invokeAndWait(), but it doesn't have a counterpart in JavaFX). All of these can be bodged together w/Runnables but are much cleaner w/modern APIs. \$\endgroup\$ – David Moles May 10 '17 at 19:53
  • \$\begingroup\$ Most Swing developers are used to thinking of the EDT as the "main" thread and everything else as temporary background jobs suitable for SwingWorker or for throwaway threads like your example. But sometimes the real work of the application is in computation or network I/O or some such and it's the UI that's the afterthought. \$\endgroup\$ – David Moles May 10 '17 at 19:56
  • \$\begingroup\$ And also, yes, I do have non-Swing/JavaFX use cases for this, but they're too complicated to go into here. \$\endgroup\$ – David Moles May 10 '17 at 19:56
  • \$\begingroup\$ I tend to dissagree with your last comment. Your question as it stands now is either about javaFX/swing or it's entirely hypothetical. In the first case my answer points out that it's overkill. In the second case you're better of adding those cases here to see if this what you're asking about actually helps or if there's a more elegant solution to those cases. As it stands now I don't really see a use for your class. \$\endgroup\$ – Imus May 11 '17 at 7:28
  • \$\begingroup\$ Luckily this is Code Review Stack Exchange, not Use Case Justification Stack Exchange. If you're not interested, don't review it. \$\endgroup\$ – David Moles May 12 '17 at 17:40
0
\$\begingroup\$

It looks good to me. However threading is very subtle and I might have missed a bug.

Very minor point: I would call pending.clear() in shutdownNow(). Also shutdownNow() calls maybeTerminate() which might have already been called indirectly through the call to shutdown(). There is probably a way to call it only once by re-ordering calls and calling pending.clear() first (having copied its elements in pendingList first). But it might make the code less legible.

Unit tests would be nice, especially since multi-threading is so tricky. However writing useful tests for threaded code might be harder than writing threaded code.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.