add chunk stitching; twice as fast long-distance uploads:

rather than sending each file chunk as a separate HTTP request,
sibling chunks will now be fused together into larger HTTP POSTs
which results in unreasonably huge speed boosts on some routes
( `2.6x` from Norway to US-East,  `1.6x` from US-West to Finland )

the `x-up2k-hash` request header now takes a comma-separated list
of chunk hashes, which must all be sibling chunks, resulting in
one large consecutive range of file data as the post body

a new global-option `--u2sz`, default `1,64,96`, sets the target
request size as 64 MiB, allowing the settings ui to specify any
value between 1 and 96 MiB, which is cloudflare's max value

this does not cause any issues for resumable uploads; thanks to the
streaming HTTP POST parser, each chunk will be verified and written
to disk as they arrive, meaning only the untransmitted chunks will
have to be resent in the event of a connection drop -- of course
assuming there are no misconfigured WAFs or caching-proxies

the previous up2k approach of uploading each chunk in a separate HTTP
POST was inefficient in many real-world scenarios, mainly due to TCP
window-scaling behaving erratically in some IXPs / along some routes

a particular link from Norway to Virginia,US is unusably slow for
the first 4 MiB, only reaching optimal speeds after 100 MiB, and
then immediately resets the scale when the request has been sent;
connection reuse does not help in this case

on this route, the basic-uploader was somehow faster than up2k
with 6 parallel uploads; only time i've seen this

Author: 9001

bin/u2c.py

@@ -1,8 +1,8 @@
 #!/usr/bin/env python3
 from __future__ import print_function, unicode_literals
 
-S_VERSION = "1.18"
-S_BUILD_DT = "2024-06-01"
+S_VERSION = "1.19"
+S_BUILD_DT = "2024-07-21"
 
 """
 u2c.py: upload to copyparty
@@ -119,6 +119,7 @@ def __init__(self, top, rel, size, lmod):
         self.nhs = 0
 
         # set by upload
+        self.nojoin = 0  # type: int
         self.up_b = 0  # type: int
         self.up_c = 0  # type: int
         self.cd = 0
@@ -130,10 +131,20 @@ def __init__(self, top, rel, size, lmod):
 class FileSlice(object):
     """file-like object providing a fixed window into a file"""
 
-    def __init__(self, file, cid):
+    def __init__(self, file, cids):
         # type: (File, str) -> None
 
-        self.car, self.len = file.kchunks[cid]
+        self.file = file
+        self.cids = cids
+
+        self.car, tlen = file.kchunks[cids[0]]
+        for cid in cids[1:]:
+            ofs, clen = file.kchunks[cid]
+            if ofs != self.car + tlen:
+                raise Exception(9)
+            tlen += clen
+
+        self.len = tlen
         self.cdr = self.car + self.len
         self.ofs = 0  # type: int
         self.f = open(file.abs, "rb", 512 * 1024)
@@ -636,13 +647,13 @@ def handshake(ar, file, search):
     return r["hash"], r["sprs"]
 
 
-def upload(file, cid, pw, stats):
-    # type: (File, str, str, str) -> None
-    """upload one specific chunk, `cid` (a chunk-hash)"""
+def upload(fsl, pw, stats):
+    # type: (FileSlice, str, str) -> None
+    """upload a range of file data, defined by one or more `cid` (chunk-hash)"""
 
     headers = {
-        "X-Up2k-Hash": cid,
-        "X-Up2k-Wark": file.wark,
+        "X-Up2k-Hash": ",".join(fsl.cids),
+        "X-Up2k-Wark": fsl.file.wark,
         "Content-Type": "application/octet-stream",
     }
 
@@ -652,15 +663,20 @@ def upload(file, cid, pw, stats):
     if pw:
         headers["Cookie"] = "=".join(["cppwd", pw])
 
-    f = FileSlice(file, cid)
     try:
-        r = req_ses.post(file.url, headers=headers, data=f)
+        r = req_ses.post(fsl.file.url, headers=headers, data=fsl)
+
+        if r.status_code == 400:
+            txt = r.text
+            if "already got that" in txt or "already being written" in txt:
+                fsl.file.nojoin = 1
+
         if not r:
             raise Exception(repr(r))
 
         _ = r.content
     finally:
-        f.f.close()
+        fsl.f.close()
 
 
 class Ctl(object):
@@ -743,7 +759,7 @@ def __init__(self, ar, stats=None):
 
             self.mutex = threading.Lock()
             self.q_handshake = Queue()  # type: Queue[File]
-            self.q_upload = Queue()  # type: Queue[tuple[File, str]]
+            self.q_upload = Queue()  # type: Queue[FileSlice]
 
             self.st_hash = [None, "(idle, starting...)"]  # type: tuple[File, int]
             self.st_up = [None, "(idle, starting...)"]  # type: tuple[File, int]
@@ -788,7 +804,8 @@ def _safe(self):
                 for nc, cid in enumerate(hs):
                     print("  {0} up {1}".format(ncs - nc, cid))
                     stats = "{0}/0/0/{1}".format(nf, self.nfiles - nf)
-                    upload(file, cid, self.ar.a, stats)
+                    fslice = FileSlice(file, [cid])
+                    upload(fslice, self.ar.a, stats)
 
             print("  ok!")
             if file.recheck:
@@ -1062,13 +1079,24 @@ def handshaker(self):
             if not hs:
                 kw = "uploaded" if file.up_b else "   found"
                 print("{0} {1}".format(kw, upath))
-            for cid in hs:
-                self.q_upload.put([file, cid])
+
+            cs = hs[:]
+            while cs:
+                fsl = FileSlice(file, cs[:1])
+                try:
+                    if file.nojoin:
+                        raise Exception()
+                    for n in range(2, self.ar.sz + 1):
+                        fsl = FileSlice(file, cs[:n])
+                except:
+                    pass
+                cs = cs[len(fsl.cids):]
+                self.q_upload.put(fsl)
 
     def uploader(self):
         while True:
-            task = self.q_upload.get()
-            if not task:
+            fsl = self.q_upload.get()
+            if not fsl:
                 self.st_up = [None, "(finished)"]
                 break
 
@@ -1086,22 +1114,23 @@ def uploader(self):
                 self.eta,
             )
 
-            file, cid = task
+            file = fsl.file
+            cids = fsl.cids
             try:
-                upload(file, cid, self.ar.a, stats)
+                upload(fsl, self.ar.a, stats)
             except Exception as ex:
-                t = "upload failed, retrying: {0} #{1} ({2})\n"
-                eprint(t.format(file.name, cid[:8], ex))
+                t = "upload failed, retrying: %s #%d+%d (%s)\n"
+                eprint(t % (file.name, cids[0][:8], len(cids) - 1, ex))
                 file.cd = time.time() + self.ar.cd
                 # handshake will fix it
 
             with self.mutex:
-                sz = file.kchunks[cid][1]
-                file.ucids = [x for x in file.ucids if x != cid]
+                sz = fsl.len
+                file.ucids = [x for x in file.ucids if x not in cids]
                 if not file.ucids:
                     self.q_handshake.put(file)
 
-                self.st_up = [file, cid]
+                self.st_up = [file, cids[0]]
                 file.up_b += sz
                 self.up_b += sz
                 self.up_br += sz
@@ -1164,6 +1193,7 @@ def main():
     ap = app.add_argument_group("performance tweaks")
     ap.add_argument("-j", type=int, metavar="CONNS", default=2, help="parallel connections")
     ap.add_argument("-J", type=int, metavar="CORES", default=hcores, help="num cpu-cores to use for hashing; set 0 or 1 for single-core hashing")
+    ap.add_argument("--sz", type=int, metavar="MiB", default=64, help="try to make each POST this big")
     ap.add_argument("-nh", action="store_true", help="disable hashing while uploading")
     ap.add_argument("-ns", action="store_true", help="no status panel (for slow consoles and macos)")
     ap.add_argument("--cd", type=float, metavar="SEC", default=5, help="delay before reattempting a failed handshake/upload")

copyparty/__main__.py

@@ -942,6 +942,7 @@ def add_upload(ap):
     ap2.add_argument("--sparse", metavar="MiB", type=int, default=4, help="windows-only: minimum size of incoming uploads through up2k before they are made into sparse files")
     ap2.add_argument("--turbo", metavar="LVL", type=int, default=0, help="configure turbo-mode in up2k client; [\033[32m-1\033[0m] = forbidden/always-off, [\033[32m0\033[0m] = default-off and warn if enabled, [\033[32m1\033[0m] = default-off, [\033[32m2\033[0m] = on, [\033[32m3\033[0m] = on and disable datecheck")
     ap2.add_argument("--u2j", metavar="JOBS", type=int, default=2, help="web-client: number of file chunks to upload in parallel; 1 or 2 is good for low-latency (same-country) connections, 4-8 for android clients, 16 for cross-atlantic (max=64)")
+    ap2.add_argument("--u2sz", metavar="N,N,N", type=u, default="1,64,96", help="web-client: default upload chunksize (MiB); sets \033[33mmin,default,max\033[0m in the settings gui. Each HTTP POST will aim for this size. Cloudflare max is 96. Big values are good for cross-atlantic but may increase HDD fragmentation on some FS. Disable this optimization with [\033[32m1,1,1\033[0m]")
     ap2.add_argument("--u2sort", metavar="TXT", type=u, default="s", help="upload order; [\033[32ms\033[0m]=smallest-first, [\033[32mn\033[0m]=alphabetical, [\033[32mfs\033[0m]=force-s, [\033[32mfn\033[0m]=force-n -- alphabetical is a bit slower on fiber/LAN but makes it easier to eyeball if everything went fine")
     ap2.add_argument("--write-uplog", action="store_true", help="write POST reports to textfiles in working-directory")
 

copyparty/httpcli.py

@@ -2199,33 +2199,36 @@ def handle_search(self, body: dict[str, Any]) -> bool:
 
     def handle_post_binary(self) -> bool:
         try:
-            remains = int(self.headers["content-length"])
+            postsize = remains = int(self.headers["content-length"])
         except:
             raise Pebkac(400, "you must supply a content-length for binary POST")
 
         try:
-            chash = self.headers["x-up2k-hash"]
+            chashes = self.headers["x-up2k-hash"].split(",")
             wark = self.headers["x-up2k-wark"]
         except KeyError:
             raise Pebkac(400, "need hash and wark headers for binary POST")
 
+        chashes = [x.strip() for x in chashes]
+
         vfs, _ = self.asrv.vfs.get(self.vpath, self.uname, False, True)
         ptop = (vfs.dbv or vfs).realpath
 
-        x = self.conn.hsrv.broker.ask("up2k.handle_chunk", ptop, wark, chash)
+        x = self.conn.hsrv.broker.ask("up2k.handle_chunks", ptop, wark, chashes)
         response = x.get()
-        chunksize, cstart, path, lastmod, sprs = response
+        chunksize, cstarts, path, lastmod, sprs = response
+        maxsize = chunksize * len(chashes)
+        cstart0 = cstarts[0]
 
         try:
             if self.args.nw:
                 path = os.devnull
 
-            if remains > chunksize:
-                raise Pebkac(400, "your chunk is too big to fit")
-
-            self.log("writing {} #{} @{} len {}".format(path, chash, cstart, remains))
+            if remains > maxsize:
+                t = "your client is sending %d bytes which is too much (server expected %d bytes at most)"
+                raise Pebkac(400, t % (remains, maxsize))
 
-            reader = read_socket(self.sr, self.args.s_rd_sz, remains)
+            self.log("writing {} {} @{} len {}".format(path, chashes, cstart0, remains))
 
             f = None
             fpool = not self.args.no_fpool and sprs
@@ -2239,37 +2242,43 @@ def handle_post_binary(self) -> bool:
             f = f or open(fsenc(path), "rb+", self.args.iobuf)
 
             try:
-                f.seek(cstart[0])
-                post_sz, _, sha_b64 = hashcopy(reader, f, self.args.s_wr_slp)
+                for chash, cstart in zip(chashes, cstarts):
+                    f.seek(cstart[0])
+                    reader = read_socket(
+                        self.sr, self.args.s_rd_sz, min(remains, chunksize)
+                    )
+                    post_sz, _, sha_b64 = hashcopy(reader, f, self.args.s_wr_slp)
 
-                if sha_b64 != chash:
-                    try:
-                        self.bakflip(f, cstart[0], post_sz, sha_b64, vfs.flags)
-                    except:
-                        self.log("bakflip failed: " + min_ex())
+                    if sha_b64 != chash:
+                        try:
+                            self.bakflip(f, cstart[0], post_sz, sha_b64, vfs.flags)
+                        except:
+                            self.log("bakflip failed: " + min_ex())
 
-                    t = "your chunk got corrupted somehow (received {} bytes); expected vs received hash:\n{}\n{}"
-                    raise Pebkac(400, t.format(post_sz, chash, sha_b64))
+                        t = "your chunk got corrupted somehow (received {} bytes); expected vs received hash:\n{}\n{}"
+                        raise Pebkac(400, t.format(post_sz, chash, sha_b64))
 
-                if len(cstart) > 1 and path != os.devnull:
-                    self.log(
-                        "clone {} to {}".format(
-                            cstart[0], " & ".join(unicode(x) for x in cstart[1:])
+                    remains -= chunksize
+
+                    if len(cstart) > 1 and path != os.devnull:
+                        self.log(
+                            "clone {} to {}".format(
+                                cstart[0], " & ".join(unicode(x) for x in cstart[1:])
+                            )
                         )
-                    )
-                    ofs = 0
-                    while ofs < chunksize:
-                        bufsz = max(4 * 1024 * 1024, self.args.iobuf)
-                        bufsz = min(chunksize - ofs, bufsz)
-                        f.seek(cstart[0] + ofs)
-                        buf = f.read(bufsz)
-                        for wofs in cstart[1:]:
-                            f.seek(wofs + ofs)
-                            f.write(buf)
+                        ofs = 0
+                        while ofs < chunksize:
+                            bufsz = max(4 * 1024 * 1024, self.args.iobuf)
+                            bufsz = min(chunksize - ofs, bufsz)
+                            f.seek(cstart[0] + ofs)
+                            buf = f.read(bufsz)
+                            for wofs in cstart[1:]:
+                                f.seek(wofs + ofs)
+                                f.write(buf)
 
-                        ofs += len(buf)
+                            ofs += len(buf)
 
-                    self.log("clone {} done".format(cstart[0]))
+                        self.log("clone {} done".format(cstart[0]))
 
                 if not fpool:
                     f.close()
@@ -2281,10 +2290,10 @@ def handle_post_binary(self) -> bool:
                 f.close()
                 raise
         finally:
-            x = self.conn.hsrv.broker.ask("up2k.release_chunk", ptop, wark, chash)
+            x = self.conn.hsrv.broker.ask("up2k.release_chunks", ptop, wark, chashes)
             x.get()  # block client until released
 
-        x = self.conn.hsrv.broker.ask("up2k.confirm_chunk", ptop, wark, chash)
+        x = self.conn.hsrv.broker.ask("up2k.confirm_chunks", ptop, wark, chashes)
         ztis = x.get()
         try:
             num_left, fin_path = ztis
@@ -2303,7 +2312,7 @@ def handle_post_binary(self) -> bool:
 
         cinf = self.headers.get("x-up2k-stat", "")
 
-        spd = self._spd(post_sz)
+        spd = self._spd(postsize)
         self.log("{:70} thank {}".format(spd, cinf))
         self.reply(b"thank")
         return True
@@ -4500,6 +4509,7 @@ def tx_browser(self) -> bool:
             "themes": self.args.themes,
             "turbolvl": self.args.turbo,
             "u2j": self.args.u2j,
+            "u2sz": self.args.u2sz,
             "idxh": int(self.args.ih),
             "u2sort": self.args.u2sort,
         }