| 593 | } |
| 594 | |
| 595 | Status KrpcDataStreamSender::Channel::TransmitData( |
| 596 | unique_ptr<OutboundRowBatch>* outbound_batch, bool swap_batch) { |
| 597 | VLOG_ROW << "Channel::TransmitData() fragment_instance_id=" |
| 598 | << PrintId(fragment_instance_id_) << " dest_node=" << dest_node_id_ |
| 599 | << " #rows=" << outbound_batch->get()->header()->num_rows(); |
| 600 | std::unique_lock<SpinLock> l(lock_); |
| 601 | RETURN_IF_ERROR(WaitForRpcLocked(&l)); |
| 602 | // Measure the time needed after getting the lock. |
| 603 | SCOPED_TIMER(parent_->transmit_data_timer_); |
| 604 | DCHECK(!rpc_in_flight_); |
| 605 | DCHECK(rpc_in_flight_batch_ == nullptr); |
| 606 | // If the remote receiver is closed already, there is no point in sending anything. |
| 607 | // TODO: Needs better solution for IMPALA-3990 in the long run. |
| 608 | if (UNLIKELY(remote_recvr_closed_)) return Status::OK(); |
| 609 | rpc_in_flight_ = true; |
| 610 | rpc_in_flight_batch_ = outbound_batch->get(); |
| 611 | RETURN_IF_ERROR(DoTransmitDataRpc()); |
| 612 | // At this point the previous RPC must be already finished and the previous buffer |
| 613 | // in outbound_batch_ can be reused. |
| 614 | DCHECK_NE(rpc_in_flight_batch_, outbound_batch_.get()); |
| 615 | if (swap_batch) outbound_batch_.swap(*outbound_batch); |
| 616 | return Status::OK(); |
| 617 | } |
| 618 | |
| 619 | Status KrpcDataStreamSender::Channel::SerializeAndSendBatch(RowBatch* batch) { |
| 620 | unique_ptr<OutboundRowBatch>* serialization_batch = GetSerializationBatch(); |