| 473 | } |
| 474 | |
| 475 | Maybe<Result> invoke(FuncId id, const Param& param, double timeout) override { |
| 476 | MGB_LOCK_GUARD(m_global_mtx); |
| 477 | mgb_assert(timeout >= 0); |
| 478 | auto iter = m_func_registry.find(id); |
| 479 | mgb_assert(iter != m_func_registry.end(), "id %zu does not exist", id); |
| 480 | if (!timeout && !check_worker_alive()) |
| 481 | return iter->second.direct_call(param); |
| 482 | |
| 483 | if (!m_fork_exec_impl) { |
| 484 | mgb_log_debug( |
| 485 | "timeout is set, but no fork_exec_impl not given; " |
| 486 | "timeout would be ignored"); |
| 487 | return iter->second.direct_call(param); |
| 488 | } |
| 489 | |
| 490 | // start worker and write init param; reading init_done sometimes fails |
| 491 | // with connection reset, so we retry for some times |
| 492 | constexpr int MAX_TRY = 5; |
| 493 | for (int cur_try = 0; cur_try < MAX_TRY; ++cur_try) { |
| 494 | ensure_worker_alive(); |
| 495 | write_pod(id); |
| 496 | write_pod(param.size); |
| 497 | write(param.data, param.size); |
| 498 | std::remove_cv_t<decltype(INIT_DONE_FLAG)> init_done; |
| 499 | if (!read(&init_done, sizeof(init_done), false)) { |
| 500 | mgb_assert(cur_try < MAX_TRY - 1, "can not read init_done flag"); |
| 501 | kill_worker(); |
| 502 | continue; |
| 503 | } |
| 504 | mgb_assert(init_done == INIT_DONE_FLAG); |
| 505 | break; |
| 506 | } |
| 507 | m_watcher_should_stop = false; |
| 508 | |
| 509 | std::future<bool> watcher; |
| 510 | if (timeout) { |
| 511 | watcher = std::async( |
| 512 | std::launch::async, &TimedFuncInvokerImpl::watcher_impl, this, |
| 513 | timeout); |
| 514 | } |
| 515 | |
| 516 | // stop watcher, return whether worker killed by watcher |
| 517 | auto stop_watcher = [&]() { |
| 518 | if (!timeout) |
| 519 | return false; |
| 520 | |
| 521 | { |
| 522 | MGB_LOCK_GUARD(m_watcher_stop_mtx); |
| 523 | m_watcher_should_stop = true; |
| 524 | m_watcher_stop_cv.notify_all(); |
| 525 | } |
| 526 | return watcher.get(); |
| 527 | }; |
| 528 | |
| 529 | auto read_safe = [&](void* dest, size_t size) { |
| 530 | if (!read(dest, size, false)) { |
| 531 | if (!stop_watcher()) |
| 532 | kill_worker(); |