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