-
Notifications
You must be signed in to change notification settings - Fork 476
/
GMS.java
1445 lines (1196 loc) · 52.6 KB
/
GMS.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
package org.jgroups.protocols.pbcast;
import org.jgroups.*;
import org.jgroups.annotations.*;
import org.jgroups.conf.PropertyConverters;
import org.jgroups.logging.Log;
import org.jgroups.protocols.TP;
import org.jgroups.protocols.pbcast.GmsImpl.Request;
import org.jgroups.stack.Protocol;
import org.jgroups.util.*;
import org.jgroups.util.Queue;
import org.jgroups.util.UUID;
import java.io.*;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
/**
* Group membership protocol. Handles joins/leaves/crashes (suspicions) and
* emits new views accordingly. Use VIEW_ENFORCER on top of this layer to make
* sure new members don't receive any messages until they are members
*
* @author Bela Ban
*/
@MBean(description="Group membership protocol")
public class GMS extends Protocol implements TP.ProbeHandler {
private static final String CLIENT="Client";
private static final String COORD="Coordinator";
private static final String PART="Participant";
/* ------------------------------------------ Properties ------------------------------------------ */
@Property(description="Join timeout")
long join_timeout=5000;
@Property(description="Leave timeout")
long leave_timeout=5000;
@Property(description="Timeout to complete merge")
long merge_timeout=5000; // time to wait for all MERGE_RSPS
@Property(description="Print local address of this member after connect. Default is true")
private boolean print_local_addr=true;
@Property(description="Print physical address(es) on startup")
private boolean print_physical_addrs=true;
@Property(description="If true this member can never become coordinator. Default is false",
deprecatedMessage="This method will be deprecated in 3.0")
boolean disable_initial_coord=false; // can the member become a coord on startup or not ?
/**
* Setting this to false disables concurrent startups. This is only used by
* unit testing code for testing merging. To everybody else: don't change it
* to false !
*/
@Property(description="Temporary switch. Default is true and should not be changed")
boolean handle_concurrent_startup=true;
/**
* Whether view bundling (http://jira.jboss.com/jira/browse/JGRP-144) should
* be enabled or not. Setting this to false forces each JOIN/LEAVE/SUPSECT
* request to be handled separately. By default these requests are processed
* together if they are queued at approximately the same time
*/
@Property(description="View bundling toggle")
private boolean view_bundling=true;
@Property(description="Max view bundling timeout if view bundling is turned on. Default is 50 msec")
private long max_bundling_time=50; // 50ms max to wait for other JOIN, LEAVE or SUSPECT requests
@Property(description="Max number of old members to keep in history. Default is 50")
protected int num_prev_mbrs=50;
@Property(description="Time in ms to wait for all VIEW acks (0 == wait forever. Default is 2000 msec" )
long view_ack_collection_timeout=2000;
@Property(description="Timeout to resume ViewHandler. Default is 10000 msec")
long resume_task_timeout=10000;
@Property(description="Use flush for view changes. Default is true")
boolean use_flush_if_present=true;
@Property(description="Logs failures for collecting all view acks if true")
boolean log_collect_msgs=true;
/* --------------------------------------------- JMX ---------------------------------------------- */
private int num_views=0;
/** Stores the last 20 views */
private final BoundedList<View> prev_views=new BoundedList<View>(20);
/* --------------------------------------------- Fields ------------------------------------------------ */
@Property(converter=PropertyConverters.FlushInvoker.class,name="flush_invoker_class")
protected Class<Callable<Boolean>> flushInvokerClass;
private GmsImpl impl=null;
private final Object impl_mutex=new Object(); // synchronizes event entry into impl
private final Hashtable<String,GmsImpl> impls=new Hashtable<String,GmsImpl>(3);
// Handles merge related tasks
final Merger merger=new Merger(this, log);
protected Address local_addr=null;
protected final Membership members=new Membership(); // real membership
private final Membership tmp_members=new Membership(); // base for computing next view
/** Members joined but for which no view has been received yet */
private final List<Address> joining=new ArrayList<Address>(7);
/** Members excluded from group, but for which no view has been received yet */
private final List<Address> leaving=new ArrayList<Address>(7);
/** Keeps track of old members (up to num_prev_mbrs) */
private BoundedList<Address> prev_members=null;
protected View view=null;
protected ViewId view_id=null;
protected long ltime=0;
protected TimeScheduler timer=null;
/** Class to process JOIN, LEAVE and MERGE requests */
private final ViewHandler view_handler=new ViewHandler();
/** To collect VIEW_ACKs from all members */
protected final AckCollector ack_collector=new AckCollector();
//[JGRP-700] - FLUSH: flushing should span merge
protected final AckCollector merge_ack_collector=new AckCollector();
boolean flushProtocolInStack=false;
public GMS() {
initState();
}
@ManagedAttribute
public String getView() {return view_id != null? view_id.toString() : "null";}
@ManagedAttribute
public int getNumberOfViews() {return num_views;}
@ManagedAttribute
public String getLocalAddress() {return local_addr != null? local_addr.toString() : "null";}
@ManagedAttribute
public String getMembers() {return members.toString();}
@ManagedAttribute
public int getNumMembers() {return members.size();}
public long getJoinTimeout() {return join_timeout;}
public void setJoinTimeout(long t) {join_timeout=t;}
public long getMergeTimeout() {
return merge_timeout;
}
public void setMergeTimeout(long timeout) {merge_timeout=timeout;}
@ManagedOperation
public String printPreviousMembers() {
StringBuilder sb=new StringBuilder();
if(prev_members != null) {
for(Address addr: prev_members) {
sb.append(addr).append("\n");
}
}
return sb.toString();
}
public void setPrintLocalAddress(boolean flag) {print_local_addr=flag;}
public void setPrintLocalAddr(boolean flag) {setPrintLocalAddress(flag);}
public long getViewAckCollectionTimeout() {
return view_ack_collection_timeout;
}
public void setViewAckCollectionTimeout(long view_ack_collection_timeout) {
if(view_ack_collection_timeout <= 0)
throw new IllegalArgumentException("view_ack_collection_timeout has to be greater than 0");
this.view_ack_collection_timeout=view_ack_collection_timeout;
}
public boolean isViewBundling() {
return view_bundling;
}
public void setViewBundling(boolean view_bundling) {
this.view_bundling=view_bundling;
}
public long getMaxBundlingTime() {
return max_bundling_time;
}
public void setMaxBundlingTime(long max_bundling_time) {
this.max_bundling_time=max_bundling_time;
}
@ManagedAttribute
public int getViewHandlerSize() {return view_handler.size();}
@ManagedAttribute
public boolean isViewHandlerSuspended() {return view_handler.suspended();}
@ManagedOperation
public String dumpViewHandlerQueue() {
return view_handler.dumpQueue();
}
@ManagedOperation
public String dumpViewHandlerHistory() {
return view_handler.dumpHistory();
}
@ManagedOperation
public void suspendViewHandler() {
view_handler.suspend(null);
}
@ManagedOperation
public void resumeViewHandler() {
view_handler.resumeForce();
}
Log getLog() {return log;}
ViewHandler getViewHandler() {return view_handler;}
@ManagedOperation
public String printPreviousViews() {
StringBuilder sb=new StringBuilder();
for(View view: prev_views) {
sb.append(view).append("\n");
}
return sb.toString();
}
@ManagedOperation
public void suspect(String suspected_member) {
if(suspected_member == null)
return;
Map<Address,String> contents=UUID.getContents();
for(Map.Entry<Address,String> entry: contents.entrySet()) {
String logical_name=entry.getValue();
if(logical_name != null && logical_name.equals(suspected_member)) {
Address suspect=entry.getKey();
if(suspect != null)
up(new Event(Event.SUSPECT, suspect));
}
}
}
public boolean isCoordinator() {
Address coord=determineCoordinator();
return coord != null && local_addr != null && local_addr.equals(coord);
}
public MergeId getMergeId() {
return impl instanceof CoordGmsImpl? ((CoordGmsImpl)impl).getMergeId() : null;
}
public void setLogCollectMessages(boolean flag) {
log_collect_msgs=flag;
}
public boolean getLogCollectMessages() {
return log_collect_msgs;
}
public void resetStats() {
super.resetStats();
num_views=0;
prev_views.clear();
}
public Vector<Integer> requiredDownServices() {
Vector<Integer> retval=new Vector<Integer>(3);
retval.addElement(new Integer(Event.GET_DIGEST));
retval.addElement(new Integer(Event.SET_DIGEST));
retval.addElement(new Integer(Event.FIND_INITIAL_MBRS));
return retval;
}
public void setImpl(GmsImpl new_impl) {
synchronized(impl_mutex) {
if(impl == new_impl) // unnecessary ?
return;
impl=new_impl;
if(log.isDebugEnabled())
log.debug(local_addr != null? local_addr + ": " : "" + "changed role to " + new_impl.getClass().getName());
}
}
public GmsImpl getImpl() {
return impl;
}
public void init() throws Exception {
if(view_ack_collection_timeout <= 0)
throw new IllegalArgumentException("view_ack_collection_timeout has to be greater than 0");
prev_members=new BoundedList<Address>(num_prev_mbrs);
TP transport=getTransport();
timer=transport.getTimer();
if(timer == null)
throw new Exception("timer is null");
if(impl != null)
impl.init();
transport.registerProbeHandler(this);
}
public void start() throws Exception {
if(impl != null) impl.start();
}
public void stop() {
view_handler.stop(true);
if(impl != null) impl.stop();
if(prev_members != null)
prev_members.clear();
}
public void becomeCoordinator() {
CoordGmsImpl tmp=(CoordGmsImpl)impls.get(COORD);
if(tmp == null) {
tmp=new CoordGmsImpl(this);
impls.put(COORD, tmp);
}
try {
tmp.init();
}
catch(Exception e) {
log.error("exception switching to coordinator role", e);
}
setImpl(tmp);
}
public void becomeParticipant() {
ParticipantGmsImpl tmp=(ParticipantGmsImpl)impls.get(PART);
if(tmp == null) {
tmp=new ParticipantGmsImpl(this);
impls.put(PART, tmp);
}
try {
tmp.init();
}
catch(Exception e) {
log.error("exception switching to participant", e);
}
setImpl(tmp);
}
public void becomeClient() {
ClientGmsImpl tmp=(ClientGmsImpl)impls.get(CLIENT);
if(tmp == null) {
tmp=new ClientGmsImpl(this);
impls.put(CLIENT, tmp);
}
try {
tmp.init();
}
catch(Exception e) {
log.error("exception switching to client role", e);
}
setImpl(tmp);
}
boolean haveCoordinatorRole() {
return impl != null && impl instanceof CoordGmsImpl;
}
@ManagedOperation(description="Fetches digests from all members and installs them, unblocking blocked members")
public void fixDigests() {
if(impl instanceof CoordGmsImpl)
((CoordGmsImpl)impl).fixDigests();
}
/**
* Computes the next view. Returns a copy that has <code>old_mbrs</code> and
* <code>suspected_mbrs</code> removed and <code>new_mbrs</code> added.
*/
public View getNextView(Collection<Address> new_mbrs, Collection<Address> old_mbrs, Collection<Address> suspected_mbrs) {
Vector<Address> mbrs;
long vid;
View v;
Membership tmp_mbrs;
synchronized(members) {
if(view_id == null) {
log.error("view_id is null");
return null; // this should *never* happen !
}
vid=Math.max(view_id.getId(), ltime) + 1;
ltime=vid;
tmp_mbrs=tmp_members.copy(); // always operate on the temporary membership
tmp_mbrs.remove(suspected_mbrs);
tmp_mbrs.remove(old_mbrs);
tmp_mbrs.add(new_mbrs);
mbrs=tmp_mbrs.getMembers();
Address new_coord=local_addr;
if(!mbrs.isEmpty())
new_coord=mbrs.firstElement();
v=new View(new_coord, vid, mbrs);
// Update membership (see DESIGN for explanation):
tmp_members.set(mbrs);
// Update joining list (see DESIGN for explanation)
if(new_mbrs != null) {
for(Address tmp_mbr: new_mbrs) {
if(!joining.contains(tmp_mbr))
joining.add(tmp_mbr);
}
}
// Update leaving list (see DESIGN for explanations)
if(old_mbrs != null) {
for(Address addr: old_mbrs) {
if(!leaving.contains(addr))
leaving.add(addr);
}
}
if(suspected_mbrs != null) {
for(Address addr:suspected_mbrs) {
if(!leaving.contains(addr))
leaving.add(addr);
}
}
return v;
}
}
/**
* Broadcasts the new view and digest, and waits for acks from all members in the list given as argument.
* If the list is null, we take the members who are part of new_view
* @param new_view
* @param digest
* @param newMembers
*/
public void castViewChangeWithDest(View new_view, Digest digest, JoinRsp jr, Collection <Address> newMembers) {
if(log.isTraceEnabled())
log.trace(local_addr + ": mcasting view {" + new_view + "} (" + new_view.size() + " mbrs)\n");
Message view_change_msg=new Message(); // bcast to all members
GmsHeader hdr=new GmsHeader(GmsHeader.VIEW, new_view);
hdr.my_digest=digest;
view_change_msg.putHeader(this.id, hdr);
List<Address> ackMembers = new ArrayList<Address>(new_view.getMembers());
if(newMembers != null && !newMembers.isEmpty()) {
ackMembers.removeAll(newMembers);
}
if(!ackMembers.isEmpty())
ack_collector.reset(ackMembers);
else
ack_collector.reset(null);
// Send down a local TMP_VIEW event. This is needed by certain layers (e.g. NAKACK) to compute correct digest
// in case client's next request (e.g. getState()) reaches us *before* our own view change multicast.
// Check NAKACK's TMP_VIEW handling for details
down_prot.up(new Event(Event.TMP_VIEW, new_view));
down_prot.down(new Event(Event.TMP_VIEW, new_view));
down_prot.down(new Event(Event.MSG, view_change_msg));
try {
if(!ackMembers.isEmpty()) {
ack_collector.waitForAllAcks(view_ack_collection_timeout);
if(log.isTraceEnabled())
log.trace(local_addr + ": received all ACKs (" + ack_collector.expectedAcks() +
") from existing members for view " + new_view.getVid());
}
}
catch(TimeoutException e) {
if(log_collect_msgs && log.isWarnEnabled()) {
log.warn(local_addr + ": failed to collect all ACKs (expected=" + ack_collector.expectedAcks()
+ ") for view " + new_view + " after " + view_ack_collection_timeout + "ms, missing ACKs from "
+ ack_collector.printMissing());
}
}
if(jr != null && (newMembers != null && !newMembers.isEmpty())) {
ack_collector.reset(new ArrayList<Address>(newMembers));
for(Address joiner: newMembers) {
sendJoinResponse(jr, joiner);
}
try {
ack_collector.waitForAllAcks(view_ack_collection_timeout);
if(log.isTraceEnabled())
log.trace(local_addr + ": received all ACKs (" + ack_collector.expectedAcks() +
") from joiners for view " + new_view.getVid());
}
catch(TimeoutException e) {
if(log_collect_msgs && log.isWarnEnabled()) {
log.warn(local_addr + ": failed to collect all ACKs (expected=" + ack_collector.expectedAcks()
+ ") for unicast view " + new_view + " after " + view_ack_collection_timeout + "ms, missing ACKs from "
+ ack_collector.printMissing());
}
}
}
}
public void sendJoinResponse(JoinRsp rsp, Address dest) {
Message m=new Message(dest, null, null);
GMS.GmsHeader hdr=new GMS.GmsHeader(GMS.GmsHeader.JOIN_RSP, rsp);
m.putHeader(this.id, hdr);
getDownProtocol().down(new Event(Event.MSG, m));
}
public void installView(View new_view) {
installView(new_view, null);
}
/**
* Sets the new view and sends a VIEW_CHANGE event up and down the stack. If the view is a MergeView (subclass
* of View), then digest will be non-null and has to be set before installing the view.
*/
public void installView(View new_view, Digest digest) {
Address coord;
int rc;
ViewId vid=new_view.getVid();
List<Address> mbrs=new_view.getMembers();
// Discards view with id lower than our own. Will be installed without check if first view
if(view_id != null) {
rc=vid.compareTo(view_id);
if(rc <= 0) {
if(log.isWarnEnabled() && rc < 0) // only scream if view is smaller, silently discard same views
log.warn(local_addr + ": received view < current view;" +
" discarding it (current vid: " + view_id + ", new vid: " + vid + ')');
return;
}
}
if(digest != null) {
if(new_view instanceof MergeView)
mergeDigest(digest);
else
setDigest(digest);
}
if(log.isDebugEnabled()) log.debug(local_addr + ": view is " + new_view);
if(stats) {
num_views++;
prev_views.add(new_view);
}
ack_collector.handleView(new_view);
merge_ack_collector.handleView(new_view);
ltime=Math.max(vid.getId(), ltime); // compute Lamport logical time
/* Check for self-inclusion: if I'm not part of the new membership, I just discard it.
This ensures that messages sent in view V1 are only received by members of V1 */
if(checkSelfInclusion(mbrs) == false) {
if(log.isWarnEnabled()) log.warn(local_addr + ": not member of view " + new_view + "; discarding it");
return;
}
synchronized(members) { // serialize access to views
// assign new_view to view_id
if(new_view instanceof MergeView)
view=new View(new_view.getVid(), new_view.getMembers());
else
view=new_view;
view_id=vid.copy();
// Set the membership. Take into account joining members
if(mbrs != null && !mbrs.isEmpty()) {
members.set(mbrs);
tmp_members.set(members);
joining.removeAll(mbrs); // remove all members in mbrs from joining
// remove all elements from 'leaving' that are not in 'mbrs'
leaving.retainAll(mbrs);
tmp_members.add(joining); // add members that haven't yet shown up in the membership
tmp_members.remove(leaving); // remove members that haven't yet been removed from the membership
// add to prev_members
for(Address addr: mbrs) {
if(!prev_members.contains(addr))
prev_members.add(addr);
}
}
// Send VIEW_CHANGE event up and down the stack:
Event view_event=new Event(Event.VIEW_CHANGE, new_view);
// changed order of passing view up and down (http://jira.jboss.com/jira/browse/JGRP-347)
// changed it back (bela Sept 4 2007): http://jira.jboss.com/jira/browse/JGRP-564
down_prot.down(view_event); // needed e.g. by failure detector or UDP
up_prot.up(view_event);
coord=determineCoordinator();
// if(coord != null && coord.equals(local_addr) && !(coord.equals(vid.getCoordAddress()))) {
// changed on suggestion by yaronr and Nicolas Piedeloupe
if(coord != null && coord.equals(local_addr) && !haveCoordinatorRole()) {
becomeCoordinator();
}
else {
if(haveCoordinatorRole() && !local_addr.equals(coord)) {
becomeParticipant();
merge_ack_collector.reset(null); // we don't need this one anymore
}
}
}
}
protected Address determineCoordinator() {
synchronized(members) {
return members.size() > 0? members.elementAt(0) : null;
}
}
/** Checks whether the potential_new_coord would be the new coordinator (2nd in line) */
protected boolean wouldBeNewCoordinator(Address potential_new_coord) {
Address new_coord;
if(potential_new_coord == null) return false;
synchronized(members) {
if(members.size() < 2) return false;
new_coord=members.elementAt(1); // member at 2nd place
return new_coord != null && new_coord.equals(potential_new_coord);
}
}
/** Returns true if local_addr is member of mbrs, else false */
protected boolean checkSelfInclusion(List<Address> mbrs) {
if(mbrs == null)
return false;
for(Address mbr: mbrs) {
if(mbr != null && local_addr.equals(mbr))
return true;
}
return false;
}
/** Send down a SET_DIGEST event */
public void setDigest(Digest d) {
down_prot.down(new Event(Event.SET_DIGEST, d));
}
/** Send down a MERGE_DIGEST event */
public void mergeDigest(Digest d) {
down_prot.down(new Event(Event.MERGE_DIGEST, d));
}
/** Sends down a GET_DIGEST event and waits for the GET_DIGEST_OK response, or
timeout, whichever occurs first */
public Digest getDigest() {
return (Digest)down_prot.down(Event.GET_DIGEST_EVT);
}
boolean startFlush(View view) {
return _startFlush(view, 4, 1000L, 5000L);
}
boolean startFlush(View view, int maxAttempts, long floor, long ceiling) {
return _startFlush(view, maxAttempts, floor, ceiling);
}
protected boolean _startFlush(final View new_view, int maxAttempts, long randomFloor, long randomCeiling) {
if(flushInvokerClass != null) {
try {
Callable<Boolean> invoker = flushInvokerClass.getDeclaredConstructor(View.class).newInstance(new_view);
return invoker.call();
} catch (Throwable e) {
return false;
}
}
try {
boolean successfulFlush=false;
boolean validView=new_view != null && new_view.size() > 0;
if(validView && flushProtocolInStack) {
int attemptCount = 0;
while (attemptCount < maxAttempts) {
try {
up_prot.up(new Event(Event.SUSPEND, new ArrayList<Address>(new_view.getMembers())));
successfulFlush = true;
break;
} catch (Exception e) {
Util.sleepRandom(randomFloor, randomCeiling);
attemptCount++;
}
}
if(successfulFlush) {
if(log.isTraceEnabled())
log.trace(local_addr + ": successful GMS flush by coordinator");
}
else {
if(log.isWarnEnabled())
log.warn(local_addr + ": GMS flush by coordinator failed");
}
}
return successfulFlush;
} catch (Exception e) {
return false;
}
}
void stopFlush() {
if(flushProtocolInStack) {
if(log.isDebugEnabled()) {
log.debug(local_addr + ": sending RESUME event");
}
up_prot.up(new Event(Event.RESUME));
}
}
void stopFlush(List<Address> members) {
if(log.isDebugEnabled()){
log.debug(local_addr + ": sending RESUME event");
}
up_prot.up(new Event(Event.RESUME,members));
}
@SuppressWarnings("unchecked")
public Object up(Event evt) {
switch(evt.getType()) {
case Event.MSG:
Message msg=(Message)evt.getArg();
GmsHeader hdr=(GmsHeader)msg.getHeader(this.id);
if(hdr == null)
break;
switch(hdr.type) {
case GmsHeader.JOIN_REQ:
view_handler.add(new Request(Request.JOIN, hdr.mbr, false, null, hdr.useFlushIfPresent));
break;
case GmsHeader.JOIN_REQ_WITH_STATE_TRANSFER:
view_handler.add(new Request(Request.JOIN_WITH_STATE_TRANSFER, hdr.mbr, false, null, hdr.useFlushIfPresent));
break;
case GmsHeader.JOIN_RSP:
impl.handleJoinResponse(hdr.join_rsp);
break;
case GmsHeader.LEAVE_REQ:
if(log.isDebugEnabled())
log.debug("received LEAVE_REQ for " + hdr.mbr + " from " + msg.getSrc());
if(hdr.mbr == null) {
return null;
}
view_handler.add(new Request(Request.LEAVE, hdr.mbr, false));
break;
case GmsHeader.LEAVE_RSP:
impl.handleLeaveResponse();
break;
case GmsHeader.VIEW:
View new_view=hdr.view;
if(new_view == null)
return null;
Address coord=msg.getSrc();
if(!new_view.containsMember(coord)) {
sendViewAck(coord); // we need to send the ack first, otherwise the connection is removed
impl.handleViewChange(new_view, hdr.my_digest);
}
else {
impl.handleViewChange(new_view, hdr.my_digest);
sendViewAck(coord); // send VIEW_ACK to sender of view
}
break;
case GmsHeader.VIEW_ACK:
Address sender=msg.getSrc();
ack_collector.ack(sender);
return null; // don't pass further up
case GmsHeader.MERGE_REQ:
down_prot.down(new Event(Event.SUSPEND_STABLE, 20000));
impl.handleMergeRequest(msg.getSrc(), hdr.merge_id, hdr.mbrs);
break;
case GmsHeader.MERGE_RSP:
MergeData merge_data=new MergeData(msg.getSrc(), hdr.view, hdr.my_digest);
merge_data.merge_rejected=hdr.merge_rejected;
if(log.isDebugEnabled()) {
log.debug(local_addr + ": got merge response from " + msg.getSrc() +
", merge_id=" + hdr.merge_id + ", merge data is "+ merge_data);
}
impl.handleMergeResponse(merge_data, hdr.merge_id);
break;
case GmsHeader.INSTALL_MERGE_VIEW:
impl.handleMergeView(new MergeData(msg.getSrc(), hdr.view, hdr.my_digest), hdr.merge_id);
down_prot.down(new Event(Event.RESUME_STABLE));
break;
case GmsHeader.INSTALL_DIGEST:
Digest tmp=hdr.my_digest;
down_prot.down(new Event(Event.MERGE_DIGEST, tmp));
break;
case GmsHeader.INSTALL_MERGE_VIEW_OK:
//[JGRP-700] - FLUSH: flushing should span merge
merge_ack_collector.ack(msg.getSrc());
break;
case GmsHeader.CANCEL_MERGE:
//[JGRP-524] - FLUSH and merge: flush doesn't wrap entire merge process
impl.handleMergeCancelled(hdr.merge_id);
down_prot.down(new Event(Event.RESUME_STABLE));
break;
case GmsHeader.GET_DIGEST_REQ:
Digest digest=(Digest)down_prot.down(Event.GET_DIGEST_EVT);
if(digest != null) {
Digest.Entry entry=digest.get(local_addr);
if(entry != null) {
// only return my own digest information, but nobody else's !
// https://jira.jboss.org/jira/browse/JGRP-948
Digest retval=new Digest(local_addr, entry.getLow(), entry.getHighestDeliveredSeqno(),
entry.getHighestReceivedSeqno());
GmsHeader rsp_hdr=new GmsHeader(GmsHeader.GET_DIGEST_RSP);
rsp_hdr.my_digest=retval;
Message get_digest_rsp=new Message(msg.getSrc(), null, null);
get_digest_rsp.setFlag(Message.OOB);
get_digest_rsp.putHeader(this.id, rsp_hdr);
down_prot.down(new Event(Event.MSG, get_digest_rsp));
}
}
break;
case GmsHeader.GET_DIGEST_RSP:
Digest digest_rsp=hdr.my_digest;
impl.handleDigestResponse(msg.getSrc(), digest_rsp);
break;
default:
if(log.isErrorEnabled()) log.error("GmsHeader with type=" + hdr.type + " not known");
}
return null; // don't pass up
case Event.SUSPECT:
Object retval=up_prot.up(evt);
Address suspected=(Address)evt.getArg();
view_handler.add(new Request(Request.SUSPECT, suspected, true));
ack_collector.suspect(suspected);
merge_ack_collector.suspect(suspected);
return retval;
case Event.UNSUSPECT:
impl.unsuspect((Address)evt.getArg());
return null; // discard
case Event.MERGE:
view_handler.add(new Request(Request.MERGE, null, false, (Map<Address,View>)evt.getArg()));
return null; // don't pass up
}
return up_prot.up(evt);
}
@SuppressWarnings("unchecked")
public Object down(Event evt) {
int type=evt.getType();
switch(type) {
case Event.CONNECT:
case Event.CONNECT_USE_FLUSH:
case Event.CONNECT_WITH_STATE_TRANSFER:
case Event.CONNECT_WITH_STATE_TRANSFER_USE_FLUSH:
boolean use_flush=type == Event.CONNECT_USE_FLUSH || type == Event.CONNECT_WITH_STATE_TRANSFER_USE_FLUSH;
boolean state_transfer=type == Event.CONNECT_WITH_STATE_TRANSFER
|| type == Event.CONNECT_WITH_STATE_TRANSFER_USE_FLUSH;
if(print_local_addr) {
PhysicalAddress physical_addr=print_physical_addrs?
(PhysicalAddress)down(new Event(Event.GET_PHYSICAL_ADDRESS, local_addr)) : null;
System.out.println("\n-------------------------------------------------------------------\n" +
"GMS: address=" + local_addr + ", cluster=" + evt.getArg() +
(physical_addr != null? ", physical address=" + physical_addr : "") +
"\n-------------------------------------------------------------------");
}
else {
if(log.isDebugEnabled()) {
PhysicalAddress physical_addr=print_physical_addrs?
(PhysicalAddress)down(new Event(Event.GET_PHYSICAL_ADDRESS, local_addr)) : null;
log.debug("address=" + local_addr + ", cluster=" + evt.getArg() +
(physical_addr != null? ", physical address=" + physical_addr : ""));
}
}
down_prot.down(evt);
if(local_addr == null)
if(log.isFatalEnabled()) log.fatal("[CONNECT] local_addr is null");
try {
if(state_transfer)
impl.joinWithStateTransfer(local_addr, use_flush);
else
impl.join(local_addr, use_flush);
}
catch(Throwable e) {
return e;
}
return null; // don't pass down: event has already been passed down
case Event.DISCONNECT:
impl.leave((Address)evt.getArg());
if(!(impl instanceof CoordGmsImpl)) {
initState(); // in case connect() is called again
}
down_prot.down(evt); // notify the other protocols, but ignore the result
return null;
case Event.CONFIG :
Map<String,Object> config=(Map<String,Object>)evt.getArg();
if((config != null && config.containsKey("flush_supported"))){
flushProtocolInStack=true;
}
break;
case Event.SET_LOCAL_ADDRESS:
local_addr=(Address)evt.getArg();
break;
}
return down_prot.down(evt);
}
public Map<String, String> handleProbe(String... keys) {
for(String key: keys) {
if(key.equals("fix-digests")) {
fixDigests();
}
}
return null;
}
public String[] supportedKeys() {
return new String[]{"fix-digests"};
}
/* ------------------------------- Private Methods --------------------------------- */
final void initState() {
becomeClient();
view_id=null;
view=null;
}
private void sendViewAck(Address dest) {
Message view_ack=new Message(dest, null, null);
view_ack.setFlag(Message.OOB);
GmsHeader tmphdr=new GmsHeader(GmsHeader.VIEW_ACK);
view_ack.putHeader(this.id, tmphdr);
down_prot.down(new Event(Event.MSG, view_ack));
}
/* --------------------------- End of Private Methods ------------------------------- */
public static class GmsHeader extends Header {
public static final byte JOIN_REQ=1;
public static final byte JOIN_RSP=2;
public static final byte LEAVE_REQ=3;
public static final byte LEAVE_RSP=4;
public static final byte VIEW=5;
public static final byte MERGE_REQ=6;
public static final byte MERGE_RSP=7;
public static final byte INSTALL_MERGE_VIEW=8;
public static final byte CANCEL_MERGE=9;
public static final byte VIEW_ACK=10;
public static final byte JOIN_REQ_WITH_STATE_TRANSFER = 11;
public static final byte INSTALL_MERGE_VIEW_OK=12;
public static final byte GET_DIGEST_REQ=13;
public static final byte GET_DIGEST_RSP=14;
public static final byte INSTALL_DIGEST=15;
byte type=0;