tomcat-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fha...@apache.org
Subject svn commit: r384676 - in /tomcat/container/tc5.5.x/modules/groupcom: src/share/org/apache/catalina/tribes/tipis/LazyReplicatedMap.java src/share/org/apache/catalina/tribes/tipis/RpcChannel.java to-do.txt
Date Fri, 10 Mar 2006 02:04:52 GMT
Author: fhanik
Date: Thu Mar  9 18:04:49 2006
New Revision: 384676

URL: http://svn.apache.org/viewcvs?rev=384676&view=rev
Log:
Working on the replicated map

Modified:
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tipis/LazyReplicatedMap.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tipis/RpcChannel.java
    tomcat/container/tc5.5.x/modules/groupcom/to-do.txt

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tipis/LazyReplicatedMap.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tipis/LazyReplicatedMap.java?rev=384676&r1=384675&r2=384676&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tipis/LazyReplicatedMap.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tipis/LazyReplicatedMap.java
Thu Mar  9 18:04:49 2006
@@ -30,14 +30,26 @@
 import org.apache.catalina.tribes.ChannelListener;
 import java.util.Collection;
 import org.apache.catalina.tribes.MembershipListener;
+import java.io.Externalizable;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import org.apache.catalina.tribes.mcast.McastMember;
+import java.util.Iterator;
+import org.apache.catalina.tribes.ChannelException;
+import java.util.LinkedList;
+import java.util.LinkedHashSet;
+import java.util.ArrayList;
+import java.util.Arrays;
 
 /**
+ * @todo implement periodic sync/transfer 
  * @author Filip Hanik
  * @version 1.0
  */
 public class LazyReplicatedMap extends LinkedHashMap 
     implements RpcCallback, ChannelListener, MembershipListener {
     protected static org.apache.commons.logging.Log log = org.apache.commons.logging.LogFactory.getLog(LazyReplicatedMap.class);
+    protected static long TIME_OUT = 15000;//hard coded timeout
     
 //------------------------------------------------------------------------------    
 //              INSTANCE VARIABLES
@@ -45,6 +57,7 @@
 
     private Channel channel;
     private RpcChannel rpcChannel;
+    private byte[] mapContextName;
     
     
 //------------------------------------------------------------------------------    
@@ -69,13 +82,15 @@
         final String chset = "ISO-8859-1";
         this.channel = channel;
         try {
-            this.rpcChannel = new RpcChannel(mapContextName.getBytes(chset), channel, this);
+            this.mapContextName = mapContextName.getBytes(chset);
         }catch (UnsupportedEncodingException x) {
             log.warn("Unable to encode mapContextName["+mapContextName+"] using getBytes("+chset+")
using default getBytes()",x);
-            this.rpcChannel = new RpcChannel(mapContextName.getBytes(), channel, this);
+            this.mapContextName = mapContextName.getBytes();
         }
+        this.rpcChannel = new RpcChannel(this.mapContextName, channel, this);
         this.channel.addChannelListener(this);
         this.channel.addMembershipListener(this);
+        transferState();
         
     }
     
@@ -93,18 +108,30 @@
         }
         this.rpcChannel = null;
         this.channel = null;
+        super.clear();
     }
     
 //------------------------------------------------------------------------------    
 //              GROUP COM INTERFACES
 //------------------------------------------------------------------------------   
+    public void transferState() {
+        throw new UnsupportedOperationException();
+    }
+    
     /**
-     * 
+     * @todo implement state transfer
      * @param msg Serializable
      * @return Serializable - null if no reply should be sent
      */
     public Serializable replyRequest(Serializable msg, Member sender) {
-        throw new UnsupportedOperationException();
+        if ( !(msg instanceof MapMessage) ) return null;
+        MapMessage mapmsg = (MapMessage)msg;
+        if ( mapmsg.getMsgType() != mapmsg.MSG_RETRIEVE_BACKUP ) return null;
+        
+        MapEntry entry = (MapEntry)super.get(mapmsg.getKey());
+        if ( entry == null ) return null;
+        mapmsg.setValue((Serializable)entry.getValue());
+        return mapmsg;
     }
 
     /**
@@ -114,30 +141,112 @@
      * @param sender Member
      */
     public void leftOver(Serializable msg, Member sender) {
-        throw new UnsupportedOperationException();
+        //ignore left over responses
     }
 
     public void messageReceived(Serializable msg, Member sender) {
         throw new UnsupportedOperationException();
+        //todo implement all the messages that we can receive
     }
     
     public boolean accept(Serializable msg, Member sender) {
-        throw new UnsupportedOperationException();
+        if ( msg instanceof MapMessage ) {
+            return Arrays.equals(mapContextName,((MapMessage)msg).getMapId());
+        }
+        return false;
     }
     
     public void memberAdded(Member member) {
-        
+        //do nothing, we don't care
     }
     public void memberDisappeared(Member member) {
-        
+        //todo move all sessions that are primary here to and have this member as 
+        //a backup
+        Iterator i = super.entrySet().iterator();
+        while ( i.hasNext() ) {
+            Map.Entry e = (Map.Entry)i.next();
+            MapEntry entry = (MapEntry)e.getValue();
+            if ( entry.isPrimary() && member.equals(entry.getBackupNode())) {
+                try {
+                    Member backup = publishEntryInfo(entry.getKey(), entry.getValue());
+                    entry.setBackupNode(backup);
+                }catch ( ChannelException x ) {
+                    log.error("Unable to relocate["+entry.getKey()+"] to a new backup node",x);
+                }
+            }//end if
+        }//while
+    }
+    
+    int currentNode = 0;
+    public Member getNextBackupNode() {
+        Member[] members = channel.getMembers();
+        if ( members.length == 0 ) return null;
+        int node = currentNode++;
+        if ( node >= members.length ) {
+            node = 0;
+            currentNode = 0;
+        }
+        return members[node];
     }
     
+    
+    
 //------------------------------------------------------------------------------    
 //              METHODS TO OVERRIDE    
 //------------------------------------------------------------------------------
-
+    /**
+     * publish info about a map pair (key/value) to other nodes in the cluster
+     * @param key Object
+     * @param value Object
+     * @return Member
+     * @throws ChannelException
+     */
+    protected Member publishEntryInfo(Object key, Object value) throws ChannelException {
+        //select a backup node
+        Member backup = getNextBackupNode();
+        //publish the data out to all nodes
+        MapMessage msg = new MapMessage(this.mapContextName, MapMessage.MSG_PROXY, false,
+                                        (Serializable) key, null, null, backup);
+        channel.send(channel.getMembers(), msg);
+
+        //publish the backup data to one node
+        msg = new MapMessage(this.mapContextName, MapMessage.MSG_BACKUP, false,
+                             (Serializable) key, (Serializable) value, null, backup);
+        channel.send(new Member[] {backup}, msg);
+        return backup;
+    }
+    
     public Object get(Object key) {
-        return super.get(key);
+        MapEntry entry = (MapEntry)super.get(key);
+        if ( entry == null ) return null;
+        if ( !entry.isPrimary() ) {
+            try {
+                MapMessage msg = new MapMessage(mapContextName, MapMessage.MSG_RETRIEVE_BACKUP,
false,
+                                                (Serializable) key, null, null, null);
+                Response[] resp = rpcChannel.send(new Member[] {entry.getBackupNode()},
+                                                  msg, this.rpcChannel.FIRST_REPLY, TIME_OUT);
+                if (resp == null || resp.length == 0) {
+                    //no responses
+                    log.warn("Unable to retrieve object for key:" + key);
+                    return null;
+                }
+                msg = (MapMessage) resp[0].getMessage();
+                
+                Member backup = entry.getBackupNode();
+                if (entry.isBackup()) {
+                    //select a new backup node
+                    backup = publishEntryInfo(key, msg.getValue());
+                }
+                entry.setBackupNode(backup);
+                entry.setBackup(false);
+                entry.setProxy(false);
+                entry.setValue(msg.getValue());
+            } catch (ChannelException x) {
+                log.error("Unable to replicate out data for a LazyReplicatedMap.get operation",
x);
+                return null;
+            }
+        }
+        return entry.getValue();
     }
 
     public boolean containsKey(Object key) {
@@ -145,19 +254,53 @@
     }
 
     public Object put(Object key, Object value) {
-        return super.put(key,value);
+        if ( !(key instanceof Serializable) ) throw new IllegalArgumentException("Key is
not serializable:"+key.getClass().getName());
+        if ( value == null ) return remove(key);
+        if ( !(value instanceof Serializable) ) throw new IllegalArgumentException("Value
is not serializable:"+value.getClass().getName());
+
+        MapEntry entry = new MapEntry((Serializable)key,(Serializable)value);
+        entry.setBackup(false);
+        entry.setProxy(false);
+        
+        Object old = null;
+        
+        //make sure that any old values get removed
+        if ( containsKey(key) ) old = (MapEntry)remove(key);
+        try {
+            Member backup = publishEntryInfo(key, value);
+            entry.setBackupNode(backup);
+        } catch (ChannelException x) {
+            log.error("Unable to replicate out data for a LazyReplicatedMap.put operation",
x);
+        }
+        super.put(key,entry);
+        return old;
     }
 
+    
+
     public void putAll(Map m) {
-        super.putAll(m);
+        Iterator i = m.entrySet().iterator();
+        while ( i.hasNext() ) {
+            Map.Entry entry = (Map.Entry)i.next();
+            put(entry.getKey(),entry.getValue());
+        }
     }
 
     public Object remove(Object key) {
-        return super.remove(key);
+        MapEntry entry = (MapEntry)super.remove(key);
+        MapMessage msg = new MapMessage(mapContextName,MapMessage.MSG_REMOVE,false,(Serializable)key,null,null,null);
+        try {
+            channel.send(channel.getMembers(), msg);
+        } catch ( ChannelException x ) {
+            log.error("Unable to replicate out data for a LazyReplicatedMap.remove operation",x);
+        }
+        return entry!=null?entry.getValue():null;
     }
 
     public void clear() {
-        super.clear();
+        //only delete active keys
+        Iterator keys = keySet().iterator();
+        while ( keys.hasNext() ) remove(keys.next());
     }
 
     public boolean containsValue(Object value) {
@@ -165,19 +308,44 @@
     }
 
     public Object clone() {
-        return super.clone();
+        throw new UnsupportedOperationException("This operation is not valid on a replicated
map");
     }
     
     public Set entrySet() {
-        return super.entrySet();
+        LinkedHashSet set = new LinkedHashSet(super.size());
+        Iterator i = super.entrySet().iterator();
+        while ( i.hasNext() ) {
+            Map.Entry e = (Map.Entry)i.next();
+            MapEntry entry = (MapEntry)e.getValue();
+            if ( entry.isPrimary() ) set.add(entry.getValue());
+        }
+        return set;
     }
     
     public Set keySet() {
-        return super.keySet();
+        //todo implement
+        //should only return keys where this is active.
+        LinkedHashSet set = new LinkedHashSet(super.size());
+        Iterator i = super.entrySet().iterator();
+        while ( i.hasNext() ) {
+            Map.Entry e = (Map.Entry)i.next();
+            MapEntry entry = (MapEntry)e.getValue();
+            if ( entry.isPrimary() ) set.add(entry.getKey());
+        }
+        return set;
     }
     
     public int size() {
-        return super.size();
+        //todo, implement a counter variable instead
+        //only count active members in this node
+        int counter = 0;
+        Iterator i = super.entrySet().iterator();
+        while ( i.hasNext() ) {
+            Map.Entry e = (Map.Entry)i.next();
+            MapEntry entry = (MapEntry)e.getValue();
+            if ( entry.isPrimary() ) counter++;
+        }
+        return counter;
     }
     
     protected boolean removeEldestEntry(Map.Entry eldest) {
@@ -185,11 +353,18 @@
     }
     
     public boolean isEmpty() {
-        return super.isEmpty();
+        return size()==0;
     }
     
     public Collection values() {
-        return super.values();
+        ArrayList values = new ArrayList(super.size());
+        Iterator i = super.entrySet().iterator();
+        while ( i.hasNext() ) {
+            Map.Entry e = (Map.Entry)i.next();
+            MapEntry entry = (MapEntry)e.getValue();
+            if ( entry.isPrimary() ) values.add(entry.getValue());
+        }
+        return values;
     }
     
 
@@ -220,6 +395,10 @@
         public boolean isProxy() {
             return proxy;
         }
+        
+        public boolean isPrimary() {
+            return ((!proxy) && (!backup));
+        }
 
         public void setProxy(boolean proxy) {
             this.proxy = proxy;
@@ -237,8 +416,6 @@
             return backupNode;
         }
         
-        
-        
         public Object getValue() {
             return value;
         }
@@ -300,6 +477,131 @@
         }
         
     }
+//------------------------------------------------------------------------------    
+//                map message to send to and from other maps    
+//------------------------------------------------------------------------------
+    
+    public static class MapMessage implements Externalizable {
+        public static final int MSG_BACKUP = 1;
+        public static final int MSG_RETRIEVE_BACKUP = 2;
+        public static final int MSG_PROXY = 3;
+        public static final int MSG_REMOVE = 4;
+        
+        private byte[] mapId;
+        private int msgtype;
+        private boolean diff;
+        private Serializable key;
+        private Serializable value;
+        private byte[] diffvalue;
+        private Member node;
+        
+        public MapMessage(byte[] mapId,
+                          int msgtype, boolean diff, 
+                          Serializable key,Serializable value,
+                          byte[] diffvalue, Member node) {
+            this.mapId = mapId;
+            this.msgtype = msgtype;
+            this.diff = diff;
+            this.key = key;
+            this.value = value;
+            this.diffvalue = diffvalue;
+            this.node = node;
+        }
+        
+        public int getMsgType() {
+            return msgtype;
+        }
+        
+        public boolean isDiff() {
+            return diff;
+        }
+        
+        public Serializable getKey() {
+            return key;
+        }
+        
+        public Serializable getValue() {
+            return value;
+        }
+        
+        public byte[] getDiffValue() {
+            return diffvalue;
+        }
+        
+        public Member getBackupNode() {
+            return node;
+        }
+        
+        public byte[] getMapId() {
+            return mapId;
+        }
+        
+        public void setValue(Serializable value) {
+            this.value = value;
+        }
+        
+        public void readExternal(ObjectInput in) throws IOException,ClassNotFoundException
{
+            mapId = new byte[in.readInt()];
+            in.read(mapId);
+            msgtype = in.readInt();
+            switch (msgtype) {
+                case MSG_BACKUP: {
+                    diff = in.readBoolean();
+                    key = (Serializable)in.readObject();
+                    if ( diff ) {
+                        diffvalue = new byte[in.readInt()];
+                        in.read(diffvalue);
+                    } else {
+                        value = (Serializable)in.readObject();
+                    }//endif
+                    break;
+                }
+                case MSG_RETRIEVE_BACKUP:
+                case MSG_REMOVE : {
+                    key = (Serializable)in.readObject();
+                    break;
+                }
+                case MSG_PROXY: {
+                    key = (Serializable)in.readObject();
+                    byte[] d = new byte[in.readInt()];
+                    in.read(d);
+                    node = McastMember.getMember(d);
+                    break;
+                }
+            }//switch
+        }//readExternal
+
+        public void writeExternal(ObjectOutput out) throws IOException {
+            out.writeInt(mapId.length);
+            out.write(mapId);
+            out.writeInt(msgtype);
+            switch (msgtype) {
+                case MSG_BACKUP: {
+                    out.writeBoolean(diff);
+                    out.writeObject(key);
+                    if ( diff ) {
+                        out.writeInt(diffvalue.length);
+                        out.write(diffvalue);
+                    } else {
+                        out.writeObject(value);
+                    }//endif
+                    break;
+                }
+                case MSG_RETRIEVE_BACKUP:
+                case MSG_REMOVE : {
+                    out.writeObject(key);
+                    break;
+                }
+                case MSG_PROXY: {
+                    out.writeObject(key);
+                    byte[] d = ((McastMember)node).getData(false);
+                    out.writeInt(d.length);
+                    out.write(d);
+                    break;
+                }
+            }//switch
+        }//writeExternal
+    }//MapMessage
 
 //------------------------------------------------------------------------------    
 //                streamable class    

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tipis/RpcChannel.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tipis/RpcChannel.java?rev=384676&r1=384675&r2=384676&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tipis/RpcChannel.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tipis/RpcChannel.java
Thu Mar  9 18:04:49 2006
@@ -74,7 +74,7 @@
     public Response[] send(Member[] destination, 
                            Serializable message,
                            int options, 
-                           long timeout) throws ChannelException, InterruptedException {
+                           long timeout) throws ChannelException {
         
         if ( destination==null || destination.length == 0 ) return new Response[0];
         RpcCollectorKey key = new RpcCollectorKey(UUIDGenerator.randomUUID(false));
@@ -86,6 +86,8 @@
                 channel.send(destination, rmsg);
                 collector.wait(timeout);
             }
+        } catch ( InterruptedException ix ) {
+            throw new ChannelException(ix);
         }finally {
             responseMap.remove(key);
         }

Modified: tomcat/container/tc5.5.x/modules/groupcom/to-do.txt
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/to-do.txt?rev=384676&r1=384675&r2=384676&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/to-do.txt (original)
+++ tomcat/container/tc5.5.x/modules/groupcom/to-do.txt Thu Mar  9 18:04:49 2006
@@ -54,6 +54,12 @@
 
 15. remove DataSenderFactory and DataSender.properties -
     these cause the settings to be hard coded ant not pluggable.
+    
+16. Guaranteed delivery of messages, ie either all get it or none get it.
+    Meaning, that all receivers get it, then wait for a process command.
+
+17. Implement transactions - the ability to start a transaction, send several messages,
+                             and then commit the transaction
 
 Tasks Completed
 ===========================================



---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org


Mime
View raw message