Add shadow context implementation.

A shadow context is a context referring to a context of another vertx instance.

Shadow context is created when a context is obtained from a vertx instance and the current context is already associated with a different context.

A shadow context appears to be local to each vertx instance.

Author: vietj

vertx-core/src/main/java/io/vertx/core/ThreadingModel.java

@@ -11,26 +11,30 @@
 package io.vertx.core;
 
 /**
- * The threading model defines how user tasks should be executed.
+ * The threading model defines the scheduler to execute context tasks.
  *
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
 public enum ThreadingModel {
 
   /**
-   * Event-loop threading model.
+   * Tasks are scheduled on the event-loop thread.
    */
   EVENT_LOOP,
 
   /**
-   * Worker threading model
+   * Tasks are scheduled on a worker pool.
    */
   WORKER,
 
   /**
-   * Virtual thread threading model
+   * Tasks are scheduled on a virtual thread.
    */
-  VIRTUAL_THREAD
+  VIRTUAL_THREAD,
 
+  /**
+   * Tasks are scheduled on threads not managed by the current vertx instance.
+   */
+  OTHER
 }
 

vertx-core/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java

@@ -248,7 +248,7 @@ private void beginRequest(Stream stream, HttpRequestHead request, boolean chunke
       if (this.metrics != null) {
         stream.metric = this.metrics.requestBegin(request.uri, request);
       }
-      VertxTracer tracer = context.tracer();
+      VertxTracer tracer = stream.context.tracer();
       if (tracer != null) {
         BiConsumer<String, String> headers = (key, val) -> new HeadersAdaptor(nettyRequest.headers()).add(key, val);
         String operation = request.traceOperation;
@@ -901,7 +901,7 @@ private void handleResponseEnd(Stream stream, LastHttpContent trailer) {
       stream.responseEnded = true;
       check = requests.peek() != stream;
     }
-    VertxTracer tracer = context.tracer();
+    VertxTracer tracer = stream.context.tracer();
     if (tracer != null) {
       tracer.receiveResponse(stream.context, response, stream.trace, null, HttpUtils.CLIENT_RESPONSE_TAG_EXTRACTOR);
     }
@@ -1164,7 +1164,6 @@ protected void handleClosed() {
         evictionHandler.handle(null);
       }
     }
-    VertxTracer tracer = context.tracer();
     List<Stream> allocatedStreams;
     List<Stream> sentStreams;
     synchronized (this) {
@@ -1180,6 +1179,7 @@ protected void handleClosed() {
         metrics.requestReset(stream.metric);
       }
       Object trace = stream.trace;
+      VertxTracer tracer = stream.context.tracer();
       if (tracer != null && trace != null) {
         tracer.receiveResponse(stream.context, null, trace, HttpUtils.CONNECTION_CLOSED_EXCEPTION, TagExtractor.empty());
       }

vertx-core/src/main/java/io/vertx/core/impl/ContextBase.java

@@ -10,11 +10,15 @@
  */
 package io.vertx.core.impl;
 
+import io.vertx.core.Future;
+import io.vertx.core.Handler;
 import io.vertx.core.ThreadingModel;
 import io.vertx.core.internal.ContextInternal;
 import io.vertx.core.spi.context.storage.AccessMode;
 import io.vertx.core.spi.context.storage.ContextLocal;
 
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executor;
 import java.util.function.Supplier;
 
 /**
@@ -30,6 +34,9 @@ abstract class ContextBase implements ContextInternal {
     this.locals = locals;
   }
 
+  @Override
+  public abstract EventExecutor executor();
+
   public ContextInternal beginDispatch() {
     VertxImpl vertx = (VertxImpl) owner();
     return vertx.beginDispatch(this);
@@ -68,4 +75,50 @@ public final <T> void putLocal(ContextLocal<T> key, AccessMode accessMode, T val
     }
     accessMode.put(locals, index, value);
   }
+
+  @Override
+  public final boolean inThread() {
+    return executor().inThread();
+  }
+
+  @Override
+  public final <T> void emit(T argument, Handler<T> task) {
+    if (executor().inThread()) {
+      ContextInternal prev = beginDispatch();
+      try {
+        task.handle(argument);
+      } catch (Throwable t) {
+        reportException(t);
+      } finally {
+        endDispatch(prev);
+      }
+    } else {
+      executor().execute(() -> emit(argument, task));
+    }
+  }
+
+  @Override
+  public final void execute(Runnable task) {
+    if (executor().inThread()) {
+      task.run();
+    } else {
+      executor().execute(task);
+    }
+  }
+
+  /**
+   * <ul>
+   *   <li>When the current thread is event-loop thread of this context the implementation will execute the {@code task} directly</li>
+   *   <li>When the current thread is a worker thread of this context the implementation will execute the {@code task} directly</li>
+   *   <li>Otherwise the task will be scheduled on the context thread for execution</li>
+   * </ul>
+   */
+  @Override
+  public final <T> void execute(T argument, Handler<T> task) {
+    if (executor().inThread()) {
+      task.handle(argument);
+    } else {
+      executor().execute(() -> task.handle(argument));
+    }
+  }
 }

vertx-core/src/main/java/io/vertx/core/impl/ContextImpl.java

@@ -43,6 +43,7 @@ public final class ContextImpl extends ContextBase implements ContextInternal {
   private final CloseFuture closeFuture;
   private final ClassLoader tccl;
   private final EventLoop eventLoop;
+  private final ThreadingModel threadingModel;
   private final EventExecutor executor;
   private ConcurrentMap<Object, Object> data;
   private volatile Handler<Throwable> exceptionHandler;
@@ -52,6 +53,7 @@ public final class ContextImpl extends ContextBase implements ContextInternal {
   public ContextImpl(VertxInternal vertx,
                         Object[] locals,
                         EventLoop eventLoop,
+                        ThreadingModel threadingModel,
                         EventExecutor executor,
                         WorkerPool workerPool,
                         TaskQueue orderedTasks,
@@ -62,6 +64,7 @@ public ContextImpl(VertxInternal vertx,
     this.deployment = deployment;
     this.config = deployment != null ? deployment.config() : new JsonObject();
     this.eventLoop = eventLoop;
+    this.threadingModel = threadingModel;
     this.executor = executor;
     this.tccl = tccl;
     this.owner = vertx;
@@ -92,14 +95,9 @@ public VertxInternal owner() {
     return owner;
   }
 
-  @Override
-  public <T> Future<T> executeBlockingInternal(Callable<T> action) {
-    return executeBlocking(this, action, null, null);
-  }
-
   @Override
   public <T> Future<T> executeBlocking(Callable<T> blockingCodeHandler, boolean ordered) {
-    return executeBlocking(this, blockingCodeHandler, workerPool, ordered ? orderedTasks : null);
+    return workerPool.executeBlocking(this, blockingCodeHandler, ordered ? orderedTasks : null);
   }
 
   @Override
@@ -109,74 +107,16 @@ public EventExecutor executor() {
 
   @Override
   public boolean isEventLoopContext() {
-    return executor.threadingModel() == ThreadingModel.EVENT_LOOP;
+    return threadingModel() == ThreadingModel.EVENT_LOOP;
   }
 
   @Override
   public boolean isWorkerContext() {
-    return executor.threadingModel() == ThreadingModel.WORKER;
+    return threadingModel() == ThreadingModel.WORKER;
   }
 
   public ThreadingModel threadingModel() {
-    return executor.threadingModel();
-  }
-
-  @Override
-  public boolean inThread() {
-    return executor.inThread();
-  }
-
-  @Override
-  public <T> Future<T> executeBlocking(Callable<T> blockingCodeHandler, TaskQueue queue) {
-    return executeBlocking(this, blockingCodeHandler, workerPool, queue);
-  }
-
-  static <T> Future<T> executeBlocking(ContextInternal context, Callable<T> blockingCodeHandler,
-                                       WorkerPool workerPool, TaskQueue queue) {
-    return internalExecuteBlocking(context, promise -> {
-      T result;
-      try {
-        result = blockingCodeHandler.call();
-      } catch (Throwable e) {
-        promise.fail(e);
-        return;
-      }
-      promise.complete(result);
-    }, workerPool, queue);
-  }
-
-  private static <T> Future<T> internalExecuteBlocking(ContextInternal context, Handler<Promise<T>> blockingCodeHandler,
-      WorkerPool workerPool, TaskQueue queue) {
-    PoolMetrics metrics = workerPool.metrics();
-    Object queueMetric = metrics != null ? metrics.enqueue() : null;
-    Promise<T> promise = context.promise();
-    Future<T> fut = promise.future();
-    try {
-      Runnable command = () -> {
-        Object execMetric = null;
-        if (metrics != null) {
-          metrics.dequeue(queueMetric);
-          execMetric = metrics.begin();
-        }
-        context.dispatch(promise, blockingCodeHandler);
-        if (metrics != null) {
-          metrics.end(execMetric);
-        }
-      };
-      Executor exec = workerPool.executor();
-      if (queue != null) {
-        queue.execute(command, exec);
-      } else {
-        exec.execute(command);
-      }
-    } catch (RejectedExecutionException e) {
-      // Pool is already shut down
-      if (metrics != null) {
-        metrics.dequeue(queueMetric);
-      }
-      throw e;
-    }
-    return fut;
+    return threadingModel;
   }
 
   @Override
@@ -225,68 +165,6 @@ public Handler<Throwable> exceptionHandler() {
     return exceptionHandler;
   }
 
-  protected void runOnContext(ContextInternal ctx, Handler<Void> action) {
-    try {
-      Executor exec = ctx.executor();
-      exec.execute(() -> ctx.dispatch(action));
-    } catch (RejectedExecutionException ignore) {
-      // Pool is already shut down
-    }
-  }
-
-  @Override
-  public void execute(Runnable task) {
-    execute(this, task);
-  }
-
-  @Override
-  public <T> void execute(T argument, Handler<T> task) {
-    execute(this, argument, task);
-  }
-
-  protected void execute(ContextInternal ctx, Runnable task) {
-    if (inThread()) {
-      task.run();
-    } else {
-      executor.execute(task);
-    }
-  }
-
-  /**
-   * <ul>
-   *   <li>When the current thread is event-loop thread of this context the implementation will execute the {@code task} directly</li>
-   *   <li>When the current thread is a worker thread of this context the implementation will execute the {@code task} directly</li>
-   *   <li>Otherwise the task will be scheduled on the context thread for execution</li>
-   * </ul>
-   */
-  protected <T> void execute(ContextInternal ctx, T argument, Handler<T> task) {
-    if (inThread()) {
-      task.handle(argument);
-    } else {
-      executor.execute(() -> task.handle(argument));
-    }
-  }
-
-  @Override
-  public <T> void emit(T argument, Handler<T> task) {
-    emit(this, argument, task);
-  }
-
-  protected <T> void emit(ContextInternal ctx, T argument, Handler<T> task) {
-    if (inThread()) {
-      ContextInternal prev = ctx.beginDispatch();
-      try {
-        task.handle(argument);
-      } catch (Throwable t) {
-        reportException(t);
-      } finally {
-        ctx.endDispatch(prev);
-      }
-    } else {
-      executor.execute(() -> emit(ctx, argument, task));
-    }
-  }
-
   @Override
   public ContextInternal duplicate() {
     return new DuplicatedContext(this, locals.length == 0 ? VertxImpl.EMPTY_CONTEXT_LOCALS : new Object[locals.length]);

vertx-core/src/main/java/io/vertx/core/impl/DuplicatedContext.java

@@ -23,7 +23,6 @@
 
 import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.Executor;
 
 /**
  * A context that forwards most operations to a delegate. This context
@@ -48,11 +47,6 @@ public ThreadingModel threadingModel() {
     return delegate.threadingModel();
   }
 
-  @Override
-  public boolean inThread() {
-    return delegate.inThread();
-  }
-
   @Override
   public CloseFuture closeFuture() {
     return delegate.closeFuture();
@@ -75,7 +69,7 @@ public Context exceptionHandler(Handler<Throwable> handler) {
   }
 
   @Override
-  public Executor executor() {
+  public EventExecutor executor() {
     return delegate.executor();
   }
 
@@ -119,34 +113,9 @@ public ConcurrentMap<Object, Object> contextData() {
     return delegate.contextData();
   }
 
-  @Override
-  public <T> Future<T> executeBlockingInternal(Callable<T> action) {
-    return ContextImpl.executeBlocking(this, action, delegate.owner().getInternalWorkerPool(), null);
-  }
-
   @Override
   public <T> Future<T> executeBlocking(Callable<T> blockingCodeHandler, boolean ordered) {
-    return ContextImpl.executeBlocking(this, blockingCodeHandler, delegate.workerPool, ordered ? delegate.orderedTasks : null);
-  }
-
-  @Override
-  public <T> Future<T> executeBlocking(Callable<T> blockingCodeHandler, TaskQueue queue) {
-    return ContextImpl.executeBlocking(this, blockingCodeHandler, delegate.workerPool, queue);
-  }
-
-  @Override
-  public <T> void execute(T argument, Handler<T> task) {
-    delegate.execute(this, argument, task);
-  }
-
-  @Override
-  public <T> void emit(T argument, Handler<T> task) {
-    delegate.emit(this, argument, task);
-  }
-
-  @Override
-  public void execute(Runnable task) {
-    delegate.execute(this, task);
+    return delegate.workerPool.executeBlocking(this, blockingCodeHandler, ordered ? delegate.orderedTasks : null);
   }
 
   @Override