feat: finish operators interface

internal/interpreter/transformer.go

@@ -100,7 +100,7 @@ func (t transformer) Transform(ctx context.Context, ll *ir.LLProgram) (runtime.P
 				In:  rIOIn,
 				Out: rIOOut,
 			},
-			Msg: rMsg,
+			MetaMsg: rMsg,
 		})
 	}
 

internal/runtime/func_runner.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 )
 
 var (
@@ -26,26 +27,39 @@ func NewDefaultFuncRunner(repo map[FuncRef]Func) (DefaultFuncRunner, error) {
 	}, nil
 }
 
-func (d DefaultFuncRunner) Run(ctx context.Context, funcRoutines []FuncRoutine) error {
-	for i := range funcRoutines {
-		funcRoutine := funcRoutines[i]
+func (d DefaultFuncRunner) Run(ctx context.Context, funcRoutines []FuncRoutine) (err error) {
+	ctx, cancel := context.WithCancel(ctx)
+	wg := sync.WaitGroup{}
+	wg.Add(len(funcRoutines))
 
-		if funcRoutine.Msg != nil {
-			ctx = context.WithValue(ctx, CtxMsgKey, funcRoutine.Msg)
+	defer func() {
+		if err != nil {
+			cancel()
+		}
+	}()
+
+	for _, routine := range funcRoutines {
+		if routine.MetaMsg != nil {
+			ctx = context.WithValue(ctx, CtxMsgKey, routine.MetaMsg)
 		}
 
-		f, ok := d.repo[funcRoutine.Ref]
+		constructor, ok := d.repo[routine.Ref]
 		if !ok {
-			return fmt.Errorf("%w: %v", ErrRepo, funcRoutine.Ref)
+			return fmt.Errorf("%w: %v", ErrRepo, routine.Ref)
 		}
 
-		cb, err := f(funcRoutine.IO)
+		fun, err := constructor(ctx, routine.IO)
 		if err != nil {
-			return fmt.Errorf("%w: %v", errors.Join(ErrFunc, err), funcRoutine.Ref)
+			return fmt.Errorf("%w: %v", errors.Join(ErrFunc, err), routine.Ref)
 		}
 
-		go cb(context.WithValue(ctx, "msg", funcRoutine.Msg))
+		go func() {
+			fun() // will return at ctx.Done()
+			wg.Done()
+		}()
 	}
 
+	wg.Wait()
+
 	return nil
 }

internal/runtime/program.go

@@ -72,9 +72,9 @@ type Selector struct {
 }
 
 type FuncRoutine struct { // Func spec/def?
-	Ref FuncRef
-	IO  FuncIO
-	Msg Msg
+	Ref     FuncRef
+	IO      FuncIO
+	MetaMsg Msg
 }
 
 type FuncRef struct {

internal/runtime/runtime.go

@@ -33,7 +33,7 @@ type (
 	FuncRunner interface {
 		Run(context.Context, []FuncRoutine) error
 	}
-	Func func(FuncIO) (func(context.Context), error)
+	Func func(context.Context, FuncIO) (func(), error)
 )
 
 var (

internal/runtime/std/std.go

@@ -2,29 +2,35 @@ package std
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/nevalang/neva/internal/runtime"
 )
 
-func Print(io runtime.FuncIO) (func(context.Context), error) {
-	ch, err := io.In.Port("v")
+func Print(ctx context.Context, io runtime.FuncIO) (func(), error) {
+	in, err := io.In.Port("v")
 	if err != nil {
 		return nil, err
 	}
-	return func(ctx context.Context) {
+	out, err := io.Out.Port("v")
+	if err != nil {
+		return nil, err
+	}
+	return func() {
 		for {
 			select {
 			case <-ctx.Done():
 				return
-			case v := <-ch:
+			case v := <-in:
 				fmt.Println(v.String())
+				out <- v
 			}
 		}
 	}, nil
 }
 
-func Lock(io runtime.FuncIO) (func(context.Context), error) {
+func Lock(ctx context.Context, io runtime.FuncIO) (func(), error) {
 	vin, err := io.In.Port("v")
 	if err != nil {
 		return nil, err
@@ -37,7 +43,7 @@ func Lock(io runtime.FuncIO) (func(context.Context), error) {
 	if err != nil {
 		return nil, err
 	}
-	return func(ctx context.Context) {
+	return func() {
 		for {
 			select {
 			case <-ctx.Done():
@@ -50,19 +56,24 @@ func Lock(io runtime.FuncIO) (func(context.Context), error) {
 	}, nil
 }
 
-func Const(io runtime.FuncIO) (func(context.Context), error) {
-	vout, err := io.Out.Port("v")
+func Const(ctx context.Context, io runtime.FuncIO) (func(), error) {
+	msg, ok := ctx.Value("msg").(runtime.Msg)
+	if !ok {
+		return nil, errors.New("ctx msg not found")
+	}
+
+	out, err := io.Out.Port("v")
 	if err != nil {
 		return nil, err
 	}
-	return func(ctx context.Context) {
-		msg := ctx.Value("msg").(runtime.Msg)
+
+	return func() {
 		for {
 			select {
 			case <-ctx.Done():
 				return
 			default:
-				vout <- msg
+				out <- msg
 			}
 		}
 	}, nil