event_pipe: added pipe_event enum and callbacks

Make the event_pipe (formerly main_notify) send/receive a set of
events, with a callback for each one.

The default event PIPE_EVENT_SIGNAL does not have a callback.  It
is still there for waking up the main thread, when it is waiting for
the player thread.

src/database.c

@@ -59,7 +59,6 @@ db_init(void)
 
 	do {
 		event_pipe_wait();
-		reap_update_task();
 	} while (isUpdatingDB());
 
 	stats.numberOfSongs = countSongsIn(NULL);

src/event_pipe.c

@@ -28,14 +28,45 @@
 
 GThread *main_task;
 static int event_pipe[2];
+static GMutex *event_pipe_mutex;
+static bool pipe_events[PIPE_EVENT_MAX];
+static event_pipe_callback_t event_pipe_callbacks[PIPE_EVENT_MAX];
 
-static void consume_pipe(void)
+/**
+ * Invoke the callback for a certain event.
+ */
+static void
+event_pipe_invoke(enum pipe_event event)
+{
+	assert((unsigned)event < PIPE_EVENT_MAX);
+	assert(event != PIPE_EVENT_SIGNAL);
+	assert(event_pipe_callbacks[event] != NULL);
+
+	event_pipe_callbacks[event]();
+}
+
+static bool consume_pipe(void)
 {
 	char buffer[256];
 	ssize_t r = read(event_pipe[0], buffer, sizeof(buffer));
+	bool events[PIPE_EVENT_MAX];
 
 	if (r < 0 && errno != EAGAIN && errno != EINTR)
 		FATAL("error reading from pipe: %s\n", strerror(errno));
+
+	g_mutex_lock(event_pipe_mutex);
+	memcpy(events, pipe_events, sizeof(events));
+	memset(pipe_events, 0, sizeof(pipe_events));
+	g_mutex_unlock(event_pipe_mutex);
+
+	for (unsigned i = 0; i < PIPE_EVENT_MAX; ++i)
+		if (i != PIPE_EVENT_SIGNAL && events[i])
+			/* invoke the event handler; the SIGNAL event
+			   has no handler, because it is handled by
+			   the event_pipe_wait() caller */
+			event_pipe_invoke(i);
+
+	return events[PIPE_EVENT_SIGNAL];
 }
 
 static gboolean
@@ -44,7 +75,6 @@ main_notify_event(G_GNUC_UNUSED GIOChannel *source,
 		  G_GNUC_UNUSED gpointer data)
 {
 	consume_pipe();
-	main_notify_triggered();
 	return true;
 }
 
@@ -63,22 +93,55 @@ void event_pipe_init(void)
 	g_io_add_watch(channel, G_IO_IN, main_notify_event, NULL);
 	g_io_channel_unref(channel);
 
+	event_pipe_mutex = g_mutex_new();
+
 	main_task = g_thread_self();
 }
 
 void event_pipe_deinit(void)
 {
+	g_mutex_free(event_pipe_mutex);
+
 	xclose(event_pipe[0]);
 	xclose(event_pipe[1]);
 }
 
-void event_pipe_signal(void)
+void
+event_pipe_register(enum pipe_event event, event_pipe_callback_t callback)
 {
-	ssize_t w = write(event_pipe[1], "", 1);
+	assert(event != PIPE_EVENT_SIGNAL);
+	assert((unsigned)event < PIPE_EVENT_MAX);
+	assert(event_pipe_callbacks[event] == NULL);
+
+	event_pipe_callbacks[event] = callback;
+}
+
+void event_pipe_emit(enum pipe_event event)
+{
+	ssize_t w;
+
+	assert((unsigned)event < PIPE_EVENT_MAX);
+
+	g_mutex_lock(event_pipe_mutex);
+	if (pipe_events[event]) {
+		/* already set: don't write */
+		g_mutex_unlock(event_pipe_mutex);
+		return;
+	}
+
+	pipe_events[event] = true;
+	g_mutex_unlock(event_pipe_mutex);
+
+	w = write(event_pipe[1], "", 1);
 	if (w < 0 && errno != EAGAIN && errno != EINTR)
 		g_error("error writing to pipe: %s", strerror(errno));
 }
 
+void event_pipe_signal(void)
+{
+	event_pipe_emit(PIPE_EVENT_SIGNAL);
+}
+
 void event_pipe_wait(void)
 {
 	consume_pipe();

src/event_pipe.h

@@ -23,17 +23,41 @@
 
 #include <glib.h>
 
+enum pipe_event {
+	/** the default event: the main thread is waiting for somebody,
+	    and this event wakes up the main thread */
+	PIPE_EVENT_SIGNAL = 0,
+
+	/** database update was finished */
+	PIPE_EVENT_UPDATE,
+
+	/** during database update, a song was deleted */
+	PIPE_EVENT_DELETE,
+
+	/** an idle event was emitted */
+	PIPE_EVENT_IDLE,
+
+	/** must call syncPlayerAndPlaylist() */
+	PIPE_EVENT_PLAYLIST,
+
+	PIPE_EVENT_MAX
+};
+
+typedef void (*event_pipe_callback_t)(void);
+
 extern GThread *main_task;
 
 void event_pipe_init(void);
 
 void event_pipe_deinit(void);
 
+void
+event_pipe_register(enum pipe_event event, event_pipe_callback_t callback);
+
+void event_pipe_emit(enum pipe_event event);
+
 void event_pipe_signal(void);
 
 void event_pipe_wait(void);
 
-void
-main_notify_triggered(void);
-
 #endif /* MAIN_NOTIFY_H */

src/idle.c

@@ -66,7 +66,7 @@ idle_add(unsigned flags)
 	idle_flags |= flags;
 	g_mutex_unlock(idle_mutex);
 
-	event_pipe_signal();
+	event_pipe_emit(PIPE_EVENT_IDLE);
 }
 
 unsigned

src/main.c

@@ -190,17 +190,15 @@ timer_save_state_file(G_GNUC_UNUSED gpointer data)
 	return true;
 }
 
-void
-main_notify_triggered(void)
+/**
+ * event_pipe callback function for PIPE_EVENT_IDLE
+ */
+static void
+idle_event_emitted(void)
 {
-	unsigned flags;
-
-	syncPlayerAndPlaylist();
-	reap_update_task();
-
 	/* send "idle" notificaions to all subscribed
 	   clients */
-	flags = idle_get();
+	unsigned flags = idle_get();
 	if (flags != 0)
 		client_manager_idle_add(flags);
 }
@@ -243,6 +241,10 @@ int main(int argc, char *argv[])
 
 	main_loop = g_main_loop_new(NULL, FALSE);
 
+	event_pipe_init();
+	event_pipe_register(PIPE_EVENT_IDLE, idle_event_emitted);
+	event_pipe_register(PIPE_EVENT_PLAYLIST, syncPlayerAndPlaylist);
+
 	path_global_init();
 	mapper_init();
 	initPermissions();
@@ -253,8 +255,6 @@ int main(int argc, char *argv[])
 	decoder_plugin_init_all();
 	update_global_init();
 
-	event_pipe_init();
-
 	openDB(&options, argv[0]);
 
 	command_init();

src/player_thread.c

@@ -90,7 +90,7 @@ static void player_stop_decoder(void)
 {
 	dc_stop(&pc.notify);
 	pc.state = PLAYER_STATE_STOP;
-	event_pipe_signal();
+	event_pipe_emit(PIPE_EVENT_PLAYLIST);
 }
 
 static int player_wait_for_decoder(struct player *player)
@@ -369,7 +369,7 @@ static void do_play(void)
 			   request the next song from the playlist */
 
 			pc.next_song = NULL;
-			event_pipe_signal();
+			event_pipe_emit(PIPE_EVENT_PLAYLIST);
 		}
 
 		if (decoder_is_idle() && player.queued) {
@@ -476,7 +476,7 @@ static void do_play(void)
 			if (player_wait_for_decoder(&player) < 0)
 				return;
 
-			event_pipe_signal();
+			event_pipe_emit(PIPE_EVENT_PLAYLIST);
 		} else if (decoder_is_idle()) {
 			break;
 		} else {

src/update.c

@@ -98,7 +98,7 @@ delete_song(struct directory *dir, struct song *del)
 	cond_enter(&delete_cond);
 	assert(!delete);
 	delete = del;
-	event_pipe_signal();
+	event_pipe_emit(PIPE_EVENT_DELETE);
 	do { cond_wait(&delete_cond); } while (delete);
 	cond_leave(&delete_cond);
 
@@ -603,7 +603,7 @@ static void * update_task(void *_path)
 	if (modified)
 		db_save();
 	progress = UPDATE_PROGRESS_DONE;
-	event_pipe_signal();
+	event_pipe_emit(PIPE_EVENT_UPDATE);
 	return NULL;
 }
 
@@ -646,7 +646,7 @@ directory_update_init(char *path)
 	return update_task_id;
 }
 
-void reap_update_task(void)
+static void reap_update_task(void)
 {
 	assert(g_thread_self() == main_task);
 
@@ -693,6 +693,9 @@ void update_global_init(void)
 				DEFAULT_FOLLOW_OUTSIDE_SYMLINKS);
 
 	cond_init(&delete_cond);
+
+	event_pipe_register(PIPE_EVENT_DELETE, reap_update_task);
+	event_pipe_register(PIPE_EVENT_UPDATE, reap_update_task);
 }
 
 void update_global_finish(void)

src/update.h

@@ -35,6 +35,4 @@ isUpdatingDB(void);
 unsigned
 directory_update_init(char *path);
 
-void reap_update_task(void);
-
 #endif