From 2fec5b39b8a7a51970960b033a81e60481db01be Mon Sep 17 00:00:00 2001 From: Ryan Ward Date: Mon, 23 Mar 2026 20:40:58 -0700 Subject: [PATCH] fixing bugs --- docs/changes.md | 9 + init.lua | 481 +++++++++++++++--------- integration/lanesManager/extensions.lua | 16 +- 3 files changed, 320 insertions(+), 186 deletions(-) diff --git a/docs/changes.md b/docs/changes.md index 0fd92a6..624bdd2 100644 --- a/docs/changes.md +++ b/docs/changes.md @@ -66,10 +66,19 @@ Table of contents Added --- +- New processor scoped to track connections that exist +- Better Destroy logic for connections +- Added UPTIME to all objects in the library via the create hook +- added `proc:isPaused()` to processors Changed --- - `thread.kill(msg)` now accepts a message. If omitted will use default `thread killed!` +- Modified multiple locations to not create a function during a thread.hold() and reuse. Only functions containing an upvalue currently have functions being created during hold. This will eventually be fixed +- Updated `multi:getStats()` with more information +- `multi.forwardConnection()` now returns the link to the source connection so you can do cleanup +- When connecting to a connection you can unconnect using ref:Unconnect() it has an internal reference to the root connection +- Unnamed threads have a shorter postfix of only 4 random characters Fixed --- diff --git a/init.lua b/init.lua index b6f322e..c4ff878 100644 --- a/init.lua +++ b/init.lua @@ -154,11 +154,18 @@ function multi:isType(type) return self.Type == type end +multi.connection_count = 0 +multi.connection_subscriptions = 0 + function multi:getStats() local stats = { [multi.Name] = { threads = multi:getThreads(), - tasks = multi.Mainloop + tasks = multi.Mainloop, + name = "root", + fullName = "root", + connections = self.connection_count, + subscriptions = self.connection_subscriptions } } local procs = multi:getProcessors() @@ -166,7 +173,11 @@ function multi:getStats() local proc = procs[i] stats[proc:getFullName()] = { threads = proc:getThreads(), - tasks = proc.Mainloop + tasks = proc.Mainloop, + name = proc:getName(), + fullName = proc:getFullName(), + connections = proc.connection_count, + subscriptions = proc.connection_subscriptions } end return stats @@ -202,7 +213,7 @@ end function multi.forwardConnection(src, dest) if multi.isMulitObj(src) and multi.isMulitObj(dest) then - src(function(...) + return src(function(...) dest:Fire(...) end) else @@ -215,163 +226,243 @@ local ignoreconn = true local empty_func = function() end function multi:newConnection(protect,func,kill) - local processor = self - local c={} - local lock = false - local fast = {} - c.__connectionAdded = function() end - c.rawadd = false - c.Parent = self + self.connection_count = self.connection_count + 1 + local processor = self + local c = {} + local lock = false + local fast = {} + c.__connectionAdded = function() end + c.rawadd = false + c.Parent = self + c._child_conns = {} -- tracks connections spawned by operators + c.destroyed = false - setmetatable(c,{ - __call=function(self,...) - local t = ... - if type(t)=="table" then - for i,v in pairs(t) do - if v==self then - local ref = self:Connect(select(2,...)) - if ref then - ref.root_link = select(1,...) - return ref - end - return self - end - end - return self:Connect(...) - else - return self:Connect(...) - end - end, - __unm = function(obj) -- -obj Reverses the order of connected events - local conns = obj:Bind({}) - for i = #conns, 1, -1 do - obj.rawadd = true - obj(conns[i]) - obj.rawadd = false - end - return obj - end, - __mod = function(obj1, obj2) -- % - local cn = self:newConnection() - if (type(obj1) == "function" or type(obj1) == "table" and obj1.Type == multi.registerType("function", "functions")) and type(obj2) == "table" then - obj2(function(...) - cn:Fire(obj1(...)) - end) - elseif type(obj1) == "table" and (type(obj2) == "function" or type(obj2) == "table" and obj2.Type == multi.registerType("function", "functions")) then - local conns = obj1:Bind({}) - for i = 1,#conns do - obj1(function(...) - conns[i](obj2(...)) - end) - end - obj1.__connectionAdded = function(conn, func) - obj1:Unconnect(conn) - obj1.rawadd = true - obj1:Connect(function(...) - func(obj2(...)) - end) - obj1.rawadd = false - end - return obj1 - else - multi.error("Invalid mod!", type(obj1), type(obj2),"Expected function, connection(table)") - end - return cn - end, - __div = function(obj1, obj2) -- / - local cn = self:newConnection() - local ref - if type(obj1) == "function" and type(obj2) == "table" then - obj2(function(...) - local args = {obj1(...)} - if args[1] then - table.remove(args,1) - cn:Fire(multi.unpack(args)) - end - end) - else - multi.error("Invalid divide!", type(obj1), type(obj2),"Expected function/connection(table)") - end - return cn - end, - __concat = function(obj1, obj2) -- .. - local cn = self:newConnection() - local ref - if type(obj1) == "function" and type(obj2) == "table" then - cn(function(...) - if obj1(...) == true then - obj2:Fire(...) - end - end) - cn.__connectionAdded = function(conn, func) - cn:Unconnect(conn) - obj2:Connect(func) - end - elseif type(obj1) == "table" and type(obj2) == "function" then - ref = cn(function(...) - obj1:Fire(...) - obj2(...) - end) - cn.__connectionAdded = function() - cn.rawadd = true - cn:Unconnect(ref) - ref = cn(function(...) - if obj2(...) then - obj1:Fire(...) - end - end) - end - return cn - elseif type(obj1) == "table" and type(obj2) == "table" then - -- - else - error("Invalid concat!", type(obj1), type(obj2),"Expected function/connection(table), connection(table)/function") - end - return cn - end, - __add = function(c1,c2) -- Or - local cn = self:newConnection() - c1(function(...) - cn:Fire(...) - end) - c2(function(...) - cn:Fire(...) - end) - return cn - end, - __mul = function(c1,c2) -- And - local cn = self:newConnection() - local ref1, ref2 - if c1.__hasInstances == nil then - cn.__hasInstances = {2} - cn.__count = {0} - else - cn.__hasInstances = c1.__hasInstances - cn.__hasInstances[1] = cn.__hasInstances[1] + 1 - cn.__count = c1.__count - end + -- Helper: register a child connection for cleanup + local function trackChild(cn) + c._child_conns[#c._child_conns + 1] = cn + return cn + end - ref1 = c1(function(...) - cn.__count[1] = cn.__count[1] + 1 - c1:Lock(ref1) - if cn.__count[1] == cn.__hasInstances[1] then - cn:Fire(...) - cn.__count[1] = 0 - c1:Unlock(ref1) - c2:Unlock(ref2) - end - end) + setmetatable(c, { + __call = function(self, ...) + local t = ... + if type(t) == "table" then + for i, v in pairs(t) do + if v == self then + local ref = self:Connect(select(2, ...)) + if ref then + ref.root_link = select(1, ...) + return ref + end + return self + end + end + return self:Connect(...) + else + return self:Connect(...) + end + end, - ref2 = c2(function(...) - cn.__count[1] = cn.__count[1] + 1 - c2:Lock(ref2) - if cn.__count[1] == cn.__hasInstances[1] then - cn:Fire(...) - cn.__count[1] = 0 - end - end) - return cn - end}) + __unm = function(obj) + local conns = obj:Bind({}) + for i = #conns, 1, -1 do + obj.rawadd = true + obj(conns[i]) + obj.rawadd = false + end + return obj + end, + + __mod = function(obj1, obj2) + local cn = trackChild(self:newConnection()) + if (type(obj1) == "function" or type(obj1) == "table" and obj1.Type == multi.registerType("function", "functions")) and type(obj2) == "table" then + obj2(function(...) + cn:Fire(obj1(...)) + end) + elseif type(obj1) == "table" and (type(obj2) == "function" or type(obj2) == "table" and obj2.Type == multi.registerType("function", "functions")) then + local conns = obj1:Bind({}) + for i = 1, #conns do + obj1(function(...) + conns[i](obj2(...)) + end) + end + obj1.__connectionAdded = function(conn, func) + obj1:Unconnect(conn) + obj1.rawadd = true + obj1:Connect(function(...) + func(obj2(...)) + end) + obj1.rawadd = false + end + return obj1 + else + multi.error("Invalid mod!", type(obj1), type(obj2), "Expected function, connection(table)") + end + return cn + end, + + __div = function(obj1, obj2) + local cn = trackChild(self:newConnection()) + if type(obj1) == "function" and type(obj2) == "table" then + obj2(function(...) + local args = { obj1(...) } + if args[1] then + table.remove(args, 1) + cn:Fire(multi.unpack(args)) + end + end) + else + multi.error("Invalid divide!", type(obj1), type(obj2), "Expected function/connection(table)") + end + return cn + end, + + __concat = function(obj1, obj2) + local cn = trackChild(self:newConnection()) + local ref + if type(obj1) == "function" and type(obj2) == "table" then + cn(function(...) + if obj1(...) == true then + obj2:Fire(...) + end + end) + cn.__connectionAdded = function(conn, func) + cn:Unconnect(conn) + obj2:Connect(func) + end + elseif type(obj1) == "table" and type(obj2) == "function" then + ref = cn(function(...) + obj1:Fire(...) + obj2(...) + end) + cn.__connectionAdded = function() + cn.rawadd = true + cn:Unconnect(ref) + ref = cn(function(...) + if obj2(...) then + obj1:Fire(...) + end + end) + end + return cn + elseif type(obj1) == "table" and type(obj2) == "table" then + -- reserved + else + error("Invalid concat!", type(obj1), type(obj2), "Expected function/connection(table), connection(table)/function") + end + return cn + end, + + __add = function(c1, c2) -- Or + local cn = trackChild(self:newConnection()) + c1(function(...) cn:Fire(...) end) + c2(function(...) cn:Fire(...) end) + return cn + end, + + __mul = function(c1, c2) -- And + local cn = trackChild(self:newConnection()) + local ref1, ref2 + if c1.__hasInstances == nil then + cn.__hasInstances = {2} + cn.__count = {0} + else + cn.__hasInstances = c1.__hasInstances + cn.__hasInstances[1] = cn.__hasInstances[1] + 1 + cn.__count = c1.__count + end + + ref1 = c1(function(...) + cn.__count[1] = cn.__count[1] + 1 + c1:Lock(ref1) + if cn.__count[1] == cn.__hasInstances[1] then + cn:Fire(...) + cn.__count[1] = 0 + c1:Unlock(ref1) + c2:Unlock(ref2) + end + end) + + ref2 = c2(function(...) + cn.__count[1] = cn.__count[1] + 1 + c2:Lock(ref2) + if cn.__count[1] == cn.__hasInstances[1] then + cn:Fire(...) + cn.__count[1] = 0 + end + end) + return cn + end, + }) + + -- ... (Type, ID, FC setup, Lock/Unlock, Fire, Connect, Bind, etc. unchanged) ... + + --- Destroy this connection and all resources it owns. + -- Recursively destroys any connections created by operator overloads (+, *, %, /, ..). + -- Safe to call multiple times. + function c:Destroy() + if self.destroyed then return end + self.destroyed = true + + -- Unlock first so any in-progress Fire() calls drain cleanly + lock = false + + -- Destroy operator-spawned child connections recursively + for i = 1, #self._child_conns do + local child = self._child_conns[i] + if child and type(child.Destroy) == "function" and not child.destroyed then + child:Destroy() + end + end + self._child_conns = {} + + -- Null out root_link back-references on all connection handles + for key, _ in pairs(fast) do + if type(key) == "string" then + local handle = fast[key] + if type(handle) == "table" and rawget(handle, "root_link") then + handle.root_link = nil + end + end + end + + -- Adjust subscription count on parent before wiping fast[] + if self.Parent and self.Parent.connection_subscriptions then + self.Parent.connection_subscriptions = + math.max(0, self.Parent.connection_subscriptions - #fast) + end + + -- Clear all stored callbacks + fast = {} + + -- Remove from parent's object list + if self.Parent then + for i = 1, #self.Parent do + if self.Parent[i] == self then + table.remove(self.Parent, i) + break + end + end + end + + -- Detach hooks so stale references can't re-fire into dead state + self.__connectionAdded = function() end + self.Connect = function() multi:warning("Connect called on destroyed connection") end + self.Fire = function() end + self.Bind = function() return {} end + self.Unconnect = function() end + self.Lock = function() return self end + self.Unlock = function() return self end + + if self.Parent and self.Parent.connection_count then + self.Parent.connection_count = math.max(0, self.Parent.connection_count - 1) + end + + self.Parent = nil + end + + -- Alias + c.destroy = c.Destroy c.Type=multi.registerType("connector", "connections") c.func={} @@ -447,6 +538,7 @@ function multi:newConnection(protect,func,kill) for i = 1, #fast do if fast[conn.ref] == fast[i] then table.remove(self) + self.Parent.connection_subscriptions = self.Parent.connection_subscriptions - 1 return table.remove(fast, i), i end end @@ -481,6 +573,7 @@ function multi:newConnection(protect,func,kill) end function c:Connect(func, name) + self.Parent.connection_subscriptions = self.Parent.connection_subscriptions + 1 local th if thread.getRunningThread then th = thread.getRunningThread() @@ -520,6 +613,10 @@ function multi:newConnection(protect,func,kill) temp.ref = multi.randomString(24) fast[temp.ref] = func temp.name = name + temp.link = self + function temp:Unconnect() + self.link:Unconnect(self) + end if self.rawadd then self.rawadd = false else @@ -531,7 +628,9 @@ function multi:newConnection(protect,func,kill) function c:Bind(t) local temp = fast + self.Parent.connection_subscriptions = self.Parent.connection_subscriptions - #fast fast=t + self.Parent.connection_subscriptions = self.Parent.connection_subscriptions + #t return temp end @@ -541,6 +640,7 @@ function multi:newConnection(protect,func,kill) function c:Remove() local temp = fast + self.Parent.connection_subscriptions = self.Parent.connection_subscriptions - #fast fast={} return temp end @@ -699,6 +799,7 @@ end function multi:create(ref) ref.UID = multi.generate_uuid7() + ref.UPTIME = clock() self.OnObjectCreated:Fire(ref, self) return self end @@ -1171,9 +1272,11 @@ function multi:newProcessor(name, opts, priority) end sandcount = sandcount + 1 + c.connection_count = 0 + c.connection_subscriptions = 0 c.Mainloop = {} c.Type = multi.registerType("process", "processes") - local Active = nothread or false + local Active = nothread or false local task_delay = 0 c.Name = name or "" c.tasks = {} @@ -1299,6 +1402,10 @@ function multi:newProcessor(name, opts, priority) return Active end + function c:isPaused() + return not(Active) + end + function c.Start() Active = true return c @@ -1541,7 +1648,8 @@ function thread.get(name) end function thread.waitFor(name) - thread.hold(function() return thread.get(name)~=nil end) + local check = function() return thread.get(name)~=nil end + thread.hold(check) return thread.get(name) end @@ -1569,6 +1677,7 @@ function thread:newFunctionBase(generator, holdme, TYPE) local tfunc = { GetCreationTimestamp = function() return multi.extract_uuid7_timestamp(UID).iso8601 end, } + tfunc.UPTIME = clock() tfunc.Active = true function tfunc:Pause() self.Active = false @@ -1583,17 +1692,18 @@ function thread:newFunctionBase(generator, holdme, TYPE) return nil, "Function is paused" end local rets, err + local check = function() + if err then + return multi.NIL, err + elseif rets then + local g = rets + rets = nil + return cleanReturns((g[1] or multi.NIL),g[2],g[3],g[4],g[5],g[6],g[7],g[8],g[9],g[10],g[11],g[12],g[13],g[14],g[15],g[16]) + end + end local function wait() if thread.isThread() then - return thread.hold(function() - if err then - return multi.NIL, err - elseif rets then - local g = rets - rets = nil - return cleanReturns((g[1] or multi.NIL),g[2],g[3],g[4],g[5],g[6],g[7],g[8],g[9],g[10],g[11],g[12],g[13],g[14],g[15],g[16]) - end - end) + return thread.hold(check) else while not rets and not err do multi:uManager() @@ -1742,7 +1852,7 @@ function thread:newThread(name, func, ...) multi.OnLoad:Fire() -- This was done incase a threaded function was called before mainloop/uManager was called if type(name) == "function" then func = name - name = "UnnamedThread_"..multi.randomString(16) + name = "UnnamedThread_"..multi.randomString(4) end local c={nil,nil,nil,nil,nil,nil,nil} c.TempRets = {nil,nil,nil,nil,nil,nil,nil,nil,nil,nil} @@ -2388,7 +2498,7 @@ function multi:enableLoadDetection() local temp = self:newProcessor() local t = clock() local stop = false - temp:benchMark(.01):OnBench(function(time,steps) + temp:benchMark(.1):OnBench(function(time,steps) stop = steps end) while not stop do @@ -2401,28 +2511,39 @@ end local lastVal = 0 local last_step = 0 -function multi:getLoad() - if not multi.maxSpd then multi:enableLoadDetection() end +function multi:getLoad(loops) + local proc = proc or multi + if not proc.maxSpd then proc:enableLoadDetection() end local val = nil local bench local bb - self:benchMark(.01).OnBench(function(time,steps) - bench = steps - bb = steps - end) - _,timeout = multi.hold(function() - return bench - end,{sleep=.012}) + local avg = 0 + local loops = loops or 5 + for i=1,loops do + self:benchMark(1).OnBench(function(time,steps) + bench = steps + bb = steps + avg = avg + steps + if steps > proc.maxSpd then + proc.maxSpd = steps + end + end) + _,timeout = multi.hold(function() + return bench + end,{sleep=1.1}) + end + avg = avg/loops if timeout or not bench then bench = 0 bb = 0 end bench = bench^1.5 - val = math.ceil((1-(bench/(multi.maxSpd/2.2)))*100) + val = math.ceil((1-(bench/(proc.maxSpd/2.2)))*100) if val<0 then val = 0 end if val > 100 then val = 100 end lastVal = val last_step = bb*100 + print(proc.maxSpd, bench, bench/proc.maxSpd/2.2,val) return val,last_step end diff --git a/integration/lanesManager/extensions.lua b/integration/lanesManager/extensions.lua index 6a1db82..20a1dfa 100644 --- a/integration/lanesManager/extensions.lua +++ b/integration/lanesManager/extensions.lua @@ -58,16 +58,20 @@ function multi:newSystemThreadedQueue(name) GLOBAL[name] = c end + local peek = function() + return c:peek() + end + + local pop = function() + return c:pop() + end + function c:Hold(opt) local multi, thread = require("multi"):init() if opt.peek then - return thread.hold(function() - return self:peek() - end) + return thread.hold(peek) else - return thread.hold(function() - return self:pop() - end) + return thread.hold(pop) end end