2015-03-04 10:47:53 +00:00
|
|
|
|
#include <DB/IO/WriteBufferAIO.h>
|
|
|
|
|
#include <DB/Common/ProfileEvents.h>
|
|
|
|
|
#include <DB/Core/ErrorCodes.h>
|
|
|
|
|
|
2015-03-11 17:27:34 +00:00
|
|
|
|
#include <limits>
|
2015-03-04 10:47:53 +00:00
|
|
|
|
#include <sys/types.h>
|
|
|
|
|
#include <sys/stat.h>
|
|
|
|
|
|
|
|
|
|
namespace DB
|
|
|
|
|
{
|
|
|
|
|
|
|
|
|
|
WriteBufferAIO::WriteBufferAIO(const std::string & filename_, size_t buffer_size_, int flags_, mode_t mode_,
|
|
|
|
|
char * existing_memory_)
|
2015-03-16 21:39:27 +00:00
|
|
|
|
: BufferWithOwnMemory<WriteBuffer>(buffer_size_, existing_memory_, DEFAULT_AIO_FILE_BLOCK_SIZE),
|
2015-03-13 11:02:43 +00:00
|
|
|
|
flush_buffer(BufferWithOwnMemory(buffer_size_, nullptr, DEFAULT_AIO_FILE_BLOCK_SIZE)),
|
2015-03-16 10:49:27 +00:00
|
|
|
|
filename(filename_)
|
2015-03-04 10:47:53 +00:00
|
|
|
|
{
|
|
|
|
|
ProfileEvents::increment(ProfileEvents::FileOpen);
|
|
|
|
|
|
2015-03-24 11:03:26 +00:00
|
|
|
|
int open_flags = (flags_ == -1) ? (O_WRONLY | O_TRUNC | O_CREAT) : flags_;
|
2015-03-04 10:47:53 +00:00
|
|
|
|
open_flags |= O_DIRECT;
|
|
|
|
|
|
2015-03-06 10:29:58 +00:00
|
|
|
|
fd = ::open(filename.c_str(), open_flags, mode_);
|
2015-03-04 10:47:53 +00:00
|
|
|
|
if (fd == -1)
|
2015-03-05 11:57:54 +00:00
|
|
|
|
{
|
|
|
|
|
got_exception = true;
|
2015-03-11 11:31:09 +00:00
|
|
|
|
auto error_code = (errno == ENOENT) ? ErrorCodes::FILE_DOESNT_EXIST : ErrorCodes::CANNOT_OPEN_FILE;
|
|
|
|
|
throwFromErrno("Cannot open file " + filename, error_code);
|
2015-03-05 11:57:54 +00:00
|
|
|
|
}
|
2015-03-11 13:40:15 +00:00
|
|
|
|
|
2015-03-24 11:03:26 +00:00
|
|
|
|
fd2 = ::open(filename.c_str(), O_RDONLY, mode_);
|
|
|
|
|
if (fd2 == -1)
|
|
|
|
|
{
|
|
|
|
|
got_exception = true;
|
|
|
|
|
auto error_code = (errno == ENOENT) ? ErrorCodes::FILE_DOESNT_EXIST : ErrorCodes::CANNOT_OPEN_FILE;
|
|
|
|
|
throwFromErrno("Cannot open file " + filename, error_code);
|
|
|
|
|
}
|
|
|
|
|
|
2015-03-11 13:40:15 +00:00
|
|
|
|
::memset(&request, 0, sizeof(request));
|
2015-03-04 10:47:53 +00:00
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
WriteBufferAIO::~WriteBufferAIO()
|
|
|
|
|
{
|
2015-03-05 11:57:54 +00:00
|
|
|
|
if (!got_exception)
|
2015-03-04 10:47:53 +00:00
|
|
|
|
{
|
2015-03-05 11:57:54 +00:00
|
|
|
|
try
|
|
|
|
|
{
|
2015-03-13 11:46:04 +00:00
|
|
|
|
flush();
|
2015-03-05 11:57:54 +00:00
|
|
|
|
}
|
|
|
|
|
catch (...)
|
|
|
|
|
{
|
|
|
|
|
tryLogCurrentException(__PRETTY_FUNCTION__);
|
|
|
|
|
}
|
2015-03-04 10:47:53 +00:00
|
|
|
|
}
|
|
|
|
|
|
2015-03-05 11:57:54 +00:00
|
|
|
|
if (fd != -1)
|
|
|
|
|
::close(fd);
|
2015-03-25 14:15:50 +00:00
|
|
|
|
if (fd2 != -1)
|
|
|
|
|
::close(fd2);
|
2015-03-04 10:47:53 +00:00
|
|
|
|
}
|
|
|
|
|
|
2015-03-10 16:34:00 +00:00
|
|
|
|
off_t WriteBufferAIO::seek(off_t off, int whence)
|
2015-03-05 14:21:55 +00:00
|
|
|
|
{
|
2015-03-13 11:46:04 +00:00
|
|
|
|
flush();
|
2015-03-05 14:21:55 +00:00
|
|
|
|
|
2015-03-11 17:27:34 +00:00
|
|
|
|
if (whence == SEEK_SET)
|
|
|
|
|
{
|
|
|
|
|
if (off < 0)
|
|
|
|
|
{
|
|
|
|
|
got_exception = true;
|
|
|
|
|
throw Exception("SEEK_SET underflow", ErrorCodes::ARGUMENT_OUT_OF_BOUND);
|
|
|
|
|
}
|
|
|
|
|
pos_in_file = off;
|
|
|
|
|
}
|
|
|
|
|
else if (whence == SEEK_CUR)
|
|
|
|
|
{
|
|
|
|
|
if (off >= 0)
|
|
|
|
|
{
|
|
|
|
|
if (off > (std::numeric_limits<off_t>::max() - pos_in_file))
|
|
|
|
|
{
|
|
|
|
|
got_exception = true;
|
|
|
|
|
throw Exception("SEEK_CUR overflow", ErrorCodes::ARGUMENT_OUT_OF_BOUND);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
else if (off < -pos_in_file)
|
|
|
|
|
{
|
|
|
|
|
got_exception = true;
|
|
|
|
|
throw Exception("SEEK_CUR underflow", ErrorCodes::ARGUMENT_OUT_OF_BOUND);
|
|
|
|
|
}
|
|
|
|
|
pos_in_file += off;
|
|
|
|
|
}
|
|
|
|
|
else
|
2015-03-05 14:21:55 +00:00
|
|
|
|
{
|
|
|
|
|
got_exception = true;
|
2015-03-11 17:27:34 +00:00
|
|
|
|
throw Exception("WriteBufferAIO::seek expects SEEK_SET or SEEK_CUR as whence", ErrorCodes::ARGUMENT_OUT_OF_BOUND);
|
2015-03-05 14:21:55 +00:00
|
|
|
|
}
|
2015-03-11 17:27:34 +00:00
|
|
|
|
|
2015-03-25 14:14:06 +00:00
|
|
|
|
if (pos_in_file > max_pos)
|
|
|
|
|
max_pos = pos_in_file;
|
|
|
|
|
|
2015-03-11 17:27:34 +00:00
|
|
|
|
return pos_in_file;
|
2015-03-05 14:21:55 +00:00
|
|
|
|
}
|
|
|
|
|
|
2015-03-13 11:02:43 +00:00
|
|
|
|
off_t WriteBufferAIO::getPositionInFile()
|
|
|
|
|
{
|
|
|
|
|
return seek(0, SEEK_CUR);
|
|
|
|
|
}
|
|
|
|
|
|
2015-03-05 14:21:55 +00:00
|
|
|
|
void WriteBufferAIO::truncate(off_t length)
|
|
|
|
|
{
|
2015-03-13 11:46:04 +00:00
|
|
|
|
flush();
|
2015-03-05 14:21:55 +00:00
|
|
|
|
|
|
|
|
|
int res = ::ftruncate(fd, length);
|
|
|
|
|
if (res == -1)
|
|
|
|
|
{
|
|
|
|
|
got_exception = true;
|
2015-03-06 10:29:58 +00:00
|
|
|
|
throwFromErrno("Cannot truncate file " + filename, ErrorCodes::CANNOT_TRUNCATE_FILE);
|
2015-03-05 14:21:55 +00:00
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
2015-03-10 16:34:00 +00:00
|
|
|
|
void WriteBufferAIO::sync()
|
2015-03-04 10:47:53 +00:00
|
|
|
|
{
|
2015-03-13 11:46:04 +00:00
|
|
|
|
flush();
|
2015-03-10 17:14:07 +00:00
|
|
|
|
|
|
|
|
|
/// Попросим ОС сбросить данные на диск.
|
|
|
|
|
int res = ::fsync(fd);
|
|
|
|
|
if (res == -1)
|
2015-03-11 13:15:19 +00:00
|
|
|
|
{
|
|
|
|
|
got_exception = true;
|
2015-03-10 17:14:07 +00:00
|
|
|
|
throwFromErrno("Cannot fsync " + getFileName(), ErrorCodes::CANNOT_FSYNC);
|
2015-03-11 13:15:19 +00:00
|
|
|
|
}
|
2015-03-04 10:47:53 +00:00
|
|
|
|
}
|
|
|
|
|
|
2015-03-13 11:46:04 +00:00
|
|
|
|
void WriteBufferAIO::flush()
|
|
|
|
|
{
|
|
|
|
|
next();
|
|
|
|
|
waitForAIOCompletion();
|
|
|
|
|
}
|
|
|
|
|
|
2015-03-04 10:47:53 +00:00
|
|
|
|
void WriteBufferAIO::nextImpl()
|
|
|
|
|
{
|
|
|
|
|
if (!offset())
|
|
|
|
|
return;
|
|
|
|
|
|
2015-03-13 11:46:04 +00:00
|
|
|
|
waitForAIOCompletion();
|
2015-03-04 10:47:53 +00:00
|
|
|
|
swapBuffers();
|
|
|
|
|
|
2015-03-20 16:58:57 +00:00
|
|
|
|
truncate_count = 0;
|
|
|
|
|
|
2015-03-25 14:14:06 +00:00
|
|
|
|
/// Регион диска, в который хотим записать данные.
|
2015-03-24 11:03:26 +00:00
|
|
|
|
off_t region_begin = pos_in_file;
|
|
|
|
|
off_t region_end = pos_in_file + flush_buffer.offset();
|
|
|
|
|
size_t region_size = region_end - region_begin;
|
2015-03-19 10:06:11 +00:00
|
|
|
|
|
2015-03-25 14:14:06 +00:00
|
|
|
|
/// Регион диска, в который действительно записываем данные.
|
2015-03-24 11:03:26 +00:00
|
|
|
|
size_t region_left_padding = region_begin % DEFAULT_AIO_FILE_BLOCK_SIZE;
|
|
|
|
|
size_t region_right_padding = 0;
|
|
|
|
|
if (region_end % DEFAULT_AIO_FILE_BLOCK_SIZE != 0)
|
|
|
|
|
region_right_padding = DEFAULT_AIO_FILE_BLOCK_SIZE - (region_end % DEFAULT_AIO_FILE_BLOCK_SIZE);
|
2015-03-19 10:06:11 +00:00
|
|
|
|
|
2015-03-24 11:03:26 +00:00
|
|
|
|
off_t region_aligned_begin = region_begin - region_left_padding;
|
|
|
|
|
off_t region_aligned_end = region_end + region_right_padding;
|
|
|
|
|
size_t region_aligned_size = region_aligned_end - region_aligned_begin;
|
2015-03-04 10:47:53 +00:00
|
|
|
|
|
2015-03-24 11:03:26 +00:00
|
|
|
|
/// Буфер данных, которые хотим записать на диск.
|
|
|
|
|
Position buffer_begin = flush_buffer.buffer().begin();
|
|
|
|
|
Position buffer_end = buffer_begin + region_size;
|
|
|
|
|
size_t buffer_size = buffer_end - buffer_begin;
|
|
|
|
|
size_t buffer_capacity = flush_buffer.buffer().size();
|
2015-03-19 10:06:11 +00:00
|
|
|
|
|
2015-03-25 14:17:51 +00:00
|
|
|
|
/// Обработать буфер, чтобы он отражал структуру региона диска.
|
2015-03-19 10:06:11 +00:00
|
|
|
|
|
2015-03-25 14:14:06 +00:00
|
|
|
|
size_t excess = 0;
|
2015-03-24 11:03:26 +00:00
|
|
|
|
if (region_left_padding > 0)
|
2015-03-06 10:29:58 +00:00
|
|
|
|
{
|
2015-03-24 11:03:26 +00:00
|
|
|
|
if ((region_left_padding + buffer_size) > buffer_capacity)
|
|
|
|
|
{
|
2015-03-25 14:14:06 +00:00
|
|
|
|
excess = region_left_padding + buffer_size - buffer_capacity;
|
2015-03-24 11:03:26 +00:00
|
|
|
|
::memset(&memory_page[0], 0, memory_page.size());
|
2015-03-25 14:14:06 +00:00
|
|
|
|
::memcpy(&memory_page[0], buffer_end - excess, excess);
|
2015-03-24 11:03:26 +00:00
|
|
|
|
buffer_end = buffer_begin + buffer_capacity;
|
|
|
|
|
buffer_size = buffer_capacity;
|
|
|
|
|
}
|
|
|
|
|
else
|
|
|
|
|
{
|
|
|
|
|
buffer_size += region_left_padding;
|
|
|
|
|
buffer_end = buffer_begin + buffer_size;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
::memmove(buffer_begin + region_left_padding, buffer_begin, buffer_size - region_left_padding);
|
|
|
|
|
::memset(buffer_begin, 0, region_left_padding);
|
|
|
|
|
|
|
|
|
|
ssize_t read_count = ::pread(fd2, buffer_begin, region_left_padding, region_aligned_begin);
|
2015-03-19 10:06:11 +00:00
|
|
|
|
if (read_count < 0)
|
2015-03-24 11:03:26 +00:00
|
|
|
|
{
|
|
|
|
|
got_exception = true;
|
2015-03-20 16:58:57 +00:00
|
|
|
|
throw Exception("Read error", ErrorCodes::AIO_READ_ERROR);
|
2015-03-24 11:03:26 +00:00
|
|
|
|
}
|
2015-03-19 10:06:11 +00:00
|
|
|
|
}
|
|
|
|
|
|
2015-03-24 11:03:26 +00:00
|
|
|
|
if (region_right_padding > 0)
|
2015-03-19 10:06:11 +00:00
|
|
|
|
{
|
2015-03-25 14:14:06 +00:00
|
|
|
|
Position end_ptr;
|
|
|
|
|
if (excess > 0)
|
|
|
|
|
end_ptr = &memory_page[excess];
|
|
|
|
|
else
|
|
|
|
|
end_ptr = buffer_end;
|
|
|
|
|
|
2015-03-24 11:03:26 +00:00
|
|
|
|
::memset(end_ptr, 0, region_right_padding);
|
|
|
|
|
ssize_t read_count = ::pread(fd2, end_ptr, region_right_padding, region_end);
|
|
|
|
|
if (read_count < 0)
|
2015-03-25 14:14:06 +00:00
|
|
|
|
{
|
|
|
|
|
got_exception = true;
|
|
|
|
|
throw Exception("Read error", ErrorCodes::AIO_READ_ERROR);
|
|
|
|
|
}
|
|
|
|
|
truncate_count = region_right_padding - read_count;
|
2015-03-19 10:06:11 +00:00
|
|
|
|
}
|
|
|
|
|
|
2015-03-24 11:03:26 +00:00
|
|
|
|
/// Создать запрос на асинхронную запись.
|
2015-03-19 10:06:11 +00:00
|
|
|
|
|
2015-03-24 11:03:26 +00:00
|
|
|
|
size_t i = 0;
|
2015-03-19 10:06:11 +00:00
|
|
|
|
|
2015-03-24 11:03:26 +00:00
|
|
|
|
iov[i].iov_base = buffer_begin;
|
2015-03-25 14:14:06 +00:00
|
|
|
|
iov[i].iov_len = ((excess > 0) ? buffer_capacity : region_aligned_size);
|
2015-03-19 10:06:11 +00:00
|
|
|
|
++i;
|
|
|
|
|
|
2015-03-25 14:14:06 +00:00
|
|
|
|
if (excess > 0)
|
2015-03-19 10:06:11 +00:00
|
|
|
|
{
|
2015-03-24 11:03:26 +00:00
|
|
|
|
iov[i].iov_base = &memory_page[0];
|
2015-03-25 14:14:06 +00:00
|
|
|
|
iov[i].iov_len = memory_page.size();
|
2015-03-23 09:43:06 +00:00
|
|
|
|
++i;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
bytes_to_write = 0;
|
|
|
|
|
for (size_t j = 0; j < i; ++j)
|
|
|
|
|
{
|
2015-03-23 09:52:13 +00:00
|
|
|
|
if ((iov[j].iov_len > std::numeric_limits<off_t>::max()) ||
|
|
|
|
|
(static_cast<off_t>(iov[j].iov_len) > (std::numeric_limits<off_t>::max() - bytes_to_write)))
|
2015-03-23 09:43:06 +00:00
|
|
|
|
{
|
|
|
|
|
got_exception = true;
|
|
|
|
|
throw Exception("Overflow on bytes to write", ErrorCodes::LOGICAL_ERROR);
|
|
|
|
|
}
|
2015-03-23 09:52:13 +00:00
|
|
|
|
bytes_to_write += iov[j].iov_len;
|
2015-03-06 10:29:58 +00:00
|
|
|
|
}
|
|
|
|
|
|
2015-03-19 10:06:11 +00:00
|
|
|
|
request.aio_lio_opcode = IOCB_CMD_PWRITEV;
|
|
|
|
|
request.aio_fildes = fd;
|
|
|
|
|
request.aio_buf = reinterpret_cast<UInt64>(iov);
|
2015-03-23 09:43:06 +00:00
|
|
|
|
request.aio_nbytes = i;
|
2015-03-19 10:06:11 +00:00
|
|
|
|
request.aio_offset = region_aligned_begin;
|
|
|
|
|
|
2015-03-10 17:14:07 +00:00
|
|
|
|
/// Отправить запрос.
|
2015-03-04 10:47:53 +00:00
|
|
|
|
while (io_submit(aio_context.ctx, request_ptrs.size(), &request_ptrs[0]) < 0)
|
2015-03-16 10:41:58 +00:00
|
|
|
|
{
|
2015-03-04 10:47:53 +00:00
|
|
|
|
if (errno != EINTR)
|
2015-03-05 11:57:54 +00:00
|
|
|
|
{
|
|
|
|
|
got_exception = true;
|
2015-03-06 10:29:58 +00:00
|
|
|
|
throw Exception("Cannot submit request for asynchronous IO on file " + filename, ErrorCodes::AIO_SUBMIT_ERROR);
|
2015-03-05 11:57:54 +00:00
|
|
|
|
}
|
2015-03-16 10:41:58 +00:00
|
|
|
|
}
|
2015-03-04 10:47:53 +00:00
|
|
|
|
|
|
|
|
|
is_pending_write = true;
|
|
|
|
|
}
|
|
|
|
|
|
2015-03-13 11:46:04 +00:00
|
|
|
|
void WriteBufferAIO::waitForAIOCompletion()
|
2015-03-04 10:47:53 +00:00
|
|
|
|
{
|
|
|
|
|
if (is_pending_write)
|
|
|
|
|
{
|
|
|
|
|
while (io_getevents(aio_context.ctx, events.size(), events.size(), &events[0], nullptr) < 0)
|
2015-03-16 10:41:58 +00:00
|
|
|
|
{
|
2015-03-04 10:47:53 +00:00
|
|
|
|
if (errno != EINTR)
|
2015-03-05 11:57:54 +00:00
|
|
|
|
{
|
|
|
|
|
got_exception = true;
|
2015-03-06 10:29:58 +00:00
|
|
|
|
throw Exception("Failed to wait for asynchronous IO completion on file " + filename, ErrorCodes::AIO_COMPLETION_ERROR);
|
2015-03-05 11:57:54 +00:00
|
|
|
|
}
|
2015-03-16 10:41:58 +00:00
|
|
|
|
}
|
2015-03-05 11:57:54 +00:00
|
|
|
|
|
2015-03-11 13:15:19 +00:00
|
|
|
|
is_pending_write = false;
|
2015-03-13 12:27:31 +00:00
|
|
|
|
off_t bytes_written = events[0].res;
|
2015-03-13 13:14:31 +00:00
|
|
|
|
|
2015-03-23 09:43:06 +00:00
|
|
|
|
if (bytes_written < bytes_to_write)
|
2015-03-11 13:15:19 +00:00
|
|
|
|
{
|
|
|
|
|
got_exception = true;
|
|
|
|
|
throw Exception("Asynchronous write error on file " + filename, ErrorCodes::AIO_WRITE_ERROR);
|
|
|
|
|
}
|
2015-03-20 16:58:57 +00:00
|
|
|
|
|
2015-03-24 11:03:26 +00:00
|
|
|
|
bytes_written -= truncate_count;
|
|
|
|
|
if (pos_in_file > (std::numeric_limits<off_t>::max() - bytes_written))
|
2015-03-20 16:58:57 +00:00
|
|
|
|
{
|
2015-03-24 11:03:26 +00:00
|
|
|
|
got_exception = true;
|
|
|
|
|
throw Exception("File position overflowed", ErrorCodes::LOGICAL_ERROR);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
off_t delta = pos_in_file - request.aio_offset;
|
|
|
|
|
pos_in_file += bytes_written - delta;
|
|
|
|
|
|
|
|
|
|
if (pos_in_file > max_pos)
|
|
|
|
|
max_pos = pos_in_file;
|
2015-03-23 09:43:06 +00:00
|
|
|
|
|
2015-03-24 11:03:26 +00:00
|
|
|
|
if (truncate_count > 0)
|
|
|
|
|
{
|
|
|
|
|
// Delete the trailing zeroes that were added for alignment purposes.
|
|
|
|
|
int res = ::ftruncate(fd, max_pos);
|
2015-03-20 16:58:57 +00:00
|
|
|
|
if (res == -1)
|
|
|
|
|
{
|
|
|
|
|
got_exception = true;
|
|
|
|
|
throwFromErrno("Cannot truncate file " + filename, ErrorCodes::CANNOT_TRUNCATE_FILE);
|
|
|
|
|
}
|
|
|
|
|
}
|
2015-03-04 10:47:53 +00:00
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
2015-03-05 11:57:54 +00:00
|
|
|
|
void WriteBufferAIO::swapBuffers() noexcept
|
2015-03-04 10:47:53 +00:00
|
|
|
|
{
|
2015-03-05 08:18:16 +00:00
|
|
|
|
buffer().swap(flush_buffer.buffer());
|
|
|
|
|
std::swap(position(), flush_buffer.position());
|
2015-03-04 10:47:53 +00:00
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|