diff --git a/docs/server-runtime.md b/docs/server-runtime.md index d35b6b34ab..0bc0e98ed8 100644 --- a/docs/server-runtime.md +++ b/docs/server-runtime.md @@ -463,3 +463,23 @@ Structure or Process suites must execute tests; an all-skipped report is not compatibility evidence. Run the read-only application examples against representative existing data and validate the serializers, client versions and authentication combinations actually used by the application. + + +## Deleting a graph with local tasks + +Stop application writes before deletion and keep them stopped during retries. The local +scheduler closes task admission and waits for actual executions, cancellation callbacks +and their transaction cleanup before data or metadata is removed. If draining times out, +the graph stays registered with its data and configuration, but task admission remains +closed: new asynchronous tasks are rejected even after the existing owners exit. Keep +application writes stopped and retry deletion after those owners exit. If no destructive +cleanup has begun, restart the Server to resume task admission instead of completing +the deletion. Closing the scheduler alone leaves unstarted persistent work in storage. + +This is a local shutdown guarantee. For a graph shared by several Servers, stop writes +and tasks on every Server before deletion; notification delivery is not a cluster-wide +barrier. Do not recreate the same graph name until deletion finishes. Use a restart for +configuration changes rather than hot reload, whose recovery is handled separately. +Deletion is not transactional: a failure after destructive cleanup needs operator +inspection, and retry does not restore removed data. An owner-cleanup failure must not +be treated as a successful drain; inspect it instead of force-removing the scheduler. 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 1a8c1b5d1e..c7d821bf38 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 @@ -36,6 +36,7 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -114,6 +115,7 @@ import org.apache.hugegraph.space.register.dto.ServiceDTO; import org.apache.hugegraph.space.register.registerImpl.PdRegister; import org.apache.hugegraph.task.TaskManager; +import org.apache.hugegraph.task.TaskScheduler; import org.apache.hugegraph.testutil.Whitebox; import org.apache.hugegraph.traversal.optimize.HugeScriptTraversal; import org.apache.hugegraph.type.define.CollectionType; @@ -1107,23 +1109,21 @@ public void getSpaceStorage(String graphSpace) { } public void clearGraphSpace(String name) { - // Clear all roles - this.metaManager.clearGraphAuth(name); - - // Clear all schemaTemplate - this.metaManager.clearSchemaTemplate(name); - // Clear all graphs for (String key : this.graphs.keySet()) { - if (key.startsWith(name)) { + if (key.startsWith(name + DELIMITER)) { String[] parts = key.split(DELIMITER); this.dropGraph(parts[0], parts[1], true); } } + // Shared metadata remains available if a graph refuses deletion. + this.metaManager.clearGraphAuth(name); + this.metaManager.clearSchemaTemplate(name); + // Clear all services for (String key : this.services.keySet()) { - if (key.startsWith(name)) { + if (key.startsWith(name + DELIMITER)) { String[] parts = key.split(DELIMITER); this.dropService(parts[0], parts[1]); } @@ -1457,12 +1457,13 @@ private HugeGraph createGraphLocal(HugeConfig config, String name) { LOG.error("Failed to create graph '{}' due to: {}", name, e.getMessage(), e); if (graph != null) { - this.graphs.remove(graph.spaceGraphName(), graph); try { this.dropGraphLocal(graph); - } finally { - // The create event may have partially registered the graph + this.graphs.remove(graph.spaceGraphName(), graph); this.notifyEventLenient(Events.GRAPH_DROP, graph); + } catch (Throwable cleanup) { + e.addSuppressed(cleanup); + this.graphs.putIfAbsent(graph.spaceGraphName(), graph); } } throw e; @@ -1472,16 +1473,8 @@ private HugeGraph createGraphLocal(HugeConfig config, String name) { } private void dropGraphLocal(HugeGraph graph) { - try { - // Clear data and config files - graph.drop(); - } finally { - /* - * Will fill graph instance into HugeFactory.graphs after - * GraphFactory.open() succeed, remove it when the graph drops - */ - HugeFactory.remove(graph); - } + graph.drop(); + HugeFactory.remove(graph); } public HugeGraph createGraph(String graphSpace, String name, String creator, @@ -1614,16 +1607,14 @@ public HugeGraph createGraph(String graphSpace, String name, String creator, try { this.notifyEvent(Events.GRAPH_CREATE, graph); } catch (Throwable e) { - this.notifyEventLenient(Events.GRAPH_DROP, graph); - this.graphs.remove(graphName, graph); try { graph.close(); + this.notifyEventLenient(Events.GRAPH_DROP, graph); + this.graphs.remove(graphName, graph); + HugeFactory.remove(graph); } catch (Exception e1) { - if (graph instanceof StandardHugeGraph) { - ((StandardHugeGraph) graph).clearSchedulerAndLock(); - } + e.addSuppressed(e1); } - HugeFactory.remove(graph); throw e; } @@ -2086,12 +2077,11 @@ private HugeGraph createGraph(String graphSpace, HugeConfig config, } catch (BackendException e) { try { graph.close(); + HugeFactory.remove(graph); } catch (Exception e1) { - if (graph instanceof StandardHugeGraph) { - ((StandardHugeGraph) graph).clearSchedulerAndLock(); - } + e.addSuppressed(e1); + this.graphs.putIfAbsent(graph.spaceGraphName(), graph); } - HugeFactory.remove(graph); throw e; } } @@ -2269,6 +2259,21 @@ public void dropGraph(String graphSpace, String name, boolean clear) { String graphName = spaceGraphName(graphSpace, name); if (clear) { + // The auth proxy checks ADMIN before the local deletion fence. + TaskScheduler scheduler = g.taskScheduler(); + if (!scheduler.close()) { + try { + scheduler.waitUntilAllTasksCompleted(g.configuration().get(CoreOptions.TASK_WAIT_TIMEOUT)); + } catch (TimeoutException e) { + throw new HugeException("Tasks are still active in graph '%s'; task admission remains closed. " + + "Retry deletion after owners exit, or restart to resume task admission", + e, graphName); + } + E.checkState(scheduler.close(), + "Tasks are still active in graph '%s'; task admission remains closed. " + + "Retry deletion after owners exit, or restart to resume task admission", + graphName); + } this.removingGraphs.add(graphName); try { this.metaManager.removeGraphConfig(graphSpace, name); @@ -2278,24 +2283,13 @@ public void dropGraph(String graphSpace, String name, boolean clear) { "Failed to remove graph config of '%s'", name, e); } - /** - * close task scheduler before clear data, - * because taskinfo stored in backend in - * {@link org.apache.hugegraph.task.DistributedTaskScheduler} - */ - try { - g.taskScheduler().close(); - } catch (Throwable t) { - LOG.warn(String.format( - "Error when close TaskScheduler of %s", - graphName), - t); - } - g.clearBackend(); try { g.close(); } catch (Exception e) { + if (!g.closed()) { + throw new HugeException("Failed to close graph '%s', please retry later", e, graphName); + } LOG.warn("Failed to close graph", e); } } @@ -2309,13 +2303,17 @@ public void dropGraph(String graphSpace, String name, boolean clear) { } // Let gremlin server and rest server context remove graph LOG.info("Notify remove graph {} by GRAPH_DROP event", name); - Graph graph = this.graphs.remove(graphName); + Graph graph = this.graphs.get(graphName); if (graph != null) { try { graph.close(); } catch (Exception e) { + if (!((HugeGraph) graph).closed()) { + throw new HugeException("Failed to close graph '%s', please retry later", e, graphName); + } LOG.warn("Failed to close graph", e); } + this.graphs.remove(graphName, graph); try { // Delete alias graph in HugeFactory HugeFactory.remove((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 4649dd1fc9..157ec6b74b 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 @@ -537,6 +537,7 @@ public void initBackend() { @Override public void clearBackend() { + this.checkGraphNotClosed(); this.waitUntilAllTasksCompleted(); this.loadSchemaStore().open(this.configuration); @@ -1212,12 +1213,24 @@ public void create(String configPath, GlobalMasterInfo nodeInfo) { @Override public void drop() { + this.checkGraphNotClosed(); + TaskScheduler scheduler = this.taskManager.getScheduler(this.params); + if (scheduler != null && !scheduler.close()) { + // Fence admission before waiting for actual task owners to exit. + try { + scheduler.waitUntilAllTasksCompleted(this.configuration().get(CoreOptions.TASK_WAIT_TIMEOUT)); + } catch (TimeoutException e) { + throw new HugeException("Tasks are still active in graph '%s'; task admission remains closed. " + + "Retry deletion after owners exit, or restart to resume task admission", + e, this.spaceGraphName()); + } + E.checkState(scheduler.close(), + "Tasks are still active in graph '%s'; task admission remains closed. " + + "Retry deletion after owners exit, or restart to resume task admission", + this.spaceGraphName()); + } this.clearBackend(); - HugeConfig config = this.configuration(); - this.storeProvider.onDeleteConfig(config); - ConfigUtil.deleteFile(config.file()); - try { /* * It's hard to ensure all threads close the tx. @@ -1228,8 +1241,15 @@ public void drop() { */ this.close(); } catch (Throwable e) { + if (!this.closed()) { + throw new HugeException("Failed to close graph '%s', please retry later", + e, this.spaceGraphName()); + } LOG.warn("Failed to close graph {} {}", this, e); } + HugeConfig config = this.configuration(); + this.storeProvider.onDeleteConfig(config); + ConfigUtil.deleteFile(config.file()); } @Override 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 d89a32040f..0fe0e5b8f2 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 @@ -25,6 +25,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ScheduledThreadPoolExecutor; @@ -32,6 +33,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.hugegraph.exception.HugeException; import org.apache.hugegraph.HugeGraph; @@ -67,8 +69,11 @@ public class DistributedTaskScheduler extends TaskAndResultScheduler { */ private final AtomicBoolean closed = new AtomicBoolean(true); private volatile boolean closeCompleted; + private final AtomicInteger cancellations = new AtomicInteger(); private final ConcurrentHashMap> runningTasks = new ConcurrentHashMap<>(); + // Local execution owners, including accepted work that has not acquired its task lock. + private final ConcurrentHashMap executions = new ConcurrentHashMap<>(); private final Set deletingTasks = ConcurrentHashMap.newKeySet(); public DistributedTaskScheduler(HugeGraphParams graph, @@ -194,7 +199,8 @@ public void cronSchedule() { initTaskParams(cancelling); LOG.info("Try to cancel task({})@({}/{})", cancelling.id(), this.graphSpace, this.graphName); - if (!cancelling.cancel(true)) { + if (!this.cancelOwned(this.executions.get(cancellingId), + () -> cancelling.cancel(true), true)) { // Task already completed normally; force CANCELLED so // it doesn't stay stuck in CANCELLING forever. updateStatusWithLock(cancellingId, TaskStatus.CANCELLING, @@ -217,8 +223,10 @@ public void cronSchedule() { Id deletingId = deletings.next().id(); HugeTask deleting = runningTasks.get(deletingId); if (deleting != null) { - this.markTaskDeleting(deleting); - deleting.cancel(true); + this.cancelOwned(this.executions.get(deletingId), () -> { + this.markTaskDeleting(deleting); + return deleting.cancel(true); + }, true); } else { // Local has no task execution, but the current task has no nodes executing anymore. if (!isLockedTask(deletingId.toString())) { @@ -242,7 +250,7 @@ public HugeGraph graph() { @Override public int pendingTasks() { - return this.runningTasks.size(); + return this.executions.size() + this.cancellations.get(); } @Override @@ -251,14 +259,21 @@ public void restoreTasks() { } @Override - public Future schedule(HugeTask task) { + public synchronized Future schedule(HugeTask task) { E.checkArgumentNotNull(task, "Task can't be null"); + E.checkState(!this.closed.get(), "TaskScheduler has closed"); + PendingTask pending = this.executions.get(task.id()); + if (pending != null && pending.task == task && task.status() == TaskStatus.NEW && + pending.started.get() && !pending.ownerFinished) { + // HugeTask checks dependencies before running and may reschedule itself. + return pending; + } initTaskParams(task); if (task.ephemeralTask()) { // Handle ephemeral tasks, no scheduling needed, execute directly - return this.ephemeralTaskExecutor.submit(task); + return this.submitTask(task, this.ephemeralTaskExecutor, task); } // Validate task state before saving to ensure correct exception type @@ -305,6 +320,59 @@ protected void initTaskParams(HugeTask task) { @Override public void cancel(HugeTask task) { E.checkArgumentNotNull(task, "Task can't be null"); + PendingTask pending; + synchronized (this) { + E.checkState(!this.closed.get(), "TaskScheduler has closed"); + pending = this.executions.get(task.id()); + if (pending != null) { + synchronized (pending) { + pending.callbacks++; + } + } + this.cancellations.incrementAndGet(); + } + String context = TaskManager.getContext(); + AtomicBoolean cleaned = new AtomicBoolean(true); + TaskManager.ContextCallable owned = new TaskManager.ContextCallable<>(() -> { + try { + this.cancelTask(task, true); + return null; + } finally { + try { + this.graph.closeTx(); + } catch (RuntimeException | Error error) { + cleaned.set(false); + throw error; + } + } + }); + try { + // Keep the original synchronous callback semantics, including nested cancel requests. + owned.call(); + } catch (RuntimeException | Error error) { + throw error; + } catch (Exception e) { + throw new HugeException("Failed to cancel task '%s'", e, task.id()); + } finally { + TaskManager.setContext(context); + if (cleaned.get()) { + this.releaseCancellation(pending); + } + } + } + + private void releaseCancellation(PendingTask pending) { + if (pending != null) { + synchronized (pending) { + pending.callbacks--; + pending.release(); + } + } + this.cancellations.decrementAndGet(); + } + + private void cancelTask(HugeTask task, boolean direct) { + E.checkArgumentNotNull(task, "Task can't be null"); if (task.completed() || task.cancelling()) { return; @@ -314,12 +382,19 @@ public void cancel(HugeTask task) { // 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); - } + PendingTask pending = this.executions.get(task.id()); + if (runningTask != null || task.ephemeralTask()) { + HugeTask local = runningTask != null ? runningTask : task; + Callable cancellation = () -> { + boolean accepted = local.cancel(true); + if (accepted && !local.ephemeralTask()) { + task.overwriteStatus(TaskStatus.CANCELLED); + this.save(local); + } + return accepted; + }; + boolean cancelled = pending == null ? local.cancel(true) : + this.cancelOwned(pending, cancellation, direct); LOG.info("Cancel local running task '{}' result: {}", task.id(), cancelled); return; } @@ -475,8 +550,10 @@ public HugeTask delete(Id id, boolean force) { HugeTask running = this.runningTasks.get(id); if (running != null) { - this.markTaskDeleting(running); - running.cancel(true); + this.cancelOwned(this.executions.get(id), () -> { + this.markTaskDeleting(running); + return running.cancel(true); + }, false); @SuppressWarnings("unchecked") HugeTask result = (HugeTask) running; return result; @@ -532,8 +609,10 @@ public boolean close() { return true; } - // set closed - this.closed.set(true); + // Fence schedule() and cron admission before starting the drain. + synchronized (this) { + this.closed.set(true); + } // cancel cron thread if (!cronFuture.isDone() && !cronFuture.isCancelled()) { @@ -557,17 +636,22 @@ public boolean close() { LOG.warn("Exception while waiting for cron task to complete", e); return false; } catch (RejectedExecutionException e) { - if (!this.schedulerExecutor.isTerminated()) { - LOG.warn("Scheduler executor has not drained when closing scheduler", e); - return false; - } + LOG.warn("Scheduler executor was stopped before its graphs drained", e); + return false; } // cancel all running tasks - for (HugeTask task : this.runningTasks.values()) { + for (PendingTask pending : this.executions.values()) { + // Durable work that never ran stays NEW for a later scheduler. + if (!pending.task.ephemeralTask() && + !this.runningTasks.containsKey(pending.task.id())) { + pending.cancelBeforeStart(); + // A wrapper already entering its lock/read phase cleans up without running the job. + continue; + } LOG.info("cancel task({}) @({}/{}) when closing scheduler", - task.id(), graphSpace, graphName); - this.cancel(task); + pending.task.id(), graphSpace, graphName); + this.cancelTask(pending.task, false); } try { @@ -767,7 +851,10 @@ protected boolean updateStatusWithLock(Id id, TaskStatus prestatus, * @param task * @return true if the task have start */ - private boolean tryStartHugeTask(HugeTask task) { + private synchronized boolean tryStartHugeTask(HugeTask task) { + if (this.closed.get() || this.executions.containsKey(task.id())) { + return false; + } // Print Scheduler status logCurrentState(); @@ -795,7 +882,7 @@ private boolean tryStartHugeTask(HugeTask task) { ThreadPoolExecutor executor = (ThreadPoolExecutor) chosenExecutor; if (executor.getActiveCount() < executor.getMaximumPoolSize()) { TaskRunner runner = new TaskRunner<>(task); - chosenExecutor.submit(runner); + this.submitTask(task, chosenExecutor, runner); LOG.info("Submit task({})@({}/{})", task.id(), this.graphSpace, this.graphName); @@ -805,6 +892,176 @@ private boolean tryStartHugeTask(HugeTask task) { return false; } + private Future submitTask(HugeTask task, ExecutorService executor, Runnable work) { + TaskManager.ContextCallable owned = new TaskManager.ContextCallable<>(() -> { + try { + work.run(); + return null; + } finally { + this.closeOwner(this.executions.get(task.id())); + } + }); + PendingTask pending = new PendingTask(task, executor, owned); + E.checkState(this.executions.putIfAbsent(task.id(), pending) == null, + "Task '%s' is already pending", task.id()); + try { + executor.execute(pending); + return pending; + } catch (RuntimeException | Error error) { + pending.cancelBeforeStart(); + throw error; + } + } + + private boolean cancelOwned(PendingTask pending, Callable action, boolean direct) { + if (pending == null) { + return false; + } + synchronized (pending) { + if (this.executions.get(pending.task.id()) != pending) { + return false; + } + pending.callbacks++; + } + TaskManager.ContextCallable owned = new TaskManager.ContextCallable<>(() -> { + try { + return action.call(); + } finally { + this.closeOwner(pending); + } + }); + Callable cancellation = () -> { + try { + return owned.call(); + } finally { + synchronized (pending) { + pending.callbacks--; + if (pending.task.isDone()) { + pending.cancelBeforeStart(); + } + pending.release(); + } + } + }; + if (direct) { + try { + return cancellation.call(); + } catch (Exception e) { + throw new HugeException("Failed to cancel task '%s'", e, pending.task.id()); + } + } + Future future; + try { + future = this.schedulerExecutor.submit(cancellation); + } catch (RuntimeException | Error error) { + synchronized (pending) { + pending.callbacks--; + pending.release(); + } + throw error; + } + try { + return future.get(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new HugeException("Interrupted when cancelling task '%s'", e, pending.task.id()); + } catch (ExecutionException e) { + throw new HugeException("Failed to cancel task '%s'", e, pending.task.id()); + } + } + + private void closeOwner(PendingTask pending) { + try { + this.graph.closeTx(); + } catch (RuntimeException | Error error) { + synchronized (pending) { + // A failed owner cleanup must never become a successful graph drain. + pending.cleanupFailed = true; + } + throw error; + } + } + + private class PendingTask extends FutureTask { + + private final HugeTask task; + private final ExecutorService executor; + private final AtomicBoolean started = new AtomicBoolean(); + private volatile boolean ownerFinished; + private int callbacks; + private boolean cleanupFailed; + + PendingTask(HugeTask task, ExecutorService executor, Callable work) { + super(work); + this.task = task; + this.executor = executor; + } + + @Override + public void run() { + if (!this.started.compareAndSet(false, true)) { + return; + } + try { + if (this.task.ephemeralTask()) { + if (super.runAndReset() && this.task.completed()) { + this.set(null); + } + } else { + super.run(); + } + } finally { + synchronized (this) { + this.ownerFinished = true; + if (this.task.ephemeralTask() && !this.task.completed() && + !this.isDone() && !closed.get() && this.callbacks == 0) { + this.started.set(false); + this.ownerFinished = false; + try { + this.executor.execute(this); + } catch (RuntimeException | Error error) { + this.ownerFinished = true; + this.setException(error); + } + } + // A dependency retry fenced by close still needs its cancellation callback. + if (!this.task.ephemeralTask() || this.task.completed() || !closed.get()) { + this.release(); + } + } + } + } + + @Override + protected void done() { + if (this.isCancelled()) { + this.cancelBeforeStart(); + } + } + + private synchronized boolean cancelBeforeStart() { + if (!this.started.compareAndSet(false, true)) { + return false; + } + this.cancel(false); + if (this.executor instanceof ThreadPoolExecutor) { + ((ThreadPoolExecutor) this.executor).remove(this); + } + this.ownerFinished = true; + this.release(); + return true; + } + + private void release() { + if (this.ownerFinished && this.callbacks == 0 && !this.cleanupFailed) { + executions.remove(this.task.id(), this); + if (this.task.completed() && !this.isDone()) { + this.set(null); + } + } + } + } + protected void logCurrentState() { int gremlinActive = ((ThreadPoolExecutor) gremlinTaskExecutor).getActiveCount(); @@ -894,7 +1151,12 @@ public void run() { return; } - runningTasks.put(task.id(), task); + synchronized (DistributedTaskScheduler.this) { + if (closed.get()) { + return; + } + runningTasks.put(task.id(), task); + } // Task execution will not throw exceptions, HugeTask will catch exceptions // during execution and store them in the DB. @@ -902,7 +1164,7 @@ public void run() { } catch (Throwable t) { LOG.warn("exception when execute task", t); } finally { - runningTasks.remove(task.id()); + runningTasks.remove(task.id(), task); deletingTasks.remove(task.id()); unlockTask(task.id().asString(), lockResult); 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 15249067bc..8d91069553 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 @@ -26,7 +26,10 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import org.apache.hugegraph.exception.HugeException; import org.apache.hugegraph.HugeGraph; @@ -69,8 +72,11 @@ public class StandardTaskScheduler implements TaskScheduler { private final ExecutorService taskDbExecutor; private final Map> tasks; + private final Map> executions; private volatile TaskTransaction taskTx; + private boolean admissionClosed; + private int cancellations; public StandardTaskScheduler(HugeGraphParams graph, ExecutorService taskExecutor, @@ -85,6 +91,7 @@ public StandardTaskScheduler(HugeGraphParams graph, this.serverManager = new ServerInfoManager(graph); this.tasks = new ConcurrentHashMap<>(); + this.executions = new ConcurrentHashMap<>(); this.taskTx = null; } @@ -115,8 +122,8 @@ public String spaceGraphName() { } @Override - public int pendingTasks() { - return this.tasks.size(); + public synchronized int pendingTasks() { + return this.tasks.size() + this.cancellations; } private TaskTransaction tx() { @@ -141,7 +148,8 @@ private TaskTransaction tx() { } @Override - public void restoreTasks() { + public synchronized void restoreTasks() { + this.checkAdmission(); List> taskList = new ArrayList<>(); // Single-node mode: restore pending tasks without server filtering. // Don't restore legacy SCHEDULING/SCHEDULED tasks globally; they were @@ -184,7 +192,8 @@ private Future restore(HugeTask task) { } @Override - public Future schedule(HugeTask task) { + public synchronized Future schedule(HugeTask task) { + this.checkAdmission(); E.checkArgumentNotNull(task, "Task can't be null"); /* @@ -230,7 +239,7 @@ private Future submitTask(HugeTask task, boolean saveAfterQueued) { throw e; } } - return this.taskExecutor.submit(task); + return this.submitExecution(task); } private Future resubmitTask(HugeTask task) { @@ -240,7 +249,88 @@ private Future resubmitTask(HugeTask task) { E.checkArgument(this.tasks.containsKey(task.id()), "Can't resubmit task '%s' not been submitted before", task.id()); - return this.taskExecutor.submit(task); + return this.submitExecution(task); + } + + private Future submitExecution(HugeTask task) { + PendingTask pending = new PendingTask(task); + this.executions.computeIfAbsent(task.id(), id -> new ArrayList<>()).add(pending); + try { + this.taskExecutor.execute(pending); + } catch (RuntimeException | Error error) { + this.finishExecution(pending); + throw error; + } + return pending; + } + + private synchronized void finishExecution(PendingTask pending) { + List outstanding = this.executions.get(pending.task.id()); + if (outstanding != null && outstanding.remove(pending) && outstanding.isEmpty()) { + this.executions.remove(pending.task.id()); + this.tasks.remove(pending.task.id(), pending.task); + } + } + + private synchronized void cancelQueuedExecutions(Id id) { + List outstanding = this.executions.get(id); + if (outstanding != null) { + for (PendingTask pending : new ArrayList<>(outstanding)) { + pending.cancelBeforeStart(); + } + } + } + + private final class PendingTask extends FutureTask { + + private final HugeTask task; + private final AtomicBoolean started = new AtomicBoolean(); + + private PendingTask(HugeTask task) { + super(new ContextCallable<>(() -> { + task.run(); + return null; + })); + this.task = task; + } + + @Override + public void run() { + if (!this.started.compareAndSet(false, true)) { + return; + } + try { + super.run(); + } finally { + // A cleanup failure must retain this owner and prevent close ACK. + StandardTaskScheduler.this.graph.closeTx(); + StandardTaskScheduler.this.finishExecution(this); + } + } + + @Override + protected void done() { + if (this.isCancelled()) { + // Cancelling the dispatch Future must not run HugeTask callbacks + // on its caller or cancel a later dependency retry. + this.cancelBeforeStart(); + } + } + + private void cancelBeforeStart() { + if (this.started.compareAndSet(false, true)) { + super.cancel(false); + if (StandardTaskScheduler.this.taskExecutor instanceof ThreadPoolExecutor) { + ((ThreadPoolExecutor) StandardTaskScheduler.this.taskExecutor).remove(this); + } + StandardTaskScheduler.this.finishExecution(this); + } + } + } + + private void checkAdmission() { + E.checkState(!this.admissionClosed, "Task scheduler for graph '%s' is closing", + this.spaceGraphName()); } public void initTaskCallable(HugeTask task) { @@ -256,32 +346,45 @@ public void initTaskCallable(HugeTask task) { } @Override - public synchronized void cancel(HugeTask task) { + public void cancel(HugeTask task) { E.checkArgumentNotNull(task, "Task can't be null"); - - if (task.completed() || task.cancelling()) { - return; - } - - LOG.info("Cancel task '{}' in status {}", task.id(), task.status()); - - HugeTask memTask = this.tasks.get(task.id()); - if (memTask != null) { - boolean cancelled = memTask.cancel(true); - if (cancelled) { - task.overwriteStatus(TaskStatus.CANCELLED); + HugeTask memTask; + synchronized (this) { + if (task.completed() || task.cancelling()) { + return; } - LOG.info("Task '{}' cancel result: {}", task.id(), cancelled); - return; + memTask = this.tasks.get(task.id()); + if (memTask == null) { + this.checkAdmission(); + } + this.cancellations++; } - - if (task.status(TaskStatus.CANCELLED)) { - this.save(task); - return; + // HugeTask.cancel() runs done/cancelled callbacks on this caller. + // Do not hold the scheduler monitor while they wait for storage/workers. + try { + if (memTask != null) { + boolean cancelled = memTask.cancel(true); + if (cancelled) { + task.overwriteStatus(TaskStatus.CANCELLED); + } + return; + } + if (task.status(TaskStatus.CANCELLED)) { + this.save(task); + return; + } + throw new HugeException("Can't cancel task '%s' in status %s", + task.id(), task.status()); + } finally { + if (memTask != null && memTask.isCancelled()) { + this.cancelQueuedExecutions(task.id()); + } + // Keep the callback registered until its transaction really closes. + this.graph.closeTx(); + synchronized (this) { + this.cancellations--; + } } - - throw new HugeException("Can't cancel task '%s' in status %s", - task.id(), task.status()); } @Override @@ -290,7 +393,10 @@ public ServerInfoManager serverManager() { } @Override public void taskDone(HugeTask task) { - this.remove(task); + // Logical completion/cancellation can precede the worker's final cleanup. + if (!this.executions.containsKey(task.id())) { + this.remove(task); + } // Single-node mode: no need to manage load LOG.debug("Task '{}' done", task.id()); } @@ -301,6 +407,9 @@ protected void remove(HugeTask task) { protected void remove(HugeTask task, boolean force) { E.checkNotNull(task, "remove task"); + if (this.executions.containsKey(task.id())) { + return; + } HugeTask delTask = this.tasks.remove(task.id()); if (delTask != null && delTask != task) { LOG.warn("Task '{}' may be inconsistent status {}(expect {})", @@ -331,7 +440,8 @@ public void init() { } @Override - public boolean close() { + public synchronized boolean close() { + this.admissionClosed = true; // Running tasks still need the task DB transaction to persist done(). // Retain the scheduler and its owners until a later close attempt. if (this.pendingTasks() != 0) { 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 0342eb63bc..6d864d288d 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 @@ -127,28 +127,17 @@ public void closeScheduler(HugeGraphParams graph) { Throwable failure = null; boolean drained = scheduler == null; if (scheduler != null) { - /* - * Keep close+remove exclusive with scheduler iteration: in their gap - * a scheduler DB worker could otherwise reopen the graph transaction. - */ - synchronized (scheduler) { - boolean stopped = false; - try { - stopped = scheduler.close(); - } catch (RuntimeException | Error error) { - failure = error; - // Standard closes its server manager in finally; Distributed - // sets its dispatch-stop flag before cancellation/drain begins. - stopped = true; - } - // A timeout leaves running jobs registered for the next drain attempt. - if (stopped && scheduler.pendingTasks() == 0) { + try { + // Distributed close waits for cron without holding its monitor. + if (scheduler.close() && scheduler.pendingTasks() == 0) { this.schedulers.remove(graph, scheduler); drained = true; } + } catch (RuntimeException | Error error) { + failure = error; } } - if (!drained) { + if (!drained && (failure == null || scheduler.pendingTasks() != 0)) { // A running cron/job must finish before owner callbacks are queued // on its executor. In particular, do not rejoin a timed-out cron. if (failure instanceof Error) { @@ -159,6 +148,7 @@ public void closeScheduler(HugeGraphParams graph) { } return; } + // An owner-close failure retains the scheduler, but idle workers can still clean up. for (Runnable close : new Runnable[]{() -> { if (!this.taskExecutor.isTerminated()) { this.closeTaskTx(graph); @@ -261,7 +251,9 @@ public ServerInfoManager getServerInfoManager(HugeGraphParams graph) { } public void shutdown(long timeout) { - assert this.schedulers.isEmpty() : this.schedulers.size(); + if (!this.schedulers.isEmpty()) { + LOG.warn("Stopping task executors with {} schedulers still registered", this.schedulers.size()); + } Throwable ex = null; boolean terminated = this.distributedSchedulerExecutor.isTerminated(); diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/DistributedTaskAdmissionTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/DistributedTaskAdmissionTest.java new file mode 100644 index 0000000000..bddb0c3ec3 --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/DistributedTaskAdmissionTest.java @@ -0,0 +1,570 @@ +/* + * 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; + +import java.util.Iterator; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.dist.RegisterUtil; +import org.apache.hugegraph.id.Id; +import org.apache.hugegraph.id.IdGenerator; +import org.apache.hugegraph.meta.MetaManager; +import org.apache.hugegraph.meta.lock.LockResult; +import org.apache.hugegraph.meta.managers.TaskMetaManager; +import org.apache.hugegraph.job.EphemeralJob; +import org.apache.hugegraph.masterelection.GlobalMasterInfo; +import org.apache.hugegraph.task.DistributedTaskScheduler; +import org.apache.hugegraph.task.HugeTask; +import org.apache.hugegraph.task.TaskAndResultSchedulerTest.EmptyCallable; +import org.apache.hugegraph.task.TaskCallable; +import org.apache.hugegraph.task.TaskManager; +import org.apache.hugegraph.task.TaskScheduler; +import org.apache.hugegraph.task.TaskStatus; +import org.apache.hugegraph.testutil.Assert; +import org.apache.hugegraph.testutil.Whitebox; +import org.apache.hugegraph.unit.FakeObjects; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +public class DistributedTaskAdmissionTest { + + private TaskMetaManager originalLocks; + + @Before + public void prepareLocks() { + this.originalLocks = Whitebox.getInternalState(MetaManager.instance(), "taskMetaManager"); + TaskMetaManager locks = Mockito.mock(TaskMetaManager.class); + Mockito.when(locks.tryLockTask(Mockito.anyString(), Mockito.anyString(), Mockito.anyString())) + .thenAnswer(invocation -> { + LockResult acquired = new LockResult(); + acquired.lockSuccess(true); + return acquired; + }); + Whitebox.setInternalState(MetaManager.instance(), "taskMetaManager", locks); + } + + @After + public void restoreLocks() { + Whitebox.setInternalState(MetaManager.instance(), "taskMetaManager", this.originalLocks); + } + + @Test + public void testCloseDuringPersistentRegistrationPreservesNewTask() throws Exception { + try (Fixture fixture = new Fixture("register_new")) { + CountDownLatch reading = new CountDownLatch(1); + CountDownLatch finishRead = new CountDownLatch(1); + fixture.scheduler.beforeTaskRead = () -> { + reading.countDown(); + awaitUninterruptibly(finishRead); + }; + HugeTask task = task(false, new AtomicBoolean(), 9999917L); + ExecutorService requests = Executors.newSingleThreadExecutor(); + try { + fixture.scheduler.schedule(task); + Assert.assertTrue(reading.await(10L, TimeUnit.SECONDS)); + Future close = requests.submit(fixture.scheduler::close); + AtomicBoolean closed = Whitebox.getInternalState(fixture.scheduler, "closed"); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10L); + while (!closed.get() && System.nanoTime() < deadline) { + Thread.yield(); + } + Assert.assertTrue(closed.get()); + Assert.assertTrue(fixture.scheduler.pendingTasks() > 0); + Assert.assertFalse(close.isDone()); + finishRead.countDown(); + Assert.assertTrue(close.get(10L, TimeUnit.SECONDS)); + fixture.scheduler.beforeTaskRead = () -> { }; + Assert.assertEquals(TaskStatus.NEW, fixture.scheduler.task(task.id(), false).status()); + Assert.assertEquals(TaskStatus.NEW, task.status()); + fixture.assertWorkerClean(); + } finally { + finishRead.countDown(); + stop(requests); + } + } + } + + @Test + public void testAcceptedRemoteCancellationPreventsEarlyDrain() throws Exception { + try (Fixture fixture = new Fixture("cancel_gate")) { + CountDownLatch busy = new CountDownLatch(1); + CountDownLatch finishWorker = new CountDownLatch(1); + Future occupied = fixture.worker.submit(() -> { + busy.countDown(); + awaitUninterruptibly(finishWorker); + }); + ExecutorService requests = Executors.newFixedThreadPool(2); + GatedTask task = new GatedTask(9999918L, new EmptyCallable()); + try { + Assert.assertTrue(busy.await(10L, TimeUnit.SECONDS)); + fixture.scheduler.schedule(task); + Assert.assertEquals(0, fixture.scheduler.pendingTasks()); + task.armed.set(true); + Future cancel = requests.submit(() -> fixture.scheduler.cancel(task)); + Assert.assertTrue(task.reached.await(10L, TimeUnit.SECONDS)); + Future close = requests.submit(fixture.scheduler::close); + Assert.assertTrue(fixture.scheduler.pendingTasks() > 0); + Assert.assertFalse(close.isDone()); + task.release.countDown(); + cancel.get(10L, TimeUnit.SECONDS); + Assert.assertTrue(close.get(10L, TimeUnit.SECONDS)); + Assert.assertEquals(TaskStatus.CANCELLING, fixture.scheduler.task(task.id(), false).status()); + } finally { + task.release.countDown(); + finishWorker.countDown(); + occupied.get(10L, TimeUnit.SECONDS); + stop(requests); + } + } + } + + @Test + public void testQueuedPersistentClosePreservesNewTask() throws Exception { + try (Fixture fixture = new Fixture("queued_new")) { + fixture.worker.armed.set(true); + HugeTask task = task(false, new AtomicBoolean(), 9999910L); + fixture.scheduler.schedule(task); + Assert.assertTrue(fixture.worker.reached.await(10L, TimeUnit.SECONDS)); + Assert.assertEquals(1, fixture.scheduler.pendingTasks()); + Assert.assertTrue(fixture.scheduler.close()); + Assert.assertFalse(task.isCancelled()); + Assert.assertEquals(TaskStatus.NEW, task.status()); + Assert.assertEquals(TaskStatus.NEW, fixture.scheduler.task(task.id(), false).status()); + Assert.assertEquals(0, fixture.scheduler.pendingTasks()); + for (boolean ephemeral : new boolean[]{true, false}) { + Assert.assertThrows(IllegalStateException.class, + () -> fixture.scheduler.schedule(task(ephemeral, new AtomicBoolean(), 9999911L))); + } + } + } + + @Test + public void testBusyWorkerLeavesPersistentBacklogInDatabase() throws Exception { + try (Fixture fixture = new Fixture("busy_new")) { + CountDownLatch busy = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + Future occupied = fixture.worker.submit(() -> { + busy.countDown(); + Assert.assertTrue(release.await(10L, TimeUnit.SECONDS)); + return null; + }); + try { + Assert.assertTrue(busy.await(10L, TimeUnit.SECONDS)); + HugeTask task = task(false, new AtomicBoolean(), 9999912L); + fixture.scheduler.schedule(task); + Assert.assertEquals(0, fixture.scheduler.pendingTasks()); + Assert.assertTrue(fixture.worker.getQueue().isEmpty()); + Assert.assertEquals(TaskStatus.NEW, fixture.scheduler.task(task.id(), false).status()); + Assert.assertTrue(fixture.scheduler.close()); + } finally { + release.countDown(); + occupied.get(10L, TimeUnit.SECONDS); + } + } + } + + @Test + public void testQueuedOuterCancellationPreservesCallerTransaction() throws Exception { + try (Fixture fixture = new Fixture("queued_future")) { + fixture.worker.armed.set(true); + AtomicBoolean ran = new AtomicBoolean(); + HugeTask task = task(true, ran, 9999913L); + Future dispatch = fixture.scheduler.schedule(task); + Assert.assertTrue(fixture.worker.reached.await(10L, TimeUnit.SECONDS)); + fixture.graph.schema().vertexLabel("caller").useCustomizeStringId().create(); + fixture.graph.addVertex(org.apache.tinkerpop.gremlin.structure.T.id, "pending", + org.apache.tinkerpop.gremlin.structure.T.label, "caller"); + Assert.assertTrue(dispatch.cancel(false)); + Assert.assertTrue(fixture.graph.tx().isOpen()); + Assert.assertTrue(fixture.graph.vertices("pending").hasNext()); + fixture.graph.tx().rollback(); + Assert.assertFalse(task.isCancelled()); + Assert.assertEquals(0, fixture.scheduler.pendingTasks()); + Assert.assertTrue(fixture.scheduler.close()); + fixture.worker.release.countDown(); + fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS); + Assert.assertFalse(ran.get()); + } + } + + @Test + public void testActiveOuterCancellationRetainsActualWorkerOwner() throws Exception { + try (Fixture fixture = new Fixture("active_future")) { + CountDownLatch started = new CountDownLatch(1); + CountDownLatch finish = new CountDownLatch(1); + HugeTask task = new HugeTask<>(IdGenerator.of(-9999914L), null, new EphemeralJob() { + @Override + public String type() { + return "test"; + } + + @Override + public Object execute() { + this.graph().vertices().hasNext(); + started.countDown(); + awaitUninterruptibly(finish); + return null; + } + }); + task.type("test"); + task.name("active-owner"); + Future dispatch = fixture.scheduler.schedule(task); + ExecutorService requests = Executors.newSingleThreadExecutor(); + try { + Assert.assertTrue(started.await(10L, TimeUnit.SECONDS)); + Assert.assertTrue(dispatch.cancel(true)); + Future close = requests.submit(fixture.scheduler::close); + Assert.assertTrue(fixture.scheduler.pendingTasks() > 0); + Assert.assertFalse(close.isDone()); + finish.countDown(); + Assert.assertTrue(close.get(10L, TimeUnit.SECONDS)); + fixture.assertWorkerClean(); + } finally { + finish.countDown(); + stop(requests); + } + } + } + + @Test + public void testAcceptedLocalCancellationRetainsCallbackAndSupportsNestedCancel() throws Exception { + try (Fixture fixture = new Fixture("nested_callback")) { + fixture.worker.armed.set(true); + HugeTask other = task(true, new AtomicBoolean(), 9999919L); + GatedTask task = new GatedTask(-9999915L, new EphemeralJob() { + @Override + public String type() { + return "test"; + } + + @Override + public Object execute() { + Assert.fail("Cancelled queued work must not execute"); + return null; + } + + @Override + protected void cancelled() { + fixture.scheduler.cancel(other); + } + }); + Future dispatch = fixture.scheduler.schedule(task); + Assert.assertTrue(fixture.worker.reached.await(10L, TimeUnit.SECONDS)); + fixture.scheduler.schedule(other); + task.armed.set(true); + ExecutorService requests = Executors.newSingleThreadExecutor(); + try { + Future cancel = requests.submit(() -> { + TaskManager.setContext("nested-cancel"); + try { + fixture.scheduler.cancel(task); + Assert.assertEquals("nested-cancel", TaskManager.getContext()); + } finally { + TaskManager.resetContext(); + } + }); + Assert.assertTrue(task.reached.await(10L, TimeUnit.SECONDS)); + Assert.assertTrue(dispatch.cancel(false)); + Assert.assertTrue(fixture.scheduler.pendingTasks() > 0); + task.release.countDown(); + cancel.get(10L, TimeUnit.SECONDS); + Assert.assertTrue(task.isCancelled()); + Assert.assertTrue(other.isCancelled()); + Assert.assertEquals(0, fixture.scheduler.pendingTasks()); + } finally { + task.release.countDown(); + stop(requests); + } + } + } + + @Test + public void testCallbackCleanupCompletesBeforeGraphDrain() throws Exception { + try (Fixture fixture = new Fixture("callback_owner")) { + fixture.worker.armed.set(true); + CountDownLatch callback = new CountDownLatch(1); + CountDownLatch finish = new CountDownLatch(1); + HugeTask task = new HugeTask<>(IdGenerator.of(-9999916L), null, new EphemeralJob() { + @Override + public String type() { + return "test"; + } + + @Override + public Object execute() { + Assert.fail("Cancelled queued work must not execute"); + return null; + } + + @Override + protected void done() { + this.graph().vertices().hasNext(); + callback.countDown(); + awaitUninterruptibly(finish); + super.done(); + } + }); + task.type("test"); + task.name("callback-owner"); + Future dispatch = fixture.scheduler.schedule(task); + Assert.assertTrue(fixture.worker.reached.await(10L, TimeUnit.SECONDS)); + ExecutorService requests = Executors.newSingleThreadExecutor(); + ExecutorService callbacks = Executors.newSingleThreadExecutor(); + try { + Future cancel = callbacks.submit(() -> fixture.scheduler.cancel(task)); + Assert.assertTrue(callback.await(10L, TimeUnit.SECONDS)); + Assert.assertTrue(dispatch.cancel(true)); + Future close = requests.submit(fixture.scheduler::close); + Assert.assertTrue(fixture.scheduler.pendingTasks() > 0); + Assert.assertFalse(close.isDone()); + finish.countDown(); + cancel.get(10L, TimeUnit.SECONDS); + Assert.assertTrue(close.get(10L, TimeUnit.SECONDS)); + ThreadLocal owners = Whitebox.getInternalState(fixture.graph.tx(), "transactions"); + Assert.assertNull(callbacks.submit(owners::get).get(10L, TimeUnit.SECONDS)); + Assert.assertNull(callbacks.submit(TaskManager::getContext).get(10L, TimeUnit.SECONDS)); + } finally { + finish.countDown(); + stop(requests); + stop(callbacks); + } + } + } + + @Test + public void testBlockedCronStillFencesAdmissionAndTimesOutClose() throws Exception { + try (Fixture fixture = new Fixture("blocked_cron", 1L)) { + CountDownLatch querying = new CountDownLatch(1); + CountDownLatch finishQuery = new CountDownLatch(1); + fixture.scheduler.beforeTaskQuery = () -> { + querying.countDown(); + awaitUninterruptibly(finishQuery); + }; + ExecutorService requests = Executors.newSingleThreadExecutor(); + Future cron = fixture.cron.submit(fixture.scheduler::cronSchedule); + try { + Assert.assertTrue(querying.await(10L, TimeUnit.SECONDS)); + Future close = requests.submit(fixture.scheduler::close); + AtomicBoolean closed = Whitebox.getInternalState(fixture.scheduler, "closed"); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2L); + while (!closed.get() && System.nanoTime() < deadline) { + Thread.yield(); + } + Assert.assertTrue(closed.get()); + Assert.assertThrows(IllegalStateException.class, + () -> fixture.scheduler.schedule(task(true, new AtomicBoolean(), 9999919L))); + // This exercises the real cron barrier, not a mocked drain method. + Assert.assertFalse(close.get(10L, TimeUnit.SECONDS)); + Assert.assertFalse(cron.isDone()); + finishQuery.countDown(); + cron.get(10L, TimeUnit.SECONDS); + Assert.assertTrue(fixture.scheduler.close()); + fixture.assertWorkerClean(); + } finally { + finishQuery.countDown(); + cron.get(10L, TimeUnit.SECONDS); + stop(requests); + } + } + } + + private static void awaitUninterruptibly(CountDownLatch release) { + while (release.getCount() != 0L) { + try { + Assert.assertTrue(release.await(10L, TimeUnit.SECONDS)); + } catch (InterruptedException ignored) { + // Keep the real owner active after logical cancellation. + } + } + } + + private static void stop(ExecutorService executor) throws InterruptedException { + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(10L, TimeUnit.SECONDS)); + } + + private static HugeTask task(boolean ephemeral, AtomicBoolean ran, long id) { + HugeTask task; + if (ephemeral) { + task = new HugeTask<>(IdGenerator.of(-id), null, new EphemeralJob() { + @Override + public String type() { + return "test"; + } + + @Override + public Object execute() { + ran.set(true); + return null; + } + }); + } else { + task = new HugeTask<>(IdGenerator.of(id), null, new EmptyCallable()); + } + task.type("test"); + task.name("admission-test"); + return task; + } + + private static class Fixture implements AutoCloseable { + + private final HugeGraph graph; + private final ScheduledThreadPoolExecutor cron = new ScheduledThreadPoolExecutor(1); + private final ExecutorService database = Executors.newSingleThreadExecutor(); + private final GatedExecutor worker = new GatedExecutor(); + private final TestScheduler scheduler; + private final HugeGraphParams params; + private final TaskScheduler originalScheduler; + private final Map registry; + + Fixture(String suffix) { + this(suffix, 600L); + } + + Fixture(String suffix, long schedulePeriod) { + RegisterUtil.registerBackends(); + HugeConfig config = FakeObjects.newConfig(); + config.setProperty("backend", "memory"); + config.setProperty("serializer", "text"); + config.setProperty("store", "distributed_admission_" + suffix); + config.setProperty("task.schedule_period", schedulePeriod); + config.setProperty("task.retry", 0); + this.graph = HugeFactory.open(config); + this.graph.initBackend(); + this.graph.serverStarted(GlobalMasterInfo.master("distributed-admission-test")); + this.params = Whitebox.getInternalState(this.graph, "params"); + this.registry = Whitebox.getInternalState(TaskManager.instance(), "schedulers"); + this.originalScheduler = this.registry.get(this.params); + Assert.assertTrue(this.originalScheduler.close()); + this.scheduler = new TestScheduler(this.params, this.cron, this.database, this.worker); + this.scheduler.init(); + Assert.assertTrue(this.registry.replace(this.params, this.originalScheduler, this.scheduler)); + } + + void assertWorkerClean() throws Exception { + this.worker.submit(() -> null).get(10L, TimeUnit.SECONDS); + Assert.assertEquals(0, this.scheduler.pendingTasks()); + ThreadLocal owners = Whitebox.getInternalState(this.graph.tx(), "transactions"); + Assert.assertNull(this.worker.submit(owners::get).get(10L, TimeUnit.SECONDS)); + Assert.assertNull(this.worker.submit(TaskManager::getContext).get(10L, TimeUnit.SECONDS)); + } + + @Override + public void close() throws Exception { + this.worker.release.countDown(); + try { + Assert.assertTrue(this.scheduler.close()); + this.scheduler.closeDatabaseOwner(); + } finally { + Assert.assertTrue(this.registry.replace(this.params, this.scheduler, this.originalScheduler)); + for (ExecutorService executor : new ExecutorService[]{this.worker, this.database, this.cron}) { + stop(executor); + } + try { + this.graph.close(); + } finally { + HugeFactory.remove(this.graph); + } + } + } + } + + private static class GatedTask extends HugeTask { + + private final AtomicBoolean armed = new AtomicBoolean(); + private final CountDownLatch reached = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + + GatedTask(long id, TaskCallable callable) { + super(IdGenerator.of(id), null, callable); + this.type("test"); + this.name("accepted-cancel"); + } + + @Override + public boolean completed() { + if (this.armed.compareAndSet(true, false)) { + this.reached.countDown(); + awaitUninterruptibly(this.release); + } + return super.completed(); + } + } + + private static class GatedExecutor extends ThreadPoolExecutor { + + private final AtomicBoolean armed = new AtomicBoolean(); + private final CountDownLatch reached = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + + GatedExecutor() { + super(1, 1, 0L, TimeUnit.SECONDS, new LinkedBlockingQueue<>()); + } + + @Override + protected void beforeExecute(Thread thread, Runnable work) { + if (this.armed.compareAndSet(true, false)) { + this.reached.countDown(); + awaitUninterruptibly(this.release); + } + } + } + + private static class TestScheduler extends DistributedTaskScheduler { + + private volatile Runnable beforeTaskRead = () -> { }; + private volatile Runnable beforeTaskQuery = () -> { }; + + @Override + protected Iterator> queryTaskWithoutResultByStatus(TaskStatus status) { + this.beforeTaskQuery.run(); + return super.queryTaskWithoutResultByStatus(status); + } + + @Override + public HugeTask task(Id id, boolean withResult) { + this.beforeTaskRead.run(); + return super.task(id, withResult); + } + + TestScheduler(HugeGraphParams graph, ScheduledThreadPoolExecutor cron, + ExecutorService database, ExecutorService worker) { + super(graph, cron, database, worker, worker, worker, worker); + } + + void closeDatabaseOwner() { + this.call(() -> { + this.tx().close(); + this.graph.closeTx(); + return null; + }); + } + } +} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/GraphDropPendingTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/GraphDropPendingTest.java new file mode 100644 index 0000000000..a7a0c57125 --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/GraphDropPendingTest.java @@ -0,0 +1,342 @@ +/* + * 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; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.commons.io.FileUtils; +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.dist.RegisterUtil; +import org.apache.hugegraph.event.EventHub; +import org.apache.hugegraph.exception.HugeException; +import org.apache.hugegraph.id.IdGenerator; +import org.apache.hugegraph.job.SysJob; +import org.apache.hugegraph.masterelection.GlobalMasterInfo; +import org.apache.hugegraph.meta.MetaManager; +import org.apache.hugegraph.space.GraphSpace; +import org.apache.hugegraph.task.HugeTask; +import org.apache.hugegraph.task.TaskScheduler; +import org.apache.hugegraph.testutil.Assert; +import org.apache.hugegraph.testutil.Whitebox; +import org.apache.hugegraph.unit.FakeObjects; +import org.apache.hugegraph.util.Events; +import org.apache.tinkerpop.gremlin.structure.Graph; +import org.apache.tinkerpop.gremlin.structure.T; +import org.junit.Test; +import org.mockito.Mockito; + +public class GraphDropPendingTest { + + @Test + public void testLocalDropKeepsDataAndRegistrationUntilCancelledWorkerExits() throws Exception { + try (Fixture fixture = new Fixture("local_pending", false)) { + BlockingJob job = fixture.start(); + fixture.scheduler.cancel(job.task()); + Assert.assertTrue(job.task().isCancelled()); + HugeException refusal = Assert.assertThrows(HugeException.class, fixture::drop); + Assert.assertContains("Tasks are still active", refusal.getMessage()); + Assert.assertContains("task admission remains closed", refusal.getMessage()); + Assert.assertContains("restart to resume task admission", refusal.getMessage()); + BlockingJob late = new BlockingJob(); + late.release.countDown(); + Assert.assertThrows(IllegalStateException.class, () -> fixture.scheduler.schedule( + new HugeTask<>(IdGenerator.of(9997001L), null, late))); + fixture.assertRetained(); + job.release.countDown(); + fixture.scheduler.waitUntilAllTasksCompleted(10L); + fixture.drop(); + Assert.assertTrue(fixture.graph.closed()); + Assert.assertNull(fixture.manager.graph(fixture.graph.spaceGraphName())); + Assert.assertFalse(Files.exists(fixture.file)); + Assert.assertEquals(1, fixture.dropEvents); + } + } + + @Test + public void testPdDropPreservesMetadataUntilLocalTasksDrain() throws Exception { + try (Fixture fixture = new Fixture("pd_pending", true)) { + BlockingJob job = fixture.start(); + HugeException refusal = Assert.assertThrows(HugeException.class, fixture::drop); + Assert.assertContains("Tasks are still active", refusal.getMessage()); + Assert.assertContains("task admission remains closed", refusal.getMessage()); + Assert.assertContains("restart to resume task admission", refusal.getMessage()); + BlockingJob late = new BlockingJob(); + late.release.countDown(); + Assert.assertThrows(IllegalStateException.class, () -> fixture.scheduler.schedule( + new HugeTask<>(IdGenerator.of(9997001L), null, late))); + fixture.assertRetained(); + Assert.assertTrue(fixture.metadataPresent.get()); + job.release.countDown(); + fixture.scheduler.waitUntilAllTasksCompleted(10L); + fixture.drop(); + Assert.assertFalse(fixture.metadataPresent.get()); + Assert.assertTrue(fixture.graph.closed()); + Assert.assertNull(fixture.manager.graph(fixture.graph.spaceGraphName())); + } + } + + @Test + public void testConfiguredWaitAllowsShortTaskToFinishBeforePdDrop() throws Exception { + try (Fixture fixture = new Fixture("pd_wait", true)) { + fixture.graph.configuration().setProperty(CoreOptions.TASK_WAIT_TIMEOUT.name(), 2L); + BlockingJob job = fixture.start(); + CountDownLatch fenced = new CountDownLatch(1); + TaskScheduler scheduler = Mockito.mock(TaskScheduler.class, + org.mockito.AdditionalAnswers.delegatesTo(fixture.scheduler)); + Mockito.doAnswer(invocation -> { + boolean drained = fixture.scheduler.close(); + if (!drained) { + fenced.countDown(); + } + return drained; + }).when(scheduler).close(); + HugeGraph proxy = Mockito.mock(HugeGraph.class, org.mockito.AdditionalAnswers.delegatesTo(fixture.graph)); + Mockito.doReturn(scheduler).when(proxy).taskScheduler(); + fixture.graphs.put(fixture.graph.spaceGraphName(), proxy); + ExecutorService caller = Executors.newSingleThreadExecutor(); + try { + Future deletion = caller.submit(fixture::drop); + Assert.assertTrue(fenced.await(10L, TimeUnit.SECONDS)); + Assert.assertTrue(fixture.metadataPresent.get()); + job.release.countDown(); + deletion.get(10L, TimeUnit.SECONDS); + Assert.assertTrue(fixture.graph.closed()); + Assert.assertFalse(fixture.metadataPresent.get()); + } finally { + job.release.countDown(); + caller.shutdown(); + Assert.assertTrue(caller.awaitTermination(10L, TimeUnit.SECONDS)); + fixture.graphs.replace(fixture.graph.spaceGraphName(), proxy, fixture.graph); + } + } + } + + @Test + public void testSchedulerCloseFailureDoesNotRemovePdMetadata() throws Exception { + try (Fixture fixture = new Fixture("pd_close_failure", true)) { + Graph proxy = Mockito.mock(HugeGraph.class, Mockito.withSettings().defaultAnswer( + org.mockito.AdditionalAnswers.delegatesTo(fixture.graph))); + TaskScheduler failed = Mockito.spy(fixture.scheduler); + Mockito.doThrow(new IllegalStateException("scheduler close failed")).when(failed).close(); + Mockito.doReturn(failed).when((HugeGraph) proxy).taskScheduler(); + fixture.graphs.put(fixture.graph.spaceGraphName(), proxy); + try { + RuntimeException error = Assert.assertThrows(RuntimeException.class, fixture::drop); + Assert.assertContains("scheduler close failed", error.getMessage()); + Assert.assertTrue(fixture.metadataPresent.get()); + Assert.assertFalse(fixture.graph.closed()); + Assert.assertTrue(Files.exists(fixture.file)); + } finally { + fixture.graphs.put(fixture.graph.spaceGraphName(), fixture.graph); + } + } + } + + @Test + public void testMissingGraphReportsArgumentErrorFromInitializedManager() throws Exception { + try (Fixture fixture = new Fixture("missing_graph", true)) { + IllegalArgumentException error = Assert.assertThrows(IllegalArgumentException.class, + () -> fixture.manager.dropGraph("DEFAULT", "missing", true)); + Assert.assertContains("doesn't exist", error.getMessage()); + fixture.assertRetained(); + Assert.assertTrue(fixture.metadataPresent.get()); + } + } + + @Test + public void testGraphSpaceSharedMetadataSurvivesRefusedDrop() throws Exception { + try (Fixture fixture = new Fixture("space_pending", true)) { + BlockingJob job = fixture.start(); + HugeException refusal = Assert.assertThrows(HugeException.class, + () -> fixture.manager.clearGraphSpace("DEFAULT")); + Assert.assertContains("Tasks are still active", refusal.getMessage()); + Assert.assertFalse(fixture.sharedMetadataCleared.get()); + Assert.assertSame(fixture.graph, fixture.manager.graph(fixture.graph.spaceGraphName())); + job.release.countDown(); + } + } + + private static final class Fixture implements AutoCloseable { + + private final Path directory; + private final Path file; + private final GraphManager manager; + private final Map graphs; + private final HugeGraph graph; + private final HugeGraph neighbor; + private final HugeGraphParams params; + private final TaskScheduler scheduler; + private final boolean pd; + private final AtomicBoolean metadataPresent = new AtomicBoolean(true); + private final AtomicBoolean sharedMetadataCleared = new AtomicBoolean(); + private final List jobs = new ArrayList<>(); + private int dropEvents; + + private Fixture(String name, boolean pd) throws Exception { + RegisterUtil.registerBackends(); + this.pd = pd; + this.directory = Files.createTempDirectory("graph-drop-core-"); + HugeConfig config = FakeObjects.newConfig(); + config.setProperty(ServerOptions.USE_PD.name(), false); + config.setProperty(ServerOptions.GRAPH_LOAD_FROM_LOCAL_CONFIG.name(), false); + config.setProperty(ServerOptions.ENABLE_DYNAMIC_CREATE_DROP.name(), true); + config.setProperty(ServerOptions.GRAPHS.name(), this.directory.toString()); + EventHub events = new EventHub("drop-core-" + name); + this.manager = new GraphManager(config, events); + this.graphs = Whitebox.getInternalState(this.manager, "graphs"); + this.graph = this.open(name); + this.neighbor = this.open(name + "_neighbor"); + this.params = Whitebox.getInternalState(this.graph, "params"); + this.scheduler = this.graph.taskScheduler(); + this.file = this.directory.resolve(name + ".properties"); + events.listen(Events.GRAPH_DROP, event -> { + if (event.args()[0] == this.graph) { + this.dropEvents++; + } + return null; + }); + if (pd) { + // Replace the external PD service, never bypass manager initialization. + Whitebox.setInternalState(this.manager, "PDExist", true); + MetaManager meta = Mockito.mock(MetaManager.class); + Whitebox.setInternalState(this.manager, "metaManager", meta); + Mockito.when(meta.graphConfigs("DEFAULT")).thenReturn(java.util.Collections.emptyMap()); + Mockito.doAnswer(invocation -> { + this.metadataPresent.set(false); + return null; + }).when(meta).removeGraphConfig(Mockito.eq("DEFAULT"), Mockito.eq(name)); + Mockito.doAnswer(invocation -> { + this.sharedMetadataCleared.set(true); + return null; + }).when(meta).clearGraphAuth("DEFAULT"); + Map spaces = Whitebox.getInternalState(this.manager, "graphSpaces"); + spaces.put("DEFAULT", Mockito.mock(GraphSpace.class)); + } + } + + private HugeGraph open(String name) throws Exception { + HugeConfig config = FakeObjects.newConfig(); + config.setProperty(Graph.GRAPH, HugeFactory.class.getName()); + config.setProperty("backend", "memory"); + config.setProperty("serializer", "text"); + config.setProperty("store", name); + config.setProperty(CoreOptions.TASK_WAIT_TIMEOUT.name(), 0L); + Path path = this.directory.resolve(name + ".properties"); + config.save(path.toFile()); + config.file(path.toString()); + HugeGraph result = HugeFactory.open(config); + result.initBackend(); + result.serverStarted(GlobalMasterInfo.master("drop-core-test")); + result.schema().vertexLabel("person").useCustomizeStringId().create(); + result.addVertex(T.id, "kept", T.label, "person"); + result.tx().commit(); + ((HugeGraphParams) Whitebox.getInternalState(result, "params")).closeTx(); + this.graphs.put(result.spaceGraphName(), result); + return result; + } + + private BlockingJob start() throws Exception { + BlockingJob job = new BlockingJob(); + this.jobs.add(job); + HugeTask task = new HugeTask<>(IdGenerator.of(9997000L), null, job); + task.type("test"); + task.name("drop-owner"); + this.scheduler.schedule(task); + Assert.assertTrue(job.started.await(10L, TimeUnit.SECONDS)); + return job; + } + + private void drop() { + if (this.pd) { + this.manager.dropGraph("DEFAULT", this.graph.name(), true); + } else { + this.manager.dropGraphLocal(this.graph.name()); + } + } + + private void assertRetained() { + Assert.assertSame(this.graph, this.manager.graph(this.graph.spaceGraphName())); + Assert.assertFalse(this.graph.closed()); + Assert.assertTrue(Files.exists(this.file)); + Assert.assertEquals(IdGenerator.of("kept"), this.graph.vertices("kept").next().id()); + this.params.closeTx(); + Map registered = Whitebox.getInternalState(HugeFactory.class, "GRAPHS"); + Assert.assertSame(this.graph, registered.get(this.graph.spaceGraphName())); + Assert.assertEquals(0, this.dropEvents); + } + + @Override + public void close() throws Exception { + for (BlockingJob job : this.jobs) { + job.release.countDown(); + } + this.scheduler.waitUntilAllTasksCompleted(10L); + this.manager.close(); + for (HugeGraph current : new HugeGraph[]{this.graph, this.neighbor}) { + if (!current.closed()) { + current.close(); + } + HugeFactory.remove(current); + } + this.graphs.clear(); + FileUtils.deleteDirectory(this.directory.toFile()); + } + } + + public static class BlockingJob extends SysJob { + + private final CountDownLatch started = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + + @Override + public String type() { + return "test"; + } + + @Override + public Object execute() { + this.graph().vertices("kept").next(); + this.started.countDown(); + boolean interrupted = false; + while (this.release.getCount() != 0L) { + try { + this.release.await(); + } catch (InterruptedException ignored) { + interrupted = true; + } + } + if (interrupted) { + Thread.currentThread().interrupt(); + } + return null; + } + } +} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java index ba34359d46..e6251122e1 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java @@ -18,6 +18,7 @@ package org.apache.hugegraph; import java.util.ArrayList; +import java.lang.reflect.Constructor; import java.util.Arrays; import java.util.List; import java.util.Map; @@ -76,12 +77,12 @@ public void testCleanupContinuesAfterGraphFailure() { } @Test - public void testGraphCloseContinuesAfterReleasedWorkerOwnerFails() throws Exception { + public void testGraphCloseRetainsSchedulerAfterReleasedWorkerOwnerFails() throws Exception { this.checkGraphCloseAfterOwnerFailure(false); } @Test - public void testGraphCloseContinuesAfterWorkerOwnerFailsBeforeRelease() throws Exception { + public void testGraphCloseRetainsSchedulerAfterWorkerOwnerFailsBeforeRelease() throws Exception { this.checkGraphCloseAfterOwnerFailure(true); } @@ -129,14 +130,27 @@ private void checkGraphCloseAfterOwnerFailure(boolean beforeRelease) throws Exce Throwable failure = Assert.assertThrows(HugeException.class, graph::close); Assert.assertSame(first, HugeException.rootCause(failure)); Assert.assertTrue(Arrays.asList(failure.getSuppressed()).contains(second)); - Assert.assertTrue(graph.closed()); - Assert.assertNull(manager.getScheduler(params)); + Assert.assertFalse(graph.closed()); + Assert.assertSame(scheduler, manager.getScheduler(params)); Assert.assertTrue((boolean) Whitebox.getInternalState(scheduler.serverManager(), "closed")); Assert.assertNull(scheduler.call(owners::get)); Assert.assertNull(owners.get()); Assert.assertEquals(1, workerClosed.get()); Assert.assertEquals(1, callerClosed.get()); Assert.assertEquals(1, otherWorkerClosed.get()); + Mockito.verify(provider, Mockito.never()).close(); + + // A close failure is not an ACK, even if its owner released first. + if (beforeRelease) { + scheduler.call(() -> { + workerOwner.get().close(); + return null; + }); + } + graph.close(); + Assert.assertTrue(graph.closed()); + Assert.assertNull(manager.getScheduler(params)); + Assert.assertEquals(1, otherWorkerClosed.get()); Mockito.verify(provider).close(); } finally { try { @@ -363,6 +377,53 @@ public void testIncompleteSchedulerCloseDoesNotQueueOwnerCleanup() { } } + @Test + public void testShutdownDrainsWorkersWithRegisteredScheduler() throws Exception { + Constructor constructor = TaskManager.class.getDeclaredConstructor(int.class); + constructor.setAccessible(true); + TaskManager manager = constructor.newInstance(1); + Map schedulers = Whitebox.getInternalState(manager, "schedulers"); + HugeGraphParams params = Mockito.mock(HugeGraphParams.class); + TaskScheduler scheduler = Mockito.mock(TaskScheduler.class); + schedulers.put(params, scheduler); + ExecutorService workers = Whitebox.getInternalState(manager, "taskExecutor"); + ExecutorService database = Whitebox.getInternalState(manager, "taskDbExecutor"); + CountDownLatch started = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicBoolean saved = new AtomicBoolean(); + ExecutorService stoppedWorkers = Mockito.spy(workers); + Mockito.doAnswer(invocation -> { + release.countDown(); + return invocation.callRealMethod(); + }).when(stoppedWorkers).shutdown(); + Whitebox.setInternalState(manager, "taskExecutor", stoppedWorkers); + try { + Future task = workers.submit(() -> { + started.countDown(); + Assert.assertTrue(release.await(10L, TimeUnit.SECONDS)); + database.submit(() -> saved.set(true)).get(10L, TimeUnit.SECONDS); + return null; + }); + Assert.assertTrue(started.await(10L, TimeUnit.SECONDS)); + manager.closeScheduler(params); + Assert.assertSame(scheduler, manager.getScheduler(params)); + manager.shutdown(10L); + task.get(10L, TimeUnit.SECONDS); + Assert.assertTrue(saved.get()); + Assert.assertSame(scheduler, manager.getScheduler(params)); + for (String field : new String[]{"taskExecutor", "taskDbExecutor", "schemaTaskExecutor", + "olapTaskExecutor", "ephemeralTaskExecutor", + "distributedSchedulerExecutor"}) { + ExecutorService executor = Whitebox.getInternalState(manager, field); + Assert.assertTrue(executor.isTerminated()); + } + } finally { + release.countDown(); + schedulers.clear(); + manager.shutdown(10L); + } + } + @Test public void testDistributedCloseRetriesDrainAfterTimeout() throws Exception { HugeGraphParams params = Mockito.mock(HugeGraphParams.class); diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/StandardTaskAdmissionTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/StandardTaskAdmissionTest.java new file mode 100644 index 0000000000..1425693fa3 --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/StandardTaskAdmissionTest.java @@ -0,0 +1,331 @@ +/* + * 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; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.dist.RegisterUtil; +import org.apache.hugegraph.id.IdGenerator; +import org.apache.hugegraph.job.SysJob; +import org.apache.hugegraph.masterelection.GlobalMasterInfo; +import org.apache.hugegraph.task.HugeTask; +import org.apache.hugegraph.task.StandardTaskScheduler; +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.unit.FakeObjects; +import org.apache.tinkerpop.gremlin.structure.T; +import org.junit.Test; +import org.mockito.Mockito; + +public class StandardTaskAdmissionTest { + + @Test + public void testCancelledWorkerFencesAdmissionUntilActualExit() throws Exception { + try (Fixture fixture = new Fixture("standard_cancel_worker")) { + BlockingJob job = new BlockingJob(); + HugeTask task = fixture.task(job); + fixture.scheduler.schedule(task); + Assert.assertTrue(job.started.await(10L, TimeUnit.SECONDS)); + fixture.scheduler.cancel(task); + Assert.assertTrue(task.isCancelled()); + Assert.assertEquals(1, fixture.scheduler.pendingTasks()); + Assert.assertFalse(fixture.scheduler.close()); + HugeTask late = fixture.task(new EmptyJob()); + Assert.assertThrows(IllegalStateException.class, () -> fixture.scheduler.schedule(late)); + Assert.assertThrows(IllegalStateException.class, () -> { + fixture.scheduler.restoreTasks(); + }, error -> Assert.assertContains("closing", error.getMessage())); + Assert.assertEquals(TaskStatus.NEW, late.status()); + job.release.countDown(); + fixture.scheduler.waitUntilAllTasksCompleted(10L); + Assert.assertEquals(TaskStatus.CANCELLED, fixture.scheduler.findTask(task.id()).status()); + Assert.assertTrue(fixture.scheduler.close()); + } + } + + @Test + public void testDispatchFutureCancellationKeepsRunningOwner() throws Exception { + try (Fixture fixture = new Fixture("standard_cancel_future")) { + BlockingJob job = new BlockingJob(); + HugeTask task = fixture.task(job); + Future dispatch = fixture.scheduler.schedule(task); + Assert.assertTrue(job.started.await(10L, TimeUnit.SECONDS)); + Assert.assertTrue(dispatch.cancel(false)); + Assert.assertFalse(task.isCancelled()); + Assert.assertFalse(fixture.scheduler.close()); + job.release.countDown(); + fixture.scheduler.waitUntilAllTasksCompleted(10L); + Assert.assertEquals(TaskStatus.SUCCESS, fixture.scheduler.findTask(task.id()).status()); + Assert.assertTrue(fixture.scheduler.close()); + } + } + + @Test + public void testCancellationCallbackRetainsItsTransactionAfterWorkerExits() throws Exception { + try (Fixture fixture = new Fixture("standard_cancel_callback")) { + fixture.graph.schema().vertexLabel("person").useCustomizeStringId().create(); + CountDownLatch callback = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + BlockingJob job = new BlockingJob() { + @Override + protected void cancelled() { + this.graph().addVertex(T.id, "callback", T.label, "person"); + callback.countDown(); + await(release); + this.graph().tx().commit(); + super.cancelled(); + } + }; + HugeTask task = fixture.task(job); + fixture.scheduler.schedule(task); + Assert.assertTrue(job.started.await(10L, TimeUnit.SECONDS)); + ExecutorService caller = Executors.newSingleThreadExecutor(); + try { + Future cancellation = caller.submit(() -> fixture.scheduler.cancel(task)); + Assert.assertTrue(callback.await(10L, TimeUnit.SECONDS)); + job.release.countDown(); + fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS); + Assert.assertEquals(1, fixture.scheduler.pendingTasks()); + Assert.assertFalse(fixture.scheduler.close()); + release.countDown(); + cancellation.get(10L, TimeUnit.SECONDS); + fixture.scheduler.waitUntilAllTasksCompleted(10L); + Assert.assertTrue(fixture.graph.vertices("callback").hasNext()); + Assert.assertEquals(TaskStatus.CANCELLED, fixture.scheduler.findTask(task.id()).status()); + } finally { + release.countDown(); + caller.shutdownNow(); + Assert.assertTrue(caller.awaitTermination(10L, TimeUnit.SECONDS)); + } + } + } + + @Test + public void testQueuedCancellationAndSubmissionRejectionReleaseOwners() throws Exception { + try (Fixture fixture = new Fixture("standard_cancel_queued")) { + CountDownLatch occupied = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + fixture.worker.submit(() -> { + occupied.countDown(); + release.await(); + return null; + }); + try { + Assert.assertTrue(occupied.await(10L, TimeUnit.SECONDS)); + BlockingJob job = new BlockingJob(); + Future queued = fixture.scheduler.schedule(fixture.task(job)); + Assert.assertTrue(queued.cancel(false)); + Assert.assertEquals(0, fixture.scheduler.pendingTasks()); + release.countDown(); + fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS); + Assert.assertEquals(1L, job.started.getCount()); + fixture.worker.shutdown(); + Assert.assertTrue(fixture.worker.awaitTermination(10L, TimeUnit.SECONDS)); + Assert.assertThrows(RejectedExecutionException.class, + () -> fixture.scheduler.schedule(fixture.task(new EmptyJob()))); + Assert.assertEquals(0, fixture.scheduler.pendingTasks()); + } finally { + release.countDown(); + } + } + } + + @Test + public void testDependencyRetryDoesNotReleaseEarlierCleanupOwner() throws Exception { + try (Fixture fixture = new Fixture("standard_dependency_owner")) { + CountDownLatch closing = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicBoolean first = new AtomicBoolean(true); + Mockito.doAnswer(invocation -> { + if (Thread.currentThread().getName().equals("standard-test-worker") && + first.compareAndSet(true, false)) { + closing.countDown(); + Assert.assertTrue(release.await(10L, TimeUnit.SECONDS)); + } + return invocation.callRealMethod(); + }).when(fixture.params).closeTx(); + HugeTask dependency = fixture.task(new EmptyJob()); + fixture.scheduler.save(dependency); + HugeTask task = fixture.task(new EmptyJob()); + task.depends(dependency.id()); + fixture.scheduler.schedule(task); + try { + Assert.assertTrue(closing.await(10L, TimeUnit.SECONDS)); + fixture.scheduler.cancel(task); + Assert.assertEquals(1, fixture.scheduler.pendingTasks()); + Assert.assertFalse(fixture.scheduler.close()); + } finally { + release.countDown(); + } + fixture.scheduler.waitUntilAllTasksCompleted(10L); + Assert.assertEquals(TaskStatus.CANCELLED, fixture.scheduler.findTask(task.id()).status()); + } + } + + @Test + public void testCloseFencesDependencyResubmission() throws Exception { + try (Fixture fixture = new Fixture("standard_dependency_fence")) { + CountDownLatch occupied = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + fixture.worker.submit(() -> { + occupied.countDown(); + release.await(); + return null; + }); + try { + Assert.assertTrue(occupied.await(10L, TimeUnit.SECONDS)); + HugeTask dependency = fixture.task(new EmptyJob()); + fixture.scheduler.save(dependency); + HugeTask task = fixture.task(new EmptyJob()); + task.depends(dependency.id()); + Future dispatch = fixture.scheduler.schedule(task); + Assert.assertFalse(fixture.scheduler.close()); + release.countDown(); + dispatch.get(10L, TimeUnit.SECONDS); + fixture.scheduler.waitUntilAllTasksCompleted(10L); + Assert.assertEquals(TaskStatus.FAILED, fixture.scheduler.findTask(task.id()).status()); + Assert.assertContains("is closing", task.result()); + Assert.assertTrue(fixture.scheduler.close()); + } finally { + release.countDown(); + } + } + } + + @Test + public void testCloseFailureRemainsRetryable() throws Exception { + try (Fixture fixture = new Fixture("standard_close_failure")) { + Mockito.doThrow(new IllegalStateException("close failed")).doCallRealMethod() + .when(fixture.params).closeTx(); + Assert.assertThrows(RuntimeException.class, fixture.scheduler::close); + HugeTask late = fixture.task(new EmptyJob()); + Assert.assertThrows(IllegalStateException.class, () -> fixture.scheduler.schedule(late)); + Assert.assertTrue(fixture.scheduler.close()); + } + } + + private static void await(CountDownLatch latch) { + boolean interrupted = false; + while (true) { + try { + latch.await(); + break; + } catch (InterruptedException ignored) { + interrupted = true; + } + } + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + + private static class Fixture implements AutoCloseable { + + private final HugeGraph graph; + private final HugeGraphParams params; + private final ExecutorService worker; + private final ExecutorService database; + private final StandardTaskScheduler scheduler; + private final List jobs = new ArrayList<>(); + private long nextId = 9998000L; + + private Fixture(String name) { + RegisterUtil.registerBackends(); + HugeConfig config = FakeObjects.newConfig(); + config.setProperty("backend", "memory"); + config.setProperty("serializer", "text"); + config.setProperty("store", name); + this.graph = HugeFactory.open(config); + this.graph.initBackend(); + this.graph.serverStarted(GlobalMasterInfo.master("standard-owner-test")); + this.params = Mockito.spy((HugeGraphParams) Whitebox.getInternalState(this.graph, "params")); + this.worker = Executors.newSingleThreadExecutor(runnable -> + new Thread(runnable, "standard-test-worker")); + this.database = Executors.newSingleThreadExecutor(); + this.scheduler = new StandardTaskScheduler(this.params, this.worker, this.database); + } + + private HugeTask task(TaskCallable job) { + if (job instanceof BlockingJob) { + this.jobs.add((BlockingJob) job); + } + HugeTask task = new HugeTask<>(IdGenerator.of(this.nextId++), null, job); + task.type("test"); + task.name("standard-owner"); + return task; + } + + @Override + public void close() throws Exception { + for (BlockingJob job : this.jobs) { + job.release.countDown(); + } + this.scheduler.close(); + this.worker.shutdown(); + Assert.assertTrue(this.worker.awaitTermination(10L, TimeUnit.SECONDS)); + Assert.assertTrue(this.scheduler.close()); + this.database.shutdownNow(); + Assert.assertTrue(this.database.awaitTermination(10L, TimeUnit.SECONDS)); + this.graph.close(); + HugeFactory.remove(this.graph); + } + } + + public static class EmptyJob extends SysJob { + + @Override + public String type() { + return "test"; + } + + @Override + public Object execute() { + return null; + } + } + + public static class BlockingJob extends SysJob { + + private final CountDownLatch started = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + + @Override + public String type() { + return "test"; + } + + @Override + public Object execute() { + this.graph().vertices().hasNext(); + this.started.countDown(); + await(this.release); + return null; + } + } +} 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 0622630ecc..3a3837a2e1 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 @@ -234,7 +234,8 @@ public void testDeleteIncompleteTask() { TaskScheduler scheduler = graph.taskScheduler(); Id id = IdGenerator.of(88891); - HugeTask task = new HugeTask<>(id, null, new SleepCallable<>()); + BlockingCallable.reset(); + HugeTask task = new HugeTask<>(id, null, new BlockingCallable<>()); task.type("test"); task.name("delete-incomplete-task"); scheduler.schedule(task); @@ -248,13 +249,14 @@ public void testDeleteIncompleteTask() { e.getMessage()); }); } else { - waitUntilTaskRunning(scheduler); + Assert.assertTrue(BlockingCallable.awaitStarted()); HugeTask deleted = scheduler.delete(id, false); 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) { @@ -604,6 +606,9 @@ public Object execute() throws Exception { Assert.assertEquals(TaskStatus.SUCCESS, task.status()); Assert.assertEquals("{\"k1\":13579,\"k2\":\"24680\"}", task.result()); + // Future.get() publishes the result before done() and native owner cleanup finish. + scheduler.waitUntilAllTasksCompleted(10); + // Ephemeral tasks are not stored in DB, so these should throw NotFoundException Assert.assertThrows(NotFoundException.class, () -> { scheduler.waitUntilTaskCompleted(task.id(), 10); diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/VertexCoreTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/VertexCoreTest.java index d0c22e2f22..8268291555 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/VertexCoreTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/VertexCoreTest.java @@ -1309,11 +1309,14 @@ public void testAddVertexWithTtl() { @Test public void testAddVertexWithTtlAndTtlStartTime() { + // The input date must not reuse a previously closed transaction's timestamp. + long birth = DateUtil.now().getTime() - 1000L; Vertex vertex = graph().addVertex( T.label, "follower", "name", "Baby", "age", 3, "city", "Beijing", - "birth", graph().now() - 1000L); + "birth", birth); this.commitTx(); + Assert.assertEquals(birth + 3000L, ((HugeElement) vertex).expiredTime()); Iterator vertices = graph().vertices(vertex); Assert.assertTrue(vertices.hasNext()); 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 b83e083a77..c36bed15f0 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 @@ -18,6 +18,9 @@ package org.apache.hugegraph.unit; import org.apache.hugegraph.HugeFactoryTest; +import org.apache.hugegraph.GraphDropPendingTest; +import org.apache.hugegraph.StandardTaskAdmissionTest; +import org.apache.hugegraph.DistributedTaskAdmissionTest; import org.apache.hugegraph.api.auth.GraphSpaceAuthPayloadTest; import org.apache.hugegraph.api.auth.GraphSpaceGroupAPITest; import org.apache.hugegraph.api.cypher.CypherClientTest; @@ -150,6 +153,9 @@ StandardAuthManagerV2Test.class, ContextTaskTest.class, HugeFactoryTest.class, + GraphDropPendingTest.class, + StandardTaskAdmissionTest.class, + DistributedTaskAdmissionTest.class, AuthMetaManagerTest.class, /* api space */ diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/space/GraphSpaceAPITest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/space/GraphSpaceAPITest.java index 4d893e3199..1a0b585b0c 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/space/GraphSpaceAPITest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/space/GraphSpaceAPITest.java @@ -19,7 +19,6 @@ import static org.apache.hugegraph.HugeFactory.SYS_GRAPH; -import java.lang.reflect.Field; import java.lang.reflect.Method; import java.util.Collections; import java.util.Map; @@ -37,11 +36,15 @@ import org.apache.hugegraph.auth.HugeUser; import org.apache.hugegraph.auth.RolePermission; import org.apache.hugegraph.core.GraphManager; +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.config.ServerOptions; +import org.apache.hugegraph.event.EventHub; import org.apache.hugegraph.meta.MetaManager; import org.apache.hugegraph.space.GraphSpace; import org.apache.hugegraph.testutil.Assert; import org.apache.hugegraph.testutil.Whitebox; import org.apache.hugegraph.unit.BaseUnitTest; +import org.apache.hugegraph.unit.FakeObjects; import org.apache.tinkerpop.gremlin.structure.Graph; import org.junit.After; import org.junit.Test; @@ -49,7 +52,6 @@ import jakarta.ws.rs.ForbiddenException; import jakarta.ws.rs.core.SecurityContext; -import sun.misc.Unsafe; public class GraphSpaceAPITest extends BaseUnitTest { @@ -221,7 +223,7 @@ public void testSpaceManagerCannotDeleteSpaceDefaultRole() { @Test public void testPdGraphsIncludesLoadedLocalConfigGraphs() { - GraphManager manager = allocateGraphManager(); + GraphManager manager = newGraphManager(); Whitebox.setInternalState(manager, "PDExist", true); Whitebox.setInternalState(manager, "localGraphs", Collections.emptySet()); @@ -276,7 +278,7 @@ private static GraphManager managerWithDefaultRoleContext(String operator, HugeAuthenticator authenticator = Mockito.mock(HugeAuthenticator.class); Mockito.when(authenticator.authManager()).thenReturn(authManager); - GraphManager manager = allocateGraphManager(); + GraphManager manager = newGraphManager(); Whitebox.setInternalState(manager, "PDExist", true); Whitebox.setInternalState(manager, "authenticator", authenticator); Map spaces = new ConcurrentHashMap<>(); @@ -301,7 +303,7 @@ private static GraphManager managerWithDefaultRoleContext(String operator, private static GraphManager managerWithLoadedDefaultGraph(String operator, boolean admin) { - GraphManager manager = allocateGraphManager(); + GraphManager manager = newGraphManager(); Whitebox.setInternalState(manager, "PDExist", true); Whitebox.setInternalState(manager, "localGraphs", Collections.singleton("hugegraph")); @@ -326,15 +328,11 @@ private static GraphManager managerWithLoadedDefaultGraph(String operator, return manager; } - private static GraphManager allocateGraphManager() { - try { - Field field = Unsafe.class.getDeclaredField("theUnsafe"); - field.setAccessible(true); - Unsafe unsafe = (Unsafe) field.get(null); - return (GraphManager) unsafe.allocateInstance(GraphManager.class); - } catch (Exception e) { - throw new AssertionError(e); - } + private static GraphManager newGraphManager() { + HugeConfig config = FakeObjects.newConfig(); + config.setProperty(ServerOptions.USE_PD.name(), false); + config.setProperty(ServerOptions.GRAPH_LOAD_FROM_LOCAL_CONFIG.name(), false); + return new GraphManager(config, new EventHub("graph-space-api-test")); } private static void setContext(String username) { diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/util/ConsumersTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/util/ConsumersTest.java index 9db801be19..ea2cb0e0bc 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/util/ConsumersTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/util/ConsumersTest.java @@ -86,15 +86,18 @@ public void testAwaitPreservesFatalConsumerError() throws Throwable { @Test(timeout = 1000) public void testAwaitThrowsWhenConsumerThrows() throws Throwable { ExecutorService executor = Executors.newFixedThreadPool(2); + CountDownLatch release = new CountDownLatch(1); try { final String msg = "Injected exception for test"; Consumers consumers = new Consumers<>(executor, v -> { + awaitGate(release); throw new RuntimeException(msg); }); consumers.start("test"); consumers.provide(1); + release.countDown(); try { consumers.await(); @@ -108,6 +111,7 @@ public void testAwaitThrowsWhenConsumerThrows() throws Throwable { root.getMessage().contains(msg)); } } finally { + release.countDown(); executor.shutdownNow(); } }