Follow up to executor example comments

rclpy/executors/examples_rclpy_executors/custom_executor.py

@@ -12,7 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import threading
+from concurrent.futures import ThreadPoolExecutor
+import multiprocessing
 
 from examples_rclpy_executors.listener import Listener
 from examples_rclpy_executors.talker import Talker
@@ -45,16 +46,14 @@ class PriorityExecutor(Executor):
     def __init__(self):
         super().__init__()
         self.high_priority_nodes = set()
-        self.low_priority_thread = None
+        self.hp_executor = ThreadPoolExecutor(max_workers=multiprocessing.cpu_count())
+        self.lp_executor = ThreadPoolExecutor(max_workers=1)
 
     def add_high_priority_node(self, node):
         self.high_priority_nodes.add(node)
         # add_node inherited
         self.add_node(node)
 
-    def can_run_low_priority(self):
-        return self.low_priority_thread is None or not self.low_priority_thread.is_alive()
-
     def spin_once(self, timeout_sec=None):
         """
         Execute a single callback, then return.
@@ -66,27 +65,16 @@ def spin_once(self, timeout_sec=None):
         :param timeout_sec: Seconds to wait. Block forever if None. Don't wait if <= 0
         :type timeout_sec: float or None
         """
-        # Wait only on high priority nodes if the low priority thread is taken.
-        # this avoids spinning rapidly when low priority callbacks are available but
-        # can't be acted on
-        nodes = self.high_priority_nodes
-        if self.can_run_low_priority():
-            # get_nodes returns all nodes added to executor
-            nodes = self.get_nodes()
-
         # wait_for_ready_callbacks yields callbacks that are ready to be executed
         try:
-            handler, group, node = next(self.wait_for_ready_callbacks(
-                timeout_sec=timeout_sec, nodes=nodes))
+            handler, group, node = next(self.wait_for_ready_callbacks(timeout_sec=timeout_sec))
         except StopIteration:
             pass
         else:
             if node in self.high_priority_nodes:
-                t = threading.Thread(target=handler)
-                t.start()
+                self.hp_executor.submit(handler)
             else:
-                self.low_priority_thread = threading.Thread(target=handler)
-                self.low_priority_thread.start()
+                self.lp_executor.submit(handler)
 
 
 def main(args=None):
@@ -97,11 +85,7 @@ def main(args=None):
         executor.add_node(Listener())
         executor.add_node(Talker())
         try:
-            # TODO(sloretz) use executor.spin() once guard conditions become available to users
-            while rclpy.ok():
-                # A timeout is used to make the executor periodically reevaluate if low priority
-                # nodes can be executed
-                executor.spin_once(timeout_sec=0.5)
+            executor.spin()
         finally:
             executor.shutdown()
     finally:

rclpy/executors/test/test_copyright.py

@@ -12,12 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
-
 from ament_copyright.main import main
 
 
 def test_copyright():
-    root_path = os.path.realpath(os.path.join(os.path.dirname(__file__), '..'))
-    rc = main(argv=[root_path])
+    # Test is called from package root
+    rc = main(argv=['.'])
     assert rc == 0, 'Found errors'

rclpy/executors/test/test_flake8.py

@@ -12,12 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
-
 from ament_flake8.main import main
 
 
 def test_flake8():
-    root_path = os.path.realpath(os.path.join(os.path.dirname(__file__), '..'))
-    rc = main(argv=[root_path])
+    # Test is called from package root
+    rc = main(argv=['.'])
     assert rc == 0, 'Found errors'

rclpy/executors/test/test_pep257.py

@@ -12,12 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
-
 from ament_pep257.main import main
 
 
 def test_pep257():
-    root_path = os.path.realpath(os.path.join(os.path.dirname(__file__), '..'))
-    rc = main(argv=[root_path])
+    # Test is called from package root
+    rc = main(argv=['.'])
     assert rc == 0, 'Found code style errors / warnings'