64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
"SELECT test_id FROM queue WHERE start_time > ?;" mystart)
res))
(define (lock-queue:get-lock db test-id)
(let ((res #f)
(lckqry (sqlite3:prepare db "SELECT test_id,run_lock FROM runlocks WHERE run_lock='locked';"))
(mklckqry (sqlite3:prepare db "INSERT INTO runlocks (test_id,run_lock) VALUES (?,'locked');")))
(handle-exceptions
exn
#f
(sqlite3:with-transaction
db
(lambda ()
(sqlite3:for-each-row (lambda (tid lockstate)
(set! res (list tid lockstate)))
lckqry)
(if res
(if (equal? (car res) test-id)
#t ;; already have the lock
#f)
(begin
(sqlite3:execute mklckqry test-id)
;; if no error handled then return #t for got the lock
#t)))))))
(let loop ((younger-waiting (lock-queue:any-younger? db mystart test-id)))
(if younger-waiting
(begin
;; no need for us to wait. mark in the lock queue db as skipping
(lock-queue:set-state db test-id "skipping")
(begin
;; (sqlite3:finalize! db)
#f)) ;; let the calling process know that nothing needs to be done
(if (lock-queue:get-lock db test-id)
(begin
;; (sqlite3:finalize! db)
#t)
(if (> (- (current-seconds) mystart) 36000) ;; waited too long, steal the lock
(lock-queue:steal-lock db test-id)
(begin
(thread-sleep! 1)
(loop (lock-queue:any-younger? db mystart test-id)))))))))
|
>
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
>
>
>
>
>
>
>
>
|
|
|
|
|
<
<
|
|
<
<
|
|
|
|
|
|
>
>
|
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
"SELECT test_id FROM queue WHERE start_time > ?;" mystart)
res))
(define (lock-queue:get-lock db test-id)
(let ((res #f)
(lckqry (sqlite3:prepare db "SELECT test_id,run_lock FROM runlocks WHERE run_lock='locked';"))
(mklckqry (sqlite3:prepare db "INSERT INTO runlocks (test_id,run_lock) VALUES (?,'locked');")))
(let ((result
(handle-exceptions
exn
#f
(sqlite3:with-transaction
db
(lambda ()
(sqlite3:for-each-row (lambda (tid lockstate)
(set! res (list tid lockstate)))
lckqry)
(if res
(if (equal? (car res) test-id)
#t ;; already have the lock
#f)
(begin
(sqlite3:execute mklckqry test-id)
;; if no error handled then return #t for got the lock
#t)))))))
(sqlite3:finalize! lckqry)
(sqlite3:finalize! mklckqry)
result)))
(define (lock-queue:release-lock fname test-id)
(let ((db (lock-queue:open-db fname)))
(sqlite3:execute db "DELETE FROM runlocks WHERE test_id=?;" test-id)
(sqlite3:finalize! db)))
(define (lock-queue:steal-lock db test-id)
(sqlite3:execute db "DELETE FROM runlocks WHERE run_lock='locked';")
(lock-queue:get-lock db test-it))
;; returns #f if ok to skip the task
;; returns #t if ok to proceed with task
;; otherwise waits
;;
(define (lock-queue:wait-turn fname test-id)
(let ((db (lock-queue:open-db fname))
(mystart (current-seconds)))
(sqlite3:execute
db
"INSERT OR REPLACE INTO queue (test_id,start_time,state) VALUES (?,?,'waiting');"
test-id mystart)
(thread-sleep! 1) ;; give other tests a chance to register
(let ((result
(let loop ((younger-waiting (lock-queue:any-younger? db mystart test-id)))
(if younger-waiting
(begin
;; no need for us to wait. mark in the lock queue db as skipping
(lock-queue:set-state db test-id "skipping")
#f) ;; let the calling process know that nothing needs to be done
(if (lock-queue:get-lock db test-id)
#t
(if (> (- (current-seconds) mystart) 36000) ;; waited too long, steal the lock
(lock-queue:steal-lock db test-id)
(begin
(thread-sleep! 1)
(loop (lock-queue:any-younger? db mystart test-id)))))))))
(sqlite3:finalize! db)
result)))
;; (use trace)
;; (trace lock-queue:get-lock lock-queue:release-lock lock-queue:wait-turn lock-queue:any-younger? lock-queue:set-state)
|