Wie ist der Verarbeitungsablauf des ereignisgesteuerten Nginx-Frameworks?

Die Methode ngx_event_process_init des Moduls ngx_event_core_module führt eine Initialisierung des Ereignismoduls durch. Dazu gehört das Festlegen des Handlers, der einem Leseereignis wie „Verbindung anfordern“ entspricht, für die Funktion ngx_event_accept und das Hinzufügen dieses Ereignisses zum Epoll-Modul. Wenn ein neues Verbindungsereignis auftritt, wird ngx_event_accept aufgerufen. Der allgemeine Prozess ist wie folgt:

Der Arbeitsprozess ruft kontinuierlich die Funktion ngx_process_events_and_timers in der Methode ngx_worker_process_cycle auf, um Ereignisse zu verarbeiten. Diese Funktion ist der allgemeine Einstiegspunkt für die Ereignisverarbeitung.

ngx_process_events_and_timers ruft ngx_process_events auf, ein Makro, das ngx_event_actions.process_events entspricht, eine globale Struktur, die 10 Funktionsschnittstellen speichert, die dem ereignisgesteuerten Modul entsprechen (hier ist das Epoll-Modul). Daher wird hier die Funktion ngx_epoll_module_ctx.actions.process_events aufgerufen, bei der es sich um die Funktion ngx_epoll_process_events zum Verarbeiten von Ereignissen handelt.

ngx_epoll_process_events ruft die Linux-Funktionsschnittstelle epoll_wait auf, um das Ereignis „Neue Verbindung“ abzurufen, und ruft dann die Handler-Handlerfunktion dieses Ereignisses auf, um dieses Ereignis zu verarbeiten.

Wie oben erwähnt, wurde der Handler auf die Funktion ngx_event_accept eingestellt, sodass ngx_event_accept für die tatsächliche Verarbeitung aufgerufen wird.

Die ngx_event_accept-Methode wird unten analysiert:

Der optimierte Code lautet wie folgt die Kommentare Die Seriennummer entspricht der Seriennummer im Bild oben:

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决定 */
}

Das „Shock Group“-Problem in Nginx

Nginx allgemein führt mehrere Arbeitsprozesse aus, die gleichzeitig denselben Port abhören. Wenn eine neue Verbindung eintrifft, weckt der Kernel alle diese Prozesse, aber nur ein Prozess kann erfolgreich eine Verbindung zum Client herstellen, was dazu führt, dass andere Prozesse beim Aufwachen viel Overhead verschwenden. Dies wird als „donnernde Herde“ bezeichnet. Die Art und Weise, wie Nginx das „Schock“-Problem löst, besteht darin, den Prozess die Mutex-Sperre ngx_accept_mutex erhalten zu lassen und den Prozess gegenseitig in einen bestimmten kritischen Abschnitt eintreten zu lassen. Im kritischen Abschnitt fügt der Prozess dem Epoll-Modul das Leseereignis hinzu, das der Verbindung entspricht, die er überwachen möchte, sodass der Arbeitsprozess reagiert, wenn ein Ereignis „Neue Verbindung“ auftritt. Dieser Vorgang des Sperrens und Hinzufügens von Ereignissen wird in der Funktion ngx_trylock_accept_mutex abgeschlossen. Wenn andere Prozesse diese Funktion ebenfalls aufrufen und Leseereignisse hinzufügen möchten, stellen sie fest, dass der Mutex von einem anderen Prozess gehalten wird und daher nur zurückgegeben werden kann. Die Ereignisse, auf die er lauscht, können nicht zum Epoll-Modul hinzugefügt werden, sodass er nicht darauf reagieren kann das Ereignis „Neue Verbindung“. Dies wirft jedoch eine Frage auf: Wann gibt der Prozess, der die Mutex-Sperre hält, die Mutex-Sperre frei? Wenn Sie warten müssen, bis alle Ereignisse verarbeitet sind, bevor Sie die Sperre aufheben, kann dies lange dauern. Während dieser Zeit können andere Workerprozesse keine neuen Verbindungen aufbauen, was offensichtlich unerwünscht ist. Die Lösung für Nginx lautet: Der Prozess, der die Mutex-Sperre über ngx_trylock_accept_mutex erhalten hat, stellt nach Erhalt des Lese-/Schreibereignisses „Ready“ und Rückkehr von epoll_wait diese Ereignisse in die Warteschlange:

Neues Verbindungsereignis Setzen Sie die vorhandene Verbindungsereignisse in die ngx_posted_accept_events-Warteschlange
Fügen Sie die vorhandenen Verbindungsereignisse in die ngx_posted_events-Warteschlange ein

Der Code lautet wie folgt:

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); /* 不需要延后，则立即处理事件 */
}

Schreiben Sie Ereignisse und führen Sie eine ähnliche Verarbeitung durch. Als nächstes verarbeitet der Prozess die Ereignisse in der Warteschlange ngx_posted_accept_events und gibt die Mutex-Sperre nach der Verarbeitung sofort frei, sodass die Zeit, die der Prozess die Sperre belegt, minimiert wird.

Lastausgleichsproblem in Nginx

Jeder Prozess in Nginx verwendet einen Schwellenwert ngx_accept_disabled für die Handhabung des Lastausgleichs, der in der Abbildung oben dargestellt ist. Initialisiert in Schritt 2:

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

Sein Anfangswert ist eine negative Zahl, und die negative Zahl ist der Absolute Der Wert entspricht 7/8 der Gesamtzahl der Verbindungen. Wenn der Schwellenwert kleiner als 0 ist, wird auf neue Verbindungsereignisse normal reagiert. Wenn der Schwellenwert größer als 0 ist, wird auf neue Verbindungsereignisse nicht mehr reagiert ngx_accept_disabled wird um 1 reduziert. Der Code lautet wie folgt:

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

#🎜 🎜#Dies zeigt, dass die Last sinkt, wenn die aktuelle Anzahl der Verbindungen eines Prozesses 7/8 der Gesamtzahl der Verbindungen erreicht, die er verarbeiten kann Der Ausgleichsmechanismus wird ausgelöst und der Prozess reagiert nicht mehr auf neue Verbindungen.

Das obige ist der detaillierte Inhalt vonWie ist der Verarbeitungsablauf des ereignisgesteuerten Nginx-Frameworks?. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!