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
  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 };
  45
  46 static struct list_head targets;
  47 static int sender_status;
  48
  49 static void udp_delete_target(struct udp_target *ut, const char *msg)
  50 {
  51         PARA_NOTICE_LOG("deleting %s:%d (%s) from list\n", TARGET_ADDR(ut),
  52                 ut->port, msg);
  53         if (ut->fd >= 0) {
  54                 close(ut->fd);
  55                 del_close_on_fork_list(ut->fd);
  56         }
  57         list_del(&ut->node);
  58         free(ut);
  59 }
  60
  61 static void udp_send_buf(char *buf, size_t len)
  62 {
  63         struct udp_target *ut, *tmp;
  64         int ret;
  65
  66         list_for_each_entry_safe(ut, tmp, &targets, node) {
  67                 if (ut->fd < 0)
  68                         continue;
  69                 ret = write_nonblock(ut->fd, buf, len, len);
  70                 if (ret < 0) /* TODO: Use chunk queueing */
  71                         return udp_delete_target(ut, "send error");
  72                 if (ret != len)
  73                         PARA_WARNING_LOG("short write %zu/%zu\n", ret, len);
  74         }
  75 }
  76
  77 static int udp_init_session(struct udp_target *ut)
  78 {
  79         int ret;
  80
  81         if (ut->fd >= 0) /* nothing to do */
  82                 return 0;
  83         PARA_NOTICE_LOG("sending to udp %s:%d\n", TARGET_ADDR(ut), ut->port);
  84         ret = create_udp_send_socket(TARGET_ADDR(ut), ut->port,
  85                 conf.udp_ttl_arg);
  86         if (ret < 0)
  87                 return ret;
  88         ut->fd = ret;
  89         ret = mark_fd_nonblocking(ut->fd);
  90         if (ret < 0) {
  91                 close(ut->fd);
  92                 return ret;
  93         }
  94         add_close_on_fork_list(ut->fd);
  95         return 1;
  96 }
  97
  98 static void udp_shutdown_targets(void)
  99 {
 100         char buf[UDP_AUDIO_HEADER_LEN];
 101         struct udp_target *ut, *tmp;
 102
 103         udp_write_packet_type(buf, UDP_EOF_PACKET);
 104         udp_write_magic(buf);
 105         list_for_each_entry_safe(ut, tmp, &targets, node) {
 106                 if (ut->fd < 0)
 107                         continue;
 108                 write(ut->fd, buf, UDP_AUDIO_HEADER_LEN);
 109                 close(ut->fd);
 110                 del_close_on_fork_list(ut->fd);
 111                 ut->fd = -1;
 112         }
 113 }
 114
 115 static int need_extra_header(long unsigned current_chunk)
 116 {
 117         static struct timeval last_header;
 118         struct timeval diff;
 119
 120         if (!current_chunk)
 121                 return 0;
 122         tv_diff(now, &last_header, &diff);
 123         if (tv2ms(&diff) < conf.udp_header_interval_arg)
 124                 return 0;
 125         last_header = *now;
 126         return 1;
 127 }
 128
 129 static void udp_send(long unsigned current_chunk, __a_unused long unsigned chunks_sent,
 130                 const char *buf, size_t len, const char *header_buf,
 131                 size_t header_len)
 132 {
 133         struct udp_target *ut, *tmp;
 134         size_t sendbuf_len;
 135         uint8_t packet_type = UDP_DATA_PACKET;
 136         int ret;
 137         char *sendbuf;
 138         struct timeval *chunk_tv;
 139         uint8_t stream_type = header_len? UDP_HEADER_STREAM : UDP_PLAIN_STREAM;
 140
 141 //      PARA_NOTICE_LOG("header_len: %zd, header_buf: %p\n", header_len,
 142 //              header_buf);
 143         if (sender_status != SENDER_ON)
 144                 return;
 145
 146         /* we might not yet know the chunk time */
 147         chunk_tv = vss_chunk_time();
 148         if (!chunk_tv)
 149                 return;
 150         if (list_empty(&targets))
 151                 return;
 152         list_for_each_entry_safe(ut, tmp, &targets, node) {
 153                 ret = udp_init_session(ut);
 154                 if (ret < 0)
 155                         udp_delete_target(ut, para_strerror(-ret));
 156         }
 157         if (!need_extra_header(current_chunk))
 158                 header_len = 0;
 159         if (!current_chunk)
 160                 packet_type = UDP_BOF_PACKET;
 161         else if (header_len)
 162                 packet_type = UDP_HEADER_PACKET;
 163         sendbuf_len = UDP_AUDIO_HEADER_LEN + header_len + len;
 164         sendbuf = para_malloc(sendbuf_len);
 165         udp_write_magic(sendbuf);
 166         udp_write_stream_type(sendbuf, stream_type);
 167         udp_write_packet_type(sendbuf, packet_type);
 168         udp_write_header_len(sendbuf, header_len);
 169         if (header_len)
 170                 memcpy(sendbuf + UDP_AUDIO_HEADER_LEN, header_buf,
 171                         header_len);
 172         memcpy(sendbuf + UDP_AUDIO_HEADER_LEN + header_len, buf, len);
 173         udp_send_buf(sendbuf, sendbuf_len);
 174         free(sendbuf);
 175 }
 176
 177 static int udp_com_on(__a_unused struct sender_command_data *scd)
 178 {
 179         sender_status = SENDER_ON;
 180         return 1;
 181 }
 182
 183 static int udp_com_off(__a_unused struct sender_command_data *scd)
 184 {
 185         udp_shutdown_targets();
 186         sender_status = SENDER_OFF;
 187         return 1;
 188 }
 189
 190 static int udp_com_delete(struct sender_command_data *scd)
 191 {
 192         char *a = para_strdup(inet_ntoa(scd->addr));
 193         struct udp_target *ut, *tmp;
 194         list_for_each_entry_safe(ut, tmp, &targets, node) {
 195                 if (scd->port != ut->port)
 196                         continue;
 197                 if (strcmp(TARGET_ADDR(ut), a))
 198                         continue;
 199                 udp_delete_target(ut, "com_delete");
 200         }
 201         return 1;
 202 }
 203
 204 static void udp_add_target(int port, struct in_addr *addr)
 205 {
 206         struct udp_target *ut = para_calloc(sizeof(struct udp_target));
 207         ut->port = port;
 208         ut->addr = *addr;
 209         ut->fd = -1; /* not yet connected */
 210         PARA_INFO_LOG("adding to target list (%s:%d)\n",
 211                 TARGET_ADDR(ut), ut->port);
 212         para_list_add(&ut->node, &targets);
 213 }
 214
 215 static int udp_com_add(struct sender_command_data *scd)
 216 {
 217         int port = (scd->port > 0)? scd->port : conf.udp_default_port_arg;
 218         udp_add_target(port, &scd->addr);
 219         return 1;
 220 }
 221
 222 static char *udp_info(void)
 223 {
 224         struct udp_target *ut;
 225         char *ret, *tgts = NULL;
 226
 227         list_for_each_entry(ut, &targets, node) {
 228                 char *tmp = make_message("%s%s:%d ", tgts? tgts : "",
 229                         TARGET_ADDR(ut), ut->port);
 230                 free(tgts);
 231                 tgts = tmp;
 232         }
 233         ret = make_message(
 234                 "udp sender:\n"
 235                 "\tstatus: %s\n"
 236                 "\tport: udp %d\n"
 237                 "\ttargets: %s\n",
 238                 (sender_status == SENDER_ON)? "on" : "off",
 239                 conf.udp_default_port_arg,
 240                 tgts? tgts : "(none)"
 241         );
 242         free(tgts);
 243         return ret;
 244 }
 245
 246 static void udp_init_target_list(void)
 247 {
 248         int i;
 249
 250         INIT_LIST_HEAD(&targets);
 251         for (i = 0; i < conf.udp_target_given; i++) {
 252                 char *arg = para_strdup(conf.udp_target_arg[i]);
 253                 char *p = strchr(arg, ':');
 254                 int port;
 255                 struct in_addr addr;
 256
 257                 if (!p)
 258                         goto err;
 259                 *p = '\0';
 260                 if (!inet_pton(AF_INET, arg, &addr))
 261                         goto err;
 262                 port = atoi(++p);
 263                 if (port < 0 || port > 65535)
 264                         port = conf.udp_default_port_arg;
 265                 udp_add_target(port, &addr);
 266                 goto success;
 267 err:
 268                 PARA_CRIT_LOG("syntax error for udp target option "
 269                         "#%d, ignoring\n", i);
 270 success:
 271                 free(arg);
 272                 continue;
 273         }
 274 }
 275
 276 static char *udp_help(void)
 277 {
 278         return make_message(
 279                 "usage: {on|off}\n"
 280                 "usage: {add|delete} IP port\n"
 281                 "example: add 224.0.1.38 8000 (multicast streaming)\n"
 282         );
 283 }
 284
 285 /**
 286  * The init function of para_server's udp sender.
 287  *
 288  * \param s Pointer to the http sender struct.
 289  *
 290  * It initializes all function pointers of \a s and the list of udp targets.
 291  */
 292 void udp_send_init(struct sender *s)
 293 {
 294         INIT_LIST_HEAD(&targets);
 295         s->info = udp_info;
 296         s->help = udp_help;
 297         s->send = udp_send;
 298         s->pre_select = NULL;
 299         s->post_select = NULL;
 300         s->shutdown_clients = udp_shutdown_targets;
 301         s->client_cmds[SENDER_ON] = udp_com_on;
 302         s->client_cmds[SENDER_OFF] = udp_com_off;
 303         s->client_cmds[SENDER_DENY] = NULL;
 304         s->client_cmds[SENDER_ALLOW] = NULL;
 305         s->client_cmds[SENDER_ADD] = udp_com_add;
 306         s->client_cmds[SENDER_DELETE] = udp_com_delete;
 307         sender_status = SENDER_OFF;
 308         udp_init_target_list();
 309         if (!conf.udp_no_autostart_given)
 310                 sender_status = SENDER_ON;
 311         PARA_DEBUG_LOG("udp sender init complete\n");
 312 }