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