Future.hpp

// SPDX-License-Identifier: LGPL-3.0-or-later
// Copyright © 2020 Florian Fischer
#pragma once

#include <liburing.h>
#include <sys/socket.h>
#include <sys/types.h>

#include <cerrno>
#include <cstddef>
#include <cstdint>

#include "BinaryPrivateSemaphore.hpp"
#include "CallerEnvironment.hpp"
#include "Debug.hpp"
#include "Emper.hpp"
#include "io/Operation.hpp"

struct __kernel_timespec;

namespace emper::io {

/*
 * @brief Future representing an IO request which can be awaited
 */
class Future : public Logger<LogSubsystem::IO> {
	friend class IoContext;

 protected:
	BPS sem;

	/* IO operation to perform */
	Operation op;

	/* Arguments for the IO operation */
	int fd;
	void* buf;
	size_t len;
	int offsetOrFlags;

	/**
	 * Return value of the operation.
	 * __s32 is used by liburing for the res field in its cqe struct
	 * This value is returned by Future::wait() and can only differentiate between
	 * failure or full success.
	 */
	int32_t returnValue = 0;

	// Dependency futures will be submitted to the io_uring before
	// this and will be linked to this IO request using IOSQE_IO_LINK
	Future* dependency = nullptr;

	/*
	 * @brief prepare a sqe for this IO operation
	 */
	virtual void prepareSqe(struct io_uring_sqe* sqe) = 0;

	/**
	 * @brief Try to complete this IO request using the result of an IO operation.
	 *        If the IO request is completed signal its semaphore otherwise resubmit it.
	 *
	 * @param res Result from an IO operation from a io_uring_cqe or a common syscall
	 * @param syscall Boolean to indicate if the result originated from a syscall or io_uring
	 */
	template <CallerEnvironment callerEnvironment = CallerEnvironment::EMPER>
	void complete(int32_t res, bool syscall = false);

 protected:
	Future(Operation op, int fd, void* buf, size_t len, int offsetOrFlags)
			: op(op), fd(fd), buf(buf), len(len), offsetOrFlags(offsetOrFlags){};

 public:
	virtual ~Future(){};
	/*
	 * @brief reset the Future
	 *
	 * Resetting a Future is useful if we wan't to use the same Future object
	 * multiple times.
	 * A Future for a specific reoccurring IO operation can be created and reused
	 * in a submit, reset loop.
	 */
	inline void reset() {
		LOGD("Resetting Future");

		sem.reset();
	}

	/*
	 * @brief add Future as dependency
	 *
	 * Dependencies which must complete before this future.
	 * This is assured by linking the futures' sqes in the io_uring using the IOSQE_IO_LINK flag.
	 * If a request in the link fails all dependent futures will return -1 and
	 * set errno to ECANCELED
	 *
	 * See: <a href="https://unixism.net/loti/tutorial/link_liburing.html">Liburing linking
	 * requests</a>
	 * The emper equivalent to the example from the liburing documentation.
	 *
	 * @code
	 * int fd = open(FILE_NAME, O_RDWR|O_TRUNC|O_CREAT, 0644);
	 * if (fd < 0 ) {
	 * perror("open");
	 * return 1;
	 * }
	 *
	 * char* msg = "Hallo emper!\n";
	 * WriteFuture writeFuture(fd, &msg, strlen(msg), 0);
	 *
	 * char buf[32];
	 * ReadFuture readFuture(fd, &buf, sizeof(buf), 0);
	 * readFuture.addDependency(writeFuture);
	 *
	 * CloseFuture closeFuture(fd);
	 * closeFuture.addDependency(readFuture);
	 *
	 * // Blocks till all three futures are completed
	 * ssize_t bytes_read = closeFuture.submitAndWait();
	 * ...
	 * @endcode
	 *
	 */
	inline void addDependency(Future& dependency) { this->dependency = &dependency; }

	/*
	 * @brief submit Future for asynchronous completion
	 *
	 * submit must be called withn the runtime because it submitts the future to
	 * the current worker's IoContext
	 */
	virtual void submit();

	/**
	 * @brief Block till the IO request is completed
	 *
	 * @return return the result received from the io_uring
	 */
	auto wait() -> int32_t;

	/**
	 * @brief Block till the IO request is completed and set errno on error
	 *
	 * @return -1 on error and errno will be set, otherwise the return value of the IO request
	 */
	auto waitAndSetErrno() -> ssize_t;

	/*
	 * @brief Equivalent to calling wait() after calling submit()
	 */
	inline auto submitAndWait() -> int32_t {
		submit();
		return wait();
	}
};

class SendFuture : public Future {
 public:
	SendFuture(int socket, const void* buffer, size_t length, int flags)
			: Future(Operation::SEND, socket, const_cast<void*>(buffer), length, flags){};

	void prepareSqe(struct io_uring_sqe* sqe) {
		io_uring_prep_send(sqe, fd, buf, len, offsetOrFlags);
	};

	void submit() {
		if constexpr (emper::IO_TRY_SYSCALL) {
			ssize_t res = ::send(fd, buf, len, offsetOrFlags | MSG_DONTWAIT);

			if (res > -1 || (errno != EAGAIN && errno != EWOULDBLOCK)) {
				complete(res > 0 ? res : -errno);
			}
		}

		Future::submit();
	}
};

class RecvFuture : public Future {
 public:
	RecvFuture(int socket, void* buffer, size_t length, int flags)
			: Future(Operation::RECV, socket, buffer, length, flags){};

	void prepareSqe(struct io_uring_sqe* sqe) {
		io_uring_prep_recv(sqe, fd, buf, len, offsetOrFlags);
	};

	void submit() {
		if constexpr (emper::IO_TRY_SYSCALL) {
			ssize_t res = ::recv(fd, buf, len, offsetOrFlags | MSG_DONTWAIT);

			if (res > -1 || (errno != EAGAIN && errno != EWOULDBLOCK)) {
				complete(res > 0 ? res : -errno);
			}
		}

		Future::submit();
	}
};

class ConnectFuture : public Future {
 public:
	ConnectFuture(int socket, const struct sockaddr* address, socklen_t address_len)
			: Future(Operation::CONNECT, socket, (void*)address, address_len, 0){};

	void prepareSqe(struct io_uring_sqe* sqe) {
		io_uring_prep_connect(sqe, fd, (const struct sockaddr*)buf, (socklen_t)len);
	};
};

class AcceptFuture : public Future {
 public:
	AcceptFuture(int socket, const struct sockaddr* address, socklen_t* address_len)
			: Future(Operation::ACCEPT, socket, (void*)address, (size_t)address_len, 0){};

	void prepareSqe(struct io_uring_sqe* sqe) {
		io_uring_prep_accept(sqe, fd, (struct sockaddr*)buf, (socklen_t*)len, 0);
	};
};

class ReadFuture : public Future {
 public:
	ReadFuture(int fildes, void* buf, size_t nbyte, int offset)
			: Future(Operation::READ, fildes, buf, nbyte, offset){};

	void prepareSqe(struct io_uring_sqe* sqe) {
		io_uring_prep_read(sqe, fd, buf, len, offsetOrFlags);
	};
};

class WriteFuture : public Future {
 public:
	WriteFuture(int fildes, const void* buf, size_t nbyte, int offset)
			: Future(Operation::WRITE, fildes, const_cast<void*>(buf), nbyte, offset){};

	void prepareSqe(struct io_uring_sqe* sqe) {
		io_uring_prep_write(sqe, fd, buf, len, offsetOrFlags);
	};
};

class CloseFuture : public Future {
 public:
	CloseFuture(int fildes) : Future(Operation::CLOSE, fildes, nullptr, 0, 0){};

	void prepareSqe(struct io_uring_sqe* sqe) { io_uring_prep_close(sqe, fd); };
};

/*
 * @brief Add a timeout to any Future not already submitted.
 *
 * A Timeout is added to a Future by inserting an dependent IOURING_OP_LINK_TIMEOUT
 * sqe after the actual IO operation.
 * The dependent timeout will not be started like usual dependent requests after
 * the previous request is completed, it is instead armed immediately when the
 * actual request is started.
 * See: https://lwn.net/Articles/803932/
 *
 * @code
 * ReadFuture rf(fd, buf, buf_len, 0);
 * struct __kernel_timespec ts = {.tv_sec = 0, .tv_nsec = 0);
 * TimeoutWrapper t(rf, ts);
 * // Both futures should be awaited by calling its wait method to
 * // prevent use after free problems.
 * t.submitAndWait()
 * rf.Wait()
 * @endcode
 */
class TimeoutWrapper : public Future {
 public:
	TimeoutWrapper(Future& future, struct __kernel_timespec& ts)
			: Future(Operation::LINK_TIMEOUT, 0, (void*)&ts, 0, 0) {
		addDependency(future);
	};

	void prepareSqe(struct io_uring_sqe* sqe) {
		io_uring_prep_link_timeout(sqe, (struct ::__kernel_timespec*)buf, 0);
	};
};

/*
 * @brief Arm a timeout which will signal the future when it is reached
 *
 * If the timeout was reached it will wait() will return -ETIME.
 */
class AlarmFuture : public Future {
 public:
	AlarmFuture(struct __kernel_timespec& ts) : Future(Operation::TIMEOUT, 0, (void*)&ts, 0, 0){};

	void prepareSqe(struct io_uring_sqe* sqe) {
		io_uring_prep_timeout(sqe, (struct ::__kernel_timespec*)buf, 0, 0);
	};
};
}	 // namespace emper::io