PersistentDiskCache extends AbstractPersistentCache { @Override public void cleanUp() { if (!cleanupInProgress.getAndSet(true)) { try { cleanUpInternal(); } finally { cleanupInProgress.set(false); } } } PersistentDiskCache(File directory, int cacheMaxSizeMB, IOMonitor diskCacheIOMonitor); @Override boolean containsSegment(long msb, long lsb); @Override void writeSegment(long msb, long lsb, Buffer buffer); @Override void cleanUp(); static final int DEFAULT_MAX_CACHE_SIZE_MB; static final String NAME; }
@Test public void cleanupTest() throws Exception { persistentCache.close(); persistentCache = new PersistentDiskCache(temporaryFolder.newFolder(), 0, new IOMonitorAdapter()); final List<TestSegment> testSegments = new ArrayList<>(SEGMENTS); final List<Map<String, Buffer>> segmentsRead = new ArrayList<>(THREADS); for (int i = 0; i < SEGMENTS; ++i) { testSegments.add(TestSegment.createSegment()); } for (int i = 0; i < THREADS; ++i) { final Map<String, Buffer> segmentsReadThisThread = new HashMap<>(SEGMENTS_PER_THREAD); segmentsRead.add(segmentsReadThisThread); } runConcurrently.accept((nThread, nSegment) -> { TestSegment segment = testSegments.get(nSegment); long[] id = segment.getSegmentId(); try { persistentCache.writeSegment(id[0], id[1], segment.getSegmentBuffer()); } catch (Throwable t) { errors.incrementAndGet(); } finally { done.incrementAndGet(); } }); waitWhile.accept(() -> done.get() < SEGMENTS); waitWhile.accept(() -> persistentCache.getWritesPending() > 0); assertEquals("Errors have occurred while writing", 0, errors.get()); assertNoTimeout(); done.set(0); waitWhile.accept(() -> ((PersistentDiskCache) persistentCache).cleanupInProgress.get()); persistentCache.cleanUp(); runConcurrently.accept((nThread, nSegment) -> { final TestSegment segment = testSegments.get(nSegment); final long[] id = segment.getSegmentId(); try { final Map<String, Buffer> segmentsReadThisThread = segmentsRead.get(nThread); final Buffer segmentRead = persistentCache.readSegment(id[0], id[1], () -> null); segmentsReadThisThread.put(new UUID(id[0], id[1]).toString(), segmentRead); } catch (Throwable t) { errors.incrementAndGet(); } finally { done.incrementAndGet(); } }); waitWhile.accept(() -> done.get() < SEGMENTS); assertNoTimeout(); assertEquals("Errors have occurred while reading", 0, errors.get()); errors.set(0); for (int i = 0; i < THREADS; ++i) { for (int j = i * SEGMENTS_PER_THREAD; j < (i + 1) * SEGMENTS_PER_THREAD; ++j) { TestSegment testSegment = testSegments.get(j); byte[] testSegmentBytes = testSegment.getSegmentBytes(); Map<String, Buffer> segmentsReadThisThread = segmentsRead.get(i); long[] segmentReadId = testSegment.getSegmentId(); Buffer segmentRead = segmentsReadThisThread.get(new UUID(segmentReadId[0], segmentReadId[1]).toString()); if (segmentRead == null) { errors.incrementAndGet(); } } } assertEquals("Segment(s) not cleaned up in cache", 0, SEGMENTS - errors.get()); }