1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
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
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
|
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
+
+
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
+
-
+
+
+
-
-
+
+
+
+
-
-
+
+
+
-
+
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
-
-
-
+
+
+
+
+
+
-
+
-
-
-
-
+
+
+
+
+
+
+
+
-
-
+
+
+
+
-
+
+
+
+
-
+
+
+
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
-
+
+
+
+
+
-
+
-
-
+
+
-
-
-
+
+
+
+
+
+
+
+
+
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
|
;; Copyright 2006-2013, Matthew Welland.
;;
;; This program is made available under the GNU GPL version 2.0 or
;; greater. See the accompanying file COPYING for details.
;;
;; This program is distributed WITHOUT ANY WARRANTY; without even the
;; implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
;; PURPOSE.
;;======================================================================
;; launch a task - this runs on the originating host, tests themselves
;;
;;======================================================================
db
(vector db actualfname)
(begin
(handle-exceptions
exn
(begin
(thread-sleep! 10)
(if (> count 0)
(lock-queue:open-db fname count: (- count 1))
(vector db actualfname)))
(sqlite3:with-transaction
db))
(sqlite3:execute
db
"CREATE TABLE IF NOT EXISTS queue (
id INTEGER PRIMARY KEY,
test_id INTEGER,
start_time INTEGER,
state TEXT,
CONSTRAINT queue_constraint UNIQUE (test_id));")
(sqlite3:execute
db
"CREATE TABLE IF NOT EXISTS runlocks (
id INTEGER PRIMARY KEY,
test_id INTEGER,
run_lock TEXT,
CONSTRAINT runlock_constraint UNIQUE (run_lock));"))))
db
(lambda ()
(sqlite3:execute
db
"CREATE TABLE IF NOT EXISTS queue (
id INTEGER PRIMARY KEY,
test_id INTEGER,
start_time INTEGER,
state TEXT,
CONSTRAINT queue_constraint UNIQUE (test_id));")
(sqlite3:execute
db
"CREATE TABLE IF NOT EXISTS runlocks (
id INTEGER PRIMARY KEY,
test_id INTEGER,
run_lock TEXT,
CONSTRAINT runlock_constraint UNIQUE (run_lock));"))))))
(sqlite3:set-busy-handler! db handler)
db))
(vector db actualfname)))
(define (lock-queue:set-state db test-id newstate #!key (count 10))
(define (lock-queue:set-state dbdat test-id newstate #!key (remtries 10))
(tasks:wait-on-journal (lock-queue:db-dat-get-path dbdat) 1200)
(handle-exceptions
exn
(if (> remtries 0)
(begin
(thread-sleep! 10)
(begin
(debug:print 0 "WARNING: exception on lock-queue:set-state. Trying again in 30 seconds.")
(debug:print 0 " message: " ((condition-property-accessor 'exn 'message) exn))
(thread-sleep! 30)
(if (> count 0)
(lock-queue:set-state db test-id newstate (- count 1))
(sqlite3:execute db "UPDATE queue SET state=? WHERE test_id=?;"
(sqlite3:execute (lock-queue:db-dat-get-db dbdat) "UPDATE queue SET state=? WHERE test_id=?;"
newstate
test-id)))
(define (lock-queue:any-younger? db mystart test-id #!key (count 10))
(let ((res #f))
(handle-exceptions
exn
(begin
(thread-sleep! 10)
(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)
(handle-exceptions
exn
(if (> remtries 0)
(begin
(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! 5)
(if (> count 0)
(lock-queue:any-younger? db mystart test-id count: (- count 1))
#f))
db
(lock-queue:db-dat-get-db dbdat)
"SELECT test_id FROM queue WHERE start_time > ?;" mystart)
res)))
(define (lock-queue:get-lock db test-id #!key (count 10))
(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');")))
(define (lock-queue:get-lock dbdat test-id #!key (count 10)(waiting-msg #f))
(tasks:wait-on-journal (lock-queue:db-dat-get-path dbdat) 1200 remove: #t waiting-msg: "lock-queue:get-lock, waiting on journal")
(let* ((res #f)
(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. 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 db 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)
(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 #!key (count 10))
(let ((db (lock-queue:open-db fname)))
(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! 10)
(thread-sleep! (/ count 10))
(if (> count 0)
(begin
(sqlite3:finalize! (lock-queue:db-dat-get-db dbdat))
(lock-queue:release-lock fname test-id count: (- count 1))
#f))
(sqlite3:execute db "DELETE FROM runlocks WHERE test_id=?;" test-id)
(sqlite3:finalize! db))))
(lock-queue:release-lock fname test-id count: (- count 1)))
(let ((journal (conc fname "-journal")))
;; If we've tried ten times and failed there is a serious problem
;; try to remove the lock db and allow it to be recreated
(handle-exceptions
exn
#f
(if (file-exists? journal)(delete-file journal))
(if (file-exists? fname) (delete-file fname))
#f))))
(sqlite3:execute (lock-queue:db-dat-get-db dbdat) "DELETE FROM runlocks WHERE test_id=?;" test-id)
(sqlite3:finalize! (lock-queue:db-dat-get-db dbdat)))))
(define (lock-queue:steal-lock db test-id #!key (count 10))
(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
(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 db test-id count: (- count 1))
(lock-queue:steal-lock dbdat test-id count: (- count 1))
#f))
(sqlite3:execute db "DELETE FROM runlocks WHERE run_lock='locked';"))
(lock-queue:get-lock db test-it))
(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))
(let ((db (lock-queue:open-db fname))
(mystart (current-seconds)))
(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))
#f))
(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))))
(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
;; no point in marking anything in the queue - simply never register this
;; test as it is *covered* by a previously started update to the html file
;; (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))))))
;; (use trace)
;; (trace lock-queue:get-lock lock-queue:release-lock lock-queue:wait-turn lock-queue:any-younger? lock-queue:set-state)
|