delayed run-db thread shutdown 
@arjclark - while testing #613 with the massive "phil" suite I noticed a severe exacerbation, and a nasty consequence, of some strange behaviour that I have seen before.  At shutdown cylc stops the request handling and command execution threads, calls db.close(), and then prints "Main thread shutting down" and exits.  Except that it doesn't exit immediately - a single thread associated with the suite hangs about for a number of minutes (for this large suite at least).  Users won't notice this in daemon mode because after the intended exit point everything (cylc scan etc.) says the suite has stopped, and you can go ahead and re-run or restart it.   BUT if you do run or restart the suite while this left-over process is still running (see threads with `ps -eLf`) the new suite will take _much longer_ to load its task proxies: on the phil suite, 4 min vs 30 sec for a restart on my desktop box (and I killed "cylc run" after 8 min).  The probable culprit here is the rundb, because it gets queried with each task proxy initialization. 

diff --git a/bin/cylc-restart b/bin/cylc-restart
index e9e62899a..268a413f0 100755
--- a/bin/cylc-restart
+++ b/bin/cylc-restart
@@ -321,7 +321,7 @@ where they got to while the suite was down."""
 
             try:
                 # startup is True only for a cold start
-                itask = self.config.get_task_proxy( name, tag, state, stopctime=None, startup=False )
+                itask = self.config.get_task_proxy( name, tag, state, stopctime=None, startup=False, submit_num=self.db.get_task_current_submit_num(name, tag), exists=self.db.get_task_state_exists(name, tag) )
             except Exception, x:
                 print >> sys.stderr, str(x)
                 print >> sys.stderr, "WARNING: ignoring task", name, "from the suite state dump file"
@@ -347,15 +347,9 @@ where they got to while the suite was down."""
 
             elif itask.state.is_currently('submitted', 'running'):  
                 itask.prerequisites.set_all_satisfied()
-                c = itask.db.connect()
                 # get user_at_host from run-db
                 user_at_host = None
-                c.execute(r"""
-                        SELECT misc FROM task_events
-                        WHERE name==? AND cycle==? AND
-                              event=="submission succeeded" AND misc!=""
-                        ORDER BY submit_num DESC LIMIT 1""", (name, tag))
-                row = c.fetchone()
+                row = self.db.get_task_location(name, tag)
                 if row and row[0]:
                     user_at_host = row[0]
                     self.old_user_at_host_set.add(str(user_at_host))
@@ -364,16 +358,11 @@ where they got to while the suite was down."""
 
                 # get submit_method_id and try_num from run-db
                 submit_method_id = try_num = None
-                c.execute(r"""
-                        SELECT submit_method_id, try_num FROM task_states
-                        WHERE name==? AND cycle==? 
-                        ORDER BY submit_num DESC LIMIT 1""", (name, tag))
-                row = c.fetchone()
+                row = self.db.get_task_sumbit_method_id_and_try(name, tag)
                 if row and row[0]:
                     submit_method_id, try_num = row
                 else:
                     print >> sys.stderr, "WARNING:", id, "failed to read submit_method_id and try_num from run-db!"
-                c.close()
 
                 if None in [ user_at_host, submit_method_id, try_num ]:
                     print >> sys.stderr, "WARNING: cannot determine what happened to " + id
diff --git a/bin/cylc-run b/bin/cylc-run
index 06433584e..d42606ccb 100755
--- a/bin/cylc-run
+++ b/bin/cylc-run
@@ -116,7 +116,7 @@ initial and final cycle time variables persists across suite restarts."""
                 else:
                     # no initial cycle time: we can't load cycling tasks
                     continue
-            itask = self.config.get_task_proxy( name, tag, 'waiting', stopctime=None, startup=True )
+            itask = self.config.get_task_proxy( name, tag, 'waiting', stopctime=None, startup=True, submit_num=0, exists=False )
             # check stop cycle in case the user has set a very quick stop
             if self.stop_tag and int( itask.tag ) > int( self.stop_tag ):
                 # we've reached the stop cycle already: delete the new task 
@@ -147,7 +147,7 @@ initial and final cycle time variables persists across suite restarts."""
                 else:
                     # no initial cycle time: we can't load cycling tasks
                     continue
-            itask = self.config.get_task_proxy( name, tag, 'waiting', stopctime=None, startup=False )
+            itask = self.config.get_task_proxy( name, tag, 'waiting', stopctime=None, startup=False, submit_num=0, exists=False )
             if name in coldstart_tasks:
                 itask.log( "NORMAL", "warm start: starting in succeeded state" )
                 itask.state.set_status( 'succeeded' )
@@ -190,7 +190,7 @@ initial and final cycle time variables persists across suite restarts."""
                 else:
                     # no initial cycle time: we can't load cycling tasks
                     continue
-            itask = self.config.get_task_proxy( name, tag, 'waiting', stopctime=None, startup=False )
+            itask = self.config.get_task_proxy( name, tag, 'waiting', stopctime=None, startup=False, submit_num=0, exists=False )
             if name in coldstart_tasks or name in startup_tasks:
                 itask.log( "NORMAL", "This is a raw start: I will self-destruct." )
                 itask.prepare_for_death()
diff --git a/bin/cylc-submit b/bin/cylc-submit
index 4a1ade48b..7b210f42b 100755
--- a/bin/cylc-submit
+++ b/bin/cylc-submit
@@ -224,14 +224,14 @@ else:
     tag = ctime
 try:
     # try to get a graphed task of the given name
-    task_proxy = config.get_task_proxy( task_name, tag, 'waiting', stopctime=None, startup=True )
+    task_proxy = config.get_task_proxy( task_name, tag, 'waiting', stopctime=None, startup=True, submit_num=0, exists=False )
 except TaskNotDefinedError, x:
     print >> sys.stderr, 'WARNING: "' + task_name + \
             '" type and validity are not constrained by the graph.'
     print >> sys.stderr, "Continuing under the assumption that the given task ID is valid."
     try:
         # try to get an ungraphed task of the given name
-        task_proxy = config.get_task_proxy_raw( task_name, tag, 'waiting', stoptag=None, startup=True )
+        task_proxy = config.get_task_proxy_raw( task_name, tag, 'waiting', stoptag=None, startup=True, submit_num=0, exists=False )
     except TaskNotDefinedError, y:
         raise SystemExit('ERROR: task "' + task_name + '" runtime is not defined either, aborting.')
 
diff --git a/lib/cylc/config.py b/lib/cylc/config.py
index 1cfdb9582..5b27ac3d2 100644
--- a/lib/cylc/config.py
+++ b/lib/cylc/config.py
@@ -1879,14 +1879,14 @@ Some translations were performed on the fly."""
 
         return taskd
    
-    def get_task_proxy( self, name, ctime, state, stopctime, startup ):
+    def get_task_proxy( self, name, ctime, state, stopctime, startup, submit_num, exists ):
         try:
             tdef = self.taskdefs[name]
         except KeyError:
             raise TaskNotDefinedError("ERROR, No such task name: " + name )
-        return tdef.get_task_class()( ctime, state, stopctime, startup )
+        return tdef.get_task_class()( ctime, state, stopctime, startup, submit_num=submit_num, exists=exists )
 
-    def get_task_proxy_raw( self, name, tag, state, stoptag, startup ):
+    def get_task_proxy_raw( self, name, tag, state, stoptag, startup, submit_num, exists ):
         # Used by 'cylc submit' to submit tasks defined by runtime
         # config but not currently present in the graph (so we must
         # assume that the given tag is valid for the task).
@@ -1903,7 +1903,7 @@ Some translations were performed on the fly."""
         else:
             # assume input cycle is valid
             tdef.hours = [ int( foo.hour ) ]
-        return tdef.get_task_class()( tag, state, stoptag, startup )
+        return tdef.get_task_class()( tag, state, stoptag, startup, submit_num=submit_num, exists=exists )
 
     def get_task_class( self, name ):
         return self.taskdefs[name].get_task_class()
diff --git a/lib/cylc/rundb.py b/lib/cylc/rundb.py
index ffa5e13e8..a9f69c07b 100644
--- a/lib/cylc/rundb.py
+++ b/lib/cylc/rundb.py
@@ -222,10 +222,25 @@ class CylcRuntimeDAO(object):
 
     def get_task_state_exists(self, name, cycle):
         s_fmt = "SELECT COUNT(*) FROM task_states WHERE name==? AND cycle==?"
-        args = [name, cycle,]
+        args = [name, cycle]
         count = self.c.select(s_fmt, args).next()[0]
         return count > 0
 
+    def get_task_location(self, name, cycle):
+        s_fmt = """SELECT misc FROM task_events WHERE name==? AND cycle==? 
+                   AND event=="submission succeeded" AND misc!=""
+                   ORDER BY submit_num DESC LIMIT 1"""
+        args = [name, cycle]
+        res = self.c.select(s_fmt, args).next()
+        return res
+
+    def get_task_sumbit_method_id_and_try(self, name, cycle):
+        s_fmt = """SELECT submit_method_id, try_num FROM task_states WHERE name==? AND cycle==? 
+                   ORDER BY submit_num DESC LIMIT 1"""
+        args = [name, cycle]
+        res = self.c.select(s_fmt, args).next()
+        return res        
+
     def run_db_op(self, db_oper):
         self.c.execute(db_oper.s_fmt, db_oper.args)
     
diff --git a/lib/cylc/scheduler.py b/lib/cylc/scheduler.py
index ec39a2572..de37c56c1 100644
--- a/lib/cylc/scheduler.py
+++ b/lib/cylc/scheduler.py
@@ -663,7 +663,7 @@ class scheduler(object):
         for task_id in task_ids:
             name, tag = task_id.split( TaskID.DELIM )
             # TODO - insertion of start-up tasks? (startup=False is assumed here)
-            new_task = self.config.get_task_proxy( name, tag, 'waiting', stop_tag, startup=False )
+            new_task = self.config.get_task_proxy( name, tag, 'waiting', stop_tag, startup=False, submit_num=self.db.get_task_current_submit_num(name, tag), exists=self.db.get_task_state_exists(name, tag))
             self.add_new_task_proxy( new_task )
 
     def command_nudge( self ):
@@ -747,7 +747,7 @@ class scheduler(object):
                         self.log.warning( 'orphaned task will not continue: ' + itask.id  )
                 else:
                     self.log.warning( 'RELOADING TASK DEFINITION FOR ' + itask.id  )
-                    new_task = self.config.get_task_proxy( itask.name, itask.tag, itask.state.get_status(), None, itask.startup )
+                    new_task = self.config.get_task_proxy( itask.name, itask.tag, itask.state.get_status(), None, itask.startup, submit_num=self.db.get_task_current_submit_num(itask.name, itask.tag), exists=self.db.get_task_state_exists(itask.name, itask.tag) )
                     if itask.state.has_spawned():
                         new_task.state.set_spawned()
                     # succeeded tasks need their outputs set completed:
diff --git a/lib/cylc/task_types/task.py b/lib/cylc/task_types/task.py
index 58f132eac..b7577a664 100644
--- a/lib/cylc/task_types/task.py
+++ b/lib/cylc/task_types/task.py
@@ -223,25 +223,14 @@ class task( object ):
         self.submission_poll_timer = None
         self.execution_poll_timer = None
 
-        # sets submit num for restarts or when triggering state prior to submission
         if self.validate: # if in validate mode bypass db operations
             self.submit_num = 0
         else:
-            self.db_path = gcfg.get_derived_host_item( self.suite_name, 'suite run directory' )
-            self.db = cylc.rundb.CylcRuntimeDAO(suite_dir=self.db_path)
-            submits = self.db.get_task_current_submit_num(self.name, self.c_time)
-            if submits > 0:
-                self.submit_num = submits
-                self.record_db_update("task_states", self.name, self.c_time, status=self.state.get_status()) #is this redundant?
-            else:
-                self.submit_num = 0
+            if not self.exists:
+                self.record_db_state(self.name, self.c_time, submit_num=self.submit_num, try_num=self.try_number, status=self.state.get_status())
+            if self.submit_num > 0:
+                self.record_db_update("task_states", self.name, self.c_time, status=self.state.get_status())
 
-            if not self.db.get_task_state_exists(self.name, self.c_time):
-                try:
-                    self.record_db_state(self.name, self.c_time, submit_num=self.submit_num, try_num=self.try_number, status=self.state.get_status()) #queued call
-                except:
-                    pass
-            self.db.close()
 
     def log( self, priority, message ):
         logger = logging.getLogger( "main" )
@@ -967,7 +956,6 @@ class task( object ):
 
             outp = self.id + " submitted" # hack: see github #476
             self.outputs.set_completed( outp )
-            self.record_db_update("task_states", self.name, self.c_time, status="submitted")
             self.record_db_event(event="submission succeeded" )
             handler = self.event_handlers['submitted']
             if handler:
@@ -980,6 +968,7 @@ class task( object ):
                 # to 'submitted' and set the job submission timer if
                 # currently still in the 'submitting'state.
                 self.set_status( 'submitted' )
+                self.record_db_update("task_states", self.name, self.c_time, status="submitted")
                 self.submission_timer_start = self.submitted_time
                 self.submission_poll_timer.set_timer()
 
diff --git a/lib/cylc/taskdef.py b/lib/cylc/taskdef.py
index b87b54101..fc308f365 100644
--- a/lib/cylc/taskdef.py
+++ b/lib/cylc/taskdef.py
@@ -246,11 +246,13 @@ class taskdef(object):
         tclass.add_prerequisites = tclass_add_prerequisites
 
         # class init function
-        def tclass_init( sself, start_tag, initial_state, stop_c_time=None, startup=False, validate=False ):
+        def tclass_init( sself, start_tag, initial_state, stop_c_time=None, startup=False, validate=False, submit_num=0, exists=False ):
 
             sself.cycon = container.cycon( self.cyclers )
             sself.intercycle_offset = self.intercycle_offset
             sself.startup = startup
+            sself.submit_num = submit_num
+            sself.exists=exists
             if self.cycling: # and startup:
                 # adjust only needed at start-up but it does not hurt to
                 # do it every time as after the first adjust we're already
['bin/cylc-submit', 'lib/cylc/taskdef.py', 'lib/cylc/scheduler.py', 'lib/cylc/task_types/task.py', 'bin/cylc-run', 'lib/cylc/config.py', 'lib/cylc/rundb.py', 'bin/cylc-restart']