source/comm/internal/BinderSocket.cpp

   1 // ANNA - Anna is Not Nothingness Anymore                                                         //
   2 //                                                                                                //
   3 // (c) Copyright 2005-2015 Eduardo Ramos Testillano & Francisco Ruiz Rayo                         //
   4 //                                                                                                //
   5 // See project site at http://redmine.teslayout.com/projects/anna-suite                           //
   6 // See accompanying file LICENSE or copy at http://www.teslayout.com/projects/public/anna.LICENSE //
   7
   8
   9 #include <sys/time.h>
  10 #include <fcntl.h>
  11 #include <strings.h>
  12 #include <sys/types.h>
  13 #include <poll.h>
  14
  15 #ifdef __CYGWIN__
  16 #include <sys/uio.h>
  17 #endif
  18
  19 #include <anna/core/functions.hpp>
  20 #include <anna/core/tracing/Logger.hpp>
  21
  22 #include <anna/comm/internal/BinderSocket.hpp>
  23 #include <anna/comm/ServerSocket.hpp>
  24
  25 using namespace std;
  26 using namespace anna;
  27
  28 //-------------------------------------------------------------------------------------------------
  29 // No sabemos si tenemos que actuar como Cliente o Servidor; por lo que tenemos que establecer
  30 // ambos extremos del Socket.
  31 //-------------------------------------------------------------------------------------------------
  32 comm::BinderSocket::BinderSocket(comm::ServerSocket* serverSocket) :
  33   ClientSocket(
  34     string("/tmp/").append(serverSocket->getLocalAccessPoint().serialize()),
  35     Type::Stream
  36   ),
  37   a_serverSocket(*serverSocket) {
  38 }
  39
  40 //-------------------------------------------------------------------------------------------------
  41 // (1) Si consigue hacer el bind => es el primero. A partir de ahora dara acceso al resto.
  42
  43 // (0) a_bindDelay esta expresando en milisegundos => maxDelay estara en nanosegundos.
  44 // (0.1) Guardara los fds asociados al socket (real) y al socket por el que enviamos la informacion
  45 //     para poder hacer multiples binds sobre una misma direccion.
  46 // (0.2) Vamos a enviar 2 fds (el socket real y el socket UNIX).
  47 // (1) El Communicator tendra que tener en cuenta que se trata de un socket bind-compartido
  48 //     para meter el fd del publicador en la lista de descriptores por los que es posible que nos
  49 //     lleguen mensajes.
  50 // (2) Si falla el bind => cierra los recursos. Ya que nos llegara el nuevo fd atraves del mensaje
  51 //     del recvmsg.
  52 // (3) Conectamos como cliente con el socket UNIX => cuando la parte servidora (a traves del
  53 //     Communicator) detecte nuestra presencia deberia enviarnos la informacion para que podamos
  54 //     crear el socket con el bind-compartido. Limpiamos la direccion local para evitar hacer otro
  55 //      bind
  56 // (4) Si el 'connect' falla => se cierra el socket automaticamente => lo abre de nuevo.
  57 // (5) "Cierra" el socket UNIX ya que no volvera a usarlo. A partir de ahora el valor bueno sera
  58 //     el indicado en la estructura que nos pasa alguno de los servidores que estan escuchando
  59 //     el socket UNIX.
  60 //
  61 // Visto en: http://mail-index.netbsd.org/current-users/1998/01/16/0010.html
  62 //-------------------------------------------------------------------------------------------------
  63 void comm::BinderSocket::requestBind(const struct sockaddr* s, const  int len)
  64 throw(RuntimeException) {
  65   if(::bind(a_serverSocket.a_fd, s, len) != -1) {                // (1)
  66     unlink(getRemoteAccessPoint().getPath().c_str());
  67     this->bind();
  68     anna_socket_assert(::listen(a_fd, 10) == -1, "Error preparing listen access point");
  69     return;
  70   }
  71
  72   const int xerrno = errno;
  73
  74   a_serverSocket.close();                                        // (2)
  75
  76   if(xerrno != EADDRINUSE)
  77     throw RuntimeException(asString(), xerrno, ANNA_FILE_LOCATION);
  78
  79   this->a_localAccessPoint.clear();                           // (3)
  80   this->connect();                                               // (4)
  81   waitBind(a_serverSocket.getBindDelay());
  82 }
  83
  84 //-------------------------------------------------------------------------------------------------
  85 // Este metodo se invoca desde el Communicator cuando se detecta activada en el socket UNIX
  86 // empleado para comunicar todos los procesos que quieran compartir una determinada direccion.
  87 //
  88 // (1) Acepta la conexion por el socket UNIX. Ver (3) de sharedBind.
  89 // (2) En el IOV debemos mandar al menos 1 byte.
  90 // (3) Rellenamos la estructura que recojera algun otro proceso que esta esperando a leer del
  91 //     socket UNIX.
  92 // (4) Cierra la nueva conexion ya que no vamos a volver a usarla para nada.
  93 //-------------------------------------------------------------------------------------------------
  94 void comm::BinderSocket::responseBind()
  95 throw(RuntimeException) {
  96   struct iovec iov;
  97   struct msghdr mh;
  98   int fds [2];
  99   int garbage(0);
 100   sockaddr_un s;
 101   socklen_t len(sizeof(sockaddr_un));
 102   anna_memset(&s, 0, sizeof(sockaddr_un));
 103   int newSocket = ::accept(a_fd, (sockaddr*) & s, &len);      // (1)
 104
 105   if(newSocket == -1) {
 106     const int xerrno = errno;
 107     throw RuntimeException(asString(), xerrno, ANNA_FILE_LOCATION);
 108   }
 109
 110   iov.iov_len = sizeof(garbage);                                                   // (2)
 111   iov.iov_base = (char*) & garbage;
 112   mh.msg_iov = &iov;
 113   mh.msg_iovlen = 1;
 114   mh.msg_name = NULL;
 115   mh.msg_namelen = 0;
 116   fds [0] = a_serverSocket.a_fd;                                                  // (3)
 117   fds [1] = a_fd;
 118 #ifndef __sun
 119   struct cmsghdr cmh;
 120   char buffer [sizeof(struct cmsghdr) + sizeof(fds)];
 121   mh.msg_control = &buffer;
 122   mh.msg_controllen = cmh.cmsg_len = sizeof(struct cmsghdr) + sizeof(fds);
 123   cmh.cmsg_level = SOL_SOCKET;
 124   cmh.cmsg_type = SCM_RIGHTS;
 125   anna_memcpy(buffer, &cmh, sizeof(cmh));
 126   anna_memcpy(buffer + sizeof(struct cmsghdr), fds, sizeof(fds));
 127   mh.msg_flags = 0;
 128 #else
 129   mh.msg_accrights = (caddr_t) fds;
 130   mh.msg_accrightslen = sizeof(fds);
 131 #endif
 132
 133   if(sendmsg(newSocket, &mh, 0) < 0) {
 134     int xerrno = errno;
 135     ::close(newSocket);
 136     throw RuntimeException(asString(), xerrno, ANNA_FILE_LOCATION);
 137   }
 138
 139   ::close(newSocket);                                                              // (4)
 140   LOGDEBUG(
 141     string msg("anna::comm::BinderSocket::responseBind | ");
 142     msg += functions::asText("ServerSocket's fd: ", fds [0]);
 143     msg += functions::asText(" | BinderSocket's fd: ", fds [1]);
 144     Logger::debug(msg, ANNA_FILE_LOCATION);
 145   );
 146 }
 147
 148 void comm::BinderSocket::waitBind(const Millisecond &maxDelay)
 149 throw(RuntimeException) {
 150   struct msghdr mh;
 151   struct iovec iov;
 152   const int garbage(0);
 153   int fds [2];
 154   iov.iov_len = sizeof(garbage);
 155   iov.iov_base = (char*) & garbage;
 156   mh.msg_iov = &iov;
 157   mh.msg_iovlen = 1;
 158   mh.msg_name = NULL;
 159   mh.msg_namelen = 0;
 160 #ifndef __sun
 161   char buffer [sizeof(struct cmsghdr) + sizeof(fds)];
 162   mh.msg_control = &buffer;
 163   mh.msg_controllen = sizeof(struct cmsghdr) + sizeof(fds);
 164   mh.msg_flags = 0;
 165 #else
 166   mh.msg_accrights = (caddr_t) fds;
 167   mh.msg_accrightslen = sizeof(fds);
 168 #endif
 169   pollfd poll;
 170   int r;
 171   poll.fd = a_fd;
 172   poll.events = POLLIN | POLLRDNORM;
 173
 174   if(::poll(&poll, 1, maxDelay) != 1) {
 175     string msg(a_serverSocket.asString());
 176     msg += " | No data is received from shared bind";
 177     throw RuntimeException(msg, ANNA_FILE_LOCATION);
 178   }
 179
 180   r = recvmsg(a_fd, &mh, 0);
 181
 182   if(r <= 0) {
 183     const int xerrno = errno;
 184     std::string msg(a_serverSocket.asString());
 185     msg += " | ";
 186     msg += asString();
 187
 188     if(r == -1) {
 189       msg += " | Error in shared bind reception";
 190       throw RuntimeException(msg, xerrno, ANNA_FILE_LOCATION);
 191     } else if(r == 0) {
 192       msg += " | No item received to obtain shared bind";
 193       throw RuntimeException(msg, ANNA_FILE_LOCATION);
 194     }
 195   }
 196
 197   close();
 198 #ifndef __sun
 199   anna_memcpy(&fds, buffer + sizeof(cmsghdr), sizeof(fds));
 200 #endif
 201   a_serverSocket.a_fd = fds [0];
 202   a_fd = fds [1];
 203   LOGDEBUG(
 204     string msg("anna::comm::BinderSocket::waitBind | ");
 205     msg += functions::asText("ServerSocket's fd: ", fds [0]);
 206     msg += functions::asText(" | BinderSocket's fd: ", fds [1]);
 207     Logger::debug(msg, ANNA_FILE_LOCATION);
 208   );
 209 }
 210