TarFiles implements Closeable { public CleanupResult cleanup(CleanupContext context) throws IOException { CleanupResult result = new CleanupResult(); result.removableFiles = new ArrayList<>(); result.reclaimedSegmentIds = new HashSet<>(); Set<UUID> references; Node head; lock.writeLock().lock(); lock.readLock().lock(); try { try { internalNewWriter(); } finally { lock.writeLock().unlock(); } head = readers; references = new HashSet<>(context.initialReferences()); } finally { lock.readLock().unlock(); } Map<TarReader, TarReader> cleaned = new LinkedHashMap<>(); for (TarReader reader : iterable(head)) { cleaned.put(reader, reader); result.reclaimedSize += reader.size(); } Set<UUID> reclaim = newHashSet(); for (TarReader reader : cleaned.keySet()) { if (shutdown) { result.interrupted = true; return result; } reader.mark(references, reclaim, context); } for (TarReader reader : cleaned.keySet()) { if (shutdown) { result.interrupted = true; return result; } cleaned.put(reader, reader.sweep(reclaim, result.reclaimedSegmentIds)); } Node closeables; long reclaimed; Node swept; while (true) { closeables = null; reclaimed = 0; swept = null; for (TarReader reader : iterable(head)) { if (cleaned.containsKey(reader)) { TarReader cleanedReader = cleaned.get(reader); if (cleanedReader != null) { swept = new Node(cleanedReader, swept); reclaimed += cleanedReader.size(); } if (cleanedReader != reader) { closeables = new Node(reader, closeables); } } else { swept = new Node(reader, swept); } } swept = reverse(swept); lock.writeLock().lock(); try { if (readers == head) { readers = swept; break; } else { head = readers; } } finally { lock.writeLock().unlock(); } } readerCount.dec(getSize(head) - getSize(swept)); segmentCount.dec(getSegmentCount(head) - getSegmentCount(swept)); result.reclaimedSize -= reclaimed; for (TarReader closeable : iterable(closeables)) { try { closeable.close(); } catch (IOException e) { log.warn("Unable to close swept TAR reader", e); } result.removableFiles.add(closeable.getFileName()); } return result; } private  TarFiles(Builder builder); static Builder builder(); @Override void close(); @Override String toString(); long size(); int readerCount(); int segmentCount(); void flush(); boolean containsSegment(long msb, long lsb); Buffer readSegment(long msb, long lsb); void writeSegment(UUID id, byte[] buffer, int offset, int length, GCGeneration generation, Set<UUID> references, Set<String> binaryReferences); CleanupResult cleanup(CleanupContext context); void collectBlobReferences(Consumer<String> collector, Predicate<GCGeneration> reclaim); Iterable<UUID> getSegmentIds(); Map<UUID, Set<UUID>> getGraph(String fileName); Map<String, Set<UUID>> getIndices(); FileReaper createFileReaper();  }
@Test public void testCleanup() throws Exception { UUID a = randomUUID(); UUID b = randomUUID(); UUID c = randomUUID(); UUID d = randomUUID(); UUID e = randomUUID(); writeSegment(a); writeSegment(b); writeSegmentWithReferences(c, a, b); writeSegment(d); writeSegmentWithReferences(e, a, d); CleanupResult result = tarFiles.cleanup(new CleanupContext() { @Override public Collection<UUID> initialReferences() { return singletonList(e); } @Override public boolean shouldReclaim(UUID id, GCGeneration generation, boolean referenced) { return !referenced; } @Override public boolean shouldFollow(UUID from, UUID to) { return true; } }); assertFalse(result.isInterrupted()); assertFalse(result.getRemovableFiles().isEmpty()); assertEquals(new HashSet<>(asList(c, b)), result.getReclaimedSegmentIds()); assertTrue(result.getReclaimedSize() > 0); }