src/acedia_network.cpp

/*                                                                            *\
**                          Copyright (C) 2009-2010                           **
**       Dominik Charousset < Dominik.Charousset (at) haw-hamburg.de >        **
**                                                                            **
**            MMM       MMMMMM   MMMMMMMMM MMMMMMM   MMM       MMM            **
**           MMMMM     MMMMMMMM  MMMMMMMMM MMMMMMMM  MMM      MMMMM           **
**          MMM MMM   MMM        MMMM      MMM   MMM MMM     MMM MMM          **
**         MMM   MMM  MMM        MMMMMMMMM MMM   MMM MMM    MMM   MMM         **
**        MMMMMMMMMMM MMMM   MMM MMMM      MMM   MMM MMM   MMMMMMMMMMM        **
**       MMM       MMM MMMMMMMM  MMMMMMMMM MMMMMMMM  MMM  MMM       MMM       **
**      MMM         MMM MMMMMM   MMMMMMMMM MMMMMMM   MMM MMM         MMM      **
**                                                                            **
**                                                                            **
** This file is part of the acedia library.                                   **
**                                                                            **
** 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, either version 3 of the License, or       **
** (at your option) any later version.                                        **
**                                                                            **
** 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 acedia. If not, see <http://www.gnu.org/licenses/>.             **
\*                                                                            */

//#define ACEDIA_REDUCED_ANNOUNCE

#include "acedia_util.hpp"
#include "acedia_network.hpp"

#include <stdio.h>
#include <stdlib.h>
#include <cstring>

#include <map>

#include "actor.hpp"
#include "actorproxy.hpp"
#include <boost/thread.hpp>

ACEDIA_DECLARE_CASE_CLASS((acedia)(Ping))

#include <iostream>
using std::cout;
using std::cerr;
using std::endl;

ACEDIA_DECLARE_CASE_TUPLE2((acedia)(details)(ProxyActorIds), originalActorId, boost::uint32_t, localProxyId, boost::uint32_t)
ACEDIA_DECLARE_CASE_TUPLE2((acedia)(details)(AddLinkToMap), localActor, boost::uint32_t, proxyActor, acedia::details::ProxyActorIds)
ACEDIA_DECLARE_CASE_TUPLE2((acedia)(details)(RemoveLinkFromMap), localActor, boost::uint32_t, proxyActor, acedia::details::ProxyActorIds)

namespace acedia
{
    PublishingResult::~PublishingResult() { }

    const char *asString(PublishingError pe)
    {
        switch (pe)
        {
            case NO_PUBLISHING_ERROR: return "NO_PUBLISHING_ERROR";
            case SERVER_SOCKET_ERROR:
                return "SERVER_SOCKET_ERROR (socket can't be created/opened)";
            case BINDING_ERROR:
                return "BINDING_ERROR (port already in use)";
            default: return "-?-";
        }
    }

    const char *PublishingResult::errorAsString() const throw()
    {
        return asString(error());
    }

    const char *asString(ConnectionError ce)
    {
        switch (ce)
        {
            case CONNECTION_SOCKET_ERROR:
                return "CONNECTION_SOCKET_ERROR (socket can't be created/opened)";

            case NO_SUCH_HOST_ERROR:
                return "NO_SUCH_HOST_ERROR (host name not found)";

            case COULD_NOT_CONNECT_TO_HOST_ERROR:
                return "COULD_NOT_CONNECT_TO_HOST_ERROR";

            case INCOMPATIBLE_ACEDIA_VERSIONS_ERROR:
                return "INCOMPATIBLE_ACEDIA_VERSIONS_ERROR";

            case INCOMPATIBLE_HOST_ERROR:
                return "INCOMPATIBLE_HOST_ERROR";

            default: return "-?-";
        }
    }

    namespace details
    {

        typedef std::map< ActorRef, std::list<ActorRef> > LinkMap;

        // send exit messages from all remote links to the local link
        void fakeExitFromAllDisconnectedLinks(LinkMap& links)
        {
            cerr << "fakeExitFromAllDisconnectedLinks" << endl;
            // oi = outer iterator
            for (LinkMap::iterator oi = links.begin(); oi != links.end(); ++oi)
            {
                ActorRef localActor = oi->first;
                if (localActor.isFarRef()) cerr << "LOCAL ACTOR IS A FAR REF!!" << endl;
                std::list<ActorRef>& remoteActors = oi->second;
                std::list<ActorRef>::iterator ii; // inner iterator
                for (ii = remoteActors.begin(); ii != remoteActors.end(); ++ii)
                {
                    ActorRef remoteActor = *ii;
                    Message msg(remoteActor, localActor,
                                Exit(exit_reasons::REMOTE_LINK_UNREACHABLE));
                    if (!remoteActor.isFarRef()) cerr << "REMOTE ACTOR IS NOT A FAR REF!!!" << endl;
                    cerr << "Sended a message from remote actor "
                         << remoteActor.realActorId()
                         << " to the local actor "
                         << localActor.actorId() << endl;
                    localActor.send(msg);
                }
            }
        }

        // A Mailman forwards messages to the PostOffice of the remote actor
        // and monitors links between local and remote links
        class Mailman : public acedia::Actor
        {

            LinkMap links;
            NativeSocketType m_socket;

            void addLinkToMap(const ActorRef& localActor, const ActorRef& remoteActor)
            {
                links[localActor].push_back(remoteActor);
            }

            void removeLinkFromMap(const ActorRef& localActor, const ActorRef& remoteActor)
            {
                links[localActor].remove(remoteActor);
            }

        protected:

            virtual bool filterIncomingMessage(const Message& msg)
            {
                return filterTimeoutEvents(msg);
            }

        public:

            virtual void linkTo(const ActorRef& who)
            {
                // do not feed m_links; PostOffice only needs single-side-links
                (void) (who.actor->backlinkTo(self()));
            }

            virtual bool backlinkTo(ActorRef&)
            {
                // do not handle backLink calls
                // PostOffice only needs single-side-links
                return true;
            }

            // trap exits
            Mailman(NativeSocketType socketDescr) :
                    Actor(HIDDEN, true), m_socket(socketDescr)
            {
            }

            virtual void act()
            {
                util::SocketIOStream sio(m_socket);
                Message msg;
                String str;
                Message pingMsg = Message(ActorRef(), ActorRef(), Ping());
                String serializedPing = pingMsg.serializeToString();
                try
                {
                    for (;;)
                    {
                        // send a ping every second to check connection
                        msg = receive();
//                      if (receiveWithin(msg, 1000))
                        {
                            // store all links of local actors to remote actors
                            if (msg.match<AddRemoteLink>())
                            {
                                addLinkToMap(msg.sender(), msg.receiver());
                            }
                            else if (msg.match<Exit>())
                            {
                                removeLinkFromMap(msg.sender(), msg.receiver());
                            }
                            else if (msg.match<AddLinkToMap>())
                            {
                                const AddLinkToMap& cmd = msg.uncheckedValueAt<AddLinkToMap>(0);
                                const ProxyActorIds& proxyIds = cmd.proxyActor();
                                ActorRef a1 = getInstanceById(cmd.localActor());
                                ActorRef a2 = getInstanceById(proxyIds.localProxyId());
                                addLinkToMap(a1, a2);
                                continue;
                            }
                            else if (msg.match<RemoveLinkFromMap>())
                            {
                                const RemoveLinkFromMap& cmd = msg.uncheckedValueAt<RemoveLinkFromMap>(0);
                                const ProxyActorIds& proxyIds = cmd.proxyActor();
                                ActorRef a1 = getInstanceById(cmd.localActor());
                                ActorRef a2 = getInstanceById(proxyIds.localProxyId());
                                removeLinkFromMap(a1, a2);
                                continue;
                            }
//                          cerr << "Mailman: " << msg.toString().const_str() << endl;
                            // forward the message
                            str = msg.serializeToString();
//cout << "SEND: " << str.const_str() << endl;
                            sio << (str);
                        }
//                      else
//                      {
//cout << "SEND: " << serializedPing.const_str() << endl;
//                          sio << serializedPing.const_str();
//                      }
                    }
                }
//              catch (acedia::Exception& e)
//              {
//                  cerr << e.what() << endl;
//              }
                catch (...)
                {
//                  cerr << "Mailman exited with an unknown exception!" << endl;
                }
//              closesocket(m_socket);
                sio.close();
                fakeExitFromAllDisconnectedLinks(links);
            }

        };

        // receives messages from a socket and forward the received messages
        // to the local actors
        class PostOffice : public acedia::Actor, public details::ProxyFactory
        {
            typedef std::map<boost::uint32_t, ActorRef> ProxyMap;

            NativeSocketType m_socket;
            ProxyMap m_proxys;
            ActorRef m_mailman;

        public:

            PostOffice(NativeSocketType socketDescr) :
                    Actor(HIDDEN, true),
                    m_socket(socketDescr)
            {
                m_mailman = spawn<Mailman>(socketDescr);
            }

            void act()
            {
                util::SocketIOStream sio(m_socket);
                try
                {
                    for (;;)
                    {
                        // read the serialized message
                        sio.readStringToBuffer();
//cout << "RECEIVED: " << sio.get_buffer() << endl;
                        Message msg = Message::deserializeFrom(sio.buffer(),
                                                               this);
                        // monitor links between local and remote actors
                        if (msg.match<AddRemoteLink>())
                        {
                            const ActorRef& a1 = msg.receiver();
                            const ActorRef& a2 = msg.sender();
                            send(m_mailman, AddLinkToMap(a1.actorId(), ProxyActorIds(a2.realActorId(), a2.actorId())));
                        }
                        else if (msg.match<Exit>())
                        {
                            const ActorRef& a1 = msg.receiver();
                            const ActorRef& a2 = msg.sender();
                            send(m_mailman, RemoveLinkFromMap(a1.actorId(), ProxyActorIds(a2.realActorId(), a2.actorId())));
                        }
                        // skip ping messages
                        else if (msg.match<Ping>()) continue;
//                      cerr << "PostOffice: " << msg.toString().const_str() << endl;
                        // forward message
                        msg.receiver().send(msg);
                        // yield READY in a detached actors does only check
                        // if we need to die because a link exited
                        yield(READY);
                    }
                }
//              catch (acedia::Exception& e)
//              {
//                  cerr << "PostOffice died with an acedia::Exception: " << e.what() << endl;
//              }
//              catch (std::exception& e)
//              {
//                  cerr << "PostOffice died with a std::exception: " << e.what() << endl;
//              }
                catch (...)
                {
//                  cerr << "PostOffice died with an unknown exception" << endl;
                }
//              closesocket(m_socket);
                sio.close();
            }

            ActorRef get(boost::uint32_t originalActorId)
            {
                if (originalActorId == 0) return ActorRef();
                else
                {
                    ProxyMap::iterator i = m_proxys.find(originalActorId);
                    if (i == m_proxys.end())
                    {
                        // create a new proxy
                        ActorRef proxy(new ActorProxy(originalActorId, m_mailman));
                        m_proxys.insert(ProxyMap::value_type(originalActorId, proxy));
                        return proxy;
                    }
                    else
                    {
                        // found a proxy in the map
                        return i->second;
                    }
                }
            }

        };

        class MiddleMan : public acedia::Actor
        {

            typedef std::map<int /* socket */, ActorRef /* post office */> ChildrenMap;

            NativeSocketType m_serverSocket;
            ChildrenMap m_children;

            ActorRef m_publishedActor;

        public:

            MiddleMan(NativeSocketType serverSocket, const ActorRef& publishedActor) : Actor(HIDDEN), m_serverSocket(serverSocket), m_publishedActor(publishedActor)
            {
            }

            void act()
            {
                sockaddr addr;
                socklen_t addrlen;
                for (;;)
                {
                    // wait for new connections
                    int newsockfd = accept(m_serverSocket, &addr, &addrlen);
                    if (newsockfd < 0)
                        ACEDIA_THROW(IOException, "Invalid socket accepted");
                    // check if the connected client is compatible
                    util::SocketIOStream sio(newsockfd);
                    boost::uint32_t major, minor;
                    std::string app_name;
                    sio >> major;
                    sio >> minor;
                    sio >> app_name;
                    if (acedia::details::isCompatible(app_name, major, minor))
                    {
cout << "Compatible client connected.\n";
                        // write ok
                        sio << (char) 1;
                        // write published actor id
                        sio << (boost::uint32_t) m_publishedActor.actorId();
                        // spawn PostOffice and add to children
                        ActorRef postOff = spawn<PostOffice>(newsockfd);
                        m_children.insert(ChildrenMap::value_type(newsockfd,
                                                                  postOff));
                    }
                    else
                    {
cout << "Client rejected.\n";
                        // reject client
                        sio << (char) 0;
                        sio.close();
//                      std::cerr << "client rejected\n";
//                      closesocket(newsockfd);
                    }
                }
            }

        };
    }

    PublishingResult publish(ActorRef who, boost::uint16_t port)
    {
        NativeSocketType sockfd;
//      socklen_t clilen;
        struct sockaddr_in serv_addr;//, cli_addr;
        sockfd = socket(AF_INET, SOCK_STREAM, 0);
        if (sockfd < 0)
        {
            return SERVER_SOCKET_ERROR;
        }
        memset((char*) &serv_addr, 0, sizeof(serv_addr));
        serv_addr.sin_family = AF_INET;
        serv_addr.sin_addr.s_addr = INADDR_ANY;
        serv_addr.sin_port = htons(port);
        if (bind(sockfd, (struct sockaddr*) &serv_addr, sizeof(serv_addr)) < 0)
        {
            return BINDING_ERROR;
        }
        listen(sockfd, 5);
//      clilen = sizeof(cli_addr);
        (void) spawn<details::MiddleMan>(sockfd, who);
        return NO_PUBLISHING_ERROR;
    }

    inline std::pair<ActorRef, ConnectionError> getResult(ConnectionError err)
    {
        return std::make_pair(ActorRef(), err);
    }

    inline std::pair<ActorRef, ConnectionError> getResult(ActorRef actorRef)
    {
        return std::make_pair(actorRef, NO_CONNECTION_ERROR);
    }

    std::pair<ActorRef, ConnectionError> remoteActor(const char *host, boost::uint16_t port)
    {
        NativeSocketType sockfd;
        struct sockaddr_in serv_addr;
        struct hostent *server;
        sockfd = socket(AF_INET, SOCK_STREAM, 0);
        if (sockfd < 0)
            return getResult(CONNECTION_SOCKET_ERROR);
        server = gethostbyname(host);
        if (server == NULL)
            return getResult(NO_SUCH_HOST_ERROR);
        memset((char *) &serv_addr, 0, sizeof(serv_addr));
        serv_addr.sin_family = AF_INET;
        (void) memmove((char *)&serv_addr.sin_addr.s_addr,
                       (char *)server->h_addr,
                       server->h_length);
        serv_addr.sin_port = htons(port);
        if (connect(sockfd, (const sockaddr*) &serv_addr,sizeof(serv_addr)) < 0)
            return getResult(COULD_NOT_CONNECT_TO_HOST_ERROR);
        // socket ready
        util::SocketIOStream sio(sockfd);
        // write major version
        sio << details::appMajorVersion();
        // write minor version
        sio << details::appMinorVersion();
        // write app name
        sio << details::appName();
        boost::uint8_t login_ok;
        sio >> login_ok;
//      cerr << "login_ok = " << (int) login_ok << endl;
        if (login_ok)
        {
            boost::uint32_t remote_actor_id;
            sio >> remote_actor_id;
            details::PostOffice *po = new details::PostOffice(sockfd);
            ActorRef result = po->get(remote_actor_id);
            details::doSpawn(po);
            return getResult(result);
        }
        else return getResult(INCOMPATIBLE_HOST_ERROR);
    }

} // namespace acedia

---