disk-io-mgr-internal.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_DISK_IO_MGR_INTERNAL_H
#define IMPALA_RUNTIME_DISK_IO_MGR_INTERNAL_H

#include "disk-io-mgr.h"
#include <queue>
#include <boost/thread/locks.hpp>
#include <unistd.h>
#include <gutil/strings/substitute.h>

#include "common/logging.h"
#include "runtime/mem-tracker.h"
#include "runtime/thread-resource-mgr.h"
#include "util/cpu-info.h"
#include "util/debug-util.h"
#include "util/disk-info.h"
#include "util/hdfs-util.h"
#include "util/filesystem-util.h"
#include "util/impalad-metrics.h"

namespace impala {

struct DiskIoMgr::DiskQueue {
  int disk_id;

  boost::mutex lock;

  boost::condition_variable work_available;

  std::list<RequestContext*> request_contexts;

  inline void EnqueueContext(RequestContext* worker) {
    {
      boost::unique_lock<boost::mutex> disk_lock(lock);
      DCHECK(find(request_contexts.begin(), request_contexts.end(), worker) ==
          request_contexts.end());
      request_contexts.push_back(worker);
    }
    work_available.notify_all();
  }

  DiskQueue(int id) : disk_id(id) { }
};

//
//
//
//
//
//
class DiskIoMgr::RequestContext {
 public:
  enum State {
    Active,

    Cancelled,

    Inactive,
  };

  RequestContext(DiskIoMgr* parent, int num_disks);

  void Reset(MemTracker* tracker);

  void DecrementDiskRefCount() {
    // boost doesn't let us dcheck that the reader lock is taken
    DCHECK_GT(num_disks_with_ranges_, 0);
    if (--num_disks_with_ranges_ == 0) {
      disks_complete_cond_var_.notify_one();
    }
    DCHECK(Validate()) << std::endl << DebugString();
  }


  void ScheduleScanRange(DiskIoMgr::ScanRange* range) {
    DCHECK_EQ(state_, Active);
    DCHECK(range != NULL);
    RequestContext::PerDiskState& state = disk_states_[range->disk_id()];
    state.in_flight_ranges()->Enqueue(range);
    state.ScheduleContext(this, range->disk_id());
  }

  void Cancel(const Status& status);

  void AddRequestRange(DiskIoMgr::RequestRange* range, bool schedule_immediately);

  int initial_scan_range_queue_capacity() const { return initial_queue_capacity_; }

  bool Validate() const;

  std::string DebugString() const;

 private:
  friend class DiskIoMgr;
  class PerDiskState;

  DiskIoMgr* parent_;

  MemTracker* mem_tracker_;

  RuntimeProfile::Counter* bytes_read_counter_;

  RuntimeProfile::Counter* read_timer_;

  RuntimeProfile::Counter* active_read_thread_counter_;

  RuntimeProfile::Counter* disks_accessed_bitmap_;

  AtomicInt<int64_t> bytes_read_local_;

  AtomicInt<int64_t> bytes_read_short_circuit_;

  AtomicInt<int64_t> bytes_read_dn_cache_;

  AtomicInt<int64_t> unexpected_remote_bytes_;

  AtomicInt<int> num_buffers_in_reader_;

  AtomicInt<int> num_finished_ranges_;

  AtomicInt<int> num_remote_ranges_;

  AtomicInt<int> num_unstarted_scan_ranges_;

  AtomicInt<int> num_used_buffers_;

  AtomicInt<int> num_ready_buffers_;

  AtomicInt<int> total_range_queue_capacity_;

  int initial_queue_capacity_;

  boost::mutex lock_;

  State state_;

  Status status_;

  int num_disks_with_ranges_;

  InternalQueue<ScanRange> cached_ranges_;

  InternalQueue<ScanRange> ready_to_start_ranges_;
  boost::condition_variable ready_to_start_ranges_cv_;  // used with lock_

  InternalQueue<ScanRange> blocked_ranges_;

  boost::condition_variable disks_complete_cond_var_;

  class PerDiskState {
   public:
    bool done() const { return done_; }
    void set_done(bool b) { done_ = b; }

    int num_remaining_ranges() const { return num_remaining_ranges_; }
    int& num_remaining_ranges() { return num_remaining_ranges_; }

    ScanRange* next_scan_range_to_start() { return next_scan_range_to_start_; }
    void set_next_scan_range_to_start(ScanRange* range) {
      next_scan_range_to_start_ = range;
    }

    bool is_on_queue() const {
      bool b = is_on_queue_;
      __sync_synchronize();
      return b;
    }

    int num_threads_in_op() const {
      int v = num_threads_in_op_;
      __sync_synchronize();
      return v;
    }

    const InternalQueue<ScanRange>* unstarted_scan_ranges() const {
      return &unstarted_scan_ranges_;
    }
    const InternalQueue<WriteRange>* unstarted_write_ranges() const {
      return &unstarted_write_ranges_;
    }
    const InternalQueue<RequestRange>* in_flight_ranges() const {
      return &in_flight_ranges_;
    }

    InternalQueue<ScanRange>* unstarted_scan_ranges() { return &unstarted_scan_ranges_; }
    InternalQueue<WriteRange>* unstarted_write_ranges() {
      return &unstarted_write_ranges_;
    }
    InternalQueue<RequestRange>* in_flight_ranges() { return &in_flight_ranges_; }

    PerDiskState() {
      Reset();
    }

    void ScheduleContext(RequestContext* context, int disk_id) {
      if (!is_on_queue_ && !done_) {
        is_on_queue_ = true;
        context->parent_->disk_queues_[disk_id]->EnqueueContext(context);
      }
    }

    void IncrementRequestThreadAndDequeue() {
      ++num_threads_in_op_;
      is_on_queue_ = false;
    }

    void DecrementRequestThread() {
      --num_threads_in_op_;
    }

    void DecrementRequestThreadAndCheckDone(RequestContext* context) {
      --num_threads_in_op_;
      // We don't need to worry about reordered loads here because updating
      // num_threads_in_request_ uses an atomic, which is a barrier.
      if (!is_on_queue_ && num_threads_in_op_ == 0 && !done_) {
        // This thread is the last one for this reader on this disk, do final cleanup
        context->DecrementDiskRefCount();
        done_ = true;
      }
    }

    void Reset() {
      DCHECK(in_flight_ranges_.empty());
      DCHECK(unstarted_scan_ranges_.empty());
      DCHECK(unstarted_write_ranges_.empty());

      done_ = true;
      num_remaining_ranges_ = 0;
      is_on_queue_ = false;
      num_threads_in_op_ = 0;
      next_scan_range_to_start_ = NULL;
    }

   private:
    bool done_;

    bool is_on_queue_;

    int num_remaining_ranges_;

    InternalQueue<ScanRange> unstarted_scan_ranges_;

    InternalQueue<RequestRange> in_flight_ranges_;

    ScanRange* next_scan_range_to_start_;

    AtomicInt<int> num_threads_in_op_;

    InternalQueue<WriteRange> unstarted_write_ranges_;
  };

  std::vector<PerDiskState> disk_states_;
};

}

#endif