2 * Copyright (c) 2011, Google Inc.
5 #include "bulk-checkin.h"
7 #include "repository.h"
11 #include "string-list.h"
12 #include "tmp-objdir.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
{
24 struct pack_idx_option pack_idx_opts
;
26 struct pack_idx_entry
**written
;
27 uint32_t alloc_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
,
35 struct pack_idx_option
*pack_idx_opts
,
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
);
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
;
56 if (state
->nr_written
== 0) {
58 unlink(state
->pack_tmp_name
);
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
);
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
,
71 strbuf_addf(&packname
, "%s/pack/pack-%s.", get_object_directory(),
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
]);
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
)
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
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
)
126 /* The object may already exist in the repository */
127 if (has_object_file(oid
))
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
))
135 /* This is a new object we need to keep */
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
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
)
160 unsigned char ibuf
[16384];
161 unsigned char obuf
[16384];
164 int write_object
= (flags
& HASH_WRITE_OBJECT
);
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
);
178 die_errno("failed to read from '%s'", path
);
179 if (read_result
!= rsize
)
180 die("failed to read %d bytes from '%s'",
183 if (*already_hashed_to
< offset
) {
184 size_t hsize
= offset
- *already_hashed_to
;
188 the_hash_algo
->update_fn(ctx
, ibuf
, hsize
);
189 *already_hashed_to
= offset
;
196 status
= git_deflate(&s
, size
? 0 : Z_FINISH
);
198 if (!s
.avail_out
|| status
== Z_STREAM_END
) {
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
);
210 hashwrite(state
->f
, obuf
, written
);
211 state
->offset
+= written
;
214 s
.avail_out
= sizeof(obuf
);
223 die("unexpected deflate failure: %d", status
);
230 /* Lazily create backing packfile for the state */
231 static void prepare_to_stream(struct bulk_checkin_packfile
*state
,
234 if (!(flags
& HASH_WRITE_OBJECT
) || state
->f
)
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);
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
,
249 enum object_type type
, const char *path
,
252 off_t seekback
, already_hashed_to
;
254 unsigned char obuf
[16384];
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
),
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;
275 prepare_to_stream(state
, flags
);
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
))
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.
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
);
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
;
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
;
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
)
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
341 if (!bulk_fsync_objdir
||
342 git_fsync(fd
, FSYNC_WRITEOUT_ONLY
) < 0) {
343 fsync_or_die(fd
, filename
);
347 int index_bulk_checkin(struct object_id
*oid
,
348 int fd
, size_t size
, enum object_type type
,
349 const char *path
, unsigned flags
)
351 int status
= deflate_to_pack(&bulk_checkin_packfile
, oid
, fd
, size
, type
,
353 if (!odb_transaction_nesting
)
354 flush_bulk_checkin_packfile(&bulk_checkin_packfile
);
358 void begin_odb_transaction(void)
360 odb_transaction_nesting
+= 1;
363 void flush_odb_transaction(void)
366 flush_bulk_checkin_packfile(&bulk_checkin_packfile
);
369 void end_odb_transaction(void)
371 odb_transaction_nesting
-= 1;
372 if (odb_transaction_nesting
< 0)
373 BUG("Unbalanced ODB transaction nesting");
375 if (odb_transaction_nesting
)
378 flush_odb_transaction();