Quel est le flux de traitement du framework événementiel Nginx ?

La méthode ngx_event_process_init du module ngx_event_core_module effectue une initialisation du module d'événement. Cela inclut la définition du gestionnaire correspondant à un événement de lecture tel que "demande de connexion" à la fonction ngx_event_accept et l'ajout de cet événement au module epoll. Lorsqu'un nouvel événement de connexion se produit, ngx_event_accept sera appelé. Le processus général est le suivant : Le processus de travail appelle en permanence la fonction ngx_process_events_and_timers dans la méthode ngx_worker_process_cycle pour traiter les événements. Cette fonction est le point d'entrée général pour le traitement des événements.

ngx_process_events_and_timers appellera ngx_process_events, qui est une macro, équivalente à ngx_event_actions.process_events est une structure globale qui stocke 10 interfaces de fonction correspondant au module événementiel (voici le module epoll). Ainsi, la fonction ngx_epoll_module_ctx.actions.process_events est appelée ici, qui est la fonction ngx_epoll_process_events pour traiter les événements.

ngx_epoll_process_events appelle l'interface de fonction Linux epoll_wait pour obtenir l'événement "nouvelle connexion", puis appelle la fonction de traitement du gestionnaire de cet événement pour traiter cet événement.

Comme mentionné ci-dessus, le gestionnaire a été défini sur la fonction ngx_event_accept, donc ngx_event_accept est appelé pour le traitement réel.

La méthode ngx_event_accept est analysée ci-dessous. Son organigramme est le suivant :

Le numéro de série dans le commentaire correspond au numéro de série dans l'image ci-dessus :

void
ngx_event_accept(ngx_event_t *ev)
{
 socklen_t  socklen;
 ngx_err_t  err;
 ngx_log_t  *log;
 ngx_uint_t  level;
 ngx_socket_t  s;
 ngx_event_t  *rev, *wev;
 ngx_listening_t  *ls;
 ngx_connection_t *c, *lc;
 ngx_event_conf_t *ecf;
 u_char  sa[ngx_sockaddrlen];
 
 if (ev->timedout) {
  if (ngx_enable_accept_events((ngx_cycle_t *) ngx_cycle) != ngx_ok) {
   return;
  }
 
  ev->timedout = 0;
 }
 
 ecf = ngx_event_get_conf(ngx_cycle->conf_ctx, ngx_event_core_module);
 
 if (ngx_event_flags & ngx_use_rtsig_event) {
  ev->available = 1;
 
 } else if (!(ngx_event_flags & ngx_use_kqueue_event)) {
  ev->available = ecf->multi_accept;
 }
 
 lc = ev->data;
 ls = lc->listening;
 ev->ready = 0;
 
 do {
  socklen = ngx_sockaddrlen;
 
  /* 1、accept方法试图建立连接，非阻塞调用 */
  s = accept(lc->fd, (struct sockaddr *) sa, &socklen);
 
  if (s == (ngx_socket_t) -1)
  {
   err = ngx_socket_errno;
 
   if (err == ngx_eagain)
   {
    /* 没有连接，直接返回 */
    return;
   }
 
   level = ngx_log_alert;
 
   if (err == ngx_econnaborted) {
    level = ngx_log_err;
 
   } else if (err == ngx_emfile || err == ngx_enfile) {
    level = ngx_log_crit;
   }
 
   if (err == ngx_econnaborted) {
    if (ngx_event_flags & ngx_use_kqueue_event) {
     ev->available--;
    }
 
    if (ev->available) {
     continue;
    }
   }
 
   if (err == ngx_emfile || err == ngx_enfile) {
    if (ngx_disable_accept_events((ngx_cycle_t *) ngx_cycle)
     != ngx_ok)
    {
     return;
    }
 
    if (ngx_use_accept_mutex) {
     if (ngx_accept_mutex_held) {
      ngx_shmtx_unlock(&ngx_accept_mutex);
      ngx_accept_mutex_held = 0;
     }
 
     ngx_accept_disabled = 1;
 
    } else {
     ngx_add_timer(ev, ecf->accept_mutex_delay);
    }
   }
 
   return;
  }
 
  /* 2、设置负载均衡阈值 */
  ngx_accept_disabled = ngx_cycle->connection_n / 8
        - ngx_cycle->free_connection_n;
 
  /* 3、从连接池获得一个连接对象 */
  c = ngx_get_connection(s, ev->log);
 
  /* 4、为连接创建内存池 */
  c->pool = ngx_create_pool(ls->pool_size, ev->log);
 
  c->sockaddr = ngx_palloc(c->pool, socklen);
 
  ngx_memcpy(c->sockaddr, sa, socklen);
 
  log = ngx_palloc(c->pool, sizeof(ngx_log_t));
 
  /* set a blocking mode for aio and non-blocking mode for others */
  /* 5、设置套接字属性为阻塞或非阻塞 */
  if (ngx_inherited_nonblocking) {
   if (ngx_event_flags & ngx_use_aio_event) {
    if (ngx_blocking(s) == -1) {
     ngx_log_error(ngx_log_alert, ev->log, ngx_socket_errno,
         ngx_blocking_n " failed");
     ngx_close_accepted_connection(c);
     return;
    }
   }
 
  } else {
   if (!(ngx_event_flags & (ngx_use_aio_event|ngx_use_rtsig_event))) {
    if (ngx_nonblocking(s) == -1) {
     ngx_log_error(ngx_log_alert, ev->log, ngx_socket_errno,
         ngx_nonblocking_n " failed");
     ngx_close_accepted_connection(c);
     return;
    }
   }
  }
 
  *log = ls->log;
 
  c->recv = ngx_recv;
  c->send = ngx_send;
  c->recv_chain = ngx_recv_chain;
  c->send_chain = ngx_send_chain;
 
  c->log = log;
  c->pool->log = log;
 
  c->socklen = socklen;
  c->listening = ls;
  c->local_sockaddr = ls->sockaddr;
  c->local_socklen = ls->socklen;
 
  c->unexpected_eof = 1;
 
  rev = c->read;
  wev = c->write;
 
  wev->ready = 1;
 
  if (ngx_event_flags & (ngx_use_aio_event|ngx_use_rtsig_event)) {
   /* rtsig, aio, iocp */
   rev->ready = 1;
  }
 
  if (ev->deferred_accept) {
   rev->ready = 1;
 
  }
 
  rev->log = log;
  wev->log = log;
 
  /*
   * todo: mt: - ngx_atomic_fetch_add()
   *  or protection by critical section or light mutex
   *
   * todo: mp: - allocated in a shared memory
   *   - ngx_atomic_fetch_add()
   *  or protection by critical section or light mutex
   */
 
  c->number = ngx_atomic_fetch_add(ngx_connection_counter, 1);
 
  if (ls->addr_ntop) {
   c->addr_text.data = ngx_pnalloc(c->pool, ls->addr_text_max_len);
   if (c->addr_text.data == null) {
    ngx_close_accepted_connection(c);
    return;
   }
 
   c->addr_text.len = ngx_sock_ntop(c->sockaddr, c->socklen,
            c->addr_text.data,
            ls->addr_text_max_len, 0);
   if (c->addr_text.len == 0) {
    ngx_close_accepted_connection(c);
    return;
   }
  }
 
  /* 6、将新连接对应的读写事件添加到epoll对象中 */
  if (ngx_add_conn && (ngx_event_flags & ngx_use_epoll_event) == 0) {
   if (ngx_add_conn(c) == ngx_error) {
    ngx_close_accepted_connection(c);
    return;
   }
  }
 
  log->data = null;
  log->handler = null;
 
  /* 7、tcp建立成功调用的方法，这个方法在ngx_listening_t结构体中 */
  ls->handler(c);
 
 } while (ev->available); /* available标志表示一次尽可能多的建立连接，由配置项multi_accept决定 */
}

nginx exécute généralement plusieurs processus de travail, et ces processus écoutent le même port en même temps. Lorsqu'une nouvelle connexion arrive, le noyau réveille tous ces processus, mais un seul processus peut se connecter avec succès au client, ce qui entraîne une perte importante de temps système pour les autres processus au réveil. C'est ce qu'on appelle le phénomène du « troupeau tonitruant ». La façon dont nginx résout le problème du "choc" est de laisser le processus obtenir le verrou mutex ngx_accept_mutex et de laisser le processus entrer mutuellement dans une certaine section critique. Dans la section critique, le processus ajoute l'événement read correspondant à la connexion qu'il souhaite surveiller au module epoll, de sorte que lorsqu'un événement "nouvelle connexion" se produit, le processus de travail répond. Ce processus de verrouillage et d'ajout d'événements est complété dans la fonction ngx_trylock_accept_mutex. Lorsque d'autres processus entrent également dans cette fonction et souhaitent ajouter des événements de lecture, ils découvrent que le mutex est détenu par un autre processus, il ne peut donc que revenir, et les événements qu'il écoute ne peuvent pas être ajoutés au module epoll, il ne peut donc pas répondre à l'événement "nouvelle connexion". Mais cela soulèvera une question : quand le processus détenant le verrou mutex libère-t-il le verrou mutex ? Si vous devez attendre que tous les événements soient traités avant de déverrouiller le verrou, cela prendra beaucoup de temps. Pendant cette période, les autres processus de travail ne peuvent pas établir de nouvelles connexions, ce qui est évidemment indésirable. La solution pour nginx est la suivante : le processus qui a obtenu le verrouillage mutex via ngx_trylock_accept_mutex, après avoir obtenu l'événement de lecture/écriture prêt et être revenu de epoll_wait, met ces événements dans la file d'attente :

Les nouveaux événements de connexion sont placés dans la file d'attente ngx_posted_accept_events

Le code est le suivant :

if (flags & ngx_post_events)
{
 /* 延后处理这批事件 */
 queue = (ngx_event_t **) (rev->accept ? &ngx_posted_accept_events : &ngx_posted_events);
 
 /* 将事件添加到延后执行队列中 */
 ngx_locked_post_event(rev, queue);
}
else
{
 rev->handler(rev); /* 不需要延后，则立即处理事件 */
}

Écrivez des événements pour un traitement similaire. Le processus traite ensuite les événements dans la file d'attente ngx_posted_accept_events et libère immédiatement le verrou mutex après le traitement, minimisant ainsi le temps pendant lequel le processus prend le verrou.

Chaque processus dans nginx utilise un seuil ngx_accept_disabled pour gérer l'équilibrage de charge, qui est initialisé à l'étape 2 de la figure ci-dessus :

ngx_accept_disabled = ngx_cycle->connection_n / 8 - ngx_cycle-> ;free_connection_n;

Sa valeur initiale est un nombre négatif, et la valeur absolue du nombre négatif est égale à 7/8 du nombre total de connexions. Lorsque le seuil est inférieur à 0, il répondra normalement aux nouveaux événements de connexion. , et lorsque le seuil est supérieur à 0, il ne répondra plus. Nouvel événement de connexion, et décrémenter ngx_accept_disabled de 1, le code est le suivant :

if (ngx_accept_disabled > 0)
{
  ngx_accept_disabled--;
}
else
{
 if (ngx_trylock_accept_mutex(cycle) == ngx_error)
 {
  return;
 }
 ....
}

Cela montre que lorsque le nombre actuel de connexions d'un processus atteint 7/ 8 du nombre total de connexions pouvant être traitées, le mécanisme d'équilibrage de charge est déclenché et le processus cesse de répondre aux nouvelles connexions.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!