src/event/ngx_event_connect.c

   1
   2 /*
   3  * Copyright (C) Igor Sysoev
   4  */
   5
   6
   7 #include <ngx_config.h>
   8 #include <ngx_core.h>
   9 #include <ngx_event.h>
  10 #include <ngx_event_connect.h>
  11 #include <nginx.h>
  12
  13
  14 /* AF_INET only */
  15
  16 ngx_int_t ngx_event_connect_peer(ngx_peer_connection_t *pc)
  17 {
  18     int                  rc;
  19     ngx_uint_t           instance;
  20     u_int                event;
  21     time_t               now;
  22     ngx_err_t            err;
  23     ngx_peer_t          *peer;
  24     ngx_socket_t         s;
  25     ngx_event_t         *rev, *wev;
  26     ngx_connection_t    *c;
  27     ngx_event_conf_t    *ecf;
  28     struct sockaddr_in   addr;
  29
  30     now = ngx_time();
  31
  32     /* ngx_lock_mutex(pc->peers->mutex); */
  33
  34     if (pc->peers->last_cached) {
  35
  36         /* cached connection */
  37
  38         c = pc->peers->cached[pc->peers->last_cached];
  39         pc->peers->last_cached--;
  40
  41         /* ngx_unlock_mutex(pc->peers->mutex); */
  42
  43 #if (NGX_THREADS)
  44         c->read->lock = c->read->own_lock;
  45         c->write->lock = c->write->own_lock;
  46 #endif
  47
  48         pc->connection = c;
  49         pc->cached = 1;
  50         return NGX_OK;
  51     }
  52
  53     pc->cached = 0;
  54     pc->connection = NULL;
  55
  56     if (pc->peers->number == 1) {
  57         peer = &pc->peers->peers[0];
  58
  59     } else {
  60
  61         /* there are several peers */
  62
  63         if (pc->tries == pc->peers->number) {
  64
  65             /* it's a first try - get a current peer */
  66
  67             pc->cur_peer = pc->peers->current++;
  68
  69             if (pc->peers->current >= pc->peers->number) {
  70                 pc->peers->current = 0;
  71             }
  72         }
  73
  74         if (pc->peers->max_fails == 0) {
  75             peer = &pc->peers->peers[pc->cur_peer];
  76
  77         } else {
  78
  79             /* the peers support a fault tolerance */
  80
  81             for ( ;; ) {
  82                 peer = &pc->peers->peers[pc->cur_peer];
  83
  84                 if (peer->fails <= pc->peers->max_fails) {
  85                     break;
  86                 }
  87
  88                 if (now - peer->accessed > pc->peers->fail_timeout) {
  89                     peer->fails = 0;
  90                     break;
  91                 }
  92
  93                 pc->cur_peer++;
  94
  95                 if (pc->cur_peer >= pc->peers->number) {
  96                     pc->cur_peer = 0;
  97                 }
  98
  99                 pc->tries--;
 100
 101                 if (pc->tries == 0) {
 102                     /* ngx_unlock_mutex(pc->peers->mutex); */
 103
 104                     return NGX_ERROR;
 105                 }
 106             }
 107         }
 108     }
 109
 110     /* ngx_unlock_mutex(pc->peers->mutex); */
 111
 112
 113     s = ngx_socket(AF_INET, SOCK_STREAM, IPPROTO_IP);
 114
 115     if (s == -1) {
 116         ngx_log_error(NGX_LOG_ALERT, pc->log, ngx_socket_errno,
 117                       ngx_socket_n " failed");
 118         return NGX_ERROR;
 119     }
 120
 121
 122     ecf = ngx_event_get_conf(ngx_cycle->conf_ctx, ngx_event_core_module);
 123
 124     /* disable warning: Win32 SOCKET is u_int while UNIX socket is int */
 125
 126     if ((ngx_uint_t) s >= ecf->connections) {
 127
 128         ngx_log_error(NGX_LOG_ALERT, pc->log, 0,
 129                       "socket() returned socket #%d while only %d "
 130                       "connections was configured, closing the socket",
 131                       s, ecf->connections);
 132
 133         if (ngx_close_socket(s) == -1) {
 134             ngx_log_error(NGX_LOG_ALERT, pc->log, ngx_socket_errno,
 135                           ngx_close_socket_n "failed");
 136         }
 137
 138         /* TODO: sleep for some time */
 139
 140         return NGX_ERROR;
 141     }
 142
 143
 144     if (pc->rcvbuf) {
 145         if (setsockopt(s, SOL_SOCKET, SO_RCVBUF,
 146                        (const void *) &pc->rcvbuf, sizeof(int)) == -1) {
 147             ngx_log_error(NGX_LOG_ALERT, pc->log, ngx_socket_errno,
 148                           "setsockopt(SO_RCVBUF) failed");
 149
 150             if (ngx_close_socket(s) == -1) {
 151                 ngx_log_error(NGX_LOG_ALERT, pc->log, ngx_socket_errno,
 152                               ngx_close_socket_n " failed");
 153             }
 154
 155             return NGX_ERROR;
 156         }
 157     }
 158
 159     if (ngx_nonblocking(s) == -1) {
 160         ngx_log_error(NGX_LOG_ALERT, pc->log, ngx_socket_errno,
 161                       ngx_nonblocking_n " failed");
 162
 163         if (ngx_close_socket(s) == -1) {
 164             ngx_log_error(NGX_LOG_ALERT, pc->log, ngx_socket_errno,
 165                           ngx_close_socket_n " failed");
 166         }
 167
 168         return NGX_ERROR;
 169     }
 170
 171 #if (NGX_WIN32)
 172     /*
 173      * Winsock assignes a socket number divisible by 4
 174      * so to find a connection we divide a socket number by 4.
 175      */
 176
 177     if (s % 4) {
 178         ngx_log_error(NGX_LOG_EMERG, pc->log, 0,
 179                       ngx_socket_n
 180                       " created socket %d, not divisible by 4", s);
 181         exit(1);
 182     }
 183
 184     c = &ngx_cycle->connections[s / 4];
 185     rev = &ngx_cycle->read_events[s / 4];
 186     wev = &ngx_cycle->write_events[s / 4];
 187
 188 #else
 189
 190     c = &ngx_cycle->connections[s];
 191     rev = &ngx_cycle->read_events[s];
 192     wev = &ngx_cycle->write_events[s];
 193
 194 #endif
 195
 196     instance = rev->instance;
 197
 198 #if (NGX_THREADS)
 199
 200     if (*(&c->lock)) {
 201         ngx_log_debug1(NGX_LOG_DEBUG_EVENT, pc->log, 0,
 202                        "spinlock in connect, fd:%d", s);
 203         ngx_spinlock(&c->lock, 1000);
 204         ngx_unlock(&c->lock);
 205     }
 206
 207 #endif
 208
 209     ngx_memzero(c, sizeof(ngx_connection_t));
 210     ngx_memzero(rev, sizeof(ngx_event_t));
 211     ngx_memzero(wev, sizeof(ngx_event_t));
 212
 213     rev->instance = !instance;
 214     wev->instance = !instance;
 215
 216     rev->index = NGX_INVALID_INDEX;
 217     wev->index = NGX_INVALID_INDEX;
 218
 219     rev->data = c;
 220     wev->data = c;
 221
 222     c->read = rev;
 223     c->write = wev;
 224     wev->write = 1;
 225
 226     c->log = pc->log;
 227     rev->log = pc->log;
 228     wev->log = pc->log;
 229
 230     c->fd = s;
 231
 232     c->log_error = pc->log_error;
 233
 234     pc->connection = c;
 235
 236     /*
 237      * TODO: MT: - atomic increment (x86: lock xadd)
 238      *             or protection by critical section or mutex
 239      *
 240      * TODO: MP: - allocated in a shared memory
 241      *           - atomic increment (x86: lock xadd)
 242      *             or protection by critical section or mutex
 243      */
 244
 245     c->number = ngx_atomic_inc(ngx_connection_counter);
 246
 247 #if (NGX_THREADS)
 248     rev->lock = pc->lock;
 249     wev->lock = pc->lock;
 250     rev->own_lock = &c->lock;
 251     wev->own_lock = &c->lock;
 252 #endif
 253
 254     if (ngx_add_conn) {
 255         if (ngx_add_conn(c) == NGX_ERROR) {
 256             return NGX_ERROR;
 257         }
 258     }
 259
 260     ngx_memzero(&addr, sizeof(struct sockaddr_in));
 261
 262     addr.sin_family = AF_INET;
 263     addr.sin_port = peer->port;
 264     addr.sin_addr.s_addr = peer->addr;
 265
 266     ngx_log_debug2(NGX_LOG_DEBUG_EVENT, pc->log, 0,
 267                    "connect to %V, #%d", &peer->addr_port_text, c->number);
 268
 269     rc = connect(s, (struct sockaddr *) &addr, sizeof(struct sockaddr_in));
 270
 271     if (rc == -1) {
 272         err = ngx_socket_errno;
 273
 274         /* Winsock returns WSAEWOULDBLOCK (NGX_EAGAIN) */
 275
 276         if (err != NGX_EINPROGRESS && err != NGX_EAGAIN) {
 277             ngx_connection_error(c, err, "connect() failed");
 278
 279             if (ngx_close_socket(s) == -1) {
 280                 ngx_log_error(NGX_LOG_ALERT, pc->log, ngx_socket_errno,
 281                               ngx_close_socket_n " failed");
 282             }
 283
 284             c->fd = (ngx_socket_t) -1;
 285
 286             return NGX_CONNECT_ERROR;
 287         }
 288     }
 289
 290     if (ngx_add_conn) {
 291         if (rc == -1) {
 292             /* NGX_EINPROGRESS */
 293             return NGX_AGAIN;
 294         }
 295
 296         ngx_log_debug0(NGX_LOG_DEBUG_EVENT, pc->log, 0, "connected");
 297         return NGX_OK;
 298     }
 299
 300     if (ngx_event_flags & NGX_USE_AIO_EVENT) {
 301
 302         ngx_log_debug1(NGX_LOG_DEBUG_EVENT, pc->log, ngx_socket_errno,
 303                        "connect(): %d", rc);
 304
 305         /* aio, iocp */
 306
 307         if (ngx_blocking(s) == -1) {
 308             ngx_log_error(NGX_LOG_ALERT, pc->log, ngx_socket_errno,
 309                           ngx_blocking_n " failed");
 310
 311             if (ngx_close_socket(s) == -1) {
 312                 ngx_log_error(NGX_LOG_ALERT, pc->log, ngx_socket_errno,
 313                               ngx_close_socket_n " failed");
 314             }
 315
 316             return NGX_ERROR;
 317         }
 318
 319         /*
 320          * FreeBSD aio allows to post operation on non-connected socket.
 321          * NT does not support it.
 322          *
 323          * TODO: check in Win32, etc. As workaround we can use NGX_ONESHOT_EVENT
 324          */
 325
 326         rev->ready = 1;
 327         wev->ready = 1;
 328
 329         return NGX_OK;
 330     }
 331
 332     if (ngx_event_flags & NGX_USE_CLEAR_EVENT) {     /* kqueue */
 333         event = NGX_CLEAR_EVENT;
 334
 335     } else {                                  /* select, poll, /dev/poll */
 336         event = NGX_LEVEL_EVENT;
 337     }
 338
 339     if (ngx_add_event(rev, NGX_READ_EVENT, event) != NGX_OK) {
 340         return NGX_ERROR;
 341     }
 342
 343     if (rc == -1) {
 344
 345         /* NGX_EINPROGRESS */
 346
 347         if (ngx_add_event(wev, NGX_WRITE_EVENT, event) != NGX_OK) {
 348             return NGX_ERROR;
 349         }
 350
 351         return NGX_AGAIN;
 352     }
 353
 354     ngx_log_debug0(NGX_LOG_DEBUG_EVENT, pc->log, 0, "connected");
 355
 356     wev->ready = 1;
 357
 358     return NGX_OK;
 359 }
 360
 361
 362 void ngx_event_connect_peer_failed(ngx_peer_connection_t *pc)
 363 {
 364     time_t  now;
 365
 366     now = ngx_time();
 367
 368     /* ngx_lock_mutex(pc->peers->mutex); */
 369
 370     pc->peers->peers[pc->cur_peer].fails++;
 371     pc->peers->peers[pc->cur_peer].accessed = now;
 372
 373     /* ngx_unlock_mutex(pc->peers->mutex); */
 374
 375     pc->cur_peer++;
 376
 377     if (pc->cur_peer >= pc->peers->number) {
 378         pc->cur_peer = 0;
 379     }
 380
 381     pc->tries--;
 382
 383     return;
 384 }