aync

5 files changed, 127 insertions, 88 deletions

diff --git a/server/src/async.lua b/server/src/async.lua
new file mode 100644
index 00000000..045109ce
--- /dev/null
+++ b/server/src/async.lua
@@ -0,0 +1,97 @@
+local thread = require 'bee.thread'
+local errlog = thread.channel 'errlog'
+local taskId = 0
+local idlePool = {}
+local runningList = {}
+
+local function createTask()
+    taskId = taskId + 1
+    local id = taskId
+    local requestName  = 'request'  .. tostring(id)
+    local responseName = 'response' .. tostring(id)
+    thread.newchannel(requestName)
+    thread.newchannel(responseName)
+    local buf = ([[
+package.cpath  = %q
+package.path   = %q
+local thread   = require 'bee.thread'
+local request  = thread.channel(%q)
+local response = thread.channel(%q)
+local errlog   = thread.channel 'errlog'
+
+local function task()
+    local dump, upvalues = request:bpop()
+    local f, err = load(dump, '=request', 'b')
+    if not f then
+        errlog:push(err)
+        return
+    end
+    for i, value in ipairs(upvalues) do
+        debug.setupvalue(f, i+1, value)
+    end
+    local results = table.pack(pcall(f))
+    local ok = table.remove(results, 1)
+    if not ok then
+        local err = table.remove(results, 1)
+        errlog:push(err)
+        return
+    end
+    results.n = results.n - 1
+    response:push(results)
+end
+
+while true do
+    task()
+end
+]]):format(package.cpath, package.path, requestName, responseName)
+    log.debug('Create thread, id: ', id)
+    return {
+        id       = id,
+        thread   = thread.thread(buf),
+        request  = thread.channel(requestName),
+        response = thread.channel(responseName),
+    }
+end
+
+local function call(func, callback)
+    local task = table.remove(idlePool)
+    if not task then
+        task = createTask()
+    end
+    local upvalues = {}
+    local upId = 2 -- 跳过第一个上值 _ENV
+    while true do
+        local k, v = debug.getupvalue(func, upId)
+        if not k then
+            break
+        end
+        upvalues[#upvalues+1] = v
+        upId = upId + 1
+    end
+    local dump = string.dump(func)
+    runningList[task.id] = {
+        task     = task,
+        callback = callback,
+    }
+    task.request:push(dump, upvalues)
+end
+
+local function onTick()
+    local ok, msg = errlog:pop()
+    if ok then
+        log.error(msg)
+    end
+    for id, running in pairs(runningList) do
+        local ok, results = running.task.response:pop()
+        if ok then
+            runningList[id] = nil
+            idlePool[#idlePool+1] = running.task
+            xpcall(running.callback, log.debug, table.unpack(results))
+        end
+    end
+end
+
+return {
+    onTick = onTick,
+    call   = call,
+}
diff --git a/server/src/method/workspace/didChangeWorkspaceFolders.lua b/server/src/method/workspace/didChangeWorkspaceFolders.lua
index 12fe915d..6daca4c6 100644
--- a/server/src/method/workspace/didChangeWorkspaceFolders.lua
+++ b/server/src/method/workspace/didChangeWorkspaceFolders.lua
@@ -4,7 +4,7 @@ return function ()
     -- 暂不支持多个工作目录，因此当工作目录切换时，暴力结束服务，让前端重启服务
     rpc:requestWait('window/showMessageRequest', {
         type = 3,
-        message = '[Lua] dose not support multi workspace now, I may need to restart to support the new workspace ...',
+        message = '[Lua] dose not support multi workspace for now, I may need to restart to support the new workspace ...',
         actions = {
             {
                 title = 'Restart'
diff --git a/server/src/proto.lua b/server/src/proto.lua
index 78d347a1..4570de2e 100644
--- a/server/src/proto.lua
+++ b/server/src/proto.lua
@@ -1,21 +1,16 @@
-local thread   = require 'bee.thread'
-local json     = require 'json'
-local response = thread.channel 'response'
+local thread = require 'bee.thread'
+local json   = require 'json'
+local proto  = thread.channel 'proto'
+local errlog = thread.channel 'errlog'
 
-local log = setmetatable({}, {__index = function (self, level)
-    self[level] = function (...)
-        local t = table.pack(...)
-        for i = 1, t.n do
-            t[i] = tostring(t[i])
-        end
-        local buf = table.concat(t, '\t')
-        response:push('log', {
-            level = level,
-            buf   = buf,
-        })
+local function pushError(...)
+    local t = table.pack(...)
+    for i = 1, t.n do
+        t[i] = tostring(t[i])
     end
-    return self[level]
-end})
+    local buf = table.concat(t, '\t')
+    errlog:push(buf)
+end
 
 local function readProtoHeader()
     local header = io.read 'l'
@@ -24,7 +19,7 @@ local function readProtoHeader()
     elseif header:sub(1, #'Content-Type') == 'Content-Type' then
         return nil
     else
-        log.error('Proto header error:', header)
+        pushError('Proto header error:', header)
         return nil
     end
 end
@@ -32,7 +27,7 @@ end
 local function readProtoContent(header)
     local len = tonumber(header:match('%d+'))
     if not len then
-        log.error('Proto header error:', header)
+        pushError('Proto header error:', header)
         return nil
     end
     local buf = io.read(len+2)
@@ -41,7 +36,7 @@ local function readProtoContent(header)
     end
     local suc, res = pcall(json.decode, buf)
     if not suc then
-        log.error('Proto error:', buf)
+        pushError('Proto error:', buf)
         return nil
     end
     return res
@@ -56,7 +51,7 @@ local function readProto()
     if not data then
         return
     end
-    response:push('proto', data)
+    proto:push(data)
 end
 
 while true do
diff --git a/server/src/service.lua b/server/src/service.lua
index 1b77c2a9..1ec7625a 100644
--- a/server/src/service.lua
+++ b/server/src/service.lua
@@ -1,10 +1,13 @@
 local subprocess = require 'bee.subprocess'
 local method     = require 'method'
-local thread     = require 'thread'
+local thread     = require 'bee.thread'
+local async      = require 'async'
 local rpc        = require 'rpc'
 local parser     = require 'parser'
 local matcher    = require 'matcher'
 
+thread.newchannel 'proto'
+
 local ErrorCodes = {
     -- Defined by JSON RPC
     ParseError           = -32700,
@@ -214,10 +217,10 @@ function mt:removeText(uri)
     self._needCompile[uri] = nil
 end
 
-function mt:on_tick()
+function mt:onTick()
     while true do
-        local proto = thread.proto()
-        if not proto then
+        local ok, proto = self._proto:pop()
+        if not ok then
             break
         end
         if proto.method then
@@ -252,11 +255,15 @@ function mt:listen()
     io.stdin:setvbuf 'no'
     io.stdout:setvbuf 'no'
 
-    thread.require 'proto'
+    self._proto = thread.channel 'proto'
+
+    async.call(function ()
+        require 'proto'
+    end)
 
     while true do
-        thread.on_tick()
-        self:on_tick()
+        async.onTick()
+        self:onTick()
         thread.sleep(0.001)
     end
 end
diff --git a/server/src/thread.lua b/server/src/thread.lua
deleted file mode 100644
index af206a50..00000000
--- a/server/src/thread.lua
+++ /dev/null
@@ -1,60 +0,0 @@
-local thread     = require 'bee.thread'
-local response, errlog
-local proto_cache
-
-local api = {
-    log = function (data)
-        local level = data.level
-        local buf   = data.buf
-        log[level](buf)
-    end,
-    proto = function (data)
-        if not proto_cache then
-            proto_cache = {}
-        end
-        proto_cache[#proto_cache+1] = data
-    end,
-}
-
-local function on_tick()
-    if not response then
-        return
-    end
-    local ok, msg = errlog:pop()
-    if ok then
-        log.error(msg)
-    end
-    local ok, who, data = response:pop()
-    if ok then
-        api[who](data)
-    end
-end
-
-local function require(name)
-    if not response then
-        thread.newchannel 'response'
-        response = thread.channel 'response'
-        errlog   = thread.channel 'errlog'
-    end
-
-    local buf = ("\z
-        package.cpath = %q\n\z
-        package.path = %q\n\z
-        require %q\n\z
-    "):format(package.cpath, package.path, name)
-    return thread.thread(buf)
-end
-
-local function proto()
-    if not proto_cache then
-        return nil
-    end
-    return table.remove(proto_cache, 1)
-end
-
-return {
-    require = require,
-    on_tick = on_tick,
-    proto   = proto,
-    sleep   = thread.sleep,
-}