FLUSH


1   // $Id: FLUSH.java,v 1.8 2004/09/23 16:29:41 belaban Exp $
2   
3   
4   
5   package org.jgroups.protocols;
6   
7   
8   import org.jgroups.Address;
9   import org.jgroups.Event;
10  import org.jgroups.View;
11  import org.jgroups.blocks.GroupRequest;
12  import org.jgroups.blocks.MethodCall;
13  import org.jgroups.stack.RpcProtocol;
14  import org.jgroups.util.List;
15  import org.jgroups.util.Rsp;
16  import org.jgroups.util.RspList;
17  import org.jgroups.util.Util;
18  
19  import java.util.Enumeration  ;
20  import java.util.Properties  ;
21  import java.util.Vector  ;
22  
23  
24  
25  
26  /**
27     The task of the FLUSH protocol is to flush all pending messages out of the system. This is
28     done before a view change by stopping all senders and then agreeing on what messages
29     should be delivered in the current view (before switching to the new view). A coordinator
30     broadcasts a FLUSH message. The message contains an array of the highest sequence number for each member
31     as seen by the coordinator so far. Each member responds with its highest sequence numbers seen so far (for
32     each member): if its sequence number for a member P is higher than the one sent by the coordinator, it
33     will append the messages apparently not received by the coordinator to its reply. The coordinator (when
34     all replies have been received), computes for each member the lowest and highest sequence number and
35     re-broadcasts messages accordingly (using ACKs rather then NAKs to ensure reliable delivery).<p> Example:
36     <pre>
37  
38     FLUSH  ---> (p=10, q=22, r=7)
39  
40     <-- (p=10, q=20, r=7)    (did not receive 2 messages from q)
41     <-- (p=12, q=23, r=7)    (attached are messages p11, p12, and q23)
42     <-- (p=10, q=22, r=8)    (attached is message r8)
43     ---------------------
44     min:   11    21    8
45     max:   12    23    8
46     </pre>
47  
48     The coordinator now computes the range for each member and re-broadcasts messages
49     p11, p12, q21, q22, q23 and r8.
50     This is essentially the exclusive min and inclusive max of all replies. Note that messages p11, p12 and q23
51     were not received by the coordinator itself before. They were only returned as result of the FLUSH replies
52     and the coordinator now re-broadcasts them.
53  
54  */
55  public class FLUSH extends RpcProtocol {
56      final Vector     mbrs=new Vector  ();
57      boolean  is_server=false;
58      final Object     block_mutex=new Object  ();
59      long     block_timeout=5000;
60      Address  local_addr=null;
61      boolean  blocked=false;  // BLOCK: true, VIEW_CHANGE: false
62      final Object     digest_mutex=new Object  ();
63      long     digest_timeout=2000;   // time to wait for retrieval of unstable msgs
64  
65      final Object     highest_delivered_mutex=new Object  ();
66      long[]   highest_delivered_msgs;
67  
68      Digest   digest=null;
69  
70      final Object     get_msgs_mutex=new Object  ();
71      final long     get_msgs_timeout=4000;
72      List     get_msgs=null;
73  
74  
75  
76      public String    getName() {return "FLUSH";}
77  
78  
79      public Vector   providedUpServices() {
80      Vector   retval=new Vector  ();
81      retval.addElement(new Integer  (Event.FLUSH));
82      return retval;
83      }
84  
85      public Vector   requiredDownServices() {
86      Vector   retval=new Vector  ();
87      retval.addElement(new Integer  (Event.GET_MSGS_RECEIVED));  // NAKACK layer
88      retval.addElement(new Integer  (Event.GET_MSG_DIGEST));     // NAKACK layer
89      retval.addElement(new Integer  (Event.GET_MSGS));           // NAKACK layer
90      return retval;
91      }
92  
93  
94      public void start() throws Exception   {
95          super.start();
96          if(_corr != null) {
97              _corr.setDeadlockDetection(true);
98          }
99          else
100             throw new Exception  ("FLUSH.start(): cannot set deadlock detection in corr, as it is null !");
101     }
102 
103 
104     /**
105        Triggered by reception of FLUSH event from GMS layer (must be coordinator). Calls
106        <code>HandleFlush</code> in all members and returns FLUSH_OK event.
107        @param dests A list of members to which the FLUSH is to be sent
108        @return FlushRsp Contains result (true or false), list of unstable messages and list of members
109            failed during the FLUSH.
110      */
111     private FlushRsp flush(Vector   dests) {
112     RspList     rsp_list;
113     FlushRsp    retval=new FlushRsp();
114     Digest      digest;
115     long[]      min, max;
116     long[]      lower[];
117     List        unstable_msgs=new List();
118     boolean     get_lower_msgs=false;
119 
120     highest_delivered_msgs=new long[members.size()];
121     min=new long[members.size()];
122     max=new long[members.size()];
123 
124 
125     /* Determine the highest seqno (for each member) that was delivered to the application
126        (i.e., consumed by the application). Stores result in array 'highest_delivered_msgs' */
127     getHighestDeliveredSeqnos();
128 
129     for(int i=0; i < highest_delivered_msgs.length; i++)
130         min[i]=max[i]=highest_delivered_msgs[i];
131 
132 
133     /* Call the handleFlush() method of all existing members. The highest seqnos seen by the coord
134        is the argument */
135      if(log.isInfoEnabled()) log.info("calling handleFlush(" + dests + ')');
136     passDown(new Event(Event.SWITCH_OUT_OF_BAND)); // we need out-of-band control for FLUSH ...
137     MethodCall call = new MethodCall("handleFlush", new Object  [] {dests, highest_delivered_msgs.clone()}, 
138         new String  [] {Vector  .class.getName(), long[].class.getName()});
139     rsp_list=callRemoteMethods(dests, call, GroupRequest.GET_ALL, 0);
140      if(log.isInfoEnabled()) log.info("flush done");
141 
142 
143     /* Process all the responses (Digest): compute a range of messages (min and max seqno) for each
144        member that has to be re-broadcast; FlushRsp contains those messages. They will be re-braodcast
145        by the cordinator (in the GMS protocol). */
146     for(int i=0; i < rsp_list.size(); i++) {
147         Rsp rsp=(Rsp)rsp_list.elementAt(i);
148         if(rsp.wasReceived()) {
149         digest=(Digest)rsp.getValue();
150         if(digest != null) {
151             for(int j=0; j < digest.highest_seqnos.length && j < min.length; j++) {
152             min[j]=Math.min(min[j], digest.highest_seqnos[j]);
153             max[j]=Math.max(max[j], digest.highest_seqnos[j]);
154             }
155             if(digest.msgs.size() > 0) {
156             for(Enumeration   e=digest.msgs.elements(); e.hasMoreElements();)
157                 unstable_msgs.add(e.nextElement());
158             }
159         }
160         }
161     } // end for-loop
162 
163 
164 
165     /* If any of the highest msgs of the flush replies were lower than the ones sent by this
166        coordinator, we have to re-broadcast them. (This won't occur often)
167        Compute the range between min and highest_delivered_msgs */
168     lower=new long[min.length][]; // stores (for each mbr) the range of seqnos (e.g. 20 24): send msgs
169                       // 21, 22 and 23 and 24 (excluding lower and including upper range)
170 
171     for(int i=0; i < min.length; i++) {
172         if(min[i] < highest_delivered_msgs[i]) {    // will almost never be the case
173         lower[i]=new long[2];
174         lower[i][0]=min[i];                     // lower boundary (excluding)
175         lower[i][1]=highest_delivered_msgs[i];  // upper boundary (including)
176         get_lower_msgs=true;
177         }
178     }
179     if(get_lower_msgs) {
180         get_msgs=null;
181         synchronized(get_msgs_mutex) {
182         passDown(new Event(Event.GET_MSGS, lower));
183         try {
184             get_msgs_mutex.wait(get_msgs_timeout);
185         }
186         catch(Exception   e) {}
187         }
188         if(get_msgs != null) {
189         for(Enumeration   e=get_msgs.elements(); e.hasMoreElements();)
190             unstable_msgs.add(e.nextElement());
191         }
192     }
193     retval.unstable_msgs=unstable_msgs.getContents();
194     if(rsp_list.numSuspectedMembers() > 0) {
195         retval.result=false;
196         retval.failed_mbrs=rsp_list.getSuspectedMembers();
197     }
198 
199     return retval;
200     }
201 
202 
203 
204 
205 
206     /**
207        Called by coordinator running the FLUSH protocol. Argument is an array of the highest seqnos as seen
208        by the coordinator (for each member). <code>handleFlush()</code> checks for each member its
209        own highest seqno seen for that member. If it is higher than the one seen by the coordinator,
210        all higher messages are attached to the return value (a message digest).
211        @param flush_dests  The members to which this message is sent. Processes not in this list just
212                ignore the handleFlush() message.
213        @param highest_seqnos The highest sequence numbers (order corresponding to membership) as seen
214                  by coordinator.
215        @return Digest An array of the highest seqnos for each member, as seen by this member. If this
216               member's seqno for a member P is higher than the one in <code>highest_seqnos</code>,
217               the missing messages are added to the message digest as well. This allows the
218               coordinator to re-broadcast missing messages.
219      */
220     public synchronized Digest handleFlush(Vector   flush_dests, long[] highest_seqnos) {
221     digest=null;
222 
223      if(log.isInfoEnabled()) log.info("flush_dests=" + flush_dests +
224                    " , highest_seqnos=" + Util.array2String(highest_seqnos));
225 
226     if(!is_server) // don't handle the FLUSH if not yet joined to the group
227         return digest;
228 
229     if(flush_dests == null) {
230          if(log.isWarnEnabled()) log.warn("flush dest is null, ignoring flush !");
231         return digest;
232     }
233 
234     if(flush_dests.size() == 0) {
235          if(log.isWarnEnabled()) log.warn("flush dest is empty, ignoring flush !");
236         return digest;
237     }
238 
239     if(!flush_dests.contains(local_addr)) {
240 
241         if(log.isWarnEnabled()) log.warn("am not in the flush dests, ignoring flush");
242         return digest;
243     }
244 
245     // block sending of messages (only if not already blocked !)
246     if(!blocked) {
247         blocked=true;
248         synchronized(block_mutex) {
249         passUp(new Event(Event.BLOCK));
250         try {block_mutex.wait(block_timeout);}
251         catch(Exception   e) {}
252         }
253     }
254 
255     // asks NAKACK layer for unstable messages and saves result in 'digest'
256     getMessageDigest(highest_seqnos);
257      if(log.isInfoEnabled()) log.info("returning digest : " + digest);
258     return digest;
259     }
260 
261 
262 
263 
264 
265 
266     /** Returns the highest seqnos (for each member) seen so far (using the NAKACK layer) */
267     void getHighestDeliveredSeqnos() {
268     synchronized(highest_delivered_mutex) {
269         passDown(new Event(Event.GET_MSGS_RECEIVED));
270         try {
271         highest_delivered_mutex.wait(4000);
272         }
273         catch(Exception   e) {
274         if(log.isDebugEnabled()) log.debug("exception is " + e);
275         }
276     }
277     }
278 
279 
280 
281 
282 
283     /** Interacts with a lower layer to retrieve unstable messages (e.g. NAKACK) */
284     void getMessageDigest(long[] highest_seqnos) {
285     synchronized(digest_mutex) {
286         passDown(new Event(Event.GET_MSG_DIGEST, highest_seqnos));
287         try {
288         digest_mutex.wait(digest_timeout);
289         }
290         catch(Exception   e) {}
291     }
292     }
293 
294 
295 
296 
297 
298 
299 
300     /**
301        <b>Callback</b>. Called by superclass when event may be handled.<p>
302        <b>Do not use <code>PassUp</code> in this method as the event is passed up
303        by default by the superclass after this method returns !</b>
304        @return boolean Defaults to true. If false, event will not be passed up the stack.
305      */
306     public boolean handleUpEvent(Event evt) {
307     switch(evt.getType()) {
308 
309     case Event.SET_LOCAL_ADDRESS:
310         local_addr=(Address)evt.getArg();
311         break;
312 
313     case Event.GET_MSG_DIGEST_OK:
314         synchronized(digest_mutex) {
315         digest=(Digest)evt.getArg();
316         digest_mutex.notifyAll();
317         }
318         return false;  // don't pass further up
319 
320     case Event.GET_MSGS_RECEIVED_OK:
321         long[] tmp=(long[])evt.getArg();
322         if(tmp != null)
323             System.arraycopy(tmp, 0, highest_delivered_msgs, 0, tmp.length);
324         synchronized(highest_delivered_mutex) {
325         highest_delivered_mutex.notifyAll();
326         }
327         return false; // don't pass up any further !
328 
329     case Event.GET_MSGS_OK:
330         synchronized(get_msgs_mutex) {
331         get_msgs=(List)evt.getArg();
332         get_msgs_mutex.notifyAll();
333         }
334         break;
335 
336     }
337     return true;
338     }
339 
340 
341     /**
342        <b>Callback</b>. Called by superclass when event may be handled.<p>
343        <b>Do not use <code>PassDown</code> in this method as the event is passed down
344        by default by the superclass after this method returns !</b>
345        @return boolean Defaults to true. If false, event will not be passed down the stack.
346     */
347     public boolean handleDownEvent(Event evt) {
348     Vector      dests;
349     FlushRsp  rsp;
350 
351     switch(evt.getType()) {
352     case Event.FLUSH:
353         dests=(Vector  )evt.getArg();
354         if(dests == null) dests=new Vector  ();
355         rsp=flush(dests);
356         passUp(new Event(Event.FLUSH_OK, rsp));
357         return false; // don't pass down
358 
359     case Event.BECOME_SERVER:
360         is_server=true;
361         break;
362 
363     case Event.VIEW_CHANGE:
364         blocked=false;
365 
366         Vector   tmp=((View)evt.getArg()).getMembers();
367         if(tmp != null) {
368         mbrs.removeAllElements();
369         for(int i=0; i < tmp.size(); i++)
370             mbrs.addElement(tmp.elementAt(i));
371         }
372         break;
373     }
374     return true;
375     }
376 
377 
378 
379 
380 
381     /**
382        The default handling adds the event to the down-queue where events are handled in order of
383        addition by a thread. However, there exists a deadlock between the FLUSH and BLOCK_OK down
384        events: when a FLUSH event is received, a BLOCK is sent up, which triggers a BLOCK_OK event
385        to be sent down to be handled by the FLUSH layer. However, the FLUSH layer's thread is still
386        processing the FLUSH down event and is therefore blocked, waiting for a BLOCK_OK event.
387        Therefore, the BLOCK_OK event has to 'preempt' the FLUSH event processing. This is done by
388        overriding this method: when a BLOCK_OK event is received, it is processed immediately
389        (in parallel to the FLUSH event), which causes the FLUSH event processing to return.
390     */
391     public void receiveDownEvent(Event evt) {
392     if(evt.getType() == Event.BLOCK_OK) { // priority handling, otherwise FLUSH would block !
393         synchronized(down_queue) {
394         Event event;
395         try {
396             while(down_queue.size() > 0) {
397             event=(Event)down_queue.remove(10); // wait 10ms at most; queue is *not* empty !
398             down(event);
399             }
400         }
401         catch(Exception   e) {}
402         }
403 
404         synchronized(block_mutex) {
405         block_mutex.notifyAll();
406         }
407         return;
408     }
409     super.receiveDownEvent(evt);
410     }
411 
412 
413 
414     public boolean setProperties(Properties   props) {super.setProperties(props);
415     String       str;
416 
417     str=props.getProperty("block_timeout");
418     if(str != null) {
419         block_timeout=Long.parseLong(str);
420         props.remove("block_timeout");
421     }
422 
423     str=props.getProperty("digest_timeout");
424     if(str != null) {
425         digest_timeout=Long.parseLong(str);
426         props.remove("digest_timeout");
427     }
428 
429     if(props.size() > 0) {
430         System.err.println("EXAMPLE.setProperties(): these properties are not recognized:");
431         props.list(System.out);
432         return false;
433     }
434     return true;
435     }
436 
437 
438 
439 }
440 
441
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags