/* * Copyright (C) Max Romanov * Copyright (C) Igor Sysoev * Copyright (C) Valentin V. Bartenev * Copyright (C) NGINX, Inc. */ #include #include #include #include #include typedef struct { nxt_str_t type; nxt_str_t version; nxt_str_t file; } nxt_module_t; static nxt_buf_t *nxt_discovery_modules(nxt_task_t *task, const char *path); static nxt_int_t nxt_discovery_module(nxt_task_t *task, nxt_mp_t *mp, nxt_array_t *modules, const char *name); static nxt_app_module_t *nxt_app_module_load(nxt_task_t *task, const char *name); static nxt_thread_mutex_t nxt_app_mutex; static nxt_thread_cond_t nxt_app_cond; static nxt_http_fields_hash_entry_t nxt_app_request_fields[]; static nxt_http_fields_hash_t *nxt_app_request_fields_hash; static nxt_application_module_t *nxt_app; static uint32_t compat[] = { NXT_VERNUM, }; nxt_int_t nxt_discovery_start(nxt_task_t *task, void *data) { nxt_buf_t *b; nxt_port_t *main_port; nxt_runtime_t *rt; nxt_debug(task, "DISCOVERY"); rt = task->thread->runtime; b = nxt_discovery_modules(task, rt->modules); if (nxt_slow_path(b == NULL)) { exit(1); } main_port = rt->port_by_type[NXT_PROCESS_MAIN]; nxt_port_socket_write(task, main_port, NXT_PORT_MSG_MODULES, -1, 0, -1, b); return NXT_OK; } static void nxt_discovery_completion_handler(nxt_task_t *task, void *obj, void *data) { nxt_mp_t *mp; nxt_buf_t *b; b = obj; mp = b->data; nxt_mp_destroy(mp); exit(0); } static nxt_buf_t * nxt_discovery_modules(nxt_task_t *task, const char *path) { char *name; u_char *p, *end; size_t size; glob_t glb; nxt_mp_t *mp; nxt_buf_t *b; nxt_int_t ret; nxt_uint_t i, n; nxt_array_t *modules; nxt_module_t *module; b = NULL; mp = nxt_mp_create(1024, 128, 256, 32); if (mp == NULL) { return b; } ret = glob(path, 0, NULL, &glb); n = glb.gl_pathc; if (ret != 0) { nxt_log(task, NXT_LOG_NOTICE, "no modules matching: \"%s\" found", path); n = 0; } modules = nxt_array_create(mp, n, sizeof(nxt_module_t)); if (modules == NULL) { goto fail; } for (i = 0; i < n; i++) { name = glb.gl_pathv[i]; ret = nxt_discovery_module(task, mp, modules, name); if (ret != NXT_OK) { goto fail; } } size = sizeof("[]") - 1; module = modules->elts; n = modules->nelts; for (i = 0; i < n; i++) { nxt_debug(task, "module: %V %V %V", &module[i].type, &module[i].version, &module[i].file); size += sizeof("{\"type\": \"\",") - 1; size += sizeof(" \"version\": \"\",") - 1; size += sizeof(" \"file\": \"\"},") - 1; size += module[i].type.length + module[i].version.length + module[i].file.length; } b = nxt_buf_mem_alloc(mp, size, 0); if (b == NULL) { goto fail; } b->completion_handler = nxt_discovery_completion_handler; p = b->mem.free; end = b->mem.end; *p++ = '['; for (i = 0; i < n; i++) { p = nxt_sprintf(p, end, "{\"type\": \"%V\", \"version\": \"%V\", \"file\": \"%V\"},", &module[i].type, &module[i].version, &module[i].file); } *p++ = ']'; b->mem.free = p; fail: globfree(&glb); return b; } static nxt_int_t nxt_discovery_module(nxt_task_t *task, nxt_mp_t *mp, nxt_array_t *modules, const char *name) { void *dl; nxt_str_t *s; nxt_int_t ret; nxt_uint_t i, n; nxt_module_t *module; nxt_application_module_t *app; /* * Only memory allocation failure should return NXT_ERROR. * Any module processing errors are ignored. */ ret = NXT_ERROR; dl = dlopen(name, RTLD_GLOBAL | RTLD_NOW); if (dl == NULL) { nxt_log(task, NXT_LOG_CRIT, "dlopen(\"%s\"), failed: \"%s\"", name, dlerror()); return NXT_OK; } app = dlsym(dl, "nxt_app_module"); if (app != NULL) { nxt_log(task, NXT_LOG_NOTICE, "module: %V %V \"%s\"", &app->type, &app->version, name); if (app->compat_length != sizeof(compat) || nxt_memcmp(app->compat, compat, sizeof(compat)) != 0) { nxt_log(task, NXT_LOG_NOTICE, "incompatible module %s", name); goto done; } module = modules->elts; n = modules->nelts; for (i = 0; i < n; i++) { if (nxt_strstr_eq(&app->type, &module[i].type) && nxt_strstr_eq(&app->version, &module[i].version)) { nxt_log(task, NXT_LOG_NOTICE, "ignoring %s module with the same " "application language version %V %V as in %V", name, &module[i].type, &module[i].version, &module[i].file); goto done; } } module = nxt_array_add(modules); if (module == NULL) { goto fail; } s = nxt_str_dup(mp, &module->type, &app->type); if (s == NULL) { goto fail; } s = nxt_str_dup(mp, &module->version, &app->version); if (s == NULL) { goto fail; } module->file.length = nxt_strlen(name); module->file.start = nxt_mp_alloc(mp, module->file.length); if (module->file.start == NULL) { goto fail; } nxt_memcpy(module->file.start, name, module->file.length); } else { nxt_log(task, NXT_LOG_CRIT, "dlsym(\"%s\"), failed: \"%s\"", name, dlerror()); } done: ret = NXT_OK; fail: if (dlclose(dl) != 0) { nxt_log(task, NXT_LOG_CRIT, "dlclose(\"%s\"), failed: \"%s\"", name, dlerror()); } return ret; } nxt_int_t nxt_app_start(nxt_task_t *task, void *data) { nxt_int_t ret; nxt_app_lang_module_t *lang; nxt_common_app_conf_t *app_conf; app_conf = data; lang = nxt_app_lang_module(task->thread->runtime, &app_conf->type); if (nxt_slow_path(lang == NULL)) { nxt_log(task, NXT_LOG_CRIT, "unknown application type: \"%V\"", &app_conf->type); return NXT_ERROR; } nxt_app = lang->module; if (nxt_app == NULL) { nxt_debug(task, "application language module: %V \"%s\"", &lang->version, lang->file); nxt_app = nxt_app_module_load(task, lang->file); } if (app_conf->working_directory != NULL && app_conf->working_directory[0] != 0) { ret = chdir(app_conf->working_directory); if (nxt_slow_path(ret != 0)) { nxt_log(task, NXT_LOG_WARN, "chdir(%s) failed %E", app_conf->working_directory, nxt_errno); return NXT_ERROR; } } if (nxt_slow_path(nxt_thread_mutex_create(&nxt_app_mutex) != NXT_OK)) { return NXT_ERROR; } if (nxt_slow_path(nxt_thread_cond_create(&nxt_app_cond) != NXT_OK)) { return NXT_ERROR; } ret = nxt_app->init(task, data); if (nxt_slow_path(ret != NXT_OK)) { nxt_debug(task, "application init failed"); } else { nxt_debug(task, "application init done"); } return ret; } static nxt_app_module_t * nxt_app_module_load(nxt_task_t *task, const char *name) { void *dl; dl = dlopen(name, RTLD_GLOBAL | RTLD_LAZY); if (dl != NULL) { return dlsym(dl, "nxt_app_module"); } nxt_log(task, NXT_LOG_CRIT, "dlopen(\"%s\"), failed: \"%s\"", name, dlerror()); return NULL; } nxt_int_t nxt_app_http_init(nxt_task_t *task, nxt_runtime_t *rt) { nxt_http_fields_hash_t *hash; hash = nxt_http_fields_hash_create(nxt_app_request_fields, rt->mem_pool); if (nxt_slow_path(hash == NULL)) { return NXT_ERROR; } nxt_app_request_fields_hash = hash; return NXT_OK; } void nxt_port_app_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) { size_t dump_size; nxt_buf_t *b; nxt_port_t *port; nxt_app_rmsg_t rmsg = { msg->buf }; nxt_app_wmsg_t wmsg; b = msg->buf; dump_size = b->mem.free - b->mem.pos; if (dump_size > 300) { dump_size = 300; } nxt_debug(task, "app data: %*s ...", dump_size, b->mem.pos); port = nxt_runtime_port_find(task->thread->runtime, msg->port_msg.pid, msg->port_msg.reply_port); if (nxt_slow_path(port == NULL)) { // } wmsg.port = port; wmsg.write = NULL; wmsg.buf = &wmsg.write; wmsg.stream = msg->port_msg.stream; nxt_app->run(task, &rmsg, &wmsg); } nxt_inline nxt_port_t * nxt_app_msg_get_port(nxt_task_t *task, nxt_app_wmsg_t *msg) { return msg->port; } u_char * nxt_app_msg_write_get_buf(nxt_task_t *task, nxt_app_wmsg_t *msg, size_t size) { size_t free_size; u_char *res; nxt_buf_t *b; nxt_port_t *port; res = NULL; do { b = *msg->buf; if (b == NULL) { port = nxt_app_msg_get_port(task, msg); if (nxt_slow_path(port == NULL)) { return NULL; } b = nxt_port_mmap_get_buf(task, port, size); if (nxt_slow_path(b == NULL)) { return NULL; } *msg->buf = b; free_size = nxt_buf_mem_free_size(&b->mem); if (nxt_slow_path(free_size < size)) { nxt_log(task, NXT_LOG_WARN, "requested buffer too big " "(%z < %z)", free_size, size); return NULL; } } free_size = nxt_buf_mem_free_size(&b->mem); if (free_size >= size) { res = b->mem.free; b->mem.free += size; return res; } if (nxt_port_mmap_increase_buf(task, b, size, size) == NXT_OK) { res = b->mem.free; b->mem.free += size; return res; } msg->buf = &b->next; } while(1); } nxt_int_t nxt_app_msg_write(nxt_task_t *task, nxt_app_wmsg_t *msg, u_char *c, size_t size) { u_char *dst; size_t dst_length; if (c != NULL) { dst_length = size + (size < 128 ? 1 : 4) + 1; dst = nxt_app_msg_write_get_buf(task, msg, dst_length); if (nxt_slow_path(dst == NULL)) { nxt_debug(task, "nxt_app_msg_write: get_buf(%uz) failed", dst_length); return NXT_ERROR; } dst = nxt_app_msg_write_length(dst, size + 1); /* +1 for trailing 0 */ nxt_memcpy(dst, c, size); dst[size] = 0; nxt_debug(task, "nxt_app_msg_write: %uz %*s", size, (int) size, c); } else { dst_length = 1; dst = nxt_app_msg_write_get_buf(task, msg, dst_length); if (nxt_slow_path(dst == NULL)) { nxt_debug(task, "nxt_app_msg_write: get_buf(%uz) failed", dst_length); return NXT_ERROR; } dst = nxt_app_msg_write_length(dst, 0); nxt_debug(task, "nxt_app_msg_write: NULL"); } return NXT_OK; } nxt_int_t nxt_app_msg_write_prefixed_upcase(nxt_task_t *task, nxt_app_wmsg_t *msg, const nxt_str_t *prefix, const nxt_str_t *v) { u_char *dst, *src; size_t i, length, dst_length; length = prefix->length + v->length; dst_length = length + (length < 128 ? 1 : 4) + 1; dst = nxt_app_msg_write_get_buf(task, msg, dst_length); if (nxt_slow_path(dst == NULL)) { return NXT_ERROR; } dst = nxt_app_msg_write_length(dst, length + 1); /* +1 for trailing 0 */ nxt_memcpy(dst, prefix->start, prefix->length); dst += prefix->length; src = v->start; for (i = 0; i < v->length; i++, dst++, src++) { if (*src >= 'a' && *src <= 'z') { *dst = *src & ~0x20; continue; } if (*src == '-') { *dst = '_'; continue; } *dst = *src; } *dst = 0; return NXT_OK; } nxt_int_t nxt_app_msg_read_str(nxt_task_t *task, nxt_app_rmsg_t *msg, nxt_str_t *str) { size_t length; nxt_buf_t *buf; do { buf = msg->buf; if (nxt_slow_path(buf == NULL)) { return NXT_DONE; } if (nxt_slow_path(nxt_buf_mem_used_size(&buf->mem) < 1)) { if (nxt_fast_path(nxt_buf_mem_used_size(&buf->mem) == 0)) { msg->buf = buf->next; continue; } return NXT_ERROR; } if (buf->mem.pos[0] >= 128) { if (nxt_slow_path(nxt_buf_mem_used_size(&buf->mem) < 4)) { return NXT_ERROR; } } break; } while (1); buf->mem.pos = nxt_app_msg_read_length(buf->mem.pos, &length); if (nxt_slow_path(nxt_buf_mem_used_size(&buf->mem) < (intptr_t) length)) { return NXT_ERROR; } if (length > 0) { str->start = buf->mem.pos; str->length = length - 1; buf->mem.pos += length; nxt_debug(task, "nxt_read_str: %d %*s", (int) length - 1, (int) length - 1, str->start); } else { str->start = NULL; str->length = 0; nxt_debug(task, "nxt_read_str: NULL"); } return NXT_OK; } size_t nxt_app_msg_read_raw(nxt_task_t *task, nxt_app_rmsg_t *msg, void *dst, size_t size) { size_t res, read_size; nxt_buf_t *buf; res = 0; while (size > 0) { buf = msg->buf; if (nxt_slow_path(buf == NULL)) { break; } if (nxt_slow_path(nxt_buf_mem_used_size(&buf->mem) == 0)) { msg->buf = buf->next; continue; } read_size = nxt_buf_mem_used_size(&buf->mem); read_size = nxt_min(read_size, size); dst = nxt_cpymem(dst, buf->mem.pos, read_size); size -= read_size; buf->mem.pos += read_size; res += read_size; } nxt_debug(task, "nxt_read_raw: %uz", res); return res; } nxt_int_t nxt_app_msg_read_nvp(nxt_task_t *task, nxt_app_rmsg_t *rmsg, nxt_str_t *n, nxt_str_t *v) { nxt_int_t rc; rc = nxt_app_msg_read_str(task, rmsg, n); if (nxt_slow_path(rc != NXT_OK)) { return rc; } rc = nxt_app_msg_read_str(task, rmsg, v); if (nxt_slow_path(rc != NXT_OK)) { return rc; } return rc; } nxt_int_t nxt_app_msg_read_size(nxt_task_t *task, nxt_app_rmsg_t *msg, size_t *size) { nxt_buf_t *buf; do { buf = msg->buf; if (nxt_slow_path(buf == NULL)) { return NXT_DONE; } if (nxt_slow_path(nxt_buf_mem_used_size(&buf->mem) < 1)) { if (nxt_fast_path(nxt_buf_mem_used_size(&buf->mem) == 0)) { msg->buf = buf->next; continue; } return NXT_ERROR; } if (buf->mem.pos[0] >= 128) { if (nxt_slow_path(nxt_buf_mem_used_size(&buf->mem) < 4)) { return NXT_ERROR; } } break; } while (1); buf->mem.pos = nxt_app_msg_read_length(buf->mem.pos, size); nxt_debug(task, "nxt_read_size: %d", (int) *size); return NXT_OK; } static nxt_int_t nxt_app_request_content_length(void *ctx, nxt_http_field_t *field, nxt_log_t *log) { nxt_str_t *v; nxt_app_parse_ctx_t *c; nxt_app_request_header_t *h; c = ctx; h = &c->r.header; v = &field->value; h->content_length = *v; h->parsed_content_length = nxt_off_t_parse(v->start, v->length); return NXT_OK; } static nxt_int_t nxt_app_request_content_type(void *ctx, nxt_http_field_t *field, nxt_log_t *log) { nxt_app_parse_ctx_t *c; nxt_app_request_header_t *h; c = ctx; h = &c->r.header; h->content_type = field->value; return NXT_OK; } static nxt_int_t nxt_app_request_cookie(void *ctx, nxt_http_field_t *field, nxt_log_t *log) { nxt_app_parse_ctx_t *c; nxt_app_request_header_t *h; c = ctx; h = &c->r.header; h->cookie = field->value; return NXT_OK; } static nxt_int_t nxt_app_request_host(void *ctx, nxt_http_field_t *field, nxt_log_t *log) { nxt_app_parse_ctx_t *c; nxt_app_request_header_t *h; c = ctx; h = &c->r.header; h->host = field->value; return NXT_OK; } static nxt_http_fields_hash_entry_t nxt_app_request_fields[] = { { nxt_string("Content-Length"), &nxt_app_request_content_length, 0 }, { nxt_string("Content-Type"), &nxt_app_request_content_type, 0 }, { nxt_string("Cookie"), &nxt_app_request_cookie, 0 }, { nxt_string("Host"), &nxt_app_request_host, 0 }, { nxt_null_string, NULL, 0 } }; nxt_app_parse_ctx_t * nxt_app_http_req_init(nxt_task_t *task) { nxt_mp_t *mp; nxt_int_t rc; nxt_app_parse_ctx_t *ctx; mp = nxt_mp_create(1024, 128, 256, 32); if (nxt_slow_path(mp == NULL)) { return NULL; } ctx = nxt_mp_retain(mp, sizeof(nxt_app_parse_ctx_t)); if (nxt_slow_path(ctx == NULL)) { nxt_mp_destroy(mp); return NULL; } nxt_memzero(ctx, sizeof(nxt_app_parse_ctx_t)); ctx->mem_pool = mp; rc = nxt_http_parse_request_init(&ctx->parser, mp); if (nxt_slow_path(rc != NXT_OK)) { nxt_mp_release(mp, ctx); return NULL; } ctx->parser.fields_hash = nxt_app_request_fields_hash; return ctx; } nxt_int_t nxt_app_http_req_header_parse(nxt_task_t *task, nxt_app_parse_ctx_t *ctx, nxt_buf_t *buf) { nxt_int_t rc; nxt_app_request_body_t *b; nxt_http_request_parse_t *p; nxt_app_request_header_t *h; p = &ctx->parser; b = &ctx->r.body; h = &ctx->r.header; nxt_assert(h->done == 0); rc = nxt_http_parse_request(p, &buf->mem); if (nxt_slow_path(rc != NXT_DONE)) { return rc; } rc = nxt_http_fields_process(p->fields, ctx, task->log); if (nxt_slow_path(rc != NXT_OK)) { return rc; } h->fields = p->fields; h->done = 1; h->version.start = p->version.str; h->version.length = nxt_strlen(p->version.str); h->method = p->method; h->target.start = p->target_start; h->target.length = p->target_end - p->target_start; h->path = p->path; h->query = p->args; if (h->parsed_content_length == 0) { b->done = 1; } if (buf->mem.free == buf->mem.pos) { return NXT_DONE; } b->buf = buf; b->done = nxt_buf_mem_used_size(&buf->mem) >= h->parsed_content_length; if (b->done == 1) { b->preread_size = nxt_buf_mem_used_size(&buf->mem); } return NXT_DONE; } nxt_int_t nxt_app_http_req_body_read(nxt_task_t *task, nxt_app_parse_ctx_t *ctx, nxt_buf_t *buf) { nxt_app_request_body_t *b; nxt_app_request_header_t *h; b = &ctx->r.body; h = &ctx->r.header; nxt_assert(h->done == 1); nxt_assert(b->done == 0); b->done = nxt_buf_mem_used_size(&buf->mem) + b->preread_size >= (size_t) h->parsed_content_length; if (b->done == 1) { b->preread_size += nxt_buf_mem_used_size(&buf->mem); } return b->done == 1 ? NXT_DONE : NXT_AGAIN; } nxt_int_t nxt_app_http_req_done(nxt_task_t *task, nxt_app_parse_ctx_t *ctx) { nxt_mp_release(ctx->mem_pool, ctx); return NXT_OK; } nxt_int_t nxt_app_msg_flush(nxt_task_t *task, nxt_app_wmsg_t *msg, nxt_bool_t last) { nxt_int_t rc; nxt_buf_t *b; nxt_port_t *port; rc = NXT_OK; port = nxt_app_msg_get_port(task, msg); if (nxt_slow_path(port == NULL)) { return NXT_ERROR; } if (nxt_slow_path(last == 1)) { do { b = *msg->buf; if (b == NULL) { b = nxt_buf_sync_alloc(port->mem_pool, NXT_BUF_SYNC_LAST); *msg->buf = b; break; } msg->buf = &b->next; } while(1); } if (nxt_slow_path(msg->write != NULL)) { rc = nxt_port_socket_write(task, port, NXT_PORT_MSG_DATA, -1, msg->stream, 0, msg->write); msg->write = NULL; msg->buf = &msg->write; } return rc; } nxt_int_t nxt_app_msg_write_raw(nxt_task_t *task, nxt_app_wmsg_t *msg, const u_char *c, size_t size) { size_t free_size, copy_size; nxt_buf_t *b; nxt_port_t *port; nxt_debug(task, "nxt_app_msg_write_raw: %uz", size); while (size > 0) { b = *msg->buf; if (b == NULL) { port = nxt_app_msg_get_port(task, msg); if (nxt_slow_path(port == NULL)) { return NXT_ERROR; } b = nxt_port_mmap_get_buf(task, port, size); if (nxt_slow_path(b == NULL)) { return NXT_ERROR; } *msg->buf = b; } do { free_size = nxt_buf_mem_free_size(&b->mem); if (free_size > 0) { copy_size = nxt_min(free_size, size); b->mem.free = nxt_cpymem(b->mem.free, c, copy_size); size -= copy_size; c += copy_size; if (size == 0) { return NXT_OK; } } } while (nxt_port_mmap_increase_buf(task, b, size, 1) == NXT_OK); msg->buf = &b->next; } return NXT_OK; } nxt_app_lang_module_t * nxt_app_lang_module(nxt_runtime_t *rt, nxt_str_t *name) { u_char *p, *end, *version; size_t type_length, version_length; nxt_uint_t i, n; nxt_app_lang_module_t *lang; end = name->start + name->length; version = end; for (p = name->start; p < end; p++) { if (*p == ' ') { version = p + 1; break; } if (*p >= '0' && *p <= '9') { version = p; break; } } type_length = p - name->start; version_length = end - version; lang = rt->languages->elts; n = rt->languages->nelts; for (i = 0; i < n; i++) { if (nxt_str_eq(&lang[i].type, name->start, type_length) && nxt_str_start(&lang[i].version, version, version_length)) { return &lang[i]; } } return NULL; } nxt_app_type_t nxt_app_parse_type(nxt_str_t *str) { if (nxt_str_eq(str, "python", 6)) { return NXT_APP_PYTHON; } else if (nxt_str_eq(str, "php", 3)) { return NXT_APP_PHP; } else if (nxt_str_eq(str, "go", 2)) { return NXT_APP_GO; } return NXT_APP_UNKNOWN; }