24
25
26
27
28
29
30
31
32
33
34
35
36
37
|
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
|
+
+
+
+
|
;;======================================================================
(define (make-lock-queue:db-dat)(make-vector 3))
(define-inline (lock-queue:db-dat-get-db vec) (vector-ref vec 0))
(define-inline (lock-queue:db-dat-get-path vec) (vector-ref vec 1))
(define-inline (lock-queue:db-dat-set-db! vec val)(vector-set! vec 0 val))
(define-inline (lock-queue:db-dat-set-path! vec val)(vector-set! vec 1 val))
(define (lock-queue:delete-lock-db dbdat)
(let ((fname (lock-queue:db-dat-get-path dbdat)))
(system (conc "rm -f " fname "*"))))
(define (lock-queue:open-db fname #!key (count 10))
(let* ((actualfname (conc fname ".lockdb"))
(dbexists (file-exists? actualfname))
(db (sqlite3:open-database actualfname))
(handler (make-busy-timeout 136000)))
(if dbexists
|
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
|
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
|
+
-
+
-
+
-
+
+
|
(debug:print 0 "ERROR: Failed to set lock state for test with id " test-id ", error: " ((condition-property-accessor 'exn 'message) exn) ", giving up.")
#f))
(sqlite3:execute (lock-queue:db-dat-get-db dbdat) "UPDATE queue SET state=? WHERE test_id=?;"
newstate
test-id)))
(define (lock-queue:any-younger? dbdat mystart test-id #!key (remtries 10))
;; no need to wait on journal on read only queries
(tasks:wait-on-journal (lock-queue:db-dat-get-path dbdat) 1200)
;; (tasks:wait-on-journal (lock-queue:db-dat-get-path dbdat) 1200)
(handle-exceptions
exn
(if (> remtries 0)
(begin
(debug:print 0 "WARNING: exception on lock-queue:any-younger. Trying again in 30 seconds.")
(debug:print 0 "WARNING: exception on lock-queue:any-younger. Removing lockdb and trying again in 5 seconds.")
(debug:print 0 " message: " ((condition-property-accessor 'exn 'message) exn))
(thread-sleep! 30)
(thread-sleep! 5)
(lock-queue:delete-lock-db dbdat)
(lock-queue:any-younger? dbdat mystart test-id remtries: (- remtries 1)))
(begin
(debug:print 0 "ERROR: Failed to find younger locks for test with id " test-id ", error: " ((condition-property-accessor 'exn 'message) exn) ", giving up.")
#f))
(let ((res #f))
(sqlite3:for-each-row
(lambda (tid)
|
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
-
+
-
-
+
+
+
+
|
(db (lock-queue:db-dat-get-db dbdat))
(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
(begin
(debug:print 0 "WARNING: failed to get queue lock. Will try again in a few seconds")
(debug:print 0 "WARNING: failed to get queue lock. Removing lock db and returning fail") ;; Will try again in a few seconds")
(debug:print 0 " message: " ((condition-property-accessor 'exn 'message) exn))
(thread-sleep! 10)
(if (> count 0)
(lock-queue:get-lock dbdat test-id count: (- count 1)))
;; (if (> count 0)
;; #f ;; (lock-queue:get-lock dbdat test-id count: (- count 1)) - give up on retries
;; (begin ;; never recovered, remote the lock file and return #f, no lock obtained
(lock-queue:delete-lock-db dbdat)
#f)
(sqlite3:with-transaction
db
(lambda ()
(sqlite3:for-each-row (lambda (tid lockstate)
(set! res (list tid lockstate)))
lckqry)
|
137
138
139
140
141
142
143
144
145
146
147
148
149
150
|
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
|
+
|
#t)))))))
(sqlite3:finalize! lckqry)
(sqlite3:finalize! mklckqry)
result)))
(define (lock-queue:release-lock fname test-id #!key (count 10))
(let* ((dbdat (lock-queue:open-db fname)))
(tasks:wait-on-journal (lock-queue:db-dat-get-path dbdat) 1200 "lock-queue:release-lock; waiting on journal")
(handle-exceptions
exn
(begin
(debug:print 0 "WARNING: Failed to release queue lock. Will try again in few seconds")
(debug:print 0 " message: " ((condition-property-accessor 'exn 'message) exn))
(thread-sleep! (/ count 10))
(if (> count 0)
|
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
|
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
|
-
+
+
+
+
+
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
|
(define (lock-queue:steal-lock dbdat test-id #!key (count 10))
(debug:print-info 0 "Attempting to steal lock at " (lock-queue:db-dat-get-path dbdat))
(tasks:wait-on-journal (lock-queue:db-dat-get-path dbdat) 1200 "lock-queue:steal-lock; waiting on journal")
(handle-exceptions
exn
(begin
(debug:print 0 "WARNING: Failed to steal queue lock. Will try again in few seconds")
(tadebug:print 0 "WARNING: Failed to steal queue lock. Will try again in few seconds")
(debug:print 0 " message: " ((condition-property-accessor 'exn 'message) exn))
(thread-sleep! 10)
(if (> count 0)
(lock-queue:steal-lock dbdat test-id count: (- count 1))
#f))
(sqlite3:execute (lock-queue:db-dat-get-db dbdat) "DELETE FROM runlocks WHERE run_lock='locked';"))
(lock-queue:get-lock dbdat 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 #!key (count 10)(waiting-msg #f))
(let* ((dbdat (lock-queue:open-db fname))
(mystart (current-seconds))
(db (lock-queue:db-dat-get-db dbdat)))
;; (tasks:wait-on-journal (lock-queue:db-dat-get-path dbdat) 1200 waiting-msg: "lock-queue:wait-turn; waiting on journal file")
(handle-exceptions
exn
(begin
(debug:print 0 "WARNING: Failed to find out if it is ok to skip the wait queue. Will try again in few seconds")
(debug:print 0 " message: " ((condition-property-accessor 'exn 'message) exn))
(print-call-chain (current-error-port))
(thread-sleep! 10)
(if (> count 0)
(begin
(sqlite3:finalize! db)
(lock-queue:wait-turn fname test-id count: (- count 1)))
(begin
(debug:print 0 "Giving up calls to lock-queue:wait-turn for test-id " test-id " at path " fname ", printing call chain")
(print-call-chain (current-error-port))
#f)))
;; wait 10 seconds and then check to see if someone is already updating the html
(thread-sleep! 10)
(if (not (lock-queue:any-younger? dbdat mystart test-id)) ;; no processing in flight, must try to start processing
(begin
(tasks:wait-on-journal (lock-queue:db-dat-get-path dbdat) 1200 waiting-msg: "lock-queue:wait-turn; waiting on journal file")
(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? dbdat 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 dbdat test-id "skipping")
#f) ;; let the calling process know that nothing needs to be done
(if (lock-queue:get-lock dbdat test-id)
#t
(if (> (- (current-seconds) mystart) 36000) ;; waited too long, steal the lock
(lock-queue:steal-lock dbdat test-id)
(begin
(thread-sleep! 1)
(loop (lock-queue:any-younger? dbdat mystart test-id)))))))))
(sqlite3:finalize! db)
result))))
|