Standardize bsp synchronization between master and worker #48
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Codecov Report
@@ Coverage Diff @@ ## master #48 +/- ## ============================================ + Coverage 84.56% 84.76% +0.19% - Complexity 1368 1384 +16 ============================================ Files 150 150 Lines 4530 4615 +85 Branches 376 378 +2 ============================================ + Hits 3831 3912 +81 - Misses 471 474 +3 - Partials 228 229 +1
Continue to review full report at Codecov.
|
javeme reviewed May 13, 2021
| LOG.info("Master waited all workers registered, workers: {}", | ||
| LOG.info("Master waited all workers init-done, workers: {}", | ||
| containers); | ||
| this.assignContainerId(containers); |
| * {@link #waitMasterAllInitDone()}. | ||
| */ | ||
| String path = this.constructPath(BspEvent.BSP_WORKER_INIT_DONE, | ||
| this.workerInfo.toString()); |
There was a problem hiding this comment.
define tmpWorkerId = this.workerInfo.host+port and comment why it's ok
| LOG.info("Worker({}) waited master all-registered, workers: {}", | ||
| String path = this.constructPath(BspEvent.BSP_MASTER_ALL_INIT_DONE); | ||
| byte[] serializedContainers = this.bspClient().get( | ||
| path, |
| */ | ||
| public int waitMasterSuperstepResume() { | ||
| LOG.info("Worker({}) is waiting for master superstep-resume", | ||
| public int waitMasterResume() { |
| if (this.workerInfo.equalsExceptId(container)) { | ||
| this.workerInfo.id(container.id()); | ||
| LOG.info("Worker({}) assigned id {} from master", | ||
| this.workerInfo, this.workerInfo.id()); |
There was a problem hiding this comment.
this.workerInfo.uniqueName()
| // Note: The workerInfo in Bsp4Worker is the same object in WorkerService. | ||
| private void assignThisWorkerId(List<ContainerInfo> workersFromMaster) { | ||
| for (ContainerInfo container : workersFromMaster) { | ||
| if (this.workerInfo.equalsExceptId(container)) { |
There was a problem hiding this comment.
compare this.workerInfo.uniqueName()
| this.bsp4Master.masterStepComputeDone(superstep); | ||
| List<WorkerStat> workerStats = | ||
| this.bsp4Master.waitWorkersSuperstepDone(superstep); | ||
| this.bsp4Master.waitWorkersStepDone(superstep); |
| } | ||
| | ||
| public static <V extends Readable> List<V> fromBytes(byte[] bytes, | ||
| ObjectFactory<V> factory) { |
javeme reviewed May 13, 2021
| BSP_WORKER_INIT_DONE(2, "/worker"), | ||
| BSP_MASTER_ALL_INIT_DONE(3, "/master/all_init_done"), | ||
| BSP_MASTER_RESUME(4, "/master/resume"), | ||
| BSP_MASTER_RESUME_DONE(4, "/master/resume"), |
There was a problem hiding this comment.
keep the same style: remove "_done" suffix
| this.rpcPort == other.rpcPort && | ||
| this.dataPort == other.dataPort; | ||
| public String uniqueName() { | ||
| return String.format("%s:%s", this.hostname, this.dataPort); |
houzhizhen changed the title houzhizhen changed the title javeme previously approved these changes May 13, 2021
…() instead of workerInfo.id()
javeme approved these changes May 13, 2021
Linary approved these changes May 13, 2021
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit. This suggestion is invalid because no changes were made to the code. Suggestions cannot be applied while the pull request is closed. Suggestions cannot be applied while viewing a subset of changes. Only one suggestion per line can be applied in a batch. Add this suggestion to a batch that can be applied as a single commit. Applying suggestions on deleted lines is not supported. You must change the existing code in this line in order to create a valid suggestion. Outdated suggestions cannot be applied. This suggestion has been applied or marked resolved. Suggestions cannot be applied from pending reviews. Suggestions cannot be applied on multi-line comments. Suggestions cannot be applied while the pull request is queued to merge. Suggestion cannot be applied right now. Please check back later.
unify bsp key to xxxDONE
master set workers info after all workers inited and assign worker id
add compute_done signal between master and workers for every superstep
worker set close_done signal after close the managers
worker signal bsp event BSP_WORKER_INIT_DONE with workerInfo.toString() instead of workerInfo.id()