Run the scheduler.
It is a blocking call, and it will return in 2 cases:
1. The topology is requested to kill
2. Unexpected exceptions happen

@return true if scheduled successfully
963510
public boolean runScheduler() {
    IScheduler scheduler = null;

    String statemgrClass = Context.stateManagerClass(config);
    IStateManager statemgr;

    try {
      // create an instance of state manager
      statemgr = ReflectionUtils.newInstance(statemgrClass);
    } catch (IllegalAccessException | InstantiationException | ClassNotFoundException e) {
      LOG.log(Level.SEVERE, "Failed to instantiate instances using config: " + config, e);
      return false;
    }

    SchedulerServer server = null;
    boolean isSuccessful = false;

    // Put it in a try block so that we can always clean resources
    try {
      // initialize the state manager
      statemgr.initialize(config);

      // TODO(mfu): timeout should read from config
      SchedulerStateManagerAdaptor adaptor = new SchedulerStateManagerAdaptor(statemgr, 5000);

      // get a packed plan and schedule it
      PackingPlans.PackingPlan serializedPackingPlan = adaptor.getPackingPlan(topology.getName());
      if (serializedPackingPlan == null) {
        LOG.log(Level.SEVERE, "Failed to fetch PackingPlan for topology:{0} from the state manager",
            topology.getName());
        return false;
      }
      LOG.log(Level.INFO, "Packing plan fetched from state: {0}", serializedPackingPlan);
      PackingPlan packedPlan = new PackingPlanProtoDeserializer().fromProto(serializedPackingPlan);

      // build the runtime config
      LauncherUtils launcherUtils = LauncherUtils.getInstance();
      Config runtime = Config.newBuilder()
          .putAll(launcherUtils.createPrimaryRuntime(topology))
          .putAll(launcherUtils.createAdaptorRuntime(adaptor))
          .put(Key.SCHEDULER_SHUTDOWN, getShutdown())
          .put(Key.SCHEDULER_PROPERTIES, properties)
          .build();

      Config ytruntime = launcherUtils.createConfigWithPackingDetails(runtime, packedPlan);

      // invoke scheduler
      scheduler = launcherUtils.getSchedulerInstance(config, ytruntime);
      if (scheduler == null) {
        return false;
      }

      isSuccessful = scheduler.onSchedule(packedPlan);
      if (!isSuccessful) {
        LOG.severe("Failed to schedule topology");
        return false;
      }

      // Failures in server initialization throw exceptions
      // get the scheduler server endpoint for receiving requests
      server = getServer(ytruntime, scheduler, schedulerServerPort);
      // start the server to manage runtime requests
      server.start();

      // write the scheduler location to state manager
      // Make sure it happens after IScheduler.onScheduler
      isSuccessful = SchedulerUtils.setSchedulerLocation(
          runtime, String.format("%s:%d", server.getHost(), server.getPort()), scheduler);

      if (isSuccessful) {
        // wait until kill request or some interrupt occurs if the scheduler starts successfully
        LOG.info("Waiting for termination... ");
        Runtime.schedulerShutdown(ytruntime).await();
      }
    } catch (IOException e) {
      LOG.log(Level.SEVERE, "Failed to start server", e);
      return false;
    } finally {
      // Clean the resources
      if (server != null) {
        server.stop();
      }

      // 4. Close the resources
      SysUtils.closeIgnoringExceptions(scheduler);
      SysUtils.closeIgnoringExceptions(statemgr);
    }

    return isSuccessful;
  }
1189762
public synchronized void initTaskSchedulerIfNot() {

        if (scheduler == null) {
            scheduler = Executors
                    .newSingleThreadScheduledExecutor(DaemonThreadFactory
                            .getInstance());
            CapacityAwareTaskScheduler runner = new CapacityAwareTaskScheduler();
            scheduler.scheduleAtFixedRate(runner,
                    ParallecGlobalConfig.schedulerInitDelay,
                    ParallecGlobalConfig.schedulerCheckInterval,
                    TimeUnit.MILLISECONDS);
            logger.info("initialized daemon task scheduler to evaluate waitQ tasks.");
            
        }
    }
1312990
@Override
    public synchronized void run() {
        try {
            notifier.beforePollingCycle(new BeforePollingCycleEvent(dp));
            if (!executor.isShutdown()) {
                executor.invokeAll(pollers);
            }
            notifier.afterPollingCycle(new AfterPollingCycleEvent(dp));
        } catch (InterruptedException e) {
            // allow thread to exit gracefully
        } catch (Throwable t) {
            LOG.error("Unexpected error!", t);
        }
    }
1126678
@Override
	public void run() {
		Stopwatch stopwatch = SimonManager.getStopwatch(NAME);
		try (Split ignored = stopwatch.start();) {
			sleep(SLEEP);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		// signal to latch that the thread is finished
		latch.countDown();
	}
756537
@Override
  public void execute(Runnable r) {
    Preconditions.checkNotNull(r, "'r' must not be null.");
    boolean scheduleTaskRunner = false;
    synchronized (internalLock) {
      waitQueue.add(r);

      if (!isThreadScheduled) {
        isThreadScheduled = true;
        scheduleTaskRunner = true;
      }
    }
    if (scheduleTaskRunner) {
      boolean threw = true;
      try {
        executor.execute(taskRunner);
        threw = false;
      } finally {
        if (threw) {
          synchronized (internalLock) {
            // It is possible that at this point that there are still tasks in
            // the queue, it would be nice to keep trying but the error may not
            // be recoverable.  So we update our state and propogate so that if
            // our caller deems it recoverable we won't be stuck.
            isThreadScheduled = false;
          }
        }
      }
    }
  }
815666
public void run() {
        System.out.print(HORIZONTAL_RULE);
        System.out.println(" Starting Processing");
        System.out.println(HORIZONTAL_RULE);

        // Print periodic stats/analysis to the console
        scheduler.scheduleWithFixedDelay(reporter,
                config.displayinterval,
                config.displayinterval,
                TimeUnit.SECONDS);

        // Update the partition key set, row counts and redundancy level 4 times per second
        scheduler.scheduleWithFixedDelay(partitionTracker,
                0,
                250,
                TimeUnit.MILLISECONDS);

        // Start tracking changes to the maximum value after a 5 second delay to
        // let things settle a bit. Then check up to 100 times per second.
        scheduler.scheduleWithFixedDelay(maxTracker,
                5000,
                10,
                TimeUnit.MILLISECONDS);

        // Run the benchmark loop for the requested duration
        // The throughput may be throttled depending on client configuration
        inserter.run();

        System.out.print(HORIZONTAL_RULE);
        System.out.println(" Processing Complete");
        System.out.println(HORIZONTAL_RULE);

        shutdown();
    }
996384
public boolean pauseQuartzScheduler() {
        try {
            this.scheduler.standby();
            return true;
        } catch (SchedulerException e) {
            logger.error("error pausing quartz scheduler", e);
        }
        return false;
    }
1167937
public void run() {
        int emptyCount = 0;
        long scheduleTickTime = System.nanoTime();
        boolean isShutDown = false;
        activeDispatchers.incrementAndGet();
        try {
            while( ! isShutDown ) {
                try {
                    if ( pollQs() ) {
                        emptyCount = 0;
                        if ( System.nanoTime() - scheduleTickTime > SCHEDULE_TICK_NANOS) {
                            if ( emptySinceLastCheck == 0 ) // no idle during last interval
                            {
                                checkForSplit();
                            }
                            emptySinceLastCheck = 0;
                            scheduleTickTime = System.nanoTime();
                            schedulePendingAdds();
                        }
                    }
                    else {
                        emptyCount++;
                        emptySinceLastCheck++;
                        scheduler.pollDelay(emptyCount);
                        if (shutDown) // access volatile only when idle
                            isShutDown = true;
                        if ( scheduler.getBackoffStrategy().isSleeping(emptyCount) ) {
                            scheduleTickTime = 0;
                            schedulePendingAdds();
                            if ( System.currentTimeMillis()-created > 5000 ) {
                                if ( autoShutDown && actors.length == 0 && toAdd.peek() == null) {
                                    shutDown();
                                } else {
                                    scheduler.tryStopThread(this);
                                }
                            }
                        }
                    }
                } catch (Throwable th) {
                    Log.Warn(this, th, "from main poll loop");
                }
            }
            scheduler.threadStopped(this);
            LockSupport.parkNanos(1000*1000*1000);
            if ( actors.length > 0 || toAdd.peek() != null ) {
                if ( SimpleScheduler.DEBUG_SCHEDULING)
                    Log.Lg.warn(this, "zombie dispatcher thread detected. This can be a debugging artifact.");
                scheduler.tryStopThread(this);
            }
            if ( SimpleScheduler.DEBUG_SCHEDULING)
                Log.Debug(this,"dispatcher thread terminated "+getName());
        } finally {
            activeDispatchers.decrementAndGet();
        }
    }
1173230
@Override
  public void run() {
    synchronized (lock) {
      if (!callables.hasNext()) {
        checkEnd();
        return;
      }

      for (int i = 0; i < parallelism && callables.hasNext(); i++) {
        setupNext(callables.next());
      }
    }

    future.whenCancelled(() -> {
      cancel = true;
      checkNext();
    });
  }
706801
public boolean start() {
        if (!this.isStarted.compareAndSet(false, true)) {
            return false;
        }
        final long delay = this.delayBetweenSweeps;
        this.timer.scheduleWithFixedDelay(this, delay, delay, TimeUnit.MILLISECONDS);
        LogWatchStorageSweeper.LOGGER.info(
                "Scheduled automated unreachable message sweep in {} to run every {} millisecond(s).",
                this.messaging.getLogWatch(), delay);
        return true;
    }
617371
def start(self):
        
        with self._lock:
            if self._running:
                return

            self._stopped = False
            self._running = True
            self._start_idle_timer()
            _logger.debug(u'starting thread of type %s to run the scheduler',
                          _THREAD_CLASS)
            self._thread = create_thread(target=self._schedule_flushes)
            try:
                self._thread.start()
            except Exception:  # pylint: disable=broad-except
                _logger.warn(
                    u'no scheduler thread, scheduler.run() will be invoked by report(...)',
                    exc_info=True)
                self._thread = None
                self._initialize_flushing()
501271
public Boolean call() throws Exception {
    Boolean 	result;

    try {
      result = doRun();
    }
    catch (Exception e) {
      m_Exception = e;
      throw e;
    }
    finally {
      cleanUp();
    }
    return result;
  }
1024868
private void executeAsyncCallRunnable(Runnable runnable, String callName, boolean blocking) {
		// make sure the executor is initialized. lock against concurrent calls to this function
		synchronized (this) {
			if (executionState != ExecutionState.RUNNING) {
				return;
			}

			// get ourselves a reference on the stack that cannot be concurrently modified
			BlockingCallMonitoringThreadPool executor = this.asyncCallDispatcher;
			if (executor == null) {
				// first time use, initialize
				checkState(userCodeClassLoader != null, "userCodeClassLoader must not be null");

				// Under normal execution, we expect that one thread will suffice, this is why we
				// keep the core threads to 1. In the case of a synchronous savepoint, we will block
				// the checkpointing thread, so we need an additional thread to execute the
				// notifyCheckpointComplete() callback. Finally, we aggressively purge (potentially)
				// idle thread so that we do not risk to have many idle thread on machines with multiple
				// tasks on them. Either way, only one of them can execute at a time due to the
				// checkpoint lock.

				executor = new BlockingCallMonitoringThreadPool(
						new DispatcherThreadFactory(
							TASK_THREADS_GROUP,
							"Async calls on " + taskNameWithSubtask,
							userCodeClassLoader));
				this.asyncCallDispatcher = executor;

				// double-check for execution state, and make sure we clean up after ourselves
				// if we created the dispatcher while the task was concurrently canceled
				if (executionState != ExecutionState.RUNNING) {
					executor.shutdown();
					asyncCallDispatcher = null;
					return;
				}
			}

			LOG.debug("Invoking async call {} on task {}", callName, taskNameWithSubtask);

			try {
				executor.submit(runnable, blocking);
			}
			catch (RejectedExecutionException e) {
				// may be that we are concurrently finished or canceled.
				// if not, report that something is fishy
				if (executionState == ExecutionState.RUNNING) {
					throw new RuntimeException("Async call with a " + (blocking ? "" : "non-") + "blocking call was rejected, even though the task is running.", e);
				}
			}
		}
	}
88451
public void run() {
        pollThread = Thread.currentThread();
        if ( underway )
            return;
        underway = true;
        try {
            int count = 1;
            while( count > 0 ) { // as long there are messages, keep sending them
                count = onePoll();
                if ( sendJobs.size() > 0 ) {
                    if ( count > 0 ) {
                        int debug =1;
                    }
                    else {
                        if ( remoteRefCounter == 0 ) // no remote actors registered
                        {
                            Actor.current().delayed(100, this); // backoff massively
                        } else {
                            Actor.current().delayed(1, this); // backoff a bit (remoteactors present, no messages)
                        }
                    }
                } else {
                    // no schedule entries (== no clients)
                    Actor.current().delayed(100, this );
                }
            }
        } finally {
            underway = false;
        }
    }
358319
@Override
    public void run() {
        final ScheduledFuture<?> future = this.executor.scheduleWithFixedDelay(this.scheduled, 0, this.delayMillis,
                TimeUnit.MILLISECONDS);
        try {
            this.runTrigger.await();
        } catch (final InterruptedException e) {
            this.listener.handle(e);
        } finally {
            future.cancel(true); // stop the periodic reading
            this.scheduled.cleanup();
            this.executor.shutdownNow();
            this.listener.destroy();
        }
    }
825045
@Override
  public void run() {

    boolean lastAcquireFailed = false;

    while (!halted.get()) {
      try {
        // check if we're supposed to pause...
        synchronized (sigLock) {
          while (paused && !halted.get()) {
            try {
              // wait until togglePause(false) is called...
              sigLock.wait(1000L);
            } catch (InterruptedException ignore) {
            }
          }

          if (halted.get()) {
            break;
          }
        }

        int availThreadCount = quartzSchedulerResources.getThreadPool().blockForAvailableThreads();
        if (availThreadCount
            > 0) { // will always be true, due to semantics of blockForAvailableThreads...

          List<OperableTrigger> triggers = null;

          long now = System.currentTimeMillis();

          clearSignaledSchedulingChange();
          try {
            triggers =
                quartzSchedulerResources
                    .getJobStore()
                    .acquireNextTriggers(
                        now + idleWaitTime,
                        Math.min(availThreadCount, quartzSchedulerResources.getMaxBatchSize()),
                        quartzSchedulerResources.getBatchTimeWindow());
            lastAcquireFailed = false;
            logger.debug(
                "batch acquisition of " + (triggers == null ? 0 : triggers.size()) + " triggers");
          } catch (JobPersistenceException jpe) {

            lastAcquireFailed = true;
          } catch (RuntimeException e) {
            if (!lastAcquireFailed) {
              logger.error("quartzSchedulerThreadLoop: RuntimeException " + e.getMessage(), e);
            }
            lastAcquireFailed = true;
          }

          if (triggers != null && !triggers.isEmpty()) {

            now = System.currentTimeMillis();
            long triggerTime = triggers.get(0).getNextFireTime().getTime();
            long timeUntilTrigger = triggerTime - now;
            while (timeUntilTrigger > 2) {
              synchronized (sigLock) {
                if (halted.get()) {
                  break;
                }
                if (!isCandidateNewTimeEarlierWithinReason(triggerTime, false)) {
                  try {
                    // we could have blocked a long while
                    // on 'synchronize', so we must recompute
                    now = System.currentTimeMillis();
                    timeUntilTrigger = triggerTime - now;
                    if (timeUntilTrigger >= 1) {
                      sigLock.wait(timeUntilTrigger);
                    }
                  } catch (InterruptedException ignore) {
                  }
                }
              }
              if (releaseIfScheduleChangedSignificantly(triggers, triggerTime)) {
                break;
              }
              now = System.currentTimeMillis();
              timeUntilTrigger = triggerTime - now;
            }

            // this happens if releaseIfScheduleChangedSignificantly decided to release triggers
            if (triggers.isEmpty()) {
              continue;
            }

            // set triggers to 'executing'
            List<TriggerFiredResult> bndles = new ArrayList<TriggerFiredResult>();

            boolean goAhead = true;
            synchronized (sigLock) {
              goAhead = !halted.get();
            }
            if (goAhead) {
              try {
                List<TriggerFiredResult> res =
                    quartzSchedulerResources.getJobStore().triggersFired(triggers);
                if (res != null) {
                  bndles = res;
                }
              } catch (SchedulerException se) {
                quartzScheduler.notifySchedulerListenersError(
                    "An error occurred while firing triggers '" + triggers + "'", se);
              }
            }

            for (int i = 0; i < bndles.size(); i++) {

              TriggerFiredResult result = bndles.get(i);
              TriggerFiredBundle bndle = result.getTriggerFiredBundle();
              Exception exception = result.getException();

              if (exception instanceof RuntimeException) {
                logger.error("RuntimeException while firing trigger " + triggers.get(i), exception);

                continue;
              }

              // it's possible to get 'null' if the triggers was paused,
              // blocked, or other similar occurrences that prevent it being
              // fired at this time... or if the scheduler was shutdown (halted)
              if (bndle == null) {
                try {
                  quartzSchedulerResources.getJobStore().releaseAcquiredTrigger(triggers.get(i));
                } catch (SchedulerException se) {
                  quartzScheduler.notifySchedulerListenersError(
                      "An error occurred while releasing triggers '"
                          + triggers.get(i).getName()
                          + "'",
                      se);
                }
                continue;
              }

              // TODO: improvements:
              //
              // 2- make sure we can get a job runshell before firing triggers, or
              // don't let that throw an exception (right now it never does,
              // but the signature says it can).
              // 3- acquire more triggers at a time (based on num threads available?)

              JobRunShell shell = null;
              try {
                shell = quartzSchedulerResources.getJobRunShellFactory().createJobRunShell(bndle);
                shell.initialize(quartzScheduler);
              } catch (SchedulerException se) {
                try {
                  quartzSchedulerResources
                      .getJobStore()
                      .triggeredJobComplete(
                          triggers.get(i),
                          bndle.getJobDetail(),
                          CompletedExecutionInstruction.SET_ALL_JOB_TRIGGERS_ERROR);
                } catch (SchedulerException se2) {
                  quartzScheduler.notifySchedulerListenersError(
                      "An error occurred while placing job's triggers in error state '"
                          + triggers.get(i).getName()
                          + "'",
                      se2);
                }
                continue;
              }

              if (quartzSchedulerResources.getThreadPool().runInThread(shell) == false) {
                try {
                  // this case should never happen, as it is indicative of the
                  // scheduler being shutdown or a bug in the thread pool or
                  // a thread pool being used concurrently - which the docs
                  // say not to do...
                  logger.error("ThreadPool.runInThread() return false!");
                  quartzSchedulerResources
                      .getJobStore()
                      .triggeredJobComplete(
                          triggers.get(i),
                          bndle.getJobDetail(),
                          CompletedExecutionInstruction.SET_ALL_JOB_TRIGGERS_ERROR);
                } catch (SchedulerException se2) {
                  quartzScheduler.notifySchedulerListenersError(
                      "An error occurred while placing job's triggers in error state '"
                          + triggers.get(i).getName()
                          + "'",
                      se2);
                }
              }
            }

            continue; // while (!halted)
          }
        } else { // if(availThreadCount > 0)
          // should never happen, if threadPool.blockForAvailableThreads() follows contract
          continue; // while (!halted)
        }

        long now = System.currentTimeMillis();
        long waitTime = now + getRandomizedIdleWaitTime();
        long timeUntilContinue = waitTime - now;
        synchronized (sigLock) {
          try {
            sigLock.wait(timeUntilContinue);
          } catch (InterruptedException ignore) {
          }
        }

      } catch (RuntimeException re) {
        logger.error("Runtime error occurred in main trigger firing loop.", re);
      }
    } // while (!halted)

    // drop references to scheduler stuff to aid garbage collection...
    quartzScheduler = null;
    quartzSchedulerResources = null;
  }
1131847
public void run() {
		
		task.setSetTimerTransactionalAction(null);
		
		if (!canceled) {
			
			TimerTask previousTask = scheduler.getLocalRunningTasksMap().putIfAbsent(task.getData().getTaskID(), task);
			if(previousTask != null) {
				if (logger.isDebugEnabled()) {
					logger.debug("A task with id " + task.getData().getTaskID() + " has already been added to the local tasks, not rescheduling");
				}
			}
			
			final TimerTaskData taskData = task.getData();
			// calculate delay
			long delay = taskData.getStartTime() - System.currentTimeMillis();
			if (delay < 0) {
				delay = 0;
			}
			
			try {
				// schedule runnable
				if (taskData.getPeriod() < 0) {
					if (logger.isDebugEnabled()) {
						logger.debug("Scheduling one-shot timer with id "
								+ task.getData().getTaskID() + ", delay " + delay);
					}
					task.setScheduledFuture(scheduler.getExecutor().schedule(task, delay, TimeUnit.MILLISECONDS));
				} else {
					if (logger.isDebugEnabled()) {
						logger.debug("Scheduling periodic timer with id "
								+ task.getData().getTaskID() + ", scheduling strategy " + taskData.getPeriodicScheduleStrategy() + ", delay " + delay + ", period " + taskData.getPeriod());
					}
					if (taskData.getPeriodicScheduleStrategy() == PeriodicScheduleStrategy.withFixedDelay) {
						task.setScheduledFuture(scheduler.getExecutor().scheduleWithFixedDelay(task, delay, taskData.getPeriod(),TimeUnit.MILLISECONDS));
					}
					else {
						// default
						task.setScheduledFuture(scheduler.getExecutor().scheduleAtFixedRate(task, delay, taskData.getPeriod(),TimeUnit.MILLISECONDS));
					}					
				}		
			} catch (Throwable e) {
				logger.error(e.getMessage(), e);
				scheduler.remove(taskData.getTaskID(),true);
			}
		} else {
			if (logger.isDebugEnabled()) {
				logger.debug("Canceled scheduling periodic timer with id "
						+ task.getData().getTaskID());
			}
		}
	}