casync-http: add max-active-chunks option

This commit introduces a new option --max-active-chunks=<MAX> that
limits the number of simultaneous chunk transfers from the remote.

The MAX number is the sum of:

    1. the number of chunks added to the cURL multi interface, and
    2. chunks downloaded and waiting to be sent to the remote.

It limits the number of chunks stored in memory that are ready to be
sent to the casync process; this limit the memory usage in the situation
where the network is faster than the pipe communication between the
helper and casync.

Author: gportay

doc/casync.rst

@@ -164,6 +164,7 @@ General options:
 --cache=<PATH>                  Directory to use as encoder cache
 --cache-auto, -c                Pick encoder cache directory automatically
 --rate-limit-bps=<LIMIT>        Maximum bandwidth in bytes/s for remote communication
+--max-active-chunks=<MAX>       Maximum number of simultaneously active chunks for remote communication
 --exclude-nodump=no             Don't exclude files with chattr(1)'s +d **nodump** flag when creating archive
 --exclude-submounts=yes         Exclude submounts when creating archive
 --exclude-file=no               Don't respect .caexclude files in the file tree

src/caremote.c

@@ -60,6 +60,7 @@ struct CaRemote {
 
         int log_level;
         uint64_t rate_limit_bps;
+        unsigned max_active_chunks;
 
         ReallocBuffer input_buffer;
         ReallocBuffer output_buffer;
@@ -248,6 +249,15 @@ int ca_remote_set_rate_limit_bps(CaRemote *rr, uint64_t rate_limit_bps) {
         return 0;
 }
 
+int ca_remote_set_max_active_chunks(CaRemote *rr, unsigned max_active_chunks) {
+        if (!rr)
+                return -EINVAL;
+
+        rr->max_active_chunks = max_active_chunks;
+
+        return 0;
+}
+
 int ca_remote_set_local_feature_flags(CaRemote *rr, uint64_t flags) {
         if (!rr)
                 return -EINVAL;
@@ -1001,6 +1011,9 @@ static int ca_remote_start(CaRemote *rr) {
                         if (rr->rate_limit_bps != UINT64_MAX)
                                 argc++;
 
+                        if (rr->max_active_chunks)
+                                argc++;
+
                         args = newa(char*, argc + 1);
 
                         if (rr->callout) {
@@ -1046,6 +1059,14 @@ static int ca_remote_start(CaRemote *rr) {
                                 i++;
                         }
 
+                        if (rr->max_active_chunks) {
+                                r = asprintf(args + i, "--max-active-chunks=%u", rr->max_active_chunks);
+                                if (r < 0)
+                                        return log_oom();
+
+                                i++;
+                        }
+
                         args[i + CA_REMOTE_ARG_OPERATION] = (char*) ((rr->local_feature_flags & (CA_PROTOCOL_PUSH_CHUNKS|CA_PROTOCOL_PUSH_INDEX|CA_PROTOCOL_PUSH_ARCHIVE)) ? "push" : "pull");
                         args[i + CA_REMOTE_ARG_BASE_URL] = /* rr->base_url ? rr->base_url + skip :*/ (char*) "-";
                         args[i + CA_REMOTE_ARG_ARCHIVE_URL] = rr->archive_url ? rr->archive_url + skip : (char*) "-";

src/caremote.h

@@ -52,6 +52,7 @@ int ca_remote_get_digest_type(CaRemote *rr, CaDigestType *ret);
 
 int ca_remote_set_log_level(CaRemote *rr, int log_level);
 int ca_remote_set_rate_limit_bps(CaRemote *rr, uint64_t rate_limit_bps);
+int ca_remote_set_max_active_chunks(CaRemote *rr, unsigned max_active_chunks);
 
 int ca_remote_set_io_fds(CaRemote *rr, int input_fd, int output_fd);
 int ca_remote_get_io_fds(CaRemote *rr, int *ret_input_fd, int *ret_output_fd);

src/casync-http.c

@@ -13,11 +13,38 @@
 #include "util.h"
 #include "list.h"
 
+/* The maximum number of active chunks is defined as the sum of:
+ * - number of chunks added to curl multi for download
+ * - number of chunks downloaded, and waiting to be sent to remote
+ *
+ * In situations where the server is local and super fast (ie. we receive chunks
+ * faster than we can send them to the remote), around 95% of the active chunks
+ * are chunks waiting to be sent to remote, hence this number can be seen as
+ * "maximum number of chunks sitting in ram".
+ *
+ * In situations where the server is away, around 95% of the active chunks are
+ * chunks added to curl multi. It doesn't mean "being downloaded" though, it's more
+ * a "maximum limit for concurrent downloads". The real number of running downloads
+ * might be lower, because:
+ * - if we're doing HTTP/1 and parallel connections, the hard limit is actually
+ *   defined by `MAX_HOST_CONNECTIONS`.
+ * - if we're doing HTTP/2 over a multiplexed connection, the number of parallel
+ *   streams is negociated between client and server.
+ *
+ * In effect, *I think* it's best to make this number quite large, because we
+ * don't want to underfeed libcurl and underperform. I think it's better to feed
+ * too many handles to the curl multi, and let libcurl decide internally what's
+ * best to do with it. Libcurl knows every details about the HTTP connection and
+ * will handle (parallel/multiplex/whatever) downloads better than us.
+ */
+#define MAX_ACTIVE_CHUNKS 64
+
 static volatile sig_atomic_t quit = false;
 
 static int arg_log_level = -1;
 static bool arg_verbose = false;
 static curl_off_t arg_rate_limit_bps = 0;
+static unsigned arg_max_active_chunks = MAX_ACTIVE_CHUNKS;
 static bool arg_ssl_trust_peer = false;
 
 typedef enum Protocol {
@@ -506,7 +533,7 @@ static CaChunkDownloader *ca_chunk_downloader_new(CaRemote *rr, const char *stor
         if (!dl->completed)
                 goto fail;
 
-        for (i = 0; i < 64; i++) {
+        for (i = 0; i < arg_max_active_chunks; i++) {
                 CURL *handle = NULL;
                 ChunkData *cd = NULL;
 
@@ -1203,6 +1230,7 @@ static int parse_argv(int argc, char *argv[]) {
 
         enum {
                 ARG_RATE_LIMIT_BPS = 0x100,
+                ARG_MAX_ACTIVE_CHUNKS,
                 ARG_SSL_TRUST_PEER,
         };
 
@@ -1211,6 +1239,7 @@ static int parse_argv(int argc, char *argv[]) {
                 { "log-level",      required_argument, NULL, 'l'                },
                 { "verbose",        no_argument,       NULL, 'v'                },
                 { "rate-limit-bps", required_argument, NULL, ARG_RATE_LIMIT_BPS },
+                { "max-active-chunks",    required_argument, NULL, ARG_MAX_ACTIVE_CHUNKS    },
                 { "ssl-trust-peer", no_argument,       NULL, ARG_SSL_TRUST_PEER },
                 {}
         };
@@ -1261,6 +1290,14 @@ static int parse_argv(int argc, char *argv[]) {
                         arg_rate_limit_bps = strtoll(optarg, NULL, 10);
                         break;
 
+                case ARG_MAX_ACTIVE_CHUNKS:
+                        r = safe_atou(optarg, &arg_max_active_chunks);
+                        if (r < 0 || arg_max_active_chunks == 0) {
+                                log_error("Invalid value for max-active-chunks, refusing");
+                                return -EINVAL;
+                        }
+                        break;
+
                 case ARG_SSL_TRUST_PEER:
                         arg_ssl_trust_peer = true;
                         break;

src/casync-tool.c

@@ -66,6 +66,7 @@ static size_t arg_chunk_size_min = 0;
 static size_t arg_chunk_size_avg = 0;
 static size_t arg_chunk_size_max = 0;
 static uint64_t arg_rate_limit_bps = UINT64_MAX;
+static unsigned arg_max_active_chunks = 0;
 static uint64_t arg_with = 0;
 static uint64_t arg_without = 0;
 static uid_t arg_uid_shift = 0, arg_uid_range = 0x10000U;
@@ -107,6 +108,8 @@ static void help(void) {
                "  -c --cache-auto            Pick encoder cache directory automatically\n"
                "     --rate-limit-bps=LIMIT  Maximum bandwidth in bytes/s for remote\n"
                "                             communication\n"
+               "     --max-active-chunks=MAX Maximum number of simultaneously active chunks for\n"
+               "                             remote communication\n"
                "     --exclude-nodump=no     Don't exclude files with chattr(1)'s +d 'nodump'\n"
                "                             flag when creating archive\n"
                "     --exclude-submounts=yes Exclude submounts when creating archive\n"
@@ -328,6 +331,7 @@ static int parse_argv(int argc, char *argv[]) {
                 ARG_SEED,
                 ARG_CACHE,
                 ARG_RATE_LIMIT_BPS,
+                ARG_MAX_ACTIVE_CHUNKS,
                 ARG_WITH,
                 ARG_WITHOUT,
                 ARG_WHAT,
@@ -362,6 +366,7 @@ static int parse_argv(int argc, char *argv[]) {
                 { "cache",             required_argument, NULL, ARG_CACHE             },
                 { "cache-auto",        no_argument,       NULL, 'c'                   },
                 { "rate-limit-bps",    required_argument, NULL, ARG_RATE_LIMIT_BPS    },
+                { "max-active-chunks", required_argument, NULL, ARG_MAX_ACTIVE_CHUNKS },
                 { "with",              required_argument, NULL, ARG_WITH              },
                 { "without",           required_argument, NULL, ARG_WITHOUT           },
                 { "what",              required_argument, NULL, ARG_WHAT              },
@@ -475,6 +480,14 @@ static int parse_argv(int argc, char *argv[]) {
 
                         break;
 
+                case ARG_MAX_ACTIVE_CHUNKS:
+                        r = safe_atou(optarg, &arg_max_active_chunks);
+                        if (r < 0) {
+                                log_error("Failed to parse --max-active-chunks= value %s", optarg);
+                                return -EINVAL;
+                        }
+                        break;
+
                 case ARG_WITH: {
                         uint64_t u;
 
@@ -1330,6 +1343,12 @@ static int verb_make(int argc, char *argv[]) {
                         return log_error_errno(r, "Failed to set rate limit: %m");
         }
 
+        if (arg_max_active_chunks) {
+                r = ca_sync_set_max_active_chunks(s, arg_max_active_chunks);
+                if (r < 0)
+                        return log_error_errno(r, "Failed to set max active chunks: %m");
+        }
+
         r = ca_sync_set_base_fd(s, input_fd);
         if (r < 0)
                 return log_error_errno(r, "Failed to set sync base: %m");
@@ -1635,6 +1654,12 @@ static int verb_extract(int argc, char *argv[]) {
                         return log_error_errno(r, "Failed to set rate limit: %m");
         }
 
+        if (arg_max_active_chunks) {
+                r = ca_sync_set_max_active_chunks(s, arg_max_active_chunks);
+                if (r < 0)
+                        return log_error_errno(r, "Failed to set max active chunks: %m");
+        }
+
         if (seek_path) {
                 if (output_fd >= 0)
                         r = ca_sync_set_boundary_fd(s, output_fd);
@@ -2796,6 +2821,12 @@ static int verb_mount(int argc, char *argv[]) {
                         return log_error_errno(r, "Failed to set rate limit: %m");
         }
 
+        if (arg_max_active_chunks) {
+                r = ca_sync_set_max_active_chunks(s, arg_max_active_chunks);
+                if (r < 0)
+                        return log_error_errno(r, "Failed to set max active chunks: %m");
+        }
+
         if (operation == MOUNT_ARCHIVE) {
                 if (input_fd >= 0)
                         r = ca_sync_set_archive_fd(s, input_fd);
@@ -2922,6 +2953,12 @@ static int verb_mkdev(int argc, char *argv[]) {
                         return log_error_errno(r, "Failed to set rate limit: %m");
         }
 
+        if (arg_max_active_chunks) {
+                r = ca_sync_set_max_active_chunks(s, arg_max_active_chunks);
+                if (r < 0)
+                        return log_error_errno(r, "Failed to set max active chunks: %m");
+        }
+
         if (operation == MKDEV_BLOB) {
                 if (input_fd >= 0)
                         r = ca_sync_set_archive_fd(s, input_fd);
@@ -3489,6 +3526,12 @@ static int verb_pull(int argc, char *argv[]) {
                         return log_error_errno(r, "Failed to set rate limit: %m");
         }
 
+        if (arg_max_active_chunks) {
+                r = ca_remote_set_max_active_chunks(rr, arg_max_active_chunks);
+                if (r < 0)
+                        return log_error_errno(r, "Failed to set max active chunks: %m");
+        }
+
         r = ca_remote_set_io_fds(rr, STDIN_FILENO, STDOUT_FILENO);
         if (r < 0)
                 return log_error_errno(r, "Failed to set I/O file descriptors: %m");
@@ -3648,6 +3691,12 @@ static int verb_push(int argc, char *argv[]) {
                         log_error_errno(r, "Failed to set rate limit: %m");
         }
 
+        if (arg_max_active_chunks) {
+                r = ca_remote_set_max_active_chunks(rr, arg_max_active_chunks);
+                if (r < 0)
+                        return log_error_errno(r, "Failed to set max active chunks: %m");
+        }
+
         r = ca_remote_set_io_fds(rr, STDIN_FILENO, STDOUT_FILENO);
         if (r < 0)
                 log_error_errno(r, "Failed to set I/O file descriptors: %m");

src/casync.c

@@ -112,6 +112,7 @@ struct CaSync {
 
         int log_level;
         size_t rate_limit_bps;
+        unsigned max_active_chunks;
 
         uint64_t feature_flags;
         uint64_t feature_flags_mask;
@@ -522,6 +523,15 @@ int ca_sync_set_log_level(CaSync *s, int log_level) {
         return 0;
 }
 
+int ca_sync_set_max_active_chunks(CaSync *s, unsigned max_active_chunks) {
+        if (!s)
+                return -EINVAL;
+
+        s->max_active_chunks = max_active_chunks;
+
+        return 0;
+}
+
 int ca_sync_set_rate_limit_bps(CaSync *s, uint64_t rate_limit_bps) {
         if (!s)
                 return -EINVAL;
@@ -694,6 +704,12 @@ int ca_sync_set_index_remote(CaSync *s, const char *url) {
                         return r;
         }
 
+        if (s->max_active_chunks > 0) {
+                r = ca_remote_set_max_active_chunks(s->remote_index, s->max_active_chunks);
+                if (r < 0)
+                        return r;
+        }
+
         r = ca_remote_set_index_url(s->remote_index, url);
         if (r < 0)
                 return r;

src/casync.h

@@ -33,6 +33,7 @@ DEFINE_TRIVIAL_CLEANUP_FUNC(CaSync *, ca_sync_unref);
 
 int ca_sync_set_log_level(CaSync *s, int log_level);
 int ca_sync_set_rate_limit_bps(CaSync *s, uint64_t rate_limit_bps);
+int ca_sync_set_max_active_chunks(CaSync *s, unsigned max_active_chunks);
 
 int ca_sync_set_feature_flags(CaSync *s, uint64_t flags);
 int ca_sync_get_feature_flags(CaSync *s, uint64_t *ret);