From f6fce5de847541fbe3c993bc4316278fda5ef3e6 Mon Sep 17 00:00:00 2001 From: shingjan Date: Wed, 25 Aug 2021 17:33:26 -0700 Subject: [PATCH 1/9] use popenpool in local_executor --- python/tvm/autotvm/measure/local_executor.py | 73 ++++--------------- .../python/unittest/test_autotvm_executor.py | 9 ++- 2 files changed, 20 insertions(+), 62 deletions(-) diff --git a/python/tvm/autotvm/measure/local_executor.py b/python/tvm/autotvm/measure/local_executor.py index a9aeb790c82a..a94312b49407 100644 --- a/python/tvm/autotvm/measure/local_executor.py +++ b/python/tvm/autotvm/measure/local_executor.py @@ -18,19 +18,14 @@ import signal -from multiprocessing import Process, Queue - -try: - from queue import Empty -except ImportError: - from Queue import Empty - try: import psutil except ImportError: psutil = None from . import executor +from ..env import GLOBAL_SCOPE +from ...contrib.popen_pool import PopenPoolExecutor def kill_child_processes(parent_pid, sig=signal.SIGTERM): @@ -47,65 +42,28 @@ def kill_child_processes(parent_pid, sig=signal.SIGTERM): return -def _execute_func(func, queue, args, kwargs): - """execute function and return the result or exception to a queue""" - try: - res = func(*args, **kwargs) - except Exception as exc: # pylint: disable=broad-except - res = exc - queue.put(res) - - -def call_with_timeout(queue, timeout, func, args, kwargs): - """A wrapper to support timeout of a function call""" - - # start a new process for timeout (cannot use thread because we have c function) - p = Process(target=_execute_func, args=(func, queue, args, kwargs)) - p.start() - p.join(timeout=timeout) - - queue.put(executor.TimeoutError()) - - kill_child_processes(p.pid) - p.terminate() - p.join() +def _popen_initializer(global_scope): + global GLOBAL_SCOPE + GLOBAL_SCOPE = global_scope class LocalFuture(executor.Future): """Local wrapper for the future - Parameters ---------- - process: multiprocessing.Process - process for running this task - queue: multiprocessing.Queue - queue for receiving the result of this task + future: concurrent.futures.Future + A future returned by PopenPoolExecutor. """ - def __init__(self, process, queue): + def __init__(self, future): self._done = False - self._process = process - self._queue = queue + self._future = future def done(self): - self._done = self._done or not self._queue.empty() - return self._done + return self._future.done() def get(self, timeout=None): - try: - res = self._queue.get(block=True, timeout=timeout) - except Empty: - raise executor.TimeoutError() - if self._process.is_alive(): - kill_child_processes(self._process.pid) - self._process.terminate() - self._process.join() - self._queue.close() - self._queue.join_thread() - self._done = True - del self._queue - del self._process - return res + return self._future.result(timeout) class LocalFutureNoFork(executor.Future): @@ -126,7 +84,6 @@ def get(self, timeout=None): class LocalExecutor(executor.Executor): """Local executor that runs workers on the same machine with multiprocessing. - Parameters ---------- timeout: float, optional @@ -151,7 +108,7 @@ def submit(self, func, *args, **kwargs): if not self.do_fork: return LocalFutureNoFork(func(*args, **kwargs)) - queue = Queue(2) # Size of 2 to avoid a race condition with size 1. - process = Process(target=call_with_timeout, args=(queue, self.timeout, func, args, kwargs)) - process.start() - return LocalFuture(process, queue) + pool = PopenPoolExecutor( + timeout=self.timeout, initializer=_popen_initializer, initargs=(GLOBAL_SCOPE,) + ) + return LocalFuture(pool.submit(func, args, kwargs)) diff --git a/tests/python/unittest/test_autotvm_executor.py b/tests/python/unittest/test_autotvm_executor.py index 9757576be9e3..92ad2bf765e1 100644 --- a/tests/python/unittest/test_autotvm_executor.py +++ b/tests/python/unittest/test_autotvm_executor.py @@ -58,10 +58,11 @@ def test_timeout(): ex = LocalExecutor(timeout=timeout) f1 = ex.submit(timeout_job, timeout) - while not f1.done(): - pass - res = f1.get() - assert isinstance(res, executor.TimeoutError) + try: + res = f1.get() + except Exception as ex: + res = ex + assert isinstance(res, executor.TimeoutError) if __name__ == "__main__": From 5c2c09e785b4568fae05f79c32e1a884fdddffa6 Mon Sep 17 00:00:00 2001 From: shingjan Date: Wed, 25 Aug 2021 17:56:16 -0700 Subject: [PATCH 2/9] move auto_tvm_common to tvm.testing --- python/tvm/testing/__init__.py | 1 + .../test_autotvm_common.py => python/tvm/testing/autotvm.py | 1 + tests/python/unittest/test_autotvm_database.py | 2 +- tests/python/unittest/test_autotvm_index_tuner.py | 2 +- tests/python/unittest/test_autotvm_measure.py | 2 +- tests/python/unittest/test_autotvm_record.py | 2 +- tests/python/unittest/test_autotvm_xgboost_model.py | 2 +- 7 files changed, 7 insertions(+), 5 deletions(-) rename tests/python/unittest/test_autotvm_common.py => python/tvm/testing/autotvm.py (97%) diff --git a/python/tvm/testing/__init__.py b/python/tvm/testing/__init__.py index 3b4e743d6f07..6d8bc70eb519 100644 --- a/python/tvm/testing/__init__.py +++ b/python/tvm/testing/__init__.py @@ -27,3 +27,4 @@ from .popen_pool import call_py_ffi, call_cpp_py_ffi from . import auto_scheduler +from . import autotvm diff --git a/tests/python/unittest/test_autotvm_common.py b/python/tvm/testing/autotvm.py similarity index 97% rename from tests/python/unittest/test_autotvm_common.py rename to python/tvm/testing/autotvm.py index 60f7d8bafb1b..6f7bb13fe6dc 100644 --- a/tests/python/unittest/test_autotvm_common.py +++ b/python/tvm/testing/autotvm.py @@ -14,6 +14,7 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +# pylint: disable=invalid-name, missing-function-docstring, missing-class-docstring """Common utilities for testing autotvm""" import time diff --git a/tests/python/unittest/test_autotvm_database.py b/tests/python/unittest/test_autotvm_database.py index 197243ed47c0..d5980022811f 100644 --- a/tests/python/unittest/test_autotvm_database.py +++ b/tests/python/unittest/test_autotvm_database.py @@ -21,7 +21,7 @@ from tvm.autotvm import database from tvm.autotvm.record import encode, MeasureResult -from test_autotvm_common import get_sample_records +from tvm.testing.autotvm import get_sample_records def test_save_load(): diff --git a/tests/python/unittest/test_autotvm_index_tuner.py b/tests/python/unittest/test_autotvm_index_tuner.py index c433d8fb7297..be89ee2506fc 100644 --- a/tests/python/unittest/test_autotvm_index_tuner.py +++ b/tests/python/unittest/test_autotvm_index_tuner.py @@ -17,7 +17,7 @@ """Test index based tuners""" import multiprocessing -from test_autotvm_common import DummyRunner, get_sample_task +from tvm.testing.autotvm import DummyRunner, get_sample_task from tvm import autotvm from tvm.autotvm.tuner import GridSearchTuner, RandomTuner diff --git a/tests/python/unittest/test_autotvm_measure.py b/tests/python/unittest/test_autotvm_measure.py index a89c69c37d64..3f56243a5137 100644 --- a/tests/python/unittest/test_autotvm_measure.py +++ b/tests/python/unittest/test_autotvm_measure.py @@ -23,7 +23,7 @@ import tvm from tvm import te -from test_autotvm_common import DummyRunner, bad_matmul, get_sample_task +from tvm.testing.autotvm import DummyRunner, bad_matmul, get_sample_task from tvm import autotvm from tvm.autotvm.measure.measure import MeasureErrorNo, MeasureResult from tvm.autotvm import measure diff --git a/tests/python/unittest/test_autotvm_record.py b/tests/python/unittest/test_autotvm_record.py index 51cc9074a4fe..65739df52cd9 100644 --- a/tests/python/unittest/test_autotvm_record.py +++ b/tests/python/unittest/test_autotvm_record.py @@ -25,7 +25,7 @@ from tvm.autotvm.measure import MeasureInput, MeasureResult, MeasureErrorNo from tvm.autotvm.record import encode, decode, ApplyHistoryBest, measure_str_key -from test_autotvm_common import get_sample_task +from tvm.testing.autotvm import get_sample_task def test_load_dump(): diff --git a/tests/python/unittest/test_autotvm_xgboost_model.py b/tests/python/unittest/test_autotvm_xgboost_model.py index 445cff8759ab..baecdaceab6d 100644 --- a/tests/python/unittest/test_autotvm_xgboost_model.py +++ b/tests/python/unittest/test_autotvm_xgboost_model.py @@ -25,7 +25,7 @@ from tvm.autotvm import MeasureInput, MeasureResult from tvm.autotvm.tuner.xgboost_cost_model import XGBoostCostModel -from test_autotvm_common import get_sample_task, get_sample_records +from tvm.testing.autotvm import get_sample_task, get_sample_records def test_fit(): From 4d92aea6704f9c77e8d3300de9970735dca6e07c Mon Sep 17 00:00:00 2001 From: shingjan Date: Thu, 26 Aug 2021 16:14:15 -0700 Subject: [PATCH 3/9] refactor --- python/tvm/autotvm/measure/__init__.py | 1 - python/tvm/autotvm/measure/executor.py | 19 +++ python/tvm/autotvm/measure/local_executor.py | 114 ------------------ python/tvm/autotvm/measure/measure_methods.py | 10 +- python/tvm/testing/__init__.py | 3 +- python/tvm/testing/popen_pool.py | 16 +++ tests/python/contrib/test_popen_pool.py | 36 ++++++ .../python/unittest/test_autotvm_executor.py | 70 ----------- tests/python/unittest/test_autotvm_measure.py | 4 +- 9 files changed, 81 insertions(+), 192 deletions(-) delete mode 100644 python/tvm/autotvm/measure/local_executor.py delete mode 100644 tests/python/unittest/test_autotvm_executor.py diff --git a/python/tvm/autotvm/measure/__init__.py b/python/tvm/autotvm/measure/__init__.py index c4c0dc92b116..10b0843402ea 100644 --- a/python/tvm/autotvm/measure/__init__.py +++ b/python/tvm/autotvm/measure/__init__.py @@ -31,4 +31,3 @@ request_remote, ) from .executor import Executor -from .local_executor import LocalExecutor diff --git a/python/tvm/autotvm/measure/executor.py b/python/tvm/autotvm/measure/executor.py index f8eca7298c64..85362727410c 100644 --- a/python/tvm/autotvm/measure/executor.py +++ b/python/tvm/autotvm/measure/executor.py @@ -87,6 +87,25 @@ def get(self, timeout=None): raise NotImplementedError() +class LocalFutureNoFork(Future): + """Local wrapper for the future. + This is a none-fork version of LocalFuture. + Use this for unittest and runtimes that does not support fork (like cudnn) + """ + + def __init__(self, result): + self._result = result + + def done(self): + return True + + def get(self, timeout=None): # pylint: disable=unused-argument + return self._result + + def result(self, timeout=None): # pylint: disable=unused-argument + return self._result + + class FutureError(RuntimeError): """Base error class of all future events""" diff --git a/python/tvm/autotvm/measure/local_executor.py b/python/tvm/autotvm/measure/local_executor.py deleted file mode 100644 index a94312b49407..000000000000 --- a/python/tvm/autotvm/measure/local_executor.py +++ /dev/null @@ -1,114 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. -"""Local based implementation of the executor using multiprocessing""" - -import signal - -try: - import psutil -except ImportError: - psutil = None - -from . import executor -from ..env import GLOBAL_SCOPE -from ...contrib.popen_pool import PopenPoolExecutor - - -def kill_child_processes(parent_pid, sig=signal.SIGTERM): - """kill all child processes recursively""" - try: - parent = psutil.Process(parent_pid) - children = parent.children(recursive=True) - except psutil.NoSuchProcess: - return - for process in children: - try: - process.send_signal(sig) - except psutil.NoSuchProcess: - return - - -def _popen_initializer(global_scope): - global GLOBAL_SCOPE - GLOBAL_SCOPE = global_scope - - -class LocalFuture(executor.Future): - """Local wrapper for the future - Parameters - ---------- - future: concurrent.futures.Future - A future returned by PopenPoolExecutor. - """ - - def __init__(self, future): - self._done = False - self._future = future - - def done(self): - return self._future.done() - - def get(self, timeout=None): - return self._future.result(timeout) - - -class LocalFutureNoFork(executor.Future): - """Local wrapper for the future. - This is a none-fork version of LocalFuture. - Use this for the runtime that does not support fork (like cudnn) - """ - - def __init__(self, result): - self._result = result - - def done(self): - return True - - def get(self, timeout=None): - return self._result - - -class LocalExecutor(executor.Executor): - """Local executor that runs workers on the same machine with multiprocessing. - Parameters - ---------- - timeout: float, optional - timeout of a job. If time is out. A TimeoutError will be returned (not raised) - do_fork: bool, optional - For some runtime systems that do not support fork after initialization - (e.g. cuda runtime, cudnn). Set this to False if you have used these runtime - before submitting jobs. - """ - - def __init__(self, timeout=None, do_fork=True): - self.timeout = timeout or executor.Executor.DEFAULT_TIMEOUT - self.do_fork = do_fork - - if self.do_fork: - if not psutil: - raise RuntimeError( - "Python package psutil is missing. " "please try `pip install psutil`" - ) - - def submit(self, func, *args, **kwargs): - if not self.do_fork: - return LocalFutureNoFork(func(*args, **kwargs)) - - pool = PopenPoolExecutor( - timeout=self.timeout, initializer=_popen_initializer, initargs=(GLOBAL_SCOPE,) - ) - return LocalFuture(pool.submit(func, args, kwargs)) diff --git a/python/tvm/autotvm/measure/measure_methods.py b/python/tvm/autotvm/measure/measure_methods.py index eab6822b63b8..59b2898feab6 100644 --- a/python/tvm/autotvm/measure/measure_methods.py +++ b/python/tvm/autotvm/measure/measure_methods.py @@ -39,6 +39,7 @@ from tvm import nd from tvm import rpc as _rpc from tvm.contrib import ndk, nvcc, stackvm, tar +from tvm.contrib.popen_pool import PopenPoolExecutor from tvm.driver import build from tvm.error import TVMError from tvm.target import Target @@ -46,7 +47,6 @@ from ..env import AutotvmGlobalScope from ..task.space import InstantiationError from ..utils import get_const_tuple -from .local_executor import LocalExecutor from .measure import Builder, MeasureErrorNo, MeasureResult, Runner logger = logging.getLogger("autotvm") @@ -98,7 +98,7 @@ def __init__(self, timeout=10, n_parallel=None, build_func="default"): else: raise ValueError("Invalid build_func" + build_func) self.build_func = _WrappedBuildFunc(build_func) - self.executor = LocalExecutor(timeout=timeout) + self.executor = PopenPoolExecutor(timeout=timeout) self.tmp_dir = tempfile.mkdtemp() def build(self, measure_inputs): @@ -114,7 +114,7 @@ def build(self, measure_inputs): futures.append(ret) for future in futures: - res = future.get() + res = future.result() if isinstance(res, Exception): # timeout or fleet error, return MeasureResult directly @@ -242,7 +242,7 @@ def __init__( self.cooldown_interval = cooldown_interval self.module_loader = module_loader - self.executor = LocalExecutor(timeout=timeout * (self.n_parallel + 1)) + self.executor = PopenPoolExecutor(timeout=timeout * (self.n_parallel + 1)) @property def ref_input(self): @@ -337,7 +337,7 @@ def run(self, measure_inputs, build_results): futures.append(ret) for future in futures: - res = future.get() + res = future.result() if isinstance(res, Exception): # executor error or timeout results.append( MeasureResult( diff --git a/python/tvm/testing/__init__.py b/python/tvm/testing/__init__.py index 6d8bc70eb519..91ffa031d06c 100644 --- a/python/tvm/testing/__init__.py +++ b/python/tvm/testing/__init__.py @@ -24,7 +24,8 @@ from ._ffi_api import ErrorTest, FrontendTestModule, identity_cpp from .popen_pool import initializer, after_initializer, register_ffi, call_cpp_ffi -from .popen_pool import call_py_ffi, call_cpp_py_ffi +from .popen_pool import call_py_ffi, call_cpp_py_ffi, fast_summation, slow_summation +from .popen_pool import timeout_job from . import auto_scheduler from . import autotvm diff --git a/python/tvm/testing/popen_pool.py b/python/tvm/testing/popen_pool.py index 20345a2218fe..b646d7a89e94 100644 --- a/python/tvm/testing/popen_pool.py +++ b/python/tvm/testing/popen_pool.py @@ -16,6 +16,7 @@ # under the License. # pylint: disable=invalid-name, missing-function-docstring """Common functions for popen_pool test cases""" +import time import tvm TEST_GLOBAL_STATE_1 = 0 @@ -57,3 +58,18 @@ def call_cpp_ffi(arg): def call_cpp_py_ffi(arg): return tvm.testing.identity_cpp(arg) + + +def fast_summation(n): + return n * (n + 1) // 2 + + +def slow_summation(n): + r = 0 + for i in range(0, n + 1): + r += i + return r + + +def timeout_job(n): + time.sleep(n * 1.5) diff --git a/tests/python/contrib/test_popen_pool.py b/tests/python/contrib/test_popen_pool.py index 9ebe4c11c118..8d79ea221008 100644 --- a/tests/python/contrib/test_popen_pool.py +++ b/tests/python/contrib/test_popen_pool.py @@ -27,6 +27,9 @@ call_py_ffi, call_cpp_ffi, call_cpp_py_ffi, + fast_summation, + slow_summation, + timeout_job, ) @@ -104,8 +107,41 @@ def test_popen_ffi(): assert proc.recv() == initargs[0] +def test_popen_pool_executor_async(): + pool = PopenPoolExecutor() + f1 = pool.submit(slow_summation, 9999999) + f2 = pool.submit(fast_summation, 9999999) + t1 = 0 + t2 = 0 + while True: + if t1 == 0 and f1.done(): + t1 = time.time() + if t2 == 0 and f2.done(): + t2 = time.time() + if t1 != 0 and t2 != 0: + break + assert t2 < t1, "Expected fast async job to finish first!" + assert f1.result() == f2.result() + + +def test_popen_pool_timeout(): + timeout = 0.5 + + pool = PopenPoolExecutor(timeout=timeout) + + f1 = pool.submit(timeout_job, timeout) + while not f1.done(): + pass + try: + res = f1.result() + except Exception as ex: + assert isinstance(ex, TimeoutError) + + if __name__ == "__main__": test_popen_worker() test_popen_pool_executor() test_popen_initializer() test_popen_ffi() + test_popen_pool_executor_async() + test_popen_pool_timeout() diff --git a/tests/python/unittest/test_autotvm_executor.py b/tests/python/unittest/test_autotvm_executor.py deleted file mode 100644 index 92ad2bf765e1..000000000000 --- a/tests/python/unittest/test_autotvm_executor.py +++ /dev/null @@ -1,70 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. -"""Test local executor""" -import time - -from tvm.autotvm.measure import LocalExecutor, executor - - -def slow(n): - r = 0 - for i in range(0, n + 1): - r += i - return r - - -def fast(n): - return n * (n + 1) // 2 - - -def test_local_measure_async(): - ex = LocalExecutor() - f1 = ex.submit(slow, 9999999) - f2 = ex.submit(fast, 9999999) - t1 = 0 - t2 = 0 - while True: - if t1 == 0 and f1.done(): - t1 = time.time() - if t2 == 0 and f2.done(): - t2 = time.time() - if t1 != 0 and t2 != 0: - break - assert t2 < t1, "Expected fast async job to finish first!" - assert f1.get() == f2.get() - - -def timeout_job(n): - time.sleep(n * 1.5) - - -def test_timeout(): - timeout = 0.5 - - ex = LocalExecutor(timeout=timeout) - - f1 = ex.submit(timeout_job, timeout) - try: - res = f1.get() - except Exception as ex: - res = ex - assert isinstance(res, executor.TimeoutError) - - -if __name__ == "__main__": - test_local_measure_async() - test_timeout() diff --git a/tests/python/unittest/test_autotvm_measure.py b/tests/python/unittest/test_autotvm_measure.py index 3f56243a5137..7a72992b3ab9 100644 --- a/tests/python/unittest/test_autotvm_measure.py +++ b/tests/python/unittest/test_autotvm_measure.py @@ -18,11 +18,13 @@ import logging import multiprocessing import time +import concurrent import numpy as np import tvm from tvm import te +from tvm.autotvm.measure import executor from tvm.testing.autotvm import DummyRunner, bad_matmul, get_sample_task from tvm import autotvm from tvm.autotvm.measure.measure import MeasureErrorNo, MeasureResult @@ -76,7 +78,7 @@ def submit(self, func, *args, **kwargs): self.ran_dummy_executor = True sig = Signature.from_callable(func) assert sig.bind(*args, **kwargs).arguments["ref_input"] == refinp - return measure.local_executor.LocalFutureNoFork(None) + return executor.LocalFutureNoFork(None) runner.executor = DummyExecutor() runner.run([None], [None]) From fc1230455daebc9f451db478773c3440f154faaa Mon Sep 17 00:00:00 2001 From: shingjan Date: Thu, 26 Aug 2021 16:23:14 -0700 Subject: [PATCH 4/9] nit --- tests/python/contrib/test_popen_pool.py | 4 ++-- tests/python/unittest/test_autotvm_measure.py | 2 -- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/tests/python/contrib/test_popen_pool.py b/tests/python/contrib/test_popen_pool.py index 8d79ea221008..b3a91e176a32 100644 --- a/tests/python/contrib/test_popen_pool.py +++ b/tests/python/contrib/test_popen_pool.py @@ -124,7 +124,7 @@ def test_popen_pool_executor_async(): assert f1.result() == f2.result() -def test_popen_pool_timeout(): +def test_popen_pool_executor_timeout(): timeout = 0.5 pool = PopenPoolExecutor(timeout=timeout) @@ -144,4 +144,4 @@ def test_popen_pool_timeout(): test_popen_initializer() test_popen_ffi() test_popen_pool_executor_async() - test_popen_pool_timeout() + test_popen_pool_executor_timeout() diff --git a/tests/python/unittest/test_autotvm_measure.py b/tests/python/unittest/test_autotvm_measure.py index 7a72992b3ab9..89a8ccde065a 100644 --- a/tests/python/unittest/test_autotvm_measure.py +++ b/tests/python/unittest/test_autotvm_measure.py @@ -17,8 +17,6 @@ """Test builder and runner""" import logging import multiprocessing -import time -import concurrent import numpy as np From 0bb30ebeb16a3861957d7a9cc8e93febf6efeee9 Mon Sep 17 00:00:00 2001 From: shingjan Date: Fri, 27 Aug 2021 12:40:24 -0700 Subject: [PATCH 5/9] remove LocalFutureNoFork --- python/tvm/autotvm/measure/executor.py | 19 ------------------- tests/python/unittest/test_autotvm_measure.py | 5 ++++- 2 files changed, 4 insertions(+), 20 deletions(-) diff --git a/python/tvm/autotvm/measure/executor.py b/python/tvm/autotvm/measure/executor.py index 85362727410c..f8eca7298c64 100644 --- a/python/tvm/autotvm/measure/executor.py +++ b/python/tvm/autotvm/measure/executor.py @@ -87,25 +87,6 @@ def get(self, timeout=None): raise NotImplementedError() -class LocalFutureNoFork(Future): - """Local wrapper for the future. - This is a none-fork version of LocalFuture. - Use this for unittest and runtimes that does not support fork (like cudnn) - """ - - def __init__(self, result): - self._result = result - - def done(self): - return True - - def get(self, timeout=None): # pylint: disable=unused-argument - return self._result - - def result(self, timeout=None): # pylint: disable=unused-argument - return self._result - - class FutureError(RuntimeError): """Base error class of all future events""" diff --git a/tests/python/unittest/test_autotvm_measure.py b/tests/python/unittest/test_autotvm_measure.py index 89a8ccde065a..3ef5cbdad635 100644 --- a/tests/python/unittest/test_autotvm_measure.py +++ b/tests/python/unittest/test_autotvm_measure.py @@ -17,6 +17,7 @@ """Test builder and runner""" import logging import multiprocessing +import concurrent import numpy as np @@ -76,7 +77,9 @@ def submit(self, func, *args, **kwargs): self.ran_dummy_executor = True sig = Signature.from_callable(func) assert sig.bind(*args, **kwargs).arguments["ref_input"] == refinp - return executor.LocalFutureNoFork(None) + dummy_future = concurrent.futures.Future() + dummy_future.set_result(None) + return dummy_future runner.executor = DummyExecutor() runner.run([None], [None]) From 798c91d1da0b7ea2d8fdb1fdd8ce9f23028325b5 Mon Sep 17 00:00:00 2001 From: shingjan Date: Fri, 27 Aug 2021 17:10:21 -0700 Subject: [PATCH 6/9] exception handling --- python/tvm/autotvm/measure/measure_methods.py | 84 ++++++++----------- 1 file changed, 37 insertions(+), 47 deletions(-) diff --git a/python/tvm/autotvm/measure/measure_methods.py b/python/tvm/autotvm/measure/measure_methods.py index 59b2898feab6..2c9a93e624ac 100644 --- a/python/tvm/autotvm/measure/measure_methods.py +++ b/python/tvm/autotvm/measure/measure_methods.py @@ -114,53 +114,43 @@ def build(self, measure_inputs): futures.append(ret) for future in futures: - res = future.result() - - if isinstance(res, Exception): - # timeout or fleet error, return MeasureResult directly + try: + res = future.result() + results.append(res) + except TimeoutError as ex: results.append( MeasureResult( - (res,), MeasureErrorNo.BUILD_TIMEOUT, self.timeout, time.time() + (ex,), MeasureErrorNo.BUILD_TIMEOUT, self.timeout, time.time() ) ) - elif res.error is not None: - # instantiation error - if isinstance(res.error, InstantiationError): - results.append( - MeasureResult( - (res.error,), - MeasureErrorNo.INSTANTIATION_ERROR, - res.time_cost, - time.time(), - ) + except ChildProcessError as ex: + results.append( + MeasureResult( + (ex,), + MeasureErrorNo.RUNTIME_DEVICE, + self.timeout, + time.time(), ) - else: - if "InstantiationError" in str(res.error): - msg = str(res.error) - try: - msg = msg.split("\n")[-2].split(": ")[1] - except Exception: # pylint: disable=broad-except - pass - results.append( - MeasureResult( - (InstantiationError(msg),), - MeasureErrorNo.INSTANTIATION_ERROR, - res.time_cost, - time.time(), - ) - ) - else: # tvm error - results.append( - MeasureResult( - (res.error,), - MeasureErrorNo.COMPILE_HOST, - res.time_cost, - time.time(), - ) - ) - else: - # return BuildResult - results.append(res) + ) + except InstantiationError as ex: + results.append( + MeasureResult( + (ex,), + MeasureErrorNo.INSTANTIATION_ERROR, + None, + time.time(), + ) + ) + except Exception as ex: # pylint: disable=broad-except + # tvm error + results.append( + MeasureResult( + (ex,), + MeasureErrorNo.COMPILE_HOST, + ex.time_cost, + time.time(), + ) + ) return results @@ -337,15 +327,15 @@ def run(self, measure_inputs, build_results): futures.append(ret) for future in futures: - res = future.result() - if isinstance(res, Exception): # executor error or timeout + try: + res = future.result() + results.append(res) + except Exception as ex: # pylint: disable=broad-except results.append( MeasureResult( - (str(res),), MeasureErrorNo.RUN_TIMEOUT, self.timeout, time.time() + (str(ex),), MeasureErrorNo.RUN_TIMEOUT, self.timeout, time.time() ) ) - else: - results.append(res) return results From bc0dca121728231bf88f799c61414f634aedce36 Mon Sep 17 00:00:00 2001 From: shingjan Date: Wed, 1 Sep 2021 14:53:34 -0700 Subject: [PATCH 7/9] handling two exceptions --- python/tvm/autotvm/measure/measure_methods.py | 19 ------------------- 1 file changed, 19 deletions(-) diff --git a/python/tvm/autotvm/measure/measure_methods.py b/python/tvm/autotvm/measure/measure_methods.py index 2c9a93e624ac..0d79248d640b 100644 --- a/python/tvm/autotvm/measure/measure_methods.py +++ b/python/tvm/autotvm/measure/measure_methods.py @@ -132,25 +132,6 @@ def build(self, measure_inputs): time.time(), ) ) - except InstantiationError as ex: - results.append( - MeasureResult( - (ex,), - MeasureErrorNo.INSTANTIATION_ERROR, - None, - time.time(), - ) - ) - except Exception as ex: # pylint: disable=broad-except - # tvm error - results.append( - MeasureResult( - (ex,), - MeasureErrorNo.COMPILE_HOST, - ex.time_cost, - time.time(), - ) - ) return results From d74bc8deba1aa42e3cb6bf6e9d74c02f737e8495 Mon Sep 17 00:00:00 2001 From: shingjan Date: Wed, 1 Sep 2021 16:09:38 -0700 Subject: [PATCH 8/9] handling error --- python/tvm/autotvm/measure/measure_methods.py | 52 ++++++++++++++----- 1 file changed, 40 insertions(+), 12 deletions(-) diff --git a/python/tvm/autotvm/measure/measure_methods.py b/python/tvm/autotvm/measure/measure_methods.py index 0d79248d640b..6ae23713612b 100644 --- a/python/tvm/autotvm/measure/measure_methods.py +++ b/python/tvm/autotvm/measure/measure_methods.py @@ -116,23 +116,51 @@ def build(self, measure_inputs): for future in futures: try: res = future.result() - results.append(res) + if res.error is not None: + # instantiation error + if isinstance(res.error, InstantiationError): + res = MeasureResult( + (res.error,), + MeasureErrorNo.INSTANTIATION_ERROR, + res.time_cost, + time.time(), + ) + + else: + if "InstantiationError" in str(res.error): + msg = str(res.error) + try: + msg = msg.split("\n")[-2].split(": ")[1] + except Exception: # pylint: disable=broad-except + pass + res = MeasureResult( + (InstantiationError(msg),), + MeasureErrorNo.INSTANTIATION_ERROR, + res.time_cost, + time.time(), + ) + + else: # tvm error + res = MeasureResult( + (res.error,), + MeasureErrorNo.COMPILE_HOST, + res.time_cost, + time.time(), + ) except TimeoutError as ex: - results.append( - MeasureResult( - (ex,), MeasureErrorNo.BUILD_TIMEOUT, self.timeout, time.time() - ) + res = MeasureResult( + (ex,), MeasureErrorNo.BUILD_TIMEOUT, self.timeout, time.time() ) except ChildProcessError as ex: - results.append( - MeasureResult( - (ex,), - MeasureErrorNo.RUNTIME_DEVICE, - self.timeout, - time.time(), - ) + res = MeasureResult( + (ex,), + MeasureErrorNo.RUNTIME_DEVICE, + self.timeout, + time.time(), ) + results.append(res) + return results From 34e9d1236a4b955a07985d49d080c5b5d77f7f5b Mon Sep 17 00:00:00 2001 From: shingjan Date: Tue, 7 Sep 2021 10:18:40 -0700 Subject: [PATCH 9/9] add initiazlier --- python/tvm/autotvm/measure/measure_methods.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/python/tvm/autotvm/measure/measure_methods.py b/python/tvm/autotvm/measure/measure_methods.py index 6ae23713612b..42e046aefb4a 100644 --- a/python/tvm/autotvm/measure/measure_methods.py +++ b/python/tvm/autotvm/measure/measure_methods.py @@ -38,6 +38,7 @@ import tvm.ir.transform from tvm import nd from tvm import rpc as _rpc +from tvm.autotvm.env import AutotvmGlobalScope, reset_global_scope from tvm.contrib import ndk, nvcc, stackvm, tar from tvm.contrib.popen_pool import PopenPoolExecutor from tvm.driver import build @@ -98,7 +99,9 @@ def __init__(self, timeout=10, n_parallel=None, build_func="default"): else: raise ValueError("Invalid build_func" + build_func) self.build_func = _WrappedBuildFunc(build_func) - self.executor = PopenPoolExecutor(timeout=timeout) + self.executor = PopenPoolExecutor( + timeout=timeout, initializer=reset_global_scope, initargs=(AutotvmGlobalScope.current,) + ) self.tmp_dir = tempfile.mkdtemp() def build(self, measure_inputs): @@ -241,7 +244,11 @@ def __init__( self.cooldown_interval = cooldown_interval self.module_loader = module_loader - self.executor = PopenPoolExecutor(timeout=timeout * (self.n_parallel + 1)) + self.executor = PopenPoolExecutor( + timeout=timeout * (self.n_parallel + 1), + initializer=reset_global_scope, + initargs=(AutotvmGlobalScope.current,), + ) @property def ref_input(self):