<commit_before>package org.jgroups.blocks;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.BindException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.NetworkInterface;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.SocketException;
import java.util.Collection;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jgroups.Address;
import org.jgroups.Version;
import org.jgroups.stack.IpAddress;
import org.jgroups.util.PortsManager;
import org.jgroups.util.ThreadFactory;
import org.jgroups.util.Util;

public class TCPConnectionMap{

    private final Mapper mapper;
    private final InetAddress bind_addr;
    private final Address local_addr; // bind_addr + port of srv_sock
    private final ThreadGroup thread_group=new ThreadGroup(Util.getGlobalThreadGroup(),"ConnectionTable");     
    private final ServerSocket srv_sock;
    private final Receiver receiver;
    private final long conn_expire_time;
    private final Log log=LogFactory.getLog(getClass());
    private int recv_buf_size=120000;
    private int send_buf_size=60000;
    private int send_queue_size = 0;
    private int sock_conn_timeout=1000; // max time in millis to wait for Socket.connect() to return    
    private boolean tcp_nodelay=false;
    private int linger=-1;    
    private Thread acceptor;
    private PortsManager pm=null;       

    private volatile boolean running = false;    

    public TCPConnectionMap(ThreadFactory f,
                            Receiver r,
                            InetAddress bind_addr,
                            InetAddress external_addr,
                            int srv_port,
                            int max_port,
                            PortsManager pm) throws Exception {
        this(f,r,bind_addr,external_addr,srv_port,max_port,0,0,pm);
    }

    public TCPConnectionMap(ThreadFactory f,
                            Receiver r,
                            InetAddress bind_addr,
                            InetAddress external_addr,
                            int srv_port,
                            int max_port,
                            long reaper_interval,
                            long conn_expire_time,
                            PortsManager pm) throws Exception {
        this.mapper = new Mapper(f,reaper_interval);
        this.receiver=r;
        this.bind_addr=bind_addr;               
        this.pm=pm;   
        this.conn_expire_time = conn_expire_time;
        this.srv_sock=createServerSocket(srv_port, max_port);      

        if(external_addr != null)
            local_addr=new IpAddress(external_addr, srv_sock.getLocalPort());
        else if(bind_addr != null)
            local_addr=new IpAddress(bind_addr, srv_sock.getLocalPort());
        else
            local_addr=new IpAddress(srv_sock.getLocalPort());
    }    
    
    public Address getLocalAddress() {       
        return local_addr;
    }

    /**
     * Calls the receiver callback. We do not serialize access to this method,
     * and it may be called concurrently by several Connection handler threads.
     * Therefore the receiver needs to be reentrant.
     */
    public void receive(Address sender, byte[] data, int offset, int length) {
        receiver.receive(sender, data, offset, length);
    }

    public void send(Address dest, byte[] data, int offset, int length) throws Exception {        
        if(dest == null) {
            if(log.isErrorEnabled())
                log.error("destination is null");
            return;
        }

        if(data == null) {
            log.warn("data is null; discarding packet");
            return;
        }      
        
        if(!running ) {
            if(log.isWarnEnabled())
                log.warn("connection table is not running, discarding message to " + dest);
            return;
        }

        if(dest.equals(local_addr)) {
            receive(local_addr, data, offset, length);
            return;
        }

        // 1. Try to obtain correct Connection (or create one if not yet existent)
        TCPConnection conn;
        try {
            conn=mapper.getConnection(dest);           
        }
        catch(Throwable ex) {
            throw new Exception("connection to " + dest + " could not be established", ex);
        }

        // 2. Send the message using that connection
        try {
            conn.send(data, offset, length);
        }
        catch(Exception ex) {            
            mapper.removeConnection(dest);
            throw ex;
        }
    }

    public synchronized void start() throws Exception {
        mapper.start();
        if(!running) {
            running=true;
            acceptor=mapper.getThreadFactory().newThread(thread_group, new ConnectionAcceptor(),"ConnectionTable.Acceptor");
            acceptor.start();   
        }               
    }

    public synchronized void stop() {
        if(running) {      
            running = false;
            if(pm != null) {
                pm.removePort(srv_sock.getLocalPort());
            }
            Util.close(srv_sock);
            if(acceptor != null)
                Util.interruptAndWaitToDie(acceptor);
            mapper.stop();
        }        
    }

    /**
     * Finds first available port starting at start_port and returns server
     * socket. Will not bind to port >end_port. Sets srv_port
     */
    protected ServerSocket createServerSocket(int start_port, int end_port) throws Exception {
        ServerSocket ret=null;

        while(true) {
            try {
                if(start_port > 0 && pm != null)
                    start_port=pm.getNextAvailablePort(start_port);
                if(bind_addr == null)
                    ret=new ServerSocket(start_port);
                else {
                    // changed (bela Sept 7 2007): we accept connections on all NICs
                    ret=new ServerSocket(start_port, 20, null);
                }
            }
            catch(BindException bind_ex) {
                if(start_port == end_port)
                    throw new BindException("No available port to bind to");
                if(bind_addr != null) {
                    NetworkInterface nic=NetworkInterface.getByInetAddress(bind_addr);
                    if(nic == null)
                        throw new BindException("bind_addr " + bind_addr
                                                + " is not a valid interface");
                }
                start_port++;
                continue;
            }
            catch(IOException io_ex) {
                if(log.isErrorEnabled())
                    log.error("exception is " + io_ex);
            }           
            break;
        }
        return ret;
    }
    
    private void setSocketParameters(Socket client_sock) throws SocketException {
        if(log.isTraceEnabled())
            log.trace("[" + local_addr
                      + "] accepted connection from "
                      + client_sock.getInetAddress()
                      + ":"
                      + client_sock.getPort());
        try {
            client_sock.setSendBufferSize(send_buf_size);
        }
        catch(IllegalArgumentException ex) {
            if(log.isErrorEnabled())
                log.error("exception setting send buffer size to " + send_buf_size + " bytes",
                          ex);
        }
        try {
            client_sock.setReceiveBufferSize(recv_buf_size);
        }
        catch(IllegalArgumentException ex) {
            if(log.isErrorEnabled())
                log.error("exception setting receive buffer size to " + send_buf_size
                          + " bytes", ex);
        }

        client_sock.setKeepAlive(true);
        client_sock.setTcpNoDelay(tcp_nodelay);
        if(linger > 0)
            client_sock.setSoLinger(true, linger);
        else
            client_sock.setSoLinger(false, -1);
    }

    /** Used for message reception. */
    public interface Receiver {
        void receive(Address sender, byte[] data, int offset, int length);
    }

    private class ConnectionAcceptor implements Runnable {

        /**
         * Acceptor thread. Continuously accept new connections. Create a new
         * thread for each new connection and put it in conns. When the thread
         * should stop, it is interrupted by the thread creator.
         */
        public void run() {
            while(running) {
                TCPConnection conn=null;
                try {
                    Socket client_sock=srv_sock.accept();                    
                    setSocketParameters(client_sock);

                    // create new thread and add to conn table
                    conn=new TCPConnection(receiver, client_sock, null);
                    Address peer_addr = conn.getPeerAddress();                    
                    
                    mapper.getLock().lock();
                    try{
                        Connection tmp=mapper.getConnection(peer_addr);
                        //<NAME>, 5th, 2007
                        //we might have a connection to peer but is that 
                        //connection still open? 
                        boolean connectionOpen=tmp != null && tmp.isOpen();
                        if(connectionOpen) {
                            if(peer_addr.compareTo(local_addr) > 0) {
                                if(log.isTraceEnabled())
                                    log.trace("peer's address (" + peer_addr
                                              + ") is greater than our local address ("
                                              + local_addr
                                              + "), replacing our existing connection");
                                // peer's address is greater, add peer's connection to ConnectionTable, destroy existing connection
                                mapper.removeConnection(peer_addr);
                                mapper.addConnection(peer_addr, conn);                                      
                            }
                            else {
                                if(log.isTraceEnabled())
                                    log.trace("peer's address (" + peer_addr
                                              + ") is smaller than our local address ("
                                              + local_addr
                                              + "), rejecting peer connection request");
                                conn.close();                                
                                continue;
                            }
                        }
                        else {
                            mapper.addConnection(peer_addr, conn);                                         
                        }
                    }
                    finally{
                        mapper.getLock().unlock();
                    }

                    conn.start(mapper.getThreadFactory()); // starts handler thread on this socket
                }<commit_msg>simplify connection setup<commit_after>package org.jgroups.blocks;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.BindException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.NetworkInterface;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.SocketException;
import java.util.Collection;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jgroups.Address;
import org.jgroups.Version;
import org.jgroups.stack.IpAddress;
import org.jgroups.util.PortsManager;
import org.jgroups.util.ThreadFactory;
import org.jgroups.util.Util;

public class TCPConnectionMap{

    private final Mapper mapper;
    private final InetAddress bind_addr;
    private final Address local_addr; // bind_addr + port of srv_sock
    private final ThreadGroup thread_group=new ThreadGroup(Util.getGlobalThreadGroup(),"ConnectionTable");     
    private final ServerSocket srv_sock;
    private final Receiver receiver;
    private final long conn_expire_time;
    private final Log log=LogFactory.getLog(getClass());
    private int recv_buf_size=120000;
    private int send_buf_size=60000;
    private int send_queue_size = 0;
    private int sock_conn_timeout=1000; // max time in millis to wait for Socket.connect() to return    
    private boolean tcp_nodelay=false;
    private int linger=-1;    
    private Thread acceptor;
    private PortsManager pm=null;       

    private volatile boolean running = false;    

    public TCPConnectionMap(ThreadFactory f,
                            Receiver r,
                            InetAddress bind_addr,
                            InetAddress external_addr,
                            int srv_port,
                            int max_port,
                            PortsManager pm) throws Exception {
        this(f,r,bind_addr,external_addr,srv_port,max_port,0,0,pm);
    }

    public TCPConnectionMap(ThreadFactory f,
                            Receiver r,
                            InetAddress bind_addr,
                            InetAddress external_addr,
                            int srv_port,
                            int max_port,
                            long reaper_interval,
                            long conn_expire_time,
                            PortsManager pm) throws Exception {
        this.mapper = new Mapper(f,reaper_interval);
        this.receiver=r;
        this.bind_addr=bind_addr;               
        this.pm=pm;   
        this.conn_expire_time = conn_expire_time;
        this.srv_sock=createServerSocket(srv_port, max_port);      

        if(external_addr != null)
            local_addr=new IpAddress(external_addr, srv_sock.getLocalPort());
        else if(bind_addr != null)
            local_addr=new IpAddress(bind_addr, srv_sock.getLocalPort());
        else
            local_addr=new IpAddress(srv_sock.getLocalPort());
    }    
    
    public Address getLocalAddress() {       
        return local_addr;
    }

    /**
     * Calls the receiver callback. We do not serialize access to this method,
     * and it may be called concurrently by several Connection handler threads.
     * Therefore the receiver needs to be reentrant.
     */
    public void receive(Address sender, byte[] data, int offset, int length) {
        receiver.receive(sender, data, offset, length);
    }

    public void send(Address dest, byte[] data, int offset, int length) throws Exception {        
        if(dest == null) {
            if(log.isErrorEnabled())
                log.error("destination is null");
            return;
        }

        if(data == null) {
            log.warn("data is null; discarding packet");
            return;
        }      
        
        if(!running ) {
            if(log.isWarnEnabled())
                log.warn("connection table is not running, discarding message to " + dest);
            return;
        }

        if(dest.equals(local_addr)) {
            receive(local_addr, data, offset, length);
            return;
        }

        // 1. Try to obtain correct Connection (or create one if not yet existent)
        TCPConnection conn;
        try {
            conn=mapper.getConnection(dest);           
        }
        catch(Throwable ex) {
            throw new Exception("connection to " + dest + " could not be established", ex);
        }

        // 2. Send the message using that connection
        try {
            conn.send(data, offset, length);
        }
        catch(Exception ex) {            
            mapper.removeConnection(dest);
            throw ex;
        }
    }

    public synchronized void start() throws Exception {
        mapper.start();
        if(!running) {
            running=true;
            acceptor=mapper.getThreadFactory().newThread(thread_group, new ConnectionAcceptor(),"ConnectionTable.Acceptor");
            acceptor.start();   
        }               
    }

    public synchronized void stop() {
        if(running) {      
            running = false;
            if(pm != null) {
                pm.removePort(srv_sock.getLocalPort());
            }
            Util.close(srv_sock);
            if(acceptor != null)
                Util.interruptAndWaitToDie(acceptor);
            mapper.stop();
        }        
    }

    /**
     * Finds first available port starting at start_port and returns server
     * socket. Will not bind to port >end_port. Sets srv_port
     */
    protected ServerSocket createServerSocket(int start_port, int end_port) throws Exception {
        ServerSocket ret=null;

        while(true) {
            try {
                if(start_port > 0 && pm != null)
                    start_port=pm.getNextAvailablePort(start_port);
                if(bind_addr == null)
                    ret=new ServerSocket(start_port);
                else {
                    // changed (bela Sept 7 2007): we accept connections on all NICs
                    ret=new ServerSocket(start_port, 20, null);
                }
            }
            catch(BindException bind_ex) {
                if(start_port == end_port)
                    throw new BindException("No available port to bind to");
                if(bind_addr != null) {
                    NetworkInterface nic=NetworkInterface.getByInetAddress(bind_addr);
                    if(nic == null)
                        throw new BindException("bind_addr " + bind_addr
                                                + " is not a valid interface");
                }
                start_port++;
                continue;
            }
            catch(IOException io_ex) {
                if(log.isErrorEnabled())
                    log.error("exception is " + io_ex);
            }           
            break;
        }
        return ret;
    }
    
    private void setSocketParameters(Socket client_sock) throws SocketException {
        if(log.isTraceEnabled())
            log.trace("[" + local_addr
                      + "] accepted connection from "
                      + client_sock.getInetAddress()
                      + ":"
                      + client_sock.getPort());
        try {
            client_sock.setSendBufferSize(send_buf_size);
        }
        catch(IllegalArgumentException ex) {
            if(log.isErrorEnabled())
                log.error("exception setting send buffer size to " + send_buf_size + " bytes",
                          ex);
        }
        try {
            client_sock.setReceiveBufferSize(recv_buf_size);
        }
        catch(IllegalArgumentException ex) {
            if(log.isErrorEnabled())
                log.error("exception setting receive buffer size to " + send_buf_size
                          + " bytes", ex);
        }

        client_sock.setKeepAlive(true);
        client_sock.setTcpNoDelay(tcp_nodelay);
        if(linger > 0)
            client_sock.setSoLinger(true, linger);
        else
            client_sock.setSoLinger(false, -1);
    }

    /** Used for message reception. */
    public interface Receiver {
        void receive(Address sender, byte[] data, int offset, int length);
    }

    private class ConnectionAcceptor implements Runnable {

        /**
         * Acceptor thread. Continuously accept new connections. Create a new
         * thread for each new connection and put it in conns. When the thread
         * should stop, it is interrupted by the thread creator.
         */
        public void run() {
            while(running) {
                TCPConnection conn=null;
                try {
                    Socket client_sock=srv_sock.accept();                    
                    setSocketParameters(client_sock);

                    // create new thread and add to conn table
                    conn=new TCPConnection(receiver, client_sock, null);
                    Address peer_addr = conn.getPeerAddress();                    
                    
                    mapper.getLock().lock();
                    try{
                        Connection currentConnection=mapper.getConnection(peer_addr);                       
                        boolean currentConnectionOpen=currentConnection != null && currentConnection.isOpen();
                        boolean replaceWithNewConnection = false;
                        if(currentConnectionOpen) {
                            replaceWithNewConnection = peer_addr.compareTo(local_addr) > 0;                               
                        }
                        if(!currentConnectionOpen || replaceWithNewConnection){
                            mapper.removeConnection(peer_addr);
                            mapper.addConnection(peer_addr, conn);
                            conn.start(mapper.getThreadFactory()); // starts handler thread on this socket
                        }else{
                            conn.close();
                        }
                    }
                    finally{
                        mapper.getLock().unlock();
                    }                   
                }
{'fraction_non_alphanumeric': 0.07787949173384343, 'fraction_numerical': 0.0031425058068042084, 'mean_word_length': 1.4890047608252097, 'pattern_counts': {'":': 2, '<': 4, '<?xml version=': 0, '>': 12, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '924283', 'n_tokens_mistral': 5488, 'n_tokens_neox': 5230, 'n_words': 1292}