|| || || || || ||

---

Channel.h

Go to the documentation of this file.

#ifndef Impala_Util_Channel_h
#define Impala_Util_Channel_h

#ifdef unix
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netdb.h>
#include <stdio.h>
#include <unistd.h>
#include <sys/ioctl.h>
#include <sys/signal.h>
#include <errno.h>
#ifdef sun
#include <signal.h>
#include <sys/filio.h>
#endif
#else
#include <winsock2.h>
#endif

#include "Basis/String.h"
#include <iostream>
#include <vector>
#include <cstring>

#include "Basis/ILog.h"

namespace Impala
{
namespace Util
{

class ChannelListener
{
public:

    virtual void HandleInfo(CString msg) = 0;
    virtual void HandleError(CString msg) = 0;
    virtual void HandleIdle() = 0;
    virtual int  HandleRequest(char* buf, int len, int bufSize) = 0;
    //virtual void HandleConnect(CString conn, int port) = 0;
    virtual void HandleDisconnect(int port) = 0;
};

class Channel
{
public: 

    Channel(int port, int nrPorts, ChannelListener* listener = 0)
    {
        mNrError = 0;
        mPort = port;
        mNrPorts = nrPorts;
        mIsServer = true;
        mFpInfo = 0;
        mFpError = 0;
        mListener = listener;
        mCallbackOnIdle = false;

        InitSocket();

        mBuffer = new char[DATA_BUFFER_SIZE + CTR_HEADER_SIZE];
        if (mBuffer == 0)
            HandleError("buffer allocation failed", false);
    }

    Channel(String serverName, int port)
    {
        mNrError = 0;
        mServerName = serverName;
        mPort = port;
        mNrPorts = 1;
        mIsServer = false;
        mFpInfo = 0;
        mFpError = 0;
        mListener = 0;
        mCallbackOnIdle = false;

        InitSocket();

        mBuffer = new char[DATA_BUFFER_SIZE + CTR_HEADER_SIZE];
        if (mBuffer == 0)
            HandleError("buffer allocation failed", false);
    }

    ~Channel()
    {
        Close();
        if (mBuffer)
            delete mBuffer;
    }

    int
    GetNrPorts() const
    {
        return mNrPorts;
    }

    int
    GetPortNumber(int idx) const
    {
        return mPort + idx;
    }

    String
    GetCurRequestConnection() const
    {
        return mConnection[mCurRequestSock];
    }

    int
    GetCurRequestPort() const
    {
        return mPort + mCurRequestSock;
    }

    String
    GetServerInfo()
    {
        return mServerName + ":" + MakeString(mPort);
    }

    bool
    Valid()
    {
        return mNrError == 0;
    }

    void
    SetListener(ChannelListener* listener)
    {
        mListener = listener;
    }

    typedef void (*FP_INFO)(CString msg);

    typedef void (*FP_ERROR)(CString msg);

    void
    SetMessageFunc(FP_INFO fpInfo, FP_ERROR fpError)
    {
        mFpInfo = fpInfo;
        mFpError = fpError;
    }

    static const int DATA_BUFFER_SIZE = 10485760; // 10 Mb

    char*
    Buffer()
    {
        return mBuffer + CTR_HEADER_SIZE;
    }

    typedef int (*FP_REQUEST)(char* buf, int len, int bufSize);

    typedef void (*FP_IDLE)();

    void
    Serve(FP_REQUEST fpRequest, FP_IDLE fpIdle = 0)
    {
        mFpRequest = fpRequest;
        mFpIdle = fpIdle;
        Serve(fpIdle != 0);
    }

    void
    Serve(bool callbackOnIdle = false)
    {
        if (!mListener && !mFpRequest)
        {
            String msg = "Either listener or function pointer(s) must be specified";
            ILOG_ERROR(msg);
            return;
        }

        mCallbackOnIdle = callbackOnIdle;

        // Start accepting connections.
        for (int i=0 ; i<mSock.size() ; i++)
            listen(mSock[i], 100);
        String msg = "Listening at port(s) " + MakeString(mPort);
        if (mNrPorts > 1)
            msg += "-" + MakeString(mPort + mNrPorts - 1);
        HandleInfo(msg);
        do
        {
            int res = WaitForSocketEvent();
            if (res == -1)
                continue;
            for (int s=0 ; s<mSockEvent.size() ; s++)
            {
                if (mSockEvent[s])
                {
                    ILOG_DEBUG("Accepting connection at port " << mPort+s);
                    // org, no address info
                    //mMsgSock[s] = accept(mSock[s], 0, 0);

                    // new, obtain address
                    struct sockaddr_in address;
                    int addressLen = sizeof(address);
                    mMsgSock[s] = accept(mSock[s], (sockaddr *) &address,
                                         (socklen_t*) &addressLen);
                    //ILOG_DEBUG("accept on port " << ntohs(address.sin_port));

                    if (mMsgSock[s] == INVALID_SOCKET)
                    {
                        HandleError("Serve: accept", true);
                    }
                    else
                    {
                        mSockConnected[s] = true;
                        in_addr_t a = address.sin_addr.s_addr;
                        int a1 = (a & 0xff000000) >> 24;
                        int a2 = (a & 0x00ff0000) >> 16;
                        int a3 = (a & 0x0000ff00) >> 8;
                        int a4 = (a & 0x000000ff);
                        std::ostringstream oss;
                        oss << a4 << "." << a3 << "." << a2 << "." << a1;
                        mConnection[s] = oss.str();
                        HandleInfo("Established connection with " +
                                   mConnection[s] + " at port " +
                                   MakeString(mPort + s));
                    }
                }
                if (mMsgSockEvent[s])
                {
                    ILOG_DEBUG("Receiving at port " << mPort+s);
                    mCurRequestSock = s;
                    int rval = RecvLarge(mMsgSock[s]);
                    if (rval > 0)
                    {
                        int nrToSend = HandleRequest(Buffer(),
                                                   rval - CTR_HEADER_SIZE,
                                                   DATA_BUFFER_SIZE);
                        rval = SendLarge(mMsgSock[s], nrToSend);
                    }
                    if (rval <= 0)
                    {
                        HandleInfo("Terminating connection with " +
                                   GetCurRequestConnection() + " at port " +
                                   MakeString(GetCurRequestPort()));
                        CloseSocket(mMsgSock[s]);
                        HandleDisconnect(GetCurRequestPort());
                        mSockConnected[s] = false;
                        mConnection[s] = "not connected";
                    }
                }
            }
        }
        while(1);
    }

    int
    SendRequest(int len)
    {
        int rval;
        rval = SendLarge(mSock[0], len);
        rval = RecvLarge(mSock[0]);
        if ((rval > 5) && (strncmp(Buffer(), "ERROR", 5) == 0))
        {
            HandleError("SendRequest to " + mServerName + " got " +
                        String(Buffer()), false);
            mNrError--; // this was not a Channel error so stay Valid()
        }
        return rval - CTR_HEADER_SIZE;
    }

    bool
    LastSendHadError()
    {
        return strncmp(Buffer(), "ERROR", 5) == 0;
    }

    void
    MaximizeSendRecvBuffer()
    {
        int nOptVal;
        socklen_t nOptLen = sizeof(int);
        for (int i=0 ; i<mSock.size() ; i++)
        {
            int nRet;
            nRet = getsockopt(mSock[i], SOL_SOCKET, SO_SNDBUF, (char*)&nOptVal,
                              &nOptLen);
            if (nRet == SOCKET_ERROR)
                HandleError("MaximizeSendRecvBuffer", true);
            else
                HandleInfo("SNDBUF : " + MakeString(nOptVal));
            nRet = getsockopt(mSock[i], SOL_SOCKET, SO_RCVBUF, (char*)&nOptVal,
                              &nOptLen);
            if (nRet == SOCKET_ERROR)
                HandleError("MaximizeSendRecvBuffer", true);
            else
                HandleInfo("RCVBUF : " + MakeString(nOptVal));
        }
    }

    void
    Close()
    {
        for (int i=0 ; i<mSock.size() ; i++)
        {
            String conn = mServerName;
            int port = mPort + i;
            HandleInfo("Closing connection to " + conn + ":" + MakeString(port));
            CloseSocket(mSock[i]);
            HandleDisconnect(port);
        }

#ifndef unix
        if (mIsServer)
            WSACleanup();
#endif
    }

private:

    static const int CTR_HEADER_SIZE = 20;

#ifdef unix
    static const int INVALID_SOCKET = -1;
    static const int SOCKET_ERROR = -1;
    typedef int SOCKET;
#else
    typedef int socklen_t;
    typedef u_long in_addr_t;
#endif

    void
    InitSocket()
    {
#ifdef unix
        signal(SIGPIPE, SIG_IGN); // block broken pipe signal
#else
        WORD wVersionRequested = MAKEWORD(1,1);
        WSADATA wsaData;

        // Initialize WinSock and check version
        if (WSAStartup(wVersionRequested, &wsaData) != 0)
            HandleError("WSAStartup", true);

        // Check socket DLL supports 1.1 or higher
        double socklib_ver = HIBYTE(wsaData.wVersion) / 10.0;
        socklib_ver += LOBYTE(wsaData.wVersion);
        if (socklib_ver < 1.1)
            HandleError("socket library must support 1.1 or greater", true);
#endif

        for (int i=0 ; i<mNrPorts ; i++)
        {
            struct sockaddr_in address;
            address.sin_family = AF_INET;
            address.sin_port = htons(mPort + i);
            struct hostent* hp;
            if (mIsServer)
            {
                address.sin_addr.s_addr = htonl(INADDR_ANY);
            }
            else
            {
                // gethostbyname returns a structure including the network
                // address of the specified host.
                if ((hp = gethostbyname(mServerName.c_str())) == NULL)
                    HandleError("InitSocket: gethostbyname", true);
                // initialise the address structure with the resolved IP address
                memcpy((char *) &address.sin_addr, (char *) hp->h_addr,
                       hp->h_length);
            }

            SOCKET sock;
            // Create socket.
            if ((sock = socket(AF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET)
                HandleError("InitSocket: open stream socket", true);
            if (mIsServer)
            {
                if (bind(sock, (struct sockaddr *) &address,
                         sizeof(address)) == SOCKET_ERROR)
                {
                    HandleError("InitSocket: bind socket", true);
                }
            }
            else
            {
                int nrTries = 5;
                bool done = false;
                while (!done)
                {
                    int res = connect(sock, (struct sockaddr *) &address,
                                      sizeof(address));
                    if (res == SOCKET_ERROR)
                    {
                        nrTries--;
                        if (nrTries < 0)
                        {
                            HandleError("InitSocket: connect socket", true);
                            done = true;
                        }
                        else
                        {
                            ILOG_INFO("Connect failed, re-trying " << nrTries);
                        }
                    }
                    else
                    {
                        done = true;
                    }
                }
            }
            mSock.push_back(sock);
            mSockConnected.push_back(false);
            mConnection.push_back("not connected");
        }
        mSockEvent.resize(mNrPorts);
        mMsgSock.resize(mNrPorts);
        mMsgSockEvent.resize(mNrPorts);
    }

    void
    CloseSocket(SOCKET sock)
    {
#ifdef unix
        close(sock);
#else
        closesocket(sock);
#endif
    }

    int
    WaitForSocketEvent()
    {
        ILOG_DEBUG("Waiting for socket event");
        int nRet;
        int i = 0;
        do
        {
            // Wait for readable state. After accept() this indicates that
            // data has been received, so recv() will succeed.
            fd_set readSet;
            FD_ZERO(&readSet);
            for (int s=0 ; s<mSock.size() ; s++)
            {
                if (mSockConnected[s])
                {
                    FD_SET(mMsgSock[s], &readSet);
                    ILOG_DEBUG("  checking on " << mPort+s);
                }
            }
            // Wait for writable state. After listen() this indicates that
            // a connection is pending, so accept() will succeed.
            // Oops, on linux this doesn't work, so use readSet.
            //fd_set writeSet;
            //FD_ZERO(&writeSet);
            for (int s=0 ; s<mSock.size() ; s++)
            {
                if (!mSockConnected[s])
                {
                    //FD_SET(mSock[s], &writeSet);
                    FD_SET(mSock[s], &readSet);
                    ILOG_DEBUG("  checking on " << mPort+s);
                }
            }
            if (mCallbackOnIdle)
            {
                struct timeval timeOut;
                timeOut.tv_sec = 0;
                timeOut.tv_usec = 100000 * i;
                if (i < 9)
                    i++;
                //nRet = select(FD_SETSIZE, &readSet, &writeSet, 0, &timeOut);
                nRet = select(FD_SETSIZE, &readSet, 0, 0, &timeOut);
            }
            else
            {
                //nRet = select(FD_SETSIZE, &readSet, &writeSet, 0, 0);
                nRet = select(FD_SETSIZE, &readSet, 0, 0, 0);
            }
            if (nRet == SOCKET_ERROR)
            {
                HandleError("WaitForSocketEvent: select", true);
                return -1;
            }
            if (nRet == 0)
            {
                if (mCallbackOnIdle)
                    HandleIdle();
            }
            else
            {
                ILOG_DEBUG(nRet << " socket event(s)");
                // find readable/writable socket, there may be more than one
                for (int s=0 ; s<mSock.size() ; s++)
                {
                    mSockEvent[s] = false;
                    mMsgSockEvent[s] = false;
                    if (!mSockConnected[s])
                    {
                        //if (FD_ISSET(mSock[s], &writeSet))
                        if (FD_ISSET(mSock[s], &readSet))
                            mSockEvent[s] = true;
                    }
                    else
                    {
                        if (FD_ISSET(mMsgSock[s], &readSet))
                            mMsgSockEvent[s] = true;
                    }
                }
            }
        }
        while (nRet == 0);
        return nRet;
    }

    int
    SendLarge(SOCKET sock, int nrData)
    {
        int rval;
        ILOG_DEBUG("sending " << nrData << " data bytes");

        // assemble CTR header to tell receiver the number of bytes to expect
        int nrToSend = nrData + CTR_HEADER_SIZE;
        sprintf(mBuffer, "CTR %d\0", nrToSend);
        ILOG_DEBUG("send : CTR buf [" << mBuffer << "]");

        // now send the buffer
        int nrSend = 0;
        char* bufPtr = mBuffer;
        do
        {
            rval = send(sock, bufPtr, nrToSend - nrSend, 0);
            if (rval == SOCKET_ERROR)
            {
                HandleError("SendLarge", true);
                return -1;
            }
            ILOG_DEBUG("sent " << rval << " bytes");
            bufPtr += rval;
            nrSend += rval;
        }
        while (nrSend < nrToSend);
        if (nrSend != nrToSend)
        {
            HandleError("sent incorrect number of bytes!", true);
            return -1;
        }
        return nrSend;
    }

    int
    RecvLarge(SOCKET sock)
    {
        int rval;

        // first recv CTR message to find out the number of bytes to expect
        ILOG_DEBUG("receiving...");
        int nrToRecv = 0;
        int nrRecv = 0;
        char* bufPtr = mBuffer;
        int bufSize = DATA_BUFFER_SIZE + CTR_HEADER_SIZE;
        do
        {
            rval = recv(sock, bufPtr, bufSize-nrRecv, 0);
            if (rval == SOCKET_ERROR)
            {
                HandleError("RecvLarge", true);
                return -1;
            }
            if (rval == 0)
            {
                ILOG_DEBUG("recv got a 0 bytes message");
                return 0;
            }
            if (nrToRecv == 0)
            { // read CTR header for number of bytes to receive
                sscanf(bufPtr, "CTR %d", &nrToRecv);
                ILOG_DEBUG("recv expects " << nrToRecv << " bytes"
                             );
            }
            ILOG_DEBUG("    recv " << rval << " bytes");
            bufPtr += rval;
            nrRecv += rval;
        }
        while (nrRecv < nrToRecv);

        ILOG_DEBUG("received " << nrRecv << " bytes");
        if (nrRecv > nrToRecv)
        {
            HandleError("received MORE bytes than expected!", true);
            return -1;
        }
        return nrRecv;
    }

    int 
    HandleRequest(char* buf, int len, int bufSize)
    {
        if (mListener)
            return mListener->HandleRequest(buf, len, bufSize);
        else
            return (mFpRequest)(buf, len, bufSize);
    }

    void
    HandleIdle()
    {
        ILOG_DEBUG("Calling idle function");

        if (mListener)
            mListener->HandleIdle();
        else
            (mFpIdle)();
    }

    void
    HandleInfo(CString msg)
    {
        if (mListener)
            mListener->HandleInfo(msg);
        else if (mFpInfo)
            (mFpInfo)(msg);
        else
            ILOG_DEBUG(msg);
    }

    void
    HandleError(CString msg, bool isSockError)
    {
        mNrError++;
        String err;

#ifdef unix

        err = String(strerror(errno));

#else

        // Errors are handled by calling the WSAGetLastError routine which
        // will return the last error as one of the following.
        switch (WSAGetLastError())
        {
        case WSANOTINITIALISED :
            err = "Unable to initialise socket.";
            break;
        case WSAEAFNOSUPPORT :
            err = "The specified address family is not supported.";
            break;
        case WSAEADDRNOTAVAIL :
            err = "Specified address is not available from the local machine.";
            break;
        case WSAECONNREFUSED :
            err = "The attempt to connect was forcefully rejected.";
            break;
        case WSAEDESTADDRREQ :
            err = "Address destination address is required.";
            break;
        case WSAEFAULT :
            err = "The namelen argument is incorrect.";
            break;
        case WSAEINVAL :
            err = "The socket is not already bound to an address.";
            break;
        case WSAEISCONN :
            err = "The socket is already connected.";
            break;
        case WSAEADDRINUSE :
            err = "The specified address is already in use.";
            break;
        case WSAEMFILE :
            err = "No more file descriptors are available.";
            break;
        case WSAENOBUFS :
            err = "No buffer space available. The socket cannot be created.";
            break;
        case WSAEPROTONOSUPPORT :
            err = "The specified protocol is not supported.";
            break;
        case WSAEPROTOTYPE :
            err = "The specified protocol is the wrong type for this socket.";
            break;
        case WSAENETUNREACH :
            err = "The network can't be reached from this host at this time.";
            break;
        case WSAENOTSOCK :
            err = "The descriptor is not a socket.";
            break;
        case WSAETIMEDOUT :
            err = "Attempt timed out without establishing a connection.";
            break;
        case WSAESOCKTNOSUPPORT :
            err = "Socket type is not supported in this address family.";
            break;
        case WSAENETDOWN :
            err = "Network subsystem failure.";
            break;
        case WSAHOST_NOT_FOUND :
            err = "Authoritative Answer Host not found.";
            break;
        case WSATRY_AGAIN :
            err = "Non-Authoritative Host not found or SERVERFAIL.";
            break;
        case WSANO_RECOVERY :
            err = "Non recoverable errors, FORMERR, REFUSED, NOTIMP.";
            break;
        case WSANO_DATA :
            err = "Valid name, no data record of requested type.";
            break;
        case WSAEINPROGRESS :
            err = "Address blocking Windows Sockets operation is in progress.";
            break;
        case WSAEINTR :
            err = "The (blocking) call was canceled via WSACancelBlockingCall.";
            break;
        default :
            err = "Unknown error.";
            break;
        }

#endif

        String finalMsg = msg;
        if (isSockError)
            finalMsg += ": " + err;

        if (mListener)
            mListener->HandleError(finalMsg);
        else if (mFpError)
            (mFpError)(finalMsg);
        else
            ILOG_ERROR(finalMsg);
    }

    //void
    //HandleConnect(CString conn, int port)
    //{
    //    if (mListener)
    //        mListener->HandleConnect(conn, port);
    //}

    void
    HandleDisconnect(int port)
    {
        if (mListener)
            mListener->HandleDisconnect(port);
    }

    bool                mIsServer;
    String              mServerName;
    int                 mPort;
    int                 mNrPorts;
    std::vector<SOCKET> mSock; // connection sockets
    std::vector<bool>   mSockConnected;
    std::vector<String> mConnection;
    std::vector<bool>   mSockEvent;
    std::vector<SOCKET> mMsgSock; // sockets for sending messages
    std::vector<bool>   mMsgSockEvent;
    char*               mBuffer;
    int                 mNrError;
    int                 mCurRequestSock;
    FP_INFO             mFpInfo;
    FP_ERROR            mFpError;
    FP_REQUEST          mFpRequest;
    FP_IDLE             mFpIdle;
    bool                mCallbackOnIdle;
    ChannelListener*    mListener;

    ILOG_VAR_DECL;

}; // class

ILOG_VAR_INIT(Channel, Impala.Util);

} // namespace Util
} // namespace Impala

#endif

---