From 4ca9d6d812e027735a6a5cf4fa3099afcb43265a Mon Sep 17 00:00:00 2001
From: Thijs Schreijer <thijs@thijsschreijer.nl>
Date: Tue, 26 Jul 2022 12:32:47 +0200
Subject: [PATCH] feat(naming) adds naming to sockets and threads for debugging

also adds some debug functions to have a detailed view of
yielding and resuming of threads in the scheduler.
---
 docs/index.html     |   2 +-
 src/copas.lua       | 249 +++++++++++++++++++++++++++++++++++++-------
 src/copas/queue.lua |  23 +++-
 src/copas/timer.lua |  12 ++-
 4 files changed, 241 insertions(+), 45 deletions(-)

diff --git a/docs/index.html b/docs/index.html
index 13d14f9..4ab01f1 100644
--- a/docs/index.html
+++ b/docs/index.html
@@ -109,7 +109,7 @@ <h2><a name="history"></a>History</h2>
 
     <dt><strong>Copas 3.0.0</strong> [12/Nov/2021]</dt>
 	<dd><ul>
-        <li>[breaking] Change: copas.addserver() now uses the timeout value as a copas timeout,
+        <li>[breaking] Change: copas.addserver() now uses the timeout value as a copas timeout,
             instead of a luasocket timeout. The client sockets for incoming connections will
             inherit the timeout from the server socket.</li>
         <li>Added: support for SNI on TLS connections #81 (@amyspark)</li>
diff --git a/src/copas.lua b/src/copas.lua
index 20df61b..5df24b0 100644
--- a/src/copas.lua
+++ b/src/copas.lua
@@ -61,6 +61,19 @@ function socket.newtry(finalizer)
          end
 end
 
+
+-- nil-safe versions for pack/unpack
+local _unpack = unpack or table.unpack
+local unpack = function(t, i, j) return _unpack(t, i or 1, j or t.n or #t) end
+local pack = function(...) return { n = select("#", ...), ...} end
+
+
+local coroutine_create = coroutine.create
+local coroutine_running = coroutine.running
+local coroutine_yield = coroutine.yield
+local coroutine_resume = coroutine.resume
+
+
 local copas = {}
 
 -- Meta information is public even if beginning with an "_"
@@ -75,6 +88,18 @@ copas.autoclose = true
 copas.running = false
 
 
+-------------------------------------------------------------------------------
+-- Object names, to track names of thread/coroutines and sockets
+-------------------------------------------------------------------------------
+local object_names = setmetatable({}, {
+  __mode = "k",
+  __index = function(self, key)
+    local name = tostring(key)
+    rawset(self, key, name)
+    return name
+  end
+})
+
 -------------------------------------------------------------------------------
 -- Simple set implementation
 -- adds a FIFO queue for each socket in the set
@@ -316,7 +341,7 @@ local sto_timeout, sto_timed_out, sto_change_queue, sto_error do
   -- is truthy, it is the connect timeout.
   -- @return true
   function sto_timeout(skt, queue, use_connect_to)
-    local co = coroutine.running()
+    local co = coroutine_running()
     socket_register[co] = skt
     operation_register[co] = queue
     timeout_flags[co] = nil
@@ -338,20 +363,20 @@ local sto_timeout, sto_timed_out, sto_change_queue, sto_error do
   -- @param queue (string) the new queue the socket is in, must be either "read" or "write"
   -- @return true
   function sto_change_queue(queue)
-    operation_register[coroutine.running()] = queue
+    operation_register[coroutine_running()] = queue
     return true
   end
 
 
   -- Responds with `true` if the operation timed-out.
   function sto_timed_out()
-    return timeout_flags[coroutine.running()]
+    return timeout_flags[coroutine_running()]
   end
 
 
   -- Returns the poroper timeout error
   function sto_error(err)
-    return useSocketTimeoutErrors[coroutine.running()] and err or "timeout"
+    return useSocketTimeoutErrors[coroutine_running()] and err or "timeout"
   end
 end
 
@@ -462,12 +487,12 @@ function copas.receive(client, pattern, part)
       current_log = _writing_log
       current_log[client] = gettime()
       sto_change_queue("write")
-      coroutine.yield(client, _writing)
+      coroutine_yield(client, _writing)
     else
       current_log = _reading_log
       current_log[client] = gettime()
       sto_change_queue("read")
-      coroutine.yield(client, _reading)
+      coroutine_yield(client, _reading)
     end
   until false
 end
@@ -498,7 +523,7 @@ function copas.receivefrom(client, size)
     end
 
     _reading_log[client] = gettime()
-    coroutine.yield(client, _reading)
+    coroutine_yield(client, _reading)
   until false
 end
 
@@ -532,12 +557,12 @@ function copas.receivePartial(client, pattern, part)
       current_log = _writing_log
       current_log[client] = gettime()
       sto_change_queue("write")
-      coroutine.yield(client, _writing)
+      coroutine_yield(client, _writing)
     else
       current_log = _reading_log
       current_log[client] = gettime()
       sto_change_queue("read")
-      coroutine.yield(client, _reading)
+      coroutine_yield(client, _reading)
     end
   until false
 end
@@ -560,9 +585,9 @@ function copas.send(client, data, from, to)
     if (math.random(100) > 90) then
       current_log[client] = gettime()   -- TODO: how to handle this??
       if current_log == _writing_log then
-        coroutine.yield(client, _writing)
+        coroutine_yield(client, _writing)
       else
-        coroutine.yield(client, _reading)
+        coroutine_yield(client, _reading)
       end
     end
 
@@ -585,12 +610,12 @@ function copas.send(client, data, from, to)
       current_log = _reading_log
       current_log[client] = gettime()
       sto_change_queue("read")
-      coroutine.yield(client, _reading)
+      coroutine_yield(client, _reading)
     else
       current_log = _writing_log
       current_log[client] = gettime()
       sto_change_queue("write")
-      coroutine.yield(client, _writing)
+      coroutine_yield(client, _writing)
     end
   until false
 end
@@ -631,7 +656,7 @@ function copas.connect(skt, host, port)
 
     tried_more_than_once = tried_more_than_once or true
     _writing_log[skt] = gettime()
-    coroutine.yield(skt, _writing)
+    coroutine_yield(skt, _writing)
   until false
 end
 
@@ -752,7 +777,7 @@ function copas.dohandshake(skt, wrap_params)
       error("TLS/SSL handshake failed: " .. tostring(err))
     end
 
-    coroutine.yield(nskt, queue)
+    coroutine_yield(nskt, queue)
   until false
 end
 
@@ -932,7 +957,7 @@ function copas.setErrorHandler (err, default)
   if default then
     _deferror = err
   else
-    _errhandlers[coroutine.running()] = err
+    _errhandlers[coroutine_running()] = err
   end
 end
 
@@ -940,7 +965,7 @@ end
 -- `timeout, wantread, wantwrite, Operation already in progress`. If falsy, it will always
 -- return `timeout`.
 function copas.useSocketTimeoutErrors(bool)
-  useSocketTimeoutErrors[coroutine.running()] = not not bool -- force to a boolean
+  useSocketTimeoutErrors[coroutine_running()] = not not bool -- force to a boolean
 end
 
 -------------------------------------------------------------------------------
@@ -957,7 +982,7 @@ local function _doTick (co, skt, ...)
     return
   end
 
-  local ok, res, new_q = coroutine.resume(co, skt, ...)
+  local ok, res, new_q = coroutine_resume(co, skt, ...)
 
   if ok and res and new_q then
     new_q:insert (res)
@@ -972,19 +997,26 @@ local function _doTick (co, skt, ...)
 end
 
 
--- accepts a connection on socket input
-local function _accept(server_skt, handler)
-  local client_skt = server_skt:accept()
-  if client_skt then
-    client_skt:settimeout(0)
-    copas.settimeouts(client_skt, user_timeouts_connect[server_skt],  -- copy server socket timeout settings
-      user_timeouts_send[server_skt], user_timeouts_receive[server_skt])
-    local co = coroutine.create(handler)
-    _doTick(co, client_skt)
+local _accept do
+  local client_counters = setmetatable({}, { __mode = "k" })
+
+  -- accepts a connection on socket input
+  function _accept(server_skt, handler)
+    local client_skt = server_skt:accept()
+    if client_skt then
+      local count = (client_counters[server_skt] or 0) + 1
+      client_counters[server_skt] = count
+      object_names[client_skt] = object_names[server_skt] .. ":client_" .. count
+
+      client_skt:settimeout(0)
+      copas.settimeouts(client_skt, user_timeouts_connect[server_skt],  -- copy server socket timeout settings
+        user_timeouts_send[server_skt], user_timeouts_receive[server_skt])
+      local co = coroutine_create(handler)
+      _doTick(co, client_skt)
+    end
   end
 end
 
-
 -------------------------------------------------------------------------------
 -- Adds a server/handler pair to Copas dispatcher
 -------------------------------------------------------------------------------
@@ -1001,7 +1033,7 @@ do
 
   local function addUDPserver(server, handler, timeout)
     server:settimeout(0)
-    local co = coroutine.create(handler)
+    local co = coroutine_create(handler)
     _reading:insert(server)
     if timeout then
       copas.settimeout(server, timeout)
@@ -1010,11 +1042,14 @@ do
   end
 
 
-  function copas.addserver(server, handler, timeout)
+  function copas.addserver(server, handler, timeout, name)
+    if name then
+      object_names[server] = name
+    end
     if isTCP(server) then
-      addTCPserver(server, handler, timeout)
+      addTCPserver(server, handler, timeout, name)
     else
-      addUDPserver(server, handler, timeout)
+      addUDPserver(server, handler, timeout, name)
     end
   end
 end
@@ -1041,21 +1076,31 @@ end
 -------------------------------------------------------------------------------
 -- Adds an new coroutine thread to Copas dispatcher
 -------------------------------------------------------------------------------
-function copas.addthread(handler, ...)
+function copas.addnamedthread(handler, name, ...)
   -- create a coroutine that skips the first argument, which is always the socket
   -- passed by the scheduler, but `nil` in case of a task/thread
-  local thread = coroutine.create(function(_, ...)
+  local thread = coroutine_create(function(_, ...)
     -- TODO: this should be added to not immediately execute the thread
     -- it should only schedule and then return to the calling code
     -- Enabling this breaks the "limitset".
     -- copas.sleep(0)
     return handler(...)
   end)
+  if name then
+    object_names[thread] = name
+  end
+
   _threads[thread] = true -- register this thread so it can be removed
   _doTick (thread, nil, ...)
   return thread
 end
 
+
+function copas.addthread(handler, ...)
+  return copas.addnamedthread(handler, nil, ...)
+end
+
+
 function copas.removethread(thread)
   -- if the specified coroutine is registered, add it to the canceled table so
   -- that next time it tries to resume it exits.
@@ -1071,7 +1116,7 @@ end
 -- yields the current coroutine and wakes it after 'sleeptime' seconds.
 -- If sleeptime < 0 then it sleeps until explicitly woken up using 'wakeup'
 function copas.sleep(sleeptime)
-  coroutine.yield((sleeptime or 0), _sleeping)
+  coroutine_yield((sleeptime or 0), _sleeping)
 end
 
 -- Wakes up a sleeping coroutine 'co'.
@@ -1093,24 +1138,24 @@ do
       err_handler = function(...) return _deferror(...) end,
     })
 
-  copas.addthread(function()
+  copas.addnamedthread(function()
     while true do
       copas.sleep(TIMEOUT_PRECISION)
       timerwheel:step()
     end
-  end)
+  end, "copas_core_timer")
 
   -- get the number of timeouts running
   function copas.gettimeouts()
     return timerwheel:count()
   end
 
-  --- Sets the timeout for the current coroutine.
+  --- Sets the timeout for the current coroutine_
   -- @param delay delay (seconds), use 0 to cancel the timerout
   -- @param callback function with signature: `function(coroutine)` where coroutine is the routine that timed-out
   -- @return true
   function copas.timeout(delay, callback)
-    local co = coroutine.running()
+    local co = coroutine_running()
     local existing_timer = timeout_register[co]
 
     if existing_timer then
@@ -1339,7 +1384,7 @@ end
 -------------------------------------------------------------------------------
 function copas.loop(initializer, timeout)
   if type(initializer) == "function" then
-    copas.addthread(initializer)
+    copas.addnamedthread(initializer, "copas_initializer")
   else
     timeout = initializer or timeout
   end
@@ -1349,4 +1394,128 @@ function copas.loop(initializer, timeout)
   copas.running = false
 end
 
+
+-------------------------------------------------------------------------------
+-- Naming sockets and coroutines.
+-------------------------------------------------------------------------------
+do
+  local function realsocket(skt)
+    local mt = getmetatable(skt)
+    if mt == _skt_mt_tcp or mt == _skt_mt_udp then
+      return skt.socket
+    else
+      return skt
+    end
+  end
+
+
+  function copas.setsocketname(name, skt)
+    assert(type(name) == "string", "expected arg #1 to be a string")
+    skt = assert(realsocket(skt), "expected arg #2 to be a socket")
+    object_names[skt] = name
+  end
+
+
+  function copas.getsocketname(skt)
+    skt = assert(realsocket(skt), "expected arg #1 to be a socket")
+    return object_names[skt]
+  end
+end
+
+
+function copas.setthreadname(name, coro)
+  assert(type(name) == "string", "expected arg #1 to be a string")
+  coro = coro or coroutine_running()
+  assert(type(coro) == "thread", "expected arg #2 to be a coroutine or nil")
+  object_names[coro] = name
+end
+
+
+function copas.getthreadname(coro)
+  coro = coro or coroutine_running()
+  assert(type(coro) == "thread", "expected arg #1 to be a coroutine or nil")
+  return object_names[coro]
+end
+
+-------------------------------------------------------------------------------
+-- Debug functionality.
+-------------------------------------------------------------------------------
+do
+  copas.debug = {}
+
+  local log_core    -- if truthy, the core-timer will also be logged
+  local debug_log   -- function used as logger
+
+
+  local debug_yield = function(skt, queue)
+    local name = object_names[coroutine.running()]
+
+    if log_core or name ~= "copas_core_timer" then
+      if queue == _sleeping then
+        debug_log("yielding '", name, "' to SLEEP for ", skt," seconds")
+
+      elseif queue == _writing then
+        debug_log("yielding '", name, "' to WRITE to '", object_names[skt], "'")
+
+      elseif queue == _reading then
+        debug_log("yielding '", name, "' to READ from '", object_names[skt], "'")
+
+      else
+        debug_log("thread '", name, "' yielding to unexpected queue; ", tostring(queue), " (", type(queue), ")", debug.traceback())
+      end
+    end
+
+    return coroutine.yield(skt, queue)
+  end
+
+
+  local debug_resume = function(coro, skt, ...)
+    local name = object_names[coro]
+
+    if skt then
+      debug_log("resuming '", name, "' for socket '", object_names[skt], "'")
+    else
+      if log_core or name ~= "copas_core_timer" then
+        debug_log("resuming '", name, "'")
+      end
+    end
+    return coroutine.resume(coro, skt, ...)
+  end
+
+
+  local debug_create = function(f)
+    local f_wrapped = function(...)
+      local results = pack(f(...))
+      debug_log("exiting '", object_names[coroutine.running()], "'")
+      return unpack(results)
+    end
+
+    return coroutine.create(f_wrapped)
+  end
+
+
+  debug_log = function() end
+
+
+  function copas.debug.start(logger, core)
+    log_core = core
+    debug_log = logger or function(...)
+      print(table.concat {...} )
+    end
+    coroutine_yield = debug_yield
+    coroutine_resume = debug_resume
+    coroutine_create = debug_create
+  end
+
+
+  function copas.debug.stop()
+    debug_log = function() end
+    coroutine_yield = coroutine.yield
+    coroutine_resume = coroutine.resume
+    coroutine_create = coroutine.create
+  end
+
+end
+
+
 return copas
diff --git a/src/copas/queue.lua b/src/copas/queue.lua
index 53273a2..dbc4d7f 100644
--- a/src/copas/queue.lua
+++ b/src/copas/queue.lua
@@ -7,16 +7,29 @@ local Queue = {}
 Queue.__index = Queue
 
 
+local new_name do
+  local count = 0
+
+  function new_name()
+    count = count + 1
+    return "copas_queue_" .. count
+  end
+end
+
+
 -- Creates a new Queue instance
-function Queue.new()
+function Queue.new(opts)
+  opts = opts or {}
   local self = {}
   setmetatable(self, Queue)
+  self.name = opts.name or new_name()
   self.sema = Sema.new(10^9)
   self.head = 1
   self.tail = 1
   self.list = {}
   self.workers = setmetatable({}, { __mode = "k" })
   self.stopping = false
+  self.worker_id = 0
   return self
 end
 
@@ -128,7 +141,11 @@ end
 function Queue:add_worker(worker)
   assert(type(worker) == "function", "expected worker to be a function")
   local coro
-  coro = copas.addthread(function()
+
+  self.worker_id = self.worker_id + 1
+  local worker_name = self.name .. ":worker_" .. self.worker_id
+
+  coro = copas.addnamedthread(function()
     copas.sleep(0) -- TODO: remove after adding into copas.addthread
     while true do
       local item = self:pop(10*365*24*60*60) -- wait forever (10yr)
@@ -138,7 +155,7 @@ function Queue:add_worker(worker)
       worker(item)
     end
     self.workers[coro] = nil
-  end)
+  end, worker_name)
 
   self.workers[coro] = true
   return coro
diff --git a/src/copas/timer.lua b/src/copas/timer.lua
index 1ed7948..3c26fbf 100644
--- a/src/copas/timer.lua
+++ b/src/copas/timer.lua
@@ -6,6 +6,15 @@ local timer = {}
 timer.__index = timer
 
 
+local new_name do
+  local count = 0
+
+  function new_name()
+    count = count + 1
+    return "copas_timer_" .. count
+  end
+end
+
 
 do
   local function expire_func(self, initial_delay)
@@ -37,7 +46,7 @@ do
     end
 
     self.cancelled = false
-    self.co = copas.addthread(expire_func, self, initial_delay or self.delay)
+    self.co = copas.addnamedthread(expire_func, self.name, self, initial_delay or self.delay)
     return self
   end
 end
@@ -73,6 +82,7 @@ function timer.new(opts)
   assert(opts.delay >= 0, "delay must be greater than or equal to 0")
   assert(type(opts.callback) == "function", "expected callback to be a function")
   return setmetatable({
+    name = opts.name or new_name(),
     delay = opts.delay,
     callback = opts.callback,
     recurring = not not opts.recurring,