feat: Phase 8.3 Concurrency — tasks, channels, spawn

- Add async/await/spawn keywords to lexer
- C runtime: pthread wrappers (bux_task_spawn, bux_task_join, bux_task_sleep)
- C runtime: mutex/condvar channel implementation (bux_channel_new/send/recv/close/free)
- Build command adds -pthread flag for C compilation
- Std::Task module: TaskHandle, Task_Spawn, Task_Join, Task_Sleep
- Std::Channel module: generic Channel<T> with Send, Recv, Close, Free
- Parser: spawn Expr() expression
- Sema: ekSpawn returns *void
- HIR: hSpawn node lowered from ekSpawn
- C backend: emit bux_task_spawn(FuncName, arg) for spawn expressions
- Example: examples/concurrency.bux (thread + channel demo)
- Example: examples_pkg/concurrency working project
This commit is contained in:
2026-06-01 00:04:48 +03:00
parent 8e255b2125
commit ac8b25935f
12 changed files with 271 additions and 2 deletions
+15
View File
@@ -0,0 +1,15 @@
import Std::Io::{PrintLine, PrintInt};
import Std::Task::TaskHandle;
import Std::Channel::Channel;
func Worker(arg: *void) -> *void {
PrintLine("Hello from thread!");
return null;
}
func Main() -> int {
let handle: *void = spawn Worker();
Task_Join(TaskHandle { handle: handle });
PrintLine("Thread finished");
return 0;
}
+4
View File
@@ -117,6 +117,7 @@ type
ekIs ekIs
ekTry ekTry
ekUnwrap ## expr! — unwrap or panic ekUnwrap ## expr! — unwrap or panic
ekSpawn ## spawn expr — create a new task
ekBlock ekBlock
ekMatch ekMatch
@@ -196,6 +197,9 @@ type
exprTryType*: TypeExpr # nil for Result?, or explicit target type exprTryType*: TypeExpr # nil for Result?, or explicit target type
of ekUnwrap: of ekUnwrap:
exprUnwrapOperand*: Expr exprUnwrapOperand*: Expr
of ekSpawn:
exprSpawnCallee*: Expr
exprSpawnArgs*: seq[Expr]
of ekBlock: of ekBlock:
exprBlock*: Block exprBlock*: Block
of ekMatch: of ekMatch:
+10
View File
@@ -252,6 +252,16 @@ proc emitExpr(be: var CBackend, node: HirNode): string =
let typ = typeToC(node.sizeOfType) let typ = typeToC(node.sizeOfType)
return &"sizeof({typ})" return &"sizeof({typ})"
of hSpawn:
var argsStr = ""
if node.spawnArgs.len > 0:
# Package arguments into a heap-allocated struct (simplified: single arg)
let arg = be.emitExpr(node.spawnArgs[0])
argsStr = &"(void*){arg}"
else:
argsStr = "NULL"
return &"bux_task_spawn({node.spawnCallee}, {argsStr})"
of hIf: of hIf:
# Ternary expression # Ternary expression
let cond = be.emitExpr(node.ifCond) let cond = be.emitExpr(node.ifCond)
+1 -1
View File
@@ -536,7 +536,7 @@ proc cmdBuild*(args: seq[string], opts: GlobalOptions): int =
# Compile with cc # Compile with cc
let outputName = if man.name != "": man.name else: "bux_out" let outputName = if man.name != "": man.name else: "bux_out"
let outputFile = buildDir / outputName let outputFile = buildDir / outputName
let ccCmd = &"cc -o {outputFile} {cFile} {runtimeDst} {ioDst} -lm 2>&1" let ccCmd = &"cc -pthread -o {outputFile} {cFile} {runtimeDst} {ioDst} -lm 2>&1"
if opts.verbose: if opts.verbose:
printInfo(&"running: {ccCmd}", useColor) printInfo(&"running: {ccCmd}", useColor)
let (output, exitCode) = execCmdEx(ccCmd) let (output, exitCode) = execCmdEx(ccCmd)
+5
View File
@@ -31,6 +31,8 @@ type
hCast hCast
hIs hIs
hSizeOf hSizeOf
# Concurrency
hSpawn
# Composite # Composite
hBlock hBlock
hStructInit hStructInit
@@ -107,6 +109,9 @@ type
isType*: Type isType*: Type
of hSizeOf: of hSizeOf:
sizeOfType*: Type sizeOfType*: Type
of hSpawn:
spawnCallee*: string
spawnArgs*: seq[HirNode]
of hBlock: of hBlock:
blockStmts*: seq[HirNode] blockStmts*: seq[HirNode]
blockExpr*: HirNode blockExpr*: HirNode
+12
View File
@@ -756,6 +756,18 @@ proc lowerExpr(ctx: var LowerCtx, expr: Expr): HirNode =
return HirNode(kind: hLit, litToken: Token(kind: tkStringLiteral, text: "\"\"", loc: loc), return HirNode(kind: hLit, litToken: Token(kind: tkStringLiteral, text: "\"\"", loc: loc),
typ: makeStr(), loc: loc) typ: makeStr(), loc: loc)
of ekSpawn:
var calleeName = ""
if expr.exprSpawnCallee.kind == ekIdent:
calleeName = expr.exprSpawnCallee.exprIdent
elif expr.exprSpawnCallee.kind == ekPath:
calleeName = expr.exprSpawnCallee.exprPath.join("_")
var args: seq[HirNode] = @[]
for arg in expr.exprSpawnArgs:
args.add(ctx.lowerExpr(arg))
return HirNode(kind: hSpawn, spawnCallee: calleeName, spawnArgs: args,
typ: makePointer(makeVoid()), loc: loc)
else: else:
return HirNode(kind: hLit, litToken: Token(kind: tkIntLiteral, text: "0", loc: loc), return HirNode(kind: hLit, litToken: Token(kind: tkIntLiteral, text: "0", loc: loc),
typ: makeVoid(), loc: loc) typ: makeVoid(), loc: loc)
+12
View File
@@ -444,6 +444,18 @@ proc parsePrimary(p: var Parser): Expr =
let ty = p.parseType() let ty = p.parseType()
discard p.expect(tkRParen, "expected ')'") discard p.expect(tkRParen, "expected ')'")
return Expr(kind: ekSizeOf, loc: loc, exprSizeOfType: ty) return Expr(kind: ekSizeOf, loc: loc, exprSizeOfType: ty)
of tkSpawn:
discard p.advance() # spawn
let callee = p.parsePrimary()
var args: seq[Expr] = @[]
if p.check(tkLParen):
discard p.advance() # (
while not p.check(tkRParen) and not p.isAtEnd:
args.add(p.parseExpr())
if p.check(tkComma):
discard p.advance()
discard p.expect(tkRParen, "expected ')' after spawn arguments")
return Expr(kind: ekSpawn, loc: loc, exprSpawnCallee: callee, exprSpawnArgs: args)
of tkHashLine: of tkHashLine:
discard p.advance() discard p.advance()
return Expr(kind: ekIntrinsic, loc: loc, exprIntrinsic: ikLine) return Expr(kind: ekIntrinsic, loc: loc, exprIntrinsic: ikLine)
+5
View File
@@ -1000,6 +1000,11 @@ proc checkExpr(sema: var Sema, expr: Expr, scope: Scope): Type =
case expr.exprIntrinsic case expr.exprIntrinsic
of ikLine, ikColumn: return makeInt() of ikLine, ikColumn: return makeInt()
of ikFile, ikFunction, ikDate, ikTime, ikModule: return makeStr() of ikFile, ikFunction, ikDate, ikTime, ikModule: return makeStr()
of ekSpawn:
discard sema.checkExpr(expr.exprSpawnCallee, scope)
for arg in expr.exprSpawnArgs:
discard sema.checkExpr(arg, scope)
return makePointer(makeVoid())
of ekSpread: of ekSpread:
return sema.checkExpr(expr.exprSpreadOperand, scope) return sema.checkExpr(expr.exprSpreadOperand, scope)
+10 -1
View File
@@ -52,6 +52,9 @@ type
tkOwn # own (gradual ownership transfer) tkOwn # own (gradual ownership transfer)
tkMut # mut (mutable reference) tkMut # mut (mutable reference)
tkDiscard # discard (evaluate and throw away) tkDiscard # discard (evaluate and throw away)
tkAsync # async
tkAwait # await
tkSpawn # spawn
##Punctuation ##Punctuation
tkLParen # ( tkLParen # (
@@ -143,7 +146,7 @@ proc isKeyword*(kind: TokenKind): bool =
tkBreak, tkContinue, tkReturn, tkMatch, tkBreak, tkContinue, tkReturn, tkMatch,
tkFunc, tkLet, tkVar, tkConst, tkType, tkStruct, tkEnum, tkFunc, tkLet, tkVar, tkConst, tkType, tkStruct, tkEnum,
tkUnion, tkInterface, tkExtend, tkModule, tkImport, tkUnion, tkInterface, tkExtend, tkModule, tkImport,
tkPub, tkExtern, tkAs, tkIs, tkNull, tkSelf, tkSuper, tkOwn, tkMut, tkDiscard: tkPub, tkExtern, tkAs, tkIs, tkNull, tkSelf, tkSuper, tkOwn, tkMut, tkDiscard, tkAsync, tkAwait, tkSpawn:
true true
else: else:
false false
@@ -200,6 +203,9 @@ proc keywordKind*(text: string): TokenKind =
of "own": tkOwn of "own": tkOwn
of "mut": tkMut of "mut": tkMut
of "discard": tkDiscard of "discard": tkDiscard
of "async": tkAsync
of "await": tkAwait
of "spawn": tkSpawn
of "true", "false": tkBoolLiteral of "true", "false": tkBoolLiteral
else: tkIdent else: tkIdent
@@ -246,6 +252,9 @@ proc tokenKindName*(kind: TokenKind): string =
of tkOwn: "'own'" of tkOwn: "'own'"
of tkMut: "'mut'" of tkMut: "'mut'"
of tkDiscard: "'discard'" of tkDiscard: "'discard'"
of tkAsync: "'async'"
of tkAwait: "'await'"
of tkSpawn: "'spawn'"
of tkLParen: "'('" of tkLParen: "'('"
of tkRParen: "')'" of tkRParen: "')'"
of tkLBrace: "'{'" of tkLBrace: "'{'"
+35
View File
@@ -0,0 +1,35 @@
module Std::Channel {
extern func bux_channel_new(capacity: int64, elem_size: int64) -> *void;
extern func bux_channel_send(handle: *void, elem: *void);
extern func bux_channel_recv(handle: *void, out: *void) -> int;
extern func bux_channel_close(handle: *void);
extern func bux_channel_free(handle: *void);
struct Channel<T> {
handle: *void;
}
func Channel_New<T>(capacity: int64) -> Channel<T> {
return Channel<T> { handle: bux_channel_new(capacity, sizeof(T)) };
}
func Channel_Send<T>(ch: *Channel<T>, value: T) {
bux_channel_send(ch.handle, &value);
}
func Channel_Recv<T>(ch: *Channel<T>) -> T {
var result: T;
bux_channel_recv(ch.handle, &result);
return result;
}
func Channel_Close<T>(ch: *Channel<T>) {
bux_channel_close(ch.handle);
}
func Channel_Free<T>(ch: *Channel<T>) {
bux_channel_free(ch.handle);
}
}
+23
View File
@@ -0,0 +1,23 @@
module Std::Task {
extern func bux_task_spawn(fn: *void, arg: *void) -> *void;
extern func bux_task_join(handle: *void);
extern func bux_task_sleep(ms: int64);
struct TaskHandle {
handle: *void;
}
func Task_Spawn(fn: *void, arg: *void) -> TaskHandle {
return TaskHandle { handle: bux_task_spawn(fn, arg) };
}
func Task_Join(t: TaskHandle) {
bux_task_join(t.handle);
}
func Task_Sleep(ms: int64) {
bux_task_sleep(ms);
}
}
+139
View File
@@ -6,6 +6,7 @@
#include <stdint.h> #include <stdint.h>
#include <stdbool.h> #include <stdbool.h>
#include <string.h> #include <string.h>
#include <pthread.h>
/* Command-line argument storage */ /* Command-line argument storage */
int g_argc = 0; int g_argc = 0;
@@ -655,3 +656,141 @@ int bux_dir_exists(const char* path) {
struct stat st; struct stat st;
return (stat(path, &st) == 0 && S_ISDIR(st.st_mode)); return (stat(path, &st) == 0 && S_ISDIR(st.st_mode));
} }
/* ============================================================================
* Concurrency primitives (Phase 8.3)
* ============================================================================ */
typedef struct {
pthread_t thread;
} BuxTask;
typedef struct {
uint8_t* buffer;
size_t capacity;
size_t elem_size;
size_t head;
size_t tail;
size_t count;
pthread_mutex_t mutex;
pthread_cond_t not_empty;
pthread_cond_t not_full;
int closed;
} BuxChannel;
/* Task / thread spawning */
void* bux_task_spawn(void* (*func)(void*), void* arg) {
BuxTask* task = (BuxTask*)malloc(sizeof(BuxTask));
if (!task) {
fprintf(stderr, "bux runtime: out of memory (task spawn)\n");
abort();
}
int rc = pthread_create(&task->thread, NULL, func, arg);
if (rc != 0) {
fprintf(stderr, "bux runtime: pthread_create failed (%d)\n", rc);
free(task);
return NULL;
}
return task;
}
void bux_task_join(void* handle) {
if (!handle) return;
BuxTask* task = (BuxTask*)handle;
pthread_join(task->thread, NULL);
free(task);
}
void bux_task_sleep(int64_t ms) {
if (ms <= 0) return;
struct timespec ts;
ts.tv_sec = ms / 1000;
ts.tv_nsec = (ms % 1000) * 1000000;
nanosleep(&ts, NULL);
}
/* Channel implementation */
void* bux_channel_new(int64_t capacity, int64_t elem_size) {
if (capacity <= 0) capacity = 1;
if (elem_size <= 0) elem_size = 1;
BuxChannel* ch = (BuxChannel*)malloc(sizeof(BuxChannel));
if (!ch) {
fprintf(stderr, "bux runtime: out of memory (channel new)\n");
abort();
}
ch->buffer = (uint8_t*)malloc((size_t)capacity * (size_t)elem_size);
if (!ch->buffer) {
fprintf(stderr, "bux runtime: out of memory (channel buffer)\n");
free(ch);
abort();
}
ch->capacity = (size_t)capacity;
ch->elem_size = (size_t)elem_size;
ch->head = 0;
ch->tail = 0;
ch->count = 0;
ch->closed = 0;
pthread_mutex_init(&ch->mutex, NULL);
pthread_cond_init(&ch->not_empty, NULL);
pthread_cond_init(&ch->not_full, NULL);
return ch;
}
void bux_channel_send(void* handle, void* elem) {
if (!handle || !elem) return;
BuxChannel* ch = (BuxChannel*)handle;
pthread_mutex_lock(&ch->mutex);
while (ch->count >= ch->capacity && !ch->closed) {
pthread_cond_wait(&ch->not_full, &ch->mutex);
}
if (ch->closed) {
pthread_mutex_unlock(&ch->mutex);
return;
}
uint8_t* dst = ch->buffer + ch->tail * ch->elem_size;
memcpy(dst, elem, ch->elem_size);
ch->tail = (ch->tail + 1) % ch->capacity;
ch->count++;
pthread_cond_signal(&ch->not_empty);
pthread_mutex_unlock(&ch->mutex);
}
int bux_channel_recv(void* handle, void* out) {
if (!handle || !out) return 0;
BuxChannel* ch = (BuxChannel*)handle;
pthread_mutex_lock(&ch->mutex);
while (ch->count == 0 && !ch->closed) {
pthread_cond_wait(&ch->not_empty, &ch->mutex);
}
if (ch->count == 0) {
pthread_mutex_unlock(&ch->mutex);
return 0; /* channel empty and closed */
}
uint8_t* src = ch->buffer + ch->head * ch->elem_size;
memcpy(out, src, ch->elem_size);
ch->head = (ch->head + 1) % ch->capacity;
ch->count--;
pthread_cond_signal(&ch->not_full);
pthread_mutex_unlock(&ch->mutex);
return 1;
}
void bux_channel_close(void* handle) {
if (!handle) return;
BuxChannel* ch = (BuxChannel*)handle;
pthread_mutex_lock(&ch->mutex);
ch->closed = 1;
pthread_cond_broadcast(&ch->not_empty);
pthread_cond_broadcast(&ch->not_full);
pthread_mutex_unlock(&ch->mutex);
}
void bux_channel_free(void* handle) {
if (!handle) return;
BuxChannel* ch = (BuxChannel*)handle;
pthread_mutex_destroy(&ch->mutex);
pthread_cond_destroy(&ch->not_empty);
pthread_cond_destroy(&ch->not_full);
free(ch->buffer);
free(ch);
}