View Javadoc

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 net.sf.ehcache.Ehcache;
20  import net.sf.ehcache.Element;
21  import net.sf.ehcache.bootstrap.BootstrapCacheLoader;
22  import org.apache.commons.logging.Log;
23  import org.apache.commons.logging.LogFactory;
24  
25  import java.io.Serializable;
26  import java.rmi.RemoteException;
27  import java.util.ArrayList;
28  import java.util.List;
29  import java.util.Random;
30  
31  /**
32   * Loads Elements from a random Cache Peer
33   *
34   * @author Greg Luck
35   * @version $Id: RMIBootstrapCacheLoader.java 512 2007-07-10 09:18:45Z gregluck $
36   */
37  public class RMIBootstrapCacheLoader implements BootstrapCacheLoader {
38  
39      private static final int ONE_SECOND = 1000;
40  
41      private static final Log LOG = LogFactory.getLog(RMIBootstrapCacheLoader.class.getName());
42  
43      /**
44       * Whether to load asynchronously
45       */
46      protected boolean asynchronous;
47  
48      /**
49       * The maximum serialized size of the elements to request from a remote cache peer during bootstrap.
50       */
51      protected int maximumChunkSizeBytes;
52  
53      /**
54       * Creates a boostrap cache loader that will work with RMI based distribution
55       *
56       * @param asynchronous Whether to load asynchronously
57       */
58      public RMIBootstrapCacheLoader(boolean asynchronous, int maximumChunkSize) {
59          this.asynchronous = asynchronous;
60          this.maximumChunkSizeBytes = maximumChunkSize;
61      }
62  
63  
64      /**
65       * Bootstraps the cache from a random CachePeer. Requests are done in chunks estimated at 5MB Serializable
66       * size. This balances memory use on each end and network performance.
67       *
68       * @throws RemoteCacheException
69       *          if anything goes wrong with the remote call
70       */
71      public void load(Ehcache cache) throws RemoteCacheException {
72          if (asynchronous) {
73              BootstrapThread bootstrapThread = new BootstrapThread(cache);
74              bootstrapThread.start();
75          } else {
76              doLoad(cache);
77          }
78      }
79  
80      /**
81       * @return true if this bootstrap loader is asynchronous
82       */
83      public boolean isAsynchronous() {
84          return asynchronous;
85      }
86  
87  
88      /**
89       * A background daemon thread that asynchronously calls doLoad
90       */
91      private final class BootstrapThread extends Thread {
92          private Ehcache cache;
93  
94          public BootstrapThread(Ehcache cache) {
95              super("Bootstrap Thread for cache " + cache.getName());
96              this.cache = cache;
97              setDaemon(true);
98              setPriority(Thread.NORM_PRIORITY);
99          }
100 
101         /**
102          * RemoteDebugger thread method.
103          */
104         public final void run() {
105             try {
106                 doLoad(cache);
107             } catch (RemoteCacheException e) {
108                 LOG.warn("Error asynchronously performing bootstrap. The cause was: " + e.getMessage(), e);
109             } finally {
110                 cache = null;
111             }
112 
113         }
114 
115     }
116 
117 
118     /**
119      * Bootstraps the cache from a random CachePeer. Requests are done in chunks estimated at 5MB Serializable
120      * size. This balances memory use on each end and network performance.
121      * <p/>
122      * Bootstrapping requires the establishment of a cluster. This can be instantaneous for manually configued
123      * clusters or may take a number of seconds for multicast ones. This method waits up to 11 seconds for a cluster
124      * to form.
125      *
126      * @throws RemoteCacheException
127      *          if anything goes wrong with the remote call
128      */
129     public void doLoad(Ehcache cache) throws RemoteCacheException {
130 
131         List cachePeers = acquireCachePeers(cache);
132         if (cachePeers == null || cachePeers.size() == 0) {
133             LOG.debug("Empty list of cache peers for cache " + cache.getName() + ". No cache peer to bootstrap from.");
134             return;
135         }
136         Random random = new Random();
137         int randomPeerNumber = random.nextInt(cachePeers.size());
138         CachePeer cachePeer = (CachePeer) cachePeers.get(randomPeerNumber);
139         LOG.debug("Bootstrapping " + cache.getName() + " from " + cachePeer);
140 
141         try {
142 
143             //Estimate element size
144             Element sampleElement = null;
145             List keys = cachePeer.getKeys();
146             for (int i = 0; i < keys.size(); i++) {
147                 Serializable key = (Serializable) keys.get(i);
148                 sampleElement = cachePeer.getQuiet(key);
149                 if (sampleElement != null) {
150                     break;
151                 }
152             }
153             if (sampleElement == null) {
154                 LOG.debug("All cache peer elements were null. Nothing to bootstrap from. Cache was "
155                         + cache.getName() + ". Cache peer was " + cachePeer);
156                 return;
157             }
158             long size = sampleElement.getSerializedSize();
159             int chunkSize = (int) (maximumChunkSizeBytes / size);
160 
161             List requestChunk = new ArrayList();
162             for (int i = 0; i < keys.size(); i++) {
163                 Serializable serializable = (Serializable) keys.get(i);
164                 requestChunk.add(serializable);
165                 if (requestChunk.size() == chunkSize) {
166                     fetchAndPutElements(cache, requestChunk, cachePeer);
167                     requestChunk.clear();
168                 }
169             }
170             //get leftovers
171             fetchAndPutElements(cache, requestChunk, cachePeer);
172             LOG.debug("Bootstrap of " + cache.getName() + " from " + cachePeer + " finished. "
173                     + keys.size() + " keys requested.");
174         } catch (Throwable t) {
175             throw new RemoteCacheException("Error bootstrapping from remote peer. Message was: " + t.getMessage(), t);
176         }
177     }
178 
179     /**
180      * Acquires the cache peers for this cache.
181      *
182      * @param cache
183      */
184     protected List acquireCachePeers(Ehcache cache) {
185 
186         long timeForClusterToForm = 0;
187         CacheManagerPeerProvider cacheManagerPeerProvider = cache.getCacheManager().getCacheManagerPeerProvider();
188         if (cacheManagerPeerProvider != null) {
189             timeForClusterToForm = cacheManagerPeerProvider.getTimeForClusterToForm();
190         }
191         if (LOG.isDebugEnabled()) {
192             LOG.debug("Attempting to acquire cache peers for cache " + cache.getName()
193                     + " to bootstrap from. Will wait up to " + timeForClusterToForm + "ms for cache to join cluster.");
194         }
195         List cachePeers = null;
196         for (int i = 0; i <= timeForClusterToForm; i = i + ONE_SECOND) {
197             cachePeers = listRemoteCachePeers(cache);
198             if (cachePeers == null) {
199                 break;
200             }
201             if (cachePeers.size() > 0) {
202                 break;
203             }
204             try {
205                 Thread.sleep(ONE_SECOND);
206             } catch (InterruptedException e) {
207                 LOG.debug("doLoad for " + cache.getName() + " interrupted.");
208             }
209         }
210         if (LOG.isDebugEnabled()) {
211             LOG.debug("cache peers: " + cachePeers);
212         }
213         return cachePeers;
214     }
215 
216     /**
217      * Fetches a chunk of elements from a remote cache peer
218      *
219      * @param cache        the cache to put elements in
220      * @param requestChunk the chunk of keys to request
221      * @param cachePeer    the peer to fetch from
222      * @throws java.rmi.RemoteException
223      */
224     protected void fetchAndPutElements(Ehcache cache, List requestChunk, CachePeer cachePeer) throws RemoteException {
225         List receivedChunk = cachePeer.getElements(requestChunk);
226         for (int i = 0; i < receivedChunk.size(); i++) {
227             Element element = (Element) receivedChunk.get(i);
228             // element could be expired at the peer
229             if (element != null) {
230                 cache.put(element, true);
231             }
232         }
233     }
234 
235     /**
236      * Package protected List of cache peers
237      *
238      * @param cache
239      */
240     protected List listRemoteCachePeers(Ehcache cache) {
241         CacheManagerPeerProvider provider = cache.getCacheManager().getCachePeerProvider();
242         if (provider == null) {
243             return null;
244         } else {
245             return provider.listRemoteCachePeers(cache);
246         }
247 
248     }
249 
250     /**
251      * Gets the maximum chunk size
252      */
253     public int getMaximumChunkSizeBytes() {
254         return maximumChunkSizeBytes;
255     }
256 
257     /**
258      * Clones this loader
259      */
260     public Object clone() throws CloneNotSupportedException {
261         //checkstyle
262         return new RMIBootstrapCacheLoader(asynchronous, maximumChunkSizeBytes);
263     }
264 
265 }