@Override
    public void updateStats(final PhysicalGraph graph) throws ClientException {
        List<WorkerSignature> workers = getAllWorkers();
        HashMap<Integer, LocalJobStatistics> statsMap = new HashMap<>();
        for (WorkerSignature worker : workers) {
            final LocalJobStatistics localStat = workerCore.getJobStatistics(masterCore.findJobDescripter(graph
                    .getJobId()));
            if (localStat == null) {
                return;
            }
            statsMap.put(worker.getServerId(), localStat);
        }

        Iterator<PhysicalVertex> iter = graph.iteratorOfVertex();
        while (iter.hasNext()) {
            final PhysicalVertex pv = iter.next();
            final LocalJobStatistics stats = statsMap.get(pv.getWorkerSignature().getServerId());
            final List<LocalVertexStat> list = stats.getLocalVertexStat(pv.getPhysicalId());
            long inputCount = pv.getInputCountStat(), outputCount = pv.getOutputCountStat();
            for (LocalVertexStat stat : list) {
                inputCount += stat.getInputCount();
                outputCount += stat.getOutputCount();
            }
            pv.setInputCountStat(inputCount);
            pv.setOutputCountStat(outputCount);
        }
    }