[2201.2.x] Add ability to get strand dump during code execution

bvm/ballerina-runtime/src/main/java/io/ballerina/runtime/internal/launch/LaunchUtils.java

@@ -29,10 +29,13 @@
 import io.ballerina.runtime.internal.configurable.providers.toml.TomlDetails;
 import io.ballerina.runtime.internal.configurable.providers.toml.TomlFileProvider;
 import io.ballerina.runtime.internal.diagnostics.RuntimeDiagnosticLog;
+import io.ballerina.runtime.internal.troubleshoot.StrandDump;
 import io.ballerina.runtime.internal.util.RuntimeUtils;
 import org.apache.commons.lang3.ArrayUtils;
+import sun.misc.Signal;
 
 import java.io.File;
+import java.io.PrintStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -54,9 +57,24 @@
  */
 public class LaunchUtils {
 
+    private static final PrintStream outStream = System.out;
+
     private LaunchUtils() {
     }
 
+    public static void startListenersAndSignalHandler(boolean isService) {
+        // starts all listeners
+        startListeners(isService);
+
+        // start TRAP signal handler which produces the strand dump
+        try {
+            Signal.handle(new Signal("TRAP"), signal -> outStream.println(StrandDump.getStrandDump()));
+        } catch (IllegalArgumentException ignored) {
+            // In some Operating Systems like Windows, "TRAP" POSIX signal is not supported.
+            // There getting the strand dump using kill signals is not expected, hence this exception is ignored.
+        }
+    }
+
     public static void startListeners(boolean isService) {
         ServiceLoader<LaunchListener> listeners = ServiceLoader.load(LaunchListener.class);
         listeners.forEach(listener -> listener.beforeRunProgram(isService));

bvm/ballerina-runtime/src/main/java/io/ballerina/runtime/internal/scheduling/FunctionFrame.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright (c) 2022, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+ *
+ * WSO2 Inc. licenses this file to you under the Apache License,
+ * Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.ballerina.runtime.internal.scheduling;
+
+/**
+ * This interface represents the function frame which saves the existing
+ * state when a function yields.
+ *
+ * @since 2201.2.0
+ */
+public interface FunctionFrame {
+
+    String getYieldLocation();
+
+    String getYieldStatus();
+
+}

bvm/ballerina-runtime/src/main/java/io/ballerina/runtime/internal/scheduling/Scheduler.java

@@ -32,11 +32,13 @@
 import io.ballerina.runtime.internal.values.FutureValue;
 
 import java.io.PrintStream;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 import java.util.Stack;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -55,7 +57,7 @@
  */
 public class Scheduler {
 
-    private static PrintStream err = System.err;
+    private static final PrintStream err = System.err;
 
     /**
      * Scheduler does not get killed if the immortal value is true. Specific to services.
@@ -65,12 +67,13 @@ public class Scheduler {
     /**
      * Strands that are ready for execution.
      */
-    private BlockingQueue<ItemGroup> runnableList = new LinkedBlockingDeque<>();
+    private final BlockingQueue<ItemGroup> runnableList = new LinkedBlockingDeque<>();
 
     private static final ThreadLocal<StrandHolder> strandHolder = ThreadLocal.withInitial(StrandHolder::new);
+    private static final ConcurrentHashMap<Integer, Strand> currentStrands = new ConcurrentHashMap<>();
     private final Strand previousStrand;
 
-    private AtomicInteger totalStrands = new AtomicInteger();
+    private final AtomicInteger totalStrands = new AtomicInteger();
 
     private static String poolSizeConf = System.getenv(RuntimeConstants.BALLERINA_MAX_POOL_SIZE_ENV_VAR);
 
@@ -112,6 +115,10 @@ public static Strand getStrandNoException() {
         return strandHolder.get().strand;
     }
 
+    public static Map<Integer, Strand> getCurrentStrands() {
+        return new HashMap<>(currentStrands);
+    }
+
     /**
      * Schedules given function by creating a new strand group.
      *
@@ -456,6 +463,8 @@ private void cleanUp(Strand justCompleted) {
         justCompleted.scheduler = null;
         justCompleted.frames = null;
         justCompleted.waitingContexts = null;
+
+        currentStrands.remove(justCompleted.getId());
         //TODO: more cleanup , eg channels
     }
 
@@ -503,13 +512,15 @@ private void addToRunnableList(SchedulerItem item, ItemGroup group) {
     public FutureValue createFuture(Strand parent, Callback callback, Map<String, Object> properties,
                                     Type constraint, String name, StrandMetadata metadata) {
         Strand newStrand = new Strand(name, metadata, this, parent, properties);
+        currentStrands.put(newStrand.getId(), newStrand);
         return createFuture(parent, callback, constraint, newStrand);
     }
 
     public FutureValue createTransactionalFuture(Strand parent, Callback callback, Map<String, Object> properties,
                                     Type constraint, String name, StrandMetadata metadata) {
         Strand newStrand = new Strand(name, metadata, this, parent, properties, parent != null ?
                 parent.currentTrxContext : null);
+        currentStrands.put(newStrand.getId(), newStrand);
         return createFuture(parent, callback, constraint, newStrand);
     }
 
@@ -633,6 +644,10 @@ public String toString() {
  */
 class ItemGroup {
 
+    private static final AtomicInteger nextItemGroupId = new AtomicInteger(0);
+
+    private final int id;
+
     /**
      * Keep the list of items that should run on same thread.
      * Using a stack to get advantage of the locality.
@@ -649,10 +664,12 @@ class ItemGroup {
     public static final ItemGroup POISON_PILL = new ItemGroup();
 
     public ItemGroup(SchedulerItem item) {
+        this();
         items.push(item);
     }
 
     public ItemGroup() {
+        this.id = nextItemGroupId.incrementAndGet();
     }
 
     public void add(SchedulerItem item) {
@@ -670,4 +687,12 @@ public void lock() {
     public void unlock() {
         this.groupLock.unlock();
     }
+
+    public int getId() {
+        return id;
+    }
+
+    public boolean isScheduled() {
+        return scheduled.get();
+    }
 }

bvm/ballerina-runtime/src/main/java/io/ballerina/runtime/internal/scheduling/Strand.java

@@ -32,6 +32,7 @@
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.ConcurrentModificationException;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -46,6 +47,7 @@
 import static io.ballerina.runtime.api.constants.RuntimeConstants.CURRENT_TRANSACTION_CONTEXT_PROPERTY;
 import static io.ballerina.runtime.internal.scheduling.State.BLOCK_AND_YIELD;
 import static io.ballerina.runtime.internal.scheduling.State.BLOCK_ON_AND_YIELD;
+import static io.ballerina.runtime.internal.scheduling.State.DONE;
 import static io.ballerina.runtime.internal.scheduling.State.RUNNABLE;
 import static io.ballerina.runtime.internal.scheduling.State.YIELD;
 
@@ -57,13 +59,13 @@
 
 public class Strand {
 
-    private static AtomicInteger nextStrandId = new AtomicInteger(0);
+    private static final AtomicInteger nextStrandId = new AtomicInteger(0);
 
-    private int id;
-    private String name;
-    private StrandMetadata metadata;
+    private final int id;
+    private final String name;
+    private final StrandMetadata metadata;
 
-    public Stack<Object> frames;
+    public Stack<FunctionFrame> frames;
     public int resumeIndex;
     public Object returnValue;
     public BError panic;
@@ -397,6 +399,14 @@ public int getId() {
         return id;
     }
 
+    public int getStrandGroupId() {
+        return strandGroup.getId();
+    }
+
+    public boolean isStrandGroupScheduled() {
+        return strandGroup.isScheduled();
+    }
+
     /**
      * Gets the strand name. This will be optional. Strand name can be either name given in strand annotation or async
      * call or function pointer variable name.
@@ -416,6 +426,70 @@ public StrandMetadata getMetadata() {
         return metadata;
     }
 
+    public String dumpState() {
+        StringBuilder strandInfo = new StringBuilder("\tstrand " + this.id);
+        if (this.name != null && this.getName().isPresent()) {
+            strandInfo.append(" \"").append(this.getName().get()).append("\"");
+        }
+
+        strandInfo.append(" [");
+        strandInfo.append(this.metadata.getModuleOrg()).append(".").append(this.metadata.getModuleName()).append(".")
+                .append(this.metadata.getModuleVersion()).append(":").append(this.metadata.getParentFunctionName());
+        if (this.parent != null) {
+            strandInfo.append("][").append(this.parent.getId());
+        }
+        strandInfo.append("] [");
+
+        String closingBracketWithNewLines = "]\n\n";
+        if (this.isYielded()) {
+            getInfoFromYieldedState(strandInfo, closingBracketWithNewLines);
+        } else if (this.getState().equals(DONE)) {
+            strandInfo.append(DONE).append(closingBracketWithNewLines);
+        } else {
+            strandInfo.append(RUNNABLE).append(closingBracketWithNewLines);
+        }
+
+        return strandInfo.toString();
+    }
+
+    private void getInfoFromYieldedState(StringBuilder strandInfo, String closingBracketWithNewLines) {
+        Stack<FunctionFrame> strandFrames = this.frames;
+        if ((strandFrames == null) || (strandFrames.isEmpty())) {
+            // this means the strand frames is changed, hence the state is runnable
+            strandInfo.append(RUNNABLE).append(closingBracketWithNewLines);
+            return;
+        }
+
+        StringBuilder frameStackTrace = new StringBuilder();
+        String stringPrefix = "\t\tat\t";
+        String yieldStatus = "BLOCKED";
+        boolean noPickedYieldStatus = true;
+        try {
+            for (FunctionFrame frame : strandFrames) {
+                if (noPickedYieldStatus) {
+                    yieldStatus = frame.getYieldStatus();
+                    noPickedYieldStatus = false;
+                }
+                String yieldLocation = frame.getYieldLocation();
+                frameStackTrace.append(stringPrefix).append(yieldLocation);
+                frameStackTrace.append("\n");
+                stringPrefix = "\t\t  \t";
+            }
+        } catch (ConcurrentModificationException ce) {
+            // this exception can be thrown when frames get added or removed while it is being iterated
+            // that means now the strand state is changed from yielded state to runnable state
+            strandInfo.append(RUNNABLE).append(closingBracketWithNewLines);
+            return;
+        }
+        if (!this.isYielded() || noPickedYieldStatus) {
+            // if frames have got empty, noPickedYieldStatus is true, then the state has changed to runnable
+            strandInfo.append(RUNNABLE).append(closingBracketWithNewLines);
+            return;
+        }
+        strandInfo.append(yieldStatus).append("]:\n");
+        strandInfo.append(frameStackTrace).append("\n");
+    }
+
     /**
      * Class to hold flush action related details.
      *

bvm/ballerina-runtime/src/main/java/io/ballerina/runtime/internal/troubleshoot/StrandDump.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright (c) 2022, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+ *
+ * WSO2 Inc. licenses this file to you under the Apache License,
+ * Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.ballerina.runtime.internal.troubleshoot;
+
+import io.ballerina.runtime.internal.scheduling.Scheduler;
+import io.ballerina.runtime.internal.scheduling.Strand;
+
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Used to get the status of current Ballerina strands.
+ *
+ * @since 2201.2.0
+ */
+public class StrandDump {
+
+    public static String getStrandDump() {
+        Map<Integer, Strand> availableStrands = Scheduler.getCurrentStrands();
+        int availableStrandCount = availableStrands.size();
+        Map<Integer, List<String>> availableStrandGroups = new HashMap<>();
+        populateAvailableStrandGroups(availableStrands, availableStrandGroups);
+
+        StringBuilder infoStr = new StringBuilder("Ballerina Strand Dump [");
+        DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy/MM/dd HH:mm:ss");
+        LocalDateTime localDateTime = LocalDateTime.now();
+        infoStr.append(dateTimeFormatter.format(localDateTime));
+        infoStr.append("]\n===========================================\n\n");
+        infoStr.append("Current no. of strand groups\t:\t").append(availableStrandGroups.size()).append("\n");
+        infoStr.append("Current no. of strands      \t:\t").append(availableStrandCount).append("\n\n");
+        availableStrandGroups.forEach((strandGroupId, strandList) -> {
+            infoStr.append("group ").append(strandGroupId).append(" [").append(strandList.get(0)).append("]: [")
+                    .append(strandList.size() - 1).append("]\n");
+            strandList.subList(1, strandList.size()).forEach(infoStr::append);
+        });
+        infoStr.append("===========================================\n");
+        cleanUp(availableStrands, availableStrandGroups);
+        return infoStr.toString();
+    }
+
+    private static void populateAvailableStrandGroups(Map<Integer, Strand> availableStrands,
+                                                      Map<Integer, List<String>> availableStrandGroups) {
+        for (Strand strand : availableStrands.values()) {
+            int strandGroupId = strand.getStrandGroupId();
+            String strandState = strand.dumpState();
+            availableStrandGroups.computeIfAbsent(strandGroupId, k -> {
+                ArrayList<String> strandDataList = new ArrayList<>();
+                strandDataList.add(getStrandGroupStatus(strand.isStrandGroupScheduled()));
+                return strandDataList;
+            }).add(strandState);
+        }
+    }
+
+    private static void cleanUp(Map<Integer, Strand> availableStrands,
+                                Map<Integer, List<String>> availableStrandGroups) {
+        availableStrands.clear();
+        availableStrandGroups.clear();
+    }
+
+    private static String getStrandGroupStatus(boolean isStrandGroupScheduled) {
+        if (isStrandGroupScheduled) {
+            return "RUNNABLE";
+        } else {
+            return "QUEUED";
+        }
+    }
+
+    private StrandDump() {}
+ }