This is an automated email from the ASF dual-hosted git repository. jermy pushed a commit to branch server-info-to-global in repository https://gitbox.apache.org/repos/asf/incubator-hugegraph.git
commit 34eab0b8c2ad78f6f86088d7f35f0d854f1931df Author: Jermy Li <[email protected]> AuthorDate: Sun Dec 3 06:52:16 2023 +0800 chore: move server info into GlobalMasterInfo Change-Id: Id854892333115fd45d7c8b9799d255627541b2ad --- .../hugegraph/api/filter/RedirectFilter.java | 12 +-- .../apache/hugegraph/auth/HugeGraphAuthProxy.java | 10 +-- .../org/apache/hugegraph/core/GraphManager.java | 94 ++++++++++------------ .../main/java/org/apache/hugegraph/HugeGraph.java | 8 +- .../org/apache/hugegraph/StandardHugeGraph.java | 44 +++++----- .../hugegraph/job/computer/AbstractComputer.java | 8 +- .../hugegraph/masterelection/GlobalMasterInfo.java | 89 +++++++++++++++----- .../masterelection/RoleElectionStateMachine.java | 2 +- ...StateMachineCallback.java => RoleListener.java} | 2 +- .../StandardRoleElectionStateMachine.java | 53 +++++++----- ...hineCallback.java => StandardRoleListener.java} | 63 +++++++-------- .../masterelection/StateMachineContext.java | 2 +- .../apache/hugegraph/task/ServerInfoManager.java | 88 ++++++++++---------- .../hugegraph/task/StandardTaskScheduler.java | 35 ++++---- .../org/apache/hugegraph/task/TaskManager.java | 50 ++++++------ .../org/apache/hugegraph/example/ExampleUtil.java | 7 +- .../org/apache/hugegraph/core/BaseCoreTest.java | 5 +- .../org/apache/hugegraph/core/MultiGraphsTest.java | 11 ++- .../core/RoleElectionStateMachineTest.java | 10 +-- .../org/apache/hugegraph/tinkerpop/TestGraph.java | 22 +++-- .../hugegraph/unit/core/SecurityManagerTest.java | 13 ++- 21 files changed, 339 insertions(+), 289 deletions(-) diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/RedirectFilter.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/RedirectFilter.java index e675dd955..3fdfd7689 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/RedirectFilter.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/RedirectFilter.java @@ -82,16 +82,16 @@ public class RedirectFilter implements ContainerRequestFilter { return; } - GlobalMasterInfo globalMasterInfo = manager.globalMasterInfo(); - if (globalMasterInfo == null || !globalMasterInfo.isFeatureSupport()) { + GlobalMasterInfo globalNodeInfo = manager.globalNodeRoleInfo(); + if (globalNodeInfo == null || !globalNodeInfo.supportElection()) { return; } - GlobalMasterInfo.NodeInfo masterNodeInfo = globalMasterInfo.nodeInfo(); - if (masterNodeInfo == null || masterNodeInfo.isMaster() || - StringUtils.isEmpty(masterNodeInfo.url())) { + GlobalMasterInfo.NodeInfo masterInfo = globalNodeInfo.masterInfo(); + if (masterInfo == null || masterInfo.isMaster() || + StringUtils.isEmpty(masterInfo.nodeUrl())) { return; } - String url = masterNodeInfo.url(); + String url = masterInfo.nodeUrl(); URI redirectUri; try { diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java index 2435e2667..6fa34ff18 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java @@ -55,6 +55,7 @@ import org.apache.hugegraph.config.TypedOption; import org.apache.hugegraph.exception.NotSupportException; import org.apache.hugegraph.iterator.FilterIterator; import org.apache.hugegraph.iterator.MapperIterator; +import org.apache.hugegraph.masterelection.GlobalMasterInfo; import org.apache.hugegraph.masterelection.RoleElectionStateMachine; import org.apache.hugegraph.rpc.RpcServiceConfig4Client; import org.apache.hugegraph.rpc.RpcServiceConfig4Server; @@ -78,7 +79,6 @@ import org.apache.hugegraph.type.HugeType; import org.apache.hugegraph.type.Nameable; import org.apache.hugegraph.type.define.GraphMode; import org.apache.hugegraph.type.define.GraphReadMode; -import org.apache.hugegraph.type.define.NodeRole; import org.apache.hugegraph.util.E; import org.apache.hugegraph.util.Log; import org.apache.hugegraph.util.RateLimiter; @@ -669,9 +669,9 @@ public final class HugeGraphAuthProxy implements HugeGraph { } @Override - public void serverStarted(Id serverId, NodeRole serverRole) { + public void serverStarted(GlobalMasterInfo serverInfo) { this.verifyAdminPermission(); - this.hugegraph.serverStarted(serverId, serverRole); + this.hugegraph.serverStarted(serverInfo); } @Override @@ -776,9 +776,9 @@ public final class HugeGraphAuthProxy implements HugeGraph { } @Override - public void create(String configPath, Id server, NodeRole role) { + public void create(String configPath, GlobalMasterInfo serverInfo) { this.verifyPermission(HugePermission.WRITE, ResourceType.STATUS); - this.hugegraph.create(configPath, server, role); + this.hugegraph.create(configPath, serverInfo); } @Override diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java index 2c73b5ee9..5ca7d0a0f 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java @@ -23,8 +23,6 @@ import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.Executor; -import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -41,7 +39,6 @@ import org.apache.hugegraph.auth.StandardAuthenticator; import org.apache.hugegraph.backend.BackendException; import org.apache.hugegraph.backend.cache.Cache; import org.apache.hugegraph.backend.cache.CacheManager; -import org.apache.hugegraph.backend.id.Id; import org.apache.hugegraph.backend.id.IdGenerator; import org.apache.hugegraph.backend.store.BackendStoreInfo; import org.apache.hugegraph.config.CoreOptions; @@ -53,7 +50,7 @@ import org.apache.hugegraph.exception.NotSupportException; import org.apache.hugegraph.masterelection.GlobalMasterInfo; import org.apache.hugegraph.masterelection.RoleElectionOptions; import org.apache.hugegraph.masterelection.RoleElectionStateMachine; -import org.apache.hugegraph.masterelection.StandardStateMachineCallback; +import org.apache.hugegraph.masterelection.StandardRoleListener; import org.apache.hugegraph.metrics.MetricsUtil; import org.apache.hugegraph.metrics.ServerReporter; import org.apache.hugegraph.rpc.RpcClientProvider; @@ -88,14 +85,11 @@ public final class GraphManager { private final HugeAuthenticator authenticator; private final RpcServer rpcServer; private final RpcClientProvider rpcClient; - private final HugeConfig conf; - - private RoleElectionStateMachine roleStateWorker; - private GlobalMasterInfo globalMasterInfo; - private Id server; - private NodeRole role; + private RoleElectionStateMachine roleStateMachine; + private GlobalMasterInfo globalNodeRoleInfo; + private final HugeConfig conf; private final EventHub eventHub; public GraphManager(HugeConfig conf, EventHub hub) { @@ -104,6 +98,10 @@ public final class GraphManager { this.authenticator = HugeAuthenticator.loadAuthenticator(conf); this.rpcServer = new RpcServer(conf); this.rpcClient = new RpcClientProvider(conf); + + this.roleStateMachine = null; + this.globalNodeRoleInfo = new GlobalMasterInfo(); + this.eventHub = hub; this.conf = conf; } @@ -141,8 +139,7 @@ public final class GraphManager { } } - public HugeGraph cloneGraph(String name, String newName, - String configText) { + public HugeGraph cloneGraph(String name, String newName, String configText) { /* * 0. check and modify params * 1. create graph instance @@ -270,6 +267,10 @@ public final class GraphManager { return this.authenticator().authManager(); } + public GlobalMasterInfo globalNodeRoleInfo() { + return this.globalNodeRoleInfo; + } + public void close() { for (Graph graph : this.graphs.values()) { try { @@ -280,8 +281,8 @@ public final class GraphManager { } this.destroyRpcServer(); this.unlistenChanges(); - if (this.roleStateWorker != null) { - this.roleStateWorker.shutdown(); + if (this.roleStateMachine != null) { + this.roleStateMachine.shutdown(); } } @@ -414,8 +415,7 @@ public final class GraphManager { LOG.info("RpcServer is not enabled, skip wait graphs ready"); return; } - com.alipay.remoting.rpc.RpcServer remotingRpcServer = - this.remotingRpcServer(); + com.alipay.remoting.rpc.RpcServer remotingRpcServer = this.remotingRpcServer(); for (String graphName : this.graphs.keySet()) { HugeGraph graph = this.graph(graphName); graph.waitReady(remotingRpcServer); @@ -433,7 +433,7 @@ public final class GraphManager { if (this.requireAuthentication()) { String token = config.get(ServerOptions.AUTH_ADMIN_TOKEN); try { - this.authenticator.initAdminUser(token); + this.authenticator().initAdminUser(token); } catch (Exception e) { throw new BackendException( "The backend store of '%s' can't " + @@ -455,65 +455,57 @@ public final class GraphManager { } private void serverStarted(HugeConfig config) { - String server = config.get(ServerOptions.SERVER_ID); + String serverId = config.get(ServerOptions.SERVER_ID); String role = config.get(ServerOptions.SERVER_ROLE); - E.checkArgument(StringUtils.isNotEmpty(server), + E.checkArgument(StringUtils.isNotEmpty(serverId), "The server name can't be null or empty"); E.checkArgument(StringUtils.isNotEmpty(role), "The server role can't be null or empty"); - this.server = IdGenerator.of(server); - this.role = NodeRole.valueOf(role.toUpperCase()); - boolean supportRoleStateWorker = this.supportRoleStateWorker(); - if (supportRoleStateWorker) { - this.role = NodeRole.WORKER; + NodeRole serverRole = NodeRole.valueOf(role.toUpperCase()); + boolean supportRoleElection = !serverRole.computer() && + this.supportRoleElection(); + if (supportRoleElection) { + // Init any server as Worker role, then do role election + serverRole = NodeRole.WORKER; } + this.globalNodeRoleInfo.serverId(IdGenerator.of(serverId)); + this.globalNodeRoleInfo.initServerRole(serverRole); + for (String graph : this.graphs()) { HugeGraph hugegraph = this.graph(graph); assert hugegraph != null; - hugegraph.serverStarted(this.server, this.role); + hugegraph.serverStarted(this.globalNodeRoleInfo); } - if (supportRoleStateWorker) { - this.initRoleStateWorker(); + if (supportRoleElection) { + this.initRoleStateMachine(); } } - private void initRoleStateWorker() { - E.checkArgument(this.roleStateWorker == null, "Repetition init"); - Executor applyThread = Executors.newSingleThreadExecutor(); - this.roleStateWorker = this.authenticator().graph().roleElectionStateMachine(); - this.globalMasterInfo = new GlobalMasterInfo(); - StandardStateMachineCallback stateMachineCallback = new StandardStateMachineCallback( - TaskManager.instance(), - this.globalMasterInfo); - applyThread.execute(() -> { - this.roleStateWorker.apply(stateMachineCallback); - }); - } - - public GlobalMasterInfo globalMasterInfo() { - return this.globalMasterInfo; + private void initRoleStateMachine() { + E.checkArgument(this.roleStateMachine == null, + "Repeated initialization of role state worker"); + this.globalNodeRoleInfo.supportElection(true); + this.roleStateMachine = this.authenticator().graph().roleElectionStateMachine(); + StandardRoleListener listener = new StandardRoleListener(TaskManager.instance(), + this.globalNodeRoleInfo); + this.roleStateMachine.start(listener); } - private boolean supportRoleStateWorker() { - if (this.role.computer()) { - return false; - } - + private boolean supportRoleElection() { try { if (!(this.authenticator() instanceof StandardAuthenticator)) { LOG.info("{} authenticator does not support role election currently", this.authenticator().getClass().getSimpleName()); return false; } + return true; } catch (IllegalStateException e) { - LOG.info("Unconfigured StandardAuthenticator, not support role election currently"); + LOG.info("{}, does not support role election currently", e.getMessage()); return false; } - - return true; } private void addMetrics(HugeConfig config) { @@ -591,7 +583,7 @@ public final class GraphManager { graph = (HugeGraph) GraphFactory.open(config); // Init graph and start it - graph.create(this.graphsDir, this.server, this.role); + graph.create(this.graphsDir, this.globalNodeRoleInfo); } catch (Throwable e) { LOG.error("Failed to create graph '{}' due to: {}", name, e.getMessage(), e); diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/HugeGraph.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/HugeGraph.java index cd287c47b..420f8bd4d 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/HugeGraph.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/HugeGraph.java @@ -30,6 +30,7 @@ import org.apache.hugegraph.backend.store.BackendStoreInfo; import org.apache.hugegraph.backend.store.raft.RaftGroupManager; import org.apache.hugegraph.config.HugeConfig; import org.apache.hugegraph.config.TypedOption; +import org.apache.hugegraph.masterelection.GlobalMasterInfo; import org.apache.hugegraph.masterelection.RoleElectionStateMachine; import org.apache.hugegraph.rpc.RpcServiceConfig4Client; import org.apache.hugegraph.rpc.RpcServiceConfig4Server; @@ -44,12 +45,11 @@ import org.apache.hugegraph.structure.HugeFeatures; import org.apache.hugegraph.task.TaskScheduler; import org.apache.hugegraph.traversal.optimize.HugeCountStepStrategy; import org.apache.hugegraph.traversal.optimize.HugeGraphStepStrategy; -import org.apache.hugegraph.traversal.optimize.HugeVertexStepStrategy; import org.apache.hugegraph.traversal.optimize.HugePrimaryKeyStrategy; +import org.apache.hugegraph.traversal.optimize.HugeVertexStepStrategy; import org.apache.hugegraph.type.HugeType; import org.apache.hugegraph.type.define.GraphMode; import org.apache.hugegraph.type.define.GraphReadMode; -import org.apache.hugegraph.type.define.NodeRole; import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategies; import org.apache.tinkerpop.gremlin.structure.Edge; import org.apache.tinkerpop.gremlin.structure.Graph; @@ -201,7 +201,7 @@ public interface HugeGraph extends Graph { void waitReady(RpcServer rpcServer); - void serverStarted(Id serverId, NodeRole serverRole); + void serverStarted(GlobalMasterInfo serverInfo); boolean started(); @@ -221,7 +221,7 @@ public interface HugeGraph extends Graph { void resumeSnapshot(); - void create(String configPath, Id server, NodeRole role); + void create(String configPath, GlobalMasterInfo serverInfo); void drop(); diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java index db37d0a4b..df3de8671 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java @@ -51,7 +51,6 @@ import org.apache.hugegraph.backend.store.BackendStoreProvider; import org.apache.hugegraph.backend.store.raft.RaftBackendStoreProvider; import org.apache.hugegraph.backend.store.raft.RaftGroupManager; import org.apache.hugegraph.backend.store.ram.RamTable; -import org.apache.hugegraph.task.EphemeralJobQueue; import org.apache.hugegraph.backend.tx.GraphTransaction; import org.apache.hugegraph.backend.tx.SchemaTransaction; import org.apache.hugegraph.config.CoreOptions; @@ -64,6 +63,7 @@ import org.apache.hugegraph.io.HugeGraphIoRegistry; import org.apache.hugegraph.job.EphemeralJob; import org.apache.hugegraph.masterelection.ClusterRoleStore; import org.apache.hugegraph.masterelection.Config; +import org.apache.hugegraph.masterelection.GlobalMasterInfo; import org.apache.hugegraph.masterelection.RoleElectionConfig; import org.apache.hugegraph.masterelection.RoleElectionOptions; import org.apache.hugegraph.masterelection.RoleElectionStateMachine; @@ -84,13 +84,13 @@ import org.apache.hugegraph.structure.HugeEdgeProperty; import org.apache.hugegraph.structure.HugeFeatures; import org.apache.hugegraph.structure.HugeVertex; import org.apache.hugegraph.structure.HugeVertexProperty; +import org.apache.hugegraph.task.EphemeralJobQueue; import org.apache.hugegraph.task.ServerInfoManager; import org.apache.hugegraph.task.TaskManager; import org.apache.hugegraph.task.TaskScheduler; import org.apache.hugegraph.type.HugeType; import org.apache.hugegraph.type.define.GraphMode; import org.apache.hugegraph.type.define.GraphReadMode; -import org.apache.hugegraph.type.define.NodeRole; import org.apache.hugegraph.util.ConfigUtil; import org.apache.hugegraph.util.DateUtil; import org.apache.hugegraph.util.E; @@ -267,15 +267,15 @@ public class StandardHugeGraph implements HugeGraph { } @Override - public void serverStarted(Id serverId, NodeRole serverRole) { + public void serverStarted(GlobalMasterInfo serverInfo) { LOG.info("Init system info for graph '{}'", this.name); this.initSystemInfo(); LOG.info("Init server info [{}-{}] for graph '{}'...", - serverId, serverRole, this.name); - this.serverInfoManager().initServerInfo(serverId, serverRole); + serverInfo.serverId(), serverInfo.serverRole(), this.name); + this.serverInfoManager().initServerInfo(serverInfo); - this.initRoleStateWorker(serverId); + this.initRoleStateMachine(serverInfo.serverId()); // TODO: check necessary? LOG.info("Check olap property-key tables for graph '{}'", this.name); @@ -291,16 +291,18 @@ public class StandardHugeGraph implements HugeGraph { this.started = true; } - private void initRoleStateWorker(Id serverId) { - Config roleStateMachineConfig = new RoleElectionConfig(serverId.toString(), - this.configuration.get(RoleElectionOptions.NODE_EXTERNAL_URL), - this.configuration.get(RoleElectionOptions.EXCEEDS_FAIL_COUNT), - this.configuration.get(RoleElectionOptions.RANDOM_TIMEOUT_MILLISECOND), - this.configuration.get(RoleElectionOptions.HEARTBEAT_INTERVAL_SECOND), - this.configuration.get(RoleElectionOptions.MASTER_DEAD_TIMES), - this.configuration.get(RoleElectionOptions.BASE_TIMEOUT_MILLISECOND)); - ClusterRoleStore clusterRoleStore = new StandardClusterRoleStore(this.params); - this.roleElectionStateMachine = new StandardRoleElectionStateMachine(roleStateMachineConfig, clusterRoleStore); + private void initRoleStateMachine(Id serverId) { + HugeConfig conf = this.configuration; + Config roleConfig = new RoleElectionConfig(serverId.toString(), + conf.get(RoleElectionOptions.NODE_EXTERNAL_URL), + conf.get(RoleElectionOptions.EXCEEDS_FAIL_COUNT), + conf.get(RoleElectionOptions.RANDOM_TIMEOUT_MILLISECOND), + conf.get(RoleElectionOptions.HEARTBEAT_INTERVAL_SECOND), + conf.get(RoleElectionOptions.MASTER_DEAD_TIMES), + conf.get(RoleElectionOptions.BASE_TIMEOUT_MILLISECOND)); + ClusterRoleStore roleStore = new StandardClusterRoleStore(this.params); + this.roleElectionStateMachine = new StandardRoleElectionStateMachine(roleConfig, + roleStore); } @Override @@ -399,8 +401,7 @@ public class StandardHugeGraph implements HugeGraph { try { this.storeProvider.truncate(); // TODO: remove this after serverinfo saved in etcd - this.serverStarted(this.serverInfoManager().selfServerId(), - this.serverInfoManager().selfServerRole()); + this.serverStarted(this.serverInfoManager().serverInfo()); } finally { LockUtil.unlock(this.name, LockUtil.GRAPH_LOCK); } @@ -974,9 +975,9 @@ public class StandardHugeGraph implements HugeGraph { } @Override - public void create(String configPath, Id server, NodeRole role) { + public void create(String configPath, GlobalMasterInfo serverInfo) { this.initBackend(); - this.serverStarted(server, role); + this.serverStarted(serverInfo); // Write config to disk file String confPath = ConfigUtil.writeToFile(configPath, this.name(), @@ -1052,8 +1053,7 @@ public class StandardHugeGraph implements HugeGraph { } private ServerInfoManager serverInfoManager() { - ServerInfoManager manager = this.taskManager - .getServerInfoManager(this.params); + ServerInfoManager manager = this.taskManager.getServerInfoManager(this.params); E.checkState(manager != null, "Can't find server info manager for graph '%s'", this); return manager; diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/job/computer/AbstractComputer.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/job/computer/AbstractComputer.java index a40d0001b..11207ebae 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/job/computer/AbstractComputer.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/job/computer/AbstractComputer.java @@ -32,16 +32,15 @@ import org.apache.commons.configuration2.io.FileHandler; import org.apache.commons.configuration2.tree.ImmutableNode; import org.apache.commons.configuration2.tree.NodeHandler; import org.apache.commons.configuration2.tree.NodeModel; -import org.apache.hugegraph.type.define.Directions; -import org.apache.hugegraph.util.ParameterUtil; -import org.slf4j.Logger; - import org.apache.hugegraph.HugeException; import org.apache.hugegraph.job.ComputerJob; import org.apache.hugegraph.job.Job; import org.apache.hugegraph.traversal.algorithm.HugeTraverser; +import org.apache.hugegraph.type.define.Directions; import org.apache.hugegraph.util.E; import org.apache.hugegraph.util.Log; +import org.apache.hugegraph.util.ParameterUtil; +import org.slf4j.Logger; public abstract class AbstractComputer implements Computer { @@ -85,7 +84,6 @@ public abstract class AbstractComputer implements Computer { @Override public Object call(Job<Object> job, Map<String, Object> parameters) { - this.checkAndCollectParameters(parameters); // Read configuration try { diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/GlobalMasterInfo.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/GlobalMasterInfo.java index d7dd127af..709b1b75d 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/GlobalMasterInfo.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/GlobalMasterInfo.java @@ -17,50 +17,103 @@ package org.apache.hugegraph.masterelection; -public class GlobalMasterInfo { +import org.apache.hugegraph.backend.id.Id; +import org.apache.hugegraph.backend.id.IdGenerator; +import org.apache.hugegraph.type.define.NodeRole; +import org.apache.hugegraph.util.E; - private NodeInfo nodeInfo; - private volatile boolean featureSupport; +// TODO: rename to GlobalNodeRoleInfo +public final class GlobalMasterInfo { + + private final static NodeInfo NO_MASTER = new NodeInfo(false, ""); + + private volatile boolean supportElection; + private volatile NodeInfo masterNodeInfo; + + private volatile Id serverId; + private volatile NodeRole serverRole; public GlobalMasterInfo() { - this.featureSupport = false; - this.nodeInfo = new NodeInfo(false, ""); + this(NO_MASTER); + } + + public GlobalMasterInfo(NodeInfo masterInfo) { + this.supportElection = false; + this.masterNodeInfo = masterInfo; + + this.serverId = null; + this.serverRole = null; + } + + public void supportElection(boolean featureSupport) { + this.supportElection = featureSupport; + } + + public boolean supportElection() { + return this.supportElection; } - public final void nodeInfo(boolean isMaster, String url) { + public void resetMasterInfo() { + this.masterNodeInfo = NO_MASTER; + } + + public void masterInfo(boolean isMaster, String nodeUrl) { // final can avoid instruction rearrangement, visibility can be ignored - final NodeInfo tmp = new NodeInfo(isMaster, url); - this.nodeInfo = tmp; + this.masterNodeInfo = new NodeInfo(isMaster, nodeUrl); + } + + public NodeInfo masterInfo() { + return this.masterNodeInfo; + } + + public Id serverId() { + return this.serverId; + } + + public NodeRole serverRole() { + return this.serverRole; + } + + public void serverId(Id id) { + this.serverId = id; } - public final NodeInfo nodeInfo() { - return this.nodeInfo; + public void initServerRole(NodeRole role) { + E.checkArgument(role != null, "The server role can't be null"); + E.checkArgument(this.serverRole == null, + "The server role can't be init twice"); + this.serverRole = role; } - public void isFeatureSupport(boolean featureSupport) { - this.featureSupport = featureSupport; + public void changeServerRole(NodeRole role) { + E.checkArgument(role != null, "The server role can't be null"); + this.serverRole = role; } - public boolean isFeatureSupport() { - return this.featureSupport; + public static GlobalMasterInfo master(String serverId) { + NodeInfo masterInfo = new NodeInfo(true, serverId); + GlobalMasterInfo serverInfo = new GlobalMasterInfo(masterInfo); + serverInfo.serverId = IdGenerator.of(serverId); + serverInfo.serverRole = NodeRole.MASTER; + return serverInfo; } public static class NodeInfo { private final boolean isMaster; - private final String url; + private final String nodeUrl; public NodeInfo(boolean isMaster, String url) { this.isMaster = isMaster; - this.url = url; + this.nodeUrl = url; } public boolean isMaster() { return this.isMaster; } - public String url() { - return this.url; + public String nodeUrl() { + return this.nodeUrl; } } } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/RoleElectionStateMachine.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/RoleElectionStateMachine.java index 920bc104f..2a33d1bf6 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/RoleElectionStateMachine.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/RoleElectionStateMachine.java @@ -21,5 +21,5 @@ public interface RoleElectionStateMachine { void shutdown(); - void apply(StateMachineCallback stateMachineCallback); + void start(RoleListener callback); } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StateMachineCallback.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/RoleListener.java similarity index 96% rename from hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StateMachineCallback.java rename to hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/RoleListener.java index 35abbe340..e99db4c16 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StateMachineCallback.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/RoleListener.java @@ -17,7 +17,7 @@ package org.apache.hugegraph.masterelection; -public interface StateMachineCallback { +public interface RoleListener { void onAsRoleMaster(StateMachineContext context); diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardRoleElectionStateMachine.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardRoleElectionStateMachine.java index aa284def6..a0e2601d4 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardRoleElectionStateMachine.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardRoleElectionStateMachine.java @@ -19,6 +19,8 @@ package org.apache.hugegraph.masterelection; import java.security.SecureRandom; import java.util.Optional; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.locks.LockSupport; import org.apache.hugegraph.util.E; @@ -29,25 +31,36 @@ public class StandardRoleElectionStateMachine implements RoleElectionStateMachin private static final Logger LOG = Log.logger(StandardRoleElectionStateMachine.class); - private volatile boolean shutdown; private final Config config; + private final ClusterRoleStore roleStore; + private final ExecutorService applyThread; + + private volatile boolean shutdown; private volatile RoleState state; - private final ClusterRoleStore clusterRoleStore; - public StandardRoleElectionStateMachine(Config config, ClusterRoleStore clusterRoleStore) { + public StandardRoleElectionStateMachine(Config config, ClusterRoleStore roleStore) { this.config = config; - this.clusterRoleStore = clusterRoleStore; + this.roleStore = roleStore; + this.applyThread = Executors.newSingleThreadExecutor(); this.state = new UnknownState(null); this.shutdown = false; } @Override public void shutdown() { + if (this.shutdown) { + return; + } this.shutdown = true; + this.applyThread.shutdown(); } @Override - public void apply(StateMachineCallback stateMachineCallback) { + public void start(RoleListener stateMachineCallback) { + this.applyThread.execute(() -> this.apply(stateMachineCallback)); + } + + private void apply(RoleListener stateMachineCallback) { int failCount = 0; StateMachineContextImpl context = new StateMachineContextImpl(this); while (!this.shutdown) { @@ -73,13 +86,17 @@ public class StandardRoleElectionStateMachine implements RoleElectionStateMachin } } + protected ClusterRoleStore roleStore() { + return this.roleStore; + } + private interface RoleState { SecureRandom SECURE_RANDOM = new SecureRandom(); RoleState transform(StateMachineContext context); - Callback callback(StateMachineCallback callback); + Callback callback(RoleListener callback); static void heartBeatPark(StateMachineContext context) { long heartBeatIntervalSecond = context.config().heartBeatIntervalSecond(); @@ -110,7 +127,7 @@ public class StandardRoleElectionStateMachine implements RoleElectionStateMachin @Override public RoleState transform(StateMachineContext context) { - ClusterRoleStore adapter = context.adapter(); + ClusterRoleStore adapter = context.roleStore(); Optional<ClusterRole> clusterRoleOpt = adapter.query(); if (!clusterRoleOpt.isPresent()) { context.reset(); @@ -137,7 +154,7 @@ public class StandardRoleElectionStateMachine implements RoleElectionStateMachin } @Override - public Callback callback(StateMachineCallback callback) { + public Callback callback(RoleListener callback) { return callback::unknown; } } @@ -158,7 +175,7 @@ public class StandardRoleElectionStateMachine implements RoleElectionStateMachin } @Override - public Callback callback(StateMachineCallback callback) { + public Callback callback(RoleListener callback) { return callback::onAsRoleAbdication; } } @@ -175,7 +192,7 @@ public class StandardRoleElectionStateMachine implements RoleElectionStateMachin public RoleState transform(StateMachineContext context) { this.clusterRole.increaseClock(); RoleState.heartBeatPark(context); - if (context.adapter().updateIfNodePresent(this.clusterRole)) { + if (context.roleStore().updateIfNodePresent(this.clusterRole)) { return this; } context.reset(); @@ -184,7 +201,7 @@ public class StandardRoleElectionStateMachine implements RoleElectionStateMachin } @Override - public Callback callback(StateMachineCallback callback) { + public Callback callback(RoleListener callback) { return callback::onAsRoleMaster; } } @@ -216,7 +233,7 @@ public class StandardRoleElectionStateMachine implements RoleElectionStateMachin } @Override - public Callback callback(StateMachineCallback callback) { + public Callback callback(RoleListener callback) { return callback::onAsRoleWorker; } @@ -255,7 +272,7 @@ public class StandardRoleElectionStateMachine implements RoleElectionStateMachin context.config().url(), epoch); // The master failover completed context.epoch(clusterRole.epoch()); - if (context.adapter().updateIfNodePresent(clusterRole)) { + if (context.roleStore().updateIfNodePresent(clusterRole)) { context.master(new MasterServerInfoImpl(clusterRole.node(), clusterRole.url())); return new MasterState(clusterRole); } else { @@ -264,7 +281,7 @@ public class StandardRoleElectionStateMachine implements RoleElectionStateMachin } @Override - public Callback callback(StateMachineCallback callback) { + public Callback callback(RoleListener callback) { return callback::onAsRoleCandidate; } } @@ -303,8 +320,8 @@ public class StandardRoleElectionStateMachine implements RoleElectionStateMachin } @Override - public ClusterRoleStore adapter() { - return this.machine.adapter(); + public ClusterRoleStore roleStore() { + return this.machine.roleStore(); } @Override @@ -348,8 +365,4 @@ public class StandardRoleElectionStateMachine implements RoleElectionStateMachin return this.node; } } - - protected ClusterRoleStore adapter() { - return this.clusterRoleStore; - } } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardStateMachineCallback.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardRoleListener.java similarity index 67% rename from hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardStateMachineCallback.java rename to hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardRoleListener.java index 28e01d291..f2bb94f52 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardStateMachineCallback.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardRoleListener.java @@ -23,84 +23,83 @@ import org.apache.hugegraph.task.TaskManager; import org.apache.hugegraph.util.Log; import org.slf4j.Logger; -public class StandardStateMachineCallback implements StateMachineCallback { +public class StandardRoleListener implements RoleListener { - private static final Logger LOG = Log.logger(StandardStateMachineCallback.class); + private static final Logger LOG = Log.logger(StandardRoleListener.class); private final TaskManager taskManager; - private final GlobalMasterInfo globalMasterInfo; + private final GlobalMasterInfo roleInfo; - private boolean isMaster = false; + private volatile boolean selfIsMaster; - public StandardStateMachineCallback(TaskManager taskManager, - GlobalMasterInfo globalMasterInfo) { + public StandardRoleListener(TaskManager taskManager, + GlobalMasterInfo roleInfo) { this.taskManager = taskManager; - this.taskManager.enableRoleElected(true); - this.globalMasterInfo = globalMasterInfo; - this.globalMasterInfo.isFeatureSupport(true); + this.taskManager.enableRoleElection(); + this.roleInfo = roleInfo; + this.selfIsMaster = false; } @Override public void onAsRoleMaster(StateMachineContext context) { - if (!isMaster) { + if (!selfIsMaster) { this.taskManager.onAsRoleMaster(); LOG.info("Server {} change to master role", context.config().node()); } - this.initGlobalMasterInfo(context); - this.isMaster = true; + this.updateMasterInfo(context); + this.selfIsMaster = true; } @Override public void onAsRoleWorker(StateMachineContext context) { - if (isMaster) { + if (this.selfIsMaster) { this.taskManager.onAsRoleWorker(); LOG.info("Server {} change to worker role", context.config().node()); } - this.initGlobalMasterInfo(context); - this.isMaster = false; + this.updateMasterInfo(context); + this.selfIsMaster = false; } @Override public void onAsRoleCandidate(StateMachineContext context) { + // pass } @Override - public void unknown(StateMachineContext context) { - if (isMaster) { + public void onAsRoleAbdication(StateMachineContext context) { + if (this.selfIsMaster) { this.taskManager.onAsRoleWorker(); LOG.info("Server {} change to worker role", context.config().node()); } - this.initGlobalMasterInfo(context); + this.updateMasterInfo(context); + this.selfIsMaster = false; + } - isMaster = false; + @Override + public void error(StateMachineContext context, Throwable e) { + LOG.error("Server {} exception occurred", context.config().node(), e); } @Override - public void onAsRoleAbdication(StateMachineContext context) { - if (isMaster) { + public void unknown(StateMachineContext context) { + if (this.selfIsMaster) { this.taskManager.onAsRoleWorker(); LOG.info("Server {} change to worker role", context.config().node()); } - this.initGlobalMasterInfo(context); - - isMaster = false; - } + this.updateMasterInfo(context); - @Override - public void error(StateMachineContext context, Throwable e) { - LOG.error("Server {} exception occurred", context.config().node(), e); + this.selfIsMaster = false; } - public void initGlobalMasterInfo(StateMachineContext context) { + public void updateMasterInfo(StateMachineContext context) { StateMachineContext.MasterServerInfo master = context.master(); if (master == null) { - this.globalMasterInfo.nodeInfo(false, ""); + this.roleInfo.resetMasterInfo(); return; } boolean isMaster = Objects.equals(context.node(), master.node()); - String url = master.url(); - this.globalMasterInfo.nodeInfo(isMaster, url); + this.roleInfo.masterInfo(isMaster, master.url()); } } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StateMachineContext.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StateMachineContext.java index a3eaf8962..587cd417f 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StateMachineContext.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StateMachineContext.java @@ -33,7 +33,7 @@ public interface StateMachineContext { void master(MasterServerInfo info); - ClusterRoleStore adapter(); + ClusterRoleStore roleStore(); void reset(); diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/ServerInfoManager.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/ServerInfoManager.java index e8cccf88e..edf3e4973 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/ServerInfoManager.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/ServerInfoManager.java @@ -37,6 +37,7 @@ import org.apache.hugegraph.backend.tx.GraphTransaction; import org.apache.hugegraph.exception.ConnectionException; import org.apache.hugegraph.iterator.ListIterator; import org.apache.hugegraph.iterator.MapperIterator; +import org.apache.hugegraph.masterelection.GlobalMasterInfo; import org.apache.hugegraph.schema.PropertyKey; import org.apache.hugegraph.schema.VertexLabel; import org.apache.hugegraph.structure.HugeVertex; @@ -61,8 +62,7 @@ public class ServerInfoManager { private final HugeGraphParams graph; private final ExecutorService dbExecutor; - private Id selfServerId; - private NodeRole selfServerRole; + private GlobalMasterInfo globalServerInfo; private volatile boolean onlySingleNode; private volatile boolean closed; @@ -75,8 +75,7 @@ public class ServerInfoManager { this.graph = graph; this.dbExecutor = dbExecutor; - this.selfServerId = null; - this.selfServerRole = NodeRole.MASTER; + this.globalServerInfo = null; this.onlySingleNode = false; this.closed = false; @@ -103,40 +102,24 @@ public class ServerInfoManager { return true; } - public synchronized void forceInitServerInfo(Id server, NodeRole role) { - if (this.closed) { - return; - } - - E.checkArgument(server != null && role != null, - "The server id or role can't be null"); - this.selfServerId = server; - this.selfServerRole = role; - - this.saveServerInfo(this.selfServerId, this.selfServerRole); - } - - public synchronized void initServerInfo(Id server, NodeRole role) { - E.checkArgument(server != null && role != null, - "The server id or role can't be null"); - this.selfServerId = server; - this.selfServerRole = role; + public synchronized void initServerInfo(GlobalMasterInfo serverInfo) { + E.checkArgument(serverInfo != null, "The global node info can't be null"); + this.globalServerInfo = serverInfo; + Id serverId = this.globalServerInfo.serverId(); - HugeServerInfo existed = this.serverInfo(server); + HugeServerInfo existed = this.serverInfo(serverId); E.checkArgument(existed == null || !existed.alive(), "The server with name '%s' already in cluster", - server); - if (role.master()) { + serverId); + if (this.globalServerInfo.serverRole().master()) { String page = this.supportsPaging() ? PageInfo.PAGE_NONE : null; do { - Iterator<HugeServerInfo> servers = this.serverInfos(PAGE_SIZE, - page); + Iterator<HugeServerInfo> servers = this.serverInfos(PAGE_SIZE, page); while (servers.hasNext()) { existed = servers.next(); - E.checkArgument(!existed.role().master() || - !existed.alive(), - "Already existed master '%s' in current " + - "cluster", existed.id()); + E.checkArgument(!existed.role().master() || !existed.alive(), + "Already existed master '%s' in current cluster", + existed.id()); } if (page != null) { page = PageInfo.pageInfo(servers); @@ -145,19 +128,39 @@ public class ServerInfoManager { } // TODO: save ServerInfo at AuthServer - this.saveServerInfo(this.selfServerId, this.selfServerRole); + this.saveServerInfo(this.selfServerId(), this.selfServerRole()); + } + + public synchronized void changeServerRole(NodeRole serverRole) { + if (this.closed) { + return; + } + + this.globalServerInfo.changeServerRole(serverRole); + + this.saveServerInfo(this.selfServerId(), this.selfServerRole()); + } + + public GlobalMasterInfo serverInfo() { + return this.globalServerInfo; } public Id selfServerId() { - return this.selfServerId; + if (this.globalServerInfo == null) { + return null; + } + return this.globalServerInfo.serverId(); } public NodeRole selfServerRole() { - return this.selfServerRole; + if (this.globalServerInfo == null) { + return null; + } + return this.globalServerInfo.serverRole(); } public boolean master() { - return this.selfServerRole != null && this.selfServerRole.master(); + return this.selfServerRole() != null && this.selfServerRole().master(); } public boolean onlySingleNode() { @@ -167,10 +170,9 @@ public class ServerInfoManager { public void heartbeat() { HugeServerInfo serverInfo = this.selfServerInfo(); - if (serverInfo == null && this.selfServerId != null && - this.selfServerRole != NodeRole.MASTER) { - serverInfo = this.saveServerInfo(this.selfServerId, - this.selfServerRole); + if (serverInfo == null && this.selfServerId() != null && + this.selfServerRole() != NodeRole.MASTER) { + serverInfo = this.saveServerInfo(this.selfServerId(), this.selfServerRole()); } serverInfo.updateTime(DateUtil.now()); this.save(serverInfo); @@ -310,9 +312,9 @@ public class ServerInfoManager { } private HugeServerInfo selfServerInfo() { - HugeServerInfo selfServerInfo = this.serverInfo(this.selfServerId); + HugeServerInfo selfServerInfo = this.serverInfo(this.selfServerId()); if (selfServerInfo == null) { - LOG.warn("ServerInfo is missing: {}", this.selfServerId); + LOG.warn("ServerInfo is missing: {}", this.selfServerId()); } return selfServerInfo; } @@ -335,8 +337,8 @@ public class ServerInfoManager { * backend store, initServerInfo() is not called in this case, so * this.selfServerId is null at this time. */ - if (this.selfServerId != null && this.graph.initialized()) { - return this.removeServerInfo(this.selfServerId); + if (this.selfServerId() != null && this.graph.initialized()) { + return this.removeServerInfo(this.selfServerId()); } return null; } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java index 9eda3f6b0..bf0a16314 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java @@ -177,32 +177,32 @@ public class StandardTaskScheduler implements TaskScheduler { public <V> Future<?> schedule(HugeTask<V> task) { E.checkArgumentNotNull(task, "Task can't be null"); + /* + * Just submit to queue if status=QUEUED (means re-schedule task) + * NOTE: schedule() method may be called multi times by + * HugeTask.checkDependenciesSuccess() method + */ if (task.status() == TaskStatus.QUEUED) { - /* - * Just submit to queue if status=QUEUED (means re-schedule task) - * NOTE: schedule() method may be called multi times by - * HugeTask.checkDependenciesSuccess() method - */ return this.resubmitTask(task); } + /* + * Due to EphemeralJob won't be serialized and deserialized through + * shared storage, submit EphemeralJob immediately on any node + */ if (task.callable() instanceof EphemeralJob) { - /* - * Due to EphemeralJob won't be serialized and deserialized through - * shared storage, submit EphemeralJob immediately on master - * NOTE: don't need to save EphemeralJob task - */ + // NOTE: we don't need to save EphemeralJob task task.status(TaskStatus.QUEUED); return this.submitTask(task); } - // Only check if not EphemeralJob + // Check this is on master for normal task schedule this.checkOnMasterNode("schedule"); if (this.serverManager().onlySingleNode() && !task.computer()) { /* * Speed up for single node, submit task immediately, - * this code can be removed without affecting logic + * this code can be removed without affecting code logic */ task.status(TaskStatus.QUEUED); task.server(this.serverManager().selfServerId()); @@ -303,10 +303,10 @@ public class StandardTaskScheduler implements TaskScheduler { return this.serverManager; } - protected synchronized void scheduleTasks() { + protected synchronized void scheduleTasksOnMaster() { // Master server schedule all scheduling tasks to suitable worker nodes - Collection<HugeServerInfo> scheduleInfos = this.serverManager() - .allServerInfos(); + Collection<HugeServerInfo> serverInfos = this.serverManager() + .allServerInfos(); String page = this.supportsPaging() ? PageInfo.PAGE_NONE : null; do { Iterator<HugeTask<Object>> tasks = this.tasks(TaskStatus.SCHEDULING, @@ -323,7 +323,7 @@ public class StandardTaskScheduler implements TaskScheduler { } HugeServerInfo server = this.serverManager().pickWorkerNode( - scheduleInfos, task); + serverInfos, task); if (server == null) { LOG.info("The master can't find suitable servers to " + "execute task '{}', wait for next schedule", @@ -348,7 +348,8 @@ public class StandardTaskScheduler implements TaskScheduler { } } while (page != null); - this.serverManager().updateServerInfos(scheduleInfos); + // Save to store + this.serverManager().updateServerInfos(serverInfos); } protected void executeTasksOnWorker(Id server) { diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskManager.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskManager.java index 0ad96f443..bb3af44f4 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskManager.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskManager.java @@ -243,19 +243,8 @@ public final class TaskManager { return size; } - protected void notifyNewTask(HugeTask<?> task) { - Queue<Runnable> queue = ((ThreadPoolExecutor) this.schedulerExecutor) - .getQueue(); - if (queue.size() <= 1) { - /* - * Notify to schedule tasks initiatively when have new task - * It's OK to not notify again if there are more than one task in - * queue(like two, one is timer task, one is immediate task), - * we don't want too many immediate tasks to be inserted into queue, - * one notify will cause all the tasks to be processed. - */ - this.schedulerExecutor.submit(this::scheduleOrExecuteJob); - } + public void enableRoleElection() { + this.enableRoleElected = true; } public void onAsRoleMaster() { @@ -263,7 +252,7 @@ public final class TaskManager { for (TaskScheduler entry : this.schedulers.values()) { StandardTaskScheduler scheduler = (StandardTaskScheduler) entry; ServerInfoManager serverInfoManager = scheduler.serverManager(); - serverInfoManager.forceInitServerInfo(serverInfoManager.selfServerId(), NodeRole.MASTER); + serverInfoManager.changeServerRole(NodeRole.MASTER); } } catch (Throwable e) { LOG.error("Exception occurred when change to master role", e); @@ -276,7 +265,7 @@ public final class TaskManager { for (TaskScheduler entry : this.schedulers.values()) { StandardTaskScheduler scheduler = (StandardTaskScheduler) entry; ServerInfoManager serverInfoManager = scheduler.serverManager(); - serverInfoManager.forceInitServerInfo(serverInfoManager.selfServerId(), NodeRole.WORKER); + serverInfoManager.changeServerRole(NodeRole.WORKER); } } catch (Throwable e) { LOG.error("Exception occurred when change to worker role", e); @@ -284,8 +273,19 @@ public final class TaskManager { } } - public void enableRoleElected(boolean enableRoleElected) { - this.enableRoleElected = enableRoleElected; + protected void notifyNewTask(HugeTask<?> task) { + Queue<Runnable> queue = ((ThreadPoolExecutor) this.schedulerExecutor) + .getQueue(); + if (queue.size() <= 1) { + /* + * Notify to schedule tasks initiatively when have new task + * It's OK to not notify again if there are more than one task in + * queue(like two, one is timer task, one is immediate task), + * we don't want too many immediate tasks to be inserted into queue, + * one notify will cause all the tasks to be processed. + */ + this.schedulerExecutor.submit(this::scheduleOrExecuteJob); + } } private void scheduleOrExecuteJob() { @@ -332,21 +332,21 @@ public final class TaskManager { serverManager.heartbeat(); /* - * Master schedule tasks to suitable servers. - * Worker maybe become Master, so Master also need perform tasks assigned by - * previous Master when enableRoleElected is true. - * However, the master only needs to take the assignment, - * because the master stays the same when enableRoleElected is false. - * There is no suitable server when these tasks are created + * Master will schedule tasks to suitable servers. + * Note a Worker may become to a Master, so elected-Master also needs to + * execute tasks assigned by previous Master when enableRoleElected=true. + * However, when enableRoleElected=false, a Master is only set by the + * config assignment, assigned-Master always stays the same state. */ if (serverManager.master()) { - scheduler.scheduleTasks(); + scheduler.scheduleTasksOnMaster(); if (!this.enableRoleElected && !serverManager.onlySingleNode()) { + // assigned-Master + non-single-node don't need to execute tasks return; } } - // Schedule queued tasks scheduled to current server + // Execute queued tasks scheduled to current server scheduler.executeTasksOnWorker(serverManager.selfServerId()); // Cancel tasks scheduled to current server diff --git a/hugegraph-server/hugegraph-example/src/main/java/org/apache/hugegraph/example/ExampleUtil.java b/hugegraph-server/hugegraph-example/src/main/java/org/apache/hugegraph/example/ExampleUtil.java index 7cb148262..8317f9ba7 100644 --- a/hugegraph-server/hugegraph-example/src/main/java/org/apache/hugegraph/example/ExampleUtil.java +++ b/hugegraph-server/hugegraph-example/src/main/java/org/apache/hugegraph/example/ExampleUtil.java @@ -20,18 +20,17 @@ package org.apache.hugegraph.example; import java.io.File; import java.util.Iterator; import java.util.concurrent.TimeoutException; -import org.slf4j.Logger; import org.apache.hugegraph.HugeException; import org.apache.hugegraph.HugeFactory; import org.apache.hugegraph.HugeGraph; -import org.apache.hugegraph.backend.id.IdGenerator; import org.apache.hugegraph.dist.RegisterUtil; +import org.apache.hugegraph.masterelection.GlobalMasterInfo; import org.apache.hugegraph.perf.PerfUtil; import org.apache.hugegraph.task.HugeTask; import org.apache.hugegraph.task.TaskScheduler; -import org.apache.hugegraph.type.define.NodeRole; import org.apache.hugegraph.util.Log; +import org.slf4j.Logger; public class ExampleUtil { private static final Logger LOG = Log.logger(ExampleUtil.class); @@ -81,7 +80,7 @@ public class ExampleUtil { graph.clearBackend(); } graph.initBackend(); - graph.serverStarted(IdGenerator.of("server1"), NodeRole.MASTER); + graph.serverStarted(GlobalMasterInfo.master("server1")); return graph; } diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/BaseCoreTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/BaseCoreTest.java index d51e1b595..df9932ab8 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/BaseCoreTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/BaseCoreTest.java @@ -21,13 +21,12 @@ import java.util.Random; import org.apache.hugegraph.HugeGraph; import org.apache.hugegraph.HugeGraphParams; -import org.apache.hugegraph.backend.id.IdGenerator; import org.apache.hugegraph.backend.store.BackendFeatures; import org.apache.hugegraph.dist.RegisterUtil; +import org.apache.hugegraph.masterelection.GlobalMasterInfo; import org.apache.hugegraph.schema.SchemaManager; import org.apache.hugegraph.testutil.Utils; import org.apache.hugegraph.testutil.Whitebox; -import org.apache.hugegraph.type.define.NodeRole; import org.apache.hugegraph.util.Log; import org.apache.tinkerpop.gremlin.structure.Edge; import org.apache.tinkerpop.gremlin.structure.Vertex; @@ -66,7 +65,7 @@ public class BaseCoreTest { graph = Utils.open(); graph.clearBackend(); graph.initBackend(); - graph.serverStarted(IdGenerator.of("server1"), NodeRole.MASTER); + graph.serverStarted(GlobalMasterInfo.master("server-test")); } @AfterClass diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/MultiGraphsTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/MultiGraphsTest.java index 23fb122f4..3b468ba45 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/MultiGraphsTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/MultiGraphsTest.java @@ -31,8 +31,8 @@ import org.apache.hugegraph.backend.id.IdGenerator; import org.apache.hugegraph.backend.store.BackendStoreInfo; import org.apache.hugegraph.backend.store.rocksdb.RocksDBOptions; import org.apache.hugegraph.config.CoreOptions; -import org.apache.hugegraph.exception.ConnectionException; import org.apache.hugegraph.exception.ExistedException; +import org.apache.hugegraph.masterelection.GlobalMasterInfo; import org.apache.hugegraph.schema.EdgeLabel; import org.apache.hugegraph.schema.IndexLabel; import org.apache.hugegraph.schema.PropertyKey; @@ -40,7 +40,6 @@ import org.apache.hugegraph.schema.SchemaManager; import org.apache.hugegraph.schema.VertexLabel; import org.apache.hugegraph.testutil.Assert; import org.apache.hugegraph.testutil.Utils; -import org.apache.hugegraph.type.define.NodeRole; import org.apache.tinkerpop.gremlin.structure.T; import org.apache.tinkerpop.gremlin.structure.Vertex; import org.apache.tinkerpop.gremlin.structure.util.GraphFactory; @@ -87,9 +86,9 @@ public class MultiGraphsTest extends BaseCoreTest { graph.initBackend(); } HugeGraph g1 = graphs.get(0); - g1.serverStarted(IdGenerator.of("server-g2"), NodeRole.MASTER); + g1.serverStarted(GlobalMasterInfo.master("server-g2")); HugeGraph g2 = graphs.get(1); - g2.serverStarted(IdGenerator.of("server-g3"), NodeRole.MASTER); + g2.serverStarted(GlobalMasterInfo.master("server-g3")); SchemaManager schema = g1.schema(); @@ -209,8 +208,8 @@ public class MultiGraphsTest extends BaseCoreTest { } HugeGraph g1 = graphs.get(0); HugeGraph g2 = graphs.get(1); - g1.serverStarted(IdGenerator.of("server-g1c"), NodeRole.MASTER); - g2.serverStarted(IdGenerator.of("server-g2c"), NodeRole.MASTER); + g1.serverStarted(GlobalMasterInfo.master("server-g1c")); + g2.serverStarted(GlobalMasterInfo.master("server-g2c")); g1.schema().propertyKey("id").asInt().create(); g2.schema().propertyKey("id").asText().create(); diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/RoleElectionStateMachineTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/RoleElectionStateMachineTest.java index 9fbbb5e62..3eb059f06 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/RoleElectionStateMachineTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/RoleElectionStateMachineTest.java @@ -35,7 +35,7 @@ import org.apache.hugegraph.masterelection.ClusterRoleStore; import org.apache.hugegraph.masterelection.Config; import org.apache.hugegraph.masterelection.RoleElectionStateMachine; import org.apache.hugegraph.masterelection.StandardRoleElectionStateMachine; -import org.apache.hugegraph.masterelection.StateMachineCallback; +import org.apache.hugegraph.masterelection.RoleListener; import org.apache.hugegraph.masterelection.StateMachineContext; import org.apache.hugegraph.testutil.Assert; import org.apache.hugegraph.testutil.Utils; @@ -143,7 +143,7 @@ public class RoleElectionStateMachineTest { final int MAX_COUNT = 200; final List<LogEntry> logRecords = Collections.synchronizedList(new ArrayList<>(MAX_COUNT)); final List<String> masterNodes = Collections.synchronizedList(new ArrayList<>(MAX_COUNT)); - final StateMachineCallback callback = new StateMachineCallback() { + final RoleListener callback = new RoleListener() { @Override public void onAsRoleMaster(StateMachineContext context) { @@ -264,7 +264,7 @@ public class RoleElectionStateMachineTest { RoleElectionStateMachine stateMachine = new StandardRoleElectionStateMachine(config, clusterRoleStore); machines[1] = stateMachine; - stateMachine.apply(callback); + stateMachine.start(callback); stop.countDown(); }); @@ -273,7 +273,7 @@ public class RoleElectionStateMachineTest { RoleElectionStateMachine stateMachine = new StandardRoleElectionStateMachine(config, clusterRoleStore); machines[2] = stateMachine; - stateMachine.apply(callback); + stateMachine.start(callback); stop.countDown(); }); @@ -282,7 +282,7 @@ public class RoleElectionStateMachineTest { RoleElectionStateMachine stateMachine = new StandardRoleElectionStateMachine(config, clusterRoleStore); machines[3] = stateMachine; - stateMachine.apply(callback); + stateMachine.start(callback); stop.countDown(); }); diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/tinkerpop/TestGraph.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/tinkerpop/TestGraph.java index c6dcff4a8..415e80462 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/tinkerpop/TestGraph.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/tinkerpop/TestGraph.java @@ -24,27 +24,24 @@ import java.util.List; import java.util.Set; import org.apache.commons.configuration2.Configuration; -import org.apache.tinkerpop.gremlin.process.computer.GraphComputer; -import org.apache.tinkerpop.gremlin.structure.Edge; -import org.apache.tinkerpop.gremlin.structure.Graph; -import org.apache.tinkerpop.gremlin.structure.T; -import org.apache.tinkerpop.gremlin.structure.Transaction; -import org.apache.tinkerpop.gremlin.structure.Vertex; -import org.apache.tinkerpop.gremlin.structure.io.Io; - import org.apache.hugegraph.HugeGraph; -import org.apache.hugegraph.backend.id.Id; -import org.apache.hugegraph.backend.id.IdGenerator; import org.apache.hugegraph.backend.store.BackendStoreInfo; import org.apache.hugegraph.io.HugeGraphIoRegistry; import org.apache.hugegraph.io.HugeGraphSONModule; +import org.apache.hugegraph.masterelection.GlobalMasterInfo; import org.apache.hugegraph.perf.PerfUtil.Watched; import org.apache.hugegraph.schema.PropertyKey; import org.apache.hugegraph.schema.SchemaManager; import org.apache.hugegraph.task.TaskScheduler; import org.apache.hugegraph.testutil.Whitebox; import org.apache.hugegraph.type.define.IdStrategy; -import org.apache.hugegraph.type.define.NodeRole; +import org.apache.tinkerpop.gremlin.process.computer.GraphComputer; +import org.apache.tinkerpop.gremlin.structure.Edge; +import org.apache.tinkerpop.gremlin.structure.Graph; +import org.apache.tinkerpop.gremlin.structure.T; +import org.apache.tinkerpop.gremlin.structure.Transaction; +import org.apache.tinkerpop.gremlin.structure.Vertex; +import org.apache.tinkerpop.gremlin.structure.io.Io; import com.google.common.collect.ImmutableSet; @@ -85,8 +82,7 @@ public class TestGraph implements Graph { assert sysInfo.exists() && !this.graph.closed(); } - Id id = IdGenerator.of("server-tinkerpop"); - this.graph.serverStarted(id, NodeRole.MASTER); + this.graph.serverStarted(GlobalMasterInfo.master("server-tinkerpop")); this.initedBackend = true; } diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/SecurityManagerTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/SecurityManagerTest.java index 403bc62e9..ae431480c 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/SecurityManagerTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/SecurityManagerTest.java @@ -32,23 +32,22 @@ import java.util.HashMap; import java.util.Map; import java.util.concurrent.TimeoutException; -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.Test; - import org.apache.hugegraph.HugeException; import org.apache.hugegraph.HugeFactory; import org.apache.hugegraph.HugeGraph; -import org.apache.hugegraph.backend.id.IdGenerator; import org.apache.hugegraph.config.HugeConfig; import org.apache.hugegraph.job.GremlinJob; import org.apache.hugegraph.job.JobBuilder; +import org.apache.hugegraph.masterelection.GlobalMasterInfo; import org.apache.hugegraph.security.HugeSecurityManager; import org.apache.hugegraph.task.HugeTask; import org.apache.hugegraph.testutil.Assert; -import org.apache.hugegraph.type.define.NodeRole; import org.apache.hugegraph.unit.FakeObjects; import org.apache.hugegraph.util.JsonUtil; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + import com.google.common.collect.ImmutableMap; public class SecurityManagerTest { @@ -319,7 +318,7 @@ public class SecurityManagerTest { graph.clearBackend(); } graph.initBackend(); - graph.serverStarted(IdGenerator.of("server1"), NodeRole.MASTER); + graph.serverStarted(GlobalMasterInfo.master("server1")); return graph; }
