| 556 | } |
| 557 | |
| 558 | std::unique_ptr<ral::frame::BlazingTable> CacheMachine::pullUnorderedFromCache() { |
| 559 | if (is_array_access) { |
| 560 | return this->pullFromCache(); |
| 561 | } |
| 562 | |
| 563 | CodeTimer cacheEventTimer; |
| 564 | cacheEventTimer.start(); |
| 565 | |
| 566 | std::unique_ptr<message> message_data = nullptr; |
| 567 | { // scope for lock |
| 568 | std::unique_lock<std::mutex> lock = this->waitingCache->lock(); |
| 569 | std::vector<std::unique_ptr<message>> all_messages = this->waitingCache->get_all_unsafe(); |
| 570 | std::vector<std::unique_ptr<message>> remaining_messages; |
| 571 | for(size_t i = 0; i < all_messages.size(); i++) { |
| 572 | if (all_messages[i]->get_data().get_type() == CacheDataType::GPU && message_data == nullptr){ |
| 573 | message_data = std::move(all_messages[i]); |
| 574 | } else { |
| 575 | remaining_messages.push_back(std::move(all_messages[i])); |
| 576 | } |
| 577 | } |
| 578 | this->waitingCache->put_all_unsafe(std::move(remaining_messages)); |
| 579 | } |
| 580 | if (message_data){ |
| 581 | std::string message_id = message_data->get_message_id(); |
| 582 | size_t num_rows = message_data->get_data().num_rows(); |
| 583 | size_t num_bytes = message_data->get_data().sizeInBytes(); |
| 584 | int dataType = static_cast<int>(message_data->get_data().get_type()); |
| 585 | std::unique_ptr<ral::frame::BlazingTable> output = message_data->get_data().decache(); |
| 586 | |
| 587 | cacheEventTimer.stop(); |
| 588 | if(cache_events_logger) { |
| 589 | cache_events_logger->info("{ral_id}|{query_id}|{message_id}|{cache_id}|{num_rows}|{num_bytes}|{event_type}|{timestamp_begin}|{timestamp_end}|{description}", |
| 590 | "ral_id"_a=(ctx ? ctx->getNodeIndex(ral::communication::CommunicationData::getInstance().getSelfNode()) : -1), |
| 591 | "query_id"_a=(ctx ? ctx->getContextToken() : -1), |
| 592 | "message_id"_a=message_id, |
| 593 | "cache_id"_a=cache_id, |
| 594 | "num_rows"_a=num_rows, |
| 595 | "num_bytes"_a=num_bytes, |
| 596 | "event_type"_a="PullUnorderedFromCache", |
| 597 | "timestamp_begin"_a=cacheEventTimer.start_time(), |
| 598 | "timestamp_end"_a=cacheEventTimer.end_time(), |
| 599 | "description"_a="Pull Unordered from CacheMachine type {}"_format(dataType)); |
| 600 | } |
| 601 | |
| 602 | return output; |
| 603 | } else { |
| 604 | return pullFromCache(); |
| 605 | } |
| 606 | } |
| 607 | |
| 608 | std::unique_ptr<ral::cache::CacheData> CacheMachine::pullCacheData() { |
| 609 | CodeTimer cacheEventTimer; |
no test coverage detected