alex/swayfx
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1313 from ilyaluk/nonblock-ipc-server

Browse source 
Implement nonblocking IO in IPC server
This commit is contained in:
Drew DeVault 2017-08-08 17:57:44 -04:00 committed by GitHub
parent e12d1cf003 1c5a96c112
commit 8208e2505a
1 changed files with 96 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

101
sway/ipc-server.c
View file

@ -40,11 +40,15 @@ static const char ipc_magic[] = {'i', '3', '-', 'i', 'p', 'c'};
struct ipc_client { struct ipc_client {
struct wlc_event_source *event_source; struct wlc_event_source *event_source;
struct wlc_event_source *writable_event_source;
int fd; int fd;
uint32_t payload_length; uint32_t payload_length;
uint32_t security_policy; uint32_t security_policy;
enum ipc_command_type current_command; enum ipc_command_type current_command;
enum ipc_command_type subscribed_events; enum ipc_command_type subscribed_events;
size_t write_buffer_len;
size_t write_buffer_size;
char *write_buffer;
}; };
static list_t *ipc_get_pixel_requests = NULL; static list_t *ipc_get_pixel_requests = NULL;
@ -58,6 +62,7 @@ struct get_pixels_request {
struct sockaddr_un *ipc_user_sockaddr(void); struct sockaddr_un *ipc_user_sockaddr(void);
int ipc_handle_connection(int fd, uint32_t mask, void *data); int ipc_handle_connection(int fd, uint32_t mask, void *data);
int ipc_client_handle_readable(int client_fd, uint32_t mask, void *data); int ipc_client_handle_readable(int client_fd, uint32_t mask, void *data);
int ipc_client_handle_writable(int client_fd, uint32_t mask, void *data);
void ipc_client_disconnect(struct ipc_client *client); void ipc_client_disconnect(struct ipc_client *client);
void ipc_client_handle_command(struct ipc_client *client); void ipc_client_handle_command(struct ipc_client *client);
bool ipc_send_reply(struct ipc_client *client, const char *payload, uint32_t payload_length); bool ipc_send_reply(struct ipc_client *client, const char *payload, uint32_t payload_length);
@ -168,6 +173,12 @@ int ipc_handle_connection(int fd, uint32_t mask, void *data) {
close(client_fd); close(client_fd);
return 0; return 0;
} }
if ((flags = fcntl(client_fd, F_GETFL)) == -1
|| fcntl(client_fd, F_SETFL, flags|O_NONBLOCK) == -1) {
sway_log_errno(L_ERROR, "Unable to set NONBLOCK on IPC client socket");
close(client_fd);
return 0;
}
struct ipc_client* client = malloc(sizeof(struct ipc_client)); struct ipc_client* client = malloc(sizeof(struct ipc_client));
if (!client) { if (!client) {
@ -179,10 +190,22 @@ int ipc_handle_connection(int fd, uint32_t mask, void *data) {
client->fd = client_fd; client->fd = client_fd;
client->subscribed_events = 0; client->subscribed_events = 0;
client->event_source = wlc_event_loop_add_fd(client_fd, WLC_EVENT_READABLE, ipc_client_handle_readable, client); client->event_source = wlc_event_loop_add_fd(client_fd, WLC_EVENT_READABLE, ipc_client_handle_readable, client);
client->writable_event_source = NULL;
client->write_buffer_size = 128;
client->write_buffer_len = 0;
client->write_buffer = malloc(client->write_buffer_size);
if (!client->write_buffer) {
sway_log(L_ERROR, "Unable to allocate ipc client write buffer");
close(client_fd);
return 0;
}
pid_t pid = get_client_pid(client->fd); pid_t pid = get_client_pid(client->fd);
client->security_policy = get_ipc_policy_mask(pid); client->security_policy = get_ipc_policy_mask(pid);
sway_log(L_DEBUG, "New client: fd %d, pid %d", client_fd, pid);
list_add(ipc_client_list, client); list_add(ipc_client_list, client);
return 0; return 0;
@ -205,6 +228,8 @@ int ipc_client_handle_readable(int client_fd, uint32_t mask, void *data) {
return 0; return 0;
} }
sway_log(L_DEBUG, "Client %d readable", client->fd);
int read_available; int read_available;
if (ioctl(client_fd, FIONREAD, &read_available) == -1) { if (ioctl(client_fd, FIONREAD, &read_available) == -1) {
sway_log_errno(L_INFO, "Unable to read IPC socket buffer size"); sway_log_errno(L_INFO, "Unable to read IPC socket buffer size");
@ -226,6 +251,7 @@ int ipc_client_handle_readable(int client_fd, uint32_t mask, void *data) {
uint8_t buf[ipc_header_size]; uint8_t buf[ipc_header_size];
uint32_t *buf32 = (uint32_t*)(buf + sizeof(ipc_magic)); uint32_t *buf32 = (uint32_t*)(buf + sizeof(ipc_magic));
// Should be fully available, because read_available >= ipc_header_size
ssize_t received = recv(client_fd, buf, ipc_header_size, 0); ssize_t received = recv(client_fd, buf, ipc_header_size, 0);
if (received == -1) { if (received == -1) {
sway_log_errno(L_INFO, "Unable to receive header from IPC client"); sway_log_errno(L_INFO, "Unable to receive header from IPC client");
@ -249,6 +275,48 @@ int ipc_client_handle_readable(int client_fd, uint32_t mask, void *data) {
return 0; return 0;
} }
int ipc_client_handle_writable(int client_fd, uint32_t mask, void *data) {
struct ipc_client *client = data;
if (mask & WLC_EVENT_ERROR) {
sway_log(L_ERROR, "IPC Client socket error, removing client");
ipc_client_disconnect(client);
return 0;
}
if (mask & WLC_EVENT_HANGUP) {
sway_log(L_DEBUG, "Client %d hung up", client->fd);
ipc_client_disconnect(client);
return 0;
}
if (client->write_buffer_len <= 0) {
return 0;
}
sway_log(L_DEBUG, "Client %d writable", client->fd);
ssize_t written = write(client->fd, client->write_buffer, client->write_buffer_len);
if (written == -1 && errno == EAGAIN) {
return 0;
} else if (written == -1) {
sway_log_errno(L_INFO, "Unable to send data from queue to IPC client");
ipc_client_disconnect(client);
return 0;
}
memmove(client->write_buffer, client->write_buffer + written, client->write_buffer_len - written);
client->write_buffer_len -= written;
if (client->write_buffer_len == 0 && client->writable_event_source) {
wlc_event_source_remove(client->writable_event_source);
client->writable_event_source = NULL;
}
return 0;
}
void ipc_client_disconnect(struct ipc_client *client) { void ipc_client_disconnect(struct ipc_client *client) {
if (!sway_assert(client != NULL, "client != NULL")) { if (!sway_assert(client != NULL, "client != NULL")) {
return; return;
@ -260,9 +328,13 @@ void ipc_client_disconnect(struct ipc_client *client) {
sway_log(L_INFO, "IPC Client %d disconnected", client->fd); sway_log(L_INFO, "IPC Client %d disconnected", client->fd);
wlc_event_source_remove(client->event_source); wlc_event_source_remove(client->event_source);
if (client->writable_event_source) {
wlc_event_source_remove(client->writable_event_source);
}
int i = 0; int i = 0;
while (i < ipc_client_list->length && ipc_client_list->items[i] != client) i++; while (i < ipc_client_list->length && ipc_client_list->items[i] != client) i++;
list_del(ipc_client_list, i); list_del(ipc_client_list, i);
free(client->write_buffer);
close(client->fd); close(client->fd);
free(client); free(client);
} }
@ -334,6 +406,7 @@ void ipc_client_handle_command(struct ipc_client *client) {
return; return;
} }
if (client->payload_length > 0) { if (client->payload_length > 0) {
// Payload should be fully available
ssize_t received = recv(client->fd, buf, client->payload_length, 0); ssize_t received = recv(client->fd, buf, client->payload_length, 0);
if (received == -1) if (received == -1)
{ {
@ -590,17 +663,35 @@ bool ipc_send_reply(struct ipc_client *client, const char *payload, uint32_t pay
data32[0] = payload_length; data32[0] = payload_length;
data32[1] = client->current_command; data32[1] = client->current_command;
if (write(client->fd, data, ipc_header_size) == -1) { while (client->write_buffer_len + ipc_header_size + payload_length >=
sway_log_errno(L_INFO, "Unable to send header to IPC client"); client->write_buffer_size) {
client->write_buffer_size *= 2;
}
if (client->write_buffer_size > (1 << 22)) { // 4 MB
sway_log(L_ERROR, "Client write buffer too big, disconnecting client");
ipc_client_disconnect(client);
return false; return false;
} }
if (write(client->fd, payload, payload_length) == -1) { char *new_buffer = realloc(client->write_buffer, client->write_buffer_size);
sway_log_errno(L_INFO, "Unable to send payload to IPC client"); if (!new_buffer) {
sway_log(L_ERROR, "Unable to reallocate ipc client write buffer");
ipc_client_disconnect(client);
return false; return false;
} }
client->write_buffer = new_buffer;
sway_log(L_DEBUG, "Send IPC reply: %s", payload); memcpy(client->write_buffer + client->write_buffer_len, data, ipc_header_size);
client->write_buffer_len += ipc_header_size;
memcpy(client->write_buffer + client->write_buffer_len, payload, payload_length);
client->write_buffer_len += payload_length;
if (!client->writable_event_source) {
client->writable_event_source = wlc_event_loop_add_fd(client->fd, WLC_EVENT_WRITABLE, ipc_client_handle_writable, client);
}
sway_log(L_DEBUG, "Added IPC reply to client %d queue: %s", client->fd, payload);
return true; return true;
} }