#!/usr/bin/env python3
#
# Execute runs of tasks and put the results into the database.
#

import daemon
import datetime
import errno
import multiprocessing
import optparse
import os
import pscheduler
import psycopg2
import psycopg2.extensions
import psycopg2.pool
import select
import signal
import socket
import tempfile
import threading
import time


pscheduler.set_graceful_exit()

# Gargle the arguments

opt_parser = optparse.OptionParser()

# Daemon-related options

opt_parser.add_option("--daemon",
                      help="Daemonize",
                      action="store_true",
                      dest="daemon", default=False)
opt_parser.add_option("--pid-file",
                      help="Location of PID file",
                      action="store", type="string", dest="pidfile",
                      default=None)

# Program options

opt_parser.add_option("-d", "--dsn",
                      help="Database connection string",
                      action="store", type="string", dest="dsn",
                      default="")
opt_parser.add_option("-m", "--max-parallel",
                      help="Maximum concurrent runs",
                      action="store", type="int", dest="max_parallel",
                      default=15)
opt_parser.add_option("-r", "--refresh",
                      help="Forced refresh interval (ISO8601)",
                      action="store", type="string", dest="refresh",
                      default="PT1M")
opt_parser.add_option("--terse-logging",
                      help="Don't log run details",
                      action="store_true",
                      dest="terse",
                      default=False)


opt_parser.add_option("--verbose", action="store_true", dest="verbose", default=False)
opt_parser.add_option("--debug", action="store_true", dest="debug", default=False)

(options, args) = opt_parser.parse_args()

refresh = pscheduler.iso8601_as_timedelta(options.refresh)
if refresh is None:
    opt_parser.error('Invalid refresh interval "' + options.refresh + '"')
if pscheduler.timedelta_as_seconds(refresh) == 0:
    opt_parser.error("Refresh interval must be calculable as seconds.")


log = pscheduler.Log(verbose=options.verbose, debug=options.debug, propagate=True)

dsn = pscheduler.string_from_file(options.dsn)

log_details = not options.terse

# Get this once so we don't have to do a function call every time.
delimiter = pscheduler.api_result_delimiter()



# Current value of run start margin according to the database

__run_start_margin = None

def run_start_margin_set(db):
    global __run_start_margin
    with db.cursor() as cursor:
        try:
            cursor.execute("SELECT run_start_margin FROM configurables")
        except Exception as ex:
            raise RuntimeError("Unable to get run start margin: %s" % (str(ex)))

        if cursor.rowcount != 1:
            raise RuntimeError("Failed to get run start margin")

        __run_start_margin = cursor.fetchone()[0]
        log.debug("Run start margin set to %s", __run_start_margin)


def run_start_margin():
    global __run_start_margin
    assert __run_start_margin is not None, "Run start margin was never set"
    return __run_start_margin




#
# Directory of what we believe to be running and what we don't.
#
# Even though Python de-parallelizes threads, force this to become
# thread-safe in case it gets smarter later.
#

# TODO: Use the ThreadSafeSet and ThreadSafeSetHold instead

class RunDictionary(object):

    def __init__(self):
        self.dict = {}
        self.lock = threading.Lock()

    def start(self, id, worker):
        with self.lock:
            self.dict[id] = worker

    def finish(self, id):
        with self.lock:
            del self.dict[id]

    def is_running(self, id):
        with self.lock:
            result = id in self.dict
        return result

run_dict = RunDictionary()


#
# Clock Survey
#

def get_clock(arg):
    slot, url, bind = arg
    status, result = pscheduler.url_get(url, throw=False, bind=bind)
    if status != 200:
        result = { "error": status }
    return (slot, result)


def clock_survey(hosts, bind=None):

    if len(hosts) == 0:
        return []

    # Any null hosts become the local host
    hosts = [ pscheduler.api_local_host() if host is None else host
              for host in hosts ]

    host_args = []
    for slot in range(0,len(hosts)):
        host = hosts[slot]
        if host is None:
            continue
        host_args.append((slot, pscheduler.api_url(host, "/clock"), bind))

    # Prime the result with empties for anything that didn't get
    # tested.
    result = [None] * len(hosts)

    # Run the lot of tests in parallel
    pool = multiprocessing.pool.ThreadPool(processes=len(host_args))
    for slot, clock in pool.imap(get_clock, host_args, chunksize=1):
        result[slot] = clock
    pool.close()

    return pscheduler.json_dump(result)



#
# Class that does the test runs
#

class RunWorker(object):

    def __init__(self, dbpool, log, id, start_at):

        # Per http://initd.org/psycopg/docs/usage.html#thread-safety,
        # Psycopg is thread-safe when you use multiple cursors against the
        # same connection.

        self.dbpool = dbpool
        self.log = log
        self.id = id
        self.start_at = start_at
        self.finished = False
        self.output = []

        self.worker = threading.Thread(target=lambda: self.run())
        self.worker.setDaemon(True)


    def start(self):
        self.worker.start()


    def _put_db_conn(self, db):
        """
        Commits any changes made with a database connection and restores to the pool
        """
        if not db:
            return
    
        try:
            db.commit()
        except Exception as ex:
            self.log.warning("Error committing DB transaction: %s" % ex)
        finally:
            self.dbpool.putconn(db)
    
    def _get_db_conn(self, attempts=60, wait=1):
        """
        Get connection from pool. If cannot get connection retry a few times before giving-up
        """
        db = None
        for attempt in range(0, attempts):
            try:
                db = self.dbpool.getconn()
                break
            except psycopg2.pool.PoolError:
                time.sleep(wait)
        
        #Unable to get connection, throw exception
        if db is None:
            raise Exception("Database connection pool exhausted. Unable to get connection after %d attempts." % attempts)
        
        return db
    
    def __post_new_result(self, result):
        """
        Post a finished run for this task using the result provided.
        """
        self.log.debug("%d: Got result: %s", self.id, result)
        try:
            json = pscheduler.json_load(result, max_schema=1)
        except ValueError:
            log.warning("%d: Discarding bogus result %s", self.id, result)
            return
        
        db = None
        try:
            db = self._get_db_conn()
            with db.cursor() as cursor:
                cursor.execute("""
                WITH inserted_row AS (
                    INSERT INTO run (task, uuid, times, state, status, result_merged)
                    VALUES (%s,
                            NULL,
                            tstzrange(normalized_now(), normalized_now(), '[]'),
                            run_state_finished(),
                            0,
                            %s)
                    RETURNING *
                ) SELECT uuid FROM inserted_row
                """, [ self.task_id, result ])

                # Should get exactly one row back with the UUID of
                # the new result.

                if cursor.rowcount != 1:
                    self.log.error("%d: Failed to get UUID of posted run.",
                                   self.id)
                    return

                if log_details:
                    self.log.info("%d: Posted result to %s/runs/%s",
                                  self.id, self.task_url,
                                  cursor.fetchone()[0])


        except Exception as ex:
            self.log.error("%d: Failed to post run for result: %s", self.id, str(ex))
        finally:
            self._put_db_conn(db)


    def __accumulate_output(self, line):
        """
        Accumulate lines of output from the tool in an array until the
        magic delimiter appears.  When it does, use it to post a
        finished run for the same task.
        """
        # TODO: This should be available in the pScheduler module
        if line == delimiter:
            self.__post_new_result("\n".join(self.output))
            self.output = []
        else:
            self.output.append(line)
        
    def run(self):
        """
        Run the tool in an exception-safe way
        """
        self.log.debug("%d: Thread running", self.id)
        run_dict.start(self.id, self)
        try:
            with tempfile.TemporaryDirectory() as self.temp:
                self.log.debug("%d: Temp space in %s", self.id, self.temp)
                self.__run()
            self.temp = None
        except Exception as ex:
            # Don't worry about the result here.  If __run() failed to
            # post anything, that will be the end of it.  If it did,
            # it might be salvageable.
            self.log.debug("%d: Exception: %s", self.id, ex)
            log.exception()

        self.log.debug("%d: Thread finished", self.id)
        run_dict.finish(self.id)


    def __run(self):
        """
        Run the tool and, if the lead participant, gather, aggregate
        and post the results.
        """

        failures = 0


        # Do as much preparation as possible before going to sleep.

        db = None
        try:
            db = self._get_db_conn()
            with db.cursor() as cursor:
                cursor.execute("""
                           SELECT            
                               test.name,
                               tool.name,
                               task.uuid,
                               task.id,
                               task.participant,
                               task.participants,
                               task.limits_passed,
                               lower(run.times),
                               upper(run.times),
                               task.json #> '{test}',
                               task.cli,
                               run.uuid,
                               run.part_data_full,
                               scheduling_class.enum,
                               task.json ->> 'lead-bind',
                               task.json #> '{contexts,contexts}',
                               task.json ->> '_key'
                           FROM
                               run
                               JOIN task ON task.id = run.task
                               JOIN test ON test.id = task.test
                               JOIN scheduling_class
                                    ON scheduling_class.id = test.scheduling_class
                               JOIN tool ON tool.id = task.tool
                           WHERE run.id = %s
                           """, [self.id])

                # Should get exactly one row back.  If not, the run probably
                # vanished.

                if cursor.rowcount != 1:
                    if log_details:
                        self.log.info("%d: Run is gone.  Stopping.", self.id)
                    return

                row = cursor.fetchone()

                test, tool, task_uuid, task_id, participant, participants, \
                    limits_passed, start, end, test_spec, cli, run_uuid, \
                    partdata, scheduling_class, lead_bind, contexts, key = row

        except Exception as ex:
            self.log.error("%d: Failed to fetch run info: %s",
                           self.id, str(ex))
            return
        finally:
            self._put_db_conn(db)

        if lead_bind is not None:
            self.log.debug("%d: Lead bind for this run is %s",
                           self.id, lead_bind)

        # This gets used by __post_result, above.
        self.task_url = pscheduler.api_url_hostport(
            hostport=participants[0],
            path="/tasks/%s" % (task_uuid) )

        # This is only used for debut messages.
        run_url = "%s/runs/%s" % (self.task_url, run_uuid)

        # This will be used when a background-multi run produces a result.
        self.task_id = task_id

        if partdata is None:
            # TODO: Should simply bail out here.
            self.log.error("%d: Got NULL part_data_full for %s",
                           self.id, run_url)

        tool_input = pscheduler.json_dump({
            'schema': 1,
            'task-uuid': task_uuid,
            'schedule': {
                'start': pscheduler.datetime_as_iso8601(start),
                'duration': pscheduler.timedelta_as_iso8601(end - start)
                },
            'test': test_spec,
            'participant': participant,
            'participant-data': partdata,
            'limits-passed': limits_passed
            })


        # If there are contexts, do the advance work for that.

        # TODO: Might be good to see if we can just do all of te
        # running with the ChainedExecRunner.

        if contexts is not None and len(contexts[participant]):
            context_args = [
                {
                    "program": [
                        pscheduler.plugin_method_path("context", context["context"], "change")
                    ],
                    "input": context.get("data", {})
                }
                for context in contexts[participant]
                ]

            context_runner = pscheduler.ChainedExecRunner(
                context_args,
                argv=[ pscheduler.plugin_method_path("tool", tool, "run") ],
                stdin=tool_input)

        else:

            context_runner = None

        #
        # Wait for the start time to roll around
        #

        self.log.debug("%d: Start at %s", self.id, self.start_at)
        start_time = self.start_at - run_start_margin()
        self.log.debug("%d: Sleeping until test start at %s", self.id,
                       start_time)
        pscheduler.sleep_until(start_time)

        #
        # Get cracking
        #

        if log_details:
            self.log.info("%d: Running %s", self.id, run_url)
            self.log.info("%d: With %s: %s %s", self.id, tool, test, " ".join(cli))

        # Tell the database we're proceeding

        db = None
        try:
            db = self._get_db_conn()
            with db.cursor() as cursor:
                cursor.execute("SELECT run_start(%s)", [self.id])
                assert cursor.rowcount == 1
                result = cursor.fetchone()[0]
                if result is not None:
                    self.log.info("%d: Aborting run: %s", self.id, result)
                    return
        except AssertionError as ae:
            self.log.error("%d: run_start did not return exactly one row.", self.id)
            raise ae
        except Exception as ex:
            self.log.error("%d: Failed to start run: %s", self.id, str(ex))
            raise ex
        finally:
            self._put_db_conn(db)


        #
        # Do the local tool run
        #

        self.log.debug("%d: Tool input: %s", self.id, tool_input)

        how_late = pscheduler.time_now() - start_time
        self.log.debug("%d: Start time difference is %s",
                       self.id, how_late)

        if how_late > datetime.timedelta(seconds=0.5):
            self.log.warning("%d: Starting %s later than scheduled",
                             self.id, how_late)

        timeout = pscheduler.timedelta_as_seconds(end - start + run_start_margin()) + 1
        if context_runner is None:

            returncode, stdout, stderr = pscheduler.plugin_invoke(
                "tool", tool, "run",
                stdin=tool_input,
                env_add={ "TMPDIR": self.temp },
                timeout=timeout,
                line_call=lambda l: self.__accumulate_output(l))

        else:

            returncode, stdout, stderr = context_runner.run(
                env_add={ "TMPDIR": self.temp },
                timeout=timeout,
                line_call=lambda l: self.__accumulate_output(l)
            )


        stdout = "\n".join(self.output)

        if len(stdout) == 0:
            stdout = None
        else:
            # See if the test claimed failure
            try:
                result_json = pscheduler.json_load(stdout, max_schema=1)
                if "succeeded" in result_json \
                        and isinstance(result_json["succeeded"], bool) \
                        and result_json["succeeded"] == False:
                    failures += 1
            except ValueError:
                self.log.error("%d: Test returned invalid JSON: %s", self.id, stdout)


        if len(stderr) == 0:
            stderr = None

        if returncode == 0:
            if log_details:
                self.log.info("%d: Run succeeded.", self.id)
            self.log.debug("%d: Run returned %s", self.id, stdout)
        else:
            failures += 1
            if log_details:
                self.log.info("%d: Run failed %d: %s", self.id,
                              returncode, stderr)
            if stderr is None:
                stderr = "Tool exited with status %d" % (returncode)
        
        db = None
        try:
            db = self._get_db_conn()
            state_selector = 2 if  (
                (participant > 0) or (scheduling_class == "background-multi")
                ) else returncode
            with db.cursor() as cursor:
                cursor.execute("""
                UPDATE run
                SET
                status = %s,
                result = %s,
                errors = %s,
                state = CASE
                            WHEN %s = 0 THEN run_state_cleanup()
                            WHEN %s = 1 THEN run_state_failed()
                            ELSE run_state_finished()
                        END
                WHERE id = %s
                """,
                [returncode,
                 stdout,
                 stderr,
                 state_selector,
                 state_selector,
                 self.id])
        except Exception as ex:
            self.log.error("%d: Failed to store local result: %s",
                           self.id, str(ex))

            return
        finally:
            #finally should ensure this is executed  even if return is an exception
            self._put_db_conn(db)
        
        self.log.debug("%d: Stored local result", self.id)

        # The lead participant in non-background-multi tasks takes care of
        # gathering and merging the finished results.  Background-multi
        # tasks take care of inserting their own results.

        if participant == 0 and scheduling_class != "background-multi":

            self.log.debug("%d: Doing lead participant duties", self.id)

            # Wait until the scheduled time has passed, which is the
            # only time we can be sure results might be available.

            if len(participants) > 1:
                wait_time = pscheduler.time_until_seconds(end)
                self.log.debug("%d: Waiting for task end time to pass (%s)",
                               self.id, wait_time)
                time.sleep(wait_time)
                self.log.debug("%d: Task end time has passed", self.id)
            else:
                self.log.debug("%d: Only one participant; not waiting.", self.id)

            # Fetch and combine the results.

            runs = [ pscheduler.api_url_hostport(
                hostport = host,
                path = '/tasks/%s/runs/%s'
                % (task_uuid, run_uuid) )
                     for host in participants ]

            self.log.debug("%d: Runs are %s", self.id, runs)
            self.log.debug("%d: Local run returned %d",
                           self.id, returncode)

            if returncode == 0:
                try:
                    local_result = pscheduler.json_load(stdout)
                except ValueError:
                    self.log.error("%d: Tool %s returned invalid JSON %s",
                                   self.id, tool, stdout)
                    local_result = None
                    failures += 1
            else:
                self.log.debug("%d: Tool returned failure: %s",
                               self.id, stderr)
                local_result = None
                failures += 1


            # Assemble the results from each participant into an
            # array.

            result_full = [ {
                "succeeded": False,
                "error": "No result was produced",
                "diags": "No result was produced"
            } for run in runs]

            # We have this on hand.
            result_full[0] = local_result
            self.log.debug("%d: Accumulated local result", self.id)

            # Wait up to 15 seconds for all of the participants to
            # produce results.

            to_get = dict([(runs[index], index)
                           for index in range(1, len(runs))])

            deadline = pscheduler.time_now() + datetime.timedelta(seconds=15)

            for get in list(to_get):
                self.log.debug("%d: Fetching run %s", self.id, get)

                timeout = int(pscheduler.timedelta_as_seconds(
                    deadline - pscheduler.time_now()))

                status, run_result = pscheduler.url_get(
                    get,
                    params={ 'wait-local': True, 'wait': timeout },
                    bind=lead_bind,
                    throw=False,
                    timeout=timeout + 0.5
                )

                if status == 200:
                    self.log.debug("%d: Retrieved %s", self.id, run_result)
                    index = to_get[get]
                    got = run_result["result"]
                    if got is not None:
                        result_full[index] = got
                else:
                    self.log.warning("%d: Unable to retrieve run %s",
                                         self.id, get)
                    got = { "succeeded": False,
                            "error": run_result,
                            "diags": run_result
                        }

                del to_get[get]


            failures = len(list([result for result in result_full if result is None or not result.get("succeeded",False)]))
            self.log.debug("%d: %d failures", self.id, failures)


            # If there were any failures, survey all of the
            # particpants' clocks.

            survey = clock_survey(participants, lead_bind) if failures > 0 else None

            # Merge the results

            if failures:
                result_merged = { "succeeded": False }
            else:
                self.log.debug("%d: Merging results: %s", self.id, result_full)
                try:
                    merge_input = {
                        "test": test_spec,
                        "results": result_full
                    }
                    merge_input_text = pscheduler.json_dump(merge_input)
                    self.log.debug("%d: Merging %s", self.id, merge_input_text)

                    returncode, stdout, stderr = pscheduler.plugin_invoke(
                        "tool", tool, "merged-results",
                        stdin=merge_input_text, timeout=5)

                    self.log.debug("%d: Merged results: %d %s %s", self.id,
                                   returncode, stdout, stderr)

                    if returncode != 0:
                        raise Exception(stderr)

                    result_merged = pscheduler.json_load(stdout)

                except Exception as ex:
                    result_merged = { "succeeded": False }

                    # This overwrites the run result but is a good way to get
                    # the error message in front of the user.
                    result_full[0]["succeeded"] = False
                    result_full[0]["diags"] = "Run succeeded, but failed to merge results: %s" % (str(ex))


            result_full_text = pscheduler.json_dump(result_full)
            self.log.debug("%d: Full result: %s ", self.id, result_full_text)

            result_merged_text = pscheduler.json_dump(result_merged)
            self.log.debug("%d: Merged result: %s ", self.id, result_merged_text)

            # Store full and merged results and the clock survey in
            # the local database.

            try:
                self.log.debug("%d: Merged result: Setting final state ", self.id)
                db = self._get_db_conn()
                with db.cursor() as cursor:
                    # TODO: Need to figure out succeeded.
                    cursor.execute("""
                                    UPDATE run
                                    SET
                                        state = CASE
                                                    WHEN %s = 0 THEN run_state_finished()
                                                    ELSE run_state_failed()
                                                END,
                                        result_full = %s,
                                        result_merged = %s,
                                        clock_survey = %s
                                    WHERE id = %s
                                    """,
                                    [failures,
                                     pscheduler.json_dump(result_full),
                                     result_merged_text,
                                     survey,
                                     self.id])
            except Exception as ex:
                self.log.error("%d: Failed to store run: %s", self.id, str(ex))
            finally:
                self._put_db_conn(db)


        self.log.debug("%d: Run complete", self.id)
        self.finished = True



#
# Main Program
#


def main_program():

    log.debug("Begin main")

    # This is for local use.
    db = pscheduler.pg_connection(dsn)
    log.debug("Connected to DB")

    with db.cursor() as cursor:
        cursor.execute("SELECT heartbeat_boot('runner')")

    with db.cursor() as cursor:

        # Get the maximum number of connections available and limit use of
        # them to half.

        try:
            cursor.execute("SHOW max_connections")

            # Should get exactly one row back.  If not, the run probably
            # vanished.

            if cursor.rowcount != 1:
                raise Exception("Got %d rows back when expecting one"
                                % (cursor.rowcount))

            max_connections = int(cursor.fetchone()[0])
        except Exception as ex:
            log.error("Failed to fetch maximum connections: %s", str(ex))
            # This is the PgSQL default
            max_connections = 100
            log.error("Using default of %d", max_connections)

    pool_size = int(max_connections / 2)
    log.debug("Using a pool size of %d", pool_size)

    # Pool of connections for use by the threads.  Note that these
    # connections do not have autocommit, so anything using them will
    # need to do its own commits.

    dbpool = psycopg2.pool.ThreadedConnectionPool(
        int(pool_size/100),   # Minimum connections
        pool_size,            # Maximum connections
        dsn="%s application_name=runner-pool" % (dsn))


    # Listen for notifications.

    for listen in ["run_new", "run_change", "configurables_changed" ]:
        log.debug("Listening for notification %s" % (listen))
        with db.cursor() as cursor:
            cursor.execute("LISTEN %s" % (listen))

    # Prime this for the first run
    wait_time = datetime.timedelta()
    run_start_margin_set(db)


    while True:

        log.debug("Next run or check in %s", wait_time)
        if not pscheduler.timedelta_is_zero(wait_time):

            # Wait for a notification or the wait time to elapse.  Eat all
            # notifications as a group; we only care that we were notified.

            # TODO: This try needs to be brought to the other programs.
            # Better, make it a function in db.py.

            with db.cursor() as cursor:
                cursor.execute("SELECT heartbeat('runner', %s)", [wait_time])

            try:
                if pscheduler.polled_select(
                        [db],[],[],
                        pscheduler.timedelta_as_seconds(wait_time)) \
                    != ([],[],[]):
                    # Notified
                    db.poll()
                    notifies = [ notify.channel for notify in db.notifies]
                    log.debug("Notifications: %s", notifies)
                    if 'configurables_changed' in notifies:
                        log.debug("Configurables changed.")
                        run_start_margin_set(db)
                    else:
                        log.debug("Schedule change.")
                    del db.notifies[:]

            except select.error as ex:

                err_no, message = ex.args
                if err_no != errno.EINTR:
                    log.exception()
                    raise ex


        with db.cursor() as cursor:
            cursor.execute("SELECT heartbeat('runner')")

        with db.cursor() as cursor:

            # Operate only on runs that are scheduled to start before the next
            # forced refresh.
            # TODO: Error check this.
            cursor.execute("""
                       SELECT * FROM (

                       -- Tasks that haven't started
                       SELECT
                           id AS run,
                           lower(times) - normalized_wall_clock() AS start_in,
                           lower(times) AS start_at,
                           FALSE as background_multi
                       FROM
                           run
                       WHERE
                           lower(run.times) < (normalized_wall_clock() + %s)
                           AND state = run_state_pending()
                           AND part_data_full IS NOT NULL

                       UNION

                       -- Background tasks that should be running.
                       SELECT
                           run.id AS run,
                           'PT1S'::INTERVAL AS start_in,
                           normalized_wall_clock() + 'PT1S'::INTERVAL AS start_at,
                           TRUE as background_multi
                       FROM
                           run
                           JOIN task ON task.id = run.task
                           JOIN test ON test.id = task.test
                       WHERE
                           times @> normalized_now()
                           AND task.enabled
                           AND test.scheduling_class = scheduling_class_background_multi()
                           AND run.state IN (run_state_pending(), run_state_running())
                       ) t
                       WHERE start_in > '0'::INTERVAL
                       ORDER BY start_in
                   """, [refresh]);


            wait_time = refresh

            already_running = 0
            run_ids = []
            run_workers = []
            for row in cursor:

                run_id, start_in, start_at, background_multi = row

                if run_dict.is_running(run_id):
                    log.debug("%d is already running" % (run_id))
                    already_running += 1
                    continue

                log.debug("Run %d, starts at %s", run_id, start_at)

                run_ids.append(run_id)

                worker = RunWorker(dbpool, log, run_id, start_at)
                run_workers.append(worker)
                log.debug("%d: Created worker", run_id)

                if not background_multi and start_in < wait_time:
                    log.debug("Dropping wait time to %s", start_in)
                    wait_time = start_in

        if already_running > 0:
            log.debug("Skipped %d already-running runs", already_running)

        if not run_workers:
            log.debug("No workers to run.")
            wait_time = refresh
            continue

        with db.cursor() as cursor:
            # Do this here to guarantee that we don't pick up rows for
            # runs we just started in the next iteration of the loop.
            if run_ids:
                log.debug("Putting runs on deck: %s", run_ids)
                try:
                    cursor.execute("""
                                   UPDATE run
                                   SET state = run_state_on_deck()
                                   WHERE
                                       id in %s
                                       AND state = run_state_pending()
                                   """, (tuple(run_ids),))
                except Exception as ex:
                    log.error("Unable to set on-deck status: %s" % (str(ex)))

        for worker in run_workers:
            worker.start()


    # Not that this will ever be reached...
    db.close()


if options.daemon:
    pidfile = pscheduler.PidFile(options.pidfile)
    with daemon.DaemonContext(pidfile=pidfile):
        pscheduler.safe_run(lambda: main_program())
else:
    pscheduler.safe_run(lambda: main_program())


