@Override
		public void run() {
			while (!stopRequested) {
				checkJobStatus();

				for (JobCache job : targetJobs) {
//					final LocalScheduler scheduler = job.scheduler;
//					final VertexStat[] stats = new VertexStat[job.vertices.length];
//					for (int i = 0; i < job.vertices.length; i++) {
//						final LocalPhysicalVertex lv = job.vertices[i];
//						final ThreadContextBase context = lv.getContext();
//						final long cpuCycles = context.getCpuCycles();
//						final long inputCount = context.getInputCount();
//						final long outputCount = context.getOutputCount();
//						context.resetCpuCyelces();
//						context.resetInputCount();
//						context.resetOutputCount();
//
//						stats[i] = new VertexStat(cpuCycles, inputCount, outputCount);
//						scheduler.setOperatorStat(lv, stats[i]);
//					}
//
//					LocalPhysicalGraph newGraph = scheduler.schedule();
//					if (job.graph != newGraph) {
//						deploymentor.changeDeploy(job.graph, newGraph);
//						buildJobCache(job.jobDesc);
//					}

					final long currentNano = System.nanoTime();
					if (currentNano - job.lastScheduledTime >= GLOBAL_TICK) {
//						sendStats(job, stats);
						worker.sendStat(job.jobDesc);
						job.lastScheduledTime = currentNano;
					}
				}

				try {
					Thread.sleep(LOCAL_TICK);
				} catch (InterruptedException ignore) {	}
			}
		}