mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-12-12 17:32:32 +00:00
276 lines
8.1 KiB
C++
276 lines
8.1 KiB
C++
#include "AsynchronousReadBufferFromHDFS.h"
|
|
|
|
#if USE_HDFS
|
|
#include <mutex>
|
|
#include <Common/logger_useful.h>
|
|
#include <Storages/HDFS/ReadBufferFromHDFS.h>
|
|
#include <Disks/IO/ThreadPoolRemoteFSReader.h>
|
|
#include <IO/AsynchronousReader.h>
|
|
|
|
|
|
namespace CurrentMetrics
|
|
{
|
|
extern const Metric AsynchronousReadWait;
|
|
}
|
|
|
|
namespace ProfileEvents
|
|
{
|
|
extern const Event AsynchronousReadWaitMicroseconds;
|
|
extern const Event RemoteFSSeeks;
|
|
extern const Event RemoteFSPrefetches;
|
|
extern const Event RemoteFSCancelledPrefetches;
|
|
extern const Event RemoteFSUnusedPrefetches;
|
|
extern const Event RemoteFSPrefetchedReads;
|
|
extern const Event RemoteFSUnprefetchedReads;
|
|
extern const Event RemoteFSBuffers;
|
|
}
|
|
|
|
namespace DB
|
|
{
|
|
|
|
namespace ErrorCodes
|
|
{
|
|
extern const int CANNOT_SEEK_THROUGH_FILE;
|
|
extern const int SEEK_POSITION_OUT_OF_BOUND;
|
|
extern const int LOGICAL_ERROR;
|
|
}
|
|
|
|
AsynchronousReadBufferFromHDFS::AsynchronousReadBufferFromHDFS(
|
|
IAsynchronousReader & reader_, const ReadSettings & settings_, std::shared_ptr<ReadBufferFromHDFS> impl_)
|
|
: BufferWithOwnMemory<SeekableReadBuffer>(settings_.remote_fs_buffer_size)
|
|
, reader(reader_)
|
|
, base_priority(settings_.priority)
|
|
, impl(std::move(impl_))
|
|
, prefetch_buffer(settings_.remote_fs_buffer_size)
|
|
, read_until_position(impl->getFileSize())
|
|
, use_prefetch(settings_.remote_fs_prefetch)
|
|
, log(&Poco::Logger::get("AsynchronousReadBufferFromHDFS"))
|
|
{
|
|
ProfileEvents::increment(ProfileEvents::RemoteFSBuffers);
|
|
}
|
|
|
|
bool AsynchronousReadBufferFromHDFS::hasPendingDataToRead()
|
|
{
|
|
if (read_until_position)
|
|
{
|
|
/// Everything is already read.
|
|
if (file_offset_of_buffer_end == *read_until_position)
|
|
return false;
|
|
|
|
if (file_offset_of_buffer_end > *read_until_position)
|
|
throw Exception(ErrorCodes::LOGICAL_ERROR, "Read beyond last offset ({} > {}, info: {})",
|
|
file_offset_of_buffer_end, *read_until_position, impl->getInfoForLog());
|
|
}
|
|
return true;
|
|
}
|
|
|
|
std::future<IAsynchronousReader::Result> AsynchronousReadBufferFromHDFS::asyncReadInto(char * data, size_t size, Priority priority)
|
|
{
|
|
IAsynchronousReader::Request request;
|
|
request.descriptor = std::make_shared<RemoteFSFileDescriptor>(*impl, nullptr);
|
|
request.buf = data;
|
|
request.size = size;
|
|
request.offset = file_offset_of_buffer_end;
|
|
request.priority = Priority{base_priority.value + priority.value};
|
|
request.ignore = 0;
|
|
return reader.submit(request);
|
|
}
|
|
|
|
void AsynchronousReadBufferFromHDFS::prefetch(Priority priority)
|
|
{
|
|
interval_watch.restart();
|
|
|
|
if (prefetch_future.valid())
|
|
return;
|
|
|
|
if (!hasPendingDataToRead())
|
|
return;
|
|
|
|
prefetch_future = asyncReadInto(prefetch_buffer.data(), prefetch_buffer.size(), priority);
|
|
ProfileEvents::increment(ProfileEvents::RemoteFSPrefetches);
|
|
}
|
|
|
|
|
|
size_t AsynchronousReadBufferFromHDFS::getFileSize()
|
|
{
|
|
return impl->getFileSize();
|
|
}
|
|
|
|
String AsynchronousReadBufferFromHDFS::getFileName() const
|
|
{
|
|
return impl->getFileName();
|
|
}
|
|
|
|
|
|
bool AsynchronousReadBufferFromHDFS::nextImpl()
|
|
{
|
|
if (!hasPendingDataToRead())
|
|
return false;
|
|
|
|
++next_times;
|
|
sum_interval += interval_watch.elapsedMicroseconds();
|
|
|
|
Stopwatch next_watch;
|
|
Int64 wait = -1;
|
|
size_t size = 0;
|
|
size_t bytes_read = 0;
|
|
|
|
if (prefetch_future.valid())
|
|
{
|
|
ProfileEvents::increment(ProfileEvents::RemoteFSPrefetchedReads);
|
|
|
|
size_t offset = 0;
|
|
{
|
|
Stopwatch watch;
|
|
CurrentMetrics::Increment metric_increment{CurrentMetrics::AsynchronousReadWait};
|
|
auto result = prefetch_future.get();
|
|
size = result.size;
|
|
offset = result.offset;
|
|
LOG_TEST(log, "Current size: {}, offset: {}", size, offset);
|
|
|
|
/// If prefetch_future is valid, size should always be greater than zero.
|
|
assert(offset <= size);
|
|
bytes_read = size - offset;
|
|
|
|
wait = watch.elapsedMicroseconds();
|
|
ProfileEvents::increment(ProfileEvents::AsynchronousReadWaitMicroseconds, wait);
|
|
}
|
|
|
|
prefetch_buffer.swap(memory);
|
|
|
|
/// Adjust the working buffer so that it ignores `offset` bytes.
|
|
internal_buffer = Buffer(memory.data(), memory.data() + memory.size());
|
|
working_buffer = Buffer(memory.data() + offset, memory.data() + size);
|
|
pos = working_buffer.begin();
|
|
}
|
|
else
|
|
{
|
|
ProfileEvents::increment(ProfileEvents::RemoteFSUnprefetchedReads);
|
|
|
|
auto result = asyncReadInto(memory.data(), memory.size(), DEFAULT_PREFETCH_PRIORITY).get();
|
|
size = result.size;
|
|
auto offset = result.offset;
|
|
|
|
LOG_TEST(log, "Current size: {}, offset: {}", size, offset);
|
|
assert(offset <= size);
|
|
bytes_read = size - offset;
|
|
|
|
if (bytes_read)
|
|
{
|
|
/// Adjust the working buffer so that it ignores `offset` bytes.
|
|
internal_buffer = Buffer(memory.data(), memory.data() + memory.size());
|
|
working_buffer = Buffer(memory.data() + offset, memory.data() + size);
|
|
pos = working_buffer.begin();
|
|
}
|
|
}
|
|
|
|
file_offset_of_buffer_end = impl->getFileOffsetOfBufferEnd();
|
|
prefetch_future = {};
|
|
|
|
if (use_prefetch && bytes_read)
|
|
prefetch(DEFAULT_PREFETCH_PRIORITY);
|
|
|
|
sum_duration += next_watch.elapsedMicroseconds();
|
|
sum_wait += wait;
|
|
return bytes_read;
|
|
}
|
|
|
|
off_t AsynchronousReadBufferFromHDFS::seek(off_t offset, int whence)
|
|
{
|
|
ProfileEvents::increment(ProfileEvents::RemoteFSSeeks);
|
|
|
|
if (whence != SEEK_SET)
|
|
throw Exception(ErrorCodes::CANNOT_SEEK_THROUGH_FILE, "Only SEEK_SET mode is allowed.");
|
|
|
|
if (offset < 0)
|
|
throw Exception(ErrorCodes::SEEK_POSITION_OUT_OF_BOUND, "Seek position is out of bounds. Offset: {}", offset);
|
|
|
|
size_t new_pos = offset;
|
|
|
|
/// Position is unchanged.
|
|
if (new_pos + (working_buffer.end() - pos) == file_offset_of_buffer_end)
|
|
return new_pos;
|
|
|
|
bool read_from_prefetch = false;
|
|
while (true)
|
|
{
|
|
if (file_offset_of_buffer_end - working_buffer.size() <= new_pos && new_pos <= file_offset_of_buffer_end)
|
|
{
|
|
/// Position is still inside the buffer.
|
|
/// Probably it is at the end of the buffer - then we will load data on the following 'next' call.
|
|
pos = working_buffer.end() - file_offset_of_buffer_end + new_pos;
|
|
assert(pos >= working_buffer.begin());
|
|
assert(pos <= working_buffer.end());
|
|
return new_pos;
|
|
}
|
|
else if (prefetch_future.valid())
|
|
{
|
|
/// Read from prefetch buffer and recheck if the new position is valid inside.
|
|
/// TODO we can judge quickly without waiting for prefetch
|
|
if (nextImpl())
|
|
{
|
|
read_from_prefetch = true;
|
|
continue;
|
|
}
|
|
}
|
|
|
|
/// Prefetch is cancelled because of seek.
|
|
if (read_from_prefetch)
|
|
ProfileEvents::increment(ProfileEvents::RemoteFSCancelledPrefetches);
|
|
break;
|
|
}
|
|
|
|
assert(!prefetch_future.valid());
|
|
|
|
/// First reset the buffer so the next read will fetch new data to the buffer.
|
|
resetWorkingBuffer();
|
|
|
|
++seek_times;
|
|
impl->seek(new_pos, SEEK_SET);
|
|
file_offset_of_buffer_end = new_pos;
|
|
|
|
return new_pos;
|
|
}
|
|
|
|
void AsynchronousReadBufferFromHDFS::finalize()
|
|
{
|
|
if (prefetch_future.valid())
|
|
{
|
|
ProfileEvents::increment(ProfileEvents::RemoteFSUnusedPrefetches);
|
|
prefetch_future.wait();
|
|
prefetch_future = {};
|
|
}
|
|
}
|
|
|
|
AsynchronousReadBufferFromHDFS::~AsynchronousReadBufferFromHDFS()
|
|
{
|
|
LOG_TEST(
|
|
log,
|
|
"object:{} next_times:{} seek_times:{} interval:{}|{} duration:{}|{} wait:{}|{}",
|
|
reinterpret_cast<std::uintptr_t>(this),
|
|
next_times,
|
|
seek_times,
|
|
sum_interval,
|
|
double(sum_interval)/next_times,
|
|
sum_duration,
|
|
double(sum_duration)/next_times,
|
|
sum_wait,
|
|
double(sum_wait)/next_times);
|
|
|
|
finalize();
|
|
}
|
|
|
|
off_t AsynchronousReadBufferFromHDFS::getPosition()
|
|
{
|
|
return file_offset_of_buffer_end - available();
|
|
}
|
|
|
|
size_t AsynchronousReadBufferFromHDFS::getFileOffsetOfBufferEnd() const
|
|
{
|
|
return file_offset_of_buffer_end;
|
|
}
|
|
|
|
}
|
|
|
|
#endif
|