Skip to content

Commit a61b61c

Browse files
ThomasWaldmannThomasWaldmann
committed
refactor: create transfer_chunks function
1 parent 5c304ed commit a61b61c

1 file changed

Lines changed: 66 additions & 59 deletions

File tree

‎src/borg/archiver/transfer_cmd.py‎

Lines changed: 66 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,60 @@
1717
logger = create_logger()
1818

1919

20+
def transfer_chunks(upgrader, other_repository, other_manifest, other_chunks, archive, cache, recompress, dry_run):
21+
transfer = 0
22+
present = 0
23+
chunks = []
24+
for chunk_id, size in other_chunks:
25+
chunk_present = cache.seen_chunk(chunk_id, size)
26+
if not chunk_present: # target repo does not yet have this chunk
27+
if not dry_run:
28+
try:
29+
cdata = other_repository.get(chunk_id)
30+
except (Repository.ObjectNotFound, LegacyRepository.ObjectNotFound):
31+
# missing correct chunk in other_repository (source) will result in
32+
# a missing chunk in repository (destination).
33+
# we do NOT want to transfer all-zero replacement chunks from borg1 repos.
34+
pass
35+
else:
36+
if recompress == "never":
37+
# keep compressed payload same, verify via assert_id (that will
38+
# decompress, but avoid needing to compress it again):
39+
meta, data = other_manifest.repo_objs.parse(
40+
chunk_id, cdata, decompress=True, want_compressed=True, ro_type=ROBJ_FILE_STREAM
41+
)
42+
meta, data = upgrader.upgrade_compressed_chunk(meta, data)
43+
chunk_entry = cache.add_chunk(
44+
chunk_id,
45+
meta,
46+
data,
47+
stats=archive.stats,
48+
wait=False,
49+
compress=False,
50+
size=size,
51+
ctype=meta["ctype"],
52+
clevel=meta["clevel"],
53+
ro_type=ROBJ_FILE_STREAM,
54+
)
55+
elif recompress == "always":
56+
# always decompress and re-compress file data chunks
57+
meta, data = other_manifest.repo_objs.parse(chunk_id, cdata, ro_type=ROBJ_FILE_STREAM)
58+
chunk_entry = cache.add_chunk(
59+
chunk_id, meta, data, stats=archive.stats, wait=False, ro_type=ROBJ_FILE_STREAM
60+
)
61+
else:
62+
raise ValueError(f"unsupported recompress mode: {recompress}")
63+
cache.repository.async_response(wait=False)
64+
chunks.append(chunk_entry)
65+
transfer += size
66+
else:
67+
if not dry_run:
68+
chunk_entry = cache.reuse_chunk(chunk_id, size, archive.stats)
69+
chunks.append(chunk_entry)
70+
present += size
71+
return chunks, transfer, present
72+
73+
2074
class TransferMixIn:
2175
@with_other_repository(manifest=True, compatibility=(Manifest.Operation.READ,))
2276
@with_repository(manifest=True, cache=True, compatibility=(Manifest.Operation.WRITE,))
@@ -120,68 +174,21 @@ def do_transfer(self, args, *, repository, manifest, cache, other_repository=Non
120174
else:
121175
other_chunks = None
122176
if other_chunks is not None:
123-
chunks = []
124-
for chunk_id, size in other_chunks:
125-
chunk_present = cache.seen_chunk(chunk_id, size)
126-
if not chunk_present: # target repo does not yet have this chunk
127-
if not dry_run:
128-
try:
129-
cdata = other_repository.get(chunk_id)
130-
except (Repository.ObjectNotFound, LegacyRepository.ObjectNotFound):
131-
# missing correct chunk in other_repository (source) will result in
132-
# a missing chunk in repository (destination).
133-
# we do NOT want to transfer all-zero replacement chunks from borg1 repos.
134-
pass
135-
else:
136-
if args.recompress == "never":
137-
# keep compressed payload same, verify via assert_id (that will
138-
# decompress, but avoid needing to compress it again):
139-
meta, data = other_manifest.repo_objs.parse(
140-
chunk_id,
141-
cdata,
142-
decompress=True,
143-
want_compressed=True,
144-
ro_type=ROBJ_FILE_STREAM,
145-
)
146-
meta, data = upgrader.upgrade_compressed_chunk(meta, data)
147-
chunk_entry = cache.add_chunk(
148-
chunk_id,
149-
meta,
150-
data,
151-
stats=archive.stats,
152-
wait=False,
153-
compress=False,
154-
size=size,
155-
ctype=meta["ctype"],
156-
clevel=meta["clevel"],
157-
ro_type=ROBJ_FILE_STREAM,
158-
)
159-
elif args.recompress == "always":
160-
# always decompress and re-compress file data chunks
161-
meta, data = other_manifest.repo_objs.parse(
162-
chunk_id, cdata, ro_type=ROBJ_FILE_STREAM
163-
)
164-
chunk_entry = cache.add_chunk(
165-
chunk_id,
166-
meta,
167-
data,
168-
stats=archive.stats,
169-
wait=False,
170-
ro_type=ROBJ_FILE_STREAM,
171-
)
172-
else:
173-
raise ValueError(f"unsupported recompress mode: {args.recompress}")
174-
cache.repository.async_response(wait=False)
175-
chunks.append(chunk_entry)
176-
transfer_size += size
177-
else:
178-
if not dry_run:
179-
chunk_entry = cache.reuse_chunk(chunk_id, size, archive.stats)
180-
chunks.append(chunk_entry)
181-
present_size += size
177+
chunks, transfer, present = transfer_chunks(
178+
upgrader,
179+
other_repository,
180+
other_manifest,
181+
other_chunks,
182+
archive,
183+
cache,
184+
args.recompress,
185+
dry_run,
186+
)
182187
if not dry_run:
183188
item.chunks = chunks
184189
archive.stats.nfiles += 1
190+
transfer_size += transfer
191+
present_size += present
185192
if not dry_run:
186193
item = upgrader.upgrade_item(item=item)
187194
archive.add_item(item, show_progress=args.progress)

0 commit comments

Comments
 (0)