mirror of
https://github.com/rayaman/multi.git
synced 2026-09-04 23:17:35 -04:00
Working on 16.0.0 (#53)
* Fixed spelling, started ideaing for 16.0.0 * Updated files * Updated readme * Updated version * Concat conns now properly transfer events * Testing types * Connections can be % with functions * Updated connections * Fixed issue with double thread activations (Looking for another solution) * Working on issue with love threaded functions not waiting when in a thread * Working on issue where threads created in threads don't work * Fixed broken threads for love * Fixed some issues with threads * removed test * Updated changes.md * Plan on testing parity between the threading modules * Writing tests for system threading * Added test cases for threading, fixed issues. Todo test love2d * Fixed love2d to succeed with tests * All tests working * Updated files for testing * Modified tests to make it more seamless * removed extra __cores in lanes/pseudo * Working on new priority scheme * Working on priority management * Working on custom prioritySchemes * Fixed issues with missing code * Threaded processors * THREAD.exposeENV(), thread:newProcessor() * Typo in changes.md * Fixed typo in pseudoManager * fixing * Trying to fix exposeENV with pseudoThreading * Changes to threads * updated changes.md * Working on systemthreadedprocess, and experimental newProxy for threading * newProxy and STP work * newProxy implemented * Proxies work with connections now :D * Added tstep to STP, updated changes.md * thread.hold(proxy.conn) * Clean up connection events when holding, working on scheduling tasks/threads to system threaded processors * Getting loads of processors implemented * Finished getLoad(type) * Fixed some bugs * Added an easy way to share a table, found some limitations with lanes threading. * THREAD_NAME set for main thread, connections break the rules for proxies * Testing * Really close to portable proxies, currently extreamly unstable! * Debugging what is going on... * Fixed critical issue with coroutine based threads * Removed extra bloat, proxies are portable now! * Started work on the debugManager * Testing actions, fixing bugs with lanes * Testing... * fixing actions * typo fixed * Throw an error when things break * fixing stuff * Fixed issue with errors not going through * Removed system threaded connections, soon to be replaced by proxies * Testing love2d tests * Test love2d * Use later love-build * Use ubuntu for build * Fixed path * Use appimage * use sudo * No window for love2d * Fixed love2d tests * Testing love2d * Use workspace * Moved other tests while testing * actually pull the repo * packagepath set * Fixed pull * Update multi * Removed link * Edited symlink * Added timeout to build * Rewriting loveManager, too much outdated code * Still implementing new love2d threading code * Rewriting love2d threading binding * Working on adding a Hold method to all objects. Will document how they all work when done. * jobqueues having isues with stp * new pack/unpack for tables, current issue is things being turned into strings * Fixed packing of values into threads, need to fix system proxies and system threaded processors * testing... * Not hard crashing when error is encountered * Should now push non 0 exit codes * Push error when an error happens * Closer to getting things working... * Working on new type system, planning out debugmanager * Fixed error code issue * Test for 5.1 * Planning out debugManager * Some work on the debug manager, proxies working on lanes, todo get pseudo manager and love2d working * Working on getting pseudoThreading tests to work * Added function / connection * Added boost method * Document new features to conns, todo fix newTask * Fixed newTask() * Updated changes.md and fixed some bugs * Added thread.defer(func) * Fixed tests on lanes and pseudo threading, todo fix love2d threading * Fixed paths * Working on paths * Testing paths * Add test for rockspec * Fixed issues * Fixed typo * Added test for defer * Threading working in love2d * Fixed, conf * lanes uses a threaded function like waitfor function * Cleaned up changes.md * added priorityManager to rockspec
This commit is contained in:
@@ -0,0 +1,106 @@
|
||||
local multi, thread = require("multi"):init()
|
||||
|
||||
multi.defaultSettings.debugging = true
|
||||
|
||||
local dbg = {}
|
||||
dbg.__index = dbg
|
||||
dbg.processors = {}
|
||||
|
||||
-- Hooks to all on object created events!
|
||||
local c_cache = {}
|
||||
local d_cache = {}
|
||||
|
||||
local proc = multi:newProcessor("Debug_Processor").Start()
|
||||
|
||||
dbg.OnObjectCreated = function(obj, process)
|
||||
if c_cache[obj] then
|
||||
return false
|
||||
else
|
||||
c_cache[obj] = true
|
||||
proc:newTask(function()
|
||||
c_cache[obj] = false
|
||||
end)
|
||||
return true
|
||||
end
|
||||
end .. multi:newConnection()
|
||||
|
||||
dbg.OnObjectDestroyed = function(obj, process)
|
||||
if d_cache[obj] then
|
||||
return false
|
||||
else
|
||||
d_cache[obj] = true
|
||||
proc:newTask(function()
|
||||
d_cache[obj] = false
|
||||
end)
|
||||
return true
|
||||
end
|
||||
end .. multi:newConnection()
|
||||
|
||||
local creation_hook, destruction_hook
|
||||
local types
|
||||
local objects = {}
|
||||
|
||||
creation_hook = function(obj, process)
|
||||
types = multi:getTypes()
|
||||
if obj.Type == multi.PROCESS and not dbg.processors[obj] then
|
||||
obj.OnObjectCreated(creation_hook)
|
||||
obj.OnObjectDestroyed(destruction_hook)
|
||||
end
|
||||
|
||||
table.insert(objects, obj)
|
||||
|
||||
dbg.OnObjectCreated:Fire(obj, process)
|
||||
end
|
||||
|
||||
destruction_hook = function(obj, process)
|
||||
for i = 1, #objects do
|
||||
if objects[i] == obj then
|
||||
table.remove(objects, i)
|
||||
break
|
||||
end
|
||||
end
|
||||
dbg.OnObjectDestroyed:Fire(obj, process)
|
||||
end
|
||||
|
||||
function dbg:getObjects(typ)
|
||||
if type(typ) == "string" then
|
||||
local objs = {}
|
||||
for i = 1, #objects do
|
||||
if objects[i].Type == typ then
|
||||
objs[#objs+1] = objects[i]
|
||||
end
|
||||
end
|
||||
return objs
|
||||
elseif type(typ) == "table" then -- Process
|
||||
local objs = {}
|
||||
for i = 1, #objects do
|
||||
if objects[i].Parent == typ then
|
||||
objs[#objs+1] = objects[i]
|
||||
end
|
||||
end
|
||||
return objs
|
||||
elseif type(typ) == "function" then
|
||||
local objs = {}
|
||||
-- Keep objects local/private, return true to add to list, false to reject, "break" to break loop
|
||||
for i = 1, #objects do
|
||||
local ret = typ(objects[i])
|
||||
if ret then
|
||||
objs[#objs+1] = objects[i]
|
||||
elseif ret == "break" then
|
||||
break
|
||||
end
|
||||
end
|
||||
return objs
|
||||
end
|
||||
end
|
||||
|
||||
local debug_stats = {}
|
||||
|
||||
local tmulti = multi:getThreadManagerProcess()
|
||||
multi.OnObjectCreated(creation_hook)
|
||||
tmulti.OnObjectCreated(creation_hook)
|
||||
multi.OnObjectDestroyed(destroction_hook)
|
||||
tmulti.OnObjectDestroyed(destroction_hook)
|
||||
|
||||
-- We write to a debug interface in the multi namespace
|
||||
multi.debugging = dbg
|
||||
@@ -0,0 +1,46 @@
|
||||
local multi, thread = require("multi"):init{error=true}
|
||||
multi.error("Currntly not supported!")
|
||||
os.exit(1)
|
||||
local effil = require("effil")
|
||||
|
||||
-- I like some of the things that this library offers.
|
||||
-- Current limitations prevent me from being able to use effil,
|
||||
-- but I might fork and work on it myself.
|
||||
|
||||
-- Configs
|
||||
effil.allow_table_upvalues(false)
|
||||
|
||||
local GLOBAL,THREAD = require("multi.integration.effilManager.threads").init()
|
||||
local count = 1
|
||||
local started = false
|
||||
local livingThreads = {}
|
||||
|
||||
function multi:newSystemThread(name, func, ...)
|
||||
local name = name or multi.randomString(16)
|
||||
local rand = math.random(1, 10000000)
|
||||
c = {}
|
||||
c.name = name
|
||||
c.Name = name
|
||||
c.Id = count
|
||||
end
|
||||
|
||||
function THREAD:newFunction(func, holdme)
|
||||
return thread:newFunctionBase(function(...)
|
||||
return multi:newSystemThread("TempSystemThread",func,...)
|
||||
end, holdme, multi.SFUNCTION)()
|
||||
end
|
||||
|
||||
THREAD.newSystemThread = function(...)
|
||||
multi:newSystemThread(...)
|
||||
end
|
||||
|
||||
multi.print("Integrated Effil Threading!")
|
||||
multi.integration = {} -- for module creators
|
||||
multi.integration.GLOBAL = GLOBAL
|
||||
multi.integration.THREAD = THREAD
|
||||
require("multi.integration.effilManager.extensions")
|
||||
return {
|
||||
init = function()
|
||||
return GLOBAL, THREAD
|
||||
end
|
||||
}
|
||||
@@ -22,29 +22,57 @@ 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
|
||||
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()
|
||||
c.Type = multi.registerType("s_queue")
|
||||
|
||||
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
|
||||
|
||||
if multi.isMainThread then
|
||||
multi.integration.GLOBAL[name] = c
|
||||
else
|
||||
GLOBAL[name] = c
|
||||
end
|
||||
|
||||
function c:Hold(opt)
|
||||
local multi, thread = require("multi"):init()
|
||||
if opt.peek then
|
||||
return thread.hold(function()
|
||||
return self:peek()
|
||||
end)
|
||||
else
|
||||
return thread.hold(function()
|
||||
return self:pop()
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
||||
self:create(c)
|
||||
|
||||
return c
|
||||
end
|
||||
|
||||
@@ -53,37 +81,60 @@ function multi:newSystemThreadedTable(name)
|
||||
local c = {}
|
||||
c.link = lanes.linda()
|
||||
c.Name = name
|
||||
setmetatable(c,{
|
||||
c.Type = multi.registerType("s_table")
|
||||
|
||||
function c:init()
|
||||
return self
|
||||
end
|
||||
|
||||
setmetatable(c,{
|
||||
__index = function(t,k)
|
||||
return c.link:get(k)
|
||||
end,
|
||||
__newindex = function(t,k,v)
|
||||
c.link:set(k,v)
|
||||
c.link:set(k, v)
|
||||
end
|
||||
})
|
||||
function c:init()
|
||||
return self
|
||||
|
||||
if multi.isMainThread then
|
||||
multi.integration.GLOBAL[name] = c
|
||||
else
|
||||
GLOBAL[name] = c
|
||||
end
|
||||
|
||||
function c:Hold(opt)
|
||||
local multi, thread = require("multi"):init()
|
||||
if opt.key then
|
||||
return thread.hold(function()
|
||||
return self.tab[opt.key]
|
||||
end)
|
||||
else
|
||||
multi.error("Must provide a key to check opt.key = 'key'")
|
||||
end
|
||||
end
|
||||
GLOBAL[name or "_"] = c
|
||||
|
||||
self:create(c)
|
||||
|
||||
return c
|
||||
end
|
||||
|
||||
function multi:newSystemThreadedJobQueue(n)
|
||||
local c = {}
|
||||
c.cores = n or THREAD.getCores()*2
|
||||
c.Type = multi.registerType("s_jobqueue")
|
||||
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 funcs = multi:newSystemThreadedTable()
|
||||
local queueJob = multi:newSystemThreadedQueue()
|
||||
local queueReturn = multi:newSystemThreadedQueue()
|
||||
local doAll = multi:newSystemThreadedQueue()
|
||||
local ID=1
|
||||
local jid = 1
|
||||
function c:isEmpty()
|
||||
return queueJob:peek()==nil
|
||||
end
|
||||
function c:doToAll(func)
|
||||
function c:doToAll(func,...)
|
||||
for i=1,c.cores do
|
||||
doAll:push{ID,func}
|
||||
doAll:push{ID,func,...}
|
||||
end
|
||||
ID = ID + 1
|
||||
return self
|
||||
@@ -93,12 +144,12 @@ function multi:newSystemThreadedJobQueue(n)
|
||||
return self
|
||||
end
|
||||
function c:pushJob(name,...)
|
||||
queueJob:push{name,jid,{...}}
|
||||
queueJob:push{name,jid,multi.pack(...)}
|
||||
jid = jid + 1
|
||||
return jid-1
|
||||
end
|
||||
local nFunc = 0
|
||||
function c:newFunction(name,func,holup) -- This registers with the queue
|
||||
function c:newFunction(name, func, holup) -- This registers with the queue
|
||||
if type(name)=="function" then
|
||||
holup = func
|
||||
func = name
|
||||
@@ -112,32 +163,38 @@ function multi:newSystemThreadedJobQueue(n)
|
||||
local rets
|
||||
link = c.OnJobCompleted(function(jid,...)
|
||||
if id==jid then
|
||||
rets = {...}
|
||||
link:Destroy()
|
||||
rets = multi.pack(...)
|
||||
end
|
||||
end)
|
||||
return thread.hold(function()
|
||||
if rets then
|
||||
return unpack(rets) or multi.NIL
|
||||
if #rets == 0 then
|
||||
return multi.NIL
|
||||
else
|
||||
return multi.unpack(rets)
|
||||
end
|
||||
end
|
||||
end)
|
||||
end,holup),name
|
||||
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))
|
||||
if job then
|
||||
local id = table.remove(job,1)
|
||||
c.OnJobCompleted:Fire(id,multi.unpack(job))
|
||||
end
|
||||
end
|
||||
end)
|
||||
for i=1,c.cores do
|
||||
multi:newSystemThread("SystemThreadedJobQueue",function(queue)
|
||||
local multi,thread = require("multi"):init()
|
||||
multi:newSystemThread("STJQ_"..multi.randomString(8),function(queue)
|
||||
local multi, thread = require("multi"):init()
|
||||
local idle = os.clock()
|
||||
local clock = os.clock
|
||||
local ref = 0
|
||||
_G["__QR"] = queueReturn
|
||||
setmetatable(_G,{__index = funcs})
|
||||
thread:newThread("JobHandler",function()
|
||||
while true do
|
||||
@@ -145,10 +202,12 @@ function multi:newSystemThreadedJobQueue(n)
|
||||
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}
|
||||
thread:newThread("JobQueue-Spawn",function()
|
||||
local name = table.remove(dat, 1)
|
||||
local jid = table.remove(dat, 1)
|
||||
local args = table.remove(dat, 1)
|
||||
queueReturn:push{jid, funcs[name](args[1],args[2],args[3],args[4],args[5],args[6],args[7],args[8]), queue}
|
||||
end)
|
||||
end
|
||||
end)
|
||||
thread:newThread("DoAllHandler",function()
|
||||
@@ -158,9 +217,10 @@ function multi:newSystemThreadedJobQueue(n)
|
||||
end)
|
||||
if dat then
|
||||
if dat[1]>ref then
|
||||
ref = table.remove(dat, 1)
|
||||
func = table.remove(dat, 1)
|
||||
idle = clock()
|
||||
ref = dat[1]
|
||||
dat[2]()
|
||||
func(unpack(dat))
|
||||
doAll:pop()
|
||||
end
|
||||
end
|
||||
@@ -175,14 +235,22 @@ function multi:newSystemThreadedJobQueue(n)
|
||||
end
|
||||
end)
|
||||
multi:mainloop()
|
||||
end,i).priority = thread.Priority_Core
|
||||
end,i)
|
||||
end
|
||||
|
||||
function c:Hold(opt)
|
||||
return thread.hold(self.OnJobCompleted)
|
||||
end
|
||||
|
||||
self:create(c)
|
||||
|
||||
return c
|
||||
end
|
||||
|
||||
function multi:newSystemThreadedConnection(name)
|
||||
local name = name or multi.randomString(16)
|
||||
local c = {}
|
||||
c.Type = multi.registerType("s_connection")
|
||||
c.CONN = 0x00
|
||||
c.TRIG = 0x01
|
||||
c.PING = 0x02
|
||||
@@ -196,7 +264,7 @@ function multi:newSystemThreadedConnection(name)
|
||||
end
|
||||
return r
|
||||
end
|
||||
c.CID = THREAD.getID()
|
||||
c.CID = THREAD_ID
|
||||
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.
|
||||
@@ -233,7 +301,7 @@ function multi:newSystemThreadedConnection(name)
|
||||
|
||||
local function fire(...)
|
||||
for _, link in pairs(c.links) do
|
||||
link:push {c.TRIG, {...}}
|
||||
link:push {c.TRIG, multi.pack(...)}
|
||||
end
|
||||
end
|
||||
|
||||
@@ -253,16 +321,16 @@ function multi:newSystemThreadedConnection(name)
|
||||
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]))
|
||||
fire(multi.unpack(item[2]))
|
||||
c.proxy_conn:Fire(multi.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
|
||||
local args = multi.pack(...)
|
||||
if self.CID == THREAD_ID then -- Host Call
|
||||
for _, link in pairs(self.links) do
|
||||
link:push {self.TRIG, args}
|
||||
end
|
||||
@@ -277,8 +345,14 @@ function multi:newSystemThreadedConnection(name)
|
||||
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
|
||||
local tempMT = {}
|
||||
for i,v in pairs(mt) do
|
||||
tempMT[i] = v
|
||||
end
|
||||
tempMT.__index = self.proxy_conn
|
||||
tempMT.__call = function(t,func) self.proxy_conn(func) end
|
||||
setmetatable(self, tempMT)
|
||||
if self.CID == THREAD_ID then return self end
|
||||
thread:newThread("STC_CONN_MAN"..name,function()
|
||||
local item
|
||||
local link_self_ref = multi:newSystemThreadedQueue()
|
||||
@@ -296,7 +370,7 @@ function multi:newSystemThreadedConnection(name)
|
||||
end
|
||||
link_self_ref:pop()
|
||||
elseif item[1] == self.TRIG then
|
||||
self.proxy_conn:Fire(unpack(item[2]))
|
||||
self.proxy_conn:Fire(multi.unpack(item[2]))
|
||||
link_self_ref:pop()
|
||||
else
|
||||
-- This shouldn't be the case
|
||||
@@ -306,7 +380,14 @@ function multi:newSystemThreadedConnection(name)
|
||||
return self
|
||||
end
|
||||
|
||||
GLOBAL[name] = c
|
||||
if multi.isMainThread then
|
||||
multi.integration.GLOBAL[name] = c
|
||||
else
|
||||
GLOBAL[name] = c
|
||||
end
|
||||
|
||||
self:create(c)
|
||||
|
||||
return c
|
||||
end
|
||||
end
|
||||
require("multi.integration.sharedExtensions")
|
||||
@@ -36,6 +36,9 @@ lanes = require("lanes").configure()
|
||||
multi.SystemThreads = {}
|
||||
multi.isMainThread = true
|
||||
|
||||
_G.THREAD_NAME = "MAIN_THREAD"
|
||||
_G.THREAD_ID = 0
|
||||
|
||||
function multi:canSystemThread()
|
||||
return true
|
||||
end
|
||||
@@ -55,10 +58,10 @@ local count = 1
|
||||
local started = false
|
||||
local livingThreads = {}
|
||||
|
||||
function THREAD:newFunction(func,holdme)
|
||||
function THREAD:newFunction(func, holdme)
|
||||
return thread:newFunctionBase(function(...)
|
||||
return multi:newSystemThread("TempSystemThread",func,...)
|
||||
end,holdme)()
|
||||
end, holdme, multi.registerType("s_function"))()
|
||||
end
|
||||
|
||||
function multi:newSystemThread(name, func, ...)
|
||||
@@ -67,35 +70,43 @@ function multi:newSystemThread(name, func, ...)
|
||||
local rand = math.random(1, 10000000)
|
||||
local return_linda = lanes.linda()
|
||||
c = {}
|
||||
c.name = name
|
||||
c.Name = name
|
||||
c.Id = count
|
||||
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.Type = multi.registerType("s_thread")
|
||||
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)
|
||||
local globe = {
|
||||
THREAD_NAME = name,
|
||||
THREAD_ID = count,
|
||||
THREAD = THREAD,
|
||||
GLOBAL = GLOBAL,
|
||||
_Console = __ConsoleLinda,
|
||||
_DEFER = {}
|
||||
}
|
||||
if GLOBAL["__env"] then
|
||||
for i,v in pairs(GLOBAL["__env"]) do
|
||||
globe[i] = v
|
||||
end
|
||||
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
|
||||
globals = globe,
|
||||
priority = c.priority
|
||||
},function(...)
|
||||
require("multi"):init(multi_settings)
|
||||
multi, thread = require("multi"):init(multi_settings)
|
||||
require("multi.integration.lanesManager.extensions")
|
||||
require("multi.integration.sharedExtensions")
|
||||
local has_error = true
|
||||
return_linda:set("returns",{func(...)})
|
||||
returns = {pcall(func, ...)}
|
||||
return_linda:set("returns", returns)
|
||||
for i,v in pairs(_DEFER) do
|
||||
pcall(v)
|
||||
end
|
||||
has_error = false
|
||||
end)(...)
|
||||
count = count + 1
|
||||
@@ -110,10 +121,20 @@ function multi:newSystemThread(name, func, ...)
|
||||
c.OnDeath = multi:newConnection()
|
||||
c.OnError = multi:newConnection()
|
||||
GLOBAL["__THREADS__"] = livingThreads
|
||||
c.OnError(multi.error)
|
||||
|
||||
if self.isActor then
|
||||
self:create(c)
|
||||
else
|
||||
multi.create(multi, c)
|
||||
end
|
||||
|
||||
return c
|
||||
end
|
||||
|
||||
THREAD.newSystemThread = multi.newSystemThread
|
||||
THREAD.newSystemThread = function(...)
|
||||
multi:newSystemThread(...)
|
||||
end
|
||||
|
||||
function multi.InitSystemThreadErrorHandler()
|
||||
if started == true then
|
||||
@@ -126,18 +147,26 @@ function multi.InitSystemThreadErrorHandler()
|
||||
while true do
|
||||
thread.yield()
|
||||
_,data = __ConsoleLinda:receive(0, "Q")
|
||||
if data then print(unpack(data)) end
|
||||
if data then
|
||||
--print(data[1])
|
||||
end
|
||||
for i = #threads, 1, -1 do
|
||||
temp = threads[i]
|
||||
status = temp.thread.status
|
||||
push = __StatusLinda:get(temp.Id)
|
||||
push = __StatusLinda:get(temp.ID)
|
||||
if push then
|
||||
temp.statusconnector:Fire(unpack(({__StatusLinda:receive(nil, temp.Id)})[2]))
|
||||
temp.statusconnector:Fire(multi.unpack(({__StatusLinda:receive(nil, temp.ID)})[2]))
|
||||
end
|
||||
if status == "done" or temp.returns:get("returns") then
|
||||
livingThreads[temp.Id] = {false, temp.Name}
|
||||
returns = ({temp.returns:receive(0, "returns")})[2]
|
||||
livingThreads[temp.ID] = {false, temp.Name}
|
||||
temp.alive = false
|
||||
temp.OnDeath:Fire(unpack(({temp.returns:receive(0, "returns")})[2]))
|
||||
if returns[1] == false then
|
||||
temp.OnError:Fire(temp, returns[2])
|
||||
else
|
||||
table.remove(returns,1)
|
||||
temp.OnDeath:Fire(multi.unpack(returns))
|
||||
end
|
||||
GLOBAL["__THREADS__"] = livingThreads
|
||||
table.remove(threads, i)
|
||||
elseif status == "running" then
|
||||
@@ -145,19 +174,15 @@ function multi.InitSystemThreadErrorHandler()
|
||||
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)
|
||||
-- The thread never really errors, we handle this through our linda object
|
||||
elseif status == "cancelled" then
|
||||
livingThreads[temp.Id] = {false, temp.Name}
|
||||
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}
|
||||
livingThreads[temp.ID] = {false, temp.Name}
|
||||
temp.alive = false
|
||||
temp.OnError:Fire(temp,"thread_killed")
|
||||
GLOBAL["__THREADS__"] = livingThreads
|
||||
@@ -165,10 +190,10 @@ function multi.InitSystemThreadErrorHandler()
|
||||
end
|
||||
end
|
||||
end
|
||||
end)
|
||||
end).OnError(multi.error)
|
||||
end
|
||||
|
||||
multi.print("Integrated Lanes!")
|
||||
multi.print("Integrated Lanes Threading!")
|
||||
multi.integration = {} -- for module creators
|
||||
multi.integration.GLOBAL = GLOBAL
|
||||
multi.integration.THREAD = THREAD
|
||||
|
||||
@@ -48,20 +48,12 @@ local function INIT(__GlobalLinda, __SleepingLinda, __StatusLinda, __Console)
|
||||
end
|
||||
|
||||
function THREAD.waitFor(name)
|
||||
local function wait()
|
||||
local multi, thread = require("multi"):init()
|
||||
return multi.hold(function()
|
||||
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"))
|
||||
return __GlobalLinda:get(name)
|
||||
end)
|
||||
end
|
||||
|
||||
function THREAD.getCores()
|
||||
@@ -72,11 +64,11 @@ local function INIT(__GlobalLinda, __SleepingLinda, __StatusLinda, __Console)
|
||||
local c = {}
|
||||
c.queue = __Console
|
||||
function c.print(...)
|
||||
c.queue:send("Q", {...})
|
||||
c.queue:push("Q", table.concat(multi.pack(...), "\t"))
|
||||
end
|
||||
function c.error(err)
|
||||
c.queue:push("Q",{"ERROR in <"..__THREADNAME__..">: "..err,__THREADID__})
|
||||
error(err)
|
||||
c.queue:push("Q", "Error in <"..THREAD_NAME..":" .. THREAD_ID .. ">: ".. err)
|
||||
multi.error(err)
|
||||
end
|
||||
return c
|
||||
end
|
||||
@@ -95,16 +87,12 @@ local function INIT(__GlobalLinda, __SleepingLinda, __StatusLinda, __Console)
|
||||
error("Thread was killed!\1")
|
||||
end
|
||||
|
||||
function THREAD.getName()
|
||||
return THREAD_NAME
|
||||
end
|
||||
|
||||
function THREAD.getID()
|
||||
return THREAD_ID
|
||||
function THREAD.sync()
|
||||
-- Maybe do something...
|
||||
end
|
||||
|
||||
function THREAD.pushStatus(...)
|
||||
local args = {...}
|
||||
local args = multi.pack(...)
|
||||
__StatusLinda:send(nil,THREAD_ID, args)
|
||||
end
|
||||
|
||||
@@ -134,9 +122,30 @@ local function INIT(__GlobalLinda, __SleepingLinda, __StatusLinda, __Console)
|
||||
__GlobalLinda:set(k, v)
|
||||
end
|
||||
})
|
||||
|
||||
function THREAD.setENV(env, name)
|
||||
GLOBAL[name or "__env"] = env
|
||||
end
|
||||
|
||||
function THREAD.getENV(name)
|
||||
return GLOBAL[name or "__env"]
|
||||
end
|
||||
|
||||
function THREAD.exposeENV(name)
|
||||
name = name or "__env"
|
||||
local env = THREAD.getENV(name)
|
||||
for i,v in pairs(env) do
|
||||
_G[i] = v
|
||||
end
|
||||
end
|
||||
|
||||
function THREAD.defer(func)
|
||||
table.insert(_DEFER, func)
|
||||
end
|
||||
|
||||
return GLOBAL, THREAD
|
||||
end
|
||||
|
||||
return {init = function(g,s,st,c)
|
||||
return INIT(g,s,st,c)
|
||||
return {init = function(g,s,st,c,onexit)
|
||||
return INIT(g,s,st,c,onexit)
|
||||
end}
|
||||
@@ -1,106 +1,131 @@
|
||||
--[[
|
||||
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
|
||||
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
|
||||
c.Type = multi.registerType("s_queue")
|
||||
c.chan = love.thread.getChannel(name)
|
||||
|
||||
function c:push(dat)
|
||||
self.chan:push(THREAD.packValue(dat))
|
||||
end
|
||||
|
||||
function c:pop()
|
||||
return THREAD.unpackValue(self.chan:pop())
|
||||
end
|
||||
|
||||
function c:peek()
|
||||
return THREAD.unpackValue(self.chan:peek())
|
||||
end
|
||||
|
||||
function c:init()
|
||||
self.chan = love.thread.getChannel(self.Name)
|
||||
return self
|
||||
end
|
||||
|
||||
function c:Hold(opt)
|
||||
local multi, thread = require("multi"):init()
|
||||
if opt.peek then
|
||||
return thread.hold(function()
|
||||
return self:peek()
|
||||
end)
|
||||
else
|
||||
return thread.hold(function()
|
||||
return self:pop()
|
||||
end)
|
||||
end
|
||||
end
|
||||
THREAD.package(name,c)
|
||||
|
||||
GLOBAL[name] = c
|
||||
|
||||
self:create(c)
|
||||
|
||||
return c
|
||||
end
|
||||
|
||||
function multi:newSystemThreadedTable(name)
|
||||
local name = name or multi.randomString(16)
|
||||
local c = {}
|
||||
|
||||
local c = {}
|
||||
|
||||
c.Name = name
|
||||
c.Type = multi.registerType("s_table")
|
||||
c.tab = THREAD.createTable(name)
|
||||
|
||||
function c:init()
|
||||
return THREAD.createTable(self.Name)
|
||||
self.tab = THREAD.createTable(self.Name)
|
||||
setmetatable(self,{
|
||||
__index = function(t, k)
|
||||
return self.tab[k]
|
||||
end,
|
||||
__newindex = function(t,k,v)
|
||||
self.tab[k] = v
|
||||
end
|
||||
})
|
||||
return self
|
||||
end
|
||||
THREAD.package(name,c)
|
||||
|
||||
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]
|
||||
end)
|
||||
else
|
||||
multi.error("Must provide a key to check opt.key = 'key'")
|
||||
end
|
||||
end
|
||||
|
||||
setmetatable(c,{
|
||||
__index = function(t, k)
|
||||
return c.tab[k]
|
||||
end,
|
||||
__newindex = function(t,k,v)
|
||||
c.tab[k] = v
|
||||
end
|
||||
})
|
||||
|
||||
GLOBAL[name] = c
|
||||
|
||||
self:create(c)
|
||||
|
||||
return c
|
||||
end
|
||||
|
||||
local jqc = 1
|
||||
function multi:newSystemThreadedJobQueue(n)
|
||||
local c = {}
|
||||
|
||||
c.cores = n or THREAD.getCores()
|
||||
c.registerQueue = {}
|
||||
c.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.Type = multi.registerType("s_jobqueue")
|
||||
c.funcs = THREAD.createTable("__JobQueue_"..jqc.."_table")
|
||||
c.queue = multi:newSystemThreadedQueue("__JobQueue_"..jqc.."_queue")
|
||||
c.queueReturn = multi:newSystemThreadedQueue("__JobQueue_"..jqc.."_queueReturn")
|
||||
c.queueAll = multi:newSystemThreadedQueue("__JobQueue_"..jqc.."_queueAll")
|
||||
c.id = 0
|
||||
c.OnJobCompleted = multi:newConnection()
|
||||
|
||||
local allfunc = 0
|
||||
|
||||
function c:doToAll(func)
|
||||
local f = THREAD.dump(func)
|
||||
for i = 1, self.cores do
|
||||
self.queueAll:push({allfunc,f})
|
||||
self.queueAll:push({allfunc, func})
|
||||
end
|
||||
allfunc = allfunc + 1
|
||||
end
|
||||
function c:registerFunction(name,func)
|
||||
function c:registerFunction(name, func)
|
||||
if self.funcs[name] then
|
||||
error("A function by the name "..name.." has already been registered!")
|
||||
multi.error("A function by the name "..name.." has already been registered!")
|
||||
end
|
||||
self.funcs[name] = func
|
||||
end
|
||||
@@ -127,13 +152,12 @@ function multi:newSystemThreadedJobQueue(n)
|
||||
local rets
|
||||
link = c.OnJobCompleted(function(jid,...)
|
||||
if id==jid then
|
||||
rets = {...}
|
||||
link:Destroy()
|
||||
rets = multi.pack(...)
|
||||
end
|
||||
end)
|
||||
return thread.hold(function()
|
||||
if rets then
|
||||
return unpack(rets) or multi.NIL
|
||||
return multi.unpack(rets) or multi.NIL
|
||||
end
|
||||
end)
|
||||
end,holup),name
|
||||
@@ -143,7 +167,7 @@ function multi:newSystemThreadedJobQueue(n)
|
||||
thread.yield()
|
||||
local dat = c.queueReturn:pop()
|
||||
if dat then
|
||||
c.OnJobCompleted:Fire(unpack(dat))
|
||||
c.OnJobCompleted:Fire(multi.unpack(dat))
|
||||
end
|
||||
end
|
||||
end)
|
||||
@@ -151,16 +175,15 @@ function multi:newSystemThreadedJobQueue(n)
|
||||
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
|
||||
love.timer.sleep(1)
|
||||
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 funcs = THREAD.createTable("__JobQueue_"..jqc.."_table")
|
||||
local queue = THREAD.waitFor("__JobQueue_"..jqc.."_queue")
|
||||
local queueReturn = THREAD.waitFor("__JobQueue_"..jqc.."_queueReturn")
|
||||
local lastProc = clock()
|
||||
local queueAll = love.thread.getChannel("__JobQueue_"..jqc.."_queueAll")
|
||||
local queueAll = THREAD.waitFor("__JobQueue_"..jqc.."_queueAll")
|
||||
local registry = {}
|
||||
_G["__QR"] = queueReturn
|
||||
setmetatable(_G,{__index = funcs})
|
||||
thread:newThread("startUp",function()
|
||||
while true do
|
||||
@@ -168,7 +191,7 @@ function multi:newSystemThreadedJobQueue(n)
|
||||
local all = queueAll:peek()
|
||||
if all and not registry[all[1]] then
|
||||
lastProc = os.clock()
|
||||
THREAD.loadDump(queueAll:pop()[2])()
|
||||
queueAll:pop()[2]()
|
||||
end
|
||||
end
|
||||
end)
|
||||
@@ -179,20 +202,21 @@ function multi:newSystemThreadedJobQueue(n)
|
||||
local all = queueAll:peek()
|
||||
if all and not registry[all[1]] then
|
||||
lastProc = os.clock()
|
||||
THREAD.loadDump(queueAll:pop()[2])()
|
||||
queueAll:pop()[2]()
|
||||
end
|
||||
local dat = queue:performAtomic(atomic)
|
||||
local dat = thread.hold(queue)
|
||||
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)
|
||||
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)
|
||||
--local test = queueReturn.push
|
||||
queueReturn:push(tab)
|
||||
end)
|
||||
end
|
||||
end
|
||||
end):OnError(function(...)
|
||||
error(...)
|
||||
end)
|
||||
thread:newThread("Idler",function()
|
||||
while true do
|
||||
@@ -207,145 +231,14 @@ function multi:newSystemThreadedJobQueue(n)
|
||||
multi:mainloop()
|
||||
end,jqc)
|
||||
end
|
||||
|
||||
function c:Hold(opt)
|
||||
return thread.hold(self.OnJobCompleted)
|
||||
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 subscribe = love.thread.getChannel("SUB_STC_" .. name)
|
||||
|
||||
function c:init()
|
||||
|
||||
self.subscribe = love.thread.getChannel("SUB_STC_" .. self.Name)
|
||||
|
||||
function self:Fire(...)
|
||||
local args = {...}
|
||||
if self.CID == THREAD.getID() 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.getID() 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(unpack(item[2]))
|
||||
link_self_ref:pop()
|
||||
else
|
||||
-- This shouldn't be the case
|
||||
end
|
||||
end
|
||||
end).OnError(print)
|
||||
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.getID()
|
||||
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, {...}}
|
||||
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(unpack(item[2]))
|
||||
c.proxy_conn:Fire(unpack(item[2]))
|
||||
end
|
||||
end
|
||||
end).OnError(print)
|
||||
--- ^^^ This will only exist in the init thread
|
||||
|
||||
THREAD.package(name,c)
|
||||
self:create(c)
|
||||
|
||||
return c
|
||||
end
|
||||
@@ -1,122 +1,137 @@
|
||||
--[[
|
||||
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)
|
||||
__THREADID__=table.remove(__IMPORTS,1)
|
||||
__THREADNAME__=table.remove(__IMPORTS,1)
|
||||
math.randomseed(__THREADID__)
|
||||
args = {...}
|
||||
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()
|
||||
math.random()
|
||||
stab = THREAD.createStaticTable(__THREADNAME__)
|
||||
GLOBAL = THREAD.getGlobal()
|
||||
multi, thread = require("multi").init()
|
||||
multi.integration={}
|
||||
multi.integration.GLOBAL = GLOBAL
|
||||
multi.integration.THREAD = THREAD
|
||||
pcall(require,"multi.integration.loveManager.extensions")
|
||||
stab["returns"] = {THREAD.loadDump(__FUNC__)(unpack(__IMPORTS))}
|
||||
stab = THREAD.createTable(THREAD_NAME .. THREAD_ID)
|
||||
if GLOBAL["__env"] then
|
||||
local env = THREAD.getENV()
|
||||
for i,v in pairs(env) do
|
||||
_G[i] = v
|
||||
end
|
||||
end
|
||||
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")
|
||||
local returns = {pcall(__FUNC, multi.unpack(ARGS))}
|
||||
table.remove(returns,1)
|
||||
stab["returns"] = returns
|
||||
for i,v in pairs(DEFER) do
|
||||
pcall(v)
|
||||
end
|
||||
]]
|
||||
local multi, thread = require("multi"):init()
|
||||
local THREAD = {}
|
||||
__THREADID__ = 0
|
||||
__THREADNAME__ = "MainThread"
|
||||
multi.integration={}
|
||||
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,...)
|
||||
_G.THREAD_NAME = "MAIN_THREAD"
|
||||
_G.THREAD_ID = 0
|
||||
|
||||
local multi, thread = require("multi"):init()
|
||||
local GLOBAL, THREAD = require("multi.integration.loveManager.threads"):init()
|
||||
|
||||
multi.registerType("s_function")
|
||||
multi.registerType("s_thread")
|
||||
|
||||
multi.integration = {}
|
||||
multi.isMainThread = true
|
||||
local threads = {}
|
||||
local tid = 0
|
||||
function multi:newSystemThread(name, func, ...)
|
||||
multi.InitSystemThreadErrorHandler()
|
||||
local name = name or multi.randomString(16)
|
||||
tid = tid + 1
|
||||
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
|
||||
c.Type = multi.STHREAD
|
||||
c.Name = name
|
||||
c.ID = tid
|
||||
c.thread = love.thread.newThread(ThreadFileData)
|
||||
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()
|
||||
c.OnError = multi:newConnection()
|
||||
c.status_channel = love.thread.getChannel("__status_channel__" .. c.ID)
|
||||
|
||||
function c:getName() return c.name end
|
||||
|
||||
table.insert(threads, c)
|
||||
|
||||
c.OnError(multi.error)
|
||||
|
||||
if self.isActor then
|
||||
self:create(c)
|
||||
else
|
||||
multi.create(multi, c)
|
||||
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)()
|
||||
local started = false
|
||||
local console_channel = love.thread.getChannel("__console_channel__")
|
||||
|
||||
function THREAD:newFunction(func, holdme)
|
||||
return thread:newFunctionBase(function(...)
|
||||
return multi:newSystemThread("SystemThreaded Function Handler", func, ...)
|
||||
end, holdme, multi.SFUNCTION)()
|
||||
end
|
||||
|
||||
THREAD.newSystemThread = multi.newSystemThread
|
||||
|
||||
function love.threaderror(thread, errorstr)
|
||||
multi.print("Thread error!\n"..errorstr)
|
||||
multi.error("Thread error! " .. errorstr)
|
||||
end
|
||||
|
||||
function multi.InitSystemThreadErrorHandler()
|
||||
if started == true then return end
|
||||
started = true
|
||||
thread:newThread("Love System Thread Handler", function()
|
||||
while true do
|
||||
thread.yield()
|
||||
for i = #threads, 1, -1 do
|
||||
local th = threads[i]
|
||||
if th.status_channel:peek() ~= nil then
|
||||
th.statusconnector:Fire(multi.unpack(th.status_channel:pop()))
|
||||
end
|
||||
local th_err = th.thread:getError()
|
||||
if th_err == "Thread Killed!\1" then
|
||||
th.OnDeath:Fire("Thread Killed!")
|
||||
table.remove(threads, i)
|
||||
elseif th_err then
|
||||
th.OnError:Fire(th, th_err)
|
||||
table.remove(threads, i)
|
||||
elseif th.stab.returns then
|
||||
th.OnDeath:Fire(multi.unpack(th.stab.returns))
|
||||
th.stab.returns = nil
|
||||
table.remove(threads, i)
|
||||
end
|
||||
end
|
||||
end
|
||||
end)
|
||||
end
|
||||
|
||||
THREAD.newSystemThread = function(...)
|
||||
multi:newSystemThread(...)
|
||||
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}
|
||||
|
||||
return {
|
||||
init = function()
|
||||
return GLOBAL, THREAD
|
||||
end
|
||||
}
|
||||
|
||||
+165
-187
@@ -25,208 +25,96 @@ require("love.timer")
|
||||
require("love.system")
|
||||
require("love.data")
|
||||
require("love.thread")
|
||||
local socket = require("socket")
|
||||
local multi, thread = require("multi").init()
|
||||
local threads = {}
|
||||
local multi, thread = require("multi"):init()
|
||||
|
||||
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)
|
||||
-- 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
|
||||
|
||||
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
|
||||
-- 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
|
||||
|
||||
function threads.waitFor(name)
|
||||
if thread.isThread() then
|
||||
return thread.hold(function()
|
||||
return threads.get(name)
|
||||
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
|
||||
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]
|
||||
if t.init then
|
||||
t:init()
|
||||
end
|
||||
return t
|
||||
end
|
||||
|
||||
function threads.getThread(n)
|
||||
return GLOBAL["__THREAD_"..n]
|
||||
local function packValue(t)
|
||||
return tableToFunctionString(t)
|
||||
end
|
||||
|
||||
function threads.getName()
|
||||
return __THREADNAME__
|
||||
local function unpackValue(t)
|
||||
return functionStringToTable(t)
|
||||
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)
|
||||
local function createTable(n)
|
||||
if not n then
|
||||
n = "STAB"..multi.randomString(8)
|
||||
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
|
||||
local function set(name, val)
|
||||
local chan = love.thread.getChannel(n .. name)
|
||||
if chan:getCount() == 1 then chan:pop() end
|
||||
__proxy[name] = true
|
||||
chan:push(packValue(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
|
||||
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({},
|
||||
{
|
||||
@@ -240,11 +128,101 @@ function threads.createStaticTable(n)
|
||||
)
|
||||
end
|
||||
|
||||
function threads.hold(n)
|
||||
local dat
|
||||
while not(dat) do
|
||||
dat = n()
|
||||
function INIT()
|
||||
local GLOBAL, THREAD, DEFER = createTable("__GLOBAL__"), {}, {}
|
||||
local status_channel, console_channel = love.thread.getChannel("__status_channel__" .. THREAD_ID),
|
||||
love.thread.getChannel("__console_channel__")
|
||||
|
||||
-- Non portable methods, shouldn't be used unless you know what you are doing
|
||||
THREAD.packValue = packValue
|
||||
THREAD.unpackValue = unpackValue
|
||||
THREAD.createTable = createTable
|
||||
|
||||
function THREAD.set(name, val)
|
||||
GLOBAL[name] = val
|
||||
end
|
||||
|
||||
function THREAD.get(name, val)
|
||||
return GLOBAL[name]
|
||||
end
|
||||
|
||||
THREAD.waitFor = thread:newFunction(function(name)
|
||||
local function wait()
|
||||
math.randomseed(os.time())
|
||||
thread.yield()
|
||||
end
|
||||
repeat
|
||||
wait()
|
||||
until GLOBAL[name] ~= nil
|
||||
return GLOBAL[name]
|
||||
end, true)
|
||||
|
||||
function THREAD.getCores()
|
||||
return love.system.getProcessorCount()
|
||||
end
|
||||
|
||||
function THREAD.getConsole()
|
||||
local c = {}
|
||||
c.queue = console_channel
|
||||
function c.print(...)
|
||||
c.queue:push(table.concat(multi.pack(...), "\t"))
|
||||
end
|
||||
function c.error(err)
|
||||
c.queue:push("Error in <"..THREAD_NAME..":" .. THREAD_ID .. ">: ".. err)
|
||||
multi.error(err)
|
||||
end
|
||||
return c
|
||||
end
|
||||
|
||||
function THREAD.getThreads()
|
||||
--
|
||||
end
|
||||
|
||||
function THREAD.kill() -- trigger the lane destruction
|
||||
error("Thread was killed!\1")
|
||||
end
|
||||
|
||||
function THREAD.pushStatus(...)
|
||||
status_channel:push(multi.pack(...))
|
||||
end
|
||||
|
||||
function THREAD.sleep(n)
|
||||
love.timer.sleep(n)
|
||||
end
|
||||
|
||||
THREAD.hold = thread:newFunction(function(n)
|
||||
thread.hold(n)
|
||||
end, true)
|
||||
|
||||
function THREAD.setENV(env, name)
|
||||
GLOBAL[name or "__env"] = env
|
||||
end
|
||||
|
||||
function THREAD.getENV(name)
|
||||
return GLOBAL[name or "__env"]
|
||||
end
|
||||
|
||||
function THREAD.exposeENV(name)
|
||||
name = name or "__env"
|
||||
local env = THREAD.getENV(name)
|
||||
for i,v in pairs(env) do
|
||||
_G[i] = v
|
||||
end
|
||||
end
|
||||
|
||||
function THREAD.defer(func)
|
||||
table.insert(DEFER, func)
|
||||
end
|
||||
|
||||
function THREAD.sync()
|
||||
-- Maybe do something...
|
||||
end
|
||||
|
||||
return GLOBAL, THREAD, DEFER
|
||||
end
|
||||
|
||||
return threads
|
||||
return {
|
||||
init = function()
|
||||
return INIT()
|
||||
end
|
||||
}
|
||||
@@ -118,13 +118,13 @@ function multi:newSystemThreadedJobQueue(n)
|
||||
local rets
|
||||
link = c.OnJobCompleted(function(jid,...)
|
||||
if id==jid then
|
||||
rets = {...}
|
||||
rets = multi.pack(...)
|
||||
link:Destroy()
|
||||
end
|
||||
end)
|
||||
return thread.hold(function()
|
||||
if rets then
|
||||
return unpack(rets) or multi.NIL
|
||||
return multi.unpack(rets) or multi.NIL
|
||||
end
|
||||
end)
|
||||
end,holup),name
|
||||
@@ -134,7 +134,7 @@ function multi:newSystemThreadedJobQueue(n)
|
||||
thread.yield()
|
||||
local dat = c.queueReturn:pop()
|
||||
if dat then
|
||||
c.OnJobCompleted:Fire(unpack(dat))
|
||||
c.OnJobCompleted:Fire(multi.unpack(dat))
|
||||
end
|
||||
end
|
||||
end)
|
||||
@@ -152,6 +152,7 @@ function multi:newSystemThreadedJobQueue(n)
|
||||
local lastProc = clock()
|
||||
local queueAll = lovr.thread.getChannel("__JobQueue_"..jqc.."_queueAll")
|
||||
local registry = {}
|
||||
_G["__QR"] = queueReturn
|
||||
setmetatable(_G,{__index = funcs})
|
||||
thread:newThread("startUp",function()
|
||||
while true do
|
||||
@@ -177,7 +178,7 @@ function multi:newSystemThreadedJobQueue(n)
|
||||
lastProc = os.clock()
|
||||
local name = table.remove(dat,1)
|
||||
local id = table.remove(dat,1)
|
||||
local tab = {funcs[name](unpack(dat))}
|
||||
local tab = {funcs[name](multi.unpack(dat))}
|
||||
table.insert(tab,1,id)
|
||||
queueReturn:push(tab)
|
||||
end
|
||||
|
||||
@@ -36,12 +36,14 @@ __THREADNAME__=table.remove(__IMPORTS,1)
|
||||
stab = THREAD.createStaticTable(__THREADNAME__)
|
||||
GLOBAL = THREAD.getGlobal()
|
||||
multi, thread = require("multi").init()
|
||||
stab["returns"] = {THREAD.loadDump(__FUNC__)(unpack(__IMPORTS))}
|
||||
stab["returns"] = {THREAD.loadDump(__FUNC__)(multi.unpack(__IMPORTS))}
|
||||
]]
|
||||
local multi, thread = require("multi.compat.lovr2d"):init()
|
||||
local THREAD = {}
|
||||
__THREADID__ = 0
|
||||
__THREADNAME__ = "MainThread"
|
||||
_G.THREAD_NAME = "MAIN_THREAD"
|
||||
_G.THREAD_ID = 0
|
||||
multi.integration={}
|
||||
multi.integration.lovr2d={}
|
||||
local THREAD = require("multi.integration.lovrManager.threads")
|
||||
@@ -58,7 +60,7 @@ function THREAD:newFunction(func,holup)
|
||||
if t.stab["returns"] then
|
||||
local dat = t.stab.returns
|
||||
t.stab.returns = nil
|
||||
return unpack(dat)
|
||||
return multi.unpack(dat)
|
||||
end
|
||||
end)
|
||||
end,holup)()
|
||||
@@ -74,16 +76,23 @@ function multi:newSystemThread(name,func,...)
|
||||
GLOBAL["__THREAD_"..c.ID] = {ID=c.ID,Name=c.name,Thread=c.thread}
|
||||
GLOBAL["__THREAD_COUNT"] = THREAD_ID
|
||||
THREAD_ID=THREAD_ID+1
|
||||
|
||||
if self.isActor then
|
||||
self:create(c)
|
||||
else
|
||||
multi.create(multi, c)
|
||||
end
|
||||
|
||||
return c
|
||||
end
|
||||
THREAD.newSystemThread = multi.newSystemThread
|
||||
function lovr.threaderror(thread, errorstr)
|
||||
print("Thread error!\n"..errorstr)
|
||||
multi.print("Thread error!\n"..errorstr)
|
||||
end
|
||||
multi.integration.GLOBAL = GLOBAL
|
||||
multi.integration.THREAD = THREAD
|
||||
require("multi.integration.lovrManager.extensions")
|
||||
print("Integrated lovr Threading!")
|
||||
multi.print("Integrated lovr Threading!")
|
||||
return {init=function()
|
||||
return GLOBAL,THREAD
|
||||
end}
|
||||
@@ -156,7 +156,7 @@ function threads.getConsole()
|
||||
local c = {}
|
||||
c.queue = lovr.thread.getChannel("__CONSOLE__")
|
||||
function c.print(...)
|
||||
c.queue:push{...}
|
||||
c.queue:push(multi.pack(...))
|
||||
end
|
||||
function c.error(err)
|
||||
c.queue:push{"ERROR in <"..__THREADNAME__..">: "..err,__THREADID__}
|
||||
@@ -174,7 +174,7 @@ if not ISTHREAD then
|
||||
dat = queue:pop()
|
||||
if dat then
|
||||
lastproc = clock()
|
||||
print(unpack(dat))
|
||||
print(multi.unpack(dat))
|
||||
end
|
||||
if clock()-lastproc>2 then
|
||||
thread.sleep(.1)
|
||||
|
||||
@@ -35,7 +35,7 @@ local function _INIT(luvitThread, timer)
|
||||
end
|
||||
-- Step 1 get setup threads on luvit... Sigh how do i even...
|
||||
local multi, thread = require("multi").init()
|
||||
isMainThread = true
|
||||
multi.isMainThread = true
|
||||
function multi:canSystemThread()
|
||||
return true
|
||||
end
|
||||
@@ -107,7 +107,7 @@ local function _INIT(luvitThread, timer)
|
||||
local c = {}
|
||||
local __self = c
|
||||
c.name = name
|
||||
c.Type = "sthread"
|
||||
c.Type = multi.STHREAD
|
||||
c.thread = {}
|
||||
c.func = string.dump(func)
|
||||
function c:kill()
|
||||
|
||||
@@ -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
|
||||
@@ -0,0 +1,213 @@
|
||||
-- Advanced process management. Mutates the multi namespace
|
||||
local multi, thread = require("multi"):init()
|
||||
local ok, chronos = pcall(require, "chronos") -- hpc
|
||||
|
||||
if not ok then chronos = nil end
|
||||
|
||||
-- This is an integration, we cannot directly access locals that are in the main file.
|
||||
|
||||
local PList = {
|
||||
multi.Priority_Core,
|
||||
multi.Priority_Very_High,
|
||||
multi.Priority_High,
|
||||
multi.Priority_Above_Normal,
|
||||
multi.Priority_Normal,
|
||||
multi.Priority_Below_Normal,
|
||||
multi.Priority_Low,
|
||||
multi.Priority_Very_Low,
|
||||
multi.Priority_Idle
|
||||
}
|
||||
|
||||
-- Restructered these functions since they rely on local variables from the core library
|
||||
|
||||
local mainloop = multi.mainloopRef
|
||||
local mainloop_p = multi.mainloop_p
|
||||
local uManagerRef = multi.uManagerRef
|
||||
local uManagerRefP = multi.uManagerRefP1
|
||||
|
||||
local PROFILE_COUNT = 5
|
||||
|
||||
-- self:setCurrentProcess() a bit slower than using the local var, but there isn't another option
|
||||
|
||||
local priorityManager = multi:newProcessor("Priority Manager", true)
|
||||
priorityManager.newThread = function() multi.warn("You cannot spawn threads on the priority manager!") end
|
||||
|
||||
priorityManager.setPriorityScheme = function() multi.warn("You cannot set priority on the priorityManager!") end
|
||||
|
||||
local function average(t)
|
||||
local sum = 0
|
||||
for _,v in pairs(t) do
|
||||
sum = sum + v
|
||||
end
|
||||
return sum / #t
|
||||
end
|
||||
|
||||
local function getPriority(obj)
|
||||
local avg = average(obj.__profiling)
|
||||
if avg < 0.0002 then
|
||||
return PList[1]
|
||||
elseif avg < 0.0004 then
|
||||
return PList[2]
|
||||
elseif avg < 0.0008 then
|
||||
return PList[3]
|
||||
elseif avg < 0.001 then
|
||||
return PList[4]
|
||||
elseif avg < 0.0025 then
|
||||
return PList[5]
|
||||
elseif avg < 0.005 then
|
||||
return PList[6]
|
||||
elseif avg < 0.008 then
|
||||
return PList[7]
|
||||
elseif avg < 0.01 then
|
||||
return PList[8]
|
||||
else
|
||||
return PList[9]
|
||||
end
|
||||
end
|
||||
|
||||
local start, stop
|
||||
|
||||
priorityManager.uManager = function(self)
|
||||
-- proc.run already checks if the processor is active
|
||||
self:setCurrentProcess()
|
||||
local Loop=self.Mainloop
|
||||
local ctask
|
||||
for _D=#Loop,1,-1 do
|
||||
ctask = Loop[_D]
|
||||
ctask:setCurrentTask()
|
||||
start = chronos.nanotime()
|
||||
if ctask:Act() then
|
||||
stop = chronos.nanotime()
|
||||
if ctask.__profiling then
|
||||
table.insert(ctask.__profiling, stop - start)
|
||||
end
|
||||
if ctask.__profiling and #ctask.__profiling == PROFILE_COUNT then
|
||||
ctask:setPriority(getPriority(ctask))
|
||||
ctask:reallocate(ctask.__restoreProc)
|
||||
ctask.__restoreProc = nil
|
||||
ctask.__profiling = nil
|
||||
end
|
||||
end
|
||||
self:setCurrentProcess()
|
||||
end
|
||||
end
|
||||
|
||||
local function processHook(obj, proc)
|
||||
if obj.Type == multi.registerType("process", "processes") or not(obj.IsAnActor) then return end
|
||||
obj.__restoreProc = proc
|
||||
obj.__profiling = {}
|
||||
obj:reallocate(priorityManager)
|
||||
end
|
||||
|
||||
local function init()
|
||||
local registry = {}
|
||||
|
||||
multi.priorityScheme = {
|
||||
RoundRobin = "RoundRobin",
|
||||
PriorityBased = "PriorityBased",
|
||||
TimeBased = "TimeBased"
|
||||
}
|
||||
|
||||
function multi:setProfilerCount(count)
|
||||
PROFILE_COUNT = count
|
||||
end
|
||||
|
||||
function multi:recalibrate()
|
||||
if self.__processConn then
|
||||
local items = self.Mainloop
|
||||
for i,v in pairs(items) do
|
||||
processHook(v, self)
|
||||
end
|
||||
else
|
||||
multi.error("Cannot recalibrate the priority if not using Time based mangement!")
|
||||
end
|
||||
end
|
||||
|
||||
function multi:isRegistredScheme(scheme)
|
||||
return registry[name] ~= nil
|
||||
end
|
||||
|
||||
function multi:getRegisteredScheme(scheme)
|
||||
return registry[name].mainloop, registry[name].umanager, registry[name].condition
|
||||
end
|
||||
|
||||
local empty_func = function() return true end
|
||||
function multi:registerScheme(name,options)
|
||||
local mainloop = options.mainloop or multi.error("You must provide a mainloop option when registring a scheme!")
|
||||
local umanager = options.umanager or multi.error("You must provide a umanager option when registring a scheme!")
|
||||
|
||||
if not options.condition then
|
||||
multi.warn("You might want to use condition when registring a scheme! A function that returns true has been auto generated for you!")
|
||||
end
|
||||
|
||||
local condition = options.condition or empty_func
|
||||
|
||||
if registry[name] and not registry[name].static then
|
||||
multi.warn("A scheme named: \"" .. name .. "\" has already been registred, overriting!")
|
||||
else
|
||||
multi.error("A scheme named: \"" .. name .. "\" has already been registred!")
|
||||
end
|
||||
|
||||
registry[name] = {
|
||||
mainloop = mainloop,
|
||||
umanager = umanger,
|
||||
condition = condition,
|
||||
static = options.static or false
|
||||
}
|
||||
|
||||
multi.priorityScheme[name] = name
|
||||
|
||||
return true
|
||||
end
|
||||
|
||||
function multi:setPriorityScheme(scheme)
|
||||
|
||||
if not self.Type == multi.registerType("process", "processes") or not self.Type == multi.registerType("rootprocess") then
|
||||
multi.warn("You should only invoke setPriorityScheme on a processor object!")
|
||||
end
|
||||
|
||||
if scheme == multi.priorityScheme.RoundRobin then
|
||||
if self.__processConn then self.OnObjectCreated:Unconnect(self.__processConn) self.__processConn = nil end
|
||||
self.mainloop = mainloop
|
||||
self.uManager = uManagerRef
|
||||
elseif scheme == multi.priorityScheme.PriorityBased then
|
||||
if self.__processConn then self.OnObjectCreated:Unconnect(self.__processConn) self.__processConn = nil end
|
||||
self.mainloop = mainloop_p
|
||||
self.uManager = uManagerRefP
|
||||
elseif scheme == multi.priorityScheme.TimeBased then
|
||||
if not chronos then return multi.warn("Unable to use TimeBased Priority without the chronos library!") end
|
||||
if self.__processConn then multi.warn("Already enabled TimeBased Priority!") end
|
||||
self.__processConn = self.OnObjectCreated(processHook)
|
||||
self.mainloop = mainloop_p
|
||||
self.uManager = uManagerRefP
|
||||
elseif self:isRegistredScheme(scheme) then
|
||||
local mainloop, umanager, condition = self:getRegisteredScheme(scheme)
|
||||
if condition() then
|
||||
self.mainloop = mainloop
|
||||
self.uManager = umanager
|
||||
end
|
||||
else
|
||||
self.error("Invalid priority scheme selected!")
|
||||
end
|
||||
|
||||
end
|
||||
end
|
||||
|
||||
local function init_chronos()
|
||||
-- Let's implement a higher precision clock
|
||||
multi.setClock(chronos.nanotime) -- This is also in .000 format. So a plug and play works.
|
||||
thread:newThread("System Priority Manager", function()
|
||||
while true do
|
||||
thread.yield()
|
||||
priorityManager.run()
|
||||
end
|
||||
end)
|
||||
end
|
||||
|
||||
if chronos then
|
||||
init_chronos()
|
||||
else
|
||||
multi.warn("In order to have time based priority management, you need to install the chronos library!")
|
||||
end
|
||||
|
||||
init()
|
||||
@@ -0,0 +1,221 @@
|
||||
--[[
|
||||
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 = {}
|
||||
c.data = {}
|
||||
c.Type = multi.registerType("s_queue")
|
||||
function c:push(v)
|
||||
table.insert(self.data,v)
|
||||
end
|
||||
function c:pop()
|
||||
return table.remove(self.data,1)
|
||||
end
|
||||
function c:peek()
|
||||
return self.data[1]
|
||||
end
|
||||
function c:init()
|
||||
return self
|
||||
end
|
||||
function c:Hold(opt)
|
||||
if opt.peek then
|
||||
return thread.hold(function()
|
||||
return self:peek()
|
||||
end)
|
||||
else
|
||||
return thread.hold(function()
|
||||
return self:pop()
|
||||
end)
|
||||
end
|
||||
end
|
||||
GLOBAL[name or "_"] = c
|
||||
return c
|
||||
end
|
||||
|
||||
function multi:newSystemThreadedTable(name)
|
||||
local c = {}
|
||||
c.Type = multi.registerType("s_table")
|
||||
function c:init()
|
||||
return self
|
||||
end
|
||||
function c:Hold(opt)
|
||||
if opt.key then
|
||||
return thread.hold(function()
|
||||
return self.tab[opt.key]
|
||||
end)
|
||||
else
|
||||
multi.error("Must provide a key to check opt.key = 'key'")
|
||||
end
|
||||
end
|
||||
GLOBAL[name or "_"] = c
|
||||
return c
|
||||
end
|
||||
|
||||
local setfenv = multi.isolateFunction
|
||||
|
||||
local jqc = 1
|
||||
function multi:newSystemThreadedJobQueue(n)
|
||||
local c = {}
|
||||
|
||||
c.cores = n or THREAD.getCores()
|
||||
c.registerQueue = {}
|
||||
c.Type = multi.registerType("s_jobqueue")
|
||||
c.funcs = multi:newSystemThreadedTable("__JobQueue_"..jqc.."_table")
|
||||
c.queue = multi:newSystemThreadedQueue("__JobQueue_"..jqc.."_queue")
|
||||
c.queueReturn = multi:newSystemThreadedQueue("__JobQueue_"..jqc.."_queueReturn")
|
||||
c.queueAll = multi:newSystemThreadedQueue("__JobQueue_"..jqc.."_queueAll")
|
||||
c.id = 0
|
||||
c.OnJobCompleted = multi:newConnection()
|
||||
|
||||
local allfunc = 0
|
||||
|
||||
function c:doToAll(func)
|
||||
for i = 1, self.cores do
|
||||
self.queueAll:push({allfunc, func})
|
||||
end
|
||||
allfunc = allfunc + 1
|
||||
end
|
||||
function c:registerFunction(name, func)
|
||||
if self.funcs[name] then
|
||||
multi.error("A function by the name "..name.." has already been registered!")
|
||||
end
|
||||
self.funcs[name] = func
|
||||
end
|
||||
function c:pushJob(name,...)
|
||||
self.id = self.id + 1
|
||||
self.queue:push{name,self.id,...}
|
||||
return self.id
|
||||
end
|
||||
function c:isEmpty()
|
||||
return queueJob:peek()==nil
|
||||
end
|
||||
local nFunc = 0
|
||||
function c:newFunction(name,func,holup) -- This registers with the queue
|
||||
if type(name)=="function" then
|
||||
holup = func
|
||||
func = name
|
||||
name = "JQ_Function_"..nFunc
|
||||
end
|
||||
nFunc = nFunc + 1
|
||||
c:registerFunction(name,func)
|
||||
return thread:newFunction(function(...)
|
||||
local id = c:pushJob(name,...)
|
||||
local link
|
||||
local rets
|
||||
link = c.OnJobCompleted(function(jid,...)
|
||||
if id==jid then
|
||||
rets = multi.pack(...)
|
||||
end
|
||||
end)
|
||||
return thread.hold(function()
|
||||
if rets then
|
||||
return multi.unpack(rets) or multi.NIL
|
||||
end
|
||||
end)
|
||||
end,holup),name
|
||||
end
|
||||
thread:newThread("jobManager",function()
|
||||
while true do
|
||||
thread.yield()
|
||||
local dat = c.queueReturn:pop()
|
||||
if dat then
|
||||
c.OnJobCompleted:Fire(multi.unpack(dat))
|
||||
end
|
||||
end
|
||||
end)
|
||||
for i=1,c.cores do
|
||||
multi:newSystemThread("STJQ_"..multi.randomString(8),function(jqc)
|
||||
local GLOBAL, THREAD = require("multi.integration.pseudoManager"):init()
|
||||
local multi, thread = require("multi"):init()
|
||||
local clock = os.clock
|
||||
local funcs = THREAD.waitFor("__JobQueue_"..jqc.."_table")
|
||||
local queue = THREAD.waitFor("__JobQueue_"..jqc.."_queue")
|
||||
local queueReturn = THREAD.waitFor("__JobQueue_"..jqc.."_queueReturn")
|
||||
local queueAll = THREAD.waitFor("__JobQueue_"..jqc.."_queueAll")
|
||||
local registry = {}
|
||||
_G["__QR"] = queueReturn
|
||||
setmetatable(_G,{__index = funcs})
|
||||
thread:newThread("startUp",function()
|
||||
while true do
|
||||
thread.yield()
|
||||
local all = queueAll:peek()
|
||||
if all and not registry[all[1]] then
|
||||
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
|
||||
queueAll:pop()[2]()
|
||||
end
|
||||
local dat = thread.hold(queue)
|
||||
if dat then
|
||||
multi:newThread("JobSubRunner",function()
|
||||
local name = table.remove(dat,1)
|
||||
local id = table.remove(dat,1)
|
||||
local tab = {multi.isolateFunction(funcs[name],_G)(multi.unpack(dat))}
|
||||
table.insert(tab,1,id)
|
||||
queueReturn:push(tab)
|
||||
end)
|
||||
end
|
||||
end
|
||||
end)
|
||||
multi:mainloop()
|
||||
end, jqc)
|
||||
end
|
||||
|
||||
function c:Hold(opt)
|
||||
return thread.hold(self.OnJobCompleted)
|
||||
end
|
||||
|
||||
jqc = jqc + 1
|
||||
|
||||
self:create(c)
|
||||
|
||||
return c
|
||||
end
|
||||
|
||||
function multi:newSystemThreadedConnection(name)
|
||||
local conn = multi:newConnection()
|
||||
conn.init = function(self) return self end
|
||||
GLOBAL[name or "_"] = conn
|
||||
return conn
|
||||
end
|
||||
|
||||
require("multi.integration.sharedExtensions")
|
||||
@@ -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()
|
||||
@@ -31,8 +33,12 @@ if multi.integration then
|
||||
end
|
||||
}
|
||||
end
|
||||
multi.isMainThread = true
|
||||
local activator = require("multi.integration.pseudoManager.threads")
|
||||
local GLOBAL, THREAD = activator.init(thread)
|
||||
|
||||
local GLOBAL, THREAD = require("multi.integration.pesudoManager.threads").init(thread)
|
||||
_G.THREAD_NAME = "MAIN_THREAD"
|
||||
_G.THREAD_ID = 0
|
||||
|
||||
function multi:canSystemThread() -- We are emulating system threading
|
||||
return true
|
||||
@@ -50,29 +56,44 @@ local function split(str)
|
||||
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]]
|
||||
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,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 = {
|
||||
|
||||
function multi:newSystemThread(name, func, ...)
|
||||
local env
|
||||
env = {
|
||||
GLOBAL = GLOBAL,
|
||||
THREAD = THREAD,
|
||||
THREAD_NAME = name,
|
||||
__THREADNAME__ = name,
|
||||
THREAD_NAME = tostring(name),
|
||||
__THREADNAME__ = tostring(name),
|
||||
THREAD_ID = id,
|
||||
thread = thread
|
||||
thread = thread,
|
||||
multi = multi,
|
||||
}
|
||||
|
||||
for i = 1,#tab do
|
||||
env[tab[i]] = _G[tab[i]]
|
||||
|
||||
for i, v in pairs(_G) do
|
||||
if not(env[i]) and not(i == "_G") and not(i == "local_global") then
|
||||
env[i] = v
|
||||
else
|
||||
multi.warn("skipping:",i)
|
||||
end
|
||||
end
|
||||
|
||||
local th = thread:newISOThread(name,func,env,...)
|
||||
if GLOBAL["__env"] then
|
||||
for i,v in pairs(GLOBAL["__env"]) do
|
||||
env[i] = v
|
||||
end
|
||||
end
|
||||
|
||||
env._G = env
|
||||
|
||||
local GLOBAL, THREAD = activator.init(thread, env)
|
||||
|
||||
local th = pseudoProcessor:newISOThread(name, func, env, ...)
|
||||
th.Type = multi.registerType("s_thread", "pseudoThreads")
|
||||
th.OnError(multi.error)
|
||||
|
||||
id = id + 1
|
||||
|
||||
@@ -86,14 +107,15 @@ THREAD.newSystemThread = multi.newSystemThread
|
||||
function THREAD:newFunction(func,holdme)
|
||||
return thread:newFunctionBase(function(...)
|
||||
return multi:newSystemThread("TempSystemThread",func,...)
|
||||
end,holdme)()
|
||||
end, holdme, multi.registerType("s_function", "pseudoFunctions"))()
|
||||
end
|
||||
|
||||
multi.print("Integrated Pesudo Threading!")
|
||||
multi.integration = {} -- for module creators
|
||||
multi.integration.GLOBAL = GLOBAL
|
||||
multi.integration.THREAD = THREAD
|
||||
require("multi.integration.pesudoManager.extensions")
|
||||
require("multi.integration.pseudoManager.extensions")
|
||||
require("multi.integration.sharedExtensions")
|
||||
return {
|
||||
init = function()
|
||||
return GLOBAL, THREAD
|
||||
@@ -33,6 +33,7 @@ end
|
||||
local function INIT(thread)
|
||||
local THREAD = {}
|
||||
local GLOBAL = {}
|
||||
|
||||
THREAD.Priority_Core = 3
|
||||
THREAD.Priority_High = 2
|
||||
THREAD.Priority_Above_Normal = 1
|
||||
@@ -80,31 +81,42 @@ local function INIT(thread)
|
||||
|
||||
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
|
||||
|
||||
THREAD.defer = thread.defer
|
||||
|
||||
function THREAD.setENV(env, name)
|
||||
name = name or "__env"
|
||||
GLOBAL[name] = env
|
||||
end
|
||||
|
||||
function THREAD.getENV(name)
|
||||
name = name or "__env"
|
||||
return GLOBAL[name]
|
||||
end
|
||||
|
||||
function THREAD.exposeENV(name)
|
||||
name = name or "__env"
|
||||
local env = THREAD.getENV(name)
|
||||
for i,v in pairs(env) do
|
||||
-- This may need to be reworked!
|
||||
local_global[i] = v
|
||||
end
|
||||
end
|
||||
|
||||
function THREAD.sync()
|
||||
thread.sleep(.5)
|
||||
end
|
||||
|
||||
return GLOBAL, THREAD
|
||||
end
|
||||
|
||||
return {init = function(thread)
|
||||
return INIT(thread)
|
||||
return {init = function(thread, global)
|
||||
return INIT(thread, global)
|
||||
end}
|
||||
@@ -0,0 +1,336 @@
|
||||
--[[ todo finish the targeted job!
|
||||
MIT License
|
||||
|
||||
Copyright (c) 2023 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()
|
||||
|
||||
-- Returns a handler that allows a user to interact with an object on another thread!
|
||||
-- Create on the thread that you want to interact with, send over the handle
|
||||
|
||||
function multi:chop(obj)
|
||||
if not _G["UIDS"] then
|
||||
_G["UIDS"] = {}
|
||||
end
|
||||
local multi, thread = require("multi"):init()
|
||||
local list = {[0] = multi.randomString(12)}
|
||||
_G[list[0]] = obj
|
||||
for i,v in pairs(obj) do
|
||||
if type(v) == "function" or type(v) == "table" and v.Type == multi.registerType("s_function") then
|
||||
table.insert(list, i)
|
||||
elseif type(v) == "table" and v.Type == multi.registerType("connector", "connections") then
|
||||
table.insert(list, {i, multi:newProxy(multi:chop(v)):init()})
|
||||
end
|
||||
end
|
||||
return list
|
||||
end
|
||||
|
||||
function multi:newProxy(list)
|
||||
|
||||
local c = {}
|
||||
|
||||
c.name = multi.randomString(12)
|
||||
c.is_init = false
|
||||
local multi, thread = nil, nil
|
||||
function c:init()
|
||||
local multi, thread = nil, nil
|
||||
local function copy(obj)
|
||||
if type(obj) ~= 'table' then return obj end
|
||||
local res = {}
|
||||
for k, v in pairs(obj) do res[copy(k)] = copy(v) end
|
||||
return res
|
||||
end
|
||||
if not(self.is_init) then
|
||||
THREAD.sync()
|
||||
self.is_init = true
|
||||
local multi, thread = require("multi"):init()
|
||||
self.proxy_link = "PL" .. multi.randomString(12)
|
||||
|
||||
if multi.integration then
|
||||
GLOBAL = multi.integration.GLOBAL
|
||||
THREAD = multi.integration.THREAD
|
||||
end
|
||||
|
||||
GLOBAL[self.proxy_link] = self
|
||||
|
||||
local function check()
|
||||
return self.send:pop()
|
||||
end
|
||||
|
||||
self.send = multi:newSystemThreadedQueue(self.name.."_S"):init()
|
||||
self.recv = multi:newSystemThreadedQueue(self.name.."_R"):init()
|
||||
self.funcs = list
|
||||
self._funcs = copy(list)
|
||||
self.Type = multi.registerType("proxy", "proxies")
|
||||
self.TID = THREAD_ID
|
||||
|
||||
thread:newThread("Proxy_Handler_" .. multi.randomString(4), function()
|
||||
while true do
|
||||
local data = thread.hold(check)
|
||||
if data then
|
||||
-- Let's not hold the main threadloop
|
||||
thread:newThread("Temp_Thread", function()
|
||||
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
|
||||
ret = {_G[list[0]][func](multi.unpack(data))}
|
||||
end
|
||||
|
||||
for i = 1,#ret do
|
||||
if type(ret[i]) == "table" and ret[i].Type ~= nil and ret[i].Type ~= multi.registerType("proxy", "proxies") then
|
||||
ret[i] = "\1PARENT_REF"
|
||||
end
|
||||
if type(ret[i]) == "table" and getmetatable(ret[i]) then
|
||||
setmetatable(ret[i],nil) -- remove that metatable, we do not need it on the other side!
|
||||
end
|
||||
if ret[i] == _G[list[0]] then
|
||||
-- We cannot return itself, that return can contain bad values.
|
||||
ret[i] = "\1SELF_REF"
|
||||
end
|
||||
end
|
||||
table.insert(ret, 1, func)
|
||||
self.recv:push(ret)
|
||||
end)
|
||||
end
|
||||
end
|
||||
end)
|
||||
return self
|
||||
else
|
||||
THREAD.sync()
|
||||
if not self.funcs then return self end
|
||||
local function copy(obj)
|
||||
if type(obj) ~= 'table' then return obj end
|
||||
local res = {}
|
||||
for k, v in pairs(obj) do res[copy(k)] = copy(v) end
|
||||
return res
|
||||
end
|
||||
local multi, thread = require("multi"):init()
|
||||
local me = self
|
||||
local funcs = copy(self.funcs)
|
||||
if multi.integration then
|
||||
GLOBAL = multi.integration.GLOBAL
|
||||
THREAD = multi.integration.THREAD
|
||||
end
|
||||
self.send = THREAD.waitFor(self.name.."_S"):init()
|
||||
self.recv = THREAD.waitFor(self.name.."_R"):init()
|
||||
self.Type = multi.registerType("proxy", "proxies")
|
||||
for _,v in pairs(funcs) do
|
||||
if type(v) == "table" then
|
||||
-- We have a connection
|
||||
v[2]:init(proc_name)
|
||||
self[v[1]] = v[2]
|
||||
v[2].Parent = self
|
||||
setmetatable(v[2],getmetatable(multi:newConnection()))
|
||||
else
|
||||
self[v] = thread:newFunction(function(self,...)
|
||||
if self == me then
|
||||
me.send:push({v, true, ...})
|
||||
else
|
||||
me.send:push({v, false, ...})
|
||||
end
|
||||
return thread.hold(function()
|
||||
local data = me.recv:peek()
|
||||
if data and data[1] == v then
|
||||
me.recv:pop()
|
||||
table.remove(data, 1)
|
||||
for i=1,#data do
|
||||
if data[i] == "\1SELF_REF" then
|
||||
data[i] = me
|
||||
elseif data[i] == "\1PARENT_REF" then
|
||||
data[i] = me.Parent
|
||||
end
|
||||
end
|
||||
return multi.unpack(data)
|
||||
end
|
||||
end)
|
||||
end, true)
|
||||
end
|
||||
end
|
||||
return self
|
||||
end
|
||||
end
|
||||
function c:getTransferable()
|
||||
local cp = {}
|
||||
local multi, thread = require("multi"):init()
|
||||
local function copy(obj)
|
||||
if type(obj) ~= 'table' then return obj end
|
||||
local res = {}
|
||||
for k, v in pairs(obj) do res[copy(k)] = copy(v) end
|
||||
return res
|
||||
end
|
||||
cp.is_init = true
|
||||
cp.proxy_link = self.proxy_link
|
||||
cp.name = self.name
|
||||
cp.funcs = copy(self._funcs)
|
||||
cp.init = function(self)
|
||||
local multi, thread = require("multi"):init()
|
||||
if multi.integration then
|
||||
GLOBAL = multi.integration.GLOBAL
|
||||
THREAD = multi.integration.THREAD
|
||||
end
|
||||
local proxy = THREAD.waitFor(self.proxy_link)
|
||||
proxy.funcs = self.funcs
|
||||
return proxy:init()
|
||||
end
|
||||
return cp
|
||||
end
|
||||
self:create(c)
|
||||
return c
|
||||
end
|
||||
|
||||
local targets = {}
|
||||
local references = {}
|
||||
|
||||
local jid = -1
|
||||
function multi:newSystemThreadedProcessor(cores)
|
||||
|
||||
local name = "STP_"..multi.randomString(4) -- set a random name if none was given.
|
||||
|
||||
local autoscale = autoscale or false -- Will scale up the number of cores that the process uses.
|
||||
local c = {}
|
||||
|
||||
setmetatable(c,{__index = multi})
|
||||
|
||||
c.Type = multi.registerType("s_process", "s_processes")
|
||||
c.threads = {}
|
||||
c.cores = cores or 8
|
||||
c.Name = name
|
||||
c.Mainloop = {}
|
||||
c.__count = 0
|
||||
c.processors = {}
|
||||
c.proc_list = {}
|
||||
c.OnObjectCreated = multi:newConnection()
|
||||
c.parent = self
|
||||
c.jobqueue = multi:newSystemThreadedJobQueue(c.cores)
|
||||
|
||||
function c:pushJob(ID, name, ...)
|
||||
local tq = THREAD.waitFor(self.Name .. "_target_tq_" .. ID):init()
|
||||
tq:push{name, jid, multi.pack(...)}
|
||||
jid = jid - 1
|
||||
return jid + 1
|
||||
end
|
||||
|
||||
c.jobqueue:registerFunction("packObj",function(obj)
|
||||
local multi, thread = require("multi"):init()
|
||||
|
||||
local list = multi:chop(obj)
|
||||
obj.__link_name = list[0]
|
||||
|
||||
local proxy = multi:newProxy(list):init()
|
||||
|
||||
return proxy
|
||||
end)
|
||||
|
||||
c.spawnThread = c.jobqueue:newFunction("__spawnThread__", function(name, func, ...)
|
||||
local multi, thread = require("multi"):init()
|
||||
local obj = thread:newThread(name, func, ...)
|
||||
return packObj(obj)
|
||||
end, true)
|
||||
|
||||
c.spawnTask = c.jobqueue:newFunction("__spawnTask__", function(obj, func, ...)
|
||||
local multi, thread = require("multi"):init()
|
||||
local obj = multi[obj](multi, func, ...)
|
||||
return packObj(obj)
|
||||
end, true)
|
||||
|
||||
local implement = {
|
||||
"newLoop",
|
||||
"newTLoop",
|
||||
"newUpdater",
|
||||
"newEvent",
|
||||
"newAlarm",
|
||||
"newStep",
|
||||
"newTStep",
|
||||
"newService"
|
||||
}
|
||||
|
||||
for _, method in pairs(implement) do
|
||||
c[method] = thread:newFunction(function(self, ...)
|
||||
proxy = self.spawnTask(method, ...)
|
||||
proxy:init()
|
||||
references[proxy] = self
|
||||
return proxy
|
||||
end, true)
|
||||
end
|
||||
|
||||
function c:newThread(name, func, ...)
|
||||
proxy = self.spawnThread(name, func, ...):init(self.Name)
|
||||
references[proxy] = self
|
||||
table.insert(self.threads, proxy)
|
||||
return proxy
|
||||
end
|
||||
|
||||
function c:newFunction(func, holdme)
|
||||
return self.jobqueue:newFunction(func, holdme)
|
||||
end
|
||||
|
||||
function c:newSharedTable(name)
|
||||
if not name then multi.error("You must provide a name when creating a table!") end
|
||||
local tbl_name = "TABLE_"..multi.randomString(8)
|
||||
self.jobqueue:doToAll(function(tbl_name, interaction)
|
||||
_G[interaction] = THREAD.waitFor(tbl_name):init()
|
||||
end, tbl_name, name)
|
||||
return multi:newSystemThreadedTable(tbl_name):init()
|
||||
end
|
||||
|
||||
function c:getHandler()
|
||||
return function() end -- return empty function
|
||||
end
|
||||
|
||||
function c:getThreads()
|
||||
return self.threads
|
||||
end
|
||||
|
||||
function c:getFullName()
|
||||
return self.parent:getFullName() .. "." .. c.Name
|
||||
end
|
||||
|
||||
function c:getName()
|
||||
return self.Name
|
||||
end
|
||||
|
||||
function c.run()
|
||||
return self
|
||||
end
|
||||
|
||||
function c.isActive()
|
||||
return true
|
||||
end
|
||||
|
||||
function c.Start()
|
||||
return self
|
||||
end
|
||||
|
||||
function c.Stop()
|
||||
return self
|
||||
end
|
||||
|
||||
function c:Destroy()
|
||||
return false
|
||||
end
|
||||
|
||||
return c
|
||||
end
|
||||
|
||||
Reference in New Issue
Block a user