1
2
3
4
5
6
7
8
9
10
11
12
13
14
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
33
34
35
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
45
46 protected boolean asynchronous;
47
48
49
50
51 protected int maximumChunkSizeBytes;
52
53
54
55
56
57
58 public RMIBootstrapCacheLoader(boolean asynchronous, int maximumChunkSize) {
59 this.asynchronous = asynchronous;
60 this.maximumChunkSizeBytes = maximumChunkSize;
61 }
62
63
64
65
66
67
68
69
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
82
83 public boolean isAsynchronous() {
84 return asynchronous;
85 }
86
87
88
89
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
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
120
121
122
123
124
125
126
127
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
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
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
181
182
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
218
219
220
221
222
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
229 if (element != null) {
230 cache.put(element, true);
231 }
232 }
233 }
234
235
236
237
238
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
252
253 public int getMaximumChunkSizeBytes() {
254 return maximumChunkSizeBytes;
255 }
256
257
258
259
260 public Object clone() throws CloneNotSupportedException {
261
262 return new RMIBootstrapCacheLoader(asynchronous, maximumChunkSizeBytes);
263 }
264
265 }