<commit_before>/*
 * Copyright (c) 2009-2011 by <NAME>,
 *               Zuse Institute Berlin
 *
 * Licensed under the BSD License, see LICENSE file for details.
 *
 */

package org.xtreemfs.foundation.pbrpc.server;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedByInterruptException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

import org.xtreemfs.foundation.LifeCycleThread;
import org.xtreemfs.foundation.SSLOptions;
import org.xtreemfs.foundation.buffer.BufferPool;
import org.xtreemfs.foundation.buffer.ReusableBuffer;
import org.xtreemfs.foundation.logging.Logging;
import org.xtreemfs.foundation.logging.Logging.Category;
import org.xtreemfs.foundation.pbrpc.channels.ChannelIO;
import org.xtreemfs.foundation.pbrpc.channels.SSLChannelIO;
import org.xtreemfs.foundation.pbrpc.channels.SSLHandshakeOnlyChannelIO;
import org.xtreemfs.foundation.util.OutputUtils;

/**
 * 
 * @author bjko
 */
public class RPCNIOSocketServer extends LifeCycleThread implements RPCServerInterface {
    
    /**
     * Maximum number of record fragments supported.
     */
    public static final int                MAX_FRAGMENTS     = 1;
    
    /**
     * Maximum fragment size to accept. If the size is larger, the connection is
     * closed.
     */
    public static final int                MAX_FRAGMENT_SIZE = 1024 * 1024 * 32;
    
    /**
     * the server socket
     */
    private final ServerSocketChannel      socket;
    
    /**
     * Selector for server socket
     */
    private final Selector                 selector;
    
    /**
     * If set to true thei main loop will exit upon next invocation
     */
    private volatile boolean               quit;
    
    /**
     * The receiver that gets all incoming requests.
     */
    private volatile RPCServerRequestListener receiver;
    
    /**
     * sslOptions if SSL is enabled, null otherwise
     */
    private final SSLOptions               sslOptions;
    
    /**
     * Connection count
     */
    private final AtomicInteger            numConnections;
    
    /**
     * Number of requests received but not answered
     */
    private long                           pendingRequests;
    
    /**
     * Port on which the server listens for incoming connections.
     */
    private final int                      bindPort;
    
    private final List<RPCNIOSocketServerConnection>   connections;
    
    /**
     * maximum number of pending client requests to allow
     */
    public static int                      MAX_CLIENT_QUEUE  = 20000;
    
    /**
     * if the Q was full we need at least CLIENT_Q_THR spaces before we start
     * reading from the client again. This is to prevent it from oscillating
     */
    public static int                      CLIENT_Q_THR      = 5000;

    // Connections with pending writes.
    private final ConcurrentLinkedQueue<RPCNIOSocketServerConnection>  writeableConnections;
    
    public RPCNIOSocketServer(int bindPort, InetAddress bindAddr, RPCServerRequestListener rl,
        SSLOptions sslOptions) throws IOException {
        this(bindPort, bindAddr, rl, sslOptions, 256 * 1024);
    }
    
    public RPCNIOSocketServer(int bindPort, InetAddress bindAddr, RPCServerRequestListener rl,
        SSLOptions sslOptions, int receiveBufferSize) throws IOException {
        super("PBRPCSrv@" + bindPort);
        
        // open server socket
        socket = ServerSocketChannel.open();
        socket.configureBlocking(false);
        
        if (receiveBufferSize != -1) {
            socket.socket().setReceiveBufferSize(receiveBufferSize);
            if (socket.socket().getReceiveBufferSize() != receiveBufferSize) {
                Logging.logMessage(Logging.LEVEL_WARN, Category.net, this,
                    "could not set socket receive buffer size to " + receiveBufferSize
                        + ", using default size of " + socket.socket().getReceiveBufferSize());
            }
        } else {
            socket.socket().setReceiveBufferSize(256 * 1024);
        }
        
        socket.socket().setReuseAddress(true);
        socket.socket().bind(
            bindAddr == null ? new InetSocketAddress(bindPort) : new InetSocketAddress(bindAddr, bindPort));
        this.bindPort = bindPort;<commit_msg>foundation: RPCNIOSocketServer no longer fails if the socket's receive buffer size could not be verified<commit_after>/*
 * Copyright (c) 2009-2011 by <NAME>,
 *               Zuse Institute Berlin
 *
 * Licensed under the BSD License, see LICENSE file for details.
 *
 */

package org.xtreemfs.foundation.pbrpc.server;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedByInterruptException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

import org.xtreemfs.foundation.LifeCycleThread;
import org.xtreemfs.foundation.SSLOptions;
import org.xtreemfs.foundation.buffer.BufferPool;
import org.xtreemfs.foundation.buffer.ReusableBuffer;
import org.xtreemfs.foundation.logging.Logging;
import org.xtreemfs.foundation.logging.Logging.Category;
import org.xtreemfs.foundation.pbrpc.channels.ChannelIO;
import org.xtreemfs.foundation.pbrpc.channels.SSLChannelIO;
import org.xtreemfs.foundation.pbrpc.channels.SSLHandshakeOnlyChannelIO;
import org.xtreemfs.foundation.util.OutputUtils;

/**
 * 
 * @author bjko
 */
public class RPCNIOSocketServer extends LifeCycleThread implements RPCServerInterface {
    
    /**
     * Maximum number of record fragments supported.
     */
    public static final int                MAX_FRAGMENTS     = 1;
    
    /**
     * Maximum fragment size to accept. If the size is larger, the connection is
     * closed.
     */
    public static final int                MAX_FRAGMENT_SIZE = 1024 * 1024 * 32;
    
    /**
     * the server socket
     */
    private final ServerSocketChannel      socket;
    
    /**
     * Selector for server socket
     */
    private final Selector                 selector;
    
    /**
     * If set to true thei main loop will exit upon next invocation
     */
    private volatile boolean               quit;
    
    /**
     * The receiver that gets all incoming requests.
     */
    private volatile RPCServerRequestListener receiver;
    
    /**
     * sslOptions if SSL is enabled, null otherwise
     */
    private final SSLOptions               sslOptions;
    
    /**
     * Connection count
     */
    private final AtomicInteger            numConnections;
    
    /**
     * Number of requests received but not answered
     */
    private long                           pendingRequests;
    
    /**
     * Port on which the server listens for incoming connections.
     */
    private final int                      bindPort;
    
    private final List<RPCNIOSocketServerConnection>   connections;
    
    /**
     * maximum number of pending client requests to allow
     */
    public static int                      MAX_CLIENT_QUEUE  = 20000;
    
    /**
     * if the Q was full we need at least CLIENT_Q_THR spaces before we start
     * reading from the client again. This is to prevent it from oscillating
     */
    public static int                      CLIENT_Q_THR      = 5000;

    // Connections with pending writes.
    private final ConcurrentLinkedQueue<RPCNIOSocketServerConnection>  writeableConnections;
    
    public RPCNIOSocketServer(int bindPort, InetAddress bindAddr, RPCServerRequestListener rl,
        SSLOptions sslOptions) throws IOException {
        this(bindPort, bindAddr, rl, sslOptions, 256 * 1024);
    }
    
    public RPCNIOSocketServer(int bindPort, InetAddress bindAddr, RPCServerRequestListener rl,
        SSLOptions sslOptions, int receiveBufferSize) throws IOException {
        super("PBRPCSrv@" + bindPort);
        
        // open server socket
        socket = ServerSocketChannel.open();
        socket.configureBlocking(false);
        
        if (receiveBufferSize != -1) {
            socket.socket().setReceiveBufferSize(receiveBufferSize);
            try {
                if (socket.socket().getReceiveBufferSize() != receiveBufferSize) {
                    Logging.logMessage(Logging.LEVEL_WARN, Category.net, this,
                            "could not set socket receive buffer size to " + receiveBufferSize
                                    + ", using default size of " + socket.socket().getReceiveBufferSize());
                }
            } catch (SocketException exc) {
                Logging.logMessage(Logging.LEVEL_WARN, this,
                        "could not check whether receive buffer size was successfully set to %d bytes", receiveBufferSize);
            }
        } else {
            socket.socket().setReceiveBufferSize(256 * 1024);
        }
        
        socket.socket().setReuseAddress(true);
        socket.socket().bind(
            bindAddr == null ? new InetSocketAddress(bindPort) : new InetSocketAddress(bindAddr, bindPort));
        this.bindPort = bindPort;
{'fraction_non_alphanumeric': 0.07858073569209181, 'fraction_numerical': 0.008619825598877418, 'mean_word_length': 2.4730247128437175, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7062724', 'n_tokens_mistral': 2794, 'n_tokens_neox': 2654, 'n_words': 699}