1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
|
(define (db:process-cached-writes db)
(let ((queries (make-hash-table))
(data #f))
(mutex-lock! *incoming-mutex*)
;; data is a list of query packets <vector qry-sig query params
(set! data (reverse *incoming-writes*)) ;; (sort ... (lambda (a b)(< (vector-ref a 1)(vector-ref b 1)))))
(set! *incoming-writes* '())
(mutex-unlock! *incoming-mutex*)
(if (> (length data) 0)
;; Process if we have data
(begin
(debug:print-info 7 "Writing cached data " data)
|
>
|
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
|
(define (db:process-cached-writes db)
(let ((queries (make-hash-table))
(data #f))
(mutex-lock! *incoming-mutex*)
;; data is a list of query packets <vector qry-sig query params
(set! data (reverse *incoming-writes*)) ;; (sort ... (lambda (a b)(< (vector-ref a 1)(vector-ref b 1)))))
(set! *server:last-write-flush* (current-milliseconds))
(set! *incoming-writes* '())
(mutex-unlock! *incoming-mutex*)
(if (> (length data) 0)
;; Process if we have data
(begin
(debug:print-info 7 "Writing cached data " data)
|
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
|
(let ((cache-size (length data)))
(if (> cache-size *max-cache-size*)
(set! *max-cache-size* cache-size)))
#t)
#f)))
(define *db:process-queue-mutex* (make-mutex))
;; The queue is a list of vectors where the zeroth slot indicates the type of query to
;; apply and the second slot is the time of the query and the third entry is a list of
;; values to be applied
;;
(define (db:queue-write-and-wait db qry-sig query params)
(let ((queue-len 0)
(res #f)
(got-it #f)
(qry-pkt (vector qry-sig query params))
(timeout (+ 10 (current-seconds)))) ;; set the time out to 10 secs in future
;; Put the item in the queue *incoming-writes*
(mutex-lock! *incoming-mutex*)
(set! *incoming-writes* (cons qry-pkt *incoming-writes*))
(set! queue-len (length *incoming-writes*))
(mutex-unlock! *incoming-mutex*)
|
>
>
>
>
|
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
|
(let ((cache-size (length data)))
(if (> cache-size *max-cache-size*)
(set! *max-cache-size* cache-size)))
#t)
#f)))
(define *db:process-queue-mutex* (make-mutex))
(define *number-of-writes* 0)
(define *writes-total-delay* 0)
;; The queue is a list of vectors where the zeroth slot indicates the type of query to
;; apply and the second slot is the time of the query and the third entry is a list of
;; values to be applied
;;
(define (db:queue-write-and-wait db qry-sig query params)
(let ((queue-len 0)
(res #f)
(got-it #f)
(qry-pkt (vector qry-sig query params))
(start-time (current-milliseconds))
(timeout (+ 10 (current-seconds)))) ;; set the time out to 10 secs in future
;; Put the item in the queue *incoming-writes*
(mutex-lock! *incoming-mutex*)
(set! *incoming-writes* (cons qry-pkt *incoming-writes*))
(set! queue-len (length *incoming-writes*))
(mutex-unlock! *incoming-mutex*)
|
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
|
(begin
(hash-table-delete! *completed-writes* qry-sig)
(set! got-it #t)))
(mutex-unlock! *completed-mutex*)
(if (and (not got-it)
(< (current-seconds) timeout))
(loop)))
got-it))
(define (db:process-queue-item db item)
(let* ((stmt-key (cdb:packet-get-qtype item))
(qry-sig (cdb:packet-get-query-sig item))
(return-address (cdb:packet-get-client-sig item))
(params (cdb:packet-get-params item))
|
>
>
|
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
|
(begin
(hash-table-delete! *completed-writes* qry-sig)
(set! got-it #t)))
(mutex-unlock! *completed-mutex*)
(if (and (not got-it)
(< (current-seconds) timeout))
(loop)))
(set! *number-of-writes* (+ *number-of-writes* 1))
(set! *writes-total-delay* (+ *writes-total-delay* 1))
got-it))
(define (db:process-queue-item db item)
(let* ((stmt-key (cdb:packet-get-qtype item))
(qry-sig (cdb:packet-get-query-sig item))
(return-address (cdb:packet-get-client-sig item))
(params (cdb:packet-get-params item))
|