Issue OpenLiberty#8611 add introspector for persistent executor

dev/com.ibm.ws.concurrent.persistent/bnd.bnd

@@ -1,5 +1,5 @@
 #*******************************************************************************
-# Copyright (c) 2017,2019 IBM Corporation and others.
+# Copyright (c) 2017,2020 IBM Corporation and others.
 # All rights reserved. This program and the accompanying materials
 # are made available under the terms of the Eclipse Public License v1.0
 # which accompanies this distribution, and is available at
@@ -44,6 +44,7 @@ Service-Component:\
 -dsannotations:\
   com.ibm.ws.concurrent.persistent.internal.ApplicationTracker,\
   com.ibm.ws.concurrent.persistent.internal.PersistentExecutorImpl,\
+  com.ibm.ws.concurrent.persistent.internal.PersistentExecutorIntrospector,\
   com.ibm.ws.concurrent.persistent.internal.PersistentExecutorMBeanImpl
 
 instrument.classesExcludes: com/ibm/ws/concurrent/persistent/resources/*.class

dev/com.ibm.ws.concurrent.persistent/src/com/ibm/ws/concurrent/persistent/internal/ApplicationTracker.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2015 IBM Corporation and others.
+ * Copyright (c) 2015,2020 IBM Corporation and others.
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
  * which accompanies this distribution, and is available at
@@ -10,6 +10,7 @@
  *******************************************************************************/
 package com.ibm.ws.concurrent.persistent.internal;
 
+import java.io.PrintWriter;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
@@ -67,7 +68,7 @@ public class ApplicationTracker {
 
     /**
      * Declarative Services method for setting a started Application instance
-     * 
+     *
      * @param ref reference to the service
      */
     @Reference(service = Application.class,
@@ -94,7 +95,7 @@ protected void addStartedApplication(ServiceReference<Application> ref) {
 
     /**
      * Declarative Services method for setting a starting Application instance
-     * 
+     *
      * @param ref reference to the service
      */
     @Reference(service = Application.class,
@@ -143,9 +144,33 @@ void deferTask(Runnable task, String appName, PersistentExecutorImpl persistentE
             executor.submit(task);
     }
 
+    /**
+     * Dump internal state to the introspector.
+     *
+     * @param out writer for the introspector.
+     */
+    void introspect(PrintWriter out) {
+        if (lock.readLock().tryLock())
+            try {
+                for (Map.Entry<String, Set<Runnable>> entry : deferredTasks.entrySet()) {
+                    out.print("Deferred tasks for ");
+                    out.print(entry.getKey());
+                    out.print(": ");
+                    out.println(entry.getValue());
+                }
+                for (Map.Entry<String, ApplicationState> entry : appStates.entrySet()) {
+                    out.print(entry.getKey());
+                    out.print(" is ");
+                    out.println(entry.getValue());
+                }
+            } finally {
+                lock.readLock().unlock();
+            }
+    }
+
     /**
      * Returns true if the application with the specified name is started, otherwise false.
-     * 
+     *
      * @return true if the application with the specified name is started, otherwise false.
      */
     boolean isStarted(String appName) {
@@ -159,7 +184,7 @@ boolean isStarted(String appName) {
 
     /**
      * Declarative Services method for unsetting a started Application instance
-     * 
+     *
      * @param ref reference to the service
      */
     protected void removeStartedApplication(ServiceReference<Application> ref) {
@@ -174,15 +199,15 @@ protected void removeStartedApplication(ServiceReference<Application> ref) {
 
     /**
      * Declarative Services method for unsetting a starting Application instance
-     * 
+     *
      * @param ref reference to the service
      */
     protected void removeStartingApplication(ServiceReference<Application> ref) {
     }
 
     /**
      * Declarative Services method for setting the Liberty executor.
-     * 
+     *
      * @param svc the service
      */
     @Reference(target = "(component.name=com.ibm.ws.threading)")
@@ -192,7 +217,7 @@ protected void setExecutor(ExecutorService svc) {
 
     /**
      * Declarative Services method for unsetting the Liberty executor.
-     * 
+     *
      * @param svc the service
      */
     protected void unsetExecutor(ExecutorService svc) {

dev/com.ibm.ws.concurrent.persistent/src/com/ibm/ws/concurrent/persistent/internal/Config.java

@@ -143,6 +143,8 @@ public String toString() {
         return new StringBuilder(300)
                         .append("instance=")
                         .append(Integer.toHexString(System.identityHashCode(this)))
+                        .append(",id=")
+                        .append(id)
                         .append(",jndiName=")
                         .append(jndiName)
                         .append(",enableTaskExecution=")
@@ -161,8 +163,6 @@ public String toString() {
                         .append(retryLimit)
                         .append(",xpathId=")
                         .append(xpathId)
-                        .append(",id=")
-                        .append(id)
                         .toString();
     }
 

dev/com.ibm.ws.concurrent.persistent/src/com/ibm/ws/concurrent/persistent/internal/PersistentExecutorImpl.java

@@ -16,6 +16,7 @@
 import java.io.InputStream;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
+import java.io.PrintWriter;
 import java.io.Serializable;
 import java.io.Writer;
 import java.net.InetAddress;
@@ -39,6 +40,7 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -250,7 +252,7 @@ public String run() throws UnknownHostException {
     /**
      * Reference to the future for the next (or current) poll.
      */
-    private final AtomicReference<Future<?>> pollingFutureRef = new AtomicReference<Future<?>>();
+    private final AtomicReference<ScheduledFuture<?>> pollingFutureRef = new AtomicReference<ScheduledFuture<?>>();
 
     /**
      * Indicates if we received a signal from the user to start polling.
@@ -926,6 +928,49 @@ private final String getOwner() {
         return name;
     }
 
+    /**
+     * Dump internal state to the introspector.
+     *
+     * @param out writer for the introspector.
+     */
+    void introspect(PrintWriter out) {
+        out.println(toString() + ' ' + name + (deactivated ? " is deactivated" : ""));
+
+        out.print("  Partition ");
+        if (partitionIdLock.readLock().tryLock())
+            try {
+                out.println(partitionId);
+            } finally {
+                partitionIdLock.readLock().unlock();
+            }
+        else
+            out.println("lock temporarily unavailable");
+
+        out.println("  Config " + configRef.get());
+
+        out.println("  Accessed from " + applications);
+
+        out.println("  Signalled to poll? " + pollingStartSignalReceived);
+        out.println("  PollingManager state " + readyForPollingTask.bits);
+
+        ScheduledFuture<?> pollFuture = pollingFutureRef.get();
+        if (pollFuture != null)
+            out.println("  Next poll in " + pollFuture.getDelay(TimeUnit.MILLISECONDS) + "ms");
+
+        if (configUpdatePendingQueueLock.readLock().tryLock())
+            try {
+                out.print("  Config updates (" + configUpdatesInProgress + ") in progress, which block tasks:");
+                for (InvokerTask task : configUpdatePendingQueue)
+                    out.print(' ' + task.taskId);
+                out.println();
+            } finally {
+                configUpdatePendingQueueLock.readLock().unlock();
+            }
+
+        out.println("  In-memory list of known pending (or active) tasks: " + inMemoryTaskIds.keySet());
+        out.println();
+    }
+
     /** {@inheritDoc} */
     @Override
     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> callables) throws InterruptedException {
@@ -1924,7 +1969,7 @@ protected synchronized void unsetServerStarted(ServiceReference<ServerStarted> r
      * @param config snapshot of configuration.
      */
     private void startPollingTask(Config config) {
-        Future<?> future;
+        ScheduledFuture<?> future;
         PollingTask pollingTask = new PollingTask(config);
 
         future = scheduledExecutor.schedule(pollingTask, config.initialPollDelay, TimeUnit.MILLISECONDS);
@@ -1997,7 +2042,7 @@ public void failedCompletion(Future<Boolean> future, Throwable t) {
     /**
      * List of Tasks to execute after configuration updates have completed.
      */
-    private final LinkedList<Runnable> configUpdatePendingQueue = new LinkedList<Runnable>();
+    private final LinkedList<InvokerTask> configUpdatePendingQueue = new LinkedList<InvokerTask>();
 
     /**
      * futureMonitor used to track the outcome of a configuration update
@@ -2017,11 +2062,11 @@ protected void unsetFutureMonitor(FutureMonitor futureMonitor) {
      * If a configuration update is currently in progress add the targetRunnable to a
      * local queue to drive its run method after the configuration update(s) is complete.
      *
-     * @param targetRunnable Runnable eligible for execution.
+     * @param targetRunnable task eligible for execution.
      * @return return true if a configuration update is in progress, false if not.
      */
     @Trivial
-    boolean deferExecutionForConfigUpdate(Runnable targetRunnable) {
+    boolean deferExecutionForConfigUpdate(InvokerTask targetRunnable) {
         boolean returnValue = false;
 
         configUpdatePendingQueueLock.readLock().lock();
@@ -2099,7 +2144,7 @@ int configUpdateCompleted(String notificationName) {
         @Override
         @Trivial
         public void run() {
-            Runnable r;
+            InvokerTask r;
             for (;;) {
 
                 configUpdatePendingQueueLock.writeLock().lock();
@@ -2287,7 +2332,7 @@ public void run() {
                         long delay = config.pollInterval - TimeUnit.NANOSECONDS.toMillis(duration);
                         if (trace && tc.isDebugEnabled())
                             Tr.debug(PersistentExecutorImpl.this, tc, "Poll completed in " + duration + "ns. Next poll " + delay + "ms from now");
-                        Future<?> future;
+                        ScheduledFuture<?> future;
                         future = scheduledExecutor.schedule(this, delay, TimeUnit.MILLISECONDS);
 
                         pollingFutureRef.getAndSet(future);

dev/com.ibm.ws.concurrent.persistent/src/com/ibm/ws/concurrent/persistent/internal/PersistentExecutorIntrospector.java

@@ -0,0 +1,63 @@
+/*******************************************************************************
+ * Copyright (c) 2020 IBM Corporation and others.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ *     IBM Corporation - initial API and implementation
+ *******************************************************************************/
+package com.ibm.ws.concurrent.persistent.internal;
+
+import java.io.PrintWriter;
+import java.security.AccessController;
+
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.FrameworkUtil;
+import org.osgi.framework.ServiceReference;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.ConfigurationPolicy;
+
+import com.ibm.websphere.concurrent.persistent.PersistentExecutor;
+import com.ibm.websphere.ras.annotation.Trivial;
+import com.ibm.ws.kernel.service.util.SecureAction;
+import com.ibm.wsspi.logging.Introspector;
+
+/**
+ * Introspector for persistent executors.
+ */
+@Component(configurationPolicy = ConfigurationPolicy.IGNORE)
+public class PersistentExecutorIntrospector implements Introspector {
+    @Override
+    @Trivial
+    public String getIntrospectorName() {
+        System.out.println("getIntrospectorName");
+        return "PersistentExecutorIntrospector";
+    }
+
+    @Override
+    @Trivial
+    public String getIntrospectorDescription() {
+        System.out.println("getIntrospectorDescription");
+        return "Persistent timers/tasks diagnostics";
+    }
+
+    @Override
+    public void introspect(PrintWriter out) throws Exception {
+        SecureAction priv = AccessController.doPrivileged(SecureAction.get());
+        BundleContext bundleContext = priv.getBundleContext(FrameworkUtil.getBundle(getClass()));
+
+        for (ServiceReference<PersistentExecutor> ref : priv.getServiceReferences(bundleContext, PersistentExecutor.class, "(!(com.ibm.wsspi.resource.ResourceFactory=true))")) {
+            System.out.println("Found service reference: " + ref);
+            PersistentExecutorImpl executor = (PersistentExecutorImpl) priv.getService(bundleContext, ref);
+            executor.introspect(out);
+        }
+
+        ServiceReference<ApplicationTracker> appTrackerRef = bundleContext.getServiceReference(ApplicationTracker.class);
+        bundleContext.getService(appTrackerRef);
+
+        ApplicationTracker appTracker = priv.getService(bundleContext, ApplicationTracker.class);
+        appTracker.introspect(out);
+    }
+}

dev/com.ibm.ws.concurrent.persistent/src/com/ibm/ws/concurrent/persistent/internal/PollingManager.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2014 IBM Corporation and others.
+ * Copyright (c) 2014,2020 IBM Corporation and others.
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
  * which accompanies this distribution, and is available at
@@ -47,7 +47,7 @@ class PollingManager {
     /**
      * Combination of bits that represent if we are ready for polling.
      */
-    private final AtomicInteger bits = new AtomicInteger();
+    final AtomicInteger bits = new AtomicInteger();
 
     /**
      * Add an event without checking if we are ready to start polling.