Merge remote-tracking branch 'origin/master' into benc-remove-channels

parsl/executors/taskvine/manager.py

@@ -44,11 +44,17 @@ def _set_manager_attributes(m, config):
     # Enable peer transfer feature between workers if specified
     if config.enable_peer_transfers:
         m.enable_peer_transfers()
+    else:
+        m.disable_peer_transfers()
 
     # Set catalog report to parsl if project name exists
     if m.name:
         m.set_property("framework", "parsl")
 
+    if config.tune_parameters is not None:
+        for k, v in config.tune_parameters.items():
+            m.tune(k, v)
+
 
 def _prepare_environment_serverless(manager_config, env_cache_dir, poncho_create_script):
     # Return path to a packaged poncho environment

parsl/executors/taskvine/manager_config.py

@@ -156,6 +156,10 @@ class TaskVineManagerConfig:
         Directory to store TaskVine logging facilities.
         Default is None, in which all TaskVine logs will be contained
         in the Parsl logging directory.
+
+    tune_parameters: Optional[dict]
+        Extended vine_tune parameters, expressed in a dictionary
+        by { 'tune-parameter' : value }.
     """
 
     # Connection and communication settings
@@ -181,6 +185,7 @@ class TaskVineManagerConfig:
     autocategory: bool = True
     enable_peer_transfers: bool = True
     wait_for_workers: Optional[int] = None
+    tune_parameters: Optional[dict] = None
 
     # Logging settings
     vine_log_dir: Optional[str] = None

parsl/monitoring/monitoring.py

@@ -258,7 +258,7 @@ def close(self) -> None:
 
 
 @wrap_with_logs
-def filesystem_receiver(logdir: str, q: "queue.Queue[TaggedMonitoringMessage]", run_dir: str) -> None:
+def filesystem_receiver(logdir: str, q: Queue[TaggedMonitoringMessage], run_dir: str) -> None:
     logger = set_file_logger("{}/monitoring_filesystem_radio.log".format(logdir),
                              name="monitoring_filesystem_radio",
                              level=logging.INFO)