diff options
Diffstat (limited to 'java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/AbstractReplicatedMap.java')
| -rw-r--r-- | java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/AbstractReplicatedMap.java | 1564 |
1 files changed, 0 insertions, 1564 deletions
diff --git a/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/AbstractReplicatedMap.java b/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/AbstractReplicatedMap.java deleted file mode 100644 index 98c6739854..0000000000 --- a/java/sca/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/AbstractReplicatedMap.java +++ /dev/null @@ -1,1564 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.tuscany.sca.endpoint.tribes; - -import java.io.IOException; -import java.io.ObjectInput; -import java.io.ObjectOutput; -import java.io.Serializable; -import java.io.UnsupportedEncodingException; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; -import java.util.HashMap; -import java.util.Iterator; -import java.util.LinkedHashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; - -import org.apache.catalina.tribes.Channel; -import org.apache.catalina.tribes.ChannelException; -import org.apache.catalina.tribes.ChannelListener; -import org.apache.catalina.tribes.Heartbeat; -import org.apache.catalina.tribes.Member; -import org.apache.catalina.tribes.MembershipListener; -import org.apache.catalina.tribes.group.Response; -import org.apache.catalina.tribes.group.RpcCallback; -import org.apache.catalina.tribes.group.RpcChannel; -import org.apache.catalina.tribes.io.XByteBuffer; -import org.apache.catalina.tribes.membership.MemberImpl; -import org.apache.catalina.tribes.tipis.ReplicatedMapEntry; -import org.apache.catalina.tribes.util.Arrays; -import org.apache.juli.logging.Log; -import org.apache.juli.logging.LogFactory; - -/** - * This file is copied from: - * https://svn.apache.org/repos/asf/tomcat/tc6.0.x/tags/TOMCAT_6_0_20/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java - * - * We need to intercept the put/remove calls on the base Map to fire events - */ -public abstract class AbstractReplicatedMap extends MapStore implements RpcCallback, ChannelListener, - MembershipListener, Heartbeat { - private static final long serialVersionUID = -2703358020113899074L; - - protected static Log log = LogFactory.getLog(AbstractReplicatedMap.class); - - /** - * The default initial capacity - MUST be a power of two. - */ - public static final int DEFAULT_INITIAL_CAPACITY = 16; - - /** - * The load factor used when none specified in constructor. - **/ - public static final float DEFAULT_LOAD_FACTOR = 0.75f; - - /** - * Used to identify the map - */ - final String chset = "ISO-8859-1"; - - //------------------------------------------------------------------------------ - // INSTANCE VARIABLES - //------------------------------------------------------------------------------ - protected abstract int getStateMessageType(); - - /** - * Timeout for RPC messages, how long we will wait for a reply - */ - protected transient long rpcTimeout = 5000; - /** - * Reference to the channel for sending messages - */ - protected transient Channel channel; - /** - * The RpcChannel to send RPC messages through - */ - protected transient RpcChannel rpcChannel; - /** - * The Map context name makes this map unique, this - * allows us to have more than one map shared - * through one channel - */ - protected transient byte[] mapContextName; - /** - * Has the state been transferred - */ - protected transient boolean stateTransferred = false; - /** - * Simple lock object for transfers - */ - protected transient Object stateMutex = new Object(); - /** - * A list of members in our map - */ - protected transient Map<Member, Long> mapMembers = new HashMap<Member, Long>(); - /** - * Our default send options - */ - protected transient int channelSendOptions = Channel.SEND_OPTIONS_DEFAULT; - /** - * The owner of this map, ala a SessionManager for example - */ - protected transient MapOwner mapOwner; - /** - * External class loaders if serialization and deserialization is to be performed successfully. - */ - protected transient ClassLoader[] externalLoaders; - - /** - * The node we are currently backing up data to, this index will rotate - * on a round robin basis - */ - protected transient int currentNode = 0; - - /** - * Since the map keeps internal membership - * this is the timeout for a ping message to be responded to - * If a remote map doesn't respond within this timeframe, - * its considered dead. - */ - protected transient long accessTimeout = 5000; - - /** - * Readable string of the mapContextName value - */ - protected transient String mapname = ""; - - //------------------------------------------------------------------------------ - // map owner interface - //------------------------------------------------------------------------------ - - public static interface MapOwner { - public void objectMadePrimay(Object key, Object value); - } - - //------------------------------------------------------------------------------ - // CONSTRUCTORS - //------------------------------------------------------------------------------ - - /** - * Creates a new map - * @param channel The channel to use for communication - * @param timeout long - timeout for RPC messags - * @param mapContextName String - unique name for this map, to allow multiple maps per channel - * @param initialCapacity int - the size of this map, see HashMap - * @param loadFactor float - load factor, see HashMap - * @param cls - a list of classloaders to be used for deserialization of objects. - */ - public AbstractReplicatedMap(MapOwner owner, - Channel channel, - long timeout, - String mapContextName, - int initialCapacity, - float loadFactor, - int channelSendOptions, - ClassLoader[] cls) { - super(initialCapacity, loadFactor, 15); - init(owner, channel, mapContextName, timeout, channelSendOptions, cls); - - } - - /** - * Helper methods, wraps a single member in an array - * @param m Member - * @return Member[] - */ - protected Member[] wrap(Member m) { - if (m == null) - return new Member[0]; - else - return new Member[] {m}; - } - - /** - * Initializes the map by creating the RPC channel, registering itself as a channel listener - * This method is also responsible for initiating the state transfer - * @param owner Object - * @param channel Channel - * @param mapContextName String - * @param timeout long - * @param channelSendOptions int - * @param cls ClassLoader[] - */ - protected void init(MapOwner owner, - Channel channel, - String mapContextName, - long timeout, - int channelSendOptions, - ClassLoader[] cls) { - log.info("Initializing AbstractReplicatedMap with context name:" + mapContextName); - this.mapOwner = owner; - this.externalLoaders = cls; - this.channelSendOptions = channelSendOptions; - this.channel = channel; - this.rpcTimeout = timeout; - - try { - this.mapname = mapContextName; - //unique context is more efficient if it is stored as bytes - this.mapContextName = mapContextName.getBytes(chset); - } catch (UnsupportedEncodingException x) { - log.warn("Unable to encode mapContextName[" + mapContextName - + "] using getBytes(" - + chset - + ") using default getBytes()", x); - this.mapContextName = mapContextName.getBytes(); - } - if (log.isTraceEnabled()) - log.trace("Created Lazy Map with name:" + mapContextName - + ", bytes:" - + Arrays.toString(this.mapContextName)); - - //create an rpc channel and add the map as a listener - this.rpcChannel = new RpcChannel(this.mapContextName, channel, this); - //add this map as a message listener - this.channel.addChannelListener(this); - //listen for membership notifications - this.channel.addMembershipListener(this); - - try { - //broadcast our map, this just notifies other members of our existence - broadcast(MapMessage.MSG_INIT, true); - //transfer state from another map - transferState(); - //state is transferred, we are ready for messaging - broadcast(MapMessage.MSG_START, true); - } catch (ChannelException x) { - log.warn("Unable to send map start message."); - throw new RuntimeException("Unable to start replicated map.", x); - } - } - - /** - * Sends a ping out to all the members in the cluster, not just map members - * that this map is alive. - * @param timeout long - * @throws ChannelException - */ - protected void ping(long timeout) throws ChannelException { - //send out a map membership message, only wait for the first reply - MapMessage msg = - new MapMessage(this.mapContextName, MapMessage.MSG_INIT, false, null, null, null, channel - .getLocalMember(false), null); - if (channel.getMembers().length > 0) { - //send a ping, wait for all nodes to reply - Response[] resp = - rpcChannel.send(channel.getMembers(), - msg, - rpcChannel.ALL_REPLY, - (channelSendOptions), - (int)accessTimeout); - for (int i = 0; i < resp.length; i++) { - memberAlive(resp[i].getSource()); - } //for - } - //update our map of members, expire some if we didn't receive a ping back - synchronized (mapMembers) { - Iterator it = mapMembers.entrySet().iterator(); - long now = System.currentTimeMillis(); - while (it.hasNext()) { - Map.Entry entry = (Map.Entry)it.next(); - long access = ((Long)entry.getValue()).longValue(); - if ((now - access) > timeout) { - it.remove(); - memberDisappeared((Member)entry.getKey()); - } - } - }//synch - } - - /** - * We have received a member alive notification - * @param member Member - */ - protected void memberAlive(Member member) { - synchronized (mapMembers) { - if (!mapMembers.containsKey(member)) { - mapMemberAdded(member); - } //end if - mapMembers.put(member, new Long(System.currentTimeMillis())); - } - } - - /** - * Helper method to broadcast a message to all members in a channel - * @param msgtype int - * @param rpc boolean - * @throws ChannelException - */ - protected void broadcast(int msgtype, boolean rpc) throws ChannelException { - //send out a map membership message, only wait for the first reply - MapMessage msg = - new MapMessage(this.mapContextName, msgtype, false, null, null, null, channel.getLocalMember(false), null); - if (rpc) { - Response[] resp = - rpcChannel.send(channel.getMembers(), msg, rpcChannel.FIRST_REPLY, (channelSendOptions), rpcTimeout); - for (int i = 0; i < resp.length; i++) { - mapMemberAdded(resp[i].getSource()); - messageReceived(resp[i].getMessage(), resp[i].getSource()); - } - } else { - channel.send(channel.getMembers(), msg, channelSendOptions); - } - } - - public void breakdown() { - finalize(); - } - - public void finalize() { - try { - broadcast(MapMessage.MSG_STOP, false); - } catch (Exception ignore) { - } - //cleanup - if (this.rpcChannel != null) { - this.rpcChannel.breakdown(); - } - if (this.channel != null) { - this.channel.removeChannelListener(this); - this.channel.removeMembershipListener(this); - } - this.rpcChannel = null; - this.channel = null; - this.mapMembers.clear(); - super.clear(); - this.stateTransferred = false; - this.externalLoaders = null; - } - - public int hashCode() { - return Arrays.hashCode(this.mapContextName); - } - - public boolean equals(Object o) { - if (o == null) - return false; - if (!(o instanceof AbstractReplicatedMap)) - return false; - if (!(o.getClass().equals(this.getClass()))) - return false; - AbstractReplicatedMap other = (AbstractReplicatedMap)o; - return Arrays.equals(mapContextName, other.mapContextName); - } - - //------------------------------------------------------------------------------ - // GROUP COM INTERFACES - //------------------------------------------------------------------------------ - public Member[] getMapMembers(Map<Member, Long> members) { - synchronized (members) { - Member[] result = new Member[members.size()]; - members.keySet().toArray(result); - return result; - } - } - - public Member[] getMapMembers() { - return getMapMembers(this.mapMembers); - } - - public Member[] getMapMembersExcl(Member[] exclude) { - synchronized (mapMembers) { - Map<Member, Long> list = (Map<Member, Long>)((HashMap<Member, Long>)mapMembers).clone(); - for (int i = 0; i < exclude.length; i++) - list.remove(exclude[i]); - return getMapMembers(list); - } - } - - /** - * Replicates any changes to the object since the last time - * The object has to be primary, ie, if the object is a proxy or a backup, it will not be replicated<br> - * @param complete - if set to true, the object is replicated to its backup - * if set to false, only objects that implement ReplicatedMapEntry and the isDirty() returns true will - * be replicated - */ - public void replicate(Object key, boolean complete) { - if (log.isTraceEnabled()) - log.trace("Replicate invoked on key:" + key); - MapEntry entry = (MapEntry)super.get(key); - if (entry == null) - return; - if (!entry.isSerializable()) - return; - if (entry != null && entry.isPrimary() && entry.getBackupNodes() != null && entry.getBackupNodes().length > 0) { - Object value = entry.getValue(); - //check to see if we need to replicate this object isDirty()||complete - boolean repl = complete || ((value instanceof ReplicatedMapEntry) && ((ReplicatedMapEntry)value).isDirty()); - - if (!repl) { - if (log.isTraceEnabled()) - log.trace("Not replicating:" + key + ", no change made"); - - return; - } - //check to see if the message is diffable - boolean diff = ((value instanceof ReplicatedMapEntry) && ((ReplicatedMapEntry)value).isDiffable()); - MapMessage msg = null; - if (diff) { - ReplicatedMapEntry rentry = (ReplicatedMapEntry)entry.getValue(); - try { - rentry.lock(); - //construct a diff message - msg = - new MapMessage(mapContextName, MapMessage.MSG_BACKUP, true, (Serializable)entry.getKey(), null, - rentry.getDiff(), entry.getPrimary(), entry.getBackupNodes()); - } catch (IOException x) { - log.error("Unable to diff object. Will replicate the entire object instead.", x); - } finally { - rentry.unlock(); - } - - } - if (msg == null) { - //construct a complete - msg = - new MapMessage(mapContextName, MapMessage.MSG_BACKUP, false, (Serializable)entry.getKey(), - (Serializable)entry.getValue(), null, entry.getPrimary(), entry.getBackupNodes()); - - } - try { - if (channel != null && entry.getBackupNodes() != null && entry.getBackupNodes().length > 0) { - channel.send(entry.getBackupNodes(), msg, channelSendOptions); - } - } catch (ChannelException x) { - log.error("Unable to replicate data.", x); - } - } //end if - - } - - /** - * This can be invoked by a periodic thread to replicate out any changes. - * For maps that don't store objects that implement ReplicatedMapEntry, this - * method should be used infrequently to avoid large amounts of data transfer - * @param complete boolean - */ - public void replicate(boolean complete) { - Iterator<Map.Entry<Object, Object>> i = super.entrySet().iterator(); - while (i.hasNext()) { - Map.Entry<Object, Object> e = i.next(); - replicate(e.getKey(), complete); - } //while - - } - - public void transferState() { - try { - Member[] members = getMapMembers(); - Member backup = members.length > 0 ? (Member)members[0] : null; - if (backup != null) { - MapMessage msg = - new MapMessage(mapContextName, getStateMessageType(), false, null, null, null, null, null); - Response[] resp = - rpcChannel.send(new Member[] {backup}, msg, RpcChannel.FIRST_REPLY, channelSendOptions, rpcTimeout); - if (resp.length > 0) { - synchronized (stateMutex) { - msg = (MapMessage)resp[0].getMessage(); - msg.deserialize(getExternalLoaders()); - List list = (List)msg.getValue(); - for (int i = 0; i < list.size(); i++) { - messageReceived((Serializable)list.get(i), resp[0].getSource()); - } //for - } - } else { - log.warn("Transfer state, 0 replies, probably a timeout."); - } - } - } catch (ChannelException x) { - log.error("Unable to transfer LazyReplicatedMap state.", x); - } catch (IOException x) { - log.error("Unable to transfer LazyReplicatedMap state.", x); - } catch (ClassNotFoundException x) { - log.error("Unable to transfer LazyReplicatedMap state.", x); - } - stateTransferred = true; - } - - /** - * @todo implement state transfer - * @param msg Serializable - * @return Serializable - null if no reply should be sent - */ - public Serializable replyRequest(Serializable msg, final Member sender) { - if (!(msg instanceof MapMessage)) - return null; - MapMessage mapmsg = (MapMessage)msg; - - //map init request - if (mapmsg.getMsgType() == MapMessage.MSG_INIT) { - mapmsg.setPrimary(channel.getLocalMember(false)); - return mapmsg; - } - - //map start request - if (mapmsg.getMsgType() == MapMessage.MSG_START) { - mapmsg.setPrimary(channel.getLocalMember(false)); - mapMemberAdded(sender); - return mapmsg; - } - - //backup request - if (mapmsg.getMsgType() == MapMessage.MSG_RETRIEVE_BACKUP) { - MapEntry entry = (MapEntry)super.get(mapmsg.getKey()); - if (entry == null || (!entry.isSerializable())) - return null; - mapmsg.setValue((Serializable)entry.getValue()); - return mapmsg; - } - - //state transfer request - if (mapmsg.getMsgType() == MapMessage.MSG_STATE || mapmsg.getMsgType() == MapMessage.MSG_STATE_COPY) { - synchronized (stateMutex) { //make sure we dont do two things at the same time - ArrayList<MapMessage> list = new ArrayList<MapMessage>(); - Iterator<Map.Entry<Object, Object>> i = super.entrySet().iterator(); - while (i.hasNext()) { - Map.Entry<Object, Object> e = i.next(); - MapEntry entry = (MapEntry)super.get(e.getKey()); - if (entry != null && entry.isSerializable()) { - boolean copy = (mapmsg.getMsgType() == MapMessage.MSG_STATE_COPY); - MapMessage me = - new MapMessage(mapContextName, copy ? MapMessage.MSG_COPY : MapMessage.MSG_PROXY, false, - (Serializable)entry.getKey(), copy ? (Serializable)entry.getValue() : null, - null, entry.getPrimary(), entry.getBackupNodes()); - list.add(me); - } - } - mapmsg.setValue(list); - return mapmsg; - - } //synchronized - } - - return null; - - } - - /** - * If the reply has already been sent to the requesting thread, - * the rpc callback can handle any data that comes in after the fact. - * @param msg Serializable - * @param sender Member - */ - public void leftOver(Serializable msg, Member sender) { - //left over membership messages - if (!(msg instanceof MapMessage)) - return; - - MapMessage mapmsg = (MapMessage)msg; - try { - mapmsg.deserialize(getExternalLoaders()); - if (mapmsg.getMsgType() == MapMessage.MSG_START) { - mapMemberAdded(mapmsg.getPrimary()); - } else if (mapmsg.getMsgType() == MapMessage.MSG_INIT) { - memberAlive(mapmsg.getPrimary()); - } - } catch (IOException x) { - log.error("Unable to deserialize MapMessage.", x); - } catch (ClassNotFoundException x) { - log.error("Unable to deserialize MapMessage.", x); - } - } - - public void messageReceived(Serializable msg, Member sender) { - if (!(msg instanceof MapMessage)) - return; - - MapMessage mapmsg = (MapMessage)msg; - if (log.isTraceEnabled()) { - log.trace("Map[" + mapname + "] received message:" + mapmsg); - } - - try { - mapmsg.deserialize(getExternalLoaders()); - } catch (IOException x) { - log.error("Unable to deserialize MapMessage.", x); - return; - } catch (ClassNotFoundException x) { - log.error("Unable to deserialize MapMessage.", x); - return; - } - if (log.isTraceEnabled()) - log.trace("Map message received from:" + sender.getName() + " msg:" + mapmsg); - if (mapmsg.getMsgType() == MapMessage.MSG_START) { - mapMemberAdded(mapmsg.getPrimary()); - } - - if (mapmsg.getMsgType() == MapMessage.MSG_STOP) { - memberDisappeared(mapmsg.getPrimary()); - } - - if (mapmsg.getMsgType() == MapMessage.MSG_PROXY) { - MapEntry entry = (MapEntry)super.get(mapmsg.getKey()); - if (entry == null) { - entry = new MapEntry(mapmsg.getKey(), mapmsg.getValue()); - entry.setBackup(false); - entry.setProxy(true); - entry.setBackupNodes(mapmsg.getBackupNodes()); - entry.setPrimary(mapmsg.getPrimary()); - super.put(entry.getKey(), entry); - } else { - entry.setProxy(true); - entry.setBackup(false); - entry.setBackupNodes(mapmsg.getBackupNodes()); - entry.setPrimary(mapmsg.getPrimary()); - } - } - - if (mapmsg.getMsgType() == MapMessage.MSG_REMOVE) { - super.remove(mapmsg.getKey()); - } - - if (mapmsg.getMsgType() == MapMessage.MSG_BACKUP || mapmsg.getMsgType() == MapMessage.MSG_COPY) { - MapEntry entry = (MapEntry)super.get(mapmsg.getKey()); - if (entry == null) { - entry = new MapEntry(mapmsg.getKey(), mapmsg.getValue()); - entry.setBackup(mapmsg.getMsgType() == MapMessage.MSG_BACKUP); - entry.setProxy(false); - entry.setBackupNodes(mapmsg.getBackupNodes()); - entry.setPrimary(mapmsg.getPrimary()); - if (mapmsg.getValue() != null && mapmsg.getValue() instanceof ReplicatedMapEntry) { - ((ReplicatedMapEntry)mapmsg.getValue()).setOwner(getMapOwner()); - } - } else { - entry.setBackup(mapmsg.getMsgType() == MapMessage.MSG_BACKUP); - entry.setProxy(false); - entry.setBackupNodes(mapmsg.getBackupNodes()); - entry.setPrimary(mapmsg.getPrimary()); - if (entry.getValue() instanceof ReplicatedMapEntry) { - ReplicatedMapEntry diff = (ReplicatedMapEntry)entry.getValue(); - if (mapmsg.isDiff()) { - try { - diff.lock(); - diff.applyDiff(mapmsg.getDiffValue(), 0, mapmsg.getDiffValue().length); - } catch (Exception x) { - log.error("Unable to apply diff to key:" + entry.getKey(), x); - } finally { - diff.unlock(); - } - } else { - if (mapmsg.getValue() != null) - entry.setValue(mapmsg.getValue()); - ((ReplicatedMapEntry)entry.getValue()).setOwner(getMapOwner()); - } //end if - } else if (mapmsg.getValue() instanceof ReplicatedMapEntry) { - ReplicatedMapEntry re = (ReplicatedMapEntry)mapmsg.getValue(); - re.setOwner(getMapOwner()); - entry.setValue(re); - } else { - if (mapmsg.getValue() != null) - entry.setValue(mapmsg.getValue()); - } //end if - } //end if - super.put(entry.getKey(), entry); - } //end if - } - - public boolean accept(Serializable msg, Member sender) { - boolean result = false; - if (msg instanceof MapMessage) { - if (log.isTraceEnabled()) - log.trace("Map[" + mapname + "] accepting...." + msg); - result = Arrays.equals(mapContextName, ((MapMessage)msg).getMapId()); - if (log.isTraceEnabled()) - log.trace("Msg[" + mapname + "] accepted[" + result + "]...." + msg); - } - return result; - } - - public void mapMemberAdded(Member member) { - Member self = getChannel().getLocalMember(false); - if (member.equals(self)) - return; - boolean memberAdded = false; - //select a backup node if we don't have one - synchronized (mapMembers) { - if (!mapMembers.containsKey(member)) { - mapMembers.put(member, new Long(System.currentTimeMillis())); - memberAdded = true; - } - } - if (memberAdded) { - synchronized (stateMutex) { - Iterator<Map.Entry<Object, Object>> i = super.entrySet().iterator(); - while (i.hasNext()) { - Map.Entry<Object, Object> e = i.next(); - MapEntry entry = (MapEntry)super.get(e.getKey()); - if (entry == null) - continue; - // if (entry.isPrimary() && (entry.getBackupNodes() == null || entry.getBackupNodes().length == 0)) { - // [rfeng] Change the behavior to replicate to all nodes - if (entry.isPrimary() && self.equals(entry.getPrimary())) { - try { - Member[] backup = publishEntryInfo(entry.getKey(), entry.getValue(), entry.getBackupNodes()); - entry.setBackupNodes(backup); - entry.setPrimary(self); - } catch (ChannelException x) { - log.error("Unable to select backup node.", x); - } //catch - } //end if - } //while - } //synchronized - }//end if - } - - public boolean inSet(Member m, Member[] set) { - if (set == null) - return false; - boolean result = false; - for (int i = 0; i < set.length && (!result); i++) - if (m.equals(set[i])) - result = true; - return result; - } - - public Member[] excludeFromSet(Member[] mbrs, Member[] set) { - List<Member> result = new ArrayList<Member>(); - for (int i = 0; i < set.length; i++) { - boolean include = true; - for (int j = 0; j < mbrs.length; j++) - if (mbrs[j].equals(set[i])) - include = false; - if (include) - result.add(set[i]); - } - return (Member[])result.toArray(new Member[result.size()]); - } - - public void memberAdded(Member member) { - //do nothing - } - - public void memberDisappeared(Member member) { - boolean removed = false; - synchronized (mapMembers) { - removed = (mapMembers.remove(member) != null); - if (!removed) { - if (log.isDebugEnabled()) - log.debug("Member[" + member + "] disappeared, but was not present in the map."); - return; //the member was not part of our map. - } - } - - Iterator<Map.Entry<Object, Object>> i = super.entrySet().iterator(); - while (i.hasNext()) { - Map.Entry<Object, Object> e = i.next(); - MapEntry entry = (MapEntry)super.get(e.getKey()); - if (entry == null) - continue; - if (entry.isPrimary() && inSet(member, entry.getBackupNodes())) { - if (log.isDebugEnabled()) - log.debug("[1] Primary choosing a new backup"); - try { - Member[] backup = publishEntryInfo(entry.getKey(), entry.getValue(), entry.getBackupNodes()); - entry.setBackupNodes(backup); - entry.setPrimary(channel.getLocalMember(false)); - } catch (ChannelException x) { - log.error("Unable to relocate[" + entry.getKey() + "] to a new backup node", x); - } - } else if (member.equals(entry.getPrimary())) { - if (log.isDebugEnabled()) - log.debug("[2] Primary disappeared"); - entry.setPrimary(null); - } //end if - - if (entry.isProxy() && entry.getPrimary() == null - && entry.getBackupNodes() != null - && entry.getBackupNodes().length == 1 - && entry.getBackupNodes()[0].equals(member)) { - //remove proxies that have no backup nor primaries - if (log.isDebugEnabled()) - log.debug("[3] Removing orphaned proxy"); - i.remove(); - } else if (entry.getPrimary() == null && entry.isBackup() - && entry.getBackupNodes() != null - && entry.getBackupNodes().length == 1 - && entry.getBackupNodes()[0].equals(channel.getLocalMember(false))) { - try { - if (log.isDebugEnabled()) - log.debug("[4] Backup becoming primary"); - entry.setPrimary(channel.getLocalMember(false)); - entry.setBackup(false); - entry.setProxy(false); - Member[] backup = publishEntryInfo(entry.getKey(), entry.getValue(), entry.getBackupNodes()); - entry.setBackupNodes(backup); - if (mapOwner != null) - mapOwner.objectMadePrimay(entry.getKey(), entry.getValue()); - - } catch (ChannelException x) { - log.error("Unable to relocate[" + entry.getKey() + "] to a new backup node", x); - } - } - - } //while - } - - public int getNextBackupIndex() { - int size = mapMembers.size(); - if (mapMembers.size() == 0) - return -1; - int node = currentNode++; - if (node >= size) { - node = 0; - currentNode = 0; - } - return node; - } - - public Member getNextBackupNode() { - Member[] members = getMapMembers(); - int node = getNextBackupIndex(); - if (members.length == 0 || node == -1) - return null; - if (node >= members.length) - node = 0; - return members[node]; - } - - protected abstract Member[] publishEntryInfo(Object key, Object value, Member[] backupNodes) throws ChannelException; - - public void heartbeat() { - try { - ping(accessTimeout); - } catch (Exception x) { - log.error("Unable to send AbstractReplicatedMap.ping message", x); - } - } - - //------------------------------------------------------------------------------ - // METHODS TO OVERRIDE - //------------------------------------------------------------------------------ - - /** - * Removes an object from this map, it will also remove it from - * - * @param key Object - * @return Object - */ - public Object remove(Object key) { - return remove(key, true); - } - - public Object remove(Object key, boolean notify) { - MapEntry entry = (MapEntry)super.remove(key); - - try { - if (getMapMembers().length > 0 && notify) { - MapMessage msg = - new MapMessage(getMapContextName(), MapMessage.MSG_REMOVE, false, (Serializable)key, null, null, - null, null); - getChannel().send(getMapMembers(), msg, getChannelSendOptions()); - } - } catch (ChannelException x) { - log.error("Unable to replicate out data for a LazyReplicatedMap.remove operation", x); - } - return entry != null ? entry.getValue() : null; - } - - public MapEntry getInternal(Object key) { - return (MapEntry)super.get(key); - } - - public Object get(Object key) { - MapEntry entry = (MapEntry)super.get(key); - if (log.isTraceEnabled()) - log.trace("Requesting id:" + key + " entry:" + entry); - if (entry == null) - return null; - if (!entry.isPrimary()) { - //if the message is not primary, we need to retrieve the latest value - try { - Member[] backup = null; - MapMessage msg = null; - if (!entry.isBackup()) { - //make sure we don't retrieve from ourselves - msg = - new MapMessage(getMapContextName(), MapMessage.MSG_RETRIEVE_BACKUP, false, (Serializable)key, - null, null, null, null); - Response[] resp = - getRpcChannel().send(entry.getBackupNodes(), - msg, - RpcChannel.FIRST_REPLY, - Channel.SEND_OPTIONS_DEFAULT, - getRpcTimeout()); - if (resp == null || resp.length == 0) { - //no responses - log.warn("Unable to retrieve remote object for key:" + key); - return null; - } - msg = (MapMessage)resp[0].getMessage(); - msg.deserialize(getExternalLoaders()); - backup = entry.getBackupNodes(); - if (entry.getValue() instanceof ReplicatedMapEntry) { - ReplicatedMapEntry val = (ReplicatedMapEntry)entry.getValue(); - val.setOwner(getMapOwner()); - } - if (msg.getValue() != null) - entry.setValue(msg.getValue()); - } - if (entry.isBackup()) { - //select a new backup node - backup = publishEntryInfo(key, entry.getValue(), entry.getBackupNodes()); - } else if (entry.isProxy()) { - //invalidate the previous primary - msg = - new MapMessage(getMapContextName(), MapMessage.MSG_PROXY, false, (Serializable)key, null, null, - channel.getLocalMember(false), backup); - Member[] dest = getMapMembersExcl(backup); - if (dest != null && dest.length > 0) { - getChannel().send(dest, msg, getChannelSendOptions()); - } - } - entry.setPrimary(channel.getLocalMember(false)); - entry.setBackupNodes(backup); - entry.setBackup(false); - entry.setProxy(false); - - } catch (Exception x) { - log.error("Unable to replicate out data for a LazyReplicatedMap.get operation", x); - return null; - } - } - if (log.isTraceEnabled()) - log.trace("Requesting id:" + key + " result:" + entry.getValue()); - if (entry.getValue() != null && entry.getValue() instanceof ReplicatedMapEntry) { - ReplicatedMapEntry val = (ReplicatedMapEntry)entry.getValue(); - //hack, somehow this is not being set above - val.setOwner(getMapOwner()); - - } - return entry.getValue(); - } - - protected void printMap(String header) { - try { - System.out.println("\nDEBUG MAP:" + header); - System.out.println("Map[" + new String(mapContextName, chset) + ", Map Size:" + super.size()); - Member[] mbrs = getMapMembers(); - for (int i = 0; i < mbrs.length; i++) { - System.out.println("Mbr[" + (i + 1) + "=" + mbrs[i].getName()); - } - Iterator i = super.entrySet().iterator(); - int cnt = 0; - - while (i.hasNext()) { - Map.Entry e = (Map.Entry)i.next(); - System.out.println((++cnt) + ". " + super.get(e.getKey())); - } - System.out.println("EndMap]\n\n"); - } catch (Exception ignore) { - ignore.printStackTrace(); - } - } - - /** - * Returns true if the key has an entry in the map. - * The entry can be a proxy or a backup entry, invoking <code>get(key)</code> - * will make this entry primary for the group - * @param key Object - * @return boolean - */ - public boolean containsKey(Object key) { - return super.containsKey(key); - } - - public Object put(Object key, Object value) { - return put(key, value, true); - } - - public Object put(Object key, Object value, boolean notify) { - MapEntry entry = new MapEntry(key, value); - entry.setBackup(false); - entry.setProxy(false); - entry.setPrimary(channel.getLocalMember(false)); - - Object old = null; - - //make sure that any old values get removed - if (containsKey(key)) - old = remove(key); - try { - if (notify) { - Member[] backup = publishEntryInfo(key, value, entry.getBackupNodes()); - entry.setBackupNodes(backup); - } - } catch (ChannelException x) { - log.error("Unable to replicate out data for a LazyReplicatedMap.put operation", x); - } - super.put(key, entry); - return old; - } - - /** - * Copies all values from one map to this instance - * @param m Map - */ - public void putAll(Map m) { - Iterator i = m.entrySet().iterator(); - while (i.hasNext()) { - Map.Entry entry = (Map.Entry)i.next(); - put(entry.getKey(), entry.getValue()); - } - } - - public void clear() { - clear(true); - } - - public void clear(boolean notify) { - if (notify) { - //only delete active keys - Iterator<Object> keys = keySet().iterator(); - while (keys.hasNext()) - remove(keys.next()); - } else { - super.clear(); - } - } - - public boolean containsValue(Object value) { - if (value == null) { - return super.containsValue(value); - } else { - Iterator<Map.Entry<Object, Object>> i = super.entrySet().iterator(); - while (i.hasNext()) { - Map.Entry<Object, Object> e = i.next(); - MapEntry entry = (MapEntry)super.get(e.getKey()); - if (entry != null && entry.isPrimary() && value.equals(entry.getValue())) - return true; - }//while - return false; - }//end if - } - - public Object clone() { - throw new UnsupportedOperationException("This operation is not valid on a replicated map"); - } - - /** - * Returns the entire contents of the map - * Map.Entry.getValue() will return a LazyReplicatedMap.MapEntry object containing all the information - * about the object. - * @return Set - */ - public Set<Map.Entry<Object, Object>> entrySetFull() { - return super.entrySet(); - } - - public Set<Object> keySetFull() { - return super.keySet(); - } - - public int sizeFull() { - return super.size(); - } - - public Set<Map.Entry<Object, Object>> entrySet() { - Set<Map.Entry<Object, Object>> set = new LinkedHashSet<Map.Entry<Object, Object>>(super.size()); - Iterator<Map.Entry<Object, Object>> i = super.entrySet().iterator(); - while (i.hasNext()) { - Map.Entry<Object, Object> e = i.next(); - Object key = e.getKey(); - MapEntry entry = (MapEntry)super.get(key); - if (entry != null && entry.isPrimary()) { - set.add(new MapEntry(key, entry.getValue())); - } - } - return Collections.unmodifiableSet(set); - } - - public Set<Object> keySet() { - //todo implement - //should only return keys where this is active. - LinkedHashSet<Object> set = new LinkedHashSet<Object>(super.size()); - Iterator<Map.Entry<Object, Object>> i = super.entrySet().iterator(); - while (i.hasNext()) { - Map.Entry<Object, Object> e = i.next(); - Object key = e.getKey(); - MapEntry entry = (MapEntry)super.get(key); - if (entry != null && entry.isPrimary()) - set.add(key); - } - return Collections.unmodifiableSet(set); - - } - - public int size() { - //todo, implement a counter variable instead - //only count active members in this node - int counter = 0; - Iterator<Map.Entry<Object, Object>> it = super.entrySet().iterator(); - while (it != null && it.hasNext()) { - Map.Entry<Object, Object> e = it.next(); - if (e != null) { - MapEntry entry = (MapEntry)super.get(e.getKey()); - if (entry != null && entry.isPrimary() && entry.getValue() != null) - counter++; - } - } - return counter; - } - - protected boolean removeEldestEntry(Map.Entry<Object, Object> eldest) { - return false; - } - - public boolean isEmpty() { - return size() == 0; - } - - public Collection<Object> values() { - ArrayList<Object> values = new ArrayList<Object>(); - Iterator<Map.Entry<Object, Object>> i = super.entrySet().iterator(); - while (i.hasNext()) { - Map.Entry<Object, Object> e = i.next(); - MapEntry entry = (MapEntry)super.get(e.getKey()); - if (entry != null && entry.isPrimary() && entry.getValue() != null) - values.add(entry.getValue()); - } - return Collections.unmodifiableCollection(values); - } - - //------------------------------------------------------------------------------ - // Map Entry class - //------------------------------------------------------------------------------ - public static class MapEntry implements Map.Entry<Object, Object> { - private boolean backup; - private boolean proxy; - private Member[] backupNodes; - private Member primary; - private Object key; - private Object value; - - public MapEntry(Object key, Object value) { - setKey(key); - setValue(value); - - } - - public boolean isKeySerializable() { - return (key == null) || (key instanceof Serializable); - } - - public boolean isValueSerializable() { - return (value == null) || (value instanceof Serializable); - } - - public boolean isSerializable() { - return isKeySerializable() && isValueSerializable(); - } - - public boolean isBackup() { - return backup; - } - - public void setBackup(boolean backup) { - this.backup = backup; - } - - public boolean isProxy() { - return proxy; - } - - public boolean isPrimary() { - return ((!proxy) && (!backup)); - } - - public void setProxy(boolean proxy) { - this.proxy = proxy; - } - - public boolean isDiffable() { - return (value instanceof ReplicatedMapEntry) && ((ReplicatedMapEntry)value).isDiffable(); - } - - public void setBackupNodes(Member[] nodes) { - this.backupNodes = nodes; - } - - public Member[] getBackupNodes() { - return backupNodes; - } - - public void setPrimary(Member m) { - primary = m; - } - - public Member getPrimary() { - return primary; - } - - public Object getValue() { - return value; - } - - public Object setValue(Object value) { - Object old = this.value; - this.value = value; - return old; - } - - public Object getKey() { - return key; - } - - public Object setKey(Object key) { - Object old = this.key; - this.key = key; - return old; - } - - public int hashCode() { - return key.hashCode(); - } - - public boolean equals(Object o) { - return key.equals(o); - } - - /** - * apply a diff, or an entire object - * @param data byte[] - * @param offset int - * @param length int - * @param diff boolean - * @throws IOException - * @throws ClassNotFoundException - */ - public void apply(byte[] data, int offset, int length, boolean diff) throws IOException, ClassNotFoundException { - if (isDiffable() && diff) { - ReplicatedMapEntry rentry = (ReplicatedMapEntry)value; - try { - rentry.lock(); - rentry.applyDiff(data, offset, length); - } finally { - rentry.unlock(); - } - } else if (length == 0) { - value = null; - proxy = true; - } else { - value = XByteBuffer.deserialize(data, offset, length); - } - } - - public String toString() { - StringBuffer buf = new StringBuffer("MapEntry[key:"); - buf.append(getKey()).append("; "); - buf.append("value:").append(getValue()).append("; "); - buf.append("primary:").append(isPrimary()).append("; "); - buf.append("backup:").append(isBackup()).append("; "); - buf.append("proxy:").append(isProxy()).append(";]"); - return buf.toString(); - } - - } - - //------------------------------------------------------------------------------ - // map message to send to and from other maps - //------------------------------------------------------------------------------ - - public static class MapMessage implements Serializable { - private static final long serialVersionUID = -7847288807489375686L; - 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; - public static final int MSG_STATE = 5; - public static final int MSG_START = 6; - public static final int MSG_STOP = 7; - public static final int MSG_INIT = 8; - public static final int MSG_COPY = 9; - public static final int MSG_STATE_COPY = 10; - - private byte[] mapId; - private int msgtype; - private boolean diff; - private transient Serializable key; - private transient Serializable value; - private byte[] valuedata; - private byte[] keydata; - private byte[] diffvalue; - private Member[] nodes; - private Member primary; - - public String toString() { - StringBuffer buf = new StringBuffer("MapMessage[context="); - buf.append(new String(mapId)); - buf.append("; type="); - buf.append(getTypeDesc()); - buf.append("; key="); - buf.append(key); - buf.append("; value="); - buf.append(value); - return buf.toString(); - } - - public String getTypeDesc() { - switch (msgtype) { - case MSG_BACKUP: - return "MSG_BACKUP"; - case MSG_RETRIEVE_BACKUP: - return "MSG_RETRIEVE_BACKUP"; - case MSG_PROXY: - return "MSG_PROXY"; - case MSG_REMOVE: - return "MSG_REMOVE"; - case MSG_STATE: - return "MSG_STATE"; - case MSG_START: - return "MSG_START"; - case MSG_STOP: - return "MSG_STOP"; - case MSG_INIT: - return "MSG_INIT"; - case MSG_STATE_COPY: - return "MSG_STATE_COPY"; - case MSG_COPY: - return "MSG_COPY"; - default: - return "UNKNOWN"; - } - } - - public MapMessage() { - } - - public MapMessage(byte[] mapId, - int msgtype, - boolean diff, - Serializable key, - Serializable value, - byte[] diffvalue, - Member primary, - Member[] nodes) { - this.mapId = mapId; - this.msgtype = msgtype; - this.diff = diff; - this.key = key; - this.value = value; - this.diffvalue = diffvalue; - this.nodes = nodes; - this.primary = primary; - setValue(value); - setKey(key); - } - - public void deserialize(ClassLoader[] cls) throws IOException, ClassNotFoundException { - key(cls); - value(cls); - } - - public int getMsgType() { - return msgtype; - } - - public boolean isDiff() { - return diff; - } - - public Serializable getKey() { - try { - return key(null); - } catch (Exception x) { - log.error("Deserialization error of the MapMessage.key", x); - return null; - } - } - - public Serializable key(ClassLoader[] cls) throws IOException, ClassNotFoundException { - if (key != null) - return key; - if (keydata == null || keydata.length == 0) - return null; - key = XByteBuffer.deserialize(keydata, 0, keydata.length, cls); - keydata = null; - return key; - } - - public byte[] getKeyData() { - return keydata; - } - - public Serializable getValue() { - try { - return value(null); - } catch (Exception x) { - log.error("Deserialization error of the MapMessage.value", x); - return null; - } - } - - public Serializable value(ClassLoader[] cls) throws IOException, ClassNotFoundException { - if (value != null) - return value; - if (valuedata == null || valuedata.length == 0) - return null; - value = XByteBuffer.deserialize(valuedata, 0, valuedata.length, cls); - valuedata = null; - ; - return value; - } - - public byte[] getValueData() { - return valuedata; - } - - public byte[] getDiffValue() { - return diffvalue; - } - - public Member[] getBackupNodes() { - return nodes; - } - - private void setBackUpNodes(Member[] nodes) { - this.nodes = nodes; - } - - public Member getPrimary() { - return primary; - } - - private void setPrimary(Member m) { - primary = m; - } - - public byte[] getMapId() { - return mapId; - } - - public void setValue(Serializable value) { - try { - if (value != null) - valuedata = XByteBuffer.serialize(value); - this.value = value; - } catch (IOException x) { - throw new RuntimeException(x); - } - } - - public void setKey(Serializable key) { - try { - if (key != null) - keydata = XByteBuffer.serialize(key); - this.key = key; - } catch (IOException x) { - throw new RuntimeException(x); - } - } - - protected Member[] readMembers(ObjectInput in) throws IOException, ClassNotFoundException { - int nodecount = in.readInt(); - Member[] members = new Member[nodecount]; - for (int i = 0; i < members.length; i++) { - byte[] d = new byte[in.readInt()]; - in.read(d); - if (d.length > 0) - members[i] = MemberImpl.getMember(d); - } - return members; - } - - protected void writeMembers(ObjectOutput out, Member[] members) throws IOException { - if (members == null) - members = new Member[0]; - out.writeInt(members.length); - for (int i = 0; i < members.length; i++) { - if (members[i] != null) { - byte[] d = members[i] != null ? ((MemberImpl)members[i]).getData(false) : new byte[0]; - out.writeInt(d.length); - out.write(d); - } - } - } - - /** - * shallow clone - * @return Object - */ - public Object clone() { - MapMessage msg = - new MapMessage(this.mapId, this.msgtype, this.diff, this.key, this.value, this.diffvalue, this.primary, - this.nodes); - msg.keydata = this.keydata; - msg.valuedata = this.valuedata; - return msg; - } - } //MapMessage - - public Channel getChannel() { - return channel; - } - - public byte[] getMapContextName() { - return mapContextName; - } - - public RpcChannel getRpcChannel() { - return rpcChannel; - } - - public long getRpcTimeout() { - return rpcTimeout; - } - - public Object getStateMutex() { - return stateMutex; - } - - public boolean isStateTransferred() { - return stateTransferred; - } - - public MapOwner getMapOwner() { - return mapOwner; - } - - public ClassLoader[] getExternalLoaders() { - return externalLoaders; - } - - public int getChannelSendOptions() { - return channelSendOptions; - } - - public long getAccessTimeout() { - return accessTimeout; - } - - public void setMapOwner(MapOwner mapOwner) { - this.mapOwner = mapOwner; - } - - public void setExternalLoaders(ClassLoader[] externalLoaders) { - this.externalLoaders = externalLoaders; - } - - public void setChannelSendOptions(int channelSendOptions) { - this.channelSendOptions = channelSendOptions; - } - - public void setAccessTimeout(long accessTimeout) { - this.accessTimeout = accessTimeout; - } - -} |