code: <TInput, TOutput> void launchTasklet(final Tasklet<TInput, TOutput> tasklet) {
      assert !runningTasklets.containsKey(tasklet.getId());
      runningTasklets.put(tasklet.getId(), tasklet);
  
      if (tasklet.getAggregateFunctionId().isPresent()) {
        // function is aggregateable.
        final TaskletAggregateExecutionRequest<TInput> taskletAggregateExecutionRequest =
            new TaskletAggregateExecutionRequest<>(tasklet.getId(), tasklet.getAggregateFunctionId().get(),
                tasklet.getInput());
        vortexRequestor.sendAsync(reefTask, taskletAggregateExecutionRequest);
      } else {
        // function is not aggregateable.
        final TaskletExecutionRequest<TInput, TOutput> taskletExecutionRequest
            = new TaskletExecutionRequest<>(tasklet.getId(), tasklet.getUserFunction(), tasklet.getInput());
        vortexRequestor.sendAsync(reefTask, taskletExecutionRequest);
      }
    }
docs: Sends a request to launch a Tasklet on a {@link org.apache.reef.vortex.evaluator.VortexWorker}.