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.CacheException;
20 import net.sf.ehcache.Ehcache;
21 import net.sf.ehcache.Element;
22 import net.sf.ehcache.Status;
23 import org.apache.commons.logging.Log;
24 import org.apache.commons.logging.LogFactory;
25
26 import java.io.Serializable;
27 import java.rmi.UnmarshalException;
28 import java.util.ArrayList;
29 import java.util.LinkedList;
30 import java.util.List;
31
32 /**
33 * Listens to {@link net.sf.ehcache.CacheManager} and {@link net.sf.ehcache.Cache} events and propagates those to
34 * {@link CachePeer} peers of the Cache asynchronously.
35 * <p/>
36 * Updates are guaranteed to be replicated in the order in which they are received.
37 * <p/>
38 * While much faster in operation than {@link RMISynchronousCacheReplicator}, it does suffer from a number
39 * of problems. Elements, which may be being spooled to DiskStore may stay around in memory because references
40 * are being held to them from {@link EventMessage}s which are queued up. The replication thread runs once
41 * per second, limiting the build up. However a lot of elements can be put into a cache in that time. We do not want
42 * to get an {@link OutOfMemoryError} using distribution in circumstances when it would not happen if we were
43 * just using the DiskStore.
44 * <p/>
45 * Accordingly, the Element values in {@link EventMessage}s are held by {@link java.lang.ref.SoftReference} in the queue,
46 * so that they can be discarded if required by the GC to avoid an {@link OutOfMemoryError}. A log message
47 * will be issued on each flush of the queue if there were any forced discards. One problem with GC collection
48 * of SoftReferences is that the VM (JDK1.5 anyway) will do that rather than grow the heap size to the maximum.
49 * The workaround is to either set minimum heap size to the maximum heap size to force heap allocation at start
50 * up, or put up with a few lost messages while the heap grows.
51 *
52 * @author Greg Luck
53 * @version $Id: RMIAsynchronousCacheReplicator.java 512 2007-07-10 09:18:45Z gregluck $
54 */
55 public class RMIAsynchronousCacheReplicator extends RMISynchronousCacheReplicator {
56
57
58 private static final Log LOG = LogFactory.getLog(RMIAsynchronousCacheReplicator.class.getName());
59
60 /**
61 * A thread which handles replication, so that replication can take place asynchronously and not hold up the cache
62 */
63 protected Thread replicationThread = new ReplicationThread();
64
65 /**
66 * The amount of time the replication thread sleeps after it detects the replicationQueue is empty
67 * before checking again.
68 */
69 protected int asynchronousReplicationInterval;
70
71 /**
72 * A queue of updates.
73 */
74 protected final List replicationQueue = new LinkedList();
75
76 /**
77 * Constructor for internal and subclass use
78 *
79 * @param replicatePuts
80 * @param replicateUpdates
81 * @param replicateUpdatesViaCopy
82 * @param replicateRemovals
83 * @param asynchronousReplicationInterval
84 *
85 */
86 public RMIAsynchronousCacheReplicator(
87 boolean replicatePuts,
88 boolean replicateUpdates,
89 boolean replicateUpdatesViaCopy,
90 boolean replicateRemovals,
91 int asynchronousReplicationInterval) {
92 super(replicatePuts,
93 replicateUpdates,
94 replicateUpdatesViaCopy,
95 replicateRemovals);
96 this.asynchronousReplicationInterval = asynchronousReplicationInterval;
97 status = Status.STATUS_ALIVE;
98 replicationThread.start();
99 }
100
101 /**
102 * RemoteDebugger method for the replicationQueue thread.
103 * <p/>
104 * Note that the replicationQueue thread locks the cache for the entire time it is writing elements to the disk.
105 */
106 private void replicationThreadMain() {
107 while (true) {
108 // Wait for elements in the replicationQueue
109 while (alive() && replicationQueue != null && replicationQueue.size() == 0) {
110 try {
111 Thread.sleep(asynchronousReplicationInterval);
112 } catch (InterruptedException e) {
113 LOG.debug("Spool Thread interrupted.");
114 return;
115 }
116 }
117 if (notAlive()) {
118 return;
119 }
120 try {
121 if (replicationQueue.size() != 0) {
122 flushReplicationQueue();
123 }
124 } catch (Throwable e) {
125 LOG.warn("Exception on flushing of replication queue: " + e.getMessage()
126 + ". Continuing...", e);
127 }
128 }
129 }
130
131
132 /**
133 * {@inheritDoc}
134 * <p/>
135 * This implementation queues the put notification for in-order replication to peers.
136 *
137 * @param cache the cache emitting the notification
138 * @param element the element which was just put into the cache.
139 */
140 public final void notifyElementPut(final Ehcache cache, final Element element) throws CacheException {
141 if (notAlive()) {
142 return;
143 }
144
145 if (!replicatePuts) {
146 return;
147 }
148
149 if (!element.isSerializable()) {
150 if (LOG.isWarnEnabled()) {
151 LOG.warn("Object with key " + element.getObjectKey() + " is not Serializable and cannot be replicated");
152 }
153 return;
154 }
155 addToReplicationQueue(new CacheEventMessage(EventMessage.PUT, cache, element, null));
156 }
157
158 /**
159 * Called immediately after an element has been put into the cache and the element already
160 * existed in the cache. This is thus an update.
161 * <p/>
162 * The {@link net.sf.ehcache.Cache#put(net.sf.ehcache.Element)} method
163 * will block until this method returns.
164 * <p/>
165 * Implementers may wish to have access to the Element's fields, including value, so the element is provided.
166 * Implementers should be careful not to modify the element. The effect of any modifications is undefined.
167 *
168 * @param cache the cache emitting the notification
169 * @param element the element which was just put into the cache.
170 */
171 public final void notifyElementUpdated(final Ehcache cache, final Element element) throws CacheException {
172 if (notAlive()) {
173 return;
174 }
175 if (!replicateUpdates) {
176 return;
177 }
178
179 if (replicateUpdatesViaCopy) {
180 if (!element.isSerializable()) {
181 if (LOG.isWarnEnabled()) {
182 LOG.warn("Object with key " + element.getObjectKey() + " is not Serializable and cannot be updated via copy");
183 }
184 return;
185 }
186 addToReplicationQueue(new CacheEventMessage(EventMessage.PUT, cache, element, null));
187 } else {
188 if (!element.isKeySerializable()) {
189 if (LOG.isWarnEnabled()) {
190 LOG.warn("Key " + element.getObjectKey() + " is not Serializable and cannot be replicated.");
191 }
192 return;
193 }
194 addToReplicationQueue(new CacheEventMessage(EventMessage.REMOVE, cache, null, element.getKey()));
195 }
196 }
197
198 /**
199 * Called immediately after an attempt to remove an element. The remove method will block until
200 * this method returns.
201 * <p/>
202 * This notification is received regardless of whether the cache had an element matching
203 * the removal key or not. If an element was removed, the element is passed to this method,
204 * otherwise a synthetic element, with only the key set is passed in.
205 * <p/>
206 *
207 * @param cache the cache emitting the notification
208 * @param element the element just deleted, or a synthetic element with just the key set if
209 * no element was removed.
210 */
211 public final void notifyElementRemoved(final Ehcache cache, final Element element) throws CacheException {
212 if (notAlive()) {
213 return;
214 }
215
216 if (!replicateRemovals) {
217 return;
218 }
219
220 if (!element.isKeySerializable()) {
221 if (LOG.isWarnEnabled()) {
222 LOG.warn("Key " + element.getObjectKey() + " is not Serializable and cannot be replicated.");
223 }
224 return;
225 }
226 addToReplicationQueue(new CacheEventMessage(EventMessage.REMOVE, cache, null, element.getKey()));
227 }
228
229
230 /**
231 * Called during {@link net.sf.ehcache.Ehcache#removeAll()} to indicate that the all
232 * elements have been removed from the cache in a bulk operation. The usual
233 * {@link #notifyElementRemoved(net.sf.ehcache.Ehcache,net.sf.ehcache.Element)}
234 * is not called.
235 * <p/>
236 * This notification exists because clearing a cache is a special case. It is often
237 * not practical to serially process notifications where potentially millions of elements
238 * have been bulk deleted.
239 *
240 * @param cache the cache emitting the notification
241 */
242 public void notifyRemoveAll(final Ehcache cache) {
243 if (notAlive()) {
244 return;
245 }
246
247 if (!replicateRemovals) {
248 return;
249 }
250
251 addToReplicationQueue(new CacheEventMessage(EventMessage.REMOVE_ALL, cache, null, null));
252 }
253
254
255 /**
256 * Adds a message to the queue.
257 * <p/>
258 * This method checks the state of the replication thread and warns
259 * if it has stopped and then discards the message.
260 *
261 * @param cacheEventMessage
262 */
263 protected void addToReplicationQueue(CacheEventMessage cacheEventMessage) {
264 if (!replicationThread.isAlive()) {
265 LOG.error("CacheEventMessages cannot be added to the replication queue"
266 + " because the replication thread has died.");
267 } else {
268 synchronized (replicationQueue) {
269 replicationQueue.add(cacheEventMessage);
270 }
271 }
272 }
273
274
275 /**
276 * Gets called once per {@link #asynchronousReplicationInterval}.
277 * <p/>
278 * Sends accumulated messages in bulk to each peer. i.e. if ther are 100 messages and 1 peer,
279 * 1 RMI invocation results, not 100. Also, if a peer is unavailable this is discovered in only 1 try.
280 * <p/>
281 * Makes a copy of the queue so as not to hold up the enqueue operations.
282 * <p/>
283 * Any exceptions are caught so that the replication thread does not die, and because errors are expected,
284 * due to peers becoming unavailable.
285 * <p/>
286 * This method issues warnings for problems that can be fixed with configuration changes.
287 */
288 private void flushReplicationQueue() {
289 List replicationQueueCopy;
290 synchronized (replicationQueue) {
291 if (replicationQueue.size() == 0) {
292 return;
293 }
294
295 replicationQueueCopy = new ArrayList(replicationQueue.size());
296 for (int i = 0; i < replicationQueue.size(); i++) {
297 CacheEventMessage cacheEventMessage = (CacheEventMessage) replicationQueue.get(i);
298 replicationQueueCopy.add(cacheEventMessage);
299 }
300 replicationQueue.clear();
301 }
302
303
304 Ehcache cache = ((CacheEventMessage) replicationQueueCopy.get(0)).cache;
305 List cachePeers = listRemoteCachePeers(cache);
306
307 List resolvedEventMessages = extractAndResolveEventMessages(replicationQueueCopy);
308
309
310 for (int j = 0; j < cachePeers.size(); j++) {
311 CachePeer cachePeer = (CachePeer) cachePeers.get(j);
312 try {
313 cachePeer.send(resolvedEventMessages);
314 } catch (UnmarshalException e) {
315 String message = e.getMessage();
316 if (message.indexOf("Read time out") != 0) {
317 LOG.warn("Unable to send message to remote peer due to socket read timeout. Consider increasing" +
318 " the socketTimeoutMillis setting in the cacheManagerPeerListenerFactory. " +
319 "Message was: " + e.getMessage());
320 } else {
321 LOG.debug("Unable to send message to remote peer. Message was: " + e.getMessage());
322 }
323 } catch (Throwable t) {
324 LOG.warn("Unable to send message to remote peer. Message was: " + t.getMessage(), t);
325 }
326 }
327 if (LOG.isWarnEnabled()) {
328 int eventMessagesNotResolved = replicationQueueCopy.size() - resolvedEventMessages.size();
329 if (eventMessagesNotResolved > 0) {
330 LOG.warn(eventMessagesNotResolved + " messages were discarded on replicate due to reclamation of " +
331 "SoftReferences by the VM. Consider increasing the maximum heap size and/or setting the " +
332 "starting heap size to a higher value.");
333 }
334
335 }
336 }
337
338 /**
339 * Extracts CacheEventMessages and attempts to get a hard reference to the underlying EventMessage
340 * <p/>
341 * If an EventMessage has been invalidated due to SoftReference collection of the Element, it is not
342 * propagated. This only affects puts and updates via copy.
343 *
344 * @param replicationQueueCopy
345 * @return a list of EventMessages which were able to be resolved
346 */
347 private static List extractAndResolveEventMessages(List replicationQueueCopy) {
348 List list = new ArrayList();
349 for (int i = 0; i < replicationQueueCopy.size(); i++) {
350 EventMessage eventMessage = ((CacheEventMessage) replicationQueueCopy.get(i)).getEventMessage();
351 if (eventMessage != null && eventMessage.isValid()) {
352 list.add(eventMessage);
353 }
354 }
355 return list;
356 }
357
358 /**
359 * A background daemon thread that writes objects to the file.
360 */
361 private final class ReplicationThread extends Thread {
362 public ReplicationThread() {
363 super("Replication Thread");
364 setDaemon(true);
365 setPriority(Thread.NORM_PRIORITY);
366 }
367
368 /**
369 * RemoteDebugger thread method.
370 */
371 public final void run() {
372 replicationThreadMain();
373 }
374 }
375
376
377 /**
378 * A wrapper around an EventMessage, which enables the element to be enqueued along with
379 * what is to be done with it.
380 * <p/>
381 * The wrapper holds a {@link java.lang.ref.SoftReference} to the {@link EventMessage}, so that the queue is never
382 * the cause of an {@link OutOfMemoryError}
383 */
384 private static class CacheEventMessage {
385
386 private final Ehcache cache;
387 private final EventMessage eventMessage;
388
389 public CacheEventMessage(int event, Ehcache cache, Element element, Serializable key) {
390 eventMessage = new EventMessage(event, key, element);
391 this.cache = cache;
392 }
393
394 /**
395 * Gets the component EventMessage
396 */
397 public final EventMessage getEventMessage() {
398 return eventMessage;
399 }
400
401 }
402
403 /**
404 * Give the replicator a chance to flush the replication queue, then cleanup and free resources when no longer needed
405 */
406 public final void dispose() {
407 status = Status.STATUS_SHUTDOWN;
408 flushReplicationQueue();
409 }
410
411
412 /**
413 * Creates a clone of this listener. This method will only be called by ehcache before a cache is initialized.
414 * <p/>
415 * This may not be possible for listeners after they have been initialized. Implementations should throw
416 * CloneNotSupportedException if they do not support clone.
417 *
418 * @return a clone
419 * @throws CloneNotSupportedException if the listener could not be cloned.
420 */
421 public Object clone() throws CloneNotSupportedException {
422 //shutup checkstyle
423 super.clone();
424 return new RMIAsynchronousCacheReplicator(replicatePuts, replicateUpdates,
425 replicateUpdatesViaCopy, replicateRemovals, asynchronousReplicationInterval);
426 }
427
428
429 }