Reduce between worker nodes, with network traffic (if greater than 1 nodes)
After all reduce()'s are done, postGlobal() will be called
@param drt task to reduce
333122
@Override
  public void reduce(DeepWaterTask2 drt) {
    if (_res == null) _res = drt._res;
    else {
//      _res._chunk_node_count += drt._res._chunk_node_count;
      _res.model_info().add(drt._res.model_info()); //add models, but don't average yet
    }
    assert(_res.model_info().get_params()._replicate_training_data);
  }
543957
public void runPostProcessing(boolean multiRoot) {
        // make identical vertices have the same set of properties
        // two vertices are identical if they have the same address value.
        // Id for every vertex is unique, so this is not a flag here
        for (Set<ExecutionGraphVertex> vs : this.addressToVertices.values()) {

            // the logic below is very fragile and based on some empirical model
            // and may not work for other type of graphs
            if (vs != null && vs.size() > 1) {
                for (ExecutionGraphVertex vertex : vs) {
                    if (CELL_WITH_FORMULA == vertex.properties().getType() || null != vertex.properties().getAlias()) {
                        copyProperties(vertex, vs);
                        break;
                    }
                }
            }
        }

        // copy or link subgraphs to identical vertices and modify Formula field with additional values
        Map<String, AtomicInteger> adressToCount = new HashMap<>();

        for (ExecutionGraphVertex vertex : this.graph.getVertices()) {

            // restore/add subgraphs to identical vertices
            Type type = vertex.properties().getType();

            if (isCell(type)) {
                String address = vertex.properties().getName();

                adressToCount.putIfAbsent(address, new AtomicInteger(0));

                if (adressToCount.get(address).incrementAndGet() > 1) { // count > 1
                    // need to link
                    Set<ExecutionGraphVertex> subgraphTops = new HashSet<>();

                    this.graph.getVertices().stream()
                                            .filter(v -> address.equals(v.properties.getName())) // check for subgraph
                                            .forEach(v -> this.graph.getIncomingEdgesOf(v).forEach(e -> subgraphTops.add(this.graph.getEdgeSource(e))));

                    for (ExecutionGraphVertex subVertex : subgraphTops) {
                        if (!this.addressToVertices.containsKey(address)) { continue; }
                        this.addressToVertices.get(address).forEach(v -> this.graph.addEdge(subVertex, v));
                    }
                }
            }

            if (RANGE == type) { connectValuesToRange(vertex, this); }

            /* Adding IF Value */
            if (IF == type) {
                Collection<ExecutionGraphEdge> two = this.graph.getIncomingEdgesOf(vertex);
                if (two.size() != 2) { throw new CalculationEngineException("IF must have only two incoming edges."); }

                Object ifBranchValue = null;
                for (ExecutionGraphEdge e : two) {
                    ExecutionGraphVertex oneOfTwo = this.graph.getEdgeSource(e);
                    if (!isCompareOperand(oneOfTwo.getName())) {
                        ifBranchValue = oneOfTwo.properties().getValue();
                        break;
                    }
                }
                vertex.properties().setValue(ifBranchValue);
            }
        }

        if (this.config.getDuplicatesNumberThreshold() != -1) { removeAllDuplicates(this); }
    }
703752
void performStreamingRepair()
    {
        InetAddress local = FBUtilities.getBroadcastAddress();
        // We can take anyone of the node as source or destination, however if one is localhost, we put at source to avoid a forwarding
        InetAddress src = r2.endpoint.equals(local) ? r2.endpoint : r1.endpoint;
        InetAddress dst = r2.endpoint.equals(local) ? r1.endpoint : r2.endpoint;

        SyncRequest request = new SyncRequest(desc, local, src, dst, differences);
        StreamingRepairTask task = new StreamingRepairTask(desc, request);
        task.run();
    }
1241746
@Override
  protected void postGlobal() {
    assert(_res.model_info().get_params()._replicate_training_data);
    super.postGlobal();
    // model averaging (DeepLearningTask only computed the per-node models, each on all the data)
    _res.model_info().div(_res._chunk_node_count);
    _res.model_info().add_processed_global(_res.model_info().get_processed_local()); //switch from local counters to global counters
    _res.model_info().set_processed_local(0l);
    DeepLearningModelInfo nodeAverageModel = _res.model_info();
    if (nodeAverageModel.get_params()._elastic_averaging)
      _sharedmodel = DeepLearningModelInfo.timeAverage(nodeAverageModel);
    else
      _sharedmodel = nodeAverageModel;
  }
38886
public void distributeAfter(int status)
    {
        if (tc.isEntryEnabled()) Tr.entry(tc, "distributeAfter", new Object[] { this, status});
        
        // Issue the RRS syncs first - these need to be as close to the completion as possible
        final List RRSsyncs = _syncs[SYNC_TIER_RRS];
                
        if (RRSsyncs != null)
        {       
            final int RRSstatus = (status == Status.STATUS_UNKNOWN ? Status.STATUS_COMMITTED : status); // @281425A
            for (int j = RRSsyncs.size(); --j >= 0;)
            {
                final Synchronization sync = (Synchronization)RRSsyncs.get(j);
    
                try
                {
                    if (tc.isEntryEnabled()) Tr.event(tc, "driving RRS after sync[" + j + "]", Util.identity(sync));
                    sync.afterCompletion(RRSstatus); // @281425C
                }
                catch (Throwable exc)
                {
                    // No FFDC Code Needed.
    
                    // Discard any exceptions at this point.
                    Tr.error(tc, "WTRN0074_SYNCHRONIZATION_EXCEPTION", new Object[] {"after_completion", exc});
                }
            }
        }
        
        coreDistributeAfter(status);

        if (tc.isEntryEnabled()) Tr.exit(tc, "distributeAfter");
    }
799976
void sendDeltaDoWorkAccepted()
   {
      doWorkAccepted.incrementAndGet();

      if (trace)
         log.tracef("sendDeltaDoWorkAccepted: %s", workManagers);

      if (own != null && transport != null && transport.isInitialized())
      {
         for (Address address : workManagers)
         {
            if (!own.equals(address))
               transport.deltaDoWorkAccepted(address);
         }
      }
   }
1295535
protected void transmitAtEndOfTask(final Task<Revision> task,
			final Task<Diff> result)
		throws TimeoutException
	{

		if (task.getTaskType() == TaskTypes.TASK_FULL
				|| task.getTaskType() == TaskTypes.TASK_PARTIAL_LAST) {

			if (this.partCounter > 1) {
				this.result.setTaskType(TaskTypes.TASK_PARTIAL_LAST);
				this.taskTransmitter.transmitPartialDiff(result);
			}
			else {
				this.result.setTaskType(TaskTypes.TASK_FULL);
				this.taskTransmitter.transmitDiff(result);
			}

			this.result = null;
		}
	}
1209424
@Override
  public void process(Node externs, Node root) {
    checkState(topLevelStatements.isEmpty(), "process() called more than once");
    NodeTraversal t = new NodeTraversal(compiler, this, scopeCreator);
    t.traverseRoots(externs, root);
  }
97364
protected Tree processGossipRequest(Tree data) throws Exception {

		// Debug
		String sender = data.get("sender", (String) null);
		if (debugHeartbeats) {
			logger.info("Gossip request received from \"" + sender + "\" node:\r\n" + data);
		}

		// Add "online" and "offline" response blocks
		LinkedList<NodeDescriptor> allNodes = new LinkedList<>(nodes.values());
		NodeDescriptor descriptor = getDescriptor();
		allNodes.add(descriptor);
		int size = allNodes.size() + 1;
		FastBuildTree onlineRsp = new FastBuildTree(size);
		FastBuildTree offlineRsp = new FastBuildTree(size);

		// Online / offline nodes in request
		Tree onlineReq = data.get("online");
		Tree offlineReq = data.get("offline");

		// Loop in nodes
		LinkedList<NodeDescriptor> disconnectedNodes = new LinkedList<>();
		for (NodeDescriptor node : allNodes) {
			node.writeLock.lock();
			try {

				Tree online = onlineReq == null ? null : onlineReq.get(node.nodeID);
				Tree offline = offlineReq == null ? null : offlineReq.get(node.nodeID);

				// Online or offline sequence number
				long seq = 0;

				// CPU data
				long cpuSeq = 0;
				int cpu = 0;

				if (offline != null) {
					if (!offline.isPrimitive()) {
						logger.warn("Invalid \"offline\" block: " + offline.toString(false));
						continue;
					}
					seq = offline.asLong();
				} else if (online != null) {
					if (!online.isEnumeration() || online.size() != 3) {
						logger.warn("Invalid \"online\" block: " + online.toString(false));
						continue;
					}
					seq = online.get(0).asLong();
					cpuSeq = online.get(1).asLong();
					cpu = online.get(2).asInteger();
				}

				if ((seq == 0 || seq < node.seq) && node.seq > 0) {

					// We have newer info or requester doesn't know it
					if (node.offlineSince == 0) {
						if (!node.info.isEmpty()) {
							if ((cpuSeq == 0 || cpuSeq < node.cpuSeq) && node.cpuSeq > 0) {
								ArrayList<Object> array = new ArrayList<>(3);
								array.add(node.info.asObject());
								array.add(node.cpuSeq);
								array.add(node.cpu);
								onlineRsp.putUnsafe(node.nodeID, array);
							} else {
								onlineRsp.putUnsafe(node.nodeID, Collections.singletonList(node.info.asObject()));
							}
						}
					} else {
						offlineRsp.putUnsafe(node.nodeID, node.seq);
					}
				}

				if (offline != null) {

					// Requester said it is OFFLINE
					if (node.offlineSince > 0) {

						// We also knew it as offline
						node.markAsOffline(seq);
						continue;
					}
					if (!node.local) {
						if (node.offlineSince == 0) {

							// We know it is online, so we change it to offline
							if (node.markAsOffline(seq)) {

								// Remove remote actions and listeners
								registry.removeActions(node.nodeID);
								eventbus.removeListeners(node.nodeID);
								writer.close(node.nodeID);
								disconnectedNodes.add(node);

							} else if (seq == node.seq) {

								// We send back that this node is online
								node.seq = seq + 1;
								node.info.put("seq", node.seq);
								if (cpuSeq < node.cpuSeq && node.cpuSeq > 0) {
									ArrayList<Object> array = new ArrayList<>(3);
									array.add(node.info.asObject());
									array.add(node.cpuSeq);
									array.add(node.cpu);
									onlineRsp.putUnsafe(node.nodeID, array);
								} else {
									onlineRsp.putUnsafe(node.nodeID, Collections.singletonList(node.info.asObject()));
								}
							}
						}
						continue;
					}
				} else if (online != null) {

					// Requester said it is ONLINE
					if (node.offlineSince == 0) {
						if (cpuSeq > node.cpuSeq) {

							// We update our CPU info
							node.updateCpu(cpuSeq, cpu);

						} else if (cpuSeq < node.cpuSeq && node.cpuSeq > 0) {

							// We have newer CPU value, send back
							ArrayList<Object> array = new ArrayList<>(2);
							array.add(node.cpuSeq);
							array.add(node.cpu);
							onlineRsp.putUnsafe(node.nodeID, array);
						}
					} else {

						// We knew it as offline. We do nothing, because we'll
						// request it and we'll receive its INFO
						continue;
					}
				}
			} finally {
				node.writeLock.unlock();
			}
		}

		// Create gossip response
		FastBuildTree root = new FastBuildTree(4);
		root.putUnsafe("ver", ServiceBroker.PROTOCOL_VERSION);
		root.putUnsafe("sender", nodeID);

		// Remove empty blocks
		boolean emptyOnlineBlock = onlineRsp.isEmpty();
		boolean emptyOfflineBlock = offlineRsp.isEmpty();
		if (emptyOnlineBlock && emptyOfflineBlock) {

			// Message is empty
			return root;
		}
		if (!emptyOnlineBlock) {
			root.putUnsafe("online", onlineRsp.asObject());
		}
		if (!emptyOfflineBlock) {
			root.putUnsafe("offline", offlineRsp.asObject());
		}

		// Debug
		if (debugHeartbeats) {
			logger.info("Gossip response submitting to \"" + sender + "\" node:\r\n" + root);
		}

		// Serialize response
		byte[] packet = serialize(PACKET_GOSSIP_RSP_ID, root);

		// Send response
		writer.send(sender, packet);

		// Notify listeners (unexpected disconnection)
		for (NodeDescriptor node : disconnectedNodes) {
			logger.info("Node \"" + node.nodeID + "\" disconnected.");
			broadcastNodeDisconnected(node.info, true);
		}

		// For unit testing
		return root;
	}
983571
@Override
   public void rebalance() {
       final int target = fairShare();
       final int mine = cluster.myWorkUnits.size();

       if (mine > target) {
           LOG.info("Simple Rebalance triggered. My Share: {}. Target: {}.", mine, target);
           super.drainToCount(target);
       }
   }
271801
@Override public void reduce(DeepWaterTask other){
    if (_localmodel != null && other._localmodel != null && other._localmodel.get_processed_local() > 0 //other DLTask was active (its model_info should be used for averaging)
        && other._localmodel != _localmodel) //other DLTask worked on a different model_info
    {
      // avoid adding remote model info to unprocessed local data, still random
      // (this can happen if we have no chunks on the master node)
      if (_localmodel.get_processed_local() == 0) {
        _localmodel = other._localmodel;
        _chunk_node_count = other._chunk_node_count;
      } else {
        _localmodel.add(other._localmodel);
        _chunk_node_count += other._chunk_node_count;
      }
    }
  }
269641
private synchronized boolean datanodeReportsReceived(boolean checkDatanodes) {
    try {
      boolean received = this.getDatanodeReportRatio() >=
        this.outStandingReportThreshold;
      if (!received && checkDatanodes) {
        checkDatanodes();
        return this.getDatanodeReportRatio() >= this.outStandingReportThreshold;
      }
      return received;
    } catch (Exception e) {
      LOG.warn("Failover - caught exception when checking reports", e);
      return false;
    }
  }
1267670
@Override
    public void onConnect(int rst, HuaweiApiClient client) {
        if (isCurFloatShow && client != null) {
            showFinal(true, null, client);
        }
    }
424169
public void setSubnetwork(int nodeId, int subnetwork) {
        if (subnetwork > 127)
            throw new IllegalArgumentException("Number of subnetworks is currently limited to 127 but requested " + subnetwork);

        byte[] bytes = new byte[1];
        bytes[0] = (byte) subnetwork;
        da.setBytes(nodeId, bytes, bytes.length);
    }
969378
public synchronized void onTransportTermination() {
    if (state != State.DISCONNECTED) {
      state = State.DISCONNECTED;
      if (shutdownFuture != null) {
        shutdownFuture.cancel(false);
      }
      if (pingFuture != null) {
        pingFuture.cancel(false);
        pingFuture = null;
      }
    }
  }
1071877
@Override
    public R visitSynchronized(SynchronizedTree node, P p) {
        R r = scan(node.getExpression(), p);
        r = scanAndReduce(node.getBlock(), p, r);
        return r;
    }
1251052
private void addReduceTask(TaskTracker taskTracker, List<Task> tasks,
                                  int maxReduceSlots, int currentReduceSlots) 
                    throws IOException {
    if (maxReduceSlots > currentReduceSlots) {
      reduceScheduler.updateCollectionOfQSIs();
      TaskLookupResult tlr = reduceScheduler.assignTasks(taskTracker);
      if (TaskLookupResult.LookUpStatus.TASK_FOUND == tlr.getLookUpStatus()) {
        tasks.add(tlr.getTask());
      }
    }
  }