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