Add support for calling a flow function

Author: RainingComputers

pkg/exports.go

@@ -15,6 +15,7 @@ type Context = runtime.Context
 
 func GetVar[T any](c Context, name string) T  { return runtime.GetVar[T](c, name) }
 func GetError(c Context, name string) *string { return runtime.GetError(c, name) }
+func GetReturn[T any](c Context, idx int) T   { return runtime.GetReturn[T](c, idx) }
 
 /* -=[ Async Primitives ]=- */
 
@@ -58,8 +59,9 @@ func Impure(line int, stmt func(Context)) AwaitBlock {
 }
 func SubProgram(
 	line int, program string, argsFn func(Context) map[string]any,
+	ret func(Context, string),
 ) AwaitBlock {
-	return runtime.SubProgram(line, program, argsFn)
+	return runtime.SubProgram(line, program, argsFn, ret)
 }
 func Program(body Block, source string, name string) ProgramBlock {
 	return runtime.Program(body, source, name)
@@ -73,6 +75,9 @@ func Send(line int, id func(Context) string, value func(Context) any) AwaitBlock
 func Recv[T any](line int, id func(Context) string, ret func(Context, T)) AwaitBlock {
 	return runtime.Recv[T](line, id, ret)
 }
+func Join(line int, id func(Context) string, rets []func(Context, Context)) AwaitBlock {
+	return runtime.Join(line, id, rets)
+}
 func RecvAny(
 	line int,
 	ids []func(Context) string,

pkg/platforms/interface.go

@@ -151,6 +151,10 @@ type MutexReleaseEvent struct {
 	Id string `json:"id"`
 }
 
+type JoinEvent struct {
+	Id string `json:"id"`
+}
+
 func (e HttpRequestEvent) Key() string { return e.Url }
 func (e HttpListenEvent) Key() string  { return e.Url }
 func (e TimerEvent) Key() string       { return e.Id }
@@ -167,6 +171,7 @@ func (e WaitGroupWaitEvent) Key() string   { return e.Id }
 func (e MutexCreateEvent) Key() string     { return e.Id }
 func (e MutexAcquireEvent) Key() string    { return e.Id }
 func (e MutexReleaseEvent) Key() string    { return e.Id }
+func (e JoinEvent) Key() string            { return e.Id }
 
 type EventNotify interface {
 	Target() uint
@@ -260,6 +265,13 @@ type MutexReleaseEventNotify struct {
 	Id   string
 	Err  string
 }
+type JoinEventNotify struct {
+	Targ uint
+	Idx  uint
+	Id   string
+	Ctx  Context
+	Err  string
+}
 
 func (e HttpListenEventNotify) Target() uint      { return e.Targ }
 func (e TimerEventNotify) Target() uint           { return e.Targ }
@@ -275,6 +287,7 @@ func (e WaitGroupWaitEventNotify) Target() uint   { return e.Targ }
 func (e MutexCreateEventNotify) Target() uint     { return e.Targ }
 func (e MutexAcquireEventNotify) Target() uint    { return e.Targ }
 func (e MutexReleaseEventNotify) Target() uint    { return e.Targ }
+func (e JoinEventNotify) Target() uint            { return e.Targ }
 func (e HttpListenEventNotify) Index() uint       { return e.Idx }
 func (e TimerEventNotify) Index() uint            { return e.Idx }
 func (e CheckpointEventNotify) Index() uint       { return e.Idx }
@@ -289,6 +302,7 @@ func (e WaitGroupWaitEventNotify) Index() uint    { return e.Idx }
 func (e MutexCreateEventNotify) Index() uint      { return e.Idx }
 func (e MutexAcquireEventNotify) Index() uint     { return e.Idx }
 func (e MutexReleaseEventNotify) Index() uint     { return e.Idx }
+func (e JoinEventNotify) Index() uint             { return e.Idx }
 func (e HttpListenEventNotify) Key() string       { return e.Url }
 func (e TimerEventNotify) Key() string            { return e.Id }
 func (e CheckpointEventNotify) Key() string       { return e.Id }
@@ -303,6 +317,7 @@ func (e WaitGroupWaitEventNotify) Key() string    { return e.Id }
 func (e MutexCreateEventNotify) Key() string      { return e.Id }
 func (e MutexAcquireEventNotify) Key() string     { return e.Id }
 func (e MutexReleaseEventNotify) Key() string     { return e.Id }
+func (e JoinEventNotify) Key() string             { return e.Id }
 
 type Trace struct {
 	ID           uint   `db:"id"`

pkg/platforms/postgres/postgres.go

@@ -470,6 +470,7 @@ func (p *Postgres) Listen(
 		go mutexCreates(p, platformCh, shutdownCtx)
 		go mutexAcquires(p, platformCh, shutdownCtx)
 		go mutexReleases(p, platformCh, shutdownCtx)
+		go joins(p, platformCh, shutdownCtx)
 		go tracer(p, shutdownCtx)
 	}
 
@@ -1782,6 +1783,76 @@ func mutexReleases(
 	)
 }
 
+func joins(
+	p *Postgres,
+	platformCh chan platforms.PlatformNotification,
+	shutdownCtx context.Context,
+) {
+	type joinRow struct {
+		isolate
+		JoinID   string `db:"join_id"`
+		ChildCtx []byte `db:"child_context"`
+	}
+
+	pollLoop(
+		p, platformCh, shutdownCtx, p.interval,
+
+		"joins", `
+		WITH locked AS (
+			UPDATE isolates
+			SET lock = $1
+			WHERE id IN (
+				SELECT id FROM isolates
+				WHERE status = 'awaiting'
+					AND lock = -1
+					AND events @> '[{"type":"JoinEvent"}]'
+				FOR UPDATE SKIP LOCKED
+				LIMIT $2
+			)
+			RETURNING id, events, pc, line, tick, program,
+				context
+		)
+		SELECT locked.id, event_match.idx,
+			event_match.join_id,
+			child.context AS child_context,
+			locked.pc, locked.line, locked.tick, locked.program,
+			locked.context
+		FROM locked,
+		LATERAL (
+			SELECT (e.idx - 1)::INT AS idx,
+				e.elem->>'id' AS join_id
+			FROM jsonb_array_elements(locked.events)
+				WITH ORDINALITY AS e(elem, idx)
+			WHERE e.elem->>'type' = 'JoinEvent'
+			ORDER BY e.idx
+			LIMIT 1
+		) event_match
+		JOIN isolates child
+			ON child.name = event_match.join_id
+			AND child.status = 'done'
+	`, func() []any { return []any{p.id, p.limit} },
+
+		func(row joinRow, m notifyMatch) platforms.EventNotify {
+			ctx, err := platforms.DeserCtxJSON(row.ChildCtx)
+			if err != nil {
+				log.Printf("platform: joins: unable to convert child context: %v", err)
+				return nil
+			}
+
+			return platforms.JoinEventNotify{
+				Targ: m.target,
+				Idx:  m.index,
+				Id:   row.JoinID,
+				Ctx:  ctx,
+			}
+		},
+
+		func(_ joinRow, pgTxn pgx.Tx) error {
+			return pgTxn.Commit(context.Background())
+		},
+	)
+}
+
 func makeRequest(
 	url string,
 	method string,