Finish up rsync for file receives

kittens/transfer/receive.py

@@ -29,8 +29,8 @@ from ..tui.utils import human_size
 from .librsync import PatchFile, signature_of_file
 from .send import Transfer
 from .utils import (
-    expand_home, random_id, render_progress_in_width, safe_divide,
+    expand_home, print_rsync_stats, random_id, render_progress_in_width,
-    should_be_compressed
+    safe_divide, should_be_compressed
 )
 
 debug
@@ -50,7 +50,9 @@ class File:
         self.expected_size = ftc.size
         self.expect_diff = False
         self.transmit_started_at = self.done_at = 0.
-        self.transmitted_bytes = 0
+        self.written_bytes = 0
+        self.received_bytes = 0
+        self.sent_bytes = 0
         self.ftype = ftc.ftype
         self.mtime = ftc.mtime
         self.spec_id = int(ftc.file_id)
@@ -71,6 +73,7 @@ class File:
         return f'File(rpath={self.remote_path!r}, lpath={self.expanded_local_path!r})'
 
     def write_data(self, data: bytes, is_last: bool) -> int:
+        self.received_bytes += len(data)
         data = self.decompressor(data, is_last)
         if self.ftype is FileType.symlink:
             self.remote_symlink_value += data
@@ -185,7 +188,6 @@ class ProgressTracker:
         self.transfered_stats_amt = 0
         self.transfered_stats_interval = 0.
         self.started_at = 0.
-        self.signature_bytes = 0
         self.done_files: List[File] = []
 
     def change_active_file(self, nf: File) -> None:
@@ -200,7 +202,7 @@ class ProgressTracker:
     def file_written(self, af: File, amt: int, is_done: bool) -> None:
         if self.active_file is not af:
             self.change_active_file(af)
-        af.transmitted_bytes += amt
+        af.written_bytes += amt
         self.total_transferred += amt
         self.transfers.append(Transfer(amt))
         now = self.transfers[-1].at
@@ -300,6 +302,7 @@ class Manager:
                 f.expect_diff = True
                 fs = signature_of_file(f.expanded_local_path)
                 for chunk in fs:
+                    f.sent_bytes += len(chunk)
                     for data in split_for_transfer(chunk, file_id=f.file_id):
                         yield data.serialize()
                 yield FileTransmissionCommand(file_id=f.file_id, action=Action.end_data).serialize()
@@ -544,7 +547,7 @@ class Receive(Handler):
         now = monotonic()
         self.render_progress(
             af.display_name, spinner_char=spinner_char, is_complete=is_complete,
-            bytes_so_far=af.transmitted_bytes, total_bytes=af.expected_size,
+            bytes_so_far=af.written_bytes, total_bytes=af.expected_size,
             secs_so_far=(af.done_at or now) - af.transmit_started_at,
             bytes_per_sec=safe_divide(p.transfered_stats_amt, p.transfered_stats_interval)
         )
@@ -628,3 +631,11 @@ def receive_main(cli_opts: TransferCLIOptions, args: List[str]) -> None:
     loop = Loop()
     handler = Receive(cli_opts, spec, dest)
     loop.loop(handler)
+    tsf = dsz = ssz = 0
+    for f in handler.manager.files:
+        if f.expect_diff:
+            tsf += f.expected_size
+            dsz += f.received_bytes
+            ssz += f.sent_bytes
+    if tsf:
+        print_rsync_stats(tsf, dsz, ssz)

kittens/transfer/send.py

@@ -31,7 +31,8 @@ from ..tui.utils import human_size
 from .librsync import LoadSignature, delta_for_file
 from .utils import (
     IdentityCompressor, ZlibCompressor, abspath, expand_home, home_path,
-    random_id, render_progress_in_width, safe_divide, should_be_compressed
+    print_rsync_stats, random_id, render_progress_in_width, safe_divide,
+    should_be_compressed
 )
 
 debug
@@ -725,10 +726,8 @@ def send_main(cli_opts: TransferCLIOptions, args: List[str]) -> None:
         for f in files:
             if f.ttype is TransmissionType.rsync:
                 tsf += f.file_size
-        print(
+        if tsf:
-            f'Rsync stats: Delta size: {human_size(p.total_transferred)} Signature size: {human_size(p.signature_bytes)}',
+            print_rsync_stats(tsf, p.total_transferred, p.signature_bytes)
-            f'Total rsynced files size: {human_size(tsf)}'
-        )
     if handler.failed_files:
         print(f'Transfer of {len(handler.failed_files)} out of {len(handler.manager.files)} files failed')
         for ff in handler.failed_files:

kittens/transfer/utils.py

@@ -192,3 +192,10 @@ class ZlibCompressor:
 
     def flush(self) -> bytes:
         return self.c.flush()
+
+
+def print_rsync_stats(total_bytes: int, delta_bytes: int, signature_bytes: int) -> None:
+    print('Rsync stats:')
+    print(f'  Delta size: {human_size(delta_bytes)} Signature size: {human_size(signature_bytes)}')
+    frac = (delta_bytes + signature_bytes) / max(1, total_bytes)
+    print(f'  Transmitted: {human_size(delta_bytes + signature_bytes)} of a total of {human_size(total_bytes)} ({frac:.1%})')

kitty/file_transmission.py

@@ -669,11 +669,19 @@ class ActiveSend:
             if af is None:
                 return None
             self.queued_files_map.pop(af.file_id, None)
-        chunk, uncompressed_sz = af.next_chunk()
+        while True:
-        if af.transmitted:
+            chunk, uncompressed_sz = af.next_chunk()
-            self.active_file = None
+            if af.transmitted:
-        self.pending_chunks.extend(split_for_transfer(chunk, file_id=af.file_id, mark_last=af.transmitted))
+                self.active_file = None
-        return self.pending_chunks.popleft() if self.pending_chunks else None
+                break
+            if chunk:
+                break
+        if chunk:
+            self.pending_chunks.extend(split_for_transfer(chunk, file_id=af.file_id, mark_last=af.transmitted))
+            return self.pending_chunks.popleft()
+        elif af.transmitted:
+            return FileTransmissionCommand(action=Action.end_data, file_id=af.file_id)
+        return None
 
     def return_chunk(self, ftc: FileTransmissionCommand) -> None:
         self.pending_chunks.insert(0, ftc)