You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hugegraph.apache.org by zh...@apache.org on 2022/05/19 06:10:30 UTC
[incubator-hugegraph-computer] branch master updated: feature: support partition concurrent compute (#177)
This is an automated email from the ASF dual-hosted git repository.
zhaocong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-hugegraph-computer.git
The following commit(s) were added to refs/heads/master by this push:
new bcaef856 feature: support partition concurrent compute (#177)
bcaef856 is described below
commit bcaef8563f15730b37b4d445cdf97a1f25c1e39f
Author: ShouJing <10...@qq.com>
AuthorDate: Thu May 19 14:10:26 2022 +0800
feature: support partition concurrent compute (#177)
---
.../computer/core/compute/ComputeManager.java | 83 +++++---
.../computer/core/compute/FileGraphPartition.java | 44 +++--
.../computer/core/config/ComputerOptions.java | 8 +
.../computer/core/input/WorkerInputManager.java | 1 +
.../computer/core/network/DataClientManager.java | 2 +-
.../computer/core/sender/MessageSendBuffers.java | 19 +-
.../computer/core/sender/MessageSendManager.java | 60 +++---
.../computer/core/sender/MessageSendPartition.java | 78 ++++++++
.../computer/core/sort/sorter/JavaInputSorter.java | 14 +-
.../hugegraph/computer/core/util/Consumers.java | 217 +++++++++++++++++++++
.../computer/core/worker/WorkerService.java | 30 +--
.../computer/core/compute/ComputeManagerTest.java | 9 +-
.../core/compute/input/EdgesInputTest.java | 4 +-
.../core/sender/MessageSendBuffersTest.java | 2 +-
14 files changed, 467 insertions(+), 104 deletions(-)
diff --git a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/compute/ComputeManager.java b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/compute/ComputeManager.java
index 7376e07b..a78a26f6 100644
--- a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/compute/ComputeManager.java
+++ b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/compute/ComputeManager.java
@@ -21,13 +21,16 @@ package com.baidu.hugegraph.computer.core.compute;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
import org.slf4j.Logger;
import com.baidu.hugegraph.computer.core.common.ComputerContext;
import com.baidu.hugegraph.computer.core.common.exception.ComputerException;
+import com.baidu.hugegraph.computer.core.config.ComputerOptions;
+import com.baidu.hugegraph.computer.core.config.Config;
import com.baidu.hugegraph.computer.core.graph.partition.PartitionStat;
-import com.baidu.hugegraph.computer.core.graph.value.Value;
import com.baidu.hugegraph.computer.core.manager.Managers;
import com.baidu.hugegraph.computer.core.network.message.MessageType;
import com.baidu.hugegraph.computer.core.receiver.MessageRecvManager;
@@ -35,31 +38,41 @@ import com.baidu.hugegraph.computer.core.receiver.MessageStat;
import com.baidu.hugegraph.computer.core.sender.MessageSendManager;
import com.baidu.hugegraph.computer.core.sort.flusher.PeekableIterator;
import com.baidu.hugegraph.computer.core.store.hgkvfile.entry.KvEntry;
-import com.baidu.hugegraph.computer.core.worker.Computation;
-import com.baidu.hugegraph.computer.core.worker.ComputationContext;
+import com.baidu.hugegraph.computer.core.util.Consumers;
+import com.baidu.hugegraph.computer.core.worker.WorkerContext;
import com.baidu.hugegraph.computer.core.worker.WorkerStat;
+import com.baidu.hugegraph.util.ExecutorUtil;
import com.baidu.hugegraph.util.Log;
-public class ComputeManager<M extends Value> {
+public class ComputeManager {
private static final Logger LOG = Log.logger(ComputeManager.class);
+ private static final String PREFIX = "partition-compute-executor-%s";
private final ComputerContext context;
private final Managers managers;
- private final Map<Integer, FileGraphPartition<M>> partitions;
- private final Computation<M> computation;
+ private final Map<Integer, FileGraphPartition> partitions;
private final MessageRecvManager recvManager;
private final MessageSendManager sendManager;
+ private final ExecutorService computeExecutor;
- public ComputeManager(ComputerContext context, Managers managers,
- Computation<M> computation) {
+ public ComputeManager(ComputerContext context, Managers managers) {
this.context = context;
this.managers = managers;
- this.computation = computation;
this.partitions = new HashMap<>();
this.recvManager = this.managers.get(MessageRecvManager.NAME);
this.sendManager = this.managers.get(MessageSendManager.NAME);
+
+ int computeThreadNum = this.partitionComputeThreadNum(context.config());
+ this.computeExecutor = ExecutorUtil.newFixedThreadPool(
+ computeThreadNum, PREFIX);
+ LOG.info("Created partition compute thread poll, thread num: {}",
+ computeThreadNum);
+ }
+
+ private Integer partitionComputeThreadNum(Config config) {
+ return config.get(ComputerOptions.PARTITIONS_COMPUTE_THREAD_NUMS);
}
public WorkerStat input() {
@@ -70,6 +83,7 @@ public class ComputeManager<M extends Value> {
this.recvManager.vertexPartitions();
Map<Integer, PeekableIterator<KvEntry>> edges =
this.recvManager.edgePartitions();
+
// TODO: parallel input process
for (Map.Entry<Integer, PeekableIterator<KvEntry>> entry :
vertices.entrySet()) {
@@ -80,9 +94,9 @@ public class ComputeManager<M extends Value> {
partition,
PeekableIterator.emptyIterator());
- FileGraphPartition<M> part = new FileGraphPartition<>(this.context,
- this.managers,
- partition);
+ FileGraphPartition part = new FileGraphPartition(this.context,
+ this.managers,
+ partition);
PartitionStat partitionStat = null;
ComputerException inputException = null;
try {
@@ -123,30 +137,45 @@ public class ComputeManager<M extends Value> {
public void takeRecvedMessages() {
Map<Integer, PeekableIterator<KvEntry>> messages =
this.recvManager.messagePartitions();
- for (FileGraphPartition<M> partition : this.partitions.values()) {
+ for (FileGraphPartition partition : this.partitions.values()) {
partition.messages(messages.get(partition.partition()));
}
}
- public WorkerStat compute(ComputationContext context, int superstep) {
+ public WorkerStat compute(WorkerContext context, int superstep) {
this.sendManager.startSend(MessageType.MSG);
+
WorkerStat workerStat = new WorkerStat();
- Map<Integer, PartitionStat> partitionStats = new HashMap<>(
- this.partitions.size());
- // TODO: parallel compute process.
- for (FileGraphPartition<M> partition : this.partitions.values()) {
- PartitionStat stat = partition.compute(context,
- this.computation,
- superstep);
- partitionStats.put(stat.partitionId(), stat);
+ Map<Integer, PartitionStat> stats = new ConcurrentHashMap<>();
+
+ /*
+ * Remark: The main thread can perceive the partition compute exception
+ * only after all partition compute completed, and only record the last
+ * exception.
+ */
+ Consumers<FileGraphPartition> consumers =
+ new Consumers<>(this.computeExecutor, partition -> {
+ PartitionStat stat = partition.compute(context,
+ superstep);
+ stats.put(stat.partitionId(), stat);
+ });
+ consumers.start("partition-compute");
+
+ try {
+ for (FileGraphPartition partition : this.partitions.values()) {
+ consumers.provide(partition);
+ }
+ consumers.await();
+ } catch (Throwable t) {
+ throw new ComputerException("An exception occurred when " +
+ "partition parallel compute", t);
}
this.sendManager.finishSend(MessageType.MSG);
// After compute and send finish signal.
Map<Integer, MessageStat> recvStats = this.recvManager.messageStats();
- for (Map.Entry<Integer, PartitionStat> entry :
- partitionStats.entrySet()) {
+ for (Map.Entry<Integer, PartitionStat> entry : stats.entrySet()) {
PartitionStat partStat = entry.getValue();
int partitionId = partStat.partitionId();
@@ -165,10 +194,14 @@ public class ComputeManager<M extends Value> {
public void output() {
// TODO: Write results back parallel
- for (FileGraphPartition<M> partition : this.partitions.values()) {
+ for (FileGraphPartition partition : this.partitions.values()) {
PartitionStat stat = partition.output();
LOG.info("Output partition {} complete, stat='{}'",
partition.partition(), stat);
}
}
+
+ public void close() {
+ this.computeExecutor.shutdown();
+ }
}
diff --git a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/compute/FileGraphPartition.java b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/compute/FileGraphPartition.java
index d9ac3e33..8f8bab64 100644
--- a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/compute/FileGraphPartition.java
+++ b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/compute/FileGraphPartition.java
@@ -23,6 +23,8 @@ import java.io.File;
import java.io.IOException;
import java.util.Iterator;
+import org.slf4j.Logger;
+
import com.baidu.hugegraph.computer.core.common.ComputerContext;
import com.baidu.hugegraph.computer.core.common.Constants;
import com.baidu.hugegraph.computer.core.common.exception.ComputerException;
@@ -47,9 +49,13 @@ import com.baidu.hugegraph.computer.core.store.hgkvfile.entry.KvEntry;
import com.baidu.hugegraph.computer.core.store.hgkvfile.entry.Pointer;
import com.baidu.hugegraph.computer.core.worker.Computation;
import com.baidu.hugegraph.computer.core.worker.ComputationContext;
+import com.baidu.hugegraph.computer.core.worker.WorkerContext;
import com.baidu.hugegraph.util.E;
+import com.baidu.hugegraph.util.Log;
+
+public class FileGraphPartition {
-public class FileGraphPartition<M extends Value> {
+ private static final Logger LOG = Log.logger(FileGraphPartition.class);
private static final String VERTEX = "vertex";
private static final String EDGE = "edge";
@@ -57,6 +63,7 @@ public class FileGraphPartition<M extends Value> {
private static final String VALUE = "value";
private final ComputerContext context;
+ private final Computation<Value> computation;
private final FileGenerator fileGenerator;
private final int partition;
@@ -79,12 +86,16 @@ public class FileGraphPartition<M extends Value> {
private VertexInput vertexInput;
private EdgesInput edgesInput;
- private MessageInput<M> messageInput;
+ private MessageInput<Value> messageInput;
public FileGraphPartition(ComputerContext context,
Managers managers,
int partition) {
this.context = context;
+ this.computation = context.config()
+ .createObject(
+ ComputerOptions.WORKER_COMPUTATION_CLASS);
+ this.computation.init(context.config());
this.fileGenerator = managers.get(FileManager.NAME);
this.partition = partition;
this.vertexFile = new File(this.fileGenerator.randomDirectory(VERTEX));
@@ -121,9 +132,10 @@ public class FileGraphPartition<M extends Value> {
this.edgeCount, 0L);
}
- protected PartitionStat compute(ComputationContext context,
- Computation<M> computation,
+ protected PartitionStat compute(WorkerContext context,
int superstep) {
+ LOG.info("Partition {} begin compute in superstep {}",
+ this.partition, superstep);
try {
this.beforeCompute(superstep);
} catch (IOException e) {
@@ -134,9 +146,11 @@ public class FileGraphPartition<M extends Value> {
long activeVertexCount;
try {
+ this.computation.beforeSuperstep(context);
activeVertexCount = superstep == 0 ?
- this.compute0(context, computation) :
- this.compute1(context, computation, superstep);
+ this.compute0(context) :
+ this.compute1(context);
+ this.computation.afterSuperstep(context);
} catch (Exception e) {
throw new ComputerException(
"Error occurred when compute at superstep %s",
@@ -151,14 +165,16 @@ public class FileGraphPartition<M extends Value> {
e, superstep);
}
+ LOG.info("Partition {} finish compute in superstep {}",
+ this.partition, superstep);
+
return new PartitionStat(this.partition, this.vertexCount,
this.edgeCount,
this.vertexCount - activeVertexCount);
}
- private long compute0(ComputationContext context,
- Computation<M> computation) {
+ private long compute0(ComputationContext context) {
long activeVertexCount = 0L;
while (this.vertexInput.hasNext()) {
Vertex vertex = this.vertexInput.next();
@@ -167,7 +183,7 @@ public class FileGraphPartition<M extends Value> {
Edges edges = this.edgesInput.edges(this.vertexInput.idPointer());
vertex.edges(edges);
- computation.compute0(context, vertex);
+ this.computation.compute0(context, vertex);
if (vertex.active()) {
activeVertexCount++;
@@ -183,9 +199,7 @@ public class FileGraphPartition<M extends Value> {
return activeVertexCount;
}
- private long compute1(ComputationContext context,
- Computation<M> computation,
- int superstep) {
+ private long compute1(ComputationContext context) {
Value result = this.context.config().createObject(
ComputerOptions.ALGORITHM_RESULT_CLASS);
long activeVertexCount = 0L;
@@ -193,8 +207,8 @@ public class FileGraphPartition<M extends Value> {
Vertex vertex = this.vertexInput.next();
this.readVertexStatusAndValue(vertex, result);
- Iterator<M> messageIter = this.messageInput.iterator(
- this.vertexInput.idPointer());
+ Iterator<Value> messageIter = this.messageInput.iterator(
+ this.vertexInput.idPointer());
if (messageIter.hasNext()) {
vertex.reactivate();
}
@@ -207,7 +221,7 @@ public class FileGraphPartition<M extends Value> {
Edges edges = this.edgesInput.edges(
this.vertexInput.idPointer());
vertex.edges(edges);
- computation.compute(context, vertex, messageIter);
+ this.computation.compute(context, vertex, messageIter);
}
// The vertex status may be changed after computation
diff --git a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/config/ComputerOptions.java b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/config/ComputerOptions.java
index 4b9c17b5..e4e9fb76 100644
--- a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/config/ComputerOptions.java
+++ b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/config/ComputerOptions.java
@@ -433,6 +433,14 @@ public class ComputerOptions extends OptionHolder {
1
);
+ public static final ConfigOption<Integer> PARTITIONS_COMPUTE_THREAD_NUMS =
+ new ConfigOption<>(
+ "job.partitions_thread_nums",
+ "The number of threads for partition parallel compute.",
+ positiveInt(),
+ 4
+ );
+
public static final ConfigOption<Integer> BSP_MAX_SUPER_STEP =
new ConfigOption<>(
"bsp.max_super_step",
diff --git a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/input/WorkerInputManager.java b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/input/WorkerInputManager.java
index 833d28eb..35acf011 100644
--- a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/input/WorkerInputManager.java
+++ b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/input/WorkerInputManager.java
@@ -92,5 +92,6 @@ public class WorkerInputManager implements Manager {
this.sendManager.sendEdge(vertex);
}
this.sendManager.finishSend(MessageType.EDGE);
+ this.sendManager.clearBuffer();
}
}
diff --git a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/network/DataClientManager.java b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/network/DataClientManager.java
index 249d6cbc..ae292b92 100644
--- a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/network/DataClientManager.java
+++ b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/network/DataClientManager.java
@@ -87,7 +87,7 @@ public class DataClientManager implements Manager {
this.sender.addWorkerClient(workerId, client);
} catch (TransportException e) {
throw new ComputerException(
- "Failed to connect to worker: {}({}:{})",
+ "Failed to connect to worker: %s(%s:%s)",
workerId, hostname, dataPort);
}
}
diff --git a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/sender/MessageSendBuffers.java b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/sender/MessageSendBuffers.java
index 4d2882d3..37abf03e 100644
--- a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/sender/MessageSendBuffers.java
+++ b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/sender/MessageSendBuffers.java
@@ -34,7 +34,7 @@ public class MessageSendBuffers {
* Add a MessageSendPartition class when find that we really need it
* to encapsulate more objects.
*/
- private final WriteBuffers[] buffers;
+ private final MessageSendPartition[] buffers;
public MessageSendBuffers(ComputerContext context) {
Config config = context.config();
@@ -43,13 +43,14 @@ public class MessageSendBuffers {
ComputerOptions.WORKER_WRITE_BUFFER_THRESHOLD);
int capacity = config.get(
ComputerOptions.WORKER_WRITE_BUFFER_INIT_CAPACITY);
- this.buffers = new WriteBuffers[partitionCount];
+ this.buffers = new MessageSendPartition[partitionCount];
for (int i = 0; i < partitionCount; i++) {
/*
* It depends on the concrete implementation of the
* partition algorithm, which is not elegant.
*/
- this.buffers[i] = new WriteBuffers(context, threshold, capacity);
+ this.buffers[i] = new MessageSendPartition(context, threshold,
+ capacity);
}
}
@@ -57,14 +58,20 @@ public class MessageSendBuffers {
if (partitionId < 0 || partitionId >= this.buffers.length) {
throw new ComputerException("Invalid partition id %s", partitionId);
}
- return this.buffers[partitionId];
+ return this.buffers[partitionId].buffersForCurrentThread();
}
- public Map<Integer, WriteBuffers> all() {
- Map<Integer, WriteBuffers> all = InsertionOrderUtil.newMap();
+ public Map<Integer, MessageSendPartition> all() {
+ Map<Integer, MessageSendPartition> all = InsertionOrderUtil.newMap();
for (int i = 0; i < this.buffers.length; i++) {
all.put(i, this.buffers[i]);
}
return all;
}
+
+ public void clear() {
+ for (MessageSendPartition partition : this.buffers) {
+ partition.clear();
+ }
+ }
}
diff --git a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/sender/MessageSendManager.java b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/sender/MessageSendManager.java
index d11dd01c..7a287070 100644
--- a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/sender/MessageSendManager.java
+++ b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/sender/MessageSendManager.java
@@ -96,8 +96,11 @@ public class MessageSendManager implements Manager {
public void sendVertex(Vertex vertex) {
this.checkException();
- WriteBuffers buffer = this.sortIfTargetBufferIsFull(vertex.id(),
- MessageType.VERTEX);
+ int partitionId = this.partitioner.partitionId(vertex.id());
+ WriteBuffers buffer = this.buffers.get(partitionId);
+
+ this.sortIfTargetBufferIsFull(buffer, partitionId,
+ MessageType.VERTEX);
try {
// Write vertex to buffer
buffer.writeVertex(vertex);
@@ -110,22 +113,27 @@ public class MessageSendManager implements Manager {
public void sendEdge(Vertex vertex) {
this.checkException();
- WriteBuffers buffer = this.sortIfTargetBufferIsFull(vertex.id(),
- MessageType.EDGE);
+ int partitionId = this.partitioner.partitionId(vertex.id());
+ WriteBuffers buffer = this.buffers.get(partitionId);
+
+ this.sortIfTargetBufferIsFull(buffer, partitionId,
+ MessageType.EDGE);
try {
// Write edge to buffer
buffer.writeEdges(vertex);
} catch (IOException e) {
- throw new ComputerException("Failed to write edges of vertex '%s'",
- e, vertex.id());
+ throw new ComputerException("Failed to write edges of " +
+ "vertex '%s'", e, vertex.id());
}
}
public void sendMessage(Id targetId, Value value) {
this.checkException();
- WriteBuffers buffer = this.sortIfTargetBufferIsFull(targetId,
- MessageType.MSG);
+ int partitionId = this.partitioner.partitionId(targetId);
+ WriteBuffers buffer = this.buffers.get(partitionId);
+
+ this.sortIfTargetBufferIsFull(buffer, partitionId, MessageType.MSG);
try {
// Write message to buffer
buffer.writeMessage(targetId, value);
@@ -139,8 +147,8 @@ public class MessageSendManager implements Manager {
* @param type the message type
*/
public void startSend(MessageType type) {
- Map<Integer, WriteBuffers> all = this.buffers.all();
- all.values().stream().forEach(WriteBuffers::resetMessageWritten);
+ Map<Integer, MessageSendPartition> all = this.buffers.all();
+ all.values().forEach(MessageSendPartition::resetMessageWritten);
Set<Integer> workerIds = all.keySet().stream()
.map(this.partitioner::workerId)
.collect(Collectors.toSet());
@@ -154,7 +162,7 @@ public class MessageSendManager implements Manager {
* @param type the message type
*/
public void finishSend(MessageType type) {
- Map<Integer, WriteBuffers> all = this.buffers.all();
+ Map<Integer, MessageSendPartition> all = this.buffers.all();
MessageStat stat = this.sortAndSendLastBuffer(all, type);
Set<Integer> workerIds = all.keySet().stream()
@@ -169,9 +177,13 @@ public class MessageSendManager implements Manager {
return this.buffers.get(partitionId).messageWritten();
}
- private WriteBuffers sortIfTargetBufferIsFull(Id id, MessageType type) {
- int partitionId = this.partitioner.partitionId(id);
- WriteBuffers buffer = this.buffers.get(partitionId);
+ public void clearBuffer() {
+ this.buffers.clear();
+ }
+
+ private void sortIfTargetBufferIsFull(WriteBuffers buffer,
+ int partitionId,
+ MessageType type) {
if (buffer.reachThreshold()) {
/*
* Switch buffers if writing buffer is full,
@@ -180,7 +192,6 @@ public class MessageSendManager implements Manager {
buffer.switchForSorting();
this.sortThenSend(partitionId, type, buffer);
}
- return buffer;
}
private Future<?> sortThenSend(int partitionId, MessageType type,
@@ -208,24 +219,27 @@ public class MessageSendManager implements Manager {
});
}
- private MessageStat sortAndSendLastBuffer(Map<Integer, WriteBuffers> all,
- MessageType type) {
+ private MessageStat sortAndSendLastBuffer(
+ Map<Integer, MessageSendPartition> all,
+ MessageType type) {
MessageStat messageWritten = new MessageStat();
List<Future<?>> futures = new ArrayList<>(all.size());
// Sort and send the last buffer
- for (Map.Entry<Integer, WriteBuffers> entry : all.entrySet()) {
+ for (Map.Entry<Integer, MessageSendPartition> entry : all.entrySet()) {
int partitionId = entry.getKey();
- WriteBuffers buffer = entry.getValue();
+ MessageSendPartition partition = entry.getValue();
/*
* If the last buffer has already been sorted and sent (empty),
* there is no need to send again here
*/
- if (!buffer.isEmpty()) {
- buffer.prepareSorting();
- futures.add(this.sortThenSend(partitionId, type, buffer));
+ for (WriteBuffers buffer : partition.buffers()) {
+ if (!buffer.isEmpty()) {
+ buffer.prepareSorting();
+ futures.add(this.sortThenSend(partitionId, type, buffer));
+ }
}
// Record total message count & bytes
- messageWritten.increase(buffer.messageWritten());
+ messageWritten.increase(partition.messageWritten());
}
this.checkException();
diff --git a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/sender/MessageSendPartition.java b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/sender/MessageSendPartition.java
new file mode 100644
index 00000000..88d9368d
--- /dev/null
+++ b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/sender/MessageSendPartition.java
@@ -0,0 +1,78 @@
+/*
+ * Copyright 2017 HugeGraph Authors
+ *
+ * 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 com.baidu.hugegraph.computer.core.sender;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.baidu.hugegraph.computer.core.common.ComputerContext;
+import com.baidu.hugegraph.computer.core.receiver.MessageStat;
+
+public class MessageSendPartition {
+
+ private final ComputerContext context;
+ private final int threshold;
+ private final int capacity;
+
+ private final Map<Thread, WriteBuffers> buffers;
+
+ public MessageSendPartition(ComputerContext context,
+ int threshold, int capacity) {
+ this.context = context;
+ this.threshold = threshold;
+ this.capacity = capacity;
+
+ this.buffers = new ConcurrentHashMap<>();
+ }
+
+ public WriteBuffers buffersForCurrentThread() {
+ Thread current = Thread.currentThread();
+ WriteBuffers buffer = this.buffers.get(current);
+ if (buffer == null) {
+ buffer = new WriteBuffers(this.context, this.threshold,
+ this.capacity);
+ this.buffers.put(current, buffer);
+ }
+ return buffer;
+ }
+
+ public synchronized void clear() {
+ this.buffers.clear();
+ }
+
+ public synchronized void resetMessageWritten() {
+ for (WriteBuffers buffer : this.buffers.values()) {
+ buffer.resetMessageWritten();
+ }
+ }
+
+ public synchronized MessageStat messageWritten() {
+ MessageStat partitionStat = new MessageStat();
+ for (WriteBuffers buffer : this.buffers.values()) {
+ partitionStat.increase(buffer.messageWritten());
+ }
+ return partitionStat;
+ }
+
+ public Collection<WriteBuffers> buffers() {
+ return this.buffers.values();
+ }
+}
diff --git a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/sort/sorter/JavaInputSorter.java b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/sort/sorter/JavaInputSorter.java
index f7b77120..818d4ec2 100644
--- a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/sort/sorter/JavaInputSorter.java
+++ b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/sort/sorter/JavaInputSorter.java
@@ -28,13 +28,19 @@ import com.baidu.hugegraph.computer.core.store.hgkvfile.entry.KvEntry;
public class JavaInputSorter implements InputSorter {
- // TODO improve it by reuse list object
- private static final int DEFAULT_CAPACITY = 10;
+ private static final ThreadLocal<List<KvEntry>> SORT_LOCAL =
+ ThreadLocal.withInitial(ArrayList::new);
+
+ private static List<KvEntry> threadLocalSortList() {
+ List<KvEntry> list = SORT_LOCAL.get();
+ list.clear();
+ return list;
+ }
@Override
public Iterator<KvEntry> sort(Iterator<KvEntry> entries)
- throws IOException {
- List<KvEntry> kvEntries = new ArrayList<>(DEFAULT_CAPACITY);
+ throws IOException {
+ List<KvEntry> kvEntries = threadLocalSortList();
while (entries.hasNext()) {
kvEntries.add(entries.next());
}
diff --git a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/util/Consumers.java b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/util/Consumers.java
new file mode 100644
index 00000000..278c9dd9
--- /dev/null
+++ b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/util/Consumers.java
@@ -0,0 +1,217 @@
+/*
+ * Copyright 2017 HugeGraph Authors
+ *
+ * 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 com.baidu.hugegraph.computer.core.util;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+import org.slf4j.Logger;
+
+import com.baidu.hugegraph.computer.core.common.exception.ComputerException;
+import com.baidu.hugegraph.util.Log;
+
+public final class Consumers<V> {
+
+ private static final int CPU_CORE_NUM = 4;
+ public static final int THREADS = 4 + CPU_CORE_NUM / 4;
+ public static final int QUEUE_WORKER_SIZE = 1000;
+ public static final long CONSUMER_WAKE_PERIOD = 1;
+
+ private static final Logger LOG = Log.logger(Consumers.class);
+
+ private final ExecutorService executor;
+ private final Consumer<V> consumer;
+ private final Runnable done;
+
+ private final int workers;
+ private final int queueSize;
+ private final CountDownLatch latch;
+ private final BlockingQueue<V> queue;
+
+ private volatile boolean ending = false;
+ private volatile Throwable exception = null;
+
+ public Consumers(ExecutorService executor, Consumer<V> consumer) {
+ this(executor, consumer, null);
+ }
+
+ public Consumers(ExecutorService executor,
+ Consumer<V> consumer, Runnable done) {
+ this.executor = executor;
+ this.consumer = consumer;
+ this.done = done;
+
+ int workers = THREADS;
+ if (this.executor instanceof ThreadPoolExecutor) {
+ workers = ((ThreadPoolExecutor) this.executor).getCorePoolSize();
+ }
+ this.workers = workers;
+ this.queueSize = QUEUE_WORKER_SIZE * workers;
+ this.latch = new CountDownLatch(workers);
+ this.queue = new ArrayBlockingQueue<>(this.queueSize);
+ }
+
+ public void start(String name) {
+ this.ending = false;
+ this.exception = null;
+ if (this.executor == null) {
+ return;
+ }
+ LOG.info("Starting {} workers[{}] with queue size {}...",
+ this.workers, name, this.queueSize);
+ for (int i = 0; i < this.workers; i++) {
+ this.executor.execute(this::runAndDone);
+ }
+ }
+
+ private void runAndDone() {
+ try {
+ this.run();
+ } catch (Throwable e) {
+ // Only the first exception of one thread can be stored
+ this.exception = e;
+ if (!(e instanceof StopExecution)) {
+ LOG.error("Error when running task", e);
+ }
+ } finally {
+ this.done();
+ this.latch.countDown();
+ }
+ }
+
+ private void run() {
+ LOG.debug("Start to work...");
+ while (!this.ending) {
+ this.consume();
+ }
+ assert this.ending;
+ while (this.consume()) {
+
+ }
+
+ LOG.debug("Worker finished");
+ }
+
+ private boolean consume() {
+ V elem;
+ try {
+ elem = this.queue.poll(CONSUMER_WAKE_PERIOD, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ // ignore
+ return true;
+ }
+ if (elem == null) {
+ return false;
+ }
+ // do job
+ this.consumer.accept(elem);
+ return true;
+ }
+
+ private void done() {
+ if (this.done == null) {
+ return;
+ }
+
+ try {
+ this.done.run();
+ } catch (Throwable e) {
+ if (this.exception == null) {
+ this.exception = e;
+ } else {
+ LOG.warn("Error while calling done()", e);
+ }
+ }
+ }
+
+ private Throwable throwException() {
+ assert this.exception != null;
+ Throwable e = this.exception;
+ this.exception = null;
+ return e;
+ }
+
+ public void provide(V v) throws Throwable {
+ if (this.executor == null) {
+ assert this.exception == null;
+ // do job directly if without thread pool
+ this.consumer.accept(v);
+ } else if (this.exception != null) {
+ throw this.throwException();
+ } else {
+ try {
+ this.queue.put(v);
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted while enqueue", e);
+ }
+ }
+ }
+
+ public void await() throws Throwable {
+ this.ending = true;
+ if (this.executor == null) {
+ // call done() directly if without thread pool
+ this.done();
+ } else {
+ try {
+ this.latch.await();
+ } catch (InterruptedException e) {
+ String error = "Interrupted while waiting for consumers";
+ this.exception = new ComputerException(error, e);
+ LOG.warn(error, e);
+ }
+ }
+
+ if (this.exception != null) {
+ throw this.throwException();
+ }
+ }
+
+ public ExecutorService executor() {
+ return this.executor;
+ }
+
+ public static RuntimeException wrapException(Throwable e) {
+ if (e instanceof RuntimeException) {
+ throw (RuntimeException) e;
+ }
+ throw new ComputerException("Error when running task: %s",
+ ComputerException.rootCause(e).getMessage(),
+ e);
+ }
+
+ public static class StopExecution extends ComputerException {
+
+ private static final long serialVersionUID = -371829356182454517L;
+
+ public StopExecution(String message) {
+ super(message);
+ }
+
+ public StopExecution(String message, Object... args) {
+ super(message, args);
+ }
+ }
+}
diff --git a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/worker/WorkerService.java b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/worker/WorkerService.java
index 61a19130..201a473e 100644
--- a/computer-core/src/main/java/com/baidu/hugegraph/computer/core/worker/WorkerService.java
+++ b/computer-core/src/main/java/com/baidu/hugegraph/computer/core/worker/WorkerService.java
@@ -71,10 +71,9 @@ public class WorkerService implements Closeable {
private volatile boolean closed;
private Config config;
private Bsp4Worker bsp4Worker;
- private ComputeManager<?> computeManager;
+ private ComputeManager computeManager;
private ContainerInfo workerInfo;
- private Computation<Value> computation;
private Combiner<Value> combiner;
private ContainerInfo masterInfo;
@@ -116,20 +115,19 @@ public class WorkerService implements Closeable {
InetSocketAddress address = this.initManagers(this.masterInfo);
this.workerInfo.updateAddress(address);
- this.computation = this.config.createObject(
- ComputerOptions.WORKER_COMPUTATION_CLASS);
- this.computation.init(this.config);
+ Computation<?> computation = this.config.createObject(
+ ComputerOptions.WORKER_COMPUTATION_CLASS);
LOG.info("Loading computation '{}' in category '{}'",
- this.computation.name(), this.computation.category());
+ computation.name(), computation.category());
this.combiner = this.config.createObject(
ComputerOptions.WORKER_COMBINER_CLASS, false);
if (this.combiner == null) {
LOG.info("None combiner is provided for computation '{}'",
- this.computation.name());
+ computation.name());
} else {
LOG.info("Combiner '{}' is provided for computation '{}'",
- this.combiner.name(), this.computation.name());
+ this.combiner.name(), computation.name());
}
LOG.info("{} register WorkerService", this);
@@ -141,11 +139,7 @@ public class WorkerService implements Closeable {
dm.connect(worker.id(), worker.hostname(), worker.dataPort());
}
- @SuppressWarnings({ "unchecked", "rawtypes" })
- ComputeManager<?> computeManager = new ComputeManager(this.context,
- this.managers,
- this.computation);
- this.computeManager = computeManager;
+ this.computeManager = new ComputeManager(this.context, this.managers);
this.managers.initedAll(this.config);
LOG.info("{} WorkerService initialized", this);
@@ -171,8 +165,7 @@ public class WorkerService implements Closeable {
return;
}
- this.computation.close(this.config);
-
+ this.computeManager.close();
/*
* Seems managers.closeAll() would do the following actions:
* TODO: close the connection to other workers.
@@ -245,11 +238,10 @@ public class WorkerService implements Closeable {
/*
* Call beforeSuperstep() before all workers compute() called.
*
- * NOTE: keep computation.beforeSuperstep() called after
+ * NOTE: keep computeManager.compute() called after
* managers.beforeSuperstep().
*/
this.managers.beforeSuperstep(this.config, superstep);
- this.computation.beforeSuperstep(context);
/*
* Notify master by each worker, when the master received all
@@ -267,11 +259,10 @@ public class WorkerService implements Closeable {
* Call afterSuperstep() after all workers compute() is done.
*
* NOTE: keep managers.afterSuperstep() called after
- * computation.afterSuperstep(), because managers may rely on
+ * computeManager.compute(), because managers may rely on
* computation, like WorkerAggrManager send aggregators to master
* after called aggregateValue(String name, V value) in computation.
*/
- this.computation.afterSuperstep(context);
this.managers.afterSuperstep(this.config, superstep);
this.bsp4Worker.workerStepDone(superstep, workerStat);
@@ -321,7 +312,6 @@ public class WorkerService implements Closeable {
DataServerManager serverManager = new DataServerManager(connManager,
recvManager);
this.managers.add(serverManager);
- this.managers.add(serverManager);
DataClientManager clientManager = new DataClientManager(connManager,
this.context);
diff --git a/computer-test/src/main/java/com/baidu/hugegraph/computer/core/compute/ComputeManagerTest.java b/computer-test/src/main/java/com/baidu/hugegraph/computer/core/compute/ComputeManagerTest.java
index 09058e69..88fbd80a 100644
--- a/computer-test/src/main/java/com/baidu/hugegraph/computer/core/compute/ComputeManagerTest.java
+++ b/computer-test/src/main/java/com/baidu/hugegraph/computer/core/compute/ComputeManagerTest.java
@@ -58,7 +58,6 @@ import com.baidu.hugegraph.computer.core.sort.sorting.SortManager;
import com.baidu.hugegraph.computer.core.store.FileManager;
import com.baidu.hugegraph.computer.core.store.hgkvfile.entry.EntryOutput;
import com.baidu.hugegraph.computer.core.store.hgkvfile.entry.EntryOutputImpl;
-import com.baidu.hugegraph.computer.core.worker.Computation;
import com.baidu.hugegraph.computer.suite.unit.UnitTestBase;
import com.baidu.hugegraph.testutil.Whitebox;
@@ -69,10 +68,9 @@ public class ComputeManagerTest extends UnitTestBase {
private Config config;
private Managers managers;
private ConnectionId connectionId;
- private ComputeManager<?> computeManager;
+ private ComputeManager computeManager;
@Before
- @SuppressWarnings({ "rawtypes", "unchecked" })
public void setup() {
this.config = UnitTestBase.updateWithRequiredOptions(
ComputerOptions.JOB_ID, "local_001",
@@ -112,10 +110,7 @@ public class ComputeManagerTest extends UnitTestBase {
this.connectionId = new ConnectionId(new InetSocketAddress("localhost",
8081),
0);
- Computation<?> computation = this.config.createObject(
- ComputerOptions.WORKER_COMPUTATION_CLASS);
- this.computeManager = new ComputeManager(context(), this.managers,
- computation);
+ this.computeManager = new ComputeManager(context(), this.managers);
}
@After
diff --git a/computer-test/src/main/java/com/baidu/hugegraph/computer/core/compute/input/EdgesInputTest.java b/computer-test/src/main/java/com/baidu/hugegraph/computer/core/compute/input/EdgesInputTest.java
index 990a5027..11625b49 100644
--- a/computer-test/src/main/java/com/baidu/hugegraph/computer/core/compute/input/EdgesInputTest.java
+++ b/computer-test/src/main/java/com/baidu/hugegraph/computer/core/compute/input/EdgesInputTest.java
@@ -143,8 +143,8 @@ public class EdgesInputTest extends UnitTestBase {
ConnectionId connectionId = new ConnectionId(new InetSocketAddress(
"localhost", 8081),
0);
- FileGraphPartition<?> partition = new FileGraphPartition<>(
- context(), this.managers, 0);
+ FileGraphPartition partition = new FileGraphPartition(
+ context(), this.managers, 0);
receiveManager.onStarted(connectionId);
add200VertexBuffer((ManagedBuffer buffer) -> {
receiveManager.handle(MessageType.VERTEX, 0, buffer);
diff --git a/computer-test/src/main/java/com/baidu/hugegraph/computer/core/sender/MessageSendBuffersTest.java b/computer-test/src/main/java/com/baidu/hugegraph/computer/core/sender/MessageSendBuffersTest.java
index b1d4c798..becc53e3 100644
--- a/computer-test/src/main/java/com/baidu/hugegraph/computer/core/sender/MessageSendBuffersTest.java
+++ b/computer-test/src/main/java/com/baidu/hugegraph/computer/core/sender/MessageSendBuffersTest.java
@@ -41,7 +41,7 @@ public class MessageSendBuffersTest extends UnitTestBase {
MessageSendBuffers buffers = new MessageSendBuffers(
ComputerContext.instance());
- Map<Integer, WriteBuffers> innerBuffers = buffers.all();
+ Map<Integer, MessageSendPartition> innerBuffers = buffers.all();
Assert.assertEquals(3, innerBuffers.size());
}