dccp_send.c

   1 /*
   2  * Copyright (C) 2006-2008 Andre Noll <maan@systemlinux.org>
   3  *
   4  * Licensed under the GPL v2. For licencing details see COPYING.
   5  */
   6
   7 /** \file dccp_send.c paraslash's dccp sender */
   8
   9 /*
  10  * based on server.c of dccp-cs-0.01.tar.bz2,
  11  * (C) 2005 Ian McDonald <imcdnzl@gmail.com>
  12  */
  13
  14 #include <sys/types.h>
  15 #include <dirent.h>
  16
  17 #include "para.h"
  18 #include "error.h"
  19 #include "string.h"
  20 #include "afh.h"
  21 #include "afs.h"
  22 #include "server.h"
  23 #include "net.h"
  24 #include "list.h"
  25 #include "vss.h"
  26 #include "send.h"
  27 #include "fd.h"
  28 #include "close_on_fork.h"
  29 #include "chunk_queue.h"
  30 #include "server.cmdline.h"
  31
  32 /** the list of connected clients **/
  33 static struct list_head clients;
  34 static int listen_fd = -1;
  35
  36 /** Maximal number of bytes in a chunk queue. */
  37 #define DCCP_MAX_PENDING_BYTES 40000
  38
  39 /** describes one connected client */
  40 struct dccp_client {
  41         /** the dccp socket */
  42         int fd;
  43         /** The socket `name' of the client. */
  44         char *name;
  45         /** the position of this client in the client list */
  46         struct list_head node;
  47         /** non-zero if audio file header has been sent */
  48         int header_sent;
  49         /** The list of pending chunks for this client. */
  50         struct chunk_queue *cq;
  51 };
  52
  53 static void dccp_pre_select(int *max_fileno, fd_set *rfds,
  54                 __a_unused fd_set *wfds)
  55 {
  56         if (listen_fd < 0)
  57                 return;
  58         FD_SET(listen_fd, rfds);
  59         *max_fileno = PARA_MAX(*max_fileno, listen_fd);
  60 }
  61
  62 static void dccp_post_select(fd_set *rfds, __a_unused fd_set *wfds)
  63 {
  64         struct dccp_client *dc;
  65         int ret, fd;
  66
  67         if (listen_fd < 0 || !FD_ISSET(listen_fd, rfds))
  68                 return;
  69         ret = para_accept(listen_fd, NULL, 0);
  70         if (ret < 0) {
  71                 PARA_ERROR_LOG("%s\n", para_strerror(-ret));
  72                 return;
  73         }
  74         fd = ret;
  75         /*
  76          * Bypass unused CCID paths: the sender does not receive application data
  77          * from the client; by shutting down this unused communication path we can
  78          * reduce processing costs a bit. See analogous comment in dccp_recv.c.
  79          */
  80         if (shutdown(fd, SHUT_RD) < 0) {
  81                 PARA_ERROR_LOG("shutdown(SHUT_RD): %s\n", strerror(errno));
  82                 goto err;
  83         }
  84         ret = mark_fd_nonblocking(fd);
  85         if (ret < 0) {
  86                 PARA_ERROR_LOG("%s\n", para_strerror(-ret));
  87                 goto err;
  88         }
  89         dc = para_calloc(sizeof(struct dccp_client));
  90         dc->fd = fd;
  91         dc->name = make_message("%s", remote_name(dc->fd));
  92         PARA_NOTICE_LOG("connection from %s\n", dc->name);
  93         para_list_add(&dc->node, &clients);
  94         add_close_on_fork_list(dc->fd);
  95         dc->cq = cq_new(DCCP_MAX_PENDING_BYTES);
  96         return;
  97 err:
  98         close(fd);
  99 }
 100
 101 static int dccp_open(void)
 102 {
 103         int ret = para_listen(AF_UNSPEC, IPPROTO_DCCP, conf.dccp_port_arg);
 104
 105         if (ret < 0)
 106                 return ret;
 107         listen_fd = ret;
 108         ret = mark_fd_nonblocking(listen_fd);
 109         if (ret < 0) {
 110                 PARA_EMERG_LOG("%s\n", para_strerror(-ret));
 111                 exit(EXIT_FAILURE);
 112         }
 113         add_close_on_fork_list(listen_fd);
 114         return 1;
 115 }
 116
 117 static void dccp_shutdown_client(struct dccp_client *dc)
 118 {
 119         PARA_DEBUG_LOG("shutting down %s (fd %d)\n", dc->name, dc->fd);
 120         free(dc->name);
 121         close(dc->fd);
 122         del_close_on_fork_list(dc->fd);
 123         cq_destroy(dc->cq);
 124         list_del(&dc->node);
 125         free(dc);
 126 }
 127
 128 /*
 129  * ret: Negative on errors, zero if nothing was written and write returned
 130  * EAGAIN, number of bytes written else.
 131  */
 132 static int dccp_write(int fd, const char *buf, size_t len)
 133 {
 134         size_t written = 0;
 135         int ret = 0;
 136
 137         while (written < len) {
 138                 ret = write(fd, buf + written, PARA_MIN(1024, len - written));
 139                 /*
 140                  * Error handling: CCID3 has a sending wait queue which fills
 141                  * up and is emptied asynchronously. The EAGAIN case means that
 142                  * there is currently no space in the wait queue, but this can
 143                  * change at any moment and is thus not an error condition.
 144                  */
 145                 if (ret < 0 && errno == EAGAIN)
 146                         return written;
 147                 if (ret < 0)
 148                         return -ERRNO_TO_PARA_ERROR(errno);
 149                 written += ret;
 150         }
 151         return written;
 152 }
 153
 154 static int queue_chunk_or_shutdown(struct dccp_client *dc, long unsigned chunk_num,
 155         size_t sent)
 156 {
 157         int ret = cq_enqueue(dc->cq, chunk_num, sent);
 158         if (ret < 0) {
 159                 PARA_NOTICE_LOG("enqueue error\n");
 160                 dccp_shutdown_client(dc);
 161         }
 162         return ret;
 163 }
 164
 165 static int send_queued_chunks(struct dccp_client *dc)
 166 {
 167         struct queued_chunk *qc;
 168         while ((qc = cq_peek(dc->cq))) {
 169                 char *buf;
 170                 size_t len;
 171                 int ret;
 172                 cq_get(qc, &buf, &len);
 173                 ret = dccp_write(dc->fd, buf, len);
 174                 if (ret < 0)
 175                         return ret;
 176                 cq_update(dc->cq, ret);
 177                 if (ret != len)
 178                         return 1;
 179                 cq_dequeue(dc->cq);
 180         }
 181         return 1;
 182 }
 183
 184 static void dccp_send(long unsigned current_chunk,
 185                 __a_unused long unsigned chunks_sent, const char *buf, size_t len)
 186 {
 187         struct dccp_client *dc, *tmp;
 188         int ret;
 189         char *header_buf;
 190
 191         list_for_each_entry_safe(dc, tmp, &clients, node) {
 192                 if (!dc->header_sent && current_chunk) {
 193                         size_t header_len;
 194                         header_buf = vss_get_header(&header_len);
 195                         if (header_buf && header_len > 0) {
 196                                 if (queue_chunk_or_shutdown(dc, -1U, 0) < 0)
 197                                         continue;
 198                         }
 199                         dc->header_sent = 1;
 200                 }
 201                 ret = send_queued_chunks(dc);
 202                 if (ret < 0) {
 203                         dccp_shutdown_client(dc);
 204                         continue;
 205                 }
 206                 if (!len)
 207                         continue;
 208 //              PARA_DEBUG_LOG("writing %d bytes to fd %d\n", len, dc->fd);
 209                 ret = dccp_write(dc->fd, buf, len);
 210                 if (ret < 0) {
 211                         PARA_NOTICE_LOG("%s\n", para_strerror(-ret));
 212                         dccp_shutdown_client(dc);
 213                         continue;
 214                 }
 215                 if (ret != len)
 216                         queue_chunk_or_shutdown(dc, current_chunk, ret);
 217         }
 218 }
 219
 220 static void dccp_shutdown_clients(void)
 221 {
 222         struct dccp_client *dc, *tmp;
 223
 224         list_for_each_entry_safe(dc, tmp, &clients, node)
 225                 dccp_shutdown_client(dc);
 226 }
 227
 228 static char *dccp_info(void)
 229 {
 230         static char *buf;
 231         int num_clients = 0;
 232         struct dccp_client *dc, *tmp;
 233
 234         free(buf);
 235         list_for_each_entry_safe(dc, tmp, &clients, node)
 236                 num_clients++;
 237         buf = make_message("dccp connected clients: %d\n", num_clients);
 238         return buf;
 239 }
 240
 241 static char *dccp_help(void)
 242 {
 243         return make_message("no help available\n");
 244 }
 245
 246 /**
 247  * the init function of the dccp sender
 248  *
 249  * \param s pointer to the dccp sender struct
 250  *
 251  * It initializes all function pointers of \a s and starts
 252  * listening on the given port.
 253  */
 254 void dccp_send_init(struct sender *s)
 255 {
 256         int ret;
 257
 258         INIT_LIST_HEAD(&clients);
 259         s->info = dccp_info;
 260         s->send = dccp_send;
 261         s->pre_select = dccp_pre_select;
 262         s->post_select = dccp_post_select;
 263         s->shutdown_clients = dccp_shutdown_clients;
 264         s->help = dccp_help;
 265         s->client_cmds[SENDER_ON] = NULL;
 266         s->client_cmds[SENDER_OFF] = NULL;
 267         s->client_cmds[SENDER_DENY] = NULL;
 268         s->client_cmds[SENDER_ALLOW] = NULL;
 269         s->client_cmds[SENDER_ADD] = NULL;
 270         s->client_cmds[SENDER_DELETE] = NULL;
 271         ret = dccp_open();
 272         if (ret < 0)
 273                 PARA_ERROR_LOG("%s\n", para_strerror(-ret));
 274 }