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

(1.8.4) Update

Browse Source
This commit is contained in:
server
2017-07-03 13:59:44 -04:00
parent 367982898a
commit 584e8be3a2
21 changed files with 1027 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/love2d Threading Example/multi/init.lua
+115 -7
View File
@@ -45,8 +45,8 @@ function print(...)
end
end
multi = {}
multi.Version="1.8.2"
multi._VERSION="1.8.2"
multi.Version="1.8.4"
multi._VERSION="1.8.4"
multi.stage='stable'
multi.__index = multi
multi.Mainloop={}
@@ -108,11 +108,13 @@ function multi:setThrestimed(n)
end
function multi:getLoad()
return multi:newFunction(function(self)
multi.scheduler:Pause()
local sample=#multi.Mainloop
local FFloadtest=0
multi:benchMark(multi.threstimed):OnBench(function(_,l3) FFloadtest=l3*(1/multi.threstimed) end)
self:hold(function() return FFloadtest~=0 end)
local val=FFloadtest/sample
multi.scheduler:Resume()
if val>multi.threshold then
return 0
else
@@ -617,7 +619,11 @@ function multi:hold(task)
env:OnEvent(function(envt) envt:Pause() envt.Active=false end)
while env.Active do
if love then
self.Parent:lManager()
if love.graphics then
self.Parent:lManager()
else
self.Parent:Do_Order()
end
else
self.Parent:Do_Order()
end
@@ -978,7 +984,110 @@ function multi:mainloop()
else
return "Already Running!"
end
--print("Did you call multi:Stop()? This method should not be used when using multi:mainloop() unless of course you wanted to stop it! you can restart the multi, by using multi:reboot() and calling multi:mainloop() again or by using multi:uManager()")
end
function multi:protectedMainloop()
multi:protect()
if not multi.isRunning then
multi.isRunning=true
for i=1,#self.Tasks do
self.Tasks[i](self)
end
rawset(self,'Start',self.clock())
while self.Active do
self:Do_Order()
end
else
return "Already Running!"
end
end
function multi:unprotectedMainloop()
multi:unProtect()
if not multi.isRunning then
multi.isRunning=true
for i=1,#self.Tasks do
self.Tasks[i](self)
end
rawset(self,'Start',self.clock())
while self.Active do
local Loop=self.Mainloop
_G.ID=0
for _D=#Loop,1,-1 do
if Loop[_D] then
if Loop[_D].Active then
Loop[_D].Id=_D
self.CID=_D
Loop[_D]:Act()
end
end
end
end
else
return "Already Running!"
end
end
function multi:prioritizedMainloop1()
multi:enablePriority()
if not multi.isRunning then
multi.isRunning=true
for i=1,#self.Tasks do
self.Tasks[i](self)
end
rawset(self,'Start',self.clock())
while self.Active do
local Loop=self.Mainloop
_G.ID=0
local PS=self
for _D=#Loop,1,-1 do
if Loop[_D] then
if (PS.PList[PS.PStep])%Loop[_D].Priority==0 then
if Loop[_D].Active then
Loop[_D].Id=_D
self.CID=_D
Loop[_D]:Act()
end
end
end
end
PS.PStep=PS.PStep+1
if PS.PStep>7 then
PS.PStep=1
end
end
else
return "Already Running!"
end
end
function multi:prioritizedMainloop2()
multi:enablePriority2()
if not multi.isRunning then
multi.isRunning=true
for i=1,#self.Tasks do
self.Tasks[i](self)
end
rawset(self,'Start',self.clock())
while self.Active do
local Loop=self.Mainloop
_G.ID=0
local PS=self
for _D=#Loop,1,-1 do
if Loop[_D] then
if (PS.PStep)%Loop[_D].Priority==0 then
if Loop[_D].Active then
Loop[_D].Id=_D
self.CID=_D
Loop[_D]:Act()
end
end
end
end
PS.PStep=PS.PStep+1
if PS.PStep>self.Priority_Idle then
PS.PStep=1
end
end
else
return "Already Running!"
end
end
function multi._tFunc(self,dt)
for i=1,#self.Tasks do
@@ -1494,7 +1603,7 @@ function multi:newThread(name,func)
end
multi:setDomainName("Threads")
multi:setDomainName("Globals")
multi.scheduler=multi:newUpdater()
multi.scheduler=multi:newLoop()
multi.scheduler.Type="scheduler"
function multi.scheduler:setStep(n)
self.skip=tonumber(n) or 24
@@ -1503,7 +1612,7 @@ multi.scheduler.skip=0
multi.scheduler.counter=0
multi.scheduler.Threads=multi:linkDomain("Threads")
multi.scheduler.Globals=multi:linkDomain("Globals")
multi.scheduler:OnUpdate(function(self)
multi.scheduler:OnLoop(function(self)
self.counter=self.counter+1
for i=#self.Threads,1,-1 do
ret={}
@@ -1557,7 +1666,6 @@ multi.scheduler:OnUpdate(function(self)
end
end
end)
multi.scheduler:setStep()
multi.scheduler:Pause()
multi.OnError=multi:newConnection()
function multi:newThreadedAlarm(name,set)
examples/love2d Threading Example/multi/integration/lanesManager.lua
+5 -8
View File
@@ -125,12 +125,9 @@ function multi:newSystemThread(name,func)
end)
return c
end
print("Intergrated Lanes!")
multi.intergration={} -- for module creators
multi.intergration.GLOBAL=GLOBAL
multi.intergration.THREAD=THREAD
multi.intergration.lanes={}
multi.intergration.lanes.GLOBAL=GLOBAL -- for module creators
multi.intergration.lanes.THREAD=THREAD -- for module creators
require("multi.intergration.shared.shared")
print("Integrated Lanes!")
multi.integration={} -- for module creators
multi.integration.GLOBAL=GLOBAL
multi.integration.THREAD=THREAD
require("multi.integration.shared.shared")
return {init=function() return GLOBAL,THREAD end}
examples/love2d Threading Example/multi/integration/loveManager.lua
+34 -39
View File
@@ -2,14 +2,14 @@ require("multi.compat.love2d")
function multi:canSystemThread()
return true
end
multi.intergration={}
multi.intergration.love2d={}
multi.intergration.love2d.ThreadBase=[[
multi.integration={}
multi.integration.love2d={}
multi.integration.love2d.ThreadBase=[[
__THREADNAME__=({...})[1]
require("love.filesystem")
require("love.system")
require("love.timer")
require("multi.all")
require("multi")
GLOBAL={}
setmetatable(GLOBAL,{
__index=function(t,k)
@@ -29,6 +29,7 @@ setmetatable(GLOBAL,{
function __sync__()
local data=__mythread__:pop()
while data do
love.timer.sleep(.001)
if type(data)=="string" then
local cmd,tp,name,d=data:match("(%S-) (%S-) (%S-) (.+)")
if name=="__DIEPLZ"..__THREADNAME__.."__" then
@@ -79,7 +80,7 @@ function resolveType(tp,d)
elseif tp=="bool" then
return (d=="true")
elseif tp=="function" then
return loadDump(d)
return loadDump("[==["..d.."]==]")
elseif tp=="table" then
return loadstring("return "..d)()
elseif tp=="nil" then
@@ -91,11 +92,11 @@ end
function resolveData(v)
local data=""
if type(v)=="table" then
data=ToStr(v)
return ToStr(v)
elseif type(v)=="function" then
data=dump(v)
return dump(v)
elseif type(v)=="string" or type(v)=="number" or type(v)=="bool" or type(v)=="nil" then
data=tostring(v)
return tostring(v)
end
return data
end
@@ -152,24 +153,16 @@ end
function sThread.hold(n)
repeat __sync__() until n()
end
updater=multi:newUpdater()
updater:OnUpdate(function(self)
local data=__mythread__:pop()
while data do
if type(data)=="string" then
local cmd,tp,name,d=data:match("(%S-) (%S-) (%S-) (.+)")
if name=="__DIEPLZ"..__THREADNAME__.."__" then
error("Thread: "..__THREADNAME__.." has been stopped!")
end
if cmd=="SYNC" then
__proxy__[name]=resolveType(tp,d)
end
else
__proxy__[name]=data
end
data=__mythread__:pop()
multi:newLoop(function(self)
self:Pause()
local ld=multi:getLoad()
self:Resume()
if ld<80 then
love.timer.sleep(.01)
end
end)
updater=multi:newUpdater()
updater:OnUpdate(__sync__)
func=loadDump([=[INSERT_USER_CODE]=])()
multi:mainloop()
]]
@@ -191,7 +184,7 @@ setmetatable(GLOBAL,{
end,
})
THREAD={} -- Allow main thread to interact with these objects as well
multi.intergration.love2d.mainChannel=love.thread.getChannel("__MainChan__")
multi.integration.love2d.mainChannel=love.thread.getChannel("__MainChan__")
function ToStr(val, name, skipnewlines, depth)
skipnewlines = skipnewlines or false
depth = depth or 0
@@ -228,7 +221,7 @@ function resolveType(tp,d)
elseif tp=="bool" then
return (d=="true")
elseif tp=="function" then
return loadDump(d)
return loadDump("[==["..d.."]==]")
elseif tp=="table" then
return loadstring("return "..d)()
elseif tp=="nil" then
@@ -240,11 +233,11 @@ end
function resolveData(v)
local data=""
if type(v)=="table" then
data=ToStr(v)
return ToStr(v)
elseif type(v)=="function" then
data=dump(v)
return dump(v)
elseif type(v)=="string" or type(v)=="number" or type(v)=="bool" or type(v)=="nil" then
data=tostring(v)
return tostring(v)
end
return data
end
@@ -274,10 +267,10 @@ function multi:newSystemThread(name,func) -- the main method
local c={}
c.name=name
c.ID=c.name.."<ID|"..randomString(8)..">"
c.thread=love.thread.newThread(multi.intergration.love2d.ThreadBase:gsub("INSERT_USER_CODE",dump(func)))
c.thread=love.thread.newThread(multi.integration.love2d.ThreadBase:gsub("INSERT_USER_CODE",dump(func)))
c.thread:start(c.ID)
function c:kill()
multi.intergration.GLOBAL["__DIEPLZ"..self.ID.."__"]="__DIEPLZ"..self.ID.."__"
multi.integration.GLOBAL["__DIEPLZ"..self.ID.."__"]="__DIEPLZ"..self.ID.."__"
end
return c
end
@@ -310,11 +303,11 @@ function THREAD.hold(n)
multi.OBJ_REF:Resume()
end
__channels__={}
multi.intergration.GLOBAL=GLOBAL
multi.intergration.THREAD=THREAD
multi.integration.GLOBAL=GLOBAL
multi.integration.THREAD=THREAD
updater=multi:newUpdater()
updater:OnUpdate(function(self)
local data=multi.intergration.love2d.mainChannel:pop()
local data=multi.integration.love2d.mainChannel:pop()
while data do
--print("MAIN:",data)
if type(data)=="string" then
@@ -342,19 +335,21 @@ updater:OnUpdate(function(self)
else
__proxy__[name]=data
end
data=multi.intergration.love2d.mainChannel:pop()
data=multi.integration.love2d.mainChannel:pop()
end
end)
require("multi.intergration.shared.shared")
print("Intergrated Love2d!")
require("multi.integration.shared.shared")
print("Integrated Love2d!")
return {
init=function(t)
if t then
if t.threadNamespace then
multi.intergration.love2d.ThreadBase:gsub("sThread",t.threadNamespace)
multi.integration.THREADNAME=t.threadNamespace
multi.integration.love2d.ThreadBase:gsub("sThread",t.threadNamespace)
end
if t.globalNamespace then
multi.intergration.love2d.ThreadBase:gsub("GLOBAL",t.globalNamespace)
multi.integration.GLOBALNAME=t.globalNamespace
multi.integration.love2d.ThreadBase:gsub("GLOBAL",t.globalNamespace)
end
end
return GLOBAL,THREAD
examples/love2d Threading Example/multi/integration/shared/shared.lua
+139 -8
View File
@@ -29,10 +29,16 @@ function multi:newSystemThreadedQueue(name) -- in love2d this will spawn a chann
function c:init() -- create an init function so we can mimic on bith love2d and lanes
self.chan=love.thread.getChannel(self.name) -- create channel by the name self.name
function self:push(v) -- push to the channel
self.chan:push(v)
self.chan:push({type(v),resolveData(v)})
end
function self:pop() -- pop from the channel
return self.chan:pop()
local tab=self.chan:pop()
if not tab then return end
return resolveType(tab[1],tab[2])
end
function self:peek()
local tp,d=unpack{self.chan:peek()}
return resolveType(tp,d)
end
GLOBAL[self.name]=self -- send the object to the thread through the global interface
return self -- return the object
@@ -49,20 +55,23 @@ function multi:newSystemThreadedQueue(name) -- in love2d this will spawn a chann
function c:pop() -- pop the queue
return ({self.linda:receive(0,"Q")})[2]
end
function c:peek()
return self.linda:get("Q")
end
function c:init() -- mimic the feature that love2d requires, so code can be consistent
return self
end
multi.intergration.GLOBAL[name]=c -- send the object to the thread through the global interface
multi.integration.GLOBAL[name]=c -- send the object to the thread through the global interface
end
return c
end
function multi:systemThreadedBenchmark(n,p)
n=n or 1
local cores=multi.intergration.THREAD.getCores()
local cores=multi.integration.THREAD.getCores()
local queue=multi:newSystemThreadedQueue("QUEUE")
multi.intergration.GLOBAL["__SYSTEMBENCHMARK__"]=n
local sThread=multi.intergration.THREAD
local GLOBAL=multi.intergration.GLOBAL
multi.integration.GLOBAL["__SYSTEMBENCHMARK__"]=n
local sThread=multi.integration.THREAD
local GLOBAL=multi.integration.GLOBAL
for i=1,cores do
multi:newSystemThread("STHREAD_BENCH",function()
require("multi")
@@ -170,7 +179,129 @@ function multi:newSystemThreadedTable(name)
})
return self
end
multi.intergration.GLOBAL[name]=c -- send the object to the thread through the global interface
multi.integration.GLOBAL[name]=c -- send the object to the thread through the global interface
end
return c
end
function multi:newSystemThreadedJobQueue(numOfCores)
local c={}
c.jobnum=1
c.cores=numOfCores or multi.integration.THREAD.getCores()
c.queueIN=multi:newSystemThreadedQueue("THREADED_JQ"):init()
c.queueOUT=multi:newSystemThreadedQueue("THREADED_JQO"):init()
c.REG=multi:newSystemThreadedTable("THREADED_JQ_F_REG"):init()
-- registerJob(name,func)
-- pushJob(...)
function c:registerJob(name,func)
self.REG[name]=func
end
function c:pushJob(name,...)
self.queueOUT:push({self.jobnum,name,...})
self.jobnum=self.jobnum+1
end
local GLOBAL=multi.integration.GLOBAL -- set up locals incase we are using lanes
local sThread=multi.integration.THREAD -- set up locals incase we are using lanes
GLOBAL["__JQ_COUNT__"]=c.cores
for i=1,c.cores do
multi:newSystemThread("System Threaded Job Queue Worker Thread #"..i,function()
require("multi")
__sleep__=.001
if love then -- lets make sure we don't reference upvalues if using love2d
GLOBAL=_G.GLOBAL
sThread=_G.sThread
__sleep__=.1
end
JQI=sThread.waitFor("THREADED_JQO"):init() -- Grab it
JQO=sThread.waitFor("THREADED_JQ"):init() -- Grab it
FGLOBAL=sThread.waitFor("THREADED_JQ_F_REG"):init() -- Grab it
sThread.sleep(.1) -- lets wait for things to work out
setmetatable(_G,{
__index=FGLOBAL
})
GLOBAL["THREADED_JQ"]=nil -- remove it
GLOBAL["THREADED_JQO"]=nil -- remove it
GLOBAL["THREADED_JQ_F_REG"]=nil -- remove it
multi:newLoop(function()
sThread.sleep(__sleep__) -- lets allow cpu time for other processes on our system!
local job=JQI:pop()
if job then
local ID=table.remove(job,1) -- return and remove
local name=table.remove(job,1) -- return and remove
local ret={FGLOBAL:waitFor(name)(unpack(job))} -- unpack the rest
JQO:push({ID,ret})
end
end)
multi:mainloop()
end)
end
c.OnJobCompleted=multi:newConnection()
c.updater=multi:newLoop(function(self)
local data=self.link.queueIN:pop()
while data do
if data then
self.link.OnJobCompleted:Fire(unpack(data))
end
data=self.link.queueIN:pop()
end
end)
c.updater.link=c
return c
end
if love then
if love.thread then
function multi:newSystemThreadedJobQueue(numOfCores)
local c={}
c.jobnum=1
c.cores=numOfCores or multi.integration.THREAD.getCores()
c.queueIN=multi:newSystemThreadedQueue("THREADED_JQ"):init()
c.queueOUT=multi:newSystemThreadedQueue("THREADED_JQO"):init()
function c:registerJob(name,func)
GLOBAL["__TJQ__"..name.."__"]=func
end
function c:pushJob(name,...)
self.queueOUT:push({self.jobnum,name,...})
self.jobnum=self.jobnum+1
end
local GLOBAL=multi.integration.GLOBAL -- set up locals incase we are using lanes
local sThread=multi.integration.THREAD -- set up locals incase we are using lanes
GLOBAL["__JQ_COUNT__"]=c.cores
for i=1,c.cores do
multi:newSystemThread("System Threaded Job Queue Worker Thread #"..i,function()
GLOBAL=_G.GLOBAL
sThread=_G.sThread
local JQI=sThread.waitFor("THREADED_JQO"):init() -- Grab it
local JQO=sThread.waitFor("THREADED_JQ"):init() -- Grab it
sThread.sleep(.1) -- lets wait for things to work out
setmetatable(_G,{
__index=function(t,k,v)
return GLOBAL["__TJQ__"..k.."__"]
end
})
GLOBAL["THREADED_JQ"]=nil -- remove it
GLOBAL["THREADED_JQO"]=nil -- remove it
multi:newLoop(function()
local job=JQI:pop()
if job then
local ID=table.remove(job,1) -- return and remove
local name=table.remove(job,1) -- return and remove
local ret={sThread.waitFor("__TJQ__"..name.."__")(unpack(job))} -- unpack the rest
JQO:push({ID,ret})
end
end)
end)
end
c.OnJobCompleted=multi:newConnection()
c.updater=multi:newLoop(function(self)
local data=self.link.queueIN:pop()
while data do
if data then
self.link.OnJobCompleted:Fire(unpack(data))
end
data=self.link.queueIN:pop()
end
end)
c.updater.link=c
return c
end
end
end