Re: [PATCH v13 5/7] object-file.c: add "stream_loose_object()" to handle large object

[Neeraj Singh <nksingh85@gmail.com>]

On 6/4/2022 3:10 AM, Ævar Arnfjörð Bjarmason wrote:
> From: Han Xin <hanxin.hx@alibaba-inc.com>
> 
> If we want unpack and write a loose object using "write_loose_object",
> we have to feed it with a buffer with the same size of the object, which
> will consume lots of memory and may cause OOM. This can be improved by
> feeding data to "stream_loose_object()" in a stream.
> 
> Add a new function "stream_loose_object()", which is a stream version of
> "write_loose_object()" but with a low memory footprint. We will use this
> function to unpack large blob object in later commit.
> 
> Another difference with "write_loose_object()" is that we have no chance
> to run "write_object_file_prepare()" to calculate the oid in advance.
> In "write_loose_object()", we know the oid and we can write the
> temporary file in the same directory as the final object, but for an
> object with an undetermined oid, we don't know the exact directory for
> the object.
> 
> Still, we need to save the temporary file we're preparing
> somewhere. We'll do that in the top-level ".git/objects/"
> directory (or whatever "GIT_OBJECT_DIRECTORY" is set to). Once we've
> streamed it we'll know the OID, and will move it to its canonical
> path.
> 

I think this new logic doesn't play well with batched-fsync. Even 
through we don't know the final OID, we should still call 
prepare_loose_object_bulk_checkin to potentially create the bulk checkin 
objdir.


> diff --git a/object-file.c b/object-file.c
> index 7946fa5e088..9fd449693c4 100644
> --- a/object-file.c
> +++ b/object-file.c
> @@ -2119,6 +2119,106 @@ static int freshen_packed_object(const struct object_id *oid)
>   	return 1;
>   }
>   
> +int stream_loose_object(struct input_stream *in_stream, size_t len,
> +			struct object_id *oid)
> +{
> +	int fd, ret, err = 0, flush = 0;
> +	unsigned char compressed[4096];
> +	git_zstream stream;
> +	git_hash_ctx c;
> +	struct strbuf tmp_file = STRBUF_INIT;
> +	struct strbuf filename = STRBUF_INIT;
> +	int dirlen;
> +	char hdr[MAX_HEADER_LEN];
> +	int hdrlen;
> +
> +	/* Since oid is not determined, save tmp file to odb path. */
> +	strbuf_addf(&filename, "%s/", get_object_directory());
> +	hdrlen = format_object_header(hdr, sizeof(hdr), OBJ_BLOB, len);
> +
> +	/*
> +	 * Common steps for write_loose_object and stream_loose_object to
> +	 * start writing loose objects:
> +	 *
> +	 *  - Create tmpfile for the loose object.
> +	 *  - Setup zlib stream for compression.
> +	 *  - Start to feed header to zlib stream.
> +	 */
> +	fd = start_loose_object_common(&tmp_file, filename.buf, 0,
> +				       &stream, compressed, sizeof(compressed),
> +				       &c, hdr, hdrlen);
> +	if (fd < 0) {
> +		err = -1;
> +		goto cleanup;
> +	}
> +
> +	/* Then the data itself.. */
> +	do {
> +		unsigned char *in0 = stream.next_in;
> +
> +		if (!stream.avail_in && !in_stream->is_finished) {
> +			const void *in = in_stream->read(in_stream, &stream.avail_in);
> +			stream.next_in = (void *)in;
> +			in0 = (unsigned char *)in;
> +			/* All data has been read. */
> +			if (in_stream->is_finished)
> +				flush = 1;
> +		}
> +		ret = write_loose_object_common(&c, &stream, flush, in0, fd,
> +						compressed, sizeof(compressed));
> +		/*
> +		 * Unlike write_loose_object(), we do not have the entire
> +		 * buffer. If we get Z_BUF_ERROR due to too few input bytes,
> +		 * then we'll replenish them in the next input_stream->read()
> +		 * call when we loop.
> +		 */
> +	} while (ret == Z_OK || ret == Z_BUF_ERROR);
> +
> +	if (stream.total_in != len + hdrlen)
> +		die(_("write stream object %ld != %"PRIuMAX), stream.total_in,
> +		    (uintmax_t)len + hdrlen);
> +
> +	/* Common steps for write_loose_object and stream_loose_object to
> +	 * end writing loose oject:
> +	 *
> +	 *  - End the compression of zlib stream.
> +	 *  - Get the calculated oid.
> +	 */
> +	if (ret != Z_STREAM_END)
> +		die(_("unable to stream deflate new object (%d)"), ret);
> +	ret = end_loose_object_common(&c, &stream, oid);
> +	if (ret != Z_OK)
> +		die(_("deflateEnd on stream object failed (%d)"), ret);
> +	close_loose_object(fd, tmp_file.buf);
> +

If batch fsync is enabled, the close_loose_object call will refrain from 
syncing the tmp file.

> +	if (freshen_packed_object(oid) || freshen_loose_object(oid)) {
> +		unlink_or_warn(tmp_file.buf);
> +		goto cleanup;
> +	}
> +
> +	loose_object_path(the_repository, &filename, oid);
> +

We expect this loose_object_path call to return a path in the bulk fsync 
object directory. It might not do so if we don't call 
prepare_loose_object_bulk_checkin.

In the new test case introduced in (7/7), we seem to be getting lucky
in that there are some small objects (commits) earlier in the packfile,
so we go through write_loose_object first.

Thanks for including me on the review!

-Neeraj