<commit_before>/*Copyright (C) 2010-2012 Institute for Software Integrated Systems (ISIS)
This software was developed by the Institute for Software Integrated
Systems (ISIS) at Vanderbilt University, Tennessee, USA for the 
Transformative Apps program under DARPA, Contract # HR011-10-C-0175.
The United States Government has unlimited rights to this software. 
The US government has the right to use, modify, reproduce, release, 
perform, display, or disclose computer software or computer software 
documentation in whole or in part, in any manner and for any 
purpose whatsoever, and to have or authorize others to do so.
*/
package edu.vu.isis.ammo.gateway;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.NetworkInterface;
import java.net.Socket;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.BufferUnderflowException;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SocketChannel;
import java.util.Enumeration;
import java.util.LinkedList;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.zip.CRC32;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import ammo.gateway.protocol.GatewayPrivateMessages;


/**
 * Two long running threads and one short.
 * The long threads are for sending and receiving messages.
 * The short thread is to connect the socket.
 * The sent messages are placed into a queue if the socket is connected.
 *
 */
class NetworkConnector {
    private static final Logger logger = LoggerFactory.getLogger(NetworkConnector.class);

    private static final int BURP_TIME = 5 * 1000; // 5 seconds expressed in milliseconds
    
    /**
     * $ sysctl net.ipv4.tcp_rmem 
     * or
     * $ cat /proc/sys/net/ipv4/tcp_rmem
     * 4096   87380   4194304
     * 0x1000 0x15554 0x400000
     * 
     * The first value tells the kernel the minimum receive buffer for each TCP connection, and 
     * this buffer is always allocated to a TCP socket, even under high pressure on the system.
     * 
     * The second value specified tells the kernel the default receive buffer allocated for each TCP socket. 
     * This value overrides the /proc/sys/net/core/rmem_default value used by other protocols.
     * 
     * The third and last value specified in this variable specifies the maximum receive buffer 
     * that can be allocated for a TCP socket.
     * 
     */
    private static final int TCP_RECV_BUFF_SIZE = 0x15554; // the maximum receive buffer size
    private static final int MAX_MESSAGE_SIZE = 0x100000;  // arbitrary max size

    private SocketChannel mSocketChannel;

    // Three threads
    private ConnectorThread connectorThread; // long running
    private SenderThread mSender;	     // created after connection
    private ReceiverThread mReceiver;	     // created after connection

    private String gatewayHost = null;
    private int gatewayPort = -1;

    private ByteOrder endian = ByteOrder.LITTLE_ENDIAN;

    private final SenderQueue mSenderQueue;
    private final GatewayConnector mGatewayConnector;

    /**
     * Constants
     **/
    static final int EXCEPTION       =  1; // the run failed by some unhandled exception

    static final int CONNECTING      = 20; // trying to connect
    static final int CONNECTED       = 21; // the socket is good an active

    static final int DISCONNECTED    = 30; // the socket is disconnected
    static final int STALE           = 31; // indicating there is a message

    static final int WAIT_CONNECT    = 40; // waiting for connection
    static final int SENDING         = 41; // indicating the next thing is the size
    static final int TAKING          = 42; // indicating the next thing is the size
    static final int INTERRUPTED     = 43; // the run was canceled via an interrupt

    static final int START           = 52; // indicating the next thing is the size
    static final int RESTART         = 53; // indicating the next thing is the size
    static final int WAIT_RECONNECT  = 54; // waiting for connection
    static final int STARTED         = 55; // indicating there is a message
    static final int SIZED           = 56; // indicating the next thing is a checksum
    static final int CHECKED         = 57; // indicating the bytes are being read
    static final int DELIVER         = 58; // indicating the message has been read


    static final int GATEWAY_HEADER_SIZE =
	  4 // magic
	+ 4 // message size
	+ 1 // priority
	+ 3 // reserved
	+ 4 // payload checksum
	+ 4; // header checksum
    static final int GATEWAY_MESSAGE_MAGIC = 0xdeadbeef;
    static final byte[] GATEWAY_MESSAGE_MAGICB = { (byte)0xde, (byte)0xad, (byte)0xbe, (byte)0xef };
    

    public NetworkConnector(GatewayConnector gatewayConnector, String gatewayHost, int gatewayPort) {
	logger.info("Thread <{}>NetworkConnector::<constructor>", Thread.currentThread().getId());
	this.mGatewayConnector = gatewayConnector;
	this.gatewayHost = gatewayHost;
	this.gatewayPort = gatewayPort;

	this.connectorThread = new ConnectorThread(this);
	mSenderQueue = new SenderQueue( this );
	this.connectorThread.start();
    }

    public boolean isConnected() { 
	return this.connectorThread.isConnected(); 
    }

    public boolean close() { return false; }

    public String toString() {
	return new StringBuilder().append("channel ").append(super.toString())
            .append("socket: host[").append(this.gatewayHost).append("] ")
            .append("port[").append(this.gatewayPort).append("]").toString();
    }

    /**
     * forces a reconnection.
     */
    // Called by ReceiverThread to send an incoming message to the
    // appropriate destination.
    private boolean deliverMessage( GatewayPrivateMessages.GatewayWrapper message )
    {
	logger.info( "deliverMessage() {} ", message );
	// hand it off to parentconnector
	if (message.getType() == GatewayPrivateMessages.GatewayWrapper.MessageType.ASSOCIATE_RESULT) {
	    logger.debug("Received Associate Result ...");
	    mGatewayConnector.onAssociateResultReceived(message.getAssociateResult() );
	} else if (message.getType() == GatewayPrivateMessages.GatewayWrapper.MessageType.PUSH_DATA) {
	    logger.debug("Received Push Data ...");
	    mGatewayConnector.onPushDataReceived(message.getPushData() );
	} else if (message.getType() == GatewayPrivateMessages.GatewayWrapper.MessageType.PULL_REQUEST) {
	    logger.debug("Received Pull Request ...");
	    mGatewayConnector.onPullRequestReceived(message.getPullRequest() );
	} else if (message.getType() == GatewayPrivateMessages.GatewayWrapper.MessageType.PULL_RESPONSE) {
	    logger.debug("Received Pull Response ...");
	    mGatewayConnector.onPullResponseReceived(message.getPullResponse() );
	}

	return true;
    }

    /**
     * do your best to send the message.
     * This makes use of the blocking "put" call.
     * A proper producer-consumer should use put or add and not offer.
     * "put" is blocking call.
     * If this were on the UI thread then offer would be used.
     *
     * @param agm GatewayPrivateMessages.GatewayWrapper
     * @return
     */
    public boolean sendMessage( GatewayPrivateMessages.GatewayWrapper agm )
    {
        return mSenderQueue.put( agm );
    }

    ///////////////////////////////////////////////////////////////////////////
    //
    class SenderQueue
    {
        public SenderQueue( NetworkConnector iChannel )
        {
            mChannel = iChannel;
            mDistQueue = new PriorityBlockingQueue<GatewayPrivateMessages.GatewayWrapper>( 20 );
        }

        public boolean put(GatewayPrivateMessages.GatewayWrapper iMessage )
        {<commit_msg>Fixed the missing comparator bug in SenderQueue<commit_after>/*Copyright (C) 2010-2012 Institute for Software Integrated Systems (ISIS)
This software was developed by the Institute for Software Integrated
Systems (ISIS) at Vanderbilt University, Tennessee, USA for the 
Transformative Apps program under DARPA, Contract # HR011-10-C-0175.
The United States Government has unlimited rights to this software. 
The US government has the right to use, modify, reproduce, release, 
perform, display, or disclose computer software or computer software 
documentation in whole or in part, in any manner and for any 
purpose whatsoever, and to have or authorize others to do so.
*/
package edu.vu.isis.ammo.gateway;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.NetworkInterface;
import java.net.Socket;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.BufferUnderflowException;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SocketChannel;
import java.util.Enumeration;
import java.util.LinkedList;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.Comparator;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.zip.CRC32;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import ammo.gateway.protocol.GatewayPrivateMessages;


/**
 * Two long running threads and one short.
 * The long threads are for sending and receiving messages.
 * The short thread is to connect the socket.
 * The sent messages are placed into a queue if the socket is connected.
 *
 */
class NetworkConnector {
    private static final Logger logger = LoggerFactory.getLogger(NetworkConnector.class);

    private static final int BURP_TIME = 5 * 1000; // 5 seconds expressed in milliseconds
    
    /**
     * $ sysctl net.ipv4.tcp_rmem 
     * or
     * $ cat /proc/sys/net/ipv4/tcp_rmem
     * 4096   87380   4194304
     * 0x1000 0x15554 0x400000
     * 
     * The first value tells the kernel the minimum receive buffer for each TCP connection, and 
     * this buffer is always allocated to a TCP socket, even under high pressure on the system.
     * 
     * The second value specified tells the kernel the default receive buffer allocated for each TCP socket. 
     * This value overrides the /proc/sys/net/core/rmem_default value used by other protocols.
     * 
     * The third and last value specified in this variable specifies the maximum receive buffer 
     * that can be allocated for a TCP socket.
     * 
     */
    private static final int TCP_RECV_BUFF_SIZE = 0x15554; // the maximum receive buffer size
    private static final int MAX_MESSAGE_SIZE = 0x100000;  // arbitrary max size

    private SocketChannel mSocketChannel;

    // Three threads
    private ConnectorThread connectorThread; // long running
    private SenderThread mSender;	     // created after connection
    private ReceiverThread mReceiver;	     // created after connection

    private String gatewayHost = null;
    private int gatewayPort = -1;

    private ByteOrder endian = ByteOrder.LITTLE_ENDIAN;

    private final SenderQueue mSenderQueue;
    private final GatewayConnector mGatewayConnector;

    /**
     * Constants
     **/
    static final int EXCEPTION       =  1; // the run failed by some unhandled exception

    static final int CONNECTING      = 20; // trying to connect
    static final int CONNECTED       = 21; // the socket is good an active

    static final int DISCONNECTED    = 30; // the socket is disconnected
    static final int STALE           = 31; // indicating there is a message

    static final int WAIT_CONNECT    = 40; // waiting for connection
    static final int SENDING         = 41; // indicating the next thing is the size
    static final int TAKING          = 42; // indicating the next thing is the size
    static final int INTERRUPTED     = 43; // the run was canceled via an interrupt

    static final int START           = 52; // indicating the next thing is the size
    static final int RESTART         = 53; // indicating the next thing is the size
    static final int WAIT_RECONNECT  = 54; // waiting for connection
    static final int STARTED         = 55; // indicating there is a message
    static final int SIZED           = 56; // indicating the next thing is a checksum
    static final int CHECKED         = 57; // indicating the bytes are being read
    static final int DELIVER         = 58; // indicating the message has been read


    static final int GATEWAY_HEADER_SIZE =
	  4 // magic
	+ 4 // message size
	+ 1 // priority
	+ 3 // reserved
	+ 4 // payload checksum
	+ 4; // header checksum
    static final int GATEWAY_MESSAGE_MAGIC = 0xdeadbeef;
    static final byte[] GATEWAY_MESSAGE_MAGICB = { (byte)0xde, (byte)0xad, (byte)0xbe, (byte)0xef };
    

    public NetworkConnector(GatewayConnector gatewayConnector, String gatewayHost, int gatewayPort) {
	logger.info("Thread <{}>NetworkConnector::<constructor>", Thread.currentThread().getId());
	this.mGatewayConnector = gatewayConnector;
	this.gatewayHost = gatewayHost;
	this.gatewayPort = gatewayPort;

	this.connectorThread = new ConnectorThread(this);
	mSenderQueue = new SenderQueue( this );
	this.connectorThread.start();
    }

    public boolean isConnected() { 
	return this.connectorThread.isConnected(); 
    }

    public boolean close() { return false; }

    public String toString() {
	return new StringBuilder().append("channel ").append(super.toString())
            .append("socket: host[").append(this.gatewayHost).append("] ")
            .append("port[").append(this.gatewayPort).append("]").toString();
    }

    /**
     * forces a reconnection.
     */
    // Called by ReceiverThread to send an incoming message to the
    // appropriate destination.
    private boolean deliverMessage( GatewayPrivateMessages.GatewayWrapper message )
    {
	logger.info( "deliverMessage() {} ", message );
	// hand it off to parentconnector
	if (message.getType() == GatewayPrivateMessages.GatewayWrapper.MessageType.ASSOCIATE_RESULT) {
	    logger.debug("Received Associate Result ...");
	    mGatewayConnector.onAssociateResultReceived(message.getAssociateResult() );
	} else if (message.getType() == GatewayPrivateMessages.GatewayWrapper.MessageType.PUSH_DATA) {
	    logger.debug("Received Push Data ...");
	    mGatewayConnector.onPushDataReceived(message.getPushData() );
	} else if (message.getType() == GatewayPrivateMessages.GatewayWrapper.MessageType.PULL_REQUEST) {
	    logger.debug("Received Pull Request ...");
	    mGatewayConnector.onPullRequestReceived(message.getPullRequest() );
	} else if (message.getType() == GatewayPrivateMessages.GatewayWrapper.MessageType.PULL_RESPONSE) {
	    logger.debug("Received Pull Response ...");
	    mGatewayConnector.onPullResponseReceived(message.getPullResponse() );
	}

	return true;
    }

    /**
     * do your best to send the message.
     * This makes use of the blocking "put" call.
     * A proper producer-consumer should use put or add and not offer.
     * "put" is blocking call.
     * If this were on the UI thread then offer would be used.
     *
     * @param agm GatewayPrivateMessages.GatewayWrapper
     * @return
     */
    public boolean sendMessage( GatewayPrivateMessages.GatewayWrapper agm )
    {
        return mSenderQueue.put( agm );
    }

    ///////////////////////////////////////////////////////////////////////////
    //
    class SenderQueue
    {
	class PriorityComparator implements Comparator<GatewayPrivateMessages.GatewayWrapper> {
	    @Override
		public int  compare(GatewayPrivateMessages.GatewayWrapper x, GatewayPrivateMessages.GatewayWrapper y)
	    {
		if (x.getMessagePriority() > y.getMessagePriority() )
		    return 1;
		else if (x.getMessagePriority() == y.getMessagePriority() )
		    return 0;

		return -1;
	    }
	    
	}

        public SenderQueue( NetworkConnector iChannel )
        {
            mChannel = iChannel;
            mDistQueue = new PriorityBlockingQueue<GatewayPrivateMessages.GatewayWrapper>( 20, new PriorityComparator() );
        }

        public boolean put(GatewayPrivateMessages.GatewayWrapper iMessage )
        {
{'fraction_non_alphanumeric': 0.08971804624766044, 'fraction_numerical': 0.014792006279055726, 'mean_word_length': 3.1733434114386494, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '376750', 'n_tokens_mistral': 4702, 'n_tokens_neox': 4407, 'n_words': 1488}