test introspector when failover enabled

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/db/DatabaseTaskStore.java

@@ -16,7 +16,6 @@
 import java.sql.SQLIntegrityConstraintViolationException;
 import java.sql.SQLTimeoutException;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -840,38 +839,6 @@ public Long getPartition(long taskId) throws Exception {
         return partitionId;
     }
 
-    /** {@inheritDoc} */
-    @Override
-    public Long getPartitionWithState(long stateBits) throws Exception {
-        String select = "SELECT p.ID,p.EXECUTOR,p.HOSTNAME,p.ID,p.LSERVER,p.USERDIR,p.EXPIRY,p.STATES FROM Partition p WHERE p.STATES-p.STATES/:d*:d=:r AND p.EXPIRY>:t ORDER BY p.EXPIRY DESC";
-
-        final boolean trace = TraceComponent.isAnyTracingEnabled();
-        if (trace && tc.isEntryEnabled())
-            Tr.entry(this, tc, "getPartitionWithState", stateBits, select);
-
-        long denominator = stateBits < 2 ? 2 : stateBits < 4 ? 4 : -1;
-        if (denominator == -1)
-            throw new IllegalArgumentException(Long.toString(stateBits)); // internal error: no states > 3 are currently defined
-
-        Object[] partitionInfo;
-        EntityManager em = getPersistenceServiceUnit().createEntityManager();
-        try {
-            TypedQuery<Object[]> query = em.createQuery(select.toString(), Object[].class);
-            query.setParameter("d", denominator);
-            query.setParameter("r", stateBits);
-            query.setParameter("t", System.currentTimeMillis());
-            query.setMaxResults(1);
-            List<Object[]> results = query.getResultList();
-            partitionInfo = results == null || results.isEmpty() ? null : results.get(0);
-        } finally {
-            em.close();
-        }
-
-        if (trace && tc.isEntryEnabled())
-            Tr.exit(this, tc, "getPartitionWithState", partitionInfo == null ? null : Arrays.toString(partitionInfo));
-        return partitionInfo == null ? null : (Long) partitionInfo[0];
-    }
-
     /**
      * Returns the persistence service unit, lazily initializing if necessary.
      *

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,70 @@
+/*******************************************************************************
+ * 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() {
+        return "PersistentExecutorIntrospector";
+    }
+
+    @Override
+    @Trivial
+    public String 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))")) {
+            PersistentExecutorImpl executor = (PersistentExecutorImpl) priv.getService(bundleContext, ref);
+            if (executor == null) {
+                String displayId = (String) ref.getProperty("config.displayId");
+                String name = displayId.contains("]/persistentExecutor[") ? displayId : (String) ref.getProperty("id");
+                if (name == null)
+                    name = (String) ref.getProperty("jndiName");
+                out.println("PersistentExecutor " + name + " is not available");
+                out.println("Properties: " + ref.getProperties());
+                out.println();
+            } else {
+                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.