@SuppressWarnings({"ForLoopReplaceableByForEach","unchecked"}) private Map<K,EntryProcessorResult> updateWithBatch(GridCacheOperation op,Collection<? extends K> keys,@Nullable Iterable<?> vals,@Nullable Object[] invokeArgs,@Nullable ExpiryPolicy expiryPlc,GridCacheVersion ver,@Nullable CacheEntryPredicate[] filter,boolean keepBinary,UUID subjId,String taskName) throws IgniteCheckedException {
  List<GridCacheEntryEx> locked=lockEntries(keys);
  try {
    int size=locked.size();
    Map<Object,Object> putMap=null;
    Collection<Object> rmvKeys=null;
    List<CacheObject> writeVals=null;
    Map<K,EntryProcessorResult> invokeResMap=op == TRANSFORM ? U.<K,EntryProcessorResult>newHashMap(size) : null;
    List<GridCacheEntryEx> filtered=new ArrayList<>(size);
    CachePartialUpdateCheckedException err=null;
    Iterator<?> valsIter=vals != null ? vals.iterator() : null;
    boolean intercept=ctx.config().getInterceptor() != null;
    for (int i=0; i < size; i++) {
      GridCacheEntryEx entry=locked.get(i);
      Object val=valsIter != null ? valsIter.next() : null;
      if (val == null && op != DELETE)       throw new NullPointerException("Null value.");
      try {
        try {
          if (!ctx.isAllLocked(entry,filter)) {
            if (log.isDebugEnabled())             log.debug("Entry did not pass the filter (will skip write) [entry=" + entry + ", filter="+ Arrays.toString(filter)+ ']');
            continue;
          }
        }
 catch (        IgniteCheckedException e) {
          if (err == null)           err=partialUpdateException();
          err.add(F.asList(entry.key()),e);
          continue;
        }
        if (op == TRANSFORM) {
          ctx.kernalContext().resource().inject(val,GridResourceIoc.AnnotationSet.ENTRY_PROCESSOR,ctx.name());
          EntryProcessor<Object,Object,Object> entryProcessor=(EntryProcessor<Object,Object,Object>)val;
          CacheObject old=entry.innerGet(null,null,true,true,true,true,true,subjId,entryProcessor,taskName,null,keepBinary);
          Object oldVal=null;
          CacheInvokeEntry<Object,Object> invokeEntry=new CacheInvokeEntry<>(entry.key(),old,entry.version(),keepBinary,entry);
          CacheObject updated;
          Object updatedVal=null;
          CacheInvokeResult invokeRes=null;
          try {
            Object computed=entryProcessor.process(invokeEntry,invokeArgs);
            updatedVal=ctx.unwrapTemporary(invokeEntry.getValue());
            updated=ctx.toCacheObject(updatedVal);
            if (computed != null)             invokeRes=CacheInvokeResult.fromResult(ctx.unwrapTemporary(computed));
          }
 catch (          Exception e) {
            invokeRes=CacheInvokeResult.fromError(e);
            updated=old;
          }
          if (invokeRes != null)           invokeResMap.put((K)entry.key().value(ctx.cacheObjectContext(),false),invokeRes);
          if (updated == null) {
            if (intercept) {
              IgniteBiTuple<Boolean,?> interceptorRes=ctx.config().getInterceptor().onBeforeRemove(new CacheLazyEntry(ctx,entry.key(),invokeEntry.key(),old,oldVal,keepBinary));
              if (ctx.cancelRemove(interceptorRes))               continue;
            }
            if (putMap != null) {
              err=updatePartialBatch(filtered,ver,writeVals,putMap,null,expiryPlc,keepBinary,err,subjId,taskName);
              putMap=null;
              writeVals=null;
              filtered=new ArrayList<>();
            }
            if (rmvKeys == null)             rmvKeys=new ArrayList<>(size);
            rmvKeys.add(entry.key().value(ctx.cacheObjectContext(),false));
          }
 else {
            if (intercept) {
              Object interceptorVal=ctx.config().getInterceptor().onBeforePut(new CacheLazyEntry(ctx,entry.key(),invokeEntry.getKey(),old,oldVal,keepBinary),updatedVal);
              if (interceptorVal == null)               continue;
              updated=ctx.toCacheObject(ctx.unwrapTemporary(interceptorVal));
            }
            if (rmvKeys != null) {
              err=updatePartialBatch(filtered,ver,null,null,rmvKeys,expiryPlc,keepBinary,err,subjId,taskName);
              rmvKeys=null;
              filtered=new ArrayList<>();
            }
            if (putMap == null) {
              putMap=new LinkedHashMap<>(size,1.0f);
              writeVals=new ArrayList<>(size);
            }
            putMap.put(CU.value(entry.key(),ctx,false),CU.value(updated,ctx,false));
            writeVals.add(updated);
          }
        }
 else         if (op == UPDATE) {
          CacheObject cacheVal=ctx.toCacheObject(val);
          if (intercept) {
            CacheObject old=entry.innerGet(null,null,true,ctx.loadPreviousValue(),true,true,true,subjId,null,taskName,null,keepBinary);
            Object interceptorVal=ctx.config().getInterceptor().onBeforePut(new CacheLazyEntry(ctx,entry.key(),old,keepBinary),val);
            if (interceptorVal == null)             continue;
            cacheVal=ctx.toCacheObject(ctx.unwrapTemporary(interceptorVal));
          }
          if (putMap == null) {
            putMap=new LinkedHashMap<>(size,1.0f);
            writeVals=new ArrayList<>(size);
          }
          putMap.put(CU.value(entry.key(),ctx,false),CU.value(cacheVal,ctx,false));
          writeVals.add(cacheVal);
        }
 else {
          assert op == DELETE;
          if (intercept) {
            CacheObject old=entry.innerGet(null,null,true,ctx.loadPreviousValue(),true,true,true,subjId,null,taskName,null,keepBinary);
            IgniteBiTuple<Boolean,?> interceptorRes=ctx.config().getInterceptor().onBeforeRemove(new CacheLazyEntry(ctx,entry.key(),old,keepBinary));
            if (ctx.cancelRemove(interceptorRes))             continue;
          }
          if (rmvKeys == null)           rmvKeys=new ArrayList<>(size);
          rmvKeys.add(entry.key().value(ctx.cacheObjectContext(),false));
        }
        filtered.add(entry);
      }
 catch (      IgniteCheckedException e) {
        if (err == null)         err=partialUpdateException();
        err.add(F.asList(entry.key()),e);
      }
catch (      GridCacheEntryRemovedException ignore) {
        assert false : "Entry cannot become obsolete while holding lock.";
      }
    }
    if (putMap != null || rmvKeys != null) {
      err=updatePartialBatch(filtered,ver,writeVals,putMap,rmvKeys,expiryPlc,keepBinary,err,subjId,taskName);
    }
 else     assert filtered.isEmpty();
    if (err != null)     throw err;
    return invokeResMap;
  }
  finally {
    unlockEntries(locked);
  }
}

@ SuppressWarnings ( { "ForLoopReplaceableByForEach" , "unchecked" } ) private Map < K , EntryProcessorResult > updateWithBatch ( GridCacheOperation op , Collection < ? extends K > keys , @ Nullable Iterable < ? > vals , @ Nullable Object [ ] invokeArgs , @ Nullable ExpiryPolicy expiryPlc , GridCacheVersion ver , @ Nullable CacheEntryPredicate [ ] filter , boolean keepBinary , UUID subjId , String taskName ) throws IgniteCheckedException { List < GridCacheEntryEx > locked = lockEntries ( keys ) ; try { int size = locked . size ( ) ; Map < Object , Object > putMap = null ; Collection < Object > rmvKeys = null ; List < CacheObject > writeVals = null ; Map < K , EntryProcessorResult > invokeResMap = op == TRANSFORM ? U . < K , EntryProcessorResult > newHashMap ( size ) : null ; List < GridCacheEntryEx > filtered = new ArrayList < > ( size ) ; CachePartialUpdateCheckedException err = null ; Iterator < ? > valsIter = vals != null ? vals . iterator ( ) : null ; boolean intercept = ctx . config ( ) . getInterceptor ( ) != null ; for ( int i = 0 ; i < size ; i ++ ) { GridCacheEntryEx entry = locked . get ( i ) ; Object val = valsIter != null ? valsIter . next ( ) : null ; if ( val == null && op != DELETE ) throw new NullPointerException ( "Null value." ) ; try { try { if ( ! ctx . isAllLocked ( entry , filter ) ) { if ( log . isDebugEnabled ( ) ) log . debug ( "Entry did not pass the filter (will skip write) [entry=" + entry + ", filter=" + Arrays . toString ( filter ) + ']' ) ; continue ; } } catch ( IgniteCheckedException e ) { if ( err == null ) err = partialUpdateException ( ) ; err . add ( F . asList ( entry . key ( ) ) , e ) ; continue ; } if ( op == TRANSFORM ) { ctx . kernalContext ( ) . resource ( ) . inject ( val , GridResourceIoc . AnnotationSet . ENTRY_PROCESSOR , ctx . name ( ) ) ; EntryProcessor < Object , Object , Object > entryProcessor = ( EntryProcessor < Object , Object , Object > ) val ; CacheObject old = entry . innerGet ( null , null , true , true , true , true , true , subjId , entryProcessor , taskName , null , keepBinary ) ; Object oldVal = null ; CacheInvokeEntry < Object , Object > invokeEntry = new CacheInvokeEntry < > ( entry . key ( ) , old , entry . version ( ) , keepBinary , entry ) ; CacheObject updated ; Object updatedVal = null ; CacheInvokeResult invokeRes = null ; try { Object computed = entryProcessor . process ( invokeEntry , invokeArgs ) ; updatedVal = ctx . unwrapTemporary ( invokeEntry . getValue ( ) ) ; updated = ctx . toCacheObject ( updatedVal ) ; if ( computed != null ) invokeRes = CacheInvokeResult . fromResult ( ctx . unwrapTemporary ( computed ) ) ; } catch ( Exception e ) { invokeRes = CacheInvokeResult . fromError ( e ) ; updated = old ; } if ( invokeRes != null ) invokeResMap . put ( ( K ) entry . key ( ) . value ( ctx . cacheObjectContext ( ) , false ) , invokeRes ) ; if ( updated == null ) { if ( intercept ) { IgniteBiTuple < Boolean , ? > interceptorRes = ctx . config ( ) . getInterceptor ( ) . onBeforeRemove ( new CacheLazyEntry ( ctx , entry . key ( ) , invokeEntry . key ( ) , old , oldVal , keepBinary ) ) ; if ( ctx . cancelRemove ( interceptorRes ) ) continue ; } if ( putMap != null ) { err = updatePartialBatch ( filtered , ver , writeVals , putMap , null , expiryPlc , keepBinary , err , subjId , taskName ) ; putMap = null ; writeVals = null ; filtered = new ArrayList < > ( ) ; } if ( rmvKeys == null ) rmvKeys = new ArrayList < > ( size ) ; rmvKeys . add ( entry . key ( ) . value ( ctx . cacheObjectContext ( ) , false ) ) ; } else { if ( intercept ) { Object interceptorVal = ctx . config ( ) . getInterceptor ( ) . onBeforePut ( new CacheLazyEntry ( ctx , entry . key ( ) , invokeEntry . getKey ( ) , old , oldVal , keepBinary ) , updatedVal ) ; if ( interceptorVal == null ) continue ; updated = ctx . toCacheObject ( ctx . unwrapTemporary ( interceptorVal ) ) ; } if ( rmvKeys != null ) { err = updatePartialBatch ( filtered , ver , null , null , rmvKeys , expiryPlc , keepBinary , err , subjId , taskName ) ; rmvKeys = null ; filtered = new ArrayList < > ( ) ; } if ( putMap == null ) { putMap = new LinkedHashMap < > ( size , 1.0f ) ; writeVals = new ArrayList < > ( size ) ; } putMap . put ( CU . value ( entry . key ( ) , ctx , false ) , CU . value ( updated , ctx , false ) ) ; writeVals . add ( updated ) ; } } else if ( op == UPDATE ) { CacheObject cacheVal = ctx . toCacheObject ( val ) ; if ( intercept ) { CacheObject old = entry . innerGet ( null , null , true , ctx . loadPreviousValue ( ) , true , true , true , subjId , null , taskName , null , keepBinary ) ; Object interceptorVal = ctx . config ( ) . getInterceptor ( ) . onBeforePut ( new CacheLazyEntry ( ctx , entry . key ( ) , old , keepBinary ) , val ) ; if ( interceptorVal == null ) continue ; cacheVal = ctx . toCacheObject ( ctx . unwrapTemporary ( interceptorVal ) ) ; } if ( putMap == null ) { putMap = new LinkedHashMap < > ( size , 1.0f ) ; writeVals = new ArrayList < > ( size ) ; } putMap . put ( CU . value ( entry . key ( ) , ctx , false ) , CU . value ( cacheVal , ctx , false ) ) ; writeVals . add ( cacheVal ) ; } else { assert op == DELETE ; if ( intercept ) { CacheObject old = entry . innerGet ( null , null , true , ctx . loadPreviousValue ( ) , true , true , true , subjId , null , taskName , null , keepBinary ) ; IgniteBiTuple < Boolean , ? > interceptorRes = ctx . config ( ) . getInterceptor ( ) . onBeforeRemove ( new CacheLazyEntry ( ctx , entry . key ( ) , old , keepBinary ) ) ; if ( ctx . cancelRemove ( interceptorRes ) ) continue ; } if ( rmvKeys == null ) rmvKeys = new ArrayList < > ( size ) ; rmvKeys . add ( entry . key ( ) . value ( ctx . cacheObjectContext ( ) , false ) ) ; } filtered . add ( entry ) ; } catch ( IgniteCheckedException e ) { if ( err == null ) err = partialUpdateException ( ) ; err . add ( F . asList ( entry . key ( ) ) , e ) ; } catch ( GridCacheEntryRemovedException ignore ) { assert false : "Entry cannot become obsolete while holding lock." ; } } if ( putMap != null || rmvKeys != null ) { err = updatePartialBatch ( filtered , ver , writeVals , putMap , rmvKeys , expiryPlc , keepBinary , err , subjId , taskName ) ; } else assert filtered . isEmpty ( ) ; if ( err != null ) throw err ; return invokeResMap ; } finally { unlockEntries ( locked ) ; } }
updates entries using batched write - through .
