***************
*** 1,6 ****
  
  package org.jgroups.protocols;
  
  import org.jgroups.*;
  import org.jgroups.stack.Protocol;
  import org.jgroups.util.Streamable;
--- 1,8 ----
  
  package org.jgroups.protocols;
  
+ import java.util.concurrent.ConcurrentHashMap;
+ 
  import org.jgroups.*;
  import org.jgroups.stack.Protocol;
  import org.jgroups.util.Streamable;
***************
*** 8,15 ****
  
  import java.io.*;
  import java.util.*;
- import java.util.concurrent.ConcurrentHashMap;
- 
  
  /**
   * Implementation of total order protocol using a sequencer. Consult doc/design/SEQUENCER.txt for details
--- 10,15 ----
  
  import java.io.*;
  import java.util.*;
  
  /**
   * Implementation of total order protocol using a sequencer. Consult doc/design/SEQUENCER.txt for details
***************
*** 73,117 ****
      }
  
      private final long nextSeqno() {
-         synchronized(this) {
              return seqno++;
          }
      }
  
- 
      public Object down(Event evt) {
-         switch(evt.getType()) {
-             case Event.MSG:
                  Message msg=(Message)evt.getArg();
                  Address dest=msg.getDest();
-                 if(dest == null || dest.isMulticastAddress()) { // only handle multicasts
-                     long next_seqno=nextSeqno();
-                     SequencerHeader hdr=new SequencerHeader(SequencerHeader.FORWARD, local_addr, next_seqno);
-                     msg.putHeader(name, hdr);
-                     if(!is_coord) {
-                         forwardToCoord(msg, next_seqno);
                      }
-                     else {
-                         broadcast(msg);
                      }
-                     return null; // don't pass down
-                 }
-                 break;
  
-             case Event.VIEW_CHANGE:
-                 handleViewChange((View)evt.getArg());
-                 break;
-         }
          return down_prot.down(evt);
      }
  
- 
- 
- 
      public Object up(Event evt) {
-         Message msg;
-         SequencerHeader hdr;
- 
          switch(evt.getType()) {
  
              case Event.SET_LOCAL_ADDRESS:
--- 53,87 ----
      }
  
      private final long nextSeqno() {
+         synchronized(lock) {
              return seqno++;
          }
      }
  
      public Object down(Event evt) {
+         if (evt.getType()!=Event.MSG) {
+             return down_prot.down(evt);
+         }
+         
                  Message msg=(Message)evt.getArg();
                  Address dest=msg.getDest();
+         if(dest == null || dest.isMulticastAddress()) {
+             // Only handle multicasts
+             if (msg.isFlagSet(Message.OOB)) {
+                 // Unordered multicast: just pass down
+                 msg.clearFlag(Message.OOB);
+             } else {
+                 // Ordered multicast
+                 handleTotalBroadcast(msg, true);
+                 return null; // don't pass down here
                      }
                      }
  
+         // Unicast or unordered multicast
          return down_prot.down(evt);
      }
  
      public Object up(Event evt) {
          switch(evt.getType()) {
  
              case Event.SET_LOCAL_ADDRESS:
***************
*** 119,148 ****
                  break;
  
              case Event.MSG:
-                 msg=(Message)evt.getArg();
-                 hdr=(SequencerHeader)msg.getHeader(name);
-                 if(hdr == null)
-                     break; // pass up
  
                  switch(hdr.type) {
                      case SequencerHeader.FORWARD:
-                         if(!is_coord) {
-                             if(log.isErrorEnabled())
-                                 log.warn("I (" + local_addr + ") am not the coord and don't handle " +
-                                         "FORWARD requests, ignoring request");
-                             return null;
-                         }
-                         broadcast(msg);
                          received_forwards++;
                          return null;
                      case SequencerHeader.BCAST:
-                         deliver(msg, hdr);  // deliver a copy and return (discard the original msg)
                          received_bcasts++;
                          return null;
                  }
                  break;
  
              case Event.VIEW_CHANGE:
                  handleViewChange((View)evt.getArg());
                  break;
          }
--- 89,128 ----
                  break;
  
              case Event.MSG:
+                 Message msg=(Message)evt.getArg();
+                 SequencerHeader hdr=(SequencerHeader)msg.getHeader(name);
+ 
+                 if (hdr == null) break; // pass up
  
                  switch(hdr.type) {
                      case SequencerHeader.FORWARD:
+                         handleTotalBroadcast(msg, false);
                          received_forwards++;
                          return null;
                      case SequencerHeader.BCAST:
+                         evt=deliver(msg, hdr);  // deliver a copy and return (discard the original msg)
                          received_bcasts++;
+                         if (evt==null) return null;
+                         break;
+                     case SequencerHeader.SYNC_REQUEST:
+                         ViewId viewId=null;
+                         synchronized(lock) {
+                             log.debug("Got sync request from "+msg.getSrc()+"; lastView="+lastView);
+                             if (lastView!=null) viewId=lastView.getVid();
+                         }
+                         if (viewId!=null) passDownSync(SequencerHeader.SYNC_ACK, viewId);
+                         return null;
+                     case SequencerHeader.SYNC_ACK:
+                         receivedAck(msg.getSrc(), hdr);
+                         return null;
+                     case SequencerHeader.SYNC_COMPLETE:
+                         installView(hdr.tag);
                          return null;
                  }
                  break;
  
              case Event.VIEW_CHANGE:
+                 log.debug("Up: VIEW_CHANGE: "+(View)evt.getArg());
                  handleViewChange((View)evt.getArg());
                  break;
          }
***************
*** 154,217 ****
  
      /* --------------------------------- Private Methods ----------------------------------- */
  
-     private void handleViewChange(View v) {
-         Vector members=v.getMembers();
          if(members.isEmpty()) return;
  
-         Address prev_coord=coord;
          coord=(Address)members.firstElement();
          is_coord=local_addr != null && local_addr.equals(coord);
  
-         boolean coord_changed=prev_coord != null && !prev_coord.equals(coord);
-         if(coord_changed) {
-             resendMessagesInForwardTable(); // maybe optimize in the future: broadcast directly if coord
-         }
          // remove left members from received_table
-         int size=received_table.size();
-         Set keys=received_table.keySet();
          keys.retainAll(members);
-         if(keys.size() != size) {
-             if(trace)
-                 log.trace("adjusted received_table, keys are " + keys);
-         }
      }
  
-     /**
-      * Sends all messages currently in forward_table to the new coordinator (changing the dest field).
-      * This needs to be done, so the underlying reliable unicast protocol (e.g. UNICAST) adds these messages
-      * to its retransmission mechanism<br/>
-      * Note that we need to resend the messages in order of their seqnos ! We also need to prevent other message
-      * from being inserted until we're done, that's why there's synchronization.
-      */
-     private void resendMessagesInForwardTable() {
-         Message   msg;
-         synchronized(forward_table) {
-             for(Iterator it=forward_table.values().iterator(); it.hasNext();) {
-                 msg=(Message)it.next();
-                 msg.setDest(coord);
                  down_prot.down(new Event(Event.MSG, msg));
              }
          }
      }
  
  
-     private void forwardToCoord(Message msg, long seqno) {
          msg.setDest(coord);  // we change the message dest from multicast to unicast (to coord)
-         synchronized(forward_table) {
-             forward_table.put(new Long(seqno), msg);
-         }
-         down_prot.down(new Event(Event.MSG, msg));
          forwarded_msgs++;
-     }
  
-     private void broadcast(Message msg) {
-         SequencerHeader hdr=(SequencerHeader)msg.getHeader(name);
          hdr.type=SequencerHeader.BCAST; // we change the type of header, but leave the tag intact
          msg.setDest(null); // mcast
          msg.setSrc(local_addr); // the coord is sending it - this will be replaced with sender in deliver()
-         down_prot.down(new Event(Event.MSG, msg));
          bcast_msgs++;
      }
  
      /**
       * We copy the message in order to change the sender's address. If we did this on the original message,
--- 134,269 ----
  
      /* --------------------------------- Private Methods ----------------------------------- */
  
+     private void handleViewChange(View view) {
+         synchronized(lock) {
+             lastView=view;
+             hold=true;
+             is_coord=false;
+             
+             Vector members=lastView.getMembers();
+             
          if(members.isEmpty()) return;
  
+             // If we're the new coordinator, fire up a synchronizing thread
+             if (local_addr!=null && local_addr.equals(members.firstElement())) {
+                 nonSynchronizedMembers=new HashSet<Address>();
+                 nonSynchronizedMembers.addAll(view.getMembers());
+                 synchronizingThread=new SynchronizingThread();
+                 synchronizingThread.start();
+             }
+         }
+     }
+ 
+     private void receivedAck(Address src, SequencerHeader hdr) {
+         log.debug("Got ack from "+src+" for: "+hdr.tag);
+         
+         synchronized(lock) {
+             if (lastView==null) return; // Not in this view (yet)
+             if (nonSynchronizedMembers==null) return; // Not coordinator
+             
+             if (!lastView.getVid().equals(hdr.tag)) {
+                 if (lastView.getMembers().contains(src)) nonSynchronizedMembers.add(src);
+                 return;
+             }
+             
+             nonSynchronizedMembers.remove(src);
+             
+             if (!nonSynchronizedMembers.isEmpty()) return;
+     
+             log.debug("Completing "+hdr.tag);
+             
+             synchronizingThread=null;
+         }
+         
+         passDownSync(SequencerHeader.SYNC_COMPLETE, lastView.getVid());
+     }
+ 
+     private void installView(ViewId id) {
+         Object toResend[];
+         
+         synchronized(lock) {
+             if (lastView==null || !id.equals(lastView.getVid())) return;
+             
+             Vector members=lastView.getMembers();
+            
+             hold=false;        
          coord=(Address)members.firstElement();
          is_coord=local_addr != null && local_addr.equals(coord);
  
          // remove left members from received_table
+             Set<Address> keys=received_table.keySet();
          keys.retainAll(members);
+             
+             if (trace) log.trace("adjusted received_table, keys are " + keys);
+             
+             lastView=null;
+             synchronizingThread=null;
+             nonSynchronizedMembers=null;
+             
+             // Sends all messages currently in forward_table to the new coordinator (changing the dest field).
+             // This needs to be done, so the underlying reliable unicast protocol (e.g. UNICAST) adds these messages
+             // to its retransmission mechanism<br/>
+             // Note that we need to resend the messages in order of their seqnos ! We also need to prevent other message
+             // from being inserted until we're done, that's why there's synchronization.
+             // maybe optimize in the future: broadcast directly if coord
+             log.debug("Resending "+forward_table.size()+" messages");
+             
+             toResend=forward_table.values().toArray();
+             for(int i=0;i<toResend.length;i++) ((Message)toResend[i]).setDest(coord);
      }
  
+         for(int i=0;i<toResend.length;i++) {
+             Message msg=(Message)toResend[i];
                  down_prot.down(new Event(Event.MSG, msg));
+             
+             if (log.isDebugEnabled()) log.debug("Resent 1 message to "+coord);
+         } 
              }
+     
+     private void handleTotalBroadcast(Message msg, boolean local) {
+         Event event;
+         
+         synchronized(lock) {
+             SequencerHeader hdr;
+             
+             if (local) {
+                 long next_seqno=nextSeqno();
+                 hdr=new SequencerHeader(SequencerHeader.FORWARD, local_addr, next_seqno);
+                 msg.putHeader(name, hdr);
+             } else {
+                 hdr=(SequencerHeader)msg.getHeader(name);
          }
+     
+             if (!is_coord) {
+                 if(!local) {
+                     if(log.isErrorEnabled())
+                         log.warn("I (" + local_addr + ") am not the coord and don't handle " +
+                                 "FORWARD requests, ignoring request");
+                     return;
      }
  
+                 forward_table.put(new Long(hdr.getSeqno()), msg);
+         
+                 if (hold) return;
  
          msg.setDest(coord);  // we change the message dest from multicast to unicast (to coord)
+                 event=new Event(Event.MSG, msg);
          forwarded_msgs++;
  
+                 if (log.isDebugEnabled()) log.debug("Forwarding 1 message to "+coord);
+             } else {
+                 if (hold) return; // Discard
+                 
          hdr.type=SequencerHeader.BCAST; // we change the type of header, but leave the tag intact
          msg.setDest(null); // mcast
          msg.setSrc(local_addr); // the coord is sending it - this will be replaced with sender in deliver()
+                 event=new Event(Event.MSG, msg);
          bcast_msgs++;
      }
+         }
+         
+         down_prot.down(event);
+     }
  
      /**
       * We copy the message in order to change the sender's address. If we did this on the original message,
***************
*** 242,257 ****
                  if(log.isWarnEnabled())
                  log.warn("message seqno (" + original_sender + "::" + msg_seqno + " has already " +
                          "been received (highest received=" + highest_seqno_seen + "); discarding duplicate message");
-                 return;
              }
          }
          // update the table with the new seqno
          received_table.put(original_sender, new Long(msg_seqno));
  
          // pass a copy of the message up the stack
          Message tmp=msg.copy(true);
          tmp.setSrc(original_sender);
-         up_prot.up(new Event(Event.MSG, tmp));
      }
  
      /* ----------------------------- End of Private Methods -------------------------------- */
--- 293,309 ----
                  if(log.isWarnEnabled())
                  log.warn("message seqno (" + original_sender + "::" + msg_seqno + " has already " +
                          "been received (highest received=" + highest_seqno_seen + "); discarding duplicate message");
+                     return null;
              }
          }
          // update the table with the new seqno
          received_table.put(original_sender, new Long(msg_seqno));
+         }
  
          // pass a copy of the message up the stack
          Message tmp=msg.copy(true);
          tmp.setSrc(original_sender);
+         return new Event(Event.MSG, tmp);
      }
  
      /* ----------------------------- End of Private Methods -------------------------------- */
