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.CacheManager;
21 import net.sf.ehcache.Ehcache;
22 import org.apache.commons.logging.Log;
23 import org.apache.commons.logging.LogFactory;
24
25 import java.io.IOException;
26 import java.net.InetAddress;
27 import java.rmi.NotBoundException;
28 import java.util.ArrayList;
29 import java.util.Date;
30 import java.util.Iterator;
31 import java.util.List;
32
33 /**
34 * A peer provider which discovers peers using Multicast.
35 * <p/>
36 * Hosts can be in three different levels of conformance with the Multicast specification (RFC1112), according to the requirements they meet.
37 * <ol>
38 * <li>Level 0 is the "no support for IP Multicasting" level. Lots of hosts and routers in the Internet are in this state,
39 * as multicast support is not mandatory in IPv4 (it is, however, in IPv6).
40 * Not too much explanation is needed here: hosts in this level can neither send nor receive multicast packets.
41 * They must ignore the ones sent by other multicast capable hosts.
42 * <li>Level 1 is the "support for sending but not receiving multicast IP datagrams" level.
43 * Thus, note that it is not necessary to join a multicast group to be able to send datagrams to it.
44 * Very few additions are needed in the IP module to make a "Level 0" host "Level 1-compliant".
45 * <li>Level 2 is the "full support for IP multicasting" level.
46 * Level 2 hosts must be able to both send and receive multicast traffic.
47 * They must know the way to join and leave multicast groups and to propagate this information to multicast routers.
48 * Thus, they must include an Internet Group Management Protocol (IGMP) implementation in their TCP/IP stack.
49 * </ol>
50 * <p/>
51 * The list of CachePeers is maintained via heartbeats. rmiUrls are looked up using RMI and converted to CachePeers on
52 * registration. On lookup any stale references are removed.
53 *
54 * @author Greg Luck
55 * @version $Id: MulticastRMICacheManagerPeerProvider.java 512 2007-07-10 09:18:45Z gregluck $
56 */
57 public final class MulticastRMICacheManagerPeerProvider extends RMICacheManagerPeerProvider implements CacheManagerPeerProvider {
58
59 /**
60 * One second, in ms
61 */
62 protected static final int SHORT_DELAY = 100;
63
64 private static final Log LOG = LogFactory.getLog(MulticastRMICacheManagerPeerProvider.class.getName());
65
66
67 private final MulticastKeepaliveHeartbeatReceiver heartBeatReceiver;
68 private final MulticastKeepaliveHeartbeatSender heartBeatSender;
69
70 /**
71 * Creates and starts a multicast peer provider
72 *
73 * @param groupMulticastAddress 224.0.0.1 to 239.255.255.255 e.g. 230.0.0.1
74 * @param groupMulticastPort 1025 to 65536 e.g. 4446
75 */
76 public MulticastRMICacheManagerPeerProvider(CacheManager cacheManager, InetAddress groupMulticastAddress,
77 Integer groupMulticastPort, Integer timeToLive) {
78 super(cacheManager);
79 heartBeatReceiver = new MulticastKeepaliveHeartbeatReceiver(this, groupMulticastAddress, groupMulticastPort);
80 heartBeatSender =
81 new MulticastKeepaliveHeartbeatSender(cacheManager, groupMulticastAddress, groupMulticastPort, timeToLive);
82 }
83
84 /**
85 * {@inheritDoc}
86 */
87 public final void init() throws CacheException {
88 try {
89 heartBeatReceiver.init();
90 heartBeatSender.init();
91 } catch (IOException exception) {
92 LOG.error("Error starting heartbeat. Error was: " + exception.getMessage(), exception);
93 throw new CacheException(exception.getMessage());
94 }
95 }
96
97 /**
98 * Register a new peer, but only if the peer is new, otherwise the last seen timestamp is updated.
99 * <p/>
100 * This method is thread-safe. It relies on peerUrls being a synchronizedMap
101 *
102 * @param rmiUrl
103 */
104 public final void registerPeer(String rmiUrl) {
105 try {
106 CachePeerEntry cachePeerEntry = (CachePeerEntry) peerUrls.get(rmiUrl);
107 if (cachePeerEntry == null || stale(cachePeerEntry.date)) {
108 //can take seconds if there is a problem
109 CachePeer cachePeer = lookupRemoteCachePeer(rmiUrl);
110 cachePeerEntry = new CachePeerEntry(cachePeer, new Date());
111 //synchronized due to peerUrls being a synchronizedMap
112 peerUrls.put(rmiUrl, cachePeerEntry);
113 } else {
114 cachePeerEntry.date = new Date();
115 }
116 } catch (IOException e) {
117 if (LOG.isDebugEnabled()) {
118 LOG.debug("Unable to lookup remote cache peer for " + rmiUrl + ". Removing from peer list. Cause was: "
119 + e.getMessage());
120 }
121 unregisterPeer(rmiUrl);
122 } catch (NotBoundException e) {
123 peerUrls.remove(rmiUrl);
124 if (LOG.isDebugEnabled()) {
125 LOG.debug("Unable to lookup remote cache peer for " + rmiUrl + ". Removing from peer list. Cause was: "
126 + e.getMessage());
127 }
128 } catch (Throwable t) {
129 LOG.error("Unable to lookup remote cache peer for " + rmiUrl
130 + ". Cause was not due to an IOException or NotBoundException which will occur in normal operation:" +
131 " " + t.getMessage());
132 }
133 }
134
135 /**
136 * @return a list of {@link CachePeer} peers, excluding the local peer.
137 */
138 public final synchronized List listRemoteCachePeers(Ehcache cache) throws CacheException {
139 List remoteCachePeers = new ArrayList();
140 List staleList = new ArrayList();
141 synchronized (peerUrls) {
142 for (Iterator iterator = peerUrls.keySet().iterator(); iterator.hasNext();) {
143 String rmiUrl = (String) iterator.next();
144 String rmiUrlCacheName = extractCacheName(rmiUrl);
145 try {
146 if (!rmiUrlCacheName.equals(cache.getName())) {
147 continue;
148 }
149 CachePeerEntry cachePeerEntry = (CachePeerEntry) peerUrls.get(rmiUrl);
150 Date date = cachePeerEntry.date;
151 if (!stale(date)) {
152 CachePeer cachePeer = cachePeerEntry.cachePeer;
153 remoteCachePeers.add(cachePeer);
154 } else {
155 if (LOG.isDebugEnabled()) {
156 LOG.debug("rmiUrl " + rmiUrl + " is stale. Either the remote peer is shutdown or the " +
157 "network connectivity has been interrupted. Will be removed from list of remote cache peers");
158 }
159 staleList.add(rmiUrl);
160 }
161 } catch (Exception exception) {
162 LOG.error(exception.getMessage(), exception);
163 throw new CacheException("Unable to list remote cache peers. Error was " + exception.getMessage());
164 }
165 }
166 //Must remove entries after we have finished iterating over them
167 for (int i = 0; i < staleList.size(); i++) {
168 String rmiUrl = (String) staleList.get(i);
169 peerUrls.remove(rmiUrl);
170 }
171 }
172 return remoteCachePeers;
173 }
174
175
176 /**
177 * Shutdown the heartbeat
178 */
179 public final void dispose() {
180 heartBeatSender.dispose();
181 heartBeatReceiver.dispose();
182 }
183
184 /**
185 * Time for a cluster to form. This varies considerably, depending on the implementation.
186 *
187 * @return the time in ms, for a cluster to form
188 */
189 public long getTimeForClusterToForm() {
190 return getStaleTime();
191 }
192
193 /**
194 * The time after which an unrefreshed peer provider entry is considered stale.
195 */
196 protected long getStaleTime() {
197 return MulticastKeepaliveHeartbeatSender.getHeartBeatInterval() * 2 + SHORT_DELAY;
198 }
199
200 /**
201 * Whether the entry should be considered stale.
202 * This will depend on the type of RMICacheManagerPeerProvider.
203 * This method should be overridden for implementations that go stale based on date
204 *
205 * @param date the date the entry was created
206 * @return true if stale
207 */
208 protected final boolean stale(Date date) {
209 long now = System.currentTimeMillis();
210 return date.getTime() < (now - getStaleTime());
211 }
212
213
214 /**
215 * Entry containing a looked up CachePeer and date
216 */
217 protected static final class CachePeerEntry {
218
219 private final CachePeer cachePeer;
220 private Date date;
221
222 /**
223 * Constructor
224 *
225 * @param cachePeer the cache peer part of this entry
226 * @param date the date part of this entry
227 */
228 public CachePeerEntry(CachePeer cachePeer, Date date) {
229 this.cachePeer = cachePeer;
230 this.date = date;
231 }
232
233 /**
234 * @return the cache peer part of this entry
235 */
236 public final CachePeer getCachePeer() {
237 return cachePeer;
238 }
239
240
241 /**
242 * @return the date part of this entry
243 */
244 public final Date getDate() {
245 return date;
246 }
247
248 }
249
250 /**
251 * @return the MulticastKeepaliveHeartbeatReceiver
252 */
253 public MulticastKeepaliveHeartbeatReceiver getHeartBeatReceiver() {
254 return heartBeatReceiver;
255 }
256
257 /**
258 * @return the MulticastKeepaliveHeartbeatSender
259 */
260 public MulticastKeepaliveHeartbeatSender getHeartBeatSender() {
261 return heartBeatSender;
262 }
263 }