sigmawall/sigmawall.cpp

#include <stdio.h>
#include <llhttp.h>
#include <uv.h>

#include <coroutine>
#include <type_traits>
#include <optional>
#include <memory>
#include <queue>
#include <cassert>
#include <variant>
#include <unordered_map>

using owned_coroutine = std::shared_ptr<std::coroutine_handle<>>;

template <typename T>
class task;

template <typename T>
class task_internal {
	public:
		class promise_type;
		using handle_type = std::coroutine_handle<promise_type>;
		using value_type = T;

	private:
		struct handle_holder {
			std::coroutine_handle<> handle;
			handle_type h_typed;
		};
		static void handle_deleter(void *p) {
			auto ptr = (handle_holder *)p;
			ptr->handle.destroy();
			delete ptr;
		}

	public:
		class promise_type {
			private:
				friend class task_internal;
				friend class task<T>::promise_type;
				bool returned = false;
				std::queue<std::coroutine_handle<>> resumables;
				std::weak_ptr<handle_holder> handle;
				std::shared_ptr<handle_holder> parked_handle;
				owned_coroutine continuation;
				void return_any() {
					returned = true;
				}
			public:
				task_internal get_return_object() {
					auto handle_ret = handle.lock();
					assert(handle_ret);
					if (parked_handle) {
						parked_handle = {};
					}
					return { handle_ret };
				}
				promise_type() {
					handle_type handle_ret = handle_type::from_promise(*this);
					handle_holder *hol = new handle_holder;
					hol->handle = handle_ret;
					hol->h_typed = handle_ret;
					parked_handle = std::shared_ptr<handle_holder>(
						hol, handle_deleter
					);
					handle = parked_handle;
				}
				class suspend_return {
					private:
						friend class promise_type;
						owned_coroutine continuation;
						suspend_return(owned_coroutine continuation) : continuation(continuation) {}
					public:
						bool await_ready() noexcept { return false; }
						template <class U>
						bool await_suspend(std::coroutine_handle<U> &paused) noexcept {
							(void)paused;
							if (continuation)
								continuation->resume();
							return true;
						}
						void await_resume() noexcept {}
				};
				std::suspend_never initial_suspend() noexcept { return {}; }
				suspend_return final_suspend() noexcept {
					return { continuation };
				}

				void unhandled_exception() {}

				owned_coroutine owned_handle() {
					std::shared_ptr<handle_holder> xhandle = handle.lock();
					assert(xhandle);
					auto yhandle = std::reinterpret_pointer_cast<std::coroutine_handle<>>(xhandle);
					assert(yhandle);
					return yhandle;
				}
		};

		bool await_ready() { return handle->h_typed.promise().returned; }

		template <class U>
		bool await_suspend(std::coroutine_handle<U> &paused) {
			assert(!handle->h_typed.promise().continuation);
			handle->h_typed.promise().continuation = paused.promise().owned_handle();
			assert(handle->h_typed.promise().continuation.use_count());
			return true;
		}

	private:
		friend class task<T>;
		std::shared_ptr<handle_holder> handle;
		task_internal(std::shared_ptr<handle_holder> handle) : handle(handle) {}
		task_internal(const task_internal &src) : handle(src.handle) {}
};

template <typename T>
class task : public task_internal<T> {
	public:
		class promise_type;
		using handle_type = std::coroutine_handle<promise_type>;
		class promise_type : public task_internal<T>::promise_type {
			private:
				friend class task;
				std::optional<T> value;
			public:
				task get_return_object() {
					return { task_internal<T>::promise_type::get_return_object() };
				}
				void return_value(T uvalue) {
					value = uvalue;
					return task_internal<T>::promise_type::return_any();
				}
		};

		T await_resume() {
			std::optional<T> &val = handle_type::from_address(task_internal<T>::handle->h_typed.address()).promise().value;
			assert(val.has_value());
			return *val;
		}

	private:
		task(const task_internal<T> &src) : task_internal<T>(src) {}
};

template <>
class task<void> : public task_internal<void> {
	public:
		class promise_type : public task_internal<void>::promise_type {
			public:
				task get_return_object() {
					return { task_internal<void>::promise_type::get_return_object() };
				}
				void return_void() {
					return task_internal<void>::promise_type::return_any();
				}
		};

		void await_resume() { }

	private:
		task(const task_internal<void> &src) : task_internal<void>(src) {}
};

template <typename F>
class awaiter {
	F fun;
	public:
		awaiter(F fun) : fun(fun) {}
		bool await_ready() { return false; }

		template <typename U>
		bool await_suspend(std::coroutine_handle<U> &handle) {
			auto xhandle = handle.promise().owned_handle();
			return fun(xhandle);
		}

		void await_resume() {}
};

template<class T> struct R_Ok { T v; R_Ok(T v) : v(v) {} };
template<class T> struct R_Err { T v; R_Err(T v) : v(v) {} };
template<> struct R_Ok<void> {};
template<> struct R_Err<void> {};
using R_VOk = R_Ok<void>;
using R_VErr = R_Err<void>;

template <class TOk, class TErr>
class Result {
	private:
		struct Ok_h { TOk v; };
		struct Err_h { TErr v; };
		std::variant<Ok_h, Err_h> val;
		Result() {}

	public:
		Result(R_Ok<TOk> v) : val((Ok_h){v.v}) {}
		Result(R_Err<TErr> v) : val((Err_h){v.v}) {}
		static Result Ok(TOk v) { return { R_Ok(v) }; }
		static Result Err(TErr v) { return { R_Err(v) }; }
		operator bool() { return !!std::get_if<Ok_h>(&val); }
		TOk &ok() { return std::get<Ok_h>(val).v; }
		TErr &err() { return std::get<Err_h>(val).v; }

		template<class UOk, class F>
		Result<UOk, TErr> map_ok(F f) {
			if (*this) {
				return R_Ok(f(ok()));
			}
			return R_Err(err());
		}
};

template<class TErr>
class Result<void, TErr> {
	private:
		std::optional<TErr> val;
		Result() {}

	public:
		Result(R_Ok<void> v) { (void)v; }
		Result(R_Err<TErr> v) : val(v.v) {}
		static Result Ok() { return { R_Ok<void>() }; }
		static Result Err(TErr v) { return { R_Err(v) }; }
		operator bool() { return !val; }
		TErr &err() { return *val; }
};

class comutex {
	private:
		std::queue<owned_coroutine> coros;
		bool locked = false;

	public:
		class lock_type {
			private:
				friend class comutex;
				std::optional<comutex*> owner;

			public:
				void unlock() {
					if (owner) {
						(*owner)->locked = false;
						auto &coros = (*owner)->coros;
						for (; !coros.empty(); coros.pop()) {
							auto &coro = coros.front();
							coro->resume();
						}
						owner = {};
					}
				}

				~lock_type() {
					unlock();
				}
		};

		task<lock_type> lock() {
			if (locked) {
				co_await awaiter([&](owned_coroutine handle){
					coros.push(handle);
					return true;
				});
			}
			locked = true;
			lock_type lock_v;
			lock_v.owner = this;
			co_return lock_v;
		}
};

class UVLoop {
	private:
		uv_loop_t loop;

		struct tireq {
			uv_timer_t req;
			owned_coroutine coro;
		};
		static void on_timer(uv_timer_t *req) {
			auto &tire = *(tireq*)req;
			if (tire.coro)
				tire.coro->resume();
		}

	public:
		UVLoop() {
			uv_loop_init(&loop);
		}
		~UVLoop() {
			if (uv_loop_close(&loop)) {
				abort();
			}
		}
		uv_loop_t &operator*() {
			return loop;
		}
		task<void> delay(uint64_t time) {
			tireq req;
			uv_timer_init(&**this, &req.req);
			uv_timer_start(&req.req, on_timer, time, 0);
			co_await awaiter([&](owned_coroutine handle){
				req.coro = handle;
				return true;
			});
		}
};

class UVHandle {
	protected:
		struct innerT {
			uv_any_handle handle;
			owned_coroutine closer;
			int (*close_fn)(uv_handle_t*, uv_close_cb) = NULL;
			bool closed = false;
		};
		std::shared_ptr<innerT> inner;

		static task<void> close_inno(innerT &inno) {
			if (inno.closed) co_return;
			if ((!inno.close_fn)||(inno.close_fn((uv_handle_t*)&inno, on_close)))
				uv_close((uv_handle_t*)&inno, on_close);
			co_await awaiter([&](owned_coroutine handle){
				inno.closer = handle;
				return true;
			});
			inno.closed = true;
		}

		template <class T>
		static task<void> destroy(T *ptr) {
			std::unique_ptr<T> holder { ptr };
			innerT &inno = *(innerT*)ptr;
			co_await close_inno(inno);
		}

		template <class innerT>
		innerT &init_inner() {
			auto inno = new innerT;
			std::shared_ptr<innerT> inner_o = { inno, destroy<innerT> };
			inner = std::reinterpret_pointer_cast<UVHandle::innerT>(inner_o);
			return *(innerT*)&*inner;
		}

		void init_inner_from(const UVHandle &src) {
			inner = src.inner;
		}

		static void on_close(uv_handle_t *handle) {
			innerT &inno = *(innerT*)handle;
			auto coro = inno.closer;
			inno.closer = {};
			coro->resume();
		}

	public:
		UVHandle() {}

		task<void> close() {
			innerT &inno = *(innerT*)&*inner;
			return close_inno(inno);
		}
};

class UVStream : public UVHandle {
	protected:
		struct innerT {
			UVHandle::innerT base;
			struct respT {
				owned_coroutine coro;
				union out_type {
					int status;
				} *out;
			};
			std::unordered_map<uv_req_t*, respT> reqs;
			bool listening = false;
			const int *stream_type_tag;
			std::queue<int> pending_accepts;
			owned_coroutine acceptor;
			owned_coroutine reader;
			ssize_t *nreadout;
			char *rbuf;
			size_t rbufsize;
			comutex readlock;
			comutex acceptlock;
			UVStream (*make_stream)(uv_loop_t*);
			uv_loop_t *loop;
		};

		void init_with_tag(const UVStream &rhs, int *stream_type_tag) {
			init_inner_from(rhs);
			if (!inner) return;
			auto &inno = *(innerT*)&*inner;
			if (inno.stream_type_tag != stream_type_tag)
				throw std::bad_cast();
		}

		static void on_alloc(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf) {
			(void)suggested_size;
			innerT &inno = *(innerT*)handle;
			buf->base = inno.rbuf;
			buf->len = inno.rbufsize;
		}

		static void on_read(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf) {
			(void)buf;
			innerT &inno = *(innerT*)stream;
			if (inno.reader) {
				auto coro = inno.reader;
				inno.reader = {};
				*inno.nreadout = nread;
				coro->resume();
			}
		}

		static void on_new_connection(uv_stream_t *server, int status) {
			innerT &inno = *(innerT*)server;
			inno.pending_accepts.push(status);
			if (inno.acceptor) {
				auto coro = inno.acceptor;
				inno.acceptor = {};
				coro->resume();
			}
		}

		struct poreque {
			uv_any_req req;
			innerT *inno;
			poreque(innerT &inno) : inno(&inno) {}
		};

		template <class reqT>
		static void on_req(reqT *req, int status) {
			innerT &inno = *((poreque*)req)->inno;
			auto freq = inno.reqs.find((uv_req_t*)req);
			if (freq != inno.reqs.end()) {
				auto coro = freq->second.coro;
				auto out = freq->second.out;
				inno.reqs.erase(freq);
				out->status = status;
				coro->resume();
			}
		}

	public:
		UVStream() {}

		task<Result<void, int>> shutdown() {
			innerT &inno = *(innerT*)&*inner;
			poreque req(inno);
			int ret = uv_shutdown((uv_shutdown_t*)&req, (uv_stream_t*)&inno, on_req<uv_shutdown_t>);
			if (ret) {
				co_return R_Err(ret);
			}
			co_await awaiter([&](owned_coroutine handle){
				inno.reqs.insert(std::pair((uv_req_t*)&req, (innerT::respT){
					handle, (innerT::respT::out_type*)&ret
				}));
				return true;
			});
			if (ret)
				co_return R_Err(ret);
			co_return R_VOk();
		}

		Result<void, int> listen(int backlog) {
			innerT &inno = *(innerT*)&*inner;
			int ret = uv_listen((uv_stream_t*)&inno, backlog, on_new_connection);
			if (ret)
				return R_Err(ret);
			inno.listening = true;
			return R_VOk();
		}

		task<Result<UVStream, int>> accept() {
			innerT &inno = *(innerT*)&*inner;
			int status;
			if (inno.listening) {
				auto lock = co_await inno.acceptlock.lock();
				while (inno.pending_accepts.empty())
					co_await awaiter([&](owned_coroutine handle){
						inno.acceptor = handle;
						return true;
					});

				int status = inno.pending_accepts.front();
				inno.pending_accepts.pop();
				if (status)
					co_return R_Err(status);
			}
			UVStream stream = inno.make_stream(inno.loop);
			status = uv_accept((uv_stream_t*)&inno, (uv_stream_t*)&*stream.inner);
			if (status)
				co_return R_Err(status);
			co_return R_Ok(stream);
		}

		task<Result<size_t, int>> read(char *buf, size_t size) {
			innerT &inno = *(innerT*)&*inner;
			auto lock = co_await inno.readlock.lock();
			ssize_t nread;
			inno.rbuf = buf;
			inno.rbufsize = size;
			inno.nreadout = &nread;
			uv_read_start((uv_stream_t*)&inno, on_alloc, on_read);
			co_await awaiter([&](owned_coroutine handle){
				inno.reader = handle;
				return true;
			});
			uv_read_stop((uv_stream_t*)&inno);
			if (nread < 0)
				co_return R_Err((int)nread);
			co_return R_Ok((size_t)nread);
		}

		task<Result<void, int>> write(char *buf, size_t size) {
			innerT &inno = *(innerT*)&*inner;
			poreque req(inno);
			uv_buf_t ubuf;
			ubuf.base = buf;
			ubuf.len = size;
			int ret = uv_write((uv_write_t*)&req, (uv_stream_t*)&inno, &ubuf, 1, on_req<uv_write_t>);
			if (ret) {
				co_return R_Err(ret);
			}
			co_await awaiter([&](owned_coroutine handle){
				inno.reqs.insert(std::pair((uv_req_t*)&req, (innerT::respT){
					handle, (innerT::respT::out_type*)&ret
				}));
				return true;
			});
			if (ret)
				co_return R_Err(ret);
			co_return R_VOk();
		}
};

class UVTcp : public UVStream {
	protected:
		struct innerT {
			UVStream::innerT base;
			innerT() {
				base.make_stream = make_stream;
				base.stream_type_tag = &stream_type_tag;
				base.base.close_fn = tcp_closer;
			}
		};

		static int stream_type_tag;

		static int tcp_closer(uv_handle_t *handle, uv_close_cb close_cb) {
			return uv_tcp_close_reset((uv_tcp_t*)handle, close_cb);
		}

		static UVStream make_stream(uv_loop_t *loop) {
			UVTcp stream;
			auto &inno = stream.init_inner<innerT>();
			inno.base.loop = loop;
			inno.base.stream_type_tag = &stream_type_tag;
			uv_tcp_init(loop, (uv_tcp_t*)&inno);
			return stream;
		}

	public:
		UVTcp() {}
		UVTcp(const UVStream &rhs) { init_with_tag(rhs, &stream_type_tag); }

		static Result<UVTcp, int> bind(uv_loop_t *loop, const char *ip, short port) {
			UVTcp tcp = {make_stream(loop)};
			auto &inno = *(innerT*)&*tcp.inner;
			struct sockaddr_in addr;
			struct sockaddr_in6 addr6;
			int err = 0;
			if (!(err = uv_ip6_addr(ip, port, &addr6))) {
				err = uv_tcp_bind((uv_tcp_t*)&inno, (const struct sockaddr*)&addr6, 0);
			} else if (!(err = uv_ip4_addr(ip, port, &addr))) {
				err = uv_tcp_bind((uv_tcp_t*)&inno, (const struct sockaddr*)&addr, 0);
			}
			if (err) {
				auto &innoh = *(UVHandle::innerT*)&inno;
				innoh.closed = true;
				tcp.inner = {};
				return R_Err(err);
			}
			return R_Ok(tcp);
		}

		task<Result<UVTcp, int>> accept() {
			co_return (co_await UVStream::accept()).map_ok<UVTcp>([](auto val) -> UVTcp {
				return {val};
			});
		}
};

template <class Stream, class Handler>
task<void> serve_http(Stream server, Handler func) {
	struct hstate {
		llhttp_t parser;
		owned_coroutine coro;
	};
	llhttp_settings_t settings;
	llhttp_settings_init(&settings);
	while (1) {
		auto client_r = co_await server.accept();
		if (!client_r) {
			co_await delay(100);
			continue;
		}
		auto client = client_r.ok();
		([](auto client) -> task<void> {
			func(client);
		})(client);
	}
}

int UVTcp::stream_type_tag = 42;


void log_uverr(const char *fname, int code) {
	char errmsg[256];
	fprintf(stderr, "sigmawall: %s: %s\n", fname, uv_strerror_r(code, errmsg, sizeof(errmsg)));
}

template <class Ret, class Out = void>
bool err_unw(const char *fname, Result<Ret, int> res, Out *out) {
	if (!res) {
		log_uverr(fname, res.err());
		return false;
	}
	if constexpr (!std::is_void<Out>())
		*out = res.ok();
	return true;
}

task<void> mrrower(UVTcp client) {
	char buf[65536];
	size_t nread;
	while (err_unw("UVStream::read", co_await client.read(buf, sizeof(buf)), &nread)) {
		co_await client.write(buf, nread);
	}
	co_await client.shutdown();
}

task<void> meower(uv_loop_t *loop) {
	UVTcp server;
	if (!err_unw("UVTcp::bind", UVTcp::bind(loop, "127.0.0.1", 12423), &server))
		co_return;
	fprintf(stderr, "Server!!\n");
	if (!err_unw("UVTcp::listen", server.listen(128), (void*)NULL))
		co_return;
	while (1) {
		UVTcp client;
		if (!err_unw("UVTcp::accept", co_await server.accept(), &client))
			continue;
		mrrower(client);
		fprintf(stderr, "Client!!\n");
	}
}

int main() {
	UVLoop loop;
	auto sig = meower(&*loop);
	while (uv_loop_alive(&*loop))
		uv_run(&*loop, UV_RUN_DEFAULT);
}