implement aggregator module #46

javeme · 2021-05-08T05:58:00Z

No description provided.

codecov · 2021-05-08T22:52:58Z

Codecov Report

Merging #46 (6c41731) into master (b1408eb) will increase coverage by 0.21%.
The diff coverage is 89.89%.

@@ Coverage Diff @@ ## master #46 +/- ## ============================================ + Coverage 85.54% 85.76% +0.21%  - Complexity 1429 1490 +61  ============================================ Files 153 156 +3 Lines 4671 4918 +247 Branches 381 405 +24 ============================================ + Hits 3996 4218 +222  - Misses 458 465 +7  - Partials 217 235 +18

Impacted Files	Coverage Δ	Complexity Δ
...hugegraph/computer/core/aggregator/Aggregator.java	`0.00% <ø> (ø)`	`0.00 <0.00> (ø)`
...ugegraph/computer/core/config/ComputerOptions.java	`99.36% <ø> (-0.02%)`	`2.00 <0.00> (ø)`
...ugegraph/computer/core/aggregator/Aggregators.java	`79.16% <79.16%> (ø)`	`8.00 <8.00> (?)`
...ph/computer/core/aggregator/DefaultAggregator.java	`82.85% <82.85%> (ø)`	`22.00 <22.00> (?)`
.../hugegraph/computer/core/master/MasterService.java	`91.30% <87.17%> (+1.92%)`	`16.00 <2.00> (+1.00)`
.../computer/core/aggregator/RegisterAggregators.java	`90.90% <90.90%> (ø)`	`10.00 <10.00> (?)`
...ph/computer/core/aggregator/MasterAggrManager.java	`95.65% <95.55%> (-4.35%)`	`11.00 <10.00> (+8.00)`	⬇️
...ph/computer/core/aggregator/WorkerAggrManager.java	`95.65% <95.55%> (-4.35%)`	`13.00 <12.00> (+10.00)`	⬇️
.../hugegraph/computer/core/worker/WorkerService.java	`91.26% <96.66%> (+3.08%)`	`14.00 <2.00> (+1.00)`
.../baidu/hugegraph/computer/core/bsp/Bsp4Master.java	`100.00% <100.00%> (ø)`	`22.00 <0.00> (ø)`
... and 15 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update b1408eb...6c41731. Read the comment docs.

houzhizhen · 2021-05-10T02:28:22Z

computer-api/src/main/java/com/baidu/hugegraph/computer/core/aggregator/Aggregator4Worker.java

+ * master. Can be called in each superstep.
+ * Throws ComputerException if master does not register the aggregator
+ * with specified name.
+ * @param value The value to be aggregated


remove this line

houzhizhen · 2021-05-10T02:29:25Z

computer-api/src/main/java/com/baidu/hugegraph/computer/core/aggregator/Aggregator4Worker.java

 /**
- * Set aggregate value after superstep. The value will be sent to
+ * Create aggregator by name in worker, the aggregator is registered by
+ * master. Can be called in each superstep.


Is master and master-computation interchangeable?

you mean aslo add createAggregator() to Aggregator4Master?

the aggregator is registered by master-computation.

houzhizhen · 2021-05-10T02:31:53Z

computer-api/src/main/java/com/baidu/hugegraph/computer/core/graph/value/BooleanValue.java


+ @Override
+ public BooleanValue copy() {
+ return new BooleanValue(this.value);


It is better if to create new object.
return this.value ? TRUE : FALSE;

houzhizhen · 2021-05-10T02:55:57Z

computer-core/src/main/java/com/baidu/hugegraph/computer/core/aggregator/MasterAggrManager.java

+ @Override
+ public <V extends Value<?>> Aggregator<V> getAggregator(String name) {
+ Aggregator<?> aggr = this.aggregators.get(name, null);
+ assert aggr != null;


Throw exception to the client.

aggregators.get() would throw if not found

computer-core/src/main/java/com/baidu/hugegraph/computer/core/aggregator/WorkerAggrManager.java

houzhizhen · 2021-05-10T03:43:34Z

computer-test/src/main/java/com/baidu/hugegraph/computer/core/worker/MockComputation.java

+ this.aggrFloat.aggregatedValue());
+
+ this.aggrFloat.aggregateValue(1f);
+ Assert.assertEquals(new FloatValue(1),


houzhizhen · 2021-05-10T03:43:46Z

computer-test/src/main/java/com/baidu/hugegraph/computer/core/worker/MockComputation.java

+ Assert.assertEquals(new FloatValue(1),
+ this.aggrFloat.aggregatedValue());
+
+ this.aggrFloat.aggregateValue(new FloatValue(1));


houzhizhen · 2021-05-10T03:47:32Z

computer-test/src/main/java/com/baidu/hugegraph/computer/core/worker/MockComputation2.java

+ Assert.assertEquals(new LongValue(8L), context.aggregatedValue(
+ MockMasterComputation.AGGR_TEST_LONG_MAX));
+
+ Assert.assertEquals(new DoubleValue(20.8), context.aggregatedValue(


20.8D and ditto in the following lines.

double value 20.8 is ok

houzhizhen · 2021-05-10T03:50:01Z

computer-test/src/main/java/com/baidu/hugegraph/computer/core/worker/MockMasterComputation.java

+ Assert.assertEquals(new LongValue(8L), context.aggregatedValue(
+ MockMasterComputation.AGGR_TEST_LONG_MAX));
+
+ Assert.assertEquals(new DoubleValue(10.4), context.aggregatedValue(


10.4D and ditto in following lines

houzhizhen · 2021-05-10T03:52:22Z

...uter-test/src/main/java/com/baidu/hugegraph/computer/core/worker/MockMasterComputation2.java

+ Assert.assertEquals(new LongValue(8L), context.aggregatedValue(
+ MockMasterComputation2.AGGR_TEST_LONG_MAX));
+
+ Assert.assertEquals(new DoubleValue(20.8), context.aggregatedValue(


javeme · 2021-05-10T07:12:42Z

wip: add test case

javeme · 2021-05-13T08:14:21Z

computer-core/src/main/java/com/baidu/hugegraph/computer/core/config/ComputerOptions.java

 TimeUnit.SECONDS.toMillis(30L)
 );

+ public static final ConfigOption<Integer> WORKER_ID =


let master assign id for workers

javeme · 2021-05-13T08:15:11Z

computer-core/src/main/java/com/baidu/hugegraph/computer/core/master/MasterService.java

+ * get aggregators from master.
+ */
+ MasterAggrManager manager = this.managers.get(MasterAggrManager.NAME);
+ manager.applyAggregators();


add manager.inited() event

javeme · 2021-05-13T08:17:21Z

computer-test/src/main/java/com/baidu/hugegraph/computer/core/worker/MockComputation2.java

+ Assert.assertEquals(new LongValue(8L), context.aggregatedValue(
+ MockMasterComputation.AGGR_TEST_LONG_MAX));
+
+ Assert.assertEquals(new DoubleValue(20.8), context.aggregatedValue(


double value 20.8 is ok

houzhizhen · 2021-05-13T09:56:33Z

computer-test/src/main/java/com/baidu/hugegraph/computer/core/worker/WorkerServiceTest.java

 MasterService masterService = new MasterService();
 WorkerService workerService = new MockWorkerService();
+
+// Config configm = UnitTestBase.updateWithRequiredOptions(


remove unused line and ditto in the following lines

houzhizhen · 2021-05-13T09:56:48Z

computer-test/src/main/java/com/baidu/hugegraph/computer/core/worker/WorkerServiceTest.java

 CountDownLatch countDownLatch = new CountDownLatch(3);
 Throwable[] exceptions = new Throwable[3];

+// Config configm = UnitTestBase.updateWithRequiredOptions(


remove unused line and ditto in the following lines

also add two-tier structure to store current-step and previous-step aggregators

houzhizhen · 2021-05-20T02:30:25Z

...ter-core/src/main/java/com/baidu/hugegraph/computer/core/aggregator/RegisterAggregators.java

+ this.aggregators.put(name, aggregator);
+ }
+
+ public Aggregator<? extends Value<?>> get(String name) {


rename to copy to consist with #copyAll

houzhizhen · 2021-05-20T02:53:37Z

computer-core/src/main/java/com/baidu/hugegraph/computer/core/worker/WorkerService.java

 // Create managers
 WorkerRpcManager rpcManager = new WorkerRpcManager();
 this.managers.add(rpcManager);
+ /*


set master's rpc address to config, and remove the comments.
config.hugeConfig().setProperty(RpcOptions.RPC_REMOTE_URL.

it's not the best way

houzhizhen · 2021-05-20T02:54:04Z

computer-core/src/main/java/com/baidu/hugegraph/computer/core/worker/WorkerService.java

+ * config.hugeConfig().setProperty(this.masterInfo.rpcPort())
+ * NOTE: this init() method will be called twice, ignore the 2nd call
+ */
 rpcManager.init(this.config);


It seems rpcManager.init need't be called twice.

it must init before calling rpcManager.inputSplitService()

houzhizhen · 2021-05-20T02:56:14Z

computer-test/src/main/java/com/baidu/hugegraph/computer/core/worker/MockComputation.java

 import com.baidu.hugegraph.computer.core.graph.vertex.Vertex;
 import com.baidu.hugegraph.testutil.Assert;

 public class MockComputation implements Computation<DoubleValue> {


Rename to MockAggregatorComputation? We be need different type of computation test.

houzhizhen · 2021-05-20T02:57:13Z

computer-test/src/main/java/com/baidu/hugegraph/computer/core/worker/MockComputation2.java

+import com.baidu.hugegraph.computer.core.graph.value.LongValue;
+import com.baidu.hugegraph.testutil.Assert;
+
+public class MockComputation2 extends MockComputation {


MockAggregatorComputation2 ?

there is still other test cases

houzhizhen · 2021-05-20T03:05:48Z

computer-test/src/main/java/com/baidu/hugegraph/computer/core/worker/WorkerServiceTest.java


 pool.submit(() -> {
 Config config = UnitTestBase.updateWithRequiredOptions(
 RpcOptions.RPC_REMOTE_URL, "127.0.0.1:8090",


Can remove this line if get hostname and port from masterInfo

houzhizhen · 2021-05-20T03:07:32Z

computer-test/src/main/java/com/baidu/hugegraph/computer/core/worker/WorkerServiceTest.java


 pool.submit(() -> {
 Config config = UnitTestBase.updateWithRequiredOptions(
 RpcOptions.RPC_SERVER_HOST, "localhost",


add config ' RpcOptions.RPC_REMOTE_URL, "127.0.0.1:0" ' to listen on random port

houzhizhen · 2021-05-20T03:09:13Z

computer-test/src/main/java/com/baidu/hugegraph/computer/core/worker/MockMasterComputation.java

+import com.baidu.hugegraph.computer.core.master.MasterContext;
 import com.baidu.hugegraph.testutil.Assert;

 public class MockMasterComputation extends DefaultMasterComputation {


rename MockMasterAggregatorComputation ?

there is still other test cases

Linary · 2021-05-20T09:31:20Z

computer-core/src/main/java/com/baidu/hugegraph/computer/core/aggregator/DefaultAggregator.java

+ this.combineAndSwapIfNeeded(localValue, this.value);
+ }
+
+ private void combineAndSwapIfNeeded(V localValue, V value2) {


What mean? doesn't need value2?

Linary · 2021-05-20T09:46:07Z

computer-core/src/main/java/com/baidu/hugegraph/computer/core/aggregator/WorkerAggrManager.java

+ public <V extends Value<?>> Aggregator<V> createAggregator(String name) {
+ /*
+ * Create aggregator for the current superstep,
+ * Generally called when computation.beforeSuperstep()


Is this method called every time when computation.beforeSuperstep()

yes, this method would be called once per superstep for each aggregator

Linary · 2021-05-20T09:46:15Z

computer-core/src/main/java/com/baidu/hugegraph/computer/core/aggregator/WorkerAggrManager.java

+ public <V extends Value<?>> void aggregateValue(String name, V value) {
+ /*
+ * Update aggregator for the current superstep,
+ * Generally called when computation.afterSuperstep()


Linary · 2021-05-20T09:51:17Z

computer-core/src/main/java/com/baidu/hugegraph/computer/core/config/ComputerOptions.java

- "The server bind port, if it is zero " +
- "will let the system pick up an ephemeral port.",
+ "The server port to listen on to transfer data. " +
+ "The system will assign a random port if server_port=0.",


server_port -> transport.server_port

Linary · 2021-05-20T09:52:27Z

computer-core/src/main/java/com/baidu/hugegraph/computer/core/manager/Manager.java


 /**
- * Used to add the resources the computation needed. Be called only one
+ * Used to add the resources needed by the computation. Be called only one


Be called ... move to line 39

Linary · 2021-05-20T13:17:53Z

computer-core/src/main/java/com/baidu/hugegraph/computer/core/aggregator/WorkerAggrManager.java

+ public WorkerAggrManager(ComputerContext context) {
+ this.context = context;
+ this.service = null;
+ this.registerAggregators = new RegisterAggregators();


seems no need to new RegisterAggregators(); it will be assigned at init()

just to avoid the potential NPE

Linary · 2021-05-21T04:01:45Z

computer-core/src/main/java/com/baidu/hugegraph/computer/core/aggregator/WorkerAggrManager.java

+ // Registered aggregators from master
+ private RegisterAggregators registerAggregators;
+ // Cache the aggregators of the previous superstep
+ private Map<String, Value<?>> lastAggregators;


What is the function of lastAggregator values? I found that it was not involved in the calculation, just get the value.

used by https://github.com/hugegraph-ee/hugegraph-computer/pull/46/files/13620f95b694cac38a2bf0edbd16dee9a5c40e75#diff-91ebc1ed9fd5e9336d92accfdc06c4a11682c14c71a4e558a537c982d2856450R362

javeme · 2021-05-21T06:43:26Z

computer-core/src/main/java/com/baidu/hugegraph/computer/core/aggregator/WorkerAggrManager.java

+ // Registered aggregators from master
+ private RegisterAggregators registerAggregators;
+ // Cache the aggregators of the previous superstep
+ private Map<String, Value<?>> lastAggregators;


used by https://github.com/hugegraph-ee/hugegraph-computer/pull/46/files/13620f95b694cac38a2bf0edbd16dee9a5c40e75#diff-91ebc1ed9fd5e9336d92accfdc06c4a11682c14c71a4e558a537c982d2856450R362

javeme · 2021-05-21T06:45:20Z

computer-core/src/main/java/com/baidu/hugegraph/computer/core/aggregator/WorkerAggrManager.java

+ // Cache the aggregators of the previous superstep
+ private Map<String, Value<?>> lastAggregators;
+ // Cache the aggregators of the current superstep
+ private Aggregators currentAggregators;


used by https://github.com/hugegraph-ee/hugegraph-computer/pull/46/files/13620f95b694cac38a2bf0edbd16dee9a5c40e75#diff-91ebc1ed9fd5e9336d92accfdc06c4a11682c14c71a4e558a537c982d2856450R145

houzhizhen reviewed May 10, 2021

View reviewed changes

houzhizhen previously approved these changes May 10, 2021

View reviewed changes

javeme dismissed houzhizhen’s stale review via 9b08c2e May 13, 2021 07:46

javeme commented May 13, 2021

View reviewed changes

houzhizhen reviewed May 13, 2021

View reviewed changes

javeme force-pushed the implement-aggregator branch 2 times, most recently from 6d2fc14 to 0245bd2 Compare May 13, 2021 15:32

javeme added 19 commits May 19, 2021 20:02

implement aggregator module

94d4f22

add DefaultAggregator.newValue()

3a9d2b5

integrate aggregator into master-service and worker-service

e1d6812

ensure worker init() executed after master

02c8191

add rpc interface registeredAggregators()

99acf6a

also add two-tier structure to store current-step and previous-step aggregators

add repair() for Aggregator

3acffaa

let master context use config of outer class

00762f5

fix test error

a54d9ad

add aggregator test

1d32448

test for multi workers

3479657

fix style

e75b814

tiny fix

25c0b1e

support register aggregator with default value

51cd76d

add more error check and fix/improve some code

4a63796

add more test and error test

acdc777

fix WorkerServiceTest process to improve fault tolerance

9b1646d

remove master clean() code before test (done in master init())

431486a

add Manager.inited()

54e7833

remove worker_id option and use server_port instead

5493b60

javeme force-pushed the implement-aggregator branch from 0245bd2 to 5493b60 Compare May 19, 2021 12:13

houzhizhen reviewed May 20, 2021

View reviewed changes

use rpc-url from master-info

6e49914

javeme force-pushed the implement-aggregator branch from 1f9cc6d to 6e49914 Compare May 20, 2021 06:13

houzhizhen previously approved these changes May 20, 2021

View reviewed changes

Linary requested changes May 20, 2021

View reviewed changes

tiny improve

13620f9

javeme dismissed houzhizhen’s stale review via 13620f9 May 20, 2021 11:54

Linary requested changes May 20, 2021

View reviewed changes

Linary requested changes May 21, 2021

View reviewed changes

javeme commented May 21, 2021

View reviewed changes

Linary previously approved these changes May 21, 2021

View reviewed changes

javeme dismissed Linary’s stale review via d6e000d May 21, 2021 07:39

improve interface java-doc

6c41731

javeme force-pushed the implement-aggregator branch from d6e000d to 6c41731 Compare May 21, 2021 07:41

Linary approved these changes May 21, 2021

View reviewed changes

houzhizhen approved these changes May 21, 2021

View reviewed changes

houzhizhen merged commit a087d27 into master May 21, 2021

houzhizhen deleted the implement-aggregator branch May 21, 2021 07:50

implement aggregator module #46

implement aggregator module #46

Uh oh!

Conversation

javeme commented May 8, 2021

codecov bot commented May 8, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

javeme May 10, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

javeme commented May 10, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

houzhizhen May 13, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

houzhizhen May 13, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Labels

3 participants

codecov bot commented May 8, 2021 •

edited

Loading

javeme May 10, 2021 •

edited

Loading

houzhizhen May 13, 2021 •

edited

Loading

houzhizhen May 13, 2021 •

edited

Loading