Add support for go statement and create_task statements

Author: RainingComputers

.vscode/settings.json

@@ -14,6 +14,7 @@
     "dbmate",
     "dburl",
     "Dcomplex",
+    "DECREF",
     "deepslate",
     "denorm",
     "deser",
@@ -99,7 +100,8 @@
     "uvcli",
     "varsser",
     "venv",
-    "vmihailenco"
+    "vmihailenco",
+    "XDECREF"
   ],
   "python.defaultInterpreterPath": "./.venv/bin/python",
   "python.terminal.activateEnvironment": false,

pkg/exports.go

@@ -52,16 +52,21 @@ func Goto(line int, name string) GotoBlock { return runtime.Goto(line, name) }
 func Impure(line int, stmt func(Context)) AwaitBlock {
 	return runtime.Impure(line, stmt)
 }
-func Program(body Block, source string) ProgramBlock {
-	return runtime.Program(body, source)
+func SubProgram(
+	line int, program string, argsFn func(Context) map[string]any,
+) AwaitBlock {
+	return runtime.SubProgram(line, program, argsFn)
+}
+func Program(body Block, source string, name string) ProgramBlock {
+	return runtime.Program(body, source, name)
 }
 
 /* -=[ Worker Pool ]=- */
 
 func SpawnWorkers(
 	parallelism uint,
 	platform platforms.Platform,
-	transpiled map[string]ProgramBlock,
+	transpiled []ProgramBlock,
 ) (
 	func(string, string, map[string]any) (uint, bool, error),
 	func(string, string) (any, error),

pkg/libdpsl/libdpsl.go

@@ -16,6 +16,33 @@ static int call_import_metadata_func(
 ) {
     return fn(opaque, func_name, result, result_len);
 }
+
+typedef int (*fork_fn_func)(
+	void* opaque,
+	char* name,
+	char* program,
+	char* args_data,
+	int args_len,
+	uint64_t txn_id,
+	char** err_out
+);
+
+__attribute__((unused))
+static int call_fork_fn_func(
+	fork_fn_func fn,
+	void* opaque,
+	char* name,
+	char* program,
+	char* args_data,
+	int args_len,
+	uint64_t txn_id,
+	char** err_out
+) {
+    return fn(
+		opaque, name, program,
+		args_data, args_len, txn_id, err_out
+	);
+}
 */
 import "C"
 
@@ -151,6 +178,8 @@ type listenHandle struct {
 func DSPlatformListen(
 	platformHandle C.uint64_t,
 	parallelism C.uint64_t,
+	forkFn C.fork_fn_func,
+	forkFnOpaque unsafe.Pointer,
 	listenHandleOut *C.uint64_t,
 	errOut **C.char,
 ) C.int {
@@ -162,7 +191,9 @@ func DSPlatformListen(
 
 	shutdownCtx, shutdown := context.WithCancel(context.Background())
 
-	ch, err := platform.Listen(uint(parallelism), shutdownCtx)
+	ch, err := platform.Listen(
+		uint(parallelism), goForkFn(forkFn, forkFnOpaque), shutdownCtx,
+	)
 	if err != nil {
 		shutdown()
 		setError(errOut, err)
@@ -218,6 +249,7 @@ func DSPlatformSpawn(
 	allocID C.uint64_t,
 	contextBytes *C.char,
 	contextLen C.int,
+	txnID *C.uint64_t,
 	idOut *C.uint64_t,
 	existsOut *C.int,
 	errOut **C.char,
@@ -248,6 +280,15 @@ func DSPlatformSpawn(
 		isolateErr = &FFIWorkerError{Message: errMsgStr}
 	}
 
+	var txn any
+	if txnID != nil {
+		txn, ok = getHandle[any](uint64(*txnID))
+		if !ok {
+			setError(errOut, nil)
+			return 0
+		}
+	}
+
 	id, exists, err := platform.Spawn(
 		int(pc),
 		int(line),
@@ -260,6 +301,7 @@ func DSPlatformSpawn(
 		C.GoString(lang),
 		uint(allocID),
 		ctx,
+		txn,
 	)
 	if err != nil {
 		setError(errOut, err)
@@ -581,4 +623,47 @@ func goMetadataFunc(
 	}
 }
 
+func goForkFn(
+	forkFnC C.fork_fn_func,
+	forkFnOpaque unsafe.Pointer,
+) func(string, string, []byte, any) error {
+	return func(
+		name string, program string,
+		args []byte, txn any,
+	) error {
+		cName := C.CString(name)
+		defer C.free(unsafe.Pointer(cName))
+
+		cProgram := C.CString(program)
+		defer C.free(unsafe.Pointer(cProgram))
+
+		var argsData *C.char
+		var argsLen C.int
+		if len(args) > 0 {
+			argsData = (*C.char)(C.CBytes(args))
+			defer C.free(unsafe.Pointer(argsData))
+			argsLen = C.int(len(args))
+		}
+
+		var txnID C.uint64_t
+		if txn != nil {
+			txnID = C.uint64_t(storeHandle(txn))
+			defer deleteHandle(uint64(txnID))
+		}
+
+		var cErr *C.char
+		ok := C.call_fork_fn_func(
+			forkFnC, forkFnOpaque, cName, cProgram, argsData, argsLen, txnID, &cErr,
+		)
+		if ok == 0 {
+			errStr := C.GoString(cErr)
+			C.free(unsafe.Pointer(cErr))
+
+			return &FFIWorkerError{Message: errStr}
+		}
+
+		return nil
+	}
+}
+
 func main() {}

pkg/libdpsl/libdpsl.h

@@ -39,6 +39,33 @@ static int call_import_metadata_func(
     return fn(opaque, func_name, result, result_len);
 }
 
+typedef int (*fork_fn_func)(
+	void* opaque,
+	char* name,
+	char* program,
+	char* args_data,
+	int args_len,
+	uint64_t txn_id,
+	char** err_out
+);
+
+__attribute__((unused))
+static int call_fork_fn_func(
+	fork_fn_func fn,
+	void* opaque,
+	char* name,
+	char* program,
+	char* args_data,
+	int args_len,
+	uint64_t txn_id,
+	char** err_out
+) {
+    return fn(
+		opaque, name, program,
+		args_data, args_len, txn_id, err_out
+	);
+}
+
 #line 1 "cgo-generated-wrapper"
 
 
@@ -105,9 +132,9 @@ extern int DSTranspilePython(char* source, import_metadata_func metadataFunc, vo
 extern int DSTranspileGo(char* source, char* dir, char** result, char** errOut);
 extern int DSNewPostgres(uint64_t id, char* dburl, char* addr, uint64_t interval, uint64_t timerRes, uint64_t limit, uint64_t* handle, char** errOut);
 extern int DSSpawnInspector(uint64_t platformHandle, char* addr, uint64_t* shutdownHandle, char** errOut);
-extern int DSPlatformListen(uint64_t platformHandle, uint64_t parallelism, uint64_t* listenHandleOut, char** errOut);
+extern int DSPlatformListen(uint64_t platformHandle, uint64_t parallelism, fork_fn_func forkFn, void* forkFnOpaque, uint64_t* listenHandleOut, char** errOut);
 extern int DSPlatformAllocate(uint64_t platformHandle, uint64_t* allocID, char** errOut);
-extern int DSPlatformSpawn(uint64_t platformHandle, int pc, int line, int tick, char* eventsBytes, int eventsLen, char* errMsg, char* name, char* program, char* source, char* lang, uint64_t allocID, char* contextBytes, int contextLen, uint64_t* idOut, int* existsOut, char** errOut);
+extern int DSPlatformSpawn(uint64_t platformHandle, int pc, int line, int tick, char* eventsBytes, int eventsLen, char* errMsg, char* name, char* program, char* source, char* lang, uint64_t allocID, char* contextBytes, int contextLen, uint64_t* txnID, uint64_t* idOut, int* existsOut, char** errOut);
 extern int DSPlatformUpdate(uint64_t platformHandle, char* notificationBytes, int notificationLen, uint64_t isolateID, int currTick, int pc, int line, int nextTick, char* eventsBytes, int eventsLen, char* errMsg, char* source, char* contextBytes, int contextLen, uint64_t* txnOut, char** errOut);
 extern int DSPlatformGetVar(uint64_t platformHandle, char* name, char* varName, char** resultOut, int* resultLen, char** errOut);
 extern int DSPlatformGetTrace(uint64_t platformHandle, char* name, char** resultOut, int* resultLen, char** errOut);

pkg/platforms/interface.go

@@ -101,10 +101,18 @@ type CheckpointEvent struct {
 	Id string `json:"id"`
 }
 
+type SpawnEvent struct {
+	Id      string `json:"id"`
+	Program string `json:"program"`
+	Name    string `json:"name"`
+	Args    []byte `json:"args"`
+}
+
 func (e HttpRequestEvent) Key() string { return e.Url }
 func (e HttpListenEvent) Key() string  { return e.Url }
 func (e TimerEvent) Key() string       { return e.Id }
 func (e CheckpointEvent) Key() string  { return e.Id }
+func (e SpawnEvent) Key() string       { return e.Id }
 
 type EventNotify interface {
 	Target() uint
@@ -131,16 +139,25 @@ type CheckpointEventNotify struct {
 	Id   string
 	Err  string
 }
+type SpawnEventNotify struct {
+	Targ uint
+	Idx  uint
+	Id   string
+	Err  string
+}
 
 func (e HttpListenEventNotify) Target() uint { return e.Targ }
 func (e TimerEventNotify) Target() uint      { return e.Targ }
 func (e CheckpointEventNotify) Target() uint { return e.Targ }
+func (e SpawnEventNotify) 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 }
+func (e SpawnEventNotify) 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 }
+func (e SpawnEventNotify) Key() string       { return e.Id }
 
 type Trace struct {
 	ID           uint   `db:"id"`
@@ -217,6 +234,7 @@ type Platform interface {
 		lang string,
 		allocid uint,
 		ctx Context,
+		txn any,
 	) (uint, bool, error)
 	Update(
 		notification PlatformNotification,
@@ -232,6 +250,7 @@ type Platform interface {
 	GetTrace(name string) ([]Trace, error)
 	Listen(
 		parallelism uint,
+		forkFn func(name string, program string, args []byte, txn any) error,
 		shutdownCtx context.Context,
 	) (chan PlatformNotification, error)
 	InspectorVersion() (string, string)