You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2017/11/18 00:31:55 UTC
[12/16] incubator-impala git commit: IMPALA-4835 (prep only): create
io subfolder and namespace
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/b840137c/be/src/runtime/disk-io-mgr.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/disk-io-mgr.h b/be/src/runtime/disk-io-mgr.h
deleted file mode 100644
index 49de0ff..0000000
--- a/be/src/runtime/disk-io-mgr.h
+++ /dev/null
@@ -1,972 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you 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_H
-#define IMPALA_RUNTIME_DISK_IO_MGR_H
-
-#include <deque>
-#include <functional>
-#include <vector>
-
-#include <boost/scoped_ptr.hpp>
-#include <boost/unordered_set.hpp>
-#include <boost/thread/mutex.hpp>
-
-#include "common/atomic.h"
-#include "common/hdfs.h"
-#include "common/object-pool.h"
-#include "common/status.h"
-#include "runtime/disk-io-mgr-handle-cache.h"
-#include "runtime/thread-resource-mgr.h"
-#include "util/aligned-new.h"
-#include "util/bit-util.h"
-#include "util/condition-variable.h"
-#include "util/error-util.h"
-#include "util/internal-queue.h"
-#include "util/runtime-profile.h"
-#include "util/thread.h"
-
-namespace impala {
-
-class MemTracker;
-
-/// Manager object that schedules IO for all queries on all disks and remote filesystems
-/// (such as S3). Each query maps to one or more DiskIoRequestContext objects, each of which
-/// has its own queue of scan ranges and/or write ranges.
-//
-/// The API splits up requesting scan/write ranges (non-blocking) and reading the data
-/// (blocking). The DiskIoMgr has worker threads that will read from and write to
-/// disk/hdfs/remote-filesystems, allowing interleaving of IO and CPU. This allows us to
-/// keep all disks and all cores as busy as possible.
-//
-/// All public APIs are thread-safe. It is not valid to call any of the APIs after
-/// UnregisterContext() returns.
-//
-/// For Readers:
-/// We can model this problem as a multiple producer (threads for each disk), multiple
-/// consumer (scan ranges) problem. There are multiple queues that need to be
-/// synchronized. Conceptually, there are two queues:
-/// 1. The per disk queue: this contains a queue of readers that need reads.
-/// 2. The per scan range ready-buffer queue: this contains buffers that have been
-/// read and are ready for the caller.
-/// The disk queue contains a queue of readers and is scheduled in a round robin fashion.
-/// Readers map to scan nodes. The reader then contains a queue of scan ranges. The caller
-/// asks the IoMgr for the next range to process. The IoMgr then selects the best range
-/// to read based on disk activity and begins reading and queuing buffers for that range.
-/// TODO: We should map readers to queries. A reader is the unit of scheduling and queries
-/// that have multiple scan nodes shouldn't have more 'turns'.
-//
-/// For Writers:
-/// Data is written via AddWriteRange(). This is non-blocking and adds a WriteRange to a
-/// per-disk queue. After the write is complete, a callback in WriteRange is invoked.
-/// No memory is allocated within IoMgr for writes and no copies are made. It is the
-/// responsibility of the client to ensure that the data to be written is valid and that
-/// the file to be written to exists until the callback is invoked.
-//
-/// The IoMgr provides three key APIs.
-/// 1. AddScanRanges: this is non-blocking and tells the IoMgr all the ranges that
-/// will eventually need to be read.
-/// 2. GetNextRange: returns to the caller the next scan range it should process.
-/// This is based on disk load. This also begins reading the data in this scan
-/// range. This is blocking.
-/// 3. ScanRange::GetNext: returns the next buffer for this range. This is blocking.
-//
-/// The disk threads do not synchronize with each other. The readers and writers don't
-/// synchronize with each other. There is a lock and condition variable for each request
-/// context queue and each disk queue.
-/// IMPORTANT: whenever both locks are needed, the lock order is to grab the context lock
-/// before the disk lock.
-//
-/// Scheduling: If there are multiple request contexts with work for a single disk, the
-/// request contexts are scheduled in round-robin order. Multiple disk threads can
-/// operate on the same request context. Exactly one request range is processed by a
-/// disk thread at a time. If there are multiple scan ranges scheduled via
-/// GetNextRange() for a single context, these are processed in round-robin order.
-/// If there are multiple scan and write ranges for a disk, a read is always followed
-/// by a write, and a write is followed by a read, i.e. reads and writes alternate.
-/// If multiple write ranges are enqueued for a single disk, they will be processed
-/// by the disk threads in order, but may complete in any order. No guarantees are made
-/// on ordering of writes across disks.
-//
-/// Resource Management: effective resource management in the IoMgr is key to good
-/// performance. The IoMgr helps coordinate two resources: CPU and disk. For CPU,
-/// spinning up too many threads causes thrashing.
-/// Memory usage in the IoMgr comes from queued read buffers. If we queue the minimum
-/// (i.e. 1), then the disks are idle while we are processing the buffer. If we don't
-/// limit the queue, then it possible we end up queueing the entire data set (i.e. CPU
-/// is slower than disks) and run out of memory.
-/// For both CPU and memory, we want to model the machine as having a fixed amount of
-/// resources. If a single query is running, it should saturate either CPU or Disk
-/// as well as using as little memory as possible. With multiple queries, each query
-/// should get less CPU. In that case each query will need fewer queued buffers and
-/// therefore have less memory usage.
-//
-/// The IoMgr defers CPU management to the caller. The IoMgr provides a GetNextRange
-/// API which will return the next scan range the caller should process. The caller
-/// can call this from the desired number of reading threads. Once a scan range
-/// has been returned via GetNextRange, the IoMgr will start to buffer reads for
-/// that range and it is expected the caller will pull those buffers promptly. For
-/// example, if the caller would like to have 1 scanner thread, the read loop
-/// would look like:
-/// while (more_ranges)
-/// range = GetNextRange()
-/// while (!range.eosr)
-/// buffer = range.GetNext()
-/// To have multiple reading threads, the caller would simply spin up the threads
-/// and each would process the loops above.
-//
-/// To control the number of IO buffers, each scan range has a limit of two queued
-/// buffers (SCAN_RANGE_READY_BUFFER_LIMIT). If the number of buffers is at capacity,
-/// the IoMgr will no longer read for that scan range until the caller has processed
-/// a buffer. Assuming the client returns each buffer before requesting the next one
-/// from the scan range, then this will consume up to 3 * 8MB = 24MB of I/O buffers per
-/// scan range.
-//
-/// Buffer Management:
-/// Buffers for reads are either a) allocated by the IoMgr and transferred to the caller,
-/// b) cached HDFS buffers if the scan range uses HDFS caching, or c) provided by the
-/// caller when constructing the scan range.
-///
-/// As a caller reads from a scan range, these buffers are wrapped in BufferDescriptors
-/// and returned to the caller. The caller must always call ReturnBuffer() on the buffer
-/// descriptor to allow recycling of the associated buffer (if there is an
-/// IoMgr-allocated or HDFS cached buffer).
-///
-/// Caching support:
-/// Scan ranges contain metadata on whether or not it is cached on the DN. In that
-/// case, we use the HDFS APIs to read the cached data without doing any copies. For these
-/// ranges, the reads happen on the caller thread (as opposed to the disk threads).
-/// It is possible for the cached read APIs to fail, in which case the ranges are then
-/// queued on the disk threads and behave identically to the case where the range
-/// is not cached.
-/// Resources for these ranges are also not accounted against the reader because none
-/// are consumed.
-/// While a cached block is being processed, the block is mlocked. We want to minimize
-/// the time the mlock is held.
-/// - HDFS will time us out if we hold onto the mlock for too long
-/// - Holding the lock prevents uncaching this file due to a caching policy change.
-/// Therefore, we only issue the cached read when the caller is ready to process the
-/// range (GetNextRange()) instead of when the ranges are issued. This guarantees that
-/// there will be a CPU available to process the buffer and any throttling we do with
-/// the number of scanner threads properly controls the amount of files we mlock.
-/// With cached scan ranges, we cannot close the scan range until the cached buffer
-/// is returned (HDFS does not allow this). We therefore need to defer the close until
-/// the cached buffer is returned (ReturnBuffer()).
-//
-/// Remote filesystem support (e.g. S3):
-/// Remote filesystems are modeled as "remote disks". That is, there is a seperate disk
-/// queue for each supported remote filesystem type. In order to maximize throughput,
-/// multiple connections are opened in parallel by having multiple threads running per
-/// queue. Also note that reading from a remote filesystem service can be more CPU
-/// intensive than local disk/hdfs because of non-direct I/O and SSL processing, and can
-/// be CPU bottlenecked especially if not enough I/O threads for these queues are
-/// started.
-//
-/// TODO: IoMgr should be able to request additional scan ranges from the coordinator
-/// to help deal with stragglers.
-/// TODO: look into using a lock free queue
-/// TODO: simplify the common path (less locking, memory allocations).
-/// TODO: Break this up the .h/.cc into multiple files under an /io subdirectory.
-//
-/// Structure of the Implementation:
-/// - All client APIs are defined in this file
-/// - Internal classes are defined in disk-io-mgr-internal.h
-/// - ScanRange APIs are implemented in disk-io-mgr-scan-range.cc
-/// This contains the ready buffer queue logic
-/// - DiskIoRequestContext APIs are implemented in disk-io-mgr-reader-context.cc
-/// This contains the logic for picking scan ranges for a reader.
-/// - Disk Thread and general APIs are implemented in disk-io-mgr.cc.
-
-class DiskIoRequestContext;
-
-// This is cache line aligned because the FileHandleCache needs cache line alignment
-// for its partitions.
-class DiskIoMgr : public CacheLineAligned {
- public:
- class ScanRange;
-
- /// Buffer struct that is used by the caller and IoMgr to pass read buffers.
- /// It is is expected that only one thread has ownership of this object at a
- /// time.
- class BufferDescriptor {
- public:
- ~BufferDescriptor() {
- DCHECK(buffer_ == nullptr); // Check we didn't leak a buffer.
- }
-
- ScanRange* scan_range() { return scan_range_; }
- uint8_t* buffer() { return buffer_; }
- int64_t buffer_len() { return buffer_len_; }
- int64_t len() { return len_; }
- bool eosr() { return eosr_; }
-
- /// Returns the offset within the scan range that this buffer starts at
- int64_t scan_range_offset() const { return scan_range_offset_; }
-
- /// Transfer ownership of buffer memory from 'mem_tracker_' to 'dst' and set
- /// 'mem_tracker_' to 'dst'. 'mem_tracker_' and 'dst' must be non-NULL. Does not
- /// check memory limits on 'dst': the caller should check the memory limit if a
- /// different memory limit may apply to 'dst'. If the buffer was a client-provided
- /// buffer, transferring is not allowed.
- /// TODO: IMPALA-3209: revisit this as part of scanner memory usage revamp.
- void TransferOwnership(MemTracker* dst);
-
- private:
- friend class DiskIoMgr;
- friend class DiskIoMgr::ScanRange;
- friend class DiskIoRequestContext;
-
- /// Create a buffer descriptor for a new reader, range and data buffer. The buffer
- /// memory should already be accounted against 'mem_tracker'.
- BufferDescriptor(DiskIoMgr* io_mgr, DiskIoRequestContext* reader,
- ScanRange* scan_range, uint8_t* buffer, int64_t buffer_len,
- MemTracker* mem_tracker);
-
- /// Return true if this is a cached buffer owned by HDFS.
- bool is_cached() const {
- return scan_range_->external_buffer_tag_
- == ScanRange::ExternalBufferTag::CACHED_BUFFER;
- }
-
- /// Return true if this is a buffer owner by the client that was provided when
- /// constructing the scan range.
- bool is_client_buffer() const {
- return scan_range_->external_buffer_tag_
- == ScanRange::ExternalBufferTag::CLIENT_BUFFER;
- }
-
- DiskIoMgr* const io_mgr_;
-
- /// Reader that this buffer is for.
- DiskIoRequestContext* const reader_;
-
- /// The current tracker this buffer is associated with. After initialisation,
- /// NULL for cached buffers and non-NULL for all other buffers.
- MemTracker* mem_tracker_;
-
- /// Scan range that this buffer is for. Non-NULL when initialised.
- ScanRange* const scan_range_;
-
- /// buffer with the read contents
- uint8_t* buffer_;
-
- /// length of buffer_. For buffers from cached reads, the length is 0.
- const int64_t buffer_len_;
-
- /// length of read contents
- int64_t len_ = 0;
-
- /// true if the current scan range is complete
- bool eosr_ = false;
-
- /// Status of the read to this buffer. if status is not ok, 'buffer' is nullptr
- Status status_;
-
- int64_t scan_range_offset_ = 0;
- };
-
- /// The request type, read or write associated with a request range.
- struct RequestType {
- enum type {
- READ,
- WRITE,
- };
- };
-
- /// Represents a contiguous sequence of bytes in a single file.
- /// This is the common base class for read and write IO requests - ScanRange and
- /// WriteRange. Each disk thread processes exactly one RequestRange at a time.
- class RequestRange : public InternalQueue<RequestRange>::Node {
- public:
- hdfsFS fs() const { return fs_; }
- const char* file() const { return file_.c_str(); }
- std::string* file_string() { return &file_; }
- int64_t offset() const { return offset_; }
- int64_t len() const { return len_; }
- int disk_id() const { return disk_id_; }
- RequestType::type request_type() const { return request_type_; }
-
- protected:
- RequestRange(RequestType::type request_type)
- : fs_(nullptr), offset_(-1), len_(-1), disk_id_(-1), request_type_(request_type) {}
-
- /// Hadoop filesystem that contains file_, or set to nullptr for local filesystem.
- hdfsFS fs_;
-
- /// Path to file being read or written.
- std::string file_;
-
- /// Offset within file_ being read or written.
- int64_t offset_;
-
- /// Length of data read or written.
- int64_t len_;
-
- /// Id of disk containing byte range.
- int disk_id_;
-
- /// The type of IO request, READ or WRITE.
- RequestType::type request_type_;
- };
-
- /// Param struct for different combinations of buffering.
- struct BufferOpts {
- public:
- // Set options for a read into an IoMgr-allocated or HDFS-cached buffer. Caching is
- // enabled if 'try_cache' is true, the file is in the HDFS cache and 'mtime' matches
- // the modified time of the cached file in the HDFS cache.
- BufferOpts(bool try_cache, int64_t mtime)
- : try_cache_(try_cache),
- mtime_(mtime),
- client_buffer_(nullptr),
- client_buffer_len_(-1) {}
-
- /// Set options for an uncached read into an IoMgr-allocated buffer.
- static BufferOpts Uncached() {
- return BufferOpts(false, NEVER_CACHE, nullptr, -1);
- }
-
- /// Set options to read the entire scan range into 'client_buffer'. The length of the
- /// buffer, 'client_buffer_len', must fit the entire scan range. HDFS caching is not
- /// enabled in this case.
- static BufferOpts ReadInto(uint8_t* client_buffer, int64_t client_buffer_len) {
- return BufferOpts(false, NEVER_CACHE, client_buffer, client_buffer_len);
- }
-
- private:
- friend class ScanRange;
-
- BufferOpts(
- bool try_cache, int64_t mtime, uint8_t* client_buffer, int64_t client_buffer_len)
- : try_cache_(try_cache),
- mtime_(mtime),
- client_buffer_(client_buffer),
- client_buffer_len_(client_buffer_len) {}
-
- /// If 'mtime_' is set to NEVER_CACHE, the file handle will never be cached, because
- /// the modification time won't match.
- const static int64_t NEVER_CACHE = -1;
-
- /// If true, read from HDFS cache if possible.
- const bool try_cache_;
-
- /// Last modified time of the file associated with the scan range. If set to
- /// NEVER_CACHE, caching is disabled.
- const int64_t mtime_;
-
- /// A destination buffer provided by the client, nullptr and -1 if no buffer.
- uint8_t* const client_buffer_;
- const int64_t client_buffer_len_;
- };
-
- /// ScanRange description. The caller must call Reset() to initialize the fields
- /// before calling AddScanRanges(). The private fields are used internally by
- /// the IoMgr.
- class ScanRange : public RequestRange {
- public:
- ScanRange();
-
- virtual ~ScanRange();
-
- /// Resets this scan range object with the scan range description. The scan range
- /// is for bytes [offset, offset + len) in 'file' on 'fs' (which is nullptr for the
- /// local filesystem). The scan range must fall within the file bounds (offset >= 0
- /// and offset + len <= file_length). 'disk_id' is the disk queue to add the range
- /// to. If 'expected_local' is true, a warning is generated if the read did not
- /// come from a local disk. 'buffer_opts' specifies buffer management options -
- /// see the DiskIoMgr class comment and the BufferOpts comments for details.
- /// 'meta_data' is an arbitrary client-provided pointer for any auxiliary data.
- void Reset(hdfsFS fs, const char* file, int64_t len, int64_t offset, int disk_id,
- bool expected_local, const BufferOpts& buffer_opts, void* meta_data = nullptr);
-
- void* meta_data() const { return meta_data_; }
- bool try_cache() const { return try_cache_; }
- bool expected_local() const { return expected_local_; }
-
- /// Returns the next buffer for this scan range. buffer is an output parameter.
- /// This function blocks until a buffer is ready or an error occurred. If this is
- /// called when all buffers have been returned, *buffer is set to nullptr and Status::OK
- /// is returned.
- /// Only one thread can be in GetNext() at any time.
- Status GetNext(std::unique_ptr<BufferDescriptor>* buffer) WARN_UNUSED_RESULT;
-
- /// Cancel this scan range. This cleans up all queued buffers and
- /// wakes up any threads blocked on GetNext().
- /// Status is the reason the range was cancelled. Must not be ok().
- /// Status is returned to the user in GetNext().
- void Cancel(const Status& status);
-
- /// return a descriptive string for debug.
- std::string DebugString() const;
-
- int64_t mtime() const { return mtime_; }
-
- private:
- friend class DiskIoMgr;
- friend class DiskIoRequestContext;
-
- /// Initialize internal fields
- void InitInternal(DiskIoMgr* io_mgr, DiskIoRequestContext* reader);
-
- /// Enqueues a buffer for this range. This does not block.
- /// Returns true if this scan range has hit the queue capacity, false otherwise.
- /// The caller passes ownership of buffer to the scan range and it is not
- /// valid to access buffer after this call. The reader lock must be held by the
- /// caller.
- bool EnqueueBuffer(const boost::unique_lock<boost::mutex>& reader_lock,
- std::unique_ptr<BufferDescriptor> buffer);
-
- /// Cleanup any queued buffers (i.e. due to cancellation). This cannot
- /// be called with any locks taken.
- void CleanupQueuedBuffers();
-
- /// Validates the internal state of this range. lock_ must be taken
- /// before calling this.
- bool Validate();
-
- /// Maximum length in bytes for hdfsRead() calls.
- int64_t MaxReadChunkSize() const;
-
- /// Opens the file for this range. This function only modifies state in this range.
- /// If 'use_file_handle_cache' is true and this is a local hdfs file, then this scan
- /// range will not maintain an exclusive file handle. It will borrow an hdfs file
- /// handle from the file handle cache for each Read(), so Open() does nothing.
- /// If 'use_file_handle_cache' is false or this is a remote hdfs file or this is
- /// a local OS file, Open() will maintain a file handle on the scan range for
- /// exclusive use by this scan range. An exclusive hdfs file handle still comes
- /// from the cache, but it is a newly opened file handle that is held for the
- /// entire duration of a scan range's lifetime and destroyed in Close().
- /// All local OS files are opened using normal OS file APIs.
- Status Open(bool use_file_handle_cache) WARN_UNUSED_RESULT;
-
- /// Closes the file for this range. This function only modifies state in this range.
- void Close();
-
- /// Reads from this range into 'buffer', which has length 'buffer_len' bytes. Returns
- /// the number of bytes read. The read position in this scan range is updated.
- Status Read(uint8_t* buffer, int64_t buffer_len, int64_t* bytes_read,
- bool* eosr) WARN_UNUSED_RESULT;
-
- /// Get the read statistics from the Hdfs file handle and aggregate them to
- /// the DiskIoRequestContext. This clears the statistics on this file handle.
- /// It is safe to pass hdfsFile by value, as hdfsFile's underlying type is a
- /// pointer.
- void GetHdfsStatistics(hdfsFile fh);
-
- /// Reads from the DN cache. On success, sets cached_buffer_ to the DN buffer
- /// and *read_succeeded to true.
- /// If the data is not cached, returns ok() and *read_succeeded is set to false.
- /// Returns a non-ok status if it ran into a non-continuable error.
- /// The reader lock must be held by the caller.
- Status ReadFromCache(const boost::unique_lock<boost::mutex>& reader_lock,
- bool* read_succeeded) WARN_UNUSED_RESULT;
-
- /// Pointer to caller specified metadata. This is untouched by the io manager
- /// and the caller can put whatever auxiliary data in here.
- void* meta_data_ = nullptr;
-
- /// If true, this scan range is expected to be cached. Note that this might be wrong
- /// since the block could have been uncached. In that case, the cached path
- /// will fail and we'll just put the scan range on the normal read path.
- bool try_cache_ = false;
-
- /// If true, we expect this scan range to be a local read. Note that if this is false,
- /// it does not necessarily mean we expect the read to be remote, and that we never
- /// create scan ranges where some of the range is expected to be remote and some of it
- /// local.
- /// TODO: we can do more with this
- bool expected_local_ = false;
-
- /// Total number of bytes read remotely. This is necessary to maintain a count of
- /// the number of remote scan ranges. Since IO statistics can be collected multiple
- /// times for a scan range, it is necessary to keep some state about whether this
- /// scan range has already been counted as remote. There is also a requirement to
- /// log the number of unexpected remote bytes for a scan range. To solve both
- /// requirements, maintain num_remote_bytes_ on the ScanRange and push it to the
- /// reader_ once at the close of the scan range.
- int64_t num_remote_bytes_;
-
- DiskIoMgr* io_mgr_ = nullptr;
-
- /// Reader/owner of the scan range
- DiskIoRequestContext* reader_ = nullptr;
-
- /// File handle either to hdfs or local fs (FILE*)
- /// The hdfs file handle is only stored here in three cases:
- /// 1. The file handle cache is off (max_cached_file_handles == 0).
- /// 2. The scan range is using hdfs caching.
- /// -OR-
- /// 3. The hdfs file is expected to be remote (expected_local_ == false)
- /// In each case, the scan range gets a new file handle from the file handle cache
- /// at Open(), holds it exclusively, and destroys it in Close().
- union {
- FILE* local_file_ = nullptr;
- HdfsFileHandle* exclusive_hdfs_fh_;
- };
-
- /// Tagged union that holds a buffer for the cases when there is a buffer allocated
- /// externally from DiskIoMgr that is associated with the scan range.
- enum class ExternalBufferTag { CLIENT_BUFFER, CACHED_BUFFER, NO_BUFFER };
- ExternalBufferTag external_buffer_tag_;
- union {
- /// Valid if the 'external_buffer_tag_' is CLIENT_BUFFER.
- struct {
- /// Client-provided buffer to read the whole scan range into.
- uint8_t* data;
-
- /// Length of the client-provided buffer.
- int64_t len;
- } client_buffer_;
-
- /// Valid and non-NULL if the external_buffer_tag_ is CACHED_BUFFER, which means
- /// that a cached read succeeded and all the bytes for the range are in this buffer.
- struct hadoopRzBuffer* cached_buffer_ = nullptr;
- };
-
- /// Lock protecting fields below.
- /// This lock should not be taken during Open()/Read()/Close().
- /// If DiskIoRequestContext::lock_ and this lock need to be held simultaneously,
- /// DiskIoRequestContext::lock_ must be taken first.
- boost::mutex lock_;
-
- /// Number of bytes read so far for this scan range
- int bytes_read_;
-
- /// Status for this range. This is non-ok if is_cancelled_ is true.
- /// Note: an individual range can fail without the DiskIoRequestContext being
- /// cancelled. This allows us to skip individual ranges.
- Status status_;
-
- /// If true, the last buffer for this scan range has been queued.
- bool eosr_queued_ = false;
-
- /// If true, the last buffer for this scan range has been returned.
- bool eosr_returned_ = false;
-
- /// If true, this scan range has been removed from the reader's in_flight_ranges
- /// queue because the ready_buffers_ queue is full.
- bool blocked_on_queue_ = false;
-
- /// IO buffers that are queued for this scan range.
- /// Condition variable for GetNext
- ConditionVariable buffer_ready_cv_;
- std::deque<std::unique_ptr<BufferDescriptor>> ready_buffers_;
-
- /// Lock that should be taken during hdfs calls. Only one thread (the disk reading
- /// thread) calls into hdfs at a time so this lock does not have performance impact.
- /// This lock only serves to coordinate cleanup. Specifically it serves to ensure
- /// that the disk threads are finished with HDFS calls before is_cancelled_ is set
- /// to true and cleanup starts.
- /// If this lock and lock_ need to be taken, lock_ must be taken first.
- boost::mutex hdfs_lock_;
-
- /// If true, this scan range has been cancelled.
- bool is_cancelled_ = false;
-
- /// Last modified time of the file associated with the scan range
- int64_t mtime_;
- };
-
- /// Used to specify data to be written to a file and offset.
- /// It is the responsibility of the client to ensure that the data to be written is
- /// valid and that the file to be written to exists until the callback is invoked.
- /// A callback is invoked to inform the client when the write is done.
- class WriteRange : public RequestRange {
- public:
- /// This callback is invoked on each WriteRange after the write is complete or the
- /// context is cancelled. The status returned by the callback parameter indicates
- /// if the write was successful (i.e. Status::OK), if there was an error
- /// TStatusCode::RUNTIME_ERROR) or if the context was cancelled
- /// (TStatusCode::CANCELLED). The callback is only invoked if this WriteRange was
- /// successfully added (i.e. AddWriteRange() succeeded). No locks are held while
- /// the callback is invoked.
- typedef std::function<void(const Status&)> WriteDoneCallback;
- WriteRange(const std::string& file, int64_t file_offset, int disk_id,
- WriteDoneCallback callback);
-
- /// Change the file and offset of this write range. Data and callbacks are unchanged.
- /// Can only be called when the write is not in flight (i.e. before AddWriteRange()
- /// is called or after the write callback was called).
- void SetRange(const std::string& file, int64_t file_offset, int disk_id);
-
- /// Set the data and number of bytes to be written for this WriteRange.
- /// Can only be called when the write is not in flight (i.e. before AddWriteRange()
- /// is called or after the write callback was called).
- void SetData(const uint8_t* buffer, int64_t len);
-
- const uint8_t* data() const { return data_; }
-
- private:
- friend class DiskIoMgr;
- friend class DiskIoRequestContext;
-
- /// Data to be written. RequestRange::len_ contains the length of data
- /// to be written.
- const uint8_t* data_;
-
- /// Callback to invoke after the write is complete.
- WriteDoneCallback callback_;
- };
-
- /// Create a DiskIoMgr object. This constructor is only used for testing.
- /// - num_disks: The number of disks the IoMgr should use. This is used for testing.
- /// Specify 0, to have the disk IoMgr query the os for the number of disks.
- /// - threads_per_rotational_disk: number of read threads to create per rotational
- /// disk. This is also the max queue depth.
- /// - threads_per_solid_state_disk: number of read threads to create per solid state
- /// disk. This is also the max queue depth.
- /// - min_buffer_size: minimum io buffer size (in bytes)
- /// - max_buffer_size: maximum io buffer size (in bytes). Also the max read size.
- DiskIoMgr(int num_disks, int threads_per_rotational_disk,
- int threads_per_solid_state_disk, int min_buffer_size, int max_buffer_size);
-
- /// Create DiskIoMgr with default configs.
- DiskIoMgr();
-
- /// Clean up all threads and resources. This is mostly useful for testing since
- /// for impalad, this object is never destroyed.
- ~DiskIoMgr();
-
- /// Initialize the IoMgr. Must be called once before any of the other APIs.
- Status Init(MemTracker* process_mem_tracker) WARN_UNUSED_RESULT;
-
- /// Allocates tracking structure for a request context.
- /// Register a new request context and return it to the caller. The caller must call
- /// UnregisterContext() for each context.
- /// reader_mem_tracker: Is non-null only for readers. IO buffers
- /// used for this reader will be tracked by this. If the limit is exceeded
- /// the reader will be cancelled and MEM_LIMIT_EXCEEDED will be returned via
- /// GetNext().
- std::unique_ptr<DiskIoRequestContext> RegisterContext(MemTracker* reader_mem_tracker);
-
- /// Unregisters context from the disk IoMgr by first cancelling it then blocking until
- /// all references to the context are removed from I/O manager internal data structures.
- /// This must be called for every RegisterContext() to ensure that the context object
- /// can be safely destroyed. It is invalid to add more request ranges to 'context' after
- /// after this call. This call blocks until all the disk threads have finished cleaning
- /// up.
- void UnregisterContext(DiskIoRequestContext* context);
-
- /// This function cancels the context asychronously. All outstanding requests
- /// are aborted and tracking structures cleaned up. This does not need to be
- /// called if the context finishes normally.
- /// This will also fail any outstanding GetNext()/Read requests.
- void CancelContext(DiskIoRequestContext* context);
-
- /// Adds the scan ranges to the queues. This call is non-blocking. The caller must
- /// not deallocate the scan range pointers before UnregisterContext().
- /// If schedule_immediately, the ranges are immediately put on the read queue
- /// (i.e. the caller should not/cannot call GetNextRange for these ranges).
- /// This can be used to do synchronous reads as well as schedule dependent ranges,
- /// as in the case for columnar formats.
- Status AddScanRanges(DiskIoRequestContext* reader,
- const std::vector<ScanRange*>& ranges,
- bool schedule_immediately = false) WARN_UNUSED_RESULT;
- Status AddScanRange(DiskIoRequestContext* reader, ScanRange* range,
- bool schedule_immediately = false) WARN_UNUSED_RESULT;
-
- /// Add a WriteRange for the writer. This is non-blocking and schedules the context
- /// on the IoMgr disk queue. Does not create any files.
- Status AddWriteRange(
- DiskIoRequestContext* writer, WriteRange* write_range) WARN_UNUSED_RESULT;
-
- /// Returns the next unstarted scan range for this reader. When the range is returned,
- /// the disk threads in the IoMgr will already have started reading from it. The
- /// caller is expected to call ScanRange::GetNext on the returned range.
- /// If there are no more unstarted ranges, nullptr is returned.
- /// This call is blocking.
- Status GetNextRange(DiskIoRequestContext* reader, ScanRange** range) WARN_UNUSED_RESULT;
-
- /// Reads the range and returns the result in buffer.
- /// This behaves like the typical synchronous read() api, blocking until the data
- /// is read. This can be called while there are outstanding ScanRanges and is
- /// thread safe. Multiple threads can be calling Read() per reader at a time.
- /// range *cannot* have already been added via AddScanRanges.
- /// This can only be used if the scan range fits in a single IO buffer (i.e. is smaller
- /// than max_read_buffer_size()) or if reading into a client-provided buffer.
- Status Read(DiskIoRequestContext* reader, ScanRange* range,
- std::unique_ptr<BufferDescriptor>* buffer) WARN_UNUSED_RESULT;
-
- /// Returns the buffer to the IoMgr. This must be called for every buffer
- /// returned by GetNext()/Read() that did not return an error. This is non-blocking.
- /// After calling this, the buffer descriptor is invalid and cannot be accessed.
- void ReturnBuffer(std::unique_ptr<BufferDescriptor> buffer);
-
- /// Determine which disk queue this file should be assigned to. Returns an index into
- /// disk_queues_. The disk_id is the volume ID for the local disk that holds the
- /// files, or -1 if unknown. Flag expected_local is true iff this impalad is
- /// co-located with the datanode for this file.
- int AssignQueue(const char* file, int disk_id, bool expected_local);
-
- /// TODO: The functions below can be moved to DiskIoRequestContext.
- /// Returns the current status of the context.
- Status context_status(DiskIoRequestContext* context) const WARN_UNUSED_RESULT;
-
- void set_bytes_read_counter(DiskIoRequestContext*, RuntimeProfile::Counter*);
- void set_read_timer(DiskIoRequestContext*, RuntimeProfile::Counter*);
- void set_active_read_thread_counter(DiskIoRequestContext*, RuntimeProfile::Counter*);
- void set_disks_access_bitmap(DiskIoRequestContext*, RuntimeProfile::Counter*);
-
- int64_t queue_size(DiskIoRequestContext* reader) const;
- int64_t bytes_read_local(DiskIoRequestContext* reader) const;
- int64_t bytes_read_short_circuit(DiskIoRequestContext* reader) const;
- int64_t bytes_read_dn_cache(DiskIoRequestContext* reader) const;
- int num_remote_ranges(DiskIoRequestContext* reader) const;
- int64_t unexpected_remote_bytes(DiskIoRequestContext* reader) const;
- int cached_file_handles_hit_count(DiskIoRequestContext* reader) const;
- int cached_file_handles_miss_count(DiskIoRequestContext* reader) const;
-
- /// Returns the read throughput across all readers.
- /// TODO: should this be a sliding window? This should report metrics for the
- /// last minute, hour and since the beginning.
- int64_t GetReadThroughput();
-
- /// Returns the maximum read buffer size
- int max_read_buffer_size() const { return max_buffer_size_; }
-
- /// Returns the total number of disk queues (both local and remote).
- int num_total_disks() const { return disk_queues_.size(); }
-
- /// Returns the total number of remote "disk" queues.
- int num_remote_disks() const { return REMOTE_NUM_DISKS; }
-
- /// Returns the number of local disks attached to the system.
- int num_local_disks() const { return num_total_disks() - num_remote_disks(); }
-
- /// The disk ID (and therefore disk_queues_ index) used for DFS accesses.
- int RemoteDfsDiskId() const { return num_local_disks() + REMOTE_DFS_DISK_OFFSET; }
-
- /// The disk ID (and therefore disk_queues_ index) used for S3 accesses.
- int RemoteS3DiskId() const { return num_local_disks() + REMOTE_S3_DISK_OFFSET; }
-
- /// The disk ID (and therefore disk_queues_ index) used for ADLS accesses.
- int RemoteAdlsDiskId() const { return num_local_disks() + REMOTE_ADLS_DISK_OFFSET; }
-
- /// Dumps the disk IoMgr queues (for readers and disks)
- std::string DebugString();
-
- /// Validates the internal state is consistent. This is intended to only be used
- /// for debugging.
- bool Validate() const;
-
- /// Given a FS handle, name and last modified time of the file, gets an HdfsFileHandle
- /// from the file handle cache. If 'require_new_handle' is true, the cache will open
- /// a fresh file handle. On success, records statistics about whether this was
- /// a cache hit or miss in the 'reader' as well as at the system level. In case of an
- /// error returns nullptr.
- HdfsFileHandle* GetCachedHdfsFileHandle(const hdfsFS& fs,
- std::string* fname, int64_t mtime, DiskIoRequestContext *reader,
- bool require_new_handle);
-
- /// Releases a file handle back to the file handle cache when it is no longer in use.
- /// If 'destroy_handle' is true, the file handle cache will close the file handle
- /// immediately.
- void ReleaseCachedHdfsFileHandle(std::string* fname, HdfsFileHandle* fid,
- bool destroy_handle);
-
- /// Reopens a file handle by destroying the file handle and getting a fresh
- /// file handle from the cache. Returns an error if the file could not be reopened.
- Status ReopenCachedHdfsFileHandle(const hdfsFS& fs, std::string* fname, int64_t mtime,
- HdfsFileHandle** fid);
-
- /// Garbage collect unused I/O buffers up to 'bytes_to_free', or all the buffers if
- /// 'bytes_to_free' is -1.
- void GcIoBuffers(int64_t bytes_to_free = -1);
-
- /// The maximum number of ready buffers that can be queued in a scan range. Having two
- /// queued buffers (plus the buffer that is returned to the client) gives good
- /// performance in most scenarios:
- /// 1. If the consumer is consuming data faster than we can read from disk, then the
- /// queue will be empty most of the time because the buffer will be immediately
- /// pulled off the queue as soon as it is added. There will always be an I/O request
- /// in the disk queue to maximize I/O throughput, which is the bottleneck in this
- /// case.
- /// 2. If we can read from disk faster than the consumer is consuming data, the queue
- /// will fill up and there will always be a buffer available for the consumer to
- /// read, so the consumer will not block and we maximize consumer throughput, which
- /// is the bottleneck in this case.
- /// 3. If the consumer is consuming data at approximately the same rate as we are
- /// reading from disk, then the steady state is that the consumer is processing one
- /// buffer and one buffer is in the disk queue. The additional buffer can absorb
- /// bursts where the producer runs faster than the consumer or the consumer runs
- /// faster than the producer without blocking either the producer or consumer.
- static const int SCAN_RANGE_READY_BUFFER_LIMIT = 2;
-
- /// "Disk" queue offsets for remote accesses. Offset 0 corresponds to
- /// disk ID (i.e. disk_queue_ index) of num_local_disks().
- enum {
- REMOTE_DFS_DISK_OFFSET = 0,
- REMOTE_S3_DISK_OFFSET,
- REMOTE_ADLS_DISK_OFFSET,
- REMOTE_NUM_DISKS
- };
-
- private:
- friend class BufferDescriptor;
- friend class DiskIoRequestContext;
- struct DiskQueue;
-
- friend class DiskIoMgrTest_Buffers_Test;
- friend class DiskIoMgrTest_VerifyNumThreadsParameter_Test;
-
- /// Memory tracker for unused I/O buffers owned by DiskIoMgr.
- boost::scoped_ptr<MemTracker> free_buffer_mem_tracker_;
-
- /// Memory tracker for I/O buffers where the DiskIoRequestContext has no MemTracker.
- /// TODO: once IMPALA-3200 is fixed, there should be no more cases where readers don't
- /// provide a MemTracker.
- boost::scoped_ptr<MemTracker> unowned_buffer_mem_tracker_;
-
- /// Number of worker(read) threads per rotational disk. Also the max depth of queued
- /// work to the disk.
- const int num_io_threads_per_rotational_disk_;
-
- /// Number of worker(read) threads per solid state disk. Also the max depth of queued
- /// work to the disk.
- const int num_io_threads_per_solid_state_disk_;
-
- /// Maximum read size. This is also the maximum size of each allocated buffer.
- const int max_buffer_size_;
-
- /// The minimum size of each read buffer.
- const int min_buffer_size_;
-
- /// Thread group containing all the worker threads.
- ThreadGroup disk_thread_group_;
-
- /// Options object for cached hdfs reads. Set on startup and never modified.
- struct hadoopRzOptions* cached_read_options_ = nullptr;
-
- /// True if the IoMgr should be torn down. Worker threads watch for this to
- /// know to terminate. This variable is read/written to by different threads.
- volatile bool shut_down_;
-
- /// Total bytes read by the IoMgr.
- RuntimeProfile::Counter total_bytes_read_counter_;
-
- /// Total time spent in hdfs reading
- RuntimeProfile::Counter read_timer_;
-
- /// Protects free_buffers_
- boost::mutex free_buffers_lock_;
-
- /// Free buffers that can be handed out to clients. There is one list for each buffer
- /// size, indexed by the Log2 of the buffer size in units of min_buffer_size_. The
- /// maximum buffer size is max_buffer_size_, so the maximum index is
- /// Log2(max_buffer_size_ / min_buffer_size_).
- //
- /// E.g. if min_buffer_size_ = 1024 bytes:
- /// free_buffers_[0] => list of free buffers with size 1024 B
- /// free_buffers_[1] => list of free buffers with size 2048 B
- /// free_buffers_[10] => list of free buffers with size 1 MB
- /// free_buffers_[13] => list of free buffers with size 8 MB
- /// free_buffers_[n] => list of free buffers with size 2^n * 1024 B
- std::vector<std::deque<uint8_t*>> free_buffers_;
-
- /// Total number of allocated buffers, used for debugging.
- AtomicInt32 num_allocated_buffers_;
-
- /// Total number of buffers in readers
- AtomicInt32 num_buffers_in_readers_;
-
- /// Per disk queues. This is static and created once at Init() time. One queue is
- /// allocated for each local disk on the system and for each remote filesystem type.
- /// It is indexed by disk id.
- std::vector<DiskQueue*> disk_queues_;
-
- /// The next disk queue to write to if the actual 'disk_id_' is unknown (i.e. the file
- /// is not associated with a particular local disk or remote queue). Used to implement
- /// round-robin assignment for that case.
- static AtomicInt32 next_disk_id_;
-
- // Number of file handle cache partitions to use
- static const size_t NUM_FILE_HANDLE_CACHE_PARTITIONS = 16;
-
- // Caching structure that maps file names to cached file handles. The cache has an upper
- // limit of entries defined by FLAGS_max_cached_file_handles. Evicted cached file
- // handles are closed.
- FileHandleCache<NUM_FILE_HANDLE_CACHE_PARTITIONS> file_handle_cache_;
-
- /// Returns the index into free_buffers_ for a given buffer size
- int free_buffers_idx(int64_t buffer_size);
-
- /// Returns a buffer to read into with size between 'buffer_size' and
- /// 'max_buffer_size_', If there is an appropriately-sized free buffer in the
- /// 'free_buffers_', that is returned, otherwise a new one is allocated.
- /// The returned *buffer_size must be between 0 and 'max_buffer_size_'.
- /// The buffer memory is tracked against reader's mem tracker, or
- /// 'unowned_buffer_mem_tracker_' if the reader does not have one.
- std::unique_ptr<BufferDescriptor> GetFreeBuffer(
- DiskIoRequestContext* reader, ScanRange* range, int64_t buffer_size);
-
- /// Disassociates the desc->buffer_ memory from 'desc' (which cannot be nullptr), either
- /// freeing it or returning it to 'free_buffers_'. Memory tracking is updated to
- /// reflect the transfer of ownership from desc->mem_tracker_ to the disk I/O mgr.
- void FreeBufferMemory(BufferDescriptor* desc);
-
- /// Disk worker thread loop. This function retrieves the next range to process on
- /// the disk queue and invokes ReadRange() or Write() depending on the type of Range().
- /// There can be multiple threads per disk running this loop.
- void WorkLoop(DiskQueue* queue);
-
- /// This is called from the disk thread to get the next range to process. It will
- /// wait until a scan range and buffer are available, or a write range is available.
- /// This functions returns the range to process.
- /// Only returns false if the disk thread should be shut down.
- /// No locks should be taken before this function call and none are left taken after.
- bool GetNextRequestRange(DiskQueue* disk_queue, RequestRange** range,
- DiskIoRequestContext** request_context);
-
- /// Updates disk queue and reader state after a read is complete. The read result
- /// is captured in the buffer descriptor.
- void HandleReadFinished(DiskQueue* disk_queue, DiskIoRequestContext* reader,
- std::unique_ptr<BufferDescriptor> buffer);
-
- /// Invokes write_range->callback_ after the range has been written and
- /// updates per-disk state and handle state. The status of the write OK/RUNTIME_ERROR
- /// etc. is passed via write_status and to the callback.
- /// The write_status does not affect the writer->status_. That is, an write error does
- /// not cancel the writer context - that decision is left to the callback handler.
- /// TODO: On the read path, consider not canceling the reader context on error.
- void HandleWriteFinished(
- DiskIoRequestContext* writer, WriteRange* write_range, const Status& write_status);
-
- /// Validates that range is correctly initialized
- Status ValidateScanRange(ScanRange* range) WARN_UNUSED_RESULT;
-
- /// Write the specified range to disk and calls HandleWriteFinished when done.
- /// Responsible for opening and closing the file that is written.
- void Write(DiskIoRequestContext* writer_context, WriteRange* write_range);
-
- /// Helper method to write a range using the specified FILE handle. Returns Status:OK
- /// if the write succeeded, or a RUNTIME_ERROR with an appropriate message otherwise.
- /// Does not open or close the file that is written.
- Status WriteRangeHelper(FILE* file_handle, WriteRange* write_range) WARN_UNUSED_RESULT;
-
- /// Reads the specified scan range and calls HandleReadFinished when done.
- void ReadRange(DiskQueue* disk_queue, DiskIoRequestContext* reader, ScanRange* range);
-
- /// Try to allocate the next buffer for the scan range, returning the new buffer
- /// if successful. If 'reader' is cancelled, cancels the range and returns nullptr.
- /// If there is memory pressure and buffers are already queued, adds the range
- /// to the blocked ranges and returns nullptr.
- std::unique_ptr<BufferDescriptor> TryAllocateNextBufferForRange(DiskQueue* disk_queue,
- DiskIoRequestContext* reader, ScanRange* range, int64_t buffer_size);
-};
-}
-
-#endif
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/b840137c/be/src/runtime/exec-env.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/exec-env.cc b/be/src/runtime/exec-env.cc
index 999b56a..6fb572c 100644
--- a/be/src/runtime/exec-env.cc
+++ b/be/src/runtime/exec-env.cc
@@ -36,7 +36,7 @@
#include "runtime/client-cache.h"
#include "runtime/coordinator.h"
#include "runtime/data-stream-mgr.h"
-#include "runtime/disk-io-mgr.h"
+#include "runtime/io/disk-io-mgr.h"
#include "runtime/hbase-table-factory.h"
#include "runtime/hdfs-fs-cache.h"
#include "runtime/krpc-data-stream-mgr.h"
@@ -164,7 +164,7 @@ ExecEnv::ExecEnv(const string& hostname, int backend_port, int krpc_port,
FLAGS_catalog_client_rpc_timeout_ms, FLAGS_catalog_client_rpc_timeout_ms, "",
!FLAGS_ssl_client_ca_certificate.empty())),
htable_factory_(new HBaseTableFactory()),
- disk_io_mgr_(new DiskIoMgr()),
+ disk_io_mgr_(new io::DiskIoMgr()),
webserver_(new Webserver(webserver_port)),
pool_mem_trackers_(new PoolMemTrackerRegistry),
thread_mgr_(new ThreadResourceMgr),
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/b840137c/be/src/runtime/exec-env.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/exec-env.h b/be/src/runtime/exec-env.h
index 8fafdc5..193fdde 100644
--- a/be/src/runtime/exec-env.h
+++ b/be/src/runtime/exec-env.h
@@ -43,7 +43,6 @@ class BufferPool;
class CallableThreadPool;
class DataStreamMgrBase;
class DataStreamMgr;
-class DiskIoMgr;
class QueryExecMgr;
class Frontend;
class HBaseTableFactory;
@@ -65,6 +64,10 @@ class ThreadResourceMgr;
class TmpFileMgr;
class Webserver;
+namespace io {
+ class DiskIoMgr;
+}
+
/// Execution environment for Impala daemon. Contains all required global structures, and
/// handles to singleton services. Clients must call StartServices() exactly once to
/// properly initialise service state.
@@ -116,7 +119,7 @@ class ExecEnv {
return catalogd_client_cache_.get();
}
HBaseTableFactory* htable_factory() { return htable_factory_.get(); }
- DiskIoMgr* disk_io_mgr() { return disk_io_mgr_.get(); }
+ io::DiskIoMgr* disk_io_mgr() { return disk_io_mgr_.get(); }
Webserver* webserver() { return webserver_.get(); }
MetricGroup* metrics() { return metrics_.get(); }
MemTracker* process_mem_tracker() { return mem_tracker_.get(); }
@@ -174,7 +177,7 @@ class ExecEnv {
boost::scoped_ptr<ImpalaBackendClientCache> impalad_client_cache_;
boost::scoped_ptr<CatalogServiceClientCache> catalogd_client_cache_;
boost::scoped_ptr<HBaseTableFactory> htable_factory_;
- boost::scoped_ptr<DiskIoMgr> disk_io_mgr_;
+ boost::scoped_ptr<io::DiskIoMgr> disk_io_mgr_;
boost::scoped_ptr<Webserver> webserver_;
boost::scoped_ptr<MemTracker> mem_tracker_;
boost::scoped_ptr<PoolMemTrackerRegistry> pool_mem_trackers_;
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/b840137c/be/src/runtime/io/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/CMakeLists.txt b/be/src/runtime/io/CMakeLists.txt
new file mode 100644
index 0000000..ae89509
--- /dev/null
+++ b/be/src/runtime/io/CMakeLists.txt
@@ -0,0 +1,36 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+
+# where to put generated libraries
+set(LIBRARY_OUTPUT_PATH "${BUILD_OUTPUT_ROOT_DIRECTORY}/runtime/io")
+
+# where to put generated binaries
+set(EXECUTABLE_OUTPUT_PATH "${BUILD_OUTPUT_ROOT_DIRECTORY}/runtime/io")
+
+add_library(Io
+ disk-io-mgr.cc
+ disk-io-mgr-stress.cc
+ request-context.cc
+ scan-range.cc
+)
+add_dependencies(Io gen-deps)
+
+# This test runs forever so should not be part of 'make test'
+add_executable(disk-io-mgr-stress-test disk-io-mgr-stress-test.cc)
+target_link_libraries(disk-io-mgr-stress-test ${IMPALA_TEST_LINK_LIBS})
+
+ADD_BE_TEST(disk-io-mgr-test)
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/b840137c/be/src/runtime/io/disk-io-mgr-internal.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/disk-io-mgr-internal.h b/be/src/runtime/io/disk-io-mgr-internal.h
new file mode 100644
index 0000000..3fc3895
--- /dev/null
+++ b/be/src/runtime/io/disk-io-mgr-internal.h
@@ -0,0 +1,78 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 <unistd.h>
+#include <queue>
+#include <boost/thread/locks.hpp>
+#include <gutil/strings/substitute.h>
+
+#include "common/logging.h"
+#include "runtime/io/request-context.h"
+#include "runtime/io/disk-io-mgr.h"
+#include "runtime/mem-tracker.h"
+#include "runtime/thread-resource-mgr.h"
+#include "util/condition-variable.h"
+#include "util/cpu-info.h"
+#include "util/debug-util.h"
+#include "util/disk-info.h"
+#include "util/filesystem-util.h"
+#include "util/hdfs-util.h"
+#include "util/impalad-metrics.h"
+
+/// This file contains internal structures shared between submodules of the IoMgr. Users
+/// of the IoMgr do not need to include this file.
+namespace impala {
+namespace io {
+
+/// Per disk state
+struct DiskIoMgr::DiskQueue {
+ /// Disk id (0-based)
+ int disk_id;
+
+ /// Lock that protects access to 'request_contexts' and 'work_available'
+ boost::mutex lock;
+
+ /// Condition variable to signal the disk threads that there is work to do or the
+ /// thread should shut down. A disk thread will be woken up when there is a reader
+ /// added to the queue. A reader is only on the queue when it has at least one
+ /// scan range that is not blocked on available buffers.
+ ConditionVariable work_available;
+
+ /// list of all request contexts that have work queued on this disk
+ std::list<RequestContext*> request_contexts;
+
+ /// Enqueue the request context to the disk queue. The DiskQueue lock must not be taken.
+ inline void EnqueueContext(RequestContext* worker) {
+ {
+ boost::unique_lock<boost::mutex> disk_lock(lock);
+ /// Check that the reader is not already on the queue
+ DCHECK(find(request_contexts.begin(), request_contexts.end(), worker) ==
+ request_contexts.end());
+ request_contexts.push_back(worker);
+ }
+ work_available.NotifyAll();
+ }
+
+ DiskQueue(int id) : disk_id(id) {}
+};
+}
+}
+
+#endif
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/b840137c/be/src/runtime/io/disk-io-mgr-stress-test.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/disk-io-mgr-stress-test.cc b/be/src/runtime/io/disk-io-mgr-stress-test.cc
new file mode 100644
index 0000000..45b36ed
--- /dev/null
+++ b/be/src/runtime/io/disk-io-mgr-stress-test.cc
@@ -0,0 +1,61 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+#include "runtime/io/disk-io-mgr-stress.h"
+#include "util/cpu-info.h"
+#include "util/string-parser.h"
+
+#include "common/names.h"
+
+using namespace impala;
+using namespace impala::io;
+
+// Simple utility to run the disk io stress test. A optional second parameter
+// can be passed to control how long to run this test (0 for forever).
+
+// TODO: make these configurable once we decide how to run BE tests with args
+const int DEFAULT_DURATION_SEC = 1;
+const int NUM_DISKS = 5;
+const int NUM_THREADS_PER_DISK = 5;
+const int NUM_CLIENTS = 10;
+const bool TEST_CANCELLATION = true;
+
+int main(int argc, char** argv) {
+ google::InitGoogleLogging(argv[0]);
+ CpuInfo::Init();
+ OsInfo::Init();
+ impala::InitThreading();
+ int duration_sec = DEFAULT_DURATION_SEC;
+
+ if (argc == 2) {
+ StringParser::ParseResult status;
+ duration_sec = StringParser::StringToInt<int>(argv[1], strlen(argv[1]), &status);
+ if (status != StringParser::PARSE_SUCCESS) {
+ printf("Invalid arg: %s\n", argv[1]);
+ return 1;
+ }
+ }
+ if (duration_sec != 0) {
+ printf("Running stress test for %d seconds.\n", duration_sec);
+ } else {
+ printf("Running stress test indefinitely.\n");
+ }
+ DiskIoMgrStress test(NUM_DISKS, NUM_THREADS_PER_DISK, NUM_CLIENTS, TEST_CANCELLATION);
+ test.Run(duration_sec);
+
+ return 0;
+}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/b840137c/be/src/runtime/io/disk-io-mgr-stress.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/disk-io-mgr-stress.cc b/be/src/runtime/io/disk-io-mgr-stress.cc
new file mode 100644
index 0000000..8815357
--- /dev/null
+++ b/be/src/runtime/io/disk-io-mgr-stress.cc
@@ -0,0 +1,247 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+#include <boost/thread/mutex.hpp>
+
+#include "runtime/io/disk-io-mgr-stress.h"
+
+#include "runtime/io/request-context.h"
+#include "util/time.h"
+
+#include "common/names.h"
+
+using namespace impala;
+using namespace impala::io;
+
+static const float ABORT_CHANCE = .10f;
+static const int MIN_READ_LEN = 1;
+static const int MAX_READ_LEN = 20;
+
+static const int MIN_FILE_LEN = 10;
+static const int MAX_FILE_LEN = 1024;
+
+// Make sure this is between MIN/MAX FILE_LEN to test more cases
+static const int MIN_READ_BUFFER_SIZE = 64;
+static const int MAX_READ_BUFFER_SIZE = 128;
+
+static const int CANCEL_READER_PERIOD_MS = 20; // in ms
+
+static void CreateTempFile(const char* filename, const char* data) {
+ FILE* file = fopen(filename, "w");
+ CHECK(file != NULL);
+ fwrite(data, 1, strlen(data), file);
+ fclose(file);
+}
+
+string GenerateRandomData() {
+ int rand_len = rand() % (MAX_FILE_LEN - MIN_FILE_LEN) + MIN_FILE_LEN;
+ stringstream ss;
+ for (int i = 0; i < rand_len; ++i) {
+ char c = rand() % 26 + 'a';
+ ss << c;
+ }
+ return ss.str();
+}
+
+struct DiskIoMgrStress::Client {
+ boost::mutex lock;
+ unique_ptr<RequestContext> reader;
+ int file_idx;
+ vector<ScanRange*> scan_ranges;
+ int abort_at_byte;
+ int files_processed;
+};
+
+DiskIoMgrStress::DiskIoMgrStress(int num_disks, int num_threads_per_disk,
+ int num_clients, bool includes_cancellation) :
+ num_clients_(num_clients),
+ includes_cancellation_(includes_cancellation) {
+
+ time_t rand_seed = time(NULL);
+ LOG(INFO) << "Running with rand seed: " << rand_seed;
+ srand(rand_seed);
+
+ io_mgr_.reset(new DiskIoMgr(num_disks, num_threads_per_disk, num_threads_per_disk,
+ MIN_READ_BUFFER_SIZE, MAX_READ_BUFFER_SIZE));
+ Status status = io_mgr_->Init(&mem_tracker_);
+ CHECK(status.ok());
+
+ // Initialize some data files. It doesn't really matter how many there are.
+ files_.resize(num_clients * 2);
+ for (int i = 0; i < files_.size(); ++i) {
+ stringstream ss;
+ ss << "/tmp/disk_io_mgr_stress_file" << i;
+ files_[i].filename = ss.str();
+ files_[i].data = GenerateRandomData();
+ CreateTempFile(files_[i].filename.c_str(), files_[i].data.c_str());
+ }
+
+ clients_ = new Client[num_clients_];
+ client_mem_trackers_.resize(num_clients_);
+ for (int i = 0; i < num_clients_; ++i) {
+ NewClient(i);
+ }
+}
+
+void DiskIoMgrStress::ClientThread(int client_id) {
+ Client* client = &clients_[client_id];
+ Status status;
+ char read_buffer[MAX_FILE_LEN];
+
+ while (!shutdown_) {
+ bool eos = false;
+ int bytes_read = 0;
+
+ const string& expected = files_[client->file_idx].data;
+
+ while (!eos) {
+ ScanRange* range;
+ Status status = io_mgr_->GetNextRange(client->reader.get(), &range);
+ CHECK(status.ok() || status.IsCancelled());
+ if (range == NULL) break;
+
+ while (true) {
+ unique_ptr<BufferDescriptor> buffer;
+ status = range->GetNext(&buffer);
+ CHECK(status.ok() || status.IsCancelled());
+ if (buffer == NULL) break;
+
+ int64_t scan_range_offset = buffer->scan_range_offset();
+ int len = buffer->len();
+ CHECK_GE(scan_range_offset, 0);
+ CHECK_LT(scan_range_offset, expected.size());
+ CHECK_GT(len, 0);
+
+ // We get scan ranges back in arbitrary order so the scan range to the file
+ // offset.
+ int64_t file_offset = scan_range_offset + range->offset();
+
+ // Validate the bytes read
+ CHECK_LE(file_offset + len, expected.size());
+ CHECK_EQ(strncmp(reinterpret_cast<char*>(buffer->buffer()),
+ &expected.c_str()[file_offset], len), 0);
+
+ // Copy the bytes from this read into the result buffer.
+ memcpy(read_buffer + file_offset, buffer->buffer(), buffer->len());
+ io_mgr_->ReturnBuffer(move(buffer));
+ bytes_read += len;
+
+ CHECK_GE(bytes_read, 0);
+ CHECK_LE(bytes_read, expected.size());
+
+ if (bytes_read > client->abort_at_byte) {
+ eos = true;
+ break;
+ }
+ } // End of buffer
+ } // End of scan range
+
+ if (bytes_read == expected.size()) {
+ // This entire file was read without being cancelled, validate the entire result
+ CHECK(status.ok());
+ CHECK_EQ(strncmp(read_buffer, expected.c_str(), bytes_read), 0);
+ }
+
+ // Unregister the old client and get a new one
+ unique_lock<mutex> lock(client->lock);
+ io_mgr_->UnregisterContext(client->reader.get());
+ NewClient(client_id);
+ }
+
+ unique_lock<mutex> lock(client->lock);
+ io_mgr_->UnregisterContext(client->reader.get());
+ client->reader = NULL;
+}
+
+// Cancel a random reader
+void DiskIoMgrStress::CancelRandomReader() {
+ if (!includes_cancellation_) return;
+
+ int rand_client = rand() % num_clients_;
+
+ unique_lock<mutex> lock(clients_[rand_client].lock);
+ io_mgr_->CancelContext(clients_[rand_client].reader.get());
+}
+
+void DiskIoMgrStress::Run(int sec) {
+ shutdown_ = false;
+ for (int i = 0; i < num_clients_; ++i) {
+ readers_.add_thread(
+ new thread(&DiskIoMgrStress::ClientThread, this, i));
+ }
+
+ // Sleep and let the clients do their thing for 'sec'
+ for (int loop_count = 1; sec == 0 || loop_count <= sec; ++loop_count) {
+ int iter = (1000) / CANCEL_READER_PERIOD_MS;
+ for (int i = 0; i < iter; ++i) {
+ SleepForMs(CANCEL_READER_PERIOD_MS);
+ CancelRandomReader();
+ }
+ LOG(ERROR) << "Finished iteration: " << loop_count;
+ }
+
+ // Signal shutdown for the client threads
+ shutdown_ = true;
+
+ for (int i = 0; i < num_clients_; ++i) {
+ unique_lock<mutex> lock(clients_[i].lock);
+ if (clients_[i].reader != NULL) io_mgr_->CancelContext(clients_[i].reader.get());
+ }
+
+ readers_.join_all();
+}
+
+// Initialize a client to read one of the files at random. The scan ranges are
+// assigned randomly.
+void DiskIoMgrStress::NewClient(int i) {
+ Client& client = clients_[i];
+ ++client.files_processed;
+ client.file_idx = rand() % files_.size();
+ int file_len = files_[client.file_idx].data.size();
+
+ client.abort_at_byte = file_len;
+
+ if (includes_cancellation_) {
+ float rand_value = rand() / (float)RAND_MAX;
+ if (rand_value < ABORT_CHANCE) {
+ // Abort at a random byte inside the file
+ client.abort_at_byte = rand() % file_len;
+ }
+ }
+
+ for (int i = 0; i < client.scan_ranges.size(); ++i) {
+ delete client.scan_ranges[i];
+ }
+ client.scan_ranges.clear();
+
+ int assigned_len = 0;
+ while (assigned_len < file_len) {
+ int range_len = rand() % (MAX_READ_LEN - MIN_READ_LEN) + MIN_READ_LEN;
+ range_len = min(range_len, file_len - assigned_len);
+
+ ScanRange* range = new ScanRange();
+ range->Reset(NULL, files_[client.file_idx].filename.c_str(), range_len, assigned_len,
+ 0, false, BufferOpts::Uncached());
+ client.scan_ranges.push_back(range);
+ assigned_len += range_len;
+ }
+
+ client_mem_trackers_[i].reset(new MemTracker(-1, "", &mem_tracker_));
+ client.reader = io_mgr_->RegisterContext(client_mem_trackers_[i].get());
+ Status status = io_mgr_->AddScanRanges(client.reader.get(), client.scan_ranges);
+ CHECK(status.ok());
+}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/b840137c/be/src/runtime/io/disk-io-mgr-stress.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/disk-io-mgr-stress.h b/be/src/runtime/io/disk-io-mgr-stress.h
new file mode 100644
index 0000000..b872694
--- /dev/null
+++ b/be/src/runtime/io/disk-io-mgr-stress.h
@@ -0,0 +1,95 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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_STRESS_H
+#define IMPALA_RUNTIME_DISK_IO_MGR_STRESS_H
+
+#include <memory>
+#include <vector>
+#include <boost/scoped_ptr.hpp>
+#include <boost/thread/thread.hpp>
+
+#include "runtime/io/disk-io-mgr.h"
+#include "runtime/mem-tracker.h"
+#include "runtime/thread-resource-mgr.h"
+
+namespace impala {
+namespace io {
+
+/// Test utility to stress the disk io mgr. It allows for a configurable
+/// number of clients. The clients continuously issue work to the io mgr and
+/// asynchronously get cancelled. The stress test can be run forever or for
+/// a fixed duration. The unit test runs this for a fixed duration.
+class DiskIoMgrStress {
+ public:
+ DiskIoMgrStress(int num_disks, int num_threads_per_disk, int num_clients,
+ bool includes_cancellation);
+
+ /// Run the test for 'sec'. If 0, run forever
+ void Run(int sec);
+
+ private:
+ struct Client;
+
+ struct File {
+ std::string filename;
+ std::string data; // the data in the file, used to validate
+ };
+
+
+ /// Files used for testing. These are created at startup and recycled
+ /// during the test
+ std::vector<File> files_;
+
+ /// Root mem tracker.
+ MemTracker mem_tracker_;
+
+ /// io manager
+ boost::scoped_ptr<DiskIoMgr> io_mgr_;
+
+ /// Thread group for reader threads
+ boost::thread_group readers_;
+
+ /// Array of clients
+ int num_clients_;
+ Client* clients_;
+
+ /// Client MemTrackers, one per client.
+ std::vector<std::unique_ptr<MemTracker>> client_mem_trackers_;
+
+ /// If true, tests cancelling readers
+ bool includes_cancellation_;
+
+ /// Flag to signal that client reader threads should exit
+ volatile bool shutdown_;
+
+ /// Helper to initialize a new reader client, registering a new reader with the
+ /// io mgr and initializing the scan ranges
+ void NewClient(int i);
+
+ /// Thread running the reader. When the current reader is done (either normally
+ /// or cancelled), it picks up a new reader
+ void ClientThread(int client_id);
+
+ /// Possibly cancels a random reader.
+ void CancelRandomReader();
+};
+}
+}
+
+#endif