1 files changed, 579 insertions, 0 deletions

diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/matcher/ReteEngine.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/matcher/ReteEngine.java
new file mode 100644
index 00000000..9bd499f4
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/matcher/ReteEngine.java
@@ -0,0 +1,579 @@
+/*******************************************************************************
+ * Copyright (c) 2004-2008 Gabor Bergmann and Daniel Varro
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v. 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-v20.html.
+ * 
+ * SPDX-License-Identifier: EPL-2.0
+ *******************************************************************************/
+
+package tools.refinery.viatra.runtime.rete.matcher;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+import org.apache.log4j.Logger;
+import tools.refinery.viatra.runtime.matchers.ViatraQueryRuntimeException;
+import tools.refinery.viatra.runtime.matchers.backend.IQueryBackend;
+import tools.refinery.viatra.runtime.matchers.backend.IQueryBackendFactory;
+import tools.refinery.viatra.runtime.matchers.backend.IQueryBackendHintProvider;
+import tools.refinery.viatra.runtime.matchers.backend.IQueryResultProvider;
+import tools.refinery.viatra.runtime.matchers.backend.QueryEvaluationHint;
+import tools.refinery.viatra.runtime.matchers.context.IQueryBackendContext;
+import tools.refinery.viatra.runtime.matchers.context.IQueryRuntimeContext;
+import tools.refinery.viatra.runtime.matchers.psystem.queries.PQuery;
+import tools.refinery.viatra.runtime.matchers.tuple.TupleMask;
+import tools.refinery.viatra.runtime.matchers.util.CollectionsFactory;
+import tools.refinery.viatra.runtime.rete.boundary.Disconnectable;
+import tools.refinery.viatra.runtime.rete.boundary.ReteBoundary;
+import tools.refinery.viatra.runtime.rete.construction.RetePatternBuildException;
+import tools.refinery.viatra.runtime.rete.construction.plancompiler.ReteRecipeCompiler;
+import tools.refinery.viatra.runtime.rete.index.Indexer;
+import tools.refinery.viatra.runtime.rete.network.Network;
+import tools.refinery.viatra.runtime.rete.network.NodeProvisioner;
+import tools.refinery.viatra.runtime.rete.network.ReteContainer;
+import tools.refinery.viatra.runtime.rete.traceability.RecipeTraceInfo;
+
+/**
+ * @author Gabor Bergmann
+ *
+ */
+public class ReteEngine implements IQueryBackend {
+
+    protected Network reteNet;
+    protected final int reteThreads;
+    protected ReteBoundary boundary;
+    
+    /**
+     * @since 2.2
+     */
+    protected final boolean deleteAndRederiveEvaluation;
+    /**
+     * @since 2.4
+     */
+    protected final TimelyConfiguration timelyConfiguration;
+
+    private IQueryBackendContext context;
+    private Logger logger;
+    protected IQueryRuntimeContext runtimeContext;
+
+    protected Collection<Disconnectable> disconnectables;
+
+    protected Map<PQuery, RetePatternMatcher> matchers;
+
+    protected ReteRecipeCompiler compiler;
+
+    protected final boolean parallelExecutionEnabled; // TRUE if model manipulation can go on
+
+    private boolean disposedOrUninitialized = true;
+    
+    private HintConfigurator hintConfigurator;
+
+    /**
+     * @param context
+     *            the context of the pattern matcher, conveying all information from the outside world.
+     * @param reteThreads
+     *            the number of threads to operate the RETE network with; 0 means single-threaded operation, 1 starts an
+     *            asynchronous thread to operate the RETE net, >1 uses multiple RETE containers.
+     */
+    public ReteEngine(IQueryBackendContext context, int reteThreads) {
+        this(context, reteThreads, false, null);
+    }
+    
+    /**
+     * @since 2.4
+     */
+    public ReteEngine(IQueryBackendContext context, int reteThreads, boolean deleteAndRederiveEvaluation, TimelyConfiguration timelyConfiguration) {
+        super();
+        this.context = context;
+        this.logger = context.getLogger();
+        this.runtimeContext = context.getRuntimeContext();
+        this.reteThreads = reteThreads;
+        this.parallelExecutionEnabled = reteThreads > 0;
+        this.deleteAndRederiveEvaluation = deleteAndRederiveEvaluation;
+        this.timelyConfiguration = timelyConfiguration;
+        initEngine();
+        this.compiler = null;
+    }
+    
+    /**
+     * @since 1.6
+     */
+    public IQueryBackendContext getBackendContext() {
+        return context;
+    }
+    
+    /**
+     * @since 2.2
+     */
+    public boolean isDeleteAndRederiveEvaluation() {
+        return this.deleteAndRederiveEvaluation;
+    }
+    
+    /**
+     * @since 2.4
+     */
+    public TimelyConfiguration getTimelyConfiguration() {
+        return this.timelyConfiguration;
+    }
+
+    /**
+     * initializes engine components
+     */
+    private synchronized void initEngine() {
+        this.disposedOrUninitialized = false;
+        this.disconnectables = new LinkedList<Disconnectable>();
+        // this.caughtExceptions = new LinkedBlockingQueue<Throwable>();
+
+        
+        this.hintConfigurator = new HintConfigurator(context.getHintProvider());
+        
+        this.reteNet = new Network(reteThreads, this);
+        this.boundary = new ReteBoundary(this); // prerequisite: network
+
+        this.matchers = CollectionsFactory.createMap();
+        /* this.matchersScoped = new HashMap<PatternDescription, Map<Map<Integer,Scope>,RetePatternMatcher>>(); */
+
+        // prerequisite: network, framework, boundary, disconnectables
+        //context.subscribeBackendForUpdates(this.boundary);
+        // prerequisite: boundary, disconnectables
+//        this.traceListener = context.subscribePatternMatcherForTraceInfluences(this);
+
+    }
+    
+    @Override
+    public void flushUpdates() {
+        for (ReteContainer container : this.reteNet.getContainers()) {
+            container.deliverMessagesSingleThreaded();
+        }
+    }
+
+    /**
+     * deconstructs engine components
+     */
+    private synchronized void deconstructEngine() {
+        ensureInitialized();
+        reteNet.kill();
+
+        //context.unSubscribeBackendFromUpdates(this.boundary);
+        for (Disconnectable disc : disconnectables) {
+            disc.disconnect();
+        }
+
+        this.matchers = null;
+        this.disconnectables = null;
+
+        this.reteNet = null;
+        this.boundary = null;
+        
+        this.hintConfigurator = null;
+
+        // this.machineListener = new MachineListener(this); // prerequisite:
+        // framework, disconnectables
+//        this.traceListener = null;
+
+        this.disposedOrUninitialized = true;
+    }
+
+    /**
+     * Deconstructs the engine to get rid of it finally
+     */
+    public void killEngine() {
+        deconstructEngine();
+        // this.framework = null;
+        this.compiler = null;
+        this.logger = null;
+    }
+
+    /**
+     * Resets the engine to an after-initialization phase
+     *
+     */
+    public void reset() {
+        deconstructEngine();
+
+        initEngine();
+
+        compiler.reset();
+    }
+
+    /**
+     * Accesses the patternmatcher for a given pattern, constructs one if a matcher is not available yet.
+     *
+     * @pre: builder is set.
+     * @param query
+     *            the pattern to be matched.
+     * @return a patternmatcher object that can match occurences of the given pattern.
+     * @throws ViatraQueryRuntimeException
+     *             if construction fails.
+     */
+    public synchronized RetePatternMatcher accessMatcher(final PQuery query) {
+        ensureInitialized();
+        RetePatternMatcher matcher;
+        // String namespace = gtPattern.getNamespace().getName();
+        // String name = gtPattern.getName();
+        // String fqn = namespace + "." + name;
+        matcher = matchers.get(query);
+        if (matcher == null) {
+            constructionWrapper(() -> {
+                RecipeTraceInfo prodNode;
+                prodNode = boundary.accessProductionTrace(query);
+
+                RetePatternMatcher retePatternMatcher = new RetePatternMatcher(ReteEngine.this,
+                        prodNode);
+                retePatternMatcher.setTag(query);
+                matchers.put(query, retePatternMatcher);
+                return null;
+            });
+            matcher = matchers.get(query);
+        }
+        
+        executeDelayedCommands();
+
+        return matcher;
+    }
+
+    
+    /**
+     * Constructs RETE pattern matchers for a collection of patterns, if they are not available yet. Model traversal
+     * during the whole construction period is coalesced (which may have an effect on performance, depending on the
+     * matcher context).
+     *
+     * @pre: builder is set.
+     * @param specifications
+     *            the patterns to be matched.
+     * @throws ViatraQueryRuntimeException
+     *             if construction fails.
+     */
+    public synchronized void buildMatchersCoalesced(final Collection<PQuery> specifications) {
+        ensureInitialized();
+        constructionWrapper(() -> {
+            for (PQuery specification : specifications) {
+                boundary.accessProductionNode(specification);
+            }
+            return null;
+        });
+    }
+
+    /**
+     * @since 2.4
+     */
+    public <T> T constructionWrapper(final Callable<T> payload) {
+        T result = null;
+//              context.modelReadLock();
+//                  try {
+                if (parallelExecutionEnabled)
+                    reteNet.getStructuralChangeLock().lock();
+                try {
+                    try {
+                        result = runtimeContext.coalesceTraversals(() -> { 
+                            T innerResult = payload.call(); 
+                            this.executeDelayedCommands(); 
+                            return innerResult;
+                        });
+                    } catch (InvocationTargetException ex) {
+                        final Throwable cause = ex.getCause();
+                        if (cause instanceof RetePatternBuildException)
+                            throw (RetePatternBuildException) cause;
+                        if (cause instanceof RuntimeException)
+                            throw (RuntimeException) cause;
+                        assert (false);
+                    }
+                } finally {
+                   if (parallelExecutionEnabled)
+                        reteNet.getStructuralChangeLock().unlock();
+                   reteNet.waitForReteTermination();
+                }
+//                  } finally {
+//                      context.modelReadUnLock();
+//                  }
+            return result;
+    }
+
+    // /**
+    // * Accesses the patternmatcher for a given pattern with additional scoping, constructs one if
+    // * a matcher is not available yet.
+    // *
+    // * @param gtPattern
+    // * the pattern to be matched.
+    // * @param additionalScopeMap
+    // * additional, optional scopes for the symbolic parameters
+    // * maps the position of the symbolic parameter to its additional scope (if any)
+    // * @pre: scope.parent is non-root, i.e. this is a nontrivial constraint
+    // * use the static method RetePatternMatcher.buildAdditionalScopeMap() to create from PatternCallSignature
+    // * @return a patternmatcher object that can match occurences of the given
+    // * pattern.
+    // * @throws PatternMatcherCompileTimeException
+    // * if construction fails.
+    // */
+    // public synchronized RetePatternMatcher accessMatcherScoped(PatternDescription gtPattern, Map<Integer, Scope>
+    // additionalScopeMap)
+    // throws PatternMatcherCompileTimeException {
+    // if (additionalScopeMap.isEmpty()) return accessMatcher(gtPattern);
+    //
+    // RetePatternMatcher matcher;
+    //
+    // Map<Map<Integer, Scope>, RetePatternMatcher> scopes = matchersScoped.get(gtPattern);
+    // if (scopes == null) {
+    // scopes = new HashMap<Map<Integer, Scope>, RetePatternMatcher>();
+    // matchersScoped.put(gtPattern, scopes);
+    // }
+    //
+    // matcher = scopes.get(additionalScopeMap);
+    // if (matcher == null) {
+    // context.modelReadLock();
+    // try {
+    // reteNet.getStructuralChangeLock().lock();
+    // try {
+    // Address<? extends Production> prodNode;
+    // prodNode = boundary.accessProductionScoped(gtPattern, additionalScopeMap);
+    //
+    // matcher = new RetePatternMatcher(this, prodNode);
+    // scopes.put(additionalScopeMap, matcher);
+    // } finally {
+    // reteNet.getStructuralChangeLock().unlock();
+    // }
+    // } finally {
+    // context.modelReadUnLock();
+    // }
+    // // reteNet.flushUpdates();
+    // }
+    //
+    // return matcher;
+    // }
+
+    /**
+     * Returns an indexer that groups the contents of this Production node by their projections to a given mask.
+     * Designed to be called by a RetePatternMatcher.
+     *
+     * @param production
+     *            the production node to be indexed.
+     * @param mask
+     *            the mask that defines the projection.
+     * @return the Indexer.
+     */
+    synchronized Indexer accessProjection(RecipeTraceInfo production, TupleMask mask) {
+        ensureInitialized();
+        NodeProvisioner nodeProvisioner = reteNet.getHeadContainer().getProvisioner();
+        Indexer result = nodeProvisioner.peekProjectionIndexer(production, mask);
+        if (result == null) {
+            result = constructionWrapper(() -> 
+                nodeProvisioner.accessProjectionIndexerOnetime(production, mask)
+            );
+        }
+
+        return result;
+    }
+
+    // /**
+    // * Retrieves the patternmatcher for a given pattern fqn, returns null if
+    // the matching network hasn't been constructed yet.
+    // *
+    // * @param fqn the fully qualified name of the pattern to be matched.
+    // * @return the previously constructed patternmatcher object that can match
+    // occurences of the given pattern, or null if it doesn't exist.
+    // */
+    // public RetePatternMatcher getMatcher(String fqn)
+    // {
+    // RetePatternMatcher matcher = matchersByFqn.get(fqn);
+    // if (matcher == null)
+    // {
+    // Production prodNode = boundary.getProduction(fqn);
+    //
+    // matcher = new RetePatternMatcher(this, prodNode);
+    // matchersByFqn.put(fqn, matcher);
+    // }
+    //
+    // return matcher;
+    // }
+
+    /**
+     * @since 2.3
+     */
+    public void executeDelayedCommands() {
+        for (final ReteContainer container : this.reteNet.getContainers()) {
+            container.executeDelayedCommands();
+        }
+    }
+    
+    /**
+     * Waits until the pattern matcher is in a steady state and output can be retrieved.
+     */
+    public void settle() {
+        ensureInitialized();
+        reteNet.waitForReteTermination();
+    }
+
+    /**
+     * Waits until the pattern matcher is in a steady state and output can be retrieved. When steady state is reached, a
+     * retrieval action is executed before the steady state ceases.
+     *
+     * @param action
+     *            the action to be run when reaching the steady-state.
+     */
+    public void settle(Runnable action) {
+        ensureInitialized();
+        reteNet.waitForReteTermination(action);
+    }
+
+    // /**
+    // * @return the framework
+    // */
+    // public IFramework getFramework() {
+    // return framework.get();
+    // }
+
+    /**
+     * @return the reteNet
+     */
+    public Network getReteNet() {
+       ensureInitialized();
+       return reteNet;
+    }
+
+    /**
+     * @return the boundary
+     */
+    public ReteBoundary getBoundary() {
+        ensureInitialized();
+        return boundary;
+    }
+
+    // /**
+    // * @return the pattern matcher builder
+    // */
+    // public IRetePatternBuilder getBuilder() {
+    // return builder;
+    // }
+
+    /**
+     * @param builder
+     *            the pattern matcher builder to set
+     */
+    public void setCompiler(ReteRecipeCompiler builder) {
+        ensureInitialized();
+        this.compiler = builder;
+    }
+
+//    /**
+//     * @return the manipulationListener
+//     */
+//    public IManipulationListener getManipulationListener() {
+//      ensureInitialized();
+//       return manipulationListener;
+//    }
+
+//    /**
+//     * @return the traceListener
+//     */
+//    public IPredicateTraceListener geTraceListener() {
+//      ensureInitialized();
+//        return traceListener;
+//    }
+
+    /**
+     * @param disc
+     *            the new Disconnectable adapter.
+     */
+    public void addDisconnectable(Disconnectable disc) {
+        ensureInitialized();
+        disconnectables.add(disc);
+    }
+
+    /**
+     * @return the parallelExecutionEnabled
+     */
+    public boolean isParallelExecutionEnabled() {
+        return parallelExecutionEnabled;
+    }
+
+
+    public Logger getLogger() {
+        ensureInitialized();
+        return logger;
+    }
+
+    public IQueryRuntimeContext getRuntimeContext() {
+        ensureInitialized();
+        return runtimeContext;
+    }
+
+    public ReteRecipeCompiler getCompiler() {
+        ensureInitialized();
+       return compiler;
+    }
+
+    // /**
+    // * For internal use only: logs exceptions occurring during term evaluation inside the RETE net.
+    // * @param e
+    // */
+    // public void logEvaluatorException(Throwable e) {
+    // try {
+    // caughtExceptions.put(e);
+    // } catch (InterruptedException e1) {
+    // logEvaluatorException(e);
+    // }
+    // }
+    // /**
+    // * Polls the exceptions caught and logged during term evaluation by this RETE engine.
+    // * Recommended usage: iterate polling until null is returned.
+    // *
+    // * @return the next caught exception, or null if there are no more.
+    // */
+    // public Throwable getNextLoggedEvaluatorException() {
+    // return caughtExceptions.poll();
+    // }
+
+    void ensureInitialized() {
+        if (disposedOrUninitialized)
+            throw new IllegalStateException("Trying to use a Rete engine that has been disposed or has not yet been initialized.");
+
+    }
+    
+    @Override
+    public IQueryResultProvider getResultProvider(PQuery query)  {
+        return accessMatcher(query);
+    }
+    
+    /**
+     * @since 1.4
+     */
+    @Override
+    public IQueryResultProvider getResultProvider(PQuery query, QueryEvaluationHint hints) {
+        hintConfigurator.storeHint(query, hints);
+        return accessMatcher(query);
+    }
+
+    @Override
+    public IQueryResultProvider peekExistingResultProvider(PQuery query) {
+        ensureInitialized();
+        return matchers.get(query);
+    }
+
+    @Override
+    public void dispose() {
+        killEngine();
+    }
+    
+    @Override
+    public boolean isCaching() {
+        return true;
+    }
+
+    /**
+     * @since 1.5
+     * @noreference Internal API, subject to change
+     */
+    public IQueryBackendHintProvider getHintConfiguration() {
+        return hintConfigurator;
+    }
+
+    @Override
+    public IQueryBackendFactory getFactory() {
+        return ReteBackendFactory.INSTANCE;
+    }
+
+}