refactor: speed up parsing/transform with stringbuf/worker

Author: phanen

lua/fzf-lua/lib/stringbuffer.lua

@@ -0,0 +1,113 @@
+---@diagnostic disable
+-- https://github.com/neovim/neovim/blob/20e77c5d886af54d1f7b6844cffc11129f579ad9/runtime/lua/vim/_core/stringbuffer.lua
+-- Basic shim for LuaJIT's stringbuffer.
+-- Note this does not implement the full API.
+-- This is intentionally internal-only. If we want to expose it, we should
+-- reimplement this a userdata and ship it as `string.buffer`
+-- (minus the FFI stuff) for Lua 5.1.
+local M = {}
+
+local has_strbuffer, strbuffer = pcall(require, "string.buffer")
+
+if has_strbuffer then
+  M.new = strbuffer.new
+
+  -- Lua 5.1 does not have __len metamethod so we need to provide a len()
+  -- function to use instead.
+
+  --- @param buf vim._core.stringbuffer
+  --- @return integer
+  function M.len(buf)
+    return #buf
+  end
+
+  return M
+end
+
+--- @class vim._core.stringbuffer
+--- @field private buf string[]
+--- @field package len integer absolute length of the `buf`
+--- @field package skip_ptr integer
+local StrBuffer = {}
+StrBuffer.__index = StrBuffer
+
+--- @return string
+function StrBuffer:tostring()
+  if #self.buf > 1 then
+    self.buf = { table.concat(self.buf) }
+  end
+
+  -- assert(self.len == #(self.buf[1] or ''), 'len mismatch')
+
+  if self.skip_ptr > 0 then
+    if self.buf[1] then
+      self.buf[1] = self.buf[1]:sub(self.skip_ptr + 1)
+      self.len = self.len - self.skip_ptr
+    end
+    self.skip_ptr = 0
+  end
+
+  -- assert(self.len == #(self.buf[1] or ''), 'len mismatch')
+
+  return self.buf[1] or ""
+end
+
+StrBuffer.__tostring = StrBuffer.tostring
+
+--- @private
+--- Efficiently peak at the first `n` characters of the buffer.
+--- @param n integer
+--- @return string
+function StrBuffer:_peak(n)
+  local skip, buf1 = self.skip_ptr, self.buf[1]
+  if buf1 and (n + skip) < #buf1 then
+    return buf1:sub(skip + 1, skip + n)
+  end
+  return self:tostring():sub(1, n)
+end
+
+--- @param chunk string
+function StrBuffer:put(chunk)
+  local s = tostring(chunk)
+  self.buf[#self.buf + 1] = s
+  self.len = self.len + #s
+  return self
+end
+
+--- @param str string
+function StrBuffer:set(str)
+  return self:reset():put(str)
+end
+
+--- @param n? integer
+--- @return string
+function StrBuffer:get(n)
+  n = n or self.len
+  local r = self:_peak(n)
+  self:skip(n)
+  return r
+end
+
+--- @param n integer
+function StrBuffer:skip(n)
+  self.skip_ptr = math.min(self.len, self.skip_ptr + n)
+  return self
+end
+
+function StrBuffer:reset()
+  self.buf = {}
+  self.skip_ptr = 0
+  self.len = 0
+  return self
+end
+
+function M.new()
+  return setmetatable({}, StrBuffer):reset()
+end
+
+--- @param buf vim._core.stringbuffer
+function M.len(buf)
+  return buf.len - buf.skip_ptr
+end
+
+return M

lua/fzf-lua/libuv.lua

@@ -84,7 +84,6 @@ end
 ---@field EOL_data? string
 ---@field process1? boolean
 ---@field profiler? boolean
----@field use_queue? boolean
 
 ---@param opts fzf-lua.SpawnOpts
 ---@param fn_transform function?
@@ -103,34 +102,31 @@ M.spawn = function(opts, fn_transform, fn_done)
         or opts.cmd:match("%s%-%-null")
         or opts.cmd:match("%s%-Z"))
       and "\0" or "\n"
+  local EOL_byte = EOL_data:byte()
   local output_pipe = assert(uv.new_pipe(false))
   local error_pipe = assert(uv.new_pipe(false))
-  local write_cb_count, read_cb_count = 0, 0
-  local prev_line_content ---@type string?
+  local write_cb_count = 0
   local handle, pid ---@type uv.uv_process_t, integer
-  local co = coroutine.running()
-  local queue = require("fzf-lua.lib.queue").new()
+  local strbuf = (vim.F.nil_wrap(require)("vim._core.stringbuffer") or
+    require("fzf-lua.lib.stringbuffer")).new()
   local work_ctx
 
-  -- Disable queue if running headless due to
-  -- "Attempt to yield across a C-call boundary"
-  opts.use_queue = not _G._fzf_lua_is_headless and opts.use_queue
-
   -- cb_write_lines trumps cb_write
   ---@diagnostic disable-next-line: assign-type-mismatch
   if opts.cb_write_lines then opts.cb_write = opts.cb_write_lines end
 
   local can_finish = function()
     if not output_pipe:is_active() -- EOF signalled or process is aborting
-        and read_cb_count == 0     -- no outstanding read_cb data processing
         and write_cb_count == 0    -- no outstanding write callbacks
+        and #strbuf == 0
     then
       return true
     end
   end
 
   ---@diagnostic disable-next-line: redefined-local
   local finish = function(code, sig, from, pid)
+    os.execute("notify-send " .. write_cb_count)
     -- Uncomment to debug pipe closure timing issues (#1521)
     -- output_pipe:close(function() print("closed o") end)
     -- error_pipe:close(function() print("closed e") end)
@@ -139,7 +135,7 @@ M.spawn = function(opts, fn_transform, fn_done)
     if opts.cb_finish then
       opts.cb_finish(code, sig, from, pid)
     end
-    queue:clear()
+    strbuf:reset()
     if not handle:is_closing() then
       handle:kill("sigterm")
       vim.defer_fn(function()
@@ -187,7 +183,7 @@ M.spawn = function(opts, fn_transform, fn_done)
   if opts.cb_pid then opts.cb_pid(pid) end
 
   local function write_cb(data)
-    write_cb_count = write_cb_count + 1
+    -- write_cb_count = write_cb_count + 1
     opts.cb_write(data, function(err)
       write_cb_count = write_cb_count - 1
       if err then
@@ -204,126 +200,75 @@ M.spawn = function(opts, fn_transform, fn_done)
   end
 
   ---@param data string data stream
-  ---@param prev string? rest of line from previous call
-  ---@param trans function? line transformation function
-  ---@return table, string? line array, partial last line (no EOL)
-  local function split_lines(data, prev, trans)
+  ---@return string, string? line array, partial last line (no EOL)
+  local function split_lines(data)
+    -- io.stderr:write("[DEBUG] worker init")
+    if not _G.fzf_lua_worker_init then
+      _G.fzf_lua_worker_init = true
+      -- local __FILE__ = assert(debug.getinfo(1, "S")).source:gsub("^@", "")
+      -- package.path = ("%s/?.lua;"):format(vim.fs.dirname(vim.fs.dirname(__FILE__))) .. package.path
+      -- require("fzf-lua")
+      -- pcall(require, "fzf-lua.make_entry")
+    end
     local ret = {}
     local start_idx = 1
     repeat
-      local nl_idx = data:find(EOL_data, start_idx, true)
+      local nl_idx = data:find(EOL_data or "\n", start_idx, true)
       if nl_idx then
         local cr = data:byte(nl_idx - 1, nl_idx - 1) == 13 -- \r
         local line = data:sub(start_idx, nl_idx - (cr and 2 or 1))
-        if prev then
-          line = prev .. line
-          prev = nil
-        end
-        if trans then line = trans(line) end
-        if line then table.insert(ret, line) end
+        -- if trans then line = trans(line) end
+        if line then ret[#ret + 1] = line end
         start_idx = nl_idx + 1
-      else
-        -- assert(start_idx <= #data)
-        if prev and #prev > 4096 then
-          -- chunk size is 64K, limit previous line length to 4K
-          -- max line length is therefor 4K + 64K (leftover + full chunk)
-          -- without this we can memory fault on extremely long lines (#185)
-          -- or have UI freezes (#211)
-          prev = prev:sub(1, 4096)
-        end
-        prev = (prev or "") .. data:sub(start_idx)
       end
     until not nl_idx or start_idx > #data
-    return ret, prev
-  end
-
-  --- Called with nil to process the leftover data
-  ---@param data string?
-  local process_data = function(data)
-    if not data and prev_line_content then
-      data = prev_line_content .. EOL
-      prev_line_content = nil
-    end
-    if not data then
-      -- NOTE: this isn't called when prev_line_content is not nil but that's
-      -- not a problem as the write_cb will call finish once the callback is done
-      -- since the output_pipe is already in "closing" state
-      if can_finish() then
-        finish(0, 0, "[EOF]", pid)
-      end
-      return
-    end
-    if not fn_transform then
-      write_cb(data)
-    else
-      -- NOTE: cannot use due to "yield across a C-call boundary"
-      -- if co and not work_ctx then
-      --   work_ctx = uv.new_work(split_lines, function(lines, prev)
-      --     coroutine.resume(co, lines, prev)
-      --   end)
-      -- end
-      local nlines, lines = 0, nil
-      local t_st = opts.profiler and uv.hrtime()
-      if t_st then write_cb(string.format("[DEBUG] start: %.0f (ns)" .. EOL, t_st)) end
-      if work_ctx then
-        -- should never get here, work_ctx is never initialized
-        -- code remains as a solemn reminder to my efforts of making
-        -- multiprocess=false a lag free experience
-        if prev_line_content then uv.queue_work(work_ctx, data, prev_line_content) end
-        lines, prev_line_content = coroutine.yield()
-      else
-        lines, prev_line_content = split_lines(data, prev_line_content,
-          -- NOTE `fn_transform=true` is used to force line split without transformation
-          type(fn_transform) == "function" and fn_transform or nil)
+    ret[#ret + 1] = ""
+    return table.concat(ret, "\n")
+  end
+
+  work_ctx = uv.new_work(split_lines, write_cb)
+
+  local co = coroutine.create(function()
+    local stop = 0
+    while true do
+      local len = #strbuf
+      local ref = strbuf:ref()
+      if output_pipe:is_closing() then
+        if len == 0 then return end
+        if ref[len - 1] ~= EOL_byte then strbuf:put(EOL_byte) end -- make split_lines happy
+        write_cb_count = write_cb_count + 1
+        return work_ctx:queue(strbuf:get())
       end
-      nlines = nlines + #lines
-      if #lines > 0 then
-        if opts.cb_write_lines then
-          write_cb(lines)
-        else
-          -- Testing shows better performance writing the entire table at once as opposed to
-          -- calling 'write_cb' for every line after 'fn_transform', we therefore only use
-          -- `process1` when using "mini.icons" as `vim.filetype.match` causes a signigicant
-          -- delay and having to wait for all lines to be processed has an apparent lag
-          if opts.process1 then
-            vim.tbl_map(function(l) write_cb(l .. EOL) end, lines)
-          else
-            write_cb(table.concat(lines, EOL) .. EOL)
-          end
+      local eol = len
+      for i = len - 1, stop, -1 do
+        if ref[i] == EOL_byte then
+          eol = i
+          break
         end
       end
-      if t_st then
-        local t_e = vim.uv.hrtime()
-        write_cb(string.format("[DEBUG] finish:%.0f (ns) %d lines took %.0f (ms)" .. EOL,
-          t_e, nlines, (t_e - t_st) / 1e6))
+      if eol == len then
+        stop = len -- no EOL found, wait for more data
+        coroutine.yield()
+      else
+        local data = strbuf:get(eol + 1)
+        stop = #strbuf
+        write_cb_count = write_cb_count + 1
+        work_ctx:queue(data)
       end
     end
-  end
+    if can_finish() then finish(0, 0, "[EOF]", pid) end
+  end)
 
   local read_cb = function(err, data)
     if err then
-      finish(130, 0, "[read_cb: err]", pid)
-      return
-    end
-    if not data then
-      -- EOF signalled, we can close the pipe
+      return finish(130, 0, "[read_cb: err]", pid)
+    elseif data then
+      strbuf:put(data)
+    else -- EOF signalled, we can close the pipe
       output_pipe:close()
     end
-    if opts.use_queue then
-      if data then queue:push(data) end
-      -- Either we have outstanding data enqueued or the pipe is closing
-      -- due to the above `output_pipe:close`, in both cases we need to
-      -- resume the dequeue loop
-      coroutine.resume(co)
-    else
-      read_cb_count = read_cb_count + 1
-      local process = function()
-        read_cb_count = read_cb_count - 1
-        process_data(data)
-      end
-      -- Schedule data processing if we're in fast event
-      -- avoids "attempt to yield across C-call boundary" by using vim.schedule
-      if vim.in_fast_event() then vim.schedule(process) else process() end
+    if #strbuf > 0 or output_pipe:is_closing() then
+      assert(coroutine.resume(co))
     end
   end
 
@@ -352,20 +297,6 @@ M.spawn = function(opts, fn_transform, fn_done)
     error_pipe:read_start(err_cb)
   end
 
-  if opts.use_queue then
-    while not (output_pipe:is_closing() and queue:empty()) do
-      if queue:empty() then
-        coroutine.yield()
-      else
-        process_data(queue:pop())
-      end
-    end
-    -- process the leftover line from `processs_data`
-    -- will call `finish` immediately if there's no last line
-    -- otherwise, finish is called in the write callback
-    process_data(nil)
-  end
-
   return handle, pid
 end
 
@@ -628,6 +559,7 @@ M.spawn_stdio = function(opts)
       cmd = cmd,
       cb_finish = on_finish,
       cb_write = on_write,
+      -- cb_write_lines = on_write_lines,
       cb_err = on_err,
       process1 = opts.process1,
       profiler = opts.profiler,

lua/fzf-lua/make_entry.lua

@@ -1,5 +1,6 @@
 local M = {}
 
+-- do return { file = nil } end
 ---@diagnostic disable-next-line: deprecated
 local uv = vim.uv or vim.loop
 local path = require "fzf-lua.path"