1
#include "cache.h"
2
#include "entry.h"
3
#include "parallel-checkout.h"
4
+#include "pkt-line.h"
5
+#include "run-command.h"
6
+#include "sigchain.h"
7
#include "streaming.h"
8
6
-enum pc_item_status {
7
- PC_ITEM_PENDING = 0,
8
- PC_ITEM_WRITTEN,
9
- /*
10
- * The entry could not be written because there was another file
11
- * already present in its path or leading directories. Since
12
- * checkout_entry_ca() removes such files from the working tree before
13
- * enqueueing the entry for parallel checkout, it means that there was
14
- * a path collision among the entries being written.
15
- */
16
- PC_ITEM_COLLIDED,
17
- PC_ITEM_FAILED,
18
-};
19
-
20
-struct parallel_checkout_item {
21
- /* pointer to a istate->cache[] entry. Not owned by us. */
22
- struct cache_entry *ce;
23
- struct conv_attrs ca;
24
- struct stat st;
25
- enum pc_item_status status;
9
+struct pc_worker {
10
+ struct child_process cp;
11
+ size_t next_item_to_complete, nr_items_to_complete;
12
};
13
14
struct parallel_checkout {
45
const struct conv_attrs *ca)
46
{
47
enum conv_attrs_classification c;
48
+ size_t packed_item_size;
49
50
/*
51
* Symlinks cannot be checked out in parallel as, in case of path
56
if (!S_ISREG(ce->ce_mode))
57
return 0;
58
59
+ packed_item_size = sizeof(struct pc_item_fixed_portion) + ce->ce_namelen +
60
+ (ca->working_tree_encoding ? strlen(ca->working_tree_encoding) : 0);
61
+
62
+ /*
63
+ * The amount of data we send to the workers per checkout item is
64
+ * typically small (75~300B). So unless we find an insanely huge path
65
+ * of 64KB, we should never reach the 65KB limit of one pkt-line. If
66
+ * that does happen, we let the sequential code handle the item.
67
+ */
68
+ if (packed_item_size > LARGE_PACKET_DATA_MAX)
69
+ return 0;
70
+
71
c = classify_conv_attrs(ca);
72
switch (c) {
73
case CA_CLASS_INCORE:
120
ALLOC_GROW(parallel_checkout.items, parallel_checkout.nr + 1,
121
parallel_checkout.alloc);
122
124
- pc_item = ¶llel_checkout.items[parallel_checkout.nr++];
123
+ pc_item = ¶llel_checkout.items[parallel_checkout.nr];
124
pc_item->ce = ce;
125
memcpy(&pc_item->ca, ca, sizeof(pc_item->ca));
126
pc_item->status = PC_ITEM_PENDING;
127
+ pc_item->id = parallel_checkout.nr;
128
+ parallel_checkout.nr++;
129
130
return 0;
131
}
237
/*
238
* checkout metadata is used to give context for external process
239
* filters. Files requiring such filters are not eligible for parallel
239
- * checkout, so pass NULL.
240
+ * checkout, so pass NULL. Note: if that changes, the metadata must also
241
+ * be passed from the main process to the workers.
242
*/
243
ret = convert_to_working_tree_ca(&pc_item->ca, pc_item->ce->name,
244
blob, size, &buf, NULL);
270
return ret;
271
}
272
271
-static void write_pc_item(struct parallel_checkout_item *pc_item,
272
- struct checkout *state)
273
+void write_pc_item(struct parallel_checkout_item *pc_item,
274
+ struct checkout *state)
275
{
276
unsigned int mode = (pc_item->ce->ce_mode & 0100) ? 0777 : 0666;
277
int fd = -1, fstat_done = 0;
342
strbuf_release(&path);
343
}
344
345
+static void send_one_item(int fd, struct parallel_checkout_item *pc_item)
346
+{
347
+ size_t len_data;
348
+ char *data, *variant;
349
+ struct pc_item_fixed_portion *fixed_portion;
350
+ const char *working_tree_encoding = pc_item->ca.working_tree_encoding;
351
+ size_t name_len = pc_item->ce->ce_namelen;
352
+ size_t working_tree_encoding_len = working_tree_encoding ?
353
+ strlen(working_tree_encoding) : 0;
354
+
355
+ /*
356
+ * Any changes in the calculation of the message size must also be made
357
+ * in is_eligible_for_parallel_checkout().
358
+ */
359
+ len_data = sizeof(struct pc_item_fixed_portion) + name_len +
360
+ working_tree_encoding_len;
361
+
362
+ data = xcalloc(1, len_data);
363
+
364
+ fixed_portion = (struct pc_item_fixed_portion *)data;
365
+ fixed_portion->id = pc_item->id;
366
+ fixed_portion->ce_mode = pc_item->ce->ce_mode;
367
+ fixed_portion->crlf_action = pc_item->ca.crlf_action;
368
+ fixed_portion->ident = pc_item->ca.ident;
369
+ fixed_portion->name_len = name_len;
370
+ fixed_portion->working_tree_encoding_len = working_tree_encoding_len;
371
+ /*
372
+ * We use hashcpy() instead of oidcpy() because the hash[] positions
373
+ * after `the_hash_algo->rawsz` might not be initialized. And Valgrind
374
+ * would complain about passing uninitialized bytes to a syscall
375
+ * (write(2)). There is no real harm in this case, but the warning could
376
+ * hinder the detection of actual errors.
377
+ */
378
+ hashcpy(fixed_portion->oid.hash, pc_item->ce->oid.hash);
379
+
380
+ variant = data + sizeof(*fixed_portion);
381
+ if (working_tree_encoding_len) {
382
+ memcpy(variant, working_tree_encoding, working_tree_encoding_len);
383
+ variant += working_tree_encoding_len;
384
+ }
385
+ memcpy(variant, pc_item->ce->name, name_len);
386
+
387
+ packet_write(fd, data, len_data);
388
+
389
+ free(data);
390
+}
391
+
392
+static void send_batch(int fd, size_t start, size_t nr)
393
+{
394
+ size_t i;
395
+ sigchain_push(SIGPIPE, SIG_IGN);
396
+ for (i = 0; i < nr; i++)
397
+ send_one_item(fd, ¶llel_checkout.items[start + i]);
398
+ packet_flush(fd);
399
+ sigchain_pop(SIGPIPE);
400
+}
401
+
402
+static struct pc_worker *setup_workers(struct checkout *state, int num_workers)
403
+{
404
+ struct pc_worker *workers;
405
+ int i, workers_with_one_extra_item;
406
+ size_t base_batch_size, batch_beginning = 0;
407
+
408
+ ALLOC_ARRAY(workers, num_workers);
409
+
410
+ for (i = 0; i < num_workers; i++) {
411
+ struct child_process *cp = &workers[i].cp;
412
+
413
+ child_process_init(cp);
414
+ cp->git_cmd = 1;
415
+ cp->in = -1;
416
+ cp->out = -1;
417
+ cp->clean_on_exit = 1;
418
+ strvec_push(&cp->args, "checkout--worker");
419
+ if (state->base_dir_len)
420
+ strvec_pushf(&cp->args, "--prefix=%s", state->base_dir);
421
+ if (start_command(cp))
422
+ die("failed to spawn checkout worker");
423
+ }
424
+
425
+ base_batch_size = parallel_checkout.nr / num_workers;
426
+ workers_with_one_extra_item = parallel_checkout.nr % num_workers;
427
+
428
+ for (i = 0; i < num_workers; i++) {
429
+ struct pc_worker *worker = &workers[i];
430
+ size_t batch_size = base_batch_size;
431
+
432
+ /* distribute the extra work evenly */
433
+ if (i < workers_with_one_extra_item)
434
+ batch_size++;
435
+
436
+ send_batch(worker->cp.in, batch_beginning, batch_size);
437
+ worker->next_item_to_complete = batch_beginning;
438
+ worker->nr_items_to_complete = batch_size;
439
+
440
+ batch_beginning += batch_size;
441
+ }
442
+
443
+ return workers;
444
+}
445
+
446
+static void finish_workers(struct pc_worker *workers, int num_workers)
447
+{
448
+ int i;
449
+
450
+ /*
451
+ * Close pipes before calling finish_command() to let the workers
452
+ * exit asynchronously and avoid spending extra time on wait().
453
+ */
454
+ for (i = 0; i < num_workers; i++) {
455
+ struct child_process *cp = &workers[i].cp;
456
+ if (cp->in >= 0)
457
+ close(cp->in);
458
+ if (cp->out >= 0)
459
+ close(cp->out);
460
+ }
461
+
462
+ for (i = 0; i < num_workers; i++) {
463
+ int rc = finish_command(&workers[i].cp);
464
+ if (rc > 128) {
465
+ /*
466
+ * For a normal non-zero exit, the worker should have
467
+ * already printed something useful to stderr. But a
468
+ * death by signal should be mentioned to the user.
469
+ */
470
+ error("checkout worker %d died of signal %d", i, rc - 128);
471
+ }
472
+ }
473
+
474
+ free(workers);
475
+}
476
+
477
+static inline void assert_pc_item_result_size(int got, int exp)
478
+{
479
+ if (got != exp)
480
+ BUG("wrong result size from checkout worker (got %dB, exp %dB)",
481
+ got, exp);
482
+}
483
+
484
+static void parse_and_save_result(const char *buffer, int len,
485
+ struct pc_worker *worker)
486
+{
487
+ struct pc_item_result *res;
488
+ struct parallel_checkout_item *pc_item;
489
+ struct stat *st = NULL;
490
+
491
+ if (len < PC_ITEM_RESULT_BASE_SIZE)
492
+ BUG("too short result from checkout worker (got %dB, exp >=%dB)",
493
+ len, (int)PC_ITEM_RESULT_BASE_SIZE);
494
+
495
+ res = (struct pc_item_result *)buffer;
496
+
497
+ /*
498
+ * Worker should send either the full result struct on success, or
499
+ * just the base (i.e. no stat data), otherwise.
500
+ */
501
+ if (res->status == PC_ITEM_WRITTEN) {
502
+ assert_pc_item_result_size(len, (int)sizeof(struct pc_item_result));
503
+ st = &res->st;
504
+ } else {
505
+ assert_pc_item_result_size(len, (int)PC_ITEM_RESULT_BASE_SIZE);
506
+ }
507
+
508
+ if (!worker->nr_items_to_complete)
509
+ BUG("received result from supposedly finished checkout worker");
510
+ if (res->id != worker->next_item_to_complete)
511
+ BUG("unexpected item id from checkout worker (got %"PRIuMAX", exp %"PRIuMAX")",
512
+ (uintmax_t)res->id, (uintmax_t)worker->next_item_to_complete);
513
+
514
+ worker->next_item_to_complete++;
515
+ worker->nr_items_to_complete--;
516
+
517
+ pc_item = ¶llel_checkout.items[res->id];
518
+ pc_item->status = res->status;
519
+ if (st)
520
+ pc_item->st = *st;
521
+}
522
+
523
+static void gather_results_from_workers(struct pc_worker *workers,
524
+ int num_workers)
525
+{
526
+ int i, active_workers = num_workers;
527
+ struct pollfd *pfds;
528
+
529
+ CALLOC_ARRAY(pfds, num_workers);
530
+ for (i = 0; i < num_workers; i++) {
531
+ pfds[i].fd = workers[i].cp.out;
532
+ pfds[i].events = POLLIN;
533
+ }
534
+
535
+ while (active_workers) {
536
+ int nr = poll(pfds, num_workers, -1);
537
+
538
+ if (nr < 0) {
539
+ if (errno == EINTR)
540
+ continue;
541
+ die_errno("failed to poll checkout workers");
542
+ }
543
+
544
+ for (i = 0; i < num_workers && nr > 0; i++) {
545
+ struct pc_worker *worker = &workers[i];
546
+ struct pollfd *pfd = &pfds[i];
547
+
548
+ if (!pfd->revents)
549
+ continue;
550
+
551
+ if (pfd->revents & POLLIN) {
552
+ int len = packet_read(pfd->fd, NULL, NULL,
553
+ packet_buffer,
554
+ sizeof(packet_buffer), 0);
555
+
556
+ if (len < 0) {
557
+ BUG("packet_read() returned negative value");
558
+ } else if (!len) {
559
+ pfd->fd = -1;
560
+ active_workers--;
561
+ } else {
562
+ parse_and_save_result(packet_buffer,
563
+ len, worker);
564
+ }
565
+ } else if (pfd->revents & POLLHUP) {
566
+ pfd->fd = -1;
567
+ active_workers--;
568
+ } else if (pfd->revents & (POLLNVAL | POLLERR)) {
569
+ die("error polling from checkout worker");
570
+ }
571
+
572
+ nr--;
573
+ }
574
+ }
575
+
576
+ free(pfds);
577
+}
578
+
579
static void write_items_sequentially(struct checkout *state)
580
{
581
size_t i;
584
write_pc_item(¶llel_checkout.items[i], state);
585
}
586
587
+static const int DEFAULT_NUM_WORKERS = 2;
588
+
589
int run_parallel_checkout(struct checkout *state)
590
{
353
- int ret;
591
+ int ret, num_workers = DEFAULT_NUM_WORKERS;
592
593
if (parallel_checkout.status != PC_ACCEPTING_ENTRIES)
594
BUG("cannot run parallel checkout: uninitialized or already running");
595
596
parallel_checkout.status = PC_RUNNING;
597
360
- write_items_sequentially(state);
598
+ if (parallel_checkout.nr < num_workers)
599
+ num_workers = parallel_checkout.nr;
600
+
601
+ if (num_workers <= 1) {
602
+ write_items_sequentially(state);
603
+ } else {
604
+ struct pc_worker *workers = setup_workers(state, num_workers);
605
+ gather_results_from_workers(workers, num_workers);
606
+ finish_workers(workers, num_workers);
607
+ }
608
+
609
ret = handle_results(state);
610
611
finish_parallel_checkout();