X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=lib%2Fsocket.c;h=038f6ca33201643a2b008067c6d50ffb407393d0;hb=b990de23c6cb1c172487f21574b3725e44461734;hp=760da2f352a2983480485eb903966ef6ded46a3a;hpb=4b0e8df9a9c5cf23c93de12f2c59d33dda9f3a9d;p=deb_libnfs.git diff --git a/lib/socket.c b/lib/socket.c index 760da2f..038f6ca 100644 --- a/lib/socket.c +++ b/lib/socket.c @@ -25,21 +25,6 @@ #include #endif/*WIN32*/ -#if defined(WIN32) -#include -#include -#include -#define ssize_t SSIZE_T -#define MSG_DONTWAIT 0 -#else -#include -#include -#include -#include -#include -#include -#endif - #ifdef HAVE_CONFIG_H #include "config.h" #endif @@ -62,7 +47,14 @@ #include "libnfs-private.h" #include "slist.h" -static int rpc_disconnect_requeue(struct rpc_context *rpc); +#ifdef WIN32 +//has to be included after stdlib!! +#include "win32_errnowrapper.h" +#endif + + +static int rpc_reconnect_requeue(struct rpc_context *rpc); +static int rpc_connect_sockaddr_async(struct rpc_context *rpc, struct sockaddr_storage *s); static void set_nonblocking(int fd) { @@ -267,7 +259,11 @@ static int rpc_read_from_socket(struct rpc_context *rpc) int rpc_service(struct rpc_context *rpc, int revents) { if (revents & POLLERR) { +#ifdef WIN32 char err = 0; +#else + int err = 0; +#endif socklen_t err_size = sizeof(err); if (getsockopt(rpc->fd, SOL_SOCKET, SO_ERROR, @@ -282,12 +278,16 @@ int rpc_service(struct rpc_context *rpc, int revents) rpc_set_error(rpc, "rpc_service: POLLERR, " "Unknown socket error."); } - rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data); + if (rpc->connect_cb != NULL) { + rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data); + } return -1; } if (revents & POLLHUP) { rpc_set_error(rpc, "Socket failed with POLLHUP"); - rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data); + if (rpc->connect_cb != NULL) { + rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data); + } return -1; } @@ -303,19 +303,23 @@ int rpc_service(struct rpc_context *rpc, int revents) rpc_set_error(rpc, "rpc_service: socket error " "%s(%d) while connecting.", strerror(err), err); - rpc->connect_cb(rpc, RPC_STATUS_ERROR, + if (rpc->connect_cb != NULL) { + rpc->connect_cb(rpc, RPC_STATUS_ERROR, NULL, rpc->connect_data); + } return -1; } rpc->is_connected = 1; - rpc->connect_cb(rpc, RPC_STATUS_SUCCESS, NULL, rpc->connect_data); + if (rpc->connect_cb != NULL) { + rpc->connect_cb(rpc, RPC_STATUS_SUCCESS, NULL, rpc->connect_data); + } return 0; } if (revents & POLLIN) { if (rpc_read_from_socket(rpc) != 0) { - rpc_disconnect_requeue(rpc); + rpc_reconnect_requeue(rpc); return 0; } } @@ -330,13 +334,94 @@ int rpc_service(struct rpc_context *rpc, int revents) return 0; } +void rpc_set_autoreconnect(struct rpc_context *rpc) +{ + rpc->auto_reconnect = 1; +} -int rpc_connect_async(struct rpc_context *rpc, const char *server, int port, rpc_cb cb, void *private_data) +void rpc_unset_autoreconnect(struct rpc_context *rpc) +{ + rpc->auto_reconnect = 0; +} + +static int rpc_connect_sockaddr_async(struct rpc_context *rpc, struct sockaddr_storage *s) { - struct sockaddr_storage s; - struct sockaddr_in *sin = (struct sockaddr_in *)&s; int socksize; + switch (s->ss_family) { + case AF_INET: + socksize = sizeof(struct sockaddr_in); + rpc->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + break; + default: + rpc_set_error(rpc, "Can not handle AF_FAMILY:%d", s->ss_family); + return -1; + } + + if (rpc->fd == -1) { + rpc_set_error(rpc, "Failed to open socket"); + return -1; + } + + +#if !defined(WIN32) + /* Some systems allow you to set capabilities on an executable + * to allow the file to be executed with privilege to bind to + * privileged system ports, even if the user is not root. + * + * Opportunistically try to bind the socket to a low numbered + * system port in the hope that the user is either root or the + * executable has the CAP_NET_BIND_SERVICE. + * + * As soon as we fail the bind() with EACCES we know we will never + * be able to bind to a system port so we terminate the loop. + * + * On linux, use + * sudo setcap 'cap_net_bind_service=+ep' /path/executable + * to make the executable able to bind to a system port. + */ + if (1) { + int port; + int one = 1; + + setsockopt(rpc->fd, SOL_SOCKET, SO_REUSEADDR, (char *)&one, sizeof(one)); + + for (port = 200; port < 500; port++) { + struct sockaddr_in sin; + + memset(&sin, 0, sizeof(sin)); + sin.sin_port = htons(port); + sin.sin_family = AF_INET; + sin.sin_addr.s_addr = 0; + + if (bind(rpc->fd, (struct sockaddr *)&sin, sizeof(struct sockaddr_in)) != 0 && errno != EACCES) { + /* we didnt get EACCES, so try again */ + continue; + } + break; + } + } +#endif + + set_nonblocking(rpc->fd); + +#if defined(WIN32) + if (connect(rpc->fd, (struct sockaddr *)s, socksize) == 0 && errno != EINPROGRESS ) +#else + if (connect(rpc->fd, (struct sockaddr *)s, socksize) != 0 && errno != EINPROGRESS) +#endif + { + rpc_set_error(rpc, "connect() to server failed"); + return -1; + } + + return 0; +} + +int rpc_connect_async(struct rpc_context *rpc, const char *server, int port, rpc_cb cb, void *private_data) +{ + struct sockaddr_in *sin = (struct sockaddr_in *)&rpc->s; + if (rpc->fd != -1) { rpc_set_error(rpc, "Trying to connect while already connected"); return -1; @@ -347,6 +432,8 @@ int rpc_connect_async(struct rpc_context *rpc, const char *server, int port, rpc return -1; } + rpc->auto_reconnect = 0; + sin->sin_family = AF_INET; sin->sin_port = htons(port); if (inet_pton(AF_INET, server, &sin->sin_addr) != 1) { @@ -354,40 +441,29 @@ int rpc_connect_async(struct rpc_context *rpc, const char *server, int port, rpc return -1; } - switch (s.ss_family) { + + switch (rpc->s.ss_family) { case AF_INET: - socksize = sizeof(struct sockaddr_in); #ifdef HAVE_SOCKADDR_LEN - sin->sin_len = socksize; + sin->sin_len = sizeof(struct sockaddr_in); #endif - rpc->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); break; } - if (rpc->fd == -1) { - rpc_set_error(rpc, "Failed to open socket"); - return -1; - } - rpc->connect_cb = cb; rpc->connect_data = private_data; - set_nonblocking(rpc->fd); -#if defined(WIN32) - if (connect(rpc->fd, (struct sockaddr *)&s, socksize) == 0 && GetLastError() != WSAEINPROGRESS ) -#else - if (connect(rpc->fd, (struct sockaddr *)&s, socksize) != 0 && errno != EINPROGRESS) -#endif - { - rpc_set_error(rpc, "connect() to server failed"); + if (rpc_connect_sockaddr_async(rpc, &rpc->s) != 0) { return -1; - } + } return 0; } int rpc_disconnect(struct rpc_context *rpc, char *error) { + rpc_unset_autoreconnect(rpc); + if (rpc->fd != -1) { #if defined(WIN32) closesocket(rpc->fd); @@ -404,8 +480,19 @@ int rpc_disconnect(struct rpc_context *rpc, char *error) return 0; } -/* disconnect but do not error all PDUs, just move pdus in-flight back to the outqueue */ -static int rpc_disconnect_requeue(struct rpc_context *rpc) +static void reconnect_cb(struct rpc_context *rpc, int status, void *data _U_, void *private_data) +{ + if (status != RPC_STATUS_SUCCESS) { + rpc_error_all_pdus(rpc, "RPC ERROR: Failed to reconnect async"); + return; + } + + rpc->is_connected = 1; + rpc->connect_cb = NULL; +} + +/* disconnect but do not error all PDUs, just move pdus in-flight back to the outqueue and reconnect */ +static int rpc_reconnect_requeue(struct rpc_context *rpc) { struct rpc_pdu *pdu; @@ -426,6 +513,17 @@ static int rpc_disconnect_requeue(struct rpc_context *rpc) for (pdu=rpc->waitpdu; pdu; pdu=pdu->next) { SLIST_REMOVE(&rpc->waitpdu, pdu); SLIST_ADD(&rpc->outqueue, pdu); + /* we have to re-send the whole pdu again */ + pdu->written = 0; + } + + if (rpc->auto_reconnect != 0) { + rpc->connect_cb = reconnect_cb; + + if (rpc_connect_sockaddr_async(rpc, &rpc->s) != 0) { + rpc_error_all_pdus(rpc, "RPC ERROR: Failed to reconnect async"); + return -1; + } } return 0; @@ -514,3 +612,17 @@ struct sockaddr *rpc_get_recv_sockaddr(struct rpc_context *rpc) { return (struct sockaddr *)&rpc->udp_src; } + +int rpc_queue_length(struct rpc_context *rpc) +{ + int i=0; + struct rpc_pdu *pdu; + + for(pdu = rpc->outqueue; pdu; pdu = pdu->next) { + i++; + } + for(pdu = rpc->waitpdu; pdu; pdu = pdu->next) { + i++; + } + return i; +}