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

refact: update the code to 2023.6.19 

Change-Id: Ib861b47cab873322d9427d6a1f654f5cae92ddb6

GraphPlatform-1749 修复显示的版本号

Change-Id: I09b072b802d60bf1820f5a6dd18e2570eee4835b

GraphPlatform-1657 修改grpc消息大小参数设置

Change-Id: Ifb51b841c5be7690407cc90b34a72d4b2bdea360

GraphPlatform-1595 调整ci及减少锁粒度

Change-Id: I496c814cac580a51125e382b538eaf57445121dc

FIN GraphPlatform-2141 增加按表查询接口

Change-Id: Iee2d3503d2a7fd4815ecf7dec828afa28a24e46a

GraphPlatform-2147 迭代完成及时释放rocksdb迭代器，部分并发控制改进

Change-Id: I69f3985fd7a29b718f10eb6718488d63e6a1a077

GraphPlatform-2161 PD leader变更，客户端重连

Change-Id: Ibfeb837e94ecc9dfc19e6fdc85b421df725a6cb7

FIN GraphPlatform-2147 处理并发

Change-Id: I7a41969a8689eb216d2a7537a7a2f94846427c0e

FIN GraphPlatform-2140 修改并发控制

Change-Id: I89cbee229c111716b092a6df903aa692be03b9bb

FIN GraphPlatform-2140 并发控制改进

Change-Id: Ie9c7755b330d2df653c7e93b5592a514bb7aa099

GraphPlatform-2119 修改线程默认配置

Change-Id: Ia6107168b381562a767e34443328391e5da1e1a0
This commit is contained in:
imbajin 2023-06-20 19:04:20 +08:00
parent 9ccbeb7de7
commit 1252f5097c
19 changed files with 274 additions and 361 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hg-store-client/src/main/java/org/apache/hugegraph/store/HgKvStore.java
View File

@ -97,6 +97,8 @@ public interface HgKvStore {
HgKvIterator<HgKvEntry> scanIterator(ScanStreamReq.Builder scanReqBuilder);
long count(String table);
boolean truncate();
default boolean existsTable(String table) {

11
hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxSessionProxy.java
View File

@ -505,6 +505,17 @@ class NodeTxSessionProxy implements HgStoreSession {
return this.toHgKvIteratorProxy(iterators, scanReqBuilder.getLimit());
}
@Override
public long count(String table) {
return this.toNodeTkvList(table)
.parallelStream()
.map(
e -> this.getStoreNode(e.getNodeId()).openSession(this.graphName)
.count(e.getTable())
)
.collect(Collectors.summingLong(l -> l));
}
@Override
public List<HgKvIterator<HgKvEntry>> scanBatch(HgScanQuery scanQuery) {
HgAssert.isArgumentNotNull(scanQuery, "scanQuery");

85
hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/AbstractGrpcClient.java
View File

@ -19,10 +19,13 @@ package org.apache.hugegraph.store.client.grpc;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.IntStream;
import org.apache.hugegraph.store.client.util.ExecutorPool;
import org.apache.hugegraph.store.client.util.HgStoreClientConfig;
import org.apache.hugegraph.store.term.HgPair;
@ -30,22 +33,27 @@ import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.AbstractAsyncStub;
import io.grpc.stub.AbstractBlockingStub;
import io.grpc.stub.AbstractStub;
/**
* @date 2023/3/28
**/
public abstract class AbstractGrpcClient {
private static final Map<String, ManagedChannel[]> channels = new ConcurrentHashMap<>();
private static final int n = 5;
private static final int concurrency = 1 << n;
private static final AtomicLong counter = new AtomicLong(0);
private static final long limit = Long.MAX_VALUE >> 1;
private static final HgStoreClientConfig config = HgStoreClientConfig.of();
private final Map<String, HgPair<ManagedChannel, AbstractBlockingStub>[]> blockingStubs =
new ConcurrentHashMap<>();
private final Map<String, HgPair<ManagedChannel, AbstractAsyncStub>[]> asyncStubs =
private static Map<String, ManagedChannel[]> channels = new ConcurrentHashMap<>();
private static int n = 5;
private static int concurrency = 1 << n;
private static AtomicLong counter = new AtomicLong(0);
private static long limit = Long.MAX_VALUE >> 1;
private Map<String, HgPair<ManagedChannel, AbstractBlockingStub>[]> blockingStubs =
new ConcurrentHashMap<>();
private Map<String, HgPair<ManagedChannel, AbstractAsyncStub>[]> asyncStubs = new ConcurrentHashMap<>();
private static HgStoreClientConfig config = HgStoreClientConfig.of();
private ThreadPoolExecutor executor;
{
executor = ExecutorPool.createExecutor("common", 60, concurrency, concurrency);
}
public AbstractGrpcClient() {
@ -56,10 +64,26 @@ public abstract class AbstractGrpcClient {
if ((tc = channels.get(target)) == null) {
synchronized (channels) {
if ((tc = channels.get(target)) == null) {
ManagedChannel[] value = new ManagedChannel[concurrency];
IntStream.range(0, concurrency).parallel()
.forEach(i -> value[i] = getManagedChannel(target));
channels.put(target, tc = value);
try {
ManagedChannel[] value = new ManagedChannel[concurrency];
CountDownLatch latch = new CountDownLatch(concurrency);
for (int i = 0; i < concurrency; i++) {
int fi = i;
executor.execute(() -> {
try{
value[fi] = getManagedChannel(target);
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
latch.countDown();
}
});
}
latch.await();
channels.put(target, tc = value);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
}
@ -81,25 +105,27 @@ public abstract class AbstractGrpcClient {
pairs = blockingStubs.get(target);
if (pairs == null) {
HgPair<ManagedChannel, AbstractBlockingStub>[] value = new HgPair[concurrency];
IntStream.range(0, concurrency).parallel().forEach(i -> {
IntStream.range(0, concurrency).forEach(i -> {
ManagedChannel channel = channels[index];
AbstractBlockingStub stub = getBlockingStub(channel);
stub.withMaxInboundMessageSize(config.getGrpcMaxInboundMessageSize())
.withMaxOutboundMessageSize(config.getGrpcMaxOutboundMessageSize());
value[i] = new HgPair<>(channel, stub);
// log.info("create channel for {}",target);
});
blockingStubs.put(target, value);
AbstractBlockingStub stub = value[index].getValue();
return (AbstractBlockingStub) stub.withDeadlineAfter(
config.getGrpcTimeoutSeconds(),
TimeUnit.SECONDS);
return (AbstractBlockingStub) setBlockingStubOption(stub);
}
}
}
return (AbstractBlockingStub) pairs[index].getValue()
.withDeadlineAfter(config.getGrpcTimeoutSeconds(),
TimeUnit.SECONDS);
return (AbstractBlockingStub) setBlockingStubOption(pairs[index].getValue());
}
private AbstractStub setBlockingStubOption(AbstractBlockingStub stub) {
return stub.withDeadlineAfter(config.getGrpcTimeoutSeconds(),TimeUnit.SECONDS)
.withMaxInboundMessageSize(
config.getGrpcMaxInboundMessageSize())
.withMaxOutboundMessageSize(
config.getGrpcMaxOutboundMessageSize());
}
public AbstractAsyncStub getAsyncStub(ManagedChannel channel) {
@ -122,21 +148,28 @@ public abstract class AbstractGrpcClient {
IntStream.range(0, concurrency).parallel().forEach(i -> {
ManagedChannel channel = channels[index];
AbstractAsyncStub stub = getAsyncStub(channel);
stub.withMaxInboundMessageSize(config.getGrpcMaxInboundMessageSize())
.withMaxOutboundMessageSize(config.getGrpcMaxOutboundMessageSize());
// stub.withMaxInboundMessageSize(config.getGrpcMaxInboundMessageSize())
// .withMaxOutboundMessageSize(config.getGrpcMaxOutboundMessageSize());
value[i] = new HgPair<>(channel, stub);
// log.info("create channel for {}",target);
});
asyncStubs.put(target, value);
AbstractAsyncStub stub = value[index].getValue();
AbstractAsyncStub stub =
(AbstractAsyncStub) setStubOption(value[index].getValue());
return stub;
}
}
}
return pairs[index].getValue();
return (AbstractAsyncStub) setStubOption(pairs[index].getValue());
}
private AbstractStub setStubOption(AbstractStub value) {
return value.withMaxInboundMessageSize(
config.getGrpcMaxInboundMessageSize())
.withMaxOutboundMessageSize(
config.getGrpcMaxOutboundMessageSize());
}
private ManagedChannel getManagedChannel(String target) {
return ManagedChannelBuilder.forTarget(target).usePlaintext().build();

5
hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/GrpcStoreNodeSessionImpl.java
View File

@ -368,6 +368,11 @@ class GrpcStoreNodeSessionImpl implements HgStoreNodeSession {
return GrpcKvIteratorImpl.of(this, scanner);
}
@Override
public long count(String table) {
return this.storeSessionClient.count(this, table);
}
@Override
public HgKvIterator<HgKvEntry> scanIterator(String table, byte[] query) {

13
hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/GrpcStoreSessionClient.java
View File

@ -18,6 +18,7 @@
package org.apache.hugegraph.store.client.grpc;
import java.util.List;
import java.util.concurrent.TimeUnit;
import javax.annotation.concurrent.ThreadSafe;
@ -37,6 +38,7 @@ import org.apache.hugegraph.store.grpc.session.HgStoreSessionGrpc;
import org.apache.hugegraph.store.grpc.session.HgStoreSessionGrpc.HgStoreSessionBlockingStub;
import org.apache.hugegraph.store.grpc.session.TableReq;
import io.grpc.Deadline;
import io.grpc.ManagedChannel;
import lombok.extern.slf4j.Slf4j;
@ -137,6 +139,17 @@ class GrpcStoreSessionClient extends AbstractGrpcClient {
.build()
);
}
public long count(HgStoreNodeSession nodeSession, String table) {
Agg agg = this.getBlockingStub(nodeSession).withDeadline(Deadline.after(24, TimeUnit.HOURS))
.count(ScanStreamReq.newBuilder()
.setHeader(getHeader(nodeSession))
.setTable(table)
.setMethod(ScanMethod.ALL)
.build()
);
return agg.getCount();
}
}

97
hg-store-client/src/main/java/org/apache/hugegraph/store/client/util/ExecutorPool.java
View File

@ -25,20 +25,12 @@ import java.util.concurrent.atomic.AtomicInteger;
import lombok.extern.slf4j.Slf4j;
/**
* 2021/11/22
*/
@Slf4j
public final class ExecutorPool {
public static ThreadFactory newThreadFactory(String namePrefix, int priority) {
HgAssert.isArgumentNotNull(namePrefix, "namePrefix");
return new HgThreadFactory(namePrefix, priority);
}
public static ThreadFactory newThreadFactory(String namePrefix) {
HgAssert.isArgumentNotNull(namePrefix, "namePrefix");
return new HgDefaultThreadFactory(namePrefix);
return new DefaultThreadFactory(namePrefix);
}
public static ThreadPoolExecutor createExecutor(String name, long keepAliveTime,
@ -50,97 +42,20 @@ public final class ExecutorPool {
);
}
//
// private final static ExecutorService executor = new ThreadPoolExecutor(0, Integer.MAX_VALUE,
// 10L, TimeUnit.SECONDS,
// new
// LinkedBlockingQueue<>(),
// new
// HgDefaultThreadFactory
// ("store-common"));
// private final static ExecutorService grpcExecutor = createExecutor("store-grpc", 60L, 600,
// 10240);
//// public final static ExecutorService scannerExecutor = createExecutor("scanner", 10l,
// 200, 1024);
//
// static {
// Thread hook = new Thread(() -> {
// try {
// executor.shutdown();
// executor.awaitTermination(30, TimeUnit.SECONDS);
// return;
// } catch (InterruptedException e) {
// log.error("failed to await executorService.shutdown()", e);
// }
// executor.shutdownNow().forEach(r -> log.error("not run task:" + r.toString()));
// try {
// executor.awaitTermination(30, TimeUnit.SECONDS);
// } catch (InterruptedException e) {
// log.error("failed to await executorService.shutdownNow()", e);
// throw HgStoreClientException.of(e);
// }
// }
// );
// Runtime.getRuntime().addShutdownHook(hook);
//}
//
//// public static void execute(Runnable command) {
//// isArgumentNotNull(command, "command");
//// executor.execute(command);
//// }
////
// public static ExecutorService getGrpcCommonExecutor() {
// return grpcExecutor;
// }
//
public static class DefaultThreadFactory implements ThreadFactory {
/**
* The default thread factory
*/
static class HgThreadFactory implements ThreadFactory {
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
private final int priority;
HgThreadFactory(String namePrefix, int priority) {
this.namePrefix = namePrefix;
this.priority = priority;
}
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(null, r, namePrefix + "-" + threadNumber.getAndIncrement(), 0);
if (t.isDaemon()) {
t.setDaemon(false);
}
if (t.getPriority() != priority) {
t.setPriority(priority);
}
return t;
}
}
/**
* The default thread factory, which added threadNamePrefix in construction method.
*/
static class HgDefaultThreadFactory implements ThreadFactory {
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
HgDefaultThreadFactory(String threadNamePrefix) {
this.namePrefix = threadNamePrefix + "-" + poolNumber.getAndIncrement() + "-thread-";
public DefaultThreadFactory(String threadNamePrefix) {
this.namePrefix = threadNamePrefix + "-";
}
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(null, r, namePrefix + threadNumber.getAndIncrement(), 0);
if (t.isDaemon()) {
t.setDaemon(false);
}
if (t.getPriority() != Thread.NORM_PRIORITY) {
t.setPriority(Thread.NORM_PRIORITY);
}
t.setDaemon(true);
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}

10
hg-store-core/src/main/java/org/apache/hugegraph/store/HgStoreEngine.java
View File

@ -80,9 +80,7 @@ public class HgStoreEngine implements Lifecycle<HgStoreEngineOptions>, HgStoreSt
private HgMetricService metricService;
private DataMover dataMover;
private HgStoreEngine() {
}
private static ConcurrentHashMap<Integer, Object> engineLocks = new ConcurrentHashMap<>();
public static HgStoreEngine getInstance() {
return instance;
@ -290,7 +288,8 @@ public class HgStoreEngine implements Lifecycle<HgStoreEngineOptions>, HgStoreSt
Configuration conf) {
PartitionEngine engine;
if ((engine = partitionEngines.get(groupId)) == null) {
synchronized (this) {
engineLocks.computeIfAbsent(groupId, k -> new Object());
synchronized (engineLocks.get(groupId)) {
// 分区分裂时特殊情况(集群中图分区数量不一样)，会导致分裂的分区，可能不在本机器上.
if (conf != null) {
var list = conf.listPeers();
@ -566,7 +565,8 @@ public class HgStoreEngine implements Lifecycle<HgStoreEngineOptions>, HgStoreSt
RaftClosure closure) {
PartitionEngine engine = getPartitionEngine(graphName, partId);
if (engine == null) {
synchronized (this) {
engineLocks.computeIfAbsent(partId, k -> new Object());
synchronized (engineLocks.get(partId)) {
engine = getPartitionEngine(graphName, partId);
if (engine == null) {
Partition partition = partitionManager.findPartition(graphName, partId);

27
hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandler.java
View File

@ -22,8 +22,6 @@ import java.util.Map;
import java.util.function.Consumer;
import java.util.function.Supplier;
import javax.annotation.concurrent.NotThreadSafe;
import org.apache.hugegraph.pd.grpc.pulse.CleanType;
import org.apache.hugegraph.rocksdb.access.ScanIterator;
import org.apache.hugegraph.store.grpc.Graphpb;
@ -198,28 +196,5 @@ public interface BusinessHandler extends DBSessionBuilder {
void destroyGraphDB(String graphName, int partId) throws HgStoreException;
@NotThreadSafe
interface TxBuilder {
TxBuilder put(int code, String table, byte[] key, byte[] value) throws HgStoreException;
TxBuilder del(int code, String table, byte[] key) throws HgStoreException;
TxBuilder delSingle(int code, String table, byte[] key) throws HgStoreException;
TxBuilder delPrefix(int code, String table, byte[] prefix) throws HgStoreException;
TxBuilder delRange(int code, String table, byte[] start, byte[] end) throws
HgStoreException;
TxBuilder merge(int code, String table, byte[] key, byte[] value) throws HgStoreException;
Tx build();
}
interface Tx {
void commit() throws HgStoreException;
void rollback() throws HgStoreException;
}
long count(String graphName, String table);
}

136
hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java
View File

@ -33,8 +33,6 @@ import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import javax.annotation.concurrent.NotThreadSafe;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hugegraph.config.HugeConfig;
@ -802,117 +800,35 @@ public class BusinessHandlerImpl implements BusinessHandler {
keyCreator.clearCache(partId);
}
@NotThreadSafe
private class TxBuilderImpl implements TxBuilder {
private final String graph;
private final int partId;
private final RocksDBSession dbSession;
private final SessionOperator op;
private TxBuilderImpl(String graph, int partId, RocksDBSession dbSession) {
this.graph = graph;
this.partId = partId;
this.dbSession = dbSession;
this.op = this.dbSession.sessionOp();
this.op.prepare();
}
@Override
public TxBuilder put(int code, String table, byte[] key, byte[] value) throws
HgStoreException {
try {
byte[] targetKey = keyCreator.getKey(this.partId, graph, code, key);
this.op.put(table, targetKey, value);
} catch (DBStoreException e) {
throw new HgStoreException(HgStoreException.EC_RKDB_DOPUT_FAIL, e.toString());
}
return this;
}
@Override
public TxBuilder del(int code, String table, byte[] key) throws HgStoreException {
try {
byte[] targetKey = keyCreator.getKey(this.partId, graph, code, key);
this.op.delete(table, targetKey);
} catch (DBStoreException e) {
throw new HgStoreException(HgStoreException.EC_RKDB_DODEL_FAIL, e.toString());
}
return this;
}
@Override
public TxBuilder delSingle(int code, String table, byte[] key) throws HgStoreException {
try {
byte[] targetKey = keyCreator.getKey(this.partId, graph, code, key);
op.deleteSingle(table, targetKey);
} catch (DBStoreException e) {
throw new HgStoreException(HgStoreException.EC_RDKDB_DOSINGLEDEL_FAIL,
e.toString());
}
return this;
}
@Override
public TxBuilder delPrefix(int code, String table, byte[] prefix) throws HgStoreException {
try {
this.op.deletePrefix(table, keyCreator.getPrefixKey(this.partId, graph, prefix));
} catch (DBStoreException e) {
throw new HgStoreException(HgStoreException.EC_RKDB_DODELPREFIX_FAIL, e.toString());
}
return this;
}
@Override
public TxBuilder delRange(int code, String table, byte[] start, byte[] end) throws
HgStoreException {
try {
this.op.deleteRange(table, keyCreator.getStartKey(this.partId, graph, start),
keyCreator.getEndKey(this.partId, graph, end));
} catch (DBStoreException e) {
throw new HgStoreException(HgStoreException.EC_RKDB_DODELRANGE_FAIL, e.toString());
}
return this;
}
@Override
public TxBuilder merge(int code, String table, byte[] key, byte[] value) throws
HgStoreException {
try {
byte[] targetKey = keyCreator.getKey(this.partId, graph, code, key);
op.merge(table, targetKey, value);
} catch (DBStoreException e) {
throw new HgStoreException(HgStoreException.EC_RKDB_DOMERGE_FAIL, e.toString());
}
return this;
}
@Override
public Tx build() {
return new Tx() {
@Override
public void commit() throws HgStoreException {
op.commit(); // commit发生异常后，必须调用rollback，否则造成锁未释放
dbSession.close();
@Override
public long count(String graph, String table) {
List<Integer> ids = this.getLeaderPartitionIds(graph);
Long all = ids.parallelStream().map((id) -> {
InnerKeyFilter it = null;
try (RocksDBSession dbSession = getSession(graph, table, id)) {
long count = 0;
SessionOperator op = dbSession.sessionOp();
it = new InnerKeyFilter(op.scan(table,
keyCreator.getStartKey(id, graph),
keyCreator.getEndKey(id, graph),
ScanIterator.Trait.SCAN_LT_END));
while (it.hasNext()) {
it.next();
count++;
}
@Override
public void rollback() throws HgStoreException {
return count;
} catch (Exception e) {
throw e;
} finally {
if (it != null) {
try {
op.rollback();
} finally {
dbSession.close();
it.close();
} catch (Exception e) {
}
}
};
}
}
}).collect(Collectors.summingLong(l -> l));
return all;
}
}

110
hg-store-core/src/main/java/org/apache/hugegraph/store/pd/DefaultPdProvider.java
View File

@ -21,8 +21,6 @@ package org.apache.hugegraph.store.pd;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Consumer;
import org.apache.hugegraph.pd.client.PDClient;
@ -54,12 +52,14 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public class DefaultPdProvider implements PdProvider {
private static final Logger LOG = Log.logger(DefaultPdProvider.class);
private static final ConcurrentMap<String, Boolean> quotas =
new ConcurrentHashMap<>();
private final PDClient pdClient;
private final String pdServerAddress;
private Consumer<Throwable> hbOnError = null;
private List<PartitionInstructionListener> partitionCommandListeners =
Collections.synchronizedList(new ArrayList());
Collections.synchronizedList(new ArrayList<>());
private final PDPulse pulseClient;
private PDPulse.Notifier<PartitionHeartbeatRequest.Builder> pdPulse;
private GraphManager graphManager = null;
PDClient.PDEventListener listener = new PDClient.PDEventListener() {
@ -70,6 +70,11 @@ public class DefaultPdProvider implements PdProvider {
log.info("store raft group changed!, {}", event);
pdClient.invalidStoreCache(event.getNodeId());
HgStoreEngine.getInstance().rebuildRaftGroup(event.getNodeId());
} else if (event.getEventType() == NodeEvent.EventType.NODE_PD_LEADER_CHANGE) {
log.info("pd leader changed!, {}. restart heart beat", event);
if (pulseClient.resetStub(event.getGraph(), pdPulse)) {
startHeartbeatStream(hbOnError);
}
}
}
@ -94,6 +99,8 @@ public class DefaultPdProvider implements PdProvider {
this.pdClient.addEventListener(listener);
this.pdServerAddress = pdAddress;
partitionCommandListeners = Collections.synchronizedList(new ArrayList());
log.info("pulse client connect to {}", pdClient.getLeaderIp());
this.pulseClient = new PDPulseImpl(pdClient.getLeaderIp());
}
@Override
@ -244,61 +251,77 @@ public class DefaultPdProvider implements PdProvider {
*/
@Override
public boolean startHeartbeatStream(Consumer<Throwable> onError) {
PDPulse pulse = pdClient.getPulseClient();
pdPulse = pulse.connectPartition(new PDPulse.Listener<PartitionHeartbeatResponse>() {
this.hbOnError = onError;
pdPulse = pulseClient.connectPartition(new PDPulse.Listener<>() {
@Override
public void onNotice(PulseServerNotice<PartitionHeartbeatResponse> response) {
PartitionHeartbeatResponse instruction = response.getContent();
LOG.debug("Partition heartbeat receive instruction: {}", instruction);
Partition partition = new Partition(instruction.getPartition());
public void onNotice(PulseServerNotice<PulseResponse> response) {
PulseResponse content = response.getContent();
// 消息消费应答，能够正确消费消息，调用accept返回状态码，否则不要调用accept
Consumer<Integer> consumer = new Consumer<>() {
@Override
public void accept(Integer integer) {
LOG.debug("Partition heartbeat accept instruction: {}", instruction);
// LOG.info("accept notice id : {}, ts:{}", response.getNoticeId(),
// System.currentTimeMillis());
// http2 并发问题，需要加锁
// synchronized (pdPulse) {
response.ack();
// }
}
Consumer<Integer> consumer = integer -> {
LOG.debug("Partition heartbeat accept instruction: {}", content);
// LOG.info("accept notice id : {}, ts:{}", response.getNoticeId(), System
// .currentTimeMillis());
// http2 并发问题，需要加锁
// synchronized (pdPulse) {
response.ack();
// }
};
if (content.hasInstructionResponse()) {
var pdInstruction = content.getInstructionResponse();
consumer.accept(0);
// 当前的链接变成了follower，重新链接
if (pdInstruction.getInstructionType() ==
PdInstructionType.CHANGE_TO_FOLLOWER) {
onCompleted();
log.info("got pulse instruction, change leader to {}",
pdInstruction.getLeaderIp());
if (pulseClient.resetStub(pdInstruction.getLeaderIp(), pdPulse)) {
startHeartbeatStream(hbOnError);
}
}
return;
}
PartitionHeartbeatResponse instruct = content.getPartitionHeartbeatResponse();
LOG.debug("Partition heartbeat receive instruction: {}", instruct);
Partition partition = new Partition(instruct.getPartition());
for (PartitionInstructionListener event : partitionCommandListeners) {
if (instruction.hasChangeShard()) {
event.onChangeShard(instruction.getId(), partition,
instruction.getChangeShard(), consumer);
if (instruct.hasChangeShard()) {
event.onChangeShard(instruct.getId(), partition, instruct.getChangeShard(),
consumer);
}
if (instruction.hasSplitPartition()) {
event.onSplitPartition(instruction.getId(), partition,
instruction.getSplitPartition(), consumer);
if (instruct.hasSplitPartition()) {
event.onSplitPartition(instruct.getId(), partition,
instruct.getSplitPartition(), consumer);
}
if (instruction.hasTransferLeader()) {
event.onTransferLeader(instruction.getId(), partition,
instruction.getTransferLeader(), consumer);
if (instruct.hasTransferLeader()) {
event.onTransferLeader(instruct.getId(), partition,
instruct.getTransferLeader(), consumer);
}
if (instruction.hasDbCompaction()) {
event.onDbCompaction(instruction.getId(), partition,
instruction.getDbCompaction(), consumer);
if (instruct.hasDbCompaction()) {
event.onDbCompaction(instruct.getId(), partition,
instruct.getDbCompaction(), consumer);
}
if (instruction.hasMovePartition()) {
event.onMovePartition(instruction.getId(), partition,
instruction.getMovePartition(), consumer);
if (instruct.hasMovePartition()) {
event.onMovePartition(instruct.getId(), partition,
instruct.getMovePartition(), consumer);
}
if (instruction.hasCleanPartition()) {
event.onCleanPartition(instruction.getId(), partition,
instruction.getCleanPartition(),
if (instruct.hasCleanPartition()) {
event.onCleanPartition(instruct.getId(), partition,
instruct.getCleanPartition(),
consumer);
}
if (instruction.hasKeyRange()) {
event.onPartitionKeyRangeChanged(instruction.getId(), partition,
instruction.getKeyRange(),
if (instruct.hasKeyRange()) {
event.onPartitionKeyRangeChanged(instruct.getId(), partition,
instruct.getKeyRange(),
consumer);
}
}
@ -307,6 +330,7 @@ public class DefaultPdProvider implements PdProvider {
@Override
public void onError(Throwable throwable) {
LOG.error("Partition heartbeat stream error. {}", throwable);
pulseClient.resetStub(pdClient.getLeaderIp(), pdPulse);
onError.accept(throwable);
}

2
hg-store-core/src/test/resources/version.txt
View File

@ -1 +1 @@
3.6.2
3.6.5

7
hg-store-grpc/src/main/proto/store_session.proto
View File

@ -5,6 +5,7 @@ option java_package = "org.apache.hugegraph.store.grpc.session";
option java_outer_classname = "HgStoreSessionProto";
import "store_common.proto";
import "store_stream_meta.proto";
service HgStoreSession {
rpc Get2(GetReq) returns (FeedbackRes) {}
@ -13,6 +14,7 @@ service HgStoreSession {
rpc Table(TableReq) returns (FeedbackRes){};
rpc Graph(GraphReq) returns (FeedbackRes){};
rpc Clean(CleanReq) returns (FeedbackRes) {}
rpc Count(ScanStreamReq) returns (Agg) {}
}
message TableReq{
@ -112,5 +114,8 @@ enum PartitionFaultType{
PARTITION_FAULT_TYPE_NOT_LOCAL = 3;
}
message Agg {
Header header = 1;
int64 count = 2;
}

23
hg-store-node/src/main/java/org/apache/hugegraph/store/node/grpc/HgStoreSessionImpl.java
View File

@ -26,6 +26,8 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hugegraph.pd.common.PDException;
import org.apache.hugegraph.pd.grpc.Metapb;
import org.apache.hugegraph.pd.grpc.Metapb.GraphMode;
import org.apache.hugegraph.rocksdb.access.ScanIterator;
import org.apache.hugegraph.store.business.BusinessHandler;
import org.apache.hugegraph.store.grpc.common.Key;
import org.apache.hugegraph.store.grpc.common.Kv;
import org.apache.hugegraph.store.grpc.common.ResCode;
@ -523,4 +525,25 @@ public class HgStoreSessionImpl extends HgStoreSessionGrpc.HgStoreSessionImplBas
}
GrpcClosure.setResult(response, builder.build());
}
@Override
public void count(ScanStreamReq request, StreamObserver<Agg> observer) {
ScanIterator it = null;
try {
BusinessHandler handler = storeService.getStoreEngine().getBusinessHandler();
long count = handler.count(request.getHeader().getGraph(), request.getTable());
observer.onNext(Agg.newBuilder().setCount(count).build());
observer.onCompleted();
} catch (Exception e) {
observer.onError(e);
} finally {
if (it != null) {
try {
it.close();
} catch (Exception e) {
}
}
}
}
}

92
hg-store-node/src/main/java/org/apache/hugegraph/store/node/grpc/ScanBatchResponse.java
View File

@ -20,7 +20,6 @@ package org.apache.hugegraph.store.node.grpc;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.hugegraph.rocksdb.access.ScanIterator;
import org.apache.hugegraph.store.buffer.ByteBufferAllocator;
@ -44,24 +43,23 @@ import lombok.extern.slf4j.Slf4j;
public class ScanBatchResponse implements StreamObserver<ScanStreamBatchReq> {
static ByteBufferAllocator bfAllocator =
new ByteBufferAllocator(ParallelScanIterator.maxBodySize * 3 / 2, 1000);
static ByteBufferAllocator alloc =
new ByteBufferAllocator(ParallelScanIterator.maxBodySize * 3 / 2, 1000);
private final int maxInFlightCount = PropertyUtil.getInt("app.scan.stream.inflight", 16);
private final int activeTimeout = PropertyUtil.getInt("app.scan.stream.timeout", 60); //单位秒
private final Object stateLock = new Object();
private final ReentrantLock iteratorLock = new ReentrantLock();
private final StreamObserver<KvStream> sender;
private final HgStoreWrapperEx wrapper;
private final ThreadPoolExecutor executor;
private final long logId;
// 当前正在遍历的迭代器
private ScanIterator iterator;
// 下一次发送的序号
private volatile int nextSeqNo;
private volatile int seqNo;
// Client已消费的序号
private volatile int clientSeqNo;
// 已经发送的条目数
private volatile long entriesCounter;
private volatile long count;
// 客户端要求返回的最大条目数
private volatile long clientLimit; // 客户端要求的最大条目数
private volatile long limit;
private ScanQueryRequest query;
// 上次读取数据时间
private long activeTime;
@ -73,10 +71,9 @@ public class ScanBatchResponse implements StreamObserver<ScanStreamBatchReq> {
this.wrapper = wrapper;
this.executor = executor;
this.iterator = null;
this.nextSeqNo = 1;
this.seqNo = 1;
this.state = State.IDLE;
this.activeTime = System.currentTimeMillis();
this.logId = response.hashCode();
}
/**
@ -95,7 +92,7 @@ public class ScanBatchResponse implements StreamObserver<ScanStreamBatchReq> {
break;
case RECEIPT_REQUEST: // 消息异步应答
this.clientSeqNo = request.getReceiptRequest().getTimes();
if (nextSeqNo - clientSeqNo < maxInFlightCount) {
if (seqNo - clientSeqNo < maxInFlightCount) {
synchronized (stateLock) {
if (state == State.IDLE) {
state = State.DOING;
@ -136,16 +133,9 @@ public class ScanBatchResponse implements StreamObserver<ScanStreamBatchReq> {
*/
private void startQuery(String graphName, ScanQueryRequest request) {
this.query = request;
// log.info("Stream {} startQuery graphName is {}, query degree/keylimit/limit is " +
// "{}/{}/{}, scanType is {}, orderType is {}",
// this.logId, graphName, query.getSkipDegree(), query.getPerKeyLimit(), query
// .getLimit(),
// query.getScanType(), query.getOrderType());
this.clientLimit = request.getLimit();
this.entriesCounter = 0;
this.iterator = ScanUtil.getParallelIterator(graphName, request, this.wrapper, executor);
this.limit = request.getLimit();
this.count = 0;
this.iterator = getParallelIterator(graphName, request, this.wrapper, executor);
synchronized (stateLock) {
if (state == State.IDLE) {
state = State.DOING;
@ -161,63 +151,75 @@ public class ScanBatchResponse implements StreamObserver<ScanStreamBatchReq> {
*/
private void closeQuery() {
setStateDone();
iteratorLock.lock();
try {
closeIter();
this.sender.onCompleted();
} catch (Exception e) {
log.error("exception ", e);
}
int active = ScanBatchResponseFactory.getInstance().removeStreamObserver(this);
log.info("ScanBatchResponse closeQuery, active count is {}", active);
}
private void closeIter() {
try {
if (this.iterator != null) {
this.iterator.close();
this.iterator = null;
}
this.sender.onCompleted();
} catch (Exception e) {
log.error("exception ", e);
} finally {
iteratorLock.unlock();
}
int active = ScanBatchResponseFactory.getInstance().removeStreamObserver(this);
log.info("ScanBatchResponse closeQuery, active count is {}", active);
}
/**
* 发送数据
*/
private void sendEntries() {
if (state == State.DONE || iterator == null) {
setStateIdle();
return;
}
iteratorLock.lock();
try {
if (state == State.DONE || iterator == null) {
setStateIdle();
return;
}
KvStream.Builder dataBuilder = KvStream.newBuilder()
.setVersion(1);
while (iterator.hasNext()
&& (nextSeqNo - clientSeqNo < maxInFlightCount)
&& this.entriesCounter < clientLimit
&& state != State.DONE) {
KVByteBuffer buffer = new KVByteBuffer(bfAllocator.get());
List<ParallelScanIterator.KV> dataList = iterator.next();
KvStream.Builder dataBuilder = KvStream.newBuilder().setVersion(1);
while (state != State.DONE && iterator.hasNext()
&& (seqNo - clientSeqNo < maxInFlightCount)
&& this.count < limit) {
KVByteBuffer buffer = new KVByteBuffer(alloc.get());
List<KV> dataList = iterator.next();
dataList.forEach(kv -> {
kv.write(buffer);
this.entriesCounter++;
this.count++;
});
dataBuilder.setStream(buffer.flip().getBuffer());
dataBuilder.setSeqNo(nextSeqNo++);
dataBuilder.complete(e -> {
bfAllocator.release(buffer.getBuffer());
});
dataBuilder.setSeqNo(seqNo++);
dataBuilder.complete(e -> alloc.release(buffer.getBuffer()));
this.sender.onNext(dataBuilder.build());
this.activeTime = System.currentTimeMillis();
}
if (!iterator.hasNext() || this.entriesCounter >= clientLimit || state == State.DONE) {
if (!iterator.hasNext() || this.count >= limit || state == State.DONE) {
closeIter();
this.sender.onNext(KvStream.newBuilder().setOver(true).build());
setStateDone();
} else {
setStateIdle();
}
} catch (Throwable e) {
log.error("exception ", e);
setStateIdle();
if (this.sender != null) {
this.sender.onError(e);
if (this.state != State.DONE) {
log.error(" send data exception: ", e);
setStateIdle();
if (this.sender != null) {
try {
this.sender.onError(e);
} catch (Exception ex) {
}
}
}
} finally {
iteratorLock.unlock();

2
hg-store-node/src/main/resources/version.txt
View File

@ -1 +1 @@
3.6.3
3.6.5

3
hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java
View File

@ -63,9 +63,6 @@ import org.rocksdb.Slice;
import org.rocksdb.Statistics;
import org.rocksdb.WriteBufferManager;
import org.rocksdb.WriteOptions;
import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.util.Bytes;
import org.apache.hugegraph.util.E;
import lombok.extern.slf4j.Slf4j;

1
hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/SessionOperatorImpl.java
View File

@ -32,7 +32,6 @@ import org.rocksdb.RocksIterator;
import org.rocksdb.Slice;
import org.rocksdb.Snapshot;
import org.rocksdb.WriteBatch;
import org.apache.hugegraph.util.Bytes;
import lombok.extern.slf4j.Slf4j;

7
hg-store-rocksdb/src/test/java/org/apache/hugegraph/rocksdb/access/SnapshotManagerTest.java
View File

@ -23,13 +23,6 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.hugegraph.store.term.HgPair;
import org.apache.commons.io.FileUtils;
import org.junit.AfterClass;
import org.junit.Assert;
import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.config.OptionSpace;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;

2
hg-store-test/src/main/resources/version.txt
View File

@ -1 +1 @@
3.6.0
3.6.5