Merge lp:~cr3/checkbox-core/shadowd into lp:checkbox-core

[Marc Tardif]

[Sylvain Pineau (sylvain-pineau)]

I would add this function to properly close the connection in enqueue():

    def disconnect(self, sender):
        if not self.is_active:
            return

        if queue_disconnect(sender) < 0:
            raise ApplicationError("Failed to close connection.")

        self.is_active = False

And probably check for errors and raise ApplicationError in the reschedule command.

But both can be implemented later, so approved !

review: Approve

[Marc Tardif (cr3)]

I've applied both your suggestions and merged into trunk. Thanks!

=== added file 'bin/checkbox-shadowd'
--- bin/checkbox-shadowd	1970-01-01 00:00:00 +0000
+++ bin/checkbox-shadowd	2012-05-01 02:07:19 +0000
@@ -0,0 +1,30 @@
+#!/usr/bin/python
+#
+# Copyright (c) 2012 Canonical
+#
+# This file is part of Checkbox.
+#
+# Checkbox is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as
+# published by the Free Software Foundation; either version 2.1 of
+# the License, or (at your option) any later version.
+#
+# Checkbox is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+import sys
+
+try:
+    import _preamble
+except ImportError:
+    sys.exc_clear()
+
+from checkbox.scripts import shadowd
+
+
+shadowd.run()

=== modified file 'checkbox/daemon/client.py'
--- checkbox/daemon/client.py	2012-04-24 22:43:09 +0000
+++ checkbox/daemon/client.py	2012-05-01 02:07:19 +0000
@@ -325,7 +325,13 @@
             messenger.startCommand(message)
 
     def _doLocate(self):
-        return True
+        if self.type == ClientType.ANY:
+            return True
+        elif self.type == ClientType.SHADOWD:
+            return self._getDaemonInfo()
+        else:
+            raise Exception(
+                "Unsupported client type: %s" % ClientType[self.type])
 
     def _initializeHostname(self):
         if self._initialized_hostname:
@@ -471,7 +477,7 @@
         return True
 
     def _getConnectorInfo(self):
-        if self._address and is_ipaddress(self._address)[0]:
+        if self._address and address_is_valid(self._address):
             self._port = address_to_port(self._address)
             if self._port > 0:
                 logging.debug("Already have address, no info to locate.")

=== modified file 'checkbox/daemon/operation.py'
--- checkbox/daemon/operation.py	2012-04-24 22:43:09 +0000
+++ checkbox/daemon/operation.py	2012-05-01 02:07:19 +0000
@@ -45,8 +45,8 @@
     "OFF_PEACEFUL",
     "SET_PEACEFUL_SHUTDOWN",
     "TIME_OFFSET",
-    ("SHADOWD_BASE", 100),
-    "QUEUE",
+    ("SHADOWD_BASE", 200),
+    "ENQUEUE",
     "CONTINUE_CLAIM",
     "SUSPEND_CLAIM",
     "DEACTIVATE_CLAIM",

=== modified file 'checkbox/daemon/server.py'
--- checkbox/daemon/server.py	2012-04-23 19:36:19 +0000
+++ checkbox/daemon/server.py	2012-05-01 02:07:19 +0000
@@ -560,7 +560,10 @@
         if sock_handler.function:
             logging.info("Calling handler for sock")
 
-            result = sock_handler.function(sock)
+            if sock_handler.data is not None:
+                result = sock_handler.function(sock, sock_handler.data)
+            else:
+                result = sock_handler.function(sock)
             logging.info("Return from handler")
 
         elif handle_command:
@@ -1176,11 +1179,11 @@
                             or sock_timed_out):
                             if (sock_timed_out.sock._finish_connect()
                                 != StreamError.WOULDBLOCK):
-                                sock_timed_out.call_handler = True
+                                sock_handler.call_handler = True
                     else:
                         if (selector.check(sock, SelectorIO.READ)
                             or sock_timed_out):
-                            sock_timed_out.call_handler = True
+                            sock_handler.call_handler = True
 
                 # Scan through the pipe table to find which ones are callable
                 for pipe_handler in self._pipe_table:
@@ -1239,7 +1242,7 @@
 
                         # Enqueue this packet into the buffers if this
                         # is a udp sock
-                        if sock.get_protocol() == "udp":
+                        if sock.protocol == "udp":
                             if not sock.handle_incoming_packet():
                                 # There is not yet a complete message ready
                                 continue
@@ -1287,7 +1290,7 @@
             self._sock_udp.unserialize(string)
 
     def _handleRequest(self, sock):
-        protocol = sock.get_protocol()
+        protocol = sock.protocol
         if protocol == "tcp":
             is_tcp = True
         elif protocol == "udp":
@@ -1376,7 +1379,7 @@
             return True
 
     def _handleRequestSockTimerHandler(self, sock):
-        assert(sock.get_protocol() == "tcp")
+        assert(sock.protocol == "tcp")
 
         self.unregisterSock(sock)
 
@@ -1706,7 +1709,7 @@
             socket.IPPROTO_TCP, socket.TCP_NODELAY, 1):
             logging.info("Failed to set TCP_NODELAY on TCP command port.")
 
-        if not sock_tcp.listen_on_port(port):
+        if not sock_tcp.listen(port):
             return cleanup(
                 "Failed to listen on port %d for TCP command socket.", port)
 

=== modified file 'checkbox/daemon/subsystem.py'
--- checkbox/daemon/subsystem.py	2012-04-24 22:43:09 +0000
+++ checkbox/daemon/subsystem.py	2012-05-01 02:07:19 +0000
@@ -126,6 +126,10 @@
         self.addEntry(
             SubsystemType.MASTER, SubsystemCategory.DAEMON, "MASTER")
         self.addEntry(
+            SubsystemType.SHADOW, SubsystemCategory.DAEMON, "SHADOW")
+        self.addEntry(
+            SubsystemType.SHADOWD, SubsystemCategory.DAEMON, "SHADOWD")
+        self.addEntry(
             SubsystemType.RUNNER, SubsystemCategory.DAEMON, "RUNNER")
         self.addEntry(
             SubsystemType.RUNNERD, SubsystemCategory.DAEMON, "RUNNERD")

=== modified file 'checkbox/journal/event.py'
--- checkbox/journal/event.py	2012-04-23 19:36:19 +0000
+++ checkbox/journal/event.py	2012-05-01 02:07:19 +0000
@@ -143,12 +143,13 @@
 
     type = None
 
-    def __init__(self):
+    def __init__(self, cluster=-1, proc=-1, subproc=-1):
+        self.cluster = cluster
+        self.proc = proc
+        self.subproc = subproc
+
         self.event_clock = time.time()
         self.event_time = time.localtime(self.event_clock)
-        self.cluster = -1
-        self.proc = -1
-        self.subproc = -1
 
     def _initializeFromRecord(self, record):
         self.cluster = record.evaluateInteger("cluster", -1)
@@ -263,8 +264,8 @@
 
     type = EventType.GENERIC
 
-    def __init__(self):
-        super(GenericEvent, self).__init__()
+    def __init__(self, *args, **kwargs):
+        super(GenericEvent, self).__init__(*args, **kwargs)
         self.data = ""
 
     def _initializeFromRecord(self, record):
@@ -276,7 +277,9 @@
         self.data = record.evaluateString("generic_data", "")
 
     def _writeEvent(self, file):
-        file.write("%s\n" % self.data)
+        if file.write("%s\n" % self.data) < 0:
+            return False
+
         return True
 
     def _readEvent(self, file):
@@ -292,8 +295,8 @@
 
     type = EventType.EXECUTE
 
-    def __init__(self):
-        super(ExecuteEvent, self).__init__()
+    def __init__(self, *args, **kwargs):
+        super(ExecuteEvent, self).__init__(*args, **kwargs)
         self.host = ""
 
     def _initializeFromRecord(self, record):
@@ -305,7 +308,9 @@
         self.host = record.evaluateString("execute_host", "")
 
     def _writeEvent(self, file):
-        file.write("Job executing on host: %s\n" % self.host)
+        if file.write("Job executing on host: %s\n" % self.host) < 0:
+            return False
+
         return True
 
     def _readEvent(self, file):
@@ -326,8 +331,8 @@
 
     type = EventType.SUSPEND
 
-    def __init__(self):
-        super(SuspendEvent, self).__init__()
+    def __init__(self, *args, **kwargs):
+        super(SuspendEvent, self).__init__(*args, **kwargs)
         self.pid_count = 0
 
     def _initializeFromRecord(self, record):
@@ -339,8 +344,13 @@
         self.pid_count = record.evaluateString("pid_count", 0)
 
     def _writeEvent(self, file):
-        file.write("Job was suspended\n")
-        file.write("Number of processes suspended: %d\n" % self.pid_count)
+        if file.write("Job was suspended\n") < 0:
+            return False
+
+        if file.write(
+            "Number of processes suspended: %d\n" % self.pid_count) < 0:
+            return False
+
         return True
 
     def _readEvent(self, file):
@@ -363,7 +373,9 @@
     type = EventType.RESUME
 
     def _writeEvent(self, file):
-        file.write("Job was resumed\n")
+        if file.write("Job was resumed\n") < 0:
+            return False
+
         return True
 
     def _readEvent(self, file):
@@ -374,6 +386,42 @@
         return True
 
 
+class SubmitEvent(EventInfo):
+
+    __slots__ = (
+        "host",
+        )
+
+    type = EventType.SUBMIT
+
+    def __init__(self, *args, **kwargs):
+        super(SubmitEvent, self).__init__(*args, **kwargs)
+        self.host = ""
+
+    def _writeEvent(self, file):
+        if file.write("Job submitted from host: %s\n" % self.host) < 0:
+            return False
+
+        return True
+
+    def _readEvent(self, file):
+        line = file.readline()
+        match = re.match(r"Job submitted from host: (?P<host>[^\s]*)\n", line)
+        if not match:
+            return False
+
+        self.host = match.group("host")
+        return True
+
+    def _initializeFromRecord(self, record):
+        super(SubmitEvent, self)._initializeFromRecord(record)
+
+        if not record:
+            return
+
+        self.host = record.evaluateString("host", "")
+
+
 class TerminateEvent(EventInfo):
 
     __slots__ = (
@@ -395,8 +443,8 @@
         ("total_sent_bytes", "Total Bytes Sent"),
         ("total_received_bytes", "Total Bytes Received"))
 
-    def __init__(self):
-        super(TerminateEvent, self).__init__()
+    def __init__(self, *args, **kwargs):
+        super(TerminateEvent, self).__init__(*args, **kwargs)
         self.sent_bytes = self.received_bytes = 0.0
         self.total_sent_bytes = self.total_received_bytes = 0.0
         self.return_value = self.signal_number = -1
@@ -404,25 +452,33 @@
         self.core_file = None
 
     def _writeEvent(self, file):
-        file.write("Job terminated\n")
+        if file.write("Job terminated\n") < 0:
+            return False
 
         if self.normal:
-            file.write(
+            if file.write(
                 "\t(1) Normal termination (return value %d)\n"
-                % self.return_value)
+                % self.return_value) < 0:
+                return False
         else:
-            file.write(
+            if file.write(
                 "\t(0) Abnormal termination (signal %d)\n"
-                % self.signal_number)
+                % self.signal_number) < 0:
+                return False
 
             if self.core_file:
-                file.write("\t(1) Core file in: %s\n" % self.core_file)
+                ret = file.write("\t(1) Core file in: %s\n" % self.core_file)
             else:
-                file.write("\t(0) No core file\n")
+                ret = file.write("\t(0) No core file\n")
+
+            if ret < 0:
+                return False
 
         # Write bytes
         for attribute, string in self.attribute_strings:
-            file.write("\t%.0f  -  %s\n" % (getattr(self, attribute), string))
+            if file.write(
+                "\t%.0f  -  %s\n" % (getattr(self, attribute), string)) < 0:
+                return False
 
         return True
 

=== modified file 'checkbox/message/attributes.py'
--- checkbox/message/attributes.py	2012-04-23 19:36:19 +0000
+++ checkbox/message/attributes.py	2012-05-01 02:07:19 +0000
@@ -43,6 +43,8 @@
     JOB_STATE="state",
     JOB_START_TIME="start_time",
     JOB_EXIT_TIME="exit_time",
+    JOB_ROOT_DIR="root_dir",
+    JOB_WORKING_DIR="working_dir",
 
     CLUSTER_ID="cluster_id",
     PROC_ID="proc_id",

=== added file 'checkbox/scripts/shadowd.py'
--- checkbox/scripts/shadowd.py	1970-01-01 00:00:00 +0000
+++ checkbox/scripts/shadowd.py	2012-05-01 02:07:19 +0000
@@ -0,0 +1,78 @@
+#
+# Copyright (c) 2012 Canonical
+#
+# This file is part of Checkbox.
+#
+# Checkbox is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as
+# published by the Free Software Foundation; either version 2.1 of
+# the License, or (at your option) any later version.
+#
+# Checkbox is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+__metaclass__ = type
+
+__all__ = [
+    "run",
+    ]
+
+from optparse import OptionGroup
+
+from checkbox.daemon.subsystem import (
+    SubsystemType,
+    set_subsystem,
+    )
+
+from checkbox.shadowd.starter import Starter
+
+from checkbox.scripts.application import Application
+from checkbox.scripts.mixins.config import ConfigMixin
+from checkbox.scripts.mixins.daemon import DaemonMixin
+from checkbox.scripts.mixins.logger import LoggerMixin
+
+
+class ShadowdApplication(ConfigMixin, DaemonMixin, LoggerMixin, Application):
+
+    # Application defaults
+    usage = "Usage: %prog [OPTIONS]"
+
+    # Shadowd defaults
+    default_name = None
+
+    def addOptions(self, parser):
+        """See Application."""
+        super(ShadowdApplication, self).addOptions(parser)
+
+        group = OptionGroup(parser, "Shadowd options")
+        group.add_option("--name",
+            default=self.default_name,
+            help="Shadow daemon name.")
+
+        parser.add_option_group(group)
+
+    def parseOptions(self, options, args):
+        super(ShadowdApplication, self).parseOptions(options, args)
+
+        self.shadow_name = options.name
+
+    def preProcess(self):
+        # Set the subsystem for the daemon server.
+        set_subsystem("SHADOWD", SubsystemType.SHADOWD)
+
+        super(ShadowdApplication, self).preProcess()
+
+    def process(self):
+        starter = Starter()
+        starter.initialize()
+        starter.register()
+
+
+def run():
+    application = ShadowdApplication()
+    application.run()

=== modified file 'checkbox/scripts/submit.py'
--- checkbox/scripts/submit.py	2012-04-24 22:43:09 +0000
+++ checkbox/scripts/submit.py	2012-05-01 02:07:19 +0000
@@ -22,6 +22,7 @@
     "run",
     ]
 
+import os
 import sys
 
 from optparse import (
@@ -29,7 +30,9 @@
     OptionValueError,
     )
 from signal import SIGPIPE
+from StringIO import StringIO
 
+from checkbox.daemon.operation import OperationCode
 from checkbox.daemon.subsystem import (
     SubsystemType,
     set_subsystem,
@@ -40,6 +43,13 @@
     sigaction,
     )
 
+from checkbox.message.attributes import Attributes
+from checkbox.message.parser import (
+    ParserError,
+    ParserStream,
+    )
+from checkbox.message.record import Record
+
 from checkbox.scripts.application import (
     Application,
     ApplicationError,
@@ -49,10 +59,6 @@
 
 from checkbox.shadowd.client import Client
 from checkbox.shadowd.queue import queue_connect
-from checkbox.shadowd.sender import (
-    send_create_cluster,
-    send_create_process,
-    )
 
 
 class SubmitApplication(ConfigMixin, LoggerMixin, Application):
@@ -71,10 +77,6 @@
         super(SubmitApplication, self).addOptions(parser)
 
         group = OptionGroup(parser, "Submit options")
-        group.add_option("--append",
-            action="append",
-            type="string",
-            help="Append expressions to each job in the file.")
         group.add_option("--pool",
             metavar="HOST",
             help="Host as the central manager to query.")
@@ -102,9 +104,6 @@
         else:
             self.file = sys.stdin
 
-        # Append option
-        self.append = options.append
-
         if options.local and options.remote:
             raise OptionValueError(
                 "Must not specify both --local and --remote.")
@@ -113,7 +112,7 @@
         elif options.remote:
             self.name = options.remote
 
-        self.client = Client(self.name, options.pool)
+        self.pool = options.pool
 
     def preProcess(self):
         # Install signal handlers.
@@ -125,15 +124,50 @@
         super(SubmitApplication, self).preProcess()
 
     def process(self):
-        self.queue()
-
-    def connect(self):
+        jobs = self.read(self.file)
+
+        client = Client(self.name, self.pool)
+        if not client.locate():
+            raise ApplicationError("Failed to find address of local shadowd")
+
+        self.enqueue(client, jobs)
+        self.reschedule(client)
+
+    def read(self, file):
+        job = Record()
+        parser = ParserStream()
+        for line in file.readlines():
+            # Skip over comments
+            line = line.strip()
+            if line[0] == "#":
+                continue
+
+            if not line:
+                if job.references:
+                    yield job
+                job = Record()
+
+            try:
+                tree = parser.parseRecord(StringIO(line))
+            except ParserError, error:
+                raise ApplicationError(error)
+
+            if not tree:
+                raise ApplicationError("Failed to parse line: %s" % line)
+
+            for key, value in tree.references.iteritems():
+                job.insert(key, value)
+
+        if job.references:
+            yield job
+
+    def connect(self, client):
         if self.is_active:
             return
 
-        sock = queue_connect(self.client.address, 0)
-        if not sock:
-            name = self.client.name
+        sender = queue_connect(client.address, 0)
+        if not sender:
+            name = client.name
             if name:
                 raise ApplicationError(
                     "Failed to connect to queue manager %s" % name)
@@ -143,21 +177,42 @@
 
         self.is_active = True
 
-        return
+        return sender
 
-    def queue(self, count=1):
+    def enqueue(self, client, jobs):
         """Place one or more copies of the jobs in the queue.
 
         :param count: Number of times to submit the jobs to the queue.
         """
-        sock = self.connect()
-
-        for i in xrange(count):
-            cluster = send_create_cluster(sock)
-            if cluster < 0:
-                raise ApplicationError("Failed to create cluster.")
-
-            process = send_create_process(sock, cluster)
+        sender = self.connect(client)
+        cluster = sender.createCluster()
+        if cluster < 0:
+            raise ApplicationError("Failed to create cluster.")
+
+        for job in jobs:
+            process = sender.createProcess(cluster)
+            if process < 0:
+                raise ApplicationError("Failed to create process.")
+
+            self.setRootDir(job)
+            self.setWorkingDir(job)
+
+    def reschedule(self, client):
+        client.sendCommand(OperationCode.RESCHEDULE)
+
+    def setRootDir(self, job):
+        root_dir = self.computerRootDir()
+        job.insertString(Attributes.JOB_ROOT_DIR, root_dir)
+
+    def setWorkingDir(self, job):
+        working_dir = self.computeWorkingDir(self)
+        job.insertString(Attributes.JOB_WORKING_DIR, working_dir)
+
+    def computeRootDir(self):
+        return "/"
+
+    def computeWorkingDir(self):
+        return os.getcwd()
 
 
 def run():

=== modified file 'checkbox/shadowd/queue.py'
--- checkbox/shadowd/queue.py	2012-04-24 22:43:09 +0000
+++ checkbox/shadowd/queue.py	2012-05-01 02:07:19 +0000
@@ -30,10 +30,7 @@
     )
 from checkbox.daemon.operation import OperationCode
 
-from checkbox.shadowd.sender import (
-    send_close_connection,
-    send_close_socket,
-    )
+from checkbox.shadowd.sender import Sender
 
 
 def queue_connect(location, timeout=0):
@@ -57,26 +54,26 @@
 
         return None
     else:
-        sock = client.startCommand(OperationCode.QUEUE, "tcp", timeout)
+        sock = client.startCommand(OperationCode.ENQUEUE, "tcp", timeout)
         if not sock:
             logging.info("Failed to connect to queue manager.")
             return None
 
-    return sock
-
-
-def queue_disconnect(sock, commit=True):
+    return Sender(sock)
+
+
+def queue_disconnect(sender, commit=True):
     """Close the connection to the shadow daemon job queue and optionally
     commit the transaction.
 
     :param sock: Sock object returned by `queue_connect`.
     :param commit: Optional transaction commit, defaults to True.
     """
-    if not sock:
+    if not sender:
         return False
 
     if commit:
-        ret = send_close_connection()
+        ret = sender.closeConnection()
 
-    send_close_socket()
+    sender.closeSocket()
     return ret

=== modified file 'checkbox/shadowd/sender.py'
--- checkbox/shadowd/sender.py	2012-04-24 22:43:09 +0000
+++ checkbox/shadowd/sender.py	2012-05-01 02:07:19 +0000
@@ -19,10 +19,7 @@
 __metaclass__ = type
 
 __all__ = [
-    "send_create_cluster",
-    "send_create_process",
-    "send_close_connection",
-    "send_close_socket",
+    "Sender",
     ]
 
 import logging
@@ -30,47 +27,61 @@
 from checkbox.shadowd.operation import OperationCode
 
 
-def send_create_cluster(sock):
-    assert(sock.putInteger(OperationCode.CREATE_CLUSTER))
-    assert(sock.putEnd())
-
-    ret = sock.getInteger()
-    if ret < 0:
-        error = sock.getInteger()
-        logging.info("Received create cluster error: %d", error)
-
-    assert(sock.getEnd())
-    return ret
-
-
-def send_create_process(sock, cluster):
-    assert(sock.putInteger(OperationCode.CREATE_PROCESS))
-    assert(sock.putInteger(cluster))
-    assert(sock.putEnd())
-
-    ret = sock.getInteger()
-    if ret < 0:
-        error = sock.getInteger()
-        logging.info("Received create process error: %d", error)
-
-    assert(sock.getEnd())
-    return ret
-
-
-def send_close_connection(sock):
-    assert(sock.putInteger(OperationCode.CLOSE_CONNECTION))
-    assert(sock.putEnd())
-
-    ret = sock.getInteger()
-    if ret < 0:
-        error = sock.getInteger()
-        logging.info("Received close connection error: %d", error)
-
-    assert(sock.getEnd())
-    return ret
-
-
-def send_close_socket(sock):
-    assert(sock.putInteger(OperationCode.CLOSE_SOCKET))
-    assert(sock.putEnd())
-    return 0
+class Sender:
+
+    __slots__ = (
+        "_sock",
+        )
+
+    def __init__(self, sock):
+        self._sock = sock
+
+    def createCluster(self):
+        sock = self._sock
+
+        assert(sock.putInteger(OperationCode.CREATE_CLUSTER))
+        assert(sock.putEnd())
+
+        ret = sock.getInteger()
+        if ret < 0:
+            error = sock.getInteger()
+            logging.info("Received create cluster error: %d", error)
+
+        assert(sock.getEnd())
+        return ret
+
+    def createProcess(self, cluster):
+        sock = self._sock
+
+        assert(sock.putInteger(OperationCode.CREATE_PROCESS))
+        assert(sock.putInteger(cluster))
+        assert(sock.putEnd())
+
+        ret = sock.getInteger()
+        if ret < 0:
+            error = sock.getInteger()
+            logging.info("Received create process error: %d", error)
+
+        assert(sock.getEnd())
+        return ret
+
+    def closeConnection(self):
+        sock = self._sock
+
+        assert(sock.putInteger(OperationCode.CLOSE_CONNECTION))
+        assert(sock.putEnd())
+
+        ret = sock.getInteger()
+        if ret < 0:
+            error = sock.getInteger()
+            logging.info("Received close connection error: %d", error)
+
+        assert(sock.getEnd())
+        return ret
+
+    def closeSocket(self):
+        sock = self._sock
+
+        assert(sock.putInteger(OperationCode.CLOSE_SOCKET))
+        assert(sock.putEnd())
+        return 0

=== added file 'checkbox/shadowd/starter.py'
--- checkbox/shadowd/starter.py	1970-01-01 00:00:00 +0000
+++ checkbox/shadowd/starter.py	2012-05-01 02:07:19 +0000
@@ -0,0 +1,128 @@
+#
+# Copyright (c) 2012 Canonical
+#
+# This file is part of Checkbox.
+#
+# Checkbox is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as
+# published by the Free Software Foundation; either version 2.1 of
+# the License, or (at your option) any later version.
+#
+# Checkbox is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+__metaclass__ = type
+
+__all__ = [
+    "Starter",
+    ]
+
+import logging
+
+from checkbox.lib import param
+from checkbox.lib.hostname import fullhostname
+
+from checkbox.daemon.operation import OperationCode
+from checkbox.daemon.server import server
+
+from checkbox.process.timeslice import Timeslice
+
+
+class Starter:
+
+    __slots__ = (
+        "_name",
+        "_spool",
+        "_interval",
+        "_exit_when_done",
+        "_timer_id",
+        "_start_delay",
+        "_start_count",
+        )
+
+    min_interval_timer_set = False
+
+    def __init__(self):
+        self._name = None
+        self._spool = None
+        self._interval = Timeslice()
+        self._exit_when_done = False
+        self._timer_id = -1
+
+    def initialize(self):
+        self._spool = param.get("SPOOL")
+        if not self._spool:
+            raise Exception("No spool directory specified in config file.")
+
+        self._name = param.get("SHADOWD_NAME")
+        if not self._name:
+            self._name = fullhostname()
+
+        logging.debug("Using name: %s", self._name)
+
+        interval = self._interval
+        interval.default_interval = param.getInteger("SHADOWD_INTERVAL", 300)
+        interval.max_interval = interval.default_interval
+        interval.min_interval = param.getInteger("SHADOWD_MIN_INTERVAL", 5)
+
+        self._start_delay = param.getInteger("SHADOWD_START_DELAY", 5)
+        self._start_count = param.getInteger("SHADOWD_START_COUNT", 1)
+
+        # TODO: initialize queue
+
+    def register(self):
+        server.registerCommand(OperationCode.ENQUEUE, self._enqueue)
+        server.registerCommand(OperationCode.RESCHEDULE, self._reschedule)
+
+        self.registerTimers()
+
+    def registerTimers(self):
+        if self._timer_id >= 0:
+            server.unregisterTimer(self._timer_id)
+
+        self._timer_id = server.registerTimer(self.timeout, 10)
+
+    def timeout(self):
+        # If called too frequently, delay.
+        self._interval.expediteNextRun()
+        time_to_next_run = self._interval.time_to_next_run
+        if time_to_next_run > 0:
+            if not Starter.min_interval_timer_set:
+                logging.debug(
+                    "Setting delay until next queue scan to %d seconds",
+                    time_to_next_run)
+                server.resetTimer(self._timer_id, time_to_next_run, 1)
+                Starter.min_interval_timer_set = True
+            return
+
+        # TODO: Check if in walk job queue
+
+        Starter.min_interval_timer_set = False
+
+        # TODO: finish this
+
+    def _enqueue(self, command, stream):
+        print 'ENQUEUE!'
+        return 0
+
+    def _reschedule(self, command, stream):
+        if stream and not stream.end():
+            logging.info("Failed to receive end of message for RESCHEDULE.")
+            return 0
+
+        if self._exit_when_done:
+            return 0
+
+        self._needReschedule()
+        return 0
+
+    def _needReschedule(self):
+        self._need_reschedule = True
+
+        # Update the portal and request a reschedule.
+        self.timeout()

=== modified file 'checkbox/shadowd/tests/test_sender.py'
--- checkbox/shadowd/tests/test_sender.py	2012-04-24 22:43:09 +0000
+++ checkbox/shadowd/tests/test_sender.py	2012-05-01 02:07:19 +0000
@@ -28,13 +28,13 @@
 from checkbox.io.sock import SockAddress
 from checkbox.io.socktcp import SockTCP
 
-from checkbox.shadowd.sender import send_close_connection
-
-
-class TestSendCloseConnection(ProcessMixin, TestCase):
+from checkbox.shadowd.sender import Sender
+
+
+class TestSender(ProcessMixin, TestCase):
 
     def setUp(self):
-        super(TestSendCloseConnection, self).setUp()
+        super(TestSender, self).setUp()
         self.server = SockTCP()
         self.server.bind(SockAddress.LOOPBACK)
         self.server.listen()
@@ -43,8 +43,10 @@
         self.client = SockTCP()
         self.client.connect(host, port)
 
+        self.sender = Sender(self.client)
+
     def tearDown(self):
-        super(TestSendCloseConnection, self).setUp()
+        super(TestSender, self).setUp()
         self.client.close()
         self.server.close()
 
@@ -58,7 +60,7 @@
             sock.putEnd()
             os._exit(0)
 
-        self.assertEquals(send_close_connection(self.client), 0)
+        self.assertEquals(self.sender.closeConnection(), 0)
 
     def test_failure(self):
         if not self.forkChild():
@@ -71,4 +73,4 @@
             sock.putEnd()
             os._exit(0)
 
-        self.assertEquals(send_close_connection(self.client), -1)
+        self.assertEquals(self.sender.closeConnection(), -1)

=== modified file 'configs/checkbox.conf'
--- configs/checkbox.conf	2012-04-23 19:34:06 +0000
+++ configs/checkbox.conf	2012-05-01 02:07:19 +0000
@@ -6,6 +6,7 @@
 
 ## Pathnames
 LOG = %(LOCAL_DIR)s/log
+SPOOL = %(LOCAL_DIR)s/spool
 BIN = %(LOCAL_DIR)s/bin
 EXECUTE = %(LOCAL_DIR)s/execute
 
@@ -56,3 +57,11 @@
 
 ## Path to the starter binary.
 STARTER = %(BIN)s/checkbox-runner
+
+[SHADOWD]
+
+## How often should the shadowd send an update to the portal?
+#INTERVAL = 300
+
+## How long should the shadowd wait between spawning each shadow?
+#START_DELAY = 2