ShortCircuitCache implements Closeable { @Override public void close() { try { lock.lock(); if (closed) return; closed = true; LOG.info(this + ": closing"); maxNonMmappedEvictableLifespanMs = 0; maxEvictableMmapedSize = 0; IOUtils.cleanup(LOG, cacheCleaner); while (true) { Entry<Long, ShortCircuitReplica> entry = evictable.firstEntry(); if (entry == null) break; purge(entry.getValue()); } while (true) { Entry<Long, ShortCircuitReplica> entry = evictableMmapped.firstEntry(); if (entry == null) break; purge(entry.getValue()); } } finally { lock.unlock(); } releaserExecutor.shutdown(); cleanerExecutor.shutdown(); try { if (!releaserExecutor.awaitTermination(30, TimeUnit.SECONDS)) { LOG.error("Forcing SlotReleaserThreadPool to shutdown!"); releaserExecutor.shutdownNow(); } } catch (InterruptedException e) { releaserExecutor.shutdownNow(); Thread.currentThread().interrupt(); LOG.error("Interrupted while waiting for SlotReleaserThreadPool " + "to terminate", e); } try { if (!cleanerExecutor.awaitTermination(30, TimeUnit.SECONDS)) { LOG.error("Forcing CleanerThreadPool to shutdown!"); cleanerExecutor.shutdownNow(); } } catch (InterruptedException e) { cleanerExecutor.shutdownNow(); Thread.currentThread().interrupt(); LOG.error("Interrupted while waiting for CleanerThreadPool " + "to terminate", e); } IOUtils.cleanup(LOG, shmManager); } ShortCircuitCache(int maxTotalSize, long maxNonMmappedEvictableLifespanMs,
      int maxEvictableMmapedSize, long maxEvictableMmapedLifespanMs,
      long mmapRetryTimeoutMs, long staleThresholdMs, int shmInterruptCheckMs); static ShortCircuitCache fromConf(ShortCircuitConf conf); long getStaleThresholdMs(); ShortCircuitReplicaInfo fetchOrCreate(ExtendedBlockId key,
      ShortCircuitReplicaCreator creator); @Override void close(); @VisibleForTesting // ONLY for testing void accept(CacheVisitor visitor); @Override String toString(); Slot allocShmSlot(DatanodeInfo datanode,
        DomainPeer peer, MutableBoolean usedPeer,
        ExtendedBlockId blockId, String clientName); void freeSlot(Slot slot); void scheduleSlotReleaser(Slot slot); @VisibleForTesting DfsClientShmManager getDfsClientShmManager(); static final Log LOG; }
@Test(timeout=60000) public void testCreateAndDestroy() throws Exception { ShortCircuitCache cache = new ShortCircuitCache(10, 1, 10, 1, 1, 10000, 0); cache.close(); }
@Test(timeout=60000) public void testDataXceiverCleansUpSlotsOnFailure() throws Exception { BlockReaderTestUtil.enableShortCircuitShmTracing(); TemporarySocketDirectory sockDir = new TemporarySocketDirectory(); Configuration conf = createShortCircuitConf( "testDataXceiverCleansUpSlotsOnFailure", sockDir); conf.setLong( HdfsClientConfigKeys.Read.ShortCircuit.STREAMS_CACHE_EXPIRY_MS_KEY, 1000000000L); MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build(); cluster.waitActive(); DistributedFileSystem fs = cluster.getFileSystem(); final Path TEST_PATH1 = new Path("/test_file1"); final Path TEST_PATH2 = new Path("/test_file2"); final int TEST_FILE_LEN = 4096; final int SEED = 0xFADE1; DFSTestUtil.createFile(fs, TEST_PATH1, TEST_FILE_LEN, (short)1, SEED); DFSTestUtil.createFile(fs, TEST_PATH2, TEST_FILE_LEN, (short)1, SEED); DFSTestUtil.readFileBuffer(fs, TEST_PATH1); fs.getClient().getConf().getShortCircuitConf().brfFailureInjector = new TestCleanupFailureInjector(); try { DFSTestUtil.readFileBuffer(fs, TEST_PATH2); } catch (Throwable t) { GenericTestUtils.assertExceptionContains("TCP reads were disabled for " + "testing, but we failed to do a non-TCP read.", t); } ShortCircuitRegistry registry = cluster.getDataNodes().get(0).getShortCircuitRegistry(); registry.visit(new ShortCircuitRegistry.Visitor() { @Override public void accept(HashMap<ShmId, RegisteredShm> segments, HashMultimap<ExtendedBlockId, Slot> slots) { Assert.assertEquals(1, segments.size()); Assert.assertEquals(1, slots.size()); } }); cluster.shutdown(); sockDir.close(); }
@Test(timeout=60000) public void testPreReceiptVerificationDfsClientCanDoScr() throws Exception { BlockReaderTestUtil.enableShortCircuitShmTracing(); TemporarySocketDirectory sockDir = new TemporarySocketDirectory(); Configuration conf = createShortCircuitConf( "testPreReceiptVerificationDfsClientCanDoScr", sockDir); conf.setLong( HdfsClientConfigKeys.Read.ShortCircuit.STREAMS_CACHE_EXPIRY_MS_KEY, 1000000000L); MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build(); cluster.waitActive(); DistributedFileSystem fs = cluster.getFileSystem(); fs.getClient().getConf().getShortCircuitConf().brfFailureInjector = new TestPreReceiptVerificationFailureInjector(); final Path TEST_PATH1 = new Path("/test_file1"); DFSTestUtil.createFile(fs, TEST_PATH1, 4096, (short)1, 0xFADE2); final Path TEST_PATH2 = new Path("/test_file2"); DFSTestUtil.createFile(fs, TEST_PATH2, 4096, (short)1, 0xFADE2); DFSTestUtil.readFileBuffer(fs, TEST_PATH1); DFSTestUtil.readFileBuffer(fs, TEST_PATH2); ShortCircuitRegistry registry = cluster.getDataNodes().get(0).getShortCircuitRegistry(); registry.visit(new ShortCircuitRegistry.Visitor() { @Override public void accept(HashMap<ShmId, RegisteredShm> segments, HashMultimap<ExtendedBlockId, Slot> slots) { Assert.assertEquals(1, segments.size()); Assert.assertEquals(2, slots.size()); } }); cluster.shutdown(); sockDir.close(); }