X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=lib%2Fsocket.c;h=3f7f89f90c9a368dc5ff6c48c0ab87f790d4e096;hb=18c94b4633c6157c3e75890c15ea51b73e7bf876;hp=c518603c5ef7896378bbd72e67ace1dda42dcfa4;hpb=aec45c6274bffc92fc1595a95d043e8aae292451;p=deb_libnfs.git

diff --git a/lib/socket.c b/lib/socket.c
index c518603..3f7f89f 100644
--- a/lib/socket.c
+++ b/lib/socket.c
@@ -95,6 +95,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)
 {
@@ -174,10 +182,14 @@ static int rpc_write_to_socket(struct rpc_context *rpc)
 
 		pdu->written += count;
 		if (pdu->written == total) {
+			unsigned int hash;
+
 			rpc->outqueue.head = pdu->next;
 			if (pdu->next == NULL)
 				rpc->outqueue.tail = NULL;
-			rpc_enqueue(&rpc->waitpdu, pdu);
+
+			hash = rpc_hash_xid(pdu->xid);
+			rpc_enqueue(&rpc->waitpdu[hash], pdu);
 		}
 	}
 	return 0;
@@ -494,7 +506,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;
 				}
@@ -510,6 +522,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);
@@ -547,6 +560,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
@@ -554,6 +568,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
@@ -607,6 +622,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);
 
@@ -620,12 +636,16 @@ 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.head; pdu; pdu=pdu->next) {
-		rpc_return_to_queue(&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);
 	}
-	rpc_reset_queue(&rpc->waitpdu);
 
 	if (rpc->auto_reconnect != 0) {
 		rpc->connect_cb  = reconnect_cb;
@@ -734,14 +754,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.head; pdu; pdu = pdu->next) {
 		i++;
 	}
-	for(pdu = rpc->waitpdu.head; 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;
 }