mirror of
https://github.com/upa/mscp.git
synced 2026-02-14 09:14:47 +08:00
configurable sftp_buf_sz.
default value 131072 is derived from qemu/block/ssh.c. we need more investigaion.
This commit is contained in:
37
src/file.c
37
src/file.c
@@ -588,19 +588,23 @@ static sftp_file chunk_open_remote(const char *path, int flags, mode_t mode, siz
|
||||
}
|
||||
|
||||
static int chunk_copy_internal(struct chunk *c, int fd, sftp_file sf,
|
||||
size_t buf_sz, bool reverse, size_t *counter)
|
||||
size_t sftp_buf_sz, size_t io_buf_sz,
|
||||
bool reverse, size_t *counter)
|
||||
{
|
||||
int remaind, read_bytes, write_bytes;
|
||||
char buf[buf_sz];
|
||||
char buf[io_buf_sz];
|
||||
|
||||
/* if reverse is false, copy fd->sf. if true, copy sf->fd */
|
||||
/* 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, buf_sz));
|
||||
read_bytes = read(fd, buf, min(remaind, io_buf_sz));
|
||||
else
|
||||
read_bytes = sftp_read2(sf, buf, min(remaind, buf_sz));
|
||||
read_bytes = sftp_read2(sf, buf, min(remaind, io_buf_sz),
|
||||
sftp_buf_sz);
|
||||
|
||||
if (read_bytes < 0) {
|
||||
pr_err("failed to read %s: %s\n", c->f->dst_path,
|
||||
@@ -610,7 +614,7 @@ static int chunk_copy_internal(struct chunk *c, int fd, sftp_file sf,
|
||||
}
|
||||
|
||||
if (!reverse)
|
||||
write_bytes = sftp_write2(sf, buf, read_bytes);
|
||||
write_bytes = sftp_write2(sf, buf, read_bytes, sftp_buf_sz);
|
||||
else
|
||||
write_bytes = write(fd, buf, read_bytes);
|
||||
|
||||
@@ -633,7 +637,8 @@ static int chunk_copy_internal(struct chunk *c, int fd, sftp_file sf,
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int chunk_copy_local_to_remote(struct chunk *c, sftp_session sftp, size_t buf_sz,
|
||||
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)
|
||||
{
|
||||
struct file *f = c->f;
|
||||
@@ -657,7 +662,8 @@ static int chunk_copy_local_to_remote(struct chunk *c, sftp_session sftp, size_t
|
||||
goto out;
|
||||
}
|
||||
|
||||
if ((ret = chunk_copy_internal(c, fd, sf, buf_sz, false, counter)) < 0)
|
||||
ret = chunk_copy_internal(c, fd, sf, sftp_buf_sz, io_buf_sz, false, counter);
|
||||
if (ret< 0)
|
||||
goto out;
|
||||
|
||||
if ((mode = chunk_get_mode(f->path, NULL)) < 0) {
|
||||
@@ -676,7 +682,8 @@ out:
|
||||
return ret;
|
||||
}
|
||||
|
||||
static int chunk_copy_remote_to_local(struct chunk *c, sftp_session sftp, size_t buf_sz,
|
||||
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)
|
||||
{
|
||||
struct file *f = c->f;
|
||||
@@ -700,7 +707,8 @@ static int chunk_copy_remote_to_local(struct chunk *c, sftp_session sftp, size_t
|
||||
goto out;
|
||||
}
|
||||
|
||||
if ((ret = chunk_copy_internal(c, fd, sf, buf_sz, true, counter)) < 0)
|
||||
ret = chunk_copy_internal(c, fd, sf, sftp_buf_sz, io_buf_sz, true, counter);
|
||||
if (ret< 0)
|
||||
goto out;
|
||||
|
||||
out:
|
||||
@@ -714,7 +722,8 @@ out:
|
||||
|
||||
|
||||
|
||||
int chunk_copy(struct chunk *c, sftp_session sftp, size_t buf_sz, size_t *counter)
|
||||
int chunk_copy(struct chunk *c, sftp_session sftp, size_t sftp_buf_sz, size_t io_buf_sz,
|
||||
size_t *counter)
|
||||
{
|
||||
struct file *f = c->f;
|
||||
int ret = 0;
|
||||
@@ -724,9 +733,11 @@ int chunk_copy(struct chunk *c, sftp_session sftp, size_t buf_sz, size_t *counte
|
||||
f->dst_path, f->dst_remote ? "(remote)" : "(local)", c->off);
|
||||
|
||||
if (f->dst_remote)
|
||||
ret = chunk_copy_local_to_remote(c, sftp, buf_sz, counter);
|
||||
ret = chunk_copy_local_to_remote(c, sftp,
|
||||
sftp_buf_sz, io_buf_sz, counter);
|
||||
else
|
||||
ret = chunk_copy_remote_to_local(c, sftp, buf_sz, counter);
|
||||
ret = chunk_copy_remote_to_local(c, sftp,
|
||||
sftp_buf_sz, io_buf_sz, counter);
|
||||
|
||||
if (ret < 0)
|
||||
return ret;
|
||||
|
||||
@@ -70,7 +70,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 buf_sz, size_t *counter);
|
||||
int chunk_copy(struct chunk *c, sftp_session sftp,
|
||||
size_t sftp_buf_sz, size_t io_buf_sz, size_t *counter);
|
||||
|
||||
#ifdef DEBUG
|
||||
void file_dump(struct list_head *file_list);
|
||||
|
||||
34
src/main.c
34
src/main.c
@@ -19,22 +19,22 @@ int verbose = 0; /* util.h */
|
||||
|
||||
|
||||
#define DEFAULT_MIN_CHUNK_SZ (64 << 20) /* 64MB */
|
||||
#define DEFAULT_BUF_SZ 32768 /* CHANNEL_MAX_PACKET in libssh */
|
||||
/* XXX: passing over CHANNEL_MAX_PACKET bytes to sftp_write stalls */
|
||||
#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 */
|
||||
|
||||
struct sscp {
|
||||
char *host; /* remote host (and username) */
|
||||
struct ssh_opts *opts; /* ssh parameters */
|
||||
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 */
|
||||
|
||||
char *target;
|
||||
|
||||
int buf_sz;
|
||||
int sftp_buf_sz, io_buf_sz;
|
||||
};
|
||||
|
||||
struct sscp_thread {
|
||||
@@ -69,7 +69,7 @@ void usage(bool print_help) {
|
||||
printf("sscp: super scp, copy files over multiple ssh connections\n"
|
||||
"\n"
|
||||
"Usage: sscp [Cvh] [-n max_conns] [-s min_chunk_sz] [-S max_chunk_sz]\n"
|
||||
" [-b buf_sz]\n"
|
||||
" [-b sftp_buf_sz] [-B io_buf_sz]\n"
|
||||
" [-l login_name] [-p port] [-i identity_file]\n"
|
||||
" [-c cipher_spec] source ... target_directory\n"
|
||||
"\n");
|
||||
@@ -80,9 +80,10 @@ void usage(bool print_help) {
|
||||
printf(" -n NR_CONNECTIONS max number of connections (default: # 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"
|
||||
" -b BUFFER_SIZE buffer size for read/write (default 32768B)\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"
|
||||
" CHANNEL_MAX_PACKET in libssh.\n"
|
||||
" qemu/block/ssh.c. need investigation...\n"
|
||||
"\n"
|
||||
" -l LOGIN_NAME login name\n"
|
||||
" -p PORT port number\n"
|
||||
@@ -157,11 +158,12 @@ int main(int argc, char **argv)
|
||||
INIT_LIST_HEAD(&sscp.file_list);
|
||||
INIT_LIST_HEAD(&sscp.chunk_list);
|
||||
lock_init(&sscp.chunk_lock);
|
||||
sscp.buf_sz = DEFAULT_BUF_SZ;
|
||||
sscp.sftp_buf_sz = DEFAULT_SFTP_BUF_SZ;
|
||||
sscp.io_buf_sz = DEFAULT_IO_BUF_SZ;
|
||||
|
||||
nr_threads = nr_cpus();
|
||||
|
||||
while ((ch = getopt(argc, argv, "n:s:S:b:l:p:i:c:Cvh")) != -1) {
|
||||
while ((ch = getopt(argc, argv, "n:s:S:b:B:l:p:i:c:Cvh")) != -1) {
|
||||
switch (ch) {
|
||||
case 'n':
|
||||
nr_threads = atoi(optarg);
|
||||
@@ -201,8 +203,15 @@ int main(int argc, char **argv)
|
||||
}
|
||||
break;
|
||||
case 'b':
|
||||
sscp.buf_sz = atoi(optarg);
|
||||
if (sscp.buf_sz < 1) {
|
||||
sscp.sftp_buf_sz = atoi(optarg);
|
||||
if (sscp.sftp_buf_sz < 1) {
|
||||
pr_err("invalid buffer size: %s\n", optarg);
|
||||
return -1;
|
||||
}
|
||||
break;
|
||||
case 'B':
|
||||
sscp.io_buf_sz = atoi(optarg);
|
||||
if (sscp.io_buf_sz < 1) {
|
||||
pr_err("invalid buffer size: %s\n", optarg);
|
||||
return -1;
|
||||
}
|
||||
@@ -373,7 +382,8 @@ void *sscp_copy_thread(void *arg)
|
||||
if (chunk_prepare(c, sftp) < 0)
|
||||
break;
|
||||
|
||||
if (chunk_copy(c, sftp, sscp->buf_sz, &t->done) < 0)
|
||||
if (chunk_copy(c, sftp,
|
||||
sscp->sftp_buf_sz, sscp->io_buf_sz, &t->done) < 0)
|
||||
break;
|
||||
}
|
||||
|
||||
|
||||
20
src/ssh.c
20
src/ssh.c
@@ -245,21 +245,13 @@ void ssh_sftp_close(sftp_session sftp)
|
||||
}
|
||||
|
||||
|
||||
/* from libssh/src/channels.c:
|
||||
*
|
||||
* All implementations MUST be able to process packets with an
|
||||
* uncompressed payload length of 32768 bytes or less and a total packet
|
||||
* size of 35000 bytes or less.
|
||||
*/
|
||||
#define CHANNEL_MAX_PACKET 32768
|
||||
|
||||
int sftp_write2(sftp_file sf, const void *buf, size_t len)
|
||||
int sftp_write2(sftp_file sf, const void *buf, size_t len, size_t sftp_buf_sz)
|
||||
{
|
||||
int n, ret, nbytes;
|
||||
int ret, nbytes;
|
||||
|
||||
for (nbytes = 0; nbytes < len;) {
|
||||
ret = sftp_write(sf, buf + nbytes,
|
||||
min(len - nbytes, CHANNEL_MAX_PACKET));
|
||||
min(len - nbytes, sftp_buf_sz));
|
||||
if (ret < 0)
|
||||
return ret;
|
||||
nbytes += ret;
|
||||
@@ -267,13 +259,13 @@ int sftp_write2(sftp_file sf, const void *buf, size_t len)
|
||||
return nbytes;
|
||||
}
|
||||
|
||||
int sftp_read2(sftp_file sf, void *buf, size_t len)
|
||||
int sftp_read2(sftp_file sf, void *buf, size_t len, size_t sftp_buf_sz)
|
||||
{
|
||||
int n, ret, nbytes;
|
||||
int ret, nbytes;
|
||||
|
||||
for (nbytes = 0; nbytes < len;) {
|
||||
ret = sftp_read(sf, buf + nbytes,
|
||||
min(len - nbytes, CHANNEL_MAX_PACKET));
|
||||
min(len - nbytes, sftp_buf_sz));
|
||||
if (ret < 0)
|
||||
return ret;
|
||||
nbytes += ret;
|
||||
|
||||
@@ -25,7 +25,7 @@ void ssh_sftp_close(sftp_session sftp);
|
||||
#define sftp_ssh(sftp) (sftp)->session
|
||||
|
||||
/* wrapping multiple sftp_read|write */
|
||||
int sftp_write2(sftp_file sf, const void *buf, size_t len);
|
||||
int sftp_read2(sftp_file sf, void *buf, size_t len);
|
||||
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);
|
||||
|
||||
#endif /* _SSH_H_ */
|
||||
|
||||
Reference in New Issue
Block a user