libnfs: Don't perform operations on nfsfh after user callback
[deb_libnfs.git] / lib / socket.c
... / ...
CommitLineData
1/*
2 Copyright (C) 2010 by Ronnie Sahlberg <ronniesahlberg@gmail.com>
3
4 This program is free software; you can redistribute it and/or modify
5 it under the terms of the GNU Lesser General Public License as published by
6 the Free Software Foundation; either version 2.1 of the License, or
7 (at your option) any later version.
8
9 This program is distributed in the hope that it will be useful,
10 but WITHOUT ANY WARRANTY; without even the implied warranty of
11 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 GNU Lesser General Public License for more details.
13
14 You should have received a copy of the GNU Lesser General Public License
15 along with this program; if not, see <http://www.gnu.org/licenses/>.
16*/
17#ifdef HAVE_CONFIG_H
18#include "config.h"
19#endif
20
21#ifdef AROS
22#include "aros_compat.h"
23#endif
24
25#ifdef WIN32
26#include "win32_compat.h"
27#endif
28
29#ifdef HAVE_ARPA_INET_H
30#include <arpa/inet.h>
31#endif
32
33#ifdef HAVE_POLL_H
34#include <poll.h>
35#endif
36
37#ifdef HAVE_UNISTD_H
38#include <unistd.h>
39#endif
40
41#ifdef HAVE_SYS_IOCTL_H
42#include <sys/ioctl.h>
43#endif
44
45#ifdef HAVE_SYS_SOCKET_H
46#include <sys/socket.h>
47#endif
48
49#ifdef HAVE_NETINET_TCP_H
50#include <netinet/tcp.h>
51#endif
52
53#ifdef HAVE_NETDB_H
54#include <netdb.h>
55#endif
56
57#ifdef HAVE_SYS_FILIO_H
58#include <sys/filio.h>
59#endif
60
61#ifdef HAVE_SYS_SOCKIO_H
62#include <sys/sockio.h>
63#endif
64
65#include <stdio.h>
66#include <stdlib.h>
67#include <assert.h>
68#include <fcntl.h>
69#include <string.h>
70#include <errno.h>
71#include <sys/types.h>
72#include "libnfs-zdr.h"
73#include "libnfs.h"
74#include "libnfs-raw.h"
75#include "libnfs-private.h"
76#include "slist.h"
77
78#ifdef WIN32
79//has to be included after stdlib!!
80#include "win32_errnowrapper.h"
81#endif
82
83static int rpc_reconnect_requeue(struct rpc_context *rpc);
84static int rpc_connect_sockaddr_async(struct rpc_context *rpc, struct sockaddr_storage *s);
85
86static void set_nonblocking(int fd)
87{
88 int v = 0;
89#if defined(WIN32)
90 long nonblocking=1;
91 v = ioctl(fd, FIONBIO, &nonblocking);
92#else
93 v = fcntl(fd, F_GETFL, 0);
94 fcntl(fd, F_SETFL, v | O_NONBLOCK);
95#endif //FIXME
96}
97
98static void set_nolinger(int fd)
99{
100 struct linger lng;
101 lng.l_onoff = 1;
102 lng.l_linger = 0;
103 setsockopt(fd, SOL_SOCKET, SO_LINGER, &lng, sizeof(lng));
104}
105
106#ifdef HAVE_NETINET_TCP_H
107int set_tcp_sockopt(int sockfd, int optname, int value)
108{
109 int level;
110
111 #if defined(__FreeBSD__) || defined(__sun) || (defined(__APPLE__) && defined(__MACH__))
112 struct protoent *buf;
113
114 if ((buf = getprotobyname("tcp")) != NULL)
115 level = buf->p_proto;
116 else
117 return -1;
118 #else
119 level = SOL_TCP;
120 #endif
121
122 return setsockopt(sockfd, level, optname, (char *)&value, sizeof(value));
123}
124#endif
125
126int rpc_get_fd(struct rpc_context *rpc)
127{
128 assert(rpc->magic == RPC_CONTEXT_MAGIC);
129
130 return rpc->fd;
131}
132
133static int rpc_has_queue(struct rpc_queue *q)
134{
135 return q->head != NULL;
136}
137
138int rpc_which_events(struct rpc_context *rpc)
139{
140 int events;
141
142 assert(rpc->magic == RPC_CONTEXT_MAGIC);
143
144 events = rpc->is_connected ? POLLIN : POLLOUT;
145
146 if (rpc->is_udp != 0) {
147 /* for udp sockets we only wait for pollin */
148 return POLLIN;
149 }
150
151 if (rpc_has_queue(&rpc->outqueue)) {
152 events |= POLLOUT;
153 }
154 return events;
155}
156
157static int rpc_write_to_socket(struct rpc_context *rpc)
158{
159 int32_t count;
160 struct rpc_pdu *pdu;
161
162 assert(rpc->magic == RPC_CONTEXT_MAGIC);
163
164 if (rpc->fd == -1) {
165 rpc_set_error(rpc, "trying to write but not connected");
166 return -1;
167 }
168
169 while ((pdu = rpc->outqueue.head) != NULL) {
170 int64_t total;
171
172 total = pdu->outdata.size;
173
174 count = send(rpc->fd, pdu->outdata.data + pdu->written, total - pdu->written, 0);
175 if (count == -1) {
176 if (errno == EAGAIN || errno == EWOULDBLOCK) {
177 return 0;
178 }
179 rpc_set_error(rpc, "Error when writing to socket :%s(%d)", strerror(errno), errno);
180 return -1;
181 }
182
183 pdu->written += count;
184 if (pdu->written == total) {
185 unsigned int hash;
186
187 rpc->outqueue.head = pdu->next;
188 if (pdu->next == NULL)
189 rpc->outqueue.tail = NULL;
190
191 hash = rpc_hash_xid(pdu->xid);
192 rpc_enqueue(&rpc->waitpdu[hash], pdu);
193 }
194 }
195 return 0;
196}
197
198static int rpc_read_from_socket(struct rpc_context *rpc)
199{
200 int available;
201 int size;
202 int pdu_size;
203 int32_t count;
204
205 assert(rpc->magic == RPC_CONTEXT_MAGIC);
206
207 if (ioctl(rpc->fd, FIONREAD, &available) != 0) {
208 rpc_set_error(rpc, "Ioctl FIONREAD returned error : %d. Closing socket.", errno);
209 return -1;
210 }
211
212 if (available == 0) {
213 rpc_set_error(rpc, "Socket has been closed");
214 return -1;
215 }
216
217 if (rpc->is_udp) {
218 char *buf;
219 socklen_t socklen = sizeof(rpc->udp_src);
220
221 buf = malloc(available);
222 if (buf == NULL) {
223 rpc_set_error(rpc, "Failed to malloc buffer for recvfrom");
224 return -1;
225 }
226 count = recvfrom(rpc->fd, buf, available, MSG_DONTWAIT, (struct sockaddr *)&rpc->udp_src, &socklen);
227 if (count < 0) {
228 rpc_set_error(rpc, "Failed recvfrom: %s", strerror(errno));
229 free(buf);
230 return -1;
231 }
232 if (rpc_process_pdu(rpc, buf, count) != 0) {
233 rpc_set_error(rpc, "Invalid/garbage pdu received from server. Ignoring PDU");
234 free(buf);
235 return -1;
236 }
237 free(buf);
238 return 0;
239 }
240
241 /* read record marker, 4 bytes at the beginning of every pdu */
242 if (rpc->inbuf == NULL) {
243 rpc->insize = 4;
244 rpc->inbuf = malloc(rpc->insize);
245 if (rpc->inbuf == NULL) {
246 rpc_set_error(rpc, "Failed to allocate buffer for record marker, errno:%d. Closing socket.", errno);
247 return -1;
248 }
249 }
250 if (rpc->inpos < 4) {
251 size = 4 - rpc->inpos;
252
253 count = recv(rpc->fd, rpc->inbuf + rpc->inpos, size, 0);
254 if (count == -1) {
255 if (errno == EINTR) {
256 return 0;
257 }
258 rpc_set_error(rpc, "Read from socket failed, errno:%d. Closing socket.", errno);
259 return -1;
260 }
261 available -= count;
262 rpc->inpos += count;
263 }
264
265 if (available == 0) {
266 return 0;
267 }
268
269 pdu_size = rpc_get_pdu_size(rpc->inbuf);
270 if (rpc->insize < pdu_size) {
271 unsigned char *buf;
272
273 buf = malloc(pdu_size);
274 if (buf == NULL) {
275 rpc_set_error(rpc, "Failed to allocate buffer of %d bytes for pdu, errno:%d. Closing socket.", pdu_size, errno);
276 return -1;
277 }
278 memcpy(buf, rpc->inbuf, rpc->insize);
279 free(rpc->inbuf);
280 rpc->inbuf = buf;
281 rpc->insize = rpc_get_pdu_size(rpc->inbuf);
282 }
283
284 size = available;
285 if (size > rpc->insize - rpc->inpos) {
286 size = rpc->insize - rpc->inpos;
287 }
288
289 count = recv(rpc->fd, rpc->inbuf + rpc->inpos, size, 0);
290 if (count == -1) {
291 if (errno == EINTR) {
292 return 0;
293 }
294 rpc_set_error(rpc, "Read from socket failed, errno:%d. Closing socket.", errno);
295 return -1;
296 }
297 available -= count;
298 rpc->inpos += count;
299
300 if (rpc->inpos == rpc->insize) {
301 char *buf = rpc->inbuf;
302
303 rpc->inbuf = NULL;
304 rpc->insize = 0;
305 rpc->inpos = 0;
306
307 if (rpc_process_pdu(rpc, buf, pdu_size) != 0) {
308 rpc_set_error(rpc, "Invalid/garbage pdu received from server. Closing socket");
309 return -1;
310 }
311 free(buf);
312 }
313
314 return 0;
315}
316
317
318
319int rpc_service(struct rpc_context *rpc, int revents)
320{
321 assert(rpc->magic == RPC_CONTEXT_MAGIC);
322
323 if (revents & POLLERR) {
324#ifdef WIN32
325 char err = 0;
326#else
327 int err = 0;
328#endif
329 socklen_t err_size = sizeof(err);
330
331 if (getsockopt(rpc->fd, SOL_SOCKET, SO_ERROR,
332 (char *)&err, &err_size) != 0 || err != 0) {
333 if (err == 0) {
334 err = errno;
335 }
336 rpc_set_error(rpc, "rpc_service: socket error "
337 "%s(%d).",
338 strerror(err), err);
339 } else {
340 rpc_set_error(rpc, "rpc_service: POLLERR, "
341 "Unknown socket error.");
342 }
343 if (rpc->connect_cb != NULL) {
344 rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data);
345 }
346 return -1;
347 }
348 if (revents & POLLHUP) {
349 rpc_set_error(rpc, "Socket failed with POLLHUP");
350 if (rpc->connect_cb != NULL) {
351 rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data);
352 }
353 return -1;
354 }
355
356 if (rpc->is_connected == 0 && rpc->fd != -1 && revents&POLLOUT) {
357 int err = 0;
358 socklen_t err_size = sizeof(err);
359
360 if (getsockopt(rpc->fd, SOL_SOCKET, SO_ERROR,
361 (char *)&err, &err_size) != 0 || err != 0) {
362 if (err == 0) {
363 err = errno;
364 }
365 rpc_set_error(rpc, "rpc_service: socket error "
366 "%s(%d) while connecting.",
367 strerror(err), err);
368 if (rpc->connect_cb != NULL) {
369 rpc->connect_cb(rpc, RPC_STATUS_ERROR,
370 NULL, rpc->connect_data);
371 }
372 return -1;
373 }
374
375 rpc->is_connected = 1;
376 if (rpc->connect_cb != NULL) {
377 rpc->connect_cb(rpc, RPC_STATUS_SUCCESS, NULL, rpc->connect_data);
378 }
379 return 0;
380 }
381
382 if (revents & POLLIN) {
383 if (rpc_read_from_socket(rpc) != 0) {
384 rpc_reconnect_requeue(rpc);
385 return 0;
386 }
387 }
388
389 if (revents & POLLOUT && rpc_has_queue(&rpc->outqueue)) {
390 if (rpc_write_to_socket(rpc) != 0) {
391 rpc_set_error(rpc, "write to socket failed");
392 return -1;
393 }
394 }
395
396 return 0;
397}
398
399void rpc_set_autoreconnect(struct rpc_context *rpc)
400{
401 assert(rpc->magic == RPC_CONTEXT_MAGIC);
402
403 rpc->auto_reconnect = 1;
404}
405
406void rpc_unset_autoreconnect(struct rpc_context *rpc)
407{
408 assert(rpc->magic == RPC_CONTEXT_MAGIC);
409
410 rpc->auto_reconnect = 0;
411}
412
413void rpc_set_tcp_syncnt(struct rpc_context *rpc, int v)
414{
415 assert(rpc->magic == RPC_CONTEXT_MAGIC);
416
417 rpc->tcp_syncnt = v;
418}
419
420#ifndef TCP_SYNCNT
421#define TCP_SYNCNT 7
422#endif
423
424static int rpc_connect_sockaddr_async(struct rpc_context *rpc, struct sockaddr_storage *s)
425{
426 int socksize;
427
428 assert(rpc->magic == RPC_CONTEXT_MAGIC);
429
430 switch (s->ss_family) {
431 case AF_INET:
432 socksize = sizeof(struct sockaddr_in);
433 rpc->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
434#ifdef HAVE_NETINET_TCP_H
435 if (rpc->tcp_syncnt != RPC_PARAM_UNDEFINED) {
436 set_tcp_sockopt(rpc->fd, TCP_SYNCNT, rpc->tcp_syncnt);
437 }
438#endif
439 break;
440 case AF_INET6:
441 socksize = sizeof(struct sockaddr_in6);
442 rpc->fd = socket(AF_INET6, SOCK_STREAM, IPPROTO_TCP);
443#ifdef HAVE_NETINET_TCP_H
444 if (rpc->tcp_syncnt != RPC_PARAM_UNDEFINED) {
445 set_tcp_sockopt(rpc->fd, TCP_SYNCNT, rpc->tcp_syncnt);
446 }
447#endif
448 break;
449 default:
450 rpc_set_error(rpc, "Can not handle AF_FAMILY:%d", s->ss_family);
451 return -1;
452 }
453
454 if (rpc->fd == -1) {
455 rpc_set_error(rpc, "Failed to open socket");
456 return -1;
457 }
458
459 /* Some systems allow you to set capabilities on an executable
460 * to allow the file to be executed with privilege to bind to
461 * privileged system ports, even if the user is not root.
462 *
463 * Opportunistically try to bind the socket to a low numbered
464 * system port in the hope that the user is either root or the
465 * executable has the CAP_NET_BIND_SERVICE.
466 *
467 * As soon as we fail the bind() with EACCES we know we will never
468 * be able to bind to a system port so we terminate the loop.
469 *
470 * On linux, use
471 * sudo setcap 'cap_net_bind_service=+ep' /path/executable
472 * to make the executable able to bind to a system port.
473 *
474 * On Windows, there is no concept of privileged ports. Thus
475 * binding will usually succeed.
476 */
477 {
478 struct sockaddr_storage ss;
479 static int portOfs = 0;
480 const int firstPort = 512; /* >= 512 according to Sun docs */
481 const int portCount = IPPORT_RESERVED - firstPort;
482 int startOfs, port, rc;
483
484 if (portOfs == 0) {
485 portOfs = time(NULL) % 400;
486 }
487 startOfs = portOfs;
488 do {
489 rc = -1;
490 port = htons(firstPort + portOfs);
491 portOfs = (portOfs + 1) % portCount;
492
493 /* skip well-known ports */
494 if (!getservbyport(port, "tcp")) {
495 memset(&ss, 0, sizeof(ss));
496
497 switch (s->ss_family) {
498 case AF_INET:
499 ((struct sockaddr_in *)&ss)->sin_port = port;
500 ((struct sockaddr_in *)&ss)->sin_family = AF_INET;
501#ifdef HAVE_SOCKADDR_LEN
502 ((struct sockaddr_in *)&ss)->sin_len = sizeof(struct sockaddr_in);
503#endif
504 break;
505 case AF_INET6:
506 ((struct sockaddr_in6 *)&ss)->sin6_port = port;
507 ((struct sockaddr_in6 *)&ss)->sin6_family = AF_INET6;
508#ifdef HAVE_SOCKADDR_LEN
509 ((struct sockaddr_in6 *)&ss)->sin6_len = sizeof(struct sockaddr_in6);
510#endif
511 break;
512 }
513
514 rc = bind(rpc->fd, (struct sockaddr *)&ss, socksize);
515#if !defined(WIN32)
516 /* we got EACCES, so don't try again */
517 if (rc != 0 && errno == EACCES)
518 break;
519#endif
520 }
521 } while (rc != 0 && portOfs != startOfs);
522 }
523
524 set_nonblocking(rpc->fd);
525 set_nolinger(rpc->fd);
526
527 if (connect(rpc->fd, (struct sockaddr *)s, socksize) != 0 && errno != EINPROGRESS) {
528 rpc_set_error(rpc, "connect() to server failed. %s(%d)", strerror(errno), errno);
529 return -1;
530 }
531
532 return 0;
533}
534
535int rpc_connect_async(struct rpc_context *rpc, const char *server, int port, rpc_cb cb, void *private_data)
536{
537 struct addrinfo *ai = NULL;
538
539 assert(rpc->magic == RPC_CONTEXT_MAGIC);
540
541 if (rpc->fd != -1) {
542 rpc_set_error(rpc, "Trying to connect while already connected");
543 return -1;
544 }
545
546 if (rpc->is_udp != 0) {
547 rpc_set_error(rpc, "Trying to connect on UDP socket");
548 return -1;
549 }
550
551 rpc->auto_reconnect = 0;
552
553 if (getaddrinfo(server, NULL, NULL, &ai) != 0) {
554 rpc_set_error(rpc, "Invalid address:%s. "
555 "Can not resolv into IPv4/v6 structure.", server);
556 return -1;
557 }
558
559 switch (ai->ai_family) {
560 case AF_INET:
561 ((struct sockaddr_in *)&rpc->s)->sin_family = ai->ai_family;
562 ((struct sockaddr_in *)&rpc->s)->sin_port = htons(port);
563 ((struct sockaddr_in *)&rpc->s)->sin_addr = ((struct sockaddr_in *)(ai->ai_addr))->sin_addr;
564#ifdef HAVE_SOCKADDR_LEN
565 ((struct sockaddr_in *)&rpc->s)->sin_len = sizeof(struct sockaddr_in);
566#endif
567 break;
568 case AF_INET6:
569 ((struct sockaddr_in6 *)&rpc->s)->sin6_family = ai->ai_family;
570 ((struct sockaddr_in6 *)&rpc->s)->sin6_port = htons(port);
571 ((struct sockaddr_in6 *)&rpc->s)->sin6_addr = ((struct sockaddr_in6 *)(ai->ai_addr))->sin6_addr;
572#ifdef HAVE_SOCKADDR_LEN
573 ((struct sockaddr_in6 *)&rpc->s)->sin6_len = sizeof(struct sockaddr_in6);
574#endif
575 break;
576 }
577
578 rpc->connect_cb = cb;
579 rpc->connect_data = private_data;
580
581 freeaddrinfo(ai);
582
583 if (rpc_connect_sockaddr_async(rpc, &rpc->s) != 0) {
584 return -1;
585 }
586
587 return 0;
588}
589
590int rpc_disconnect(struct rpc_context *rpc, char *error)
591{
592 assert(rpc->magic == RPC_CONTEXT_MAGIC);
593
594 rpc_unset_autoreconnect(rpc);
595
596 if (rpc->fd != -1) {
597 close(rpc->fd);
598 }
599 rpc->fd = -1;
600
601 rpc->is_connected = 0;
602
603 rpc_error_all_pdus(rpc, error);
604
605 return 0;
606}
607
608static void reconnect_cb(struct rpc_context *rpc, int status, void *data _U_, void *private_data)
609{
610 assert(rpc->magic == RPC_CONTEXT_MAGIC);
611
612 if (status != RPC_STATUS_SUCCESS) {
613 rpc_error_all_pdus(rpc, "RPC ERROR: Failed to reconnect async");
614 return;
615 }
616
617 rpc->is_connected = 1;
618 rpc->connect_cb = NULL;
619}
620
621/* disconnect but do not error all PDUs, just move pdus in-flight back to the outqueue and reconnect */
622static int rpc_reconnect_requeue(struct rpc_context *rpc)
623{
624 struct rpc_pdu *pdu;
625 unsigned int i;
626
627 assert(rpc->magic == RPC_CONTEXT_MAGIC);
628
629 if (rpc->fd != -1) {
630 close(rpc->fd);
631 }
632 rpc->fd = -1;
633
634 rpc->is_connected = 0;
635
636 /* socket is closed so we will not get any replies to any commands
637 * in flight. Move them all over from the waitpdu queue back to the out queue
638 */
639 for (i = 0; i < HASHES; i++) {
640 struct rpc_queue *q = &rpc->waitpdu[i];
641
642 for (pdu=q->head; pdu; pdu=pdu->next) {
643 rpc_return_to_queue(&rpc->outqueue, pdu);
644 /* we have to re-send the whole pdu again */
645 pdu->written = 0;
646 }
647 rpc_reset_queue(q);
648 }
649
650 if (rpc->auto_reconnect != 0) {
651 rpc->connect_cb = reconnect_cb;
652
653 if (rpc_connect_sockaddr_async(rpc, &rpc->s) != 0) {
654 rpc_error_all_pdus(rpc, "RPC ERROR: Failed to reconnect async");
655 return -1;
656 }
657 }
658
659 return 0;
660}
661
662
663int rpc_bind_udp(struct rpc_context *rpc, char *addr, int port)
664{
665 struct addrinfo *ai = NULL;
666 char service[6];
667
668 assert(rpc->magic == RPC_CONTEXT_MAGIC);
669
670 if (rpc->is_udp == 0) {
671 rpc_set_error(rpc, "Cant not bind UDP. Not UDP context");
672 return -1;
673 }
674
675 sprintf(service, "%d", port);
676 if (getaddrinfo(addr, service, NULL, &ai) != 0) {
677 rpc_set_error(rpc, "Invalid address:%s. "
678 "Can not resolv into IPv4/v6 structure.", addr);
679 return -1;
680 }
681
682 switch(ai->ai_family) {
683 case AF_INET:
684 rpc->fd = socket(ai->ai_family, SOCK_DGRAM, 0);
685 if (rpc->fd == -1) {
686 rpc_set_error(rpc, "Failed to create UDP socket: %s", strerror(errno));
687 freeaddrinfo(ai);
688 return -1;
689 }
690
691 if (bind(rpc->fd, (struct sockaddr *)ai->ai_addr, sizeof(struct sockaddr_in)) != 0) {
692 rpc_set_error(rpc, "Failed to bind to UDP socket: %s",strerror(errno));
693 freeaddrinfo(ai);
694 return -1;
695 }
696 break;
697 default:
698 rpc_set_error(rpc, "Can not handle UPD sockets of family %d yet", ai->ai_family);
699 freeaddrinfo(ai);
700 return -1;
701 }
702
703 freeaddrinfo(ai);
704
705 return 0;
706}
707
708int rpc_set_udp_destination(struct rpc_context *rpc, char *addr, int port, int is_broadcast)
709{
710 struct addrinfo *ai = NULL;
711 char service[6];
712
713 assert(rpc->magic == RPC_CONTEXT_MAGIC);
714
715 if (rpc->is_udp == 0) {
716 rpc_set_error(rpc, "Can not set destination sockaddr. Not UDP context");
717 return -1;
718 }
719
720 sprintf(service, "%d", port);
721 if (getaddrinfo(addr, service, NULL, &ai) != 0) {
722 rpc_set_error(rpc, "Invalid address:%s. "
723 "Can not resolv into IPv4/v6 structure.", addr);
724 return -1;
725 }
726
727 if (rpc->udp_dest) {
728 free(rpc->udp_dest);
729 rpc->udp_dest = NULL;
730 }
731 rpc->udp_dest = malloc(ai->ai_addrlen);
732 if (rpc->udp_dest == NULL) {
733 rpc_set_error(rpc, "Out of memory. Failed to allocate sockaddr structure");
734 freeaddrinfo(ai);
735 return -1;
736 }
737 memcpy(rpc->udp_dest, ai->ai_addr, ai->ai_addrlen);
738 freeaddrinfo(ai);
739
740 rpc->is_broadcast = is_broadcast;
741 setsockopt(rpc->fd, SOL_SOCKET, SO_BROADCAST, (char *)&is_broadcast, sizeof(is_broadcast));
742
743 return 0;
744}
745
746struct sockaddr *rpc_get_recv_sockaddr(struct rpc_context *rpc)
747{
748 assert(rpc->magic == RPC_CONTEXT_MAGIC);
749
750 return (struct sockaddr *)&rpc->udp_src;
751}
752
753int rpc_queue_length(struct rpc_context *rpc)
754{
755 int i=0;
756 struct rpc_pdu *pdu;
757 unsigned int n;
758
759 assert(rpc->magic == RPC_CONTEXT_MAGIC);
760
761 for(pdu = rpc->outqueue.head; pdu; pdu = pdu->next) {
762 i++;
763 }
764
765 for (n = 0; n < HASHES; n++) {
766 struct rpc_queue *q = &rpc->waitpdu[n];
767
768 for(pdu = q->head; pdu; pdu = pdu->next)
769 i++;
770 }
771 return i;
772}
773
774void rpc_set_fd(struct rpc_context *rpc, int fd)
775{
776 assert(rpc->magic == RPC_CONTEXT_MAGIC);
777
778 rpc->fd = fd;
779}