feat: implement parallel send data in load graph step #248

Radeity · 2023-05-25T07:42:43Z

Purpose of the PR

close: [Feature] Implement parallel loading of data #84

Main Changes

During graph loading step, introduce thread pool to parallel send vertices and edges to target Worker.

Verifying these changes

This change is already covered by existing tests, such as (please describe tests).

Does this PR potentially affect the following parts?

Documentation Status

Doc - TODO
Doc - Done
Doc - NO Need

coderzc

LGTM

javeme

tiny comments

javeme · 2023-05-25T15:44:12Z

computer-core/src/main/java/org/apache/hugegraph/computer/core/input/WorkerInputManager.java

        while (iterator.hasNext()) {
            Vertex vertex = iterator.next();
-            this.sendManager.sendEdge(vertex);
+            futures.add(CompletableFuture.runAsync(() -> consumer.accept(vertex),
+                    this.sendExecutor));


prefer to align with CompletableFuture

javeme · 2023-05-25T15:50:06Z

computer-core/src/main/java/org/apache/hugegraph/computer/core/input/WorkerInputManager.java

+                sendThreadNum);
+    }
+
+    private Integer inputLoaderThreadNum(Config config) {


inputLoaderThreadNum => inputSendThreadNum ?

Radeity · 2023-05-25T15:57:01Z

Hi, @javeme , I find that create a future for each vertex(edge) will cause OOM, I will change the implementation to parallel handle inputSplit tomorrow, thus, I draft this PR temporarily. I would appreciate it if you can help review after modification.

codecov · 2023-05-26T10:42:23Z

Codecov Report

Merging #248 (e5475dd) into master (f7d10a8) will increase coverage by 0.04%.
The diff coverage is 96.36%.

@@             Coverage Diff              @@
##             master     #248      +/-   ##
============================================
+ Coverage     85.78%   85.82%   +0.04%     
- Complexity     3230     3237       +7     
============================================
  Files           344      344              
  Lines         12076    12105      +29     
  Branches       1088     1090       +2     
============================================
+ Hits          10359    10389      +30     
+ Misses         1190     1188       -2     
- Partials        527      528       +1

Impacted Files	Coverage Δ
...egraph/computer/core/input/WorkerInputManager.java	`95.74% <93.33%> (-4.26%)`	⬇️
...ugegraph/computer/core/config/ComputerOptions.java	`98.88% <100.00%> (+0.01%)`	⬆️
...ugegraph/computer/core/compute/ComputeManager.java	`86.51% <100.00%> (ø)`
...gegraph/computer/core/worker/load/LoadService.java	`91.48% <100.00%> (+0.79%)`	⬆️

... and 6 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

javeme · 2023-05-26T12:38:14Z

computer-core/src/main/java/org/apache/hugegraph/computer/core/input/WorkerInputManager.java

+
+        this.loadServices = new ArrayList<>(this.sendThreadNum);
+        for (int i = 0; i < this.sendThreadNum; i++) {
+            this.loadServices.add(new LoadService(context));


seems don't need to create multi LoadService, just try calling LoadService.createIteratorFromVertex() with sendThreadNum times.

Hi, @javeme , thanks for your comment, I move fetcher into IteratorFromVertex and IteratorFromEdge, also replace creation of multi LoadService as you suggested.

computer-core/src/main/java/org/apache/hugegraph/computer/core/worker/load/LoadService.java

computer-core/src/main/java/org/apache/hugegraph/computer/core/input/WorkerInputManager.java

computer-core/src/main/java/org/apache/hugegraph/computer/core/worker/load/LoadService.java

javeme · 2023-05-27T13:21:25Z

computer-core/src/main/java/org/apache/hugegraph/computer/core/worker/load/LoadService.java

+         */
+        private GraphFetcher fetcher;
+
+        public IteratorFromVertex(Config config, InputSplitRpcService rpcService) {


can we just pass into the GraphFetcher fetcher arg

It seems that we can't, fetcher has property localBatch, different threads can not share the same localBatch.

we can create GraphFetcher when new IteratorFromVertex()

Do you mean like this?

public Iterator<Vertex> createIteratorFromVertex() { GraphFetcher fetcher = InputSourceFactory.createGraphFetcher(this.config, this.rpcService); return new IteratorFromVertex(fetcher); }

computer-api/src/main/java/org/apache/hugegraph/computer/core/config/ComputerOptions.java

javeme

LGTM

computer-core/src/main/java/org/apache/hugegraph/computer/core/input/WorkerInputManager.java

javeme · 2023-05-28T12:25:22Z

ci error:

Error:  Tests run: 563, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 218.399 s <<< FAILURE! - in org.apache.hugegraph.computer.suite.unit.UnitTestSuite
Error:  testRunAlgorithmFromHdfs(org.apache.hugegraph.computer.algorithm.centrality.degree.DegreeCentralityTest)  Time elapsed: 4.379 s  <<< FAILURE!
java.lang.AssertionError: [java.util.concurrent.CompletionException: org.apache.hugegraph.computer.core.common.exception.ComputerException: An exception occurred during parallel sending edges, null]
	at org.apache.hugegraph.computer.algorithm.centrality.degree.DegreeCentralityTest.testRunAlgorithmFromHdfs(DegreeCentralityTest.java:50)

[INFO] 
[INFO] Results:
[INFO] 
Error:  Failures: 
Error:    DegreeCentralityTest.testRunAlgorithmFromHdfs:50->AlgorithmTestBase.runAlgorithm:147 [java.util.concurrent.CompletionException: org.apache.hugegraph.computer.core.common.exception.ComputerException:
  An exception occurred during parallel sending edges, null]
[INFO] 
Error:  Tests run: 563, Failures: 1, Errors: 0, Skipped: 0

Radeity · 2023-05-28T16:33:59Z

ci error:

Error:  Tests run: 563, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 218.399 s <<< FAILURE! - in org.apache.hugegraph.computer.suite.unit.UnitTestSuite
Error:  testRunAlgorithmFromHdfs(org.apache.hugegraph.computer.algorithm.centrality.degree.DegreeCentralityTest)  Time elapsed: 4.379 s  <<< FAILURE!
java.lang.AssertionError: [java.util.concurrent.CompletionException: org.apache.hugegraph.computer.core.common.exception.ComputerException: An exception occurred during parallel sending edges, null]
	at org.apache.hugegraph.computer.algorithm.centrality.degree.DegreeCentralityTest.testRunAlgorithmFromHdfs(DegreeCentralityTest.java:50)

[INFO] 
[INFO] Results:
[INFO] 
Error:  Failures: 
Error:    DegreeCentralityTest.testRunAlgorithmFromHdfs:50->AlgorithmTestBase.runAlgorithm:147 [java.util.concurrent.CompletionException: org.apache.hugegraph.computer.core.common.exception.ComputerException:
  An exception occurred during parallel sending edges, null]
[INFO] 
Error:  Tests run: 563, Failures: 1, Errors: 0, Skipped: 0

Hi, @javeme , I've checked recent CI logs, the error is caused by IOException: Filesystem closed when reading data from HDFS and checking whether it's open. In addition, I find that some TransportException were thrown before, thus I guess maybe there's something wrong with CI environment?

Radeity · 2023-05-29T05:18:54Z

Sry, it's my own fault, I'm doing some code profiling now ...

Radeity · 2023-05-29T08:31:02Z

Hi, @javeme , I've careful debug, and find the real reason is that FileElementFetcher maintains inputReader, when the type of inputReader is HDFSFileReader, only one client for the same URI will be cached (according to the setting), also, if one thread close fetcher and inputReader, the client will be removed from the map, the other thread can not get this client and then throw IOException: Filesystem closed.

Thus, I use a fetcher array to manage all fetchers in LoadService, and close them together after sending all edges and vertices, PTAL. cc @coderzc @imbajin

javeme

Great, it's clear now.
Thanks for your analysis and feedback of the root cause.

computer-core/src/main/java/org/apache/hugegraph/computer/core/worker/load/LoadService.java

computer-core/src/main/java/org/apache/hugegraph/computer/core/input/WorkerInputManager.java

javeme · 2023-05-29T11:18:49Z

computer-core/src/main/java/org/apache/hugegraph/computer/core/worker/load/LoadService.java

+    public void init() {
+        assert this.rpcService != null;
+        for (int i = 0; i < this.fetcherNum; i++) {
+            this.fetchers[i] = InputSourceFactory.createGraphFetcher(this.config, this.rpcService);


could you please also add some comment for the reason why we hold a fetchers array

javeme

LGTM

coderzc

LGTM

coderzc assigned Radeity May 25, 2023

Radeity force-pushed the Feature-84 branch from ab8eb29 to 8ecb808 Compare May 25, 2023 08:39

coderzc requested review from coderzc, javeme and imbajin May 25, 2023 09:45

coderzc previously approved these changes May 25, 2023

View reviewed changes

Radeity dismissed coderzc’s stale review via e5335f5 May 25, 2023 11:22

coderzc previously approved these changes May 25, 2023

View reviewed changes

Radeity marked this pull request as draft May 25, 2023 12:47

javeme reviewed May 25, 2023

View reviewed changes

Radeity dismissed coderzc’s stale review via 19b9b11 May 26, 2023 10:24

Radeity force-pushed the Feature-84 branch from e5335f5 to 19b9b11 Compare May 26, 2023 10:24

javeme reviewed May 26, 2023

View reviewed changes

Radeity marked this pull request as ready for review May 27, 2023 09:27

coderzc reviewed May 27, 2023

View reviewed changes

computer-core/src/main/java/org/apache/hugegraph/computer/core/worker/load/LoadService.java Outdated Show resolved Hide resolved

Radeity force-pushed the Feature-84 branch from 97c64f8 to d0c18aa Compare May 27, 2023 12:15

javeme reviewed May 27, 2023

View reviewed changes

Radeity force-pushed the Feature-84 branch 2 times, most recently from ab11331 to 5f74db7 Compare May 27, 2023 13:55

feat: implement parallel send data in load graph step

ca8b0df

Radeity force-pushed the Feature-84 branch from 108d97c to ca8b0df Compare May 27, 2023 14:25

javeme previously approved these changes May 27, 2023

View reviewed changes

coderzc reviewed May 27, 2023

View reviewed changes

computer-core/src/main/java/org/apache/hugegraph/computer/core/input/WorkerInputManager.java Show resolved Hide resolved

call close()

5d4d4f6

Radeity dismissed javeme’s stale review via 5d4d4f6 May 28, 2023 02:45

imbajin reviewed May 28, 2023

View reviewed changes

computer-core/src/main/java/org/apache/hugegraph/computer/core/input/WorkerInputManager.java Outdated Show resolved Hide resolved

computer-core/src/main/java/org/apache/hugegraph/computer/core/input/WorkerInputManager.java Outdated Show resolved Hide resolved

Apply align suggestions from code review

53d5565

imbajin previously approved these changes May 28, 2023

View reviewed changes

improve: use AutoCloseableIterator

025f714

Radeity dismissed imbajin’s stale review via 025f714 May 28, 2023 14:23

Radeity marked this pull request as draft May 29, 2023 04:28

fix: Filesystem closed caused by cache only one HDFS client

5499f7e

Radeity marked this pull request as ready for review May 29, 2023 08:04

improve: reuse fetcher instance for vertex and edge sending

20aee89

javeme reviewed May 29, 2023

View reviewed changes

This comment was marked as duplicate.

Sign in to view

Radeity added 2 commits May 29, 2023 19:35

improve: remove blank line & add comment

ba8d55f

modify comment

e5475dd

javeme approved these changes May 29, 2023

View reviewed changes

coderzc approved these changes May 29, 2023

View reviewed changes

imbajin merged commit 7df3b43 into apache:master May 29, 2023

Radeity deleted the Feature-84 branch May 31, 2023 02:26

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: implement parallel send data in load graph step #248

feat: implement parallel send data in load graph step #248

Radeity commented May 25, 2023

coderzc left a comment

javeme left a comment

javeme May 25, 2023

javeme May 25, 2023

Radeity commented May 25, 2023 •

edited

Loading

codecov bot commented May 26, 2023 •

edited

Loading

javeme May 26, 2023

Radeity May 27, 2023

javeme May 27, 2023

Radeity May 27, 2023 •

edited

Loading

javeme May 27, 2023

Radeity May 27, 2023

javeme left a comment

javeme commented May 28, 2023

Radeity commented May 28, 2023

Radeity commented May 29, 2023

Radeity commented May 29, 2023 •

edited

Loading

javeme left a comment

javeme May 29, 2023

This comment was marked as duplicate.

javeme left a comment

coderzc left a comment

feat: implement parallel send data in load graph step #248

feat: implement parallel send data in load graph step #248

Conversation

Radeity commented May 25, 2023

Purpose of the PR

Main Changes

Verifying these changes

Does this PR potentially affect the following parts?

Documentation Status

coderzc left a comment

Choose a reason for hiding this comment

javeme left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Radeity commented May 25, 2023 • edited Loading

codecov bot commented May 26, 2023 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Radeity May 27, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

javeme left a comment

Choose a reason for hiding this comment

javeme commented May 28, 2023

Radeity commented May 28, 2023

Radeity commented May 29, 2023

Radeity commented May 29, 2023 • edited Loading

javeme left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

This comment was marked as duplicate.

javeme left a comment

Choose a reason for hiding this comment

coderzc left a comment

Choose a reason for hiding this comment

Radeity commented May 25, 2023 •

edited

Loading

codecov bot commented May 26, 2023 •

edited

Loading

Radeity May 27, 2023 •

edited

Loading

Radeity commented May 29, 2023 •

edited

Loading