schedulerbase.py@ 90802

Last change on this file since 90802 was 83341, checked in by vboxsync, 5 years ago
TestManager: Don't generate scheduling entries for disabled scheduling groups. Remove items from SchedQueues belonging to deleted scheduling groups when regenerating all queues.
Property svn:eol-style set to `native` Property svn:keywords set to `Author Date Id Revision`
File size: 69.0 KB

Line
1	# -- coding: utf-8 --
2	# $Id: schedulerbase.py 83341 2020-03-19 20:40:17Z vboxsync $
3	# pylint: disable=too-many-lines
4
5
6	"""
7	Test Manager - Base class and utilities for the schedulers.
8	"""
9
10	__copyright__ = \
11	"""
12	Copyright (C) 2012-2020 Oracle Corporation
13
14	This file is part of VirtualBox Open Source Edition (OSE), as
15	available from http://www.virtualbox.org. This file is free software;
16	you can redistribute it and/or modify it under the terms of the GNU
17	General Public License (GPL) as published by the Free Software
18	Foundation, in version 2 as it comes in the "COPYING" file of the
19	VirtualBox OSE distribution. VirtualBox OSE is distributed in the
20	hope that it will be useful, but WITHOUT ANY WARRANTY of any kind.
21
22	The contents of this file may alternatively be used under the terms
23	of the Common Development and Distribution License Version 1.0
24	(CDDL) only, as it comes in the "COPYING.CDDL" file of the
25	VirtualBox OSE distribution, in which case the provisions of the
26	CDDL are applicable instead of those of the GPL.
27
28	You may elect to license modified versions of this file under the
29	terms and conditions of either the GPL or the CDDL or both.
30	"""
31	__version__ = "$Revision: 83341 $"
32
33
34	# Standard python imports.
35	import sys;
36	import unittest;
37
38	# Validation Kit imports.
39	from common import utils, constants;
40	from testmanager import config;
41	from testmanager.core.build import BuildDataEx, BuildLogic;
42	from testmanager.core.base import ModelDataBase, ModelDataBaseTestCase, TMExceptionBase;
43	from testmanager.core.buildsource import BuildSourceData, BuildSourceLogic;
44	from testmanager.core.globalresource import GlobalResourceLogic;
45	from testmanager.core.schedgroup import SchedGroupData, SchedGroupLogic;
46	from testmanager.core.systemlog import SystemLogData, SystemLogLogic;
47	from testmanager.core.testbox import TestBoxData, TestBoxDataEx;
48	from testmanager.core.testboxstatus import TestBoxStatusData, TestBoxStatusLogic;
49	from testmanager.core.testcase import TestCaseLogic;
50	from testmanager.core.testcaseargs import TestCaseArgsDataEx, TestCaseArgsLogic;
51	from testmanager.core.testset import TestSetData, TestSetLogic;
52
53	# Python 3 hacks:
54	if sys.version_info[0] >= 3:
55	xrange = range; # pylint: disable=redefined-builtin,invalid-name
56
57
58
59	class ReCreateQueueData(object):
60	"""
61	Data object for recreating a scheduling queue.
62
63	It's mostly a storage object, but has a few data checking operation
64	associated with it.
65	"""
66
67	def __init__(self, oDb, idSchedGroup):
68	#
69	# Load data from the database.
70	#
71	oSchedGroupLogic = SchedGroupLogic(oDb);
72	self.oSchedGroup = oSchedGroupLogic.cachedLookup(idSchedGroup);
73
74	# Will extend the entries with aoTestCases and dTestCases members
75	# further down (SchedGroupMemberDataEx). checkForGroupDepCycles
76	# will add aidTestGroupPreReqs.
77	self.aoTestGroups = oSchedGroupLogic.getMembers(idSchedGroup);
78
79	# aoTestCases entries are TestCaseData instance with iSchedPriority
80	# and idTestGroup added for our purposes.
81	# We will add oTestGroup and aoArgsVariations members to each further down.
82	self.aoTestCases = oSchedGroupLogic.getTestCasesForGroup(idSchedGroup, cMax = 4096);
83
84	# Load dependencies.
85	oTestCaseLogic = TestCaseLogic(oDb)
86	for oTestCase in self.aoTestCases:
87	oTestCase.aidPreReqs = oTestCaseLogic.getTestCasePreReqIds(oTestCase.idTestCase, cMax = 4096);
88
89	# aoTestCases entries are TestCaseArgsData instance with iSchedPriority
90	# and idTestGroup added for our purposes.
91	# We will add oTestGroup and oTestCase members to each further down.
92	self.aoArgsVariations = oSchedGroupLogic.getTestCaseArgsForGroup(idSchedGroup, cMax = 65536);
93
94	#
95	# Generate global lookups.
96	#
97
98	# Generate a testcase lookup dictionary for use when working on
99	# argument variations.
100	self.dTestCases = dict();
101	for oTestCase in self.aoTestCases:
102	self.dTestCases[oTestCase.idTestCase] = oTestCase;
103	assert len(self.dTestCases) <= len(self.aoTestCases); # Note! Can be shorter!
104
105	# Generate a testgroup lookup dictionary.
106	self.dTestGroups = dict();
107	for oTestGroup in self.aoTestGroups:
108	self.dTestGroups[oTestGroup.idTestGroup] = oTestGroup;
109	assert len(self.dTestGroups) == len(self.aoTestGroups);
110
111	#
112	# Associate extra members with the base data.
113	#
114	if self.aoTestGroups:
115	# Prep the test groups.
116	for oTestGroup in self.aoTestGroups:
117	oTestGroup.aoTestCases = list();
118	oTestGroup.dTestCases = dict();
119
120	# Link testcases to their group, both directions. Prep testcases for
121	# argument varation association.
122	oTestGroup = self.aoTestGroups[0];
123	for oTestCase in self.aoTestCases:
124	if oTestGroup.idTestGroup != oTestCase.idTestGroup:
125	oTestGroup = self.dTestGroups[oTestCase.idTestGroup];
126
127	assert oTestCase.idTestCase not in oTestGroup.dTestCases;
128	oTestGroup.dTestCases[oTestCase.idTestCase] = oTestCase;
129	oTestGroup.aoTestCases.append(oTestCase);
130	oTestCase.oTestGroup = oTestGroup;
131	oTestCase.aoArgsVariations = list();
132
133	# Associate testcase argument variations with their testcases (group)
134	# in both directions.
135	oTestGroup = self.aoTestGroups[0];
136	oTestCase = self.aoTestCases[0] if self.aoTestCases else None;
137	for oArgVariation in self.aoArgsVariations:
138	if oTestGroup.idTestGroup != oArgVariation.idTestGroup:
139	oTestGroup = self.dTestGroups[oArgVariation.idTestGroup];
140	if oTestCase.idTestCase != oArgVariation.idTestCase or oTestCase.idTestGroup != oArgVariation.idTestGroup:
141	oTestCase = oTestGroup.dTestCases[oArgVariation.idTestCase];
142
143	oTestCase.aoArgsVariations.append(oArgVariation);
144	oArgVariation.oTestCase = oTestCase;
145	oArgVariation.oTestGroup = oTestGroup;
146
147	else:
148	assert not self.aoTestCases;
149	assert not self.aoArgsVariations;
150	# done.
151
152	@staticmethod
153	def _addPreReqError(aoErrors, aidChain, oObj, sMsg):
154	""" Returns a chain of IDs error entry. """
155
156	sMsg += ' Dependency chain: %s' % (aidChain[0],);
157	for i in range(1, len(aidChain)):
158	sMsg += ' -> %s' % (aidChain[i],);
159
160	aoErrors.append([sMsg, oObj]);
161	return aoErrors;
162
163	def checkForGroupDepCycles(self):
164	"""
165	Checks for testgroup depencency cycles and any missing testgroup
166	dependencies.
167	Returns array of errors (see SchedulderBase.recreateQueue()).
168	"""
169	aoErrors = list();
170	for oTestGroup in self.aoTestGroups:
171	idPreReq = oTestGroup.idTestGroupPreReq;
172	if idPreReq is None:
173	oTestGroup.aidTestGroupPreReqs = list();
174	continue;
175
176	aidChain = [oTestGroup.idTestGroup,];
177	while idPreReq is not None:
178	aidChain.append(idPreReq);
179	if len(aidChain) >= 10:
180	self._addPreReqError(aoErrors, aidChain, oTestGroup,
181	'TestGroup #%s prerequisite chain is too long!'
182	% (oTestGroup.idTestGroup,));
183	break;
184
185	oDep = self.dTestGroups.get(idPreReq, None);
186	if oDep is None:
187	self._addPreReqError(aoErrors, aidChain, oTestGroup,
188	'TestGroup #%s prerequisite #%s is not in the scheduling group!'
189	% (oTestGroup.idTestGroup, idPreReq,));
190	break;
191
192	idPreReq = oDep.idTestGroupPreReq;
193	oTestGroup.aidTestGroupPreReqs = aidChain[1:];
194
195	return aoErrors;
196
197
198	def checkForMissingTestCaseDeps(self):
199	"""
200	Checks that testcase dependencies stays within bounds. We do not allow
201	dependencies outside a testgroup, no dependency cycles or even remotely
202	long dependency chains.
203
204	Returns array of errors (see SchedulderBase.recreateQueue()).
205	"""
206	aoErrors = list();
207	for oTestGroup in self.aoTestGroups:
208	for oTestCase in oTestGroup.aoTestCases:
209	if not oTestCase.aidPreReqs:
210	continue;
211
212	# Stupid recursion code using special stack(s).
213	aiIndexes = [[oTestCase, 0], ];
214	aidChain = [oTestCase.idTestGroup,];
215	while aiIndexes:
216	(oCur, i) = aiIndexes[-1];
217	if i >= len(oCur.aidPreReqs):
218	aiIndexes.pop();
219	aidChain.pop();
220	else:
221	aiIndexes[-1][1] = i + 1; # whatever happens, we'll advance on the current level.
222
223	idPreReq = oTestCase.aidPreReqs[i];
224	oDep = oTestGroup.dTestCases.get(idPreReq, None);
225	if oDep is None:
226	self._addPreReqError(aoErrors, aidChain, oTestCase,
227	'TestCase #%s prerequisite #%s is not in the scheduling group!'
228	% (oTestCase.idTestCase, idPreReq));
229	elif idPreReq in aidChain:
230	self._addPreReqError(aoErrors, aidChain, oTestCase,
231	'TestCase #%s prerequisite #%s creates a cycle!'
232	% (oTestCase.idTestCase, idPreReq));
233	elif not oDep.aiPreReqs:
234	pass;
235	elif len(aidChain) >= 10:
236	self._addPreReqError(aoErrors, aidChain, oTestCase,
237	'TestCase #%s prerequisite chain is too long!' % (oTestCase.idTestCase,));
238	else:
239	aiIndexes.append([oDep, 0]);
240	aidChain.append(idPreReq);
241
242	return aoErrors;
243
244	def deepTestGroupSort(self):
245	"""
246	Sorts the testgroups and their testcases by priority and dependencies.
247	Note! Don't call this before checking for dependency cycles!
248	"""
249	if not self.aoTestGroups:
250	return;
251
252	#
253	# ASSUMES groups as well as testcases are sorted by priority by the
254	# database. So we only have to concern ourselves with the dependency
255	# sorting.
256	#
257	iGrpPrio = self.aoTestGroups[0].iSchedPriority;
258	for iTestGroup, oTestGroup in enumerate(self.aoTestGroups):
259	if oTestGroup.iSchedPriority > iGrpPrio:
260	raise TMExceptionBase('Incorrectly sorted testgroups returned by database: iTestGroup=%s prio=%s %s'
261	% ( iTestGroup, iGrpPrio,
262	', '.join(['(%s: %s)' % (oCur.idTestGroup, oCur.iSchedPriority)
263	for oCur in self.aoTestGroups]), ) );
264	iGrpPrio = oTestGroup.iSchedPriority;
265
266	if oTestGroup.aoTestCases:
267	iTstPrio = oTestGroup.aoTestCases[0].iSchedPriority;
268	for iTestCase, oTestCase in enumerate(oTestGroup.aoTestCases):
269	if oTestCase.iSchedPriority > iTstPrio:
270	raise TMExceptionBase('Incorrectly sorted testcases returned by database: i=%s prio=%s idGrp=%s %s'
271	% ( iTestCase, iTstPrio, oTestGroup.idTestGroup,
272	', '.join(['(%s: %s)' % (oCur.idTestCase, oCur.iSchedPriority)
273	for oCur in oTestGroup.aoTestCases]),));
274
275	#
276	# Sort the testgroups by dependencies.
277	#
278	i = 0;
279	while i < len(self.aoTestGroups):
280	oTestGroup = self.aoTestGroups[i];
281	if oTestGroup.idTestGroupPreReq is not None:
282	iPreReq = self.aoTestGroups.index(self.dTestGroups[oTestGroup.idTestGroupPreReq]);
283	if iPreReq > i:
284	# The prerequisite is after the current entry. Move the
285	# current entry so that it's following it's prereq entry.
286	self.aoTestGroups.insert(iPreReq + 1, oTestGroup);
287	self.aoTestGroups.pop(i);
288	continue;
289	assert iPreReq < i;
290	i += 1; # Advance.
291
292	#
293	# Sort the testcases by dependencies.
294	# Same algorithm as above, just more prerequisites.
295	#
296	for oTestGroup in self.aoTestGroups:
297	i = 0;
298	while i < len(oTestGroup.aoTestCases):
299	oTestCase = oTestGroup.aoTestCases[i];
300	if oTestCase.aidPreReqs:
301	for idPreReq in oTestCase.aidPreReqs:
302	iPreReq = oTestGroup.aoTestCases.index(oTestGroup.dTestCases[idPreReq]);
303	if iPreReq > i:
304	# The prerequisite is after the current entry. Move the
305	# current entry so that it's following it's prereq entry.
306	oTestGroup.aoTestGroups.insert(iPreReq + 1, oTestCase);
307	oTestGroup.aoTestGroups.pop(i);
308	i -= 1; # Don't advance.
309	break;
310	assert iPreReq < i;
311	i += 1; # Advance.
312
313
314
315	class SchedQueueData(ModelDataBase):
316	"""
317	Scheduling queue data item.
318	"""
319
320	ksIdAttr = 'idSchedGroup';
321
322	ksParam_idSchedGroup = 'SchedQueueData_idSchedGroup';
323	ksParam_idItem = 'SchedQueueData_idItem';
324	ksParam_offQueue = 'SchedQueueData_offQueue';
325	ksParam_idGenTestCaseArgs = 'SchedQueueData_idGenTestCaseArgs';
326	ksParam_idTestGroup = 'SchedQueueData_idTestGroup';
327	ksParam_aidTestGroupPreReqs = 'SchedQueueData_aidTestGroupPreReqs';
328	ksParam_bmHourlySchedule = 'SchedQueueData_bmHourlySchedule';
329	ksParam_tsConfig = 'SchedQueueData_tsConfig';
330	ksParam_tsLastScheduled = 'SchedQueueData_tsLastScheduled';
331	ksParam_idTestSetGangLeader = 'SchedQueueData_idTestSetGangLeader';
332	ksParam_cMissingGangMembers = 'SchedQueueData_cMissingGangMembers';
333
334	kasAllowNullAttributes = [ 'idItem', 'offQueue', 'aidTestGroupPreReqs', 'bmHourlySchedule', 'idTestSetGangLeader',
335	'tsConfig', 'tsLastScheduled' ];
336
337
338	def __init__(self):
339	ModelDataBase.__init__(self);
340
341	#
342	# Initialize with defaults.
343	# See the database for explanations of each of these fields.
344	#
345	self.idSchedGroup = None;
346	self.idItem = None;
347	self.offQueue = None;
348	self.idGenTestCaseArgs = None;
349	self.idTestGroup = None;
350	self.aidTestGroupPreReqs = None;
351	self.bmHourlySchedule = None;
352	self.tsConfig = None;
353	self.tsLastScheduled = None;
354	self.idTestSetGangLeader = None;
355	self.cMissingGangMembers = 1;
356
357	def initFromValues(self, idSchedGroup, idGenTestCaseArgs, idTestGroup, aidTestGroupPreReqs, # pylint: disable=too-many-arguments
358	bmHourlySchedule, cMissingGangMembers,
359	idItem = None, offQueue = None, tsConfig = None, tsLastScheduled = None, idTestSetGangLeader = None):
360	"""
361	Reinitialize with all attributes potentially given as inputs.
362	Return self.
363	"""
364	self.idSchedGroup = idSchedGroup;
365	self.idItem = idItem;
366	self.offQueue = offQueue;
367	self.idGenTestCaseArgs = idGenTestCaseArgs;
368	self.idTestGroup = idTestGroup;
369	self.aidTestGroupPreReqs = aidTestGroupPreReqs;
370	self.bmHourlySchedule = bmHourlySchedule;
371	self.tsConfig = tsConfig;
372	self.tsLastScheduled = tsLastScheduled;
373	self.idTestSetGangLeader = idTestSetGangLeader;
374	self.cMissingGangMembers = cMissingGangMembers;
375	return self;
376
377	def initFromDbRow(self, aoRow):
378	"""
379	Initialize from database row (SELECT * FROM SchedQueues).
380	Returns self.
381	Raises exception if no row is specfied.
382	"""
383	if aoRow is None:
384	raise TMExceptionBase('SchedQueueData not found.');
385
386	self.idSchedGroup = aoRow[0];
387	self.idItem = aoRow[1];
388	self.offQueue = aoRow[2];
389	self.idGenTestCaseArgs = aoRow[3];
390	self.idTestGroup = aoRow[4];
391	self.aidTestGroupPreReqs = aoRow[5];
392	self.bmHourlySchedule = aoRow[6];
393	self.tsConfig = aoRow[7];
394	self.tsLastScheduled = aoRow[8];
395	self.idTestSetGangLeader = aoRow[9];
396	self.cMissingGangMembers = aoRow[10];
397	return self;
398
399
400
401
402
403
404	class SchedulerBase(object):
405	"""
406	The scheduler base class.
407
408	The scheduler classes have two functions:
409	1. Recreate the scheduling queue.
410	2. Pick the next task from the queue.
411
412	The first is scheduler specific, the latter isn't.
413	"""
414
415	class BuildCache(object):
416	""" Build cache. """
417
418	class BuildCacheIterator(object):
419	""" Build class iterator. """
420	def __init__(self, oCache):
421	self.oCache = oCache;
422	self.iCur = 0;
423
424	def __iter__(self):
425	"""Returns self, required by the language."""
426	return self;
427
428	def __next__(self):
429	"""Returns the next build, raises StopIteration when the end has been reached."""
430	while True:
431	if self.iCur >= len(self.oCache.aoEntries):
432	oEntry = self.oCache.fetchFromCursor();
433	if oEntry is None:
434	raise StopIteration;
435	else:
436	oEntry = self.oCache.aoEntries[self.iCur];
437	self.iCur += 1;
438	if not oEntry.fRemoved:
439	return oEntry;
440	return None; # not reached, but make pylint happy (for now).
441
442	def next(self):
443	""" For python 2.x. """
444	return self.__next__();
445
446	class BuildCacheEntry(object):
447	""" Build cache entry. """
448
449	def __init__(self, oBuild, fMaybeBlacklisted):
450	self.oBuild = oBuild;
451	self._fBlacklisted = None if fMaybeBlacklisted is True else False;
452	self.fRemoved = False;
453	self._dPreReqDecisions = dict();
454
455	def remove(self):
456	"""
457	Marks the cache entry as removed.
458	This doesn't actually remove it from the cache array, only marks
459	it as removed. It has no effect on open iterators.
460	"""
461	self.fRemoved = True;
462
463	def getPreReqDecision(self, sPreReqSet):
464	"""
465	Retrieves a cached prerequisite decision.
466	Returns boolean if found, None if not.
467	"""
468	return self._dPreReqDecisions.get(sPreReqSet);
469
470	def setPreReqDecision(self, sPreReqSet, fDecision):
471	"""
472	Caches a prerequistie decision.
473	"""
474	self._dPreReqDecisions[sPreReqSet] = fDecision;
475	return fDecision;
476
477	def isBlacklisted(self, oDb):
478	""" Checks if the build is blacklisted. """
479	if self._fBlacklisted is None:
480	self._fBlacklisted = BuildLogic(oDb).isBuildBlacklisted(self.oBuild);
481	return self._fBlacklisted;
482
483
484	def __init__(self):
485	self.aoEntries = [];
486	self.oCursor = None;
487
488	def setupSource(self, oDb, idBuildSrc, sOs, sCpuArch, tsNow):
489	""" Configures the build cursor for the cache. """
490	if not self.aoEntries and self.oCursor is None:
491	oBuildSource = BuildSourceData().initFromDbWithId(oDb, idBuildSrc, tsNow);
492	self.oCursor = BuildSourceLogic(oDb).openBuildCursor(oBuildSource, sOs, sCpuArch, tsNow);
493	return True;
494
495	def __iter__(self):
496	"""Return an iterator."""
497	return self.BuildCacheIterator(self);
498
499	def fetchFromCursor(self):
500	""" Fetches a build from the cursor and adds it to the cache."""
501	if self.oCursor is None:
502	return None;
503
504	try:
505	aoRow = self.oCursor.fetchOne();
506	except:
507	return None;
508	if aoRow is None:
509	return None;
510
511	oBuild = BuildDataEx().initFromDbRow(aoRow);
512	oEntry = self.BuildCacheEntry(oBuild, aoRow[-1]);
513	self.aoEntries.append(oEntry);
514	return oEntry;
515
516	def __init__(self, oDb, oSchedGrpData, iVerbosity = 0, tsSecStart = None):
517	self._oDb = oDb;
518	self._oSchedGrpData = oSchedGrpData;
519	self._iVerbosity = iVerbosity;
520	self._asMessages = [];
521	self._tsSecStart = tsSecStart if tsSecStart is not None else utils.timestampSecond();
522	self.oBuildCache = self.BuildCache();
523	self.dTestGroupMembers = dict();
524
525	@staticmethod
526	def _instantiate(oDb, oSchedGrpData, iVerbosity = 0, tsSecStart = None):
527	"""
528	Instantiate the scheduler specified by the scheduling group.
529	Returns scheduler child class instance. May raise exception if
530	the input is invalid.
531	"""
532	if oSchedGrpData.enmScheduler == SchedGroupData.ksScheduler_BestEffortContinuousIntegration:
533	from testmanager.core.schedulerbeci import SchdulerBeci;
534	oScheduler = SchdulerBeci(oDb, oSchedGrpData, iVerbosity, tsSecStart);
535	else:
536	raise oDb.integrityException('Invalid scheduler "%s", idSchedGroup=%d' \
537	% (oSchedGrpData.enmScheduler, oSchedGrpData.idSchedGroup));
538	return oScheduler;
539
540
541	#
542	# Misc.
543	#
544
545	def msgDebug(self, sText):
546	"""Debug printing."""
547	if self._iVerbosity > 1:
548	self._asMessages.append('debug:' + sText);
549	return None;
550
551	def msgInfo(self, sText):
552	"""Info printing."""
553	if self._iVerbosity > 1:
554	self._asMessages.append('info: ' + sText);
555	return None;
556
557	def dprint(self, sMsg):
558	"""Prints a debug message to the srv glue log (see config.py). """
559	if config.g_kfSrvGlueDebugScheduler:
560	self._oDb.dprint(sMsg);
561	return None;
562
563	def getElapsedSecs(self):
564	""" Returns the number of seconds this scheduling task has been running. """
565	tsSecNow = utils.timestampSecond();
566	if tsSecNow < self._tsSecStart: # paranoia
567	self._tsSecStart = tsSecNow;
568	return tsSecNow - self._tsSecStart;
569
570
571	#
572	# Create schedule.
573	#
574
575	def _recreateQueueCancelGatherings(self):
576	"""
577	Cancels all pending gang gatherings on the current queue.
578	"""
579	self._oDb.execute('SELECT idTestSetGangLeader\n'
580	'FROM SchedQueues\n'
581	'WHERE idSchedGroup = %s\n'
582	' AND idTestSetGangLeader is not NULL\n'
583	, (self._oSchedGrpData.idSchedGroup,));
584	if self._oDb.getRowCount() > 0:
585	oTBStatusLogic = TestBoxStatusLogic(self._oDb);
586	for aoRow in self._oDb.fetchAll():
587	idTestSetGangLeader = aoRow[0];
588	oTBStatusLogic.updateGangStatus(idTestSetGangLeader,
589	TestBoxStatusData.ksTestBoxState_GangGatheringTimedOut,
590	fCommit = False);
591	return True;
592
593	def _recreateQueueItems(self, oData):
594	"""
595	Returns an array of queue items (SchedQueueData).
596	Child classes must override this.
597	"""
598	_ = oData;
599	return [];
600
601	def recreateQueueWorker(self):
602	"""
603	Worker for recreateQueue.
604	"""
605
606	#
607	# Collect the necessary data and validate it.
608	#
609	oData = ReCreateQueueData(self._oDb, self._oSchedGrpData.idSchedGroup);
610	aoErrors = oData.checkForGroupDepCycles();
611	aoErrors.extend(oData.checkForMissingTestCaseDeps());
612	if not aoErrors:
613	oData.deepTestGroupSort();
614
615	#
616	# The creation of the scheduling queue is done by the child class.
617	#
618	# We will try guess where in queue we're currently at and rotate
619	# the items such that we will resume execution in the approximately
620	# same position. The goal of the scheduler is to provide a 100%
621	# deterministic result so that if we regenerate the queue when there
622	# are no changes to the testcases, testgroups or scheduling groups
623	# involved, test execution will be unchanged (save for maybe just a
624	# little for gang gathering).
625	#
626	aoItems = list();
627	if not oData.oSchedGroup.fEnabled:
628	self.msgInfo('Disabled.');
629	elif not oData.aoArgsVariations:
630	self.msgInfo('Found no test case argument variations.');
631	else:
632	aoItems = self._recreateQueueItems(oData);
633	self.msgDebug('len(aoItems)=%s' % (len(aoItems),));
634	#for i in range(len(aoItems)):
635	# self.msgDebug('aoItems[%2d]=%s' % (i, aoItems[i]));
636	if aoItems:
637	self._oDb.execute('SELECT offQueue FROM SchedQueues WHERE idSchedGroup = %s ORDER BY idItem LIMIT 1'
638	, (self._oSchedGrpData.idSchedGroup,));
639	if self._oDb.getRowCount() > 0:
640	offQueue = self._oDb.fetchOne()[0];
641	self._oDb.execute('SELECT COUNT(*) FROM SchedQueues WHERE idSchedGroup = %s'
642	, (self._oSchedGrpData.idSchedGroup,));
643	cItems = self._oDb.fetchOne()[0];
644	offQueueNew = (offQueue * cItems) // len(aoItems);
645	if offQueueNew != 0:
646	aoItems = aoItems[offQueueNew:] + aoItems[:offQueueNew];
647
648	#
649	# Replace the scheduling queue.
650	# Care need to be take to first timeout/abort any gangs in the
651	# gathering state since these use the queue to set up the date.
652	#
653	self._recreateQueueCancelGatherings();
654	self._oDb.execute('DELETE FROM SchedQueues WHERE idSchedGroup = %s\n', (self._oSchedGrpData.idSchedGroup,));
655	if aoItems:
656	self._oDb.insertList('INSERT INTO SchedQueues (\n'
657	' idSchedGroup,\n'
658	' offQueue,\n'
659	' idGenTestCaseArgs,\n'
660	' idTestGroup,\n'
661	' aidTestGroupPreReqs,\n'
662	' bmHourlySchedule,\n'
663	' cMissingGangMembers )\n',
664	aoItems, self._formatItemForInsert);
665	return (aoErrors, self._asMessages);
666
667	def _formatItemForInsert(self, oItem):
668	"""
669	Used by recreateQueueWorker together with TMDatabaseConnect::insertList
670	"""
671	return self._oDb.formatBindArgs('(%s,%s,%s,%s,%s,%s,%s)'
672	, ( oItem.idSchedGroup,
673	oItem.offQueue,
674	oItem.idGenTestCaseArgs,
675	oItem.idTestGroup,
676	oItem.aidTestGroupPreReqs if oItem.aidTestGroupPreReqs else None,
677	oItem.bmHourlySchedule,
678	oItem.cMissingGangMembers
679	));
680
681	@staticmethod
682	def recreateQueue(oDb, uidAuthor, idSchedGroup, iVerbosity = 1):
683	"""
684	(Re-)creates the scheduling queue for the given group.
685
686	Returns (asMessages, asMessages). On success the array with the error
687	will be empty, on failure it will contain (sError, oRelatedObject)
688	entries. The messages is for debugging and are simple strings.
689
690	Raises exception database error.
691	"""
692
693	aoExtraMsgs = [];
694	if oDb.debugIsExplainEnabled():
695	aoExtraMsgs += ['Warning! Disabling SQL explain to avoid deadlocking against locked tables.'];
696	oDb.debugDisableExplain();
697
698	aoErrors = [];
699	asMessages = [];
700	try:
701	#
702	# To avoid concurrency issues (SchedQueues) and inconsistent data (*),
703	# we lock quite a few tables while doing this work. We access more
704	# data than scheduleNewTask so we lock some additional tables.
705	#
706	oDb.rollback();
707	oDb.begin();
708	oDb.execute('LOCK TABLE SchedGroups, SchedGroupMembers, TestGroups, TestGroupMembers IN SHARE MODE');
709	oDb.execute('LOCK TABLE TestBoxes, TestCaseArgs, TestCases IN SHARE MODE');
710	oDb.execute('LOCK TABLE TestBoxStatuses, SchedQueues IN EXCLUSIVE MODE');
711
712	#
713	# Instantiate the scheduler and call the worker function.
714	#
715	oSchedGrpData = SchedGroupData().initFromDbWithId(oDb, idSchedGroup);
716	oScheduler = SchedulerBase._instantiate(oDb, oSchedGrpData, iVerbosity);
717
718	(aoErrors, asMessages) = oScheduler.recreateQueueWorker();
719	if not aoErrors:
720	SystemLogLogic(oDb).addEntry(SystemLogData.ksEvent_SchedQueueRecreate,
721	'User #%d recreated sched queue #%d.' % (uidAuthor, idSchedGroup,));
722	oDb.commit();
723	else:
724	oDb.rollback();
725
726	except:
727	oDb.rollback();
728	raise;
729
730	return (aoErrors, aoExtraMsgs + asMessages);
731
732
733	@staticmethod
734	def cleanUpOrphanedQueues(oDb):
735	"""
736	Removes orphan scheduling queues from the SchedQueues table.
737
738	Queues becomes orphaned when the scheduling group they belongs to has been deleted.
739
740	Returns number of orphaned queues.
741	Raises exception database error.
742	"""
743	cRet = 0;
744	try:
745	oDb.rollback();
746	oDb.begin();
747	oDb.execute('''
748	SELECT SchedQueues.idSchedGroup
749	FROM SchedQueues
750	LEFT OUTER JOIN SchedGroups
751	ON SchedGroups.idSchedGroup = SchedQueues.idSchedGroup
752	AND SchedGroups.tsExpire = 'infinity'::TIMESTAMP
753	WHERE SchedGroups.idSchedGroup is NULL
754	GROUP BY SchedQueues.idSchedGroup''');
755	aaoOrphanRows = oDb.fetchAll();
756	cRet = len(aaoOrphanRows);
757	if cRet > 0:
758	oDb.execute('DELETE FROM SchedQueues WHERE idSchedGroup IN (%s)'
759	% (','.join([str(aoRow[0]) for aoRow in aaoOrphanRows]),));
760	oDb.commit();
761	except:
762	oDb.rollback();
763	raise;
764	return cRet;
765
766
767	#
768	# Schedule Task.
769	#
770
771	def _composeGangArguments(self, idTestSet):
772	"""
773	Composes the gang specific testdriver arguments.
774	Returns command line string, including a leading space.
775	"""
776
777	oTestSet = TestSetData().initFromDbWithId(self._oDb, idTestSet);
778	aoGangMembers = TestSetLogic(self._oDb).getGang(oTestSet.idTestSetGangLeader);
779
780	sArgs = ' --gang-member-no %s --gang-members %s' % (oTestSet.iGangMemberNo, len(aoGangMembers));
781	for i, _ in enumerate(aoGangMembers):
782	sArgs = ' --gang-ipv4-%s %s' % (i, aoGangMembers[i].ip); ## @todo IPv6
783
784	return sArgs;
785
786
787	def composeExecResponseWorker(self, idTestSet, oTestEx, oTestBox, oBuild, oValidationKitBuild, sBaseUrl):
788	"""
789	Given all the bits of data, compose an EXEC command response to the testbox.
790	"""
791	sScriptZips = oTestEx.oTestCase.sValidationKitZips;
792	if sScriptZips is None or sScriptZips.find('@VALIDATIONKIT_ZIP@') >= 0:
793	assert oValidationKitBuild;
794	if sScriptZips is None:
795	sScriptZips = oValidationKitBuild.sBinaries;
796	else:
797	sScriptZips = sScriptZips.replace('@VALIDATIONKIT_ZIP@', oValidationKitBuild.sBinaries);
798	sScriptZips = sScriptZips.replace('@DOWNLOAD_BASE_URL@', sBaseUrl + config.g_ksTmDownloadBaseUrlRel);
799
800	sCmdLine = oTestEx.oTestCase.sBaseCmd + ' ' + oTestEx.sArgs;
801	sCmdLine = sCmdLine.replace('@BUILD_BINARIES@', oBuild.sBinaries);
802	sCmdLine = sCmdLine.strip();
803	if oTestEx.cGangMembers > 1:
804	sCmdLine += ' ' + self._composeGangArguments(idTestSet);
805
806	cSecTimeout = oTestEx.cSecTimeout if oTestEx.cSecTimeout is not None else oTestEx.oTestCase.cSecTimeout;
807	cSecTimeout = cSecTimeout * oTestBox.pctScaleTimeout // 100;
808
809	dResponse = \
810	{
811	constants.tbresp.ALL_PARAM_RESULT: constants.tbresp.CMD_EXEC,
812	constants.tbresp.EXEC_PARAM_RESULT_ID: idTestSet,
813	constants.tbresp.EXEC_PARAM_SCRIPT_ZIPS: sScriptZips,
814	constants.tbresp.EXEC_PARAM_SCRIPT_CMD_LINE: sCmdLine,
815	constants.tbresp.EXEC_PARAM_TIMEOUT: cSecTimeout,
816	};
817	return dResponse;
818
819	@staticmethod
820	def composeExecResponse(oDb, idTestSet, sBaseUrl, iVerbosity = 0):
821	"""
822	Composes an EXEC response for a gang member (other than the last).
823	Returns a EXEC response or raises an exception (DB/input error).
824	"""
825	#
826	# Gather the necessary data.
827	#
828	oTestSet = TestSetData().initFromDbWithId(oDb, idTestSet);
829	oTestBox = TestBoxData().initFromDbWithGenId(oDb, oTestSet.idGenTestBox);
830	oTestEx = TestCaseArgsDataEx().initFromDbWithGenIdEx(oDb, oTestSet.idGenTestCaseArgs,
831	tsConfigEff = oTestSet.tsConfig,
832	tsRsrcEff = oTestSet.tsConfig);
833	oBuild = BuildDataEx().initFromDbWithId(oDb, oTestSet.idBuild);
834	oValidationKitBuild = None;
835	if oTestSet.idBuildTestSuite is not None:
836	oValidationKitBuild = BuildDataEx().initFromDbWithId(oDb, oTestSet.idBuildTestSuite);
837
838	#
839	# Instantiate the specified scheduler and let it do the rest.
840	#
841	oSchedGrpData = SchedGroupData().initFromDbWithId(oDb, oTestSet.idSchedGroup, oTestSet.tsCreated);
842	assert oSchedGrpData.fEnabled is True;
843	assert oSchedGrpData.idBuildSrc is not None;
844	oScheduler = SchedulerBase._instantiate(oDb, oSchedGrpData, iVerbosity);
845
846	return oScheduler.composeExecResponseWorker(idTestSet, oTestEx, oTestBox, oBuild, oValidationKitBuild, sBaseUrl);
847
848
849	def _updateTask(self, oTask, tsNow):
850	"""
851	Updates a gang schedule task.
852	"""
853	assert oTask.cMissingGangMembers >= 1;
854	assert oTask.idTestSetGangLeader is not None;
855	assert oTask.idTestSetGangLeader >= 1;
856	if tsNow is not None:
857	self._oDb.execute('UPDATE SchedQueues\n'
858	' SET idTestSetGangLeader = %s,\n'
859	' cMissingGangMembers = %s,\n'
860	' tsLastScheduled = %s\n'
861	'WHERE idItem = %s\n'
862	, (oTask.idTestSetGangLeader, oTask.cMissingGangMembers, tsNow, oTask.idItem,) );
863	else:
864	self._oDb.execute('UPDATE SchedQueues\n'
865	' SET cMissingGangMembers = %s\n'
866	'WHERE idItem = %s\n'
867	, (oTask.cMissingGangMembers, oTask.idItem,) );
868	return True;
869
870	def _moveTaskToEndOfQueue(self, oTask, cGangMembers, tsNow):
871	"""
872	The task has been scheduled successfully, reset it's data move it to
873	the end of the queue.
874	"""
875	if cGangMembers > 1:
876	self._oDb.execute('UPDATE SchedQueues\n'
877	' SET idItem = NEXTVAL(\'SchedQueueItemIdSeq\'),\n'
878	' idTestSetGangLeader = NULL,\n'
879	' cMissingGangMembers = %s\n'
880	'WHERE idItem = %s\n'
881	, (cGangMembers, oTask.idItem,) );
882	else:
883	self._oDb.execute('UPDATE SchedQueues\n'
884	' SET idItem = NEXTVAL(\'SchedQueueItemIdSeq\'),\n'
885	' idTestSetGangLeader = NULL,\n'
886	' cMissingGangMembers = 1,\n'
887	' tsLastScheduled = %s\n'
888	'WHERE idItem = %s\n'
889	, (tsNow, oTask.idItem,) );
890	return True;
891
892
893
894
895	def _createTestSet(self, oTask, oTestEx, oTestBoxData, oBuild, oValidationKitBuild, tsNow):
896	# type: (SchedQueueData, TestCaseArgsDataEx, TestBoxData, BuildDataEx, BuildDataEx, datetime.datetime) -> int
897	"""
898	Creates a test set for using the given data.
899	Will not commit, someone up the callstack will that later on.
900
901	Returns the test set ID, may raise an exception on database error.
902	"""
903	# Lazy bird doesn't want to write testset.py and does it all here.
904
905	#
906	# We're getting the TestSet ID first in order to include it in the base
907	# file name (that way we can directly relate files on the disk to the
908	# test set when doing batch work), and also for idTesetSetGangLeader.
909	#
910	self._oDb.execute('SELECT NEXTVAL(\'TestSetIdSeq\')');
911	idTestSet = self._oDb.fetchOne()[0];
912
913	sBaseFilename = '%04d/%02d/%02d/%02d/TestSet-%s' \
914	% (tsNow.year, tsNow.month, tsNow.day, (tsNow.hour // 6) * 6, idTestSet);
915
916	#
917	# Gang scheduling parameters. Changes the oTask data for updating by caller.
918	#
919	iGangMemberNo = 0;
920
921	if oTestEx.cGangMembers <= 1:
922	assert oTask.idTestSetGangLeader is None;
923	assert oTask.cMissingGangMembers <= 1;
924	elif oTask.idTestSetGangLeader is None:
925	assert oTask.cMissingGangMembers == oTestEx.cGangMembers;
926	oTask.cMissingGangMembers = oTestEx.cGangMembers - 1;
927	oTask.idTestSetGangLeader = idTestSet;
928	else:
929	assert oTask.cMissingGangMembers > 0 and oTask.cMissingGangMembers < oTestEx.cGangMembers;
930	oTask.cMissingGangMembers -= 1;
931
932	#
933	# Do the database stuff.
934	#
935	self._oDb.execute('INSERT INTO TestSets (\n'
936	' idTestSet,\n'
937	' tsConfig,\n'
938	' tsCreated,\n'
939	' idBuild,\n'
940	' idBuildCategory,\n'
941	' idBuildTestSuite,\n'
942	' idGenTestBox,\n'
943	' idTestBox,\n'
944	' idSchedGroup,\n'
945	' idTestGroup,\n'
946	' idGenTestCase,\n'
947	' idTestCase,\n'
948	' idGenTestCaseArgs,\n'
949	' idTestCaseArgs,\n'
950	' sBaseFilename,\n'
951	' iGangMemberNo,\n'
952	' idTestSetGangLeader )\n'
953	'VALUES ( %s,\n' # idTestSet
954	' %s,\n' # tsConfig
955	' %s,\n' # tsCreated
956	' %s,\n' # idBuild
957	' %s,\n' # idBuildCategory
958	' %s,\n' # idBuildTestSuite
959	' %s,\n' # idGenTestBox
960	' %s,\n' # idTestBox
961	' %s,\n' # idSchedGroup
962	' %s,\n' # idTestGroup
963	' %s,\n' # idGenTestCase
964	' %s,\n' # idTestCase
965	' %s,\n' # idGenTestCaseArgs
966	' %s,\n' # idTestCaseArgs
967	' %s,\n' # sBaseFilename
968	' %s,\n' # iGangMemberNo
969	' %s)\n' # idTestSetGangLeader
970	, ( idTestSet,
971	oTask.tsConfig,
972	tsNow,
973	oBuild.idBuild,
974	oBuild.idBuildCategory,
975	oValidationKitBuild.idBuild if oValidationKitBuild is not None else None,
976	oTestBoxData.idGenTestBox,
977	oTestBoxData.idTestBox,
978	oTask.idSchedGroup,
979	oTask.idTestGroup,
980	oTestEx.oTestCase.idGenTestCase,
981	oTestEx.oTestCase.idTestCase,
982	oTestEx.idGenTestCaseArgs,
983	oTestEx.idTestCaseArgs,
984	sBaseFilename,
985	iGangMemberNo,
986	oTask.idTestSetGangLeader,
987	));
988
989	self._oDb.execute('INSERT INTO TestResults (\n'
990	' idTestResultParent,\n'
991	' idTestSet,\n'
992	' tsCreated,\n'
993	' idStrName,\n'
994	' cErrors,\n'
995	' enmStatus,\n'
996	' iNestingDepth)\n'
997	'VALUES ( NULL,\n' # idTestResultParent
998	' %s,\n' # idTestSet
999	' %s,\n' # tsCreated
1000	' 0,\n' # idStrName
1001	' 0,\n' # cErrors
1002	' \'running\'::TestStatus_T,\n'
1003	' 0)\n' # iNestingDepth
1004	'RETURNING idTestResult'
1005	, ( idTestSet, tsNow, ));
1006	idTestResult = self._oDb.fetchOne()[0];
1007
1008	self._oDb.execute('UPDATE TestSets\n'
1009	' SET idTestResult = %s\n'
1010	'WHERE idTestSet = %s\n'
1011	, (idTestResult, idTestSet, ));
1012
1013	return idTestSet;
1014
1015	def _tryFindValidationKitBit(self, oTestBoxData, tsNow):
1016	"""
1017	Tries to find the most recent validation kit build suitable for the given testbox.
1018	Returns BuildDataEx or None. Raise exception on database error.
1019
1020	Can be overridden by child classes to change the default build requirements.
1021	"""
1022	oBuildLogic = BuildLogic(self._oDb);
1023	oBuildSource = BuildSourceData().initFromDbWithId(self._oDb, self._oSchedGrpData.idBuildSrcTestSuite, tsNow);
1024	oCursor = BuildSourceLogic(self._oDb).openBuildCursor(oBuildSource, oTestBoxData.sOs, oTestBoxData.sCpuArch, tsNow);
1025	for _ in range(oCursor.getRowCount()):
1026	oBuild = BuildDataEx().initFromDbRow(oCursor.fetchOne());
1027	if not oBuildLogic.isBuildBlacklisted(oBuild):
1028	return oBuild;
1029	return None;
1030
1031	def _tryFindBuild(self, oTask, oTestEx, oTestBoxData, tsNow):
1032	"""
1033	Tries to find a fitting build.
1034	Returns BuildDataEx or None. Raise exception on database error.
1035
1036	Can be overridden by child classes to change the default build requirements.
1037	"""
1038
1039	#
1040	# Gather the set of prerequisites we have and turn them into a value
1041	# set for use in the loop below.
1042	#
1043	# Note! We're scheduling on testcase level and ignoring argument variation
1044	# selections in TestGroupMembers is intentional.
1045	#
1046	dPreReqs = {};
1047
1048	# Direct prerequisites. We assume they're all enabled as this can be
1049	# checked at queue creation time.
1050	for oPreReq in oTestEx.aoTestCasePreReqs:
1051	dPreReqs[oPreReq.idTestCase] = 1;
1052
1053	# Testgroup dependencies from the scheduling group config.
1054	if oTask.aidTestGroupPreReqs is not None:
1055	for iTestGroup in oTask.aidTestGroupPreReqs:
1056	# Make sure the _active_ test group members are in the cache.
1057	if iTestGroup not in self.dTestGroupMembers:
1058	self._oDb.execute('SELECT DISTINCT TestGroupMembers.idTestCase\n'
1059	'FROM TestGroupMembers, TestCases\n'
1060	'WHERE TestGroupMembers.idTestGroup = %s\n'
1061	' AND TestGroupMembers.tsExpire > %s\n'
1062	' AND TestGroupMembers.tsEffective <= %s\n'
1063	' AND TestCases.idTestCase = TestGroupMembers.idTestCase\n'
1064	' AND TestCases.tsExpire > %s\n'
1065	' AND TestCases.tsEffective <= %s\n'
1066	' AND TestCases.fEnabled is TRUE\n'
1067	, (iTestGroup, oTask.tsConfig, oTask.tsConfig, oTask.tsConfig, oTask.tsConfig,));
1068	aidTestCases = [];
1069	for aoRow in self._oDb.fetchAll():
1070	aidTestCases.append(aoRow[0]);
1071	self.dTestGroupMembers[iTestGroup] = aidTestCases;
1072
1073	# Add the testgroup members to the prerequisites.
1074	for idTestCase in self.dTestGroupMembers[iTestGroup]:
1075	dPreReqs[idTestCase] = 1;
1076
1077	# Create a SQL values table out of them.
1078	sPreReqSet = ''
1079	if dPreReqs:
1080	for idPreReq in sorted(dPreReqs):
1081	sPreReqSet += ', (' + str(idPreReq) + ')';
1082	sPreReqSet = sPreReqSet[2:]; # drop the leading ', '.
1083
1084	#
1085	# Try the builds.
1086	#
1087	self.oBuildCache.setupSource(self._oDb, self._oSchedGrpData.idBuildSrc, oTestBoxData.sOs, oTestBoxData.sCpuArch, tsNow);
1088	for oEntry in self.oBuildCache:
1089	#
1090	# Check build requirements set by the test.
1091	#
1092	if not oTestEx.matchesBuildProps(oEntry.oBuild):
1093	continue;
1094
1095	if oEntry.isBlacklisted(self._oDb):
1096	oEntry.remove();
1097	continue;
1098
1099	#
1100	# Check prerequisites. The default scheduler is satisfied if one
1101	# argument variation has been executed successfully. It is not
1102	# satisfied if there are any failure runs.
1103	#
1104	if sPreReqSet:
1105	fDecision = oEntry.getPreReqDecision(sPreReqSet);
1106	if fDecision is None:
1107	# Check for missing prereqs.
1108	self._oDb.execute('SELECT COUNT(*)\n'
1109	'FROM (VALUES ' + sPreReqSet + ') AS PreReqs(idTestCase)\n'
1110	'LEFT OUTER JOIN (SELECT idTestSet\n'
1111	' FROM TestSets\n'
1112	' WHERE enmStatus IN (%s, %s)\n'
1113	' AND idBuild = %s\n'
1114	' ) AS TestSets\n'
1115	' ON (PreReqs.idTestCase = TestSets.idTestCase)\n'
1116	'WHERE TestSets.idTestSet is NULL\n'
1117	, ( TestSetData.ksTestStatus_Success, TestSetData.ksTestStatus_Skipped,
1118	oEntry.oBuild.idBuild, ));
1119	cMissingPreReqs = self._oDb.fetchOne()[0];
1120	if cMissingPreReqs > 0:
1121	self.dprint('build %s is missing %u prerequisites (out of %s)'
1122	% (oEntry.oBuild.idBuild, cMissingPreReqs, sPreReqSet,));
1123	oEntry.setPreReqDecision(sPreReqSet, False);
1124	continue;
1125
1126	# Check for failed prereq runs.
1127	self._oDb.execute('SELECT COUNT(*)\n'
1128	'FROM (VALUES ' + sPreReqSet + ') AS PreReqs(idTestCase),\n'
1129	' TestSets\n'
1130	'WHERE PreReqs.idTestCase = TestSets.idTestCase\n'
1131	' AND TestSets.idBuild = %s\n'
1132	' AND TestSets.enmStatus IN (%s, %s, %s)\n'
1133	, ( oEntry.oBuild.idBuild,
1134	TestSetData.ksTestStatus_Failure,
1135	TestSetData.ksTestStatus_TimedOut,
1136	TestSetData.ksTestStatus_Rebooted,
1137	)
1138	);
1139	cFailedPreReqs = self._oDb.fetchOne()[0];
1140	if cFailedPreReqs > 0:
1141	self.dprint('build %s is has %u prerequisite failures (out of %s)'
1142	% (oEntry.oBuild.idBuild, cFailedPreReqs, sPreReqSet,));
1143	oEntry.setPreReqDecision(sPreReqSet, False);
1144	continue;
1145
1146	oEntry.setPreReqDecision(sPreReqSet, True);
1147	elif not fDecision:
1148	continue;
1149
1150	#
1151	# If we can, check if the build files still exist.
1152	#
1153	if oEntry.oBuild.areFilesStillThere() is False:
1154	self.dprint('build %s no longer exists' % (oEntry.oBuild.idBuild,));
1155	oEntry.remove();
1156	continue;
1157
1158	self.dprint('found oBuild=%s' % (oEntry.oBuild,));
1159	return oEntry.oBuild;
1160	return None;
1161
1162	def _tryFindMatchingBuild(self, oLeaderBuild, oTestBoxData, idBuildSrc):
1163	"""
1164	Tries to find a matching build for gang scheduling.
1165	Returns BuildDataEx or None. Raise exception on database error.
1166
1167	Can be overridden by child classes to change the default build requirements.
1168	"""
1169	#
1170	# Note! Should probably check build prerequisites if we get a different
1171	# build back, so that we don't use a build which hasn't passed
1172	# the smoke test.
1173	#
1174	_ = idBuildSrc;
1175	return BuildLogic(self._oDb).tryFindSameBuildForOsArch(oLeaderBuild, oTestBoxData.sOs, oTestBoxData.sCpuArch);
1176
1177
1178	def _tryAsLeader(self, oTask, oTestEx, oTestBoxData, tsNow, sBaseUrl):
1179	"""
1180	Try schedule the task as a gang leader (can be a gang of one).
1181	Returns response or None. May raise exception on DB error.
1182	"""
1183
1184	# We don't wait for busy resources, we just try the next test.
1185	oTestArgsLogic = TestCaseArgsLogic(self._oDb);
1186	if not oTestArgsLogic.areResourcesFree(oTestEx):
1187	self.dprint('Cannot get global test resources!');
1188	return None;
1189
1190	#
1191	# Find a matching build (this is the difficult bit).
1192	#
1193	oBuild = self._tryFindBuild(oTask, oTestEx, oTestBoxData, tsNow);
1194	if oBuild is None:
1195	self.dprint('No build!');
1196	return None;
1197	if oTestEx.oTestCase.needValidationKitBit():
1198	oValidationKitBuild = self._tryFindValidationKitBit(oTestBoxData, tsNow);
1199	if oValidationKitBuild is None:
1200	self.dprint('No validation kit build!');
1201	return None;
1202	else:
1203	oValidationKitBuild = None;
1204
1205	#
1206	# Create a testset, allocate the resources and update the state.
1207	# Note! Since resource allocation may still fail, we create a nested
1208	# transaction so we can roll back. (Heed lock warning in docs!)
1209	#
1210	self._oDb.execute('SAVEPOINT tryAsLeader');
1211	idTestSet = self._createTestSet(oTask, oTestEx, oTestBoxData, oBuild, oValidationKitBuild, tsNow);
1212
1213	if GlobalResourceLogic(self._oDb).allocateResources(oTestBoxData.idTestBox, oTestEx.aoGlobalRsrc, fCommit = False) \
1214	is not True:
1215	self._oDb.execute('ROLLBACK TO SAVEPOINT tryAsLeader');
1216	self.dprint('Failed to allocate global resources!');
1217	return False;
1218
1219	if oTestEx.cGangMembers <= 1:
1220	# We're alone, put the task back at the end of the queue and issue EXEC cmd.
1221	self._moveTaskToEndOfQueue(oTask, oTestEx.cGangMembers, tsNow);
1222	dResponse = self.composeExecResponseWorker(idTestSet, oTestEx, oTestBoxData, oBuild, oValidationKitBuild, sBaseUrl);
1223	sTBState = TestBoxStatusData.ksTestBoxState_Testing;
1224	else:
1225	# We're missing gang members, issue WAIT cmd.
1226	self._updateTask(oTask, tsNow if idTestSet == oTask.idTestSetGangLeader else None);
1227	dResponse = { constants.tbresp.ALL_PARAM_RESULT: constants.tbresp.CMD_WAIT, };
1228	sTBState = TestBoxStatusData.ksTestBoxState_GangGathering;
1229
1230	TestBoxStatusLogic(self._oDb).updateState(oTestBoxData.idTestBox, sTBState, idTestSet, fCommit = False);
1231	self._oDb.execute('RELEASE SAVEPOINT tryAsLeader');
1232	return dResponse;
1233
1234	def _tryAsGangMember(self, oTask, oTestEx, oTestBoxData, tsNow, sBaseUrl):
1235	"""
1236	Try schedule the task as a gang member.
1237	Returns response or None. May raise exception on DB error.
1238	"""
1239
1240	#
1241	# The leader has choosen a build, we need to find a matching one for our platform.
1242	# (It's up to the scheduler decide upon how strict dependencies are to be enforced
1243	# upon subordinate group members.)
1244	#
1245	oLeaderTestSet = TestSetData().initFromDbWithId(self._oDb, oTestBoxData.idTestSetGangLeader);
1246
1247	oLeaderBuild = BuildDataEx().initFromDbWithId(self._oDb, oLeaderTestSet.idBuild);
1248	oBuild = self._tryFindMatchingBuild(oLeaderBuild, oTestBoxData, self._oSchedGrpData.idBuildSrc);
1249	if oBuild is None:
1250	return None;
1251
1252	oValidationKitBuild = None;
1253	if oLeaderTestSet.idBuildTestSuite is not None:
1254	oLeaderValidationKitBit = BuildDataEx().initFromDbWithId(self._oDb, oLeaderTestSet.idBuildTestSuite);
1255	oValidationKitBuild = self._tryFindMatchingBuild(oLeaderValidationKitBit, oTestBoxData,
1256	self._oSchedGrpData.idBuildSrcTestSuite);
1257
1258	#
1259	# Create a testset and update the state(s).
1260	#
1261	idTestSet = self._createTestSet(oTask, oTestEx, oTestBoxData, oBuild, oValidationKitBuild, tsNow);
1262
1263	oTBStatusLogic = TestBoxStatusLogic(self._oDb);
1264	if oTask.cMissingGangMembers < 1:
1265	# The whole gang is there, move the task to the end of the queue
1266	# and update the status on the other gang members.
1267	self._moveTaskToEndOfQueue(oTask, oTestEx.cGangMembers, tsNow);
1268	dResponse = self.composeExecResponseWorker(idTestSet, oTestEx, oTestBoxData, oBuild, oValidationKitBuild, sBaseUrl);
1269	sTBState = TestBoxStatusData.ksTestBoxState_GangTesting;
1270	oTBStatusLogic.updateGangStatus(oTask.idTestSetGangLeader, sTBState, fCommit = False);
1271	else:
1272	# We're still missing some gang members, issue WAIT cmd.
1273	self._updateTask(oTask, tsNow if idTestSet == oTask.idTestSetGangLeader else None);
1274	dResponse = { constants.tbresp.ALL_PARAM_RESULT: constants.tbresp.CMD_WAIT, };
1275	sTBState = TestBoxStatusData.ksTestBoxState_GangGathering;
1276
1277	oTBStatusLogic.updateState(oTestBoxData.idTestBox, sTBState, idTestSet, fCommit = False);
1278	return dResponse;
1279
1280
1281	def scheduleNewTaskWorker(self, oTestBoxData, tsNow, sBaseUrl):
1282	"""
1283	Worker for schduling a new task.
1284	"""
1285
1286	#
1287	# Iterate the scheduler queue (fetch all to avoid having to concurrent
1288	# queries), trying out each task to see if the testbox can execute it.
1289	#
1290	dRejected = {}; # variations we've already checked out and rejected.
1291	self._oDb.execute('SELECT *\n'
1292	'FROM SchedQueues\n'
1293	'WHERE idSchedGroup = %s\n'
1294	' AND ( bmHourlySchedule IS NULL\n'
1295	' OR get_bit(bmHourlySchedule, %s) = 1 )\n'
1296	'ORDER BY idItem ASC\n'
1297	, (self._oSchedGrpData.idSchedGroup, utils.getLocalHourOfWeek()) );
1298	aaoRows = self._oDb.fetchAll();
1299	for aoRow in aaoRows:
1300	# Don't loop forever.
1301	if self.getElapsedSecs() >= config.g_kcSecMaxNewTask:
1302	break;
1303
1304	# Unpack the data and check if we've rejected the testcasevar/group variation already (they repeat).
1305	oTask = SchedQueueData().initFromDbRow(aoRow);
1306	if config.g_kfSrvGlueDebugScheduler:
1307	self.dprint('** Considering: idItem=%s idGenTestCaseArgs=%s idTestGroup=%s Deps=%s last=%s cfg=%s\n'
1308	% ( oTask.idItem, oTask.idGenTestCaseArgs, oTask.idTestGroup, oTask.aidTestGroupPreReqs,
1309	oTask.tsLastScheduled, oTask.tsConfig,));
1310
1311	sRejectNm = '%s:%s' % (oTask.idGenTestCaseArgs, oTask.idTestGroup,);
1312	if sRejectNm in dRejected:
1313	self.dprint('Duplicate, already rejected! (%s)' % (sRejectNm,));
1314	continue;
1315	dRejected[sRejectNm] = 1;
1316
1317	# Fetch all the test case info (too much, but who cares right now).
1318	oTestEx = TestCaseArgsDataEx().initFromDbWithGenIdEx(self._oDb, oTask.idGenTestCaseArgs,
1319	tsConfigEff = oTask.tsConfig,
1320	tsRsrcEff = oTask.tsConfig);
1321	if config.g_kfSrvGlueDebugScheduler:
1322	self.dprint('TestCase "%s": %s %s' % (oTestEx.oTestCase.sName, oTestEx.oTestCase.sBaseCmd, oTestEx.sArgs,));
1323
1324	# This shouldn't happen, but just in case it does...
1325	if oTestEx.oTestCase.fEnabled is not True:
1326	self.dprint('Testcase is not enabled!!');
1327	continue;
1328
1329	# Check if the testbox properties matches the test.
1330	if not oTestEx.matchesTestBoxProps(oTestBoxData):
1331	self.dprint('Testbox mismatch!');
1332	continue;
1333
1334	# Try schedule it.
1335	if oTask.idTestSetGangLeader is None or oTestEx.cGangMembers <= 1:
1336	dResponse = self._tryAsLeader(oTask, oTestEx, oTestBoxData, tsNow, sBaseUrl);
1337	elif oTask.cMissingGangMembers > 1:
1338	dResponse = self._tryAsGangMember(oTask, oTestEx, oTestBoxData, tsNow, sBaseUrl);
1339	else:
1340	dResponse = None; # Shouldn't happen!
1341	if dResponse is not None:
1342	self.dprint('Found a task! dResponse=%s' % (dResponse,));
1343	return dResponse;
1344
1345	# Found no suitable task.
1346	return None;
1347
1348	@staticmethod
1349	def _pickSchedGroup(oTestBoxDataEx, iWorkItem, dIgnoreSchedGroupIds):
1350	"""
1351	Picks the next scheduling group for the given testbox.
1352	"""
1353	if len(oTestBoxDataEx.aoInSchedGroups) == 1:
1354	oSchedGroup = oTestBoxDataEx.aoInSchedGroups[0].oSchedGroup;
1355	if oSchedGroup.fEnabled \
1356	and oSchedGroup.idBuildSrc is not None \
1357	and oSchedGroup.idSchedGroup not in dIgnoreSchedGroupIds:
1358	return (oSchedGroup, 0);
1359	iWorkItem = 0;
1360
1361	elif oTestBoxDataEx.aoInSchedGroups:
1362	# Construct priority table of currently enabled scheduling groups.
1363	aaoList1 = [];
1364	for oInGroup in oTestBoxDataEx.aoInSchedGroups:
1365	oSchedGroup = oInGroup.oSchedGroup;
1366	if oSchedGroup.fEnabled and oSchedGroup.idBuildSrc is not None:
1367	iSchedPriority = oInGroup.iSchedPriority;
1368	if iSchedPriority > 31: # paranoia
1369	iSchedPriority = 31;
1370	elif iSchedPriority < 0: # paranoia
1371	iSchedPriority = 0;
1372
1373	for iSchedPriority in xrange(min(iSchedPriority, len(aaoList1))):
1374	aaoList1[iSchedPriority].append(oSchedGroup);
1375	while len(aaoList1) <= iSchedPriority:
1376	aaoList1.append([oSchedGroup,]);
1377
1378	# Flatten it into a single list, mixing the priorities a little so it doesn't
1379	# take forever before low priority stuff is executed.
1380	aoFlat = [];
1381	iLo = 0;
1382	iHi = len(aaoList1) - 1;
1383	while iHi >= iLo:
1384	aoFlat += aaoList1[iHi];
1385	if iLo < iHi:
1386	aoFlat += aaoList1[iLo];
1387	iLo += 1;
1388	iHi -= 1;
1389
1390	# Pick the next one.
1391	cLeft = len(aoFlat);
1392	while cLeft > 0:
1393	cLeft -= 1;
1394	iWorkItem += 1;
1395	if iWorkItem >= len(aoFlat) or iWorkItem < 0:
1396	iWorkItem = 0;
1397	if aoFlat[iWorkItem].idSchedGroup not in dIgnoreSchedGroupIds:
1398	return (aoFlat[iWorkItem], iWorkItem);
1399	else:
1400	iWorkItem = 0;
1401
1402	# No active group.
1403	return (None, iWorkItem);
1404
1405	@staticmethod
1406	def scheduleNewTask(oDb, oTestBoxData, iWorkItem, sBaseUrl, iVerbosity = 0):
1407	# type: (TMDatabaseConnection, TestBoxData, int, str, int) -> None
1408	"""
1409	Schedules a new task for a testbox.
1410	"""
1411	oTBStatusLogic = TestBoxStatusLogic(oDb);
1412
1413	try:
1414	#
1415	# To avoid concurrency issues in SchedQueues we lock all the rows
1416	# related to our scheduling queue. Also, since this is a very
1417	# expensive operation we lock the testbox status row to fend of
1418	# repeated retires by faulty testbox scripts.
1419	#
1420	tsSecStart = utils.timestampSecond();
1421	oDb.rollback();
1422	oDb.begin();
1423	oDb.execute('SELECT idTestBox FROM TestBoxStatuses WHERE idTestBox = %s FOR UPDATE NOWAIT'
1424	% (oTestBoxData.idTestBox,));
1425	oDb.execute('SELECT SchedQueues.idSchedGroup\n'
1426	' FROM SchedQueues, TestBoxesInSchedGroups\n'
1427	'WHERE TestBoxesInSchedGroups.idTestBox = %s\n'
1428	' AND TestBoxesInSchedGroups.tsExpire = \'infinity\'::TIMESTAMP\n'
1429	' AND TestBoxesInSchedGroups.idSchedGroup = SchedQueues.idSchedGroup\n'
1430	' FOR UPDATE'
1431	% (oTestBoxData.idTestBox,));
1432
1433	# We need the current timestamp.
1434	tsNow = oDb.getCurrentTimestamp();
1435
1436	# Re-read the testbox data with scheduling group relations.
1437	oTestBoxDataEx = TestBoxDataEx().initFromDbWithId(oDb, oTestBoxData.idTestBox, tsNow);
1438	if oTestBoxDataEx.fEnabled \
1439	and oTestBoxDataEx.idGenTestBox == oTestBoxData.idGenTestBox:
1440
1441	# We may have to skip scheduling groups that are out of work (e.g. 'No build').
1442	iInitialWorkItem = iWorkItem;
1443	dIgnoreSchedGroupIds = {};
1444	while True:
1445	# Now, pick the scheduling group.
1446	(oSchedGroup, iWorkItem) = SchedulerBase._pickSchedGroup(oTestBoxDataEx, iWorkItem, dIgnoreSchedGroupIds);
1447	if oSchedGroup is None:
1448	break;
1449	assert oSchedGroup.fEnabled and oSchedGroup.idBuildSrc is not None;
1450
1451	# Instantiate the specified scheduler and let it do the rest.
1452	oScheduler = SchedulerBase._instantiate(oDb, oSchedGroup, iVerbosity, tsSecStart);
1453	dResponse = oScheduler.scheduleNewTaskWorker(oTestBoxDataEx, tsNow, sBaseUrl);
1454	if dResponse is not None:
1455	oTBStatusLogic.updateWorkItem(oTestBoxDataEx.idTestBox, iWorkItem);
1456	oDb.commit();
1457	return dResponse;
1458
1459	# Check out the next work item?
1460	if oScheduler.getElapsedSecs() > config.g_kcSecMaxNewTask:
1461	break;
1462	dIgnoreSchedGroupIds[oSchedGroup.idSchedGroup] = oSchedGroup;
1463
1464	# No luck, but best if we update the work item if we've made progress.
1465	# Note! In case of a config.g_kcSecMaxNewTask timeout, this may accidentally skip
1466	# a work item with actually work to do. But that's a small price to pay.
1467	if iWorkItem != iInitialWorkItem:
1468	oTBStatusLogic.updateWorkItem(oTestBoxDataEx.idTestBox, iWorkItem);
1469	oDb.commit();
1470	return None;
1471	except:
1472	oDb.rollback();
1473	raise;
1474
1475	# Not enabled, rollback and return no task.
1476	oDb.rollback();
1477	return None;
1478
1479	@staticmethod
1480	def tryCancelGangGathering(oDb, oStatusData):
1481	"""
1482	Try canceling a gang gathering.
1483
1484	Returns True if successfully cancelled.
1485	Returns False if not (someone raced us to the SchedQueue table).
1486
1487	Note! oStatusData is re-initialized.
1488	"""
1489	assert oStatusData.enmState == TestBoxStatusData.ksTestBoxState_GangGathering;
1490	try:
1491	#
1492	# Lock the tables we're updating so we don't run into concurrency
1493	# issues (we're racing both scheduleNewTask and other callers of
1494	# this method).
1495	#
1496	oDb.rollback();
1497	oDb.begin();
1498	oDb.execute('LOCK TABLE TestBoxStatuses, SchedQueues IN EXCLUSIVE MODE');
1499
1500	#
1501	# Re-read the testbox data and check that we're still in the same state.
1502	#
1503	oStatusData.initFromDbWithId(oDb, oStatusData.idTestBox);
1504	if oStatusData.enmState == TestBoxStatusData.ksTestBoxState_GangGathering:
1505	#
1506	# Get the leader thru the test set and change the state of the whole gang.
1507	#
1508	oTestSetData = TestSetData().initFromDbWithId(oDb, oStatusData.idTestSet);
1509
1510	oTBStatusLogic = TestBoxStatusLogic(oDb);
1511	oTBStatusLogic.updateGangStatus(oTestSetData.idTestSetGangLeader,
1512	TestBoxStatusData.ksTestBoxState_GangGatheringTimedOut,
1513	fCommit = False);
1514
1515	#
1516	# Move the scheduling queue item to the end.
1517	#
1518	oDb.execute('SELECT *\n'
1519	'FROM SchedQueues\n'
1520	'WHERE idTestSetGangLeader = %s\n'
1521	, (oTestSetData.idTestSetGangLeader,) );
1522	oTask = SchedQueueData().initFromDbRow(oDb.fetchOne());
1523	oTestEx = TestCaseArgsDataEx().initFromDbWithGenIdEx(oDb, oTask.idGenTestCaseArgs,
1524	tsConfigEff = oTask.tsConfig,
1525	tsRsrcEff = oTask.tsConfig);
1526	oDb.execute('UPDATE SchedQueues\n'
1527	' SET idItem = NEXTVAL(\'SchedQueueItemIdSeq\'),\n'
1528	' idTestSetGangLeader = NULL,\n'
1529	' cMissingGangMembers = %s\n'
1530	'WHERE idItem = %s\n'
1531	, (oTestEx.cGangMembers, oTask.idItem,) );
1532
1533	oDb.commit();
1534	return True;
1535
1536	if oStatusData.enmState == TestBoxStatusData.ksTestBoxState_GangGatheringTimedOut:
1537	oDb.rollback();
1538	return True;
1539	except:
1540	oDb.rollback();
1541	raise;
1542
1543	# Not enabled, rollback and return no task.
1544	oDb.rollback();
1545	return False;
1546
1547
1548	#
1549	# Unit testing.
1550	#
1551
1552	# pylint: disable=missing-docstring
1553	class SchedQueueDataTestCase(ModelDataBaseTestCase):
1554	def setUp(self):
1555	self.aoSamples = [SchedQueueData(),];
1556
1557	if __name__ == '__main__':
1558	unittest.main();
1559	# not reached.
1560

Note: See TracBrowser for help on using the repository browser.

source: vbox/trunk/src/VBox/ValidationKit/testmanager/core/schedulerbase.py@ 90802

Download in other formats: