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.CacheManager;
20 import org.apache.commons.logging.Log;
21 import org.apache.commons.logging.LogFactory;
22
23 import java.io.IOException;
24 import java.net.DatagramPacket;
25 import java.net.InetAddress;
26 import java.net.MulticastSocket;
27 import java.util.List;
28 import java.util.ArrayList;
29 import java.util.Iterator;
30
31 /**
32 * Sends heartbeats to a multicast group containing a compressed list of URLs.
33 * <p/>
34 * You can control how far the multicast packets propagate by setting the badly misnamed "TTL".
35 * Using the multicast IP protocol, the TTL value indicates the scope or range in which a packet may be forwarded.
36 * By convention:
37 * <ul>
38 * <li>0 is restricted to the same host
39 * <li>1 is restricted to the same subnet
40 * <li>32 is restricted to the same site
41 * <li>64 is restricted to the same region
42 * <li>128 is restricted to the same continent
43 * <li>255 is unrestricted
44 * </ul>
45 * You can also control how often the heartbeat sends by setting the interval.
46 *
47 * @author Greg Luck
48 * @version $Id: MulticastKeepaliveHeartbeatSender.java 512 2007-07-10 09:18:45Z gregluck $
49 */
50 public final class MulticastKeepaliveHeartbeatSender {
51
52
53 private static final Log LOG = LogFactory.getLog(MulticastKeepaliveHeartbeatSender.class.getName());
54
55 private static final int DEFAULT_HEARTBEAT_INTERVAL = 5000;
56 private static final int MINIMUM_HEARTBEAT_INTERVAL = 1000;
57 private static final int MAXIMUM_PEERS_PER_SEND = 150;
58
59 private static long heartBeatInterval = DEFAULT_HEARTBEAT_INTERVAL;
60
61 private final InetAddress groupMulticastAddress;
62 private final Integer groupMulticastPort;
63 private final Integer timeToLive;
64 private MulticastServerThread serverThread;
65 private boolean stopped;
66 private final CacheManager cacheManager;
67
68 /**
69 * Constructor.
70 *
71 * @param cacheManager the bound CacheManager. Each CacheManager has a maximum of one sender
72 * @param multicastAddress
73 * @param multicastPort
74 * @param timeToLive See class description for the meaning of this parameter.
75 */
76 public MulticastKeepaliveHeartbeatSender(CacheManager cacheManager,
77 InetAddress multicastAddress, Integer multicastPort,
78 Integer timeToLive) {
79 this.cacheManager = cacheManager;
80 this.groupMulticastAddress = multicastAddress;
81 this.groupMulticastPort = multicastPort;
82 this.timeToLive = timeToLive;
83
84 }
85
86 /**
87 * Start the heartbeat thread
88 */
89 public final void init() {
90 serverThread = new MulticastServerThread();
91 serverThread.start();
92 }
93
94 /**
95 * Shutdown this heartbeat sender
96 */
97 public final synchronized void dispose() {
98 stopped = true;
99 notifyAll();
100 serverThread.interrupt();
101 }
102
103 /**
104 * A thread which sends a multicast heartbeat every second
105 */
106 private final class MulticastServerThread extends Thread {
107
108 private MulticastSocket socket;
109 private List compressedUrlListList = new ArrayList();
110 private int cachePeersHash;
111
112
113 /**
114 * Constructor
115 */
116 public MulticastServerThread() {
117 super("Multicast Heartbeat Sender Thread");
118 setDaemon(true);
119 }
120
121 public final void run() {
122 while (!stopped) {
123 try {
124 socket = new MulticastSocket(groupMulticastPort.intValue());
125 socket.setTimeToLive(timeToLive.intValue());
126 socket.joinGroup(groupMulticastAddress);
127
128 while (!stopped) {
129 List buffers = createCachePeersPayload();
130 for (Iterator iter = buffers.iterator(); iter.hasNext();) {
131 byte[] buffer = (byte[]) iter.next();
132 DatagramPacket packet = new DatagramPacket(buffer, buffer.length, groupMulticastAddress,
133 groupMulticastPort.intValue());
134 socket.send(packet);
135 }
136 try {
137 synchronized (this) {
138 wait(heartBeatInterval);
139 }
140 } catch (InterruptedException e) {
141 if (!stopped) {
142 LOG.error("Error receiving heartbeat. Initial cause was " + e.getMessage(), e);
143 }
144 }
145 }
146 } catch (IOException e) {
147 LOG.debug(e);
148 } catch (Throwable e) {
149 LOG.info("Unexpected throwable in run thread. Continuing..." + e.getMessage(), e);
150 } finally {
151 closeSocket();
152 }
153 }
154 }
155
156 /**
157 * Creates a gzipped payload.
158 * <p/>
159 * The last gzipped payload is retained and only recalculated if the list of cache peers
160 * has changed.
161 *
162 * @return a gzipped byte[]
163 */
164 private List createCachePeersPayload() {
165 List localCachePeers = cacheManager.getCachePeerListener().getBoundCachePeers();
166 int newCachePeersHash = localCachePeers.hashCode();
167 if (cachePeersHash != newCachePeersHash) {
168 cachePeersHash = newCachePeersHash;
169
170 compressedUrlListList = new ArrayList();
171 while (localCachePeers.size() > 0) {
172 int endIndex = Math.min(localCachePeers.size(), MAXIMUM_PEERS_PER_SEND);
173 List localCachePeersSubList = localCachePeers.subList(0, endIndex);
174 localCachePeers = localCachePeers.subList(endIndex, localCachePeers.size());
175
176 byte[] uncompressedUrlList = PayloadUtil.assembleUrlList(localCachePeersSubList);
177 byte[] compressedUrlList = PayloadUtil.gzip(uncompressedUrlList);
178 if (compressedUrlList.length > PayloadUtil.MTU) {
179 LOG.fatal("Heartbeat is not working. Configure fewer caches for replication. " +
180 "Size is " + compressedUrlList.length + " but should be no greater than" +
181 PayloadUtil.MTU);
182 }
183 compressedUrlListList.add(compressedUrlList);
184 }
185 }
186 return compressedUrlListList;
187 }
188
189
190 /**
191 * Interrupts this thread.
192 * <p/>
193 * <p> Unless the current thread is interrupting itself, which is
194 * always permitted, the {@link #checkAccess() checkAccess} method
195 * of this thread is invoked, which may cause a {@link
196 * SecurityException} to be thrown.
197 * <p/>
198 * <p> If this thread is blocked in an invocation of the {@link
199 * Object#wait() wait()}, {@link Object#wait(long) wait(long)}, or {@link
200 * Object#wait(long,int) wait(long, int)} methods of the {@link Object}
201 * class, or of the {@link #join()}, {@link #join(long)}, {@link
202 * #join(long,int)}, {@link #sleep(long)}, or {@link #sleep(long,int)},
203 * methods of this class, then its interrupt status will be cleared and it
204 * will receive an {@link InterruptedException}.
205 * <p/>
206 * <p> If this thread is blocked in an I/O operation upon an {@link
207 * java.nio.channels.InterruptibleChannel </code>interruptible
208 * channel<code>} then the channel will be closed, the thread's interrupt
209 * status will be set, and the thread will receive a {@link
210 * java.nio.channels.ClosedByInterruptException}.
211 * <p/>
212 * <p> If this thread is blocked in a {@link java.nio.channels.Selector}
213 * then the thread's interrupt status will be set and it will return
214 * immediately from the selection operation, possibly with a non-zero
215 * value, just as if the selector's {@link
216 * java.nio.channels.Selector#wakeup wakeup} method were invoked.
217 * <p/>
218 * <p> If none of the previous conditions hold then this thread's interrupt
219 * status will be set. </p>
220 *
221 * @throws SecurityException if the current thread cannot modify this thread
222 */
223 public final void interrupt() {
224 closeSocket();
225 super.interrupt();
226 }
227
228 private void closeSocket() {
229 try {
230 if (socket != null && !socket.isClosed()) {
231 try {
232 socket.leaveGroup(groupMulticastAddress);
233 } catch (IOException e) {
234 LOG.error("Error leaving multicast group. Message was " + e.getMessage());
235 }
236 socket.close();
237 }
238 } catch (NoSuchMethodError e) {
239 LOG.debug("socket.isClosed is not supported by JDK1.3");
240 try {
241 socket.leaveGroup(groupMulticastAddress);
242 } catch (IOException ex) {
243 LOG.error("Error leaving multicast group. Message was " + ex.getMessage());
244 }
245 socket.close();
246 }
247 }
248
249 }
250
251 /**
252 * Sets the heartbeat interval to something other than the default of 5000ms. This is useful for testing,
253 * but not recommended for production. This method is static and so affects the heartbeat interval of all
254 * senders. The change takes effect after the next scheduled heartbeat.
255 *
256 * @param heartBeatInterval a time in ms, greater than 1000
257 */
258 public static void setHeartBeatInterval(long heartBeatInterval) {
259 if (heartBeatInterval < MINIMUM_HEARTBEAT_INTERVAL) {
260 LOG.warn("Trying to set heartbeat interval too low. Using MINIMUM_HEARTBEAT_INTERVAL instead.");
261 MulticastKeepaliveHeartbeatSender.heartBeatInterval = MINIMUM_HEARTBEAT_INTERVAL;
262 } else {
263 MulticastKeepaliveHeartbeatSender.heartBeatInterval = heartBeatInterval;
264 }
265 }
266
267 /**
268 * Returns the heartbeat interval.
269 */
270 public static long getHeartBeatInterval() {
271 return heartBeatInterval;
272 }
273
274 /**
275 * @return the TTL
276 */
277 public Integer getTimeToLive() {
278 return timeToLive;
279 }
280 }