Add a rpc_set_fd() fucntion which can be used to swap the underlying socket file...
[deb_libnfs.git] / lib / socket.c
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
83
84 static int rpc_reconnect_requeue(struct rpc_context *rpc);
85 static int rpc_connect_sockaddr_async(struct rpc_context *rpc, struct sockaddr_storage *s);
86
87 static void set_nonblocking(int fd)
88 {
89 int v = 0;
90 #if defined(WIN32)
91 long nonblocking=1;
92 v = ioctl(fd, FIONBIO, &nonblocking);
93 #else
94 v = fcntl(fd, F_GETFL, 0);
95 fcntl(fd, F_SETFL, v | O_NONBLOCK);
96 #endif //FIXME
97 }
98
99 #ifdef HAVE_NETINET_TCP_H
100 int set_tcp_sockopt(int sockfd, int optname, int value)
101 {
102 int level;
103
104 #if defined(__FreeBSD__) || defined(__sun) || (defined(__APPLE__) && defined(__MACH__))
105 struct protoent *buf;
106
107 if ((buf = getprotobyname("tcp")) != NULL)
108 level = buf->p_proto;
109 else
110 return -1;
111 #else
112 level = SOL_TCP;
113 #endif
114
115 return setsockopt(sockfd, level, optname, (char *)&value, sizeof(value));
116 }
117 #endif
118
119 int rpc_get_fd(struct rpc_context *rpc)
120 {
121 assert(rpc->magic == RPC_CONTEXT_MAGIC);
122
123 return rpc->fd;
124 }
125
126 int rpc_which_events(struct rpc_context *rpc)
127 {
128 int events;
129
130 assert(rpc->magic == RPC_CONTEXT_MAGIC);
131
132 events = rpc->is_connected ? POLLIN : POLLOUT;
133
134 if (rpc->is_udp != 0) {
135 /* for udp sockets we only wait for pollin */
136 return POLLIN;
137 }
138
139 if (rpc->outqueue) {
140 events |= POLLOUT;
141 }
142 return events;
143 }
144
145 static int rpc_write_to_socket(struct rpc_context *rpc)
146 {
147 int32_t count;
148
149 assert(rpc->magic == RPC_CONTEXT_MAGIC);
150
151 if (rpc->fd == -1) {
152 rpc_set_error(rpc, "trying to write but not connected");
153 return -1;
154 }
155
156 while (rpc->outqueue != NULL) {
157 int64_t total;
158
159 total = rpc->outqueue->outdata.size;
160
161 count = send(rpc->fd, rpc->outqueue->outdata.data + rpc->outqueue->written, total - rpc->outqueue->written, 0);
162 if (count == -1) {
163 if (errno == EAGAIN || errno == EWOULDBLOCK) {
164 return 0;
165 }
166 rpc_set_error(rpc, "Error when writing to socket :%s(%d)", strerror(errno), errno);
167 return -1;
168 }
169
170 rpc->outqueue->written += count;
171 if (rpc->outqueue->written == total) {
172 struct rpc_pdu *pdu = rpc->outqueue;
173
174 SLIST_REMOVE(&rpc->outqueue, pdu);
175 SLIST_ADD_END(&rpc->waitpdu, pdu);
176 }
177 }
178 return 0;
179 }
180
181 static int rpc_read_from_socket(struct rpc_context *rpc)
182 {
183 int available;
184 int size;
185 int pdu_size;
186 int32_t count;
187
188 assert(rpc->magic == RPC_CONTEXT_MAGIC);
189
190 if (ioctl(rpc->fd, FIONREAD, &available) != 0) {
191 rpc_set_error(rpc, "Ioctl FIONREAD returned error : %d. Closing socket.", errno);
192 return -1;
193 }
194
195 if (available == 0) {
196 rpc_set_error(rpc, "Socket has been closed");
197 return -1;
198 }
199
200 if (rpc->is_udp) {
201 char *buf;
202 socklen_t socklen = sizeof(rpc->udp_src);
203
204 buf = malloc(available);
205 if (buf == NULL) {
206 rpc_set_error(rpc, "Failed to malloc buffer for recvfrom");
207 return -1;
208 }
209 count = recvfrom(rpc->fd, buf, available, MSG_DONTWAIT, (struct sockaddr *)&rpc->udp_src, &socklen);
210 if (count < 0) {
211 rpc_set_error(rpc, "Failed recvfrom: %s", strerror(errno));
212 free(buf);
213 }
214 if (rpc_process_pdu(rpc, buf, count) != 0) {
215 rpc_set_error(rpc, "Invalid/garbage pdu received from server. Ignoring PDU");
216 free(buf);
217 return -1;
218 }
219 free(buf);
220 return 0;
221 }
222
223 /* read record marker, 4 bytes at the beginning of every pdu */
224 if (rpc->inbuf == NULL) {
225 rpc->insize = 4;
226 rpc->inbuf = malloc(rpc->insize);
227 if (rpc->inbuf == NULL) {
228 rpc_set_error(rpc, "Failed to allocate buffer for record marker, errno:%d. Closing socket.", errno);
229 return -1;
230 }
231 }
232 if (rpc->inpos < 4) {
233 size = 4 - rpc->inpos;
234
235 count = recv(rpc->fd, rpc->inbuf + rpc->inpos, size, 0);
236 if (count == -1) {
237 if (errno == EINTR) {
238 return 0;
239 }
240 rpc_set_error(rpc, "Read from socket failed, errno:%d. Closing socket.", errno);
241 return -1;
242 }
243 available -= count;
244 rpc->inpos += count;
245 }
246
247 if (available == 0) {
248 return 0;
249 }
250
251 pdu_size = rpc_get_pdu_size(rpc->inbuf);
252 if (rpc->insize < pdu_size) {
253 unsigned char *buf;
254
255 buf = malloc(pdu_size);
256 if (buf == NULL) {
257 rpc_set_error(rpc, "Failed to allocate buffer of %d bytes for pdu, errno:%d. Closing socket.", pdu_size, errno);
258 return -1;
259 }
260 memcpy(buf, rpc->inbuf, rpc->insize);
261 free(rpc->inbuf);
262 rpc->inbuf = buf;
263 rpc->insize = rpc_get_pdu_size(rpc->inbuf);
264 }
265
266 size = available;
267 if (size > rpc->insize - rpc->inpos) {
268 size = rpc->insize - rpc->inpos;
269 }
270
271 count = recv(rpc->fd, rpc->inbuf + rpc->inpos, size, 0);
272 if (count == -1) {
273 if (errno == EINTR) {
274 return 0;
275 }
276 rpc_set_error(rpc, "Read from socket failed, errno:%d. Closing socket.", errno);
277 return -1;
278 }
279 available -= count;
280 rpc->inpos += count;
281
282 if (rpc->inpos == rpc->insize) {
283 char *buf = rpc->inbuf;
284
285 rpc->inbuf = NULL;
286 rpc->insize = 0;
287 rpc->inpos = 0;
288
289 if (rpc_process_pdu(rpc, buf, pdu_size) != 0) {
290 rpc_set_error(rpc, "Invalid/garbage pdu received from server. Closing socket");
291 return -1;
292 }
293 free(buf);
294 }
295
296 return 0;
297 }
298
299
300
301 int rpc_service(struct rpc_context *rpc, int revents)
302 {
303 assert(rpc->magic == RPC_CONTEXT_MAGIC);
304
305 if (revents & POLLERR) {
306 #ifdef WIN32
307 char err = 0;
308 #else
309 int err = 0;
310 #endif
311 socklen_t err_size = sizeof(err);
312
313 if (getsockopt(rpc->fd, SOL_SOCKET, SO_ERROR,
314 (char *)&err, &err_size) != 0 || err != 0) {
315 if (err == 0) {
316 err = errno;
317 }
318 rpc_set_error(rpc, "rpc_service: socket error "
319 "%s(%d).",
320 strerror(err), err);
321 } else {
322 rpc_set_error(rpc, "rpc_service: POLLERR, "
323 "Unknown socket error.");
324 }
325 if (rpc->connect_cb != NULL) {
326 rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data);
327 }
328 return -1;
329 }
330 if (revents & POLLHUP) {
331 rpc_set_error(rpc, "Socket failed with POLLHUP");
332 if (rpc->connect_cb != NULL) {
333 rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data);
334 }
335 return -1;
336 }
337
338 if (rpc->is_connected == 0 && rpc->fd != -1 && revents&POLLOUT) {
339 int err = 0;
340 socklen_t err_size = sizeof(err);
341
342 if (getsockopt(rpc->fd, SOL_SOCKET, SO_ERROR,
343 (char *)&err, &err_size) != 0 || err != 0) {
344 if (err == 0) {
345 err = errno;
346 }
347 rpc_set_error(rpc, "rpc_service: socket error "
348 "%s(%d) while connecting.",
349 strerror(err), err);
350 if (rpc->connect_cb != NULL) {
351 rpc->connect_cb(rpc, RPC_STATUS_ERROR,
352 NULL, rpc->connect_data);
353 }
354 return -1;
355 }
356
357 rpc->is_connected = 1;
358 if (rpc->connect_cb != NULL) {
359 rpc->connect_cb(rpc, RPC_STATUS_SUCCESS, NULL, rpc->connect_data);
360 }
361 return 0;
362 }
363
364 if (revents & POLLIN) {
365 if (rpc_read_from_socket(rpc) != 0) {
366 rpc_reconnect_requeue(rpc);
367 return 0;
368 }
369 }
370
371 if (revents & POLLOUT && rpc->outqueue != NULL) {
372 if (rpc_write_to_socket(rpc) != 0) {
373 rpc_set_error(rpc, "write to socket failed");
374 return -1;
375 }
376 }
377
378 return 0;
379 }
380
381 void rpc_set_autoreconnect(struct rpc_context *rpc)
382 {
383 assert(rpc->magic == RPC_CONTEXT_MAGIC);
384
385 rpc->auto_reconnect = 1;
386 }
387
388 void rpc_unset_autoreconnect(struct rpc_context *rpc)
389 {
390 assert(rpc->magic == RPC_CONTEXT_MAGIC);
391
392 rpc->auto_reconnect = 0;
393 }
394
395 void rpc_set_tcp_syncnt(struct rpc_context *rpc, int v)
396 {
397 assert(rpc->magic == RPC_CONTEXT_MAGIC);
398
399 rpc->tcp_syncnt = v;
400 }
401
402 #ifndef TCP_SYNCNT
403 #define TCP_SYNCNT 7
404 #endif
405
406 static int rpc_connect_sockaddr_async(struct rpc_context *rpc, struct sockaddr_storage *s)
407 {
408 int socksize;
409
410 assert(rpc->magic == RPC_CONTEXT_MAGIC);
411
412 switch (s->ss_family) {
413 case AF_INET:
414 socksize = sizeof(struct sockaddr_in);
415 rpc->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
416 #ifdef HAVE_NETINET_TCP_H
417 if (rpc->tcp_syncnt != RPC_PARAM_UNDEFINED) {
418 set_tcp_sockopt(rpc->fd, TCP_SYNCNT, rpc->tcp_syncnt);
419 }
420 #endif
421 break;
422 default:
423 rpc_set_error(rpc, "Can not handle AF_FAMILY:%d", s->ss_family);
424 return -1;
425 }
426
427 if (rpc->fd == -1) {
428 rpc_set_error(rpc, "Failed to open socket");
429 return -1;
430 }
431
432 /* Some systems allow you to set capabilities on an executable
433 * to allow the file to be executed with privilege to bind to
434 * privileged system ports, even if the user is not root.
435 *
436 * Opportunistically try to bind the socket to a low numbered
437 * system port in the hope that the user is either root or the
438 * executable has the CAP_NET_BIND_SERVICE.
439 *
440 * As soon as we fail the bind() with EACCES we know we will never
441 * be able to bind to a system port so we terminate the loop.
442 *
443 * On linux, use
444 * sudo setcap 'cap_net_bind_service=+ep' /path/executable
445 * to make the executable able to bind to a system port.
446 *
447 * On Windows, there is no concept of privileged ports. Thus
448 * binding will usually succeed.
449 */
450 {
451 struct sockaddr_in sin;
452 static int portOfs = 0;
453 const int firstPort = 512; /* >= 512 according to Sun docs */
454 const int portCount = IPPORT_RESERVED - firstPort;
455 int startOfs, port, rc;
456
457 if (portOfs == 0) {
458 portOfs = time(NULL) % 400;
459 }
460 startOfs = portOfs;
461 do {
462 rc = -1;
463 port = htons(firstPort + portOfs);
464 portOfs = (portOfs + 1) % portCount;
465
466 /* skip well-known ports */
467 if (!getservbyport(port, "tcp")) {
468 memset(&sin, 0, sizeof(sin));
469 sin.sin_port = port;
470 sin.sin_family = AF_INET;
471 sin.sin_addr.s_addr = 0;
472
473 rc = bind(rpc->fd, (struct sockaddr *)&sin, sizeof(struct sockaddr_in));
474 #if !defined(WIN32)
475 /* we got EACCES, so don't try again */
476 if (rc != 0 && errno == EACCES)
477 break;
478 #endif
479 }
480 } while (rc != 0 && portOfs != startOfs);
481 }
482
483 set_nonblocking(rpc->fd);
484
485 if (connect(rpc->fd, (struct sockaddr *)s, socksize) != 0 && errno != EINPROGRESS) {
486 rpc_set_error(rpc, "connect() to server failed. %s(%d)", strerror(errno), errno);
487 return -1;
488 }
489
490 return 0;
491 }
492
493 int rpc_connect_async(struct rpc_context *rpc, const char *server, int port, rpc_cb cb, void *private_data)
494 {
495 struct sockaddr_in *sin = (struct sockaddr_in *)&rpc->s;
496
497 assert(rpc->magic == RPC_CONTEXT_MAGIC);
498
499 if (rpc->fd != -1) {
500 rpc_set_error(rpc, "Trying to connect while already connected");
501 return -1;
502 }
503
504 if (rpc->is_udp != 0) {
505 rpc_set_error(rpc, "Trying to connect on UDP socket");
506 return -1;
507 }
508
509 rpc->auto_reconnect = 0;
510
511 sin->sin_family = AF_INET;
512 sin->sin_port = htons(port);
513 if (inet_pton(AF_INET, server, &sin->sin_addr) != 1) {
514 rpc_set_error(rpc, "Not a valid server ip address");
515 return -1;
516 }
517
518
519 switch (rpc->s.ss_family) {
520 case AF_INET:
521 #ifdef HAVE_SOCKADDR_LEN
522 sin->sin_len = sizeof(struct sockaddr_in);
523 #endif
524 break;
525 }
526
527 rpc->connect_cb = cb;
528 rpc->connect_data = private_data;
529
530 if (rpc_connect_sockaddr_async(rpc, &rpc->s) != 0) {
531 return -1;
532 }
533
534 return 0;
535 }
536
537 int rpc_disconnect(struct rpc_context *rpc, char *error)
538 {
539 assert(rpc->magic == RPC_CONTEXT_MAGIC);
540
541 rpc_unset_autoreconnect(rpc);
542
543 if (rpc->fd != -1) {
544 close(rpc->fd);
545 }
546 rpc->fd = -1;
547
548 rpc->is_connected = 0;
549
550 rpc_error_all_pdus(rpc, error);
551
552 return 0;
553 }
554
555 static void reconnect_cb(struct rpc_context *rpc, int status, void *data _U_, void *private_data)
556 {
557 assert(rpc->magic == RPC_CONTEXT_MAGIC);
558
559 if (status != RPC_STATUS_SUCCESS) {
560 rpc_error_all_pdus(rpc, "RPC ERROR: Failed to reconnect async");
561 return;
562 }
563
564 rpc->is_connected = 1;
565 rpc->connect_cb = NULL;
566 }
567
568 /* disconnect but do not error all PDUs, just move pdus in-flight back to the outqueue and reconnect */
569 static int rpc_reconnect_requeue(struct rpc_context *rpc)
570 {
571 struct rpc_pdu *pdu;
572
573 assert(rpc->magic == RPC_CONTEXT_MAGIC);
574
575 if (rpc->fd != -1) {
576 close(rpc->fd);
577 }
578 rpc->fd = -1;
579
580 rpc->is_connected = 0;
581
582 /* socket is closed so we will not get any replies to any commands
583 * in flight. Move them all over from the waitpdu queue back to the out queue
584 */
585 for (pdu=rpc->waitpdu; pdu; pdu=pdu->next) {
586 SLIST_REMOVE(&rpc->waitpdu, pdu);
587 SLIST_ADD(&rpc->outqueue, pdu);
588 /* we have to re-send the whole pdu again */
589 pdu->written = 0;
590 }
591
592 if (rpc->auto_reconnect != 0) {
593 rpc->connect_cb = reconnect_cb;
594
595 if (rpc_connect_sockaddr_async(rpc, &rpc->s) != 0) {
596 rpc_error_all_pdus(rpc, "RPC ERROR: Failed to reconnect async");
597 return -1;
598 }
599 }
600
601 return 0;
602 }
603
604
605 int rpc_bind_udp(struct rpc_context *rpc, char *addr, int port)
606 {
607 struct addrinfo *ai = NULL;
608 char service[6];
609
610 assert(rpc->magic == RPC_CONTEXT_MAGIC);
611
612 if (rpc->is_udp == 0) {
613 rpc_set_error(rpc, "Cant not bind UDP. Not UDP context");
614 return -1;
615 }
616
617 sprintf(service, "%d", port);
618 if (getaddrinfo(addr, service, NULL, &ai) != 0) {
619 rpc_set_error(rpc, "Invalid address:%s. "
620 "Can not resolv into IPv4/v6 structure.");
621 return -1;
622 }
623
624 switch(ai->ai_family) {
625 case AF_INET:
626 rpc->fd = socket(ai->ai_family, SOCK_DGRAM, 0);
627 if (rpc->fd == -1) {
628 rpc_set_error(rpc, "Failed to create UDP socket: %s", strerror(errno));
629 freeaddrinfo(ai);
630 return -1;
631 }
632
633 if (bind(rpc->fd, (struct sockaddr *)ai->ai_addr, sizeof(struct sockaddr_in)) != 0) {
634 rpc_set_error(rpc, "Failed to bind to UDP socket: %s",strerror(errno));
635 freeaddrinfo(ai);
636 return -1;
637 }
638 break;
639 default:
640 rpc_set_error(rpc, "Can not handle UPD sockets of family %d yet", ai->ai_family);
641 freeaddrinfo(ai);
642 return -1;
643 }
644
645 freeaddrinfo(ai);
646
647 return 0;
648 }
649
650 int rpc_set_udp_destination(struct rpc_context *rpc, char *addr, int port, int is_broadcast)
651 {
652 struct addrinfo *ai = NULL;
653 char service[6];
654
655 assert(rpc->magic == RPC_CONTEXT_MAGIC);
656
657 if (rpc->is_udp == 0) {
658 rpc_set_error(rpc, "Can not set destination sockaddr. Not UDP context");
659 return -1;
660 }
661
662 sprintf(service, "%d", port);
663 if (getaddrinfo(addr, service, NULL, &ai) != 0) {
664 rpc_set_error(rpc, "Invalid address:%s. "
665 "Can not resolv into IPv4/v6 structure.");
666 return -1;
667 }
668
669 if (rpc->udp_dest) {
670 free(rpc->udp_dest);
671 rpc->udp_dest = NULL;
672 }
673 rpc->udp_dest = malloc(ai->ai_addrlen);
674 if (rpc->udp_dest == NULL) {
675 rpc_set_error(rpc, "Out of memory. Failed to allocate sockaddr structure");
676 freeaddrinfo(ai);
677 return -1;
678 }
679 memcpy(rpc->udp_dest, ai->ai_addr, ai->ai_addrlen);
680 freeaddrinfo(ai);
681
682 rpc->is_broadcast = is_broadcast;
683 setsockopt(rpc->fd, SOL_SOCKET, SO_BROADCAST, (char *)&is_broadcast, sizeof(is_broadcast));
684
685 return 0;
686 }
687
688 struct sockaddr *rpc_get_recv_sockaddr(struct rpc_context *rpc)
689 {
690 assert(rpc->magic == RPC_CONTEXT_MAGIC);
691
692 return (struct sockaddr *)&rpc->udp_src;
693 }
694
695 int rpc_queue_length(struct rpc_context *rpc)
696 {
697 int i=0;
698 struct rpc_pdu *pdu;
699
700 assert(rpc->magic == RPC_CONTEXT_MAGIC);
701
702 for(pdu = rpc->outqueue; pdu; pdu = pdu->next) {
703 i++;
704 }
705 for(pdu = rpc->waitpdu; pdu; pdu = pdu->next) {
706 i++;
707 }
708 return i;
709 }
710
711 void rpc_set_fd(struct rpc_context *rpc, int fd)
712 {
713 assert(rpc->magic == RPC_CONTEXT_MAGIC);
714
715 rpc->fd = fd;
716 }