Overview
Comment: | Partial re-implementation (again) of transaction wrapped writes, now working except for transaction within transaction issue |
---|---|
Downloads: | Tarball | ZIP archive | SQL archive |
Timelines: | family | ancestors | descendants | both | newdashboard |
Files: | files | file ages | folders |
SHA1: |
7bbc43c98de6974b9702afbfcb3216a3 |
User & Date: | mrwellan on 2013-03-19 11:22:46 |
Other Links: | branch diff | manifest | tags |
Context
2013-03-19
| ||
11:26 | Partial re-implementation (again) of transaction wrapped writes, now working except for transaction within transaction issue check-in: 69c6ef28ac user: mrwellan tags: newdashboard | |
11:22 | Partial re-implementation (again) of transaction wrapped writes, now working except for transaction within transaction issue check-in: 7bbc43c98d user: mrwellan tags: newdashboard | |
10:24 | Partial re-implementation (again) of transaction wrapped writes check-in: 5de9fdab73 user: mrwellan tags: newdashboard | |
Changes
Modified db.scm from [2cf85fa9f1] to [5fd4db1042].
︙ | ︙ | |||
1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 | ;; not used, intended to indicate to run in calling process (define db:run-local-queries '()) ;; rollup-tests-pass-fail)) (define (db:process-cached-writes db) (let ((queries (make-hash-table)) (data #f)) (mutex-lock! *incoming-mutex*) (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) | > > > | | | | | < | < | < | | | | | | | < | | | | > | | > | | | | | | > | | | | | | | | | | | | | | > | | > | | | > > > | 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 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 1378 1379 1380 1381 | ;; not used, intended to indicate to run in calling process (define db:run-local-queries '()) ;; rollup-tests-pass-fail)) (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) ;; Prepare the needed sql statements ;; (for-each (lambda (request-item) (let ((stmt-key (vector-ref request-item 0)) (query (vector-ref request-item 1))) (hash-table-set! queries stmt-key (sqlite3:prepare db query)))) data) ;; No outer loop needed. Single loop for write items only. Reads trigger flush of queue ;; and then are executed. (sqlite3:with-transaction db (lambda () (for-each (lambda (hed) (let* ((params (vector-ref hed 2)) (stmt-key (vector-ref hed 0)) (stmt (hash-table-ref/default queries stmt-key #f))) (if stmt (apply sqlite3:execute stmt params) (debug:print 0 "ERROR: Problem Executing " stmt-key " for " params)))) data))) ;; let all the waiting calls know all is done (mutex-lock! *completed-mutex*) (for-each (lambda (item) (let ((qry-sig (cdb:packet-get-client-sig item))) (debug:print-info 7 "Registering query " qry-sig " as done") (hash-table-set! *completed-writes* qry-sig #t))) data) (mutex-unlock! *completed-mutex*) ;; Finalize the statements. Should this be done inside the mutex above? ;; I think sqlite3 mutexes will keep the data safe (for-each (lambda (stmt-key) (sqlite3:finalize! (hash-table-ref queries stmt-key))) (hash-table-keys queries)) ;; Do a little record keeping (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*) (debug:print-info 7 "Current write queue length is " queue-len) ;; poll for the write to complete, timeout after 10 seconds ;; periodic flushing of the queue is taken care of by ;; db:flush-queue (let loop () (thread-sleep! 0.1) (mutex-lock! *completed-mutex*) |
︙ | ︙ | |||
1388 1389 1390 1391 1392 1393 1394 | (return-address (cdb:packet-get-client-sig item)) (params (cdb:packet-get-params item)) (query (let ((q (alist-ref stmt-key db:queries))) (if q (car q) #f)))) (debug:print-info 11 "Special queries/requests stmt-key=" stmt-key ", return-address=" return-address ", query=" query ", params=" params) (if query ;; hand queries off to the write queue | < | > > | | | > > | | 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 | (return-address (cdb:packet-get-client-sig item)) (params (cdb:packet-get-params item)) (query (let ((q (alist-ref stmt-key db:queries))) (if q (car q) #f)))) (debug:print-info 11 "Special queries/requests stmt-key=" stmt-key ", return-address=" return-address ", query=" query ", params=" params) (if query ;; hand queries off to the write queue (let ((response (case *transport-type* ((http) (debug:print-info 7 "Queuing item " item " for wrapped write") (db:queue-write-and-wait db qry-sig query params)) (else (apply sqlite3:execute db query params) #t)))) (debug:print-info 7 "Received " response " from wrapped write") (server:reply return-address qry-sig response response)) ;; otherwise if appropriate flush the queue (this is a read or complex query) (begin (case *transport-type* ((http)(db:process-cached-writes db))) (cond ((member stmt-key db:special-queries) (debug:print-info 11 "Handling special statement " stmt-key) |
︙ | ︙ |