Index: src/org/jgroups/protocols/SEQUENCER.java
===================================================================
RCS file: /cvsroot/javagroups/JGroups/src/org/jgroups/protocols/SEQUENCER.java,v
retrieving revision 1.15
diff -w -u -r1.15 SEQUENCER.java
--- src/org/jgroups/protocols/SEQUENCER.java	12 Jan 2007 14:19:12 -0000	1.15
+++ src/org/jgroups/protocols/SEQUENCER.java	24 Jan 2007 20:18:41 -0000
@@ -1,6 +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,8 +10,6 @@
 
 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
@@ -17,50 +17,30 @@
  * @version $Id: SEQUENCER.java,v 1.15 2007/01/12 14:19:12 belaban Exp $
  */
 public class SEQUENCER extends Protocol {
-    private Address           local_addr=null, coord=null;
+    
     static final String       name="SEQUENCER";
-    private boolean           is_coord=false;
+
+    private final Object lock=new Object();
+    
+    private Address           local_addr=null, coord=null;
+    private boolean           is_coord=false, hold=true;
     private long              seqno=0;
 
     /** Map<seqno, Message>: maintains messages forwarded to the coord which which no ack has been received yet */
-    private final Map               forward_table=new TreeMap();
+    private final Map<Long, Message> forward_table=new TreeMap<Long, Message>();
 
     /** Map<Address, seqno>: maintains the highest seqnos seen for a given member */
-    private final ConcurrentHashMap received_table=new ConcurrentHashMap();
+    private final ConcurrentHashMap<Address, Long> received_table=new ConcurrentHashMap<Address, Long>();
 
-    private long forwarded_msgs=0;
-    private long bcast_msgs=0;
-    private long received_forwards=0;
-    private long received_bcasts=0;
+    private View lastView;
 
+    private SynchronizingThread synchronizingThread;
+    private Set<Address> nonSynchronizedMembers;
+    
+    public String getName() {return name;}
+    public Address getLocalAddress() {return local_addr;}
     public boolean isCoordinator() {return is_coord;}
     public Address getCoordinator() {return coord;}
-    public Address getLocalAddress() {return local_addr;}
-    public String getName() {return name;}
-    public long getForwarded() {return forwarded_msgs;}
-    public long getBroadcast() {return bcast_msgs;}
-    public long getReceivedForwards() {return received_forwards;}
-    public long getReceivedBroadcasts() {return received_bcasts;}
-
-    public void resetStats() {
-        forwarded_msgs=bcast_msgs=received_forwards=received_bcasts=0L;
-    }
-
-    public Map dumpStats() {
-        Map m=super.dumpStats();
-        if(m == null)
-            m=new HashMap();
-        m.put("forwarded", new Long(forwarded_msgs));
-        m.put("broadcast", new Long(bcast_msgs));
-        m.put("received_forwards", new Long(received_forwards));
-        m.put("received_bcasts", new Long(received_bcasts));
-        return m;
-    }
-
-    public String printStats() {
-        return dumpStats().toString();
-    }
-
 
     public boolean setProperties(Properties props) {
         super.setProperties(props);
@@ -73,45 +53,35 @@
     }
 
     private final long nextSeqno() {
-        synchronized(this) {
+        synchronized(lock) {
             return seqno++;
         }
     }
 
-
     public Object down(Event evt) {
-        switch(evt.getType()) {
-            case Event.MSG:
+        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
-                    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);
+        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
                     }
-                    else {
-                        broadcast(msg);
                     }
-                    return null; // don't pass down
-                }
-                break;
 
-            case Event.VIEW_CHANGE:
-                handleViewChange((View)evt.getArg());
-                break;
-        }
+        // Unicast or unordered multicast
         return down_prot.down(evt);
     }
 
-
-
-
     public Object up(Event evt) {
-        Message msg;
-        SequencerHeader hdr;
-
         switch(evt.getType()) {
 
             case Event.SET_LOCAL_ADDRESS:
@@ -119,30 +89,40 @@
                 break;
 
             case Event.MSG:
-                msg=(Message)evt.getArg();
-                hdr=(SequencerHeader)msg.getHeader(name);
-                if(hdr == null)
-                    break; // pass up
+                Message msg=(Message)evt.getArg();
+                SequencerHeader 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);
+                        handleTotalBroadcast(msg, false);
                         received_forwards++;
                         return null;
                     case SequencerHeader.BCAST:
-                        deliver(msg, hdr);  // deliver a copy and return (discard the original msg)
+                        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,64 +134,136 @@
 
     /* --------------------------------- Private Methods ----------------------------------- */
 
-    private void handleViewChange(View v) {
-        Vector members=v.getMembers();
+    private void handleViewChange(View view) {
+        synchronized(lock) {
+            lastView=view;
+            hold=true;
+            is_coord=false;
+            
+            Vector members=lastView.getMembers();
+            
         if(members.isEmpty()) return;
 
-        Address prev_coord=coord;
+            // 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);
 
-        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();
+            Set<Address> keys=received_table.keySet();
         keys.retainAll(members);
-        if(keys.size() != size) {
-            if(trace)
-                log.trace("adjusted received_table, keys are " + keys);
-        }
+            
+            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);
     }
 
-    /**
-     * 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);
+        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;
 
-    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));
+                event=new Event(Event.MSG, msg);
         forwarded_msgs++;
-    }
 
-    private void broadcast(Message msg) {
-        SequencerHeader hdr=(SequencerHeader)msg.getHeader(name);
+                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()
-        down_prot.down(new Event(Event.MSG, msg));
+                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,
@@ -219,21 +271,20 @@
      * @param msg
      * @param hdr
      */
-    private void deliver(Message msg, SequencerHeader hdr) {
+    private Event deliver(Message msg, SequencerHeader hdr) {
         Address original_sender=hdr.getOriginalSender();
         if(original_sender == null) {
             if(log.isErrorEnabled())
                 log.error("original sender is null, cannot swap sender address back to original sender");
-            return;
+            return null;
         }
         long msg_seqno=hdr.getSeqno();
 
+        synchronized(lock) {
         // this is the ack for the message sent by myself
         if(original_sender.equals(local_addr)) {
-            synchronized(forward_table) {
                 forward_table.remove(new Long(msg_seqno));
             }
-        }
 
         // if msg was already delivered, discard it
         Long highest_seqno_seen=(Long)received_table.get(original_sender);
@@ -242,16 +293,17 @@
                 if(log.isWarnEnabled())
                 log.warn("message seqno (" + original_sender + "::" + msg_seqno + " has already " +
                         "been received (highest received=" + highest_seqno_seen + "); discarding duplicate message");
-                return;
+                    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);
-        up_prot.up(new Event(Event.MSG, tmp));
+        return new Event(Event.MSG, tmp);
     }
 
     /* ----------------------------- End of Private Methods -------------------------------- */
@@ -260,14 +312,17 @@
 
 
 
-    public static class SequencerHeader extends Header implements Streamable {
+    public static final class SequencerHeader extends Header implements Streamable {
         static final byte FORWARD = 1;
         static final byte BCAST   = 2;
+        static final byte SYNC_REQUEST = 3;
+        static final byte SYNC_ACK = 4;
+        static final byte SYNC_COMPLETE = 5;
 
-        byte    type=-1;
-        /** the original sender's address and a seqno */
-        ViewId  tag=null;
+        private byte type=-1;
 
+        /** the original sender's address and a seqno, or the sync ViewId */
+        private ViewId tag=null;
 
         public SequencerHeader() {
         }
@@ -277,6 +332,11 @@
             this.tag=new ViewId(original_sender, seqno);
         }
 
+        public SequencerHeader(byte type, ViewId viewId) {
+            this.type=type;
+            this.tag=viewId;
+        }
+
         public Address getOriginalSender() {
             return tag != null? tag.getCoordAddress() : null;
         }
@@ -297,6 +357,9 @@
             switch(type) {
                 case FORWARD: return "FORWARD";
                 case BCAST:   return "BCAST";
+                case SYNC_REQUEST: return "SYNC_REQUEST";
+                case SYNC_ACK: return "SYNC_ACK";
+                case SYNC_COMPLETE: return "SYNC_COMPLETE";
                 default:      return "n/a";
             }
         }
@@ -306,7 +369,6 @@
             out.writeObject(tag);
         }
 
-
         public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
             type=in.readByte();
             tag=(ViewId)in.readObject();
@@ -323,13 +385,82 @@
         }
 
         public long size() {
-            long size=Global.BYTE_SIZE *2; // type + presence byte
-            if(tag != null)
-                size+=tag.serializedSize();
+            long size=Global.BYTE_SIZE; // type
+            size+=tag!=null ? (Global.BYTE_SIZE+tag.serializedSize()) : Global.BYTE_SIZE; // tag
             return size;
         }
 
     }
 
+    private void passDownSync(byte type, ViewId viewId) {
+        Message msg=new Message();
+        msg.putHeader(name, new SequencerHeader(type, viewId));       
+        
+        down_prot.down(new Event(Event.MSG, msg));
+    }
+    
+    private class SynchronizingThread extends Thread {
+
+        public SynchronizingThread() {
+            log.debug("New sync thread: "+this);
+        }
+
+        public void run() {
+            // As long as we're the synchronizing thread, send a SYNC_REQUEST message down every 100 ms.
+            for(;;) {
+                ViewId viewId;
+                
+                synchronized(lock) {
+                    if (synchronizingThread!=this) {
+                        log.debug("Sync thread is: "+synchronizingThread+" != "+this);
+                        return; // New view...
+                    }
+                    viewId=lastView.getVid();
+                    
+                    log.debug("Sending sync request; lastView="+lastView);
+                }
+                
+                passDownSync(SequencerHeader.SYNC_REQUEST, viewId);
+                
+                synchronized(lock) {
+                    try {
+                        lock.wait(100);
+                    } catch (InterruptedException e) {
+                        // Ignore
+                    }
+                }
+            }
+        }
+        
+    }
+
+    //
+    //  Stats
+    //
+    
+    private long forwarded_msgs=0, bcast_msgs=0, received_forwards=0, received_bcasts=0;
+
+    public long getForwarded() {return forwarded_msgs;}
+    public long getBroadcast() {return bcast_msgs;}
+    public long getReceivedForwards() {return received_forwards;}
+    public long getReceivedBroadcasts() {return received_bcasts;}
+
+    public void resetStats() {
+        forwarded_msgs=bcast_msgs=received_forwards=received_bcasts=0L;
+    }
+
+    public Map dumpStats() {
+        Map m=super.dumpStats();
+        if(m == null) m=new HashMap();
+        m.put("forwarded", new Long(forwarded_msgs));
+        m.put("broadcast", new Long(bcast_msgs));
+        m.put("received_forwards", new Long(received_forwards));
+        m.put("received_bcasts", new Long(received_bcasts));
+        return m;
+    }
+
+    public String printStats() {
+        return dumpStats().toString();
+    }
 
 }
\ No newline at end of file
