MasterService extends AbstractService implements MasterProtocol { public MasterService(AMContext context) { super(MasterService.class.getName()); this.context = context; this.stopped = new AtomicBoolean(false); matrics = new ArrayList<>(); Configuration conf = context.getConf(); yarnNMWebPort = getYarnNMWebPort(conf); } MasterService(AMContext context); @Override long getProtocolVersion(String protocol, long clientVersion); @SuppressWarnings("unchecked") @Override PSReportResponse psReport(
      RpcController controller, PSReportRequest request); @Override PSAgentMasterServiceProtos.FetchMinClockResponse fetchMinClock(RpcController controller,
      PSAgentMasterServiceProtos.FetchMinClockRequest request); @SuppressWarnings("unchecked") @Override PSRegisterResponse psRegister(
      RpcController controller, PSRegisterRequest request); RpcServer getRpcServer(); @Override GetJobReportResponse getJobReport(RpcController controller,
      GetJobReportRequest request); @Override GetWorkerLogDirResponse getWorkerLogDir(RpcController controller,
      GetWorkerLogDirRequest request); @Override CreateMatricesResponse createMatrices(RpcController controller,
      CreateMatricesRequest request); @Override GetMatricesResponse getMatrices(RpcController controller,
      GetMatricesRequest request); @Override ReleaseMatricesResponse releaseMatrices(RpcController controller,
      ReleaseMatricesRequest request); InetSocketAddress getRPCListenAddr(); @SuppressWarnings("unchecked") @Override PSDoneResponse psDone(RpcController controller,
      PSDoneRequest request); @SuppressWarnings("unchecked") @Override PSErrorResponse psError(RpcController controller,
      PSErrorRequest request); @Override GetAllMatrixMetaResponse getAllMatrixMeta(RpcController controller,
      GetAllMatrixMetaRequest request); @Override GetAllPSLocationResponse getAllPSLocation(RpcController controller,
      GetAllPSLocationRequest request); @Override GetPSLocationReponse getPSLocation(RpcController controller,
      GetPSLocationRequest request); @Override GetPartLocationResponse getPartLocation(RpcController controller,
      GetPartLocationRequest request); @Override GetIterationResponse getIteration(RpcController controller,
      GetIterationRequest request); @Override GetPSMatricesResponse getPSMatricesMeta(RpcController controller,
      GetPSMatricesMetaRequest request); @Override SaveFinishResponse saveFinish(RpcController controller,
      SaveFinishRequest request); @Override LoadFinishResponse loadFinish(RpcController controller,
      LoadFinishRequest request); @Override SaveStartResponse saveStart(RpcController controller, SaveStartRequest request); @Override LoadStartResponse loadStart(RpcController controller, LoadStartRequest request); @Override GetSaveResultsResponse getSaveResults(RpcController controller,
      GetSaveResultsRequest request); @Override GetCheckpointsResponse getCheckpointResults(RpcController controller,
      GetCheckpointsRequest request); @Override GetStoredPssResponse getStoredPss(RpcController controller,
      GetStoredPssRequest request); @Override GetPSAgentIdResponse getPSAgentId(RpcController controller,
      GetPSAgentIdRequest request); @Override CheckPSExitResponse checkPSExited(RpcController controller,
      CheckPSExitRequest request); @SuppressWarnings("unchecked") @Override PSAgentReportResponse psAgentReport(
      RpcController controller, PSAgentReportRequest request); @SuppressWarnings("unchecked") @Override PSAgentRegisterResponse psAgentRegister(
      RpcController controller, PSAgentRegisterRequest request); @SuppressWarnings("unchecked") @Override PSAgentDoneResponse psAgentDone(
      RpcController controller, PSAgentDoneRequest request); @SuppressWarnings("unchecked") @Override PSAgentErrorResponse psAgentError(
      RpcController controller, PSAgentErrorRequest request); @Override GetExecuteUnitDescResponse getExecuteUnitDesc(RpcController controller,
      GetExecuteUnitDescRequest request); @SuppressWarnings("unchecked") @Override WorkerReportResponse workerReport(
      RpcController controller, WorkerReportRequest request); @SuppressWarnings("unchecked") @Override WorkerRegisterResponse workerRegister(
      RpcController controller, WorkerRegisterRequest request); @Override GetWorkerGroupMetaInfoResponse getWorkerGroupMetaInfo(RpcController controller,
      GetWorkerGroupMetaInfoRequest request); @SuppressWarnings("unchecked") @Override WorkerDoneResponse workerDone(
      RpcController controller, WorkerDoneRequest request); @SuppressWarnings("unchecked") @Override WorkerErrorResponse workerError(
      RpcController controller, WorkerErrorRequest request); @Override GetWorkerGroupSuccessNumResponse getWorkerGroupSuccessNum(
      RpcController controller, GetWorkerGroupSuccessNumRequest request); List<MatrixMeta> getMatrics(); @Override TaskClockResponse taskClock(RpcController controller, TaskClockRequest request); @Override TaskIterationResponse taskIteration(RpcController controller,
      TaskIterationRequest request); @Override PSAgentMasterServiceProtos.TaskCountersUpdateResponse taskCountersUpdate(
      RpcController controller, PSAgentMasterServiceProtos.TaskCounterUpdateRequest request); @Override PSAgentMasterServiceProtos.SetAlgoMetricsResponse setAlgoMetrics(
      RpcController controller, PSAgentMasterServiceProtos.SetAlgoMetricsRequest request); @Override PSFailedReportResponse psFailedReport(RpcController controller,
      PSFailedReportRequest request); @Override GetTaskMatrixClockResponse getTaskMatrixClocks(RpcController controller,
      GetTaskMatrixClockRequest request); @Override PingResponse ping(RpcController controller, PingRequest request); Location getLocation(); @Override StartResponse start(RpcController controller, StartRequest request); @SuppressWarnings("unchecked") @Override SaveResponse save(RpcController controller,
      SaveRequest request); @Override LoadResponse load(RpcController controller, LoadRequest request); @Override ClientMasterServiceProtos.StopResponse stop(RpcController controller,
      ClientMasterServiceProtos.StopRequest request); void stop(int exitStatus); @Override CheckMatricesCreatedResponse checkMatricesCreated(RpcController controller,
      CheckMatricesCreatedRequest request); @Override CheckModelSavedResponse checkModelSaved(RpcController controller,
      CheckModelSavedRequest request); @Override CheckModelLoadedResponse checkModelLoaded(RpcController controller,
      CheckModelLoadedRequest request); @Override SetParamsResponse setParams(RpcController controller, SetParamsRequest request); @Override GetClientIdResponse getClientId(RpcController controller, GetClientIdRequest request); @Override KeepAliveResponse keepAlive(RpcController controller, KeepAliveRequest request); @Override ClientRegisterResponse clientRegister(RpcController controller,
      ClientRegisterRequest request); @Override KillPSResponse killPS(RpcController controller, KillPSRequest request);  }
@Test public void testMasterService() throws Exception { try { LOG.info("===========================testMasterService==============================="); Worker worker = LocalClusterContext.get().getWorker(worker0Attempt0Id).getWorker(); Location masterLoc = LocalClusterContext.get().getMaster().getAppMaster().getAppContext().getMasterService() .getLocation(); TConnection connection = TConnectionManager.getConnection(worker.getConf()); MasterProtocol master = connection.getMasterService(masterLoc.getIp(), masterLoc.getPort()); int psAgentId = master .getPSAgentId(null, PSAgentMasterServiceProtos.GetPSAgentIdRequest.getDefaultInstance()) .getPsAgentId(); WorkerAttemptId worker1Attempt0Id = new WorkerAttemptId(new WorkerId(new WorkerGroupId(1), 0), 0); WorkerRegisterRequest registeRequest = WorkerRegisterRequest.newBuilder().setPsAgentId(psAgentId) .setWorkerAttemptId(ProtobufUtil.convertToIdProto(worker1Attempt0Id)) .setLocation(LocationProto.newBuilder().setIp("10.10.10.10").setPort(10000).build()) .build(); WorkerRegisterResponse registerResponse = master.workerRegister(null, registeRequest); assertTrue(registerResponse.getCommand() == WorkerCommandProto.W_SHUTDOWN); WorkerReportRequest.Builder reportBuilder = WorkerReportRequest.newBuilder(); Pair.Builder kvBuilder = Pair.newBuilder(); TaskStateProto.Builder taskBuilder = TaskStateProto.newBuilder(); reportBuilder.setWorkerAttemptId(ProtobufUtil.convertToIdProto(worker0Attempt0Id)); taskBuilder.setProgress(0.20f); taskBuilder.setState("RUNNING"); taskBuilder.setTaskId(ProtobufUtil.convertToIdProto(task0Id)); kvBuilder.setKey("task_key1"); kvBuilder.setValue("100"); taskBuilder.addCounters(kvBuilder.build()); kvBuilder.setKey("task_key2"); kvBuilder.setValue("200"); taskBuilder.addCounters(kvBuilder.build()); reportBuilder.addTaskReports(taskBuilder.build()); taskBuilder.setProgress(0.30f); taskBuilder.setState("RUNNING"); taskBuilder.setTaskId(ProtobufUtil.convertToIdProto(task1Id)); kvBuilder.setKey("task_key1"); kvBuilder.setValue("1000"); taskBuilder.addCounters(kvBuilder.build()); kvBuilder.setKey("task_key2"); kvBuilder.setValue("2000"); taskBuilder.addCounters(kvBuilder.build()); reportBuilder.addTaskReports(taskBuilder.build()); kvBuilder.setKey("worker_key1"); kvBuilder.setValue("100"); reportBuilder.addPairs(kvBuilder.build()); kvBuilder.setKey("worker_key2"); kvBuilder.setValue("200"); reportBuilder.addPairs(kvBuilder.build()); WorkerReportResponse reportResponse = master.workerReport(null, reportBuilder.build()); assertTrue(reportResponse.getCommand() == WorkerCommandProto.W_SUCCESS); assertEquals(reportResponse.getActiveTaskNum(), 2); AngelApplicationMaster angelAppMaster = LocalClusterContext.get().getMaster().getAppMaster(); WorkerAttempt worker0Attempt = angelAppMaster.getAppContext().getWorkerManager().getWorker(worker0Attempt0Id.getWorkerId()) .getWorkerAttempt(worker0Attempt0Id); assertTrue(worker0Attempt != null); Map<String, String> workerMetrics = worker0Attempt.getMetrics(); String valueForWorkerKey1 = workerMetrics.get("worker_key1"); String valueForWorkerKey2 = workerMetrics.get("worker_key2"); assertNotNull(valueForWorkerKey1); assertNotNull(valueForWorkerKey2); assertEquals(valueForWorkerKey1, "100"); assertEquals(valueForWorkerKey2, "200"); AMTaskManager amTaskManager = angelAppMaster.getAppContext().getTaskManager(); AMTask task0 = amTaskManager.getTask(task0Id); AMTask task1 = amTaskManager.getTask(task1Id); assertTrue(task0 != null); assertTrue(task1 != null); Map<String, String> task0Metrics = task0.getMetrics(); Map<String, String> task1Metrics = task1.getMetrics(); String valueForTask0Key1 = task0Metrics.get("task_key1"); String valueForTask0Key2 = task0Metrics.get("task_key2"); String valueForTask1Key1 = task1Metrics.get("task_key1"); String valueForTask1Key2 = task1Metrics.get("task_key2"); assertTrue(valueForTask0Key1 != null); assertTrue(valueForTask0Key2 != null); assertTrue(valueForTask1Key1 != null); assertTrue(valueForTask1Key2 != null); assertEquals(valueForTask0Key1, "100"); assertEquals(valueForTask0Key2, "200"); assertEquals(valueForTask1Key1, "1000"); assertEquals(valueForTask1Key2, "2000"); assertEquals(task0.getProgress(), 0.20f, 0.000001); assertEquals(task1.getProgress(), 0.30f, 0.000001); } catch (Exception x) { LOG.error("run testMasterService failed ", x); throw x; } }