Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion TODO.md
Original file line number Diff line number Diff line change
Expand Up @@ -3,4 +3,12 @@
## Handling of negative return code

In msocket, when stream data processing returns an error, its internal io_task thread breaks out of the loop and terminates immediately without calling msocket_common_on_disconnected().
Can we improve this design?
Can we improve this design?

## TCP_NODELAY should be optional

In msocket, all TCP sockets are automatically using TCP_NODELAY. This feature should be optional using a setting or config.

## Remove state MSOCKET_STATE_ACCEPTING

The MSOCKET_STATE_ACCEPTING doesn't serve any purpose except for debugging. It should be removed
29 changes: 29 additions & 0 deletions include/msocket.h
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,12 @@ typedef uint8_t msocket_state_t;
#define MSOCKET_ADDR_INET 0u
#define MSOCKET_ADDR_UNIX 1u
#define MSOCKET_ADDR_INET6 3u
#define MSOCKET_ADDR_VSOCK 4u

#define MSOCKET_VMADDR_CID_ANY 0xFFFFFFFFU
#define MSOCKET_VMADDR_CID_HYPERVISOR 0U
#define MSOCKET_VMADDR_CID_LOCAL 1U
#define MSOCKET_VMADDR_CID_HOST 2U

#define MSOCKET_ADDRSTRLEN 46u

Expand Down Expand Up @@ -224,6 +230,16 @@ msocket_error_t msocket_listen(msocket_t *self, uint8_t mode, uint16_t port, con
*/
msocket_error_t msocket_unix_listen(msocket_t *self, const char *socket_path);

/**
* Binds and configures a listening VSOCK socket (Linux only).
*
* @param self Pointer to msocket_t instance.
* @param cid Context ID to bind to (e.g. MSOCKET_VMADDR_CID_ANY).
* @param port VSOCK port number to bind to.
* @return MSOCKET_NO_ERROR on success, or error code on failure.
*/
msocket_error_t msocket_vsock_listen(msocket_t *self, uint32_t cid, uint32_t port);

/**
* Accepts an incoming client connection on a listening TCP or UNIX socket.
*
Expand Down Expand Up @@ -295,6 +311,19 @@ msocket_error_t msocket_connect(msocket_t *self, const char *addr, uint16_t port
*/
msocket_error_t msocket_unix_connect(msocket_t *self, const char *socket_path);

/**
* Initiates an outgoing VSOCK connection (Linux only).
*
* Requires a handler table to be registered beforehand via msocket_set_handler().
* Upon connection, the background I/O event loop is launched automatically.
*
* @param self Pointer to msocket_t instance.
* @param cid Target Context ID (e.g. MSOCKET_VMADDR_CID_HOST or specific guest CID).
* @param port Target VSOCK port.
* @return MSOCKET_NO_ERROR on success, or error code on failure.
*/
msocket_error_t msocket_vsock_connect(msocket_t *self, uint32_t cid, uint32_t port);

/**
* Sends data to a specific destination over UDP.
*
Expand Down
11 changes: 11 additions & 0 deletions include/msocket_server.h
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,8 @@ typedef struct msocket_server_tag {
uint16_t udp_port;
char *udp_addr;
char *socket_path;
uint32_t vsock_cid;
uint32_t vsock_port;
adt_ary_t cleanup_items;
uint8_t cleanup_stop;
uint8_t address_family;
Expand Down Expand Up @@ -124,6 +126,15 @@ void msocket_server_start(msocket_server_t *self, const char *udp_addr, uint16_t
*/
void msocket_server_unix_start(msocket_server_t *self, const char *socket_path);

/**
* Binds a listening VSOCK socket and starts the server threads (Linux only).
*
* @param self Pointer to msocket_server_t instance.
* @param cid Context ID to bind to (e.g. MSOCKET_VMADDR_CID_ANY).
* @param port VSOCK port to listen on.
*/
void msocket_server_vsock_start(msocket_server_t *self, uint32_t cid, uint32_t port);

/**
* Binds and starts a listening TCP server with TLS enabled.
*
Expand Down
89 changes: 89 additions & 0 deletions source/msocket_common.c
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,12 @@ msocket_error_t msocket_create(msocket_t *self, uint8_t address_family)
#endif
self->address_family = MSOCKET_ADDR_UNIX;
break;
case MSOCKET_ADDR_VSOCK:
#if defined(AF_VSOCK) && (AF_VSOCK != MSOCKET_ADDR_VSOCK)
case AF_VSOCK:
#endif
self->address_family = MSOCKET_ADDR_VSOCK;
break;
default:
return MSOCKET_INVALID_ARGUMENT_ERROR;
}
Expand Down Expand Up @@ -332,6 +338,14 @@ msocket_error_t msocket_unix_listen(msocket_t *self, const char *socket_path)
return msocket_os_unix_listen(self, socket_path);
}

msocket_error_t msocket_vsock_listen(msocket_t *self, uint32_t cid, uint32_t port)
{
if (self == NULL || self->address_family != MSOCKET_ADDR_VSOCK || port == 0u) {
return MSOCKET_INVALID_ARGUMENT_ERROR;
}
return msocket_os_vsock_listen(self, cid, port);
}

msocket_t *msocket_accept(msocket_t *self, msocket_t *child)
{
if (self == NULL || self->os == NULL || self->state != MSOCKET_STATE_LISTENING) {
Expand All @@ -351,6 +365,15 @@ msocket_t *msocket_accept(msocket_t *self, msocket_t *child)
}

msocket_os_mutex_lock(self->os);
if (self->state == MSOCKET_STATE_CLOSING) {
msocket_os_mutex_unlock(self->os);
if (placement_new) {
msocket_destroy(child);
} else {
msocket_delete(child);
}
return NULL;
}
self->state = MSOCKET_STATE_ACCEPTING;
os_socket_t accept_fd = self->os->tcp_sockfd;
msocket_os_mutex_unlock(self->os);
Expand All @@ -364,6 +387,45 @@ msocket_t *msocket_accept(msocket_t *self, msocket_t *child)
return NULL;
}

while (1) {
fd_set readfds;
FD_ZERO(&readfds);
FD_SET(accept_fd, &readfds);

struct timeval timeout;
timeout.tv_sec = 0;
timeout.tv_usec = TIMEOUT_US;

int activity = select((int)accept_fd + 1, &readfds, NULL, NULL, &timeout);
if (activity > 0) {
break;
} else if (activity == 0) {
msocket_state_t state;
msocket_os_mutex_lock(self->os);
state = self->state;
msocket_os_mutex_unlock(self->os);

if (state == MSOCKET_STATE_CLOSING) {
if (placement_new) {
msocket_destroy(child);
} else {
msocket_delete(child);
}
return NULL;
}
} else {
if (OS_SOCKET_ERRNO_IS_INTR()) {
continue;
}
if (placement_new) {
msocket_destroy(child);
} else {
msocket_delete(child);
}
return NULL;
}
}

os_socket_t sockfd;
int one = 1;
msocket_error_t result = MSOCKET_NO_ERROR;
Expand Down Expand Up @@ -392,6 +454,22 @@ msocket_t *msocket_accept(msocket_t *self, msocket_t *child)
setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, OS_SOCKOPT_CAST(&one), sizeof(one));
child->os->tcp_sockfd = sockfd;
}
} else if (self->address_family == MSOCKET_ADDR_VSOCK) {
#if defined(__linux__) && defined(AF_VSOCK)
struct sockaddr_vm cli_addr_vm;
OS_SOCK_LEN_T cli_len = (OS_SOCK_LEN_T)sizeof(cli_addr_vm);
memset(&cli_addr_vm, 0, sizeof(cli_addr_vm));
sockfd = accept(accept_fd, (struct sockaddr *)&cli_addr_vm, &cli_len);
if (OS_SOCKET_IS_INVALID(sockfd)) {
result = MSOCKET_SOCKET_ERROR;
} else {
snprintf(child->stream_info.addr, MSOCKET_ADDRSTRLEN, "cid:%u", cli_addr_vm.svm_cid);
child->stream_info.port = (uint16_t)(cli_addr_vm.svm_port & 0xFFFF);
child->os->tcp_sockfd = sockfd;
}
#else
result = MSOCKET_NOT_IMPLEMENTED_ERROR;
#endif
} else {
struct sockaddr_in cli_addr;
OS_SOCK_LEN_T cli_len = (OS_SOCK_LEN_T)sizeof(cli_addr);
Expand Down Expand Up @@ -542,6 +620,17 @@ msocket_error_t msocket_unix_connect(msocket_t *self, const char *socket_path)
return msocket_os_unix_connect(self, socket_path);
}

msocket_error_t msocket_vsock_connect(msocket_t *self, uint32_t cid, uint32_t port)
{
if (self == NULL || port == 0u || (self->socket_mode & MSOCKET_MODE_STREAM) != 0) {
return MSOCKET_INVALID_ARGUMENT_ERROR;
}
if (self->handler_table == NULL) {
return MSOCKET_INVALID_ARGUMENT_ERROR;
}
return msocket_os_vsock_connect(self, cid, port);
}

msocket_error_t msocket_send(msocket_t *self, const void *msg_data, uint32_t msg_len)
{
if (self == NULL || msg_data == NULL || (self->socket_mode & MSOCKET_MODE_STREAM) == 0) {
Expand Down
6 changes: 6 additions & 0 deletions source/msocket_internal.h
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,9 @@ typedef SOCKET os_socket_t;
# include <unistd.h>
# include <pthread.h>
# include <semaphore.h>
# if defined(__linux__)
# include <linux/vm_sockets.h>
# endif

/* Sockets */
typedef int os_socket_t;
Expand Down Expand Up @@ -185,6 +188,9 @@ void msocket_os_mutex_unlock(msocket_os_t *os);
msocket_error_t msocket_os_unix_listen(msocket_t *self, const char *socket_path);
msocket_error_t msocket_os_unix_connect(msocket_t *self, const char *socket_path);

msocket_error_t msocket_os_vsock_listen(msocket_t *self, uint32_t cid, uint32_t port);
msocket_error_t msocket_os_vsock_connect(msocket_t *self, uint32_t cid, uint32_t port);

//////////////////////////////////////////////////////////////////////////////
// COMMON CALLBACKS FROM PLATFORM
//////////////////////////////////////////////////////////////////////////////
Expand Down
88 changes: 88 additions & 0 deletions source/msocket_posix.c
Original file line number Diff line number Diff line change
Expand Up @@ -223,3 +223,91 @@ msocket_error_t msocket_os_unix_connect(msocket_t *self, const char *socket_path
}
return MSOCKET_NO_ERROR;
}

//////////////////////////////////////////////////////////////////////////////
// VSOCK SOCKETS
//////////////////////////////////////////////////////////////////////////////

msocket_error_t msocket_os_vsock_listen(msocket_t *self, uint32_t cid, uint32_t port)
{
#if defined(__linux__) && defined(AF_VSOCK)
struct sockaddr_vm saddr;
int sockvsock;
int rc;

memset(&saddr, 0, sizeof(saddr));
saddr.svm_family = AF_VSOCK;
saddr.svm_cid = cid;
saddr.svm_port = port;

sockvsock = socket(AF_VSOCK, SOCK_STREAM, 0);
if (sockvsock < 0) {
return MSOCKET_SOCKET_ERROR;
}

rc = bind(sockvsock, (struct sockaddr *)&saddr, sizeof(saddr));
if (rc < 0) {
close(sockvsock);
return MSOCKET_SOCKET_ERROR;
}

rc = listen(sockvsock, 5);
if (rc < 0) {
close(sockvsock);
return MSOCKET_SOCKET_ERROR;
}

self->os->tcp_sockfd = sockvsock;
self->state = MSOCKET_STATE_LISTENING;
self->socket_mode |= MSOCKET_MODE_STREAM;
return MSOCKET_NO_ERROR;
#else
(void)self;
(void)cid;
(void)port;
return MSOCKET_NOT_IMPLEMENTED_ERROR;
#endif
}

msocket_error_t msocket_os_vsock_connect(msocket_t *self, uint32_t cid, uint32_t port)
{
#if defined(__linux__) && defined(AF_VSOCK)
struct sockaddr_vm saddr;
int sockfd = socket(AF_VSOCK, SOCK_STREAM, 0);
if (sockfd < 0) {
return MSOCKET_SOCKET_ERROR;
}

memset(&saddr, 0, sizeof(saddr));
saddr.svm_family = AF_VSOCK;
saddr.svm_cid = cid;
saddr.svm_port = port;

int rc = connect(sockfd, (struct sockaddr *)&saddr, sizeof(saddr));
if (rc < 0) {
close(sockfd);
return MSOCKET_SOCKET_ERROR;
}

snprintf(self->stream_info.addr, MSOCKET_ADDRSTRLEN, "cid:%u", cid);
self->stream_info.port = (uint16_t)(port & 0xFFFF);
self->os->tcp_sockfd = sockfd;
self->socket_mode |= MSOCKET_MODE_STREAM;
self->state = MSOCKET_STATE_ESTABLISHED;
self->os->new_connection = true;

msocket_error_t io_rc = msocket_start_io(self);
if (io_rc != MSOCKET_NO_ERROR) {
close(sockfd);
self->os->tcp_sockfd = -1;
self->state = MSOCKET_STATE_CLOSED;
return io_rc;
}
return MSOCKET_NO_ERROR;
#else
(void)self;
(void)cid;
(void)port;
return MSOCKET_NOT_IMPLEMENTED_ERROR;
#endif
}
23 changes: 23 additions & 0 deletions source/msocket_server.c
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,8 @@ void msocket_server_create(msocket_server_t *self, uint8_t address_family, void
self->udp_port = 0u;
self->udp_addr = NULL;
self->socket_path = NULL;
self->vsock_cid = 0u;
self->vsock_port = 0u;
self->accept_socket = NULL;
self->cleanup_stop = 0u;
memset(&self->handler_table, 0, sizeof(self->handler_table));
Expand Down Expand Up @@ -169,6 +171,17 @@ void msocket_server_unix_start(msocket_server_t *self, const char *socket_path)
}
}

void msocket_server_vsock_start(msocket_server_t *self, uint32_t cid, uint32_t port)
{
if (self != NULL && port != 0u) {
self->tcp_port = 0u;
self->udp_port = 0u;
self->vsock_cid = cid;
self->vsock_port = port;
msocket_server_start_threads(self);
}
}

msocket_error_t msocket_server_start_tls(msocket_server_t *self, uint16_t tcp_port, const struct msocket_tls_config_tag *tls_config)
{
#if defined(MSOCKET_ENABLE_TLS)
Expand Down Expand Up @@ -258,6 +271,16 @@ static msocket_error_t msocket_server_bind(msocket_server_t *self)
}
}

if (self->vsock_port != 0u) {
rc = msocket_vsock_listen(self->accept_socket, self->vsock_cid, self->vsock_port);
if (rc != MSOCKET_NO_ERROR) {
fprintf(stderr, "[MSOCKET_SERVER] Failed to bind VSOCK port %u (CID %u)\n", self->vsock_port, self->vsock_cid);
msocket_delete(self->accept_socket);
self->accept_socket = NULL;
return rc;
}
}

return MSOCKET_NO_ERROR;
}

Expand Down
Loading
Loading