Delegate to the JournalWorker.
964408
public Date[] purgeDatastream(Context context,
                                  String pid,
                                  String datastreamID,
                                  Date startDT,
                                  Date endDT,
                                  String logMessage) throws ServerException {
        return worker.purgeDatastream(context,
                                      pid,
                                      datastreamID,
                                      startDT,
                                      endDT,
                                      logMessage);
    }
1283772
private synchronized void putDelegate(final List<Tasklet> tasklets, final VortexFutureDelegate delegate) {
    for (final Tasklet tasklet : tasklets) {
      taskletFutureMap.put(tasklet.getId(), delegate);
    }
  }
946866
private void processTask(ChildTaskContainer parent, MapRow row) throws IOException
   {
      Task task = parent.addTask();
      task.setName(row.getString("NAME"));
      task.setGUID(row.getUUID("UUID"));
      task.setText(1, row.getString("ID"));
      task.setDuration(row.getDuration("PLANNED_DURATION"));
      task.setRemainingDuration(row.getDuration("REMAINING_DURATION"));
      task.setHyperlink(row.getString("URL"));
      task.setPercentageComplete(row.getDouble("PERCENT_COMPLETE"));
      task.setNotes(getNotes(row.getRows("COMMENTARY")));
      task.setMilestone(task.getDuration().getDuration() == 0);

      ProjectCalendar calendar = m_calendarMap.get(row.getUUID("CALENDAR_UUID"));
      if (calendar != m_project.getDefaultCalendar())
      {
         task.setCalendar(calendar);
      }

      switch (row.getInteger("STATUS").intValue())
      {
         case 1: // Planned
         {
            task.setStart(row.getDate("PLANNED_START"));
            task.setFinish(task.getEffectiveCalendar().getDate(task.getStart(), task.getDuration(), false));
            break;
         }

         case 2: // Started
         {
            task.setActualStart(row.getDate("ACTUAL_START"));
            task.setStart(task.getActualStart());
            task.setFinish(row.getDate("ESTIMATED_FINISH"));
            if (task.getFinish() == null)
            {
               task.setFinish(row.getDate("PLANNED_FINISH"));
            }
            break;
         }

         case 3: // Finished
         {
            task.setActualStart(row.getDate("ACTUAL_START"));
            task.setActualFinish(row.getDate("ACTUAL_FINISH"));
            task.setPercentageComplete(Double.valueOf(100.0));
            task.setStart(task.getActualStart());
            task.setFinish(task.getActualFinish());
            break;
         }
      }

      setConstraints(task, row);

      processChildTasks(task, row);

      m_taskMap.put(task.getGUID(), task);

      List<MapRow> predecessors = row.getRows("PREDECESSORS");
      if (predecessors != null && !predecessors.isEmpty())
      {
         m_predecessorMap.put(task, predecessors);
      }

      List<MapRow> resourceAssignmnets = row.getRows("RESOURCE_ASSIGNMENTS");
      if (resourceAssignmnets != null && !resourceAssignmnets.isEmpty())
      {
         processResourceAssignments(task, resourceAssignmnets);
      }
   }
1053797
protected void writeTo( JournalFile journalFile ) throws JournalException
    {
    	journalFile.writeByte(type);
    	journalFile.writeLong(transactionId); 
    }
1301504
@Override
	protected void writeTo(JournalFile journalFile) throws JournalException
    {
        super.writeTo(journalFile);
        journalFile.writeInt(blockData.length);
        journalFile.write(blockData);
    }
93419
public synchronized void start() {
    if (mStopped) {
      Preconditions.checkState(mLoggingWorkerThread == null);
      mStopped = false;
      mLoggingWorkerThread = new Thread(new AuditLoggingWorker());
      mLoggingWorkerThread.setName(AUDIT_LOG_THREAD_NAME);
      mLoggingWorkerThread.start();
      LOG.info("AsyncUserAccessAuditLogWriter thread started.");
    }
  }
93568
def _assign_work_unit(self, node):
        """"""
        assert self.workqueue

        # Grab a unit of work
        scope, work_unit = self.workqueue.popitem(last=False)

        # Keep track of the assigned work
        assigned_to_node = self.assigned_work.setdefault(node, default=OrderedDict())
        assigned_to_node[scope] = work_unit

        # Ask the node to execute the workload
        worker_collection = self.registered_collections[node]
        nodeids_indexes = [
            worker_collection.index(nodeid)
            for nodeid, completed in work_unit.items()
            if not completed
        ]

        node.send_runtest_some(nodeids_indexes)
1283918
@Override
    public void writeJournalEntry(CreatorJournalEntry journalEntry,
                                  XMLEventWriter writer)
            throws JournalException {
        super.writeJournalEntry(journalEntry, writer);
    }
801471
private void schedulePeriodicWorks() {
        // JMX stats broadcast
        m_periodicWorks.add(scheduleWork(new Runnable() {
            @Override
            public void run() {
                // A null here was causing a steady stream of annoying but apparently inconsequential
                // NPEs during a debug session of an unrelated unit test.
                if (m_statsManager != null) {
                    m_statsManager.sendNotification();
                }
            }
        }, 0, StatsManager.POLL_INTERVAL, TimeUnit.MILLISECONDS));

        // clear login count
        m_periodicWorks.add(scheduleWork(new Runnable() {
            @Override
            public void run() {
                ScheduledExecutorService es = VoltDB.instance().getSES(false);
                if (es != null && !es.isShutdown()) {
                    es.submit(new Runnable() {
                        @Override
                        public void run()
                        {
                            long timestamp = System.currentTimeMillis();
                            m_flc.checkCounter(timestamp);
                        }
                    });
                }
            }
        }, 0, 10, TimeUnit.SECONDS));

        // small stats samples
        m_periodicWorks.add(scheduleWork(new Runnable() {
            @Override
            public void run() {
                SystemStatsCollector.asyncSampleSystemNow(false, false);
            }
        }, 0, 5, TimeUnit.SECONDS));

        // medium stats samples
        m_periodicWorks.add(scheduleWork(new Runnable() {
            @Override
            public void run() {
                SystemStatsCollector.asyncSampleSystemNow(true, false);
            }
        }, 0, 1, TimeUnit.MINUTES));

        // large stats samples
        m_periodicWorks.add(scheduleWork(new Runnable() {
            @Override
            public void run() {
                SystemStatsCollector.asyncSampleSystemNow(true, true);
            }
        }, 0, 6, TimeUnit.MINUTES));

        // export stream master check
        m_periodicWorks.add(scheduleWork(new Runnable() {
            @Override
            public void run() {
                checkExportStreamMastership();
            }
        }, 0, 1, TimeUnit.MINUTES));

        // other enterprise setup
        EnterpriseMaintenance em = EnterpriseMaintenance.get();
        if (em != null) { em.setupMaintenaceTasks(); }

        GCInspector.instance.start(m_periodicPriorityWorkThread, m_gcStats);
    }
18253
private void delegate(RequestDelegationService service, HttpServletRequest request, HttpServletResponse response,
        FilterChain filterChain) {
        try {
            service.delegate(request, response, filterChain);
        } catch (Exception e) {
            throw new RequestDelegationException(
                String.format("The request processing delegation failed: %s", e.getCause()), e);
        }
    }
723040
@Override
	public void collect(T record)
	{
		this.delegate.setInstance(record);
		try {
			for (int i = 0; i < writers.length; i++) {
				this.writers[i].emit(this.delegate);
			}
		}
		catch (IOException e) {
			throw new RuntimeException("Emitting the record caused an I/O exception: " + e.getMessage(), e);
		}
		catch (InterruptedException e) {
			throw new RuntimeException("Emitting the record was interrupted: " + e.getMessage(), e);
		}
	}
594451
private void updateWork(Task parentTask)
   {
      if (parentTask.hasChildTasks())
      {
         ProjectProperties properties = m_project.getProjectProperties();

         Duration actualWork = null;
         Duration baselineWork = null;
         Duration remainingWork = null;
         Duration work = null;

         for (Task task : parentTask.getChildTasks())
         {
            updateWork(task);

            actualWork = Duration.add(actualWork, task.getActualWork(), properties);
            baselineWork = Duration.add(baselineWork, task.getBaselineWork(), properties);
            remainingWork = Duration.add(remainingWork, task.getRemainingWork(), properties);
            work = Duration.add(work, task.getWork(), properties);
         }

         parentTask.setActualWork(actualWork);
         parentTask.setBaselineWork(baselineWork);
         parentTask.setRemainingWork(remainingWork);
         parentTask.setWork(work);
      }
   }
715359
private void executeAggregateTasklet(final ExecutorService commandExecutor,
                                       final MasterToWorkerRequest masterToWorkerRequest) {
    final TaskletAggregateExecutionRequest taskletAggregateExecutionRequest =
        (TaskletAggregateExecutionRequest) masterToWorkerRequest;

    assert aggregates.containsKey(taskletAggregateExecutionRequest.getAggregateFunctionId());

    final AggregateContainer aggregateContainer =
        aggregates.get(taskletAggregateExecutionRequest.getAggregateFunctionId());
    final TaskletAggregationRequest aggregationRequest = aggregateContainer.getTaskletAggregationRequest();

    commandExecutor.submit(new Runnable() {
      @Override
      public void run() {
        try {
          aggregateContainer.scheduleTasklet(taskletAggregateExecutionRequest.getTaskletId());
          final Object result = aggregationRequest.executeFunction(taskletAggregateExecutionRequest.getInput());
          aggregateContainer.taskletComplete(taskletAggregateExecutionRequest.getTaskletId(), result);
        } catch (final Exception e) {
          aggregateContainer.taskletFailed(taskletAggregateExecutionRequest.getTaskletId(), e);
        }
      }
    });
  }
411937
private void insertOrUpdateTasks(TaskStartEvent event) {
        Assignment old = event.getOldAssignment();
        Assignment current = event.getNewAssignment();

        Map<Integer, String> task2Component = event.getTask2Component();
        List<KVSerializable> taskTrackList = new ArrayList<>();

        // assign
        if (old == null) {
            Set<ResourceWorkerSlot> workers = current.getWorkers();
            logger.info("old workers are null, assigned workers:{}", Joiner.on(",").join(workers));
            for (ResourceWorkerSlot worker : workers) {
                Set<Integer> tasks = worker.getTasks();
                for (Integer task : tasks) {
                    TaskTrack track = new TaskTrack(
                            event.getClusterMetricsContext().getClusterName(), event.getTopologyId());
                    track.setStart(new Date(event.getTimestamp()));
                    track.setComponent(task2Component.get(task));
                    track.setHost(worker.getHostname());
                    track.setPort(worker.getPort());
                    track.setTaskId(task);

                    taskTrackList.add(track);
                }
            }
        } else { // rebalance, we only insert newly assigned tasks
            Set<ResourceWorkerSlot> oldWorkers = old.getWorkers();
            Joiner joiner = Joiner.on(",");
            logger.info("old workers:{}, new workers:{}", joiner.join(oldWorkers), joiner.join(current.getWorkers()));
            for (ResourceWorkerSlot worker : current.getWorkers()) {
                // a new worker, insert all tasks
                if (!oldWorkers.contains(worker)) {
                    for (Integer task : worker.getTasks()) {
                        TaskTrack track = new TaskTrack(
                                event.getClusterMetricsContext().getClusterName(),
                                event.getTopologyId());
                        track.setStart(new Date(event.getTimestamp()));
                        track.setComponent(task2Component.get(task));
                        track.setHost(worker.getHostname());
                        track.setPort(worker.getPort());
                        track.setTaskId(task);

                        taskTrackList.add(track);
                    }
                } else {
                    for (Integer task : worker.getTasks()) {
                        ResourceWorkerSlot oldWorker = old.getWorkerByTaskId(task);
                        if (oldWorker != null) {

                            // update end time of old task
                            TaskTrack oldTrack = new TaskTrack(
                                    event.getClusterMetricsContext().getClusterName(),
                                    event.getTopologyId());
                            oldTrack.setEnd(new Date(event.getTimestamp()));
                            oldTrack.setTaskId(task);
                            oldTrack.setHost(oldWorker.getHostname());
                            oldTrack.setPort(oldWorker.getPort());
                            taskTrackList.add(oldTrack);

                            // insert new task
                            TaskTrack track = new TaskTrack(
                                    event.getClusterMetricsContext().getClusterName(),
                                    event.getTopologyId());
                            track.setStart(new Date());
                            track.setComponent(task2Component.get(task));
                            track.setHost(worker.getHostname());
                            track.setPort(worker.getPort());
                            track.setTaskId(task);
                            taskTrackList.add(track);
                        }
                    }
                }
            }
        }
        if (taskTrackList.size() > 0) {
            batchAdd(taskTrackList, TABLE_TASK_TRACK);
        }
    }
226735
private void writeResourceAssignmentWorkgroupFields(ResourceAssignmentWorkgroupFields record) throws IOException
   {
      m_buffer.setLength(0);

      m_buffer.append(MPXConstants.RESOURCE_ASSIGNMENT_WORKGROUP_FIELDS_RECORD_NUMBER);
      m_buffer.append(m_delimiter);
      m_buffer.append(format(record.getMessageUniqueID()));
      m_buffer.append(m_delimiter);
      m_buffer.append(record.getConfirmed() ? "1" : "0");
      m_buffer.append(m_delimiter);
      m_buffer.append(record.getResponsePending() ? "1" : "0");
      m_buffer.append(m_delimiter);
      m_buffer.append(format(formatDateTimeNull(record.getUpdateStart())));
      m_buffer.append(m_delimiter);
      m_buffer.append(format(formatDateTimeNull(record.getUpdateFinish())));
      m_buffer.append(m_delimiter);
      m_buffer.append(format(record.getScheduleID()));

      stripTrailingDelimiters(m_buffer);
      m_buffer.append(MPXConstants.EOL);

      m_writer.write(m_buffer.toString());
   }
390956
def start(self):
        """"""
        assert not self.interrupted
        for thread in self.worker_threads:
            thread.start()
        WorkerThread.start(self)
317476
private void startWorkerIfNeeded() {
    // Perform a compare-and-swap retry loop for synchronization to make sure we don't start more
    // workers than desired.
    int currentCount = mPendingWorkers.get();
    while (currentCount < mMaxConcurrency) {
      int updatedCount = currentCount + 1;
      if (mPendingWorkers.compareAndSet(currentCount, updatedCount)) {
        // Start a new worker.
        FLog.v(TAG, "%s: starting worker %d of %d", mName, updatedCount, mMaxConcurrency);
        mExecutor.execute(mTaskRunner);
        break;
      }
      // else: compareAndSet failed due to race; snapshot the new count and try again
      FLog.v(TAG, "%s: race in startWorkerIfNeeded; retrying", mName);
      currentCount = mPendingWorkers.get();
    }
  }