001: // $Id: DistributedHashtable.java,v 1.26.2.2 2007/04/26 16:43:00 vlada 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.persistence.CannotPersistException;
009: import org.jgroups.persistence.CannotRemoveException;
010: import org.jgroups.persistence.PersistenceFactory;
011: import org.jgroups.persistence.PersistenceManager;
012: import org.jgroups.util.Promise;
013: import org.jgroups.util.Util;
014:
015: import java.io.Serializable;
016: import java.util.*;
017:
018: /**
019: * Provides the abstraction of a java.util.Hashtable that is replicated at several
020: * locations. Any change to the hashtable (clear, put, remove etc) will transparently be
021: * propagated to all replicas in the group. All read-only methods will always access the
022: * local replica.<p>
023: * Both keys and values added to the hashtable <em>must be serializable</em>, the reason
024: * being that they will be sent across the network to all replicas of the group. Having said
025: * this, it is now for example possible to add RMI remote objects to the hashtable as they
026: * are derived from <code>java.rmi.server.RemoteObject</code> which in turn is serializable.
027: * This allows to lookup shared distributed objects by their name and invoke methods on them,
028: * regardless of one's onw location. A <code>DistributedHashtable</code> thus allows to
029: * implement a distributed naming service in just a couple of lines.<p>
030: * An instance of this class will contact an existing member of the group to fetch its
031: * initial state (using the state exchange funclet <code>StateExchangeFunclet</code>.
032: * @author Bela Ban
033: * @author <a href="mailto:aolias@yahoo.com">Alfonso Olias-Sanz</a>
034: * @version $Id: DistributedHashtable.java,v 1.26.2.2 2007/04/26 16:43:00 vlada Exp $
035: */
036: public class DistributedHashtable extends Hashtable implements
037: MessageListener, MembershipListener {
038:
039: public interface Notification {
040: void entrySet(Object key, Object value);
041:
042: void entryRemoved(Object key);
043:
044: void viewChange(Vector new_mbrs, Vector old_mbrs);
045:
046: void contentsSet(Map new_entries);
047:
048: void contentsCleared();
049: }
050:
051: private transient Channel channel;
052: protected transient RpcDispatcher disp = null;
053: private transient String groupname = null;
054: private final transient Vector notifs = new Vector(); // to be notified when mbrship changes
055: private final transient Vector members = new Vector(); // keeps track of all DHTs
056: private transient Class[] put_signature = null;
057: private transient Class[] putAll_signature = null;
058: private transient Class[] clear_signature = null;
059: private transient Class[] remove_signature = null;
060: private transient boolean persistent = false; // whether to use PersistenceManager to save state
061: private transient PersistenceManager persistence_mgr = null;
062:
063: /** Determines when the updates have to be sent across the network, avoids sending unnecessary
064: * messages when there are no member in the group */
065: private transient boolean send_message = false;
066:
067: protected final transient Promise state_promise = new Promise();
068:
069: protected final Log log = LogFactory.getLog(this .getClass());
070:
071: /**
072: * Creates a DistributedHashtable
073: * @param groupname The name of the group to join
074: * @param factory The ChannelFactory which will be used to create a channel
075: * @param properties The property string to be used to define the channel. This will override the properties of
076: * the factory. If null, then the factory properties will be used
077: * @param state_timeout The time to wait until state is retrieved in milliseconds. A value of 0 means wait forever.
078: */
079: public DistributedHashtable(String groupname,
080: ChannelFactory factory, String properties,
081: long state_timeout) throws ChannelException {
082: this .groupname = groupname;
083: initSignatures();
084: if (factory != null) {
085: channel = properties != null ? factory
086: .createChannel(properties) : factory
087: .createChannel();
088: } else {
089: channel = new JChannel(properties);
090: }
091: disp = new RpcDispatcher(channel, this , this , this );
092: channel.connect(groupname);
093: start(state_timeout);
094: }
095:
096: /**
097: * Creates a DisttributedHashtable. Optionally the contents can be saved to
098: * persistemt storage using the {@link PersistenceManager}.
099: * @param groupname Name of the group to join
100: * @param factory Instance of a ChannelFactory to create the channel
101: * @param properties Protocol stack properties. This will override the properties of the factory. If
102: * null, then the factory properties will be used
103: * @param persistent Whether the contents should be persisted
104: * @param state_timeout Max number of milliseconds to wait until state is
105: * retrieved
106: */
107: public DistributedHashtable(String groupname,
108: ChannelFactory factory, String properties,
109: boolean persistent, long state_timeout)
110: throws ChannelException {
111: this .groupname = groupname;
112: this .persistent = persistent;
113: initSignatures();
114: if (factory != null) {
115: channel = properties != null ? factory
116: .createChannel(properties) : factory
117: .createChannel();
118: } else {
119: channel = new JChannel(properties);
120: }
121: disp = new RpcDispatcher(channel, this , this , this );
122: channel.connect(groupname);
123: start(state_timeout);
124: }
125:
126: public DistributedHashtable(Channel channel, long state_timeout) {
127: this (channel, false, state_timeout);
128: }
129:
130: public DistributedHashtable(Channel channel, boolean persistent,
131: long state_timeout) {
132: this .groupname = channel.getClusterName();
133: this .channel = channel;
134: this .persistent = persistent;
135: init(state_timeout);
136: }
137:
138: /**
139: * Uses a user-provided PullPushAdapter to create the dispatcher rather than a Channel. If id is non-null, it will be
140: * used to register under that id. This is typically used when another building block is already using
141: * PullPushAdapter, and we want to add this building block in addition. The id is the used to discriminate
142: * between messages for the various blocks on top of PullPushAdapter. If null, we will assume we are the
143: * first block created on PullPushAdapter.
144: * @param adapter The PullPushAdapter which to use as underlying transport
145: * @param id A serializable object (e.g. an Integer) used to discriminate (multiplex/demultiplex) between
146: * requests/responses for different building blocks on top of PullPushAdapter.
147: * @param state_timeout Max number of milliseconds to wait until state is
148: * retrieved
149: */
150: public DistributedHashtable(PullPushAdapter adapter,
151: Serializable id, long state_timeout)
152: throws ChannelNotConnectedException, ChannelClosedException {
153: initSignatures();
154: this .channel = (Channel) adapter.getTransport();
155: this .groupname = this .channel.getClusterName();
156: disp = new RpcDispatcher(adapter, id, this , this , this );
157: start(state_timeout);
158: }
159:
160: public DistributedHashtable(PullPushAdapter adapter, Serializable id) {
161: initSignatures();
162: this .channel = (Channel) adapter.getTransport();
163: this .groupname = this .channel.getClusterName();
164: disp = new RpcDispatcher(adapter, id, this , this , this );
165: }
166:
167: protected final void init(long state_timeout) {
168: initSignatures();
169: disp = new RpcDispatcher(channel, this , this , this );
170:
171: // Changed by bela (jan 20 2003): start() has to be called by user (only when providing
172: // own channel). First, Channel.connect() has to be called, then start().
173: // start(state_timeout);
174: }
175:
176: /**
177: * Fetches the state
178: * @param state_timeout
179: * @throws ChannelClosedException
180: * @throws ChannelNotConnectedException
181: */
182: public final void start(long state_timeout)
183: throws ChannelClosedException, ChannelNotConnectedException {
184: boolean rc;
185: if (persistent) {
186: if (log.isInfoEnabled())
187: log.info("fetching state from database");
188: try {
189: persistence_mgr = PersistenceFactory.getInstance()
190: .createManager();
191: } catch (Throwable ex) {
192: if (log.isErrorEnabled())
193: log.error("failed creating PersistenceManager, "
194: + "turning persistency off. Exception: "
195: + Util.printStackTrace(ex));
196: persistent = false;
197: }
198: }
199:
200: state_promise.reset();
201: rc = channel.getState(null, state_timeout);
202: if (rc) {
203: if (log.isInfoEnabled())
204: log
205: .info("state was retrieved successfully, waiting for setState()");
206: Boolean result = (Boolean) state_promise
207: .getResult(state_timeout);
208: if (result == null) {
209: if (log.isErrorEnabled())
210: log.error("setState() never got called");
211: } else {
212: if (log.isInfoEnabled())
213: log.info("setState() was called");
214: }
215: } else {
216: if (log.isInfoEnabled())
217: log.info("state could not be retrieved (first member)");
218: if (persistent) {
219: if (log.isInfoEnabled())
220: log.info("fetching state from database");
221: try {
222: Map m = persistence_mgr.retrieveAll();
223: if (m != null) {
224: Map.Entry entry;
225: Object key, val;
226: for (Iterator it = m.entrySet().iterator(); it
227: .hasNext();) {
228: entry = (Map.Entry) it.next();
229: key = entry.getKey();
230: val = entry.getValue();
231: if (log.isInfoEnabled())
232: log.info("inserting " + key + " --> "
233: + val);
234: put(key, val); // will replicate key and value
235: }
236: }
237: } catch (Throwable ex) {
238: if (log.isErrorEnabled())
239: log
240: .error("failed creating PersistenceManager, "
241: + "turning persistency off. Exception: "
242: + Util.printStackTrace(ex));
243: persistent = false;
244: }
245: }
246: }
247: }
248:
249: public Address getLocalAddress() {
250: return channel != null ? channel.getLocalAddress() : null;
251: }
252:
253: public String getGroupName() {
254: return groupname;
255: }
256:
257: public Channel getChannel() {
258: return channel;
259: }
260:
261: public boolean getPersistent() {
262: return persistent;
263: }
264:
265: public void setPersistent(boolean p) {
266: persistent = p;
267: }
268:
269: public void setDeadlockDetection(boolean flag) {
270: if (disp != null)
271: disp.setDeadlockDetection(flag);
272: }
273:
274: public void addNotifier(Notification n) {
275: if (!notifs.contains(n))
276: notifs.addElement(n);
277: }
278:
279: public void removeNotifier(Notification n) {
280: if (notifs.contains(n))
281: notifs.removeElement(n);
282: }
283:
284: public void stop() {
285: if (disp != null) {
286: disp.stop();
287: disp = null;
288: }
289: if (channel != null) {
290: channel.close();
291: channel = null;
292: }
293: }
294:
295: /**
296: * Maps the specified key to the specified value in the hashtable. Neither of both parameters can be null
297: * @param key - the hashtable key
298: * @param value - the value
299: * @return the previous value of the specified key in this hashtable, or null if it did not have one
300: */
301: public Object put(Object key, Object value) {
302: Object prev_val = get(key);
303:
304: //Changes done by <aos>
305: //if true, propagate action to the group
306: if (send_message == true) {
307: try {
308: disp.callRemoteMethods(null, "_put", new Object[] {
309: key, value }, put_signature,
310: GroupRequest.GET_ALL, 0);
311: } catch (Exception e) {
312: //return null;
313: }
314: } else {
315: _put(key, value);
316: //don't have to do prev_val = super.put(..) as is done at the beginning
317: }
318: return prev_val;
319: }
320:
321: /**
322: * 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.
323: * @param m - Mappings to be stored in this map
324: */
325: public void putAll(Map m) {
326: //Changes done by <aos>
327: //if true, propagate action to the group
328: if (send_message == true) {
329: try {
330: disp.callRemoteMethods(null, "_putAll",
331: new Object[] { m }, putAll_signature,
332: GroupRequest.GET_ALL, 0);
333: } catch (Throwable t) {
334: }
335: } else {
336: _putAll(m);
337: }
338: }
339:
340: /**
341: * Clears this hashtable so that it contains no keys
342: */
343: public void clear() {
344: //Changes done by <aos>
345: //if true, propagate action to the group
346: if (send_message == true) {
347: try {
348: disp.callRemoteMethods(null, "_clear", null,
349: clear_signature, GroupRequest.GET_ALL, 0);
350: } catch (Exception e) {
351: if (log.isErrorEnabled())
352: log.error("exception=" + e);
353: }
354: } else {
355: _clear();
356: }
357: }
358:
359: /**
360: * Removes the key (and its corresponding value) from the Hashtable.
361: * @param key - the key to be removed.
362: * @return the value to which the key had been mapped in this hashtable, or null if the key did not have a mapping.
363: */
364: public Object remove(Object key) {
365: Object retval = get(key);
366:
367: //Changes done by <aos>
368: //if true, propagate action to the group
369: if (send_message == true) {
370: try {
371: disp.callRemoteMethods(null, "_remove",
372: new Object[] { key }, remove_signature,
373: GroupRequest.GET_ALL, 0);
374: //return retval;
375: } catch (Exception e) {
376: //return null;
377: }
378: } else {
379: _remove(key);
380: //don't have to do retval = super.remove(..) as is done at the beginning
381: }
382: return retval;
383: }
384:
385: /*------------------------ Callbacks -----------------------*/
386:
387: public Object _put(Object key, Object value) {
388: Object retval = super .put(key, value);
389: if (persistent) {
390: try {
391: persistence_mgr.save((Serializable) key,
392: (Serializable) value);
393: } catch (CannotPersistException cannot_persist_ex) {
394: if (log.isErrorEnabled())
395: log.error("failed persisting " + key + " + "
396: + value + ", exception="
397: + cannot_persist_ex);
398: } catch (Throwable t) {
399: if (log.isErrorEnabled())
400: log.error("failed persisting " + key + " + "
401: + value + ", exception="
402: + Util.printStackTrace(t));
403: }
404: }
405: for (int i = 0; i < notifs.size(); i++)
406: ((Notification) notifs.elementAt(i)).entrySet(key, value);
407: return retval;
408: }
409:
410: /**
411: * @see java.util.Map#putAll(java.util.Map)
412: */
413: public void _putAll(Map m) {
414: if (m == null)
415: return;
416:
417: // Calling the method below seems okay, but would result in ... deadlock !
418: // The reason is that Map.putAll() calls put(), which we override, which results in
419: // lock contention for the map.
420:
421: // ---> super.putAll(m); <--- CULPRIT !!!@#$%$
422:
423: // That said let's do it the stupid way:
424: Map.Entry entry;
425: for (Iterator it = m.entrySet().iterator(); it.hasNext();) {
426: entry = (Map.Entry) it.next();
427: super .put(entry.getKey(), entry.getValue());
428: }
429:
430: if (persistent) {
431: try {
432: persistence_mgr.saveAll(m);
433: } catch (CannotPersistException persist_ex) {
434: if (log.isErrorEnabled())
435: log.error("failed persisting contents: "
436: + persist_ex);
437: } catch (Throwable t) {
438: if (log.isErrorEnabled())
439: log.error("failed persisting contents: " + t);
440: }
441: }
442: for (int i = 0; i < notifs.size(); i++)
443: ((Notification) notifs.elementAt(i)).contentsSet(m);
444: }
445:
446: public void _clear() {
447: super .clear();
448: if (persistent) {
449: try {
450: persistence_mgr.clear();
451: } catch (CannotRemoveException cannot_remove_ex) {
452: if (log.isErrorEnabled())
453: log.error("failed clearing contents, exception="
454: + cannot_remove_ex);
455: } catch (Throwable t) {
456: if (log.isErrorEnabled())
457: log.error("failed clearing contents, exception="
458: + t);
459: }
460: }
461: for (int i = 0; i < notifs.size(); i++)
462: ((Notification) notifs.elementAt(i)).contentsCleared();
463: }
464:
465: public Object _remove(Object key) {
466: Object retval = super .remove(key);
467: if (persistent) {
468: try {
469: persistence_mgr.remove((Serializable) key);
470: } catch (CannotRemoveException cannot_remove_ex) {
471: if (log.isErrorEnabled())
472: log.error("failed clearing contents, exception="
473: + cannot_remove_ex);
474: } catch (Throwable t) {
475: if (log.isErrorEnabled())
476: log.error("failed clearing contents, exception="
477: + t);
478: }
479: }
480: for (int i = 0; i < notifs.size(); i++)
481: ((Notification) notifs.elementAt(i)).entryRemoved(key);
482:
483: return retval;
484: }
485:
486: /*----------------------------------------------------------*/
487:
488: /*-------------------- State Exchange ----------------------*/
489:
490: public void receive(Message msg) {
491: }
492:
493: public byte[] getState() {
494: Object key, val;
495: Hashtable copy = new Hashtable();
496:
497: for (Enumeration e = keys(); e.hasMoreElements();) {
498: key = e.nextElement();
499: val = get(key);
500: copy.put(key, val);
501: }
502: try {
503: return Util.objectToByteBuffer(copy);
504: } catch (Throwable ex) {
505: if (log.isErrorEnabled())
506: log.error("exception marshalling state: " + ex);
507: return null;
508: }
509: }
510:
511: public void setState(byte[] new_state) {
512: Hashtable new_copy;
513:
514: try {
515: new_copy = (Hashtable) Util.objectFromByteBuffer(new_state);
516: if (new_copy == null)
517: return;
518: } catch (Throwable ex) {
519: if (log.isErrorEnabled())
520: log.error("exception unmarshalling state: " + ex);
521: return;
522: }
523: _putAll(new_copy);
524: state_promise.setResult(Boolean.TRUE);
525: }
526:
527: /*------------------- Membership Changes ----------------------*/
528:
529: public void viewAccepted(View new_view) {
530: Vector new_mbrs = new_view.getMembers();
531:
532: if (new_mbrs != null) {
533: sendViewChangeNotifications(new_mbrs, members); // notifies observers (joined, left)
534: members.removeAllElements();
535: for (int i = 0; i < new_mbrs.size(); i++)
536: members.addElement(new_mbrs.elementAt(i));
537: }
538: //if size is bigger than one, there are more peers in the group
539: //otherwise there is only one server.
540: send_message = members.size() > 1;
541: }
542:
543: /** Called when a member is suspected */
544: public void suspect(Address suspected_mbr) {
545: ;
546: }
547:
548: /** Block sending and receiving of messages until ViewAccepted is called */
549: public void block() {
550: }
551:
552: void sendViewChangeNotifications(Vector new_mbrs, Vector old_mbrs) {
553: Vector joined, left;
554: Object mbr;
555: Notification n;
556:
557: if (notifs.size() == 0 || old_mbrs == null || new_mbrs == null
558: || old_mbrs.size() == 0 || new_mbrs.size() == 0)
559: return;
560:
561: // 1. Compute set of members that joined: all that are in new_mbrs, but not in old_mbrs
562: joined = new Vector();
563: for (int i = 0; i < new_mbrs.size(); i++) {
564: mbr = new_mbrs.elementAt(i);
565: if (!old_mbrs.contains(mbr))
566: joined.addElement(mbr);
567: }
568:
569: // 2. Compute set of members that left: all that were in old_mbrs, but not in new_mbrs
570: left = new Vector();
571: for (int i = 0; i < old_mbrs.size(); i++) {
572: mbr = old_mbrs.elementAt(i);
573: if (!new_mbrs.contains(mbr)) {
574: left.addElement(mbr);
575: }
576: }
577:
578: for (int i = 0; i < notifs.size(); i++) {
579: n = (Notification) notifs.elementAt(i);
580: n.viewChange(joined, left);
581: }
582: }
583:
584: final void initSignatures() {
585: try {
586: if (put_signature == null) {
587: put_signature = new Class[] { Object.class,
588: Object.class };
589: }
590:
591: if (putAll_signature == null) {
592: putAll_signature = new Class[] { Map.class };
593: }
594:
595: if (clear_signature == null)
596: clear_signature = new Class[0];
597:
598: if (remove_signature == null) {
599: remove_signature = new Class[] { Object.class };
600: }
601: } catch (Throwable ex) {
602: if (log.isErrorEnabled())
603: log.error("exception=" + ex);
604: }
605: }
606:
607: public static void main(String[] args) {
608: try {
609: // The setup here is kind of weird:
610: // 1. Create a channel
611: // 2. Create a DistributedHashtable (on the channel)
612: // 3. Connect the channel (so the HT gets a VIEW_CHANGE)
613: // 4. Start the HT
614: //
615: // A simpler setup is
616: // DistributedHashtable ht = new DistributedHashtable("demo", null,
617: // "file://c:/JGroups-2.0/conf/state_transfer.xml", 5000);
618:
619: JChannel c = new JChannel(
620: "file:/c:/JGroups-2.0/conf/state_transfer.xml");
621: DistributedHashtable ht = new DistributedHashtable(c,
622: false, 5000);
623: c.connect("demo");
624: ht.start(5000);
625:
626: ht.put("name", "Michelle Ban");
627: Object old_key = ht.remove("name");
628: System.out.println("old key was " + old_key);
629: ht.put("newkey", "newvalue");
630:
631: Map m = new HashMap();
632: m.put("k1", "v1");
633: m.put("k2", "v2");
634:
635: ht.putAll(m);
636:
637: System.out.println("hashmap is " + ht);
638: } catch (Throwable t) {
639: t.printStackTrace();
640: }
641: }
642:
643: }
|