##// END OF EJS Templates
wireprotov2peer: stream decoded responses...
wireprotov2peer: stream decoded responses Previously, wire protocol version 2 would buffer all response data. Only once all data was received did we CBOR decode it and resolve the future associated with the command. This was obviously not desirable. In future commits that introduce large response payloads, this caused significant memory bloat and slowed down client operations due to waiting on the server. This commit refactors the response handling code so that response data can be streamed. Command response objects now contain a buffered CBOR decoder. As new data arrives, it is fed into the decoder. Decoded objects are made available to the generator as they are decoded. Because there is a separate thread processing incoming frames and feeding data into the response object, there is the potential for race conditions when mutating response objects. So a lock has been added to guard access to critical state variables. Because the generator emitting decoded objects needs to wait on those objects to become available, we've added an Event for the generator to wait on so it doesn't busy loop. This does mean there is the potential for deadlocks. And I'm pretty sure they can occur in some scenarios. We already have a handful of TODOs around this. But I've added some more. Fixing this will likely require moving the background thread receiving frames into clienthandler. We likely would have done this anyway when implementing the client bits for the SSH transport. Test output changes because the initial CBOR map holding the overall response state is now always handled internally by the response object. Differential Revision: https://phab.mercurial-scm.org/D4474

File last commit:

r28836:3f45488d default
r39597:d06834e0 default
Show More
hgclient.py
123 lines | 3.3 KiB | text/x-python | PythonLexer
Yuya Nishihara
test-commandserver: split helper functions to new hgclient module...
r22566 # A minimal client for Mercurial's command server
Pulkit Goyal
hgclient: use absolute_import and print_function
r28355 from __future__ import absolute_import, print_function
import os
import signal
import socket
import struct
import subprocess
import sys
import time
Yuya Nishihara
test-commandserver: split helper functions to new hgclient module...
r22566
timeless
test-commandserver: handle cStringIO.StringIO/io.StringIO divergence
r28836 try:
import cStringIO as io
stringio = io.StringIO
except ImportError:
import io
stringio = io.StringIO
Yuya Nishihara
test-commandserver: allow check() to make connection in different way...
r22992 def connectpipe(path=None):
Yuya Nishihara
test-commandserver: split helper functions to new hgclient module...
r22566 cmdline = ['hg', 'serve', '--cmdserver', 'pipe']
if path:
cmdline += ['-R', path]
server = subprocess.Popen(cmdline, stdin=subprocess.PIPE,
stdout=subprocess.PIPE)
return server
Yuya Nishihara
test-commandserver: add connector for unix domain socket server...
r22993 class unixconnection(object):
def __init__(self, sockpath):
self.sock = sock = socket.socket(socket.AF_UNIX)
sock.connect(sockpath)
self.stdin = sock.makefile('wb')
self.stdout = sock.makefile('rb')
def wait(self):
self.stdin.close()
self.stdout.close()
self.sock.close()
class unixserver(object):
def __init__(self, sockpath, logpath=None, repopath=None):
self.sockpath = sockpath
cmdline = ['hg', 'serve', '--cmdserver', 'unix', '-a', sockpath]
if repopath:
cmdline += ['-R', repopath]
if logpath:
stdout = open(logpath, 'a')
stderr = subprocess.STDOUT
else:
stdout = stderr = None
self.server = subprocess.Popen(cmdline, stdout=stdout, stderr=stderr)
# wait for listen()
while self.server.poll() is None:
if os.path.exists(sockpath):
break
time.sleep(0.1)
def connect(self):
return unixconnection(self.sockpath)
def shutdown(self):
os.kill(self.server.pid, signal.SIGTERM)
self.server.wait()
Yuya Nishihara
test-commandserver: split helper functions to new hgclient module...
r22566 def writeblock(server, data):
server.stdin.write(struct.pack('>I', len(data)))
server.stdin.write(data)
server.stdin.flush()
def readchannel(server):
data = server.stdout.read(5)
if not data:
raise EOFError
channel, length = struct.unpack('>cI', data)
if channel in 'IL':
return channel, length
else:
return channel, server.stdout.read(length)
def sep(text):
return text.replace('\\', '/')
def runcommand(server, args, output=sys.stdout, error=sys.stderr, input=None,
outfilter=lambda x: x):
Pulkit Goyal
hgclient: use absolute_import and print_function
r28355 print('*** runcommand', ' '.join(args))
Yuya Nishihara
test-commandserver: split helper functions to new hgclient module...
r22566 sys.stdout.flush()
server.stdin.write('runcommand\n')
writeblock(server, '\0'.join(args))
if not input:
timeless
test-commandserver: handle cStringIO.StringIO/io.StringIO divergence
r28836 input = stringio()
Yuya Nishihara
test-commandserver: split helper functions to new hgclient module...
r22566
while True:
ch, data = readchannel(server)
if ch == 'o':
output.write(outfilter(data))
output.flush()
elif ch == 'e':
error.write(data)
error.flush()
elif ch == 'I':
writeblock(server, input.read(data))
elif ch == 'L':
writeblock(server, input.readline(data))
elif ch == 'r':
ret, = struct.unpack('>i', data)
if ret != 0:
Pulkit Goyal
hgclient: use absolute_import and print_function
r28355 print(' [%d]' % ret)
Yuya Nishihara
test-commandserver: split helper functions to new hgclient module...
r22566 return ret
else:
Pulkit Goyal
hgclient: use absolute_import and print_function
r28355 print("unexpected channel %c: %r" % (ch, data))
Yuya Nishihara
test-commandserver: split helper functions to new hgclient module...
r22566 if ch.isupper():
return
Yuya Nishihara
test-commandserver: allow check() to make connection in different way...
r22992 def check(func, connect=connectpipe):
Yuya Nishihara
test-commandserver: split helper functions to new hgclient module...
r22566 sys.stdout.flush()
Yuya Nishihara
test-commandserver: remove unused repopath argument from check()...
r22991 server = connect()
Yuya Nishihara
test-commandserver: split helper functions to new hgclient module...
r22566 try:
return func(server)
finally:
server.stdin.close()
server.wait()