| 270 | namespace { |
| 271 | |
| 272 | class TimedFuncInvokerImpl final : public TimedFuncInvoker { |
| 273 | /* |
| 274 | * server-client protocol: |
| 275 | * |
| 276 | * server is the main megbrain process which calls invoke() |
| 277 | * |
| 278 | * client is the worker process that executes the function and may get |
| 279 | * killed |
| 280 | * |
| 281 | * s: hello: rand uint32 |
| 282 | * c: hello + 1 |
| 283 | * |
| 284 | * while true: |
| 285 | * s: func id, func arg len <size_t>, func arg |
| 286 | * c: init_done<uint8:1>, err<bool>, func result len <size_t>, |
| 287 | * func result; if error happens, err would be true and result is |
| 288 | * the error message |
| 289 | */ |
| 290 | struct FuncRegistry { |
| 291 | Func func; |
| 292 | FuncInit init; |
| 293 | |
| 294 | Result direct_call(const Param& param) const { |
| 295 | if (init) |
| 296 | init(param); |
| 297 | return func(param); |
| 298 | } |
| 299 | }; |
| 300 | static constexpr uint8_t INIT_DONE_FLAG = 23; |
| 301 | ForkExecImpl m_fork_exec_impl; |
| 302 | pid_t m_worker_pid = 0; |
| 303 | int m_sock_fd = 0, m_peer_fd = 0, m_sock_name_cnt = 0; |
| 304 | ThinHashMap<FuncId, FuncRegistry> m_func_registry; |
| 305 | |
| 306 | bool m_watcher_should_stop = false; |
| 307 | std::condition_variable m_watcher_stop_cv; |
| 308 | MGB_MUTEX m_watcher_stop_mtx, m_global_mtx; |
| 309 | |
| 310 | void clear_sock_fd() { |
| 311 | if (m_peer_fd) |
| 312 | close(m_peer_fd); |
| 313 | if (m_sock_fd && m_sock_fd != m_peer_fd) |
| 314 | close(m_sock_fd); |
| 315 | m_sock_fd = m_peer_fd = 0; |
| 316 | } |
| 317 | |
| 318 | void set_fork_exec_impl(const ForkExecImpl& impl) override { |
| 319 | mgb_assert(!m_fork_exec_impl); |
| 320 | m_fork_exec_impl = impl; |
| 321 | } |
| 322 | |
| 323 | //! create an abstract AF_UNIX socket and bind to it |
| 324 | void create_sock_and_bind( |
| 325 | const char* name, int (*do_bind)(int, const sockaddr*, socklen_t)) { |
| 326 | clear_sock_fd(); |
| 327 | |
| 328 | m_sock_fd = socket(AF_UNIX, SOCK_STREAM, 0); |
| 329 | CHECK_SYS_ERR(m_sock_fd); |