Changeset 4126

Timestamp:

03/09/08 15:15:13 (10 months ago)

Author:

pmoura

Message:

When executing a threaded/1 call, use the message queue of the calling thread for collecting individual thread results instead of creating a new message queue. In addition, join individual threads as soon as they return their results, freeing resources. Individual thread cancellation is still buggy, at least when using SWI-Prolog on MacOS X 10.5.

Files:

• trunk/compiler/logtalk.pl (modified) (7 diffs)

trunk/compiler/logtalk.pl

@@ -7336 +7336 @@
 '$lgt_tr_threaded_and_call'(TGoals, MTCalls, MTExits) :-
     '$lgt_tr_threaded_and_call'(TGoals, Queue, MTGoals, Ids, Results),
-    MTCalls = (message_queue_create(Queue), MTGoals, '$lgt_mt_check_threads'(Ids, Queue)),
-    MTExits = ('$lgt_mt_threaded_and_exit'(TGoals, Queue, Results), message_queue_destroy(Queue)).
+    MTCalls = (thread_self(Queue), MTGoals, thread_send_message(Queue, '$lgt_workers'(Ids))),
+    MTExits = ('$lgt_mt_check_threads'(Ids, Queue), '$lgt_mt_threaded_and_exit'(TGoals, Ids, Results)).
 
 
@@ -7351 +7351 @@
 '$lgt_tr_threaded_or_call'(TGoals, MTCalls, MTExits) :-
     '$lgt_tr_threaded_or_call'(TGoals, Queue, MTGoals, Ids, Results),
-    MTCalls = (message_queue_create(Queue), MTGoals, '$lgt_mt_check_threads'(Ids, Queue)),
-    MTExits = ('$lgt_mt_threaded_or_exit'(TGoals, Queue, Results), message_queue_destroy(Queue)).
+    MTCalls = (thread_self(Queue), MTGoals, thread_send_message(Queue, '$lgt_workers'(Ids))),
+    MTExits = ('$lgt_mt_check_threads'(Ids, Queue), '$lgt_mt_threaded_or_exit'(TGoals, Ids, Results)).
 
 
@@ -12938 +12938 @@
 '$lgt_mt_threaded_and_call'(TGoal, Queue) :-
     thread_self(Id),
-    (   catch(TGoal, Error, (thread_send_message(Queue, Id::exception(Error)), throw(Error))) ->
-        thread_send_message(Queue, Id::true(TGoal))
-    ;   thread_send_message(Queue, Id::fail)
+    (   catch(TGoal, Error, (thread_send_message(Queue, '$lgt_result'(Id, exception(Error))), throw(Error))) ->
+        thread_send_message(Queue, '$lgt_result'(Id, true(TGoal)))
+    ;   thread_send_message(Queue, '$lgt_result'(Id, fail))
     ).
 
@@ -12955 +12955 @@
 '$lgt_mt_check_threads'([Id| Ids], Queue) :-
     (   thread_property(Id, status(exception(Error))) ->
-        thread_send_message(Queue, Id::exception(Error))
+        thread_send_message(Queue, '$lgt_result'(Id, exception(Error)))
+    ;   '$lgt_mt_check_threads'(Ids)
+    ).
+
+
+'$lgt_mt_check_threads'([]).
+
+'$lgt_mt_check_threads'([Id| Ids]) :-
+    (   thread_property(Id, status(exception(Error))) ->
+        thread_join(Id, _)
     ;   true
     ),
-    '$lgt_mt_check_threads'(Ids, Queue).
-
-
-
-% '$lgt_mt_threaded_and_exit'(+callable, +message_queue_identifier, +list)
+    '$lgt_mt_check_threads'(Ids).
+
+
+
+% '$lgt_mt_threaded_and_exit'(+callable, +list(thread_identifier), +list)
 %
 % retrieves the result of proving a conjunction of goals using a threaded/1 predicate
 % call by collecting the individual thread results posted to the call message queue
 
-'$lgt_mt_threaded_and_exit'(TGoals, Queue, Results) :-
-    thread_get_message(Queue, Id::Result),
-    '$lgt_mt_threaded_and_exit'(Result, Id, TGoals, Queue, Results).
-
-
-'$lgt_mt_threaded_and_exit'(exception(Error), _, _, Queue, Results) :-
-    '$lgt_mt_threaded_call_abort'(Results),
-    message_queue_destroy(Queue),
+'$lgt_mt_threaded_and_exit'(TGoals, Ids, Results) :-
+    thread_get_message('$lgt_result'(Id, Result)),
+    thread_join(Id, _),
+    '$lgt_mt_threaded_and_exit'(Result, Id, TGoals, Ids, Results).
+
+
+'$lgt_mt_threaded_and_exit'(exception(Error), Id, _, Ids, _) :-
+    '$lgt_mt_threaded_call_abort'(Ids, Id),
     throw(Error).
 
-'$lgt_mt_threaded_and_exit'(true(Tgoal), Id, TGoals, Queue, Results) :-
+'$lgt_mt_threaded_and_exit'(true(Tgoal), Id, TGoals, Ids, Results) :-
     '$lgt_mt_threaded_and_add_result'(Results, Id, Tgoal, Continue),
     (   Continue == false ->
-        '$lgt_mt_threaded_and_clean'(Results, Queue),
         '$lgt_mt_threaded_and_exit_unify'(TGoals, Results)
-    ;   '$lgt_mt_threaded_and_exit'(TGoals, Queue, Results)
-    ).
-
-'$lgt_mt_threaded_and_exit'(fail, _, _, Queue, Results) :-
-    '$lgt_mt_threaded_call_abort'(Results),
-    message_queue_destroy(Queue),
+    ;   '$lgt_mt_threaded_and_exit'(TGoals, Ids, Results)
+    ).
+
+'$lgt_mt_threaded_and_exit'(fail, Id, _, Ids, _) :-
+    '$lgt_mt_threaded_call_abort'(Ids, Id),
     fail.
 
@@ -12999 +13006 @@
 
 '$lgt_mt_threaded_and_exit_unify'(TGoal, [id(_, TGoal)]).
-
-
-
-% joins all threads
-
-'$lgt_mt_threaded_and_clean'([], _).
-
-'$lgt_mt_threaded_and_clean'([id(Id, _)| Results], Queue) :-
-    thread_join(Id, _),
-    '$lgt_mt_threaded_and_clean'(Results, Queue).
 
 
@@ -13051 +13048 @@
 '$lgt_mt_threaded_or_call'(TGoal, Queue) :-
     thread_self(Id),
-    (   catch(TGoal, Error, (thread_send_message(Queue, Id::exception(Error)), throw(Error))) ->
-        thread_send_message(Queue, Id::true(TGoal))
-    ;   thread_send_message(Queue, Id::fail)
-    ).
-
-
-
-% '$lgt_mt_threaded_or_exit'(+callable, +message_queue_identifier, +list)
+    (   catch(TGoal, Error, (thread_send_message(Queue, '$lgt_result'(Id, exception(Error))), throw(Error))) ->
+        thread_send_message(Queue, '$lgt_result'(Id, true(TGoal)))
+    ;   thread_send_message(Queue, '$lgt_result'(Id, fail))
+    ).
+
+
+
+% '$lgt_mt_threaded_or_exit'(+callable, +list(thread_identifier), +list)
 %
 % retrieves the result of proving a disjunction of goals using a threaded/1 predicate
 % call by collecting the individual thread results posted to the call message queue
 
-'$lgt_mt_threaded_or_exit'(TGoals, Queue, Results) :-
-    thread_get_message(Queue, Id::Result),
-    '$lgt_mt_threaded_or_exit'(Result, Id, TGoals, Queue, Results).
-
-
-'$lgt_mt_threaded_or_exit'(exception(Error), _, _, Queue, Results) :-
-    '$lgt_mt_threaded_call_abort'(Results),
-    message_queue_destroy(Queue),
+'$lgt_mt_threaded_or_exit'(TGoals, Ids, Results) :-
+    thread_get_message('$lgt_result'(Id, Result)),
+    thread_join(Id, _),
+    '$lgt_mt_threaded_or_exit'(Result, Id, TGoals, Ids, Results).
+
+
+'$lgt_mt_threaded_or_exit'(exception(Error), Id, _, Ids, _) :-
+    '$lgt_mt_threaded_call_abort'(Ids, Id),
     throw(Error).
 
-'$lgt_mt_threaded_or_exit'(true(TGoal), Id, TGoals, _, Results) :-
-    '$lgt_mt_threaded_call_abort'(Results),
+'$lgt_mt_threaded_or_exit'(true(TGoal), Id, TGoals, Ids, Results) :-
+    '$lgt_mt_threaded_call_abort'(Ids, Id),
     '$lgt_mt_threaded_or_exit_unify'(TGoals, Results, Id, TGoal).
 
-'$lgt_mt_threaded_or_exit'(fail, Id, TGoals, Queue, Results) :-
+'$lgt_mt_threaded_or_exit'(fail, Id, TGoals, Ids, Results) :-
     '$lgt_mt_threaded_or_record_failure'(Results, Id, Continue),
     (   Continue == true ->
-        '$lgt_mt_threaded_or_exit'(TGoals, Queue, Results)
+        '$lgt_mt_threaded_or_exit'(TGoals, Ids, Results)
     ;   % all goals failed
-        '$lgt_mt_threaded_call_abort'(Results),
-        message_queue_destroy(Queue),
+        '$lgt_mt_threaded_call_abort'(Ids, Id),
         fail
     ).
@@ -13136 +13132 @@
 % we must use catch/3 as some thread may already be terminated
 
+'$lgt_mt_threaded_call_abort'(Ids, Id) :-
+    '$lgt_mt_threaded_call_abort_filter'(Ids, Id, RIds),
+    '$lgt_mt_threaded_call_abort'(RIds).
+
+
+'$lgt_mt_threaded_call_abort_filter'([], _, []).
+
+'$lgt_mt_threaded_call_abort_filter'([Id| Ids], Id, Ids) :-
+    !.
+
+'$lgt_mt_threaded_call_abort_filter'([RId| Ids], Id, [RId| RIds]) :-
+    '$lgt_mt_threaded_call_abort_filter'(Ids, Id, RIds).
+
+
 '$lgt_mt_threaded_call_abort'([]).
 
-'$lgt_mt_threaded_call_abort'([id(Id, _)| Results]) :-
-    catch(thread_signal(Id, (mutex_unlock_all, thread_exit(aborted))), _, true),
-    thread_join(Id, _),
-    '$lgt_mt_threaded_call_abort'(Results).
-
+'$lgt_mt_threaded_call_abort'([Id| Ids]) :-
+    (   catch(thread_peek_message(Id, '$lgt_workers'(DIds)), _, fail) ->
+        '$lgt_mt_threaded_call_abort'(DIds)
+    ;   true
+    ),
+    catch(thread_signal(Id, thread_exit(aborted)), _, true),
+    catch(thread_join(Id, _), _, true),
+    '$lgt_mt_threaded_call_abort'(Ids).
+