#include #include #include #include #include #include #include #include #include #include #include using owned_coroutine = std::shared_ptr>; template class task; template class task_internal { public: class promise_type; using handle_type = std::coroutine_handle; 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::promise_type; bool returned = false; std::queue> resumables; std::weak_ptr handle; std::shared_ptr 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( 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 bool await_suspend(std::coroutine_handle &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 xhandle = handle.lock(); assert(xhandle); auto yhandle = std::reinterpret_pointer_cast>(xhandle); assert(yhandle); return yhandle; } }; bool await_ready() { return handle->h_typed.promise().returned; } template bool await_suspend(std::coroutine_handle &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; std::shared_ptr handle; task_internal(std::shared_ptr handle) : handle(handle) {} task_internal(const task_internal &src) : handle(src.handle) {} }; template class task : public task_internal { public: class promise_type; using handle_type = std::coroutine_handle; class promise_type : public task_internal::promise_type { private: friend class task; std::optional value; public: task get_return_object() { return { task_internal::promise_type::get_return_object() }; } void return_value(T uvalue) { value = uvalue; return task_internal::promise_type::return_any(); } }; T await_resume() { std::optional &val = handle_type::from_address(task_internal::handle->h_typed.address()).promise().value; assert(val.has_value()); return *val; } private: task(const task_internal &src) : task_internal(src) {} }; template <> class task : public task_internal { public: class promise_type : public task_internal::promise_type { public: task get_return_object() { return { task_internal::promise_type::get_return_object() }; } void return_void() { return task_internal::promise_type::return_any(); } }; void await_resume() { } private: task(const task_internal &src) : task_internal(src) {} }; template class awaiter { F fun; public: awaiter(F fun) : fun(fun) {} bool await_ready() { return false; } template bool await_suspend(std::coroutine_handle &handle) { auto xhandle = handle.promise().owned_handle(); return fun(xhandle); } void await_resume() {} }; template struct R_Ok { T v; R_Ok(T v) : v(v) {} }; template struct R_Err { T v; R_Err(T v) : v(v) {} }; template<> struct R_Ok {}; template<> struct R_Err {}; using R_VOk = R_Ok; using R_VErr = R_Err; template class Result { private: struct Ok_h { TOk v; }; struct Err_h { TErr v; }; std::variant val; Result() {} public: Result(R_Ok v) : val((Ok_h){v.v}) {} Result(R_Err 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(&val); } TOk &ok() { return std::get(val).v; } TErr &err() { return std::get(val).v; } template Result map_ok(F f) { if (*this) { return R_Ok(f(ok())); } return R_Err(err()); } }; template class Result { private: std::optional val; Result() {} public: Result(R_Ok v) { (void)v; } Result(R_Err v) : val(v.v) {} static Result Ok() { return { R_Ok() }; } static Result Err(TErr v) { return { R_Err(v) }; } operator bool() { return !val; } TErr &err() { return *val; } }; class comutex { private: std::queue coros; bool locked = false; public: class lock_type { private: friend class comutex; std::optional 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() { 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 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 inner; static task 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 static task destroy(T *ptr) { std::unique_ptr holder { ptr }; innerT &inno = *(innerT*)ptr; co_await close_inno(inno); } template innerT &init_inner() { auto inno = new innerT; std::shared_ptr inner_o = { inno, destroy }; inner = std::reinterpret_pointer_cast(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 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 reqs; bool listening = false; const int *stream_type_tag; std::queue 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 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> shutdown() { innerT &inno = *(innerT*)&*inner; poreque req(inno); int ret = uv_shutdown((uv_shutdown_t*)&req, (uv_stream_t*)&inno, on_req); 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 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> 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> 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> 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); 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(); 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 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> accept() { co_return (co_await UVStream::accept()).map_ok([](auto val) -> UVTcp { return {val}; }); } }; template task 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 { 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 bool err_unw(const char *fname, Result res, Out *out) { if (!res) { log_uverr(fname, res.err()); return false; } if constexpr (!std::is_void()) *out = res.ok(); return true; } task 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 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); }