mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-09-19 16:20:50 +00:00
Replace asyncronouos buffer with syncronouos
This commit is contained in:
parent
b67b7a62a4
commit
bc5d793382
@ -231,7 +231,6 @@ void ReadBufferFromRemoteFSGather::reset()
|
||||
{
|
||||
current_object = StoredObject();
|
||||
current_buf_idx = {};
|
||||
// buffer_cemetery_.push_back(current_buf);
|
||||
current_buf.reset();
|
||||
}
|
||||
|
||||
|
@ -78,7 +78,6 @@ std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileBase(
|
||||
|
||||
if (settings.local_fs_method == LocalFSReadMethod::read)
|
||||
{
|
||||
LOG_DEBUG(&Poco::Logger::get("Read settings"), "Read settings 1");
|
||||
res = std::make_unique<ReadBufferFromFile>(
|
||||
filename,
|
||||
buffer_size,
|
||||
@ -90,8 +89,6 @@ std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileBase(
|
||||
}
|
||||
else if (settings.local_fs_method == LocalFSReadMethod::pread || settings.local_fs_method == LocalFSReadMethod::mmap)
|
||||
{
|
||||
LOG_DEBUG(&Poco::Logger::get("Read settings"), "Read settings 2");
|
||||
|
||||
res = std::make_unique<ReadBufferFromFilePReadWithDescriptorsCache>(
|
||||
filename,
|
||||
buffer_size,
|
||||
@ -103,7 +100,6 @@ std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileBase(
|
||||
}
|
||||
else if (settings.local_fs_method == LocalFSReadMethod::io_uring)
|
||||
{
|
||||
LOG_DEBUG(&Poco::Logger::get("Read settings"), "Read settings 3");
|
||||
|
||||
#if USE_LIBURING
|
||||
auto & reader = getIOUringReaderOrThrow();
|
||||
@ -123,8 +119,6 @@ std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileBase(
|
||||
}
|
||||
else if (settings.local_fs_method == LocalFSReadMethod::pread_fake_async)
|
||||
{
|
||||
LOG_DEBUG(&Poco::Logger::get("Read settings"), "Read settings 4");
|
||||
|
||||
auto & reader = getThreadPoolReader(FilesystemReaderType::SYNCHRONOUS_LOCAL_FS_READER);
|
||||
res = std::make_unique<AsynchronousReadBufferFromFileWithDescriptorsCache>(
|
||||
reader,
|
||||
@ -139,7 +133,6 @@ std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileBase(
|
||||
}
|
||||
else if (settings.local_fs_method == LocalFSReadMethod::pread_threadpool)
|
||||
{
|
||||
LOG_DEBUG(&Poco::Logger::get("Read settings"), "Read settings 5");
|
||||
auto & reader = getThreadPoolReader(FilesystemReaderType::ASYNCHRONOUS_LOCAL_FS_READER);
|
||||
res = std::make_unique<AsynchronousReadBufferFromFileWithDescriptorsCache>(
|
||||
reader,
|
||||
@ -154,8 +147,6 @@ std::unique_ptr<ReadBufferFromFileBase> createReadBufferFromFileBase(
|
||||
}
|
||||
else
|
||||
{
|
||||
LOG_DEBUG(&Poco::Logger::get("Read settings"), "Read settings 6");
|
||||
|
||||
throw Exception(ErrorCodes::LOGICAL_ERROR, "Unknown read method");
|
||||
}
|
||||
return res;
|
||||
|
@ -43,18 +43,13 @@ bool LocalObjectStorage::exists(const StoredObject & object) const
|
||||
std::unique_ptr<ReadBufferFromFileBase> LocalObjectStorage::readObjects( /// NOLINT
|
||||
const StoredObjects & objects,
|
||||
const ReadSettings & read_settings,
|
||||
std::optional<size_t> read_hint,
|
||||
std::optional<size_t> file_size) const
|
||||
std::optional<size_t>,
|
||||
std::optional<size_t>) const
|
||||
{
|
||||
auto modified_settings = patchSettings(read_settings);
|
||||
auto global_context = Context::getGlobalContextInstance();
|
||||
auto read_buffer_creator = [=](bool /* restricted_seek */, const StoredObject & object) -> std::unique_ptr<ReadBufferFromFileBase>
|
||||
{
|
||||
LOG_DEBUG(&Poco::Logger::get("Get object path"), "Remote Path: {}", object.remote_path);
|
||||
auto kek = createReadBufferFromFileBase(object.remote_path, modified_settings, read_hint, file_size);
|
||||
LOG_DEBUG(&Poco::Logger::get("Buffer created"), "Remote Path: {}", object.remote_path);
|
||||
return kek;
|
||||
};
|
||||
{ return std::make_unique<ReadBufferFromFile>(object.remote_path); };
|
||||
|
||||
return std::make_unique<ReadBufferFromRemoteFSGather>(
|
||||
std::move(read_buffer_creator),
|
||||
|
@ -101,11 +101,7 @@ public:
|
||||
*
|
||||
* Try to read after the end should throw an exception.
|
||||
*/
|
||||
bool ALWAYS_INLINE eof()
|
||||
{
|
||||
LOG_DEBUG();
|
||||
return !hasPendingData() && !next();
|
||||
}
|
||||
bool ALWAYS_INLINE eof() { return !hasPendingData() && !next(); }
|
||||
|
||||
void ignore()
|
||||
{
|
||||
@ -183,8 +179,6 @@ public:
|
||||
|
||||
while (bytes_copied < n && !eof())
|
||||
{
|
||||
auto k = *pos;
|
||||
LOG_DEBUG(&Poco::Logger::get("Next symbol in read"), "Symbol: {}", k);
|
||||
size_t bytes_to_copy = std::min(static_cast<size_t>(working_buffer.end() - pos), n - bytes_copied);
|
||||
::memcpy(to + bytes_copied, pos, bytes_to_copy);
|
||||
pos += bytes_to_copy;
|
||||
|
Loading…
Reference in New Issue
Block a user