GH2701: Fuseki Mod to list and abort running executions.

Author: Aklakan

jena-arq/src/main/java/org/apache/jena/sparql/exec/tracker/DatasetGraphWithExecTracker.java

@@ -0,0 +1,28 @@
+package org.apache.jena.sparql.exec.tracker;
+
+import org.apache.jena.sparql.core.DatasetGraph;
+import org.apache.jena.sparql.core.DatasetGraphWrapper;
+import org.apache.jena.sparql.core.DatasetGraphWrapperView;
+import org.apache.jena.sparql.util.Context;
+
+public class DatasetGraphWithExecTracker
+    extends DatasetGraphWrapper
+    implements DatasetGraphWrapperView
+{
+    public static DatasetGraph wrap(DatasetGraph dsg) {
+        DatasetGraph result;
+        if (dsg instanceof DatasetGraphWithExecTracker) {
+            result = dsg;
+        } else {
+            // Put an exec tracker into the dataset's context.
+            Context context = dsg.getContext();
+            ExecTracker.ensureTracker(context);
+            result = new DatasetGraphWithExecTracker(dsg);
+        }
+        return result;
+    }
+
+    protected DatasetGraphWithExecTracker(DatasetGraph dsg) {
+        super(dsg);
+    }
+}

jena-arq/src/main/java/org/apache/jena/sparql/exec/tracker/ExecTracker.java

@@ -0,0 +1,153 @@
+package org.apache.jena.sparql.exec.tracker;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.Iterator;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentNavigableMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.jena.sparql.SystemARQ;
+import org.apache.jena.sparql.util.Context;
+import org.apache.jena.sparql.util.Symbol;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ExecTracker {
+    private static final Logger logger = LoggerFactory.getLogger(ExecTracker.class);
+
+    public record StartRecord(long requestId, Instant timestamp, Object requestObject, Runnable abortAction) {}
+
+    public record CompletionRecord(StartRecord start, Instant timestamp, Throwable throwable) {
+        public Duration duration() {
+            return Duration.between(start.timestamp, timestamp);
+        }
+
+        public boolean isSuccess() {
+            return throwable == null;
+        }
+    }
+
+    private Set<ExecTrackerListener> eventListeners = Collections.synchronizedSet(Collections.newSetFromMap(new IdentityHashMap<>()));
+
+    protected AtomicLong nextId = new AtomicLong();
+    protected ConcurrentNavigableMap<Long, StartRecord> idToStartRecord = new ConcurrentSkipListMap<>();
+    protected int maxHistorySize = 1000;
+    protected ConcurrentNavigableMap<Long, CompletionRecord> history = new ConcurrentSkipListMap<>();
+
+    public ConcurrentNavigableMap<Long, StartRecord> getActiveTasks() {
+        return idToStartRecord;
+    }
+
+    public ConcurrentNavigableMap<Long, CompletionRecord> getHistory() {
+        return history;
+    }
+
+    public void setMaxHistorySize(int maxHistorySize) {
+        this.maxHistorySize = maxHistorySize;
+    }
+
+    public long put(Object requestObject, Runnable abortAction) {
+        long result = nextId.getAndIncrement();
+        StartRecord record = new StartRecord(result, Instant.now(), requestObject, abortAction);
+        idToStartRecord.put(result, record);
+        broadcastStartEvent(record);
+        return result;
+    }
+
+    protected void trimHistory() {
+        if (history.size() >= maxHistorySize) {
+            Iterator<Entry<Long, CompletionRecord>> it = history.entrySet().iterator();
+            while (history.size() >= maxHistorySize && it.hasNext()) {
+                it.next();
+                it.remove();
+            }
+        }
+    }
+
+    public CompletionRecord remove(long id, Throwable t) {
+        StartRecord startRecord = idToStartRecord.remove(id);
+        CompletionRecord result = null;
+        if (startRecord != null) {
+            trimHistory();
+            Instant now = Instant.now();
+            result = new CompletionRecord(startRecord, now, t);
+            // long requestId = startRecord.requestId();
+            // history.put(now, result);
+            history.put(id, result);
+            broadcastCompletionEvent(result);
+        }
+        return result;
+    }
+
+    protected void broadcastStartEvent(StartRecord startRecord) {
+        for (ExecTrackerListener listener : eventListeners) {
+            try {
+                listener.onStart(startRecord);
+            } catch (Throwable t) {
+                if (logger.isWarnEnabled()) {
+                    logger.warn("Failure during event handler.", t);
+                }
+            }
+        }
+    }
+
+    protected void broadcastCompletionEvent(CompletionRecord completionRecord) {
+        for (ExecTrackerListener listener : eventListeners) {
+            try {
+                listener.onComplete(completionRecord);
+            } catch (Throwable t) {
+                if (logger.isWarnEnabled()) {
+                    logger.warn("Failure during event handler.", t);
+                }
+            }
+        }
+    }
+
+    public Runnable addListener(ExecTrackerListener listener) {
+        Objects.requireNonNull(listener);
+        eventListeners.add(listener);
+        return () -> eventListeners.remove(listener);
+    }
+
+    public Set<ExecTrackerListener> getEventListeners() {
+        return eventListeners;
+    }
+
+    @Override
+    public String toString() {
+        return "Active: " + idToStartRecord.size() + ", History: " + history.size() + "/" + maxHistorySize;
+    }
+
+    public static final Symbol symTracker = SystemARQ.allocSymbol("execTracker");
+
+    public static ExecTracker getTracker(Context context) {
+        return context.get(symTracker);
+    }
+
+    public static ExecTracker requireTracker(Context context) {
+        ExecTracker result = getTracker(context);
+        Objects.requireNonNull("No ExecTracker registered in context");
+        return result;
+    }
+
+    public static ExecTracker ensureTracker(Context context) {
+        // FIXME The spatial index PR adds an atomic context.compute method.
+        ExecTracker result = context.get(symTracker);
+        if (result == null) {
+            synchronized (context) {
+                result = context.get(symTracker);
+                if (result == null) {
+                    result = new ExecTracker();
+                    context.set(symTracker, result);
+                }
+            }
+        }
+        return result;
+    }
+}

jena-arq/src/main/java/org/apache/jena/sparql/exec/tracker/ExecTrackerListener.java

@@ -0,0 +1,9 @@
+package org.apache.jena.sparql.exec.tracker;
+
+import org.apache.jena.sparql.exec.tracker.ExecTracker.CompletionRecord;
+import org.apache.jena.sparql.exec.tracker.ExecTracker.StartRecord;
+
+public interface ExecTrackerListener {
+    void onStart(StartRecord startRecord);
+    void onComplete(CompletionRecord endRecord);
+}

jena-arq/src/main/java/org/apache/jena/sparql/exec/tracker/InitExecTracker.java

@@ -0,0 +1,37 @@
+package org.apache.jena.sparql.exec.tracker;
+
+import org.apache.jena.sparql.engine.QueryEngineRegistry;
+import org.apache.jena.sparql.modify.UpdateEngineRegistry;
+import org.apache.jena.sys.JenaSubsystemLifecycle;
+
+public class InitExecTracker
+    implements JenaSubsystemLifecycle
+{
+    @Override
+    public void start() {
+        QueryEngineRegistry queryReg = QueryEngineRegistry.get();
+        init(queryReg);
+
+        UpdateEngineRegistry updateReg = UpdateEngineRegistry.get();
+        init(updateReg);
+    }
+
+    @Override
+    public void stop() {
+    }
+
+    public static void init(QueryEngineRegistry reg) {
+        reg.add(new QueryEngineFactoryExecTracker());
+    }
+
+    public static void init(UpdateEngineRegistry reg) {
+        reg.add(new UpdateEngineFactoryExecTracker());
+    }
+
+    @Override
+    public int level() {
+        // Register the 'wrapper engine factories' late
+        // such that upon execution they are consulted early.
+        return 1_000_000;
+    }
+}

jena-arq/src/main/java/org/apache/jena/sparql/exec/tracker/PlanWrapper.java

@@ -0,0 +1,44 @@
+package org.apache.jena.sparql.exec.tracker;
+
+import org.apache.jena.atlas.io.IndentedWriter;
+import org.apache.jena.shared.PrefixMapping;
+import org.apache.jena.sparql.algebra.Op;
+import org.apache.jena.sparql.engine.Plan;
+import org.apache.jena.sparql.engine.QueryIterator;
+import org.apache.jena.sparql.serializer.SerializationContext;
+
+public interface PlanWrapper
+    extends Plan
+{
+    Plan getDelegate();
+
+    @Override
+    default void output(IndentedWriter out, SerializationContext sCxt) {
+        getDelegate().output(out, sCxt);
+    }
+
+    @Override
+    default String toString(PrefixMapping pmap) {
+        return getDelegate().toString(pmap);
+    }
+
+    @Override
+    default void output(IndentedWriter out) {
+        getDelegate().output(out);
+    }
+
+    @Override
+    default void close() {
+        getDelegate().close();
+    }
+
+    @Override
+    default Op getOp() {
+        return getDelegate().getOp();
+    }
+
+    @Override
+    default QueryIterator iterator() {
+        return getDelegate().iterator();
+    }
+}

jena-arq/src/main/java/org/apache/jena/sparql/exec/tracker/PlanWrapperBase.java

@@ -0,0 +1,19 @@
+package org.apache.jena.sparql.exec.tracker;
+
+import org.apache.jena.sparql.engine.Plan;
+
+public class PlanWrapperBase
+    implements PlanWrapper
+{
+    protected Plan delegate;
+
+    public PlanWrapperBase(Plan delegate) {
+        super();
+        this.delegate = delegate;
+    }
+
+    @Override
+    public Plan getDelegate() {
+        return delegate;
+    }
+}