Sync with 2.38.5
[git.git] / bulk-checkin.c
blob855b68ec23bdb1bdc85c1184a9c144c226f0245c
1 /*
2 * Copyright (c) 2011, Google Inc.
3 */
4 #include "cache.h"
5 #include "bulk-checkin.h"
6 #include "lockfile.h"
7 #include "repository.h"
8 #include "csum-file.h"
9 #include "pack.h"
10 #include "strbuf.h"
11 #include "string-list.h"
12 #include "tmp-objdir.h"
13 #include "packfile.h"
14 #include "object-store.h"
16 static int odb_transaction_nesting;
18 static struct tmp_objdir *bulk_fsync_objdir;
20 static struct bulk_checkin_packfile {
21 char *pack_tmp_name;
22 struct hashfile *f;
23 off_t offset;
24 struct pack_idx_option pack_idx_opts;
26 struct pack_idx_entry **written;
27 uint32_t alloc_written;
28 uint32_t nr_written;
29 } bulk_checkin_packfile;
31 static void finish_tmp_packfile(struct strbuf *basename,
32 const char *pack_tmp_name,
33 struct pack_idx_entry **written_list,
34 uint32_t nr_written,
35 struct pack_idx_option *pack_idx_opts,
36 unsigned char hash[])
38 char *idx_tmp_name = NULL;
40 stage_tmp_packfiles(basename, pack_tmp_name, written_list, nr_written,
41 NULL, pack_idx_opts, hash, &idx_tmp_name);
42 rename_tmp_packfile_idx(basename, &idx_tmp_name);
44 free(idx_tmp_name);
47 static void flush_bulk_checkin_packfile(struct bulk_checkin_packfile *state)
49 unsigned char hash[GIT_MAX_RAWSZ];
50 struct strbuf packname = STRBUF_INIT;
51 int i;
53 if (!state->f)
54 return;
56 if (state->nr_written == 0) {
57 close(state->f->fd);
58 unlink(state->pack_tmp_name);
59 goto clear_exit;
60 } else if (state->nr_written == 1) {
61 finalize_hashfile(state->f, hash, FSYNC_COMPONENT_PACK,
62 CSUM_HASH_IN_STREAM | CSUM_FSYNC | CSUM_CLOSE);
63 } else {
64 int fd = finalize_hashfile(state->f, hash, FSYNC_COMPONENT_PACK, 0);
65 fixup_pack_header_footer(fd, hash, state->pack_tmp_name,
66 state->nr_written, hash,
67 state->offset);
68 close(fd);
71 strbuf_addf(&packname, "%s/pack/pack-%s.", get_object_directory(),
72 hash_to_hex(hash));
73 finish_tmp_packfile(&packname, state->pack_tmp_name,
74 state->written, state->nr_written,
75 &state->pack_idx_opts, hash);
76 for (i = 0; i < state->nr_written; i++)
77 free(state->written[i]);
79 clear_exit:
80 free(state->written);
81 memset(state, 0, sizeof(*state));
83 strbuf_release(&packname);
84 /* Make objects we just wrote available to ourselves */
85 reprepare_packed_git(the_repository);
89 * Cleanup after batch-mode fsync_object_files.
91 static void flush_batch_fsync(void)
93 struct strbuf temp_path = STRBUF_INIT;
94 struct tempfile *temp;
96 if (!bulk_fsync_objdir)
97 return;
100 * Issue a full hardware flush against a temporary file to ensure
101 * that all objects are durable before any renames occur. The code in
102 * fsync_loose_object_bulk_checkin has already issued a writeout
103 * request, but it has not flushed any writeback cache in the storage
104 * hardware or any filesystem logs. This fsync call acts as a barrier
105 * to ensure that the data in each new object file is durable before
106 * the final name is visible.
108 strbuf_addf(&temp_path, "%s/bulk_fsync_XXXXXX", get_object_directory());
109 temp = xmks_tempfile(temp_path.buf);
110 fsync_or_die(get_tempfile_fd(temp), get_tempfile_path(temp));
111 delete_tempfile(&temp);
112 strbuf_release(&temp_path);
115 * Make the object files visible in the primary ODB after their data is
116 * fully durable.
118 tmp_objdir_migrate(bulk_fsync_objdir);
119 bulk_fsync_objdir = NULL;
122 static int already_written(struct bulk_checkin_packfile *state, struct object_id *oid)
124 int i;
126 /* The object may already exist in the repository */
127 if (has_object_file(oid))
128 return 1;
130 /* Might want to keep the list sorted */
131 for (i = 0; i < state->nr_written; i++)
132 if (oideq(&state->written[i]->oid, oid))
133 return 1;
135 /* This is a new object we need to keep */
136 return 0;
140 * Read the contents from fd for size bytes, streaming it to the
141 * packfile in state while updating the hash in ctx. Signal a failure
142 * by returning a negative value when the resulting pack would exceed
143 * the pack size limit and this is not the first object in the pack,
144 * so that the caller can discard what we wrote from the current pack
145 * by truncating it and opening a new one. The caller will then call
146 * us again after rewinding the input fd.
148 * The already_hashed_to pointer is kept untouched by the caller to
149 * make sure we do not hash the same byte when we are called
150 * again. This way, the caller does not have to checkpoint its hash
151 * status before calling us just in case we ask it to call us again
152 * with a new pack.
154 static int stream_to_pack(struct bulk_checkin_packfile *state,
155 git_hash_ctx *ctx, off_t *already_hashed_to,
156 int fd, size_t size, enum object_type type,
157 const char *path, unsigned flags)
159 git_zstream s;
160 unsigned char ibuf[16384];
161 unsigned char obuf[16384];
162 unsigned hdrlen;
163 int status = Z_OK;
164 int write_object = (flags & HASH_WRITE_OBJECT);
165 off_t offset = 0;
167 git_deflate_init(&s, pack_compression_level);
169 hdrlen = encode_in_pack_object_header(obuf, sizeof(obuf), type, size);
170 s.next_out = obuf + hdrlen;
171 s.avail_out = sizeof(obuf) - hdrlen;
173 while (status != Z_STREAM_END) {
174 if (size && !s.avail_in) {
175 ssize_t rsize = size < sizeof(ibuf) ? size : sizeof(ibuf);
176 ssize_t read_result = read_in_full(fd, ibuf, rsize);
177 if (read_result < 0)
178 die_errno("failed to read from '%s'", path);
179 if (read_result != rsize)
180 die("failed to read %d bytes from '%s'",
181 (int)rsize, path);
182 offset += rsize;
183 if (*already_hashed_to < offset) {
184 size_t hsize = offset - *already_hashed_to;
185 if (rsize < hsize)
186 hsize = rsize;
187 if (hsize)
188 the_hash_algo->update_fn(ctx, ibuf, hsize);
189 *already_hashed_to = offset;
191 s.next_in = ibuf;
192 s.avail_in = rsize;
193 size -= rsize;
196 status = git_deflate(&s, size ? 0 : Z_FINISH);
198 if (!s.avail_out || status == Z_STREAM_END) {
199 if (write_object) {
200 size_t written = s.next_out - obuf;
202 /* would we bust the size limit? */
203 if (state->nr_written &&
204 pack_size_limit_cfg &&
205 pack_size_limit_cfg < state->offset + written) {
206 git_deflate_abort(&s);
207 return -1;
210 hashwrite(state->f, obuf, written);
211 state->offset += written;
213 s.next_out = obuf;
214 s.avail_out = sizeof(obuf);
217 switch (status) {
218 case Z_OK:
219 case Z_BUF_ERROR:
220 case Z_STREAM_END:
221 continue;
222 default:
223 die("unexpected deflate failure: %d", status);
226 git_deflate_end(&s);
227 return 0;
230 /* Lazily create backing packfile for the state */
231 static void prepare_to_stream(struct bulk_checkin_packfile *state,
232 unsigned flags)
234 if (!(flags & HASH_WRITE_OBJECT) || state->f)
235 return;
237 state->f = create_tmp_packfile(&state->pack_tmp_name);
238 reset_pack_idx_option(&state->pack_idx_opts);
240 /* Pretend we are going to write only one object */
241 state->offset = write_pack_header(state->f, 1);
242 if (!state->offset)
243 die_errno("unable to write pack header");
246 static int deflate_to_pack(struct bulk_checkin_packfile *state,
247 struct object_id *result_oid,
248 int fd, size_t size,
249 enum object_type type, const char *path,
250 unsigned flags)
252 off_t seekback, already_hashed_to;
253 git_hash_ctx ctx;
254 unsigned char obuf[16384];
255 unsigned header_len;
256 struct hashfile_checkpoint checkpoint = {0};
257 struct pack_idx_entry *idx = NULL;
259 seekback = lseek(fd, 0, SEEK_CUR);
260 if (seekback == (off_t) -1)
261 return error("cannot find the current offset");
263 header_len = format_object_header((char *)obuf, sizeof(obuf),
264 type, size);
265 the_hash_algo->init_fn(&ctx);
266 the_hash_algo->update_fn(&ctx, obuf, header_len);
268 /* Note: idx is non-NULL when we are writing */
269 if ((flags & HASH_WRITE_OBJECT) != 0)
270 CALLOC_ARRAY(idx, 1);
272 already_hashed_to = 0;
274 while (1) {
275 prepare_to_stream(state, flags);
276 if (idx) {
277 hashfile_checkpoint(state->f, &checkpoint);
278 idx->offset = state->offset;
279 crc32_begin(state->f);
281 if (!stream_to_pack(state, &ctx, &already_hashed_to,
282 fd, size, type, path, flags))
283 break;
285 * Writing this object to the current pack will make
286 * it too big; we need to truncate it, start a new
287 * pack, and write into it.
289 if (!idx)
290 BUG("should not happen");
291 hashfile_truncate(state->f, &checkpoint);
292 state->offset = checkpoint.offset;
293 flush_bulk_checkin_packfile(state);
294 if (lseek(fd, seekback, SEEK_SET) == (off_t) -1)
295 return error("cannot seek back");
297 the_hash_algo->final_oid_fn(result_oid, &ctx);
298 if (!idx)
299 return 0;
301 idx->crc32 = crc32_end(state->f);
302 if (already_written(state, result_oid)) {
303 hashfile_truncate(state->f, &checkpoint);
304 state->offset = checkpoint.offset;
305 free(idx);
306 } else {
307 oidcpy(&idx->oid, result_oid);
308 ALLOC_GROW(state->written,
309 state->nr_written + 1,
310 state->alloc_written);
311 state->written[state->nr_written++] = idx;
313 return 0;
316 void prepare_loose_object_bulk_checkin(void)
319 * We lazily create the temporary object directory
320 * the first time an object might be added, since
321 * callers may not know whether any objects will be
322 * added at the time they call begin_odb_transaction.
324 if (!odb_transaction_nesting || bulk_fsync_objdir)
325 return;
327 bulk_fsync_objdir = tmp_objdir_create("bulk-fsync");
328 if (bulk_fsync_objdir)
329 tmp_objdir_replace_primary_odb(bulk_fsync_objdir, 0);
332 void fsync_loose_object_bulk_checkin(int fd, const char *filename)
335 * If we have an active ODB transaction, we issue a call that
336 * cleans the filesystem page cache but avoids a hardware flush
337 * command. Later on we will issue a single hardware flush
338 * before renaming the objects to their final names as part of
339 * flush_batch_fsync.
341 if (!bulk_fsync_objdir ||
342 git_fsync(fd, FSYNC_WRITEOUT_ONLY) < 0) {
343 if (errno == ENOSYS)
344 warning(_("core.fsyncMethod = batch is unsupported on this platform"));
345 fsync_or_die(fd, filename);
349 int index_bulk_checkin(struct object_id *oid,
350 int fd, size_t size, enum object_type type,
351 const char *path, unsigned flags)
353 int status = deflate_to_pack(&bulk_checkin_packfile, oid, fd, size, type,
354 path, flags);
355 if (!odb_transaction_nesting)
356 flush_bulk_checkin_packfile(&bulk_checkin_packfile);
357 return status;
360 void begin_odb_transaction(void)
362 odb_transaction_nesting += 1;
365 void flush_odb_transaction(void)
367 flush_batch_fsync();
368 flush_bulk_checkin_packfile(&bulk_checkin_packfile);
371 void end_odb_transaction(void)
373 odb_transaction_nesting -= 1;
374 if (odb_transaction_nesting < 0)
375 BUG("Unbalanced ODB transaction nesting");
377 if (odb_transaction_nesting)
378 return;
380 flush_odb_transaction();