epicknex/multi
1
0
Fork 0
Code Issues 5 Pull Requests 1 Actions Packages Projects Releases 13 Wiki Activity

Threading working in love2d

Browse Source
This commit is contained in:
Ryan Ward 2024-02-24 23:05:05 -05:00
parent 42b4e0fcf4
commit e3ac0951ab
18 changed files with 210 additions and 999 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
init.lua
View File

@ -2152,8 +2152,7 @@ local init = false
multi.settingsHook = multi:newConnection()
function multi.init(settings, realsettings)
if settings == multi then settings = realsettings end
if init then return _G["$multi"].multi,_G["$multi"].thread end
init = true
if type(settings)=="table" then
multi.defaultSettings = settings
@ -2164,6 +2163,8 @@ function multi.init(settings, realsettings)
multi.mainloop = multi.mainloopRef
end
if not init then
if settings.findopt then
find_optimization = true
doOpt()
@ -2176,6 +2177,8 @@ function multi.init(settings, realsettings)
multi.settingsHook:Fire(settings)
end
end
init = true
return _G["$multi"].multi,_G["$multi"].thread
end

19
integration/lanesManager/init.lua
View File

@ -85,7 +85,8 @@ function multi:newSystemThread(name, func, ...)
THREAD_ID = count,
THREAD = THREAD,
GLOBAL = GLOBAL,
_Console = __ConsoleLinda
_Console = __ConsoleLinda,
_DEFER = {}
}
if GLOBAL["__env"] then
for i,v in pairs(GLOBAL["__env"]) do
@ -97,26 +98,16 @@ function multi:newSystemThread(name, func, ...)
globals = globe,
priority = c.priority
},function(...)
local profi
if multi_settings.debug then
profi = require("proFI")
profi:start()
end
multi, thread = require("multi"):init(multi_settings)
require("multi.integration.lanesManager.extensions")
require("multi.integration.sharedExtensions")
local has_error = true
returns = {pcall(func, ...)}
return_linda:set("returns", returns)
has_error = false
if profi then
multi.OnExit(function(...)
profi:stop()
profi:writeReport("Profiling Report [".. THREAD_NAME .."].txt")
end)
for i,v in pairs(_DEFER) do
pcall(v)
end
has_error = false
end)(...)
count = count + 1
function c:getName()

12
integration/lanesManager/threads.lua
View File

@ -89,6 +89,10 @@ local function INIT(__GlobalLinda, __SleepingLinda, __StatusLinda, __Console)
error("Thread was killed!\1")
end
function THREAD.sync()
-- Maybe do something...
end
function THREAD.pushStatus(...)
local args = multi.pack(...)
__StatusLinda:send(nil,THREAD_ID, args)
@ -138,13 +142,7 @@ local function INIT(__GlobalLinda, __SleepingLinda, __StatusLinda, __Console)
end
function THREAD.defer(func)
local m = {onexit = function() func() end}
if _VERSION >= "Lua 5.2" then
setmetatable(m, {__gc = m.onexit})
else
m.sentinel = newproxy(true)
getmetatable(m.sentinel).__gc = m.onexit
end
table.insert(_DEFER, func)
end
return GLOBAL, THREAD

5
integration/loveManager/extensions.lua
View File

@ -11,7 +11,7 @@ function multi:newSystemThreadedQueue(name)
c.Name = name
c.Type = multi.registerType("s_queue")
c.chan = love.thread.newChannel()
c.chan = love.thread.getChannel(name)
function c:push(dat)
self.chan:push(THREAD.packValue(dat))
@ -26,10 +26,12 @@ function multi:newSystemThreadedQueue(name)
end
function c:init()
self.chan = love.thread.getChannel(self.Name)
return self
end
function c:Hold(opt)
local multi, thread = require("multi"):init()
if opt.peek then
return thread.hold(function()
return self:peek()
@ -73,6 +75,7 @@ function multi:newSystemThreadedTable(name)
c.__init = c.init
function c:Hold(opt)
local multi, thread = require("multi"):init()
if opt.key then
return thread.hold(function()
return self.tab[opt.key]

24
integration/loveManager/init.lua
View File

@ -5,11 +5,13 @@ end
local ThreadFileData = [[
ISTHREAD = true
args = {...}
THREAD_ID = table.remove(args, 1)
THREAD_NAME = table.remove(args, 1)
GLOBAL, THREAD = require("multi.integration.loveManager.threads"):init()
__FUNC = THREAD.unpackValue(table.remove(args, 1))
ARGS = THREAD.unpackValue(table.remove(args, 1))
THREAD_ID = args[1]
THREAD_NAME = args[2]
GLOBAL, THREAD, DEFER = require("multi.integration.loveManager.threads"):init()
__FUNC = THREAD.unpackValue(args[3])
ARGS = THREAD.unpackValue(args[4])
settings = args[5]
if ARGS == nil then ARGS = {} end
math.randomseed(THREAD_ID)
math.random()
math.random()
@ -21,10 +23,16 @@ if GLOBAL["__env"] then
_G[i] = v
end
end
multi, thread = require("multi"):init()
multi, thread = require("multi"):init{error=true, warning=true, print=true, priority=true}
multi.defaultSettings.print = true
require("multi.integration.loveManager.extensions")
require("multi.integration.sharedExtensions")
stab["returns"] = {__FUNC(multi.unpack(ARGS))}
local returns = {pcall(__FUNC, multi.unpack(ARGS))}
table.remove(returns,1)
stab["returns"] = returns
for i,v in pairs(DEFER) do
pcall(v)
end
]]
_G.THREAD_NAME = "MAIN_THREAD"
@ -49,7 +57,7 @@ function multi:newSystemThread(name, func, ...)
c.Name = name
c.ID = tid
c.thread = love.thread.newThread(ThreadFileData)
c.thread:start(c.ID, c.Name, THREAD.packValue(func), THREAD.packValue({...}))
c.thread:start(c.ID, c.Name, THREAD.packValue(func), THREAD.packValue({...}), multi.defaultSettings)
c.stab = THREAD.createTable(name .. c.ID)
c.creationTime = os.clock()
c.OnDeath = multi:newConnection()

94
integration/loveManager/threads.lua
View File

@ -25,19 +25,76 @@ require("love.timer")
require("love.system")
require("love.data")
require("love.thread")
local utils = require("multi.integration.loveManager.utils")
local multi, thread = require("multi"):init()
local NIL = love.data.newByteData("\3")
-- If a non table/function is supplied we just return it
local function packValue(t)
return utils.pack(t)
-- Checks if the given value is a LOVE2D object (i.e. has metatable with __index field) and if that __index field contains functions typical of LOVE2D objects
function isLoveObject(value)
-- Check if the value has metatable
if type(value) == "userdata" and getmetatable(value) then
-- Check if the metatable has the __index field
local index = getmetatable(value).__index
if type(index) == "table" then
-- Check if the metatable's __index table contains functions typical of LOVE2D objects
if index.draw or index.update or index.getWidth or index.getHeight or index.getString or index.getPointer then
return true
end
end
end
return false
end
-- If a non table/function is supplied we just return it
local function unpackValue(d)
return utils.unpack(d)
-- Converts any function values in a table to a string with the value "\1\2:func:<function_string>" where <function_string> is the Lua stringified version of the function
function tableToFunctionString(t)
if type(t) == "nil" then return "\1\2:nil:" end
if type(t) == "function" then return "\1\2:func:"..string.dump(t) end
if type(t) ~= "table" then return t end
local newtable = {}
for k, v in pairs(t) do
if type(v) == "function" then
newtable[k] = "\1\2:func:"..string.dump(v)
elseif type(v) == "table" then
newtable[k] = tableToFunctionString(v)
elseif isLoveObject(v) then
newtable[k] = v
elseif type(v) == "userdata" then
newtable[k] = tostring(v)
else
newtable[k] = v
end
end
return newtable
end
-- Converts strings with the value "\1\2:func:<function_string>" back to functions
function functionStringToTable(t)
if type(t) == "string" and t:sub(1, 8) == "\1\2:func:" then return loadstring(t:sub(9, -1)) end
if type(t) == "string" and t:sub(1, 7) == "\1\2:nil:" then return nil end
if type(t) ~= "table" then return t end
for k, v in pairs(t) do
if type(v) == "string" then
if v:sub(1, 8) == "\1\2:func:" then
t[k] = loadstring(v:sub(9, -1))
else
t[k] = v
end
elseif type(v) == "table" then
t[k] = functionStringToTable(v)
else
t[k] = v
end
end
if t.init then
t:init()
end
return t
end
local function packValue(t)
return tableToFunctionString(t)
end
local function unpackValue(t)
return functionStringToTable(t)
end
local function createTable(n)
@ -53,6 +110,11 @@ local function createTable(n)
end
local function get(name)
return unpackValue(love.thread.getChannel(n .. name):peek())
-- if type(data) == "table" and data.init then
-- return data:init()
-- else
-- return data
-- end
end
return setmetatable({},
{
@ -67,7 +129,7 @@ local function createTable(n)
end
function INIT()
local GLOBAL, THREAD = createTable("__GLOBAL__"), {}
local GLOBAL, THREAD, DEFER = createTable("__GLOBAL__"), {}, {}
local status_channel, console_channel = love.thread.getChannel("__status_channel__" .. THREAD_ID),
love.thread.getChannel("__console_channel__")
@ -92,11 +154,7 @@ function INIT()
repeat
wait()
until GLOBAL[name] ~= nil
if type(GLOBAL[name].__init) == "function" then
return GLOBAL[name]:__init()
else
return GLOBAL[name]
end
end, true)
function THREAD.getCores()
@ -153,10 +211,14 @@ function INIT()
end
function THREAD.defer(func)
multi.OnExit(func)
table.insert(DEFER, func)
end
return GLOBAL, THREAD
function THREAD.sync()
-- Maybe do something...
end
return GLOBAL, THREAD, DEFER
end
return {

58
integration/loveManager/utils.lua
View File

@ -1,58 +0,0 @@
require("love.data")
local utils = {}
local NIL = {Type="nil"}
--love.data.newByteData("\2"..serpent.dump({t,true},{safe = true}))
local ltype = function(v) return v:type() end
local t = function(value)
local v = type(value)
if v == "userdata" then
local status, return_or_err = pcall(ltype, value)
if status then return return_or_err else return "userdata" end
else return v end
end
function utils.pack(tbl, seen)
if type(tbl) == "function" then return {["__$FUNC$__"] = love.data.newByteData(string.dump(tbl))} end
if type(tbl) ~= "table" then return tbl end
local seen = seen or {}
local result = {}
result.__isPacked = true
for i,v in pairs(tbl) do
if seen[v] then
result[i] = v
elseif t(v) == "table" then
seen[v] = true
result[i] = utils.pack(v, seen)
elseif t(v) == "function" then
result["$F"..i] = love.data.newByteData(string.dump(v))
elseif t{v} == "userdata" then
result[i] = "userdata"
else -- Handle what we need to and pass the rest along as a value
result[i] = v
end
end
return result
end
function utils.unpack(tbl)
if not tbl then return nil end
if type(tbl) ~= "table" then return tbl end
if tbl["__$FUNC$__"] then return loadstring(tbl["__$FUNC$__"]:getString()) end
for i,v in pairs(tbl) do
if type(i) == "string" and i:sub(1,2) == "$F" then
local rawfunc = v:getString()
v:release()
tbl[i] = nil
tbl[i:sub(3,-1)] = loadstring(rawfunc)
end
if type(v) == "table" then
utils.unpack(v)
end
end
tbl.__isPacked = nil
return tbl
end
return utils

389
integration/loveManagerold/extensions.lua
View File

@ -1,389 +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 not ISTHREAD then
multi, thread = require("multi").init()
GLOBAL = multi.integration.GLOBAL
THREAD = multi.integration.THREAD
else
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
local fRef = {"\2",nil}
function c:init()
local q = {}
q.chan = love.thread.getChannel(self.Name)
function q:push(dat)
if type(dat) == "table" then
self.chan:push{"DATA",THREAD.packTable(dat)}
else
self.chan:push(dat)
end
-- 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]=="DATA" then
return THREAD.unpackTable(dat[2])--THREAD.loadDump(dat[2])
else
return dat
end
end
function q:peek()
local dat = self.chan:peek()
if type(dat)=="table" and dat[1]=="DATA" then
return THREAD.unpackTable(dat[2])--THREAD.loadDump(dat[2])
else
return dat
end
end
return q
end
THREAD.package(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
function c:init()
return THREAD.createTable(self.Name)
end
THREAD.package(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.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 = 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 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 = {}
_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()
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
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(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
self:create(c)
return c
end
function multi:newSystemThreadedConnection(name)
local name = name or multi.randomString(16)
local c = {}
c.Type = multi.SCONNECTION
c.CONN = 0x00
c.TRIG = 0x01
c.PING = 0x02
c.PONG = 0x03
local subscribe = love.thread.getChannel("SUB_STC_" .. name)
function c:init()
self.subscribe = love.thread.getChannel("SUB_STC_" .. self.Name)
function self:Fire(...)
local args = multi.pack(...)
if self.CID == THREAD_ID then -- Host Call
for _, link in pairs(self.links) do
love.thread.getChannel(link):push{self.TRIG, args}
end
self.proxy_conn:Fire(...)
else
self.subscribe:push{self.TRIG, args}
end
end
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_ID then return self end
thread:newThread("STC_CONN_MAN" .. self.Name,function()
local item
local string_self_ref = "LSF_" .. multi.randomString(16)
local link_self_ref = love.thread.getChannel(string_self_ref)
self.subscribe:push{self.CONN, string_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 string_self_ref ~= item[2] then
table.insert(self.links, love.thread.getChannel(item[2]))
end
link_self_ref:pop()
elseif item[1] == self.TRIG then
self.proxy_conn:Fire(multi.unpack(item[2]))
link_self_ref:pop()
else
-- This shouldn't be the case
end
end
end)
return self
end
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_ID
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 love.thread.getChannel(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
love.thread.getChannel(link):pop()
end
end
ping = thread:newFunction(function(self)
ping:Pause()
multi.ForEach(self.links, function(link) -- Sync new connections
love.thread.getChannel(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
love.thread.getChannel(link):push {c.TRIG, multi.pack(...)}
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
love.thread.getChannel(item[2]):push{c.CONN, link}
end)
c.links[#c.links+1] = item[2]
elseif item[1] == c.TRIG then
fire(multi.unpack(item[2]))
c.proxy_conn:Fire(multi.unpack(item[2]))
end
end
end)
--- ^^^ This will only exist in the init thread
THREAD.package(name,c)
self:create(c)
return c
end
require("multi.integration.sharedExtensions")

138
integration/loveManagerold/init.lua
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.
]]
if ISTHREAD then
error("You cannot require the loveManager from within a thread!")
end
local ThreadFileData = [[
ISTHREAD = true
THREAD = require("multi.integration.loveManager.threads")
sThread = THREAD
__IMPORTS = {...}
__FUNC__=table.remove(__IMPORTS,1)
THREAD_ID=table.remove(__IMPORTS,1)
THREAD_NAME=table.remove(__IMPORTS,1)
math.randomseed(THREAD_ID)
math.random()
math.random()
math.random()
stab = THREAD.createStaticTable(THREAD_NAME .. THREAD_ID)
GLOBAL = THREAD.getGlobal()
if GLOBAL["__env"] then
local env = THREAD.unpackENV(GLOBAL["__env"])
for i,v in pairs(env) do
_G[i] = v
end
end
multi, thread = require("multi").init()
multi.integration={}
multi.integration.GLOBAL = GLOBAL
multi.integration.THREAD = THREAD
pcall(require,"multi.integration.loveManager.extensions")
pcall(require,"multi.integration.sharedExtensions")
stab["returns"] = {THREAD.loadDump(__FUNC__)(multi.unpack(__IMPORTS))}
]]
local multi, thread = require("multi"):init()
local THREAD = {}
_G.THREAD_NAME = "MAIN_THREAD"
_G.THREAD_ID = 0
multi.integration = {}
local THREAD = require("multi.integration.loveManager.threads")
local GLOBAL = THREAD.getGlobal()
multi.isMainThread = true
function multi:newSystemThread(name, func, ...)
THREAD_ID = THREAD_ID + 1
local c = {}
c.Type = multi.STHREAD
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.ID)
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
function c:getName() return c.name end
thread:newThread(name .. "_System_Thread_Handler",function()
if name == "SystemThreaded Function Handler" then
local status_channel = love.thread.getChannel("STATCHAN_" .. c.ID)
thread.hold(function()
-- While the thread is running we might as well do something in the loop
if status_channel:peek() ~= nil then
c.statusconnector:Fire(multi.unpack(status_channel: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(multi.unpack(c.stab.returns))
c.stab.returns = nil
end
end)
c.OnError(multi.error)
if self.isActor then
self:create(c)
else
multi.create(multi, c)
end
return c
end
function THREAD:newFunction(func, holdme)
return thread:newFunctionBase(function(...)
return multi:newSystemThread("SystemThreaded Function Handler", func, ...)
end, holdme, multi.SFUNCTION)()
end
THREAD.newSystemThread = function(...)
multi:newSystemThread(...)
end
function love.threaderror(thread, errorstr)
multi.print("Thread error!\n" .. errorstr)
end
multi.integration.GLOBAL = GLOBAL
multi.integration.THREAD = THREAD
require("multi.integration.loveManager.extensions")
require("multi.integration.sharedExtensions")
multi.print("Integrated Love Threading!")
return {init = function() return GLOBAL, THREAD end}

257
integration/loveManagerold/threads.lua
View File

@ -1,257 +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 serpent = require("multi.integration.loveManager.serpent")
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
function threads.packTable(table)
return love.data.newByteData(serpent.dump(table))
end
function threads.unpackTable(data)
return serpent.load(data:getString())
end
local fRef = {"func",nil}
local function manage(channel, value)
channel:clear()
if type(value) == "table" then
channel:push{"DATA",threads.packTable(value)}
else
channel:push(value)
end
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]=="DATA" then
return threads.unpackTable(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("STATCHAN_" ..__THREADID__)
local args = multi.pack(...)
status_channel:push(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.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.packENV(env)
return threads.packTable(env)
end
function threads.unpackENV(env)
return threads.unpackTable(env)
end
function threads.setENV(env, name)
name = name or "__env"
(threads.getGlobal())[name] = threads.packTable(env)
end
function threads.getENV(name)
name = name or "__env"
return threads.unpackTable((threads.getGlobal())[name])
end
function threads.exposeENV(name)
name = name or "__env"
local env = threads.getENV(name)
for i,v in pairs(env) do
_G[i] = 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]=="DATA" then
return threads.unpackTable(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(multi.pack(...))
end
function c.error(err)
c.queue:push{"ERROR in <"..__THREADNAME__..">: "..err,__THREADID__}
error(err)
end
return c
end
if not ISTHREAD then
local queue = love.thread.getChannel("__CONSOLE__")
multi:newLoop(function(loop)
dat = queue:pop()
if dat then
print(multi.unpack(dat))
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] = threads.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

3
integration/pseudoManager/extensions.lua
View File

@ -156,7 +156,7 @@ function multi:newSystemThreadedJobQueue(n)
end
end)
for i=1,c.cores do
multi:newSystemThread("JobQueue_"..jqc.."_worker_"..i,function(jqc)
multi:newSystemThread("STJQ_"..multi.randomString(8),function(jqc)
local GLOBAL, THREAD = require("multi.integration.pseudoManager"):init()
local multi, thread = require("multi"):init()
local clock = os.clock
@ -217,4 +217,5 @@ function multi:newSystemThreadedConnection(name)
GLOBAL[name or "_"] = conn
return conn
end
require("multi.integration.sharedExtensions")

4
integration/pseudoManager/init.lua
View File

@ -24,6 +24,8 @@ SOFTWARE.
package.path = "?/init.lua;?.lua;" .. package.path
local multi, thread = require("multi"):init()
local pseudoProcessor = multi:newProcessor()
if multi.integration then
return {
init = function()
@ -89,7 +91,7 @@ function multi:newSystemThread(name, func, ...)
local GLOBAL, THREAD = activator.init(thread, env)
local th = thread:newISOThread(name, func, env, ...)
local th = pseudoProcessor:newISOThread(name, func, env, ...)
th.Type = multi.registerType("s_thread", "pseudoThreads")
th.OnError(multi.error)

4
integration/pseudoManager/threads.lua
View File

@ -110,6 +110,10 @@ local function INIT(thread)
end
end
function THREAD.sync()
thread.sleep(.5)
end
return GLOBAL, THREAD
end

10
integration/sharedExtensions/init.lua
View File

@ -60,7 +60,7 @@ function multi:newProxy(list)
return res
end
if not(self.is_init) then
THREAD.sleep(.3)
THREAD.sync()
self.is_init = true
local multi, thread = require("multi"):init()
self.proxy_link = "PL" .. multi.randomString(12)
@ -92,7 +92,6 @@ function multi:newProxy(list)
local func = table.remove(data, 1)
local sref = table.remove(data, 1)
local ret
if sref then
ret = {_G[list[0]][func](_G[list[0]], multi.unpack(data))}
else
@ -119,7 +118,8 @@ function multi:newProxy(list)
end)
return self
else
THREAD.sleep(.3)
THREAD.sync()
if not self.funcs then return self end
local function copy(obj)
if type(obj) ~= 'table' then return obj end
local res = {}
@ -267,12 +267,12 @@ function multi:newSystemThreadedProcessor(cores)
}
for _, method in pairs(implement) do
c[method] = function(self, ...)
c[method] = thread:newFunction(function(self, ...)
proxy = self.spawnTask(method, ...)
proxy:init()
references[proxy] = self
return proxy
end
end, true)
end
function c:newThread(name, func, ...)

21
tests/conf.lua
View File

@ -1,22 +1,3 @@
function love.conf(t)
t.identity = nil -- The name of the save directory (string)
t.version = "11.4" -- The LOVE version this game was made for (string)
t.console = true -- Attach a console (boolean, Windows only)
t.window = false
t.modules.audio = false -- Enable the audio module (boolean)
t.modules.event = false -- Enable the event module (boolean)
t.modules.graphics = false -- Enable the graphics module (boolean)
t.modules.image = false -- Enable the image module (boolean)
t.modules.joystick = false -- Enable the joystick module (boolean)
t.modules.keyboard = false -- Enable the keyboard module (boolean)
t.modules.math = false -- Enable the math module (boolean)
t.modules.mouse = false -- Enable the mouse module (boolean)
t.modules.physics = false -- Enable the physics module (boolean)
t.modules.sound = false -- Enable the sound module (boolean)
t.modules.system = true -- Enable the system module (boolean)
t.modules.timer = true -- Enable the timer module (boolean)
t.modules.window = false -- Enable the window module (boolean)
t.modules.thread = true -- Enable the thread module (boolean)
t.console = true
end

14
tests/main.lua
View File

@ -1,12 +1,10 @@
package.path = "../?/init.lua;../?.lua;"..package.path
require("runtests")
require("threadtests")
-- Allows you to run "love tests" which runs the tests
multi, thread = require("multi"):init()
if os.getenv("LOCAL_LUA_DEBUGGER_VSCODE") == "1" then
require("lldebugger").start()
end
GLOBAL, THREAD = require("multi.integration.loveManager"):init()
function love.update()
multi:uManager()
end
require("runtests")
require("threadtests")

69
tests/test.lua
View File

@ -1,8 +1,9 @@
package.path = "../?/init.lua;../?.lua;"..package.path
multi, thread = require("multi"):init{print=true,warn=true,debugging=true}
for i,v in pairs(thread) do
print(i,v)
end
-- for i,v in pairs(thread) do
-- print(i,v)
-- end
-- require("multi.integration.priorityManager")
-- multi.debugging.OnObjectCreated(function(obj, process)
@ -14,10 +15,6 @@ end
-- end)
-- test = multi:newProcessor("Test")
-- test:setPriorityScheme(multi.priorityScheme.TimeBased)
@ -102,39 +99,39 @@ end
-- multi:mainloop()
multi:setTaskDelay(.05)
multi:newTask(function()
for i = 1, 10 do
multi:newTask(function()
print("Task "..i)
end)
end
end)
-- multi:setTaskDelay(.05)
-- multi:newTask(function()
-- for i = 1, 10 do
-- multi:newTask(function()
-- print("Task "..i)
-- end)
-- end
-- end)
local conn = multi:newConnection()
conn(function() print("Test 1") end)
conn(function() print("Test 2") end)
conn(function() print("Test 3") end)
conn(function() print("Test 4") end)
-- local conn = multi:newConnection()
-- conn(function() print("Test 1") end)
-- conn(function() print("Test 2") end)
-- conn(function() print("Test 3") end)
-- conn(function() print("Test 4") end)
print("Fire 1")
conn:Fire()
conn = -conn
print("Fire 2")
conn:Fire()
-- print("Fire 1")
-- conn:Fire()
-- conn = -conn
-- print("Fire 2")
-- conn:Fire()
print(#conn)
-- print(#conn)
thread:newThread("Test thread", function()
print("Starting thread!")
thread.defer(function() -- Runs when the thread finishes execution
print("Clean up time!")
end)
--[[
Do lot's of stuff
]]
thread.sleep(3)
end)
-- thread:newThread("Test thread", function()
-- print("Starting thread!")
-- thread.defer(function() -- Runs when the thread finishes execution
-- print("Clean up time!")
-- end)
-- --[[
-- Do lot's of stuff
-- ]]
-- thread.sleep(3)
-- end)
multi:mainloop()

45
tests/threadtests.lua
View File

@ -1,6 +1,6 @@
package.path = "D:/VSCWorkspace/?/init.lua;D:/VSCWorkspace/?.lua;"..package.path
package.cpath = "C:/luaInstalls/lua5.4/lib/lua/5.4/?/core.dll;" .. package.cpath
multi, thread = require("multi"):init{error=true,warning=true,print=true}--{priority=true}
multi, thread = require("multi"):init{error=true,warning=true,print=true, priority=true}
proc = multi:newProcessor("Thread Test",true)
local LANES, LOVE, PSEUDO = 1, 2, 3
local env, we_good
@ -38,15 +38,15 @@ THREAD.setENV({
})
multi:newThread("Scheduler Thread",function()
-- multi:newThread(function()
-- thread.sleep(30)
-- print("Timeout tests took longer than 30 seconds")
-- multi:Stop()
-- os.exit(1)
-- end)
multi:newThread(function()
thread.sleep(30)
print("Timeout tests took longer than 30 seconds")
multi:Stop()
os.exit(1)
end)
queue = multi:newSystemThreadedQueue("Test_Queue"):init()
defer_queue = multi:newSystemThreadedQueue("Defer_Queue"):init()
queue = multi:newSystemThreadedQueue("Test_Queue")
defer_queue = multi:newSystemThreadedQueue("Defer_Queue")
multi:newSystemThread("Test_Thread_0", function()
defer_queue = THREAD.waitFor("Defer_Queue"):init()
@ -63,13 +63,11 @@ multi:newThread("Scheduler Thread",function()
end
end)
thread:newThread(function()
if thread.hold(function()
return defer_queue:pop() == "done"
end,{sleep=1}) == nil then
end,{sleep=3}) == nil then
multi.error("Thread.defer didn't work!")
end
end)
th1 = multi:newSystemThread("Test_Thread_1", function(a,b,c,d,e,f)
queue = THREAD.waitFor("Test_Queue"):init()
@ -135,7 +133,7 @@ multi:newThread("Scheduler Thread",function()
local ready = false
jq = multi:newSystemThreadedJobQueue(1) -- Job queue with 4 worker threads
jq = multi:newSystemThreadedJobQueue(4) -- Job queue with 4 worker threads
func2 = jq:newFunction("sleep",function(a,b)
THREAD.sleep(.2)
end)
@ -168,7 +166,7 @@ multi:newThread("Scheduler Thread",function()
--print("Test")
end, 1)
multi:newSystemThread("Testing proxy copy THREAD",function(tloop)
multi:newSystemThread("PROX_THREAD",function(tloop)
local multi, thread = require("multi"):init()
tloop = tloop:init()
multi.print("tloop type:",tloop.Type)
@ -176,19 +174,21 @@ multi:newThread("Scheduler Thread",function()
thread:newThread(function()
while true do
thread.hold(tloop.OnLoop)
--print(THREAD_NAME,"Loopy")
print(THREAD_NAME,"Loopy")
end
end)
tloop.OnLoop(function(a)
--print(THREAD_NAME, "Got loop...")
print(THREAD_NAME, "Got loop...")
end)
multi:mainloop()
end, tloop:getTransferable())
local test = tloop:getTransferable()
multi.print("tloop", tloop.Type)
multi.print("tloop.OnLoop", tloop.OnLoop.Type)
thread:newThread(function()
thread:newThread("Proxy Test Thread",function()
multi.print("Testing holding on a proxy connection!")
thread.hold(tloop.OnLoop)
multi.print("Held on proxy connection... once")
@ -197,17 +197,22 @@ multi:newThread("Scheduler Thread",function()
thread.hold(tloop.OnLoop)
multi.print("Held on proxy connection... finally")
proxy_test = true
end)
end).OnError(print)
thread:newThread(function()
thread.defer(function()
multi.print("Something happened!")
end)
while true do
thread.hold(tloop.OnLoop)
--print(THREAD_NAME,"Local Loopy")
multi.print(THREAD_NAME,"Local Loopy")
end
end).OnError(function(...)
print("Error",...)
end)
tloop.OnLoop(function()
--print("OnLoop",THREAD_NAME)
print("OnLoop", THREAD_NAME)
end)
t, val = thread.hold(function()