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

---

ChannelPool.h

Go to the documentation of this file.

#ifndef Impala_Util_ChannelPool_h
#define Impala_Util_ChannelPool_h

#include "Basis/ILog.h"
#include "Basis/String.h"
#include "Util/Channel.h"
#include "Util/ChannelAuthorization.h"
#include "Util/Cache.h"

#ifdef MPI_USED
#include "Link/Mpi/MpiFuncs.h"
#endif

namespace Impala
{
namespace Util
{


class ChannelPool
{
public:

    ~ChannelPool()
    {
        delete mCache;
    }

    static ChannelPool&
    Instance()
    {
        static ChannelPool theSystem;
        return theSystem;
    }

    // socketName == host.domain:port
    Channel*
    Get(CString socketName, CString passwordFile)
    {
        String::size_type p = socketName.find(":");
        String serverName = socketName.substr(0, p);
        int port = atoi(socketName.substr(p+1, String::npos));
        return Get(serverName, port, socketName, passwordFile);
    }

    // serverName == host.domain
    Channel*
    Get(CString serverName, int port, CString passwordFile)
    {
        return Get(serverName, port, serverName + ":" + MakeString(port),
                   passwordFile);
    }

    void
    Remove(CString serverName, int port)
    {
        String socketName = serverName + ":" + MakeString(port);
        mCache->RemoveElement(socketName);
    }

    void
    Dump() const
    {
        mCache->Dump("ChannelPool - ");
    }

private:

    ChannelPool()
    {
        mCache = new ChannelCache(CACHE_SIZE, true);
    }

    ChannelPool(const ChannelPool&)
    {
    }

    ChannelPool&
    operator=(const ChannelPool&);

    // multiple calls to this method specifying the same server+port, will 
    // be returned one and the same open channel object (assuming the channel
    // can successfully be created).
    Channel*
    Get(CString serverName, int port, CString socketName, CString passwordFile)
    {
        Channel* chan = 0;
        ILOG_DEBUG("Get for " << socketName);
        if (! mCache->GetElement(socketName, chan))
        {
            ILOG_DEBUG("Get: trying new Channel");
            chan = new Util::Channel(serverName, port);
            if (! chan->Valid())
            {
                ILOG_ERROR("Failed to open channel to " << socketName);
                delete chan;
                return 0;
            }
            ILOG_DEBUG("Get: Opened channel to " << socketName);

            ChannelAuthorization auth(passwordFile);
            if (!auth.Valid())
            {
                ILOG_ERROR("Cannot open Channel without valid authorization");
                delete chan;
                return 0;
            }
            char* buf = chan->Buffer();
            sprintf(buf, "PortMode read=\"%s\" write=\"%s\"\0",
                    auth.GetReadPassword().c_str(),
                    auth.GetWritePassword().c_str());
            int len = chan->SendRequest(strlen(buf) + 1);
            if (len < 0)
            {
                ILOG_ERROR("Port mode request failed (" << len <<
                           "); closing channel");
                delete chan;
                return 0;
            }

            buf[len] = 0;
            ILOG_DEBUG("Port mode response from server: " << buf);
            if (String(buf) == "Permission denied")
            {
                ILOG_ERROR("Permission denied; closing channel");
                delete chan;
                return 0;
            }
            else if (String(buf) == "Static")
            {
                // do nothing; the current channel/port is used for serving
                // further requests
            }
            else if (String(buf) == "Dynamic")
            {
                ILOG_DEBUG("Requesting port assignment");
                sprintf(buf, "PortAssignment write=\"%s\"\0",
                        auth.GetWritePassword().c_str());
                len = chan->SendRequest(strlen(buf) + 1);
                if (len < 0)
                {
                    ILOG_ERROR("Port assignment request failed (" << len <<
                               "); closing channel");
                    delete chan;
                    return 0;
                }
                String response(buf);
                if (response.substr(0, 15) == "NoPortAvailable")
                {
                    ILOG_ERROR("No free port available; closing channel");
                    delete chan;
                    return 0;
                }
                if (response.substr(0, 13) != "PortAssigned=")
                {
                    ILOG_ERROR("Unexpected response to port assignment request ("
                               << buf << "); closing channel");
                    delete chan;
                    return 0;
                }
                int portAssigned = atoi(response.substr(13));
                ILOG_DEBUG("Port assigned is " << portAssigned);
                delete chan;
                ILOG_DEBUG("Closed channel to " << socketName);

                chan = new Util::Channel(serverName, portAssigned);
                if (! chan->Valid())
                {
                    ILOG_ERROR("Failed to open channel to " << serverName << ":"
                               << MakeString(portAssigned));
                    delete chan;
                    return 0;
                }
                ILOG_DEBUG("Opened channel to " << serverName <<
                           " at assigned port " << portAssigned);
            }
            else
            {
                ILOG_WARN("Unexpected response to port mode request (" << buf <<
                          "); server may not support dynamic port assignment");
                ILOG_DEBUG("Opened channel to " << serverName << " at port " <<
                           port);
            }

            // Note that for dynamically assigned ports, the actual port number
            // is not the same as the port number used for the cache.
            ILOG_DEBUG("Adding chan to " << socketName << " : " << chan);
            mCache->AddElement(socketName, chan);
        }
        else
        {
            ILOG_DEBUG("Get: Returning channel from pool: " << chan);
        }

        return chan;
    }

    static const int CACHE_SIZE = 100;

    typedef Cache<String, Channel*> ChannelCache;
    ChannelCache* mCache;

    ILOG_VAR_DEC;

}; // class

ILOG_VAR_INIT(ChannelPool, Impala.Util);

} // namespace Util
} // namespace Impala

#endif

---