Perform multiple overlapping read/write requests in file transfer. Mostly

done by Tobias Ringstrom <tori@ringstrom.mine.nu>; ok markus@

5 files changed, 213 insertions, 103 deletions

diff --git a/usr.bin/ssh/sftp-client.c b/usr.bin/ssh/sftp-client.c
index 72e61db3bec..07792e96005 100644
--- a/usr.bin/ssh/sftp-client.c
+++ b/usr.bin/ssh/sftp-client.c
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2001 Damien Miller.  All rights reserved.
+ * Copyright (c) 2001-2002 Damien Miller.  All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions
@@ -24,12 +24,13 @@
 
 /* XXX: memleaks */
 /* XXX: signed vs unsigned */
-/* XXX: redesign to allow concurrent overlapped operations */
 /* XXX: we use fatal too much, error may be more appropriate in places */
 /* XXX: copy between two remote sites */
 
 #include "includes.h"
-RCSID("$OpenBSD: sftp-client.c,v 1.20 2002/02/05 00:00:46 djm Exp $");
+RCSID("$OpenBSD: sftp-client.c,v 1.21 2002/02/12 12:32:27 djm Exp $");
+
+#include <sys/queue.h>
 
 #include "buffer.h"
 #include "bufaux.h"
@@ -42,6 +43,9 @@ RCSID("$OpenBSD: sftp-client.c,v 1.20 2002/02/05 00:00:46 djm Exp $");
 #include "sftp-common.h"
 #include "sftp-client.h"
 
+/* Minimum amount of data to read at at time */
+#define MIN_READ_SIZE	512
+
 /* Message ID */
 static u_int msg_id = 1;
 
@@ -664,16 +668,44 @@ do_readlink(int fd_in, int fd_out, char *path)
 	return(filename);
 }
 
+static void
+send_read_request(int fd_out, u_int id, u_int64_t offset, u_int len,
+    char *handle, u_int handle_len)
+{
+	Buffer msg;
+	
+	buffer_init(&msg);
+	buffer_clear(&msg);
+	buffer_put_char(&msg, SSH2_FXP_READ);
+	buffer_put_int(&msg, id);
+	buffer_put_string(&msg, handle, handle_len);
+	buffer_put_int64(&msg, offset);
+	buffer_put_int(&msg, len);
+	send_msg(fd_out, &msg);
+	buffer_free(&msg);
+}	
+
 int
 do_download(int fd_in, int fd_out, char *remote_path, char *local_path,
-    int pflag, size_t buflen)
+    int pflag, size_t buflen, int num_requests)
 {
-	int local_fd, status;
-	u_int expected_id, handle_len, mode, type, id;
-	u_int64_t offset;
-	char *handle;
-	Buffer msg;
 	Attrib junk, *a;
+	Buffer msg;
+	char *handle;
+	int local_fd, status, num_req, max_req, write_error;
+	int read_error, write_errno;
+	u_int64_t offset, size;
+	u_int handle_len, mode, type, id;
+	struct request {
+		u_int id;
+		u_int len;
+		u_int64_t offset;
+		TAILQ_ENTRY(request) tq; 
+	};
+	TAILQ_HEAD(reqhead, request) requests;
+	struct request *req;
+
+	TAILQ_INIT(&requests);
 
 	a = do_stat(fd_in, fd_out, remote_path, 0);
 	if (a == NULL)
@@ -691,6 +723,11 @@ do_download(int fd_in, int fd_out, char *remote_path, char *local_path,
 		return(-1);
 	}
 
+	if (a->flags & SSH2_FILEXFER_ATTR_SIZE)
+		size = a->size;
+	else
+		size = 0;
+
 	local_fd = open(local_path, O_WRONLY | O_CREAT | O_TRUNC, mode);
 	if (local_fd == -1) {
 		error("Couldn't open local file \"%s\" for writing: %s",
@@ -719,84 +756,135 @@ do_download(int fd_in, int fd_out, char *remote_path, char *local_path,
 	}
 
 	/* Read from remote and write to local */
-	offset = 0;
-	for (;;) {
-		u_int len;
+	write_error = read_error = write_errno = num_req = offset = 0;
+	max_req = 1;
+	while (num_req > 0 || max_req > 0) {
 		char *data;
+		u_int len;
 
-		id = expected_id = msg_id++;
-
-		buffer_clear(&msg);
-		buffer_put_char(&msg, SSH2_FXP_READ);
-		buffer_put_int(&msg, id);
-		buffer_put_string(&msg, handle, handle_len);
-		buffer_put_int64(&msg, offset);
-		buffer_put_int(&msg, buflen);
-		send_msg(fd_out, &msg);
-		debug3("Sent message SSH2_FXP_READ I:%d O:%llu S:%u",
-		    id, (unsigned long long)offset, buflen);
+		/* Send some more requests */
+		while (num_req < max_req) {
+			debug3("Request range %llu -> %llu (%d/%d)", 
+			    offset, offset + buflen - 1, num_req, max_req);
+			req = xmalloc(sizeof(*req));
+			req->id = msg_id++;
+			req->len = buflen;
+			req->offset = offset;
+			offset += buflen;
+			num_req++;
+			TAILQ_INSERT_TAIL(&requests, req, tq);
+			send_read_request(fd_out, req->id, req->offset, 
+			    req->len, handle, handle_len);
+		}
 
 		buffer_clear(&msg);
-
 		get_msg(fd_in, &msg);
 		type = buffer_get_char(&msg);
 		id = buffer_get_int(&msg);
-		debug3("Received reply T:%d I:%d", type, id);
-		if (id != expected_id)
-			fatal("ID mismatch (%d != %d)", id, expected_id);
-		if (type == SSH2_FXP_STATUS) {
+		debug3("Received reply T:%d I:%d R:%d", type, id, max_req);
+
+		/* Find the request in our queue */
+		for(req = TAILQ_FIRST(&requests);
+		    req != NULL && req->id != id;
+		    req = TAILQ_NEXT(req, tq))
+			;
+		if (req == NULL)
+			fatal("Unexpected reply %u", id);
+
+		switch (type) {
+		case SSH2_FXP_STATUS:
 			status = buffer_get_int(&msg);
+			if (status != SSH2_FX_EOF)
+				read_error = 1;
+			max_req = 0;
+			TAILQ_REMOVE(&requests, req, tq);
+			xfree(req);
+			num_req--;
+			break;
+		case SSH2_FXP_DATA:
+			data = buffer_get_string(&msg, &len);
+			debug3("Received data %llu -> %llu", req->offset, 
+			    req->offset + len - 1);
+			if (len > req->len)
+				fatal("Received more data than asked for "
+				      "%d > %d", len, req->len);
+			if ((lseek(local_fd, req->offset, SEEK_SET) == -1 ||
+			     atomicio(write, local_fd, data, len) != len) &&
+			    !write_error) {
+				write_errno = errno;
+				write_error = 1;
+				max_req = 0;
+			}
+			xfree(data);
 
-			if (status == SSH2_FX_EOF)
-				break;
-			else {
-				error("Couldn't read from remote "
-				    "file \"%s\" : %s", remote_path,
-				    fx2txt(status));
-				do_close(fd_in, fd_out, handle, handle_len);
-				goto done;
+			if (len == req->len) {
+				TAILQ_REMOVE(&requests, req, tq);
+				xfree(req);
+				num_req--;
+			} else {
+				/* Resend the request for the missing data */
+				debug3("Short data block, re-requesting "
+				    "%llu -> %llu (%2d)", req->offset + len, 
+					req->offset + req->len - 1, num_req);
+				req->id = msg_id++;
+				req->len -= len;
+				req->offset += len;
+				send_read_request(fd_out, req->id, 
+				    req->offset, req->len, handle, 
+				    handle_len);
+				/* Reduce the request size */
+				if (len < buflen)
+					buflen = MAX(MIN_READ_SIZE, len);
 			}
-		} else if (type != SSH2_FXP_DATA) {
+			if (max_req > 0) { /* max_req = 0 iff EOF received */
+				if (size > 0 && offset > size) {
+					/* Only one request at a time
+					 * after the expected EOF */
+					debug3("Finish at %llu (%2d)",
+					    offset, num_req);
+					max_req = 1;
+				}
+				else if (max_req < num_requests + 1) {
+					++max_req;
+				}
+			}
+			break;
+		default:
 			fatal("Expected SSH2_FXP_DATA(%d) packet, got %d",
 			    SSH2_FXP_DATA, type);
 		}
+	}
 
-		data = buffer_get_string(&msg, &len);
-		if (len > buflen)
-			fatal("Received more data than asked for %d > %d",
-			    len, buflen);
+	/* Sanity check */
+	if (TAILQ_FIRST(&requests) != NULL)
+		fatal("Transfer complete, but requests still in queue");
 
-		debug3("In read loop, got %d offset %llu", len,
-		    (unsigned long long)offset);
-		if (atomicio(write, local_fd, data, len) != len) {
-			error("Couldn't write to \"%s\": %s", local_path,
-			    strerror(errno));
-			do_close(fd_in, fd_out, handle, handle_len);
-			status = -1;
-			xfree(data);
-			goto done;
+	if (read_error) {
+		error("Couldn't read from remote file \"%s\" : %s", 
+		    remote_path, fx2txt(status));
+		do_close(fd_in, fd_out, handle, handle_len);
+	} else if (write_error) {
+		error("Couldn't write to \"%s\": %s", local_path,
+		    strerror(write_errno));
+		status = -1;
+		do_close(fd_in, fd_out, handle, handle_len);
+	} else {
+		status = do_close(fd_in, fd_out, handle, handle_len);
+
+		/* Override umask and utimes if asked */
+		if (pflag && fchmod(local_fd, mode) == -1)
+			error("Couldn't set mode on \"%s\": %s", local_path,
+			      strerror(errno));
+		if (pflag && (a->flags & SSH2_FILEXFER_ATTR_ACMODTIME)) {
+			struct timeval tv[2];
+			tv[0].tv_sec = a->atime;
+			tv[1].tv_sec = a->mtime;
+			tv[0].tv_usec = tv[1].tv_usec = 0;
+			if (utimes(local_path, tv) == -1)
+				error("Can't set times on \"%s\": %s",
+				      local_path, strerror(errno));
 		}
-
-		offset += len;
-		xfree(data);
 	}
-	status = do_close(fd_in, fd_out, handle, handle_len);
-
-	/* Override umask and utimes if asked */
-	if (pflag && fchmod(local_fd, mode) == -1)
-		error("Couldn't set mode on \"%s\": %s", local_path,
-		    strerror(errno));
-	if (pflag && (a->flags & SSH2_FILEXFER_ATTR_ACMODTIME)) {
-		struct timeval tv[2];
-		tv[0].tv_sec = a->atime;
-		tv[1].tv_sec = a->mtime;
-		tv[0].tv_usec = tv[1].tv_usec = 0;
-		if (utimes(local_path, tv) == -1)
-			error("Can't set times on \"%s\": %s", local_path,
-			    strerror(errno));
-	}
-
-done:
 	close(local_fd);
 	buffer_free(&msg);
 	xfree(handle);
@@ -805,7 +893,7 @@ done:
 
 int
 do_upload(int fd_in, int fd_out, char *local_path, char *remote_path,
-    int pflag, size_t buflen)
+    int pflag, size_t buflen, int num_requests)
 {
 	int local_fd, status;
 	u_int handle_len, id;
@@ -814,6 +902,8 @@ do_upload(int fd_in, int fd_out, char *local_path, char *remote_path,
 	Buffer msg;
 	struct stat sb;
 	Attrib a;
+	u_int32_t startid;
+	u_int32_t ackid;
 
 	if ((local_fd = open(local_path, O_RDONLY, 0)) == -1) {
 		error("Couldn't open local file \"%s\" for reading: %s",
@@ -855,6 +945,7 @@ do_upload(int fd_in, int fd_out, char *local_path, char *remote_path,
 		return(-1);
 	}
 
+	startid = ackid = id + 1;
 	data = xmalloc(buflen);
 
 	/* Read from local and write to remote */
@@ -873,29 +964,34 @@ do_upload(int fd_in, int fd_out, char *local_path, char *remote_path,
 		if (len == -1)
 			fatal("Couldn't read from \"%s\": %s", local_path,
 			    strerror(errno));
-		if (len == 0)
+
+		if (len != 0) {
+			buffer_clear(&msg);
+			buffer_put_char(&msg, SSH2_FXP_WRITE);
+			buffer_put_int(&msg, ++id);
+			buffer_put_string(&msg, handle, handle_len);
+			buffer_put_int64(&msg, offset);
+			buffer_put_string(&msg, data, len);
+			send_msg(fd_out, &msg);
+			debug3("Sent message SSH2_FXP_WRITE I:%d O:%llu S:%u",
+			       id, (u_int64_t)offset, len);
+		} else if ( id < ackid )
 			break;
 
-		buffer_clear(&msg);
-		buffer_put_char(&msg, SSH2_FXP_WRITE);
-		buffer_put_int(&msg, ++id);
-		buffer_put_string(&msg, handle, handle_len);
-		buffer_put_int64(&msg, offset);
-		buffer_put_string(&msg, data, len);
-		send_msg(fd_out, &msg);
-		debug3("Sent message SSH2_FXP_WRITE I:%d O:%llu S:%u",
-		    id, (unsigned long long)offset, len);
-
-		status = get_status(fd_in, id);
-		if (status != SSH2_FX_OK) {
-			error("Couldn't write to remote file \"%s\": %s",
-			    remote_path, fx2txt(status));
-			do_close(fd_in, fd_out, handle, handle_len);
-			close(local_fd);
-			goto done;
+		if (id == startid || len == 0 ||
+		    id - ackid >= num_requests) {
+			status = get_status(fd_in, ackid);
+			if (status != SSH2_FX_OK) {
+				error("Couldn't write to remote file \"%s\": %s",
+				      remote_path, fx2txt(status));
+				do_close(fd_in, fd_out, handle, handle_len);
+				close(local_fd);
+				goto done;
+			}
+			debug3("In write loop, got %d offset %llu", len,
+			       (u_int64_t)offset);
+			++ackid;
 		}
-		debug3("In write loop, got %d offset %llu", len,
-		    (unsigned long long)offset);
 
 		offset += len;
 	}
@@ -920,4 +1016,3 @@ done:
 	buffer_free(&msg);
 	return status;
 }
-
diff --git a/usr.bin/ssh/sftp-client.h b/usr.bin/ssh/sftp-client.h
index 20350701c8d..477c8ed3bd2 100644
--- a/usr.bin/ssh/sftp-client.h
+++ b/usr.bin/ssh/sftp-client.h
@@ -1,4 +1,4 @@
-/* $OpenBSD: sftp-client.h,v 1.7 2002/02/05 00:00:46 djm Exp $ */
+/* $OpenBSD: sftp-client.h,v 1.8 2002/02/12 12:32:27 djm Exp $ */
 
 /*
  * Copyright (c) 2001-2002 Damien Miller.  All rights reserved.
@@ -94,10 +94,10 @@ char *do_readlink(int, int, char *);
  * Download 'remote_path' to 'local_path'. Preserve permissions and times
  * if 'pflag' is set
  */
-int do_download(int, int, char *, char *, int, size_t);
+int do_download(int, int, char *, char *, int, size_t, int);
 
 /*
  * Upload 'local_path' to 'remote_path'. Preserve permissions and times
  * if 'pflag' is set
  */
-int do_upload(int, int, char *, char *, int , size_t);
+int do_upload(int, int, char *, char *, int , size_t, int);
diff --git a/usr.bin/ssh/sftp-int.c b/usr.bin/ssh/sftp-int.c
index c84f0a93735..f2a1846d8ac 100644
--- a/usr.bin/ssh/sftp-int.c
+++ b/usr.bin/ssh/sftp-int.c
@@ -26,7 +26,7 @@
 /* XXX: recursive operations */
 
 #include "includes.h"
-RCSID("$OpenBSD: sftp-int.c,v 1.42 2002/02/05 00:00:46 djm Exp $");
+RCSID("$OpenBSD: sftp-int.c,v 1.43 2002/02/12 12:32:27 djm Exp $");
 
 #include <glob.h>
 
@@ -47,6 +47,9 @@ extern FILE *infile;
 /* Size of buffer used when copying files */
 extern size_t copy_buffer_len;
 
+/* Number of concurrent outstanding requests */
+extern int num_requests;
+
 /* Version of server we are speaking to */
 int version;
 
@@ -387,7 +390,7 @@ process_get(int in, int out, char *src, char *dst, char *pwd, int pflag)
 		}
 		printf("Fetching %s to %s\n", g.gl_pathv[0], abs_dst);
 		err = do_download(in, out, g.gl_pathv[0], abs_dst, pflag,
-		    copy_buffer_len);
+		    copy_buffer_len, num_requests);
 		goto out;
 	}
 
@@ -412,7 +415,7 @@ process_get(int in, int out, char *src, char *dst, char *pwd, int pflag)
 
 		printf("Fetching %s to %s\n", g.gl_pathv[i], abs_dst);
 		if (do_download(in, out, g.gl_pathv[i], abs_dst, pflag,
-		    copy_buffer_len) == -1)
+		    copy_buffer_len, num_requests) == -1)
 			err = -1;
 		xfree(abs_dst);
 		abs_dst = NULL;
@@ -471,7 +474,7 @@ process_put(int in, int out, char *src, char *dst, char *pwd, int pflag)
 		}
 		printf("Uploading %s to %s\n", g.gl_pathv[0], abs_dst);
 		err = do_upload(in, out, g.gl_pathv[0], abs_dst, pflag,
-		    copy_buffer_len);
+		    copy_buffer_len, num_requests);
 		goto out;
 	}
 
@@ -496,7 +499,7 @@ process_put(int in, int out, char *src, char *dst, char *pwd, int pflag)
 
 		printf("Uploading %s to %s\n", g.gl_pathv[i], abs_dst);
 		if (do_upload(in, out, g.gl_pathv[i], abs_dst, pflag, 
-		    copy_buffer_len) == -1)
+		    copy_buffer_len, num_requests) == -1)
 			err = -1;
 	}
 
diff --git a/usr.bin/ssh/sftp.1 b/usr.bin/ssh/sftp.1
index 059b46d1f05..bfdbce1e917 100644
--- a/usr.bin/ssh/sftp.1
+++ b/usr.bin/ssh/sftp.1
@@ -1,4 +1,4 @@
-.\" $OpenBSD: sftp.1,v 1.29 2002/02/06 14:22:42 markus Exp $
+.\" $OpenBSD: sftp.1,v 1.30 2002/02/12 12:32:27 djm Exp $
 .\"
 .\" Copyright (c) 2001 Damien Miller.  All rights reserved.
 .\"
@@ -37,6 +37,7 @@
 .Op Fl B Ar buffer_size
 .Op Fl F Ar ssh_config
 .Op Fl P Ar sftp_server path
+.Op Fl R Ar num_requests
 .Op Fl S Ar program
 .Ar host
 .Nm sftp
@@ -118,6 +119,10 @@ Connect directly to a local
 (rather than via
 .Nm ssh )
 This option may be useful in debugging the client and server.
+.It Fl R Ar num_requests
+Specify how many requests may be outstanding at any one time. Increasing 
+this may slightly improve file transfer speed but will increase memory
+usage. The default is 16 outstanding requests.
 .It Fl S Ar program
 Name of the
 .Ar program
diff --git a/usr.bin/ssh/sftp.c b/usr.bin/ssh/sftp.c
index 16dc983949e..0a7689c4e79 100644
--- a/usr.bin/ssh/sftp.c
+++ b/usr.bin/ssh/sftp.c
@@ -24,7 +24,7 @@
 
 #include "includes.h"
 
-RCSID("$OpenBSD: sftp.c,v 1.25 2002/02/06 14:27:23 mpech Exp $");
+RCSID("$OpenBSD: sftp.c,v 1.26 2002/02/12 12:32:27 djm Exp $");
 
 /* XXX: short-form remote directory listings (like 'ls -C') */
 
@@ -41,6 +41,7 @@ RCSID("$OpenBSD: sftp.c,v 1.25 2002/02/06 14:27:23 mpech Exp $");
 
 FILE* infile;
 size_t copy_buffer_len = 32768;
+size_t num_requests = 16;
 
 static void
 connect_to_server(char *path, char **args, int *in, int *out, pid_t *sshpid)
@@ -118,7 +119,7 @@ main(int argc, char **argv)
 	ll = SYSLOG_LEVEL_INFO;
 	infile = stdin;		/* Read from STDIN unless changed by -b */
 
-	while ((ch = getopt(argc, argv, "1hvCo:s:S:b:B:F:P:")) != -1) {
+	while ((ch = getopt(argc, argv, "1hvCo:s:S:b:B:F:P:R:")) != -1) {
 		switch (ch) {
 		case 'C':
 			addargs(&args, "-C");
@@ -161,6 +162,12 @@ main(int argc, char **argv)
 			if (copy_buffer_len == 0 || *cp != '\0')
 				fatal("Invalid buffer size \"%s\"", optarg);
 			break;
+		case 'R':
+			num_requests = strtol(optarg, &cp, 10);
+			if (num_requests == 0 || *cp != '\0')
+				fatal("Invalid number of requests \"%s\"", 
+				    optarg);
+			break;
 		case 'h':
 		default:
 			usage();