streaming: refactor interface to be object-database-centric
Refactor the streaming interface to be centered around object databases instead of centered around the repository. Rename the functions accordingly. Signed-off-by: Patrick Steinhardt <ps@pks.im> Signed-off-by: Junio C Hamano <gitster@pobox.com>
Patrick Steinhardt committed
Nov 23, 2025 at 19:59 UTC
378ec56beba161abbef6e2c87d9bc2ac43c355f3
7 files changed
+71
-51
archive-tar.c
+3
-3
@@ -135,16 +135,16 @@ static int stream_blocked(struct repository *r, const struct object_id *oid)
135
char buf[BLOCKSIZE];
136
ssize_t readlen;
137
138
- st = open_istream(r, oid, &type, &sz, NULL);
138
+ st = odb_read_stream_open(r->objects, oid, &type, &sz, NULL);
139
if (!st)
140
return error(_("cannot stream blob %s"), oid_to_hex(oid));
141
for (;;) {
142
- readlen = read_istream(st, buf, sizeof(buf));
142
+ readlen = odb_read_stream_read(st, buf, sizeof(buf));
143
if (readlen <= 0)
144
break;
145
do_write_blocked(buf, readlen);
146
}
147
- close_istream(st);
147
+ odb_read_stream_close(st);
148
if (!readlen)
149
finish_record();
150
return readlen;
archive-zip.c
+6
-6
@@ -348,8 +348,8 @@ static int write_zip_entry(struct archiver_args *args,
348
349
if (!buffer) {
350
enum object_type type;
351
- stream = open_istream(args->repo, oid, &type, &size,
352
- NULL);
351
+ stream = odb_read_stream_open(args->repo->objects, oid,
352
+ &type, &size, NULL);
353
if (!stream)
354
return error(_("cannot stream blob %s"),
355
oid_to_hex(oid));
@@ -429,7 +429,7 @@ static int write_zip_entry(struct archiver_args *args,
429
ssize_t readlen;
430
431
for (;;) {
432
- readlen = read_istream(stream, buf, sizeof(buf));
432
+ readlen = odb_read_stream_read(stream, buf, sizeof(buf));
433
if (readlen <= 0)
434
break;
435
crc = crc32(crc, buf, readlen);
@@ -439,7 +439,7 @@ static int write_zip_entry(struct archiver_args *args,
439
buf, readlen);
440
write_or_die(1, buf, readlen);
441
}
442
- close_istream(stream);
442
+ odb_read_stream_close(stream);
443
if (readlen)
444
return readlen;
445
@@ -462,7 +462,7 @@ static int write_zip_entry(struct archiver_args *args,
462
zstream.avail_out = sizeof(compressed);
463
464
for (;;) {
465
- readlen = read_istream(stream, buf, sizeof(buf));
465
+ readlen = odb_read_stream_read(stream, buf, sizeof(buf));
466
if (readlen <= 0)
467
break;
468
crc = crc32(crc, buf, readlen);
@@ -486,7 +486,7 @@ static int write_zip_entry(struct archiver_args *args,
486
}
487
488
}
489
- close_istream(stream);
489
+ odb_read_stream_close(stream);
490
if (readlen)
491
return readlen;
492
builtin/index-pack.c
+4
-4
@@ -779,7 +779,7 @@ static int compare_objects(const unsigned char *buf, unsigned long size,
779
}
780
781
while (size) {
782
- ssize_t len = read_istream(data->st, data->buf, size);
782
+ ssize_t len = odb_read_stream_read(data->st, data->buf, size);
783
if (len == 0)
784
die(_("SHA1 COLLISION FOUND WITH %s !"),
785
oid_to_hex(&data->entry->idx.oid));
@@ -807,15 +807,15 @@ static int check_collison(struct object_entry *entry)
807
808
memset(&data, 0, sizeof(data));
809
data.entry = entry;
810
- data.st = open_istream(the_repository, &entry->idx.oid, &type, &size,
811
- NULL);
810
+ data.st = odb_read_stream_open(the_repository->objects, &entry->idx.oid,
811
+ &type, &size, NULL);
812
if (!data.st)
813
return -1;
814
if (size != entry->size || type != entry->type)
815
die(_("SHA1 COLLISION FOUND WITH %s !"),
816
oid_to_hex(&entry->idx.oid));
817
unpack_data(entry, compare_objects, &data);
818
- close_istream(data.st);
818
+ odb_read_stream_close(data.st);
819
free(data.buf);
820
return 0;
821
}
builtin/pack-objects.c
+7
-7
@@ -417,7 +417,7 @@ static unsigned long write_large_blob_data(struct odb_read_stream *st, struct ha
417
for (;;) {
418
ssize_t readlen;
419
int zret = Z_OK;
420
- readlen = read_istream(st, ibuf, sizeof(ibuf));
420
+ readlen = odb_read_stream_read(st, ibuf, sizeof(ibuf));
421
if (readlen == -1)
422
die(_("unable to read %s"), oid_to_hex(oid));
423
@@ -520,8 +520,8 @@ static unsigned long write_no_reuse_object(struct hashfile *f, struct object_ent
520
if (oe_type(entry) == OBJ_BLOB &&
521
oe_size_greater_than(&to_pack, entry,
522
repo_settings_get_big_file_threshold(the_repository)) &&
523
- (st = open_istream(the_repository, &entry->idx.oid, &type,
524
- &size, NULL)) != NULL)
523
+ (st = odb_read_stream_open(the_repository->objects, &entry->idx.oid,
524
+ &type, &size, NULL)) != NULL)
525
buf = NULL;
526
else {
527
buf = odb_read_object(the_repository->objects,
@@ -577,7 +577,7 @@ static unsigned long write_no_reuse_object(struct hashfile *f, struct object_ent
577
dheader[--pos] = 128 | (--ofs & 127);
578
if (limit && hdrlen + sizeof(dheader) - pos + datalen + hashsz >= limit) {
579
if (st)
580
- close_istream(st);
580
+ odb_read_stream_close(st);
581
free(buf);
582
return 0;
583
}
@@ -591,7 +591,7 @@ static unsigned long write_no_reuse_object(struct hashfile *f, struct object_ent
591
*/
592
if (limit && hdrlen + hashsz + datalen + hashsz >= limit) {
593
if (st)
594
- close_istream(st);
594
+ odb_read_stream_close(st);
595
free(buf);
596
return 0;
597
}
@@ -601,7 +601,7 @@ static unsigned long write_no_reuse_object(struct hashfile *f, struct object_ent
601
} else {
602
if (limit && hdrlen + datalen + hashsz >= limit) {
603
if (st)
604
- close_istream(st);
604
+ odb_read_stream_close(st);
605
free(buf);
606
return 0;
607
}
@@ -609,7 +609,7 @@ static unsigned long write_no_reuse_object(struct hashfile *f, struct object_ent
609
}
610
if (st) {
611
datalen = write_large_blob_data(st, f, &entry->idx.oid);
612
- close_istream(st);
612
+ odb_read_stream_close(st);
613
} else {
614
hashwrite(f, buf, datalen);
615
free(buf);
object-file.c
+4
-4
@@ -139,7 +139,7 @@ int stream_object_signature(struct repository *r, const struct object_id *oid)
139
char hdr[MAX_HEADER_LEN];
140
int hdrlen;
141
142
- st = open_istream(r, oid, &obj_type, &size, NULL);
142
+ st = odb_read_stream_open(r->objects, oid, &obj_type, &size, NULL);
143
if (!st)
144
return -1;
145
@@ -151,10 +151,10 @@ int stream_object_signature(struct repository *r, const struct object_id *oid)
151
git_hash_update(&c, hdr, hdrlen);
152
for (;;) {
153
char buf[1024 * 16];
154
- ssize_t readlen = read_istream(st, buf, sizeof(buf));
154
+ ssize_t readlen = odb_read_stream_read(st, buf, sizeof(buf));
155
156
if (readlen < 0) {
157
- close_istream(st);
157
+ odb_read_stream_close(st);
158
return -1;
159
}
160
if (!readlen)
@@ -162,7 +162,7 @@ int stream_object_signature(struct repository *r, const struct object_id *oid)
162
git_hash_update(&c, buf, readlen);
163
}
164
git_hash_final_oid(&real_oid, &c);
165
- close_istream(st);
165
+ odb_read_stream_close(st);
166
return !oideq(oid, &real_oid) ? -1 : 0;
167
}
168
streaming.c
+22
-22
@@ -35,7 +35,7 @@ static int close_istream_filtered(struct odb_read_stream *_fs)
35
{
36
struct odb_filtered_read_stream *fs = (struct odb_filtered_read_stream *)_fs;
37
free_stream_filter(fs->filter);
38
- return close_istream(fs->upstream);
38
+ return odb_read_stream_close(fs->upstream);
39
}
40
41
static ssize_t read_istream_filtered(struct odb_read_stream *_fs, char *buf,
@@ -87,7 +87,7 @@ static ssize_t read_istream_filtered(struct odb_read_stream *_fs, char *buf,
87
88
/* refill the input from the upstream */
89
if (!fs->input_finished) {
90
- fs->i_end = read_istream(fs->upstream, fs->ibuf, FILTER_BUFFER);
90
+ fs->i_end = odb_read_stream_read(fs->upstream, fs->ibuf, FILTER_BUFFER);
91
if (fs->i_end < 0)
92
return -1;
93
if (fs->i_end)
@@ -149,7 +149,7 @@ static ssize_t read_istream_incore(struct odb_read_stream *_st, char *buf, size_
149
}
150
151
static int open_istream_incore(struct odb_read_stream **out,
152
- struct repository *r,
152
+ struct object_database *odb,
153
const struct object_id *oid)
154
{
155
struct object_info oi = OBJECT_INFO_INIT;
@@ -163,7 +163,7 @@ static int open_istream_incore(struct odb_read_stream **out,
163
oi.typep = &stream.base.type;
164
oi.sizep = &stream.base.size;
165
oi.contentp = (void **)&stream.buf;
166
- ret = odb_read_object_info_extended(r->objects, oid, &oi,
166
+ ret = odb_read_object_info_extended(odb, oid, &oi,
167
OBJECT_INFO_DIE_IF_CORRUPT);
168
if (ret)
169
return ret;
@@ -180,47 +180,47 @@ static int open_istream_incore(struct odb_read_stream **out,
180
*****************************************************************************/
181
182
static int istream_source(struct odb_read_stream **out,
183
- struct repository *r,
183
+ struct object_database *odb,
184
const struct object_id *oid)
185
{
186
struct odb_source *source;
187
188
- if (!packfile_store_read_object_stream(out, r->objects->packfiles, oid))
188
+ if (!packfile_store_read_object_stream(out, odb->packfiles, oid))
189
return 0;
190
191
- odb_prepare_alternates(r->objects);
192
- for (source = r->objects->sources; source; source = source->next)
191
+ odb_prepare_alternates(odb);
192
+ for (source = odb->sources; source; source = source->next)
193
if (!odb_source_loose_read_object_stream(out, source, oid))
194
return 0;
195
196
- return open_istream_incore(out, r, oid);
196
+ return open_istream_incore(out, odb, oid);
197
}
198
199
/****************************************************************
200
* Users of streaming interface
201
****************************************************************/
202
203
-int close_istream(struct odb_read_stream *st)
203
+int odb_read_stream_close(struct odb_read_stream *st)
204
{
205
int r = st->close(st);
206
free(st);
207
return r;
208
}
209
210
-ssize_t read_istream(struct odb_read_stream *st, void *buf, size_t sz)
210
+ssize_t odb_read_stream_read(struct odb_read_stream *st, void *buf, size_t sz)
211
{
212
return st->read(st, buf, sz);
213
}
214
215
-struct odb_read_stream *open_istream(struct repository *r,
216
- const struct object_id *oid,
217
- enum object_type *type,
218
- unsigned long *size,
219
- struct stream_filter *filter)
215
+struct odb_read_stream *odb_read_stream_open(struct object_database *odb,
216
+ const struct object_id *oid,
217
+ enum object_type *type,
218
+ unsigned long *size,
219
+ struct stream_filter *filter)
220
{
221
struct odb_read_stream *st;
222
- const struct object_id *real = lookup_replace_object(r, oid);
223
- int ret = istream_source(&st, r, real);
222
+ const struct object_id *real = lookup_replace_object(odb->repo, oid);
223
+ int ret = istream_source(&st, odb, real);
224
225
if (ret)
226
return NULL;
@@ -229,7 +229,7 @@ struct odb_read_stream *open_istream(struct repository *r,
229
/* Add "&& !is_null_stream_filter(filter)" for performance */
230
struct odb_read_stream *nst = attach_stream_filter(st, filter);
231
if (!nst) {
232
- close_istream(st);
232
+ odb_read_stream_close(st);
233
return NULL;
234
}
235
st = nst;
@@ -252,7 +252,7 @@ int odb_stream_blob_to_fd(struct object_database *odb,
252
ssize_t kept = 0;
253
int result = -1;
254
255
- st = open_istream(odb->repo, oid, &type, &sz, filter);
255
+ st = odb_read_stream_open(odb, oid, &type, &sz, filter);
256
if (!st) {
257
if (filter)
258
free_stream_filter(filter);
@@ -263,7 +263,7 @@ int odb_stream_blob_to_fd(struct object_database *odb,
263
for (;;) {
264
char buf[1024 * 16];
265
ssize_t wrote, holeto;
266
- ssize_t readlen = read_istream(st, buf, sizeof(buf));
266
+ ssize_t readlen = odb_read_stream_read(st, buf, sizeof(buf));
267
268
if (readlen < 0)
269
goto close_and_exit;
@@ -294,6 +294,6 @@ int odb_stream_blob_to_fd(struct object_database *odb,
294
result = 0;
295
296
close_and_exit:
297
- close_istream(st);
297
+ odb_read_stream_close(st);
298
return result;
299
}
streaming.h
+25
-5
@@ -24,11 +24,31 @@ struct odb_read_stream {
24
unsigned long size; /* inflated size of full object */
25
};
26
27
-struct odb_read_stream *open_istream(struct repository *, const struct object_id *,
28
- enum object_type *, unsigned long *,
29
- struct stream_filter *);
30
-int close_istream(struct odb_read_stream *);
31
-ssize_t read_istream(struct odb_read_stream *, void *, size_t);
27
+/*
28
+ * Create a new object stream for the given object database. Populates the type
29
+ * and size pointers with the object's info. An optional filter can be used to
30
+ * transform the object's content.
31
+ *
32
+ * Returns the stream on success, a `NULL` pointer otherwise.
33
+ */
34
+struct odb_read_stream *odb_read_stream_open(struct object_database *odb,
35
+ const struct object_id *oid,
36
+ enum object_type *type,
37
+ unsigned long *size,
38
+ struct stream_filter *filter);
39
+
40
+/*
41
+ * Close the given read stream and release all resources associated with it.
42
+ * Returns 0 on success, a negative error code otherwise.
43
+ */
44
+int odb_read_stream_close(struct odb_read_stream *stream);
45
+
46
+/*
47
+ * Read data from the stream into the buffer. Returns 0 on EOF and the number
48
+ * of bytes read on success. Returns a negative error code in case reading from
49
+ * the stream fails.
50
+ */
51
+ssize_t odb_read_stream_read(struct odb_read_stream *stream, void *buf, size_t len);
52
53
/*
54
* Look up the object by its ID and write the full contents to the file