private void triggerAsyncCaching ( BlockInStream stream ) throws IOException { boolean cache = ReadType . fromProto ( mOptions . getOptions ( ) . getReadType ( ) ) . isCache ( ) ; boolean overReplicated = mStatus . getReplicationMax ( ) > 0 && mStatus . getFileBlockInfos ( ) . get ( ( int ) ( getPos ( ) / mBlockSize ) ) . getBlockInfo ( ) . getLocations ( ) . size ( ) >= mStatus . getReplicationMax ( ) ; cache = cache && ! overReplicated ; // Get relevant information from the stream. WorkerNetAddress dataSource = stream . getAddress ( ) ; long blockId = stream . getId ( ) ; if ( cache && ( mLastBlockIdCached != blockId ) ) { WorkerNetAddress worker ; if ( mPassiveCachingEnabled && mContext . hasLocalWorker ( ) ) { // send request to local worker worker = mContext . getLocalWorker ( ) ; } else { // send request to data source worker = dataSource ; } try { // Construct the async cache request long blockLength = mOptions . getBlockInfo ( blockId ) . getLength ( ) ; AsyncCacheRequest request = AsyncCacheRequest . newBuilder ( ) . setBlockId ( blockId ) . setLength ( blockLength ) . setOpenUfsBlockOptions ( mOptions . getOpenUfsBlockOptions ( blockId ) ) . setSourceHost ( dataSource . getHost ( ) ) . setSourcePort ( dataSource . getDataPort ( ) ) . build ( ) ; BlockWorkerClient blockWorker = mContext . acquireBlockWorkerClient ( worker ) ; try { blockWorker . asyncCache ( request ) ; mLastBlockIdCached = blockId ; } finally { mContext . releaseBlockWorkerClient ( worker , blockWorker ) ; } } catch ( Exception e ) { LOG . warn ( "Failed to complete async cache request for block {} at worker {}: {}" , blockId , worker , e . getMessage ( ) ) ; } } }
Send an async cache request to a worker based on read type and passive cache options .