| 390 | |
| 391 | template <typename TaskType> |
| 392 | Status SharedBatchScheduler<TaskType>::AddQueue( |
| 393 | const QueueOptions& options, |
| 394 | std::function<void(std::unique_ptr<Batch<TaskType>>)> |
| 395 | process_batch_callback, |
| 396 | std::unique_ptr<BatchScheduler<TaskType>>* queue) { |
| 397 | if (options.max_batch_size == 0) { |
| 398 | return errors::InvalidArgument("max_batch_size must be positive; was ", |
| 399 | options.max_batch_size); |
| 400 | } |
| 401 | if (options.batch_timeout_micros < 0) { |
| 402 | return errors::InvalidArgument( |
| 403 | "batch_timeout_micros must be non-negative; was ", |
| 404 | options.batch_timeout_micros); |
| 405 | } |
| 406 | if (options.max_enqueued_batches < 0) { |
| 407 | return errors::InvalidArgument( |
| 408 | "max_enqueued_batches must be non-negative; was ", |
| 409 | options.max_enqueued_batches); |
| 410 | } |
| 411 | |
| 412 | auto schedulable_batch_callback = [this] { |
| 413 | mutex_lock l(mu_); |
| 414 | schedulable_batch_cv_.notify_one(); |
| 415 | }; |
| 416 | auto internal_queue = |
| 417 | std::unique_ptr<internal::Queue<TaskType>>(new internal::Queue<TaskType>( |
| 418 | options, options_.env, process_batch_callback, |
| 419 | schedulable_batch_callback)); |
| 420 | auto handle = std::unique_ptr<BatchScheduler<TaskType>>( |
| 421 | new internal::QueueHandle<TaskType>(this->shared_from_this(), |
| 422 | internal_queue.get())); |
| 423 | { |
| 424 | mutex_lock l(mu_); |
| 425 | queues_.push_back(std::move(internal_queue)); |
| 426 | if (next_queue_to_schedule_ == queues_.end()) { |
| 427 | next_queue_to_schedule_ = queues_.begin(); |
| 428 | } |
| 429 | } |
| 430 | *queue = std::move(handle); |
| 431 | return Status::OK(); |
| 432 | } |
| 433 | |
| 434 | template <typename TaskType> |
| 435 | SharedBatchScheduler<TaskType>::SharedBatchScheduler(const Options& options) |