data-stream-recvr.h

Go to the documentation of this file.

// Copyright 2012 Cloudera Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#ifndef IMPALA_RUNTIME_DATA_STREAM_RECVR_H
#define IMPALA_RUNTIME_DATA_STREAM_RECVR_H

#include <boost/scoped_ptr.hpp>
#include <boost/thread/mutex.hpp>

#include "common/object-pool.h"
#include "common/status.h"
#include "gen-cpp/Types_types.h"   // for TUniqueId
#include "gen-cpp/Results_types.h" // for TRowBatch
#include "runtime/descriptors.h"
#include "util/tuple-row-compare.h"

namespace impala {

class DataStreamMgr;
class SortedRunMerger;
class MemTracker;
class RowBatch;
class RuntimeProfile;

//
//
class DataStreamRecvr {
 public:
  ~DataStreamRecvr();

  Status GetBatch(RowBatch** next_batch);

  void Close();

  Status CreateMerger(const TupleRowComparator& less_than);

  Status GetNext(RowBatch* output_batch, bool* eos);

  void TransferAllResources(RowBatch* transfer_batch);

  const TUniqueId& fragment_instance_id() const { return fragment_instance_id_; }
  PlanNodeId dest_node_id() const { return dest_node_id_; }
  const RowDescriptor& row_desc() const { return row_desc_; }
  MemTracker* mem_tracker() const { return mem_tracker_.get(); }

 private:
  friend class DataStreamMgr;
  class SenderQueue;

  DataStreamRecvr(DataStreamMgr* stream_mgr, MemTracker* parent_tracker,
      const RowDescriptor& row_desc, const TUniqueId& fragment_instance_id,
      PlanNodeId dest_node_id, int num_senders, bool is_merging, int total_buffer_limit,
      RuntimeProfile* profile);

  void AddBatch(const TRowBatch& thrift_batch, int sender_id);

  void RemoveSender(int sender_id);

  void CancelStream();

  bool ExceedsLimit(int batch_size) {
    return num_buffered_bytes_ + batch_size > total_buffer_limit_;
  }

  DataStreamMgr* mgr_;

  TUniqueId fragment_instance_id_;
  PlanNodeId dest_node_id_;

  int total_buffer_limit_;

  RowDescriptor row_desc_;

  bool is_merging_;

  AtomicInt<int> num_buffered_bytes_;

  boost::scoped_ptr<MemTracker> mem_tracker_;

  std::vector<SenderQueue*> sender_queues_;

  boost::scoped_ptr<SortedRunMerger> merger_;

  ObjectPool sender_queue_pool_;

  RuntimeProfile* profile_;

  RuntimeProfile::Counter* bytes_received_counter_;

  RuntimeProfile::TimeSeriesCounter* bytes_received_time_series_counter_;

  RuntimeProfile::Counter* deserialize_row_batch_timer_;

  RuntimeProfile::Counter* first_batch_wait_total_timer_;

  RuntimeProfile::Counter* buffer_full_total_timer_;

  boost::try_mutex buffer_wall_timer_lock_;

  RuntimeProfile::Counter* buffer_full_wall_timer_;

  RuntimeProfile::Counter* data_arrival_timer_;
};

}

#endif