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

Reworking the networkManager

Browse Source 
some new features
This commit is contained in:
Ryan Ward 2019-12-30 20:16:27 -05:00
parent 71b56023bb
commit 9c74f6c265
22 changed files with 429 additions and 630 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
changes.md
View File

@ -1,10 +1,11 @@
# Changes
[TOC]
Update 14.0.0 Consistency, stability and some new features
Update 14.0.0 Consistency, Stability and Additions
-------------
Added:
- multi.init() -- Initlizes the library! Must be called for multiple files to have the same handle. Example below
- thread.holdFor(NUMBER sec, FUNCTION condition) -- Works like hold, but timesout when a certain amount of time has passed!
- multi.hold(function or number) -- It's back and uses context switching. Normal multi objs without threading will all be halted where threads will still run. If within a thread continue using thread.hold() and thread.sleep()
- thread.holdWithin(NUMBER; cycles,FUNCTION; condition) -- Holds until the condition is met! If the number of cycles passed is equal to cycles, hold will return a timeout error
**Note:** when hold has a timeout the first argument will return nil and the second atgument will be TIMEOUT, if not timed out hold will return the values from the conditions
- thread objects got an addition!
@ -14,8 +15,8 @@ Added:
- thread.run(function) -- Can only be used within a thread, creates another thread that can do work, but automatically returns whatever from the run function -- Use thread newfunctions for a more powerful version of thread.run()
- thread:newFunction(FUNCTION; func)
-- returns a function that gives you the option to wait or connect to the returns of the function.
-- func().wait() -- only works when within a coroutine based thread
-- func().connect() -- this can work outside of of a coroutine based thread
-- func().wait() -- waits for the function to return
-- func().connect() -- connects to the function finishing
-- func() -- If your function does not return anything you dont have to use wait or connect at all and the function will return instantly. You could also use wait() to hold until the function does it thing
-- If the created function encounters an error, it will return nil, the error message!
- special variable multi.NIL was added to allow error handling in threaded functions.
@ -82,12 +83,28 @@ Removed:
- multi:newCustomObject() -- No real use
Changed:
- multi connections connect function can now chain connections
```lua
package.path = "./?/init.lua;"..package.path
local multi, thread = require("multi").init()
test = multi:newConnection()
test(function(a)
print("test 1",a.Temp)
a.Temp = "No!"
end,function(a)
print("test 2",a.Temp)
a.Temp = "Maybe!"
end,function(a)
print("test 3",a.Temp)
end)
test:Fire({Temp="Yes!"})
```
- Ties in to the new function that has been added multi.init()
```lua
local multi, thread = require("multi").init() -- The require multi function still returns the multi object like before
```
- love/lanesManager system threading integration has been reworked. Faster and cleaner code! Consistant code as well
- l
Note: Using init allows you to get access to the thread handle. This was done because thread was modifying the global space as well as multi. I wanted to not modify the global space anymore.
internally most of your code can stay the same, you only need to change how the library is required. I do toy a bit with the global space, buy I use a variable name that is invalid as a variable name. The variable name is $multi. This is used internally to keep some records and maintain a clean space

2
multi/compat/love2d.lua
View File

@ -1,7 +1,7 @@
--[[
MIT License
Copyright (c) 2019 Ryan Ward
Copyright (c) 2020 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal

181
multi/init.lua
View File

@ -1,7 +1,7 @@
--[[
MIT License
Copyright (c) 2019 Ryan Ward
Copyright (c) 2020 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
@ -861,9 +861,13 @@ function multi:newConnection(protect,func)
self.func={}
return self
end
function c:connect(func,name,num)
local function conn_helper(self,func,name,num)
self.ID=self.ID+1
table.insert(self.func,num or #self.func+1,{func,self.ID})
if num then
table.insert(self.func,num,{func,self.ID})
else
table.insert(self.func,1,{func,self.ID})
end
local temp = {
Link=self.func,
func=func,
@ -908,6 +912,24 @@ function multi:newConnection(protect,func)
end
return temp
end
function c:connect(...)--func,name,num
local tab = {...}
local funcs={}
for i=1,#tab do
if type(tab[i])=="function" then
funcs[#funcs+1] = tab[i]
end
end
if #funcs>1 then
local ret = {}
for i=1,#funcs do
table.insert(ret,conn_helper(self,funcs[i]))
end
return ret
else
conn_helper(self,tab[1],tab[2],tab[3])
end
end
c.Connect=c.connect
c.GetConnection=c.getConnection
function c:tofile(path)
@ -1489,11 +1511,34 @@ function thread.waitFor(name)
thread.hold(function() return thread.get(name)~=nil end)
return thread.get(name)
end
function multi.hold(func)
local death = false
if type(func)=="function" then
local rets
multi:newThread("Hold_func",function()
rets = {thread.hold(func)}
death = true
end)
while not death do
multi.scheduler:Act()
end
return unpack(rets)
elseif type(func)=="number" then
multi:newThread("Hold_func",function()
thread.sleep(func)
death = true
end)
while not death do
multi.scheduler:Act()
end
end
end
function thread:newFunction(func)
local c = {Type = "tfunc"}
c.__call = function(self,...)
local rets, err
local function wait()
if thread.isThread() then
return thread.hold(function()
if err then
return multi.NIL, err
@ -1501,13 +1546,16 @@ function thread:newFunction(func)
return unpack(rets)
end
end)
else
while not rets do
multi.scheduler:Act()
end
return unpack(rets)
end
end
local t = multi:newThread("TempThread",func,...)
t.OnDeath(function(self,status,...) rets = {...} end)
t.OnError(function(self,e) err = e end)
--if thread.isThread() then
-- return wait()
--else
return {
isTFunc = true,
wait = wait,
@ -1516,7 +1564,6 @@ function thread:newFunction(func)
t.OnError(function(self,err) f(self, err) end)
end
}
--end
end
setmetatable(c,c)
return c
@ -1648,7 +1695,7 @@ function multi:newThread(name,func,...)
threadid = threadid + 1
return c
end
function multi.initThreads()
function multi.initThreads(justThreads)
initT = true
multi.scheduler=multi:newLoop():setName("multi.thread")
multi.scheduler.Type="scheduler"
@ -1764,120 +1811,14 @@ function multi.initThreads()
helper(i)
end
end)
if justThreads then
while true do
multi.scheduler:Act()
end
end
end
function multi:threadloop()
initT = true
multi.scheduler=multi:newLoop():setName("multi.thread")
multi.scheduler.Type="scheduler"
function multi.scheduler:setStep(n)
self.skip=tonumber(n) or 24
end
multi.scheduler.skip=0
local t0,t1,t2,t3,t4,t5,t6
local r1,r2,r3,r4,r5,r6
local ret,_
local function helper(i)
if ret then
if ret[1]=="_kill_" then
threads[i].OnDeath:Fire(threads[i],"killed",ret,r1,r2,r3,r4,r5,r6)
table.remove(threads,i)
ret = nil
elseif ret[1]=="_sleep_" then
threads[i].sec = ret[2]
threads[i].time = clock()
threads[i].task = "sleep"
threads[i].__ready = false
ret = nil
elseif ret[1]=="_skip_" then
threads[i].count = ret[2]
threads[i].pos = 0
threads[i].task = "skip"
threads[i].__ready = false
ret = nil
elseif ret[1]=="_hold_" then
threads[i].func = ret[2]
threads[i].task = "hold"
threads[i].__ready = false
ret = nil
elseif ret[1]=="_holdF_" then
threads[i].sec = ret[2]
threads[i].func = ret[3]
threads[i].task = "holdF"
threads[i].time = clock()
threads[i].__ready = false
ret = nil
elseif ret[1]=="_holdW_" then
threads[i].count = ret[2]
threads[i].pos = 0
threads[i].func = ret[3]
threads[i].task = "holdW"
threads[i].time = clock()
threads[i].__ready = false
ret = nil
end
end
end
while true do
for i=#threads,1,-1 do
if not threads[i].__started then
_,ret,r1,r2,r3,r4,r5,r6=coroutine.resume(threads[i].thread)
threads[i].__started = true
helper(i)
end
if not _ then
threads[i].OnError:Fire(threads[i],ret)
end
if coroutine.status(threads[i].thread)=="dead" then
threads[i].OnDeath(threads[i],"ended",ret,r1,r2,r3,r4,r5,r6)
table.remove(threads,i)
elseif threads[i].task == "skip" then
threads[i].pos = threads[i].pos + 1
if threads[i].count==threads[i].pos then
threads[i].task = ""
threads[i].__ready = true
end
elseif threads[i].task == "hold" then
t0,t1,t2,t3,t4,t5,t6 = threads[i].func()
if t0 then
threads[i].task = ""
threads[i].__ready = true
end
elseif threads[i].task == "sleep" then
if clock() - threads[i].time>=threads[i].sec then
threads[i].task = ""
threads[i].__ready = true
end
elseif threads[i].task == "holdF" then
t0,t1,t2,t3,t4,t5,t6 = threads[i].func()
if t0 then
threads[i].task = ""
threads[i].__ready = true
elseif clock() - threads[i].time>=threads[i].sec then
threads[i].task = ""
threads[i].__ready = true
t0 = nil
t1 = "TIMEOUT"
end
elseif threads[i].task == "holdW" then
threads[i].pos = threads[i].pos + 1
t0,t1,t2,t3,t4,t5,t6 = threads[i].func()
if t0 then
threads[i].task = ""
threads[i].__ready = true
elseif threads[i].count==threads[i].pos then
threads[i].task = ""
threads[i].__ready = true
t0 = nil
t1 = "TIMEOUT"
end
end
if threads[i] and threads[i].__ready then
threads[i].__ready = false
_,ret,r1,r2,r3,r4,r5,r6=coroutine.resume(threads[i].thread,t0,t1,t2,t3,t4,t5,t6)
end
helper(i)
end
end
multi.initThreads(true)
end
multi.OnError=multi:newConnection()
function multi:newThreadedProcess(name)

23
multi/integration/lanesManager/extensions.lua
View File

@ -1,3 +1,26 @@
--[[
MIT License
Copyright (c) 2020 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
local multi, thread = require("multi"):init()
local GLOBAL, THREAD = multi.integration.GLOBAL,multi.integration.THREAD
function multi:newSystemThreadedQueue(name)

114
multi/integration/lanesManager/init.lua
View File

@ -1,7 +1,7 @@
--[[
MIT License
Copyright (c) 2019 Ryan Ward
Copyright (c) 2020 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
@ -30,13 +30,6 @@ if multi.integration then -- This allows us to call the lanes manager from suppo
end
}
end
function os.getOS()
if package.config:sub(1, 1) == "\\" then
return "windows"
else
return "unix"
end
end
-- Step 1 get lanes
lanes = require("lanes").configure()
multi.SystemThreads = {}
@ -50,114 +43,11 @@ end
-- Step 2 set up the Linda objects
local __GlobalLinda = lanes.linda() -- handles global stuff
local __SleepingLinda = lanes.linda() -- handles sleeping stuff
-- For convenience a GLOBAL table will be constructed to handle requests
local GLOBAL = {}
setmetatable(
GLOBAL,
{
__index = function(t, k)
return __GlobalLinda:get(k)
end,
__newindex = function(t, k, v)
__GlobalLinda:set(k, v)
end
}
)
-- Step 3 rewrite the thread methods to use Lindas
local THREAD = {}
function THREAD.set(name, val)
__GlobalLinda:set(name, val)
end
function THREAD.get(name)
__GlobalLinda:get(name)
end
local function randomString(n)
local str = ""
local strings = {"a","b","c","d","e","f","g","h","i","j","k","l","m","n","o","p","q","r","s","t","u","v","w","x","y","z","1","2","3","4","5","6","7","8","9","0","A","B","C","D","E","F","G","H","I","J","K","L","M","N","O","P","Q","R","S","T","U","V","W","X","Y","Z"}
for i = 1, n do
str = str .. "" .. strings[math.random(1, #strings)]
end
return str
end
function THREAD.waitFor(name)
local function wait()
math.randomseed(os.time())
__SleepingLinda:receive(.001, randomString(12))
end
repeat
wait()
until __GlobalLinda:get(name)
return __GlobalLinda:get(name)
end
function THREAD.testFor(name, val, sym)
--
end
function THREAD.getCores()
return THREAD.__CORES
end
function THREAD.getThreads()
return GLOBAL.__THREADS__
end
if os.getOS() == "windows" then
THREAD.__CORES = tonumber(os.getenv("NUMBER_OF_PROCESSORS"))
else
THREAD.__CORES = tonumber(io.popen("nproc --all"):read("*n"))
end
function THREAD.kill() -- trigger the lane destruction
error("Thread was killed!")
end
function THREAD.getName()
return THREAD_NAME
end
function THREAD.getID()
return THREAD_ID
end
_G.THREAD_ID = 0
--[[ Step 4 We need to get sleeping working to handle timing... We want idle wait, not busy wait
Idle wait keeps the CPU running better where busy wait wastes CPU cycles... Lanes does not have a sleep method
however, a linda recieve will in fact be a idle wait! So we use that and wrap it in a nice package]]
function THREAD.sleep(n)
math.randomseed(os.time())
__SleepingLinda:receive(n, randomString(12))
end
function THREAD.hold(n)
local function wait()
math.randomseed(os.time())
__SleepingLinda:receive(.001, randomString(12))
end
repeat
wait()
until n()
end
local rand = math.random(1, 10000000)
-- Step 5 Basic Threads!
local GLOBAL,THREAD = require("multi.integration.lanesManager.threads").init(__GlobalLinda,__SleepingLinda)
local threads = {}
local count = 1
local started = false
local livingThreads = {}
function multi.removeUpvalues(func)
if not debug then return end
local count=1
local dat = true
while dat do
dat = debug.setupvalue(func, count, nil)
count = count+1
end
end
function multi.getUpvalues(func)
local count=1
local tab = {}
local dat = true
while dat do
dat = debug.getupvalue(func, count)
if dat then
table.insert(tab,dat)
print(dat)
end
count = count+1
end
return tab
end
function multi:newSystemThread(name, func, ...)
multi.InitSystemThreadErrorHandler()
rand = math.random(1, 10000000)

39
multi/integration/lanesManager/threads.lua
View File

@ -1,3 +1,33 @@
--[[
MIT License
Copyright (c) 2020 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
local function getOS()
if package.config:sub(1, 1) == "\\" then
return "windows"
else
return "unix"
end
end
local function INIT(__GlobalLinda,__SleepingLinda)
local THREAD = {}
function THREAD.set(name, val)
@ -16,8 +46,10 @@ local function INIT(__GlobalLinda,__SleepingLinda)
until __GlobalLinda:get(name)
return __GlobalLinda:get(name)
end
function THREAD.testFor(name, val, sym)
--
if getOS() == "windows" then
THREAD.__CORES = tonumber(os.getenv("NUMBER_OF_PROCESSORS"))
else
THREAD.__CORES = tonumber(io.popen("nproc --all"):read("*n"))
end
function THREAD.getCores()
return THREAD.__CORES
@ -59,9 +91,6 @@ local function INIT(__GlobalLinda,__SleepingLinda)
return __GlobalLinda:get(k)
end,
__newindex = function(t, k, v)
for i,v in pairs(__GlobalLinda) do
print(i,v)
end
__GlobalLinda:set(k, v)
end
})

24
multi/integration/loveManager/extensions.lua
View File

@ -1,5 +1,27 @@
--[[
MIT License
Copyright (c) 2020 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
local multi, thread = require("multi").init()
local status = require("multi.integration.loveManager.status")
local pad = require("multi.integration.loveManager.scratchpad")
GLOBAL = multi.integration.GLOBAL
THREAD = multi.integration.THREAD

35
multi/integration/loveManager/init.lua
View File

@ -1,17 +1,37 @@
--[[
MIT License
Copyright (c) 2020 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
if ISTHREAD then
error("You cannot require the loveManager from within a thread!")
end
local ThreadFileData = [[
ISTHREAD = true
THREAD = require("multi.integration.loveManager.threads") -- order is important!
scratchpad = require("multi.integration.loveManager.scratchpad")
STATUS = require("multi.integration.loveManager.status")
__IMPORTS = {...}
__FUNC__=table.remove(__IMPORTS,1)
__THREADID__=table.remove(__IMPORTS,1)
__THREADNAME__=table.remove(__IMPORTS,1)
pad=table.remove(__IMPORTS,1)
globalhpad=table.remove(__IMPORTS,1)
GLOBAL = THREAD.getGlobal()
multi, thread = require("multi").init()
THREAD.loadDump(__FUNC__)(unpack(__IMPORTS))]]
@ -21,21 +41,16 @@ __THREADID__ = 0
__THREADNAME__ = "MainThread"
multi.integration={}
multi.integration.love2d={}
multi.integration.love2d.defaultScratchpadSize = 1024
multi.integration.love2d.GlobalScratchpad = love.data.newByteData(string.rep("\0",16*multi.integration.love2d.defaultScratchpadSize))
local THREAD = require("multi.integration.loveManager.threads")
local scratchpad = require("multi.integration.loveManager.scratchpad")
local STATUS = require("multi.integration.loveManager.status")
local GLOBAL = THREAD.getGlobal()
local THREAD_ID = 1 -- The Main thread has ID of 0
local THREAD_ID = 1
local OBJECT_ID = 0
function multi:newSystemThread(name,func,...)
local c = {}
c.scratchpad = love.data.newByteData(string.rep("\0",multi.integration.love2d.defaultScratchpadSize))
c.name = name
c.ID=THREAD_ID
c.thread=love.thread.newThread(ThreadFileData)
c.thread:start(THREAD.dump(func),c.ID,c.name,c.scratchpad,multi.integration.love2d.GlobalScratchpad,...)
c.thread:start(THREAD.dump(func),c.ID,c.name,...)
GLOBAL["__THREAD_"..c.ID] = {ID=c.ID,Name=c.name,Thread=c.thread}
GLOBAL["__THREAD_COUNT"] = THREAD_ID
THREAD_ID=THREAD_ID+1

41
multi/integration/loveManager/scratchpad.lua
View File

@ -1,41 +0,0 @@
local ffi = require("ffi")
local scratchpad = {}
local mt = {
__index = function(t, k)
if type(k)=="string" then
local a, b = k:match("(%d+):(%d+)")
return t:read(tonumber(a),tonumber(b))
elseif type(k)=="number" then
return t:read(k,1)
end
end,
__newindex = function(t, k, v)
t:write(v,k)
end
}
function scratchpad:new(data, size, rep)
local c = {}
local pad
if type(data)=="string" then
pad = love.data.newByteData(data or string.rep(rep or "\0",size or 16))
elseif data:type()=="ByteData" then
pad = data
end
local ptr = ffi.cast("unsigned char*",pad:getPointer())
local size = pad:getSize()
function c:write(data, loc, len)
if (loc or 0)+(len or #data)>size then
error("Attpemting to write data outside the bounds of data byte array!")
end
ffi.copy(ptr+(loc or 0), data, len or #data)
end
function c:read(loc, len)
if (((loc or 0)+(len or size)) > size) or ((loc or 0)<0) then
error("Attempt to read outside the bounds of the scratchpad!")
end
return ffi.string(ptr+(loc or 0), len or size)
end
setmetatable(c,mt)
return c
end
return scratchpad

13
multi/integration/loveManager/status.lua
View File

@ -1,13 +0,0 @@
--[[
global pads allow us to directly write data to threads. Each thread has a unique ID which means we can allocate space in memory for each thread to relay stats
Below are codes, If there is more data that needs to be sent we can use byte 0 for that and byte 1,2 and 3 to define a channel
]]
local char = string.char
local cmds = {
OK = char(0x00), -- All is good thread is running can recieve and send data
ERR = char(0x01), -- This tells the system that an error has occured
STOP = char(0x02), -- Thread has finished
BUSY = char(0x03), -- Thread is busy and isn't responding to messages right now
POST = char(0x04), -- Important message for other threads to see, ChannelData with message MSG_TID
}
return cmds

30
multi/integration/loveManager/threads.lua
View File

@ -1,5 +1,25 @@
--[[
Shared methods for both the main thread and
MIT License
Copyright (c) 2020 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
require("love.timer")
require("love.system")
@ -47,7 +67,6 @@ function threads.get(name)
return dat
end
end
function threads.waitFor(name)
if thread.isThread() then
return thread.hold(function()
@ -63,22 +82,17 @@ function threads.waitFor(name)
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.testFor(name,val,sym)
-- Not yet implemented
end
function threads.getCores()
return love.system.getProcessorCount()
end
function threads.kill()
error("Thread Killed!")
end
function threads.getThreads()
local t = {}

71
multi/integration/luvitManager.lua
View File

@ -1,7 +1,7 @@
--[[
MIT License
Copyright (c) 2019 Ryan Ward
Copyright (c) 2020 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
@ -21,8 +21,8 @@ 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.
]]
-- I DEMAND USAGE FOR LUVIT
-- Cannot use discordia without my multitasking library (Which I love more that the luvit platform... then again i'm partial :P)
-- This module probably will not be maintained any longer!
package.path = "?/init.lua;?.lua;" .. package.path
local function _INIT(luvitThread, timer)
-- lots of this stuff should be able to stay the same
@ -66,70 +66,7 @@ local function _INIT(luvitThread, timer)
end
local function randomString(n)
local str = ""
local strings = {
"a",
"b",
"c",
"d",
"e",
"f",
"g",
"h",
"i",
"j",
"k",
"l",
"m",
"n",
"o",
"p",
"q",
"r",
"s",
"t",
"u",
"v",
"w",
"x",
"y",
"z",
"1",
"2",
"3",
"4",
"5",
"6",
"7",
"8",
"9",
"0",
"A",
"B",
"C",
"D",
"E",
"F",
"G",
"H",
"I",
"J",
"K",
"L",
"M",
"N",
"O",
"P",
"Q",
"R",
"S",
"T",
"U",
"V",
"W",
"X",
"Y",
"Z"
}
local strings = {"a","b","c","d","e","f","g","h","i","j","k","l","m","n","o","p","q","r","s","t","u","v","w","x","y","z","1","2","3","4","5","6","7","8","9","0","A","B","C","D","E","F","G","H","I","J","K","L","M","N","O","P","Q","R","S","T","U","V","W","X","Y","Z"}
for i = 1, n do
str = str .. "" .. strings[math.random(1, #strings)]
end

4
multi/integration/networkManager.lua
View File

@ -1,7 +1,7 @@
--[[
MIT License
Copyright (c) 2019 Ryan Ward
Copyright (c) 2020 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
@ -21,7 +21,7 @@ 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 multi, thread = require("multi"):init()
local net = require("net")
local bin = require("bin")
bin.setBitsInterface(infinabits) -- the bits interface does not work so well, another bug to fix

0
multi/integration/networkManager/childNode.lua Normal file
View File

15
multi/integration/networkManager/cmds.lua Normal file
View File

@ -0,0 +1,15 @@
local cmds = {
ERROR = 0x00,
PING = 0x01,
PONG = 0x02,
QUEUE = 0x03,
TASK = 0x04,
INITNODE = 0x05,
INITMASTER = 0x06,
GLOBAL = 0x07,
LOAD = 0x08,
CALL = 0x09,
REG = 0x0A,
CONSOLE = 0x0B,
}
return cmds

23
multi/integration/networkManager/extensions.lua Normal file
View File

@ -0,0 +1,23 @@
--[[
MIT License
Copyright (c) 2020 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]

57
multi/integration/networkManager/init.lua Normal file
View File

@ -0,0 +1,57 @@
--[[
MIT License
Copyright (c) 2020 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
local multi, thread = require("multi"):init()
local net = require("net")
local bin = require("bin")
local char = string.char
local byte = string.byte
bin.setBitsInterface(infinabits)
--[[
--[=[ Pre reqs:
- Network contains nodes
- Network can broadcast/has nodemanager/ is simple and can be scanned
Outline:
- multi:newMasterNode(connectionDetails)
-- master:setDefaultNode(nodeName) -- Set default node
-- master:newNetworkThread(nodeName,func,...) -- Thread is ran on a random node or the default one if set if nodeName is set to nil
-- master:newNetworkChannel(nodeName)
-- master:sendTo(nodeName,data)
- multi:newNode(connectionDetails)
- multi:newNodeManager(connectionDetails) -- This will be incharge of a lot of data handling
]=]
local nGLOBAL, nTHREAD = require("multi.integration.networkManager"):init()
local master = multi:newMasterNode()
master:newNetworkThread("simpleNode",function(a,b,c)
print(a,b,c)
end,1,2,3)
]]
-- The init file should provide the structure that all the other modules build off of
return {
init = function()
--
end
}

70
multi/integration/networkManager/masterNode.lua Normal file
View File

@ -0,0 +1,70 @@
--[[
MIT License
Copyright (c) 2020 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]
local multi, thread = require("multi"):init()
local cmd = require("multi.integration.networkManager.cmds")
local net = require("net")
local bin = require("bin")
local master = {}
master.__index = master
function multi:newMasterNode(connectionDetails)
local c = {}
c.OnDataRecieved = multi:newConnection()
c.defaultNode = ""
c.nodes = {}
if connectionDetails then
-- We do what we do
else
-- Lets make assumptions
end
setmetatable(c, master)
return c
end
function master:setDefaultNode(nodeName)
-- Verify Node is active
if self:nodeExists(nodeName) then
self.defaultNode = nodeName
end
end
function master:newNetworkThread(nodeName,func)
--
end
function master:newNetworkChannel(nodeName)
--
end
function master:sendTo(nodeName,data)
if self:nodeExists(nodeName):wait() then
print("It exists!")
end
end
master.nodeExists = thread:newFunction(function(self,nodeName)
if self.nodes[nodeName] then
local wait = nodes[nodeName]:ping()
local bool = thread.hold(function()
return wait()
end)
return bool
else
return false
end
end)

0
multi/integration/networkManager/nodeManager.lua Normal file
View File

23
multi/integration/networkManager/threads.lua Normal file
View File

@ -0,0 +1,23 @@
--[[
MIT License
Copyright (c) 2020 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sub-license, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
]]

4
multi/integration/networkManager/utils.lua Normal file
View File

@ -0,0 +1,4 @@
local bin = require("bin")
local utils = {}
-- Will contain data that handles sterilizing and managing data
return utils

229
multi/integration/shared.lua
View File

@ -1,7 +1,7 @@
--[[
MIT License
Copyright (c) 2019 Ryan Ward
Copyright (c) 2020 Ryan Ward
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
@ -167,7 +167,6 @@ function multi:newSystemThreadedConnection(name, protect)
GLOBAL[c.name] = c
return c
end
function multi:SystemThreadedBenchmark(n)
n = n or 1
local cores = multi.integration.THREAD.getCores()
@ -269,232 +268,6 @@ function multi:newSystemThreadedConsole(name)
GLOBAL[c.name] = c
return c
end
-- NEEDS WORK
function multi:newSystemThreadedTable(name)
local c = {}
c.name = name -- set the name this is important for identifying what is what
local sThread = multi.integration.THREAD
local GLOBAL = multi.integration.GLOBAL
function c:init() -- create an init function so we can mimic on both love2d and lanes
_G.__Needs_Multi = true
local multi = require("multi")
if multi:getPlatform() == "love2d" then
GLOBAL = _G.GLOBAL
sThread = _G.sThread
end
local cc = {}
cc.tab = {}
if multi.isMainThread then
if not GLOBAL[self.name .. "_Tabled_Connection"] then
cc.conn = multi:newSystemThreadedConnection(self.name .. "_Tabled_Connection"):init()
end
else
cc.conn = sThread.waitFor(self.name .. "_Tabled_Connection"):init()
end
function cc:waitFor(name)
repeat
multi:uManager()
until tab[name] ~= nil
return tab[name]
end
local link = cc
cc.conn(
function(k, v)
link.tab[k] = v
end
)
setmetatable(
cc,
{
__index = function(t, k)
return t.tab[k]
end,
__newindex = function(t, k, v)
t.tab[k] = v
t.conn:Fire(k, v)
end
}
)
return cc
end
GLOBAL[c.name] = c
return c
end
local jobqueuecount = 0
local jqueues = {}
function multi:newSystemThreadedJobQueue(a, b)
jobqueuecount = jobqueuecount + 1
local GLOBAL = multi.integration.GLOBAL
local sThread = multi.integration.THREAD
local c = {}
c.numberofcores = 4
c.idle = nil
c.name = "SYSTEM_THREADED_JOBQUEUE_" .. jobqueuecount
-- This is done to keep backwards compatibility for older code
if type(a) == "string" and not (b) then
c.name = a
elseif type(a) == "number" and not (b) then
c.numberofcores = a
elseif type(a) == "string" and type(b) == "number" then
c.name = a
c.numberofcores = b
elseif type(a) == "number" and type(b) == "string" then
c.name = b
c.numberofcores = a
end
if jqueues[c.name] then
error("A job queue by the name: " .. c.name .. " already exists!")
end
jqueues[c.name] = true
c.isReady = false
c.jobnum = 1
c.OnJobCompleted = multi:newConnection()
local queueIN = self:newSystemThreadedQueue("QUEUE_IN_" .. c.name):init()
local queueCC = self:newSystemThreadedQueue("QUEUE_CC_" .. c.name):init()
local queueREG = self:newSystemThreadedQueue("QUEUE_REG_" .. c.name):init()
local queueJD = self:newSystemThreadedQueue("QUEUE_JD_" .. c.name):init()
local queueDA = self:newSystemThreadedQueue("QUEUE_DA_" .. c.name):init()
c.OnReady = multi:newConnection()
function c:registerJob(name, func)
for i = 1, self.numberofcores do
queueREG:push({name, func})
end
end
c.tempQueue = {}
function c:pushJob(name, ...)
c.idle = os.clock()
if not self.isReady then
table.insert(c.tempQueue, {self.jobnum, name, ...})
self.jobnum = self.jobnum + 1
return self.jobnum - 1
else
queueIN:push {self.jobnum, name, ...}
self.jobnum = self.jobnum + 1
return self.jobnum - 1
end
end
function c:doToAll(func)
local r = multi.randomString(12)
for i = 1, self.numberofcores do
queueDA:push {r, func}
end
end
for i = 1, c.numberofcores do
multi:newSystemThread(
c.name .. " Worker Thread #" .. i,
function(name)
local multi = require("multi")
if love then -- lets make sure we don't reference up-values if using love2d
GLOBAL = _G.GLOBAL
sThread = _G.sThread
end
local CC = sThread.waitFor("QUEUE_CC_" .. name):init()
CC:push("ready")
local FUNCS = {}
local ids = {}
local JQI = sThread.waitFor("QUEUE_IN_" .. name):init()
local JD = sThread.waitFor("QUEUE_JD_" .. name):init()
local REG = sThread.waitFor("QUEUE_REG_" .. name):init()
local DA = sThread.waitFor("QUEUE_DA_" .. name):init()
local lastjob = os.clock()
multi:newLoop(
function()
local job = JQI:pop()
local rd = REG:peek()
local da = DA:peek()
if rd then
if not FUNCS[rd[1]] then
FUNCS[rd[1]] = rd[2]
rd = nil
REG:pop()
end
end
if da then
if not ids[da[1]] then
local meh = da[1]
ids[da[1]] = true
da[2](multi)
da = nil
DA:pop()
multi:newAlarm(60):OnRing(
function(a)
ids[meh] = nil
a:Destroy()
end
)
end
end
if job then
lastjob = os.clock()
local ID = table.remove(job, 1) -- return and remove
local _name = table.remove(job, 1) -- return and remove
if FUNCS[_name] then
JD:push({ID, FUNCS[_name](unpack(job))})
else -- making use of that new holding feature
JD:push({ID, FUNCS:waitFor(_name)(unpack(job))})
end
end
end
)
multi:newLoop(
function()
if os.clock() - lastjob > 1 then
sThread.sleep(.1)
end
end
)
setmetatable(
_G,
{
__index = function(t, k)
return FUNCS[k]
end
}
)
if not love then
multi:mainloop()
end
end,
c.name
)
end
local clock = os.clock
multi:newThread(
"JQ-" .. c.name .. " Manager",
function()
local _count = 0
while _count < c.numberofcores do
thread.skip()
if queueCC:pop() then
_count = _count + 1
end
end
c.isReady = true
for i = 1, #c.tempQueue do
queueIN:push(c.tempQueue[i])
end
c.tempQueue = nil
c.OnReady:Fire(c)
local dat
while true do
if not c.idle then
thread.sleep(.5)
else
if clock() - c.idle >= 15 then
c.idle = nil
end
thread.skip()
end
dat = queueJD:pop()
if dat then
c.idle = clock()
c.OnJobCompleted:Fire(unpack(dat))
end
end
end
)
return c
end
function multi:newSystemThreadedExecute(cmd)
local c = {}
local GLOBAL = multi.integration.GLOBAL -- set up locals incase we are using lanes