1/*  File:    threaded/throttle.pl
    2    Author:  Roy Ratcliffe
    3    Created: Jul  7 2026
    4    Purpose: Threaded throttle mechanism for Redis streams
    5
    6Copyright (c) 2026, Roy Ratcliffe, Northumberland, United Kingdom
    7
    8Permission is hereby granted, free of charge,  to any person obtaining a
    9copy  of  this  software  and    associated   documentation  files  (the
   10"Software"), to deal in  the   Software  without  restriction, including
   11without limitation the rights to  use,   copy,  modify,  merge, publish,
   12distribute, sub-license, and/or sell copies  of   the  Software,  and to
   13permit persons to whom the Software is   furnished  to do so, subject to
   14the following conditions:
   15
   16    The above copyright notice and this permission notice shall be
   17    included in all copies or substantial portions of the Software.
   18
   19THE SOFTWARE IS PROVIDED "AS IS", WITHOUT  WARRANTY OF ANY KIND, EXPRESS
   20OR  IMPLIED,  INCLUDING  BUT  NOT   LIMITED    TO   THE   WARRANTIES  OF
   21MERCHANTABILITY, FITNESS FOR A PARTICULAR   PURPOSE AND NONINFRINGEMENT.
   22IN NO EVENT SHALL THE AUTHORS  OR   COPYRIGHT  HOLDERS BE LIABLE FOR ANY
   23CLAIM, DAMAGES OR OTHER LIABILITY,  WHETHER   IN  AN ACTION OF CONTRACT,
   24TORT OR OTHERWISE, ARISING FROM,  OUT  OF   OR  IN  CONNECTION  WITH THE
   25SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
   26
   27*/
   28
   29:- module(threaded_throttle,
   30          [ consume_threaded_throttle/2 % +Key, +Field
   31          , unconsume_threaded_throttle/2 % +Key, +Field
   32          , current_threaded_throttle_timeout/4 % +When, +Key, +Field, -Timeout
   33          , set_threaded_throttle_timeout/4 % +When, +Key, +Field, +Timeout
   34          , exit_threaded_throttle/2 % +Key, +Field
   35          ]).   36:- autoload(library(broadcast), [listen/3, unlisten/1, broadcast/1]).   37:- autoload(library(redis), [redis/3]).   38:- use_module(library(settings), [setting/4, setting/2]).

Threaded Throttle

This module provides functionality for consuming throttle events from a Redis stream in a threaded manner. It allows for the creation of threads to handle throttle events, with configurable idle and wait timeouts. The module also provides predicates to set and retrieve timeout values for specific keys and fields.

Broadcast Events

The module broadcasts the following events:

   63:- setting(idle_timeout, number, env('THREADED_THROTTLE_IDLE_TIMEOUT', 5),
   64           'Timeout in seconds for idling').   65
   66:- setting(wait_timeout, number, env('THREADED_THROTTLE_WAIT_TIMEOUT', 5),
   67           'Timeout in seconds for waiting').
 consume_threaded_throttle(+Key:atom, +Field:atom) is det
Listens to a Redis stream for throttle events on the specified Key and Field. When an event is received, it creates a thread to handle the event and sends the value to that thread.
Arguments:
Key- The Redis stream key to listen to.
Field- The field within the Redis stream to listen for events.
   79consume_threaded_throttle(Key, Field) :-
   80    % Use the thread alias as the listener to avoid duplicate listeners for the
   81    % same Key and Field.
   82    key_field_alias(Key, Field, Alias),
   83    listen(Alias, redis_consume(Key, Entry, _), consume(Key, Field, Entry)).
 unconsume_threaded_throttle(+Key:atom, +Field:atom) is det
Stops listening to the Redis stream for throttle events on the specified Key and Field.
Arguments:
Key- The Redis stream key to stop listening to.
Field- The field within the Redis stream to stop listening for events.
   95unconsume_threaded_throttle(Key, Field) :-
   96    key_field_alias(Key, Field, Alias),
   97    unlisten(Alias).
   98
   99consume(Key, Field, Entry) :-
  100    get_dict(Field, Entry, Value),
  101    !,
  102    % Lazily create a thread for the Key and Field if it doesn't exist, and
  103    % send the Value to that thread.
  104    create_thread(Key, Field, Thread),
  105    thread_send_message(Thread, Value).
  106consume(_, _, _).
  107
  108% TODO: make the alias construction injective so that it is unique for
  109% each Key and Field combination. This will prevent potential conflicts
  110% if different Key and Field combinations produce the same alias.
  111key_field_alias(Key, Field, Alias) :-
  112    atomic_list_concat([threaded, Key, Field, throttle], '_', Alias).
  113
  114create_thread(Key, Field, Thread) :-
  115    key_field_alias(Key, Field, Alias),
  116    (   thread_property(Thread, alias(Alias))
  117    ->  true
  118    ;   thread_create(catch(throttle(Key, Field), quit, true), Thread,
  119                      [ alias(Alias),
  120                        detached(true)
  121                      ])
  122    ).
  123
  124throttle(Key, Field) :-
  125    (   redis(default, get(Key:Field), Value)
  126    ->  throttle_(Key, Field, Value)
  127    ;   throttle____(Key, Field)
  128    ).
  129
  130throttle_(Key, Field, Value) :-
  131    broadcast(threaded_throttle(up, Key, Field)),
  132    throttle__(Key, Field, Value).
  133
  134throttle__(Key, Field, Value) :-
  135    broadcast(threaded_throttle(adjust, Key, Field, Value)),
  136    redis(default, set(Key:Field, Value), status(ok)),
  137    throttle___(Key, Field, Value).
  138
  139throttle___(Key, Field, Value) :-
  140    thread_self(Self),
  141    current_threaded_throttle_timeout(idle, Key, Field, Timeout),
  142    (   thread_get_message(Self, Value1, [timeout(Timeout)])
  143    ->  (   thread_peek_message(Self, _)
  144        ->  throttle___(Key, Field, Value)
  145        ;   (   Value == Value1
  146            ->  throttle___(Key, Field, Value1)
  147            ;   throttle__(Key, Field, Value1)
  148            )
  149        )
  150    ;   throttle____(Key, Field)
  151    ).
  152
  153throttle____(Key, Field) :-
  154    broadcast(threaded_throttle(down, Key, Field)),
  155    thread_self(Self),
  156    current_threaded_throttle_timeout(wait, Key, Field, Timeout),
  157    (   thread_get_message(Self, Value, [timeout(Timeout)])
  158    ->  throttle_(Key, Field, Value)
  159    ;   throttle(Key, Field)
  160    ).
  161
  162:- dynamic timeout/4.
 current_threaded_throttle_timeout(+When:atom, +Key:atom, +Field:atom, -Timeout:number) is det
Retrieves the current timeout value for the specified Key and Field based on the When condition (idle or wait). If a specific timeout has been set for the Key and Field, it will be used; otherwise, the default setting will be retrieved from the application settings.
Arguments:
When- The condition for which the timeout is being retrieved (idle or wait).
Key- The Redis stream key for which the timeout is being retrieved.
Field- The field within the Redis stream for which the timeout is being retrieved.
Timeout- The timeout value in seconds.
  185current_threaded_throttle_timeout(When, Key, Field, Timeout) :-
  186    when(When, TimeoutSetting),
  187    (   timeout(When, Key, Field, Timeout)
  188    ->  true
  189    ;   setting(TimeoutSetting, Timeout)
  190    ).
 set_threaded_throttle_timeout(+When:atom, +Key:atom, +Field:atom, +Timeout:number) is det
Sets a specific timeout value for the specified Key and Field based on the When condition (idle or wait). This allows for dynamic adjustment of timeout values for different throttle events.
Arguments:
When- The condition for which the timeout is being set (idle or wait).
Key- The Redis stream key for which the timeout is being set.
Field- The field within the Redis stream for which the timeout is being set.
Timeout- The timeout value in seconds.
  211set_threaded_throttle_timeout(When, Key, Field, Timeout) :-
  212    when(When, _),
  213    retractall(timeout(When, Key, Field, _)),
  214    assertz(timeout(When, Key, Field, Timeout)).
  215
  216when(idle, idle_timeout).
  217when(wait, wait_timeout).
 exit_threaded_throttle(+Key:atom, +Field:atom) is det
Exits the threaded throttle for the specified Key and Field by signaling the associated thread to terminate. If a thread exists for the Key and Field, it will be signaled to throw an exit_thread exception, allowing for graceful termination of the thread.
  226exit_threaded_throttle(Key, Field) :-
  227    key_field_alias(Key, Field, Alias),
  228    (   thread_property(Thread, alias(Alias))
  229    ->  thread_signal(Thread, throw(quit))
  230    ;   true
  231    )