Calls method in replay context so that no journal entries are created,
    expected_side_effects are checked, and no asynchronous task is started.
    The journal entry is only used to fetch side-effects results.
184366
def replay(journal_entry, function, *args, **kwargs):
    '''
    
    '''
    # Starts the fiber section
    section = fiber.WovenSection()
    section.enter()

    # Check if this is the first recording in the fiber section
    journal_mode = section.state.get(RECMODE_TAG, None)
    is_first = journal_mode is None

    if is_first:
        section.state[RECMODE_TAG] = JournalMode.replay
        section.state[JOURNAL_ENTRY_TAG] = IJournalReplayEntry(journal_entry)

    result = function(*args, **kwargs)

    # We don't want anything asynchronous to be called,
    # so we abort the fiber section
    section.abort(result)
    # side effects are returned in sake of making sure that
    # all the side effects expected have been consumed (called)
    return result
564603
private void prepareExecutionAttempt(AsyncExecutionContextImpl<W> executionContext) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) {
            Tr.event(tc, "Execution {0} enqueuing new execution attempt", executionContext.getId());
        }

        AsyncAttemptContextImpl<W> attemptContext = new AsyncAttemptContextImpl<>(executionContext);

        TimeoutState timeout = FaultToleranceStateFactory.INSTANCE.createTimeoutState(executorService, timeoutPolicy, metricRecorder);
        attemptContext.setTimeoutState(timeout);

        if (!circuitBreaker.requestPermissionToExecute()) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) {
                Tr.event(tc, "Execution {0} Circuit Breaker open, not executing", executionContext.getId());
            }

            MethodResult<W> result = MethodResult.failure(new CircuitBreakerOpenException());
            processEndOfAttempt(attemptContext, result);
            return;
        }

        attemptContext.setCircuitBreakerPermittedExecution(true);

        timeout.start();

        ExecutionReference ref = bulkhead.submit((reservation) -> runExecutionAttempt(attemptContext, reservation), getExceptionHandler(attemptContext));

        if (!ref.wasAccepted()) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) {
                Tr.event(tc, "Execution {0} bulkhead rejected execution", executionContext.getId());
            }

            processEndOfAttempt(attemptContext, MethodResult.failure(new BulkheadException()));
            return;
        }

        timeout.setTimeoutCallback(handleExceptions(() -> timeout(attemptContext, ref), attemptContext, executionContext));

        // Update what we should do if the user cancels the execution
        executionContext.setCancelCallback((mayInterrupt) -> {
            ref.abort(mayInterrupt);
            processEndOfAttempt(attemptContext, MethodResult.failure(new CancellationException()));
        }
1227695
function reportResult(devRes, scriptRes, method) {
    		debug('in performRetry', devRes, scriptRes, method);
    		var returnData = {
    			'numAttempts': context.currentAttempt,
    			'maxAttempts': context.maxAttempts,
    			// 'startTime': new Date(),
    			// 'finishTime': new Date(),
    			'scriptRes': scriptRes,
    			'evalStr': context.evalStr,
    			'delay': context.delay,
    			'register': context.register,
    		};
    		var devResKeys = Object.keys(devRes);
    		devResKeys.forEach(function(key) {
    			returnData[key] = devRes[key];
    		});
    		valChecker.defered[method](returnData);
    	}
1373641
def call(record, reflection:, skip_record:)
      if reflection.collection?
        _reindex_collection(reflection, record, skip_record)
      else
        _reindex_single(reflection, record, skip_record)
      end
    end
423046
@Override
    protected void recordResult(final RootNode node, final boolean success, final RootNodeExecutionContext context) {

        if (success) {
            if (log.isTraceEnabled()) {

                log.trace("node success");
            }

            node.getResult().setFinished();
            context.getNotificationDistributor().onNodeFinished(node);

        } else {

            final List<SubstepExecutionFailure> failures = context.getFailures();

            if (log.isDebugEnabled()) {

                log.debug("node failures");
            }

            // have a look at the constituent features
            List<FeatureNode> featureNodes = node.getChildren();
            boolean rootNodeStateSet = false;
            for (FeatureNode featureNode : node.getChildren()){

                if(featureNode.getResult().getResult() == ExecutionResult.FAILED && !featureNode.getResult().getFailure().isNonCritical()){
                    // we've got one valid feature failure, fail the root node
                    SubstepsRuntimeException e = new SubstepsRuntimeException("At least one critical Feature failed");
                    SubstepExecutionFailure sef = new SubstepExecutionFailure(e, node, ExecutionResult.FAILED);

                    context.getNotificationDistributor().onNodeFailed(node, e);
                    rootNodeStateSet = true;
                    break;
                }
            }

            if (!rootNodeStateSet){
                // got this far, must be ok
                node.getResult().setFinished();
                context.getNotificationDistributor().onNodeFinished(node);
            }


//            final Throwable lastException;
//            boolean nonCritical = false;
//            if (!failures.isEmpty()) {
//
//                final SubstepExecutionFailure lastFailure = failures.get(failures.size() - 1);
//                // just notify on the last one in..?
//                lastException = lastFailure.getCause();
//                node.getResult().setScreenshot(lastFailure.getScreenshot());
//                nonCritical = lastFailure.isNonCritical();
//            }
//            else {
//                lastException = new SubstepsRuntimeException("Error throw during startup, initialisation issue ?");
//                lastException.fillInStackTrace();
//                SubstepExecutionFailure sef = new SubstepExecutionFailure(lastException, node, ExecutionResult.FAILED);
//            }



//            SubstepExecutionFailure sef = new SubstepExecutionFailure(lastException, node, ExecutionResult.FAILED);
//            sef.setNonCritical(nonCritical);
//            context.getNotificationDistributor().onNodeFailed(node, lastException);

        }
    }
1344535
public void replayTo(Log destination) {
		synchronized (this.lines) {
			for (Line line : this.lines) {
				logTo(destination, line.getLevel(), line.getMessage(),
						line.getThrowable());
			}
			this.lines.clear();
		}
	}
502876
protected void retryIfReady() {
        Log.v(Log.TAG_SYNC, "[retryIfReady()] stateMachine => " + stateMachine.getState().toString());
        // check if state is still IDLE (ONLINE), then retry now.
        if (stateMachine.getState().equals(ReplicationState.IDLE)) {
            Log.v(Log.TAG_SYNC, "%s RETRYING, to transfer missed revisions...", this);
            cancelRetryFuture();
            retry();
        }
    }
1255424
private void safeCustomDelay(PollingStrategyContext pollingStrategyContext) {
        try {
            pollingStrategy.getDelayStrategy().delayBeforeNextRetry(pollingStrategyContext);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
    }
987104
function () {
    var self = this;
    if (self._stopped)
      throw new Error("Called waitUntilCaughtUp on stopped handle!");

    // Calling waitUntilCaughtUp requries us to wait for the oplog connection to
    // be ready.
    self._readyFuture.wait();

    while (!self._stopped) {
      // We need to make the selector at least as restrictive as the actual
      // tailing selector (ie, we need to specify the DB name) or else we might
      // find a TS that won't show up in the actual tail stream.
      try {
        var lastEntry = self._oplogLastEntryConnection.findOne(
          OPLOG_COLLECTION, self._baseOplogSelector,
          {fields: {ts: 1}, sort: {$natural: -1}});
        break;
      } catch (e) {
        // During failover (eg) if we get an exception we should log and retry
        // instead of crashing.
        Meteor._debug("Got exception while reading last entry: " + e);
        Meteor._sleepForMs(100);
      }
    }

    if (self._stopped)
      return;

    if (!lastEntry) {
      // Really, nothing in the oplog? Well, we've processed everything.
      return;
    }

    var ts = lastEntry.ts;
    if (!ts)
      throw Error("oplog entry without ts: " + EJSON.stringify(lastEntry));

    if (self._lastProcessedTS && ts.lessThanOrEqual(self._lastProcessedTS)) {
      // We've already caught up to here.
      return;
    }


    // Insert the future into our list. Almost always, this will be at the end,
    // but it's conceivable that if we fail over from one primary to another,
    // the oplog entries we see will go backwards.
    var insertAfter = self._catchingUpFutures.length;
    while (insertAfter - 1 > 0
           && self._catchingUpFutures[insertAfter - 1].ts.greaterThan(ts)) {
      insertAfter--;
    }
    var f = new Future;
    self._catchingUpFutures.splice(insertAfter, 0, {ts: ts, future: f});
    f.wait();
  }
1043916
private void handleRunAsync(RunAsync runAsync) {
		final long timeToRun = runAsync.getTimeNanos();
		final long delayNanos;

		if (timeToRun == 0 || (delayNanos = timeToRun - System.nanoTime()) <= 0) {
			// run immediately
			try {
				runAsync.getRunnable().run();
			} catch (Throwable t) {
				log.error("Caught exception while executing runnable in main thread.", t);
				ExceptionUtils.rethrowIfFatalErrorOrOOM(t);
			}
		}
		else {
			// schedule for later. send a new message after the delay, which will then be immediately executed
			FiniteDuration delay = new FiniteDuration(delayNanos, TimeUnit.NANOSECONDS);
			RunAsync message = new RunAsync(runAsync.getRunnable(), timeToRun);

			final Object envelopedSelfMessage = envelopeSelfMessage(message);

			getContext().system().scheduler().scheduleOnce(delay, getSelf(), envelopedSelfMessage,
					getContext().dispatcher(), ActorRef.noSender());
		}
	}
3850
void runCheckAsync(final EvaluatorManager manager) {

    final String evaluatorId = manager.getId();
    LOG.log(Level.FINEST, "Idle check for Evaluator: {0}", manager);

    this.executor.submit(new Runnable() {

      @Override
      public void run() {

        LOG.log(Level.FINEST, "Idle check for Evaluator {0} - begin", evaluatorId);

        while (!manager.isClosed()) {
          try {

            LOG.log(Level.FINEST,
                "Waiting for Evaluator {0} to close: Sleep for {1} ms",
                new Object[] {evaluatorId, waitInMillis});

            Thread.sleep(waitInMillis);

          } catch (final InterruptedException e) {
            LOG.log(Level.SEVERE, "Thread interrupted while waiting for Evaluator to finish.");
            throw new RuntimeException(e);
          }
        }

        manager.checkIdlenessSource();

        LOG.log(Level.FINEST, "Idle check for Evaluator {0} - end", evaluatorId);
      }

      @Override
      public String toString() {
        return "CheckIdle: " + evaluatorId;
      }
    });
  }
160270
synchronized ShortVoid journal(RequestInfo reqInfo, 
      long segmentTxId, long firstTxnId, 
      int numTxns, byte[] records) throws IOException {
    checkJournalStorageFormatted();
    checkWriteRequest(reqInfo);

    if (curSegment == null) {
      checkSync(false, "Can't write, no segment open");
    }
    
    if (curSegmentTxId != segmentTxId) {
      // Sanity check: it is possible that the writer will fail IPCs
      // on both the finalize() and then the start() of the next segment.
      // This could cause us to continue writing to an old segment
      // instead of rolling to a new one, which breaks one of the
      // invariants in the design. If it happens, abort the segment
      // and throw an exception.
      JournalOutOfSyncException e = new JournalOutOfSyncException(
          "Writer out of sync: it thinks it is writing segment " + segmentTxId
          + " but current segment is " + curSegmentTxId);
      abortCurSegment();
      throw e;
    }
      
    if (nextTxId != firstTxnId) {
      checkSync(false, "Can't write txid " + firstTxnId
          + " expecting nextTxId=" + nextTxId);
    }
    
    long lastTxnId = firstTxnId + numTxns - 1;
    if (LOG.isTraceEnabled()) {
      LOG.trace("Writing txid " + firstTxnId + "-" + lastTxnId);
    }

    // If the edit has already been marked as committed, we know
    // it has been fsynced on a quorum of other nodes, and we are
    // "catching up" with the rest. Hence we do not need to fsync.
    boolean isLagging = lastTxnId <= committedTxnId.get();
    boolean shouldFsync = !isLagging;
    
    curSegment.writeRaw(records, 0, records.length);
    curSegment.setReadyToFlush();
    long start = System.nanoTime();
    curSegment.flush(shouldFsync);
    long time = DFSUtil.getElapsedTimeMicroSeconds(start);
    currentSegmentWrittenBytes += records.length;
    
    metrics.addSync(time);
    
    if (time / 1000 > WARN_SYNC_MILLIS_THRESHOLD) {
      LOG.warn("Sync of transaction range " + firstTxnId + "-" + lastTxnId + 
               " took " + (time / 1000) + "ms");
    }

    if (isLagging) {
      // This batch of edits has already been committed on a quorum of other
      // nodes. So, we are in "catch up" mode. This gets its own metric.
      metrics.batchesWrittenWhileLagging.inc(1);
    }
    
    metrics.batchesWritten.inc(1);
    metrics.bytesWritten.inc(records.length);
    metrics.txnsWritten.inc(numTxns);
    
    highestWrittenTxId = lastTxnId;
    metrics.setLastWrittenTxId(highestWrittenTxId);
    metrics.setCurrentTxnsLag(getCurrentLagTxns());
    nextTxId = lastTxnId + 1;
    return ShortVoid.instance;
  }
1257755
private boolean executeWrites(List<Write> toExecute) {
        log.debug("{}: Executing {} writes.", this.traceObjectId, toExecute.size());
        for (int i = 0; i < toExecute.size(); i++) {
            Write w = toExecute.get(i);
            try {
                // Record the beginning of a new attempt.
                int attemptCount = w.beginAttempt();
                if (attemptCount > this.config.getMaxWriteAttempts()) {
                    // Retried too many times.
                    throw new RetriesExhaustedException(w.getFailureCause());
                }

                // Invoke the BookKeeper write.
                w.getWriteLedger().ledger.asyncAddEntry(w.data.array(), w.data.arrayOffset(), w.data.getLength(), this::addCallback, w);
            } catch (Throwable ex) {
                // Synchronous failure (or RetriesExhausted). Fail current write.
                boolean isFinal = !isRetryable(ex);
                w.fail(ex, isFinal);

                // And fail all remaining writes as well.
                for (int j = i + 1; j < toExecute.size(); j++) {
                    toExecute.get(j).fail(new DurableDataLogException("Previous write failed.", ex), isFinal);
                }

                return false;
            }
        }

        // Success.
        return true;
    }
720964
protected void cleanAndReindex( boolean async ) {
        final IndexWriter writer = getIndexWriter();
        scan(async, getIndexWriter(), new Callable<Void>() {
            @SuppressWarnings( "synthetic-access" )
            @Override
            public Void call() throws Exception {
                writer.clearAllIndexes();
                reindexContent(true, writer);
                return null;
            }
        });
    }
419581
@Override
    public void run() {
        Object captured = capturedRef.get();
        if (captured == null || releaseTtlValueReferenceAfterRun && !capturedRef.compareAndSet(captured, null)) {
            throw new IllegalStateException("TTL value reference is released after run!");
        }

        Object backup = replay(captured);
        try {
            runnable.run();
        } finally {
            restore(backup);
        }
    }
10115
private synchronized void ensureIsRun() {
        if (!isRun) {
            isRun = true;
            try {
                result = executeProc(args);
                exception = null;
            } catch (ProcError e) {
                throw e.asRuntimeException();
            } catch (Throwable t) {
                exception = t;
                result = null;
            }
        }
    }
649920
private void performPostCompilationTasksInternal() {
    if (options.devMode == DevMode.START_AND_END) {
      runValidityCheck();
    }
    setProgress(1.0, "recordFunctionInformation");

    if (tracker != null) {
      tracker.outputTracerReport();
    }
  }