private void startNextDownloaders ( ) throws SnowflakeSQLException { long waitingTime = BASE_WAITING_MS ; // submit the chunks to be downloaded up to the prefetch slot capacity // and limited by memory while ( nextChunkToDownload - nextChunkToConsume < prefetchSlots && nextChunkToDownload < chunks . size ( ) ) { // check if memory limit allows more prefetching final SnowflakeResultChunk nextChunk = chunks . get ( nextChunkToDownload ) ; final long neededChunkMemory = nextChunk . computeNeededChunkMemory ( ) ; // each time only one thread can enter this block synchronized ( currentMemoryUsage ) { // make sure memoryLimit > neededChunkMemory; otherwise, the thread hangs if ( neededChunkMemory > memoryLimit ) { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "{}: reset memoryLimit from {} MB to current chunk size {} MB" , Thread . currentThread ( ) . getName ( ) , memoryLimit / 1024 / 1024 , neededChunkMemory / 1024 / 1024 ) ; } memoryLimit = neededChunkMemory ; } // no memory allocate when memory is not enough for prefetch if ( currentMemoryUsage + neededChunkMemory > memoryLimit && nextChunkToDownload - nextChunkToConsume > 0 ) { break ; } // only allocate memory when the future usage is less than the limit if ( currentMemoryUsage + neededChunkMemory <= memoryLimit ) { nextChunk . tryReuse ( chunkDataCache ) ; currentMemoryUsage += neededChunkMemory ; if ( logger . isDebugEnabled ( ) ) { logger . debug ( "{}: currentMemoryUsage in MB: {}, nextChunkToDownload: {}, nextChunkToConsume: {}, " + "newReservedMemory in B: {} " , Thread . currentThread ( ) . getName ( ) , currentMemoryUsage / 1024 / 1024 , nextChunkToDownload , nextChunkToConsume , neededChunkMemory ) ; } logger . debug ( "submit chunk #{} for downloading, url={}" , this . nextChunkToDownload , nextChunk . getUrl ( ) ) ; executor . submit ( getDownloadChunkCallable ( this , nextChunk , qrmk , nextChunkToDownload , chunkHeadersMap , networkTimeoutInMilli , useJsonParserV2 ) ) ; // increment next chunk to download nextChunkToDownload ++ ; // make sure reset waiting time waitingTime = BASE_WAITING_MS ; // go to next chunk continue ; } } // waiting when nextChunkToDownload is equal to nextChunkToConsume but reach memory limit try { waitingTime *= WAITING_SECS_MULTIPLIER ; waitingTime = waitingTime > MAX_WAITING_MS ? MAX_WAITING_MS : waitingTime ; long jitter = ThreadLocalRandom . current ( ) . nextLong ( 0 , waitingTime / WAITING_JITTER_RATIO ) ; waitingTime += jitter ; if ( logger . isDebugEnabled ( ) ) { logger . debug ( "{} waiting for {}s: currentMemoryUsage in MB: {}, neededChunkMemory in MB: {}, " + "nextChunkToDownload: {}, nextChunkToConsume: {} " , Thread . currentThread ( ) . getName ( ) , waitingTime / 1000.0 , currentMemoryUsage / 1024 / 1024 , neededChunkMemory / 1024 / 1024 , nextChunkToDownload , nextChunkToConsume ) ; } Thread . sleep ( waitingTime ) ; } catch ( InterruptedException ie ) { throw new SnowflakeSQLException ( SqlState . INTERNAL_ERROR , ErrorCode . INTERNAL_ERROR . getMessageCode ( ) , "Waiting SnowflakeChunkDownloader has been interrupted." ) ; } } // clear the cache, we can't download more at the moment // so we won't need them in the near future chunkDataCache . clear ( ) ; }
Submit download chunk tasks to executor . Number depends on thread and memory limit