diff --git a/py4j-python/src/py4j/clientserver.py b/py4j-python/src/py4j/clientserver.py index 1517a523..88a31c21 100644 --- a/py4j-python/src/py4j/clientserver.py +++ b/py4j-python/src/py4j/clientserver.py @@ -23,7 +23,7 @@ disable_nagle) from py4j import protocol as proto from py4j.protocol import ( - Py4JError, Py4JNetworkError, smart_decode, get_command_part, + Py4JError, Py4JNetworkError, get_command_part, get_return_value, Py4JAuthenticationError) @@ -530,7 +530,7 @@ def send_command(self, command): try: while True: - answer = smart_decode(self.stream.readline()[:-1]) + answer = self.stream.readline()[:-1].decode("utf-8") logger.debug("Answer received: {0}".format(answer)) # Happens when a the other end is dead. There might be an empty # answer before the socket raises an error. @@ -541,7 +541,7 @@ def send_command(self, command): return answer[1:] else: command = answer - obj_id = smart_decode(self.stream.readline())[:-1] + obj_id = self.stream.readline()[:-1].decode("utf-8") if command == proto.CALL_PROXY_COMMAND_NAME: return_message = self._call_proxy(obj_id, self.stream) @@ -588,7 +588,7 @@ def wait_for_commands(self): authenticated = self.python_parameters.auth_token is None try: while True: - command = smart_decode(self.stream.readline())[:-1] + command = self.stream.readline()[:-1].decode("utf-8") if not authenticated: # Will raise an exception if auth fails in any way. authenticated = do_client_auth( @@ -596,7 +596,7 @@ def wait_for_commands(self): self.python_parameters.auth_token) continue - obj_id = smart_decode(self.stream.readline())[:-1] + obj_id = self.stream.readline()[:-1].decode("utf-8") logger.info( "Received command {0} on object id {1}". format(command, obj_id)) @@ -637,7 +637,7 @@ def _call_proxy(self, obj_id, input): get_command_part('Object ID unknown', self.pool) try: - method = smart_decode(input.readline())[:-1] + method = input.readline()[:-1].decode("utf-8") params = self._get_params(input) return_value = getattr(self.pool[obj_id], method)(*params) return proto.RETURN_MESSAGE + proto.SUCCESS +\ @@ -657,11 +657,11 @@ def _call_proxy(self, obj_id, input): def _get_params(self, input): params = [] - temp = smart_decode(input.readline())[:-1] + temp = input.readline()[:-1].decode("utf-8") while temp != proto.END: param = get_return_value("y" + temp, self.java_client) params.append(param) - temp = smart_decode(input.readline())[:-1] + temp = input.readline()[:-1].decode("utf-8") return params def __del__(self): diff --git a/py4j-python/src/py4j/java_gateway.py b/py4j-python/src/py4j/java_gateway.py index 3b5a3d2f..0792fa22 100644 --- a/py4j-python/src/py4j/java_gateway.py +++ b/py4j-python/src/py4j/java_gateway.py @@ -31,7 +31,7 @@ Py4JError, Py4JJavaError, Py4JNetworkError, Py4JAuthenticationError, get_command_part, get_return_value, - register_output_converter, smart_decode, escape_new_line, + register_output_converter, escape_new_line, is_fatal_error, is_error, unescape_new_line, get_error_message, compute_exception_message) from py4j.signals import Signal @@ -361,10 +361,13 @@ def launch_gateway(port=0, jarpath="", classpath="", javaopts=[], # ephemeral ports) _port = int(proc.stdout.readline()) - # Read the auth token from the server if enabled. + # Read the auth token from the server if enabled. stdout is in + # binary mode by default; decode here so the rest of the auth flow + # (which uses string equality, escape_new_line, etc.) sees str. _auth_token = None if enable_auth: - _auth_token = proc.stdout.readline()[:-len(os.linesep)] + _auth_token = proc.stdout.readline()[:-len(os.linesep)].decode( + "utf-8") # Start consumer threads so process does not deadlock/hangs OutputConsumer( @@ -638,7 +641,7 @@ def do_client_auth(command, input_stream, sock, auth_token): raise Py4JAuthenticationError("Expected {}, received {}.".format( proto.AUTH_COMMAND_NAME, command)) - client_token = smart_decode(input_stream.readline()[:-1]) + client_token = input_stream.readline()[:-1].decode("utf-8") # Remove the END marker input_stream.readline() if auth_token == client_token: @@ -669,8 +672,8 @@ def _garbage_collect_object(gateway_client, target_id): try: try: ThreadSafeFinalizer.remove_finalizer( - smart_decode(gateway_client.address) + - smart_decode(gateway_client.port) + + str(gateway_client.address) + + str(gateway_client.port) + target_id) gateway_client.garbage_collect_object(target_id) except Exception: @@ -747,7 +750,8 @@ def _pipe_fd(self, line): def run(self): lines_iterator = iter(self.stream.readline, b"") for line in lines_iterator: - self.redirect_func(smart_decode(line)) + # The sentinel b"" above pins line to bytes; decode directly. + self.redirect_func(line.decode("utf-8")) class ProcessConsumer(Thread): @@ -1278,7 +1282,11 @@ def send_command(self, command): "Error while sending", e, proto.ERROR_ON_SEND) try: - answer = smart_decode(self.stream.readline()[:-1]) + # Stream is opened in binary mode (socket.makefile("rb")), + # so readline() returns bytes; decode at the source rather + # than dispatch through smart_decode's isinstance check. + # Every JavaGateway call hits this — the saving compounds. + answer = self.stream.readline()[:-1].decode("utf-8") logger.debug("Answer received: {0}".format(answer)) if answer.startswith(proto.RETURN_MESSAGE): answer = answer[1:] @@ -1427,8 +1435,8 @@ def __init__(self, target_id, gateway_client): self._fully_populated = False self._gateway_doc = None - key = smart_decode(self._gateway_client.address) +\ - smart_decode(self._gateway_client.port) +\ + key = str(self._gateway_client.address) +\ + str(self._gateway_client.port) +\ self._target_id if self._gateway_client.gateway_property.enable_memory_management: @@ -2352,7 +2360,7 @@ def run(self): self.server_socket.listen(5) logger.info( "Socket listening on {0}". - format(smart_decode(self.server_socket.getsockname()))) + format(self.server_socket.getsockname())) server_started.send( self, server=self) @@ -2475,7 +2483,7 @@ def run(self): authenticated = self.callback_server_parameters.auth_token is None try: while True: - command = smart_decode(self.input.readline())[:-1] + command = self.input.readline()[:-1].decode("utf-8") if not authenticated: token = self.callback_server_parameters.auth_token # Will raise an exception if auth fails in any way. @@ -2483,7 +2491,7 @@ def run(self): command, self.input, self.socket, token) continue - obj_id = smart_decode(self.input.readline())[:-1] + obj_id = self.input.readline()[:-1].decode("utf-8") logger.info( "Received command {0} on object id {1}". format(command, obj_id)) @@ -2540,7 +2548,7 @@ def _call_proxy(self, obj_id, input): get_command_part('Object ID unknown', self.pool) try: - method = smart_decode(input.readline())[:-1] + method = input.readline()[:-1].decode("utf-8") params = self._get_params(input) return_value = getattr(self.pool[obj_id], method)(*params) return proto.RETURN_MESSAGE + proto.SUCCESS +\ @@ -2559,11 +2567,11 @@ def _call_proxy(self, obj_id, input): def _get_params(self, input): params = [] - temp = smart_decode(input.readline())[:-1] + temp = input.readline()[:-1].decode("utf-8") while temp != proto.END: param = get_return_value("y" + temp, self.gateway_client) params.append(param) - temp = smart_decode(input.readline())[:-1] + temp = input.readline()[:-1].decode("utf-8") return params @@ -2596,7 +2604,7 @@ def put(self, object, force_id=None): if force_id: id = force_id else: - id = proto.PYTHON_PROXY_PREFIX + smart_decode(self.next_id) + id = proto.PYTHON_PROXY_PREFIX + str(self.next_id) self.next_id += 1 self.dict[id] = object return id diff --git a/py4j-python/src/py4j/protocol.py b/py4j-python/src/py4j/protocol.py index 059d926d..bb14d36a 100644 --- a/py4j-python/src/py4j/protocol.py +++ b/py4j-python/src/py4j/protocol.py @@ -173,9 +173,22 @@ def escape_new_line(original): Backslashes are also escaped by another backslash. - :param original: the string to escape + :param original: the string to escape (str or bytes; bytes inputs + are decoded via smart_decode for backward compatibility — see + below). :rtype: an escaped string + + .. note:: + The internal ``smart_decode(original)`` is **load-bearing**: it + accepts bytes inputs that some legacy callers (and any code path + that forgot to decode at the socket boundary) might still + produce. Removing it makes ``bytes.replace("str", "str")`` + raise ``TypeError`` — see PR #575 review for the auth-token + regression this guarded against. The replacement chain is fast + enough that the smart_decode dispatch is not a hot-path concern; + all py4j-internal callers already pass str, so the type check + is a single isinstance hit. """ if original: return smart_decode(original).replace("\\", "\\\\").\ @@ -215,7 +228,10 @@ def smart_decode(s): def encode_float(float_value): - float_str = smart_decode(repr(float_value)) + # str(float) on Python 3 already returns the same shortest- + # roundtrip repr that smart_decode(repr(...)) was producing on + # py2; smart_decode here was a no-op dispatcher. + float_str = str(float_value) if float_str == "-inf": float_str = JAVA_NEGATIVE_INFINITY elif float_str == "inf": @@ -234,8 +250,14 @@ def encode_bytearray(barray): def decode_bytearray(encoded): - new_bytes = bytes(encoded, encoding="ascii") - return bytes([b for b in standard_b64decode(new_bytes)]) + # Per @PaperTsar's analysis in issue #570: the prior + # implementation built a Python list of ints (one PyObject per + # byte) then reconstructed bytes from that list — pure overhead + # now that Python 2 is no longer a target. standard_b64decode + # already returns bytes; the bytes() wrapper preserves the return- + # type contract while skipping the intermediate list. ~7.5x on + # 256KB payloads in microbenchmarks. + return bytes(standard_b64decode(encoded.encode("ascii"))) def is_python_proxy(parameter): diff --git a/py4j-python/src/py4j/tests/protocol_test.py b/py4j-python/src/py4j/tests/protocol_test.py index 40f876f4..59f0ae7d 100644 --- a/py4j-python/src/py4j/tests/protocol_test.py +++ b/py4j-python/src/py4j/tests/protocol_test.py @@ -311,5 +311,45 @@ def test_only_special_chars(self): self.assertEqual(self._roundtrip(s), s) +class EscapeNewLineBytesInputSafetyTest(unittest.TestCase): + """Pins the bytes-input safety contract on escape_new_line. + + escape_new_line's ``smart_decode(original)`` is a defensive measure + that lets bytes inputs pass through cleanly — see the docstring of + escape_new_line for the full rationale. PR #575's perf review + proposed dropping this smart_decode; doing so makes + ``bytes.replace("str", "str")`` raise TypeError, breaking the + auth-token round-trip path (testGatewayAuth) and any other path + that hands escape_new_line a bytes input without explicit decoding. + + If a future refactor drops smart_decode from escape_new_line, these + tests fail immediately — catching the regression before CI's + integration tests need to spin up a JVM.""" + + def test_bytes_input_decoded_as_utf8(self): + # ASCII bytes round-trip through escape_new_line as if they + # were str — smart_decode does the conversion. + result = escape_new_line(b"hello\nworld") + self.assertEqual(result, "hello\\nworld") + + def test_str_input_passes_through(self): + # str inputs are the common case; smart_decode is a single + # isinstance hit for these. + result = escape_new_line("hello\nworld") + self.assertEqual(result, "hello\\nworld") + + def test_bytes_input_with_utf8_payload(self): + # Non-ASCII bytes (UTF-8 encoded) decode correctly via + # smart_decode("utf-8") — auth tokens or other identifiers + # may contain UTF-8 bytes if read from stdout in binary mode. + s = "\u4e2d\u6587" # "中文" + result = escape_new_line(s.encode("utf-8")) + self.assertEqual(result, s) + + def test_empty_bytes_passes_through(self): + # The falsy-passthrough branch handles both b"" and "". + self.assertEqual(escape_new_line(b""), b"") + + if __name__ == "__main__": unittest.main()