fix implicit decl warnings

[deb_libnfs.git] / lib / socket.c

diff --git a/lib/socket.c b/lib/socket.c
index d025e070e017091e1f796178c139e8ff44c7cf48..f8db957e0bb600510fcba27afc06dd04b2145c1b 100644 (file)
--- a/lib/socket.c
+++ b/lib/socket.c
@@ -68,6 +68,7 @@
 #include <fcntl.h>
 #include <string.h>
 #include <errno.h>
+#include <time.h>
 #include <sys/types.h>
 #include "libnfs-zdr.h"
 #include "libnfs.h"
@@ -95,6 +96,14 @@ static void set_nonblocking(int fd)
 #endif //FIXME
 }
 
+static void set_nolinger(int fd)
+{
+       struct linger lng;
+       lng.l_onoff = 1;
+       lng.l_linger = 0;
+       setsockopt(fd, SOL_SOCKET, SO_LINGER, &lng, sizeof(lng));
+}
+
 #ifdef HAVE_NETINET_TCP_H
 int set_tcp_sockopt(int sockfd, int optname, int value)
 {
@@ -122,6 +131,11 @@ int rpc_get_fd(struct rpc_context *rpc)
        return rpc->fd;
 }
 
+static int rpc_has_queue(struct rpc_queue *q)
+{
+       return q->head != NULL;
+}
+
 int rpc_which_events(struct rpc_context *rpc)
 {
        int events;
@@ -135,7 +149,7 @@ int rpc_which_events(struct rpc_context *rpc)
                return POLLIN;
        }
 
-       if (rpc->outqueue) {
+       if (rpc_has_queue(&rpc->outqueue)) {
                events |= POLLOUT;
        }
        return events;
@@ -144,6 +158,7 @@ int rpc_which_events(struct rpc_context *rpc)
 static int rpc_write_to_socket(struct rpc_context *rpc)
 {
        int32_t count;
+       struct rpc_pdu *pdu;
 
        assert(rpc->magic == RPC_CONTEXT_MAGIC);
 
@@ -152,12 +167,12 @@ static int rpc_write_to_socket(struct rpc_context *rpc)
                return -1;
        }
 
-       while (rpc->outqueue != NULL) {
+       while ((pdu = rpc->outqueue.head) != NULL) {
                int64_t total;
 
-               total = rpc->outqueue->outdata.size;
+               total = pdu->outdata.size;
 
-               count = send(rpc->fd, rpc->outqueue->outdata.data + rpc->outqueue->written, total - rpc->outqueue->written, 0);
+               count = send(rpc->fd, pdu->outdata.data + pdu->written, total - pdu->written, 0);
                if (count == -1) {
                        if (errno == EAGAIN || errno == EWOULDBLOCK) {
                                return 0;
@@ -166,12 +181,16 @@ static int rpc_write_to_socket(struct rpc_context *rpc)
                        return -1;
                }
 
-               rpc->outqueue->written += count;
-               if (rpc->outqueue->written == total) {
-                       struct rpc_pdu *pdu = rpc->outqueue;
+               pdu->written += count;
+               if (pdu->written == total) {
+                       unsigned int hash;
 
-                       SLIST_REMOVE(&rpc->outqueue, pdu);
-                       SLIST_ADD_END(&rpc->waitpdu, pdu);
+                       rpc->outqueue.head = pdu->next;
+                       if (pdu->next == NULL)
+                               rpc->outqueue.tail = NULL;
+
+                       hash = rpc_hash_xid(pdu->xid);
+                       rpc_enqueue(&rpc->waitpdu[hash], pdu);
                }
        }
        return 0;
@@ -368,7 +387,7 @@ int rpc_service(struct rpc_context *rpc, int revents)
                }
        }
 
-       if (revents & POLLOUT && rpc->outqueue != NULL) {
+       if (revents & POLLOUT && rpc_has_queue(&rpc->outqueue)) {
                if (rpc_write_to_socket(rpc) != 0) {
                        rpc_set_error(rpc, "write to socket failed");
                        return -1;
@@ -488,7 +507,7 @@ static int rpc_connect_sockaddr_async(struct rpc_context *rpc, struct sockaddr_s
                                        ((struct sockaddr_in6 *)&ss)->sin6_port = port;
                                        ((struct sockaddr_in6 *)&ss)->sin6_family      = AF_INET6;
 #ifdef HAVE_SOCKADDR_LEN
-                                       ((struct sockaddr_in6 *)&ss)->sin6_len = sizeof(struct sockaddr_in);
+                                       ((struct sockaddr_in6 *)&ss)->sin6_len = sizeof(struct sockaddr_in6);
 #endif
                                        break;
                                }
@@ -504,6 +523,7 @@ static int rpc_connect_sockaddr_async(struct rpc_context *rpc, struct sockaddr_s
        }
 
        set_nonblocking(rpc->fd);
+       set_nolinger(rpc->fd);
 
        if (connect(rpc->fd, (struct sockaddr *)s, socksize) != 0 && errno != EINPROGRESS) {
                rpc_set_error(rpc, "connect() to server failed. %s(%d)", strerror(errno), errno);
@@ -541,6 +561,7 @@ int rpc_connect_async(struct rpc_context *rpc, const char *server, int port, rpc
        case AF_INET:
                ((struct sockaddr_in *)&rpc->s)->sin_family = ai->ai_family;
                ((struct sockaddr_in *)&rpc->s)->sin_port   = htons(port);
+               ((struct sockaddr_in *)&rpc->s)->sin_addr   = ((struct sockaddr_in *)(ai->ai_addr))->sin_addr;
 #ifdef HAVE_SOCKADDR_LEN
                ((struct sockaddr_in *)&rpc->s)->sin_len = sizeof(struct sockaddr_in);
 #endif
@@ -548,6 +569,7 @@ int rpc_connect_async(struct rpc_context *rpc, const char *server, int port, rpc
        case AF_INET6:
                ((struct sockaddr_in6 *)&rpc->s)->sin6_family = ai->ai_family;
                ((struct sockaddr_in6 *)&rpc->s)->sin6_port   = htons(port);
+               ((struct sockaddr_in6 *)&rpc->s)->sin6_addr   = ((struct sockaddr_in6 *)(ai->ai_addr))->sin6_addr;
 #ifdef HAVE_SOCKADDR_LEN
                ((struct sockaddr_in6 *)&rpc->s)->sin6_len = sizeof(struct sockaddr_in6);
 #endif
@@ -601,6 +623,7 @@ static void reconnect_cb(struct rpc_context *rpc, int status, void *data _U_, vo
 static int rpc_reconnect_requeue(struct rpc_context *rpc)
 {
        struct rpc_pdu *pdu;
+       unsigned int i;
 
        assert(rpc->magic == RPC_CONTEXT_MAGIC);
 
@@ -614,11 +637,15 @@ static int rpc_reconnect_requeue(struct rpc_context *rpc)
        /* socket is closed so we will not get any replies to any commands
         * in flight. Move them all over from the waitpdu queue back to the out queue
         */
-       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;
+       for (i = 0; i < HASHES; i++) {
+               struct rpc_queue *q = &rpc->waitpdu[i];
+
+               for (pdu=q->head; pdu; pdu=pdu->next) {
+                       rpc_return_to_queue(&rpc->outqueue, pdu);
+                       /* we have to re-send the whole pdu again */
+                       pdu->written = 0;
+               }
+               rpc_reset_queue(q);
        }
 
        if (rpc->auto_reconnect != 0) {
@@ -728,14 +755,19 @@ int rpc_queue_length(struct rpc_context *rpc)
 {
        int i=0;
        struct rpc_pdu *pdu;
+       unsigned int n;
 
        assert(rpc->magic == RPC_CONTEXT_MAGIC);
 
-       for(pdu = rpc->outqueue; pdu; pdu = pdu->next) {
+       for(pdu = rpc->outqueue.head; pdu; pdu = pdu->next) {
                i++;
        }
-       for(pdu = rpc->waitpdu; pdu; pdu = pdu->next) {
-               i++;
+
+       for (n = 0; n < HASHES; n++) {
+               struct rpc_queue *q = &rpc->waitpdu[n];
+
+               for(pdu = q->head; pdu; pdu = pdu->next)
+                       i++;
        }
        return i;
 }