Merge "Fix periodic tasks not running after pickle/unpickle"

ironic/conductor/base_manager.py

@@ -71,12 +71,18 @@ class BaseConductorManager(object):
         ``_shutdown`` is a threading.Event (contains ``_thread.lock``);
         ``sensors_notifier`` and ``dbapi`` may contain unpicklable
         transport objects and connection-pool locks respectively.
-        All three are recreated during startup.
+        ``_executor``, ``_reserved_executor``, ``_periodic_tasks``, and
+        ``_periodic_tasks_worker`` contain thread pools and futures.
+        All are recreated during startup.
         """
         state = self.__dict__.copy()
         state['_shutdown'] = self._shutdown.is_set()
         state.pop('sensors_notifier', None)
         state.pop('dbapi', None)
+        state.pop('_executor', None)
+        state.pop('_reserved_executor', None)
+        state.pop('_periodic_tasks', None)
+        state.pop('_periodic_tasks_worker', None)
         return state
 
     def __setstate__(self, state):
@@ -87,6 +93,10 @@ class BaseConductorManager(object):
             self._shutdown.set()
         self.sensors_notifier = None  # recreated in init_host()
         self.dbapi = None  # recreated in prepare_host()/init_host()
+        self._executor = None  # recreated in prepare_host()
+        self._reserved_executor = None  # recreated in prepare_host()
+        self._periodic_tasks = None  # recreated in init_host()
+        self._periodic_tasks_worker = None  # recreated in init_host()
 
     def prepare_host(self):
         """Prepares host for initialization

ironic/tests/unit/conductor/test_rpc_service.py

@@ -323,6 +323,13 @@ class TestRPCService(db_base.DbTestCase):
         self.assertEqual(self.rpc_svc.manager.topic, restored.manager.topic)
         # threading.Event should be recreated as unset
         self.assertFalse(restored.manager._shutdown.is_set())
+        # Thread pool executors and periodic tasks should be excluded
+        # from pickle and set to None after unpickling (to be recreated
+        # during startup)
+        self.assertIsNone(restored.manager._executor)
+        self.assertIsNone(restored.manager._reserved_executor)
+        self.assertIsNone(restored.manager._periodic_tasks)
+        self.assertIsNone(restored.manager._periodic_tasks_worker)
 
     def test_getstate_includes_argv(self):
         """__getstate__ saves sys.argv so __setstate__ can restore CONF."""
@@ -333,6 +340,25 @@ class TestRPCService(db_base.DbTestCase):
         # tg must be excluded (contains un-picklable threading objects)
         self.assertNotIn('tg', state)
 
+    def test_manager_getstate_excludes_unpicklable(self):
+        """Manager __getstate__ excludes thread pools and periodic tasks.
+
+        Thread pool executors and periodic task objects contain threading
+        primitives that cannot be pickled. Verify they are excluded from
+        the pickled state and will be recreated during startup.
+        """
+        state = self.rpc_svc.manager.__getstate__()
+        # Threading objects must be excluded
+        self.assertNotIn('_executor', state)
+        self.assertNotIn('_reserved_executor', state)
+        self.assertNotIn('_periodic_tasks', state)
+        self.assertNotIn('_periodic_tasks_worker', state)
+        self.assertNotIn('sensors_notifier', state)
+        self.assertNotIn('dbapi', state)
+        # _shutdown should be converted to bool
+        self.assertIn('_shutdown', state)
+        self.assertIsInstance(state['_shutdown'], bool)
+
     @mock.patch('ironic.common.service.prepare_command', autospec=True)
     def test_setstate_calls_prepare_command(self, mock_prepare):
         """__setstate__ re-configures CONF when _argv is present.