diff --git a/ListTalk/networking/Socket.h b/ListTalk/networking/Socket.h new file mode 100644 index 0000000..d20b0fa --- /dev/null +++ b/ListTalk/networking/Socket.h @@ -0,0 +1,78 @@ +/* SPDX-License-Identifier: MIT */ +#ifndef H__ListTalk__networking__Socket__ +#define H__ListTalk__networking__Socket__ + +#include +#include +#include + +LT__BEGIN_DECLS + +LT_DECLARE_CLASS(LT_Socket); +LT_DECLARE_CLASS(LT_DatagramSocket); +LT_DECLARE_CLASS(LT_StreamSocket); +LT_DECLARE_CLASS(LT_ServerSocket); +LT_DECLARE_CLASS(LT_IPSocket); +LT_DECLARE_CLASS(LT_UDPSocket); +LT_DECLARE_CLASS(LT_TCPSocket); +LT_DECLARE_CLASS(LT_TCPServerSocket); +LT_DECLARE_CLASS(LT_UnixSocket); +LT_DECLARE_CLASS(LT_UnixDatagramSocket); +LT_DECLARE_CLASS(LT_UnixStreamSocket); +LT_DECLARE_CLASS(LT_UnixServerSocket); + +int LT_IPSocket_closed(LT_IPSocket* socket); +int LT_IPSocket_descriptor(LT_IPSocket* socket); +void LT_IPSocket_close(LT_IPSocket* socket); + +LT_UDPSocket* LT_UDPSocket_bind(const char* host, uint16_t port); +LT_UDPSocket* LT_UDPSocket_connect(const char* host, uint16_t port); +size_t LT_UDPSocket_send(LT_UDPSocket* socket, LT_ByteVector* bytes); +LT_ByteVector* LT_UDPSocket_receive(LT_UDPSocket* socket, + size_t maximum_length); + +LT_TCPSocket* LT_TCPSocket_connect(const char* host, uint16_t port); +size_t LT_TCPSocket_read(LT_TCPSocket* socket, + void* buffer, size_t length); +LT_Value LT_TCPSocket_readLine(LT_TCPSocket* socket); +void LT_TCPSocket_write(LT_TCPSocket* socket, + const void* buffer, size_t length); +void LT_TCPSocket_shutdown_write(LT_TCPSocket* socket); + +LT_TCPServerSocket* LT_TCPServerSocket_new(const char* host, + uint16_t port, + int backlog); +LT_TCPSocket* LT_TCPServerSocket_accept(LT_TCPServerSocket* socket); + +int LT_UnixSocket_closed(LT_UnixSocket* socket); +int LT_UnixSocket_descriptor(LT_UnixSocket* socket); +void LT_UnixSocket_close(LT_UnixSocket* socket); + +LT_UnixDatagramSocket* LT_UnixDatagramSocket_bind(const char* path); +LT_UnixDatagramSocket* LT_UnixDatagramSocket_connect(const char* path); +size_t LT_UnixDatagramSocket_send(LT_UnixDatagramSocket* socket, + LT_ByteVector* bytes); +LT_ByteVector* LT_UnixDatagramSocket_receive( + LT_UnixDatagramSocket* socket, + size_t maximum_length +); + +LT_UnixStreamSocket* LT_UnixStreamSocket_connect(const char* path); +void LT_UnixStreamSocket_pair(LT_UnixStreamSocket** first, + LT_UnixStreamSocket** second); +size_t LT_UnixStreamSocket_read(LT_UnixStreamSocket* socket, + void* buffer, + size_t length); +LT_Value LT_UnixStreamSocket_readLine(LT_UnixStreamSocket* socket); +void LT_UnixStreamSocket_write(LT_UnixStreamSocket* socket, + const void* buffer, + size_t length); +void LT_UnixStreamSocket_shutdown_write(LT_UnixStreamSocket* socket); + +LT_UnixServerSocket* LT_UnixServerSocket_new(const char* path, int backlog); +LT_UnixStreamSocket* LT_UnixServerSocket_accept( + LT_UnixServerSocket* socket +); + +LT__END_DECLS +#endif diff --git a/ListTalk/vm/eval.h b/ListTalk/vm/eval.h index 7ad6ee2..345184a 100644 --- a/ListTalk/vm/eval.h +++ b/ListTalk/vm/eval.h @@ -15,6 +15,9 @@ LT__BEGIN_DECLS typedef struct LT_TailCallUnwindMarker_s LT_TailCallUnwindMarker; +/* Run and clear the current thread's pending ListTalk signal, if any. */ +extern void LT_check_pending_signal(void); + extern LT_Value LT_eval( LT_Value expression, LT_Environment* environment, @@ -93,4 +96,4 @@ extern LT_Value LT_super_send( LT__END_DECLS -#endif \ No newline at end of file +#endif diff --git a/meson.build b/meson.build index aa02ed2..5c419a1 100644 --- a/meson.build +++ b/meson.build @@ -262,6 +262,28 @@ debugger_lib = library( install : true ) +networking_lib = library( + 'ListTalkNetworking', + [ + 'src/networking/Socket.c', + 'src/networking/IPSocket.c', + 'src/networking/SocketBuffer.c', + 'src/networking/UnixSocket.c', + ], + link_with : [vm_lib], + dependencies : listtalk_deps, + install : true +) + +pkg.generate( + networking_lib, + filebase : 'ListTalkNetworking', + name : 'ListTalkNetworking', + description : 'ListTalk high-level socket networking library', + subdirs : '.', + requires : ['ListTalkVM'] +) + pkg.generate( debugger_lib, filebase : 'ListTalkDebugger', @@ -326,6 +348,17 @@ ini_module = shared_module( install_dir : listtalk_native_module_dir ) +socket_module = shared_module( + 'socket', + 'src/modules/socket.c', + name_prefix : '', + name_suffix : 'ltm', + link_with : [vm_lib, networking_lib], + dependencies : listtalk_deps, + install : true, + install_dir : listtalk_native_module_dir +) + if md4c_dep.found() and md4c_html_dep.found() markdown_config = configuration_data() markdown_config.set10( @@ -517,6 +550,13 @@ message_queue_test_exe = executable( dependencies : [bdwgc] ) +networking_test_exe = executable( + 'networking_test', + 'tests/networking_test.c', + link_with : [vm_lib, networking_lib], + dependencies : [bdwgc] +) + test( 'reader_test', reader_test_exe @@ -545,6 +585,16 @@ test( depends : [os_module, gc_module, cmdopts_module, json_module, ini_module] ) +test( + 'eval_socket', + listtalk_exe, + args : [ + '-L', meson.current_build_dir(), + meson.current_source_dir() / 'tests' / 'eval-socket.lt' + ], + depends : [socket_module] +) + if md4c_dep.found() and md4c_html_dep.found() test( 'eval_markdown', @@ -683,6 +733,11 @@ test( message_queue_test_exe ) +test( + 'networking_test', + networking_test_exe +) + install_subdir( 'ListTalk', install_dir: get_option('includedir') diff --git a/src/modules/socket.c b/src/modules/socket.c new file mode 100644 index 0000000..62e7e65 --- /dev/null +++ b/src/modules/socket.c @@ -0,0 +1,22 @@ +/* SPDX-License-Identifier: MIT */ +#include +#include + +void ListTalk_socket_load(LT_Environment* environment){ + LT_Package* package = LT_Package_new("ListTalk:Socket"); +#define BIND_CLASS(name) LT_Environment_bind(environment, LT_Symbol_new_in(package, #name), LT_STATIC_CLASS(LT_##name), LT_ENV_BINDING_FLAG_CONSTANT) + BIND_CLASS(Socket); + BIND_CLASS(DatagramSocket); + BIND_CLASS(StreamSocket); + BIND_CLASS(ServerSocket); + BIND_CLASS(IPSocket); + BIND_CLASS(UDPSocket); + BIND_CLASS(TCPSocket); + BIND_CLASS(TCPServerSocket); + BIND_CLASS(UnixSocket); + BIND_CLASS(UnixDatagramSocket); + BIND_CLASS(UnixStreamSocket); + BIND_CLASS(UnixServerSocket); +#undef BIND_CLASS + LT_loader_provide(environment, "socket"); +} diff --git a/src/networking/IPSocket.c b/src/networking/IPSocket.c new file mode 100644 index 0000000..af92a76 --- /dev/null +++ b/src/networking/IPSocket.c @@ -0,0 +1,830 @@ +/* SPDX-License-Identifier: MIT */ +#include +#include +#include +#include +#include +#include +#include +#include + +#include "Socket_internal.h" + +#include +#include +#include +#include +#include +#include +#include + +struct LT_IPSocket_s { + LT_Object base; + int fd; +}; + +struct LT_UDPSocket_s { + LT_IPSocket base; +}; + +struct LT_TCPSocket_s { + LT_IPSocket base; + LT_SocketReadBuffer read_buffer; +}; + +struct LT_TCPServerSocket_s { + LT_IPSocket base; +}; + +static int socket_fd(LT_IPSocket* socket){ + if (socket->fd < 0){ + LT_error("Socket is closed"); + } + return socket->fd; +} + +static LT_IPSocket* socket_from_value(LT_Value value){ + if (LT_Value_class(value) != <_IPSocket_class + && !LT_Value_is_instance_of(value, LT_STATIC_CLASS(LT_IPSocket))) + LT_type_error(value, <_IPSocket_class); + return (LT_IPSocket*)LT_VALUE_POINTER_VALUE(value); +} + +int LT_IPSocket_closed(LT_IPSocket* socket){ + return socket->fd < 0; +} + +int LT_IPSocket_descriptor(LT_IPSocket* socket){ + return socket_fd(socket); +} + +void LT_IPSocket_close(LT_IPSocket* socket){ + int fd; + if (socket->fd < 0){ + return; + } + fd = socket->fd; + socket->fd = -1; + if (close(fd) != 0 && errno != EINTR){ + LT_system_error("Socket close failed", errno); + } +} + +static void socket_finalizer(void* object, void* unused){ + LT_IPSocket* socket = object; + (void)unused; + if (socket->fd >= 0){ + close(socket->fd); + socket->fd = -1; + } +} + +static void set_fd(LT_IPSocket* socket, int fd){ + socket->fd = fd; + GC_register_finalizer(socket, socket_finalizer, NULL, NULL, NULL); +} + +static uint16_t port_value(LT_Value value){ + size_t port = LT_Number_nonnegative_size_from_integer( + value, + "Invalid socket port", + "Invalid socket port" + ); + + if (port > 65535){ + LT_error("Socket port must be an integer from 0 through 65535"); + } + return (uint16_t)port; +} + +static int resolve_socket(const char* host, + uint16_t port, + int type, + int passive, + int do_listen, + int backlog){ + struct addrinfo hints = {0}; + struct addrinfo* addresses; + struct addrinfo* address; + char service[6]; + int fd = -1; + int result; + int saved_errno = 0; + int one = 1; + int zero = 0; + const char* resolved_host = host; + + if (passive && host != NULL && strcmp(host, "*") == 0){ + resolved_host = NULL; + } + + snprintf(service, sizeof(service), "%u", (unsigned)port); + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = type; + hints.ai_flags = passive ? AI_PASSIVE : 0; + result = getaddrinfo( + resolved_host && *resolved_host ? resolved_host : NULL, + service, + &hints, + &addresses + ); + if (result != 0){ + LT_error(LT_sprintf( + "Could not resolve socket address: %s", + gai_strerror(result) + )); + } + for (address = addresses; address; address = address->ai_next){ + fd = socket( + address->ai_family, + address->ai_socktype, + address->ai_protocol + ); + if (fd < 0){ + saved_errno = errno; + continue; + } + if (do_listen + && setsockopt( + fd, + SOL_SOCKET, + SO_REUSEADDR, + &one, + sizeof(one) + ) != 0){ + saved_errno = errno; + close(fd); + fd = -1; + continue; + } + if (do_listen + && address->ai_family == AF_INET6 + && setsockopt( + fd, + IPPROTO_IPV6, + IPV6_V6ONLY, + &zero, + sizeof(zero) + ) != 0){ + saved_errno = errno; + close(fd); + fd = -1; + continue; + } + int ready = 0; + + while (1){ + result = passive + ? bind(fd, address->ai_addr, address->ai_addrlen) + : connect(fd, address->ai_addr, address->ai_addrlen); + if (result == 0){ + ready = 1; + break; + } + if (errno == EINTR){ + LT_socket_interrupted(); + continue; + } + break; + } + if (ready && do_listen){ + ready = 0; + while (1){ + if (listen(fd, backlog) == 0){ + ready = 1; + break; + } + if (errno == EINTR){ + LT_socket_interrupted(); + continue; + } + break; + } + } + if (ready){ + break; + } + saved_errno = errno; + close(fd); + fd = -1; + } + freeaddrinfo(addresses); + if (fd < 0){ + LT_system_error( + passive ? "Could not bind socket" : "Could not connect socket", + saved_errno + ); + } + return fd; +} + +LT_UDPSocket* LT_UDPSocket_bind(const char* host, uint16_t port){ + LT_UDPSocket* result = LT_Class_ALLOC(LT_UDPSocket); + set_fd( + (LT_IPSocket*)result, + resolve_socket(host, port, SOCK_DGRAM, 1, 0, 0) + ); + return result; +} + +LT_UDPSocket* LT_UDPSocket_connect(const char* host, uint16_t port){ + LT_UDPSocket* result = LT_Class_ALLOC(LT_UDPSocket); + set_fd( + (LT_IPSocket*)result, + resolve_socket(host, port, SOCK_DGRAM, 0, 0, 0) + ); + return result; +} +size_t LT_UDPSocket_send(LT_UDPSocket* socket, LT_ByteVector* bytes){ + ssize_t n; + + while (1){ + n = send( + socket_fd((LT_IPSocket*)socket), + LT_ByteVector_bytes(bytes), + LT_ByteVector_length(bytes), + 0 + ); + if (n >= 0 || errno != EINTR){ + break; + } + LT_socket_interrupted(); + } + if (n < 0){ + LT_system_error("Datagram send failed", errno); + } + if ((size_t)n != LT_ByteVector_length(bytes)){ + LT_error("Datagram send was incomplete"); + } + return (size_t)n; +} +LT_ByteVector* LT_UDPSocket_receive(LT_UDPSocket* socket, + size_t maximum_length){ + uint8_t* bytes = GC_MALLOC_ATOMIC(maximum_length ? maximum_length : 1); + ssize_t n; + while (1){ + n = recv( + socket_fd((LT_IPSocket*)socket), + bytes, + maximum_length, + 0 + ); + if (n >= 0 || errno != EINTR){ + break; + } + LT_socket_interrupted(); + } + if (n < 0){ + LT_system_error("Datagram receive failed", errno); + } + return LT_ByteVector_new(bytes, (size_t)n); +} + +LT_TCPSocket* LT_TCPSocket_connect(const char* host, uint16_t port){ + LT_TCPSocket* result = LT_Class_ALLOC(LT_TCPSocket); + set_fd( + (LT_IPSocket*)result, + resolve_socket(host, port, SOCK_STREAM, 0, 0, 0) + ); + LT_socket_read_buffer_init(&result->read_buffer); + return result; +} +static LT_TCPSocket* stream_from_fd(int fd){ + LT_TCPSocket* result = LT_Class_ALLOC(LT_TCPSocket); + + set_fd((LT_IPSocket*)result, fd); + LT_socket_read_buffer_init(&result->read_buffer); + return result; +} + +size_t LT_TCPSocket_read(LT_TCPSocket* socket, + void* buffer, + size_t length){ + return LT_socket_buffered_read( + socket_fd((LT_IPSocket*)socket), + &socket->read_buffer, + buffer, + length, + "Socket read failed" + ); +} + +LT_Value LT_TCPSocket_readLine(LT_TCPSocket* socket){ + return LT_socket_buffered_read_line( + socket_fd((LT_IPSocket*)socket), + &socket->read_buffer, + "Socket read failed" + ); +} +void LT_TCPSocket_write(LT_TCPSocket* socket, + const void* buffer, + size_t length){ + const uint8_t* bytes = buffer; + size_t offset = 0; + + while (offset < length){ +#ifdef MSG_NOSIGNAL + ssize_t n = send( + socket_fd((LT_IPSocket*)socket), + bytes + offset, + length - offset, + MSG_NOSIGNAL + ); +#else + ssize_t n = send( + socket_fd((LT_IPSocket*)socket), + bytes + offset, + length - offset, + 0 + ); +#endif + if (n < 0 && errno == EINTR){ + LT_socket_interrupted(); + continue; + } + if (n <= 0){ + LT_system_error( + "Socket write failed", + n < 0 ? errno : EPIPE + ); + } + offset += (size_t)n; + } +} +void LT_TCPSocket_shutdown_write(LT_TCPSocket* socket){ + if (shutdown(socket_fd((LT_IPSocket*)socket), SHUT_WR) != 0){ + LT_system_error("Socket shutdown failed", errno); + } +} + +LT_TCPServerSocket* LT_TCPServerSocket_new(const char* host, + uint16_t port, + int backlog){ + LT_TCPServerSocket* result = LT_Class_ALLOC(LT_TCPServerSocket); + if (backlog < 1){ + LT_error("Socket backlog must be positive"); + } + set_fd( + (LT_IPSocket*)result, + resolve_socket(host, port, SOCK_STREAM, 1, 1, backlog) + ); + return result; +} +LT_TCPSocket* LT_TCPServerSocket_accept(LT_TCPServerSocket* socket){ + int fd; + + while (1){ + fd = accept(socket_fd((LT_IPSocket*)socket), NULL, NULL); + if (fd >= 0 || errno != EINTR){ + break; + } + LT_socket_interrupted(); + } + if (fd < 0){ + LT_system_error("Socket accept failed", errno); + } + return stream_from_fd(fd); +} + +static LT_String* string_arg(LT_Value* cursor){ + LT_String* value; + + LT_GENERIC_ARG(*cursor, value, LT_String*, LT_String_from_value); + return value; +} + +LT_DEFINE_PRIMITIVE( + socket_method_closed, + "IPSocket>>isClosed", + "(self)", + "Return true when the socket is closed." +){ + LT_Value cursor = arguments; + LT_IPSocket* self; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG(cursor, self, LT_IPSocket*, socket_from_value); + LT_ARG_END(cursor); + return LT_IPSocket_closed(self) ? LT_TRUE : LT_FALSE; +} + +LT_DEFINE_PRIMITIVE( + socket_method_close, + "IPSocket>>close", + "(self)", + "Close the socket." +){ + LT_Value cursor = arguments; + LT_IPSocket* self; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG(cursor, self, LT_IPSocket*, socket_from_value); + LT_ARG_END(cursor); + LT_IPSocket_close(self); + return (LT_Value)(uintptr_t)self; +} + +LT_DEFINE_PRIMITIVE( + datagram_bind, + "UDPSocket class>>bindTo:port:", + "(self host port)", + "Bind a datagram socket." +){ + LT_Value cursor = arguments; + LT_Value self; + LT_Value port; + LT_String* host; + + (void)tail_call_unwind_marker; + LT_OBJECT_ARG(cursor, self); + host = string_arg(&cursor); + LT_OBJECT_ARG(cursor, port); + LT_ARG_END(cursor); + (void)self; + return (LT_Value)(uintptr_t)LT_UDPSocket_bind( + LT_String_value_cstr(host), + port_value(port) + ); +} + +LT_DEFINE_PRIMITIVE( + datagram_connect, + "UDPSocket class>>connectTo:port:", + "(self host port)", + "Connect a datagram socket." +){ + LT_Value cursor = arguments; + LT_Value self; + LT_Value port; + LT_String* host; + + (void)tail_call_unwind_marker; + LT_OBJECT_ARG(cursor, self); + host = string_arg(&cursor); + LT_OBJECT_ARG(cursor, port); + LT_ARG_END(cursor); + (void)self; + return (LT_Value)(uintptr_t)LT_UDPSocket_connect( + LT_String_value_cstr(host), + port_value(port) + ); +} + +LT_DEFINE_PRIMITIVE( + datagram_send, + "UDPSocket>>send:", + "(self bytes)", + "Send one datagram." +){ + LT_Value cursor = arguments; + LT_UDPSocket* self; + LT_ByteVector* bytes; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG( + cursor, + self, + LT_UDPSocket*, + LT_UDPSocket_from_value + ); + LT_GENERIC_ARG(cursor, bytes, LT_ByteVector*, LT_ByteVector_from_value); + LT_ARG_END(cursor); + return LT_Number_smallinteger_from_size( + LT_UDPSocket_send(self, bytes), + "Datagram too large" + ); +} + +LT_DEFINE_PRIMITIVE( + datagram_receive, + "UDPSocket>>receive:", + "(self maximumLength)", + "Receive one datagram." +){ + LT_Value cursor = arguments; + LT_Value maximum_length; + LT_UDPSocket* self; + size_t length; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG( + cursor, + self, + LT_UDPSocket*, + LT_UDPSocket_from_value + ); + LT_OBJECT_ARG(cursor, maximum_length); + LT_ARG_END(cursor); + length = LT_Number_nonnegative_size_from_integer( + maximum_length, + "Invalid receive length", + "Invalid receive length" + ); + return (LT_Value)(uintptr_t)LT_UDPSocket_receive(self, length); +} + +LT_DEFINE_PRIMITIVE( + stream_connect, + "TCPSocket class>>connectTo:port:", + "(self host port)", + "Connect a stream socket." +){ + LT_Value cursor = arguments; + LT_Value self; + LT_Value port; + LT_String* host; + + (void)tail_call_unwind_marker; + LT_OBJECT_ARG(cursor, self); + host = string_arg(&cursor); + LT_OBJECT_ARG(cursor, port); + LT_ARG_END(cursor); + (void)self; + return (LT_Value)(uintptr_t)LT_TCPSocket_connect( + LT_String_value_cstr(host), + port_value(port) + ); +} + +LT_DEFINE_PRIMITIVE( + stream_read, + "TCPSocket>>read:", + "(self maximumLength)", + "Read exactly maximumLength bytes, or fewer at end of stream." +){ + LT_Value cursor = arguments; + LT_Value maximum_length; + LT_TCPSocket* self; + size_t length; + size_t count; + uint8_t* bytes; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG( + cursor, + self, + LT_TCPSocket*, + LT_TCPSocket_from_value + ); + LT_OBJECT_ARG(cursor, maximum_length); + LT_ARG_END(cursor); + length = LT_Number_nonnegative_size_from_integer( + maximum_length, + "Invalid read length", + "Invalid read length" + ); + bytes = GC_MALLOC_ATOMIC(length ? length : 1); + count = LT_TCPSocket_read(self, bytes, length); + return (LT_Value)(uintptr_t)LT_ByteVector_new(bytes, count); +} + +LT_DEFINE_PRIMITIVE( + stream_write, + "TCPSocket>>write:", + "(self bytes)", + "Write all bytes." +){ + LT_Value cursor = arguments; + LT_TCPSocket* self; + LT_ByteVector* bytes; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG( + cursor, + self, + LT_TCPSocket*, + LT_TCPSocket_from_value + ); + LT_GENERIC_ARG(cursor, bytes, LT_ByteVector*, LT_ByteVector_from_value); + LT_ARG_END(cursor); + LT_TCPSocket_write( + self, + LT_ByteVector_bytes(bytes), + LT_ByteVector_length(bytes) + ); + return (LT_Value)(uintptr_t)self; +} + +LT_DEFINE_PRIMITIVE( + stream_read_line, + "TCPSocket>>readLine", + "(self)", + "Read through a line feed and include it in the returned bytevector." +){ + LT_Value cursor = arguments; + LT_TCPSocket* self; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG( + cursor, + self, + LT_TCPSocket*, + LT_TCPSocket_from_value + ); + LT_ARG_END(cursor); + return LT_TCPSocket_readLine(self); +} + +LT_DEFINE_PRIMITIVE( + stream_shutdown, + "TCPSocket>>shutdownWrite", + "(self)", + "Shut down the writing half." +){ + LT_Value cursor = arguments; + LT_TCPSocket* self; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG( + cursor, + self, + LT_TCPSocket*, + LT_TCPSocket_from_value + ); + LT_ARG_END(cursor); + LT_TCPSocket_shutdown_write(self); + return (LT_Value)(uintptr_t)self; +} + +LT_DEFINE_PRIMITIVE( + server_new, + "TCPServerSocket class>>newOn:port:backlog:", + "(self host port backlog)", + "Create a listening socket." +){ + LT_Value cursor = arguments; + LT_Value self; + LT_Value port; + LT_Value backlog_value; + LT_String* host; + size_t backlog; + + (void)tail_call_unwind_marker; + LT_OBJECT_ARG(cursor, self); + host = string_arg(&cursor); + LT_OBJECT_ARG(cursor, port); + LT_OBJECT_ARG(cursor, backlog_value); + LT_ARG_END(cursor); + backlog = LT_Number_nonnegative_size_from_integer( + backlog_value, + "Invalid socket backlog", + "Invalid socket backlog" + ); + if (backlog < 1 || backlog > INT_MAX){ + LT_error("Invalid socket backlog"); + } + (void)self; + return (LT_Value)(uintptr_t)LT_TCPServerSocket_new( + LT_String_value_cstr(host), + port_value(port), + (int)backlog + ); +} + +LT_DEFINE_PRIMITIVE( + server_new_default_backlog, + "TCPServerSocket class>>newOn:port:", + "(self host port)", + "Create a listening socket using the system maximum backlog." +){ + LT_Value cursor = arguments; + LT_Value self; + LT_Value port; + LT_String* host; + + (void)tail_call_unwind_marker; + LT_OBJECT_ARG(cursor, self); + host = string_arg(&cursor); + LT_OBJECT_ARG(cursor, port); + LT_ARG_END(cursor); + (void)self; + return (LT_Value)(uintptr_t)LT_TCPServerSocket_new( + LT_String_value_cstr(host), + port_value(port), + SOMAXCONN + ); +} + +LT_DEFINE_PRIMITIVE( + server_accept, + "TCPServerSocket>>accept", + "(self)", + "Accept and return a stream socket." +){ + LT_Value cursor = arguments; + LT_TCPServerSocket* self; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG( + cursor, + self, + LT_TCPServerSocket*, + LT_TCPServerSocket_from_value + ); + LT_ARG_END(cursor); + return (LT_Value)(uintptr_t)LT_TCPServerSocket_accept(self); +} + +static LT_Method_Descriptor ip_socket_methods[] = { + {"isClosed", &socket_method_closed}, + {"close", &socket_method_close}, + LT_NULL_NATIVE_CLASS_METHOD_DESCRIPTOR +}; + +static LT_Method_Descriptor datagram_methods[] = { + {"send:", &datagram_send}, + {"receive:", &datagram_receive}, + LT_NULL_NATIVE_CLASS_METHOD_DESCRIPTOR +}; + +static LT_Method_Descriptor datagram_class_methods[] = { + {"bindTo:port:", &datagram_bind}, + {"connectTo:port:", &datagram_connect}, + LT_NULL_NATIVE_CLASS_METHOD_DESCRIPTOR +}; + +static LT_Method_Descriptor stream_methods[] = { + {"read:", &stream_read}, + {"readLine", &stream_read_line}, + {"write:", &stream_write}, + {"shutdownWrite", &stream_shutdown}, + LT_NULL_NATIVE_CLASS_METHOD_DESCRIPTOR +}; + +static LT_Method_Descriptor stream_class_methods[] = { + {"connectTo:port:", &stream_connect}, + LT_NULL_NATIVE_CLASS_METHOD_DESCRIPTOR +}; + +static LT_Method_Descriptor server_methods[] = { + {"accept", &server_accept}, + LT_NULL_NATIVE_CLASS_METHOD_DESCRIPTOR +}; + +static LT_Method_Descriptor server_class_methods[] = { + {"newOn:port:", &server_new_default_backlog}, + {"newOn:port:backlog:", &server_new}, + LT_NULL_NATIVE_CLASS_METHOD_DESCRIPTOR +}; + +static LT_Class* udp_socket_mixins[] = { + <_DatagramSocket_class, + NULL +}; + +static LT_Class* tcp_socket_mixins[] = { + <_StreamSocket_class, + NULL +}; + +static LT_Class* tcp_server_socket_mixins[] = { + <_ServerSocket_class, + NULL +}; + +LT_DEFINE_CLASS(LT_IPSocket) { + .superclass = <_Socket_class, + .metaclass_superclass = <_Class_class, + .package = "ListTalk:Socket", + .name = "IPSocket", + .documentation = "Abstract socket implemented using the IP protocol suite.", + .instance_size = sizeof(LT_IPSocket), + .class_flags = LT_CLASS_FLAG_ABSTRACT, + .methods = ip_socket_methods, +}; + +LT_DEFINE_CLASS(LT_UDPSocket) { + .superclass = <_IPSocket_class, + .mixins = udp_socket_mixins, + .metaclass_superclass = <_Class_class, + .package = "ListTalk:Socket", + .name = "UDPSocket", + .documentation = "UDP datagram socket.", + .instance_size = sizeof(LT_UDPSocket), + .methods = datagram_methods, + .class_methods = datagram_class_methods, +}; + +LT_DEFINE_CLASS(LT_TCPSocket) { + .superclass = <_IPSocket_class, + .mixins = tcp_socket_mixins, + .metaclass_superclass = <_Class_class, + .package = "ListTalk:Socket", + .name = "TCPSocket", + .documentation = "Connected TCP byte-stream socket.", + .instance_size = sizeof(LT_TCPSocket), + .methods = stream_methods, + .class_methods = stream_class_methods, +}; + +LT_DEFINE_CLASS(LT_TCPServerSocket) { + .superclass = <_IPSocket_class, + .mixins = tcp_server_socket_mixins, + .metaclass_superclass = <_Class_class, + .package = "ListTalk:Socket", + .name = "TCPServerSocket", + .documentation = "Listening TCP socket.", + .instance_size = sizeof(LT_TCPServerSocket), + .methods = server_methods, + .class_methods = server_class_methods, +}; diff --git a/src/networking/Socket.c b/src/networking/Socket.c new file mode 100644 index 0000000..1ab34a9 --- /dev/null +++ b/src/networking/Socket.c @@ -0,0 +1,44 @@ +/* SPDX-License-Identifier: MIT */ +#include +#include +#include + +LT_DEFINE_CLASS(LT_Socket) { + .superclass = <_Object_class, + .metaclass_superclass = <_Class_class, + .package = "ListTalk:Socket", + .name = "Socket", + .documentation = "Abstract socket protocol root.", + .instance_size = 0, + .class_flags = LT_CLASS_FLAG_ABSTRACT, +}; + +LT_DEFINE_CLASS(LT_DatagramSocket) { + .superclass = <_Socket_class, + .metaclass_superclass = <_Class_class, + .package = "ListTalk:Socket", + .name = "DatagramSocket", + .documentation = "Abstract message-oriented socket.", + .instance_size = 0, + .class_flags = LT_CLASS_FLAG_ABSTRACT, +}; + +LT_DEFINE_CLASS(LT_StreamSocket) { + .superclass = <_Socket_class, + .metaclass_superclass = <_Class_class, + .package = "ListTalk:Socket", + .name = "StreamSocket", + .documentation = "Abstract connected byte-stream socket.", + .instance_size = 0, + .class_flags = LT_CLASS_FLAG_ABSTRACT, +}; + +LT_DEFINE_CLASS(LT_ServerSocket) { + .superclass = <_Socket_class, + .metaclass_superclass = <_Class_class, + .package = "ListTalk:Socket", + .name = "ServerSocket", + .documentation = "Abstract socket that accepts connections.", + .instance_size = 0, + .class_flags = LT_CLASS_FLAG_ABSTRACT, +}; diff --git a/src/networking/SocketBuffer.c b/src/networking/SocketBuffer.c new file mode 100644 index 0000000..354257b --- /dev/null +++ b/src/networking/SocketBuffer.c @@ -0,0 +1,163 @@ +/* SPDX-License-Identifier: MIT */ +#include "Socket_internal.h" + +#include +#include +#include + +#include +#include +#include + +#define SOCKET_READ_BUFFER_SIZE 4096 + +void LT_socket_read_buffer_init(LT_SocketReadBuffer* buffer){ + buffer->bytes = NULL; + buffer->start = 0; + buffer->end = 0; + buffer->capacity = 0; +} + +static size_t buffered_length(LT_SocketReadBuffer* buffer){ + return buffer->end - buffer->start; +} + +static ssize_t socket_receive(int fd, + void* destination, + size_t length, + const char* error_message){ + ssize_t count; + + while (1){ + count = recv(fd, destination, length, 0); + if (count >= 0 || errno != EINTR){ + break; + } + LT_socket_interrupted(); + } + if (count < 0){ + LT_system_error(error_message, errno); + } + return count; +} + +void LT_socket_interrupted(void){ + LT_check_pending_signal(); +} + +static int fill_buffer(int fd, + LT_SocketReadBuffer* buffer, + const char* error_message){ + ssize_t count; + + if (buffer->capacity == 0){ + buffer->bytes = GC_MALLOC_ATOMIC(SOCKET_READ_BUFFER_SIZE); + buffer->capacity = SOCKET_READ_BUFFER_SIZE; + } + count = socket_receive( + fd, + buffer->bytes, + buffer->capacity, + error_message + ); + buffer->start = 0; + buffer->end = (size_t)count; + return count != 0; +} + +size_t LT_socket_buffered_read(int fd, + LT_SocketReadBuffer* buffer, + void* destination, + size_t length, + const char* error_message){ + uint8_t* output = destination; + size_t total = 0; + + while (total < length){ + size_t available = buffered_length(buffer); + + if (available > 0){ + size_t count = available < length - total + ? available + : length - total; + + memcpy(output + total, buffer->bytes + buffer->start, count); + buffer->start += count; + total += count; + } else { + ssize_t count = socket_receive( + fd, + output + total, + length - total, + error_message + ); + + if (count == 0){ + break; + } + total += (size_t)count; + } + } + return total; +} + +static void append_bytes(uint8_t** result, + size_t* length, + size_t* capacity, + const uint8_t* bytes, + size_t count){ + if (count > *capacity - *length){ + size_t new_capacity = *capacity ? *capacity : SOCKET_READ_BUFFER_SIZE; + uint8_t* new_result; + + while (count > new_capacity - *length){ + new_capacity *= 2; + } + new_result = GC_MALLOC_ATOMIC(new_capacity); + if (*length > 0){ + memcpy(new_result, *result, *length); + } + *result = new_result; + *capacity = new_capacity; + } + memcpy(*result + *length, bytes, count); + *length += count; +} + +LT_Value LT_socket_buffered_read_line(int fd, + LT_SocketReadBuffer* buffer, + const char* error_message){ + uint8_t* result = NULL; + size_t length = 0; + size_t capacity = 0; + + while (1){ + size_t available; + uint8_t* delimiter; + size_t count; + + if (buffered_length(buffer) == 0 + && !fill_buffer(fd, buffer, error_message)){ + if (length == 0){ + return LT_NIL; + } + return (LT_Value)(uintptr_t)LT_ByteVector_new(result, length); + } + available = buffered_length(buffer); + delimiter = memchr(buffer->bytes + buffer->start, '\n', available); + count = delimiter == NULL + ? available + : (size_t)(delimiter - (buffer->bytes + buffer->start)) + 1; + append_bytes( + &result, + &length, + &capacity, + buffer->bytes + buffer->start, + count + ); + buffer->start += count; + if (delimiter != NULL){ + return (LT_Value)(uintptr_t)LT_ByteVector_new(result, length); + } + } +} diff --git a/src/networking/Socket_internal.h b/src/networking/Socket_internal.h new file mode 100644 index 0000000..957ac63 --- /dev/null +++ b/src/networking/Socket_internal.h @@ -0,0 +1,25 @@ +/* SPDX-License-Identifier: MIT */ +#ifndef H__ListTalk__networking__Socket_internal__ +#define H__ListTalk__networking__Socket_internal__ + +#include + +typedef struct LT_SocketReadBuffer { + uint8_t* bytes; + size_t start; + size_t end; + size_t capacity; +} LT_SocketReadBuffer; + +void LT_socket_read_buffer_init(LT_SocketReadBuffer* buffer); +size_t LT_socket_buffered_read(int fd, + LT_SocketReadBuffer* buffer, + void* destination, + size_t length, + const char* error_message); +LT_Value LT_socket_buffered_read_line(int fd, + LT_SocketReadBuffer* buffer, + const char* error_message); +void LT_socket_interrupted(void); + +#endif diff --git a/src/networking/UnixSocket.c b/src/networking/UnixSocket.c new file mode 100644 index 0000000..79bc96c --- /dev/null +++ b/src/networking/UnixSocket.c @@ -0,0 +1,764 @@ +/* SPDX-License-Identifier: MIT */ +#include +#include +#include +#include +#include +#include +#include + +#include "Socket_internal.h" + +#include +#include +#include +#include +#include +#include +#include + +struct LT_UnixSocket_s { + LT_Object base; + int fd; +}; + +struct LT_UnixDatagramSocket_s { + LT_UnixSocket base; +}; + +struct LT_UnixStreamSocket_s { + LT_UnixSocket base; + LT_SocketReadBuffer read_buffer; +}; + +struct LT_UnixServerSocket_s { + LT_UnixSocket base; +}; + +static int unix_socket_fd(LT_UnixSocket* socket){ + if (socket->fd < 0){ + LT_error("Socket is closed"); + } + return socket->fd; +} + +static LT_UnixSocket* unix_socket_from_value(LT_Value value){ + if (LT_Value_class(value) != <_UnixSocket_class + && !LT_Value_is_instance_of( + value, + LT_STATIC_CLASS(LT_UnixSocket) + )){ + LT_type_error(value, <_UnixSocket_class); + } + return (LT_UnixSocket*)LT_VALUE_POINTER_VALUE(value); +} + +int LT_UnixSocket_closed(LT_UnixSocket* socket){ + return socket->fd < 0; +} + +int LT_UnixSocket_descriptor(LT_UnixSocket* socket){ + return unix_socket_fd(socket); +} + +void LT_UnixSocket_close(LT_UnixSocket* socket){ + int fd; + + if (socket->fd < 0){ + return; + } + fd = socket->fd; + socket->fd = -1; + if (close(fd) != 0 && errno != EINTR){ + LT_system_error("Socket close failed", errno); + } +} + +static void unix_socket_finalizer(void* object, void* unused){ + LT_UnixSocket* socket = object; + + (void)unused; + if (socket->fd >= 0){ + close(socket->fd); + socket->fd = -1; + } +} + +static void unix_socket_set_fd(LT_UnixSocket* socket, int fd){ + socket->fd = fd; + GC_register_finalizer(socket, unix_socket_finalizer, NULL, NULL, NULL); +} + +static socklen_t unix_address(const char* path, struct sockaddr_un* address){ + size_t length = strlen(path); + + if (length == 0){ + LT_error("Unix socket path must not be empty"); + } + if (length >= sizeof(address->sun_path)){ + LT_error("Unix socket path is too long"); + } + memset(address, 0, sizeof(*address)); + address->sun_family = AF_UNIX; + memcpy(address->sun_path, path, length + 1); + return (socklen_t)(offsetof(struct sockaddr_un, sun_path) + length + 1); +} + +static int unix_connect(const char* path, int type){ + struct sockaddr_un address; + socklen_t address_length = unix_address(path, &address); + int fd = socket(AF_UNIX, type, 0); + + if (fd < 0){ + LT_system_error("Could not create Unix socket", errno); + } + while (1){ + if (connect(fd, (struct sockaddr*)&address, address_length) == 0){ + break; + } + if (errno == EINTR){ + LT_socket_interrupted(); + continue; + } + if (errno == EISCONN){ + break; + } + int saved_errno = errno; + + close(fd); + LT_system_error("Could not connect Unix socket", saved_errno); + } + return fd; +} + +static int unix_bind(const char* path, int type, int backlog){ + struct sockaddr_un address; + socklen_t address_length = unix_address(path, &address); + int fd = socket(AF_UNIX, type, 0); + + if (fd < 0){ + LT_system_error("Could not create Unix socket", errno); + } + if (bind(fd, (struct sockaddr*)&address, address_length) != 0 + || (backlog > 0 && listen(fd, backlog) != 0)){ + int saved_errno = errno; + + close(fd); + LT_system_error("Could not bind Unix socket", saved_errno); + } + return fd; +} + +LT_UnixDatagramSocket* LT_UnixDatagramSocket_bind(const char* path){ + LT_UnixDatagramSocket* result = LT_Class_ALLOC(LT_UnixDatagramSocket); + + unix_socket_set_fd((LT_UnixSocket*)result, unix_bind(path, SOCK_DGRAM, 0)); + return result; +} + +LT_UnixDatagramSocket* LT_UnixDatagramSocket_connect(const char* path){ + LT_UnixDatagramSocket* result = LT_Class_ALLOC(LT_UnixDatagramSocket); + + unix_socket_set_fd( + (LT_UnixSocket*)result, + unix_connect(path, SOCK_DGRAM) + ); + return result; +} + +size_t LT_UnixDatagramSocket_send(LT_UnixDatagramSocket* socket, + LT_ByteVector* bytes){ + ssize_t count; + + while (1){ + count = send( + unix_socket_fd((LT_UnixSocket*)socket), + LT_ByteVector_bytes(bytes), + LT_ByteVector_length(bytes), + 0 + ); + if (count >= 0 || errno != EINTR){ + break; + } + LT_socket_interrupted(); + } + if (count < 0){ + LT_system_error("Unix datagram send failed", errno); + } + if ((size_t)count != LT_ByteVector_length(bytes)){ + LT_error("Unix datagram send was incomplete"); + } + return (size_t)count; +} + +LT_ByteVector* LT_UnixDatagramSocket_receive( + LT_UnixDatagramSocket* socket, + size_t maximum_length +){ + uint8_t* bytes = GC_MALLOC_ATOMIC( + maximum_length ? maximum_length : 1 + ); + ssize_t count; + + while (1){ + count = recv( + unix_socket_fd((LT_UnixSocket*)socket), + bytes, + maximum_length, + 0 + ); + if (count >= 0 || errno != EINTR){ + break; + } + LT_socket_interrupted(); + } + if (count < 0){ + LT_system_error("Unix datagram receive failed", errno); + } + return LT_ByteVector_new(bytes, (size_t)count); +} + +static LT_UnixStreamSocket* unix_stream_from_fd(int fd){ + LT_UnixStreamSocket* result = LT_Class_ALLOC(LT_UnixStreamSocket); + + unix_socket_set_fd((LT_UnixSocket*)result, fd); + LT_socket_read_buffer_init(&result->read_buffer); + return result; +} + +LT_UnixStreamSocket* LT_UnixStreamSocket_connect(const char* path){ + return unix_stream_from_fd(unix_connect(path, SOCK_STREAM)); +} + +void LT_UnixStreamSocket_pair(LT_UnixStreamSocket** first, + LT_UnixStreamSocket** second){ + int descriptors[2]; + + while (socketpair(AF_UNIX, SOCK_STREAM, 0, descriptors) != 0){ + if (errno != EINTR){ + LT_system_error("Could not create Unix socket pair", errno); + } + LT_socket_interrupted(); + } + *first = unix_stream_from_fd(descriptors[0]); + *second = unix_stream_from_fd(descriptors[1]); +} + +size_t LT_UnixStreamSocket_read(LT_UnixStreamSocket* socket, + void* buffer, + size_t length){ + return LT_socket_buffered_read( + unix_socket_fd((LT_UnixSocket*)socket), + &socket->read_buffer, + buffer, + length, + "Unix socket read failed" + ); +} + +LT_Value LT_UnixStreamSocket_readLine(LT_UnixStreamSocket* socket){ + return LT_socket_buffered_read_line( + unix_socket_fd((LT_UnixSocket*)socket), + &socket->read_buffer, + "Unix socket read failed" + ); +} + +void LT_UnixStreamSocket_write(LT_UnixStreamSocket* socket, + const void* buffer, + size_t length){ + const uint8_t* bytes = buffer; + size_t offset = 0; + + while (offset < length){ +#ifdef MSG_NOSIGNAL + ssize_t count = send( + unix_socket_fd((LT_UnixSocket*)socket), + bytes + offset, + length - offset, + MSG_NOSIGNAL + ); +#else + ssize_t count = send( + unix_socket_fd((LT_UnixSocket*)socket), + bytes + offset, + length - offset, + 0 + ); +#endif + if (count < 0 && errno == EINTR){ + LT_socket_interrupted(); + continue; + } + if (count <= 0){ + LT_system_error( + "Unix socket write failed", + count < 0 ? errno : EPIPE + ); + } + offset += (size_t)count; + } +} + +void LT_UnixStreamSocket_shutdown_write(LT_UnixStreamSocket* socket){ + if (shutdown(unix_socket_fd((LT_UnixSocket*)socket), SHUT_WR) != 0){ + LT_system_error("Unix socket shutdown failed", errno); + } +} + +LT_UnixServerSocket* LT_UnixServerSocket_new(const char* path, int backlog){ + LT_UnixServerSocket* result; + + if (backlog < 1){ + LT_error("Socket backlog must be positive"); + } + result = LT_Class_ALLOC(LT_UnixServerSocket); + unix_socket_set_fd( + (LT_UnixSocket*)result, + unix_bind(path, SOCK_STREAM, backlog) + ); + return result; +} + +LT_UnixStreamSocket* LT_UnixServerSocket_accept( + LT_UnixServerSocket* socket +){ + int fd; + + while (1){ + fd = accept(unix_socket_fd((LT_UnixSocket*)socket), NULL, NULL); + if (fd >= 0 || errno != EINTR){ + break; + } + LT_socket_interrupted(); + } + if (fd < 0){ + LT_system_error("Unix socket accept failed", errno); + } + return unix_stream_from_fd(fd); +} + +static LT_String* unix_path_arg(LT_Value* cursor){ + LT_String* path; + + LT_GENERIC_ARG(*cursor, path, LT_String*, LT_String_from_value); + return path; +} + +LT_DEFINE_PRIMITIVE( + unix_socket_closed, + "UnixSocket>>isClosed", + "(self)", + "Return true when the Unix socket is closed." +){ + LT_Value cursor = arguments; + LT_UnixSocket* self; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG(cursor, self, LT_UnixSocket*, unix_socket_from_value); + LT_ARG_END(cursor); + return LT_UnixSocket_closed(self) ? LT_TRUE : LT_FALSE; +} + +LT_DEFINE_PRIMITIVE( + unix_socket_close, + "UnixSocket>>close", + "(self)", + "Close the Unix socket." +){ + LT_Value cursor = arguments; + LT_UnixSocket* self; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG(cursor, self, LT_UnixSocket*, unix_socket_from_value); + LT_ARG_END(cursor); + LT_UnixSocket_close(self); + return (LT_Value)(uintptr_t)self; +} + +#define UNIX_PATH_CONSTRUCTOR(c_name, primitive_name, function) \ +LT_DEFINE_PRIMITIVE( \ + c_name, primitive_name, "(self path)", "Create a Unix socket." \ +){ \ + LT_Value cursor = arguments; \ + LT_Value self; \ + LT_String* path; \ + \ + (void)tail_call_unwind_marker; \ + LT_OBJECT_ARG(cursor, self); \ + path = unix_path_arg(&cursor); \ + LT_ARG_END(cursor); \ + (void)self; \ + return (LT_Value)(uintptr_t)function(LT_String_value_cstr(path)); \ +} + +UNIX_PATH_CONSTRUCTOR( + unix_datagram_bind, + "UnixDatagramSocket class>>bindTo:", + LT_UnixDatagramSocket_bind +) +UNIX_PATH_CONSTRUCTOR( + unix_datagram_connect, + "UnixDatagramSocket class>>connectTo:", + LT_UnixDatagramSocket_connect +) +UNIX_PATH_CONSTRUCTOR( + unix_stream_connect, + "UnixStreamSocket class>>connectTo:", + LT_UnixStreamSocket_connect +) + +#undef UNIX_PATH_CONSTRUCTOR + +LT_DEFINE_PRIMITIVE( + unix_datagram_send, + "UnixDatagramSocket>>send:", + "(self bytes)", + "Send one Unix datagram." +){ + LT_Value cursor = arguments; + LT_UnixDatagramSocket* self; + LT_ByteVector* bytes; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG( + cursor, + self, + LT_UnixDatagramSocket*, + LT_UnixDatagramSocket_from_value + ); + LT_GENERIC_ARG(cursor, bytes, LT_ByteVector*, LT_ByteVector_from_value); + LT_ARG_END(cursor); + return LT_Number_smallinteger_from_size( + LT_UnixDatagramSocket_send(self, bytes), + "Datagram too large" + ); +} + +LT_DEFINE_PRIMITIVE( + unix_datagram_receive, + "UnixDatagramSocket>>receive:", + "(self maximumLength)", + "Receive one Unix datagram." +){ + LT_Value cursor = arguments; + LT_Value maximum_length; + LT_UnixDatagramSocket* self; + size_t length; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG( + cursor, + self, + LT_UnixDatagramSocket*, + LT_UnixDatagramSocket_from_value + ); + LT_OBJECT_ARG(cursor, maximum_length); + LT_ARG_END(cursor); + length = LT_Number_nonnegative_size_from_integer( + maximum_length, + "Invalid receive length", + "Invalid receive length" + ); + return (LT_Value)(uintptr_t)LT_UnixDatagramSocket_receive(self, length); +} + +LT_DEFINE_PRIMITIVE( + unix_stream_pair, + "UnixStreamSocket class>>pair", + "(self)", + "Create and return a connected pair of Unix stream sockets." +){ + LT_Value cursor = arguments; + LT_Value self; + LT_UnixStreamSocket* first; + LT_UnixStreamSocket* second; + + (void)tail_call_unwind_marker; + LT_OBJECT_ARG(cursor, self); + LT_ARG_END(cursor); + (void)self; + LT_UnixStreamSocket_pair(&first, &second); + return LT_cons( + (LT_Value)(uintptr_t)first, + LT_cons((LT_Value)(uintptr_t)second, LT_NIL) + ); +} + +LT_DEFINE_PRIMITIVE( + unix_stream_read, + "UnixStreamSocket>>read:", + "(self maximumLength)", + "Read bytes from a Unix stream socket." +){ + LT_Value cursor = arguments; + LT_Value maximum_length; + LT_UnixStreamSocket* self; + size_t length; + size_t count; + uint8_t* bytes; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG( + cursor, + self, + LT_UnixStreamSocket*, + LT_UnixStreamSocket_from_value + ); + LT_OBJECT_ARG(cursor, maximum_length); + LT_ARG_END(cursor); + length = LT_Number_nonnegative_size_from_integer( + maximum_length, + "Invalid read length", + "Invalid read length" + ); + bytes = GC_MALLOC_ATOMIC(length ? length : 1); + count = LT_UnixStreamSocket_read(self, bytes, length); + return (LT_Value)(uintptr_t)LT_ByteVector_new(bytes, count); +} + +LT_DEFINE_PRIMITIVE( + unix_stream_write, + "UnixStreamSocket>>write:", + "(self bytes)", + "Write bytes to a Unix stream socket." +){ + LT_Value cursor = arguments; + LT_UnixStreamSocket* self; + LT_ByteVector* bytes; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG( + cursor, + self, + LT_UnixStreamSocket*, + LT_UnixStreamSocket_from_value + ); + LT_GENERIC_ARG(cursor, bytes, LT_ByteVector*, LT_ByteVector_from_value); + LT_ARG_END(cursor); + LT_UnixStreamSocket_write( + self, + LT_ByteVector_bytes(bytes), + LT_ByteVector_length(bytes) + ); + return (LT_Value)(uintptr_t)self; +} + +LT_DEFINE_PRIMITIVE( + unix_stream_read_line, + "UnixStreamSocket>>readLine", + "(self)", + "Read through a line feed and include it in the returned bytevector." +){ + LT_Value cursor = arguments; + LT_UnixStreamSocket* self; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG( + cursor, + self, + LT_UnixStreamSocket*, + LT_UnixStreamSocket_from_value + ); + LT_ARG_END(cursor); + return LT_UnixStreamSocket_readLine(self); +} + +LT_DEFINE_PRIMITIVE( + unix_stream_shutdown, + "UnixStreamSocket>>shutdownWrite", + "(self)", + "Shut down the writing half of a Unix stream socket." +){ + LT_Value cursor = arguments; + LT_UnixStreamSocket* self; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG( + cursor, + self, + LT_UnixStreamSocket*, + LT_UnixStreamSocket_from_value + ); + LT_ARG_END(cursor); + LT_UnixStreamSocket_shutdown_write(self); + return (LT_Value)(uintptr_t)self; +} + +LT_DEFINE_PRIMITIVE( + unix_server_new, + "UnixServerSocket class>>newOn:backlog:", + "(self path backlog)", + "Create a listening Unix socket." +){ + LT_Value cursor = arguments; + LT_Value self; + LT_Value backlog_value; + LT_String* path; + size_t backlog; + + (void)tail_call_unwind_marker; + LT_OBJECT_ARG(cursor, self); + path = unix_path_arg(&cursor); + LT_OBJECT_ARG(cursor, backlog_value); + LT_ARG_END(cursor); + backlog = LT_Number_nonnegative_size_from_integer( + backlog_value, + "Invalid socket backlog", + "Invalid socket backlog" + ); + if (backlog < 1 || backlog > INT_MAX){ + LT_error("Invalid socket backlog"); + } + (void)self; + return (LT_Value)(uintptr_t)LT_UnixServerSocket_new( + LT_String_value_cstr(path), + (int)backlog + ); +} + +LT_DEFINE_PRIMITIVE( + unix_server_new_default_backlog, + "UnixServerSocket class>>newOn:", + "(self path)", + "Create a listening Unix socket using the system maximum backlog." +){ + LT_Value cursor = arguments; + LT_Value self; + LT_String* path; + + (void)tail_call_unwind_marker; + LT_OBJECT_ARG(cursor, self); + path = unix_path_arg(&cursor); + LT_ARG_END(cursor); + (void)self; + return (LT_Value)(uintptr_t)LT_UnixServerSocket_new( + LT_String_value_cstr(path), + SOMAXCONN + ); +} + +LT_DEFINE_PRIMITIVE( + unix_server_accept, + "UnixServerSocket>>accept", + "(self)", + "Accept and return a Unix stream socket." +){ + LT_Value cursor = arguments; + LT_UnixServerSocket* self; + + (void)tail_call_unwind_marker; + LT_GENERIC_ARG( + cursor, + self, + LT_UnixServerSocket*, + LT_UnixServerSocket_from_value + ); + LT_ARG_END(cursor); + return (LT_Value)(uintptr_t)LT_UnixServerSocket_accept(self); +} + +static LT_Method_Descriptor unix_socket_methods[] = { + {"isClosed", &unix_socket_closed}, + {"close", &unix_socket_close}, + LT_NULL_NATIVE_CLASS_METHOD_DESCRIPTOR +}; + +static LT_Method_Descriptor unix_datagram_methods[] = { + {"send:", &unix_datagram_send}, + {"receive:", &unix_datagram_receive}, + LT_NULL_NATIVE_CLASS_METHOD_DESCRIPTOR +}; + +static LT_Method_Descriptor unix_datagram_class_methods[] = { + {"bindTo:", &unix_datagram_bind}, + {"connectTo:", &unix_datagram_connect}, + LT_NULL_NATIVE_CLASS_METHOD_DESCRIPTOR +}; + +static LT_Method_Descriptor unix_stream_methods[] = { + {"read:", &unix_stream_read}, + {"readLine", &unix_stream_read_line}, + {"write:", &unix_stream_write}, + {"shutdownWrite", &unix_stream_shutdown}, + LT_NULL_NATIVE_CLASS_METHOD_DESCRIPTOR +}; + +static LT_Method_Descriptor unix_stream_class_methods[] = { + {"connectTo:", &unix_stream_connect}, + {"pair", &unix_stream_pair}, + LT_NULL_NATIVE_CLASS_METHOD_DESCRIPTOR +}; + +static LT_Method_Descriptor unix_server_methods[] = { + {"accept", &unix_server_accept}, + LT_NULL_NATIVE_CLASS_METHOD_DESCRIPTOR +}; + +static LT_Method_Descriptor unix_server_class_methods[] = { + {"newOn:", &unix_server_new_default_backlog}, + {"newOn:backlog:", &unix_server_new}, + LT_NULL_NATIVE_CLASS_METHOD_DESCRIPTOR +}; + +static LT_Class* unix_datagram_mixins[] = { + <_DatagramSocket_class, + NULL +}; + +static LT_Class* unix_stream_mixins[] = { + <_StreamSocket_class, + NULL +}; + +static LT_Class* unix_server_mixins[] = { + <_ServerSocket_class, + NULL +}; + +LT_DEFINE_CLASS(LT_UnixSocket) { + .superclass = <_Socket_class, + .metaclass_superclass = <_Class_class, + .package = "ListTalk:Socket", + .name = "UnixSocket", + .documentation = "Abstract Unix-domain socket.", + .instance_size = sizeof(LT_UnixSocket), + .class_flags = LT_CLASS_FLAG_ABSTRACT, + .methods = unix_socket_methods, +}; + +LT_DEFINE_CLASS(LT_UnixDatagramSocket) { + .superclass = <_UnixSocket_class, + .mixins = unix_datagram_mixins, + .metaclass_superclass = <_Class_class, + .package = "ListTalk:Socket", + .name = "UnixDatagramSocket", + .documentation = "Unix-domain datagram socket.", + .instance_size = sizeof(LT_UnixDatagramSocket), + .methods = unix_datagram_methods, + .class_methods = unix_datagram_class_methods, +}; + +LT_DEFINE_CLASS(LT_UnixStreamSocket) { + .superclass = <_UnixSocket_class, + .mixins = unix_stream_mixins, + .metaclass_superclass = <_Class_class, + .package = "ListTalk:Socket", + .name = "UnixStreamSocket", + .documentation = "Connected Unix-domain stream socket.", + .instance_size = sizeof(LT_UnixStreamSocket), + .methods = unix_stream_methods, + .class_methods = unix_stream_class_methods, +}; + +LT_DEFINE_CLASS(LT_UnixServerSocket) { + .superclass = <_UnixSocket_class, + .mixins = unix_server_mixins, + .metaclass_superclass = <_Class_class, + .package = "ListTalk:Socket", + .name = "UnixServerSocket", + .documentation = "Listening Unix-domain stream socket.", + .instance_size = sizeof(LT_UnixServerSocket), + .methods = unix_server_methods, + .class_methods = unix_server_class_methods, +}; diff --git a/src/vm/eval.c b/src/vm/eval.c index 8b1fafc..19743a8 100644 --- a/src/vm/eval.c +++ b/src/vm/eval.c @@ -109,7 +109,7 @@ LT_DEFINE_PRIMITIVE_RESTART( LT_throw(unbound_symbol_define_variable_tag(), value); } -static void check_pending_signal(void){ +void LT_check_pending_signal(void){ LT_ThreadState* state = LT__thread_state; LT_Value signal; @@ -919,7 +919,7 @@ static LT_Value eval_form(LT_Value expression, LT_TailCallUnwindMarker* tail_call_unwind_marker){ LT_StackFrame stack_frame; - check_pending_signal(); + LT_check_pending_signal(); stack_frame.type = LT_STACK_FRAME_TYPE_EVAL; stack_frame.arguments.eval.expression = expression; diff --git a/tests/eval-socket.lt b/tests/eval-socket.lt new file mode 100644 index 0000000..5d1588d --- /dev/null +++ b/tests/eval-socket.lt @@ -0,0 +1,34 @@ +;;;;; -*- mode: scheme -*- +; SPDX-License-Identifier: MIT + +(require :socket) + +(define (check condition description) + (if condition #true (error (string-append "FAIL: " description)))) + +(define socket [ListTalk:Socket:UDPSocket bindTo: "127.0.0.1" port: 0]) +(check (not [socket isClosed]) "new datagram socket is open") +[socket close] +(check [socket isClosed] "inherited close closes a datagram socket") + +(define sockets [ListTalk:Socket:UnixStreamSocket pair]) +(check (and (pair? sockets) (pair? (cdr sockets)) (null? (cddr sockets))) + "UnixStreamSocket pair returns a two-element list") +(define first (car sockets)) +(define second (cadr sockets)) +[first write: (bytevector 1 2 3)] +(check (equal? [second read: 3] (bytevector 1 2 3)) + "Unix stream socket pair transports bytes") +[first write: (bytevector 4 10 5 6)] +(check (equal? [second readLine] (bytevector 4 10)) + "readLine includes its line-feed delimiter") +(check (equal? [second read: 2] (bytevector 5 6)) + "ordinary reads consume bytes buffered by readLine") +[first close] +[second close] + +(define server + [ListTalk:Socket:TCPServerSocket newOn: "127.0.0.1" port: 0]) +(check (not [server isClosed]) + "TCP server constructor defaults its backlog") +[server close] diff --git a/tests/networking_test.c b/tests/networking_test.c new file mode 100644 index 0000000..f54130e --- /dev/null +++ b/tests/networking_test.c @@ -0,0 +1,242 @@ +/* SPDX-License-Identifier: MIT */ +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +static int failures; + +typedef struct ShortWriteArguments { + LT_UnixStreamSocket* socket; +} ShortWriteArguments; + +static void* short_write(void* data){ + ShortWriteArguments* arguments = data; + struct timespec delay = { + .tv_sec = 0, + .tv_nsec = 10000000, + }; + + LT_UnixStreamSocket_write(arguments->socket, "12", 2); + nanosleep(&delay, NULL); + LT_UnixStreamSocket_write(arguments->socket, "34", 2); + return NULL; +} + +static void check(int condition, const char* message){ + if (!condition){ + fprintf(stderr, "FAIL: %s\n", message); + failures++; + } +} + +static uint16_t local_port(LT_IPSocket* socket){ + struct sockaddr_in address; + socklen_t length = sizeof(address); + + if (getsockname( + LT_IPSocket_descriptor(socket), + (struct sockaddr*)&address, + &length + ) != 0){ + return 0; + } + return ntohs(address.sin_port); +} + +int main(void){ + LT_TCPServerSocket* server; + LT_TCPServerSocket* wildcard_server; + LT_TCPSocket *client, *peer; + LT_UDPSocket *receiver, *sender; + LT_UnixStreamSocket *unix_first, *unix_second; + LT_UnixServerSocket* unix_server; + LT_ByteVector *message, *received; + LT_Value line; + char reply[4]; + char unix_path[96]; + int option; + socklen_t option_length; + struct sockaddr_storage wildcard_address; + socklen_t wildcard_address_length; + ShortWriteArguments short_write_arguments; + pthread_t writer; + int writer_created; + + LT_INIT(); + wildcard_server = LT_TCPServerSocket_new("*", 0, SOMAXCONN); + option = 0; + option_length = sizeof(option); + check( + getsockopt( + LT_IPSocket_descriptor((LT_IPSocket*)wildcard_server), + SOL_SOCKET, + SO_REUSEADDR, + &option, + &option_length + ) == 0 && option != 0, + "TCP wildcard server enables SO_REUSEADDR" + ); + wildcard_address_length = sizeof(wildcard_address); + check( + getsockname( + LT_IPSocket_descriptor((LT_IPSocket*)wildcard_server), + (struct sockaddr*)&wildcard_address, + &wildcard_address_length + ) == 0, + "TCP wildcard server has a local address" + ); + if (wildcard_address.ss_family == AF_INET6){ + option = 1; + option_length = sizeof(option); + check( + getsockopt( + LT_IPSocket_descriptor((LT_IPSocket*)wildcard_server), + IPPROTO_IPV6, + IPV6_V6ONLY, + &option, + &option_length + ) == 0 && option == 0, + "TCP IPv6 server disables IPV6_V6ONLY" + ); + } + LT_IPSocket_close((LT_IPSocket*)wildcard_server); + + server = LT_TCPServerSocket_new("127.0.0.1", 0, 4); + check( + LT_Value_is_instance_of( + (LT_Value)(uintptr_t)server, + LT_STATIC_CLASS(LT_ServerSocket) + ), + "TCPServerSocket implements ServerSocket" + ); + client = LT_TCPSocket_connect( + "127.0.0.1", + local_port((LT_IPSocket*)server) + ); + peer = LT_TCPServerSocket_accept(server); + check( + LT_Value_is_instance_of( + (LT_Value)(uintptr_t)peer, + LT_STATIC_CLASS(LT_StreamSocket) + ), + "TCPSocket implements StreamSocket" + ); + LT_TCPSocket_write(client, "ping", 4); + check( + LT_TCPSocket_read(peer, reply, sizeof(reply)) == 4 + && !memcmp(reply, "ping", 4), + "stream socket loopback round trip" + ); + LT_IPSocket_close((LT_IPSocket*)peer); + LT_IPSocket_close((LT_IPSocket*)client); + LT_IPSocket_close((LT_IPSocket*)server); + check( + LT_IPSocket_closed((LT_IPSocket*)server), + "closed server reports closed" + ); + + receiver = LT_UDPSocket_bind("127.0.0.1", 0); + check( + LT_Value_is_instance_of( + (LT_Value)(uintptr_t)receiver, + LT_STATIC_CLASS(LT_DatagramSocket) + ), + "UDPSocket implements DatagramSocket" + ); + sender = LT_UDPSocket_connect( + "127.0.0.1", + local_port((LT_IPSocket*)receiver) + ); + message = LT_ByteVector_new((uint8_t*)"udp", 3); + check( + LT_UDPSocket_send(sender, message) == 3, + "datagram send length" + ); + received = LT_UDPSocket_receive(receiver, 16); + check( + LT_ByteVector_length(received) == 3 + && !memcmp(LT_ByteVector_bytes(received), "udp", 3), + "datagram socket loopback receive" + ); + LT_IPSocket_close((LT_IPSocket*)sender); + LT_IPSocket_close((LT_IPSocket*)receiver); + + LT_UnixStreamSocket_pair(&unix_first, &unix_second); + LT_UnixStreamSocket_write(unix_first, "pair", 4); + check( + LT_UnixStreamSocket_read(unix_second, reply, sizeof(reply)) == 4 + && !memcmp(reply, "pair", 4), + "Unix stream socket pair round trip" + ); + LT_UnixStreamSocket_write(unix_first, "a\nbc", 4); + line = LT_UnixStreamSocket_readLine(unix_second); + check( + LT_ByteVector_p(line) + && LT_ByteVector_length(LT_ByteVector_from_value(line)) == 2 + && !memcmp( + LT_ByteVector_bytes(LT_ByteVector_from_value(line)), + "a\n", + 2 + ), + "Unix stream readLine includes the delimiter" + ); + check( + LT_UnixStreamSocket_read(unix_second, reply, 2) == 2 + && !memcmp(reply, "bc", 2), + "Unix stream read consumes readLine buffered bytes" + ); + short_write_arguments.socket = unix_first; + writer_created = pthread_create( + &writer, + NULL, + short_write, + &short_write_arguments + ) == 0; + check(writer_created, "create short-write thread"); + if (writer_created){ + check( + LT_UnixStreamSocket_read(unix_second, reply, 4) == 4 + && !memcmp(reply, "1234", 4), + "high-level read shields caller from short reads" + ); + pthread_join(writer, NULL); + } + check( + LT_Value_is_instance_of( + (LT_Value)(uintptr_t)unix_first, + LT_STATIC_CLASS(LT_StreamSocket) + ), + "UnixStreamSocket implements StreamSocket" + ); + LT_UnixSocket_close((LT_UnixSocket*)unix_first); + LT_UnixSocket_close((LT_UnixSocket*)unix_second); + + snprintf( + unix_path, + sizeof(unix_path), + "/tmp/listtalk-networking-test-%ld.sock", + (long)getpid() + ); + unlink(unix_path); + unix_server = LT_UnixServerSocket_new(unix_path, 4); + unix_first = LT_UnixStreamSocket_connect(unix_path); + unix_second = LT_UnixServerSocket_accept(unix_server); + LT_UnixStreamSocket_write(unix_first, "path", 4); + check( + LT_UnixStreamSocket_read(unix_second, reply, sizeof(reply)) == 4 + && !memcmp(reply, "path", 4), + "pathname Unix stream socket round trip" + ); + LT_UnixSocket_close((LT_UnixSocket*)unix_first); + LT_UnixSocket_close((LT_UnixSocket*)unix_second); + LT_UnixSocket_close((LT_UnixSocket*)unix_server); + unlink(unix_path); + return failures ? 1 : 0; +}