working on 1.16

This commit is contained in:
Ryan Ward 2021-12-18 12:42:14 -05:00
parent 57563688ae
commit cc20914391
7 changed files with 307 additions and 74 deletions

22
client.lua Normal file
View File

@ -0,0 +1,22 @@
package.path = "?/init.lua;?.lua;"..package.path
local multi, thread = require("multi"):init()
net = require("lnet.tcp")
client = net.newTCPClient("localhost",12345)
multi:newThread(function()
while true do
thread.sleep(1)
client:send(multi:getTasksDetails())
end
end)
multi:newThread(function()
while true do
thread.sleep(.01)
multi:newLoop()
end
end)
multi:mainloop()

View File

@ -50,6 +50,7 @@ multi.time = os.time
multi.LinkedPath = multi multi.LinkedPath = multi
multi.lastTime = clock() multi.lastTime = clock()
multi.TIMEOUT = "TIMEOUT" multi.TIMEOUT = "TIMEOUT"
multi.TID = 0
multi.Priority_Core = 1 multi.Priority_Core = 1
multi.Priority_Very_High = 4 multi.Priority_Very_High = 4
@ -92,8 +93,9 @@ end
--Processor --Processor
local priorityTable = {[0]="Round-Robin",[1]="Balanced",[2]="Top-Down",[3]="Timed-Based-Balancer"} local priorityTable = {[0]="Round-Robin",[1]="Balanced",[2]="Top-Down",[3]="Timed-Based-Balancer"}
local ProcessName = {[true]="SubProcessor",[false]="MainProcessor"} local ProcessName = {[true]="SubProcessor",[false]="MainProcessor"}
local globalThreads = {}
function multi:getTasksDetails(t) function multi:getTasksDetails(t)
if t == "string" or not t then if not(t) then
str = { str = {
{"Type <Identifier>","Uptime","Priority","TID"} {"Type <Identifier>","Uptime","Priority","TID"}
} }
@ -126,19 +128,37 @@ function multi:getTasksDetails(t)
end end
end end
local load, steps = self:getLoad() local load, steps = self:getLoad()
if thread.__threads then local thread_count = 0
for i=1,#thread.__threads do local process_count = 0
dat = dat .. "<THREAD: "..thread.__threads[i].Name.." | "..os.clock()-thread.__threads[i].creationTime..">\n" if globalThreads then
end local th_tab = {
return "Load on "..ProcessName[self.Type=="process"].."<"..(self.Name or "Unnamed")..">"..": "..multi.Round(load,2).."%\nCycles Per Second Per Task: "..steps.."\nMemory Usage: "..math.ceil(collectgarbage("count")).." KB\nThreads Running: "..#thread.__threads.."\nSystemThreads Running: "..#(multi.SystemThreads or {}).."\nPriority Scheme: "..priorityTable[multi.defaultSettings.priority or 0].."\n\n"..s.."\n\n"..dat..dat2 {"Thread Name","Uptime","TID","Attached To"}
}
local proc_tab = {
{"Process Name", "Uptime", "PID", "Load", "Cycles per Second per task"}
}
for th,process in pairs(globalThreads) do
if tostring(th.isProcessThread) == "destroyed" then
globalThreads[th] = nil
elseif th.isProcessThread then
local load, steps = process:getLoad()
process_count = process_count + 1
table.insert(proc_tab,{th.Name,os.clock()-th.creationTime,(th.PID or "-1"),load,steps})
else else
return "Load on "..ProcessName[self.Type=="process"].."<"..(self.Name or "Unnamed")..">"..": "..multi.Round(load,2).."%\nCycles Per Second Per Task: "..steps.."\n\nMemory Usage: "..math.ceil(collectgarbage("count")).." KB\nThreads Running: 0\nPriority Scheme: "..priorityTable[multi.defaultSettings.priority or 0].."\n\n"..s..dat2 thread_count = thread_count + 1
table.insert(th_tab,{th.Name,os.clock()-th.creationTime,(th.TID or "-1"),process.Name})
end end
elseif t == "t" or t == "table" then end
dat = multi.AlignTable(proc_tab).. "\n"
dat = dat .. "\n" .. multi.AlignTable(th_tab)
return "Load on "..ProcessName[self.Type=="process"].."<"..(self.Name or "Unnamed")..">"..": "..multi.Round(load,2).."%\nCycles Per Second Per Task: "..steps.."\nMemory Usage: "..math.ceil(collectgarbage("count")).." KB\nProcesses Running: "..process_count.."\nThreads Running: "..thread_count.."\nSystemThreads Running: "..#(multi.SystemThreads or {}).."\nPriority Scheme: "..priorityTable[multi.defaultSettings.priority or 0].."\n\n"..dat..dat2.."\n\n"..s
else
return "Load on "..ProcessName[self.Type=="process"].."<"..(self.Name or "Unnamed")..">"..": "..multi.Round(load,2).."%\nCycles Per Second Per Task: "..steps.."\n\nMemory Usage: "..math.ceil(collectgarbage("count")).." KB\nProcesses Running: "..process_count.."\nThreads Running: 0\nPriority Scheme: "..priorityTable[multi.defaultSettings.priority or 0].."\n\n"..dat2.."\n\n"..s
end
else
local load,steps = self:getLoad() local load,steps = self:getLoad()
str = { str = {
ProcessName = (self.Name or "Unnamed"), ProcessName = (self.Name or "Unnamed"),
ThreadCount = #thread.__threads,
MemoryUsage = math.ceil(collectgarbage("count")), MemoryUsage = math.ceil(collectgarbage("count")),
PriorityScheme = priorityTable[multi.defaultSettings.priority or 0], PriorityScheme = priorityTable[multi.defaultSettings.priority or 0],
SystemLoad = multi.Round(load,2), SystemLoad = multi.Round(load,2),
@ -148,6 +168,7 @@ function multi:getTasksDetails(t)
str.Tasks = {} str.Tasks = {}
str.PausedTasks = {} str.PausedTasks = {}
str.Threads = {} str.Threads = {}
str.Processes = {}
str.Systemthreads = {} str.Systemthreads = {}
for i,v in pairs(self.Mainloop) do for i,v in pairs(self.Mainloop) do
table.insert(str.Tasks,{Link = v, Type=v.Type,Name=v.Name,Uptime=os.clock()-v.creationTime,Priority=self.PriorityResolve[v.Priority],TID = v.TID}) table.insert(str.Tasks,{Link = v, Type=v.Type,Name=v.Name,Uptime=os.clock()-v.creationTime,Priority=self.PriorityResolve[v.Priority],TID = v.TID})
@ -155,9 +176,18 @@ function multi:getTasksDetails(t)
for v,i in pairs(multi.PausedObjects) do for v,i in pairs(multi.PausedObjects) do
table.insert(str.Tasks,{Link = v, Type=v.Type,Name=v.Name,Uptime=os.clock()-v.creationTime,Priority=self.PriorityResolve[v.Priority],TID = v.TID}) table.insert(str.Tasks,{Link = v, Type=v.Type,Name=v.Name,Uptime=os.clock()-v.creationTime,Priority=self.PriorityResolve[v.Priority],TID = v.TID})
end end
for i=1,#thread.__threads do for th,process in pairs(globalThreads) do
table.insert(str.Threads,{Uptime = os.clock()-thread.__threads[i].creationTime,Name = thread.__threads[i].Name,Link = thread.__threads[i],TID = thread.__threads[i].TID}) if tostring(th.isProcessThread) == "destroyed" then
globalThreads[th] = nil
elseif th.isProcessThread then
local load, steps = process:getLoad()
table.insert(str.Processes,{Uptime = os.clock()-th.creationTime, Name = th.Name, Link = th, TID = th.TID,Load = load,Steps = steps})
else
table.insert(str.Threads,{Uptime = os.clock()-th.creationTime,Name = th.Name,Link = th,TID = th.TID,Attached_to = process})
end end
end
str.ThreadCount = #str.Threads
str.ProcessCount = #str.Processes
if multi.SystemThreads then if multi.SystemThreads then
for i=1,#multi.SystemThreads do for i=1,#multi.SystemThreads do
table.insert(str.Systemthreads,{Uptime = os.clock()-multi.SystemThreads[i].creationTime,Name = multi.SystemThreads[i].Name,Link = multi.SystemThreads[i],TID = multi.SystemThreads[i].count}) table.insert(str.Systemthreads,{Uptime = os.clock()-multi.SystemThreads[i].creationTime,Name = multi.SystemThreads[i].Name,Link = multi.SystemThreads[i],TID = multi.SystemThreads[i].count})
@ -293,6 +323,17 @@ function multi:Destroy()
self.OnObjectDestroyed:Fire(c[i]) self.OnObjectDestroyed:Fire(c[i])
c[i]:Destroy() c[i]:Destroy()
end end
local new = {}
for th,proc in pairs(globalThreads) do
if proc == self then
th:Destroy()
table.remove(globalThreads,th)
else
new[th]=proc
end
end
globalThreads = new
multi.setType(self,multi.DestroyedObj)
else else
for i=1,#self.Parent.Mainloop do for i=1,#self.Parent.Mainloop do
if self.Parent.Mainloop[i]==self then if self.Parent.Mainloop[i]==self then
@ -806,11 +847,12 @@ function multi:newTLoop(func,set)
end end
function c:Act() function c:Act()
if self.timer:Get()>=self.set then if self.timer:Get()>=self.set then
print("Acting...")
self.life=self.life+1 self.life=self.life+1
self.timer:Reset()
for i=1,#self.func do for i=1,#self.func do
self.func[i](self,self.life) self.func[i](self,self.life)
end end
self.timer:Reset()
end end
end end
function c:Resume() function c:Resume()
@ -939,11 +981,10 @@ function multi.getCurrentProcess()
return __CurrentProcess return __CurrentProcess
end end
local globalThreads = {} local sandcount = 1
local sandcount = 0
function multi:newProcessor(name) function multi:newProcessor(name)
local c = {} local c = {}
print("Proc Created:",sandcount)
setmetatable(c,{__index = self}) setmetatable(c,{__index = self})
local multi,thread = require("multi"):init() -- We need to capture the t in thread local multi,thread = require("multi"):init() -- We need to capture the t in thread
local name = name or "Processor_"..sandcount local name = name or "Processor_"..sandcount
@ -951,7 +992,7 @@ function multi:newProcessor(name)
c.Mainloop = {} c.Mainloop = {}
c.Type = "process" c.Type = "process"
c.Active = false c.Active = false
c.Name = "multi.process<".. (name or "") .. ">" c.Name = name or ""
c.process = self:newThread(c.Name,function() c.process = self:newThread(c.Name,function()
while true do while true do
thread.hold(function() thread.hold(function()
@ -962,6 +1003,8 @@ function multi:newProcessor(name)
__CurrentProcess = self __CurrentProcess = self
end end
end) end)
c.process.isProcessThread = true
c.process.PID = sandcount
c.OnError = c.process.OnError c.OnError = c.process.OnError
function c.Start() function c.Start()
c.Active = true c.Active = true
@ -971,6 +1014,9 @@ function multi:newProcessor(name)
c.Active = false c.Active = false
return self return self
end end
function c:Destroy()
self.OnObjectDestroyed:Fire(c)
end
c:attachScheduler() c:attachScheduler()
c.initThreads() c.initThreads()
return c return c
@ -995,7 +1041,7 @@ function thread.getRunningThread()
local t = coroutine.running() local t = coroutine.running()
if t then if t then
for i,v in pairs(threads) do for i,v in pairs(threads) do
if t==v.thread then if t==i.thread then
return v return v
end end
end end
@ -1300,6 +1346,7 @@ function multi:attachScheduler()
return self return self
end end
c.Destroy = c.Kill c.Destroy = c.Kill
c.kill = c.Kill
function c.ref:send(name,val) function c.ref:send(name,val)
ret=coroutine.yield({Name=name,Value=val}) ret=coroutine.yield({Name=name,Value=val})
end end
@ -1327,7 +1374,7 @@ function multi:attachScheduler()
self.Globals=v self.Globals=v
end end
table.insert(threads,c) table.insert(threads,c)
table.insert(globalThreads,c) globalThreads[c] = self
if initT==false then if initT==false then
self.initThreads() self.initThreads()
end end
@ -1519,7 +1566,6 @@ function multi:attachScheduler()
elseif threads[i] and threads[i].task == "holdW" then elseif threads[i] and threads[i].task == "holdW" then
if clock() - threads[i].intervalR>=threads[i].interval then if clock() - threads[i].intervalR>=threads[i].interval then
threads[i].pos = threads[i].pos + 1 threads[i].pos = threads[i].pos + 1
print(threads[i].pos,threads[i].count)
t0,t1,t2,t3,t4,t5,t6,t7,t8,t9,t10,t11,t12,t13,t14,t15 = threads[i].func() t0,t1,t2,t3,t4,t5,t6,t7,t8,t9,t10,t11,t12,t13,t14,t15 = threads[i].func()
if t0 then if t0 then
threads[i].task = "" threads[i].task = ""
@ -2155,11 +2201,12 @@ end
local busy = false local busy = false
local lastVal = 0 local lastVal = 0
local last_step = 0
local bb = 0 local bb = 0
function multi:getLoad() function multi:getLoad()
if not multi.maxSpd then self:enableLoadDetection() end if not multi.maxSpd then self:enableLoadDetection() end
if busy then return lastVal end if busy then return lastVal,last_step end
local val = nil local val = nil
if thread.isThread() then if thread.isThread() then
local bench local bench
@ -2189,7 +2236,8 @@ function multi:getLoad()
if val<0 then val = 0 end if val<0 then val = 0 end
if val > 100 then val = 100 end if val > 100 then val = 100 end
lastVal = val lastVal = val
return val,bb*100 last_step = bb*100
return val,last_step
end end
function multi:setPriority(s) function multi:setPriority(s)

View File

@ -55,19 +55,84 @@ local threads = {}
local count = 1 local count = 1
local started = false local started = false
local livingThreads = {} local livingThreads = {}
function THREAD:newFunction(func,holup)
return function(...) function THREAD:newFunction(func,holdme)
local t = multi:newSystemThread("SystemThreadedFunction",function(...) local tfunc = {}
return func(...) tfunc.Active = true
end,...) function tfunc:Pause()
return thread:newFunction(function() self.Active = false
thread.hold(function() return t.thread end)
return thread.hold(function()
return t.thread:join(.001)
end)
end,holup)()
end end
function tfunc:Resume()
self.Active = true
end
function tfunc:holdMe(b)
holdme = b
end
local function noWait()
return nil, "Function is paused"
end
local rets, err
local function wait(no)
if thread.isThread() and not (no) then
return multi.hold(function()
if err then
return nil, err
elseif rets then
return unpack(rets)
end
end)
else
while not rets and not err do
multi.scheduler:Act()
end
if err then
return nil,err
end
return unpack(rets)
end
end
tfunc.__call = function(t,...)
if not t.Active then
if holdme then
return nil, "Function is paused"
end
return {
isTFunc = true,
wait = noWait,
connect = function(f)
f(nil,"Function is paused")
end
}
end
local t = multi:newSystemThread("SystemThreadedFunction",func,...)
t.OnDeath(function(self,...) rets = {...} end)
t.OnError(function(self,e) err = e end)
if holdme then
return wait()
end
local temp = {
OnStatus = multi:newConnection(),
OnError = multi:newConnection(),
OnReturn = multi:newConnection(),
isTFunc = true,
wait = wait,
connect = function(f)
local tempConn = multi:newConnection()
t.OnDeath(function(self,...) if f then f(...) else tempConn:Fire(...) end end)
t.OnError(function(self,err) if f then f(nil,err) else tempConn:Fire(nil,err) end end)
return tempConn
end
}
t.OnDeath(function(self,...) temp.OnReturn:Fire(...) end)
t.OnError(function(self,err) temp.OnError:Fire(err) end)
t.linkedFunction = temp
t.statusconnector = temp.OnStatus
return temp
end
setmetatable(tfunc,tfunc)
return tfunc
end end
function multi:newSystemThread(name, func, ...) function multi:newSystemThread(name, func, ...)
multi.InitSystemThreadErrorHandler() multi.InitSystemThreadErrorHandler()
rand = math.random(1, 10000000) rand = math.random(1, 10000000)
@ -104,48 +169,66 @@ function multi:newSystemThread(name, func, ...)
self.alive = false self.alive = false
end end
table.insert(multi.SystemThreads, c) table.insert(multi.SystemThreads, c)
c.OnDeath = multi:newConnection()
c.OnError = multi:newConnection() c.OnError = multi:newConnection()
GLOBAL["__THREADS__"] = livingThreads GLOBAL["__THREADS__"] = livingThreads
return c return c
end end
multi.OnSystemThreadDied = multi:newConnection()
local function detectLuaError(str)
return type(str)=="string" and str:match("%.lua:%d*:")
end
local function tableLen(tab)
local len = 0
for i,v in pairs(tab) do
len = len + 1
end
return len
end
function multi.InitSystemThreadErrorHandler() function multi.InitSystemThreadErrorHandler()
if started == true then if started == true then
return return
end end
started = true started = true
multi:newThread( multi:newThread("ThreadErrorHandler",function()
"ThreadErrorHandler",
function()
local threads = multi.SystemThreads local threads = multi.SystemThreads
while true do while true do
thread.sleep(.5) -- switching states often takes a huge hit on performance. half a second to tell me there is an error is good enough. thread.sleep(.1) -- switching states often takes a huge hit on performance. half a second to tell me there is an error is good enough.
for i = #threads, 1, -1 do for i = #threads, 1, -1 do
local v, err, t = threads[i].thread:join(.001) local _,data = pcall(function()
if err then return {threads[i].thread:join(1)}
if err:find("Thread was killed!") then end)
print(err) local v, err, t = data[1],data[2],data[3]
if detectLuaError(err) then
if err:find("Thread was killed!\1") then
livingThreads[threads[i].Id] = {false, threads[i].Name} livingThreads[threads[i].Id] = {false, threads[i].Name}
threads[i].alive = false threads[i].alive = false
multi.OnSystemThreadDied:Fire(threads[i].Id) threads[i].OnDeath:Fire(threads[i],nil,"Thread was killed!")
GLOBAL["__THREADS__"] = livingThreads GLOBAL["__THREADS__"] = livingThreads
table.remove(threads, i) table.remove(threads, i)
elseif err:find("stack traceback") then else
print(err)
threads[i].OnError:Fire(threads[i], err, "Error in systemThread: '" .. threads[i].name .. "' <" .. err .. ">") threads[i].OnError:Fire(threads[i], err, "Error in systemThread: '" .. threads[i].name .. "' <" .. err .. ">")
threads[i].alive = false threads[i].alive = false
livingThreads[threads[i].Id] = {false, threads[i].Name} livingThreads[threads[i].Id] = {false, threads[i].Name}
multi.OnSystemThreadDied:Fire(threads[i].Id) GLOBAL["__THREADS__"] = livingThreads
table.remove(threads, i)
end
elseif tableLen(data)>0 then
livingThreads[threads[i].Id] = {false, threads[i].Name}
threads[i].alive = false
threads[i].OnDeath:Fire(threads[i],unpack(data))
GLOBAL["__THREADS__"] = livingThreads GLOBAL["__THREADS__"] = livingThreads
table.remove(threads, i) table.remove(threads, i)
end end
end end
end end
end end).OnError(function(...)
end print("Error!",...)
) end)
end end
print("Integrated Lanes!") multi.print("Integrated Lanes!")
multi.integration = {} -- for module creators multi.integration = {} -- for module creators
multi.integration.GLOBAL = GLOBAL multi.integration.GLOBAL = GLOBAL
multi.integration.THREAD = THREAD multi.integration.THREAD = THREAD

View File

@ -82,7 +82,7 @@ local function INIT(__GlobalLinda,__SleepingLinda)
THREAD.__CORES = tonumber(io.popen("nproc --all"):read("*n")) THREAD.__CORES = tonumber(io.popen("nproc --all"):read("*n"))
end end
function THREAD.kill() -- trigger the lane destruction function THREAD.kill() -- trigger the lane destruction
error("Thread was killed!") error("Thread was killed!\1")
end end
function THREAD.getName() function THREAD.getName()
return THREAD_NAME return THREAD_NAME

View File

@ -49,19 +49,68 @@ local THREAD_ID = 1
local OBJECT_ID = 0 local OBJECT_ID = 0
local stf = 0 local stf = 0
function THREAD:newFunction(func,holup) function THREAD:newFunction(func,holup)
stf = stf + 1 local tfunc = {}
return function(...) tfunc.Active = true
local t = multi:newSystemThread("STF"..stf,func,...) function tfunc:Pause()
return thread:newFunction(function() self.Active = false
return thread.hold(function()
if t.stab["returns"] then
local dat = t.stab.returns
t.stab.returns = nil
return unpack(dat)
end end
end) function tfunc:Resume()
end,holup)() self.Active = true
end end
function tfunc:holdMe(b)
holdme = b
end
local function noWait()
return nil, "Function is paused"
end
local rets, err
local function wait(no)
if thread.isThread() and not (no) then
-- In a thread
else
-- Not in a thread
end
end
tfunc.__call = function(t,...)
if not t.Active then
if holdme then
return nil, "Function is paused"
end
return {
isTFunc = true,
wait = noWait,
connect = function(f)
f(nil,"Function is paused")
end
}
end
local t = multi:newSystemThread("SystemThreadedFunction",func,...)
t.OnDeath(function(self,...) rets = {...} end)
t.OnError(function(self,e) err = e end)
if holdme then
return wait()
end
local temp = {
OnStatus = multi:newConnection(),
OnError = multi:newConnection(),
OnReturn = multi:newConnection(),
isTFunc = true,
wait = wait,
connect = function(f)
local tempConn = multi:newConnection()
t.OnDeath(function(self,...) if f then f(...) else tempConn:Fire(...) end end)
t.OnError(function(self,err) if f then f(nil,err) else tempConn:Fire(nil,err) end end)
return tempConn
end
}
t.OnDeath(function(self,...) temp.OnReturn:Fire(...) end)
t.OnError(function(self,err) temp.OnError:Fire(err) end)
t.linkedFunction = temp
t.statusconnector = temp.OnStatus
return temp
end
setmetatable(tfunc,tfunc)
return tfunc
end end
function multi:newSystemThread(name,func,...) function multi:newSystemThread(name,func,...)
local c = {} local c = {}
@ -73,6 +122,26 @@ function multi:newSystemThread(name,func,...)
GLOBAL["__THREAD_"..c.ID] = {ID=c.ID,Name=c.name,Thread=c.thread} GLOBAL["__THREAD_"..c.ID] = {ID=c.ID,Name=c.name,Thread=c.thread}
GLOBAL["__THREAD_COUNT"] = THREAD_ID GLOBAL["__THREAD_COUNT"] = THREAD_ID
THREAD_ID=THREAD_ID+1 THREAD_ID=THREAD_ID+1
multi:newThread(function()
while true do
thread.yield()
if c.stab["returns"] then
c.OnDeath:Fire(c,unpack(t.stab.returns))
t.stab.returns = nil
thread.kill()
end
local error = c.thread:getError()
if error then
if error:find("Thread Killed!\1") then
c.OnDeath:Fire(c,"Thread Killed!")
thread.kill()
else
c.OnError:Fire(c,error)
thread.kill()
end
end
end
end)
return c return c
end end
function love.threaderror(thread, errorstr) function love.threaderror(thread, errorstr)

View File

@ -92,7 +92,7 @@ function threads.getCores()
return love.system.getProcessorCount() return love.system.getProcessorCount()
end end
function threads.kill() function threads.kill()
error("Thread Killed!") error("Thread Killed!\1")
end end
function threads.getThreads() function threads.getThreads()
local t = {} local t = {}

11
server.lua Normal file
View File

@ -0,0 +1,11 @@
package.path = "?/init.lua;?.lua;"..package.path
local multi, thread = require("multi"):init()
net = require("lnet.tcp")
server = net.newTCPServer(12345)
server.OnDataRecieved(function(self,data)
print(data)
end)
multi:mainloop()