Line data Source code
1 : /*
2 : Copyright (c) 2007, Adobe Systems, Incorporated
3 : Copyright (c) 2013, Mozilla
4 : All rights reserved.
5 :
6 : Redistribution and use in source and binary forms, with or without
7 : modification, are permitted provided that the following conditions are
8 : met:
9 :
10 : * Redistributions of source code must retain the above copyright
11 : notice, this list of conditions and the following disclaimer.
12 :
13 : * Redistributions in binary form must reproduce the above copyright
14 : notice, this list of conditions and the following disclaimer in the
15 : documentation and/or other materials provided with the distribution.
16 :
17 : * Neither the name of Adobe Systems, Network Resonance nor the names of its
18 : contributors may be used to endorse or promote products derived from
19 : this software without specific prior written permission.
20 :
21 : THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
22 : "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
23 : LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
24 : A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
25 : OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
26 : SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
27 : LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
28 : DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
29 : THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
30 : (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
31 : OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
32 : */
33 :
34 : #include <nr_api.h>
35 :
36 : #include <stdio.h>
37 : #include <string.h>
38 : #include <sys/types.h>
39 : #include <sys/queue.h>
40 : #include <assert.h>
41 : #include <inttypes.h>
42 :
43 : #include "p_buf.h"
44 : #include "nr_socket.h"
45 : #include "stun.h"
46 : #include "nr_socket_buffered_stun.h"
47 :
48 : #define NR_MAX_FRAME_SIZE 0xFFFF
49 :
50 : typedef struct nr_frame_header_ {
51 : UINT2 frame_length;
52 : char data[0];
53 : } nr_frame_header;
54 :
55 : typedef struct nr_socket_buffered_stun_ {
56 : nr_socket *inner;
57 : nr_transport_addr remote_addr;
58 : int connected;
59 :
60 : /* Read state */
61 : int read_state;
62 : #define NR_ICE_SOCKET_READ_NONE 0
63 : #define NR_ICE_SOCKET_READ_HDR 1
64 : #define NR_ICE_SOCKET_READ_FAILED 2
65 : UCHAR *buffer;
66 : size_t buffer_size;
67 : size_t bytes_needed;
68 : size_t bytes_read;
69 : NR_async_cb readable_cb;
70 : void *readable_cb_arg;
71 :
72 : /* Write state */
73 : nr_p_buf_ctx *p_bufs;
74 : nr_p_buf_head pending_writes;
75 : size_t pending;
76 : size_t max_pending;
77 : nr_framing_type framing_type;
78 : } nr_socket_buffered_stun;
79 :
80 : static int nr_socket_buffered_stun_destroy(void **objp);
81 : static int nr_socket_buffered_stun_sendto(void *obj,const void *msg, size_t len,
82 : int flags, nr_transport_addr *to);
83 : static int nr_socket_buffered_stun_recvfrom(void *obj,void * restrict buf,
84 : size_t maxlen, size_t *len, int flags, nr_transport_addr *from);
85 : static int nr_socket_buffered_stun_getfd(void *obj, NR_SOCKET *fd);
86 : static int nr_socket_buffered_stun_getaddr(void *obj, nr_transport_addr *addrp);
87 : static int nr_socket_buffered_stun_close(void *obj);
88 : static int nr_socket_buffered_stun_connect(void *sock, nr_transport_addr *addr);
89 : static int nr_socket_buffered_stun_write(void *obj,const void *msg, size_t len, size_t *written);
90 : static void nr_socket_buffered_stun_writable_cb(NR_SOCKET s, int how, void *arg);
91 : static int nr_socket_buffered_stun_listen(void *obj, int backlog);
92 : static int nr_socket_buffered_stun_accept(void *obj, nr_transport_addr *addrp, nr_socket **sockp);
93 :
94 : static nr_socket_vtbl nr_socket_buffered_stun_vtbl={
95 : 2,
96 : nr_socket_buffered_stun_destroy,
97 : nr_socket_buffered_stun_sendto,
98 : nr_socket_buffered_stun_recvfrom,
99 : nr_socket_buffered_stun_getfd,
100 : nr_socket_buffered_stun_getaddr,
101 : nr_socket_buffered_stun_connect,
102 : 0,
103 : 0,
104 : nr_socket_buffered_stun_close,
105 : nr_socket_buffered_stun_listen,
106 : nr_socket_buffered_stun_accept
107 : };
108 :
109 0 : void nr_socket_buffered_stun_set_readable_cb(nr_socket *sock,
110 : NR_async_cb readable_cb, void *readable_cb_arg)
111 : {
112 0 : nr_socket_buffered_stun *buf_sock = (nr_socket_buffered_stun *)sock->obj;
113 :
114 0 : buf_sock->readable_cb = readable_cb;
115 0 : buf_sock->readable_cb_arg = readable_cb_arg;
116 0 : }
117 :
118 0 : int nr_socket_buffered_set_connected_to(nr_socket *sock, nr_transport_addr *remote_addr)
119 : {
120 0 : nr_socket_buffered_stun *buf_sock = (nr_socket_buffered_stun *)sock->obj;
121 : int r, _status;
122 :
123 0 : if ((r=nr_transport_addr_copy(&buf_sock->remote_addr, remote_addr)))
124 0 : ABORT(r);
125 :
126 0 : buf_sock->connected = 1;
127 :
128 0 : _status=0;
129 : abort:
130 0 : return(_status);
131 : }
132 :
133 0 : int nr_socket_buffered_stun_create(nr_socket *inner, int max_pending,
134 : nr_framing_type framing_type, nr_socket **sockp)
135 : {
136 : int r, _status;
137 0 : nr_socket_buffered_stun *sock = 0;
138 : size_t frame_size;
139 :
140 0 : if (!(sock = RCALLOC(sizeof(nr_socket_buffered_stun))))
141 0 : ABORT(R_NO_MEMORY);
142 :
143 0 : sock->inner = inner;
144 0 : sock->framing_type = framing_type;
145 :
146 0 : if ((r=nr_ip4_port_to_transport_addr(INADDR_ANY, 0, IPPROTO_UDP, &sock->remote_addr)))
147 0 : ABORT(r);
148 :
149 0 : switch (framing_type) {
150 : case ICE_TCP_FRAMING:
151 0 : frame_size = sizeof(nr_frame_header);
152 0 : sock->buffer_size = sizeof(nr_frame_header) + NR_MAX_FRAME_SIZE;
153 0 : sock->bytes_needed = sizeof(nr_frame_header);
154 0 : break;
155 : case TURN_TCP_FRAMING:
156 0 : frame_size = 0;
157 0 : sock->buffer_size = NR_STUN_MAX_MESSAGE_SIZE;
158 0 : sock->bytes_needed = sizeof(nr_stun_message_header);
159 0 : break;
160 : default:
161 0 : assert(0);
162 : ABORT(R_BAD_ARGS);
163 : }
164 :
165 : /* TODO(ekr@rtfm.com): Check this */
166 0 : if (!(sock->buffer = RMALLOC(sock->buffer_size)))
167 0 : ABORT(R_NO_MEMORY);
168 :
169 0 : sock->read_state = NR_ICE_SOCKET_READ_NONE;
170 0 : sock->connected = 0;
171 :
172 0 : STAILQ_INIT(&sock->pending_writes);
173 0 : if ((r=nr_p_buf_ctx_create(sock->buffer_size, &sock->p_bufs)))
174 0 : ABORT(r);
175 0 : sock->max_pending = max_pending + frame_size;
176 :
177 0 : if ((r=nr_socket_create_int(sock, &nr_socket_buffered_stun_vtbl, sockp)))
178 0 : ABORT(r);
179 :
180 0 : _status=0;
181 : abort:
182 0 : if (_status && sock) {
183 0 : void *sock_v = sock;
184 0 : sock->inner = 0; /* Give up ownership so we don't destroy */
185 0 : nr_socket_buffered_stun_destroy(&sock_v);
186 : }
187 0 : return(_status);
188 : }
189 :
190 : /* Note: This destroys the inner socket */
191 0 : int nr_socket_buffered_stun_destroy(void **objp)
192 : {
193 : nr_socket_buffered_stun *sock;
194 : NR_SOCKET fd;
195 :
196 0 : if (!objp || !*objp)
197 0 : return 0;
198 :
199 0 : sock = (nr_socket_buffered_stun *)*objp;
200 0 : *objp = 0;
201 :
202 : /* Free the buffer if needed */
203 0 : RFREE(sock->buffer);
204 :
205 : /* Cancel waiting on the socket */
206 0 : if (sock->inner && !nr_socket_getfd(sock->inner, &fd)) {
207 0 : NR_ASYNC_CANCEL(fd, NR_ASYNC_WAIT_WRITE);
208 : }
209 :
210 0 : nr_p_buf_free_chain(sock->p_bufs, &sock->pending_writes);
211 0 : nr_p_buf_ctx_destroy(&sock->p_bufs);
212 0 : nr_socket_destroy(&sock->inner);
213 0 : RFREE(sock);
214 :
215 0 : return 0;
216 : }
217 :
218 0 : static int nr_socket_buffered_stun_sendto(void *obj,const void *msg, size_t len,
219 : int flags, nr_transport_addr *to)
220 : {
221 0 : nr_socket_buffered_stun *sock = (nr_socket_buffered_stun *)obj;
222 : int r, _status;
223 : size_t written;
224 0 : nr_frame_header *frame = NULL;
225 :
226 : /* Check that we are writing to the connected address if
227 : connected */
228 0 : if (!nr_transport_addr_is_wildcard(&sock->remote_addr)) {
229 0 : if (nr_transport_addr_cmp(&sock->remote_addr, to, NR_TRANSPORT_ADDR_CMP_MODE_ALL)) {
230 0 : r_log(LOG_GENERIC, LOG_ERR, "Sendto on connected socket doesn't match");
231 0 : ABORT(R_BAD_DATA);
232 : }
233 : }
234 :
235 0 : if (sock->framing_type == ICE_TCP_FRAMING) {
236 :
237 0 : assert(len <= NR_MAX_FRAME_SIZE);
238 0 : if (len > NR_MAX_FRAME_SIZE)
239 0 : ABORT(R_FAILED);
240 :
241 0 : if (!(frame = RMALLOC(len + sizeof(nr_frame_header))))
242 0 : ABORT(R_NO_MEMORY);
243 :
244 0 : frame->frame_length = htons(len);
245 0 : memcpy(frame->data, msg, len);
246 0 : len += sizeof(nr_frame_header);
247 0 : msg = frame;
248 : }
249 :
250 0 : if ((r=nr_socket_buffered_stun_write(obj, msg, len, &written)))
251 0 : ABORT(r);
252 :
253 0 : if (len != written)
254 0 : ABORT(R_IO_ERROR);
255 :
256 0 : _status=0;
257 : abort:
258 0 : RFREE(frame);
259 0 : return _status;
260 : }
261 :
262 0 : static void nr_socket_buffered_stun_failed(nr_socket_buffered_stun *sock)
263 : {
264 : NR_SOCKET fd;
265 :
266 0 : sock->read_state = NR_ICE_SOCKET_READ_FAILED;
267 :
268 : /* Cancel waiting on the socket */
269 0 : if (sock->inner && !nr_socket_getfd(sock->inner, &fd)) {
270 0 : NR_ASYNC_CANCEL(fd, NR_ASYNC_WAIT_WRITE);
271 0 : NR_ASYNC_CANCEL(fd, NR_ASYNC_WAIT_READ);
272 : }
273 0 : }
274 :
275 0 : static int nr_socket_buffered_stun_recvfrom(void *obj,void * restrict buf,
276 : size_t maxlen, size_t *len, int flags, nr_transport_addr *from)
277 : {
278 : int r, _status;
279 : size_t bytes_read;
280 0 : nr_socket_buffered_stun *sock = (nr_socket_buffered_stun *)obj;
281 0 : nr_frame_header *frame = (nr_frame_header *)sock->buffer;
282 0 : size_t skip_hdr_size = (sock->framing_type == ICE_TCP_FRAMING) ? sizeof(nr_frame_header) : 0;
283 :
284 0 : if (sock->read_state == NR_ICE_SOCKET_READ_FAILED) {
285 0 : ABORT(R_FAILED);
286 : }
287 :
288 0 : while (sock->bytes_needed) {
289 : /* Read all the expected bytes */
290 0 : assert(sock->bytes_needed <= sock->buffer_size - sock->bytes_read);
291 :
292 0 : if(r=nr_socket_read(sock->inner,
293 0 : sock->buffer + sock->bytes_read,
294 : sock->bytes_needed, &bytes_read, 0))
295 0 : ABORT(r);
296 :
297 0 : assert(bytes_read <= sock->bytes_needed);
298 0 : sock->bytes_needed -= bytes_read;
299 0 : sock->bytes_read += bytes_read;
300 :
301 : /* Unfinished */
302 0 : if (sock->bytes_needed)
303 0 : ABORT(R_WOULDBLOCK);
304 :
305 : /* No more bytes expected */
306 0 : if (sock->read_state == NR_ICE_SOCKET_READ_NONE) {
307 : size_t remaining_length;
308 0 : if (sock->framing_type == ICE_TCP_FRAMING) {
309 0 : if (sock->bytes_read < sizeof(nr_frame_header))
310 0 : ABORT(R_BAD_DATA);
311 0 : remaining_length = ntohs(frame->frame_length);
312 : } else {
313 : int tmp_length;
314 :
315 : /* Parse the header */
316 0 : if (r = nr_stun_message_length(sock->buffer, sock->bytes_read, &tmp_length))
317 0 : ABORT(r);
318 0 : assert(tmp_length >= 0);
319 0 : if (tmp_length < 0)
320 0 : ABORT(R_BAD_DATA);
321 0 : remaining_length = tmp_length;
322 :
323 : }
324 : /* Check to see if we have enough room */
325 0 : if ((sock->buffer_size - sock->bytes_read) < remaining_length)
326 0 : ABORT(R_BAD_DATA);
327 :
328 0 : sock->read_state = NR_ICE_SOCKET_READ_HDR;
329 : /* Set ourselves up to read the rest of the data */
330 0 : sock->bytes_needed = remaining_length;
331 : }
332 : }
333 :
334 0 : assert(skip_hdr_size <= sock->bytes_read);
335 0 : if (skip_hdr_size > sock->bytes_read)
336 0 : ABORT(R_BAD_DATA);
337 0 : sock->bytes_read -= skip_hdr_size;
338 :
339 0 : if (maxlen < sock->bytes_read)
340 0 : ABORT(R_BAD_ARGS);
341 :
342 0 : *len = sock->bytes_read;
343 0 : memcpy(buf, sock->buffer + skip_hdr_size, sock->bytes_read);
344 :
345 0 : sock->bytes_read = 0;
346 0 : sock->read_state = NR_ICE_SOCKET_READ_NONE;
347 0 : sock->bytes_needed = (sock->framing_type == ICE_TCP_FRAMING) ? sizeof(nr_frame_header) : sizeof(nr_stun_message_header);
348 :
349 0 : assert(!nr_transport_addr_is_wildcard(&sock->remote_addr));
350 0 : if (!nr_transport_addr_is_wildcard(&sock->remote_addr)) {
351 0 : if ((r=nr_transport_addr_copy(from, &sock->remote_addr)))
352 0 : ABORT(r);
353 : }
354 :
355 0 : _status=0;
356 : abort:
357 0 : if (_status && (_status != R_WOULDBLOCK)) {
358 0 : nr_socket_buffered_stun_failed(sock);
359 : }
360 :
361 0 : return(_status);
362 : }
363 :
364 0 : static int nr_socket_buffered_stun_getfd(void *obj, NR_SOCKET *fd)
365 : {
366 0 : nr_socket_buffered_stun *sock = (nr_socket_buffered_stun *)obj;
367 :
368 0 : return nr_socket_getfd(sock->inner, fd);
369 : }
370 :
371 0 : static int nr_socket_buffered_stun_getaddr(void *obj, nr_transport_addr *addrp)
372 : {
373 0 : nr_socket_buffered_stun *sock = (nr_socket_buffered_stun *)obj;
374 :
375 0 : return nr_socket_getaddr(sock->inner, addrp);
376 : }
377 :
378 0 : static int nr_socket_buffered_stun_close(void *obj)
379 : {
380 0 : nr_socket_buffered_stun *sock = (nr_socket_buffered_stun *)obj;
381 : NR_SOCKET fd;
382 :
383 : /* Cancel waiting on the socket */
384 0 : if (sock->inner && !nr_socket_getfd(sock->inner, &fd)) {
385 0 : NR_ASYNC_CANCEL(fd, NR_ASYNC_WAIT_WRITE);
386 : }
387 :
388 0 : return nr_socket_close(sock->inner);
389 : }
390 :
391 0 : static int nr_socket_buffered_stun_listen(void *obj, int backlog)
392 : {
393 : int r, _status;
394 0 : nr_socket_buffered_stun *sock = (nr_socket_buffered_stun *)obj;
395 :
396 0 : if (!sock->inner)
397 0 : ABORT(R_FAILED);
398 :
399 0 : if ((r=nr_socket_listen(sock->inner, backlog)))
400 0 : ABORT(r);
401 :
402 0 : _status=0;
403 : abort:
404 0 : return(_status);
405 : }
406 :
407 :
408 0 : static int nr_socket_buffered_stun_accept(void *obj, nr_transport_addr *addrp, nr_socket **sockp)
409 : {
410 0 : nr_socket_buffered_stun *bsock = (nr_socket_buffered_stun *)obj;
411 :
412 0 : return nr_socket_accept(bsock->inner, addrp, sockp);
413 : }
414 :
415 0 : static void nr_socket_buffered_stun_connected_cb(NR_SOCKET s, int how, void *arg)
416 : {
417 0 : nr_socket_buffered_stun *sock = (nr_socket_buffered_stun *)arg;
418 : int r, _status;
419 : NR_SOCKET fd;
420 :
421 0 : assert(!sock->connected);
422 :
423 0 : sock->connected = 1;
424 :
425 0 : if ((r=nr_socket_getfd(sock->inner, &fd)))
426 0 : ABORT(r);
427 0 : NR_ASYNC_CANCEL(fd, NR_ASYNC_WAIT_WRITE);
428 :
429 : // once connected arm for read
430 0 : if (sock->readable_cb) {
431 0 : NR_ASYNC_WAIT(fd, NR_ASYNC_WAIT_READ, sock->readable_cb, sock->readable_cb_arg);
432 : }
433 :
434 0 : if (sock->pending) {
435 0 : r_log(LOG_GENERIC, LOG_INFO, "Invoking writable_cb on connected (%u)", (uint32_t) sock->pending);
436 0 : nr_socket_buffered_stun_writable_cb(s, how, arg);
437 : }
438 :
439 0 : _status=0;
440 : abort:
441 0 : if (_status) {
442 0 : r_log(LOG_GENERIC, LOG_ERR, "Failure in nr_socket_buffered_stun_connected_cb: %d", _status);
443 :
444 : }
445 0 : }
446 :
447 0 : static int nr_socket_buffered_stun_connect(void *obj, nr_transport_addr *addr)
448 : {
449 0 : nr_socket_buffered_stun *sock = (nr_socket_buffered_stun *)obj;
450 : int r, _status;
451 :
452 0 : if ((r=nr_transport_addr_copy(&sock->remote_addr, addr)))
453 0 : ABORT(r);
454 :
455 0 : if ((r=nr_socket_connect(sock->inner, addr))) {
456 0 : if (r == R_WOULDBLOCK) {
457 : NR_SOCKET fd;
458 :
459 0 : if ((r=nr_socket_getfd(sock->inner, &fd)))
460 0 : ABORT(r);
461 :
462 0 : NR_ASYNC_WAIT(fd, NR_ASYNC_WAIT_WRITE, nr_socket_buffered_stun_connected_cb, sock);
463 0 : ABORT(R_WOULDBLOCK);
464 : }
465 0 : ABORT(r);
466 : } else {
467 0 : r_log(LOG_GENERIC, LOG_INFO, "Connected without blocking");
468 0 : sock->connected = 1;
469 : }
470 :
471 0 : _status=0;
472 : abort:
473 0 : return(_status);
474 : }
475 :
476 0 : static int nr_socket_buffered_stun_arm_writable_cb(nr_socket_buffered_stun *sock)
477 : {
478 : int r, _status;
479 : NR_SOCKET fd;
480 :
481 0 : if ((r=nr_socket_getfd(sock->inner, &fd)))
482 0 : ABORT(r);
483 :
484 0 : NR_ASYNC_WAIT(fd, NR_ASYNC_WAIT_WRITE, nr_socket_buffered_stun_writable_cb, sock);
485 :
486 0 : _status=0;
487 : abort:
488 0 : return(_status);
489 : }
490 :
491 0 : static int nr_socket_buffered_stun_write(void *obj,const void *msg, size_t len, size_t *written)
492 : {
493 0 : nr_socket_buffered_stun *sock = (nr_socket_buffered_stun *)obj;
494 0 : int already_armed = 0;
495 : int r,_status;
496 0 : size_t written2 = 0;
497 0 : size_t original_len = len;
498 :
499 : /* Buffers are close to full, report error. Do this now so we never
500 : get partial writes */
501 0 : if ((sock->pending + len) > sock->max_pending) {
502 0 : r_log(LOG_GENERIC, LOG_INFO, "Write buffer for %s full (%u + %u > %u) - re-arming @%p",
503 0 : sock->remote_addr.as_string, (uint32_t)sock->pending, (uint32_t)len, (uint32_t)sock->max_pending,
504 : &(sock->pending));
505 0 : ABORT(R_WOULDBLOCK);
506 : }
507 :
508 :
509 0 : if (sock->connected && !sock->pending) {
510 0 : r = nr_socket_write(sock->inner, msg, len, &written2, 0);
511 0 : if (r) {
512 0 : if (r != R_WOULDBLOCK) {
513 0 : r_log(LOG_GENERIC, LOG_ERR, "Write error for %s - %d",
514 0 : sock->remote_addr.as_string, r);
515 0 : ABORT(r);
516 : }
517 0 : r_log(LOG_GENERIC, LOG_INFO, "Write of %" PRIu64 " blocked for %s",
518 0 : (uint64_t) len, sock->remote_addr.as_string);
519 :
520 0 : written2=0;
521 : }
522 : } else {
523 0 : already_armed = 1;
524 : }
525 :
526 : /* Buffer what's left */
527 0 : len -= written2;
528 :
529 0 : if (len) {
530 0 : if ((r=nr_p_buf_write_to_chain(sock->p_bufs, &sock->pending_writes,
531 : ((UCHAR *)msg) + written2, len))) {
532 0 : r_log(LOG_GENERIC, LOG_ERR, "Write_to_chain error for %s - %d",
533 0 : sock->remote_addr.as_string, r);
534 :
535 0 : ABORT(r);
536 : }
537 :
538 0 : sock->pending += len;
539 : }
540 :
541 0 : if (sock->pending) {
542 0 : if (!already_armed) {
543 0 : if ((r=nr_socket_buffered_stun_arm_writable_cb(sock)))
544 0 : ABORT(r);
545 : }
546 0 : r_log(LOG_GENERIC, LOG_INFO, "Write buffer not empty for %s %u - %s armed (@%p),%s connected",
547 0 : sock->remote_addr.as_string, (uint32_t)sock->pending,
548 : already_armed ? "already" : "", &sock->pending,
549 0 : sock->connected ? "" : " not");
550 : }
551 :
552 0 : *written = original_len;
553 :
554 0 : _status=0;
555 : abort:
556 0 : return _status;
557 : }
558 :
559 0 : static void nr_socket_buffered_stun_writable_cb(NR_SOCKET s, int how, void *arg)
560 : {
561 0 : nr_socket_buffered_stun *sock = (nr_socket_buffered_stun *)arg;
562 : int r,_status;
563 : nr_p_buf *n1, *n2;
564 :
565 0 : if (sock->read_state == NR_ICE_SOCKET_READ_FAILED) {
566 0 : ABORT(R_FAILED);
567 : }
568 :
569 : /* Try to flush */
570 0 : STAILQ_FOREACH_SAFE(n1, &sock->pending_writes, entry, n2) {
571 0 : size_t written = 0;
572 :
573 0 : if ((r=nr_socket_write(sock->inner, n1->data + n1->r_offset,
574 0 : n1->length - n1->r_offset,
575 : &written, 0))) {
576 :
577 0 : r_log(LOG_GENERIC, LOG_ERR, "Write error for %s - %d",
578 0 : sock->remote_addr.as_string, r);
579 0 : ABORT(r);
580 : }
581 :
582 0 : n1->r_offset += written;
583 0 : assert(sock->pending >= written);
584 0 : sock->pending -= written;
585 :
586 0 : if (n1->r_offset < n1->length) {
587 : /* We wrote something, but not everything */
588 0 : r_log(LOG_GENERIC, LOG_INFO, "Write in callback didn't write all (remaining %u of %u) for %s",
589 0 : n1->length - n1->r_offset, n1->length,
590 0 : sock->remote_addr.as_string);
591 0 : ABORT(R_WOULDBLOCK);
592 : }
593 :
594 : /* We are done with this p_buf */
595 0 : STAILQ_REMOVE_HEAD(&sock->pending_writes, entry);
596 0 : nr_p_buf_free(sock->p_bufs, n1);
597 : }
598 :
599 0 : assert(!sock->pending);
600 0 : _status=0;
601 : abort:
602 0 : r_log(LOG_GENERIC, LOG_INFO, "Writable_cb %s (%u (%p) pending)",
603 0 : sock->remote_addr.as_string, (uint32_t)sock->pending, &(sock->pending));
604 0 : if (_status && _status != R_WOULDBLOCK) {
605 0 : r_log(LOG_GENERIC, LOG_ERR, "Failure in writable_cb: %d", _status);
606 0 : nr_socket_buffered_stun_failed(sock);
607 0 : } else if (sock->pending) {
608 0 : nr_socket_buffered_stun_arm_writable_cb(sock);
609 : }
610 0 : }
|
