RMICacheReplicatorTest xref


1   /**
2    *  Copyright 2003-2007 Greg Luck
3    *
4    *  Licensed under the Apache License, Version 2.0 (the "License");
5    *  you may not use this file except in compliance with the License.
6    *  You may obtain a copy of the License at
7    *
8    *      http://www.apache.org/licenses/LICENSE-2.0
9    *
10   *  Unless required by applicable law or agreed to in writing, software
11   *  distributed under the License is distributed on an "AS IS" BASIS,
12   *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13   *  See the License for the specific language governing permissions and
14   *  limitations under the License.
15   */
16  
17  package net.sf.ehcache.distribution;
18  
19  import junit.framework.AssertionFailedError;
20  import net.sf.ehcache.AbstractCacheTest;
21  import net.sf.ehcache.Cache;
22  import net.sf.ehcache.CacheException;
23  import net.sf.ehcache.CacheManager;
24  import net.sf.ehcache.Ehcache;
25  import net.sf.ehcache.Element;
26  import net.sf.ehcache.StopWatch;
27  import net.sf.ehcache.ThreadKiller;
28  import net.sf.ehcache.management.ManagementService;
29  import net.sf.ehcache.event.CountingCacheEventListener;
30  import org.apache.commons.logging.Log;
31  import org.apache.commons.logging.LogFactory;
32  
33  import java.io.IOException;
34  import java.io.Serializable;
35  import java.rmi.RemoteException;
36  import java.util.ArrayList;
37  import java.util.Arrays;
38  import java.util.Date;
39  import java.util.List;
40  import java.util.Random;
41  
42  /**
43   * Tests replication of Cache events
44   * <p/>
45   * Note these tests need a live network interface running in multicast mode to work
46   * <p/>
47   * If running involving RMIAsynchronousCacheReplicator individually the test will fail because
48   * the VM will gobble up the SoftReferences rather than allocating more memory. Uncomment the
49   * forceVMGrowth() method usage in setup.
50   *
51   * @author Greg Luck
52   * @version $Id: RMICacheReplicatorTest.java 512 2007-07-10 09:18:45Z gregluck $
53   * todo do one for Manual
54   */
55  public class RMICacheReplicatorTest extends AbstractCacheTest {
56  
57  
58      /**
59       * A value to represent replicate asynchronously
60       */
61      protected static final boolean ASYNCHRONOUS = true;
62  
63      /**
64       * A value to represent replicate synchronously
65       */
66      protected static final boolean SYNCHRONOUS = false;
67  
68      private static final Log LOG = LogFactory.getLog(RMICacheReplicatorTest.class.getName());
69  
70  
71      /**
72       * CacheManager 1 in the cluster
73       */
74      protected CacheManager manager1;
75      /**
76       * CacheManager 2 in the cluster
77       */
78      protected CacheManager manager2;
79      /**
80       * CacheManager 3 in the cluster
81       */
82      protected CacheManager manager3;
83      /**
84       * CacheManager 4 in the cluster
85       */
86      protected CacheManager manager4;
87      /**
88       * CacheManager 5 in the cluster
89       */
90      protected CacheManager manager5;
91      /**
92       * CacheManager 6 in the cluster
93       */
94      protected CacheManager manager6;
95  
96      /**
97       * The name of the cache under test
98       */
99      protected String cacheName = "sampleCache1";
100     /**
101      * CacheManager 1 of 2s cache being replicated
102      */
103     protected Ehcache cache1;
104 
105     /**
106      * CacheManager 2 of 2s cache being replicated
107      */
108     protected Ehcache cache2;
109 
110     /**
111      * Allows setup to be the same
112      */
113     protected String cacheNameBase = "ehcache-distributed";
114 
115     /**
116      * {@inheritDoc}
117      * Sets up two caches: cache1 is local. cache2 is to be receive updates
118      *
119      * @throws Exception
120      */
121     protected void setUp() throws Exception {
122         if (JVMUtil.isSingleRMIRegistryPerVM()) {
123             return;
124         }
125 
126         //Required to get SoftReference tests to pass. The VM clean up SoftReferences rather than allocating
127         // memory to -Xmx!
128         //forceVMGrowth();
129         //System.gc();
130         MulticastKeepaliveHeartbeatSender.setHeartBeatInterval(1000);
131 
132         CountingCacheEventListener.resetCounters();
133         manager1 = new CacheManager(AbstractCacheTest.TEST_CONFIG_DIR + "distribution/ehcache-distributed1.xml");
134         manager2 = new CacheManager(AbstractCacheTest.TEST_CONFIG_DIR + "distribution/ehcache-distributed2.xml");
135         manager3 = new CacheManager(AbstractCacheTest.TEST_CONFIG_DIR + "distribution/ehcache-distributed3.xml");
136         manager4 = new CacheManager(AbstractCacheTest.TEST_CONFIG_DIR + "distribution/ehcache-distributed4.xml");
137         manager5 = new CacheManager(AbstractCacheTest.TEST_CONFIG_DIR + "distribution/ehcache-distributed5.xml");
138 
139         //manager6 = new CacheManager(AbstractCacheTest.TEST_CONFIG_DIR + "distribution/ehcache-distributed-jndi6.xml");
140 
141         //allow cluster to be established
142         Thread.sleep(1020);
143 
144         cache1 = manager1.getCache(cacheName);
145         cache1.removeAll();
146 
147         cache2 = manager2.getCache(cacheName);
148         cache2.removeAll();
149 
150         //enable distributed removeAlls to finish
151         waitForProgagate();
152 
153 
154     }
155 
156     /**
157      * {@inheritDoc}
158      *
159      * @throws Exception
160      */
161     protected void tearDown() throws Exception {
162 
163         if (JVMUtil.isSingleRMIRegistryPerVM()) {
164             return;
165         }
166 
167         if (manager1 != null) {
168             manager1.shutdown();
169         }
170         if (manager2 != null) {
171             manager2.shutdown();
172         }
173         if (manager3 != null) {
174             manager3.shutdown();
175         }
176         if (manager4 != null) {
177             manager4.shutdown();
178         }
179         if (manager5 != null) {
180             manager5.shutdown();
181         }
182         if (manager6 != null) {
183             manager6.shutdown();
184         }
185         Thread.sleep(50);
186     }
187 
188     /**
189      * 5 cache managers should means that each cache has four remote peers
190      */
191     public void testRemoteCachePeersEqualsNumberOfCacheManagersInCluster() {
192 
193         if (JVMUtil.isSingleRMIRegistryPerVM()) {
194             return;
195         }
196 
197 
198         CacheManagerPeerProvider provider = manager1.getCachePeerProvider();
199         List remotePeersOfCache1 = provider.listRemoteCachePeers(cache1);
200         assertEquals(4, remotePeersOfCache1.size());
201     }
202 
203     /**
204      * Does a new cache manager in the cluster get detected?
205      */
206     public void testRemoteCachePeersDetectsNewCacheManager() throws InterruptedException {
207 
208         if (JVMUtil.isSingleRMIRegistryPerVM()) {
209             return;
210         }
211 
212         CacheManagerPeerProvider provider = manager1.getCachePeerProvider();
213         List remotePeersOfCache1 = provider.listRemoteCachePeers(cache1);
214         assertEquals(4, remotePeersOfCache1.size());
215 
216         //Add new CacheManager to cluster
217         manager6 = new CacheManager(AbstractCacheTest.TEST_CONFIG_DIR + "distribution/ehcache-distributed6.xml");
218 
219         //Allow detection to occur
220         Thread.sleep(10010);
221 
222         remotePeersOfCache1 = provider.listRemoteCachePeers(cache1);
223         assertEquals(5, remotePeersOfCache1.size());
224     }
225 
226     /**
227      * Does a down cache manager in the cluster get removed?
228      */
229     public void testRemoteCachePeersDetectsDownCacheManager() throws InterruptedException {
230 
231         if (JVMUtil.isSingleRMIRegistryPerVM()) {
232             return;
233         }
234 
235 
236         CacheManagerPeerProvider provider = manager1.getCachePeerProvider();
237         List remotePeersOfCache1 = provider.listRemoteCachePeers(cache1);
238         assertEquals(4, remotePeersOfCache1.size());
239 
240         //Drop a CacheManager from the cluster
241         manager5.shutdown();
242 
243         //Allow change detection to occur. Heartbeat 1 second and is not stale until 5000
244         Thread.sleep(11010);
245         remotePeersOfCache1 = provider.listRemoteCachePeers(cache1);
246 
247 
248         assertEquals(3, remotePeersOfCache1.size());
249     }
250 
251     /**
252      * Does a down cache manager in the cluster get removed?
253      */
254     public void testRemoteCachePeersDetectsDownCacheManagerSlow() throws InterruptedException {
255 
256         if (JVMUtil.isSingleRMIRegistryPerVM()) {
257             return;
258         }
259 
260         try {
261             CacheManagerPeerProvider provider = manager1.getCachePeerProvider();
262             List remotePeersOfCache1 = provider.listRemoteCachePeers(cache1);
263             assertEquals(4, remotePeersOfCache1.size());
264 
265             MulticastKeepaliveHeartbeatSender.setHeartBeatInterval(2000);
266             Thread.sleep(2000);
267 
268             //Drop a CacheManager from the cluster
269             manager5.shutdown();
270 
271             //Insufficient time for it to timeout
272             remotePeersOfCache1 = provider.listRemoteCachePeers(cache1);
273             assertEquals(4, remotePeersOfCache1.size());
274         } finally {
275             MulticastKeepaliveHeartbeatSender.setHeartBeatInterval(1000);
276             Thread.sleep(2000);
277         }
278 
279 
280     }
281 
282     /**
283      * Tests put and remove initiated from cache1 in a cluster
284      * <p/>
285      * This test goes into an infinite loop if the chain of notifications is not somehow broken.
286      */
287     public void testPutProgagatesFromAndToEveryCacheManagerAndCache() throws CacheException, InterruptedException {
288 
289         if (JVMUtil.isSingleRMIRegistryPerVM()) {
290             return;
291         }
292 
293         //Put
294         String[] cacheNames = manager1.getCacheNames();
295         int numberOfCaches = getNumberOfReplicatingCachesInCacheManager();
296         Arrays.sort(cacheNames);
297         for (int i = 0; i < cacheNames.length; i++) {
298             String name = cacheNames[i];
299             manager1.getCache(name).put(new Element("" + i, new Integer(i)));
300             //Add some non serializable elements that should not get propagated
301             manager1.getCache(name).put(new Element("nonSerializable" + i, new Object()));
302         }
303 
304         waitForProgagate();
305 
306         int count2 = 0;
307         int count3 = 0;
308         int count4 = 0;
309         int count5 = 0;
310         for (int i = 0; i < cacheNames.length; i++) {
311             String name = cacheNames[i];
312             Element element2 = manager2.getCache(name).get("" + i);
313             if (element2 != null) {
314                 count2++;
315             }
316             Element nonSerializableElement2 = manager2.getCache(name).get("nonSerializable" + i);
317             if (nonSerializableElement2 != null) {
318                 count2++;
319             }
320             Element element3 = manager3.getCache(name).get("" + i);
321             if (element3 != null) {
322                 count3++;
323             }
324             Element element4 = manager4.getCache(name).get("" + i);
325             if (element4 != null) {
326                 count4++;
327             }
328             Element element5 = manager5.getCache(name).get("" + i);
329             if (element5 != null) {
330                 count5++;
331             }
332         }
333         assertEquals(numberOfCaches, count2);
334         assertEquals(numberOfCaches, count3);
335         assertEquals(numberOfCaches, count4);
336         assertEquals(numberOfCaches, count5);
337 
338 
339     }
340 
341     /**
342      * Tests what happens when a CacheManager in the cluster comes and goes. In ehcache-1.2.4 this would cause the new RMI CachePeers in the CacheManager to
343      * be permanently corrupt.
344      */
345     public void testPutProgagatesFromAndToEveryCacheManagerAndCacheDirty() throws CacheException, InterruptedException {
346 
347         if (JVMUtil.isSingleRMIRegistryPerVM()) {
348             return;
349         }
350 
351 
352         manager3.shutdown();
353 
354         Thread.sleep(11010);
355 
356         manager3 = new CacheManager(AbstractCacheTest.TEST_CONFIG_DIR + "distribution/ehcache-distributed3.xml");
357         Thread.sleep(11010);
358 
359         //Put
360         String[] cacheNames = manager1.getCacheNames();
361         int numberOfCaches = getNumberOfReplicatingCachesInCacheManager();
362         Arrays.sort(cacheNames);
363         for (int i = 0; i < cacheNames.length; i++) {
364             String name = cacheNames[i];
365             manager1.getCache(name).put(new Element("" + i, new Integer(i)));
366             //Add some non serializable elements that should not get propagated
367             manager1.getCache(name).put(new Element("nonSerializable" + i, new Object()));
368         }
369 
370         waitForProgagate();
371 
372         int count2 = 0;
373         int count3 = 0;
374         int count4 = 0;
375         int count5 = 0;
376         for (int i = 0; i < cacheNames.length; i++) {
377             String name = cacheNames[i];
378             Element element2 = manager2.getCache(name).get("" + i);
379             if (element2 != null) {
380                 count2++;
381             }
382             Element nonSerializableElement2 = manager2.getCache(name).get("nonSerializable" + i);
383             if (nonSerializableElement2 != null) {
384                 count2++;
385             }
386             Element element3 = manager3.getCache(name).get("" + i);
387             if (element3 != null) {
388                 count3++;
389             }
390             Element element4 = manager4.getCache(name).get("" + i);
391             if (element4 != null) {
392                 count4++;
393             }
394             Element element5 = manager5.getCache(name).get("" + i);
395             if (element5 != null) {
396                 count5++;
397             }
398         }
399         assertEquals(numberOfCaches, count2);
400         assertEquals(numberOfCaches, count3);
401         assertEquals(numberOfCaches, count4);
402         assertEquals(numberOfCaches, count5);
403 
404 
405     }
406 
407     /**
408      * Enables long stabilty runs using replication to be done.
409      * <p/>
410      * This test has been run in a profile for 15 hours without any observed issues.
411      *
412      * @throws InterruptedException
413      */
414     public void manualStabilityTest() throws InterruptedException {
415         forceVMGrowth();
416 
417         ManagementService.registerMBeans(manager3, createMBeanServer(), true, true, true, true);
418         while (true) {
419             testBigPutsProgagatesAsynchronous();
420         }
421     }
422 
423     /**
424      * Non JUnit invocation of stability test to get cleaner run
425      *
426      * @param args
427      * @throws InterruptedException
428      */
429     public static void main(String[] args) throws Exception {
430         RMICacheReplicatorTest replicatorTest = new RMICacheReplicatorTest();
431         replicatorTest.setUp();
432         replicatorTest.manualStabilityTest();
433     }
434 
435     /**
436      * The number of caches there should be.
437      */
438     protected int getNumberOfReplicatingCachesInCacheManager() {
439         return 55;
440     }
441 
442 
443     /**
444      * Performance and capacity tests.
445      * <p/>
446      * The numbers given are for the remote peer tester (java -jar ehcache-1.x-remote-debugger.jar ehcache-distributed1.xml)
447      * running on a 10Mbit ethernet network and are measured from the time the peer starts receiving to when
448      * it has fully received.
449      * <p/>
450      * r37 and earlier - initial implementation
451      * 38 seconds to get all notifications with 6 peers, 2000 Elements and 400 byte payload
452      * 18 seconds to get all notifications with 2 peers, 2000 Elements and 400 byte payload
453      * 40 seconds to get all notifications with 2 peers, 2000 Elements and 10k payload
454      * 22 seconds to get all notifications with 2 peers, 2000 Elements and 1k payload
455      * 26 seconds to get all notifications with 2 peers, 200 Elements and 100k payload
456      * <p/>
457      * r38 - RMI stub lookup on registration rather than at each lookup. Saves quite a few lookups. Also change to 5 second heartbeat
458      * 38 seconds to get 2000 notifications with 6 peers, Elements with 400 byte payload (1 second heartbeat)
459      * 16 seconds to get 2000 notifications with 6 peers, Elements with 400 byte payload (5 second heartbeat)
460      * 13 seconds to get 2000 notifications with 2 peers, Elements with 400 byte payload
461      * <p/>
462      * r39 - Batching asyn replicator. Send all queued messages in one RMI call once per second.
463      * 2 seconds to get 2000 notifications with 6 peers, Elements with 400 byte payload (5 second heartbeat)
464      */
465     public void testBigPutsProgagatesAsynchronous() throws CacheException, InterruptedException {
466 
467         if (JVMUtil.isSingleRMIRegistryPerVM()) {
468             return;
469         }
470 
471         //Give everything a chance to startup
472         Thread.sleep(10000);
473         StopWatch stopWatch = new StopWatch();
474         Integer index = null;
475         for (int i = 0; i < 2; i++) {
476             for (int j = 0; j < 1000; j++) {
477                 index = new Integer(((1000 * i) + j));
478                 cache1.put(new Element(index,
479                         "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
480                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
481                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
482                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
483                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"));
484             }
485 
486         }
487         long elapsed = stopWatch.getElapsedTime();
488         long putTime = ((elapsed / 1000));
489         LOG.info("Put Elapsed time: " + putTime);
490         //assertTrue(putTime < 8);
491 
492         assertEquals(2000, cache1.getSize());
493 
494         Thread.sleep(5000);
495         assertEquals(2000, manager2.getCache("sampleCache1").getSize());
496         assertEquals(2000, manager3.getCache("sampleCache1").getSize());
497         assertEquals(2000, manager4.getCache("sampleCache1").getSize());
498         assertEquals(2000, manager5.getCache("sampleCache1").getSize());
499 
500         CountingCacheEventListener.resetCounters();
501 
502     }
503 
504 
505     /**
506      * Performance and capacity tests.
507      * <p/>
508      */
509     public void testBootstrap() throws CacheException, InterruptedException, RemoteException {
510 
511         if (JVMUtil.isSingleRMIRegistryPerVM()) {
512             return;
513         }
514 
515         //load up some data
516         StopWatch stopWatch = new StopWatch();
517         Integer index = null;
518         for (int i = 0; i < 2; i++) {
519             for (int j = 0; j < 1000; j++) {
520                 index = new Integer(((1000 * i) + j));
521                 cache1.put(new Element(index,
522                         "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
523                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
524                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
525                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
526                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"));
527             }
528 
529         }
530         long elapsed = stopWatch.getElapsedTime();
531         long putTime = ((elapsed / 1000));
532         LOG.info("Put Elapsed time: " + putTime);
533 
534         assertEquals(2000, cache1.getSize());
535 
536         Thread.sleep(7000);
537         assertEquals(2000, manager2.getCache("sampleCache1").getSize());
538         assertEquals(2000, manager3.getCache("sampleCache1").getSize());
539         assertEquals(2000, manager4.getCache("sampleCache1").getSize());
540         assertEquals(2000, manager5.getCache("sampleCache1").getSize());
541 
542         //now test bootstrap
543         manager1.addCache("bootStrapResults");
544         Cache cache = manager1.getCache("bootStrapResults");
545         List cachePeers = manager1.getCacheManagerPeerProvider().listRemoteCachePeers(cache1);
546         CachePeer cachePeer = (CachePeer) cachePeers.get(0);
547 
548         List keys = cachePeer.getKeys();
549         assertEquals(2000, keys.size());
550 
551         Element firstElement = cachePeer.getQuiet((Serializable) keys.get(0));
552         long size = firstElement.getSerializedSize();
553         assertEquals(574, size);
554 
555         int chunkSize = (int) (5000000 / size);
556 
557         List requestChunk = new ArrayList();
558         for (int i = 0; i < keys.size(); i++) {
559             Serializable serializable = (Serializable) keys.get(i);
560             requestChunk.add(serializable);
561             if (requestChunk.size() == chunkSize) {
562                 fetchAndPutElements(cache, requestChunk, cachePeer);
563                 requestChunk.clear();
564             }
565         }
566         //get leftovers
567         fetchAndPutElements(cache, requestChunk, cachePeer);
568 
569         assertEquals(keys.size(), cache.getSize());
570 
571     }
572 
573     private void fetchAndPutElements(Ehcache cache, List requestChunk, CachePeer cachePeer) throws RemoteException {
574         List receivedChunk = cachePeer.getElements(requestChunk);
575         for (int i = 0; i < receivedChunk.size(); i++) {
576             Element element = (Element) receivedChunk.get(i);
577             assertNotNull(element);
578             cache.put(element, true);
579         }
580 
581     }
582 
583 
584     /**
585      * Drive everything to point of breakage within a 64MB VM.
586      */
587     public void xTestHugePutsBreaksAsynchronous() throws CacheException, InterruptedException {
588 
589         if (JVMUtil.isSingleRMIRegistryPerVM()) {
590             return;
591         }
592 
593         //Give everything a chance to startup
594         StopWatch stopWatch = new StopWatch();
595         Integer index = null;
596         for (int i = 0; i < 500; i++) {
597             for (int j = 0; j < 1000; j++) {
598                 index = new Integer(((1000 * i) + j));
599                 cache1.put(new Element(index,
600                         "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
601                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
602                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
603                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
604                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"));
605             }
606 
607         }
608         long elapsed = stopWatch.getElapsedTime();
609         long putTime = ((elapsed / 1000));
610         LOG.info("Put Elapsed time: " + putTime);
611         //assertTrue(putTime < 8);
612 
613         assertEquals(100000, cache1.getSize());
614 
615         Thread.sleep(100000);
616         assertEquals(20000, manager2.getCache("sampleCache1").getSize());
617         assertEquals(20000, manager3.getCache("sampleCache1").getSize());
618         assertEquals(20000, manager4.getCache("sampleCache1").getSize());
619         assertEquals(20000, manager5.getCache("sampleCache1").getSize());
620 
621     }
622 
623 
624     /**
625      * Performance and capacity tests.
626      * <p/>
627      * The numbers given are for the remote peer tester (java -jar ehcache-1.x-remote-debugger.jar ehcache-distributed1.xml)
628      * running on a 10Mbit ethernet network and are measured from the time the peer starts receiving to when
629      * it has fully received.
630      * <p/>
631      * 4 seconds to get all remove notifications with 6 peers, 5000 Elements and 400 byte payload
632      */
633     public void testBigRemovesProgagatesAsynchronous() throws CacheException, InterruptedException {
634 
635         if (JVMUtil.isSingleRMIRegistryPerVM()) {
636             return;
637         }
638 
639         //Give everything a chance to startup
640         Integer index = null;
641         for (int i = 0; i < 5; i++) {
642             for (int j = 0; j < 1000; j++) {
643                 index = new Integer(((1000 * i) + j));
644                 cache1.put(new Element(index,
645                         "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
646                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
647                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
648                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
649                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"));
650             }
651 
652         }
653         Thread.sleep(8000);
654         assertEquals(5000, cache1.getSize());
655         assertEquals(5000, manager2.getCache("sampleCache1").getSize());
656         assertEquals(5000, manager3.getCache("sampleCache1").getSize());
657         assertEquals(5000, manager4.getCache("sampleCache1").getSize());
658         assertEquals(5000, manager5.getCache("sampleCache1").getSize());
659 
660         //Let the disk stores catch up before the next stage of the test
661         Thread.sleep(2000);
662 
663         StopWatch stopWatch = new StopWatch();
664 
665         for (int i = 0; i < 5; i++) {
666             for (int j = 0; j < 1000; j++) {
667                 index = new Integer(((1000 * i) + j));
668                 cache1.remove(index);
669             }
670         }
671 
672 
673         int timeForPropagate = 10000;
674 
675         Thread.sleep(timeForPropagate);
676         assertEquals(0, cache1.getSize());
677         assertEquals(0, manager2.getCache("sampleCache1").getSize());
678         assertEquals(0, manager3.getCache("sampleCache1").getSize());
679         assertEquals(0, manager4.getCache("sampleCache1").getSize());
680         assertEquals(0, manager5.getCache("sampleCache1").getSize());
681 
682         LOG.info("Remove Elapsed time: " + timeForPropagate);
683 
684 
685     }
686 
687 
688     /**
689      * Performance and capacity tests.
690      * <p/>
691      * 5 seconds to send all notifications synchronously with 5 peers, 2000 Elements and 400 byte payload
692      * The numbers given below are for the remote peer tester (java -jar ehcache-1.x-remote-debugger.jar ehcache-distributed1.xml)
693      * running on a 10Mbit ethernet network and are measured from the time the peer starts receiving to when
694      * it has fully received.
695      */
696     public void testBigPutsProgagatesSynchronous() throws CacheException, InterruptedException {
697 
698         if (JVMUtil.isSingleRMIRegistryPerVM()) {
699             return;
700         }
701 
702         //Give everything a chance to startup
703         StopWatch stopWatch = new StopWatch();
704         Integer index;
705         for (int i = 0; i < 2; i++) {
706             for (int j = 0; j < 1000; j++) {
707                 index = new Integer(((1000 * i) + j));
708                 manager1.getCache("sampleCache3").put(new Element(index,
709                         "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
710                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
711                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
712                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
713                                 + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"));
714             }
715 
716         }
717         long elapsed = stopWatch.getElapsedTime();
718         long putTime = ((elapsed / 1000));
719         LOG.info("Put and Propagate Synchronously Elapsed time: " + putTime + " seconds");
720 
721         assertEquals(2000, manager1.getCache("sampleCache3").getSize());
722         assertEquals(2000, manager2.getCache("sampleCache3").getSize());
723         assertEquals(2000, manager3.getCache("sampleCache3").getSize());
724         assertEquals(2000, manager4.getCache("sampleCache3").getSize());
725         assertEquals(2000, manager5.getCache("sampleCache3").getSize());
726 
727     }
728 
729 
730     /**
731      * manager1 adds a replicating cache, then manager2 and so on. Then we remove one. Does everything work as expected?
732      */
733     public void testPutWithNewCacheAddedProgressively() throws InterruptedException {
734 
735         manager1.addCache("progressiveAddCache");
736         manager2.addCache("progressiveAddCache");
737 
738         //The cluster will not have formed yet, so it will fail
739         try {
740             putTest(manager1.getCache("progressiveAddCache"), manager2.getCache("progressiveAddCache"), ASYNCHRONOUS);
741             fail();
742         } catch (AssertionFailedError e) {
743             //expected
744         }
745 
746         //The cluster will now have formed yet, so it will succeed
747         putTest(manager1.getCache("progressiveAddCache"), manager2.getCache("progressiveAddCache"), ASYNCHRONOUS);
748 
749         Cache secondCache = manager2.getCache("progressiveAddCache");
750 
751         //The second peer disappears. The test will fail.
752         manager2.removeCache("progressiveAddCache");
753         try {
754             putTest(manager1.getCache("progressiveAddCache"), secondCache, ASYNCHRONOUS);
755             fail();
756         } catch (IllegalStateException e) {
757             //The second cache will not alive. Expected. But no other exception is caught and this will otherwise fail.
758 
759         }
760 
761 
762     }
763 
764 
765     /**
766      * Test various cache configurations for cache1 - explicit setting of:
767      * properties="replicateAsynchronously=true, replicatePuts=true, replicateUpdates=true, replicateUpdatesViaCopy=true, replicateRemovals=true "/>
768      */
769     public void testPutWithExplicitReplicationConfig() throws InterruptedException {
770         if (JVMUtil.isSingleRMIRegistryPerVM()) {
771             return;
772         }
773         putTest(manager1.getCache("sampleCache1"), manager2.getCache("sampleCache1"), ASYNCHRONOUS);
774     }
775 
776 
777     /**
778      * Test various cache configurations for cache1 - explicit setting of:
779      * properties="replicateAsynchronously=true, replicatePuts=true, replicateUpdates=true, replicateUpdatesViaCopy=true, replicateRemovals=true "/>
780      */
781     public void testPutWithThreadKiller() throws InterruptedException {
782         if (JVMUtil.isSingleRMIRegistryPerVM()) {
783             return;
784         }
785         putTestWithThreadKiller(manager1.getCache("sampleCache1"), manager2.getCache("sampleCache1"), ASYNCHRONOUS);
786     }
787 
788     /**
789      * CacheEventListeners that are not CacheReplicators should receive cache events originated from receipt
790      * of a remote event by a CachePeer.
791      */
792     public void testRemotelyReceivedPutNotifiesCountingListener() throws InterruptedException {
793         if (JVMUtil.isSingleRMIRegistryPerVM()) {
794             return;
795         }
796         putTest(manager1.getCache("sampleCache1"), manager2.getCache("sampleCache1"), ASYNCHRONOUS);
797         assertEquals(1, CountingCacheEventListener.getCacheElementsPut(manager1.getCache("sampleCache1")).size());
798         assertEquals(1, CountingCacheEventListener.getCacheElementsPut(manager2.getCache("sampleCache1")).size());
799 
800     }
801 
802     /**
803      * Test various cache configurations for cache1 - explicit setting of:
804      * properties="replicateAsynchronously=false, replicatePuts=true, replicateUpdates=true, replicateUpdatesViaCopy=true, replicateRemovals=true "/>
805      */
806     public void testPutWithExplicitReplicationSynchronousConfig() throws InterruptedException {
807         if (JVMUtil.isSingleRMIRegistryPerVM()) {
808             return;
809         }
810         putTest(manager1.getCache("sampleCache3"), manager2.getCache("sampleCache3"), SYNCHRONOUS);
811     }
812 
813 
814     /**
815      * Test put replicated for cache4 - no properties.
816      * Defaults should be replicateAsynchronously=true, replicatePuts=true, replicateUpdates=true, replicateUpdatesViaCopy=true, replicateRemovals=true
817      */
818     public void testPutWithEmptyReplicationPropertiesConfig() throws InterruptedException {
819         if (JVMUtil.isSingleRMIRegistryPerVM()) {
820             return;
821         }
822         putTest(manager1.getCache("sampleCache4"), manager2.getCache("sampleCache4"), ASYNCHRONOUS);
823     }
824 
825     /**
826      * Test put replicated for cache4 - missing replicatePuts property.
827      * replicateAsynchronously=true, replicatePuts=true, replicateUpdates=true, replicateUpdatesViaCopy=true, replicateRemovals=true
828      * should equal replicateAsynchronously=true, replicateUpdates=true, replicateUpdatesViaCopy=true, replicateRemovals=true
829      */
830     public void testPutWithOneMissingReplicationPropertyConfig() throws InterruptedException {
831         if (JVMUtil.isSingleRMIRegistryPerVM()) {
832             return;
833         }
834         putTest(manager1.getCache("sampleCache5"), manager2.getCache("sampleCache5"), ASYNCHRONOUS);
835     }
836 
837 
838     /**
839      * Tests put and remove initiated from cache1 in a cluster
840      * <p/>
841      * This test goes into an infinite loop if the chain of notifications is not somehow broken.
842      */
843     public void putTest(Ehcache fromCache, Ehcache toCache, boolean asynchronous) throws CacheException, InterruptedException {
844 
845         Serializable key = new Date();
846         Serializable value = new Date();
847         Element sourceElement = new Element(key, value);
848 
849         //Put
850         fromCache.put(sourceElement);
851         int i = 0;
852 
853         if (asynchronous) {
854             waitForProgagate();
855         }
856 
857         //Should have been replicated to toCache.
858         Element deliveredElement = toCache.get(key);
859         assertEquals(sourceElement, deliveredElement);
860 
861     }
862 
863 
864     /**
865      * Tests put and remove initiated from cache1 in a cluster
866      * <p/>
867      * This test goes into an infinite loop if the chain of notifications is not somehow broken.
868      */
869     public void putTestWithThreadKiller(Ehcache fromCache, Ehcache toCache, boolean asynchronous)
870             throws CacheException, InterruptedException {
871 
872         fromCache.put(new Element("thread killer", new ThreadKiller()));
873         if (asynchronous) {
874             waitForProgagate();
875         }
876 
877         Serializable key = new Date();
878         Serializable value = new Date();
879         Element sourceElement = new Element(key, value);
880 
881         //Put
882         fromCache.put(sourceElement);
883 
884         if (asynchronous) {
885             waitForProgagate();
886         }
887 
888         //Should have been replicated to toCache.
889         Element deliveredElement = toCache.get(key);
890         assertEquals(sourceElement, deliveredElement);
891 
892     }
893 
894 
895     /**
896      * Checks that a put received from a remote cache notifies any registered listeners.
897      * <p/>
898      * This test goes into an infinite loop if the chain of notifications is not somehow broken.
899      */
900     public void testRemotePutNotificationGetsToOtherListeners() throws CacheException, InterruptedException {
901 
902         if (JVMUtil.isSingleRMIRegistryPerVM()) {
903             return;
904         }
905 
906         Serializable key = new Date();
907         Serializable value = new Date();
908         Element element1 = new Element(key, value);
909 
910         //Put
911         cache1.put(new Element("1", new Date()));
912         cache1.put(new Element("2", new Date()));
913         cache1.put(new Element("3", new Date()));
914 
915         //Nonserializable and non deliverable put
916         Object nonSerializableObject = new Object();
917         cache1.put(new Element(nonSerializableObject, new Object()));
918 
919 
920         waitForProgagate();
921 
922         //local initiating cache's counting listener should have been notified
923         assertEquals(4, CountingCacheEventListener.getCacheElementsPut(cache1).size());
924         //remote receiving caches' counting listener should have been notified
925         assertEquals(3, CountingCacheEventListener.getCacheElementsPut(cache2).size());
926 
927         //Update
928         cache1.put(new Element("1", new Date()));
929         cache1.put(new Element("2", new Date()));
930         cache1.put(new Element("3", new Date()));
931 
932         //Nonserializable and non deliverable put
933         cache1.put(new Element(nonSerializableObject, new Object()));
934 
935         waitForProgagate();
936 
937         //local initiating cache's counting listener should have been notified
938         assertEquals(4, CountingCacheEventListener.getCacheElementsUpdated(cache1).size());
939         //remote receiving caches' counting listener should have been notified
940         assertEquals(3, CountingCacheEventListener.getCacheElementsUpdated(cache2).size());
941 
942         //Remove
943         cache1.remove("1");
944         cache1.remove("2");
945         cache1.remove("3");
946         cache1.remove(nonSerializableObject);
947 
948         waitForProgagate();
949 
950         //local initiating cache's counting listener should have been notified
951         assertEquals(4, CountingCacheEventListener.getCacheElementsRemoved(cache1).size());
952         //remote receiving caches' counting listener should have been notified
953         assertEquals(3, CountingCacheEventListener.getCacheElementsRemoved(cache2).size());
954 
955     }
956 
957 
958     /**
959      * Test various cache configurations for cache1 - explicit setting of:
960      * properties="replicateAsynchronously=true, replicatePuts=true, replicateUpdates=true, replicateUpdatesViaCopy=true, replicateRemovals=true "/>
961      */
962     public void testRemoveWithExplicitReplicationConfig() throws InterruptedException {
963         if (JVMUtil.isSingleRMIRegistryPerVM()) {
964             return;
965         }
966         removeTest(manager1.getCache("sampleCache1"), manager2.getCache("sampleCache1"), ASYNCHRONOUS);
967     }
968 
969     /**
970      * Test various cache configurations for cache1 - explicit setting of:
971      * properties="replicateAsynchronously=true, replicatePuts=true, replicateUpdates=true, replicateUpdatesViaCopy=true, replicateRemovals=true "/>
972      */
973     public void testRemoveWithExplicitReplicationSynchronousConfig() throws InterruptedException {
974         if (JVMUtil.isSingleRMIRegistryPerVM()) {
975             return;
976         }
977         removeTest(manager1.getCache("sampleCache3"), manager2.getCache("sampleCache3"), SYNCHRONOUS);
978     }
979 
980 
981     /**
982      * Test put replicated for cache4 - no properties.
983      * Defaults should be replicateAsynchronously=true, replicatePuts=true, replicateUpdates=true, replicateUpdatesViaCopy=true, replicateRemovals=true
984      */
985     public void testRemoveWithEmptyReplicationPropertiesConfig() throws InterruptedException {
986         if (JVMUtil.isSingleRMIRegistryPerVM()) {
987             return;
988         }
989         removeTest(manager1.getCache("sampleCache4"), manager2.getCache("sampleCache4"), ASYNCHRONOUS);
990     }
991 
992     /**
993      * Tests put and remove initiated from a cache to another cache in a cluster
994      * <p/>
995      * This test goes into an infinite loop if the chain of notifications is not somehow broken.
996      */
997     public void removeTest(Ehcache fromCache, Ehcache toCache, boolean asynchronous) throws CacheException, InterruptedException {
998 
999         Serializable key = new Date();
1000         Serializable value = new Date();
1001         Element element1 = new Element(key, value);
1002 
1003         //Put
1004         fromCache.put(element1);
1005 
1006         if (asynchronous) {
1007             waitForProgagate();
1008         }
1009 
1010         //Should have been replicated to cache2.
1011         Element element2 = toCache.get(key);
1012         assertEquals(element1, element2);
1013 
1014         //Remove
1015         fromCache.remove(key);
1016         if (asynchronous) {
1017             waitForProgagate();
1018         }
1019 
1020         //Should have been replicated to cache2.
1021         element2 = toCache.get(key);
1022         assertNull(element2);
1023 
1024     }
1025 
1026 
1027     /**
1028      * test removeAll sync
1029      */
1030     public void testRemoveAllAsynchronous() throws Exception {
1031         if (JVMUtil.isSingleRMIRegistryPerVM()) {
1032             return;
1033         }
1034         removeAllTest(manager1.getCache("sampleCache1"), manager2.getCache("sampleCache1"), ASYNCHRONOUS);
1035     }
1036 
1037     /**
1038      * test removeAll async
1039      */
1040     public void testRemoveAllSynchronous() throws Exception {
1041         if (JVMUtil.isSingleRMIRegistryPerVM()) {
1042             return;
1043         }
1044         removeAllTest(manager1.getCache("sampleCache3"), manager2.getCache("sampleCache3"), SYNCHRONOUS);
1045     }
1046 
1047     /**
1048      * Tests removeAll initiated from a cache to another cache in a cluster
1049      * <p/>
1050      * This test goes into an infinite loop if the chain of notifications is not somehow broken.
1051      */
1052     public void removeAllTest(Ehcache fromCache, Ehcache toCache, boolean asynchronous) throws Exception {
1053 
1054         //removeAll is distributed. Stop it colliding with the rest of the test
1055         waitForProgagate();
1056 
1057 
1058         Serializable key = new Date();
1059         Serializable value = new Date();
1060         Element element1 = new Element(key, value);
1061 
1062         //Put
1063         fromCache.put(element1);
1064 
1065 
1066         if (asynchronous) {
1067             waitForProgagate();
1068         }
1069 
1070         //Should have been replicated to cache2.
1071         Element element2 = toCache.get(key);
1072         assertEquals(element1, element2);
1073 
1074         //Remove
1075         fromCache.removeAll();
1076         if (asynchronous) {
1077             waitForProgagate();
1078         }
1079 
1080         //Should have been replicated to cache2.
1081         element2 = toCache.get(key);
1082         assertNull(element2);
1083         assertEquals(0, toCache.getSize());
1084 
1085     }
1086 
1087 
1088     /**
1089      * Test various cache configurations for cache1 - explicit setting of:
1090      * properties="replicateAsynchronously=true, replicatePuts=true, replicateUpdates=true, replicateUpdatesViaCopy=true, replicateRemovals=true "/>
1091      */
1092     public void testUpdateWithExplicitReplicationConfig() throws Exception {
1093         if (JVMUtil.isSingleRMIRegistryPerVM()) {
1094             return;
1095         }
1096         updateViaCopyTest(manager1.getCache("sampleCache1"), manager2.getCache("sampleCache1"), ASYNCHRONOUS);
1097     }
1098 
1099     /**
1100      * Test various cache configurations for cache1 - explicit setting of:
1101      * properties="replicateAsynchronously=true, replicatePuts=true, replicateUpdates=true, replicateUpdatesViaCopy=true, replicateRemovals=true "/>
1102      */
1103     public void testUpdateWithExplicitReplicationSynchronousConfig() throws Exception {
1104         if (JVMUtil.isSingleRMIRegistryPerVM()) {
1105             return;
1106         }
1107         updateViaCopyTest(manager1.getCache("sampleCache3"), manager2.getCache("sampleCache3"), SYNCHRONOUS);
1108     }
1109 
1110 
1111     /**
1112      * Test put replicated for cache4 - no properties.
1113      * Defaults should be replicateAsynchronously=true, replicatePuts=true, replicateUpdates=true, replicateUpdatesViaCopy=true, replicateRemovals=true
1114      */
1115     public void testUpdateWithEmptyReplicationPropertiesConfig() throws Exception {
1116         if (JVMUtil.isSingleRMIRegistryPerVM()) {
1117             return;
1118         }
1119         updateViaCopyTest(manager1.getCache("sampleCache4"), manager2.getCache("sampleCache4"), ASYNCHRONOUS);
1120     }
1121 
1122     /**
1123      * Tests put and update through copy initiated from cache1 in a cluster
1124      * <p/>
1125      * This test goes into an infinite loop if the chain of notifications is not somehow broken.
1126      */
1127     public void updateViaCopyTest(Ehcache fromCache, Ehcache toCache, boolean asynchronous) throws Exception {
1128 
1129         fromCache.removeAll();
1130         toCache.removeAll();
1131 
1132         //removeAll is distributed. Stop it colliding with the rest of the test
1133         waitForProgagate();
1134 
1135         Serializable key = new Date();
1136         Serializable value = new Date();
1137         Element element1 = new Element(key, value);
1138 
1139         //Put
1140         fromCache.put(element1);
1141         if (asynchronous) {
1142             waitForProgagate();
1143         }
1144 
1145         //Should have been replicated to cache2.
1146         Element element2 = toCache.get(key);
1147         assertEquals(element1, element2);
1148 
1149         //Update
1150         Element updatedElement1 = new Element(key, new Date());
1151 
1152         fromCache.put(updatedElement1);
1153         if (asynchronous) {
1154             waitForProgagate();
1155         }
1156 
1157         //Should have been replicated to cache2.
1158         Element receivedUpdatedElement2 = toCache.get(key);
1159         assertEquals(updatedElement1, receivedUpdatedElement2);
1160 
1161     }
1162 
1163 
1164     /**
1165      * Tests put and update through invalidation initiated from cache1 in a cluster
1166      * <p/>
1167      * This test goes into an infinite loop if the chain of notifications is not somehow broken.
1168      */
1169     public void testUpdateViaInvalidate() throws CacheException, InterruptedException, IOException {
1170 
1171         if (JVMUtil.isSingleRMIRegistryPerVM()) {
1172             return;
1173         }
1174 
1175         cache1 = manager1.getCache("sampleCache2");
1176         cache1.removeAll();
1177 
1178         cache2 = manager2.getCache("sampleCache2");
1179         cache2.removeAll();
1180 
1181         //removeAll is distributed. Stop it colliding with the rest of the test
1182         waitForProgagate();
1183 
1184         Serializable key = "1";
1185         Serializable value = new Date();
1186         Element element1 = new Element(key, value);
1187 
1188         //Put
1189         cache1.put(element1);
1190         waitForProgagate();
1191 
1192         //Should have been replicated to cache2.
1193         Element element2 = cache2.get(key);
1194         assertEquals(element1, element2);
1195 
1196         //Update
1197         cache1.put(element1);
1198         waitForProgagate();
1199 
1200         //Should have been removed in cache2.
1201         element2 = cache2.get(key);
1202         assertNull(element2);
1203 
1204     }
1205 
1206     /**
1207      * What happens when two cache instances replicate to each other and a change is initiated
1208      */
1209     public void testInfiniteNotificationsLoop() throws InterruptedException {
1210 
1211         if (JVMUtil.isSingleRMIRegistryPerVM()) {
1212             return;
1213         }
1214 
1215         Serializable key = "1";
1216         Serializable value = new Date();
1217         Element element = new Element(key, value);
1218 
1219         //Put
1220         cache1.put(element);
1221         waitForProgagate();
1222 
1223         //Should have been replicated to cache2.
1224         Element element2 = cache2.get(key);
1225         assertEquals(element, element2);
1226 
1227         //Remove
1228         cache1.remove(key);
1229         assertNull(cache1.get(key));
1230 
1231         //Should have been replicated to cache2.
1232         waitForProgagate();
1233         element2 = cache2.get(key);
1234         assertNull(element2);
1235 
1236         //Put into 2
1237         Element element3 = new Element("3", "ddsfds");
1238         cache2.put(element3);
1239         waitForProgagate();
1240         Element element4 = cache2.get("3");
1241         assertEquals(element3, element4);
1242 
1243     }
1244 
1245 
1246     /**
1247      * Need to wait for async
1248      *
1249      * @throws InterruptedException
1250      */
1251     protected void waitForProgagate() throws InterruptedException {
1252         Thread.sleep(2000);
1253     }
1254 
1255     /**
1256      * Need to wait for async
1257      *
1258      * @throws InterruptedException
1259      */
1260     protected void waitForSlowProgagate() throws InterruptedException {
1261         Thread.sleep(6000);
1262     }
1263 
1264 
1265     /**
1266      * Distributed operations create extra scope for deadlock.
1267      * This test checks whether a distributed deadlock scenario exists for synchronous replication
1268      * of each distributed operation all at once.
1269      * It shows that no distributed deadlock exists for asynchronous replication. It is multi thread
1270      * and multi process safe.
1271      * <p/>
1272      * Carefully tailored to exercise:
1273      * <ol>
1274      * <li>overflow to disk. We put in 20 things and the memory size is 10
1275      * <li>each peer is working on the same set of keys thus maximising contention
1276      * <li>we do puts, gets and removes to explore all the execution paths
1277      * </ol>
1278      * If a deadlock occurs, processing will stop until a SocketTimeout exception is thrown and
1279      * the deadlock will be released.
1280      */
1281     public void testCacheOperationsSynchronousMultiThreaded() throws Exception, InterruptedException {
1282 
1283         if (JVMUtil.isSingleRMIRegistryPerVM()) {
1284             return;
1285         }
1286 
1287         // Run a set of threads, that attempt to fetch the elements
1288         final List executables = new ArrayList();
1289 
1290         executables.add(new ClusterExecutable(manager1, "sampleCache3"));
1291         executables.add(new ClusterExecutable(manager2, "sampleCache3"));
1292         executables.add(new ClusterExecutable(manager3, "sampleCache3"));
1293 
1294         runThreads(executables);
1295     }
1296 
1297 
1298     /**
1299      * Distributed operations create extra scope for deadlock.
1300      * This test checks whether a distributed deadlock scenario exists for asynchronous replication
1301      * of each distributed operation all at once.
1302      * It shows that no distributed deadlock exists for asynchronous replication. It is multi thread
1303      * and multi process safe.
1304      * It uses sampleCache2, which is configured to be asynchronous
1305      * <p/>
1306      * Carefully tailored to exercise:
1307      * <ol>
1308      * <li>overflow to disk. We put in 20 things and the memory size is 10
1309      * <li>each peer is working on the same set of keys thus maximising contention
1310      * <li>we do puts, gets and removes to explore all the execution paths
1311      * </ol>
1312      */
1313     public void testCacheOperationsAynchronousMultiThreaded() throws Exception, InterruptedException {
1314 
1315         if (JVMUtil.isSingleRMIRegistryPerVM()) {
1316             return;
1317         }
1318 
1319         // Run a set of threads, that attempt to fetch the elements
1320         final List executables = new ArrayList();
1321 
1322         executables.add(new ClusterExecutable(manager1, "sampleCache2"));
1323         executables.add(new ClusterExecutable(manager2, "sampleCache2"));
1324         executables.add(new ClusterExecutable(manager3, "sampleCache2"));
1325 
1326         runThreads(executables);
1327     }
1328 
1329     /**
1330      * An Exececutable which allows the CacheManager to be set
1331      */
1332     class ClusterExecutable implements Executable {
1333 
1334         private CacheManager manager;
1335         private String cacheName;
1336 
1337         /**
1338          * Construct with CacheManager
1339          *
1340          * @param manager
1341          */
1342         public ClusterExecutable(CacheManager manager, String cacheName) {
1343             this.manager = manager;
1344             this.cacheName = cacheName;
1345         }
1346 
1347         /**
1348          * Execute
1349          *
1350          * @throws Exception
1351          */
1352         public void execute() throws Exception {
1353             Random random = new Random();
1354 
1355             for (int i = 0; i < 20; i++) {
1356                 Integer key = new Integer((i));
1357                 int operationSelector = random.nextInt(4);
1358                 Cache cache = manager.getCache(cacheName);
1359                 if (operationSelector == 100) {
1360                     cache.get(key);
1361                     if (LOG.isDebugEnabled()) {
1362                         LOG.debug(cache.getGuid() + ": get " + key);
1363                     }
1364                 } else if (operationSelector == 100) {
1365                     cache.remove(key);
1366                     if (LOG.isDebugEnabled()) {
1367                         LOG.debug(cache.getGuid() + ": remove " + key);
1368                     }
1369                 } else if (operationSelector == 2) {
1370                     cache.put(new Element(key,
1371                             "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
1372                                     + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
1373                                     + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
1374                                     + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
1375                                     + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"));
1376                     if (LOG.isDebugEnabled()) {
1377                         LOG.debug(cache.getGuid() + ": put " + key);
1378                     }
1379                 } else {
1380                     //every twelfth time 1/4 * 1/3 = 1/12
1381                     if (random.nextInt(3) == 1) {
1382                         LOG.debug("cache.removeAll()");
1383                         cache.removeAll();
1384                     }
1385                 }
1386             }
1387 
1388         }
1389     }
1390 
1391 }