We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Remote function name is not resolved correctly | task_func_name: wrapper
**Describe the bug**
The stdout and stderr log filenames of Parsl apps should contain the function name of the corresponding app (Parsl 1.0.0 behavior). In master branch version (ec09fdc), the function name is appended as 'wrapper'. This unexpected behavior is only noticed in bash_apps. It works as expected for python_apps.

**To Reproduce**
Steps to reproduce the behavior, for e.g:
1. Setup Parsl latest (ec09fdc) with Python 3.7 on cluster
2. Run a test script with a bash_app named "my_bash_app"
3. Wait until tasks are finished
4. See generated Parsl stdout, stderr log files.
5. Check parsl.log for 'task_func_name': 'my_bash_app'. However, 'task_func_name': 'wrapper' is seen

**Expected behavior**
1. 'task_func_name': 'my_bash_app' in Parsl log
2. "my_bash_app" string contained in stdout and stderr log filenames.

**Environment**
 - OS: RHEL6.1
 - Python version: 3.7 (Anaconda 4.8.2)
 - Parsl version: ec09fdc2099e1795f41210bd1a159633b601cee3

**Distributed Environment**
- PBS Pro in NSCC's ASPIRE1 (HTEX)


--- 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/bash.py`
Content:
```
1 from functools import update_wrapper
2 from functools import partial
3 from inspect import signature, Parameter
4 
5 from parsl.app.errors import wrap_error
6 from parsl.app.app import AppBase
7 from parsl.dataflow.dflow import DataFlowKernelLoader
8 
9 
10 def remote_side_bash_executor(func, *args, **kwargs):
11     """Executes the supplied function with *args and **kwargs to get a
12     command-line to run, and then run that command-line using bash.
13     """
14     import os
15     import time
16     import subprocess
17     import logging
18     import parsl.app.errors as pe
19     from parsl import set_file_logger
20     from parsl.utils import get_std_fname_mode
21 
22     logbase = "/tmp"
23     format_string = "%(asctime)s.%(msecs)03d %(name)s:%(lineno)d [%(levelname)s]  %(message)s"
24 
25     # make this name unique per invocation so that each invocation can
26     # log to its own file. It would be better to include the task_id here
27     # but that is awkward to wire through at the moment as apps do not
28     # have access to that execution context.
29     t = time.time()
30 
31     logname = __name__ + "." + str(t)
32     logger = logging.getLogger(logname)
33     set_file_logger(filename='{0}/bashexec.{1}.log'.format(logbase, t), name=logname, level=logging.DEBUG, format_string=format_string)
34 
35     func_name = func.__name__
36 
37     executable = None
38 
39     # Try to run the func to compose the commandline
40     try:
41         # Execute the func to get the commandline
42         executable = func(*args, **kwargs)
43 
44     except AttributeError as e:
45         if executable is not None:
46             raise pe.AppBadFormatting("App formatting failed for app '{}' with AttributeError: {}".format(func_name, e))
47         else:
48             raise pe.BashAppNoReturn("Bash app '{}' did not return a value, or returned None - with this exception: {}".format(func_name, e))
49 
50     except IndexError as e:
51         raise pe.AppBadFormatting("App formatting failed for app '{}' with IndexError: {}".format(func_name, e))
52     except Exception as e:
53         logger.error("Caught exception during formatting of app '{}': {}".format(func_name, e))
54         raise e
55 
56     logger.debug("Executable: %s", executable)
57 
58     # Updating stdout, stderr if values passed at call time.
59 
60     def open_std_fd(fdname):
61         # fdname is 'stdout' or 'stderr'
62         stdfspec = kwargs.get(fdname)  # spec is str name or tuple (name, mode)
63         if stdfspec is None:
64             return None
65 
66         fname, mode = get_std_fname_mode(fdname, stdfspec)
67         try:
68             if os.path.dirname(fname):
69                 os.makedirs(os.path.dirname(fname), exist_ok=True)
70             fd = open(fname, mode)
71         except Exception as e:
72             raise pe.BadStdStreamFile(fname, e)
73         return fd
74 
75     std_out = open_std_fd('stdout')
76     std_err = open_std_fd('stderr')
77     timeout = kwargs.get('walltime')
78 
79     if std_err is not None:
80         print('--> executable follows <--\n{}\n--> end executable <--'.format(executable), file=std_err, flush=True)
81 
82     returncode = None
83     try:
84         proc = subprocess.Popen(executable, stdout=std_out, stderr=std_err, shell=True, executable='/bin/bash')
85         proc.wait(timeout=timeout)
86         returncode = proc.returncode
87 
88     except subprocess.TimeoutExpired:
89         raise pe.AppTimeout("[{}] App exceeded walltime: {}".format(func_name, timeout))
90 
91     except Exception as e:
92         raise pe.AppException("[{}] App caught exception with returncode: {}".format(func_name, returncode), e)
93 
94     if returncode != 0:
95         raise pe.BashExitFailure(func_name, proc.returncode)
96 
97     # TODO : Add support for globs here
98 
99     missing = []
100     for outputfile in kwargs.get('outputs', []):
101         fpath = outputfile.filepath
102 
103         if not os.path.exists(fpath):
104             missing.extend([outputfile])
105 
106     if missing:
107         raise pe.MissingOutputs("[{}] Missing outputs".format(func_name), missing)
108 
109     return returncode
110 
111 
112 class BashApp(AppBase):
113 
114     def __init__(self, func, data_flow_kernel=None, cache=False, executors='all', ignore_for_cache=None):
115         super().__init__(func, data_flow_kernel=data_flow_kernel, executors=executors, cache=cache, ignore_for_cache=ignore_for_cache)
116         self.kwargs = {}
117 
118         # We duplicate the extraction of parameter defaults
119         # to self.kwargs to ensure availability at point of
120         # command string format. Refer: #349
121         sig = signature(func)
122 
123         for s in sig.parameters:
124             if sig.parameters[s].default is not Parameter.empty:
125                 self.kwargs[s] = sig.parameters[s].default
126 
127         # update_wrapper allows remote_side_bash_executor to masquerade as self.func
128         # partial is used to attach the first arg the "func" to the remote_side_bash_executor
129         # this is done to avoid passing a function type in the args which parsl.serializer
130         # doesn't support
131         remote_fn = partial(update_wrapper(remote_side_bash_executor, self.func), self.func)
132         self.wrapped_remote_function = wrap_error(remote_fn)
133 
134     def __call__(self, *args, **kwargs):
135         """Handle the call to a Bash app.
136 
137         Args:
138              - Arbitrary
139 
140         Kwargs:
141              - Arbitrary
142 
143         Returns:
144                    App_fut
145 
146         """
147         invocation_kwargs = {}
148         invocation_kwargs.update(self.kwargs)
149         invocation_kwargs.update(kwargs)
150 
151         if self.data_flow_kernel is None:
152             dfk = DataFlowKernelLoader.dfk()
153         else:
154             dfk = self.data_flow_kernel
155 
156         app_fut = dfk.submit(self.wrapped_remote_function,
157                              app_args=args,
158                              executors=self.executors,
159                              fn_hash=self.func_hash,
160                              cache=self.cache,
161                              ignore_for_cache=self.ignore_for_cache,
162                              app_kwargs=invocation_kwargs)
163 
164         return app_fut
165
```
--- 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/bash.py b/parsl/app/bash.py
--- a/parsl/app/bash.py
+++ b/parsl/app/bash.py
@@ -129,6 +129,7 @@
         # this is done to avoid passing a function type in the args which parsl.serializer
         # doesn't support
         remote_fn = partial(update_wrapper(remote_side_bash_executor, self.func), self.func)
+        remote_fn.__name__ = self.func.__name__
         self.wrapped_remote_function = wrap_error(remote_fn)
 
     def __call__(self, *args, **kwargs):

{"golden_diff": "diff --git a/parsl/app/bash.py b/parsl/app/bash.py\n--- a/parsl/app/bash.py\n+++ b/parsl/app/bash.py\n@@ -129,6 +129,7 @@\n         # this is done to avoid passing a function type in the args which parsl.serializer\n         # doesn't support\n         remote_fn = partial(update_wrapper(remote_side_bash_executor, self.func), self.func)\n+        remote_fn.__name__ = self.func.__name__\n         self.wrapped_remote_function = wrap_error(remote_fn)\n \n     def __call__(self, *args, **kwargs):\n", "issue": "Remote function name is not resolved correctly | task_func_name: wrapper\n**Describe the bug**\r\nThe stdout and stderr log filenames of Parsl apps should contain the function name of the corresponding app (Parsl 1.0.0 behavior). In master branch version (ec09fdc), the function name is appended as 'wrapper'. This unexpected behavior is only noticed in bash_apps. It works as expected for python_apps.\r\n\r\n**To Reproduce**\r\nSteps to reproduce the behavior, for e.g:\r\n1. Setup Parsl latest (ec09fdc) with Python 3.7 on cluster\r\n2. Run a test script with a bash_app named \"my_bash_app\"\r\n3. Wait until tasks are finished\r\n4. See generated Parsl stdout, stderr log files.\r\n5. Check parsl.log for 'task_func_name': 'my_bash_app'. However, 'task_func_name': 'wrapper' is seen\r\n\r\n**Expected behavior**\r\n1. 'task_func_name': 'my_bash_app' in Parsl log\r\n2. \"my_bash_app\" string contained in stdout and stderr log filenames.\r\n\r\n**Environment**\r\n - OS: RHEL6.1\r\n - Python version: 3.7 (Anaconda 4.8.2)\r\n - Parsl version: ec09fdc2099e1795f41210bd1a159633b601cee3\r\n\r\n**Distributed Environment**\r\n- PBS Pro in NSCC's ASPIRE1 (HTEX)\r\n\n", "before_files": [{"content": "from functools import update_wrapper\nfrom functools import partial\nfrom inspect import signature, Parameter\n\nfrom parsl.app.errors import wrap_error\nfrom parsl.app.app import AppBase\nfrom parsl.dataflow.dflow import DataFlowKernelLoader\n\n\ndef remote_side_bash_executor(func, *args, **kwargs):\n    \"\"\"Executes the supplied function with *args and **kwargs to get a\n    command-line to run, and then run that command-line using bash.\n    \"\"\"\n    import os\n    import time\n    import subprocess\n    import logging\n    import parsl.app.errors as pe\n    from parsl import set_file_logger\n    from parsl.utils import get_std_fname_mode\n\n    logbase = \"/tmp\"\n    format_string = \"%(asctime)s.%(msecs)03d %(name)s:%(lineno)d [%(levelname)s]  %(message)s\"\n\n    # make this name unique per invocation so that each invocation can\n    # log to its own file. It would be better to include the task_id here\n    # but that is awkward to wire through at the moment as apps do not\n    # have access to that execution context.\n    t = time.time()\n\n    logname = __name__ + \".\" + str(t)\n    logger = logging.getLogger(logname)\n    set_file_logger(filename='{0}/bashexec.{1}.log'.format(logbase, t), name=logname, level=logging.DEBUG, format_string=format_string)\n\n    func_name = func.__name__\n\n    executable = None\n\n    # Try to run the func to compose the commandline\n    try:\n        # Execute the func to get the commandline\n        executable = func(*args, **kwargs)\n\n    except AttributeError as e:\n        if executable is not None:\n            raise pe.AppBadFormatting(\"App formatting failed for app '{}' with AttributeError: {}\".format(func_name, e))\n        else:\n            raise pe.BashAppNoReturn(\"Bash app '{}' did not return a value, or returned None - with this exception: {}\".format(func_name, e))\n\n    except IndexError as e:\n        raise pe.AppBadFormatting(\"App formatting failed for app '{}' with IndexError: {}\".format(func_name, e))\n    except Exception as e:\n        logger.error(\"Caught exception during formatting of app '{}': {}\".format(func_name, e))\n        raise e\n\n    logger.debug(\"Executable: %s\", executable)\n\n    # Updating stdout, stderr if values passed at call time.\n\n    def open_std_fd(fdname):\n        # fdname is 'stdout' or 'stderr'\n        stdfspec = kwargs.get(fdname)  # spec is str name or tuple (name, mode)\n        if stdfspec is None:\n            return None\n\n        fname, mode = get_std_fname_mode(fdname, stdfspec)\n        try:\n            if os.path.dirname(fname):\n                os.makedirs(os.path.dirname(fname), exist_ok=True)\n            fd = open(fname, mode)\n        except Exception as e:\n            raise pe.BadStdStreamFile(fname, e)\n        return fd\n\n    std_out = open_std_fd('stdout')\n    std_err = open_std_fd('stderr')\n    timeout = kwargs.get('walltime')\n\n    if std_err is not None:\n        print('--> executable follows <--\\n{}\\n--> end executable <--'.format(executable), file=std_err, flush=True)\n\n    returncode = None\n    try:\n        proc = subprocess.Popen(executable, stdout=std_out, stderr=std_err, shell=True, executable='/bin/bash')\n        proc.wait(timeout=timeout)\n        returncode = proc.returncode\n\n    except subprocess.TimeoutExpired:\n        raise pe.AppTimeout(\"[{}] App exceeded walltime: {}\".format(func_name, timeout))\n\n    except Exception as e:\n        raise pe.AppException(\"[{}] App caught exception with returncode: {}\".format(func_name, returncode), e)\n\n    if returncode != 0:\n        raise pe.BashExitFailure(func_name, proc.returncode)\n\n    # TODO : Add support for globs here\n\n    missing = []\n    for outputfile in kwargs.get('outputs', []):\n        fpath = outputfile.filepath\n\n        if not os.path.exists(fpath):\n            missing.extend([outputfile])\n\n    if missing:\n        raise pe.MissingOutputs(\"[{}] Missing outputs\".format(func_name), missing)\n\n    return returncode\n\n\nclass BashApp(AppBase):\n\n    def __init__(self, func, data_flow_kernel=None, cache=False, executors='all', ignore_for_cache=None):\n        super().__init__(func, data_flow_kernel=data_flow_kernel, executors=executors, cache=cache, ignore_for_cache=ignore_for_cache)\n        self.kwargs = {}\n\n        # We duplicate the extraction of parameter defaults\n        # to self.kwargs to ensure availability at point of\n        # command string format. Refer: #349\n        sig = signature(func)\n\n        for s in sig.parameters:\n            if sig.parameters[s].default is not Parameter.empty:\n                self.kwargs[s] = sig.parameters[s].default\n\n        # update_wrapper allows remote_side_bash_executor to masquerade as self.func\n        # partial is used to attach the first arg the \"func\" to the remote_side_bash_executor\n        # this is done to avoid passing a function type in the args which parsl.serializer\n        # doesn't support\n        remote_fn = partial(update_wrapper(remote_side_bash_executor, self.func), self.func)\n        self.wrapped_remote_function = wrap_error(remote_fn)\n\n    def __call__(self, *args, **kwargs):\n        \"\"\"Handle the call to a Bash app.\n\n        Args:\n             - Arbitrary\n\n        Kwargs:\n             - Arbitrary\n\n        Returns:\n                   App_fut\n\n        \"\"\"\n        invocation_kwargs = {}\n        invocation_kwargs.update(self.kwargs)\n        invocation_kwargs.update(kwargs)\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.wrapped_remote_function,\n                             app_args=args,\n                             executors=self.executors,\n                             fn_hash=self.func_hash,\n                             cache=self.cache,\n                             ignore_for_cache=self.ignore_for_cache,\n                             app_kwargs=invocation_kwargs)\n\n        return app_fut\n", "path": "parsl/app/bash.py"}], "after_files": [{"content": "from functools import update_wrapper\nfrom functools import partial\nfrom inspect import signature, Parameter\n\nfrom parsl.app.errors import wrap_error\nfrom parsl.app.app import AppBase\nfrom parsl.dataflow.dflow import DataFlowKernelLoader\n\n\ndef remote_side_bash_executor(func, *args, **kwargs):\n    \"\"\"Executes the supplied function with *args and **kwargs to get a\n    command-line to run, and then run that command-line using bash.\n    \"\"\"\n    import os\n    import time\n    import subprocess\n    import logging\n    import parsl.app.errors as pe\n    from parsl import set_file_logger\n    from parsl.utils import get_std_fname_mode\n\n    logbase = \"/tmp\"\n    format_string = \"%(asctime)s.%(msecs)03d %(name)s:%(lineno)d [%(levelname)s]  %(message)s\"\n\n    # make this name unique per invocation so that each invocation can\n    # log to its own file. It would be better to include the task_id here\n    # but that is awkward to wire through at the moment as apps do not\n    # have access to that execution context.\n    t = time.time()\n\n    logname = __name__ + \".\" + str(t)\n    logger = logging.getLogger(logname)\n    set_file_logger(filename='{0}/bashexec.{1}.log'.format(logbase, t), name=logname, level=logging.DEBUG, format_string=format_string)\n\n    func_name = func.__name__\n\n    executable = None\n\n    # Try to run the func to compose the commandline\n    try:\n        # Execute the func to get the commandline\n        executable = func(*args, **kwargs)\n\n    except AttributeError as e:\n        if executable is not None:\n            raise pe.AppBadFormatting(\"App formatting failed for app '{}' with AttributeError: {}\".format(func_name, e))\n        else:\n            raise pe.BashAppNoReturn(\"Bash app '{}' did not return a value, or returned None - with this exception: {}\".format(func_name, e))\n\n    except IndexError as e:\n        raise pe.AppBadFormatting(\"App formatting failed for app '{}' with IndexError: {}\".format(func_name, e))\n    except Exception as e:\n        logger.error(\"Caught exception during formatting of app '{}': {}\".format(func_name, e))\n        raise e\n\n    logger.debug(\"Executable: %s\", executable)\n\n    # Updating stdout, stderr if values passed at call time.\n\n    def open_std_fd(fdname):\n        # fdname is 'stdout' or 'stderr'\n        stdfspec = kwargs.get(fdname)  # spec is str name or tuple (name, mode)\n        if stdfspec is None:\n            return None\n\n        fname, mode = get_std_fname_mode(fdname, stdfspec)\n        try:\n            if os.path.dirname(fname):\n                os.makedirs(os.path.dirname(fname), exist_ok=True)\n            fd = open(fname, mode)\n        except Exception as e:\n            raise pe.BadStdStreamFile(fname, e)\n        return fd\n\n    std_out = open_std_fd('stdout')\n    std_err = open_std_fd('stderr')\n    timeout = kwargs.get('walltime')\n\n    if std_err is not None:\n        print('--> executable follows <--\\n{}\\n--> end executable <--'.format(executable), file=std_err, flush=True)\n\n    returncode = None\n    try:\n        proc = subprocess.Popen(executable, stdout=std_out, stderr=std_err, shell=True, executable='/bin/bash')\n        proc.wait(timeout=timeout)\n        returncode = proc.returncode\n\n    except subprocess.TimeoutExpired:\n        raise pe.AppTimeout(\"[{}] App exceeded walltime: {}\".format(func_name, timeout))\n\n    except Exception as e:\n        raise pe.AppException(\"[{}] App caught exception with returncode: {}\".format(func_name, returncode), e)\n\n    if returncode != 0:\n        raise pe.BashExitFailure(func_name, proc.returncode)\n\n    # TODO : Add support for globs here\n\n    missing = []\n    for outputfile in kwargs.get('outputs', []):\n        fpath = outputfile.filepath\n\n        if not os.path.exists(fpath):\n            missing.extend([outputfile])\n\n    if missing:\n        raise pe.MissingOutputs(\"[{}] Missing outputs\".format(func_name), missing)\n\n    return returncode\n\n\nclass BashApp(AppBase):\n\n    def __init__(self, func, data_flow_kernel=None, cache=False, executors='all', ignore_for_cache=None):\n        super().__init__(func, data_flow_kernel=data_flow_kernel, executors=executors, cache=cache, ignore_for_cache=ignore_for_cache)\n        self.kwargs = {}\n\n        # We duplicate the extraction of parameter defaults\n        # to self.kwargs to ensure availability at point of\n        # command string format. Refer: #349\n        sig = signature(func)\n\n        for s in sig.parameters:\n            if sig.parameters[s].default is not Parameter.empty:\n                self.kwargs[s] = sig.parameters[s].default\n\n        # update_wrapper allows remote_side_bash_executor to masquerade as self.func\n        # partial is used to attach the first arg the \"func\" to the remote_side_bash_executor\n        # this is done to avoid passing a function type in the args which parsl.serializer\n        # doesn't support\n        remote_fn = partial(update_wrapper(remote_side_bash_executor, self.func), self.func)\n        remote_fn.__name__ = self.func.__name__\n        self.wrapped_remote_function = wrap_error(remote_fn)\n\n    def __call__(self, *args, **kwargs):\n        \"\"\"Handle the call to a Bash app.\n\n        Args:\n             - Arbitrary\n\n        Kwargs:\n             - Arbitrary\n\n        Returns:\n                   App_fut\n\n        \"\"\"\n        invocation_kwargs = {}\n        invocation_kwargs.update(self.kwargs)\n        invocation_kwargs.update(kwargs)\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.wrapped_remote_function,\n                             app_args=args,\n                             executors=self.executors,\n                             fn_hash=self.func_hash,\n                             cache=self.cache,\n                             ignore_for_cache=self.ignore_for_cache,\n                             app_kwargs=invocation_kwargs)\n\n        return app_fut\n", "path": "parsl/app/bash.py"}]}