From 9e6552d42e1506bbc15e07239ad534158c709bad Mon Sep 17 00:00:00 2001 From: Ryan Ward Date: Sun, 16 Jul 2023 11:10:35 -0400 Subject: [PATCH] Working on adding a Hold method to all objects. Will document how they all work when done. --- init.lua | 42 ++-- integration/lanesManager/extensions.lua | 28 +++ integration/loveManager/extensions.lua | 242 ++++++++++++++++++++++++ integration/loveManager/threads.lua | 6 +- lovethreads/main.lua | 45 ++--- 5 files changed, 308 insertions(+), 55 deletions(-) diff --git a/init.lua b/init.lua index 59172fe..821e0f1 100644 --- a/init.lua +++ b/init.lua @@ -482,9 +482,9 @@ function multi:newConnection(protect,func,kill) return temp end - c.Hold = thread:newFunction(function(self) - return thread.hold(self) - end, true) + function c:Hold(self) + return multi.hold(self) + end c.connect=c.Connect c.GetConnection=c.getConnection @@ -1158,36 +1158,18 @@ function multi:newProcessor(name, nothread, priority) end function multi.hold(func,opt) - if thread.isThread() then - if type(func) == "function" or type(func) == "table" then - return thread.hold(func,opt) - end - return thread.sleep(func) - end - local death = false + if thread.isThread() then return thread.hold(func, opt) end local proc = multi.getCurrentTask() proc:Pause() - if type(func)=="number" then - thread:newThread("Hold_func",function() - thread.hold(func) - death = true - end) - while not death do - multi:uManager() - end - proc:Resume() - else - local rets - thread:newThread("Hold_func",function() - rets = {thread.hold(func,opt)} - death = true - end) - while not death do - multi:uManager() - end - proc:Resume() - return multi.unpack(rets) + local rets + thread:newThread("Hold_func",function() + rets = {thread.hold(func,opt)} + end) + while rets == nil do + multi:uManager() end + proc:Resume() + return multi.unpack(rets) end -- Threading stuff diff --git a/integration/lanesManager/extensions.lua b/integration/lanesManager/extensions.lua index 343dd8b..d622aab 100644 --- a/integration/lanesManager/extensions.lua +++ b/integration/lanesManager/extensions.lua @@ -58,6 +58,19 @@ function multi:newSystemThreadedQueue(name) GLOBAL[name] = c end + function c:Hold(opt) + local multi, thread = require("multi"):init() + if opt.peek then + return thread.hold(function() + return self:peek() + end) + else + return thread.hold(function() + return self:pop() + end) + end + end + self:create(c) return c @@ -89,6 +102,17 @@ function multi:newSystemThreadedTable(name) GLOBAL[name] = c end + function c:Hold(opt) + local multi, thread = require("multi"):init() + if opt.key then + return thread.hold(function() + return self.tab[opt.key] + end) + else + multi.error("Must provide a key to check opt.key = 'key'") + end + end + self:create(c) return c @@ -214,6 +238,10 @@ function multi:newSystemThreadedJobQueue(n) end,i).OnError(multi.error) end + function c:Hold(opt) + return thread.hold(self.OnJobCompleted) + end + self:create(c) return c diff --git a/integration/loveManager/extensions.lua b/integration/loveManager/extensions.lua index e69de29..2b78793 100644 --- a/integration/loveManager/extensions.lua +++ b/integration/loveManager/extensions.lua @@ -0,0 +1,242 @@ +if not ISTHREAD then + multi, thread = require("multi").init() + GLOBAL = multi.integration.GLOBAL + THREAD = multi.integration.THREAD +end + +function multi:newSystemThreadedQueue(name) + local name = name or multi.randomString(16) + + local c = {} + + c.Name = name + c.Type = multi.SQUEUE + c.chan = love.thread.getChannel(name) + + function c:push(dat) + self.chan:push(THREAD.packValue(dat)) + end + + function c:pop() + return THREAD.unpackValue(self.chan:pop()) + end + + function c:peek() + return THREAD.unpackValue(self.chan:peek()) + end + + function c:init() + self.chan = love.thread.getChannel(self.Name) + return self + end + + function c:Hold(opt) + if opt.peek then + return thread.hold(function() + return self:peek() + end) + else + return thread.hold(function() + return self:pop() + end) + end + end + + GLOBAL[name] = c + + self:create(c) + + return c +end + +function multi:newSystemThreadedTable(name) + local name = name or multi.randomString(16) + + local c = {} + + c.Name = name + c.Type = multi.STABLE + c.tab = THREAD.createTable(name) + + function c:init() + self.tab = THREAD.createTable(self.Name) + setmetatable(self,{ + __index = function(t, k) + print("Getting...", k) + return self.tab[k] + end, + __newindex = function(t,k,v) + print("Setting...", k, v) + self.tab[k] = v + end + }) + return self + end + + function c:Hold(opt) + if opt.key then + return thread.hold(function() + return self.tab[opt.key] + end) + else + multi.error("Must provide a key to check opt.key = 'key'") + end + end + + setmetatable(c,{ + __index = function(t, k) + print("Getting...", k) + return c.tab[k] + end, + __newindex = function(t,k,v) + print("Setting...", k, v) + c.tab[k] = v + end + }) + + GLOBAL[name] = c + + self:create(c) + + return c +end + +local jqc = 1 +function multi:newSystemThreadedJobQueue(n) + local c = {} + + c.cores = n or THREAD.getCores() + c.registerQueue = {} + c.Type = multi.SJOBQUEUE + c.funcs = THREAD.createTable("__JobQueue_"..jqc.."_table") + c.queue = multi:newSystemThreadedQueue("__JobQueue_"..jqc.."_queue") + c.queueReturn = multi:newSystemThreadedQueue("__JobQueue_"..jqc.."_queueReturn") + c.queueAll = multi:newSystemThreadedQueue("__JobQueue_"..jqc.."_queueAll") + c.id = 0 + c.OnJobCompleted = multi:newConnection() + + local allfunc = 0 + + function c:doToAll(func) + for i = 1, self.cores do + self.queueAll:push({allfunc, func}) + end + allfunc = allfunc + 1 + end + function c:registerFunction(name, func) + if self.funcs[name] then + multi.error("A function by the name "..name.." has already been registered!") + end + self.funcs[name] = func + end + function c:pushJob(name,...) + self.id = self.id + 1 + self.queue:push{name,self.id,...} + return self.id + end + function c:isEmpty() + return queueJob:peek()==nil + end + local nFunc = 0 + function c:newFunction(name,func,holup) -- This registers with the queue + if type(name)=="function" then + holup = func + func = name + name = "JQ_Function_"..nFunc + end + nFunc = nFunc + 1 + c:registerFunction(name,func) + return thread:newFunction(function(...) + local id = c:pushJob(name,...) + local link + local rets + link = c.OnJobCompleted(function(jid,...) + if id==jid then + rets = multi.pack(...) + end + end) + return thread.hold(function() + if rets then + return multi.unpack(rets) or multi.NIL + end + end) + end,holup),name + end + thread:newThread("jobManager",function() + while true do + thread.yield() + local dat = c.queueReturn:pop() + if dat then + c.OnJobCompleted:Fire(multi.unpack(dat)) + end + end + end) + for i=1,c.cores do + multi:newSystemThread("JobQueue_"..jqc.."_worker_"..i,function(jqc) + local multi, thread = require("multi"):init() + require("love.timer") + local clock = os.clock + local funcs = THREAD.createTable("__JobQueue_"..jqc.."_table") + local queue = THREAD.waitFor("__JobQueue_"..jqc.."_queue") + local queueReturn = THREAD.waitFor("__JobQueue_"..jqc.."_queueReturn") + local lastProc = clock() + local queueAll = THREAD.waitFor("__JobQueue_"..jqc.."_queueAll") + local registry = {} + _G["__QR"] = queueReturn + setmetatable(_G,{__index = funcs}) + thread:newThread("startUp",function() + while true do + thread.yield() + local all = queueAll:peek() + if all and not registry[all[1]] then + lastProc = os.clock() + queueAll:pop()[2]() + end + end + end) + thread:newThread("runner",function() + thread.sleep(.1) + while true do + thread.yield() + local all = queueAll:peek() + if all and not registry[all[1]] then + lastProc = os.clock() + queueAll:pop()[2]() + end + local dat = thread.hold(queue) + if dat then + multi:newThread("Test",function() + lastProc = os.clock() + local name = table.remove(dat,1) + local id = table.remove(dat,1) + local tab = {funcs[name](multi.unpack(dat))} + table.insert(tab,1,id) + queueReturn:push(tab) + end) + end + end + end).OnError(multi.error) + thread:newThread("Idler",function() + while true do + thread.yield() + if clock()-lastProc> 2 then + THREAD.sleep(.05) + else + THREAD.sleep(.001) + end + end + end) + multi:mainloop() + end,jqc) + end + + function c:Hold(opt) + return thread.hold(self.OnJobCompleted) + end + + jqc = jqc + 1 + + self:create(c) + + return c +end \ No newline at end of file diff --git a/integration/loveManager/threads.lua b/integration/loveManager/threads.lua index e7c049e..6610690 100644 --- a/integration/loveManager/threads.lua +++ b/integration/loveManager/threads.lua @@ -127,7 +127,11 @@ function INIT() repeat wait() until GLOBAL[name] - return GLOBAL[name] + if type(GLOBAL[name].init) == "function" then + return GLOBAL[name]:init() + else + return GLOBAL[name] + end end, true) function THREAD.getCores() diff --git a/lovethreads/main.lua b/lovethreads/main.lua index c06a9ff..199dc70 100644 --- a/lovethreads/main.lua +++ b/lovethreads/main.lua @@ -1,42 +1,39 @@ package.path = "../?/init.lua;../?.lua;"..package.path -local multi, thread = require("multi"):init() +local multi, thread = require("multi"):init{print=true, warning = true, error=true} GLOBAL, THREAD = require("multi.integration.loveManager"):init() -GLOBAL["Test"] = {1,2,3, function() print("HI") end} - -for i,v in pairs(GLOBAL["Test"]) do - print(i,v) - if type(v) == "function" then v() end -end +local queue = multi:newSystemThreadedQueue("TestQueue") +local tab = multi:newSystemThreadedTable("TestTable") local test = multi:newSystemThread("Test",function() + local queue = THREAD.waitFor("TestQueue") + local tab = THREAD.waitFor("TestTable") print("THREAD_ID:",THREAD_ID) - GLOBAL["Test2"] = "We did it!" - eror("hahaha") + queue:push("Did it work?") + tab["Test"] = true return 1,2,3 end) -test.OnDeath(function(a,b,c) - print("Thread finished!",a,b,c) +multi:newThread("QueueTest", function() + print(thread.hold(queue)) + print(thread.hold(tab, {key="Test"})) + print("Done!") end) -test.OnError(function(self, err) - print("Got Error!",err) +local jq = multi:newSystemThreadedJobQueue(n) + +jq:registerFunction("test",function(a, b, c) + print(a, b+c) + return a+b+c end) -local func = THREAD:newFunction(function(a,b,c) - print("let's do this!",1,2,3) - return true -end) +print("Job:",jq:pushJob("test",1,2,3)) +print("Job:",jq:pushJob("test",2,3,4)) +print("Job:",jq:pushJob("test",5,6,7)) -func(1,2,3).OnReturn(function(ret) - print("Done",ret) -end) - -thread:newThread(function() - print("Waiting...") - print(THREAD.waitFor("Test2")) +jq.OnJobCompleted(function(...) + print("Job Completed!", ...) end) function love.draw()