fixing bugs

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
 ---

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
+	self.connection_count = self.connection_count + 1
-	local c={}
+    local processor = self
-	local lock = false
+    local c = {}
-	local fast = {}
+    local lock = false
-	c.__connectionAdded = function() end
+    local fast = {}
-	c.rawadd = false
+    c.__connectionAdded = function() end
-	c.Parent = self
+    c.rawadd = false
+    c.Parent = self
+    c._child_conns = {}   -- tracks connections spawned by operators
+    c.destroyed = false
 
-	setmetatable(c,{
+    -- Helper: register a child connection for cleanup
-		__call=function(self,...)
+    local function trackChild(cn)
-		local t = ...
+        c._child_conns[#c._child_conns + 1] = cn
-		if type(t)=="table" then
+        return cn
-			for i,v in pairs(t) do
+    end
-				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
 
-		ref1 = c1(function(...)
+    setmetatable(c, {
-			cn.__count[1] = cn.__count[1] + 1
+        __call = function(self, ...)
-			c1:Lock(ref1)
+            local t = ...
-			if cn.__count[1] == cn.__hasInstances[1] then
+            if type(t) == "table" then
-				cn:Fire(...)
+                for i, v in pairs(t) do
-				cn.__count[1] = 0
+                    if v == self then
-				c1:Unlock(ref1)
+                        local ref = self:Connect(select(2, ...))
-				c2:Unlock(ref2)
+                        if ref then
-			end
+                            ref.root_link = select(1, ...)
-		end)
+                            return ref
+                        end
+                        return self
+                    end
+                end
+                return self:Connect(...)
+            else
+                return self:Connect(...)
+            end
+        end,
 
-		ref2 = c2(function(...)
+        __unm = function(obj)
-			cn.__count[1] = cn.__count[1] + 1
+            local conns = obj:Bind({})
-			c2:Lock(ref2)
+            for i = #conns, 1, -1 do
-			if cn.__count[1] == cn.__hasInstances[1] then
+                obj.rawadd = true
-				cn:Fire(...)
+                obj(conns[i])
-				cn.__count[1] = 0
+                obj.rawadd = false
-			end
+            end
-		end)
+            return obj
-		return cn
+        end,
-	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()
+				return thread.hold(check)
-					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)
 			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()
+function multi:getLoad(loops)
-	if not multi.maxSpd then multi:enableLoadDetection() end
+	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)
+	local avg = 0
-		bench = steps
+	local loops = loops or 5
-		bb = steps
+	for i=1,loops do
-	end)
+		self:benchMark(1).OnBench(function(time,steps)
-	_,timeout = multi.hold(function()
+			bench = steps
-		return bench
+			bb = steps
-	end,{sleep=.012})
+			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
 

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 thread.hold(peek)
-                return self:peek()
-            end)
         else
-            return thread.hold(function()
+            return thread.hold(pop)
-                return self:pop()
-            end)
         end
 	end