perf(winrm): free streamed PSRP records to cap host RAM during prepare
run_streaming used index-based _drain, which read new records from
ps.output / ps.streams.{information,error} but never released the consumed
ones. pypsrp retains every record (each a heavy deserialised PSObject) for
the whole invocation, so a long prepare-win (Windows Update + toolchain,
thousands of lines) grew the host python process RAM without bound.
Replace _drain with _drain_clear, which emits each pending record then
deletes the consumed prefix (`del seq[:n]`). Safe because draining runs on
the same thread as poll_invoke — nothing appends while we mutate. Track an
explicit had_error flag since the error stream is now cleared as we go.
This commit is contained in:
@@ -51,11 +51,23 @@ def _info_text(record: Any) -> str:
|
||||
return str(data) if data is not None else str(record)
|
||||
|
||||
|
||||
def _drain(seq: list[Any], start: int, emit: Callable[[str], None]) -> int:
|
||||
"""Emit ``str()`` of every item in ``seq[start:]``; return the new length."""
|
||||
for item in seq[start:]:
|
||||
emit(str(item))
|
||||
return len(seq)
|
||||
def _drain_clear(
|
||||
seq: list[Any], render: Callable[[Any], str], emit: Callable[[str], None]
|
||||
) -> None:
|
||||
"""Emit ``render(item)`` for each item in ``seq`` then drop those items.
|
||||
|
||||
Clearing the consumed prefix is essential: pypsrp keeps every output/info/
|
||||
error record in these lists for the whole invocation, and each is a heavy
|
||||
deserialised PSObject. A long prepare (Windows Update + toolchain, thousands
|
||||
of lines) would otherwise grow the host process RAM without bound. We run on
|
||||
the same thread as ``poll_invoke`` so nothing appends while we mutate.
|
||||
"""
|
||||
if not seq:
|
||||
return
|
||||
n = len(seq)
|
||||
for item in seq[:n]:
|
||||
emit(render(item))
|
||||
del seq[:n]
|
||||
|
||||
|
||||
class WinRmTransport:
|
||||
@@ -224,20 +236,23 @@ class WinRmTransport:
|
||||
ps = PowerShell(pool)
|
||||
ps.add_script(script)
|
||||
ps.begin_invoke()
|
||||
out_i = info_i = err_i = 0
|
||||
running = (PSInvocationState.NOT_STARTED, PSInvocationState.RUNNING)
|
||||
had_error = False
|
||||
while ps.state in running:
|
||||
ps.poll_invoke()
|
||||
out_i = _drain(ps.output, out_i, emit_out)
|
||||
info_i = _drain(
|
||||
[_info_text(r) for r in ps.streams.information], info_i, on_line
|
||||
)
|
||||
err_i = _drain([str(e) for e in ps.streams.error], err_i, emit_err)
|
||||
_drain_clear(ps.output, str, emit_out)
|
||||
_drain_clear(ps.streams.information, _info_text, on_line)
|
||||
if ps.streams.error:
|
||||
had_error = True
|
||||
_drain_clear(ps.streams.error, str, emit_err)
|
||||
time.sleep(0.5)
|
||||
ps.end_invoke()
|
||||
_drain(ps.output, out_i, emit_out)
|
||||
_drain([str(e) for e in ps.streams.error], err_i, emit_err)
|
||||
returncode = 1 if ps.streams.error else 0
|
||||
_drain_clear(ps.output, str, emit_out)
|
||||
_drain_clear(ps.streams.information, _info_text, on_line)
|
||||
if ps.streams.error:
|
||||
had_error = True
|
||||
_drain_clear(ps.streams.error, str, emit_err)
|
||||
returncode = 1 if had_error else 0
|
||||
finally:
|
||||
with contextlib.suppress(Exception):
|
||||
pool.close()
|
||||
|
||||
Reference in New Issue
Block a user