diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendSessionPool.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendSessionPool.java index fdc1a0ac0..dcf1bd593 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendSessionPool.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendSessionPool.java @@ -94,7 +94,7 @@ public abstract class BackendSessionPool { session.update(); } - public Pair closeSession() { + private Pair closeSession() { int sessionCount = this.sessionCount.get(); if (sessionCount <= 0) { assert sessionCount == 0 : sessionCount; @@ -104,7 +104,7 @@ public abstract class BackendSessionPool { assert sessionCount > 0 : sessionCount; BackendSession session = this.threadLocalSession.get(); if (session == null) { - LOG.warn("Current session has ever been closed: {}", this); + LOG.debug("Current session has ever been closed: {}", this); return Pair.of(sessionCount, -1); } @@ -134,7 +134,7 @@ public abstract class BackendSessionPool { } } - public void close() { + public boolean close() { Pair result = Pair.of(-1, -1); try { result = this.closeSession(); @@ -146,6 +146,7 @@ public abstract class BackendSessionPool { LOG.debug("Now(after close({})) session count is: {}, " + "current session reference is: {}", this, result.getLeft(), result.getRight()); + return result.getLeft() == 0; } public boolean closed() { diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/task/TaskManager.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/task/TaskManager.java index 6eea70542..84322941a 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/task/TaskManager.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/task/TaskManager.java @@ -19,8 +19,6 @@ package com.baidu.hugegraph.task; -import java.util.ArrayList; -import java.util.List; import java.util.Map; import java.util.Queue; import java.util.concurrent.Callable; @@ -35,6 +33,7 @@ import org.slf4j.Logger; import com.baidu.hugegraph.HugeException; import com.baidu.hugegraph.HugeGraphParams; +import com.baidu.hugegraph.util.Consumers; import com.baidu.hugegraph.util.E; import com.baidu.hugegraph.util.ExecutorUtil; import com.baidu.hugegraph.util.LockUtil; @@ -89,7 +88,7 @@ public final class TaskManager { E.checkArgumentNotNull(graph, "The graph can't be null"); TaskScheduler scheduler = new StandardTaskScheduler(graph, - this.taskExecutor,this.taskDbExecutor, + this.taskExecutor, this.taskDbExecutor, this.serverInfoDbExecutor); this.schedulers.put(graph, scheduler); } @@ -127,42 +126,14 @@ public final class TaskManager { private void closeTaskTx(HugeGraphParams graph) { final boolean selfIsTaskWorker = Thread.currentThread().getName() .startsWith(TASK_WORKER_PREFIX); - final Map threadsTimes = new ConcurrentHashMap<>(); - final List> tasks = new ArrayList<>(); final int totalThreads = selfIsTaskWorker ? THREADS - 1 : THREADS; - - final Callable closeTx = () -> { - Thread current = Thread.currentThread(); - threadsTimes.putIfAbsent(current, 0); - int times = threadsTimes.get(current); - if (times == 0) { - // Do close-tx for current thread - graph.closeTx(); - // Let other threads run - Thread.yield(); - } else { - assert times < totalThreads; - assert threadsTimes.size() < totalThreads; - E.checkState(tasks.size() == totalThreads, - "Bad tasks size: %s", tasks.size()); - // Let another thread run and wait for it - this.taskExecutor.submit(tasks.get(0)).get(); - } - threadsTimes.put(current, ++times); - return null; - }; - - // NOTE: expect each task thread to perform a close operation - for (int i = 0; i < totalThreads; i++) { - tasks.add(closeTx); - } - try { if (selfIsTaskWorker) { // Call closeTx directly if myself is task thread(ignore others) - closeTx.call(); + graph.closeTx(); } else { - this.taskExecutor.invokeAll(tasks); + Consumers.executeOncePerThread(this.taskExecutor, totalThreads, + graph::closeTx); } } catch (Exception e) { throw new HugeException("Exception when closing task tx", e); @@ -177,7 +148,6 @@ public final class TaskManager { Thread.yield(); return null; }; - try { this.schedulerExecutor.submit(closeTx).get(); } catch (Exception e) { diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/util/Consumers.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/util/Consumers.java index aedb547fb..1549ea0e3 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/util/Consumers.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/util/Consumers.java @@ -19,8 +19,13 @@ package com.baidu.hugegraph.util; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.ThreadPoolExecutor; @@ -185,6 +190,40 @@ public class Consumers { } } + public static void executeOncePerThread(ExecutorService executor, + int totalThreads, + Runnable callback) + throws InterruptedException { + // Ensure callback execute at least once for every thread + final Map threadsTimes = new ConcurrentHashMap<>(); + final List> tasks = new ArrayList<>(); + final Callable task = () -> { + Thread current = Thread.currentThread(); + threadsTimes.putIfAbsent(current, 0); + int times = threadsTimes.get(current); + if (times == 0) { + callback.run(); + // Let other threads run + Thread.yield(); + } else { + assert times < totalThreads; + assert threadsTimes.size() < totalThreads; + E.checkState(tasks.size() == totalThreads, + "Bad tasks size: %s", tasks.size()); + // Let another thread run and wait for it + executor.submit(tasks.get(0)).get(); + } + threadsTimes.put(current, ++times); + return null; + }; + + // NOTE: expect each task thread to perform a close operation + for (int i = 0; i < totalThreads; i++) { + tasks.add(task); + } + executor.invokeAll(tasks); + } + public static ExecutorService newThreadPool(String prefix, int workers) { if (workers == 0) { return null; diff --git a/hugegraph-palo/src/main/java/com/baidu/hugegraph/backend/store/palo/PaloSessions.java b/hugegraph-palo/src/main/java/com/baidu/hugegraph/backend/store/palo/PaloSessions.java index 7aa8b7e71..d349bbb51 100644 --- a/hugegraph-palo/src/main/java/com/baidu/hugegraph/backend/store/palo/PaloSessions.java +++ b/hugegraph-palo/src/main/java/com/baidu/hugegraph/backend/store/palo/PaloSessions.java @@ -107,10 +107,11 @@ public class PaloSessions extends MysqlSessions { } @Override - public void close() { + public boolean close() { this.loadTask.join(); this.timer.cancel(); super.close(); + return true; } public final class Session extends MysqlSessions.Session { diff --git a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStore.java b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStore.java index cba5a2cab..05097b40a 100644 --- a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStore.java +++ b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStore.java @@ -24,6 +24,7 @@ import java.io.IOException; import java.nio.file.Paths; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; @@ -60,6 +61,7 @@ import com.baidu.hugegraph.backend.store.rocksdb.RocksDBSessions.Session; import com.baidu.hugegraph.config.HugeConfig; import com.baidu.hugegraph.exception.ConnectionException; import com.baidu.hugegraph.type.HugeType; +import com.baidu.hugegraph.util.Consumers; import com.baidu.hugegraph.util.E; import com.baidu.hugegraph.util.ExecutorUtil; import com.baidu.hugegraph.util.GZipUtil; @@ -172,7 +174,7 @@ public abstract class RocksDBStore extends AbstractBackendStore { if (this.sessions != null && !this.sessions.closed()) { LOG.debug("Store {} has been opened before", this.store); - this.sessions.useSession(); + this.useSessions(); return; } @@ -205,8 +207,8 @@ public abstract class RocksDBStore extends AbstractBackendStore { waitOpenFinish(futures, openPool); } - private static void waitOpenFinish(List> futures, - ExecutorService openPool) { + private void waitOpenFinish(List> futures, + ExecutorService openPool) { for (Future future : futures) { try { future.get(); @@ -217,6 +219,22 @@ public abstract class RocksDBStore extends AbstractBackendStore { if (openPool.isShutdown()) { return; } + + /* + * Transfer the session holder from db-open thread to main thread, + * otherwise once the db-open thread pool is closed, we can no longer + * close the session created by it, which will cause the rocksdb + * instance fail to close + */ + this.useSessions(); + try { + Consumers.executeOncePerThread(openPool, OPEN_POOL_THREADS, + this::closeSessions); + } catch (InterruptedException e) { + throw new BackendException("Failed to close session opened by " + + "open-pool"); + } + boolean terminated = false; openPool.shutdown(); try { @@ -339,7 +357,7 @@ public abstract class RocksDBStore extends AbstractBackendStore { LOG.debug("Store close: {}", this.store); this.checkOpened(); - this.sessions.close(); + this.closeSessions(); } @Override @@ -643,6 +661,25 @@ public abstract class RocksDBStore extends AbstractBackendStore { } } + private final void useSessions() { + for (RocksDBSessions sessions : this.sessions()) { + sessions.useSession(); + } + } + + private final void closeSessions() { + Iterator> iter = dbs.entrySet() + .iterator(); + while (iter.hasNext()) { + Map.Entry entry = iter.next(); + RocksDBSessions sessions = entry.getValue(); + boolean closed = sessions.close(); + if (closed) { + iter.remove(); + } + } + } + private Session findMatchedSession(File snapshotFile) { String fileName = snapshotFile.getName(); for (Session session : this.session()) { @@ -670,6 +707,10 @@ public abstract class RocksDBStore extends AbstractBackendStore { return list; } + private final Collection sessions() { + return dbs.values(); + } + private final void parseTableDiskMapping(Map disks, String dataPath) { this.tableDiskMapping.clear();