diff --git a/hugegraph-cluster-test/hugegraph-clustertest-dist/src/assembly/static/conf/hugegraph.properties.template b/hugegraph-cluster-test/hugegraph-clustertest-dist/src/assembly/static/conf/hugegraph.properties.template index 005031fe60..f97e365748 100644 --- a/hugegraph-cluster-test/hugegraph-clustertest-dist/src/assembly/static/conf/hugegraph.properties.template +++ b/hugegraph-cluster-test/hugegraph-clustertest-dist/src/assembly/static/conf/hugegraph.properties.template @@ -45,7 +45,6 @@ store=hugegraph pd.peers=$PD_PEERS_LIST$ # task config -task.scheduler_type=local task.schedule_period=10 task.retry=0 task.wait_timeout=10 diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java index 278542854b..1aed0434da 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java @@ -556,9 +556,9 @@ public class ServerOptions extends OptionHolder { public static final ConfigOption SERVER_ID = new ConfigOption<>( "server.id", - "The id of hugegraph-server.", - disallowEmpty(), - "server-1" + "The optional legacy id of hugegraph-server.", + null, + "" ); public static final ConfigOption SERVER_ROLE = new ConfigOption<>( 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 770e75cc74..ff04359649 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 @@ -58,7 +58,6 @@ 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.IdGenerator; import org.apache.hugegraph.backend.store.AbstractBackendStoreProvider; import org.apache.hugegraph.backend.store.BackendStoreInfo; import org.apache.hugegraph.config.ConfigOption; @@ -68,6 +67,7 @@ import org.apache.hugegraph.config.TypedOption; import org.apache.hugegraph.event.EventHub; import org.apache.hugegraph.exception.ExistedException; +import org.apache.hugegraph.exception.NotFoundException; import org.apache.hugegraph.exception.NotSupportException; import org.apache.hugegraph.io.HugeGraphSONModule; import org.apache.hugegraph.k8s.K8sDriver; @@ -195,8 +195,6 @@ public final class GraphManager { public GraphManager(HugeConfig conf, EventHub hub) { LOG.info("Init graph manager"); E.checkArgumentNotNull(conf, "The config can't be null"); - String server = conf.get(ServerOptions.SERVER_ID); - String role = conf.get(ServerOptions.SERVER_ROLE); this.config = conf; this.url = conf.get(ServerOptions.REST_SERVER_URL); @@ -206,10 +204,6 @@ public GraphManager(HugeConfig conf, EventHub hub) { conf.get(ServerOptions.SERVER_DEPLOY_IN_K8S); this.startIgnoreSingleGraphError = conf.get( ServerOptions.SERVER_START_IGNORE_SINGLE_GRAPH_ERROR); - E.checkArgument(server != null && !server.isEmpty(), - "The server name can't be null or empty"); - E.checkArgument(role != null && !role.isEmpty(), - "The server role can't be null or empty"); this.graphsDir = conf.get(ServerOptions.GRAPHS); this.cluster = conf.get(ServerOptions.CLUSTER); this.graphSpaces = new ConcurrentHashMap<>(); @@ -1557,6 +1551,9 @@ private void loadGraph(String name, String graphConfPath) { String raftGroupPeers = this.conf.get(ServerOptions.RAFT_GROUP_PEERS); config.addProperty(ServerOptions.RAFT_GROUP_PEERS.name(), raftGroupPeers); + + this.transferPdPeersConfig(config); + this.transferRoleWorkerConfig(config); Graph graph = GraphFactory.open(config); @@ -1575,6 +1572,19 @@ private void loadGraph(String name, String graphConfPath) { } } + private void transferPdPeersConfig(HugeConfig config) { + if (config.containsKey(CoreOptions.PD_PEERS.name())) { + return; + } + + String backend = config.get(CoreOptions.BACKEND); + boolean needPdPeers = this.conf.get(ServerOptions.USE_PD) || + StringUtils.equalsIgnoreCase("hstore", backend); + if (needPdPeers) { + config.addProperty(CoreOptions.PD_PEERS.name(), this.pdPeers); + } + } + private void transferRoleWorkerConfig(HugeConfig config) { config.setProperty(RoleElectionOptions.NODE_EXTERNAL_URL.name(), this.conf.get(ServerOptions.REST_SERVER_URL)); @@ -1635,23 +1645,21 @@ private void checkBackendVersionOrExit(HugeConfig config) { } private void initNodeRole() { - String id = config.get(ServerOptions.SERVER_ID); + boolean enableRoleElection = config.get( + ServerOptions.ENABLE_SERVER_ROLE_ELECTION); + if (enableRoleElection) { + LOG.warn("The server.role_election option is deprecated and no " + + "longer supported (removed with server_info persistence). " + + "The configured server.role is still used for local node " + + "role initialization. Set server.role_election=false to " + + "suppress this warning."); + } + String role = config.get(ServerOptions.SERVER_ROLE); - E.checkArgument(StringUtils.isNotEmpty(id), - "The server name can't be null or empty"); E.checkArgument(StringUtils.isNotEmpty(role), "The server role can't be null or empty"); NodeRole nodeRole = NodeRole.valueOf(role.toUpperCase()); - boolean supportRoleElection = !nodeRole.computer() && - this.supportRoleElection() && - config.get(ServerOptions.ENABLE_SERVER_ROLE_ELECTION); - if (supportRoleElection) { - // Init any server as Worker role, then do role election - nodeRole = NodeRole.WORKER; - } - - this.globalNodeRoleInfo.initNodeId(IdGenerator.of(id)); this.globalNodeRoleInfo.initNodeRole(nodeRole); } @@ -1663,7 +1671,7 @@ private void serverStarted(HugeConfig conf) { } if (!this.globalNodeRoleInfo.nodeRole().computer() && this.supportRoleElection() && config.get(ServerOptions.ENABLE_SERVER_ROLE_ELECTION)) { - this.initRoleStateMachine(); + LOG.info("Skip role state machine init (deprecated with server_info)"); } } @@ -1937,26 +1945,29 @@ public Set getServiceUrls(String graphSpace, String service, public HugeGraph graph(String graphSpace, String name) { String key = String.join(DELIMITER, graphSpace, name); Graph graph = this.graphs.get(key); - if (graph == null && isPDEnabled()) { - Map> configs = - this.metaManager.graphConfigs(graphSpace); - // If current server registered graph space is not DEFAULT, only load graph creation - // under registered graph space - if (!configs.containsKey(key) || - (!"DEFAULT".equals(this.serviceGraphSpace) && - !graphSpace.equals(this.serviceGraphSpace))) { - return null; + if (graph == null) { + if (isPDEnabled()) { + Map> configs = + this.metaManager.graphConfigs(graphSpace); + // If current server registered graph space is not DEFAULT, only load graph creation + // under registered graph space + if (!configs.containsKey(key) || + (!"DEFAULT".equals(this.serviceGraphSpace) && + !graphSpace.equals(this.serviceGraphSpace))) { + return null; + } + Map config = configs.get(key); + String creator = String.valueOf(config.get("creator")); + Date createTime = parseDate(config.get("create_time")); + Date updateTime = parseDate(config.get("update_time")); + HugeGraph graph1 = this.createGraph(graphSpace, name, + creator, config, false); + graph1.createTime(createTime); + graph1.updateTime(updateTime); + this.graphs.put(key, graph1); + return graph1; } - Map config = configs.get(key); - String creator = String.valueOf(config.get("creator")); - Date createTime = parseDate(config.get("create_time")); - Date updateTime = parseDate(config.get("update_time")); - HugeGraph graph1 = this.createGraph(graphSpace, name, - creator, config, false); - graph1.createTime(createTime); - graph1.updateTime(updateTime); - this.graphs.put(key, graph1); - return graph1; + throw new NotFoundException(String.format("Graph '%s' does not exist", name)); } else if (graph instanceof HugeGraph) { return (HugeGraph) graph; } 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 aaacb799c9..6f66b4b8a9 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 @@ -177,7 +177,6 @@ public class StandardHugeGraph implements HugeGraph { private final BackendStoreProvider storeProvider; private final TinkerPopTransaction tx; private final RamTable ramtable; - private final String schedulerType; private volatile boolean started; private volatile boolean closed; private volatile GraphMode mode; @@ -226,11 +225,18 @@ public StandardHugeGraph(HugeConfig config) { this.taskManager = TaskManager.instance(); this.name = config.get(CoreOptions.STORE); + + // Keep old config files upgrade-safe while ignoring the legacy scheduler. + if (config.containsKey("task.scheduler_type")) { + LOG.warn("Config key 'task.scheduler_type' is deprecated and " + + "ignored. The scheduler is auto-selected by backend " + + "type (hstore -> distributed, others -> local)."); + } + this.started = false; this.closed = false; this.mode = GraphMode.NONE; this.readMode = GraphReadMode.OLTP_ONLY; - this.schedulerType = config.get(CoreOptions.SCHEDULER_TYPE); // Init process-wide static configs before lock, so that validation // failures won't leave stale lock groups in LockManager. @@ -323,6 +329,7 @@ public String backend() { return this.storeProvider.type(); } + @Override public BackendStoreInfo backendStoreInfo() { // Just for trigger Tx.getOrNewTransaction, then load 3 stores // TODO: pass storeProvider.metaStore() @@ -340,11 +347,10 @@ public void serverStarted(GlobalMasterInfo nodeInfo) { LOG.info("Init system info for graph '{}'", this.spaceGraphName()); this.initSystemInfo(); - LOG.info("Init server info [{}-{}] for graph '{}'...", - nodeInfo.nodeId(), nodeInfo.nodeRole(), this.spaceGraphName()); - this.serverInfoManager().initServerInfo(nodeInfo); - - this.initRoleStateMachine(nodeInfo.nodeId()); + if (nodeInfo != null && nodeInfo.nodeId() != null) { + this.serverInfoManager().initServerInfo(nodeInfo); + this.initRoleStateMachine(nodeInfo.nodeId()); + } // TODO: check necessary? LOG.info("Check olap property-key tables for graph '{}'", this.spaceGraphName()); @@ -473,6 +479,7 @@ public void updateTime(Date updateTime) { this.updateTime = updateTime; } + @Override public void waitStarted() { // Just for trigger Tx.getOrNewTransaction, then load 3 stores this.schemaTransaction(); @@ -489,9 +496,7 @@ public void initBackend() { try { this.storeProvider.init(); /* - * NOTE: The main goal is to write the serverInfo to the central - * node, such as etcd, and also create the system schema in memory, - * which has no side effects + * NOTE: Create system schema in memory, which has no side effects. */ this.initSystemInfo(); } finally { @@ -532,8 +537,7 @@ public void truncateBackend() { LockUtil.lock(this.spaceGraphName(), LockUtil.GRAPH_LOCK); try { this.storeProvider.truncate(); - // TODO: remove this after serverinfo saved in etcd - this.serverStarted(this.serverInfoManager().globalNodeRoleInfo()); + this.serverStarted(null); } finally { LockUtil.unlock(this.spaceGraphName(), LockUtil.GRAPH_LOCK); } @@ -555,7 +559,6 @@ public KvStore kvStore() { public void initSystemInfo() { try { this.taskScheduler().init(); - this.serverInfoManager().init(); this.authManager().init(); } finally { this.closeTx(); @@ -1640,7 +1643,9 @@ public void submitEphemeralJob(EphemeralJob job) { @Override public String schedulerType() { - return StandardHugeGraph.this.schedulerType; + // Use distributed scheduler for hstore backend, otherwise use local + // After the merger of rocksdb and hstore, consider whether to change this logic + return StandardHugeGraph.this.isHstore() ? "distributed" : "local"; } } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java index 07109580a0..9aa60e398a 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java @@ -313,13 +313,7 @@ public class CoreOptions extends OptionHolder { rangeInt(1, 500), 1 ); - public static final ConfigOption SCHEDULER_TYPE = - new ConfigOption<>( - "task.scheduler_type", - "The type of scheduler used in distribution system.", - allowValues("local", "distributed"), - "local" - ); + public static final ConfigOption TASK_SYNC_DELETION = new ConfigOption<>( "task.sync_deletion", 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 c345c50e60..4856744459 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 @@ -22,7 +22,7 @@ import org.apache.hugegraph.type.define.NodeRole; import org.apache.hugegraph.util.E; -// TODO: rename to GlobalNodeRoleInfo +// TODO: We need to completely delete the startup of master-worker public final class GlobalMasterInfo { private static final NodeInfo NO_MASTER = new NodeInfo(false, ""); diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardRoleListener.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardRoleListener.java index dbbea6d91e..ce91ca2d5f 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardRoleListener.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardRoleListener.java @@ -36,7 +36,6 @@ public class StandardRoleListener implements RoleListener { public StandardRoleListener(TaskManager taskManager, GlobalMasterInfo roleInfo) { this.taskManager = taskManager; - this.taskManager.enableRoleElection(); this.roleInfo = roleInfo; this.selfIsMaster = false; } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java index 3929f8414d..8b11db9b6c 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java @@ -18,8 +18,10 @@ package org.apache.hugegraph.task; import java.util.Iterator; +import java.util.Set; import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -48,6 +50,7 @@ import org.slf4j.Logger; public class DistributedTaskScheduler extends TaskAndResultScheduler { + private static final Logger LOG = Log.logger(DistributedTaskScheduler.class); private final long schedulePeriod; private final ExecutorService taskDbExecutor; @@ -64,6 +67,7 @@ public class DistributedTaskScheduler extends TaskAndResultScheduler { private final AtomicBoolean closed = new AtomicBoolean(true); private final ConcurrentHashMap> runningTasks = new ConcurrentHashMap<>(); + private final Set deletingTasks = ConcurrentHashMap.newKeySet(); public DistributedTaskScheduler(HugeGraphParams graph, ScheduledThreadPoolExecutor schedulerExecutor, @@ -118,6 +122,11 @@ private static boolean sleep(long ms) { public void cronSchedule() { // Perform periodic scheduling tasks + // Check closed flag first to exit early + if (this.closed.get()) { + return; + } + if (!this.graph.started() || this.graph.closed()) { return; } @@ -179,14 +188,17 @@ public void cronSchedule() { while (!this.closed.get() && cancellings.hasNext()) { Id cancellingId = cancellings.next().id(); - if (runningTasks.containsKey(cancellingId)) { - HugeTask cancelling = runningTasks.get(cancellingId); + HugeTask cancelling = runningTasks.get(cancellingId); + if (cancelling != null) { initTaskParams(cancelling); LOG.info("Try to cancel task({})@({}/{})", cancelling.id(), this.graphSpace, this.graphName); - cancelling.cancel(true); - - runningTasks.remove(cancellingId); + if (!cancelling.cancel(true)) { + // Task already completed normally; force CANCELLED so + // it doesn't stay stuck in CANCELLING forever. + updateStatusWithLock(cancellingId, TaskStatus.CANCELLING, + TaskStatus.CANCELLED); + } } else { // Local no execution task, but the current task has no nodes executing. if (!isLockedTask(cancellingId.toString())) { @@ -202,15 +214,10 @@ public void cronSchedule() { while (!this.closed.get() && deletings.hasNext()) { Id deletingId = deletings.next().id(); - if (runningTasks.containsKey(deletingId)) { - HugeTask deleting = runningTasks.get(deletingId); - initTaskParams(deleting); + HugeTask deleting = runningTasks.get(deletingId); + if (deleting != null) { + this.markTaskDeleting(deleting); deleting.cancel(true); - - // Delete storage information - deleteFromDB(deletingId); - - runningTasks.remove(deletingId); } else { // Local has no task execution, but the current task has no nodes executing anymore. if (!isLockedTask(deletingId.toString())) { @@ -253,6 +260,10 @@ public Future schedule(HugeTask task) { return this.ephemeralTaskExecutor.submit(task); } + // Validate task state before saving to ensure correct exception type + E.checkState(task.type() != null, "Task type can't be null"); + E.checkState(task.name() != null, "Task name can't be null"); + // Process schema task // Handle gremlin task // Handle OLAP calculation tasks @@ -284,14 +295,63 @@ protected void initTaskParams(HugeTask task) { } } + /** + * Note: This method will update the status of the input task. + * + * @param task + * @param + */ @Override public void cancel(HugeTask task) { - // Update status to CANCELLING - if (!task.completed()) { - // Task not completed, can only execute status not CANCELLING - this.updateStatus(task.id(), null, TaskStatus.CANCELLING); + E.checkArgumentNotNull(task, "Task can't be null"); + + if (task.completed() || task.cancelling()) { + return; + } + + LOG.info("Cancel task '{}' in status {}", task.id(), task.status()); + + // Check if task is running locally, cancel it directly if so + HugeTask runningTask = this.runningTasks.get(task.id()); + if (runningTask != null) { + boolean cancelled = runningTask.cancel(true); + if (cancelled) { + task.overwriteStatus(TaskStatus.CANCELLED); + this.save(runningTask); + } + LOG.info("Cancel local running task '{}' result: {}", task.id(), cancelled); + return; + } + + // Task not running locally, update status to CANCELLING + // for cronSchedule() or other nodes to handle + TaskStatus currentStatus = task.status(); + if (this.updateStatus(task.id(), currentStatus, TaskStatus.CANCELLING)) { + task.overwriteStatus(TaskStatus.CANCELLING); } else { - LOG.info("cancel task({}) error, task has completed", task.id()); + // Status race: re-read from DB and retry with fresh status + HugeTask reloaded; + try { + reloaded = this.taskWithoutResult(task.id()); + } catch (NotFoundException e) { + LOG.info("Task '{}' already deleted, skip cancel", task.id()); + return; + } + TaskStatus stored = reloaded.status(); + if (stored != TaskStatus.CANCELLING && + !TaskStatus.COMPLETED_STATUSES.contains(stored)) { + if (this.updateStatus(task.id(), stored, TaskStatus.CANCELLING)) { + task.overwriteStatus(TaskStatus.CANCELLING); + LOG.info("Retry cancel task '{}' succeeded (stored was {})", + task.id(), stored); + } else { + LOG.warn("Failed to cancel task '{}', re-read status {} changed again", + task.id(), stored); + } + } else { + LOG.info("Task '{}' already {}/terminal, skip cancel", + task.id(), stored); + } } } @@ -302,38 +362,166 @@ public void init() { protected HugeTask deleteFromDB(Id id) { // Delete Task from DB, without checking task status - return this.call(() -> { - Iterator vertices = this.tx().queryTaskInfos(id); - HugeVertex vertex = (HugeVertex) QueryResults.one(vertices); - if (vertex == null) { + try { + return this.call(() -> { + Iterator vertices = this.tx().queryTaskInfos(id); + HugeVertex vertex = (HugeVertex) QueryResults.one(vertices); + if (vertex == null) { + this.deleteTaskResultFromTx(id); + return null; + } + HugeTask result = HugeTask.fromVertex(vertex, false); + // Keep the task vertex as a retryable tombstone until its result + // vertex is removed; cronSchedule() can rediscover DELETING tasks. this.deleteTaskResultFromTx(id); + this.tx().removeTaskVertex(vertex); + return result; + }); + } finally { + if (!this.runningTasks.containsKey(id)) { + this.deletingTasks.remove(id); + } + } + } + + @Override + public void save(HugeTask task) { + E.checkArgumentNotNull(task, "Task can't be null"); + if (this.deletingTasks.contains(task.id())) { + LOG.info("Skip saving task({})@({}/{}) because it is deleting", + task.id(), this.graphSpace, this.graphName); + return; + } + + String rawResult = task.result(); + Boolean saved = this.call(() -> { + if (task.status() != TaskStatus.DELETING && + this.storedTaskDeleting(task.id())) { + LOG.info("Skip saving task({})@({}/{}) because stored status " + + "is DELETING", task.id(), this.graphSpace, + this.graphName); + return false; + } + HugeVertex vertex = this.tx().constructTaskVertex(task); + this.tx().deleteIndex(vertex); + this.tx().addVertex(vertex); + return true; + }); + + if (!saved || rawResult == null) { + return; + } + + this.call(() -> { + if (this.deletingTasks.contains(task.id()) || + !this.storedTaskAllowsResultSave(task.id())) { + LOG.info("Skip saving task({}) result@({}/{}) because it is " + + "deleting or missing", task.id(), this.graphSpace, + this.graphName); return null; } - HugeTask result = HugeTask.fromVertex(vertex, false); - // Keep the task vertex as a retryable tombstone until its result - // vertex is removed; cronSchedule() can rediscover DELETING tasks. - this.deleteTaskResultFromTx(id); - this.tx().removeTaskVertex(vertex); - return result; + HugeTaskResult result = + new HugeTaskResult(HugeTaskResult.genId(task.id())); + result.result(rawResult); + + HugeVertex vertex = this.tx().constructTaskResultVertex(result); + return this.tx().addVertex(vertex); + }); + } + + private void markTaskDeleting(HugeTask task) { + this.deletingTasks.add(task.id()); + initTaskParams(task); + HugeTask deleting; + synchronized (task) { + task.overwriteStatus(TaskStatus.DELETING); + deleting = task.copyWithoutResult(); + } + this.saveTaskWithoutResult(deleting); + } + + private boolean storedTaskDeleting(Id id) { + Iterator vertices = this.tx().queryTaskInfos(id); + Vertex vertex = QueryResults.one(vertices); + if (vertex == null) { + return false; + } + HugeTask task = HugeTask.fromVertex(vertex, false); + return task.status() == TaskStatus.DELETING; + } + + private boolean storedTaskAllowsResultSave(Id id) { + Iterator vertices = this.tx().queryTaskInfos(id); + Vertex vertex = QueryResults.one(vertices); + if (vertex == null) { + return false; + } + HugeTask task = HugeTask.fromVertex(vertex, false); + return task.status() != TaskStatus.DELETING; + } + + private void saveTaskWithoutResult(HugeTask task) { + this.call(() -> { + HugeVertex vertex = this.tx().constructTaskVertex(task); + this.tx().deleteIndex(vertex); + return this.tx().addVertex(vertex); }); } @Override public HugeTask delete(Id id, boolean force) { - if (!force) { - // Change status to DELETING, perform the deletion operation through automatic - // scheduling. - this.updateStatus(id, null, TaskStatus.DELETING); - return null; - } else { - HugeTask task = this.taskWithoutResult(id); - if (task != null && task.status() != TaskStatus.DELETING) { - initTaskParams(task); - task.overwriteStatus(TaskStatus.DELETING); - this.save(task); + HugeTask task = this.taskWithoutResult(id); + HugeTask running = this.runningTasks.get(id); + + if (running != null) { + this.markTaskDeleting(running); + running.cancel(true); + @SuppressWarnings("unchecked") + HugeTask result = (HugeTask) running; + return result; + } + + if (!force && !task.completed()) { + // Can't safely mark a remotely running task without owning its + // lock; the owner may otherwise overwrite DELETING on final save. + LockResult lockResult = tryLockTask(id.asString()); + checkDeleteLock(id, lockResult); + try { + this.markTaskDeleting(task); + } finally { + unlockTask(id.asString(), lockResult); + } + @SuppressWarnings("unchecked") + HugeTask result = (HugeTask) task; + return result; + } + + if (!task.completed()) { + LockResult lockResult = tryLockTask(id.asString()); + checkDeleteLock(id, lockResult); + try { + if (task.status() != TaskStatus.DELETING) { + this.markTaskDeleting(task); + } + return this.deleteFromDB(id); + } finally { + unlockTask(id.asString(), lockResult); } - return this.deleteFromDB(id); } + + // Write DELETING status before attempting physical delete so that a + // failed result deletion is recoverable via cronSchedule(). + if (task.status() != TaskStatus.DELETING) { + this.markTaskDeleting(task); + } + + return this.deleteFromDB(id); + } + + private static void checkDeleteLock(Id id, LockResult lockResult) { + E.checkState(lockResult.lockSuccess(), + "Can't delete task '%s' because it is locked by another " + + "server, please retry later", id); } @Override @@ -345,6 +533,26 @@ public boolean close() { // set closed this.closed.set(true); + // cancel cron thread + if (!cronFuture.isDone() && !cronFuture.isCancelled()) { + cronFuture.cancel(false); + } + + // Wait behind the scheduler thread to ensure any running cron task is completed + try { + Future barrier = this.schedulerExecutor.submit(() -> { + // pass + }); + barrier.get(schedulePeriod + 5, TimeUnit.SECONDS); + } catch (TimeoutException e) { + LOG.warn("Cron task did not complete in time when closing scheduler"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOG.warn("Interrupted while waiting for cron task to complete", e); + } catch (ExecutionException e) { + LOG.warn("Exception while waiting for cron task to complete", e); + } + // cancel all running tasks for (HugeTask task : this.runningTasks.values()) { LOG.info("cancel task({}) @({}/{}) when closing scheduler", @@ -356,11 +564,7 @@ public boolean close() { this.waitUntilAllTasksCompleted(10); } catch (TimeoutException e) { LOG.warn("Tasks not completed when close distributed task scheduler", e); - } - - // cancel cron thread - if (!cronFuture.isDone() && !cronFuture.isCancelled()) { - cronFuture.cancel(false); + return false; } if (!this.taskDbExecutor.isShutdown()) { @@ -373,7 +577,9 @@ public boolean close() { this.graph.closeTx(); }); } - return true; + + // TODO: serverInfoManager section should be removed in the future. + return this.serverManager().close(); } @Override @@ -447,9 +653,7 @@ public void waitUntilAllTasksCompleted(long seconds) @Override public void checkRequirement(String op) { - if (!this.serverManager().selfIsMaster()) { - throw new HugeException("Can't %s task on non-master server", op); - } + // Distributed scheduler uses task locks to coordinate workers. } @Override @@ -654,6 +858,7 @@ public void run() { LOG.warn("exception when execute task", t); } finally { runningTasks.remove(task.id()); + deletingTasks.remove(task.id()); unlockTask(task.id().asString(), lockResult); LOG.info("task({}) finished.", task.id().toString()); diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/HugeServerInfo.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/HugeServerInfo.java index 71feb3f688..f0485f6656 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/HugeServerInfo.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/HugeServerInfo.java @@ -209,14 +209,6 @@ public static HugeServerInfo fromVertex(Vertex vertex) { return serverInfo; } - public boolean suitableFor(HugeTask task, long now) { - if (task.computer() != this.role.computer()) { - return false; - } - return this.updateTime.getTime() + EXPIRED_INTERVAL >= now && - this.load() + task.load() <= this.maxLoad; - } - public static Schema schema(HugeGraphParams graph) { return new Schema(graph); } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/HugeTask.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/HugeTask.java index d5d703928a..f63d4c0762 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/HugeTask.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/HugeTask.java @@ -319,12 +319,20 @@ public void run() { public boolean cancel(boolean mayInterruptIfRunning) { // NOTE: Gremlin sleep() can't be interrupted by default // https://mrhaki.blogspot.com/2016/10/groovy-goodness-interrupted-sleeping.html + TaskStatus prevStatus = this.status; boolean cancelled = super.cancel(mayInterruptIfRunning); if (!cancelled) { return cancelled; } try { + // If the task is being deleted, don't transition to CANCELLED or + // fire cancelled() callback — let cronSchedule() handle the actual + // DB deletion. Prevents an async save() from resurrecting the task + // after deleteFromDB() has already removed it. + if (prevStatus == TaskStatus.DELETING) { + return cancelled; + } if (this.status(TaskStatus.CANCELLED)) { // Callback for saving status to store this.callable.cancelled(); 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 1560ec0b8f..dac78f2fcc 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 @@ -17,55 +17,26 @@ package org.apache.hugegraph.task; -import static org.apache.hugegraph.backend.query.Query.NO_LIMIT; - -import java.util.Collection; -import java.util.Iterator; -import java.util.Map; import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; import org.apache.hugegraph.HugeException; -import org.apache.hugegraph.HugeGraph; import org.apache.hugegraph.HugeGraphParams; import org.apache.hugegraph.backend.id.Id; import org.apache.hugegraph.backend.id.IdGenerator; -import org.apache.hugegraph.backend.page.PageInfo; -import org.apache.hugegraph.backend.query.Condition; -import org.apache.hugegraph.backend.query.ConditionQuery; -import org.apache.hugegraph.backend.query.QueryResults; 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; -import org.apache.hugegraph.type.HugeType; -import org.apache.hugegraph.type.define.HugeKeys; import org.apache.hugegraph.type.define.NodeRole; -import org.apache.hugegraph.util.DateUtil; import org.apache.hugegraph.util.E; -import org.apache.hugegraph.util.Log; -import org.apache.tinkerpop.gremlin.structure.Vertex; -import org.slf4j.Logger; - -import com.google.common.collect.ImmutableMap; public class ServerInfoManager { - private static final Logger LOG = Log.logger(ServerInfoManager.class); - - public static final long MAX_SERVERS = 100000L; - public static final long PAGE_SIZE = 10L; - private final HugeGraphParams graph; private final ExecutorService dbExecutor; private volatile GlobalMasterInfo globalNodeInfo; - private volatile boolean onlySingleNode; private volatile boolean closed; public ServerInfoManager(HugeGraphParams graph, ExecutorService dbExecutor) { @@ -77,28 +48,17 @@ public ServerInfoManager(HugeGraphParams graph, ExecutorService dbExecutor) { this.globalNodeInfo = null; - this.onlySingleNode = false; this.closed = false; } public void init() { - HugeServerInfo.schema(this.graph).initSchemaIfNeeded(); + // ServerInfo is soft-disabled; keep this method for compatibility. } public synchronized boolean close() { + // ServerInfo persistence is soft-deprecated; init() and heartbeat() + // are no-ops, so there's nothing to clean up in close(). this.closed = true; - if (!this.dbExecutor.isShutdown()) { - this.removeSelfServerInfo(); - this.call(() -> { - try { - this.tx().close(); - } catch (ConnectionException ignored) { - // ConnectionException means no connection established - } - this.graph.closeTx(); - return null; - }); - } return true; } @@ -106,53 +66,14 @@ public synchronized void initServerInfo(GlobalMasterInfo nodeInfo) { E.checkArgument(nodeInfo != null, "The global node info can't be null"); this.globalNodeInfo = nodeInfo; - - Id serverId = this.selfNodeId(); - HugeServerInfo existed = this.serverInfo(serverId); - if (existed != null && existed.alive()) { - final long now = DateUtil.now().getTime(); - if (existed.expireTime() > now + 30 * 1000) { - LOG.info("The node time maybe skew very much: {}", existed); - throw new HugeException("The server with name '%s' maybe skew very much", serverId); - } - try { - Thread.sleep(existed.expireTime() - now + 1); - } catch (InterruptedException e) { - throw new HugeException("Interrupted when waiting for server info expired", e); - } - } - E.checkArgument(existed == null || !existed.alive(), - "The server with name '%s' already in cluster", serverId); - - if (nodeInfo.nodeRole().master()) { - String page = this.supportsPaging() ? PageInfo.PAGE_NONE : null; - do { - Iterator 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()); - } - if (page != null) { - page = PageInfo.pageInfo(servers); - } - } while (page != null); - } - - // TODO: save ServerInfo to AuthServer - this.saveServerInfo(this.selfNodeId(), this.selfNodeRole()); } public synchronized void changeServerRole(NodeRole nodeRole) { - if (this.closed) { + if (this.closed || this.globalNodeInfo == null) { return; } this.globalNodeInfo.changeNodeRole(nodeRole); - - // TODO: save ServerInfo to AuthServer - this.saveServerInfo(this.selfNodeId(), this.selfNodeRole()); } public GlobalMasterInfo globalNodeRoleInfo() { @@ -163,9 +84,13 @@ public Id selfNodeId() { if (this.globalNodeInfo == null) { return null; } + Id nodeId = this.globalNodeInfo.nodeId(); + if (nodeId == null) { + return null; + } // Scope server id to graph to avoid cross-graph collision return IdGenerator.of(this.graph.spaceGraphName() + "/" + - this.globalNodeInfo.nodeId().asString()); + nodeId.asString()); } public NodeRole selfNodeRole() { @@ -179,98 +104,8 @@ public boolean selfIsMaster() { return this.selfNodeRole() != null && this.selfNodeRole().master(); } - public boolean onlySingleNode() { - // Only exists one node in the whole master - return this.onlySingleNode; - } - public synchronized void heartbeat() { - assert this.graphIsReady(); - - HugeServerInfo serverInfo = this.selfServerInfo(); - if (serverInfo != null) { - // Update heartbeat time for this server - serverInfo.updateTime(DateUtil.now()); - this.save(serverInfo); - return; - } - - /* ServerInfo is missing */ - if (this.selfNodeId() == null) { - // Ignore if ServerInfo is not initialized - LOG.info("ServerInfo is missing: {}, may not be initialized yet", this.selfNodeId()); - return; - } - if (this.selfIsMaster()) { - // On the master node, just wait for ServerInfo re-init - LOG.warn("ServerInfo is missing: {}, may be cleared before", this.selfNodeId()); - return; - } - /* - * Missing server info on non-master node, may be caused by graph - * truncated on master node then synced by raft. - * TODO: we just patch it here currently, to be improved. - */ - serverInfo = this.saveServerInfo(this.selfNodeId(), this.selfNodeRole()); - assert serverInfo != null; - } - - public synchronized void decreaseLoad(int load) { - assert load > 0 : load; - HugeServerInfo serverInfo = this.selfServerInfo(); - serverInfo.increaseLoad(-load); - this.save(serverInfo); - } - - public int calcMaxLoad() { - // TODO: calc max load based on CPU and Memory resources - return 10000; - } - - protected boolean graphIsReady() { - return !this.closed && this.graph.started() && this.graph.initialized(); - } - - protected synchronized HugeServerInfo pickWorkerNode(Collection servers, - HugeTask task) { - HugeServerInfo master = null; - HugeServerInfo serverWithMinLoad = null; - int minLoad = Integer.MAX_VALUE; - boolean hasWorkerNode = false; - long now = DateUtil.now().getTime(); - - // Iterate servers to find suitable one - for (HugeServerInfo server : servers) { - if (!server.alive()) { - continue; - } - if (server.role().master()) { - master = server; - continue; - } - hasWorkerNode = true; - if (!server.suitableFor(task, now)) { - continue; - } - if (server.load() < minLoad) { - minLoad = server.load(); - serverWithMinLoad = server; - } - } - - boolean singleNode = !hasWorkerNode; - if (singleNode != this.onlySingleNode) { - LOG.info("Switch only_single_node to {}", singleNode); - this.onlySingleNode = singleNode; - } - - // Only schedule to master if there are no workers and master are suitable - if (!hasWorkerNode) { - if (master != null && master.suitableFor(task, now)) { - serverWithMinLoad = master; - } - } - return serverWithMinLoad; + // ServerInfo heartbeat is deprecated for local scheduling. } private GraphTransaction tx() { @@ -278,57 +113,6 @@ private GraphTransaction tx() { return this.graph.systemTransaction(); } - private HugeServerInfo saveServerInfo(Id serverId, NodeRole serverRole) { - HugeServerInfo serverInfo = new HugeServerInfo(serverId, serverRole); - serverInfo.maxLoad(this.calcMaxLoad()); - this.save(serverInfo); - - LOG.info("Init server info: {}", serverInfo); - return serverInfo; - } - - private Id save(HugeServerInfo serverInfo) { - return this.call(() -> { - // Construct vertex from server info - HugeServerInfo.Schema schema = HugeServerInfo.schema(this.graph); - if (!schema.existVertexLabel(HugeServerInfo.P.SERVER)) { - throw new HugeException("Schema is missing for %s '%s'", - HugeServerInfo.P.SERVER, serverInfo); - } - HugeVertex vertex = this.tx().constructVertex(false, serverInfo.asArray()); - // Add or update server info in backend store - vertex = this.tx().addVertex(vertex); - return vertex.id(); - }); - } - - private int save(Collection serverInfos) { - return this.call(() -> { - if (serverInfos.isEmpty()) { - return 0; - } - HugeServerInfo.Schema schema = HugeServerInfo.schema(this.graph); - if (!schema.existVertexLabel(HugeServerInfo.P.SERVER)) { - throw new HugeException("Schema is missing for %s", HugeServerInfo.P.SERVER); - } - // Save server info in batch - GraphTransaction tx = this.tx(); - int updated = 0; - for (HugeServerInfo server : serverInfos) { - if (!server.updated()) { - continue; - } - HugeVertex vertex = tx.constructVertex(false, server.asArray()); - tx.addVertex(vertex); - updated++; - } - // NOTE: actually it is auto-commit, to be improved - tx.commitOrRollback(); - - return updated; - }); - } - private V call(Callable callable) { assert !Thread.currentThread().getName().startsWith( "server-info-db-worker") : "can't call by itself"; @@ -342,110 +126,4 @@ private V call(Callable callable) { e, e.toString()); } } - - private HugeServerInfo selfServerInfo() { - HugeServerInfo selfServerInfo = this.serverInfo(this.selfNodeId()); - if (selfServerInfo == null && this.selfNodeId() != null) { - LOG.warn("ServerInfo is missing: {}", this.selfNodeId()); - } - return selfServerInfo; - } - - private HugeServerInfo serverInfo(Id serverId) { - return this.call(() -> { - Iterator vertices = this.tx().queryServerInfos(serverId); - Vertex vertex = QueryResults.one(vertices); - if (vertex == null) { - return null; - } - return HugeServerInfo.fromVertex(vertex); - }); - } - - private HugeServerInfo removeSelfServerInfo() { - /* - * Check this.selfServerId != null to avoid graph.initialized() call. - * NOTE: graph.initialized() may throw exception if we can't connect to - * backend store, initServerInfo() is not called in this case, so - * this.selfServerId is null at this time. - */ - if (this.selfNodeId() != null && this.graph.initialized()) { - return this.removeServerInfo(this.selfNodeId()); - } - return null; - } - - private HugeServerInfo removeServerInfo(Id serverId) { - if (serverId == null) { - return null; - } - LOG.info("Remove server info: {}", serverId); - return this.call(() -> { - Iterator vertices = this.tx().queryServerInfos(serverId); - Vertex vertex = QueryResults.one(vertices); - if (vertex == null) { - return null; - } - this.tx().removeVertex((HugeVertex) vertex); - return HugeServerInfo.fromVertex(vertex); - }); - } - - protected void updateServerInfos(Collection serverInfos) { - this.save(serverInfos); - } - - protected Collection allServerInfos() { - Iterator infos = this.serverInfos(NO_LIMIT, null); - try (ListIterator iter = new ListIterator<>( - MAX_SERVERS, infos)) { - return iter.list(); - } catch (Exception e) { - throw new HugeException("Failed to close server info iterator", e); - } - } - - protected Iterator serverInfos(String page) { - return this.serverInfos(ImmutableMap.of(), PAGE_SIZE, page); - } - - protected Iterator serverInfos(long limit, String page) { - return this.serverInfos(ImmutableMap.of(), limit, page); - } - - private Iterator serverInfos(Map conditions, - long limit, String page) { - return this.call(() -> { - ConditionQuery query; - if (this.graph.backendStoreFeatures().supportsTaskAndServerVertex()) { - query = new ConditionQuery(HugeType.SERVER); - } else { - query = new ConditionQuery(HugeType.VERTEX); - } - if (page != null) { - query.page(page); - } - - HugeGraph graph = this.graph.graph(); - VertexLabel vl = graph.vertexLabel(HugeServerInfo.P.SERVER); - query.eq(HugeKeys.LABEL, vl.id()); - for (Map.Entry entry : conditions.entrySet()) { - PropertyKey pk = graph.propertyKey(entry.getKey()); - query.query(Condition.eq(pk.id(), entry.getValue())); - } - query.showHidden(true); - if (limit != NO_LIMIT) { - query.limit(limit); - } - Iterator vertices = this.tx().queryServerInfos(query); - Iterator servers = - new MapperIterator<>(vertices, HugeServerInfo::fromVertex); - // Convert iterator to list to avoid across thread tx accessed - return QueryResults.toList(servers); - }); - } - - private boolean supportsPaging() { - return this.graph.graph().backendStoreFeatures().supportsQueryByPage(); - } } 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 1438249a4a..269094bc7e 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 @@ -18,7 +18,6 @@ package org.apache.hugegraph.task; import java.util.ArrayList; -import java.util.Collection; import java.util.Iterator; import java.util.List; import java.util.Map; @@ -125,11 +124,9 @@ private TaskTransaction tx() { // NOTE: only the owner thread can access task tx if (this.taskTx == null) { /* - * NOTE: don't synchronized(this) due to scheduler thread hold - * this lock through scheduleTasks(), then query tasks and wait - * for db-worker thread after call(), the tx may not be initialized - * but can't catch this lock, then cause deadlock. - * We just use this.serverManager as a monitor here + * NOTE: don't synchronized(this) to avoid potential deadlock + * when multiple threads are accessing task transaction. + * We use this.serverManager as a monitor here for thread safety. */ synchronized (this.serverManager) { if (this.taskTx == null) { @@ -146,9 +143,10 @@ private TaskTransaction tx() { @Override public void restoreTasks() { - Id selfServer = this.serverManager().selfNodeId(); List> taskList = new ArrayList<>(); - // Restore 'RESTORING', 'RUNNING' and 'QUEUED' tasks in order. + // Single-node mode: restore pending tasks without server filtering. + // Don't restore legacy SCHEDULING/SCHEDULED tasks globally; they were + // owned by the removed master-worker scheduler. for (TaskStatus status : TaskStatus.PENDING_STATUSES) { String page = this.supportsPaging() ? PageInfo.PAGE_NONE : null; do { @@ -156,9 +154,7 @@ public void restoreTasks() { for (iter = this.findTask(status, PAGE_SIZE, page, false); iter.hasNext(); ) { HugeTask task = iter.next(); - if (selfServer.equals(task.server())) { - taskList.add(task); - } + taskList.add(task); } if (page != null) { page = PageInfo.pageInfo(iter); @@ -211,33 +207,15 @@ public Future schedule(HugeTask task) { return this.submitTask(task); } - // Check this is on master for normal task schedule - this.checkOnMasterNode("schedule"); - if (this.serverManager().onlySingleNode() && !task.computer()) { - /* - * Speed up for single node, submit the task immediately, - * this code can be removed without affecting code logic - */ - task.status(TaskStatus.QUEUED); - task.server(this.serverManager().selfNodeId()); - this.save(task); - return this.submitTask(task); - } else { - /* - * Just set the SCHEDULING status and save the task, - * it will be scheduled by periodic scheduler worker - */ - task.status(TaskStatus.SCHEDULING); - this.save(task); - - // Notify master server to schedule and execute immediately - TaskManager.instance().notifyNewTask(task); - - return task; - } + task.status(TaskStatus.QUEUED); + return this.submitTask(task, true); } private Future submitTask(HugeTask task) { + return this.submitTask(task, false); + } + + private Future submitTask(HugeTask task, boolean saveAfterQueued) { int size = this.tasks.size() + 1; E.checkArgument(size <= MAX_PENDING_TASKS, "Pending tasks size %s has exceeded the max limit %s", @@ -245,6 +223,14 @@ private Future submitTask(HugeTask task) { this.initTaskCallable(task); assert !this.tasks.containsKey(task.id()) : task; this.tasks.put(task.id(), task); + if (saveAfterQueued) { + try { + this.save(task); + } catch (RuntimeException e) { + this.tasks.remove(task.id()); + throw e; + } + } return this.taskExecutor.submit(task); } @@ -273,7 +259,6 @@ public void initTaskCallable(HugeTask task) { @Override public synchronized void cancel(HugeTask task) { E.checkArgumentNotNull(task, "Task can't be null"); - this.checkOnMasterNode("cancel"); if (task.completed() || task.cancelling()) { return; @@ -281,31 +266,18 @@ public synchronized void cancel(HugeTask task) { LOG.info("Cancel task '{}' in status {}", task.id(), task.status()); - if (task.server() == null) { - // The task not scheduled to workers, set canceled immediately - assert task.status().code() < TaskStatus.QUEUED.code(); - if (task.status(TaskStatus.CANCELLED)) { - this.save(task); - return; + HugeTask memTask = this.tasks.get(task.id()); + if (memTask != null) { + boolean cancelled = memTask.cancel(true); + if (cancelled) { + task.overwriteStatus(TaskStatus.CANCELLED); } - } else if (task.status(TaskStatus.CANCELLING)) { - // The task scheduled to workers, let the worker node to cancel + LOG.info("Task '{}' cancel result: {}", task.id(), cancelled); + return; + } + + if (task.status(TaskStatus.CANCELLED)) { this.save(task); - assert task.server() != null : task; - assert this.serverManager().selfIsMaster(); - if (!task.server().equals(this.serverManager().selfNodeId())) { - /* - * Remove the task from memory if it's running on worker node, - * but keep the task in memory if it's running on master node. - * Cancel-scheduling will read the task from backend store, if - * removed this instance from memory, there will be two task - * instances with the same id, and can't cancel the real task that - * is running but removed from memory. - */ - this.remove(task); - } - // Notify master server to schedule and execute immediately - TaskManager.instance().notifyNewTask(task); return; } @@ -317,132 +289,11 @@ public synchronized void cancel(HugeTask task) { public ServerInfoManager serverManager() { return this.serverManager; } - - protected synchronized void scheduleTasksOnMaster() { - // Master server schedule all scheduling tasks to suitable worker nodes - Collection serverInfos = this.serverManager().allServerInfos(); - String page = this.supportsPaging() ? PageInfo.PAGE_NONE : null; - do { - Iterator> tasks = this.tasks(TaskStatus.SCHEDULING, PAGE_SIZE, page, - false); - while (tasks.hasNext()) { - HugeTask task = tasks.next(); - if (task.server() != null) { - // Skip if already scheduled - continue; - } - - if (!this.serverManager.selfIsMaster()) { - return; - } - - HugeServerInfo server = this.serverManager().pickWorkerNode(serverInfos, task); - if (server == null) { - LOG.info("The master can't find suitable servers to " + - "execute task '{}', wait for next schedule", task.id()); - continue; - } - - // Found suitable server, update task status - assert server.id() != null; - task.server(server.id()); - task.status(TaskStatus.SCHEDULED); - this.save(task); - - // Update server load in memory, it will be saved at the ending - server.increaseLoad(task.load()); - - LOG.info("Scheduled task '{}' to server '{}'", task.id(), server.id()); - } - if (page != null) { - page = PageInfo.pageInfo(tasks); - } - } while (page != null); - - // Save to store - this.serverManager().updateServerInfos(serverInfos); - } - - protected void executeTasksOnWorker(Id server) { - String page = this.supportsPaging() ? PageInfo.PAGE_NONE : null; - do { - Iterator> tasks = this.tasks(TaskStatus.SCHEDULED, PAGE_SIZE, page, - false); - while (tasks.hasNext()) { - HugeTask task = tasks.next(); - this.initTaskCallable(task); - Id taskServer = task.server(); - if (taskServer == null) { - LOG.warn("Task '{}' may not be scheduled", task.id()); - continue; - } - HugeTask memTask = this.tasks.get(task.id()); - if (memTask != null) { - assert memTask.status().code() > task.status().code(); - continue; - } - if (taskServer.equals(server)) { - task.status(TaskStatus.QUEUED); - this.save(task); - this.submitTask(task); - } - } - if (page != null) { - page = PageInfo.pageInfo(tasks); - } - } while (page != null); - } - - protected void cancelTasksOnWorker(Id server) { - String page = this.supportsPaging() ? PageInfo.PAGE_NONE : null; - do { - Iterator> tasks = this.tasks(TaskStatus.CANCELLING, PAGE_SIZE, page, - false); - while (tasks.hasNext()) { - HugeTask task = tasks.next(); - Id taskServer = task.server(); - if (taskServer == null) { - LOG.warn("Task '{}' may not be scheduled", task.id()); - continue; - } - if (!taskServer.equals(server)) { - continue; - } - /* - * Task may be loaded from backend store and not initialized. - * like: A task is completed but failed to save in the last - * step, resulting in the status of the task not being - * updated to storage, the task is not in memory, so it's not - * initialized when canceled. - */ - HugeTask memTask = this.tasks.get(task.id()); - if (memTask != null) { - task = memTask; - } else { - this.initTaskCallable(task); - } - boolean cancelled = task.cancel(true); - LOG.info("Server '{}' cancel task '{}' with cancelled={}", - server, task.id(), cancelled); - } - if (page != null) { - page = PageInfo.pageInfo(tasks); - } - } while (page != null); - } - @Override public void taskDone(HugeTask task) { this.remove(task); - - Id selfServerId = this.serverManager().selfNodeId(); - try { - this.serverManager().decreaseLoad(task.load()); - } catch (Throwable e) { - LOG.error("Failed to decrease load for task '{}' on server '{}'", - task.id(), selfServerId, e); - } - LOG.debug("Task '{}' done on server '{}'", task.id(), selfServerId); + // Single-node mode: no need to manage load + LOG.debug("Task '{}' done", task.id()); } protected void remove(HugeTask task) { @@ -608,6 +459,7 @@ public Iterator> findTask(TaskStatus status, public HugeTask delete(Id id, boolean force) { this.checkOnMasterNode("delete"); + HugeTask running = this.tasks.get(id); HugeTask task = this.task(id, false); /* * The following is out of date when the task running on worker node: @@ -619,7 +471,17 @@ public HugeTask delete(Id id, boolean force) { * in fact, it is also possible to appear on the backend tasks * when the database status is inconsistent. */ - if (task != null) { + if (running != null) { + E.checkArgument(force || running.completed(), + "Can't delete incomplete task '%s' in status %s" + + ", Please try to cancel the task first", + id, running.status()); + if (force && !running.completed()) { + running.overwriteStatus(TaskStatus.DELETING); + running.cancel(true); + } + this.remove(running, force); + } else if (task != null) { E.checkArgument(force || task.completed(), "Can't delete incomplete task '%s' in status %s" + ", Please try to cancel the task first", @@ -801,10 +663,9 @@ public V call(Callable callable) { } } + @Deprecated private void checkOnMasterNode(String op) { - if (!this.serverManager().selfIsMaster()) { - throw new HugeException("Can't %s task on non-master server", op); - } + // Single-node mode: all operations are allowed, no role check needed } private boolean supportsPaging() { diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskAndResultScheduler.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskAndResultScheduler.java index 26a520bde3..70d4d2f571 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskAndResultScheduler.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskAndResultScheduler.java @@ -46,6 +46,7 @@ * Base class of task & result scheduler */ public abstract class TaskAndResultScheduler implements TaskScheduler { + /** * Which graph the scheduler belongs to */ @@ -61,8 +62,8 @@ public abstract class TaskAndResultScheduler implements TaskScheduler { private final ServerInfoManager serverManager; public TaskAndResultScheduler( - HugeGraphParams graph, - ExecutorService serverInfoDbExecutor) { + HugeGraphParams graph, + ExecutorService serverInfoDbExecutor) { E.checkNotNull(graph, "graph"); this.graph = graph; @@ -90,7 +91,7 @@ public void save(HugeTask task) { // Save result outcome if (rawResult != null) { HugeTaskResult result = - new HugeTaskResult(HugeTaskResult.genId(task.id())); + new HugeTaskResult(HugeTaskResult.genId(task.id())); result.result(rawResult); this.call(() -> { @@ -193,7 +194,7 @@ protected Iterator> queryTask(Map conditions, } Iterator vertices = this.tx().queryTaskInfos(query); Iterator> tasks = - new MapperIterator<>(vertices, HugeTask::fromVertex); + new MapperIterator<>(vertices, HugeTask::fromVertex); // Convert iterator to list to avoid across thread tx accessed return QueryResults.toList(tasks); }); @@ -209,16 +210,16 @@ protected Iterator> queryTask(Map conditions, protected Iterator> queryTask(List ids) { ListIterator> ts = this.call( - () -> { - Object[] idArray = ids.toArray(new Id[ids.size()]); - Iterator vertices = this.tx() - .queryTaskInfos(idArray); - Iterator> tasks = - new MapperIterator<>(vertices, - HugeTask::fromVertex); - // Convert iterator to list to avoid across thread tx accessed - return QueryResults.toList(tasks); - }); + () -> { + Object[] idArray = ids.toArray(new Id[ids.size()]); + Iterator vertices = this.tx() + .queryTaskInfos(idArray); + Iterator> tasks = + new MapperIterator<>(vertices, + HugeTask::fromVertex); + // Convert iterator to list to avoid across thread tx accessed + return QueryResults.toList(tasks); + }); Iterator results = queryTaskResult(ids); @@ -230,7 +231,7 @@ protected Iterator> queryTask(List ids) { return new MapperIterator<>(ts, (task) -> { HugeTaskResult taskResult = - resultCaches.get(HugeTaskResult.genId(task.id())); + resultCaches.get(HugeTaskResult.genId(task.id())); if (taskResult != null) { task.result(taskResult); } @@ -248,6 +249,10 @@ protected HugeTask taskWithoutResult(Id id) { return HugeTask.fromVertex(vertex, false); }); + if (result == null) { + throw new NotFoundException("Can't find task with id '%s'", id); + } + return result; } @@ -281,7 +286,7 @@ protected Iterator> queryTaskWithoutResult(String key, } protected Iterator> queryTaskWithoutResult(Map conditions, long limit, String page) { + Object> conditions, long limit, String page) { return this.call(() -> { ConditionQuery query = new ConditionQuery(HugeType.TASK); if (page != null) { @@ -319,7 +324,7 @@ protected void deleteTaskResultFromTx(Id taskId) { protected HugeTaskResult queryTaskResult(Id taskid) { HugeTaskResult result = this.call(() -> { Iterator vertices = - this.tx().queryTaskInfos(HugeTaskResult.genId(taskid)); + this.tx().queryTaskInfos(HugeTaskResult.genId(taskid)); Vertex vertex = QueryResults.one(vertices); if (vertex == null) { return null; @@ -334,12 +339,12 @@ protected HugeTaskResult queryTaskResult(Id taskid) { protected Iterator queryTaskResult(List taskIds) { return this.call(() -> { Object[] idArray = - taskIds.stream().map(HugeTaskResult::genId).toArray(); + taskIds.stream().map(HugeTaskResult::genId).toArray(); Iterator vertices = this.tx() .queryTaskInfos(idArray); Iterator tasks = - new MapperIterator<>(vertices, - HugeTaskResult::fromVertex); + new MapperIterator<>(vertices, + HugeTaskResult::fromVertex); // Convert iterator to list to avoid across thread tx accessed return QueryResults.toList(tasks); }); 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 277822a386..ec063754d8 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 @@ -18,7 +18,6 @@ package org.apache.hugegraph.task; import java.util.Map; -import java.util.Queue; import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; @@ -29,11 +28,9 @@ import org.apache.hugegraph.HugeException; import org.apache.hugegraph.HugeGraphParams; import org.apache.hugegraph.concurrent.PausableScheduledThreadPool; -import org.apache.hugegraph.type.define.NodeRole; import org.apache.hugegraph.util.Consumers; import org.apache.hugegraph.util.E; import org.apache.hugegraph.util.ExecutorUtil; -import org.apache.hugegraph.util.LockUtil; import org.apache.hugegraph.util.Log; import org.slf4j.Logger; @@ -76,8 +73,6 @@ public final class TaskManager { private final ExecutorService ephemeralTaskExecutor; private final PausableScheduledThreadPool distributedSchedulerExecutor; - private boolean enableRoleElected = false; - public static TaskManager instance() { return MANAGER; } @@ -102,11 +97,6 @@ private TaskManager(int pool) { // For a schedule task to run, just one thread is ok this.schedulerExecutor = ExecutorUtil.newPausableScheduledThreadPool( 1, TASK_SCHEDULER); - // Start after 10x period time waiting for HugeGraphServer startup - this.schedulerExecutor.scheduleWithFixedDelay(this::scheduleOrExecuteJob, - 10 * SCHEDULE_PERIOD, - SCHEDULE_PERIOD, - TimeUnit.MILLISECONDS); } public void addScheduler(HugeGraphParams graph) { @@ -230,14 +220,6 @@ private void closeDistributedSchedulerTx(HugeGraphParams graph) { } } - public void pauseScheduledThreadPool() { - this.schedulerExecutor.pauseSchedule(); - } - - public void resumeScheduledThreadPool() { - this.schedulerExecutor.resumeSchedule(); - } - public TaskScheduler getScheduler(HugeGraphParams graph) { return this.schedulers.get(graph); } @@ -349,125 +331,12 @@ public int pendingTasks() { return size; } - public void enableRoleElection() { - this.enableRoleElected = true; - } - public void onAsRoleMaster() { - try { - for (TaskScheduler entry : this.schedulers.values()) { - ServerInfoManager serverInfoManager = entry.serverManager(); - if (serverInfoManager != null) { - serverInfoManager.changeServerRole(NodeRole.MASTER); - } else { - LOG.warn("ServerInfoManager is null for graph {}", entry.spaceGraphName()); - } - } - } catch (Throwable e) { - LOG.error("Exception occurred when change to master role", e); - throw e; - } + // ServerInfo based role propagation is deprecated. } public void onAsRoleWorker() { - try { - for (TaskScheduler entry : this.schedulers.values()) { - ServerInfoManager serverInfoManager = entry.serverManager(); - if (serverInfoManager != null) { - serverInfoManager.changeServerRole(NodeRole.WORKER); - } else { - LOG.warn("ServerInfoManager is null for graph {}", entry.spaceGraphName()); - } - } - } catch (Throwable e) { - LOG.error("Exception occurred when change to worker role", e); - throw e; - } - } - - void notifyNewTask(HugeTask task) { - Queue queue = 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() { - // Called by scheduler timer - try { - for (TaskScheduler entry : this.schedulers.values()) { - // Maybe other threads close&remove scheduler at the same time - synchronized (entry) { - this.scheduleOrExecuteJobForGraph(entry); - } - } - } catch (Throwable e) { - LOG.error("Exception occurred when schedule job", e); - } - } - - private void scheduleOrExecuteJobForGraph(TaskScheduler scheduler) { - E.checkNotNull(scheduler, "scheduler"); - - if (scheduler instanceof StandardTaskScheduler) { - StandardTaskScheduler standardTaskScheduler = (StandardTaskScheduler) (scheduler); - ServerInfoManager serverManager = scheduler.serverManager(); - String spaceGraphName = scheduler.spaceGraphName(); - - LockUtil.lock(spaceGraphName, LockUtil.GRAPH_LOCK); - try { - /* - * Skip if: - * graph is closed (iterate schedulers before graph is closing) - * or - * graph is not initialized(maybe truncated or cleared). - * - * If graph is closing by other thread, current thread get - * serverManager and try lock graph, at the same time other - * thread deleted the lock-group, current thread would get - * exception 'LockGroup xx does not exists'. - * If graph is closed, don't call serverManager.initialized() - * due to it will reopen graph tx. - */ - if (!serverManager.graphIsReady()) { - return; - } - - // Update server heartbeat - serverManager.heartbeat(); - - /* - * 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.selfIsMaster()) { - standardTaskScheduler.scheduleTasksOnMaster(); - if (!this.enableRoleElected && !serverManager.onlySingleNode()) { - // assigned-Master + non-single-node don't need to execute tasks - return; - } - } - - // Execute queued tasks scheduled to current server - standardTaskScheduler.executeTasksOnWorker(serverManager.selfNodeId()); - - // Cancel tasks scheduled to current server - standardTaskScheduler.cancelTasksOnWorker(serverManager.selfNodeId()); - } finally { - LockUtil.unlock(spaceGraphName, LockUtil.GRAPH_LOCK); - } - } + // ServerInfo based role propagation is deprecated. } private static final ThreadLocal CONTEXTS = new ThreadLocal<>(); diff --git a/hugegraph-server/hugegraph-dist/src/assembly/static/conf/graphs/hstore.properties.template b/hugegraph-server/hugegraph-dist/src/assembly/static/conf/graphs/hstore.properties.template index f627b62029..4c78cad662 100644 --- a/hugegraph-server/hugegraph-dist/src/assembly/static/conf/graphs/hstore.properties.template +++ b/hugegraph-server/hugegraph-dist/src/assembly/static/conf/graphs/hstore.properties.template @@ -33,7 +33,6 @@ store=hugegraph pd.peers=127.0.0.1:8686 # task config -task.scheduler_type=local task.schedule_period=10 task.retry=0 task.wait_timeout=10 diff --git a/hugegraph-server/hugegraph-dist/src/assembly/static/conf/graphs/hugegraph.properties b/hugegraph-server/hugegraph-dist/src/assembly/static/conf/graphs/hugegraph.properties index bc15c39b39..26adfe0183 100644 --- a/hugegraph-server/hugegraph-dist/src/assembly/static/conf/graphs/hugegraph.properties +++ b/hugegraph-server/hugegraph-dist/src/assembly/static/conf/graphs/hugegraph.properties @@ -32,7 +32,6 @@ store=hugegraph #pd.peers=127.0.0.1:8686 # task config -task.scheduler_type=local task.schedule_period=10 task.retry=0 task.wait_timeout=10 diff --git a/hugegraph-server/hugegraph-dist/src/assembly/static/conf/rest-server.properties b/hugegraph-server/hugegraph-dist/src/assembly/static/conf/rest-server.properties index bbb956aafa..33ff0effcb 100644 --- a/hugegraph-server/hugegraph-dist/src/assembly/static/conf/rest-server.properties +++ b/hugegraph-server/hugegraph-dist/src/assembly/static/conf/rest-server.properties @@ -23,9 +23,6 @@ arthas.disabled_commands=jad #auth.admin_pa=pa #auth.graph_store=hugegraph -# lightweight load balancing (TODO: legacy mode, remove soon) -server.id=server-1 -server.role=master # use pd # usePD=true 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 4fae0f76c6..2c9c11021a 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 @@ -248,7 +248,7 @@ public void testCreateGraphsWithInvalidNames() { @Test public void testCreateGraphsWithSameName() { - List graphs = openGraphs("g", "g", "G"); + List graphs = openGraphs("gg", "gg", "GG"); HugeGraph g1 = graphs.get(0); HugeGraph g2 = graphs.get(1); HugeGraph g3 = graphs.get(2); @@ -320,6 +320,14 @@ public void testCreateGraphsWithDifferentNameDifferentBackends() { destroyGraphs(ImmutableList.of(g1, g2, graph)); } + @Test + public void testOpenGraphWithDeprecatedTaskSchedulerType() { + HugeGraph graph = openGraphWithBackend("legacySchedulerType", + "rocksdb", "binary", + "task.scheduler_type", "local"); + destroyGraphs(ImmutableList.of(graph)); + } + @Test public void testCreateGraphsWithMultiDisksForRocksDB() { HugeGraph g1 = openGraphWithBackend("g1", "rocksdb", "binary", "rocksdb.data_disks", diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/TaskCoreTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/TaskCoreTest.java index 2d106fc3b2..9e82692689 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/TaskCoreTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/TaskCoreTest.java @@ -17,9 +17,15 @@ package org.apache.hugegraph.core; -import java.util.Arrays; +import java.util.ArrayList; import java.util.Iterator; +import java.util.List; +import java.util.Map; import java.util.Random; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import org.apache.hugegraph.HugeException; @@ -33,6 +39,7 @@ import org.apache.hugegraph.job.GremlinJob; import org.apache.hugegraph.job.JobBuilder; import org.apache.hugegraph.task.HugeTask; +import org.apache.hugegraph.task.StandardTaskScheduler; import org.apache.hugegraph.task.TaskCallable; import org.apache.hugegraph.task.TaskScheduler; import org.apache.hugegraph.task.TaskStatus; @@ -53,11 +60,73 @@ public void setup() { HugeGraph graph = graph(); TaskScheduler scheduler = graph.taskScheduler(); + this.clearTasks(scheduler); + } + + private void clearTasks(TaskScheduler scheduler) { + for (int pass = 0; pass < 3; pass++) { + List> tasks = listTasks(scheduler); + if (tasks.isEmpty()) { + return; + } + + for (HugeTask task : tasks) { + if (!task.completed() && task.status() != TaskStatus.DELETING) { + scheduler.cancel(task); + } + } + + try { + scheduler.waitUntilAllTasksCompleted(10); + } catch (TimeoutException ignored) { + // Delete below will either remove completed tasks or fail loudly. + } + + for (HugeTask task : listTasks(scheduler)) { + deleteTaskAndWaitGone(scheduler, task.id()); + } + } + List> remaining = listTasks(scheduler); + if (!remaining.isEmpty()) { + Assert.fail(String.format("Failed to clean tasks: %s", remaining)); + } + } + + private static List> listTasks(TaskScheduler scheduler) { + List> tasks = new ArrayList<>(); Iterator> iter = scheduler.tasks(null, -1, null); while (iter.hasNext()) { - scheduler.delete(iter.next().id(), false); + tasks.add(iter.next()); + } + return tasks; + } + + private static void waitUntilTaskRunning(TaskScheduler scheduler) { + for (int pass = 0; pass < 1000; pass++) { + if (scheduler.pendingTasks() > 0) { + return; + } + sleepAWhile(10L); + } + Assert.fail("Timed out waiting for task to start running"); + } + + private static void deleteTaskAndWaitGone(TaskScheduler scheduler, Id id) { + for (int pass = 0; pass < 30; pass++) { + try { + scheduler.delete(id, true); + } catch (NotFoundException ignored) { + return; + } + try { + scheduler.task(id); + } catch (NotFoundException ignored) { + return; + } + sleepAWhile(100L); } + Assert.fail(String.format("Failed to delete task '%s'", id)); } @Test @@ -76,12 +145,15 @@ public void testTask() throws TimeoutException { Assert.assertEquals(id, task.id()); Assert.assertFalse(task.completed()); - Assert.assertThrows(IllegalArgumentException.class, () -> { - scheduler.delete(id, false); - }, e -> { - Assert.assertContains("Can't delete incomplete task '88888'", - e.getMessage()); - }); + if (scheduler instanceof StandardTaskScheduler) { + // StandardTaskScheduler: delete of incomplete task throws + Assert.assertThrows(IllegalArgumentException.class, () -> { + scheduler.delete(id, false); + }, e -> { + Assert.assertContains("Can't delete incomplete task '88888'", + e.getMessage()); + }); + } task = scheduler.waitUntilTaskCompleted(task.id(), 10); Assert.assertEquals(id, task.id()); @@ -89,7 +161,7 @@ public void testTask() throws TimeoutException { Assert.assertEquals(TaskStatus.SUCCESS, task.status()); Assert.assertEquals("test-task", scheduler.task(id).name()); - Assert.assertEquals("test-task", scheduler.tasks(Arrays.asList(id)) + Assert.assertEquals("test-task", scheduler.tasks(List.of(id)) .next().name()); Iterator> iter = scheduler.tasks(ImmutableList.of(id)); @@ -115,6 +187,249 @@ public void testTask() throws TimeoutException { }); } + @Test + public void testScheduleDoesNotPersistTaskWhenQueueFull() { + HugeGraph graph = graph(); + TaskScheduler scheduler = graph.taskScheduler(); + if (!(scheduler instanceof StandardTaskScheduler)) { + return; + } + + @SuppressWarnings("unchecked") + Map> tasks = Whitebox.getInternalState(scheduler, + "tasks"); + List placeholderIds = new ArrayList<>(); + Id id = IdGenerator.of(999996); + try { + for (int i = 0; tasks.size() < TaskScheduler.MAX_PENDING_TASKS; i++) { + Id placeholderId = IdGenerator.of(880000 + i); + tasks.put(placeholderId, new HugeTask<>(placeholderId, null, + new SleepCallable<>())); + placeholderIds.add(placeholderId); + } + + HugeTask task = new HugeTask<>(id, null, new SleepCallable<>()); + task.type("test"); + task.name("queue-full-task"); + + Assert.assertThrows(IllegalArgumentException.class, () -> { + scheduler.schedule(task); + }, e -> { + Assert.assertContains("Pending tasks size", e.getMessage()); + }); + Assert.assertThrows(NotFoundException.class, () -> { + scheduler.task(id); + }); + } finally { + for (Id placeholderId : placeholderIds) { + tasks.remove(placeholderId); + } + tasks.remove(id); + } + } + + @Test + public void testDeleteIncompleteTask() { + HugeGraph graph = graph(); + TaskScheduler scheduler = graph.taskScheduler(); + + Id id = IdGenerator.of(88891); + HugeTask task = new HugeTask<>(id, null, new SleepCallable<>()); + task.type("test"); + task.name("delete-incomplete-task"); + scheduler.schedule(task); + + try { + if (scheduler instanceof StandardTaskScheduler) { + Assert.assertThrows(IllegalArgumentException.class, () -> { + scheduler.delete(id, false); + }, e -> { + Assert.assertContains("Can't delete incomplete task '88891'", + e.getMessage()); + }); + } else { + waitUntilTaskRunning(scheduler); + HugeTask deleted = scheduler.delete(id, false); + Assert.assertNotNull(deleted); + Assert.assertEquals(TaskStatus.DELETING, deleted.status()); + Assert.assertEquals(TaskStatus.DELETING, scheduler.task(id).status()); + } + } finally { + try { + scheduler.waitUntilAllTasksCompleted(10); + } catch (TimeoutException ignored) { + // Force cleanup below handles non-interruptible test tasks. + } + deleteTaskAndWaitGone(scheduler, id); + } + } + + @Test + public void testForceDeleteRunningTaskDoesNotResurrectAfterDone() + throws Exception { + HugeGraph graph = graph(); + TaskScheduler scheduler = graph.taskScheduler(); + if (!(scheduler instanceof StandardTaskScheduler)) { + return; + } + + Id id = IdGenerator.of(88897); + BlockingCallable.reset(); + HugeTask task = new HugeTask<>(id, null, new BlockingCallable<>()); + task.type("test"); + task.name("force-delete-running-task"); + Future future = null; + + try { + future = scheduler.schedule(task); + waitUntilTaskRunning(scheduler); + Assert.assertTrue(BlockingCallable.awaitStarted()); + + HugeTask deleted = scheduler.delete(id, true); + Assert.assertNotNull(deleted); + Assert.assertThrows(NotFoundException.class, () -> { + scheduler.task(id); + }); + + BlockingCallable.release(); + try { + future.get(10L, TimeUnit.SECONDS); + } catch (CancellationException ignored) { + // Expected after deleting a locally running task. + } + + Assert.assertThrows(NotFoundException.class, () -> { + scheduler.task(id); + }); + } finally { + BlockingCallable.release(); + if (future != null && !future.isDone()) { + future.cancel(true); + } + deleteTaskAndWaitGone(scheduler, id); + } + } + + @Test + public void testDeleteRemoteLockedIncompleteTaskFailsFast() { + HugeGraph graph = graph(); + TaskScheduler scheduler = graph.taskScheduler(); + if (scheduler instanceof StandardTaskScheduler) { + return; + } + + Id id = IdGenerator.of(88892); + BlockingCallable.reset(); + HugeTask task = new HugeTask<>(id, null, new BlockingCallable<>()); + task.type("test"); + task.name("delete-remote-locked-incomplete-task"); + scheduler.schedule(task); + + Map> runningTasks = + Whitebox.getInternalState(scheduler, "runningTasks"); + HugeTask running = null; + try { + waitUntilTaskRunning(scheduler); + Assert.assertTrue(BlockingCallable.awaitStarted()); + running = runningTasks.remove(id); + Assert.assertNotNull(running); + + Assert.assertThrows(IllegalStateException.class, () -> { + scheduler.delete(id, false); + }, e -> { + Assert.assertContains("locked by another server", e.getMessage()); + }); + Assert.assertThrows(IllegalStateException.class, () -> { + scheduler.delete(id, true); + }, e -> { + Assert.assertContains("locked by another server", e.getMessage()); + }); + Assert.assertNotEquals(TaskStatus.DELETING, scheduler.task(id).status()); + } finally { + if (running != null && !running.completed()) { + runningTasks.put(id, running); + } + BlockingCallable.release(); + try { + scheduler.waitUntilAllTasksCompleted(10); + } catch (TimeoutException ignored) { + // Force cleanup below handles non-interruptible test tasks. + } + deleteTaskAndWaitGone(scheduler, id); + } + } + + @Test + public void testForceDeleteRunningDistributedTaskWaitsForRunnerExit() { + HugeGraph graph = graph(); + TaskScheduler scheduler = graph.taskScheduler(); + if (scheduler instanceof StandardTaskScheduler) { + return; + } + + Id id = IdGenerator.of(88895); + BlockingCallable.reset(); + HugeTask task = new HugeTask<>(id, null, new BlockingCallable<>()); + task.type("test"); + task.name("force-delete-running-distributed-task"); + scheduler.schedule(task); + + try { + waitUntilTaskRunning(scheduler); + Assert.assertTrue(BlockingCallable.awaitStarted()); + + HugeTask deleted = scheduler.delete(id, true); + Assert.assertNotNull(deleted); + Assert.assertEquals(TaskStatus.DELETING, deleted.status()); + Assert.assertEquals(TaskStatus.DELETING, scheduler.task(id).status()); + } finally { + BlockingCallable.release(); + try { + scheduler.waitUntilAllTasksCompleted(10); + } catch (TimeoutException ignored) { + // Force cleanup below handles non-interruptible test tasks. + } + deleteTaskAndWaitGone(scheduler, id); + } + } + + @Test + public void testLegacyTaskStatusesAreNotRestored() { + HugeGraph graph = graph(); + TaskScheduler scheduler = graph.taskScheduler(); + if (!(scheduler instanceof StandardTaskScheduler)) { + return; + } + + Id schedulingId = IdGenerator.of(88893); + Id scheduledId = IdGenerator.of(88894); + HugeTask scheduling = new HugeTask<>(schedulingId, null, + new SleepCallable<>()); + scheduling.type("test"); + scheduling.name("legacy-scheduling-task"); + scheduling.overwriteStatus(TaskStatus.SCHEDULING); + HugeTask scheduled = new HugeTask<>(scheduledId, null, + new SleepCallable<>()); + scheduled.type("test"); + scheduled.name("legacy-scheduled-task"); + scheduled.overwriteStatus(TaskStatus.SCHEDULED); + + try { + scheduler.save(scheduling); + scheduler.save(scheduled); + scheduler.restoreTasks(); + + Assert.assertEquals(0, scheduler.pendingTasks()); + Assert.assertEquals(TaskStatus.SCHEDULING, + scheduler.task(schedulingId).status()); + Assert.assertEquals(TaskStatus.SCHEDULED, + scheduler.task(scheduledId).status()); + } finally { + deleteTaskAndWaitGone(scheduler, schedulingId); + deleteTaskAndWaitGone(scheduler, scheduledId); + } + } + @Test public void testTaskWithoutResult() throws TimeoutException { HugeGraph graph = graph(); @@ -164,7 +479,7 @@ public void testTaskWithoutResult() throws TimeoutException { Assert.assertEquals("metadata-task", taskWithoutResult.name()); Assert.assertNull(taskWithoutResult.result()); - Iterator> iter = scheduler.tasks(ImmutableList.of(id)); + Iterator> iter = scheduler.tasks(ImmutableList.of(id), true); Assert.assertTrue(iter.hasNext()); taskWithResult = iter.next(); Assert.assertEquals("metadata-task", taskWithResult.name()); @@ -184,7 +499,7 @@ public void testTaskWithoutResult() throws TimeoutException { Assert.assertEquals("metadata-task", taskWithoutResult.name()); Assert.assertNull(taskWithoutResult.result()); - iter = scheduler.tasks(TaskStatus.SUCCESS, 10, null); + iter = scheduler.tasks(TaskStatus.SUCCESS, 10, null, true); Assert.assertTrue(iter.hasNext()); taskWithResult = iter.next(); Assert.assertEquals("metadata-task", taskWithResult.name()); @@ -278,13 +593,18 @@ public Object execute() throws Exception { Assert.assertEquals("test", task.type()); Assert.assertFalse(task.completed()); - HugeTask task2 = scheduler.waitUntilTaskCompleted(task.id(), 10); + // Ephemeral tasks are node-local and not persisted to DB. + // Use Future.get() to wait for completion instead of ID-based lookup. + try { + task.get(10, java.util.concurrent.TimeUnit.SECONDS); + } catch (Exception e) { + throw new RuntimeException("Ephemeral task execution failed", e); + } + Assert.assertEquals(TaskStatus.SUCCESS, task.status()); Assert.assertEquals("{\"k1\":13579,\"k2\":\"24680\"}", task.result()); - Assert.assertEquals(TaskStatus.SUCCESS, task2.status()); - Assert.assertEquals("{\"k1\":13579,\"k2\":\"24680\"}", task2.result()); - + // Ephemeral tasks are not stored in DB, so these should throw NotFoundException Assert.assertThrows(NotFoundException.class, () -> { scheduler.waitUntilTaskCompleted(task.id(), 10); }); @@ -639,7 +959,12 @@ public void testGremlinJobAndCancel() throws TimeoutException { scheduler.cancel(task); task = scheduler.task(task.id()); - Assert.assertEquals(TaskStatus.CANCELLING, task.status()); + // For DistributedTaskScheduler, local cancel may result in CANCELLED directly + // (task thread updates status after being interrupted) + // or CANCELLING (if task hasn't processed the interrupt yet) + Assert.assertTrue("Task status should be CANCELLING or CANCELLED, but was " + task.status(), + task.status() == TaskStatus.CANCELLING || + task.status() == TaskStatus.CANCELLED); task = scheduler.waitUntilTaskCompleted(task.id(), 10); Assert.assertEquals(TaskStatus.CANCELLED, task.status()); @@ -711,46 +1036,60 @@ public void testGremlinJobAndRestore() throws Exception { scheduler.cancel(task); task = scheduler.task(task.id()); - Assert.assertEquals(TaskStatus.CANCELLING, task.status()); + Assert.assertTrue("Task status should be CANCELLING or CANCELLED, but was " + task.status(), + task.status() == TaskStatus.CANCELLING || + task.status() == TaskStatus.CANCELLED); task = scheduler.waitUntilTaskCompleted(task.id(), 10); Assert.assertEquals(TaskStatus.CANCELLED, task.status()); Assert.assertTrue("progress=" + task.progress(), 0 < task.progress() && task.progress() < 10); Assert.assertEquals(0, task.retries()); - Assert.assertEquals(null, task.result()); + Assert.assertNull(task.result()); HugeTask finalTask = task; - Assert.assertThrows(IllegalArgumentException.class, () -> { - Whitebox.invoke(scheduler.getClass(), "restore", scheduler, - finalTask); - }, e -> { - Assert.assertContains("No need to restore completed task", - e.getMessage()); - }); - - HugeTask task2 = scheduler.task(task.id()); - Assert.assertThrows(IllegalArgumentException.class, () -> { - Whitebox.invoke(scheduler.getClass(), "restore", scheduler, task2); - }, e -> { - Assert.assertContains("No need to restore completed task", - e.getMessage()); - }); - - Whitebox.setInternalState(task2, "status", TaskStatus.RUNNING); - Whitebox.invoke(scheduler.getClass(), "restore", scheduler, task2); - Assert.assertThrows(IllegalArgumentException.class, () -> { + // because Distributed do nothing in restore + if (scheduler instanceof StandardTaskScheduler) { + Assert.assertThrows(IllegalArgumentException.class, () -> { + Whitebox.invoke(scheduler.getClass(), "restore", scheduler, + finalTask); + }, e -> { + Assert.assertContains("No need to restore completed task", + e.getMessage()); + }); + + HugeTask task2 = scheduler.task(task.id()); + Assert.assertThrows(IllegalArgumentException.class, () -> { + Whitebox.invoke(scheduler.getClass(), "restore", scheduler, task2); + }, e -> { + Assert.assertContains("No need to restore completed task", + e.getMessage()); + }); + + Whitebox.setInternalState(task2, "status", TaskStatus.RUNNING); Whitebox.invoke(scheduler.getClass(), "restore", scheduler, task2); - }, e -> { - Assert.assertContains("is already in the queue", e.getMessage()); - }); - scheduler.waitUntilTaskCompleted(task2.id(), 10); - sleepAWhile(500); - Assert.assertEquals(10, task2.progress()); - Assert.assertEquals(1, task2.retries()); - Assert.assertEquals("100", task2.result()); + Assert.assertThrows(IllegalArgumentException.class, () -> { + Whitebox.invoke(scheduler.getClass(), "restore", scheduler, task2); + }, e -> { + Assert.assertContains("is already in the queue", e.getMessage()); + }); + scheduler.waitUntilTaskCompleted(task2.id(), 10); + sleepAWhile(500); + Assert.assertEquals(10, task2.progress()); + Assert.assertEquals(1, task2.retries()); + Assert.assertEquals("100", task2.result()); + } else { + // DistributedTaskScheduler.restoreTasks() is a no-op by design — + // distributed recovery is handled by cronSchedule(), not + // restoreTasks(). Verify it does not throw and leaves task state + // untouched. + scheduler.restoreTasks(); + HugeTask recovered = scheduler.task(task.id()); + Assert.assertNotNull(recovered); + Assert.assertEquals(TaskStatus.CANCELLED, recovered.status()); + } } private HugeTask runGremlinJob(String gremlin) { @@ -779,6 +1118,46 @@ private static void sleepAWhile(long ms) { } } + public static class BlockingCallable extends TaskCallable { + + private static volatile CountDownLatch started = new CountDownLatch(1); + private static volatile CountDownLatch release = new CountDownLatch(1); + + public BlockingCallable() { + // pass + } + + public static void reset() { + started = new CountDownLatch(1); + release = new CountDownLatch(1); + } + + public static boolean awaitStarted() { + try { + return started.await(10L, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } + + public static void release() { + release.countDown(); + } + + @Override + public V call() throws Exception { + started.countDown(); + release.await(10L, TimeUnit.SECONDS); + return null; + } + + @Override + public void done() { + this.graph().taskScheduler().save(this.task()); + } + } + public static class SleepCallable extends TaskCallable { public SleepCallable() { diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java index 2bb10efaf4..f7b2729fe0 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java @@ -50,6 +50,7 @@ import org.apache.hugegraph.unit.core.SerialEnumTest; import org.apache.hugegraph.unit.core.ServerInfoManagerTest; import org.apache.hugegraph.unit.core.SystemSchemaStoreTest; +import org.apache.hugegraph.unit.core.TaskSchedulerServerInfoTest; import org.apache.hugegraph.unit.core.TraversalUtilTest; import org.apache.hugegraph.unit.id.EdgeIdTest; import org.apache.hugegraph.unit.id.IdTest; @@ -134,6 +135,7 @@ PageStateTest.class, SystemSchemaStoreTest.class, ServerInfoManagerTest.class, + TaskSchedulerServerInfoTest.class, RoleElectionStateMachineTest.class, HugeGraphAuthProxyTest.class, SchemaElementTest.class, diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/ServerInfoManagerTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/ServerInfoManagerTest.java index 85815bbc19..607d81de91 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/ServerInfoManagerTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/ServerInfoManagerTest.java @@ -50,6 +50,20 @@ public void setup() { this.hugegraphManager = new ServerInfoManager(hugegraphParams, executor); } + @Test + public void testInitDoesNotAccessBackendStore() { + HugeGraphParams graphParams = Mockito.mock(HugeGraphParams.class); + ExecutorService executor = Mockito.mock(ExecutorService.class); + ServerInfoManager manager = new ServerInfoManager(graphParams, executor); + + manager.init(); + + Mockito.verify(graphParams, Mockito.never()).systemTransaction(); + Mockito.verify(graphParams, Mockito.never()).backendStoreFeatures(); + Mockito.verify(graphParams, Mockito.never()).graph(); + Mockito.verify(graphParams, Mockito.never()).closeTx(); + } + @Test public void testSelfNodeIdScopedByGraphWithSameNodeId() { GlobalMasterInfo nodeInfo = GlobalMasterInfo.master("server-1"); @@ -73,4 +87,22 @@ public void testSelfNodeIdScopedByGraphWithSameNodeId() { public void testSelfNodeIdReturnsNullWhenNotInitialized() { Assert.assertNull(this.sysGraphManager.selfNodeId()); } + + @Test + public void testSelfNodeIdReturnsNullWhenNodeIdMissing() { + Whitebox.setInternalState(this.sysGraphManager, + "globalNodeInfo", new GlobalMasterInfo()); + + Assert.assertNull(this.sysGraphManager.selfNodeId()); + } + + @Test + public void testInitServerInfoDoesNotAccessBackendStore() { + GlobalMasterInfo nodeInfo = GlobalMasterInfo.master("server-1"); + + this.sysGraphManager.initServerInfo(nodeInfo); + + Assert.assertEquals("DEFAULT-~sys_graph/server-1", + this.sysGraphManager.selfNodeId().asString()); + } } diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/TaskSchedulerServerInfoTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/TaskSchedulerServerInfoTest.java new file mode 100644 index 0000000000..758b8c1f2d --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/TaskSchedulerServerInfoTest.java @@ -0,0 +1,219 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hugegraph.unit.core; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import org.apache.commons.configuration2.PropertiesConfiguration; +import org.apache.hugegraph.HugeGraph; +import org.apache.hugegraph.HugeGraphParams; +import org.apache.hugegraph.backend.id.Id; +import org.apache.hugegraph.backend.id.IdGenerator; +import org.apache.hugegraph.concurrent.PausableScheduledThreadPool; +import org.apache.hugegraph.config.CoreOptions; +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.config.ServerOptions; +import org.apache.hugegraph.core.GraphManager; +import org.apache.hugegraph.event.EventHub; +import org.apache.hugegraph.exception.NotFoundException; +import org.apache.hugegraph.task.DistributedTaskScheduler; +import org.apache.hugegraph.task.HugeTask; +import org.apache.hugegraph.task.TaskCallable; +import org.apache.hugegraph.task.TaskStatus; +import org.apache.hugegraph.testutil.Assert; +import org.apache.hugegraph.testutil.Whitebox; +import org.apache.hugegraph.util.ExecutorUtil; +import org.junit.Test; +import org.mockito.Mockito; + +public class TaskSchedulerServerInfoTest { + + @Test + public void testDistributedCheckRequirementDoesNotNeedServerInfo() { + HugeGraph graph = Mockito.mock(HugeGraph.class); + Mockito.when(graph.graphSpace()).thenReturn("DEFAULT"); + + HugeGraphParams params = Mockito.mock(HugeGraphParams.class); + Mockito.when(params.graph()).thenReturn(graph); + Mockito.when(params.name()).thenReturn("hugegraph"); + Mockito.when(params.spaceGraphName()).thenReturn("DEFAULT-hugegraph"); + Mockito.when(params.configuration()).thenReturn(newConfig()); + + PausableScheduledThreadPool schedulerExecutor = + ExecutorUtil.newPausableScheduledThreadPool( + 1, "distributed-scheduler-test-%d"); + ExecutorService taskDbExecutor = Executors.newSingleThreadExecutor(); + ExecutorService schemaTaskExecutor = Executors.newSingleThreadExecutor(); + ExecutorService olapTaskExecutor = Executors.newSingleThreadExecutor(); + ExecutorService gremlinTaskExecutor = Executors.newSingleThreadExecutor(); + ExecutorService ephemeralTaskExecutor = Executors.newSingleThreadExecutor(); + ExecutorService serverInfoDbExecutor = Executors.newSingleThreadExecutor(); + + try { + DistributedTaskScheduler scheduler = new DistributedTaskScheduler( + params, schedulerExecutor, taskDbExecutor, schemaTaskExecutor, + olapTaskExecutor, gremlinTaskExecutor, ephemeralTaskExecutor, + serverInfoDbExecutor); + scheduler.checkRequirement("schedule"); + } finally { + schedulerExecutor.shutdownNow(); + taskDbExecutor.shutdownNow(); + schemaTaskExecutor.shutdownNow(); + olapTaskExecutor.shutdownNow(); + gremlinTaskExecutor.shutdownNow(); + ephemeralTaskExecutor.shutdownNow(); + serverInfoDbExecutor.shutdownNow(); + } + } + + @Test + public void testDistributedCancelTreatsMissingRetryTaskAsGone() { + HugeGraph graph = Mockito.mock(HugeGraph.class); + Mockito.when(graph.graphSpace()).thenReturn("DEFAULT"); + + HugeGraphParams params = Mockito.mock(HugeGraphParams.class); + Mockito.when(params.graph()).thenReturn(graph); + Mockito.when(params.name()).thenReturn("hugegraph"); + Mockito.when(params.spaceGraphName()).thenReturn("DEFAULT-hugegraph"); + Mockito.when(params.configuration()).thenReturn(newConfig()); + + PausableScheduledThreadPool schedulerExecutor = + ExecutorUtil.newPausableScheduledThreadPool( + 1, "distributed-cancel-test-%d"); + ExecutorService taskDbExecutor = Executors.newSingleThreadExecutor(); + ExecutorService schemaTaskExecutor = Executors.newSingleThreadExecutor(); + ExecutorService olapTaskExecutor = Executors.newSingleThreadExecutor(); + ExecutorService gremlinTaskExecutor = Executors.newSingleThreadExecutor(); + ExecutorService ephemeralTaskExecutor = Executors.newSingleThreadExecutor(); + ExecutorService serverInfoDbExecutor = Executors.newSingleThreadExecutor(); + + try { + DistributedTaskScheduler scheduler = new DistributedTaskScheduler( + params, schedulerExecutor, taskDbExecutor, schemaTaskExecutor, + olapTaskExecutor, gremlinTaskExecutor, ephemeralTaskExecutor, + serverInfoDbExecutor) { + + @Override + protected boolean updateStatus(Id id, TaskStatus prestatus, + TaskStatus status) { + return false; + } + + @Override + protected HugeTask taskWithoutResult(Id id) { + throw new NotFoundException("Can't find task with id '%s'", id); + } + }; + Id id = IdGenerator.of(999997); + HugeTask task = new HugeTask<>(id, null, new TaskCallable() { + @Override + public Object call() { + return null; + } + }); + task.type("test"); + task.name("missing-retry-task"); + + scheduler.cancel(task); + } finally { + schedulerExecutor.shutdownNow(); + taskDbExecutor.shutdownNow(); + schemaTaskExecutor.shutdownNow(); + olapTaskExecutor.shutdownNow(); + gremlinTaskExecutor.shutdownNow(); + ephemeralTaskExecutor.shutdownNow(); + serverInfoDbExecutor.shutdownNow(); + } + } + + @Test + public void testGraphManagerDoesNotGenerateServerIdWhenElectionDisabled() { + HugeConfig config = newConfig(); + + GraphManager manager = new GraphManager(config, new EventHub("test")); + try { + Assert.assertEquals("", config.get(ServerOptions.SERVER_ID)); + Assert.assertNull(manager.globalNodeRoleInfo().nodeId()); + } finally { + manager.close(); + } + } + + @Test + public void testGraphManagerWarnsOnRoleElection() { + PropertiesConfiguration conf = new PropertiesConfiguration(); + conf.setProperty(ServerOptions.ENABLE_SERVER_ROLE_ELECTION.name(), true); + HugeConfig config = new HugeConfig(conf); + + GraphManager manager = new GraphManager(config, new EventHub("test")); + try { + Assert.assertNotNull(manager); + } finally { + manager.close(); + } + } + + @Test + public void testGraphManagerDoesNotInjectPdPeersForStandaloneRocksDB() { + HugeConfig serverConfig = newConfig(); + GraphManager manager = new GraphManager(serverConfig, new EventHub("test")); + HugeConfig graphConfig = newGraphConfig("rocksdb"); + + try { + Whitebox.invoke(manager.getClass(), "transferPdPeersConfig", + manager, graphConfig); + + Assert.assertFalse(graphConfig.containsKey(CoreOptions.PD_PEERS.name())); + } finally { + manager.close(); + } + } + + @Test + public void testGraphManagerInjectsPdPeersForHStoreGraph() { + HugeConfig serverConfig = newConfig(); + GraphManager manager = new GraphManager(serverConfig, new EventHub("test")); + HugeConfig graphConfig = newGraphConfig("hstore"); + HugeConfig mixedCaseGraphConfig = newGraphConfig("HStore"); + + try { + Whitebox.invoke(manager.getClass(), "transferPdPeersConfig", + manager, graphConfig); + Whitebox.invoke(manager.getClass(), "transferPdPeersConfig", + manager, mixedCaseGraphConfig); + + Assert.assertEquals(serverConfig.get(ServerOptions.PD_PEERS), + graphConfig.get(CoreOptions.PD_PEERS)); + Assert.assertEquals(serverConfig.get(ServerOptions.PD_PEERS), + mixedCaseGraphConfig.get(CoreOptions.PD_PEERS)); + } finally { + manager.close(); + } + } + + private static HugeConfig newConfig() { + return new HugeConfig(new PropertiesConfiguration()); + } + + private static HugeConfig newGraphConfig(String backend) { + PropertiesConfiguration conf = new PropertiesConfiguration(); + conf.setProperty(CoreOptions.BACKEND.name(), backend); + return new HugeConfig(conf); + } +} diff --git a/hugegraph-struct/src/main/java/org/apache/hugegraph/options/CoreOptions.java b/hugegraph-struct/src/main/java/org/apache/hugegraph/options/CoreOptions.java index 849539419b..caf0146bb9 100644 --- a/hugegraph-struct/src/main/java/org/apache/hugegraph/options/CoreOptions.java +++ b/hugegraph-struct/src/main/java/org/apache/hugegraph/options/CoreOptions.java @@ -295,13 +295,7 @@ public class CoreOptions extends OptionHolder { rangeInt(1, 500), 1 ); - public static final ConfigOption SCHEDULER_TYPE = - new ConfigOption<>( - "task.scheduler_type", - "The type of scheduler used in distribution system.", - allowValues("local", "distributed"), - "local" - ); + public static final ConfigOption TASK_SYNC_DELETION = new ConfigOption<>( "task.sync_deletion",