procutil: avoid using os.fork() to implement runbgcommand
We ran into the following deadlock:
- some command creates an ssh peer, then raises without explicitly
closing the peer (hg id + extension in our case)
- dispatch catches the exception, calls ui.log('commandfinish', ..)
(the sshpeer is still not closed), which calls logtoprocess, which
calls procutil.runbgcommand.
- in the child of runbgcommand's fork(), between the fork and the
exec, the opening of file descriptors triggers a gc which runs the
destructor for sshpeer, which waits on ssh's stderr being closed,
which never happens since ssh's stderr is held open by the parent of
the fork where said destructor hasn't run
Remotefilelog appears to have a hack around this deadlock as well.
I don't know if there's more subtlety to it, because even though the
problem is determistic, it is very fragile, so I didn't manage to
reduce it.
I can imagine three ways of tackling this problem:
1. don't run any python between fork and exec in runbgcommand
2. make the finalizer harmless after the fork
3. close the peer without relying on gc behavior
This commit goes with 1, as forking without exec'ing is tricky in
general in a language with gc finalizers. And maybe it's better in the
presence of rust threads. A future commit will try 2 or 3.
Performance wise: at low memory usage, it's an improvement. At higher
memory usage, it's about 2x faster than before when ensurestart=True,
but 2x slower when ensurestart=False. Not sure if that matters. The
reason for that last bit is that the subprocess.Popen always waits for
the execve to finish, and at high memory usage, execve is slow because
it deallocates the large page table. Numbers and script:
before after
mem=1.0GB, ensurestart=True 52.1ms 26.0ms
mem=1.0GB, ensurestart=False 14.7ms 26.0ms
mem=0.5GB, ensurestart=True 23.2ms 11.2ms
mem=0.5GB, ensurestart=False 6.2ms 11.3ms
mem=0.2GB, ensurestart=True 15.7ms 7.4ms
mem=0.2GB, ensurestart=False 4.3ms 8.1ms
mem=0.0GB, ensurestart=True 2.3ms 0.7ms
mem=0.0GB, ensurestart=False 0.8ms 0.8ms
import time
for memsize in [1_000_000_000, 500_000_000, 250_000_000, 0]:
mem = 'a' * memsize
for ensurestart in [True, False]:
now = time.time()
n = 100
for i in range(n):
procutil.runbgcommand([b'true'], {}, ensurestart=ensurestart)
after = time.time()
ms = (after - now) / float(n) * 1000
print(f'mem={memsize / 1e9:.1f}GB, ensurestart={ensurestart} -> {ms:.1f}ms')
Differential Revision: https://phab.mercurial-scm.org/D9019
/*
* A command server client that uses Unix domain socket
*
* Copyright (c) 2011 Yuya Nishihara <yuya@tcha.org>
*
* This software may be used and distributed according to the terms of the
* GNU General Public License version 2 or any later version.
*/
#include <arpa/inet.h> /* for ntohl(), htonl() */
#include <assert.h>
#include <ctype.h>
#include <errno.h>
#include <fcntl.h>
#include <signal.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/un.h>
#include <unistd.h>
#include "hgclient.h"
#include "procutil.h"
#include "util.h"
enum {
CAP_GETENCODING = 0x0001,
CAP_RUNCOMMAND = 0x0002,
/* cHg extension: */
CAP_ATTACHIO = 0x0100,
CAP_CHDIR = 0x0200,
CAP_SETENV = 0x0800,
CAP_SETUMASK2 = 0x1000,
CAP_VALIDATE = 0x2000,
CAP_SETPROCNAME = 0x4000,
};
typedef struct {
const char *name;
unsigned int flag;
} cappair_t;
static const cappair_t captable[] = {
{"getencoding", CAP_GETENCODING},
{"runcommand", CAP_RUNCOMMAND},
{"attachio", CAP_ATTACHIO},
{"chdir", CAP_CHDIR},
{"setenv", CAP_SETENV},
{"setumask2", CAP_SETUMASK2},
{"validate", CAP_VALIDATE},
{"setprocname", CAP_SETPROCNAME},
{NULL, 0}, /* terminator */
};
typedef struct {
char ch;
char *data;
size_t maxdatasize;
size_t datasize;
} context_t;
struct hgclient_tag_ {
int sockfd;
pid_t pgid;
pid_t pid;
context_t ctx;
unsigned int capflags;
};
static const size_t defaultdatasize = 4096;
static void attachio(hgclient_t *hgc);
static void initcontext(context_t *ctx)
{
ctx->ch = '\0';
ctx->data = malloc(defaultdatasize);
ctx->maxdatasize = (ctx->data) ? defaultdatasize : 0;
ctx->datasize = 0;
debugmsg("initialize context buffer with size %zu", ctx->maxdatasize);
}
static void enlargecontext(context_t *ctx, size_t newsize)
{
if (newsize <= ctx->maxdatasize) {
return;
}
newsize = defaultdatasize *
((newsize + defaultdatasize - 1) / defaultdatasize);
ctx->data = reallocx(ctx->data, newsize);
ctx->maxdatasize = newsize;
debugmsg("enlarge context buffer to %zu", ctx->maxdatasize);
}
static void freecontext(context_t *ctx)
{
debugmsg("free context buffer");
free(ctx->data);
ctx->data = NULL;
ctx->maxdatasize = 0;
ctx->datasize = 0;
}
/* Read channeled response from cmdserver */
static void readchannel(hgclient_t *hgc)
{
assert(hgc);
ssize_t rsize = recv(hgc->sockfd, &hgc->ctx.ch, sizeof(hgc->ctx.ch), 0);
if (rsize != sizeof(hgc->ctx.ch)) {
/* server would have exception and traceback would be printed */
debugmsg("failed to read channel");
exit(255);
}
uint32_t datasize_n;
rsize = recv(hgc->sockfd, &datasize_n, sizeof(datasize_n), 0);
if (rsize != sizeof(datasize_n)) {
abortmsg("failed to read data size");
}
/* datasize denotes the maximum size to write if input request */
hgc->ctx.datasize = ntohl(datasize_n);
enlargecontext(&hgc->ctx, hgc->ctx.datasize);
if (isupper(hgc->ctx.ch) && hgc->ctx.ch != 'S') {
return; /* assumes input request */
}
size_t cursize = 0;
while (cursize < hgc->ctx.datasize) {
rsize = recv(hgc->sockfd, hgc->ctx.data + cursize,
hgc->ctx.datasize - cursize, 0);
if (rsize < 1) {
abortmsg("failed to read data block");
}
cursize += rsize;
}
}
static void sendall(int sockfd, const void *data, size_t datasize)
{
const char *p = data;
const char *const endp = p + datasize;
while (p < endp) {
ssize_t r = send(sockfd, p, endp - p, 0);
if (r < 0) {
abortmsgerrno("cannot communicate");
}
p += r;
}
}
/* Write lengh-data block to cmdserver */
static void writeblock(const hgclient_t *hgc)
{
assert(hgc);
const uint32_t datasize_n = htonl(hgc->ctx.datasize);
sendall(hgc->sockfd, &datasize_n, sizeof(datasize_n));
sendall(hgc->sockfd, hgc->ctx.data, hgc->ctx.datasize);
}
static void writeblockrequest(const hgclient_t *hgc, const char *chcmd)
{
debugmsg("request %s, block size %zu", chcmd, hgc->ctx.datasize);
char buf[strlen(chcmd) + 1];
memcpy(buf, chcmd, sizeof(buf) - 1);
buf[sizeof(buf) - 1] = '\n';
sendall(hgc->sockfd, buf, sizeof(buf));
writeblock(hgc);
}
/* Build '\0'-separated list of args. argsize < 0 denotes that args are
* terminated by NULL. */
static void packcmdargs(context_t *ctx, const char *const args[],
ssize_t argsize)
{
ctx->datasize = 0;
const char *const *const end = (argsize >= 0) ? args + argsize : NULL;
for (const char *const *it = args; it != end && *it; ++it) {
const size_t n = strlen(*it) + 1; /* include '\0' */
enlargecontext(ctx, ctx->datasize + n);
memcpy(ctx->data + ctx->datasize, *it, n);
ctx->datasize += n;
}
if (ctx->datasize > 0) {
--ctx->datasize; /* strip last '\0' */
}
}
/* Extract '\0'-separated list of args to new buffer, terminated by NULL */
static const char **unpackcmdargsnul(const context_t *ctx)
{
const char **args = NULL;
size_t nargs = 0, maxnargs = 0;
const char *s = ctx->data;
const char *e = ctx->data + ctx->datasize;
for (;;) {
if (nargs + 1 >= maxnargs) { /* including last NULL */
maxnargs += 256;
args = reallocx(args, maxnargs * sizeof(args[0]));
}
args[nargs] = s;
nargs++;
s = memchr(s, '\0', e - s);
if (!s) {
break;
}
s++;
}
args[nargs] = NULL;
return args;
}
static void handlereadrequest(hgclient_t *hgc)
{
context_t *ctx = &hgc->ctx;
size_t r = fread(ctx->data, sizeof(ctx->data[0]), ctx->datasize, stdin);
ctx->datasize = r;
writeblock(hgc);
}
/* Read single-line */
static void handlereadlinerequest(hgclient_t *hgc)
{
context_t *ctx = &hgc->ctx;
if (!fgets(ctx->data, ctx->datasize, stdin)) {
ctx->data[0] = '\0';
}
ctx->datasize = strlen(ctx->data);
writeblock(hgc);
}
/* Execute the requested command and write exit code */
static void handlesystemrequest(hgclient_t *hgc)
{
context_t *ctx = &hgc->ctx;
enlargecontext(ctx, ctx->datasize + 1);
ctx->data[ctx->datasize] = '\0'; /* terminate last string */
const char **args = unpackcmdargsnul(ctx);
if (!args[0] || !args[1] || !args[2]) {
abortmsg("missing type or command or cwd in system request");
}
if (strcmp(args[0], "system") == 0) {
debugmsg("run '%s' at '%s'", args[1], args[2]);
int32_t r = runshellcmd(args[1], args + 3, args[2]);
free(args);
uint32_t r_n = htonl(r);
memcpy(ctx->data, &r_n, sizeof(r_n));
ctx->datasize = sizeof(r_n);
writeblock(hgc);
} else if (strcmp(args[0], "pager") == 0) {
setuppager(args[1], args + 3);
if (hgc->capflags & CAP_ATTACHIO) {
attachio(hgc);
}
/* unblock the server */
static const char emptycmd[] = "\n";
sendall(hgc->sockfd, emptycmd, sizeof(emptycmd) - 1);
} else {
abortmsg("unknown type in system request: %s", args[0]);
}
}
/* Read response of command execution until receiving 'r'-esult */
static void handleresponse(hgclient_t *hgc)
{
for (;;) {
readchannel(hgc);
context_t *ctx = &hgc->ctx;
debugmsg("response read from channel %c, size %zu", ctx->ch,
ctx->datasize);
switch (ctx->ch) {
case 'o':
fwrite(ctx->data, sizeof(ctx->data[0]), ctx->datasize,
stdout);
break;
case 'e':
fwrite(ctx->data, sizeof(ctx->data[0]), ctx->datasize,
stderr);
break;
case 'd':
/* assumes last char is '\n' */
ctx->data[ctx->datasize - 1] = '\0';
debugmsg("server: %s", ctx->data);
break;
case 'r':
return;
case 'I':
handlereadrequest(hgc);
break;
case 'L':
handlereadlinerequest(hgc);
break;
case 'S':
handlesystemrequest(hgc);
break;
default:
if (isupper(ctx->ch)) {
abortmsg("cannot handle response (ch = %c)",
ctx->ch);
}
}
}
}
static unsigned int parsecapabilities(const char *s, const char *e)
{
unsigned int flags = 0;
while (s < e) {
const char *t = strchr(s, ' ');
if (!t || t > e) {
t = e;
}
const cappair_t *cap;
for (cap = captable; cap->flag; ++cap) {
size_t n = t - s;
if (strncmp(s, cap->name, n) == 0 &&
strlen(cap->name) == n) {
flags |= cap->flag;
break;
}
}
s = t + 1;
}
return flags;
}
static void readhello(hgclient_t *hgc)
{
readchannel(hgc);
context_t *ctx = &hgc->ctx;
if (ctx->ch != 'o') {
char ch = ctx->ch;
if (ch == 'e') {
/* write early error and will exit */
fwrite(ctx->data, sizeof(ctx->data[0]), ctx->datasize,
stderr);
handleresponse(hgc);
}
abortmsg("unexpected channel of hello message (ch = %c)", ch);
}
enlargecontext(ctx, ctx->datasize + 1);
ctx->data[ctx->datasize] = '\0';
debugmsg("hello received: %s (size = %zu)", ctx->data, ctx->datasize);
const char *s = ctx->data;
const char *const dataend = ctx->data + ctx->datasize;
while (s < dataend) {
const char *t = strchr(s, ':');
if (!t || t[1] != ' ') {
break;
}
const char *u = strchr(t + 2, '\n');
if (!u) {
u = dataend;
}
if (strncmp(s, "capabilities:", t - s + 1) == 0) {
hgc->capflags = parsecapabilities(t + 2, u);
} else if (strncmp(s, "pgid:", t - s + 1) == 0) {
hgc->pgid = strtol(t + 2, NULL, 10);
} else if (strncmp(s, "pid:", t - s + 1) == 0) {
hgc->pid = strtol(t + 2, NULL, 10);
}
s = u + 1;
}
debugmsg("capflags=0x%04x, pid=%d", hgc->capflags, hgc->pid);
}
static void updateprocname(hgclient_t *hgc)
{
int r = snprintf(hgc->ctx.data, hgc->ctx.maxdatasize, "chg[worker/%d]",
(int)getpid());
if (r < 0 || (size_t)r >= hgc->ctx.maxdatasize) {
abortmsg("insufficient buffer to write procname (r = %d)", r);
}
hgc->ctx.datasize = (size_t)r;
writeblockrequest(hgc, "setprocname");
}
static void attachio(hgclient_t *hgc)
{
debugmsg("request attachio");
static const char chcmd[] = "attachio\n";
sendall(hgc->sockfd, chcmd, sizeof(chcmd) - 1);
readchannel(hgc);
context_t *ctx = &hgc->ctx;
if (ctx->ch != 'I') {
abortmsg("unexpected response for attachio (ch = %c)", ctx->ch);
}
static const int fds[3] = {STDIN_FILENO, STDOUT_FILENO, STDERR_FILENO};
struct msghdr msgh;
memset(&msgh, 0, sizeof(msgh));
struct iovec iov = {ctx->data, ctx->datasize}; /* dummy payload */
msgh.msg_iov = &iov;
msgh.msg_iovlen = 1;
char fdbuf[CMSG_SPACE(sizeof(fds))];
msgh.msg_control = fdbuf;
msgh.msg_controllen = sizeof(fdbuf);
struct cmsghdr *cmsg = CMSG_FIRSTHDR(&msgh);
cmsg->cmsg_level = SOL_SOCKET;
cmsg->cmsg_type = SCM_RIGHTS;
cmsg->cmsg_len = CMSG_LEN(sizeof(fds));
memcpy(CMSG_DATA(cmsg), fds, sizeof(fds));
msgh.msg_controllen = cmsg->cmsg_len;
ssize_t r = sendmsg(hgc->sockfd, &msgh, 0);
if (r < 0) {
abortmsgerrno("sendmsg failed");
}
handleresponse(hgc);
int32_t n;
if (ctx->datasize != sizeof(n)) {
abortmsg("unexpected size of attachio result");
}
memcpy(&n, ctx->data, sizeof(n));
n = ntohl(n);
if (n != sizeof(fds) / sizeof(fds[0])) {
abortmsg("failed to send fds (n = %d)", n);
}
}
static void chdirtocwd(hgclient_t *hgc)
{
if (!getcwd(hgc->ctx.data, hgc->ctx.maxdatasize)) {
abortmsgerrno("failed to getcwd");
}
hgc->ctx.datasize = strlen(hgc->ctx.data);
writeblockrequest(hgc, "chdir");
}
static void forwardumask(hgclient_t *hgc)
{
mode_t mask = umask(0);
umask(mask);
uint32_t data = htonl(mask);
enlargecontext(&hgc->ctx, sizeof(data));
memcpy(hgc->ctx.data, &data, sizeof(data));
hgc->ctx.datasize = sizeof(data);
writeblockrequest(hgc, "setumask2");
}
/*!
* Open connection to per-user cmdserver
*
* If no background server running, returns NULL.
*/
hgclient_t *hgc_open(const char *sockname)
{
int fd = socket(AF_UNIX, SOCK_STREAM, 0);
if (fd < 0) {
abortmsgerrno("cannot create socket");
}
/* don't keep fd on fork(), so that it can be closed when the parent
* process get terminated. */
fsetcloexec(fd);
struct sockaddr_un addr;
addr.sun_family = AF_UNIX;
/* use chdir to workaround small sizeof(sun_path) */
int bakfd = -1;
const char *basename = sockname;
{
const char *split = strrchr(sockname, '/');
if (split && split != sockname) {
if (split[1] == '\0') {
abortmsg("sockname cannot end with a slash");
}
size_t len = split - sockname;
char sockdir[len + 1];
memcpy(sockdir, sockname, len);
sockdir[len] = '\0';
bakfd = open(".", O_DIRECTORY);
if (bakfd == -1) {
abortmsgerrno("cannot open cwd");
}
int r = chdir(sockdir);
if (r != 0) {
abortmsgerrno("cannot chdir %s", sockdir);
}
basename = split + 1;
}
}
if (strlen(basename) >= sizeof(addr.sun_path)) {
abortmsg("sockname is too long: %s", basename);
}
strncpy(addr.sun_path, basename, sizeof(addr.sun_path));
addr.sun_path[sizeof(addr.sun_path) - 1] = '\0';
/* real connect */
int r = connect(fd, (struct sockaddr *)&addr, sizeof(addr));
if (r < 0) {
if (errno != ENOENT && errno != ECONNREFUSED) {
abortmsgerrno("cannot connect to %s", sockname);
}
}
if (bakfd != -1) {
fchdirx(bakfd);
close(bakfd);
}
if (r < 0) {
close(fd);
return NULL;
}
debugmsg("connected to %s", addr.sun_path);
hgclient_t *hgc = mallocx(sizeof(hgclient_t));
memset(hgc, 0, sizeof(*hgc));
hgc->sockfd = fd;
initcontext(&hgc->ctx);
readhello(hgc);
if (!(hgc->capflags & CAP_RUNCOMMAND)) {
abortmsg("insufficient capability: runcommand");
}
if (hgc->capflags & CAP_SETPROCNAME) {
updateprocname(hgc);
}
if (hgc->capflags & CAP_ATTACHIO) {
attachio(hgc);
}
if (hgc->capflags & CAP_CHDIR) {
chdirtocwd(hgc);
}
if (hgc->capflags & CAP_SETUMASK2) {
forwardumask(hgc);
}
return hgc;
}
/*!
* Close connection and free allocated memory
*/
void hgc_close(hgclient_t *hgc)
{
assert(hgc);
freecontext(&hgc->ctx);
close(hgc->sockfd);
free(hgc);
}
pid_t hgc_peerpgid(const hgclient_t *hgc)
{
assert(hgc);
return hgc->pgid;
}
pid_t hgc_peerpid(const hgclient_t *hgc)
{
assert(hgc);
return hgc->pid;
}
/*!
* Send command line arguments to let the server load the repo config and check
* whether it can process our request directly or not.
* Make sure hgc_setenv is called before calling this.
*
* @return - NULL, the server believes it can handle our request, or does not
* support "validate" command.
* - a list of strings, the server probably cannot handle our request
* and it sent instructions telling us what to do next. See
* chgserver.py for possible instruction formats.
* the list should be freed by the caller.
* the last string is guaranteed to be NULL.
*/
const char **hgc_validate(hgclient_t *hgc, const char *const args[],
size_t argsize)
{
assert(hgc);
if (!(hgc->capflags & CAP_VALIDATE)) {
return NULL;
}
packcmdargs(&hgc->ctx, args, argsize);
writeblockrequest(hgc, "validate");
handleresponse(hgc);
/* the server returns '\0' if it can handle our request */
if (hgc->ctx.datasize <= 1) {
return NULL;
}
/* make sure the buffer is '\0' terminated */
enlargecontext(&hgc->ctx, hgc->ctx.datasize + 1);
hgc->ctx.data[hgc->ctx.datasize] = '\0';
return unpackcmdargsnul(&hgc->ctx);
}
/*!
* Execute the specified Mercurial command
*
* @return result code
*/
int hgc_runcommand(hgclient_t *hgc, const char *const args[], size_t argsize)
{
assert(hgc);
packcmdargs(&hgc->ctx, args, argsize);
writeblockrequest(hgc, "runcommand");
handleresponse(hgc);
int32_t exitcode_n;
if (hgc->ctx.datasize != sizeof(exitcode_n)) {
abortmsg("unexpected size of exitcode");
}
memcpy(&exitcode_n, hgc->ctx.data, sizeof(exitcode_n));
return ntohl(exitcode_n);
}
/*!
* (Re-)send client's stdio channels so that the server can access to tty
*/
void hgc_attachio(hgclient_t *hgc)
{
assert(hgc);
if (!(hgc->capflags & CAP_ATTACHIO)) {
return;
}
attachio(hgc);
}
/*!
* Update server's environment variables
*
* @param envp list of environment variables in "NAME=VALUE" format,
* terminated by NULL.
*/
void hgc_setenv(hgclient_t *hgc, const char *const envp[])
{
assert(hgc && envp);
if (!(hgc->capflags & CAP_SETENV)) {
return;
}
packcmdargs(&hgc->ctx, envp, /*argsize*/ -1);
writeblockrequest(hgc, "setenv");
}