nginx-0.0.1-2003-05-12-19:52:24 import
diff --git a/src/event/modules/ngx_kqueue_module.c b/src/event/modules/ngx_kqueue_module.c
index 4fe9e6b..0c2ce35 100644
--- a/src/event/modules/ngx_kqueue_module.c
+++ b/src/event/modules/ngx_kqueue_module.c
@@ -141,7 +141,7 @@
if (nchanges > 0
&& ev->index < nchanges
- && change_list[ev->index].udata == ev)
+ && (void *) ((uintptr_t) change_list[ev->index].udata & ~1) == ev)
{
#if (NGX_DEBUG_EVENT)
ngx_connection_t *c = (ngx_connection_t *) ev->data;
@@ -159,8 +159,9 @@
return NGX_OK;
}
- /* when a socket is closed kqueue automatically deletes its filters
- so we do not need to delete a event explicity before a socket closing */
+ /* when the file descriptor is closed a kqueue automatically deletes
+ its filters so we do not need to delete explicity the event
+ before the closing the file descriptor */
if (flags & NGX_CLOSE_EVENT) {
return NGX_OK;
@@ -200,7 +201,7 @@
change_list[nchanges].ident = c->fd;
change_list[nchanges].filter = filter;
change_list[nchanges].flags = flags;
- change_list[nchanges].udata = ev;
+ change_list[nchanges].udata = (void *) ((uintptr_t) ev | ev->instance);
#if (HAVE_LOWAT_EVENT)
@@ -230,7 +231,7 @@
int ngx_kqueue_process_events(ngx_log_t *log)
{
- int events, i;
+ int events, instance, i;
ngx_msec_t timer, delta;
ngx_event_t *ev;
struct timeval tv;
@@ -310,12 +311,15 @@
}
ev = (ngx_event_t *) event_list[i].udata;
+ instance = (uintptr_t) ev & 1;
+ ev = (void *) ((uintptr_t) ev & ~1);
- /* It's a stale event from a socket
+ /* It's a stale event from a file descriptor
that was just closed in this iteration */
- if (!ev->active) {
- continue;
+ if (ev->active == 0 || ev->instance != instance) {
+ ngx_log_debug(log, "stale kevent");
+ continue;
}
switch (event_list[i].filter) {
@@ -323,28 +327,6 @@
case EVFILT_READ:
case EVFILT_WRITE:
- if (ev->first) {
- if (nchanges > 0
- && ev->index < nchanges
- && change_list[ev->index].udata == ev) {
-
- /* It's a stale event from a socket that was just closed
- in this iteration and during processing another socket
- was opened with the same number by accept() or socket()
- and its event has been added the event to the change_list
- but has not been passed to a kernel. Nevertheless
- there's small chance that ngx_kqueue_set_event() has
- flushed the new event if the change_list was filled up.
- In this very rare case we would get EAGAIN while
- a reading or a writing */
-
- continue;
-
- } else {
- ev->first = 0;
- }
- }
-
ev->available = event_list[i].data;
if (event_list[i].flags & EV_EOF) {
diff --git a/src/event/ngx_event.h b/src/event/ngx_event.h
index c3cc907..bc8cae5 100644
--- a/src/event/ngx_event.h
+++ b/src/event/ngx_event.h
@@ -63,7 +63,9 @@
#endif
unsigned write:1;
- unsigned first:1;
+ unsigned instance:1; /* used to detect stale events in kqueue,
+ rt signals and epoll */
+
unsigned active:1;
unsigned ready:1;
unsigned timedout:1;
@@ -114,6 +116,7 @@
#endif
};
+
typedef enum {
NGX_SELECT_EVENT_N = 0,
#if (HAVE_POLL)
diff --git a/src/event/ngx_event_accept.c b/src/event/ngx_event_accept.c
index aa2911c..411f2b1 100644
--- a/src/event/ngx_event_accept.c
+++ b/src/event/ngx_event_accept.c
@@ -16,6 +16,7 @@
int ngx_event_accept(ngx_event_t *ev)
{
+ int instance;
socklen_t len;
struct sockaddr *sa;
ngx_err_t err;
@@ -95,6 +96,8 @@
wev = &ngx_write_events[s];
c = &ngx_connections[s];
+ instance = rev->instance;
+
ngx_memzero(rev, sizeof(ngx_event_t));
ngx_memzero(wev, sizeof(ngx_event_t));
ngx_memzero(c, sizeof(ngx_connection_t));
@@ -108,6 +111,8 @@
c->addr_text_max_len = ls->addr_text_max_len;
c->post_accept_timeout = ls->post_accept_timeout;
+ rev->instance = wev->instance = !instance;
+
rev->index = wev->index = NGX_INVALID_INDEX;
rev->data = wev->data = c;
@@ -117,7 +122,6 @@
c->fd = s;
c->unexpected_eof = 1;
wev->write = 1;
- rev->first = wev->first = 1;
#if (USE_KQUEUE)
wev->ready = 1;
diff --git a/src/event/ngx_event_connect.c b/src/event/ngx_event_connect.c
new file mode 100644
index 0000000..ca8a4a3
--- /dev/null
+++ b/src/event/ngx_event_connect.c
@@ -0,0 +1,126 @@
+
+#include <ngx_event_connect.h>
+
+
+int ngx_event_connect_peer(ngx_connect_peer_t *cp)
+{
+
+
+ if (cp->peers->number > 1) {
+
+ /* it's a first try - get current peer */
+
+ if (cp->tries == cp->peers->number) {
+
+ /* Here is the race condition
+ when the peers are shared between
+ the threads or the processes but it should not be serious */
+
+ cp->cur_peer = cp->peers->current++;
+
+ if (cp->peers->current >= cp->peers->number) {
+ cp->peers->current = 0;
+ }
+
+ /* */
+
+#if (NGX_MULTITHREADED || NGX_MULTIPROCESSED)
+ /* eliminate the sequences of the race condition */
+ if (cp->cur_peer >= cp->peers->number) {
+ cp->cur_peer = 0;
+ }
+#endif
+ }
+
+ if (cp->peers->max_fails > 0) {
+
+ for ( ;; ) {
+ peer = &cp->peers->peers[cp->cur_peer];
+
+ /* Here is the race condition
+ when the peers are shared between
+ the threads or the processes but it should not be serious */
+
+ if (peer->fails <= cp->peers->max_fails
+ || (now - peer->accessed > cp->peers->fail_timeout))
+ {
+ break;
+ }
+
+ /* */
+
+ cp->cur_peer++;
+
+ if (cp->cur_peer >= cp->peers->number) {
+ cp->cur_peer = 0;
+ }
+
+ cp->tries--;
+
+ if (cp->tries == 0) {
+ return NGX_ERROR;
+ }
+ }
+ }
+ }
+
+
+
+
+
+ s = ngx_socket(AF_INET, SOCK_STREAM, IPPROTO_IP, 0);
+
+ if (s == -1) {
+ ngx_log_error(NGX_LOG_ALERT, cn->log, ngx_socket_errno,
+ ngx_socket_n " failed");
+ return NGX_HTTP_INTERNAL_SERVER_ERROR;
+ }
+
+ if (cn->rcvbuf) {
+ if (setsockopt(s, SOL_SOCKET, SO_RCVBUF,
+ (const void *) &cn->rcvbuf, sizeof(int)) == -1) {
+ ngx_log_error(NGX_LOG_ALERT, cn->log, ngx_socket_errno,
+ "setsockopt(SO_RCVBUF) failed");
+
+ if (ngx_close_socket(s) == -1) {
+ ngx_log_error(NGX_LOG_ALERT, cn->log, ngx_socket_errno,
+ ngx_close_socket_n " failed");
+ }
+
+ return NGX_HTTP_INTERNAL_SERVER_ERROR;
+ }
+ }
+
+ if (ngx_nonblocking(s) == -1) {
+ ngx_log_error(NGX_LOG_ALERT, cn->log, ngx_socket_errno,
+ ngx_nonblocking_n " failed");
+
+ if (ngx_close_socket(s) == -1) {
+ ngx_log_error(NGX_LOG_ALERT, cn->log, ngx_socket_errno,
+ ngx_close_socket_n " failed");
+ }
+
+ return NGX_HTTP_INTERNAL_SERVER_ERROR;
+ }
+
+ c = &ngx_connections[s];
+ rev = &ngx_read_events[s];
+ wev = &ngx_write_events[s];
+
+ instance = rev->instance;
+
+ ngx_memzero(c, sizeof(ngx_connection_t));
+ ngx_memzero(rev, sizeof(ngx_event_t));
+ ngx_memzero(wev, sizeof(ngx_event_t));
+
+ rev->index = wev->index = NGX_INVALID_INDEX;
+ rev->data = wev->data = c;
+ c->read = rev;
+ c->write = wev;
+
+ rev->instance = wev->instance = !instance;
+
+ rev->log = wev->log = c->log = cn->log;
+ c->fd = s;
+ wev->close_handler = rev->close_handler = ngx_event_close_connection;
+
diff --git a/src/event/ngx_event_connect.h b/src/event/ngx_event_connect.h
new file mode 100644
index 0000000..ecfedf9
--- /dev/null
+++ b/src/event/ngx_event_connect.h
@@ -0,0 +1,43 @@
+#ifndef _NGX_EVENT_CONNECT_H_INCLUDED_
+#define _NGX_EVENT_CONNECT_H_INCLUDED_
+
+
+#include <ngx_config.h>
+#include <ngx_string.h>
+#include <ngx_log.h>
+#include <ngx_event.h>
+#include <ngx_connection.h>
+
+
+typedef struct {
+ u_int32_t addr;
+ ngx_str_t host;
+ int port;
+ ngx_str_t addr_port_name;
+
+ int fails;
+ time_t accessed;
+} ngx_peer_t;
+
+
+typedef struct {
+ int current;
+ int number;
+ int max_fails;
+ int fail_timeout;
+
+ /* ngx_mutex_t *mutex; */
+ /* ngx_connection_t *cached; */
+
+ ngx_peer_t peers[1];
+} ngx_peers_t;
+
+
+typedef struct {
+ ngx_peers_t *peers;
+ int cur_peer;
+ int tries;
+} ngx_connect_peer_t;
+
+
+#endif /* _NGX_EVENT_CONNECT_H_INCLUDED_ */
diff --git a/src/event/ngx_event_timer.c b/src/event/ngx_event_timer.c
index aebbaf8..f32ce47 100644
--- a/src/event/ngx_event_timer.c
+++ b/src/event/ngx_event_timer.c
@@ -45,7 +45,8 @@
#if (NGX_DEBUG_EVENT)
ngx_connection_t *c = (ngx_connection_t *) ev->data;
- ngx_log_debug(ev->log, "set timer: %d:%d" _ c->fd _ timer);
+ ngx_log_debug(ev->log, "set timer: %d:%d, slot: %d" _
+ c->fd _ timer _ ngx_timer_cur_queue);
#endif
if (ev->timer_next || ev->timer_prev) {
@@ -53,10 +54,6 @@
return;
}
-#if (NGX_DEBUG_EVENT)
- ngx_log_debug(ev->log, "timer slot: %d" _ ngx_timer_cur_queue);
-#endif
-
for (e = ngx_timer_queue[ngx_timer_cur_queue].timer_next;
e != &ngx_timer_queue[ngx_timer_cur_queue] && timer > e->timer_delta;
e = e->timer_next)
