Completes any fsyncs in a batch.

Because the fsync is slow, this method will typically block while
new requests fill the queue, making the next batch efficient.
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;
    }
  }
1201790
private void invokeOnComplete() {
        for (; ; ) {
            if (pooledBuffers != null) {
                for (PooledByteBuffer buffer : pooledBuffers) {
                    buffer.close();
                }
                pooledBuffers = null;
            }
            IoCallback callback = this.callback;
            this.buffer = null;
            this.fileChannel = null;
            this.callback = null;
            inCallback = true;
            try {
                callback.onComplete(exchange, this);
            } finally {
                inCallback = false;
            }

            StreamSinkChannel channel = this.channel;
            if (this.buffer != null) {
                long t = Buffers.remaining(buffer);
                final long total = t;
                long written = 0;

                try {
                    do {
                        long res = channel.write(buffer);
                        written += res;
                        if (res == 0) {
                            if(writeListener == null) {
                                initWriteListener();
                            }
                            channel.getWriteSetter().set(writeListener);
                            channel.resumeWrites();
                            return;
                        }
                    } while (written < total);
                    //we loop and invoke onComplete again
                } catch (IOException e) {
                    invokeOnException(callback, e);
                }
            } else if (this.fileChannel != null) {
                if(transferTask == null) {
                    transferTask = new TransferTask();
                }
                if (!transferTask.run(false)) {
                    return;
                }
            } else {
                return;
            }

        }
    }
874922
final void flushComplete() throws IOException {
        synchronized (lock) {
            try {
                bufferFull = false;
                int remaining = header.getRemainingInBuffer();
                boolean finalFrame = finalFrameQueued;
                boolean channelClosed = finalFrame && remaining == 0 && !header.isAnotherFrameRequired();
                if (remaining > 0) {
                    body.getBuffer().limit(body.getBuffer().limit() + remaining);
                    if (finalFrame) {
                        //we clear the final frame flag, as it could not actually be written out
                        //note that we don't attempt to requeue, as whatever stopped it from being written will likely still
                        //be an issue
                        this.finalFrameQueued = false;
                    }
                } else if (header.isAnotherFrameRequired()) {
                    this.finalFrameQueued = false;
                    if (body != null) {
                        body.close();
                        body = null;
                        state &= ~STATE_PRE_WRITE_CALLED;
                    }
                } else if (body != null) {
                    body.close();
                    body = null;
                    state &= ~STATE_PRE_WRITE_CALLED;
                }
                if (channelClosed) {
                    fullyFlushed = true;
                    if (body != null) {
                        body.close();
                        body = null;
                        state &= ~STATE_PRE_WRITE_CALLED;
                    }
                } else if (body != null) {
                    // We still have a body, but since we just flushed, we transfer it to the write buffer.
                    // This works as long as you call write() again

                    //TODO: this code may not work if the channel has frame level compression and flow control
                    //we don't have an implementation that needs this yet so it is ok for now

                    body.getBuffer().compact();
                    writeBuffer = body;
                    body = null;
                    state &= ~STATE_PRE_WRITE_CALLED;
                }

                if (header.getByteBuffer() != null) {
                    header.getByteBuffer().close();
                }
                header = null;

                readyForFlush = false;
                if (isWriteResumed() && !channelClosed) {
                    wakeupWrites();
                } else if (isWriteResumed()) {
                    //we need to execute the write listener one last time
                    //we need to dispatch it back to the IO thread, so we don't invoke it recursivly
                    ChannelListeners.invokeChannelListener(getIoThread(), (S) this, getWriteListener());
                }

                final ChannelListener<? super S> closeListener = this.closeSetter.get();
                if (channelClosed && closeListener != null) {
                    ChannelListeners.invokeChannelListener(getIoThread(), (S) AbstractFramedStreamSinkChannel.this, closeListener);
                }
                handleFlushComplete(channelClosed);
            } finally {
                wakeupWaiters();
            }
        }
    }
482235
public void processSync(T message) {
        // If this batcher has been shutdown, do not accept any more messages
        if (isShutDown) {
            return;
        }

        try {
            queueSizeTracer.record(queue.size());
        } catch (Throwable ignored) {
        }

        try {
            Stopwatch s = batchSyncPutTracer.start();
            queue.put(message);
            s.stop();
        } catch (InterruptedException e) {
            return;
        }
        numberAdded.incrementAndGet();
    }
706134
public void complete() {
        long seqNo = this.operation.getSequenceNumber();
        Preconditions.checkState(seqNo >= 0, "About to complete a CompletableOperation that has no sequence number.");

        this.done = true;
        if (this.successHandler != null) {
            Callbacks.invokeSafely(this.successHandler, null, cex -> log.error("Success Callback invocation failure.", cex));
        }
    }
547676
def notify_batches_finished(self, statuses):
        
        with self._wait_condition:
            self._statuses = statuses
            self._wait_condition.notify()
526661
private void afterBatch(BatchBackend backend) {
		IndexedTypeSet targetedTypes = searchFactoryImplementor.getIndexedTypesPolymorphic( rootIndexedTypes );
		if ( this.optimizeAtEnd ) {
			backend.optimize( targetedTypes );
		}
		backend.flush( targetedTypes );
	}
672214
public static void afterCompletion(TransactionImpl tran, int status, int syncLevel) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            Tr.entry(tc, "afterCompletion", new Object[] { tran, status, syncLevel });

        if (syncLevel >= 0) {
            final ArrayList syncs = _syncLevels.get(syncLevel);

            if (syncs != null) {
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                    Tr.debug(tc, "syncs.length=" + syncs.size());

                final int localID = (int) tran.getLocalTID();
                final byte[] globalID = tran.getTID();
                // d281425 - status maybe committed, unknown or rolledback
                final boolean committed = !(status == Status.STATUS_ROLLEDBACK); // @281425C
                final int numSyncs = syncs.size();

                for (int s = 0; s < numSyncs; s++) {
                    try {
                        ((SynchronizationCallback) syncs.get(s)).afterCompletion(localID, globalID, committed);
                    } catch (Throwable t) {
                        // No FFDC needed
                        Tr.error(tc, "WTRN0074_SYNCHRONIZATION_EXCEPTION", new Object[] { "afterCompletion", syncs.get(s), t });
                        // absorb and allow other afterCompletions to run.
                    }
                }
            } else if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                Tr.debug(tc, "syncs is null");
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            Tr.exit(tc, "afterCompletion");
    }
551256
public void fsync(long sequence, K key, Result<Boolean> result)
  {
    scheduleFsync(sequence, key, result);
    flush();
  }
335652
private synchronized void maybeRun() {
        if (inprogress==null && pending!=null) {
            base.submit(new Callable<Void>() {
                @Override
                public Void call() throws Exception {
                    synchronized (AtmostOneTaskExecutor.this) {
                        // everyone who submits after this should form a next batch
                        inprogress = pending;
                        pending = null;
                    }

                    try {
                        inprogress.set(task.call());
                    } catch (Throwable t) {
                        LOGGER.log(Level.WARNING, null, t);
                        inprogress.setException(t);
                    } finally {
                        synchronized (AtmostOneTaskExecutor.this) {
                            // if next one is pending, get that scheduled
                            inprogress = null;
                            maybeRun();
                        }
                    }
                    return null;
                }
            });
        }
    }
397679
def fetch_async(self, limit=None, **q_options):
    
    if limit is None:
      default_options = self._make_options(q_options)
      if default_options is not None and default_options.limit is not None:
        limit = default_options.limit
      else:
        limit = _MAX_LIMIT
    q_options['limit'] = limit
    q_options.setdefault('batch_size', limit)
    if self._needs_multi_query():
      return self.map_async(None, **q_options)
    # Optimization using direct batches.
    options = self._make_options(q_options)
    qry = self._fix_namespace()
    return qry._run_to_list([], options=options)
605738
private void completeResponses() {
    // Iterate through queued responses and complete as many as possible.
    ResponseCallback response = responseCallbacks.get(responseSequence + 1);
    while (response != null) {
      // If the response was completed, remove the response callback from the response queue,
      // increment the response sequence number, and check the next response.
      if (completeResponse(response.response, response.callback)) {
        responseCallbacks.remove(++responseSequence);
        response = responseCallbacks.get(responseSequence + 1);
      } else {
        break;
      }
    }

    // Once we've completed as many responses as possible, if no more operations are outstanding
    // and events remain in the event queue, complete the events.
    if (requestSequence == responseSequence) {
      EventCallback eventCallback = eventCallbacks.poll();
      while (eventCallback != null) {
        log.trace("Completing {}", eventCallback.request);
        eventCallback.run();
        eventIndex = eventCallback.request.eventIndex();
        eventCallback = eventCallbacks.poll();
      }
    }
  }
622458
def sync(self, expectedThreads=0):
        ''
        while len(self.unfinishedThreads) > expectedThreads:
            time.sleep(.3)
            self.checkForFinishedThreads()
1097593
public ISynchronizationPoint<IOException> finishAsync() {
		if (finishing != null) return finishing;
		finishing = new SynchronizationPoint<>();
		Task<Void,NoException> task = new Task.Cpu<Void,NoException>("Finishing zip compression", priority) {
			@Override
			public Void run() {
				AsyncWork<Integer, IOException> lastWrite = null;
				deflater.finish();
				if (!deflater.finished()) {
					do {
						byte[] writeBuf = new byte[8192];
						int nb = deflater.deflate(writeBuf, 0, writeBuf.length);
						if (nb <= 0) break;
						try { lastWrite = writeOps.write(ByteBuffer.wrap(writeBuf, 0, nb)); }
						catch (IOException e) {
							finishing.error(e);
							return null;
						}
					} while (!deflater.finished());
				}
				if (lastWrite == null) lastWrite = writeOps.getLastPendingOperation();
				if (lastWrite != null)
					lastWrite.listenInline(finishing);
				else
					finishing.unblock();
				return null;
			}
		};
		task.start();
		return operation(finishing);
	}
791519
def _fetchBlock(self):
        
        # make sure that another block request is not standing
        if self._blockRequestInProgress :
           # need to wait here before returning... (TODO)
           return

        # make sure another block request has not completed meanwhile
        if self._standbyBlock is not None: 
           return

        self._blockRequestInProgress = True
        fetchReq = TFetchResultsReq(operationHandle=self.operationHandle,
                                    orientation=TFetchOrientation.FETCH_NEXT,
                                    maxRows=self.arraysize)
        self._standbyBlock = self._fetch([],fetchReq)
        self._blockRequestInProgress = False
        return
296353
private void completePendingFlush()
  {
    int size = _pendingFlushEntries.size();
    
    if (size == 0) {
      return;
    }

    // ArrayList<TableService.PageFlush> pageList = new ArrayList<>();
    
    for (int i = 0; i < size; i++) {
      PendingEntry entry = _pendingFlushEntries.get(i);
      
      entry.afterFlush();
      
      _pendingFsyncEntries.add(entry);
    }
    
    // _tableService.afterDataFlush(pageList);
    
    _pendingFlushEntries.clear();
  }
1262471
@Override
  public void flush(boolean forceSync)
      throws IOException {
    if (checksumOut != null) {
      checksumOut.flush();
      if (forceSync && (cout instanceof FileOutputStream)) {
        ((FileOutputStream) cout).getChannel().force(true);
      }
    }
    if (blockDataWriter != null) {
      blockDataWriter.flush();
      if (forceSync) {
        blockDataWriter.force(true);
      }
    }
  }