View source with raw comments or as raw
    1/*  Part of SWISH
    2
    3    Author:        Jan Wielemaker
    4    E-mail:        J.Wielemaker@cs.vu.nl
    5    WWW:           http://www.swi-prolog.org
    6    Copyright (C): 2018, VU University Amsterdam
    7			 CWI 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(http_dyn_workers,
   37          [
   38          ]).   39:- use_module(library(http/thread_httpd)).   40:- use_module(library(debug)).   41:- use_module(library(settings)).   42:- use_module(library(aggregate)).   43
   44:- setting(http:max_workers, integer, 100,
   45           "Maximum number of workers to create").   46:- setting(http:worker_idle_limit, number, 10,
   47           "Terminate a dynamic worker when idle for this time").   48:- setting(http:max_load, number, 10,
   49           "Maximum load average caused by HTTP workers").

Dynamically schedule HTTP workers.

Most code doesn't need to use this directly; instead use library(http/http_server), which combines this library with the typical HTTP libraries that most servers need.

This module defines hooks into the HTTP framework to dynamically schedule worker threads. Dynamic scheduling relieves us from finding a good value for the size of the HTTP worker pool.

The decision to add a worker follows these rules:

The policy depends on three settings:

http:max_workers
The maximum number of workers that will be created. Default is 100.
http:worker_idle_limit
The number of seconds a dynamic worker waits for a new job. If no job arrives in time it terminates. Default is 10 seconds.
http:max_load
Max load average created by the HTTP server, i.e. the amount of CPU time consumed per second. Default is 10. */
 http:schedule_workers(+Dict)
Called if there is no immediately free worker to handle the incomming request. The request is forwarded to the thread __http_scheduler as the hook is called in time critical code.
   88:- multifile
   89    http:schedule_workers/1.   90
   91http:schedule_workers(Dict) :-
   92    get_time(Now),
   93    catch(thread_send_message('__http_scheduler', no_workers(Now, Dict)),
   94          error(existence_error(message_queue, _), _),
   95          fail),
   96    !.
   97http:schedule_workers(Dict) :-
   98    create_scheduler,
   99    http:schedule_workers(Dict).
  100
  101create_scheduler :-
  102    catch(thread_create(http_scheduler, _,
  103                        [ alias('__http_scheduler'),
  104                          inherit_from(main),
  105                          debug(false),
  106                          detached(true)
  107                        ]),
  108          error(_,_),
  109          fail).
  110
  111http_scheduler :-
  112    get_time(Now),
  113    http_scheduler(_{ waiting:0,
  114                      time:Now
  115                    }).
  116
  117http_scheduler(State) :-
  118    (   thread_self(Me),
  119        thread_get_message(Me, Task, [timeout(10)])
  120    ->  true
  121    ;   Task = update_load_avg
  122    ),
  123    (   catch(reschedule(Task, State, State1),
  124              Error,
  125              ( print_message(warning, Error),
  126                fail))
  127    ->  !,
  128        http_scheduler(State1)
  129    ;   http_scheduler(State)
  130    ).
 reschedule(+Message, +State0, -State) is semidet
  134reschedule(no_workers(Reported, Dict), State0, State) :-
  135    update_load_avg(Dict, State0, State, Load),
  136    setting(http:max_load, MaxLoad),
  137    (   Load > MaxLoad
  138    ->  debug(http(scheduler), 'Load ~1f > ~1f; not adding workers',
  139              [ Load, MaxLoad ])
  140    ;   aggregate_all(count, http_current_worker(Dict.port, _), Workers),
  141        setting(http:max_workers, MaxWorkers),
  142        Wait is 0.001*(MaxWorkers/max(1, MaxWorkers-Workers)),
  143        get_time(Now),
  144        Sleep is Wait + Reported-Now,
  145        debug(http(scheduler), 'Waiting: ~w; active: ~w; sleep: ~3f; load: ~1f',
  146              [Dict.waiting, Workers, Sleep, Load]),
  147        sleep(Sleep),
  148        accept_queue(Dict, Queue),
  149        message_queue_property(Queue, size(Newsize)),
  150        (   Newsize == 0
  151        ->  debug(http(scheduler), 'Drained', [])
  152        ;   debug(http(scheduler), 'Size is ~w: adding worker', [Newsize]),
  153            setting(http:worker_idle_limit, MaxIdle),
  154            http_add_worker(Dict.port,
  155                            [ max_idle_time(MaxIdle)
  156                            ])
  157        )
  158    ).
  159reschedule(update_load_avg, State0, State) :-
  160    update_load_avg(_{}, State0, State, _).
  161
  162update_load_avg(_Dict, State, State, Load) :-
  163    _{stamp:Last, load:Load} :< State.get(load),
  164    get_time(Now),
  165    Now - Last < 10.
  166update_load_avg(Dict, State0, State, Load) :-
  167    server_port(Dict, State0, State1, Port),
  168    !,
  169    aggregate_all(sum(CPU), worker_cpu(Port, CPU), CPU1),
  170    get_time(Now),
  171    (   LoadDict = State1.get(load),
  172        _{stamp:Last, cpu:LastCPU} :< LoadDict
  173    ->  Load0 is (CPU1-LastCPU)/(Now-Last),
  174        smooth_load(LoadDict, Load0, Load),
  175        State = State1.put(load, _{stamp:Now, cpu:CPU1, load:Load})
  176    ;   State = State1.put(load, _{stamp:Now, cpu:CPU1}),
  177        Load = 0
  178    ).
  179update_load_avg(_, _, _, 0).
  180
  181worker_cpu(Port, CPU) :-
  182    http_current_worker(Port, Thread),
  183    catch(thread_statistics(Thread, cputime, CPU), _, fail).
  184
  185server_port(_Dict, State, State, Port) :-
  186    Port = State.get(port),
  187    !.
  188server_port(Dict, State0, State, Port) :-
  189    Port = Dict.get(port),
  190    State = State0.put(port, Port).
  191
  192smooth_load(LoadDict, Load0, Load) :-
  193    OldLoad = LoadDict.get(load),
  194    !,
  195    Load is (5*OldLoad+Load0)/6.
  196smooth_load(_, Load, Load).
 accept_queue(+Dict, -Queue)
As of 7.7.16, queue is a member of the provided dict. For older versions we need a hack.
  203accept_queue(Dict, Queue) :-
  204    Queue = Dict.get(queue),
  205    !.
  206accept_queue(Dict, Queue) :-
  207    thread_httpd:current_server(Dict.port, _, _, Queue, _, _),
  208    !