Merge pull request #169 from oracle/executor

Share executor service

Author: rjeberhard

operator/src/main/java/oracle/kubernetes/operator/Main.java

@@ -15,6 +15,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
@@ -70,7 +71,6 @@
 import oracle.kubernetes.operator.wlsconfig.WlsServerConfig;
 import oracle.kubernetes.operator.work.Component;
 import oracle.kubernetes.operator.work.Container;
-import oracle.kubernetes.operator.work.ContainerResolver;
 import oracle.kubernetes.operator.work.Engine;
 import oracle.kubernetes.operator.work.Fiber;
 import oracle.kubernetes.operator.work.Fiber.CompletionCallback;
@@ -111,13 +111,17 @@ public class Main {
   private static final CallBuilderFactory callBuilderFactory = new CallBuilderFactory(tuningAndConfig);
 
   private static final Container container = new Container();
+  private static final ScheduledExecutorService wrappedExecutorService = 
+      Engine.wrappedExecutorService("operator", container);
+  
   static {
     container.getComponents().put(
         ProcessingConstants.MAIN_COMPONENT_NAME,
-        Component.createFor(TuningParameters.class, tuningAndConfig, callBuilderFactory));
+        Component.createFor(ScheduledExecutorService.class, wrappedExecutorService,
+            TuningParameters.class, tuningAndConfig, callBuilderFactory));
   }
 
-  private static final Engine engine = new Engine("operator", container);
+  private static final Engine engine = new Engine(wrappedExecutorService);
   private static final FiberGate domainUpdaters = new FiberGate(engine);
 
   private static final ConcurrentMap<String, Boolean> initialized = new ConcurrentHashMap<>();

operator/src/main/java/oracle/kubernetes/operator/http/HttpClient.java

@@ -6,7 +6,6 @@
 import io.kubernetes.client.ApiException;
 import io.kubernetes.client.models.V1Service;
 import io.kubernetes.client.models.V1ServiceSpec;
-import oracle.kubernetes.operator.helpers.CallBuilder;
 import oracle.kubernetes.operator.helpers.CallBuilderFactory;
 import oracle.kubernetes.operator.helpers.SecretHelper;
 import oracle.kubernetes.operator.logging.LoggingFacade;

operator/src/main/java/oracle/kubernetes/operator/wlsconfig/WlsRetriever.java

@@ -14,6 +14,7 @@
 import oracle.kubernetes.operator.logging.LoggingFacade;
 import oracle.kubernetes.operator.logging.LoggingFactory;
 import oracle.kubernetes.operator.logging.MessageKeys;
+import oracle.kubernetes.operator.work.ContainerResolver;
 import oracle.kubernetes.operator.work.NextAction;
 import oracle.kubernetes.operator.work.Packet;
 import oracle.kubernetes.operator.work.Step;
@@ -24,9 +25,8 @@
 import java.util.Random;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
@@ -293,7 +293,7 @@ public WlsDomainConfig readConfig(String principal) {
     LOGGER.entering();
 
     long timeout = READ_CONFIG_TIMEOUT_MILLIS;
-    ExecutorService executorService = Executors.newSingleThreadExecutor();
+    ScheduledExecutorService executorService = ContainerResolver.getInstance().getContainer().getSPI(ScheduledExecutorService.class);
     long startTime = System.currentTimeMillis();
     Future<WlsDomainConfig> future = executorService.submit(() -> getWlsDomainConfig(principal, timeout));
     executorService.shutdown();

operator/src/main/java/oracle/kubernetes/operator/work/Engine.java

@@ -13,93 +13,45 @@
  * Collection of {@link Fiber}s. Owns an {@link Executor} to run them.
  */
 public class Engine {
-  private final int DEFAULT_THREAD_COUNT = 10;
-
-  private volatile ScheduledExecutorService threadPool;
-  public final String id;
-  private final Container container;
-
-  /**
-   * Returns identifier for this Engine
-   * @return identifier
-   */
-  String getId() {
-    return id;
+  private static final int DEFAULT_THREAD_COUNT = 10;
+  
+  public static ScheduledExecutorService wrappedExecutorService(String id, Container container) {
+	  return wrap(container, Executors.newScheduledThreadPool(DEFAULT_THREAD_COUNT, 
+			  new DaemonThreadFactory(id)));
   }
 
-  /**
-   * Returns the container
-   * @return container
-   */
-  Container getContainer() {
-    return container;
-  }
+  private volatile ScheduledExecutorService threadPool;
 
   /**
    * Returns the executor
    * @return executor
    */
   public ScheduledExecutorService getExecutor() {
-    if (threadPool == null) {
-      synchronized (this) {
-        threadPool = wrap(Executors.newScheduledThreadPool(DEFAULT_THREAD_COUNT, new DaemonThreadFactory()));
-      }
-    }
     return threadPool;
   }
 
   /**
-   * Creates engine with the specified id, default container and specified executor
-   * @param id Engine id
-   * @param threadPool Executor
-   */
-  public Engine(String id, ScheduledExecutorService threadPool) {
-    this(id, ContainerResolver.getDefault().getContainer(), threadPool);
-  }
-
-  /**
-   * Creates engine with the specified id, container and executor
-   * @param id Engine id
-   * @param container Container
+   * Creates engine with the specified executor
    * @param threadPool Executor
    */
-  public Engine(String id, Container container, ScheduledExecutorService threadPool) {
-    this(id, container);
-    this.threadPool = threadPool != null ? wrap(threadPool) : null;
+  public Engine(ScheduledExecutorService threadPool) {
+    this.threadPool = threadPool;
   }
 
   /**
    * Creates engine with the specified id and default container and executor
    * @param id Engine id
    */
   public Engine(String id) {
-    this(id, ContainerResolver.getDefault().getContainer());
-  }
-
-  /**
-   * Creates engine with the specified id, container and default executor
-   * @param id Engine id
-   * @param container Container
-   */
-  public Engine(String id, Container container) {
-    this.id = id;
-    this.container = container;
-  }
-
-  /**
-   * Sets the executor
-   * @param threadPool Executor
-   */
-  public void setExecutor(ScheduledExecutorService threadPool) {
-    this.threadPool = threadPool != null ? wrap(threadPool) : null;
+    this(wrappedExecutorService(id, ContainerResolver.getDefault().getContainer()));
   }
 
   void addRunnable(Fiber fiber) {
     getExecutor().execute(fiber);
   }
 
-  private ScheduledExecutorService wrap(ScheduledExecutorService ex) {
-    return ContainerResolver.getDefault().wrapExecutor(container, ex);
+  private static ScheduledExecutorService wrap(Container container, ScheduledExecutorService ex) {
+    return container != null ? ContainerResolver.getDefault().wrapExecutor(container, ex) : ex;
   }
 
   /**
@@ -120,11 +72,11 @@ Fiber createChildFiber(Fiber parent) {
     return new Fiber(this, parent);
   }
 
-  private class DaemonThreadFactory implements ThreadFactory {
+  private static class DaemonThreadFactory implements ThreadFactory {
     final AtomicInteger threadNumber = new AtomicInteger(1);
     final String namePrefix;
 
-    DaemonThreadFactory() {
+    DaemonThreadFactory(String id) {
       namePrefix = "engine-" + id  + "-thread-";
     }
 

operator/src/main/java/oracle/kubernetes/operator/work/Fiber.java

@@ -479,14 +479,12 @@ public void run() {
 
       final Fiber oldFiber = CURRENT_FIBER.get();
       CURRENT_FIBER.set(this);
-      Container oldContainer = ContainerResolver.getDefault().enterContainer(owner.getContainer());
       try {
         // doRun returns true to indicate an early exit from fiber processing
         if (!doRun(next)) {
             completionCheck();
         }
       } finally {
-        ContainerResolver.getDefault().exitContainer(oldContainer);
         CURRENT_FIBER.set(oldFiber);
       }
     }
@@ -660,9 +658,13 @@ private String getName() {
       sb.append(parent.getName());
       sb.append("-child-");
     } else {
-      sb.append("engine-");
-      sb.append(owner.id);
-      sb.append("-fiber-");
+      synchronized (this) {
+        if (currentThread != null) {
+          sb.append(currentThread.getName());
+          sb.append("-");
+    	  }
+      }
+      sb.append("fiber-");
     }
     sb.append(id);
     return sb.toString();