nl.nikhef.pdp.dynsched/trunk/EstTT.py

#--
# EstTT.py -- class for ESTimating Traversal Times via various
# algorithms.  Use "strategy" design pattern to handle alg
# framework.
# J. A. Templon, NIKHEF, PDP Group
# $Id$
# Source: $URL$

#--

# the strategy interface (ie generic algorithm interface):
class FindEstimate(object):
    def algorithm(self, lrms, vo, debug):
        """
        lrms == concrete instance of class lrms (snapshot of LRMS state)
        vo   == virtual organization for which you want estimate
                note: so far this usually maps to unix group of mapped user
        debug: 0 is don't debug, 1 is to print debugging info
        """
        pass

# put concrete algorithm instances here

class Gott(FindEstimate):
    """The original Gott ideas are embodied here.  If you want to
    use the old scheme of looking at jobs matching both the VO
    AND queue like the old system did, do a call like
       Estimator.strategy.queue = 'atlas'
    before actually using the Estimator.estimate() method.
    """
    def __init__(self):
        FindEstimate.__init__(self)
        self.queue = ''
    def algorithm(self, lrms, vo, debug):
        return ett(lrms, self.queue, vo, algorithm='average',debug=debug)

class WaitTime(FindEstimate):
    """This is Gott modified to use longest-wait times instead
    of average wait times.  If you want to
    use the old scheme of looking at jobs matching both the VO
    AND queue like the old system did, do a call like
       Estimator.strategy.queue = 'atlas'
    before actually using the Estimator.estimate() method.
    """
    def __init__(self):
        FindEstimate.__init__(self)
        self.queue = ''
    def algorithm(self, lrms, vo, debug):
        return ett(lrms, self.queue, vo, algorithm='longest',debug=debug)

class Throughput(FindEstimate):
    """This class is based on looking at job throughputs and calculating
    how long it will take to empty all waiting jobs for the desired VO
    based on the observe througput.  Motivation is that Gott & WaitTime
    do very poorly when the job population is not stable, ie when the
    rates of job arrival and job starts are not roughly the same.
    """
    def __init__(self):
        FindEstimate.__init__(self)
        self.twin = 10
        """
        Tunable parameter specifying how far back (how many jobs) in the system
        start history to look when calculating job throughput.  IE if twin
        (Througput WINdow) is 4, code will look at no more than the last four
        jobs started to see how many jobs per second are being started.
        """
    def algorithm(self, lrms, vo, debug):

        # get rid of the silliest case first

        if lrms.slotsUp < 1:  # system has no job slots at all
            return 7777777

        # gather a couple lists to help us decide

        def allrunfunc(job):
            return job.get('state') == 'running'
        def vowaitfunc(job):
            return job.get('state') == 'queued' and job.get('group') == vo
        allRunningJobs  = lrms.matchingJobs(allrunfunc)
        waitingJobsMyVO = lrms.matchingJobs(vowaitfunc)

        # another very silly case; this VO has waiting jobs, but absolutely no
        # jobs are running.  This is either a short transient that will only last
        # for one sched cycle, or else something is wrong.  Assume the latter.
        
        if len(allRunningJobs) < 1 and len(waitingJobsMyVO) > 0 :
            return 7777777
        
        # another silly case: someone just restarted the server.
        # symptom (at least on PBS): no free slots, lots of running
        # jobs, many (or all) of which have no 'start' record.

        # note this is not the only symptom ... there may be free
        # slots but still no jobs with historical info.  Some code
        # later on to handle this in most_recent_run ...

        if lrms.slotsFree == 0:
            nr = len(allRunningJobs)
            nns = 0                     # Number with No Start
            for j in allRunningJobs:
                if j.get('start') == None: nns = nns + 1
            if float(nns)/nr >= 0.5:
                return 7777777
            
        nw = len(waitingJobsMyVO)
        if nw < 0:
            print "found negative number of queued jobs!" ; sys.exit(1)
        elif nw == 0:
            if lrms.slotsFree > 0:
                return lrms.schedCycle / 2
            else:
                if len(allRunningJobs) == 0:
                    return lrms.schedCycle / 2
                else:
                    most_recent = allRunningJobs[0]
                    startList = [ ]
                    for j in allRunningJobs:
                        tval = j.get('start')
                        if tval:  # jobs started during last poll cycle won't have 'start'
                            startList.append(tval)
                        else:
                            startList.append(lrms.schedCycle / 2)  # rough approx
                    startList.sort()  # last element will be most recent start
                    time_sys_full = lrms.now - startList[-1]
                    return time_sys_full + lrms.schedCycle/2
        else: # there are waiting jobs

            # first check one special case: the one where *all* waiting jobs
            # have been in the system for less than the schedule cycle
            # if there are enough slots to run them all, skip the throughput
            # calculation.

            waitList = [ ]
            for j in waitingJobsMyVO:
                waitList.append(lrms.now - j.get('qtime'))
            waitList.sort()
            if waitList[-1] < 1.1*lrms.schedCycle \
                   and len(waitList) < lrms.slotsFree:
                return lrms.schedCycle / 2

            # calcs based on throughput.  we know since we checked above that at
            # least *some* jobs are running.  Use jobs from our VO if we have
            # them, otherwise use list of all running jobs.
            
            def vorunfunc(job):
                return job.get('state') == 'running' and job.get('group') == vo
            runningJobsMyVO = lrms.matchingJobs(vorunfunc)
            if len(runningJobsMyVO) > 0:
                runlist = runningJobsMyVO
            else:
                runlist = allRunningJobs
            startList = [ ]
            for j in runlist:
                st = j.get('start')
                if st :
                    startList.append(st)
                else:
                    startList.append(lrms.now - lrms.schedCycle/2)
            startList.sort()

            # decide how many jobs to use to calc throughput; take minimum
            # of number running, number waiting, and the 'throughput window'
            # that can be set on the algorithm.  Note we have to use the
            # startList instead of runlist since there may be starts for
            # which we have no records yet (PBS caching).
            
            num = min(nw,len(startList),self.twin)
            period = lrms.now - startList[-num]
            thruput = float(num)/period   # jobs per sec

            if len(runningJobsMyVO) > 0:
                return int((nw+1) / thruput)
            else:
                # jobs are running but not yours, so assume you need to finish all
                # waiting jobs before your new one starts
                naw = len(lrms.jobs()) - len(runlist) # counts all waiting jobs
                return int((naw+1) / thruput)

# the "context" controls the strategy:
class TTEstimator(object):
    def __init__(self, strategy):
        self.strategy = strategy
    def estimate(self, lrms, vo, debug):
        return self.strategy.algorithm(lrms, vo, debug)

# there are several algorithms for what to do with queued jobs.
# define them here and provide means to select between them.

def ett_longest_queue_time(server,list) :
    est = 7777777
    qtlist = [ ]
    
    for j in list:
        if j.get('state') == 'queued' :
            qtlist.append(server.now - j.get('qtime'))
    qtlist.sort()
    est = qtlist[-1]   # sorted list so this is the longest waiting.

    return est

def ett_avg_queue_time(server,list) :
    est = 7777777
    # find average time in queue of all queued jobs
    count = 0
    sum = 0.0
    for j in list:
        if j.get('state') == 'queued' :
            count = count + 1
            sum = sum + (server.now - j.get('qtime'))
    avgtime = sum / count
    # avg time in queue likely to be half of expected time in queue
    # => return twice avg as estimate.
    est = int(2*avgtime)

    return est

# make dict of functions to allow run-time selection
_ALGS = {
    'average' : ett_avg_queue_time,
    'longest' : ett_longest_queue_time }

# adjusted_ett provides a central place to make any changes to
# return values; right now it just puts a lower limit of 5 on
# the lower limit.  Could be used e.g. to change units.
def adjusted_ett(rawval,period):
    if rawval < period:
        return int(period / 2.)
    else:
        return int(rawval)

def ett(server,queue,voname='',algorithm='longest',debug=0) :

    if debug: print 'running for VO', voname

    jfilt = {'state' : 'queued'}
    
    if queue != '':              jfilt['queue'] = queue
    if voname not in ['', None]: jfilt['group'] = voname

    jobsrunning = {'state': 'running'}
    
    nq = server.nmatch(jfilt)
    if debug: print 'found', nq, 'queued jobs in queue \'' + queue + \
       '\' for VO', voname
    if nq > 0: # there are queued jobs for this VO
        est = _ALGS[algorithm](server,server.jobs_last_query())
    elif server.slotsFree > 0 :
        if debug: print 'found', server.slotsFree, 'free job slots'
        est = 0
    else :
        server.nmatch(jobsrunning) # get all running jobs
        est = ett_most_recent_run(server.now,
                                      server.jobs_last_query(),
                                      debug)

    return adjusted_ett(est,server.schedCycle)

def ett_most_recent_run(now,joblist,debug=0) :

    # find most recently run job
    # make list of start times
    
    starts = list()
    for j in joblist:
        st = j.get('start')
        if st :
            starts.append(st)

    if len(starts) > 0:
        starts.sort()
        last_started = starts[-1]  # last in list - start with largest timestamp
        if debug : print now, last_started
        timefilled = now - last_started
        if debug : print 'most recent run', timefilled
        return timefilled
    else:
        return 777777   # error code for 'i don't know'
1	templon	1999	#--
2			# EstTT.py -- class for ESTimating Traversal Times via various
3			# algorithms. Use "strategy" design pattern to handle alg
4			# framework.
5			# J. A. Templon, NIKHEF, PDP Group
6	templon	2007	# $Id$
7			# Source: $URL$
8
9	templon	1999	#--
10
11			# the strategy interface (ie generic algorithm interface):
12			class FindEstimate(object):
13			def algorithm(self, lrms, vo, debug):
14			"""
15			lrms == concrete instance of class lrms (snapshot of LRMS state)
16			vo == virtual organization for which you want estimate
17			note: so far this usually maps to unix group of mapped user
18			debug: 0 is don't debug, 1 is to print debugging info
19			"""
20			pass
21
22			# put concrete algorithm instances here
23
24			class Gott(FindEstimate):
25			"""The original Gott ideas are embodied here. If you want to
26			use the old scheme of looking at jobs matching both the VO
27			AND queue like the old system did, do a call like
28			Estimator.strategy.queue = 'atlas'
29			before actually using the Estimator.estimate() method.
30			"""
31			def __init__(self):
32			FindEstimate.__init__(self)
33			self.queue = ''
34			def algorithm(self, lrms, vo, debug):
35			return ett(lrms, self.queue, vo, algorithm='average',debug=debug)
36
37			class WaitTime(FindEstimate):
38			"""This is Gott modified to use longest-wait times instead
39			of average wait times. If you want to
40			use the old scheme of looking at jobs matching both the VO
41			AND queue like the old system did, do a call like
42			Estimator.strategy.queue = 'atlas'
43			before actually using the Estimator.estimate() method.
44			"""
45			def __init__(self):
46			FindEstimate.__init__(self)
47			self.queue = ''
48			def algorithm(self, lrms, vo, debug):
49			return ett(lrms, self.queue, vo, algorithm='longest',debug=debug)
50
51			class Throughput(FindEstimate):
52			"""This class is based on looking at job throughputs and calculating
53			how long it will take to empty all waiting jobs for the desired VO
54			based on the observe througput. Motivation is that Gott & WaitTime
55			do very poorly when the job population is not stable, ie when the
56			rates of job arrival and job starts are not roughly the same.
57			"""
58			def __init__(self):
59			FindEstimate.__init__(self)
60			self.twin = 10
61			"""
62			Tunable parameter specifying how far back (how many jobs) in the system
63			start history to look when calculating job throughput. IE if twin
64			(Througput WINdow) is 4, code will look at no more than the last four
65			jobs started to see how many jobs per second are being started.
66			"""
67			def algorithm(self, lrms, vo, debug):
68
69			# get rid of the silliest case first
70
71			if lrms.slotsUp < 1: # system has no job slots at all
72			return 7777777
73
74			# gather a couple lists to help us decide
75
76			def allrunfunc(job):
77			return job.get('state') == 'running'
78			def vowaitfunc(job):
79			return job.get('state') == 'queued' and job.get('group') == vo
80			allRunningJobs = lrms.matchingJobs(allrunfunc)
81			waitingJobsMyVO = lrms.matchingJobs(vowaitfunc)
82
83			# another very silly case; this VO has waiting jobs, but absolutely no
84			# jobs are running. This is either a short transient that will only last
85			# for one sched cycle, or else something is wrong. Assume the latter.
86
87			if len(allRunningJobs) < 1 and len(waitingJobsMyVO) > 0 :
88			return 7777777
89
90			# another silly case: someone just restarted the server.
91			# symptom (at least on PBS): no free slots, lots of running
92			# jobs, many (or all) of which have no 'start' record.
93
94			# note this is not the only symptom ... there may be free
95			# slots but still no jobs with historical info. Some code
96			# later on to handle this in most_recent_run ...
97
98			if lrms.slotsFree == 0:
99			nr = len(allRunningJobs)
100			nns = 0 # Number with No Start
101			for j in allRunningJobs:
102			if j.get('start') == None: nns = nns + 1
103			if float(nns)/nr >= 0.5:
104			return 7777777
105
106			nw = len(waitingJobsMyVO)
107			if nw < 0:
108			print "found negative number of queued jobs!" ; sys.exit(1)
109			elif nw == 0:
110			if lrms.slotsFree > 0:
111			return lrms.schedCycle / 2
112			else:
113			if len(allRunningJobs) == 0:
114			return lrms.schedCycle / 2
115			else:
116			most_recent = allRunningJobs[0]
117			startList = [ ]
118			for j in allRunningJobs:
119			tval = j.get('start')
120			if tval: # jobs started during last poll cycle won't have 'start'
121			startList.append(tval)
122			else:
123			startList.append(lrms.schedCycle / 2) # rough approx
124			startList.sort() # last element will be most recent start
125			time_sys_full = lrms.now - startList[-1]
126			return time_sys_full + lrms.schedCycle/2
127			else: # there are waiting jobs
128
129			# first check one special case: the one where all waiting jobs
130			# have been in the system for less than the schedule cycle
131			# if there are enough slots to run them all, skip the throughput
132			# calculation.
133
134			waitList = [ ]
135			for j in waitingJobsMyVO:
136			waitList.append(lrms.now - j.get('qtime'))
137			waitList.sort()
138			if waitList[-1] < 1.1*lrms.schedCycle \
139			and len(waitList) < lrms.slotsFree:
140			return lrms.schedCycle / 2
141
142			# calcs based on throughput. we know since we checked above that at
143			# least some jobs are running. Use jobs from our VO if we have
144			# them, otherwise use list of all running jobs.
145
146			def vorunfunc(job):
147			return job.get('state') == 'running' and job.get('group') == vo
148			runningJobsMyVO = lrms.matchingJobs(vorunfunc)
149			if len(runningJobsMyVO) > 0:
150			runlist = runningJobsMyVO
151			else:
152			runlist = allRunningJobs
153			startList = [ ]
154			for j in runlist:
155			st = j.get('start')
156			if st :
157			startList.append(st)
158			else:
159			startList.append(lrms.now - lrms.schedCycle/2)
160			startList.sort()
161
162			# decide how many jobs to use to calc throughput; take minimum
163			# of number running, number waiting, and the 'throughput window'
164			# that can be set on the algorithm. Note we have to use the
165			# startList instead of runlist since there may be starts for
166			# which we have no records yet (PBS caching).
167
168			num = min(nw,len(startList),self.twin)
169			period = lrms.now - startList[-num]
170			thruput = float(num)/period # jobs per sec
171
172			if len(runningJobsMyVO) > 0:
173			return int((nw+1) / thruput)
174			else:
175			# jobs are running but not yours, so assume you need to finish all
176			# waiting jobs before your new one starts
177			naw = len(lrms.jobs()) - len(runlist) # counts all waiting jobs
178			return int((naw+1) / thruput)
179
180			# the "context" controls the strategy:
181			class TTEstimator(object):
182			def __init__(self, strategy):
183			self.strategy = strategy
184			def estimate(self, lrms, vo, debug):
185			return self.strategy.algorithm(lrms, vo, debug)
186
187			# there are several algorithms for what to do with queued jobs.
188			# define them here and provide means to select between them.
189
190			def ett_longest_queue_time(server,list) :
191			est = 7777777
192			qtlist = [ ]
193
194			for j in list:
195			if j.get('state') == 'queued' :
196			qtlist.append(server.now - j.get('qtime'))
197			qtlist.sort()
198			est = qtlist[-1] # sorted list so this is the longest waiting.
199
200			return est
201
202			def ett_avg_queue_time(server,list) :
203			est = 7777777
204			# find average time in queue of all queued jobs
205			count = 0
206			sum = 0.0
207			for j in list:
208			if j.get('state') == 'queued' :
209			count = count + 1
210			sum = sum + (server.now - j.get('qtime'))
211			avgtime = sum / count
212			# avg time in queue likely to be half of expected time in queue
213			# => return twice avg as estimate.
214			est = int(2*avgtime)
215
216			return est
217
218			# make dict of functions to allow run-time selection
219			_ALGS = {
220			'average' : ett_avg_queue_time,
221			'longest' : ett_longest_queue_time }
222
223			# adjusted_ett provides a central place to make any changes to
224			# return values; right now it just puts a lower limit of 5 on
225			# the lower limit. Could be used e.g. to change units.
226			def adjusted_ett(rawval,period):
227			if rawval < period:
228			return int(period / 2.)
229			else:
230			return int(rawval)
231
232			def ett(server,queue,voname='',algorithm='longest',debug=0) :
233
234			if debug: print 'running for VO', voname
235
236			jfilt = {'state' : 'queued'}
237
238			if queue != '': jfilt['queue'] = queue
239			if voname not in ['', None]: jfilt['group'] = voname
240
241			jobsrunning = {'state': 'running'}
242
243			nq = server.nmatch(jfilt)
244			if debug: print 'found', nq, 'queued jobs in queue \'' + queue + \
245			'\' for VO', voname
246			if nq > 0: # there are queued jobs for this VO
247			est = _ALGS[algorithm](server,server.jobs_last_query())
248			elif server.slotsFree > 0 :
249			if debug: print 'found', server.slotsFree, 'free job slots'
250			est = 0
251			else :
252			server.nmatch(jobsrunning) # get all running jobs
253			est = ett_most_recent_run(server.now,
254			server.jobs_last_query(),
255			debug)
256
257			return adjusted_ett(est,server.schedCycle)
258
259			def ett_most_recent_run(now,joblist,debug=0) :
260
261			# find most recently run job
262			# make list of start times
263
264			starts = list()
265			for j in joblist:
266			st = j.get('start')
267			if st :
268			starts.append(st)
269
270			if len(starts) > 0:
271			starts.sort()
272			last_started = starts[-1] # last in list - start with largest timestamp
273			if debug : print now, last_started
274			timefilled = now - last_started
275			if debug : print 'most recent run', timefilled
276			return timefilled
277			else:
278			return 777777 # error code for 'i don't know'
grid.support@nikhef.nl	ViewVC Help
Powered by ViewVC 1.1.28