diff options
Diffstat (limited to 'server')
| -rw-r--r-- | server/Makefile.am | 2 | ||||
| -rw-r--r-- | server/dispatcher.c | 248 | ||||
| -rw-r--r-- | server/dispatcher.h | 81 | ||||
| -rw-r--r-- | server/main_dispatcher.c | 102 |
4 files changed, 358 insertions, 75 deletions
diff --git a/server/Makefile.am b/server/Makefile.am index 418d707c..34a6b47b 100644 --- a/server/Makefile.am +++ b/server/Makefile.am @@ -78,6 +78,8 @@ libspice_server_la_SOURCES = \ red_client_cache.h \ red_client_shared_cache.h \ red_common.h \ + dispatcher.c \ + dispatcher.h \ red_dispatcher.c \ red_dispatcher.h \ main_dispatcher.c \ diff --git a/server/dispatcher.c b/server/dispatcher.c new file mode 100644 index 00000000..1cd9e43a --- /dev/null +++ b/server/dispatcher.c @@ -0,0 +1,248 @@ +#include <unistd.h> +#include <errno.h> +#include <assert.h> +#include <string.h> +#include <pthread.h> +#include <fcntl.h> +#include <poll.h> + +#include "mem.h" +#include "spice_common.h" +#include "dispatcher.h" + +#define DISPATCHER_DEBUG_PRINTF(level, ...) \ + red_printf_debug(level, "DISP", ##__VA_ARGS__) + +//#define DEBUG_DISPATCHER + +#ifdef DEBUG_DISPATCHER +#include <signal.h> +#endif + +#define ACK 0xffffffff + +/* + * read_safe + * helper. reads until size bytes accumulated in buf, if an error other then + * EINTR is encountered returns -1, otherwise returns 0. + * @block if 1 the read will block (the fd is always blocking). + * if 0 poll first, return immediately if no bytes available, otherwise + * read size in blocking mode. + */ +static int read_safe(int fd, void *buf, size_t size, int block) +{ + int read_size = 0; + int ret; + struct pollfd pollfd = {.fd = fd, .events = POLLIN, .revents = 0}; + + if (size == 0) { + return 0; + } + + if (!block) { + while ((ret = poll(&pollfd, 1, 0)) == -1) { + if (errno == EINTR) { + DISPATCHER_DEBUG_PRINTF(3, "EINTR in poll"); + continue; + } + red_error("poll failed"); + return -1; + } + if (!(pollfd.revents & POLLIN)) { + return 0; + } + } + while (read_size < size) { + ret = read(fd, buf + read_size, size - read_size); + if (ret == -1) { + if (errno == EINTR) { + DISPATCHER_DEBUG_PRINTF(3, "EINTR in read"); + continue; + } + return -1; + } + if (ret == 0) { + red_error("broken pipe on read"); + return -1; + } + read_size += ret; + } + return read_size; +} + +/* + * write_safe + * @return -1 for error, otherwise number of written bytes. may be zero. + */ +static int write_safe(int fd, void *buf, size_t size) +{ + int written_size = 0; + int ret; + + while (written_size < size) { + ret = write(fd, buf + written_size, size - written_size); + if (ret == -1) { + if (errno != EINTR) { + DISPATCHER_DEBUG_PRINTF(3, "EINTR in write\n"); + return -1; + } + continue; + } + written_size += ret; + } + return written_size; +} + +static int dispatcher_handle_single_read(Dispatcher *dispatcher) +{ + int ret; + uint32_t type; + DispatcherMessage *msg = NULL; + uint8_t *payload = dispatcher->payload; + uint32_t ack = ACK; + + if ((ret = read_safe(dispatcher->recv_fd, &type, sizeof(type), 0)) == -1) { + red_printf("error reading from dispatcher: %d\n", errno); + return 0; + } + if (ret == 0) { + /* no messsage */ + return 0; + } + msg = &dispatcher->messages[type]; + if (read_safe(dispatcher->recv_fd, payload, msg->size, 1) == -1) { + red_printf("error reading from dispatcher: %d\n", errno); + /* TODO: close socketpair? */ + return 0; + } + if (msg->handler) { + msg->handler(dispatcher->opaque, (void *)payload); + } else { + red_printf("error: no handler for message type %d\n", type); + } + if (msg->ack && write_safe(dispatcher->recv_fd, + &ack, sizeof(ack)) == -1) { + red_printf("error writing ack for message %d\n", type); + /* TODO: close socketpair? */ + } + return 1; +} + +/* + * dispatcher_handle_recv_read + * doesn't handle being in the middle of a message. all reads are blocking. + */ +void dispatcher_handle_recv_read(Dispatcher *dispatcher) +{ + while (dispatcher_handle_single_read(dispatcher)) { + } +} + +void dispatcher_send_message(Dispatcher *dispatcher, uint32_t message_type, + void *payload) +{ + DispatcherMessage *msg; + uint32_t ack; + int send_fd = dispatcher->send_fd; + + assert(dispatcher->max_message_type > message_type); + assert(dispatcher->messages[message_type].handler); + msg = &dispatcher->messages[message_type]; + pthread_mutex_lock(&dispatcher->lock); + if (write_safe(send_fd, &message_type, sizeof(message_type)) == -1) { + red_printf("error: failed to send message type for message %d\n", + message_type); + goto unlock; + } + if (write_safe(send_fd, payload, msg->size) == -1) { + red_printf("error: failed to send message body for message %d\n", + message_type); + goto unlock; + } + if (msg->ack) { + if (read_safe(send_fd, &ack, sizeof(ack), 1) == -1) { + red_printf("error: failed to read ack"); + } else if (ack != ACK) { + red_printf("error: got wrong ack value in dispatcher " + "for message %d\n", message_type); + /* TODO handling error? */ + } + } +unlock: + pthread_mutex_unlock(&dispatcher->lock); +} + +void dispatcher_register_handler(Dispatcher *dispatcher, uint32_t message_type, + dispatcher_handle_message handler, size_t size, + int ack) +{ + DispatcherMessage *msg; + + assert(message_type < dispatcher->max_message_type); + assert(dispatcher->messages[message_type].handler == 0); + msg = &dispatcher->messages[message_type]; + msg->handler = handler; + msg->size = size; + msg->ack = ack; + if (msg->size > dispatcher->payload_size) { + dispatcher->payload = realloc(dispatcher->payload, msg->size); + dispatcher->payload_size = msg->size; + } +} + +#ifdef DEBUG_DISPATCHER +static void dummy_handler(int bla) +{ +} + +static void setup_dummy_signal_handler(void) +{ + static int inited = 0; + struct sigaction act = { + .sa_handler = &dummy_handler, + }; + if (inited) { + return; + } + inited = 1; + /* handle SIGRTMIN+10 in order to test the loops for EINTR */ + if (sigaction(SIGRTMIN + 10, &act, NULL) == -1) { + fprintf(stderr, + "failed to set dummy sigaction for DEBUG_DISPATCHER\n"); + exit(1); + } +} +#endif + +void dispatcher_init(Dispatcher *dispatcher, size_t max_message_type, + void *opaque) +{ + int channels[2]; + +#ifdef DEBUG_DISPATCHER + setup_dummy_signal_handler(); +#endif + dispatcher->opaque = opaque; + if (socketpair(AF_LOCAL, SOCK_STREAM, 0, channels) == -1) { + red_error("socketpair failed %s", strerror(errno)); + return; + } + pthread_mutex_init(&dispatcher->lock, NULL); + dispatcher->recv_fd = channels[0]; + dispatcher->send_fd = channels[1]; + dispatcher->self = pthread_self(); + + dispatcher->messages = spice_malloc0_n(max_message_type, + sizeof(dispatcher->messages[0])); + dispatcher->max_message_type = max_message_type; +} + +void dispatcher_set_opaque(Dispatcher *dispatcher, void *opaque) +{ + dispatcher->opaque = opaque; +} + +int dispatcher_get_recv_fd(Dispatcher *dispatcher) +{ + return dispatcher->recv_fd; +} diff --git a/server/dispatcher.h b/server/dispatcher.h new file mode 100644 index 00000000..04e6b467 --- /dev/null +++ b/server/dispatcher.h @@ -0,0 +1,81 @@ +#ifndef MAIN_DISPATCHER_H +#define MAIN_DISPATCHER_H + +#include <spice.h> + +typedef struct Dispatcher Dispatcher; + +typedef void (*dispatcher_handle_message)(void *opaque, + void *payload); + +typedef struct DispatcherMessage { + size_t size; + int ack; + dispatcher_handle_message handler; +} DispatcherMessage; + +struct Dispatcher { + SpiceCoreInterface *recv_core; + int recv_fd; + int send_fd; + pthread_t self; + pthread_mutex_t lock; + DispatcherMessage *messages; + int stage; /* message parser stage - sender has no stages */ + size_t max_message_type; + void *payload; /* allocated as max of message sizes */ + size_t payload_size; /* used to track realloc calls */ + void *opaque; +}; + +/* + * dispatcher_send_message + * @message_type: message type + * @payload: payload + */ +void dispatcher_send_message(Dispatcher *dispatcher, uint32_t message_type, + void *payload); + +/* + * dispatcher_init + * @max_message_type: number of message types. Allows upfront allocation + * of a DispatcherMessage list. + * up front, and registration in any order wanted. + */ +void dispatcher_init(Dispatcher *dispatcher, size_t max_message_type, + void *opaque); + +/* + * dispatcher_register_handler + * @dispatcher: dispatcher + * @messsage_type: message type + * @handler: message handler + * @size: message size. Each type has a fixed associated size. + * @ack: send an ack. This is per message type - you can't send the + * same message type with and without. Register two different + * messages if that is what you want. + */ +void dispatcher_register_handler(Dispatcher *dispatcher, uint32_t message_type, + dispatcher_handle_message handler, size_t size, + int ack); + +/* + * dispatcher_handle_recv_read + * @dispatcher: Dispatcher instance + */ +void dispatcher_handle_recv_read(Dispatcher *); + +/* + * dispatcher_get_recv_fd + * @return: receive file descriptor of the dispatcher + */ +int dispatcher_get_recv_fd(Dispatcher *); + +/* + * dispatcher_set_opaque + * @dispatcher: Dispatcher instance + * @opaque: opaque to use for callbacks + */ +void dispatcher_set_opaque(Dispatcher *dispatcher, void *opaque); + +#endif //MAIN_DISPATCHER_H diff --git a/server/main_dispatcher.c b/server/main_dispatcher.c index 73856bfe..a5967fa4 100644 --- a/server/main_dispatcher.c +++ b/server/main_dispatcher.c @@ -5,6 +5,7 @@ #include <assert.h> #include "red_common.h" +#include "dispatcher.h" #include "main_dispatcher.h" /* @@ -28,11 +29,8 @@ */ typedef struct { + Dispatcher base; SpiceCoreInterface *core; - int main_fd; - int other_fd; - pthread_t self; - pthread_mutex_t lock; } MainDispatcher; MainDispatcher main_dispatcher; @@ -43,15 +41,10 @@ enum { MAIN_DISPATCHER_NUM_MESSAGES }; -typedef struct MainDispatcherMessage { - uint32_t type; - union { - struct { - int event; - SpiceChannelEventInfo *info; - } channel_event; - } data; -} MainDispatcherMessage; +typedef struct MainDispatcherChannelEventMessage { + int event; + SpiceChannelEventInfo *info; +} MainDispatcherChannelEventMessage; /* channel_event - calls core->channel_event, must be done in main thread */ static void main_dispatcher_self_handle_channel_event( @@ -61,85 +54,44 @@ static void main_dispatcher_self_handle_channel_event( main_dispatcher.core->channel_event(event, info); } -static void main_dispatcher_handle_channel_event(MainDispatcherMessage *msg) +static void main_dispatcher_handle_channel_event(void *opaque, + void *payload) { - main_dispatcher_self_handle_channel_event(msg->data.channel_event.event, - msg->data.channel_event.info); + MainDispatcherChannelEventMessage *channel_event = payload; + + main_dispatcher_self_handle_channel_event(channel_event->event, + channel_event->info); } void main_dispatcher_channel_event(int event, SpiceChannelEventInfo *info) { - MainDispatcherMessage msg; - ssize_t written = 0; - ssize_t ret; + MainDispatcherChannelEventMessage msg; - if (pthread_self() == main_dispatcher.self) { + if (pthread_self() == main_dispatcher.base.self) { main_dispatcher_self_handle_channel_event(event, info); return; } - msg.type = MAIN_DISPATCHER_CHANNEL_EVENT; - msg.data.channel_event.event = event; - msg.data.channel_event.info = info; - pthread_mutex_lock(&main_dispatcher.lock); - while (written < sizeof(msg)) { - ret = write(main_dispatcher.other_fd, &msg + written, - sizeof(msg) - written); - if (ret == -1) { - assert(errno == EINTR); - continue; - } - written += ret; - } - pthread_mutex_unlock(&main_dispatcher.lock); + msg.event = event; + msg.info = info; + dispatcher_send_message(&main_dispatcher.base, MAIN_DISPATCHER_CHANNEL_EVENT, + &msg); } - -static void main_dispatcher_handle_read(int fd, int event, void *opaque) +static void dispatcher_handle_read(int fd, int event, void *opaque) { - int ret; - MainDispatcher *md = opaque; - MainDispatcherMessage msg; - int read_size = 0; + Dispatcher *dispatcher = opaque; - while (read_size < sizeof(msg)) { - /* blocks until sizeof(msg) is read */ - ret = read(md->main_fd, &msg + read_size, sizeof(msg) - read_size); - if (ret == -1) { - if (errno != EINTR) { - red_printf("error reading from main dispatcher: %d\n", errno); - /* TODO: close channel? */ - return; - } - continue; - } - read_size += ret; - } - switch (msg.type) { - case MAIN_DISPATCHER_CHANNEL_EVENT: - main_dispatcher_handle_channel_event(&msg); - break; - default: - red_printf("error: unhandled main dispatcher message type %d\n", - msg.type); - } + dispatcher_handle_recv_read(dispatcher); } void main_dispatcher_init(SpiceCoreInterface *core) { - int channels[2]; - memset(&main_dispatcher, 0, sizeof(main_dispatcher)); main_dispatcher.core = core; - - if (socketpair(AF_LOCAL, SOCK_STREAM, 0, channels) == -1) { - red_error("socketpair failed %s", strerror(errno)); - return; - } - pthread_mutex_init(&main_dispatcher.lock, NULL); - main_dispatcher.main_fd = channels[0]; - main_dispatcher.other_fd = channels[1]; - main_dispatcher.self = pthread_self(); - - core->watch_add(main_dispatcher.main_fd, SPICE_WATCH_EVENT_READ, - main_dispatcher_handle_read, &main_dispatcher); + dispatcher_init(&main_dispatcher.base, MAIN_DISPATCHER_NUM_MESSAGES, &main_dispatcher.base); + core->watch_add(main_dispatcher.base.recv_fd, SPICE_WATCH_EVENT_READ, + dispatcher_handle_read, &main_dispatcher.base); + dispatcher_register_handler(&main_dispatcher.base, MAIN_DISPATCHER_CHANNEL_EVENT, + main_dispatcher_handle_channel_event, + sizeof(MainDispatcherChannelEventMessage), 0 /* no ack */); } |