odb/streaming: drop `is_finished` field

The `is_finished` field is used to track whether a write stream is done
writing all of its data. Tracking this field as part of the stream
itself shouldn't be required though: callers will already know when the
stream is done when the stream's read function returns zero bytes, same
as when reading from a file descriptor.

There is one exception where it gets a bit more complicated: when
consuming data in "builtin/unpack-objects.c" it may happen that we don't
yield any new bytes after reading from the pipe. This is addressed by
looping until we have produced at least a single byte of output.

Drop the field from `struct odb_write_stream`. Again, same as in the
preceding commit, this brings the structure a bit closer to its sibling
`struct odb_read_stream`.

Signed-off-by: Patrick Steinhardt <ps@pks.im>
Signed-off-by: Junio C Hamano <gitster@pobox.com>

builtin/unpack-objects.c

@@ -368,20 +368,20 @@ static ssize_t feed_input_zstream(struct odb_write_stream *in_stream,
 {
 	struct input_zstream_data *data = in_stream->data;
 	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->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;
 }
 
@@ -397,6 +397,7 @@ static void stream_blob(unsigned long size, unsigned nr)
 	struct obj_info *info = &obj_list[nr];
 
 	data.zstream = &zstream;
+	data.status = Z_OK;
 	git_inflate_init(&zstream);
 
 	if (odb_write_object_stream(the_repository->objects, &in_stream, &info->oid))

object-file.c

@@ -716,12 +716,13 @@ static int hash_blob_stream(struct odb_write_stream *stream,
 	git_hash_init(&ctx, hash_algo);
 	git_hash_update(&ctx, buf, header_len);
 
-	while (!stream->is_finished) {
+	while (1) {
 		ssize_t read_result = odb_write_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;
@@ -749,6 +750,7 @@ 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);
@@ -758,12 +760,13 @@ static void stream_blob_to_pack(struct transaction_packfile *state,
 	s.avail_out = sizeof(obuf) - hdrlen;
 
 	while (status != Z_STREAM_END) {
-		if (!stream->is_finished && !s.avail_in) {
+		if (!is_finished && !s.avail_in) {
 			ssize_t rsize = odb_write_stream_read(stream, ibuf,
 							      sizeof(ibuf));
-
 			if (rsize < 0)
 				die("failed to read blob 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;

odb/source-inmemory.c

@@ -265,10 +265,17 @@ static int odb_source_inmemory_write_object_stream(struct odb_source *source,
 	int ret;
 
 	CALLOC_ARRAY(data, stream->size);
-	while (!stream->is_finished) {
+	while (1) {
 		ssize_t bytes_read;
 
 		bytes_read = odb_write_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;

odb/source-loose.c

@@ -859,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;
 
@@ -889,7 +890,7 @@ 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) {
+		if (!stream.avail_in && !is_finished) {
 			ssize_t read_len = odb_write_stream_read(in_stream, buf,
 								 sizeof(buf));
 			if (read_len < 0) {
@@ -898,12 +899,15 @@ static int odb_source_loose_write_object_stream(struct odb_source *source,
 				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));

odb/streaming.c

@@ -310,7 +310,7 @@ static ssize_t read_object_fd(struct odb_write_stream *stream,
 	ssize_t read_result;
 	size_t count;
 
-	if (stream->is_finished)
+	if (!data->remaining)
 		return 0;
 
 	count = data->remaining < len ? data->remaining : len;
@@ -319,8 +319,6 @@ static ssize_t read_object_fd(struct odb_write_stream *stream,
 		return -1;
 
 	data->remaining -= count;
-	if (!data->remaining)
-		stream->is_finished = 1;
 
 	return read_result;
 }
@@ -337,5 +335,4 @@ void odb_write_stream_from_fd(struct odb_write_stream *stream, int fd,
 	stream->data = data;
 	stream->read = read_object_fd;
 	stream->size = size;
-	stream->is_finished = 0;
 }

odb/streaming.h

@@ -56,7 +56,6 @@ struct odb_write_stream {
 	ssize_t (*read)(struct odb_write_stream *, unsigned char *, size_t);
 	void *data;
 	size_t size;
-	int is_finished;
 };
 
 /*

t/unit-tests/u-odb-inmemory.c

@@ -277,6 +277,9 @@ static ssize_t membuf_write_stream_read(struct odb_write_stream *stream,
 	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->base.size - s->offset)
@@ -285,8 +288,6 @@ static ssize_t membuf_write_stream_read(struct odb_write_stream *stream,
 	memcpy(buf, s->buf + s->offset, chunk_size);
 
 	s->offset += chunk_size;
-	if (s->offset == s->base.size)
-		s->base.is_finished = 1;
 
 	return chunk_size;
 }