[infinispan-issues] [JBoss JIRA] (ISPN-3938) AdvancedAsyncCacheLoader.process() concurrency issues

[William Burns (JIRA)]

    [ https://issues.jboss.org/browse/ISPN-3938?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=12939046#comment-12939046 ] 

William Burns commented on ISPN-3938:
-------------------------------------

Whether process method returns after it is iterated over the keys is defined by the cache loader implementation though, so an implementation can do whatever it wants there.  If you look at the references to allKeys which threw the ConcurrentModificationException it is only possible if the process method returned early.

{code}
      Set<Object> allKeys = new HashSet<Object>(batchSize);
      Set<Object> batch = new HashSet<Object>();
      loadAllKeys(state.get(), allKeys, keyFilter, executor);  <--- reference
      for (Iterator it = allKeys.iterator(); it.hasNext(); ) { <--- reference in for loop (doesn't pass reference to any other method invocations)
         batch.add(it.next());
         if (batch.size() == batchSize) {
            final Set<Object> toHandle = batch;
            batch = new HashSet<Object>(batchSize);
            submitProcessTask(cacheLoaderTask, eacs, taskContext, toHandle);
         }
      }
-- No more references to the HashSet past this.
{code}

The HashSet is only used in the loadAllKeys method and the iterator.  The iterator is guaranteed to not be used concurrently with the code shown.  The loadAllKeys however can with the process method.  Or is there somewhere else I am not seeing?
                
> AdvancedAsyncCacheLoader.process() concurrency issues
> -----------------------------------------------------
>
>                 Key: ISPN-3938
>                 URL: https://issues.jboss.org/browse/ISPN-3938
>             Project: Infinispan
>          Issue Type: Bug
>          Components: Core, Loaders and Stores
>    Affects Versions: 6.0.0.Final
>            Reporter: Dan Berindei
>            Assignee: Dan Berindei
>             Fix For: 7.0.0.Final
>
>
> {{AdvancedAsyncCacheLoader.process()}} calls {{advancedLoader().process()}} to collect all the keys in the store, but the HashSet used to collect the keys it not thread-safe. This can cause problems, e.g. during state transfer:
> {noformat}
> WARN  cheTopologyControlCommand | ISPN000071: Caught exception when handling command CacheTopologyControlCommand{cache=sessions, type=CH_UPDATE, sender=alfie-lt-46127, joinInfo=null, topologyId=3, currentCH=DefaultConsistentHash{numSegments=60, numOwners=1, members=[alfie-lt-46127]}, pendingCH=null, throwable=null, viewId=1}java.util.ConcurrentModificationException
>   at java.util.HashMap$HashIterator.nextEntry(HashMap.java:926)
>   at java.util.HashMap$KeyIterator.next(HashMap.java:960)
>   at org.infinispan.persistence.async.AdvancedAsyncCacheLoader.process(AdvancedAsyncCacheLoader.java:80)
>   at org.infinispan.persistence.manager.PersistenceManagerImpl.processOnAllStores(PersistenceManagerImpl.java:414)
>   at org.infinispan.statetransfer.StateConsumerImpl.invalidateSegments(StateConsumerImpl.java:910)
>   at org.infinispan.statetransfer.StateConsumerImpl.onTopologyUpdate(StateConsumerImpl.java:393)
>   at org.infinispan.statetransfer.StateTransferManagerImpl.doTopologyUpdate(StateTransferManagerImpl.java:178)
>   at org.infinispan.statetransfer.StateTransferManagerImpl.access$000(StateTransferManagerImpl.java:38)
>   at org.infinispan.statetransfer.StateTransferManagerImpl$1.updateConsistentHash(StateTransferManagerImpl.java:100)
>   at org.infinispan.topology.LocalTopologyManagerImpl.handleConsistentHashUpdate(LocalTopologyManagerImpl.java:191)
>   at org.infinispan.topology.CacheTopologyControlCommand.doPerform(CacheTopologyControlCommand.java:152)
>   at org.infinispan.topology.CacheTopologyControlCommand.perform(CacheTopologyControlCommand.java:124)
>   at org.infinispan.topology.ClusterTopologyManagerImpl$3.run(ClusterTopologyManagerImpl.java:606)
>   at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
>   at java.util.concurrent.FutureTask.run(FutureTask.java:262)
>   at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
>   at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
>   at java.lang.Thread.run(Thread.java:744)
> {noformat}

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira