We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
remote_side_bash_executor global logger conflates logs from multiple tasks/sources
`remote_side_bash_executor` configures and uses the global `logging` logger, rather than one scoped to that function. Once it has configured a log file for output, all subsequent global logs from any further `remote_side_bash_executor` invocation in that process, as well as other uses of logging, such as htex `process_worker_pool.py`, end up in earlier configured log files.

This results in `/tmp/bashexec` logs containing a confused assortment of logs from different sources, rather than being focuses on a single bash execution.

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

{"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@@ -1,4 +1,3 @@\n-import logging\n from functools import update_wrapper\n from inspect import signature, Parameter\n \n@@ -7,8 +6,6 @@\n from parsl.app.app import AppBase\n from parsl.dataflow.dflow import DataFlowKernelLoader\n \n-logger = logging.getLogger(__name__)\n-\n \n def remote_side_bash_executor(func, *args, **kwargs):\n     \"\"\"Execute the bash app type function and return the command line string.\n@@ -21,8 +18,20 @@\n     import subprocess\n     import logging\n     import parsl.app.errors as pe\n+    from parsl import set_file_logger\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-    logging.basicConfig(filename='/tmp/bashexec.{0}.log'.format(time.time()), level=logging.DEBUG)\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@@ -44,10 +53,10 @@\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-        logging.error(\"Caught exception during formatting of app '{}': {}\".format(func_name, e))\n+        logger.error(\"Caught exception during formatting of app '{}': {}\".format(func_name, e))\n         raise e\n \n-    logging.debug(\"Executable: %s\", executable)\n+    logger.debug(\"Executable: %s\", executable)\n \n     # Updating stdout, stderr if values passed at call time.\n", "issue": "remote_side_bash_executor global logger conflates logs from multiple tasks/sources\n`remote_side_bash_executor` configures and uses the global `logging` logger, rather than one scoped to that function. Once it has configured a log file for output, all subsequent global logs from any further `remote_side_bash_executor` invocation in that process, as well as other uses of logging, such as htex `process_worker_pool.py`, end up in earlier configured log files.\r\n\r\nThis results in `/tmp/bashexec` logs containing a confused assortment of logs from different sources, rather than being focuses on a single bash execution.\n", "before_files": [{"content": "import logging\nfrom functools import update_wrapper\nfrom inspect import signature, Parameter\n\nfrom parsl.app.errors import wrap_error\nfrom parsl.app.futures import DataFuture\nfrom parsl.app.app import AppBase\nfrom parsl.dataflow.dflow import DataFlowKernelLoader\n\nlogger = logging.getLogger(__name__)\n\n\ndef remote_side_bash_executor(func, *args, **kwargs):\n    \"\"\"Execute the bash app type function and return the command line string.\n\n    This string is reformatted with the *args, and **kwargs\n    from call time.\n    \"\"\"\n    import os\n    import time\n    import subprocess\n    import logging\n    import parsl.app.errors as pe\n\n    logging.basicConfig(filename='/tmp/bashexec.{0}.log'.format(time.time()), level=logging.DEBUG)\n\n    func_name = func.__name__\n\n    partial_cmdline = None\n\n    # Try to run the func to compose the commandline\n    try:\n        # Execute the func to get the commandline\n        partial_cmdline = func(*args, **kwargs)\n        # Reformat the commandline with current args and kwargs\n        executable = partial_cmdline.format(*args, **kwargs)\n\n    except AttributeError as e:\n        if partial_cmdline 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), None)\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        logging.error(\"Caught exception during formatting of app '{}': {}\".format(func_name, e))\n        raise e\n\n    logging.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        elif isinstance(stdfspec, str):\n            fname = stdfspec\n            mode = 'a+'\n        elif isinstance(stdfspec, tuple):\n            if len(stdfspec) != 2:\n                raise pe.BadStdStreamFile(\"std descriptor %s has incorrect tuple length %s\" % (fdname, len(stdfspec)), TypeError('Bad Tuple Length'))\n            fname, mode = stdfspec\n        else:\n            raise pe.BadStdStreamFile(\"std descriptor %s has unexpected type %s\" % (fdname, str(type(stdfspec))), TypeError('Bad Tuple Type'))\n\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: {}\".format(func_name, proc.returncode), e)\n\n    if returncode != 0:\n        raise pe.AppFailure(\"[{}] App failed with exit code: {}\".format(func_name, proc.returncode), proc.returncode)\n\n    # TODO : Add support for globs here\n\n    missing = []\n    for outputfile in kwargs.get('outputs', []):\n        fpath = outputfile\n        if type(outputfile) != str:\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, walltime=60, cache=False, executors='all'):\n        super().__init__(func, data_flow_kernel=data_flow_kernel, walltime=60, executors=executors, cache=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 != Parameter.empty:\n                self.kwargs[s] = sig.parameters[s].default\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             If outputs=[...] was a kwarg then:\n                   App_fut, [Data_Futures...]\n             else:\n                   App_fut\n\n        \"\"\"\n        # Update kwargs in the app definition with ones passed in at calltime\n        self.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(wrap_error(update_wrapper(remote_side_bash_executor, self.func)),\n                             self.func, *args,\n                             executors=self.executors,\n                             fn_hash=self.func_hash,\n                             cache=self.cache,\n                             **self.kwargs)\n\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/bash.py"}], "after_files": [{"content": "from functools import update_wrapper\nfrom inspect import signature, Parameter\n\nfrom parsl.app.errors import wrap_error\nfrom parsl.app.futures import DataFuture\nfrom parsl.app.app import AppBase\nfrom parsl.dataflow.dflow import DataFlowKernelLoader\n\n\ndef remote_side_bash_executor(func, *args, **kwargs):\n    \"\"\"Execute the bash app type function and return the command line string.\n\n    This string is reformatted with the *args, and **kwargs\n    from call time.\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\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    partial_cmdline = None\n\n    # Try to run the func to compose the commandline\n    try:\n        # Execute the func to get the commandline\n        partial_cmdline = func(*args, **kwargs)\n        # Reformat the commandline with current args and kwargs\n        executable = partial_cmdline.format(*args, **kwargs)\n\n    except AttributeError as e:\n        if partial_cmdline 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), None)\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        elif isinstance(stdfspec, str):\n            fname = stdfspec\n            mode = 'a+'\n        elif isinstance(stdfspec, tuple):\n            if len(stdfspec) != 2:\n                raise pe.BadStdStreamFile(\"std descriptor %s has incorrect tuple length %s\" % (fdname, len(stdfspec)), TypeError('Bad Tuple Length'))\n            fname, mode = stdfspec\n        else:\n            raise pe.BadStdStreamFile(\"std descriptor %s has unexpected type %s\" % (fdname, str(type(stdfspec))), TypeError('Bad Tuple Type'))\n\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: {}\".format(func_name, proc.returncode), e)\n\n    if returncode != 0:\n        raise pe.AppFailure(\"[{}] App failed with exit code: {}\".format(func_name, proc.returncode), proc.returncode)\n\n    # TODO : Add support for globs here\n\n    missing = []\n    for outputfile in kwargs.get('outputs', []):\n        fpath = outputfile\n        if type(outputfile) != str:\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, walltime=60, cache=False, executors='all'):\n        super().__init__(func, data_flow_kernel=data_flow_kernel, walltime=60, executors=executors, cache=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 != Parameter.empty:\n                self.kwargs[s] = sig.parameters[s].default\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             If outputs=[...] was a kwarg then:\n                   App_fut, [Data_Futures...]\n             else:\n                   App_fut\n\n        \"\"\"\n        # Update kwargs in the app definition with ones passed in at calltime\n        self.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(wrap_error(update_wrapper(remote_side_bash_executor, self.func)),\n                             self.func, *args,\n                             executors=self.executors,\n                             fn_hash=self.func_hash,\n                             cache=self.cache,\n                             **self.kwargs)\n\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/bash.py"}]}