| 483 | } |
| 484 | |
| 485 | std::unique_ptr<ral::frame::BlazingTable> CacheMachine::pullFromCache() { |
| 486 | CodeTimer cacheEventTimer; |
| 487 | cacheEventTimer.start(); |
| 488 | |
| 489 | std::string message_id; |
| 490 | |
| 491 | std::unique_ptr<message> message_data = nullptr; |
| 492 | if (is_array_access) { |
| 493 | message_id = this->cache_machine_name + "_" + std::to_string(++global_index); |
| 494 | message_data = waitingCache->get_or_wait(message_id); |
| 495 | } else { |
| 496 | message_data = waitingCache->pop_or_wait(); |
| 497 | message_id = message_data->get_message_id(); |
| 498 | } |
| 499 | |
| 500 | if (message_data == nullptr) { |
| 501 | return nullptr; |
| 502 | } |
| 503 | |
| 504 | size_t num_rows = message_data->get_data().num_rows(); |
| 505 | size_t num_bytes = message_data->get_data().sizeInBytes(); |
| 506 | int dataType = static_cast<int>(message_data->get_data().get_type()); |
| 507 | std::unique_ptr<ral::frame::BlazingTable> output = message_data->get_data().decache(); |
| 508 | |
| 509 | cacheEventTimer.stop(); |
| 510 | if(cache_events_logger) { |
| 511 | cache_events_logger->info("{ral_id}|{query_id}|{message_id}|{cache_id}|{num_rows}|{num_bytes}|{event_type}|{timestamp_begin}|{timestamp_end}|{description}", |
| 512 | "ral_id"_a=(ctx ? ctx->getNodeIndex(ral::communication::CommunicationData::getInstance().getSelfNode()) : -1), |
| 513 | "query_id"_a=(ctx ? ctx->getContextToken() : -1), |
| 514 | "message_id"_a=message_id, |
| 515 | "cache_id"_a=cache_id, |
| 516 | "num_rows"_a=num_rows, |
| 517 | "num_bytes"_a=num_bytes, |
| 518 | "event_type"_a="PullFromCache", |
| 519 | "timestamp_begin"_a=cacheEventTimer.start_time(), |
| 520 | "timestamp_end"_a=cacheEventTimer.end_time(), |
| 521 | "description"_a="Pull from CacheMachine type {}"_format(dataType)); |
| 522 | } |
| 523 | |
| 524 | return output; |
| 525 | } |
| 526 | |
| 527 | |
| 528 | std::unique_ptr<ral::cache::CacheData> CacheMachine::pullCacheData(std::string message_id) { |