@Test(timeout=60000) public void testDataXceiverHandlesRequestShortCircuitShmFailure() throws Exception { BlockReaderTestUtil.enableShortCircuitShmTracing(); TemporarySocketDirectory sockDir = new TemporarySocketDirectory(); Configuration conf = createShortCircuitConf( "testDataXceiverHandlesRequestShortCircuitShmFailure", 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(); final Path TEST_PATH1 = new Path("/test_file1"); DFSTestUtil.createFile(fs, TEST_PATH1, 4096, (short)1, 0xFADE1); LOG.info("Setting failure injector and performing a read which " + "should fail..."); DataNodeFaultInjector failureInjector = Mockito.mock(DataNodeFaultInjector.class); Mockito.doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { throw new IOException("injected error into sendShmResponse"); } }).when(failureInjector).sendShortCircuitShmResponse(); DataNodeFaultInjector prevInjector = DataNodeFaultInjector.instance; DataNodeFaultInjector.instance = failureInjector; try { DFSTestUtil.readFileBuffer(fs, TEST_PATH1); Assert.fail("expected readFileBuffer to fail, but it succeeded."); } catch (Throwable t) { GenericTestUtils.assertExceptionContains("TCP reads were disabled for " + "testing, but we failed to do a non-TCP read.", t); } checkNumberOfSegmentsAndSlots(0, 0, cluster.getDataNodes().get(0).getShortCircuitRegistry()); LOG.info("Clearing failure injector and performing another read..."); DataNodeFaultInjector.instance = prevInjector; fs.getClient().getClientContext().getDomainSocketFactory().clearPathMap(); DFSTestUtil.readFileBuffer(fs, TEST_PATH1); checkNumberOfSegmentsAndSlots(1, 1, cluster.getDataNodes().get(0).getShortCircuitRegistry()); 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; }