epicknex/multi
1
0
Fork 0
mirror of https://github.com/rayaman/multi.git synced 2026-09-04 23:17:35 -04:00
Code Issues Packages Projects Releases Wiki Activity

Reverted changes with code structure

Browse Source
This commit is contained in:
server
2022-09-29 12:29:05 -04:00
parent 57b4871a85
commit b727fb15b5
15 changed files with 16 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
integration/lanesManager/extensions.lua
-311
View File
@@ -1,311 +0,0 @@
--[[
MIT License
Copyright (c) 2022 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
local multi, thread = require("multi"):init()
if not (GLOBAL and THREAD) then
local GLOBAL, THREAD = multi.integration.GLOBAL,multi.integration.THREAD
else
lanes = require("lanes")
end
function multi:newSystemThreadedQueue(name)
local name = name or multi.randomString(16)
local c = {}
c.Name = name
c.linda = lanes.linda()
function c:push(v)
self.linda:send("Q", v)
end
function c:pop()
return ({self.linda:receive(0, "Q")})[2]
end
function c:peek()
return self.linda:get("Q")
end
function c:init()
return self
end
GLOBAL[name or "_"] = c
return c
end
function multi:newSystemThreadedTable(name)
local name = name or multi.randomString(16)
local c = {}
c.link = lanes.linda()
c.Name = name
setmetatable(c,{
__index = function(t,k)
return c.link:get(k)
end,
__newindex = function(t,k,v)
c.link:set(k,v)
end
})
function c:init()
return self
end
GLOBAL[name or "_"] = c
return c
end
function multi:newSystemThreadedJobQueue(n)
local c = {}
c.cores = n or THREAD.getCores()*2
c.OnJobCompleted = multi:newConnection()
local funcs = multi:newSystemThreadedTable():init()
local queueJob = multi:newSystemThreadedQueue():init()
local queueReturn = multi:newSystemThreadedQueue():init()
local doAll = multi:newSystemThreadedQueue():init()
local ID=1
local jid = 1
function c:isEmpty()
return queueJob:peek()==nil
end
function c:doToAll(func)
for i=1,c.cores do
doAll:push{ID,func}
end
ID = ID + 1
return self
end
function c:registerFunction(name,func)
funcs[name]=func
return self
end
function c:pushJob(name,...)
queueJob:push{name,jid,{...}}
jid = jid + 1
return jid-1
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 = {...}
link:Destroy()
end
end)
return thread.hold(function()
if rets then
return unpack(rets) or multi.NIL
end
end)
end,holup),name
end
thread:newThread("JobQueueManager",function()
while true do
local job = thread.hold(function()
return queueReturn:pop()
end)
local id = table.remove(job,1)
c.OnJobCompleted:Fire(id,unpack(job))
end
end)
for i=1,c.cores do
multi:newSystemThread("SystemThreadedJobQueue",function(queue)
local multi,thread = require("multi"):init()
local idle = os.clock()
local clock = os.clock
local ref = 0
setmetatable(_G,{__index = funcs})
thread:newThread("JobHandler",function()
while true do
local dat = thread.hold(function()
return queueJob:pop()
end)
idle = clock()
local name = table.remove(dat,1)
local jid = table.remove(dat,1)
local args = table.remove(dat,1)
queueReturn:push{jid, funcs[name](unpack(args)),queue}
end
end)
thread:newThread("DoAllHandler",function()
while true do
local dat = thread.hold(function()
return doAll:peek()
end)
if dat then
if dat[1]>ref then
idle = clock()
ref = dat[1]
dat[2]()
doAll:pop()
end
end
end
end)
thread:newThread("IdleHandler",function()
while true do
thread.hold(function()
return clock()-idle>3
end)
THREAD.sleep(.01)
end
end)
multi:mainloop()
end,i).priority = thread.Priority_Core
end
return c
end
function multi:newSystemThreadedConnection(name)
local name = name or multi.randomString(16)
local c = {}
c.CONN = 0x00
c.TRIG = 0x01
c.PING = 0x02
c.PONG = 0x03
local function remove(a, b)
local ai = {}
local r = {}
for k,v in pairs(a) do ai[v]=true end
for k,v in pairs(b) do
if ai[v]==nil then table.insert(r,a[k]) end
end
return r
end
c.CID = THREAD.getID()
c.subscribe = multi:newSystemThreadedQueue("SUB_STC_"..self.Name):init()
c.Name = name
c.links = {} -- All triggers sent from main connection. When a connection is triggered on another thread, they speak to the main then send stuff out.
-- Locals will only live in the thread that creates the original object
local ping
local pong = function(link, links)
local res = thread.hold(function()
return link:peek()[1] == c.PONG
end,{sleep=3})
if not res then
for i=1,#links do
if links[i] == link then
table.remove(links,i,link)
break
end
end
else
link:pop()
end
end
ping = thread:newFunction(function(self)
ping:Pause()
multi.ForEach(self.links, function(link) -- Sync new connections
link:push{self.PING}
multi:newThread("pong Thread", pong, link, links)
end)
thread.sleep(3)
ping:Resume()
end,false)
local function fire(...)
for _, link in pairs(c.links) do
link:push {c.TRIG, {...}}
end
end
thread:newThread("STC_SUB_MAN"..name,function()
local item
while true do
thread.yield()
-- We need to check on broken connections
ping(c) -- Should return instantlly and process this in another thread
item = thread.hold(function() -- This will keep things held up until there is something to process
return c.subscribe:pop()
end)
if item[1] == c.CONN then
multi.ForEach(c.links, function(link) -- Sync new connections
item[2]:push{c.CONN, link}
end)
c.links[#c.links+1] = item[2]
elseif item[1] == c.TRIG then
fire(unpack(item[2]))
c.proxy_conn:Fire(unpack(item[2]))
end
end
end)
--- ^^^ This will only exist in the init thread
function c:Fire(...)
local args = {...}
if self.CID == THREAD.getID() then -- Host Call
for _, link in pairs(self.links) do
link:push {self.TRIG, args}
end
self.proxy_conn:Fire(...)
else
self.subscribe:push {self.TRIG, args}
end
end
function c:init()
local multi, thread = require("multi"):init()
self.links = {}
self.proxy_conn = multi:newConnection()
local mt = getmetatable(self.proxy_conn)
setmetatable(self, {__index = self.proxy_conn, __call = function(t,func) self.proxy_conn(func) end, __add = mt.__add})
if self.CID == THREAD.getID() then return self end
thread:newThread("STC_CONN_MAN"..name,function()
local item
local link_self_ref = multi:newSystemThreadedQueue()
self.subscribe:push{self.CONN, link_self_ref}
while true do
item = thread.hold(function()
return link_self_ref:peek()
end)
if item[1] == self.PING then
link_self_ref:push{self.PONG}
link_self_ref:pop()
elseif item[1] == self.CONN then
if item[2].Name ~= link_self_ref.Name then
table.insert(self.links, item[2])
end
link_self_ref:pop()
elseif item[1] == self.TRIG then
self.proxy_conn:Fire(unpack(item[2]))
link_self_ref:pop()
else
-- This shouldn't be the case
end
end
end)
return self
end
GLOBAL[name] = c
return c
end
integration/lanesManager/init.lua
-175
View File
@@ -1,175 +0,0 @@
--[[
MIT License
Copyright (c) 2022 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
package.path = "?/init.lua;?.lua;" .. package.path
multi, thread = require("multi"):init() -- get it all and have it on all lanes
if multi.integration then -- This allows us to call the lanes manager from supporting modules without a hassle
return {
init = function()
return multi.integration.GLOBAL, multi.integration.THREAD
end
}
end
-- Step 1 get lanes
lanes = require("lanes").configure()
multi.SystemThreads = {}
multi.isMainThread = true
function multi:canSystemThread()
return true
end
function multi:getPlatform()
return "lanes"
end
-- Step 2 set up the Linda objects
local __GlobalLinda = lanes.linda() -- handles global stuff
local __SleepingLinda = lanes.linda() -- handles sleeping stuff
local __ConsoleLinda = lanes.linda() -- handles console stuff
local __StatusLinda = lanes.linda() -- handles pushstatus for stfunctions
local GLOBAL,THREAD = require("multi.integration.lanesManager.threads").init(__GlobalLinda, __SleepingLinda, __StatusLinda, __ConsoleLinda)
local count = 1
local started = false
local livingThreads = {}
function THREAD:newFunction(func,holdme)
return thread:newFunctionBase(function(...)
return multi:newSystemThread("TempSystemThread",func,...)
end,holdme)()
end
function multi:newSystemThread(name, func, ...)
local name = name or multi.randomString(16)
multi.InitSystemThreadErrorHandler()
local rand = math.random(1, 10000000)
local return_linda = lanes.linda()
c = {}
c.name = name
c.Name = name
c.Id = count
c.loadString = {"base","package","os","io","math","table","string","coroutine"}
livingThreads[count] = {true, name}
c.returns = return_linda
c.Type = "sthread"
c.creationTime = os.clock()
c.alive = true
c.priority = THREAD.Priority_Normal
c.thread = lanes.gen("*",
{
globals={ -- Set up some globals
THREAD_NAME = name,
THREAD_ID = count,
THREAD = THREAD,
GLOBAL = GLOBAL,
_Console = __ConsoleLinda
},
priority=c.priority
},function(...)
require("multi.integration.lanesManager.extensions")
local has_error = true
return_linda:set("returns",{func(...)})
has_error = false
end)(...)
count = count + 1
function c:getName()
return c.Name
end
function c:kill()
self.thread:cancel()
self.alive = false
end
table.insert(multi.SystemThreads, c)
c.OnDeath = multi:newConnection()
c.OnError = multi:newConnection()
GLOBAL["__THREADS__"] = livingThreads
return c
end
THREAD.newSystemThread = multi.newSystemThread
function multi.InitSystemThreadErrorHandler()
if started == true then
return
end
started = true
thread:newThread("SystemThreadScheduler",function()
local threads = multi.SystemThreads
local _,data,status,push,temp
while true do
thread.yield()
_,data = __ConsoleLinda:receive(0, "Q")
if data then print(unpack(data)) end
for i = #threads, 1, -1 do
temp = threads[i]
status = temp.thread.status
push = __StatusLinda:get(temp.Id)
if push then
temp.statusconnector:Fire(unpack(({__StatusLinda:receive(nil, temp.Id)})[2]))
end
if status == "done" or temp.returns:get("returns") then
livingThreads[temp.Id] = {false, temp.Name}
temp.alive = false
temp.OnDeath:Fire(unpack(({temp.returns:receive(0, "returns")})[2]))
GLOBAL["__THREADS__"] = livingThreads
table.remove(threads, i)
elseif status == "running" then
--
elseif status == "waiting" then
--
elseif status == "error" then
livingThreads[temp.Id] = {false, temp.Name}
temp.alive = false
temp.OnError:Fire(temp,unpack(temp.returns:receive(0,"returns") or {"Thread Killed!"}))
GLOBAL["__THREADS__"] = livingThreads
table.remove(threads, i)
elseif status == "cancelled" then
livingThreads[temp.Id] = {false, temp.Name}
temp.alive = false
temp.OnError:Fire(temp,"thread_cancelled")
GLOBAL["__THREADS__"] = livingThreads
table.remove(threads, i)
elseif status == "killed" then
livingThreads[temp.Id] = {false, temp.Name}
temp.alive = false
temp.OnError:Fire(temp,"thread_killed")
GLOBAL["__THREADS__"] = livingThreads
table.remove(threads, i)
end
end
end
end)
end
multi.print("Integrated Lanes!")
multi.integration = {} -- for module creators
multi.integration.GLOBAL = GLOBAL
multi.integration.THREAD = THREAD
require("multi.integration.lanesManager.extensions")
return {
init = function()
return GLOBAL, THREAD
end
}
integration/lanesManager/threads.lua
-142
View File
@@ -1,142 +0,0 @@
--[[
MIT License
Copyright (c) 2022 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
local function getOS()
if package.config:sub(1, 1) == "\\" then
return "windows"
else
return "unix"
end
end
local function INIT(__GlobalLinda, __SleepingLinda, __StatusLinda, __Console)
local THREAD = {}
THREAD.Priority_Core = 3
THREAD.Priority_High = 2
THREAD.Priority_Above_Normal = 1
THREAD.Priority_Normal = 0
THREAD.Priority_Below_Normal = -1
THREAD.Priority_Low = -2
THREAD.Priority_Idle = -3
function THREAD.set(name, val)
__GlobalLinda:set(name, val)
end
function THREAD.get(name)
return __GlobalLinda:get(name)
end
function THREAD.waitFor(name)
local function wait()
math.randomseed(os.time())
__SleepingLinda:receive(.001, "__non_existing_variable")
end
repeat
wait()
until __GlobalLinda:get(name)
return __GlobalLinda:get(name)
end
if getOS() == "windows" then
THREAD.__CORES = tonumber(os.getenv("NUMBER_OF_PROCESSORS"))
else
THREAD.__CORES = tonumber(io.popen("nproc --all"):read("*n"))
end
function THREAD.getCores()
return THREAD.__CORES
end
function THREAD.getConsole()
local c = {}
c.queue = __Console
function c.print(...)
c.queue:send("Q", {...})
end
function c.error(err)
c.queue:push("Q",{"ERROR in <"..__THREADNAME__..">: "..err,__THREADID__})
error(err)
end
return c
end
function THREAD.getThreads()
return GLOBAL.__THREADS__
end
if os.getOS() == "windows" then
THREAD.__CORES = tonumber(os.getenv("NUMBER_OF_PROCESSORS"))
else
THREAD.__CORES = tonumber(io.popen("nproc --all"):read("*n"))
end
function THREAD.kill() -- trigger the lane destruction
error("Thread was killed!\1")
end
function THREAD.getName()
return THREAD_NAME
end
function THREAD.getID()
return THREAD_ID
end
function THREAD.pushStatus(...)
local args = {...}
__StatusLinda:send(nil,THREAD_ID, args)
end
_G.THREAD_ID = 0
function THREAD.sleep(n)
math.randomseed(os.time())
__SleepingLinda:receive(n, "__non_existing_variable")
end
function THREAD.hold(n)
local function wait()
math.randomseed(os.time())
__SleepingLinda:receive(.001, "__non_existing_variable")
end
repeat
wait()
until n()
end
local GLOBAL = {}
setmetatable(GLOBAL, {
__index = function(t, k)
return __GlobalLinda:get(k)
end,
__newindex = function(t, k, v)
__GlobalLinda:set(k, v)
end
})
return GLOBAL, THREAD
end
return {init = function(g,s,st,c)
return INIT(g,s,st,c)
end}
integration/loveManager/extensions.lua
-545
View File
@@ -1,545 +0,0 @@
--[[
MIT License
Copyright (c) 2022 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
-- TODO make compatible with lovr
local multi, thread = require("multi").init()
GLOBAL = multi.integration.GLOBAL
THREAD = multi.integration.THREAD
function multi:newSystemThreadedQueue(name)
local name = name or multi.randomString(16)
local c = {}
c.Name = name
local fRef = {"func",nil}
function c:init()
local q = {}
q.chan = love.thread.getChannel(self.Name)
function q:push(dat)
if type(dat) == "function" then
fRef[2] = THREAD.dump(dat)
self.chan:push(fRef)
return
else
self.chan:push(dat)
end
end
function q:pop()
local dat = self.chan:pop()
if type(dat)=="table" and dat[1]=="func" then
return THREAD.loadDump(dat[2])
else
return dat
end
end
function q:peek()
local dat = self.chan:peek()
if type(dat)=="table" and dat[1]=="func" then
return THREAD.loadDump(dat[2])
else
return dat
end
end
return q
end
THREAD.package(name,c)
return c
end
function multi:newSystemThreadedTable(name)
local name = name or multi.randomString(16)
local c = {}
c.Name = name
function c:init()
return THREAD.createTable(self.Name)
end
THREAD.package(name,c)
return c
end
local jqc = 1
function multi:newSystemThreadedJobQueue(n)
local c = {}
c.cores = n or THREAD.getCores()
c.registerQueue = {}
c.funcs = THREAD.createStaticTable("__JobQueue_"..jqc.."_table")
c.queue = love.thread.getChannel("__JobQueue_"..jqc.."_queue")
c.queueReturn = love.thread.getChannel("__JobQueue_"..jqc.."_queueReturn")
c.queueAll = love.thread.getChannel("__JobQueue_"..jqc.."_queueAll")
c.id = 0
c.OnJobCompleted = multi:newConnection()
local allfunc = 0
function c:doToAll(func)
local f = THREAD.dump(func)
for i = 1, self.cores do
self.queueAll:push({allfunc,f})
end
allfunc = allfunc + 1
end
function c:registerFunction(name,func)
if self.funcs[name] then
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 = {...}
link:Destroy()
end
end)
return thread.hold(function()
if rets then
return 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(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 function atomic(channel)
return channel:pop()
end
local clock = os.clock
local funcs = THREAD.createStaticTable("__JobQueue_"..jqc.."_table")
local queue = love.thread.getChannel("__JobQueue_"..jqc.."_queue")
local queueReturn = love.thread.getChannel("__JobQueue_"..jqc.."_queueReturn")
local lastProc = clock()
local queueAll = love.thread.getChannel("__JobQueue_"..jqc.."_queueAll")
local registry = {}
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()
THREAD.loadDump(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()
THREAD.loadDump(queueAll:pop()[2])()
end
local dat = queue:performAtomic(atomic)
if dat then
lastProc = os.clock()
local name = table.remove(dat,1)
local id = table.remove(dat,1)
local tab = {funcs[name](unpack(dat))}
table.insert(tab,1,id)
queueReturn:push(tab)
end
end
end):OnError(function(...)
error(...)
end)
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
jqc = jqc + 1
return c
end
--[[
MIT License
Copyright (c) 2022 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
-- TODO make compatible with lovr
local multi, thread = require("multi").init()
GLOBAL = multi.integration.GLOBAL
THREAD = multi.integration.THREAD
function multi:newSystemThreadedQueue(name)
local name = name or multi.randomString(16)
local c = {}
c.Name = name
local fRef = {"func",nil}
function c:init()
local q = {}
q.chan = love.thread.getChannel(self.Name)
function q:push(dat)
if type(dat) == "function" then
fRef[2] = THREAD.dump(dat)
self.chan:push(fRef)
return
else
self.chan:push(dat)
end
end
function q:pop()
local dat = self.chan:pop()
if type(dat)=="table" and dat[1]=="func" then
return THREAD.loadDump(dat[2])
else
return dat
end
end
function q:peek()
local dat = self.chan:peek()
if type(dat)=="table" and dat[1]=="func" then
return THREAD.loadDump(dat[2])
else
return dat
end
end
return q
end
THREAD.package(name,c)
return c
end
function multi:newSystemThreadedTable(name)
local name = name or multi.randomString(16)
local c = {}
c.Name = name
function c:init()
return THREAD.createTable(self.Name)
end
THREAD.package(name,c)
return c
end
local jqc = 1
function multi:newSystemThreadedJobQueue(n)
local c = {}
c.cores = n or THREAD.getCores()
c.registerQueue = {}
c.funcs = THREAD.createStaticTable("__JobQueue_"..jqc.."_table")
c.queue = love.thread.getChannel("__JobQueue_"..jqc.."_queue")
c.queueReturn = love.thread.getChannel("__JobQueue_"..jqc.."_queueReturn")
c.queueAll = love.thread.getChannel("__JobQueue_"..jqc.."_queueAll")
c.id = 0
c.OnJobCompleted = multi:newConnection()
local allfunc = 0
function c:doToAll(func)
local f = THREAD.dump(func)
for i = 1, self.cores do
self.queueAll:push({allfunc,f})
end
allfunc = allfunc + 1
end
function c:registerFunction(name,func)
if self.funcs[name] then
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 = {...}
link:Destroy()
end
end)
return thread.hold(function()
if rets then
return 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(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 function atomic(channel)
return channel:pop()
end
local clock = os.clock
local funcs = THREAD.createStaticTable("__JobQueue_"..jqc.."_table")
local queue = love.thread.getChannel("__JobQueue_"..jqc.."_queue")
local queueReturn = love.thread.getChannel("__JobQueue_"..jqc.."_queueReturn")
local lastProc = clock()
local queueAll = love.thread.getChannel("__JobQueue_"..jqc.."_queueAll")
local registry = {}
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()
THREAD.loadDump(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()
THREAD.loadDump(queueAll:pop()[2])()
end
local dat = queue:performAtomic(atomic)
if dat then
lastProc = os.clock()
local name = table.remove(dat,1)
local id = table.remove(dat,1)
local tab = {funcs[name](unpack(dat))}
table.insert(tab,1,id)
queueReturn:push(tab)
end
end
end):OnError(function(...)
error(...)
end)
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
jqc = jqc + 1
return c
end
function multi:newSystemThreadedConnection(name)
local name = name or multi.randomString(16)
local c = {}
c.CONN = 0x00
c.TRIG = 0x01
c.PING = 0x02
c.PONG = 0x03
local function remove(a, b)
local ai = {}
local r = {}
for k,v in pairs(a) do ai[v]=true end
for k,v in pairs(b) do
if ai[v]==nil then table.insert(r,a[k]) end
end
return r
end
c.CID = THREAD.getID()
c.subscribe = multi:newSystemThreadedQueue("SUB_STC_"..self.Name):init()
c.Name = name
c.links = {} -- All triggers sent from main connection. When a connection is triggered on another thread, they speak to the main then send stuff out.
-- Locals will only live in the thread that creates the original object
local ping
local pong = function(link, links)
local res = thread.hold(function()
return link:peek()[1] == c.PONG
end,{sleep=3})
if not res then
for i=1,#links do
if links[i] == link then
table.remove(links,i,link)
break
end
end
else
link:pop()
end
end
ping = thread:newFunction(function(self)
ping:Pause()
multi.ForEach(self.links, function(link) -- Sync new connections
link:push{self.PING}
multi:newThread("pong Thread", pong, link, links)
end)
thread.sleep(3)
ping:Resume()
end,false)
local function fire(...)
for _, link in pairs(c.links) do
link:push {c.TRIG, {...}}
end
end
thread:newThread("STC_SUB_MAN"..name,function()
local item
while true do
thread.yield()
-- We need to check on broken connections
ping(c) -- Should return instantlly and process this in another thread
item = thread.hold(function() -- This will keep things held up until there is something to process
return c.subscribe:pop()
end)
if item[1] == c.CONN then
multi.ForEach(c.links, function(link) -- Sync new connections
item[2]:push{c.CONN, link}
end)
c.links[#c.links+1] = item[2]
elseif item[1] == c.TRIG then
fire(unpack(item[2]))
c.proxy_conn:Fire(unpack(item[2]))
end
end
end)
--- ^^^ This will only exist in the init thread
function c:Fire(...)
local args = {...}
if self.CID == THREAD.getID() then -- Host Call
for _, link in pairs(self.links) do
link:push {self.TRIG, args}
end
self.proxy_conn:Fire(...)
else
self.subscribe:push {self.TRIG, args}
end
end
function c:init()
local multi, thread = require("multi"):init()
self.links = {}
self.proxy_conn = multi:newConnection()
local mt = getmetatable(self.proxy_conn)
setmetatable(self, {__index = self.proxy_conn, __call = function(t,func) self.proxy_conn(func) end, __add = mt.__add})
if self.CID == THREAD.getID() then return self end
thread:newThread("STC_CONN_MAN"..name,function()
local item
local link_self_ref = multi:newSystemThreadedQueue()
self.subscribe:push{self.CONN, link_self_ref}
while true do
item = thread.hold(function()
return link_self_ref:peek()
end)
if item[1] == self.PING then
link_self_ref:push{self.PONG}
link_self_ref:pop()
elseif item[1] == self.CONN then
if item[2].Name ~= link_self_ref.Name then
table.insert(self.links, item[2])
end
link_self_ref:pop()
elseif item[1] == self.TRIG then
self.proxy_conn:Fire(unpack(item[2]))
link_self_ref:pop()
else
-- This shouldn't be the case
end
end
end)
return self
end
THREAD.package(name,c)
return c
end
integration/loveManager/init.lua
-115
View File
@@ -1,115 +0,0 @@
--[[
MIT License
Copyright (c) 2022 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
if ISTHREAD then
error("You cannot require the loveManager from within a thread!")
end
local ThreadFileData = [[
ISTHREAD = true
THREAD = require("multi.integration.loveManager.threads") -- order is important!
sThread = THREAD
__IMPORTS = {...}
__FUNC__=table.remove(__IMPORTS,1)
__THREADID__=table.remove(__IMPORTS,1)
__THREADNAME__=table.remove(__IMPORTS,1)
stab = THREAD.createStaticTable(__THREADNAME__)
GLOBAL = THREAD.getGlobal()
multi, thread = require("multi").init()
stab["returns"] = {THREAD.loadDump(__FUNC__)(unpack(__IMPORTS))}
]]
local multi, thread = require("multi"):init()
local THREAD = {}
__THREADID__ = 0
__THREADNAME__ = "MainThread"
multi.integration={}
multi.integration.love2d={}
local THREAD = require("multi.integration.loveManager.threads")
local GLOBAL = THREAD.getGlobal()
local THREAD_ID = 1
local OBJECT_ID = 0
local stf = 0
function multi:newSystemThread(name,func,...)
local c = {}
c.name = name
c.ID=THREAD_ID
c.thread=love.thread.newThread(ThreadFileData)
c.thread:start(THREAD.dump(func),c.ID,c.name,...)
c.stab = THREAD.createStaticTable(name)
c.OnDeath = multi:newConnection()
c.OnError = multi:newConnection()
GLOBAL["__THREAD_"..c.ID] = {ID=c.ID, Name=c.name, Thread=c.thread}
GLOBAL["__THREAD_COUNT"] = THREAD_ID
THREAD_ID=THREAD_ID + 1
function c:getName()
return c.name
end
thread:newThread(function()
if name:find("TempSystemThread") then
local status_channel = love.thread.getChannel("__"..c.ID.."__MULTI__STATUS_CHANNEL__")
thread.hold(function()
-- While the thread is running we might as well do something in the loop
local status = status_channel
if status:peek()~=nil then
c.statusconnector:Fire(unpack(status:pop()))
end
return not c.thread:isRunning()
end)
else
thread.hold(function()
return not c.thread:isRunning()
end)
end
-- If the thread is not running let's handle that.
local thread_err = c.thread:getError()
if thread_err == "Thread Killed!\1" then
c.OnDeath:Fire("Thread Killed!")
elseif thread_err then
c.OnError:Fire(c,thread_err)
elseif c.stab.returns then
c.OnDeath:Fire(unpack(c.stab.returns))
c.stab.returns = nil
end
end)
return c
end
function THREAD:newFunction(func)
return thread:newFunctionBase(function(...)
return multi:newSystemThread("TempSystemThread"..THREAD_ID,func,...)
end)()
end
THREAD.newSystemThread = multi.newSystemThread
function love.threaderror(thread, errorstr)
mulit.print("Thread error!\n"..errorstr)
end
multi.integration.GLOBAL = GLOBAL
multi.integration.THREAD = THREAD
require("multi.integration.loveManager.extensions")
multi.print("Integrated Love Threading!")
return {init=function()
return GLOBAL,THREAD
end}
integration/loveManager/threads.lua
-250
View File
@@ -1,250 +0,0 @@
--[[
MIT License
Copyright (c) 2022 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
require("love.timer")
require("love.system")
require("love.data")
require("love.thread")
local socket = require("socket")
local multi, thread = require("multi").init()
local threads = {}
function threads.loadDump(d)
return loadstring(d:getString())
end
function threads.dump(func)
return love.data.newByteData(string.dump(func))
end
local fRef = {"func",nil}
local function manage(channel, value)
channel:clear()
if type(value) == "function" then
fRef[2] = THREAD.dump(value)
channel:push(fRef)
return
else
channel:push(value)
end
end
local function RandomVariable(length)
local res = {}
math.randomseed(socket.gettime()*10000)
for i = 1, length do
res[#res+1] = string.char(math.random(97, 122))
end
return table.concat(res)
end
local GNAME = "__GLOBAL_"
local proxy = {}
function threads.set(name,val)
if not proxy[name] then proxy[name] = love.thread.getChannel(GNAME..name) end
proxy[name]:performAtomic(manage, val)
end
function threads.get(name)
if not proxy[name] then proxy[name] = love.thread.getChannel(GNAME..name) end
local dat = proxy[name]:peek()
if type(dat)=="table" and dat[1]=="func" then
return THREAD.loadDump(dat[2])
else
return dat
end
end
function threads.waitFor(name)
if thread.isThread() then
return thread.hold(function()
return threads.get(name)
end)
end
while threads.get(name)==nil do
love.timer.sleep(.001)
end
local dat = threads.get(name)
if type(dat) == "table" and dat.init then
dat.init = threads.loadDump(dat.init)
end
return dat
end
function threads.package(name,val)
local init = val.init
val.init=threads.dump(val.init)
GLOBAL[name]=val
val.init=init
end
function threads.getCores()
return love.system.getProcessorCount()
end
function threads.kill()
error("Thread Killed!\1")
end
function threads.pushStatus(...)
local status_channel = love.thread.getChannel("__"..__THREADID__.."__MULTI__STATUS_CHANNEL__")
local args = {...}
status_channel:push(__THREADID__, args)
end
function threads.getThreads()
local t = {}
for i=1,GLOBAL["__THREAD_COUNT"] do
t[#t+1]=GLOBAL["__THREAD_"..i]
end
return t
end
function threads.getThread(n)
return GLOBAL["__THREAD_"..n]
end
function threads.getName()
return __THREADNAME__
end
function threads.getID()
return __THREADID__
end
function threads.sleep(n)
love.timer.sleep(n)
end
function threads.getGlobal()
return setmetatable({},
{
__index = function(t, k)
return THREAD.get(k)
end,
__newindex = function(t, k, v)
THREAD.set(k,v)
end
}
)
end
function threads.createTable(n)
local _proxy = {}
local function set(name,val)
if not _proxy[name] then _proxy[name] = love.thread.getChannel(n..name) end
_proxy[name]:performAtomic(manage, val)
end
local function get(name)
if not _proxy[name] then _proxy[name] = love.thread.getChannel(n..name) end
local dat = _proxy[name]:peek()
if type(dat)=="table" and dat[1]=="func" then
return THREAD.loadDump(dat[2])
else
return dat
end
end
return setmetatable({},
{
__index = function(t, k)
return get(k)
end,
__newindex = function(t, k, v)
set(k,v)
end
}
)
end
function threads.getConsole()
local c = {}
c.queue = love.thread.getChannel("__CONSOLE__")
function c.print(...)
c.queue:push{...}
end
function c.error(err)
c.queue:push{"ERROR in <"..__THREADNAME__..">: "..err,__THREADID__}
error(err)
end
return c
end
if not ISTHREAD then
local clock = os.clock
local lastproc = clock()
local queue = love.thread.getChannel("__CONSOLE__")
thread:newThread("consoleManager",function()
while true do
thread.yield()
dat = queue:pop()
if dat then
lastproc = clock()
print(unpack(dat))
end
if clock()-lastproc>2 then
thread.sleep(.1)
end
end
end)
end
function threads.createStaticTable(n)
local __proxy = {}
local function set(name,val)
if __proxy[name] then return end
local chan = love.thread.getChannel(n..name)
if chan:getCount()>0 then return end
chan:performAtomic(manage, val)
__proxy[name] = val
end
local function get(name)
if __proxy[name] then return __proxy[name] end
local dat = love.thread.getChannel(n..name):peek()
if type(dat)=="table" and dat[1]=="func" then
__proxy[name] = THREAD.loadDump(dat[2])
return __proxy[name]
else
__proxy[name] = dat
return __proxy[name]
end
end
return setmetatable({},
{
__index = function(t, k)
return get(k)
end,
__newindex = function(t, k, v)
set(k,v)
end
}
)
end
function threads.hold(n)
local dat
while not(dat) do
dat = n()
end
end
return threads
integration/lovrManager/extensions.lua
-205
View File
@@ -1,205 +0,0 @@
--[[
MIT License
Copyright (c) 2022 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
local multi, thread = require("multi").init()
GLOBAL = multi.integration.GLOBAL
THREAD = multi.integration.THREAD
function multi:newSystemThreadedQueue(name)
local name = name or multi.randomString(16)
local c = {}
c.Name = name
local fRef = {"func",nil}
function c:init()
local q = {}
q.chan = lovr.thread.getChannel(self.Name)
function q:push(dat)
if type(dat) == "function" then
fRef[2] = THREAD.dump(dat)
self.chan:push(fRef)
return
else
self.chan:push(dat)
end
end
function q:pop()
local dat = self.chan:pop()
if type(dat)=="table" and dat[1]=="func" then
return THREAD.loadDump(dat[2])
else
return dat
end
end
function q:peek()
local dat = self.chan:peek()
if type(dat)=="table" and dat[1]=="func" then
return THREAD.loadDump(dat[2])
else
return dat
end
end
return q
end
THREAD.package(name,c)
return c
end
function multi:newSystemThreadedTable(name)
local name = name or multi.randomString(16)
local c = {}
c.Name = name
function c:init()
return THREAD.createTable(self.Name)
end
THREAD.package(name,c)
return c
end
local jqc = 1
function multi:newSystemThreadedJobQueue(n)
local c = {}
c.cores = n or THREAD.getCores()
c.registerQueue = {}
c.funcs = THREAD.createStaticTable("__JobQueue_"..jqc.."_table")
c.queue = lovr.thread.getChannel("__JobQueue_"..jqc.."_queue")
c.queueReturn = lovr.thread.getChannel("__JobQueue_"..jqc.."_queueReturn")
c.queueAll = lovr.thread.getChannel("__JobQueue_"..jqc.."_queueAll")
c.id = 0
c.OnJobCompleted = multi:newConnection()
local allfunc = 0
function c:doToAll(func)
local f = THREAD.dump(func)
for i = 1, self.cores do
self.queueAll:push({allfunc,f})
end
allfunc = allfunc + 1
end
function c:registerFunction(name,func)
if self.funcs[name] then
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 = {...}
link:Destroy()
end
end)
return thread.hold(function()
if rets then
return 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(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("lovr.timer")
local function atomic(channel)
return channel:pop()
end
local clock = os.clock
local funcs = THREAD.createStaticTable("__JobQueue_"..jqc.."_table")
local queue = lovr.thread.getChannel("__JobQueue_"..jqc.."_queue")
local queueReturn = lovr.thread.getChannel("__JobQueue_"..jqc.."_queueReturn")
local lastProc = clock()
local queueAll = lovr.thread.getChannel("__JobQueue_"..jqc.."_queueAll")
local registry = {}
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()
THREAD.loadDump(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()
THREAD.loadDump(queueAll:pop()[2])()
end
local dat = queue:performAtomic(atomic)
if dat then
lastProc = os.clock()
local name = table.remove(dat,1)
local id = table.remove(dat,1)
local tab = {funcs[name](unpack(dat))}
table.insert(tab,1,id)
queueReturn:push(tab)
end
end
end):OnError(function(...)
error(...)
end)
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
jqc = jqc + 1
return c
end
integration/luvitManager.lua
-138
View File
@@ -1,138 +0,0 @@
--[[
MIT License
Copyright (c) 2022 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
-- This module probably will not be maintained any longer!
package.path = "?/init.lua;?.lua;" .. package.path
local function _INIT(luvitThread, timer)
-- lots of this stuff should be able to stay the same
function os.getOS()
if package.config:sub(1, 1) == "\\" then
return "windows"
else
return "unix"
end
end
-- Step 1 get setup threads on luvit... Sigh how do i even...
local multi, thread = require("multi").init()
isMainThread = true
function multi:canSystemThread()
return true
end
function multi:getPlatform()
return "luvit"
end
local multi = multi
-- Step 2 set up the Global table... is this possible?
local GLOBAL = {}
setmetatable(
GLOBAL,
{
__index = function(t, k)
--print("No Global table when using luvit integration!")
return nil
end,
__newindex = function(t, k, v)
--print("No Global table when using luvit integration!")
end
}
)
local THREAD = {}
function THREAD.set(name, val)
--print("No Global table when using luvit integration!")
end
function THREAD.get(name)
--print("No Global table when using luvit integration!")
end
local function randomString(n)
local str = ""
local strings = {"a","b","c","d","e","f","g","h","i","j","k","l","m","n","o","p","q","r","s","t","u","v","w","x","y","z","1","2","3","4","5","6","7","8","9","0","A","B","C","D","E","F","G","H","I","J","K","L","M","N","O","P","Q","R","S","T","U","V","W","X","Y","Z"}
for i = 1, n do
str = str .. "" .. strings[math.random(1, #strings)]
end
return str
end
function THREAD.waitFor(name)
--print("No Global table when using luvit integration!")
end
function THREAD.testFor(name, val, sym)
--print("No Global table when using luvit integration!")
end
function THREAD.getCores()
return THREAD.__CORES
end
if os.getOS() == "windows" then
THREAD.__CORES = tonumber(os.getenv("NUMBER_OF_PROCESSORS"))
else
THREAD.__CORES = tonumber(io.popen("nproc --all"):read("*n"))
end
function THREAD.kill() -- trigger the thread destruction
error("Thread was Killed!")
end
-- hmmm if im cleaver I can get this to work... but since data passing isn't going to be a thing its probably not important
function THREAD.sleep(n)
--print("No Global table when using luvit integration!")
end
function THREAD.hold(n)
--print("No Global table when using luvit integration!")
end
-- Step 5 Basic Threads!
local function entry(path, name, func, ...)
local timer = require "timer"
local luvitThread = require "thread"
package.path = path
loadstring(func)(...)
end
function multi:newSystemThread(name, func, ...)
local c = {}
local __self = c
c.name = name
c.Type = "sthread"
c.thread = {}
c.func = string.dump(func)
function c:kill()
-- print("No Global table when using luvit integration!")
end
luvitThread.start(entry, package.path, name, c.func, ...)
return c
end
THREAD.newSystemThread = multi.newSystemThread
multi.print("Integrated Luvit!")
multi.integration = {} -- for module creators
multi.integration.GLOBAL = GLOBAL
multi.integration.THREAD = THREAD
require("multi.integration.shared")
-- Start the main mainloop... This allows you to process your multi objects, but the engine on the main thread will be limited to .001 or 1 millisecond sigh...
local interval =
timer.setInterval(
1,
function()
multi:uManager()
end
)
return multi
end
return {init = function(threadHandle, timerHandle)
local multi = _INIT(threadHandle, timerHandle)
return GLOBAL, THREAD
end}
integration/networkManager/init.lua
-57
View File
@@ -1,57 +0,0 @@
--[[
MIT License
Copyright (c) 2022 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
local multi, thread = require("multi"):init()
local net = require("net")
--local bin = require("bin")
local char = string.char
local byte = string.byte
bin.setBitsInterface(infinabits)
--[[
--[=[ Pre reqs:
- Network contains nodes
- Network can broadcast/has nodemanager/ is simple and can be scanned
Outline:
- multi:newMasterNode(connectionDetails)
-- master:setDefaultNode(nodeName) -- Set default node
-- master:newNetworkThread(nodeName,func,...) -- Thread is ran on a random node or the default one if set if nodeName is set to nil
-- master:newNetworkChannel(nodeName)
-- master:sendTo(nodeName,data)
- multi:newNode(connectionDetails)
- multi:newNodeManager(connectionDetails) -- This will be incharge of a lot of data handling
]=]
local nGLOBAL, nTHREAD = require("multi.integration.networkManager"):init()
local master = multi:newMasterNode()
master:newNetworkThread("simpleNode",function(a,b,c)
print(a,b,c)
end,1,2,3)
]]
-- The init file should provide the structure that all the other modules build off of
return {
init = function()
--
end
}
integration/pesudoManager/extensions.lua
-141
View File
@@ -1,141 +0,0 @@
--[[
MIT License
Copyright (c) 2022 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
local multi, thread = require("multi"):init()
local GLOBAL, THREAD = multi.integration.GLOBAL,multi.integration.THREAD
local function stripUpValues(func)
local dmp = string.dump(func)
if setfenv then
return loadstring(dmp,"IsolatedThread_PesudoThreading")
else
return load(dmp,"IsolatedThread_PesudoThreading","bt")
end
end
function multi:newSystemThreadedQueue(name)
local c = {}
function c:push(v)
table.insert(self,v)
end
function c:pop()
return table.remove(self,1)
end
function c:peek()
return self[1]
end
function c:init()
return self
end
GLOBAL[name or "_"] = c
return c
end
function multi:newSystemThreadedTable(name)
local c = {}
function c:init()
return self
end
GLOBAL[name or "_"] = c
return c
end
local setfenv = setfenv
if not setfenv then
if not debug then
multi.print("Unable to implement setfenv in lua 5.2+ the debug module is not available!")
else
setfenv = function(f, env)
return load(string.dump(f), nil, nil, env)
end
end
end
function multi:newSystemThreadedJobQueue(n)
local c = {}
c.cores = n or THREAD.getCores()*2
c.OnJobCompleted = multi:newConnection()
local jobs = {}
local ID=1
local jid = 1
local env = {}
setmetatable(env,{
__index = _G
})
local funcs = {}
function c:doToAll(func)
setfenv(func,env)()
return self
end
function c:registerFunction(name,func)
funcs[name] = setfenv(func,env)
return self
end
function c:pushJob(name,...)
table.insert(jobs,{name,jid,{...}})
jid = jid + 1
return jid-1
end
function c:isEmpty()
print(#jobs)
return #jobs == 0
end
local nFunc = 0
function c:newFunction(name,func,holup) -- This registers with the queue
local func = stripUpValues(func)
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 = {...}
link:Destroy()
end
end)
return thread.hold(function()
if rets then
return unpack(rets) or multi.NIL
end
end)
end,holup),name
end
for i=1,c.cores do
thread:newthread("PesudoThreadedJobQueue_"..i,function()
while true do
thread.yield()
if #jobs>0 then
local j = table.remove(jobs,1)
c.OnJobCompleted:Fire(j[2],funcs[j[1]](unpack(j[3])))
else
thread.sleep(.05)
end
end
end)
end
return c
end
integration/pesudoManager/init.lua
-101
View File
@@ -1,101 +0,0 @@
--[[
MIT License
Copyright (c) 2020 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
package.path = "?/init.lua;?.lua;" .. package.path
local multi, thread = require("multi"):init()
if multi.integration then
return {
init = function()
return multi.integration.GLOBAL, multi.integration.THREAD
end
}
end
local GLOBAL, THREAD = require("multi.integration.pesudoManager.threads").init(thread)
function multi:canSystemThread() -- We are emulating system threading
return true
end
function multi:getPlatform()
return "pesudo"
end
local function split(str)
local tab = {}
for word in string.gmatch(str, '([^,]+)') do
table.insert(tab,word)
end
return tab
end
local tab = [[_VERSION,io,os,require,load,debug,assert,collectgarbage,error,getfenv,getmetatable,ipairs,loadstring,module,next,pairs,pcall,print,rawequal,rawget,rawset,select,setfenv,setmetatable,tonumber,tostring,type,unpack,xpcall,math,coroutine,string,table]]
tab = split(tab)
local id = 0
function multi:newSystemThread(name,func,...)
GLOBAL["$THREAD_NAME"] = name
GLOBAL["$__THREADNAME__"] = name
GLOBAL["$THREAD_ID"] = id
GLOBAL["$thread"] = thread
local env = {
GLOBAL = GLOBAL,
THREAD = THREAD,
THREAD_NAME = name,
__THREADNAME__ = name,
THREAD_ID = id,
thread = thread
}
for i = 1,#tab do
env[tab[i]] = _G[tab[i]]
end
local th = thread:newISOThread(name,func,env,...)
id = id + 1
return th
end
THREAD.newSystemThread = multi.newSystemThread
-- System threads as implemented here cannot share memory, but use a message passing system.
-- An isolated thread allows us to mimic that behavior so if access data from the "main" thread happens things will not work. This behavior is in line with how the system threading works
function THREAD:newFunction(func,holdme)
return thread:newFunctionBase(function(...)
return multi:newSystemThread("TempSystemThread",func,...)
end,holdme)()
end
multi.print("Integrated Pesudo Threading!")
multi.integration = {} -- for module creators
multi.integration.GLOBAL = GLOBAL
multi.integration.THREAD = THREAD
require("multi.integration.pesudoManager.extensions")
return {
init = function()
return GLOBAL, THREAD
end
}
integration/pesudoManager/threads.lua
-110
View File
@@ -1,110 +0,0 @@
--[[
MIT License
Copyright (c) 2022 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
local function getOS()
if package.config:sub(1, 1) == "\\" then
return "windows"
else
return "unix"
end
end
local function INIT(thread)
local THREAD = {}
local GLOBAL = {}
THREAD.Priority_Core = 3
THREAD.Priority_High = 2
THREAD.Priority_Above_Normal = 1
THREAD.Priority_Normal = 0
THREAD.Priority_Below_Normal = -1
THREAD.Priority_Low = -2
THREAD.Priority_Idle = -3
function THREAD.set(name, val)
GLOBAL[name] = val
end
function THREAD.get(name)
return GLOBAL[name]
end
function THREAD.waitFor(name)
return thread.hold(function() return GLOBAL[name] end)
end
if getOS() == "windows" then
THREAD.__CORES = tonumber(os.getenv("NUMBER_OF_PROCESSORS"))
else
THREAD.__CORES = tonumber(io.popen("nproc --all"):read("*n"))
end
function THREAD.getCores()
return THREAD.__CORES
end
function THREAD.getConsole()
local c = {}
function c.print(...)
print(...)
end
function c.error(err)
error("ERROR in <"..GLOBAL["$__THREADNAME__"]..">: "..err)
end
return c
end
function THREAD.getThreads()
return {}--GLOBAL.__THREADS__
end
THREAD.pushStatus = thread.pushStatus
if os.getOS() == "windows" then
THREAD.__CORES = tonumber(os.getenv("NUMBER_OF_PROCESSORS"))
else
THREAD.__CORES = tonumber(io.popen("nproc --all"):read("*n"))
end
function THREAD.kill()
error("Thread was killed!")
end
function THREAD.getName()
return GLOBAL["$THREAD_NAME"]
end
function THREAD.getID()
return GLOBAL["$THREAD_ID"]
end
THREAD.sleep = thread.sleep
THREAD.hold = thread.hold
return GLOBAL, THREAD
end
return {init = function(thread)
return INIT(thread)
end}
integration/threading.lua
-13
View File
@@ -1,13 +0,0 @@
-- We need to detect what enviroment we are running our code in.
return {
init = function()
if love then
return require("multi.integration.loveManager"):init()
else
if pcall(require,"lanes") then
return require("multi.integration.lanesManager"):init()
end
return require("multi.integration.pesudoManager"):init()
end
end
}