perf: stm read/write sets use eq?-hashtables (O(1) per op, was O(k) assq)
ober
c630d11243dab5345f796db02f951d466a2b0428
--- a/stm.ss +++ b/stm.ss @@ -12,7 +12,7 @@ (export make-tvar tvar? tvar-ref atomically tvar-read tvar-write! retry or-else make-ref ref? ref-deref dosync alter ref-set commute ensure io!) -(import :std/sugar :std/sort) +(import :std/sugar :std/sort :std/misc/hash) ;; ---- tvar (internal struct name tvx; public names aliased below) ---- (def *tvar-id* 0) @@ -55,9 +55,13 @@ (for-each waiter-signal! ws))) ;; ---- transaction state ---- +;; reads: eq?-hashtable tvar -> (version . value) +;; writes: eq?-hashtable tvar -> value +;; Hashtables keep tvar-read/tvar-write! O(1) per op instead of O(k) assq scans. (defstruct txr (reads writes)) -(def (make-tx) (make-txr '() '())) +(def (make-tx) (make-txr (make-hash-table test: eq?) (make-hash-table test: eq?))) (def *current-tx* (make-parameter #f)) +(def *tx-missing* (list 'tx-missing)) (defstruct stm-retry-exn ()) (def (retry) (raise (make-stm-retry-exn))) @@ -70,10 +74,10 @@ (let (tx (*current-tx*)) (if (not tx) (tvx-value tv) - (let (we (assq tv (txr-writes tx))) - (if we (cdr we) - (let (re (assq tv (txr-reads tx))) - (if re (caddr re) + (let (wv (hash-ref (txr-writes tx) tv *tx-missing*)) + (if (not (eq? wv *tx-missing*)) wv + (let (re (hash-ref (txr-reads tx) tv #f)) + (if re (cdr re) (let (snap (let (mx (tvx-lock tv)) (mutex-lock! mx) (let ((v (tvx-version tv)) (x (tvx-value tv))) @@ -82,7 +86,7 @@ ;; Retain both version and value so repeated reads are opaque: ;; they never observe a value from a transaction that changed ;; the TVar after our first read. - (txr-reads-set! tx (cons (list tv (car snap) (cdr snap)) (txr-reads tx))) + (hash-put! (txr-reads tx) tv snap) (cdr snap))))))))) (def (tvar-write! tv val) @@ -94,9 +98,7 @@ (tvx-version-set! tv (+ (tvx-version tv) 1)) (notify-waiters! tv) (mutex-unlock! mx)) - (let (entry (assq tv (txr-writes tx))) - (if entry (set-cdr! entry val) - (txr-writes-set! tx (cons (cons tv val) (txr-writes tx)))))))) + (hash-put! (txr-writes tx) tv val)))) (def (tx-commit! tx) (let ((reads (txr-reads tx)) (writes (txr-writes tx))) @@ -104,22 +106,25 @@ ;; every read dependency locked through validation and publication closes ;; the window where another transaction could change a previously checked ;; TVar before our writes became visible. - (let* ((lock-set - (let loop ((pending writes) (all reads)) - (if (null? pending) all - (let (entry (car pending)) - (loop (cdr pending) - (if (assq (car entry) all) all (cons entry all))))))) + (let* ((read-list + (hash-fold (lambda (tv vv acc) (cons (list tv (car vv)) acc)) '() reads)) + (write-list + (hash-fold (lambda (tv val acc) (cons (cons tv val) acc)) '() writes)) + (lock-tvars + (let (ht (make-hash-table test: eq?)) + (for-each (lambda (e) (hash-put! ht (car e) #t)) read-list) + (for-each (lambda (e) (hash-put! ht (car e) #t)) write-list) + (hash-keys ht))) (sorted-locks - (sort lock-set (lambda (a b) (< (tvx-id (car a)) (tvx-id (car b)))))) + (sort lock-tvars (lambda (a b) (< (tvx-id a) (tvx-id b))))) (sorted-writes - (sort writes (lambda (a b) (< (tvx-id (car a)) (tvx-id (car b))))))) - (for-each (lambda (e) (mutex-lock! (tvx-lock (car e)))) sorted-locks) + (sort write-list (lambda (a b) (< (tvx-id (car a)) (tvx-id (car b))))))) + (for-each (lambda (tv) (mutex-lock! (tvx-lock tv))) sorted-locks) (try (let (valid? (every? (lambda (e) (= (tvx-version (car e)) (cadr e))) - reads)) + read-list)) (if valid? (begin (for-each (lambda (e) @@ -131,16 +136,16 @@ #t) #f)) (finally - (for-each (lambda (e) (mutex-unlock! (tvx-lock (car e)))) + (for-each (lambda (tv) (mutex-unlock! (tvx-lock tv))) (reverse sorted-locks))))))) (def (stm-unregister-waiter! reads w) - (for-each (lambda (e) - (let* ((tv (car e)) (mx (tvx-lock tv))) - (mutex-lock! mx) - (tvx-waiters-set! tv (remq-eq w (tvx-waiters tv))) - (mutex-unlock! mx))) - reads)) + (hash-for-each (lambda (tv vv) + (let (mx (tvx-lock tv)) + (mutex-lock! mx) + (tvx-waiters-set! tv (remq-eq w (tvx-waiters tv))) + (mutex-unlock! mx))) + reads)) ;; Register a waiter on every read TVar and, while still holding each TVar's ;; lock, re-validate its recorded version. A version that moved since the read @@ -151,16 +156,16 @@ ;; blocking on a delivered notification. (def (stm-wait-on-reads! tx) (let ((reads (txr-reads tx)) (w (make-thread-waiter))) - (and (not (null? reads)) + (and (not (= (hash-length reads) 0)) (let (stale? - (let loop ((rs reads) (stale? #f)) - (if (null? rs) stale? - (let* ((e (car rs)) (tv (car e)) (mx (tvx-lock tv))) - (mutex-lock! mx) - (tvx-waiters-set! tv (cons w (tvx-waiters tv))) - (let (moved? (not (= (tvx-version tv) (cadr e)))) - (mutex-unlock! mx) - (loop (cdr rs) (or stale? moved?))))))) + (hash-fold (lambda (tv vv stale?) + (let (mx (tvx-lock tv)) + (mutex-lock! mx) + (tvx-waiters-set! tv (cons w (tvx-waiters tv))) + (let (moved? (not (= (tvx-version tv) (car vv)))) + (mutex-unlock! mx) + (or stale? moved?)))) + #f reads)) (if stale? (begin (stm-unregister-waiter! reads w) #f) (begin (waiter-wait! w) (stm-unregister-waiter! reads w) #t))))))