read-eventsacquire-fd

2 files changed, 69 insertions, 121 deletions

diff --git a/src/wayland-client.c b/src/wayland-client.c
index b94dfdc..5bfff2a 100644
--- a/src/wayland-client.c
+++ b/src/wayland-client.c
@@ -92,12 +92,9 @@ struct wl_display {
 	struct wl_event_queue queue;
 	struct wl_list event_queue_list;
 	pthread_mutex_t mutex;
-	pthread_t reader;
+
+	int reader_count;
 	pthread_cond_t reader_cond;
-	pthread_cond_t pipe_cond;
-	int reader_pipe[2];
-	int reader_state;
-	int pipe_busy;
 };
 
 /** \endcond */
@@ -538,9 +535,7 @@ wl_display_connect_to_fd(int fd)
 	wl_list_init(&display->event_queue_list);
 	pthread_mutex_init(&display->mutex, NULL);
 	pthread_cond_init(&display->reader_cond, NULL);
-	pthread_cond_init(&display->pipe_cond, NULL);
-	display->reader_state = LEGACY_READER;
-	display->reader = pthread_self();
+	display->reader_count = 0;
 
 	wl_map_insert_new(&display->objects, WL_MAP_CLIENT_SIDE, NULL);
 
@@ -559,17 +554,11 @@ wl_display_connect_to_fd(int fd)
 	if (display->connection == NULL)
 		goto err_connection;
 
-	if (pipe2(display->reader_pipe, O_CLOEXEC) == -1)
-		goto err_pipe;
-
 	return display;
 
- err_pipe:
-	wl_connection_destroy(display->connection);
  err_connection:
 	pthread_mutex_destroy(&display->mutex);
 	pthread_cond_destroy(&display->reader_cond);
-	pthread_cond_destroy(&display->pipe_cond);
 	wl_map_release(&display->objects);
 	close(display->fd);
 	free(display);
@@ -625,23 +614,9 @@ wl_display_connect_to_fd(int fd)
 WL_EXPORT int 
 wl_display_acquire_fd(struct wl_display *display)
 {
-	char c = 0;
-
 	pthread_mutex_lock(&display->mutex);
 
-	if (display->reader_state == LOCKED_READER &&
-	    !pthread_equal(display->reader, pthread_self())) {
-		errno = EBUSY;
-		pthread_mutex_unlock(&display->mutex);
-		return -1;
-	}
-
-	pthread_cond_signal(&display->pipe_cond);
-	if (display->reader_state == VOLUNTEER_READER)
-		write(display->reader_pipe[1], &c, 1);
-
-	display->reader = pthread_self();
-	display->reader_state = LOCKED_READER;
+	display->reader_count++;
 
 	pthread_mutex_unlock(&display->mutex);
 
@@ -663,14 +638,9 @@ wl_display_release_fd(struct wl_display *display)
 {
 	pthread_mutex_lock(&display->mutex);
 
-	if (display->reader_state != LOCKED_READER ||
-	    !pthread_equal(display->reader, pthread_self())) {
-		pthread_mutex_unlock(&display->mutex);
-		return;
-	}
-
-	display->reader_state = NO_READER;
-	pthread_cond_broadcast(&display->reader_cond);
+	display->reader_count--;
+	if (display->reader_count == 0)
+		pthread_cond_broadcast(&display->reader_cond);
 
 	pthread_mutex_unlock(&display->mutex);
 }
@@ -728,11 +698,8 @@ wl_display_disconnect(struct wl_display *display)
 	wl_event_queue_release(&display->queue);
 	pthread_mutex_destroy(&display->mutex);
 	pthread_cond_destroy(&display->reader_cond);
-	pthread_cond_destroy(&display->pipe_cond);
 	if (display->fd > 0)
 		close(display->fd);
-	close(display->reader_pipe[0]);
-	close(display->reader_pipe[1]);
 
 	free(display);
 }
@@ -983,52 +950,12 @@ dispatch_event(struct wl_display *display, struct wl_event_queue *queue)
 }
 
 static int
-read_events(struct wl_display *display, struct wl_event_queue *queue)
+read_events(struct wl_display *display)
 {
-	struct pollfd pfd[2];
-	int len, size, ret;
-	char c;
-
-	while (display->reader_state == NO_READER && display->pipe_busy)
-		pthread_cond_wait(&display->pipe_cond, &display->mutex);
-
-	if (display->reader_state == NO_READER) {
-		/* A thread gets here when it's the first reader to
-		 * try to read and there's no LOCKED_READER.  We set
-		 * VOLUNTEER_READER and poll on the fd and the
-		 * reader_pipe, so that a locked reader can preempt if
-		 * it needs to. */
-		display->reader = pthread_self();
-		display->reader_state = VOLUNTEER_READER;
-		display->pipe_busy = 1;
-
-		pthread_mutex_unlock(&display->mutex);
-
-		pfd[0].fd = display->fd;
-		pfd[0].events = POLLIN;
-		pfd[1].fd = display->reader_pipe[0];
-		pfd[1].events = POLLIN;
-		ret = poll(pfd, 2, -1);
-
-		pthread_mutex_lock(&display->mutex);
-
-		display->pipe_busy = 0;
-		if (display->reader_state != VOLUNTEER_READER) {
-			read(display->reader_pipe[0], &c, 1);
-			pthread_cond_broadcast(&display->pipe_cond);
-		}
-
-		if (ret == -1) {
-			display_fatal_error(display, errno);
-			return -1;
-		}
-	}
+	int len, size;
 
-	if (pthread_equal(display->reader, pthread_self())) {
-		/* At this point we have a reader.  Either a
-		 * VOLUNTEER_READER and the fd is readable or a
-		 * LOCKED_READER which is allowed to block in
-		 * wl_connection_read(). */
+	display->reader_count--;
+	if (display->reader_count == 0) {
 		len = wl_connection_read(display->connection);
 		if (len == -1) {
 			display_fatal_error(display, errno);
@@ -1048,55 +975,48 @@ read_events(struct wl_display *display, struct wl_event_queue *queue)
 			len -= size;
 		}
 
-		/* Indicate we're done by setting NO_READER and wake
-		 * up other threads waiting to read. */
-		if (display->reader_state != LEGACY_READER)
-			display->reader_state = NO_READER;
 		pthread_cond_broadcast(&display->reader_cond);
 	} else {
-		/* Some other thread signed up to read from the fd, so
-		 * just wait for the reader to do the work. */
-		pthread_cond_wait(&display->reader_cond, &display->mutex);
+		while (display->reader_count > 0)
+			pthread_cond_wait(&display->reader_cond,
+					  &display->mutex);
 	}
 
 	return 0;
 }
 
-static int
-dispatch_queue(struct wl_display *display,
-	       struct wl_event_queue *queue, int block)
+WL_EXPORT int
+wl_display_read_events(struct wl_display *display)
 {
-	int ret, count;
+	int ret;
 
 	pthread_mutex_lock(&display->mutex);
 
-	if (display->last_error)
-		goto err_unlock;
+	ret = read_events(display) == -1;
 
-	ret = wl_connection_flush(display->connection);
-	if (ret < 0 && errno != EAGAIN) {
-		display_fatal_error(display, errno);
-		goto err_unlock;
-	}
+	pthread_mutex_unlock(&display->mutex);
 
-	if (block && wl_list_empty(&queue->event_list)) {
-		if (read_events(display, queue) == -1)
-			goto err_unlock;
-	}
+	return ret;
+}
+
+static int
+dispatch_queue(struct wl_display *display, struct wl_event_queue *queue)
+{
+	int count;
+
+	if (display->last_error)
+		goto err;
 
 	for (count = 0; !wl_list_empty(&queue->event_list); count++) {
 		dispatch_event(display, queue);
 		if (display->last_error)
-			goto err_unlock;
+			goto err;
 	}
 
-	pthread_mutex_unlock(&display->mutex);
-
 	return count;
 
-err_unlock:
+err:
 	errno = display->last_error;
-	pthread_mutex_unlock(&display->mutex);
 
 	return -1;
 }
@@ -1121,7 +1041,29 @@ WL_EXPORT int
 wl_display_dispatch_queue(struct wl_display *display,
 			  struct wl_event_queue *queue)
 {
-	return dispatch_queue(display, queue, 1);
+	struct pollfd pfd[2];
+
+	wl_display_acquire_fd(display);
+
+	pfd[0].fd = display->fd;
+	pfd[0].events = POLLIN;
+	poll(pfd, 1, -1);
+
+	pthread_mutex_lock(&display->mutex);
+
+	if (read_events(display) == -1)
+		goto err_unlock;
+
+	if (dispatch_queue(display, queue) == -1)
+		goto err_unlock;
+
+	pthread_mutex_unlock(&display->mutex);
+
+	return 0;
+
+ err_unlock:
+	pthread_mutex_unlock(&display->mutex);
+	return -1;
 }
 
 /** Dispatch pending events in an event queue
@@ -1141,7 +1083,18 @@ WL_EXPORT int
 wl_display_dispatch_queue_pending(struct wl_display *display,
 				  struct wl_event_queue *queue)
 {
-	return dispatch_queue(display, queue, 0);
+	pthread_mutex_lock(&display->mutex);
+
+	if (dispatch_queue(display, queue) == -1)
+		goto err_unlock;
+
+	pthread_mutex_unlock(&display->mutex);
+
+	return 0;
+
+ err_unlock:
+	pthread_mutex_unlock(&display->mutex);
+	return -1;
 }
 
 /** Process incoming events
@@ -1170,10 +1123,7 @@ wl_display_dispatch_queue_pending(struct wl_display *display,
 WL_EXPORT int
 wl_display_dispatch(struct wl_display *display)
 {
-	if (display->reader_state == LEGACY_READER)
-		display->reader = pthread_self();
-
-	return dispatch_queue(display, &display->queue, 1);
+	return wl_display_dispatch_queue(display, &display->queue);
 }
 
 /** Dispatch main queue events without reading from the display fd
@@ -1217,10 +1167,7 @@ wl_display_dispatch(struct wl_display *display)
 WL_EXPORT int
 wl_display_dispatch_pending(struct wl_display *display)
 {
-	if (display->reader_state == LEGACY_READER)
-		display->reader = pthread_self();
-
-	return dispatch_queue(display, &display->queue, 0);
+	return wl_display_dispatch_queue_pending(display, &display->queue);
 }
 
 /** Retrieve the last error occurred on a display
diff --git a/src/wayland-client.h b/src/wayland-client.h
index 2efd01c..ec4946d 100644
--- a/src/wayland-client.h
+++ b/src/wayland-client.h
@@ -154,6 +154,7 @@ struct wl_event_queue *wl_display_create_queue(struct wl_display *display);
 
 int wl_display_acquire_fd(struct wl_display *display);
 void wl_display_release_fd(struct wl_display *display);
+int wl_display_read_events(struct wl_display *display);
 
 void wl_log_set_handler_client(wl_log_func_t handler);