question: private void map ( Iterable < KeyCacheObject > keys , boolean remap , boolean topLocked ) { try { AffinityTopologyVersion topVer = this . topVer ; assert topVer != null ; assert topVer . topologyVersion ( ) > 0 : topVer ; if ( CU . affinityNodes ( cctx , topVer ) . isEmpty ( ) ) { onDone ( new ClusterTopologyServerNotFoundException ( "Failed to map keys for near-only cache (all " + "partition nodes left the grid)." ) ) ; return ; } boolean clientNode = cctx . kernalContext ( ) . clientNode ( ) ; assert ! remap || ( clientNode && ( tx == null || ! tx . hasRemoteLocks ( ) ) ) ; synchronized ( this ) { mappings = new ArrayDeque < > ( ) ; GridNearLockMapping map = null ; for ( KeyCacheObject key : keys ) { GridNearLockMapping updated = map ( key , map , topVer ) ; if ( updated != map ) { mappings . add ( updated ) ; if ( tx != null && updated . node ( ) . isLocal ( ) ) tx . nearLocallyMapped ( true ) ; } map = updated ; } if ( isDone ( ) ) { if ( log . isDebugEnabled ( ) ) log . debug ( "Abandoning (re)map because future is done: " + this ) ; return ; } if ( log . isDebugEnabled ( ) ) log . debug ( "Starting (re)map for mappings [mappings=" + mappings + ", fut=" + this + ']' ) ; boolean first = true ; for ( Iterator < GridNearLockMapping > iter = mappings . iterator ( ) ; iter . hasNext ( ) ; ) { GridNearLockMapping mapping = iter . next ( ) ; ClusterNode node = mapping . node ( ) ; Collection < KeyCacheObject > mappedKeys = mapping . mappedKeys ( ) ; assert ! mappedKeys . isEmpty ( ) ; GridNearLockRequest req = null ; Collection < KeyCacheObject > distributedKeys = new ArrayList < > ( mappedKeys . size ( ) ) ; boolean explicit = false ; for ( KeyCacheObject key : mappedKeys ) { IgniteTxKey txKey = cctx . txKey ( key ) ; while ( true ) { GridNearCacheEntry entry = null ; try { entry = cctx . near ( ) . entryExx ( key , topVer ) ; if ( ! cctx . isAll ( entry , filter ) ) { if ( log . isDebugEnabled ( ) ) log . debug ( "Entry being locked did not pass filter (will not lock): " + entry ) ; onComplete ( false , false ) ; return ; } GridCacheMvccCandidate cand = addEntry ( topVer , entry , node . id ( ) ) ; if ( isDone ( ) ) { if ( log . isDebugEnabled ( ) ) log . debug ( "Abandoning (re)map because future is done after addEntry attempt " + "[fut=" + this + ", entry=" + entry + ']' ) ; return ; } if ( cand != null ) { if ( tx == null && ! cand . reentry ( ) ) cctx . mvcc ( ) . addExplicitLock ( threadId , cand , topVer ) ; IgniteBiTuple < GridCacheVersion , CacheObject > val = entry . versionedValue ( ) ; if ( val == null ) { GridDhtCacheEntry dhtEntry = dht ( ) . peekExx ( key ) ; try { if ( dhtEntry != null ) val = dhtEntry . versionedValue ( topVer ) ; } catch ( GridCacheEntryRemovedException ignored ) { assert dhtEntry . obsolete ( ) : dhtEntry ; if ( log . isDebugEnabled ( ) ) log . debug ( "Got removed exception for DHT entry in map (will ignore): " + dhtEntry ) ; } } GridCacheVersion dhtVer = null ; if ( val != null ) { dhtVer = val . get1 ( ) ; valMap . put ( key , val ) ; } if ( ! cand . reentry ( ) ) { if ( req == null ) { boolean clientFirst = false ; if ( first ) { clientFirst = clientNode && ! topLocked && ( tx == null || ! tx . hasRemoteLocks ( ) ) ; first = false ; } req = new GridNearLockRequest ( cctx . cacheId ( ) , topVer , cctx . nodeId ( ) , threadId , futId , lockVer , inTx ( ) , implicitTx ( ) , implicitSingleTx ( ) , read , retval , isolation ( ) , isInvalidate ( ) , timeout , mappedKeys . size ( ) , inTx ( ) ? tx . size ( ) : mappedKeys . size ( ) , inTx ( ) && tx . syncMode ( ) == FULL_SYNC , inTx ( ) ? tx . subjectId ( ) : null , inTx ( ) ? tx . taskNameHash ( ) : 0 , read ? accessTtl : - 1L , skipStore , keepBinary , clientFirst , cctx . deploymentEnabled ( ) ) ; mapping . request ( req ) ; } distributedKeys . add ( key ) ; if ( tx != null ) tx . addKeyMapping ( txKey , mapping . node ( ) ) ; req . addKeyBytes ( key , retval && dhtVer == null , dhtVer , cctx ) ; } if ( cand . reentry ( ) ) explicit = tx != null && ! entry . hasLockCandidate ( tx . xidVersion ( ) ) ; } else explicit = tx != null && ! entry . hasLockCandidate ( tx . xidVersion ( ) ) ; if ( explicit ) tx . addKeyMapping ( txKey , mapping . node ( ) ) ; break ; } catch ( GridCacheEntryRemovedException ignored ) { assert entry . obsolete ( ) : "Got removed exception on non-obsolete entry: " + entry ; if ( log . isDebugEnabled ( ) ) log . debug ( "Got removed entry in lockAsync(..) method (will retry): " + entry ) ; } } if ( explicit ) { boolean marked = tx != null && tx . markExplicit ( node . id ( ) ) ; assert tx == null || marked ; } } if ( ! distributedKeys . isEmpty ( ) ) mapping . distributedKeys ( distributedKeys ) ; else { assert mapping . request ( ) == null ; iter . remove ( ) ; } } } cctx . mvcc ( ) . recheckPendingLocks ( ) ; proceedMapping ( ) ; } catch ( IgniteCheckedException ex ) { onError ( ex ) ; } }
target:  Maps keys to nodes. Note that we can not simply group keys by nodes and send lock request as such approach does not preserve order of lock acquisition. Instead, keys are split in continuous groups belonging to one primary node and locks for these groups are acquired sequentially.