Step 18 complete: Software Transactional Memory (STM)
ober
339dbe57856526975aca7b044f9c28c2ca515edf
--- a/Makefile +++ b/Makefile @@ -86,6 +86,7 @@ test-features: @$(SCHEME) --libdirs $(LIBDIRS) --script tests/test-effect.ss @$(SCHEME) --libdirs $(LIBDIRS) --script tests/test-async.ss @$(SCHEME) --libdirs $(LIBDIRS) --script tests/test-iouring.ss + @$(SCHEME) --libdirs $(LIBDIRS) --script tests/test-stm.ss test-all: test test-features test-wrappers new file mode 100644 --- /dev/null +++ b/lib/std/stm.sls @@ -0,0 +1,188 @@ +#!chezscheme +;;; (std stm) — Software Transactional Memory +;;; +;;; Optimistic concurrency using versioned transactional variables (TVars). +;;; Transactions run speculatively, validate their read-set at commit time, +;;; and retry on conflict. Lock-free reads, single global commit mutex. +;;; +;;; API: +;;; (make-tvar init) — create a transactional variable +;;; (tvar? x) — predicate +;;; (tvar-ref tv) — read outside transaction (unsafe) +;;; (atomically body ...) — run body as atomic transaction +;;; (tvar-read tv) — read within transaction +;;; (tvar-write! tv val) — write within transaction +;;; (retry) — abort and block until a TVar changes +;;; (or-else expr1 expr2) — try expr1; if it retries, try expr2 + +(library (std stm) + (export + make-tvar + tvar? + tvar-ref + atomically + tvar-read + tvar-write! + retry + or-else) + + (import (chezscheme)) + + ;; ========== TVar ========== + + (define-record-type (stm-tvar %make-tvar tvar?) + (fields + (mutable stm-value) ; current committed value + (mutable stm-version)) ; monotonically increasing version counter + (sealed #t)) + + (define (make-tvar init) + (%make-tvar init 0)) + + ;; Read the current value of a TVar outside a transaction. + ;; Not safe to use inside atomically (use tvar-read instead). + (define (tvar-ref tv) (stm-tvar-stm-value tv)) + + ;; ========== Global Commit Infrastructure ========== + ;; + ;; Single global mutex serialises all commit attempts. + ;; After each successful commit we broadcast on *commit-cond* + ;; so that threads blocked in retry can wake up and re-run. + + (define *commit-mutex* (make-mutex)) + (define *commit-cond* (make-condition)) + + ;; ========== Transaction State ========== + ;; + ;; read-set : assoc list of (tvar . version-seen-at-first-read) + ;; write-set : assoc list of (tvar . new-value) + + (define-record-type tx-rec + (fields + (mutable tx-read-set) + (mutable tx-write-set)) + (sealed #t)) + + (define (make-tx) (make-tx-rec '() '())) + + ;; Thread-local: the currently active transaction, or #f. + (define *current-tx* (make-thread-parameter #f)) + + ;; ========== Retry Condition ========== + + (define-condition-type &stm-retry &condition + make-stm-retry stm-retry?) + + (define (retry) + (raise (make-stm-retry))) + + ;; ========== tvar-read ========== + + (define (tvar-read tv) + (let ([tx (*current-tx*)]) + (if (not tx) + ;; Outside transaction: direct read (not transactional) + (stm-tvar-stm-value tv) + ;; 1. Check write-set (our own writes take priority) + (let ([we (assq tv (tx-rec-tx-write-set tx))]) + (if we + (cdr we) + ;; 2. Check read-set (already snapshotted this tvar) + (let ([re (assq tv (tx-rec-tx-read-set tx))]) + (if re + ;; Return current value (version mismatch is caught at commit) + (stm-tvar-stm-value tv) + ;; 3. First read: snapshot version + value + (let* ([ver (stm-tvar-stm-version tv)] + [val (stm-tvar-stm-value tv)]) + (tx-rec-tx-read-set-set! tx + (cons (cons tv ver) (tx-rec-tx-read-set tx))) + val)))))))) + + ;; ========== tvar-write! ========== + + (define (tvar-write! tv val) + (let ([tx (*current-tx*)]) + (if (not tx) + ;; Outside transaction: direct (non-transactional) write + (with-mutex *commit-mutex* + (stm-tvar-stm-value-set! tv val) + (stm-tvar-stm-version-set! tv (+ (stm-tvar-stm-version tv) 1)) + (condition-broadcast *commit-cond*)) + ;; Inside transaction: buffer in write-set + (let ([entry (assq tv (tx-rec-tx-write-set tx))]) + (if entry + (set-cdr! entry val) + (tx-rec-tx-write-set-set! tx + (cons (cons tv val) (tx-rec-tx-write-set tx)))))))) + + ;; ========== Commit ========== + ;; + ;; Acquires *commit-mutex*, validates read-set, applies writes. + ;; Returns #t on success, #f on conflict. + + (define (tx-commit! tx) + (with-mutex *commit-mutex* + (let ([read-set (tx-rec-tx-read-set tx)] + [write-set (tx-rec-tx-write-set tx)]) + ;; Validate: check that all snapshotted versions are still current + (let ([valid? (for-all (lambda (entry) + (= (stm-tvar-stm-version (car entry)) + (cdr entry))) + read-set)]) + (when valid? + ;; Apply writes and bump versions + (for-each (lambda (entry) + (stm-tvar-stm-value-set! (car entry) (cdr entry)) + (stm-tvar-stm-version-set! (car entry) + (+ (stm-tvar-stm-version (car entry)) 1))) + write-set) + ;; Wake all threads waiting in retry + (when (not (null? write-set)) + (condition-broadcast *commit-cond*))) + valid?)))) + + ;; ========== atomically ========== + ;; + ;; Runs thunk speculatively; commits on success; retries on conflict. + ;; Nested atomically flattens into the enclosing transaction. + + (define (%run-atomically thunk) + (if (*current-tx*) + ;; Already in a transaction: flatten (run in enclosing tx) + (thunk) + ;; New transaction + (let loop () + (let ([tx (make-tx)]) + (parameterize ([*current-tx* tx]) + (guard (exn + [(stm-retry? exn) + ;; Block until some TVar is modified, then retry + (mutex-acquire *commit-mutex*) + (condition-wait *commit-cond* *commit-mutex*) + (mutex-release *commit-mutex*) + (loop)] + [#t (raise exn)]) + (let ([result (thunk)]) + (if (tx-commit! tx) + result + ;; Conflict: retry immediately + (loop))))))))) + + (define-syntax atomically + (syntax-rules () + [(_ body ...) + (%run-atomically (lambda () body ...))])) + + ;; ========== or-else ========== + ;; + ;; Try the first expression; if it calls retry, try the second. + ;; Both run within the same enclosing transaction (if any). + + (define-syntax or-else + (syntax-rules () + [(_ expr1 expr2) + (guard (exn [(stm-retry? exn) expr2]) + expr1)])) + + ) ;; end library new file mode 100644 --- /dev/null +++ b/tests/test-stm.ss @@ -0,0 +1,187 @@ +#!chezscheme +;;; Tests for (std stm) — Software Transactional Memory + +(import (chezscheme) (std stm)) + +(define pass 0) +(define fail 0) + +(define-syntax test + (syntax-rules () + [(_ name expr expected) + (guard (exn [#t (set! fail (+ fail 1)) + (printf "FAIL ~a: ~a~%" name + (if (message-condition? exn) (condition-message exn) exn))]) + (let ([got expr]) + (if (equal? got expected) + (begin (set! pass (+ pass 1)) (printf " ok ~a~%" name)) + (begin (set! fail (+ fail 1)) + (printf "FAIL ~a: got ~s expected ~s~%" name got expected)))))])) + +(printf "--- (std stm) tests ---~%") + +;; Basic TVar operations +(printf "~%-- TVar basics --~%") + +(test "make-tvar + tvar-ref" + (let ([tv (make-tvar 42)]) + (tvar-ref tv)) + 42) + +(test "tvar?" + (tvar? (make-tvar 0)) + #t) + +(test "tvar? on non-tvar" + (tvar? 42) + #f) + +;; atomically: read and write +(printf "~%-- atomically: basic read/write --~%") + +(test "tvar-read outside transaction (falls back to direct read)" + (let ([tv (make-tvar 99)]) + (tvar-read tv)) + 99) + +(test "atomically: read" + (let ([tv (make-tvar 10)]) + (atomically (tvar-read tv))) + 10) + +(test "atomically: write then read" + (let ([tv (make-tvar 1)]) + (atomically + (tvar-write! tv 42) + (tvar-read tv))) + 42) + +(test "atomically: write persists after transaction" + (let ([tv (make-tvar 0)]) + (atomically (tvar-write! tv 100)) + (tvar-ref tv)) + 100) + +(test "atomically: multiple TVars" + (let ([a (make-tvar 1)] + [b (make-tvar 2)]) + (atomically + (tvar-write! a 10) + (tvar-write! b 20)) + (list (tvar-ref a) (tvar-ref b))) + '(10 20)) + +(test "atomically: read then write" + (let ([tv (make-tvar 5)]) + (atomically + (let ([v (tvar-read tv)]) + (tvar-write! tv (* v 2)))) + (tvar-ref tv)) + 10) + +;; Nested transactions flatten into parent +(printf "~%-- nested atomically --~%") + +(test "nested atomically: flattens into parent" + (let ([tv (make-tvar 0)]) + (atomically + (atomically (tvar-write! tv 1)) + (atomically (tvar-write! tv (+ (tvar-read tv) 1)))) + (tvar-ref tv)) + 2) + +;; Composable transfers +(printf "~%-- composable transfers --~%") + +(define (transfer! from to amount) + (atomically + (let ([f (tvar-read from)] + [t (tvar-read to)]) + (when (< f amount) + (error 'transfer! "insufficient funds" f amount)) + (tvar-write! from (- f amount)) + (tvar-write! to (+ t amount))))) + +(test "transfer: basic" + (let ([a (make-tvar 1000)] + [b (make-tvar 500)]) + (transfer! a b 300) + (list (tvar-ref a) (tvar-ref b))) + '(700 800)) + +(test "transfer: composed (two in one atomically)" + (let ([a (make-tvar 1000)] + [b (make-tvar 500)] + [c (make-tvar 200)]) + (atomically + (transfer! a b 100) + (transfer! b c 50)) + (list (tvar-ref a) (tvar-ref b) (tvar-ref c))) + '(900 550 250)) + +;; Concurrent correctness: two threads doing increments +(printf "~%-- concurrent correctness --~%") + +(test "concurrent increments" + (let ([counter (make-tvar 0)] + [n 100]) + (define (increment!) + (atomically + (let ([v (tvar-read counter)]) + (tvar-write! counter (+ v 1))))) + ;; Run n increments in two threads + (let ([t1 (fork-thread + (lambda () + (let loop ([i 0]) + (when (< i (quotient n 2)) + (increment!) + (loop (+ i 1))))))] + [t2 (fork-thread + (lambda () + (let loop ([i 0]) + (when (< i (quotient n 2)) + (increment!) + (loop (+ i 1))))))]) + ;; Wait for threads to finish + (let ([m (make-mutex)] + [c (make-condition)] + [done 0]) + (define (thread-done!) + (with-mutex m + (set! done (+ done 1)) + (condition-broadcast c))) + ;; Can't join threads in Chez directly; use a simpler approach: + ;; Just sleep and check + (sleep (make-time 'time-duration 200000000 0)) + (tvar-ref counter)))) + 100) + +;; or-else +(printf "~%-- or-else --~%") + +(test "or-else: first succeeds" + (atomically + (or-else + 42 + 99)) + 42) + +(test "or-else: first retries, second runs" + (let ([flag (make-tvar #t)]) + (atomically + (or-else + (if (tvar-read flag) 'first (retry)) + 'second))) + 'first) + +(test "or-else: first retries (flag=#f), second runs" + (let ([flag (make-tvar #f)]) + (atomically + (or-else + (if (tvar-read flag) 'first (retry)) + 'second))) + 'second) + +(printf "~%~a tests: ~a passed, ~a failed~%" + (+ pass fail) pass fail) +(when (> fail 0) (exit 1))