Skip to content

Commit 7909aa4

Browse files
RainingComputersRainingComputers
committed
Implemented sync primitives
1 parent e133bc0 commit 7909aa4

45 files changed

Lines changed: 3868 additions & 389 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎.vscode/settings.json‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,9 @@
1919
"denorm",
2020
"deser",
2121
"distclass",
22+
"Dones",
2223
"dpslc",
24+
"DPSLIO",
2325
"DSERROR",
2426
"Dupher",
2527
"dython",
@@ -72,6 +74,7 @@
7274
"pypiwheel",
7375
"pytest",
7476
"Rabbitmq",
77+
"Recvs",
7578
"relname",
7679
"reltuples",
7780
"segmentio",
@@ -84,6 +87,7 @@
8487
"Strs",
8588
"sysconfig",
8689
"targ",
90+
"TASKGROUP",
8791
"testcache",
8892
"testcontainers",
8993
"testdb",

‎pkg/exports.go‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,10 @@ type PassBlock = runtime.PassBlock
3535
type GotoBlock = runtime.GotoBlock
3636
type ProgramBlock = runtime.ProgramBlock
3737

38+
type ChanRef[T any] = runtime.ChanRef[T]
39+
type WgRef = runtime.WgRef
40+
type MutexRef = runtime.MutexRef
41+
3842
func Body(line int, stmts ...Block) BodyBlock { return runtime.Body(line, stmts...) }
3943
func BodyLabeled(line int, label string, stmts ...Block) BodyLabeledBlock {
4044
return runtime.BodyLabeled(line, label, stmts...)
@@ -60,6 +64,42 @@ func SubProgram(
6064
func Program(body Block, source string, name string) ProgramBlock {
6165
return runtime.Program(body, source, name)
6266
}
67+
func Chan(line int, ret func(Context, string)) AwaitBlock {
68+
return runtime.Chan(line, ret)
69+
}
70+
func Send(line int, id func(Context) string, value func(Context) any) AwaitBlock {
71+
return runtime.Send(line, id, value)
72+
}
73+
func Recv[T any](line int, id func(Context) string, ret func(Context, T)) AwaitBlock {
74+
return runtime.Recv[T](line, id, ret)
75+
}
76+
func WaitGroup(
77+
line int, value func(Context) int,
78+
ret func(Context, string),
79+
) AwaitBlock {
80+
return runtime.WaitGroup(line, value, ret)
81+
}
82+
func WaitGroupAdd(
83+
line int, id func(Context) string,
84+
value func(Context) int,
85+
) AwaitBlock {
86+
return runtime.WaitGroupAdd(line, id, value)
87+
}
88+
func Done(line int, id func(Context) string) AwaitBlock {
89+
return runtime.Done(line, id)
90+
}
91+
func Wait(line int, id func(Context) string) AwaitBlock {
92+
return runtime.Wait(line, id)
93+
}
94+
func MutexCreate(line int, ret func(Context, string)) AwaitBlock {
95+
return runtime.Mutex(line, ret)
96+
}
97+
func Acquire(line int, id func(Context) string) AwaitBlock {
98+
return runtime.Acquire(line, id)
99+
}
100+
func Release(line int, id func(Context) string) AwaitBlock {
101+
return runtime.Release(line, id)
102+
}
63103

64104
/* -=[ Worker Pool ]=- */
65105

‎pkg/platforms/interface.go‎

Lines changed: 157 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -108,12 +108,66 @@ type SpawnEvent struct {
108108
Args []byte `json:"args"`
109109
}
110110

111+
type ChannelCreateEvent struct {
112+
Id string `json:"id"`
113+
}
114+
115+
type ChannelSendEvent struct {
116+
Id string `json:"id"`
117+
Value []byte `json:"value"`
118+
}
119+
120+
type ChannelRecvEvent struct {
121+
Id string `json:"id"`
122+
}
123+
124+
type WaitGroupCreateEvent struct {
125+
Id string `json:"id"`
126+
Value int `json:"value"`
127+
}
128+
129+
type WaitGroupAddEvent struct {
130+
Id string `json:"id"`
131+
Value int `json:"value"`
132+
}
133+
134+
type WaitGroupDoneEvent struct {
135+
Id string `json:"id"`
136+
}
137+
138+
type WaitGroupWaitEvent struct {
139+
Id string `json:"id"`
140+
}
141+
142+
type MutexCreateEvent struct {
143+
Id string `json:"id"`
144+
}
145+
146+
type MutexAcquireEvent struct {
147+
Id string `json:"id"`
148+
}
149+
150+
type MutexReleaseEvent struct {
151+
Id string `json:"id"`
152+
}
153+
111154
func (e HttpRequestEvent) Key() string { return e.Url }
112155
func (e HttpListenEvent) Key() string { return e.Url }
113156
func (e TimerEvent) Key() string { return e.Id }
114157
func (e CheckpointEvent) Key() string { return e.Id }
115158
func (e SpawnEvent) Key() string { return e.Id }
116159

160+
func (e ChannelCreateEvent) Key() string { return e.Id }
161+
func (e ChannelSendEvent) Key() string { return e.Id }
162+
func (e ChannelRecvEvent) Key() string { return e.Id }
163+
func (e WaitGroupCreateEvent) Key() string { return e.Id }
164+
func (e WaitGroupAddEvent) Key() string { return e.Id }
165+
func (e WaitGroupDoneEvent) Key() string { return e.Id }
166+
func (e WaitGroupWaitEvent) Key() string { return e.Id }
167+
func (e MutexCreateEvent) Key() string { return e.Id }
168+
func (e MutexAcquireEvent) Key() string { return e.Id }
169+
func (e MutexReleaseEvent) Key() string { return e.Id }
170+
117171
type EventNotify interface {
118172
Target() uint
119173
Index() uint
@@ -145,19 +199,110 @@ type SpawnEventNotify struct {
145199
Id string
146200
Err string
147201
}
202+
type ChannelCreateEventNotify struct {
203+
Targ uint
204+
Idx uint
205+
Id string
206+
Err string
207+
}
208+
type ChannelSendEventNotify struct {
209+
Targ uint
210+
Idx uint
211+
Id string
212+
Err string
213+
}
214+
type ChannelRecvEventNotify struct {
215+
Targ uint
216+
Idx uint
217+
Id string
218+
Value []byte
219+
Err string
220+
}
221+
type WaitGroupCreateEventNotify struct {
222+
Targ uint
223+
Idx uint
224+
Id string
225+
Err string
226+
}
227+
type WaitGroupAddEventNotify struct {
228+
Targ uint
229+
Idx uint
230+
Id string
231+
Err string
232+
}
233+
type WaitGroupDoneEventNotify struct {
234+
Targ uint
235+
Idx uint
236+
Id string
237+
Err string
238+
}
239+
type WaitGroupWaitEventNotify struct {
240+
Targ uint
241+
Idx uint
242+
Id string
243+
Err string
244+
}
245+
type MutexCreateEventNotify struct {
246+
Targ uint
247+
Idx uint
248+
Id string
249+
Err string
250+
}
251+
type MutexAcquireEventNotify struct {
252+
Targ uint
253+
Idx uint
254+
Id string
255+
Err string
256+
}
257+
type MutexReleaseEventNotify struct {
258+
Targ uint
259+
Idx uint
260+
Id string
261+
Err string
262+
}
148263

149-
func (e HttpListenEventNotify) Target() uint { return e.Targ }
150-
func (e TimerEventNotify) Target() uint { return e.Targ }
151-
func (e CheckpointEventNotify) Target() uint { return e.Targ }
152-
func (e SpawnEventNotify) Target() uint { return e.Targ }
153-
func (e HttpListenEventNotify) Index() uint { return e.Idx }
154-
func (e TimerEventNotify) Index() uint { return e.Idx }
155-
func (e CheckpointEventNotify) Index() uint { return e.Idx }
156-
func (e SpawnEventNotify) Index() uint { return e.Idx }
157-
func (e HttpListenEventNotify) Key() string { return e.Url }
158-
func (e TimerEventNotify) Key() string { return e.Id }
159-
func (e CheckpointEventNotify) Key() string { return e.Id }
160-
func (e SpawnEventNotify) Key() string { return e.Id }
264+
func (e HttpListenEventNotify) Target() uint { return e.Targ }
265+
func (e TimerEventNotify) Target() uint { return e.Targ }
266+
func (e CheckpointEventNotify) Target() uint { return e.Targ }
267+
func (e SpawnEventNotify) Target() uint { return e.Targ }
268+
func (e ChannelCreateEventNotify) Target() uint { return e.Targ }
269+
func (e ChannelSendEventNotify) Target() uint { return e.Targ }
270+
func (e ChannelRecvEventNotify) Target() uint { return e.Targ }
271+
func (e WaitGroupCreateEventNotify) Target() uint { return e.Targ }
272+
func (e WaitGroupAddEventNotify) Target() uint { return e.Targ }
273+
func (e WaitGroupDoneEventNotify) Target() uint { return e.Targ }
274+
func (e WaitGroupWaitEventNotify) Target() uint { return e.Targ }
275+
func (e MutexCreateEventNotify) Target() uint { return e.Targ }
276+
func (e MutexAcquireEventNotify) Target() uint { return e.Targ }
277+
func (e MutexReleaseEventNotify) Target() uint { return e.Targ }
278+
func (e HttpListenEventNotify) Index() uint { return e.Idx }
279+
func (e TimerEventNotify) Index() uint { return e.Idx }
280+
func (e CheckpointEventNotify) Index() uint { return e.Idx }
281+
func (e SpawnEventNotify) Index() uint { return e.Idx }
282+
func (e ChannelCreateEventNotify) Index() uint { return e.Idx }
283+
func (e ChannelSendEventNotify) Index() uint { return e.Idx }
284+
func (e ChannelRecvEventNotify) Index() uint { return e.Idx }
285+
func (e WaitGroupCreateEventNotify) Index() uint { return e.Idx }
286+
func (e WaitGroupAddEventNotify) Index() uint { return e.Idx }
287+
func (e WaitGroupDoneEventNotify) Index() uint { return e.Idx }
288+
func (e WaitGroupWaitEventNotify) Index() uint { return e.Idx }
289+
func (e MutexCreateEventNotify) Index() uint { return e.Idx }
290+
func (e MutexAcquireEventNotify) Index() uint { return e.Idx }
291+
func (e MutexReleaseEventNotify) Index() uint { return e.Idx }
292+
func (e HttpListenEventNotify) Key() string { return e.Url }
293+
func (e TimerEventNotify) Key() string { return e.Id }
294+
func (e CheckpointEventNotify) Key() string { return e.Id }
295+
func (e SpawnEventNotify) Key() string { return e.Id }
296+
func (e ChannelCreateEventNotify) Key() string { return e.Id }
297+
func (e ChannelSendEventNotify) Key() string { return e.Id }
298+
func (e ChannelRecvEventNotify) Key() string { return e.Id }
299+
func (e WaitGroupCreateEventNotify) Key() string { return e.Id }
300+
func (e WaitGroupAddEventNotify) Key() string { return e.Id }
301+
func (e WaitGroupDoneEventNotify) Key() string { return e.Id }
302+
func (e WaitGroupWaitEventNotify) Key() string { return e.Id }
303+
func (e MutexCreateEventNotify) Key() string { return e.Id }
304+
func (e MutexAcquireEventNotify) Key() string { return e.Id }
305+
func (e MutexReleaseEventNotify) Key() string { return e.Id }
161306

162307
type Trace struct {
163308
ID uint `db:"id"`

‎pkg/platforms/postgres/20251031000424_create_isolates_table.sql‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,18 @@ CREATE INDEX idx_isolates_events ON isolates USING GIN (events)
4949
WHERE
5050
status = 'awaiting';
5151

52+
-- Used by: channelSends, channelRecvs, waitGroupAdds, waitGroupDones,
53+
-- waitGroupWaits, mutexAcquires, mutexReleases
54+
-- Query: candidates CTE filters WHERE status = 'awaiting' AND lock = -1
55+
CREATE INDEX idx_isolates_awaiting_unlocked ON isolates (id)
56+
WHERE
57+
status = 'awaiting'
58+
AND
59+
lock = -1;
60+
5261
-- migrate:down
62+
DROP INDEX idx_isolates_awaiting_unlocked;
63+
5364
DROP INDEX idx_isolates_events;
5465

5566
DROP INDEX idx_isolates_lock;

‎pkg/platforms/postgres/20260222000000_create_sync_tables.sql‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
-- migrate:up
2+
CREATE TABLE channels (
3+
id TEXT PRIMARY KEY,
4+
buffer BYTEA[] DEFAULT '{}',
5+
lock INT NOT NULL DEFAULT -1
6+
);
7+
8+
CREATE TABLE wait_groups (
9+
id TEXT PRIMARY KEY,
10+
value INT NOT NULL,
11+
lock INT NOT NULL DEFAULT -1
12+
);
13+
14+
CREATE TABLE mutexes (
15+
id TEXT PRIMARY KEY,
16+
value BOOLEAN NOT NULL,
17+
lock INT NOT NULL DEFAULT -1
18+
);
19+
20+
-- migrate:down
21+
DROP TABLE mutexes;
22+
23+
DROP TABLE wait_groups;
24+
25+
DROP TABLE channels;

0 commit comments

Comments
 (0)