We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
garbage collection vs usage tracking shutdown race condition
**Describe the bug**
I've seen this exception at least once in CI local tests. I think it comes from job completion garbage collection racing with DFK shutdown usage reporting, in examining the task dictionary.

```
parsl/tests/test_staging/test_elaborate_noop_file.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
parsl/dataflow/dflow.py:1051: in cleanup
    self.usage_tracker.send_message()
parsl/dataflow/usage_tracking/usage.py:224: in send_message
    message = self.construct_end_message()
parsl/dataflow/usage_tracking/usage.py:182: in construct_end_message
    app_fails = len([t for t in self.dfk.tasks if
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.0 = <dict_keyiterator object at 0x7f86107ca040>
>   app_fails = len([t for t in self.dfk.tasks if
                     self.dfk.tasks[t]['status'] in FINAL_FAILURE_STATES])
E   RuntimeError: dictionary changed size during iteration

```


**To Reproduce**
non-deterministic in CI

**Environment**
CI, near master 02d3b93ad4fd97b2411c000cf0b8820c76e5dfef


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `parsl/dataflow/usage_tracking/usage.py`
Content:
```
1 import uuid
2 import time
3 import hashlib
4 import os
5 import getpass
6 import json
7 import logging
8 import socket
9 import sys
10 import platform
11 import multiprocessing as mp
12 
13 from parsl.dataflow.states import FINAL_FAILURE_STATES
14 from parsl.version import VERSION as PARSL_VERSION
15 
16 logger = logging.getLogger(__name__)
17 
18 
19 def async_process(fn):
20     """ Decorator function to launch a function as a separate process """
21 
22     def run(*args, **kwargs):
23         proc = mp.Process(target=fn, args=args, kwargs=kwargs, name="Usage-Tracking")
24         proc.start()
25         return proc
26 
27     return run
28 
29 
30 @async_process
31 def udp_messenger(domain_name, UDP_IP, UDP_PORT, sock_timeout, message):
32     """Send UDP messages to usage tracker asynchronously
33 
34     This multiprocessing based messenger was written to overcome the limitations
35     of signalling/terminating a thread that is blocked on a system call. This
36     messenger is created as a separate process, and initialized with 2 queues,
37     to_send to receive messages to be sent to the internet.
38 
39     Args:
40           - domain_name (str) : Domain name string
41           - UDP_IP (str) : IP address YYY.YYY.YYY.YYY
42           - UDP_PORT (int) : UDP port to send out on
43           - sock_timeout (int) : Socket timeout
44           - to_send (multiprocessing.Queue) : Queue of outgoing messages to internet
45     """
46     try:
47         if message is None:
48             raise ValueError("message was none")
49 
50         encoded_message = bytes(message, "utf-8")
51 
52         if encoded_message is None:
53             raise ValueError("utf-8 encoding of message failed")
54 
55         if domain_name:
56             try:
57                 UDP_IP = socket.gethostbyname(domain_name)
58             except Exception:
59                 # (False, "Domain lookup failed, defaulting to {0}".format(UDP_IP))
60                 pass
61 
62         if UDP_IP is None:
63             raise Exception("UDP_IP is None")
64 
65         if UDP_PORT is None:
66             raise Exception("UDP_PORT is None")
67 
68         sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)  # UDP
69         sock.settimeout(sock_timeout)
70         sock.sendto(bytes(message, "utf-8"), (UDP_IP, UDP_PORT))
71         sock.close()
72 
73     except socket.timeout:
74         logger.debug("Failed to send usage tracking data: socket timeout")
75     except OSError as e:
76         logger.debug("Failed to send usage tracking data: OSError: {}".format(e))
77     except Exception as e:
78         logger.debug("Failed to send usage tracking data: Exception: {}".format(e))
79 
80 
81 class UsageTracker (object):
82     """Anonymized Usage Tracking for Parsl.
83 
84     Client for this is here : https://github.com/Parsl/parsl_tracking
85     This issue captures the discussion that went into functionality
86     implemented here : https://github.com/Parsl/parsl/issues/34
87 
88     """
89 
90     def __init__(self, dfk, ip='52.3.111.203', port=50077,
91                  domain_name='tracking.parsl-project.org'):
92         """Initialize usage tracking unless the user has opted-out.
93 
94         We will try to resolve the hostname specified in kwarg:domain_name
95         and if that fails attempt to use the kwarg:ip. Determining the
96         IP and sending message is threaded to avoid slowing down DFK
97         initialization.
98 
99         Tracks usage stats by inspecting the internal state of the dfk.
100 
101         Args:
102              - dfk (DFK object) : Data Flow Kernel object
103 
104         KWargs:
105              - ip (string) : IP address
106              - port (int) : Port number, Default:50077
107              - domain_name (string) : Domain name, will override IP
108                   Default: tracking.parsl-project.org
109         """
110 
111         self.domain_name = domain_name
112         self.ip = ip
113         # The sock timeout will only apply to UDP send and not domain resolution
114         self.sock_timeout = 5
115         self.UDP_PORT = port
116         self.UDP_IP = None
117         self.procs = []
118         self.dfk = dfk
119         self.config = self.dfk.config
120         self.uuid = str(uuid.uuid4())
121         self.parsl_version = PARSL_VERSION
122         self.python_version = "{}.{}.{}".format(sys.version_info.major,
123                                                 sys.version_info.minor,
124                                                 sys.version_info.micro)
125         self.tracking_enabled = self.check_tracking_enabled()
126         logger.debug("Tracking status: {}".format(self.tracking_enabled))
127         self.initialized = False  # Once first message is sent this will be True
128 
129     def check_tracking_enabled(self):
130         """By default tracking is enabled.
131 
132         Tracking is disabled if :
133             1. config["globals"]["usageTracking"] is set to False (Bool)
134             2. Environment variable PARSL_TRACKING is set to false (case insensitive)
135 
136         """
137         track = True   # By default we track usage
138 
139         if not self.config.usage_tracking:
140             track = False
141 
142         envvar = str(os.environ.get("PARSL_TRACKING", True)).lower()
143         if envvar == "false":
144             track = False
145 
146         return track
147 
148     def construct_start_message(self):
149         """Collect preliminary run info at the start of the DFK.
150 
151         Returns :
152               - Message dict dumped as json string, ready for UDP
153         """
154         uname = getpass.getuser().encode('latin1')
155         hashed_username = hashlib.sha256(uname).hexdigest()[0:10]
156         hname = socket.gethostname().encode('latin1')
157         hashed_hostname = hashlib.sha256(hname).hexdigest()[0:10]
158         message = {'uuid': self.uuid,
159                    'uname': hashed_username,
160                    'hname': hashed_hostname,
161                    'test': False,  # this field previously indicated if parsl
162                                    # was being run in test mode, and is
163                                    # retained for protocol compatibility
164                    'parsl_v': self.parsl_version,
165                    'python_v': self.python_version,
166                    'os': platform.system(),
167                    'os_v': platform.release(),
168                    'start': time.time()}
169 
170         return json.dumps(message)
171 
172     def construct_end_message(self):
173         """Collect the final run information at the time of DFK cleanup.
174 
175         Returns:
176              - Message dict dumped as json string, ready for UDP
177         """
178         app_count = self.dfk.task_count
179 
180         site_count = len([x for x in self.dfk.config.executors if x.managed])
181 
182         app_fails = len([t for t in self.dfk.tasks if
183                          self.dfk.tasks[t]['status'] in FINAL_FAILURE_STATES])
184 
185         message = {'uuid': self.uuid,
186                    'end': time.time(),
187                    't_apps': app_count,
188                    'sites': site_count,
189                    'c_time': None,
190                    'failed': app_fails,
191                    'test': False,  # see comment in construct_start_message
192                    }
193 
194         return json.dumps(message)
195 
196     def send_UDP_message(self, message):
197         """Send UDP message."""
198         x = 0
199         if self.tracking_enabled:
200             try:
201                 proc = udp_messenger(self.domain_name, self.UDP_IP, self.UDP_PORT, self.sock_timeout, message)
202                 self.procs.append(proc)
203             except Exception as e:
204                 logger.debug("Usage tracking failed: {}".format(e))
205         else:
206             x = -1
207 
208         return x
209 
210     def send_message(self):
211         """Send message over UDP.
212 
213         If tracking is disables, the bytes_sent will always be set to -1
214 
215         Returns:
216             (bytes_sent, time_taken)
217         """
218         start = time.time()
219         message = None
220         if not self.initialized:
221             message = self.construct_start_message()
222             self.initialized = True
223         else:
224             message = self.construct_end_message()
225 
226         self.send_UDP_message(message)
227         end = time.time()
228 
229         return end - start
230 
231     def __del__(self):
232         return self.close()
233 
234     def close(self):
235         """We terminate (SIGTERM) the processes added to the self.procs list """
236         for proc in self.procs:
237             proc.terminate()
238
```
--- 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/parsl/dataflow/usage_tracking/usage.py b/parsl/dataflow/usage_tracking/usage.py
--- a/parsl/dataflow/usage_tracking/usage.py
+++ b/parsl/dataflow/usage_tracking/usage.py
@@ -10,7 +10,6 @@
 import platform
 import multiprocessing as mp
 
-from parsl.dataflow.states import FINAL_FAILURE_STATES
 from parsl.version import VERSION as PARSL_VERSION
 
 logger = logging.getLogger(__name__)
@@ -179,8 +178,7 @@
 
         site_count = len([x for x in self.dfk.config.executors if x.managed])
 
-        app_fails = len([t for t in self.dfk.tasks if
-                         self.dfk.tasks[t]['status'] in FINAL_FAILURE_STATES])
+        app_fails = self.dfk.tasks_failed_count + self.dfk.tasks_dep_fail_count
 
         message = {'uuid': self.uuid,
                    'end': time.time(),

{"golden_diff": "diff --git a/parsl/dataflow/usage_tracking/usage.py b/parsl/dataflow/usage_tracking/usage.py\n--- a/parsl/dataflow/usage_tracking/usage.py\n+++ b/parsl/dataflow/usage_tracking/usage.py\n@@ -10,7 +10,6 @@\n import platform\n import multiprocessing as mp\n \n-from parsl.dataflow.states import FINAL_FAILURE_STATES\n from parsl.version import VERSION as PARSL_VERSION\n \n logger = logging.getLogger(__name__)\n@@ -179,8 +178,7 @@\n \n         site_count = len([x for x in self.dfk.config.executors if x.managed])\n \n-        app_fails = len([t for t in self.dfk.tasks if\n-                         self.dfk.tasks[t]['status'] in FINAL_FAILURE_STATES])\n+        app_fails = self.dfk.tasks_failed_count + self.dfk.tasks_dep_fail_count\n \n         message = {'uuid': self.uuid,\n                    'end': time.time(),\n", "issue": "garbage collection vs usage tracking shutdown race condition\n**Describe the bug**\r\nI've seen this exception at least once in CI local tests. I think it comes from job completion garbage collection racing with DFK shutdown usage reporting, in examining the task dictionary.\r\n\r\n```\r\nparsl/tests/test_staging/test_elaborate_noop_file.py:60: \r\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \r\nparsl/dataflow/dflow.py:1051: in cleanup\r\n    self.usage_tracker.send_message()\r\nparsl/dataflow/usage_tracking/usage.py:224: in send_message\r\n    message = self.construct_end_message()\r\nparsl/dataflow/usage_tracking/usage.py:182: in construct_end_message\r\n    app_fails = len([t for t in self.dfk.tasks if\r\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \r\n.0 = <dict_keyiterator object at 0x7f86107ca040>\r\n>   app_fails = len([t for t in self.dfk.tasks if\r\n                     self.dfk.tasks[t]['status'] in FINAL_FAILURE_STATES])\r\nE   RuntimeError: dictionary changed size during iteration\r\n\r\n```\r\n\r\n\r\n**To Reproduce**\r\nnon-deterministic in CI\r\n\r\n**Environment**\r\nCI, near master 02d3b93ad4fd97b2411c000cf0b8820c76e5dfef\r\n\n", "before_files": [{"content": "import uuid\nimport time\nimport hashlib\nimport os\nimport getpass\nimport json\nimport logging\nimport socket\nimport sys\nimport platform\nimport multiprocessing as mp\n\nfrom parsl.dataflow.states import FINAL_FAILURE_STATES\nfrom parsl.version import VERSION as PARSL_VERSION\n\nlogger = logging.getLogger(__name__)\n\n\ndef async_process(fn):\n    \"\"\" Decorator function to launch a function as a separate process \"\"\"\n\n    def run(*args, **kwargs):\n        proc = mp.Process(target=fn, args=args, kwargs=kwargs, name=\"Usage-Tracking\")\n        proc.start()\n        return proc\n\n    return run\n\n\n@async_process\ndef udp_messenger(domain_name, UDP_IP, UDP_PORT, sock_timeout, message):\n    \"\"\"Send UDP messages to usage tracker asynchronously\n\n    This multiprocessing based messenger was written to overcome the limitations\n    of signalling/terminating a thread that is blocked on a system call. This\n    messenger is created as a separate process, and initialized with 2 queues,\n    to_send to receive messages to be sent to the internet.\n\n    Args:\n          - domain_name (str) : Domain name string\n          - UDP_IP (str) : IP address YYY.YYY.YYY.YYY\n          - UDP_PORT (int) : UDP port to send out on\n          - sock_timeout (int) : Socket timeout\n          - to_send (multiprocessing.Queue) : Queue of outgoing messages to internet\n    \"\"\"\n    try:\n        if message is None:\n            raise ValueError(\"message was none\")\n\n        encoded_message = bytes(message, \"utf-8\")\n\n        if encoded_message is None:\n            raise ValueError(\"utf-8 encoding of message failed\")\n\n        if domain_name:\n            try:\n                UDP_IP = socket.gethostbyname(domain_name)\n            except Exception:\n                # (False, \"Domain lookup failed, defaulting to {0}\".format(UDP_IP))\n                pass\n\n        if UDP_IP is None:\n            raise Exception(\"UDP_IP is None\")\n\n        if UDP_PORT is None:\n            raise Exception(\"UDP_PORT is None\")\n\n        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)  # UDP\n        sock.settimeout(sock_timeout)\n        sock.sendto(bytes(message, \"utf-8\"), (UDP_IP, UDP_PORT))\n        sock.close()\n\n    except socket.timeout:\n        logger.debug(\"Failed to send usage tracking data: socket timeout\")\n    except OSError as e:\n        logger.debug(\"Failed to send usage tracking data: OSError: {}\".format(e))\n    except Exception as e:\n        logger.debug(\"Failed to send usage tracking data: Exception: {}\".format(e))\n\n\nclass UsageTracker (object):\n    \"\"\"Anonymized Usage Tracking for Parsl.\n\n    Client for this is here : https://github.com/Parsl/parsl_tracking\n    This issue captures the discussion that went into functionality\n    implemented here : https://github.com/Parsl/parsl/issues/34\n\n    \"\"\"\n\n    def __init__(self, dfk, ip='52.3.111.203', port=50077,\n                 domain_name='tracking.parsl-project.org'):\n        \"\"\"Initialize usage tracking unless the user has opted-out.\n\n        We will try to resolve the hostname specified in kwarg:domain_name\n        and if that fails attempt to use the kwarg:ip. Determining the\n        IP and sending message is threaded to avoid slowing down DFK\n        initialization.\n\n        Tracks usage stats by inspecting the internal state of the dfk.\n\n        Args:\n             - dfk (DFK object) : Data Flow Kernel object\n\n        KWargs:\n             - ip (string) : IP address\n             - port (int) : Port number, Default:50077\n             - domain_name (string) : Domain name, will override IP\n                  Default: tracking.parsl-project.org\n        \"\"\"\n\n        self.domain_name = domain_name\n        self.ip = ip\n        # The sock timeout will only apply to UDP send and not domain resolution\n        self.sock_timeout = 5\n        self.UDP_PORT = port\n        self.UDP_IP = None\n        self.procs = []\n        self.dfk = dfk\n        self.config = self.dfk.config\n        self.uuid = str(uuid.uuid4())\n        self.parsl_version = PARSL_VERSION\n        self.python_version = \"{}.{}.{}\".format(sys.version_info.major,\n                                                sys.version_info.minor,\n                                                sys.version_info.micro)\n        self.tracking_enabled = self.check_tracking_enabled()\n        logger.debug(\"Tracking status: {}\".format(self.tracking_enabled))\n        self.initialized = False  # Once first message is sent this will be True\n\n    def check_tracking_enabled(self):\n        \"\"\"By default tracking is enabled.\n\n        Tracking is disabled if :\n            1. config[\"globals\"][\"usageTracking\"] is set to False (Bool)\n            2. Environment variable PARSL_TRACKING is set to false (case insensitive)\n\n        \"\"\"\n        track = True   # By default we track usage\n\n        if not self.config.usage_tracking:\n            track = False\n\n        envvar = str(os.environ.get(\"PARSL_TRACKING\", True)).lower()\n        if envvar == \"false\":\n            track = False\n\n        return track\n\n    def construct_start_message(self):\n        \"\"\"Collect preliminary run info at the start of the DFK.\n\n        Returns :\n              - Message dict dumped as json string, ready for UDP\n        \"\"\"\n        uname = getpass.getuser().encode('latin1')\n        hashed_username = hashlib.sha256(uname).hexdigest()[0:10]\n        hname = socket.gethostname().encode('latin1')\n        hashed_hostname = hashlib.sha256(hname).hexdigest()[0:10]\n        message = {'uuid': self.uuid,\n                   'uname': hashed_username,\n                   'hname': hashed_hostname,\n                   'test': False,  # this field previously indicated if parsl\n                                   # was being run in test mode, and is\n                                   # retained for protocol compatibility\n                   'parsl_v': self.parsl_version,\n                   'python_v': self.python_version,\n                   'os': platform.system(),\n                   'os_v': platform.release(),\n                   'start': time.time()}\n\n        return json.dumps(message)\n\n    def construct_end_message(self):\n        \"\"\"Collect the final run information at the time of DFK cleanup.\n\n        Returns:\n             - Message dict dumped as json string, ready for UDP\n        \"\"\"\n        app_count = self.dfk.task_count\n\n        site_count = len([x for x in self.dfk.config.executors if x.managed])\n\n        app_fails = len([t for t in self.dfk.tasks if\n                         self.dfk.tasks[t]['status'] in FINAL_FAILURE_STATES])\n\n        message = {'uuid': self.uuid,\n                   'end': time.time(),\n                   't_apps': app_count,\n                   'sites': site_count,\n                   'c_time': None,\n                   'failed': app_fails,\n                   'test': False,  # see comment in construct_start_message\n                   }\n\n        return json.dumps(message)\n\n    def send_UDP_message(self, message):\n        \"\"\"Send UDP message.\"\"\"\n        x = 0\n        if self.tracking_enabled:\n            try:\n                proc = udp_messenger(self.domain_name, self.UDP_IP, self.UDP_PORT, self.sock_timeout, message)\n                self.procs.append(proc)\n            except Exception as e:\n                logger.debug(\"Usage tracking failed: {}\".format(e))\n        else:\n            x = -1\n\n        return x\n\n    def send_message(self):\n        \"\"\"Send message over UDP.\n\n        If tracking is disables, the bytes_sent will always be set to -1\n\n        Returns:\n            (bytes_sent, time_taken)\n        \"\"\"\n        start = time.time()\n        message = None\n        if not self.initialized:\n            message = self.construct_start_message()\n            self.initialized = True\n        else:\n            message = self.construct_end_message()\n\n        self.send_UDP_message(message)\n        end = time.time()\n\n        return end - start\n\n    def __del__(self):\n        return self.close()\n\n    def close(self):\n        \"\"\"We terminate (SIGTERM) the processes added to the self.procs list \"\"\"\n        for proc in self.procs:\n            proc.terminate()\n", "path": "parsl/dataflow/usage_tracking/usage.py"}], "after_files": [{"content": "import uuid\nimport time\nimport hashlib\nimport os\nimport getpass\nimport json\nimport logging\nimport socket\nimport sys\nimport platform\nimport multiprocessing as mp\n\nfrom parsl.version import VERSION as PARSL_VERSION\n\nlogger = logging.getLogger(__name__)\n\n\ndef async_process(fn):\n    \"\"\" Decorator function to launch a function as a separate process \"\"\"\n\n    def run(*args, **kwargs):\n        proc = mp.Process(target=fn, args=args, kwargs=kwargs, name=\"Usage-Tracking\")\n        proc.start()\n        return proc\n\n    return run\n\n\n@async_process\ndef udp_messenger(domain_name, UDP_IP, UDP_PORT, sock_timeout, message):\n    \"\"\"Send UDP messages to usage tracker asynchronously\n\n    This multiprocessing based messenger was written to overcome the limitations\n    of signalling/terminating a thread that is blocked on a system call. This\n    messenger is created as a separate process, and initialized with 2 queues,\n    to_send to receive messages to be sent to the internet.\n\n    Args:\n          - domain_name (str) : Domain name string\n          - UDP_IP (str) : IP address YYY.YYY.YYY.YYY\n          - UDP_PORT (int) : UDP port to send out on\n          - sock_timeout (int) : Socket timeout\n          - to_send (multiprocessing.Queue) : Queue of outgoing messages to internet\n    \"\"\"\n    try:\n        if message is None:\n            raise ValueError(\"message was none\")\n\n        encoded_message = bytes(message, \"utf-8\")\n\n        if encoded_message is None:\n            raise ValueError(\"utf-8 encoding of message failed\")\n\n        if domain_name:\n            try:\n                UDP_IP = socket.gethostbyname(domain_name)\n            except Exception:\n                # (False, \"Domain lookup failed, defaulting to {0}\".format(UDP_IP))\n                pass\n\n        if UDP_IP is None:\n            raise Exception(\"UDP_IP is None\")\n\n        if UDP_PORT is None:\n            raise Exception(\"UDP_PORT is None\")\n\n        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)  # UDP\n        sock.settimeout(sock_timeout)\n        sock.sendto(bytes(message, \"utf-8\"), (UDP_IP, UDP_PORT))\n        sock.close()\n\n    except socket.timeout:\n        logger.debug(\"Failed to send usage tracking data: socket timeout\")\n    except OSError as e:\n        logger.debug(\"Failed to send usage tracking data: OSError: {}\".format(e))\n    except Exception as e:\n        logger.debug(\"Failed to send usage tracking data: Exception: {}\".format(e))\n\n\nclass UsageTracker (object):\n    \"\"\"Anonymized Usage Tracking for Parsl.\n\n    Client for this is here : https://github.com/Parsl/parsl_tracking\n    This issue captures the discussion that went into functionality\n    implemented here : https://github.com/Parsl/parsl/issues/34\n\n    \"\"\"\n\n    def __init__(self, dfk, ip='52.3.111.203', port=50077,\n                 domain_name='tracking.parsl-project.org'):\n        \"\"\"Initialize usage tracking unless the user has opted-out.\n\n        We will try to resolve the hostname specified in kwarg:domain_name\n        and if that fails attempt to use the kwarg:ip. Determining the\n        IP and sending message is threaded to avoid slowing down DFK\n        initialization.\n\n        Tracks usage stats by inspecting the internal state of the dfk.\n\n        Args:\n             - dfk (DFK object) : Data Flow Kernel object\n\n        KWargs:\n             - ip (string) : IP address\n             - port (int) : Port number, Default:50077\n             - domain_name (string) : Domain name, will override IP\n                  Default: tracking.parsl-project.org\n        \"\"\"\n\n        self.domain_name = domain_name\n        self.ip = ip\n        # The sock timeout will only apply to UDP send and not domain resolution\n        self.sock_timeout = 5\n        self.UDP_PORT = port\n        self.UDP_IP = None\n        self.procs = []\n        self.dfk = dfk\n        self.config = self.dfk.config\n        self.uuid = str(uuid.uuid4())\n        self.parsl_version = PARSL_VERSION\n        self.python_version = \"{}.{}.{}\".format(sys.version_info.major,\n                                                sys.version_info.minor,\n                                                sys.version_info.micro)\n        self.tracking_enabled = self.check_tracking_enabled()\n        logger.debug(\"Tracking status: {}\".format(self.tracking_enabled))\n        self.initialized = False  # Once first message is sent this will be True\n\n    def check_tracking_enabled(self):\n        \"\"\"By default tracking is enabled.\n\n        Tracking is disabled if :\n            1. config[\"globals\"][\"usageTracking\"] is set to False (Bool)\n            2. Environment variable PARSL_TRACKING is set to false (case insensitive)\n\n        \"\"\"\n        track = True   # By default we track usage\n\n        if not self.config.usage_tracking:\n            track = False\n\n        envvar = str(os.environ.get(\"PARSL_TRACKING\", True)).lower()\n        if envvar == \"false\":\n            track = False\n\n        return track\n\n    def construct_start_message(self):\n        \"\"\"Collect preliminary run info at the start of the DFK.\n\n        Returns :\n              - Message dict dumped as json string, ready for UDP\n        \"\"\"\n        uname = getpass.getuser().encode('latin1')\n        hashed_username = hashlib.sha256(uname).hexdigest()[0:10]\n        hname = socket.gethostname().encode('latin1')\n        hashed_hostname = hashlib.sha256(hname).hexdigest()[0:10]\n        message = {'uuid': self.uuid,\n                   'uname': hashed_username,\n                   'hname': hashed_hostname,\n                   'test': False,  # this field previously indicated if parsl\n                                   # was being run in test mode, and is\n                                   # retained for protocol compatibility\n                   'parsl_v': self.parsl_version,\n                   'python_v': self.python_version,\n                   'os': platform.system(),\n                   'os_v': platform.release(),\n                   'start': time.time()}\n\n        return json.dumps(message)\n\n    def construct_end_message(self):\n        \"\"\"Collect the final run information at the time of DFK cleanup.\n\n        Returns:\n             - Message dict dumped as json string, ready for UDP\n        \"\"\"\n        app_count = self.dfk.task_count\n\n        site_count = len([x for x in self.dfk.config.executors if x.managed])\n\n        app_fails = self.dfk.tasks_failed_count + self.dfk.tasks_dep_fail_count\n\n        message = {'uuid': self.uuid,\n                   'end': time.time(),\n                   't_apps': app_count,\n                   'sites': site_count,\n                   'c_time': None,\n                   'failed': app_fails,\n                   'test': False,  # see comment in construct_start_message\n                   }\n\n        return json.dumps(message)\n\n    def send_UDP_message(self, message):\n        \"\"\"Send UDP message.\"\"\"\n        x = 0\n        if self.tracking_enabled:\n            try:\n                proc = udp_messenger(self.domain_name, self.UDP_IP, self.UDP_PORT, self.sock_timeout, message)\n                self.procs.append(proc)\n            except Exception as e:\n                logger.debug(\"Usage tracking failed: {}\".format(e))\n        else:\n            x = -1\n\n        return x\n\n    def send_message(self):\n        \"\"\"Send message over UDP.\n\n        If tracking is disables, the bytes_sent will always be set to -1\n\n        Returns:\n            (bytes_sent, time_taken)\n        \"\"\"\n        start = time.time()\n        message = None\n        if not self.initialized:\n            message = self.construct_start_message()\n            self.initialized = True\n        else:\n            message = self.construct_end_message()\n\n        self.send_UDP_message(message)\n        end = time.time()\n\n        return end - start\n\n    def __del__(self):\n        return self.close()\n\n    def close(self):\n        \"\"\"We terminate (SIGTERM) the processes added to the self.procs list \"\"\"\n        for proc in self.procs:\n            proc.terminate()\n", "path": "parsl/dataflow/usage_tracking/usage.py"}]}