bump up version: 0.0.1

Too large buffer size for sftp_async_read causes unfinished copy:
sftp_async_read returns 0 althrough data remains.

.github/workflows/build-macos.yml

@@ -36,3 +36,5 @@ jobs:
       # Build your program with the given configuration
       run: cmake --build ${{github.workspace}}/build --config ${{env.BUILD_TYPE}}
 
+    - name: Run
+      run: ${{github.workspace}}/build/mscp -h

.github/workflows/build-ubuntu.yml

@@ -31,3 +31,5 @@ jobs:
       # Build your program with the given configuration
       run: cmake --build ${{github.workspace}}/build --config ${{env.BUILD_TYPE}}
 
+    - name: Run
+      run: ${{github.workspace}}/build/mscp -h

.github/workflows/release.yml

@@ -36,3 +36,4 @@ jobs:
             ${{github.workspace}}/build/mscp_${{env.VERSION}}-ubuntu-20.04-x86_64.deb
             ${{github.workspace}}/build/mscp_${{env.VERSION}}-ubuntu-22.04-x86_64.deb
             ${{github.workspace}}/build/mscp_${{env.VERSION}}-centos-8-x86_64.rpm
+            ${{github.workspace}}/build/mscp_${{env.VERSION}}-rocky-8.6-x86_64.rpm

CMakeLists.txt

@@ -105,12 +105,23 @@ if(BUILD_PKG)
 		COMMAND docker run --rm -v ${CMAKE_BINARY_DIR}:/out mscp-centos:8
 		cp /mscp/build/mscp_${PROJECT_VERSION}-centos-8-${ARCH}.rpm /out/)
 
+	# Rocky 8.6
+	add_custom_target(package-rocky-8.6-in-docker
+		COMMENT "Build mscp in rocky 8.6 docker container"
+		WORKING_DIRECTORY ${CMAKE_SOURCE_DIR}
+		COMMAND docker build -t mscp-rocky:8.6 -f docker/rocky-8.6.Dockerfile .
+		COMMAND docker run --init --rm mscp-rocky:8.6
+			/mscp/scripts/test-in-container.sh
+		COMMAND docker run --rm -v ${CMAKE_BINARY_DIR}:/out mscp-rocky:8.6
+		cp /mscp/build/mscp_${PROJECT_VERSION}-rocky-8.6-${ARCH}.rpm /out/)
+
 	# build on all conatiners
 	add_custom_target(package-all-in-docker
 		COMMENT "Build mscp in all docker containers"
 		DEPENDS package-ubuntu-20.04-in-docker
 		DEPENDS package-ubuntu-22.04-in-docker
-		DEPENDS package-centos-8-in-docker)
+		DEPENDS package-centos-8-in-docker
+		DEPENDS package-rocky-8.6-in-docker)
 
 endif() # BUILD_PKG
 

README.md

@@ -3,24 +3,40 @@
 [![build on ubuntu](https://github.com/upa/mscp/actions/workflows/build-ubuntu.yml/badge.svg)](https://github.com/upa/mscp/actions/workflows/build-ubuntu.yml) [![build on macOS](https://github.com/upa/mscp/actions/workflows/build-macos.yml/badge.svg)](https://github.com/upa/mscp/actions/workflows/build-macos.yml) [![test](https://github.com/upa/mscp/actions/workflows/test.yml/badge.svg)](https://github.com/upa/mscp/actions/workflows/test.yml)
 
 
-`mscp`, a variant of `scp`, copies files over multiple ssh (sftp)
+`mscp`, a variant of `scp`, copies files over multiple ssh (SFTP)
 sessions. Multiple threads in mscp transfer (1) multiple files
 simultaneously and (2) a large file in parallel. It may shorten the
 waiting time for transferring a lot of/large files over networks.
 
-You can use `mscp` like `scp`, e.g., `mscp example.com:srcfile
-/tmp/dstfile`. Remote hosts only need to run `sshd` supporting the
-SFTP subsystem, and you need to be able to ssh to the hosts (as
-usual).
+You can use `mscp` like `scp`, for example, `mscp
+user@example.com:srcfile /tmp/dstfile`. Remote hosts only need to run
+standard `sshd` supporting the SFTP subsystem, and you need to be able
+to ssh to the hosts (as usual). `mscp` does not require anything else.
+
 
 Differences from `scp` are:
 
 - remote glob on remote shell expansion is not supported.
 - remote to remote copy is not supported.
 - `-r` option is not needed.
-- and any other differences I have not noticed and implemented...
+- and any other differences I have not implemented and noticed...
 
-## Build
+
+## Install
+
+- homebrew
+
+```console
+brew install upa/tap/mscp
+```
+
+- Linux
+
+Download a package for your environment from [Releases
+page](https://github.com/upa/mscp/releases).
+
+
+## Build from source
 
 mscp depends on [libssh](https://www.libssh.org/).
 
@@ -59,50 +75,30 @@ make install
 
 - Usage
 
-```shell-session
+```console
 $ mscp
 mscp v0.0.0: copy files over multiple ssh connections
 
 Usage: mscp [vqDCHdh] [-n nr_conns]
             [-s min_chunk_sz] [-S max_chunk_sz]
-            [-b sftp_buf_sz] [-B io_buf_sz]
+            [-b sftp_buf_sz] [-B io_buf_sz] [-a nr_ahead]
             [-l login_name] [-p port] [-i identity_file]
             [-c cipher_spec] source ... target
-
-    -n NR_CONNECTIONS  number of connections (default: half of # of cpu cores)
-    -s MIN_CHUNK_SIZE  min chunk size (default: 64MB)
-    -S MAX_CHUNK_SIZE  max chunk size (default: filesize / nr_conn)
-    -b SFTP_BUF_SIZE   buf size for sftp_read/write (default 131072B)
-    -B IO_BUF_SIZE     buf size for read/write (default 131072B)
-                       Note that this value is derived from
-                       qemu/block/ssh.c. need investigation...
-    -v                 increment verbose output level
-    -q                 disable output
-    -D                 dry run
-
-    -l LOGIN_NAME      login name
-    -p PORT            port number
-    -i IDENTITY        identity file for publickey authentication
-    -c CIPHER          cipher spec, see `ssh -Q cipher`
-    -C                 enable compression on libssh
-    -H                 disable hostkey check
-    -d                 increment ssh debug output level
-    -h                 print this help
 ```
 
 - Example: copy an 8GB file on tmpfs over a 100Gbps link
   - Two Intel Xeon Gold 6130 machines directly connected with Intel E810 100Gbps NICs.
 
-```shell-session
+```console
 $ mscp /tmp/test.img 10.0.0.1:/tmp/
-[===============================================================] 100% 8GB/8GB 3.02GB/s 
+[=====================================================] 100% 8GB/8GB 3.02GB/s 
 ```
 
-- `-v` options increment verbose output level.
+- `-v` option increments verbose output level.
 
-```shell-session
+```console
 $ mscp test 10.0.0.1:
-[===============================================================] 100% 13B/13B 2.41KB/s 
+[=====================================================] 100% 13B/13B 2.41KB/s 
 
 $ mscp -v test 10.0.0.1:
 file test/test.txt (local) -> ./test/test.txt (remote) 9B
@@ -114,7 +110,7 @@ copy start: test/test2/2.txt
 copy done: test/1.txt
 copy done: test/test2/2.txt
 copy done: test/test.txt
-[===============================================================] 100% 13B/13B 2.51KB/s 
+[=====================================================] 100% 13B/13B 2.51KB/s 
 
 $ mscp -vv -n 4 test 10.0.0.1:
 connecting to 10.0.0.1 for checking destinations...
@@ -131,8 +127,46 @@ copy start: test/test2/2.txt
 copy done: test/test.txt
 copy done: test/test2/2.txt
 copy done: test/1.txt
-[===============================================================] 100% 13B/13B 3.27KB/s
+[=====================================================] 100% 13B/13B 3.27KB/s
 ```
 
+- Full usage
+
+```console
+$ mscp -h
+mscp v0.0.0: copy files over multiple ssh connections
+
+Usage: mscp [vqDCHdh] [-n nr_conns]
+            [-s min_chunk_sz] [-S max_chunk_sz]
+            [-b sftp_buf_sz] [-B io_buf_sz] [-a nr_ahead]
+            [-l login_name] [-p port] [-i identity_file]
+            [-c cipher_spec] source ... target
+
+    -n NR_CONNECTIONS  number of connections (default: half of # of cpu cores)
+    -s MIN_CHUNK_SIZE  min chunk size (default: 64MB)
+    -S MAX_CHUNK_SIZE  max chunk size (default: filesize / nr_conn)
+
+    -b SFTP_BUF_SIZE   buf size for sftp_read/write (default 131072B)
+    -B IO_BUF_SIZE     buf size for read/write (default 131072B)
+                       Note that the default value is derived from
+                       qemu/block/ssh.c. need investigation...
+                       -b and -B affect only local to remote copy
+    -a NR_AHEAD        number of inflight SFTP read commands (default 16)
+
+    -v                 increment verbose output level
+    -q                 disable output
+    -D                 dry run
+
+    -l LOGIN_NAME      login name
+    -p PORT            port number
+    -i IDENTITY        identity file for publickey authentication
+    -c CIPHER          cipher spec, see `ssh -Q cipher`
+    -C                 enable compression on libssh
+    -H                 disable hostkey check
+    -d                 increment ssh debug output level
+    -h                 print this help
+```
+
+
 Note: mscp is still under development, and the author is not
-responsible for any accidents on mscp.
+responsible for any accidents due to mscp.

VERSION

@@ -1 +1 @@
-0.0.0
+0.0.1

docker/README.md

@@ -6,19 +6,23 @@ cd ..
 
 docker build -t mscp-ubuntu:20.04 -f docker/ubuntu-20.04.Dockerfile .
 
-docker build -t mscp-ubuntu:22.04 -f docker/Dockerfile-ubuntu-22.04 .
+docker build -t mscp-ubuntu:22.04 -f docker/ubuntu-22.04.Dockerfile .
 
-docker build -t mscp-centos:8 -f docker/Dockerfile-centos-8 .
+docker build -t mscp-centos:8 -f docker/centos-8.Dockerfile .
+
+docker build -t mscp-rocky:8.6 -f docker/rocky-8.6.Dockerfile .
 ```
 
 Test `mscp` in the containers.
 
 ```console
-docker run --init --rm mscp-ubuntu:20.04 /build/mscp/scripts/test-in-container.sh
+docker run --init --rm mscp-ubuntu:20.04 /mscp/scripts/test-in-container.sh
 
-docker run --init --rm mscp-ubuntu:22.04 /build/mscp/scripts/test-in-container.sh
+docker run --init --rm mscp-ubuntu:22.04 /mscp/scripts/test-in-container.sh
 
-docker run --init --rm mscp-centos:8 /build/mscp/scripts/test-in-container.sh
+docker run --init --rm mscp-centos:8 /mscp/scripts/test-in-container.sh
+
+docker run --init --rm mscp-rocky:8.6 /mscp/scripts/test-in-container.sh
 ```
 
 Retrieve deb/rpm packages.
@@ -32,6 +36,9 @@ docker run --rm -v (pwd):/out mscp-ubuntu:22.04 \
 
 docker run --rm -v (pwd):/out mscp-centos:8 \
        cp /mscp/build/mscp_0.0.0-centos-8-x86_64.rpm /out/
+
+docker run --rm -v (pwd):/out mscp-rocky:8.6 \
+       cp /mscp/build/mscp_0.0.0-rocky-8.6-x86_64.rpm /out/
 ```
 
 I don't know whether these are good way.

docker/rocky-8.6.Dockerfile

@@ -0,0 +1,30 @@
+FROM rockylinux:8.6
+
+ARG mscpdir="/mscp"
+
+COPY . ${mscpdir}
+
+# install numpy and pytest, sshd for test, and rpm-build
+RUN set -ex && yum -y update && yum -y install \
+	python3 python3-pip openssh openssh-server openssh-clients rpm-build
+
+RUN python3 -m pip install numpy pytest
+
+
+# preparation for sshd
+RUN  mkdir /var/run/sshd        \
+	&& ssh-keygen -A	\
+        && ssh-keygen -f /root/.ssh/id_rsa -N ""                \
+        && mv /root/.ssh/id_rsa.pub /root/.ssh/authorized_keys
+
+# install build dependency
+RUN ${mscpdir}/scripts/install-build-deps.sh
+
+# build
+RUN cd ${mscpdir}			\
+        && rm -rf build			\
+        && cmake -B build -DBUILD_PKG=1 \
+        && cd ${mscpdir}/build          \
+	&& make				\
+	&& cpack -G RPM CPackConfig.cmake
+

scripts/install-build-deps.sh

@@ -9,8 +9,8 @@ case $ID in
 	ubuntu*)
 		apt-get install -y gcc make cmake libssh-dev
 		;;
-	centos* | rhel*)
-		dnf install -y gcc make cmake libssh-devel rpm-build
+	centos* | rhel* | rocky*)
+		yum install -y gcc make cmake libssh-devel rpm-build
 		;;
 	*)
 		echo "unsupported dependency install: $ID"

scripts/test-in-container.sh

@@ -17,7 +17,7 @@ case $ID in
 		pkg=mscp_${project_version}-${ID}-${VERSION_ID}-${arch}.deb
 		dpkg -i ../build/$pkg
                 ;;
-        centos* | rhel*)
+        centos* | rhel* | rocky*)
 		pkg=mscp_${project_version}-${ID}-${VERSION_ID}-${arch}.rpm
 		rpm -iv ../build/$pkg
                 ;;

src/file.c

@@ -619,39 +619,25 @@ static sftp_file chunk_open_remote(const char *path, int flags, mode_t mode, siz
 	return sf;
 }
 
-static int chunk_copy_internal(struct chunk *c, int fd, sftp_file sf,
-			       size_t sftp_buf_sz, size_t io_buf_sz,
-			       bool reverse, size_t *counter)
+static int chunk_copy_internal_local_to_remote(struct chunk *c, int fd, sftp_file sf,
+					       size_t sftp_buf_sz, size_t io_buf_sz,
+					       size_t *counter)
 {
-	size_t remaind, read_bytes, write_bytes;
+	ssize_t read_bytes, write_bytes, remaind;
 	char buf[io_buf_sz];
 
-	/* if reverse is false, copy fd->sf (local to remote).
-	 * if reverse is true, copy sf->fd (remote to local)
-	 */
-
 	for (remaind = c->len; remaind > 0;) {
 
-		if (!reverse)
-			read_bytes = read(fd, buf, min(remaind, io_buf_sz));
-		else
-			read_bytes = sftp_read2(sf, buf, min(remaind, io_buf_sz),
-						sftp_buf_sz);
-
+		read_bytes = read(fd, buf, min(remaind, io_buf_sz));
 		if (read_bytes < 0) {
-			pr_err("failed to read %s: %s\n", c->f->dst_path,
-			       !reverse ? strerrno() : sftp_get_ssh_error(sf->sftp));
+			pr_err("failed to read %s: %s\n", c->f->src_path, strerrno());
 			return -1;
 		}
 
-		if (!reverse)
-			write_bytes = sftp_write2(sf, buf, read_bytes, sftp_buf_sz);
-		else
-			write_bytes = write(fd, buf, read_bytes);
-
+		write_bytes = sftp_write2(sf, buf, read_bytes, sftp_buf_sz);
 		if (write_bytes < 0) {
-			pr_err("failed to write %s: %s\n", c->f->dst_path,
-			       !reverse ? strerrno() : sftp_get_ssh_error(sf->sftp));
+			pr_err("failed to write to %s: SFTP error code %d\n",
+			       c->f->dst_path, sftp_get_error(sf->sftp));
 			return -1;
 		}
 
@@ -667,6 +653,76 @@ static int chunk_copy_internal(struct chunk *c, int fd, sftp_file sf,
 	return 0;
 }
 
+static int chunk_copy_internal_remote_to_local(struct chunk *c, int fd, sftp_file sf,
+					       int nr_ahead, size_t *counter)
+{
+#define XFER_BUF_SIZE	16384
+
+	ssize_t read_bytes, write_bytes, remaind, thrown;
+	char buf[XFER_BUF_SIZE];
+	int idx;
+	struct {
+		int id;
+		size_t len;
+	} reqs[nr_ahead];
+
+	/* TODO: sftp_buf_sz has no effect on remote to local copy. we
+	 * always use 16384 byte buffer pointed by
+	 * https://api.libssh.org/stable/libssh_tutor_sftp.html. The
+	 * larget read length from sftp_async_read is 65536 byte.
+	 * Read sizes larget than 65536 cause a situation where data
+	 * remainds but sftp_async_read returns 0.
+	 */
+
+	if (c->len == 0)
+		return 0;
+
+	remaind = thrown = c->len;
+
+	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) {
+			pr_err("sftp_async_read_begin failed: %d\n",
+			       sftp_get_error(sf->sftp));
+			return -1;
+		}
+		thrown -= reqs[idx].len;
+	}
+
+	for (idx = 0; remaind > 0;) {
+
+		read_bytes = sftp_async_read(sf, buf, reqs[idx].len, reqs[idx].id);
+		if (read_bytes == SSH_ERROR) {
+			pr_err("sftp_async_read failed: %d\n", sftp_get_error(sf->sftp));
+			return -1;
+		}
+
+		remaind -= read_bytes;
+
+		if (remaind > 0) {
+			reqs[idx].len = min(remaind, sizeof(buf));
+			reqs[idx].id = sftp_async_read_begin(sf, reqs[idx].len);
+		}
+
+		write_bytes = write(fd, buf, read_bytes);
+		if (write_bytes < 0) {
+			pr_err("write to %s failed: %s\n", c->f->dst_path, strerrno());
+			return -1;
+		}
+
+		if (write_bytes < read_bytes) {
+			pr_err("failed to write full bytes to %s\n", c->f->dst_path);
+			return -1;
+		}
+
+		*counter += write_bytes;
+		idx = (idx + 1) % nr_ahead;
+	}
+
+	return 0;
+}
+
 static int chunk_copy_local_to_remote(struct chunk *c, sftp_session sftp,
 				      size_t sftp_buf_sz, size_t io_buf_sz,
 				      size_t *counter)
@@ -692,7 +748,8 @@ static int chunk_copy_local_to_remote(struct chunk *c, sftp_session sftp,
 		goto out;
 	}
 
-	ret = chunk_copy_internal(c, fd, sf, sftp_buf_sz, io_buf_sz, false, counter);
+	ret = chunk_copy_internal_local_to_remote(c, fd, sf, sftp_buf_sz, io_buf_sz,
+						  counter);
 	if (ret < 0)
 		goto out;
 
@@ -713,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,
-				      size_t *counter)
+				      int nr_ahead, size_t *counter)
 {
 	struct file *f = c->f;
 	sftp_file sf = NULL;
@@ -737,7 +793,7 @@ static int chunk_copy_remote_to_local(struct chunk *c, sftp_session sftp,
 		goto out;
 	}
 
-	ret = chunk_copy_internal(c, fd, sf, sftp_buf_sz, io_buf_sz, true, counter);
+	ret = chunk_copy_internal_remote_to_local(c, fd, sf, nr_ahead, counter);
 	if (ret< 0)
 		goto out;
 
@@ -753,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;
@@ -770,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,
-						 sftp_buf_sz, io_buf_sz, counter);
+		ret = chunk_copy_remote_to_local(c, sftp, nr_ahead, 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,
-	       size_t sftp_buf_sz, size_t io_buf_sz, size_t *counter);
+int chunk_copy(struct chunk *c, sftp_session sftp, size_t sftp_buf_sz, size_t io_buf_sz,
+	       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 */
-	lock                    chunk_lock;  /* lock for chunk list */
+	struct list_head        chunk_list;	/* stack of chunks */
+	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 */
 };
@@ -66,7 +68,8 @@ void stop_copy_threads(int sig)
 
 	pr("stopping...\n");
 	for (n = 0; n < nr_threads; n++) {
-		pthread_cancel(threads[n].tid);
+		if (!threads[n].finished)
+			pthread_cancel(threads[n].tid);
 	}
 }
 
@@ -84,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");
@@ -95,10 +98,14 @@ void usage(bool print_help) {
 	printf("    -n NR_CONNECTIONS  number of connections (default: half of # of cpu cores)\n"
 	       "    -s MIN_CHUNK_SIZE  min chunk size (default: 64MB)\n"
 	       "    -S MAX_CHUNK_SIZE  max chunk size (default: filesize / nr_conn)\n"
+	       "\n"
 	       "    -b SFTP_BUF_SIZE   buf size for sftp_read/write (default 131072B)\n"
 	       "    -B IO_BUF_SIZE     buf size for read/write (default 131072B)\n"
-	       "                       Note that this value is derived from\n"
+	       "                       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"
 	       "    -D                 dry run\n"
@@ -175,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);
@@ -232,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;
@@ -323,13 +338,6 @@ int main(int argc, char **argv)
 	if (dryrun)
 		return 0;
 
-	/* register SIGINT to stop thrads */
-	if (signal(SIGINT, stop_copy_threads) == SIG_ERR) {
-		pr_err("cannot set signal: %s\n", strerrno());
-		ret = 1;
-		goto out;
-	}
-
 	/* prepare thread instances */
 	if ((n = list_count(&m.chunk_list)) < nr_threads) {
 		pprint3("we have only %d chunk(s). set nr_conns to %d\n", n, n);
@@ -359,6 +367,13 @@ int main(int argc, char **argv)
 		goto join_out;
 	}
 
+	/* register SIGINT to stop threads */
+	if (signal(SIGINT, stop_copy_threads) == SIG_ERR) {
+		pr_err("cannot set signal: %s\n", strerrno());
+		ret = 1;
+		goto out;
+	}
+
 	/* save start time */
 	gettimeofday(&m.start, NULL);
 
@@ -401,6 +416,7 @@ void mscp_copy_thread_cleanup(void *arg)
 	if (t->sftp)
 		ssh_sftp_close(t->sftp);
 	t->finished = true;
+	__sync_synchronize();
 }
 
 void *mscp_copy_thread(void *arg)
@@ -425,13 +441,17 @@ void *mscp_copy_thread(void *arg)
 		if ((t->ret = chunk_prepare(c, sftp)) < 0)
 			break;
 
-		if ((t->ret = chunk_copy(c, sftp,
-					 m->sftp_buf_sz, m->io_buf_sz, &t->done)) < 0)
+		if ((t->ret = chunk_copy(c, sftp, m->sftp_buf_sz, m->io_buf_sz,
+					 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;
 }
 
@@ -490,11 +510,11 @@ static void print_progress_bar(double percent, char *suffix)
 static void print_progress(struct timeval *start, struct timeval *end,
 			   size_t total, size_t last, size_t done)
 {
-	char *bps_units[] = { "B/s", "KB/s", "MB/s", "GB/s" };
-	char *byte_units[] = { "B", "KB", "MB", "GB", "TB", "PB" };
+	char *bps_units[] = { "B/s ", "KB/s", "MB/s", "GB/s" };
+	char *byte_units[] = { "B ", "KB", "MB", "GB", "TB", "PB" };
 	char suffix[128];
 	int bps_u, byte_tu, byte_du;
-	size_t total_round;
+	size_t total_round, done_round;
 	int percent;
 	double bps;
 
@@ -515,11 +535,14 @@ static void print_progress(struct timeval *start, struct timeval *end,
 		bps /= 1000;
 
 	percent = floor(((double)(done) / (double)total) * 100);
-	for (byte_du = 0; done > 1000 && byte_du < array_size(byte_units) - 1; byte_du++)
-		done /= 1024;
 
-	snprintf(suffix, sizeof(suffix), "%lu%s/%lu%s %.2f%s ",
-		 done, byte_units[byte_du], total_round, byte_units[byte_tu],
+	done_round = done;
+	for (byte_du = 0; done_round > 1000 && byte_du < array_size(byte_units) - 1;
+	     byte_du++)
+		done_round /= 1024;
+
+	snprintf(suffix, sizeof(suffix), "%lu%s/%lu%s %6.1f%s ",
+		 done_round, byte_units[byte_du], total_round, byte_units[byte_tu],
 		 bps, bps_units[bps_u]);
 
 	print_progress_bar(percent, suffix);
@@ -579,10 +602,10 @@ void *mscp_monitor_thread(void *arg)
 		}
 		gettimeofday(&b, NULL);
 
-		usleep(500000);
+		usleep(1000000);
 
 		for (n = 0; n < nr_threads; n++) {
-			done += threads[n].done;;
+			done += threads[n].done;
 			if (!threads[n].finished)
 				all_done = false;
 		}

src/ssh.c

@@ -246,9 +246,9 @@ void ssh_sftp_close(sftp_session sftp)
 }
 
 
-int sftp_write2(sftp_file sf, const void *buf, size_t len, size_t sftp_buf_sz)
+ssize_t sftp_write2(sftp_file sf, const void *buf, size_t len, size_t sftp_buf_sz)
 {
-	int ret, nbytes;
+	ssize_t ret, nbytes;
 
 	for (nbytes = 0; nbytes < len;) {
 		ret = sftp_write(sf, buf + nbytes,
@@ -260,9 +260,9 @@ int sftp_write2(sftp_file sf, const void *buf, size_t len, size_t sftp_buf_sz)
 	return nbytes;
 }
 
-int sftp_read2(sftp_file sf, void *buf, size_t len, size_t sftp_buf_sz)
+ssize_t sftp_read2(sftp_file sf, void *buf, size_t len, size_t sftp_buf_sz)
 {
-	int ret, nbytes;
+	ssize_t ret, nbytes;
 
 	for (nbytes = 0; nbytes < len;) {
 		ret = sftp_read(sf, buf + nbytes,

src/ssh.h

@@ -28,7 +28,7 @@ void ssh_sftp_close(sftp_session sftp);
 #define sftp_get_ssh_error(sftp) ssh_get_error(sftp_ssh(sftp))
 
 /* wrapping multiple sftp_read|write */
-int sftp_write2(sftp_file sf, const void *buf, size_t len, size_t sftp_buf_sz);
-int sftp_read2(sftp_file sf, void *buf, size_t len, size_t sftp_buf_sz);
+ssize_t sftp_write2(sftp_file sf, const void *buf, size_t len, size_t sftp_buf_sz);
+ssize_t sftp_read2(sftp_file sf, void *buf, size_t len, size_t sftp_buf_sz);
 
 #endif /* _SSH_H_ */