java/org/apache/catalina/tribes/membership/McastServiceImpl.java - tomcat - Git at Google

 /*
  * 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.catalina.tribes.membership;


 import java.io.IOException;
 import java.net.BindException;
 import java.net.DatagramPacket;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.MulticastSocket;
 import java.net.NetworkInterface;
 import java.net.SocketTimeoutException;
 import java.net.StandardSocketOptions;
 import java.util.Arrays;
 import java.util.concurrent.atomic.AtomicBoolean;

 import org.apache.catalina.tribes.Channel;
 import org.apache.catalina.tribes.Member;
 import org.apache.catalina.tribes.MembershipListener;
 import org.apache.catalina.tribes.MessageListener;
 import org.apache.catalina.tribes.io.ChannelData;
 import org.apache.catalina.tribes.io.XByteBuffer;
 import org.apache.catalina.tribes.util.StringManager;
 import org.apache.juli.logging.Log;
 import org.apache.juli.logging.LogFactory;

 /**
  * A <b>membership</b> implementation using simple multicast. This is the representation of a multicast membership
  * service. This class is responsible for maintaining a list of active cluster nodes in the cluster. If a node fails to
  * send out a heartbeat, the node will be dismissed. This is the low level implementation that handles the multicasting
  * sockets. Need to fix this, could use java.nio and only need one thread to send and receive, or just use a timeout on
  * the receive
  */
 public class McastServiceImpl extends MembershipProviderBase {

     private static final Log log = LogFactory.getLog(McastService.class);

     protected static final int MAX_PACKET_SIZE = 65535;

     protected static final StringManager sm = StringManager.getManager(Constants.Package);
     /**
      * Internal flag used for the listen thread that listens to the multicasting socket.
      */
     protected volatile boolean doRunSender = false;
     protected volatile boolean doRunReceiver = false;
     protected volatile int startLevel = 0;
     /**
      * Socket that we intend to listen to
      */
     protected MulticastSocket socket;
     /**
      * The local member that we intend to broadcast over and over again
      */
     protected final MemberImpl member;
     /**
      * The multicast address
      */
     protected final InetAddress address;
     /**
      * The multicast port
      */
     protected final int port;
     /**
      * The time it takes for a member to expire.
      */
     protected final long timeToExpiration;
     /**
      * How often to we send out a broadcast saying we are alive, must be smaller than timeToExpiration
      */
     protected final long sendFrequency;
     /**
      * Reuse the sendPacket, no need to create a new one every time
      */
     protected DatagramPacket sendPacket;
     /**
      * Reuse the receivePacket, no need to create a new one every time
      */
     protected DatagramPacket receivePacket;

     /**
      * The actual listener, for callback when stuff goes down
      */
     protected final MembershipListener service;
     /**
      * The actual listener for broadcast callbacks
      */
     protected final MessageListener msgservice;
     /**
      * Thread to listen for pings
      */
     protected ReceiverThread receiver;
     /**
      * Thread to send pings
      */
     protected SenderThread sender;

     /**
      * Time to live for the multicast packets that are being sent out
      */
     protected final int mcastTTL;
     /**
      * Read timeout on the mcast socket
      */
     protected int mcastSoTimeout;
     /**
      * bind address
      */
     protected final InetAddress mcastBindAddress;

     /**
      * nr of times the system has to fail before a recovery is initiated
      */
     protected int recoveryCounter = 10;

     /**
      * The time the recovery thread sleeps between recovery attempts
      */
     protected long recoverySleepTime = 5000;

     /**
      * Add the ability to turn on/off recovery
      */
     protected boolean recoveryEnabled = true;

     /**
      * disable/enable local loopback message
      */
     protected final boolean localLoopbackDisabled;

     private Channel channel;

     /**
      * Create a new mcast service instance.
      *
      * @param member                - the local member
      * @param sendFrequency         - the time (ms) in between pings sent out
      * @param expireTime            - the time (ms) for a member to expire
      * @param port                  - the mcast port
      * @param bind                  - the bind address (not sure this is used yet)
      * @param mcastAddress          - the mcast address
      * @param ttl                   multicast ttl that will be set on the socket
      * @param soTimeout             Socket timeout
      * @param service               - the callback service
      * @param msgservice            Message listener
      * @param localLoopbackDisabled - disable loopbackMode
      *
      * @throws IOException Init error
      */
     public McastServiceImpl(MemberImpl member, long sendFrequency, long expireTime, int port, InetAddress bind,
             InetAddress mcastAddress, int ttl, int soTimeout, MembershipListener service, MessageListener msgservice,
             boolean localLoopbackDisabled) throws IOException {
         this.member = member;
         this.address = mcastAddress;
         this.port = port;
         this.mcastSoTimeout = soTimeout;
         this.mcastTTL = ttl;
         this.mcastBindAddress = bind;
         this.timeToExpiration = expireTime;
         this.service = service;
         this.msgservice = msgservice;
         this.sendFrequency = sendFrequency;
         this.localLoopbackDisabled = localLoopbackDisabled;
         init();
     }

     public void init() throws IOException {
         setupSocket();
         sendPacket = new DatagramPacket(new byte[MAX_PACKET_SIZE], MAX_PACKET_SIZE);
         sendPacket.setAddress(address);
         sendPacket.setPort(port);
         receivePacket = new DatagramPacket(new byte[MAX_PACKET_SIZE], MAX_PACKET_SIZE);
         receivePacket.setAddress(address);
         receivePacket.setPort(port);
         member.setCommand(new byte[0]);
         if (membership == null) {
             membership = new Membership(member);
         }
     }

     protected void setupSocket() throws IOException {
         if (mcastBindAddress != null) {
             try {
                 log.info(sm.getString("mcastServiceImpl.bind", address, Integer.toString(port)));
                 socket = new MulticastSocket(new InetSocketAddress(address, port));
             } catch (BindException e) {
                 /*
                  * On some platforms (e.g. Linux) it is not possible to bind to the multicast address. In this case only
                  * bind to the port.
                  */
                 if (log.isDebugEnabled()) {
                     log.debug(sm.getString("mcastServiceImpl.bind.failed"), e);
                 } else {
                     log.info(sm.getString("mcastServiceImpl.bind.failed"));
                 }
                 socket = new MulticastSocket(port);
             }
         } else {
             socket = new MulticastSocket(port);
         }
         // Hint if we want to disable loop back(local machine) messages
         socket.setOption(StandardSocketOptions.IP_MULTICAST_LOOP, Boolean.valueOf(!localLoopbackDisabled));
         if (mcastBindAddress != null) {
             if (log.isInfoEnabled()) {
                 log.info(sm.getString("mcastServiceImpl.setInterface", mcastBindAddress));
             }
             NetworkInterface networkInterface = NetworkInterface.getByInetAddress(mcastBindAddress);
             socket.setNetworkInterface(networkInterface);
         } // end if
           // force a so timeout so that we don't block forever
         if (mcastSoTimeout <= 0) {
             mcastSoTimeout = (int) sendFrequency;
         }
         if (log.isInfoEnabled()) {
             log.info(sm.getString("mcastServiceImpl.setSoTimeout", Integer.toString(mcastSoTimeout)));
         }
         socket.setSoTimeout(mcastSoTimeout);

         if (mcastTTL >= 0) {
             if (log.isInfoEnabled()) {
                 log.info(sm.getString("mcastServiceImpl.setTTL", Integer.toString(mcastTTL)));
             }
             socket.setTimeToLive(mcastTTL);
         }
     }


     @Override
     public synchronized void start(int level) throws IOException {
         boolean valid = false;
         if ((level & Channel.MBR_RX_SEQ) == Channel.MBR_RX_SEQ) {
             if (receiver != null) {
                 throw new IllegalStateException(sm.getString("mcastServiceImpl.receive.running"));
             }
             try {
                 if (sender == null) {
                     socket.joinGroup(new InetSocketAddress(address, 0), null);
                 }
             } catch (IOException iox) {
                 log.error(sm.getString("mcastServiceImpl.unable.join"));
                 throw iox;
             }
             doRunReceiver = true;
             receiver = new ReceiverThread();
             receiver.setDaemon(true);
             receiver.start();
             valid = true;
         }
         if ((level & Channel.MBR_TX_SEQ) == Channel.MBR_TX_SEQ) {
             if (sender != null) {
                 throw new IllegalStateException(sm.getString("mcastServiceImpl.send.running"));
             }
             if (receiver == null) {
                 socket.joinGroup(new InetSocketAddress(address, 0), null);
             }
             // make sure at least one packet gets out there
             send(false);
             doRunSender = true;
             sender = new SenderThread(sendFrequency);
             sender.setDaemon(true);
             sender.start();
             // we have started the receiver, but not yet waited for membership to establish
             valid = true;
         }
         if (!valid) {
             throw new IllegalArgumentException(sm.getString("mcastServiceImpl.invalid.startLevel"));
         }
         // pause, once or twice
         waitForMembers(level);
         startLevel = (startLevel | level);
     }

     private void waitForMembers(int level) {
         long memberwait = sendFrequency * 2;
         if (log.isInfoEnabled()) {
             log.info(sm.getString("mcastServiceImpl.waitForMembers.start", Long.toString(memberwait),
                     Integer.toString(level)));
         }
         try {
             Thread.sleep(memberwait);
         } catch (InterruptedException ignore) {
             // Ignore
         }
         if (log.isInfoEnabled()) {
             log.info(sm.getString("mcastServiceImpl.waitForMembers.done", Integer.toString(level)));
         }
     }

     @Override
     public synchronized boolean stop(int level) throws IOException {
         boolean valid = false;

         if ((level & Channel.MBR_RX_SEQ) == Channel.MBR_RX_SEQ) {
             valid = true;
             doRunReceiver = false;
             if (receiver != null) {
                 receiver.interrupt();
             }
             receiver = null;
         }
         if ((level & Channel.MBR_TX_SEQ) == Channel.MBR_TX_SEQ) {
             valid = true;
             doRunSender = false;
             if (sender != null) {
                 sender.interrupt();
             }
             sender = null;
         }

         if (!valid) {
             throw new IllegalArgumentException(sm.getString("mcastServiceImpl.invalid.stopLevel"));
         }
         startLevel = (startLevel & (~level));
         // we're shutting down, send a shutdown message and close the socket
         if (startLevel == 0) {
             // send a stop message
             member.setCommand(Member.SHUTDOWN_PAYLOAD);
             send(false);
             // leave mcast group
             try {
                 socket.leaveGroup(new InetSocketAddress(address, 0), null);
             } catch (Exception e) {
                 // Shutting down. Only log at debug.
                 if (log.isDebugEnabled()) {
                     log.debug(sm.getString("mcastServiceImpl.error.stop"), e);
                 }
             }
             try {
                 socket.close();
             } catch (Exception e) {
                 // Shutting down. Only log at debug.
                 if (log.isDebugEnabled()) {
                     log.debug(sm.getString("mcastServiceImpl.error.stop"), e);
                 }
             }
             member.setServiceStartTime(-1);
         }
         return (startLevel == 0);
     }

     /**
      * Receive a datagram packet, locking wait
      *
      * @throws IOException Received failed
      */
     public void receive() throws IOException {
         try {
             socket.receive(receivePacket);
             if (receivePacket.getLength() > MAX_PACKET_SIZE) {
                 log.error(sm.getString("mcastServiceImpl.packet.tooLong", Integer.toString(receivePacket.getLength())));
             } else {
                 byte[] data = new byte[receivePacket.getLength()];
                 System.arraycopy(receivePacket.getData(), receivePacket.getOffset(), data, 0, data.length);
                 if (XByteBuffer.firstIndexOf(data, 0, MemberImpl.TRIBES_MBR_BEGIN) == 0) {
                     memberDataReceived(data);
                 } else {
                     memberBroadcastsReceived(data);
                 }
             }
         } catch (SocketTimeoutException ignore) {
             /*
              * Do nothing. This is normal. We don't want to block forever since the receive thread is the same thread
              * that does membership expiration.
              */
         }
         checkExpired();
     }

     private void memberDataReceived(byte[] data) {
         final Member m = MemberImpl.getMember(data);
         if (log.isTraceEnabled()) {
             log.trace("Mcast receive ping from member " + m);
         }
         Runnable t = null;
         Thread currentThread = Thread.currentThread();
         if (Arrays.equals(m.getCommand(), Member.SHUTDOWN_PAYLOAD)) {
             if (log.isDebugEnabled()) {
                 log.debug(sm.getString("mcastServiceImpl.memberShutdown", m));
             }
             membership.removeMember(m);
             t = () -> {
                 String name = currentThread.getName();
                 try {
                     currentThread.setName("Membership-MemberDisappeared");
                     service.memberDisappeared(m);
                 } finally {
                     currentThread.setName(name);
                 }
             };
         } else if (membership.memberAlive(m)) {
             if (log.isDebugEnabled()) {
                 log.debug(sm.getString("mcastServiceImpl.memberAdd", m));
             }
             t = () -> {
                 String name = currentThread.getName();
                 try {
                     currentThread.setName("Membership-MemberAdded");
                     service.memberAdded(m);
                 } finally {
                     currentThread.setName(name);
                 }
             };
         }
         if (t != null) {
             executor.execute(t);
         }
     }

     private void memberBroadcastsReceived(final byte[] b) {
         if (log.isTraceEnabled()) {
             log.trace("Mcast received broadcasts.");
         }
         XByteBuffer buffer = new XByteBuffer(b, true);
         if (buffer.countPackages(true) > 0) {
             int count = buffer.countPackages();
             final ChannelData[] data = new ChannelData[count];
             for (int i = 0; i < count; i++) {
                 try {
                     data[i] = buffer.extractPackage(true);
                 } catch (IllegalStateException ise) {
                     log.debug(sm.getString("mcastServiceImpl.messageError"), ise);
                 }
             }
             Runnable t = () -> {
                 Thread currentThread = Thread.currentThread();
                 String name = currentThread.getName();
                 try {
                     currentThread.setName("Membership-MemberAdded");
                     for (ChannelData datum : data) {
                         try {
                             if (datum != null && !member.equals(datum.getAddress())) {
                                 msgservice.messageReceived(datum);
                             }
                         } catch (Throwable t1) {
                             if (t1 instanceof VirtualMachineError) {
                                 throw (VirtualMachineError) t1;
                             }
                             log.error(sm.getString("mcastServiceImpl.unableReceive.broadcastMessage"), t1);
                         }
                     }
                 } finally {
                     currentThread.setName(name);
                 }
             };
             executor.execute(t);
         }
     }

     protected final Object expiredMutex = new Object();

     protected void checkExpired() {
         synchronized (expiredMutex) {
             Member[] expired = membership.expire(timeToExpiration);
             for (final Member member : expired) {
                 if (log.isDebugEnabled()) {
                     log.debug(sm.getString("mcastServiceImpl.memberExpire", member));
                 }
                 try {
                     Runnable t = () -> {
                         Thread currentThread = Thread.currentThread();
                         String name = currentThread.getName();
                         try {
                             currentThread.setName("Membership-MemberExpired");
                             service.memberDisappeared(member);
                         } finally {
                             currentThread.setName(name);
                         }
                     };
                     executor.execute(t);
                 } catch (Exception e) {
                     log.error(sm.getString("mcastServiceImpl.memberDisappeared.failed"), e);
                 }
             }
         }
     }

     /**
      * Send a ping.
      *
      * @param checkexpired <code>true</code> to check for expiration
      *
      * @throws IOException Send error
      */
     public void send(boolean checkexpired) throws IOException {
         send(checkexpired, null);
     }

     private final Object sendLock = new Object();

     public void send(boolean checkexpired, DatagramPacket packet) throws IOException {
         checkexpired = (checkexpired && (packet == null));
         // ignore if we haven't started the sender
         // if ( (startLevel&Channel.MBR_TX_SEQ) != Channel.MBR_TX_SEQ ) return;
         if (packet == null) {
             member.inc();
             if (log.isTraceEnabled()) {
                 log.trace("Mcast send ping from member " + member);
             }
             byte[] data = member.getData();
             packet = new DatagramPacket(data, data.length);
         } else if (log.isTraceEnabled()) {
             log.trace("Sending message broadcast " + packet.getLength() + " bytes from " + member);
         }
         packet.setAddress(address);
         packet.setPort(port);
         // TODO this operation is not thread safe
         synchronized (sendLock) {
             socket.send(packet);
         }
         if (checkexpired) {
             checkExpired();
         }
     }

     public long getServiceStartTime() {
         return (member != null) ? member.getServiceStartTime() : -1L;
     }

     public int getRecoveryCounter() {
         return recoveryCounter;
     }

     public boolean isRecoveryEnabled() {
         return recoveryEnabled;
     }

     public long getRecoverySleepTime() {
         return recoverySleepTime;
     }

     public Channel getChannel() {
         return channel;
     }

     public void setChannel(Channel channel) {
         this.channel = channel;
     }

     public class ReceiverThread extends Thread {
         int errorCounter = 0;

         public ReceiverThread() {
             super();
             String channelName = "";
             if (channel.getName() != null) {
                 channelName = "[" + channel.getName() + "]";
             }
             setName("Tribes-MembershipReceiver" + channelName);
         }

         @Override
         public void run() {
             while (doRunReceiver) {
                 try {
                     receive();
                     errorCounter = 0;
                 } catch (ArrayIndexOutOfBoundsException ax) {
                     // we can ignore this, as it means we have an invalid package
                     // but we will log it to debug
                     if (log.isDebugEnabled()) {
                         log.debug(sm.getString("mcastServiceImpl.invalidMemberPackage"), ax);
                     }
                 } catch (Exception e) {
                     if (errorCounter == 0 && doRunReceiver) {
                         log.warn(sm.getString("mcastServiceImpl.error.receiving"), e);
                     } else if (log.isDebugEnabled()) {
                         if (doRunReceiver) {
                             log.debug(sm.getString("mcastServiceImpl.error.receiving"), e);
                         } else {
                             log.debug(sm.getString("mcastServiceImpl.error.receivingNoSleep"), e);
                         }
                     }
                     if (doRunReceiver) {
                         try {
                             sleep(500);
                         } catch (Exception ignore) {
                             // Ignore
                         }
                         if ((++errorCounter) >= recoveryCounter) {
                             errorCounter = 0;
                             RecoveryThread.recover(McastServiceImpl.this);
                         }
                     }
                 }
             }
         }
     }// class ReceiverThread

     public class SenderThread extends Thread {
         final long time;
         int errorCounter = 0;

         public SenderThread(long time) {
             this.time = time;
             String channelName = "";
             if (channel.getName() != null) {
                 channelName = "[" + channel.getName() + "]";
             }
             setName("Tribes-MembershipSender" + channelName);

         }

         @Override
         public void run() {
             while (doRunSender) {
                 try {
                     send(true);
                     errorCounter = 0;
                 } catch (Exception e) {
                     if (errorCounter == 0) {
                         log.warn(sm.getString("mcastServiceImpl.send.failed"), e);
                     } else {
                         log.debug(sm.getString("mcastServiceImpl.send.failed"), e);
                     }
                     if ((++errorCounter) >= recoveryCounter) {
                         errorCounter = 0;
                         RecoveryThread.recover(McastServiceImpl.this);
                     }
                 }
                 try {
                     sleep(time);
                 } catch (Exception ignore) {
                     // Ignore
                 }
             }
         }
     }// class SenderThread

     protected static class RecoveryThread extends Thread {

         private static final AtomicBoolean running = new AtomicBoolean(false);

         public static synchronized void recover(McastServiceImpl parent) {

             if (!parent.isRecoveryEnabled()) {
                 return;
             }

             if (!running.compareAndSet(false, true)) {
                 return;
             }

             Thread t = new RecoveryThread(parent);
             String channelName = "";
             if (parent.channel.getName() != null) {
                 channelName = "[" + parent.channel.getName() + "]";
             }
             t.setName("Tribes-MembershipRecovery" + channelName);
             t.setDaemon(true);
             t.start();
         }


         final McastServiceImpl parent;

         public RecoveryThread(McastServiceImpl parent) {
             this.parent = parent;
         }

         public boolean stopService() {
             try {
                 parent.stop(Channel.MBR_RX_SEQ | Channel.MBR_TX_SEQ);
                 return true;
             } catch (Exception e) {
                 log.warn(sm.getString("mcastServiceImpl.recovery.stopFailed"), e);
                 return false;
             }
         }

         public boolean startService() {
             try {
                 parent.init();
                 parent.start(Channel.MBR_RX_SEQ | Channel.MBR_TX_SEQ);
                 return true;
             } catch (Exception e) {
                 log.warn(sm.getString("mcastServiceImpl.recovery.startFailed"), e);
                 return false;
             }
         }

         @Override
         public void run() {
             boolean success = false;
             int attempt = 0;
             try {
                 while (!success) {
                     if (log.isInfoEnabled()) {
                         log.info(sm.getString("mcastServiceImpl.recovery"));
                     }
                     if (stopService() & startService()) {
                         success = true;
                         if (log.isInfoEnabled()) {
                             log.info(sm.getString("mcastServiceImpl.recovery.successful"));
                         }
                     }
                     try {
                         if (!success) {
                             if (log.isInfoEnabled()) {
                                 log.info(sm.getString("mcastServiceImpl.recovery.failed", Integer.toString(++attempt),
                                         Long.toString(parent.recoverySleepTime)));
                             }
                             sleep(parent.recoverySleepTime);
                         }
                     } catch (InterruptedException ignore) {
                     }
                 }
             } finally {
                 running.set(false);
             }
         }
     }

     public void setRecoveryCounter(int recoveryCounter) {
         this.recoveryCounter = recoveryCounter;
     }

     public void setRecoveryEnabled(boolean recoveryEnabled) {
         this.recoveryEnabled = recoveryEnabled;
     }

     public void setRecoverySleepTime(long recoverySleepTime) {
         this.recoverySleepTime = recoverySleepTime;
     }
 }
	/*
	* 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.catalina.tribes.membership;


	import java.io.IOException;
	import java.net.BindException;
	import java.net.DatagramPacket;
	import java.net.InetAddress;
	import java.net.InetSocketAddress;
	import java.net.MulticastSocket;
	import java.net.NetworkInterface;
	import java.net.SocketTimeoutException;
	import java.net.StandardSocketOptions;
	import java.util.Arrays;
	import java.util.concurrent.atomic.AtomicBoolean;

	import org.apache.catalina.tribes.Channel;
	import org.apache.catalina.tribes.Member;
	import org.apache.catalina.tribes.MembershipListener;
	import org.apache.catalina.tribes.MessageListener;
	import org.apache.catalina.tribes.io.ChannelData;
	import org.apache.catalina.tribes.io.XByteBuffer;
	import org.apache.catalina.tribes.util.StringManager;
	import org.apache.juli.logging.Log;
	import org.apache.juli.logging.LogFactory;

	/**
	* A <b>membership</b> implementation using simple multicast. This is the representation of a multicast membership
	* service. This class is responsible for maintaining a list of active cluster nodes in the cluster. If a node fails to
	* send out a heartbeat, the node will be dismissed. This is the low level implementation that handles the multicasting
	* sockets. Need to fix this, could use java.nio and only need one thread to send and receive, or just use a timeout on
	* the receive
	*/
	public class McastServiceImpl extends MembershipProviderBase {

	private static final Log log = LogFactory.getLog(McastService.class);

	protected static final int MAX_PACKET_SIZE = 65535;

	protected static final StringManager sm = StringManager.getManager(Constants.Package);
	/**
	* Internal flag used for the listen thread that listens to the multicasting socket.
	*/
	protected volatile boolean doRunSender = false;
	protected volatile boolean doRunReceiver = false;
	protected volatile int startLevel = 0;
	/**
	* Socket that we intend to listen to
	*/
	protected MulticastSocket socket;
	/**
	* The local member that we intend to broadcast over and over again
	*/
	protected final MemberImpl member;
	/**
	* The multicast address
	*/
	protected final InetAddress address;
	/**
	* The multicast port
	*/
	protected final int port;
	/**
	* The time it takes for a member to expire.
	*/
	protected final long timeToExpiration;
	/**
	* How often to we send out a broadcast saying we are alive, must be smaller than timeToExpiration
	*/
	protected final long sendFrequency;
	/**
	* Reuse the sendPacket, no need to create a new one every time
	*/
	protected DatagramPacket sendPacket;
	/**
	* Reuse the receivePacket, no need to create a new one every time
	*/
	protected DatagramPacket receivePacket;

	/**
	* The actual listener, for callback when stuff goes down
	*/
	protected final MembershipListener service;
	/**
	* The actual listener for broadcast callbacks
	*/
	protected final MessageListener msgservice;
	/**
	* Thread to listen for pings
	*/
	protected ReceiverThread receiver;
	/**
	* Thread to send pings
	*/
	protected SenderThread sender;

	/**
	* Time to live for the multicast packets that are being sent out
	*/
	protected final int mcastTTL;
	/**
	* Read timeout on the mcast socket
	*/
	protected int mcastSoTimeout;
	/**
	* bind address
	*/
	protected final InetAddress mcastBindAddress;

	/**
	* nr of times the system has to fail before a recovery is initiated
	*/
	protected int recoveryCounter = 10;

	/**
	* The time the recovery thread sleeps between recovery attempts
	*/
	protected long recoverySleepTime = 5000;

	/**
	* Add the ability to turn on/off recovery
	*/
	protected boolean recoveryEnabled = true;

	/**
	* disable/enable local loopback message
	*/
	protected final boolean localLoopbackDisabled;

	private Channel channel;

	/**
	* Create a new mcast service instance.
	*
	* @param member - the local member
	* @param sendFrequency - the time (ms) in between pings sent out
	* @param expireTime - the time (ms) for a member to expire
	* @param port - the mcast port
	* @param bind - the bind address (not sure this is used yet)
	* @param mcastAddress - the mcast address
	* @param ttl multicast ttl that will be set on the socket
	* @param soTimeout Socket timeout
	* @param service - the callback service
	* @param msgservice Message listener
	* @param localLoopbackDisabled - disable loopbackMode
	*
	* @throws IOException Init error
	*/
	public McastServiceImpl(MemberImpl member, long sendFrequency, long expireTime, int port, InetAddress bind,
	InetAddress mcastAddress, int ttl, int soTimeout, MembershipListener service, MessageListener msgservice,
	boolean localLoopbackDisabled) throws IOException {
	this.member = member;
	this.address = mcastAddress;
	this.port = port;
	this.mcastSoTimeout = soTimeout;
	this.mcastTTL = ttl;
	this.mcastBindAddress = bind;
	this.timeToExpiration = expireTime;
	this.service = service;
	this.msgservice = msgservice;
	this.sendFrequency = sendFrequency;
	this.localLoopbackDisabled = localLoopbackDisabled;
	init();
	}

	public void init() throws IOException {
	setupSocket();
	sendPacket = new DatagramPacket(new byte[MAX_PACKET_SIZE], MAX_PACKET_SIZE);
	sendPacket.setAddress(address);
	sendPacket.setPort(port);
	receivePacket = new DatagramPacket(new byte[MAX_PACKET_SIZE], MAX_PACKET_SIZE);
	receivePacket.setAddress(address);
	receivePacket.setPort(port);
	member.setCommand(new byte[0]);
	if (membership == null) {
	membership = new Membership(member);
	}
	}

	protected void setupSocket() throws IOException {
	if (mcastBindAddress != null) {
	try {
	log.info(sm.getString("mcastServiceImpl.bind", address, Integer.toString(port)));
	socket = new MulticastSocket(new InetSocketAddress(address, port));
	} catch (BindException e) {
	/*
	* On some platforms (e.g. Linux) it is not possible to bind to the multicast address. In this case only
	* bind to the port.
	*/
	if (log.isDebugEnabled()) {
	log.debug(sm.getString("mcastServiceImpl.bind.failed"), e);
	} else {
	log.info(sm.getString("mcastServiceImpl.bind.failed"));
	}
	socket = new MulticastSocket(port);
	}
	} else {
	socket = new MulticastSocket(port);
	}
	// Hint if we want to disable loop back(local machine) messages
	socket.setOption(StandardSocketOptions.IP_MULTICAST_LOOP, Boolean.valueOf(!localLoopbackDisabled));
	if (mcastBindAddress != null) {
	if (log.isInfoEnabled()) {
	log.info(sm.getString("mcastServiceImpl.setInterface", mcastBindAddress));
	}
	NetworkInterface networkInterface = NetworkInterface.getByInetAddress(mcastBindAddress);
	socket.setNetworkInterface(networkInterface);
	} // end if
	// force a so timeout so that we don't block forever
	if (mcastSoTimeout <= 0) {
	mcastSoTimeout = (int) sendFrequency;
	}
	if (log.isInfoEnabled()) {
	log.info(sm.getString("mcastServiceImpl.setSoTimeout", Integer.toString(mcastSoTimeout)));
	}
	socket.setSoTimeout(mcastSoTimeout);

	if (mcastTTL >= 0) {
	if (log.isInfoEnabled()) {
	log.info(sm.getString("mcastServiceImpl.setTTL", Integer.toString(mcastTTL)));
	}
	socket.setTimeToLive(mcastTTL);
	}
	}


	@Override
	public synchronized void start(int level) throws IOException {
	boolean valid = false;
	if ((level & Channel.MBR_RX_SEQ) == Channel.MBR_RX_SEQ) {
	if (receiver != null) {
	throw new IllegalStateException(sm.getString("mcastServiceImpl.receive.running"));
	}
	try {
	if (sender == null) {
	socket.joinGroup(new InetSocketAddress(address, 0), null);
	}
	} catch (IOException iox) {
	log.error(sm.getString("mcastServiceImpl.unable.join"));
	throw iox;
	}
	doRunReceiver = true;
	receiver = new ReceiverThread();
	receiver.setDaemon(true);
	receiver.start();
	valid = true;
	}
	if ((level & Channel.MBR_TX_SEQ) == Channel.MBR_TX_SEQ) {
	if (sender != null) {
	throw new IllegalStateException(sm.getString("mcastServiceImpl.send.running"));
	}
	if (receiver == null) {
	socket.joinGroup(new InetSocketAddress(address, 0), null);
	}
	// make sure at least one packet gets out there
	send(false);
	doRunSender = true;
	sender = new SenderThread(sendFrequency);
	sender.setDaemon(true);
	sender.start();
	// we have started the receiver, but not yet waited for membership to establish
	valid = true;
	}
	if (!valid) {
	throw new IllegalArgumentException(sm.getString("mcastServiceImpl.invalid.startLevel"));
	}
	// pause, once or twice
	waitForMembers(level);
	startLevel = (startLevel \| level);
	}

	private void waitForMembers(int level) {
	long memberwait = sendFrequency * 2;
	if (log.isInfoEnabled()) {
	log.info(sm.getString("mcastServiceImpl.waitForMembers.start", Long.toString(memberwait),
	Integer.toString(level)));
	}
	try {
	Thread.sleep(memberwait);
	} catch (InterruptedException ignore) {
	// Ignore
	}
	if (log.isInfoEnabled()) {
	log.info(sm.getString("mcastServiceImpl.waitForMembers.done", Integer.toString(level)));
	}
	}

	@Override
	public synchronized boolean stop(int level) throws IOException {
	boolean valid = false;

	if ((level & Channel.MBR_RX_SEQ) == Channel.MBR_RX_SEQ) {
	valid = true;
	doRunReceiver = false;
	if (receiver != null) {
	receiver.interrupt();
	}
	receiver = null;
	}
	if ((level & Channel.MBR_TX_SEQ) == Channel.MBR_TX_SEQ) {
	valid = true;
	doRunSender = false;
	if (sender != null) {
	sender.interrupt();
	}
	sender = null;
	}

	if (!valid) {
	throw new IllegalArgumentException(sm.getString("mcastServiceImpl.invalid.stopLevel"));
	}
	startLevel = (startLevel & (~level));
	// we're shutting down, send a shutdown message and close the socket
	if (startLevel == 0) {
	// send a stop message
	member.setCommand(Member.SHUTDOWN_PAYLOAD);
	send(false);
	// leave mcast group
	try {
	socket.leaveGroup(new InetSocketAddress(address, 0), null);
	} catch (Exception e) {
	// Shutting down. Only log at debug.
	if (log.isDebugEnabled()) {
	log.debug(sm.getString("mcastServiceImpl.error.stop"), e);
	}
	}
	try {
	socket.close();
	} catch (Exception e) {
	// Shutting down. Only log at debug.
	if (log.isDebugEnabled()) {
	log.debug(sm.getString("mcastServiceImpl.error.stop"), e);
	}
	}
	member.setServiceStartTime(-1);
	}
	return (startLevel == 0);
	}

	/**
	* Receive a datagram packet, locking wait
	*
	* @throws IOException Received failed
	*/
	public void receive() throws IOException {
	try {
	socket.receive(receivePacket);
	if (receivePacket.getLength() > MAX_PACKET_SIZE) {
	log.error(sm.getString("mcastServiceImpl.packet.tooLong", Integer.toString(receivePacket.getLength())));
	} else {
	byte[] data = new byte[receivePacket.getLength()];
	System.arraycopy(receivePacket.getData(), receivePacket.getOffset(), data, 0, data.length);
	if (XByteBuffer.firstIndexOf(data, 0, MemberImpl.TRIBES_MBR_BEGIN) == 0) {
	memberDataReceived(data);
	} else {
	memberBroadcastsReceived(data);
	}
	}
	} catch (SocketTimeoutException ignore) {
	/*
	* Do nothing. This is normal. We don't want to block forever since the receive thread is the same thread
	* that does membership expiration.
	*/
	}
	checkExpired();
	}

	private void memberDataReceived(byte[] data) {
	final Member m = MemberImpl.getMember(data);
	if (log.isTraceEnabled()) {
	log.trace("Mcast receive ping from member " + m);
	}
	Runnable t = null;
	Thread currentThread = Thread.currentThread();
	if (Arrays.equals(m.getCommand(), Member.SHUTDOWN_PAYLOAD)) {
	if (log.isDebugEnabled()) {
	log.debug(sm.getString("mcastServiceImpl.memberShutdown", m));
	}
	membership.removeMember(m);
	t = () -> {
	String name = currentThread.getName();
	try {
	currentThread.setName("Membership-MemberDisappeared");
	service.memberDisappeared(m);
	} finally {
	currentThread.setName(name);
	}
	};
	} else if (membership.memberAlive(m)) {
	if (log.isDebugEnabled()) {
	log.debug(sm.getString("mcastServiceImpl.memberAdd", m));
	}
	t = () -> {
	String name = currentThread.getName();
	try {
	currentThread.setName("Membership-MemberAdded");
	service.memberAdded(m);
	} finally {
	currentThread.setName(name);
	}
	};
	}
	if (t != null) {
	executor.execute(t);
	}
	}

	private void memberBroadcastsReceived(final byte[] b) {
	if (log.isTraceEnabled()) {
	log.trace("Mcast received broadcasts.");
	}
	XByteBuffer buffer = new XByteBuffer(b, true);
	if (buffer.countPackages(true) > 0) {
	int count = buffer.countPackages();
	final ChannelData[] data = new ChannelData[count];
	for (int i = 0; i < count; i++) {
	try {
	data[i] = buffer.extractPackage(true);
	} catch (IllegalStateException ise) {
	log.debug(sm.getString("mcastServiceImpl.messageError"), ise);
	}
	}
	Runnable t = () -> {
	Thread currentThread = Thread.currentThread();
	String name = currentThread.getName();
	try {
	currentThread.setName("Membership-MemberAdded");
	for (ChannelData datum : data) {
	try {
	if (datum != null && !member.equals(datum.getAddress())) {
	msgservice.messageReceived(datum);
	}
	} catch (Throwable t1) {
	if (t1 instanceof VirtualMachineError) {
	throw (VirtualMachineError) t1;
	}
	log.error(sm.getString("mcastServiceImpl.unableReceive.broadcastMessage"), t1);
	}
	}
	} finally {
	currentThread.setName(name);
	}
	};
	executor.execute(t);
	}
	}

	protected final Object expiredMutex = new Object();

	protected void checkExpired() {
	synchronized (expiredMutex) {
	Member[] expired = membership.expire(timeToExpiration);
	for (final Member member : expired) {
	if (log.isDebugEnabled()) {
	log.debug(sm.getString("mcastServiceImpl.memberExpire", member));
	}
	try {
	Runnable t = () -> {
	Thread currentThread = Thread.currentThread();
	String name = currentThread.getName();
	try {
	currentThread.setName("Membership-MemberExpired");
	service.memberDisappeared(member);
	} finally {
	currentThread.setName(name);
	}
	};
	executor.execute(t);
	} catch (Exception e) {
	log.error(sm.getString("mcastServiceImpl.memberDisappeared.failed"), e);
	}
	}
	}
	}

	/**
	* Send a ping.
	*
	* @param checkexpired <code>true</code> to check for expiration
	*
	* @throws IOException Send error
	*/
	public void send(boolean checkexpired) throws IOException {
	send(checkexpired, null);
	}

	private final Object sendLock = new Object();

	public void send(boolean checkexpired, DatagramPacket packet) throws IOException {
	checkexpired = (checkexpired && (packet == null));
	// ignore if we haven't started the sender
	// if ( (startLevel&Channel.MBR_TX_SEQ) != Channel.MBR_TX_SEQ ) return;
	if (packet == null) {
	member.inc();
	if (log.isTraceEnabled()) {
	log.trace("Mcast send ping from member " + member);
	}
	byte[] data = member.getData();
	packet = new DatagramPacket(data, data.length);
	} else if (log.isTraceEnabled()) {
	log.trace("Sending message broadcast " + packet.getLength() + " bytes from " + member);
	}
	packet.setAddress(address);
	packet.setPort(port);
	// TODO this operation is not thread safe
	synchronized (sendLock) {
	socket.send(packet);
	}
	if (checkexpired) {
	checkExpired();
	}
	}

	public long getServiceStartTime() {
	return (member != null) ? member.getServiceStartTime() : -1L;
	}

	public int getRecoveryCounter() {
	return recoveryCounter;
	}

	public boolean isRecoveryEnabled() {
	return recoveryEnabled;
	}

	public long getRecoverySleepTime() {
	return recoverySleepTime;
	}

	public Channel getChannel() {
	return channel;
	}

	public void setChannel(Channel channel) {
	this.channel = channel;
	}

	public class ReceiverThread extends Thread {
	int errorCounter = 0;

	public ReceiverThread() {
	super();
	String channelName = "";
	if (channel.getName() != null) {
	channelName = "[" + channel.getName() + "]";
	}
	setName("Tribes-MembershipReceiver" + channelName);
	}

	@Override
	public void run() {
	while (doRunReceiver) {
	try {
	receive();
	errorCounter = 0;
	} catch (ArrayIndexOutOfBoundsException ax) {
	// we can ignore this, as it means we have an invalid package
	// but we will log it to debug
	if (log.isDebugEnabled()) {
	log.debug(sm.getString("mcastServiceImpl.invalidMemberPackage"), ax);
	}
	} catch (Exception e) {
	if (errorCounter == 0 && doRunReceiver) {
	log.warn(sm.getString("mcastServiceImpl.error.receiving"), e);
	} else if (log.isDebugEnabled()) {
	if (doRunReceiver) {
	log.debug(sm.getString("mcastServiceImpl.error.receiving"), e);
	} else {
	log.debug(sm.getString("mcastServiceImpl.error.receivingNoSleep"), e);
	}
	}
	if (doRunReceiver) {
	try {
	sleep(500);
	} catch (Exception ignore) {
	// Ignore
	}
	if ((++errorCounter) >= recoveryCounter) {
	errorCounter = 0;
	RecoveryThread.recover(McastServiceImpl.this);
	}
	}
	}
	}
	}
	}// class ReceiverThread

	public class SenderThread extends Thread {
	final long time;
	int errorCounter = 0;

	public SenderThread(long time) {
	this.time = time;
	String channelName = "";
	if (channel.getName() != null) {
	channelName = "[" + channel.getName() + "]";
	}
	setName("Tribes-MembershipSender" + channelName);

	}

	@Override
	public void run() {
	while (doRunSender) {
	try {
	send(true);
	errorCounter = 0;
	} catch (Exception e) {
	if (errorCounter == 0) {
	log.warn(sm.getString("mcastServiceImpl.send.failed"), e);
	} else {
	log.debug(sm.getString("mcastServiceImpl.send.failed"), e);
	}
	if ((++errorCounter) >= recoveryCounter) {
	errorCounter = 0;
	RecoveryThread.recover(McastServiceImpl.this);
	}
	}
	try {
	sleep(time);
	} catch (Exception ignore) {
	// Ignore
	}
	}
	}
	}// class SenderThread

	protected static class RecoveryThread extends Thread {

	private static final AtomicBoolean running = new AtomicBoolean(false);

	public static synchronized void recover(McastServiceImpl parent) {

	if (!parent.isRecoveryEnabled()) {
	return;
	}

	if (!running.compareAndSet(false, true)) {
	return;
	}

	Thread t = new RecoveryThread(parent);
	String channelName = "";
	if (parent.channel.getName() != null) {
	channelName = "[" + parent.channel.getName() + "]";
	}
	t.setName("Tribes-MembershipRecovery" + channelName);
	t.setDaemon(true);
	t.start();
	}


	final McastServiceImpl parent;

	public RecoveryThread(McastServiceImpl parent) {
	this.parent = parent;
	}

	public boolean stopService() {
	try {
	parent.stop(Channel.MBR_RX_SEQ \| Channel.MBR_TX_SEQ);
	return true;
	} catch (Exception e) {
	log.warn(sm.getString("mcastServiceImpl.recovery.stopFailed"), e);
	return false;
	}
	}

	public boolean startService() {
	try {
	parent.init();
	parent.start(Channel.MBR_RX_SEQ \| Channel.MBR_TX_SEQ);
	return true;
	} catch (Exception e) {
	log.warn(sm.getString("mcastServiceImpl.recovery.startFailed"), e);
	return false;
	}
	}

	@Override
	public void run() {
	boolean success = false;
	int attempt = 0;
	try {
	while (!success) {
	if (log.isInfoEnabled()) {
	log.info(sm.getString("mcastServiceImpl.recovery"));
	}
	if (stopService() & startService()) {
	success = true;
	if (log.isInfoEnabled()) {
	log.info(sm.getString("mcastServiceImpl.recovery.successful"));
	}
	}
	try {
	if (!success) {
	if (log.isInfoEnabled()) {
	log.info(sm.getString("mcastServiceImpl.recovery.failed", Integer.toString(++attempt),
	Long.toString(parent.recoverySleepTime)));
	}
	sleep(parent.recoverySleepTime);
	}
	} catch (InterruptedException ignore) {
	}
	}
	} finally {
	running.set(false);
	}
	}
	}

	public void setRecoveryCounter(int recoveryCounter) {
	this.recoveryCounter = recoveryCounter;
	}

	public void setRecoveryEnabled(boolean recoveryEnabled) {
	this.recoveryEnabled = recoveryEnabled;
	}

	public void setRecoverySleepTime(long recoverySleepTime) {
	this.recoverySleepTime = recoverySleepTime;
	}
	}