From 933aee1d5c53b0cc7d608011a29188b594c8d70b Mon Sep 17 00:00:00 2001 From: Jamey Sharp Date: Fri, 16 Apr 2010 20:18:28 -0700 Subject: Fix Xlib/XCB for multi-threaded applications (with caveats). Rather than trying to group all response processing in one monolithic process_responses function, let _XEventsQueued, _XReadEvents, and _XReply each do their own thing with a minimum of code that can all be reasoned about independently. Tested with `ico -threads 20`, which seems to be able to make many icosahedrons dance at once quite nicely now. Caveats: - Anything that was not thread-safe in Xlib before XCB probably still isn't. XListFontsWithInfo, for instance. - If one thread is waiting for events and another thread tries to read a reply, both will hang until an event arrives. Previously, if this happened it might work sometimes, but otherwise would trigger either an assertion failure or a permanent hang. - Versions of libxcb up to and including 1.6 have a bug that can cause xcb_wait_for_event or xcb_wait_for_reply to hang if they run concurrently with xcb_writev or other writers. So you'll want that fix as well. Signed-off-by: Jamey Sharp Reviewed-by: Josh Triplett --- src/Xxcbint.h | 6 +- src/xcb_disp.c | 5 +- src/xcb_io.c | 344 +++++++++++++++++++++++++++++++++++++-------------------- 3 files changed, 233 insertions(+), 122 deletions(-) diff --git a/src/Xxcbint.h b/src/Xxcbint.h index 8b6a3619..c8191620 100644 --- a/src/Xxcbint.h +++ b/src/Xxcbint.h @@ -16,6 +16,7 @@ typedef struct PendingRequest PendingRequest; struct PendingRequest { PendingRequest *next; unsigned long sequence; + unsigned reply_waiter; }; typedef struct _X11XCBPrivate { @@ -31,11 +32,10 @@ typedef struct _X11XCBPrivate { enum XEventQueueOwner event_owner; XID next_xid; - /* handle simultaneous threads waiting for events, - * used in wait_or_poll_for_event - */ + /* handle simultaneous threads waiting for responses */ xcondition_t event_notify; int event_waiter; + xcondition_t reply_notify; } _X11XCBPrivate; /* xcb_disp.c */ diff --git a/src/xcb_disp.c b/src/xcb_disp.c index 622afe76..fea1326c 100644 --- a/src/xcb_disp.c +++ b/src/xcb_disp.c @@ -98,9 +98,11 @@ int _XConnectXCB(Display *dpy, _Xconst char *display, char **fullnamep, int *scr dpy->xcb->next_xid = xcb_generate_id(dpy->xcb->connection); dpy->xcb->event_notify = xcondition_malloc(); - if (!dpy->xcb->event_notify) + dpy->xcb->reply_notify = xcondition_malloc(); + if (!dpy->xcb->event_notify || !dpy->xcb->reply_notify) return 0; xcondition_init(dpy->xcb->event_notify); + xcondition_init(dpy->xcb->reply_notify); return !xcb_connection_has_error(c); } @@ -115,5 +117,6 @@ void _XFreeX11XCBStructure(Display *dpy) free(tmp); } xcondition_free(dpy->xcb->event_notify); + xcondition_free(dpy->xcb->reply_notify); Xfree(dpy->xcb); } diff --git a/src/xcb_io.c b/src/xcb_io.c index 19dc6a97..dac7622c 100644 --- a/src/xcb_io.c +++ b/src/xcb_io.c @@ -122,6 +122,7 @@ static PendingRequest *append_pending_request(Display *dpy, unsigned long sequen assert(node); node->next = NULL; node->sequence = sequence; + node->reply_waiter = 0; if(dpy->xcb->pending_requests_tail) { assert(XLIB_SEQUENCE_COMPARE(dpy->xcb->pending_requests_tail->sequence, <, node->sequence)); @@ -166,43 +167,6 @@ static int handle_error(Display *dpy, xError *err, Bool in_XReply) return 0; } -static void call_handlers(Display *dpy, xcb_generic_reply_t *buf) -{ - _XAsyncHandler *async, *next; - for(async = dpy->async_handlers; async; async = next) - { - next = async->next; - if(async->handler(dpy, (xReply *) buf, (char *) buf, sizeof(xReply) + (buf->length << 2), async->data)) - return; - } -} - -static xcb_generic_event_t * wait_or_poll_for_event(Display *dpy, int wait) -{ - xcb_connection_t *c = dpy->xcb->connection; - xcb_generic_event_t *event; - if(wait) - { - if(dpy->xcb->event_waiter) - { - ConditionWait(dpy, dpy->xcb->event_notify); - event = xcb_poll_for_event(c); - } - else - { - dpy->xcb->event_waiter = 1; - UnlockDisplay(dpy); - event = xcb_wait_for_event(c); - InternalLockDisplay(dpy, /* don't skip user locks */ 0); - dpy->xcb->event_waiter = 0; - ConditionBroadcast(dpy, dpy->xcb->event_notify); - } - } - else - event = xcb_poll_for_event(c); - return event; -} - /* Widen a 32-bit sequence number into a native-word-size (unsigned long) * sequence number. Treating the comparison as a 1 and shifting it avoids a * conditional branch, and shifting by 16 twice avoids a compiler warning when @@ -213,93 +177,114 @@ static void widen(unsigned long *wide, unsigned int narrow) *wide = new + ((unsigned long) (new < *wide) << 16 << 16); } -static void process_responses(Display *dpy, int wait_for_first_event, xcb_generic_error_t **current_error, unsigned long current_request) -{ - void *reply; - xcb_generic_event_t *event = dpy->xcb->next_event; - xcb_generic_error_t *error; - xcb_connection_t *c = dpy->xcb->connection; - if(!event && dpy->xcb->event_owner == XlibOwnsEventQueue) - event = wait_or_poll_for_event(dpy, wait_for_first_event); +/* Thread-safety rules: + * + * At most one thread can be reading from XCB's event queue at a time. + * If you are not the current event-reading thread and you need to find + * out if an event is available, you must wait. + * + * The same rule applies for reading replies. + * + * A single thread cannot be both the the event-reading and the + * reply-reading thread at the same time. + * + * We always look at both the current event and the first pending reply + * to decide which to process next. + * + * We always process all responses in sequence-number order, which may + * mean waiting for another thread (either the event_waiter or the + * reply_waiter) to handle an earlier response before we can process or + * return a later one. If so, we wait on the corresponding condition + * variable for that thread to process the response and wake us up. + */ +static xcb_generic_reply_t *poll_for_event(Display *dpy) +{ + /* Make sure the Display's sequence numbers are valid */ require_socket(dpy); - while(1) + /* Precondition: This thread can safely get events from XCB. */ + assert(dpy->xcb->event_owner == XlibOwnsEventQueue && !dpy->xcb->event_waiter); + + if(!dpy->xcb->next_event) + dpy->xcb->next_event = xcb_poll_for_event(dpy->xcb->connection); + + if(dpy->xcb->next_event) { PendingRequest *req = dpy->xcb->pending_requests; + xcb_generic_event_t *event = dpy->xcb->next_event; unsigned long event_sequence = dpy->last_request_read; - if(event) - widen(&event_sequence, event->full_sequence); - assert(!(req && current_request && !XLIB_SEQUENCE_COMPARE(req->sequence, <=, current_request))); - if(event && (!req || XLIB_SEQUENCE_COMPARE(event_sequence, <=, req->sequence))) + widen(&event_sequence, event->full_sequence); + if(!req || XLIB_SEQUENCE_COMPARE(event_sequence, <, req->sequence) + || (event->response_type != X_Error && event_sequence == req->sequence)) { + assert(XLIB_SEQUENCE_COMPARE(event_sequence, <=, dpy->request)); dpy->last_request_read = event_sequence; - if(event->response_type != X_Error) - { - /* GenericEvents may be > 32 bytes. In this - * case, the event struct is trailed by the - * additional bytes. the xcb_generic_event_t - * struct uses 4 bytes for internal numbering, - * so we need to shift the trailing data to be - * after the first 32 bytes. */ - if (event->response_type == GenericEvent && - ((xcb_ge_event_t*)event)->length) - { - memmove(&event->full_sequence, - &event[1], - ((xcb_ge_event_t*)event)->length * 4); - } - _XEnq(dpy, (xEvent *) event); - wait_for_first_event = 0; - } - else if(current_error && event_sequence == current_request) - { - /* This can only occur when called from - * _XReply, which doesn't need a new event. */ - *current_error = (xcb_generic_error_t *) event; - event = NULL; - break; - } - else - handle_error(dpy, (xError *) event, current_error != NULL); - free(event); - event = wait_or_poll_for_event(dpy, wait_for_first_event); + dpy->xcb->next_event = NULL; + return (xcb_generic_reply_t *) event; } - else if(req && req->sequence == current_request) + } + return NULL; +} + +static xcb_generic_reply_t *poll_for_response(Display *dpy) +{ + void *response; + xcb_generic_error_t *error; + PendingRequest *req; + while(!(response = poll_for_event(dpy)) && + (req = dpy->xcb->pending_requests) && + !req->reply_waiter && + xcb_poll_for_reply(dpy->xcb->connection, req->sequence, &response, &error)) + { + assert(XLIB_SEQUENCE_COMPARE(req->sequence, <=, dpy->request)); + dpy->last_request_read = req->sequence; + if(!response) + dequeue_pending_request(dpy, req); + if(error) + return (xcb_generic_reply_t *) error; + } + return response; +} + +static void handle_response(Display *dpy, xcb_generic_reply_t *response, Bool in_XReply) +{ + _XAsyncHandler *async, *next; + switch(response->response_type) + { + case X_Reply: + for(async = dpy->async_handlers; async; async = next) { - break; + next = async->next; + if(async->handler(dpy, (xReply *) response, (char *) response, sizeof(xReply) + (response->length << 2), async->data)) + break; } - else if(req && xcb_poll_for_reply(dpy->xcb->connection, req->sequence, &reply, &error)) + break; + + case X_Error: + handle_error(dpy, (xError *) response, in_XReply); + break; + + default: /* event */ + /* GenericEvents may be > 32 bytes. In this case, the + * event struct is trailed by the additional bytes. the + * xcb_generic_event_t struct uses 4 bytes for internal + * numbering, so we need to shift the trailing data to + * be after the first 32 bytes. */ + if(response->response_type == GenericEvent && ((xcb_ge_event_t *) response)->length) { - if(reply || error) - dpy->last_request_read = req->sequence; - if(reply) - { - call_handlers(dpy, reply); - free(reply); - } - if(error) - { - handle_error(dpy, (xError *) error, current_error != NULL); - free(error); - } - if(!reply) - dequeue_pending_request(dpy, req); + xcb_ge_event_t *event = (xcb_ge_event_t *) response; + memmove(&event->full_sequence, &event[1], event->length * 4); } - else - break; + _XEnq(dpy, (xEvent *) response); + break; } - - dpy->xcb->next_event = event; - - if(xcb_connection_has_error(c)) - _XIOError(dpy); - - assert(XLIB_SEQUENCE_COMPARE(dpy->last_request_read, <=, dpy->request)); + free(response); } int _XEventsQueued(Display *dpy, int mode) { + xcb_generic_reply_t *response; if(dpy->flags & XlibDisplayIOError) return 0; if(dpy->xcb->event_owner != XlibOwnsEventQueue) @@ -309,7 +294,17 @@ int _XEventsQueued(Display *dpy, int mode) _XSend(dpy, NULL, 0); else check_internal_connections(dpy); - process_responses(dpy, 0, NULL, 0); + + /* If another thread is blocked waiting for events, then we must + * let that thread pick up the next event. Since it blocked, we + * can reasonably claim there are no new events right now. */ + if(!dpy->xcb->event_waiter) + { + while((response = poll_for_response(dpy))) + handle_response(dpy, response, False); + if(xcb_connection_has_error(dpy->xcb->connection)) + _XIOError(dpy); + } return dpy->qlen; } @@ -318,15 +313,66 @@ int _XEventsQueued(Display *dpy, int mode) */ void _XReadEvents(Display *dpy) { + xcb_generic_reply_t *response; + unsigned long serial; + if(dpy->flags & XlibDisplayIOError) return; _XSend(dpy, NULL, 0); if(dpy->xcb->event_owner != XlibOwnsEventQueue) return; check_internal_connections(dpy); - do { - process_responses(dpy, 1, NULL, 0); - } while (dpy->qlen == 0); + + serial = dpy->next_event_serial_num; + while(serial == dpy->next_event_serial_num || dpy->qlen == 0) + { + if(dpy->xcb->event_waiter) + { + ConditionWait(dpy, dpy->xcb->event_notify); + /* Maybe the other thread got us an event. */ + continue; + } + + if(!dpy->xcb->next_event) + { + xcb_generic_event_t *event; + dpy->xcb->event_waiter = 1; + UnlockDisplay(dpy); + event = xcb_wait_for_event(dpy->xcb->connection); + InternalLockDisplay(dpy, /* don't skip user locks */ 0); + dpy->xcb->event_waiter = 0; + ConditionBroadcast(dpy, dpy->xcb->event_notify); + if(!event) + _XIOError(dpy); + dpy->xcb->next_event = event; + } + + /* We've established most of the conditions for + * poll_for_response to return non-NULL. The exceptions + * are connection shutdown, and finding that another + * thread is waiting for the next reply we'd like to + * process. */ + + response = poll_for_response(dpy); + if(response) + handle_response(dpy, response, False); + else if(dpy->xcb->pending_requests->reply_waiter) + { /* need braces around ConditionWait */ + ConditionWait(dpy, dpy->xcb->reply_notify); + } + else + _XIOError(dpy); + } + + /* The preceding loop established that there is no + * event_waiter--unless we just called ConditionWait because of + * a reply_waiter, in which case another thread may have become + * the event_waiter while we slept unlocked. */ + if(!dpy->xcb->event_waiter) + while((response = poll_for_response(dpy))) + handle_response(dpy, response, False); + if(xcb_connection_has_error(dpy->xcb->connection)) + _XIOError(dpy); } /* @@ -467,19 +513,83 @@ Status _XReply(Display *dpy, xReply *rep, int extra, Bool discard) current = dpy->xcb->pending_requests_tail; else current = append_pending_request(dpy, dpy->request); - /* FIXME: drop the Display lock while waiting? - * Complicates process_responses. */ - reply = xcb_wait_for_reply(c, current->sequence, &error); + /* Don't let any other thread get this reply. */ + current->reply_waiter = 1; + + while(1) + { + PendingRequest *req = dpy->xcb->pending_requests; + xcb_generic_reply_t *response; + + if(req != current && req->reply_waiter) + { + ConditionWait(dpy, dpy->xcb->reply_notify); + /* Another thread got this reply. */ + continue; + } + req->reply_waiter = 1; + UnlockDisplay(dpy); + response = xcb_wait_for_reply(c, req->sequence, &error); + InternalLockDisplay(dpy, /* don't skip user locks */ 0); + + /* We have the response we're looking for. Now, before + * letting anyone else process this sequence number, we + * need to process any events that should have come + * earlier. */ + + if(dpy->xcb->event_owner == XlibOwnsEventQueue) + { + xcb_generic_reply_t *event; + /* If some thread is already waiting for events, + * it will get the first one. That thread must + * process that event before we can continue. */ + /* FIXME: That event might be after this reply, + * and might never even come--or there might be + * multiple threads trying to get events. */ + while(dpy->xcb->event_waiter) + { /* need braces around ConditionWait */ + ConditionWait(dpy, dpy->xcb->event_notify); + } + while((event = poll_for_event(dpy))) + handle_response(dpy, event, True); + } + req->reply_waiter = 0; + ConditionBroadcast(dpy, dpy->xcb->reply_notify); + assert(XLIB_SEQUENCE_COMPARE(req->sequence, <=, dpy->request)); + dpy->last_request_read = req->sequence; + if(!response) + dequeue_pending_request(dpy, req); + + if(req == current) + { + reply = (char *) response; + break; + } + + if(error) + handle_response(dpy, (xcb_generic_reply_t *) error, True); + else if(response) + handle_response(dpy, response, True); + } check_internal_connections(dpy); - process_responses(dpy, 0, &error, current->sequence); + + if(dpy->xcb->next_event && dpy->xcb->next_event->response_type == X_Error) + { + xcb_generic_event_t *event = dpy->xcb->next_event; + unsigned long event_sequence = dpy->last_request_read; + widen(&event_sequence, event->full_sequence); + if(event_sequence == current->sequence) + { + error = (xcb_generic_error_t *) event; + dpy->xcb->next_event = NULL; + } + } if(error) { int ret_code; - dpy->last_request_read = error->full_sequence; - /* Xlib is evil and assumes that even errors will be * copied into rep. */ memcpy(rep, error, 32); @@ -522,8 +632,6 @@ Status _XReply(Display *dpy, xReply *rep, int extra, Bool discard) return 0; } - dpy->last_request_read = current->sequence; - /* there's no error and we have a reply. */ dpy->xcb->reply_data = reply; dpy->xcb->reply_consumed = sizeof(xReply) + (extra * 4); -- cgit v1.2.1