| 299 | } |
| 300 | |
| 301 | void StarWorker::StarRunGraphAsync(StarRunGraphRequest* request, |
| 302 | StarRunGraphResponse* response, |
| 303 | StatusCallback done) { |
| 304 | // Enable prmalloc optimization |
| 305 | ScopedMemoryCollector scoped_memory_collector; |
| 306 | |
| 307 | CallOptions* opts = &(request->opts_); |
| 308 | const int64 step_id = request->step_id_; |
| 309 | WorkerSession* session = env_->session_mgr->LegacySession().get(); |
| 310 | |
| 311 | if (env()->lockless) { |
| 312 | if (pending_graph_count_.find(step_id) == |
| 313 | pending_graph_count_.end()) { |
| 314 | pending_graph_count_[step_id] = request->ps_graph_count_; |
| 315 | } |
| 316 | } else { |
| 317 | mutex_lock l(graph_count_mu_); |
| 318 | if (pending_graph_count_.find(step_id) == |
| 319 | pending_graph_count_.end()) { |
| 320 | pending_graph_count_[step_id] = request->ps_graph_count_; |
| 321 | } |
| 322 | } |
| 323 | |
| 324 | static std::atomic<int64> global_step_id(0); |
| 325 | auto current_step_id = global_step_id.fetch_add(1); |
| 326 | |
| 327 | static int64 last_time = 0; |
| 328 | if (current_step_id % 100000 == 0) { |
| 329 | int64_t cur_time = Env::Default()->NowMicros(); |
| 330 | if (last_time != 0) { |
| 331 | double qps = 100000 / ((cur_time - last_time) / 1000000.0); |
| 332 | VLOG(1) << "zero copy run graph qps:" << qps; |
| 333 | } |
| 334 | last_time = cur_time; |
| 335 | } |
| 336 | |
| 337 | std::map<std::string, bool> is_send_dead; |
| 338 | GraphMgr::NamedTensors in; |
| 339 | GraphMgr::NamedTensors* out = new GraphMgr::NamedTensors; |
| 340 | static Tensor empty_tensor(DT_FLOAT); |
| 341 | for (size_t i = 0; i < request->feed_tensors_.size(); ++i) { |
| 342 | in.insert({request->feed_names_[i], request->feed_tensors_[i]}); |
| 343 | is_send_dead.insert({request->feed_names_[i], request->is_dead_[i]}); |
| 344 | } |
| 345 | |
| 346 | for (size_t i = 0; i < request->fetch_names_.size(); ++i) { |
| 347 | out->insert({request->fetch_names_[i], empty_tensor}); |
| 348 | } |
| 349 | |
| 350 | // TODO: No collector, No GPU, No execution options |
| 351 | ExecutorOpts exec_opts; |
| 352 | StepStatsCollector* collector = nullptr; |
| 353 | |
| 354 | CancellationManager* cm = new CancellationManager; |
| 355 | opts->SetCancelCallback([this, cm, step_id]() { |
| 356 | cm->StartCancel(); |
| 357 | AbortStep(step_id); |
| 358 | }); |
no test coverage detected