diff --git a/docs/userguide/monitoring.rst b/docs/userguide/monitoring.rst
index 3404862450..02b3177ca7 100644
--- a/docs/userguide/monitoring.rst
+++ b/docs/userguide/monitoring.rst
@@ -42,7 +42,6 @@ configuration. Here the `parsl.monitoring.MonitoringHub` is specified to use por
       ],
       monitoring=MonitoringHub(
           hub_address=address_by_hostname(),
-          hub_port=55055,
           monitoring_debug=False,
           resource_monitoring_interval=10,
       ),
diff --git a/parsl/configs/ASPIRE1.py b/parsl/configs/ASPIRE1.py
index 7792f15dba..017e1061d7 100644
--- a/parsl/configs/ASPIRE1.py
+++ b/parsl/configs/ASPIRE1.py
@@ -34,7 +34,6 @@
         ],
         monitoring=MonitoringHub(
             hub_address=address_by_interface('ib0'),
-            hub_port=55055,
             resource_monitoring_interval=10,
         ),
         strategy='simple',
diff --git a/parsl/executors/base.py b/parsl/executors/base.py
index a112b9eb00..fc97db89d3 100644
--- a/parsl/executors/base.py
+++ b/parsl/executors/base.py
@@ -5,7 +5,7 @@
 
 from typing_extensions import Literal, Self
 
-from parsl.monitoring.radios import MonitoringRadioSender
+from parsl.monitoring.radios.base import MonitoringRadioSender
 
 
 class ParslExecutor(metaclass=ABCMeta):
diff --git a/parsl/executors/high_throughput/executor.py b/parsl/executors/high_throughput/executor.py
index 882776cf5c..d3fa522619 100644
--- a/parsl/executors/high_throughput/executor.py
+++ b/parsl/executors/high_throughput/executor.py
@@ -331,6 +331,9 @@ def __init__(self,
             interchange_launch_cmd = DEFAULT_INTERCHANGE_LAUNCH_CMD
         self.interchange_launch_cmd = interchange_launch_cmd
 
+        self._result_queue_thread_exit = threading.Event()
+        self._result_queue_thread: Optional[threading.Thread] = None
+
     radio_mode = "htex"
     enable_mpi_mode: bool = False
     mpi_launcher: str = "mpiexec"
@@ -455,9 +458,11 @@ def _result_queue_worker(self):
         """
         logger.debug("Result queue worker starting")
 
-        while not self.bad_state_is_set:
+        while not self.bad_state_is_set and not self._result_queue_thread_exit.is_set():
             try:
-                msgs = self.incoming_q.get()
+                msgs = self.incoming_q.get(timeout_ms=self.poll_period)
+                if msgs is None:  # timeout
+                    continue
 
             except IOError as e:
                 logger.exception("Caught broken queue with exception code {}: {}".format(e.errno, e))
@@ -515,6 +520,8 @@ def _result_queue_worker(self):
                     else:
                         raise BadMessage("Message received with unknown type {}".format(msg['type']))
 
+        logger.info("Closing result ZMQ pipe")
+        self.incoming_q.close()
         logger.info("Result queue worker finished")
 
     def _start_local_interchange_process(self) -> None:
@@ -817,6 +824,8 @@ def shutdown(self, timeout: float = 10.0):
 
         logger.info("Attempting HighThroughputExecutor shutdown")
 
+        logger.info("Terminating interchange and result queue thread")
+        self._result_queue_thread_exit.set()
         self.interchange_proc.terminate()
         try:
             self.interchange_proc.wait(timeout=timeout)
@@ -841,6 +850,10 @@ def shutdown(self, timeout: float = 10.0):
             logger.info("Closing command client")
             self.command_client.close()
 
+        logger.info("Waiting for result queue thread exit")
+        if self._result_queue_thread:
+            self._result_queue_thread.join()
+
         logger.info("Finished HighThroughputExecutor shutdown attempt")
 
     def get_usage_information(self):
diff --git a/parsl/executors/high_throughput/interchange.py b/parsl/executors/high_throughput/interchange.py
index 88bb6c7156..12d3e07f31 100644
--- a/parsl/executors/high_throughput/interchange.py
+++ b/parsl/executors/high_throughput/interchange.py
@@ -20,7 +20,8 @@
 from parsl.executors.high_throughput.manager_record import ManagerRecord
 from parsl.executors.high_throughput.manager_selector import ManagerSelector
 from parsl.monitoring.message_type import MessageType
-from parsl.monitoring.radios import MonitoringRadioSender, ZMQRadioSender
+from parsl.monitoring.radios.base import MonitoringRadioSender
+from parsl.monitoring.radios.zmq import ZMQRadioSender
 from parsl.process_loggers import wrap_with_logs
 from parsl.serialize import serialize as serialize_object
 from parsl.utils import setproctitle
diff --git a/parsl/executors/high_throughput/zmq_pipes.py b/parsl/executors/high_throughput/zmq_pipes.py
index 54ed8c1da9..a7278cf067 100644
--- a/parsl/executors/high_throughput/zmq_pipes.py
+++ b/parsl/executors/high_throughput/zmq_pipes.py
@@ -206,12 +206,21 @@ def __init__(self, ip_address, port_range, cert_dir: Optional[str] = None):
         self.port = self.results_receiver.bind_to_random_port(tcp_url(ip_address),
                                                               min_port=port_range[0],
                                                               max_port=port_range[1])
+        self.poller = zmq.Poller()
+        self.poller.register(self.results_receiver, zmq.POLLIN)
 
-    def get(self):
+    def get(self, timeout_ms=None):
+        """Get a message from the queue, returning None if timeout expires
+        without a message. timeout is measured in milliseconds.
+        """
         logger.debug("Waiting for ResultsIncoming message")
-        m = self.results_receiver.recv_multipart()
-        logger.debug("Received ResultsIncoming message")
-        return m
+        socks = dict(self.poller.poll(timeout=timeout_ms))
+        if self.results_receiver in socks and socks[self.results_receiver] == zmq.POLLIN:
+            m = self.results_receiver.recv_multipart()
+            logger.debug("Received ResultsIncoming message")
+            return m
+        else:
+            return None
 
     def close(self):
         self.results_receiver.close()
diff --git a/parsl/monitoring/monitoring.py b/parsl/monitoring/monitoring.py
index e82c8fb688..3fbe5736ba 100644
--- a/parsl/monitoring/monitoring.py
+++ b/parsl/monitoring/monitoring.py
@@ -14,7 +14,7 @@
 
 from parsl.log_utils import set_file_logger
 from parsl.monitoring.errors import MonitoringHubStartError
-from parsl.monitoring.radios import MultiprocessingQueueRadioSender
+from parsl.monitoring.radios.multiprocessing import MultiprocessingQueueRadioSender
 from parsl.monitoring.router import router_starter
 from parsl.monitoring.types import TaggedMonitoringMessage
 from parsl.multiprocessing import ForkProcess, SizedQueue
diff --git a/parsl/monitoring/radios.py b/parsl/monitoring/radios.py
deleted file mode 100644
index 14dc046557..0000000000
--- a/parsl/monitoring/radios.py
+++ /dev/null
@@ -1,191 +0,0 @@
-import logging
-import os
-import pickle
-import socket
-import uuid
-from abc import ABCMeta, abstractmethod
-from multiprocessing.queues import Queue
-
-import zmq
-
-logger = logging.getLogger(__name__)
-
-
-class MonitoringRadioSender(metaclass=ABCMeta):
-    @abstractmethod
-    def send(self, message: object) -> None:
-        pass
-
-
-class FilesystemRadioSender(MonitoringRadioSender):
-    """A MonitoringRadioSender that sends messages over a shared filesystem.
-
-    The messsage directory structure is based on maildir,
-    https://en.wikipedia.org/wiki/Maildir
-
-    The writer creates a message in tmp/ and then when it is fully
-    written, moves it atomically into new/
-
-    The reader ignores tmp/ and only reads and deletes messages from
-    new/
-
-    This avoids a race condition of reading partially written messages.
-
-    This radio is likely to give higher shared filesystem load compared to
-    the UDP radio, but should be much more reliable.
-    """
-
-    def __init__(self, *, monitoring_url: str, timeout: int = 10, run_dir: str):
-        logger.info("filesystem based monitoring channel initializing")
-        self.base_path = f"{run_dir}/monitor-fs-radio/"
-        self.tmp_path = f"{self.base_path}/tmp"
-        self.new_path = f"{self.base_path}/new"
-
-        os.makedirs(self.tmp_path, exist_ok=True)
-        os.makedirs(self.new_path, exist_ok=True)
-
-    def send(self, message: object) -> None:
-        logger.info("Sending a monitoring message via filesystem")
-
-        unique_id = str(uuid.uuid4())
-
-        tmp_filename = f"{self.tmp_path}/{unique_id}"
-        new_filename = f"{self.new_path}/{unique_id}"
-        buffer = message
-
-        # this will write the message out then atomically
-        # move it into new/, so that a partially written
-        # file will never be observed in new/
-        with open(tmp_filename, "wb") as f:
-            pickle.dump(buffer, f)
-        os.rename(tmp_filename, new_filename)
-
-
-class HTEXRadioSender(MonitoringRadioSender):
-
-    def __init__(self, monitoring_url: str, timeout: int = 10):
-        """
-        Parameters
-        ----------
-
-        monitoring_url : str
-            URL of the form <scheme>://<IP>:<PORT>
-        timeout : int
-            timeout, default=10s
-        """
-        logger.info("htex-based monitoring channel initialising")
-
-    def send(self, message: object) -> None:
-        """ Sends a message to the UDP receiver
-
-        Parameter
-        ---------
-
-        message: object
-            Arbitrary pickle-able object that is to be sent
-
-        Returns:
-            None
-        """
-
-        import parsl.executors.high_throughput.monitoring_info
-
-        result_queue = parsl.executors.high_throughput.monitoring_info.result_queue
-
-        # this message needs to go in the result queue tagged so that it is treated
-        # i) as a monitoring message by the interchange, and then further more treated
-        # as a RESOURCE_INFO message when received by monitoring (rather than a NODE_INFO
-        # which is the implicit default for messages from the interchange)
-
-        # for the interchange, the outer wrapper, this needs to be a dict:
-
-        interchange_msg = {
-            'type': 'monitoring',
-            'payload': message
-        }
-
-        if result_queue:
-            result_queue.put(pickle.dumps(interchange_msg))
-        else:
-            logger.error("result_queue is uninitialized - cannot put monitoring message")
-
-        return
-
-
-class UDPRadioSender(MonitoringRadioSender):
-
-    def __init__(self, monitoring_url: str, timeout: int = 10):
-        """
-        Parameters
-        ----------
-
-        monitoring_url : str
-            URL of the form <scheme>://<IP>:<PORT>
-        timeout : int
-            timeout, default=10s
-        """
-        self.monitoring_url = monitoring_url
-        self.sock_timeout = timeout
-        try:
-            self.scheme, self.ip, port = (x.strip('/') for x in monitoring_url.split(':'))
-            self.port = int(port)
-        except Exception:
-            raise Exception("Failed to parse monitoring url: {}".format(monitoring_url))
-
-        self.sock = socket.socket(socket.AF_INET,
-                                  socket.SOCK_DGRAM,
-                                  socket.IPPROTO_UDP)  # UDP
-        self.sock.settimeout(self.sock_timeout)
-
-    def send(self, message: object) -> None:
-        """ Sends a message to the UDP receiver
-
-        Parameter
-        ---------
-
-        message: object
-            Arbitrary pickle-able object that is to be sent
-
-        Returns:
-            None
-        """
-        try:
-            buffer = pickle.dumps(message)
-        except Exception:
-            logging.exception("Exception during pickling", exc_info=True)
-            return
-
-        try:
-            self.sock.sendto(buffer, (self.ip, self.port))
-        except socket.timeout:
-            logging.error("Could not send message within timeout limit")
-            return
-        return
-
-
-class MultiprocessingQueueRadioSender(MonitoringRadioSender):
-    """A monitoring radio which connects over a multiprocessing Queue.
-    This radio is intended to be used on the submit side, where components
-    in the submit process, or processes launched by multiprocessing, will have
-    access to a Queue shared with the monitoring database code (bypassing the
-    monitoring router).
-    """
-    def __init__(self, queue: Queue) -> None:
-        self.queue = queue
-
-    def send(self, message: object) -> None:
-        self.queue.put(message)
-
-
-class ZMQRadioSender(MonitoringRadioSender):
-    """A monitoring radio which connects over ZMQ. This radio is not
-    thread-safe, because its use of ZMQ is not thread-safe.
-    """
-
-    def __init__(self, hub_address: str, hub_zmq_port: int) -> None:
-        self._hub_channel = zmq.Context().socket(zmq.DEALER)
-        self._hub_channel.set_hwm(0)
-        self._hub_channel.connect(f"tcp://{hub_address}:{hub_zmq_port}")
-
-    def send(self, message: object) -> None:
-        self._hub_channel.send_pyobj(message)
diff --git a/parsl/tests/test_channels/__init__.py b/parsl/monitoring/radios/__init__.py
similarity index 100%
rename from parsl/tests/test_channels/__init__.py
rename to parsl/monitoring/radios/__init__.py
diff --git a/parsl/monitoring/radios/base.py b/parsl/monitoring/radios/base.py
new file mode 100644
index 0000000000..2bb799f256
--- /dev/null
+++ b/parsl/monitoring/radios/base.py
@@ -0,0 +1,13 @@
+import logging
+from abc import ABCMeta, abstractmethod
+from typing import Optional
+
+_db_manager_excepts: Optional[Exception]
+
+logger = logging.getLogger(__name__)
+
+
+class MonitoringRadioSender(metaclass=ABCMeta):
+    @abstractmethod
+    def send(self, message: object) -> None:
+        pass
diff --git a/parsl/monitoring/radios/filesystem.py b/parsl/monitoring/radios/filesystem.py
new file mode 100644
index 0000000000..accff87d36
--- /dev/null
+++ b/parsl/monitoring/radios/filesystem.py
@@ -0,0 +1,52 @@
+import logging
+import os
+import pickle
+import uuid
+
+from parsl.monitoring.radios.base import MonitoringRadioSender
+
+logger = logging.getLogger(__name__)
+
+
+class FilesystemRadioSender(MonitoringRadioSender):
+    """A MonitoringRadioSender that sends messages over a shared filesystem.
+
+    The messsage directory structure is based on maildir,
+    https://en.wikipedia.org/wiki/Maildir
+
+    The writer creates a message in tmp/ and then when it is fully
+    written, moves it atomically into new/
+
+    The reader ignores tmp/ and only reads and deletes messages from
+    new/
+
+    This avoids a race condition of reading partially written messages.
+
+    This radio is likely to give higher shared filesystem load compared to
+    the UDP radio, but should be much more reliable.
+    """
+
+    def __init__(self, *, monitoring_url: str, timeout: int = 10, run_dir: str):
+        logger.info("filesystem based monitoring channel initializing")
+        self.base_path = f"{run_dir}/monitor-fs-radio/"
+        self.tmp_path = f"{self.base_path}/tmp"
+        self.new_path = f"{self.base_path}/new"
+
+        os.makedirs(self.tmp_path, exist_ok=True)
+        os.makedirs(self.new_path, exist_ok=True)
+
+    def send(self, message: object) -> None:
+        logger.info("Sending a monitoring message via filesystem")
+
+        unique_id = str(uuid.uuid4())
+
+        tmp_filename = f"{self.tmp_path}/{unique_id}"
+        new_filename = f"{self.new_path}/{unique_id}"
+        buffer = message
+
+        # this will write the message out then atomically
+        # move it into new/, so that a partially written
+        # file will never be observed in new/
+        with open(tmp_filename, "wb") as f:
+            pickle.dump(buffer, f)
+        os.rename(tmp_filename, new_filename)
diff --git a/parsl/monitoring/radios/htex.py b/parsl/monitoring/radios/htex.py
new file mode 100644
index 0000000000..bdb893b303
--- /dev/null
+++ b/parsl/monitoring/radios/htex.py
@@ -0,0 +1,57 @@
+import logging
+import pickle
+
+from parsl.monitoring.radios.base import MonitoringRadioSender
+
+logger = logging.getLogger(__name__)
+
+
+class HTEXRadioSender(MonitoringRadioSender):
+
+    def __init__(self, monitoring_url: str, timeout: int = 10):
+        """
+        Parameters
+        ----------
+
+        monitoring_url : str
+            URL of the form <scheme>://<IP>:<PORT>
+        timeout : int
+            timeout, default=10s
+        """
+        logger.info("htex-based monitoring channel initialising")
+
+    def send(self, message: object) -> None:
+        """ Sends a message to the UDP receiver
+
+        Parameter
+        ---------
+
+        message: object
+            Arbitrary pickle-able object that is to be sent
+
+        Returns:
+            None
+        """
+
+        import parsl.executors.high_throughput.monitoring_info
+
+        result_queue = parsl.executors.high_throughput.monitoring_info.result_queue
+
+        # this message needs to go in the result queue tagged so that it is treated
+        # i) as a monitoring message by the interchange, and then further more treated
+        # as a RESOURCE_INFO message when received by monitoring (rather than a NODE_INFO
+        # which is the implicit default for messages from the interchange)
+
+        # for the interchange, the outer wrapper, this needs to be a dict:
+
+        interchange_msg = {
+            'type': 'monitoring',
+            'payload': message
+        }
+
+        if result_queue:
+            result_queue.put(pickle.dumps(interchange_msg))
+        else:
+            logger.error("result_queue is uninitialized - cannot put monitoring message")
+
+        return
diff --git a/parsl/monitoring/radios/multiprocessing.py b/parsl/monitoring/radios/multiprocessing.py
new file mode 100644
index 0000000000..6274bbfca8
--- /dev/null
+++ b/parsl/monitoring/radios/multiprocessing.py
@@ -0,0 +1,17 @@
+from multiprocessing.queues import Queue
+
+from parsl.monitoring.radios.base import MonitoringRadioSender
+
+
+class MultiprocessingQueueRadioSender(MonitoringRadioSender):
+    """A monitoring radio which connects over a multiprocessing Queue.
+    This radio is intended to be used on the submit side, where components
+    in the submit process, or processes launched by multiprocessing, will have
+    access to a Queue shared with the monitoring database code (bypassing the
+    monitoring router).
+    """
+    def __init__(self, queue: Queue) -> None:
+        self.queue = queue
+
+    def send(self, message: object) -> None:
+        self.queue.put(message)
diff --git a/parsl/monitoring/radios/udp.py b/parsl/monitoring/radios/udp.py
new file mode 100644
index 0000000000..f2a652e9ac
--- /dev/null
+++ b/parsl/monitoring/radios/udp.py
@@ -0,0 +1,56 @@
+import logging
+import pickle
+import socket
+
+from parsl.monitoring.radios.base import MonitoringRadioSender
+
+
+class UDPRadioSender(MonitoringRadioSender):
+
+    def __init__(self, monitoring_url: str, timeout: int = 10):
+        """
+        Parameters
+        ----------
+
+        monitoring_url : str
+            URL of the form <scheme>://<IP>:<PORT>
+        timeout : int
+            timeout, default=10s
+        """
+        self.monitoring_url = monitoring_url
+        self.sock_timeout = timeout
+        try:
+            self.scheme, self.ip, port = (x.strip('/') for x in monitoring_url.split(':'))
+            self.port = int(port)
+        except Exception:
+            raise Exception("Failed to parse monitoring url: {}".format(monitoring_url))
+
+        self.sock = socket.socket(socket.AF_INET,
+                                  socket.SOCK_DGRAM,
+                                  socket.IPPROTO_UDP)  # UDP
+        self.sock.settimeout(self.sock_timeout)
+
+    def send(self, message: object) -> None:
+        """ Sends a message to the UDP receiver
+
+        Parameter
+        ---------
+
+        message: object
+            Arbitrary pickle-able object that is to be sent
+
+        Returns:
+            None
+        """
+        try:
+            buffer = pickle.dumps(message)
+        except Exception:
+            logging.exception("Exception during pickling", exc_info=True)
+            return
+
+        try:
+            self.sock.sendto(buffer, (self.ip, self.port))
+        except socket.timeout:
+            logging.error("Could not send message within timeout limit")
+            return
+        return
diff --git a/parsl/monitoring/radios/zmq.py b/parsl/monitoring/radios/zmq.py
new file mode 100644
index 0000000000..397c943568
--- /dev/null
+++ b/parsl/monitoring/radios/zmq.py
@@ -0,0 +1,17 @@
+import zmq
+
+from parsl.monitoring.radios.base import MonitoringRadioSender
+
+
+class ZMQRadioSender(MonitoringRadioSender):
+    """A monitoring radio which connects over ZMQ. This radio is not
+    thread-safe, because its use of ZMQ is not thread-safe.
+    """
+
+    def __init__(self, hub_address: str, hub_zmq_port: int) -> None:
+        self._hub_channel = zmq.Context().socket(zmq.DEALER)
+        self._hub_channel.set_hwm(0)
+        self._hub_channel.connect(f"tcp://{hub_address}:{hub_zmq_port}")
+
+    def send(self, message: object) -> None:
+        self._hub_channel.send_pyobj(message)
diff --git a/parsl/monitoring/remote.py b/parsl/monitoring/remote.py
index d72b54dc3c..530b39f935 100644
--- a/parsl/monitoring/remote.py
+++ b/parsl/monitoring/remote.py
@@ -7,12 +7,10 @@
 from typing import Any, Callable, Dict, List, Sequence, Tuple
 
 from parsl.monitoring.message_type import MessageType
-from parsl.monitoring.radios import (
-    FilesystemRadioSender,
-    HTEXRadioSender,
-    MonitoringRadioSender,
-    UDPRadioSender,
-)
+from parsl.monitoring.radios.base import MonitoringRadioSender
+from parsl.monitoring.radios.filesystem import FilesystemRadioSender
+from parsl.monitoring.radios.htex import HTEXRadioSender
+from parsl.monitoring.radios.udp import UDPRadioSender
 from parsl.multiprocessing import ForkProcess
 from parsl.process_loggers import wrap_with_logs
 
diff --git a/parsl/monitoring/router.py b/parsl/monitoring/router.py
index 04e7480a7a..0926712c36 100644
--- a/parsl/monitoring/router.py
+++ b/parsl/monitoring/router.py
@@ -14,7 +14,7 @@
 import zmq
 
 from parsl.log_utils import set_file_logger
-from parsl.monitoring.radios import MultiprocessingQueueRadioSender
+from parsl.monitoring.radios.multiprocessing import MultiprocessingQueueRadioSender
 from parsl.monitoring.types import TaggedMonitoringMessage
 from parsl.process_loggers import wrap_with_logs
 from parsl.utils import setproctitle
diff --git a/parsl/tests/configs/htex_local_alternate.py b/parsl/tests/configs/htex_local_alternate.py
index ffff814b43..cc69d56186 100644
--- a/parsl/tests/configs/htex_local_alternate.py
+++ b/parsl/tests/configs/htex_local_alternate.py
@@ -60,7 +60,6 @@ def fresh_config():
         retries=2,
         monitoring=MonitoringHub(
                         hub_address="localhost",
-                        hub_port=55055,
                         monitoring_debug=False,
                         resource_monitoring_interval=1,
         ),
diff --git a/parsl/tests/configs/local_threads_monitoring.py b/parsl/tests/configs/local_threads_monitoring.py
index 81b9095285..9f105af25d 100644
--- a/parsl/tests/configs/local_threads_monitoring.py
+++ b/parsl/tests/configs/local_threads_monitoring.py
@@ -5,7 +5,6 @@
 config = Config(executors=[ThreadPoolExecutor(label='threads', max_threads=4)],
                 monitoring=MonitoringHub(
                     hub_address="localhost",
-                    hub_port=55055,
                     resource_monitoring_interval=3,
                 )
                 )
diff --git a/parsl/tests/manual_tests/test_udp_simple.py b/parsl/tests/manual_tests/test_udp_simple.py
index 8de257d8fa..847f8803bf 100644
--- a/parsl/tests/manual_tests/test_udp_simple.py
+++ b/parsl/tests/manual_tests/test_udp_simple.py
@@ -15,7 +15,6 @@ def local_setup():
         ],
         monitoring=MonitoringHub(
             hub_address="127.0.0.1",
-            hub_port=55055,
             logging_level=logging.INFO,
             resource_monitoring_interval=10))
 
diff --git a/parsl/tests/test_channels/test_local_channel.py b/parsl/tests/test_channels/test_local_channel.py
deleted file mode 100644
index 39094d0235..0000000000
--- a/parsl/tests/test_channels/test_local_channel.py
+++ /dev/null
@@ -1,18 +0,0 @@
-import pytest
-
-from parsl.utils import execute_wait
-
-
-@pytest.mark.local
-def test_env():
-    ''' Regression testing for issue #27
-    '''
-
-    rc, stdout, stderr = execute_wait("env", 1)
-
-    stdout = stdout.split('\n')
-    x = [s for s in stdout if s.startswith("PATH=")]
-    assert x, "PATH not found"
-
-    x = [s for s in stdout if s.startswith("HOME=")]
-    assert x, "HOME not found"
diff --git a/parsl/tests/test_monitoring/test_htex_init_blocks_vs_monitoring.py b/parsl/tests/test_monitoring/test_htex_init_blocks_vs_monitoring.py
index fabb172842..67db42671c 100644
--- a/parsl/tests/test_monitoring/test_htex_init_blocks_vs_monitoring.py
+++ b/parsl/tests/test_monitoring/test_htex_init_blocks_vs_monitoring.py
@@ -35,7 +35,6 @@ def fresh_config(run_dir, strategy, db_url):
         strategy_period=0.1,
         monitoring=MonitoringHub(
                         hub_address="localhost",
-                        hub_port=55055,
                         logging_endpoint=db_url
         )
     )
diff --git a/parsl/tests/test_monitoring/test_stdouterr.py b/parsl/tests/test_monitoring/test_stdouterr.py
index d1817164c0..8e1935045f 100644
--- a/parsl/tests/test_monitoring/test_stdouterr.py
+++ b/parsl/tests/test_monitoring/test_stdouterr.py
@@ -37,7 +37,6 @@ def fresh_config(run_dir):
         strategy_period=0.1,
         monitoring=MonitoringHub(
                         hub_address="localhost",
-                        hub_port=55055,
         )
     )
 
diff --git a/parsl/tests/test_channels/test_large_output.py b/parsl/tests/test_utils/test_execute_wait.py
similarity index 51%
rename from parsl/tests/test_channels/test_large_output.py
rename to parsl/tests/test_utils/test_execute_wait.py
index 0558e600c7..44488c239c 100644
--- a/parsl/tests/test_channels/test_large_output.py
+++ b/parsl/tests/test_utils/test_execute_wait.py
@@ -4,9 +4,24 @@
 
 
 @pytest.mark.local
-def test_local_large_output_2210():
+def test_env():
+    ''' Regression testing for issue #27
+    '''
+
+    rc, stdout, stderr = execute_wait("env", 1)
+
+    stdout = stdout.split('\n')
+    x = [s for s in stdout if s.startswith("PATH=")]
+    assert x, "PATH not found"
+
+    x = [s for s in stdout if s.startswith("HOME=")]
+    assert x, "HOME not found"
+
+
+@pytest.mark.local
+def test_large_output_2210():
     """Regression test for #2210.
-    The local channel was hanging if the specified command gave too
+    execute_wait was hanging if the specified command gave too
     much output, due to a race condition between process exiting and
     pipes filling up.
     """