public void submitRequest ( AsyncCacheRequest request ) { ASYNC_CACHE_REQUESTS . inc ( ) ; long blockId = request . getBlockId ( ) ; long blockLength = request . getLength ( ) ; if ( mPendingRequests . putIfAbsent ( blockId , request ) != null ) { // This block is already planned. ASYNC_CACHE_DUPLICATE_REQUESTS . inc ( ) ; return ; } try { mAsyncCacheExecutor . submit ( ( ) -> { boolean result = false ; try { // Check if the block has already been cached on this worker long lockId = mBlockWorker . lockBlockNoException ( Sessions . ASYNC_CACHE_SESSION_ID , blockId ) ; if ( lockId != BlockLockManager . INVALID_LOCK_ID ) { try { mBlockWorker . unlockBlock ( lockId ) ; } catch ( BlockDoesNotExistException e ) { LOG . error ( "Failed to unlock block on async caching. We should never reach here" , e ) ; } ASYNC_CACHE_DUPLICATE_REQUESTS . inc ( ) ; return ; } Protocol . OpenUfsBlockOptions openUfsBlockOptions = request . getOpenUfsBlockOptions ( ) ; boolean isSourceLocal = mLocalWorkerHostname . equals ( request . getSourceHost ( ) ) ; // Depends on the request, cache the target block from different sources if ( isSourceLocal ) { ASYNC_CACHE_UFS_BLOCKS . inc ( ) ; result = cacheBlockFromUfs ( blockId , blockLength , openUfsBlockOptions ) ; } else { ASYNC_CACHE_REMOTE_BLOCKS . inc ( ) ; InetSocketAddress sourceAddress = new InetSocketAddress ( request . getSourceHost ( ) , request . getSourcePort ( ) ) ; result = cacheBlockFromRemoteWorker ( blockId , blockLength , sourceAddress , openUfsBlockOptions ) ; } LOG . debug ( "Result of async caching block {}: {}" , blockId , result ) ; } catch ( Exception e ) { LOG . warn ( "Async cache request failed.\n{}\nError: {}" , request , e ) ; } finally { if ( result ) { ASYNC_CACHE_SUCCEEDED_BLOCKS . inc ( ) ; } else { ASYNC_CACHE_FAILED_BLOCKS . inc ( ) ; } mPendingRequests . remove ( blockId ) ; } } ) ; } catch ( Exception e ) { // RuntimeExceptions (e.g. RejectedExecutionException) may be thrown in extreme cases when the // gRPC thread pool is drained due to highly concurrent caching workloads. In these cases, // return as async caching is at best effort. LOG . warn ( "Failed to submit async cache request.\n{}\nError: {}" , request , e ) ; ASYNC_CACHE_FAILED_BLOCKS . inc ( ) ; mPendingRequests . remove ( blockId ) ; } }
Handles a request to cache a block asynchronously . This is a non - blocking call .