Stuck calculations in the new daemon
This is a continuation of #1274. 

Observed behaviour: Even after the fix in PR #1304, calculations still sometimes get stuck (especially long-running ones) after the daemon was restarted.

Apart from the general "it doesn't work", I also have a reproducible error:
- Submit a calculation, preferably over SSH. I'm not sure if this is essential, but it makes the timing easier.
- Stop the calculation.
- Exactly when it is marked as ``DONE`` in ``verdi calculation list``, but before it is retrieved / finished, try stopping the daemon
- The daemon now hangs indefinitely, and can only be ``kill -9``ed. The last log (on ``INFO`` loglevel) is "Querying the workflow DB".
- The calculation is now stuck in ``DONE``.

in order to more generally improve the stability of the daemon w.r.t. stuck calculations, maybe we should add a sort of "cleanup" task, that picks up calculations which are still in a valid non-finished state, but don't have a task associated with them?
diff --git a/.travis-data/workchains.py b/.travis-data/workchains.py
index 542b9cf68..3db43814a 100644
--- a/.travis-data/workchains.py
+++ b/.travis-data/workchains.py
@@ -86,7 +86,7 @@ class DynamicNonDbInput(WorkChain):
 
     def do_test(self):
         input_list = self.inputs.namespace.input
-        assert isinstance(input_list, tuple)
+        assert isinstance(input_list, list)
         assert not isinstance(input_list, List)
         self.out('output', List(list=list(input_list)))
 
diff --git a/.travis.yml b/.travis.yml
index 4e9eb82fb..4214555a5 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -41,7 +41,7 @@ install:
     # Upgrade pip setuptools and wheel to be able to run the next command
     - pip install -U pip wheel setuptools
     # Install AiiDA with some optional dependencies
-    - if [ "$TEST_TYPE" == "docs" ]; then pip install . && pip install -r docs/requirements_for_rtd.txt && pip install -e git://github.com/muhrin/plumpy.git@1dc151bbb5ee5f9a1f615589e545d29f547d93cf#egg=plumpy; else pip install .[rest,docs,atomic_tools,testing,dev_precommit] && pip install -r requirements.txt; fi
+    - if [ "$TEST_TYPE" == "docs" ]; then pip install . && pip install -r docs/requirements_for_rtd.txt && pip install -e git://github.com/muhrin/plumpy.git@2b628c331c286347fdd7ab12911ae82ad2146033#egg=plumpy; else pip install .[rest,docs,atomic_tools,testing,dev_precommit] && pip install -r requirements.txt; fi
 
 env:
 ## Build matrix to test both backends, and the docs
diff --git a/aiida/backends/tests/utils/serialize.py b/aiida/backends/tests/utils/serialize.py
index aae13b1fe..a7a6e3844 100644
--- a/aiida/backends/tests/utils/serialize.py
+++ b/aiida/backends/tests/utils/serialize.py
@@ -17,7 +17,7 @@ class TestSerialize(AiidaTestCase):
 
         data = {
             'test': 1,
-            'list': (1, 2, 3, node_a),
+            'list': [1, 2, 3, node_a],
             'dict': {
                 ('Si',): node_b,
                 'foo': 'bar'
diff --git a/aiida/cmdline/commands/calculation.py b/aiida/cmdline/commands/calculation.py
index 13abcded4..86e99c297 100644
--- a/aiida/cmdline/commands/calculation.py
+++ b/aiida/cmdline/commands/calculation.py
@@ -418,7 +418,7 @@ class Calculation(VerdiCommandWithSubcommands):
 
         if not is_dbenv_loaded():
             load_dbenv()
-        from aiida.plugins.entry_point import get_entry_from_class
+        from aiida.plugins.entry_point import get_entry_point_from_class
 
         try:
             calc = load_node(parsed_args.calc)
@@ -573,7 +573,7 @@ class Calculation(VerdiCommandWithSubcommands):
 
         if not is_dbenv_loaded():
             load_dbenv()
-        from aiida.plugins.entry_point import get_entry_from_class
+        from aiida.plugins.entry_point import get_entry_point_from_class
 
         try:
             calc = load_node(parsed_args.calc)
diff --git a/aiida/common/log.py b/aiida/common/log.py
index 08c758e39..248319cfa 100644
--- a/aiida/common/log.py
+++ b/aiida/common/log.py
@@ -83,7 +83,7 @@ LOGGING = {
                       '%(thread)d %(message)s',
         },
         'halfverbose': {
-            'format': '%(asctime)s, %(name)s: [%(levelname)s] %(message)s',
+            'format': '%(asctime)s <%(process)d> %(name)s: [%(levelname)s] %(message)s',
             'datefmt': '%m/%d/%Y %I:%M:%S %p',
         },
     },
diff --git a/aiida/daemon/runner.py b/aiida/daemon/runner.py
index 570a60dc1..2f24cbd57 100644
--- a/aiida/daemon/runner.py
+++ b/aiida/daemon/runner.py
@@ -7,17 +7,17 @@
 # For further information on the license, see the LICENSE.txt file        #
 # For further information please visit http://www.aiida.net               #
 ###########################################################################
+import logging
 import signal
 from functools import partial
 
-from aiida.common.log import aiidalogger
 from aiida.common.log import configure_logging
 from aiida.daemon.client import DaemonClient
 from aiida.work.rmq import get_rmq_config
 from aiida.work import DaemonRunner, set_runner
 
 
-logger = aiidalogger.getChild(__name__)
+logger = logging.getLogger(__name__)
 
 DAEMON_LEGACY_WORKFLOW_INTERVAL = 30
 
@@ -60,7 +60,7 @@ def tick_legacy_workflows(runner, interval=DAEMON_LEGACY_WORKFLOW_INTERVAL):
     :param runner: the DaemonRunner instance to perform the callback
     :param interval: the number of seconds to wait between callbacks
     """
-    logger.info('Ticking the legacy workflows')
+    logger.debug('Ticking the legacy workflows')
     legacy_workflow_stepper()
     runner.loop.call_later(interval, partial(tick_legacy_workflows, runner))
 
@@ -73,7 +73,7 @@ def legacy_workflow_stepper():
     from aiida.daemon.timestamps import set_daemon_timestamp, get_last_daemon_timestamp
     from aiida.daemon.workflowmanager import execute_steps
 
-    logger.info('Checking for workflows to manage')
+    logger.debug('Checking for workflows to manage')
     # RUDIMENTARY way to check if this task is already running (to avoid acting
     # again and again on the same workflow steps)
     try:
@@ -87,8 +87,8 @@ def legacy_workflow_stepper():
     if not stepper_is_running:
         set_daemon_timestamp(task_name='workflow', when='start')
         # The previous wf manager stopped already -> we can run a new one
-        logger.info('Running execute_steps')
+        logger.debug('Running execute_steps')
         execute_steps()
         set_daemon_timestamp(task_name='workflow', when='stop')
     else:
-        logger.info('Execute_steps already running')
\ No newline at end of file
+        logger.debug('Execute_steps already running')
\ No newline at end of file
diff --git a/aiida/daemon/workflowmanager.py b/aiida/daemon/workflowmanager.py
index a1d462e59..caec78ff0 100644
--- a/aiida/daemon/workflowmanager.py
+++ b/aiida/daemon/workflowmanager.py
@@ -38,7 +38,7 @@ def execute_steps():
     from aiida.orm import JobCalculation
     from aiida.orm.implementation import get_all_running_steps
  
-    logger.info("Querying the worflow DB")
+    logger.debug("Querying the worflow DB")
     
     running_steps = get_all_running_steps()
 
diff --git a/aiida/utils/serialize.py b/aiida/utils/serialize.py
index 4131016c1..59894637e 100644
--- a/aiida/utils/serialize.py
+++ b/aiida/utils/serialize.py
@@ -59,7 +59,7 @@ def serialize_data(data):
     elif isinstance(data, collections.Mapping):
         return {encode_key(key): serialize_data(value) for key, value in data.iteritems()}
     elif isinstance(data, collections.Sequence) and not isinstance(data, (str, unicode)):
-        return tuple(serialize_data(value) for value in data)
+        return [serialize_data(value) for value in data]
     else:
         return data
 
@@ -78,7 +78,7 @@ def deserialize_data(data):
     elif isinstance(data, collections.Mapping):
         return {decode_key(key): deserialize_data(value) for key, value in data.iteritems()}
     elif isinstance(data, collections.Sequence) and not isinstance(data, (str, unicode)):
-        return tuple(deserialize_data(value) for value in data)
+        return [deserialize_data(value) for value in data]
     elif isinstance(data, (str, unicode)) and data.startswith(_PREFIX_VALUE_NODE):
         return load_node(uuid=data[len(_PREFIX_VALUE_NODE):])
     elif isinstance(data, (str, unicode)) and data.startswith(_PREFIX_VALUE_GROUP):
diff --git a/aiida/work/job_processes.py b/aiida/work/job_processes.py
index d171a0c2d..e6f437097 100644
--- a/aiida/work/job_processes.py
+++ b/aiida/work/job_processes.py
@@ -166,6 +166,7 @@ class KillJob(TransportTask):
 
         if calc_state == calc_states.NEW or calc_state == calc_states.TOSUBMIT:
             calc._set_state(calc_states.FAILED)
+            calc._set_scheduler_state(job_states.DONE)
             calc.logger.warning("Calculation {} killed by the user "
                                 "(it was in {} state)".format(calc.pk, calc_state))
             return True
@@ -188,9 +189,7 @@ class KillJob(TransportTask):
         else:
             calc._set_state(calc_states.FAILED)
             calc._set_scheduler_state(job_states.DONE)
-            calc.logger.warning(
-                "Calculation {} killed by the user "
-                "(it was {})".format(calc.pk, calc_states.WITHSCHEDULER))
+            calc.logger.warning('Calculation<{}> killed by the user'.format(calc.pk))
 
         return result
 
@@ -243,18 +242,17 @@ class Waiting(plumpy.Waiting):
                     self.scheduler_update()
 
             elif self.data == UPDATE_SCHEDULER_COMMAND:
-                self._task = UpdateSchedulerState(calc, transport_queue)
-                job_done = yield self._task
-
-                if self._kill_future:
-                    yield self._do_kill()
-                else:
-                    if job_done:
-                        # Done, go on to retrieve
-                        self.retrieve()
-                    else:
-                        # Not done yet, keep getting updates
-                        self.scheduler_update()
+                job_done = False
+                # Keep geting scheduler updates until done
+                while not job_done:
+                    self._task = UpdateSchedulerState(calc, transport_queue)
+                    job_done = yield self._task
+                    if self._kill_future:
+                        yield self._do_kill()
+                        return
+
+                # Done, go on to retrieve
+                self.retrieve()
 
             elif self.data == RETRIEVE_COMMAND:
                 # Create a temporary folder that has to be deleted by JobProcess.retrieved after successful parsing
diff --git a/aiida/work/launch.py b/aiida/work/launch.py
index 05f352ca7..b65d5aa01 100644
--- a/aiida/work/launch.py
+++ b/aiida/work/launch.py
@@ -22,8 +22,7 @@ def submit(process_class, **inputs):
 
     # Use context manager to make sure connection is closed at end
     with rmq.new_blocking_control_panel() as control_panel:
-        pid = control_panel.execute_process_start(
-            process_class, init_kwargs={'inputs': inputs})
+        pid = control_panel.execute_process_start(process_class, init_kwargs={'inputs': inputs})
         return load_node(pid)
 
 
diff --git a/aiida/work/persistence.py b/aiida/work/persistence.py
index 23dfccf67..0ea616f12 100644
--- a/aiida/work/persistence.py
+++ b/aiida/work/persistence.py
@@ -8,6 +8,7 @@
 # For further information please visit http://www.aiida.net               #
 ###########################################################################
 
+import logging
 import uritools
 import os.path
 import os
@@ -20,6 +21,7 @@ from aiida import orm
 from . import class_loader
 
 Bundle = plumpy.Bundle
+LOGGER = logging.getLogger(__name__)
 
 
 # If portalocker accepts my pull request to have this incorporated into the
@@ -98,10 +100,16 @@ class AiiDAPersister(plumpy.Persister):
     """
 
     def save_checkpoint(self, process, tag=None):
+        LOGGER.info('Persisting process<{}>'.format(process.pid))
+
         if tag is not None:
             raise NotImplementedError("Checkpoint tags not supported yet")
 
-        bundle = Bundle(process, class_loader.CLASS_LOADER)
+        loader = class_loader.CLASS_LOADER
+
+        assert loader is not None, 'the class loader is not defined'
+
+        bundle = Bundle(process, loader)
         calc = process.calc
         calc._set_checkpoint(yaml.dump(bundle))
 
diff --git a/aiida/work/processes.py b/aiida/work/processes.py
index 6e670c965..ef636cbce 100644
--- a/aiida/work/processes.py
+++ b/aiida/work/processes.py
@@ -132,8 +132,7 @@ class Process(plumpy.Process):
         out_state[self.SaveKeys.CALC_ID.value] = self.pid
 
     def get_provenance_inputs_iterator(self):
-        return itertools.ifilter(lambda kv: not kv[0].startswith('_'),
-                                 self.inputs.iteritems())
+        return itertools.ifilter(lambda kv: not kv[0].startswith('_'), self.inputs.iteritems())
 
     @override
     def load_instance_state(self, saved_state, load_context):
@@ -145,24 +144,19 @@ class Process(plumpy.Process):
         load_context = load_context.copyextend(loop=self._runner.loop, communicator=self._runner.communicator)
         super(Process, self).load_instance_state(saved_state, load_context)
 
-        is_copy = saved_state.get('COPY', False)
-
         if self.SaveKeys.CALC_ID.value in saved_state:
-            if is_copy:
-                old = load_node(saved_state[self.SaveKeys.CALC_ID.value])
-                self._calc = old.copy()
-                self._calc.store()
-            else:
-                self._calc = load_node(saved_state[self.SaveKeys.CALC_ID.value])
-
+            self._calc = load_node(saved_state[self.SaveKeys.CALC_ID.value])
             self._pid = self.calc.pk
         else:
             self._pid = self._create_and_setup_db_record()
 
+        self.calc.logger.info('Loaded process<{}> from saved state'.format(self.calc.pk))
+
     def kill(self, msg=None):
         """
         Kill the process and all the children calculations it called
         """
+        self._calc.logger.info('Kill Process<{}>'.format(self._calc.pk))
         result = super(Process, self).kill(msg)
 
         for child in self.calc.called:
@@ -175,9 +169,13 @@ class Process(plumpy.Process):
         if value is None:
             # In this case assume that output_port is the actual value and there
             # is just one return value
-            return super(Process, self).out(self.SINGLE_RETURN_LINKNAME, output_port)
-        else:
-            return super(Process, self).out(output_port, value)
+            value = output_port
+            output_port = self.SINGLE_RETURN_LINKNAME
+
+        if isinstance(value, Node) and not value.is_stored:
+            value.store()
+
+        return super(Process, self).out(output_port, value)
 
     def out_many(self, out_dict):
         """
@@ -192,7 +190,7 @@ class Process(plumpy.Process):
         super(Process, self).on_entering(state)
         # Update the node attributes every time we enter a new state
         self.update_node_state(state)
-        if self._enable_persistence and not state.is_terminal():
+        if self._enable_persistence and not state.is_terminal() and not state.label == ProcessState.CREATED:
             self.call_soon(self.runner.persister.save_checkpoint, self)
 
     @override
@@ -212,6 +210,7 @@ class Process(plumpy.Process):
         except exceptions.ModificationNotAllowed:
             pass
 
+    @override
     def on_except(self, exc_info):
         """
         Log the exception by calling the report method with formatted stack trace from exception info object
@@ -227,15 +226,6 @@ class Process(plumpy.Process):
         super(Process, self).on_finish(result, successful)
         self.calc._set_finish_status(result)
 
-    @override
-    def on_fail(self, exc_info):
-        """
-        Format the exception info into a string and log it as an error
-        """
-        super(Process, self).on_fail(exc_info)
-        exception = traceback.format_exception(exc_info[0], exc_info[1], exc_info[2])
-        self.logger.error('{} failed:\n{}'.format(self.pid, ''.join(exception)))
-
     @override
     def on_output_emitting(self, output_port, value):
         """
diff --git a/aiida/work/rmq.py b/aiida/work/rmq.py
index 010558989..ed720f9d5 100644
--- a/aiida/work/rmq.py
+++ b/aiida/work/rmq.py
@@ -7,6 +7,7 @@ import plumpy.rmq
 
 from aiida.utils.serialize import serialize_data, deserialize_data
 from aiida.work.class_loader import CLASS_LOADER
+from aiida.work.persistence import AiiDAPersister
 
 __all__ = ['new_control_panel', 'new_blocking_control_panel', 'BlockingProcessControlPanel',
            'RemoteException', 'DeliveryFailed', 'ProcessLauncher']
@@ -154,14 +155,29 @@ class LaunchProcessAction(plumpy.LaunchProcessAction):
         super(LaunchProcessAction, self).__init__(*args, **kwargs)
 
 
-class ExecuteProcessAction(plumpy.ExecuteProcessAction):
-    def __init__(self, process_class, init_args=None, init_kwargs=None, nowait=False):
-        """
-        Calls through to the constructor of the plum ExecuteProcessAction while making sure that
-        any unstored nodes in the inputs are first stored and the data is then serialized
-        """
-        init_kwargs['inputs'] = store_and_serialize_inputs(init_kwargs['inputs'])
-        super(ExecuteProcessAction, self).__init__(process_class, init_args, init_kwargs, class_loader=CLASS_LOADER, nowait=nowait)
+class ExecuteProcessAction(plumpy.Action):
+
+    def __init__(self, process_class, init_args=None, init_kwargs=None, nowait=False, persister=None):
+        super(ExecuteProcessAction, self).__init__()
+        self._process_class = process_class
+        self._args = init_args or ()
+        self._kwargs = init_kwargs or {}
+        self._nowait = nowait
+
+        if persister is not None:
+            self._persister = persister
+        else:
+            self._persister = AiiDAPersister()
+
+    def execute(self, publisher):
+        proc = self._process_class(*self._args, **self._kwargs)
+        self._persister.save_checkpoint(proc)
+        proc.close()
+
+        continue_action = plumpy.ContinueProcessAction(proc.calc.pk, play=True)
+        continue_action.execute(publisher)
+
+        self.set_result(proc.calc.pk)
 
 
 class ProcessLauncher(plumpy.ProcessLauncher):
@@ -261,7 +277,7 @@ class BlockingProcessControlPanel(ProcessControlPanel):
         action = ExecuteProcessAction(process_class, init_args, init_kwargs, nowait=True)
         action.execute(self._communicator)
         self._communicator.await(action)
-        return action.get_launch_future().result()
+        return action.result()
 
     def execute_action(self, action):
         action.execute(self._communicator)
diff --git a/aiida/work/runners.py b/aiida/work/runners.py
index 0f22dc4e2..e94c22a44 100644
--- a/aiida/work/runners.py
+++ b/aiida/work/runners.py
@@ -203,6 +203,7 @@ class Runner(object):
         if self._rmq_submit:
             process = _create_process(process_class, self, input_args=args, input_kwargs=inputs)
             self.persister.save_checkpoint(process)
+            process.close()
             self.rmq.continue_process(process.pid)
             return process.calc
         else:
diff --git a/aiida/work/workchain.py b/aiida/work/workchain.py
index 5122cb9e2..62b1b2d0d 100644
--- a/aiida/work/workchain.py
+++ b/aiida/work/workchain.py
@@ -165,10 +165,10 @@ class WorkChain(processes.Process):
         """
         for awaitable in self._awaitables:
             if awaitable.target == AwaitableTarget.CALCULATION:
-                fn = functools.partial(self.on_calculation_finished, awaitable)
+                fn = functools.partial(self._run_task, self.on_calculation_finished, awaitable)
                 self.runner.call_on_calculation_finish(awaitable.pk, fn)
             elif awaitable.target == AwaitableTarget.WORKFLOW:
-                fn = functools.partial(self.on_legacy_workflow_finished, awaitable)
+                fn = functools.partial(self._run_task, self.on_legacy_workflow_finished, awaitable)
                 self.runner.call_on_legacy_workflow_finish(awaitable.pk, fn)
             else:
                 assert "invalid awaitable target '{}'".format(awaitable.target)
diff --git a/requirements.txt b/requirements.txt
index 5b319afd3..d37f4a82f 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -142,5 +142,5 @@ wcwidth==0.1.7
 Werkzeug==0.14.1
 wrapt==1.10.11
 yapf==0.19.0
--e git://github.com/muhrin/plumpy.git@349471363eb880d90f888d5dc338734e85871930#egg=plumpy
--e git://github.com/muhrin/kiwipy.git@07dbeac1e0a7de41638e3e212823ddd3ab63b7fb#egg=kiwipy
+-e git://github.com/muhrin/plumpy.git@2b628c331c286347fdd7ab12911ae82ad2146033#egg=plumpy
+-e git://github.com/muhrin/kiwipy.git@9ccea9b7ab719f042e0529239c47de3dba1cc249#egg=kiwipy
['aiida/cmdline/commands/calculation.py', 'aiida/work/persistence.py', 'aiida/work/processes.py', 'aiida/utils/serialize.py', 'aiida/work/launch.py', 'aiida/work/rmq.py', 'aiida/common/log.py', 'requirements.txt', 'aiida/work/runners.py', 'aiida/work/workchain.py', '.travis-data/workchains.py', 'aiida/work/job_processes.py', 'aiida/daemon/workflowmanager.py', '.travis.yml', 'aiida/backends/tests/utils/serialize.py', 'aiida/daemon/runner.py']