We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Implement time limits for python apps
Requested by @lgray.

--- 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/app.py`
Content:
```
1 """Definitions for the @App decorator and the App classes.
2 
3 The App class encapsulates a generic leaf task that can be executed asynchronously.
4 """
5 import logging
6 from abc import ABCMeta, abstractmethod
7 from inspect import getsource
8 from hashlib import md5
9 from inspect import signature
10 
11 from parsl.app.errors import InvalidAppTypeError
12 
13 logger = logging.getLogger(__name__)
14 
15 
16 class AppBase(metaclass=ABCMeta):
17     """This is the base class that defines the two external facing functions that an App must define.
18 
19     The  __init__ () which is called when the interpreter sees the definition of the decorated
20     function, and the __call__ () which is invoked when a decorated function is called by the user.
21 
22     """
23 
24     def __init__(self, func, data_flow_kernel=None, walltime=60, executors='all', cache=False):
25         """Construct the App object.
26 
27         Args:
28              - func (function): Takes the function to be made into an App
29 
30         Kwargs:
31              - data_flow_kernel (DataFlowKernel): The :class:`~parsl.dataflow.dflow.DataFlowKernel` responsible for
32                managing this app. This can be omitted only
33                after calling :meth:`parsl.dataflow.dflow.DataFlowKernelLoader.load`.
34              - walltime (int) : Walltime in seconds for the app execution.
35              - executors (str|list) : Labels of the executors that this app can execute over. Default is 'all'.
36              - cache (Bool) : Enable caching of this app ?
37 
38         Returns:
39              - App object.
40 
41         """
42         self.__name__ = func.__name__
43         self.func = func
44         self.data_flow_kernel = data_flow_kernel
45         self.status = 'created'
46         self.executors = executors
47         self.cache = cache
48         if not (isinstance(executors, list) or isinstance(executors, str)):
49             logger.error("App {} specifies invalid executor option, expects string or list".format(
50                 func.__name__))
51 
52         if cache is True:
53             try:
54                 self.fn_source = getsource(func)
55             except OSError:
56                 logger.debug("Unable to get source code for AppCaching. Recommend creating module")
57                 self.fn_source = func.__name__
58 
59             self.func_hash = md5(self.fn_source.encode('utf-8')).hexdigest()
60         else:
61             self.func_hash = func.__name__
62 
63         params = signature(func).parameters
64 
65         self.kwargs = {}
66         if 'stdout' in params:
67             self.kwargs['stdout'] = params['stdout'].default
68         if 'stderr' in params:
69             self.kwargs['stderr'] = params['stderr'].default
70         self.outputs = params['outputs'].default if 'outputs' in params else []
71         self.inputs = params['inputs'].default if 'inputs' in params else []
72 
73     @abstractmethod
74     def __call__(self, *args, **kwargs):
75         pass
76 
77 
78 def App(apptype, data_flow_kernel=None, walltime=60, cache=False, executors='all'):
79     """The App decorator function.
80 
81     Args:
82         - apptype (string) : Apptype can be bash|python
83 
84     Kwargs:
85         - data_flow_kernel (DataFlowKernel): The :class:`~parsl.dataflow.dflow.DataFlowKernel` responsible for
86           managing this app. This can be omitted only
87           after calling :meth:`parsl.dataflow.dflow.DataFlowKernelLoader.load`.
88         - walltime (int) : Walltime for app in seconds,
89              default=60
90         - executors (str|list) : Labels of the executors that this app can execute over. Default is 'all'.
91         - cache (Bool) : Enable caching of the app call
92              default=False
93 
94     Returns:
95          A PythonApp or BashApp object, which when called runs the apps through the executor.
96     """
97 
98     from parsl.app.python import PythonApp
99     from parsl.app.bash import BashApp
100 
101     logger.warning("The 'App' decorator will be deprecated in Parsl 0.8. Please use 'python_app' or 'bash_app' instead.")
102 
103     if apptype == 'python':
104         app_class = PythonApp
105     elif apptype == 'bash':
106         app_class = BashApp
107     else:
108         raise InvalidAppTypeError("Invalid apptype requested {}; must be 'python' or 'bash'".format(apptype))
109 
110     def wrapper(f):
111         return app_class(f,
112                          data_flow_kernel=data_flow_kernel,
113                          walltime=walltime,
114                          cache=cache,
115                          executors=executors)
116     return wrapper
117 
118 
119 def python_app(function=None, data_flow_kernel=None, walltime=60, cache=False, executors='all'):
120     """Decorator function for making python apps.
121 
122     Parameters
123     ----------
124     function : function
125         Do not pass this keyword argument directly. This is needed in order to allow for omitted parenthesis,
126         for example, `@python_app` if using all defaults or `@python_app(walltime=120)`. If the
127         decorator is used alone, function will be the actual function being decorated, whereas if it
128         is called with arguments, function will be None. Default is None.
129     data_flow_kernel : DataFlowKernel
130         The :class:`~parsl.dataflow.dflow.DataFlowKernel` responsible for managing this app. This can
131         be omitted only after calling :meth:`parsl.dataflow.dflow.DataFlowKernelLoader.load`. Default is None.
132     walltime : int
133         Walltime for app in seconds. Default is 60.
134     executors : string or list
135         Labels of the executors that this app can execute over. Default is 'all'.
136     cache : bool
137         Enable caching of the app call. Default is False.
138     """
139     from parsl.app.python import PythonApp
140 
141     def decorator(func):
142         def wrapper(f):
143             return PythonApp(f,
144                              data_flow_kernel=data_flow_kernel,
145                              walltime=walltime,
146                              cache=cache,
147                              executors=executors)
148         return wrapper(func)
149     if function is not None:
150         return decorator(function)
151     return decorator
152 
153 
154 def bash_app(function=None, data_flow_kernel=None, walltime=60, cache=False, executors='all'):
155     """Decorator function for making bash apps.
156 
157     Parameters
158     ----------
159     function : function
160         Do not pass this keyword argument directly. This is needed in order to allow for omitted parenthesis,
161         for example, `@bash_app` if using all defaults or `@bash_app(walltime=120)`. If the
162         decorator is used alone, function will be the actual function being decorated, whereas if it
163         is called with arguments, function will be None. Default is None.
164     data_flow_kernel : DataFlowKernel
165         The :class:`~parsl.dataflow.dflow.DataFlowKernel` responsible for managing this app. This can
166         be omitted only after calling :meth:`parsl.dataflow.dflow.DataFlowKernelLoader.load`. Default is None.
167     walltime : int
168         Walltime for app in seconds. Default is 60.
169     executors : string or list
170         Labels of the executors that this app can execute over. Default is 'all'.
171     cache : bool
172         Enable caching of the app call. Default is False.
173     """
174     from parsl.app.bash import BashApp
175 
176     def decorator(func):
177         def wrapper(f):
178             return BashApp(f,
179                            data_flow_kernel=data_flow_kernel,
180                            walltime=walltime,
181                            cache=cache,
182                            executors=executors)
183         return wrapper(func)
184     if function is not None:
185         return decorator(function)
186     return decorator
187
```

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.futures import DataFuture
7 from parsl.app.app import AppBase
8 from parsl.app.errors import wrap_error
9 from parsl.dataflow.dflow import DataFlowKernelLoader
10 
11 
12 logger = logging.getLogger(__name__)
13 
14 
15 class PythonApp(AppBase):
16     """Extends AppBase to cover the Python App."""
17 
18     def __init__(self, func, data_flow_kernel=None, walltime=60, cache=False, executors='all'):
19         super().__init__(
20             wrap_error(func),
21             data_flow_kernel=data_flow_kernel,
22             walltime=walltime,
23             executors=executors,
24             cache=cache
25         )
26 
27     def __call__(self, *args, **kwargs):
28         """This is where the call to a python app is handled.
29 
30         Args:
31              - Arbitrary
32         Kwargs:
33              - Arbitrary
34 
35         Returns:
36              If outputs=[...] was a kwarg then:
37                    App_fut, [Data_Futures...]
38              else:
39                    App_fut
40 
41         """
42 
43         if self.data_flow_kernel is None:
44             dfk = DataFlowKernelLoader.dfk()
45         else:
46             dfk = self.data_flow_kernel
47 
48         app_fut = dfk.submit(self.func, *args,
49                              executors=self.executors,
50                              fn_hash=self.func_hash,
51                              cache=self.cache,
52                              **kwargs)
53 
54         # logger.debug("App[{}] assigned Task[{}]".format(self.func.__name__,
55         #                                                 app_fut.tid))
56         out_futs = [DataFuture(app_fut, o, tid=app_fut.tid)
57                     for o in kwargs.get('outputs', [])]
58         app_fut._outputs = out_futs
59 
60         return app_fut
61
```
--- 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/app.py b/parsl/app/app.py
--- a/parsl/app/app.py
+++ b/parsl/app/app.py
@@ -67,6 +67,8 @@
             self.kwargs['stdout'] = params['stdout'].default
         if 'stderr' in params:
             self.kwargs['stderr'] = params['stderr'].default
+        if 'walltime' in params:
+            self.kwargs['walltime'] = params['walltime'].default
         self.outputs = params['outputs'].default if 'outputs' in params else []
         self.inputs = params['inputs'].default if 'inputs' in params else []
 
diff --git a/parsl/app/python.py b/parsl/app/python.py
--- a/parsl/app/python.py
+++ b/parsl/app/python.py
@@ -12,6 +12,27 @@
 logger = logging.getLogger(__name__)
 
 
+def timeout(f, seconds):
+    def wrapper(*args, **kwargs):
+        import threading
+        import ctypes
+        import parsl.app.errors
+
+        def inject_exception(thread):
+            ctypes.pythonapi.PyThreadState_SetAsyncExc(
+                ctypes.c_long(thread),
+                ctypes.py_object(parsl.app.errors.AppTimeout)
+            )
+
+        thread = threading.current_thread().ident
+        timer = threading.Timer(seconds, inject_exception, args=[thread])
+        timer.start()
+        result = f(*args, **kwargs)
+        timer.cancel()
+        return result
+    return wrapper
+
+
 class PythonApp(AppBase):
     """Extends AppBase to cover the Python App."""
 
@@ -45,6 +66,9 @@
         else:
             dfk = self.data_flow_kernel
 
+        walltime = self.kwargs.get('walltime')
+        if walltime is not None:
+            self.func = timeout(self.func, walltime)
         app_fut = dfk.submit(self.func, *args,
                              executors=self.executors,
                              fn_hash=self.func_hash,

{"golden_diff": "diff --git a/parsl/app/app.py b/parsl/app/app.py\n--- a/parsl/app/app.py\n+++ b/parsl/app/app.py\n@@ -67,6 +67,8 @@\n             self.kwargs['stdout'] = params['stdout'].default\n         if 'stderr' in params:\n             self.kwargs['stderr'] = params['stderr'].default\n+        if 'walltime' in params:\n+            self.kwargs['walltime'] = params['walltime'].default\n         self.outputs = params['outputs'].default if 'outputs' in params else []\n         self.inputs = params['inputs'].default if 'inputs' in params else []\n \ndiff --git a/parsl/app/python.py b/parsl/app/python.py\n--- a/parsl/app/python.py\n+++ b/parsl/app/python.py\n@@ -12,6 +12,27 @@\n logger = logging.getLogger(__name__)\n \n \n+def 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+\n class PythonApp(AppBase):\n     \"\"\"Extends AppBase to cover the Python App.\"\"\"\n \n@@ -45,6 +66,9 @@\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", "issue": "Implement time limits for python apps\nRequested by @lgray.\n", "before_files": [{"content": "\"\"\"Definitions for the @App decorator and the App classes.\n\nThe App class encapsulates a generic leaf task that can be executed asynchronously.\n\"\"\"\nimport logging\nfrom abc import ABCMeta, abstractmethod\nfrom inspect import getsource\nfrom hashlib import md5\nfrom inspect import signature\n\nfrom parsl.app.errors import InvalidAppTypeError\n\nlogger = logging.getLogger(__name__)\n\n\nclass AppBase(metaclass=ABCMeta):\n    \"\"\"This is the base class that defines the two external facing functions that an App must define.\n\n    The  __init__ () which is called when the interpreter sees the definition of the decorated\n    function, and the __call__ () which is invoked when a decorated function is called by the user.\n\n    \"\"\"\n\n    def __init__(self, func, data_flow_kernel=None, walltime=60, executors='all', cache=False):\n        \"\"\"Construct the App object.\n\n        Args:\n             - func (function): Takes the function to be made into an App\n\n        Kwargs:\n             - data_flow_kernel (DataFlowKernel): The :class:`~parsl.dataflow.dflow.DataFlowKernel` responsible for\n               managing this app. This can be omitted only\n               after calling :meth:`parsl.dataflow.dflow.DataFlowKernelLoader.load`.\n             - walltime (int) : Walltime in seconds for the app execution.\n             - executors (str|list) : Labels of the executors that this app can execute over. Default is 'all'.\n             - cache (Bool) : Enable caching of this app ?\n\n        Returns:\n             - App object.\n\n        \"\"\"\n        self.__name__ = func.__name__\n        self.func = func\n        self.data_flow_kernel = data_flow_kernel\n        self.status = 'created'\n        self.executors = executors\n        self.cache = cache\n        if not (isinstance(executors, list) or isinstance(executors, str)):\n            logger.error(\"App {} specifies invalid executor option, expects string or list\".format(\n                func.__name__))\n\n        if cache is True:\n            try:\n                self.fn_source = getsource(func)\n            except OSError:\n                logger.debug(\"Unable to get source code for AppCaching. Recommend creating module\")\n                self.fn_source = func.__name__\n\n            self.func_hash = md5(self.fn_source.encode('utf-8')).hexdigest()\n        else:\n            self.func_hash = func.__name__\n\n        params = signature(func).parameters\n\n        self.kwargs = {}\n        if 'stdout' in params:\n            self.kwargs['stdout'] = params['stdout'].default\n        if 'stderr' in params:\n            self.kwargs['stderr'] = params['stderr'].default\n        self.outputs = params['outputs'].default if 'outputs' in params else []\n        self.inputs = params['inputs'].default if 'inputs' in params else []\n\n    @abstractmethod\n    def __call__(self, *args, **kwargs):\n        pass\n\n\ndef App(apptype, data_flow_kernel=None, walltime=60, cache=False, executors='all'):\n    \"\"\"The App decorator function.\n\n    Args:\n        - apptype (string) : Apptype can be bash|python\n\n    Kwargs:\n        - data_flow_kernel (DataFlowKernel): The :class:`~parsl.dataflow.dflow.DataFlowKernel` responsible for\n          managing this app. This can be omitted only\n          after calling :meth:`parsl.dataflow.dflow.DataFlowKernelLoader.load`.\n        - walltime (int) : Walltime for app in seconds,\n             default=60\n        - executors (str|list) : Labels of the executors that this app can execute over. Default is 'all'.\n        - cache (Bool) : Enable caching of the app call\n             default=False\n\n    Returns:\n         A PythonApp or BashApp object, which when called runs the apps through the executor.\n    \"\"\"\n\n    from parsl.app.python import PythonApp\n    from parsl.app.bash import BashApp\n\n    logger.warning(\"The 'App' decorator will be deprecated in Parsl 0.8. Please use 'python_app' or 'bash_app' instead.\")\n\n    if apptype == 'python':\n        app_class = PythonApp\n    elif apptype == 'bash':\n        app_class = BashApp\n    else:\n        raise InvalidAppTypeError(\"Invalid apptype requested {}; must be 'python' or 'bash'\".format(apptype))\n\n    def wrapper(f):\n        return app_class(f,\n                         data_flow_kernel=data_flow_kernel,\n                         walltime=walltime,\n                         cache=cache,\n                         executors=executors)\n    return wrapper\n\n\ndef python_app(function=None, data_flow_kernel=None, walltime=60, cache=False, executors='all'):\n    \"\"\"Decorator function for making python apps.\n\n    Parameters\n    ----------\n    function : function\n        Do not pass this keyword argument directly. This is needed in order to allow for omitted parenthesis,\n        for example, `@python_app` if using all defaults or `@python_app(walltime=120)`. If the\n        decorator is used alone, function will be the actual function being decorated, whereas if it\n        is called with arguments, function will be None. Default is None.\n    data_flow_kernel : DataFlowKernel\n        The :class:`~parsl.dataflow.dflow.DataFlowKernel` responsible for managing this app. This can\n        be omitted only after calling :meth:`parsl.dataflow.dflow.DataFlowKernelLoader.load`. Default is None.\n    walltime : int\n        Walltime for app in seconds. Default is 60.\n    executors : string or list\n        Labels of the executors that this app can execute over. Default is 'all'.\n    cache : bool\n        Enable caching of the app call. Default is False.\n    \"\"\"\n    from parsl.app.python import PythonApp\n\n    def decorator(func):\n        def wrapper(f):\n            return PythonApp(f,\n                             data_flow_kernel=data_flow_kernel,\n                             walltime=walltime,\n                             cache=cache,\n                             executors=executors)\n        return wrapper(func)\n    if function is not None:\n        return decorator(function)\n    return decorator\n\n\ndef bash_app(function=None, data_flow_kernel=None, walltime=60, cache=False, executors='all'):\n    \"\"\"Decorator function for making bash apps.\n\n    Parameters\n    ----------\n    function : function\n        Do not pass this keyword argument directly. This is needed in order to allow for omitted parenthesis,\n        for example, `@bash_app` if using all defaults or `@bash_app(walltime=120)`. If the\n        decorator is used alone, function will be the actual function being decorated, whereas if it\n        is called with arguments, function will be None. Default is None.\n    data_flow_kernel : DataFlowKernel\n        The :class:`~parsl.dataflow.dflow.DataFlowKernel` responsible for managing this app. This can\n        be omitted only after calling :meth:`parsl.dataflow.dflow.DataFlowKernelLoader.load`. Default is None.\n    walltime : int\n        Walltime for app in seconds. Default is 60.\n    executors : string or list\n        Labels of the executors that this app can execute over. Default is 'all'.\n    cache : bool\n        Enable caching of the app call. Default is False.\n    \"\"\"\n    from parsl.app.bash import BashApp\n\n    def decorator(func):\n        def wrapper(f):\n            return BashApp(f,\n                           data_flow_kernel=data_flow_kernel,\n                           walltime=walltime,\n                           cache=cache,\n                           executors=executors)\n        return wrapper(func)\n    if function is not None:\n        return decorator(function)\n    return decorator\n", "path": "parsl/app/app.py"}, {"content": "import logging\n\nimport tblib.pickling_support\ntblib.pickling_support.install()\n\nfrom parsl.app.futures import DataFuture\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\nclass PythonApp(AppBase):\n    \"\"\"Extends AppBase to cover the Python App.\"\"\"\n\n    def __init__(self, func, data_flow_kernel=None, walltime=60, cache=False, executors='all'):\n        super().__init__(\n            wrap_error(func),\n            data_flow_kernel=data_flow_kernel,\n            walltime=walltime,\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             If outputs=[...] was a kwarg then:\n                   App_fut, [Data_Futures...]\n             else:\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        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        # logger.debug(\"App[{}] assigned Task[{}]\".format(self.func.__name__,\n        #                                                 app_fut.tid))\n        out_futs = [DataFuture(app_fut, o, tid=app_fut.tid)\n                    for o in kwargs.get('outputs', [])]\n        app_fut._outputs = out_futs\n\n        return app_fut\n", "path": "parsl/app/python.py"}], "after_files": [{"content": "\"\"\"Definitions for the @App decorator and the App classes.\n\nThe App class encapsulates a generic leaf task that can be executed asynchronously.\n\"\"\"\nimport logging\nfrom abc import ABCMeta, abstractmethod\nfrom inspect import getsource\nfrom hashlib import md5\nfrom inspect import signature\n\nfrom parsl.app.errors import InvalidAppTypeError\n\nlogger = logging.getLogger(__name__)\n\n\nclass AppBase(metaclass=ABCMeta):\n    \"\"\"This is the base class that defines the two external facing functions that an App must define.\n\n    The  __init__ () which is called when the interpreter sees the definition of the decorated\n    function, and the __call__ () which is invoked when a decorated function is called by the user.\n\n    \"\"\"\n\n    def __init__(self, func, data_flow_kernel=None, walltime=60, executors='all', cache=False):\n        \"\"\"Construct the App object.\n\n        Args:\n             - func (function): Takes the function to be made into an App\n\n        Kwargs:\n             - data_flow_kernel (DataFlowKernel): The :class:`~parsl.dataflow.dflow.DataFlowKernel` responsible for\n               managing this app. This can be omitted only\n               after calling :meth:`parsl.dataflow.dflow.DataFlowKernelLoader.load`.\n             - walltime (int) : Walltime in seconds for the app execution.\n             - executors (str|list) : Labels of the executors that this app can execute over. Default is 'all'.\n             - cache (Bool) : Enable caching of this app ?\n\n        Returns:\n             - App object.\n\n        \"\"\"\n        self.__name__ = func.__name__\n        self.func = func\n        self.data_flow_kernel = data_flow_kernel\n        self.status = 'created'\n        self.executors = executors\n        self.cache = cache\n        if not (isinstance(executors, list) or isinstance(executors, str)):\n            logger.error(\"App {} specifies invalid executor option, expects string or list\".format(\n                func.__name__))\n\n        if cache is True:\n            try:\n                self.fn_source = getsource(func)\n            except OSError:\n                logger.debug(\"Unable to get source code for AppCaching. Recommend creating module\")\n                self.fn_source = func.__name__\n\n            self.func_hash = md5(self.fn_source.encode('utf-8')).hexdigest()\n        else:\n            self.func_hash = func.__name__\n\n        params = signature(func).parameters\n\n        self.kwargs = {}\n        if 'stdout' in params:\n            self.kwargs['stdout'] = params['stdout'].default\n        if 'stderr' in params:\n            self.kwargs['stderr'] = params['stderr'].default\n        if 'walltime' in params:\n            self.kwargs['walltime'] = params['walltime'].default\n        self.outputs = params['outputs'].default if 'outputs' in params else []\n        self.inputs = params['inputs'].default if 'inputs' in params else []\n\n    @abstractmethod\n    def __call__(self, *args, **kwargs):\n        pass\n\n\ndef App(apptype, data_flow_kernel=None, walltime=60, cache=False, executors='all'):\n    \"\"\"The App decorator function.\n\n    Args:\n        - apptype (string) : Apptype can be bash|python\n\n    Kwargs:\n        - data_flow_kernel (DataFlowKernel): The :class:`~parsl.dataflow.dflow.DataFlowKernel` responsible for\n          managing this app. This can be omitted only\n          after calling :meth:`parsl.dataflow.dflow.DataFlowKernelLoader.load`.\n        - walltime (int) : Walltime for app in seconds,\n             default=60\n        - executors (str|list) : Labels of the executors that this app can execute over. Default is 'all'.\n        - cache (Bool) : Enable caching of the app call\n             default=False\n\n    Returns:\n         A PythonApp or BashApp object, which when called runs the apps through the executor.\n    \"\"\"\n\n    from parsl.app.python import PythonApp\n    from parsl.app.bash import BashApp\n\n    logger.warning(\"The 'App' decorator will be deprecated in Parsl 0.8. Please use 'python_app' or 'bash_app' instead.\")\n\n    if apptype == 'python':\n        app_class = PythonApp\n    elif apptype == 'bash':\n        app_class = BashApp\n    else:\n        raise InvalidAppTypeError(\"Invalid apptype requested {}; must be 'python' or 'bash'\".format(apptype))\n\n    def wrapper(f):\n        return app_class(f,\n                         data_flow_kernel=data_flow_kernel,\n                         walltime=walltime,\n                         cache=cache,\n                         executors=executors)\n    return wrapper\n\n\ndef python_app(function=None, data_flow_kernel=None, walltime=60, cache=False, executors='all'):\n    \"\"\"Decorator function for making python apps.\n\n    Parameters\n    ----------\n    function : function\n        Do not pass this keyword argument directly. This is needed in order to allow for omitted parenthesis,\n        for example, `@python_app` if using all defaults or `@python_app(walltime=120)`. If the\n        decorator is used alone, function will be the actual function being decorated, whereas if it\n        is called with arguments, function will be None. Default is None.\n    data_flow_kernel : DataFlowKernel\n        The :class:`~parsl.dataflow.dflow.DataFlowKernel` responsible for managing this app. This can\n        be omitted only after calling :meth:`parsl.dataflow.dflow.DataFlowKernelLoader.load`. Default is None.\n    walltime : int\n        Walltime for app in seconds. Default is 60.\n    executors : string or list\n        Labels of the executors that this app can execute over. Default is 'all'.\n    cache : bool\n        Enable caching of the app call. Default is False.\n    \"\"\"\n    from parsl.app.python import PythonApp\n\n    def decorator(func):\n        def wrapper(f):\n            return PythonApp(f,\n                             data_flow_kernel=data_flow_kernel,\n                             walltime=walltime,\n                             cache=cache,\n                             executors=executors)\n        return wrapper(func)\n    if function is not None:\n        return decorator(function)\n    return decorator\n\n\ndef bash_app(function=None, data_flow_kernel=None, walltime=60, cache=False, executors='all'):\n    \"\"\"Decorator function for making bash apps.\n\n    Parameters\n    ----------\n    function : function\n        Do not pass this keyword argument directly. This is needed in order to allow for omitted parenthesis,\n        for example, `@bash_app` if using all defaults or `@bash_app(walltime=120)`. If the\n        decorator is used alone, function will be the actual function being decorated, whereas if it\n        is called with arguments, function will be None. Default is None.\n    data_flow_kernel : DataFlowKernel\n        The :class:`~parsl.dataflow.dflow.DataFlowKernel` responsible for managing this app. This can\n        be omitted only after calling :meth:`parsl.dataflow.dflow.DataFlowKernelLoader.load`. Default is None.\n    walltime : int\n        Walltime for app in seconds. Default is 60.\n    executors : string or list\n        Labels of the executors that this app can execute over. Default is 'all'.\n    cache : bool\n        Enable caching of the app call. Default is False.\n    \"\"\"\n    from parsl.app.bash import BashApp\n\n    def decorator(func):\n        def wrapper(f):\n            return BashApp(f,\n                           data_flow_kernel=data_flow_kernel,\n                           walltime=walltime,\n                           cache=cache,\n                           executors=executors)\n        return wrapper(func)\n    if function is not None:\n        return decorator(function)\n    return decorator\n", "path": "parsl/app/app.py"}, {"content": "import logging\n\nimport tblib.pickling_support\ntblib.pickling_support.install()\n\nfrom parsl.app.futures import DataFuture\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, walltime=60, cache=False, executors='all'):\n        super().__init__(\n            wrap_error(func),\n            data_flow_kernel=data_flow_kernel,\n            walltime=walltime,\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             If outputs=[...] was a kwarg then:\n                   App_fut, [Data_Futures...]\n             else:\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        # logger.debug(\"App[{}] assigned Task[{}]\".format(self.func.__name__,\n        #                                                 app_fut.tid))\n        out_futs = [DataFuture(app_fut, o, tid=app_fut.tid)\n                    for o in kwargs.get('outputs', [])]\n        app_fut._outputs = out_futs\n\n        return app_fut\n", "path": "parsl/app/python.py"}]}