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

db: added #:use-place arg for ffi-based connections

Browse Source 
SQLite and ODBC connections can use places to avoid blocking all
Racket threads.
This commit is contained in:
Ryan Culpepper 2011-08-30 17:41:19 -06:00
parent 5f0a221a03
commit 96663d4fa4
16 changed files with 532 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
collects/db/TODO
View File

@ -67,13 +67,6 @@ Misc
- how do people want to use cursors?
- how about implicit support only in 'in-query'?
- ODBC: use async execution to avoid blocking all Racket threads
Status: Tried it. Oracle driver doesn't support async. PG, MY drivers don't support async.
DB2 driver does, but gives baffling HY010 function sequence errors, couldn't fix.
(Best theory so far: possible that DB2 requires polling args to be identical to original
call, which means (_ptr o X) args are the problem. Or maybe unixodbc's fault.)
Didn't try SQL Server. All in all, not worth it.
- add evt versions of functions
- for query functions (?)
- connection-pool-lease-evt

9
collects/db/main.rkt
View File

@ -75,7 +75,8 @@
(->* (#:database (or/c path-string? 'memory 'temporary))
(#:mode (or/c 'read-only 'read/write 'create)
#:busy-retry-limit (or/c exact-nonnegative-integer? +inf.0)
#:busy-retry-delay (and/c rational? (not/c negative?)))
#:busy-retry-delay (and/c rational? (not/c negative?))
#:use-place boolean?)
any/c)]
;; Duplicates contracts at odbc.rkt
@ -85,13 +86,15 @@
#:password (or/c string? #f)
#:notice-handler (or/c 'output 'error output-port? procedure?)
#:strict-parameter-types? boolean?
#:character-mode (or/c 'wchar 'utf-8 'latin-1))
#:character-mode (or/c 'wchar 'utf-8 'latin-1)
#:use-place boolean?)
connection?)]
[odbc-driver-connect
(->* (string?)
(#:notice-handler (or/c 'output 'error output-port? procedure?)
#:strict-parameter-types? boolean?
#:character-mode (or/c 'wchar 'utf-8 'latin-1))
#:character-mode (or/c 'wchar 'utf-8 'latin-1)
#:use-place boolean?)
connection?)]
[odbc-data-sources
(-> (listof (list/c string? string?)))]

6
collects/db/odbc.rkt
View File

@ -11,13 +11,15 @@
#:password (or/c string? #f)
#:notice-handler (or/c 'output 'error output-port? procedure?)
#:strict-parameter-types? boolean?
#:character-mode (or/c 'wchar 'utf-8 'latin-1))
#:character-mode (or/c 'wchar 'utf-8 'latin-1)
#:use-place boolean?)
connection?)]
[odbc-driver-connect
(->* (string?)
(#:notice-handler (or/c 'output 'error output-port? procedure?)
#:strict-parameter-types? boolean?
#:character-mode (or/c 'wchar 'utf-8 'latin-1))
#:character-mode (or/c 'wchar 'utf-8 'latin-1)
#:use-place boolean?)
connection?)]
[odbc-data-sources
(-> (listof (list/c string? string?)))]

10
collects/db/private/generic/dsn.rkt
View File

@ -189,12 +189,12 @@ considered important.
(define sqlite3-data-source
(mk-specialized 'sqlite3-data-source 'sqlite3 0
'(#:database #:mode #:busy-retry-limit #:busy-retry-delay)))
'(#:database #:mode #:busy-retry-limit #:busy-retry-delay #:use-place)))
(define odbc-data-source
(mk-specialized 'odbc-data-source 'odbc 0
'(#:dsn #:user #:password #:notice-handler
#:strict-parameter-types? #:character-mode)))
#:strict-parameter-types? #:character-mode #:use-place)))
(provide/contract
[struct data-source
@ -235,7 +235,8 @@ considered important.
(#:database (or/c string? 'memory 'temporary)
#:mode (or/c 'read-only 'read/write 'create)
#:busy-retry-limit (or/c exact-nonnegative-integer? +inf.0)
#:busy-retry-delay (and/c rational? (not/c negative?)))
#:busy-retry-delay (and/c rational? (not/c negative?))
#:use-place boolean?)
data-source?)]
[odbc-data-source
(->* ()
@ -244,5 +245,6 @@ considered important.
#:password string?
#:notice-handler (or/c 'output 'error)
#:strict-parameter-types? boolean?
#:character-mode (or/c 'wchar 'utf-8 'latin-1))
#:character-mode (or/c 'wchar 'utf-8 'latin-1)
#:use-place boolean?)
data-source?)])

137
collects/db/private/generic/place-client.rkt Normal file
View File

@ -0,0 +1,137 @@
#lang racket/base
(require racket/class
racket/match
racket/place
racket/promise
racket/vector
ffi/unsafe/atomic
"interfaces.rkt"
"prepared.rkt"
"sql-data.rkt")
(provide place-connect
place-proxy-connection%
sql-datum->sexpr
sexpr->sql-datum)
(define connection-server-channel
(delay/sync
(dynamic-place 'db/private/generic/place-server 'connection-server)))
(define (place-connect connection-spec proxy%)
(let-values ([(channel other-channel) (place-channel)])
(place-channel-put (force connection-server-channel)
(list 'connect other-channel connection-spec))
(match (place-channel-get channel)
[(list 'ok)
(new proxy% (channel channel))]
[(list 'error message)
(raise (make-exn:fail message (current-continuation-marks)))])))
(define place-proxy-connection%
(class* locking% (connection<%>)
(init-field channel)
(inherit call-with-lock
call-with-lock*)
(super-new)
(define/private (call method-name . args)
(call-with-lock method-name (lambda () (call* method-name args #t))))
(define/private (call/d method-name . args)
(call-with-lock* method-name (lambda () (call* method-name args #f)) #f #f))
(define/private (call* method-name args need-connected?)
(cond [channel
(place-channel-put channel (cons method-name args))
(match (place-channel-get channel)
[(cons 'values vals)
(apply values (for/list ([val (in-list vals)]) (translate-result val)))]
[(list 'error message)
(raise (make-exn:fail message (current-continuation-marks)))])]
[need-connected?
(unless channel
(error/not-connected method-name))]
[else (void)]))
(define/override (connected?)
;; FIXME: can underlying connection disconnect w/o us knowing?
(and channel #t))
(define/public (disconnect)
(call/d 'disconnect)
(set! channel #f))
(define/public (get-dbsystem) (error 'get-dbsystem "not implemented"))
(define/public (get-base) this)
(define/public (query fsym stmt)
(call 'query fsym
(match stmt
[(? string?) (list 'string stmt)]
[(statement-binding pst _meta params)
(list 'statement-binding
(send pst get-handle)
(map sql-datum->sexpr params))])))
(define/public (prepare fsym stmt close-on-exec?)
(call 'prepare fsym stmt close-on-exec?))
(define/public (free-statement pst)
(start-atomic)
(let ([handle (send pst get-handle)])
(send pst set-handle #f)
(end-atomic)
(when channel
(call/d 'free-statement handle))))
(define/public (transaction-status fsym)
(call 'transaction-status fsym))
(define/public (start-transaction fsym iso)
(call 'start-transaction fsym iso))
(define/public (end-transaction fsym mode)
(call 'end-transaction fsym mode))
(define/public (list-tables fsym schema)
(call 'list-tables fsym schema))
(define/private (translate-result x)
(match x
[(list 'simple-result y)
(simple-result y)]
[(list 'rows-result h rows)
(let ([rows
(for/list ([row (in-list rows)])
(vector-map sexpr->sql-datum row))])
(rows-result h rows))]
[(list 'prepared-statement handle close-on-exec? param-typeids result-dvecs)
(new prepared-statement%
(handle handle)
(close-on-exec? close-on-exec?)
(param-typeids param-typeids)
(result-dvecs result-dvecs)
(owner this))]
[_ x]))
))
(define (sql-datum->sexpr x)
(match x
[(? sql-null?)
'sql-null]
[(sql-date Y M D)
(list 'sql-date Y M D)]
[(sql-time h m s ns tz)
(list 'sql-time h m s ns tz)]
[(sql-timestamp Y M D h m s ns tz)
(list 'sql-timestamp Y M D h m s ns tz)]
;; FIXME: add sql-interval when implemented for odbc
[_ x]))
(define (sexpr->sql-datum x)
(match x
['sql-null sql-null]
[(list 'sql-date Y M D) (sql-date Y M D)]
[(list 'sql-time h m s ns tz) (sql-time h m s ns tz)]
[(list 'sql-timestamp Y M D h m s ns tz)
(sql-timestamp Y M D h m s ns tz)]
[else x]))

147
collects/db/private/generic/place-server.rkt Normal file
View File

@ -0,0 +1,147 @@
#lang racket/base
(require (for-syntax racket/base)
racket/class
racket/match
racket/place
"lazy-require.rkt"
"interfaces.rkt"
"prepared.rkt"
"sql-data.rkt"
"place-client.rkt")
(provide connection-server)
#|
Connection creation protocol
client -> server on client-chan: (list 'connect conn-chan <connect-options>)
server -> client on conn-chan: (or (list 'ok)
(list 'error string))
where <connect-spec> ::= (list 'sqlite3 path/sym mode-sym delay-num limit-num)
| (list 'odbc string string/#f string/#f boolean symbol)
|#
(define (connection-server client-chan)
(let loop ()
(serve client-chan)
(loop)))
(define-lazy-require-definer define-main "../../main.rkt")
(define-main
sqlite3-connect
odbc-connect
odbc-driver-connect)
(define (serve client-chan)
(match (place-channel-get client-chan)
[(list 'connect conn-chan connect-spec)
(with-handlers ([exn:fail?
(lambda (e)
(place-channel-put conn-chan
(list 'error (exn-message e))))])
(let* ([c
(match connect-spec
[(list 'sqlite3 db mode busy-retry-delay busy-retry-limit)
(sqlite3-connect #:database db
#:mode mode
#:busy-retry-delay busy-retry-delay
#:busy-retry-limit busy-retry-limit
#:use-place #f)]
[(list 'odbc dsn user password strict-param? char-mode)
(odbc-connect #:dsn dsn
#:user user
#:password password
#:strict-parameter-types? strict-param?
#:character-mode char-mode
#:use-place #f)]
[(list 'odbc-driver connection-string strict-param? char-mode)
(odbc-driver-connect connection-string
#:strict-parameter-types? strict-param?
#:character-mode char-mode
#:use-place #f)])]
[p (new proxy-server% (connection c) (channel conn-chan))])
(place-channel-put conn-chan (list 'ok))
(thread (lambda () (send p serve)))))]))
#|
Connection methods protocol
client -> server: (list '<method-name> arg ...)
server -> client: (or (list 'values result ...)
(list 'error string))
|#
(define proxy-server%
(class object%
(init-field connection
channel)
(super-new)
(define pstmt-table (make-hash)) ;; int => prepared-statement
(define pstmt-counter 0)
(define/public (serve)
(serve1)
(when connection (serve)))
(define/private (serve1)
(with-handlers ([exn?
(lambda (e)
(place-channel-put channel (list 'error (exn-message e))))])
(call-with-values
(lambda ()
(match (place-channel-get channel)
[(list 'disconnect)
(send connection disconnect)
(set! connection #f)]
[(list 'free-statement pstmt-index)
(send connection free-statement (hash-ref pstmt-table pstmt-index))
(hash-remove! pstmt-table pstmt-index)]
[msg
(define-syntax-rule (forward-methods (method (arg translate) ...) ...)
(match msg
[(list 'method arg ...)
(send connection method (translate arg) ...)]
...))
(define-syntax-rule (id x) x)
(forward-methods (connected?)
(query (w id) (s translate-in-stmt))
(prepare (w id) (s id) (m id))
(list-tables (w id) (s id))
(start-transaction (w id) (m id))
(end-transaction (w id) (m id))
(transaction-status (w id)))]))
(lambda results
(let ([results (for/list ([result (in-list results)]) (translate-result result))])
(place-channel-put channel (cons 'values results)))))))
(define/private (translate-in-stmt x)
(match x
[(list 'string s)
s]
[(list 'statement-binding pstmt-index args)
(statement-binding (hash-ref pstmt-table pstmt-index)
null
(map sexpr->sql-datum args))]))
(define/private (translate-result x)
(match x
[(simple-result y)
(list 'simple-result y)]
[(rows-result h rows)
(for ([row (in-list rows)])
(for ([i (in-range (vector-length row))])
(let* ([x (vector-ref row i)]
[nx (sql-datum->sexpr x)])
(unless (eq? x nx) (vector-set! row i nx)))))
(list 'rows-result h rows)]
;; FIXME: Assumes prepared-statement is concrete class, not interface.
[(? (lambda (x) (is-a? x prepared-statement%)))
(let ([pstmt-index (begin (set! pstmt-counter (add1 pstmt-counter)) pstmt-counter)])
(hash-set! pstmt-table pstmt-index x)
(list 'prepared-statement
pstmt-index
(get-field close-on-exec? x)
(get-field param-typeids x)
(get-field result-dvecs x)))]
[_ x]))))

2
collects/db/private/generic/prepared.rkt
View File

@ -8,7 +8,7 @@
;; prepared-statement%
(define prepared-statement%
(class* object% (prepared-statement<%>)
(init-private handle ;; handle, determined by database system, #f means closed
(init-field handle ;; handle, determined by database system, #f means closed
close-on-exec? ;; boolean
param-typeids ;; (listof typeid)
result-dvecs) ;; (listof vector), layout depends on dbsys

16
collects/db/private/odbc/connection.rkt
View File

@ -661,3 +661,19 @@
(define (field-dvec->typeid dvec)
(vector-ref dvec 1))
#|
Historical note: I tried using ODBC async execution to avoid blocking
all Racket threads for a long time.
1) The postgresql, mysql, and oracle drivers don't even support async
execution. Only DB2 (and probably SQL Server, but I didn't try it).
2) Tests using the DB2 driver gave bafflind HY010 (function sequence
error). My best theory so far is that DB2 (or maybe unixodbc) requires
poll call arguments to be identical to original call arguments, which
means that I would have to replace all uses of (_ptr o X) with
something stable across invocations.
All in all, not worth it, especially given #:use-place solution.
|#

28
collects/db/private/odbc/main.rkt
View File

@ -2,21 +2,26 @@
(require racket/class
racket/contract
"../generic/interfaces.rkt"
"../generic/place-client.rkt"
"connection.rkt"
"dbsystem.rkt"
"ffi.rkt")
(provide odbc-connect
odbc-driver-connect
odbc-data-sources
odbc-drivers
(rename-out [dbsystem odbc-dbsystem]))
odbc-drivers)
(define (odbc-connect #:dsn dsn
#:user [user #f]
#:password [auth #f]
#:notice-handler [notice-handler void]
#:strict-parameter-types? [strict-parameter-types? #f]
#:character-mode [char-mode 'wchar])
#:character-mode [char-mode 'wchar]
#:use-place [use-place #f])
(cond [use-place
(place-connect (list 'odbc dsn user auth strict-parameter-types? char-mode)
odbc-proxy%)]
[else
(let ([notice-handler (make-handler notice-handler "notice")])
(call-with-env 'odbc-connect
(lambda (env)
@ -29,12 +34,17 @@
(db db)
(notice-handler notice-handler)
(strict-parameter-types? strict-parameter-types?)
(char-mode char-mode)))))))))
(char-mode char-mode))))))))]))
(define (odbc-driver-connect connection-string
#:notice-handler [notice-handler void]
#:strict-parameter-types? [strict-parameter-types? #f]
#:character-mode [char-mode 'wchar])
#:character-mode [char-mode 'wchar]
#:use-place [use-place #f])
(cond [use-place
(place-connect (list 'odbc-driver connection-string strict-parameter-types? char-mode)
odbc-proxy%)]
[else
(let ([notice-handler (make-handler notice-handler "notice")])
(call-with-env 'odbc-driver-connect
(lambda (env)
@ -47,7 +57,7 @@
(db db)
(notice-handler notice-handler)
(strict-parameter-types? strict-parameter-types?)
(char-mode char-mode)))))))))
(char-mode char-mode))))))))]))
(define (odbc-data-sources)
(define server-buf (make-bytes 1024))
@ -97,6 +107,12 @@
(let ([=-pos (caar m)])
(cons (substring s 0 =-pos) (substring s (+ 1 =-pos))))))))
(define odbc-proxy%
(class place-proxy-connection%
(super-new)
(define/override (get-dbsystem) dbsystem)))
;; ----
;; Aux functions to free handles on error.

42
collects/db/private/sqlite3/main.rkt
View File

@ -1,34 +1,39 @@
#lang racket/base
(require racket/class
racket/contract
ffi/file
"../generic/place-client.rkt"
"connection.rkt"
"dbsystem.rkt"
"ffi.rkt")
(provide sqlite3-connect
(rename-out [dbsystem sqlite3-dbsystem]))
(provide sqlite3-connect)
(define (sqlite3-connect #:database path-or-sym
(define (sqlite3-connect #:database path
#:mode [mode 'read/write]
#:busy-retry-delay [busy-retry-delay 0.1]
#:busy-retry-limit [busy-retry-limit 10])
#:busy-retry-limit [busy-retry-limit 10]
#:use-place [use-place #f])
(let ([path
(cond [(symbol? path-or-sym)
(case path-or-sym
;; Private, temporary in-memory
[(memory) #":memory:"]
;; Private, temporary on-disk
[(temporary) #""])]
[(or (path? path-or-sym) (string? path-or-sym))
(let ([path (cleanse-path (path->complete-path path-or-sym))])
(case path
((memory temporary) path)
(else
(let ([path (cleanse-path (path->complete-path path))])
(security-guard-check-file 'sqlite3-connect
path
(case mode
((read-only) '(read))
(else '(read write))))
(path->bytes path))])])
path)))])
(cond [use-place
(place-connect (list 'sqlite3 path mode busy-retry-delay busy-retry-limit)
sqlite-place-proxy%)]
[else
(let ([path-bytes
(case path
((memory) #":memory:")
((temporary) #"")
(else (path->bytes path)))])
(let-values ([(db open-status)
(sqlite3_open_v2 path
(sqlite3_open_v2 path-bytes
(case mode
((read-only) SQLITE_OPEN_READONLY)
((read/write) SQLITE_OPEN_READWRITE)
@ -38,4 +43,9 @@
(new connection%
(db db)
(busy-retry-limit busy-retry-limit)
(busy-retry-delay busy-retry-delay)))))
(busy-retry-delay busy-retry-delay))))])))
(define sqlite-place-proxy%
(class place-proxy-connection%
(super-new)
(define/override (get-dbsystem) dbsystem)))

3
collects/db/scribblings/config.rkt
View File

@ -8,6 +8,9 @@
(for-label (all-from-out racket/base)
(all-from-out racket/contract)))
(define (tech/reference . pre-flows)
(apply tech #:doc '(lib "scribblings/reference/reference.scrbl") pre-flows))
;; ----
(define the-eval (make-base-eval))

36
collects/db/scribblings/connect.scrbl
View File

@ -16,6 +16,22 @@ administrative functions for managing connections.
@declare-exporting[db]
There are four kinds of base connection, and they are divided into two
groups: @deftech{wire-based connections} and @deftech{FFI-based
connections}. PostgreSQL and MySQL connections are wire-based, and
SQLite and ODBC connections are FFI-based.
Wire-based connections communicate using @tech/reference{ports}, which
do not cause other Racket threads to block. In contrast, all Racket
threads are blocked during an FFI call, so FFI-based connections can
seriously degrade the interactivity of a Racket program, particularly
if long-running queries are performed using the connection. This
problem can be avoided by creating the FFI-based connection in a
separate @tech/reference{place} using the @racket[#:use-place]
keyword argument. Such a connection will not block all Racket threads
during queries; the disadvantage is the cost of creating and
communicating with a separate @tech/reference{place}.
Base connections are made using the following functions.
@defproc[(postgresql-connect [#:user user string?]
@ -188,7 +204,8 @@ Base connections are made using the following functions.
[#:busy-retry-limit busy-retry-limit
(or/c exact-nonnegative-integer? +inf.0) 10]
[#:busy-retry-delay busy-retry-delay
(and/c rational? (not/c negative?)) 0.1])
(and/c rational? (not/c negative?)) 0.1]
[#:use-place use-place boolean? #f])
connection?]{
Opens the SQLite database at the file named by @racket[database], if
@ -214,6 +231,10 @@ Base connections are made using the following functions.
attempted once. If after @racket[busy-retry-limit] retries the
operation still does not succeed, an exception is raised.
If @racket[use-place] is true, the actual connection is created in
a distinct @tech/reference{place} for database connections and a
proxy is returned.
If the connection cannot be made, an exception is raised.
@(examples/results
@ -234,7 +255,8 @@ Base connections are made using the following functions.
[#:strict-parameter-types? strict-parameter-types? boolean? #f]
[#:character-mode character-mode
(or/c 'wchar 'utf-8 'latin-1)
'wchar])
'wchar]
[#:use-place use-place boolean? #f])
connection?]{
Creates a connection to the ODBC Data Source named @racket[dsn]. The
@ -258,6 +280,10 @@ Base connections are made using the following functions.
See @secref["odbc-status"] for notes on specific ODBC drivers and
recommendations for connection options.
If @racket[use-place] is true, the actual connection is created in
a distinct @tech/reference{place} for database connections and a
proxy is returned.
If the connection cannot be made, an exception is raised.
}
@ -269,7 +295,8 @@ Base connections are made using the following functions.
[#:strict-parameter-types? strict-parameter-types? boolean? #f]
[#:character-mode character-mode
(or/c 'wchar 'utf-8 'latin-1)
'wchar])
'wchar]
[#:use-place use-place boolean? #f])
connection?]{
Creates a connection using an ODBC connection string containing a
@ -606,7 +633,8 @@ ODBC's DSNs.
[#:busy-retry-limit busy-retry-limit
(or/c exact-nonnegative-integer? +inf.0) @#,absent]
[#:busy-retry-delay busy-retry-delay
(and/c rational? (not/c negative?)) @#,absent])
(and/c rational? (not/c negative?)) @#,absent]
[#:use-place use-place boolean? @#,absent])
data-source?]
@defproc[(odbc-data-source
[#:dsn dsn (or/c string? #f) @#,absent]

3
collects/db/sqlite3.rkt
View File

@ -9,5 +9,6 @@
(->* (#:database (or/c path-string? 'memory 'temporary))
(#:mode (or/c 'read-only 'read/write 'create)
#:busy-retry-limit (or/c exact-nonnegative-integer? +inf.0)
#:busy-retry-delay (and/c rational? (not/c negative?)))
#:busy-retry-delay (and/c rational? (not/c negative?))
#:use-place any/c)
any/c)])

18
collects/tests/db/all-tests.rkt
View File

@ -130,7 +130,16 @@ Testing profiles are flattened, not hierarchical.
(define sqlite-unit
(dbconf->unit
(dbconf "sqlite3, memory"
(data-source 'sqlite3 '(#:database memory) '((db:test (issl)))))))
(data-source 'sqlite3
'(#:database memory)
'((db:test (issl)))))))
(define sqlite/p-unit
(dbconf->unit
(dbconf "sqlite3, memory, with #:use-place=#t"
(data-source 'sqlite3
'(#:database memory #:use-place #t)
'((db:test (issl async)))))))
;; ----
@ -176,6 +185,9 @@ Testing profiles are flattened, not hierarchical.
(define sqlite-test
(specialize-test sqlite-unit))
(define sqlite/p-test
(specialize-test sqlite/p-unit))
(define generic-test
(make-test-suite "Generic tests (no db)"
(list gen-sql-types:test
@ -217,7 +229,9 @@ Testing profiles are flattened, not hierarchical.
(make-all-tests label (get-dbconf (string->symbol label)))))]
[tests
(cond [(or include-sqlite? (null? labels))
(cons (cons "sqlite3, memory" sqlite-test) tests)]
(list* (cons "sqlite3, memory" sqlite-test)
(cons "sqlite3, memory, #:use-place=#t" sqlite/p-test)
tests)]
[else tests])]
[tests
(cond [(or include-generic? (null? labels))

72
collects/tests/db/db/concurrent.rkt
View File

@ -62,50 +62,48 @@
(for ([t (in-hash-keys threads)])
(sync t))))))))
;; ----
(define pool-test
(test-suite "connection pools"
(test-case "lease, limit, release"
(let* ([counter 0]
[p (connection-pool (lambda () (set! counter (+ 1 counter)) (connect-for-test))
#:max-connections 2)]
[c1 (connection-pool-lease p)]
[c2 (connection-pool-lease p)])
;; Two created
(check-equal? counter 2)
;; Can't create new one yet
(check-exn exn:fail? (lambda () (connection-pool-lease p)))
;; But if we free one...
(disconnect c2)
(check-equal? (connected? c2) #f)
(let ([c3 (connection-pool-lease p)])
(check-equal? counter 2 "not new") ;; used idle, not new connection
(check-equal? (connected? c3) #t))))
(test-case "release on evt"
(let* ([p (connection-pool connect-for-test #:max-connections 2)]
(define (async-test)
(test-case "asynchronous execution"
(unless (ANYFLAGS 'ismy 'isora 'isdb2)
(call-with-connection
(lambda (c)
(query-exec c "create temporary table nums (n integer)")
(for ([i (in-range 40)])
(query-exec c (sql "insert into nums (n) values ($1)") i))
(let* ([the-sql "select cast(max(a.n * b.n *c.n * d.n) as varchar) \
from nums a, nums b, nums c, nums d"]
[pst (prepare c the-sql)]
[sema (make-semaphore 0)]
[c1 (connection-pool-lease p sema)])
(check-equal? (connected? c1) #t)
;; Closes when evt ready
(begin (semaphore-post sema) (sleep 0.1))
(check-equal? (connected? c1) #f)))
(test-case "release on custodian"
(let* ([p (connection-pool connect-for-test #:max-connections 2)]
[cust (make-custodian)]
[c1 (connection-pool-lease p cust)])
(check-equal? (connected? c1) #t)
;; Closes when custodian shutdown
(begin (custodian-shutdown-all cust) (sleep 0.1))
(check-equal? (connected? c1) #f)))))
[peek (semaphore-peek-evt sema)]
[counter 0]
[thd
(thread (lambda ()
(let loop ()
(sync peek)
(set! counter (add1 counter))
(sleep 0.01)
(loop))))])
(let ([start (current-inexact-milliseconds)])
(semaphore-post sema)
(query-value c pst)
(semaphore-wait sema)
(let ([end (current-inexact-milliseconds)])
(when (ANYFLAGS 'postgresql 'mysql 'async)
(when #f
(printf "counter = ~s\n" counter)
(printf "time elapsed = ~s\n" (- end start)))
;; If c does not execute asynchronously, expect counter to be about 0.
(check-pred positive? counter)
(let ([expected-counter (/ (- end start) (* 0.01 1000))])
(check > counter (* 0.5 expected-counter))))))))))))
;; ----
(define test
(test-suite "Concurrency"
(async-test)
;; Tests whether connections are properly locked.
(test-concurrency 1)
(test-concurrency 2)
(test-concurrency 20)
(kill-safe-test #t)
pool-test))
(kill-safe-test #t)))

36
collects/tests/db/db/query.rkt
View File

@ -300,6 +300,41 @@
(check-prep-once
(lambda () (virtual-connection (connection-pool connect-and-setup))))))))
(define pool-tests
(test-suite "connection pools"
(test-case "lease, limit, release"
(let* ([counter 0]
[p (connection-pool (lambda () (set! counter (+ 1 counter)) (connect-for-test))
#:max-connections 2)]
[c1 (connection-pool-lease p)]
[c2 (connection-pool-lease p)])
;; Two created
(check-equal? counter 2)
;; Can't create new one yet
(check-exn exn:fail? (lambda () (connection-pool-lease p)))
;; But if we free one...
(disconnect c2)
(check-equal? (connected? c2) #f)
(let ([c3 (connection-pool-lease p)])
(check-equal? counter 2 "not new") ;; used idle, not new connection
(check-equal? (connected? c3) #t))))
(test-case "release on evt"
(let* ([p (connection-pool connect-for-test #:max-connections 2)]
[sema (make-semaphore 0)]
[c1 (connection-pool-lease p sema)])
(check-equal? (connected? c1) #t)
;; Closes when evt ready
(begin (semaphore-post sema) (sleep 0.1))
(check-equal? (connected? c1) #f)))
(test-case "release on custodian"
(let* ([p (connection-pool connect-for-test #:max-connections 2)]
[cust (make-custodian)]
[c1 (connection-pool-lease p cust)])
(check-equal? (connected? c1) #t)
;; Closes when custodian shutdown
(begin (custodian-shutdown-all cust) (sleep 0.1))
(check-equal? (connected? c1) #f)))))
(define test
(test-suite "query API"
(simple-tests 'string)
@ -309,4 +344,5 @@
low-level-tests
misc-tests
virtual-statement-tests
pool-tests
error-tests))