We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
log_exception error
This error happens during log_exception
```
nvflare/private/event.py", line 62, in fire_event
    h.log_exception(
TypeError: log_exception() got an unexpected keyword argument 'local_logging'
FL client execution exception: log_exception() got an unexpected keyword argument 'local_logging'
2022-03-17 18:29:04,424 - ProcessExecutor - INFO - process finished with execution code: 0
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `nvflare/widgets/fed_event.py`
Content:
```
1 # Copyright (c) 2021-2022, NVIDIA CORPORATION.  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 import threading
16 import time
17 
18 from nvflare.apis.client_engine_spec import ClientEngineSpec
19 from nvflare.apis.event_type import EventType
20 from nvflare.apis.fl_constant import EventScope, FedEventHeader, FLContextKey, ReservedKey, ReturnCode
21 from nvflare.apis.fl_context import FLContext
22 from nvflare.apis.server_engine_spec import ServerEngineSpec
23 from nvflare.apis.shareable import Shareable, make_reply
24 from nvflare.widgets.widget import Widget
25 
26 FED_EVENT_TOPIC = "fed.event"
27 
28 
29 class FedEventRunner(Widget):
30     def __init__(self, topic=FED_EVENT_TOPIC):
31         """Init FedEventRunner.
32 
33         The FedEventRunner handles posting and receiving of fed events.
34         The system will do its best to fire off all events in the queue before shutdown
35         using the ABOUT_TO_END_RUN event and a grace period during END_RUN.
36 
37         Args:
38             topic: the fed event topic to be handled. Defaults to 'fed.event'
39         """
40         Widget.__init__(self)
41         self.topic = topic
42         self.abort_signal = None
43         self.asked_to_stop = False
44         self.asked_to_flush = False
45         self.regular_interval = 0.001
46         self.grace_period = 2
47         self.flush_wait = 2
48         self.engine = None
49         self.last_timestamps = {}  # client name => last_timestamp
50         self.in_events = []
51         self.in_lock = threading.Lock()
52         self.poster = threading.Thread(target=self._post, args=())
53 
54     def handle_event(self, event_type: str, fl_ctx: FLContext):
55         if event_type == EventType.START_RUN:
56             self.engine = fl_ctx.get_engine()
57             self.engine.register_aux_message_handler(topic=self.topic, message_handle_func=self._receive)
58             self.abort_signal = fl_ctx.get_run_abort_signal()
59             self.asked_to_stop = False
60             self.asked_to_flush = False
61             self.poster.start()
62         elif event_type == EventType.ABOUT_TO_END_RUN:
63             self.asked_to_flush = True
64             # delay self.flush_wait seconds so
65             # _post can empty the queue before
66             # END_RUN is fired
67             time.sleep(self.flush_wait)
68         elif event_type == EventType.END_RUN:
69             self.asked_to_stop = True
70             if self.poster.is_alive():
71                 self.poster.join()
72         else:
73             # handle outgoing fed events
74             event_scope = fl_ctx.get_prop(key=FLContextKey.EVENT_SCOPE, default=EventScope.LOCAL)
75             if event_scope != EventScope.FEDERATION:
76                 return
77 
78             event_data = fl_ctx.get_prop(FLContextKey.EVENT_DATA, None)
79             if not isinstance(event_data, Shareable):
80                 self.log_error(fl_ctx, "bad fed event: expect data to be Shareable but got {}".format(type(event_data)))
81                 return
82 
83             direction = event_data.get_header(FedEventHeader.DIRECTION, "out")
84             if direction != "out":
85                 # ignore incoming events
86                 return
87 
88             event_data.set_header(FedEventHeader.EVENT_TYPE, event_type)
89             event_data.set_header(FedEventHeader.ORIGIN, fl_ctx.get_identity_name())
90             event_data.set_header(FedEventHeader.TIMESTAMP, time.time())
91 
92             targets = event_data.get_header(FedEventHeader.TARGETS, None)
93             self.fire_and_forget_request(request=event_data, fl_ctx=fl_ctx, targets=targets)
94 
95     def fire_and_forget_request(self, request: Shareable, fl_ctx: FLContext, targets=None):
96         pass
97 
98     def _receive(self, topic: str, request: Shareable, fl_ctx: FLContext) -> Shareable:
99         peer_name = request.get_peer_prop(ReservedKey.IDENTITY_NAME, None)
100         if not peer_name:
101             self.log_error(fl_ctx, "missing identity name of the data sender")
102             return make_reply(ReturnCode.MISSING_PEER_CONTEXT)
103 
104         timestamp = request.get_header(FedEventHeader.TIMESTAMP, None)
105         if timestamp is None:
106             self.log_error(fl_ctx, "missing timestamp in incoming fed event")
107             return make_reply(ReturnCode.BAD_REQUEST_DATA)
108 
109         event_type = request.get_header(FedEventHeader.EVENT_TYPE, None)
110         if event_type is None:
111             self.log_error(fl_ctx, "missing event_type in incoming fed event")
112             return make_reply(ReturnCode.BAD_REQUEST_DATA)
113 
114         with self.in_lock:
115             last_timestamp = self.last_timestamps.get(peer_name, None)
116             if last_timestamp is None or timestamp > last_timestamp:
117                 # we only keep new items, in case the peer somehow sent old items
118                 request.set_header(FedEventHeader.DIRECTION, "in")
119                 self.in_events.append(request)
120                 self.last_timestamps[peer_name] = timestamp
121 
122         # NOTE: we do not fire event here since event process could take time.
123         # Instead we simply add the package to the queue and return quickly.
124         # The posting of events will be handled in the poster thread
125         return make_reply(ReturnCode.OK)
126 
127     def _post(self):
128         """Post an event.
129 
130         During ABOUT_TO_END_RUN, sleep_time is 0 and system will flush
131          in_events by firing events without delay.
132 
133         During END_RUN, system will wait for self.grace_period, even the queue is empty,
134         so any new item can be processed.
135 
136         However, since the system does not guarantee the receiving side of _post is still
137         alive, we catch the exception and show warning messages to users if events can not
138         be handled by receiving side.
139         """
140         sleep_time = self.regular_interval
141         countdown = self.grace_period
142         while True:
143             time.sleep(sleep_time)
144             if self.abort_signal.triggered:
145                 break
146             n = len(self.in_events)
147             if n > 0:
148                 if self.asked_to_flush:
149                     sleep_time = 0
150                 else:
151                     sleep_time = self.regular_interval
152                 with self.in_lock:
153                     event_to_post = self.in_events.pop(0)
154             elif self.asked_to_stop:
155                 # the queue is empty and we are asked to stop.  Give
156                 # it self.grace_period seconds to wait, then
157                 # exit.
158                 if countdown < 0:
159                     break
160                 else:
161                     countdown = countdown - 1
162                     time.sleep(1)
163                     continue
164             else:
165                 sleep_time = min(sleep_time * 2, 1)
166                 continue
167 
168             with self.engine.new_context() as fl_ctx:
169                 if self.asked_to_stop:
170                     self.log_warning(fl_ctx, f"{n} items remained in in_events.  Will stop when it reaches 0.")
171                 fl_ctx.set_prop(key=FLContextKey.EVENT_DATA, value=event_to_post, private=True, sticky=False)
172                 fl_ctx.set_prop(key=FLContextKey.EVENT_SCOPE, value=EventScope.FEDERATION, private=True, sticky=False)
173 
174                 event_type = event_to_post.get_header(FedEventHeader.EVENT_TYPE)
175                 try:
176                     self.engine.fire_event(event_type=event_type, fl_ctx=fl_ctx)
177                 except BaseException as e:
178                     if self.asked_to_stop:
179                         self.log_warning(fl_ctx, f"event {event_to_post} fired unsuccessfully during END_RUN")
180                     else:
181                         raise e
182 
183 
184 class ServerFedEventRunner(FedEventRunner):
185     def __init__(self, topic=FED_EVENT_TOPIC):
186         """Init ServerFedEventRunner."""
187         FedEventRunner.__init__(self, topic)
188 
189     def fire_and_forget_request(self, request: Shareable, fl_ctx: FLContext, targets=None):
190         if not isinstance(self.engine, ServerEngineSpec):
191             raise TypeError("self.engine must be ServerEngineSpec but got {}".format(type(self.engine)))
192         self.engine.fire_and_forget_aux_request(
193             topic=self.topic,
194             targets=targets,
195             request=request,
196             fl_ctx=fl_ctx,
197         )
198 
199 
200 class ClientFedEventRunner(FedEventRunner):
201     def __init__(self, topic=FED_EVENT_TOPIC):
202         """Init ClientFedEventRunner."""
203         FedEventRunner.__init__(self, topic)
204         self.ready = False
205 
206     def handle_event(self, event_type: str, fl_ctx: FLContext):
207         super().handle_event(event_type, fl_ctx)
208 
209         if event_type == EventType.START_RUN:
210             self.ready = True
211 
212     def fire_and_forget_request(self, request: Shareable, fl_ctx: FLContext, targets=None):
213         if not self.ready:
214             self.log_warning(fl_ctx, "Engine in not ready, skip the aux event firing.", local_logging=True)
215             return
216 
217         if not isinstance(self.engine, ClientEngineSpec):
218             raise TypeError("self.engine must be ClientEngineSpec but got {}".format(type(self.engine)))
219         self.engine.fire_and_forget_aux_request(topic=self.topic, request=request, fl_ctx=fl_ctx)
220
```

Path: `nvflare/private/event.py`
Content:
```
1 # Copyright (c) 2021-2022, NVIDIA CORPORATION.  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 import uuid
16 
17 from nvflare.apis.fl_component import FLComponent
18 from nvflare.apis.fl_constant import EventScope, FLContextKey
19 from nvflare.apis.fl_context import FLContext
20 
21 # do not use underscore as key name; otherwise it cannot be removed from ctx
22 _KEY_EVENT_DEPTH = "###event_depth"
23 _MAX_EVENT_DEPTH = 20
24 
25 
26 def fire_event(event: str, handlers: list, ctx: FLContext):
27     """Fires the specified event and invokes the list of handlers.
28 
29     Args:
30         event: the event to be fired
31         handlers: handlers to be invoked
32         ctx: context for cross-component data sharing
33 
34     Returns: N/A
35 
36     """
37     event_id = str(uuid.uuid4())
38     event_data = ctx.get_prop(FLContextKey.EVENT_DATA, None)
39     event_origin = ctx.get_prop(FLContextKey.EVENT_ORIGIN, None)
40     event_scope = ctx.get_prop(FLContextKey.EVENT_SCOPE, EventScope.LOCAL)
41 
42     depth = ctx.get_prop(_KEY_EVENT_DEPTH, 0)
43     if depth > _MAX_EVENT_DEPTH:
44         # too many recursive event calls
45         raise RuntimeError("Recursive event calls too deep (>{})".format(_MAX_EVENT_DEPTH))
46 
47     ctx.set_prop(key=_KEY_EVENT_DEPTH, value=depth + 1, private=True, sticky=False)
48 
49     if handlers:
50         for h in handlers:
51             if not isinstance(h, FLComponent):
52                 raise TypeError("handler must be FLComponent but got {}".format(type(h)))
53             try:
54                 # since events could be recursive (a handler fires another event) on the same fl_ctx,
55                 # we need to reset these key values into the fl_ctx
56                 ctx.set_prop(key=FLContextKey.EVENT_ID, value=event_id, private=True, sticky=False)
57                 ctx.set_prop(key=FLContextKey.EVENT_DATA, value=event_data, private=True, sticky=False)
58                 ctx.set_prop(key=FLContextKey.EVENT_ORIGIN, value=event_origin, private=True, sticky=False)
59                 ctx.set_prop(key=FLContextKey.EVENT_SCOPE, value=event_scope, private=True, sticky=False)
60                 h.handle_event(event, ctx)
61             except:
62                 h.log_exception(
63                     ctx, 'exception when handling event "{}"'.format(event), fire_event=False, local_logging=True
64                 )
65 
66     ctx.set_prop(key=_KEY_EVENT_DEPTH, value=depth, private=True, sticky=False)
67
```
--- 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/nvflare/private/event.py b/nvflare/private/event.py
--- a/nvflare/private/event.py
+++ b/nvflare/private/event.py
@@ -59,8 +59,6 @@
                 ctx.set_prop(key=FLContextKey.EVENT_SCOPE, value=event_scope, private=True, sticky=False)
                 h.handle_event(event, ctx)
             except:
-                h.log_exception(
-                    ctx, 'exception when handling event "{}"'.format(event), fire_event=False, local_logging=True
-                )
+                h.log_exception(ctx, 'exception when handling event "{}"'.format(event), fire_event=False)
 
     ctx.set_prop(key=_KEY_EVENT_DEPTH, value=depth, private=True, sticky=False)
diff --git a/nvflare/widgets/fed_event.py b/nvflare/widgets/fed_event.py
--- a/nvflare/widgets/fed_event.py
+++ b/nvflare/widgets/fed_event.py
@@ -211,7 +211,7 @@
 
     def fire_and_forget_request(self, request: Shareable, fl_ctx: FLContext, targets=None):
         if not self.ready:
-            self.log_warning(fl_ctx, "Engine in not ready, skip the aux event firing.", local_logging=True)
+            self.log_warning(fl_ctx, "Engine in not ready, skip the aux event firing.")
             return
 
         if not isinstance(self.engine, ClientEngineSpec):

{"golden_diff": "diff --git a/nvflare/private/event.py b/nvflare/private/event.py\n--- a/nvflare/private/event.py\n+++ b/nvflare/private/event.py\n@@ -59,8 +59,6 @@\n                 ctx.set_prop(key=FLContextKey.EVENT_SCOPE, value=event_scope, private=True, sticky=False)\n                 h.handle_event(event, ctx)\n             except:\n-                h.log_exception(\n-                    ctx, 'exception when handling event \"{}\"'.format(event), fire_event=False, local_logging=True\n-                )\n+                h.log_exception(ctx, 'exception when handling event \"{}\"'.format(event), fire_event=False)\n \n     ctx.set_prop(key=_KEY_EVENT_DEPTH, value=depth, private=True, sticky=False)\ndiff --git a/nvflare/widgets/fed_event.py b/nvflare/widgets/fed_event.py\n--- a/nvflare/widgets/fed_event.py\n+++ b/nvflare/widgets/fed_event.py\n@@ -211,7 +211,7 @@\n \n     def fire_and_forget_request(self, request: Shareable, fl_ctx: FLContext, targets=None):\n         if not self.ready:\n-            self.log_warning(fl_ctx, \"Engine in not ready, skip the aux event firing.\", local_logging=True)\n+            self.log_warning(fl_ctx, \"Engine in not ready, skip the aux event firing.\")\n             return\n \n         if not isinstance(self.engine, ClientEngineSpec):\n", "issue": "log_exception error\nThis error happens during log_exception\r\n```\r\nnvflare/private/event.py\", line 62, in fire_event\r\n    h.log_exception(\r\nTypeError: log_exception() got an unexpected keyword argument 'local_logging'\r\nFL client execution exception: log_exception() got an unexpected keyword argument 'local_logging'\r\n2022-03-17 18:29:04,424 - ProcessExecutor - INFO - process finished with execution code: 0\r\n```\n", "before_files": [{"content": "# Copyright (c) 2021-2022, NVIDIA CORPORATION.  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\nimport threading\nimport time\n\nfrom nvflare.apis.client_engine_spec import ClientEngineSpec\nfrom nvflare.apis.event_type import EventType\nfrom nvflare.apis.fl_constant import EventScope, FedEventHeader, FLContextKey, ReservedKey, ReturnCode\nfrom nvflare.apis.fl_context import FLContext\nfrom nvflare.apis.server_engine_spec import ServerEngineSpec\nfrom nvflare.apis.shareable import Shareable, make_reply\nfrom nvflare.widgets.widget import Widget\n\nFED_EVENT_TOPIC = \"fed.event\"\n\n\nclass FedEventRunner(Widget):\n    def __init__(self, topic=FED_EVENT_TOPIC):\n        \"\"\"Init FedEventRunner.\n\n        The FedEventRunner handles posting and receiving of fed events.\n        The system will do its best to fire off all events in the queue before shutdown\n        using the ABOUT_TO_END_RUN event and a grace period during END_RUN.\n\n        Args:\n            topic: the fed event topic to be handled. Defaults to 'fed.event'\n        \"\"\"\n        Widget.__init__(self)\n        self.topic = topic\n        self.abort_signal = None\n        self.asked_to_stop = False\n        self.asked_to_flush = False\n        self.regular_interval = 0.001\n        self.grace_period = 2\n        self.flush_wait = 2\n        self.engine = None\n        self.last_timestamps = {}  # client name => last_timestamp\n        self.in_events = []\n        self.in_lock = threading.Lock()\n        self.poster = threading.Thread(target=self._post, args=())\n\n    def handle_event(self, event_type: str, fl_ctx: FLContext):\n        if event_type == EventType.START_RUN:\n            self.engine = fl_ctx.get_engine()\n            self.engine.register_aux_message_handler(topic=self.topic, message_handle_func=self._receive)\n            self.abort_signal = fl_ctx.get_run_abort_signal()\n            self.asked_to_stop = False\n            self.asked_to_flush = False\n            self.poster.start()\n        elif event_type == EventType.ABOUT_TO_END_RUN:\n            self.asked_to_flush = True\n            # delay self.flush_wait seconds so\n            # _post can empty the queue before\n            # END_RUN is fired\n            time.sleep(self.flush_wait)\n        elif event_type == EventType.END_RUN:\n            self.asked_to_stop = True\n            if self.poster.is_alive():\n                self.poster.join()\n        else:\n            # handle outgoing fed events\n            event_scope = fl_ctx.get_prop(key=FLContextKey.EVENT_SCOPE, default=EventScope.LOCAL)\n            if event_scope != EventScope.FEDERATION:\n                return\n\n            event_data = fl_ctx.get_prop(FLContextKey.EVENT_DATA, None)\n            if not isinstance(event_data, Shareable):\n                self.log_error(fl_ctx, \"bad fed event: expect data to be Shareable but got {}\".format(type(event_data)))\n                return\n\n            direction = event_data.get_header(FedEventHeader.DIRECTION, \"out\")\n            if direction != \"out\":\n                # ignore incoming events\n                return\n\n            event_data.set_header(FedEventHeader.EVENT_TYPE, event_type)\n            event_data.set_header(FedEventHeader.ORIGIN, fl_ctx.get_identity_name())\n            event_data.set_header(FedEventHeader.TIMESTAMP, time.time())\n\n            targets = event_data.get_header(FedEventHeader.TARGETS, None)\n            self.fire_and_forget_request(request=event_data, fl_ctx=fl_ctx, targets=targets)\n\n    def fire_and_forget_request(self, request: Shareable, fl_ctx: FLContext, targets=None):\n        pass\n\n    def _receive(self, topic: str, request: Shareable, fl_ctx: FLContext) -> Shareable:\n        peer_name = request.get_peer_prop(ReservedKey.IDENTITY_NAME, None)\n        if not peer_name:\n            self.log_error(fl_ctx, \"missing identity name of the data sender\")\n            return make_reply(ReturnCode.MISSING_PEER_CONTEXT)\n\n        timestamp = request.get_header(FedEventHeader.TIMESTAMP, None)\n        if timestamp is None:\n            self.log_error(fl_ctx, \"missing timestamp in incoming fed event\")\n            return make_reply(ReturnCode.BAD_REQUEST_DATA)\n\n        event_type = request.get_header(FedEventHeader.EVENT_TYPE, None)\n        if event_type is None:\n            self.log_error(fl_ctx, \"missing event_type in incoming fed event\")\n            return make_reply(ReturnCode.BAD_REQUEST_DATA)\n\n        with self.in_lock:\n            last_timestamp = self.last_timestamps.get(peer_name, None)\n            if last_timestamp is None or timestamp > last_timestamp:\n                # we only keep new items, in case the peer somehow sent old items\n                request.set_header(FedEventHeader.DIRECTION, \"in\")\n                self.in_events.append(request)\n                self.last_timestamps[peer_name] = timestamp\n\n        # NOTE: we do not fire event here since event process could take time.\n        # Instead we simply add the package to the queue and return quickly.\n        # The posting of events will be handled in the poster thread\n        return make_reply(ReturnCode.OK)\n\n    def _post(self):\n        \"\"\"Post an event.\n\n        During ABOUT_TO_END_RUN, sleep_time is 0 and system will flush\n         in_events by firing events without delay.\n\n        During END_RUN, system will wait for self.grace_period, even the queue is empty,\n        so any new item can be processed.\n\n        However, since the system does not guarantee the receiving side of _post is still\n        alive, we catch the exception and show warning messages to users if events can not\n        be handled by receiving side.\n        \"\"\"\n        sleep_time = self.regular_interval\n        countdown = self.grace_period\n        while True:\n            time.sleep(sleep_time)\n            if self.abort_signal.triggered:\n                break\n            n = len(self.in_events)\n            if n > 0:\n                if self.asked_to_flush:\n                    sleep_time = 0\n                else:\n                    sleep_time = self.regular_interval\n                with self.in_lock:\n                    event_to_post = self.in_events.pop(0)\n            elif self.asked_to_stop:\n                # the queue is empty and we are asked to stop.  Give\n                # it self.grace_period seconds to wait, then\n                # exit.\n                if countdown < 0:\n                    break\n                else:\n                    countdown = countdown - 1\n                    time.sleep(1)\n                    continue\n            else:\n                sleep_time = min(sleep_time * 2, 1)\n                continue\n\n            with self.engine.new_context() as fl_ctx:\n                if self.asked_to_stop:\n                    self.log_warning(fl_ctx, f\"{n} items remained in in_events.  Will stop when it reaches 0.\")\n                fl_ctx.set_prop(key=FLContextKey.EVENT_DATA, value=event_to_post, private=True, sticky=False)\n                fl_ctx.set_prop(key=FLContextKey.EVENT_SCOPE, value=EventScope.FEDERATION, private=True, sticky=False)\n\n                event_type = event_to_post.get_header(FedEventHeader.EVENT_TYPE)\n                try:\n                    self.engine.fire_event(event_type=event_type, fl_ctx=fl_ctx)\n                except BaseException as e:\n                    if self.asked_to_stop:\n                        self.log_warning(fl_ctx, f\"event {event_to_post} fired unsuccessfully during END_RUN\")\n                    else:\n                        raise e\n\n\nclass ServerFedEventRunner(FedEventRunner):\n    def __init__(self, topic=FED_EVENT_TOPIC):\n        \"\"\"Init ServerFedEventRunner.\"\"\"\n        FedEventRunner.__init__(self, topic)\n\n    def fire_and_forget_request(self, request: Shareable, fl_ctx: FLContext, targets=None):\n        if not isinstance(self.engine, ServerEngineSpec):\n            raise TypeError(\"self.engine must be ServerEngineSpec but got {}\".format(type(self.engine)))\n        self.engine.fire_and_forget_aux_request(\n            topic=self.topic,\n            targets=targets,\n            request=request,\n            fl_ctx=fl_ctx,\n        )\n\n\nclass ClientFedEventRunner(FedEventRunner):\n    def __init__(self, topic=FED_EVENT_TOPIC):\n        \"\"\"Init ClientFedEventRunner.\"\"\"\n        FedEventRunner.__init__(self, topic)\n        self.ready = False\n\n    def handle_event(self, event_type: str, fl_ctx: FLContext):\n        super().handle_event(event_type, fl_ctx)\n\n        if event_type == EventType.START_RUN:\n            self.ready = True\n\n    def fire_and_forget_request(self, request: Shareable, fl_ctx: FLContext, targets=None):\n        if not self.ready:\n            self.log_warning(fl_ctx, \"Engine in not ready, skip the aux event firing.\", local_logging=True)\n            return\n\n        if not isinstance(self.engine, ClientEngineSpec):\n            raise TypeError(\"self.engine must be ClientEngineSpec but got {}\".format(type(self.engine)))\n        self.engine.fire_and_forget_aux_request(topic=self.topic, request=request, fl_ctx=fl_ctx)\n", "path": "nvflare/widgets/fed_event.py"}, {"content": "# Copyright (c) 2021-2022, NVIDIA CORPORATION.  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\nimport uuid\n\nfrom nvflare.apis.fl_component import FLComponent\nfrom nvflare.apis.fl_constant import EventScope, FLContextKey\nfrom nvflare.apis.fl_context import FLContext\n\n# do not use underscore as key name; otherwise it cannot be removed from ctx\n_KEY_EVENT_DEPTH = \"###event_depth\"\n_MAX_EVENT_DEPTH = 20\n\n\ndef fire_event(event: str, handlers: list, ctx: FLContext):\n    \"\"\"Fires the specified event and invokes the list of handlers.\n\n    Args:\n        event: the event to be fired\n        handlers: handlers to be invoked\n        ctx: context for cross-component data sharing\n\n    Returns: N/A\n\n    \"\"\"\n    event_id = str(uuid.uuid4())\n    event_data = ctx.get_prop(FLContextKey.EVENT_DATA, None)\n    event_origin = ctx.get_prop(FLContextKey.EVENT_ORIGIN, None)\n    event_scope = ctx.get_prop(FLContextKey.EVENT_SCOPE, EventScope.LOCAL)\n\n    depth = ctx.get_prop(_KEY_EVENT_DEPTH, 0)\n    if depth > _MAX_EVENT_DEPTH:\n        # too many recursive event calls\n        raise RuntimeError(\"Recursive event calls too deep (>{})\".format(_MAX_EVENT_DEPTH))\n\n    ctx.set_prop(key=_KEY_EVENT_DEPTH, value=depth + 1, private=True, sticky=False)\n\n    if handlers:\n        for h in handlers:\n            if not isinstance(h, FLComponent):\n                raise TypeError(\"handler must be FLComponent but got {}\".format(type(h)))\n            try:\n                # since events could be recursive (a handler fires another event) on the same fl_ctx,\n                # we need to reset these key values into the fl_ctx\n                ctx.set_prop(key=FLContextKey.EVENT_ID, value=event_id, private=True, sticky=False)\n                ctx.set_prop(key=FLContextKey.EVENT_DATA, value=event_data, private=True, sticky=False)\n                ctx.set_prop(key=FLContextKey.EVENT_ORIGIN, value=event_origin, private=True, sticky=False)\n                ctx.set_prop(key=FLContextKey.EVENT_SCOPE, value=event_scope, private=True, sticky=False)\n                h.handle_event(event, ctx)\n            except:\n                h.log_exception(\n                    ctx, 'exception when handling event \"{}\"'.format(event), fire_event=False, local_logging=True\n                )\n\n    ctx.set_prop(key=_KEY_EVENT_DEPTH, value=depth, private=True, sticky=False)\n", "path": "nvflare/private/event.py"}], "after_files": [{"content": "# Copyright (c) 2021-2022, NVIDIA CORPORATION.  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\nimport threading\nimport time\n\nfrom nvflare.apis.client_engine_spec import ClientEngineSpec\nfrom nvflare.apis.event_type import EventType\nfrom nvflare.apis.fl_constant import EventScope, FedEventHeader, FLContextKey, ReservedKey, ReturnCode\nfrom nvflare.apis.fl_context import FLContext\nfrom nvflare.apis.server_engine_spec import ServerEngineSpec\nfrom nvflare.apis.shareable import Shareable, make_reply\nfrom nvflare.widgets.widget import Widget\n\nFED_EVENT_TOPIC = \"fed.event\"\n\n\nclass FedEventRunner(Widget):\n    def __init__(self, topic=FED_EVENT_TOPIC):\n        \"\"\"Init FedEventRunner.\n\n        The FedEventRunner handles posting and receiving of fed events.\n        The system will do its best to fire off all events in the queue before shutdown\n        using the ABOUT_TO_END_RUN event and a grace period during END_RUN.\n\n        Args:\n            topic: the fed event topic to be handled. Defaults to 'fed.event'\n        \"\"\"\n        Widget.__init__(self)\n        self.topic = topic\n        self.abort_signal = None\n        self.asked_to_stop = False\n        self.asked_to_flush = False\n        self.regular_interval = 0.001\n        self.grace_period = 2\n        self.flush_wait = 2\n        self.engine = None\n        self.last_timestamps = {}  # client name => last_timestamp\n        self.in_events = []\n        self.in_lock = threading.Lock()\n        self.poster = threading.Thread(target=self._post, args=())\n\n    def handle_event(self, event_type: str, fl_ctx: FLContext):\n        if event_type == EventType.START_RUN:\n            self.engine = fl_ctx.get_engine()\n            self.engine.register_aux_message_handler(topic=self.topic, message_handle_func=self._receive)\n            self.abort_signal = fl_ctx.get_run_abort_signal()\n            self.asked_to_stop = False\n            self.asked_to_flush = False\n            self.poster.start()\n        elif event_type == EventType.ABOUT_TO_END_RUN:\n            self.asked_to_flush = True\n            # delay self.flush_wait seconds so\n            # _post can empty the queue before\n            # END_RUN is fired\n            time.sleep(self.flush_wait)\n        elif event_type == EventType.END_RUN:\n            self.asked_to_stop = True\n            if self.poster.is_alive():\n                self.poster.join()\n        else:\n            # handle outgoing fed events\n            event_scope = fl_ctx.get_prop(key=FLContextKey.EVENT_SCOPE, default=EventScope.LOCAL)\n            if event_scope != EventScope.FEDERATION:\n                return\n\n            event_data = fl_ctx.get_prop(FLContextKey.EVENT_DATA, None)\n            if not isinstance(event_data, Shareable):\n                self.log_error(fl_ctx, \"bad fed event: expect data to be Shareable but got {}\".format(type(event_data)))\n                return\n\n            direction = event_data.get_header(FedEventHeader.DIRECTION, \"out\")\n            if direction != \"out\":\n                # ignore incoming events\n                return\n\n            event_data.set_header(FedEventHeader.EVENT_TYPE, event_type)\n            event_data.set_header(FedEventHeader.ORIGIN, fl_ctx.get_identity_name())\n            event_data.set_header(FedEventHeader.TIMESTAMP, time.time())\n\n            targets = event_data.get_header(FedEventHeader.TARGETS, None)\n            self.fire_and_forget_request(request=event_data, fl_ctx=fl_ctx, targets=targets)\n\n    def fire_and_forget_request(self, request: Shareable, fl_ctx: FLContext, targets=None):\n        pass\n\n    def _receive(self, topic: str, request: Shareable, fl_ctx: FLContext) -> Shareable:\n        peer_name = request.get_peer_prop(ReservedKey.IDENTITY_NAME, None)\n        if not peer_name:\n            self.log_error(fl_ctx, \"missing identity name of the data sender\")\n            return make_reply(ReturnCode.MISSING_PEER_CONTEXT)\n\n        timestamp = request.get_header(FedEventHeader.TIMESTAMP, None)\n        if timestamp is None:\n            self.log_error(fl_ctx, \"missing timestamp in incoming fed event\")\n            return make_reply(ReturnCode.BAD_REQUEST_DATA)\n\n        event_type = request.get_header(FedEventHeader.EVENT_TYPE, None)\n        if event_type is None:\n            self.log_error(fl_ctx, \"missing event_type in incoming fed event\")\n            return make_reply(ReturnCode.BAD_REQUEST_DATA)\n\n        with self.in_lock:\n            last_timestamp = self.last_timestamps.get(peer_name, None)\n            if last_timestamp is None or timestamp > last_timestamp:\n                # we only keep new items, in case the peer somehow sent old items\n                request.set_header(FedEventHeader.DIRECTION, \"in\")\n                self.in_events.append(request)\n                self.last_timestamps[peer_name] = timestamp\n\n        # NOTE: we do not fire event here since event process could take time.\n        # Instead we simply add the package to the queue and return quickly.\n        # The posting of events will be handled in the poster thread\n        return make_reply(ReturnCode.OK)\n\n    def _post(self):\n        \"\"\"Post an event.\n\n        During ABOUT_TO_END_RUN, sleep_time is 0 and system will flush\n         in_events by firing events without delay.\n\n        During END_RUN, system will wait for self.grace_period, even the queue is empty,\n        so any new item can be processed.\n\n        However, since the system does not guarantee the receiving side of _post is still\n        alive, we catch the exception and show warning messages to users if events can not\n        be handled by receiving side.\n        \"\"\"\n        sleep_time = self.regular_interval\n        countdown = self.grace_period\n        while True:\n            time.sleep(sleep_time)\n            if self.abort_signal.triggered:\n                break\n            n = len(self.in_events)\n            if n > 0:\n                if self.asked_to_flush:\n                    sleep_time = 0\n                else:\n                    sleep_time = self.regular_interval\n                with self.in_lock:\n                    event_to_post = self.in_events.pop(0)\n            elif self.asked_to_stop:\n                # the queue is empty and we are asked to stop.  Give\n                # it self.grace_period seconds to wait, then\n                # exit.\n                if countdown < 0:\n                    break\n                else:\n                    countdown = countdown - 1\n                    time.sleep(1)\n                    continue\n            else:\n                sleep_time = min(sleep_time * 2, 1)\n                continue\n\n            with self.engine.new_context() as fl_ctx:\n                if self.asked_to_stop:\n                    self.log_warning(fl_ctx, f\"{n} items remained in in_events.  Will stop when it reaches 0.\")\n                fl_ctx.set_prop(key=FLContextKey.EVENT_DATA, value=event_to_post, private=True, sticky=False)\n                fl_ctx.set_prop(key=FLContextKey.EVENT_SCOPE, value=EventScope.FEDERATION, private=True, sticky=False)\n\n                event_type = event_to_post.get_header(FedEventHeader.EVENT_TYPE)\n                try:\n                    self.engine.fire_event(event_type=event_type, fl_ctx=fl_ctx)\n                except BaseException as e:\n                    if self.asked_to_stop:\n                        self.log_warning(fl_ctx, f\"event {event_to_post} fired unsuccessfully during END_RUN\")\n                    else:\n                        raise e\n\n\nclass ServerFedEventRunner(FedEventRunner):\n    def __init__(self, topic=FED_EVENT_TOPIC):\n        \"\"\"Init ServerFedEventRunner.\"\"\"\n        FedEventRunner.__init__(self, topic)\n\n    def fire_and_forget_request(self, request: Shareable, fl_ctx: FLContext, targets=None):\n        if not isinstance(self.engine, ServerEngineSpec):\n            raise TypeError(\"self.engine must be ServerEngineSpec but got {}\".format(type(self.engine)))\n        self.engine.fire_and_forget_aux_request(\n            topic=self.topic,\n            targets=targets,\n            request=request,\n            fl_ctx=fl_ctx,\n        )\n\n\nclass ClientFedEventRunner(FedEventRunner):\n    def __init__(self, topic=FED_EVENT_TOPIC):\n        \"\"\"Init ClientFedEventRunner.\"\"\"\n        FedEventRunner.__init__(self, topic)\n        self.ready = False\n\n    def handle_event(self, event_type: str, fl_ctx: FLContext):\n        super().handle_event(event_type, fl_ctx)\n\n        if event_type == EventType.START_RUN:\n            self.ready = True\n\n    def fire_and_forget_request(self, request: Shareable, fl_ctx: FLContext, targets=None):\n        if not self.ready:\n            self.log_warning(fl_ctx, \"Engine in not ready, skip the aux event firing.\")\n            return\n\n        if not isinstance(self.engine, ClientEngineSpec):\n            raise TypeError(\"self.engine must be ClientEngineSpec but got {}\".format(type(self.engine)))\n        self.engine.fire_and_forget_aux_request(topic=self.topic, request=request, fl_ctx=fl_ctx)\n", "path": "nvflare/widgets/fed_event.py"}, {"content": "# Copyright (c) 2021-2022, NVIDIA CORPORATION.  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\nimport uuid\n\nfrom nvflare.apis.fl_component import FLComponent\nfrom nvflare.apis.fl_constant import EventScope, FLContextKey\nfrom nvflare.apis.fl_context import FLContext\n\n# do not use underscore as key name; otherwise it cannot be removed from ctx\n_KEY_EVENT_DEPTH = \"###event_depth\"\n_MAX_EVENT_DEPTH = 20\n\n\ndef fire_event(event: str, handlers: list, ctx: FLContext):\n    \"\"\"Fires the specified event and invokes the list of handlers.\n\n    Args:\n        event: the event to be fired\n        handlers: handlers to be invoked\n        ctx: context for cross-component data sharing\n\n    Returns: N/A\n\n    \"\"\"\n    event_id = str(uuid.uuid4())\n    event_data = ctx.get_prop(FLContextKey.EVENT_DATA, None)\n    event_origin = ctx.get_prop(FLContextKey.EVENT_ORIGIN, None)\n    event_scope = ctx.get_prop(FLContextKey.EVENT_SCOPE, EventScope.LOCAL)\n\n    depth = ctx.get_prop(_KEY_EVENT_DEPTH, 0)\n    if depth > _MAX_EVENT_DEPTH:\n        # too many recursive event calls\n        raise RuntimeError(\"Recursive event calls too deep (>{})\".format(_MAX_EVENT_DEPTH))\n\n    ctx.set_prop(key=_KEY_EVENT_DEPTH, value=depth + 1, private=True, sticky=False)\n\n    if handlers:\n        for h in handlers:\n            if not isinstance(h, FLComponent):\n                raise TypeError(\"handler must be FLComponent but got {}\".format(type(h)))\n            try:\n                # since events could be recursive (a handler fires another event) on the same fl_ctx,\n                # we need to reset these key values into the fl_ctx\n                ctx.set_prop(key=FLContextKey.EVENT_ID, value=event_id, private=True, sticky=False)\n                ctx.set_prop(key=FLContextKey.EVENT_DATA, value=event_data, private=True, sticky=False)\n                ctx.set_prop(key=FLContextKey.EVENT_ORIGIN, value=event_origin, private=True, sticky=False)\n                ctx.set_prop(key=FLContextKey.EVENT_SCOPE, value=event_scope, private=True, sticky=False)\n                h.handle_event(event, ctx)\n            except:\n                h.log_exception(ctx, 'exception when handling event \"{}\"'.format(event), fire_event=False)\n\n    ctx.set_prop(key=_KEY_EVENT_DEPTH, value=depth, private=True, sticky=False)\n", "path": "nvflare/private/event.py"}]}