diff --git a/packages/@n8n/task-runner-python/src/constants.py b/packages/@n8n/task-runner-python/src/constants.py index 155657a69f1..d2b12fbd6f4 100644 --- a/packages/@n8n/task-runner-python/src/constants.py +++ b/packages/@n8n/task-runner-python/src/constants.py @@ -77,7 +77,7 @@ IGNORED_ERROR_TYPES = ( # Logging LOG_FORMAT = "%(asctime)s.%(msecs)03d\t%(levelname)s\t%(message)s" LOG_TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S" -LOG_TASK_COMPLETE = 'Completed task {task_id} in {duration} for node "{node_name}" ({node_id}) in workflow "{workflow_name}" ({workflow_id})' +LOG_TASK_COMPLETE = 'Completed task {task_id} in {duration} ({result_size}) for node "{node_name}" ({node_id}) in workflow "{workflow_name}" ({workflow_id})' LOG_TASK_CANCEL = 'Cancelled task {task_id} for node "{node_name}" ({node_id}) in workflow "{workflow_name}" ({workflow_id})' LOG_TASK_CANCEL_UNKNOWN = ( "Received cancel for unknown task: {task_id}. Discarding message." diff --git a/packages/@n8n/task-runner-python/src/task_executor.py b/packages/@n8n/task-runner-python/src/task_executor.py index 42e9ecdc50f..7e8bb3b3b89 100644 --- a/packages/@n8n/task-runner-python/src/task_executor.py +++ b/packages/@n8n/task-runner-python/src/task_executor.py @@ -5,7 +5,6 @@ import textwrap import json import os import sys -import tempfile import logging from src.errors import ( @@ -27,6 +26,7 @@ from src.constants import ( from typing import Any, Set from multiprocessing.context import ForkServerProcess +from multiprocessing import shared_memory logger = logging.getLogger(__name__) @@ -79,7 +79,7 @@ class TaskExecutor: queue: multiprocessing.Queue, task_timeout: int, continue_on_fail: bool, - ) -> tuple[list, PrintArgs]: + ) -> tuple[list, PrintArgs, int]: """Execute a subprocess for a Python code task.""" print_args: PrintArgs = [] @@ -119,23 +119,29 @@ class TaskExecutor: if "error" in returned: raise TaskRuntimeError(returned["error"]) - if "result_file" not in returned: + if "shm_name" not in returned: raise TaskResultMissingError() - result_file = returned["result_file"] + shm_name = returned["shm_name"] + shm_size = returned["shm_size"] try: - with open(result_file, "r", encoding="utf-8") as f: - result = json.load(f) - finally: - os.unlink(result_file) + shm = shared_memory.SharedMemory(name=shm_name) + try: + json_str = bytes(shm.buf[:shm_size]).decode("utf-8") + result = json.loads(json_str) + finally: + shm.close() + shm.unlink() + except FileNotFoundError: + raise TaskResultMissingError() print_args = returned.get("print_args", []) - return result, print_args + return result, print_args, shm_size except Exception as e: if continue_on_fail: - return [{"json": {"error": str(e)}}], print_args + return [{"json": {"error": str(e)}}], print_args, 0 raise @staticmethod @@ -249,18 +255,22 @@ class TaskExecutor: def _put_result( queue: multiprocessing.Queue, result: list[Any], print_args: PrintArgs ): - with tempfile.NamedTemporaryFile( - mode="w", - delete=False, - prefix="n8n_result_", - suffix=".json", - encoding="utf-8", - ) as f: - json.dump(result, f, default=str, ensure_ascii=False) - result_file = f.name + json_bytes = json.dumps(result, default=str, ensure_ascii=False).encode("utf-8") + json_bytes_size = len(json_bytes) + + shm = shared_memory.SharedMemory(create=True, size=json_bytes_size) + shm.buf[:json_bytes_size] = json_bytes print_args_to_send = TaskExecutor._truncate_print_args(print_args) - queue.put({"result_file": result_file, "print_args": print_args_to_send}) + queue.put( + { + "shm_name": shm.name, + "shm_size": json_bytes_size, # stay exact, shm.size can round up for alignment + "print_args": print_args_to_send, + } + ) + + shm.close() @staticmethod def _put_error(queue: multiprocessing.Queue, e: Exception, print_args: PrintArgs): diff --git a/packages/@n8n/task-runner-python/src/task_runner.py b/packages/@n8n/task-runner-python/src/task_runner.py index 5ecec71ccad..6321cf8112b 100644 --- a/packages/@n8n/task-runner-python/src/task_runner.py +++ b/packages/@n8n/task-runner-python/src/task_runner.py @@ -303,7 +303,7 @@ class TaskRunner: task_state.process = process - result, print_args = await asyncio.to_thread( + result, print_args, result_size_bytes = await asyncio.to_thread( self.executor.execute_process, process=process, queue=queue, @@ -323,6 +323,7 @@ class TaskRunner: LOG_TASK_COMPLETE.format( task_id=task_id, duration=self._get_duration(start_time), + result_size=self._get_result_size(result_size_bytes), **task_state.context(), ) ) @@ -380,6 +381,8 @@ class TaskRunner: serialized = self.serde.serialize_runner_message(message) await self.websocket_connection.send(serialized) + # ========== Formatting ========== + def _get_duration(self, start_time: float) -> str: elapsed = time.time() - start_time @@ -391,6 +394,14 @@ class TaskRunner: return f"{int(elapsed) // 60}m" + def _get_result_size(self, size_bytes: int) -> str: + if size_bytes < 1024: + return f"{size_bytes} bytes" + elif size_bytes < 1024 * 1024: + return f"{size_bytes / 1024:.1f} KB" + else: + return f"{size_bytes / (1024 * 1024):.1f} MB" + # ========== Offers ========== async def _send_offers_loop(self) -> None: