udp_send.c

   1 /*
   2  * Copyright (C) 2005-2009 Andre Noll <maan@systemlinux.org>
   3  *
   4  * Licensed under the GPL v2. For licencing details see COPYING.
   5  */
   6
   7 /** \file udp_send.c Para_server's udp sender. */
   8
   9
  10 #include <sys/time.h>
  11 #include <dirent.h>
  12
  13 #include "server.cmdline.h"
  14 #include "para.h"
  15 #include "error.h"
  16 #include "string.h"
  17 #include "afh.h"
  18 #include "afs.h"
  19 #include "server.h"
  20 #include "vss.h"
  21 #include "list.h"
  22 #include "send.h"
  23 #include "portable_io.h"
  24 #include "udp_header.h"
  25 #include "net.h"
  26 #include "fd.h"
  27 #include "sched.h"
  28 #include "close_on_fork.h"
  29 #include "chunk_queue.h"
  30
  31 /** Convert in_addr to ascii. */
  32 #define TARGET_ADDR(oc) inet_ntoa((oc)->addr)
  33
  34 /** Describes one entry in the list of targets for the udp sender. */
  35 struct udp_target {
  36         /** Address info. */
  37         struct in_addr addr;
  38         /** The position of this target in the list of targets. */
  39         struct list_head node;
  40         /** The UDP port. */
  41         int port;
  42         /** The socket fd. */
  43         int fd;
  44         /** The list of queued chunks for this fd. */
  45         struct chunk_queue *cq;
  46 };
  47
  48 static struct list_head targets;
  49 static int sender_status;
  50
  51 static void udp_close_target(struct udp_target *ut)
  52 {
  53         if (ut->fd < 0)
  54                 return;
  55         close(ut->fd);
  56         del_close_on_fork_list(ut->fd);
  57         cq_destroy(ut->cq);
  58         ut->cq = NULL;
  59         ut->fd = -1;
  60 }
  61
  62 static void udp_delete_target(struct udp_target *ut, const char *msg)
  63 {
  64         PARA_NOTICE_LOG("deleting %s:%d (%s) from list\n", TARGET_ADDR(ut),
  65                 ut->port, msg);
  66         udp_close_target(ut);
  67         list_del(&ut->node);
  68         free(ut);
  69 }
  70
  71 /**
  72  * Perform AF-independent multicast sender setup.
  73  *
  74  * \param fd    The connected socket descriptor.
  75  * \param ttl   UDPv4 multicast TTL or UDPv6 multicast number of hops.
  76  *              Use -1 to mean default, 0..255 otherwise.
  77  * \param iface The outgoing multicast interface, or NULL for the default.
  78  *
  79  * \return Zero if okay, negative on error.
  80  */
  81 static int mcast_sender_setup(struct udp_target *ut, int ttl, char *iface)
  82 {
  83         struct sockaddr_storage ss;
  84         socklen_t sslen = sizeof(ss);
  85
  86         const int on = 1;
  87         int id = iface == NULL ? 0 : if_nametoindex(iface);
  88
  89         if (getpeername(ut->fd, (struct sockaddr *)&ss, &sslen) < 0)
  90                 goto err;
  91
  92         if (iface != NULL && id == 0)
  93                 PARA_WARNING_LOG("could not resolve interface %s, using default", iface);
  94
  95         /* RFC 3493, 5.2: -1 means 'use kernel default' */
  96         if (ttl < 0 || ttl > 255)
  97                 ttl = -1;
  98
  99         switch (ss.ss_family) {
 100         case AF_INET:
 101                 if (!IN_MULTICAST(htonl(((struct sockaddr_in *)&ss)->sin_addr.s_addr)))
 102                         return 0;
 103                 if (id != 0) {
 104 #ifdef HAVE_IP_MREQN
 105                         struct ip_mreqn mn;
 106
 107                         memset(&mn, 0, sizeof(mn));
 108                         mn.imr_ifindex = id;
 109                         if (setsockopt(ut->fd, IPPROTO_IP, IP_MULTICAST_IF, &mn, sizeof(mn)) < 0)
 110                                 goto err;
 111 #else
 112                         PARA_ERROR_LOG("No support for setting outgoing IPv4 mcast interface.");
 113 #endif
 114                 }
 115                 /*
 116                  * Enable receiving multicast messages generated on the local host
 117                  * At least on Linux, this is enabled by default.
 118                  */
 119                 if (setsockopt(ut->fd, IPPROTO_IP, IP_MULTICAST_LOOP, &on, sizeof(on)) < 0)
 120                         break;
 121
 122                 /* Default: use local subnet (do not flood out into the WAN) */
 123                 if (ttl == -1)
 124                         ttl = 1;
 125                 if (setsockopt(ut->fd, IPPROTO_IP, IP_MULTICAST_TTL, &ttl, sizeof(ttl)) < 0)
 126                         break;
 127                 return 0;
 128         case AF_INET6:
 129                 if (!IN6_IS_ADDR_MULTICAST(&((struct sockaddr_in6 *)&ss)->sin6_addr))
 130                         return 0;
 131                 if (id != 0 &&
 132                     setsockopt(ut->fd, IPPROTO_IPV6, IPV6_MULTICAST_IF, &id, sizeof(id)) < 0)
 133                         break;
 134                 if (setsockopt(ut->fd, IPPROTO_IPV6, IPV6_MULTICAST_LOOP, &on, sizeof(on)) < 0)
 135                         break;
 136                 if (setsockopt(ut->fd, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, &ttl, sizeof(ttl)) < 0)
 137                         break;
 138                 return 0;
 139         default:
 140                 PARA_ERROR_LOG("address family %d not supported", ss.ss_family);
 141                 return -E_ADDRESS_LOOKUP;
 142         }
 143 err:
 144         return -ERRNO_TO_PARA_ERROR(errno);
 145 }
 146
 147 /** The maximal size of the per-target chunk queue. */
 148 #define UDP_CQ_BYTES 40000
 149
 150 static int udp_init_session(struct udp_target *ut)
 151 {
 152         int ret;
 153         char *iface = NULL;
 154
 155         if (ut->fd >= 0) /* nothing to do */
 156                 return 0;
 157
 158         ret = makesock(AF_UNSPEC, IPPROTO_UDP, 0, TARGET_ADDR(ut), ut->port);
 159         if (ret < 0)
 160                 return ret;
 161         ut->fd = ret;
 162
 163         if (conf.udp_mcast_iface_given)
 164                 iface = conf.udp_mcast_iface_arg;
 165
 166         ret = mcast_sender_setup(ut, conf.udp_ttl_arg, iface);
 167         if (ret < 0) {
 168                 close(ut->fd);
 169                 return ret;
 170         }
 171
 172         ret = mark_fd_nonblocking(ut->fd);
 173         if (ret < 0) {
 174                 close(ut->fd);
 175                 return ret;
 176         }
 177         add_close_on_fork_list(ut->fd);
 178         ut->cq = cq_new(UDP_CQ_BYTES);
 179         PARA_NOTICE_LOG("sending to udp %s:%d\n", TARGET_ADDR(ut), ut->port);
 180         return 1;
 181 }
 182
 183 static void udp_send_buf(char *buf, size_t len)
 184 {
 185         struct udp_target *ut, *tmp;
 186         int ret;
 187
 188         list_for_each_entry_safe(ut, tmp, &targets, node) {
 189                 ret = udp_init_session(ut);
 190                 if (ret < 0) {
 191                         udp_delete_target(ut, para_strerror(-ret));
 192                         continue;
 193                 }
 194                 ret = send_queued_chunks(ut->fd, ut->cq, 0);
 195                 if (ret < 0) {
 196                         udp_delete_target(ut, para_strerror(-ret));
 197                         continue;
 198                 }
 199                 if (!len)
 200                         continue;
 201                 if (!ret) { /* still data left in the queue */
 202                         ret = cq_enqueue(ut->cq, buf, len);
 203                         if (ret < 0) {
 204                                 udp_delete_target(ut, para_strerror(-ret));
 205                                 continue;
 206                         }
 207                 }
 208                 ret = write_nonblock(ut->fd, buf, len, 0);
 209                 if (ret < 0) {
 210                         udp_delete_target(ut, para_strerror(-ret));
 211                         continue;
 212                 }
 213                 if (ret != len) {
 214                         ret = cq_enqueue(ut->cq, buf + ret, len - ret);
 215                         if (ret < 0) {
 216                                 udp_delete_target(ut, para_strerror(-ret));
 217                                 continue;
 218                         }
 219                 }
 220         }
 221 }
 222
 223 static void udp_shutdown_targets(void)
 224 {
 225         char buf[UDP_AUDIO_HEADER_LEN];
 226         struct udp_target *ut, *tmp;
 227         struct udp_audio_header uah = {
 228                 .stream_type = UDP_UNKNOWN_STREAM,
 229                 .packet_type = UDP_EOF_PACKET,
 230         };
 231
 232         write_udp_audio_header(buf, &uah);
 233         list_for_each_entry_safe(ut, tmp, &targets, node) {
 234                 if (ut->fd < 0)
 235                         continue;
 236                 write(ut->fd, buf, UDP_AUDIO_HEADER_LEN);
 237                 udp_close_target(ut);
 238         }
 239 }
 240
 241 static int need_extra_header(long unsigned current_chunk)
 242 {
 243         static struct timeval last_header;
 244         struct timeval diff;
 245
 246         if (!current_chunk)
 247                 return 0;
 248         tv_diff(now, &last_header, &diff);
 249         if (tv2ms(&diff) < conf.udp_header_interval_arg)
 250                 return 0;
 251         last_header = *now;
 252         return 1;
 253 }
 254
 255 static void udp_send(long unsigned current_chunk, __a_unused long unsigned chunks_sent,
 256                 const char *buf, size_t len, const char *header_buf,
 257                 size_t header_len)
 258 {
 259         char *sendbuf;
 260         size_t sendbuf_len;
 261         struct timeval *chunk_tv;
 262         struct udp_audio_header uah;
 263
 264 //      PARA_NOTICE_LOG("len: %zd, header_len: %zd\n", len, header_len);
 265         if (sender_status != SENDER_ON)
 266                 return;
 267
 268         /* we might not yet know the chunk time */
 269         chunk_tv = vss_chunk_time();
 270         if (!chunk_tv)
 271                 return;
 272         if (list_empty(&targets))
 273                 return;
 274         uah.stream_type = header_len? UDP_HEADER_STREAM : UDP_PLAIN_STREAM;
 275         uah.header_len = need_extra_header(current_chunk)? header_len : 0;
 276         if (!current_chunk)
 277                 uah.packet_type = UDP_BOF_PACKET;
 278         else if (uah.header_len)
 279                 uah.packet_type = UDP_HEADER_PACKET;
 280         else
 281                 uah.packet_type = UDP_DATA_PACKET;
 282         uah.payload_len = uah.header_len + len;
 283         sendbuf_len = UDP_AUDIO_HEADER_LEN + uah.payload_len;
 284         sendbuf = para_malloc(sendbuf_len);
 285         write_udp_audio_header(sendbuf, &uah);
 286         if (uah.header_len)
 287                 memcpy(sendbuf + UDP_AUDIO_HEADER_LEN, header_buf,
 288                         uah.header_len);
 289         memcpy(sendbuf + UDP_AUDIO_HEADER_LEN + uah.header_len, buf, len);
 290         udp_send_buf(sendbuf, sendbuf_len);
 291         free(sendbuf);
 292 }
 293
 294 static int udp_com_on(__a_unused struct sender_command_data *scd)
 295 {
 296         sender_status = SENDER_ON;
 297         return 1;
 298 }
 299
 300 static int udp_com_off(__a_unused struct sender_command_data *scd)
 301 {
 302         udp_shutdown_targets();
 303         sender_status = SENDER_OFF;
 304         return 1;
 305 }
 306
 307 static int udp_com_delete(struct sender_command_data *scd)
 308 {
 309         char *a = para_strdup(inet_ntoa(scd->addr));
 310         struct udp_target *ut, *tmp;
 311         list_for_each_entry_safe(ut, tmp, &targets, node) {
 312                 if (scd->port != ut->port)
 313                         continue;
 314                 if (strcmp(TARGET_ADDR(ut), a))
 315                         continue;
 316                 udp_delete_target(ut, "com_delete");
 317         }
 318         return 1;
 319 }
 320
 321 static void udp_add_target(int port, struct in_addr *addr)
 322 {
 323         struct udp_target *ut = para_calloc(sizeof(struct udp_target));
 324         ut->port = port;
 325         ut->addr = *addr;
 326         ut->fd = -1; /* not yet connected */
 327         PARA_INFO_LOG("adding to target list (%s:%d)\n",
 328                 TARGET_ADDR(ut), ut->port);
 329         para_list_add(&ut->node, &targets);
 330 }
 331
 332 static int udp_com_add(struct sender_command_data *scd)
 333 {
 334         int port = (scd->port > 0)? scd->port : conf.udp_default_port_arg;
 335         udp_add_target(port, &scd->addr);
 336         return 1;
 337 }
 338
 339 static char *udp_info(void)
 340 {
 341         struct udp_target *ut;
 342         char *ret, *tgts = NULL;
 343
 344         list_for_each_entry(ut, &targets, node) {
 345                 char *tmp = make_message("%s%s:%d ", tgts? tgts : "",
 346                         TARGET_ADDR(ut), ut->port);
 347                 free(tgts);
 348                 tgts = tmp;
 349         }
 350         ret = make_message(
 351                 "udp sender:\n"
 352                 "\tstatus: %s\n"
 353                 "\tport: udp %d\n"
 354                 "\ttargets: %s\n",
 355                 (sender_status == SENDER_ON)? "on" : "off",
 356                 conf.udp_default_port_arg,
 357                 tgts? tgts : "(none)"
 358         );
 359         free(tgts);
 360         return ret;
 361 }
 362
 363 static void udp_init_target_list(void)
 364 {
 365         int i;
 366
 367         INIT_LIST_HEAD(&targets);
 368         for (i = 0; i < conf.udp_target_given; i++) {
 369                 char *arg = para_strdup(conf.udp_target_arg[i]);
 370                 char *p = strchr(arg, ':');
 371                 int port;
 372                 struct in_addr addr;
 373
 374                 if (!p)
 375                         goto err;
 376                 *p = '\0';
 377                 if (!inet_pton(AF_INET, arg, &addr))
 378                         goto err;
 379                 port = atoi(++p);
 380                 if (port < 0 || port > 65535)
 381                         port = conf.udp_default_port_arg;
 382                 udp_add_target(port, &addr);
 383                 goto success;
 384 err:
 385                 PARA_CRIT_LOG("syntax error for udp target option "
 386                         "#%d, ignoring\n", i);
 387 success:
 388                 free(arg);
 389                 continue;
 390         }
 391 }
 392
 393 static char *udp_help(void)
 394 {
 395         return make_message(
 396                 "usage: {on|off}\n"
 397                 "usage: {add|delete} IP port\n"
 398                 "example: add 224.0.1.38 8000 (multicast streaming)\n"
 399         );
 400 }
 401
 402 /**
 403  * The init function of para_server's udp sender.
 404  *
 405  * \param s Pointer to the http sender struct.
 406  *
 407  * It initializes all function pointers of \a s and the list of udp targets.
 408  */
 409 void udp_send_init(struct sender *s)
 410 {
 411         INIT_LIST_HEAD(&targets);
 412         s->info = udp_info;
 413         s->help = udp_help;
 414         s->send = udp_send;
 415         s->pre_select = NULL;
 416         s->post_select = NULL;
 417         s->shutdown_clients = udp_shutdown_targets;
 418         s->client_cmds[SENDER_ON] = udp_com_on;
 419         s->client_cmds[SENDER_OFF] = udp_com_off;
 420         s->client_cmds[SENDER_DENY] = NULL;
 421         s->client_cmds[SENDER_ALLOW] = NULL;
 422         s->client_cmds[SENDER_ADD] = udp_com_add;
 423         s->client_cmds[SENDER_DELETE] = udp_com_delete;
 424         sender_status = SENDER_OFF;
 425         udp_init_target_list();
 426         if (!conf.udp_no_autostart_given)
 427                 sender_status = SENDER_ON;
 428         PARA_DEBUG_LOG("udp sender init complete\n");
 429 }