This commit is contained in:
2022-11-05 22:21:19 -04:00
parent f240588edd
commit a5b59ee601
33 changed files with 1 additions and 50 deletions
+312
View File
@@ -0,0 +1,312 @@
--[[
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, self.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
local sub_func = function() -- This will keep things held up until there is something to process
return c.subscribe:pop()
end
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(sub_func)
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
+180
View File
@@ -0,0 +1,180 @@
--[[
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
local multi_settings = multi.defaultSettings
for i,v in pairs(multi_settings) do
print(i,v)
end
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"):init(multi_settings)
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
}
+142
View File
@@ -0,0 +1,142 @@
--[[
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}