src/xport_udp.c

   1 /*======================================================================*
   2  * Copyright (C) 2008 Light Weight Event System                         *
   3  * All rights reserved.                                                 *
   4  *                                                                      *
   5  * This program is free software; you can redistribute it and/or modify *
   6  * it under the terms of the GNU General Public License as published by *
   7  * the Free Software Foundation; either version 2 of the License, or    *
   8  * (at your option) any later version.                                  *
   9  *                                                                      *
  10  * This program is distributed in the hope that it will be useful,      *
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of       *
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the        *
  13  * GNU General Public License for more details.                         *
  14  *                                                                      *
  15  * You should have received a copy of the GNU General Public License    *
  16  * along with this program; if not, write to the Free Software          *
  17  * Foundation, Inc., 51 Franklin Street, Fifth Floor,                   *
  18  * Boston, MA 02110-1301 USA.                                           *
  19  *======================================================================*/
  20
  21 #include "config.h"
  22
  23 #include "xport.h"
  24 #include "xport_udp.h"
  25
  26 #include "perror.h"
  27 #include "opt.h"
  28
  29 #include <arpa/inet.h>
  30 #include <netinet/in.h>
  31 #include <stdlib.h>
  32 #include <sys/socket.h>
  33 #include <sys/types.h>
  34 #include <unistd.h>
  35
  36 struct priv {
  37     int                   fd;
  38
  39     struct in_addr        address;
  40     short                 port;    /* In network byte order. */
  41     char*                 iface;
  42
  43     struct sockaddr_in    ip_addr;
  44     struct ip_mreq        mreq;
  45
  46     int                   join_group;
  47
  48     int                   bind;
  49     int                   joined_multi_group;
  50 };
  51
  52
  53 static void destructor (struct xport* this_xport)
  54 {
  55   struct priv* ppriv = (struct priv*)this_xport->priv;
  56
  57   free (ppriv->iface);
  58   free (ppriv);
  59
  60   this_xport->vtbl = 0;
  61   this_xport->priv = 0;
  62 }
  63
  64
  65 static int xopen (struct xport* this_xport, int flags)
  66 {
  67   char optval = arg_ttl ; // Configured TTL
  68   struct priv* ppriv = (struct priv*)this_xport->priv;
  69   (void)flags; /* appease -Wall -Werror */
  70
  71   memset(&ppriv->ip_addr, 0, sizeof(ppriv->ip_addr));
  72
  73   ppriv->ip_addr.sin_family = AF_INET;
  74   ppriv->ip_addr.sin_addr = ppriv->address;
  75   ppriv->ip_addr.sin_port = ppriv->port;
  76
  77   ppriv->mreq.imr_multiaddr = ppriv->ip_addr.sin_addr;
  78
  79   if ( ppriv->iface && ppriv->iface[0] )
  80     {
  81       ppriv->mreq.imr_interface.s_addr = inet_addr(ppriv->iface);
  82     }
  83   else
  84     {
  85       ppriv->mreq.imr_interface.s_addr = htonl(INADDR_ANY);
  86     }
  87
  88   if ( (ppriv->fd = socket(AF_INET, SOCK_DGRAM, 0)) < 0 )
  89     {
  90       PERROR("socket");
  91       return -1;
  92     }
  93   setsockopt(ppriv->fd, SOL_IP, IP_MULTICAST_TTL, &optval, 1) ;
  94
  95   /* If the interface is specified, then we set the interface. */
  96   if ( ppriv->iface && ppriv->iface[0] )
  97     {
  98       struct in_addr addr;
  99
 100       addr.s_addr = inet_addr(ppriv->iface);
 101       if ( setsockopt(ppriv->fd, IPPROTO_IP, IP_MULTICAST_IF,
 102                       &addr, sizeof(addr)) < 0 )
 103         {
 104           PERROR("can't set interface");
 105           return -1;
 106         }
 107     }
 108
 109   return 0;
 110 }
 111
 112 static int xclose (struct xport* this_xport)
 113 {
 114   struct priv* ppriv = (struct priv*)this_xport->priv;
 115
 116   if ( -1 == ppriv->fd )
 117     {
 118       LOG_ER("Close transport called with xport already closed.\n");
 119       return -1;
 120     }
 121
 122   if ( ppriv->joined_multi_group )
 123     {
 124       /* remove ourselves from the multicast channel */
 125       if ( setsockopt(ppriv->fd, IPPROTO_IP, IP_DROP_MEMBERSHIP,
 126                       &ppriv->mreq, sizeof(ppriv->mreq)) < 0 )
 127         {
 128           PERROR("can't drop multicast group");
 129         }
 130     }
 131
 132   close(ppriv->fd);
 133   ppriv->fd = -1;
 134
 135   return 0;
 136 }
 137
 138 static int xread (struct xport* this_xport, void* buf, size_t count,
 139                   unsigned long* addr, short* port)
 140 {
 141   struct priv* ppriv = (struct priv*)this_xport->priv;
 142   int recvfrom_ret;
 143
 144   struct sockaddr_in sender_ip_addr;
 145   socklen_t          sender_ip_socket_size;
 146   struct sockaddr_in ip_addr;
 147
 148   /* Set the size for use below: */
 149   sender_ip_socket_size = (socklen_t)sizeof(sender_ip_addr);
 150
 151   /* This should be done only with multi-cast addresses. */
 152   if ( ! ppriv->bind )
 153     {
 154       int on = 1;
 155
 156       LOG_PROG("About to join multi-cast group.\n");
 157
 158       /* Set address for reuse. */
 159       if ( setsockopt(ppriv->fd, SOL_SOCKET, SO_REUSEADDR,
 160                       &on, sizeof(on)) < 0 )
 161         {
 162           PERROR("setsockopt -- can't reuse address\n");
 163           return -1;
 164         }
 165
 166       /* Set the receive buffer size. */
 167       if ( arg_sockbuffer )
 168         {
 169           if ( setsockopt(ppriv->fd, SOL_SOCKET, SO_RCVBUF,
 170                           (char*)&arg_sockbuffer, sizeof(arg_sockbuffer)) < 0 )
 171             {
 172               PERROR("Setsockopt:SO_RCVBUF");
 173             }
 174           /* We'll try to continue with the default buffer size. */
 175         }
 176
 177       /* Bind the socket to the port. */
 178       memset(&ip_addr, 0, sizeof(ip_addr));
 179       ip_addr.sin_family = AF_INET;
 180       ip_addr.sin_addr.s_addr = htonl(INADDR_ANY);
 181       ip_addr.sin_port = ppriv->port;
 182
 183       if ( bind(ppriv->fd, (struct sockaddr*)&ip_addr, sizeof(ip_addr) ) < 0 )
 184         {
 185           PERROR("can't bind to local address\n");
 186           return -1;
 187         }
 188       ppriv->bind = 1;
 189
 190       if ( ppriv->join_group && !ppriv->joined_multi_group )
 191         {
 192           /* add the multicast channel given */
 193           if ( setsockopt(ppriv->fd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
 194                           &ppriv->mreq, sizeof(ppriv->mreq)) < 0 )
 195             {
 196               PERROR("setsockopt can't add to multicast group\n");
 197               return -1;
 198             }
 199           ppriv->joined_multi_group = 1;
 200         }
 201     }
 202
 203   LOG_PROG("about to call recvfrom().\n");
 204
 205   if ( (recvfrom_ret = recvfrom(ppriv->fd, buf, count, 0,
 206                                 (struct sockaddr*)&sender_ip_addr,
 207                                 &sender_ip_socket_size)) < 0 )
 208     {
 209       switch ( errno )
 210         {
 211           default:
 212             PERROR("recvfrom");
 213
 214           case EINTR:                   /* Quiet return on interrupt. */
 215             break;
 216         }
 217     }
 218
 219   *addr = sender_ip_addr.sin_addr.s_addr;
 220   *port = ntohs(sender_ip_addr.sin_port);
 221
 222   return recvfrom_ret;
 223 }
 224
 225
 226 static int xwrite (struct xport* this_xport, const void* buf, size_t count)
 227 {
 228   struct priv* ppriv = (struct priv*)this_xport->priv;
 229
 230   LOG_PROG("about to call sendto().\n");
 231
 232   return sendto (ppriv->fd, buf, count, 0,
 233                  (struct sockaddr*)&ppriv->ip_addr, sizeof(ppriv->ip_addr));
 234 }
 235
 236
 237 int xport_udp_ctor (struct xport* this_xport,
 238                     const char*   address,
 239                     const char*   iface,
 240                     short         port,
 241                     int           join)
 242 {
 243   static struct xport_vtbl vtbl = {
 244       destructor,
 245
 246       xopen, xclose,
 247       xread, xwrite
 248   };
 249
 250   struct priv* ppriv;
 251
 252   this_xport->vtbl = 0;
 253   this_xport->priv = 0;
 254
 255   ppriv = (struct priv*)malloc(sizeof(*ppriv));
 256   if ( 0 == ppriv )
 257     {
 258       LOG_ER("Failed to allocate %d bytes for xport data.\n", sizeof(*ppriv));
 259       return -1;
 260     }
 261   memset(ppriv, 0, sizeof(*ppriv));
 262
 263   ppriv->fd = -1;
 264   if ( 0 == inet_aton(address, &ppriv->address) )
 265     {
 266       LOG_ER("invalid address \"%s\"\n", address);
 267       free(ppriv);
 268       return -1;
 269     }
 270
 271   ppriv->port = htons(port);
 272
 273   ppriv->iface = strdup(iface);
 274   if ( 0 == ppriv->iface )
 275     {
 276       LOG_ER("strdup failed attempting to allocate %d bytes\n", strlen(iface));
 277       free(ppriv);
 278       return -1;
 279     }
 280
 281   ppriv->join_group = join;
 282
 283   this_xport->vtbl = &vtbl;
 284   this_xport->priv = ppriv;
 285
 286   return 0;
 287 }