run-command: support custom fd-set in async
This patch adds the possibility to supply a set of non-0 file descriptors for async process communication instead of the default-created pipe. Additionally, we now support bi-directional communiction with the async procedure, by giving the async function both read and write file descriptors. To retain compatiblity and similar "API feel" with start_command, we require start_async callers to set .out = -1 to get a readable file descriptor. If either of .in or .out is 0, we supply no file descriptor to the async process. [sp: Note: Erik started this patch, and a huge bulk of it is his work. All bugs were introduced later by Shawn.] Signed-off-by: Erik Faye-Lund <kusmabite@gmail.com> Signed-off-by: Shawn O. Pearce <spearce@spearce.org> Signed-off-by: Junio C Hamano <gitster@pobox.com>
Erik Faye-Lund committed
Feb 5, 2010 at 12:57 UTC
ae6a5609c025d9ac79e54a3a052704e25d885314
7 files changed
+131
-37
Documentation/technical/api-run-command.txt
+40
-10
@@ -64,8 +64,8 @@ The functions above do the following:
64
`start_async`::
65
66
Run a function asynchronously. Takes a pointer to a `struct
67
- async` that specifies the details and returns a pipe FD
68
- from which the caller reads. See below for details.
67
+ async` that specifies the details and returns a set of pipe FDs
68
+ for communication with the function. See below for details.
69
70
`finish_async`::
71
@@ -180,17 +180,47 @@ The caller:
180
struct async variable;
181
2. initializes .proc and .data;
182
3. calls start_async();
183
-4. processes the data by reading from the fd in .out;
184
-5. closes .out;
183
+4. processes communicates with proc through .in and .out;
184
+5. closes .in and .out;
185
6. calls finish_async().
186
187
+The members .in, .out are used to provide a set of fd's for
188
+communication between the caller and the callee as follows:
189
+
190
+. Specify 0 to have no file descriptor passed. The callee will
191
+ receive -1 in the corresponding argument.
192
+
193
+. Specify < 0 to have a pipe allocated; start_async() replaces
194
+ with the pipe FD in the following way:
195
+
196
+ .in: Returns the writable pipe end into which the caller
197
+ writes; the readable end of the pipe becomes the function's
198
+ in argument.
199
+
200
+ .out: Returns the readable pipe end from which the caller
201
+ reads; the writable end of the pipe becomes the function's
202
+ out argument.
203
+
204
+ The caller of start_async() must close the returned FDs after it
205
+ has completed reading from/writing from them.
206
+
207
+. Specify a file descriptor > 0 to be used by the function:
208
+
209
+ .in: The FD must be readable; it becomes the function's in.
210
+ .out: The FD must be writable; it becomes the function's out.
211
+
212
+ The specified FD is closed by start_async(), even if it fails to
213
+ run the function.
214
+
215
The function pointer in .proc has the following signature:
216
189
- int proc(int fd, void *data);
217
+ int proc(int in, int out, void *data);
218
191
-. fd specifies a writable file descriptor to which the function must
192
- write the data that it produces. The function *must* close this
193
- descriptor before it returns.
219
+. in, out specifies a set of file descriptors to which the function
220
+ must read/write the data that it needs/produces. The function
221
+ *must* close these descriptors before it returns. A descriptor
222
+ may be -1 if the caller did not configure a descriptor for that
223
+ direction.
224
225
. data is the value that the caller has specified in the .data member
226
of struct async.
@@ -205,8 +235,8 @@ because this facility is implemented by a pipe to a forked process on
235
UNIX, but by a thread in the same address space on Windows:
236
237
. It cannot change the program's state (global variables, environment,
208
- etc.) in a way that the caller notices; in other words, .out is the
209
- only communication channel to the caller.
238
+ etc.) in a way that the caller notices; in other words, .in and .out
239
+ are the only communication channels to the caller.
240
241
. It must not change the program's state that the caller of the
242
facility also uses.
builtin-fetch-pack.c
+4
-3
@@ -586,12 +586,12 @@ static int everything_local(struct ref **refs, int nr_match, char **match)
586
return retval;
587
}
588
589
-static int sideband_demux(int fd, void *data)
589
+static int sideband_demux(int in, int out, void *data)
590
{
591
int *xd = data;
592
593
- int ret = recv_sideband("fetch-pack", xd[0], fd);
594
- close(fd);
593
+ int ret = recv_sideband("fetch-pack", xd[0], out);
594
+ close(out);
595
return ret;
596
}
597
@@ -613,6 +613,7 @@ static int get_pack(int xd[2], char **pack_lockfile)
613
*/
614
demux.proc = sideband_demux;
615
demux.data = xd;
616
+ demux.out = -1;
617
if (start_async(&demux))
618
die("fetch-pack: unable to fork off sideband"
619
" demultiplexer");
convert.c
+3
-2
@@ -241,7 +241,7 @@ struct filter_params {
241
const char *cmd;
242
};
243
244
-static int filter_buffer(int fd, void *data)
244
+static int filter_buffer(int in, int out, void *data)
245
{
246
/*
247
* Spawn cmd and feed the buffer contents through its stdin.
@@ -254,7 +254,7 @@ static int filter_buffer(int fd, void *data)
254
memset(&child_process, 0, sizeof(child_process));
255
child_process.argv = argv;
256
child_process.in = -1;
257
- child_process.out = fd;
257
+ child_process.out = out;
258
259
if (start_command(&child_process))
260
return error("cannot fork to run external filter %s", params->cmd);
@@ -291,6 +291,7 @@ static int apply_filter(const char *path, const char *src, size_t len,
291
memset(&async, 0, sizeof(async));
292
async.proc = filter_buffer;
293
async.data = ¶ms;
294
+ async.out = -1;
295
params.src = src;
296
params.size = len;
297
params.cmd = cmd;
remote-curl.c
+4
-3
@@ -184,13 +184,13 @@ static struct discovery* discover_refs(const char *service)
184
return last;
185
}
186
187
-static int write_discovery(int fd, void *data)
187
+static int write_discovery(int in, int out, void *data)
188
{
189
struct discovery *heads = data;
190
int err = 0;
191
- if (write_in_full(fd, heads->buf, heads->len) != heads->len)
191
+ if (write_in_full(out, heads->buf, heads->len) != heads->len)
192
err = 1;
193
- close(fd);
193
+ close(out);
194
return err;
195
}
196
@@ -202,6 +202,7 @@ static struct ref *parse_git_refs(struct discovery *heads)
202
memset(&async, 0, sizeof(async));
203
async.proc = write_discovery;
204
async.data = heads;
205
+ async.out = -1;
206
207
if (start_async(&async))
208
die("cannot start thread to parse advertised refs");
run-command.c
+70
-13
@@ -327,17 +327,51 @@ int run_command_v_opt_cd_env(const char **argv, int opt, const char *dir, const
327
static unsigned __stdcall run_thread(void *data)
328
{
329
struct async *async = data;
330
- return async->proc(async->fd_for_proc, async->data);
330
+ return async->proc(async->proc_in, async->proc_out, async->data);
331
}
332
#endif
333
334
int start_async(struct async *async)
335
{
336
- int pipe_out[2];
336
+ int need_in, need_out;
337
+ int fdin[2], fdout[2];
338
+ int proc_in, proc_out;
339
338
- if (pipe(pipe_out) < 0)
339
- return error("cannot create pipe: %s", strerror(errno));
340
- async->out = pipe_out[0];
340
+ need_in = async->in < 0;
341
+ if (need_in) {
342
+ if (pipe(fdin) < 0) {
343
+ if (async->out > 0)
344
+ close(async->out);
345
+ return error("cannot create pipe: %s", strerror(errno));
346
+ }
347
+ async->in = fdin[1];
348
+ }
349
+
350
+ need_out = async->out < 0;
351
+ if (need_out) {
352
+ if (pipe(fdout) < 0) {
353
+ if (need_in)
354
+ close_pair(fdin);
355
+ else if (async->in)
356
+ close(async->in);
357
+ return error("cannot create pipe: %s", strerror(errno));
358
+ }
359
+ async->out = fdout[0];
360
+ }
361
+
362
+ if (need_in)
363
+ proc_in = fdin[0];
364
+ else if (async->in)
365
+ proc_in = async->in;
366
+ else
367
+ proc_in = -1;
368
+
369
+ if (need_out)
370
+ proc_out = fdout[1];
371
+ else if (async->out)
372
+ proc_out = async->out;
373
+ else
374
+ proc_out = -1;
375
376
#ifndef WIN32
377
/* Flush stdio before fork() to avoid cloning buffers */
@@ -346,24 +380,47 @@ int start_async(struct async *async)
380
async->pid = fork();
381
if (async->pid < 0) {
382
error("fork (async) failed: %s", strerror(errno));
349
- close_pair(pipe_out);
350
- return -1;
383
+ goto error;
384
}
385
if (!async->pid) {
353
- close(pipe_out[0]);
354
- exit(!!async->proc(pipe_out[1], async->data));
386
+ if (need_in)
387
+ close(fdin[1]);
388
+ if (need_out)
389
+ close(fdout[0]);
390
+ exit(!!async->proc(proc_in, proc_out, async->data));
391
}
356
- close(pipe_out[1]);
392
+
393
+ if (need_in)
394
+ close(fdin[0]);
395
+ else if (async->in)
396
+ close(async->in);
397
+
398
+ if (need_out)
399
+ close(fdout[1]);
400
+ else if (async->out)
401
+ close(async->out);
402
#else
358
- async->fd_for_proc = pipe_out[1];
403
+ async->proc_in = proc_in;
404
+ async->proc_out = proc_out;
405
async->tid = (HANDLE) _beginthreadex(NULL, 0, run_thread, async, 0, NULL);
406
if (!async->tid) {
407
error("cannot create thread: %s", strerror(errno));
362
- close_pair(pipe_out);
363
- return -1;
408
+ goto error;
409
}
410
#endif
411
return 0;
412
+
413
+error:
414
+ if (need_in)
415
+ close_pair(fdin);
416
+ else if (async->in)
417
+ close(async->in);
418
+
419
+ if (need_out)
420
+ close_pair(fdout);
421
+ else if (async->out)
422
+ close(async->out);
423
+ return -1;
424
}
425
426
int finish_async(struct async *async)
run-command.h
+6
-3
@@ -64,17 +64,20 @@ int run_command_v_opt_cd_env(const char **argv, int opt, const char *dir, const
64
*/
65
struct async {
66
/*
67
- * proc writes to fd and closes it;
67
+ * proc reads from in; closes it before return
68
+ * proc writes to out; closes it before return
69
* returns 0 on success, non-zero on failure
70
*/
70
- int (*proc)(int fd, void *data);
71
+ int (*proc)(int in, int out, void *data);
72
void *data;
73
+ int in; /* caller writes here and closes it */
74
int out; /* caller reads from here and closes it */
75
#ifndef WIN32
76
pid_t pid;
77
#else
78
HANDLE tid;
77
- int fd_for_proc;
79
+ int proc_in;
80
+ int proc_out;
81
#endif
82
};
83
upload-pack.c
+4
-3
@@ -105,12 +105,12 @@ static void show_edge(struct commit *commit)
105
fprintf(pack_pipe, "-%s\n", sha1_to_hex(commit->object.sha1));
106
}
107
108
-static int do_rev_list(int fd, void *create_full_pack)
108
+static int do_rev_list(int in, int out, void *create_full_pack)
109
{
110
int i;
111
struct rev_info revs;
112
113
- pack_pipe = xfdopen(fd, "w");
113
+ pack_pipe = xfdopen(out, "w");
114
init_revisions(&revs, NULL);
115
revs.tag_objects = 1;
116
revs.tree_objects = 1;
@@ -162,8 +162,9 @@ static void create_pack_file(void)
162
int arg = 0;
163
164
if (shallow_nr) {
165
+ memset(&rev_list, 0, sizeof(rev_list));
166
rev_list.proc = do_rev_list;
166
- rev_list.data = 0;
167
+ rev_list.out = -1;
168
if (start_async(&rev_list))
169
die("git upload-pack: unable to fork git-rev-list");
170
argv[arg++] = "pack-objects";