cqm.py

#!/usr/bin/env python
# $Id$

'''Cobalt Queue Manager'''

#
# TODO:
#
# - modify progress routine to catch exceptions and report them using _sm_log_exception.  should some or all exceptions cause
#   the job to be terminated?
#
# - make a pass through the _sm_log calls and decide the correct error levels
#
# Changes:
#
# - (Job) the system_state and user_state fields no longer exist.
#
# - (Job) the admin_hold and user_hold fields are now set to True or False to respective hold or release a job.
#
# - (Job) setting the preemptable field to true specifies that a task associate with the job may be preempted.
#
# - (Job) the is_runnable field is true when the job is able to run (ready or preempted states) and false otherwise.
#
# - (Job) the is_active field is set to true when the job is in any state other than ready (queued), held, preempted or completed
#   (terminal).
#
# - (Job) the has_resources field is set to true if resources are assigned to the job.  this assumes that resources may not be
#   deallocated until after the resource epilogue scripts have been run and thus remains true until those scripts have completed.
#   in reality, the system component may reacquire control of the resources and possibly reassign them as soon as the task
#   terminates.  this may occur even before cqm is notices that the task has ended.  unfortunately, this is the best we can do
#   without acquiring resource state from the system component.
#
# - (Job) the has_completed field is set to true only once the job has finished (reached the terminal state).  a value of true
#   does not imply that the job completed successfully, only that it is not active and may never be run again.
#
# - (Job) the preempts field contains a count of the number of times a task associated with the job has been
#   preempted.  this field is read-only and only exists if the job is preemptable.
#
# - (Job) the maxcptime field specifies the time needed for a task to checkpoint.  if maxcptime is greater than zero, the task
#   will first be signaled to checkpoint (SIGUSR1).  after the specified amount of time elapses, the task will then be signaled
#   to terminate (SIGTERM).  if maxcptime is zero, the task will immediately be signaled to terminate.  (BRT: is this the right
#   thing to do?  how will the job know the difference between being preempted and killed?  is it necessary for it to know the
#   difference no checkpoint time is requested?)
#
# - (Job) the mintasktime field specifies the minimum amount of time a task must run before it may be preempted.  any
#   preemption requests made prior to the specified amount of time elapsing will be recorded and processed once the specified
#   amount of time has elapsed.  notification to checkpoint will be made prior to preemption as indicated by the maxcptime.
#
# - (Job) the maxtasktime field specifies the maximum amount of time a task will be allowed to run before being preempted.
#   after the specified amount of time has elasped the task will be signaled to terminate (SIGTERM).  notification to checkpoint
#   will be made prior to preemption as indicated by maxcptime.
#
# - (Job) setting the walltime field is no longer required (at least by the cqm component).  instead, the maxtasktime field
#   may be used to specify repeated timeslices to be given to the job until it terminates.  (BRT: this is a first stab at
#   timeslicing.  eventually, we may want the scheduler component to have some say in whether a job gets preempted after the
#   maximum amount of time has been reached, allowing a job to continue to run past it's specified timeslice if no resource
#   contention exists.)
#
# - (Job) the 'force_kill_delay' field specifies the time that must elapse before a task being preempted or killed should be
#   forcibly terminated (with SIGKILL).  if the field is not set in the job instance, the value of 'force_kill_delay' in the cqm
#   section of the config file will be used.  if 'force_kill_delay' is not found in the cqm section of the config file, then the
#   module attribute DEFAULT_FORCE_KILL_DELAY is used (see below).
#
# Assumptions:
#
# - (Job/CQM) the system component will return a non-zero exit status if the task was killed by a signal.  'None' is considered
#   to be non-zero and thus would be a valid exit status if the task was terminated.
#


import errno
import logging
import os
import pwd
import grp
import sys
import time
import math
import types
import xmlrpclib
import ConfigParser
import signal
import thread
from threading import Thread, Lock
import traceback
import string


import Cobalt
import Cobalt.Util
from Cobalt.Util import Timer, pickle_data, unpickle_data, disk_writer_thread
import Cobalt.Cqparse
from Cobalt.Data import Data, DataList, DataDict, IncrID
from Cobalt.StateMachine import StateMachine
from Cobalt.Components.base import Component, exposed, automatic, query, locking
from Cobalt.Proxy import ComponentProxy
from Cobalt.Exceptions import (QueueError, ComponentLookupError, DataStateError, DataStateTransitionError, StateMachineError,
    StateMachineIllegalEventError, StateMachineNonexistentEventError, ThreadPickledAliveException, JobProcessingError,
    JobRunError, JobPreemptionError, JobDeleteError, IncrIDError, ResourceReservationFailure)
from Cobalt import accounting
from Cobalt.Statistics import Statistics
from Cobalt.Util import get_config_option, init_cobalt_config

__revision__ = '$Revision$'
init_cobalt_config()

DEFAULT_FORCE_KILL_DELAY = 5  # (in minutes)
CLOB_SIZE = 4096

logger = logging.getLogger(__name__.split('.')[-1])

cqm_id_gen = None
run_id_gen = None #IncrID()

cqm_forker_tag = "cqm_script"
job_prescript_tag = "job prescript"
job_postscript_tag = "job postscript"
resource_postscript_tag = "resource postscript"

config = ConfigParser.ConfigParser()
config.read(Cobalt.CONFIG_FILES)
if not config.has_section('cqm'):
    print '''"cqm" section missing from cobalt config file'''
    sys.exit(1)

def get_cqm_config(option, default):
    try:
        value = config.get('cqm', option)
    except ConfigParser.NoOptionError:
        value = default
    return value

def get_bgsched_config(option, default):
    try:
        value = config.get('bgsched', option)
    except ConfigParser.NoOptionError:
        value = default
    return value

# *AdjEst*
def get_histm_config(option, default):
    try:
        value = config.get('histm', option)
    except ConfigParser.NoSectionError:
        value = default
    return value

CQM_SCALE_DEP_FRAC = str(get_cqm_config('scale_dep_frac', 'false')).lower() in Cobalt.Util.config_true_values

walltime_prediction = get_histm_config("walltime_prediction", "False").lower()   # *AdjEst*
walltime_prediction_configured = False
walltime_prediction_enabled = False
if walltime_prediction  == "true":
    walltime_prediction_configured = True
    walltime_prediction_enabled = True
#print "walltime_prediction_configured=", walltime_prediction_configured 

prediction_scheme = get_histm_config("prediction_scheme", "combined").lower()  # ["project", "user", "combined"]   # *AdjEst*

accounting_logdir = os.path.expandvars(get_cqm_config("log_dir", Cobalt.DEFAULT_LOG_DIRECTORY))
accounting_logger = logging.getLogger("cqm.accounting")
accounting_logger.addHandler(
    accounting.DatetimeFileHandler(os.path.join(accounting_logdir, "%Y%m%d")))


dbwriter = Cobalt.Logging.dbwriter(logging)
use_db_logging = get_cqm_config('use_db_logging','false')
if use_db_logging.lower() in Cobalt.Util.config_true_values:
    dbwriter.enabled = True
    overflow_filename = get_cqm_config('overflow_file', None)
    max_queued = int(get_cqm_config('max_queued_msgs', '-1'))
    if max_queued <= 0:
        max_queued = None
    if (overflow_filename == None) and (max_queued != None):
        logger.warning('No filename set for database logging messages, max_queued_msgs set to unlimited')
    if max_queued != None:
        dbwriter.overflow_filename = overflow_filename
        dbwriter.max_queued = max_queued

#writer so that cobalt log writes don't hang up scheduling.
cobalt_log_writer = disk_writer_thread()
cobalt_log_writer.daemon = True
cobalt_log_writer.start()

def cobalt_log_write(filename, msg):
    '''send the cobalt_log writer thread a filename, msg tuple.

    '''
    cobalt_log_writer.send((filename, msg))

def cobalt_log_terminate():
    '''Terminate the writer thread by sending it None.

    '''
    cobalt_log_writer.send(None)

def str_elapsed_time(elapsed_time):
    return "%d:%02d:%02d" % (elapsed_time / 3600, elapsed_time / 60 % 60, elapsed_time % 60)

def has_private_attr(obj, attr):
    assert attr[0:2] == "__"
    return hasattr(obj, "_" + obj.__class__.__name__ + attr)

def has_semi_private_attr(obj, attr):
    assert attr[0:1] == "_"
    return hasattr(obj, attr)


class Signal_Map (object):
    checkpoint = 'SIGUSR1'
    terminate = 'SIGTERM'
    force_kill = 'SIGKILL'

class Signal_Info (object):
    class Reason (object):
        delete = 'delete'
        preempt = 'preempt'
        time_limit = 'time limit'

    def __init__(self, reason = None, signame = None, user = None, pending = False):
        self.__reason = reason
        self.__signal = signame
        self.__user = user
        self.__pending = pending

    def __get_reason(self):
        return self.__reason

    def __set_reason(self, reason):
        if reason in ('delete', 'preempt', 'time_limit'):
            self.__reason = reason
        else:
            raise ValueError, "illegal signal reason: %s" % (reason,)

    reason = property(__get_reason, __set_reason)

    def __get_signal(self):
        return self.__signal

    def __set_signal(self, signame):
        if signame == None:
            self.__signal = None
        if not isinstance(signame, str):
            raise TypeError, "signal name must be a string"
        if signame[0:3] == "SIG" and hasattr(signal, signame):
            self.__signal = signame
        else:
            raise ValueError, "unknown signal name: %s" % (signame,)

    signal = property(__get_signal, __set_signal)

    def __get_user(self):
        return self.__user

    def __set_user(self, user):
        if isinstance(user, str) or user == None:
            self.__user = user
        else:
            raise TypeError, "user name must be a string or None"

    user = property(__get_user, __set_user)

    def __get_pending(self):
        return self.__pending

    def __set_pending(self, pending):
        if isinstance(pending, bool):
            self.__pending = pending
        else:
            raise TypeError, "pending flag must be a boolean value"

    pending = property(__get_pending, __set_pending)


#TODO: I want to get rid of this.  I don't want to see threaded-forks in 
# cobalt ever again. --PR
class RunScriptsThread(object):
    '''stub for restart compatibility'''
    def __init__(self):
        pass

def get_job_sm_states():

    return ['Ready',
        'Hold',
        'Job_Prologue',
        'Job_Prologue_Retry',
        'Job_Prologue_Retry_Release',
        'Resource_Prologue',
        'Resource_Prologue_Retry',
        'Resource_Prologue_Retry_Release',
        'Run_Retry',
        'Running',
        'Kill_Retry',
        'Killing',
        'Preempt_Retry',
        'Preempting',
        'Preempt_Finalize_Retry',
        'Preempt_Epilogue',
        'Preempted',
        'Preempted_Hold',
        'Finalize_Retry',
        'Resource_Epilogue',
        'Resource_Epilogue_Retry',
        'Job_Epilogue',
        'Job_Epilogue_Retry']

def get_job_sm_transitions():

    return [
        ('Ready', 'Hold'),                                  # user/admin hold
        ('Ready', 'Job_Prologue'),                          # run; start prologue scripts
        ('Ready', 'Job_Prologue_Retry'),                    # run; error contacting forker component; scripts not running, yet
        ('Ready', 'Terminal'),                              # kill
        ('Hold', 'Ready'),                                  # user/admin release, no holds
        ('Hold', 'Terminal'),                               # kill
        ('Job_Prologue', 'Resource_Prologue'),              # job_prologue scripts complete.  Starting Resource_prologue scripts
        ('Job_Prologue', 'Resource_Prologue_Retry'),        # error contacting forker component
        ('Job_Prologue', 'Job_Prologue_Retry'),             # Lost communication to forker during progress
        ('Job_Prologue', 'Job_Epilogue'),                   # kill / job prologue failed
        ('Job_Prologue', 'Job_Epilogue_Retry'),             # kill / job prologue failed.  Error communicating with forker
        ('Job_Prologue', 'Job_Prologue_Retry_Release'),     # kill / job prologue failed.  error releasing system resources
        ('Job_Prologue_Retry', 'Job_Prologue'),             # forker starting job prologue scripts
        ('Job_Prologue_Retry', 'Terminal'),                 # kill; error contacting forker component
        ('Job_Prologue_Retry_Release', 'Job_Epilogue'),     # resource successfully released
        ('Job_Prologue_Retry_Release', 'Job_Epilogue_Retry'),     # error contacting forker component
        ('Resource_Prologue', 'Run_Retry'),                 # prologue scripts complete; error contacting system component
        ('Resource_Prologue', 'Running'),                   # prologue scripts complete; system component starting task
        ('Resource_Prologue', 'Resource_Prologue_Retry'),   # Lost communication to forker during progress
        ('Resource_Prologue', 'Resource_Epilogue'),         # kill / resource prologue failed; system component released resource
        ('Resource_Prologue', 'Resource_Epilogue_Retry'),   # kill / resource prologue failed; error contacting forker component
        ('Resource_Prologue', 'Resource_Prologue_Retry_Release'), # kill / job_prologue failed.  error releasing system resources
        ('Resource_Prologue_Retry','Resource_Prologue'),    # run resource prologue scripts
        ('Resource_Prologue_Retry','Job_Epilogue'),         # kill; run any required job cleanup
        ('Resource_Prologue_Retry_Release', 'Resource_Epilogue'),       # resource successfully released
        ('Resource_Prologue_Retry_Release', 'Resource_Epilogue_Retry'), # error contacting forker component
        ('Run_Retry', 'Running'),                           # system component starting task
        ('Run_Retry', 'Resource_Epilogue'),                 # kill
        ('Run_Retry', 'Resource_Epilogue_Retry'),           # kill; error contacting forker component
        ('Running', 'Kill_Retry'),                          # kill; error contacting system component
        ('Running', 'Killing'),                             # kill; system component signaling task
        ('Running', 'Preempt_Retry'),                       # preempt; error contacting system component
        ('Running', 'Preempting'),                          # preempt; system component signaling task
        ('Running', 'Finalize_Retry'),                      # task execution complete; error finalizing task and obtaining exit
                                                            #     status
        ('Running', 'Resource_Epilogue'),                   # task execution complete; task finalized and exit status obtained
        ('Running', 'Resource_Epilogue_Retry'),             # task executioncomplete; error contacting forker component 
        ('Kill_Retry', 'Kill_Retry'),                       # handle multiple task signaling failures
        ('Kill_Retry', 'Killing'),                          # system component signaling task
        ('Kill_Retry', 'Finalize_Retry'),                   # task execution complete/terminated; task finalization failed
        ('Kill_Retry', 'Resource_Epilogue'),                # task execution completed/terminated successfully
        ('Kill_Retry', 'Resource_Epilogue_Retry'),          # task execution completed/terminated successfully; error contacting
                                                            #     forker component
        ('Killing', 'Kill_Retry'),                          # new signal, or kill timer expired; error contacting system component
        ('Killing', 'Killing'),                             # kill timer expired; escalating to a forced kill
        ('Killing', 'Finalize_Retry'),                      # task execution complete/terminated; task finalization failed
        ('Killing', 'Resource_Epilogue'),                   # task execution complete/terminated; task finalization successful
        ('Killing', 'Resource_Epilogue_Retry'),             # task execution complete/terminated; task finalization successful;
                                                            #   error contacting forker component
        ('Preempt_Retry', 'Kill_Retry'),                    # kill; error contacting system component
        ('Preempt_Retry', 'Killing'),                       # kill; system component signaling task
        ('Preempt_Retry', 'Preempt_Retry'),                 # hasndle multiple task signaling failures
        ('Preempt_Retry', 'Preempting'),                    # system component signaling task
        ('Preempt_Retry', 'Preempt_Finalize_Retry'),        # task execution terminated, task finalization failed
        ('Preempt_Retry', 'Preempt_Epilogue'),              # task execution terminated successfully
        ('Preempt_Retry', 'Finalize_Retry'),                # task execution completed, task finalization failed
        ('Preempt_Retry', 'Resource_Epilogue'),             # task execution completed successfully
        ('Preempt_Retry', 'Resource_Epilogue_Retry'),       # task execution completed; error contacting forker component
        ('Preempting', 'Kill_Retry'),                       # kill; new signal, error contacting system component
        ('Preempting', 'Killing'),                          # kill; new signal and system component signaling task,
                                                            #     same signal used for preempt, or attempted signal demotion
        ('Preempting', 'Preempting'),                       # preemption timer expired, escalating to the next signal level
        ('Preempting', 'Preempt_Retry'),                    # preemption timer expired, error contacting system component
        ('Preempting', 'Preempt_Finalize_Retry'),           # task execution terminated, task finalization failed
        ('Preempting', 'Preempt_Epilogue'),                 # task execution complete/terminated; task finalization successful
        ('Preempt_Finalize_Retry', 'Preempt_Epilogue'),     # task finalization failed successful
        ('Preempt_Epilogue', 'Preempted'),                  # task execution complete/terminated, no holds pending
        ('Preempt_Epilogue', 'Preempted_Hold'),             # task execution complete/terminated, holds pending
        ('Preempt_Epilogue', 'Job_Epilogue'),               # task execution complete/terminated, kill pending
        ('Preempted', 'Resource_Prologue'),                 # run
        ('Preempted', 'Preempted_Hold'),                    # user/admin hold
        ('Preempted', 'Job_Epilogue'),                      # kill
        ('Preempted_Hold', 'Preempted'),                    # user/admin release, no holds
        ('Preempted_Hold', 'Job_Epilogue'),                 # kill
        ('Finalize_Retry', 'Resource_Epilogue'),            # task finalized and exit status obtained (if applicable)
        ('Finalize_Retry', 'Resource_Epilogue_Retry'),      # task finalized, exit status obtained, error contacting forker component
        ('Resource_Epilogue', 'Job_Epilogue'),              # resource epilogue scripts complete
        ('Resource_Epilogue', 'Resource_Epilogue_Retry'),   # Lost communication to forker during progress
        ('Resource_Epilogue', 'Job_Epilogue_Retry'),        # resource epilogue scripts complete; error contacting forker component
        ('Resource_Epilogue_Retry', 'Resource_Epilogue'),   # starting resource epilogue scripts
        ('Job_Epilogue', 'Terminal'),                       # job epilogue scripts complete
        ('Job_Epilogue_Retry', 'Job_Epilogue'),             # starting job_epilogue scripts
        ('Job_Epilogue', 'Job_Epilogue_Retry')              # Lost communication to forker during progress
        ]

def get_job_sm_initial_state():

    return "Ready"

def get_job_sm_events():

    return ['Run', 'Hold', 'Release', 'Preempt', 'Kill', 'Task_End']

def get_job_sm_seas(job):

    return {('Ready', 'Run') : [job._sm_ready__run],
            ('Ready', 'Hold') : [job._sm_ready__hold],
            ('Ready', 'Release') : [job._sm_ready__release],
            ('Ready', 'Kill') : [job._sm_ready__kill],
            ('Hold', 'Hold') : [job._sm_hold__hold],
            ('Hold', 'Release') : [job._sm_hold__release],
            ('Hold', 'Kill') : [job._sm_hold__kill],
            ('Job_Prologue', 'Progress') : [job._sm_job_prologue__progress],
            ('Job_Prologue', 'Hold') : [job._sm_common__pending_hold], 
            ('Job_Prologue', 'Release') : [job._sm_common__pending_release],
            ('Job_Prologue', 'Preempt') : [job._sm_common__pending_preempt], #custom?
            ('Job_Prologue', 'Kill') : [job._sm_common__pending_kill],
            ('Job_Prologue_Retry', 'Progress') : [job._sm_job_prologue_retry__progress],
            ('Job_Prologue_Retry', 'Hold') : [job._sm_common__pending_hold],
            ('Job_Prologue_Retry', 'Release') : [job._sm_common__pending_release],
            ('Job_Prologue_Retry', 'Preempt') : [job._sm_common__pending_preempt], #Must go pending
            ('Job_Prologue_Retry', 'Kill') : [job._sm_job_prologue_retry__kill],
            ('Job_Prologue_Retry_Release', 'Progress') : [job._sm_job_prologue_retry_release__progress],
            ('Job_Prologue_Retry_Release', 'Hold') : [job._sm_exit_common__hold],
            ('Job_Prologue_Retry_Release', 'Release') : [job._sm_exit_common__release],
            ('Job_Prologue_Retry_Release', 'Preempt') : [job._sm_exit_common__preempt],
            ('Job_Prologue_Retry_Release', 'Kill') : [job._sm_exit_common__kill],
            ('Resource_Prologue', 'Progress') : [job._sm_resource_prologue__progress],
            ('Resource_Prologue', 'Hold') : [job._sm_common__pending_hold],
            ('Resource_Prologue', 'Release') : [job._sm_common__pending_release],
            ('Resource_Prologue', 'Preempt') : [job._sm_common__pending_preempt], #custom?
            ('Resource_Prologue', 'Kill') : [job._sm_common__pending_kill],
            ('Resource_Prologue_Retry', 'Progress') : [job._sm_resource_prologue_retry__progress],
            ('Resource_Prologue_Retry', 'Hold') : [job._sm_common__pending_hold],
            ('Resource_Prologue_Retry', 'Release') : [job._sm_common__pending_release],
            ('Resource_Prologue_Retry', 'Preempt') : [job._sm_common__pending_preempt], #custom: go directly to preempted?
            ('Resource_Prologue_Retry', 'Kill') : [job._sm_resource_prologue_retry__kill],
            ('Resource_Prologue_Retry_Release', 'Progress') : [job._sm_resource_prologue_retry_release__progress],
            ('Resource_Prologue_Retry_Release', 'Hold') : [job._sm_exit_common__hold],
            ('Resource_Prologue_Retry_Release', 'Release') : [job._sm_exit_common__release],
            ('Resource_Prologue_Retry_Release', 'Preempt') : [job._sm_exit_common__preempt],
            ('Resource_Prologue_Retry_Release', 'Kill') : [job._sm_exit_common__kill],
            ('Run_Retry', 'Progress') : [job._sm_run_retry__progress],
            ('Run_Retry', 'Hold') : [job._sm_common__pending_hold],
            ('Run_Retry', 'Release') : [job._sm_common__pending_release],
            ('Run_Retry', 'Preempt') : [job._sm_common__pending_preempt],
            ('Run_Retry', 'Kill') : [job._sm_run_retry__kill],
            ('Running', 'Progress') : [job._sm_running__progress],
            ('Running', 'Hold') : [job._sm_common__pending_hold],
            ('Running', 'Release') : [job._sm_common__pending_release],
            ('Running', 'Preempt') : [job._sm_common__pending_preempt],
            ('Running', 'Kill') : [job._sm_running__kill],
            ('Running', 'Task_End') : [job._sm_running__task_end],
            ('Kill_Retry', 'Progress') : [job._sm_kill_retry__progress],
            ('Kill_Retry', 'Hold') : [job._sm_kill_common__hold],
            ('Kill_Retry', 'Release') : [job._sm_kill_common__release],
            ('Kill_Retry', 'Preempt') : [job._sm_kill_common__preempt],
            ('Kill_Retry', 'Kill') : [job._sm_kill_retry__kill],
            ('Kill_Retry', 'Task_End') : [job._sm_kill_retry__task_end],
            ('Killing', 'Progress') : [job._sm_killing__progress],
            ('Killing', 'Hold') : [job._sm_kill_common__hold],
            ('Killing', 'Release') : [job._sm_kill_common__release],
            ('Killing', 'Preempt') : [job._sm_kill_common__preempt],
            ('Killing', 'Kill') : [job._sm_killing__kill],
            ('Killing', 'Task_End') : [job._sm_killing__task_end],
            ('Preempt_Retry', 'Progress') : [job._sm_preempt_retry__progress],
            ('Preempt_Retry', 'Hold') : [job._sm_common__pending_hold],
            ('Preempt_Retry', 'Release') : [job._sm_common__pending_release],
            ('Preempt_Retry', 'Kill') : [job._sm_preempt_retry__kill],
            ('Preempt_Retry', 'Task_End') : [job._sm_preempt_retry__task_end],
            ('Preempting', 'Progress') : [job._sm_preempting__progress],
            ('Preempting', 'Hold') : [job._sm_common__pending_hold],
            ('Preempting', 'Release') : [job._sm_common__pending_release],
            ('Preempting', 'Kill') : [job._sm_preempting__kill],
            ('Preempting', 'Task_End') : [job._sm_preempting__task_end],
            ('Preempt_Finalize_Retry', 'Progress') : [job._sm_preempt_finalize_retry__progress],
            ('Preempt_Finalize_Retry', 'Hold') : [job._sm_common__pending_hold],
            ('Preempt_Finalize_Retry', 'Release') : [job._sm_common__pending_release],
            ('Preempt_Finalize_Retry', 'Kill') : [job._sm_common__pending_kill],
            ('Preempt_Epilogue', 'Progress') : [job._sm_preempt_epilogue__progress],
            ('Preempt_Epilogue', 'Hold') : [job._sm_common__pending_hold],
            ('Preempt_Epilogue', 'Release') : [job._sm_common__pending_release],
            ('Preempt_Epilogue', 'Kill') : [job._sm_common__pending_kill],
            ('Preempted', 'Run') : [job._sm_preempted__run],
            ('Preempted', 'Hold') : [job._sm_preempted__hold],
            ('Preempted', 'Release') : [job._sm_preempted__release],
            ('Preempted', 'Kill') : [job._sm_preempted__kill],
            ('Preempted_Hold', 'Hold') : [job._sm_preempted_hold__hold],
            ('Preempted_Hold', 'Release') : [job._sm_preempted_hold__release],
            ('Preempted_Hold', 'Kill') : [job._sm_preempted_hold__kill],
            ('Finalize_Retry', 'Progress') : [job._sm_finalize_retry__progress],
            ('Finalize_Retry', 'Hold') : [job._sm_exit_common__hold],
            ('Finalize_Retry', 'Release') : [job._sm_exit_common__release],
            ('Finalize_Retry', 'Preempt') : [job._sm_exit_common__preempt],
            ('Finalize_Retry', 'Kill') : [job._sm_exit_common__kill],
            ('Resource_Epilogue', 'Progress') : [job._sm_resource_epilogue__progress],
            ('Resource_Epilogue', 'Hold') : [job._sm_exit_common__hold],
            ('Resource_Epilogue', 'Release') : [job._sm_exit_common__release],
            ('Resource_Epilogue', 'Preempt') : [job._sm_exit_common__preempt],
            ('Resource_Epilogue', 'Kill') : [job._sm_exit_common__kill],
            ('Resource_Epilogue_Retry', 'Progress') : [job._sm_resource_epilogue_retry__progress],
            ('Resource_Epilogue_Retry', 'Hold') : [job._sm_exit_common__hold],
            ('Resource_Epilogue_Retry', 'Release') : [job._sm_exit_common__release],
            ('Resource_Epilogue_Retry', 'Preempt') : [job._sm_exit_common__preempt],
            ('Resource_Epilogue_Retry', 'Kill') : [job._sm_exit_common__kill],
            ('Job_Epilogue', 'Progress') : [job._sm_job_epilogue__progress],
            ('Job_Epilogue', 'Hold') : [job._sm_exit_common__hold],
            ('Job_Epilogue', 'Release') : [job._sm_exit_common__release],
            ('Job_Epilogue', 'Preempt') : [job._sm_exit_common__preempt],
            ('Job_Epilogue', 'Kill') : [job._sm_exit_common__kill],
            ('Job_Epilogue_Retry', 'Progress') : [job._sm_job_epilogue_retry__progress],
            ('Job_Epilogue_Retry', 'Hold') : [job._sm_exit_common__hold],
            ('Job_Epilogue_Retry', 'Release') : [job._sm_exit_common__release],
            ('Job_Epilogue_Retry', 'Preempt') : [job._sm_exit_common__preempt],
            ('Job_Epilogue_Retry', 'Kill') : [job._sm_exit_common__kill],
            }

class Job (StateMachine):
    """
    The job tracks a job, driving it at a high-level.  Actual operations on the job, such as execution or termination, are
    the responsibility of the system component
    """

    acctlog = Cobalt.Util.AccountingLog('qm')

    # properties for easier accounting logging
    ctime = property(lambda self: self.__timers['queue'].start_times[0])
    qtime = property(lambda self:
        self.__timers['current_queue'].start_times[0])
    start = property(lambda self: self.__timers['user'].start_times[-1])
    exec_host = property(lambda self: ":".join(self.__locations[-1]))
    end = property(lambda self: self.__timers['user'].stop_times[-1])

    fields = Data.fields + [
        "jobid", "jobname", "state", "attribute", "location", "starttime", 
        "submittime", "endtime", "queue", "type", "user",
        "walltime", "procs", "nodes", "mode", "cwd", "command", "args", 
        "outputdir", "project", "lienID", "stagein", "stageout",
        "reservation", "host", "port", "url", "stageid", "envs", "inputfile", 
        "kernel", "kerneloptions", "ion_kernel", "ion_kerneloptions", "admin_hold",
        "user_hold", "dependencies", "notify", "adminemail", "outputpath",
        "errorpath", "cobalt_log_file", "path", "preemptable", "preempts",
        "mintasktime", "maxtasktime", "maxcptime", "force_kill_delay", 
        "is_runnable", "is_active",
        "has_completed", "sm_state", "score", "attrs", "has_resources", 
        "exit_status", "dep_frac", "walltime_p", "user_list", "runid",
        "geometry"
    ]

    _states = get_job_sm_states() + StateMachine._states

    _transitions = get_job_sm_transitions()

    _initial_state = get_job_sm_initial_state()
    _events = get_job_sm_events() + StateMachine._events

    # return codes to improve typo detection.  by using the return codes in condition statements, a typo will result in a key
    # error rather than an incorrectly followed path.
    __rc_success = "success"
    __rc_retry = "retry"
    __rc_pg_create = "pg_create"
    __rc_xmlrpc = "xmlrpc"
    __rc_unknown = "unknown"

    def __init__(self, spec):
        self.initializing = True
        seas = get_job_sm_seas(self)

        # StateMachine.__init__(self, spec, seas = seas, terminal_actions = [(self._sm_terminal, {})])
        StateMachine.__init__(self, spec, seas = seas)

        logger.info(str(spec))

        self.jobid = spec.get("jobid")
        self.umask = spec.get("umask")
        self.jobname = spec.get("jobname", "N/A")
        self.attribute = spec.get("attribute", "compute")
        self.starttime = spec.get("starttime", "-1")
        self.submittime = spec.get("submittime", time.time())
        self.endtime = spec.get("endtime", "-1")
        self.type = spec.get("type", "mpish")
        self.user = spec.get("user")
        self.__walltime = int(float(spec.get("walltime", 0)))
        self.walltime_p = spec.get("walltime_p", self.walltime)    #  *AdjEst*
        self.procs = spec.get("procs")
        self.nodes = spec.get("nodes")
        self.cwd = spec.get("cwd")
        self.command = spec.get("command")
        self.args = spec.get("args", [])
        self.project = spec.get("project")
        self.lienID = spec.get("lienID")
        self.stagein = spec.get("stagein") #does nothing
        self.stageout = spec.get("stageout") #like ze goggles, it does nothing
        self.reservation = spec.get("reservation", False) #appears to be defunct.
        self.host = spec.get("host")
        self.port = spec.get("port")
        self.url = spec.get("url")
        self.stageid = spec.get("stageid")
        self.inputfile = spec.get("inputfile")
        self.tag = spec.get("tag", "job")
        self.kernel = spec.get("kernel", get_config_option('bgsystem', 'cn_default_kernel', 'default'))
        self.kerneloptions = spec.get("kerneloptions")
        self.ion_kernel = spec.get("ion_kernel",  get_config_option('bgsystem', 'ion_default_kernel', 'default'))
        self.ion_kerneloptions = spec.get("ion_kerneloptions")
        self.notify = spec.get("notify")
        self.adminemail = spec.get("adminemail")
        self.location = spec.get("location")
        self.__locations = []
        self.outputpath = spec.get("outputpath")
        if self.outputpath and self.jobname == 'N/A':
            jname = self.outputpath.split('/')[-1].split('.output')[0]
            if jname and jname != str(self.jobid):
                self.jobname = jname
        self.outputdir = spec.get("outputdir")
        self.errorpath = spec.get("errorpath")
        self.cobalt_log_file = spec.get("cobalt_log_file")
        if not self.cobalt_log_file:
            self.cobalt_log_file = "%s/%s.cobaltlog" % (self.outputdir, self.jobid)
        else:
            t = string.Template(self.cobalt_log_file)
            self.cobalt_log_file = t.safe_substitute(jobid=self.jobid)
        self.path = spec.get("path")
        self.mode = spec.get("mode", "co")
        self.envs = spec.get("envs", {})
        self.force_kill_delay = spec.get("force_kill_delay",
                get_cqm_config('force_kill_delay', DEFAULT_FORCE_KILL_DELAY))
        self.attrs = spec.get("attrs", {})

        self.score = float(spec.get("score", 0.0))

        self.__resource_nodects = []
        self.__timers = dict(
            queue = Timer(),
            current_queue = Timer(),
            user = Timer(),
        )

        # setting the queue will cause updated accounting records to be
        #written and the current queue timer to be restarted, so
        # this needs to be done only after the object has been initialized
        self.queue = spec.get("queue", "default")
        self.resid = None #Must be obtained from the scheduler component.
        self.__timers['queue'].start()
        self.etime = time.time()

        self.__admin_hold = False
        self.__user_hold = False
        self.__dep_hold = False

        # setting the hold flags will automatically cause the appropriate hold
        #events to be triggered, so this needs to be done
        # only after the object has been completely initialized
        if spec.get("admin_hold", False):
            self.admin_hold = True
        if spec.get("user_hold", False):
            self.user_hold = True

        self.dep_fail = False
        self.dep_frac = None #float(get_cqm_config('dep_frac', 0.5))
        self.all_dependencies = spec.get("all_dependencies")
        self.satisfied_dependencies = []
        if self.all_dependencies:
            self.all_dependencies = str(self.all_dependencies).split(":")
            logger.info("Job %s/%s: dependencies set to %s", self.jobid, self.user, ":".join(self.all_dependencies))
            # read passed dependencies and set dep_hold if necessary
            # set dep_hold as needed:
            self.update_dep_state()
        else:
            self.all_dependencies = []


        self.preemptable = spec.get("preemptable", False)
        self.__preempts = 0
        if self.preemptable:
            self.mintasktime = int(float(spec.get("mintasktime", 0)))
            self.maxtasktime = int(float(spec.get("maxtasktime", 0)))
            self.maxcptime = int(float(spec.get("maxcptime", 0)))

        self.geometry = spec.get("geometry", None)

        self.taskid = None
        self.task_running = False
        self.exit_status = None
        self.max_running = False

        self.total_etime = 0.0
        self.priority_core_hours = None
        self.user_list = spec.get('user_list', [self.user])

        #for imporved script handling:
        self.job_prescript_ids = []
        self.job_postscript_ids = []
        self.resource_prescript_ids = []
        self.resource_postscript_ids = []

        self.runid = None #A job starts once.
        #prebooting is assumed for script jobs, unless the user requests that it be turned off
        #This is here for BG/Q ensemble job support.  --PMR
        self.script_preboot = spec.get("script_preboot", True) #preboot a block for a script.


        dbwriter.log_to_db(self.user, "creating", "job_data", JobDataMsg(self))
        if self.admin_hold:
            dbwriter.log_to_db(self.user, "admin_hold", "job_prog", JobProgMsg(self))
        if self.user_hold:
            dbwriter.log_to_db(self.user, "user_hold", "job_prog", JobProgMsg(self))

        self.current_task_start = time.time()

        self.initializing = False

    # end def __init__()

    def no_holds_left(self):
        '''Check for whether any holds are set on a job'''
        return not (self.admin_hold or 
                self.user_hold or 
                self.dep_hold or 
                self.max_running)

    def __getstate__(self):
        data = {}
        for key, value in self.__dict__.iteritems():
            if key not in ['log', 'comms', 'acctlog']:
                data[key] = value
        return data

    def __setstate__(self, state):
        self.__dict__.update(state)

        #reset the statemachine's states
        self.update_seas(get_job_sm_seas(self))

        # BRT: why is the current queue timer being reset?  if cqm is 
        #restarted, the job remained in the queue during that time, so I would
        #think that timer should continue to run during the restart rather 
        #than been reset.
        if not self.__timers.has_key('current_queue'):
            self.__timers['current_queue'] = Timer()
            self.__timers['current_queue'].start()

        # special case to handle missing data from old state files
        if not state.has_key("total_etime"):
            logger.info("old job missing total_etime")
            self.total_etime = 0.0

        if not state.has_key("priority_core_hours"):
            logger.info("old job missing priority_core_hours")
            self.priority_core_hours = None

        if not state.has_key("dep_fail"):
            logger.info("old job missing dep_fail")
            self.dep_fail = False

        if not state.has_key("dep_frac"):
            logger.info("old job missing dep_frac")
            self.dep_frac = None

        if not state.has_key("user_list"):
            logger.info("old job missing user_list")
            self.user_list = [self.user]

        if not state.has_key("walltime_p"):
            logger.info("old job missing walltime_p")
            self.walltime_p = self.walltime

        if not state.has_key("resid"):
            logger.info("old job missing resid")
            self.resid = None

        if not state.has_key("job_prologue_ids"):
            self.job_prologue_ids = None
        if not state.has_key("job_epilogue_ids"):
            self.job_epilogue_ids = None
        if not state.has_key("resource_prologue_ids"):
            self.resource_prologue_ids = None
        if not state.has_key("resource_epilogue_ids"):
            self.resource_epilogue_ids = None
        if not state.has_key("geometry"):
            self.geometry = None
        if not state.has_key("script_preboot"):
            self.script_preboot = True
        if not state.has_key('ion_kernel'):
            #if ion kernel doesn't exist, neither will ion_kerneloptions
            self.ion_kernel = get_config_option('bgsystem', 'ion_default_kernel', 'default')
            self.ion_kerneloptions = get_config_option('bgsystem', 'ion_default_kernel_options', False)
            if self.ion_kerneloptions == False:
                self.ion_kerneloptions = None
        self.runid = state.get("runid", None)
        self.current_task_start = state.get("current_task_start", None)
        #for old statefiles, make sure to update the dependency state on restart:
        self.__dep_hold = False
        self.initializing = False

    def __task_signal(self, retry = True):
        '''send a signal to the managed task'''
        # BRT: this routine should probably check if the task could not be 
        #signaled because it was no longer running
        try:
            self._sm_log_info("instructing the system component to send signal %s" % (self.__signaling_info.signal,))
            pgroup = ComponentProxy("system").signal_process_groups([{'id':self.taskid}], self.__signaling_info.signal)
        except (ComponentLookupError, xmlrpclib.Fault), e:
            #
            # BRT: will a ComponentLookupError ever be raised directly or will 
            # it always be buried in a XML-RPC fault?
            #
            # BRT: shouldn't we be checking the XML-RPC fault code?  which 
            # fault codes are valid for this operation?  at the very least 
            # unexpected fault code should be reported as such and the retry 
            # loop broken.
            #
            if retry:
                self._sm_log_warn("failed to communicate with the system component (%s); retry pending" % (e,))
                return Job.__rc_retry
            else:
                self._sm_log_warn("failed to communicate with the system component (%s); manual cleanup may be required" % \
                    (e,))
                return Job.__rc_xmlrpc
        except:
            self._sm_raise_exception("unexpected error from the system component; manual cleanup may be required")
            return Job.__rc_unknown

        self.__signaled_info = self.__signaling_info
        del self.__signaling_info
        return Job.__rc_success

    def __task_run(self):
        global run_id_gen
        walltime = self.walltime
        if self.runid is None: #allow us to unset this for preemption
            self.runid = run_id_gen.next() #Don't try and run this twice.

        if self.preemptable and self.maxtasktime < walltime:
            walltime = self.maxtasktime
        try:
            self._sm_log_info("instructing the system component to begin executing the task")
            pgroup = ComponentProxy("system", retry=False).add_process_groups([{
                'id':"*",
                'jobid':self.jobid,
                'user':self.user,
                'stdin':self.inputfile,
                'stdout':self.outputpath,
                'stderr':self.errorpath,
                'cobalt_log_file':self.cobalt_log_file,
                'size':self.procs,
                'nodect':"*",
                'mode':self.mode,
                'script_preboot':self.script_preboot,
                'cwd':self.outputdir,
                'executable':self.command,
                'args':self.args,
                'env':self.envs,
                'location':self.location,
                'umask':self.umask,
                'kernel':self.kernel,
                'kerneloptions':self.kerneloptions,
                'ion_kernel':self.ion_kernel,
                'ion_kerneloptions':self.ion_kerneloptions,
                'starttime':self.starttime,
                'walltime':walltime,
                'killtime':self.force_kill_delay + 1,
                'resid': self.resid,
                'runid': self.runid,
                'attrs': self.attrs,
                'queue': self.queue,
                'project': self.project
            }])
            if pgroup[0].has_key('id'):
                self.taskid = pgroup[0]['id']
                if pgroup[0].has_key('nodect') and pgroup[0]['nodect'] != None:
                    self.__resource_nodects.append(pgroup[0]['nodect'])
                else:
                    self.__resource_nodects.append(self.nodes)
            else:
                self._sm_log_error("process group creation failed", 
                        cobalt_log = True)
                return Job.__rc_pg_create
        except (ComponentLookupError, xmlrpclib.Fault), e:
            self._sm_log_warn("failed to execute the task (%s); retry pending" % (e,))
            return Job.__rc_retry
        except:
            self._sm_raise_exception("unexpected error returned from the system component when attempting to add task",
                cobalt_log = True)
            return Job.__rc_unknown
        else:
            # Start task timer.  This corresponds to the compute time
            if self.walltime > 0:
                self.__max_job_timer = Timer(self.walltime * 60)
            else:
                self.__max_job_timer = Timer()
            self.__max_job_timer.start()
            self.current_task_start = time.time()
            task_start = accounting.task_start(self.jobid, self.runid, self.current_task_start, self.location)
            accounting_logger.info(task_start)
            logger.info(task_start)
        return Job.__rc_success

    def __task_finalize(self):
        '''get exit code from system component'''
        def end_time_and_log():
            self.__max_job_timer.stop()
            task_end = accounting.task_end(self.jobid, self.runid, self.__max_job_timer.elapsed_times[-1], self.current_task_start,
                    time.time(), self.location)
            self.current_task_start = None
            accounting_logger.info(task_end)
            logger.info(task_end)
        try:
            result = ComponentProxy("system").wait_process_groups([{'id':self.taskid, 'exit_status':'*'}])
            if result:
                self.exit_status = result[0].get('exit_status')
                dbwriter.log_to_db(None, "exit_status_update", "job_prog",
                    JobProgExitStatusMsg(self))
            else:
                self._sm_log_warn("system component was unable to locate the task; exit status not obtained")
        except (ComponentLookupError, xmlrpclib.Fault), e:
            self._sm_log_warn("failed to communicate with the system component (%s); retry pending" % (e,))
            return Job.__rc_retry
        except:
            # We aren't going into a retry and anything that doesn't return a retry "ends" the task and progresses towards
            # preemption/the job terminal action.  We need the log here.
            end_time_and_log()
            self._sm_raise_exception("unexpected error returned from the system component while finalizing task")
            return Job.__rc_unknown
        else:
            end_time_and_log()
        self.taskid = None
        return Job.__rc_success

    def __release_resources(self):
        '''release computing resources that may still be reserved by the job'''
        try:
            ComponentProxy("system").reserve_resources_until(self.location, None, self.jobid)
            return Job.__rc_success
        except (ComponentLookupError, xmlrpclib.Fault), e:
            self._sm_log_warn("failed to communicate with the system component (%s); retry pending" % (e,))
            return Job.__rc_retry
        except:
            self._sm_raise_exception("unexpected error returned from the system component while releasing resources")
            return Job.__rc_unknown

    def _sm_log_debug(self, msg, cobalt_log = False):
        '''write an informational message to the CQM log that includes state machine status'''
        if self._sm_event != None:
            event_msg = "; Event=%s" % (self._sm_event,)
        else:
            event_msg = ""
        logger.debug("Job %s/%s: State=%s%s; %s" % (self.jobid, self.user, self._sm_state, event_msg, msg))
        if cobalt_log:
            self.__write_cobalt_log("Debug: %s" % (msg,))

    def _sm_log_info(self, msg, cobalt_log = False):
        '''write an informational message to the CQM log that includes state machine status'''
        if self._sm_event != None:
            event_msg = "; Event=%s" % (self._sm_event,)
        else:
            event_msg = ""
        logger.info("Job %s/%s: State=%s%s; %s" % (self.jobid, self.user, self._sm_state, event_msg, msg))
        if cobalt_log:
            self.__write_cobalt_log("Info: %s" % (msg,))

    def _sm_log_warn(self, msg, cobalt_log = False):
        '''write a warning message to the CQM log that includes state machine status'''
        if self._sm_event != None:
            event_msg = "; Event=%s" % (self._sm_event,)
        else:
            event_msg = ""
        logger.warning("Job %s/%s: State=%s%s; %s" % (self.jobid, self.user, self._sm_state, event_msg, msg))
        if cobalt_log:
            self.__write_cobalt_log("Warning: %s" % (msg,))

    def _sm_log_error(self, msg, tb_flag = True, skip_tb_entries = 1, cobalt_log = False):
        '''write an error message to the CQM log that includes state machine status and a stack trace'''
        if self._sm_event != None:
            event_msg = "; Event=%s" % (self._sm_event,)
        else:
            event_msg = ""
        full_msg = "Job %s/%s: State=%s%s; %s" % (self.jobid, self.user, self._sm_state, event_msg, msg)
        if tb_flag:
            stack = traceback.format_stack()
            last_tb_entry = len(stack) - skip_tb_entries
            for entry in stack[:last_tb_entry]:
                full_msg += "\n    " + entry[:-1]
        logger.error(full_msg)
        if cobalt_log:
            self.__write_cobalt_log("ERROR: %s" % (msg,))

    def _sm_log_exception(self, msg, cobalt_log = False):
        '''write an error message to the CQM log that includes state machine status and a stack trace'''
        if self._sm_event != None:
            event_msg = "; Event=%s" % (self._sm_event,)
        else:
            event_msg = ""
        full_msg = "Job %s/%s: State=%s%s; %s" % (self.jobid, self.user, self._sm_state, event_msg, msg)
        (exc_cls, exc, tb) = sys.exc_info()
        exc_str = traceback.format_exception_only(exc_cls, exc)[0]
        full_msg += "\n    Exception: %s" % (exc_str)
        stack = traceback.format_tb(tb)
        for entry in stack:
            full_msg += "\n    " + entry[:-1]
        logger.error(full_msg)
        if cobalt_log:
            self.__write_cobalt_log("EXCEPTION: %s" % (full_msg,))

    def _sm_raise_exception(self, msg, cobalt_log = False):
        self._sm_log_error(msg, skip_tb_entries = 2, cobalt_log = cobalt_log)
        raise JobProcessingError(msg, self.jobid, self.user, self.state, self._sm_state, self._sm_event)

    def _sm_log_user_delete(self, signame, user = None, pending = False):
        if user != None:
            umsg = " by user %s" % (user,)
        else:
            umsg = ""
        if pending == True:
            pmsg = " now pending"
        else:
            pmsg = ""
        self._sm_log_info("user delete requested with signal %s%s%s" % (signame, umsg, pmsg), cobalt_log = True)

    def _sm_signaling_info_set_user_delete(self, signame = Signal_Map.terminate, user = None, pending = False):
        self.__signaling_info = Signal_Info(Signal_Info.Reason.delete, signame, user, pending)
        self._sm_log_user_delete(signame, user, pending)

    def _sm_check_job_timers(self):
        if self.__max_job_timer.has_expired:
            # if the job execution time has exceeded the wallclock time, then inform the task that it must terminate
            self._sm_log_info("maximum execution time exceeded; initiating job termination", cobalt_log = True)
            accounting_logger.info(accounting.abort(self.jobid))
            return Signal_Info(Signal_Info.Reason.time_limit, Signal_Map.terminate)
        else:
            return None

    def _sm_kill_task(self):
        '''initiate the user deletion of a task by signaling it and then changing to the appropriate state'''
        # BRT: this routine should probably check if the task could not be signaled because it was no longer running
        rc = self.__task_signal()
        if rc == Job.__rc_success:
            # start the signal timer so that the state machine knows when to escalate to sending a force kill signal
            if self.__signaled_info.signal != Signal_Map.force_kill: