1/*  Part of SWI-Prolog
    2
    3    Author:        Jan Wielemaker
    4    E-mail:        J.Wielemaker@vu.nl
    5    WWW:           http://www.swi-prolog.org
    6    Copyright (c)  2008-2016, University of Amsterdam
    7                              VU University Amsterdam
    8    All rights reserved.
    9
   10    Redistribution and use in source and binary forms, with or without
   11    modification, are permitted provided that the following conditions
   12    are met:
   13
   14    1. Redistributions of source code must retain the above copyright
   15       notice, this list of conditions and the following disclaimer.
   16
   17    2. Redistributions in binary form must reproduce the above copyright
   18       notice, this list of conditions and the following disclaimer in
   19       the documentation and/or other materials provided with the
   20       distribution.
   21
   22    THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
   23    "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
   24    LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
   25    FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
   26    COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
   27    INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
   28    BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
   29    LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
   30    CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
   31    LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
   32    ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
   33    POSSIBILITY OF SUCH DAMAGE.
   34*/
   35
   36:- module(thread_pool,
   37          [ thread_pool_create/3,       % +Pool, +Size, +Options
   38            thread_pool_destroy/1,      % +Pool
   39            thread_create_in_pool/4,    % +Pool, :Goal, -Id, +Options
   40
   41            current_thread_pool/1,      % ?Pool
   42            thread_pool_property/2      % ?Pool, ?Property
   43          ]).   44:- use_module(library(error)).   45:- use_module(library(lists)).   46:- use_module(library(option)).   47:- use_module(library(rbtrees)).   48:- use_module(library(debug)).   49
   50
   51/** <module> Resource bounded thread management
   52
   53The module library(thread_pool) manages threads in pools. A pool defines
   54properties of its member threads and the  maximum number of threads that
   55can coexist in the pool. The   call  thread_create_in_pool/4 allocates a
   56thread in the pool, just like  thread_create/3.   If  the  pool is fully
   57allocated it can be asked to wait or raise an error.
   58
   59The library has been designed  to   deal  with  server applications that
   60receive a variety of requests, such as   HTTP servers. Simply starting a
   61thread for each request is a bit too simple minded for such servers:
   62
   63    * Creating many CPU intensive threads often leads to a slow-down
   64    rather than a speedup.
   65    * Creating many memory intensive threads may exhaust resources
   66    * Tasks that require little CPU and memory but take long waiting
   67    for external resources can run many threads.
   68
   69Using this library, one can define a  pool   for  each set of tasks with
   70comparable characteristics and create threads in   this pool. Unlike the
   71worker-pool model, threads are not started immediately. Depending on the
   72design, both approaches can be attractive.
   73
   74The library is implemented by means of   a manager thread with the fixed
   75thread id =|__thread_pool_manager|=. All  state   is  maintained in this
   76manager thread, which receives and  processes   requests  to  create and
   77destroy pools, create  threads  in  a   pool  and  handle  messages from
   78terminated threads. Thread pools are _not_ saved   in  a saved state and
   79must therefore be recreated  using   the  initialization/1  directive or
   80otherwise during startup of the application.
   81
   82@see http_handler/3 and http_spawn/2.
   83*/
   84
   85:- meta_predicate
   86    thread_create_in_pool(+, 0, -, :).   87:- predicate_options(thread_create_in_pool/4, 4,
   88                     [ wait(boolean),
   89                       pass_to(system:thread_create/3, 3)
   90                     ]).   91
   92:- multifile
   93    create_pool/1.   94
   95%!  thread_pool_create(+Pool, +Size, +Options) is det.
   96%
   97%   Create a pool of threads. A pool of threads is a declaration for
   98%   creating threads with shared  properties   (stack  sizes)  and a
   99%   limited  number  of  threads.   Threads    are   created   using
  100%   thread_create_in_pool/4. If all threads in the  pool are in use,
  101%   the   behaviour   depends   on    the     =wait=    option    of
  102%   thread_create_in_pool/4  and  the  =backlog=   option  described
  103%   below.  Options are passed to thread_create/3, except for
  104%
  105%       * backlog(+MaxBackLog)
  106%       Maximum number of requests that can be suspended.  Default
  107%       is =infinite=.  Otherwise it must be a non-negative integer.
  108%       Using backlog(0) will never delay thread creation for this
  109%       pool.
  110%
  111%   The pooling mechanism does _not_   interact  with the =detached=
  112%   state of a thread. Threads can   be  created both =detached= and
  113%   normal and must be joined using   thread_join/2  if they are not
  114%   detached.
  115
  116thread_pool_create(Name, Size, Options) :-
  117    must_be(list, Options),
  118    pool_manager(Manager),
  119    thread_self(Me),
  120    thread_send_message(Manager, create_pool(Name, Size, Options, Me)),
  121    wait_reply.
  122
  123%!  thread_pool_destroy(+Name) is det.
  124%
  125%   Destroy the thread pool named Name.
  126%
  127%   @error  existence_error(thread_pool, Name).
  128
  129thread_pool_destroy(Name) :-
  130    pool_manager(Manager),
  131    thread_self(Me),
  132    thread_send_message(Manager, destroy_pool(Name, Me)),
  133    wait_reply.
  134
  135
  136%!  current_thread_pool(?Name) is nondet.
  137%
  138%   True if Name refers to a defined thread pool.
  139
  140current_thread_pool(Name) :-
  141    pool_manager(Manager),
  142    thread_self(Me),
  143    thread_send_message(Manager, current_pools(Me)),
  144    wait_reply(Pools),
  145    (   atom(Name)
  146    ->  memberchk(Name, Pools)
  147    ;   member(Name, Pools)
  148    ).
  149
  150%!  thread_pool_property(?Name, ?Property) is nondet.
  151%
  152%   True if Property is a property of thread pool Name. Defined
  153%   properties are:
  154%
  155%       * options(Options)
  156%       Thread creation options for this pool
  157%       * free(Size)
  158%       Number of free slots on this pool
  159%       * size(Size)
  160%       Total number of slots on this pool
  161%       * members(ListOfIDs)
  162%       ListOfIDs is the list or threads running in this pool
  163%       * running(Running)
  164%       Number of running threads in this pool
  165%       * backlog(Size)
  166%       Number of delayed thread creations on this pool
  167
  168thread_pool_property(Name, Property) :-
  169    current_thread_pool(Name),
  170    pool_manager(Manager),
  171    thread_self(Me),
  172    thread_send_message(Manager, pool_properties(Me, Name, Property)),
  173    wait_reply(Props),
  174    (   nonvar(Property)
  175    ->  memberchk(Property, Props)
  176    ;   member(Property, Props)
  177    ).
  178
  179
  180%!  thread_create_in_pool(+Pool, :Goal, -Id, +Options) is det.
  181%
  182%   Create  a  thread  in  Pool.  Options  overrule  default  thread
  183%   creation options associated  to  the   pool.  In  addition,  the
  184%   following option is defined:
  185%
  186%       * wait(+Boolean)
  187%       If =true= (default) and the pool is full, wait until a
  188%       member of the pool completes.  If =false=, throw a
  189%       resource_error.
  190%
  191%   @error  resource_error(threads_in_pool(Pool)) is raised if wait
  192%           is =false= or the backlog limit has been reached.
  193%   @error  existence_error(thread_pool, Pool) if Pool does not
  194%           exist.
  195
  196thread_create_in_pool(Pool, Goal, Id, QOptions) :-
  197    meta_options(is_meta, QOptions, Options),
  198    catch(thread_create_in_pool_(Pool, Goal, Id, Options),
  199          Error, true),
  200    (   var(Error)
  201    ->  true
  202    ;   Error = error(existence_error(thread_pool, Pool), _),
  203        create_pool_lazily(Pool)
  204    ->  thread_create_in_pool_(Pool, Goal, Id, Options)
  205    ;   throw(Error)
  206    ).
  207
  208thread_create_in_pool_(Pool, Goal, Id, Options) :-
  209    select_option(wait(Wait), Options, ThreadOptions, true),
  210    pool_manager(Manager),
  211    thread_self(Me),
  212    thread_send_message(Manager,
  213                        create(Pool, Goal, Me, Wait, Id, ThreadOptions)),
  214    wait_reply(Id).
  215
  216is_meta(at_exit).
  217
  218
  219%!  create_pool_lazily(+Pool) is semidet.
  220%
  221%   Call the hook create_pool/1 to create the pool lazily.
  222
  223create_pool_lazily(Pool) :-
  224    with_mutex(Pool,
  225               ( mutex_destroy(Pool),
  226                 create_pool_sync(Pool)
  227               )).
  228
  229create_pool_sync(Pool) :-
  230    current_thread_pool(Pool),
  231    !.
  232create_pool_sync(Pool) :-
  233    create_pool(Pool).
  234
  235
  236                 /*******************************
  237                 *        START MANAGER         *
  238                 *******************************/
  239
  240%!  pool_manager(-ThreadID) is det.
  241%
  242%   ThreadID is the thread (alias) identifier of the manager. Starts
  243%   the manager if it is not running.
  244
  245pool_manager(TID) :-
  246    TID = '__thread_pool_manager',
  247    (   thread_running(TID)
  248    ->  true
  249    ;   with_mutex('__thread_pool', create_pool_manager(TID))
  250    ).
  251
  252thread_running(Thread) :-
  253    catch(thread_property(Thread, status(Status)),
  254          E, true),
  255    (   var(E)
  256    ->  (   Status == running
  257        ->  true
  258        ;   thread_join(Thread, _),
  259            print_message(warning, thread_pool(manager_died(Status))),
  260            fail
  261        )
  262    ;   E = error(existence_error(thread, Thread), _)
  263    ->  fail
  264    ;   throw(E)
  265    ).
  266
  267create_pool_manager(Thread) :-
  268    thread_running(Thread),
  269    !.
  270create_pool_manager(Thread) :-
  271    thread_create(pool_manager_main, _,
  272                  [ alias(Thread),
  273                    inherit_from(main)
  274                  ]).
  275
  276
  277pool_manager_main :-
  278    rb_new(State0),
  279    manage_thread_pool(State0).
  280
  281
  282                 /*******************************
  283                 *        MANAGER LOGIC         *
  284                 *******************************/
  285
  286%!  manage_thread_pool(+State)
  287
  288manage_thread_pool(State0) :-
  289    thread_get_message(Message),
  290    (   update_thread_pool(Message, State0, State)
  291    ->  debug(thread_pool(state), 'Message ~p --> ~p', [Message, State]),
  292        manage_thread_pool(State)
  293    ;   format(user_error, 'Update failed: ~p~n', [Message])
  294    ).
  295
  296
  297update_thread_pool(create_pool(Name, Size, Options, For), State0, State) :-
  298    !,
  299    (   rb_insert_new(State0,
  300                      Name, tpool(Options, Size, Size, WP, WP, []),
  301                      State)
  302    ->  thread_send_message(For, thread_pool(true))
  303    ;   reply_error(For, permission_error(create, thread_pool, Name)),
  304        State = State0
  305    ).
  306update_thread_pool(destroy_pool(Name, For), State0, State) :-
  307    !,
  308    (   rb_delete(State0, Name, State)
  309    ->  thread_send_message(For, thread_pool(true))
  310    ;   reply_error(For, existence_error(thread_pool, Name)),
  311        State = State0
  312    ).
  313update_thread_pool(current_pools(For), State, State) :-
  314    !,
  315    rb_keys(State, Keys),
  316    debug(thread_pool(current), 'Reply to ~w: ~p', [For, Keys]),
  317    reply(For, Keys).
  318update_thread_pool(pool_properties(For, Name, P), State, State) :-
  319    !,
  320    (   rb_lookup(Name, Pool, State)
  321    ->  findall(P, pool_property(P, Pool), List),
  322        reply(For, List)
  323    ;   reply_error(For, existence_error(thread_pool, Name))
  324    ).
  325update_thread_pool(Message, State0, State) :-
  326    arg(1, Message, Name),
  327    (   rb_lookup(Name, Pool0, State0)
  328    ->  update_pool(Message, Pool0, Pool),
  329        rb_update(State0, Name, Pool, State)
  330    ;   State = State0,
  331        (   Message = create(Name, _, For, _, _, _)
  332        ->  reply_error(For, existence_error(thread_pool, Name))
  333        ;   true
  334        )
  335    ).
  336
  337pool_property(options(Options),
  338              tpool(Options, _Free, _Size, _WP, _WPT, _Members)).
  339pool_property(backlog(Size),
  340              tpool(_, _Free, _Size, WP, WPT, _Members)) :-
  341    diff_list_length(WP, WPT, Size).
  342pool_property(free(Free),
  343              tpool(_, Free, _Size, _, _, _)).
  344pool_property(size(Size),
  345              tpool(_, _Free, Size, _, _, _)).
  346pool_property(running(Count),
  347              tpool(_, Free, Size, _, _, _)) :-
  348    Count is Size - Free.
  349pool_property(members(IDList),
  350              tpool(_, _, _, _, _, IDList)).
  351
  352diff_list_length(List, Tail, Size) :-
  353    '$skip_list'(Length, List, Rest),
  354    (   Rest == Tail
  355    ->  Size = Length
  356    ;   type_error(difference_list, List/Tail)
  357    ).
  358
  359
  360%!  update_pool(+Message, +Pool0, -Pool) is det.
  361%
  362%   Deal with create requests and  completion   messages  on a given
  363%   pool.  There are two messages:
  364%
  365%       * create(PoolName, Goal, ForThread, Wait, Id, Options)
  366%       Create a new thread on behalf of ForThread.  There are
  367%       two cases:
  368%            * Free slots: create the thread
  369%            * No free slots: error or add to waiting
  370%       * exitted(PoolName, Thread)
  371%       A thread completed.  If there is a request waiting,
  372%       create a new one.
  373
  374update_pool(create(Name, Goal, For, _, Id, MyOptions),
  375            tpool(Options, Free0, Size, WP, WPT, Members0),
  376            tpool(Options, Free, Size, WP, WPT, Members)) :-
  377    succ(Free, Free0),
  378    !,
  379    merge_options(MyOptions, Options, ThreadOptions),
  380    select_option(at_exit(AtExit), ThreadOptions, ThreadOptions1, true),
  381    catch(thread_create(Goal, Id,
  382                        [ at_exit(worker_exitted(Name, Id, AtExit))
  383                        | ThreadOptions1
  384                        ]),
  385          E, true),
  386    (   var(E)
  387    ->  Members = [Id|Members0],
  388        reply(For, Id)
  389    ;   reply_error(For, E),
  390        Members = Members0
  391    ).
  392update_pool(Create,
  393            tpool(Options, 0, Size, WP, WPT0, Members),
  394            tpool(Options, 0, Size, WP, WPT, Members)) :-
  395    Create = create(Name, _Goal, For, Wait, _, _Options),
  396    !,
  397    option(backlog(BackLog), Options, infinite),
  398    (   can_delay(Wait, BackLog, WP, WPT0)
  399    ->  WPT0 = [Create|WPT],
  400        debug(thread_pool, 'Delaying ~p', [Create])
  401    ;   WPT = WPT0,
  402        reply_error(For, resource_error(threads_in_pool(Name)))
  403    ).
  404update_pool(exitted(_Name, Id),
  405            tpool(Options, Free0, Size, WP0, WPT, Members0),
  406            Pool) :-
  407    succ(Free0, Free),
  408    delete(Members0, Id, Members1),
  409    Pool1 = tpool(Options, Free, Size, WP, WPT, Members1),
  410    (   WP0 == WPT
  411    ->  WP = WP0,
  412        Pool = Pool1
  413    ;   WP0 = [Waiting|WP],
  414        debug(thread_pool, 'Start delayed ~p', [Waiting]),
  415        update_pool(Waiting, Pool1, Pool)
  416    ).
  417
  418
  419can_delay(true, infinite, _, _) :- !.
  420can_delay(true, BackLog, WP, WPT) :-
  421    diff_list_length(WP, WPT, Size),
  422    BackLog > Size.
  423
  424%!  worker_exitted(+PoolName, +WorkerId, :AtExit)
  425%
  426%   It is possible that  '__thread_pool_manager'   no  longer exists
  427%   while closing down the process because   the  manager was killed
  428%   before the worker.
  429%
  430%   @tbd Find a way to discover that we are terminating Prolog.
  431
  432worker_exitted(Name, Id, AtExit) :-
  433    catch(thread_send_message('__thread_pool_manager', exitted(Name, Id)),
  434          _, true),
  435    call(AtExit).
  436
  437
  438                 /*******************************
  439                 *             UTIL             *
  440                 *******************************/
  441
  442reply(To, Term) :-
  443    thread_send_message(To, thread_pool(true(Term))).
  444
  445reply_error(To, Error) :-
  446    thread_send_message(To, thread_pool(error(Error, _))).
  447
  448wait_reply :-
  449    thread_get_message(thread_pool(Result)),
  450    (   Result == true
  451    ->  true
  452    ;   Result == fail
  453    ->  fail
  454    ;   throw(Result)
  455    ).
  456
  457wait_reply(Value) :-
  458    thread_get_message(thread_pool(Reply)),
  459    (   Reply = true(Value0)
  460    ->  Value = Value0
  461    ;   Reply == fail
  462    ->  fail
  463    ;   throw(Reply)
  464    ).
  465
  466
  467                 /*******************************
  468                 *             HOOKS            *
  469                 *******************************/
  470
  471%!  create_pool(+PoolName) is semidet.
  472%
  473%   Hook to create a thread  pool  lazily.   The  hook  is called if
  474%   thread_create_in_pool/4 discovers that the thread  pool does not
  475%   exist. If the  hook   succeeds,  thread_create_in_pool/4 retries
  476%   creating the thread. For  example,  we   can  use  the following
  477%   declaration to create threads in the pool =media=, which holds a
  478%   maximum of 20 threads.
  479%
  480%     ==
  481%     :- multifile thread_pool:create_pool/1.
  482%
  483%     thread_pool:create_pool(media) :-
  484%         thread_pool_create(media, 20, []).
  485%     ==
  486
  487                 /*******************************
  488                 *            MESSAGES          *
  489                 *******************************/
  490:- multifile
  491    prolog:message/3.  492
  493prolog:message(thread_pool(Message)) -->
  494    message(Message).
  495
  496message(manager_died(Status)) -->
  497    [ 'Thread-pool: manager died on status ~p; restarting'-[Status] ]