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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
|
--!A cross-platform build utility based on Lua
--
-- Licensed under the Apache License, Version 2.0 (the "License");
-- you may not use this file except in compliance with the License.
-- You may obtain a copy of the License at
--
-- http://www.apache.org/licenses/LICENSE-2.0
--
-- Unless required by applicable law or agreed to in writing, software
-- distributed under the License is distributed on an "AS IS" BASIS,
-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-- See the License for the specific language governing permissions and
-- limitations under the License.
--
-- Copyright (C) 2015 - 2019, TBOOX Open Source Group.
--
-- @author ruki
-- @file scheduler.lua
--
-- define module: scheduler
local scheduler = scheduler or {}
local _coroutine = _coroutine or {}
-- load modules
local table = require("base/table")
local option = require("base/option")
local string = require("base/string")
local poller = require("base/poller")
local timer = require("base/timer")
local coroutine = require("base/coroutine")
-- new a coroutine instance
function _coroutine.new(name, thread)
local instance = table.inherit(_coroutine)
instance._NAME = name
instance._THREAD = thread
setmetatable(instance, _coroutine)
return instance
end
-- get the coroutine name
function _coroutine:name()
return self._NAME or "none"
end
-- set the coroutine name
function _coroutine:name_set(name)
self._NAME = name
end
-- get the raw coroutine thread
function _coroutine:thread()
return self._THREAD
end
-- get the coroutine status
function _coroutine:status()
return coroutine.status(self:thread())
end
-- is dead?
function _coroutine:is_dead()
return self:status() == "dead"
end
-- is running?
function _coroutine:is_running()
return self:status() == "running"
end
-- is suspended?
function _coroutine:is_suspend()
return self:status() == "suspend"
end
-- tostring(socket)
function _coroutine:__tostring()
return string.format("<co: %s/%s>", self:thread(), self:name())
end
-- gc(coroutine)
function _coroutine:__gc()
self._THREAD = nil
end
-- get the timer of scheduler
function scheduler:_timer()
local t = self._TIMER
if t == nil then
t = timer:new()
self._TIMER = t
end
return t
end
-- wait the current coroutine
function scheduler:_co_wait(...)
return self:co_suspend(...)
end
-- wake the given coroutine
function scheduler:_co_wake(co, ...)
return self:co_resume(co, ...)
end
-- start a new coroutine task
function scheduler:co_start(cotask, ...)
return self:co_start_named(nil, cotask, ...)
end
-- start a new named coroutine task
function scheduler:co_start_named(coname, cotask, ...)
local co = _coroutine.new(coname, coroutine.create(cotask))
self:co_tasks()[co:thread()] = co
local ok, errors = scheduler:co_resume(co, ...)
self:co_tasks()[co:thread()] = nil
if not ok then
return nil, errors
end
return co
end
-- resume the given coroutine
function scheduler:co_resume(co, ...)
return coroutine.resume(co:thread(), ...)
end
-- suspend the current coroutine
function scheduler:co_suspend(...)
return coroutine.yield(...)
end
-- get the current running coroutine
function scheduler:co_running()
local running = coroutine.running()
return running and self:co_tasks()[running] or nil
end
-- get all coroutine tasks
function scheduler:co_tasks()
local cotasks = self._CO_TASKS
if not cotasks then
cotasks = {}
self._CO_TASKS = cotasks
end
return cotasks
end
-- wait socket events
function scheduler:waitsock(sock, events, timeout)
-- TODO
return 0
end
-- sleep some times (ms)
function scheduler:sleep(ms)
local running = self:co_running()
if not running then
return false, "we must call sleep() in coroutine with scheduler!"
end
self:_timer():post(function (cancel)
self:co_resume(running)
end, ms)
self:co_suspend()
return true
end
-- stop the scheduler loop
function scheduler:stop()
-- TODO post a kill signal to poller
-- stop timer and cancel all tasks
self._STARTED = false
return true
end
-- run loop, schedule coroutine with socket/io and sub-processes
function scheduler:runloop()
-- start loop
self._STARTED = true
-- run loop
opt = opt or {}
local ok = true
local errors = nil
local timeout = -1
while self._STARTED do
-- get the next timeout
timeout = self:_timer():delay() or 1000
-- wait events
local count, events = poller:wait(timeout)
if count < 0 then
ok = false
errors = events
break
end
-- spank the timer and trigger all timeout tasks
self:_timer():next()
-- resume all suspended tasks with events
-- TODO
end
-- mark the loop as stopped first
self._STARTED = false
-- TODO resume all suspended tasks
-- we cannot suspend them now, all tasks will be exited directly and free all resources.
-- ...
-- finished
return ok, errors
end
-- return module: scheduler
return scheduler
|
