__init__.py
1195 lines
| 37.4 KiB
| text/x-python
|
PythonLexer
|
r28432 | # Copyright 2014-present Facebook, Inc. | ||
# All rights reserved. | ||||
# | ||||
# Redistribution and use in source and binary forms, with or without | ||||
# modification, are permitted provided that the following conditions are met: | ||||
# | ||||
# * Redistributions of source code must retain the above copyright notice, | ||||
# this list of conditions and the following disclaimer. | ||||
# | ||||
# * Redistributions in binary form must reproduce the above copyright notice, | ||||
# this list of conditions and the following disclaimer in the documentation | ||||
# and/or other materials provided with the distribution. | ||||
# | ||||
# * Neither the name Facebook nor the names of its contributors may be used to | ||||
# endorse or promote products derived from this software without specific | ||||
# prior written permission. | ||||
# | ||||
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" | ||||
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE | ||||
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE | ||||
# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE | ||||
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL | ||||
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR | ||||
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER | ||||
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, | ||||
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE | ||||
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. | ||||
|
r30656 | # no unicode literals | ||
|
r43703 | from __future__ import absolute_import, division, print_function | ||
|
r30656 | |||
import inspect | ||||
import math | ||||
|
r28432 | import os | ||
import socket | ||||
import subprocess | ||||
import time | ||||
|
r43703 | from . import capabilities, compat, encoding, load | ||
|
r28432 | # Sometimes it's really hard to get Python extensions to compile, | ||
# so fall back to a pure Python implementation. | ||||
try: | ||||
|
r30656 | from . import bser | ||
|
r43703 | |||
|
r30656 | # Demandimport causes modules to be loaded lazily. Force the load now | ||
# so that we can fall back on pybser if bser doesn't exist | ||||
bser.pdu_info | ||||
|
r28432 | except ImportError: | ||
|
r30656 | from . import pybser as bser | ||
|
r28432 | |||
|
r39851 | |||
|
r43703 | if os.name == "nt": | ||
|
r28432 | import ctypes | ||
import ctypes.wintypes | ||||
wintypes = ctypes.wintypes | ||||
GENERIC_READ = 0x80000000 | ||||
GENERIC_WRITE = 0x40000000 | ||||
FILE_FLAG_OVERLAPPED = 0x40000000 | ||||
OPEN_EXISTING = 3 | ||||
|
r43703 | INVALID_HANDLE_VALUE = ctypes.c_void_p(-1).value | ||
|
r28432 | FORMAT_MESSAGE_FROM_SYSTEM = 0x00001000 | ||
FORMAT_MESSAGE_ALLOCATE_BUFFER = 0x00000100 | ||||
FORMAT_MESSAGE_IGNORE_INSERTS = 0x00000200 | ||||
|
r30656 | WAIT_FAILED = 0xFFFFFFFF | ||
|
r28432 | WAIT_TIMEOUT = 0x00000102 | ||
WAIT_OBJECT_0 = 0x00000000 | ||||
|
r30656 | WAIT_IO_COMPLETION = 0x000000C0 | ||
INFINITE = 0xFFFFFFFF | ||||
# Overlapped I/O operation is in progress. (997) | ||||
ERROR_IO_PENDING = 0x000003E5 | ||||
# The pointer size follows the architecture | ||||
# We use WPARAM since this type is already conditionally defined | ||||
ULONG_PTR = ctypes.wintypes.WPARAM | ||||
|
r28432 | |||
class OVERLAPPED(ctypes.Structure): | ||||
_fields_ = [ | ||||
|
r43703 | ("Internal", ULONG_PTR), | ||
("InternalHigh", ULONG_PTR), | ||||
("Offset", wintypes.DWORD), | ||||
("OffsetHigh", wintypes.DWORD), | ||||
("hEvent", wintypes.HANDLE), | ||||
|
r28432 | ] | ||
def __init__(self): | ||||
|
r30656 | self.Internal = 0 | ||
self.InternalHigh = 0 | ||||
|
r28432 | self.Offset = 0 | ||
self.OffsetHigh = 0 | ||||
self.hEvent = 0 | ||||
LPDWORD = ctypes.POINTER(wintypes.DWORD) | ||||
CreateFile = ctypes.windll.kernel32.CreateFileA | ||||
|
r43703 | CreateFile.argtypes = [ | ||
wintypes.LPSTR, | ||||
wintypes.DWORD, | ||||
wintypes.DWORD, | ||||
wintypes.LPVOID, | ||||
wintypes.DWORD, | ||||
wintypes.DWORD, | ||||
wintypes.HANDLE, | ||||
] | ||||
|
r28432 | CreateFile.restype = wintypes.HANDLE | ||
CloseHandle = ctypes.windll.kernel32.CloseHandle | ||||
CloseHandle.argtypes = [wintypes.HANDLE] | ||||
CloseHandle.restype = wintypes.BOOL | ||||
ReadFile = ctypes.windll.kernel32.ReadFile | ||||
|
r43703 | ReadFile.argtypes = [ | ||
wintypes.HANDLE, | ||||
wintypes.LPVOID, | ||||
wintypes.DWORD, | ||||
LPDWORD, | ||||
ctypes.POINTER(OVERLAPPED), | ||||
] | ||||
|
r28432 | ReadFile.restype = wintypes.BOOL | ||
WriteFile = ctypes.windll.kernel32.WriteFile | ||||
|
r43703 | WriteFile.argtypes = [ | ||
wintypes.HANDLE, | ||||
wintypes.LPVOID, | ||||
wintypes.DWORD, | ||||
LPDWORD, | ||||
ctypes.POINTER(OVERLAPPED), | ||||
] | ||||
|
r28432 | WriteFile.restype = wintypes.BOOL | ||
GetLastError = ctypes.windll.kernel32.GetLastError | ||||
GetLastError.argtypes = [] | ||||
GetLastError.restype = wintypes.DWORD | ||||
|
r30656 | SetLastError = ctypes.windll.kernel32.SetLastError | ||
SetLastError.argtypes = [wintypes.DWORD] | ||||
SetLastError.restype = None | ||||
|
r28432 | FormatMessage = ctypes.windll.kernel32.FormatMessageA | ||
|
r43703 | FormatMessage.argtypes = [ | ||
wintypes.DWORD, | ||||
wintypes.LPVOID, | ||||
wintypes.DWORD, | ||||
wintypes.DWORD, | ||||
ctypes.POINTER(wintypes.LPSTR), | ||||
wintypes.DWORD, | ||||
wintypes.LPVOID, | ||||
] | ||||
|
r28432 | FormatMessage.restype = wintypes.DWORD | ||
LocalFree = ctypes.windll.kernel32.LocalFree | ||||
|
r30656 | GetOverlappedResult = ctypes.windll.kernel32.GetOverlappedResult | ||
|
r43703 | GetOverlappedResult.argtypes = [ | ||
wintypes.HANDLE, | ||||
ctypes.POINTER(OVERLAPPED), | ||||
LPDWORD, | ||||
wintypes.BOOL, | ||||
] | ||||
|
r30656 | GetOverlappedResult.restype = wintypes.BOOL | ||
|
r28432 | |||
|
r43703 | GetOverlappedResultEx = getattr( | ||
ctypes.windll.kernel32, "GetOverlappedResultEx", None | ||||
) | ||||
|
r30656 | if GetOverlappedResultEx is not None: | ||
|
r43703 | GetOverlappedResultEx.argtypes = [ | ||
wintypes.HANDLE, | ||||
ctypes.POINTER(OVERLAPPED), | ||||
LPDWORD, | ||||
wintypes.DWORD, | ||||
wintypes.BOOL, | ||||
] | ||||
|
r30656 | GetOverlappedResultEx.restype = wintypes.BOOL | ||
WaitForSingleObjectEx = ctypes.windll.kernel32.WaitForSingleObjectEx | ||||
|
r43703 | WaitForSingleObjectEx.argtypes = [ | ||
wintypes.HANDLE, | ||||
wintypes.DWORD, | ||||
wintypes.BOOL, | ||||
] | ||||
|
r30656 | WaitForSingleObjectEx.restype = wintypes.DWORD | ||
CreateEvent = ctypes.windll.kernel32.CreateEventA | ||||
|
r43703 | CreateEvent.argtypes = [ | ||
LPDWORD, | ||||
wintypes.BOOL, | ||||
wintypes.BOOL, | ||||
wintypes.LPSTR, | ||||
] | ||||
|
r30656 | CreateEvent.restype = wintypes.HANDLE | ||
# Windows Vista is the minimum supported client for CancelIoEx. | ||||
|
r28432 | CancelIoEx = ctypes.windll.kernel32.CancelIoEx | ||
CancelIoEx.argtypes = [wintypes.HANDLE, ctypes.POINTER(OVERLAPPED)] | ||||
CancelIoEx.restype = wintypes.BOOL | ||||
# 2 bytes marker, 1 byte int size, 8 bytes int64 value | ||||
sniff_len = 13 | ||||
# This is a helper for debugging the client. | ||||
_debugging = False | ||||
if _debugging: | ||||
def log(fmt, *args): | ||||
|
r43703 | print( | ||
"[%s] %s" | ||||
% ( | ||||
time.strftime("%a, %d %b %Y %H:%M:%S", time.gmtime()), | ||||
fmt % args[:], | ||||
) | ||||
) | ||||
|
r28432 | else: | ||
def log(fmt, *args): | ||||
pass | ||||
|
r30656 | def _win32_strerror(err): | ||
""" expand a win32 error code into a human readable message """ | ||||
# FormatMessage will allocate memory and assign it here | ||||
buf = ctypes.c_char_p() | ||||
FormatMessage( | ||||
|
r43703 | FORMAT_MESSAGE_FROM_SYSTEM | ||
| FORMAT_MESSAGE_ALLOCATE_BUFFER | ||||
| FORMAT_MESSAGE_IGNORE_INSERTS, | ||||
None, | ||||
err, | ||||
0, | ||||
buf, | ||||
0, | ||||
None, | ||||
) | ||||
|
r30656 | try: | ||
return buf.value | ||||
finally: | ||||
LocalFree(buf) | ||||
|
r28432 | class WatchmanError(Exception): | ||
|
r30656 | def __init__(self, msg=None, cmd=None): | ||
self.msg = msg | ||||
self.cmd = cmd | ||||
def setCommand(self, cmd): | ||||
self.cmd = cmd | ||||
def __str__(self): | ||||
if self.cmd: | ||||
|
r43703 | return "%s, while executing %s" % (self.msg, self.cmd) | ||
|
r30656 | return self.msg | ||
|
r43703 | class BSERv1Unsupported(WatchmanError): | ||
pass | ||||
class UseAfterFork(WatchmanError): | ||||
pass | ||||
|
r30656 | class WatchmanEnvironmentError(WatchmanError): | ||
def __init__(self, msg, errno, errmsg, cmd=None): | ||||
super(WatchmanEnvironmentError, self).__init__( | ||||
|
r43703 | "{0}: errno={1} errmsg={2}".format(msg, errno, errmsg), cmd | ||
) | ||||
|
r30656 | |||
class SocketConnectError(WatchmanError): | ||||
def __init__(self, sockpath, exc): | ||||
super(SocketConnectError, self).__init__( | ||||
|
r43703 | "unable to connect to %s: %s" % (sockpath, exc) | ||
) | ||||
|
r30656 | self.sockpath = sockpath | ||
self.exc = exc | ||||
|
r28432 | |||
class SocketTimeout(WatchmanError): | ||||
"""A specialized exception raised for socket timeouts during communication to/from watchman. | ||||
This makes it easier to implement non-blocking loops as callers can easily distinguish | ||||
between a routine timeout and an actual error condition. | ||||
Note that catching WatchmanError will also catch this as it is a super-class, so backwards | ||||
compatibility in exception handling is preserved. | ||||
""" | ||||
class CommandError(WatchmanError): | ||||
"""error returned by watchman | ||||
self.msg is the message returned by watchman. | ||||
""" | ||||
|
r43703 | |||
|
r28432 | def __init__(self, msg, cmd=None): | ||
|
r30656 | super(CommandError, self).__init__( | ||
|
r43703 | "watchman command error: %s" % (msg,), cmd | ||
|
r30656 | ) | ||
|
r28432 | |||
class Transport(object): | ||||
""" communication transport to the watchman server """ | ||||
|
r43703 | |||
|
r28432 | buf = None | ||
def close(self): | ||||
""" tear it down """ | ||||
raise NotImplementedError() | ||||
def readBytes(self, size): | ||||
""" read size bytes """ | ||||
raise NotImplementedError() | ||||
def write(self, buf): | ||||
""" write some data """ | ||||
raise NotImplementedError() | ||||
def setTimeout(self, value): | ||||
pass | ||||
def readLine(self): | ||||
""" read a line | ||||
Maintains its own buffer, callers of the transport should not mix | ||||
calls to readBytes and readLine. | ||||
""" | ||||
if self.buf is None: | ||||
self.buf = [] | ||||
# Buffer may already have a line if we've received unilateral | ||||
# response(s) from the server | ||||
|
r30656 | if len(self.buf) == 1 and b"\n" in self.buf[0]: | ||
(line, b) = self.buf[0].split(b"\n", 1) | ||||
|
r28432 | self.buf = [b] | ||
return line | ||||
while True: | ||||
b = self.readBytes(4096) | ||||
|
r30656 | if b"\n" in b: | ||
|
r43703 | result = b"".join(self.buf) | ||
|
r30656 | (line, b) = b.split(b"\n", 1) | ||
|
r28432 | self.buf = [b] | ||
return result + line | ||||
self.buf.append(b) | ||||
class Codec(object): | ||||
""" communication encoding for the watchman server """ | ||||
|
r43703 | |||
|
r28432 | transport = None | ||
def __init__(self, transport): | ||||
self.transport = transport | ||||
def receive(self): | ||||
raise NotImplementedError() | ||||
def send(self, *args): | ||||
raise NotImplementedError() | ||||
def setTimeout(self, value): | ||||
self.transport.setTimeout(value) | ||||
class UnixSocketTransport(Transport): | ||||
""" local unix domain socket transport """ | ||||
|
r43703 | |||
|
r28432 | sock = None | ||
|
r43703 | def __init__(self, sockpath, timeout): | ||
|
r28432 | self.sockpath = sockpath | ||
self.timeout = timeout | ||||
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) | ||||
try: | ||||
sock.settimeout(self.timeout) | ||||
sock.connect(self.sockpath) | ||||
self.sock = sock | ||||
except socket.error as e: | ||||
|
r30656 | sock.close() | ||
raise SocketConnectError(self.sockpath, e) | ||||
|
r28432 | |||
def close(self): | ||||
|
r43703 | if self.sock: | ||
self.sock.close() | ||||
self.sock = None | ||||
|
r28432 | |||
def setTimeout(self, value): | ||||
self.timeout = value | ||||
self.sock.settimeout(self.timeout) | ||||
def readBytes(self, size): | ||||
try: | ||||
buf = [self.sock.recv(size)] | ||||
if not buf[0]: | ||||
|
r43703 | raise WatchmanError("empty watchman response") | ||
|
r28432 | return buf[0] | ||
except socket.timeout: | ||||
|
r43703 | raise SocketTimeout("timed out waiting for response") | ||
|
r28432 | |||
def write(self, data): | ||||
try: | ||||
self.sock.sendall(data) | ||||
except socket.timeout: | ||||
|
r43703 | raise SocketTimeout("timed out sending query command") | ||
|
r28432 | |||
|
r30656 | def _get_overlapped_result_ex_impl(pipe, olap, nbytes, millis, alertable): | ||
""" Windows 7 and earlier does not support GetOverlappedResultEx. The | ||||
alternative is to use GetOverlappedResult and wait for read or write | ||||
operation to complete. This is done be using CreateEvent and | ||||
WaitForSingleObjectEx. CreateEvent, WaitForSingleObjectEx | ||||
and GetOverlappedResult are all part of Windows API since WindowsXP. | ||||
This is the exact same implementation that can be found in the watchman | ||||
source code (see get_overlapped_result_ex_impl in stream_win.c). This | ||||
way, maintenance should be simplified. | ||||
""" | ||||
|
r43703 | log("Preparing to wait for maximum %dms", millis) | ||
|
r30656 | if millis != 0: | ||
waitReturnCode = WaitForSingleObjectEx(olap.hEvent, millis, alertable) | ||||
if waitReturnCode == WAIT_OBJECT_0: | ||||
# Event is signaled, overlapped IO operation result should be available. | ||||
pass | ||||
elif waitReturnCode == WAIT_IO_COMPLETION: | ||||
# WaitForSingleObjectEx returnes because the system added an I/O completion | ||||
# routine or an asynchronous procedure call (APC) to the thread queue. | ||||
SetLastError(WAIT_IO_COMPLETION) | ||||
pass | ||||
elif waitReturnCode == WAIT_TIMEOUT: | ||||
# We reached the maximum allowed wait time, the IO operation failed | ||||
# to complete in timely fashion. | ||||
SetLastError(WAIT_TIMEOUT) | ||||
return False | ||||
elif waitReturnCode == WAIT_FAILED: | ||||
# something went wrong calling WaitForSingleObjectEx | ||||
err = GetLastError() | ||||
|
r43703 | log("WaitForSingleObjectEx failed: %s", _win32_strerror(err)) | ||
|
r30656 | return False | ||
else: | ||||
# unexpected situation deserving investigation. | ||||
err = GetLastError() | ||||
|
r43703 | log("Unexpected error: %s", _win32_strerror(err)) | ||
|
r30656 | return False | ||
return GetOverlappedResult(pipe, olap, nbytes, False) | ||||
|
r28432 | class WindowsNamedPipeTransport(Transport): | ||
""" connect to a named pipe """ | ||||
|
r43703 | def __init__(self, sockpath, timeout): | ||
|
r28432 | self.sockpath = sockpath | ||
self.timeout = int(math.ceil(timeout * 1000)) | ||||
self._iobuf = None | ||||
|
r43703 | if compat.PYTHON3: | ||
sockpath = os.fsencode(sockpath) | ||||
self.pipe = CreateFile( | ||||
sockpath, | ||||
GENERIC_READ | GENERIC_WRITE, | ||||
0, | ||||
None, | ||||
OPEN_EXISTING, | ||||
FILE_FLAG_OVERLAPPED, | ||||
None, | ||||
) | ||||
|
r28432 | |||
|
r43703 | err = GetLastError() | ||
if self.pipe == INVALID_HANDLE_VALUE or self.pipe == 0: | ||||
|
r28432 | self.pipe = None | ||
|
r43703 | raise SocketConnectError(self.sockpath, self._make_win_err("", err)) | ||
|
r28432 | |||
|
r30656 | # event for the overlapped I/O operations | ||
self._waitable = CreateEvent(None, True, False, None) | ||||
|
r43703 | err = GetLastError() | ||
|
r30656 | if self._waitable is None: | ||
|
r43703 | self._raise_win_err("CreateEvent failed", err) | ||
|
r28432 | |||
|
r30656 | self._get_overlapped_result_ex = GetOverlappedResultEx | ||
|
r43703 | if ( | ||
os.getenv("WATCHMAN_WIN7_COMPAT") == "1" | ||||
or self._get_overlapped_result_ex is None | ||||
): | ||||
|
r30656 | self._get_overlapped_result_ex = _get_overlapped_result_ex_impl | ||
|
r28432 | |||
def _raise_win_err(self, msg, err): | ||||
|
r43703 | raise self._make_win_err(msg, err) | ||
def _make_win_err(self, msg, err): | ||||
return IOError( | ||||
"%s win32 error code: %d %s" % (msg, err, _win32_strerror(err)) | ||||
) | ||||
|
r28432 | |||
def close(self): | ||||
if self.pipe: | ||||
|
r43703 | log("Closing pipe") | ||
|
r28432 | CloseHandle(self.pipe) | ||
self.pipe = None | ||||
|
r30656 | if self._waitable is not None: | ||
# We release the handle for the event | ||||
CloseHandle(self._waitable) | ||||
self._waitable = None | ||||
def setTimeout(self, value): | ||||
# convert to milliseconds | ||||
self.timeout = int(value * 1000) | ||||
|
r28432 | def readBytes(self, size): | ||
""" A read can block for an unbounded amount of time, even if the | ||||
kernel reports that the pipe handle is signalled, so we need to | ||||
always perform our reads asynchronously | ||||
""" | ||||
# try to satisfy the read from any buffered data | ||||
if self._iobuf: | ||||
if size >= len(self._iobuf): | ||||
res = self._iobuf | ||||
self.buf = None | ||||
return res | ||||
res = self._iobuf[:size] | ||||
self._iobuf = self._iobuf[size:] | ||||
return res | ||||
# We need to initiate a read | ||||
buf = ctypes.create_string_buffer(size) | ||||
olap = OVERLAPPED() | ||||
|
r30656 | olap.hEvent = self._waitable | ||
|
r28432 | |||
|
r43703 | log("made read buff of size %d", size) | ||
|
r28432 | |||
# ReadFile docs warn against sending in the nread parameter for async | ||||
# operations, so we always collect it via GetOverlappedResultEx | ||||
immediate = ReadFile(self.pipe, buf, size, None, olap) | ||||
if not immediate: | ||||
err = GetLastError() | ||||
if err != ERROR_IO_PENDING: | ||||
|
r43703 | self._raise_win_err("failed to read %d bytes" % size, err) | ||
|
r28432 | |||
nread = wintypes.DWORD() | ||||
|
r43703 | if not self._get_overlapped_result_ex( | ||
self.pipe, olap, nread, 0 if immediate else self.timeout, True | ||||
): | ||||
|
r28432 | err = GetLastError() | ||
CancelIoEx(self.pipe, olap) | ||||
if err == WAIT_TIMEOUT: | ||||
|
r43703 | log("GetOverlappedResultEx timedout") | ||
raise SocketTimeout( | ||||
"timed out after waiting %dms for read" % self.timeout | ||||
) | ||||
|
r28432 | |||
|
r43703 | log("GetOverlappedResultEx reports error %d", err) | ||
self._raise_win_err("error while waiting for read", err) | ||||
|
r28432 | |||
nread = nread.value | ||||
if nread == 0: | ||||
# Docs say that named pipes return 0 byte when the other end did | ||||
# a zero byte write. Since we don't ever do that, the only | ||||
# other way this shows up is if the client has gotten in a weird | ||||
# state, so let's bail out | ||||
CancelIoEx(self.pipe, olap) | ||||
|
r43703 | raise IOError("Async read yielded 0 bytes; unpossible!") | ||
|
r28432 | |||
# Holds precisely the bytes that we read from the prior request | ||||
buf = buf[:nread] | ||||
returned_size = min(nread, size) | ||||
if returned_size == nread: | ||||
return buf | ||||
# keep any left-overs around for a later read to consume | ||||
self._iobuf = buf[returned_size:] | ||||
return buf[:returned_size] | ||||
def write(self, data): | ||||
olap = OVERLAPPED() | ||||
|
r30656 | olap.hEvent = self._waitable | ||
|
r43703 | immediate = WriteFile( | ||
self.pipe, ctypes.c_char_p(data), len(data), None, olap | ||||
) | ||||
|
r28432 | |||
if not immediate: | ||||
err = GetLastError() | ||||
if err != ERROR_IO_PENDING: | ||||
|
r43703 | self._raise_win_err( | ||
"failed to write %d bytes to handle %r" | ||||
% (len(data), self.pipe), | ||||
err, | ||||
) | ||||
|
r28432 | |||
# Obtain results, waiting if needed | ||||
nwrote = wintypes.DWORD() | ||||
|
r43703 | if self._get_overlapped_result_ex( | ||
self.pipe, olap, nwrote, 0 if immediate else self.timeout, True | ||||
): | ||||
log("made write of %d bytes", nwrote.value) | ||||
|
r28432 | return nwrote.value | ||
err = GetLastError() | ||||
# It's potentially unsafe to allow the write to continue after | ||||
# we unwind, so let's make a best effort to avoid that happening | ||||
CancelIoEx(self.pipe, olap) | ||||
if err == WAIT_TIMEOUT: | ||||
|
r43703 | raise SocketTimeout( | ||
"timed out after waiting %dms for write" % self.timeout | ||||
) | ||||
self._raise_win_err( | ||||
"error while waiting for write of %d bytes" % len(data), err | ||||
) | ||||
def _default_binpath(binpath=None): | ||||
if binpath: | ||||
return binpath | ||||
# The test harness sets WATCHMAN_BINARY to the binary under test, | ||||
# so we use that by default, otherwise, allow resolving watchman | ||||
# from the users PATH. | ||||
return os.environ.get("WATCHMAN_BINARY", "watchman") | ||||
|
r28432 | |||
class CLIProcessTransport(Transport): | ||||
""" open a pipe to the cli to talk to the service | ||||
This intended to be used only in the test harness! | ||||
The CLI is an oddball because we only support JSON input | ||||
and cannot send multiple commands through the same instance, | ||||
so we spawn a new process for each command. | ||||
We disable server spawning for this implementation, again, because | ||||
it is intended to be used only in our test harness. You really | ||||
should not need to use the CLI transport for anything real. | ||||
While the CLI can output in BSER, our Transport interface doesn't | ||||
support telling this instance that it should do so. That effectively | ||||
limits this implementation to JSON input and output only at this time. | ||||
It is the responsibility of the caller to set the send and | ||||
receive codecs appropriately. | ||||
""" | ||||
|
r43703 | |||
|
r28432 | proc = None | ||
closed = True | ||||
|
r43703 | def __init__(self, sockpath, timeout, binpath=None): | ||
|
r28432 | self.sockpath = sockpath | ||
self.timeout = timeout | ||||
|
r43703 | self.binpath = _default_binpath(binpath) | ||
|
r28432 | |||
def close(self): | ||||
if self.proc: | ||||
|
r30656 | if self.proc.pid is not None: | ||
self.proc.kill() | ||||
self.proc.stdin.close() | ||||
self.proc.stdout.close() | ||||
|
r43703 | self.proc.wait() | ||
|
r28432 | self.proc = None | ||
def _connect(self): | ||||
if self.proc: | ||||
return self.proc | ||||
args = [ | ||||
|
r43703 | self.binpath, | ||
"--sockname={0}".format(self.sockpath), | ||||
"--logfile=/BOGUS", | ||||
"--statefile=/BOGUS", | ||||
"--no-spawn", | ||||
"--no-local", | ||||
"--no-pretty", | ||||
"-j", | ||||
|
r28432 | ] | ||
|
r43703 | self.proc = subprocess.Popen( | ||
args, stdin=subprocess.PIPE, stdout=subprocess.PIPE | ||||
) | ||||
|
r28432 | return self.proc | ||
def readBytes(self, size): | ||||
self._connect() | ||||
res = self.proc.stdout.read(size) | ||||
|
r43703 | if not res: | ||
raise WatchmanError("EOF on CLI process transport") | ||||
|
r28432 | return res | ||
def write(self, data): | ||||
if self.closed: | ||||
|
r30656 | self.close() | ||
|
r28432 | self.closed = False | ||
self._connect() | ||||
res = self.proc.stdin.write(data) | ||||
self.proc.stdin.close() | ||||
self.closed = True | ||||
return res | ||||
class BserCodec(Codec): | ||||
""" use the BSER encoding. This is the default, preferred codec """ | ||||
|
r43703 | def __init__(self, transport, value_encoding, value_errors): | ||
super(BserCodec, self).__init__(transport) | ||||
self._value_encoding = value_encoding | ||||
self._value_errors = value_errors | ||||
|
r28432 | def _loads(self, response): | ||
|
r43703 | return bser.loads( | ||
response, | ||||
value_encoding=self._value_encoding, | ||||
value_errors=self._value_errors, | ||||
) | ||||
|
r28432 | |||
def receive(self): | ||||
buf = [self.transport.readBytes(sniff_len)] | ||||
if not buf[0]: | ||||
|
r43703 | raise WatchmanError("empty watchman response") | ||
|
r28432 | |||
|
r30656 | _1, _2, elen = bser.pdu_info(buf[0]) | ||
|
r28432 | |||
rlen = len(buf[0]) | ||||
while elen > rlen: | ||||
buf.append(self.transport.readBytes(elen - rlen)) | ||||
rlen += len(buf[-1]) | ||||
|
r43703 | response = b"".join(buf) | ||
|
r28432 | try: | ||
res = self._loads(response) | ||||
return res | ||||
except ValueError as e: | ||||
|
r43703 | raise WatchmanError("watchman response decode error: %s" % e) | ||
|
r28432 | |||
def send(self, *args): | ||||
|
r43703 | cmd = bser.dumps(*args) # Defaults to BSER v1 | ||
|
r28432 | self.transport.write(cmd) | ||
class ImmutableBserCodec(BserCodec): | ||||
""" use the BSER encoding, decoding values using the newer | ||||
immutable object support """ | ||||
def _loads(self, response): | ||||
|
r43703 | return bser.loads( | ||
response, | ||||
False, | ||||
value_encoding=self._value_encoding, | ||||
value_errors=self._value_errors, | ||||
) | ||||
|
r30656 | |||
class Bser2WithFallbackCodec(BserCodec): | ||||
""" use BSER v2 encoding """ | ||||
|
r43703 | def __init__(self, transport, value_encoding, value_errors): | ||
super(Bser2WithFallbackCodec, self).__init__( | ||||
transport, value_encoding, value_errors | ||||
) | ||||
if compat.PYTHON3: | ||||
bserv2_key = "required" | ||||
else: | ||||
bserv2_key = "optional" | ||||
self.send(["version", {bserv2_key: ["bser-v2"]}]) | ||||
|
r30656 | |||
capabilities = self.receive() | ||||
|
r43703 | if "error" in capabilities: | ||
raise BSERv1Unsupported( | ||||
"The watchman server version does not support Python 3. Please " | ||||
"upgrade your watchman server." | ||||
) | ||||
|
r30656 | |||
|
r43703 | if capabilities["capabilities"]["bser-v2"]: | ||
|
r30656 | self.bser_version = 2 | ||
self.bser_capabilities = 0 | ||||
else: | ||||
self.bser_version = 1 | ||||
self.bser_capabilities = 0 | ||||
def receive(self): | ||||
buf = [self.transport.readBytes(sniff_len)] | ||||
if not buf[0]: | ||||
|
r43703 | raise WatchmanError("empty watchman response") | ||
|
r30656 | |||
recv_bser_version, recv_bser_capabilities, elen = bser.pdu_info(buf[0]) | ||||
|
r43703 | if hasattr(self, "bser_version"): | ||
# Readjust BSER version and capabilities if necessary | ||||
self.bser_version = max(self.bser_version, recv_bser_version) | ||||
self.capabilities = self.bser_capabilities & recv_bser_capabilities | ||||
|
r30656 | |||
rlen = len(buf[0]) | ||||
while elen > rlen: | ||||
buf.append(self.transport.readBytes(elen - rlen)) | ||||
rlen += len(buf[-1]) | ||||
|
r43703 | response = b"".join(buf) | ||
|
r30656 | try: | ||
res = self._loads(response) | ||||
return res | ||||
except ValueError as e: | ||||
|
r43703 | raise WatchmanError("watchman response decode error: %s" % e) | ||
|
r30656 | |||
def send(self, *args): | ||||
|
r43703 | if hasattr(self, "bser_version"): | ||
cmd = bser.dumps( | ||||
*args, | ||||
version=self.bser_version, | ||||
capabilities=self.bser_capabilities | ||||
) | ||||
|
r30656 | else: | ||
cmd = bser.dumps(*args) | ||||
self.transport.write(cmd) | ||||
|
r28432 | |||
|
r43703 | class ImmutableBser2Codec(Bser2WithFallbackCodec, ImmutableBserCodec): | ||
""" use the BSER encoding, decoding values using the newer | ||||
immutable object support """ | ||||
pass | ||||
|
r28432 | class JsonCodec(Codec): | ||
""" Use json codec. This is here primarily for testing purposes """ | ||||
|
r43703 | |||
|
r28432 | json = None | ||
def __init__(self, transport): | ||||
super(JsonCodec, self).__init__(transport) | ||||
# optional dep on json, only if JsonCodec is used | ||||
import json | ||||
|
r43703 | |||
|
r28432 | self.json = json | ||
def receive(self): | ||||
line = self.transport.readLine() | ||||
try: | ||||
|
r30656 | # In Python 3, json.loads is a transformation from Unicode string to | ||
# objects possibly containing Unicode strings. We typically expect | ||||
# the JSON blob to be ASCII-only with non-ASCII characters escaped, | ||||
# but it's possible we might get non-ASCII bytes that are valid | ||||
# UTF-8. | ||||
if compat.PYTHON3: | ||||
|
r43703 | line = line.decode("utf-8") | ||
|
r28432 | return self.json.loads(line) | ||
except Exception as e: | ||||
print(e, line) | ||||
raise | ||||
def send(self, *args): | ||||
cmd = self.json.dumps(*args) | ||||
|
r30656 | # In Python 3, json.dumps is a transformation from objects possibly | ||
# containing Unicode strings to Unicode string. Even with (the default) | ||||
# ensure_ascii=True, dumps returns a Unicode string. | ||||
if compat.PYTHON3: | ||||
|
r43703 | cmd = cmd.encode("ascii") | ||
|
r30656 | self.transport.write(cmd + b"\n") | ||
|
r28432 | |||
class client(object): | ||||
""" Handles the communication with the watchman service """ | ||||
|
r43703 | |||
|
r28432 | sockpath = None | ||
transport = None | ||||
sendCodec = None | ||||
recvCodec = None | ||||
sendConn = None | ||||
recvConn = None | ||||
subs = {} # Keyed by subscription name | ||||
sub_by_root = {} # Keyed by root, then by subscription name | ||||
logs = [] # When log level is raised | ||||
|
r43703 | unilateral = ["log", "subscription"] | ||
|
r28432 | tport = None | ||
useImmutableBser = None | ||||
|
r43703 | pid = None | ||
|
r28432 | |||
|
r43703 | def __init__( | ||
self, | ||||
sockpath=None, | ||||
timeout=1.0, | ||||
transport=None, | ||||
sendEncoding=None, | ||||
recvEncoding=None, | ||||
useImmutableBser=False, | ||||
# use False for these two because None has a special | ||||
# meaning | ||||
valueEncoding=False, | ||||
valueErrors=False, | ||||
binpath=None, | ||||
): | ||||
|
r28432 | self.sockpath = sockpath | ||
self.timeout = timeout | ||||
self.useImmutableBser = useImmutableBser | ||||
|
r43703 | self.binpath = _default_binpath(binpath) | ||
|
r28432 | |||
|
r30656 | if inspect.isclass(transport) and issubclass(transport, Transport): | ||
self.transport = transport | ||||
|
r28432 | else: | ||
|
r43703 | transport = transport or os.getenv("WATCHMAN_TRANSPORT") or "local" | ||
if transport == "local" and os.name == "nt": | ||||
|
r30656 | self.transport = WindowsNamedPipeTransport | ||
|
r43703 | elif transport == "local": | ||
|
r30656 | self.transport = UnixSocketTransport | ||
|
r43703 | elif transport == "cli": | ||
|
r30656 | self.transport = CLIProcessTransport | ||
if sendEncoding is None: | ||||
|
r43703 | sendEncoding = "json" | ||
|
r30656 | if recvEncoding is None: | ||
recvEncoding = sendEncoding | ||||
else: | ||||
|
r43703 | raise WatchmanError("invalid transport %s" % transport) | ||
|
r28432 | |||
|
r43703 | sendEncoding = str( | ||
sendEncoding or os.getenv("WATCHMAN_ENCODING") or "bser" | ||||
) | ||||
recvEncoding = str( | ||||
recvEncoding or os.getenv("WATCHMAN_ENCODING") or "bser" | ||||
) | ||||
|
r28432 | |||
self.recvCodec = self._parseEncoding(recvEncoding) | ||||
self.sendCodec = self._parseEncoding(sendEncoding) | ||||
|
r43703 | # We want to act like the native OS methods as much as possible. This | ||
# means returning bytestrings on Python 2 by default and Unicode | ||||
# strings on Python 3. However we take an optional argument that lets | ||||
# users override this. | ||||
if valueEncoding is False: | ||||
if compat.PYTHON3: | ||||
self.valueEncoding = encoding.get_local_encoding() | ||||
self.valueErrors = encoding.default_local_errors | ||||
else: | ||||
self.valueEncoding = None | ||||
self.valueErrors = None | ||||
else: | ||||
self.valueEncoding = valueEncoding | ||||
if valueErrors is False: | ||||
self.valueErrors = encoding.default_local_errors | ||||
else: | ||||
self.valueErrors = valueErrors | ||||
def _makeBSERCodec(self, codec): | ||||
def make_codec(transport): | ||||
return codec(transport, self.valueEncoding, self.valueErrors) | ||||
return make_codec | ||||
|
r28432 | def _parseEncoding(self, enc): | ||
|
r43703 | if enc == "bser": | ||
|
r28432 | if self.useImmutableBser: | ||
|
r43703 | return self._makeBSERCodec(ImmutableBser2Codec) | ||
return self._makeBSERCodec(Bser2WithFallbackCodec) | ||||
elif enc == "bser-v1": | ||||
if compat.PYTHON3: | ||||
raise BSERv1Unsupported( | ||||
"Python 3 does not support the BSER v1 encoding: specify " | ||||
'"bser" or omit the sendEncoding and recvEncoding ' | ||||
"arguments" | ||||
) | ||||
if self.useImmutableBser: | ||||
return self._makeBSERCodec(ImmutableBserCodec) | ||||
return self._makeBSERCodec(BserCodec) | ||||
elif enc == "json": | ||||
|
r28432 | return JsonCodec | ||
else: | ||||
|
r43703 | raise WatchmanError("invalid encoding %s" % enc) | ||
|
r28432 | |||
def _hasprop(self, result, name): | ||||
if self.useImmutableBser: | ||||
return hasattr(result, name) | ||||
return name in result | ||||
def _resolvesockname(self): | ||||
# if invoked via a trigger, watchman will set this env var; we | ||||
# should use it unless explicitly set otherwise | ||||
|
r43703 | path = os.getenv("WATCHMAN_SOCK") | ||
|
r28432 | if path: | ||
return path | ||||
|
r43703 | cmd = [self.binpath, "--output-encoding=bser", "get-sockname"] | ||
|
r28432 | try: | ||
|
r43703 | args = dict( | ||
stdout=subprocess.PIPE, stderr=subprocess.PIPE | ||||
) # noqa: C408 | ||||
|
r30656 | |||
|
r43703 | if os.name == "nt": | ||
|
r30656 | # if invoked via an application with graphical user interface, | ||
# this call will cause a brief command window pop-up. | ||||
# Using the flag STARTF_USESHOWWINDOW to avoid this behavior. | ||||
startupinfo = subprocess.STARTUPINFO() | ||||
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW | ||||
|
r43703 | args["startupinfo"] = startupinfo | ||
|
r30656 | |||
|
r43703 | p = subprocess.Popen(cmd, **args) | ||
|
r30656 | |||
|
r28432 | except OSError as e: | ||
|
r43703 | raise WatchmanError('"watchman" executable not in PATH (%s)', e) | ||
|
r28432 | |||
stdout, stderr = p.communicate() | ||||
exitcode = p.poll() | ||||
if exitcode: | ||||
raise WatchmanError("watchman exited with code %d" % exitcode) | ||||
result = bser.loads(stdout) | ||||
|
r43703 | if "error" in result: | ||
raise WatchmanError("get-sockname error: %s" % result["error"]) | ||||
|
r28432 | |||
|
r43703 | return result["sockname"] | ||
|
r28432 | |||
def _connect(self): | ||||
""" establish transport connection """ | ||||
if self.recvConn: | ||||
|
r43703 | if self.pid != os.getpid(): | ||
raise UseAfterFork( | ||||
"do not re-use a connection after fork; open a new client instead" | ||||
) | ||||
|
r28432 | return | ||
if self.sockpath is None: | ||||
self.sockpath = self._resolvesockname() | ||||
|
r43703 | kwargs = {} | ||
if self.transport == CLIProcessTransport: | ||||
kwargs["binpath"] = self.binpath | ||||
self.tport = self.transport(self.sockpath, self.timeout, **kwargs) | ||||
|
r28432 | self.sendConn = self.sendCodec(self.tport) | ||
self.recvConn = self.recvCodec(self.tport) | ||||
|
r43703 | self.pid = os.getpid() | ||
|
r28432 | |||
def __del__(self): | ||||
self.close() | ||||
|
r43703 | def __enter__(self): | ||
self._connect() | ||||
return self | ||||
def __exit__(self, exc_type, exc_value, exc_traceback): | ||||
self.close() | ||||
|
r28432 | def close(self): | ||
if self.tport: | ||||
self.tport.close() | ||||
self.tport = None | ||||
self.recvConn = None | ||||
self.sendConn = None | ||||
def receive(self): | ||||
""" receive the next PDU from the watchman service | ||||
If the client has activated subscriptions or logs then | ||||
this PDU may be a unilateral PDU sent by the service to | ||||
inform the client of a log event or subscription change. | ||||
It may also simply be the response portion of a request | ||||
initiated by query. | ||||
There are clients in production that subscribe and call | ||||
this in a loop to retrieve all subscription responses, | ||||
so care should be taken when making changes here. | ||||
""" | ||||
self._connect() | ||||
result = self.recvConn.receive() | ||||
|
r43703 | if self._hasprop(result, "error"): | ||
raise CommandError(result["error"]) | ||||
|
r28432 | |||
|
r43703 | if self._hasprop(result, "log"): | ||
self.logs.append(result["log"]) | ||||
|
r28432 | |||
|
r43703 | if self._hasprop(result, "subscription"): | ||
sub = result["subscription"] | ||||
|
r28432 | if not (sub in self.subs): | ||
self.subs[sub] = [] | ||||
self.subs[sub].append(result) | ||||
# also accumulate in {root,sub} keyed store | ||||
|
r43703 | root = os.path.normpath(os.path.normcase(result["root"])) | ||
|
r28432 | if not root in self.sub_by_root: | ||
self.sub_by_root[root] = {} | ||||
if not sub in self.sub_by_root[root]: | ||||
self.sub_by_root[root][sub] = [] | ||||
self.sub_by_root[root][sub].append(result) | ||||
return result | ||||
def isUnilateralResponse(self, res): | ||||
|
r43703 | if "unilateral" in res and res["unilateral"]: | ||
|
r30656 | return True | ||
# Fall back to checking for known unilateral responses | ||||
|
r28432 | for k in self.unilateral: | ||
if k in res: | ||||
return True | ||||
return False | ||||
def getLog(self, remove=True): | ||||
""" Retrieve buffered log data | ||||
If remove is true the data will be removed from the buffer. | ||||
Otherwise it will be left in the buffer | ||||
""" | ||||
res = self.logs | ||||
if remove: | ||||
self.logs = [] | ||||
return res | ||||
def getSubscription(self, name, remove=True, root=None): | ||||
""" Retrieve the data associated with a named subscription | ||||
If remove is True (the default), the subscription data is removed | ||||
from the buffer. Otherwise the data is returned but left in | ||||
the buffer. | ||||
Returns None if there is no data associated with `name` | ||||
If root is not None, then only return the subscription | ||||
data that matches both root and name. When used in this way, | ||||
remove processing impacts both the unscoped and scoped stores | ||||
for the subscription data. | ||||
""" | ||||
if root is not None: | ||||
|
r43703 | root = os.path.normpath(os.path.normcase(root)) | ||
if root not in self.sub_by_root: | ||||
|
r28432 | return None | ||
|
r43703 | if name not in self.sub_by_root[root]: | ||
|
r28432 | return None | ||
sub = self.sub_by_root[root][name] | ||||
if remove: | ||||
del self.sub_by_root[root][name] | ||||
# don't let this grow unbounded | ||||
if name in self.subs: | ||||
del self.subs[name] | ||||
return sub | ||||
|
r43703 | if name not in self.subs: | ||
|
r28432 | return None | ||
sub = self.subs[name] | ||||
if remove: | ||||
del self.subs[name] | ||||
return sub | ||||
def query(self, *args): | ||||
""" Send a query to the watchman service and return the response | ||||
This call will block until the response is returned. | ||||
If any unilateral responses are sent by the service in between | ||||
the request-response they will be buffered up in the client object | ||||
and NOT returned via this method. | ||||
""" | ||||
|
r43703 | log("calling client.query") | ||
|
r28432 | self._connect() | ||
try: | ||||
self.sendConn.send(args) | ||||
res = self.receive() | ||||
while self.isUnilateralResponse(res): | ||||
res = self.receive() | ||||
return res | ||||
|
r30656 | except EnvironmentError as ee: | ||
# When we can depend on Python 3, we can use PEP 3134 | ||||
# exception chaining here. | ||||
raise WatchmanEnvironmentError( | ||||
|
r43703 | "I/O error communicating with watchman daemon", | ||
|
r30656 | ee.errno, | ||
ee.strerror, | ||||
|
r43703 | args, | ||
) | ||||
|
r30656 | except WatchmanError as ex: | ||
|
r28432 | ex.setCommand(args) | ||
|
r30656 | raise | ||
|
r28432 | |||
def capabilityCheck(self, optional=None, required=None): | ||||
""" Perform a server capability check """ | ||||
|
r43703 | res = self.query( | ||
"version", {"optional": optional or [], "required": required or []} | ||||
) | ||||
|
r28432 | |||
|
r43703 | if not self._hasprop(res, "capabilities"): | ||
|
r28432 | # Server doesn't support capabilities, so we need to | ||
# synthesize the results based on the version | ||||
capabilities.synthesize(res, optional) | ||||
|
r43703 | if "error" in res: | ||
raise CommandError(res["error"]) | ||||
|
r28432 | |||
return res | ||||
def setTimeout(self, value): | ||||
self.recvConn.setTimeout(value) | ||||
self.sendConn.setTimeout(value) | ||||