// Copyright Pan Yue 2021.

// Distributed under the Boost Software License, Version 1.0. (See

// accompanying file LICENSE_1_0.txt or copy at

// http://www.boost.org/LICENSE_1_0.txt)

// TODO:

// 1. posix::stream_descriptor need windows version

// 2. call_* need return async.Handle

// 3. _ensure_fd_no_transport

// 4. _ensure_resolve

#include <errno.h>

#include <boost/asio.hpp>

#include <boost/bind.hpp>

#include <boost/python.hpp>

#include <boost/python/eventloop.hpp>

#include <boost/mpl/vector.hpp>

#include <Python.h>

namespace boost { namespace python { namespace asio {

namespace

{

bool _hasattr(object o, const char* name)

{

return PyObject_HasAttrString(o.ptr(), name);

}

void raise_dup_error()

{

PyErr_SetString(PyExc_OSError, std::system_category().message(errno).c_str());

throw_error_already_set();

}

} // namespace

void event_loop::_sock_connect_cb(object pymod_socket, object fut, object sock, object addr)

{

try

{

object err = sock.attr("getsockopt")(

pymod_socket.attr("SOL_SOCKET"), pymod_socket.attr("SO_ERROR"));

if (err != object(0)) {

// TODO: print the address

PyErr_SetString(PyExc_OSError, "Connect call failed {address}");

throw_error_already_set();

}

fut.attr("set_result")(object());

}

catch (const error_already_set& e)

{

if (PyErr_ExceptionMatches(PyExc_BlockingIOError)

|| PyErr_ExceptionMatches(PyExc_InterruptedError))

{

PyErr_Clear();

// pass

}

else if (PyErr_ExceptionMatches(PyExc_SystemExit)

|| PyErr_ExceptionMatches(PyExc_KeyboardInterrupt))

{

// raise

}

else

{

PyErr_Clear();

fut.attr("set_exception")(std::current_exception());

}

}

}

void event_loop::_sock_accept(event_loop& loop, object fut, object sock)

{

int fd = extract<int>(sock.attr("fileno")());

object conn, address;

try

{

object ret = sock.attr("accept")();

conn = ret[0];

address = ret[1];

conn.attr("setblocking")(object(false));

fut.attr("set_result")(make_tuple(conn, address));

}

catch (const error_already_set& e)

{

if (PyErr_ExceptionMatches(PyExc_BlockingIOError)

|| PyErr_ExceptionMatches(PyExc_InterruptedError))

{

PyErr_Clear();

loop._async_wait_fd(fd, bind(_sock_accept, boost::ref(loop), fut, sock), loop._write_key(fd));

}

else if (PyErr_ExceptionMatches(PyExc_SystemExit)

|| PyErr_ExceptionMatches(PyExc_KeyboardInterrupt))

{

// raise

}

else

{

PyErr_Clear();

fut.attr("set_exception")(std::current_exception());

}

}

}

void event_loop::call_later(double delay, object f)

{

auto p_timer = std::make_shared<boost::asio::steady_timer>(

_strand.context(),

std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::duration<double>(delay)));

p_timer->async_wait(boost::asio::bind_executor(_strand,

[f, p_timer] (const boost::system::error_code& ec) {

PyEval_AcquireLock();

f();

PyEval_ReleaseLock();

}));

}

void event_loop::call_at(double when, object f)

{

using sc = std::chrono::steady_clock;

auto p_timer = std::make_shared<boost::asio::steady_timer>(

_strand.context(),

sc::duration(static_cast<sc::time_point::rep>(when)));

p_timer->async_wait(boost::asio::bind_executor(_strand,

[f, p_timer] (const boost::system::error_code& ec) {f();}));

}

// TODO: support windows socket

object event_loop::sock_recv(object sock, size_t nbytes)

{

int fd = extract<int>(sock.attr("fileno")());

int fd_dup = dup(fd);

if (fd_dup == -1)

raise_dup_error();

object py_fut = create_future();

_async_wait_fd(fd_dup,

[py_fut, nbytes, fd=fd_dup] {

PyEval_AcquireLock();

std::vector<char> buffer(nbytes);

read(fd, buffer.data(), nbytes);

py_fut.attr("set_result")(object(handle<>(PyBytes_FromStringAndSize(buffer.data(), nbytes))));

PyEval_ReleaseLock();

},

_read_key(fd));

return py_fut;

}

// TODO: support windows socket

object event_loop::sock_recv_into(object sock, object buffer)

{

int fd = extract<int>(sock.attr("fileno")());

int fd_dup = dup(fd);

if (fd_dup == -1)

raise_dup_error();

ssize_t nbytes = len(buffer);

object py_fut = create_future();

_async_wait_fd(fd_dup,

[py_fut, nbytes, fd=fd_dup] {

PyEval_AcquireLock();

std::vector<char> buffer(nbytes);

ssize_t nbytes_read = read(fd, buffer.data(), nbytes);

py_fut.attr("set_result")(nbytes_read);

PyEval_ReleaseLock();

},

_read_key(fd));

return py_fut;

}

// TODO: support windows socket

object event_loop::sock_sendall(object sock, object data)

{

int fd = extract<int>(sock.attr("fileno")());

int fd_dup = dup(fd);

if (fd_dup == -1)

raise_dup_error();

char const* py_str = extract<char const*>(data.attr("decode")());

ssize_t py_str_len = len(data);

object py_fut = create_future();

_async_wait_fd(fd_dup,

[py_fut, fd, py_str, py_str_len] {

PyEval_AcquireLock();

write(fd, py_str, py_str_len);

py_fut.attr("set_result")(object());

PyEval_ReleaseLock();

},

_write_key(fd));

return py_fut;

}

// TODO: support windows socket

object event_loop::sock_connect(object sock, object address)

{

if (!_hasattr(_pymod_socket, "AF_UNIX") || sock.attr("family") != _pymod_socket.attr("AF_UNIX"))

{

// TODO: _ensure_resolve

}

object py_fut = create_future();

int fd = extract<int>(sock.attr("fileno")());

try

{

sock.attr("connect")(address);

py_fut.attr("set_result")(object());

}

catch (const error_already_set& e)

{

if (PyErr_ExceptionMatches(PyExc_BlockingIOError)

|| PyErr_ExceptionMatches(PyExc_InterruptedError))

{

PyErr_Clear();

int fd_dup = dup(fd);

if (fd_dup == -1)

raise_dup_error();

_async_wait_fd(fd_dup, bind(_sock_connect_cb, _pymod_socket, py_fut, sock, address), _write_key(fd));

}

else if (PyErr_ExceptionMatches(PyExc_SystemExit)

|| PyErr_ExceptionMatches(PyExc_KeyboardInterrupt))

{

// raise

}

else

{

PyErr_Clear();

py_fut.attr("set_exception")(std::current_exception());

}

}

return py_fut;

}

object event_loop::sock_accept(object sock)

{

object py_fut = create_future();

_sock_accept(*this, py_fut, sock);

return py_fut;

}

// TODO: implement this

object event_loop::sock_sendfile(object sock, object file, int offset, int count, bool fallback)

{

PyErr_SetString(PyExc_NotImplementedError, "Not implemented!");

throw_error_already_set();

return object();

}

// TODO: implement this

object event_loop::start_tls(object transport, object protocol, object sslcontext,

bool server_side, object server_hostname, object ssl_handshake_timeout)

{

PyErr_SetString(PyExc_NotImplementedError, "Not implemented!");

throw_error_already_set();

return object();

}

object event_loop::getaddrinfo(object host, int port, int family, int type, int proto, int flags)

{

object py_fut = create_future();

_strand.post(

[this, py_fut, host, port, family, type, proto, flags] {

PyEval_AcquireLock();

object res = _pymod_socket.attr("getaddrinfo")(host, port, family, type, proto, flags);

py_fut.attr("set_result")(res);

PyEval_ReleaseLock();

});

return py_fut;

}

object event_loop::getnameinfo(object sockaddr, int flags)

{

object py_fut = create_future();

_strand.post(

[this, py_fut, sockaddr, flags] {

PyEval_AcquireLock();

object res = _pymod_socket.attr("getnameinfo")(sockaddr, flags);

py_fut.attr("set_result")(res);

PyEval_ReleaseLock();

});

return py_fut;

}

void event_loop::default_exception_handler(object context)

{

object message = context.attr("get")(str("message"));

if (message == object())

{

message = str("Unhandled exception in event loop");

}

object exception = context.attr("get")(str("exception"));

object exc_info;

if (exception != object())

{

exc_info = make_tuple(exception.attr("__class__"), exception, exception.attr("__traceback__"));

}

else

{

exc_info = object(false);

}

if (!PyObject_IsTrue(context.attr("__contains__")(str("source_traceback")).ptr()) &&

_exception_handler != object() &&

_exception_handler.attr("_source_traceback") != object())

{

context["handle_traceback"] = _exception_handler.attr("_source_traceback");

}

list log_lines;

log_lines.append(message);

list context_keys(context.attr("keys"));

context_keys.sort();

for (int i = 0; i < len(context_keys); i++)

{

std::string key = extract<std::string>(context_keys[i]);

if (key == "message" || key == "exception")

continue;

str value(context[key]);

if (key == "source_traceback")

{

str tb = str("").join(_pymod_traceback.attr("format_list")(value));

value = str("Object created at (most recent call last):\n");

value += tb.rstrip();

}

else if (key == "handle_traceback")

{

str tb = str("").join(_pymod_traceback.attr("format_list")(value));

value = str("Handle created at (most recent call last):\n");

value += tb.rstrip();

}

else

{

value = str(value.attr("__str__")());

}

std::ostringstream stringStream;

stringStream << key << ": " << value;

log_lines.append(str(stringStream.str()));

}

list args;

dict kwargs;

args.append(str("\n").join(log_lines));

kwargs["exc_info"] = exc_info;

_py_logger.attr("error")(tuple(args), **kwargs);

}

void event_loop::call_exception_handler(object context)

{

if (_exception_handler == object())

{

try

{

default_exception_handler(context);

}

catch (const error_already_set& e)

{

if (PyErr_ExceptionMatches(PyExc_SystemExit)

|| PyErr_ExceptionMatches(PyExc_KeyboardInterrupt))

{

// raise

}

else

{

PyErr_Clear();

list args;

dict kwargs;

args.append(str("Exception in default exception handler"));

kwargs["exc_info"] = true;

_py_logger.attr("error")(tuple(args), **kwargs);

}

}

}

else

{

try

{

_exception_handler(context);

}

catch (const error_already_set& e)

{

if (PyErr_ExceptionMatches(PyExc_SystemExit)

|| PyErr_ExceptionMatches(PyExc_KeyboardInterrupt))

{

// raise

}

else

{

PyObject *ptype, *pvalue, *ptraceback;

PyErr_Fetch(&ptype, &pvalue, &ptraceback);

PyErr_NormalizeException(&ptype, &pvalue, &ptraceback);

object type{handle<>(ptype)};

object value{handle<>(pvalue)};

object traceback{handle<>(ptraceback)};

try

{

dict tmp_dict;

tmp_dict["message"] = str("Unhandled error in exception handler");

tmp_dict["exception"] = value;

tmp_dict["context"] = context;

default_exception_handler(tmp_dict);

}

catch (const error_already_set& e)

{

if (PyErr_ExceptionMatches(PyExc_SystemExit)

|| PyErr_ExceptionMatches(PyExc_KeyboardInterrupt))

{

// raise

}

else

{

boost::python::list args;

boost::python::dict kwargs;

args.append(str("Exception in default exception handler"));

kwargs["exc_info"] = true;

_py_logger.attr("error")(tuple(args), **kwargs);

}

}

}

}

}

}

object event_loop::create_future()

{

boost::python::dict kwargs;

kwargs["loop"] = boost::ref(*this);

return _pymod_asyncio_futures.attr("Future")(*boost::python::tuple(), **kwargs);

}

void set_default_event_loop(const boost::asio::io_context::strand& strand)

{

class_<event_loop, boost::noncopyable>("BoostAsioEventLoop", init<boost::asio::io_context::strand&>())

.def("call_soon", &event_loop::call_soon)

.def("call_soon_thread_safe", &event_loop::call_soon_thread_safe)

.def("call_later", &event_loop::call_later)

.def("call_at", &event_loop::call_at)

.def("time", &event_loop::time)

.def("add_reader", &event_loop::add_reader)

.def("remove_reader", &event_loop::remove_reader)

.def("add_writer", &event_loop::add_writer)

.def("remove_writer", &event_loop::remove_writer)

.def("sock_recv", &event_loop::sock_recv)

.def("sock_recv_into", &event_loop::sock_recv_into)

.def("sock_sendall", &event_loop::sock_sendall)

.def("sock_connect", &event_loop::sock_connect)

.def("sock_accept", &event_loop::sock_accept)

.def("sock_sendfile", &event_loop::sock_sendfile)

.def("start_tls", &event_loop::start_tls)

.def("getaddrinfo", &event_loop::getaddrinfo)

.def("getnameinfo", &event_loop::getnameinfo)

.def("set_exception_handler", &event_loop::set_exception_handler)

.def("get_exception_handler", &event_loop::get_exception_handler)

.def("default_exception_handler", &event_loop::default_exception_handler)

.def("call_exception_handler", &event_loop::call_exception_handler)

.def("create_future", &event_loop::create_future)

.def("get_debug", &event_loop::get_debug);

object asyncio = import("asyncio");

object abstract_policy = asyncio.attr("AbstractEventLoopPolicy");

dict method_dict;

std::shared_ptr<event_loop> p_loop = std::make_shared<event_loop>(strand);

method_dict["get_event_loop"] = make_function(

[p_loop] (object e) {return object(boost::ref(*p_loop));},

default_call_policies(),

boost::mpl::vector<object, object>()

);

object class_boost_policy = call<object>(

(PyObject*)&PyType_Type,

str("BoostEventLoopPolicy"),

boost::python::make_tuple(abstract_policy),

method_dict);

object boost_policy_instance = class_boost_policy.attr("__call__")();

asyncio.attr("set_event_loop_policy")(boost_policy_instance);

}

}}} // namespace boost::python::asio