created guard for split task

Author: Tale152

src/interconnected_node/broker_service_socket/handlers/applyOnRecruitmentAcceptHandler.ts

@@ -28,8 +28,9 @@ export default function applyOnRecruitmentAcceptHandler(
           });
         })
         .setOnMessageHandler((msg: any) => {
-          // TODO implement actual handler
-          console.log('Slave sent this message: ' + msg);
+          jobsRepository
+            .get(payload.operationId)
+            ?.notifyNewMessage(masterP2PConnection, msg);
         })
         .setOnDisconnectionHandler(() => {
           const masterConnection = masterP2PConnectionsHub.getBySlaveId(

src/interconnected_node/contribution/common/Job.ts

@@ -13,4 +13,9 @@ export default interface Job {
   notifyNewMasterP2PConnection(
     masterP2PConnection: MasterP2PConnection
   ): Promise<void>;
+
+  notifyNewMessage(
+    masterP2PConnection: MasterP2PConnection,
+    msg: any
+  ): Promise<void>;
 }

src/interconnected_node/contribution/mapreduce/jobs/MapReduceMasterJob.ts

@@ -16,6 +16,8 @@ export default class MapReduceMasterJob implements Job {
   private currentUsedMapWorkerIndex: number;
   private mapWorkersToReach: number;
   private mapWorkers: Array<MasterP2PConnection>;
+  private mapWorkersJobAcks: Map<string, boolean>;
+  private mapWorkersJobGuard: boolean;
   private mapWorkersPerRegion: number;
   private reduceWorkersToReach: number;
   private reduceWorkers: Array<MasterP2PConnection>;
@@ -35,6 +37,8 @@ export default class MapReduceMasterJob implements Job {
       Math.random() * this.mapWorkersToReach
     );
     this.mapWorkers = new Array<MasterP2PConnection>();
+    this.mapWorkersJobAcks = new Map();
+    this.mapWorkersJobGuard = false;
     this.reduceWorkersToReach = params.reduceWorkers;
     this.reduceWorkers = new Array<MasterP2PConnection>();
     this.mapFunction = params.mapFunction;
@@ -65,8 +69,10 @@ export default class MapReduceMasterJob implements Job {
     masterP2PConnection: MasterP2PConnection
   ): Promise<void> {
     this.mapWorkers.push(masterP2PConnection);
+    this.mapWorkersJobAcks.set(masterP2PConnection.slaveId, false);
     if (this.mapWorkers.length === this.mapWorkersToReach) {
       this.status = Status.REDUCE_WORKERS_RECRUITMENT;
+      console.log('MAP WORKERS RECRUITED');
       const switchToReduceWorkersRecruitmentInterval = setInterval(() => {
         if (this.mapWorkers.every((mw) => mw.remoteDescription !== undefined)) {
           this.brokerServiceSocket.emit(
@@ -78,6 +84,12 @@ export default class MapReduceMasterJob implements Job {
               masterRole: 'NODE',
             }
           );
+          while (this.enqueuedTasks.length > 0) {
+            const poppedTask = this.enqueuedTasks.pop();
+            if (poppedTask !== undefined) {
+              this.executeSplit(poppedTask);
+            }
+          }
           clearInterval(switchToReduceWorkersRecruitmentInterval);
         }
       }, 100);
@@ -105,12 +117,6 @@ export default class MapReduceMasterJob implements Job {
     if (this.mapWorkers.length === this.mapWorkersToReach) {
       this.status = Status.RECRUITMENT_COMPLETED;
       console.log('RECRUITMENT COMPLETED');
-      while (this.enqueuedTasks.length > 0) {
-        const poppedTask = this.enqueuedTasks.pop();
-        if (poppedTask !== undefined) {
-          this.executeSplit(poppedTask);
-        }
-      }
     }
     return new Promise<void>((resolve) => {
       masterP2PConnection.sendMessage(
@@ -154,14 +160,20 @@ export default class MapReduceMasterJob implements Job {
     const mwUsed = new Array<MasterP2PConnection>();
     const remainingMWsFromIndex =
       this.mapWorkers.length - this.currentUsedMapWorkerIndex;
+    console.log('length ' + this.mapWorkers.length.toString());
+    console.log('index ' + this.currentUsedMapWorkerIndex.toString());
+    console.log('remaining ' + remainingMWsFromIndex.toString());
+    console.log('per region ' + this.mapWorkersPerRegion);
     if (remainingMWsFromIndex >= this.mapWorkersPerRegion) {
+      console.log('A');
       mwUsed.push(
         ...this.mapWorkers.slice(
           this.currentUsedMapWorkerIndex,
           this.mapWorkersPerRegion
         )
       );
     } else {
+      console.log('B');
       mwUsed.push(
         ...this.mapWorkers.slice(
           this.currentUsedMapWorkerIndex,
@@ -175,28 +187,58 @@ export default class MapReduceMasterJob implements Job {
         )
       );
     }
+    console.log(mwUsed.length.toString());
     this.currentUsedMapWorkerIndex += this.mapWorkersPerRegion;
     if (this.currentUsedMapWorkerIndex >= this.mapWorkers.length) {
       this.currentUsedMapWorkerIndex =
         this.mapWorkersPerRegion - remainingMWsFromIndex;
     }
-    task.execute(
-      { mapWorkers: mwUsed },
-      () => {
-        console.log('MASTER COMPLETED SLICE TASK');
-      },
-      () => {
-        'MASTER HAD AN ERROR COMPLETING SLICE TASK';
+    const interval = setInterval(() => {
+      if (this.mapWorkersJobGuard === true) {
+        clearInterval(interval);
+        task.execute(
+          { mapWorkers: mwUsed },
+          () => {
+            console.log('MASTER COMPLETED SLICE TASK');
+          },
+          () => {
+            'MASTER HAD AN ERROR COMPLETING SLICE TASK';
+          }
+        );
       }
-    );
+    }, 100);
   }
 
   enqueueTask(task: Task): Promise<boolean> {
     if (this.status === Status.MAP_WORKERS_RECRUITMENT) {
+      console.log('RECEIVED TASK, BUT ENQUEUED IT');
       this.enqueuedTasks.push(task);
     } else {
       this.executeSplit(task);
     }
     return new Promise<boolean>((resolve) => resolve(true));
   }
+
+  notifyNewMessage(
+    masterP2PConnection: MasterP2PConnection,
+    msg: any
+  ): Promise<void> {
+    const parsedMsg = JSON.parse(msg);
+    if (
+      parsedMsg.channel === 'START_JOB' &&
+      parsedMsg.payload.name === 'MAP_WORKER' &&
+      parsedMsg.payload.result === 'ACK'
+    ) {
+      this.mapWorkersJobAcks.set(masterP2PConnection.slaveId, true);
+      if (
+        this.mapWorkersJobAcks.size === this.mapWorkersToReach &&
+        Array.from(this.mapWorkersJobAcks.values()).every((s) => {
+          return s === true;
+        })
+      ) {
+        this.mapWorkersJobGuard = true;
+      }
+    }
+    return new Promise<void>((resolve) => resolve());
+  }
 }

src/interconnected_node/contribution/mapreduce/jobs/MapWorkerJob.ts

@@ -44,4 +44,11 @@ export default class MapWorkerJob implements Job {
       resolve();
     });
   }
+
+  notifyNewMessage(
+    masterP2PConnection: MasterP2PConnection,
+    msg: any
+  ): Promise<void> {
+    throw new Error('Method not implemented.');
+  }
 }

src/interconnected_node/contribution/mapreduce/jobs/ReduceWorkerJob.ts

@@ -41,4 +41,11 @@ export default class ReduceWorkerJob implements Job {
       resolve();
     });
   }
+
+  notifyNewMessage(
+    masterP2PConnection: MasterP2PConnection,
+    msg: any
+  ): Promise<void> {
+    throw new Error('Method not implemented.');
+  }
 }

src/interconnected_node/contribution/mapreduce/tasks/MapReduceRegionSplitsTask.ts

@@ -18,6 +18,9 @@ export default class MapReduceRegionSplitsTask implements Task {
     return new Promise<void>((resolve) => {
       try {
         const mapWorkers: MasterP2PConnection[] = jobParams.mapWorkers;
+        console.log(
+          'SPLIT TASK map workers length' + mapWorkers.length.toString()
+        );
         const roundedUpSplitsPerWorker = Math.ceil(
           this.splits.length / mapWorkers.length
         );

src/interconnected_node/p2p/message_handlers/slave/handleSlaveP2PConnectionMessage.ts

@@ -12,6 +12,7 @@ export default function handleSlaveP2PConnectionMessage(
   jobsRepository: JobsRepository
 ) {
   const parsedMsg = JSON.parse(msg);
+  console.log('channel: ' + parsedMsg.channel);
   switch (parsedMsg.channel) {
     case 'START_JOB':
       onSlaveStartJobMessageHandler(
@@ -28,6 +29,7 @@ export default function handleSlaveP2PConnectionMessage(
         slaveP2PConnection,
         jobsRepository
       );
+      break;
     default:
       throw new Error('Unrecognized P2P message received');
   }

src/interconnected_node/p2p/message_handlers/slave/handlers/onSlaveStartJobMessageHandler.ts

@@ -20,6 +20,7 @@ export default function onSlaveStartJobMessageHandler(
   const response = {
     channel: 'START_JOB',
     payload: {
+      name: payload.name,
       result: 'ACK',
     },
   };