| 2238 | } |
| 2239 | |
| 2240 | Status MasterSession::DoRunCallable(CallOptions* opts, ReffedClientGraph* rcg, |
| 2241 | const RunCallableRequest& req, |
| 2242 | RunCallableResponse* resp) { |
| 2243 | VLOG(2) << "DoRunCallable req: " << req.DebugString(); |
| 2244 | PerStepState pss; |
| 2245 | pss.start_micros = Env::Default()->NowMicros(); |
| 2246 | auto cleanup = gtl::MakeCleanup([this] { MarkRunCompletion(); }); |
| 2247 | |
| 2248 | // Prepare. |
| 2249 | int64 count = rcg->get_and_increment_execution_count(); |
| 2250 | |
| 2251 | const uint64 step_id = NewStepId(rcg->collective_graph_key()); |
| 2252 | TRACEPRINTF("stepid %llu", step_id); |
| 2253 | |
| 2254 | const RunOptions& run_options = rcg->callable_options().run_options(); |
| 2255 | |
| 2256 | if (run_options.timeout_in_ms() != 0) { |
| 2257 | opts->SetTimeout(run_options.timeout_in_ms()); |
| 2258 | } |
| 2259 | |
| 2260 | std::unique_ptr<ProfileHandler> ph; |
| 2261 | FillPerStepState(rcg, run_options, step_id, count, &pss, &ph); |
| 2262 | Status s = rcg->RunPartitions(env_, step_id, count, &pss, opts, req, resp, |
| 2263 | &cancellation_manager_); |
| 2264 | cleanup.release(); // MarkRunCompletion called in PostRunCleanup(). |
| 2265 | return PostRunCleanup(rcg, step_id, run_options, &pss, ph, s, |
| 2266 | resp->mutable_metadata()); |
| 2267 | } |
| 2268 | |
| 2269 | Status MasterSession::RunCallable(CallOptions* opts, |
| 2270 | const RunCallableRequest& req, |
nothing calls this directly
no test coverage detected