anph
/
hugegraph-sync
forked from hugegraph/hugegraph-sync
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix rocksdb instance not close properly (#1264) 

* The open-pool opened session is always exist
* Fix other disk sessions not close

Change-Id: I30504565155848659a818fbc1b57999547f0502f
This commit is contained in:
Linary 2020-11-17 15:34:50 +08:00 committed by GitHub
parent 836956df30
commit 5a68f50b9e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 95 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendSessionPool.java
View File

@ -94,7 +94,7 @@ public abstract class BackendSessionPool {
session.update();
}
public Pair<Integer, Integer> closeSession() {
private Pair<Integer, Integer> 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<Integer, Integer> 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() {

40
hugegraph-core/src/main/java/com/baidu/hugegraph/task/TaskManager.java
View File

@ -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<Thread, Integer> threadsTimes = new ConcurrentHashMap<>();
final List<Callable<Void>> tasks = new ArrayList<>();
final int totalThreads = selfIsTaskWorker ? THREADS - 1 : THREADS;
final Callable<Void> 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) {

39
hugegraph-core/src/main/java/com/baidu/hugegraph/util/Consumers.java
View File

@ -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<V> {
}
}
public static void executeOncePerThread(ExecutorService executor,
int totalThreads,
Runnable callback)
throws InterruptedException {
// Ensure callback execute at least once for every thread
final Map<Thread, Integer> threadsTimes = new ConcurrentHashMap<>();
final List<Callable<Void>> tasks = new ArrayList<>();
final Callable<Void> 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;

3
hugegraph-palo/src/main/java/com/baidu/hugegraph/backend/store/palo/PaloSessions.java
View File

@ -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 {

49
hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStore.java
View File

@ -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<Session> {
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<Session> {
waitOpenFinish(futures, openPool);
}
private static void waitOpenFinish(List<Future<?>> futures,
ExecutorService openPool) {
private void waitOpenFinish(List<Future<?>> futures,
ExecutorService openPool) {
for (Future<?> future : futures) {
try {
future.get();
@ -217,6 +219,22 @@ public abstract class RocksDBStore extends AbstractBackendStore<Session> {
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<Session> {
LOG.debug("Store close: {}", this.store);
this.checkOpened();
this.sessions.close();
this.closeSessions();
}
@Override
@ -643,6 +661,25 @@ public abstract class RocksDBStore extends AbstractBackendStore<Session> {
}
}
private final void useSessions() {
for (RocksDBSessions sessions : this.sessions()) {
sessions.useSession();
}
}
private final void closeSessions() {
Iterator<Map.Entry<String, RocksDBSessions>> iter = dbs.entrySet()
.iterator();
while (iter.hasNext()) {
Map.Entry<String, RocksDBSessions> 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<Session> {
return list;
}
private final Collection<RocksDBSessions> sessions() {
return dbs.values();
}
private final void parseTableDiskMapping(Map<String, String> disks,
String dataPath) {
this.tableDiskMapping.clear();