282 lines
10 KiB
Plaintext
282 lines
10 KiB
Plaintext
# ThreadingTest - vthreading coverage
|
|
# Thread/Lock/Event/Condition/RWLock/local/id/sleep_ms/args
|
|
from stdint import *
|
|
from win32.win32base import NULL, FALSE, TRUE
|
|
import stdlib
|
|
import testcheck
|
|
import vthreading
|
|
|
|
|
|
# ===== Shared state: module globals read/written across threads =====
|
|
_shared_count: int = 0 # Lock test shared counter
|
|
_evt_done: int = 0 # Event test done flag
|
|
_cond_ready: int = 0 # Condition test ready flag
|
|
_rw_value: int = 0 # RWLock test shared value
|
|
_main_tid: uint32 = 0 # Main thread ID
|
|
|
|
# Concurrency scale constants
|
|
THREAD_NUM: int = 4
|
|
LOCK_ITER: int = 1000
|
|
RW_ITER: int = 100
|
|
TLS_NUM: int = 4
|
|
SYS_ITER: int = 10
|
|
|
|
|
|
# ===== Thread callbacks (signature: def worker(arg: rawptr) -> rawptr) =====
|
|
|
|
# Thread basic: brief sleep then exit, keeps an is_alive window after start
|
|
def _basic_worker(arg: rawptr) -> rawptr:
|
|
vthreading.sleep_ms(100)
|
|
return NULL
|
|
|
|
|
|
# Arg passing: increment through arg pointer 100 times (c.Deref/c.DerefAs)
|
|
def _increment_worker(arg: rawptr) -> rawptr:
|
|
ip: int* = int*(arg)
|
|
i: int
|
|
for i in range(100):
|
|
*ip = (*ip + 1)
|
|
return NULL
|
|
|
|
|
|
# Lock: increment a global counter while holding the lock
|
|
def _lock_worker(arg: rawptr) -> rawptr:
|
|
lk: vthreading.Lock* = (vthreading.Lock*)(arg)
|
|
global _shared_count
|
|
for i in range(LOCK_ITER):
|
|
lk.acquire()
|
|
_shared_count += 1
|
|
lk.release()
|
|
return NULL
|
|
|
|
|
|
# Event: wait to be woken, then set the done flag
|
|
def _event_worker(arg: rawptr) -> rawptr:
|
|
ev: vthreading.Event* = (vthreading.Event*)(arg)
|
|
global _evt_done
|
|
r: int = ev.wait(5000)
|
|
if r == 1:
|
|
_evt_done = 1
|
|
return NULL
|
|
|
|
|
|
# Condition consumer: wait for _cond_ready, bounded retries to avoid hanging
|
|
def _cond_consumer(arg: rawptr) -> rawptr:
|
|
cond: vthreading.Condition* = (vthreading.Condition*)(arg)
|
|
global _cond_ready
|
|
tries: int = 0
|
|
cond.acquire()
|
|
while _cond_ready == 0 and tries < 100:
|
|
cond.wait(100)
|
|
tries += 1
|
|
cond.release()
|
|
return NULL
|
|
|
|
|
|
# Condition producer: delay, set _cond_ready, then notify
|
|
def _cond_producer(arg: rawptr) -> rawptr:
|
|
cond: vthreading.Condition* = (vthreading.Condition*)(arg)
|
|
global _cond_ready
|
|
vthreading.sleep_ms(50)
|
|
cond.acquire()
|
|
_cond_ready = 1
|
|
cond.notify()
|
|
cond.release()
|
|
return NULL
|
|
|
|
|
|
# RWLock reader: read the shared value repeatedly under the read lock
|
|
def _rw_reader(arg: rawptr) -> rawptr:
|
|
rwl: vthreading.RWLock* = (vthreading.RWLock*)(arg)
|
|
global _rw_value
|
|
i: int
|
|
for i in range(RW_ITER * 10):
|
|
rwl.acquire_read()
|
|
v: int = _rw_value
|
|
rwl.release_read()
|
|
return NULL
|
|
|
|
|
|
# RWLock writer: increment the shared value under the write lock
|
|
def _rw_writer(arg: rawptr) -> rawptr:
|
|
rwl: vthreading.RWLock* = (vthreading.RWLock*)(arg)
|
|
global _rw_value
|
|
i: int
|
|
for i in range(RW_ITER):
|
|
rwl.acquire_write()
|
|
_rw_value += 1
|
|
rwl.release_write()
|
|
return NULL
|
|
|
|
|
|
# TLS: per-thread set/get self-check (store own thread ID, then read back)
|
|
def _tls_worker(arg: rawptr) -> rawptr:
|
|
tls: vthreading.local = vthreading.local()
|
|
tls.set(rawptr(uint64(vthreading.get_current_thread_id())))
|
|
tid_val: uint64 = uint64(tls.get())
|
|
ok: int = 0
|
|
if tid_val == uint64(vthreading.get_current_thread_id()):
|
|
ok = 1
|
|
tls.free()
|
|
return rawptr(uint64(ok))
|
|
|
|
|
|
# Thread ID isolation: worker ID must differ from the main thread ID
|
|
def _tid_worker(arg: rawptr) -> rawptr:
|
|
global _main_tid
|
|
if vthreading.get_current_thread_id() == _main_tid:
|
|
return rawptr(uint64(0))
|
|
return rawptr(uint64(1))
|
|
|
|
|
|
# stdlib.system concurrency: each thread calls C system(), count failures
|
|
def _system_worker(arg: rawptr) -> rawptr:
|
|
i: int
|
|
fails: int = 0
|
|
for i in range(SYS_ITER):
|
|
r: int = stdlib.system("echo ok > NUL")
|
|
if r != 0:
|
|
fails += 1
|
|
return rawptr(uint64(fails))
|
|
|
|
|
|
# ===== main =====
|
|
export def main() -> int:
|
|
testcheck.begin("ThreadingTest: vthreading 专项测试")
|
|
|
|
# ========== Thread: start/join/is_alive/get_exit_code ==========
|
|
testcheck.section("Thread: start/join/is_alive")
|
|
th1: vthreading.Thread = vthreading.Thread(_basic_worker, NULL)
|
|
testcheck.check(th1.is_alive() == FALSE, "未 start 时不在运行", "未 start 却 is_alive")
|
|
th1.start()
|
|
testcheck.check(th1.is_alive() != FALSE, "start 后线程运行中", "start 后未运行")
|
|
testcheck.check(th1.join(5000) != FALSE, "join 正常返回", "join 超时")
|
|
testcheck.check(th1.is_alive() == FALSE, "join 后线程已结束", "join 后仍运行")
|
|
testcheck.check(th1.get_exit_code() == 0, "退出码 == 0", "退出码 != 0")
|
|
th1.close()
|
|
|
|
# ========== Thread: pass arg ==========
|
|
testcheck.section("Thread: arg 传参")
|
|
cnt: int = 0
|
|
th2: vthreading.Thread = vthreading.Thread(_increment_worker, &cnt)
|
|
th2.start()
|
|
testcheck.check(th2.join(5000) != FALSE, "join 正常返回", "join 超时")
|
|
th2.close()
|
|
testcheck.check(cnt == 100, "worker 经 arg 自增 100 次", "cnt != 100")
|
|
|
|
# ========== Lock: mutex increment ==========
|
|
testcheck.section("Lock: 多线程互斥自增")
|
|
lock: vthreading.Lock = vthreading.Lock()
|
|
t0: vthreading.Thread = vthreading.Thread(_lock_worker, rawptr(lock))
|
|
t1: vthreading.Thread = vthreading.Thread(_lock_worker, rawptr(lock))
|
|
t2: vthreading.Thread = vthreading.Thread(_lock_worker, rawptr(lock))
|
|
t3: vthreading.Thread = vthreading.Thread(_lock_worker, rawptr(lock))
|
|
t0.start()
|
|
t1.start()
|
|
t2.start()
|
|
t3.start()
|
|
j0: int = t0.join(10000)
|
|
j1: int = t1.join(10000)
|
|
j2: int = t2.join(10000)
|
|
j3: int = t3.join(10000)
|
|
t0.close()
|
|
t1.close()
|
|
t2.close()
|
|
t3.close()
|
|
testcheck.check(j0 != 0 and j1 != 0 and j2 != 0 and j3 != 0, "4 线程全部 join 成功", "有线程 join 超时")
|
|
testcheck.check(_shared_count == THREAD_NUM * LOCK_ITER, "互斥自增计数正确", "计数丢失")
|
|
|
|
# ========== Event: thread waits for wakeup ==========
|
|
testcheck.section("Event: 线程等待唤醒")
|
|
ev: vthreading.Event = vthreading.Event(FALSE, FALSE)
|
|
th4: vthreading.Thread = vthreading.Thread(_event_worker, rawptr(ev))
|
|
testcheck.check(ev.is_set() == FALSE, "Event 初始未置位", "Event 初始已置位")
|
|
th4.start()
|
|
vthreading.sleep_ms(50)
|
|
testcheck.check(_evt_done == 0, "set 前 worker 仍等待", "worker 提前完成")
|
|
ev.set()
|
|
testcheck.check(th4.join(5000) != FALSE, "唤醒后 join 成功", "join 超时")
|
|
th4.close()
|
|
ev.close()
|
|
testcheck.check(_evt_done == 1, "worker 收到事件并置位", "worker 未收到事件")
|
|
|
|
# ========== Condition: producer/consumer ==========
|
|
testcheck.section("Condition: 生产者/消费者")
|
|
cond: vthreading.Condition = vthreading.Condition()
|
|
tc: vthreading.Thread = vthreading.Thread(_cond_consumer, rawptr(cond))
|
|
tp: vthreading.Thread = vthreading.Thread(_cond_producer, rawptr(cond))
|
|
tc.start()
|
|
tp.start()
|
|
testcheck.check(tc.join(5000) != FALSE, "消费者被唤醒", "消费者等待超时")
|
|
testcheck.check(tp.join(5000) != FALSE, "生产者完成", "生产者超时")
|
|
tc.close()
|
|
tp.close()
|
|
testcheck.check(_cond_ready == 1, "条件变量值已置位", "条件变量值未置位")
|
|
cond.delete()
|
|
|
|
# ========== RWLock: multi-reader single-writer ==========
|
|
testcheck.section("RWLock: 多读单写")
|
|
rwl: vthreading.RWLock = vthreading.RWLock()
|
|
r1: vthreading.Thread = vthreading.Thread(_rw_reader, rawptr(rwl))
|
|
r2: vthreading.Thread = vthreading.Thread(_rw_reader, rawptr(rwl))
|
|
w1: vthreading.Thread = vthreading.Thread(_rw_writer, rawptr(rwl))
|
|
r1.start()
|
|
r2.start()
|
|
w1.start()
|
|
testcheck.check(r1.join(10000) != FALSE and r2.join(10000) != FALSE and w1.join(10000) != FALSE,
|
|
"读写线程全部 join", "读写线程 join 超时")
|
|
r1.close()
|
|
r2.close()
|
|
w1.close()
|
|
testcheck.check(_rw_value == RW_ITER, "写锁自增计数正确", "写锁计数丢失")
|
|
|
|
# ========== local: TLS thread-local storage ==========
|
|
testcheck.section("local: TLS 线程局部存储")
|
|
i_t: int
|
|
for i_t in range(TLS_NUM):
|
|
th_t: vthreading.Thread = vthreading.Thread(_tls_worker, NULL)
|
|
th_t.start()
|
|
testcheck.check(th_t.join(5000) != FALSE, "TLS worker join", "TLS worker 超时")
|
|
testcheck.check(th_t.get_exit_code() == 1, "TLS set/get 自校验通过", "TLS 自校验失败")
|
|
th_t.close()
|
|
|
|
# ========== get_current_thread_id: thread ID isolation ==========
|
|
testcheck.section("get_current_thread_id: 线程 ID 隔离")
|
|
global _main_tid
|
|
_main_tid = vthreading.get_current_thread_id()
|
|
testcheck.check(_main_tid != 0, "主线程 ID != 0", "主线程 ID == 0")
|
|
th_tid: vthreading.Thread = vthreading.Thread(_tid_worker, NULL)
|
|
th_tid.start()
|
|
testcheck.check(th_tid.join(5000) != FALSE, "tid worker join", "tid worker 超时")
|
|
testcheck.check(th_tid.get_exit_code() == 1, "工作线程 ID 与主线程不同", "工作线程 ID 与主线程相同")
|
|
th_tid.close()
|
|
|
|
# ========== sleep_ms ==========
|
|
testcheck.section("sleep_ms")
|
|
vthreading.sleep_ms(20)
|
|
testcheck.ok("sleep_ms(20) 正常返回")
|
|
|
|
# ========== stdlib.system concurrency ==========
|
|
testcheck.section("stdlib.system 多线程并发")
|
|
s0: vthreading.Thread = vthreading.Thread(_system_worker, NULL)
|
|
s1: vthreading.Thread = vthreading.Thread(_system_worker, NULL)
|
|
s2: vthreading.Thread = vthreading.Thread(_system_worker, NULL)
|
|
s3: vthreading.Thread = vthreading.Thread(_system_worker, NULL)
|
|
s0.start()
|
|
s1.start()
|
|
s2.start()
|
|
s3.start()
|
|
testcheck.check(s0.join(30000) != FALSE and s1.join(30000) != FALSE
|
|
and s2.join(30000) != FALSE and s3.join(30000) != FALSE,
|
|
"4 个 system 线程全部 join", "system 线程 join 超时")
|
|
testcheck.check(s0.get_exit_code() == 0, "线程0 system 全部成功", "线程0 system 有失败")
|
|
testcheck.check(s1.get_exit_code() == 0, "线程1 system 全部成功", "线程1 system 有失败")
|
|
testcheck.check(s2.get_exit_code() == 0, "线程2 system 全部成功", "线程2 system 有失败")
|
|
testcheck.check(s3.get_exit_code() == 0, "线程3 system 全部成功", "线程3 system 有失败")
|
|
s0.close()
|
|
s1.close()
|
|
s2.close()
|
|
s3.close()
|
|
|
|
return testcheck.end()
|