global/git
1
0
Fork 0
mirror of https://github.com/git/git.git synced 2026-02-26 01:43:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'ar/run-command-hook-take-2' into jch

Browse Source 
Use the hook API to replace ad-hoc invocation of hook scripts via
the run_command() API.

* ar/run-command-hook-take-2:
  receive-pack: convert receive hooks to hook API
  receive-pack: convert update hooks to new API
  run-command: poll child input in addition to output
  hook: add jobs option
  reference-transaction: use hook API instead of run-command
  transport: convert pre-push to hook API
  hook: allow separate std[out|err] streams
  hook: convert 'post-rewrite' hook in sequencer.c to hook API
  hook: provide stdin via callback
  run-command: add stdin callback for parallelization
  run-command: add helper for pp child states
  t1800: add hook output stream tests
This commit is contained in:
Junio C Hamano
2026-02-23 14:25:49 -08:00
parent 5b91ebdaac b5e9ad508c
commit 98655bd40e
11 changed files with 754 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files

281
builtin/receive-pack.c
View File

@@ -561,6 +561,48 @@ static int copy_to_sideband(int in, int out UNUSED, void *arg UNUSED)
return 0;
}
/*
* Start an async thread which redirects hook stderr over the sideband.
* The original stderr fd is saved to `saved_stderr` and STDERR_FILENO is
* redirected to the async's input pipe.
*/
static void prepare_sideband_async(struct async *sideband_async, int *saved_stderr, int *started)
{
*started = 0;
if (!use_sideband)
return;
memset(sideband_async, 0, sizeof(*sideband_async));
sideband_async->proc = copy_to_sideband;
sideband_async->in = -1;
if (!start_async(sideband_async)) {
*started = 1;
*saved_stderr = dup(STDERR_FILENO);
if (*saved_stderr >= 0)
dup2(sideband_async->in, STDERR_FILENO);
close(sideband_async->in);
}
}
/*
* Restore the original stderr and wait for the async sideband thread to finish.
*/
static void finish_sideband_async(struct async *sideband_async, int saved_stderr, int started)
{
if (!use_sideband)
return;
if (saved_stderr >= 0) {
dup2(saved_stderr, STDERR_FILENO);
close(saved_stderr);
}
if (started)
finish_async(sideband_async);
}
static void hmac_hash(unsigned char *out,
const char *key_in, size_t key_len,
const char *text, size_t text_len)
@@ -749,7 +791,7 @@ static int check_cert_push_options(const struct string_list *push_options)
return retval;
}
static void prepare_push_cert_sha1(struct child_process *proc)
static void prepare_push_cert_sha1(struct run_hooks_opt *opt)
{
static int already_done;
@@ -775,23 +817,23 @@ static void prepare_push_cert_sha1(struct child_process *proc)
nonce_status = check_nonce(sigcheck.payload);
}
if (!is_null_oid(&push_cert_oid)) {
strvec_pushf(&proc->env, "GIT_PUSH_CERT=%s",
strvec_pushf(&opt->env, "GIT_PUSH_CERT=%s",
oid_to_hex(&push_cert_oid));
strvec_pushf(&proc->env, "GIT_PUSH_CERT_SIGNER=%s",
strvec_pushf(&opt->env, "GIT_PUSH_CERT_SIGNER=%s",
sigcheck.signer ? sigcheck.signer : "");
strvec_pushf(&proc->env, "GIT_PUSH_CERT_KEY=%s",
strvec_pushf(&opt->env, "GIT_PUSH_CERT_KEY=%s",
sigcheck.key ? sigcheck.key : "");
strvec_pushf(&proc->env, "GIT_PUSH_CERT_STATUS=%c",
strvec_pushf(&opt->env, "GIT_PUSH_CERT_STATUS=%c",
sigcheck.result);
if (push_cert_nonce) {
strvec_pushf(&proc->env,
strvec_pushf(&opt->env,
"GIT_PUSH_CERT_NONCE=%s",
push_cert_nonce);
strvec_pushf(&proc->env,
strvec_pushf(&opt->env,
"GIT_PUSH_CERT_NONCE_STATUS=%s",
nonce_status);
if (nonce_status == NONCE_SLOP)
strvec_pushf(&proc->env,
strvec_pushf(&opt->env,
"GIT_PUSH_CERT_NONCE_SLOP=%ld",
nonce_stamp_slop);
}
@@ -803,94 +845,25 @@ struct receive_hook_feed_state {
struct ref_push_report *report;
int skip_broken;
struct strbuf buf;
const struct string_list *push_options;
};
typedef int (*feed_fn)(void *, const char **, size_t *);
static int run_and_feed_hook(const char *hook_name, feed_fn feed,
struct receive_hook_feed_state *feed_state)
static int feed_receive_hook_cb(int hook_stdin_fd, void *pp_cb UNUSED, void *pp_task_cb)
{
struct child_process proc = CHILD_PROCESS_INIT;
struct async muxer;
int code;
const char *hook_path = find_hook(the_repository, hook_name);
if (!hook_path)
return 0;
strvec_push(&proc.args, hook_path);
proc.in = -1;
proc.stdout_to_stderr = 1;
proc.trace2_hook_name = hook_name;
if (feed_state->push_options) {
size_t i;
for (i = 0; i < feed_state->push_options->nr; i++)
strvec_pushf(&proc.env,
"GIT_PUSH_OPTION_%"PRIuMAX"=%s",
(uintmax_t)i,
feed_state->push_options->items[i].string);
strvec_pushf(&proc.env, "GIT_PUSH_OPTION_COUNT=%"PRIuMAX"",
(uintmax_t)feed_state->push_options->nr);
} else
strvec_pushf(&proc.env, "GIT_PUSH_OPTION_COUNT");
if (tmp_objdir)
strvec_pushv(&proc.env, tmp_objdir_env(tmp_objdir));
if (use_sideband) {
memset(&muxer, 0, sizeof(muxer));
muxer.proc = copy_to_sideband;
muxer.in = -1;
code = start_async(&muxer);
if (code)
return code;
proc.err = muxer.in;
}
prepare_push_cert_sha1(&proc);
code = start_command(&proc);
if (code) {
if (use_sideband)
finish_async(&muxer);
return code;
}
sigchain_push(SIGPIPE, SIG_IGN);
while (1) {
const char *buf;
size_t n;
if (feed(feed_state, &buf, &n))
break;
if (write_in_full(proc.in, buf, n) < 0)
break;
}
close(proc.in);
if (use_sideband)
finish_async(&muxer);
sigchain_pop(SIGPIPE);
return finish_command(&proc);
}
static int feed_receive_hook(void *state_, const char **bufp, size_t *sizep)
{
struct receive_hook_feed_state *state = state_;
struct receive_hook_feed_state *state = pp_task_cb;
struct command *cmd = state->cmd;
strbuf_reset(&state->buf);
while (cmd &&
state->skip_broken && (cmd->error_string || cmd->did_not_exist))
cmd = cmd->next;
if (!cmd)
return -1; /* EOF */
if (!bufp)
return 0; /* OK, can feed something. */
strbuf_reset(&state->buf);
return 1; /* no more commands left */
if (!state->report)
state->report = cmd->report;
if (state->report) {
struct object_id *old_oid;
struct object_id *new_oid;
@@ -899,23 +872,33 @@ static int feed_receive_hook(void *state_, const char **bufp, size_t *sizep)
old_oid = state->report->old_oid ? state->report->old_oid : &cmd->old_oid;
new_oid = state->report->new_oid ? state->report->new_oid : &cmd->new_oid;
ref_name = state->report->ref_name ? state->report->ref_name : cmd->ref_name;
strbuf_addf(&state->buf, "%s %s %s\n",
oid_to_hex(old_oid), oid_to_hex(new_oid),
ref_name);
state->report = state->report->next;
if (!state->report)
state->cmd = cmd->next;
cmd = cmd->next;
} else {
strbuf_addf(&state->buf, "%s %s %s\n",
oid_to_hex(&cmd->old_oid), oid_to_hex(&cmd->new_oid),
cmd->ref_name);
state->cmd = cmd->next;
cmd = cmd->next;
}
if (bufp) {
*bufp = state->buf.buf;
*sizep = state->buf.len;
state->cmd = cmd;
if (state->buf.len > 0) {
int ret = write_in_full(hook_stdin_fd, state->buf.buf, state->buf.len);
if (ret < 0) {
if (errno == EPIPE)
return 1; /* child closed pipe */
return ret;
}
}
return 0;
return state->cmd ? 0 : 1; /* 0 = more to come, 1 = EOF */
}
static int run_receive_hook(struct command *commands,
@@ -923,47 +906,74 @@ static int run_receive_hook(struct command *commands,
int skip_broken,
const struct string_list *push_options)
{
struct receive_hook_feed_state state;
int status;
struct run_hooks_opt opt = RUN_HOOKS_OPT_INIT;
struct command *iter = commands;
struct receive_hook_feed_state feed_state;
struct async sideband_async;
int sideband_async_started = 0;
int saved_stderr = -1;
int ret;
strbuf_init(&state.buf, 0);
state.cmd = commands;
state.skip_broken = skip_broken;
state.report = NULL;
if (feed_receive_hook(&state, NULL, NULL))
/* if there are no valid commands, don't invoke the hook at all. */
while (iter && skip_broken && (iter->error_string || iter->did_not_exist))
iter = iter->next;
if (!iter)
return 0;
state.cmd = commands;
state.push_options = push_options;
status = run_and_feed_hook(hook_name, feed_receive_hook, &state);
strbuf_release(&state.buf);
return status;
if (push_options) {
for (int i = 0; i < push_options->nr; i++)
strvec_pushf(&opt.env, "GIT_PUSH_OPTION_%d=%s", i,
push_options->items[i].string);
strvec_pushf(&opt.env, "GIT_PUSH_OPTION_COUNT=%"PRIuMAX"",
(uintmax_t)push_options->nr);
} else {
strvec_push(&opt.env, "GIT_PUSH_OPTION_COUNT");
}
if (tmp_objdir)
strvec_pushv(&opt.env, tmp_objdir_env(tmp_objdir));
prepare_push_cert_sha1(&opt);
prepare_sideband_async(&sideband_async, &saved_stderr, &sideband_async_started);
/* set up stdin callback */
feed_state.cmd = commands;
feed_state.skip_broken = skip_broken;
feed_state.report = NULL;
strbuf_init(&feed_state.buf, 0);
opt.feed_pipe_cb_data = &feed_state;
opt.feed_pipe = feed_receive_hook_cb;
ret = run_hooks_opt(the_repository, hook_name, &opt);
strbuf_release(&feed_state.buf);
finish_sideband_async(&sideband_async, saved_stderr, sideband_async_started);
return ret;
}
static int run_update_hook(struct command *cmd)
{
struct child_process proc = CHILD_PROCESS_INIT;
struct run_hooks_opt opt = RUN_HOOKS_OPT_INIT;
struct async sideband_async;
int sideband_async_started = 0;
int saved_stderr = -1;
int code;
const char *hook_path = find_hook(the_repository, "update");
if (!hook_path)
return 0;
strvec_pushl(&opt.args,
cmd->ref_name,
oid_to_hex(&cmd->old_oid),
oid_to_hex(&cmd->new_oid),
NULL);
strvec_push(&proc.args, hook_path);
strvec_push(&proc.args, cmd->ref_name);
strvec_push(&proc.args, oid_to_hex(&cmd->old_oid));
strvec_push(&proc.args, oid_to_hex(&cmd->new_oid));
prepare_sideband_async(&sideband_async, &saved_stderr, &sideband_async_started);
proc.no_stdin = 1;
proc.stdout_to_stderr = 1;
proc.err = use_sideband ? -1 : 0;
proc.trace2_hook_name = "update";
code = run_hooks_opt(the_repository, "update", &opt);
code = start_command(&proc);
if (code)
return code;
if (use_sideband)
copy_to_sideband(proc.err, -1, NULL);
return finish_command(&proc);
finish_sideband_async(&sideband_async, saved_stderr, sideband_async_started);
return code;
}
static struct command *find_command_by_refname(struct command *list,
@@ -1639,34 +1649,25 @@ out:
static void run_update_post_hook(struct command *commands)
{
struct run_hooks_opt opt = RUN_HOOKS_OPT_INIT;
struct async sideband_async;
struct command *cmd;
struct child_process proc = CHILD_PROCESS_INIT;
const char *hook;
hook = find_hook(the_repository, "post-update");
if (!hook)
return;
int sideband_async_started = 0;
int saved_stderr = -1;
for (cmd = commands; cmd; cmd = cmd->next) {
if (cmd->error_string || cmd->did_not_exist)
continue;
if (!proc.args.nr)
strvec_push(&proc.args, hook);
strvec_push(&proc.args, cmd->ref_name);
strvec_push(&opt.args, cmd->ref_name);
}
if (!proc.args.nr)
if (!opt.args.nr)
return;
proc.no_stdin = 1;
proc.stdout_to_stderr = 1;
proc.err = use_sideband ? -1 : 0;
proc.trace2_hook_name = "post-update";
prepare_sideband_async(&sideband_async, &saved_stderr, &sideband_async_started);
if (!start_command(&proc)) {
if (use_sideband)
copy_to_sideband(proc.err, -1, NULL);
finish_command(&proc);
}
run_hooks_opt(the_repository, "post-update", &opt);
finish_sideband_async(&sideband_async, saved_stderr, sideband_async_started);
}
static void check_aliased_update_internal(struct command *cmd,

32
hook.c
View File

@@ -55,7 +55,7 @@ int hook_exists(struct repository *r, const char *name)
static int pick_next_hook(struct child_process *cp,
struct strbuf *out UNUSED,
void *pp_cb,
void **pp_task_cb UNUSED)
void **pp_task_cb)
{
struct hook_cb_data *hook_cb = pp_cb;
const char *hook_path = hook_cb->hook_path;
@@ -65,18 +65,35 @@ static int pick_next_hook(struct child_process *cp,
cp->no_stdin = 1;
strvec_pushv(&cp->env, hook_cb->options->env.v);
if (hook_cb->options->path_to_stdin && hook_cb->options->feed_pipe)
BUG("options path_to_stdin and feed_pipe are mutually exclusive");
/* reopen the file for stdin; run_command closes it. */
if (hook_cb->options->path_to_stdin) {
cp->no_stdin = 0;
cp->in = xopen(hook_cb->options->path_to_stdin, O_RDONLY);
}
cp->stdout_to_stderr = 1;
if (hook_cb->options->feed_pipe) {
cp->no_stdin = 0;
/* start_command() will allocate a pipe / stdin fd for us */
cp->in = -1;
}
cp->stdout_to_stderr = hook_cb->options->stdout_to_stderr;
cp->trace2_hook_name = hook_cb->hook_name;
cp->dir = hook_cb->options->dir;
strvec_push(&cp->args, hook_path);
strvec_pushv(&cp->args, hook_cb->options->args.v);
/*
* Provide per-hook internal state via task_cb for easy access, so
* hook callbacks don't have to go through hook_cb->options.
*/
*pp_task_cb = hook_cb->options->feed_pipe_cb_data;
/*
* This pick_next_hook() will be called again, we're only
* running one hook, so indicate that no more work will be
@@ -135,11 +152,12 @@ int run_hooks_opt(struct repository *r, const char *hook_name,
.tr2_category = "hook",
.tr2_label = hook_name,
.processes = 1,
.ungroup = 1,
.processes = options->jobs,
.ungroup = options->jobs == 1,
.get_next_task = pick_next_hook,
.start_failure = notify_start_failure,
.feed_pipe = options->feed_pipe,
.task_finished = notify_hook_finished,
.data = &cb_data,
@@ -148,6 +166,12 @@ int run_hooks_opt(struct repository *r, const char *hook_name,
if (!options)
BUG("a struct run_hooks_opt must be provided to run_hooks");
if (options->path_to_stdin && options->feed_pipe)
BUG("options path_to_stdin and feed_pipe are mutually exclusive");
if (!options->jobs)
BUG("run_hooks_opt must be called with options.jobs >= 1");
if (options->invoked_hook)
*options->invoked_hook = 0;

57
hook.h
View File

@@ -1,6 +1,7 @@
#ifndef HOOK_H
#define HOOK_H
#include "strvec.h"
#include "run-command.h"
struct repository;
@@ -15,6 +16,14 @@ struct run_hooks_opt
/* Emit an error if the hook is missing */
unsigned int error_if_missing:1;
/**
* Number of processes to parallelize across.
*
* If > 1, output will be buffered and de-interleaved (ungroup=0).
* If == 1, output will be real-time (ungroup=1).
*/
unsigned int jobs;
/**
* An optional initial working directory for the hook,
* translates to "struct child_process"'s "dir" member.
@@ -33,15 +42,63 @@ struct run_hooks_opt
*/
int *invoked_hook;
/**
* Send the hook's stdout to stderr.
*
* This is the default behavior for all hooks except pre-push,
* which has separate stdout and stderr streams for backwards
* compatibility reasons.
*/
unsigned int stdout_to_stderr:1;
/**
* Path to file which should be piped to stdin for each hook.
*/
const char *path_to_stdin;
/**
* Callback used to incrementally feed a child hook stdin pipe.
*
* Useful especially if a hook consumes large quantities of data
* (e.g. a list of all refs in a client push), so feeding it via
* in-memory strings or slurping to/from files is inefficient.
* While the callback allows piecemeal writing, it can also be
* used for smaller inputs, where it gets called only once.
*
* Add hook callback initalization context to `feed_pipe_ctx`.
* Add hook callback internal state to `feed_pipe_cb_data`.
*
*/
feed_pipe_fn feed_pipe;
/**
* Opaque data pointer used to pass context to `feed_pipe_fn`.
*
* It can be accessed via the second callback arg 'pp_cb':
* ((struct hook_cb_data *) pp_cb)->hook_cb->options->feed_pipe_ctx;
*
* The caller is responsible for managing the memory for this data.
* Only useful when using `run_hooks_opt.feed_pipe`, otherwise ignore it.
*/
void *feed_pipe_ctx;
/**
* Opaque data pointer used to keep internal state across callback calls.
*
* It can be accessed directly via the third callback arg 'pp_task_cb':
* struct ... *state = pp_task_cb;
*
* The caller is responsible for managing the memory for this data.
* Only useful when using `run_hooks_opt.feed_pipe`, otherwise ignore it.
*/
void *feed_pipe_cb_data;
};
#define RUN_HOOKS_OPT_INIT { \
.env = STRVEC_INIT, \
.args = STRVEC_INIT, \
.stdout_to_stderr = 1, \
.jobs = 1, \
}
struct hook_cb_data {

112
refs.c
View File

@@ -15,7 +15,6 @@
#include "iterator.h"
#include "refs.h"
#include "refs/refs-internal.h"
#include "run-command.h"
#include "hook.h"
#include "object-name.h"
#include "odb.h"
@@ -26,7 +25,6 @@
#include "strvec.h"
#include "repo-settings.h"
#include "setup.h"
#include "sigchain.h"
#include "date.h"
#include "commit.h"
#include "wildmatch.h"
@@ -2468,68 +2466,72 @@ static int ref_update_reject_duplicates(struct string_list *refnames,
return 0;
}
struct transaction_feed_cb_data {
size_t index;
struct strbuf buf;
};
static int transaction_hook_feed_stdin(int hook_stdin_fd, void *pp_cb, void *pp_task_cb)
{
struct hook_cb_data *hook_cb = pp_cb;
struct ref_transaction *transaction = hook_cb->options->feed_pipe_ctx;
struct transaction_feed_cb_data *feed_cb_data = pp_task_cb;
struct strbuf *buf = &feed_cb_data->buf;
struct ref_update *update;
size_t i = feed_cb_data->index++;
int ret;
if (i >= transaction->nr)
return 1; /* No more refs to process */
update = transaction->updates[i];
if (update->flags & REF_LOG_ONLY)
return 0;
strbuf_reset(buf);
if (!(update->flags & REF_HAVE_OLD))
strbuf_addf(buf, "%s ", oid_to_hex(null_oid(the_hash_algo)));
else if (update->old_target)
strbuf_addf(buf, "ref:%s ", update->old_target);
else
strbuf_addf(buf, "%s ", oid_to_hex(&update->old_oid));
if (!(update->flags & REF_HAVE_NEW))
strbuf_addf(buf, "%s ", oid_to_hex(null_oid(the_hash_algo)));
else if (update->new_target)
strbuf_addf(buf, "ref:%s ", update->new_target);
else
strbuf_addf(buf, "%s ", oid_to_hex(&update->new_oid));
strbuf_addf(buf, "%s\n", update->refname);
ret = write_in_full(hook_stdin_fd, buf->buf, buf->len);
if (ret < 0 && errno != EPIPE)
return ret;
return 0; /* no more input to feed */
}
static int run_transaction_hook(struct ref_transaction *transaction,
const char *state)
{
struct child_process proc = CHILD_PROCESS_INIT;
struct strbuf buf = STRBUF_INIT;
const char *hook;
struct run_hooks_opt opt = RUN_HOOKS_OPT_INIT;
struct transaction_feed_cb_data feed_ctx = { 0 };
int ret = 0;
hook = find_hook(transaction->ref_store->repo, "reference-transaction");
if (!hook)
return ret;
strvec_push(&opt.args, state);
strvec_pushl(&proc.args, hook, state, NULL);
proc.in = -1;
proc.stdout_to_stderr = 1;
proc.trace2_hook_name = "reference-transaction";
opt.feed_pipe = transaction_hook_feed_stdin;
opt.feed_pipe_ctx = transaction;
opt.feed_pipe_cb_data = &feed_ctx;
ret = start_command(&proc);
if (ret)
return ret;
strbuf_init(&feed_ctx.buf, 0);
sigchain_push(SIGPIPE, SIG_IGN);
ret = run_hooks_opt(transaction->ref_store->repo, "reference-transaction", &opt);
for (size_t i = 0; i < transaction->nr; i++) {
struct ref_update *update = transaction->updates[i];
if (update->flags & REF_LOG_ONLY)
continue;
strbuf_reset(&buf);
if (!(update->flags & REF_HAVE_OLD))
strbuf_addf(&buf, "%s ", oid_to_hex(null_oid(the_hash_algo)));
else if (update->old_target)
strbuf_addf(&buf, "ref:%s ", update->old_target);
else
strbuf_addf(&buf, "%s ", oid_to_hex(&update->old_oid));
if (!(update->flags & REF_HAVE_NEW))
strbuf_addf(&buf, "%s ", oid_to_hex(null_oid(the_hash_algo)));
else if (update->new_target)
strbuf_addf(&buf, "ref:%s ", update->new_target);
else
strbuf_addf(&buf, "%s ", oid_to_hex(&update->new_oid));
strbuf_addf(&buf, "%s\n", update->refname);
if (write_in_full(proc.in, buf.buf, buf.len) < 0) {
if (errno != EPIPE) {
/* Don't leak errno outside this API */
errno = 0;
ret = -1;
}
break;
}
}
close(proc.in);
sigchain_pop(SIGPIPE);
strbuf_release(&buf);
ret |= finish_command(&proc);
strbuf_release(&feed_ctx.buf);
return ret;
}

174
run-command.c
View File

@@ -1478,15 +1478,40 @@ enum child_state {
GIT_CP_WAIT_CLEANUP,
};
struct parallel_child {
enum child_state state;
struct child_process process;
struct strbuf err;
void *data;
};
static int child_is_working(const struct parallel_child *pp_child)
{
return pp_child->state == GIT_CP_WORKING;
}
static int child_is_ready_for_cleanup(const struct parallel_child *pp_child)
{
return child_is_working(pp_child) && !pp_child->process.in;
}
static int child_is_receiving_input(const struct parallel_child *pp_child)
{
return child_is_working(pp_child) && pp_child->process.in > 0;
}
static int child_is_sending_output(const struct parallel_child *pp_child)
{
/*
* all pp children which buffer output through run_command via ungroup=0
* redirect stdout to stderr, so we just need to check process.err.
*/
return child_is_working(pp_child) && pp_child->process.err > 0;
}
struct parallel_processes {
size_t nr_processes;
struct {
enum child_state state;
struct child_process process;
struct strbuf err;
void *data;
} *children;
struct parallel_child *children;
/*
* The struct pollfd is logically part of *children,
* but the system call expects it as its own array.
@@ -1509,7 +1534,7 @@ static void kill_children(const struct parallel_processes *pp,
int signo)
{
for (size_t i = 0; i < opts->processes; i++)
if (pp->children[i].state == GIT_CP_WORKING)
if (child_is_working(&pp->children[i]))
kill(pp->children[i].process.pid, signo);
}
@@ -1545,7 +1570,7 @@ static void pp_init(struct parallel_processes *pp,
CALLOC_ARRAY(pp->children, n);
if (!opts->ungroup)
CALLOC_ARRAY(pp->pfd, n);
CALLOC_ARRAY(pp->pfd, n * 2);
for (size_t i = 0; i < n; i++) {
strbuf_init(&pp->children[i].err, 0);
@@ -1652,21 +1677,101 @@ static int pp_start_one(struct parallel_processes *pp,
return 0;
}
static void pp_buffer_stderr(struct parallel_processes *pp,
const struct run_process_parallel_opts *opts,
int output_timeout)
static void pp_buffer_stdin(struct parallel_processes *pp,
const struct run_process_parallel_opts *opts)
{
while (poll(pp->pfd, opts->processes, output_timeout) < 0) {
/* Buffer stdin for each pipe. */
for (size_t i = 0; i < opts->processes; i++) {
struct child_process *proc = &pp->children[i].process;
int ret;
if (!child_is_receiving_input(&pp->children[i]))
continue;
/*
* child input is provided via path_to_stdin when the feed_pipe cb is
* missing, so we just signal an EOF.
*/
if (!opts->feed_pipe) {
close(proc->in);
proc->in = 0;
continue;
}
/**
* Feed the pipe:
* ret < 0 means error
* ret == 0 means there is more data to be fed
* ret > 0 means feeding finished
*/
ret = opts->feed_pipe(proc->in, opts->data, pp->children[i].data);
if (ret < 0)
die_errno("feed_pipe");
if (ret) {
close(proc->in);
proc->in = 0;
}
}
}
static void pp_buffer_io(struct parallel_processes *pp,
const struct run_process_parallel_opts *opts,
int timeout)
{
/* for each potential child slot, prepare two pollfd entries */
for (size_t i = 0; i < opts->processes; i++) {
if (child_is_sending_output(&pp->children[i])) {
pp->pfd[2*i].fd = pp->children[i].process.err;
pp->pfd[2*i].events = POLLIN | POLLHUP;
} else {
pp->pfd[2*i].fd = -1;
}
if (child_is_receiving_input(&pp->children[i])) {
pp->pfd[2*i+1].fd = pp->children[i].process.in;
pp->pfd[2*i+1].events = POLLOUT;
} else {
pp->pfd[2*i+1].fd = -1;
}
}
while (poll(pp->pfd, opts->processes * 2, timeout) < 0) {
if (errno == EINTR)
continue;
pp_cleanup(pp, opts);
die_errno("poll");
}
/* Buffer output from all pipes. */
for (size_t i = 0; i < opts->processes; i++) {
if (pp->children[i].state == GIT_CP_WORKING &&
pp->pfd[i].revents & (POLLIN | POLLHUP)) {
/* Handle input feeding (stdin) */
if (pp->pfd[2*i+1].revents & (POLLOUT | POLLHUP | POLLERR)) {
if (opts->feed_pipe) {
int ret = opts->feed_pipe(pp->children[i].process.in,
opts->data,
pp->children[i].data);
if (ret < 0)
die_errno("feed_pipe");
if (ret) {
/* done feeding */
close(pp->children[i].process.in);
pp->children[i].process.in = 0;
}
} else {
/*
* No feed_pipe means there is nothing to do, so
* close the fd. Child input can be fed by other
* methods, such as opts->path_to_stdin which
* slurps a file via dup2, so clean up here.
*/
close(pp->children[i].process.in);
pp->children[i].process.in = 0;
}
}
/* Handle output reading (stderr) */
if (child_is_working(&pp->children[i]) &&
pp->pfd[2*i].revents & (POLLIN | POLLHUP)) {
int n = strbuf_read_once(&pp->children[i].err,
pp->children[i].process.err, 0);
if (n == 0) {
@@ -1683,7 +1788,7 @@ static void pp_output(const struct parallel_processes *pp)
{
size_t i = pp->output_owner;
if (pp->children[i].state == GIT_CP_WORKING &&
if (child_is_working(&pp->children[i]) &&
pp->children[i].err.len) {
strbuf_write(&pp->children[i].err, stderr);
strbuf_reset(&pp->children[i].err);
@@ -1722,6 +1827,7 @@ static int pp_collect_finished(struct parallel_processes *pp,
pp->children[i].state = GIT_CP_FREE;
if (pp->pfd)
pp->pfd[i].fd = -1;
pp->children[i].process.in = 0;
child_process_init(&pp->children[i].process);
if (opts->ungroup) {
@@ -1748,7 +1854,7 @@ static int pp_collect_finished(struct parallel_processes *pp,
* running process time.
*/
for (i = 0; i < n; i++)
if (pp->children[(pp->output_owner + i) % n].state == GIT_CP_WORKING)
if (child_is_working(&pp->children[(pp->output_owner + i) % n]))
break;
pp->output_owner = (pp->output_owner + i) % n;
}
@@ -1756,10 +1862,25 @@ static int pp_collect_finished(struct parallel_processes *pp,
return result;
}
static void pp_handle_child_IO(struct parallel_processes *pp,
const struct run_process_parallel_opts *opts,
int timeout)
{
if (opts->ungroup) {
pp_buffer_stdin(pp, opts);
for (size_t i = 0; i < opts->processes; i++)
if (child_is_ready_for_cleanup(&pp->children[i]))
pp->children[i].state = GIT_CP_WAIT_CLEANUP;
} else {
pp_buffer_io(pp, opts, timeout);
pp_output(pp);
}
}
void run_processes_parallel(const struct run_process_parallel_opts *opts)
{
int i, code;
int output_timeout = 100;
int timeout = 100;
int spawn_cap = 4;
struct parallel_processes_for_signal pp_sig;
struct parallel_processes pp = {
@@ -1775,6 +1896,13 @@ void run_processes_parallel(const struct run_process_parallel_opts *opts)
"max:%"PRIuMAX,
(uintmax_t)opts->processes);
/*
* Child tasks might receive input via stdin, terminating early (or not), so
* ignore the default SIGPIPE which gets handled by each feed_pipe_fn which
* actually writes the data to children stdin fds.
*/
sigchain_push(SIGPIPE, SIG_IGN);
pp_init(&pp, opts, &pp_sig);
while (1) {
for (i = 0;
@@ -1792,13 +1920,7 @@ void run_processes_parallel(const struct run_process_parallel_opts *opts)
}
if (!pp.nr_processes)
break;
if (opts->ungroup) {
for (size_t i = 0; i < opts->processes; i++)
pp.children[i].state = GIT_CP_WAIT_CLEANUP;
} else {
pp_buffer_stderr(&pp, opts, output_timeout);
pp_output(&pp);
}
pp_handle_child_IO(&pp, opts, timeout);
code = pp_collect_finished(&pp, opts);
if (code) {
pp.shutdown = 1;
@@ -1809,6 +1931,8 @@ void run_processes_parallel(const struct run_process_parallel_opts *opts)
pp_cleanup(&pp, opts);
sigchain_pop(SIGPIPE);
if (do_trace2)
trace2_region_leave(tr2_category, tr2_label, NULL);
}

21
run-command.h
View File

@@ -420,6 +420,21 @@ typedef int (*start_failure_fn)(struct strbuf *out,
void *pp_cb,
void *pp_task_cb);
/**
* This callback is repeatedly called on every child process who requests
* start_command() to create a pipe by setting child_process.in < 0.
*
* pp_cb is the callback cookie as passed into run_processes_parallel, and
* pp_task_cb is the callback cookie as passed into get_next_task_fn.
*
* Returns < 0 for error
* Returns == 0 when there is more data to be fed (will be called again)
* Returns > 0 when finished (child closed fd or no more data to be fed)
*/
typedef int (*feed_pipe_fn)(int child_in,
void *pp_cb,
void *pp_task_cb);
/**
* This callback is called on every child process that finished processing.
*
@@ -473,6 +488,12 @@ struct run_process_parallel_opts
*/
start_failure_fn start_failure;
/*
* feed_pipe: see feed_pipe_fn() above. This can be NULL to omit any
* special handling.
*/
feed_pipe_fn feed_pipe;
/**
* task_finished: See task_finished_fn() above. This can be
* NULL to omit any special handling.

44
sequencer.c
View File

@@ -1292,32 +1292,40 @@ int update_head_with_reflog(const struct commit *old_head,
return ret;
}
static int pipe_from_strbuf(int hook_stdin_fd, void *pp_cb, void *pp_task_cb UNUSED)
{
struct hook_cb_data *hook_cb = pp_cb;
struct strbuf *to_pipe = hook_cb->options->feed_pipe_ctx;
int ret;
if (!to_pipe)
BUG("pipe_from_strbuf called without feed_pipe_ctx");
ret = write_in_full(hook_stdin_fd, to_pipe->buf, to_pipe->len);
if (ret < 0 && errno != EPIPE)
return ret;
return 1; /* done writing */
}
static int run_rewrite_hook(const struct object_id *oldoid,
const struct object_id *newoid)
{
struct child_process proc = CHILD_PROCESS_INIT;
struct run_hooks_opt opt = RUN_HOOKS_OPT_INIT;
int code;
struct strbuf sb = STRBUF_INIT;
const char *hook_path = find_hook(the_repository, "post-rewrite");
if (!hook_path)
return 0;
strvec_pushl(&proc.args, hook_path, "amend", NULL);
proc.in = -1;
proc.stdout_to_stderr = 1;
proc.trace2_hook_name = "post-rewrite";
code = start_command(&proc);
if (code)
return code;
strbuf_addf(&sb, "%s %s\n", oid_to_hex(oldoid), oid_to_hex(newoid));
sigchain_push(SIGPIPE, SIG_IGN);
write_in_full(proc.in, sb.buf, sb.len);
close(proc.in);
opt.feed_pipe_ctx = &sb;
opt.feed_pipe = pipe_from_strbuf;
strvec_push(&opt.args, "amend");
code = run_hooks_opt(the_repository, "post-rewrite", &opt);
strbuf_release(&sb);
sigchain_pop(SIGPIPE);
return finish_command(&proc);
return code;
}
void commit_post_rewrite(struct repository *r,

52
t/helper/test-run-command.c
View File

@@ -23,19 +23,26 @@ static int number_callbacks;
static int parallel_next(struct child_process *cp,
struct strbuf *err,
void *cb,
void **task_cb UNUSED)
void **task_cb)
{
struct child_process *d = cb;
if (number_callbacks >= 4)
return 0;
strvec_pushv(&cp->args, d->args.v);
cp->in = d->in;
cp->no_stdin = d->no_stdin;
if (err)
strbuf_addstr(err, "preloaded output of a child\n");
else
fprintf(stderr, "preloaded output of a child\n");
number_callbacks++;
/* test_stdin callback will use this to count remaining lines */
*task_cb = xmalloc(sizeof(int));
*(int*)(*task_cb) = 2;
return 1;
}
@@ -54,15 +61,48 @@ static int no_job(struct child_process *cp UNUSED,
static int task_finished(int result UNUSED,
struct strbuf *err,
void *pp_cb UNUSED,
void *pp_task_cb UNUSED)
void *pp_task_cb)
{
if (err)
strbuf_addstr(err, "asking for a quick stop\n");
else
fprintf(stderr, "asking for a quick stop\n");
FREE_AND_NULL(pp_task_cb);
return 1;
}
static int task_finished_quiet(int result UNUSED,
struct strbuf *err UNUSED,
void *pp_cb UNUSED,
void *pp_task_cb)
{
FREE_AND_NULL(pp_task_cb);
return 0;
}
static int test_stdin_pipe_feed(int hook_stdin_fd, void *cb UNUSED, void *task_cb)
{
int *lines_remaining = task_cb;
if (*lines_remaining) {
struct strbuf buf = STRBUF_INIT;
strbuf_addf(&buf, "sample stdin %d\n", --(*lines_remaining));
if (write_in_full(hook_stdin_fd, buf.buf, buf.len) < 0) {
if (errno == EPIPE) {
/* child closed stdin, nothing more to do */
strbuf_release(&buf);
return 1;
}
die_errno("write");
}
strbuf_release(&buf);
}
return !(*lines_remaining);
}
struct testsuite {
struct string_list tests, failed;
int next;
@@ -157,6 +197,7 @@ static int testsuite(int argc, const char **argv)
struct run_process_parallel_opts opts = {
.get_next_task = next_test,
.start_failure = test_failed,
.feed_pipe = test_stdin_pipe_feed,
.task_finished = test_finished,
.data = &suite,
};
@@ -460,12 +501,19 @@ int cmd__run_command(int argc, const char **argv)
if (!strcmp(argv[1], "run-command-parallel")) {
opts.get_next_task = parallel_next;
opts.task_finished = task_finished_quiet;
} else if (!strcmp(argv[1], "run-command-abort")) {
opts.get_next_task = parallel_next;
opts.task_finished = task_finished;
} else if (!strcmp(argv[1], "run-command-no-jobs")) {
opts.get_next_task = no_job;
opts.task_finished = task_finished;
} else if (!strcmp(argv[1], "run-command-stdin")) {
proc.in = -1;
proc.no_stdin = 0;
opts.get_next_task = parallel_next;
opts.task_finished = task_finished_quiet;
opts.feed_pipe = test_stdin_pipe_feed;
} else {
ret = 1;
fprintf(stderr, "check usage\n");

31
t/t0061-run-command.sh
View File

@@ -164,6 +164,37 @@ test_expect_success 'run_command runs ungrouped in parallel with more tasks than
test_line_count = 4 err
'
test_expect_success 'run_command listens to stdin' '
cat >expect <<-\EOF &&
preloaded output of a child
listening for stdin:
sample stdin 1
sample stdin 0
preloaded output of a child
listening for stdin:
sample stdin 1
sample stdin 0
preloaded output of a child
listening for stdin:
sample stdin 1
sample stdin 0
preloaded output of a child
listening for stdin:
sample stdin 1
sample stdin 0
EOF
write_script stdin-script <<-\EOF &&
echo "listening for stdin:"
while read line
do
echo "$line"
done
EOF
test-tool run-command run-command-stdin 2 ./stdin-script 2>actual &&
test_cmp expect actual
'
cat >expect <<-EOF
preloaded output of a child
asking for a quick stop

137
t/t1800-hook.sh
View File

@@ -184,4 +184,141 @@ test_expect_success 'stdin to hooks' '
test_cmp expect actual
'
check_stdout_separate_from_stderr () {
for hook in "$@"
do
# Ensure hook's stdout is only in stdout, not stderr
test_grep "Hook $hook stdout" stdout.actual || return 1
test_grep ! "Hook $hook stdout" stderr.actual || return 1
# Ensure hook's stderr is only in stderr, not stdout
test_grep "Hook $hook stderr" stderr.actual || return 1
test_grep ! "Hook $hook stderr" stdout.actual || return 1
done
}
check_stdout_merged_to_stderr () {
for hook in "$@"
do
# Ensure hook's stdout is only in stderr, not stdout
test_grep "Hook $hook stdout" stderr.actual || return 1
test_grep ! "Hook $hook stdout" stdout.actual || return 1
# Ensure hook's stderr is only in stderr, not stdout
test_grep "Hook $hook stderr" stderr.actual || return 1
test_grep ! "Hook $hook stderr" stdout.actual || return 1
done
}
setup_hooks () {
for hook in "$@"
do
test_hook $hook <<-EOF
echo >&1 Hook $hook stdout
echo >&2 Hook $hook stderr
EOF
done
}
test_expect_success 'client hooks: pre-push expects separate stdout and stderr' '
test_when_finished "rm -f stdout.actual stderr.actual" &&
git init --bare remote &&
git remote add origin remote &&
test_commit A &&
setup_hooks pre-push &&
git push origin HEAD:main >stdout.actual 2>stderr.actual &&
check_stdout_separate_from_stderr pre-push
'
test_expect_success 'client hooks: commit hooks expect stdout redirected to stderr' '
hooks="pre-commit prepare-commit-msg \
commit-msg post-commit \
reference-transaction" &&
setup_hooks $hooks &&
test_when_finished "rm -f stdout.actual stderr.actual" &&
git checkout -B main &&
git checkout -b branch-a &&
test_commit commit-on-branch-a &&
git commit --allow-empty -m "Test" >stdout.actual 2>stderr.actual &&
check_stdout_merged_to_stderr $hooks
'
test_expect_success 'client hooks: checkout hooks expect stdout redirected to stderr' '
setup_hooks post-checkout reference-transaction &&
test_when_finished "rm -f stdout.actual stderr.actual" &&
git checkout -b new-branch main >stdout.actual 2>stderr.actual &&
check_stdout_merged_to_stderr post-checkout reference-transaction
'
test_expect_success 'client hooks: merge hooks expect stdout redirected to stderr' '
setup_hooks pre-merge-commit post-merge reference-transaction &&
test_when_finished "rm -f stdout.actual stderr.actual" &&
test_commit new-branch-commit &&
git merge --no-ff branch-a >stdout.actual 2>stderr.actual &&
check_stdout_merged_to_stderr pre-merge-commit post-merge reference-transaction
'
test_expect_success 'client hooks: post-rewrite hooks expect stdout redirected to stderr' '
setup_hooks post-rewrite reference-transaction &&
test_when_finished "rm -f stdout.actual stderr.actual" &&
git commit --amend --allow-empty --no-edit >stdout.actual 2>stderr.actual &&
check_stdout_merged_to_stderr post-rewrite reference-transaction
'
test_expect_success 'client hooks: applypatch hooks expect stdout redirected to stderr' '
setup_hooks applypatch-msg pre-applypatch post-applypatch &&
test_when_finished "rm -f stdout.actual stderr.actual" &&
git checkout -b branch-b main &&
test_commit branch-b &&
git format-patch -1 --stdout >patch &&
git checkout -b branch-c main &&
git am patch >stdout.actual 2>stderr.actual &&
check_stdout_merged_to_stderr applypatch-msg pre-applypatch post-applypatch
'
test_expect_success 'client hooks: rebase hooks expect stdout redirected to stderr' '
setup_hooks pre-rebase &&
test_when_finished "rm -f stdout.actual stderr.actual" &&
git checkout -b branch-d main &&
test_commit branch-d &&
git checkout main &&
test_commit diverge-main &&
git checkout branch-d &&
git rebase main >stdout.actual 2>stderr.actual &&
check_stdout_merged_to_stderr pre-rebase
'
test_expect_success 'client hooks: post-index-change expects stdout redirected to stderr' '
setup_hooks post-index-change &&
test_when_finished "rm -f stdout.actual stderr.actual" &&
oid=$(git hash-object -w --stdin </dev/null) &&
git update-index --add --cacheinfo 100644 $oid new-file \
>stdout.actual 2>stderr.actual &&
check_stdout_merged_to_stderr post-index-change
'
test_expect_success 'server hooks expect stdout redirected to stderr' '
test_when_finished "rm -f stdout.actual stderr.actual" &&
git init --bare remote-server &&
git remote add origin-server remote-server &&
cd remote-server &&
setup_hooks pre-receive update post-receive post-update &&
cd .. &&
git push origin-server HEAD:new-branch >stdout.actual 2>stderr.actual &&
check_stdout_merged_to_stderr pre-receive update post-receive post-update
'
test_expect_success 'server push-to-checkout hook expects stdout redirected to stderr' '
test_when_finished "rm -f stdout.actual stderr.actual" &&
git init server &&
git -C server checkout -b main &&
test_config -C server receive.denyCurrentBranch updateInstead &&
git remote add origin-server-2 server &&
cd server &&
setup_hooks push-to-checkout &&
cd .. &&
git push origin-server-2 HEAD:main >stdout.actual 2>stderr.actual &&
check_stdout_merged_to_stderr push-to-checkout
'
test_done

107
transport.c
View File

@@ -1317,65 +1317,72 @@ static void die_with_unpushed_submodules(struct string_list *needs_pushing)
die(_("Aborting."));
}
struct feed_pre_push_hook_data {
struct strbuf buf;
const struct ref *refs;
};
static int pre_push_hook_feed_stdin(int hook_stdin_fd, void *pp_cb UNUSED, void *pp_task_cb)
{
struct feed_pre_push_hook_data *data = pp_task_cb;
const struct ref *r = data->refs;
int ret = 0;
if (!r)
return 1; /* no more refs */
data->refs = r->next;
switch (r->status) {
case REF_STATUS_REJECT_NONFASTFORWARD:
case REF_STATUS_REJECT_REMOTE_UPDATED:
case REF_STATUS_REJECT_STALE:
case REF_STATUS_UPTODATE:
return 0; /* skip refs which won't be pushed */
default:
break;
}
if (!r->peer_ref)
return 0;
strbuf_reset(&data->buf);
strbuf_addf(&data->buf, "%s %s %s %s\n",
r->peer_ref->name, oid_to_hex(&r->new_oid),
r->name, oid_to_hex(&r->old_oid));
ret = write_in_full(hook_stdin_fd, data->buf.buf, data->buf.len);
if (ret < 0 && errno != EPIPE)
return ret; /* We do not mind if a hook does not read all refs. */
return 0;
}
static int run_pre_push_hook(struct transport *transport,
struct ref *remote_refs)
{
int ret = 0, x;
struct ref *r;
struct child_process proc = CHILD_PROCESS_INIT;
struct strbuf buf;
const char *hook_path = find_hook(the_repository, "pre-push");
struct run_hooks_opt opt = RUN_HOOKS_OPT_INIT;
struct feed_pre_push_hook_data data;
int ret = 0;
if (!hook_path)
return 0;
strvec_push(&opt.args, transport->remote->name);
strvec_push(&opt.args, transport->url);
strvec_push(&proc.args, hook_path);
strvec_push(&proc.args, transport->remote->name);
strvec_push(&proc.args, transport->url);
strbuf_init(&data.buf, 0);
data.refs = remote_refs;
proc.in = -1;
proc.trace2_hook_name = "pre-push";
opt.feed_pipe = pre_push_hook_feed_stdin;
opt.feed_pipe_cb_data = &data;
if (start_command(&proc)) {
finish_command(&proc);
return -1;
}
/*
* pre-push hooks expect stdout & stderr to be separate, so don't merge
* them to keep backwards compatibility with existing hooks.
*/
opt.stdout_to_stderr = 0;
sigchain_push(SIGPIPE, SIG_IGN);
ret = run_hooks_opt(the_repository, "pre-push", &opt);
strbuf_init(&buf, 256);
for (r = remote_refs; r; r = r->next) {
if (!r->peer_ref) continue;
if (r->status == REF_STATUS_REJECT_NONFASTFORWARD) continue;
if (r->status == REF_STATUS_REJECT_STALE) continue;
if (r->status == REF_STATUS_REJECT_REMOTE_UPDATED) continue;
if (r->status == REF_STATUS_UPTODATE) continue;
strbuf_reset(&buf);
strbuf_addf( &buf, "%s %s %s %s\n",
r->peer_ref->name, oid_to_hex(&r->new_oid),
r->name, oid_to_hex(&r->old_oid));
if (write_in_full(proc.in, buf.buf, buf.len) < 0) {
/* We do not mind if a hook does not read all refs. */
if (errno != EPIPE)
ret = -1;
break;
}
}
strbuf_release(&buf);
x = close(proc.in);
if (!ret)
ret = x;
sigchain_pop(SIGPIPE);
x = finish_command(&proc);
if (!ret)
ret = x;
strbuf_release(&data.buf);
return ret;
}