<commit_before>--!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-2020, 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 utils     = require("base/utils")
local option    = require("base/option")
local string    = require("base/string")
local poller    = require("base/poller")
local timer     = require("base/timer")
local hashset   = require("base/hashset")
local coroutine = require("base/coroutine")
local bit       = require("bit")

-- 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 waiting poller object
function _coroutine:waitobj()
    return self._WAITOBJ
end

-- set the waiting poller object
function _coroutine:waitobj_set(obj)
    self._WAITOBJ = obj
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_suspended()
    return self:status() == "suspended"
end

-- get the current timer task
function _coroutine:_timer_task()
    return self._TIMER_TASK
end

-- set the timer task
function _coroutine:_timer_task_set(task)
    self._TIMER_TASK = task
end

-- tostring(coroutine)
function _coroutine:__tostring()
    return string.format("<co: %s/%s>", self:name(), self:status())
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

-- get poller object data for socket, pipe or process object 
function scheduler:_poller_data(obj)
    return self._POLLERDATA and self._POLLERDATA[obj] or nil
end

-- set poller object data
--
-- data.co_recv:            the suspended coroutine for waiting poller/recv
-- data.co_send:            the suspended coroutine for waiting poller/send
-- data.poller_events_wait: the waited events for poller
-- data.poller_events_save: the saved events for poller (triggered)
--
function scheduler:_poller_data_set(obj, data)
    local pollerdata = self._POLLERDATA 
    if not pollerdata then
        pollerdata = {}
        self._POLLERDATA = pollerdata
    end
    pollerdata[obj] = data
end

-- resume the suspended coroutine after poller callback
function scheduler:_poller_resume_co(co, events)

    -- cancel timer task if exists
    local timer_task = co:_timer_task()
    if timer_task then
        timer_task.cancel = true
    end

    -- the scheduler has been stopped? mark events as error to stop the coroutine
    if not self._STARTED then
        events = poller.EV_POLLER_ERROR
    end

    -- this coroutine must be suspended
    assert(co:is_suspended())

    -- resume this coroutine task
    co:waitobj_set(nil)
    return self:co_resume(co, (bit.band(events, poller.EV_POLLER_ERROR) ~= 0) and -1 or events)
end

-- the poller events callback
function scheduler:_poller_events_cb(obj, events)

    -- get poller object data
    local pollerdata = self:_poller_data(obj)
    if not pollerdata then
        -- we cannot use assert(, ""), because this will cause the object to be serialized in advance
        raise("%s: cannot get poller data!", obj)
    end

    -- is process object?
    if obj:otype() == poller.OT_PROC then

        -- resume coroutine and return the process exit status
        pollerdata.proc_status = events

        -- waiting process? resume this coroutine
        if pollerdata.co_waiting then
            local co_waiting = pollerdata.co_waiting
            pollerdata.co_waiting = nil
            return self:_poller_resume_co(co_waiting, 1)
        else 
            pollerdata.proc_pending = 1
        end
        return ;
    end

    -- get poller object events
    local events_prev_wait = pollerdata.poller_events_wait
    local events_prev_save = pollerdata.poller_events_save

    -- eof for edge trigger?
    if bit.band(events, poller.EV_POLLER_EOF) ~= 0 then
        -- cache this eof as next recv/send event
        events = bit.band(events, bit.bnot(poller.EV_POLLER_EOF))
        events_prev_save = bit.bor(events_prev_save, events_prev_wait)
        pollerdata.poller_events_save = events_prev_save
    end

    -- get the waiting coroutines
    local co_recv = bit.band(events, poller.EV_POLLER_RECV) ~= 0 and pollerdata.co_recv or nil
    local co_send = bit.band(events, poller.EV_POLLER_SEND) ~= 0 and pollerdata.co_send or nil

    -- return the events result for the waiting coroutines
    if co_recv and co_recv == co_send then
        pollerdata.co_recv = nil
        pollerdata.co_send = nil
        return self:_poller_resume_co(co_recv, events)
    else 
    
        if co_recv then
            pollerdata.co_recv = nil<commit_msg>fix scheduler<commit_after>--!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-2020, 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 utils     = require("base/utils")
local option    = require("base/option")
local string    = require("base/string")
local poller    = require("base/poller")
local timer     = require("base/timer")
local hashset   = require("base/hashset")
local coroutine = require("base/coroutine")
local bit       = require("bit")

-- 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 waiting poller object
function _coroutine:waitobj()
    return self._WAITOBJ
end

-- set the waiting poller object
function _coroutine:waitobj_set(obj)
    self._WAITOBJ = obj
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_suspended()
    return self:status() == "suspended"
end

-- get the current timer task
function _coroutine:_timer_task()
    return self._TIMER_TASK
end

-- set the timer task
function _coroutine:_timer_task_set(task)
    self._TIMER_TASK = task
end

-- tostring(coroutine)
function _coroutine:__tostring()
    return string.format("<co: %s/%s>", self:name(), self:status())
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

-- get poller object data for socket, pipe or process object 
function scheduler:_poller_data(obj)
    return self._POLLERDATA and self._POLLERDATA[obj] or nil
end

-- set poller object data
--
-- data.co_recv:            the suspended coroutine for waiting poller/recv
-- data.co_send:            the suspended coroutine for waiting poller/send
-- data.poller_events_wait: the waited events for poller
-- data.poller_events_save: the saved events for poller (triggered)
--
function scheduler:_poller_data_set(obj, data)
    local pollerdata = self._POLLERDATA 
    if not pollerdata then
        pollerdata = {}
        self._POLLERDATA = pollerdata
    end
    pollerdata[obj] = data
end

-- resume the suspended coroutine after poller callback
function scheduler:_poller_resume_co(co, events)

    -- cancel timer task if exists
    local timer_task = co:_timer_task()
    if timer_task then
        timer_task.cancel = true
    end

    -- the scheduler has been stopped? mark events as error to stop the coroutine
    if not self._STARTED then
        events = poller.EV_POLLER_ERROR
    end

    -- this coroutine must be suspended
    assert(co:is_suspended())

    -- resume this coroutine task
    co:waitobj_set(nil)
    return self:co_resume(co, (bit.band(events, poller.EV_POLLER_ERROR) ~= 0) and -1 or events)
end

-- the poller events callback
function scheduler:_poller_events_cb(obj, events)

    -- get poller object data
    local pollerdata = self:_poller_data(obj)
    if not pollerdata then
        return false, string.format("%s: cannot get poller data!", obj)
    end

    -- is process object?
    if obj:otype() == poller.OT_PROC then

        -- resume coroutine and return the process exit status
        pollerdata.proc_status = events

        -- waiting process? resume this coroutine
        if pollerdata.co_waiting then
            local co_waiting = pollerdata.co_waiting
            pollerdata.co_waiting = nil
            return self:_poller_resume_co(co_waiting, 1)
        else 
            pollerdata.proc_pending = 1
        end
        return true
    end

    -- get poller object events
    local events_prev_wait = pollerdata.poller_events_wait
    local events_prev_save = pollerdata.poller_events_save

    -- eof for edge trigger?
    if bit.band(events, poller.EV_POLLER_EOF) ~= 0 then
        -- cache this eof as next recv/send event
        events = bit.band(events, bit.bnot(poller.EV_POLLER_EOF))
        events_prev_save = bit.bor(events_prev_save, events_prev_wait)
        pollerdata.poller_events_save = events_prev_save
    end

    -- get the waiting coroutines
    local co_recv = bit.band(events, poller.EV_POLLER_RECV) ~= 0 and pollerdata.co_recv or nil
    local co_send = bit.band(events, poller.EV_POLLER_SEND) ~= 0 and pollerdata.co_send or nil

    -- return the events result for the waiting coroutines
    if co_recv and co_recv == co_send then
        pollerdata.co_recv = nil
        pollerdata.co_send = nil
        return self:_poller_resume_co(co_recv, events)
    else 
    
        if co_recv then
            pollerdata.co_recv = nil
{'fraction_non_alphanumeric': 0.08642590286425902, 'fraction_numerical': 0.00315483603154836, 'mean_word_length': 3.047715053763441, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2336902', 'n_tokens_mistral': 3688, 'n_tokens_neox': 3713, 'n_words': 1169}