Use vf_share_layers=False for PPO; add includes to ray_submit

Can disable COMET summary print.

Always use gymnasium SYNC envs when using GYM_SYNC_VECTOR_MODE

Author: Daraan

ray_submit.py

@@ -943,7 +943,6 @@ async def monitor_job_statuses(
         jobs_tracked_left = jobs_tracked.copy()
         final_states: dict[str, JobStatus] = {}
         failed_jobs: dict[str, tuple[str, str]] = {}
-        global running_jobs_count
 
         while jobs_tracked_left or (pending_submissions and len(pending_submissions) > 0):
             print("-" * 80)
@@ -1243,7 +1242,13 @@ def collect_restore_jobs(yaml_file: str, excludes: Sequence[str] = ()) -> dict[s
         "--excludes",
         nargs="+",
         default=[],
-        help="Exclude submissions whose entrypoint contains any of these patterns.",
+        help="Exclude submissions whose entrypoint or submission_id contains any of these patterns.",
+    )
+    parser.add_argument(
+        "--includes",
+        nargs="+",
+        default=[],
+        help="Only include submissions whose entrypoint or submission_id contains any of these patterns.",
     )
     parser.add_argument(
         "--include-running",
@@ -1379,6 +1384,12 @@ def collect_restore_jobs(yaml_file: str, excludes: Sequence[str] = ()) -> dict[s
                 excludes=args.excludes,
                 include_running=args.include_running,
             )
+        if args.includes:
+            submissions_dict = {
+                job_id: settings
+                for job_id, settings in submissions_dict.items()
+                if any(inc in settings.get("entrypoint", "") or inc in job_id for inc in args.includes)
+            }
         submissions = list(submissions_dict.items())
 
         if not args.test and args.max_jobs is not None and args.max_jobs > 0:
@@ -1438,7 +1449,7 @@ def submit_next():
                 )
                 args.max_jobs = max(1, running_jobs_count)
                 return False
-            while pending_submissions:
+            while pending_submissions and running_jobs_count < args.max_jobs:
                 next_job_id, next_settings = pending_submissions.pop(0)
                 submission_id_out = submit_single_job(next_job_id, next_settings, args, track_for_run_id=False)
                 if submission_id_out:

ray_utilities/callbacks/comet.py

@@ -196,17 +196,27 @@ def get_default_workspace() -> str:
 
 
 @contextmanager
-def _catch_comet_offline_logger():
+def _catch_comet_offline_logger(*, disable_print: bool | None = None):
     """Context manager to temporarily add a stream handler to the comet_ml logger and yield the log stream."""
     from comet_ml.offline import LOGGER as COMET_LOGGER
 
     log_stream = io.StringIO()
     handler = logging.StreamHandler(log_stream)
+    if disable_print is True or (disable_print is None and "DISABLE_COMET_SUMMARY_PRINT" in os.environ):
+        # disable existing loggers
+        comet_handlers = COMET_LOGGER.handlers
+        levels = {}
+        for handler in comet_handlers:
+            levels[handler] = handler.level
+            COMET_LOGGER.setLevel(logging.WARNING)
     COMET_LOGGER.addHandler(handler)
     try:
         yield log_stream
     finally:
         COMET_LOGGER.removeHandler(handler)
+        # Reset levels
+        for handler in COMET_LOGGER.handlers:
+            handler.setLevel(levels[handler])
 
 
 def comet_upload_offline_experiments(tracker: Optional[CometArchiveTracker] = None):

ray_utilities/callbacks/tuner/adv_comet_callback.py

@@ -608,6 +608,7 @@ def log_trial_end(self, trial: "Trial", failed: bool = False):  # noqa: FBT001,
                     self._processes.append(process)
 
     def on_experiment_end(self, trials: list[Trial], **info):
+        # Suppress output
         super().on_experiment_end(trials, **info)
         # if there are any trials left (maybe because of an error and log_trial_end was not called)
         for experiment in self._trial_experiments.values():

ray_utilities/config/_tuner_callbacks_setup.py

@@ -211,6 +211,9 @@ def create_comet_logger(
             auto_metric_step_rate=10,  # How often batch metrics are logged. Default 10
             auto_histogram_epoch_rate=1,  # How often histograms are logged. Default 1
             parse_args=False,
+            display_summary_level=(
+                0 if args.command_str == "pbt" and (not args.comet or "upload" not in args.comet) else 1
+            ),
             log_git_metadata=not args.test,  # disabled by rllib; might cause throttling -> needed for Reproduce button
             log_git_patch=False,
             log_graph=False,  # computation graph, Default True

ray_utilities/config/create_algorithm.py

@@ -24,6 +24,7 @@
 from __future__ import annotations
 
 import logging
+import os
 import sys
 from typing import TYPE_CHECKING, Any, Callable, Final, Literal, Optional, TypeVar, cast
 
@@ -207,9 +208,12 @@ def create_algorithm_config(
         # increase time in case of debugging the sampler
         config.env_runners(sample_timeout_s=1000)
     try:
+        vector_mode = VectorizeMode.ASYNC if args["num_envs_per_env_runner"] > 1 else VectorizeMode.SYNC
+        if "GYM_SYNC_VECTOR_MODE" in os.environ:
+            vector_mode = VectorizeMode.SYNC
         config.env_runners(
             # experimental
-            gym_env_vectorize_mode=(VectorizeMode.ASYNC if args["num_envs_per_env_runner"] > 1 else VectorizeMode.SYNC),  # pyright: ignore[reportArgumentType]
+            gym_env_vectorize_mode=vector_mode,  # pyright: ignore[reportArgumentType]
         )
     except TypeError:
         logger.error("Current ray version does not support AlgorithmConfig.env_runners(gym_env_vectorize_mode=...)")
@@ -405,7 +409,7 @@ def create_algorithm_config(
             use_kl_loss=args.get("use_kl_loss", False) or (bool(args["tune"]) and "kl_coeff" in args["tune"]),
             use_gae=True,  # Must be true to use "truncate_episodes"
             # As long as this is not fully deprecated keep it here.
-            vf_share_layers=args.get("vf_share_layers", True),
+            vf_share_layers=args.get("vf_share_layers", False),
         )
     elif algorithm_type == "dqn":
         assert isinstance(config, DQNConfig)
@@ -446,7 +450,7 @@ def create_algorithm_config(
         # Workaround for https://github.com/ray-project/ray/issues/58715 avoid no sync mishaps
         from ray.rllib.core.rl_module.default_model_config import DefaultModelConfig  # noqa: PLC0415
 
-        model_config["vf_share_layers"] = DefaultModelConfig.vf_share_layers
+        model_config["vf_share_layers"] = DefaultModelConfig.vf_share_layers if algorithm_type != "ppo" else False
     # Create a single agent RL module spec.
     # Note: legacy keys are updated below
     module_spec = RLModuleSpec(

ray_utilities/config/parser/mlp_argument_parser.py

@@ -120,7 +120,7 @@ class SimpleMLPParser(Tap):
     the default initializer defined by `torch` is used.
     """
 
-    vf_share_layers: bool = True
+    vf_share_layers: bool = False
     """
     Whether encoder layers (defined by `fcnet_hiddens` or `conv_filters`) should be
     shared between policy- and value function.
@@ -139,6 +139,10 @@ def process_args(self) -> None:
         # flatten list of lists
         self.fcnet_hiddens = tree.flatten(self.fcnet_hiddens)
         self.head_fcnet_hiddens = tree.flatten(self.head_fcnet_hiddens)
+        import sys  # XXX
+
+        if "--no_vf_share_layers" in sys.argv:
+            assert not self.vf_share_layers
         super().process_args()