|
1
7 package org.jboss.cache;
8
9 import org.jboss.cache.eviction.LRUPolicy;
10 import org.jboss.cache.interceptors.Interceptor;
11 import org.jboss.cache.loader.CacheLoader;
12 import org.jboss.cache.lock.IsolationLevel;
13 import org.jboss.cache.lock.LockStrategyFactory;
14 import org.jboss.cache.lock.LockingException;
15 import org.jboss.cache.lock.TimeoutException;
16 import org.jboss.logging.Logger;
17 import org.jboss.system.ServiceMBeanSupport;
18 import org.jboss.util.NestedRuntimeException;
19 import org.jgroups.*;
20 import org.jgroups.blocks.GroupRequest;
21 import org.jgroups.blocks.MethodCall;
22 import org.jgroups.blocks.RpcDispatcher;
23 import org.jgroups.util.Rsp;
24 import org.jgroups.util.RspList;
25 import org.jgroups.util.Util;
26 import org.w3c.dom.Attr  ;
27 import org.w3c.dom.Element ;
28 import org.w3c.dom.NamedNodeMap ;
29 import org.w3c.dom.NodeList ;
30
31 import javax.transaction.Status ;
32 import javax.transaction.SystemException ;
33 import javax.transaction.Transaction ;
34 import javax.transaction.TransactionManager ;
35 import java.lang.reflect.Method ;
36 import java.util.*;
37
38
50 public class TreeCache extends ServiceMBeanSupport implements TreeCacheMBean, Cloneable , MembershipListener {
51 protected Node root=new Node(SEPARATOR, Fqn.fromString(SEPARATOR), null, null, this);
52 protected final Vector listeners=new Vector();
53 protected JChannel channel=null;
54
55
56 protected boolean coordinator=false;
57
58 protected String cluster_name="TreeCache-Group";
59 protected String cluster_props=null;
60 protected final Vector members=new Vector();
61 protected RpcDispatcher disp=null;
62 protected MessageListener ml=new MessageListenerAdaptor(this, log);
63 protected long state_fetch_timeout=5000;
64 protected long sync_repl_timeout=15000;
65 protected boolean use_repl_queue=false;
66 protected int repl_queue_max_elements=1000;
67 protected long repl_queue_interval=5000;
68
69
70 private final TransactionTable tx_table=new TransactionTable();
71
72
73 private final HashMap lock_table=new HashMap();
74
75 protected boolean fetch_state_on_startup=true;
76 protected long lock_acquisition_timeout=10000;
77 protected String eviction_policy_class=null;
78 protected TreeCacheListener eviction_policy_provider = null;
79 protected int cache_mode=LOCAL;
80
81
82 public static Method putDataMethodLocal=null;
83 public static Method putDataEraseMethodLocal=null;
84 public static Method putKeyValMethodLocal=null;
85 public static Method putFailFastKeyValueMethodLocal=null;
86 public static Method removeNodeMethodLocal=null;
87 public static Method removeKeyMethodLocal=null;
88 public static Method removeDataMethodLocal=null;
89 public static Method evictNodeMethodLocal=null;
90 public static Method prepareMethod=null;
92 public static Method commitMethod=null;
93 public static Method rollbackMethod=null;
94 public static Method replicateMethod=null;
95 public static Method replicateAllMethod=null;
96 public static Method addChildMethodLocal=null;
98 public static Method getKeyValueMethodLocal=null;
99 public static Method getNodeMethodLocal=null;
100 public static Method getKeysMethodLocal=null;
101 public static Method getChildrenNamesMethodLocal=null;
102 public static Method releaseAllLocksMethodLocal=null;
103 public static Method printMethodLocal=null;
104 public static Method lockMethodLocal=null;
105
106 static LinkedList crud_methods=new LinkedList();
107 protected boolean isStateSet=false;
108 private final Object stateLock=new Object ();
109 protected IsolationLevel isolationLevel=IsolationLevel.REPEATABLE_READ;
110
111
112 protected Element evictConfig_ = null;
113
114
115
116 public MessageListener getMessageListener() {
117 return ml;
118 }
119
120
123 protected Interceptor interceptor_chain=null;
124
125
126
132 protected Replicatable replication_handler=null;
133
134
135
137 protected TransactionManagerLookup tm_lookup=null;
138
139
140 protected String tm_lookup_class=null;
141
142
143 protected TransactionManager tm=null;
144
145
146 protected String cache_loader_class=null;
147
148
149 protected CacheLoader cache_loader=null;
150
151
152 protected Properties cache_loader_config=null;
153
154
155 protected boolean cache_loader_shared=true;
156
157
158 protected List cache_loader_preload=null;
159
160
161 protected boolean cache_loader_fetch_transient_state=true;
162
163
165 protected boolean cache_loader_fetch_persistent_state=true;
166
167
168 protected boolean sync_commit_phase=false;
169
170
171 protected boolean sync_rollback_phase=false;
172
173 protected boolean deadlockDetection=false;
174
175
176 protected ReplicationQueue repl_queue=null;
177
178 public static final String SEPARATOR="/";
179
180
181 public static final int LOCAL=1;
182
183
184 public static final int REPL_ASYNC=2;
185
186
187 public static final int REPL_SYNC=3;
188
189 static public final String UNINITIALIZED="jboss:internal:uninitialized"; static final String JNDI_LOCATOR_URI="socket://localhost:6789";
191
192
193 static {
194 try {
195 putDataMethodLocal=TreeCache.class.getDeclaredMethod("_put",
196 new Class []{GlobalTransaction.class,
197 Fqn.class,
198 Map .class,
199 boolean.class});
200 putDataEraseMethodLocal=TreeCache.class.getDeclaredMethod("_put",
201 new Class []{GlobalTransaction.class,
202 Fqn.class,
203 Map .class,
204 boolean.class,
205 boolean.class});
206 putKeyValMethodLocal=TreeCache.class.getDeclaredMethod("_put",
207 new Class []{GlobalTransaction.class,
208 Fqn.class,
209 Object .class,
210 Object .class,
211 boolean.class});
212 putFailFastKeyValueMethodLocal=TreeCache.class.getDeclaredMethod("_put",
213 new Class []{GlobalTransaction.class,
214 Fqn.class,
215 Object .class,
216 Object .class,
217 boolean.class,
218 long.class});
219 removeNodeMethodLocal=TreeCache.class.getDeclaredMethod("_remove",
220 new Class []{GlobalTransaction.class,
221 Fqn.class,
222 boolean.class});
223 removeKeyMethodLocal=TreeCache.class.getDeclaredMethod("_remove",
224 new Class []{GlobalTransaction.class,
225 Fqn.class,
226 Object .class,
227 boolean.class});
228 removeDataMethodLocal=TreeCache.class.getDeclaredMethod("_removeData",
229 new Class []{GlobalTransaction.class,
230 Fqn.class,
231 boolean.class});
232 evictNodeMethodLocal=TreeCache.class.getDeclaredMethod("_evict", new Class [] {Fqn.class});
233 prepareMethod=TreeCache.class.getDeclaredMethod("prepare",
235 new Class []{GlobalTransaction.class,
236 List.class,
237 Address.class,
238 boolean.class});
239 commitMethod=TreeCache.class.getDeclaredMethod("commit",
240 new Class []{GlobalTransaction.class});
241 rollbackMethod=TreeCache.class.getDeclaredMethod("rollback",
242 new Class []{GlobalTransaction.class});
243 addChildMethodLocal=TreeCache.class.getDeclaredMethod("_addChild",
244 new Class []{GlobalTransaction.class,
245 Fqn.class, Object .class, Node.class});
246 getKeyValueMethodLocal=TreeCache.class.getDeclaredMethod("_get",
247 new Class []{Fqn.class, Object .class, boolean.class});
248 getNodeMethodLocal=TreeCache.class.getDeclaredMethod("_get", new Class []{Fqn.class});
249 getKeysMethodLocal=TreeCache.class.getDeclaredMethod("_getKeys", new Class []{Fqn.class});
250 getChildrenNamesMethodLocal=TreeCache.class.getDeclaredMethod("_getChildrenNames", new Class []{Fqn.class});
251 replicateMethod=TreeCache.class.getDeclaredMethod("_replicate", new Class []{MethodCall.class});
252 replicateAllMethod=TreeCache.class.getDeclaredMethod("_replicate", new Class []{List.class});
253 releaseAllLocksMethodLocal=TreeCache.class.getDeclaredMethod("_releaseAllLocks", new Class []{Fqn.class});
254 printMethodLocal=TreeCache.class.getDeclaredMethod("_print", new Class []{Fqn.class});
255 lockMethodLocal=TreeCache.class.getDeclaredMethod("_lock", new Class []{Fqn.class,
256 int.class,
257 boolean.class});
258 }
259 catch(NoSuchMethodException ex) {
260 ex.printStackTrace();
261 throw new ExceptionInInitializerError (ex.toString());
262 }
263
264 crud_methods.add(putDataMethodLocal);
265 crud_methods.add(putDataEraseMethodLocal);
266 crud_methods.add(putKeyValMethodLocal);
267 crud_methods.add(putFailFastKeyValueMethodLocal);
268 crud_methods.add(removeNodeMethodLocal);
269 crud_methods.add(removeKeyMethodLocal);
270 crud_methods.add(removeDataMethodLocal);
271 }
272
273
274
275
276 public static boolean isCrudMethod(Method m) {
277 return m == null? false : crud_methods.contains(m);
278 }
279
280
281
282
286 public TreeCache(String cluster_name, String props, long state_fetch_timeout) throws Exception {
287 super();
288 if(cluster_name != null)
289 this.cluster_name=cluster_name;
290 if(props != null)
291 this.cluster_props=props;
292 this.state_fetch_timeout=state_fetch_timeout;
293 }
294
295 public TreeCache() throws Exception {
296 super();
297 }
304
305
308 public TreeCache(JChannel channel) throws Exception {
309 super();
310 this.channel=channel;
311 }
312
313
318 public Node getRoot() {
319 return root;
320 }
321
322
326 public Object getLocalAddress() {
327 return channel != null ? channel.getLocalAddress() : null;
328 }
329
330
334 public Vector getMembers() {
335 return members;
336 }
337
338
343 public boolean isCoordinator() {
344 return coordinator;
345 }
346
347
351 public String getClusterName() {
352 return cluster_name;
353 }
354
355
359 public void setClusterName(String name) {
360 cluster_name=name;
361 }
362
363
367 public String getClusterProperties() {
368 return cluster_props;
369 }
370
371
376 public void setClusterProperties(String cluster_props) {
377 this.cluster_props=cluster_props;
378 }
379
380
381 public TransactionTable getTransactionTable() {
382 return tx_table;
383 }
384
385
386 public HashMap getLockTable() {
387 return lock_table;
388 }
389
390
395 public String dumpTransactionTable() {
396 return tx_table.toString(true);
397 }
398
399
404 public boolean getDeadlockDetection() {
405 return deadlockDetection;
406 }
407
408
413 public void setDeadlockDetection(boolean dt) {
414 deadlockDetection=dt;
415 if(disp != null)
416 disp.setDeadlockDetection(dt);
417 }
418
419
423 public String getInterceptorChain() {
424 String retval=printInterceptorChain(interceptor_chain);
425 if(retval == null || retval.length() == 0)
426 return "<empty>";
427 else
428 return retval;
429 }
430
431
432
436 public List getInterceptors() {
437 if(interceptor_chain == null)
438 return null;
439 int num=1;
440 Interceptor tmp=interceptor_chain;
441 while((tmp=tmp.getNext()) != null) {
442 num++;
443 }
444 List retval=new ArrayList(num);
445 tmp=interceptor_chain;
446 num=0;
447 do {
448 retval.add(tmp);
449 tmp=tmp.getNext();
450 }
451 while(tmp != null);
452 return retval;
453 }
454
455
456
457
458
462 public String getCacheLoaderClass() {
463 return cache_loader_class;
464 }
465
466
471 public void setCacheLoaderClass(String cache_loader_class) {
472 this.cache_loader_class=cache_loader_class;
473 }
474
475
479 public Properties getCacheLoaderConfig() {
480 return cache_loader_config;
481 }
482
483
487 public void setCacheLoaderConfig(Properties cache_loader_config) {
488 this.cache_loader_config=cache_loader_config;
489 }
490
491
495 public CacheLoader getCacheLoader() {
496 return cache_loader;
497 }
498
499
503 public void setCacheLoader(CacheLoader cache_loader) {
504 this.cache_loader=cache_loader;
505 }
506
507
511 public boolean getCacheLoaderShared() {
512 return cache_loader_shared;
513 }
514
515
519 public void setCacheLoaderShared(boolean shared) {
520 this.cache_loader_shared=shared;
521 }
522
523
524
528 public void setCacheLoaderPreload(String list) {
529 if(list == null) return;
530 ArrayList l;
531 StringTokenizer st=new StringTokenizer(list, ",");
532 String tok;
533 Fqn fqn;
534 l=new ArrayList();
535 while(st.hasMoreTokens()) {
536 tok=st.nextToken();
537 fqn=Fqn.fromString(tok.trim());
538 l.add(fqn);
539 }
540 if(l.size() > 0)
541 this.cache_loader_preload=l;
542 }
543
544
545
546
550 public String getCacheLoaderPreload() {
551 return cache_loader_preload != null? cache_loader_preload.toString() : null;
552 }
553
554
558 public void setCacheLoaderFetchPersistentState(boolean flag) {
559 cache_loader_fetch_persistent_state=flag;
560 }
561
562
566 public boolean getCacheLoaderFetchPersistentState() {
567 return cache_loader_fetch_persistent_state;
568 }
569
570
574 public void setCacheLoaderFetchTransientState(boolean flag) {
575 cache_loader_fetch_transient_state=flag;
576 }
577
578
582 public boolean getCacheLoaderFetchTransientState() {
583 return cache_loader_fetch_transient_state;
584 }
585
586
587
591 public boolean getSyncCommitPhase() {
592 return sync_commit_phase;
593 }
594
595
599 public void setSyncCommitPhase(boolean sync_commit_phase) {
600 this.sync_commit_phase=sync_commit_phase;
601 }
602
603
607 public boolean getSyncRollbackPhase() {
608 return sync_rollback_phase;
609 }
610
611
615 public void setSyncRollbackPhase(boolean sync_rollback_phase) {
616 this.sync_rollback_phase=sync_rollback_phase;
617 }
618
619
620
624 public void setEvictionPolicyConfig(Element config) {
625 evictConfig_ = config;
626 log.info("setEvictionPolicyConfig(): " +config);
627 }
628
629 public Element getEvictionPolicyConfig() {
630 return evictConfig_;
631 }
632
633
637 public void setClusterConfig(Element config) {
638 StringBuffer buffer=new StringBuffer ();
639 NodeList stack=config.getChildNodes();
640 int length=stack.getLength();
641
642 for(int s=0; s < length; s++) {
643 org.w3c.dom.Node node=stack.item(s);
644 if(node.getNodeType() != org.w3c.dom.Node.ELEMENT_NODE)
645 continue;
646
647 Element tag=(Element )node;
648 String protocol=tag.getTagName();
649 buffer.append(protocol);
650 NamedNodeMap attrs=tag.getAttributes();
651 int attrLength=attrs.getLength();
652 if(attrLength > 0)
653 buffer.append('(');
654 for(int a=0; a < attrLength; a++) {
655 Attr attr=(Attr )attrs.item(a);
656 String name=attr.getName();
657 String value=attr.getValue();
658 buffer.append(name);
659 buffer.append('=');
660 buffer.append(value);
661 if(a < attrLength - 1)
662 buffer.append(';');
663 }
664 if(attrLength > 0)
665 buffer.append(')');
666 buffer.append(':');
667 }
668 buffer.setLength(buffer.length() - 1);
670 setClusterProperties(buffer.toString());
671 log.info("setting cluster properties from xml to: " + cluster_props);
672 }
673
674
675
676
677
687 public long getInitialStateRetrievalTimeout() {
688 return state_fetch_timeout;
689 }
690
691
696 public void setInitialStateRetrievalTimeout(long timeout) {
697 state_fetch_timeout=timeout;
698 }
699
700
710 public String getCacheMode() {
711 return mode2String(cache_mode);
712 }
713
714 public int getCacheModeInternal() {
715 return cache_mode;
716 }
717
718 private String mode2String(int mode) {
719 switch(mode) {
720 case LOCAL:
721 return "LOCAL";
722 case REPL_ASYNC:
723 return "REPL_ASYNC";
724 case REPL_SYNC:
725 return "REPL_SYNC";
726 default:
727 throw new RuntimeException ("setCacheMode(): caching mode " + mode + " is invalid");
728 }
729 }
730
731
735 public void setCacheMode(String mode) throws Exception {
736 int m=string2Mode(mode);
737 setCacheMode(m);
738 }
739
740
741
750 public void setCacheMode(int mode) {
751 if(mode == LOCAL || mode == REPL_ASYNC || mode == REPL_SYNC)
752 this.cache_mode=mode;
753 else
754 throw new IllegalArgumentException ("setCacheMode(): caching mode " + mode + " is invalid");
755 }
756
757
758
763 public long getSyncReplTimeout() {
764 return sync_repl_timeout;
765 }
766
767
771 public void setSyncReplTimeout(long timeout) {
772 sync_repl_timeout=timeout;
773 }
774
775
779 public boolean getUseReplQueue() {
780 return use_repl_queue;
781 }
782
783
787 public void setUseReplQueue(boolean flag) {
788 use_repl_queue=flag;
789 if(flag) {
790 if(repl_queue == null) {
791 repl_queue=new ReplicationQueue(this, repl_queue_interval, repl_queue_max_elements);
792 if(repl_queue_interval >= 0)
793 repl_queue.start();
794 }
795 }
796 else {
797 if(repl_queue != null) {
798 repl_queue.stop();
799 repl_queue=null;
800 }
801 }
802 }
803
804
808 public long getReplQueueInterval() {
809 return repl_queue_interval;
810 }
811
812
816 public void setReplQueueInterval(long interval) {
817 this.repl_queue_interval=interval;
818 if(repl_queue != null)
819 repl_queue.setInterval(interval);
820 }
821
822
826 public int getReplQueueMaxElements() {
827 return repl_queue_max_elements;
828 }
829
830
834 public void setReplQueueMaxElements(int max_elements) {
835 this.repl_queue_max_elements=max_elements;
836 if(repl_queue != null)
837 repl_queue.setMax_elements(max_elements);
838 }
839
840
841 public ReplicationQueue getReplQueue() {
842 return repl_queue;
843 }
844
845
849 public String getIsolationLevel() {
850 return isolationLevel.toString();
851 }
852
853
857 public void setIsolationLevel(String level) {
858 IsolationLevel tmp_level=IsolationLevel.stringToIsolationLevel(level);
859
860 if(tmp_level == null) {
861 throw new IllegalArgumentException ("TreeCache.setIsolationLevel(): level \"" + level + "\" is invalid");
862 }
863 setIsolationLevel(tmp_level);
864 }
865
866
869 public void setIsolationLevel(IsolationLevel level) {
870 isolationLevel=level;
871 LockStrategyFactory.setIsolationLevel(level);
872 }
873
874 public IsolationLevel getIsolationLevelClass() {
875 return isolationLevel;
876 }
877
878
882 public boolean getFetchStateOnStartup() {
883 return fetch_state_on_startup;
884 }
885
886
887
891 public void setFetchStateOnStartup(boolean flag) {
892 fetch_state_on_startup=flag;
893 }
894
895
896
901 public long getLockAcquisitionTimeout() {
902 return lock_acquisition_timeout;
903 }
904
905
911 public void setLockAcquisitionTimeout(long timeout) {
912 this.lock_acquisition_timeout=timeout;
913 }
914
915
916
917
922 public String getEvictionPolicyClass() {
923 return eviction_policy_class;
924 }
925
926
930 public void setEvictionPolicyClass(String eviction_policy_class) {
931 if(eviction_policy_class == null || eviction_policy_class.length() ==0)
932 return;
933 try {
934 this.eviction_policy_class=eviction_policy_class;
935 eviction_policy_provider =(TreeCacheListener)
936 getClass().getClassLoader().loadClass(eviction_policy_class).newInstance();
937 this.addTreeCacheListener(eviction_policy_provider );
938 }
939 catch(Throwable t) {
940 log.error("setEvictionPolicyClass(): failed creating instance of " + eviction_policy_class, t);
941 }
942 }
943
944
948 public int getEvictionThreadWakeupIntervalSeconds() {
949 if( eviction_policy_provider == null ) return -1;
950 else
951 return ((LRUPolicy)eviction_policy_provider).getWakeupIntervalSeconds();
952 }
953
954
955
960 public void setTransactionManagerLookup(TransactionManagerLookup l) {
961 this.tm_lookup=l;
962 }
963
964
965
969 public String getTransactionManagerLookupClass() {
970 return tm_lookup_class;
971 }
972
973
980 public void setTransactionManagerLookupClass(String cl) throws Exception {
981 this.tm_lookup_class=cl;
982 }
983
984
988 public TransactionManager getTransactionManager() {
989 return tm;
990 }
991
992
996 public TreeCache getInstance() {
997 return this;
998 }
999
1000
1001
1002 public void setReplicationHandler(Replicatable handler) {
1003 replication_handler=handler;
1004 }
1005
1006 public Replicatable getReplicationHandler() {
1007 return replication_handler;
1008 }
1009
1010
1014 public void fetchState(long timeout) throws ChannelClosedException, ChannelNotConnectedException {
1015 if(channel == null)
1016 throw new ChannelNotConnectedException();
1017 boolean rc=channel.getState(null, timeout);
1018 if(rc)
1019 log.info("fetchState(): state was retrieved successfully");
1020 else
1021 log.info("fetchState(): state could not be retrieved (first member)");
1022 }
1023
1024
1028 public void addTreeCacheListener(TreeCacheListener listener) {
1029 if(!listeners.contains(listener))
1030 listeners.addElement(listener);
1031 }
1032
1033
1037 public void removeTreeCacheListener(TreeCacheListener listener) {
1038 listeners.removeElement(listener);
1039 }
1040
1041
1042
1043
1048 public void createService() throws Exception {
1049 }
1050
1051
1054 public void destroyService() {
1055 }
1056
1057
1058
1063 public void startService() throws Exception {
1064 if(this.tm_lookup == null && this.tm_lookup_class != null) {
1065 Class clazz=Thread.currentThread().getContextClassLoader().loadClass(this.tm_lookup_class);
1066 this.tm_lookup=(TransactionManagerLookup)clazz.newInstance();
1067 }
1068
1069 try {
1070 if(tm_lookup != null)
1071 tm=tm_lookup.getTransactionManager();
1072 else
1073 log.warn("No transaction manager lookup class has been defined. Transactions cannot be used");
1074 }
1075 catch(Exception e) {
1076 log.debug("failed looking up TransactionManager, will not use transactions", e);
1077 }
1078
1079 createCacheLoader();
1082
1083 createInterceptorChain();
1085
1086 createEvictionPolicy();
1087
1088 switch(cache_mode) {
1089 case LOCAL:
1090 log.info("cache mode is local, will not create the channel");
1091 break;
1092 case REPL_SYNC:
1093 case REPL_ASYNC:
1094 log.info("cache mode is " + mode2String(cache_mode));
1095 if(channel != null) { log.info("channel is already running");
1097 return;
1098 }
1099 if(cluster_props == null) {
1100 cluster_props=getDefaultProperties();
1101 log.debug("setting cluster properties to default value");
1102 }
1103
1104 channel=new JChannel(cluster_props);
1105 channel.setOpt(Channel.AUTO_RECONNECT, Boolean.TRUE);
1106 channel.setOpt(Channel.AUTO_GETSTATE, Boolean.TRUE);
1107 if(log.isTraceEnabled())
1108 log.trace("cache properties: " + cluster_props);
1109 channel.setOpt(Channel.GET_STATE_EVENTS, Boolean.TRUE);
1110 disp=new RpcDispatcher(channel, ml, this, this);
1111 disp.setDeadlockDetection(deadlockDetection);
1112 channel.connect(cluster_name);
1113 if(fetch_state_on_startup) {
1114 fetchStateOnStartup();
1115 }
1116 break;
1117 default:
1118 throw new IllegalArgumentException ("cache mode " + cache_mode + " is invalid");
1119 }
1120
1121 cacheLoaderPreload();
1123
1124 coordinator=determineCoordinator();
1125 notifyCacheStarted();
1126 }
1127
1128 private void createEvictionPolicy() {
1129 if(eviction_policy_provider != null)
1131 ((LRUPolicy)eviction_policy_provider).configure(this);
1132 }
1133
1134
1158 protected void createInterceptorChain() throws IllegalAccessException , InstantiationException , ClassNotFoundException {
1159 Interceptor call_interceptor=null;
1160 Interceptor lock_interceptor=null;
1161 Interceptor repl_interceptor=null;
1163 Interceptor cache_loader_interceptor=null;
1164 Interceptor cache_store_interceptor=null;
1165 Interceptor unlock_interceptor=null;
1166 Interceptor first=null;
1167
1168 call_interceptor=createInterceptor("org.jboss.cache.interceptors.CallInterceptor");
1169 call_interceptor.setCache(this);
1170
1171 lock_interceptor=createInterceptor("org.jboss.cache.interceptors.LockInterceptor");
1172 lock_interceptor.setCache(this);
1173
1174
1177 unlock_interceptor=createInterceptor("org.jboss.cache.interceptors.UnlockInterceptor");
1178 unlock_interceptor.setCache(this);
1179
1180 if(cache_mode != LOCAL) {
1181 repl_interceptor=createInterceptor("org.jboss.cache.interceptors.ReplicationInterceptor");
1182 repl_interceptor.setCache(this);
1183 }
1184
1185 if(cache_loader_class != null || cache_loader != null) {
1186 cache_loader_interceptor=createInterceptor("org.jboss.cache.interceptors.CacheLoaderInterceptor");
1187 cache_loader_interceptor.setCache(this);
1188 cache_store_interceptor=createInterceptor("org.jboss.cache.interceptors.CacheStoreInterceptor");
1189 cache_store_interceptor.setCache(this);
1190 }
1191
1192
1193 if(cache_loader_interceptor != null) {
1195 if(cache_loader_shared == true) {
1196 if(first == null)
1197 first=cache_store_interceptor;
1198 else
1199 addInterceptor(first, cache_store_interceptor);
1200 }
1201 }
1202
1203 if(repl_interceptor != null) {
1204 if(first == null)
1205 first=repl_interceptor;
1206 else
1207 addInterceptor(first, repl_interceptor);
1208 }
1209
1210 if(unlock_interceptor != null) {
1211 if(first == null)
1212 first=unlock_interceptor;
1213 else
1214 addInterceptor(first, unlock_interceptor);
1215 }
1216
1217 if(cache_loader_interceptor != null) {
1218 if(cache_loader_shared == true) {
1219 if(first == null)
1220 first=cache_loader_interceptor;
1221 else
1222 addInterceptor(first, cache_loader_interceptor);
1223 }
1224 else {
1225 if(first == null)
1226 first=cache_loader_interceptor;
1227 else
1228 addInterceptor(first, cache_loader_interceptor);
1229 if(first == null)
1230 first=cache_store_interceptor;
1231 else
1232 addInterceptor(first, cache_store_interceptor);
1233 }
1234 }
1235
1236
1241 if(first == null)
1242 first=lock_interceptor;
1243 else
1244 addInterceptor(first, lock_interceptor);
1245
1246 if(first == null)
1247 first=call_interceptor;
1248 else
1249 addInterceptor(first, call_interceptor);
1250
1251 interceptor_chain=first;
1252 if(log.isInfoEnabled())
1253 log.info("interceptor chain is:\n" + printInterceptorChain(first));
1254 }
1255
1256
1257
1258
1259 private String printInterceptorChain(Interceptor i) {
1260 StringBuffer sb=new StringBuffer ();
1261 if(i != null) {
1262 if(i.getNext() != null) {
1263 sb.append(printInterceptorChain(i.getNext())).append("\n");
1264 }
1265 sb.append(i.getClass());
1266 }
1267 return sb.toString();
1268 }
1269
1270
1271
1272 private void addInterceptor(Interceptor first, Interceptor i) {
1273 if(first == null) {
1274 return;
1275 }
1276 do {
1277 if(first.getNext() != null)
1278 first=first.getNext();
1279 else
1280 break;
1281 }
1282 while(first != null); first.setNext(i);
1284 }
1285
1286
1287 private Interceptor createInterceptor(String classname) throws ClassNotFoundException , IllegalAccessException , InstantiationException {
1288 Class clazz=getClass().getClassLoader().loadClass(classname);
1289 return (Interceptor)clazz.newInstance();
1290 }
1291
1292
1293
1295
1304 protected void createCacheLoader() throws Exception {
1305 if(cache_loader == null && cache_loader_class != null) {
1306 Class cl=Thread.currentThread().getContextClassLoader().loadClass(cache_loader_class);
1307 cache_loader=(CacheLoader)cl.newInstance();
1308 cache_loader.setConfig(cache_loader_config);
1309 cache_loader.setCache(this);
1310 cache_loader.create();
1311 cache_loader.start();
1312 }
1313 }
1314
1315 protected void cacheLoaderPreload() throws Exception {
1316 if(cache_loader != null) {
1317 if(log.isTraceEnabled())
1318 log.trace("preloading " + cache_loader_preload);
1319 if(cache_loader_preload != null) {
1320 for(Iterator it=cache_loader_preload.iterator(); it.hasNext();) {
1321 Fqn fqn=(Fqn)it.next();
1322 preload(fqn, true, true);
1323 }
1324 }
1325 }
1326 }
1327
1328
1329
1336 public void load(String fqn) throws Exception {
1337 if(cache_loader != null)
1338 preload(Fqn.fromString(fqn), true, true);
1339 }
1340
1341 void preload(Fqn fqn, boolean preload_parents, boolean preload_children) throws Exception {
1342
1343 this.get(fqn, "bla");
1345
1346 if(preload_parents) {
1348 Fqn tmp_fqn=new Fqn();
1349 for(int i=0; i < fqn.size()-1; i++) {
1350 tmp_fqn=new Fqn(tmp_fqn, fqn.get(i));
1351 this.get(tmp_fqn, "bla");
1352 }
1353 }
1354
1355 if(preload_children == false)
1356 return;
1357
1358 Set children=cache_loader.getChildrenNames(fqn);
1360 if(children == null)
1361 return;
1362 for(Iterator it=children.iterator(); it.hasNext();) {
1363 String child_name=(String )it.next();
1364 Fqn child_fqn=new Fqn(fqn, child_name);
1365 preload(child_fqn, false, true);
1366 }
1367 }
1368
1369 void destroyCacheLoader() {
1370 if(cache_loader != null) {
1371 cache_loader.stop();
1372 cache_loader.destroy();
1373 cache_loader=null;
1374 }
1375 }
1376
1377 protected boolean determineCoordinator() {
1378 if(channel == null)
1379 return false;
1380 Object local_addr=getLocalAddress();
1381 if(local_addr == null)
1382 return false;
1383 View view=channel.getView();
1384 if(view == null) return false;
1385 ViewId vid=view.getVid();
1386 if(vid == null) return false;
1387 Object coord=vid.getCoordAddress();
1388 if(coord == null) return false;
1389 return local_addr.equals(coord);
1390 }
1391
1392 public Address getCoordinator() {
1393 if(channel == null) return null;
1394 View view=channel.getView();
1395 if(view == null) return null;
1396 ViewId vid=view.getVid();
1397 if(vid == null) return null;
1398 Address coord=vid.getCoordAddress();
1399 return coord;
1400 }
1401
1402 public byte[] getStateBytes() {
1403 return this.getMessageListener().getState();
1404 }
1405
1406 public void setStateBytes(byte[] state) {
1407 this.getMessageListener().setState(state);
1408 }
1409
1410 protected void fetchStateOnStartup() throws Exception {
1411 long start, stop;
1412 synchronized(stateLock) {
1413 isStateSet=false;
1414 start=System.currentTimeMillis();
1415 boolean rc=channel.getState(null, state_fetch_timeout);
1416 if(rc) {
1417 while(!isStateSet) {
1418 try {
1419 stateLock.wait();
1420 }
1421 catch(InterruptedException iex) {
1422 }
1423 }
1424 stop=System.currentTimeMillis();
1425 log.info("state was retrieved successfully (in " + (stop-start) + " milliseconds)");
1426 }
1427 else
1428 log.info("state could not be retrieved (must be first member in group)");
1429 }
1430 }
1431
1432
1435 public void stopService() {
1436 if(channel != null) {
1437 log.info("stopService(): closing the channel");
1438 channel.close();
1439 channel=null;
1440 }
1441 if(disp != null) {
1442 log.info("stopService(): stopping the dispatcher");
1443 disp.stop();
1444 disp=null;
1445 }
1446 if(members != null && members.size() > 0)
1447 members.clear();
1448
1449 if(repl_queue != null)
1450 repl_queue.stop();
1451
1452 destroyCacheLoader();
1453
1454 notifyCacheStopped();
1455
1456 listeners.clear();
1458 }
1459
1460
1461
1462
1467 public Node get(String fqn) throws CacheException {
1468 return get(Fqn.fromString(fqn));
1469 }
1470
1471
1476 public Node get(Fqn fqn) throws CacheException {
1477 MethodCall m=new MethodCall(getNodeMethodLocal, new Object []{fqn});
1478 return (Node)invokeMethod(m);
1479 }
1480
1481 public Node _get(Fqn fqn) throws CacheException {
1482 return findNode(fqn);
1483 }
1484
1485
1486
1491 public Set getKeys(String fqn) throws CacheException {
1492 return getKeys(Fqn.fromString(fqn));
1493 }
1494
1495
1501 public Set getKeys(Fqn fqn) throws CacheException {
1502 MethodCall m=new MethodCall(getKeysMethodLocal, new Object []{fqn});
1503 return (Set)invokeMethod(m);
1504 }
1505
1506
1507 public Set _getKeys(Fqn fqn) throws CacheException {
1508 Set retval=null;
1509 Node n=findNode(fqn);
1510 if(n == null)
1511 return null;
1512 retval=n.getDataKeys();
1513 return retval != null? new LinkedHashSet(retval) : null;
1514 }
1515
1516
1524 public Object get(String fqn, Object key) throws CacheException {
1525 return get(Fqn.fromString(fqn), key);
1526 }
1527
1528
1529
1537 public Object get(Fqn fqn, Object key) throws CacheException {
1538 return get(fqn, key, true);
1539 }
1540
1541 public Object _get(Fqn fqn, Object key, boolean sendNodeEvent) throws CacheException {
1542 if(log.isTraceEnabled())
1543 log.trace("_get(" + ", \"" + fqn + "\", " + key + ", \"" +sendNodeEvent +"\")");
1544 Node n=findNode(fqn);
1545 if(n == null) return null;
1546 if(sendNodeEvent)
1547 notifyNodeVisisted(fqn);
1548 return n.get(key);
1549 }
1550
1551
1552 protected Object get(Fqn fqn, Object key, boolean sendNodeEvent) throws CacheException {
1553 MethodCall m=new MethodCall(getKeyValueMethodLocal, new Object []{fqn, key, new Boolean (sendNodeEvent)});
1554 return invokeMethod(m);
1555 }
1556
1557
1564 public Object peek(Fqn fqn, Object key) throws CacheException {
1565 return get(fqn, key, false);
1566 }
1567
1568
1569
1570
1577 public boolean exists(String fqn) {
1578 return exists(Fqn.fromString(fqn));
1579 }
1580
1581
1582
1588 public boolean exists(Fqn fqn) {
1589 Node n=findInternal(fqn);
1590 return n != null;
1591 }
1592
1593
1598 private Node findInternal(Fqn fqn) {
1599 if(fqn == null || fqn.size() == 0) return root;
1600 Node n=root, retval=null;
1601 Object obj;
1602 for(int i=0; i < fqn.size(); i++) {
1603 obj=fqn.get(i);
1604 n=n.getChild(obj);
1605 if(n == null)
1606 return null;
1607 else
1608 retval=n;
1609 }
1610 return retval;
1611 }
1612
1613
1614
1621 public boolean exists(String fqn, Object key) {
1622 return exists(Fqn.fromString(fqn), key);
1623 }
1624
1625
1626
1634 public boolean exists(Fqn fqn, Object key) {
1635 Node n=findInternal(fqn);
1636 if(n == null)
1637 return false;
1638 else
1639 return n.containsKey(key);
1640 }
1641
1642
1643
1653 public void put(String fqn, Map data) throws CacheException {
1654 put(Fqn.fromString(fqn), data);
1655 }
1656
1657
1667 public void put(Fqn fqn, Map data) throws CacheException {
1668 GlobalTransaction tx=getCurrentTransaction();
1669 MethodCall m=new MethodCall(putDataMethodLocal, new Object []{tx, fqn, data, Boolean.TRUE});
1670 invokeMethod(m);
1671 }
1672
1673
1684 public Object put(String fqn, Object key, Object value) throws CacheException {
1685 return put(Fqn.fromString(fqn), key, value);
1686 }
1687
1688
1689
1706 public Object putFailFast(Fqn fqn, Object key, Object value, long timeout) throws CacheException {
1707 GlobalTransaction tx=getCurrentTransaction();
1708 MethodCall m=new MethodCall(putFailFastKeyValueMethodLocal,
1709 new Object []{tx, fqn, key, value, Boolean.TRUE, new Long (timeout)});
1710 return invokeMethod(m);
1711 }
1712
1713
1723 public Object putFailFast(String fqn, Object key, Object value, long timeout) throws CacheException {
1724 GlobalTransaction tx=getCurrentTransaction();
1725 Fqn fqntmp=Fqn.fromString(fqn);
1726 MethodCall m=new MethodCall(putFailFastKeyValueMethodLocal,
1727 new Object []{tx, fqntmp, key, value, Boolean.TRUE, new Long (timeout)});
1728 return invokeMethod(m);
1729 }
1730
1731
1742 public Object put(Fqn fqn, Object key, Object value) throws CacheException {
1743 GlobalTransaction tx=getCurrentTransaction();
1744 MethodCall m=new MethodCall(putKeyValMethodLocal, new Object []{tx, fqn, key, value, Boolean.TRUE});
1745 return invokeMethod(m);
1746 }
1747
1748
1754 public void remove(String fqn) throws CacheException {
1755 remove(Fqn.fromString(fqn));
1756 }
1757
1758
1764 public void remove(Fqn fqn) throws CacheException {
1765 GlobalTransaction tx=getCurrentTransaction();
1766 MethodCall m=new MethodCall(removeNodeMethodLocal, new Object []{tx, fqn, Boolean.TRUE});
1767 invokeMethod(m);
1768 }
1769
1770
1779 public void evict(Fqn fqn) throws CacheException {
1780 MethodCall m=new MethodCall(evictNodeMethodLocal, new Object []{fqn});
1781 invokeMethod(m);
1782 }
1783
1784
1785
1792
1797
1798
1799
1800
1808 public Object remove(String fqn, Object key) throws CacheException {
1809 return remove(Fqn.fromString(fqn), key);
1810 }
1811
1812
1820 public Object remove(Fqn fqn, Object key) throws CacheException {
1821 GlobalTransaction tx=getCurrentTransaction();
1822 MethodCall m=new MethodCall(removeKeyMethodLocal, new Object []{tx, fqn, key, Boolean.TRUE});
1823 return invokeMethod(m);
1824 }
1825
1826
1832 public void removeData(String fqn) throws CacheException {
1833 removeData(Fqn.fromString(fqn));
1834 }
1835
1836
1842 public void removeData(Fqn fqn) throws CacheException {
1843 GlobalTransaction tx=getCurrentTransaction();
1844 MethodCall m=new MethodCall(removeDataMethodLocal, new Object []{tx, fqn, Boolean.TRUE});
1845 invokeMethod(m);
1846 }
1847
1848
1849
1859
1863
1871
1875
1881 public void releaseAllLocks(String fqn) {
1882 releaseAllLocks(Fqn.fromString(fqn));
1883 }
1884
1885
1891 public void releaseAllLocks(Fqn fqn) {
1892 MethodCall m=new MethodCall(releaseAllLocksMethodLocal, new Object []{fqn});
1893 try {
1894 invokeMethod(m);
1895 }
1896 catch(CacheException e) {
1897 log.error("failed releasing all locks for " + fqn, e);
1898 }
1899 }
1900
1901
1907 public String print(String fqn) {
1908 return print(Fqn.fromString(fqn));
1909 }
1910
1911
1917 public String print(Fqn fqn) {
1918 MethodCall m=new MethodCall(printMethodLocal, new Object []{fqn});
1919 Object retval=null;
1920 try {
1921 retval=invokeMethod(m);
1922 }
1923 catch(Throwable e) {
1924 retval=e;
1925 }
1926 if(retval != null)
1927 return retval.toString();
1928 else return "";
1929 }
1930
1931
1932
1939 public Set getChildrenNames(String fqn) throws CacheException {
1940 return getChildrenNames(Fqn.fromString(fqn));
1941 }
1942
1943
1952 public Set getChildrenNames(Fqn fqn) throws CacheException {
1953 MethodCall m=new MethodCall(getChildrenNamesMethodLocal, new Object []{fqn});
1954 return (Set)invokeMethod(m);
1955 }
1956
1957 public Set _getChildrenNames(Fqn fqn) throws CacheException {
1958 Node n=findNode(fqn);
1959 if(n == null) return null;
1960 Map m=n.getChildren();
1961 if(m != null)
1962 return new LinkedHashSet(m.keySet());
1963 else
1964 return null;
1965 }
1966
1967
1968
1969 public boolean hasChild(Fqn fqn) {
1970 if(fqn == null) return false;
1971
1972 Node n=root;
1973 Object obj;
1974 for(int i=0; i < fqn.size(); i++) {
1975 obj=fqn.get(i);
1976 n=n.getChild(obj);
1977 if(n == null)
1978 return false;
1979 }
1980 return n.hasChildren();
1981 }
1982
1983
1987 public String toString() {
1988 return toString(false);
1989 }
1990
1991
1992
1996 public String toString(boolean details) {
1997 StringBuffer sb=new StringBuffer ();
1998 int indent=0;
1999 Map children;
2000
2001 if(!details) {
2002 sb.append(getClass().getName()).append(" [").append(getNumberOfNodes()).append(" nodes, ");
2003 sb.append(getNumberOfLocksHeld()).append(" locks]");
2004 }
2005 else {
2006 children=root.getChildren();
2007 if(children != null && children.size() > 0) {
2008 Collection nodes=children.values();
2009 for(Iterator it=nodes.iterator(); it.hasNext();) {
2010 ((Node)it.next()).print(sb, indent);
2011 sb.append("\n");
2012 }
2013 }
2014 else
2015 sb.append(SEPARATOR);
2016 }
2017 return sb.toString();
2018 }
2019
2020
2021
2022
2023
2027 public String printDetails() {
2028 StringBuffer sb=new StringBuffer ();
2029 int indent=0;
2030 Map children;
2031
2032 children=root.getChildren();
2033 if(children != null && children.size() > 0) {
2034 Collection nodes=children.values();
2035 for(Iterator it=nodes.iterator(); it.hasNext();) {
2036 ((Node)it.next()).printDetails(sb, indent);
2037 sb.append("\n");
2038 }
2039 }
2040 else
2041 sb.append(SEPARATOR);
2042 return sb.toString();
2043 }
2044
2045
2049 public String printLockInfo() {
2050 StringBuffer sb=new StringBuffer ("\n");
2051 int indent=0;
2052 Map children;
2053
2054 children=root.getChildren();
2055 if(children != null && children.size() > 0) {
2056 Collection nodes=children.values();
2057 for(Iterator it=nodes.iterator(); it.hasNext();) {
2058 ((Node)it.next()).printLockInfo(sb, indent);
2059 sb.append("\n");
2060 }
2061 }
2062 else
2063 sb.append(SEPARATOR);
2064 return sb.toString();
2065 }
2066
2067
2072 public int getNumberOfLocksHeld() {
2073 return numLocks(root);
2074 }
2075
2076
2077 int numLocks(Node n) {
2078 int num=0;
2079 Map children;
2080 if(n.isLocked())
2081 num++;
2082 if((children=n.getChildren()) != null) {
2083 for(Iterator it=children.values().iterator(); it.hasNext();) {
2084 num+=numLocks((Node)it.next());
2085 }
2086 }
2087 return num;
2088 }
2089
2090
2096 public int getNumberOfNodes() {
2097 return numNodes(root)-1;
2098 }
2099
2100
2101 int numNodes(Node n) {
2102 if(n == null)
2103 return 0;
2104 int count=1; if(n.hasChildren()) {
2106 Map children=n.getChildren();
2107 if(children != null && children.size() > 0) {
2108 Collection child_nodes=children.values();
2109 Node child;
2110 for(Iterator it=child_nodes.iterator(); it.hasNext();) {
2111 child=(Node)it.next();
2112 count+=numNodes(child);
2113 }
2114 }
2115 }
2116 return count;
2117 }
2118
2119
2125 public int getNumberOfAttributes() {
2126 return numAttributes(root);
2127 }
2128
2129 int numAttributes(Node n) {
2130 if(n == null)
2131 return 0;
2132 int count=n.numAttributes();
2133 if(n.hasChildren()) {
2134 Map children=n.getChildren();
2135 if(children != null && children.size() > 0) {
2136 Collection child_nodes=children.values();
2137 Node child;
2138 for(Iterator it=child_nodes.iterator(); it.hasNext();) {
2139 child=(Node)it.next();
2140 count+=child.numAttributes();
2141 }
2142 }
2143 }
2144 return count;
2145 }
2146
2147
2148
2149
2150 public List callRemoteMethods(Vector mbrs, MethodCall method_call,
2151 boolean synchronous, boolean exclude_self, long timeout)
2152 throws Exception {
2153 RspList rsps;
2154 Rsp rsp;
2155 List retval;
2156 Vector validMembers;
2157 int mode=synchronous ? GroupRequest.GET_ALL : GroupRequest.GET_NONE;
2158
2159 if(disp == null)
2160 return null;
2161
2162 validMembers=mbrs != null ? new Vector(mbrs) : new Vector(this.members);
2163 if(exclude_self && validMembers.size() > 0) {
2164 Object local_addr=getLocalAddress();
2165 if(local_addr != null)
2166 validMembers.remove(local_addr);
2167 }
2168 if(validMembers.size() == 0) {
2169 if(log.isTraceEnabled())
2170 log.trace("destination list is empty, discarding call");
2171 return null;
2172 }
2173
2174 if(log.isTraceEnabled())
2175 log.trace("callRemoteMethods(): valid members are " + validMembers);
2176
2177 rsps=disp.callRemoteMethods(validMembers, method_call, mode, timeout);
2178 if(log.isTraceEnabled())
2179 log.trace("(" + getLocalAddress() + "): responses for method " + method_call.getName() + ":\n" + rsps);
2180 if(rsps == null)
2181 return null;
2182 retval=new ArrayList(rsps.size());
2183 for(int i=0; i < rsps.size(); i++) {
2184 rsp=(Rsp)rsps.elementAt(i);
2185 if(rsp.wasSuspected() || !rsp.wasReceived())
2186 retval.add(new TimeoutException("rsp=" + rsp));
2187 else
2188 retval.add(rsp.getValue());
2189 }
2190 return retval;
2191 }
2192
2193
2194
2205 public List callRemoteMethods(Vector members, Method method, Object [] args,
2206 boolean synchronous, boolean exclude_self, long timeout)
2207 throws Exception {
2208 return callRemoteMethods(members, new MethodCall(method, args), synchronous, exclude_self, timeout);
2209 }
2210
2211
2223 public List callRemoteMethods(Vector members, String method_name,
2224 Class [] types, Object [] args,
2225 boolean synchronous, boolean exclude_self, long timeout)
2226 throws Exception {
2227 Method method=getClass().getDeclaredMethod(method_name, types);
2228 return callRemoteMethods(members, method, args, synchronous, exclude_self, timeout);
2229 }
2230
2231
2232
2233
2234
2235
2250 public void _put(GlobalTransaction tx, String fqn, Map data, boolean create_undo_ops)
2251 throws CacheException {
2252 _put(tx, Fqn.fromString(fqn), data, create_undo_ops);
2253 }
2254
2255
2256
2271 public void _put(GlobalTransaction tx, Fqn fqn, Map data, boolean create_undo_ops)
2272 throws CacheException {
2273 _put(tx, fqn, data, create_undo_ops, false);
2274 }
2275
2276
2277
2293 public void _put(GlobalTransaction tx, Fqn fqn, Map data, boolean create_undo_ops, boolean erase_contents)
2294 throws CacheException {
2295 Node n;
2296 MethodCall undo_op=null;
2297 Map old_data;
2298
2299 if(log.isTraceEnabled())
2300 log.trace(new StringBuffer ().append("_put(").append(tx).append(", \"").append(fqn)
2301 .append("\", ").append(data).append(")").toString());
2302
2303 n=findNode(fqn);
2306 if(n == null) {
2307 String errStr="node " + fqn + " not found (gtx=" + tx + ", caller=" + Thread.currentThread() + ")";
2308 if(log.isTraceEnabled())
2309 log.trace(errStr);
2310 throw new NodeNotExistsException(errStr);
2311 }
2312
2313 if(tx != null && create_undo_ops) {
2317 if((old_data=n.getData()) == null) {
2319 undo_op=new MethodCall(removeDataMethodLocal,
2320 new Object []{tx, fqn, Boolean.FALSE});
2321 }
2322 else {
2323 undo_op=new MethodCall(putDataEraseMethodLocal,
2324 new Object []{tx, fqn,
2325 new HashMap(old_data),
2326 Boolean.FALSE,
2327 Boolean.TRUE}); }
2329 }
2330
2331 n.put(data, erase_contents);
2332
2333 if(tx != null && create_undo_ops) {
2334 tx_table.addUndoOperation(tx, undo_op);
2336 }
2337 notifyNodeModified(fqn);
2338 }
2339
2340
2341
2348 public Object _put(GlobalTransaction tx, String fqn, Object key, Object value, boolean create_undo_ops)
2349 throws CacheException {
2350 return _put(tx, Fqn.fromString(fqn), key, value, create_undo_ops);
2351 }
2352
2353
2354
2355
2356
2363 public Object _put(GlobalTransaction tx, Fqn fqn, Object key, Object value, boolean create_undo_ops)
2364 throws CacheException {
2365 Node n=null;
2366 MethodCall undo_op=null;
2367 Object old_value=null;
2368
2369 if(log.isTraceEnabled()) {
2370 log.trace(new StringBuffer ().append("_put(").append(tx).append(", \"").
2371 append(fqn).append("\", ").append(key).append(", ").append(value).append(")").toString());
2372 }
2373
2374 n=findNode(fqn);
2375 if(n == null) {
2376 String errStr="node " + fqn + " not found (gtx=" + tx + ", caller=" + Thread.currentThread() + ")";
2377 if(log.isTraceEnabled())
2378 log.trace(errStr);
2379 throw new NodeNotExistsException(errStr);
2380 }
2381
2382 old_value=n.get(key);
2383 n.put(key, value);
2384
2385 if(tx != null && create_undo_ops) {
2388 if(old_value == null) {
2389 undo_op=new MethodCall(removeKeyMethodLocal,
2390 new Object []{tx, fqn, key, Boolean.FALSE});
2391 }
2392 else {
2393 undo_op=new MethodCall(putKeyValMethodLocal,
2394 new Object []{tx, fqn, key, old_value,
2395 Boolean.FALSE});
2396 }
2397 tx_table.addUndoOperation(tx, undo_op);
2399 }
2400
2401 notifyNodeModified(fqn);
2402 return old_value;
2403 }
2404
2405
2406 public Object _put(GlobalTransaction tx, Fqn fqn, Object key, Object value, boolean create_undo_ops, long timeout)
2407 throws CacheException {
2408 return _put(tx, fqn, key, value, create_undo_ops);
2409 }
2410
2411
2415 public void _remove(GlobalTransaction tx, String fqn, boolean create_undo_ops) throws CacheException {
2416 _remove(tx, Fqn.fromString(fqn), create_undo_ops);
2417 }
2418
2419
2423 public void _remove(GlobalTransaction tx, Fqn fqn, boolean create_undo_ops) throws CacheException {
2424 _remove(tx, fqn, create_undo_ops, true);
2425 }
2426
2427 public void _remove(GlobalTransaction tx, Fqn fqn, boolean create_undo_ops, boolean sendNodeEvent)
2428 throws CacheException {
2429 _remove(tx, fqn, create_undo_ops, sendNodeEvent, false);
2430 }
2431
2432
2439 public void _remove(GlobalTransaction tx, Fqn fqn, boolean create_undo_ops, boolean sendNodeEvent, boolean eviction)
2440 throws CacheException {
2441 Node n, parent_node;
2442 MethodCall undo_op=null;
2443
2444 if(log.isTraceEnabled())
2445 log.trace("_remove(" + tx + ", \"" + fqn + "\")");
2446
2447 if(fqn.size() == 0) {
2448 Set children=getChildrenNames(fqn);
2449 if(children != null) {
2450 Object [] kids=children.toArray();
2451
2452 for(int i=0; i < kids.length; i++) {
2453 Object s=kids[i];
2454 Fqn tmp=new Fqn(fqn, s);
2455 try {
2456 _remove(tx, tmp, create_undo_ops, true, eviction);
2457 }
2458 catch(Exception e) {
2459 log.error("failure removing node " + tmp);
2460 }
2461 }
2462 }
2463 return;
2464 }
2465
2466 n=findNode(fqn);
2468 if(n == null) {
2469 log.warn("node " + fqn + " not found");
2470 return;
2471 }
2472 parent_node=n.getParent();
2473
2474 parent_node.removeChild(n.getName());
2476
2477 n.releaseAll(tx != null? tx : (Object )Thread.currentThread());
2479
2480 if(tx != null && create_undo_ops && n != null && eviction == false) {
2483 undo_op=new MethodCall(addChildMethodLocal, new Object []{tx, parent_node.getFqn(), n.getName(), n});
2484
2485 tx_table.addUndoOperation(tx, undo_op);
2487 }
2488
2489 if(sendNodeEvent)
2490 notifyNodeRemoved(fqn);
2491 else
2492 notifyNodeEvicted(fqn);
2493 }
2494
2495
2501 public Object _remove(GlobalTransaction tx, String fqn, Object key, boolean create_undo_ops)
2502 throws CacheException {
2503 return _remove(tx, Fqn.fromString(fqn), key, create_undo_ops);
2504 }
2505
2506
2512 public Object _remove(GlobalTransaction tx, Fqn fqn, Object key, boolean create_undo_ops)
2513 throws CacheException {
2514 return _remove(tx, fqn, key, create_undo_ops, false); }
2516
2517 public Object _remove(GlobalTransaction tx, Fqn fqn, Object key, boolean create_undo_ops, boolean sendNodeEvent)
2518 throws CacheException {
2519 Node n=null;
2520 MethodCall undo_op=null;
2521 Object old_value=null;
2522
2523 if(log.isTraceEnabled())
2524 log.trace(new StringBuffer ().append("_remove(").append(tx).append(", \"")
2525 .append(fqn).append("\", ").append(key).append(")").toString());
2526
2527 n=findNode(fqn);
2530 if(n == null) {
2531 log.warn("node " + fqn + " not found");
2532 return null;
2533 }
2534 old_value=n.remove(key);
2535
2536 if(tx != null && create_undo_ops && old_value != null) {
2539 undo_op=new MethodCall(putKeyValMethodLocal,
2540 new Object []{tx, fqn, key, old_value,
2541 Boolean.FALSE});
2542 tx_table.addUndoOperation(tx, undo_op);
2544 }
2545
2546 if(sendNodeEvent)
2547 notifyNodeModified(fqn); return old_value;
2549 }
2550
2551
2554 public void _removeData(GlobalTransaction tx, String fqn, boolean create_undo_ops)
2555 throws CacheException {
2556 _removeData(tx, Fqn.fromString(fqn), create_undo_ops);
2557 }
2558
2559
2562 public void _removeData(GlobalTransaction tx, Fqn fqn, boolean create_undo_ops)
2563 throws CacheException {
2564 _removeData(tx, fqn, create_undo_ops, true);
2565 }
2566
2567
2568 public void _removeData(GlobalTransaction tx, Fqn fqn, boolean create_undo_ops, boolean sendNodeEvent)
2569 throws CacheException {
2570 _removeData(tx, fqn, create_undo_ops, sendNodeEvent, false);
2571 }
2572
2573 public void _removeData(GlobalTransaction tx, Fqn fqn, boolean create_undo_ops, boolean sendNodeEvent, boolean eviction)
2574 throws CacheException {
2575 Node n=null;
2576 MethodCall undo_op=null;
2577 Map old_data=null;
2578
2579 if(log.isTraceEnabled())
2580 log.trace("_removeData(" + tx + ", \"" + fqn + "\")");
2581
2582 n=findNode(fqn);
2585 if(n == null) {
2586 log.warn("node " + fqn + " not found");
2587 return;
2588 }
2589
2590 if(tx != null && create_undo_ops && (old_data=n.getData()) != null && !eviction) {
2593 undo_op=new MethodCall(putDataMethodLocal, new Object []{tx, fqn, new HashMap(old_data), Boolean.FALSE});
2594 }
2595
2596 n.clear();
2597 if(eviction)
2598 n.put(UNINITIALIZED, null);
2600 if(sendNodeEvent) {
2601 notifyNodeVisisted(fqn);
2602 }
2603 else { if(eviction)
2605 notifyNodeEvicted(fqn);
2606 else
2607 notifyNodeModified(fqn); }
2609
2610 if(tx != null && create_undo_ops) {
2612 tx_table.addUndoOperation(tx, undo_op);
2613 }
2614 }
2615
2616
2617
2625 public void _evict(Fqn fqn) throws CacheException {
2626 if(!exists(fqn)) return; boolean create_undo_ops = false;
2629 boolean sendNodeEvent = false;
2630 boolean eviction=true;
2631 if(log.isTraceEnabled())
2632 log.trace("_evict(" + fqn + ")");
2633 if(hasChild(fqn)) {
2634 _removeData(null, fqn, create_undo_ops, sendNodeEvent, eviction);
2635 }
2636 else {
2637 _remove(null, fqn, create_undo_ops, sendNodeEvent, eviction);
2638 }
2639 }
2640
2641
2642
2648
2656
2657
2658
2659
2668 public void _addChild(GlobalTransaction tx, Fqn parent_fqn, Object child_name, Node old_node)
2669 throws CacheException {
2670 if(log.isTraceEnabled())
2671 log.trace("_addChild(" + tx + ", \"" + parent_fqn + "\", \"" + child_name + "\")");
2672
2673 if(parent_fqn == null || child_name == null || old_node == null) {
2674 log.error("parent_fqn or child_name or node was null");
2675 return;
2676 }
2677 Node tmp=findNode(parent_fqn);
2678 if(tmp == null) {
2679 log.warn("node " + parent_fqn + " not found");
2680 return;
2681 }
2682
2683 tmp.addChild(child_name, old_node);
2684 }
2685
2686
2687
2688
2689
2697 public Object _replicate(MethodCall method_call) throws Throwable {
2698 if(replication_handler != null) {
2699 return replication_handler.replicate(method_call);
2700 }
2701 else {
2702 throw new UnsupportedOperationException ("no replication handler is installed");
2703 }
2704 }
2705
2706 public void _replicate(List method_calls) throws Throwable {
2707 if(replication_handler != null) {
2708 replication_handler.replicate(method_calls);
2709 }
2710 else {
2711 throw new UnsupportedOperationException ("no replication handler is installed");
2712 }
2713 }
2714
2715 public void _releaseAllLocks(Fqn fqn) {
2716 Node n;
2717
2718 try {
2719 n=findNode(fqn);
2720 if(n == null) {
2721 log.error("releaseAllLocks(): node " + fqn + " not found");
2722 return;
2723 }
2724 n.releaseAllForce();
2725 }
2726 catch(Throwable t) {
2727 log.error("releaseAllLocks(): failed", t);
2728 }
2729 }
2730
2731
2732 public String _print(Fqn fqn) {
2733 try {
2734 Node n=findNode(fqn);
2735 if(n == null) return null;
2736 return n.toString();
2737 }
2738 catch(Throwable t) {
2739 return null;
2740 }
2741 }
2742
2743
2744 public void _lock(Fqn fqn, int lock_type, boolean recursive)
2745 throws TimeoutException, LockingException {
2746 log.warn("method _lock() should not be invoked on TreeCache");
2747 }
2748
2749 public void prepare(GlobalTransaction global_tx, List modifications, Address coord, boolean commit) {
2751 throw new UnsupportedOperationException ("prepare() should not be called on TreeCache directly");
2752 }
2753
2754 public void commit(GlobalTransaction tx) {
2755 throw new UnsupportedOperationException ("commit() should not be called on TreeCache directly");
2756 }
2757
2758 public void rollback(GlobalTransaction tx) {
2759 throw new UnsupportedOperationException ("rollback() should not be called on TreeCache directly");
2760 }
2761
2762
2763
2764
2765
2766 public void addNode(GlobalTransaction gtx, Fqn node) {
2767 tx_table.addNode(gtx, node);
2768 }
2769
2770
2771
2772
2773
2774 class MessageListenerAdaptor implements MessageListener {
2775 final TreeCache cache;
2776 final Logger log;
2778 MessageListenerAdaptor(TreeCache cache, Logger log) {
2779 this.cache = cache;
2780 this.log = log;
2781 }
2782
2783
2788 public void receive(Message msg) {
2789 }
2790
2791
2795 public byte[] getState() {
2796 Object owner=null;
2797 boolean fetch_persistent_state=cache_loader != null &&
2798 cache_loader_shared == false &&
2799 cache_loader_fetch_persistent_state;
2800 byte[] transient_state=null;
2801 byte[] persistent_state=null;
2802 byte[][] states=new byte[2][];
2803 byte[] retval=null;
2804 boolean locked=false;
2805
2806 owner=getCurrentTransaction();
2807 if(owner == null) owner=Thread.currentThread();
2808
2809 states[0]=states[1]=null;
2810 try {
2811 if(cache_loader_fetch_transient_state) {
2812 log.info("locking the tree to obtain transient state");
2813 root.acquireAll(owner, state_fetch_timeout, Node.LOCK_TYPE_READ);
2814 locked=true;
2815 transient_state=Util.objectToByteBuffer(root);
2816 states[0]=transient_state;
2817 log.info("returning the transient state (" + transient_state.length + " bytes)");
2818 }
2819 }
2820 catch(Throwable t) {
2821 log.error("failed getting the transient state", t);
2822 }
2823 try {
2824 if(fetch_persistent_state) {
2825 if(!locked) {
2826 log.info("locking the tree to obtain persistent state");
2827 root.acquireAll(owner, state_fetch_timeout, Node.LOCK_TYPE_READ);
2828 locked=true;
2829 }
2830 log.info("getting the persistent state");
2831 persistent_state=cache_loader.loadEntireState();
2832 states[1]=persistent_state;
2833 log.info("returning the persistent state (" + persistent_state.length + " bytes)");
2834 }
2835 }
2836 catch(Throwable t) {
2837 log.error("failed getting the persistent state", t);
2838 }
2839
2840 try {
2841 retval=Util.objectToByteBuffer(states);
2842 return retval;
2843 }
2844 catch(Throwable t) {
2845 log.error("failed serializing transient and persistent state", t);
2846 return retval;
2847 }
2848
2849 finally {
2850 root.releaseAll(owner);
2851 }
2852 }
2853
2854
2855 public void setState(byte[] new_state) {
2856 try {
2857 _setState(new_state);
2858 }
2859 finally {
2860 synchronized(stateLock) {
2861 stateLock.notifyAll();
2863 if(root != null)
2864 root.releaseAllForce();
2865 }
2866 }
2867 }
2868
2869
2873 void _setState(byte[] new_state) {
2874 Node new_root=null, old_root=null;
2875 Object obj;
2876 Object owner=null;
2877 byte[][] states=null;
2878 byte[] transient_state=null;
2879 byte[] persistent_state=null;
2880 boolean locked=false;
2881
2882 if(new_state == null) {
2883 log.info("new cache is null (maybe first member in cluster)");
2884 return;
2885 }
2886
2887 try {
2889 log.info("received the state (size=" + new_state.length + " bytes)");
2890 states=(byte[][])Util.objectFromByteBuffer(new_state);
2891 transient_state=states[0];
2892 persistent_state=states[1];
2893 if(transient_state != null)
2894 log.info("transient state: " + transient_state.length + " bytes");
2895 if(persistent_state != null)
2896 log.info("persistent state: " + persistent_state.length + " bytes");
2897 owner=getCurrentTransaction();
2898 if(owner == null) owner=Thread.currentThread();
2899 }
2900 catch(Throwable t) {
2901 log.error("failed unserializing state", t);
2902 }
2903
2904 if(transient_state != null) {
2906 try {
2907 log.info("setting transient state");
2908 obj=Util.objectFromByteBuffer(transient_state);
2909 new_root=(Node)obj;
2910 new_root.setRecursiveTreeCacheInstance(cache); log.info("locking the old tree");
2912 root.acquireAll(owner, state_fetch_timeout, Node.LOCK_TYPE_WRITE);
2913 locked=true;
2914 log.info("locking the old tree was successful");
2915 old_root=root;
2916 root=new_root;
2917 log.info("setting the transient state was successful");
2918 notifyAllNodesCreated(root);
2919 }
2920 catch(Throwable t) {
2921 log.error("failed setting transient state", t);
2922 }
2923 }
2924
2925 if(persistent_state != null) {
2927 if(cache_loader == null) {
2928 log.error("cache loader is null, cannot set persistent state");
2929 }
2930 else {
2931 try {
2932 if(!locked) {
2933 log.info("locking the old tree");
2934 root.acquireAll(owner, state_fetch_timeout, Node.LOCK_TYPE_WRITE);
2935 old_root=root; locked=true;
2937 root=new Node(SEPARATOR, Fqn.fromString(SEPARATOR), null, null, cache);
2938 log.info("locking the old tree was successful");
2939 }
2940 log.info("setting the persistent state");
2941 cache_loader.storeEntireState(persistent_state);
2943 log.info("setting the persistent state was successful");
2944 }
2945 catch(Throwable t) {
2946 log.error("failed setting persistent state", t);
2947 }
2948 }
2949 }
2950
2951 if(old_root != null) {
2952 log.info("forcing release of all locks in old tree");
2953 try {old_root.releaseAllForce();} catch(Throwable t) {log.error("failed releasing locks", t);}
2954 }
2955
2956 isStateSet=true;
2957 }
2958
2959
2960
2961 }
2962
2963
2964
2965
2966
2967
2968
2969
2970 public void viewAccepted(View new_view) {
2971 Vector new_mbrs=new_view.getMembers();
2972
2973
2977 log.info("viewAccepted(): new members: " + new_mbrs);
2978 if(new_mbrs != null) {
2979 members.removeAllElements();
2980 members.addAll(new_view.getMembers());
2981 notifyViewChange(new_view);
2982 }
2983
2984 if(cache_loader_shared) {
2985 coordinator=determineCoordinator();
2987 }
2988 }
2989
2990
2991
2994 public void suspect(Address suspected_mbr) {
2995 }
2996
2997
2998
3001 public void block() {
3002 }
3003
3004
3005
3006
3007
3008
3009
3016 protected Transaction getLocalTransaction() {
3017 if(tm == null) {
3018 return null;
3019 }
3020 try {
3021 return tm.getTransaction();
3022 }
3023 catch(Throwable t) {
3024 return null;
3025 }
3026 }
3027
3028
3029
3030
3031
3032 boolean isValid(Transaction tx) {
3033 if(tx == null) return false;
3034 int status=-1;
3035 try {
3036 status=tx.getStatus();
3037 return status == Status.STATUS_ACTIVE || status == Status.STATUS_PREPARING;
3038 }
3039 catch(SystemException e) {
3040 log.error("failed getting transaction status", e);
3041 return false;
3042 }
3043 }
3044
3045
3046
3054 public GlobalTransaction getCurrentTransaction() {
3055 Transaction tx;
3056
3057 if((tx=getLocalTransaction()) == null) { return null;
3059 }
3060
3061 if(!isValid(tx)) { int status=-1;
3063 try {status=tx.getStatus();} catch(SystemException e) {}
3064 log.warn("status is " + status + " (not ACTIVE or PREPARING); returning null)");
3065 return null;
3066 }
3067
3068 return getCurrentTransaction(tx);
3069 }
3070
3071 public GlobalTransaction getCurrentTransaction(Transaction tx) {
3072 synchronized(tx_table) {
3073 GlobalTransaction gtx=tx_table.get(tx);
3074 if(gtx == null) {
3075 Address addr=(Address)getLocalAddress();
3076 gtx=GlobalTransaction.create(addr);
3077 tx_table.put(tx, gtx);
3078 TransactionEntry ent=new TransactionEntry();
3079 ent.setTransaction(tx);
3080 tx_table.put(gtx, ent);
3081 if(log.isTraceEnabled())
3082 log.trace("created new GTX: " + gtx + ", local TX=" + tx);
3083 }
3084 return gtx;
3085 }
3086 }
3087
3088
3089
3090
3099 protected Object invokeMethod(MethodCall m) throws CacheException {
3100 try {
3102 return interceptor_chain.invoke(m);
3103 }
3104 catch(Throwable t) {
3105 if(t instanceof CacheException)
3106 throw (CacheException)t;
3107 throw new NestedRuntimeException(t);
3108 }
3109 }
3110
3111
3112
3127 private Node findNode(Fqn fqn) {
3128 Node n, child_node=null;
3129 Object child_name;
3130 int treeNodeSize;
3131 Fqn tmp_fqn=new Fqn();
3132
3133 if(fqn == null) return null;
3134 if((treeNodeSize=fqn.size()) == 0)
3135 return root;
3136
3137 n=root;
3138 for(int i=0; i < treeNodeSize; i++) {
3139 child_name=fqn.get(i);
3140 tmp_fqn=new Fqn(tmp_fqn, child_name);
3141 child_node=n.getChild(child_name);
3142 if(child_node == null)
3143 return null;
3144 n=child_node;
3145 }
3146 return child_node;
3147 }
3148
3149
3150
3151 public void notifyNodeCreated(Fqn fqn) {
3152 for(int i=0; i < listeners.size(); i++)
3153 ((TreeCacheListener)listeners.elementAt(i)).nodeCreated(fqn);
3154 }
3155
3156 public void notifyNodeLoaded(Fqn fqn) {
3157 for(int i=0; i < listeners.size(); i++)
3158 ((TreeCacheListener)listeners.elementAt(i)).nodeLoaded(fqn);
3159 }
3160
3161 protected void notifyNodeRemoved(Fqn fqn) {
3162 for(int i=0; i < listeners.size(); i++)
3163 ((TreeCacheListener)listeners.elementAt(i)).nodeRemoved(fqn);
3164 }
3165
3166 protected void notifyNodeEvicted(Fqn fqn) {
3167 for(int i=0; i < listeners.size(); i++)
3168 ((TreeCacheListener)listeners.elementAt(i)).nodeEvicted(fqn);
3169 }
3170
3171 protected void notifyNodeModified(Fqn fqn) {
3172 for(int i=0; i < listeners.size(); i++)
3173 ((TreeCacheListener)listeners.elementAt(i)).nodeModified(fqn);
3174 }
3175
3176 protected void notifyNodeVisisted(Fqn fqn) {
3177 for(int i=0; i < listeners.size(); i++)
3178 ((TreeCacheListener)listeners.elementAt(i)).nodeVisited(fqn);
3179 }
3180
3181 protected void notifyCacheStarted() {
3182 for(int i=0; i < listeners.size(); i++)
3183 ((TreeCacheListener)listeners.elementAt(i)).cacheStarted(this);
3184 }
3185
3186 protected void notifyCacheStopped() {
3187 for(int i=0; i < listeners.size(); i++)
3188 ((TreeCacheListener)listeners.elementAt(i)).cacheStopped(this);
3189 }
3190
3191 protected void notifyViewChange(View v) {
3192 for(int i=0; i < listeners.size(); i++)
3193 ((TreeCacheListener)listeners.elementAt(i)).viewChange(v);
3194 }
3195
3196
3200 protected void notifyAllNodesCreated(Node curr) {
3201 Node n;
3202 Map children;
3203
3204 if(curr == null) return;
3205 notifyNodeCreated(curr.fqn);
3206 if((children=curr.getChildren()) != null) {
3207 for(Iterator it=children.values().iterator(); it.hasNext();) {
3208 n=(Node)it.next();
3209 notifyAllNodesCreated(n);
3210 }
3211 }
3212 }
3213
3214 protected String getDefaultProperties() {
3215 return "UDP(mcast_addr=224.0.0.36;mcast_port=55566;ip_ttl=32;" +
3216 "mcast_send_buf_size=150000;mcast_recv_buf_size=80000):" +
3217 "PING(timeout=1000;num_initial_members=2):" +
3218 "MERGE2(min_interval=5000;max_interval=10000):" +
3219 "FD_SOCK:" +
3220 "VERIFY_SUSPECT(timeout=1500):" +
3221 "pbcast.NAKACK(gc_lag=50;max_xmit_size=8192;retransmit_timeout=600,1200,2400,4800):" +
3222 "UNICAST(timeout=600,1200,2400,4800):" +
3223 "pbcast.STABLE(desired_avg_gossip=20000):" +
3224 "FRAG(frag_size=8192;down_thread=false;up_thread=false):" +
3225 "pbcast.GMS(join_timeout=5000;join_retry_timeout=2000;" +
3226 "shun=false;print_local_addr=true):" +
3227 "pbcast.STATE_TRANSFER";
3228 }
3229
3230
3231 protected int string2Mode(String mode) {
3232 if(mode == null) return -1;
3233 String m=mode.toLowerCase().trim();
3234 if(m.equals("local"))
3235 return LOCAL;
3236 else
3237 if(m.equals("repl_async") || m.equals("repl-async"))
3238 return REPL_ASYNC;
3239 else
3240 if(m.equals("repl_sync") || m.equals("repl-sync"))
3241 return REPL_SYNC;
3242 else
3243 return -1;
3244 }
3245
3246
3247
3248}
3249
3250
3251 |
Popular Tags |
Java API By Example, From Geeks To Geeks.