Mercurial > hg > nginx
annotate src/stream/ngx_stream_upstream.c @ 6435:d1c791479bbb
Stream: post first read events from client and upstream.
The main proxy function ngx_stream_proxy_process() can terminate the stream
session. The code, following it, should check its return code to make sure the
session still exists. This happens in client and upstream initialization
functions. Swapping ngx_stream_proxy_process() call with the code, that
follows it, leaves the same problem vice versa.
In future ngx_stream_proxy_process() will call ngx_stream_proxy_next_upstream()
making it too complicated to know if stream session still exists after this
call.
Now ngx_stream_proxy_process() is called from posted event handlers in both
places with no code following it. The posted event is automatically removed
once session is terminated.
| author | Roman Arutyunyan <arut@nginx.com> |
|---|---|
| date | Tue, 15 Mar 2016 15:55:23 +0300 |
| parents | cf5e822cf470 |
| children | 78fc2dce69e7 |
| rev | line source |
|---|---|
| 6115 | 1 |
| 2 /* | |
| 3 * Copyright (C) Igor Sysoev | |
| 4 * Copyright (C) Nginx, Inc. | |
| 5 */ | |
| 6 | |
| 7 | |
| 8 #include <ngx_config.h> | |
| 9 #include <ngx_core.h> | |
| 10 #include <ngx_stream.h> | |
| 11 | |
| 12 | |
| 13 static char *ngx_stream_upstream(ngx_conf_t *cf, ngx_command_t *cmd, | |
| 14 void *dummy); | |
| 15 static char *ngx_stream_upstream_server(ngx_conf_t *cf, ngx_command_t *cmd, | |
| 16 void *conf); | |
| 17 static void *ngx_stream_upstream_create_main_conf(ngx_conf_t *cf); | |
| 18 static char *ngx_stream_upstream_init_main_conf(ngx_conf_t *cf, void *conf); | |
| 19 | |
| 20 | |
| 21 static ngx_command_t ngx_stream_upstream_commands[] = { | |
| 22 | |
| 23 { ngx_string("upstream"), | |
| 24 NGX_STREAM_MAIN_CONF|NGX_CONF_BLOCK|NGX_CONF_TAKE1, | |
| 25 ngx_stream_upstream, | |
| 26 0, | |
| 27 0, | |
| 28 NULL }, | |
| 29 | |
| 30 { ngx_string("server"), | |
| 31 NGX_STREAM_UPS_CONF|NGX_CONF_1MORE, | |
| 32 ngx_stream_upstream_server, | |
| 33 NGX_STREAM_SRV_CONF_OFFSET, | |
| 34 0, | |
| 35 NULL }, | |
| 36 | |
| 37 ngx_null_command | |
| 38 }; | |
| 39 | |
| 40 | |
| 41 static ngx_stream_module_t ngx_stream_upstream_module_ctx = { | |
|
6174
68c106e6fa0a
Stream: added postconfiguration method to stream modules.
Vladimir Homutov <vl@nginx.com>
parents:
6115
diff
changeset
|
42 NULL, /* postconfiguration */ |
|
68c106e6fa0a
Stream: added postconfiguration method to stream modules.
Vladimir Homutov <vl@nginx.com>
parents:
6115
diff
changeset
|
43 |
| 6115 | 44 ngx_stream_upstream_create_main_conf, /* create main configuration */ |
| 45 ngx_stream_upstream_init_main_conf, /* init main configuration */ | |
| 46 | |
| 47 NULL, /* create server configuration */ | |
| 48 NULL, /* merge server configuration */ | |
| 49 }; | |
| 50 | |
| 51 | |
| 52 ngx_module_t ngx_stream_upstream_module = { | |
| 53 NGX_MODULE_V1, | |
| 54 &ngx_stream_upstream_module_ctx, /* module context */ | |
| 55 ngx_stream_upstream_commands, /* module directives */ | |
| 56 NGX_STREAM_MODULE, /* module type */ | |
| 57 NULL, /* init master */ | |
| 58 NULL, /* init module */ | |
| 59 NULL, /* init process */ | |
| 60 NULL, /* init thread */ | |
| 61 NULL, /* exit thread */ | |
| 62 NULL, /* exit process */ | |
| 63 NULL, /* exit master */ | |
| 64 NGX_MODULE_V1_PADDING | |
| 65 }; | |
| 66 | |
| 67 | |
| 68 static char * | |
| 69 ngx_stream_upstream(ngx_conf_t *cf, ngx_command_t *cmd, void *dummy) | |
| 70 { | |
| 71 char *rv; | |
| 72 void *mconf; | |
| 73 ngx_str_t *value; | |
| 74 ngx_url_t u; | |
| 75 ngx_uint_t m; | |
| 76 ngx_conf_t pcf; | |
| 77 ngx_stream_module_t *module; | |
| 78 ngx_stream_conf_ctx_t *ctx, *stream_ctx; | |
| 79 ngx_stream_upstream_srv_conf_t *uscf; | |
| 80 | |
| 81 ngx_memzero(&u, sizeof(ngx_url_t)); | |
| 82 | |
| 83 value = cf->args->elts; | |
| 84 u.host = value[1]; | |
| 85 u.no_resolve = 1; | |
| 86 u.no_port = 1; | |
| 87 | |
| 88 uscf = ngx_stream_upstream_add(cf, &u, NGX_STREAM_UPSTREAM_CREATE | |
| 89 |NGX_STREAM_UPSTREAM_WEIGHT | |
| 90 |NGX_STREAM_UPSTREAM_MAX_FAILS | |
| 91 |NGX_STREAM_UPSTREAM_FAIL_TIMEOUT | |
| 92 |NGX_STREAM_UPSTREAM_DOWN | |
| 93 |NGX_STREAM_UPSTREAM_BACKUP); | |
| 94 if (uscf == NULL) { | |
| 95 return NGX_CONF_ERROR; | |
| 96 } | |
| 97 | |
| 98 | |
| 99 ctx = ngx_pcalloc(cf->pool, sizeof(ngx_stream_conf_ctx_t)); | |
| 100 if (ctx == NULL) { | |
| 101 return NGX_CONF_ERROR; | |
| 102 } | |
| 103 | |
| 104 stream_ctx = cf->ctx; | |
| 105 ctx->main_conf = stream_ctx->main_conf; | |
| 106 | |
| 107 /* the upstream{}'s srv_conf */ | |
| 108 | |
| 109 ctx->srv_conf = ngx_pcalloc(cf->pool, | |
| 110 sizeof(void *) * ngx_stream_max_module); | |
| 111 if (ctx->srv_conf == NULL) { | |
| 112 return NGX_CONF_ERROR; | |
| 113 } | |
| 114 | |
| 115 ctx->srv_conf[ngx_stream_upstream_module.ctx_index] = uscf; | |
| 116 | |
| 117 uscf->srv_conf = ctx->srv_conf; | |
| 118 | |
|
6379
cf5e822cf470
Dynamic modules: changed ngx_modules to cycle->modules.
Maxim Dounin <mdounin@mdounin.ru>
parents:
6174
diff
changeset
|
119 for (m = 0; cf->cycle->modules[m]; m++) { |
|
cf5e822cf470
Dynamic modules: changed ngx_modules to cycle->modules.
Maxim Dounin <mdounin@mdounin.ru>
parents:
6174
diff
changeset
|
120 if (cf->cycle->modules[m]->type != NGX_STREAM_MODULE) { |
| 6115 | 121 continue; |
| 122 } | |
| 123 | |
|
6379
cf5e822cf470
Dynamic modules: changed ngx_modules to cycle->modules.
Maxim Dounin <mdounin@mdounin.ru>
parents:
6174
diff
changeset
|
124 module = cf->cycle->modules[m]->ctx; |
| 6115 | 125 |
| 126 if (module->create_srv_conf) { | |
| 127 mconf = module->create_srv_conf(cf); | |
| 128 if (mconf == NULL) { | |
| 129 return NGX_CONF_ERROR; | |
| 130 } | |
| 131 | |
|
6379
cf5e822cf470
Dynamic modules: changed ngx_modules to cycle->modules.
Maxim Dounin <mdounin@mdounin.ru>
parents:
6174
diff
changeset
|
132 ctx->srv_conf[cf->cycle->modules[m]->ctx_index] = mconf; |
| 6115 | 133 } |
| 134 } | |
| 135 | |
| 136 uscf->servers = ngx_array_create(cf->pool, 4, | |
| 137 sizeof(ngx_stream_upstream_server_t)); | |
| 138 if (uscf->servers == NULL) { | |
| 139 return NGX_CONF_ERROR; | |
| 140 } | |
| 141 | |
| 142 | |
| 143 /* parse inside upstream{} */ | |
| 144 | |
| 145 pcf = *cf; | |
| 146 cf->ctx = ctx; | |
| 147 cf->cmd_type = NGX_STREAM_UPS_CONF; | |
| 148 | |
| 149 rv = ngx_conf_parse(cf, NULL); | |
| 150 | |
| 151 *cf = pcf; | |
| 152 | |
| 153 if (rv != NGX_CONF_OK) { | |
| 154 return rv; | |
| 155 } | |
| 156 | |
| 157 if (uscf->servers->nelts == 0) { | |
| 158 ngx_conf_log_error(NGX_LOG_EMERG, cf, 0, | |
| 159 "no servers are inside upstream"); | |
| 160 return NGX_CONF_ERROR; | |
| 161 } | |
| 162 | |
| 163 return rv; | |
| 164 } | |
| 165 | |
| 166 | |
| 167 static char * | |
| 168 ngx_stream_upstream_server(ngx_conf_t *cf, ngx_command_t *cmd, void *conf) | |
| 169 { | |
| 170 ngx_stream_upstream_srv_conf_t *uscf = conf; | |
| 171 | |
| 172 time_t fail_timeout; | |
| 173 ngx_str_t *value, s; | |
| 174 ngx_url_t u; | |
| 175 ngx_int_t weight, max_fails; | |
| 176 ngx_uint_t i; | |
| 177 ngx_stream_upstream_server_t *us; | |
| 178 | |
| 179 us = ngx_array_push(uscf->servers); | |
| 180 if (us == NULL) { | |
| 181 return NGX_CONF_ERROR; | |
| 182 } | |
| 183 | |
| 184 ngx_memzero(us, sizeof(ngx_stream_upstream_server_t)); | |
| 185 | |
| 186 value = cf->args->elts; | |
| 187 | |
| 188 weight = 1; | |
| 189 max_fails = 1; | |
| 190 fail_timeout = 10; | |
| 191 | |
| 192 for (i = 2; i < cf->args->nelts; i++) { | |
| 193 | |
| 194 if (ngx_strncmp(value[i].data, "weight=", 7) == 0) { | |
| 195 | |
| 196 if (!(uscf->flags & NGX_STREAM_UPSTREAM_WEIGHT)) { | |
| 197 goto not_supported; | |
| 198 } | |
| 199 | |
| 200 weight = ngx_atoi(&value[i].data[7], value[i].len - 7); | |
| 201 | |
| 202 if (weight == NGX_ERROR || weight == 0) { | |
| 203 goto invalid; | |
| 204 } | |
| 205 | |
| 206 continue; | |
| 207 } | |
| 208 | |
| 209 if (ngx_strncmp(value[i].data, "max_fails=", 10) == 0) { | |
| 210 | |
| 211 if (!(uscf->flags & NGX_STREAM_UPSTREAM_MAX_FAILS)) { | |
| 212 goto not_supported; | |
| 213 } | |
| 214 | |
| 215 max_fails = ngx_atoi(&value[i].data[10], value[i].len - 10); | |
| 216 | |
| 217 if (max_fails == NGX_ERROR) { | |
| 218 goto invalid; | |
| 219 } | |
| 220 | |
| 221 continue; | |
| 222 } | |
| 223 | |
| 224 if (ngx_strncmp(value[i].data, "fail_timeout=", 13) == 0) { | |
| 225 | |
| 226 if (!(uscf->flags & NGX_STREAM_UPSTREAM_FAIL_TIMEOUT)) { | |
| 227 goto not_supported; | |
| 228 } | |
| 229 | |
| 230 s.len = value[i].len - 13; | |
| 231 s.data = &value[i].data[13]; | |
| 232 | |
| 233 fail_timeout = ngx_parse_time(&s, 1); | |
| 234 | |
| 235 if (fail_timeout == (time_t) NGX_ERROR) { | |
| 236 goto invalid; | |
| 237 } | |
| 238 | |
| 239 continue; | |
| 240 } | |
| 241 | |
| 242 if (ngx_strcmp(value[i].data, "backup") == 0) { | |
| 243 | |
| 244 if (!(uscf->flags & NGX_STREAM_UPSTREAM_BACKUP)) { | |
| 245 goto not_supported; | |
| 246 } | |
| 247 | |
| 248 us->backup = 1; | |
| 249 | |
| 250 continue; | |
| 251 } | |
| 252 | |
| 253 if (ngx_strcmp(value[i].data, "down") == 0) { | |
| 254 | |
| 255 if (!(uscf->flags & NGX_STREAM_UPSTREAM_DOWN)) { | |
| 256 goto not_supported; | |
| 257 } | |
| 258 | |
| 259 us->down = 1; | |
| 260 | |
| 261 continue; | |
| 262 } | |
| 263 | |
| 264 goto invalid; | |
| 265 } | |
| 266 | |
| 267 ngx_memzero(&u, sizeof(ngx_url_t)); | |
| 268 | |
| 269 u.url = value[1]; | |
| 270 | |
| 271 if (ngx_parse_url(cf->pool, &u) != NGX_OK) { | |
| 272 if (u.err) { | |
| 273 ngx_conf_log_error(NGX_LOG_EMERG, cf, 0, | |
| 274 "%s in upstream \"%V\"", u.err, &u.url); | |
| 275 } | |
| 276 | |
| 277 return NGX_CONF_ERROR; | |
| 278 } | |
| 279 | |
| 280 if (u.no_port) { | |
| 281 ngx_conf_log_error(NGX_LOG_EMERG, cf, 0, | |
| 282 "no port in upstream \"%V\"", &u.url); | |
| 283 return NGX_CONF_ERROR; | |
| 284 } | |
| 285 | |
| 286 us->name = u.url; | |
| 287 us->addrs = u.addrs; | |
| 288 us->naddrs = u.naddrs; | |
| 289 us->weight = weight; | |
| 290 us->max_fails = max_fails; | |
| 291 us->fail_timeout = fail_timeout; | |
| 292 | |
| 293 return NGX_CONF_OK; | |
| 294 | |
| 295 invalid: | |
| 296 | |
| 297 ngx_conf_log_error(NGX_LOG_EMERG, cf, 0, | |
| 298 "invalid parameter \"%V\"", &value[i]); | |
| 299 | |
| 300 return NGX_CONF_ERROR; | |
| 301 | |
| 302 not_supported: | |
| 303 | |
| 304 ngx_conf_log_error(NGX_LOG_EMERG, cf, 0, | |
| 305 "balancing method does not support parameter \"%V\"", | |
| 306 &value[i]); | |
| 307 | |
| 308 return NGX_CONF_ERROR; | |
| 309 } | |
| 310 | |
| 311 | |
| 312 ngx_stream_upstream_srv_conf_t * | |
| 313 ngx_stream_upstream_add(ngx_conf_t *cf, ngx_url_t *u, ngx_uint_t flags) | |
| 314 { | |
| 315 ngx_uint_t i; | |
| 316 ngx_stream_upstream_server_t *us; | |
| 317 ngx_stream_upstream_srv_conf_t *uscf, **uscfp; | |
| 318 ngx_stream_upstream_main_conf_t *umcf; | |
| 319 | |
| 320 if (!(flags & NGX_STREAM_UPSTREAM_CREATE)) { | |
| 321 | |
| 322 if (ngx_parse_url(cf->pool, u) != NGX_OK) { | |
| 323 if (u->err) { | |
| 324 ngx_conf_log_error(NGX_LOG_EMERG, cf, 0, | |
| 325 "%s in upstream \"%V\"", u->err, &u->url); | |
| 326 } | |
| 327 | |
| 328 return NULL; | |
| 329 } | |
| 330 } | |
| 331 | |
| 332 umcf = ngx_stream_conf_get_module_main_conf(cf, ngx_stream_upstream_module); | |
| 333 | |
| 334 uscfp = umcf->upstreams.elts; | |
| 335 | |
| 336 for (i = 0; i < umcf->upstreams.nelts; i++) { | |
| 337 | |
| 338 if (uscfp[i]->host.len != u->host.len | |
| 339 || ngx_strncasecmp(uscfp[i]->host.data, u->host.data, u->host.len) | |
| 340 != 0) | |
| 341 { | |
| 342 continue; | |
| 343 } | |
| 344 | |
| 345 if ((flags & NGX_STREAM_UPSTREAM_CREATE) | |
| 346 && (uscfp[i]->flags & NGX_STREAM_UPSTREAM_CREATE)) | |
| 347 { | |
| 348 ngx_conf_log_error(NGX_LOG_EMERG, cf, 0, | |
| 349 "duplicate upstream \"%V\"", &u->host); | |
| 350 return NULL; | |
| 351 } | |
| 352 | |
| 353 if ((uscfp[i]->flags & NGX_STREAM_UPSTREAM_CREATE) && !u->no_port) { | |
| 354 ngx_conf_log_error(NGX_LOG_WARN, cf, 0, | |
| 355 "upstream \"%V\" may not have port %d", | |
| 356 &u->host, u->port); | |
| 357 return NULL; | |
| 358 } | |
| 359 | |
| 360 if ((flags & NGX_STREAM_UPSTREAM_CREATE) && !uscfp[i]->no_port) { | |
| 361 ngx_log_error(NGX_LOG_WARN, cf->log, 0, | |
| 362 "upstream \"%V\" may not have port %d in %s:%ui", | |
| 363 &u->host, uscfp[i]->port, | |
| 364 uscfp[i]->file_name, uscfp[i]->line); | |
| 365 return NULL; | |
| 366 } | |
| 367 | |
| 368 if (uscfp[i]->port != u->port) { | |
| 369 continue; | |
| 370 } | |
| 371 | |
| 372 if (flags & NGX_STREAM_UPSTREAM_CREATE) { | |
| 373 uscfp[i]->flags = flags; | |
| 374 } | |
| 375 | |
| 376 return uscfp[i]; | |
| 377 } | |
| 378 | |
| 379 uscf = ngx_pcalloc(cf->pool, sizeof(ngx_stream_upstream_srv_conf_t)); | |
| 380 if (uscf == NULL) { | |
| 381 return NULL; | |
| 382 } | |
| 383 | |
| 384 uscf->flags = flags; | |
| 385 uscf->host = u->host; | |
| 386 uscf->file_name = cf->conf_file->file.name.data; | |
| 387 uscf->line = cf->conf_file->line; | |
| 388 uscf->port = u->port; | |
| 389 uscf->no_port = u->no_port; | |
| 390 | |
| 391 if (u->naddrs == 1) { | |
| 392 uscf->servers = ngx_array_create(cf->pool, 1, | |
| 393 sizeof(ngx_stream_upstream_server_t)); | |
| 394 if (uscf->servers == NULL) { | |
| 395 return NULL; | |
| 396 } | |
| 397 | |
| 398 us = ngx_array_push(uscf->servers); | |
| 399 if (us == NULL) { | |
| 400 return NULL; | |
| 401 } | |
| 402 | |
| 403 ngx_memzero(us, sizeof(ngx_stream_upstream_server_t)); | |
| 404 | |
| 405 us->addrs = u->addrs; | |
| 406 us->naddrs = 1; | |
| 407 } | |
| 408 | |
| 409 uscfp = ngx_array_push(&umcf->upstreams); | |
| 410 if (uscfp == NULL) { | |
| 411 return NULL; | |
| 412 } | |
| 413 | |
| 414 *uscfp = uscf; | |
| 415 | |
| 416 return uscf; | |
| 417 } | |
| 418 | |
| 419 | |
| 420 static void * | |
| 421 ngx_stream_upstream_create_main_conf(ngx_conf_t *cf) | |
| 422 { | |
| 423 ngx_stream_upstream_main_conf_t *umcf; | |
| 424 | |
| 425 umcf = ngx_pcalloc(cf->pool, sizeof(ngx_stream_upstream_main_conf_t)); | |
| 426 if (umcf == NULL) { | |
| 427 return NULL; | |
| 428 } | |
| 429 | |
| 430 if (ngx_array_init(&umcf->upstreams, cf->pool, 4, | |
| 431 sizeof(ngx_stream_upstream_srv_conf_t *)) | |
| 432 != NGX_OK) | |
| 433 { | |
| 434 return NULL; | |
| 435 } | |
| 436 | |
| 437 return umcf; | |
| 438 } | |
| 439 | |
| 440 | |
| 441 static char * | |
| 442 ngx_stream_upstream_init_main_conf(ngx_conf_t *cf, void *conf) | |
| 443 { | |
| 444 ngx_stream_upstream_main_conf_t *umcf = conf; | |
| 445 | |
| 446 ngx_uint_t i; | |
| 447 ngx_stream_upstream_init_pt init; | |
| 448 ngx_stream_upstream_srv_conf_t **uscfp; | |
| 449 | |
| 450 uscfp = umcf->upstreams.elts; | |
| 451 | |
| 452 for (i = 0; i < umcf->upstreams.nelts; i++) { | |
| 453 | |
| 454 init = uscfp[i]->peer.init_upstream | |
| 455 ? uscfp[i]->peer.init_upstream | |
| 456 : ngx_stream_upstream_init_round_robin; | |
| 457 | |
| 458 if (init(cf, uscfp[i]) != NGX_OK) { | |
| 459 return NGX_CONF_ERROR; | |
| 460 } | |
| 461 } | |
| 462 | |
| 463 return NGX_CONF_OK; | |
| 464 } |