001: // $Id: ReplicatedHashtable.java,v 1.14 2006/03/27 08:34:24 belaban Exp $
002:
003: package org.jgroups.blocks;
004:
005: import org.apache.commons.logging.Log;
006: import org.apache.commons.logging.LogFactory;
007: import org.jgroups.*;
008: import org.jgroups.util.Util;
009:
010: import java.io.Serializable;
011: import java.util.*;
012:
013: /**
014: * Provides the abstraction of a java.util.Hashtable that is replicated at several
015: * locations. Any change to the hashtable (clear, put, remove etc) will transparently be
016: * propagated to all replicas in the group. All read-only methods will always access the
017: * local replica.<p>
018: * Both keys and values added to the hashtable <em>must be serializable</em>, the reason
019: * being that they will be sent across the network to all replicas of the group. Having said
020: * this, it is now for example possible to add RMI remote objects to the hashtable as they
021: * are derived from <code>java.rmi.server.RemoteObject</code> which in turn is serializable.
022: * This allows to lookup shared distributed objects by their name and invoke methods on them,
023: * regardless of one's onw location. A <code>ReplicatedHashtable</code> thus allows to
024: * implement a distributed naming service in just a couple of lines.<p>
025: * An instance of this class will contact an existing member of the group to fetch its
026: * initial state.<p>
027: * Contrary to DistributedHashtable, this class does not make use of RpcDispatcher (and RequestCorrelator)
028: * but uses plain asynchronous messages instead.
029: * @author Bela Ban
030: * @author <a href="mailto:aolias@yahoo.com">Alfonso Olias-Sanz</a>
031: * @todo implement putAll() [similar to DistributedHashtable].
032: */
033: public class ReplicatedHashtable extends Hashtable implements
034: MessageListener, MembershipListener {
035:
036: public interface Notification {
037: void entrySet(Object key, Object value);
038:
039: void entryRemoved(Object key);
040:
041: void viewChange(Vector new_mbrs, Vector old_mbrs);
042:
043: void contentsSet(Map new_entries);
044: }
045:
046: public interface StateTransferListener {
047: void stateTransferStarted();
048:
049: void stateTransferCompleted(boolean success);
050: }
051:
052: transient Channel channel;
053: transient PullPushAdapter adapter = null;
054: final transient Vector notifs = new Vector();
055: // to be notified when mbrship changes
056: final transient Vector members = new Vector(); // keeps track of all DHTs
057: final transient List state_transfer_listeners = new ArrayList();
058: transient boolean state_transfer_running = false;
059:
060: /** Determines when the updates have to be sent across the network, avoids sending unnecessary
061: * messages when there are no member in the group */
062: private transient boolean send_message = false;
063:
064: protected final transient Log log = LogFactory.getLog(this
065: .getClass());
066:
067: /**
068: * Creates a ReplicatedHashtable
069: * @param groupname The name of the group to join
070: * @param factory The ChannelFactory which will be used to create a channel
071: * @param properties The property string to be used to define the channel
072: * @param state_timeout The time to wait until state is retrieved in milliseconds. A value of 0 means wait forever.
073: */
074: public ReplicatedHashtable(String groupname,
075: ChannelFactory factory, StateTransferListener l,
076: String properties, long state_timeout) {
077: if (l != null)
078: addStateTransferListener(l);
079: try {
080: channel = factory != null ? factory
081: .createChannel(properties) : new JChannel(
082: properties);
083: channel.connect(groupname);
084: adapter = new PullPushAdapter(channel, this , this );
085: adapter.setListener(this );
086: getInitState(channel, state_timeout);
087: } catch (Exception e) {
088: if (log.isErrorEnabled())
089: log.error("exception=" + e);
090: }
091: }
092:
093: private void getInitState(Channel channel, long state_timeout)
094: throws ChannelClosedException, ChannelNotConnectedException {
095: try {
096: notifyStateTransferStarted();
097: boolean rc = channel.getState(null, state_timeout);
098: if (rc) {
099: if (log.isInfoEnabled())
100: log.info("state was retrieved successfully");
101: } else {
102: if (log.isInfoEnabled())
103: log
104: .info("state could not be retrieved (first member)");
105: notifyStateTransferCompleted(false);
106: }
107: } catch (ChannelClosedException ex) {
108: notifyStateTransferCompleted(false);
109: throw ex;
110: } catch (ChannelNotConnectedException ex2) {
111: notifyStateTransferCompleted(false);
112: throw ex2;
113: }
114: }
115:
116: public ReplicatedHashtable(String groupname,
117: ChannelFactory factory, String properties,
118: long state_timeout) {
119: this (groupname, factory, null, properties, state_timeout);
120: }
121:
122: public ReplicatedHashtable(JChannel channel, long state_timeout)
123: throws ChannelClosedException, ChannelNotConnectedException {
124: this (channel, null, state_timeout);
125: }
126:
127: public ReplicatedHashtable(JChannel channel,
128: StateTransferListener l, long state_timeout)
129: throws ChannelClosedException, ChannelNotConnectedException {
130: this .channel = channel;
131: this .adapter = new PullPushAdapter(channel, this , this );
132: this .adapter.setListener(this );
133: if (l != null)
134: addStateTransferListener(l);
135: getInitState(channel, state_timeout);
136: // boolean rc=channel.getState(null, state_timeout);
137: // if(rc)
138: // if(log.isInfoEnabled()) log.info("state was retrieved successfully");
139: // else
140: // if(log.isInfoEnabled()) log.info("state could not be retrieved (first member)");
141: }
142:
143: public boolean stateTransferRunning() {
144: return state_transfer_running;
145: }
146:
147: public Address getLocalAddress() {
148: return channel != null ? channel.getLocalAddress() : null;
149: }
150:
151: public Channel getChannel() {
152: return channel;
153: }
154:
155: public void addNotifier(Notification n) {
156: if (!notifs.contains(n))
157: notifs.addElement(n);
158: }
159:
160: public final void addStateTransferListener(StateTransferListener l) {
161: if (l != null && !(state_transfer_listeners.contains(l)))
162: state_transfer_listeners.add(l);
163: }
164:
165: public void removeStateTransferListener(StateTransferListener l) {
166: if (l != null)
167: state_transfer_listeners.remove(l);
168: }
169:
170: /**
171: * Maps the specified key to the specified value in the hashtable. Neither of both parameters can be null
172: * @param key - the hashtable key
173: * @param value - the value
174: * @return the previous value of the specified key in this hashtable, or null if it did not have one
175: */
176: public Object put(Object key, Object value) {
177: Message msg;
178: Object prev_val = null;
179: prev_val = get(key);
180:
181: //Changes done by <aos>
182: //if true, send message to the group
183: if (send_message == true) {
184: try {
185: msg = new Message(null, null, new Request(Request.PUT,
186: key, value));
187: channel.send(msg);
188: //return prev_val;
189: } catch (Exception e) {
190: //return null;
191: }
192: } else {
193: super .put(key, value);
194: //don't have to do prev_val = super.put(..) as is done at the beginning
195: }
196: return prev_val;
197: }
198:
199: /**
200: * Copies all of the mappings from the specified Map to this Hashtable These mappings will replace any mappings that this Hashtable had for any of the keys currently in the specified Map.
201: * @param m - Mappings to be stored in this map
202: */
203: public void putAll(Map m) {
204: Message msg;
205: //Changes done by <aos>
206: //if true, send message to the group
207: if (send_message == true) {
208: try {
209: msg = new Message(null, null, new Request(
210: Request.PUT_ALL, null, m));
211: channel.send(msg);
212: } catch (Exception e) {
213: if (log.isErrorEnabled())
214: log.error("exception=" + e);
215: }
216: } else {
217: super .putAll(m);
218: }
219: }
220:
221: /**
222: * Clears this hashtable so that it contains no keys
223: */
224: public void clear() {
225: Message msg;
226: //Changes done by <aos>
227: //if true, send message to the group
228: if (send_message == true) {
229: try {
230: msg = new Message(null, null, new Request(
231: Request.CLEAR, null, null));
232: channel.send(msg);
233: } catch (Exception e) {
234: if (log.isErrorEnabled())
235: log.error("exception=" + e);
236: }
237: } else {
238: super .clear();
239: }
240: }
241:
242: /**
243: * Removes the key (and its corresponding value) from the Hashtable.
244: * @param key - the key to be removed.
245: * @return the value to which the key had been mapped in this hashtable, or null if the key did not have a mapping.
246: */
247: public Object remove(Object key) {
248: Message msg;
249: Object retval = null;
250: retval = get(key);
251:
252: //Changes done by <aos>
253: //if true, propagate action to the group
254: if (send_message == true) {
255: try {
256: msg = new Message(null, null, new Request(
257: Request.REMOVE, key, null));
258: channel.send(msg);
259: //return retval;
260: } catch (Exception e) {
261: //return null;
262: }
263: } else {
264: super .remove(key);
265: //don't have to do retval = super.remove(..) as is done at the beginning
266: }
267: return retval;
268: }
269:
270: /*------------------------ Callbacks -----------------------*/
271: Object _put(Object key, Object value) {
272: Object retval = super .put(key, value);
273: for (int i = 0; i < notifs.size(); i++)
274: ((Notification) notifs.elementAt(i)).entrySet(key, value);
275: return retval;
276: }
277:
278: void _clear() {
279: super .clear();
280: }
281:
282: Object _remove(Object key) {
283: Object retval = super .remove(key);
284: for (int i = 0; i < notifs.size(); i++)
285: ((Notification) notifs.elementAt(i)).entryRemoved(key);
286: return retval;
287: }
288:
289: /**
290: * @see java.util.Map#putAll(java.util.Map)
291: */
292: public void _putAll(Map m) {
293: if (m == null)
294: return;
295: //######## The same way as in the DistributedHashtable
296: // Calling the method below seems okay, but would result in ... deadlock !
297: // The reason is that Map.putAll() calls put(), which we override, which results in
298: // lock contention for the map.
299: // ---> super.putAll(m); <--- CULPRIT !!!@#$%$
300: // That said let's do it the stupid way:
301: //######## The same way as in the DistributedHashtable
302: Map.Entry entry;
303: for (Iterator it = m.entrySet().iterator(); it.hasNext();) {
304: entry = (Map.Entry) it.next();
305: super .put(entry.getKey(), entry.getValue());
306: }
307:
308: for (int i = 0; i < notifs.size(); i++)
309: ((Notification) notifs.elementAt(i)).contentsSet(m);
310: }
311:
312: /*----------------------------------------------------------*/
313:
314: /*-------------------- MessageListener ----------------------*/
315:
316: public void receive(Message msg) {
317: Request req = null;
318:
319: if (msg == null)
320: return;
321: req = (Request) msg.getObject();
322: if (req == null)
323: return;
324: switch (req.req_type) {
325: case Request.PUT:
326: if (req.key != null && req.val != null)
327: _put(req.key, req.val);
328: break;
329: case Request.REMOVE:
330: if (req.key != null)
331: _remove(req.key);
332: break;
333: case Request.CLEAR:
334: _clear();
335: break;
336:
337: case Request.PUT_ALL:
338: if (req.val != null)
339: _putAll((Map) req.val);
340: break;
341: default:
342: // error
343: }
344: }
345:
346: public byte[] getState() {
347: Object key, val;
348: Hashtable copy = new Hashtable();
349:
350: for (Enumeration e = keys(); e.hasMoreElements();) {
351: key = e.nextElement();
352: val = get(key);
353: copy.put(key, val);
354: }
355: try {
356: return Util.objectToByteBuffer(copy);
357: } catch (Exception ex) {
358: if (log.isErrorEnabled())
359: log.error("exception marshalling state: " + ex);
360: return null;
361: }
362: }
363:
364: public void setState(byte[] new_state) {
365: Hashtable new_copy;
366: Object key;
367:
368: try {
369: new_copy = (Hashtable) Util.objectFromByteBuffer(new_state);
370: if (new_copy == null) {
371: notifyStateTransferCompleted(true);
372: return;
373: }
374: } catch (Throwable ex) {
375: if (log.isErrorEnabled())
376: log.error("exception unmarshalling state: " + ex);
377: notifyStateTransferCompleted(false);
378: return;
379: }
380:
381: _clear(); // remove all elements
382: for (Enumeration e = new_copy.keys(); e.hasMoreElements();) {
383: key = e.nextElement();
384: _put(key, new_copy.get(key));
385: }
386: notifyStateTransferCompleted(true);
387: }
388:
389: /*-------------------- End of MessageListener ----------------------*/
390:
391: /*----------------------- MembershipListener ------------------------*/
392:
393: public void viewAccepted(View new_view) {
394: Vector new_mbrs = new_view.getMembers();
395:
396: if (new_mbrs != null) {
397: sendViewChangeNotifications(new_mbrs, members);
398: // notifies observers (joined, left)
399: members.removeAllElements();
400: for (int i = 0; i < new_mbrs.size(); i++)
401: members.addElement(new_mbrs.elementAt(i));
402: }
403: //if size is bigger than one, there are more peers in the group
404: //otherwise there is only one server.
405: if (members.size() > 1) {
406: send_message = true;
407: } else {
408: send_message = false;
409: }
410: }
411:
412: /** Called when a member is suspected */
413: public void suspect(Address suspected_mbr) {
414: ;
415: }
416:
417: /** Block sending and receiving of messages until ViewAccepted is called */
418: public void block() {
419: }
420:
421: /*------------------- End of MembershipListener ----------------------*/
422:
423: void sendViewChangeNotifications(Vector new_mbrs, Vector old_mbrs) {
424: Vector joined, left;
425: Object mbr;
426: Notification n;
427:
428: if (notifs.size() == 0 || old_mbrs == null || new_mbrs == null
429: || old_mbrs.size() == 0 || new_mbrs.size() == 0)
430: return;
431:
432: // 1. Compute set of members that joined: all that are in new_mbrs, but not in old_mbrs
433: joined = new Vector();
434: for (int i = 0; i < new_mbrs.size(); i++) {
435: mbr = new_mbrs.elementAt(i);
436: if (!old_mbrs.contains(mbr))
437: joined.addElement(mbr);
438: }
439:
440: // 2. Compute set of members that left: all that were in old_mbrs, but not in new_mbrs
441: left = new Vector();
442: for (int i = 0; i < old_mbrs.size(); i++) {
443: mbr = old_mbrs.elementAt(i);
444: if (!new_mbrs.contains(mbr)) {
445: left.addElement(mbr);
446: }
447: }
448:
449: for (int i = 0; i < notifs.size(); i++) {
450: n = (Notification) notifs.elementAt(i);
451: n.viewChange(joined, left);
452: }
453: }
454:
455: void notifyStateTransferStarted() {
456: state_transfer_running = true;
457: for (Iterator it = state_transfer_listeners.iterator(); it
458: .hasNext();) {
459: StateTransferListener listener = (StateTransferListener) it
460: .next();
461: try {
462: listener.stateTransferStarted();
463: } catch (Throwable t) {
464: }
465: }
466: }
467:
468: void notifyStateTransferCompleted(boolean success) {
469: state_transfer_running = false;
470: for (Iterator it = state_transfer_listeners.iterator(); it
471: .hasNext();) {
472: StateTransferListener listener = (StateTransferListener) it
473: .next();
474: try {
475: listener.stateTransferCompleted(success);
476: } catch (Throwable t) {
477: }
478: }
479: }
480:
481: private static class Request implements Serializable {
482: static final int PUT = 1;
483: static final int REMOVE = 2;
484: static final int CLEAR = 3;
485: static final int PUT_ALL = 4;
486:
487: int req_type = 0;
488: Object key = null;
489: Object val = null;
490:
491: Request(int req_type, Object key, Object val) {
492: this .req_type = req_type;
493: this .key = key;
494: this .val = val;
495: }
496:
497: public String toString() {
498: StringBuffer sb = new StringBuffer();
499: sb.append(type2String(req_type));
500: if (key != null)
501: sb.append("\nkey=" + key);
502: if (val != null)
503: sb.append("\nval=" + val);
504: return sb.toString();
505: }
506:
507: String type2String(int t) {
508: switch (t) {
509: case PUT:
510: return "PUT";
511: case REMOVE:
512: return "REMOVE";
513: case CLEAR:
514: return "CLEAR";
515: case PUT_ALL:
516: return "PUT_ALL";
517: default:
518: return "<unknown>";
519: }
520: }
521:
522: }
523: }
|