Skip to content

Commit 5d03828

Browse files
digdig
committed
Add working version
1 parent 561d06e commit 5d03828

6 files changed

Lines changed: 331 additions & 0 deletions

File tree

‎default.project.json‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
{
2+
"name": "roblox-lua-parallel",
3+
"tree": {
4+
"$path": "lib"
5+
}
6+
}

‎lib/ClientActor/Worker.lua‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,69 @@
1+
local actor = script:GetActor()
2+
3+
local connections = {}
4+
local preparedRunnable = nil
5+
local bindableEvent = nil
6+
7+
local tasks = {}
8+
9+
local destroyed = false
10+
11+
local function add(connection: RBXScriptConnection)
12+
table.insert(connections, connection)
13+
end
14+
15+
add(actor:BindToMessage("Parallel:Init", function(bindable: BindableEvent, runnable: (...any?) -> any?)
16+
assert(not destroyed, "Parallel destroyed")
17+
bindableEvent = bindable
18+
preparedRunnable = runnable
19+
end))
20+
21+
add(actor:BindToMessage("Parallel:Run", function(...: any?)
22+
assert(preparedRunnable, "Parallel not initialized")
23+
assert(not destroyed, "Parallel destroyed")
24+
preparedRunnable(...)
25+
end))
26+
27+
add(actor:BindToMessage("Parallel:SubmitTask", function(uuid: string, ...: any?)
28+
assert(preparedRunnable, "Parallel not initialized")
29+
assert(not destroyed, "Parallel destroyed")
30+
local args = {...}
31+
32+
if tasks[uuid] ~= nil then
33+
task.cancel(tasks[uuid])
34+
tasks[uuid] = nil
35+
end
36+
tasks[uuid] = task.spawn(function()
37+
bindableEvent:Fire(uuid, preparedRunnable(table.unpack(args)))
38+
end)
39+
end))
40+
41+
add(actor:BindToMessage("Parallel:CancelTask", function(uuid: string)
42+
assert(preparedRunnable, "Parallel not initialized")
43+
assert(not destroyed, "Parallel destroyed")
44+
if tasks[uuid] ~= nil then
45+
task.cancel(tasks[uuid])
46+
tasks[uuid] = nil
47+
end
48+
end))
49+
50+
add(actor:BindToMessage("Parallel:Destroy", function()
51+
assert(preparedRunnable, "Parallel not initialized")
52+
assert(not destroyed, "Parallel destroyed")
53+
destroyed = true
54+
55+
for _, connection in connections do
56+
connection:Disconnect()
57+
end
58+
connections = {}
59+
60+
for _, t in tasks do
61+
task.cancel(t)
62+
end
63+
tasks = {}
64+
65+
preparedRunnable = nil
66+
bindableEvent = nil
67+
end))
68+
69+
return nil

‎lib/ClientActor/init.meta.json‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
{
2+
"className": "Actor"
3+
}

‎lib/ServerActor/Worker.lua‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,69 @@
1+
local actor = script:GetActor()
2+
3+
local connections = {}
4+
local preparedRunnable = nil
5+
local bindableEvent = nil
6+
7+
local tasks = {}
8+
9+
local destroyed = false
10+
11+
local function add(connection: RBXScriptConnection)
12+
table.insert(connections, connection)
13+
end
14+
15+
add(actor:BindToMessage("Parallel:Init", function(bindable: BindableEvent, runnable: (...any?) -> any?)
16+
assert(not destroyed, "Parallel destroyed")
17+
bindableEvent = bindable
18+
preparedRunnable = runnable
19+
end))
20+
21+
add(actor:BindToMessage("Parallel:Run", function(...: any?)
22+
assert(preparedRunnable, "Parallel not initialized")
23+
assert(not destroyed, "Parallel destroyed")
24+
preparedRunnable(...)
25+
end))
26+
27+
add(actor:BindToMessage("Parallel:SubmitTask", function(uuid: string, ...: any?)
28+
assert(preparedRunnable, "Parallel not initialized")
29+
assert(not destroyed, "Parallel destroyed")
30+
local args = {...}
31+
32+
if tasks[uuid] ~= nil then
33+
task.cancel(tasks[uuid])
34+
tasks[uuid] = nil
35+
end
36+
tasks[uuid] = task.spawn(function()
37+
bindableEvent:Fire(uuid, preparedRunnable(table.unpack(args)))
38+
end)
39+
end))
40+
41+
add(actor:BindToMessage("Parallel:CancelTask", function(uuid: string)
42+
assert(preparedRunnable, "Parallel not initialized")
43+
assert(not destroyed, "Parallel destroyed")
44+
if tasks[uuid] ~= nil then
45+
task.cancel(tasks[uuid])
46+
tasks[uuid] = nil
47+
end
48+
end))
49+
50+
add(actor:BindToMessage("Parallel:Destroy", function()
51+
assert(preparedRunnable, "Parallel not initialized")
52+
assert(not destroyed, "Parallel destroyed")
53+
destroyed = true
54+
55+
for _, connection in connections do
56+
connection:Disconnect()
57+
end
58+
connections = {}
59+
60+
for _, t in tasks do
61+
task.cancel(t)
62+
end
63+
tasks = {}
64+
65+
preparedRunnable = nil
66+
bindableEvent = nil
67+
end))
68+
69+
return nil

‎lib/ServerActor/init.meta.json‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
{
2+
"className": "Actor"
3+
}

‎lib/init.luau‎

Lines changed: 181 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,181 @@
1+
local RunService = game:GetService("RunService")
2+
local HttpService = game:GetService("HttpService")
3+
4+
local Promise = require(script.Parent.Promise)
5+
6+
local ClientActor = script.ClientActor
7+
local ServerActor = script.ServerActor
8+
9+
export type Promise = typeof(Promise)
10+
export type PreparedParallel = {
11+
withName: (name: string) -> PreparedParallel,
12+
withActors: (actorCount: number) -> PreparedParallel,
13+
run: (...any?) -> (),
14+
submit: (...any?) -> Promise,
15+
destroy: () -> (),
16+
}
17+
18+
type ActorState = {
19+
count: number,
20+
running: {Promise},
21+
}
22+
23+
-- Parallel class
24+
local Parallel = {}
25+
Parallel.__index = Parallel
26+
27+
function Parallel.of(runnable: (...any?) -> any?): PreparedParallel
28+
local self = setmetatable({
29+
_runnable = runnable,
30+
_name = "ParallelWorker",
31+
_actorCount = 1,
32+
_bindableEvent = Instance.new("BindableEvent"),
33+
_folder = Instance.new("Folder"),
34+
35+
_actors = {},
36+
_actorState = {} :: ActorState,
37+
_results = {},
38+
39+
_connection = nil,
40+
_destroyed = false,
41+
}, Parallel)
42+
43+
self._connection = self._bindableEvent.Event:Connect(function(uuid: string, ...: any)
44+
self._results[uuid] = {...}
45+
end)
46+
47+
self._folder.Parent = script
48+
self:_createActors()
49+
return self
50+
end
51+
52+
function Parallel:withName(name: string): PreparedParallel
53+
assert(not self._destroyed, "Parallel destroyed")
54+
self._name = name
55+
self._folder.Name = name
56+
return self
57+
end
58+
59+
function Parallel:withActors(actorCount: number): PreparedParallel
60+
assert(not self._destroyed, "Parallel destroyed")
61+
self._actorCount = actorCount
62+
self:_createActors()
63+
return self
64+
end
65+
66+
function Parallel:run(...: any?)
67+
assert(not self._destroyed, "Parallel destroyed")
68+
local args = {...}
69+
return Promise.promisify(self._findAvailableActor)(self)
70+
:andThen(function(actor: Actor)
71+
actor:SendMessage("Parallel:Run", table.unpack(args))
72+
end)
73+
end
74+
75+
function Parallel:submit(...: any?): Promise
76+
assert(not self._destroyed, "Parallel destroyed")
77+
local args = {...}
78+
local actor = self:_findAvailableActor()
79+
local uuid = HttpService:GenerateGUID(false)
80+
81+
local promise
82+
local function cleanup(status: string)
83+
if status == "Cancelled" or status == "Rejected" then
84+
actor:SendMessage("Parallel:CancelTask", uuid)
85+
end
86+
87+
self._results[uuid] = nil
88+
self._actorState[actor].count -= 1
89+
self._actorState[actor].running[promise] = nil
90+
end
91+
92+
promise = Promise.new(function(resolve)
93+
actor:SendMessage("Parallel:SubmitTask", uuid, table.unpack(args))
94+
resolve()
95+
end)
96+
:andThen(function()
97+
while self._results[uuid] == nil do
98+
task.wait()
99+
end
100+
return table.unpack(self._results[uuid])
101+
end)
102+
:finally(cleanup)
103+
104+
self._actorState[actor].count += 1
105+
self._actorState[actor].running[promise] = true
106+
return promise
107+
end
108+
109+
function Parallel:destroy()
110+
assert(not self._destroyed, "Parallel already destroyed")
111+
self._destroyed = true
112+
113+
self._connection:Disconnect()
114+
self._connection = nil
115+
116+
for _, actor in self._actors do
117+
for runningPromise, _ in self._actorState[actor].running do
118+
runningPromise:cancel()
119+
end
120+
121+
actor:SendMessage("Parallel:Destroy")
122+
end
123+
124+
self._actors = {}
125+
self._actorState = {}
126+
self._results = {}
127+
128+
self._bindableEvent:Destroy()
129+
self._bindableEvent = nil
130+
131+
self._folder:Destroy()
132+
self._folder = nil
133+
end
134+
135+
function Parallel:_findAvailableActor(): Actor
136+
assert(#self._actors > 0, "No available actors")
137+
138+
-- Find actor with least running promises
139+
local min = 999_999
140+
local minActor = nil
141+
for _, actor in self._actors do
142+
local count = self._actorState[actor].count
143+
if count < min then
144+
min = count
145+
minActor = actor
146+
end
147+
end
148+
149+
return minActor
150+
end
151+
152+
function Parallel:_createActors()
153+
for _ = 1, self._actorCount - #self._actors do
154+
local actor = Parallel._createTemplatedActor()
155+
actor.Parent = self._folder
156+
actor:SendMessage("Parallel:Init", self._bindableEvent, self._runnable)
157+
158+
self._actorState[actor] = {
159+
count = 0,
160+
running = {},
161+
}
162+
table.insert(self._actors, actor)
163+
end
164+
end
165+
166+
function Parallel._createTemplatedActor(): Actor
167+
local actor = nil
168+
if RunService:IsServer() then
169+
actor = ServerActor:Clone()
170+
else
171+
actor = ClientActor:Clone()
172+
end
173+
require(actor.Worker)
174+
return actor
175+
end
176+
177+
function Parallel:__tostring()
178+
return string.format("Parallel(%s)", self._name)
179+
end
180+
181+
return Parallel

0 commit comments

Comments
 (0)