One Level Up Top Level

src/stream/ngx_stream_upstream_least_time_module.c - nginx

Global variables defined

Data types defined

Functions defined

Macros defined

Source code


  2. /*
  3. * Copyright (C) Nginx, Inc.
  4. */


  7. #include <ngx_config.h>
  8. #include <ngx_core.h>
  9. #include <ngx_stream.h>


  12. ‌#define NGX_STREAM_UPSTREAM_LT_CONNECT     0
  13. ‌#define NGX_STREAM_UPSTREAM_LT_FIRST_BYTE  1
  14. ‌#define NGX_STREAM_UPSTREAM_LT_LAST_BYTE   2


  17. ‌typedef struct {
  18.     ngx_uint_t                           mode;
  19.     ngx_uint_t                           use_inflight;
  20.                                                 /* unsigned  use_inflight:1; */
  21. ‌} ngx_stream_upstream_lt_conf_t;


  24. ‌typedef struct {
  25.     /* the round robin data must be first */
  26.     ngx_stream_upstream_rr_peer_data_t   rrp;

  28.     ngx_stream_upstream_lt_conf_t       *conf;
  29.     ngx_uint_t                           inflight;  /* unsigned  inflight:1; */
  30.     ngx_stream_upstream_t               *upstream;
  31. ‌} ngx_stream_upstream_lt_peer_data_t;


  34. static ngx_int_t ngx_stream_upstream_init_least_time_peer(
  35.     ngx_stream_session_t *s, ngx_stream_upstream_srv_conf_t *us);
  36. static ngx_int_t ngx_stream_upstream_get_least_time_peer(
  37.     ngx_peer_connection_t *pc, void *data);
  38. static ngx_uint_t ngx_stream_upstream_least_time_eta(
  39.     ngx_stream_upstream_lt_peer_data_t *ltp,
  40.     ngx_stream_upstream_rr_peer_t *peer);
  41. static void ngx_stream_upstream_least_time_notify(ngx_peer_connection_t *pc,
  42.     void *data, ngx_uint_t type);
  43. static void ngx_stream_upstream_least_time_inflight_done(
  44.     ngx_stream_upstream_lt_peer_data_t *ltp,
  45.     ngx_stream_upstream_rr_peers_t *peers, ngx_stream_upstream_rr_peer_t *peer,
  46.     ngx_msec_t last);
  47. static void ngx_stream_upstream_free_least_time_peer(ngx_peer_connection_t *pc,
  48.     void *data, ngx_uint_t state);

  50. static void *ngx_stream_upstream_least_time_create_conf(ngx_conf_t *cf);
  51. static char *ngx_stream_upstream_least_time(ngx_conf_t *cf, ngx_command_t *cmd,
  52.     void *conf);


  55. ‌static ngx_conf_enum_t  ngx_stream_upstream_least_time_mode[] = {
  56.     { ngx_string("connect"), NGX_STREAM_UPSTREAM_LT_CONNECT },
  57.     { ngx_string("first_byte"), NGX_STREAM_UPSTREAM_LT_FIRST_BYTE },
  58.     { ngx_string("last_byte"), NGX_STREAM_UPSTREAM_LT_LAST_BYTE },
  59.     { ngx_null_string, 0 }
  60. };


  63. ‌static ngx_command_t  ngx_stream_upstream_least_time_commands[] = {

  65.     { ngx_string("least_time"),
  66.       NGX_STREAM_UPS_CONF|NGX_CONF_TAKE12,
  67.       ngx_stream_upstream_least_time,
  68.       NGX_STREAM_SRV_CONF_OFFSET,
  69.       offsetof(ngx_stream_upstream_lt_conf_t, mode),
  70.       &ngx_stream_upstream_least_time_mode },

  72.       ngx_null_command
  73. };


  76. ‌static ngx_stream_module_t  ngx_stream_upstream_least_time_module_ctx = {
  77.     NULL,                                  /* preconfiguration */
  78.     NULL,                                  /* postconfiguration */

  80.     NULL,                                  /* create main configuration */
  81.     NULL,                                  /* init main configuration */

  83.     ngx_stream_upstream_least_time_create_conf,
  84.                                            /* create server configuration */
  85.     NULL,                                  /* merge server configuration */
  86. };


  89. ‌ngx_module_t  ngx_stream_upstream_least_time_module = {
  90.     NGX_MODULE_V1,
  91.     &ngx_stream_upstream_least_time_module_ctx, /* module context */
  92.     ngx_stream_upstream_least_time_commands,    /* module directives */
  93.     NGX_STREAM_MODULE,                          /* module type */
  94.     NULL,                                       /* init master */
  95.     NULL,                                       /* init module */
  96.     NULL,                                       /* init process */
  97.     NULL,                                       /* init thread */
  98.     NULL,                                       /* exit thread */
  99.     NULL,                                       /* exit process */
  100.     NULL,                                       /* exit master */
  101.     NGX_MODULE_V1_PADDING
  102. };


  105. static ngx_int_t
  106. ‌ngx_stream_upstream_init_least_time(ngx_conf_t *cf,
  107.     ngx_stream_upstream_srv_conf_t *us)
  108. {
  109.     ngx_log_debug0(NGX_LOG_DEBUG_STREAM, cf->log, 0,
  110.                    "init least time");

  112.     if (ngx_stream_upstream_init_round_robin(cf, us) != NGX_OK) {
  113.         return NGX_ERROR;
  114.     }

  116.     us->peer.init = ngx_stream_upstream_init_least_time_peer;

  118.     return NGX_OK;
  119. }


  122. static ngx_int_t
  123. ‌ngx_stream_upstream_init_least_time_peer(ngx_stream_session_t *s,
  124.     ngx_stream_upstream_srv_conf_t *us)
  125. {
  126.     ngx_stream_upstream_lt_conf_t       *ltcf;
  127.     ngx_stream_upstream_lt_peer_data_t  *ltp;

  129.     ngx_log_debug0(NGX_LOG_DEBUG_STREAM, s->connection->log, 0,
  130.                    "init least time peer");

  132.     ltp = ngx_pcalloc(s->connection->pool,
  133.                       sizeof(ngx_stream_upstream_lt_peer_data_t));
  134.     if (ltp == NULL) {
  135.         return NGX_ERROR;
  136.     }

  138.     s->upstream->peer.data = &ltp->rrp;

  140.     if (ngx_stream_upstream_init_round_robin_peer(s, us) != NGX_OK) {
  141.         return NGX_ERROR;
  142.     }

  144.     s->upstream->peer.get = ngx_stream_upstream_get_least_time_peer;
  145.     s->upstream->peer.free = ngx_stream_upstream_free_least_time_peer;

  147.     ltp->upstream = s->upstream;

  149.     ltcf = ngx_stream_conf_upstream_srv_conf(us,
  150.                                         ngx_stream_upstream_least_time_module);

  152.     if (ltcf->use_inflight) {
  153.         s->upstream->peer.notify = ngx_stream_upstream_least_time_notify;
  154.     }

  156.     ltp->conf = ltcf;

  158.     return NGX_OK;
  159. }


  162. static ngx_int_t
  163. ‌ngx_stream_upstream_get_least_time_peer(ngx_peer_connection_t *pc, void *data)
  164. {
  165.     ngx_stream_upstream_lt_peer_data_t *ltp = data;

  167.     time_t                               now;
  168.     uintptr_t                            m;
  169.     ngx_int_t                            rc, total;
  170.     ngx_uint_t                           i, n, p, many, eta, best_eta;
  171.     ngx_msec_t                           ift;
  172.     ngx_stream_upstream_rr_peer_t       *peer, *best;
  173.     ngx_stream_upstream_rr_peers_t      *peers;
  174.     ngx_stream_upstream_rr_peer_data_t  *rrp;

  176.     ngx_log_debug1(NGX_LOG_DEBUG_STREAM, pc->log, 0,
  177.                    "get least time peer, try: %ui", pc->tries);

  179.     rrp = &ltp->rrp;

  181.     if (rrp->peers->single) {
  182.         return ngx_stream_upstream_get_round_robin_peer(pc, rrp);
  183.     }

  185.     pc->cached = 0;
  186.     pc->connection = NULL;

  188.     now = ngx_time();

  190.     peers = rrp->peers;

  192.     ngx_stream_upstream_rr_peers_wlock(peers);

  194. #if (NGX_STREAM_UPSTREAM_ZONE)
  195.     if (peers->config && rrp->config != *peers->config) {
  196.         goto busy;
  197.     }
  198. #endif

  200.     best = NULL;
  201.     total = 0;

  203. #if (NGX_SUPPRESS_WARN)
  204.     many = 0;
  205.     p = 0;
  206.     best_eta = 0;
  207. #endif

  209.     for (peer = peers->peer, i = 0;
  210.          peer;
  211.          peer = peer->next, i++)
  212.     {
  213.         n = i / (8 * sizeof(uintptr_t));
  214.         m = (uintptr_t) 1 << i % (8 * sizeof(uintptr_t));

  216.         if (rrp->tried[n] & m) {
  217.             continue;
  218.         }

  220.         if (peer->down) {
  221.             continue;
  222.         }

  224.         if (peer->max_fails
  225.             && peer->fails >= peer->max_fails
  226.             && now - peer->checked <= peer->fail_timeout)
  227.         {
  228.             continue;
  229.         }

  231.         if (peer->max_conns && peer->conns >= peer->max_conns) {
  232.             continue;
  233.         }

  235.         if (peer->inflight_reqs > 0) {

  237.             ift = peer->inflight_last / peer->inflight_reqs
  238.                   + (ngx_current_msec - peer->inflight_reqs_changed);

  240.             ngx_stream_upstream_response_time_avg(&peer->inflight_time, ift);
  241.         }

  243.         /*
  244.          * select peer with least estimated time of processing; if there are
  245.          * multiple peers with the same time, select based on round-robin
  246.          */

  248.         eta = ngx_stream_upstream_least_time_eta(ltp, peer);

  250.         if (best == NULL
  251.             || eta * best->weight < best_eta * peer->weight)
  252.         {
  253.             best = peer;
  254.             best_eta = eta;
  255.             many = 0;
  256.             p = i;

  258.         } else if (eta * best->weight == best_eta * peer->weight) {
  259.             many = 1;
  260.         }
  261.     }

  263.     if (best == NULL) {
  264.         ngx_log_debug0(NGX_LOG_DEBUG_STREAM, pc->log, 0,
  265.                        "get least time peer, no peer found");

  267.         goto failed;
  268.     }

  270.     if (many) {
  271.         ngx_log_debug0(NGX_LOG_DEBUG_STREAM, pc->log, 0,
  272.                        "get least time peer, many");

  274.         for (peer = best, i = p;
  275.              peer;
  276.              peer = peer->next, i++)
  277.         {
  278.             n = i / (8 * sizeof(uintptr_t));
  279.             m = (uintptr_t) 1 << i % (8 * sizeof(uintptr_t));

  281.             if (rrp->tried[n] & m) {
  282.                 continue;
  283.             }

  285.             if (peer->down) {
  286.                 continue;
  287.             }

  289.             eta = ngx_stream_upstream_least_time_eta(ltp, peer);

  291.             if (eta * best->weight != best_eta * peer->weight) {
  292.                 continue;
  293.             }

  295.             if (peer->max_fails
  296.                 && peer->fails >= peer->max_fails
  297.                 && now - peer->checked <= peer->fail_timeout)
  298.             {
  299.                 continue;
  300.             }

  302.             if (peer->max_conns && peer->conns >= peer->max_conns) {
  303.                 continue;
  304.             }

  306.             peer->current_weight += peer->effective_weight;
  307.             total += peer->effective_weight;

  309.             if (peer->effective_weight < peer->weight) {
  310.                 peer->effective_weight++;
  311.             }

  313.             if (peer->current_weight > best->current_weight) {
  314.                 best = peer;
  315.                 p = i;
  316.             }
  317.         }
  318.     }

  320.     best->current_weight -= total;

  322.     if (ltp->conf->use_inflight) {
  323.         if (best->inflight_reqs > 0) {
  324.             /* account time spent by inflight requests */
  325.             best->inflight_last +=
  326.                                (ngx_current_msec - best->inflight_reqs_changed)
  327.                                * best->inflight_reqs;
  328.         }

  330.         best->inflight_reqs_changed = ngx_current_msec;
  331.         best->inflight_reqs++;

  333.         ltp->inflight = 1;
  334.     }

  336.     if (now - best->checked > best->fail_timeout) {
  337.         best->checked = now;
  338.     }

  340.     pc->sockaddr = best->sockaddr;
  341.     pc->socklen = best->socklen;
  342.     pc->name = &best->name;

  344.     best->conns++;

  346.     rrp->current = best;
  347.     ngx_stream_upstream_rr_peer_ref(peers, best);

  349.     n = p / (8 * sizeof(uintptr_t));
  350.     m = (uintptr_t) 1 << p % (8 * sizeof(uintptr_t));

  352.     rrp->tried[n] |= m;

  354.     ngx_stream_upstream_rr_peers_unlock(peers);

  356.     return NGX_OK;

  358. failed:

  360.     if (peers->next) {
  361.         ngx_log_debug0(NGX_LOG_DEBUG_STREAM, pc->log, 0,
  362.                        "get least time peer, backup servers");

  364.         rrp->peers = peers->next;

  366.         n = (rrp->peers->number + (8 * sizeof(uintptr_t) - 1))
  367.                 / (8 * sizeof(uintptr_t));

  369.         for (i = 0; i < n; i++) {
  370.              rrp->tried[i] = 0;
  371.         }

  373.         ngx_stream_upstream_rr_peers_unlock(peers);

  375.         rc = ngx_stream_upstream_get_least_time_peer(pc, rrp);

  377.         if (rc != NGX_BUSY) {
  378.             return rc;
  379.         }

  381.         ngx_stream_upstream_rr_peers_wlock(peers);
  382.     }

  384. #if (NGX_STREAM_UPSTREAM_ZONE)
  385. busy:
  386. #endif

  388.     ngx_stream_upstream_rr_peers_unlock(peers);

  390.     pc->name = peers->name;

  392.     return NGX_BUSY;
  393. }


  396. static ngx_uint_t
  397. ‌ngx_stream_upstream_least_time_eta(ngx_stream_upstream_lt_peer_data_t *ltp,
  398.     ngx_stream_upstream_rr_peer_t *peer)
  399. {
  400.     time_t      now;
  401.     ngx_msec_t  rt;

  403.     switch (ltp->conf->mode) {

  405.     case NGX_STREAM_UPSTREAM_LT_FIRST_BYTE:
  406.         rt = peer->first_byte_time;
  407.         break;

  409.     case NGX_STREAM_UPSTREAM_LT_CONNECT:
  410.         rt = peer->connect_time;
  411.         break;

  413.     default: /* NGX_STREAM_UPSTREAM_LT_LAST_BYTE */
  414.         rt = peer->response_time;
  415.     }

  417.     now = ngx_time();

  419.     if (now - peer->checked > peer->fail_timeout) {
  420.         /*
  421.          * once in fail_timeout make response time of a peer 2 times
  422.          * lower to give chances to slow peers
  423.          */
  424.         rt >>= (now - peer->checked) / (peer->fail_timeout + 1);
  425.     }

  427.     if (peer->inflight_reqs > 0) {
  428.         /*
  429.          * average inflight time exceeding average response time indicates
  430.          * bad (low priority) peer
  431.          */
  432.         rt = ngx_max(rt, peer->inflight_time);
  433.     }

  435.     if (rt > 5000) {
  436.         /*
  437.          * consider peers with response time greater than max equally bad
  438.          * and thus fallback to least_conns
  439.          */
  440.         rt = 5000;

  442.     } else {
  443.         /*
  444.          * divide response times into clusters to allow round-robin for peers
  445.          * with close response times
  446.          */
  447.         rt += 20 - rt % 20;
  448.     }

  450.     /*
  451.      * estimated time peer has to spend to finish processing current requests
  452.      */
  453.     return rt * (1 + peer->conns);
  454. }


  457. static void
  458. ‌ngx_stream_upstream_least_time_notify(ngx_peer_connection_t *pc, void *data,
  459.     ngx_uint_t type)
  460. {
  461.     ngx_stream_upstream_lt_peer_data_t *ltp = data;

  463.     ngx_msec_t                           last, *metric;
  464.     ngx_stream_upstream_t               *u;
  465.     ngx_stream_upstream_rr_peer_t       *peer;
  466.     ngx_stream_upstream_rr_peers_t      *peers;
  467.     ngx_stream_upstream_rr_peer_data_t  *rrp;

  469.     rrp = &ltp->rrp;

  471.     peers = rrp->peers;
  472.     peer = rrp->current;

  474.     u = ltp->upstream;

  476.     /*
  477.      * Only update average time here if needed for balancing.
  478.      * Otherwise, it will be updated in peer.free().
  479.      */

  481. #if (NGX_SUPPRESS_WARN)
  482.     last = 0;
  483. #endif
  484.     metric = NULL;

  486.     switch (type) {

  488.     case NGX_STREAM_UPSTREAM_NOTIFY_CONNECT:

  490.         if (ltp->conf->mode == NGX_STREAM_UPSTREAM_LT_CONNECT) {
  491.             last = u->state->connect_time;
  492.             metric = &peer->connect_time;
  493.         }

  495.         break;

  497.     case NGX_STREAM_UPSTREAM_NOTIFY_FIRST_BYTE:

  499.         if (ltp->conf->mode == NGX_STREAM_UPSTREAM_LT_FIRST_BYTE) {
  500.             last = u->state->first_byte_time;
  501.             metric = &peer->first_byte_time;
  502.         }

  504.         break;
  505.     }

  507.     ngx_stream_upstream_rr_peers_rlock(peers);
  508.     ngx_stream_upstream_rr_peer_lock(peers, peer);

  510.     if (metric) {
  511.         ngx_stream_upstream_response_time_avg(metric, last);

  513.         if (ltp->inflight) {
  514.             ngx_stream_upstream_least_time_inflight_done(ltp, peers, peer,
  515.                                                          last);
  516.         }
  517.     }

  519.     ngx_stream_upstream_notify_round_robin_peer_locked(pc, data, type);
  520. }


  523. static void
  524. ‌ngx_stream_upstream_least_time_inflight_done(
  525.     ngx_stream_upstream_lt_peer_data_t *ltp,
  526.     ngx_stream_upstream_rr_peers_t *peers, ngx_stream_upstream_rr_peer_t *peer,
  527.     ngx_msec_t last)
  528. {
  529.     if (peer->inflight_reqs == 1) {
  530.         /* no more inflight requests */
  531.         peer->inflight_last = 0;

  533.     } else {

  535.         /*
  536.          * account time spent by inflight requests and forget about
  537.          * request "completed" right now
  538.          */
  539.         peer->inflight_last += (ngx_current_msec - peer->inflight_reqs_changed)
  540.                                * peer->inflight_reqs - last;
  541.         peer->inflight_reqs_changed = ngx_current_msec;
  542.     }

  544.     peer->inflight_reqs--;
  545.     ltp->inflight = 0;
  546. }


  549. static void
  550. ‌ngx_stream_upstream_free_least_time_peer(ngx_peer_connection_t *pc,
  551.     void *data, ngx_uint_t state)
  552. {
  553.     ngx_stream_upstream_lt_peer_data_t *ltp = data;

  555.     ngx_stream_upstream_t               *u;
  556.     ngx_stream_upstream_rr_peer_t       *peer;
  557.     ngx_stream_upstream_rr_peers_t      *peers;
  558.     ngx_stream_upstream_rr_peer_data_t  *rrp;

  560.     ngx_log_debug0(NGX_LOG_DEBUG_STREAM, pc->log, 0, "free least time peer");

  562.     rrp = &ltp->rrp;
  563.     peers = rrp->peers;
  564.     peer = rrp->current;

  566.     u = ltp->upstream;

  568.     ngx_stream_upstream_rr_peers_rlock(peers);
  569.     ngx_stream_upstream_rr_peer_lock(peers, peer);

  571.     if (ltp->inflight) {
  572.         ngx_stream_upstream_least_time_inflight_done(ltp, peers, peer,
  573.                                                      u->state->response_time);
  574.     }

  576.     /*
  577.      * only successful attempts are accounted to mitigate preferring
  578.      * of failing peers
  579.      */
  580.     if (!(state & (NGX_PEER_FAILED|NGX_PEER_NEXT))) {
  581.         ngx_stream_upstream_response_time_avg(&peer->response_time,
  582.                                               u->state->response_time);

  584.         if (!ltp->conf->use_inflight
  585.             || ltp->conf->mode != NGX_STREAM_UPSTREAM_LT_CONNECT)
  586.         {
  587.             ngx_stream_upstream_response_time_avg(&peer->connect_time,
  588.                                                   u->state->connect_time);
  589.         }

  591.         if (u->state->first_byte_time != (ngx_msec_t) -1
  592.             && (!ltp->conf->use_inflight
  593.                 || ltp->conf->mode != NGX_STREAM_UPSTREAM_LT_FIRST_BYTE))
  594.         {
  595.             ngx_stream_upstream_response_time_avg(&peer->first_byte_time,
  596.                                                   u->state->first_byte_time);
  597.         }
  598.     }

  600.     ngx_stream_upstream_free_round_robin_peer_locked(pc, rrp, state);
  601. }


  604. static void *
  605. ‌ngx_stream_upstream_least_time_create_conf(ngx_conf_t *cf)
  606. {
  607.     ngx_stream_upstream_lt_conf_t  *conf;

  609.     conf = ngx_pcalloc(cf->pool, sizeof(ngx_stream_upstream_lt_conf_t));
  610.     if (conf == NULL) {
  611.         return NULL;
  612.     }

  614.     /*
  615.      * set by ngx_pcalloc():
  616.      *
  617.      *     conf->use_inflight = 0;
  618.      */

  620.     conf->mode = NGX_CONF_UNSET_UINT;

  622.     return conf;
  623. }


  626. static char *
  627. ‌ngx_stream_upstream_least_time(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
  628. {
  629.     ngx_str_t                       *value;
  630.     ngx_stream_upstream_lt_conf_t   *ltcf;
  631.     ngx_stream_upstream_srv_conf_t  *uscf;

  633.     uscf = ngx_stream_conf_get_module_srv_conf(cf, ngx_stream_upstream_module);

  635.     if (uscf->peer.init_upstream) {
  636.         ngx_conf_log_error(NGX_LOG_WARN, cf, 0,
  637.                            "load balancing method redefined");
  638.     }

  640.     uscf->peer.init_upstream = ngx_stream_upstream_init_least_time;

  642.     uscf->flags = NGX_STREAM_UPSTREAM_CREATE
  643.                   |NGX_STREAM_UPSTREAM_MODIFY
  644.                   |NGX_STREAM_UPSTREAM_WEIGHT
  645.                   |NGX_STREAM_UPSTREAM_MAX_CONNS
  646.                   |NGX_STREAM_UPSTREAM_MAX_FAILS
  647.                   |NGX_STREAM_UPSTREAM_FAIL_TIMEOUT
  648.                   |NGX_STREAM_UPSTREAM_DOWN
  649.                   |NGX_STREAM_UPSTREAM_BACKUP;

  651.     if (cf->args->nelts == 3) {
  652.         value = cf->args->elts;
  653.         ltcf = ngx_stream_conf_upstream_srv_conf(uscf,
  654.                                         ngx_stream_upstream_least_time_module);
  655.         if (ngx_strcmp(value[2].data, "inflight") == 0) {
  656.             ltcf->use_inflight = 1;

  658.         } else {
  659.             ngx_conf_log_error(NGX_LOG_EMERG, cf, 0,
  660.                                "invalid parameter \"%V\"", &value[2]);
  661.             return NGX_CONF_ERROR;
  662.         }
  663.     }

  665.     return ngx_conf_set_enum_slot(cf, cmd, conf);
  666. }

One Level Up Top Level