Request an fsync for the allocated sequence, and request a flush
to occur at the next service batch.

@param sequence the sequence id allocated from allocateFsyncSequence()
@param result the completion to be called after the fsync completes
551256
public void fsync(long sequence, K key, Result<Boolean> result)
  {
    scheduleFsync(sequence, key, result);
    flush();
  }
776184
@Override
    public IPromise then(Callback resultCB) {
        // FIXME: this can be implemented more efficient
        while( !lock.compareAndSet(false,true) ) {}
        try {
            if (resultReceiver != null)
                throw new RuntimeException("Double register of promise listener");
            resultReceiver = resultCB;
            if (hadResult) {
                hasFired = true;
                if (nextFuture == null) {
                    nextFuture = new Promise(result, error);
                    lock.set(false);
                    resultCB.complete(result, error);
                } else {
                    lock.set(false);
                    resultCB.complete(result, error);
                    nextFuture.complete(result, error);
                    return nextFuture;
                }
            }
            if (resultCB instanceof IPromise) {
                return (IPromise) resultCB;
            }
            lock.set(false);
            while( !lock.compareAndSet(false,true) ) {}
            if (nextFuture == null) {
                return nextFuture = new Promise();
            } else {
                return nextFuture;
            }
        } finally {
            lock.set(false);
        }
    }
939367
@Override
  protected void flushAndSync(boolean durable) throws IOException {
    if (outputStream == null) {
      throw new IOException("Trying to use aborted output stream!");
    }
    if (doubleBuf.isFlushed()) {
      return;
    }
    // Under the covers, this invokes a method that does not return until
    // we have successfully persisted the buffer to a quorum of bookies.
    doubleBuf.flushTo(outputStream);
  }
1277452
public void close(Result<Boolean> result)
  {
    _lifecycle.toDestroy();
    
    SegmentStream nodeStream = _nodeStream;
    _nodeStream = null;
      
    if (nodeStream != null) {
      nodeStream.closeFsync(result.then(v->closeImpl()));
    }
    else {
      result.ok(true);
    }
  }
685712
public byte[] sendEnquireLinkResp(OutputStream os, int sequenceNumber)
            throws IOException {
        byte[] b = pduComposer.enquireLinkResp(sequenceNumber);
        writeAndFlush(os, b);
        return b;
    }
477039
public void registerSequenceQuery(long sequence, Runnable query) {
    // Add a query to be run once the session's sequence number reaches the given sequence number.
    List<Runnable> queries = this.sequenceQueries.computeIfAbsent(sequence, v -> new LinkedList<Runnable>());
    queries.add(query);
  }
918515
protected AsyncResponseImpl checkFutureResponse(Message inMessage, Object result) {
        if (result instanceof CompletionStage) {
            final CompletionStage<?> stage = (CompletionStage<?>)result;
            final AsyncResponseImpl asyncResponse = new AsyncResponseImpl(inMessage);
            stage.whenComplete((v, t) -> {
                if (t instanceof CancellationException) {
                    asyncResponse.cancel();
                } else {
                    asyncResponse.resume(v != null ? v : t);
                }
            });
            return asyncResponse;
        }
        return null;
    }
37887
protected void success(T result) {
    this.result = result;
    this.completed = true;
    for (Callback<T, E> callback : toBeNotified) {
      callback.success(result);
    }
  }
805399
private void afterDataFsync(Result<Boolean> result,
                              int position,
                              FsyncType fsyncType,
                              ArrayList<SegmentFsyncCallback> listeners)
  {
    try {
      completeIndex(position);
      
      Result<Boolean> cont = result.then((v,r)->
        afterIndexFsync(r, fsyncType, listeners));

      if (fsyncType.isSchedule()) {
        _sOut.fsyncSchedule(cont);
      }
      else {
        _sOut.fsync(cont);
        
      }
    } catch (Throwable e) {
      result.fail(e);
    }
  }
645210
public void processFlushQuery(ControlAreYouFlushed flushQuery) throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "processFlushQuery", new Object[] { flushQuery });

    SIBUuid12 streamID = flushQuery.getGuaranteedStreamUUID();

    try
    {
      //synchronize to give a consistent view of the flushed state
      synchronized(this)
      {
        SIBUuid8 requestor = flushQuery.getGuaranteedSourceMessagingEngineUUID();
        if(isFlushed(streamID))
        {
          // It's flushed.  Send a message saying as much.
          downControl.sendFlushedMessage(requestor, streamID);
        }
        else
        {
          // Not flushed.  Send a message saying as much.
          downControl.sendNotFlushedMessage(requestor, streamID, flushQuery.getRequestID());
        }
      }
    }
    catch (SIResourceException e)
    {
      // FFDC
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.gd.SourceStreamManager.processFlushQuery",
        "1:1333:1.102",
        this);

      SibTr.exception(tc, e);

      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "processFlushQuery", e);

      throw e;
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "processFlushQuery");

  }
47832
public void waitToFinish(PipelineResult result) {
    pipelinesToCancel.add(result);
    if (!options.as(ExampleOptions.class).getKeepJobsRunning()) {
      addShutdownHook(pipelinesToCancel);
    }
    try {
      result.waitUntilFinish();
    } catch (UnsupportedOperationException e) {
      // Do nothing if the given PipelineResult doesn't support waitUntilFinish(),
      // such as EvaluationResults returned by DirectRunner.
      tearDown();
      printPendingMessages();
    } catch (Exception e) {
      throw new RuntimeException("Failed to wait the pipeline until finish: " + result);
    }
  }
7863
@AfterBatch
  public void afterBatch()
  {
    long requestSequence = _requestSequence;
    
    if (! _isFsync) {
      return;
    }
    
    _isFsync = false;
    
    try {
      _storeFsync.fsync();
      /*
      if (_tailSequence < requestSequence) {
        _storeFsync.fsync();
      }
      */
    } catch (Throwable e) {
      e.printStackTrace();
    } finally {
      _tailSequence = requestSequence;
    }
  }
134178
public static <T> CompletableFuture<T> completedFutureAsync(T result, Executor executor) {
    CompletableFuture<T> future = new CompletableFuture<>();
    executor.execute(() -> future.complete(result));
    return future;
  }
216669
public void requestCompleted (T result)
    {
        if (_chain != null) {
            _chain.requestCompleted(result);

        } else if (_invChain instanceof InvocationService.ResultListener) {
            ((InvocationService.ResultListener)_invChain).requestProcessed(result);

        } else if (_invChain instanceof InvocationService.ConfirmListener) {
            ((InvocationService.ConfirmListener)_invChain).requestProcessed();
        }
    }
903555
protected CompletableFuture<ExecutionResult> onFailureAsync(ExecutionResult result, Scheduler scheduler,
      FailsafeFuture<Object> future) {
    return CompletableFuture.completedFuture(onFailure(result));
  }
870257
private void fsyncImpl(Result<Boolean> result, FsyncType fsyncType)
  {
    try {
      flushData();
      
      ArrayList<SegmentFsyncCallback> fsyncListeners
        = new ArrayList<>(_fsyncListeners);
      _fsyncListeners.clear();
      
      Result<Boolean> resultNext = result.then((v,r)->
        afterDataFsync(r, _position, fsyncType, fsyncListeners));
        
      if (_isDirty || ! fsyncType.isSchedule()) {
        _isDirty = false;
      
        try (OutStore sOut = _readWrite.openWrite(_segment.extent())) {
          if (fsyncType.isSchedule()) {
            sOut.fsyncSchedule(resultNext);
          }
          else {
            sOut.fsync(resultNext);
          }
        }
      }
      else {
        resultNext.ok(true);
      }
    } catch (Throwable e) {
      e.printStackTrace();
      result.fail(e);
    }
  }
1253923
protected boolean setResult(Object result) {
        for (; ; ) {
            Object currentState = this.state;

            if (isDoneState(currentState)) {
                return false;
            }

            if (STATE.compareAndSet(this, currentState, result)) {
                done();
                notifyThreadsWaitingOnGet();
                runAsynchronous((ExecutionCallbackNode) currentState, result);
                break;
            }
        }
        return true;
    }