test: don't block to wait for previous session's exit (#35885)

This reduces the time taken by clear() by over 20% with ASAN.

test/client/uv_stream.lua

@@ -9,7 +9,7 @@ local uv = vim.uv
 --- @field write fun(self, data: string|string[])
 --- @field read_start fun(self, cb: fun(chunk: string))
 --- @field read_stop fun(self)
---- @field close fun(self, signal?: string)
+--- @field close fun(self, signal?: string, noblock?: boolean)
 
 --- Stream over given pipes.
 ---
@@ -126,6 +126,8 @@ end
 --- @field private _child_stdin uv.uv_pipe_t
 --- @field private _child_stdout uv.uv_pipe_t
 --- @field private _child_stderr uv.uv_pipe_t
+--- @field package _closed integer
+--- @field package _on_exit fun(closed: integer?)
 --- Collects stdout (if `collect_text=true`). Treats data as text (CRLF converted to LF).
 --- @field stdout string
 --- Collects stderr as raw data.
@@ -147,8 +149,10 @@ ProcStream.__index = ProcStream
 --- @param argv string[]
 --- @param env string[]?
 --- @param io_extra uv.uv_pipe_t?
+--- @param on_exit fun(closed: integer?)?  Called after the child process exits.
+--- `closed` is the timestamp (uv.now()) when close() was called, or nil if it wasn't.
 --- @return test.ProcStream
-function ProcStream.spawn(argv, env, io_extra)
+function ProcStream.spawn(argv, env, io_extra, on_exit)
   local self = setmetatable({
     collect_text = false,
     output = function(self)
@@ -165,6 +169,7 @@ function ProcStream.spawn(argv, env, io_extra)
     _child_stdout = assert(uv.new_pipe(false)),
     _child_stderr = assert(uv.new_pipe(false)),
     _exiting = false,
+    _on_exit = on_exit,
   }, ProcStream)
   local prog = argv[1]
   local args = {} --- @type string[]
@@ -181,6 +186,9 @@ function ProcStream.spawn(argv, env, io_extra)
     self.signal = signal
     -- "Abort" exit may not set status; force to nonzero in that case.
     self.status = (0 ~= (status or 0) or 0 == (signal or 0)) and status or (128 + (signal or 0))
+    if self._on_exit then
+      self._on_exit(self._closed)
+    end
   end)
 
   if not self._proc then
@@ -238,11 +246,11 @@ function ProcStream:read_stop()
   self._child_stderr:read_stop()
 end
 
-function ProcStream:close(signal)
+function ProcStream:close(signal, noblock)
   if self._closed then
     return
   end
-  self._closed = true
+  self._closed = uv.now()
   self:read_stop()
   self._child_stdin:close()
   self._child_stdout:close()
@@ -250,10 +258,12 @@ function ProcStream:close(signal)
   if type(signal) == 'string' then
     self._proc:kill('sig' .. signal)
   end
-  while self.status == nil do
-    uv.run 'once'
+  if not noblock then
+    while self.status == nil do
+      uv.run 'once'
+    end
+    return self.status, self.signal
   end
-  return self.status, self.signal
 end
 
 return {