We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Improve datafuture representation
Here is an example: `<DataFuture at 0x7f305c0b1da0 state=finished returned /home/annawoodard/pnpfit/results/best-fit-cuB.root_file>`

I do not think we should append `_file` to the end of the filepath, it makes it confusing what the actual filepath is.

--- 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/app/futures.py`
Content:
```
1 """This module implements DataFutures.
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 import os
8 import logging
9 from concurrent.futures import Future
10 
11 from parsl.dataflow.futures import AppFuture
12 from parsl.app.errors import *
13 from parsl.data_provider.files import File
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 DataFuture(Future):
36     """A datafuture points at an AppFuture.
37 
38     We are simply wrapping a AppFuture, and adding the specific case where, if
39     the future is resolved i.e file exists, then the DataFuture is assumed to be
40     resolved.
41     """
42 
43     def parent_callback(self, parent_fu):
44         """Callback from executor future to update the parent.
45 
46         Args:
47             - parent_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         if parent_fu.done() is True:
55             e = parent_fu._exception
56             if e:
57                 super().set_exception(e)
58             else:
59                 super().set_result(parent_fu.result())
60         return
61 
62     def __init__(self, fut, file_obj, parent=None, tid=None):
63         """Construct the DataFuture object.
64 
65         If the file_obj is a string convert to a File.
66 
67         Args:
68             - fut (AppFuture) : AppFuture that this DataFuture will track
69             - file_obj (string/File obj) : Something representing file(s)
70 
71         Kwargs:
72             - parent ()
73             - tid (task_id) : Task id that this DataFuture tracks
74         """
75         super().__init__()
76         self._tid = tid
77         if isinstance(file_obj, str) and not isinstance(file_obj, File):
78             self.file_obj = File(file_obj)
79         else:
80             self.file_obj = file_obj
81         self.parent = parent
82         self._exception = None
83 
84         if fut is None:
85             logger.debug("Setting result to filepath since no future was passed")
86             self.set_result = self.file_obj
87 
88         else:
89             if isinstance(fut, Future):
90                 self.parent = fut
91                 self.parent.add_done_callback(self.parent_callback)
92             else:
93                 raise NotFutureError("DataFuture can be created only with a FunctionFuture on None")
94 
95         logger.debug("Creating DataFuture with parent: %s", parent)
96         logger.debug("Filepath: %s", self.filepath)
97 
98     @property
99     def tid(self):
100         """Returns the task_id of the task that will resolve this DataFuture."""
101         return self._tid
102 
103     @property
104     def filepath(self):
105         """Filepath of the File object this datafuture represents."""
106         return self.file_obj.filepath
107 
108     @property
109     def filename(self):
110         """Filepath of the File object this datafuture represents."""
111         return self.filepath
112 
113     def result(self, timeout=None):
114         """A blocking call that returns either the result or raises an exception.
115 
116         Assumptions : A DataFuture always has a parent AppFuture. The AppFuture does callbacks when
117         setup.
118 
119         Kwargs:
120             - timeout (int): Timeout in seconds
121 
122         Returns:
123             - If App completed successfully returns the filepath.
124 
125         Raises:
126             - Exception raised by app if failed.
127 
128         """
129         if self.parent:
130             if self.parent.done():
131                 # This explicit call to raise exceptions might be redundant.
132                 # the result() call *should* raise an exception if there's one
133                 e = self.parent._exception
134                 if e:
135                     raise e
136                 else:
137                     self.parent.result(timeout=timeout)
138             else:
139                 self.parent.result(timeout=timeout)
140 
141         return self.file_obj
142 
143     def cancel(self):
144         """Cancel the task that this DataFuture is tracking.
145 
146             Note: This may not work
147         """
148         if self.parent:
149             return self.parent.cancel
150         else:
151             return False
152 
153     def cancelled(self):
154         if self.parent:
155             return self.parent.cancelled()
156         else:
157             return False
158 
159     def running(self):
160         if self.parent:
161             return self.parent.running()
162         else:
163             return False
164 
165     def done(self):
166         if self.parent:
167             return self.parent.done()
168         else:
169             return True
170 
171     def exception(self, timeout=None):
172         if self.parent:
173             return self.parent.exception(timeout=timeout)
174         else:
175             return True
176 
177     def add_done_callback(self, fn):
178         if self.parent:
179             return self.parent.add_done_callback(fn)
180         else:
181             return None
182 
183     def __repr__(self):
184 
185         # The DataFuture could be wrapping an AppFuture whose parent is a Future
186         # check to find the top level parent
187         if isinstance(self.parent, AppFuture):
188             parent = self.parent.parent
189         else:
190             parent = self.parent
191 
192         if parent:
193             with parent._condition:
194                 if parent._state == FINISHED:
195                     if parent._exception:
196                         return '<%s at %#x state=%s raised %s>' % (
197                             self.__class__.__name__,
198                             id(self),
199                             _STATE_TO_DESCRIPTION_MAP[parent._state],
200                             parent._exception.__class__.__name__)
201                     else:
202                         return '<%s at %#x state=%s returned %s>' % (
203                             self.__class__.__name__,
204                             id(self),
205                             _STATE_TO_DESCRIPTION_MAP[parent._state],
206                             self.filepath + '_file')
207                 return '<%s at %#x state=%s>' % (
208                     self.__class__.__name__,
209                     id(self),
210                     _STATE_TO_DESCRIPTION_MAP[parent._state])
211 
212         else:
213             return '<%s at %#x state=%s>' % (
214                 self.__class__.__name__,
215                 id(self),
216                 _STATE_TO_DESCRIPTION_MAP[self._state])
217 
218 
219 def testing_nonfuture():
220     fpath = '~/shuffled.txt'
221     df = DataFuture(None, fpath)
222     print(df)
223     print("Result: ", df.filepath)
224     assert df.filepath == os.path.abspath(os.path.expanduser(fpath))
225 
226 
227 if __name__ == "__main__":
228     # logging.basicConfig(filename='futures.testing.log',level=logging.DEBUG)
229     import sys
230     import random
231     logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
232     logger.debug("Begin Testing")
233 
234     with open('shuffled.txt', 'w') as testfile:
235         nums = list(range(0, 10000))
236         random.shuffle(nums)
237         for item in nums:
238             testfile.write("{0}\n".format(item))
239 
240     foo = Future()
241     df = DataFuture(foo, './shuffled.txt')
242     dx = DataFuture(foo, '~/shuffled.txt')
243 
244     print(foo.done())
245     print(df.done())
246 
247     testing_nonfuture()
248
```
--- 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/app/futures.py b/parsl/app/futures.py
--- a/parsl/app/futures.py
+++ b/parsl/app/futures.py
@@ -203,7 +203,7 @@
                             self.__class__.__name__,
                             id(self),
                             _STATE_TO_DESCRIPTION_MAP[parent._state],
-                            self.filepath + '_file')
+                            self.filepath)
                 return '<%s at %#x state=%s>' % (
                     self.__class__.__name__,
                     id(self),

{"golden_diff": "diff --git a/parsl/app/futures.py b/parsl/app/futures.py\n--- a/parsl/app/futures.py\n+++ b/parsl/app/futures.py\n@@ -203,7 +203,7 @@\n                             self.__class__.__name__,\n                             id(self),\n                             _STATE_TO_DESCRIPTION_MAP[parent._state],\n-                            self.filepath + '_file')\n+                            self.filepath)\n                 return '<%s at %#x state=%s>' % (\n                     self.__class__.__name__,\n                     id(self),\n", "issue": "Improve datafuture representation\nHere is an example: `<DataFuture at 0x7f305c0b1da0 state=finished returned /home/annawoodard/pnpfit/results/best-fit-cuB.root_file>`\r\n\r\nI do not think we should append `_file` to the end of the filepath, it makes it confusing what the actual filepath is.\n", "before_files": [{"content": "\"\"\"This module implements DataFutures.\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\"\"\"\nimport os\nimport logging\nfrom concurrent.futures import Future\n\nfrom parsl.dataflow.futures import AppFuture\nfrom parsl.app.errors import *\nfrom parsl.data_provider.files import File\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 DataFuture(Future):\n    \"\"\"A datafuture points at an AppFuture.\n\n    We are simply wrapping a AppFuture, and adding the specific case where, if\n    the future is resolved i.e file exists, then the DataFuture is assumed to be\n    resolved.\n    \"\"\"\n\n    def parent_callback(self, parent_fu):\n        \"\"\"Callback from executor future to update the parent.\n\n        Args:\n            - parent_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        if parent_fu.done() is True:\n            e = parent_fu._exception\n            if e:\n                super().set_exception(e)\n            else:\n                super().set_result(parent_fu.result())\n        return\n\n    def __init__(self, fut, file_obj, parent=None, tid=None):\n        \"\"\"Construct the DataFuture object.\n\n        If the file_obj is a string convert to a File.\n\n        Args:\n            - fut (AppFuture) : AppFuture that this DataFuture will track\n            - file_obj (string/File obj) : Something representing file(s)\n\n        Kwargs:\n            - parent ()\n            - tid (task_id) : Task id that this DataFuture tracks\n        \"\"\"\n        super().__init__()\n        self._tid = tid\n        if isinstance(file_obj, str) and not isinstance(file_obj, File):\n            self.file_obj = File(file_obj)\n        else:\n            self.file_obj = file_obj\n        self.parent = parent\n        self._exception = None\n\n        if fut is None:\n            logger.debug(\"Setting result to filepath since no future was passed\")\n            self.set_result = self.file_obj\n\n        else:\n            if isinstance(fut, Future):\n                self.parent = fut\n                self.parent.add_done_callback(self.parent_callback)\n            else:\n                raise NotFutureError(\"DataFuture can be created only with a FunctionFuture on None\")\n\n        logger.debug(\"Creating DataFuture with parent: %s\", parent)\n        logger.debug(\"Filepath: %s\", self.filepath)\n\n    @property\n    def tid(self):\n        \"\"\"Returns the task_id of the task that will resolve this DataFuture.\"\"\"\n        return self._tid\n\n    @property\n    def filepath(self):\n        \"\"\"Filepath of the File object this datafuture represents.\"\"\"\n        return self.file_obj.filepath\n\n    @property\n    def filename(self):\n        \"\"\"Filepath of the File object this datafuture represents.\"\"\"\n        return self.filepath\n\n    def result(self, timeout=None):\n        \"\"\"A blocking call that returns either the result or raises an exception.\n\n        Assumptions : A DataFuture always has a parent AppFuture. The AppFuture does callbacks when\n        setup.\n\n        Kwargs:\n            - timeout (int): Timeout in seconds\n\n        Returns:\n            - If App completed successfully returns the filepath.\n\n        Raises:\n            - Exception raised by app if failed.\n\n        \"\"\"\n        if self.parent:\n            if self.parent.done():\n                # This explicit call to raise exceptions might be redundant.\n                # the result() call *should* raise an exception if there's one\n                e = self.parent._exception\n                if e:\n                    raise e\n                else:\n                    self.parent.result(timeout=timeout)\n            else:\n                self.parent.result(timeout=timeout)\n\n        return self.file_obj\n\n    def cancel(self):\n        \"\"\"Cancel the task that this DataFuture is tracking.\n\n            Note: This may not work\n        \"\"\"\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        if self.parent:\n            return self.parent.done()\n        else:\n            return True\n\n    def exception(self, timeout=None):\n        if self.parent:\n            return self.parent.exception(timeout=timeout)\n        else:\n            return True\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    def __repr__(self):\n\n        # The DataFuture could be wrapping an AppFuture whose parent is a Future\n        # check to find the top level parent\n        if isinstance(self.parent, AppFuture):\n            parent = self.parent.parent\n        else:\n            parent = self.parent\n\n        if parent:\n            with parent._condition:\n                if parent._state == FINISHED:\n                    if parent._exception:\n                        return '<%s at %#x state=%s raised %s>' % (\n                            self.__class__.__name__,\n                            id(self),\n                            _STATE_TO_DESCRIPTION_MAP[parent._state],\n                            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[parent._state],\n                            self.filepath + '_file')\n                return '<%s at %#x state=%s>' % (\n                    self.__class__.__name__,\n                    id(self),\n                    _STATE_TO_DESCRIPTION_MAP[parent._state])\n\n        else:\n            return '<%s at %#x state=%s>' % (\n                self.__class__.__name__,\n                id(self),\n                _STATE_TO_DESCRIPTION_MAP[self._state])\n\n\ndef testing_nonfuture():\n    fpath = '~/shuffled.txt'\n    df = DataFuture(None, fpath)\n    print(df)\n    print(\"Result: \", df.filepath)\n    assert df.filepath == os.path.abspath(os.path.expanduser(fpath))\n\n\nif __name__ == \"__main__\":\n    # logging.basicConfig(filename='futures.testing.log',level=logging.DEBUG)\n    import sys\n    import random\n    logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)\n    logger.debug(\"Begin Testing\")\n\n    with open('shuffled.txt', 'w') as testfile:\n        nums = list(range(0, 10000))\n        random.shuffle(nums)\n        for item in nums:\n            testfile.write(\"{0}\\n\".format(item))\n\n    foo = Future()\n    df = DataFuture(foo, './shuffled.txt')\n    dx = DataFuture(foo, '~/shuffled.txt')\n\n    print(foo.done())\n    print(df.done())\n\n    testing_nonfuture()\n", "path": "parsl/app/futures.py"}], "after_files": [{"content": "\"\"\"This module implements DataFutures.\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\"\"\"\nimport os\nimport logging\nfrom concurrent.futures import Future\n\nfrom parsl.dataflow.futures import AppFuture\nfrom parsl.app.errors import *\nfrom parsl.data_provider.files import File\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 DataFuture(Future):\n    \"\"\"A datafuture points at an AppFuture.\n\n    We are simply wrapping a AppFuture, and adding the specific case where, if\n    the future is resolved i.e file exists, then the DataFuture is assumed to be\n    resolved.\n    \"\"\"\n\n    def parent_callback(self, parent_fu):\n        \"\"\"Callback from executor future to update the parent.\n\n        Args:\n            - parent_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        if parent_fu.done() is True:\n            e = parent_fu._exception\n            if e:\n                super().set_exception(e)\n            else:\n                super().set_result(parent_fu.result())\n        return\n\n    def __init__(self, fut, file_obj, parent=None, tid=None):\n        \"\"\"Construct the DataFuture object.\n\n        If the file_obj is a string convert to a File.\n\n        Args:\n            - fut (AppFuture) : AppFuture that this DataFuture will track\n            - file_obj (string/File obj) : Something representing file(s)\n\n        Kwargs:\n            - parent ()\n            - tid (task_id) : Task id that this DataFuture tracks\n        \"\"\"\n        super().__init__()\n        self._tid = tid\n        if isinstance(file_obj, str) and not isinstance(file_obj, File):\n            self.file_obj = File(file_obj)\n        else:\n            self.file_obj = file_obj\n        self.parent = parent\n        self._exception = None\n\n        if fut is None:\n            logger.debug(\"Setting result to filepath since no future was passed\")\n            self.set_result = self.file_obj\n\n        else:\n            if isinstance(fut, Future):\n                self.parent = fut\n                self.parent.add_done_callback(self.parent_callback)\n            else:\n                raise NotFutureError(\"DataFuture can be created only with a FunctionFuture on None\")\n\n        logger.debug(\"Creating DataFuture with parent: %s\", parent)\n        logger.debug(\"Filepath: %s\", self.filepath)\n\n    @property\n    def tid(self):\n        \"\"\"Returns the task_id of the task that will resolve this DataFuture.\"\"\"\n        return self._tid\n\n    @property\n    def filepath(self):\n        \"\"\"Filepath of the File object this datafuture represents.\"\"\"\n        return self.file_obj.filepath\n\n    @property\n    def filename(self):\n        \"\"\"Filepath of the File object this datafuture represents.\"\"\"\n        return self.filepath\n\n    def result(self, timeout=None):\n        \"\"\"A blocking call that returns either the result or raises an exception.\n\n        Assumptions : A DataFuture always has a parent AppFuture. The AppFuture does callbacks when\n        setup.\n\n        Kwargs:\n            - timeout (int): Timeout in seconds\n\n        Returns:\n            - If App completed successfully returns the filepath.\n\n        Raises:\n            - Exception raised by app if failed.\n\n        \"\"\"\n        if self.parent:\n            if self.parent.done():\n                # This explicit call to raise exceptions might be redundant.\n                # the result() call *should* raise an exception if there's one\n                e = self.parent._exception\n                if e:\n                    raise e\n                else:\n                    self.parent.result(timeout=timeout)\n            else:\n                self.parent.result(timeout=timeout)\n\n        return self.file_obj\n\n    def cancel(self):\n        \"\"\"Cancel the task that this DataFuture is tracking.\n\n            Note: This may not work\n        \"\"\"\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        if self.parent:\n            return self.parent.done()\n        else:\n            return True\n\n    def exception(self, timeout=None):\n        if self.parent:\n            return self.parent.exception(timeout=timeout)\n        else:\n            return True\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    def __repr__(self):\n\n        # The DataFuture could be wrapping an AppFuture whose parent is a Future\n        # check to find the top level parent\n        if isinstance(self.parent, AppFuture):\n            parent = self.parent.parent\n        else:\n            parent = self.parent\n\n        if parent:\n            with parent._condition:\n                if parent._state == FINISHED:\n                    if parent._exception:\n                        return '<%s at %#x state=%s raised %s>' % (\n                            self.__class__.__name__,\n                            id(self),\n                            _STATE_TO_DESCRIPTION_MAP[parent._state],\n                            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[parent._state],\n                            self.filepath)\n                return '<%s at %#x state=%s>' % (\n                    self.__class__.__name__,\n                    id(self),\n                    _STATE_TO_DESCRIPTION_MAP[parent._state])\n\n        else:\n            return '<%s at %#x state=%s>' % (\n                self.__class__.__name__,\n                id(self),\n                _STATE_TO_DESCRIPTION_MAP[self._state])\n\n\ndef testing_nonfuture():\n    fpath = '~/shuffled.txt'\n    df = DataFuture(None, fpath)\n    print(df)\n    print(\"Result: \", df.filepath)\n    assert df.filepath == os.path.abspath(os.path.expanduser(fpath))\n\n\nif __name__ == \"__main__\":\n    # logging.basicConfig(filename='futures.testing.log',level=logging.DEBUG)\n    import sys\n    import random\n    logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)\n    logger.debug(\"Begin Testing\")\n\n    with open('shuffled.txt', 'w') as testfile:\n        nums = list(range(0, 10000))\n        random.shuffle(nums)\n        for item in nums:\n            testfile.write(\"{0}\\n\".format(item))\n\n    foo = Future()\n    df = DataFuture(foo, './shuffled.txt')\n    dx = DataFuture(foo, '~/shuffled.txt')\n\n    print(foo.done())\n    print(df.done())\n\n    testing_nonfuture()\n", "path": "parsl/app/futures.py"}]}