Merge pull request JanusGraph#897 from dpitera/graph-cache-evictions

Evict graph from cache upon configuration updates

docs/configuredgraphfactory.adoc

@@ -85,7 +85,7 @@ for which you have instantiated _and_ the references are stored inside the `Janu
 
 IMPORTANT: This is an irreversible operation that will delete all graph and index data.
 
-IMPORTANT: To ensure all graph representations are consistent across all JanusGraph nodes in your cluster, remove the graph from the `JanusGraphManager` graph reference tracker on all nodes in your cluster: `ConfiguredGraphFactory.close("graphName");`.
+IMPORTANT: To ensure all graph representations are consistent across all JanusGraph nodes in your cluster, this removes the graph from the `JanusGraphManager` graph cache on every node in the cluster, assuming each node has been properly configured to use the `JanusGraphManager`.
 
 [[configuring-JanusGraph-server-for-configuredgraphfactory]]
 === Configuring JanusGraph Server for ConfiguredGraphFactory
@@ -210,16 +210,7 @@ the property `graph.graphname` go through the `JanusGraphManager` which
 keeps track of graph references created on the given JVM. Think of it as
 a graph cache. For this reason:
 
-IMPORTANT: Any updates to a configuration are not guaranteed to take effect until
-you remove the graph in question on every JanusGraph node in your
-cluster.
-
-You can do so by calling:
-
-[source, gremlin]
-----
-ConfiguredGraphFactory.close("graph2");
-----
+IMPORTANT: Any updates to a graph configuration results in the eviction of the relevant graph from the graph cache on every node in the JanusGraph cluster, assuming each node has been configured properly to use the `JanusGraphManager`.
 
 Since graphs created using the template configuration first create a
 configuration for that graph in question using a copy and create method,
@@ -232,8 +223,7 @@ configuration are not guaranteed to take effect on the specific graph
 until:
 
 1. The relevant configuration is removed: `ConfiguredGraphFactory.removeConfiguration("graph2");`
-2. The graph in question has been closed on every JanusGraph node: `ConfiguredGraphFactory.close("graph2");`
-3. The graph is recreated using the template configuration: `ConfiguredGraphFactory.create("graph2");`
+2. The graph is recreated using the template configuration: `ConfiguredGraphFactory.create("graph2");`
 ====
 
 [[update-examples]]
@@ -259,9 +249,6 @@ map.put("storage.hostname", "10.0.0.1");
 ConfiguredGraphFactory.updateConfiguration("graph1",
 map);
 
-// Close graph
-ConfiguredGraphFactory.close("graph1");
-
 // We are now guaranteed to use the updated configuration
 def g1 = ConfiguredGraphFactory.open("graph1");
 ----
@@ -287,9 +274,6 @@ map.put("index.search.elasticsearch.transport-scheme", "http");
 ConfiguredGraphFactory.updateConfiguration("graph1",
 map);
 
-// Close graph
-ConfiguredGraphFactory.close("graph1");
-
 // We are now guaranteed to use the updated configuration
 def g1 = ConfiguredGraphFactory.open("graph1");
 ----
@@ -317,9 +301,6 @@ MapConfiguration(map));
 // Remove Configuration
 ConfiguredGraphFactory.removeConfiguration("graph1");
 
-// Close graph on all JanusGraph nodes
-ConfiguredGraphFactory.close("graph1");
-
 // Recreate
 ConfiguredGraphFactory.create("graph1");
 // Now this graph's configuration is guaranteed to be updated

janusgraph-core/src/main/java/org/janusgraph/core/ConfiguredGraphFactory.java

@@ -16,6 +16,7 @@
 
 import org.janusgraph.graphdb.management.ConfigurationManagementGraph;
 import org.janusgraph.graphdb.management.JanusGraphManager;
+import org.janusgraph.graphdb.database.management.ManagementSystem;
 import org.janusgraph.graphdb.database.StandardJanusGraph;
 import org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration;
 import org.janusgraph.diskstorage.BackendException;
@@ -34,6 +35,8 @@
 import java.util.Set;
 import java.util.Objects;
 import java.util.stream.Collectors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * This class provides static methods to: 1) create graph references denoted by a
@@ -53,6 +56,9 @@
  */
 public class ConfiguredGraphFactory {
 
+    private static final Logger log =
+    LoggerFactory.getLogger(ConfiguredGraphFactory.class);
+
     /**
      * Creates a {@link JanusGraph} configuration stored in the {@ConfigurationGraphManagament}
      * graph and a {@link JanusGraph} graph reference according to the single
@@ -152,7 +158,7 @@ public static JanusGraph close(String graphName) throws Exception {
     public static void drop(String graphName) throws BackendException, ConfigurationManagementGraphNotEnabledException, Exception {
         final StandardJanusGraph graph = (StandardJanusGraph) ConfiguredGraphFactory.close(graphName);
         JanusGraphFactory.drop(graph);
-        ConfigurationManagementGraph.getInstance().removeConfiguration(graphName);
+        removeConfiguration(graphName);
     }
 
     private static ConfigurationManagementGraph getConfigGraphManagementInstance() {
@@ -197,6 +203,14 @@ public static void createTemplateConfiguration(final Configuration config) {
      */
     public static void updateConfiguration(final String graphName, final Configuration config) {
         final ConfigurationManagementGraph configManagementGraph = getConfigGraphManagementInstance();
+        try {
+            final JanusGraph graph = open(graphName);
+            removeGraphFromCache(graph);
+        } catch (Exception e) {
+            // cannot open graph, do nothing
+            log.error(String.format("Failed to open graph %s with the following error:\n %s.\n" +
+            "Thus, it and its traversal will not be bound on this server.", graphName, e.toString()));
+        }
         configManagementGraph.updateConfiguration(graphName, config);
     }
 
@@ -217,9 +231,26 @@ public static void updateTemplateConfiguration(final Configuration config) {
      */
     public static void removeConfiguration(final String graphName) {
         final ConfigurationManagementGraph configManagementGraph = getConfigGraphManagementInstance();
+        try {
+            final JanusGraph graph = open(graphName);
+            removeGraphFromCache(graph);
+        } catch (Exception e) {
+            // cannot open graph, do nothing
+            log.error(String.format("Failed to open graph %s with the following error:\n %s.\n" +
+            "Thus, it and its traversal will not be bound on this server.", graphName, e.toString()));
+        }
         configManagementGraph.removeConfiguration(graphName);
     }
 
+    private static void removeGraphFromCache(final JanusGraph graph) {
+        final JanusGraphManager jgm = JanusGraphManagerUtility.getInstance();
+        Preconditions.checkState(jgm != null, JANUS_GRAPH_MANAGER_EXPECTED_STATE_MSG);
+        jgm.removeGraph(((StandardJanusGraph) graph).getGraphName());
+        final ManagementSystem mgmt = (ManagementSystem) graph.openManagement();
+        mgmt.evictGraphFromCache();
+        mgmt.commit();
+    }
+
     /**
      * Remove template configuration
      */

janusgraph-core/src/main/java/org/janusgraph/graphdb/configuration/GraphDatabaseConfiguration.java

@@ -1775,7 +1775,7 @@ public Backend getBackend() {
     }
 
     public String getGraphName() {
-        return getConfigurationAtOpen().getString(GRAPH_NAME.toString());
+        return getConfigurationAtOpen().getString(GRAPH_NAME.toStringWithoutRoot());
     }
 
     public StoreFeatures getStoreFeatures() {

janusgraph-core/src/main/java/org/janusgraph/graphdb/database/management/GraphCacheEvictionAction.java

@@ -0,0 +1,22 @@
+// Copyright 2018 JanusGraph Authors
+//
+// Licensed 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 org.janusgraph.graphdb.database.management;
+
+public enum GraphCacheEvictionAction {
+
+    EVICT,
+    DO_NOT_EVICT
+}
+

janusgraph-core/src/main/java/org/janusgraph/graphdb/database/management/ManagementLogger.java

@@ -31,7 +31,10 @@
 import org.janusgraph.graphdb.database.idhandling.VariableLong;
 import org.janusgraph.graphdb.database.serialize.DataOutput;
 import org.janusgraph.graphdb.database.serialize.Serializer;
+import org.janusgraph.graphdb.management.JanusGraphManager;
 import org.janusgraph.graphdb.types.vertices.JanusGraphSchemaVertex;
+import static org.janusgraph.graphdb.database.management.GraphCacheEvictionAction.EVICT;
+import static org.janusgraph.graphdb.database.management.GraphCacheEvictionAction.DO_NOT_EVICT;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -89,7 +92,9 @@ public void read(Message message) {
                     long typeId = VariableLong.readPositive(in);
                     schemaCache.expireSchemaElement(typeId);
                 }
-                Thread ack = new Thread(new SendAckOnTxClose(evictionId, senderId, graph.getOpenTransactions()));
+                final GraphCacheEvictionAction action = serializer.readObjectNotNull(in, GraphCacheEvictionAction.class);
+                Preconditions.checkNotNull(action);
+                final Thread ack = new Thread(new SendAckOnTxClose(evictionId, senderId, graph.getOpenTransactions(), action, graph.getGraphName()));
                 ack.setDaemon(true);
                 ack.start();
                 break;
@@ -115,6 +120,7 @@ public void read(Message message) {
     }
 
     public void sendCacheEviction(Set<JanusGraphSchemaVertex> updatedTypes,
+                                             final boolean evictGraphFromCache,
                                              List<Callable<Boolean>> updatedTypeTriggers,
                                              Set<String> openInstances) {
         Preconditions.checkArgument(!openInstances.isEmpty());
@@ -128,6 +134,11 @@ public void sendCacheEviction(Set<JanusGraphSchemaVertex> updatedTypes,
             assert type.hasId();
             VariableLong.writePositive(out,type.longId());
         }
+        if (evictGraphFromCache) {
+            out.writeObjectNotNull(EVICT);
+        } else {
+            out.writeObjectNotNull(DO_NOT_EVICT);
+        }
         sysLog.add(out.getStaticBuffer());
     }
 
@@ -197,11 +208,19 @@ private class SendAckOnTxClose implements Runnable {
         private final long evictionId;
         private final Set<? extends JanusGraphTransaction> openTx;
         private final String originId;
+        private final GraphCacheEvictionAction action;
+        private final String graphName;
 
-        private SendAckOnTxClose(long evictionId, String originId, Set<? extends JanusGraphTransaction> openTx) {
+        private SendAckOnTxClose(long evictionId,
+                                 String originId,
+                                 Set<? extends JanusGraphTransaction> openTx,
+                                 GraphCacheEvictionAction action,
+                                 String graphName) {
             this.evictionId = evictionId;
             this.openTx = openTx;
             this.originId = originId;
+            this.action = action;
+            this.graphName = graphName;
         }
 
         @Override
@@ -218,12 +237,24 @@ public void run() {
                         txStillOpen = true;
                     }
                 }
-                if (!txStillOpen) {
+                final JanusGraphManager jgm = JanusGraphManager.getInstance();
+                final boolean janusGraphManagerIsInBadState = null == jgm && action.equals(EVICT);
+                if (!txStillOpen && janusGraphManagerIsInBadState) {
+                    log.error("JanusGraphManager should be instantiated on this server, but it is not. " +
+                              "Please restart with proper server settings. " +
+                              "As a result, we could not evict graph {} from the cache.", graphName);
+                    break;
+                }
+                else if (!txStillOpen) {
                     //Send ack and finish up
                     DataOutput out = graph.getDataSerializer().getDataOutput(64);
                     out.writeObjectNotNull(MgmtLogType.CACHED_TYPE_EVICTION_ACK);
                     out.writeObjectNotNull(originId);
                     VariableLong.writePositive(out,evictionId);
+                    if (null != jgm && action.equals(EVICT)) {
+                        jgm.removeGraph(graphName);
+                        log.debug("Graph {} has been removed from the JanusGraphManager graph cache.", graphName);
+                    }
                     try {
                         sysLog.add(out.getStaticBuffer());
                         log.debug("Sent {}: evictionID={} originID={}", MgmtLogType.CACHED_TYPE_EVICTION_ACK, evictionId, originId);

janusgraph-core/src/main/java/org/janusgraph/graphdb/database/management/ManagementSystem.java

@@ -64,6 +64,7 @@
 import org.janusgraph.graphdb.internal.Order;
 import org.janusgraph.graphdb.internal.JanusGraphSchemaCategory;
 import org.janusgraph.graphdb.internal.Token;
+import org.janusgraph.graphdb.management.JanusGraphManager;
 import org.janusgraph.graphdb.olap.VertexJobConverter;
 import org.janusgraph.graphdb.olap.job.IndexRemoveJob;
 import org.janusgraph.graphdb.olap.job.IndexRepairJob;
@@ -142,6 +143,7 @@ public class ManagementSystem implements JanusGraphManagement {
     private final StandardJanusGraphTx transaction;
 
     private final Set<JanusGraphSchemaVertex> updatedTypes;
+    private boolean evictGraphFromCache;
     private final List<Callable<Boolean>> updatedTypeTriggers;
 
     private final Instant txStartTime;
@@ -161,6 +163,7 @@ public ManagementSystem(StandardJanusGraph graph, KCVSConfiguration config, Log
         this.userConfig = new UserModifiableConfiguration(modifyConfig, configVerifier);
 
         this.updatedTypes = new HashSet<>();
+        this.evictGraphFromCache = false;
         this.updatedTypeTriggers = new ArrayList<>();
         this.graphShutdownRequired = false;
 
@@ -236,8 +239,8 @@ public synchronized void commit() {
         transaction.commit();
 
         //Communicate schema changes
-        if (!updatedTypes.isEmpty()) {
-            managementLogger.sendCacheEviction(updatedTypes, updatedTypeTriggers, getOpenInstancesInternal());
+        if (!updatedTypes.isEmpty() || evictGraphFromCache) {
+            managementLogger.sendCacheEviction(updatedTypes, evictGraphFromCache, updatedTypeTriggers, getOpenInstancesInternal());
             for (JanusGraphSchemaVertex schemaVertex : updatedTypes) {
                 schemaCache.expireSchemaElement(schemaVertex.longId());
             }
@@ -744,6 +747,31 @@ public IndexJobFuture updateIndex(Index index, SchemaAction updateAction) {
         return future;
     }
 
+    /**
+     * Upon the open managementsystem's commit, this graph will be asynchronously evicted from the cache on all JanusGraph nodes in your
+     * cluster, once there are no open transactions on this graph on each respective JanusGraph node
+     * and assuming each node is correctly configured to use the {@link JanusGraphManager}.
+     */
+    public void evictGraphFromCache() {
+        this.evictGraphFromCache = true;
+        setUpdateTrigger(new GraphCacheEvictionCompleteTrigger(this.graph.getGraphName()));
+    }
+
+    private static class GraphCacheEvictionCompleteTrigger implements Callable<Boolean> {
+        private static final Logger log = LoggerFactory.getLogger(GraphCacheEvictionCompleteTrigger.class);
+        private final String graphName;
+
+        private GraphCacheEvictionCompleteTrigger(String graphName) {
+            this.graphName = graphName;
+        }
+
+        @Override
+        public Boolean call() {
+            log.info("Graph {} has been removed from the graph cache on every JanusGraph node in the cluster.", graphName);
+            return true;
+        }
+    }
+
     private static class EmptyIndexJobFuture implements IndexJobFuture {
 
         @Override

janusgraph-core/src/main/java/org/janusgraph/graphdb/database/serialize/StandardSerializer.java

@@ -28,6 +28,7 @@
 import org.janusgraph.diskstorage.util.time.TimestampProviders;
 import org.janusgraph.graphdb.database.idhandling.VariableLong;
 import org.janusgraph.graphdb.database.log.LogTxStatus;
+import org.janusgraph.graphdb.database.management.GraphCacheEvictionAction;
 import org.janusgraph.graphdb.database.management.MgmtLogType;
 import org.janusgraph.graphdb.database.serialize.attribute.*;
 import org.janusgraph.graphdb.internal.ElementCategory;
@@ -130,6 +131,7 @@ public StandardSerializer() {
         registerClassInternal(66,StandardTransactionId.class, new StandardTransactionIdSerializer());
         registerClassInternal(67,TraverserSet.class, new SerializableSerializer());
         registerClassInternal(68,HashMap.class, new SerializableSerializer());
+        registerClassInternal(69,GraphCacheEvictionAction.class, new EnumSerializer<>(GraphCacheEvictionAction.class));
 
     }
 

janusgraph-core/src/main/java/org/janusgraph/graphdb/management/JanusGraphManager.java

@@ -131,6 +131,11 @@ public void run() {
         }
     }
 
+    // To be used for testing purposes
+    protected static void shutdownJanusGraphManager() {
+        instance = null;
+    }
+
     /**
      * @Deprecated
      */