From 394591783006d5edbd891558895708e4b9c622b4 Mon Sep 17 00:00:00 2001 From: Datong Sun Date: Thu, 26 Apr 2018 17:36:43 -0700 Subject: [PATCH 1/4] feature: ngx.semaphore support for stream_lua. #05a3d36 --- ngx_stream_lua-0.0.5/config | 2 + ngx_stream_lua-0.0.5/src/ddebug.h | 24 +- ngx_stream_lua-0.0.5/src/ngx_stream_lua_initworkerby.c | 4 +- ngx_stream_lua-0.0.5/src/ngx_stream_lua_module.c | 38 +- ngx_stream_lua-0.0.5/src/ngx_stream_lua_probe.h | 3 +- ngx_stream_lua-0.0.5/src/ngx_stream_lua_regex.c | 4 +- ngx_stream_lua-0.0.5/src/ngx_stream_lua_semaphore.c | 568 ++++++++++++++++++++++++++++++ ngx_stream_lua-0.0.5/src/ngx_stream_lua_semaphore.h | 51 +++ ngx_stream_lua-0.0.5/src/ngx_stream_lua_util.h | 2 - 9 files changed, 656 insertions(+), 40 deletions(-) create mode 100644 ngx_stream_lua-0.0.5/src/ngx_stream_lua_semaphore.c create mode 100644 ngx_stream_lua-0.0.5/src/ngx_stream_lua_semaphore.h diff --git a/ngx_stream_lua-0.0.5/config b/ngx_stream_lua-0.0.5/config index 758f5dc..5563736 100644 --- a/ngx_stream_lua-0.0.5/config +++ b/ngx_stream_lua-0.0.5/config @@ -341,6 +341,7 @@ NGX_ADDON_SRCS="$NGX_ADDON_SRCS \ $ngx_addon_dir/src/ngx_stream_lua_balancer.c \ $ngx_addon_dir/src/ngx_stream_lua_logby.c \ $ngx_addon_dir/src/ngx_stream_lua_prereadby.c \ + $ngx_addon_dir/src/ngx_stream_lua_semaphore.c \ " NGX_ADDON_DEPS="$NGX_ADDON_DEPS \ @@ -383,6 +384,7 @@ NGX_ADDON_DEPS="$NGX_ADDON_DEPS \ $ngx_addon_dir/src/ngx_stream_lua_balancer.h \ $ngx_addon_dir/src/ngx_stream_lua_logby.h \ $ngx_addon_dir/src/ngx_stream_lua_prereadby.h \ + $ngx_addon_dir/src/ngx_stream_lua_semaphore.h \ " ngx_feature="export symbols by default (-E)" diff --git a/ngx_stream_lua-0.0.5/src/ddebug.h b/ngx_stream_lua-0.0.5/src/ddebug.h index 7bf22e9..7f01608 100644 --- a/ngx_stream_lua-0.0.5/src/ddebug.h +++ b/ngx_stream_lua-0.0.5/src/ddebug.h @@ -56,21 +56,21 @@ dd(const char *fmt, ...) { #define dd_check_read_event_handler(r) \ dd("r->read_event_handler = %s", \ - r->read_event_handler == ngx_http_block_reading ? \ - "ngx_http_block_reading" : \ - r->read_event_handler == ngx_http_test_reading ? \ - "ngx_http_test_reading" : \ - r->read_event_handler == ngx_http_request_empty_handler ? \ - "ngx_http_request_empty_handler" : "UNKNOWN") + r->read_event_handler == ngx_http_block_reading ? \ + "ngx_http_block_reading" : \ + r->read_event_handler == ngx_http_test_reading ? \ + "ngx_http_test_reading" : \ + r->read_event_handler == ngx_http_request_empty_handler ? \ + "ngx_http_request_empty_handler" : "UNKNOWN") #define dd_check_write_event_handler(r) \ dd("r->write_event_handler = %s", \ - r->write_event_handler == ngx_http_handler ? \ - "ngx_http_handler" : \ - r->write_event_handler == ngx_http_core_run_phases ? \ - "ngx_http_core_run_phases" : \ - r->write_event_handler == ngx_http_request_empty_handler ? \ - "ngx_http_request_empty_handler" : "UNKNOWN") + r->write_event_handler == ngx_http_handler ? \ + "ngx_http_handler" : \ + r->write_event_handler == ngx_http_core_run_phases ? \ + "ngx_http_core_run_phases" : \ + r->write_event_handler == ngx_http_request_empty_handler ? \ + "ngx_http_request_empty_handler" : "UNKNOWN") #else diff --git a/ngx_stream_lua-0.0.5/src/ngx_stream_lua_initworkerby.c b/ngx_stream_lua-0.0.5/src/ngx_stream_lua_initworkerby.c index a6e8de1..8c7a7d4 100644 --- a/ngx_stream_lua-0.0.5/src/ngx_stream_lua_initworkerby.c +++ b/ngx_stream_lua-0.0.5/src/ngx_stream_lua_initworkerby.c @@ -32,7 +32,6 @@ ngx_stream_lua_init_worker(ngx_cycle_t *cycle) ngx_open_file_t *file, *ofile; ngx_list_part_t *part; ngx_connection_t *c = NULL; - ngx_conf_file_t *conf_file; ngx_stream_module_t *module; ngx_stream_lua_request_t *r = NULL; ngx_stream_lua_ctx_t *ctx; @@ -40,6 +39,7 @@ ngx_stream_lua_init_worker(ngx_cycle_t *cycle) ngx_stream_lua_main_conf_t *lmcf; + ngx_conf_file_t *conf_file; ngx_stream_session_t *s; ngx_stream_core_srv_conf_t *clcf, *top_clcf; @@ -177,12 +177,12 @@ ngx_stream_lua_init_worker(ngx_cycle_t *cycle) conf_file->file.name.data = (u_char *) "dummy"; conf_file->file.name.len = sizeof("dummy") - 1; conf_file->line = 1; + conf.conf_file = conf_file; conf.ctx = &stream_ctx; conf.cycle = fake_cycle; conf.pool = fake_cycle->pool; conf.log = cycle->log; - conf.conf_file = conf_file; stream_ctx.srv_conf = ngx_pcalloc(conf.pool, diff --git a/ngx_stream_lua-0.0.5/src/ngx_stream_lua_module.c b/ngx_stream_lua-0.0.5/src/ngx_stream_lua_module.c index a3b81c5..a035d0a 100644 --- a/ngx_stream_lua-0.0.5/src/ngx_stream_lua_module.c +++ b/ngx_stream_lua-0.0.5/src/ngx_stream_lua_module.c @@ -19,6 +19,7 @@ #include "ngx_stream_lua_probe.h" #include "ngx_stream_lua_balancer.h" #include "ngx_stream_lua_logby.h" +#include "ngx_stream_lua_semaphore.h" #include "ngx_stream_lua_prereadby.h" @@ -431,6 +432,7 @@ ngx_stream_lua_init(ngx_conf_t *cf) ngx_stream_handler_pt *h; ngx_stream_core_main_conf_t *cmcf; + ngx_pool_cleanup_t *cln; lmcf = ngx_stream_conf_get_module_main_conf(cf, ngx_stream_lua_module); @@ -476,7 +478,6 @@ ngx_stream_lua_init(ngx_conf_t *cf) } -#ifndef NGX_LUA_NO_FFI_API /* add the cleanup of semaphores after the lua_close */ cln = ngx_pool_cleanup_add(cf->pool, 0); if (cln == NULL) { @@ -485,7 +486,6 @@ ngx_stream_lua_init(ngx_conf_t *cf) cln->data = lmcf; cln->handler = ngx_stream_lua_sema_mm_cleanup; -#endif if (lmcf->lua == NULL) { @@ -553,9 +553,7 @@ ngx_stream_lua_lowat_check(ngx_conf_t *cf, void *post, void *data) static void * ngx_stream_lua_create_main_conf(ngx_conf_t *cf) { -#ifndef NGX_LUA_NO_FFI_API ngx_int_t rc; -#endif ngx_stream_lua_main_conf_t *lmcf; @@ -602,14 +600,12 @@ ngx_stream_lua_create_main_conf(ngx_conf_t *cf) lmcf->malloc_trim_cycle = NGX_CONF_UNSET_UINT; #endif -#ifndef NGX_LUA_NO_FFI_API rc = ngx_stream_lua_sema_mm_init(cf, lmcf); if (rc != NGX_OK) { return NULL; } dd("nginx Lua module main config structure initialized!"); -#endif return lmcf; } @@ -777,18 +773,18 @@ ngx_stream_lua_merge_srv_conf(ngx_conf_t *cf, void *parent, void *child) #if (NGX_STREAM_SSL) static ngx_int_t -ngx_stream_lua_set_ssl(ngx_conf_t *cf, ngx_stream_lua_srv_conf_t *lxcf) +ngx_stream_lua_set_ssl(ngx_conf_t *cf, ngx_stream_lua_srv_conf_t *lscf) { ngx_pool_cleanup_t *cln; - lxcf->ssl = ngx_pcalloc(cf->pool, sizeof(ngx_ssl_t)); - if (lxcf->ssl == NULL) { + lscf->ssl = ngx_pcalloc(cf->pool, sizeof(ngx_ssl_t)); + if (lscf->ssl == NULL) { return NGX_ERROR; } - lxcf->ssl->log = cf->log; + lscf->ssl->log = cf->log; - if (ngx_ssl_create(lxcf->ssl, lxcf->ssl_protocols, NULL) != NGX_OK) { + if (ngx_ssl_create(lscf->ssl, lscf->ssl_protocols, NULL) != NGX_OK) { return NGX_ERROR; } @@ -798,25 +794,25 @@ ngx_stream_lua_set_ssl(ngx_conf_t *cf, ngx_stream_lua_srv_conf_t *lxcf) } cln->handler = ngx_ssl_cleanup_ctx; - cln->data = lxcf->ssl; + cln->data = lscf->ssl; - if (SSL_CTX_set_cipher_list(lxcf->ssl->ctx, - (const char *) lxcf->ssl_ciphers.data) + if (SSL_CTX_set_cipher_list(lscf->ssl->ctx, + (const char *) lscf->ssl_ciphers.data) == 0) { ngx_ssl_error(NGX_LOG_EMERG, cf->log, 0, "SSL_CTX_set_cipher_list(\"%V\") failed", - &lxcf->ssl_ciphers); + &lscf->ssl_ciphers); return NGX_ERROR; } - if (lxcf->ssl_trusted_certificate.len) { + if (lscf->ssl_trusted_certificate.len) { #if defined(nginx_version) && nginx_version >= 1003007 - if (ngx_ssl_trusted_certificate(cf, lxcf->ssl, - &lxcf->ssl_trusted_certificate, - lxcf->ssl_verify_depth) + if (ngx_ssl_trusted_certificate(cf, lscf->ssl, + &lscf->ssl_trusted_certificate, + lscf->ssl_verify_depth) != NGX_OK) { return NGX_ERROR; @@ -832,9 +828,9 @@ ngx_stream_lua_set_ssl(ngx_conf_t *cf, ngx_stream_lua_srv_conf_t *lxcf) #endif } - dd("ssl crl: %.*s", (int) lxcf->ssl_crl.len, lxcf->ssl_crl.data); + dd("ssl crl: %.*s", (int) lscf->ssl_crl.len, lscf->ssl_crl.data); - if (ngx_ssl_crl(cf, lxcf->ssl, &lxcf->ssl_crl) != NGX_OK) { + if (ngx_ssl_crl(cf, lscf->ssl, &lscf->ssl_crl) != NGX_OK) { return NGX_ERROR; } diff --git a/ngx_stream_lua-0.0.5/src/ngx_stream_lua_probe.h b/ngx_stream_lua-0.0.5/src/ngx_stream_lua_probe.h index 14df542..b379002 100644 --- a/ngx_stream_lua-0.0.5/src/ngx_stream_lua_probe.h +++ b/ngx_stream_lua-0.0.5/src/ngx_stream_lua_probe.h @@ -46,7 +46,8 @@ NGINX_LUA_HTTP_LUA_SOCKET_TCP_RECEIVE_DONE(r, u, data, len) #define ngx_stream_lua_probe_socket_tcp_setkeepalive_buf_unread(r, u, \ - data, len) \ + data, \ + len) \ NGINX_LUA_HTTP_LUA_SOCKET_TCP_SETKEEPALIVE_BUF_UNREAD(r, u, data, len) #define ngx_stream_lua_probe_user_thread_spawn(r, creator, newthread) \ diff --git a/ngx_stream_lua-0.0.5/src/ngx_stream_lua_regex.c b/ngx_stream_lua-0.0.5/src/ngx_stream_lua_regex.c index 6cea5bd..cebe7d8 100644 --- a/ngx_stream_lua-0.0.5/src/ngx_stream_lua_regex.c +++ b/ngx_stream_lua-0.0.5/src/ngx_stream_lua_regex.c @@ -106,13 +106,13 @@ static void ngx_stream_lua_re_collect_named_captures(lua_State *L, #define ngx_stream_lua_regex_exec(re, e, s, start, captures, size, \ - opts) \ + opts) \ pcre_exec(re, e, (const char *) (s)->data, (s)->len, start, opts, \ captures, size) #define ngx_stream_lua_regex_dfa_exec(re, e, s, start, captures, size, \ - ws, wscount, opts) \ + ws, wscount, opts) \ pcre_dfa_exec(re, e, (const char *) (s)->data, (s)->len, start, opts, \ captures, size, ws, wscount) diff --git a/ngx_stream_lua-0.0.5/src/ngx_stream_lua_semaphore.c b/ngx_stream_lua-0.0.5/src/ngx_stream_lua_semaphore.c new file mode 100644 index 0000000..207015b --- /dev/null +++ b/ngx_stream_lua-0.0.5/src/ngx_stream_lua_semaphore.c @@ -0,0 +1,568 @@ + +/* + * Copyright (C) Yichun Zhang (agentzh) + * Copyright (C) cuiweixie + * I hereby assign copyright in this code to the lua-nginx-module project, + * to be licensed under the same terms as the rest of the code. + */ + + + + +#ifndef DDEBUG +#define DDEBUG 0 +#endif +#include "ddebug.h" + + +#include "ngx_stream_lua_util.h" +#include "ngx_stream_lua_semaphore.h" +#include "ngx_stream_lua_contentby.h" + + +ngx_int_t ngx_stream_lua_sema_mm_init(ngx_conf_t *cf, + ngx_stream_lua_main_conf_t *lmcf); +void ngx_stream_lua_sema_mm_cleanup(void *data); +static ngx_stream_lua_sema_t *ngx_stream_lua_alloc_sema(void); +static void ngx_stream_lua_free_sema(ngx_stream_lua_sema_t *sem); +static ngx_int_t ngx_stream_lua_sema_resume(ngx_stream_lua_request_t *r); +int ngx_stream_lua_ffi_sema_new(ngx_stream_lua_sema_t **psem, + int n, char **errmsg); +int ngx_stream_lua_ffi_sema_post(ngx_stream_lua_sema_t *sem, int n); +int ngx_stream_lua_ffi_sema_wait(ngx_stream_lua_request_t *r, + ngx_stream_lua_sema_t *sem, int wait_ms, u_char *err, size_t *errlen); +static void ngx_stream_lua_sema_cleanup(void *data); +static void ngx_stream_lua_sema_handler(ngx_event_t *ev); +static void ngx_stream_lua_sema_timeout_handler(ngx_event_t *ev); +void ngx_stream_lua_ffi_sema_gc(ngx_stream_lua_sema_t *sem); + + +enum { + SEMAPHORE_WAIT_SUCC = 0, + SEMAPHORE_WAIT_TIMEOUT = 1 +}; + + +ngx_int_t +ngx_stream_lua_sema_mm_init(ngx_conf_t *cf, ngx_stream_lua_main_conf_t *lmcf) +{ + ngx_stream_lua_sema_mm_t *mm; + + mm = ngx_palloc(cf->pool, sizeof(ngx_stream_lua_sema_mm_t)); + if (mm == NULL) { + return NGX_ERROR; + } + + lmcf->sema_mm = mm; + mm->lmcf = lmcf; + + ngx_queue_init(&mm->free_queue); + mm->cur_epoch = 0; + mm->total = 0; + mm->used = 0; + + /* it's better to be 4096, but it needs some space for + * ngx_stream_lua_sema_mm_block_t, one is enough, so it is 4095 + */ + mm->num_per_block = 4095; + + return NGX_OK; +} + + +static ngx_stream_lua_sema_t * +ngx_stream_lua_alloc_sema(void) +{ + ngx_uint_t i, n; + ngx_queue_t *q; + ngx_stream_lua_sema_t *sem, *iter; + ngx_stream_lua_sema_mm_t *mm; + ngx_stream_lua_main_conf_t *lmcf; + ngx_stream_lua_sema_mm_block_t *block; + + ngx_stream_lua_assert(ngx_cycle && ngx_cycle->conf_ctx); + + lmcf = ngx_stream_cycle_get_module_main_conf(ngx_cycle, + ngx_stream_lua_module); + + mm = lmcf->sema_mm; + + if (!ngx_queue_empty(&mm->free_queue)) { + q = ngx_queue_head(&mm->free_queue); + ngx_queue_remove(q); + + sem = ngx_queue_data(q, ngx_stream_lua_sema_t, chain); + + sem->block->used++; + + ngx_memzero(&sem->sem_event, sizeof(ngx_event_t)); + + sem->sem_event.handler = ngx_stream_lua_sema_handler; + sem->sem_event.data = sem; + sem->sem_event.log = ngx_cycle->log; + + mm->used++; + + ngx_log_debug1(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0, + "from head of free queue, alloc semaphore: %p", sem); + + return sem; + } + + /* free_queue is empty */ + + n = sizeof(ngx_stream_lua_sema_mm_block_t) + + mm->num_per_block * sizeof(ngx_stream_lua_sema_t); + + dd("block size: %d, item size: %d", + (int) sizeof(ngx_stream_lua_sema_mm_block_t), + (int) sizeof(ngx_stream_lua_sema_t)); + + block = ngx_alloc(n, ngx_cycle->log); + if (block == NULL) { + return NULL; + } + + mm->cur_epoch++; + mm->total += mm->num_per_block; + mm->used++; + + block->mm = mm; + block->epoch = mm->cur_epoch; + + sem = (ngx_stream_lua_sema_t *) (block + 1); + sem->block = block; + sem->block->used = 1; + + ngx_memzero(&sem->sem_event, sizeof(ngx_event_t)); + + sem->sem_event.handler = ngx_stream_lua_sema_handler; + sem->sem_event.data = sem; + sem->sem_event.log = ngx_cycle->log; + + for (iter = sem + 1, i = 1; i < mm->num_per_block; i++, iter++) { + iter->block = block; + ngx_queue_insert_tail(&mm->free_queue, &iter->chain); + } + + ngx_log_debug2(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0, + "new block, alloc semaphore: %p block: %p", sem, block); + + return sem; +} + + +void +ngx_stream_lua_sema_mm_cleanup(void *data) +{ + ngx_uint_t i; + ngx_queue_t *q; + ngx_stream_lua_sema_t *sem, *iter; + ngx_stream_lua_sema_mm_t *mm; + ngx_stream_lua_main_conf_t *lmcf; + ngx_stream_lua_sema_mm_block_t *block; + + lmcf = (ngx_stream_lua_main_conf_t *) data; + mm = lmcf->sema_mm; + + while (!ngx_queue_empty(&mm->free_queue)) { + q = ngx_queue_head(&mm->free_queue); + + sem = ngx_queue_data(q, ngx_stream_lua_sema_t, chain); + block = sem->block; + + if (block->used == 0) { + iter = (ngx_stream_lua_sema_t *) (block + 1); + + for (i = 0; i < block->mm->num_per_block; i++, iter++) { + ngx_queue_remove(&iter->chain); + } + + dd("free sema block: %p at final", block); + + ngx_free(block); + + } else { + /* just return directly when some thing goes wrong */ + + ngx_log_error(NGX_LOG_ALERT, ngx_cycle->log, 0, + "lua sema mm: freeing a block %p that is still " + " used by someone", block); + + return; + } + } + + dd("lua sema mm cleanup done"); +} + + +static void +ngx_stream_lua_free_sema(ngx_stream_lua_sema_t *sem) +{ + ngx_stream_lua_sema_t *iter; + ngx_uint_t i, mid_epoch; + ngx_stream_lua_sema_mm_block_t *block; + ngx_stream_lua_sema_mm_t *mm; + + block = sem->block; + block->used--; + + mm = block->mm; + mm->used--; + + mid_epoch = mm->cur_epoch - ((mm->total / mm->num_per_block) >> 1); + + if (block->epoch < mid_epoch) { + ngx_queue_insert_tail(&mm->free_queue, &sem->chain); + ngx_log_debug4(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0, + "add to free queue tail semaphore: %p epoch: %d" + "mid_epoch: %d cur_epoch: %d", sem, (int) block->epoch, + (int) mid_epoch, (int) mm->cur_epoch); + + } else { + ngx_queue_insert_head(&mm->free_queue, &sem->chain); + ngx_log_debug4(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0, + "add to free queue head semaphore: %p epoch: %d" + "mid_epoch: %d cur_epoch: %d", sem, (int) block->epoch, + (int) mid_epoch, (int) mm->cur_epoch); + } + + dd("used: %d", (int) block->used); + + if (block->used == 0 + && mm->used <= (mm->total >> 1) + && block->epoch < mid_epoch) + { + /* load <= 50% and it's on the older side */ + iter = (ngx_stream_lua_sema_t *) (block + 1); + + for (i = 0; i < mm->num_per_block; i++, iter++) { + ngx_queue_remove(&iter->chain); + } + + mm->total -= mm->num_per_block; + + ngx_log_debug1(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0, + "free semaphore block: %p", block); + + ngx_free(block); + } +} + + +static ngx_int_t +ngx_stream_lua_sema_resume(ngx_stream_lua_request_t *r) +{ + lua_State *vm; + ngx_connection_t *c; + ngx_int_t rc; + ngx_uint_t nreqs; + ngx_stream_lua_ctx_t *ctx; + + ctx = ngx_stream_lua_get_module_ctx(r, ngx_stream_lua_module); + if (ctx == NULL) { + return NGX_ERROR; + } + + ctx->resume_handler = ngx_stream_lua_wev_handler; + + c = r->connection; + vm = ngx_stream_lua_get_lua_vm(r, ctx); + nreqs = c->requests; + + if (ctx->cur_co_ctx->sem_resume_status == SEMAPHORE_WAIT_SUCC) { + lua_pushboolean(ctx->cur_co_ctx->co, 1); + lua_pushnil(ctx->cur_co_ctx->co); + + } else { + lua_pushboolean(ctx->cur_co_ctx->co, 0); + lua_pushliteral(ctx->cur_co_ctx->co, "timeout"); + } + + rc = ngx_stream_lua_run_thread(vm, r, ctx, 2); + + ngx_log_debug1(NGX_LOG_DEBUG_STREAM, r->connection->log, 0, + "lua run thread returned %d", rc); + + if (rc == NGX_AGAIN) { + return ngx_stream_lua_run_posted_threads(c, vm, r, ctx, nreqs); + } + + if (rc == NGX_DONE) { + ngx_stream_lua_finalize_request(r, NGX_DONE); + return ngx_stream_lua_run_posted_threads(c, vm, r, ctx, nreqs); + } + + /* rc == NGX_ERROR || rc >= NGX_OK */ + + if (ctx->entered_content_phase) { + ngx_stream_lua_finalize_request(r, rc); + return NGX_DONE; + } + + return rc; +} + + +int +ngx_stream_lua_ffi_sema_new(ngx_stream_lua_sema_t **psem, + int n, char **errmsg) +{ + ngx_stream_lua_sema_t *sem; + + sem = ngx_stream_lua_alloc_sema(); + if (sem == NULL) { + *errmsg = "no memory"; + return NGX_ERROR; + } + + ngx_queue_init(&sem->wait_queue); + + sem->resource_count = n; + sem->wait_count = 0; + *psem = sem; + + ngx_log_debug2(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0, + "stream lua semaphore new: %p, resources: %d", + sem, sem->resource_count); + + return NGX_OK; +} + + +int +ngx_stream_lua_ffi_sema_post(ngx_stream_lua_sema_t *sem, int n) +{ + ngx_log_debug3(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0, + "stream lua semaphore post: %p, n: %d, resources: %d", + sem, n, sem->resource_count); + + sem->resource_count += n; + + if (!ngx_queue_empty(&sem->wait_queue)) { + /* we need the extra paranthese around the first argument of + * ngx_post_event() just to work around macro issues in nginx + * cores older than nginx 1.7.12 (exclusive). + */ + ngx_post_event((&sem->sem_event), &ngx_posted_events); + } + + return NGX_OK; +} + + +int +ngx_stream_lua_ffi_sema_wait(ngx_stream_lua_request_t *r, + ngx_stream_lua_sema_t *sem, int wait_ms, u_char *err, size_t *errlen) +{ + ngx_stream_lua_ctx_t *ctx; + ngx_stream_lua_co_ctx_t *wait_co_ctx; + ngx_int_t rc; + + ngx_log_debug4(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0, + "stream lua semaphore wait: %p, timeout: %d, " + "resources: %d, event posted: %d", + sem, wait_ms, sem->resource_count, +#if (nginx_version >= 1007005) + (int) sem->sem_event.posted +#else + sem->sem_event.prev ? 1 : 0 +#endif + ); + + ctx = ngx_stream_lua_get_module_ctx(r, ngx_stream_lua_module); + if (ctx == NULL) { + *errlen = ngx_snprintf(err, *errlen, "no request ctx found") - err; + return NGX_ERROR; + } + + rc = ngx_stream_lua_ffi_check_context(ctx, NGX_STREAM_LUA_CONTEXT_CONTENT + | NGX_STREAM_LUA_CONTEXT_PREREAD + | NGX_STREAM_LUA_CONTEXT_TIMER, + err, errlen); + + if (rc != NGX_OK) { + return NGX_ERROR; + } + + /* we keep the order, will first resume the thread waiting for the + * longest time in ngx_stream_lua_sema_handler + */ + + if (ngx_queue_empty(&sem->wait_queue) && sem->resource_count > 0) { + sem->resource_count--; + return NGX_OK; + } + + if (wait_ms == 0) { + return NGX_DECLINED; + } + + sem->wait_count++; + wait_co_ctx = ctx->cur_co_ctx; + + wait_co_ctx->sleep.handler = ngx_stream_lua_sema_timeout_handler; + wait_co_ctx->sleep.data = ctx->cur_co_ctx; + wait_co_ctx->sleep.log = r->connection->log; + + ngx_add_timer(&wait_co_ctx->sleep, (ngx_msec_t) wait_ms); + + dd("ngx_stream_lua_ffi_sema_wait add timer coctx:%p wait: %d(ms)", + wait_co_ctx, wait_ms); + + ngx_queue_insert_tail(&sem->wait_queue, &wait_co_ctx->sem_wait_queue); + + wait_co_ctx->data = sem; + wait_co_ctx->cleanup = ngx_stream_lua_sema_cleanup; + + ngx_log_debug0(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0, + "stream lua semaphore wait yielding"); + + return NGX_AGAIN; +} + + +int +ngx_stream_lua_ffi_sema_count(ngx_stream_lua_sema_t *sem) +{ + return sem->resource_count - sem->wait_count; +} + + +static void +ngx_stream_lua_sema_cleanup(void *data) +{ + ngx_stream_lua_co_ctx_t *coctx = data; + ngx_queue_t *q; + ngx_stream_lua_sema_t *sem; + + sem = coctx->data; + + ngx_log_debug0(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0, + "stream lua semaphore cleanup"); + + if (coctx->sleep.timer_set) { + ngx_del_timer(&coctx->sleep); + } + + q = &coctx->sem_wait_queue; + + ngx_queue_remove(q); + sem->wait_count--; + coctx->cleanup = NULL; +} + + +static void +ngx_stream_lua_sema_handler(ngx_event_t *ev) +{ + ngx_stream_lua_sema_t *sem; + ngx_stream_lua_request_t *r; + ngx_stream_lua_ctx_t *ctx; + ngx_stream_lua_co_ctx_t *wait_co_ctx; + ngx_queue_t *q; + + sem = ev->data; + + while (!ngx_queue_empty(&sem->wait_queue) && sem->resource_count > 0) { + + q = ngx_queue_head(&sem->wait_queue); + ngx_queue_remove(q); + + sem->wait_count--; + + wait_co_ctx = ngx_queue_data(q, ngx_stream_lua_co_ctx_t, sem_wait_queue); + wait_co_ctx->cleanup = NULL; + + if (wait_co_ctx->sleep.timer_set) { + ngx_del_timer(&wait_co_ctx->sleep); + } + + r = ngx_stream_lua_get_req(wait_co_ctx->co); + + ctx = ngx_stream_lua_get_module_ctx(r, ngx_stream_lua_module); + ngx_stream_lua_assert(ctx != NULL); + + sem->resource_count--; + + ctx->cur_co_ctx = wait_co_ctx; + + wait_co_ctx->sem_resume_status = SEMAPHORE_WAIT_SUCC; + + if (ctx->entered_content_phase) { + (void) ngx_stream_lua_sema_resume(r); + + } else { + ctx->resume_handler = ngx_stream_lua_sema_resume; + ngx_stream_lua_core_run_phases(r); + } + + } +} + + +static void +ngx_stream_lua_sema_timeout_handler(ngx_event_t *ev) +{ + ngx_stream_lua_co_ctx_t *wait_co_ctx; + ngx_stream_lua_request_t *r; + ngx_stream_lua_ctx_t *ctx; + ngx_stream_lua_sema_t *sem; + + wait_co_ctx = ev->data; + wait_co_ctx->cleanup = NULL; + + dd("ngx_stream_lua_sema_timeout_handler timeout coctx:%p", wait_co_ctx); + + sem = wait_co_ctx->data; + + ngx_queue_remove(&wait_co_ctx->sem_wait_queue); + sem->wait_count--; + + r = ngx_stream_lua_get_req(wait_co_ctx->co); + + ctx = ngx_stream_lua_get_module_ctx(r, ngx_stream_lua_module); + ngx_stream_lua_assert(ctx != NULL); + + ctx->cur_co_ctx = wait_co_ctx; + + wait_co_ctx->sem_resume_status = SEMAPHORE_WAIT_TIMEOUT; + + if (ctx->entered_content_phase) { + (void) ngx_stream_lua_sema_resume(r); + + } else { + ctx->resume_handler = ngx_stream_lua_sema_resume; + ngx_stream_lua_core_run_phases(r); + } + +} + + +void +ngx_stream_lua_ffi_sema_gc(ngx_stream_lua_sema_t *sem) +{ + ngx_log_debug1(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0, + "in lua gc, semaphore %p", sem); + + if (sem == NULL) { + return; + } + + if (!ngx_terminate + && !ngx_quit + && !ngx_queue_empty(&sem->wait_queue)) + { + ngx_log_error(NGX_LOG_ERR, ngx_cycle->log, 0, + "in lua semaphore gc wait queue is" + " not empty while the semaphore %p is being " + "destroyed", sem); + } + + ngx_stream_lua_free_sema(sem); +} + + + +/* vi:set ft=c ts=4 sw=4 et fdm=marker: */ diff --git a/ngx_stream_lua-0.0.5/src/ngx_stream_lua_semaphore.h b/ngx_stream_lua-0.0.5/src/ngx_stream_lua_semaphore.h new file mode 100644 index 0000000..6841950 --- /dev/null +++ b/ngx_stream_lua-0.0.5/src/ngx_stream_lua_semaphore.h @@ -0,0 +1,51 @@ + +/* + * Copyright (C) Yichun Zhang (agentzh) + * Copyright (C) cuiweixie + * I hereby assign copyright in this code to the lua-nginx-module project, + * to be licensed under the same terms as the rest of the code. + */ + + +#ifndef _NGX_STREAM_LUA_SEMAPHORE_H_INCLUDED_ +#define _NGX_STREAM_LUA_SEMAPHORE_H_INCLUDED_ + + +#include "ngx_stream_lua_common.h" + + +typedef struct ngx_stream_lua_sema_mm_block_s { + ngx_uint_t used; + ngx_stream_lua_sema_mm_t *mm; + ngx_uint_t epoch; +} ngx_stream_lua_sema_mm_block_t; + + +struct ngx_stream_lua_sema_mm_s { + ngx_queue_t free_queue; + ngx_uint_t total; + ngx_uint_t used; + ngx_uint_t num_per_block; + ngx_uint_t cur_epoch; + ngx_stream_lua_main_conf_t *lmcf; +}; + + +typedef struct ngx_stream_lua_sema_s { + ngx_queue_t wait_queue; + ngx_queue_t chain; + ngx_event_t sem_event; + ngx_stream_lua_sema_mm_block_t *block; + int resource_count; + unsigned wait_count; +} ngx_stream_lua_sema_t; + + +void ngx_stream_lua_sema_mm_cleanup(void *data); +ngx_int_t ngx_stream_lua_sema_mm_init(ngx_conf_t *cf, + ngx_stream_lua_main_conf_t *lmcf); + + +#endif /* _NGX_STREAM_LUA_SEMAPHORE_H_INCLUDED_ */ + +/* vi:set ft=c ts=4 sw=4 et fdm=marker: */ diff --git a/ngx_stream_lua-0.0.5/src/ngx_stream_lua_util.h b/ngx_stream_lua-0.0.5/src/ngx_stream_lua_util.h index 135648a..49be95b 100644 --- a/ngx_stream_lua-0.0.5/src/ngx_stream_lua_util.h +++ b/ngx_stream_lua-0.0.5/src/ngx_stream_lua_util.h @@ -89,7 +89,6 @@ extern char ngx_stream_lua_headers_metatable_key; } -#ifndef NGX_LUA_NO_FFI_API static ngx_inline ngx_int_t ngx_stream_lua_ffi_check_context(ngx_stream_lua_ctx_t *ctx, unsigned flags, u_char *err, size_t *errlen) @@ -105,7 +104,6 @@ ngx_stream_lua_ffi_check_context(ngx_stream_lua_ctx_t *ctx, return NGX_OK; } -#endif #define ngx_stream_lua_check_fake_request(L, r) \ -- 2.19.1