We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
If gRPC server is running behing a load balancer, clients may not be able to connect
### Describe the bug

Flower is using the `peer()` method from `grpc.ServicerContext` in order to identify unique flower clients. However, in some situations (like when running the gRPC server behind a load balancer or proxy) different clients can have the same peer identifier (i.e. the same `IP:PORT`), as HTTP/2 connections are multiplexed.

### Steps/Code to Reproduce

This can be tested with a simple [Traefik](https://doc.traefik.io/traefik/) setup as follows (https://doc.traefik.io/traefik/user-guides/grpc/):

`traefik.yaml`:
```
entryPoints:
  web:
    address: :80

providers:
  file:
    filename: grpc.yaml

api: {}
```

`grpc.yaml`:
```
http:
  routers:
    to-grpc:
      entryPoints: 
        - "web"
      service: "srv-grpc"
      rule: "HostRegexp(`.*`)"

  services:
    srv-grpc:
      loadBalancer:
        servers:
           - url: h2c://127.0.0.1:5000
```

The Traefik server can be executed as follows:
```
 ./traefik --configFile=traefik.yaml
```

Then we can launch then a Flower sever with the following code:
```
import flwr as fl

def acc_wavg(metrics):
    n = sum([i for i, _ in metrics])
    acc = sum([i * metric["accuracy"] / n for i, metric in metrics])
    return {"accuracy": acc}
 
strategy = fl.server.strategy.FedAvg(
    min_available_clients=2,
    evaluate_metrics_aggregation_fn=acc_wavg
)
 
fl.server.start_server(
    server_address="0.0.0.0:5000",
    config=fl.server.ServerConfig(num_rounds=3),
    strategy=strategy,
)
```

### Expected Results

When we connect two clients to the loadbalancer IP, on port 80, the first one connects and keeps waiting:

```
INFO flwr 2023-06-28 10:03:14,848 | grpc.py:50 | Opened insecure gRPC connection (no certificates were passed)
DEBUG flwr 2023-06-28 10:03:14,849 | connection.py:39 | ChannelConnectivity.IDLE
DEBUG flwr 2023-06-28 10:03:14,849 | connection.py:39 | ChannelConnectivity.CONNECTING
DEBUG flwr 2023-06-28 10:03:14,875 | connection.py:39 | ChannelConnectivity.READY
```

However, the second one is not connected:
```
DEBUG flwr 2023-06-28 09:37:32,554 | connection.py:39 | ChannelConnectivity.IDLE
DEBUG flwr 2023-06-28 09:37:32,554 | connection.py:39 | ChannelConnectivity.CONNECTING
DEBUG flwr 2023-06-28 09:37:32,578 | connection.py:39 | ChannelConnectivity.READY
DEBUG flwr 2023-06-28 09:37:32,784 | connection.py:113 | gRPC channel closed
Traceback (most recent call last):
  File "/home/alvaro/w/prj/fl/fl.py", line 68, in <module>
    fl.client.start_numpy_client(
  File "/home/alvaro/w/prj/fl/VENV/lib/python3.11/site-packages/flwr/client/app.py", line 252, in start_numpy_client
    start_client(
  File "/home/alvaro/w/prj/fl/VENV/lib/python3.11/site-packages/flwr/client/app.py", line 174, in start_client
    server_message = receive()
                     ^^^^^^^^^
  File "/home/alvaro/w/prj/fl/VENV/lib/python3.11/site-packages/flwr/client/grpc_client/connection.py", line 105, in <lambda>
    receive: Callable[[], ServerMessage] = lambda: next(server_message_iterator)
                                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/alvaro/w/prj/fl/VENV/lib/python3.11/site-packages/grpc/_channel.py", line 475, in __next__
    return self._next()
           ^^^^^^^^^^^^
  File "/home/alvaro/w/prj/fl/VENV/lib/python3.11/site-packages/grpc/_channel.py", line 879, in _next
    raise StopIteration()
StopIteration
```

This is because the second client connection is discarded [here]https://github.com/adap/flower/blob/main/src/py/flwr/server/client_manager.py#L149) as it is coming from the same gRPC peer (i.e. the load balancer) due to this code [here](https://github.com/adap/flower/blob/main/src/py/flwr/server/fleet/grpc_bidi/flower_service_servicer.py#L91), as due to the above I assume that the tuple `ip:port` is not unique for gRPC clients, therefore it cannot be used to identify a unique flower client (as a matter of fact, the `ip:port` are those of the load balancer, not those of the actual client).

As a test and quick hack we have moved to use uuids to identify the clients [here](https://github.com/AI4EOSC/flower/commit/b215d9f3cce1ad8806e296db4fe105a8b7f5c6c9). To my understanding, it is safe to do that way, as the peer identifier is not used elsewhere, but I do not know if it is planned to do something else with them.

### Actual Results

Two and more clients should be able to connect when running behind a load balancer.

--- 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/server/fleet/grpc_bidi/flower_service_servicer.py`
Content:
```
1 # Copyright 2020 Flower Labs 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 """Servicer for FlowerService.
16 
17 Relevant knowledge for reading this modules code:
18 - https://github.com/grpc/grpc/blob/master/doc/statuscodes.md
19 """
20 
21 
22 from typing import Callable, Iterator
23 
24 import grpc
25 from iterators import TimeoutIterator
26 
27 from flwr.proto import transport_pb2_grpc  # pylint: disable=E0611
28 from flwr.proto.transport_pb2 import (  # pylint: disable=E0611
29     ClientMessage,
30     ServerMessage,
31 )
32 from flwr.server.client_manager import ClientManager
33 from flwr.server.fleet.grpc_bidi.grpc_bridge import GrpcBridge, InsWrapper, ResWrapper
34 from flwr.server.fleet.grpc_bidi.grpc_client_proxy import GrpcClientProxy
35 
36 
37 def default_bridge_factory() -> GrpcBridge:
38     """Return GrpcBridge instance."""
39     return GrpcBridge()
40 
41 
42 def default_grpc_client_proxy_factory(cid: str, bridge: GrpcBridge) -> GrpcClientProxy:
43     """Return GrpcClientProxy instance."""
44     return GrpcClientProxy(cid=cid, bridge=bridge)
45 
46 
47 def register_client_proxy(
48     client_manager: ClientManager,
49     client_proxy: GrpcClientProxy,
50     context: grpc.ServicerContext,
51 ) -> bool:
52     """Try registering GrpcClientProxy with ClientManager."""
53     is_success = client_manager.register(client_proxy)
54     if is_success:
55 
56         def rpc_termination_callback() -> None:
57             client_proxy.bridge.close()
58             client_manager.unregister(client_proxy)
59 
60         context.add_callback(rpc_termination_callback)
61     return is_success
62 
63 
64 class FlowerServiceServicer(transport_pb2_grpc.FlowerServiceServicer):
65     """FlowerServiceServicer for bi-directional gRPC message stream."""
66 
67     def __init__(
68         self,
69         client_manager: ClientManager,
70         grpc_bridge_factory: Callable[[], GrpcBridge] = default_bridge_factory,
71         grpc_client_proxy_factory: Callable[
72             [str, GrpcBridge], GrpcClientProxy
73         ] = default_grpc_client_proxy_factory,
74     ) -> None:
75         self.client_manager: ClientManager = client_manager
76         self.grpc_bridge_factory = grpc_bridge_factory
77         self.client_proxy_factory = grpc_client_proxy_factory
78 
79     def Join(  # pylint: disable=invalid-name
80         self,
81         request_iterator: Iterator[ClientMessage],
82         context: grpc.ServicerContext,
83     ) -> Iterator[ServerMessage]:
84         """Facilitate bi-directional streaming of messages between server and client.
85 
86         Invoked by each gRPC client which participates in the network.
87 
88         Protocol:
89         - The first message is sent from the server to the client
90         - Both `ServerMessage` and `ClientMessage` are message "wrappers"
91           wrapping the actual message
92         - The `Join` method is (pretty much) unaware of the protocol
93         """
94         peer: str = context.peer()
95         bridge = self.grpc_bridge_factory()
96         client_proxy = self.client_proxy_factory(peer, bridge)
97         is_success = register_client_proxy(self.client_manager, client_proxy, context)
98 
99         if is_success:
100             # Get iterators
101             client_message_iterator = TimeoutIterator(
102                 iterator=request_iterator, reset_on_next=True
103             )
104             ins_wrapper_iterator = bridge.ins_wrapper_iterator()
105 
106             # All messages will be pushed to client bridge directly
107             while True:
108                 try:
109                     # Get ins_wrapper from bridge and yield server_message
110                     ins_wrapper: InsWrapper = next(ins_wrapper_iterator)
111                     yield ins_wrapper.server_message
112 
113                     # Set current timeout, might be None
114                     if ins_wrapper.timeout is not None:
115                         client_message_iterator.set_timeout(ins_wrapper.timeout)
116 
117                     # Wait for client message
118                     client_message = next(client_message_iterator)
119 
120                     if client_message is client_message_iterator.get_sentinel():
121                         # Important: calling `context.abort` in gRPC always
122                         # raises an exception so that all code after the call to
123                         # `context.abort` will not run. If subsequent code should
124                         # be executed, the `rpc_termination_callback` can be used
125                         # (as shown in the `register_client` function).
126                         details = f"Timeout of {ins_wrapper.timeout}sec was exceeded."
127                         context.abort(
128                             code=grpc.StatusCode.DEADLINE_EXCEEDED,
129                             details=details,
130                         )
131                         # This return statement is only for the linter so it understands
132                         # that client_message in subsequent lines is not None
133                         # It does not understand that `context.abort` will terminate
134                         # this execution context by raising an exception.
135                         return
136 
137                     bridge.set_res_wrapper(
138                         res_wrapper=ResWrapper(client_message=client_message)
139                     )
140                 except StopIteration:
141                     break
142
```
--- 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/server/fleet/grpc_bidi/flower_service_servicer.py b/src/py/flwr/server/fleet/grpc_bidi/flower_service_servicer.py
--- a/src/py/flwr/server/fleet/grpc_bidi/flower_service_servicer.py
+++ b/src/py/flwr/server/fleet/grpc_bidi/flower_service_servicer.py
@@ -18,7 +18,7 @@
 - https://github.com/grpc/grpc/blob/master/doc/statuscodes.md
 """
 
-
+import uuid
 from typing import Callable, Iterator
 
 import grpc
@@ -88,7 +88,11 @@
           wrapping the actual message
         - The `Join` method is (pretty much) unaware of the protocol
         """
-        peer: str = context.peer()
+        # NOTE(aloga): this is a hack. when running flower behind a proxy, the peer can
+        # be the same for different clients (i.e. ip:port) so we use a uuid that is
+        # unique.
+        # peer: str = context.peer()
+        peer = uuid.uuid4().hex
         bridge = self.grpc_bridge_factory()
         client_proxy = self.client_proxy_factory(peer, bridge)
         is_success = register_client_proxy(self.client_manager, client_proxy, context)

{"golden_diff": "diff --git a/src/py/flwr/server/fleet/grpc_bidi/flower_service_servicer.py b/src/py/flwr/server/fleet/grpc_bidi/flower_service_servicer.py\n--- a/src/py/flwr/server/fleet/grpc_bidi/flower_service_servicer.py\n+++ b/src/py/flwr/server/fleet/grpc_bidi/flower_service_servicer.py\n@@ -18,7 +18,7 @@\n - https://github.com/grpc/grpc/blob/master/doc/statuscodes.md\n \"\"\"\n \n-\n+import uuid\n from typing import Callable, Iterator\n \n import grpc\n@@ -88,7 +88,11 @@\n           wrapping the actual message\n         - The `Join` method is (pretty much) unaware of the protocol\n         \"\"\"\n-        peer: str = context.peer()\n+        # NOTE(aloga): this is a hack. when running flower behind a proxy, the peer can\n+        # be the same for different clients (i.e. ip:port) so we use a uuid that is\n+        # unique.\n+        # peer: str = context.peer()\n+        peer = uuid.uuid4().hex\n         bridge = self.grpc_bridge_factory()\n         client_proxy = self.client_proxy_factory(peer, bridge)\n         is_success = register_client_proxy(self.client_manager, client_proxy, context)\n", "issue": "If gRPC server is running behing a load balancer, clients may not be able to connect\n### Describe the bug\r\n\r\nFlower is using the `peer()` method from `grpc.ServicerContext` in order to identify unique flower clients. However, in some situations (like when running the gRPC server behind a load balancer or proxy) different clients can have the same peer identifier (i.e. the same `IP:PORT`), as HTTP/2 connections are multiplexed.\r\n\r\n### Steps/Code to Reproduce\r\n\r\nThis can be tested with a simple [Traefik](https://doc.traefik.io/traefik/) setup as follows (https://doc.traefik.io/traefik/user-guides/grpc/):\r\n\r\n`traefik.yaml`:\r\n```\r\nentryPoints:\r\n  web:\r\n    address: :80\r\n\r\nproviders:\r\n  file:\r\n    filename: grpc.yaml\r\n\r\napi: {}\r\n```\r\n\r\n`grpc.yaml`:\r\n```\r\nhttp:\r\n  routers:\r\n    to-grpc:\r\n      entryPoints: \r\n        - \"web\"\r\n      service: \"srv-grpc\"\r\n      rule: \"HostRegexp(`.*`)\"\r\n\r\n  services:\r\n    srv-grpc:\r\n      loadBalancer:\r\n        servers:\r\n           - url: h2c://127.0.0.1:5000\r\n```\r\n\r\nThe Traefik server can be executed as follows:\r\n```\r\n ./traefik --configFile=traefik.yaml\r\n```\r\n\r\nThen we can launch then a Flower sever with the following code:\r\n```\r\nimport flwr as fl\r\n\r\ndef acc_wavg(metrics):\r\n    n = sum([i for i, _ in metrics])\r\n    acc = sum([i * metric[\"accuracy\"] / n for i, metric in metrics])\r\n    return {\"accuracy\": acc}\r\n \r\nstrategy = fl.server.strategy.FedAvg(\r\n    min_available_clients=2,\r\n    evaluate_metrics_aggregation_fn=acc_wavg\r\n)\r\n \r\nfl.server.start_server(\r\n    server_address=\"0.0.0.0:5000\",\r\n    config=fl.server.ServerConfig(num_rounds=3),\r\n    strategy=strategy,\r\n)\r\n```\r\n\r\n### Expected Results\r\n\r\nWhen we connect two clients to the loadbalancer IP, on port 80, the first one connects and keeps waiting:\r\n\r\n```\r\nINFO flwr 2023-06-28 10:03:14,848 | grpc.py:50 | Opened insecure gRPC connection (no certificates were passed)\r\nDEBUG flwr 2023-06-28 10:03:14,849 | connection.py:39 | ChannelConnectivity.IDLE\r\nDEBUG flwr 2023-06-28 10:03:14,849 | connection.py:39 | ChannelConnectivity.CONNECTING\r\nDEBUG flwr 2023-06-28 10:03:14,875 | connection.py:39 | ChannelConnectivity.READY\r\n```\r\n\r\nHowever, the second one is not connected:\r\n```\r\nDEBUG flwr 2023-06-28 09:37:32,554 | connection.py:39 | ChannelConnectivity.IDLE\r\nDEBUG flwr 2023-06-28 09:37:32,554 | connection.py:39 | ChannelConnectivity.CONNECTING\r\nDEBUG flwr 2023-06-28 09:37:32,578 | connection.py:39 | ChannelConnectivity.READY\r\nDEBUG flwr 2023-06-28 09:37:32,784 | connection.py:113 | gRPC channel closed\r\nTraceback (most recent call last):\r\n  File \"/home/alvaro/w/prj/fl/fl.py\", line 68, in <module>\r\n    fl.client.start_numpy_client(\r\n  File \"/home/alvaro/w/prj/fl/VENV/lib/python3.11/site-packages/flwr/client/app.py\", line 252, in start_numpy_client\r\n    start_client(\r\n  File \"/home/alvaro/w/prj/fl/VENV/lib/python3.11/site-packages/flwr/client/app.py\", line 174, in start_client\r\n    server_message = receive()\r\n                     ^^^^^^^^^\r\n  File \"/home/alvaro/w/prj/fl/VENV/lib/python3.11/site-packages/flwr/client/grpc_client/connection.py\", line 105, in <lambda>\r\n    receive: Callable[[], ServerMessage] = lambda: next(server_message_iterator)\r\n                                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"/home/alvaro/w/prj/fl/VENV/lib/python3.11/site-packages/grpc/_channel.py\", line 475, in __next__\r\n    return self._next()\r\n           ^^^^^^^^^^^^\r\n  File \"/home/alvaro/w/prj/fl/VENV/lib/python3.11/site-packages/grpc/_channel.py\", line 879, in _next\r\n    raise StopIteration()\r\nStopIteration\r\n```\r\n\r\nThis is because the second client connection is discarded [here]https://github.com/adap/flower/blob/main/src/py/flwr/server/client_manager.py#L149) as it is coming from the same gRPC peer (i.e. the load balancer) due to this code [here](https://github.com/adap/flower/blob/main/src/py/flwr/server/fleet/grpc_bidi/flower_service_servicer.py#L91), as due to the above I assume that the tuple `ip:port` is not unique for gRPC clients, therefore it cannot be used to identify a unique flower client (as a matter of fact, the `ip:port` are those of the load balancer, not those of the actual client).\r\n\r\nAs a test and quick hack we have moved to use uuids to identify the clients [here](https://github.com/AI4EOSC/flower/commit/b215d9f3cce1ad8806e296db4fe105a8b7f5c6c9). To my understanding, it is safe to do that way, as the peer identifier is not used elsewhere, but I do not know if it is planned to do something else with them.\r\n\r\n### Actual Results\r\n\r\nTwo and more clients should be able to connect when running behind a load balancer.\n", "before_files": [{"content": "# Copyright 2020 Flower Labs 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\"\"\"Servicer for FlowerService.\n\nRelevant knowledge for reading this modules code:\n- https://github.com/grpc/grpc/blob/master/doc/statuscodes.md\n\"\"\"\n\n\nfrom typing import Callable, Iterator\n\nimport grpc\nfrom iterators import TimeoutIterator\n\nfrom flwr.proto import transport_pb2_grpc  # pylint: disable=E0611\nfrom flwr.proto.transport_pb2 import (  # pylint: disable=E0611\n    ClientMessage,\n    ServerMessage,\n)\nfrom flwr.server.client_manager import ClientManager\nfrom flwr.server.fleet.grpc_bidi.grpc_bridge import GrpcBridge, InsWrapper, ResWrapper\nfrom flwr.server.fleet.grpc_bidi.grpc_client_proxy import GrpcClientProxy\n\n\ndef default_bridge_factory() -> GrpcBridge:\n    \"\"\"Return GrpcBridge instance.\"\"\"\n    return GrpcBridge()\n\n\ndef default_grpc_client_proxy_factory(cid: str, bridge: GrpcBridge) -> GrpcClientProxy:\n    \"\"\"Return GrpcClientProxy instance.\"\"\"\n    return GrpcClientProxy(cid=cid, bridge=bridge)\n\n\ndef register_client_proxy(\n    client_manager: ClientManager,\n    client_proxy: GrpcClientProxy,\n    context: grpc.ServicerContext,\n) -> bool:\n    \"\"\"Try registering GrpcClientProxy with ClientManager.\"\"\"\n    is_success = client_manager.register(client_proxy)\n    if is_success:\n\n        def rpc_termination_callback() -> None:\n            client_proxy.bridge.close()\n            client_manager.unregister(client_proxy)\n\n        context.add_callback(rpc_termination_callback)\n    return is_success\n\n\nclass FlowerServiceServicer(transport_pb2_grpc.FlowerServiceServicer):\n    \"\"\"FlowerServiceServicer for bi-directional gRPC message stream.\"\"\"\n\n    def __init__(\n        self,\n        client_manager: ClientManager,\n        grpc_bridge_factory: Callable[[], GrpcBridge] = default_bridge_factory,\n        grpc_client_proxy_factory: Callable[\n            [str, GrpcBridge], GrpcClientProxy\n        ] = default_grpc_client_proxy_factory,\n    ) -> None:\n        self.client_manager: ClientManager = client_manager\n        self.grpc_bridge_factory = grpc_bridge_factory\n        self.client_proxy_factory = grpc_client_proxy_factory\n\n    def Join(  # pylint: disable=invalid-name\n        self,\n        request_iterator: Iterator[ClientMessage],\n        context: grpc.ServicerContext,\n    ) -> Iterator[ServerMessage]:\n        \"\"\"Facilitate bi-directional streaming of messages between server and client.\n\n        Invoked by each gRPC client which participates in the network.\n\n        Protocol:\n        - The first message is sent from the server to the client\n        - Both `ServerMessage` and `ClientMessage` are message \"wrappers\"\n          wrapping the actual message\n        - The `Join` method is (pretty much) unaware of the protocol\n        \"\"\"\n        peer: str = context.peer()\n        bridge = self.grpc_bridge_factory()\n        client_proxy = self.client_proxy_factory(peer, bridge)\n        is_success = register_client_proxy(self.client_manager, client_proxy, context)\n\n        if is_success:\n            # Get iterators\n            client_message_iterator = TimeoutIterator(\n                iterator=request_iterator, reset_on_next=True\n            )\n            ins_wrapper_iterator = bridge.ins_wrapper_iterator()\n\n            # All messages will be pushed to client bridge directly\n            while True:\n                try:\n                    # Get ins_wrapper from bridge and yield server_message\n                    ins_wrapper: InsWrapper = next(ins_wrapper_iterator)\n                    yield ins_wrapper.server_message\n\n                    # Set current timeout, might be None\n                    if ins_wrapper.timeout is not None:\n                        client_message_iterator.set_timeout(ins_wrapper.timeout)\n\n                    # Wait for client message\n                    client_message = next(client_message_iterator)\n\n                    if client_message is client_message_iterator.get_sentinel():\n                        # Important: calling `context.abort` in gRPC always\n                        # raises an exception so that all code after the call to\n                        # `context.abort` will not run. If subsequent code should\n                        # be executed, the `rpc_termination_callback` can be used\n                        # (as shown in the `register_client` function).\n                        details = f\"Timeout of {ins_wrapper.timeout}sec was exceeded.\"\n                        context.abort(\n                            code=grpc.StatusCode.DEADLINE_EXCEEDED,\n                            details=details,\n                        )\n                        # This return statement is only for the linter so it understands\n                        # that client_message in subsequent lines is not None\n                        # It does not understand that `context.abort` will terminate\n                        # this execution context by raising an exception.\n                        return\n\n                    bridge.set_res_wrapper(\n                        res_wrapper=ResWrapper(client_message=client_message)\n                    )\n                except StopIteration:\n                    break\n", "path": "src/py/flwr/server/fleet/grpc_bidi/flower_service_servicer.py"}], "after_files": [{"content": "# Copyright 2020 Flower Labs 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\"\"\"Servicer for FlowerService.\n\nRelevant knowledge for reading this modules code:\n- https://github.com/grpc/grpc/blob/master/doc/statuscodes.md\n\"\"\"\n\nimport uuid\nfrom typing import Callable, Iterator\n\nimport grpc\nfrom iterators import TimeoutIterator\n\nfrom flwr.proto import transport_pb2_grpc\nfrom flwr.proto.transport_pb2 import ClientMessage, ServerMessage\nfrom flwr.server.client_manager import ClientManager\nfrom flwr.server.fleet.grpc_bidi.grpc_bridge import GrpcBridge, InsWrapper, ResWrapper\nfrom flwr.server.fleet.grpc_bidi.grpc_client_proxy import GrpcClientProxy\n\n\ndef default_bridge_factory() -> GrpcBridge:\n    \"\"\"Return GrpcBridge instance.\"\"\"\n    return GrpcBridge()\n\n\ndef default_grpc_client_proxy_factory(cid: str, bridge: GrpcBridge) -> GrpcClientProxy:\n    \"\"\"Return GrpcClientProxy instance.\"\"\"\n    return GrpcClientProxy(cid=cid, bridge=bridge)\n\n\ndef register_client_proxy(\n    client_manager: ClientManager,\n    client_proxy: GrpcClientProxy,\n    context: grpc.ServicerContext,\n) -> bool:\n    \"\"\"Try registering GrpcClientProxy with ClientManager.\"\"\"\n    is_success = client_manager.register(client_proxy)\n    if is_success:\n\n        def rpc_termination_callback() -> None:\n            client_proxy.bridge.close()\n            client_manager.unregister(client_proxy)\n\n        context.add_callback(rpc_termination_callback)\n    return is_success\n\n\nclass FlowerServiceServicer(transport_pb2_grpc.FlowerServiceServicer):\n    \"\"\"FlowerServiceServicer for bi-directional gRPC message stream.\"\"\"\n\n    def __init__(\n        self,\n        client_manager: ClientManager,\n        grpc_bridge_factory: Callable[[], GrpcBridge] = default_bridge_factory,\n        grpc_client_proxy_factory: Callable[\n            [str, GrpcBridge], GrpcClientProxy\n        ] = default_grpc_client_proxy_factory,\n    ) -> None:\n        self.client_manager: ClientManager = client_manager\n        self.grpc_bridge_factory = grpc_bridge_factory\n        self.client_proxy_factory = grpc_client_proxy_factory\n\n    def Join(  # pylint: disable=invalid-name\n        self,\n        request_iterator: Iterator[ClientMessage],\n        context: grpc.ServicerContext,\n    ) -> Iterator[ServerMessage]:\n        \"\"\"Facilitate bi-directional streaming of messages between server and client.\n\n        Invoked by each gRPC client which participates in the network.\n\n        Protocol:\n        - The first message is sent from the server to the client\n        - Both `ServerMessage` and `ClientMessage` are message \"wrappers\"\n          wrapping the actual message\n        - The `Join` method is (pretty much) unaware of the protocol\n        \"\"\"\n        # NOTE(aloga): this is a hack. when running flower behind a proxy, the peer can\n        # be the same for different clients (i.e. ip:port) so we use a uuid that is\n        # unique.\n        # peer: str = context.peer()\n        peer = uuid.uuid4().hex\n        bridge = self.grpc_bridge_factory()\n        client_proxy = self.client_proxy_factory(peer, bridge)\n        is_success = register_client_proxy(self.client_manager, client_proxy, context)\n\n        if is_success:\n            # Get iterators\n            client_message_iterator = TimeoutIterator(\n                iterator=request_iterator, reset_on_next=True\n            )\n            ins_wrapper_iterator = bridge.ins_wrapper_iterator()\n\n            # All messages will be pushed to client bridge directly\n            while True:\n                try:\n                    # Get ins_wrapper from bridge and yield server_message\n                    ins_wrapper: InsWrapper = next(ins_wrapper_iterator)\n                    yield ins_wrapper.server_message\n\n                    # Set current timeout, might be None\n                    if ins_wrapper.timeout is not None:\n                        client_message_iterator.set_timeout(ins_wrapper.timeout)\n\n                    # Wait for client message\n                    client_message = next(client_message_iterator)\n\n                    if client_message is client_message_iterator.get_sentinel():\n                        # Important: calling `context.abort` in gRPC always\n                        # raises an exception so that all code after the call to\n                        # `context.abort` will not run. If subsequent code should\n                        # be executed, the `rpc_termination_callback` can be used\n                        # (as shown in the `register_client` function).\n                        details = f\"Timeout of {ins_wrapper.timeout}sec was exceeded.\"\n                        context.abort(\n                            code=grpc.StatusCode.DEADLINE_EXCEEDED,\n                            details=details,\n                        )\n                        # This return statement is only for the linter so it understands\n                        # that client_message in subsequent lines is not None\n                        # It does not understand that `context.abort` will terminate\n                        # this execution context by raising an exception.\n                        return\n\n                    bridge.set_res_wrapper(\n                        res_wrapper=ResWrapper(client_message=client_message)\n                    )\n                except StopIteration:\n                    break\n", "path": "src/py/flwr/server/fleet/grpc_bidi/flower_service_servicer.py"}]}