commandserver.py
727 lines
| 22.6 KiB
| text/x-python
|
PythonLexer
/ mercurial / commandserver.py
|
r14647 | # commandserver.py - communicate with Mercurial's API over a pipe | ||
# | ||||
# Copyright Matt Mackall <mpm@selenic.com> | ||||
# | ||||
# This software may be used and distributed according to the terms of the | ||||
# GNU General Public License version 2 or any later version. | ||||
|
r27351 | from __future__ import absolute_import | ||
import errno | ||||
|
r29513 | import gc | ||
|
r27351 | import os | ||
|
r29513 | import random | ||
|
r29544 | import signal | ||
import socket | ||||
|
r14647 | import struct | ||
|
r27351 | import traceback | ||
|
r36958 | try: | ||
import selectors | ||||
|
r43346 | |||
|
r36958 | selectors.BaseSelector | ||
except ImportError: | ||||
from .thirdparty import selectors2 as selectors | ||||
|
r27351 | from .i18n import _ | ||
|
r43359 | from .pycompat import getattr | ||
|
r27351 | from . import ( | ||
encoding, | ||||
error, | ||||
|
r40859 | loggingutil, | ||
|
r40625 | pycompat, | ||
|
r41035 | repocache, | ||
|
r27351 | util, | ||
|
r40859 | vfs as vfsmod, | ||
|
r27351 | ) | ||
|
r37137 | from .utils import ( | ||
|
r40625 | cborutil, | ||
|
r37137 | procutil, | ||
) | ||||
|
r14647 | |||
|
r43346 | |||
|
r14647 | class channeledoutput(object): | ||
""" | ||||
|
r22561 | Write data to out in the following format: | ||
|
r14647 | |||
data length (unsigned int), | ||||
data | ||||
""" | ||||
|
r43346 | |||
|
r22563 | def __init__(self, out, channel): | ||
|
r14647 | self.out = out | ||
self.channel = channel | ||||
|
r27415 | @property | ||
def name(self): | ||||
|
r43347 | return b'<%c-channel>' % self.channel | ||
|
r27415 | |||
|
r14647 | def write(self, data): | ||
if not data: | ||||
return | ||||
|
r30263 | # single write() to guarantee the same atomicity as the underlying file | ||
|
r43347 | self.out.write(struct.pack(b'>cI', self.channel, len(data)) + data) | ||
|
r14647 | self.out.flush() | ||
def __getattr__(self, attr): | ||||
|
r43906 | if attr in ('isatty', 'fileno', 'tell', 'seek'): | ||
|
r18174 | raise AttributeError(attr) | ||
|
r22563 | return getattr(self.out, attr) | ||
|
r14647 | |||
|
r43346 | |||
|
r40625 | class channeledmessage(object): | ||
""" | ||||
Write encoded message and metadata to out in the following format: | ||||
data length (unsigned int), | ||||
encoded message and metadata, as a flat key-value dict. | ||||
|
r40626 | |||
Each message should have 'type' attribute. Messages of unknown type | ||||
should be ignored. | ||||
|
r40625 | """ | ||
# teach ui that write() can take **opts | ||||
structured = True | ||||
def __init__(self, out, channel, encodename, encodefn): | ||||
self._cout = channeledoutput(out, channel) | ||||
self.encoding = encodename | ||||
self._encodefn = encodefn | ||||
def write(self, data, **opts): | ||||
opts = pycompat.byteskwargs(opts) | ||||
|
r40630 | if data is not None: | ||
opts[b'data'] = data | ||||
|
r40625 | self._cout.write(self._encodefn(opts)) | ||
def __getattr__(self, attr): | ||||
return getattr(self._cout, attr) | ||||
|
r43346 | |||
|
r14647 | class channeledinput(object): | ||
""" | ||||
Read data from in_. | ||||
Requests for input are written to out in the following format: | ||||
channel identifier - 'I' for plain input, 'L' line based (1 byte) | ||||
how many bytes to send at most (unsigned int), | ||||
The client replies with: | ||||
data length (unsigned int), 0 meaning EOF | ||||
data | ||||
""" | ||||
maxchunksize = 4 * 1024 | ||||
def __init__(self, in_, out, channel): | ||||
self.in_ = in_ | ||||
self.out = out | ||||
self.channel = channel | ||||
|
r27415 | @property | ||
def name(self): | ||||
|
r43347 | return b'<%c-channel>' % self.channel | ||
|
r27415 | |||
|
r14647 | def read(self, size=-1): | ||
if size < 0: | ||||
# if we need to consume all the clients input, ask for 4k chunks | ||||
# so the pipe doesn't fill up risking a deadlock | ||||
size = self.maxchunksize | ||||
s = self._read(size, self.channel) | ||||
buf = s | ||||
while s: | ||||
|
r14728 | s = self._read(size, self.channel) | ||
|
r14647 | buf += s | ||
return buf | ||||
else: | ||||
return self._read(size, self.channel) | ||||
def _read(self, size, channel): | ||||
if not size: | ||||
|
r43347 | return b'' | ||
|
r14647 | assert size > 0 | ||
# tell the client we need at most size bytes | ||||
|
r43347 | self.out.write(struct.pack(b'>cI', channel, size)) | ||
|
r14647 | self.out.flush() | ||
length = self.in_.read(4) | ||||
|
r43347 | length = struct.unpack(b'>I', length)[0] | ||
|
r14647 | if not length: | ||
|
r43347 | return b'' | ||
|
r14647 | else: | ||
return self.in_.read(length) | ||||
def readline(self, size=-1): | ||||
if size < 0: | ||||
size = self.maxchunksize | ||||
|
r43347 | s = self._read(size, b'L') | ||
|
r14647 | buf = s | ||
# keep asking for more until there's either no more or | ||||
# we got a full line | ||||
|
r44857 | while s and not s.endswith(b'\n'): | ||
|
r43347 | s = self._read(size, b'L') | ||
|
r14647 | buf += s | ||
return buf | ||||
else: | ||||
|
r43347 | return self._read(size, b'L') | ||
|
r14647 | |||
def __iter__(self): | ||||
return self | ||||
def next(self): | ||||
l = self.readline() | ||||
if not l: | ||||
raise StopIteration | ||||
return l | ||||
|
r40394 | __next__ = next | ||
|
r14647 | def __getattr__(self, attr): | ||
|
r43906 | if attr in ('isatty', 'fileno', 'tell', 'seek'): | ||
|
r18174 | raise AttributeError(attr) | ||
|
r14647 | return getattr(self.in_, attr) | ||
|
r43346 | |||
|
r40625 | _messageencoders = { | ||
b'cbor': lambda v: b''.join(cborutil.streamencode(v)), | ||||
} | ||||
|
r43346 | |||
|
r40625 | def _selectmessageencoder(ui): | ||
# experimental config: cmdserver.message-encodings | ||||
encnames = ui.configlist(b'cmdserver', b'message-encodings') | ||||
for n in encnames: | ||||
f = _messageencoders.get(n) | ||||
if f: | ||||
return n, f | ||||
|
r43346 | raise error.Abort( | ||
b'no supported message encodings: %s' % b' '.join(encnames) | ||||
) | ||||
|
r40625 | |||
|
r14647 | class server(object): | ||
""" | ||||
|
r22990 | Listens for commands on fin, runs them and writes the output on a channel | ||
based stream to fout. | ||||
|
r14647 | """ | ||
|
r43346 | |||
|
r40911 | def __init__(self, ui, repo, fin, fout, prereposetups=None): | ||
|
r39843 | self.cwd = encoding.getcwd() | ||
|
r14647 | |||
|
r20650 | if repo: | ||
# the ui here is really the repo ui so take its baseui so we don't | ||||
# end up with its local configuration | ||||
self.ui = repo.baseui | ||||
self.repo = repo | ||||
self.repoui = repo.ui | ||||
else: | ||||
self.ui = ui | ||||
self.repo = self.repoui = None | ||||
|
r40911 | self._prereposetups = prereposetups | ||
|
r14647 | |||
|
r43347 | self.cdebug = channeledoutput(fout, b'd') | ||
self.cerr = channeledoutput(fout, b'e') | ||||
self.cout = channeledoutput(fout, b'o') | ||||
self.cin = channeledinput(fin, fout, b'I') | ||||
self.cresult = channeledoutput(fout, b'r') | ||||
|
r14647 | |||
|
r40859 | if self.ui.config(b'cmdserver', b'log') == b'-': | ||
# switch log stream of server's ui to the 'd' (debug) channel | ||||
# (don't touch repo.ui as its lifetime is longer than the server) | ||||
self.ui = self.ui.copy() | ||||
setuplogging(self.ui, repo=None, fp=self.cdebug) | ||||
|
r40625 | # TODO: add this to help/config.txt when stabilized | ||
# ``channel`` | ||||
# Use separate channel for structured output. (Command-server only) | ||||
self.cmsg = None | ||||
if ui.config(b'ui', b'message-output') == b'channel': | ||||
encname, encfn = _selectmessageencoder(ui) | ||||
self.cmsg = channeledmessage(fout, b'm', encname, encfn) | ||||
|
r22990 | self.client = fin | ||
|
r14647 | |||
|
r29512 | def cleanup(self): | ||
"""release and restore resources taken during server session""" | ||||
|
r14647 | def _read(self, size): | ||
|
r14706 | if not size: | ||
|
r43347 | return b'' | ||
|
r14706 | |||
|
r14647 | data = self.client.read(size) | ||
# is the other end closed? | ||||
if not data: | ||||
|
r16687 | raise EOFError | ||
|
r14647 | |||
return data | ||||
|
r28156 | def _readstr(self): | ||
"""read a string from the channel | ||||
format: | ||||
data length (uint32), data | ||||
""" | ||||
|
r43347 | length = struct.unpack(b'>I', self._read(4))[0] | ||
|
r28156 | if not length: | ||
|
r43347 | return b'' | ||
|
r28156 | return self._read(length) | ||
def _readlist(self): | ||||
"""read a list of NULL separated strings from the channel""" | ||||
s = self._readstr() | ||||
if s: | ||||
|
r43347 | return s.split(b'\0') | ||
|
r28156 | else: | ||
return [] | ||||
|
r14647 | def runcommand(self): | ||
""" reads a list of \0 terminated arguments, executes | ||||
and writes the return code to the result channel """ | ||||
|
r27352 | from . import dispatch # avoid cycle | ||
|
r14647 | |||
|
r28157 | args = self._readlist() | ||
|
r14647 | |||
|
r14750 | # copy the uis so changes (e.g. --config or --verbose) don't | ||
# persist between requests | ||||
|
r14751 | copiedui = self.ui.copy() | ||
|
r21195 | uis = [copiedui] | ||
|
r20650 | if self.repo: | ||
self.repo.baseui = copiedui | ||||
# clone ui without using ui.copy because this is protected | ||||
repoui = self.repoui.__class__(self.repoui) | ||||
|
r43346 | repoui.copy = copiedui.copy # redo copy protection | ||
|
r21195 | uis.append(repoui) | ||
|
r20650 | self.repo.ui = self.repo.dirstate._ui = repoui | ||
self.repo.invalidateall() | ||||
|
r14751 | |||
|
r21195 | for ui in uis: | ||
|
r29366 | ui.resetstate() | ||
|
r27565 | # any kind of interaction must use server channels, but chg may | ||
# replace channels by fully functional tty files. so nontty is | ||||
# enforced only if cin is a channel. | ||||
|
r43347 | if not util.safehasattr(self.cin, b'fileno'): | ||
ui.setconfig(b'ui', b'nontty', b'true', b'commandserver') | ||||
|
r21195 | |||
|
r43346 | req = dispatch.request( | ||
args[:], | ||||
copiedui, | ||||
self.repo, | ||||
self.cin, | ||||
self.cout, | ||||
self.cerr, | ||||
self.cmsg, | ||||
prereposetups=self._prereposetups, | ||||
) | ||||
|
r14647 | |||
|
r35670 | try: | ||
|
r38015 | ret = dispatch.dispatch(req) & 255 | ||
|
r43347 | self.cresult.write(struct.pack(b'>i', int(ret))) | ||
|
r35670 | finally: | ||
# restore old cwd | ||||
|
r43347 | if b'--cwd' in args: | ||
|
r35670 | os.chdir(self.cwd) | ||
|
r14647 | |||
def getencoding(self): | ||||
""" writes the current encoding to the result channel """ | ||||
self.cresult.write(encoding.encoding) | ||||
def serveone(self): | ||||
cmd = self.client.readline()[:-1] | ||||
if cmd: | ||||
handler = self.capabilities.get(cmd) | ||||
if handler: | ||||
handler(self) | ||||
else: | ||||
# clients are expected to check what commands are supported by | ||||
# looking at the servers capabilities | ||||
|
r43347 | raise error.Abort(_(b'unknown command %s') % cmd) | ||
|
r14647 | |||
|
r43347 | return cmd != b'' | ||
|
r14647 | |||
|
r43347 | capabilities = {b'runcommand': runcommand, b'getencoding': getencoding} | ||
|
r14647 | |||
def serve(self): | ||||
|
r43347 | hellomsg = b'capabilities: ' + b' '.join(sorted(self.capabilities)) | ||
hellomsg += b'\n' | ||||
hellomsg += b'encoding: ' + encoding.encoding | ||||
hellomsg += b'\n' | ||||
|
r40625 | if self.cmsg: | ||
|
r43347 | hellomsg += b'message-encoding: %s\n' % self.cmsg.encoding | ||
hellomsg += b'pid: %d' % procutil.getpid() | ||||
if util.safehasattr(os, b'getpgid'): | ||||
hellomsg += b'\n' | ||||
hellomsg += b'pgid: %d' % os.getpgid(0) | ||||
|
r14719 | |||
# write the hello msg in -one- chunk | ||||
self.cout.write(hellomsg) | ||||
|
r14647 | |||
try: | ||||
while self.serveone(): | ||||
pass | ||||
except EOFError: | ||||
# we'll get here if the client disconnected while we were reading | ||||
# its request | ||||
return 1 | ||||
return 0 | ||||
|
r22988 | |||
|
r43346 | |||
|
r40859 | def setuplogging(ui, repo=None, fp=None): | ||
|
r40858 | """Set up server logging facility | ||
|
r40859 | If cmdserver.log is '-', log messages will be sent to the given fp. | ||
It should be the 'd' channel while a client is connected, and otherwise | ||||
is the stderr of the server process. | ||||
|
r40858 | """ | ||
# developer config: cmdserver.log | ||||
logpath = ui.config(b'cmdserver', b'log') | ||||
if not logpath: | ||||
return | ||||
|
r40862 | # developer config: cmdserver.track-log | ||
tracked = set(ui.configlist(b'cmdserver', b'track-log')) | ||||
|
r40858 | |||
|
r40859 | if logpath == b'-' and fp: | ||
logger = loggingutil.fileobjectlogger(fp, tracked) | ||||
elif logpath == b'-': | ||||
logger = loggingutil.fileobjectlogger(ui.ferr, tracked) | ||||
else: | ||||
|
r40861 | logpath = os.path.abspath(util.expandpath(logpath)) | ||
|
r40862 | # developer config: cmdserver.max-log-files | ||
maxfiles = ui.configint(b'cmdserver', b'max-log-files') | ||||
# developer config: cmdserver.max-log-size | ||||
maxsize = ui.configbytes(b'cmdserver', b'max-log-size') | ||||
|
r40859 | vfs = vfsmod.vfs(os.path.dirname(logpath)) | ||
|
r43346 | logger = loggingutil.filelogger( | ||
vfs, | ||||
os.path.basename(logpath), | ||||
tracked, | ||||
maxfiles=maxfiles, | ||||
maxsize=maxsize, | ||||
) | ||||
|
r40859 | |||
targetuis = {ui} | ||||
if repo: | ||||
targetuis.add(repo.baseui) | ||||
targetuis.add(repo.ui) | ||||
for u in targetuis: | ||||
u.setlogger(b'cmdserver', logger) | ||||
|
r43346 | |||
|
r22988 | class pipeservice(object): | ||
def __init__(self, ui, repo, opts): | ||||
|
r23323 | self.ui = ui | ||
self.repo = repo | ||||
|
r22988 | |||
def init(self): | ||||
pass | ||||
def run(self): | ||||
|
r23323 | ui = self.ui | ||
|
r23324 | # redirect stdio to null device so that broken extensions or in-process | ||
# hooks will never cause corruption of channel protocol. | ||||
|
r41320 | with ui.protectedfinout() as (fin, fout): | ||
|
r40624 | sv = server(ui, self.repo, fin, fout) | ||
|
r37142 | try: | ||
return sv.serve() | ||||
finally: | ||||
sv.cleanup() | ||||
|
r22989 | |||
|
r43346 | |||
|
r29586 | def _initworkerprocess(): | ||
|
r29609 | # use a different process group from the master process, in order to: | ||
# 1. make the current process group no longer "orphaned" (because the | ||||
# parent of this process is in a different process group while | ||||
# remains in a same session) | ||||
# according to POSIX 2.2.2.52, orphaned process group will ignore | ||||
# terminal-generated stop signals like SIGTSTP (Ctrl+Z), which will | ||||
# cause trouble for things like ncurses. | ||||
# 2. the client can use kill(-pgid, sig) to simulate terminal-generated | ||||
# SIGINT (Ctrl+C) and process-exit-generated SIGHUP. our child | ||||
# processes like ssh will be killed properly, without affecting | ||||
# unrelated processes. | ||||
|
r29585 | os.setpgid(0, 0) | ||
# change random state otherwise forked request handlers would have a | ||||
# same state inherited from parent. | ||||
random.seed() | ||||
|
r29542 | |||
|
r43346 | |||
|
r40911 | def _serverequest(ui, repo, conn, createcmdserver, prereposetups): | ||
|
r43906 | fin = conn.makefile('rb') | ||
fout = conn.makefile('wb') | ||||
|
r29585 | sv = None | ||
try: | ||||
|
r40911 | sv = createcmdserver(repo, conn, fin, fout, prereposetups) | ||
|
r22994 | try: | ||
|
r29585 | sv.serve() | ||
# handle exceptions that may be raised by command server. most of | ||||
# known exceptions are caught by dispatch. | ||||
except error.Abort as inst: | ||||
|
r43347 | ui.error(_(b'abort: %s\n') % inst) | ||
|
r29585 | except IOError as inst: | ||
if inst.errno != errno.EPIPE: | ||||
raise | ||||
except KeyboardInterrupt: | ||||
pass | ||||
|
r29513 | finally: | ||
|
r29585 | sv.cleanup() | ||
|
r43346 | except: # re-raises | ||
|
r29585 | # also write traceback to error channel. otherwise client cannot | ||
# see it because it is written to server's stderr by default. | ||||
if sv: | ||||
cerr = sv.cerr | ||||
else: | ||||
|
r43347 | cerr = channeledoutput(fout, b'e') | ||
|
r40397 | cerr.write(encoding.strtolocal(traceback.format_exc())) | ||
|
r29585 | raise | ||
finally: | ||||
fin.close() | ||||
try: | ||||
fout.close() # implicit flush() may cause another EPIPE | ||||
except IOError as inst: | ||||
if inst.errno != errno.EPIPE: | ||||
raise | ||||
|
r22994 | |||
|
r43346 | |||
|
r29544 | class unixservicehandler(object): | ||
"""Set of pluggable operations for unix-mode services | ||||
Almost all methods except for createcmdserver() are called in the main | ||||
process. You can't pass mutable resource back from createcmdserver(). | ||||
""" | ||||
pollinterval = None | ||||
def __init__(self, ui): | ||||
self.ui = ui | ||||
def bindsocket(self, sock, address): | ||||
util.bindunixsocket(sock, address) | ||||
|
r32232 | sock.listen(socket.SOMAXCONN) | ||
|
r43347 | self.ui.status(_(b'listening at %s\n') % address) | ||
|
r32233 | self.ui.flush() # avoid buffering of status message | ||
|
r29544 | |||
def unlinksocket(self, address): | ||||
os.unlink(address) | ||||
def shouldexit(self): | ||||
"""True if server should shut down; checked per pollinterval""" | ||||
return False | ||||
def newconnection(self): | ||||
"""Called when main process notices new connection""" | ||||
|
r40911 | def createcmdserver(self, repo, conn, fin, fout, prereposetups): | ||
|
r29544 | """Create new command server instance; called in the process that | ||
serves for the current connection""" | ||||
|
r40911 | return server(self.ui, repo, fin, fout, prereposetups) | ||
|
r29544 | |||
|
r43346 | |||
|
r29548 | class unixforkingservice(object): | ||
|
r22994 | """ | ||
Listens on unix domain socket and forks server per connection | ||||
""" | ||||
|
r29548 | |||
def __init__(self, ui, repo, opts, handler=None): | ||||
|
r22994 | self.ui = ui | ||
self.repo = repo | ||||
|
r43347 | self.address = opts[b'address'] | ||
if not util.safehasattr(socket, b'AF_UNIX'): | ||||
raise error.Abort(_(b'unsupported platform')) | ||||
|
r22994 | if not self.address: | ||
|
r43347 | raise error.Abort(_(b'no socket path specified with --address')) | ||
|
r29544 | self._servicehandler = handler or unixservicehandler(ui) | ||
self._sock = None | ||||
|
r41034 | self._mainipc = None | ||
self._workeripc = None | ||||
|
r29544 | self._oldsigchldhandler = None | ||
self._workerpids = set() # updated by signal handler; do not iterate | ||||
|
r30887 | self._socketunlinked = None | ||
|
r41035 | # experimental config: cmdserver.max-repo-cache | ||
maxlen = ui.configint(b'cmdserver', b'max-repo-cache') | ||||
if maxlen < 0: | ||||
|
r43347 | raise error.Abort(_(b'negative max-repo-cache size not allowed')) | ||
|
r41035 | self._repoloader = repocache.repoloader(ui, maxlen) | ||
|
r29544 | |||
def init(self): | ||||
self._sock = socket.socket(socket.AF_UNIX) | ||||
|
r41034 | # IPC channel from many workers to one main process; this is actually | ||
# a uni-directional pipe, but is backed by a DGRAM socket so each | ||||
# message can be easily separated. | ||||
o = socket.socketpair(socket.AF_UNIX, socket.SOCK_DGRAM) | ||||
self._mainipc, self._workeripc = o | ||||
|
r29544 | self._servicehandler.bindsocket(self._sock, self.address) | ||
|
r43347 | if util.safehasattr(procutil, b'unblocksignal'): | ||
|
r37138 | procutil.unblocksignal(signal.SIGCHLD) | ||
|
r29544 | o = signal.signal(signal.SIGCHLD, self._sigchldhandler) | ||
self._oldsigchldhandler = o | ||||
|
r30887 | self._socketunlinked = False | ||
|
r41035 | self._repoloader.start() | ||
|
r30887 | |||
def _unlinksocket(self): | ||||
if not self._socketunlinked: | ||||
self._servicehandler.unlinksocket(self.address) | ||||
self._socketunlinked = True | ||||
|
r29544 | |||
def _cleanup(self): | ||||
signal.signal(signal.SIGCHLD, self._oldsigchldhandler) | ||||
self._sock.close() | ||||
|
r41034 | self._mainipc.close() | ||
self._workeripc.close() | ||||
|
r30887 | self._unlinksocket() | ||
|
r41035 | self._repoloader.stop() | ||
|
r29544 | # don't kill child processes as they have active clients, just wait | ||
self._reapworkers(0) | ||||
def run(self): | ||||
try: | ||||
self._mainloop() | ||||
finally: | ||||
self._cleanup() | ||||
def _mainloop(self): | ||||
|
r30891 | exiting = False | ||
|
r29544 | h = self._servicehandler | ||
|
r36958 | selector = selectors.DefaultSelector() | ||
|
r43346 | selector.register( | ||
self._sock, selectors.EVENT_READ, self._acceptnewconnection | ||||
) | ||||
selector.register( | ||||
self._mainipc, selectors.EVENT_READ, self._handlemainipc | ||||
) | ||||
|
r30891 | while True: | ||
if not exiting and h.shouldexit(): | ||||
# clients can no longer connect() to the domain socket, so | ||||
# we stop queuing new requests. | ||||
# for requests that are queued (connect()-ed, but haven't been | ||||
# accept()-ed), handle them before exit. otherwise, clients | ||||
# waiting for recv() will receive ECONNRESET. | ||||
self._unlinksocket() | ||||
exiting = True | ||||
|
r40833 | try: | ||
|
r40914 | events = selector.select(timeout=h.pollinterval) | ||
|
r40833 | except OSError as inst: | ||
# selectors2 raises ETIMEDOUT if timeout exceeded while | ||||
# handling signal interrupt. That's probably wrong, but | ||||
# we can easily get around it. | ||||
if inst.errno != errno.ETIMEDOUT: | ||||
raise | ||||
|
r40914 | events = [] | ||
if not events: | ||||
|
r33543 | # only exit if we completed all queued requests | ||
if exiting: | ||||
break | ||||
continue | ||||
|
r40914 | for key, _mask in events: | ||
key.data(key.fileobj, selector) | ||||
|
r40912 | selector.close() | ||
def _acceptnewconnection(self, sock, selector): | ||||
h = self._servicehandler | ||||
|
r40913 | try: | ||
conn, _addr = sock.accept() | ||||
except socket.error as inst: | ||||
if inst.args[0] == errno.EINTR: | ||||
return | ||||
raise | ||||
|
r41035 | # Future improvement: On Python 3.7, maybe gc.freeze() can be used | ||
# to prevent COW memory from being touched by GC. | ||||
# https://instagram-engineering.com/ | ||||
# copy-on-write-friendly-python-garbage-collection-ad6ed5233ddf | ||||
|
r40913 | pid = os.fork() | ||
if pid: | ||||
|
r29544 | try: | ||
|
r43346 | self.ui.log( | ||
b'cmdserver', b'forked worker process (pid=%d)\n', pid | ||||
) | ||||
|
r40913 | self._workerpids.add(pid) | ||
h.newconnection() | ||||
finally: | ||||
conn.close() # release handle in parent process | ||||
else: | ||||
try: | ||||
selector.close() | ||||
sock.close() | ||||
|
r41034 | self._mainipc.close() | ||
|
r40913 | self._runworker(conn) | ||
conn.close() | ||||
|
r41034 | self._workeripc.close() | ||
|
r40913 | os._exit(0) | ||
except: # never return, hence no re-raises | ||||
|
r29544 | try: | ||
|
r40913 | self.ui.traceback(force=True) | ||
|
r29544 | finally: | ||
|
r40913 | os._exit(255) | ||
|
r29544 | |||
|
r41034 | def _handlemainipc(self, sock, selector): | ||
"""Process messages sent from a worker""" | ||||
try: | ||||
path = sock.recv(32768) # large enough to receive path | ||||
except socket.error as inst: | ||||
if inst.args[0] == errno.EINTR: | ||||
return | ||||
raise | ||||
|
r41035 | self._repoloader.load(path) | ||
|
r41034 | |||
|
r29544 | def _sigchldhandler(self, signal, frame): | ||
self._reapworkers(os.WNOHANG) | ||||
def _reapworkers(self, options): | ||||
while self._workerpids: | ||||
try: | ||||
pid, _status = os.waitpid(-1, options) | ||||
except OSError as inst: | ||||
if inst.errno == errno.EINTR: | ||||
continue | ||||
if inst.errno != errno.ECHILD: | ||||
raise | ||||
# no child processes at all (reaped by other waitpid()?) | ||||
self._workerpids.clear() | ||||
return | ||||
if pid == 0: | ||||
# no waitable child processes | ||||
return | ||||
|
r40863 | self.ui.log(b'cmdserver', b'worker process exited (pid=%d)\n', pid) | ||
|
r29544 | self._workerpids.discard(pid) | ||
|
r29587 | def _runworker(self, conn): | ||
|
r29544 | signal.signal(signal.SIGCHLD, self._oldsigchldhandler) | ||
|
r29586 | _initworkerprocess() | ||
|
r29544 | h = self._servicehandler | ||
|
r29586 | try: | ||
|
r43346 | _serverequest( | ||
self.ui, | ||||
self.repo, | ||||
conn, | ||||
h.createcmdserver, | ||||
prereposetups=[self._reposetup], | ||||
) | ||||
|
r29586 | finally: | ||
gc.collect() # trigger __del__ since worker process uses os._exit | ||||
|
r41034 | |||
def _reposetup(self, ui, repo): | ||||
if not repo.local(): | ||||
return | ||||
class unixcmdserverrepo(repo.__class__): | ||||
def close(self): | ||||
super(unixcmdserverrepo, self).close() | ||||
try: | ||||
self._cmdserveripc.send(self.root) | ||||
except socket.error: | ||||
|
r43346 | self.ui.log( | ||
b'cmdserver', b'failed to send repo root to master\n' | ||||
) | ||||
|
r41034 | |||
repo.__class__ = unixcmdserverrepo | ||||
repo._cmdserveripc = self._workeripc | ||||
|
r41035 | |||
cachedrepo = self._repoloader.get(repo.root) | ||||
if cachedrepo is None: | ||||
return | ||||
repo.ui.log(b'repocache', b'repo from cache: %s\n', repo.root) | ||||
repocache.copycache(cachedrepo, repo) | ||||