SWI-Prolog -- Message queues

10.3.1 Message queues

Prolog threads can exchange data using dynamic predicates, database records, and other globally shared data. These provide no suitable means to wait for data or a condition as they can only be checked in an expensive polling loop. Message queues provide a means for threads to wait for data or conditions without using the CPU.

Each thread has a message queue attached to it that is identified by the thread. Additional queues are created using message_queue_create/1. Explicitly created queues come in two flavours. When given an alias, they must be destroyed by the user. Anonymous message queues are identified by a blob (see section 12.4.10) and subject to garbage collection.

thread_send_message(+QueueOrThreadId, +Term)

Place Term in the given queue or default queue of the indicated thread (which can even be the message queue of itself, see thread_self/1). Any term can be placed in a message queue, but note that the term is copied to the receiving thread and variable bindings are thus lost. This call returns immediately.

If more than one thread is waiting for messages on the given queue and at least one of these is waiting with a partially instantiated Term, the waiting threads are all sent a wake-up signal, starting a rush for the available messages in the queue. This behaviour can seriously harm performance with many threads waiting on the same queue as all-but-the-winner perform a useless scan of the queue. If there is only one waiting thread or all waiting threads wait with an unbound variable, an arbitrary thread is restarted to scan the queue.^{199See the documentation for
the POSIX thread functions pthread_cond_signal() v.s. pthread_cond_broadcast()
for background information.}

[semidet]thread_send_message(+Queue, +Term, +Options)

As thread_send_message/2, but providing additional Options. These are to deal with the case that the queue has a finite maximum size and is full: whereas thread_send_message/2 will block until the queue has drained sufficiently to accept a new message, thread_send_message/3 can accept a time-out or deadline analogously to thread_get_message/3. The options are:

deadline(+AbsTime)

The call fails (silently) if no space has become available before AbsTime. See get_time/1 for the representation of absolute time. If AbsTime is earlier then the current time, thread_send_message/3 fails immediately. Both resolution and maximum wait time is platform-dependent.^{200The
implementation uses MsgWaitForMultipleObjects() on MS-Windows and pthread_cond_timedwait()
on other systems.}

timeout(+Time)

Time is a float or integer and specifies the maximum time to wait in seconds. This is a relative-time version of the deadline option. If both options are provided, the earlier time is effective.

If Time is 0 or 0.0, thread_send_message/3 examines the queue and sends the message if space is available, but does not suspend if no space is available, failing immediately instead.

If Time < 0, thread_send_message/3 fails immediately without sending the message.

thread_get_message(?Term)

Examines the thread message queue and if necessary blocks execution until a term that unifies to Term arrives in the queue. After a term from the queue has been unified to Term, the term is deleted from the queue.

Please note that non-unifying messages remain in the queue. After the following has been executed, thread 1 has the term b(gnu) in its queue and continues execution using A = gnat.

   <thread 1>
   thread_get_message(a(A)),

   <thread 2>
   thread_send_message(Thread_1, b(gnu)),
   thread_send_message(Thread_1, a(gnat)),

Term may contain attributed variables (see section 8), in which case only terms for which the constraints successfully execute are returned. Handle constraints applies for all predicates that extract terms from message queues. For example, we can get the even numbers from a queue using this code:

get_matching_messages(Queue, Pattern, [H|T]) :-
    copy_term(Pattern, H),
    thread_get_message(Queue, H, [timeout(0)]),
    !,
    get_matching_messages(Queue, Pattern, T).
get_matching_messages(_, _, []).

even_numbers(Q, List) :-
    freeze(Even, Even mod 2 =:= 0),
    get_matching_messages(Q, Even, List).

See also thread_peek_message/1.

thread_peek_message(?Term)

Examines the thread message queue and compares the queued terms with Term until one unifies or the end of the queue has been reached. In the first case the call succeeds, possibly instantiating Term. If no term from the queue unifies, this call fails. I.e., thread_peek_message/1 never waits and does not remove any term from the queue. See also thread_get_message/3.

message_queue_create(?Queue)

Equivalent to message_queue_create(Queue,[]). For compatibility, calling message_queue_create(+Atom) is equivalent to message_queue_create(Queue, [alias(Atom)]). New code should use message_queue_create/2 to create a named queue.

message_queue_create(-Queue, +Options)

Create a message queue from Options. Defined options are:

alias(+Alias): Create a message queue that is identified by the atom Alias. Message queues created this way must be explicitly destroyed by the user. If the alias option is omitted, an Anonymous queue is created that is identified by a blob (see section 12.4.10) and subject to garbage collection.^{201Garbage
collecting anonymous message queues is not part of the ISO proposal and
most likely not a widely implemented feature.}
max_size(+Size): Maximum number of terms in the queue. If this number is reached, thread_send_message/2 will suspend until the queue is drained. The option can be used if the source, sending messages to the queue, is faster than the drain, consuming the messages.

[det]message_queue_destroy(+Queue)

Destroy a message queue created with message_queue_create/1. A permission error is raised if Queue refers to (the default queue of) a thread. Other threads that are waiting for Queue using thread_get_message/2 receive an existence error.

[det]thread_get_message(+Queue, ?Term)

As thread_get_message/1, operating on a given queue. It is allowed (but not advised) to get messages from the queue of other threads. This predicate raises an existence error exception if Queue doesn't exist or is destroyed using message_queue_destroy/1 while this predicate is waiting.

[semidet]thread_get_message(+Queue, ?Term, +Options)

As thread_get_message/2, but providing additional Options:

deadline(+AbsTime)

The call fails (silently) if no message has arrived before AbsTime. See get_time/1 for the representation of absolute time. If AbsTime is earlier then the current time, thread_get_message/3 fails immediately. Both resolution and maximum wait time is platform-dependent.^{202The
implementation uses MsgWaitForMultipleObjects() on MS-Windows and pthread_cond_timedwait()
on other systems.}

timeout(+Time)

Time is a float or integer and specifies the maximum time to wait in seconds. This is a relative-time version of the deadline option. If both options are provided, the earlier time is effective.

If Time is 0 or 0.0, thread_get_message/3 examines the queue but does not suspend if no matching term is available. Note that unlike thread_peek_message/2, a matching term is removed from the queue.

If Time < 0, thread_get_message/3 fails immediately without removing any message from the queue.

[semidet]thread_peek_message(+Queue, ?Term)

As thread_peek_message/1, operating on a given queue. It is allowed to peek into another thread's message queue, an operation that can be used to check whether a thread has swallowed a message sent to it.

message_queue_property(?Queue, ?Property)

True if Property is a property of Queue. Defined properties are:

alias(Alias): Queue has the given alias name.
max_size(Size): Maximum number of terms that can be in the queue. See message_queue_create/2. This property is not present if there is no limit (default).
size(Size): Queue currently contains Size terms. Note that due to concurrent access the returned value may be outdated before it is returned. It can be used for debugging purposes as well as work distribution purposes.
waiting(-Count): Number of threads waiting for this queue. This property is not present if no threads waits for this queue.

The size(Size) property is always present and may be used to enumerate the created message queues. Note that this predicate does not enumerate threads, but can be used to query the properties of the default queue of a thread.

message_queue_set(+Queue, +Property)

Set a property on the queue. Supported properties are:

max_size(+Size): Change the number of terms that may appear in the message queue before the next thread_send_message/[2,3] blocks on it. If the value is higher then the current maximum and the queue has writers waiting, wakeup the writers. The value can be lower than the current number of terms in the queue. In that case writers will block until the queue is drained below the new maximum.

Explicit message queues are designed with the worker-pool model in mind, where multiple threads wait on a single queue and pick up the first goal to execute. Below is a simple implementation where the workers execute arbitrary Prolog goals. Note that this example provides no means to tell when all work is done. This must be realised using additional synchronisation.

%%      create_workers(?Id, +N)
%
%       Create a pool with Id and number of workers.
%       After the pool is created, post_job/1 can be used to
%       send jobs to the pool.

create_workers(Id, N) :-
        message_queue_create(Id),
        forall(between(1, N, _),
               thread_create(do_work(Id), _, [])).

do_work(Id) :-
        repeat,
          thread_get_message(Id, Goal),
          (   catch(Goal, E, print_message(error, E))
          ->  true
          ;   print_message(error, goal_failed(Goal, worker(Id)))
          ),
        fail.

%%      post_job(+Id, +Goal)
%
%       Post a job to be executed by one of the pool's workers.

post_job(Id, Goal) :-
        thread_send_message(Id, Goal).