To do offset storage we let the topology drain itself out. Then we commit.
13669
public void maybeDoOffset(){
    long seen = tuplesSeen;
    if (offsetCommitInterval > 0 && seen % offsetCommitInterval == 0 && offsetStorage != null && fp.supportsOffsetManagement()){
      doOffsetInternal();
    }
  }
235702
public synchronized void close() {
		if (storage.isOpen()) {
			if (useRedoThread && (redoThread != null)) {
				stopRedoThread(redoThread);
			}
			sync();
			writeMetaData(true);
			if (enableDirtyCheck) {
				log.info("dirtyCheck=" + dirtyCheck);
			}
		}
		storage.close();
		redoStore.close();
		clearReadCaches();
		clearWriteCaches();
		validState = false;
		// log.debug(this.hashCode() + "::close() done");
	}
259160
public void commit() {
    if (directory == null)
      return;

    try {
      if (reader != null)
        reader.close();

      // it turns out that IndexWriter.optimize actually slows
      // searches down, because it invalidates the cache. therefore
      // not calling it any more.
      // http://www.searchworkings.org/blog/-/blogs/uwe-says%3A-is-your-reader-atomic
      // iwriter.optimize();

      iwriter.commit();
      openSearchers();
    } catch (IOException e) {
      throw new DukeException(e);
    }
  }
350460
private CompletableFuture<Void> commitInitialEntries() {
    // The Raft protocol dictates that leaders cannot commit entries from previous terms until
    // at least one entry from their current term has been stored on a majority of servers. Thus,
    // we force entries to be appended up to the leader's no-op entry. The LeaderAppender will ensure
    // that the commitIndex is not increased until the no-op entry (appender.index()) is committed.
    CompletableFuture<Void> future = new CompletableFuture<>();
    appender.appendEntries(appender.index()).whenComplete((resultIndex, error) -> {
      context.checkThread();
      if (isOpen()) {
        if (error == null) {
          context.getStateMachine().apply(resultIndex);
          future.complete(null);
        } else {
          context.setLeader(0);
          context.transition(CopycatServer.State.FOLLOWER);
        }
      }
    });
    return future;
  }
992492
def disconnect_storage(network, storage):
    
    # does only remove from network.pypsa, not from network.pypsa_lopf
    # remove from pypsa (buses, storage_units, storage_units_t, lines)
    neighbor = storage.grid.graph.neighbors(storage)[0]
    if network.pypsa is not None:
        line = storage.grid.graph.line_from_nodes(storage, neighbor)
        network.pypsa.storage_units = network.pypsa.storage_units.loc[
                                      network.pypsa.storage_units.index.drop(
                                          repr(storage)), :]
        network.pypsa.storage_units_t.p_set.drop([repr(storage)], axis=1,
                                                 inplace=True)
        network.pypsa.storage_units_t.q_set.drop([repr(storage)], axis=1,
                                                 inplace=True)
        network.pypsa.buses = network.pypsa.buses.loc[
                              network.pypsa.buses.index.drop(
                                  '_'.join(['Bus', repr(storage)])), :]
        network.pypsa.lines = network.pypsa.lines.loc[
                              network.pypsa.lines.index.drop(
                                  repr(line)), :]
    # delete line
    neighbor = storage.grid.graph.neighbors(storage)[0]
    storage.grid.graph.remove_edge(storage, neighbor)
    # delete storage
    storage.grid.graph.remove_node(storage)
1245306
private void blockIfNecessary() {
        if (diskPermits == null || diskLimit == Long.MAX_VALUE)
            return;
        try {
            diskPermits.acquire();
        } catch (InterruptedException e) {
            // don't want to re-interrupt so it won't cause SDK stream to be
            // closed in case the thread is reused for a different request
            throw new AbortedException(e);
        }
    }
879825
function purgeDependencies(unit, error, onPurgedDependentJoint, onDone){
	db.takeConnectionFromPool(function(conn){
		var arrQueries = [];
		conn.addQuery(arrQueries, "BEGIN");
		collectQueriesToPurgeDependentJoints(conn, arrQueries, unit, error, onPurgedDependentJoint, function(){
			conn.addQuery(arrQueries, "COMMIT");
			async.series(arrQueries, function(){
				conn.release();
				if (onDone)
					onDone();
			})
		});
	});
}
1360989
final void rebalance() {
        // Let's start distribute the vbuckets across the servers
        configurationRwLock.writeLock().lock();
        try {
            Info.incrementConfigRevision();
            List<MemcachedServer> nodes = activeServers();
            for (int ii = 0; ii < numVBuckets; ++ii) {
                Collections.shuffle(nodes);
                vbInfo[ii].setOwner(nodes.get(0));
                if (nodes.size() < 2) {
                    continue;
                }
                List<MemcachedServer> replicas = nodes.subList(1, nodes.size());
                if (replicas.size() > numReplicas) {
                    replicas = replicas.subList(0, numReplicas);
                }
                vbInfo[ii].setReplicas(replicas);
            }
        } finally {
            Info.incrementConfigRevision();
            configurationRwLock.writeLock().unlock();
        }
    }
1082706
public function commit()
	{
		$this->sourceWidth = $this->frameWidth;
		$this->sourceHeight = $this->frameHeight;
		$this->offsetX = 0;
		$this->offsetY = 0;
		$this->scaleWidth = $this->frameWidth;
		$this->scaleHeight = $this->frameHeight;

		$this->getFocalPoint();
	}
968910
public void commit ()
    {
        // first clear our target
        clearTarget();

        // then post this event onto the queue (but only if we actually
        // accumulated some events)
        int size = _events.size();
        switch (size) {
        case 0: // nothing doing
            break;
        case 1: // no point in being compound
            _omgr.postEvent(_events.get(0));
            break;
        default: // now we're talking
            _transport = _events.get(0).getTransport();
            for (int ii = 1; ii < size; ii++) {
                _transport = _events.get(ii).getTransport().combine(_transport);
            }
            _omgr.postEvent(this);
            break;
        }
    }
1212039
void channelCloses(final SynchronizeFXWebsocketChannel synchronizeFXTomcatChannel) {
        synchronized (channels) {
            if (isCurrentlyShutingDown) {
                return;
            }
            // Maybe a bit inefficient but the alternative would be to pass cannelName and SynchronizeFxServer to the
            // channel just to be able to close itself.
            final Iterator<Entry<String, SynchronizeFXWebsocketChannel>> channelIterator =
                    channels.entrySet().iterator();
            while (channelIterator.hasNext()) {
                if (channelIterator.next().getValue().equals(synchronizeFXTomcatChannel)) {
                    channelIterator.remove();
                }
            }

            final Iterator<Entry<SynchronizeFxServer, SynchronizeFXWebsocketChannel>> serverIterator =
                    servers.entrySet().iterator();
            while (serverIterator.hasNext()) {
                if (serverIterator.next().getValue().equals(synchronizeFXTomcatChannel)) {
                    serverIterator.remove();
                }
            }

            final Iterator<Entry<Session, SynchronizeFXWebsocketChannel>> clientIterator =
                    clients.entrySet().iterator();
            while (clientIterator.hasNext()) {
                if (clientIterator.next().getValue().equals(synchronizeFXTomcatChannel)) {
                    clientIterator.remove();
                }
            }
        }
    }
1224465
def low_fileoff
      offset = @raw_data.size

      segments.each do |seg|
        seg.sections.each do |sect|
          next if sect.empty?
          next if sect.flag?(:S_ZEROFILL)
          next if sect.flag?(:S_THREAD_LOCAL_ZEROFILL)
          next unless sect.offset < offset

          offset = sect.offset
        end
      end

      offset
    end
7935
private void discardIdleConnections() {
        List<Connection<CL>> connections = Lists.newArrayList();
        availableConnections.drainTo(connections);
        activeCount.addAndGet(-connections.size());

        for (Connection<CL> connection : connections) {
            try {
                closedConnections.incrementAndGet();
                connection.close(); // This is usually an async operation
            }
            catch (Throwable t) {
                // TODO
            }
        }
    }
35684
public void wrap(final long offset, final long length)
    {
        if (offset == addressOffset && length == capacity)
        {
            return;
        }

        wrap(fileChannel, offset, length);
    }
434574
def _finalize_chunk(self, dd, offsets):
        # type: (Downloader, blobxfer.models.download.Descriptor,
        #        blobxfer.models.download.Offsets) -> None
        
        if dd.entity.is_encrypted:
            dd.mark_unchecked_chunk_decrypted(offsets.chunk_num)
        # integrity check data and write to disk (this is called
        # regardless of md5/hmac enablement for resume purposes)
        dd.perform_chunked_integrity_check()
        # remove from disk set and add bytes to counter
        with self._disk_operation_lock:
            self._disk_set.remove(
                blobxfer.operations.download.Downloader.
                create_unique_disk_operation_id(dd, offsets))
            self._download_bytes_sofar += offsets.num_bytes
508252
public void releaseAllChunkMemoryUsage()
  {
    if (chunks == null || chunks.size() == 0)
    {
      return;
    }

    for (int i = 0; i < chunks.size(); i++)
    {
      releaseCurrentMemoryUsage(i, chunks.get(i).computeNeededChunkMemory());
    }
  }
674605
def reset_offsets_if_needed(self, partitions):
        
        for tp in partitions:
            # TODO: If there are several offsets to reset, we could submit offset requests in parallel
            if self._subscriptions.is_assigned(tp) and self._subscriptions.is_offset_reset_needed(tp):
                self._reset_offset(tp)