Explicitly manage FIFO thread resource

src/fifo.pl

@@ -2,18 +2,20 @@
 
 :- module(fifo, []).
 
-%! setup_fifo() is det
+%! setup_fifo(Tid) is det
 %
 %  If config:fifo_enabled/1 and config:fifo_path/1 are set, attempts to create
 %  a named pipe with mkfifo(1).
-%  If the fifo is created, its path is passed to fifo:process_fifo/1 on a detached thread.
-setup_fifo() :-
+%  If the fifo is created, its path is passed to fifo:process_fifo/1.
+%
+%  @arg Tid thread id of the fifo handler
+setup_fifo(Tid) :-
 	optcnf_then(fifo_enabled(true), optcnf_then(fifo_path(FifoPath), (
 		catch(delete_file(FifoPath), _, true), % cleanup from previous execution
 		string_concat("mkfifo ", FifoPath, MkFifoCmd), % no swipl predicate for this
 		shell(MkFifoCmd, ExitCode),
 		(ExitCode == 0 ->
-			thread_create(fifo:process_fifo(FifoPath), _, [detached(true)])
+			thread_create(fifo:process_fifo(FifoPath), Tid)
 		; writeln(user_error, "Could not spawn command fifo!"))
 	)))
 .
@@ -30,12 +32,21 @@
 %
 %  @arg FifoPath file path to the command fifo
 process_fifo(FifoPath) :-
-	open(FifoPath, read, Fifo),
-	(read_terms(Fifo, Jobs) ->
-		jobs_notify(Jobs)
-	; true),
-	close(Fifo),
-	process_fifo(FifoPath)
+	thread_peek_message(shutdown) ->
+		thread_get_message(shutdown)
+		;
+		catch(
+			% open/3 blocks when FIFO's other end isn't open for
+			% writing. In order to poll the message queue in a
+			% timely manner, we need this timeout.
+			(call_with_time_limit(0.1, open(FifoPath, read, Fifo)),
+				(read_terms(Fifo, Jobs) ->
+					jobs_notify(Jobs)
+				; true),
+				close(Fifo)
+			), _, true
+		),
+		process_fifo(FifoPath)
 .
 
 %! read_terms(++S:stream, --Terms:[term]) is semidet

src/plwm.pl

@@ -2341,11 +2341,13 @@
 	update_free_win_space,
 	update_ws_atoms,
 
-	fifo:setup_fifo,
+	fifo:setup_fifo(FifoThread),
 
 	setup_hooks,
 	run_hook(start),
 
-	eventloop
+	eventloop,
+	thread_send_message(FifoThread, shutdown),
+	thread_join(FifoThread)
 .
 

tests/unit_tests/fifo.plt

@@ -9,39 +9,35 @@ optcnf_then(fifo_path(FifoPath), Then) :- FifoPath = "/tmp/test-fifo", Then.
 :- use_module("../../src/fifo").
 
 test("setup_fifo +", [
-	setup(
-		optcnf_then(fifo_path(FifoPath), true)
-	),
-	cleanup(
+	setup((
+		optcnf_then(fifo_path(FifoPath), true),
+		fifo:setup_fifo(FifoThread)
+	)),
+	cleanup((
+		thread_send_message(FifoThread, shutdown),
+		thread_join(FifoThread),
 		delete_file(FifoPath)
-		% Note: thread spawned by setup_fifo is detached, so no need to join
-	)
+	))
 ]) :-
-	assertion(fifo:setup_fifo),
 	assertion(access_file(FifoPath, read)),
-
-	% Check if setup_fifo started the thread for process_fifo,
-	% i.e. an extra thread is running next to 'main' and 'gc'
-	findall(T, thread_property(T, _), Threads),
-	assertion(list_to_set(Threads, [main, gc, _]))
+	assertion(thread_property(FifoThread, status(running)))
 .
 
 test("setup_fifo + (already exists)", [
 	setup((
 		optcnf_then(fifo_path(FifoPath), true),
-		open(FifoPath, write, _) % create empty file
+		open(FifoPath, write, Stream), % create empty file
+		fifo:setup_fifo(FifoThread)
 	)),
-	cleanup(
+	cleanup((
+		close(Stream),
+		thread_send_message(FifoThread, shutdown),
+		thread_join(FifoThread),
 		delete_file(FifoPath)
-	)
+	))
 ]) :-
-	assertion(fifo:setup_fifo),
 	assertion(access_file(FifoPath, read)),
-
-	findall(T, thread_property(T, _), Threads),
-	assertion(list_to_set(Threads, [main, gc, _, _]))
-	% Note: one more thread due to execution of prior testcase,
-	% unfortunately we cannot destroy detached threads:/
+	assertion(thread_property(FifoThread, status(running)))
 .
 
 test("process_fifo", [