Differs from the above because it can be used on any file, rather than only
managed commit log segments (and thus cannot call waitForFinalSync).

Used to archive files present in the commit log directory at startup (CASSANDRA-6904)
551791
public void maybeArchive(final String path, final String name)
    {
        if (Strings.isNullOrEmpty(archiveCommand))
            return;

        archivePending.put(name, executor.submit(new WrappedRunnable()
        {
            protected void runMayThrow() throws IOException
            {
                String command = archiveCommand.replace("%name", name);
                command = command.replace("%path", path);
                exec(command);
            }
        }));
    }
1086063
@Override
    public void storeUpdate(INDArray array, int iterationNumber, int epochNumber) {
        /*
            Here we want to do 4 things:
            1) update accumulated values
            2) invoke extractor, that'll (optionally) pull all updates above threshold
            3) ???
            4) PROFIT!
         */

        try {
            // commit should happen in each individual thread
            Nd4j.getExecutioner().commit();
            firstOne.compareAndSet(-1L, Thread.currentThread().getId());

            // TODO: since we know number of elements in advance, we don't really need CopyOnWrite list here.
            candidates.add(array);
            barrier.await();

            if (firstOne.get() == Thread.currentThread().getId()) {
                // if accum is null, let's just create it here
                if (storage == null) {
                    // we don't want state array to be attached to any workspace
                    shape = array.shape();
                    ordering = array.ordering();

                    try (MemoryWorkspace workspace = Nd4j.getMemoryManager().scopeOutOfWorkspaces()) {
                        // TODO: if p2p isn't supported, this should become HOST-only allocation
                        storage = Nd4j.create(shape, ordering);
                    }
                }


                // accumulate our values, a
                //storage.addi(array);
                Nd4j.accumulate(storage, candidates);

                // we ensure storage was updated successfully
                Nd4j.getExecutioner().commit();

                // if there's something to send - send it. Skip otherwise!!!
                if (handler.broadcastUpdates(storage, iterationNumber, epochNumber)) {
                    ownCounter.getAndIncrement();
                }

                // reset "first one" :)
                firstOne.set(-1L);
                candidates.clear();
            }

            barrier.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        } catch (BrokenBarrierException e) {
            throw new RuntimeException(e);
        }
    }
914989
File getSyncLogTemporaryFile(long segmentTxId, long endTxId, long stamp) {
    String name = NNStorage.getFinalizedEditsFileName(segmentTxId, endTxId) +
        ".tmp=" + stamp; 
    return new File(sd.getCurrentDir(), name);
  }
241785
static SourceLocation createSrcLoc(String filePath, Token first, Token... rest) {
    int beginLine = first.beginLine;
    int beginColumn = first.beginColumn;
    int endLine = first.endLine;
    int endColumn = first.endColumn;

    for (Token next : rest) {
      checkArgument(startsLaterThan(next, beginLine, beginColumn));
      checkArgument(endsLaterThan(next, endLine, endColumn));
      endLine = next.endLine;
      endColumn = next.endColumn;
    }
    // this special case happens for completely empty files.
    if (beginLine == 0 && endLine == 0 && beginColumn == 0 && endColumn == 0) {
      return new SourceLocation(filePath);
    }
    return new SourceLocation(filePath, beginLine, beginColumn, endLine, endColumn);
  }
1018225
@Override
    public CompletableFuture<Void> setAsync(int expireSeconds, String key, V value) {
        return (CompletableFuture) setAsync(key, value).thenCompose(v -> setExpireSecondsAsync(key, expireSeconds));
    }
585166
public synchronized void recordCommit(int commitLength) {
        Preconditions.checkArgument(commitLength >= 0, "commitLength must be a non-negative number.");

        // Update counters.
        this.commitCount++;
        this.accumulatedLength += commitLength;

        int minCount = this.config.getCheckpointMinCommitCount();
        int countThreshold = this.config.getCheckpointCommitCountThreshold();
        long lengthThreshold = this.config.getCheckpointTotalCommitLengthThreshold();
        if (this.commitCount >= minCount && (this.commitCount >= countThreshold || this.accumulatedLength >= lengthThreshold)) {
            // Reset counters.
            this.commitCount = 0;
            this.accumulatedLength = 0;

            // Invoke callback.
            this.executor.execute(this.createCheckpointCallback);
        }
    }
162546
File syncLog(long stamp, final long startTxId, final URL url, String name, File tmpFile)
      throws IOException {
    final File[] localPaths = new File[] { tmpFile };

    // TODO add security if needed.
    LOG.info("Synchronizing log " + name + " from " + url);
    boolean success = false;
    try {
      TransferFsImage.doGetUrl(
          url,
          ImageSet.convertFilesToStreams(localPaths, journalStorage,
              url.toString()), journalStorage, true);
      assert tmpFile.exists();
      success = true;
    } finally {
      if (!success) {
        if (!tmpFile.delete()) {
          LOG.warn("Failed to delete temporary file " + tmpFile);
        }
      }
    }
    return tmpFile;
  }
1122727
@Override
	public Object visitPrepareCommand(TxInvocationContext ctx, PrepareCommand command) throws Throwable {
		if (ctx.isOriginLocal()) {
			// We can't wait to commit phase to remove the entry locally (invalidations are processed in 1pc
			// on remote nodes, so only local case matters here). The problem is that while the entry is locked
			// reads still can take place and we can read outdated collection after reading updated entity
			// owning this collection from DB; when this happens, the version lock on entity cannot protect
			// us against concurrent modification of the collection. Therefore, we need to remove the entry
			// here (even without lock!) and let possible update happen in commit phase.
			for (WriteCommand wc : command.getModifications()) {
				for (Object key : wc.getAffectedKeys()) {
					dataContainer.remove(key);
				}
			}
		}
		else {
			for (WriteCommand wc : command.getModifications()) {
            Collection<?> keys = wc.getAffectedKeys();
				if (log.isTraceEnabled()) {
					log.tracef("Invalidating keys %s with lock owner %s", keys, ctx.getLockOwner());
				}
				for (Object key : keys ) {
					putFromLoadValidator.beginInvalidatingKey(ctx.getLockOwner(), key);
				}
			}
		}
		return invokeNext(ctx, command);
	}
1349375
@SuppressWarnings("ConstantConditions")
    @Override
    public void storeSnapshot(ExtendedSnapshotWriter snapshotWriter) throws StorageException {
        checkInitialized();

        checkArgument(snapshotWriter instanceof SnapshotFileWriter, "unknown snapshot request type:%s", snapshotWriter.getClass().getSimpleName());

        final SnapshotFileWriter snapshotFileWriter = (SnapshotFileWriter) snapshotWriter;
        checkArgument(snapshotFileWriter.snapshotStarted(), "snapshot was never started");

        try {
            // first close the output stream (the snapshotFileWriter may already have done this, but nbd)
            snapshotFileWriter.getSnapshotOutputStream().close();

            // the time we'll assign to this snapshot (both in the filename and the db)
            long snapshotTimestamp = System.currentTimeMillis();

            // setup the name/path of the final snapshot file
            String snapshotFilename = String.format("%d-%s.snap", snapshotTimestamp, UUID.randomUUID().toString());
            File snapshotFile = new File(snapshotsDirectory, snapshotFilename);

            // move the temporary file to the final location
            File tempSnapshotFile = snapshotFileWriter.getSnapshotFile();
            Files.move(tempSnapshotFile.toPath(), snapshotFile.toPath(), StandardCopyOption.ATOMIC_MOVE);

            // create the metadata entry for this snapshot
            // note that this filename does _not_ include the path, only the filename itself
            final SnapshotMetadata metadata = new SnapshotMetadata(
                            snapshotFilename,
                            snapshotTimestamp,
                            snapshotFileWriter.getTerm(),
                            snapshotFileWriter.getIndex());

            // actually add this snapshot to the db
            dbi.withHandle(new HandleCallback<Void>() {
                @Override
                public Void withHandle(Handle handle) throws Exception {
                    SnapshotsDAO dao = handle.attach(SnapshotsDAO.class);
                    dao.addSnapshot(metadata);
                    return null;
                }
            });
        } catch (IOException e) {
            throw new StorageException("fail finalize snapshot file req:" + snapshotFileWriter, e);
        } catch (CallbackFailedException e) {
            throw new StorageException("fail store snapshot metadata req:" + snapshotFileWriter, e.getCause());
        } catch (Exception e) {
            throw new StorageException("fail store snapshot req:" + snapshotFileWriter, e);
        }
    }
1048462
function getCurrentFileSize(counter, timeout, cb) {
  var fd = fs.openSync(__filename, 'r');

  console.log("Trying to aquire lock for the %s time", counter);

  fs.flock(fd, 'exnb', function(err) {
    if (err) {
      return console.log("Couldn't lock file", counter);
    }

    console.log('Aquired lock', counter);

    // unlock after `timeout`
    setTimeout(function() {
      fs.flock(fd, 'un', function(err) {
        if (err) {
          return console.log("Couldn't unlock file", counter);
        }

        if (cb) { cb(); }
      });
    }, timeout);
  });
}
654486
private CompletableFuture<Void> takeSnapshots(boolean rescheduleAfterCompletion, boolean force) {
    // If compaction is already in progress, return the existing future and reschedule if this is a scheduled compaction.
    if (compactFuture != null) {
      if (rescheduleAfterCompletion) {
        compactFuture.whenComplete((r, e) -> scheduleSnapshots());
      }
      return compactFuture;
    }

    long lastApplied = raft.getLastApplied();

    // Only take snapshots if segments can be removed from the log below the lastApplied index.
    if (raft.getLog().isCompactable(lastApplied) && raft.getLog().getCompactableIndex(lastApplied) > lastCompacted) {

      // Determine whether the node is running out of disk space.
      boolean runningOutOfDiskSpace = isRunningOutOfDiskSpace();

      // Determine whether the node is running out of memory.
      boolean runningOutOfMemory = isRunningOutOfMemory();

      // If compaction is not already being forced...
      if (!force
          // And the node isn't running out of memory (we need to free up memory if it is)...
          && !runningOutOfMemory
          // And dynamic compaction is enabled (we need to compact immediately if it's disabled)...
          && raft.getStorage().dynamicCompaction()
          // And the node isn't running out of disk space (we need to compact immediately if it is)...
          && !runningOutOfDiskSpace
          // And the server is under high load (we can skip compaction at this point)...
          && raft.getLoadMonitor().isUnderHighLoad()) {
        // We can skip taking a snapshot for now.
        logger.debug("Skipping compaction due to high load");
        if (rescheduleAfterCompletion) {
          scheduleSnapshots();
        }
        return CompletableFuture.completedFuture(null);
      }

      logger.debug("Snapshotting services");

      // Update the index at which the log was last compacted.
      this.lastCompacted = lastApplied;

      // We need to ensure that callbacks added to the compaction future are completed in the order in which they
      // were added in order to preserve the order of retries when appending to the log.
      compactFuture = new OrderedFuture<>();

      // Wait for snapshots in all state machines to be completed before compacting the log at the last applied index.
      takeSnapshots().whenComplete((snapshot, error) -> {
        if (error == null) {
          scheduleCompletion(snapshot.persist());
        }
      });

      // Reschedule snapshots after completion if necessary.
      if (rescheduleAfterCompletion) {
        compactFuture.whenComplete((r, e) -> scheduleSnapshots());
      }
      return compactFuture;
    }
    // Otherwise, if the log can't be compacted anyways, just reschedule snapshots.
    else {
      if (rescheduleAfterCompletion) {
        scheduleSnapshots();
      }
      return CompletableFuture.completedFuture(null);
    }
  }
1259300
function (commit, until) {
            var checker = this;

            until = !isNaN(until) ? until : (new Date()).getTime() + (this.getOption('timeout'));

            if ((new Date()).getTime() > until) {
                return this.handleError('Timeout');
            }

            return this.getBuilds()
                .then(function (builds) {
                    if (undefined !== builds.message) {
                        return checker.handleError(builds.message);
                    }

                    var finder = new BuildFinder(builds),
                        builds = finder.findByCommit(commit);

                    if (null === builds) {
                        return checker.handleError('Build not found');
                    }

                    return builds;
                }, this.handleError)

                .then(function (builds) {
                    var successfulBuilds = 0;

                    for (var i = 0; i < builds.length; i++) {
                        if (builds[i].isSuccess()) {
                            successfulBuilds++;
                        } else if (builds[i].isFailure() || !checker.getOption('retryOnRunning')) {
                            return checker.handleError('Invalid status for CircleCI build: "' + builds[i].getStatus() + '"')
                        }
                    }

                    if (successfulBuilds === builds.length) {
                        return true;
                    }

                    // Sleep for 10 seconds by default
                    var promise = sleep(checker.getOption('retryAfter'));

                    return promise.then(function () {
                        return checker.checkCommit(commit, until);
                    });
                }, this.handleError);
        }
1202333
public synchronized void startLogSegment(RequestInfo reqInfo, long txid)
      throws IOException {
    assert fjm != null;
    checkJournalStorageFormatted();
    checkRequest(reqInfo);
    
    if (curSegment != null) {
      LOG.warn("Client is requesting a new log segment " + txid + 
          " though we are already writing " + curSegment + ". " +
          "Aborting the current segment in order to begin the new one.");
      // The writer may have lost a connection to us and is now
      // re-connecting after the connection came back.
      // We should abort our own old segment.
      abortCurSegment();
    }

    // Paranoid sanity check: we should never overwrite a finalized log file.
    // Additionally, if it's in-progress, it should have at most 1 transaction.
    // This can happen if the writer crashes exactly at the start of a segment.
    EditLogFile existing = fjm.getLogFile(txid);
    if (existing != null) {
      if (!existing.isInProgress()) {
        throw new IllegalStateException("Already have a finalized segment " +
            existing + " beginning at " + txid);
      }
      
      // If it's in-progress, it should only contain one transaction,
      // because the "startLogSegment" transaction is written alone at the
      // start of each segment. 
      existing.validateLog();
      if (existing.getLastTxId() != existing.getFirstTxId()) {
        throw new IllegalStateException("The log file " +
            existing + " seems to contain valid transactions");
      }
    }
    
    long curLastWriterEpoch = lastWriterEpoch.get();
    if (curLastWriterEpoch != reqInfo.getEpoch()) {
      LOG.info("Updating lastWriterEpoch from " + curLastWriterEpoch +
          " to " + reqInfo.getEpoch() + " for client " +
          Server.getRemoteIp());
      lastWriterEpoch.set(reqInfo.getEpoch());
      metrics.lastWriterEpoch.set(reqInfo.getEpoch());
    }

    // The fact that we are starting a segment at this txid indicates
    // that any previous recovery for this same segment was aborted.
    // Otherwise, no writer would have started writing. So, we can
    // remove the record of the older segment here.
    purgePaxosDecision(txid);
    
    curSegment = fjm.startLogSegment(txid);
    curSegmentTxId = txid;
    nextTxId = txid;
    // the layout version has only been written
    // plus OP_INVALID
    currentSegmentWrittenBytes = 5L;
    
    // inform the syncer service that we might have some work to do
    if (journalNode != null) {
      journalNode.addSyncTask(this, curSegmentTxId);
    }
  }
301621
@Override
	public void externalizeCacheObjectAsync(final String fileNamePrefix, final Object object,
			final CreateCompletionCallback callback) {
		_aggregator.getExecutors().getFileCreateExecutor().submit(new Runnable() {
			File file = null;
			public void run() {
				try {
					file = File.createTempFile(fileNamePrefix, ".cache", _directory); //$NON-NLS-1$
					ObjectOutputStream os;
					os = new ObjectOutputStream(new FileOutputStream(file));
					try {
						os.writeObject(object);
					} finally {
						try { os.close(); } catch (Exception ignore) {}
					}
					if (callback != null) {
						callback.completed(file.getName(), null);
					}
				} catch (IOException e) {
					if (log.isLoggable(Level.WARNING))
						log.log(Level.WARNING, MessageFormat.format(
								Messages.CacheManagerImpl_4, new Object[]{file.getPath()}), e);
					if (callback != null) {
						callback.completed(file != null ? file.getName() : null, e);
					}
				}
			}
		});
	}
737623
@Override
    public synchronized long commit(long offset) {
        if (offset <= submittedOffset && offset > safeOffset) {
            int ggap = (int)Math.min(committedOffsets.length, offset-safeOffset);
            if (ggap == committedOffsets.length) {
                LOGGER.rateLimitedLog(LOG_SUPPRESSION_INTERVAL_SECONDS ,Level.WARN,
                        null, "CommitTracker moving topic commit point from %d to %d for topic "
                                + topic + " partition " + partition + " group:" + consumerGroup, safeOffset, (offset - committedOffsets.length + 1)
                        );
                safeOffset = offset - committedOffsets.length + 1;
                committedOffsets[idx(safeOffset)] = safeOffset;
            }
            committedOffsets[idx(offset)] = offset;
            while (ggap > 0 && committedOffsets[idx(safeOffset)]+1 == committedOffsets[idx(safeOffset+1)]) {
                ++safeOffset;
            }
            if (offerOffset >=0 && (offerOffset-safeOffset) < committedOffsets.length) {
                offerOffset = -1L;
                notify();
            }
        }

        if (offset == firstOffset) {
            firstOffsetCommitted = true;
        }
        return safeOffset;
    }
790099
void waitNewLeaderAckFromQuorum()
      throws TimeoutException, InterruptedException, IOException {
    LOG.debug("Waiting for synchronization to followers complete.");
    int completeCount = 0;
    Zxid lastZxid = persistence.getLatestZxid();
    while (completeCount < this.quorumMap.size()) {
      // Here we should use sync_timeout.
      MessageTuple tuple = filter.getExpectedMessage(MessageType.ACK, null,
                                                     getSyncTimeoutMs());
      ZabMessage.Ack ack = tuple.getMessage().getAck();
      String source =tuple.getServerId();
      Zxid zxid = MessageBuilder.fromProtoZxid(ack.getZxid());
      if (!this.quorumMap.containsKey(source)) {
        LOG.warn("Quorum set doesn't contain {}, a bug?", source);
        continue;
      }
      if (zxid.compareTo(lastZxid) != 0) {
        LOG.error("The follower {} is not correctly synchronized.", source);
        throw new RuntimeException("The synchronized follower's last zxid"
            + "doesn't match last zxid of current leader.");
      }
      PeerHandler ph = this.quorumMap.get(source);
      ph.setLastAckedZxid(zxid);
      completeCount++;
    }
  }
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();
        }
      }
    }
  }