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