We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
using walltime for python app results in Recursion Error
I've been trying to use the timeout functionality of parsl via the walltime=X argument. To narrow it down to a minimum example, I took the test in the test suite and modified it to invoke the app many times:

```
import parsl
from parsl.app.errors import AppTimeout
import pytest
from parsl.configs.local_threads import config

from parsl.providers import LocalProvider
from parsl.channels import LocalChannel
from parsl.config import Config
from parsl.executors import HighThroughputExecutor


local_htex = Config(
    executors=[
        HighThroughputExecutor(
            label="htex_Local",
            worker_debug=True,
            cores_per_worker=1,
            provider=LocalProvider(
                channel=LocalChannel(),
                init_blocks=1,
                max_blocks=1,
            ),
        )
    ],
    strategy=None,
)

parsl.load(local_htex)


@parsl.python_app
def my_app(duration, bar=1.0, walltime=1):
    import time
    time.sleep(duration*bar)
    return True


def test_python_walltime():
    fs = []
    for i in range(1024):
        f = my_app(2.0)
        fs.append(f)
    print("all invoked")
    for f in fs:
        with pytest.raises(AppTimeout):
            f.result()

if __name__ == "__main__":
    test_python_walltime()

```

This gives a (deep) stack trace with repeated 
```
  File "/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py", line 393, in can
    return canner(obj)
  File "/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py", line 186, in __init__
    self.closure = tuple( can(cell) for cell in closure )
  File "/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py", line 186, in <genexpr>
    self.closure = tuple( can(cell) for cell in closure )
  File "/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py", line 393, in can
    return canner(obj)
  File "/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py", line 165, in __init__
    self.cell_contents = can(cell.cell_contents)
  File "/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py", line 393, in can
    return canner(obj)
  File "/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py", line 186, in __init__
    self.closure = tuple( can(cell) for cell in closure )
  File "/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py", line 186, in <genexpr>
    self.closure = tuple( can(cell) for cell in closure )
  File "/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py", line 393, in can
    return canner(obj)
  File "/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py", line 165, in __init__
    self.cell_contents = can(cell.cell_contents)
  File "/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py", line 393, in can
    return canner(obj)
  File "/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py", line 177, in __init__
    self._check_type(f)
RecursionError: maximum recursion depth exceeded

```

style of errors. Note that this only appears when the `walltime` arg is present in `my_app` -- if you delete that, this example works flawlessly.  Also note the recursion error does NOT occur when using the local thread executor. 

I currently see this with both master and the pip-installable version of parsl on a linux host (tacc login node) installed via the newest anaconda with python 3.7. 

Unfortunately this is currently blocking experiments as I have ~1% of my jobs that take 100x longer than the others and thus I'd like them to die when they exceed a limit. 


--- 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/python.py`
Content:
```
1 import logging
2 
3 import tblib.pickling_support
4 tblib.pickling_support.install()
5 
6 from parsl.app.app import AppBase
7 from parsl.app.errors import wrap_error
8 from parsl.dataflow.dflow import DataFlowKernelLoader
9 
10 
11 logger = logging.getLogger(__name__)
12 
13 
14 def timeout(f, seconds):
15     def wrapper(*args, **kwargs):
16         import threading
17         import ctypes
18         import parsl.app.errors
19 
20         def inject_exception(thread):
21             ctypes.pythonapi.PyThreadState_SetAsyncExc(
22                 ctypes.c_long(thread),
23                 ctypes.py_object(parsl.app.errors.AppTimeout)
24             )
25 
26         thread = threading.current_thread().ident
27         timer = threading.Timer(seconds, inject_exception, args=[thread])
28         timer.start()
29         result = f(*args, **kwargs)
30         timer.cancel()
31         return result
32     return wrapper
33 
34 
35 class PythonApp(AppBase):
36     """Extends AppBase to cover the Python App."""
37 
38     def __init__(self, func, data_flow_kernel=None, cache=False, executors='all'):
39         super().__init__(
40             wrap_error(func),
41             data_flow_kernel=data_flow_kernel,
42             executors=executors,
43             cache=cache
44         )
45 
46     def __call__(self, *args, **kwargs):
47         """This is where the call to a python app is handled.
48 
49         Args:
50              - Arbitrary
51         Kwargs:
52              - Arbitrary
53 
54         Returns:
55                    App_fut
56 
57         """
58 
59         if self.data_flow_kernel is None:
60             dfk = DataFlowKernelLoader.dfk()
61         else:
62             dfk = self.data_flow_kernel
63 
64         walltime = self.kwargs.get('walltime')
65         if walltime is not None:
66             self.func = timeout(self.func, walltime)
67         app_fut = dfk.submit(self.func, *args,
68                              executors=self.executors,
69                              fn_hash=self.func_hash,
70                              cache=self.cache,
71                              **kwargs)
72 
73         return app_fut
74
```
--- 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/python.py b/parsl/app/python.py
--- a/parsl/app/python.py
+++ b/parsl/app/python.py
@@ -63,8 +63,11 @@
 
         walltime = self.kwargs.get('walltime')
         if walltime is not None:
-            self.func = timeout(self.func, walltime)
-        app_fut = dfk.submit(self.func, *args,
+            func = timeout(self.func, walltime)
+        else:
+            func = self.func
+
+        app_fut = dfk.submit(func, *args,
                              executors=self.executors,
                              fn_hash=self.func_hash,
                              cache=self.cache,

{"golden_diff": "diff --git a/parsl/app/python.py b/parsl/app/python.py\n--- a/parsl/app/python.py\n+++ b/parsl/app/python.py\n@@ -63,8 +63,11 @@\n \n         walltime = self.kwargs.get('walltime')\n         if walltime is not None:\n-            self.func = timeout(self.func, walltime)\n-        app_fut = dfk.submit(self.func, *args,\n+            func = timeout(self.func, walltime)\n+        else:\n+            func = self.func\n+\n+        app_fut = dfk.submit(func, *args,\n                              executors=self.executors,\n                              fn_hash=self.func_hash,\n                              cache=self.cache,\n", "issue": "using walltime for python app results in Recursion Error\nI've been trying to use the timeout functionality of parsl via the walltime=X argument. To narrow it down to a minimum example, I took the test in the test suite and modified it to invoke the app many times:\r\n\r\n```\r\nimport parsl\r\nfrom parsl.app.errors import AppTimeout\r\nimport pytest\r\nfrom parsl.configs.local_threads import config\r\n\r\nfrom parsl.providers import LocalProvider\r\nfrom parsl.channels import LocalChannel\r\nfrom parsl.config import Config\r\nfrom parsl.executors import HighThroughputExecutor\r\n\r\n\r\nlocal_htex = Config(\r\n    executors=[\r\n        HighThroughputExecutor(\r\n            label=\"htex_Local\",\r\n            worker_debug=True,\r\n            cores_per_worker=1,\r\n            provider=LocalProvider(\r\n                channel=LocalChannel(),\r\n                init_blocks=1,\r\n                max_blocks=1,\r\n            ),\r\n        )\r\n    ],\r\n    strategy=None,\r\n)\r\n\r\nparsl.load(local_htex)\r\n\r\n\r\n@parsl.python_app\r\ndef my_app(duration, bar=1.0, walltime=1):\r\n    import time\r\n    time.sleep(duration*bar)\r\n    return True\r\n\r\n\r\ndef test_python_walltime():\r\n    fs = []\r\n    for i in range(1024):\r\n        f = my_app(2.0)\r\n        fs.append(f)\r\n    print(\"all invoked\")\r\n    for f in fs:\r\n        with pytest.raises(AppTimeout):\r\n            f.result()\r\n\r\nif __name__ == \"__main__\":\r\n    test_python_walltime()\r\n\r\n```\r\n\r\nThis gives a (deep) stack trace with repeated \r\n```\r\n  File \"/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py\", line 393, in can\r\n    return canner(obj)\r\n  File \"/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py\", line 186, in __init__\r\n    self.closure = tuple( can(cell) for cell in closure )\r\n  File \"/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py\", line 186, in <genexpr>\r\n    self.closure = tuple( can(cell) for cell in closure )\r\n  File \"/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py\", line 393, in can\r\n    return canner(obj)\r\n  File \"/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py\", line 165, in __init__\r\n    self.cell_contents = can(cell.cell_contents)\r\n  File \"/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py\", line 393, in can\r\n    return canner(obj)\r\n  File \"/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py\", line 186, in __init__\r\n    self.closure = tuple( can(cell) for cell in closure )\r\n  File \"/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py\", line 186, in <genexpr>\r\n    self.closure = tuple( can(cell) for cell in closure )\r\n  File \"/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py\", line 393, in can\r\n    return canner(obj)\r\n  File \"/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py\", line 165, in __init__\r\n    self.cell_contents = can(cell.cell_contents)\r\n  File \"/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py\", line 393, in can\r\n    return canner(obj)\r\n  File \"/work/04372/ejonas/stampede2/anaconda/envs/nmr-abinitio/lib/python3.7/site-packages/ipyparallel/serialize/canning.py\", line 177, in __init__\r\n    self._check_type(f)\r\nRecursionError: maximum recursion depth exceeded\r\n\r\n```\r\n\r\nstyle of errors. Note that this only appears when the `walltime` arg is present in `my_app` -- if you delete that, this example works flawlessly.  Also note the recursion error does NOT occur when using the local thread executor. \r\n\r\nI currently see this with both master and the pip-installable version of parsl on a linux host (tacc login node) installed via the newest anaconda with python 3.7. \r\n\r\nUnfortunately this is currently blocking experiments as I have ~1% of my jobs that take 100x longer than the others and thus I'd like them to die when they exceed a limit. \r\n\n", "before_files": [{"content": "import logging\n\nimport tblib.pickling_support\ntblib.pickling_support.install()\n\nfrom parsl.app.app import AppBase\nfrom parsl.app.errors import wrap_error\nfrom parsl.dataflow.dflow import DataFlowKernelLoader\n\n\nlogger = logging.getLogger(__name__)\n\n\ndef timeout(f, seconds):\n    def wrapper(*args, **kwargs):\n        import threading\n        import ctypes\n        import parsl.app.errors\n\n        def inject_exception(thread):\n            ctypes.pythonapi.PyThreadState_SetAsyncExc(\n                ctypes.c_long(thread),\n                ctypes.py_object(parsl.app.errors.AppTimeout)\n            )\n\n        thread = threading.current_thread().ident\n        timer = threading.Timer(seconds, inject_exception, args=[thread])\n        timer.start()\n        result = f(*args, **kwargs)\n        timer.cancel()\n        return result\n    return wrapper\n\n\nclass PythonApp(AppBase):\n    \"\"\"Extends AppBase to cover the Python App.\"\"\"\n\n    def __init__(self, func, data_flow_kernel=None, cache=False, executors='all'):\n        super().__init__(\n            wrap_error(func),\n            data_flow_kernel=data_flow_kernel,\n            executors=executors,\n            cache=cache\n        )\n\n    def __call__(self, *args, **kwargs):\n        \"\"\"This is where the call to a python app is handled.\n\n        Args:\n             - Arbitrary\n        Kwargs:\n             - Arbitrary\n\n        Returns:\n                   App_fut\n\n        \"\"\"\n\n        if self.data_flow_kernel is None:\n            dfk = DataFlowKernelLoader.dfk()\n        else:\n            dfk = self.data_flow_kernel\n\n        walltime = self.kwargs.get('walltime')\n        if walltime is not None:\n            self.func = timeout(self.func, walltime)\n        app_fut = dfk.submit(self.func, *args,\n                             executors=self.executors,\n                             fn_hash=self.func_hash,\n                             cache=self.cache,\n                             **kwargs)\n\n        return app_fut\n", "path": "parsl/app/python.py"}], "after_files": [{"content": "import logging\n\nimport tblib.pickling_support\ntblib.pickling_support.install()\n\nfrom parsl.app.app import AppBase\nfrom parsl.app.errors import wrap_error\nfrom parsl.dataflow.dflow import DataFlowKernelLoader\n\n\nlogger = logging.getLogger(__name__)\n\n\ndef timeout(f, seconds):\n    def wrapper(*args, **kwargs):\n        import threading\n        import ctypes\n        import parsl.app.errors\n\n        def inject_exception(thread):\n            ctypes.pythonapi.PyThreadState_SetAsyncExc(\n                ctypes.c_long(thread),\n                ctypes.py_object(parsl.app.errors.AppTimeout)\n            )\n\n        thread = threading.current_thread().ident\n        timer = threading.Timer(seconds, inject_exception, args=[thread])\n        timer.start()\n        result = f(*args, **kwargs)\n        timer.cancel()\n        return result\n    return wrapper\n\n\nclass PythonApp(AppBase):\n    \"\"\"Extends AppBase to cover the Python App.\"\"\"\n\n    def __init__(self, func, data_flow_kernel=None, cache=False, executors='all'):\n        super().__init__(\n            wrap_error(func),\n            data_flow_kernel=data_flow_kernel,\n            executors=executors,\n            cache=cache\n        )\n\n    def __call__(self, *args, **kwargs):\n        \"\"\"This is where the call to a python app is handled.\n\n        Args:\n             - Arbitrary\n        Kwargs:\n             - Arbitrary\n\n        Returns:\n                   App_fut\n\n        \"\"\"\n\n        if self.data_flow_kernel is None:\n            dfk = DataFlowKernelLoader.dfk()\n        else:\n            dfk = self.data_flow_kernel\n\n        walltime = self.kwargs.get('walltime')\n        if walltime is not None:\n            func = timeout(self.func, walltime)\n        else:\n            func = self.func\n\n        app_fut = dfk.submit(func, *args,\n                             executors=self.executors,\n                             fn_hash=self.func_hash,\n                             cache=self.cache,\n                             **kwargs)\n\n        return app_fut\n", "path": "parsl/app/python.py"}]}