Add support for rpc timeouts to the async interface

Signed-off-by: Ronnie Sahlberg <ronniesahlberg@gmail.com>
sahlberg · May 9, 2017 · b5c5d59 · b5c5d59 · earlchew · May 9, 2017
1 parent 4abadb7
commit b5c5d59
Show file tree

Hide file tree

Showing 5 changed files with 63 additions and 2 deletions.
diff --git a/include/libnfs-private.h b/include/libnfs-private.h
@@ -173,6 +173,8 @@ struct rpc_pdu {
 
 #define PDU_DISCARD_AFTER_SENDING 0x00000001
         uint32_t flags;
+
+	time_t timeout;
 };
 
 void rpc_reset_queue(struct rpc_queue *q);

diff --git a/include/nfsc/libnfs-raw.h b/include/nfsc/libnfs-raw.h
@@ -167,6 +167,7 @@ void rpc_set_fd(struct rpc_context *rpc, int fd);
 #define RPC_STATUS_SUCCESS	   	0
 #define RPC_STATUS_ERROR		1
 #define RPC_STATUS_CANCEL		2
+#define RPC_STATUS_TIMEOUT		3
 
 /*
  * Async connection to the tcp port at server:port.

diff --git a/include/nfsc/libnfs.h b/include/nfsc/libnfs.h
@@ -1525,6 +1525,8 @@ void free_nfs_srvr_list(struct nfs_server_list *srv);
  * Function returns nothing.
  *
  * int milliseconds : timeout to be applied in milliseconds (-1 no timeout)
+ *                    timeouts must currently be set in whole seconds,
+ *                    i.e. units of 1000
  */
 EXTERN void nfs_set_timeout(struct nfs_context *nfs, int milliseconds);
 

diff --git a/lib/pdu.c b/lib/pdu.c
@@ -222,6 +222,12 @@ int rpc_queue_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu)
 
 	assert(rpc->magic == RPC_CONTEXT_MAGIC);
 
+	if (rpc->timeout > 0) {
+		pdu->timeout = time(NULL) + rpc->timeout / 1000;
+	} else {
+		pdu->timeout = 0;
+	}
+
 	size = zdr_getpos(&pdu->zdr);
 
 	/* for udp we dont queue, we just send it straight away */

diff --git a/lib/socket.c b/lib/socket.c
@@ -344,10 +344,59 @@ maybe_call_connect_cb(struct rpc_context *rpc, int status)
 	tmp_cb(rpc, status, rpc->error_string, rpc->connect_data);
 }
 
+static void
+rpc_timeout_scan(struct rpc_context *rpc)
+{
+	struct rpc_pdu *pdu;
+	struct rpc_pdu *next_pdu;
+	time_t t = time(NULL);
+	unsigned int i;
+
+	for (pdu = rpc->outqueue.head; pdu; pdu = next_pdu) {
+		next_pdu = pdu->next;
+
+		if (pdu->timeout == 0) {
+			/* no timeout for this pdu */
+			continue;
+		}
+		if (t < pdu->timeout) {
+			/* not expired yet */
+			continue;
+		}
+		LIBNFS_LIST_REMOVE(&rpc->outqueue.head, pdu);
+		rpc_set_error(rpc, "command timed out");
+		pdu->cb(rpc, RPC_STATUS_TIMEOUT,
+			NULL, pdu->private_data);
+		rpc_free_pdu(rpc, pdu);
+	}
+	for (i = 0; i < HASHES; i++) {
+		struct rpc_queue *q = &rpc->waitpdu[i];
+
+		for (pdu = q->head; pdu; pdu = next_pdu) {
+			next_pdu = pdu->next;
+
+			if (pdu->timeout == 0) {
+				/* no timeout for this pdu */
+				continue;
+			}
+			if (t < pdu->timeout) {
+				/* not expired yet */
+				continue;
+			}
+			LIBNFS_LIST_REMOVE(&q->head, pdu);
+			rpc_set_error(rpc, "command timed out");
+			pdu->cb(rpc, RPC_STATUS_TIMEOUT,
+				NULL, pdu->private_data);
+			rpc_free_pdu(rpc, pdu);
+		}
+	}
+}
+
 int rpc_service(struct rpc_context *rpc, int revents)
 {
 	assert(rpc->magic == RPC_CONTEXT_MAGIC);
-	if (revents & (POLLERR|POLLHUP)) {
+
+	if (revents & (POLLERR | POLLHUP)) {
 		if (revents & POLLERR) {
 #ifdef WIN32
 			char err = 0;
@@ -380,7 +429,7 @@ int rpc_service(struct rpc_context *rpc, int revents)
 
 	}
 
-	if (rpc->is_connected == 0 && rpc->fd != -1 && revents&POLLOUT) {
+	if (rpc->is_connected == 0 && rpc->fd != -1 && (revents & POLLOUT)) {
 		int err = 0;
 		socklen_t err_size = sizeof(err);
 
@@ -422,6 +471,7 @@ int rpc_service(struct rpc_context *rpc, int revents)
 		}
 	}
 
+	rpc_timeout_scan(rpc);
 	return 0;
 }