src/proto_inet.c

   1 /**************************************************************************/
   2 /* ---------------------------------------------------------------------- */
   3 /* Copyright (C) 2006 - 2008 FRESCOR consortium partners:                 */
   4 /*                                                                        */
   5 /*   Universidad de Cantabria,              SPAIN                         */
   6 /*   University of York,                    UK                            */
   7 /*   Scuola Superiore Sant'Anna,            ITALY                         */
   8 /*   Kaiserslautern University,             GERMANY                       */
   9 /*   Univ. Politécnica  Valencia,           SPAIN                        */
  10 /*   Czech Technical University in Prague,  CZECH REPUBLIC                */
  11 /*   ENEA                                   SWEDEN                        */
  12 /*   Thales Communication S.A.              FRANCE                        */
  13 /*   Visual Tools S.A.                      SPAIN                         */
  14 /*   Rapita Systems Ltd                     UK                            */
  15 /*   Evidence                               ITALY                         */
  16 /*                                                                        */
  17 /*   See http://www.frescor.org for a link to partners' websites          */
  18 /*                                                                        */
  19 /*          FRESCOR project (FP6/2005/IST/5-034026) is funded             */
  20 /*       in part by the European Union Sixth Framework Programme          */
  21 /*       The European Union is not liable of any use that may be          */
  22 /*       made of this code.                                               */
  23 /*                                                                        */
  24 /*                                                                        */
  25 /*  This file is part of FORB (Frescor Object Request Broker)             */
  26 /*                                                                        */
  27 /* FORB is free software; you can redistribute it and/or modify it        */
  28 /* under terms of the GNU General Public License as published by the      */
  29 /* Free Software Foundation; either version 2, or (at your option) any    */
  30 /* later version.  FORB is distributed in the hope that it will be        */
  31 /* useful, but WITHOUT ANY WARRANTY; without even the implied warranty    */
  32 /* of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU    */
  33 /* General Public License for more details. You should have received a    */
  34 /* copy of the GNU General Public License along with FORB; see file       */
  35 /* COPYING. If not, write to the Free Software Foundation, 675 Mass Ave,  */
  36 /* Cambridge, MA 02139, USA.                                              */
  37 /*                                                                        */
  38 /* As a special exception, including FORB header files in a file,         */
  39 /* instantiating FORB generics or templates, or linking other files       */
  40 /* with FORB objects to produce an executable application, does not       */
  41 /* by itself cause the resulting executable application to be covered     */
  42 /* by the GNU General Public License. This exception does not             */
  43 /* however invalidate any other reasons why the executable file might be  */
  44 /* covered by the GNU Public License.                                     */
  45 /**************************************************************************/
  46
  47 #include "proto.h"
  48 #include <arpa/inet.h>
  49 #include <dirent.h>
  50 #include <fcntl.h>
  51 #include <forb/proto_inet.h>
  52 #include <net/if.h>
  53 #include <netinet/in.h>
  54 #include <stdio.h>
  55 #include <sys/epoll.h>
  56 #include <sys/ioctl.h>
  57 #include <sys/socket.h>
  58 #include <sys/types.h>
  59 #include <ul_log.h>
  60 #include <unistd.h>
  61
  62 /**
  63  * @file   proto_inet.c
  64  * @author Michal Sojka <sojkam1@fel.cvut.cz>
  65  * @date   Sun Oct 12 16:10:23 2008
  66  *
  67  * @brief  FORB transport protocol based on INET family sockets.
  68  *
  69  * UDP is used for broadcasts and TCP for requests/replies. There
  70  * exist two uni-drectional connections between any two communicating
  71  * peers.
  72  */
  73
  74 extern UL_LOG_CUST(ulogd_forb_proto_inet);
  75
  76 #define PORT 15514
  77 #define MCAST_ADDR "225.15.5.14"
  78
  79 /** Address used by inet protocol. */
  80 struct inet_addr {
  81         struct in_addr addr;
  82 };
  83
  84 /** INET protocol data for ports. */
  85 struct inet_port {
  86         int udp_socket;         /**< Socket for sending and receiving broadcasts */
  87         int listen_socket;      /*  */
  88         int epoll_fd;           /**< File descriptor used by epoll() in inet_receive(). */
  89         struct inet_addr addr;  /**< Address of this port */
  90         int last_recv_fd;       /**< Used in inet_recv() to read data longer than receiving buffer */
  91         struct in_addr multicast_addr;
  92 };
  93
  94 /** INET protocol data associated with every peer */
  95 struct inet_peer {
  96         int socket;             /**< Connected socket to the peer */
  97 };
  98
  99 /* static struct inet_port* */
 100 /* peer_to_inet_port(forb_peer_t *peer) { return peer->port->desc.proto_priv; } */
 101
 102 static CORBA_boolean
 103 inet_serialize_addr(FORB_CDR_Codec *codec, const void *addr)
 104 {
 105         const struct inet_addr *a = addr;
 106         CORBA_long haddr = ntohl(a->addr.s_addr);
 107         return CORBA_long_serialize(codec, &haddr);
 108 }
 109
 110 static CORBA_boolean
 111 inet_deserialize_addr(FORB_CDR_Codec *codec, void **addr)
 112 {
 113         struct inet_addr *a;
 114         CORBA_long s_addr;
 115         CORBA_boolean ret;
 116
 117         a = forb_malloc(sizeof(*a));
 118         if (!a)
 119                 return CORBA_FALSE;
 120         ret = CORBA_long_deserialize(codec, &s_addr);
 121         a->addr.s_addr = htonl(s_addr);
 122         *addr = a;
 123         return ret;
 124 }
 125
 126 static struct inet_peer *
 127 inet_connect(forb_peer_t *peer)
 128 {
 129         struct inet_peer *ipeer;
 130         struct sockaddr_in sa;
 131         struct inet_addr *addr = peer->addr;
 132         int ret;
 133
 134         if (!addr)
 135                 return NULL;
 136         ipeer = forb_malloc(sizeof(*ipeer));
 137         if (!ipeer)
 138                 goto err;
 139         ipeer->socket = socket(PF_INET, SOCK_STREAM, 0);
 140         if (!ipeer->socket)
 141                 goto err_free;
 142         sa.sin_family = AF_INET;
 143         sa.sin_port = htons(PORT);
 144         sa.sin_addr = addr->addr;
 145         ret = connect(ipeer->socket, (struct sockaddr*)&sa, sizeof(sa));
 146         if (ret)
 147                 goto err_close;
 148
 149         return ipeer;
 150 err_close:
 151         close(ipeer->socket);
 152 err_free:
 153         forb_free(ipeer);
 154 err:
 155         return NULL;
 156
 157 }
 158
 159 static size_t
 160 inet_send(forb_peer_t *peer, const void *buf, size_t len)
 161 {
 162         struct inet_peer *ipeer = peer->proto_priv;
 163
 164         if (!ipeer) {
 165                 ipeer = inet_connect(peer);
 166                 if (!ipeer)
 167                         return -1;
 168                 peer->proto_priv = ipeer;
 169
 170         }
 171
 172         return send(ipeer->socket, buf, len, 0);
 173 }
 174
 175 /*----------------------------------------------------------------------
 176  Portable function to set a socket into nonblocking mode.
 177  Calling this on a socket causes all future read() and write() calls on
 178  that socket to do only as much as they can immediately, and return
 179  without waiting.
 180  If no data can be read or written, they return -1 and set errno
 181  to EAGAIN (or EWOULDBLOCK).
 182  Thanks to Bjorn Reese for this code.
 183 ----------------------------------------------------------------------*/
 184 int setnonblocking(int fd)
 185 {
 186     int flags;
 187
 188     /* If they have O_NONBLOCK, use the Posix way to do it */
 189 #if defined(O_NONBLOCK)
 190     /* Fixme: O_NONBLOCK is defined but broken on SunOS 4.1.x and AIX 3.2.5. */
 191     if (-1 == (flags = fcntl(fd, F_GETFL, 0)))
 192         flags = 0;
 193     return fcntl(fd, F_SETFL, flags | O_NONBLOCK);
 194 #else
 195     /* Otherwise, use the old way of doing it */
 196     flags = 1;
 197     return ioctl(fd, FIOBIO, &flags);
 198 #endif
 199 }
 200
 201 int inet_accept_connection(struct inet_port *p)
 202 {
 203         int client;
 204         struct sockaddr_in addr;
 205         socklen_t addrlen;
 206         struct epoll_event ev;
 207         int ret;
 208
 209         client = accept(p->listen_socket, (struct sockaddr *) &addr, &addrlen);
 210         if (client < 0){
 211                 //perror("accept");
 212                 return -1;
 213         }
 214         ret = setnonblocking(client);
 215         if (ret) {
 216                 close(client);
 217                 return -1;
 218         }
 219         ev.events = EPOLLIN | EPOLLET;
 220         ev.data.fd = client;
 221         return epoll_ctl(p->epoll_fd, EPOLL_CTL_ADD, client, &ev);
 222 }
 223
 224 static size_t
 225 inet_recv(forb_port_t *port, void *buf, size_t len)
 226 {
 227         struct inet_port *iport = port->desc.proto_priv;
 228 #if 1
 229         struct epoll_event ev;
 230         int ret, nfds;
 231         for (;;) {
 232                 if (iport->last_recv_fd == -1) {
 233                         nfds = epoll_wait(iport->epoll_fd, &ev, 1, -1);
 234                         if (nfds < 1)
 235                                 return -1;
 236                         if (ev.data.fd == iport->listen_socket) {
 237                                 ret = inet_accept_connection(iport);
 238                                 if (ret)
 239                                         return -1;
 240                                 else
 241                                         continue;
 242                         } else {
 243                                 iport->last_recv_fd = ev.data.fd;
 244                         }
 245                 }
 246                 ret = recv(iport->last_recv_fd, buf, len, 0);
 247                 if (ret == -1 && errno == EAGAIN) {
 248                         iport->last_recv_fd = -1;
 249                         continue;
 250                 }
 251                 return ret;
 252         }
 253 #else
 254         return recv(iport->udp_socket, buf, len, 0);
 255 #endif
 256 }
 257
 258 static int
 259 inet_port_destroy(forb_port_t * port)
 260 {
 261         struct inet_port *pd = port->desc.proto_priv;
 262         close(pd->epoll_fd);
 263         close(pd->udp_socket);
 264         close(pd->listen_socket);
 265         forb_free(pd);
 266         return 0;
 267 }
 268
 269 static size_t
 270 inet_broadcast(forb_port_t *port, const void *buf, size_t len)
 271 {
 272         struct inet_port *p = port->desc.proto_priv;
 273         struct sockaddr_in addr;
 274         int ret;
 275
 276         addr.sin_family = AF_INET;
 277         addr.sin_port = htons(PORT);
 278         addr.sin_addr = p->multicast_addr;
 279
 280         ret = sendto(p->udp_socket, buf, len, 0,
 281                      (struct sockaddr*)&addr, sizeof(addr));
 282         return ret;
 283 }
 284
 285 static void
 286 inet_peer_destroy(forb_peer_t *peer)
 287 {
 288         struct inet_peer *ipeer = peer->proto_priv;
 289         peer->proto_priv = NULL;
 290         close(ipeer->socket);
 291         free(ipeer);
 292 }
 293
 294 static const forb_proto_t proto_inet = {
 295         .hello_interval = 40 /* seconds */,
 296         .port_destroy = inet_port_destroy,
 297         .peer_destroy = inet_peer_destroy,
 298         .send = inet_send,
 299         .recv = inet_recv,
 300         .broadcast = inet_broadcast,
 301         .serialize_addr = inet_serialize_addr,
 302         .deserialize_addr = inet_deserialize_addr,
 303 };
 304
 305 #define MAX_INTERFACES 10
 306 int get_local_address(int sock, struct inet_addr *addr)
 307 {
 308         struct ifconf  ifc;
 309         struct ifreq   *ifr, req[MAX_INTERFACES];
 310
 311
 312         ifc.ifc_len = sizeof(req);
 313         ifc.ifc_req = req;
 314
 315         if (ioctl(sock, SIOCGIFCONF, &ifc) < 0)
 316                 return -1;
 317         for (ifr = req; ifr < &req[MAX_INTERFACES]; ifr++) {
 318                 struct sockaddr_in ia;
 319                 memcpy(&ia, &ifr->ifr_addr, sizeof(ia));
 320                 ioctl(sock, SIOCGIFFLAGS, ifr);
 321                 if ((ifr->ifr_flags & IFF_UP) && !(ifr->ifr_flags & IFF_LOOPBACK)) {
 322                         addr->addr = ia.sin_addr;
 323                         return 0;
 324                 }
 325         }
 326         return -1;
 327 }
 328
 329 /**
 330  * Initializes INET protocol port.
 331  *
 332  * @param port_desc Port description to initialize.
 333  * @return Zero on success, -1 on error and errno is set
 334  * appropriately.
 335  */
 336 int
 337 forb_inet_port_init(struct forb_port_desc *port_desc, struct in_addr listen_on)
 338 {
 339
 340         int ret;
 341         struct inet_port *port_priv;
 342         struct sockaddr_in addr;
 343         socklen_t len;
 344         struct epoll_event ev;
 345
 346         port_priv = forb_malloc(sizeof(*port_priv));
 347         if (!port_priv)
 348                 return -1;
 349
 350         memset(port_priv, 0, sizeof(*port_priv));
 351         port_priv->last_recv_fd = -1;
 352
 353         port_priv->udp_socket = socket(PF_INET, SOCK_DGRAM, 0);
 354         if (port_priv->udp_socket == -1) goto err_free;
 355         port_priv->listen_socket = socket(PF_INET, SOCK_STREAM, 0);
 356         if (port_priv->listen_socket == -1) goto err_close_udp;
 357
 358         int yes = 1;
 359         ret = setsockopt(port_priv->udp_socket, SOL_SOCKET, SO_BROADCAST, &yes, sizeof(yes));
 360         if (ret)
 361                 goto err_close_listen;
 362
 363         addr.sin_family = AF_INET;
 364         addr.sin_port = htons(PORT);
 365         addr.sin_addr = listen_on;
 366
 367         ret = bind(port_priv->listen_socket, (struct sockaddr *)&addr, sizeof(addr));
 368         if (ret != 0) goto err_close_listen;
 369         if (setnonblocking(port_priv->listen_socket))
 370                 goto err_close_listen;
 371
 372         ret = bind(port_priv->udp_socket, (struct sockaddr *)&addr, sizeof(addr));
 373         if (ret != 0) goto err_close_listen;
 374         setnonblocking(port_priv->udp_socket);
 375         inet_aton(MCAST_ADDR, &port_priv->multicast_addr);
 376         struct ip_mreq mreq;
 377         mreq.imr_multiaddr = port_priv->multicast_addr;
 378         mreq.imr_interface.s_addr = INADDR_ANY;
 379         ret = setsockopt(port_priv->udp_socket, IPPROTO_IP,
 380                          IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
 381         if (ret)
 382                 goto err_close_listen;
 383
 384         char loop = 1;
 385         unsigned loop_size = sizeof(loop);
 386         ret = setsockopt(port_priv->udp_socket, IPPROTO_IP,
 387                          IP_MULTICAST_LOOP, &loop, loop_size);
 388         if (ret)
 389                 goto err_close_listen;
 390
 391
 392         ret = listen(port_priv->listen_socket, 10);
 393         if (ret)
 394                 goto err_close_listen;
 395
 396         len = sizeof(addr);
 397         ret = getsockname(port_priv->listen_socket, (struct sockaddr *)&addr, &len);
 398         if (ret) {
 399                 ul_logerr("Non-loopback inet address not found\n");
 400                 goto err_close_listen;
 401         }
 402
 403         port_priv->epoll_fd = epoll_create(10);
 404         if (port_priv->epoll_fd == -1)
 405                 goto err_close_listen;
 406
 407         memset(&ev, 0, sizeof(ev));
 408         ev.events = EPOLLIN | EPOLLET;
 409         ev.data.fd = port_priv->listen_socket;
 410         ret = epoll_ctl(port_priv->epoll_fd, EPOLL_CTL_ADD,
 411                         port_priv->listen_socket, &ev);
 412         if (ret)
 413                 goto err_close_epoll;
 414
 415         ev.events = EPOLLIN | EPOLLET;
 416         ev.data.fd = port_priv->udp_socket;
 417         ret = epoll_ctl(port_priv->epoll_fd, EPOLL_CTL_ADD,
 418                         port_priv->udp_socket, &ev);
 419         if (ret)
 420                 goto err_close_epoll;
 421
 422         if (listen_on.s_addr == INADDR_ANY)
 423                 get_local_address(port_priv->listen_socket, &port_priv->addr);
 424         else
 425                 port_priv->addr.addr = listen_on;
 426
 427
 428         port_desc->proto = &proto_inet;
 429         port_desc->proto_priv = port_priv;
 430         port_desc->addr = &port_priv->addr;
 431         return 0;
 432 err_close_epoll:
 433         close(port_priv->epoll_fd);
 434 err_close_listen:
 435         ret = errno;
 436         close(port_priv->listen_socket);
 437         errno = ret;
 438 err_close_udp:
 439         ret = errno;
 440         close(port_priv->udp_socket);
 441         errno = ret;
 442 err_free:
 443         ret = errno;
 444         forb_free(port_priv);
 445         errno = ret;
 446         return -1;
 447 }