import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
@Override
@SuppressForbidden(reason = "debug")
protected void stopInternal() {
    Releasables.close(serverOpenChannels, () -> {
        final List<Tuple<String, Future<?>>> serverBootstrapCloseFutures = new ArrayList<>(serverBootstraps.size());
        for (final Map.Entry<String, ServerBootstrap> entry : serverBootstraps.entrySet()) {
            serverBootstrapCloseFutures.add(
                Tuple.tuple(entry.getKey(), entry.getValue().config().group().shutdownGracefully(0, 5, TimeUnit.SECONDS)));
        }
        for (final Tuple<String, Future<?>> future : serverBootstrapCloseFutures) {
            future.v2().awaitUninterruptibly();
            if (!future.v2().isSuccess()) {
                logger.debug(
                    (Supplier<?>) () -> new ParameterizedMessage(
                        "Error closing server bootstrap for profile [{}]", future.v1()), future.v2().cause());
            }
        }
        serverBootstraps.clear();

        if (bootstrap != null) {
            bootstrap.config().group().shutdownGracefully(0, 5, TimeUnit.SECONDS).awaitUninterruptibly();
            bootstrap = null;
        }
    });
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
/**
 * Writes all operations in the given iterable to the given output stream including the size of the array
 * use {@link #readOperations(StreamInput)} to read it back.
 */
public static void writeOperations(StreamOutput outStream, List<Operation> toWrite) throws IOException {
    final ReleasableBytesStreamOutput out = new ReleasableBytesStreamOutput(BigArrays.NON_RECYCLING_INSTANCE);
    try {
        outStream.writeInt(toWrite.size());
        final BufferedChecksumStreamOutput checksumStreamOutput = new BufferedChecksumStreamOutput(out);
        for (Operation op : toWrite) {
            out.reset();
            final long start = out.position();
            out.skip(Integer.BYTES);
            writeOperationNoSize(checksumStreamOutput, op);
            long end = out.position();
            int operationSize = (int) (out.position() - Integer.BYTES - start);
            out.seek(start);
            out.writeInt(operationSize);
            out.seek(end);
            ReleasablePagedBytesReference bytes = out.bytes();
            bytes.writeTo(outStream);
        }
    } finally {
        Releasables.close(out.bytes());
    }

}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
public void setReallocation(final ClusterService clusterService, final RoutingService routingService) {
    this.routingService = routingService;
    clusterService.addStateApplier(event -> {
        boolean cleanCache = false;
        DiscoveryNode localNode = event.state().nodes().getLocalNode();
        if (localNode != null) {
            if (localNode.isMasterNode() && event.localNodeMaster() == false) {
                cleanCache = true;
            }
        } else {
            cleanCache = true;
        }
        if (cleanCache) {
            Releasables.close(asyncFetchStarted.values());
            asyncFetchStarted.clear();
            Releasables.close(asyncFetchStore.values());
            asyncFetchStore.clear();
        }
    });
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
protected <A extends InternalAggregation, C extends Aggregator> A search(IndexSearcher searcher,
                                                                         Query query,
                                                                         AggregationBuilder builder,
                                                                         MappedFieldType... fieldTypes) throws IOException {
    C a = createAggregator(builder, searcher, fieldTypes);
    try {
        a.preCollection();
        searcher.search(query, a);
        a.postCollection();
        @SuppressWarnings("unchecked")
        A internalAgg = (A) a.buildAggregation(0L);
        return internalAgg;
    } finally {
        Releasables.close(releasables);
        releasables.clear();
    }
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
private static Snapshot createSnapshot(TranslogReader... translogs) {
    Snapshot[] snapshots = new Snapshot[translogs.length];
    boolean success = false;
    try {
        for (int i = 0; i < translogs.length; i++) {
            snapshots[i] = translogs[i].newSnapshot();
        }

        Snapshot snapshot = new MultiSnapshot(snapshots);
        success = true;
        return snapshot;
    } finally {
        if (success == false) {
            Releasables.close(snapshots);
        }
    }
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
static void stop() {
    try {
        Releasables.close(INSTANCE.node);
    } finally {
        INSTANCE.keepAliveLatch.countDown();
    }
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
void resetClient() throws IOException {
    if (closed.get() == false) {
        Releasables.close(nodeClient, transportClient);
        nodeClient = null;
        transportClient = null;
    }
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
public void merge(long thisBucket, HyperLogLogPlusPlus other, long otherBucket) {
    if (p != other.p) {
        throw new IllegalArgumentException();
    }
    ensureCapacity(thisBucket + 1);
    if (other.algorithm.get(otherBucket) == LINEAR_COUNTING) {
        final IntArray values = other.hashSet.values(otherBucket);
        try {
            for (long i = 0; i < values.size(); ++i) {
                final int encoded = values.get(i);
                if (algorithm.get(thisBucket) == LINEAR_COUNTING) {
                    collectLcEncoded(thisBucket, encoded);
                } else {
                    collectHllEncoded(thisBucket, encoded);
                }
            }
        } finally {
            Releasables.close(values);
        }
    } else {
        if (algorithm.get(thisBucket) != HYPERLOGLOG) {
            upgradeToHll(thisBucket);
        }
        final long thisStart = thisBucket << p;
        final long otherStart = otherBucket << p;
        for (int i = 0; i < m; ++i) {
            runLens.set(thisStart + i, (byte) Math.max(runLens.get(thisStart + i), other.runLens.get(otherStart + i)));
        }
    }
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
void upgradeToHll(long bucket) {
    ensureCapacity(bucket + 1);
    final IntArray values = hashSet.values(bucket);
    try {
        runLens.fill(bucket << p, (bucket << p) + m, (byte) 0);
        for (long i = 0; i < values.size(); ++i) {
            final int encoded = values.get(i);
            collectHllEncoded(bucket, encoded);
        }
        algorithm.set(bucket);
    } finally {
        Releasables.close(values);
    }
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
@Override
protected final void doClose() {
    if (recycler != null) {
        Releasables.close(cache);
        cache = null;
    }
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
@Override
public void sendResponse(TransportResponse response, TransportResponseOptions options) throws IOException {
    if (transport.compress) {
        options = TransportResponseOptions.builder(options).withCompress(transport.compress).build();
    }

    byte status = 0;
    status = TransportStatus.setResponse(status);

    ReleasableBytesStreamOutput bStream = new ReleasableBytesStreamOutput(transport.bigArrays);
    boolean addedReleaseListener = false;
    try {
        bStream.skip(NettyHeader.HEADER_SIZE);
        StreamOutput stream = bStream;
        if (options.compress()) {
            status = TransportStatus.setCompress(status);
            stream = CompressorFactory.defaultCompressor().streamOutput(stream);
        }
        stream.setVersion(version);
        response.writeTo(stream);
        stream.close();

        ReleasablePagedBytesReference bytes = bStream.bytes();
        ChannelBuffer buffer = bytes.toChannelBuffer();
        NettyHeader.writeHeader(buffer, requestId, status, version);
        ChannelFuture future = channel.write(buffer);
        ReleaseChannelFutureListener listener = new ReleaseChannelFutureListener(bytes);
        future.addListener(listener);
        addedReleaseListener = true;
        transportServiceAdapter.onResponseSent(requestId, action, response, options);
    } finally {
        if (!addedReleaseListener) {
            Releasables.close(bStream.bytes());
        }
    }
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
@Override
public void close() {
    List<Releasable> closeables = new ArrayList<>();
    for (OrdinalMapAndAtomicFieldData fds : atomicFD.values()) {
        closeables.addAll(Arrays.asList(fds.fieldData));
    }
    Releasables.close(closeables);
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
@Override
public void close() {
    try {
        Releasables.close(cachedTermOrds, termDocFreqs, termsTotalFreqs);
    } finally {
        cachedTermOrds = null;
        termDocFreqs = null;
        termsTotalFreqs = null;
    }
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
private <T extends BigArray> T validate(T array) {
    boolean success = false;
    try {
        adjustBreaker(array.ramBytesUsed());
        success = true;
    } finally {
        if (!success) {
            Releasables.closeWhileHandlingException(array);
        }
    }
    return array;
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
@Override
public void onResponse(Releasable releasable) {
    try {
        ReplicaResult replicaResult = shardOperationOnReplica(request, replica);
        releasable.close(); // release shard operation lock before responding to caller
        final TransportReplicationAction.ReplicaResponse response =
            new ReplicaResponse(replica.routingEntry().allocationId().getId(), replica.getLocalCheckpoint());
        replicaResult.respond(new ResponseListener(response));
    } catch (Exception e) {
        Releasables.closeWhileHandlingException(releasable); // release shard operation lock before responding to caller
        AsyncReplicaAction.this.onFailure(e);
    }
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
/**
 * Return a tuple representing the sequence ID for the given {@code Get}
 * operation. The first value in the tuple is the sequence number, the
 * second is the primary term.
 */
private Tuple<Long, Long> getSequenceID(Engine engine, Engine.Get get) throws EngineException {
    final Searcher searcher = engine.acquireSearcher("get");
    try {
        long seqNum = Versions.loadSeqNo(searcher.reader(), get.uid());
        long primaryTerm = Versions.loadPrimaryTerm(searcher.reader(), get.uid());
        return new Tuple(seqNum, primaryTerm);
    } catch (Exception e) {
        Releasables.closeWhileHandlingException(searcher);
        throw new EngineException(shardId, "unable to retrieve sequence id", e);
    } finally {
        searcher.close();
    }
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
public void testReleaseOnce() {
    AtomicInteger count = new AtomicInteger(0);
    Releasable releasable = Releasables.releaseOnce(count::incrementAndGet, count::incrementAndGet);
    assertEquals(0, count.get());
    releasable.close();
    assertEquals(2, count.get());
    releasable.close();
    assertEquals(2, count.get());
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
/**
 * performs the recovery from the local engine to the target
 */
public RecoveryResponse recoverToTarget() {
    final SnapshotIndexCommit phase1Snapshot;
    phase1Snapshot = shard.snapshotIndex(false);
    
    try {
        recoverLuceneFiles(phase1Snapshot);
    } catch (Throwable e) {
        logger.error("errors while recovery to target", e);
        throw new RecoveryEngineException(shard.shardId(), 1, "phase1 failed", e);
    } finally {
        Releasables.closeWhileHandlingException(phase1Snapshot);
    }
    return response;
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
public static void stop() {
    try {
        Releasables.close(INSTANCE.node);
    } finally {
        INSTANCE.keepAliveLatch.countDown();
    }
}
 

import org.elasticsearch.common.lease.Releasables; //导入依赖的package包/类
public void merge(long thisBucket, HyperLogLogPlusPlus other, long otherBucket) {
    Preconditions.checkArgument(p == other.p);
    ensureCapacity(thisBucket + 1);
    if (other.algorithm.get(otherBucket) == LINEAR_COUNTING) {
        final IntArray values = other.hashSet.values(otherBucket);
        try {
            for (long i = 0; i < values.size(); ++i) {
                final int encoded = values.get(i);
                if (algorithm.get(thisBucket) == LINEAR_COUNTING) {
                    collectLcEncoded(thisBucket, encoded);
                } else {
                    collectHllEncoded(thisBucket, encoded);
                }
            }
        } finally {
            Releasables.close(values);
        }
    } else {
        if (algorithm.get(thisBucket) != HYPERLOGLOG) {
            upgradeToHll(thisBucket);
        }
        final long thisStart = thisBucket << p;
        final long otherStart = otherBucket << p;
        for (int i = 0; i < m; ++i) {
            runLens.set(thisStart + i, (byte) Math.max(runLens.get(thisStart + i), other.runLens.get(otherStart + i)));
        }
    }
}