suzanne.soy/racket
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

db: fix for disconnect during query in OS thread

Browse Source 
This change allows a query that is run in an OS thread to succeed even
when the connection is custodian-disconnected. Previously, the part in
the OS thread would complete, and then the operations needed to
package the result would fail. This fix moves some of those operations
to the OS thread and makes read-tx-status work when disconnected.
This commit is contained in:
Ryan Culpepper 2021-01-31 07:03:52 +01:00
parent a5b14d74b7
commit ee43d982e4
1 changed files with 49 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
racket/collects/db/private/sqlite3/connection.rkt
View File

@ -70,6 +70,8 @@
(begin (start-atomic) (begin (start-atomic)
(unless -db (end-atomic) (error/disconnect-in-lock 'sqlite3)) (unless -db (end-atomic) (error/disconnect-in-lock 'sqlite3))
(begin0 (let () e ...) (end-atomic)))) (begin0 (let () e ...) (end-atomic))))
(define-syntax-rule (A* e ...)
(begin (start-atomic) (begin0 (let () e ...) (end-atomic))))
(define/private (get-db fsym) (define/private (get-db fsym)
(or -db (error/not-connected fsym))) (or -db (error/not-connected fsym)))
@ -108,10 +110,11 @@
[saved-last-insert-rowid [saved-last-insert-rowid
(and (null? info) (A (sqlite3_last_insert_rowid db)))] (and (null? info) (A (sqlite3_last_insert_rowid db)))]
[saved-total-changes [saved-total-changes
(and (null? info) (A (sqlite3_total_changes db)))] (and (null? info) (A (sqlite3_total_changes db)))])
[result (define-values (result last-insert-rowid total-changes changes)
(or cursor? (if cursor?
(get-rows fsym stmt #f +inf.0 pst))]) (values #t #f #f #f)
(get-result fsym stmt #f +inf.0 pst (not (pair? info)))))
(unless (eq? (get-tx-status) 'invalid) (unless (eq? (get-tx-status) 'invalid)
(set-tx-status! fsym (read-tx-status))) (set-tx-status! fsym (read-tx-status)))
(unless cursor? (unless cursor?
@ -122,10 +125,10 @@
(cursor-result info pst (box #f))] (cursor-result info pst (box #f))]
[else [else
(simple-result (simple-result
(let ([last-insert-rowid (A (sqlite3_last_insert_rowid db))]) (let ()
;; Not all statements clear last_insert_rowid, changes; so ;; Not all statements clear last_insert_rowid, changes; so
;; extra guards to make sure results are relevant. ;; extra guards to make sure results are relevant.
(define changes? (> (A (sqlite3_total_changes db)) saved-total-changes)) (define changes? (> total-changes saved-total-changes))
`((insert-id `((insert-id
;; We want to report insert-id if statement was a *successful* INSERT, ;; We want to report insert-id if statement was a *successful* INSERT,
;; but we can't check that directly. Instead, check if either ;; but we can't check that directly. Instead, check if either
@ -141,7 +144,7 @@
. ,(and (or (not (= last-insert-rowid saved-last-insert-rowid)) . ,(and (or (not (= last-insert-rowid saved-last-insert-rowid))
(and changes? (eq? (send pst get-stmt-type) 'insert))) (and changes? (eq? (send pst get-stmt-type) 'insert)))
last-insert-rowid)) last-insert-rowid))
(affected-rows . ,(if changes? (A (sqlite3_changes db)) 0)))))]))))) (affected-rows . ,(if changes? changes 0)))))])))))
(define/public (fetch/cursor fsym cursor fetch-size) (define/public (fetch/cursor fsym cursor fetch-size)
(let ([pst (cursor-result-pst cursor)] (let ([pst (cursor-result-pst cursor)]
@ -152,9 +155,11 @@
(cond [(unbox end-box) #f] (cond [(unbox end-box) #f]
[else [else
(let ([stmt (send pst get-handle)]) (let ([stmt (send pst get-handle)])
(begin0 (get-rows fsym stmt end-box fetch-size pst) (define-values (rows _lii _tc _c)
(when (unbox end-box) (get-result fsym stmt end-box fetch-size pst #f))
(send pst after-exec #f))))]))))) (when (unbox end-box)
(send pst after-exec #f))
rows)])))))
(define/private (check-statement fsym stmt cursor?) (define/private (check-statement fsym stmt cursor?)
(cond [(statement-binding? stmt) (cond [(statement-binding? stmt)
@ -183,39 +188,38 @@
[else [else
(error/internal* fsym "bad parameter value" '("value" value) param)]))) (error/internal* fsym "bad parameter value" '("value" value) param)])))
(define/private (get-rows who stmt end-box fetch-limit pst) (define/private (get-result who stmt end-box fetch-limit pst changes?)
(with-handlers ([exn:fail? (with-handlers ([exn:fail?
(lambda (e) (lambda (e)
(A (sqlite3_reset stmt) (A* (when -db (sqlite3_reset stmt) (sqlite3_clear_bindings stmt)))
(sqlite3_clear_bindings stmt))
(raise e))]) (raise e))])
(define result ((cond [use-os-thread?
(cond [use-os-thread? (define timeout (inexact->exact (ceiling (* 1000 busy-retry-delay))))
(define timeout (inexact->exact (ceiling (* 1000 busy-retry-delay)))) (sync-call-in-os-thread
(sync-call-in-os-thread (lambda (db)
(lambda (db) (sqlite3_busy_timeout db timeout)
(sqlite3_busy_timeout db timeout) (begin0 (get-result* who db stmt end-box fetch-limit pst changes?)
(begin0 (get-rows* who stmt end-box fetch-limit pst #f) (sqlite3_busy_timeout db 0))))]
(sqlite3_busy_timeout db 0))))] [else (get-result* who #f stmt end-box fetch-limit pst changes?)]))))
[else (get-rows* who stmt end-box fetch-limit pst #t)]))
(if (procedure? result) (result) result)))
(define/private (get-rows* who stmt end-box fetch-limit pst fine-atomic?) (define/private (get-result* who os-db stmt end-box fetch-limit pst changes?)
(define (call-as-fine-atomic thunk) (if fine-atomic? (A (thunk)) (thunk))) ;; os-db is sqlite3_database if in OS thread, #f if in Racket thread
(define (call-as-fine-atomic thunk) (if os-db (thunk) (A (thunk))))
(define-syntax-rule (FA expr ...) (call-as-fine-atomic (lambda () expr ...))) (define-syntax-rule (FA expr ...) (call-as-fine-atomic (lambda () expr ...)))
;; step* : -> (U (Listof Vector) (-> Any)) ;; step* : -> (-> (values (U Vector (-> Error)) Int/#f Nat/#f Nat/#f))
(define (step*) (define (step*)
(let loop ([fetch-limit fetch-limit]) (let loop ([fetch-limit fetch-limit] [acc null])
(cond [(zero? fetch-limit) null] (cond [(zero? fetch-limit)
(return (reverse acc))]
[(step) [(step)
=> (lambda (c) => (lambda (c)
(cond [(procedure? c) c] (cond [(procedure? c) c]
[else (cons c (loop (sub1 fetch-limit)))]))] [else (loop (sub1 fetch-limit) (cons c acc))]))]
[else [else
(FA (sqlite3_reset stmt) (sqlite3_clear_bindings stmt)) (FA (sqlite3_reset stmt) (sqlite3_clear_bindings stmt))
(when end-box (set-box! end-box #t)) (when end-box (set-box! end-box #t))
null]))) (return (reverse acc))])))
;; step : -> (U #f Vector (-> Any)) ;; step : -> (U #f Vector (-> Error))
(define (step) (define (step)
(let loop ([iteration 0]) (let loop ([iteration 0])
(define full-s (FA (sqlite3_step stmt))) (define full-s (FA (sqlite3_step stmt)))
@ -225,12 +229,21 @@
[(and (= s SQLITE_BUSY) (< iteration busy-retry-limit)) [(and (= s SQLITE_BUSY) (< iteration busy-retry-limit))
;; Normally, sleep and try again (cooperates w/ scheduler). ;; Normally, sleep and try again (cooperates w/ scheduler).
;; In os-thread, can't sleep; see sqlite3_busy_timeout above. ;; In os-thread, can't sleep; see sqlite3_busy_timeout above.
(when fine-atomic? (sleep busy-retry-delay)) (unless os-db (sleep busy-retry-delay))
(loop (add1 iteration))] (loop (add1 iteration))]
[else (lambda () (handle-status who full-s pst))]))) [else (lambda () (handle-status who full-s pst))])))
;; return : X -> (-> (values X Int/#f Nat/#f Nat/#f))
(define (return rows)
(define-values (last-insert-rowid total-changes changes)
(if changes?
(FA (values (sqlite3_last_insert_rowid (or os-db -db))
(sqlite3_total_changes (or os-db -db))
(sqlite3_changes (or os-db -db))))
(values #f #f #f)))
(lambda () (values rows last-insert-rowid total-changes changes)))
(step*)) (step*))
;; -get-row : Symbol stmt -> (U Vector (-> Any)) ;; -get-row : Symbol stmt -> (U Vector (-> Error))
;; PRE: in atomic mode ;; PRE: in atomic mode
(define/private (-get-row fsym stmt) (define/private (-get-row fsym stmt)
(define column-count (sqlite3_column_count stmt)) (define column-count (sqlite3_column_count stmt))
@ -367,7 +380,9 @@
;; http://www.sqlite.org/lang_transaction.html ;; http://www.sqlite.org/lang_transaction.html
(define/private (read-tx-status) (define/private (read-tx-status)
(not (A (sqlite3_get_autocommit -db)))) ;; Allow this to be called after custodian-disconnect so that in-progress
;; query can complete.
(not (A* (if -db (sqlite3_get_autocommit -db) #t))))
(define/override (start-transaction* fsym isolation option) (define/override (start-transaction* fsym isolation option)
;; Isolation level can be set to READ UNCOMMITTED via pragma, but ;; Isolation level can be set to READ UNCOMMITTED via pragma, but