rewrite scheduler

4 files changed, 241 insertions, 152 deletions

diff --git a/xmake/core/base/poller.lua b/xmake/core/base/poller.lua
index 22275e2ff..a5cb02df2 100644
--- a/xmake/core/base/poller.lua
+++ b/xmake/core/base/poller.lua
@@ -27,32 +27,32 @@ local string = require("base/string")
 
 -- the poller object type
 poller.OT_SOCK = 1
-poller.OT_PROC = 2
-poller.OT_PIPE = 3
+poller.OT_PIPE = 2
+poller.OT_PROC = 3
 
 -- the poller events, @see tbox/platform/poller.h
-poller.EV_SOCK_RECV    = 1
-poller.EV_SOCK_SEND    = 2
-poller.EV_SOCK_CONN    = poller.EV_SOCK_SEND
-poller.EV_SOCK_ACPT    = poller.EV_SOCK_RECV
-poller.EV_SOCK_CLEAR   = 0x0010 -- edge trigger. after the event is retrieved by the user, its state is reset
-poller.EV_SOCK_ONESHOT = 0x0010 -- causes the event to return only the first occurrence of the filter being triggered
-poller.EV_SOCK_EOF     = 0x0100 -- the event flag will be marked if the connection be closed in the edge trigger
-poller.EV_SOCK_ERROR   = 0x0200 -- socket error after waiting
+poller.EV_POLLER_RECV    = 1
+poller.EV_POLLER_SEND    = 2
+poller.EV_POLLER_CONN    = poller.EV_POLLER_SEND
+poller.EV_POLLER_ACPT    = poller.EV_POLLER_RECV
+poller.EV_POLLER_CLEAR   = 0x0010 -- edge trigger. after the event is retrieved by the user, its state is reset
+poller.EV_POLLER_ONESHOT = 0x0010 -- causes the event to return only the first occurrence of the filter being triggered
+poller.EV_POLLER_EOF     = 0x0100 -- the event flag will be marked if the connection be closed in the edge trigger
+poller.EV_POLLER_ERROR   = 0x0200 -- socket error after waiting
 
 -- get socket data 
-function poller:_sockdata(csock)
-    return self._SOCKDATA and self._SOCKDATA[csock] or nil
+function poller:_sockdata(cdata)
+    return self._SOCKDATA and self._SOCKDATA[cdata] or nil
 end
 
 -- set socket data
-function poller:_sockdata_set(csock, data)
+function poller:_sockdata_set(cdata, data)
     local sockdata = self._SOCKDATA 
     if not sockdata then
         sockdata = {}
         self._SOCKDATA = sockdata
     end
-    sockdata[csock] = data
+    sockdata[cdata] = data
 end
 
 -- insert socket events to poller
@@ -65,12 +65,12 @@ function poller:_insert_sock(sock, events, udata)
     end
 
     -- insert it
-    if not io.poller_insert(sock:csock(), events) then
+    if not io.poller_insert(sock:cdata(), events) then
         return false, string.format("%s: insert events(%d) to poller failed!", sock, events)
     end
 
     -- save socket data and save sock/ref for gc
-    self:_sockdata_set(sock:csock(), {sock, udata})
+    self:_sockdata_set(sock:cdata(), {sock, udata})
     return true
 end
 
@@ -84,12 +84,12 @@ function poller:_modify_sock(sock, events, udata)
     end
 
     -- modify it
-    if not io.poller_modify(sock:csock(), events) then
+    if not io.poller_modify(sock:cdata(), events) then
         return false, string.format("%s: modify events(%d) to poller failed!", sock, events)
     end
 
     -- update socket data for this socket
-    self:_sockdata_set(sock:csock(), {sock, udata})
+    self:_sockdata_set(sock:cdata(), {sock, udata})
     return true
 end
 
@@ -103,7 +103,7 @@ function poller:_remove_sock(sock)
     end
 
     -- remove it
-    if not io.poller_remove(sock:csock()) then
+    if not io.poller_remove(sock:cdata()) then
         return false, string.format("%s: remove events from poller failed!", sock)
     end
 
@@ -168,11 +168,11 @@ function poller:wait(timeout)
     if events then
         for _, v in ipairs(events) do
             -- TODO only socket events now. It will be proc/pipe events in the future
-            local csock      = v[1]
+            local cdata      = v[1]
             local sockevents = v[2]
-            local sockdata   = self:_sockdata(csock)
+            local sockdata   = self:_sockdata(cdata)
             if not sockdata then
-                return -1, string.format("no socket data for csock(%d)!", csock)
+                return -1, string.format("no socket data for cdata(%d)!", cdata)
             end
             table.insert(results, {poller.OT_SOCK,  sockdata[1], sockevents, sockdata[2]})
         end
diff --git a/xmake/core/base/process.lua b/xmake/core/base/process.lua
index 68d02466b..0d4cdbf26 100644
--- a/xmake/core/base/process.lua
+++ b/xmake/core/base/process.lua
@@ -54,10 +54,15 @@ function _subprocess:name()
 end
 
 -- get cdata of process
-function _subprocess:cproc()
+function _subprocess:cdata()
     return self._PROC
 end
 
+-- get poller object type, poller.OT_PROC
+function _subprocess:otype()
+    return 3
+end
+
 -- wait subprocess
 --
 -- @param timeout   the timeout
diff --git a/xmake/core/base/scheduler.lua b/xmake/core/base/scheduler.lua
index 7ebf0178d..1e5c03f5d 100644
--- a/xmake/core/base/scheduler.lua
+++ b/xmake/core/base/scheduler.lua
@@ -28,6 +28,7 @@ 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")
 
@@ -85,7 +86,7 @@ function _coroutine:_timer_task_set(task)
     self._TIMER_TASK = task
 end
 
--- tostring(socket)
+-- tostring(coroutine)
 function _coroutine:__tostring()
     return string.format("<co: %s/%s>", self:thread(), self:name())
 end
@@ -105,89 +106,117 @@ function scheduler:_timer()
     return t
 end
 
--- get socket events 
-function scheduler:_sockevents(csock)
-    return self._SOCKEVENTS and self._SOCKEVENTS[csock] or 0
+-- 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 socket events
-function scheduler:_sockevents_set(csock, data)
-    local sockevents = self._SOCKEVENTS 
-    if not sockevents then
-        sockevents = {}
-        self._SOCKEVENTS = sockevents
+-- 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
-    sockevents[csock] = data
+
+    -- the scheduler has been stopped? mark events as error to stop the coroutine
+    if not self._STARTED then
+        events = poller.EV_POLLER_ERROR
+    end
+
+    -- resume this coroutine task
+    self:_co_tasks_suspended():remove(co)
+    self:co_resume(co, (bit.band(events, poller.EV_POLLER_ERROR) ~= 0) and -1 or events)
 end
 
--- the socket events callback
-function scheduler:_sockevents_cb(sock, sockevents)
+-- the poller events callback
+function scheduler:_poller_events_cb(obj, events)
+
+    -- get poller object data
+    local pollerdata = self:_poller_data(obj)
+    assert(pollerdata, string.format("%s: cannot get poller data!", obj))
+
+    -- get poller object events
+    local events_prev_wait = pollerdata.poller_events_wait
+    local events_prev_save = pollerdata.poller_events_save
 
-    -- get the previous socket events
-    local events_prev = self:_sockevents(sock:csock())
-    local events_prev_wait = bit.band(events_prev, 0xffff)
-    local events_prev_save = bit.rshift(events_prev, 16)
+    -- 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
 
-    -- is waiting?
-    local running = self:_co_sock_suspended(sock)
-    if running and running:is_suspended() then
+    -- 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
+        self:_poller_resume_co(co_recv, events)
+    else 
     
-        -- eof for edge trigger?
-        if bit.band(sockevents, poller.EV_SOCK_EOF) ~= 0 then
-            -- cache this eof as next recv/send event
-            sockevents  = bit.band(sockevents, bit.bnot(poller.EV_SOCK_EOF))
-            events_prev_save = bit.bor(events_prev_save, events_prev_wait)
-            self:_sockevents_set(sock:csock(), bit.bor(bit.lshift(events_prev_save, 16), events_prev_wait))
+        if co_recv then
+            pollerdata.co_recv = nil
+            self:_poller_resume_co(co_recv, bit.band(events, bit.bnot(poller.EV_POLLER_SEND)))
+            events = bit.band(events, bit.bnot(poller.EV_POLLER_RECV))
         end
-
-        -- cancel timer task if exists
-        local timer_task = running:_timer_task()
-        if timer_task then
-            timer_task.cancel = true
+        if co_send then
+            pollerdata.co_send = nil
+            self:_poller_resume_co(co_send, bit.band(events, bit.bnot(poller.EV_POLLER_RECV)))
+            events = bit.band(events, bit.bnot(poller.EV_POLLER_SEND))
         end
 
-        -- the scheduler has been stopped? mark events as error to stop the coroutine
-        if not self._STARTED then
-            sockevents = poller.EV_SOCK_ERROR
-        end
+        -- no coroutines are waiting? cache this events
+        if bit.band(events, poller.EV_POLLER_RECV) ~= 0 or bit.band(events, poller.EV_POLLER_SEND) ~= 0 then
 
-        -- resume this coroutine task
-        self:_co_sock_suspended_set(sock, nil)
-        self:co_resume(running, (bit.band(sockevents, poller.EV_SOCK_ERROR) ~= 0) and -1 or sockevents)
-    else
-        -- cache socket events
-        events_prev_save = events
-        self:_sockevents_set(sock:csock(), bit.bor(bit.lshift(events_prev_save, 16), events_prev_wait))
+            -- cache this events
+            events_prev_save = bit.bor(events_prev_save, events)
+            pollerdata.poller_events_save = events_prev_save
+        end
     end
 end
 
--- get the suspended coroutine task
-function scheduler:_co_sock_suspended(sock)
-    return self._CO_SOCK_SUSPENDED_TASKS and self._CO_SOCK_SUSPENDED_TASKS[sock] or nil
-end
-
--- set the suspended coroutine task
-function scheduler:_co_sock_suspended_set(sock, co)
-    local co_sock_suspended_tasks = self._CO_SOCK_SUSPENDED_TASKS 
-    if not co_sock_suspended_tasks then
-        co_sock_suspended_tasks = {}
-        self._CO_SOCK_SUSPENDED_TASKS = co_sock_suspended_tasks
+-- get all suspended coroutine tasks
+function scheduler:_co_tasks_suspended()
+    local co_tasks_suspended = self._CO_TASKS_SUSPENDED 
+    if not co_tasks_suspended then
+        co_tasks_suspended = hashset.new()
+        self._CO_TASKS_SUSPENDED = co_tasks_suspended
     end
-    co_sock_suspended_tasks[sock] = co
+    return co_tasks_suspended
 end
 
--- cancel and resume all suspended socket tasks after stopping scheduler
+-- cancel and resume all suspended tasks after stopping scheduler
 -- we cannot suspend them forever, all tasks will be exited directly and free all resources.
-function scheduler:_co_sock_suspended_cancel_all()
-    local co_sock_suspended_tasks = self._CO_SOCK_SUSPENDED_TASKS
-    if co_sock_suspended_tasks then
-        for _, co in pairs(co_sock_suspended_tasks) do
-            local ok, errors = self:co_resume(co, -1) 
-            if not ok then
-                return false, errors
-            end
+function scheduler:_co_tasks_suspended_cancel_all()
+    for co in self:_co_tasks_suspended():keys() do
+        local ok, errors = self:co_resume(co, -1) 
+        if not ok then
+            return false, errors
         end
     end
+    return true
 end
 
 -- start a new coroutine task
@@ -250,13 +279,13 @@ function scheduler:co_count()
     return self._CO_COUNT or 0
 end
 
--- wait socket events
-function scheduler:sock_wait(sock, events, timeout)
+-- wait poller object io events, only for socket and pipe object
+function scheduler:poller_wait(obj, events, timeout)
 
     -- get the running coroutine
     local running = self:co_running()
     if not running then
-        return -1, "we must call waitsock() in coroutine with scheduler!"
+        return -1, "we must call poller_wait() in coroutine with scheduler!"
     end
 
     -- is stopped?
@@ -264,44 +293,71 @@ function scheduler:sock_wait(sock, events, timeout)
         return -1, "the scheduler is stopped!"
     end
 
-    -- enable edge-trigger mode if be supported
-    if self._SUPPORT_EV_SOCK_CLEAR then
-        events = bit.bor(events, poller.EV_SOCK_CLEAR)
+    -- check the object type
+    local otype = obj:otype()
+    if otype ~= poller.OT_SOCK and otype ~= poller.OT_PIPE then
+        return -1, string.format("%s: invalid object type(%d)!", obj, otype)
     end
 
-    -- get the previous socket events
-    local events_prev = self:_sockevents(sock:csock())
-    if events_prev ~= 0 then
-        local events_prev_wait = bit.band(events_prev, 0xffff)
-        local events_prev_save = bit.rshift(events_prev, 16)
+    -- get and allocate poller object data
+    local pollerdata = self:_poller_data(obj)
+    if not pollerdata then
+        pollerdata = {poller_events_wait = 0, poller_events_save = 0}
+        self:_poller_data_set(obj, pollerdata)
+    end
+
+    -- enable edge-trigger mode if be supported
+    if otype == poller.OT_SOCK and self._SUPPORT_EV_POLLER_CLEAR then
+        events = bit.bor(events, poller.EV_POLLER_CLEAR)
+    end
 
+    -- get the previous poller object events
+    local events_wait = events
+    if pollerdata.poller_events_wait ~= 0 then
+        
         -- return the cached events directly if the waiting events exists cache
+        local events_prev_wait = pollerdata.poller_events_wait
+        local events_prev_save = pollerdata.poller_events_save
         if events_prev_save ~= 0 and bit.band(events_prev_wait, events) ~= 0 then
 
             -- check error?
-            if bit.band(events_prev_save, poller.EV_SOCK_ERROR) ~= 0 then
-                self:_sockevents_set(sock:csock(), events_prev_wait)
-                return -1, string.format("%s: socket events error!", sock)
+            if bit.band(events_prev_save, poller.EV_POLLER_ERROR) ~= 0 then
+                pollerdata.poller_events_wait = events_prev_wait
+                pollerdata.poller_events_save = 0
+                return -1, string.format("%s: events error!", obj)
             end
 
             -- clear cache events
-            self:_sockevents_set(sock:csock(), bit.bor(bit.lshift(bit.band(events_prev_save, bit.bnot(events)), 16), events_prev_wait))
+            pollerdata.poller_events_wait = events_prev_wait
+            pollerdata.poller_events_save = bit.band(events_prev_save, bit.bnot(events))
 
             -- return the cached events
             return bit.band(events_prev_save, events)
         end
 
-        -- modify socket from poller for waiting events if the waiting events has been changed 
-        if events_prev_wait ~= events then
-            -- modify socket events
-            local ok, errors = poller:modify(poller.OT_SOCK, sock, events, self._sockevents_cb)
+        -- modify the wait events and reserve the pending events in other coroutine
+        events_wait = events_prev_wait
+        if bit.band(events_wait, poller.EV_POLLER_RECV) ~= 0 and not pollerdata.co_recv then
+            events_wait = bit.band(events_wait, bit.bnot(poller.EV_POLLER_RECV))
+        end
+        if bit.band(events_wait, poller.EV_POLLER_SEND) ~= 0 and not pollerdata.co_send then
+            events_wait = bit.band(events_wait, bit.bnot(poller.EV_POLLER_SEND))
+        end
+        events_wait = bit.bor(events_wait, events)
+
+        -- modify poller object from poller for waiting events if the waiting events has been changed 
+        if bit.band(events_prev_wait, events_wait) ~= events_wait then
+
+            -- maybe wait recv/send at same time
+            local ok, errors = poller:modify(poller.OT_SOCK, obj, events_wait, self._sockevents_cb)
             if not ok then
                 return -1, errors
             end
         end
     else
-        -- insert socket events
-        local ok, errors = poller:insert(poller.OT_SOCK, sock, events, self._sockevents_cb)
+
+        -- insert poller object events
+        local ok, errors = poller:insert(poller.OT_SOCK, obj, events_wait, self._sockevents_cb)
         if not ok then
             return -1, errors
         end
@@ -312,37 +368,56 @@ function scheduler:sock_wait(sock, events, timeout)
     if timeout > 0 then
         timer_task = self:_timer():post(function (cancel) 
             if not cancel and running:is_suspended() then
-                self:_co_sock_suspended_set(sock, nil)
+                self:_co_tasks_suspended():remove(running)
                 self:co_resume(running, 0)
             end
         end, timeout)
     end
     running:_timer_task_set(timer_task)
 
-    -- save the waiting events 
-    self:_sockevents_set(sock:csock(), events)
+    -- save waiting events to coroutine
+    pollerdata.poller_events_wait = events_wait
+    pollerdata.poller_events_save = 0
+
+    -- save the current coroutine 
+    if bit.band(events, poller.EV_POLLER_RECV) ~= 0 then
+        pollerdata.co_recv = running
+    end
+    if bit.band(events, poller.EV_POLLER_SEND) ~= 0 then
+        pollerdata.co_send = running
+    end
 
     -- save the suspended coroutine
-    self:_co_sock_suspended_set(sock, running)
+    self:_co_tasks_suspended():insert(running)
 
     -- wait
     return self:co_suspend()
 end
 
--- cancel socket events
-function scheduler:sock_cancel(sock)
+-- cancel poller object events
+function scheduler:poller_cancel(obj)
 
-    -- get the previous socket events
-    local events_prev = self:_sockevents(sock:csock())
-    if events_prev ~= 0 then
+    -- reset the pollerdata data
+    local pollerdata = self:_poller_data(obj)
+    if pollerdata then
 
-        -- remove the waiting socket from the poller
-        local ok, errors = poller:remove(poller.OT_SOCK, sock)
-        if not ok then
-            return false, errors
+        -- clear the waiting coroutines
+        pollerdata.co_recv = nil
+        pollerdata.co_send = nil
+
+        -- remove the this poller object from poller
+        if pollerdata.poller_events_wait ~= 0 then
+
+            -- remove the previous poller object first if exists
+            local ok, errors = poller:remove(poller.OT_SOCK, obj)
+            if not ok then
+                return false, errors
+            end
+
+            -- remove the poller object events
+            pollerdata.poller_events_wait = 0
+            pollerdata.poller_events_save = 0
         end
-        self:_sockevents_set(sock:csock(), 0)
-        self:_co_sock_suspended_set(sock, nil)
     end
     return true
 end
@@ -393,8 +468,8 @@ function scheduler:runloop()
     self._STARTED = true
 
     -- ensure poller has been initialized first (for windows/iocp) and check edge-trigger mode (for epoll/kqueue)
-    if poller:support(poller.OT_SOCK, poller.EV_SOCK_CLEAR) then
-        self._SUPPORT_EV_SOCK_CLEAR = true
+    if poller:support(poller.OT_SOCK, poller.EV_POLLER_CLEAR) then
+        self._SUPPORT_EV_POLLER_CLEAR = true
     end
 
     -- start all ready coroutine tasks
@@ -454,7 +529,11 @@ function scheduler:runloop()
     self._STARTED = false
 
     -- cancel all suspended tasks after stopping scheduler
-    self:_co_sock_suspended_cancel_all()
+    local ok2, errors2 = self:_co_tasks_suspended_cancel_all()
+    if ok and not ok2 then
+        ok = ok2
+        errors = errors2
+    end
 
     -- cancel all timeout tasks and trigger them
     self:_timer():kill()
diff --git a/xmake/core/base/socket.lua b/xmake/core/base/socket.lua
index 1fe895c20..7669ac455 100644
--- a/xmake/core/base/socket.lua
+++ b/xmake/core/base/socket.lua
@@ -66,10 +66,15 @@ function _instance:family()
 end
 
 -- get cdata of socket
-function _instance:csock()
+function _instance:cdata()
     return self._SOCK
 end
 
+-- get poller object type, poller.OT_SOCK
+function _instance:otype()
+    return 1
+end
+
 -- get socket rawfd
 function _instance:rawfd()
 
@@ -80,7 +85,7 @@ function _instance:rawfd()
     end
 
     -- get rawfd
-    local result, errors = io.socket_rawfd(self:csock())
+    local result, errors = io.socket_rawfd(self:cdata())
     if not result and errors then
         errors = string.format("%s: %s", self, errors)
     end
@@ -97,7 +102,7 @@ function _instance:bind(addr, port)
     end
 
     -- bind it
-    local ok, errors = io.socket_bind(self:csock(), addr, port, self:family())
+    local ok, errors = io.socket_bind(self:cdata(), addr, port, self:family())
     if not ok and errors then
         errors = string.format("%s: %s", self, errors)
     end
@@ -120,7 +125,7 @@ function _instance:bind_unix(addr, opt)
  
     -- bind it
     opt = opt or {}
-    local ok, errors = io.socket_bind(self:csock(), addr, opt.is_abstract, self:family())
+    local ok, errors = io.socket_bind(self:cdata(), addr, opt.is_abstract, self:family())
     if not ok and errors then
         errors = string.format("%s: %s", self, errors)
     end
@@ -137,7 +142,7 @@ function _instance:listen(backlog)
     end
 
     -- listen it
-    local ok, errors = io.socket_listen(self:csock(), backlog or 10)
+    local ok, errors = io.socket_listen(self:cdata(), backlog or 10)
     if not ok and errors then
         errors = string.format("%s: %s", self, errors)
     end
@@ -154,12 +159,12 @@ function _instance:accept(opt)
     end
 
     -- accept it
-    local sock, errors = io.socket_accept(self:csock())
+    local sock, errors = io.socket_accept(self:cdata())
     if not sock and not errors then
         opt = opt or {}
         local events, waiterrs = self:wait(socket.EV_ACPT, opt.timeout or -1)
         if events == socket.EV_ACPT then
-            sock, errors = io.socket_accept(self:csock())
+            sock, errors = io.socket_accept(self:cdata())
         else
             errors = waiterrs
         end
@@ -183,12 +188,12 @@ function _instance:connect(addr, port, opt)
     end
 
     -- connect it
-    local ok, errors = io.socket_connect(self:csock(), addr, port, self:family())
+    local ok, errors = io.socket_connect(self:cdata(), addr, port, self:family())
     if ok == 0 then
         opt = opt or {}
         local events, waiterrs = self:wait(socket.EV_CONN, opt.timeout or -1)
         if events == socket.EV_CONN then
-            ok, errors = io.socket_connect(self:csock(), addr, port, self:family())
+            ok, errors = io.socket_connect(self:cdata(), addr, port, self:family())
         else
             errors = waiterrs
         end
@@ -215,11 +220,11 @@ function _instance:connect_unix(addr, opt)
 
     -- connect it
     opt = opt or {}
-    local ok, errors = io.socket_connect(self:csock(), addr, opt.is_abstract, self:family())
+    local ok, errors = io.socket_connect(self:cdata(), addr, opt.is_abstract, self:family())
     if ok == 0 then
         local events, waiterrs = self:wait(socket.EV_CONN, opt.timeout or -1)
         if events == socket.EV_CONN then
-            ok, errors = io.socket_connect(self:csock(), addr, opt.is_abstract, self:family())
+            ok, errors = io.socket_connect(self:cdata(), addr, opt.is_abstract, self:family())
         else
             errors = waiterrs
         end
@@ -264,7 +269,7 @@ function _instance:send(data, opt)
     if opt.block then
         local size = last + 1 - start
         while start <= last do
-            real, errors = io.socket_send(self:csock(), data, start, last)
+            real, errors = io.socket_send(self:cdata(), data, start, last)
             if real > 0 then
                 send = send + real
                 start = start + real
@@ -285,7 +290,7 @@ function _instance:send(data, opt)
             send = -1
         end
     else
-        send, errors = io.socket_send(self:csock(), data, start, last)
+        send, errors = io.socket_send(self:cdata(), data, start, last)
         if send < 0 and errors then
             errors = string.format("%s: %s", self, errors)
         end
@@ -326,7 +331,7 @@ function _instance:sendfile(file, opt)
     if opt.block then
         local size = last + 1 - start
         while start <= last do
-            real, errors = io.socket_sendfile(self:csock(), file._FILE, start, last)
+            real, errors = io.socket_sendfile(self:cdata(), file._FILE, start, last)
             if real > 0 then
                 send = send + real
                 start = start + real
@@ -347,7 +352,7 @@ function _instance:sendfile(file, opt)
             send = -1
         end
     else
-        send, errors = io.socket_sendfile(self:csock(), file._FILE, start, last)
+        send, errors = io.socket_sendfile(self:cdata(), file._FILE, start, last)
         if send < 0 and errors then
             errors = string.format("%s: %s", self, errors)
         end
@@ -381,7 +386,7 @@ function _instance:recv(size, opt)
         local results = {}
         while recv < size do
             local buff = self:_recvbuff()
-            real, data_or_errors = io.socket_recv(self:csock(), buff:caddr(), math.min(buff:size(), size - recv))
+            real, data_or_errors = io.socket_recv(self:cdata(), buff:caddr(), math.min(buff:size(), size - recv))
             if real > 0 then
                 recv = recv + real
                 wait = false
@@ -406,7 +411,7 @@ function _instance:recv(size, opt)
         end
     else
         local buff = self:_recvbuff()
-        recv, data_or_errors = io.socket_recv(self:csock(), buff:caddr(), math.min(buff:size(), size))
+        recv, data_or_errors = io.socket_recv(self:cdata(), buff:caddr(), math.min(buff:size(), size))
         if recv > 0 then
             data_or_errors = bytes(buff, 1, recv)
             self:_recvbuff_clear()
@@ -449,7 +454,7 @@ function _instance:sendto(data, addr, port, opt)
     local errors = nil
     if opt.block then
         while true do
-            send, errors = io.socket_sendto(self:csock(), data, addr, port, self:family())
+            send, errors = io.socket_sendto(self:cdata(), data, addr, port, self:family())
             if send == 0 and not wait then
                 local events, waiterrs = self:wait(socket.EV_SEND, opt.timeout or -1)
                 if events == socket.EV_SEND then
@@ -463,7 +468,7 @@ function _instance:sendto(data, addr, port, opt)
             end
         end
     else
-        send, errors = io.socket_sendto(self:csock(), data, addr, port, self:family())
+        send, errors = io.socket_sendto(self:cdata(), data, addr, port, self:family())
         if send < 0 and errors then
             errors = string.format("%s: %s", self, errors)
         end
@@ -500,7 +505,7 @@ function _instance:recvfrom(size, opt)
     if opt.block then
         while true do
             local buff = self:_recvbuff()
-            recv, data_or_errors, addr, port = io.socket_recvfrom(self:csock(), buff:caddr(), math.min(buff:size(), size))
+            recv, data_or_errors, addr, port = io.socket_recvfrom(self:cdata(), buff:caddr(), math.min(buff:size(), size))
             if recv > 0 then
                 data_or_errors = bytes(buff, 1, recv)
                 self:_recvbuff_clear()
@@ -520,7 +525,7 @@ function _instance:recvfrom(size, opt)
         end
     else
         local buff = self:_recvbuff()
-        recv, data_or_errors, addr, port = io.socket_recvfrom(self:csock(), buff:caddr(), math.min(buff:size(), size))
+        recv, data_or_errors, addr, port = io.socket_recvfrom(self:cdata(), buff:caddr(), math.min(buff:size(), size))
         if recv > 0 then
             data_or_errors = bytes(buff, 1, recv)
             self:_recvbuff_clear()
@@ -545,9 +550,9 @@ function _instance:wait(events, timeout)
     local result = -1
     local errors = nil
     if scheduler:co_running() then
-        result, errors = scheduler:sock_wait(self, events, timeout or -1)
+        result, errors = scheduler:poller_wait(self, events, timeout or -1)
     else
-        result, errors = io.socket_wait(self:csock(), events, timeout or -1)
+        result, errors = io.socket_wait(self:cdata(), events, timeout or -1)
     end
     if result < 0 and errors then
         errors = string.format("%s: %s", self, errors)
@@ -566,14 +571,14 @@ function _instance:close()
 
     -- cancel socket events from the scheduler
     if scheduler:co_running() then
-        ok, errors = scheduler:sock_cancel(self)
+        ok, errors = scheduler:poller_cancel(self)
         if not ok then
             return false, errors
         end
     end
 
     -- close it
-    ok = io.socket_close(self:csock())
+    ok = io.socket_close(self:cdata())
     if ok then
         self._SOCK = nil
     end
@@ -597,7 +602,7 @@ end
 
 -- ensure the socket is opened
 function _instance:_ensure_opened()
-    if not self:csock() then
+    if not self:cdata() then
         return false, string.format("%s: has been closed!", self)
     end
     return true
@@ -612,7 +617,7 @@ end
 
 -- gc(socket)
 function _instance:__gc()
-    if self:csock() and io.socket_close(self:csock()) then
+    if self:cdata() and io.socket_close(self:cdata()) then
         self._SOCK = nil
     end
 end