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