| 272 | } |
| 273 | |
| 274 | void RandomShuffleQueue::TryDequeueMany(int num_elements, OpKernelContext* ctx, |
| 275 | bool allow_small_batch, |
| 276 | CallbackWithTuple callback) { |
| 277 | if (!specified_shapes()) { |
| 278 | ctx->SetStatus(errors::InvalidArgument( |
| 279 | "RandomShuffleQueue's DequeueMany and DequeueUpTo require the " |
| 280 | "components to have specified shapes.")); |
| 281 | callback(Tuple()); |
| 282 | return; |
| 283 | } |
| 284 | if (num_elements == 0) { |
| 285 | Tuple tuple; |
| 286 | tuple.reserve(num_components()); |
| 287 | for (int i = 0; i < num_components(); ++i) { |
| 288 | // TODO(josh11b,misard): Switch to allocate_output(). Problem is |
| 289 | // this breaks the abstraction boundary since we don't *really* |
| 290 | // know if and how the Tensors in the tuple we pass to callback |
| 291 | // correspond to the outputs of *ctx. For example, the |
| 292 | // ReaderRead Op uses TryDequeue() to get a filename out of a |
| 293 | // queue that is used internally by the reader and is not |
| 294 | // associated with any output of the ReaderRead. |
| 295 | // mrry@ adds: |
| 296 | // Maybe we need to pass a std::function<Tensor*(...)> (or |
| 297 | // better signature) that calls the appropriate allocator |
| 298 | // function in addition to ctx? (Or support a shim Allocator |
| 299 | // that has an internal OpKernelContext*, and dispatches to the |
| 300 | // appropriate method?) |
| 301 | // misard@ adds: |
| 302 | // I don't see that a std::function would help. The problem is |
| 303 | // that at this point (allocation time) the system doesn't know |
| 304 | // what is going to happen to the element read out of the |
| 305 | // queue. As long as we keep the generality that TensorFlow Ops |
| 306 | // do their own dynamic allocation in arbitrary C++ code, we |
| 307 | // need to preserve robustness to allocating output Tensors with |
| 308 | // the 'wrong' attributes, and fixing up with a copy. The only |
| 309 | // improvement I can see here in the future would be to support |
| 310 | // an optimized case where the queue 'knows' what attributes to |
| 311 | // use, and plumbs them through here. |
| 312 | Tensor element; |
| 313 | Status s = ctx->allocate_temp(component_dtypes_[i], ManyOutShape(i, 0), |
| 314 | &element); |
| 315 | if (!s.ok()) { |
| 316 | ctx->SetStatus(s); |
| 317 | callback(Tuple()); |
| 318 | return; |
| 319 | } |
| 320 | tuple.emplace_back(element); |
| 321 | } |
| 322 | callback(tuple); |
| 323 | return; |
| 324 | } |
| 325 | |
| 326 | CancellationManager* cm = ctx->cancellation_manager(); |
| 327 | CancellationToken token = cm->get_cancellation_token(); |
| 328 | bool already_cancelled; |
| 329 | { |
| 330 | mutex_lock l(mu_); |
| 331 | already_cancelled = !cm->RegisterCallback( |
nothing calls this directly
no test coverage detected