add -a nr_ahead option

src/file.c

@@ -654,9 +654,8 @@ static int chunk_copy_internal_local_to_remote(struct chunk *c, int fd, sftp_fil
 }
 
 static int chunk_copy_internal_remote_to_local(struct chunk *c, int fd, sftp_file sf,
-					       size_t *counter)
+					       int nr_ahead, size_t *counter)
 {
-#define AHEAD		8
 #define XFER_BUF_SIZE	16384
 
 	ssize_t read_bytes, write_bytes, remaind, thrown;
@@ -665,7 +664,7 @@ static int chunk_copy_internal_remote_to_local(struct chunk *c, int fd, sftp_fil
 	struct {
 		int id;
 		size_t len;
-	} reqs[AHEAD];
+	} reqs[nr_ahead];
 
 	/* TODO: sftp_buf_sz has no effect on remote to local copy. we
 	 * always use 16384 byte buffer pointed by
@@ -680,7 +679,7 @@ static int chunk_copy_internal_remote_to_local(struct chunk *c, int fd, sftp_fil
 
 	remaind = thrown = c->len;
 
-	for (idx = 0; idx < AHEAD && thrown > 0; idx++) {
+	for (idx = 0; idx < nr_ahead && thrown > 0; idx++) {
 		reqs[idx].len = min(thrown, sizeof(buf));
 		reqs[idx].id = sftp_async_read_begin(sf, reqs[idx].len);
 		if (reqs[idx].id < 0) {
@@ -718,7 +717,7 @@ static int chunk_copy_internal_remote_to_local(struct chunk *c, int fd, sftp_fil
 		}
 
 		*counter += write_bytes;
-		idx = (idx + 1) % AHEAD;
+		idx = (idx + 1) % nr_ahead;
 	}
 
 	return 0;
@@ -771,8 +770,7 @@ out:
 }
 
 static int chunk_copy_remote_to_local(struct chunk *c, sftp_session sftp,
-				      size_t sftp_buf_sz, size_t io_buf_sz,
+				      int nr_ahead, size_t *counter)
-				      size_t *counter)
 {
 	struct file *f = c->f;
 	sftp_file sf = NULL;
@@ -795,7 +793,7 @@ static int chunk_copy_remote_to_local(struct chunk *c, sftp_session sftp,
 		goto out;
 	}
 
-	ret = chunk_copy_internal_remote_to_local(c, fd, sf, counter);
+	ret = chunk_copy_internal_remote_to_local(c, fd, sf, nr_ahead, counter);
 	if (ret< 0)
 		goto out;
 
@@ -811,7 +809,7 @@ out:
 
 
 int chunk_copy(struct chunk *c, sftp_session sftp, size_t sftp_buf_sz, size_t io_buf_sz,
-	       size_t *counter)
+	       int nr_ahead, size_t *counter)
 {
 	struct file *f = c->f;
 	int ret = 0;
@@ -828,8 +826,7 @@ int chunk_copy(struct chunk *c, sftp_session sftp, size_t sftp_buf_sz, size_t io
 		ret = chunk_copy_local_to_remote(c, sftp,
 						 sftp_buf_sz, io_buf_sz, counter);
 	else
-		ret = chunk_copy_remote_to_local(c, sftp,
+		ret = chunk_copy_remote_to_local(c, sftp, nr_ahead, counter);
-						 sftp_buf_sz, io_buf_sz, counter);
 
 	if (ret < 0)
 		return ret;

src/file.h

@@ -71,8 +71,8 @@ int chunk_fill(struct list_head *file_list, struct list_head *chunk_list,
 
 struct chunk *chunk_acquire(struct list_head *chunk_list);
 int chunk_prepare(struct chunk *c, sftp_session sftp);
-int chunk_copy(struct chunk *c, sftp_session sftp,
+int chunk_copy(struct chunk *c, sftp_session sftp, size_t sftp_buf_sz, size_t io_buf_sz,
-	       size_t sftp_buf_sz, size_t io_buf_sz, size_t *counter);
+	       int nr_ahead, size_t *counter);
 
 #ifdef DEBUG
 void file_dump(struct list_head *file_list);

src/main.c

@@ -26,6 +26,7 @@
 #define DEFAULT_SFTP_BUF_SZ     131072          /* derived from qemu/block/ssh.c */
 #define DEFAULT_IO_BUF_SZ       DEFAULT_SFTP_BUF_SZ
 /* XXX: need to investigate max buf size for sftp_read/sftp_write */
+#define DEFAULT_NR_AHEAD	16
 
 struct mscp {
 	char                    *host;  /* remote host (and username) */
@@ -33,12 +34,13 @@ struct mscp {
 	sftp_session            ctrl;   /* control sftp session */
 
 	struct list_head        file_list;
-	struct list_head        chunk_list;     /* stack of chunks */
+	struct list_head        chunk_list;	/* stack of chunks */
-	lock                    chunk_lock;  /* lock for chunk list */
+	lock                    chunk_lock;	/* lock for chunk list */
 
 	char    *target;
 
 	int     sftp_buf_sz, io_buf_sz;
+	int	nr_ahead;	/* # of ahead read command for remote to local copy */
 
 	struct timeval start;   /* timestamp of starting copy */
 };
@@ -85,7 +87,7 @@ void usage(bool print_help) {
 	       "\n"
 	       "Usage: mscp [vqDCHdh] [-n nr_conns]\n"
 	       "            [-s min_chunk_sz] [-S max_chunk_sz]\n"
-	       "            [-b sftp_buf_sz] [-B io_buf_sz]\n"
+	       "            [-b sftp_buf_sz] [-B io_buf_sz] [-a nr_ahead]\n"
 	       "            [-l login_name] [-p port] [-i identity_file]\n"
 	       "            [-c cipher_spec] source ... target\n"
 	       "\n");
@@ -102,6 +104,7 @@ void usage(bool print_help) {
 	       "                       Note that the default value is derived from\n"
 	       "                       qemu/block/ssh.c. need investigation...\n"
 	       "                       -b and -B affect only local to remote copy\n"
+	       "    -a NR_AHEAD        number of inflight SFTP read commands (default 16)\n"
 	       "\n"
 	       "    -v                 increment verbose output level\n"
 	       "    -q                 disable output\n"
@@ -179,11 +182,12 @@ int main(int argc, char **argv)
 	lock_init(&m.chunk_lock);
 	m.sftp_buf_sz = DEFAULT_SFTP_BUF_SZ;
 	m.io_buf_sz = DEFAULT_IO_BUF_SZ;
+	m.nr_ahead = DEFAULT_NR_AHEAD;
 
 	nr_threads = (int)(nr_cpus() / 2);
 	nr_threads = nr_threads == 0 ? 1 : nr_threads;
 
-	while ((ch = getopt(argc, argv, "n:s:S:b:B:vqDl:p:i:c:CHdh")) != -1) {
+	while ((ch = getopt(argc, argv, "n:s:S:b:B:a:vqDl:p:i:c:CHdh")) != -1) {
 		switch (ch) {
 		case 'n':
 			nr_threads = atoi(optarg);
@@ -236,6 +240,13 @@ int main(int argc, char **argv)
 				return -1;
 			}
 			break;
+		case 'a':
+			m.nr_ahead = atoi(optarg);
+			if (m.nr_ahead < 1) {
+				pr_err("invalid number of ahead: %s\n", optarg);
+				return -1;
+			}
+			break;
 		case 'v':
 			verbose++;
 			break;
@@ -429,13 +440,17 @@ void *mscp_copy_thread(void *arg)
 		if ((t->ret = chunk_prepare(c, sftp)) < 0)
 			break;
 
-		if ((t->ret = chunk_copy(c, sftp,
+		if ((t->ret = chunk_copy(c, sftp, m->sftp_buf_sz, m->io_buf_sz,
-					 m->sftp_buf_sz, m->io_buf_sz, &t->done)) < 0)
+					 m->nr_ahead, &t->done)) < 0)
 			break;
 	}
 
 	pthread_cleanup_pop(1);
 
+	if (t->ret < 0)
+		pr_err("copy failed: chunk %s 0x%010lx-0x%010lx\n",
+		       c->f->src_path, c->off, c->off + c->len);
+
 	return NULL;
 }