gaocegege
commented
4 years ago 自称的下一代超参数优化框架，特点主要是：
- Define-by-run programming that allows the user to dynamically construct the search space,
- Efficient sampling algorithm and pruning algorithm that allows some user-customization,
- Easy-to-setup, versatile architecture that can be deployed for tasks of various types, ranging from light-weight experiments conducted via interactive interfaces to heavy-weight distributed computations.
值得一提的是，它的 Distributed 训练，是用 Python 的 multiprocess 实现的，更像是并发而非并行。哦多进程称为并行也没问题。但是它没法解决单机的资源瓶颈问题。
https://github.com/pfnet/optuna/blob/451fe257342c50a11940ec380f06f60f94324237/optuna/study.py#L416
def _optimize_parallel(
self,
func, # type: ObjectiveFuncType
n_trials, # type: Optional[int]
timeout, # type: Optional[float]
n_jobs, # type: int
catch, # type: Union[Tuple[()], Tuple[Type[Exception]]]
callbacks # type: Optional[List[Callable[[Study, structs.FrozenTrial], None]]]
):
# type: (...) -> None
self.start_datetime = datetime.datetime.now()
if n_jobs == -1:
n_jobs = multiprocessing.cpu_count()
if n_trials is not None:
# The number of threads needs not to be larger than trials.
n_jobs = min(n_jobs, n_trials)
if n_trials == 0:
return # When n_jobs is zero, ThreadPool fails.
pool = multiprocessing.pool.ThreadPool(n_jobs) # type: ignore
# A queue is passed to each thread. When True is received, then the thread continues
# the evaluation. When False is received, then it quits.
def func_child_thread(que):
# type: (Queue) -> None
while que.get():
self._run_trial_and_callbacks(func, catch, callbacks)
self._storage.remove_session()
que = multiprocessing.Queue(maxsize=n_jobs) # type: ignore
for _ in range(n_jobs):
que.put(True)
n_enqueued_trials = n_jobs
imap_ite = pool.imap(func_child_thread, [que] * n_jobs, chunksize=1)
while True:
if timeout is not None:
elapsed_timedelta = datetime.datetime.now() - self.start_datetime
elapsed_seconds = elapsed_timedelta.total_seconds()
if elapsed_seconds > timeout:
break
if n_trials is not None:
if n_enqueued_trials >= n_trials:
break
try:
que.put_nowait(True)
n_enqueued_trials += 1
except queue.Full:
time.sleep(1)
for _ in range(n_jobs):
que.put(False)
collections.deque(imap_ite, maxlen=0) # Consume the iterator to wait for all threads.
pool.terminate()
que.close()
que.join_thread()
https://arxiv.org/pdf/1907.10902.pdf
https://github.com/pfnet/optuna/
SIGKDD'19 Applied Data Science Track Papers