We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Consider removing emphasis on the `DataFlowKernel` in the docs
We might want to relegate discussion of the DFK to the more advanced section of the docs. For many applications, simply running `parsl.load` is sufficient. I strongly feel that as part of the class-based configuration shuffling we're doing, it will be simpler and less bug prone (on our side and user-side) to have the `Config` object be the only mode of specifying options (in other words, dispense with the ability to override options in the DFK constructor-- that removes places people can get confused about what precedence options have. There isn't really a cost; if they are importing a config and then overriding various options, it should be the same if they do it in the `Config` before loading or passing options to the constructor. Overriding options in the `Config` makes it explicit on their side what is going in.) If we _do_ go that route, then exposing the DFK to users seems even less necessary. I'm putting this out there now so I don't forget, but we should make the call after the class-based configs are worked out.

--- 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/futures.py`
Content:
```
1 """This module implements the AppFutures.
2 
3     We have two basic types of futures:
4     1. DataFutures which represent data objects
5     2. AppFutures which represent the futures on App/Leaf tasks.
6 
7 """
8 
9 from concurrent.futures import Future
10 import logging
11 import threading
12 
13 from parsl.app.errors import RemoteException
14 
15 logger = logging.getLogger(__name__)
16 
17 # Possible future states (for internal use by the futures package).
18 PENDING = 'PENDING'
19 RUNNING = 'RUNNING'
20 # The future was cancelled by the user...
21 CANCELLED = 'CANCELLED'
22 # ...and _Waiter.add_cancelled() was called by a worker.
23 CANCELLED_AND_NOTIFIED = 'CANCELLED_AND_NOTIFIED'
24 FINISHED = 'FINISHED'
25 
26 _STATE_TO_DESCRIPTION_MAP = {
27     PENDING: "pending",
28     RUNNING: "running",
29     CANCELLED: "cancelled",
30     CANCELLED_AND_NOTIFIED: "cancelled",
31     FINISHED: "finished"
32 }
33 
34 
35 class AppFuture(Future):
36     """An AppFuture points at a Future returned from an Executor.
37 
38     We are simply wrapping a AppFuture, and adding the specific case where, if the future
39     is resolved i.e file exists, then the DataFuture is assumed to be resolved.
40 
41     """
42 
43     def parent_callback(self, executor_fu):
44         """Callback from executor future to update the parent.
45 
46         Args:
47             - executor_fu (Future): Future returned by the executor along with callback
48 
49         Returns:
50             - None
51 
52         Updates the super() with the result() or exception()
53         """
54         # print("[RETRY:TODO] parent_Callback for {0}".format(executor_fu))
55         if executor_fu.done() is True:
56             try:
57                 super().set_result(executor_fu.result())
58             except Exception as e:
59                 super().set_exception(e)
60 
61     def __init__(self, parent, tid=None, stdout=None, stderr=None):
62         """Initialize the AppFuture.
63 
64         Args:
65              - parent (Future) : The parent future if one exists
66                A default value of None should be passed in if app is not launched
67 
68         KWargs:
69              - tid (Int) : Task id should be any unique identifier. Now Int.
70              - stdout (str) : Stdout file of the app.
71                    Default: None
72              - stderr (str) : Stderr file of the app.
73                    Default: None
74         """
75         self._tid = tid
76         super().__init__()
77         self.prev_parent = None
78         self.parent = parent
79         self._parent_update_lock = threading.Lock()
80         self._parent_update_event = threading.Event()
81         self._outputs = []
82         self._stdout = stdout
83         self._stderr = stderr
84 
85     @property
86     def stdout(self):
87         return self._stdout
88 
89     @property
90     def stderr(self):
91         return self._stderr
92 
93     @property
94     def tid(self):
95         return self._tid
96 
97     def update_parent(self, fut):
98         """Add a callback to the parent to update the state.
99 
100         This handles the case where the user has called result on the AppFuture
101         before the parent exists.
102         """
103         # with self._parent_update_lock:
104         self.parent = fut
105         fut.add_done_callback(self.parent_callback)
106         self._parent_update_event.set()
107 
108     def result(self, timeout=None):
109         """Result.
110 
111         Waits for the result of the AppFuture
112         KWargs:
113               timeout (int): Timeout in seconds
114         """
115         try:
116             if self.parent:
117                 res = self.parent.result(timeout=timeout)
118             else:
119                 res = super().result(timeout=timeout)
120             if isinstance(res, RemoteException):
121                 res.reraise()
122             return res
123 
124         except Exception as e:
125             if self.parent.retries_left > 0:
126                 self._parent_update_event.wait()
127                 self._parent_update_event.clear()
128                 return self.result(timeout=timeout)
129             else:
130                 if isinstance(e, RemoteException):
131                     e.reraise()
132                 else:
133                     raise
134 
135     def cancel(self):
136         if self.parent:
137             return self.parent.cancel
138         else:
139             return False
140 
141     def cancelled(self):
142         if self.parent:
143             return self.parent.cancelled()
144         else:
145             return False
146 
147     def running(self):
148         if self.parent:
149             return self.parent.running()
150         else:
151             return False
152 
153     def done(self):
154         """Check if the future is done.
155 
156         If a parent is set, we return the status of the parent.
157         else, there is no parent assigned, meaning the status is False.
158 
159         Returns:
160               - True : If the future has successfully resolved.
161               - False : Pending resolution
162         """
163         if self.parent:
164             return self.parent.done()
165         else:
166             return False
167 
168     def exception(self, timeout=None):
169         if self.parent:
170             return self.parent.exception(timeout=timeout)
171         else:
172             return False
173 
174     def add_done_callback(self, fn):
175         if self.parent:
176             return self.parent.add_done_callback(fn)
177         else:
178             return None
179 
180     @property
181     def outputs(self):
182         return self._outputs
183 
184     def __repr__(self):
185         if self.parent:
186             with self.parent._condition:
187                 if self.parent._state == FINISHED:
188                     if self.parent._exception:
189                         return '<%s at %#x state=%s raised %s>' % (
190                             self.__class__.__name__,
191                             id(self),
192                             _STATE_TO_DESCRIPTION_MAP[self.parent._state],
193                             self.parent._exception.__class__.__name__)
194                     else:
195                         return '<%s at %#x state=%s returned %s>' % (
196                             self.__class__.__name__,
197                             id(self),
198                             _STATE_TO_DESCRIPTION_MAP[self.parent._state],
199                             self.parent._result.__class__.__name__)
200                 return '<%s at %#x state=%s>' % (
201                     self.__class__.__name__,
202                     id(self),
203                     _STATE_TO_DESCRIPTION_MAP[self.parent._state])
204         else:
205             return '<%s at %#x state=%s>' % (
206                 self.__class__.__name__,
207                 id(self),
208                 _STATE_TO_DESCRIPTION_MAP[self._state])
209
```
--- 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/futures.py b/parsl/dataflow/futures.py
--- a/parsl/dataflow/futures.py
+++ b/parsl/dataflow/futures.py
@@ -1,6 +1,6 @@
 """This module implements the AppFutures.
 
-    We have two basic types of futures:
+We have two basic types of futures:
     1. DataFutures which represent data objects
     2. AppFutures which represent the futures on App/Leaf tasks.
 
@@ -110,7 +110,7 @@
 
         Waits for the result of the AppFuture
         KWargs:
-              timeout (int): Timeout in seconds
+        timeout (int): Timeout in seconds
         """
         try:
             if self.parent:

{"golden_diff": "diff --git a/parsl/dataflow/futures.py b/parsl/dataflow/futures.py\n--- a/parsl/dataflow/futures.py\n+++ b/parsl/dataflow/futures.py\n@@ -1,6 +1,6 @@\n \"\"\"This module implements the AppFutures.\n \n-    We have two basic types of futures:\n+We have two basic types of futures:\n     1. DataFutures which represent data objects\n     2. AppFutures which represent the futures on App/Leaf tasks.\n \n@@ -110,7 +110,7 @@\n \n         Waits for the result of the AppFuture\n         KWargs:\n-              timeout (int): Timeout in seconds\n+        timeout (int): Timeout in seconds\n         \"\"\"\n         try:\n             if self.parent:\n", "issue": "Consider removing emphasis on the `DataFlowKernel` in the docs\nWe might want to relegate discussion of the DFK to the more advanced section of the docs. For many applications, simply running `parsl.load` is sufficient. I strongly feel that as part of the class-based configuration shuffling we're doing, it will be simpler and less bug prone (on our side and user-side) to have the `Config` object be the only mode of specifying options (in other words, dispense with the ability to override options in the DFK constructor-- that removes places people can get confused about what precedence options have. There isn't really a cost; if they are importing a config and then overriding various options, it should be the same if they do it in the `Config` before loading or passing options to the constructor. Overriding options in the `Config` makes it explicit on their side what is going in.) If we _do_ go that route, then exposing the DFK to users seems even less necessary. I'm putting this out there now so I don't forget, but we should make the call after the class-based configs are worked out.\n", "before_files": [{"content": "\"\"\"This module implements the AppFutures.\n\n    We have two basic types of futures:\n    1. DataFutures which represent data objects\n    2. AppFutures which represent the futures on App/Leaf tasks.\n\n\"\"\"\n\nfrom concurrent.futures import Future\nimport logging\nimport threading\n\nfrom parsl.app.errors import RemoteException\n\nlogger = logging.getLogger(__name__)\n\n# Possible future states (for internal use by the futures package).\nPENDING = 'PENDING'\nRUNNING = 'RUNNING'\n# The future was cancelled by the user...\nCANCELLED = 'CANCELLED'\n# ...and _Waiter.add_cancelled() was called by a worker.\nCANCELLED_AND_NOTIFIED = 'CANCELLED_AND_NOTIFIED'\nFINISHED = 'FINISHED'\n\n_STATE_TO_DESCRIPTION_MAP = {\n    PENDING: \"pending\",\n    RUNNING: \"running\",\n    CANCELLED: \"cancelled\",\n    CANCELLED_AND_NOTIFIED: \"cancelled\",\n    FINISHED: \"finished\"\n}\n\n\nclass AppFuture(Future):\n    \"\"\"An AppFuture points at a Future returned from an Executor.\n\n    We are simply wrapping a AppFuture, and adding the specific case where, if the future\n    is resolved i.e file exists, then the DataFuture is assumed to be resolved.\n\n    \"\"\"\n\n    def parent_callback(self, executor_fu):\n        \"\"\"Callback from executor future to update the parent.\n\n        Args:\n            - executor_fu (Future): Future returned by the executor along with callback\n\n        Returns:\n            - None\n\n        Updates the super() with the result() or exception()\n        \"\"\"\n        # print(\"[RETRY:TODO] parent_Callback for {0}\".format(executor_fu))\n        if executor_fu.done() is True:\n            try:\n                super().set_result(executor_fu.result())\n            except Exception as e:\n                super().set_exception(e)\n\n    def __init__(self, parent, tid=None, stdout=None, stderr=None):\n        \"\"\"Initialize the AppFuture.\n\n        Args:\n             - parent (Future) : The parent future if one exists\n               A default value of None should be passed in if app is not launched\n\n        KWargs:\n             - tid (Int) : Task id should be any unique identifier. Now Int.\n             - stdout (str) : Stdout file of the app.\n                   Default: None\n             - stderr (str) : Stderr file of the app.\n                   Default: None\n        \"\"\"\n        self._tid = tid\n        super().__init__()\n        self.prev_parent = None\n        self.parent = parent\n        self._parent_update_lock = threading.Lock()\n        self._parent_update_event = threading.Event()\n        self._outputs = []\n        self._stdout = stdout\n        self._stderr = stderr\n\n    @property\n    def stdout(self):\n        return self._stdout\n\n    @property\n    def stderr(self):\n        return self._stderr\n\n    @property\n    def tid(self):\n        return self._tid\n\n    def update_parent(self, fut):\n        \"\"\"Add a callback to the parent to update the state.\n\n        This handles the case where the user has called result on the AppFuture\n        before the parent exists.\n        \"\"\"\n        # with self._parent_update_lock:\n        self.parent = fut\n        fut.add_done_callback(self.parent_callback)\n        self._parent_update_event.set()\n\n    def result(self, timeout=None):\n        \"\"\"Result.\n\n        Waits for the result of the AppFuture\n        KWargs:\n              timeout (int): Timeout in seconds\n        \"\"\"\n        try:\n            if self.parent:\n                res = self.parent.result(timeout=timeout)\n            else:\n                res = super().result(timeout=timeout)\n            if isinstance(res, RemoteException):\n                res.reraise()\n            return res\n\n        except Exception as e:\n            if self.parent.retries_left > 0:\n                self._parent_update_event.wait()\n                self._parent_update_event.clear()\n                return self.result(timeout=timeout)\n            else:\n                if isinstance(e, RemoteException):\n                    e.reraise()\n                else:\n                    raise\n\n    def cancel(self):\n        if self.parent:\n            return self.parent.cancel\n        else:\n            return False\n\n    def cancelled(self):\n        if self.parent:\n            return self.parent.cancelled()\n        else:\n            return False\n\n    def running(self):\n        if self.parent:\n            return self.parent.running()\n        else:\n            return False\n\n    def done(self):\n        \"\"\"Check if the future is done.\n\n        If a parent is set, we return the status of the parent.\n        else, there is no parent assigned, meaning the status is False.\n\n        Returns:\n              - True : If the future has successfully resolved.\n              - False : Pending resolution\n        \"\"\"\n        if self.parent:\n            return self.parent.done()\n        else:\n            return False\n\n    def exception(self, timeout=None):\n        if self.parent:\n            return self.parent.exception(timeout=timeout)\n        else:\n            return False\n\n    def add_done_callback(self, fn):\n        if self.parent:\n            return self.parent.add_done_callback(fn)\n        else:\n            return None\n\n    @property\n    def outputs(self):\n        return self._outputs\n\n    def __repr__(self):\n        if self.parent:\n            with self.parent._condition:\n                if self.parent._state == FINISHED:\n                    if self.parent._exception:\n                        return '<%s at %#x state=%s raised %s>' % (\n                            self.__class__.__name__,\n                            id(self),\n                            _STATE_TO_DESCRIPTION_MAP[self.parent._state],\n                            self.parent._exception.__class__.__name__)\n                    else:\n                        return '<%s at %#x state=%s returned %s>' % (\n                            self.__class__.__name__,\n                            id(self),\n                            _STATE_TO_DESCRIPTION_MAP[self.parent._state],\n                            self.parent._result.__class__.__name__)\n                return '<%s at %#x state=%s>' % (\n                    self.__class__.__name__,\n                    id(self),\n                    _STATE_TO_DESCRIPTION_MAP[self.parent._state])\n        else:\n            return '<%s at %#x state=%s>' % (\n                self.__class__.__name__,\n                id(self),\n                _STATE_TO_DESCRIPTION_MAP[self._state])\n", "path": "parsl/dataflow/futures.py"}], "after_files": [{"content": "\"\"\"This module implements the AppFutures.\n\nWe have two basic types of futures:\n    1. DataFutures which represent data objects\n    2. AppFutures which represent the futures on App/Leaf tasks.\n\n\"\"\"\n\nfrom concurrent.futures import Future\nimport logging\nimport threading\n\nfrom parsl.app.errors import RemoteException\n\nlogger = logging.getLogger(__name__)\n\n# Possible future states (for internal use by the futures package).\nPENDING = 'PENDING'\nRUNNING = 'RUNNING'\n# The future was cancelled by the user...\nCANCELLED = 'CANCELLED'\n# ...and _Waiter.add_cancelled() was called by a worker.\nCANCELLED_AND_NOTIFIED = 'CANCELLED_AND_NOTIFIED'\nFINISHED = 'FINISHED'\n\n_STATE_TO_DESCRIPTION_MAP = {\n    PENDING: \"pending\",\n    RUNNING: \"running\",\n    CANCELLED: \"cancelled\",\n    CANCELLED_AND_NOTIFIED: \"cancelled\",\n    FINISHED: \"finished\"\n}\n\n\nclass AppFuture(Future):\n    \"\"\"An AppFuture points at a Future returned from an Executor.\n\n    We are simply wrapping a AppFuture, and adding the specific case where, if the future\n    is resolved i.e file exists, then the DataFuture is assumed to be resolved.\n\n    \"\"\"\n\n    def parent_callback(self, executor_fu):\n        \"\"\"Callback from executor future to update the parent.\n\n        Args:\n            - executor_fu (Future): Future returned by the executor along with callback\n\n        Returns:\n            - None\n\n        Updates the super() with the result() or exception()\n        \"\"\"\n        # print(\"[RETRY:TODO] parent_Callback for {0}\".format(executor_fu))\n        if executor_fu.done() is True:\n            try:\n                super().set_result(executor_fu.result())\n            except Exception as e:\n                super().set_exception(e)\n\n    def __init__(self, parent, tid=None, stdout=None, stderr=None):\n        \"\"\"Initialize the AppFuture.\n\n        Args:\n             - parent (Future) : The parent future if one exists\n               A default value of None should be passed in if app is not launched\n\n        KWargs:\n             - tid (Int) : Task id should be any unique identifier. Now Int.\n             - stdout (str) : Stdout file of the app.\n                   Default: None\n             - stderr (str) : Stderr file of the app.\n                   Default: None\n        \"\"\"\n        self._tid = tid\n        super().__init__()\n        self.prev_parent = None\n        self.parent = parent\n        self._parent_update_lock = threading.Lock()\n        self._parent_update_event = threading.Event()\n        self._outputs = []\n        self._stdout = stdout\n        self._stderr = stderr\n\n    @property\n    def stdout(self):\n        return self._stdout\n\n    @property\n    def stderr(self):\n        return self._stderr\n\n    @property\n    def tid(self):\n        return self._tid\n\n    def update_parent(self, fut):\n        \"\"\"Add a callback to the parent to update the state.\n\n        This handles the case where the user has called result on the AppFuture\n        before the parent exists.\n        \"\"\"\n        # with self._parent_update_lock:\n        self.parent = fut\n        fut.add_done_callback(self.parent_callback)\n        self._parent_update_event.set()\n\n    def result(self, timeout=None):\n        \"\"\"Result.\n\n        Waits for the result of the AppFuture\n        KWargs:\n        timeout (int): Timeout in seconds\n        \"\"\"\n        try:\n            if self.parent:\n                res = self.parent.result(timeout=timeout)\n            else:\n                res = super().result(timeout=timeout)\n            if isinstance(res, RemoteException):\n                res.reraise()\n            return res\n\n        except Exception as e:\n            if self.parent.retries_left > 0:\n                self._parent_update_event.wait()\n                self._parent_update_event.clear()\n                return self.result(timeout=timeout)\n            else:\n                if isinstance(e, RemoteException):\n                    e.reraise()\n                else:\n                    raise\n\n    def cancel(self):\n        if self.parent:\n            return self.parent.cancel\n        else:\n            return False\n\n    def cancelled(self):\n        if self.parent:\n            return self.parent.cancelled()\n        else:\n            return False\n\n    def running(self):\n        if self.parent:\n            return self.parent.running()\n        else:\n            return False\n\n    def done(self):\n        \"\"\"Check if the future is done.\n\n        If a parent is set, we return the status of the parent.\n        else, there is no parent assigned, meaning the status is False.\n\n        Returns:\n              - True : If the future has successfully resolved.\n              - False : Pending resolution\n        \"\"\"\n        if self.parent:\n            return self.parent.done()\n        else:\n            return False\n\n    def exception(self, timeout=None):\n        if self.parent:\n            return self.parent.exception(timeout=timeout)\n        else:\n            return False\n\n    def add_done_callback(self, fn):\n        if self.parent:\n            return self.parent.add_done_callback(fn)\n        else:\n            return None\n\n    @property\n    def outputs(self):\n        return self._outputs\n\n    def __repr__(self):\n        if self.parent:\n            with self.parent._condition:\n                if self.parent._state == FINISHED:\n                    if self.parent._exception:\n                        return '<%s at %#x state=%s raised %s>' % (\n                            self.__class__.__name__,\n                            id(self),\n                            _STATE_TO_DESCRIPTION_MAP[self.parent._state],\n                            self.parent._exception.__class__.__name__)\n                    else:\n                        return '<%s at %#x state=%s returned %s>' % (\n                            self.__class__.__name__,\n                            id(self),\n                            _STATE_TO_DESCRIPTION_MAP[self.parent._state],\n                            self.parent._result.__class__.__name__)\n                return '<%s at %#x state=%s>' % (\n                    self.__class__.__name__,\n                    id(self),\n                    _STATE_TO_DESCRIPTION_MAP[self.parent._state])\n        else:\n            return '<%s at %#x state=%s>' % (\n                self.__class__.__name__,\n                id(self),\n                _STATE_TO_DESCRIPTION_MAP[self._state])\n", "path": "parsl/dataflow/futures.py"}]}