@Test(timeout=60000) public void testShmBasedStaleness() throws Exception { BlockReaderTestUtil.enableShortCircuitShmTracing(); TemporarySocketDirectory sockDir = new TemporarySocketDirectory(); Configuration conf = createShortCircuitConf("testShmBasedStaleness", sockDir); MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build(); cluster.waitActive(); DistributedFileSystem fs = cluster.getFileSystem(); final ShortCircuitCache cache = fs.getClient().getClientContext().getShortCircuitCache(); String TEST_FILE = "/test_file"; final int TEST_FILE_LEN = 8193; final int SEED = 0xFADED; DFSTestUtil.createFile(fs, new Path(TEST_FILE), TEST_FILE_LEN, (short)1, SEED); FSDataInputStream fis = fs.open(new Path(TEST_FILE)); int first = fis.read(); final ExtendedBlock block = DFSTestUtil.getFirstBlock(fs, new Path(TEST_FILE)); Assert.assertTrue(first != -1); cache.accept(new CacheVisitor() { @Override public void visit(int numOutstandingMmaps, Map<ExtendedBlockId, ShortCircuitReplica> replicas, Map<ExtendedBlockId, InvalidToken> failedLoads, Map<Long, ShortCircuitReplica> evictable, Map<Long, ShortCircuitReplica> evictableMmapped) { ShortCircuitReplica replica = replicas.get( ExtendedBlockId.fromExtendedBlock(block)); Assert.assertNotNull(replica); Assert.assertTrue(replica.getSlot().isValid()); } }); cluster.getDataNodes().get(0).shutdown(); cache.accept(new CacheVisitor() { @Override public void visit(int numOutstandingMmaps, Map<ExtendedBlockId, ShortCircuitReplica> replicas, Map<ExtendedBlockId, InvalidToken> failedLoads, Map<Long, ShortCircuitReplica> evictable, Map<Long, ShortCircuitReplica> evictableMmapped) { ShortCircuitReplica replica = replicas.get( ExtendedBlockId.fromExtendedBlock(block)); Assert.assertNotNull(replica); Assert.assertFalse(replica.getSlot().isValid()); } }); cluster.shutdown(); }
@VisibleForTesting public void accept(CacheVisitor visitor) { lock.lock(); try { Map<ExtendedBlockId, ShortCircuitReplica> replicas = new HashMap<ExtendedBlockId, ShortCircuitReplica>(); Map<ExtendedBlockId, InvalidToken> failedLoads = new HashMap<ExtendedBlockId, InvalidToken>(); for (Entry<ExtendedBlockId, Waitable<ShortCircuitReplicaInfo>> entry : replicaInfoMap.entrySet()) { Waitable<ShortCircuitReplicaInfo> waitable = entry.getValue(); if (waitable.hasVal()) { if (waitable.getVal().getReplica() != null) { replicas.put(entry.getKey(), waitable.getVal().getReplica()); } else { failedLoads.put(entry.getKey(), waitable.getVal().getInvalidTokenException()); } } } if (LOG.isDebugEnabled()) { StringBuilder builder = new StringBuilder(); builder.append("visiting ").append(visitor.getClass().getName()). append("with outstandingMmapCount=").append(outstandingMmapCount). append(", replicas="); String prefix = ""; for (Entry<ExtendedBlockId, ShortCircuitReplica> entry : replicas.entrySet()) { builder.append(prefix).append(entry.getValue()); prefix = ","; } prefix = ""; builder.append(", failedLoads="); for (Entry<ExtendedBlockId, InvalidToken> entry : failedLoads.entrySet()) { builder.append(prefix).append(entry.getValue()); prefix = ","; } prefix = ""; builder.append(", evictable="); for (Entry<Long, ShortCircuitReplica> entry : evictable.entrySet()) { builder.append(prefix).append(entry.getKey()). append(":").append(entry.getValue()); prefix = ","; } prefix = ""; builder.append(", evictableMmapped="); for (Entry<Long, ShortCircuitReplica> entry : evictableMmapped.entrySet()) { builder.append(prefix).append(entry.getKey()). append(":").append(entry.getValue()); prefix = ","; } LOG.debug(builder.toString()); } visitor.visit(outstandingMmapCount, replicas, failedLoads, evictable, evictableMmapped); } finally { lock.unlock(); } }
ShortCircuitCache implements Closeable { @VisibleForTesting public void accept(CacheVisitor visitor) { lock.lock(); try { Map<ExtendedBlockId, ShortCircuitReplica> replicas = new HashMap<ExtendedBlockId, ShortCircuitReplica>(); Map<ExtendedBlockId, InvalidToken> failedLoads = new HashMap<ExtendedBlockId, InvalidToken>(); for (Entry<ExtendedBlockId, Waitable<ShortCircuitReplicaInfo>> entry : replicaInfoMap.entrySet()) { Waitable<ShortCircuitReplicaInfo> waitable = entry.getValue(); if (waitable.hasVal()) { if (waitable.getVal().getReplica() != null) { replicas.put(entry.getKey(), waitable.getVal().getReplica()); } else { failedLoads.put(entry.getKey(), waitable.getVal().getInvalidTokenException()); } } } if (LOG.isDebugEnabled()) { StringBuilder builder = new StringBuilder(); builder.append("visiting ").append(visitor.getClass().getName()). append("with outstandingMmapCount=").append(outstandingMmapCount). append(", replicas="); String prefix = ""; for (Entry<ExtendedBlockId, ShortCircuitReplica> entry : replicas.entrySet()) { builder.append(prefix).append(entry.getValue()); prefix = ","; } prefix = ""; builder.append(", failedLoads="); for (Entry<ExtendedBlockId, InvalidToken> entry : failedLoads.entrySet()) { builder.append(prefix).append(entry.getValue()); prefix = ","; } prefix = ""; builder.append(", evictable="); for (Entry<Long, ShortCircuitReplica> entry : evictable.entrySet()) { builder.append(prefix).append(entry.getKey()). append(":").append(entry.getValue()); prefix = ","; } prefix = ""; builder.append(", evictableMmapped="); for (Entry<Long, ShortCircuitReplica> entry : evictableMmapped.entrySet()) { builder.append(prefix).append(entry.getKey()). append(":").append(entry.getValue()); prefix = ","; } LOG.debug(builder.toString()); } visitor.visit(outstandingMmapCount, replicas, failedLoads, evictable, evictableMmapped); } finally { lock.unlock(); } } }
ShortCircuitCache implements Closeable { @VisibleForTesting public void accept(CacheVisitor visitor) { lock.lock(); try { Map<ExtendedBlockId, ShortCircuitReplica> replicas = new HashMap<ExtendedBlockId, ShortCircuitReplica>(); Map<ExtendedBlockId, InvalidToken> failedLoads = new HashMap<ExtendedBlockId, InvalidToken>(); for (Entry<ExtendedBlockId, Waitable<ShortCircuitReplicaInfo>> entry : replicaInfoMap.entrySet()) { Waitable<ShortCircuitReplicaInfo> waitable = entry.getValue(); if (waitable.hasVal()) { if (waitable.getVal().getReplica() != null) { replicas.put(entry.getKey(), waitable.getVal().getReplica()); } else { failedLoads.put(entry.getKey(), waitable.getVal().getInvalidTokenException()); } } } if (LOG.isDebugEnabled()) { StringBuilder builder = new StringBuilder(); builder.append("visiting ").append(visitor.getClass().getName()). append("with outstandingMmapCount=").append(outstandingMmapCount). append(", replicas="); String prefix = ""; for (Entry<ExtendedBlockId, ShortCircuitReplica> entry : replicas.entrySet()) { builder.append(prefix).append(entry.getValue()); prefix = ","; } prefix = ""; builder.append(", failedLoads="); for (Entry<ExtendedBlockId, InvalidToken> entry : failedLoads.entrySet()) { builder.append(prefix).append(entry.getValue()); prefix = ","; } prefix = ""; builder.append(", evictable="); for (Entry<Long, ShortCircuitReplica> entry : evictable.entrySet()) { builder.append(prefix).append(entry.getKey()). append(":").append(entry.getValue()); prefix = ","; } prefix = ""; builder.append(", evictableMmapped="); for (Entry<Long, ShortCircuitReplica> entry : evictableMmapped.entrySet()) { builder.append(prefix).append(entry.getKey()). append(":").append(entry.getValue()); prefix = ","; } LOG.debug(builder.toString()); } visitor.visit(outstandingMmapCount, replicas, failedLoads, evictable, evictableMmapped); } finally { lock.unlock(); } } ShortCircuitCache(int maxTotalSize, long maxNonMmappedEvictableLifespanMs,
      int maxEvictableMmapedSize, long maxEvictableMmapedLifespanMs,
      long mmapRetryTimeoutMs, long staleThresholdMs, int shmInterruptCheckMs); }
ShortCircuitCache implements Closeable { @VisibleForTesting public void accept(CacheVisitor visitor) { lock.lock(); try { Map<ExtendedBlockId, ShortCircuitReplica> replicas = new HashMap<ExtendedBlockId, ShortCircuitReplica>(); Map<ExtendedBlockId, InvalidToken> failedLoads = new HashMap<ExtendedBlockId, InvalidToken>(); for (Entry<ExtendedBlockId, Waitable<ShortCircuitReplicaInfo>> entry : replicaInfoMap.entrySet()) { Waitable<ShortCircuitReplicaInfo> waitable = entry.getValue(); if (waitable.hasVal()) { if (waitable.getVal().getReplica() != null) { replicas.put(entry.getKey(), waitable.getVal().getReplica()); } else { failedLoads.put(entry.getKey(), waitable.getVal().getInvalidTokenException()); } } } if (LOG.isDebugEnabled()) { StringBuilder builder = new StringBuilder(); builder.append("visiting ").append(visitor.getClass().getName()). append("with outstandingMmapCount=").append(outstandingMmapCount). append(", replicas="); String prefix = ""; for (Entry<ExtendedBlockId, ShortCircuitReplica> entry : replicas.entrySet()) { builder.append(prefix).append(entry.getValue()); prefix = ","; } prefix = ""; builder.append(", failedLoads="); for (Entry<ExtendedBlockId, InvalidToken> entry : failedLoads.entrySet()) { builder.append(prefix).append(entry.getValue()); prefix = ","; } prefix = ""; builder.append(", evictable="); for (Entry<Long, ShortCircuitReplica> entry : evictable.entrySet()) { builder.append(prefix).append(entry.getKey()). append(":").append(entry.getValue()); prefix = ","; } prefix = ""; builder.append(", evictableMmapped="); for (Entry<Long, ShortCircuitReplica> entry : evictableMmapped.entrySet()) { builder.append(prefix).append(entry.getKey()). append(":").append(entry.getValue()); prefix = ","; } LOG.debug(builder.toString()); } visitor.visit(outstandingMmapCount, replicas, failedLoads, evictable, evictableMmapped); } finally { lock.unlock(); } } 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 { @VisibleForTesting public void accept(CacheVisitor visitor) { lock.lock(); try { Map<ExtendedBlockId, ShortCircuitReplica> replicas = new HashMap<ExtendedBlockId, ShortCircuitReplica>(); Map<ExtendedBlockId, InvalidToken> failedLoads = new HashMap<ExtendedBlockId, InvalidToken>(); for (Entry<ExtendedBlockId, Waitable<ShortCircuitReplicaInfo>> entry : replicaInfoMap.entrySet()) { Waitable<ShortCircuitReplicaInfo> waitable = entry.getValue(); if (waitable.hasVal()) { if (waitable.getVal().getReplica() != null) { replicas.put(entry.getKey(), waitable.getVal().getReplica()); } else { failedLoads.put(entry.getKey(), waitable.getVal().getInvalidTokenException()); } } } if (LOG.isDebugEnabled()) { StringBuilder builder = new StringBuilder(); builder.append("visiting ").append(visitor.getClass().getName()). append("with outstandingMmapCount=").append(outstandingMmapCount). append(", replicas="); String prefix = ""; for (Entry<ExtendedBlockId, ShortCircuitReplica> entry : replicas.entrySet()) { builder.append(prefix).append(entry.getValue()); prefix = ","; } prefix = ""; builder.append(", failedLoads="); for (Entry<ExtendedBlockId, InvalidToken> entry : failedLoads.entrySet()) { builder.append(prefix).append(entry.getValue()); prefix = ","; } prefix = ""; builder.append(", evictable="); for (Entry<Long, ShortCircuitReplica> entry : evictable.entrySet()) { builder.append(prefix).append(entry.getKey()). append(":").append(entry.getValue()); prefix = ","; } prefix = ""; builder.append(", evictableMmapped="); for (Entry<Long, ShortCircuitReplica> entry : evictableMmapped.entrySet()) { builder.append(prefix).append(entry.getKey()). append(":").append(entry.getValue()); prefix = ","; } LOG.debug(builder.toString()); } visitor.visit(outstandingMmapCount, replicas, failedLoads, evictable, evictableMmapped); } finally { lock.unlock(); } } 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; }