path: root/indra/llmessage/lliosocket.cpp

/**
 * @file lliosocket.cpp
 * @author Phoenix
 * @date 2005-07-31
 * @brief Sockets declarations for use with the io pipes
 *
 * $LicenseInfo:firstyear=2005&license=viewerlgpl$
 * Second Life Viewer Source Code
 * Copyright (C) 2010, Linden Research, Inc.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation;
 * version 2.1 of the License only.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
 *
 * Linden Research, Inc., 945 Battery Street, San Francisco, CA  94111  USA
 * $/LicenseInfo$
 */

#include "linden_common.h"
#include "lliosocket.h"

#include "llapr.h"

#include "llbuffer.h"
#include "llfasttimer.h"
#include "llhost.h"
#include "llpumpio.h"

//
// constants
//

const std::string CONTEXT_REMOTE_HOST("remote-host");
const std::string CONTEXT_REMOTE_PORT("remote-port");

static const S32 LL_DEFAULT_LISTEN_BACKLOG = 10;
static const S32 LL_SEND_BUFFER_SIZE = 40000;
static const S32 LL_RECV_BUFFER_SIZE = 40000;
//static const U16 LL_PORT_DISCOVERY_RANGE_MIN = 13000;
//static const U16 LL_PORT_DISCOVERY_RANGE_MAX = 13050;

//
// local methods
//

bool is_addr_in_use(apr_status_t status)
{
#if LL_WINDOWS
    return (WSAEADDRINUSE == APR_TO_OS_ERROR(status));
#else
    return (EADDRINUSE == APR_TO_OS_ERROR(status));
#endif
}

#if ! LL_WINDOWS
// Define this to see the actual file descriptors being tossed around.
#define LL_DEBUG_SOCKET_FILE_DESCRIPTORS 1
#if LL_DEBUG_SOCKET_FILE_DESCRIPTORS
#include "apr_portable.h"
#endif
#endif


// Quick function
void ll_debug_socket(const char* msg, apr_socket_t* apr_sock)
{
#if LL_DEBUG_SOCKET_FILE_DESCRIPTORS
    if(!apr_sock)
    {
        LL_DEBUGS("Socket") << "Socket -- " << (msg?msg:"") << ": no socket." << LL_ENDL;
        return;
    }
    // *TODO: Why doesn't this work?
    //apr_os_sock_t os_sock;
    int os_sock;
    if(APR_SUCCESS == apr_os_sock_get(&os_sock, apr_sock))
    {
        LL_DEBUGS("Socket") << "Socket -- " << (msg?msg:"") << " on fd " << os_sock
            << " at " << apr_sock << LL_ENDL;
    }
    else
    {
        LL_DEBUGS("Socket") << "Socket -- " << (msg?msg:"") << " no fd "
            << " at " << apr_sock << LL_ENDL;
    }
#endif
}

///
/// LLSocket
///

// static
LLSocket::ptr_t LLSocket::create(apr_pool_t* pool, EType type, U16 port, const char *hostname)
{
    LLSocket::ptr_t rv;
    apr_socket_t* socket = NULL;
    apr_pool_t* new_pool = NULL;
    apr_status_t status = APR_EGENERAL;

    // create a pool for the socket
    status = apr_pool_create(&new_pool, pool);
    if(ll_apr_warn_status(status))
    {
        if(new_pool) apr_pool_destroy(new_pool);
        return rv;
    }

    if(STREAM_TCP == type)
    {
        status = apr_socket_create(
            &socket,
            APR_INET,
            SOCK_STREAM,
            APR_PROTO_TCP,
            new_pool);
    }
    else if(DATAGRAM_UDP == type)
    {
        status = apr_socket_create(
            &socket,
            APR_INET,
            SOCK_DGRAM,
            APR_PROTO_UDP,
            new_pool);
    }
    else
    {
        if(new_pool) apr_pool_destroy(new_pool);
        return rv;
    }
    if(ll_apr_warn_status(status))
    {
        if(new_pool) apr_pool_destroy(new_pool);
        return rv;
    }
    // At this point, the new LLSocket instance takes ownership of new_pool,
    // which is why no early return below this call explicitly destroys it: it
    // is instead cleaned up by ~LLSocket().
    rv = ptr_t(new LLSocket(socket, new_pool));
    if(port > 0)
    {
        apr_sockaddr_t* sa = NULL;
        status = apr_sockaddr_info_get(
            &sa,
            hostname,
            APR_UNSPEC,
            port,
            0,
            new_pool);
        if(ll_apr_warn_status(status))
        {
            rv.reset();
            return rv;
        }
        // This allows us to reuse the address on quick down/up. This
        // is unlikely to create problems.
        ll_apr_warn_status(apr_socket_opt_set(socket, APR_SO_REUSEADDR, 1));
        status = apr_socket_bind(socket, sa);
        if(ll_apr_warn_status(status))
        {
            rv.reset();
            return rv;
        }
        LL_DEBUGS() << "Bound " << ((DATAGRAM_UDP == type) ? "udp" : "tcp")
                 << " socket to port: " << sa->port << LL_ENDL;
        if(STREAM_TCP == type)
        {
            // If it's a stream based socket, we need to tell the OS
            // to keep a queue of incoming connections for ACCEPT.
            LL_DEBUGS() << "Setting listen state for socket." << LL_ENDL;
            status = apr_socket_listen(
                socket,
                LL_DEFAULT_LISTEN_BACKLOG);
            if(ll_apr_warn_status(status))
            {
                rv.reset();
                return rv;
            }
        }
    }
    else // port <= 0
    {
        // we need to indicate that we have an ephemeral port if the
        // previous calls were successful. It will
        port = PORT_EPHEMERAL;
    }
    rv->mPort = port;
    rv->setNonBlocking();
    return rv;
}

// static
LLSocket::ptr_t LLSocket::create(apr_socket_t* socket, apr_pool_t* pool)
{
    LLSocket::ptr_t rv;
    if(!socket)
    {
        return rv;
    }
    rv = ptr_t(new LLSocket(socket, pool));
    rv->mPort = PORT_EPHEMERAL;
    rv->setNonBlocking();
    return rv;
}


bool LLSocket::blockingConnect(const LLHost& host)
{
    if(!mSocket) return false;
    apr_sockaddr_t* sa = NULL;
    std::string ip_address;
    ip_address = host.getIPString();
    if(ll_apr_warn_status(apr_sockaddr_info_get(
        &sa,
        ip_address.c_str(),
        APR_UNSPEC,
        host.getPort(),
        0,
        mPool)))
    {
        return false;
    }
    setBlocking(1000);
    ll_debug_socket("Blocking connect", mSocket);
    if(ll_apr_warn_status(apr_socket_connect(mSocket, sa))) return false;
    setNonBlocking();
    return true;
}

LLSocket::LLSocket(apr_socket_t* socket, apr_pool_t* pool) :
    mSocket(socket),
    mPool(pool),
    mPort(PORT_INVALID)
{
    ll_debug_socket("Constructing wholely formed socket", mSocket);
}

LLSocket::~LLSocket()
{
    // *FIX: clean up memory we are holding.
    if(mSocket)
    {
        ll_debug_socket("Destroying socket", mSocket);
        apr_socket_close(mSocket);
        mSocket = NULL;
    }
    if(mPool)
    {
        apr_pool_destroy(mPool);
    }
}

// See http://dev.ariel-networks.com/apr/apr-tutorial/html/apr-tutorial-13.html#ss13.4
// for an explanation of how to get non-blocking sockets and timeouts with
// consistent behavior across platforms.

void LLSocket::setBlocking(S32 timeout)
{
    // set up the socket options
    ll_apr_warn_status(apr_socket_timeout_set(mSocket, timeout)); // Sets both receive and send timeout SO_RCVTIMEO, SO_SNDTIMEO
    ll_apr_warn_status(apr_socket_opt_set(mSocket, APR_SO_NONBLOCK, 0));
    ll_apr_warn_status(apr_socket_opt_set(mSocket, APR_SO_SNDBUF, LL_SEND_BUFFER_SIZE));
    ll_apr_warn_status(apr_socket_opt_set(mSocket, APR_SO_RCVBUF, LL_RECV_BUFFER_SIZE));

}

void LLSocket::setNonBlocking()
{
    // set up the socket options
    ll_apr_warn_status(apr_socket_timeout_set(mSocket, 0));
    ll_apr_warn_status(apr_socket_opt_set(mSocket, APR_SO_NONBLOCK, 1));
    ll_apr_warn_status(apr_socket_opt_set(mSocket, APR_SO_SNDBUF, LL_SEND_BUFFER_SIZE));
    ll_apr_warn_status(apr_socket_opt_set(mSocket, APR_SO_RCVBUF, LL_RECV_BUFFER_SIZE));

}

///
/// LLIOSocketReader
///

LLIOSocketReader::LLIOSocketReader(LLSocket::ptr_t socket) :
    mSource(socket),
    mInitialized(false)
{
}

LLIOSocketReader::~LLIOSocketReader()
{
    //LL_DEBUGS() << "Destroying LLIOSocketReader" << LL_ENDL;
}

// virtual
LLIOPipe::EStatus LLIOSocketReader::process_impl(
    const LLChannelDescriptors& channels,
    buffer_ptr_t& buffer,
    bool& eos,
    LLSD& context,
    LLPumpIO* pump)
{
    LL_PROFILE_ZONE_SCOPED;
    PUMP_DEBUG;
    if(!mSource) return STATUS_PRECONDITION_NOT_MET;
    if(!mInitialized)
    {
        PUMP_DEBUG;
        // Since the read will not block, it's ok to initialize and
        // attempt to read off the descriptor immediately.
        mInitialized = true;
        if(pump)
        {
            PUMP_DEBUG;
            LL_DEBUGS() << "Initializing poll descriptor for LLIOSocketReader."
                     << LL_ENDL;
            apr_pollfd_t poll_fd;
            poll_fd.p = NULL;
            poll_fd.desc_type = APR_POLL_SOCKET;
            poll_fd.reqevents = APR_POLLIN;
            poll_fd.rtnevents = 0x0;
            poll_fd.desc.s = mSource->getSocket();
            poll_fd.client_data = NULL;
            pump->setConditional(this, &poll_fd);
        }
    }
    //if(!buffer)
    //{
    //  buffer = new LLBufferArray;
    //}
    PUMP_DEBUG;
    const apr_size_t READ_BUFFER_SIZE = 1024;
    char read_buf[READ_BUFFER_SIZE]; /*Flawfinder: ignore*/
    apr_size_t len;
    apr_status_t status = APR_SUCCESS;
    do
    {
        PUMP_DEBUG;
        len = READ_BUFFER_SIZE;
        status = apr_socket_recv(mSource->getSocket(), read_buf, &len);
        buffer->append(channels.out(), (U8*)read_buf, static_cast<S32>(len));
    } while((APR_SUCCESS == status) && (READ_BUFFER_SIZE == len));
    LL_DEBUGS() << "socket read status: " << status << LL_ENDL;
    LLIOPipe::EStatus rv = STATUS_OK;

    PUMP_DEBUG;
    // *FIX: Also need to check for broken pipe
    if(APR_STATUS_IS_EOF(status))
    {
        // *FIX: Should we shut down the socket read?
        if(pump)
        {
            pump->setConditional(this, NULL);
        }
        rv = STATUS_DONE;
        eos = true;
    }
    else if(APR_STATUS_IS_EAGAIN(status))
    {
/*Commented out by Aura 9-9-8 for DEV-19961.
        // everything is fine, but we can terminate this process pump.

        rv = STATUS_BREAK;
*/
    }
    else
    {
        if(ll_apr_warn_status(status))
        {
            rv = STATUS_ERROR;
        }
    }
    PUMP_DEBUG;
    return rv;
}

///
/// LLIOSocketWriter
///

LLIOSocketWriter::LLIOSocketWriter(LLSocket::ptr_t socket) :
    mDestination(socket),
    mLastWritten(NULL),
    mInitialized(false)
{
}

LLIOSocketWriter::~LLIOSocketWriter()
{
    //LL_DEBUGS() << "Destroying LLIOSocketWriter" << LL_ENDL;
}

// virtual
LLIOPipe::EStatus LLIOSocketWriter::process_impl(
    const LLChannelDescriptors& channels,
    buffer_ptr_t& buffer,
    bool& eos,
    LLSD& context,
    LLPumpIO* pump)
{
    LL_PROFILE_ZONE_SCOPED;
    PUMP_DEBUG;
    if(!mDestination) return STATUS_PRECONDITION_NOT_MET;
    if(!mInitialized)
    {
        PUMP_DEBUG;
        // Since the write will not block, it's ok to initialize and
        // attempt to write immediately.
        mInitialized = true;
        if(pump)
        {
            PUMP_DEBUG;
            LL_DEBUGS() << "Initializing poll descriptor for LLIOSocketWriter."
                     << LL_ENDL;
            apr_pollfd_t poll_fd;
            poll_fd.p = NULL;
            poll_fd.desc_type = APR_POLL_SOCKET;
            poll_fd.reqevents = APR_POLLOUT;
            poll_fd.rtnevents = 0x0;
            poll_fd.desc.s = mDestination->getSocket();
            poll_fd.client_data = NULL;
            pump->setConditional(this, &poll_fd);
        }
    }

    PUMP_DEBUG;
    // *FIX: Some sort of writev implementation would be much more
    // efficient - not only because writev() is better, but also
    // because we won't have to do as much work to find the start
    // address.
    buffer->lock();
    LLBufferArray::segment_iterator_t it;
    LLBufferArray::segment_iterator_t end = buffer->endSegment();
    LLSegment segment;
    it = buffer->constructSegmentAfter(mLastWritten, segment);
    /*
    if(NULL == mLastWritten)
    {
        it = buffer->beginSegment();
        segment = (*it);
    }
    else
    {
        it = buffer->getSegment(mLastWritten);
        segment = (*it);
        S32 size = segment.size();
        U8* data = segment.data();
        if((data + size) == mLastWritten)
        {
            ++it;
            segment = (*it);
        }
        else
        {
            // *FIX: check the math on this one
            segment = LLSegment(
                (*it).getChannelMask(),
                mLastWritten + 1,
                size - (mLastWritten - data));
        }
    }
    */

    PUMP_DEBUG;
    apr_size_t len;
    bool done = false;
    apr_status_t status = APR_SUCCESS;
    while(it != end)
    {

        PUMP_DEBUG;
        if((*it).isOnChannel(channels.in()))
        {
            PUMP_DEBUG;
            len = (apr_size_t)segment.size();
            status = apr_socket_send(
                mDestination->getSocket(),
                (const char*)segment.data(),
                &len);
            // We sometimes get a 'non-blocking socket operation could not be
            // completed immediately' error from apr_socket_send.  In this
            // case we break and the data will be sent the next time the chain
            // is pumped.
            if(APR_STATUS_IS_EAGAIN(status))
            {
                ll_apr_warn_status(status);
                break;
            }

            mLastWritten = segment.data() + len - 1;

            PUMP_DEBUG;
            if((S32)len < segment.size())
            {
                break;
            }

        }

        ++it;
        if(it != end)
        {
            segment = (*it);
        }
        else
        {
            done = true;
        }

    }
    buffer->unlock();

    PUMP_DEBUG;
    if(done && eos)
    {
        return STATUS_DONE;
    }
    return STATUS_OK;
}


///
/// LLIOServerSocket
///

LLIOServerSocket::LLIOServerSocket(
    apr_pool_t* pool,
    LLIOServerSocket::socket_t listener,
    factory_t factory) :
    mPool(pool),
    mListenSocket(listener),
    mReactor(factory),
    mInitialized(false),
    mResponseTimeout(DEFAULT_CHAIN_EXPIRY_SECS)
{
}

LLIOServerSocket::~LLIOServerSocket()
{
    //LL_DEBUGS() << "Destroying LLIOServerSocket" << LL_ENDL;
}

void LLIOServerSocket::setResponseTimeout(F32 timeout_secs)
{
    mResponseTimeout = timeout_secs;
}

// virtual
LLIOPipe::EStatus LLIOServerSocket::process_impl(
    const LLChannelDescriptors& channels,
    buffer_ptr_t& buffer,
    bool& eos,
    LLSD& context,
    LLPumpIO* pump)
{
    LL_PROFILE_ZONE_SCOPED;
    PUMP_DEBUG;
    if(!pump)
    {
        LL_WARNS() << "Need a pump for server socket." << LL_ENDL;
        return STATUS_ERROR;
    }
    if(!mInitialized)
    {
        PUMP_DEBUG;
        // This segment sets up the pump so that we do not call
        // process again until we have an incoming read, aka connect()
        // from a remote host.
        LL_DEBUGS() << "Initializing poll descriptor for LLIOServerSocket."
                 << LL_ENDL;
        apr_pollfd_t poll_fd;
        poll_fd.p = NULL;
        poll_fd.desc_type = APR_POLL_SOCKET;
        poll_fd.reqevents = APR_POLLIN;
        poll_fd.rtnevents = 0x0;
        poll_fd.desc.s = mListenSocket->getSocket();
        poll_fd.client_data = NULL;
        pump->setConditional(this, &poll_fd);
        mInitialized = true;
        return STATUS_OK;
    }

    // we are initialized, and told to process, so we must have a
    // socket waiting for a connection.
    LL_DEBUGS() << "accepting socket" << LL_ENDL;

    PUMP_DEBUG;
    apr_pool_t* new_pool = NULL;
    apr_status_t status = apr_pool_create(&new_pool, mPool);
    if(ll_apr_warn_status(status))
    {
        if(new_pool)
        {
            apr_pool_destroy(new_pool);
        }
        return STATUS_ERROR;
    }

    apr_socket_t* socket = NULL;
    status = apr_socket_accept(
        &socket,
        mListenSocket->getSocket(),
        new_pool);
    LLSocket::ptr_t llsocket(LLSocket::create(socket, new_pool));
    //EStatus rv = STATUS_ERROR;
    if(llsocket)
    {
        PUMP_DEBUG;

        apr_sockaddr_t* remote_addr;
        apr_socket_addr_get(&remote_addr, APR_REMOTE, socket);

        char* remote_host_string;
        apr_sockaddr_ip_get(&remote_host_string, remote_addr);

        LLSD context;
        context[CONTEXT_REMOTE_HOST] = remote_host_string;
        context[CONTEXT_REMOTE_PORT] = remote_addr->port;

        LLPumpIO::chain_t chain;
        chain.push_back(LLIOPipe::ptr_t(new LLIOSocketReader(llsocket)));
        if(mReactor->build(chain, context))
        {
            chain.push_back(LLIOPipe::ptr_t(new LLIOSocketWriter(llsocket)));
            pump->addChain(chain, mResponseTimeout);
            status = STATUS_OK;
        }
        else
        {
            LL_WARNS() << "Unable to build reactor to socket." << LL_ENDL;
        }
    }
    else
    {
        LL_WARNS() << "Unable to create linden socket." << LL_ENDL;
    }

    PUMP_DEBUG;
    // This needs to always return success, lest it get removed from
    // the pump.
    return STATUS_OK;
}


#if 0
LLIODataSocket::LLIODataSocket(
    U16 suggested_port,
    U16 start_discovery_port,
    apr_pool_t* pool) :
    mSocket(NULL)
{
    if(!pool || (PORT_INVALID == suggested_port)) return;
    if(ll_apr_warn_status(apr_socket_create(&mSocket, APR_INET, SOCK_DGRAM, APR_PROTO_UDP, pool))) return;
    apr_sockaddr_t* sa = NULL;
    if(ll_apr_warn_status(apr_sockaddr_info_get(&sa, APR_ANYADDR, APR_UNSPEC, suggested_port, 0, pool))) return;
    apr_status_t status = apr_socket_bind(mSocket, sa);
    if((start_discovery_port > 0) && is_addr_in_use(status))
    {
        const U16 MAX_ATTEMPT_PORTS = 50;
        for(U16 attempt_port = start_discovery_port;
            attempt_port < (start_discovery_port + MAX_ATTEMPT_PORTS);
            ++attempt_port)
        {
            sa->port = attempt_port;
            sa->sa.sin.sin_port = htons(attempt_port);
            status = apr_socket_bind(mSocket, sa);
            if(APR_SUCCESS == status) break;
            if(is_addr_in_use(status)) continue;
            (void)ll_apr_warn_status(status);
        }
    }
    if(ll_apr_warn_status(status)) return;
    if(sa->port)
    {
        LL_DEBUGS() << "Bound datagram socket to port: " << sa->port << LL_ENDL;
        mPort = sa->port;
    }
    else
    {
        mPort = LLIOSocket::PORT_EPHEMERAL;
    }

    // set up the socket options options
    ll_apr_warn_status(apr_socket_timeout_set(mSocket, 0));
    ll_apr_warn_status(apr_socket_opt_set(mSocket, APR_SO_SNDBUF, LL_SEND_BUFFER_SIZE));
    ll_apr_warn_status(apr_socket_opt_set(mSocket, APR_SO_RCVBUF, LL_RECV_BUFFER_SIZE));
}

LLIODataSocket::~LLIODataSocket()
{
}


#endif