commandserver.py
398 lines
| 11.2 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 SocketServer | ||||
import errno | ||||
import os | ||||
|
r14647 | import struct | ||
|
r27351 | import sys | ||
import traceback | ||||
from .i18n import _ | ||||
from . import ( | ||||
encoding, | ||||
error, | ||||
util, | ||||
) | ||||
|
r14647 | |||
logfile = None | ||||
def log(*args): | ||||
if not logfile: | ||||
return | ||||
for a in args: | ||||
logfile.write(str(a)) | ||||
logfile.flush() | ||||
class channeledoutput(object): | ||||
""" | ||||
|
r22561 | Write data to out in the following format: | ||
|
r14647 | |||
data length (unsigned int), | ||||
data | ||||
""" | ||||
|
r22563 | def __init__(self, out, channel): | ||
|
r14647 | self.out = out | ||
self.channel = channel | ||||
|
r27415 | @property | ||
def name(self): | ||||
return '<%c-channel>' % self.channel | ||||
|
r14647 | def write(self, data): | ||
if not data: | ||||
return | ||||
self.out.write(struct.pack('>cI', self.channel, len(data))) | ||||
self.out.write(data) | ||||
self.out.flush() | ||||
def __getattr__(self, attr): | ||||
|
r27915 | if attr in ('isatty', 'fileno', 'tell', 'seek'): | ||
|
r18174 | raise AttributeError(attr) | ||
|
r22563 | return getattr(self.out, attr) | ||
|
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): | ||||
return '<%c-channel>' % self.channel | ||||
|
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: | ||||
return '' | ||||
assert size > 0 | ||||
# tell the client we need at most size bytes | ||||
self.out.write(struct.pack('>cI', channel, size)) | ||||
self.out.flush() | ||||
length = self.in_.read(4) | ||||
length = struct.unpack('>I', length)[0] | ||||
if not length: | ||||
return '' | ||||
else: | ||||
return self.in_.read(length) | ||||
def readline(self, size=-1): | ||||
if size < 0: | ||||
size = self.maxchunksize | ||||
s = self._read(size, 'L') | ||||
buf = s | ||||
# keep asking for more until there's either no more or | ||||
# we got a full line | ||||
while s and s[-1] != '\n': | ||||
|
r14728 | s = self._read(size, 'L') | ||
|
r14647 | buf += s | ||
return buf | ||||
else: | ||||
return self._read(size, 'L') | ||||
def __iter__(self): | ||||
return self | ||||
def next(self): | ||||
l = self.readline() | ||||
if not l: | ||||
raise StopIteration | ||||
return l | ||||
def __getattr__(self, attr): | ||||
|
r27915 | if attr in ('isatty', 'fileno', 'tell', 'seek'): | ||
|
r18174 | raise AttributeError(attr) | ||
|
r14647 | return getattr(self.in_, attr) | ||
class server(object): | ||||
""" | ||||
|
r22990 | Listens for commands on fin, runs them and writes the output on a channel | ||
based stream to fout. | ||||
|
r14647 | """ | ||
|
r22990 | def __init__(self, ui, repo, fin, fout): | ||
|
r14864 | self.cwd = os.getcwd() | ||
|
r14647 | |||
|
r25832 | # developer config: cmdserver.log | ||
|
r14647 | logpath = ui.config("cmdserver", "log", None) | ||
if logpath: | ||||
global logfile | ||||
if logpath == '-': | ||||
|
r17425 | # write log on a special 'd' (debug) channel | ||
|
r22990 | logfile = channeledoutput(fout, 'd') | ||
|
r14647 | else: | ||
logfile = open(logpath, 'a') | ||||
|
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 | ||||
|
r14647 | |||
|
r22990 | self.cerr = channeledoutput(fout, 'e') | ||
self.cout = channeledoutput(fout, 'o') | ||||
self.cin = channeledinput(fin, fout, 'I') | ||||
self.cresult = channeledoutput(fout, 'r') | ||||
|
r14647 | |||
|
r22990 | self.client = fin | ||
|
r14647 | |||
def _read(self, size): | ||||
|
r14706 | if not size: | ||
return '' | ||||
|
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 | ||||
""" | ||||
length = struct.unpack('>I', self._read(4))[0] | ||||
if not length: | ||||
return '' | ||||
return self._read(length) | ||||
def _readlist(self): | ||||
"""read a list of NULL separated strings from the channel""" | ||||
s = self._readstr() | ||||
if s: | ||||
return s.split('\0') | ||||
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) | ||||
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. | ||||
if not util.safehasattr(self.cin, 'fileno'): | ||||
ui.setconfig('ui', 'nontty', 'true', 'commandserver') | ||||
|
r21195 | |||
|
r14864 | req = dispatch.request(args[:], copiedui, self.repo, self.cin, | ||
|
r14647 | self.cout, self.cerr) | ||
|
r20631 | ret = (dispatch.dispatch(req) or 0) & 255 # might return None | ||
|
r14647 | |||
|
r14864 | # restore old cwd | ||
if '--cwd' in args: | ||||
os.chdir(self.cwd) | ||||
|
r14647 | self.cresult.write(struct.pack('>i', int(ret))) | ||
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 | ||||
|
r26587 | raise error.Abort(_('unknown command %s') % cmd) | ||
|
r14647 | |||
return cmd != '' | ||||
capabilities = {'runcommand' : runcommand, | ||||
'getencoding' : getencoding} | ||||
def serve(self): | ||||
|
r18359 | hellomsg = 'capabilities: ' + ' '.join(sorted(self.capabilities)) | ||
|
r14719 | hellomsg += '\n' | ||
hellomsg += 'encoding: ' + encoding.encoding | ||||
|
r23036 | hellomsg += '\n' | ||
|
r28027 | hellomsg += 'pid: %d' % util.getpid() | ||
|
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 | |||
|
r23324 | def _protectio(ui): | ||
""" duplicates streams and redirect original to null if ui uses stdio """ | ||||
ui.flush() | ||||
newfiles = [] | ||||
nullfd = os.open(os.devnull, os.O_RDWR) | ||||
for f, sysf, mode in [(ui.fin, sys.stdin, 'rb'), | ||||
(ui.fout, sys.stdout, 'wb')]: | ||||
if f is sysf: | ||||
newfd = os.dup(f.fileno()) | ||||
os.dup2(nullfd, f.fileno()) | ||||
f = os.fdopen(newfd, mode) | ||||
newfiles.append(f) | ||||
os.close(nullfd) | ||||
return tuple(newfiles) | ||||
def _restoreio(ui, fin, fout): | ||||
""" restores streams from duplicated ones """ | ||||
ui.flush() | ||||
for f, uif in [(fin, ui.fin), (fout, ui.fout)]: | ||||
if f is not uif: | ||||
os.dup2(f.fileno(), uif.fileno()) | ||||
f.close() | ||||
|
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. | ||||
fin, fout = _protectio(ui) | ||||
try: | ||||
sv = server(ui, self.repo, fin, fout) | ||||
return sv.serve() | ||||
finally: | ||||
_restoreio(ui, fin, fout) | ||||
|
r22989 | |||
|
r22994 | class _requesthandler(SocketServer.StreamRequestHandler): | ||
def handle(self): | ||||
ui = self.server.ui | ||||
repo = self.server.repo | ||||
|
r28511 | sv = None | ||
|
r22994 | try: | ||
|
r28511 | sv = server(ui, repo, self.rfile, self.wfile) | ||
|
r22994 | try: | ||
sv.serve() | ||||
# handle exceptions that may be raised by command server. most of | ||||
# known exceptions are caught by dispatch. | ||||
|
r26587 | except error.Abort as inst: | ||
|
r22994 | ui.warn(_('abort: %s\n') % inst) | ||
|
r25660 | except IOError as inst: | ||
|
r22994 | if inst.errno != errno.EPIPE: | ||
raise | ||||
except KeyboardInterrupt: | ||||
pass | ||||
except: # re-raises | ||||
# also write traceback to error channel. otherwise client cannot | ||||
# see it because it is written to server's stderr by default. | ||||
|
r28511 | if sv: | ||
cerr = sv.cerr | ||||
else: | ||||
cerr = channeledoutput(self.wfile, 'e') | ||||
traceback.print_exc(file=cerr) | ||||
|
r22994 | raise | ||
class unixservice(object): | ||||
""" | ||||
Listens on unix domain socket and forks server per connection | ||||
""" | ||||
def __init__(self, ui, repo, opts): | ||||
self.ui = ui | ||||
self.repo = repo | ||||
self.address = opts['address'] | ||||
if not util.safehasattr(SocketServer, 'UnixStreamServer'): | ||||
|
r26587 | raise error.Abort(_('unsupported platform')) | ||
|
r22994 | if not self.address: | ||
|
r26587 | raise error.Abort(_('no socket path specified with --address')) | ||
|
r22994 | |||
def init(self): | ||||
class cls(SocketServer.ForkingMixIn, SocketServer.UnixStreamServer): | ||||
ui = self.ui | ||||
repo = self.repo | ||||
self.server = cls(self.address, _requesthandler) | ||||
self.ui.status(_('listening at %s\n') % self.address) | ||||
self.ui.flush() # avoid buffering of status message | ||||
def run(self): | ||||
try: | ||||
self.server.serve_forever() | ||||
finally: | ||||
os.unlink(self.address) | ||||
|
r22989 | _servicemap = { | ||
'pipe': pipeservice, | ||||
|
r22994 | 'unix': unixservice, | ||
|
r22989 | } | ||
def createservice(ui, repo, opts): | ||||
mode = opts['cmdserver'] | ||||
try: | ||||
return _servicemap[mode](ui, repo, opts) | ||||
except KeyError: | ||||
|
r26587 | raise error.Abort(_('unknown mode %s') % mode) | ||