Delegate to the JournalWorker.
5282
public String ingest(Context context,
                         InputStream serialization,
                         String logMessage,
                         String format,
                         String encoding,
                         String pid) throws ServerException {
        return worker.ingest(context,
                             serialization,
                             logMessage,
                             format,
                             encoding,
                             pid);
    }
230107
function worker () {
  var indexedDB = self.indexedDB
  var db
  var methodMap = {
    connect: connect,
    disconnect: disconnect,
    create: create,
    update: update,
    delete: remove,
    deleteAll: removeAll
  }

  self.addEventListener('message', function (event) {
    var data = event.data
    var id = data.id
    var method = data.method

    methodMap[method](data, function (error, result, transfer) {
      if (error) {
        self.postMessage({
          id: id, error: '' + error
        })
        return
      }

      self.postMessage({
        id: id, result: result
      }, transfer)
    })
  })


  function connect (data, callback) {
    var request = indexedDB.open(data.name)
    var typesArray = data.typesArray

    request.onerror = errorConnection
    request.onupgradeneeded = handleUpgrade
    request.onsuccess = handleSuccess

    function handleSuccess (event) {
      var i, j

      db = event.target.result

      for (i = 0, j = typesArray.length; i < j; i++)
        if (!~Array.prototype.indexOf.call(
          db.objectStoreNames, typesArray[i])) {
          reconnect()
          return
        }

      loadRecords()
    }

    function handleUpgrade (event) {
      var i, j, type

      db = event.target.result

      for (i = 0, j = typesArray.length; i < j; i++) {
        type = typesArray[i]
        if (!~Array.prototype.indexOf.call(db.objectStoreNames, type))
          db.createObjectStore(type, { keyPath: primaryKey })
      }

      for (i = 0, j = db.objectStoreNames.length; i < j; i++) {
        type = db.objectStoreNames[i]
        if (!~Array.prototype.indexOf.call(typesArray, type))
          db.deleteObjectStore(type)
      }
    }

    function reconnect () {
      var version = (db.version || 1) + 1

      db.close()
      request = indexedDB.open(data.name, version)
      request.onerror = errorReconnection
      request.onupgradeneeded = handleUpgrade
      request.onsuccess = function (event) {
        db = event.target.result
        loadRecords(db)
      }
    }

    function loadRecords () {
      var counter = 0
      var payload = {}
      var transfer = []
      var i, j

      for (i = 0, j = typesArray.length; i < j; i++)
        loadType(typesArray[i])

      function loadType (type) {
        var transaction = db.transaction(type, 'readonly')
        var objectStore = transaction.objectStore(type)
        var cursor = objectStore.openCursor()

        payload[type] = []
        cursor.onsuccess = function (event) {
          var iterator = event.target.result
          if (iterator) {
            payload[type].push(iterator.value[dataKey])
            transfer.push(iterator.value[dataKey].buffer)
            iterator.continue()
            return
          }
          counter++
          if (counter === typesArray.length)
            callback(null, payload, transfer)
        }
        cursor.onerror = errorIteration
      }
    }

    function errorConnection () {
      callback('The database connection could not be established.')
    }

    function errorReconnection () {
      callback('An attempt to reconnect failed.')
    }

    function errorIteration () {
      callback('Failed to read record.')
    }
  }


  function disconnect () {
    db.close()
  }


  function create (data, callback) {
    var recordsLength = Object.keys(data.records).length
    var type = data.type
    var transaction = db.transaction(type, 'readwrite')
    var objectStore = transaction.objectStore(type)
    var id, record, object, request, counter = 0

    for (id in data.records) {
      record = data.records[id]
      object = {}
      object[primaryKey] = type + delimiter + id
      object[dataKey] = record
      request = objectStore.add(object)
      request.onsuccess = check
      request.onerror = error
    }

    function check () {
      counter++
      if (counter === recordsLength) callback()
    }

    function error () {
      callback('A record could not be created.')
    }
  }


  function update (data, callback) {
    var recordsLength = Object.keys(data.records).length
    var type = data.type
    var transaction = db.transaction(type, 'readwrite')
    var objectStore = transaction.objectStore(type)
    var id, record, object, request, counter = 0

    for (id in data.records) {
      record = data.records[id]
      object = {}

      // Here it should be assumed that the type and delimiter are already
      // supplied, rather than needing to add it here.
      object[primaryKey] = id

      object[dataKey] = record
      request = objectStore.put(object)
      request.onsuccess = check
      request.onerror = error
    }

    function check () {
      counter++
      if (counter === recordsLength) callback()
    }

    function error () {
      callback('A record could not be updated.')
    }
  }


  function remove (data, callback) {
    var type = data.type
    var ids = data.ids
    var transaction = db.transaction(type, 'readwrite')
    var objectStore = transaction.objectStore(type)
    var i, j, id, request, counter = 0

    for (i = 0, j = ids.length; i < j; i++) {
      id = ids[i]
      request = objectStore.delete(type + delimiter + id)
      request.onsuccess = check
      request.onerror = error
    }

    function check () {
      counter++
      if (counter === ids.length) callback()
    }

    function error () {
      callback('A record could not be deleted.')
    }
  }


  function removeAll (data, callback) {
    var type = data.type
    var transaction = db.transaction(type, 'readwrite')
    var objectStore = transaction.objectStore(type)
    var request = objectStore.clear()
    request.onsuccess = function () { callback() }
    request.onerror = error

    function error () {
      callback('Not all records could be deleted.')
    }
  }
}
374264
void attach() {
    editors.addAll(editorSource.create(workingCopy.size(), 0));
    final int size = workingCopy.size();
    for (int i = 0; i < size; i++) {
      chain.attach(workingCopy.get(i), editors.get(i));
    }
  }
104677
@Override
	protected void writeTo(JournalFile journalFile) throws JournalException
    {
    	super.writeTo(journalFile);
    	journalFile.writeInt(blockSize);
    	journalFile.writeInt(oldBlockCount);
    	journalFile.writeInt(newBlockCount);
    }
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);
		}
	}
1250316
def worker_thread(context):
    """"""
    queue = context.task_queue
    parameters = context.worker_parameters

    if parameters.initializer is not None:
        if not run_initializer(parameters.initializer, parameters.initargs):
            context.state = ERROR
            return

    for task in get_next_task(context, parameters.max_tasks):
        execute_next_task(task)
        queue.task_done()
1281030
@Override
    public void prepareToWriteJournalEntry() throws JournalException {
        if (open) {
            currentJournal.closeIfAppropriate();

            if (!currentJournal.isOpen()) {
                currentJournal =
                        new JournalOutputFile(this,
                                              filenamePrefix,
                                              journalDirectory,
                                              sizeLimit,
                                              ageLimit);
            }
        }
    }
883320
protected void writeJournalEntry(CreatorJournalEntry journalEntry,
                                     XMLEventWriter writer)
            throws JournalException {
        try {
            writeJournaEntryStartTag(journalEntry, writer);

            new ContextXmlWriter().writeContext(journalEntry.getContext(),
                                                writer);

            writeArguments(journalEntry.getArgumentsMap(), writer);

            putEndTag(writer, QNAME_TAG_JOURNAL_ENTRY);
            writer.flush();
        } catch (XMLStreamException e) {
            throw new JournalException(e);
        }
    }
484991
void addWorkWrapper(WorkWrapper ww)
   {
      synchronized (activeWorkWrappers)
      {
         activeWorkWrappers.add(ww);

         if (statisticsEnabled)
            statistics.setWorkActive(activeWorkWrappers.size());
      }
   }
1099207
@Override
  public void commitBlock(long workerId, long usedBytesOnTier, String tierAlias, long blockId,
      long length) throws NotFoundException, UnavailableException {
    LOG.debug("Commit block from workerId: {}, usedBytesOnTier: {}, blockId: {}, length: {}",
        workerId, usedBytesOnTier, blockId, length);

    MasterWorkerInfo worker = mWorkers.getFirstByField(ID_INDEX, workerId);
    // TODO(peis): Check lost workers as well.
    if (worker == null) {
      throw new NotFoundException(ExceptionMessage.NO_WORKER_FOUND.getMessage(workerId));
    }

    // Lock the worker metadata first.
    try (JournalContext journalContext = createJournalContext()) {
      synchronized (worker) {
        try (LockResource lr = lockBlock(blockId)) {
          Optional<BlockMeta> block = mBlockStore.getBlock(blockId);
          if (!block.isPresent() || block.get().getLength() != length) {
            if (block.isPresent() && block.get().getLength() != Constants.UNKNOWN_SIZE) {
              LOG.warn("Rejecting attempt to change block length from {} to {}",
                  block.get().getLength(), length);
            } else {
              mBlockStore.putBlock(blockId, BlockMeta.newBuilder().setLength(length).build());
              BlockInfoEntry blockInfo =
                  BlockInfoEntry.newBuilder().setBlockId(blockId).setLength(length).build();
              journalContext.append(JournalEntry.newBuilder().setBlockInfo(blockInfo).build());
            }
          }

          // Update the block metadata with the new worker location.
          mBlockStore.addLocation(blockId, BlockLocation.newBuilder()
              .setWorkerId(workerId)
              .setTier(tierAlias)
              .build());
          // This worker has this block, so it is no longer lost.
          mLostBlocks.remove(blockId);

          // Update the worker information for this new block.
          // TODO(binfan): when retry commitBlock on master is expected, make sure metrics are not
          // double counted.
          worker.addBlock(blockId);
          worker.updateUsedBytes(tierAlias, usedBytesOnTier);
          worker.updateLastUpdatedTimeMs();
        }
      }
    }
  }
1066997
void scheduleHeronWorkers(PackingPlan topologyPacking) throws ContainerAllocationException {
    this.componentRamMap = topologyPacking.getComponentRamDistribution();
    scheduleHeronWorkers(topologyPacking.getContainers());
  }
1030157
void sendDeltaDoWorkRejected()
   {
      doWorkRejected.incrementAndGet();

      if (trace)
         log.tracef("sendDeltaDoWorkRejected: %s", workManagers);

      if (own != null && transport != null && transport.isInitialized())
      {
         for (Address address : workManagers)
         {
            if (!own.equals(address))
               transport.deltaDoWorkRejected(address);
         }
      }
   }
556481
protected void delegate(StreamMessage<T> delegate) {
        requireNonNull(delegate, "delegate");

        if (!delegateUpdater.compareAndSet(this, null, delegate)) {
            throw new IllegalStateException("delegate set already");
        }

        if (abortPending != 0) {
            delegate.abort();
        }

        if (!completionFuture().isDone()) {
            delegate.completionFuture().handle((unused, cause) -> {
                if (cause == null) {
                    completionFuture().complete(null);
                } else {
                    completionFuture().completeExceptionally(cause);
                }
                return null;
            }).exceptionally(CompletionActions::log);
        }

        safeOnSubscribeToDelegate();
    }
315854
def delegate(self, success, **kwargs):
        """"""
        if success:
            self.succeeded(**kwargs)
        else:
            self.failed(**kwargs)
1273751
public void commit( SynchronizationBarrier barrier ) throws JournalException
    {
    	// Refuse any further operation in case of previous failure
    	if (failing)
    		throw new JournalException("Store journal is failing");
    	
    	// Register as a barrier participant
    	if (barrier != null)
    		barrier.addParty();
        
        boolean newFlushRequired;
        synchronized (journalWriteQueue)
        {
        	writeCommit(barrier);
        	
        	newFlushRequired = !flushingJournal;
        	if (newFlushRequired)
        		flushingJournal = true;
        	unflushedJournalSize = 0;
        }
        
        if (newFlushRequired)
        {
	        // Flush journal to store (asynchronous)
	        try
	        {
	        	asyncTaskManager.execute(flushJournalAsyncTask);
	        }
	    	catch (JMSException e)
	    	{
	    		throw new JournalException("Cannot flush journal asynchronously : "+e);
	    	}
        }
    }
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();
    }
  }
734109
def schedule(self, task):
        """"""
        self.task_manager.register(task)
        self.worker_manager.dispatch(task)