We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[tune] Ray Tune fails to parse typing hints of the function for experiment
### What is the problem?
If the function for experiment has a [typing hint](https://docs.python.org/3/library/typing.html) for its argument `config`, then Ray Tune fails to parse the argument and assumes that there is a reporter signature. The cause of this problem is in this source code:
https://github.com/ray-project/ray/blob/1eaa57c98f8870a43e1ea14ec011b6bd4be97c8d/python/ray/tune/function_runner.py#L250-L257
Changing 
`func_args = inspect.getargspec(train_func).args` 
to
`func_args = inspect.getfullargspec(train_func).args` 
might solve the problem.

*Ray version and other system information (Python version, TensorFlow version, OS):*
Ray: 0.8.0
Python: 3.7.5
OS: Ubuntu 18.04

*Does the problem occur on the [latest wheels](https://ray.readthedocs.io/en/latest/installation.html)?*
I couldn't install the latest wheel. So I can't confirm it.

### Reproduction
The following is a modification of the first examples in the [Ray Tune Documentation](https://ray.readthedocs.io/en/latest/tune.html#quick-start), where I added a typing hint  `config: Dict[str, Any]` for the argument of function `train_mnist`.

```
from typing import Dict, Any

import torch.optim as optim
from ray import tune
from ray.tune.examples.mnist_pytorch import get_data_loaders, ConvNet, train, test


def train_mnist(config: Dict[str, Any]):
    train_loader, test_loader = get_data_loaders()
    model = ConvNet()
    optimizer = optim.SGD(model.parameters(), lr=config["lr"])
    for i in range(10):
        train(model, optimizer, train_loader)
        acc = test(model, test_loader)
        tune.track.log(mean_accuracy=acc)


analysis = tune.run(train_mnist, config={"lr": tune.grid_search([0.001, 0.01, 0.1])})

print("Best config: ", analysis.get_best_config(metric="mean_accuracy"))

# Get a dataframe for analyzing trial results.
df = analysis.dataframe()
```
When running the code you get this error message:
**TypeError: train_mnist() takes 1 positional argument but 2 were given**


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python/ray/tune/function_runner.py`
Content:
```
1 from __future__ import absolute_import
2 from __future__ import division
3 from __future__ import print_function
4 
5 import logging
6 import time
7 import inspect
8 import threading
9 import traceback
10 from six.moves import queue
11 
12 from ray.tune import track
13 from ray.tune import TuneError
14 from ray.tune.trainable import Trainable
15 from ray.tune.result import TIME_THIS_ITER_S, RESULT_DUPLICATE
16 
17 logger = logging.getLogger(__name__)
18 
19 # Time between FunctionRunner checks when fetching
20 # new results after signaling the reporter to continue
21 RESULT_FETCH_TIMEOUT = 0.2
22 
23 ERROR_REPORT_TIMEOUT = 10
24 ERROR_FETCH_TIMEOUT = 1
25 
26 
27 class StatusReporter(object):
28     """Object passed into your function that you can report status through.
29 
30     Example:
31         >>> def trainable_function(config, reporter):
32         >>>     assert isinstance(reporter, StatusReporter)
33         >>>     reporter(timesteps_this_iter=1)
34     """
35 
36     def __init__(self, result_queue, continue_semaphore, logdir=None):
37         self._queue = result_queue
38         self._last_report_time = None
39         self._continue_semaphore = continue_semaphore
40         self._logdir = logdir
41 
42     def __call__(self, **kwargs):
43         """Report updated training status.
44 
45         Pass in `done=True` when the training job is completed.
46 
47         Args:
48             kwargs: Latest training result status.
49 
50         Example:
51             >>> reporter(mean_accuracy=1, training_iteration=4)
52             >>> reporter(mean_accuracy=1, training_iteration=4, done=True)
53 
54         Raises:
55             StopIteration: A StopIteration exception is raised if the trial has
56                 been signaled to stop.
57         """
58 
59         assert self._last_report_time is not None, (
60             "StatusReporter._start() must be called before the first "
61             "report __call__ is made to ensure correct runtime metrics.")
62 
63         # time per iteration is recorded directly in the reporter to ensure
64         # any delays in logging results aren't counted
65         report_time = time.time()
66         if TIME_THIS_ITER_S not in kwargs:
67             kwargs[TIME_THIS_ITER_S] = report_time - self._last_report_time
68         self._last_report_time = report_time
69 
70         # add results to a thread-safe queue
71         self._queue.put(kwargs.copy(), block=True)
72 
73         # This blocks until notification from the FunctionRunner that the last
74         # result has been returned to Tune and that the function is safe to
75         # resume training.
76         self._continue_semaphore.acquire()
77 
78     def _start(self):
79         self._last_report_time = time.time()
80 
81     @property
82     def logdir(self):
83         return self._logdir
84 
85 
86 class _RunnerThread(threading.Thread):
87     """Supervisor thread that runs your script."""
88 
89     def __init__(self, entrypoint, error_queue):
90         threading.Thread.__init__(self)
91         self._entrypoint = entrypoint
92         self._error_queue = error_queue
93         self.daemon = True
94 
95     def run(self):
96         try:
97             self._entrypoint()
98         except StopIteration:
99             logger.debug(
100                 ("Thread runner raised StopIteration. Interperting it as a "
101                  "signal to terminate the thread without error."))
102         except Exception as e:
103             logger.exception("Runner Thread raised error.")
104             try:
105                 # report the error but avoid indefinite blocking which would
106                 # prevent the exception from being propagated in the unlikely
107                 # case that something went terribly wrong
108                 err_tb_str = traceback.format_exc()
109                 self._error_queue.put(
110                     err_tb_str, block=True, timeout=ERROR_REPORT_TIMEOUT)
111             except queue.Full:
112                 logger.critical(
113                     ("Runner Thread was unable to report error to main "
114                      "function runner thread. This means a previous error "
115                      "was not processed. This should never happen."))
116             raise e
117 
118 
119 class FunctionRunner(Trainable):
120     """Trainable that runs a user function reporting results.
121 
122     This mode of execution does not support checkpoint/restore."""
123 
124     _name = "func"
125 
126     def _setup(self, config):
127         # Semaphore for notifying the reporter to continue with the computation
128         # and to generate the next result.
129         self._continue_semaphore = threading.Semaphore(0)
130 
131         # Queue for passing results between threads
132         self._results_queue = queue.Queue(1)
133 
134         # Queue for passing errors back from the thread runner. The error queue
135         # has a max size of one to prevent stacking error and force error
136         # reporting to block until finished.
137         self._error_queue = queue.Queue(1)
138 
139         self._status_reporter = StatusReporter(
140             self._results_queue, self._continue_semaphore, self.logdir)
141         self._last_result = {}
142         config = config.copy()
143 
144         def entrypoint():
145             return self._trainable_func(config, self._status_reporter)
146 
147         # the runner thread is not started until the first call to _train
148         self._runner = _RunnerThread(entrypoint, self._error_queue)
149 
150     def _trainable_func(self):
151         """Subclasses can override this to set the trainable func."""
152 
153         raise NotImplementedError
154 
155     def _train(self):
156         """Implements train() for a Function API.
157 
158         If the RunnerThread finishes without reporting "done",
159         Tune will automatically provide a magic keyword __duplicate__
160         along with a result with "done=True". The TrialRunner will handle the
161         result accordingly (see tune/trial_runner.py).
162         """
163         if self._runner.is_alive():
164             # if started and alive, inform the reporter to continue and
165             # generate the next result
166             self._continue_semaphore.release()
167         else:
168             # if not alive, try to start
169             self._status_reporter._start()
170             try:
171                 self._runner.start()
172             except RuntimeError:
173                 # If this is reached, it means the thread was started and is
174                 # now done or has raised an exception.
175                 pass
176 
177         result = None
178         while result is None and self._runner.is_alive():
179             # fetch the next produced result
180             try:
181                 result = self._results_queue.get(
182                     block=True, timeout=RESULT_FETCH_TIMEOUT)
183             except queue.Empty:
184                 pass
185 
186         # if no result were found, then the runner must no longer be alive
187         if result is None:
188             # Try one last time to fetch results in case results were reported
189             # in between the time of the last check and the termination of the
190             # thread runner.
191             try:
192                 result = self._results_queue.get(block=False)
193             except queue.Empty:
194                 pass
195 
196         # check if error occured inside the thread runner
197         if result is None:
198             # only raise an error from the runner if all results are consumed
199             self._report_thread_runner_error(block=True)
200 
201             # Under normal conditions, this code should never be reached since
202             # this branch should only be visited if the runner thread raised
203             # an exception. If no exception were raised, it means that the
204             # runner thread never reported any results which should not be
205             # possible when wrapping functions with `wrap_function`.
206             raise TuneError(
207                 ("Wrapped function ran until completion without reporting "
208                  "results or raising an exception."))
209 
210         else:
211             if not self._error_queue.empty():
212                 logger.warning(
213                     ("Runner error waiting to be raised in main thread. "
214                      "Logging all available results first."))
215 
216         # This keyword appears if the train_func using the Function API
217         # finishes without "done=True". This duplicates the last result, but
218         # the TrialRunner will not log this result again.
219         if "__duplicate__" in result:
220             new_result = self._last_result.copy()
221             new_result.update(result)
222             result = new_result
223 
224         self._last_result = result
225         return result
226 
227     def _stop(self):
228         # If everything stayed in synch properly, this should never happen.
229         if not self._results_queue.empty():
230             logger.warning(
231                 ("Some results were added after the trial stop condition. "
232                  "These results won't be logged."))
233 
234         # Check for any errors that might have been missed.
235         self._report_thread_runner_error()
236 
237     def _report_thread_runner_error(self, block=False):
238         try:
239             err_tb_str = self._error_queue.get(
240                 block=block, timeout=ERROR_FETCH_TIMEOUT)
241             raise TuneError(("Trial raised an exception. Traceback:\n{}"
242                              .format(err_tb_str)))
243         except queue.Empty:
244             pass
245 
246 
247 def wrap_function(train_func):
248 
249     use_track = False
250     try:
251         func_args = inspect.getargspec(train_func).args
252         use_track = ("reporter" not in func_args and len(func_args) == 1)
253         if use_track:
254             logger.info("tune.track signature detected.")
255     except Exception:
256         logger.info(
257             "Function inspection failed - assuming reporter signature.")
258 
259     class WrappedFunc(FunctionRunner):
260         def _trainable_func(self, config, reporter):
261             output = train_func(config, reporter)
262             # If train_func returns, we need to notify the main event loop
263             # of the last result while avoiding double logging. This is done
264             # with the keyword RESULT_DUPLICATE -- see tune/trial_runner.py.
265             reporter(**{RESULT_DUPLICATE: True})
266             return output
267 
268     class WrappedTrackFunc(FunctionRunner):
269         def _trainable_func(self, config, reporter):
270             track.init(_tune_reporter=reporter)
271             output = train_func(config)
272             reporter(**{RESULT_DUPLICATE: True})
273             track.shutdown()
274             return output
275 
276     return WrappedTrackFunc if use_track else WrappedFunc
277
```
--- 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/python/ray/tune/function_runner.py b/python/ray/tune/function_runner.py
--- a/python/ray/tune/function_runner.py
+++ b/python/ray/tune/function_runner.py
@@ -7,6 +7,7 @@
 import inspect
 import threading
 import traceback
+import sys
 from six.moves import queue
 
 from ray.tune import track
@@ -248,7 +249,10 @@
 
     use_track = False
     try:
-        func_args = inspect.getargspec(train_func).args
+        if sys.version_info >= (3, 3):
+            func_args = inspect.getfullargspec(train_func).args
+        else:
+            func_args = inspect.getargspec(train_func).args
         use_track = ("reporter" not in func_args and len(func_args) == 1)
         if use_track:
             logger.info("tune.track signature detected.")

{"golden_diff": "diff --git a/python/ray/tune/function_runner.py b/python/ray/tune/function_runner.py\n--- a/python/ray/tune/function_runner.py\n+++ b/python/ray/tune/function_runner.py\n@@ -7,6 +7,7 @@\n import inspect\n import threading\n import traceback\n+import sys\n from six.moves import queue\n \n from ray.tune import track\n@@ -248,7 +249,10 @@\n \n     use_track = False\n     try:\n-        func_args = inspect.getargspec(train_func).args\n+        if sys.version_info >= (3, 3):\n+            func_args = inspect.getfullargspec(train_func).args\n+        else:\n+            func_args = inspect.getargspec(train_func).args\n         use_track = (\"reporter\" not in func_args and len(func_args) == 1)\n         if use_track:\n             logger.info(\"tune.track signature detected.\")\n", "issue": "[tune] Ray Tune fails to parse typing hints of the function for experiment\n### What is the problem?\r\nIf the function for experiment has a [typing hint](https://docs.python.org/3/library/typing.html) for its argument `config`, then Ray Tune fails to parse the argument and assumes that there is a reporter signature. The cause of this problem is in this source code:\r\nhttps://github.com/ray-project/ray/blob/1eaa57c98f8870a43e1ea14ec011b6bd4be97c8d/python/ray/tune/function_runner.py#L250-L257\r\nChanging \r\n`func_args = inspect.getargspec(train_func).args` \r\nto\r\n`func_args = inspect.getfullargspec(train_func).args` \r\nmight solve the problem.\r\n\r\n*Ray version and other system information (Python version, TensorFlow version, OS):*\r\nRay: 0.8.0\r\nPython: 3.7.5\r\nOS: Ubuntu 18.04\r\n\r\n*Does the problem occur on the [latest wheels](https://ray.readthedocs.io/en/latest/installation.html)?*\r\nI couldn't install the latest wheel. So I can't confirm it.\r\n\r\n### Reproduction\r\nThe following is a modification of the first examples in the [Ray Tune Documentation](https://ray.readthedocs.io/en/latest/tune.html#quick-start), where I added a typing hint  `config: Dict[str, Any]` for the argument of function `train_mnist`.\r\n\r\n```\r\nfrom typing import Dict, Any\r\n\r\nimport torch.optim as optim\r\nfrom ray import tune\r\nfrom ray.tune.examples.mnist_pytorch import get_data_loaders, ConvNet, train, test\r\n\r\n\r\ndef train_mnist(config: Dict[str, Any]):\r\n    train_loader, test_loader = get_data_loaders()\r\n    model = ConvNet()\r\n    optimizer = optim.SGD(model.parameters(), lr=config[\"lr\"])\r\n    for i in range(10):\r\n        train(model, optimizer, train_loader)\r\n        acc = test(model, test_loader)\r\n        tune.track.log(mean_accuracy=acc)\r\n\r\n\r\nanalysis = tune.run(train_mnist, config={\"lr\": tune.grid_search([0.001, 0.01, 0.1])})\r\n\r\nprint(\"Best config: \", analysis.get_best_config(metric=\"mean_accuracy\"))\r\n\r\n# Get a dataframe for analyzing trial results.\r\ndf = analysis.dataframe()\r\n```\r\nWhen running the code you get this error message:\r\n**TypeError: train_mnist() takes 1 positional argument but 2 were given**\r\n\n", "before_files": [{"content": "from __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nimport logging\nimport time\nimport inspect\nimport threading\nimport traceback\nfrom six.moves import queue\n\nfrom ray.tune import track\nfrom ray.tune import TuneError\nfrom ray.tune.trainable import Trainable\nfrom ray.tune.result import TIME_THIS_ITER_S, RESULT_DUPLICATE\n\nlogger = logging.getLogger(__name__)\n\n# Time between FunctionRunner checks when fetching\n# new results after signaling the reporter to continue\nRESULT_FETCH_TIMEOUT = 0.2\n\nERROR_REPORT_TIMEOUT = 10\nERROR_FETCH_TIMEOUT = 1\n\n\nclass StatusReporter(object):\n    \"\"\"Object passed into your function that you can report status through.\n\n    Example:\n        >>> def trainable_function(config, reporter):\n        >>>     assert isinstance(reporter, StatusReporter)\n        >>>     reporter(timesteps_this_iter=1)\n    \"\"\"\n\n    def __init__(self, result_queue, continue_semaphore, logdir=None):\n        self._queue = result_queue\n        self._last_report_time = None\n        self._continue_semaphore = continue_semaphore\n        self._logdir = logdir\n\n    def __call__(self, **kwargs):\n        \"\"\"Report updated training status.\n\n        Pass in `done=True` when the training job is completed.\n\n        Args:\n            kwargs: Latest training result status.\n\n        Example:\n            >>> reporter(mean_accuracy=1, training_iteration=4)\n            >>> reporter(mean_accuracy=1, training_iteration=4, done=True)\n\n        Raises:\n            StopIteration: A StopIteration exception is raised if the trial has\n                been signaled to stop.\n        \"\"\"\n\n        assert self._last_report_time is not None, (\n            \"StatusReporter._start() must be called before the first \"\n            \"report __call__ is made to ensure correct runtime metrics.\")\n\n        # time per iteration is recorded directly in the reporter to ensure\n        # any delays in logging results aren't counted\n        report_time = time.time()\n        if TIME_THIS_ITER_S not in kwargs:\n            kwargs[TIME_THIS_ITER_S] = report_time - self._last_report_time\n        self._last_report_time = report_time\n\n        # add results to a thread-safe queue\n        self._queue.put(kwargs.copy(), block=True)\n\n        # This blocks until notification from the FunctionRunner that the last\n        # result has been returned to Tune and that the function is safe to\n        # resume training.\n        self._continue_semaphore.acquire()\n\n    def _start(self):\n        self._last_report_time = time.time()\n\n    @property\n    def logdir(self):\n        return self._logdir\n\n\nclass _RunnerThread(threading.Thread):\n    \"\"\"Supervisor thread that runs your script.\"\"\"\n\n    def __init__(self, entrypoint, error_queue):\n        threading.Thread.__init__(self)\n        self._entrypoint = entrypoint\n        self._error_queue = error_queue\n        self.daemon = True\n\n    def run(self):\n        try:\n            self._entrypoint()\n        except StopIteration:\n            logger.debug(\n                (\"Thread runner raised StopIteration. Interperting it as a \"\n                 \"signal to terminate the thread without error.\"))\n        except Exception as e:\n            logger.exception(\"Runner Thread raised error.\")\n            try:\n                # report the error but avoid indefinite blocking which would\n                # prevent the exception from being propagated in the unlikely\n                # case that something went terribly wrong\n                err_tb_str = traceback.format_exc()\n                self._error_queue.put(\n                    err_tb_str, block=True, timeout=ERROR_REPORT_TIMEOUT)\n            except queue.Full:\n                logger.critical(\n                    (\"Runner Thread was unable to report error to main \"\n                     \"function runner thread. This means a previous error \"\n                     \"was not processed. This should never happen.\"))\n            raise e\n\n\nclass FunctionRunner(Trainable):\n    \"\"\"Trainable that runs a user function reporting results.\n\n    This mode of execution does not support checkpoint/restore.\"\"\"\n\n    _name = \"func\"\n\n    def _setup(self, config):\n        # Semaphore for notifying the reporter to continue with the computation\n        # and to generate the next result.\n        self._continue_semaphore = threading.Semaphore(0)\n\n        # Queue for passing results between threads\n        self._results_queue = queue.Queue(1)\n\n        # Queue for passing errors back from the thread runner. The error queue\n        # has a max size of one to prevent stacking error and force error\n        # reporting to block until finished.\n        self._error_queue = queue.Queue(1)\n\n        self._status_reporter = StatusReporter(\n            self._results_queue, self._continue_semaphore, self.logdir)\n        self._last_result = {}\n        config = config.copy()\n\n        def entrypoint():\n            return self._trainable_func(config, self._status_reporter)\n\n        # the runner thread is not started until the first call to _train\n        self._runner = _RunnerThread(entrypoint, self._error_queue)\n\n    def _trainable_func(self):\n        \"\"\"Subclasses can override this to set the trainable func.\"\"\"\n\n        raise NotImplementedError\n\n    def _train(self):\n        \"\"\"Implements train() for a Function API.\n\n        If the RunnerThread finishes without reporting \"done\",\n        Tune will automatically provide a magic keyword __duplicate__\n        along with a result with \"done=True\". The TrialRunner will handle the\n        result accordingly (see tune/trial_runner.py).\n        \"\"\"\n        if self._runner.is_alive():\n            # if started and alive, inform the reporter to continue and\n            # generate the next result\n            self._continue_semaphore.release()\n        else:\n            # if not alive, try to start\n            self._status_reporter._start()\n            try:\n                self._runner.start()\n            except RuntimeError:\n                # If this is reached, it means the thread was started and is\n                # now done or has raised an exception.\n                pass\n\n        result = None\n        while result is None and self._runner.is_alive():\n            # fetch the next produced result\n            try:\n                result = self._results_queue.get(\n                    block=True, timeout=RESULT_FETCH_TIMEOUT)\n            except queue.Empty:\n                pass\n\n        # if no result were found, then the runner must no longer be alive\n        if result is None:\n            # Try one last time to fetch results in case results were reported\n            # in between the time of the last check and the termination of the\n            # thread runner.\n            try:\n                result = self._results_queue.get(block=False)\n            except queue.Empty:\n                pass\n\n        # check if error occured inside the thread runner\n        if result is None:\n            # only raise an error from the runner if all results are consumed\n            self._report_thread_runner_error(block=True)\n\n            # Under normal conditions, this code should never be reached since\n            # this branch should only be visited if the runner thread raised\n            # an exception. If no exception were raised, it means that the\n            # runner thread never reported any results which should not be\n            # possible when wrapping functions with `wrap_function`.\n            raise TuneError(\n                (\"Wrapped function ran until completion without reporting \"\n                 \"results or raising an exception.\"))\n\n        else:\n            if not self._error_queue.empty():\n                logger.warning(\n                    (\"Runner error waiting to be raised in main thread. \"\n                     \"Logging all available results first.\"))\n\n        # This keyword appears if the train_func using the Function API\n        # finishes without \"done=True\". This duplicates the last result, but\n        # the TrialRunner will not log this result again.\n        if \"__duplicate__\" in result:\n            new_result = self._last_result.copy()\n            new_result.update(result)\n            result = new_result\n\n        self._last_result = result\n        return result\n\n    def _stop(self):\n        # If everything stayed in synch properly, this should never happen.\n        if not self._results_queue.empty():\n            logger.warning(\n                (\"Some results were added after the trial stop condition. \"\n                 \"These results won't be logged.\"))\n\n        # Check for any errors that might have been missed.\n        self._report_thread_runner_error()\n\n    def _report_thread_runner_error(self, block=False):\n        try:\n            err_tb_str = self._error_queue.get(\n                block=block, timeout=ERROR_FETCH_TIMEOUT)\n            raise TuneError((\"Trial raised an exception. Traceback:\\n{}\"\n                             .format(err_tb_str)))\n        except queue.Empty:\n            pass\n\n\ndef wrap_function(train_func):\n\n    use_track = False\n    try:\n        func_args = inspect.getargspec(train_func).args\n        use_track = (\"reporter\" not in func_args and len(func_args) == 1)\n        if use_track:\n            logger.info(\"tune.track signature detected.\")\n    except Exception:\n        logger.info(\n            \"Function inspection failed - assuming reporter signature.\")\n\n    class WrappedFunc(FunctionRunner):\n        def _trainable_func(self, config, reporter):\n            output = train_func(config, reporter)\n            # If train_func returns, we need to notify the main event loop\n            # of the last result while avoiding double logging. This is done\n            # with the keyword RESULT_DUPLICATE -- see tune/trial_runner.py.\n            reporter(**{RESULT_DUPLICATE: True})\n            return output\n\n    class WrappedTrackFunc(FunctionRunner):\n        def _trainable_func(self, config, reporter):\n            track.init(_tune_reporter=reporter)\n            output = train_func(config)\n            reporter(**{RESULT_DUPLICATE: True})\n            track.shutdown()\n            return output\n\n    return WrappedTrackFunc if use_track else WrappedFunc\n", "path": "python/ray/tune/function_runner.py"}], "after_files": [{"content": "from __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nimport logging\nimport time\nimport inspect\nimport threading\nimport traceback\nimport sys\nfrom six.moves import queue\n\nfrom ray.tune import track\nfrom ray.tune import TuneError\nfrom ray.tune.trainable import Trainable\nfrom ray.tune.result import TIME_THIS_ITER_S, RESULT_DUPLICATE\n\nlogger = logging.getLogger(__name__)\n\n# Time between FunctionRunner checks when fetching\n# new results after signaling the reporter to continue\nRESULT_FETCH_TIMEOUT = 0.2\n\nERROR_REPORT_TIMEOUT = 10\nERROR_FETCH_TIMEOUT = 1\n\n\nclass StatusReporter(object):\n    \"\"\"Object passed into your function that you can report status through.\n\n    Example:\n        >>> def trainable_function(config, reporter):\n        >>>     assert isinstance(reporter, StatusReporter)\n        >>>     reporter(timesteps_this_iter=1)\n    \"\"\"\n\n    def __init__(self, result_queue, continue_semaphore, logdir=None):\n        self._queue = result_queue\n        self._last_report_time = None\n        self._continue_semaphore = continue_semaphore\n        self._logdir = logdir\n\n    def __call__(self, **kwargs):\n        \"\"\"Report updated training status.\n\n        Pass in `done=True` when the training job is completed.\n\n        Args:\n            kwargs: Latest training result status.\n\n        Example:\n            >>> reporter(mean_accuracy=1, training_iteration=4)\n            >>> reporter(mean_accuracy=1, training_iteration=4, done=True)\n\n        Raises:\n            StopIteration: A StopIteration exception is raised if the trial has\n                been signaled to stop.\n        \"\"\"\n\n        assert self._last_report_time is not None, (\n            \"StatusReporter._start() must be called before the first \"\n            \"report __call__ is made to ensure correct runtime metrics.\")\n\n        # time per iteration is recorded directly in the reporter to ensure\n        # any delays in logging results aren't counted\n        report_time = time.time()\n        if TIME_THIS_ITER_S not in kwargs:\n            kwargs[TIME_THIS_ITER_S] = report_time - self._last_report_time\n        self._last_report_time = report_time\n\n        # add results to a thread-safe queue\n        self._queue.put(kwargs.copy(), block=True)\n\n        # This blocks until notification from the FunctionRunner that the last\n        # result has been returned to Tune and that the function is safe to\n        # resume training.\n        self._continue_semaphore.acquire()\n\n    def _start(self):\n        self._last_report_time = time.time()\n\n    @property\n    def logdir(self):\n        return self._logdir\n\n\nclass _RunnerThread(threading.Thread):\n    \"\"\"Supervisor thread that runs your script.\"\"\"\n\n    def __init__(self, entrypoint, error_queue):\n        threading.Thread.__init__(self)\n        self._entrypoint = entrypoint\n        self._error_queue = error_queue\n        self.daemon = True\n\n    def run(self):\n        try:\n            self._entrypoint()\n        except StopIteration:\n            logger.debug(\n                (\"Thread runner raised StopIteration. Interperting it as a \"\n                 \"signal to terminate the thread without error.\"))\n        except Exception as e:\n            logger.exception(\"Runner Thread raised error.\")\n            try:\n                # report the error but avoid indefinite blocking which would\n                # prevent the exception from being propagated in the unlikely\n                # case that something went terribly wrong\n                err_tb_str = traceback.format_exc()\n                self._error_queue.put(\n                    err_tb_str, block=True, timeout=ERROR_REPORT_TIMEOUT)\n            except queue.Full:\n                logger.critical(\n                    (\"Runner Thread was unable to report error to main \"\n                     \"function runner thread. This means a previous error \"\n                     \"was not processed. This should never happen.\"))\n            raise e\n\n\nclass FunctionRunner(Trainable):\n    \"\"\"Trainable that runs a user function reporting results.\n\n    This mode of execution does not support checkpoint/restore.\"\"\"\n\n    _name = \"func\"\n\n    def _setup(self, config):\n        # Semaphore for notifying the reporter to continue with the computation\n        # and to generate the next result.\n        self._continue_semaphore = threading.Semaphore(0)\n\n        # Queue for passing results between threads\n        self._results_queue = queue.Queue(1)\n\n        # Queue for passing errors back from the thread runner. The error queue\n        # has a max size of one to prevent stacking error and force error\n        # reporting to block until finished.\n        self._error_queue = queue.Queue(1)\n\n        self._status_reporter = StatusReporter(\n            self._results_queue, self._continue_semaphore, self.logdir)\n        self._last_result = {}\n        config = config.copy()\n\n        def entrypoint():\n            return self._trainable_func(config, self._status_reporter)\n\n        # the runner thread is not started until the first call to _train\n        self._runner = _RunnerThread(entrypoint, self._error_queue)\n\n    def _trainable_func(self):\n        \"\"\"Subclasses can override this to set the trainable func.\"\"\"\n\n        raise NotImplementedError\n\n    def _train(self):\n        \"\"\"Implements train() for a Function API.\n\n        If the RunnerThread finishes without reporting \"done\",\n        Tune will automatically provide a magic keyword __duplicate__\n        along with a result with \"done=True\". The TrialRunner will handle the\n        result accordingly (see tune/trial_runner.py).\n        \"\"\"\n        if self._runner.is_alive():\n            # if started and alive, inform the reporter to continue and\n            # generate the next result\n            self._continue_semaphore.release()\n        else:\n            # if not alive, try to start\n            self._status_reporter._start()\n            try:\n                self._runner.start()\n            except RuntimeError:\n                # If this is reached, it means the thread was started and is\n                # now done or has raised an exception.\n                pass\n\n        result = None\n        while result is None and self._runner.is_alive():\n            # fetch the next produced result\n            try:\n                result = self._results_queue.get(\n                    block=True, timeout=RESULT_FETCH_TIMEOUT)\n            except queue.Empty:\n                pass\n\n        # if no result were found, then the runner must no longer be alive\n        if result is None:\n            # Try one last time to fetch results in case results were reported\n            # in between the time of the last check and the termination of the\n            # thread runner.\n            try:\n                result = self._results_queue.get(block=False)\n            except queue.Empty:\n                pass\n\n        # check if error occured inside the thread runner\n        if result is None:\n            # only raise an error from the runner if all results are consumed\n            self._report_thread_runner_error(block=True)\n\n            # Under normal conditions, this code should never be reached since\n            # this branch should only be visited if the runner thread raised\n            # an exception. If no exception were raised, it means that the\n            # runner thread never reported any results which should not be\n            # possible when wrapping functions with `wrap_function`.\n            raise TuneError(\n                (\"Wrapped function ran until completion without reporting \"\n                 \"results or raising an exception.\"))\n\n        else:\n            if not self._error_queue.empty():\n                logger.warning(\n                    (\"Runner error waiting to be raised in main thread. \"\n                     \"Logging all available results first.\"))\n\n        # This keyword appears if the train_func using the Function API\n        # finishes without \"done=True\". This duplicates the last result, but\n        # the TrialRunner will not log this result again.\n        if \"__duplicate__\" in result:\n            new_result = self._last_result.copy()\n            new_result.update(result)\n            result = new_result\n\n        self._last_result = result\n        return result\n\n    def _stop(self):\n        # If everything stayed in synch properly, this should never happen.\n        if not self._results_queue.empty():\n            logger.warning(\n                (\"Some results were added after the trial stop condition. \"\n                 \"These results won't be logged.\"))\n\n        # Check for any errors that might have been missed.\n        self._report_thread_runner_error()\n\n    def _report_thread_runner_error(self, block=False):\n        try:\n            err_tb_str = self._error_queue.get(\n                block=block, timeout=ERROR_FETCH_TIMEOUT)\n            raise TuneError((\"Trial raised an exception. Traceback:\\n{}\"\n                             .format(err_tb_str)))\n        except queue.Empty:\n            pass\n\n\ndef wrap_function(train_func):\n\n    use_track = False\n    try:\n        if sys.version_info >= (3, 3):\n            func_args = inspect.getfullargspec(train_func).args\n        else:\n            func_args = inspect.getargspec(train_func).args\n        use_track = (\"reporter\" not in func_args and len(func_args) == 1)\n        if use_track:\n            logger.info(\"tune.track signature detected.\")\n    except Exception:\n        logger.info(\n            \"Function inspection failed - assuming reporter signature.\")\n\n    class WrappedFunc(FunctionRunner):\n        def _trainable_func(self, config, reporter):\n            output = train_func(config, reporter)\n            # If train_func returns, we need to notify the main event loop\n            # of the last result while avoiding double logging. This is done\n            # with the keyword RESULT_DUPLICATE -- see tune/trial_runner.py.\n            reporter(**{RESULT_DUPLICATE: True})\n            return output\n\n    class WrappedTrackFunc(FunctionRunner):\n        def _trainable_func(self, config, reporter):\n            track.init(_tune_reporter=reporter)\n            output = train_func(config)\n            reporter(**{RESULT_DUPLICATE: True})\n            track.shutdown()\n            return output\n\n    return WrappedTrackFunc if use_track else WrappedFunc\n", "path": "python/ray/tune/function_runner.py"}]}