| 44 | } |
| 45 | |
| 46 | Status QueueRunner::Init(const QueueRunnerDef& queue_runner_def) { |
| 47 | queue_name_ = queue_runner_def.queue_name(); |
| 48 | enqueue_op_names_.clear(); |
| 49 | enqueue_op_names_.insert(enqueue_op_names_.end(), |
| 50 | queue_runner_def.enqueue_op_name().begin(), |
| 51 | queue_runner_def.enqueue_op_name().end()); |
| 52 | size_t op_names_size = enqueue_op_names_.size(); |
| 53 | if (op_names_size > kint32max) { |
| 54 | return Status(error::INVALID_ARGUMENT, |
| 55 | "Enqueue ops to run cannot exceed kint32max"); |
| 56 | } |
| 57 | runs_ = static_cast<int>(op_names_size); |
| 58 | if (runs_ == 0) { |
| 59 | return Status(error::INVALID_ARGUMENT, "Empty enqueue ops to run."); |
| 60 | } |
| 61 | close_op_name_ = queue_runner_def.close_op_name(); |
| 62 | cancel_op_name_ = queue_runner_def.cancel_op_name(); |
| 63 | if (queue_runner_def.queue_closed_exception_types_size() == 0) { |
| 64 | queue_closed_exception_types_.insert(error::OUT_OF_RANGE); |
| 65 | } else { |
| 66 | for (const auto& code : queue_runner_def.queue_closed_exception_types()) { |
| 67 | queue_closed_exception_types_.insert(static_cast<int>(code)); |
| 68 | } |
| 69 | } |
| 70 | |
| 71 | int nthreads = runs_; |
| 72 | if (coord_) { |
| 73 | // One more thread to call Stop() |
| 74 | nthreads++; |
| 75 | } |
| 76 | thread_pool_.reset(new thread::ThreadPool( |
| 77 | Env::Default(), SanitizeThreadSuffix(queue_name_), nthreads)); |
| 78 | |
| 79 | return Status::OK(); |
| 80 | } |
| 81 | |
| 82 | QueueRunner::~QueueRunner() { |
| 83 | // Cannot run Stop() here because the session might already be closed or |