dilim/git
1
0
Fork 0
mirror of https://github.com/git/git.git synced 2026-08-07 08:31:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'ps/odb-streams' into seen

Browse Source 
The 'struct odb_read_stream' and 'struct odb_write_stream'
structures have been consolidated into a single unified 'struct
odb_stream' structure, simplifying object database streaming APIs
and enabling streaming of arbitrary object types.

* ps/odb-streams:
  odb/streaming: unify function names to create new streams
  odb/streaming: rename `struct input_zstream_data`
  odb/streaming: rename `struct read_object_fd_data`
  odb/streaming: consolidate read and write streams
  odb/streaming: rename `struct odb_read_stream`
  odb/streaming: support streaming arbitrary object types
  odb/streaming: drop `is_finished` field
  odb/streaming: track write stream size in the structure
This commit is contained in:
Junio C Hamano
2026-08-06 11:49:28 -07:00
parent b2b6e0ff20 ebdd7e10d6
commit 0dbd2c6bfc
23 changed files with 253 additions and 266 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
archive-tar.c
View File

@@ -129,20 +129,20 @@ static void write_trailer(void)
*/
static int stream_blocked(struct repository *r, const struct object_id *oid)
{
struct odb_read_stream *st;
struct odb_stream *st;
char buf[BLOCKSIZE];
ssize_t readlen;
st = odb_read_stream_open(r->objects, oid, NULL);
st = odb_stream_from_object(r->objects, oid, NULL);
if (!st)
return error(_("cannot stream blob %s"), oid_to_hex(oid));
for (;;) {
readlen = odb_read_stream_read(st, buf, sizeof(buf));
readlen = odb_stream_read(st, buf, sizeof(buf));
if (readlen <= 0)
break;
do_write_blocked(buf, readlen);
}
odb_read_stream_close(st);
odb_stream_close(st);
if (!readlen)
finish_record();
return readlen;

12
archive-zip.c
View File

@@ -309,7 +309,7 @@ static int write_zip_entry(struct archiver_args *args,
enum zip_method method;
unsigned char *out;
void *deflated = NULL;
struct odb_read_stream *stream = NULL;
struct odb_stream *stream = NULL;
unsigned long flags = 0;
int is_binary = -1;
const char *path_without_prefix = path + args->baselen;
@@ -347,7 +347,7 @@ static int write_zip_entry(struct archiver_args *args,
method = ZIP_METHOD_DEFLATE;
if (!buffer) {
stream = odb_read_stream_open(args->repo->objects, oid, NULL);
stream = odb_stream_from_object(args->repo->objects, oid, NULL);
if (!stream)
return error(_("cannot stream blob %s"),
oid_to_hex(oid));
@@ -428,7 +428,7 @@ static int write_zip_entry(struct archiver_args *args,
ssize_t readlen;
for (;;) {
readlen = odb_read_stream_read(stream, buf, sizeof(buf));
readlen = odb_stream_read(stream, buf, sizeof(buf));
if (readlen <= 0)
break;
crc = crc32(crc, buf, readlen);
@@ -438,7 +438,7 @@ static int write_zip_entry(struct archiver_args *args,
buf, readlen);
write_or_die(1, buf, readlen);
}
odb_read_stream_close(stream);
odb_stream_close(stream);
if (readlen)
return readlen;
@@ -461,7 +461,7 @@ static int write_zip_entry(struct archiver_args *args,
zstream.avail_out = sizeof(compressed);
for (;;) {
readlen = odb_read_stream_read(stream, buf, sizeof(buf));
readlen = odb_stream_read(stream, buf, sizeof(buf));
if (readlen <= 0)
break;
crc = crc32(crc, buf, readlen);
@@ -485,7 +485,7 @@ static int write_zip_entry(struct archiver_args *args,
}
}
odb_read_stream_close(stream);
odb_stream_close(stream);
if (readlen)
return readlen;

8
builtin/index-pack.c
View File

@@ -763,7 +763,7 @@ static void find_ref_delta_children(const struct object_id *oid,
struct compare_data {
struct object_entry *entry;
struct odb_read_stream *st;
struct odb_stream *st;
unsigned char *buf;
unsigned long buf_size;
};
@@ -780,7 +780,7 @@ static int compare_objects(const unsigned char *buf, unsigned long size,
}
while (size) {
ssize_t len = odb_read_stream_read(data->st, data->buf, size);
ssize_t len = odb_stream_read(data->st, data->buf, size);
if (len == 0)
die(_("SHA1 COLLISION FOUND WITH %s !"),
oid_to_hex(&data->entry->idx.oid));
@@ -806,14 +806,14 @@ static int check_collison(struct object_entry *entry)
memset(&data, 0, sizeof(data));
data.entry = entry;
data.st = odb_read_stream_open(the_repository->objects, &entry->idx.oid, NULL);
data.st = odb_stream_from_object(the_repository->objects, &entry->idx.oid, NULL);
if (!data.st)
return -1;
if (data.st->size != entry->size || data.st->type != entry->type)
die(_("SHA1 COLLISION FOUND WITH %s !"),
oid_to_hex(&entry->idx.oid));
unpack_data(entry, compare_objects, &data);
odb_read_stream_close(data.st);
odb_stream_close(data.st);
free(data.buf);
return 0;
}

18
builtin/pack-objects.c
View File

@@ -416,7 +416,7 @@ static unsigned long do_compress(void **pptr, unsigned long size)
return stream.total_out;
}
static unsigned long write_large_blob_data(struct odb_read_stream *st, struct hashfile *f,
static unsigned long write_large_blob_data(struct odb_stream *st, struct hashfile *f,
const struct object_id *oid)
{
git_zstream stream;
@@ -430,7 +430,7 @@ static unsigned long write_large_blob_data(struct odb_read_stream *st, struct ha
for (;;) {
ssize_t readlen;
int zret = Z_OK;
readlen = odb_read_stream_read(st, ibuf, sizeof(ibuf));
readlen = odb_stream_read(st, ibuf, sizeof(ibuf));
if (readlen == -1)
die(_("unable to read %s"), oid_to_hex(oid));
@@ -526,15 +526,15 @@ static unsigned long write_no_reuse_object(struct hashfile *f, struct object_ent
unsigned hdrlen;
enum object_type type;
void *buf;
struct odb_read_stream *st = NULL;
struct odb_stream *st = NULL;
const unsigned hashsz = the_hash_algo->rawsz;
if (!usable_delta) {
if (oe_type(entry) == OBJ_BLOB &&
oe_size_greater_than(&to_pack, entry,
repo_settings_get_big_file_threshold(the_repository)) &&
(st = odb_read_stream_open(the_repository->objects, &entry->idx.oid,
NULL)) != NULL) {
(st = odb_stream_from_object(the_repository->objects, &entry->idx.oid,
NULL)) != NULL) {
buf = NULL;
type = st->type;
size = st->size;
@@ -594,7 +594,7 @@ static unsigned long write_no_reuse_object(struct hashfile *f, struct object_ent
dheader[--pos] = 128 | (--ofs & 127);
if (limit && hdrlen + sizeof(dheader) - pos + datalen + hashsz >= limit) {
if (st)
odb_read_stream_close(st);
odb_stream_close(st);
free(buf);
return 0;
}
@@ -608,7 +608,7 @@ static unsigned long write_no_reuse_object(struct hashfile *f, struct object_ent
*/
if (limit && hdrlen + hashsz + datalen + hashsz >= limit) {
if (st)
odb_read_stream_close(st);
odb_stream_close(st);
free(buf);
return 0;
}
@@ -618,7 +618,7 @@ static unsigned long write_no_reuse_object(struct hashfile *f, struct object_ent
} else {
if (limit && hdrlen + datalen + hashsz >= limit) {
if (st)
odb_read_stream_close(st);
odb_stream_close(st);
free(buf);
return 0;
}
@@ -626,7 +626,7 @@ static unsigned long write_no_reuse_object(struct hashfile *f, struct object_ent
}
if (st) {
datalen = write_large_blob_data(st, f, &entry->idx.oid);
odb_read_stream_close(st);
odb_stream_close(st);
} else {
hashwrite(f, buf, datalen);
free(buf);

44
builtin/unpack-objects.c
View File

@@ -358,51 +358,55 @@ static void unpack_non_delta_entry(enum object_type type, unsigned long size,
write_object(nr, type, buf, size);
}
struct input_zstream_data {
struct zlib_stream {
struct odb_stream base;
git_zstream *zstream;
int status;
};
static ssize_t feed_input_zstream(struct odb_write_stream *in_stream,
unsigned char *buf, size_t buf_len)
static ssize_t zlib_stream_read(struct odb_stream *in_stream,
char *buf, size_t buf_len)
{
struct input_zstream_data *data = in_stream->data;
struct zlib_stream *data = container_of(in_stream, struct zlib_stream, base);
git_zstream *zstream = data->zstream;
void *in = fill(1);
if (in_stream->is_finished)
if (data->status != Z_OK)
return 0;
zstream->next_out = buf;
zstream->next_out = (unsigned char *) buf;
zstream->avail_out = buf_len;
zstream->next_in = in;
zstream->avail_in = len;
data->status = git_inflate(zstream, 0);
while (data->status == Z_OK && zstream->avail_out == buf_len) {
zstream->next_in = fill(1);
zstream->avail_in = len;
data->status = git_inflate(zstream, 0);
use(len - zstream->avail_in);
}
in_stream->is_finished = data->status != Z_OK;
use(len - zstream->avail_in);
return buf_len - zstream->avail_out;
}
static void stream_blob(unsigned long size, unsigned nr)
{
git_zstream zstream = { 0 };
struct input_zstream_data data = { 0 };
struct odb_write_stream in_stream = {
.read = feed_input_zstream,
.data = &data,
struct zlib_stream in_stream = {
.base = {
.read = zlib_stream_read,
.size = size,
.type = OBJ_BLOB,
},
.zstream = &zstream,
.status = Z_OK,
};
struct obj_info *info = &obj_list[nr];
data.zstream = &zstream;
git_inflate_init(&zstream);
if (odb_write_object_stream(the_repository->objects, &in_stream, size, &info->oid))
if (odb_write_object_stream(the_repository->objects, &in_stream.base, &info->oid))
die(_("failed to write object in stream"));
if (data.status != Z_STREAM_END)
die(_("inflate returned (%d)"), data.status);
if (in_stream.status != Z_STREAM_END)
die(_("inflate returned (%d)"), in_stream.status);
git_inflate_end(&zstream);
if (strict) {

76
object-file.c
View File

@@ -122,7 +122,7 @@ int check_object_signature(struct repository *r, const struct object_id *oid,
}
int stream_object_signature(struct repository *r,
struct odb_read_stream *st,
struct odb_stream *st,
const struct object_id *oid)
{
struct object_id real_oid;
@@ -138,7 +138,7 @@ int stream_object_signature(struct repository *r,
git_hash_update(&c, hdr, hdrlen);
for (;;) {
char buf[1024 * 16];
ssize_t readlen = odb_read_stream_read(st, buf, sizeof(buf));
ssize_t readlen = odb_stream_read(st, buf, sizeof(buf));
if (readlen < 0)
return -1;
if (!readlen)
@@ -702,9 +702,9 @@ static void prepare_packfile_transaction(struct odb_transaction_files *transacti
die_errno("unable to write pack header");
}
static int hash_blob_stream(struct odb_write_stream *stream,
const struct git_hash_algo *hash_algo,
struct object_id *result_oid, size_t size)
static int hash_stream(struct odb_stream *stream,
const struct git_hash_algo *hash_algo,
struct object_id *result_oid)
{
unsigned char buf[16384];
struct git_hash_ctx ctx;
@@ -712,22 +712,23 @@ static int hash_blob_stream(struct odb_write_stream *stream,
size_t bytes_hashed = 0;
header_len = format_object_header((char *)buf, sizeof(buf),
OBJ_BLOB, size);
stream->type, stream->size);
git_hash_init(&ctx, hash_algo);
git_hash_update(&ctx, buf, header_len);
while (!stream->is_finished) {
ssize_t read_result = odb_write_stream_read(stream, buf,
sizeof(buf));
while (1) {
ssize_t read_result = odb_stream_read(stream, buf,
sizeof(buf));
if (read_result < 0)
return -1;
if (!read_result)
break;
git_hash_update(&ctx, buf, read_result);
bytes_hashed += read_result;
}
if (bytes_hashed != size)
if (bytes_hashed != stream->size)
return -1;
git_hash_final_oid(result_oid, &ctx);
@@ -739,9 +740,9 @@ static int hash_blob_stream(struct odb_write_stream *stream,
* Read the contents from the stream provided, streaming it to the
* packfile in state while updating the hash in ctx.
*/
static void stream_blob_to_pack(struct transaction_packfile *state,
struct git_hash_ctx *ctx, size_t size,
struct odb_write_stream *stream)
static void stream_to_pack(struct transaction_packfile *state,
struct git_hash_ctx *ctx,
struct odb_stream *stream)
{
git_zstream s;
unsigned char ibuf[16384];
@@ -749,21 +750,23 @@ static void stream_blob_to_pack(struct transaction_packfile *state,
unsigned hdrlen;
int status = Z_OK;
struct repo_config_values *cfg = repo_config_values(the_repository);
bool is_finished = false;
size_t bytes_read = 0;
git_deflate_init(&s, cfg->pack_compression_level);
hdrlen = encode_in_pack_object_header(obuf, sizeof(obuf), OBJ_BLOB, size);
hdrlen = encode_in_pack_object_header(obuf, sizeof(obuf), stream->type, stream->size);
s.next_out = obuf + hdrlen;
s.avail_out = sizeof(obuf) - hdrlen;
while (status != Z_STREAM_END) {
if (!stream->is_finished && !s.avail_in) {
ssize_t rsize = odb_write_stream_read(stream, ibuf,
sizeof(ibuf));
if (!is_finished && !s.avail_in) {
ssize_t rsize = odb_stream_read(stream, ibuf,
sizeof(ibuf));
if (rsize < 0)
die("failed to read blob data");
die("failed to read object data");
if (!rsize)
is_finished = true;
git_hash_update(ctx, ibuf, rsize);
@@ -772,7 +775,7 @@ static void stream_blob_to_pack(struct transaction_packfile *state,
bytes_read += rsize;
}
status = git_deflate(&s, stream->is_finished ? Z_FINISH : 0);
status = git_deflate(&s, is_finished ? Z_FINISH : 0);
if (!s.avail_out || status == Z_STREAM_END) {
size_t written = s.next_out - obuf;
@@ -793,9 +796,9 @@ static void stream_blob_to_pack(struct transaction_packfile *state,
}
}
if (bytes_read != size)
die("read %" PRIuMAX " bytes of blob data, but expected %" PRIuMAX " bytes",
(uintmax_t)bytes_read, (uintmax_t)size);
if (bytes_read != stream->size)
die("read %" PRIuMAX " bytes of object data, but expected %" PRIuMAX " bytes",
(uintmax_t)bytes_read, (uintmax_t)stream->size);
git_deflate_end(&s);
}
@@ -865,12 +868,11 @@ clear_exit:
* result, which we need to know beforehand when writing a git object.
* Since the primary motivation for trying to stream from the working
* tree file and to avoid mmaping it in core is to deal with large
* binary blobs, they generally do not want to get any conversion, and
* objects, they generally do not want to get any conversion, and
* callers should avoid this code path when filters are requested.
*/
static int odb_transaction_files_write_object_stream(struct odb_transaction *base,
struct odb_write_stream *stream,
size_t size,
struct odb_stream *stream,
struct object_id *result_oid)
{
struct odb_transaction_files *transaction = container_of(base,
@@ -884,7 +886,7 @@ static int odb_transaction_files_write_object_stream(struct odb_transaction *bas
struct pack_idx_entry *idx;
header_len = format_object_header((char *)obuf, sizeof(obuf),
OBJ_BLOB, size);
stream->type, stream->size);
git_hash_init(&ctx, transaction->base.source->odb->repo->hash_algo);
git_hash_update(&ctx, obuf, header_len);
@@ -899,7 +901,7 @@ static int odb_transaction_files_write_object_stream(struct odb_transaction *bas
* to zlib compression and is sufficient for this check.
*/
if (state->nr_written && pack_size_limit_cfg &&
pack_size_limit_cfg < state->offset + size)
pack_size_limit_cfg < state->offset + stream->size)
flush_packfile_transaction(transaction);
CALLOC_ARRAY(idx, 1);
@@ -909,7 +911,7 @@ static int odb_transaction_files_write_object_stream(struct odb_transaction *bas
hashfile_checkpoint(state->f, &checkpoint);
idx->offset = state->offset;
crc32_begin(state->f);
stream_blob_to_pack(state, &ctx, size, stream);
stream_to_pack(state, &ctx, stream);
git_hash_final_oid(result_oid, &ctx);
idx->crc32 = crc32_end(state->f);
@@ -950,8 +952,8 @@ int index_fd(struct index_state *istate, struct object_id *oid,
ret = index_core(istate, oid, fd, xsize_t(st->st_size),
type, path, flags);
} else {
struct odb_write_stream stream;
odb_write_stream_from_fd(&stream, fd, xsize_t(st->st_size));
struct odb_stream *stream = odb_stream_from_fd(fd, xsize_t(st->st_size),
OBJ_BLOB);
if (flags & INDEX_WRITE_OBJECT) {
struct object_database *odb = the_repository->objects;
@@ -961,18 +963,14 @@ int index_fd(struct index_state *istate, struct object_id *oid,
if (!inflight)
odb_transaction_begin_or_die(odb, &transaction, 0);
ret = odb_transaction_write_object_stream(transaction,
&stream,
xsize_t(st->st_size),
oid);
stream, oid);
if (!inflight)
odb_transaction_commit(transaction);
} else {
ret = hash_blob_stream(&stream,
the_repository->hash_algo, oid,
xsize_t(st->st_size));
ret = hash_stream(stream, the_repository->hash_algo, oid);
}
odb_write_stream_release(&stream);
odb_stream_close(stream);
}
close(fd);

2
object-file.h
View File

@@ -101,7 +101,7 @@ int check_object_signature(struct repository *r, const struct object_id *oid,
* the streaming interface and rehash it to do the same.
*/
int stream_object_signature(struct repository *r,
struct odb_read_stream *stream,
struct odb_stream *stream,
const struct object_id *oid);
enum finalize_object_file_flags {

6
object.c
View File

@@ -345,7 +345,7 @@ struct object *parse_object_with_flags(struct repository *r,
if ((!obj || obj->type == OBJ_NONE || obj->type == OBJ_BLOB) &&
odb_read_object_info(r->objects, oid, NULL) == OBJ_BLOB) {
if (!skip_hash) {
struct odb_read_stream *stream = odb_read_stream_open(r->objects, oid, NULL);
struct odb_stream *stream = odb_stream_from_object(r->objects, oid, NULL);
if (!stream) {
error(_("unable to open object stream for %s"), oid_to_hex(oid));
@@ -354,11 +354,11 @@ struct object *parse_object_with_flags(struct repository *r,
if (stream_object_signature(r, stream, repl) < 0) {
error(_("hash mismatch %s"), oid_to_hex(oid));
odb_read_stream_close(stream);
odb_stream_close(stream);
return NULL;
}
odb_read_stream_close(stream);
odb_stream_close(stream);
}
parse_blob_buffer(lookup_blob(r, oid));
return lookup_object(r, oid);

4
odb.c
View File

@@ -1029,10 +1029,10 @@ int odb_write_object_ext(struct object_database *odb,
}
int odb_write_object_stream(struct object_database *odb,
struct odb_write_stream *stream, size_t len,
struct odb_stream *stream,
struct object_id *oid)
{
return odb_source_write_object_stream(odb->sources, stream, len, oid);
return odb_source_write_object_stream(odb->sources, stream, oid);
}
int odb_optimize(struct object_database *odb,

4
odb.h
View File

@@ -688,10 +688,10 @@ static inline int odb_write_object(struct object_database *odb,
return odb_write_object_ext(odb, buf, len, type, oid, NULL, 0);
}
struct odb_write_stream;
struct odb_stream;
int odb_write_object_stream(struct object_database *odb,
struct odb_write_stream *stream, size_t len,
struct odb_stream *stream,
struct object_id *oid);
void parse_alternates(const char *string,

7
odb/source-files.c
View File

@@ -90,7 +90,7 @@ static int odb_source_files_read_object_info(struct odb_source *source,
return -1;
}
static int odb_source_files_read_object_stream(struct odb_read_stream **out,
static int odb_source_files_read_object_stream(struct odb_stream **out,
struct odb_source *source,
const struct object_id *oid)
{
@@ -201,12 +201,11 @@ static int odb_source_files_write_object(struct odb_source *source,
}
static int odb_source_files_write_object_stream(struct odb_source *source,
struct odb_write_stream *stream,
size_t len,
struct odb_stream *stream,
struct object_id *oid)
{
struct odb_source_files *files = odb_source_files_downcast(source);
return odb_source_write_object_stream(&files->loose->base, stream, len, oid);
return odb_source_write_object_stream(&files->loose->base, stream, oid);
}
static int odb_source_files_begin_transaction(struct odb_source *source,

35
odb/source-inmemory.c
View File

@@ -73,12 +73,12 @@ static int odb_source_inmemory_read_object_info(struct odb_source *source,
}
struct odb_read_stream_inmemory {
struct odb_read_stream base;
struct odb_stream base;
const unsigned char *buf;
size_t offset;
};
static ssize_t odb_read_stream_inmemory_read(struct odb_read_stream *stream,
static ssize_t odb_read_stream_inmemory_read(struct odb_stream *stream,
char *buf, size_t buf_len)
{
struct odb_read_stream_inmemory *inmemory =
@@ -94,12 +94,12 @@ static ssize_t odb_read_stream_inmemory_read(struct odb_read_stream *stream,
return bytes;
}
static int odb_read_stream_inmemory_close(struct odb_read_stream *stream UNUSED)
static int odb_read_stream_inmemory_close(struct odb_stream *stream UNUSED)
{
return 0;
}
static int odb_source_inmemory_read_object_stream(struct odb_read_stream **out,
static int odb_source_inmemory_read_object_stream(struct odb_stream **out,
struct odb_source *source,
const struct object_id *oid)
{
@@ -256,8 +256,7 @@ static int odb_source_inmemory_write_object(struct odb_source *source,
}
static int odb_source_inmemory_write_object_stream(struct odb_source *source,
struct odb_write_stream *stream,
size_t len,
struct odb_stream *stream,
struct object_id *oid)
{
char buf[16384];
@@ -265,12 +264,19 @@ static int odb_source_inmemory_write_object_stream(struct odb_source *source,
char *data;
int ret;
CALLOC_ARRAY(data, len);
while (!stream->is_finished) {
CALLOC_ARRAY(data, stream->size);
while (1) {
ssize_t bytes_read;
bytes_read = odb_write_stream_read(stream, buf, sizeof(buf));
if (total_read + bytes_read > len) {
bytes_read = odb_stream_read(stream, buf, sizeof(buf));
if (bytes_read < 0) {
ret = error("failed to read object stream");
goto out;
}
if (!bytes_read)
break;
if (total_read + bytes_read > stream->size) {
ret = error("object stream yielded more bytes than expected");
goto out;
}
@@ -279,15 +285,16 @@ static int odb_source_inmemory_write_object_stream(struct odb_source *source,
total_read += bytes_read;
}
if (total_read != len) {
if (total_read != stream->size) {
ret = error("object stream yielded less bytes than expected");
goto out;
}
hash_object_file(source->odb->repo->hash_algo, data, total_read, OBJ_BLOB, oid);
hash_object_file(source->odb->repo->hash_algo, data, total_read,
stream->type, oid);
ret = odb_source_inmemory_write_object(source, data, len, OBJ_BLOB, oid,
NULL, NULL, 0);
ret = odb_source_inmemory_write_object(source, data, stream->size,
stream->type, oid, NULL, NULL, 0);
if (ret < 0)
goto out;

33
odb/source-loose.c
View File

@@ -278,7 +278,7 @@ out:
}
struct odb_loose_read_stream {
struct odb_read_stream base;
struct odb_stream base;
git_zstream z;
enum {
ODB_LOOSE_READ_STREAM_INUSE,
@@ -292,7 +292,7 @@ struct odb_loose_read_stream {
int hdr_used;
};
static ssize_t read_istream_loose(struct odb_read_stream *_st, char *buf, size_t sz)
static ssize_t read_istream_loose(struct odb_stream *_st, char *buf, size_t sz)
{
struct odb_loose_read_stream *st =
container_of(_st, struct odb_loose_read_stream, base);
@@ -339,7 +339,7 @@ static ssize_t read_istream_loose(struct odb_read_stream *_st, char *buf, size_t
return total_read;
}
static int close_istream_loose(struct odb_read_stream *_st)
static int close_istream_loose(struct odb_stream *_st)
{
struct odb_loose_read_stream *st =
container_of(_st, struct odb_loose_read_stream, base);
@@ -350,7 +350,7 @@ static int close_istream_loose(struct odb_read_stream *_st)
return 0;
}
static int odb_source_loose_read_object_stream(struct odb_read_stream **out,
static int odb_source_loose_read_object_stream(struct odb_stream **out,
struct odb_source *source,
const struct object_id *oid)
{
@@ -845,8 +845,7 @@ static int odb_source_loose_write_object(struct odb_source *source,
}
static int odb_source_loose_write_object_stream(struct odb_source *source,
struct odb_write_stream *in_stream,
size_t len,
struct odb_stream *in_stream,
struct object_id *oid)
{
struct odb_source_loose *loose = odb_source_loose_downcast(source);
@@ -860,6 +859,7 @@ static int odb_source_loose_write_object_stream(struct odb_source *source,
struct strbuf filename = STRBUF_INIT;
unsigned char buf[8192];
int dirlen;
bool is_finished = false;
char hdr[MAX_HEADER_LEN];
int hdrlen;
@@ -868,7 +868,7 @@ static int odb_source_loose_write_object_stream(struct odb_source *source,
/* Since oid is not determined, save tmp file to odb path. */
strbuf_addf(&filename, "%s/", loose->base.path);
hdrlen = format_object_header(hdr, sizeof(hdr), OBJ_BLOB, len);
hdrlen = format_object_header(hdr, sizeof(hdr), in_stream->type, in_stream->size);
/*
* Common steps for write_loose_object and stream_loose_object to
@@ -890,21 +890,24 @@ static int odb_source_loose_write_object_stream(struct odb_source *source,
do {
unsigned char *in0 = stream.next_in;
if (!stream.avail_in && !in_stream->is_finished) {
ssize_t read_len = odb_write_stream_read(in_stream, buf,
sizeof(buf));
if (!stream.avail_in && !is_finished) {
ssize_t read_len = odb_stream_read(in_stream, buf,
sizeof(buf));
if (read_len < 0) {
close(fd);
err = -1;
goto cleanup;
}
/* All data has been read. */
if (!read_len) {
is_finished = true;
flush = 1;
}
stream.avail_in = read_len;
stream.next_in = buf;
in0 = buf;
/* All data has been read. */
if (in_stream->is_finished)
flush = 1;
}
ret = write_loose_object_common(loose, &c, &compat_c, &stream, flush, in0, fd,
compressed, sizeof(compressed));
@@ -916,9 +919,9 @@ static int odb_source_loose_write_object_stream(struct odb_source *source,
*/
} while (ret == Z_OK || ret == Z_BUF_ERROR);
if (stream.total_in != len + hdrlen)
if (stream.total_in != in_stream->size + hdrlen)
die(_("write stream object %"PRIuMAX" != %"PRIuMAX), (uintmax_t)stream.total_in,
(uintmax_t)len + hdrlen);
(uintmax_t)in_stream->size + hdrlen);
/*
* Common steps for write_loose_object and stream_loose_object to

5
odb/source-packed.c
View File

@@ -70,7 +70,7 @@ static int odb_source_packed_read_object_info(struct odb_source *source,
return 0;
}
static int odb_source_packed_read_object_stream(struct odb_read_stream **out,
static int odb_source_packed_read_object_stream(struct odb_stream **out,
struct odb_source *source,
const struct object_id *oid)
{
@@ -609,8 +609,7 @@ static int odb_source_packed_write_object(struct odb_source *source UNUSED,
}
static int odb_source_packed_write_object_stream(struct odb_source *source UNUSED,
struct odb_write_stream *stream UNUSED,
size_t len UNUSED,
struct odb_stream *stream UNUSED,
struct object_id *oid UNUSED)
{
return error("packed backend cannot write object streams");

13
odb/source.h
View File

@@ -32,7 +32,7 @@ enum odb_source_type {
const char *odb_source_type_to_name(enum odb_source_type type);
struct object_id;
struct odb_read_stream;
struct odb_stream;
struct strvec;
/*
@@ -143,7 +143,7 @@ struct odb_source {
* The callback is expected to return a negative error code in case
* creating the object stream has failed, 0 otherwise.
*/
int (*read_object_stream)(struct odb_read_stream **out,
int (*read_object_stream)(struct odb_stream **out,
struct odb_source *source,
const struct object_id *oid);
@@ -239,7 +239,7 @@ struct odb_source {
* otherwise.
*/
int (*write_object_stream)(struct odb_source *source,
struct odb_write_stream *stream, size_t len,
struct odb_stream *stream,
struct object_id *oid);
/*
@@ -383,7 +383,7 @@ static inline int odb_source_read_object_info(struct odb_source *source,
* Create a new read stream for the given object ID. Returns 0 on success, a
* negative error code otherwise.
*/
static inline int odb_source_read_object_stream(struct odb_read_stream **out,
static inline int odb_source_read_object_stream(struct odb_stream **out,
struct odb_source *source,
const struct object_id *oid)
{
@@ -480,11 +480,10 @@ static inline int odb_source_write_object(struct odb_source *source,
* out pointer for the object ID.
*/
static inline int odb_source_write_object_stream(struct odb_source *source,
struct odb_write_stream *stream,
size_t len,
struct odb_stream *stream,
struct object_id *oid)
{
return source->write_object_stream(source, stream, len, oid);
return source->write_object_stream(source, stream, oid);
}
/*

108
odb/streaming.c
View File

@@ -20,8 +20,8 @@
*****************************************************************/
struct odb_filtered_read_stream {
struct odb_read_stream base;
struct odb_read_stream *upstream;
struct odb_stream base;
struct odb_stream *upstream;
struct stream_filter *filter;
char ibuf[FILTER_BUFFER];
char obuf[FILTER_BUFFER];
@@ -30,14 +30,14 @@ struct odb_filtered_read_stream {
int input_finished;
};
static int close_istream_filtered(struct odb_read_stream *_fs)
static int close_istream_filtered(struct odb_stream *_fs)
{
struct odb_filtered_read_stream *fs = (struct odb_filtered_read_stream *)_fs;
free_stream_filter(fs->filter);
return odb_read_stream_close(fs->upstream);
return odb_stream_close(fs->upstream);
}
static ssize_t read_istream_filtered(struct odb_read_stream *_fs, char *buf,
static ssize_t read_istream_filtered(struct odb_stream *_fs, char *buf,
size_t sz)
{
struct odb_filtered_read_stream *fs = (struct odb_filtered_read_stream *)_fs;
@@ -86,7 +86,7 @@ static ssize_t read_istream_filtered(struct odb_read_stream *_fs, char *buf,
/* refill the input from the upstream */
if (!fs->input_finished) {
fs->i_end = odb_read_stream_read(fs->upstream, fs->ibuf, FILTER_BUFFER);
fs->i_end = odb_stream_read(fs->upstream, fs->ibuf, FILTER_BUFFER);
if (fs->i_end < 0)
return -1;
if (fs->i_end)
@@ -97,8 +97,8 @@ static ssize_t read_istream_filtered(struct odb_read_stream *_fs, char *buf,
return filled;
}
static struct odb_read_stream *attach_stream_filter(struct odb_read_stream *st,
struct stream_filter *filter)
static struct odb_stream *attach_stream_filter(struct odb_stream *st,
struct stream_filter *filter)
{
struct odb_filtered_read_stream *fs;
@@ -120,19 +120,19 @@ static struct odb_read_stream *attach_stream_filter(struct odb_read_stream *st,
*****************************************************************/
struct odb_incore_read_stream {
struct odb_read_stream base;
struct odb_stream base;
char *buf; /* from odb_read_object_info_extended() */
unsigned long read_ptr;
};
static int close_istream_incore(struct odb_read_stream *_st)
static int close_istream_incore(struct odb_stream *_st)
{
struct odb_incore_read_stream *st = (struct odb_incore_read_stream *)_st;
free(st->buf);
return 0;
}
static ssize_t read_istream_incore(struct odb_read_stream *_st, char *buf, size_t sz)
static ssize_t read_istream_incore(struct odb_stream *_st, char *buf, size_t sz)
{
struct odb_incore_read_stream *st = (struct odb_incore_read_stream *)_st;
size_t read_size = sz;
@@ -147,7 +147,7 @@ static ssize_t read_istream_incore(struct odb_read_stream *_st, char *buf, size_
return read_size;
}
static int open_istream_incore(struct odb_read_stream **out,
static int open_istream_incore(struct odb_stream **out,
struct object_database *odb,
const struct object_id *oid)
{
@@ -178,7 +178,7 @@ static int open_istream_incore(struct odb_read_stream **out,
* static helpers variables and functions for users of streaming interface
*****************************************************************************/
static int istream_source(struct odb_read_stream **out,
static int istream_source(struct odb_stream **out,
struct object_database *odb,
const struct object_id *oid)
{
@@ -196,23 +196,23 @@ static int istream_source(struct odb_read_stream **out,
* Users of streaming interface
****************************************************************/
int odb_read_stream_close(struct odb_read_stream *st)
int odb_stream_close(struct odb_stream *st)
{
int r = st->close(st);
free(st);
return r;
}
ssize_t odb_read_stream_read(struct odb_read_stream *st, void *buf, size_t sz)
ssize_t odb_stream_read(struct odb_stream *st, void *buf, size_t sz)
{
return st->read(st, buf, sz);
}
struct odb_read_stream *odb_read_stream_open(struct object_database *odb,
const struct object_id *oid,
struct stream_filter *filter)
struct odb_stream *odb_stream_from_object(struct object_database *odb,
const struct object_id *oid,
struct stream_filter *filter)
{
struct odb_read_stream *st;
struct odb_stream *st;
const struct object_id *real = lookup_replace_object(odb->repo, oid);
int ret = istream_source(&st, odb, real);
@@ -221,9 +221,9 @@ struct odb_read_stream *odb_read_stream_open(struct object_database *odb,
if (filter) {
/* Add "&& !is_null_stream_filter(filter)" for performance */
struct odb_read_stream *nst = attach_stream_filter(st, filter);
struct odb_stream *nst = attach_stream_filter(st, filter);
if (!nst) {
odb_read_stream_close(st);
odb_stream_close(st);
return NULL;
}
st = nst;
@@ -232,27 +232,17 @@ struct odb_read_stream *odb_read_stream_open(struct object_database *odb,
return st;
}
ssize_t odb_write_stream_read(struct odb_write_stream *st, void *buf, size_t sz)
{
return st->read(st, buf, sz);
}
void odb_write_stream_release(struct odb_write_stream *st)
{
free(st->data);
}
int odb_stream_blob_to_fd(struct object_database *odb,
int fd,
const struct object_id *oid,
struct stream_filter *filter,
int can_seek)
{
struct odb_read_stream *st;
struct odb_stream *st;
ssize_t kept = 0;
int result = -1;
st = odb_read_stream_open(odb, oid, filter);
st = odb_stream_from_object(odb, oid, filter);
if (!st) {
if (filter)
free_stream_filter(filter);
@@ -263,7 +253,7 @@ int odb_stream_blob_to_fd(struct object_database *odb,
for (;;) {
char buf[1024 * 16];
ssize_t wrote, holeto;
ssize_t readlen = odb_read_stream_read(st, buf, sizeof(buf));
ssize_t readlen = odb_stream_read(st, buf, sizeof(buf));
if (readlen < 0)
goto close_and_exit;
@@ -294,47 +284,53 @@ int odb_stream_blob_to_fd(struct object_database *odb,
result = 0;
close_and_exit:
odb_read_stream_close(st);
odb_stream_close(st);
return result;
}
struct read_object_fd_data {
struct fd_stream {
struct odb_stream base;
int fd;
size_t remaining;
};
static ssize_t read_object_fd(struct odb_write_stream *stream,
unsigned char *buf, size_t len)
static ssize_t fd_stream_read(struct odb_stream *stream,
char *buf, size_t len)
{
struct read_object_fd_data *data = stream->data;
struct fd_stream *fds = container_of(stream, struct fd_stream, base);
ssize_t read_result;
size_t count;
if (stream->is_finished)
if (!fds->remaining)
return 0;
count = data->remaining < len ? data->remaining : len;
read_result = read_in_full(data->fd, buf, count);
count = fds->remaining < len ? fds->remaining : len;
read_result = read_in_full(fds->fd, buf, count);
if (read_result < 0 || (size_t)read_result != count)
return -1;
data->remaining -= count;
if (!data->remaining)
stream->is_finished = 1;
fds->remaining -= count;
return read_result;
}
void odb_write_stream_from_fd(struct odb_write_stream *stream, int fd,
size_t size)
static int fd_stream_close(struct odb_stream *stream UNUSED)
{
struct read_object_fd_data *data;
CALLOC_ARRAY(data, 1);
data->fd = fd;
data->remaining = size;
stream->data = data;
stream->read = read_object_fd;
stream->is_finished = 0;
/* The file descriptor is owned by the caller for now. */
return 0;
}
struct odb_stream *odb_stream_from_fd(int fd, size_t size, enum object_type type)
{
struct fd_stream *fds;
CALLOC_ARRAY(fds, 1);
fds->base.read = fd_stream_read;
fds->base.close = fd_stream_close;
fds->base.size = size;
fds->base.type = type;
fds->fd = fd;
fds->remaining = size;
return &fds->base;
}

69
odb/streaming.h
View File

@@ -8,68 +8,53 @@
#include "odb.h"
struct object_database;
struct odb_read_stream;
struct odb_stream;
struct stream_filter;
typedef int (*odb_read_stream_close_fn)(struct odb_read_stream *);
typedef ssize_t (*odb_read_stream_read_fn)(struct odb_read_stream *, char *, size_t);
typedef int (*odb_stream_close_fn)(struct odb_stream *);
typedef ssize_t (*odb_stream_read_fn)(struct odb_stream *, char *, size_t);
/*
* A stream that can be used to read an object from the object database without
* loading all of it into memory.
* A stream that can be used to read an object from or write an object into the
* object database without loading all of it into memory.
*/
struct odb_read_stream {
odb_read_stream_close_fn close;
odb_read_stream_read_fn read;
struct odb_stream {
odb_stream_close_fn close;
odb_stream_read_fn read;
enum object_type type;
size_t size; /* inflated size of full object */
};
/*
* Create a new object stream for the given object database. An optional filter
* can be used to transform the object's content.
* Create a new object stream for the given object. An optional filter can be
* used to transform the object's content.
*
* Returns the stream on success, a `NULL` pointer otherwise.
*/
struct odb_read_stream *odb_read_stream_open(struct object_database *odb,
const struct object_id *oid,
struct stream_filter *filter);
struct odb_stream *odb_stream_from_object(struct object_database *odb,
const struct object_id *oid,
struct stream_filter *filter);
/*
* Close the given read stream and release all resources associated with it.
* Create a new object stream for the given file descriptor. This can be used
* to, for example, stream an object into the object database. This function
* does _not_ take ownership of the file descriptor. It's the responsibility of
* the caller to close it after the stream has been closed.
*/
struct odb_stream *odb_stream_from_fd(int fd, size_t size, enum object_type type);
/*
* Close the given object stream and release all resources associated with it.
* Returns 0 on success, a negative error code otherwise.
*/
int odb_read_stream_close(struct odb_read_stream *stream);
int odb_stream_close(struct odb_stream *stream);
/*
* Read data from the stream into the buffer. Returns 0 on EOF and the number
* of bytes read on success. Returns a negative error code in case reading from
* the stream fails.
*/
ssize_t odb_read_stream_read(struct odb_read_stream *stream, void *buf, size_t len);
/*
* A stream that provides an object to be written to the object database without
* loading all of it into memory.
*/
struct odb_write_stream {
ssize_t (*read)(struct odb_write_stream *, unsigned char *, size_t);
void *data;
int is_finished;
};
/*
* Read data from the stream into the buffer. Returns 0 when finished and the
* number of bytes read on success. Returns a negative error code in case
* reading from the stream fails.
*/
ssize_t odb_write_stream_read(struct odb_write_stream *stream, void *buf,
size_t len);
/*
* Releases memory allocated for underlying stream data.
*/
void odb_write_stream_release(struct odb_write_stream *stream);
ssize_t odb_stream_read(struct odb_stream *stream, void *buf, size_t len);
/*
* Look up the object by its ID and write the full contents to the file
@@ -88,10 +73,4 @@ int odb_stream_blob_to_fd(struct object_database *odb,
struct stream_filter *filter,
int can_seek);
/*
* Sets up an ODB write stream that reads from an fd.
*/
void odb_write_stream_from_fd(struct odb_write_stream *stream, int fd,
size_t size);
#endif /* STREAMING_H */

6
odb/transaction.c
View File

@@ -39,10 +39,10 @@ int odb_transaction_commit(struct odb_transaction *transaction)
}
int odb_transaction_write_object_stream(struct odb_transaction *transaction,
struct odb_write_stream *stream,
size_t len, struct object_id *oid)
struct odb_stream *stream,
struct object_id *oid)
{
return transaction->write_object_stream(transaction, stream, len, oid);
return transaction->write_object_stream(transaction, stream, oid);
}
int odb_transaction_env(struct odb_transaction *transaction, struct strvec *env)

8
odb/transaction.h
View File

@@ -24,14 +24,14 @@ struct odb_transaction {
/*
* This callback is expected to write the given object stream into
* the ODB transaction. Note that for now, only blobs support streaming.
* the ODB transaction.
*
* The resulting object ID shall be written into the out pointer. The
* callback is expected to return 0 on success, a negative error code
* otherwise.
*/
int (*write_object_stream)(struct odb_transaction *transaction,
struct odb_write_stream *stream, size_t len,
struct odb_stream *stream,
struct object_id *oid);
/*
@@ -81,8 +81,8 @@ int odb_transaction_commit(struct odb_transaction *transaction);
* error code otherwise.
*/
int odb_transaction_write_object_stream(struct odb_transaction *transaction,
struct odb_write_stream *stream,
size_t len, struct object_id *oid);
struct odb_stream *stream,
struct object_id *oid);
/*
* Populates the provided strvec with the environment variables that a child

4
pack-check.c
View File

@@ -106,7 +106,7 @@ static int verify_packfile(struct repository *r,
QSORT(entries, nr_objects, compare_entries);
for (i = 0; i < nr_objects; i++) {
struct odb_read_stream *stream = NULL;
struct odb_stream *stream = NULL;
void *data;
struct object_id oid;
enum object_type type;
@@ -171,7 +171,7 @@ static int verify_packfile(struct repository *r,
display_progress(progress, base_count + i);
if (stream)
odb_read_stream_close(stream);
odb_stream_close(stream);
free(data);
}

8
packfile.c
View File

@@ -2314,7 +2314,7 @@ int parse_pack_header_option(const char *in, unsigned char *out, unsigned int *l
}
struct odb_packed_read_stream {
struct odb_read_stream base;
struct odb_stream base;
struct packed_git *pack;
git_zstream z;
enum {
@@ -2326,7 +2326,7 @@ struct odb_packed_read_stream {
off_t pos;
};
static ssize_t read_istream_pack_non_delta(struct odb_read_stream *_st, char *buf,
static ssize_t read_istream_pack_non_delta(struct odb_stream *_st, char *buf,
size_t sz)
{
struct odb_packed_read_stream *st = (struct odb_packed_read_stream *)_st;
@@ -2386,7 +2386,7 @@ static ssize_t read_istream_pack_non_delta(struct odb_read_stream *_st, char *bu
return total_read;
}
static int close_istream_pack_non_delta(struct odb_read_stream *_st)
static int close_istream_pack_non_delta(struct odb_stream *_st)
{
struct odb_packed_read_stream *st = (struct odb_packed_read_stream *)_st;
if (st->z_state == ODB_PACKED_READ_STREAM_INUSE)
@@ -2394,7 +2394,7 @@ static int close_istream_pack_non_delta(struct odb_read_stream *_st)
return 0;
}
int packfile_read_object_stream(struct odb_read_stream **out,
int packfile_read_object_stream(struct odb_stream **out,
const struct object_id *oid,
struct packed_git *pack,
off_t offset)

4
packfile.h
View File

@@ -12,7 +12,7 @@
/* in odb.h */
struct object_info;
struct odb_read_stream;
struct odb_stream;
struct packed_git {
struct pack_window *windows;
@@ -307,7 +307,7 @@ off_t get_delta_base(struct packed_git *p, struct pack_window **w_curs,
off_t *curpos, enum object_type type,
off_t delta_obj_offset);
int packfile_read_object_stream(struct odb_read_stream **out,
int packfile_read_object_stream(struct odb_stream **out,
const struct object_id *oid,
struct packed_git *pack,
off_t offset);

37
t/unit-tests/u-odb-inmemory.c
View File

@@ -100,7 +100,7 @@ void test_odb_inmemory__read_written_object(void)
void test_odb_inmemory__read_stream_object(void)
{
struct odb_source_inmemory *source = odb_source_inmemory_new(odb);
struct odb_read_stream *stream;
struct odb_stream *stream;
struct object_id written_oid;
const char data[] = "foobar";
char buf[3] = { 0 };
@@ -112,15 +112,15 @@ void test_odb_inmemory__read_stream_object(void)
cl_assert_equal_i(stream->type, OBJ_BLOB);
cl_assert_equal_u(stream->size, 6);
cl_assert_equal_i(odb_read_stream_read(stream, buf, 2), 2);
cl_assert_equal_i(odb_stream_read(stream, buf, 2), 2);
cl_assert_equal_s(buf, "fo");
cl_assert_equal_i(odb_read_stream_read(stream, buf, 2), 2);
cl_assert_equal_i(odb_stream_read(stream, buf, 2), 2);
cl_assert_equal_s(buf, "ob");
cl_assert_equal_i(odb_read_stream_read(stream, buf, 2), 2);
cl_assert_equal_i(odb_stream_read(stream, buf, 2), 2);
cl_assert_equal_s(buf, "ar");
cl_assert_equal_i(odb_read_stream_read(stream, buf, 2), 0);
cl_assert_equal_i(odb_stream_read(stream, buf, 2), 0);
odb_read_stream_close(stream);
odb_stream_close(stream);
odb_source_free(&source->base);
}
@@ -266,28 +266,28 @@ void test_odb_inmemory__freshen_object(void)
}
struct membuf_write_stream {
struct odb_write_stream base;
struct odb_stream base;
const char *buf;
size_t offset;
size_t size;
};
static ssize_t membuf_write_stream_read(struct odb_write_stream *stream,
unsigned char *buf, size_t len)
static ssize_t membuf_write_stream_read(struct odb_stream *stream,
char *buf, size_t len)
{
struct membuf_write_stream *s = container_of(stream, struct membuf_write_stream, base);
size_t chunk_size = 2;
if (s->offset == s->base.size)
return 0;
if (chunk_size > len)
chunk_size = len;
if (chunk_size > s->size - s->offset)
chunk_size = s->size - s->offset;
if (chunk_size > s->base.size - s->offset)
chunk_size = s->base.size - s->offset;
memcpy(buf, s->buf + s->offset, chunk_size);
s->offset += chunk_size;
if (s->offset == s->size)
s->base.is_finished = 1;
return chunk_size;
}
@@ -297,14 +297,17 @@ void test_odb_inmemory__write_object_stream(void)
struct odb_source_inmemory *source = odb_source_inmemory_new(odb);
const char data[] = "foobar";
struct membuf_write_stream stream = {
.base.read = membuf_write_stream_read,
.base = {
.read = membuf_write_stream_read,
.size = strlen(data),
.type = OBJ_BLOB,
},
.buf = data,
.size = strlen(data),
};
struct object_id written_oid;
cl_must_pass(odb_source_write_object_stream(&source->base, &stream.base,
strlen(data), &written_oid));
&written_oid));
cl_assert_equal_s(oid_to_hex(&written_oid), FOOBAR_OID);
cl_assert_object_info(source, &written_oid, OBJ_BLOB, "foobar");