Fix context propagation in gRPC SerializingExecutor

[richardstartin]

What Does This Do

SerializingExecutor is implemented as follows:

public final class SerializingExecutor implements Executor, Runnable {
...

  /**
   * Runs the given runnable strictly after all Runnables that were submitted
   * before it, and using the {@code executor} passed to the constructor.     .
   */
  @Override
  public void execute(Runnable r) {
    runQueue.add(checkNotNull(r, "'r' must not be null."));
    schedule(r);
  }

  private void schedule(@Nullable Runnable removable) {
    if (atomicHelper.runStateCompareAndSet(this, STOPPED, RUNNING)) {
      boolean success = false;
      try {
        executor.execute(this);
        success = true;
      } finally {
        // It is possible that at this point that there are still tasks in
        // the queue, it would be nice to keep trying but the error may not
        // be recoverable.  So we update our state and propagate so that if
        // our caller deems it recoverable we won't be stuck.
        if (!success) {
          if (removable != null) {
            // This case can only be reached if 'this' was not currently running, and we failed to
            // reschedule.  The item should still be in the queue for removal.
            // ConcurrentLinkedQueue claims that null elements are not allowed, but seems to not
            // throw if the item to remove is null.  If removable is present in the queue twice,
            // the wrong one may be removed.  It doesn't seem possible for this case to exist today.
            // This is important to run in case of RejectedExectuionException, so that future calls
            // to execute don't succeed and accidentally run a previous runnable.
            runQueue.remove(removable);
          }
          atomicHelper.runStateSet(this, STOPPED);
        }
      }
    }
  }

  @Override
  public void run() {
    Runnable r;
    try {
      Executor oldExecutor = executor;
      while (oldExecutor == executor && (r = runQueue.poll()) != null ) {
        try {
          r.run();
        } catch (RuntimeException e) {
          // Log it and keep going.
          log.log(Level.SEVERE, "Exception while executing runnable " + r, e);
        }
      }
    } finally {
      atomicHelper.runStateSet(this, STOPPED);
    }
    if (!runQueue.isEmpty()) {
      // we didn't enqueue anything but someone else did.
      schedule(null);
    }
  }
...

Currently, the active span is propagated when the SerializingExecutor is scheduled, and is active while all tasks are executed, instead of injecting the active span into the executed task.

The solution is simple:

• Disable async propagation in SerializingExecutor#schedule so it doesn't get associated with the executor
• Whitelist SerializingExecutor so it propagates the active span with the provided task

This allows a lot of scope management instrumentation to be removed, simplifying the instrumentation at the same time as ensuring the profiler is aware of all request/response activity.

Motivation

Additional Notes