Working on systemthreadedprocess, and experimental newProxy for threading

integration/lanesManager/extensions.lua

@@ -22,8 +22,9 @@ 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
@@ -54,17 +55,24 @@ function multi:newSystemThreadedTable(name)
     local c = {}
     c.link = lanes.linda()
 	c.Name = name
-    setmetatable(c,{
+
+	-- function c:getIndex()
+	-- 	return c.link:dump()
+	-- end
+
+    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
-    end
+
     GLOBAL[name or "_"] = c
 	return c
 end
@@ -134,7 +142,7 @@ function multi:newSystemThreadedJobQueue(n)
     end)
     for i=1,c.cores do
         multi:newSystemThread("SystemThreadedJobQueue",function(queue)
-            local multi,thread = require("multi"):init()
+            local multi, thread = require("multi"):init()
             local idle = os.clock()
             local clock = os.clock
             local ref = 0
@@ -145,12 +153,14 @@ 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("test",function()
+						local name = table.remove(dat, 1)
+						local jid = table.remove(dat, 1)
+						local args = table.remove(dat, 1)
+						queueReturn:push{jid, funcs[name](unpack(args)), queue}
+					end)
                 end
-            end)
+            end).OnError(multi.error)
             thread:newThread("DoAllHandler",function()
                 while true do
                     local dat = thread.hold(function()

integration/lanesManager/init.lua

@@ -97,6 +97,7 @@ function multi:newSystemThread(name, func, ...)
 	},function(...)
 		multi, thread = require("multi"):init(multi_settings)
 		require("multi.integration.lanesManager.extensions")
+		require("multi.integration.sharedExtensions")
 		local has_error = true
 		returns = {pcall(func, ...)}
 		return_linda:set("returns", returns)
@@ -179,6 +180,7 @@ multi.integration = {} -- for module creators
 multi.integration.GLOBAL = GLOBAL
 multi.integration.THREAD = THREAD
 require("multi.integration.lanesManager.extensions")
+require("multi.integration.sharedExtensions")
 return {
 	init = function()
 		return GLOBAL, THREAD

integration/lanesManager/threads.lua

@@ -122,13 +122,11 @@ local function INIT(__GlobalLinda, __SleepingLinda, __StatusLinda, __Console)
 	})
 
     function THREAD.setENV(env, name)
-        name = name or "__env"
-        GLOBAL[name] = env
+        GLOBAL[name or "__env"] = env
     end
 
     function THREAD.getENV(name)
-        name = name or "__env"
-        return GLOBAL[name]
+        return GLOBAL[name or "__env"]
     end
 
     function THREAD.exposeENV(name)

integration/loveManager/extensions.lua

@@ -184,12 +184,14 @@ function multi:newSystemThreadedJobQueue(n)
 					end
 					local dat = queue:performAtomic(atomic)
 					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](unpack(dat))}
+							table.insert(tab,1,id)
+							queueReturn:push(tab)
+						end)
 					end
 				end
 			end):OnError(function(...)

integration/loveManager/init.lua

@@ -53,6 +53,7 @@ multi.integration={}
 multi.integration.GLOBAL = GLOBAL
 multi.integration.THREAD = THREAD
 pcall(require,"multi.integration.loveManager.extensions")
+pcall(require,"multi.integration.sharedExtensions")
 stab["returns"] = {THREAD.loadDump(__FUNC__)(unpack(__IMPORTS))}
 ]]
 
@@ -121,5 +122,6 @@ 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}

integration/pseudoManager/init.lua

@@ -110,6 +110,7 @@ multi.integration = {} -- for module creators
 multi.integration.GLOBAL = GLOBAL
 multi.integration.THREAD = THREAD
 require("multi.integration.pseudoManager.extensions")
+require("multi.integration.sharedExtensions")
 return {
 	init = function()
 		return GLOBAL, THREAD

integration/sharedExtensions/init.lua

@@ -0,0 +1,127 @@
+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:newProxy(obj)
+	
+	local c = {
+		__index = function()
+			--
+		end,
+		__newindex = function()
+			--
+		end,
+		__call = function()
+			--
+		end
+	}
+
+	c.name = multi.randomString(12)
+	
+	function c:init()
+		if not multi.isMainThread then
+			c.send = multi:newSystemThreadedQueue(self.name.."_S"):init()
+			c.recv = multi:newSystemThreadedQueue(self.name.."_R"):init()
+			c.ref = obj
+		else
+			GLOBAL = multi.integration.GLOBAL
+			THREAD = multi.integration.THREAD
+			c.send = THREAD.waitFor(self.name.."_S")
+			c.recv = THREAD.waitFor(self.name.."_R")
+		end
+	end
+
+	return c
+end
+
+function multi:newSystemThreadedProcessor(name, cores)
+
+	local name = name or "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.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)
+	
+	c.jobqueue:registerFunction("__spawnThread__", function(name, func, ...)
+		local multi, thread = require("multi"):init()
+		thread:newThread(name, func, ...)
+		return true
+	end)
+
+	c.jobqueue:registerFunction("__spawnTask__", function(obj, ...)
+		local multi, thread = require("multi"):init()
+		multi[obj](multi, func)
+		return true
+	end)
+
+	c.OnObjectCreated(function(proc, obj)
+		if obj.Type == multi.UPDATER then
+			local func = obj.OnUpdate:Remove()[1]
+			c.jobqueue:pushJob("__spawnTask__", "newUpdater", func)
+		elseif obj.Type == multi.LOOP then
+			local func = obj.OnLoop:Remove()[1]
+			c.jobqueue:pushJob("__spawnTask__", "newLoop", func)
+		else
+			return multi.error("Invalid type!")
+		end
+	end)
+
+	function c:getHandler()
+		-- Not needed
+	end
+
+	function c:getThreads()
+		-- We might want to keep track of the number of threads we have
+	end
+
+	function c:getFullName()
+		return self.parent:getFullName() .. "." .. c.Name
+	end
+
+	function c:getName()
+		return self.Name
+	end
+
+	function c:newThread(name, func, ...)
+		c.jobqueue:pushJob("__spawnThread__", name, func, ...)
+	end
+
+	function c:newFunction(func, holdme)
+		return c.jobqueue:newFunction(func, holdme)
+	end
+
+	function c.run()
+		-- Not needed
+	end
+
+	function c.isActive()
+		-- 
+	end
+	
+	function c.Start()
+		--
+	end
+
+	function c.Stop()
+		--
+	end
+
+	function c:Destroy()
+		--
+	end
+
+	return c
+end
+