McSocket.h

Go to the documentation of this file.

// -*- mode: C++; indent-tabs-mode: nil; c-basic-offset: 4; tab-width: 4; -*-
// vim: set shiftwidth=4 softtabstop=4 expandtab:
/*
 ********************************************************************
 ** NIDAS: NCAR In-situ Data Acquistion Software
 **
 ** 2006, Copyright University Corporation for Atmospheric Research
 **
 ** This program is free software; you can redistribute it and/or modify
 ** it under the terms of the GNU General Public License as published by
 ** the Free Software Foundation; either version 2 of the License, or
 ** (at your option) any later version.
 **
 ** This program 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 General Public License for more details.
 **
 ** The LICENSE.txt file accompanying this software contains
 ** a copy of the GNU General Public License. If it is not found,
 ** write to the Free Software Foundation, Inc.,
 ** 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 **
 ********************************************************************
*/
/*
 ********************************************************************
    Copyright by the National Center for Atmospheric Research

    $LastChangedDate$

    $LastChangedRevision$

    $LastChangedBy$

    $HeadURL$
 ********************************************************************

*/

#ifndef NIDAS_UTIL_MCSOCKET_H
#define NIDAS_UTIL_MCSOCKET_H

#include <nidas/Config.h>   // HAVE_PPOLL

#include "Socket.h"
#include "DatagramPacket.h"
#include "Thread.h"
#include "UTime.h"

#ifdef HAVE_PPOLL
#include <poll.h>
#else
#include <sys/select.h>
#endif

#include <list>

#include <unistd.h>   // need declaration of sleep()

namespace nidas { namespace util {

struct McSocketData {

    int magic;
    int _requestType;

    unsigned short _listenPort;

    short _socketType;

    McSocketData() : magic(0),_requestType(htonl((unsigned)-1)),_listenPort(0),_socketType(htons(SOCK_STREAM)) {}
};

class McSocketDatagram: public DatagramPacketT<McSocketData>
{
public:
    McSocketDatagram(int requestType=-1);

    McSocketDatagram(const McSocketDatagram& x);

    McSocketDatagram& operator=(const McSocketDatagram& rhs);

    int getMagic() const { return ntohl(mcdata.magic); }

    void setMagic(int val) { mcdata.magic = htonl(val); }

    int getRequestType() const { return ntohl(mcdata._requestType); }

    void setRequestType(int val) { mcdata._requestType = htonl(val); }

    int getRequesterListenPort() const { return ntohs(mcdata._listenPort); }

    void setRequesterListenPort(int val) { mcdata._listenPort = htons(val); }

    int getSocketType() const { return ntohs(mcdata._socketType); }

    void setSocketType(int val) { mcdata._socketType = htons(val); }

    static const int magicVal;

protected:
    struct McSocketData mcdata;
};


template <class SocketT>
class McSocket;

inline int getMcSocketType(McSocket<Socket>*)
{
    return SOCK_STREAM;
}

inline int getMcSocketType(McSocket<DatagramSocket>*)
{
    return SOCK_DGRAM;
}

template <class SocketT>
class McSocket
{
    friend class McSocketListener;

    template<class SocketTT>
    friend class McSocketMulticaster;
public:
    McSocket();

    McSocket(const McSocket<SocketT>&);

    McSocket<SocketT>& operator=(const McSocket<SocketT>& rhs);

    virtual ~McSocket();

    void setInterface(Inet4NetworkInterface iaddr) {
        _iface = iaddr;
    }

    Inet4NetworkInterface getInterface() const { return _iface; }

    std::list<Inet4NetworkInterface> getInterfaces() const throw(IOException);

    const Inet4SocketAddress& getInet4McastSocketAddress() const
    {
        return _mcastAddr;
    }

    void setInet4McastSocketAddress(const Inet4SocketAddress& val) { _mcastAddr = val; }

    void setRequestType(int val) { _requestType = val; }

    int getRequestType() const { return _requestType; }

    void listen() throw(IOException);

    SocketT* accept(Inet4PacketInfoX&) throw(IOException);

    virtual void connected(SocketT*,const Inet4PacketInfoX&) {}

    void request() throw(IOException);

    SocketT* connect(Inet4PacketInfoX&) throw(IOException);

    virtual void close() throw(IOException);

private:
    void offer(SocketT* sock,const Inet4PacketInfoX& pktinfo);

    void offer(int err);

    void joinMulticaster();

    Inet4SocketAddress _mcastAddr;

    Inet4NetworkInterface _iface;

    int _requestType;

    Cond _connectCond;

    SocketT* _newsocket;

    Inet4PacketInfoX _newpktinfo;

    bool _socketOffered;

    int _offerErrno;

    Thread* _multicaster;

    Mutex _multicaster_mutex;

    // Inet4PacketInfoX _pktinfo;

};

class McSocketListener: private Thread
{
    template<class SocketT>
    friend class McSocket;

public:

    static int check() throw();

private:
    static void accept(McSocket<Socket>* sock) throw(Exception);

    static void accept(McSocket<DatagramSocket>* sock) throw(Exception);

    static void close(McSocket<Socket>* sock) throw(Exception);

    static void close(McSocket<DatagramSocket>* sock) throw(Exception);

    int run() throw(Exception);

    void interrupt();

private:
    McSocketListener(const Inet4SocketAddress& addr);

    ~McSocketListener();

    void add(McSocket<Socket>* mcsocket);

    void add(McSocket<DatagramSocket>* mcsocket);

    int remove(McSocket<Socket>* mcsocket);

    int remove(McSocket<DatagramSocket>* mcsocket);

    McSocketListener(const McSocketListener&);

    McSocketListener& operator=(const McSocketListener&);

private:

    Inet4SocketAddress _mcastAddr;

    Mutex _mcsocket_mutex;

    DatagramSocket* _readsock;

    std::map<int,McSocket<Socket>*> _tcpMcSockets;

    std::map<int,McSocket<DatagramSocket>*> _udpMcSockets;

};

template<class SocketTT>
class McSocketMulticaster: private Thread
{
    template<class SocketT>
    friend class McSocket;

private:
    McSocketMulticaster(McSocket<SocketTT>* mcsocket);

    McSocketMulticaster(const McSocketMulticaster&);

    McSocketMulticaster& operator=(const McSocketMulticaster&);

    virtual ~McSocketMulticaster();

    int run() throw(Exception);

    void interrupt();

    McSocket<SocketTT>* _mcsocket;

    ServerSocket* _serverSocket;

    DatagramSocket* _datagramSocket;

    DatagramSocket* _requestSocket;

    Mutex _mcsocketMutex;
};

}}  // namespace nidas namespace util

#include "Logger.h"
#include <memory>

namespace nidas { namespace util {

template<class SocketT>
McSocket<SocketT>::McSocket(): _mcastAddr(),_iface(),_requestType(-1),
    _connectCond(),
    _newsocket(0),_newpktinfo(),
    _socketOffered(false),_offerErrno(0),
    _multicaster(0),_multicaster_mutex()
{
}

template<class SocketT>
McSocket<SocketT>::McSocket(const McSocket<SocketT>& x) :
    _mcastAddr(x._mcastAddr),_iface(x._iface),
    _requestType(x._requestType), _connectCond(),
    _newsocket(0),_newpktinfo(),
    _socketOffered(false),_offerErrno(0),
    _multicaster(0),_multicaster_mutex()
{
}

template<class SocketT>
McSocket<SocketT>& McSocket<SocketT>::operator=(const McSocket<SocketT>& rhs)
{
    if (&rhs != this) {
        _mcastAddr = rhs._mcastAddr;
        _iface = rhs._iface;
        _requestType = rhs._requestType;
        _newsocket = 0;
        _socketOffered = false;
        _multicaster = 0;
    }
    return *this;
}

template<class SocketT>
McSocket<SocketT>::~McSocket() {
}

template<class SocketT>
std::list<Inet4NetworkInterface> McSocket<SocketT>::getInterfaces() const throw(IOException)
{
    MulticastSocket tmpsock(_mcastAddr.getPort());
    std::list<Inet4NetworkInterface> ifcs = tmpsock.getInterfaces();
    tmpsock.close();
    return ifcs;
}

template<class SocketT>
void McSocket<SocketT>::listen() throw(IOException)
{
    if (getRequestType() < 0)
        throw IOException(_mcastAddr.toString(),"listen",
        "request number has not been set");
    _newsocket = 0;
    _socketOffered = false;
    try {
    McSocketListener::accept(this);
    }
    catch (const Exception& e) {
    throw IOException(_mcastAddr.toString(),"accept",e.what());
    }
}

/*
 * Does a listen and then waits for the connection.
 */
template<class SocketT>
SocketT* McSocket<SocketT>::accept(Inet4PacketInfoX& pktinfo) throw(IOException)
{
    listen();
    _connectCond.lock();
    while(!_socketOffered) _connectCond.wait();
    SocketT* socket = _newsocket;
    pktinfo = _newpktinfo;
    _newsocket = 0;
    _connectCond.unlock();
    VLOG(("accept offerErrno=%d", _offerErrno));
    if (!socket) throw IOException("McSocket","accept",_offerErrno);
    return socket;
}

template<class SocketT>
void McSocket<SocketT>::request() throw(IOException)
{
    if (getRequestType() < 0)
        throw IOException(_mcastAddr.toString(),"listen",
                          "request number has not been set");
    _newsocket = 0;
    _socketOffered = false;
    // if (!_mcastAddr.getInet4Address().isMultiCastAddress())
    //  throw IOException(_mcastAddr.toString(),"accept","is not a multicast address");

    _multicaster_mutex.lock();
    if (!_multicaster) {
        _multicaster = new McSocketMulticaster<SocketT>(this);
        try {
            _multicaster->start();
        }
        catch(const Exception& e) {
            throw IOException("McSocket","request",e.what());
        }
    }
    _multicaster_mutex.unlock();
}

/*
 * Does a request() and then waits for the connection.
 */
template<class SocketT>
SocketT* McSocket<SocketT>::connect(Inet4PacketInfoX& pktinfo) throw(IOException)
{
    request();
    _connectCond.lock();
    while(!_socketOffered) _connectCond.wait();
    SocketT* socket = _newsocket;
    pktinfo = _newpktinfo;
    _newsocket = 0;
    _connectCond.unlock();
    VLOG(("connect offerErrno=%d",_offerErrno));
    if (!socket) throw IOException("McSocket","connect",_offerErrno);
    return socket;
}

template<class SocketT>
void McSocket<SocketT>::joinMulticaster()
{
    // can't do a multicaster->join here, that would deadlock
    // because we'd be waiting for ourselves,
    // so we spawn a detached thread to join and delete it.

    _multicaster_mutex.lock();

    if (_multicaster) {
        VLOG(("Mcsocket::offer creating/starting joiner"));
        ThreadJoiner* joiner = new ThreadJoiner(_multicaster);
        _multicaster = 0;
        try {
            joiner->start();
        }
        // shouldn't happen
        catch(const Exception& e) {
            Logger::getInstance()->log(LOG_ERR,"%s",e.what());
        }
        VLOG(("Mcsocket::offer joiner started"));
    }
    VLOG(("Mcsocket::offer doing connectCond.lock"));
    _multicaster_mutex.unlock();
}
/*
 * Method that executes in the thread of the McSocketListener
 * or the McSocketMulticaster, and notifies whoever did
 * the listen() or request() that the socket is connected.
 */
template<class SocketT>
void McSocket<SocketT>::offer(SocketT* socket,const Inet4PacketInfoX& pktinfo)
{
    joinMulticaster();

    if (socket) connected(socket,pktinfo);

    _connectCond.lock();
    _socketOffered = true;
    _offerErrno = 0;
    _newsocket = socket;
    _newpktinfo = pktinfo;
    _connectCond.signal();
    _connectCond.unlock();

    // note: don't access any class variables or virtual methods
    // after the above unlock.  The calling thread may have
    // deleted this object after it received the connected Socket.
}

template<class SocketT>
void McSocket<SocketT>::offer(int err)
{
    joinMulticaster();

    _connectCond.lock();
    _socketOffered = true;
    _offerErrno = err;
    _newsocket = 0;
    _connectCond.signal();
    _connectCond.unlock();

    // note: don't access any class variables or virtual methods
    // after the above unlock.  The calling thread may have
    // deleted this object after it received the connected Socket.
}

template<class SocketT>
void McSocket<SocketT>::close() throw(IOException)
{
    VLOG(("Mcsocket::close"));
    _multicaster_mutex.lock();
    if (_multicaster && _multicaster->isRunning()) _multicaster->interrupt();
    _multicaster_mutex.unlock();

    offer(EINTR);

    try {
        McSocketListener::close(this);
    }
    catch (const Exception& e) {
        VLOG(("Mcsocket::close exception: %s", e.what()));
        throw IOException(_mcastAddr.toString(), "close", e.what());
    }
    VLOG(("Mcsocket::close done, this=0x%x", this));
}

template<class SocketTT>
McSocketMulticaster<SocketTT>::McSocketMulticaster(McSocket<SocketTT>* mcsock) :
    Thread("McSocketMulticaster"),
    _mcsocket(mcsock),_serverSocket(0),_datagramSocket(0),_requestSocket(0),
    _mcsocketMutex()
{
    switch (getMcSocketType(mcsock)) {
    case SOCK_STREAM:
        _serverSocket = new ServerSocket();
        _datagramSocket = 0;
        break;
    default:
        _datagramSocket = new DatagramSocket();
        _serverSocket = 0;
        break;
    }
    // block SIGUSR1, then unblock it in pselect/ppoll
    blockSignal(SIGUSR1);
}

template<class SocketT>
McSocketMulticaster<SocketT>::~McSocketMulticaster()
{
    if (_serverSocket) {
        _serverSocket->close();
        delete _serverSocket;
    }
    if (_datagramSocket) {
        _datagramSocket->close();
        delete _datagramSocket;
    }
    if (_requestSocket) {
        _requestSocket->close();
        delete _requestSocket;
    }
}

template<class SocketT>
void McSocketMulticaster<SocketT>::interrupt()
{
    _mcsocketMutex.lock();
    _mcsocket = 0;
    _mcsocketMutex.unlock();
    try {
        kill(SIGUSR1);
    }
    catch(const Exception& e) {
        PLOG(("%s",e.what()));
    }
}


void
listMulticastInterfaces(MulticastSocket* requestmsock,
                        std::vector<Inet4NetworkInterface>& ifaces);

template<class SocketT>
int McSocketMulticaster<SocketT>::run() throw(Exception)
{
    MulticastSocket* requestmsock = 0;

    int sockfd = (_serverSocket ? _serverSocket->getFd() :
        _datagramSocket->getFd());

#ifdef HAVE_PPOLL
    struct pollfd fds;
    fds.fd =  sockfd;
#ifdef POLLRDHUP
    fds.events = POLLIN | POLLRDHUP;
#else
    fds.events = POLLIN;
#endif
#else
    fd_set fdset;
    FD_ZERO(&fdset);
#endif

    // wait for this amount of time after a datagram send to receive a
    // connection or receipt of UDP data
    struct timespec timeout;
    timeout.tv_sec = 0;
    timeout.tv_nsec = NSECS_PER_SEC / 4;             // portion of a second

    // get the existing signal mask
    sigset_t sigmask;
    pthread_sigmask(SIG_BLOCK,NULL,&sigmask);
    // unblock SIGUSR1 in pselect/ppoll
    sigdelset(&sigmask,SIGUSR1);

    Inet4SocketAddress mcsockaddr =
        _mcsocket->getInet4McastSocketAddress();
    Inet4Address mcaddr = mcsockaddr.getInet4Address();

    std::vector<Inet4NetworkInterface> ifaces;

    if (mcaddr.isMultiCastAddress())
    {
        VLOG(("") << mcsockaddr.toString() << " is a multicast address.");
        _requestSocket = requestmsock = new MulticastSocket();
        if (_mcsocket->getInterface().getAddress() == Inet4Address(INADDR_ANY))
        {
            listMulticastInterfaces(requestmsock, ifaces);
        }
    }
    else
    {
        VLOG(("") << mcsockaddr.toString()
             << " is not a multicast address, using a datagram request socket.");
        _requestSocket = new DatagramSocket();
    }

    McSocketDatagram dgram;
    dgram.setMagic(dgram.magicVal);
    dgram.setSocketAddress(mcsockaddr);
    dgram.setRequestType(_mcsocket->getRequestType());
    dgram.setSocketType(getMcSocketType(_mcsocket));
    if (_serverSocket)
        dgram.setRequesterListenPort(_serverSocket->getLocalPort());
    else
        dgram.setRequesterListenPort(_datagramSocket->getLocalPort());

    for (int numCasts=0; !isInterrupted() ; numCasts++) {
        // If multicast, loop over interfaces
        try {
            if (requestmsock && ifaces.size() > 0) {
                Inet4NetworkInterface iface = ifaces[numCasts % ifaces.size()];
                requestmsock->setInterface(mcaddr,iface);
                _requestSocket->send(dgram);
            }
            else {
                _requestSocket->send(dgram);
            }
            if (!(numCasts % 300)) {
                ILOG(("") << "sent " << numCasts << " dgrams" <<
                     ", requestType=" << dgram.getRequestType() <<
                     ", port=" << dgram.getRequesterListenPort() <<
                     ", socketType=" << dgram.getSocketType() <<
                     ", len=" << dgram.getLength() <<
                     ", #mcifaces=" << ifaces.size());
            }
        }
        catch(const IOException& e)
        {
            // perhaps the interface has disappeared. Log the error.
            WLOG(("McSocketMulticaster: ")
                 << _requestSocket->getLocalSocketAddress().toString()
                 << ": " << e.what());
            sleep(10);
            if (requestmsock)
            {
                _requestSocket->close();
                delete _requestSocket;
                // close and re-create the socket
                _requestSocket = requestmsock = new MulticastSocket();
                // re-create the list of interfaces.
                listMulticastInterfaces(requestmsock, ifaces);
            }
            continue;
        }

        int res;
#ifdef HAVE_PPOLL
        res = ::ppoll(&fds,1,&timeout,&sigmask);

        if (res > 0) {

            if (fds.revents & POLLERR) {
                _mcsocketMutex.lock();
                if (_mcsocket) _mcsocket->offer(errno);
                _mcsocketMutex.unlock();
                if (_serverSocket) _serverSocket->close(); // OK to close twice
                if (_datagramSocket) _datagramSocket->close();
                _requestSocket->close();
                throw IOException("McSocket","ppoll",errno);
            }

#ifdef POLLRDHUP
            if (fds.revents & (POLLHUP | POLLRDHUP))
#else
            if (fds.revents & POLLHUP)
#endif
                WLOG(("%s POLLHUP","McSocket"));

        }

#else
        assert(sockfd >= 0 && sockfd < FD_SETSIZE);     // FD_SETSIZE=1024
        FD_SET(sockfd, &fdset);
        res = ::pselect(sockfd+1,&fdset,0,0,&timeout,&sigmask);
#endif
        if (res <= 0) {
            if (res == 0) continue;
            if (errno == EINTR) break;
            _mcsocketMutex.lock();
            if (_mcsocket) _mcsocket->offer(errno);
            _mcsocketMutex.unlock();
            if (_serverSocket) _serverSocket->close(); // OK to close twice
            if (_datagramSocket) _datagramSocket->close();
            _requestSocket->close();
            throw IOException("McSocket","poll/select",errno);
        }

        if (_serverSocket) {
            Socket* socket = _serverSocket->accept();
            DLOG(("accepted socket connection from ") <<
                socket->getRemoteSocketAddress().toString() << " on " <<
                _serverSocket->getLocalSocketAddress().toString());
            _serverSocket->close();
            nidas::util::Inet4PacketInfoX pktinfo;
            if (socket->getRemoteSocketAddress().getFamily() == AF_INET) {
                nidas::util::Inet4SocketAddress remoteAddr =
                    nidas::util::Inet4SocketAddress((const struct sockaddr_in*)
                    socket->getRemoteSocketAddress().getConstSockAddrPtr());
                pktinfo.setRemoteSocketAddress(remoteAddr);
            }
            if (socket->getLocalSocketAddress().getFamily() == AF_INET) {
                nidas::util::Inet4SocketAddress localAddr =
                    nidas::util::Inet4SocketAddress((const struct sockaddr_in*)
                    socket->getLocalSocketAddress().getConstSockAddrPtr());
                pktinfo.setLocalAddress(localAddr.getInet4Address());
                pktinfo.setDestinationAddress(localAddr.getInet4Address());
            }
            _mcsocketMutex.lock();
            if (_mcsocket) _mcsocket->offer((SocketT*)socket,pktinfo);
            _mcsocketMutex.unlock();
        }
        else {
            // If fishing for UDP responses, send out at least 3 multicasts
            // to see if we get more than one response.
            if (numCasts < 3) continue;

            // We know there is data at the socket, do a MSG_PEEK to get
            // information about the first packet.
            nidas::util::Inet4PacketInfoX pktinfo;
            _datagramSocket->receive(dgram,pktinfo,MSG_PEEK);
            if (dgram.getSocketAddress().getFamily() == AF_INET) {
                nidas::util::Inet4SocketAddress remoteAddr =
                    nidas::util::Inet4SocketAddress((const struct sockaddr_in*)
                    dgram.getSocketAddress().getConstSockAddrPtr());
                pktinfo.setRemoteSocketAddress(remoteAddr);
            }

            _mcsocketMutex.lock();
            if (_mcsocket) _mcsocket->offer((SocketT*)_datagramSocket,pktinfo);
            _datagramSocket = 0;    // we no longer own _datagramSocket
            _mcsocketMutex.unlock();
        }
        _requestSocket->close();
        return RUN_OK;
    }   // loop until receipt of connection or UDP packet

    VLOG(("McSocketMulticaster break"));
    _mcsocketMutex.lock();
    if (_mcsocket) _mcsocket->offer(EINTR);
    _mcsocketMutex.unlock();
    if (_serverSocket) _serverSocket->close();
    if (_datagramSocket) _datagramSocket->close();
    _requestSocket->close();
    VLOG(("McSocketMulticaster run method exiting"));
    return RUN_OK;
}

}}  // namespace nidas namespace util
#endif