This is an archive of the discontinued Mercurial Phabricator instance.

Differential D2404

util: enable observing of util.bufferedinputpipe
ClosedPublic

Authored by indygreg on Feb 23 2018, 2:01 PM.

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHG3158052720ae: util: enable observing of util.bufferedinputpipe
rHGcafa60632281: util: enable observing of util.bufferedinputpipe
Summary

Our file object proxy is useful. But it doesn't capture all I/O.
The "os" module offers low-level interfaces to various system calls.
For example, os.read() exposes read(2) to read from a file
descriptor.

bufferedinputpipe is special in a few ways. First, it acts as a
proxy of sorts around our [potentially proxied] file object. In
addition, it uses os.read() to satisfy all I/O. This means that
our observer doesn't see notifications for reads on this type.
This is preventing us from properly instrumenting reads on ssh
peers.

This commit teaches bufferedinputpipe to be aware of our
observed file objects. We do this by introducing a class variation
that notifies our observer of os.read() events. Since read()
and readline() bypass os.read(), we also teach this instance
to notify the observer for buffered variations of these reads as
well. We don't report them as actual read() and readline() calls
because these methods are never called on the actual file object
but rather a buffered version of it.

We introduce bufferedinputpipe.new to swap in the new class
if the passed file object is a fileobjectproxy. This makes hooking
up the observer automatic. And it is a zero cost abstraction for
I/O operations on non-proxied file objects.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

indygreg created this revision.Feb 23 2018, 2:01 PM
indygreg planned changes to this revision.Feb 23 2018, 3:12 PM

I have some revisions to this...

indygreg edited the summary of this revision. (Show Details)Feb 23 2018, 3:51 PM
indygreg retitled this revision from util: enable observing of os.read() from bufferedinputpipe to util: enable observing of util.bufferedinputpipe.
indygreg updated this revision to Diff 6025.
indygreg updated this revision to Diff 6247.Mar 1 2018, 11:30 AM
durin42 accepted this revision.Mar 1 2018, 4:13 PM
This revision is now accepted and ready to land.Mar 1 2018, 4:13 PM
Closed by commit rHGcafa60632281: util: enable observing of util.bufferedinputpipe (authored by indygreg). · Explain WhyMar 1 2018, 4:30 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/util.py (61 lines)

Diff 6025

mercurial/util.py

This class let us work around that by implementing its own buffering This class let us work around that by implementing its own buffering
(allowing efficient readline) while offering a way to know if the buffer is (allowing efficient readline) while offering a way to know if the buffer is
empty from the output (allowing collaboration of the buffer with polling). empty from the output (allowing collaboration of the buffer with polling).
This class lives in the 'util' module because it makes use of the 'os' This class lives in the 'util' module because it makes use of the 'os'
module from the python stdlib. module from the python stdlib.
""" """
def __new__(cls, fh):
# If we receive a fileobjectproxy, we need to use a variation of this
# class that notifies observers about activity.
if isinstance(fh, fileobjectproxy):
cls = observedbufferedinputpipe
return super(bufferedinputpipe, cls).__new__(cls)
def __init__(self, input): def __init__(self, input):
self._input = input self._input = input
self._buffer = [] self._buffer = []
self._eof = False self._eof = False
self._lenbuf = 0 self._lenbuf = 0
@property @property
"""read data to the buffer""" """read data to the buffer"""
data = os.read(self._input.fileno(), _chunksize) data = os.read(self._input.fileno(), _chunksize)
if not data: if not data:
self._eof = True self._eof = True
else: else:
self._lenbuf += len(data) self._lenbuf += len(data)
self._buffer.append(data) self._buffer.append(data)
return data
def mmapread(fp): def mmapread(fp):
try: try:
fd = getattr(fp, 'fileno', lambda: fp)() fd = getattr(fp, 'fileno', lambda: fp)()
return mmap.mmap(fd, 0, access=mmap.ACCESS_READ) return mmap.mmap(fd, 0, access=mmap.ACCESS_READ)
except ValueError: except ValueError:
# Empty files cannot be mmapped, but mmapread should still work. Check # Empty files cannot be mmapped, but mmapread should still work. Check
# if the file is empty, and if so, return an empty buffer. # if the file is empty, and if so, return an empty buffer.
if os.fstat(fd).st_size == 0: if os.fstat(fd).st_size == 0:
) )
def __init__(self, fh, observer): def __init__(self, fh, observer):
object.__setattr__(self, '_orig', fh) object.__setattr__(self, '_orig', fh)
object.__setattr__(self, '_observer', observer) object.__setattr__(self, '_observer', observer)
def __getattribute__(self, name): def __getattribute__(self, name):
ours = { ours = {
r'_observer',
# IOBase # IOBase
r'close', r'close',
# closed if a property # closed if a property
r'fileno', r'fileno',
r'flush', r'flush',
r'isatty', r'isatty',
r'readable', r'readable',
r'readline', r'readline',
def detach(self, *args, **kwargs): def detach(self, *args, **kwargs):
return object.__getattribute__(self, r'_observedcall')( return object.__getattribute__(self, r'_observedcall')(
r'detach', *args, **kwargs) r'detach', *args, **kwargs)
def read1(self, *args, **kwargs): def read1(self, *args, **kwargs):
return object.__getattribute__(self, r'_observedcall')( return object.__getattribute__(self, r'_observedcall')(
r'read1', *args, **kwargs) r'read1', *args, **kwargs)
class observedbufferedinputpipe(bufferedinputpipe):
"""A variation of bufferedinputpipe that is aware of fileobjectproxy.
``bufferedinputpipe`` makes low-level calls to ``os.read()`` that
bypass ``fileobjectproxy``. Because of this, we need to make
``bufferedinputpipe`` aware of these operations.
This variation of ``bufferedinputpipe`` can notify observers about
``os.read()`` events. It also re-publishes other events, such as
``read()`` and ``readline()``.
"""
def _fillbuffer(self):
res = super(observedbufferedinputpipe, self)._fillbuffer()
fn = getattr(self._input._observer, r'osread', None)
if fn:
fn(res, _chunksize)
return res
# We use different observer methods because the operation isn't
# performed on the actual file object but on us.
def read(self, size):
res = super(observedbufferedinputpipe, self).read(size)
fn = getattr(self._input._observer, r'bufferedread', None)
if fn:
fn(res, size)
return res
def readline(self, *args, **kwargs):
res = super(observedbufferedinputpipe, self).readline(*args, **kwargs)
fn = getattr(self._input._observer, r'bufferedreadline', None)
if fn:
fn(res)
return res
DATA_ESCAPE_MAP = {pycompat.bytechr(i): br'\x%02x' % i for i in range(256)} DATA_ESCAPE_MAP = {pycompat.bytechr(i): br'\x%02x' % i for i in range(256)}
DATA_ESCAPE_MAP.update({ DATA_ESCAPE_MAP.update({
b'\\': b'\\\\', b'\\': b'\\\\',
b'\r': br'\r', b'\r': br'\r',
b'\n': br'\n', b'\n': br'\n',
}) })
DATA_ESCAPE_RE = remod.compile(br'[\x00-\x08\x0a-\x1f\\\x7f-\xff]') DATA_ESCAPE_RE = remod.compile(br'[\x00-\x08\x0a-\x1f\\\x7f-\xff]')
def write(self, res, data): def write(self, res, data):
self.fh.write('%s> write(%d) -> %r' % (self.name, len(data), res)) self.fh.write('%s> write(%d) -> %r' % (self.name, len(data), res))
self._writedata(data) self._writedata(data)
def flush(self, res): def flush(self, res):
self.fh.write('%s> flush() -> %r\n' % (self.name, res)) self.fh.write('%s> flush() -> %r\n' % (self.name, res))
# For observedbufferedinputpipe.
def bufferedread(self, res, size):
self.fh.write('%s> bufferedread(%d) -> %d' % (
self.name, size, len(res)))
self._writedata(res)
def bufferedreadline(self, res):
self.fh.write('%s> bufferedreadline() -> %d' % (self.name, len(res)))
self._writedata(res)
def makeloggingfileobject(logh, fh, name, logdata=False): def makeloggingfileobject(logh, fh, name, logdata=False):
"""Turn a file object into a logging file object.""" """Turn a file object into a logging file object."""
observer = fileobjectobserver(logh, name, logdata=logdata) observer = fileobjectobserver(logh, name, logdata=logdata)
return fileobjectproxy(fh, observer) return fileobjectproxy(fh, observer)
def version(): def version():
"""Return version information if available.""" """Return version information if available."""