dccp_send.c

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