This is an automated email from the ASF dual-hosted git repository.
imbajin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hugegraph.git
The following commit(s) were added to refs/heads/master by this push:
new 2d6380420 refactor(server): disable legacy master-worker scheduler
logic (#3082)
2d6380420 is described below
commit 2d638042064b28dcbe6fe7cc39176c09d740fe5c
Author: Tsukilc <[email protected]>
AuthorDate: Wed Jul 8 13:03:56 2026 +0800
refactor(server): disable legacy master-worker scheduler logic (#3082)
This PR soft-disables the pre-PD master-worker task scheduling path while
keeping old configs and task data upgrade-safe. Scheduler selection is now
backend-driven: `hstore` uses `DistributedTaskScheduler`, and other backends
use the local `StandardTaskScheduler`.
---------
Co-authored-by: Himanshu Verma <[email protected]>
Co-authored-by: imbajin <[email protected]>
---
.../static/conf/hugegraph.properties.template | 1 -
.../org/apache/hugegraph/config/ServerOptions.java | 6 +-
.../org/apache/hugegraph/core/GraphManager.java | 89 ++--
.../org/apache/hugegraph/StandardHugeGraph.java | 33 +-
.../org/apache/hugegraph/config/CoreOptions.java | 8 +-
.../hugegraph/masterelection/GlobalMasterInfo.java | 2 +-
.../masterelection/StandardRoleListener.java | 1 -
.../hugegraph/task/DistributedTaskScheduler.java | 303 ++++++++++---
.../org/apache/hugegraph/task/HugeServerInfo.java | 8 -
.../java/org/apache/hugegraph/task/HugeTask.java | 8 +
.../apache/hugegraph/task/ServerInfoManager.java | 342 +--------------
.../hugegraph/task/StandardTaskScheduler.java | 233 +++-------
.../hugegraph/task/TaskAndResultScheduler.java | 45 +-
.../org/apache/hugegraph/task/TaskManager.java | 135 +-----
.../static/conf/graphs/hstore.properties.template | 1 -
.../static/conf/graphs/hugegraph.properties | 1 -
.../assembly/static/conf/rest-server.properties | 3 -
.../org/apache/hugegraph/core/MultiGraphsTest.java | 10 +-
.../org/apache/hugegraph/core/TaskCoreTest.java | 469 +++++++++++++++++++--
.../org/apache/hugegraph/unit/UnitTestSuite.java | 2 +
.../hugegraph/unit/core/ServerInfoManagerTest.java | 32 ++
.../unit/core/TaskSchedulerServerInfoTest.java | 219 ++++++++++
.../org/apache/hugegraph/options/CoreOptions.java | 8 +-
23 files changed, 1107 insertions(+), 852 deletions(-)
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 005031fe6..f97e36574 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 278542854..1aed0434d 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<String> 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<String> 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 770e75cc7..ff0435964 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.auth.StandardAuthenticator;
import org.apache.hugegraph.backend.BackendException;
import org.apache.hugegraph.backend.cache.Cache;
import org.apache.hugegraph.backend.cache.CacheManager;
-import org.apache.hugegraph.backend.id.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.ServerOptions;
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 final class GraphManager {
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 @@ public final class GraphManager {
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 @@ public final class GraphManager {
}
}
+ 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 @@ public final class GraphManager {
}
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 @@ public final class GraphManager {
}
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 final class GraphManager {
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<String, Map<String, Object>> 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<String, Map<String, Object>> 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<String, Object> 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<String, Object> 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 aaacb799c..6f66b4b8a 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 class StandardHugeGraph implements HugeGraph {
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 class StandardHugeGraph implements HugeGraph {
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 class StandardHugeGraph implements HugeGraph {
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 class StandardHugeGraph implements HugeGraph {
this.updateTime = updateTime;
}
+ @Override
public void waitStarted() {
// Just for trigger Tx.getOrNewTransaction, then load 3 stores
this.schemaTransaction();
@@ -489,9 +496,7 @@ public class StandardHugeGraph implements HugeGraph {
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 class StandardHugeGraph implements HugeGraph {
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 class StandardHugeGraph implements HugeGraph {
public void initSystemInfo() {
try {
this.taskScheduler().init();
- this.serverInfoManager().init();
this.authManager().init();
} finally {
this.closeTx();
@@ -1640,7 +1643,9 @@ public class StandardHugeGraph implements HugeGraph {
@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 07109580a..9aa60e398 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<String> SCHEDULER_TYPE =
- new ConfigOption<>(
- "task.scheduler_type",
- "The type of scheduler used in distribution system.",
- allowValues("local", "distributed"),
- "local"
- );
+
public static final ConfigOption<Boolean> 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 c345c50e6..485674445 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.backend.id.IdGenerator;
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 dbbea6d91..ce91ca2d5 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 3929f8414..8b11db9b6 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.apache.tinkerpop.gremlin.structure.Vertex;
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<Id, HugeTask<?>> runningTasks = new
ConcurrentHashMap<>();
+ private final Set<Id> deletingTasks = ConcurrentHashMap.newKeySet();
public DistributedTaskScheduler(HugeGraphParams graph,
ScheduledThreadPoolExecutor
schedulerExecutor,
@@ -118,6 +122,11 @@ public class DistributedTaskScheduler extends
TaskAndResultScheduler {
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 class DistributedTaskScheduler extends
TaskAndResultScheduler {
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 class DistributedTaskScheduler extends
TaskAndResultScheduler {
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 class DistributedTaskScheduler extends
TaskAndResultScheduler {
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 @@ public class DistributedTaskScheduler extends
TaskAndResultScheduler {
}
}
+ /**
+ * Note: This method will update the status of the input task.
+ *
+ * @param task
+ * @param <V>
+ */
@Override
public <V> void cancel(HugeTask<V> 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<Object> 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 class DistributedTaskScheduler extends
TaskAndResultScheduler {
protected <V> HugeTask<V> deleteFromDB(Id id) {
// Delete Task from DB, without checking task status
- return this.call(() -> {
- Iterator<Vertex> vertices = this.tx().queryTaskInfos(id);
- HugeVertex vertex = (HugeVertex) QueryResults.one(vertices);
- if (vertex == null) {
+ try {
+ return this.call(() -> {
+ Iterator<Vertex> vertices = this.tx().queryTaskInfos(id);
+ HugeVertex vertex = (HugeVertex) QueryResults.one(vertices);
+ if (vertex == null) {
+ this.deleteTaskResultFromTx(id);
+ return null;
+ }
+ HugeTask<V> 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 <V> void save(HugeTask<V> 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<V> 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<Vertex> 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<Vertex> 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 <V> HugeTask<V> 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<V> result = (HugeTask<V>) 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<V> result = (HugeTask<V>) 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 class DistributedTaskScheduler extends
TaskAndResultScheduler {
// 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 class DistributedTaskScheduler extends
TaskAndResultScheduler {
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 class DistributedTaskScheduler extends
TaskAndResultScheduler {
this.graph.closeTx();
});
}
- return true;
+
+ // TODO: serverInfoManager section should be removed in the future.
+ return this.serverManager().close();
}
@Override
@@ -447,9 +653,7 @@ public class DistributedTaskScheduler extends
TaskAndResultScheduler {
@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 class DistributedTaskScheduler extends
TaskAndResultScheduler {
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 71feb3f68..f0485f665 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 class HugeServerInfo {
return serverInfo;
}
- public <V> boolean suitableFor(HugeTask<V> 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 d5d703928..f63d4c076 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 class HugeTask<V> extends FutureTask<V> {
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 1560ec0b8..dac78f2fc 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 class ServerInfoManager {
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 class ServerInfoManager {
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<HugeServerInfo> servers = this.serverInfos(PAGE_SIZE,
page);
- while (servers.hasNext()) {
- existed = servers.next();
- E.checkArgument(!existed.role().master() ||
!existed.alive(),
- "Already existed master '%s' in current
cluster",
- existed.id());
- }
- 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 class ServerInfoManager {
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 class ServerInfoManager {
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<HugeServerInfo> 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 @@ public class ServerInfoManager {
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<HugeServerInfo> 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> V call(Callable<V> callable) {
assert !Thread.currentThread().getName().startsWith(
"server-info-db-worker") : "can't call by itself";
@@ -342,110 +126,4 @@ public class ServerInfoManager {
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<Vertex> 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<Vertex> 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<HugeServerInfo> serverInfos) {
- this.save(serverInfos);
- }
-
- protected Collection<HugeServerInfo> allServerInfos() {
- Iterator<HugeServerInfo> infos = this.serverInfos(NO_LIMIT, null);
- try (ListIterator<HugeServerInfo> iter = new ListIterator<>(
- MAX_SERVERS, infos)) {
- return iter.list();
- } catch (Exception e) {
- throw new HugeException("Failed to close server info iterator", e);
- }
- }
-
- protected Iterator<HugeServerInfo> serverInfos(String page) {
- return this.serverInfos(ImmutableMap.of(), PAGE_SIZE, page);
- }
-
- protected Iterator<HugeServerInfo> serverInfos(long limit, String page) {
- return this.serverInfos(ImmutableMap.of(), limit, page);
- }
-
- private Iterator<HugeServerInfo> serverInfos(Map<String, Object>
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<String, Object> 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<Vertex> vertices = this.tx().queryServerInfos(query);
- Iterator<HugeServerInfo> 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 1438249a4..269094bc7 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 @@ public class StandardTaskScheduler implements
TaskScheduler {
// 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 @@ public class StandardTaskScheduler implements
TaskScheduler {
@Override
public <V> void restoreTasks() {
- Id selfServer = this.serverManager().selfNodeId();
List<HugeTask<V>> 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 class StandardTaskScheduler implements TaskScheduler
{
for (iter = this.findTask(status, PAGE_SIZE, page, false);
iter.hasNext(); ) {
HugeTask<V> 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 class StandardTaskScheduler implements
TaskScheduler {
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 <V> Future<?> submitTask(HugeTask<V> task) {
+ return this.submitTask(task, false);
+ }
+
+ private <V> Future<?> submitTask(HugeTask<V> 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 @@ public class StandardTaskScheduler implements
TaskScheduler {
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 class StandardTaskScheduler implements TaskScheduler
{
@Override
public synchronized <V> void cancel(HugeTask<V> task) {
E.checkArgumentNotNull(task, "Task can't be null");
- this.checkOnMasterNode("cancel");
if (task.completed() || task.cancelling()) {
return;
@@ -281,31 +266,18 @@ public class StandardTaskScheduler implements
TaskScheduler {
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 class StandardTaskScheduler implements
TaskScheduler {
public ServerInfoManager serverManager() {
return this.serverManager;
}
-
- protected synchronized void scheduleTasksOnMaster() {
- // Master server schedule all scheduling tasks to suitable worker nodes
- Collection<HugeServerInfo> serverInfos =
this.serverManager().allServerInfos();
- String page = this.supportsPaging() ? PageInfo.PAGE_NONE : null;
- do {
- Iterator<HugeTask<Object>> 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<HugeTask<Object>> 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<HugeTask<Object>> 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 class StandardTaskScheduler implements TaskScheduler
{
public <V> HugeTask<V> 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 class StandardTaskScheduler implements
TaskScheduler {
* 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 class StandardTaskScheduler implements
TaskScheduler {
}
}
+ @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 26a520bde..70d4d2f57 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 @@ import com.google.common.collect.ImmutableMap;
* 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 abstract class TaskAndResultScheduler implements
TaskScheduler {
// 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 @@ public abstract class TaskAndResultScheduler implements
TaskScheduler {
}
Iterator<Vertex> vertices = this.tx().queryTaskInfos(query);
Iterator<HugeTask<V>> 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 @@ public abstract class TaskAndResultScheduler implements
TaskScheduler {
protected <V> Iterator<HugeTask<V>> queryTask(List<Id> ids) {
ListIterator<HugeTask<V>> ts = this.call(
- () -> {
- Object[] idArray = ids.toArray(new Id[ids.size()]);
- Iterator<Vertex> vertices = this.tx()
- .queryTaskInfos(idArray);
- Iterator<HugeTask<V>> 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<Vertex> vertices = this.tx()
+ .queryTaskInfos(idArray);
+ Iterator<HugeTask<V>> tasks =
+ new MapperIterator<>(vertices,
+ HugeTask::fromVertex);
+ // Convert iterator to list to avoid across thread tx
accessed
+ return QueryResults.toList(tasks);
+ });
Iterator<HugeTaskResult> results = queryTaskResult(ids);
@@ -230,7 +231,7 @@ public abstract class TaskAndResultScheduler implements
TaskScheduler {
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 @@ public abstract class TaskAndResultScheduler implements
TaskScheduler {
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 @@ public abstract class TaskAndResultScheduler implements
TaskScheduler {
}
protected <V> Iterator<HugeTask<V>> queryTaskWithoutResult(Map<String,
- Object> 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 @@ public abstract class TaskAndResultScheduler implements
TaskScheduler {
protected HugeTaskResult queryTaskResult(Id taskid) {
HugeTaskResult result = this.call(() -> {
Iterator<Vertex> 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 @@ public abstract class TaskAndResultScheduler implements
TaskScheduler {
protected Iterator<HugeTaskResult> queryTaskResult(List<Id> taskIds) {
return this.call(() -> {
Object[] idArray =
- taskIds.stream().map(HugeTaskResult::genId).toArray();
+ taskIds.stream().map(HugeTaskResult::genId).toArray();
Iterator<Vertex> vertices = this.tx()
.queryTaskInfos(idArray);
Iterator<HugeTaskResult> 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 277822a38..ec063754d 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 java.util.concurrent.TimeoutException;
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 @@ public final class TaskManager {
// 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 @@ public final class TaskManager {
}
}
- 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 final class TaskManager {
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<Runnable> 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<String> 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 f627b6202..4c78cad66 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 bc15c39b3..26adfe018 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 bbb956aaf..33ff0effc 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 4fae0f76c..2c9c11021 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 class MultiGraphsTest extends BaseCoreTest {
@Test
public void testCreateGraphsWithSameName() {
- List<HugeGraph> graphs = openGraphs("g", "g", "G");
+ List<HugeGraph> 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 class MultiGraphsTest extends BaseCoreTest {
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 2d106fc3b..9e8269268 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.EphemeralJobBuilder;
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 class TaskCoreTest extends BaseCoreTest {
HugeGraph graph = graph();
TaskScheduler scheduler = graph.taskScheduler();
+ this.clearTasks(scheduler);
+ }
+
+ private void clearTasks(TaskScheduler scheduler) {
+ for (int pass = 0; pass < 3; pass++) {
+ List<HugeTask<Object>> tasks = listTasks(scheduler);
+ if (tasks.isEmpty()) {
+ return;
+ }
+
+ for (HugeTask<Object> 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<Object> task : listTasks(scheduler)) {
+ deleteTaskAndWaitGone(scheduler, task.id());
+ }
+ }
+ List<HugeTask<Object>> remaining = listTasks(scheduler);
+ if (!remaining.isEmpty()) {
+ Assert.fail(String.format("Failed to clean tasks: %s", remaining));
+ }
+ }
+
+ private static List<HugeTask<Object>> listTasks(TaskScheduler scheduler) {
+ List<HugeTask<Object>> tasks = new ArrayList<>();
Iterator<HugeTask<Object>> 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 class TaskCoreTest extends BaseCoreTest {
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 class TaskCoreTest extends BaseCoreTest {
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<HugeTask<Object>> iter =
scheduler.tasks(ImmutableList.of(id));
@@ -115,6 +187,249 @@ public class TaskCoreTest extends BaseCoreTest {
});
}
+ @Test
+ public void testScheduleDoesNotPersistTaskWhenQueueFull() {
+ HugeGraph graph = graph();
+ TaskScheduler scheduler = graph.taskScheduler();
+ if (!(scheduler instanceof StandardTaskScheduler)) {
+ return;
+ }
+
+ @SuppressWarnings("unchecked")
+ Map<Id, HugeTask<?>> tasks = Whitebox.getInternalState(scheduler,
+ "tasks");
+ List<Id> 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<Id, HugeTask<?>> 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 class TaskCoreTest extends BaseCoreTest {
Assert.assertEquals("metadata-task", taskWithoutResult.name());
Assert.assertNull(taskWithoutResult.result());
- Iterator<HugeTask<Object>> iter =
scheduler.tasks(ImmutableList.of(id));
+ Iterator<HugeTask<Object>> 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 class TaskCoreTest extends BaseCoreTest {
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 class TaskCoreTest extends BaseCoreTest {
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 class TaskCoreTest extends BaseCoreTest {
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 class TaskCoreTest extends BaseCoreTest {
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<Object> 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<Object> 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<Object> 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<Object> runGremlinJob(String gremlin) {
@@ -779,6 +1118,46 @@ public class TaskCoreTest extends BaseCoreTest {
}
}
+ public static class BlockingCallable<V> extends TaskCallable<V> {
+
+ 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<V> extends TaskCallable<V> {
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 2bb10efaf..f7b2729fe 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.SecurityManagerTest;
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 @@ import org.junit.runners.Suite;
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 85815bbc1..607d81de9 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 class ServerInfoManagerTest {
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 class ServerInfoManagerTest {
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 000000000..758b8c1f2
--- /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 <V> HugeTask<V> 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<Object>() {
+ @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 849539419..caf0146bb 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<String> SCHEDULER_TYPE =
- new ConfigOption<>(
- "task.scheduler_type",
- "The type of scheduler used in distribution system.",
- allowValues("local", "distributed"),
- "local"
- );
+
public static final ConfigOption<Boolean> TASK_SYNC_DELETION =
new ConfigOption<>(
"task.sync_deletion",
