We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Pytorch RayClientProxy suppresses errors
### Describe the bug

When running in PyTorch simulated and an error occurs on the client side, the error message gets buried under `DEBUG` and may never be shown to failing clients and thus the simulated_server process.

I assume that the following lines should change their loglevel flags to `ERROR`:

https://github.com/adap/flower/blob/3918d62ce8c346dc9c8cbf7471d0c8912c1e9e8f/src/py/flwr/simulation/ray_transport/ray_client_proxy.py#L55
https://github.com/adap/flower/blob/3918d62ce8c346dc9c8cbf7471d0c8912c1e9e8f/src/py/flwr/simulation/ray_transport/ray_client_proxy.py#L72
https://github.com/adap/flower/blob/3918d62ce8c346dc9c8cbf7471d0c8912c1e9e8f/src/py/flwr/simulation/ray_transport/ray_client_proxy.py#L87
https://github.com/adap/flower/blob/3918d62ce8c346dc9c8cbf7471d0c8912c1e9e8f/src/py/flwr/simulation/ray_transport/ray_client_proxy.py#L104

### Steps/Code to Reproduce

Taking the code from `examples/simulation_pytorch/main.py`

```python
import argparse
import flwr as fl
from flwr.common.typing import Scalar
import ray
import torch
import torchvision
import numpy as np
from collections import OrderedDict
from pathlib import Path
from typing import Dict, Callable, Optional, Tuple, List
from dataset_utils import get_cifar_10, do_fl_partitioning, get_dataloader
from utils import Net, train, test

import logging
from flwr.common.logger import logger as flwr_logger
flwr_logger.setLevel(logging.INFO)

parser = argparse.ArgumentParser(description="Flower Simulation with PyTorch")

parser.add_argument("--num_client_cpus", type=int, default=1)
parser.add_argument("--num_rounds", type=int, default=5)


# Flower client, adapted from Pytorch quickstart example
class FlowerClient(fl.client.NumPyClient):
    def __init__(self, cid: str, fed_dir_data: str):
        self.cid = cid
        self.fed_dir = Path(fed_dir_data)
        self.properties: Dict[str, Scalar] = {"tensor_type": "numpy.ndarray"}

        # Instantiate model
        self.net = Net()

        # Determine device
        self.device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")

    def get_parameters(self, config):
        assert(1 == 0)  # This will be hidden
        return get_params(self.net)

    def fit(self, parameters, config):
        set_params(self.net, parameters)
        assert(1 == 0)  # This will be hidden 
        
        # Load data for this client and get trainloader
        num_workers = int(ray.get_runtime_context().get_assigned_resources()["CPU"])
        trainloader = get_dataloader(
            self.fed_dir,
            self.cid,
            is_train=True,
            batch_size=config["batch_size"],
            workers=num_workers,
        )

        # Send model to device
        self.net.to(self.device)

        # Train
        train(self.net, trainloader, epochs=config["epochs"], device=self.device)

        # Return local model and statistics
        return get_params(self.net), len(trainloader.dataset), {}

    def evaluate(self, parameters, config):
        set_params(self.net, parameters)
        assert(1 == 0)  # This will be hidden

        # Load data for this client and get trainloader
        num_workers = int(ray.get_runtime_context().get_assigned_resources()["CPU"])
        valloader = get_dataloader(
            self.fed_dir, self.cid, is_train=False, batch_size=50, workers=num_workers
        )

        # Send model to device
        self.net.to(self.device)

        # Evaluate
        loss, accuracy = test(self.net, valloader, device=self.device)

        # Return statistics
        return float(loss), len(valloader.dataset), {"accuracy": float(accuracy)}


def fit_config(server_round: int) -> Dict[str, Scalar]:
    """Return a configuration with static batch size and (local) epochs."""
    config = {
        "epochs": 5,  # number of local epochs
        "batch_size": 64,
    }
    return config


def get_params(model: torch.nn.ModuleList) -> List[np.ndarray]:
    """Get model weights as a list of NumPy ndarrays."""
    return [val.cpu().numpy() for _, val in model.state_dict().items()]


def set_params(model: torch.nn.ModuleList, params: List[np.ndarray]):
    """Set model weights from a list of NumPy ndarrays."""
    params_dict = zip(model.state_dict().keys(), params)
    state_dict = OrderedDict({k: torch.from_numpy(np.copy(v)) for k, v in params_dict})
    model.load_state_dict(state_dict, strict=True)


def get_evaluate_fn(
    testset: torchvision.datasets.CIFAR10,
) -> Callable[[fl.common.NDArrays], Optional[Tuple[float, float]]]:
    """Return an evaluation function for centralized evaluation."""

    def evaluate(
        server_round: int, parameters: fl.common.NDArrays, config: Dict[str, Scalar]
    ) -> Optional[Tuple[float, float]]:
        """Use the entire CIFAR-10 test set for evaluation."""

        # determine device
        device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")

        model = Net()
        set_params(model, parameters)
        model.to(device)

        testloader = torch.utils.data.DataLoader(testset, batch_size=50)
        loss, accuracy = test(model, testloader, device=device)

        # return statistics
        return loss, {"accuracy": accuracy}

    return evaluate


# Start simulation (a _default server_ will be created)
# This example does:
# 1. Downloads CIFAR-10
# 2. Partitions the dataset into N splits, where N is the total number of
#    clients. We refere to this as `pool_size`. The partition can be IID or non-IID
# 3. Starts a simulation where a % of clients are sample each round.
# 4. After the M rounds end, the global model is evaluated on the entire testset.
#    Also, the global model is evaluated on the valset partition residing in each
#    client. This is useful to get a sense on how well the global model can generalise
#    to each client's data.
if __name__ == "__main__":
    # parse input arguments
    args = parser.parse_args()

    pool_size = 100  # number of dataset partions (= number of total clients)
    client_resources = {
        "num_cpus": args.num_client_cpus
    }  # each client will get allocated 1 CPUs

    # Download CIFAR-10 dataset
    train_path, testset = get_cifar_10()

    # partition dataset (use a large `alpha` to make it IID;
    # a small value (e.g. 1) will make it non-IID)
    # This will create a new directory called "federated": in the directory where
    # CIFAR-10 lives. Inside it, there will be N=pool_size sub-directories each with
    # its own train/set split.
    fed_dir = do_fl_partitioning(
        train_path, pool_size=pool_size, alpha=1000, num_classes=10, val_ratio=0.1
    )

    # configure the strategy
    strategy = fl.server.strategy.FedAvg(
        fraction_fit=0.1,
        fraction_evaluate=0.1,
        min_fit_clients=10,
        min_evaluate_clients=10,
        min_available_clients=pool_size,  # All clients should be available
        on_fit_config_fn=fit_config,
        evaluate_fn=get_evaluate_fn(testset),  # centralised evaluation of global model
    )

    def client_fn(cid: str):
        # create a single client instance
        return FlowerClient(cid, fed_dir)

    # (optional) specify Ray config
    ray_init_args = {"include_dashboard": False}

    # start simulation
    fl.simulation.start_simulation(
        client_fn=client_fn,
        num_clients=pool_size,
        client_resources=client_resources,
        config=fl.server.ServerConfig(num_rounds=args.num_rounds),
        strategy=strategy,
        ray_init_args=ray_init_args,
    )
```

And running `examples/simulation_pytorch/run.sh`.

### Expected Results

Expectation is to see the assertions errors, as this is unintended behaviour.

### Actual Results

Errors are suppressed because of `flwr_logger.setLevel(logging.INFO)`, but should not.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/py/flwr/simulation/ray_transport/ray_client_proxy.py`
Content:
```
1 # Copyright 2020 Adap GmbH. All Rights Reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 # ==============================================================================
15 """Ray-based Flower ClientProxy implementation."""
16 
17 
18 from logging import DEBUG
19 from typing import Callable, Dict, Optional, cast
20 
21 import ray
22 
23 from flwr import common
24 from flwr.client import Client, ClientLike, to_client
25 from flwr.client.client import (
26     maybe_call_evaluate,
27     maybe_call_fit,
28     maybe_call_get_parameters,
29     maybe_call_get_properties,
30 )
31 from flwr.common.logger import log
32 from flwr.server.client_proxy import ClientProxy
33 
34 ClientFn = Callable[[str], ClientLike]
35 
36 
37 class RayClientProxy(ClientProxy):
38     """Flower client proxy which delegates work using Ray."""
39 
40     def __init__(self, client_fn: ClientFn, cid: str, resources: Dict[str, float]):
41         super().__init__(cid)
42         self.client_fn = client_fn
43         self.resources = resources
44 
45     def get_properties(
46         self, ins: common.GetPropertiesIns, timeout: Optional[float]
47     ) -> common.GetPropertiesRes:
48         """Returns client's properties."""
49         future_get_properties_res = launch_and_get_properties.options(  # type: ignore
50             **self.resources,
51         ).remote(self.client_fn, self.cid, ins)
52         try:
53             res = ray.get(future_get_properties_res, timeout=timeout)
54         except Exception as ex:
55             log(DEBUG, ex)
56             raise ex
57         return cast(
58             common.GetPropertiesRes,
59             res,
60         )
61 
62     def get_parameters(
63         self, ins: common.GetParametersIns, timeout: Optional[float]
64     ) -> common.GetParametersRes:
65         """Return the current local model parameters."""
66         future_paramseters_res = launch_and_get_parameters.options(  # type: ignore
67             **self.resources,
68         ).remote(self.client_fn, self.cid, ins)
69         try:
70             res = ray.get(future_paramseters_res, timeout=timeout)
71         except Exception as ex:
72             log(DEBUG, ex)
73             raise ex
74         return cast(
75             common.GetParametersRes,
76             res,
77         )
78 
79     def fit(self, ins: common.FitIns, timeout: Optional[float]) -> common.FitRes:
80         """Train model parameters on the locally held dataset."""
81         future_fit_res = launch_and_fit.options(  # type: ignore
82             **self.resources,
83         ).remote(self.client_fn, self.cid, ins)
84         try:
85             res = ray.get(future_fit_res, timeout=timeout)
86         except Exception as ex:
87             log(DEBUG, ex)
88             raise ex
89         return cast(
90             common.FitRes,
91             res,
92         )
93 
94     def evaluate(
95         self, ins: common.EvaluateIns, timeout: Optional[float]
96     ) -> common.EvaluateRes:
97         """Evaluate model parameters on the locally held dataset."""
98         future_evaluate_res = launch_and_evaluate.options(  # type: ignore
99             **self.resources,
100         ).remote(self.client_fn, self.cid, ins)
101         try:
102             res = ray.get(future_evaluate_res, timeout=timeout)
103         except Exception as ex:
104             log(DEBUG, ex)
105             raise ex
106         return cast(
107             common.EvaluateRes,
108             res,
109         )
110 
111     def reconnect(
112         self, ins: common.ReconnectIns, timeout: Optional[float]
113     ) -> common.DisconnectRes:
114         """Disconnect and (optionally) reconnect later."""
115         return common.DisconnectRes(reason="")  # Nothing to do here (yet)
116 
117 
118 @ray.remote
119 def launch_and_get_properties(
120     client_fn: ClientFn, cid: str, get_properties_ins: common.GetPropertiesIns
121 ) -> common.GetPropertiesRes:
122     """Exectue get_properties remotely."""
123     client: Client = _create_client(client_fn, cid)
124     return maybe_call_get_properties(
125         client=client,
126         get_properties_ins=get_properties_ins,
127     )
128 
129 
130 @ray.remote
131 def launch_and_get_parameters(
132     client_fn: ClientFn, cid: str, get_parameters_ins: common.GetParametersIns
133 ) -> common.GetParametersRes:
134     """Exectue get_parameters remotely."""
135     client: Client = _create_client(client_fn, cid)
136     return maybe_call_get_parameters(
137         client=client,
138         get_parameters_ins=get_parameters_ins,
139     )
140 
141 
142 @ray.remote
143 def launch_and_fit(
144     client_fn: ClientFn, cid: str, fit_ins: common.FitIns
145 ) -> common.FitRes:
146     """Exectue fit remotely."""
147     client: Client = _create_client(client_fn, cid)
148     return maybe_call_fit(
149         client=client,
150         fit_ins=fit_ins,
151     )
152 
153 
154 @ray.remote
155 def launch_and_evaluate(
156     client_fn: ClientFn, cid: str, evaluate_ins: common.EvaluateIns
157 ) -> common.EvaluateRes:
158     """Exectue evaluate remotely."""
159     client: Client = _create_client(client_fn, cid)
160     return maybe_call_evaluate(
161         client=client,
162         evaluate_ins=evaluate_ins,
163     )
164 
165 
166 def _create_client(client_fn: ClientFn, cid: str) -> Client:
167     """Create a client instance."""
168     client_like: ClientLike = client_fn(cid)
169     return to_client(client_like=client_like)
170
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/py/flwr/simulation/ray_transport/ray_client_proxy.py b/src/py/flwr/simulation/ray_transport/ray_client_proxy.py
--- a/src/py/flwr/simulation/ray_transport/ray_client_proxy.py
+++ b/src/py/flwr/simulation/ray_transport/ray_client_proxy.py
@@ -15,7 +15,7 @@
 """Ray-based Flower ClientProxy implementation."""
 
 
-from logging import DEBUG
+from logging import ERROR
 from typing import Callable, Dict, Optional, cast
 
 import ray
@@ -52,7 +52,7 @@
         try:
             res = ray.get(future_get_properties_res, timeout=timeout)
         except Exception as ex:
-            log(DEBUG, ex)
+            log(ERROR, ex)
             raise ex
         return cast(
             common.GetPropertiesRes,
@@ -69,7 +69,7 @@
         try:
             res = ray.get(future_paramseters_res, timeout=timeout)
         except Exception as ex:
-            log(DEBUG, ex)
+            log(ERROR, ex)
             raise ex
         return cast(
             common.GetParametersRes,
@@ -84,7 +84,7 @@
         try:
             res = ray.get(future_fit_res, timeout=timeout)
         except Exception as ex:
-            log(DEBUG, ex)
+            log(ERROR, ex)
             raise ex
         return cast(
             common.FitRes,
@@ -101,7 +101,7 @@
         try:
             res = ray.get(future_evaluate_res, timeout=timeout)
         except Exception as ex:
-            log(DEBUG, ex)
+            log(ERROR, ex)
             raise ex
         return cast(
             common.EvaluateRes,

{"golden_diff": "diff --git a/src/py/flwr/simulation/ray_transport/ray_client_proxy.py b/src/py/flwr/simulation/ray_transport/ray_client_proxy.py\n--- a/src/py/flwr/simulation/ray_transport/ray_client_proxy.py\n+++ b/src/py/flwr/simulation/ray_transport/ray_client_proxy.py\n@@ -15,7 +15,7 @@\n \"\"\"Ray-based Flower ClientProxy implementation.\"\"\"\n \n \n-from logging import DEBUG\n+from logging import ERROR\n from typing import Callable, Dict, Optional, cast\n \n import ray\n@@ -52,7 +52,7 @@\n         try:\n             res = ray.get(future_get_properties_res, timeout=timeout)\n         except Exception as ex:\n-            log(DEBUG, ex)\n+            log(ERROR, ex)\n             raise ex\n         return cast(\n             common.GetPropertiesRes,\n@@ -69,7 +69,7 @@\n         try:\n             res = ray.get(future_paramseters_res, timeout=timeout)\n         except Exception as ex:\n-            log(DEBUG, ex)\n+            log(ERROR, ex)\n             raise ex\n         return cast(\n             common.GetParametersRes,\n@@ -84,7 +84,7 @@\n         try:\n             res = ray.get(future_fit_res, timeout=timeout)\n         except Exception as ex:\n-            log(DEBUG, ex)\n+            log(ERROR, ex)\n             raise ex\n         return cast(\n             common.FitRes,\n@@ -101,7 +101,7 @@\n         try:\n             res = ray.get(future_evaluate_res, timeout=timeout)\n         except Exception as ex:\n-            log(DEBUG, ex)\n+            log(ERROR, ex)\n             raise ex\n         return cast(\n             common.EvaluateRes,\n", "issue": "Pytorch RayClientProxy suppresses errors\n### Describe the bug\n\nWhen running in PyTorch simulated and an error occurs on the client side, the error message gets buried under `DEBUG` and may never be shown to failing clients and thus the simulated_server process.\r\n\r\nI assume that the following lines should change their loglevel flags to `ERROR`:\r\n\r\nhttps://github.com/adap/flower/blob/3918d62ce8c346dc9c8cbf7471d0c8912c1e9e8f/src/py/flwr/simulation/ray_transport/ray_client_proxy.py#L55\r\nhttps://github.com/adap/flower/blob/3918d62ce8c346dc9c8cbf7471d0c8912c1e9e8f/src/py/flwr/simulation/ray_transport/ray_client_proxy.py#L72\r\nhttps://github.com/adap/flower/blob/3918d62ce8c346dc9c8cbf7471d0c8912c1e9e8f/src/py/flwr/simulation/ray_transport/ray_client_proxy.py#L87\r\nhttps://github.com/adap/flower/blob/3918d62ce8c346dc9c8cbf7471d0c8912c1e9e8f/src/py/flwr/simulation/ray_transport/ray_client_proxy.py#L104\n\n### Steps/Code to Reproduce\n\nTaking the code from `examples/simulation_pytorch/main.py`\r\n\r\n```python\r\nimport argparse\r\nimport flwr as fl\r\nfrom flwr.common.typing import Scalar\r\nimport ray\r\nimport torch\r\nimport torchvision\r\nimport numpy as np\r\nfrom collections import OrderedDict\r\nfrom pathlib import Path\r\nfrom typing import Dict, Callable, Optional, Tuple, List\r\nfrom dataset_utils import get_cifar_10, do_fl_partitioning, get_dataloader\r\nfrom utils import Net, train, test\r\n\r\nimport logging\r\nfrom flwr.common.logger import logger as flwr_logger\r\nflwr_logger.setLevel(logging.INFO)\r\n\r\nparser = argparse.ArgumentParser(description=\"Flower Simulation with PyTorch\")\r\n\r\nparser.add_argument(\"--num_client_cpus\", type=int, default=1)\r\nparser.add_argument(\"--num_rounds\", type=int, default=5)\r\n\r\n\r\n# Flower client, adapted from Pytorch quickstart example\r\nclass FlowerClient(fl.client.NumPyClient):\r\n    def __init__(self, cid: str, fed_dir_data: str):\r\n        self.cid = cid\r\n        self.fed_dir = Path(fed_dir_data)\r\n        self.properties: Dict[str, Scalar] = {\"tensor_type\": \"numpy.ndarray\"}\r\n\r\n        # Instantiate model\r\n        self.net = Net()\r\n\r\n        # Determine device\r\n        self.device = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\r\n\r\n    def get_parameters(self, config):\r\n        assert(1 == 0)  # This will be hidden\r\n        return get_params(self.net)\r\n\r\n    def fit(self, parameters, config):\r\n        set_params(self.net, parameters)\r\n        assert(1 == 0)  # This will be hidden \r\n        \r\n        # Load data for this client and get trainloader\r\n        num_workers = int(ray.get_runtime_context().get_assigned_resources()[\"CPU\"])\r\n        trainloader = get_dataloader(\r\n            self.fed_dir,\r\n            self.cid,\r\n            is_train=True,\r\n            batch_size=config[\"batch_size\"],\r\n            workers=num_workers,\r\n        )\r\n\r\n        # Send model to device\r\n        self.net.to(self.device)\r\n\r\n        # Train\r\n        train(self.net, trainloader, epochs=config[\"epochs\"], device=self.device)\r\n\r\n        # Return local model and statistics\r\n        return get_params(self.net), len(trainloader.dataset), {}\r\n\r\n    def evaluate(self, parameters, config):\r\n        set_params(self.net, parameters)\r\n        assert(1 == 0)  # This will be hidden\r\n\r\n        # Load data for this client and get trainloader\r\n        num_workers = int(ray.get_runtime_context().get_assigned_resources()[\"CPU\"])\r\n        valloader = get_dataloader(\r\n            self.fed_dir, self.cid, is_train=False, batch_size=50, workers=num_workers\r\n        )\r\n\r\n        # Send model to device\r\n        self.net.to(self.device)\r\n\r\n        # Evaluate\r\n        loss, accuracy = test(self.net, valloader, device=self.device)\r\n\r\n        # Return statistics\r\n        return float(loss), len(valloader.dataset), {\"accuracy\": float(accuracy)}\r\n\r\n\r\ndef fit_config(server_round: int) -> Dict[str, Scalar]:\r\n    \"\"\"Return a configuration with static batch size and (local) epochs.\"\"\"\r\n    config = {\r\n        \"epochs\": 5,  # number of local epochs\r\n        \"batch_size\": 64,\r\n    }\r\n    return config\r\n\r\n\r\ndef get_params(model: torch.nn.ModuleList) -> List[np.ndarray]:\r\n    \"\"\"Get model weights as a list of NumPy ndarrays.\"\"\"\r\n    return [val.cpu().numpy() for _, val in model.state_dict().items()]\r\n\r\n\r\ndef set_params(model: torch.nn.ModuleList, params: List[np.ndarray]):\r\n    \"\"\"Set model weights from a list of NumPy ndarrays.\"\"\"\r\n    params_dict = zip(model.state_dict().keys(), params)\r\n    state_dict = OrderedDict({k: torch.from_numpy(np.copy(v)) for k, v in params_dict})\r\n    model.load_state_dict(state_dict, strict=True)\r\n\r\n\r\ndef get_evaluate_fn(\r\n    testset: torchvision.datasets.CIFAR10,\r\n) -> Callable[[fl.common.NDArrays], Optional[Tuple[float, float]]]:\r\n    \"\"\"Return an evaluation function for centralized evaluation.\"\"\"\r\n\r\n    def evaluate(\r\n        server_round: int, parameters: fl.common.NDArrays, config: Dict[str, Scalar]\r\n    ) -> Optional[Tuple[float, float]]:\r\n        \"\"\"Use the entire CIFAR-10 test set for evaluation.\"\"\"\r\n\r\n        # determine device\r\n        device = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\r\n\r\n        model = Net()\r\n        set_params(model, parameters)\r\n        model.to(device)\r\n\r\n        testloader = torch.utils.data.DataLoader(testset, batch_size=50)\r\n        loss, accuracy = test(model, testloader, device=device)\r\n\r\n        # return statistics\r\n        return loss, {\"accuracy\": accuracy}\r\n\r\n    return evaluate\r\n\r\n\r\n# Start simulation (a _default server_ will be created)\r\n# This example does:\r\n# 1. Downloads CIFAR-10\r\n# 2. Partitions the dataset into N splits, where N is the total number of\r\n#    clients. We refere to this as `pool_size`. The partition can be IID or non-IID\r\n# 3. Starts a simulation where a % of clients are sample each round.\r\n# 4. After the M rounds end, the global model is evaluated on the entire testset.\r\n#    Also, the global model is evaluated on the valset partition residing in each\r\n#    client. This is useful to get a sense on how well the global model can generalise\r\n#    to each client's data.\r\nif __name__ == \"__main__\":\r\n    # parse input arguments\r\n    args = parser.parse_args()\r\n\r\n    pool_size = 100  # number of dataset partions (= number of total clients)\r\n    client_resources = {\r\n        \"num_cpus\": args.num_client_cpus\r\n    }  # each client will get allocated 1 CPUs\r\n\r\n    # Download CIFAR-10 dataset\r\n    train_path, testset = get_cifar_10()\r\n\r\n    # partition dataset (use a large `alpha` to make it IID;\r\n    # a small value (e.g. 1) will make it non-IID)\r\n    # This will create a new directory called \"federated\": in the directory where\r\n    # CIFAR-10 lives. Inside it, there will be N=pool_size sub-directories each with\r\n    # its own train/set split.\r\n    fed_dir = do_fl_partitioning(\r\n        train_path, pool_size=pool_size, alpha=1000, num_classes=10, val_ratio=0.1\r\n    )\r\n\r\n    # configure the strategy\r\n    strategy = fl.server.strategy.FedAvg(\r\n        fraction_fit=0.1,\r\n        fraction_evaluate=0.1,\r\n        min_fit_clients=10,\r\n        min_evaluate_clients=10,\r\n        min_available_clients=pool_size,  # All clients should be available\r\n        on_fit_config_fn=fit_config,\r\n        evaluate_fn=get_evaluate_fn(testset),  # centralised evaluation of global model\r\n    )\r\n\r\n    def client_fn(cid: str):\r\n        # create a single client instance\r\n        return FlowerClient(cid, fed_dir)\r\n\r\n    # (optional) specify Ray config\r\n    ray_init_args = {\"include_dashboard\": False}\r\n\r\n    # start simulation\r\n    fl.simulation.start_simulation(\r\n        client_fn=client_fn,\r\n        num_clients=pool_size,\r\n        client_resources=client_resources,\r\n        config=fl.server.ServerConfig(num_rounds=args.num_rounds),\r\n        strategy=strategy,\r\n        ray_init_args=ray_init_args,\r\n    )\r\n```\r\n\r\nAnd running `examples/simulation_pytorch/run.sh`.\n\n### Expected Results\n\nExpectation is to see the assertions errors, as this is unintended behaviour.\n\n### Actual Results\n\nErrors are suppressed because of `flwr_logger.setLevel(logging.INFO)`, but should not.\n", "before_files": [{"content": "# Copyright 2020 Adap GmbH. All Rights Reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n# ==============================================================================\n\"\"\"Ray-based Flower ClientProxy implementation.\"\"\"\n\n\nfrom logging import DEBUG\nfrom typing import Callable, Dict, Optional, cast\n\nimport ray\n\nfrom flwr import common\nfrom flwr.client import Client, ClientLike, to_client\nfrom flwr.client.client import (\n    maybe_call_evaluate,\n    maybe_call_fit,\n    maybe_call_get_parameters,\n    maybe_call_get_properties,\n)\nfrom flwr.common.logger import log\nfrom flwr.server.client_proxy import ClientProxy\n\nClientFn = Callable[[str], ClientLike]\n\n\nclass RayClientProxy(ClientProxy):\n    \"\"\"Flower client proxy which delegates work using Ray.\"\"\"\n\n    def __init__(self, client_fn: ClientFn, cid: str, resources: Dict[str, float]):\n        super().__init__(cid)\n        self.client_fn = client_fn\n        self.resources = resources\n\n    def get_properties(\n        self, ins: common.GetPropertiesIns, timeout: Optional[float]\n    ) -> common.GetPropertiesRes:\n        \"\"\"Returns client's properties.\"\"\"\n        future_get_properties_res = launch_and_get_properties.options(  # type: ignore\n            **self.resources,\n        ).remote(self.client_fn, self.cid, ins)\n        try:\n            res = ray.get(future_get_properties_res, timeout=timeout)\n        except Exception as ex:\n            log(DEBUG, ex)\n            raise ex\n        return cast(\n            common.GetPropertiesRes,\n            res,\n        )\n\n    def get_parameters(\n        self, ins: common.GetParametersIns, timeout: Optional[float]\n    ) -> common.GetParametersRes:\n        \"\"\"Return the current local model parameters.\"\"\"\n        future_paramseters_res = launch_and_get_parameters.options(  # type: ignore\n            **self.resources,\n        ).remote(self.client_fn, self.cid, ins)\n        try:\n            res = ray.get(future_paramseters_res, timeout=timeout)\n        except Exception as ex:\n            log(DEBUG, ex)\n            raise ex\n        return cast(\n            common.GetParametersRes,\n            res,\n        )\n\n    def fit(self, ins: common.FitIns, timeout: Optional[float]) -> common.FitRes:\n        \"\"\"Train model parameters on the locally held dataset.\"\"\"\n        future_fit_res = launch_and_fit.options(  # type: ignore\n            **self.resources,\n        ).remote(self.client_fn, self.cid, ins)\n        try:\n            res = ray.get(future_fit_res, timeout=timeout)\n        except Exception as ex:\n            log(DEBUG, ex)\n            raise ex\n        return cast(\n            common.FitRes,\n            res,\n        )\n\n    def evaluate(\n        self, ins: common.EvaluateIns, timeout: Optional[float]\n    ) -> common.EvaluateRes:\n        \"\"\"Evaluate model parameters on the locally held dataset.\"\"\"\n        future_evaluate_res = launch_and_evaluate.options(  # type: ignore\n            **self.resources,\n        ).remote(self.client_fn, self.cid, ins)\n        try:\n            res = ray.get(future_evaluate_res, timeout=timeout)\n        except Exception as ex:\n            log(DEBUG, ex)\n            raise ex\n        return cast(\n            common.EvaluateRes,\n            res,\n        )\n\n    def reconnect(\n        self, ins: common.ReconnectIns, timeout: Optional[float]\n    ) -> common.DisconnectRes:\n        \"\"\"Disconnect and (optionally) reconnect later.\"\"\"\n        return common.DisconnectRes(reason=\"\")  # Nothing to do here (yet)\n\n\n@ray.remote\ndef launch_and_get_properties(\n    client_fn: ClientFn, cid: str, get_properties_ins: common.GetPropertiesIns\n) -> common.GetPropertiesRes:\n    \"\"\"Exectue get_properties remotely.\"\"\"\n    client: Client = _create_client(client_fn, cid)\n    return maybe_call_get_properties(\n        client=client,\n        get_properties_ins=get_properties_ins,\n    )\n\n\n@ray.remote\ndef launch_and_get_parameters(\n    client_fn: ClientFn, cid: str, get_parameters_ins: common.GetParametersIns\n) -> common.GetParametersRes:\n    \"\"\"Exectue get_parameters remotely.\"\"\"\n    client: Client = _create_client(client_fn, cid)\n    return maybe_call_get_parameters(\n        client=client,\n        get_parameters_ins=get_parameters_ins,\n    )\n\n\n@ray.remote\ndef launch_and_fit(\n    client_fn: ClientFn, cid: str, fit_ins: common.FitIns\n) -> common.FitRes:\n    \"\"\"Exectue fit remotely.\"\"\"\n    client: Client = _create_client(client_fn, cid)\n    return maybe_call_fit(\n        client=client,\n        fit_ins=fit_ins,\n    )\n\n\n@ray.remote\ndef launch_and_evaluate(\n    client_fn: ClientFn, cid: str, evaluate_ins: common.EvaluateIns\n) -> common.EvaluateRes:\n    \"\"\"Exectue evaluate remotely.\"\"\"\n    client: Client = _create_client(client_fn, cid)\n    return maybe_call_evaluate(\n        client=client,\n        evaluate_ins=evaluate_ins,\n    )\n\n\ndef _create_client(client_fn: ClientFn, cid: str) -> Client:\n    \"\"\"Create a client instance.\"\"\"\n    client_like: ClientLike = client_fn(cid)\n    return to_client(client_like=client_like)\n", "path": "src/py/flwr/simulation/ray_transport/ray_client_proxy.py"}], "after_files": [{"content": "# Copyright 2020 Adap GmbH. All Rights Reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n# ==============================================================================\n\"\"\"Ray-based Flower ClientProxy implementation.\"\"\"\n\n\nfrom logging import ERROR\nfrom typing import Callable, Dict, Optional, cast\n\nimport ray\n\nfrom flwr import common\nfrom flwr.client import Client, ClientLike, to_client\nfrom flwr.client.client import (\n    maybe_call_evaluate,\n    maybe_call_fit,\n    maybe_call_get_parameters,\n    maybe_call_get_properties,\n)\nfrom flwr.common.logger import log\nfrom flwr.server.client_proxy import ClientProxy\n\nClientFn = Callable[[str], ClientLike]\n\n\nclass RayClientProxy(ClientProxy):\n    \"\"\"Flower client proxy which delegates work using Ray.\"\"\"\n\n    def __init__(self, client_fn: ClientFn, cid: str, resources: Dict[str, float]):\n        super().__init__(cid)\n        self.client_fn = client_fn\n        self.resources = resources\n\n    def get_properties(\n        self, ins: common.GetPropertiesIns, timeout: Optional[float]\n    ) -> common.GetPropertiesRes:\n        \"\"\"Returns client's properties.\"\"\"\n        future_get_properties_res = launch_and_get_properties.options(  # type: ignore\n            **self.resources,\n        ).remote(self.client_fn, self.cid, ins)\n        try:\n            res = ray.get(future_get_properties_res, timeout=timeout)\n        except Exception as ex:\n            log(ERROR, ex)\n            raise ex\n        return cast(\n            common.GetPropertiesRes,\n            res,\n        )\n\n    def get_parameters(\n        self, ins: common.GetParametersIns, timeout: Optional[float]\n    ) -> common.GetParametersRes:\n        \"\"\"Return the current local model parameters.\"\"\"\n        future_paramseters_res = launch_and_get_parameters.options(  # type: ignore\n            **self.resources,\n        ).remote(self.client_fn, self.cid, ins)\n        try:\n            res = ray.get(future_paramseters_res, timeout=timeout)\n        except Exception as ex:\n            log(ERROR, ex)\n            raise ex\n        return cast(\n            common.GetParametersRes,\n            res,\n        )\n\n    def fit(self, ins: common.FitIns, timeout: Optional[float]) -> common.FitRes:\n        \"\"\"Train model parameters on the locally held dataset.\"\"\"\n        future_fit_res = launch_and_fit.options(  # type: ignore\n            **self.resources,\n        ).remote(self.client_fn, self.cid, ins)\n        try:\n            res = ray.get(future_fit_res, timeout=timeout)\n        except Exception as ex:\n            log(ERROR, ex)\n            raise ex\n        return cast(\n            common.FitRes,\n            res,\n        )\n\n    def evaluate(\n        self, ins: common.EvaluateIns, timeout: Optional[float]\n    ) -> common.EvaluateRes:\n        \"\"\"Evaluate model parameters on the locally held dataset.\"\"\"\n        future_evaluate_res = launch_and_evaluate.options(  # type: ignore\n            **self.resources,\n        ).remote(self.client_fn, self.cid, ins)\n        try:\n            res = ray.get(future_evaluate_res, timeout=timeout)\n        except Exception as ex:\n            log(ERROR, ex)\n            raise ex\n        return cast(\n            common.EvaluateRes,\n            res,\n        )\n\n    def reconnect(\n        self, ins: common.ReconnectIns, timeout: Optional[float]\n    ) -> common.DisconnectRes:\n        \"\"\"Disconnect and (optionally) reconnect later.\"\"\"\n        return common.DisconnectRes(reason=\"\")  # Nothing to do here (yet)\n\n\n@ray.remote\ndef launch_and_get_properties(\n    client_fn: ClientFn, cid: str, get_properties_ins: common.GetPropertiesIns\n) -> common.GetPropertiesRes:\n    \"\"\"Exectue get_properties remotely.\"\"\"\n    client: Client = _create_client(client_fn, cid)\n    return maybe_call_get_properties(\n        client=client,\n        get_properties_ins=get_properties_ins,\n    )\n\n\n@ray.remote\ndef launch_and_get_parameters(\n    client_fn: ClientFn, cid: str, get_parameters_ins: common.GetParametersIns\n) -> common.GetParametersRes:\n    \"\"\"Exectue get_parameters remotely.\"\"\"\n    client: Client = _create_client(client_fn, cid)\n    return maybe_call_get_parameters(\n        client=client,\n        get_parameters_ins=get_parameters_ins,\n    )\n\n\n@ray.remote\ndef launch_and_fit(\n    client_fn: ClientFn, cid: str, fit_ins: common.FitIns\n) -> common.FitRes:\n    \"\"\"Exectue fit remotely.\"\"\"\n    client: Client = _create_client(client_fn, cid)\n    return maybe_call_fit(\n        client=client,\n        fit_ins=fit_ins,\n    )\n\n\n@ray.remote\ndef launch_and_evaluate(\n    client_fn: ClientFn, cid: str, evaluate_ins: common.EvaluateIns\n) -> common.EvaluateRes:\n    \"\"\"Exectue evaluate remotely.\"\"\"\n    client: Client = _create_client(client_fn, cid)\n    return maybe_call_evaluate(\n        client=client,\n        evaluate_ins=evaluate_ins,\n    )\n\n\ndef _create_client(client_fn: ClientFn, cid: str) -> Client:\n    \"\"\"Create a client instance.\"\"\"\n    client_like: ClientLike = client_fn(cid)\n    return to_client(client_like=client_like)\n", "path": "src/py/flwr/simulation/ray_transport/ray_client_proxy.py"}]}