1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
|
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, env = request:bpop()
if env then
setmetatable(env, { __index = _ENV })
else
env = _ENV
end
local f, err = load(dump, '=task', 't', env)
if not f then
errlog:push(err)
return
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(dump, env, callback)
local task = table.remove(idlePool)
if not task then
task = createTask()
end
runningList[task.id] = {
task = task,
callback = callback,
}
task.request:push(dump, env)
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,
require = require,
}
|
