[RLlib] Fix "seed" setting to work in all frameworks and w/ all CUDA versions.
diff --git a/rllib/BUILD b/rllib/BUILD
index 1acd31b8dc0eb..4dc3148cb159a 100644
--- a/rllib/BUILD
+++ b/rllib/BUILD
@@ -2075,6 +2075,33 @@ py_test(
     args = ["--stop-iters=2", "--num-cpus=4"]
 )
 
+py_test(
+    name = "examples/deterministic_training_tf",
+    main = "examples/deterministic_training.py",
+    tags = ["examples", "examples_D"],
+    size = "medium",
+    srcs = ["examples/deterministic_training.py"],
+    args = ["--as-test", "--stop-iters=1", "--framework=tf"]
+)
+
+py_test(
+    name = "examples/deterministic_training_tf2",
+    main = "examples/deterministic_training.py",
+    tags = ["examples", "examples_D"],
+    size = "medium",
+    srcs = ["examples/deterministic_training.py"],
+    args = ["--as-test", "--stop-iters=1", "--framework=tf2"]
+)
+
+py_test(
+    name = "examples/deterministic_training_torch",
+    main = "examples/deterministic_training.py",
+    tags = ["examples", "examples_D"],
+    size = "medium",
+    srcs = ["examples/deterministic_training.py"],
+    args = ["--as-test", "--stop-iters=1", "--framework=torch"]
+)
+
 py_test(
     name = "examples/eager_execution",
     tags = ["examples", "examples_E"],
diff --git a/rllib/evaluation/rollout_worker.py b/rllib/evaluation/rollout_worker.py
index 680587589ea85..47d8b62680b70 100644
--- a/rllib/evaluation/rollout_worker.py
+++ b/rllib/evaluation/rollout_worker.py
@@ -48,7 +48,7 @@
 
 if TYPE_CHECKING:
     from ray.rllib.evaluation.observation_function import ObservationFunction
-    from ray.rllib.agents.callbacks import DefaultCallbacks
+    from ray.rllib.agents.callbacks import DefaultCallbacks  # noqa
 
 # Generic type var for foreach_* methods.
 T = TypeVar("T")
@@ -356,7 +356,7 @@ def gen_rollouts():
         if callbacks:
             self.callbacks: "DefaultCallbacks" = callbacks()
         else:
-            from ray.rllib.agents.callbacks import DefaultCallbacks
+            from ray.rllib.agents.callbacks import DefaultCallbacks  # noqa
             self.callbacks: DefaultCallbacks = DefaultCallbacks()
         self.worker_index: int = worker_index
         self.num_workers: int = num_workers
@@ -482,20 +482,39 @@ def make_env(vector_index):
         self.policy_map: Dict[PolicyID, Policy] = None
         self.preprocessors: Dict[PolicyID, Preprocessor] = None
 
-        # set numpy and python seed
+        # Set Python random, numpy, env, and torch/tf seeds.
         if seed is not None:
-            np.random.seed(seed)
+            # Python random module.
             random.seed(seed)
+            # Numpy.
+            np.random.seed(seed)
+            # Gym.env.
             if not hasattr(self.env, "seed"):
                 logger.info("Env doesn't support env.seed(): {}".format(
                     self.env))
             else:
                 self.env.seed(seed)
-            try:
-                assert torch is not None
+
+            # Torch.
+            if torch and policy_config.get("framework") == "torch":
                 torch.manual_seed(seed)
-            except AssertionError:
-                logger.info("Could not seed torch")
+                # See https://github.com/pytorch/pytorch/issues/47672.
+                cuda_version = torch.version.cuda
+                if cuda_version is not None and float(
+                        torch.version.cuda) >= 10.2:
+                    os.environ["CUBLAS_WORKSPACE_CONFIG"] = "4096:8"
+                else:
+                    # Not all Operations support this.
+                    torch.use_deterministic_algorithms(True)
+                # This is only for Convolution no problem.
+                torch.backends.cudnn.deterministic = True
+            # Tf2.x.
+            elif tf and policy_config.get("framework") == "tf2":
+                tf.random.set_seed(seed)
+            # Tf-eager.
+            elif tf1 and policy_config.get("framework") == "tfe":
+                tf1.set_random_seed(seed)
+
         if _has_tensorflow_graph(policy_dict) and not (
                 tf1 and tf1.executing_eagerly()):
             if not tf1:
diff --git a/rllib/examples/deterministic_training.py b/rllib/examples/deterministic_training.py
new file mode 100644
index 0000000000000..c9951ae605f58
--- /dev/null
+++ b/rllib/examples/deterministic_training.py
@@ -0,0 +1,61 @@
+"""
+Example of a fully deterministic, repeatable RLlib train run using
+the "seed" config key.
+"""
+import argparse
+import os
+
+import ray
+from ray import tune
+from ray.rllib.examples.env.env_using_remote_actor import \
+    CartPoleWithRemoteParamServer, ParameterStorage
+from ray.rllib.utils.framework import try_import_tf, try_import_torch
+from ray.rllib.utils.test_utils import check
+
+tf1, tf, tfv = try_import_tf()
+torch, nn = try_import_torch()
+
+parser = argparse.ArgumentParser()
+parser.add_argument("--run", type=str, default="PPO")
+parser.add_argument(
+    "--framework", choices=["tf2", "tf", "tfe", "torch"], default="tf")
+parser.add_argument("--seed", type=int, default=42)
+parser.add_argument("--as-test", action="store_true")
+parser.add_argument("--stop-iters", type=int, default=2)
+
+if __name__ == "__main__":
+    args = parser.parse_args()
+    ray.init()
+
+    param_storage = ParameterStorage.options(name="param-server").remote()
+
+    config = {
+        "env": CartPoleWithRemoteParamServer,
+        "env_config": {
+            "param_server": "param-server",
+        },
+        # Use GPUs iff `RLLIB_NUM_GPUS` env var set to > 0.
+        "num_gpus": int(os.environ.get("RLLIB_NUM_GPUS", "0")),
+        "num_workers": 2,  # parallelism
+        "framework": args.framework,
+        "seed": args.seed,
+
+        # Simplify to run this example script faster.
+        "train_batch_size": 100,
+        "sgd_minibatch_size": 10,
+        "num_sgd_iter": 5,
+        "rollout_fragment_length": 50,
+    }
+
+    stop = {
+        "training_iteration": args.stop_iters,
+    }
+
+    results = tune.run(args.run, config=config, stop=stop, verbose=1)
+    results2 = tune.run(args.run, config=config, stop=stop, verbose=1)
+
+    if args.as_test:
+        check(
+            list(results.results.values())[0]["hist_stats"],
+            list(results2.results.values())[0]["hist_stats"])
+    ray.shutdown()
diff --git a/rllib/examples/env/env_using_remote_actor.py b/rllib/examples/env/env_using_remote_actor.py
new file mode 100644
index 0000000000000..15c5c4f610ec7
--- /dev/null
+++ b/rllib/examples/env/env_using_remote_actor.py
@@ -0,0 +1,56 @@
+"""
+Example of an environment that uses a named remote actor as parameter
+server.
+
+"""
+from gym.envs.classic_control.cartpole import CartPoleEnv
+from gym.utils import seeding
+
+import ray
+
+
+@ray.remote
+class ParameterStorage:
+    def get_params(self, rng):
+        return {
+            "MASSCART": rng.uniform(low=0.5, high=2.0),
+        }
+
+
+class CartPoleWithRemoteParamServer(CartPoleEnv):
+    """CartPoleMassEnv varies the weights of the cart and the pole.
+    """
+
+    def __init__(self, env_config):
+        self.env_config = env_config
+        super().__init__()
+        # Get our param server (remote actor) by name.
+        self._handler = ray.get_actor(
+            env_config.get("param_server", "param-server"))
+
+    # def seed(self, seed=None):
+    #     print(f"Seeding env (worker={self.env_config.worker_index}) "
+    #           f"with {seed}")
+    #     self.np_random, seed = seeding.np_random(seed)
+    #     return [seed]
+
+    def reset(self):
+        # Pass in our RNG to guarantee no race conditions.
+        # If `self._handler` had its own RNG, this may clash with other
+        # envs trying to use the same param-server.
+        params = ray.get(self._handler.get_params.remote(self.np_random))
+
+        # IMPORTANT: Advance the state of our RNG (self._rng was passed
+        # above via ray (serialized) and thus not altered locally here!).
+        # Or create a new RNG from another random number:
+        new_seed = self.np_random.randint(0, 1000000)
+        self.np_random, _ = seeding.np_random(new_seed)
+
+        print(f"Env worker-idx={self.env_config.worker_index} "
+              f"mass={params['MASSCART']}")
+
+        self.masscart = params["MASSCART"]
+        self.total_mass = (self.masspole + self.masscart)
+        self.polemass_length = (self.masspole * self.length)
+
+        return super().reset()

<!-- Thank you for your contribution! Please review https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before opening a pull request. -->

Fix "seed" setting to work in all frameworks and w/ all CUDA versions.
- Added example script that tests determinism in the parallel setting for all frameworks.

<!-- Please add a reviewer to the assignee section when you create a PR. If you don't have the access to it, we will shortly find a reviewer and assign them to your PR. -->

## Why are these changes needed?

<!-- Please give a short summary of the change and the problem this solves. -->

## Related issue number

<!-- For example: "Closes #1234" -->

## Checks

- [x] I've run `scripts/format.sh` to lint the changes in this PR.
- [x] I've included any doc changes needed for https://docs.ray.io/en/master/.
- [x] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [x] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(
