code: void launchTasklet(final Tasklet tasklet) {
      lock.lock();
      try {
        if (!terminated) {
          Optional<String> workerId;
          while(true) {
            workerId = schedulingPolicy.trySchedule(tasklet);
            if (!workerId.isPresent()) {
              try {
                noWorkerOrResource.await();
              } catch (InterruptedException e) {
                throw new RuntimeException(e);
              }
            } else {
              break;
            }
          }
  
          // TODO[JIRA REEF-500]: Will need to support duplicate tasklets.
          if (taskletsToCancel.contains(tasklet.getId())) {
            tasklet.cancelled();
            taskletsToCancel.remove(tasklet.getId());
            LOG.log(Level.FINE, "Cancelled tasklet {0}.", tasklet.getId());
            return;
          }
  
          final Optional<Integer> taskletAggFunctionId =  tasklet.getAggregateFunctionId();
          final VortexWorkerManager vortexWorkerManager = runningWorkers.get(workerId.get());
  
          if (taskletAggFunctionId.isPresent() &&
              !workerHasAggregateFunction(vortexWorkerManager.getId(), taskletAggFunctionId.get())) {
  
            // This assumes that all aggregate tasklets share the same user function.
            vortexWorkerManager.sendAggregateFunction(
                taskletAggFunctionId.get(),
                aggregateFunctionRepository.getAggregateFunction(taskletAggFunctionId.get()),
                tasklet.getUserFunction(),
                aggregateFunctionRepository.getPolicy(taskletAggFunctionId.get()));
            workerAggregateFunctionMap.get(vortexWorkerManager.getId()).add(taskletAggFunctionId.get());
          }
  
          vortexWorkerManager.launchTasklet(tasklet);
          schedulingPolicy.taskletLaunched(vortexWorkerManager, tasklet);
        }
      } finally {
        lock.unlock();
      }
    }
docs: Concurrency: Called by single scheduler thread.
  Parameter: Same tasklet can be launched multiple times.