java.nio.channels.FileLockInterruptionException#com.indeed.util.core.io.Closeables2源码实例Demo

@Override
public final void close() {
    synchronized (session) {
        if (intTermDocIterator != null) {
            Closeables2.closeQuietly(intTermDocIterator, ImhotepLocalSession.log);
            intTermDocIterator = null;
        }
        if (stringTermDocIterator != null) {
            Closeables2.closeQuietly(stringTermDocIterator, ImhotepLocalSession.log);
            stringTermDocIterator = null;
        }
        if (flamdexReader != null) {
            Closeables2.closeQuietly(flamdexReader, ImhotepLocalSession.log);
            flamdexReader = null;
        }
    }
}
 

@Override
public final void close() {
    synchronized (session) {
        if (docIdStream != null) {
            Closeables2.closeQuietly(docIdStream, ImhotepLocalSession.log);
        }
        if (intTermIterator != null) {
            Closeables2.closeQuietly(intTermIterator, ImhotepLocalSession.log);
            intTermIterator = null;
        }
        if (stringTermIterator != null) {
            Closeables2.closeQuietly(stringTermIterator, ImhotepLocalSession.log);
            stringTermIterator = null;
        }
        if (flamdexReader != null) {
            Closeables2.closeQuietly(flamdexReader, ImhotepLocalSession.log);
            flamdexReader = null;
        }
    }
}
 

@Override
public void close() {
    try {
        if (readLockRef == null) {
            Closeables2.closeAll(log, metricCache, wrapped);
        } else {
            Closeables2.closeAll(log, metricCache, wrapped, readLockRef);                
        }
    } finally {
        if (memory == null) {
            return;
        }
        if (memory.usedMemory() > 0) {
            log.error("CachedFlamdexReader is leaking! memory reserved after all memory has been freed: "+memory.usedMemory());
        }
        Closeables2.closeQuietly(memory, log);
    }
}
 

@Override
public void close() {
    synchronized (loadedMetrics) {
        if (!closed) {
            closed = true;
            try {
                Closeables2.closeAll(log, Collections2.transform(loadedMetrics.entrySet(), new Function<Map.Entry<String, IntValueLookup>, Closeable>() {
                    public Closeable apply(final Map.Entry<String, IntValueLookup> metric) {
                        return closeMetric.asCloseable(metric);
                    }
                }));
            } finally {
                loadedMetrics.clear();
            }
        }
    }
}
 

public RawFTGSIterator mergeFTGSSplit(final String[] intFields, final String[] stringFields, final String sessionId, final InetSocketAddress[] nodes, final int splitIndex) {
    final RawFTGSIterator[] splits = new RawFTGSIterator[nodes.length];
    try {
        execute(splits, nodes, new ThrowingFunction<InetSocketAddress, RawFTGSIterator>() {
            public RawFTGSIterator apply(final InetSocketAddress node) throws Exception {
                final ImhotepRemoteSession remoteSession = new ImhotepRemoteSession(node.getHostName(), node.getPort(), sessionId, tempFileSizeBytesLeft);
                remoteSession.setNumStats(numStats);
                return remoteSession.getFTGSIteratorSplit(intFields, stringFields, splitIndex, nodes.length);
            }
        });
    } catch (Throwable t) {
        Closeables2.closeAll(log, splits);
        throw Throwables.propagate(t);
    }
    return mergeFTGSSplits(splits);
}
 

@Override
public RawFTGSIterator mergeSubsetFTGSSplit(final Map<String, long[]> intFields, final Map<String, String[]> stringFields, final String sessionId, final InetSocketAddress[] nodes, final int splitIndex) {
    final RawFTGSIterator[] splits = new RawFTGSIterator[nodes.length];
    try {
        execute(splits, nodes, new ThrowingFunction<InetSocketAddress, RawFTGSIterator>() {
            public RawFTGSIterator apply(final InetSocketAddress node) throws Exception {
                final ImhotepRemoteSession remoteSession = new ImhotepRemoteSession(node.getHostName(), node.getPort(), sessionId, tempFileSizeBytesLeft);
                remoteSession.setNumStats(numStats);
                return remoteSession.getSubsetFTGSIteratorSplit(intFields, stringFields, splitIndex, nodes.length);
            }
        });
    } catch (Throwable t) {
        Closeables2.closeAll(log, splits);
        throw Throwables.propagate(t);
    }
    return mergeFTGSSplits(splits);
}
 

public BlockCompressedStableGeneration(
        BloomFilter.MemoryManager memoryManager, File file, Comparator<K> comparator, Serializer<K> keySerializer, Serializer<V> valueSerializer, final CompressionCodec codec, final boolean mlockBTree
) throws IOException {
    this.file = file;
    reader = new ImmutableBTreeIndex.Reader(file, comparator, keySerializer, new LongSerializer(), mlockBTree);
    final File valuesFile = new File(file, "values.bin");
    recordFile =
            new BlockCompressedRecordFile.Builder(valuesFile, valueSerializer, codec).setMlockFiles(mlockBTree).build();
    final File bloomFilterFile = new File(file, "bloomfilter.bin");
    if (bloomFilterFile.exists()) {
        bloomFilter = new BloomFilter.Reader(memoryManager, bloomFilterFile, keySerializer);
    } else {
        bloomFilter = null;
    }
    sizeInBytes = reader.sizeInBytes()+valuesFile.length()+(bloomFilter == null ? 0 : bloomFilter.sizeInBytes());
    stuffToClose = SharedReference.create((Closeable)new Closeable() {
        public void close() throws IOException {
            Closeables2.closeQuietly(reader, log);
            if (bloomFilter != null) Closeables2.closeQuietly(bloomFilter, log);
            Closeables2.closeQuietly(recordFile, log);
        }
    });
}
 

public void checkpoint(File checkpointDir) throws IOException {
    final SharedReference<GenerationState<K, V>> localState = generationState.getCopy();
    try {
        if (localState == null) {
            throw new IOException("store is closed");
        }
        checkpointDir.mkdirs();
        localState.get().volatileGeneration.checkpoint(checkpointDir);
        for (Generation<K, V> generation : localState.get().stableGenerations) {
            generation.checkpoint(checkpointDir);
        }
        PosixFileOperations.cplr(new File(localState.get().path, "state"), checkpointDir);
    } finally {
        Closeables2.closeQuietly(localState, log);
    }
}
 

/**
 * Flushes volatile generation to disk.
 *
 * @throws IOException  if an I/O error occurs
 */
public void sync() throws IOException {
    final SharedReference<GenerationState<K, V>> localState = generationState.getCopy();
    try {
        if (localState == null) {
            throw new IOException("store is closed");
        }
        try {
            localState.get().volatileGeneration.sync();
        } catch (IOException e) {
            compactor.compact();
            throw e;
        }
    } finally {
        Closeables2.closeQuietly(localState, log);
    }
}
 

public void compact() throws IOException {
    lock.lock();
    try {
        if (!closed) {
            final SharedReference<GenerationState<K, V>> localStateReference = generationState.getCopy();
            try {
                if (localStateReference == null) return;
                final GenerationState<K, V> localState = localStateReference.get();
                //this is double checked locking but in this case it doesn't really matter since it's just a heuristic
                if (localState.volatileGeneration.sizeInBytes() > maxVolatileGenerationSize) {
                    final GenerationState<K, V> nextState = startNewLog(localState);
                    startCompaction(nextState);
                }
            } finally {
                Closeables2.closeQuietly(localStateReference, log);
            }
        }
    } finally {
        lock.unlock();
    }
}
 

private GenerationState<K, V> startNewLog(final GenerationState<K, V> localState) throws IOException {
    //create new volatile generation and checkpoint
    final File newLog = getNextLogFile();
    final VolatileGeneration<K,V> nextVolatileGeneration = new VolatileGeneration<K, V>(newLog, keySerializer, valueSerializer, comparator);
    final List<SharedReference<? extends Generation<K,V>>> nextStableGenerations = Lists.newArrayList();
    nextStableGenerations.add(localState.volatileGenerationReference.copy());
    for (SharedReference<? extends Generation<K, V>> reference : localState.stableGenerationReferences) {
        nextStableGenerations.add(reference.copy());
    }
    final File checkpointDir = getNextCheckpointDir();
    checkpointDir.mkdirs();
    final GenerationState<K,V> nextState = new GenerationState<K, V>(nextStableGenerations, SharedReference.create(nextVolatileGeneration), checkpointDir);
    checkpointGenerationState(nextState, checkpointDir);
    //there will be a brief period of time where there is no writable generation, put and delete will block during this time
    localState.volatileGeneration.closeWriter();
    PosixFileOperations.atomicLink(checkpointDir, new File(root, "latest"));
    final SharedReference<GenerationState<K, V>> oldState = Preconditions.checkNotNull(generationState.getAndSet(nextState));
    oldState.get().delete();
    Closeables2.closeQuietly(oldState, log);
    return nextState;
}
 

public BlockCompressedRecordFile(final Supplier<? extends Either<IOException, ? extends RandomAccessDataInput>> inputSupplier, final Closeable closeable, String file, Serializer<E> serializer, CompressionCodec codec, BlockingQueue<Decompressor> decompressorPool, int blockSize, int recordIndexBits, int padBits, int maxChunkSize) throws IOException {
    this.inputSupplier = inputSupplier;
    this.file = file;
    this.serializer = serializer;
    this.codec = codec;
    this.blockSize = blockSize;
    this.padBits = padBits;
    this.maxChunkSize = maxChunkSize;
    pad = 1<<padBits;
    padMask = ~(long)(pad-1);
    shift = Math.max(recordIndexBits - padBits, 0);
    mask = (1L<<recordIndexBits)-1;
    closeableRef = SharedReference.create(closeable);
    try {
        blockCache = new BlockCache(decompressorPool);
    } catch (Throwable t) {
        Closeables2.closeQuietly(closeableRef, log);
        Throwables.propagateIfInstanceOf(t, IOException.class);
        throw Throwables.propagate(t);
    }
}
 

@Override
public E get(long address) throws IOException {
    final int segmentNum = (int)(address>>> segmentShift);
    final Option<SharedReference<BlockCompressedRecordFile<E>>> option = fileCache.get(segmentNum);
    if (option.isNone()) {
        throw new IOException("address is invalid: "+address);
    }
    final SharedReference<BlockCompressedRecordFile<E>> reference = option.some();
    final E ret;
    try {
        ret = reference.get().get(address & segmentMask);
    } finally {
        Closeables2.closeQuietly(reference, log);
    }
    return ret;
}
 

private IntValueLookup stringTermCountLookup(final String field) throws ImhotepOutOfMemoryException {
    final long memoryUsage = flamdexReader.getNumDocs();

    if (!memory.claimMemory(memoryUsage)) {
        throw new ImhotepOutOfMemoryException();
    }

    final byte[] array = new byte[flamdexReader.getNumDocs()];

    final StringTermDocIterator iterator = flamdexReader.getStringTermDocIterator(field);
    try {
        while (iterator.nextTerm()) {
            while (true) {
                final int n = iterator.fillDocIdBuffer(docIdBuf);
                for (int i = 0; i < n; ++i) {
                    final int doc = docIdBuf[i];
                    if (array[doc] != (byte) 255) {
                        ++array[doc];
                    }
                }
                if (n < BUFFER_SIZE) {
                    break;
                }
            }
        }
    } finally {
        Closeables2.closeQuietly(iterator, log);
    }

    return new MemoryReservingIntValueLookupWrapper(new ByteArrayIntValueLookup(array, 0, 255));
}
 

@Override
public final boolean nextField() {
    // todo: reset/cleanup term iterators etc that are in progress
    synchronized (session) {
        if (intFieldPtr < intFields.length) {
            currentField = intFields[intFieldPtr++];
            currentFieldIsIntType = true;
            if (intTermDocIterator != null) Closeables2.closeQuietly(intTermDocIterator, ImhotepLocalSession.log);
            intTermDocIterator = flamdexReader.get().getIntTermDocIterator(currentField);
            if (session.fieldZeroDocBitsets != null) {
                fieldZeroDocBitset = session.fieldZeroDocBitsets.get(Pair.of(currentField, currentFieldIsIntType));
            }
            termIndex = 0;
            return true;
        }
        if (stringFieldPtr < stringFields.length) {
            currentField = stringFields[stringFieldPtr++];
            currentFieldIsIntType = false;
            if (stringTermDocIterator != null) Closeables2.closeQuietly(stringTermDocIterator, ImhotepLocalSession.log);
            stringTermDocIterator = flamdexReader.get().getStringTermDocIterator(currentField);
            if (session.fieldZeroDocBitsets != null) {
                fieldZeroDocBitset = session.fieldZeroDocBitsets.get(Pair.of(currentField, currentFieldIsIntType));
            }
            termIndex = 0;
            return true;
        }
        currentField = null;
        close();
        if (ImhotepLocalSession.logTiming) {
            ImhotepLocalSession.log.info("intTermsTime: "+intTermsTime/1000000d+" ms, stringTermsTime: "+stringTermsTime/1000000d+" ms, docsTime: "+docsTime/1000000d+" ms, lookupsTime: "+lookupsTime/1000000d+" ms, timingErrorTime: "+timingErrorTime/1000000d+" ms");
        }
        return false;
    }
}
 

public <Z, T extends Throwable> Z doWithSession(String sessionId, ThrowingFunction<ImhotepSession, Z, T> f) throws T {
    final SharedReference<ImhotepSession> sessionRef = getSessionManager().getSession(sessionId);
    try {
        return f.apply(sessionRef.get());
    } finally {
        Closeables2.closeQuietly(sessionRef, log);
    }
}
 

public <Z> Z doWithSession(String sessionId, Function<ImhotepSession, Z> f)  {
    final SharedReference<ImhotepSession> sessionRef = getSessionManager().getSession(sessionId);
    try {
        return f.apply(sessionRef.get());
    } finally {
        Closeables2.closeQuietly(sessionRef, log);
    }
}
 

@Override
public void close() {
    if (!closed) {
        closed = true;
        Closeables2.closeQuietly(reference, log);
    }
}
 

@Override
protected void postClose() {
    if (memory.usedMemory() > 0) {
        log.error("MTImhotepMultiSession is leaking! usedMemory = "+memory.usedMemory());
    }
    Closeables2.closeQuietly(memory, log);
}
 

@Override
public void removeAndCloseIfExists(final String sessionId) {
    final SharedReference<ImhotepSession> imhotepSession;
    synchronized (sessionMap) {
        final Session<E> session = sessionMap.remove(sessionId);
        if (session == null) {
            return;
        }
        imhotepSession = session.imhotepSession;
    }
    Closeables2.closeQuietly(imhotepSession, log);
}
 

@Override
public void removeAndCloseIfExists(final String sessionId, Exception e) {
    final SharedReference<ImhotepSession> imhotepSession;
    synchronized (sessionMap) {
        final Session<E> session = sessionMap.remove(sessionId);
        if (session == null) {
            return;
        }
        imhotepSession = session.imhotepSession;
    }
    failureCauseMap.put(sessionId, e);
    Closeables2.closeQuietly(imhotepSession, log);
}
 

private List<ShardInfo> buildShardList() throws IOException {
    final Map<String, Map<String, AtomicSharedReference<Shard>>> localShards = shards;
    final List<ShardInfo> ret = new ArrayList<ShardInfo>();
    for (final Map<String, AtomicSharedReference<Shard>> map : localShards.values()) {
        for (final String shardName : map.keySet()) {
            final SharedReference<Shard> ref = map.get(shardName).get();
            try {
                if (ref != null) {
                    final Shard shard = ref.get();
                    ret.add(new ShardInfo(shard.getDataset(), shardName,
                                          shard.getLoadedMetrics(), shard.getNumDocs(),
                                          shard.getShardVersion()));
                }
            } finally {
                Closeables2.closeQuietly(ref, log);
            }
        }
    }
    Collections.sort(ret, new Comparator<ShardInfo>() {
        @Override
        public int compare(ShardInfo o1, ShardInfo o2) {
            final int c = o1.dataset.compareTo(o2.dataset);
            if (c != 0)
                return c;
            return o1.shardId.compareTo(o2.shardId);
        }
    });
    return ret;
}
 

private List<DatasetInfo> buildDatasetList() throws IOException {
    final Map<String, Map<String, AtomicSharedReference<Shard>>> localShards = shards;
    final List<DatasetInfo> ret = Lists.newArrayList();
    for (final Map.Entry<String, Map<String, AtomicSharedReference<Shard>>> e : localShards.entrySet()) {
        final String dataset = e.getKey();
        final Map<String, AtomicSharedReference<Shard>> map = e.getValue();
        final List<ShardInfo> shardList = Lists.newArrayList();
        final Set<String> intFields = Sets.newHashSet();
        final Set<String> stringFields = Sets.newHashSet();
        final Set<String> metrics = Sets.newHashSet();
        for (final String shardName : map.keySet()) {
            final SharedReference<Shard> ref = map.get(shardName).get();
            try {
                if (ref != null) {
                    final Shard shard = ref.get();
                    shardList.add(new ShardInfo(shard.getDataset(), shardName,
                                                shard.getLoadedMetrics(), shard.getNumDocs(),
                                                shard.getShardVersion()));
                    intFields.addAll(shard.getIntFields());
                    stringFields.addAll(shard.getStringFields());
                    metrics.addAll(shard.getAvailableMetrics());
                }
            } finally {
                Closeables2.closeQuietly(ref, log);
            }
        }
        ret.add(new DatasetInfo(dataset, shardList, intFields, stringFields, metrics));
    }
    return ret;
}
 

private List<ShardInfo> buildShardList() throws IOException {
    final Map<String, Map<String, AtomicSharedReference<Shard>>> localShards = shards;
    final List<ShardInfo> ret = new ArrayList<ShardInfo>();
    for (final Map<String, AtomicSharedReference<Shard>> map : localShards.values()) {
        for (final String shardName : map.keySet()) {
            final SharedReference<Shard> ref = map.get(shardName).getCopy();
            try {
                if (ref != null) {
                    final Shard shard = ref.get();
                    ret.add(new ShardInfo(shard.getDataset(), shardName,
                                          shard.getLoadedMetrics(), shard.getNumDocs(),
                                          shard.getShardVersion()));
                }
            } finally {
                Closeables2.closeQuietly(ref, log);
            }
        }
    }
    Collections.sort(ret, new Comparator<ShardInfo>() {
        @Override
        public int compare(ShardInfo o1, ShardInfo o2) {
            final int c = o1.dataset.compareTo(o2.dataset);
            if (c != 0) {
                return c;
            }
            return o1.shardId.compareTo(o2.shardId);
        }
    });
    return ret;
}
 

private List<DatasetInfo> buildDatasetList() throws IOException {
    final Map<String, Map<String, AtomicSharedReference<Shard>>> localShards = shards;
    final List<DatasetInfo> ret = Lists.newArrayList();
    for (final Map.Entry<String, Map<String, AtomicSharedReference<Shard>>> e : localShards.entrySet()) {
        final String dataset = e.getKey();
        final Map<String, AtomicSharedReference<Shard>> map = e.getValue();
        final List<ShardInfo> shardList = Lists.newArrayList();
        final Set<String> intFields = Sets.newHashSet();
        final Set<String> stringFields = Sets.newHashSet();
        final Set<String> metrics = Sets.newHashSet();
        for (final String shardName : map.keySet()) {
            final SharedReference<Shard> ref = map.get(shardName).getCopy();
            try {
                if (ref != null) {
                    final Shard shard = ref.get();
                    shardList.add(new ShardInfo(shard.getDataset(), shardName,
                                                shard.getLoadedMetrics(), shard.getNumDocs(),
                                                shard.getShardVersion()));
                    intFields.addAll(shard.getIntFields());
                    stringFields.addAll(shard.getStringFields());
                    metrics.addAll(shard.getAvailableMetrics());
                }
            } finally {
                Closeables2.closeQuietly(ref, log);
            }
        }
        ret.add(new DatasetInfo(dataset, shardList, intFields, stringFields, metrics));
    }
    return ret;
}
 

public Set<String> getLoadedMetrics() {
    final SharedReference<CachedFlamdexReader> copy = ref.copyIfLoaded();
    if (copy != null) {
        try {
            return copy.get().getLoadedMetrics();
        } finally {
            Closeables2.closeQuietly(copy, log);
        }
    }
    return Collections.emptySet();
}
 

public List<ImhotepStatusDump.MetricDump> getMetricDump() {
    final SharedReference<CachedFlamdexReader> copy = ref.copyIfLoaded();
    if (copy != null) {
        try {
            return copy.get().getMetricDump();
        } finally {
            Closeables2.closeQuietly(copy, log);
        }
    }
    return Collections.emptyList();
}
 

@Override
public InputStream getInputStreamForFile(String fullPath, long startOffset, long maxReadLength) throws IOException {
    final String relativePath = mounter.getMountRelativePath(fullPath, mountPoint);
    final Path hdfsPath = new Path(hdfsBasePath, relativePath);
    final FSDataInputStream stream = fs.open(hdfsPath);
    try {
        stream.seek(startOffset);
    } catch (final IOException e) {
        Closeables2.closeQuietly(stream, log);
        throw new IOException("Failed to open " + fullPath + " with offset " + startOffset, e);
    }
    return stream;
}
 

@Override
public void close() throws IOException {
    synchronized (lockFileMap) {
        final RandomAccessFile randomAccessFile = lockFileMap.get(indexDir);
        Closeables2.closeQuietly(randomAccessFile, log);
        lockFileMap.remove(indexDir);
    }
}
 

@Override
public void close() {
    if (!closed) {
        closed = true;
        Closeables2.closeQuietly(file, log);
        Closeables2.closeQuietly(buffer, log);
    }
}