@Test public void testShortCircuitReadFromClientWithoutShm() throws Exception { TemporarySocketDirectory sockDir = new TemporarySocketDirectory(); Configuration clientConf = createShortCircuitConf( "testShortCircuitReadWithoutShm", sockDir); Configuration serverConf = new Configuration(clientConf); DFSInputStream.tcpReadsDisabledForTesting = true; final MiniDFSCluster cluster = new MiniDFSCluster.Builder(serverConf).numDataNodes(1).build(); cluster.waitActive(); clientConf.setInt( DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS, 0); clientConf.set(DFS_CLIENT_CONTEXT, "testShortCircuitReadFromClientWithoutShm_clientContext"); final DistributedFileSystem fs = (DistributedFileSystem)FileSystem.get(cluster.getURI(0), clientConf); final String TEST_FILE = "/test_file"; final int TEST_FILE_LEN = 4000; final int SEED = 0xFADEC; DFSTestUtil.createFile(fs, new Path(TEST_FILE), TEST_FILE_LEN, (short)1, SEED); byte contents[] = DFSTestUtil.readFileBuffer(fs, new Path(TEST_FILE)); byte expected[] = DFSTestUtil. calculateFileContentsFromSeed(SEED, TEST_FILE_LEN); Assert.assertTrue(Arrays.equals(contents, expected)); final ShortCircuitCache cache = fs.dfs.getClientContext().getShortCircuitCache(); Assert.assertEquals(null, cache.getDfsClientShmManager()); cluster.shutdown(); }
public BlockReader build() throws IOException { BlockReader reader = null; Preconditions.checkNotNull(configuration); if (conf.shortCircuitLocalReads && allowShortCircuitLocalReads) { if (clientContext.getUseLegacyBlockReaderLocal()) { reader = getLegacyBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new legacy block reader local."); } return reader; } } else { reader = getBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new block reader local."); } return reader; } } } if (conf.domainSocketDataTraffic) { reader = getRemoteBlockReaderFromDomain(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new remote block reader using " + "UNIX domain socket on " + pathInfo.getPath()); } return reader; } } Preconditions.checkState(!DFSInputStream.tcpReadsDisabledForTesting, "TCP reads were disabled for testing, but we failed to " + "do a non-TCP read."); return getRemoteBlockReaderFromTcp(); }
BlockReaderFactory implements ShortCircuitReplicaCreator { public BlockReader build() throws IOException { BlockReader reader = null; Preconditions.checkNotNull(configuration); if (conf.shortCircuitLocalReads && allowShortCircuitLocalReads) { if (clientContext.getUseLegacyBlockReaderLocal()) { reader = getLegacyBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new legacy block reader local."); } return reader; } } else { reader = getBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new block reader local."); } return reader; } } } if (conf.domainSocketDataTraffic) { reader = getRemoteBlockReaderFromDomain(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new remote block reader using " + "UNIX domain socket on " + pathInfo.getPath()); } return reader; } } Preconditions.checkState(!DFSInputStream.tcpReadsDisabledForTesting, "TCP reads were disabled for testing, but we failed to " + "do a non-TCP read."); return getRemoteBlockReaderFromTcp(); } }
BlockReaderFactory implements ShortCircuitReplicaCreator { public BlockReader build() throws IOException { BlockReader reader = null; Preconditions.checkNotNull(configuration); if (conf.shortCircuitLocalReads && allowShortCircuitLocalReads) { if (clientContext.getUseLegacyBlockReaderLocal()) { reader = getLegacyBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new legacy block reader local."); } return reader; } } else { reader = getBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new block reader local."); } return reader; } } } if (conf.domainSocketDataTraffic) { reader = getRemoteBlockReaderFromDomain(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new remote block reader using " + "UNIX domain socket on " + pathInfo.getPath()); } return reader; } } Preconditions.checkState(!DFSInputStream.tcpReadsDisabledForTesting, "TCP reads were disabled for testing, but we failed to " + "do a non-TCP read."); return getRemoteBlockReaderFromTcp(); } BlockReaderFactory(DFSClient.Conf conf); }
BlockReaderFactory implements ShortCircuitReplicaCreator { public BlockReader build() throws IOException { BlockReader reader = null; Preconditions.checkNotNull(configuration); if (conf.shortCircuitLocalReads && allowShortCircuitLocalReads) { if (clientContext.getUseLegacyBlockReaderLocal()) { reader = getLegacyBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new legacy block reader local."); } return reader; } } else { reader = getBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new block reader local."); } return reader; } } } if (conf.domainSocketDataTraffic) { reader = getRemoteBlockReaderFromDomain(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new remote block reader using " + "UNIX domain socket on " + pathInfo.getPath()); } return reader; } } Preconditions.checkState(!DFSInputStream.tcpReadsDisabledForTesting, "TCP reads were disabled for testing, but we failed to " + "do a non-TCP read."); return getRemoteBlockReaderFromTcp(); } BlockReaderFactory(DFSClient.Conf conf); BlockReaderFactory setFileName(String fileName); BlockReaderFactory setBlock(ExtendedBlock block); BlockReaderFactory setBlockToken(Token<BlockTokenIdentifier> token); BlockReaderFactory setStartOffset(long startOffset); BlockReaderFactory setVerifyChecksum(boolean verifyChecksum); BlockReaderFactory setClientName(String clientName); BlockReaderFactory setDatanodeInfo(DatanodeInfo datanode); BlockReaderFactory setStorageType(StorageType storageType); BlockReaderFactory setAllowShortCircuitLocalReads(
      boolean allowShortCircuitLocalReads); BlockReaderFactory setClientCacheContext(
      ClientContext clientContext); BlockReaderFactory setLength(long length); BlockReaderFactory setCachingStrategy(
      CachingStrategy cachingStrategy); BlockReaderFactory setInetSocketAddress(
      InetSocketAddress inetSocketAddress); BlockReaderFactory setUserGroupInformation(
      UserGroupInformation userGroupInformation); BlockReaderFactory setRemotePeerFactory(
      RemotePeerFactory remotePeerFactory); BlockReaderFactory setConfiguration(
      Configuration configuration); BlockReader build(); @Override ShortCircuitReplicaInfo createShortCircuitReplicaInfo(); @Override String toString(); static String getFileName(final InetSocketAddress s,
      final String poolId, final long blockId); }
BlockReaderFactory implements ShortCircuitReplicaCreator { public BlockReader build() throws IOException { BlockReader reader = null; Preconditions.checkNotNull(configuration); if (conf.shortCircuitLocalReads && allowShortCircuitLocalReads) { if (clientContext.getUseLegacyBlockReaderLocal()) { reader = getLegacyBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new legacy block reader local."); } return reader; } } else { reader = getBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new block reader local."); } return reader; } } } if (conf.domainSocketDataTraffic) { reader = getRemoteBlockReaderFromDomain(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new remote block reader using " + "UNIX domain socket on " + pathInfo.getPath()); } return reader; } } Preconditions.checkState(!DFSInputStream.tcpReadsDisabledForTesting, "TCP reads were disabled for testing, but we failed to " + "do a non-TCP read."); return getRemoteBlockReaderFromTcp(); } BlockReaderFactory(DFSClient.Conf conf); BlockReaderFactory setFileName(String fileName); BlockReaderFactory setBlock(ExtendedBlock block); BlockReaderFactory setBlockToken(Token<BlockTokenIdentifier> token); BlockReaderFactory setStartOffset(long startOffset); BlockReaderFactory setVerifyChecksum(boolean verifyChecksum); BlockReaderFactory setClientName(String clientName); BlockReaderFactory setDatanodeInfo(DatanodeInfo datanode); BlockReaderFactory setStorageType(StorageType storageType); BlockReaderFactory setAllowShortCircuitLocalReads(
      boolean allowShortCircuitLocalReads); BlockReaderFactory setClientCacheContext(
      ClientContext clientContext); BlockReaderFactory setLength(long length); BlockReaderFactory setCachingStrategy(
      CachingStrategy cachingStrategy); BlockReaderFactory setInetSocketAddress(
      InetSocketAddress inetSocketAddress); BlockReaderFactory setUserGroupInformation(
      UserGroupInformation userGroupInformation); BlockReaderFactory setRemotePeerFactory(
      RemotePeerFactory remotePeerFactory); BlockReaderFactory setConfiguration(
      Configuration configuration); BlockReader build(); @Override ShortCircuitReplicaInfo createShortCircuitReplicaInfo(); @Override String toString(); static String getFileName(final InetSocketAddress s,
      final String poolId, final long blockId);  }