summaryrefslogtreecommitdiff
path: root/tpool/aio_liburing.cc
diff options
context:
space:
mode:
Diffstat (limited to 'tpool/aio_liburing.cc')
-rw-r--r--tpool/aio_liburing.cc209
1 files changed, 209 insertions, 0 deletions
diff --git a/tpool/aio_liburing.cc b/tpool/aio_liburing.cc
new file mode 100644
index 00000000000..447c2335c74
--- /dev/null
+++ b/tpool/aio_liburing.cc
@@ -0,0 +1,209 @@
+/* Copyright (C) 2021, 2022, MariaDB Corporation.
+
+This program is free software; you can redistribute itand /or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation; version 2 of the License.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.See the
+GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License
+along with this program; if not, write to the Free Software
+Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02111 - 1301 USA*/
+
+#include "tpool_structs.h"
+#include "tpool.h"
+#include "mysql/service_my_print_error.h"
+#include "mysqld_error.h"
+
+#include <liburing.h>
+
+#include <algorithm>
+#include <vector>
+#include <thread>
+#include <mutex>
+#include <stdexcept>
+
+namespace
+{
+
+class aio_uring final : public tpool::aio
+{
+public:
+ aio_uring(tpool::thread_pool *tpool, int max_aio) : tpool_(tpool)
+ {
+ if (io_uring_queue_init(max_aio, &uring_, 0) != 0)
+ {
+ switch (const auto e= errno) {
+ case ENOMEM:
+ my_printf_error(ER_UNKNOWN_ERROR,
+ "io_uring_queue_init() failed with ENOMEM:"
+ " try larger memory locked limit, ulimit -l"
+ ", or https://mariadb.com/kb/en/systemd/#configuring-limitmemlock"
+ " under systemd"
+#ifdef HAVE_IO_URING_MLOCK_SIZE
+ " (%zd bytes required)", ME_ERROR_LOG | ME_WARNING,
+ io_uring_mlock_size(max_aio, 0));
+#else
+ , ME_ERROR_LOG | ME_WARNING);
+#endif
+ break;
+ case ENOSYS:
+ my_printf_error(ER_UNKNOWN_ERROR,
+ "io_uring_queue_init() failed with ENOSYS:"
+ " check seccomp filters, and the kernel version "
+ "(newer than 5.1 required)",
+ ME_ERROR_LOG | ME_WARNING);
+ break;
+ default:
+ my_printf_error(ER_UNKNOWN_ERROR,
+ "io_uring_queue_init() failed with errno %d",
+ ME_ERROR_LOG | ME_WARNING, e);
+ }
+ throw std::runtime_error("aio_uring()");
+ }
+ if (io_uring_ring_dontfork(&uring_) != 0)
+ {
+ my_printf_error(ER_UNKNOWN_ERROR,
+ "io_uring_dontfork() failed with errno %d (continuing)",
+ ME_ERROR_LOG | ME_WARNING, errno);
+ }
+
+ thread_= std::thread(thread_routine, this);
+ }
+
+ ~aio_uring() noexcept
+ {
+ {
+ std::lock_guard<std::mutex> _(mutex_);
+ io_uring_sqe *sqe= io_uring_get_sqe(&uring_);
+ io_uring_prep_nop(sqe);
+ io_uring_sqe_set_data(sqe, nullptr);
+ auto ret= io_uring_submit(&uring_);
+ if (ret != 1)
+ {
+ my_printf_error(ER_UNKNOWN_ERROR,
+ "io_uring_submit() returned %d during shutdown:"
+ " this may cause a hang\n",
+ ME_ERROR_LOG | ME_FATAL, ret);
+ abort();
+ }
+ }
+ thread_.join();
+ io_uring_queue_exit(&uring_);
+ }
+
+ int submit_io(tpool::aiocb *cb) final
+ {
+ cb->iov_base= cb->m_buffer;
+ cb->iov_len= cb->m_len;
+
+ // The whole operation since io_uring_get_sqe() and till io_uring_submit()
+ // must be atomical. This is because liburing provides thread-unsafe calls.
+ std::lock_guard<std::mutex> _(mutex_);
+
+ io_uring_sqe *sqe= io_uring_get_sqe(&uring_);
+ if (cb->m_opcode == tpool::aio_opcode::AIO_PREAD)
+ io_uring_prep_readv(sqe, cb->m_fh, static_cast<struct iovec *>(cb), 1,
+ cb->m_offset);
+ else
+ io_uring_prep_writev(sqe, cb->m_fh, static_cast<struct iovec *>(cb), 1,
+ cb->m_offset);
+ io_uring_sqe_set_data(sqe, cb);
+
+ return io_uring_submit(&uring_) == 1 ? 0 : -1;
+ }
+
+ int bind(native_file_handle &fd) final
+ {
+ std::lock_guard<std::mutex> _(files_mutex_);
+ auto it= std::lower_bound(files_.begin(), files_.end(), fd);
+ assert(it == files_.end() || *it != fd);
+ files_.insert(it, fd);
+ return io_uring_register_files_update(&uring_, 0, files_.data(),
+ files_.size());
+ }
+
+ int unbind(const native_file_handle &fd) final
+ {
+ std::lock_guard<std::mutex> _(files_mutex_);
+ auto it= std::lower_bound(files_.begin(), files_.end(), fd);
+ assert(*it == fd);
+ files_.erase(it);
+ return io_uring_register_files_update(&uring_, 0, files_.data(),
+ files_.size());
+ }
+
+private:
+ static void thread_routine(aio_uring *aio)
+ {
+ for (;;)
+ {
+ io_uring_cqe *cqe;
+ if (int ret= io_uring_wait_cqe(&aio->uring_, &cqe))
+ {
+ if (ret == -EINTR)
+ continue;
+ my_printf_error(ER_UNKNOWN_ERROR,
+ "io_uring_wait_cqe() returned %d\n",
+ ME_ERROR_LOG | ME_FATAL, ret);
+ abort();
+ }
+
+ auto *iocb= static_cast<tpool::aiocb*>(io_uring_cqe_get_data(cqe));
+ if (!iocb)
+ break; // ~aio_uring() told us to terminate
+
+ int res= cqe->res;
+ if (res < 0)
+ {
+ iocb->m_err= -res;
+ iocb->m_ret_len= 0;
+ }
+ else
+ {
+ iocb->m_err= 0;
+ iocb->m_ret_len= res;
+ }
+
+ io_uring_cqe_seen(&aio->uring_, cqe);
+ finish_synchronous(iocb);
+
+ // If we need to resubmit the IO operation, but the ring is full,
+ // we will follow the same path as for any other error codes.
+ if (res == -EAGAIN && !aio->submit_io(iocb))
+ continue;
+
+ iocb->m_internal_task.m_func= iocb->m_callback;
+ iocb->m_internal_task.m_arg= iocb;
+ iocb->m_internal_task.m_group= iocb->m_group;
+ aio->tpool_->submit_task(&iocb->m_internal_task);
+ }
+ }
+
+ io_uring uring_;
+ std::mutex mutex_;
+ tpool::thread_pool *tpool_;
+ std::thread thread_;
+
+ std::vector<native_file_handle> files_;
+ std::mutex files_mutex_;
+};
+
+} // namespace
+
+namespace tpool
+{
+
+aio *create_linux_aio(thread_pool *pool, int max_aio)
+{
+ try {
+ return new aio_uring(pool, max_aio);
+ } catch (std::runtime_error& error) {
+ return nullptr;
+ }
+}
+
+} // namespace tpool