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);
  }
}

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 ) ; } }
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 .
