@Test(timeout=60000) public void testPreReceiptVerificationDfsClientCanDoScr() throws Exception { BlockReaderTestUtil.enableShortCircuitShmTracing(); TemporarySocketDirectory sockDir = new TemporarySocketDirectory(); Configuration conf = createShortCircuitConf( "testPreReceiptVerificationDfsClientCanDoScr", sockDir); conf.setLong(DFS_CLIENT_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().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(); }
@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(); } IOUtils.cleanup(LOG, shmManager); }
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(); } IOUtils.cleanup(LOG, shmManager); } }
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(); } IOUtils.cleanup(LOG, shmManager); } ShortCircuitCache(int maxTotalSize, long maxNonMmappedEvictableLifespanMs,
      int maxEvictableMmapedSize, long maxEvictableMmapedLifespanMs,
      long mmapRetryTimeoutMs, long staleThresholdMs, int shmInterruptCheckMs); }
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(); } IOUtils.cleanup(LOG, shmManager); } ShortCircuitCache(int maxTotalSize, long maxNonMmappedEvictableLifespanMs,
      int maxEvictableMmapedSize, long maxEvictableMmapedLifespanMs,
      long mmapRetryTimeoutMs, long staleThresholdMs, int shmInterruptCheckMs); static ShortCircuitCache fromConf(Configuration 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(); }
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(); } IOUtils.cleanup(LOG, shmManager); } ShortCircuitCache(int maxTotalSize, long maxNonMmappedEvictableLifespanMs,
      int maxEvictableMmapedSize, long maxEvictableMmapedLifespanMs,
      long mmapRetryTimeoutMs, long staleThresholdMs, int shmInterruptCheckMs); static ShortCircuitCache fromConf(Configuration 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; }