From 80caf07e09efe14c67f89a3c01501a6a39681167 Mon Sep 17 00:00:00 2001 From: Alon Levy Date: Thu, 8 Sep 2011 02:16:24 +0300 Subject: [0.8 branch] server: add main_dispatcher add main_dispatcher, a message passing mechanism for sending messages to the main thread. The main thread is the thread that implements SpiceCoreInterface, which is assumed to be a single thread. Similar to the async operation of red_worker, a socket pair is created and used to pass messages. The messages are a fixed size to ease parsing. A single message is defined to pass a channel_event. RHBZ: 746950 FDBZ: 41858 This patch is 0.8 branch only, for the master branch there should be a better approach to share code with red_dispatcher and ready the way for later adding more threads. --- server/Makefile.am | 2 + server/main_dispatcher.c | 145 +++++++++++++++++++++++++++++++++++++++++++++++ server/main_dispatcher.h | 9 +++ server/reds.c | 4 +- 4 files changed, 159 insertions(+), 1 deletion(-) create mode 100644 server/main_dispatcher.c create mode 100644 server/main_dispatcher.h diff --git a/server/Makefile.am b/server/Makefile.am index 93ed3120..ebb0d3f2 100644 --- a/server/Makefile.am +++ b/server/Makefile.am @@ -114,6 +114,8 @@ libspice_server_la_SOURCES = \ red_common.h \ red_dispatcher.c \ red_dispatcher.h \ + main_dispatcher.c \ + main_dispatcher.h \ red_memslots.c \ red_memslots.h \ red_parse_qxl.c \ diff --git a/server/main_dispatcher.c b/server/main_dispatcher.c new file mode 100644 index 00000000..73856bfe --- /dev/null +++ b/server/main_dispatcher.c @@ -0,0 +1,145 @@ +#include +#include +#include +#include +#include + +#include "red_common.h" +#include "main_dispatcher.h" + +/* + * Main Dispatcher + * =============== + * + * Communication channel between any non main thread and the main thread. + * + * The main thread is that from which spice_server_init is called. + * + * Messages are single sized, sent from the non-main thread to the main-thread. + * No acknowledge is sent back. This prevents a possible deadlock with the main + * thread already waiting on a response for the existing red_dispatcher used + * by the worker thread. + * + * All events have three functions: + * main_dispatcher_ - non static, public function + * main_dispatcher_self_ - handler for main thread + * main_dispatcher_handle_ - handler for callback from main thread + * seperate from self because it may send an ack or do other work in the future. + */ + +typedef struct { + SpiceCoreInterface *core; + int main_fd; + int other_fd; + pthread_t self; + pthread_mutex_t lock; +} MainDispatcher; + +MainDispatcher main_dispatcher; + +enum { + MAIN_DISPATCHER_CHANNEL_EVENT = 0, + + MAIN_DISPATCHER_NUM_MESSAGES +}; + +typedef struct MainDispatcherMessage { + uint32_t type; + union { + struct { + int event; + SpiceChannelEventInfo *info; + } channel_event; + } data; +} MainDispatcherMessage; + +/* channel_event - calls core->channel_event, must be done in main thread */ +static void main_dispatcher_self_handle_channel_event( + int event, + SpiceChannelEventInfo *info) +{ + main_dispatcher.core->channel_event(event, info); +} + +static void main_dispatcher_handle_channel_event(MainDispatcherMessage *msg) +{ + main_dispatcher_self_handle_channel_event(msg->data.channel_event.event, + msg->data.channel_event.info); +} + +void main_dispatcher_channel_event(int event, SpiceChannelEventInfo *info) +{ + MainDispatcherMessage msg; + ssize_t written = 0; + ssize_t ret; + + if (pthread_self() == main_dispatcher.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); +} + + +static void main_dispatcher_handle_read(int fd, int event, void *opaque) +{ + int ret; + MainDispatcher *md = opaque; + MainDispatcherMessage msg; + int read_size = 0; + + 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); + } +} + +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); +} diff --git a/server/main_dispatcher.h b/server/main_dispatcher.h new file mode 100644 index 00000000..2c201c75 --- /dev/null +++ b/server/main_dispatcher.h @@ -0,0 +1,9 @@ +#ifndef MAIN_DISPATCHER_H +#define MAIN_DISPATCHER_H + +#include + +void main_dispatcher_channel_event(int event, SpiceChannelEventInfo *info); +void main_dispatcher_init(SpiceCoreInterface *core); + +#endif //MAIN_DISPATCHER_H diff --git a/server/reds.c b/server/reds.c index 8e83b990..0bb7e962 100644 --- a/server/reds.c +++ b/server/reds.c @@ -53,6 +53,7 @@ #include "red_common.h" #include "red_dispatcher.h" +#include "main_dispatcher.h" #include "snd_worker.h" #include #include "stat.h" @@ -415,7 +416,7 @@ static void reds_channel_event(RedsStream *stream, int event) { if (core->base.minor_version < 3 || core->channel_event == NULL) return; - core->channel_event(event, &stream->info); + main_dispatcher_channel_event(event, &stream->info); } static ssize_t stream_write_cb(RedsStream *s, const void *buf, size_t size) @@ -4685,6 +4686,7 @@ static int do_spice_init(SpiceCoreInterface *core_interface) reds->outgoing.vec = reds->outgoing.vec_buf; init_vd_agent_resources(); + main_dispatcher_init(core); if (!(reds->mig_timer = core->timer_add(migrate_timeout, NULL))) { red_error("migration timer create failed"); -- cgit