technocloud-public/nginx-unit
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
fa81d7a11a7ba31ddded934eac8e58e7c801d44a
nginx-unit/src/perl/nxt_perl_psgi.c
Andrew Clayton fa81d7a11a Perl: Fix a crash in the language module. 
User @bes-internal reported a Perl module crasher on GitHub.

This was due to a Perl application sending back two responses, for each
response we would call down into XS_NGINX__Unit__Sandbox_cb(), the first
time pctx->req would point to a valid nxt_unit_request_info_t, the
second time pctx->req would be NULL.

Add an invalid responses check which covers this case.

Closes: <https://github.com/nginx/unit/issues/841>
Reviewed-by: Alejandro Colomar <alx@nginx.com>
Signed-off-by: Andrew Clayton <a.clayton@nginx.com>
2023-03-10 21:40:28 +00:00

1436 lines
36 KiB
C
Raw Blame History

/*
* Copyright (C) Alexander Borisov
* Copyright (C) NGINX, Inc.
*/
#include <perl/nxt_perl_psgi_layer.h>
#include <nxt_main.h>
#include <nxt_router.h>
#include <nxt_runtime.h>
#include <nxt_application.h>
#include <nxt_file.h>
#include <nxt_unit.h>
#include <nxt_unit_request.h>
#include <nxt_unit_response.h>
typedef struct {
PerlInterpreter *my_perl;
nxt_perl_psgi_io_arg_t arg_input;
nxt_perl_psgi_io_arg_t arg_error;
SV *app;
CV *cb;
nxt_unit_request_info_t *req;
pthread_t thread;
nxt_unit_ctx_t *ctx;
} nxt_perl_psgi_ctx_t;
static SSize_t nxt_perl_psgi_io_input_read(PerlInterpreter *my_perl,
nxt_perl_psgi_io_arg_t *arg, void *vbuf, size_t length);
static SSize_t nxt_perl_psgi_io_input_write(PerlInterpreter *my_perl,
nxt_perl_psgi_io_arg_t *arg, const void *vbuf, size_t length);
static SSize_t nxt_perl_psgi_io_error_read(PerlInterpreter *my_perl,
nxt_perl_psgi_io_arg_t *arg, void *vbuf, size_t length);
static SSize_t nxt_perl_psgi_io_error_write(PerlInterpreter *my_perl,
nxt_perl_psgi_io_arg_t *arg, const void *vbuf, size_t length);
/*
static void nxt_perl_psgi_xs_core_global_changes(PerlInterpreter *my_perl,
const char *core, const char *sub, XSUBADDR_t sub_addr);
*/
static void nxt_perl_psgi_xs_init(pTHX);
static SV *nxt_perl_psgi_call_var_application(PerlInterpreter *my_perl,
SV *env, SV *app, nxt_unit_request_info_t *req);
static SV *nxt_perl_psgi_call_method(PerlInterpreter *my_perl, SV *obj,
const char *method, nxt_unit_request_info_t *req);
/* For currect load XS modules */
EXTERN_C void boot_DynaLoader(pTHX_ CV *cv);
static int nxt_perl_psgi_io_init(PerlInterpreter *my_perl,
nxt_perl_psgi_io_arg_t *arg, const char *mode, void *req);
static int nxt_perl_psgi_ctx_init(const char *script,
nxt_perl_psgi_ctx_t *pctx);
static SV *nxt_perl_psgi_env_create(PerlInterpreter *my_perl,
nxt_unit_request_info_t *req);
nxt_inline int nxt_perl_psgi_add_sptr(PerlInterpreter *my_perl, HV *hash_env,
const char *name, uint32_t name_len, nxt_unit_sptr_t *sptr, uint32_t len);
nxt_inline int nxt_perl_psgi_add_str(PerlInterpreter *my_perl, HV *hash_env,
const char *name, uint32_t name_len, const char *str, uint32_t len);
nxt_inline int nxt_perl_psgi_add_value(PerlInterpreter *my_perl, HV *hash_env,
const char *name, uint32_t name_len, void *value);
static char *nxt_perl_psgi_module_create(const char *script);
static nxt_int_t nxt_perl_psgi_result_status(PerlInterpreter *my_perl,
SV *result);
static int nxt_perl_psgi_result_head(PerlInterpreter *my_perl,
SV *sv_head, nxt_unit_request_info_t *req, uint16_t status);
static int nxt_perl_psgi_result_body(PerlInterpreter *my_perl,
SV *result, nxt_unit_request_info_t *req);
static int nxt_perl_psgi_result_body_ref(PerlInterpreter *my_perl,
SV *sv_body, nxt_unit_request_info_t *req);
static int nxt_perl_psgi_result_body_fh(PerlInterpreter *my_perl, SV *sv_body,
nxt_unit_request_info_t *req);
static ssize_t nxt_perl_psgi_io_read(nxt_unit_read_info_t *read_info, void *dst,
size_t size);
static int nxt_perl_psgi_result_array(PerlInterpreter *my_perl,
SV *result, nxt_unit_request_info_t *req);
static void nxt_perl_psgi_result_cb(PerlInterpreter *my_perl, SV *result,
nxt_unit_request_info_t *req);
static nxt_int_t nxt_perl_psgi_start(nxt_task_t *task,
nxt_process_data_t *data);
static void nxt_perl_psgi_request_handler(nxt_unit_request_info_t *req);
static int nxt_perl_psgi_ready_handler(nxt_unit_ctx_t *ctx);
static void *nxt_perl_psgi_thread_func(void *main_ctx);
static int nxt_perl_psgi_init_threads(nxt_perl_app_conf_t *c);
static void nxt_perl_psgi_join_threads(nxt_unit_ctx_t *ctx,
nxt_perl_app_conf_t *c);
static void nxt_perl_psgi_ctx_free(nxt_perl_psgi_ctx_t *pctx);
static CV *nxt_perl_psgi_write;
static CV *nxt_perl_psgi_close;
static CV *nxt_perl_psgi_cb;
static pthread_attr_t *nxt_perl_psgi_thread_attr;
static nxt_perl_psgi_ctx_t *nxt_perl_psgi_ctxs;
static uint32_t nxt_perl_psgi_compat[] = {
NXT_VERNUM, NXT_DEBUG,
};
NXT_EXPORT nxt_app_module_t nxt_app_module = {
sizeof(nxt_perl_psgi_compat),
nxt_perl_psgi_compat,
nxt_string("perl"),
PERL_VERSION_STRING,
NULL,
0,
NULL,
nxt_perl_psgi_start,
};
const nxt_perl_psgi_io_tab_t nxt_perl_psgi_io_tab_input = {
.read = nxt_perl_psgi_io_input_read,
.write = nxt_perl_psgi_io_input_write,
};
const nxt_perl_psgi_io_tab_t nxt_perl_psgi_io_tab_error = {
.read = nxt_perl_psgi_io_error_read,
.write = nxt_perl_psgi_io_error_write,
};
static SSize_t
nxt_perl_psgi_io_input_read(PerlInterpreter *my_perl,
nxt_perl_psgi_io_arg_t *arg, void *vbuf, size_t length)
{
return nxt_unit_request_read(arg->req, vbuf, length);
}
static SSize_t
nxt_perl_psgi_io_input_write(PerlInterpreter *my_perl,
nxt_perl_psgi_io_arg_t *arg, const void *vbuf, size_t length)
{
return 0;
}
static SSize_t
nxt_perl_psgi_io_error_read(PerlInterpreter *my_perl,
nxt_perl_psgi_io_arg_t *arg, void *vbuf, size_t length)
{
return 0;
}
static SSize_t
nxt_perl_psgi_io_error_write(PerlInterpreter *my_perl,
nxt_perl_psgi_io_arg_t *arg, const void *vbuf, size_t length)
{
nxt_unit_req_error(arg->req, "Perl: %s", (const char*) vbuf);
return (SSize_t) length;
}
/* In the future it will be necessary to change some Perl functions. */
/*
static void
nxt_perl_psgi_xs_core_global_changes(PerlInterpreter *my_perl,
const char *core, const char *sub, XSUBADDR_t sub_addr)
{
GV *gv;
gv = gv_fetchpv(core, TRUE, SVt_PVCV);
#ifdef MUTABLE_CV
GvCV_set(gv, MUTABLE_CV(SvREFCNT_inc(get_cv(sub, TRUE))));
#else
GvCV_set(gv, (CV *) (SvREFCNT_inc(get_cv(sub, TRUE))));
#endif
GvIMPORTED_CV_on(gv);
newXS(sub, sub_addr, __FILE__);
}
*/
XS(XS_NGINX__Unit__PSGI_exit);
XS(XS_NGINX__Unit__PSGI_exit)
{
I32 ax = POPMARK;
Perl_croak(aTHX_ (char *) NULL);
XSRETURN_EMPTY;
}
XS(XS_NGINX__Unit__Sandbox_write);
XS(XS_NGINX__Unit__Sandbox_write)
{
int rc;
char *body;
size_t len;
nxt_perl_psgi_ctx_t *pctx;
dXSARGS;
if (nxt_slow_path(items != 2)) {
Perl_croak(aTHX_ "Wrong number of arguments. Need one string");
XSRETURN_EMPTY;
}
body = SvPV(ST(1), len);
pctx = CvXSUBANY(cv).any_ptr;
rc = nxt_unit_response_write(pctx->req, body, len);
if (nxt_slow_path(rc != NXT_UNIT_OK)) {
Perl_croak(aTHX_ "Failed to write response body");
XSRETURN_EMPTY;
}
XSRETURN_IV(len);
}
nxt_inline void
nxt_perl_psgi_cb_request_done(nxt_perl_psgi_ctx_t *pctx, int status)
{
if (pctx->req != NULL) {
nxt_unit_request_done(pctx->req, status);
pctx->req = NULL;
}
}
XS(XS_NGINX__Unit__Sandbox_close);
XS(XS_NGINX__Unit__Sandbox_close)
{
I32 ax;
ax = POPMARK;
nxt_perl_psgi_cb_request_done(CvXSUBANY(cv).any_ptr, NXT_UNIT_OK);
XSRETURN_NO;
}
XS(XS_NGINX__Unit__Sandbox_cb);
XS(XS_NGINX__Unit__Sandbox_cb)
{
SV *obj;
int rc;
long array_len;
nxt_perl_psgi_ctx_t *pctx;
dXSARGS;
if (nxt_slow_path(items != 1)) {
nxt_perl_psgi_cb_request_done(CvXSUBANY(cv).any_ptr, NXT_UNIT_ERROR);
Perl_croak(aTHX_ "Wrong number of arguments");
XSRETURN_EMPTY;
}
pctx = CvXSUBANY(cv).any_ptr;
if (nxt_slow_path(SvOK(ST(0)) == 0 || SvROK(ST(0)) == 0
|| SvTYPE(SvRV(ST(0))) != SVt_PVAV
|| pctx->req == NULL))
{
nxt_perl_psgi_cb_request_done(CvXSUBANY(cv).any_ptr, NXT_UNIT_ERROR);
Perl_croak(aTHX_ "PSGI: An unexpected response was received "
"from Perl Application");
XSRETURN_EMPTY;
}
rc = nxt_perl_psgi_result_array(PERL_GET_CONTEXT, ST(0), pctx->req);
if (nxt_slow_path(rc != NXT_UNIT_OK)) {
nxt_perl_psgi_cb_request_done(CvXSUBANY(cv).any_ptr, NXT_UNIT_ERROR);
Perl_croak(aTHX_ (char *) NULL);
XSRETURN_EMPTY;
}
array_len = av_len((AV *) SvRV(ST(0)));
if (array_len < 2) {
obj = sv_bless(newRV_noinc((SV *) newHV()),
gv_stashpv("NGINX::Unit::Sandbox", GV_ADD));
ST(0) = obj;
XSRETURN(1);
}
nxt_perl_psgi_cb_request_done(CvXSUBANY(cv).any_ptr, NXT_UNIT_OK);
XSRETURN_EMPTY;
}
static void
nxt_perl_psgi_xs_init(pTHX)
{
/*
nxt_perl_psgi_xs_core_global_changes(my_perl, "CORE::GLOBAL::exit",
"NGINX::Unit::PSGI::exit",
XS_NGINX__Unit__PSGI_exit);
*/
nxt_perl_psgi_layer_stream_init(aTHX);
/* DynaLoader for Perl modules who use XS */
newXS("DynaLoader::boot_DynaLoader", boot_DynaLoader, __FILE__);
nxt_perl_psgi_write = newXS("NGINX::Unit::Sandbox::write",
XS_NGINX__Unit__Sandbox_write, __FILE__);
nxt_perl_psgi_close = newXS("NGINX::Unit::Sandbox::close",
XS_NGINX__Unit__Sandbox_close, __FILE__);
nxt_perl_psgi_cb = newXS("NGINX::Unit::Sandbox::cb",
XS_NGINX__Unit__Sandbox_cb, __FILE__);
}
static SV *
nxt_perl_psgi_call_var_application(PerlInterpreter *my_perl,
SV *env, SV *app, nxt_unit_request_info_t *req)
{
SV *result;
dSP;
ENTER;
SAVETMPS;
PUSHMARK(sp);
XPUSHs(env);
PUTBACK;
call_sv(app, G_EVAL|G_SCALAR);
SPAGAIN;
if (SvTRUE(ERRSV)) {
nxt_unit_req_error(req, "PSGI: Failed to run Perl Application: \n%s",
SvPV_nolen(ERRSV));
}
result = POPs;
SvREFCNT_inc(result);
PUTBACK;
FREETMPS;
LEAVE;
return result;
}
static SV *
nxt_perl_psgi_call_method(PerlInterpreter *my_perl, SV *obj, const char *method,
nxt_unit_request_info_t *req)
{
SV *result;
dSP;
ENTER;
SAVETMPS;
PUSHMARK(sp);
XPUSHs(obj);
PUTBACK;
call_method(method, G_EVAL|G_SCALAR);
SPAGAIN;
if (SvTRUE(ERRSV)) {
nxt_unit_req_error(req, "PSGI: Failed to call method '%s':\n%s",
method, SvPV_nolen(ERRSV));
result = NULL;
} else {
result = SvREFCNT_inc(POPs);
}
PUTBACK;
FREETMPS;
LEAVE;
return result;
}
static char *
nxt_perl_psgi_module_create(const char *script)
{
char *buf, *p;
size_t length;
static nxt_str_t prefix = nxt_string(
"package NGINX::Unit::Sandbox;"
"sub new {"
" return bless {}, $_[0];"
"}"
"{my $app = do \""
);
static nxt_str_t suffix = nxt_string_zero(
"\";"
"unless ($app) {"
" if($@ || $1) {die $@ || $1}"
" else {die \"File not found or compilation error.\"}"
"} "
"return $app}"
);
length = strlen(script);
buf = nxt_unit_malloc(NULL, prefix.length + length + suffix.length);
if (nxt_slow_path(buf == NULL)) {
nxt_unit_alert(NULL, "PSGI: Failed to allocate memory "
"for Perl script file %s", script);
return NULL;
}
p = nxt_cpymem(buf, prefix.start, prefix.length);
p = nxt_cpymem(p, script, length);
nxt_memcpy(p, suffix.start, suffix.length);
return buf;
}
static int
nxt_perl_psgi_io_init(PerlInterpreter *my_perl,
nxt_perl_psgi_io_arg_t *arg, const char *mode, void *req)
{
SV *io;
PerlIO *fp;
if (arg->io == NULL) {
fp = nxt_perl_psgi_layer_stream_fp_create(aTHX_ arg->rv, mode);
if (nxt_slow_path(fp == NULL)) {
return NXT_UNIT_ERROR;
}
io = nxt_perl_psgi_layer_stream_io_create(aTHX_ fp);
if (nxt_slow_path(io == NULL)) {
nxt_perl_psgi_layer_stream_fp_destroy(aTHX_ fp);
return NXT_UNIT_ERROR;
}
arg->io = io;
arg->fp = fp;
}
arg->req = req;
return NXT_UNIT_OK;
}
static void
nxt_perl_psgi_io_release(PerlInterpreter *my_perl, nxt_perl_psgi_io_arg_t *arg)
{
if (arg->io != NULL) {
SvREFCNT_dec(arg->io);
arg->io = NULL;
}
}
static int
nxt_perl_psgi_ctx_init(const char *script, nxt_perl_psgi_ctx_t *pctx)
{
int status, res;
char *run_module;
PerlInterpreter *my_perl;
static char argv[] = "\0""-e\0""0";
static char *embedding[] = { &argv[0], &argv[1], &argv[4] };
my_perl = perl_alloc();
if (nxt_slow_path(my_perl == NULL)) {
nxt_unit_alert(NULL,
"PSGI: Failed to allocate memory for Perl interpreter");
return NXT_UNIT_ERROR;
}
pctx->my_perl = my_perl;
run_module = NULL;
perl_construct(my_perl);
PERL_SET_CONTEXT(my_perl);
status = perl_parse(my_perl, nxt_perl_psgi_xs_init, 3, embedding, NULL);
if (nxt_slow_path(status != 0)) {
nxt_unit_alert(NULL, "PSGI: Failed to parse Perl Script");
goto fail;
}
CvXSUBANY(nxt_perl_psgi_write).any_ptr = pctx;
CvXSUBANY(nxt_perl_psgi_close).any_ptr = pctx;
CvXSUBANY(nxt_perl_psgi_cb).any_ptr = pctx;
pctx->cb = nxt_perl_psgi_cb;
PL_exit_flags |= PERL_EXIT_DESTRUCT_END;
PL_origalen = 1;
status = perl_run(my_perl);
if (nxt_slow_path(status != 0)) {
nxt_unit_alert(NULL, "PSGI: Failed to run Perl");
goto fail;
}
sv_setsv(get_sv("0", 0), newSVpv(script, 0));
run_module = nxt_perl_psgi_module_create(script);
if (nxt_slow_path(run_module == NULL)) {
goto fail;
}
pctx->arg_input.rv = newSV_type(SVt_RV);
sv_setptrref(pctx->arg_input.rv, &pctx->arg_input);
SvSETMAGIC(pctx->arg_input.rv);
pctx->arg_input.io_tab = &nxt_perl_psgi_io_tab_input;
res = nxt_perl_psgi_io_init(my_perl, &pctx->arg_input, "r", NULL);
if (nxt_slow_path(res != NXT_UNIT_OK)) {
nxt_unit_alert(NULL, "PSGI: Failed to init io.psgi.input");
goto fail;
}
pctx->arg_error.rv = newSV_type(SVt_RV);
sv_setptrref(pctx->arg_error.rv, &pctx->arg_error);
SvSETMAGIC(pctx->arg_error.rv);
pctx->arg_error.io_tab = &nxt_perl_psgi_io_tab_error;
res = nxt_perl_psgi_io_init(my_perl, &pctx->arg_error, "w", NULL);
if (nxt_slow_path(res != NXT_UNIT_OK)) {
nxt_unit_alert(NULL, "PSGI: Failed to init io.psgi.error");
goto fail;
}
pctx->app = eval_pv(run_module, FALSE);
if (SvTRUE(ERRSV)) {
nxt_unit_alert(NULL, "PSGI: Failed to parse script: %s\n%s",
script, SvPV_nolen(ERRSV));
goto fail;
}
nxt_unit_free(NULL, run_module);
return NXT_UNIT_OK;
fail:
nxt_perl_psgi_io_release(my_perl, &pctx->arg_input);
nxt_perl_psgi_io_release(my_perl, &pctx->arg_error);
if (run_module != NULL) {
nxt_unit_free(NULL, run_module);
}
perl_destruct(my_perl);
perl_free(my_perl);
pctx->my_perl = NULL;
return NXT_UNIT_ERROR;
}
static SV *
nxt_perl_psgi_env_create(PerlInterpreter *my_perl,
nxt_unit_request_info_t *req)
{
HV *hash_env;
AV *array_version;
uint32_t i;
nxt_unit_field_t *f;
nxt_unit_request_t *r;
nxt_perl_psgi_ctx_t *pctx;
pctx = req->ctx->data;
hash_env = newHV();
if (nxt_slow_path(hash_env == NULL)) {
return NULL;
}
#define RC(FNS) \
do { \
if (nxt_slow_path((FNS) != NXT_UNIT_OK)) \
goto fail; \
} while (0)
#define NL(S) (S), sizeof(S)-1
r = req->request;
RC(nxt_perl_psgi_add_str(my_perl, hash_env, NL("SERVER_SOFTWARE"),
(char *) nxt_server.start, nxt_server.length));
RC(nxt_perl_psgi_add_sptr(my_perl, hash_env, NL("REQUEST_METHOD"),
&r->method, r->method_length));
RC(nxt_perl_psgi_add_sptr(my_perl, hash_env, NL("REQUEST_URI"),
&r->target, r->target_length));
RC(nxt_perl_psgi_add_sptr(my_perl, hash_env, NL("PATH_INFO"),
&r->path, r->path_length));
array_version = newAV();
if (nxt_slow_path(array_version == NULL)) {
goto fail;
}
av_push(array_version, newSViv(1));
av_push(array_version, newSViv(1));
RC(nxt_perl_psgi_add_value(my_perl, hash_env, NL("psgi.version"),
newRV_noinc((SV *) array_version)));
RC(nxt_perl_psgi_add_value(my_perl, hash_env, NL("psgi.url_scheme"),
r->tls ? newSVpv("https", 5)
: newSVpv("http", 4)));
RC(nxt_perl_psgi_io_init(my_perl, &pctx->arg_input, "r", req));
RC(nxt_perl_psgi_add_value(my_perl, hash_env, NL("psgi.input"),
SvREFCNT_inc(pctx->arg_input.io)));
RC(nxt_perl_psgi_io_init(my_perl, &pctx->arg_error, "w", req));
RC(nxt_perl_psgi_add_value(my_perl, hash_env, NL("psgi.errors"),
SvREFCNT_inc(pctx->arg_error.io)));
RC(nxt_perl_psgi_add_value(my_perl, hash_env, NL("psgi.multithread"),
nxt_perl_psgi_ctxs != NULL
? &PL_sv_yes : &PL_sv_no));
RC(nxt_perl_psgi_add_value(my_perl, hash_env, NL("psgi.multiprocess"),
&PL_sv_yes));
RC(nxt_perl_psgi_add_value(my_perl, hash_env, NL("psgi.run_once"),
&PL_sv_no));
RC(nxt_perl_psgi_add_value(my_perl, hash_env, NL("psgi.nonblocking"),
&PL_sv_no));
RC(nxt_perl_psgi_add_value(my_perl, hash_env, NL("psgi.streaming"),
&PL_sv_yes));
RC(nxt_perl_psgi_add_sptr(my_perl, hash_env, NL("QUERY_STRING"),
&r->query, r->query_length));
RC(nxt_perl_psgi_add_sptr(my_perl, hash_env, NL("SERVER_PROTOCOL"),
&r->version, r->version_length));
RC(nxt_perl_psgi_add_sptr(my_perl, hash_env, NL("REMOTE_ADDR"),
&r->remote, r->remote_length));
RC(nxt_perl_psgi_add_sptr(my_perl, hash_env, NL("SERVER_ADDR"),
&r->local_addr, r->local_addr_length));
RC(nxt_perl_psgi_add_sptr(my_perl, hash_env, NL("SERVER_NAME"),
&r->server_name, r->server_name_length));
RC(nxt_perl_psgi_add_str(my_perl, hash_env, NL("SERVER_PORT"), "80", 2));
for (i = 0; i < r->fields_count; i++) {
f = r->fields + i;
RC(nxt_perl_psgi_add_sptr(my_perl, hash_env,
nxt_unit_sptr_get(&f->name), f->name_length,
&f->value, f->value_length));
}
if (r->content_length_field != NXT_UNIT_NONE_FIELD) {
f = r->fields + r->content_length_field;
RC(nxt_perl_psgi_add_sptr(my_perl, hash_env, NL("CONTENT_LENGTH"),
&f->value, f->value_length));
}
if (r->content_type_field != NXT_UNIT_NONE_FIELD) {
f = r->fields + r->content_type_field;
RC(nxt_perl_psgi_add_sptr(my_perl, hash_env, NL("CONTENT_TYPE"),
&f->value, f->value_length));
}
#undef NL
#undef RC
return newRV_noinc((SV *) hash_env);
fail:
SvREFCNT_dec(hash_env);
return NULL;
}
nxt_inline int
nxt_perl_psgi_add_sptr(PerlInterpreter *my_perl, HV *hash_env,
const char *name, uint32_t name_len, nxt_unit_sptr_t *sptr, uint32_t len)
{
return nxt_perl_psgi_add_str(my_perl, hash_env, name, name_len,
nxt_unit_sptr_get(sptr), len);
}
nxt_inline int
nxt_perl_psgi_add_str(PerlInterpreter *my_perl, HV *hash_env,
const char *name, uint32_t name_len, const char *str, uint32_t len)
{
SV **ha;
ha = hv_store(hash_env, name, (I32) name_len,
newSVpv(str, (STRLEN) len), 0);
if (nxt_slow_path(ha == NULL)) {
return NXT_UNIT_ERROR;
}
return NXT_UNIT_OK;
}
nxt_inline int
nxt_perl_psgi_add_value(PerlInterpreter *my_perl, HV *hash_env,
const char *name, uint32_t name_len, void *value)
{
SV **ha;
ha = hv_store(hash_env, name, (I32) name_len, value, 0);
if (nxt_slow_path(ha == NULL)) {
return NXT_UNIT_ERROR;
}
return NXT_UNIT_OK;
}
static nxt_int_t
nxt_perl_psgi_result_status(PerlInterpreter *my_perl, SV *result)
{
SV **sv_status;
AV *array;
u_char *space;
nxt_str_t status;
array = (AV *) SvRV(result);
sv_status = av_fetch(array, 0, 0);
status.start = (u_char *) SvPV(*sv_status, status.length);
space = memchr(status.start, ' ', status.length);
if (space != NULL) {
status.length = space - status.start;
}
return nxt_int_parse(status.start, status.length);
}
static int
nxt_perl_psgi_result_head(PerlInterpreter *my_perl, SV *sv_head,
nxt_unit_request_info_t *req, uint16_t status)
{
AV *array_head;
SV **entry;
int rc;
long i, array_len;
char *name, *value;
STRLEN name_len, value_len;
uint32_t fields, size;
if (nxt_slow_path(SvROK(sv_head) == 0
|| SvTYPE(SvRV(sv_head)) != SVt_PVAV))
{
nxt_unit_req_error(req,
"PSGI: An unsupported format was received from "
"Perl Application for head part");
return NXT_UNIT_ERROR;
}
array_head = (AV *) SvRV(sv_head);
array_len = av_len(array_head);
if (array_len < 1) {
return nxt_unit_response_init(req, status, 0, 0);
}
if (nxt_slow_path((array_len % 2) == 0)) {
nxt_unit_req_error(req, "PSGI: Bad format for head from "
"Perl Application");
return NXT_UNIT_ERROR;
}
fields = 0;
size = 0;
for (i = 0; i <= array_len; i++) {
entry = av_fetch(array_head, i, 0);
if (nxt_fast_path(entry == NULL)) {
nxt_unit_req_error(req, "PSGI: Failed to get head entry from "
"Perl Application");
return NXT_UNIT_ERROR;
}
value = SvPV(*entry, value_len);
size += value_len;
if ((i % 2) == 0) {
fields++;
}
}
rc = nxt_unit_response_init(req, status, fields, size);
if (nxt_slow_path(rc != NXT_UNIT_OK)) {
return rc;
}
for (i = 0; i <= array_len; i += 2) {
entry = av_fetch(array_head, i, 0);
name = SvPV(*entry, name_len);
entry = av_fetch(array_head, i + 1, 0);
value = SvPV(*entry, value_len);
rc = nxt_unit_response_add_field(req, name, name_len, value, value_len);
if (nxt_slow_path(rc != NXT_UNIT_OK)) {
return rc;
}
}
return NXT_UNIT_OK;
}
static int
nxt_perl_psgi_result_body(PerlInterpreter *my_perl, SV *sv_body,
nxt_unit_request_info_t *req)
{
SV **entry;
AV *body_array;
int rc;
long i;
nxt_str_t body;
if (nxt_slow_path(SvROK(sv_body) == 0
|| SvTYPE(SvRV(sv_body)) != SVt_PVAV))
{
nxt_unit_req_error(req, "PSGI: An unsupported format was received from "
"Perl Application for a body part");
return NXT_UNIT_ERROR;
}
body_array = (AV *) SvRV(sv_body);
for (i = 0; i <= av_len(body_array); i++) {
entry = av_fetch(body_array, i, 0);
if (nxt_fast_path(entry == NULL)) {
nxt_unit_req_error(req, "PSGI: Failed to get body entry from "
"Perl Application");
return NXT_UNIT_ERROR;
}
body.start = (u_char *) SvPV(*entry, body.length);
if (body.length == 0) {
continue;
}
rc = nxt_unit_response_write(req, body.start, body.length);
if (nxt_slow_path(rc != NXT_UNIT_OK)) {
nxt_unit_req_error(req, "PSGI: Failed to write content from "
"Perl Application");
return rc;
}
}
return NXT_UNIT_OK;
}
static int
nxt_perl_psgi_result_body_ref(PerlInterpreter *my_perl, SV *sv_body,
nxt_unit_request_info_t *req)
{
SV *data, *old_rs, *old_perl_rs;
int rc;
size_t len;
const char *body;
/*
* Servers should set the $/ special variable to the buffer size
* when reading content from $body using the getline method.
* This is done by setting $/ with a reference to an integer ($/ = \8192).
*/
old_rs = PL_rs;
old_perl_rs = get_sv("/", GV_ADD);
PL_rs = sv_2mortal(newRV_noinc(newSViv(nxt_unit_buf_min())));
sv_setsv(old_perl_rs, PL_rs);
rc = NXT_UNIT_OK;
for ( ;; ) {
data = nxt_perl_psgi_call_method(my_perl, sv_body, "getline", req);
if (nxt_slow_path(data == NULL)) {
rc = NXT_UNIT_ERROR;
break;
}
body = SvPV(data, len);
if (len == 0) {
SvREFCNT_dec(data);
data = nxt_perl_psgi_call_method(my_perl, sv_body, "close", req);
if (nxt_fast_path(data != NULL)) {
SvREFCNT_dec(data);
}
break;
}
rc = nxt_unit_response_write(req, body, len);
SvREFCNT_dec(data);
if (nxt_slow_path(rc != NXT_UNIT_OK)) {
nxt_unit_req_error(req, "PSGI: Failed to write content from "
"Perl Application");
break;
}
};
PL_rs = old_rs;
sv_setsv(get_sv("/", GV_ADD), old_perl_rs);
return rc;
}
typedef struct {
PerlInterpreter *my_perl;
PerlIO *fp;
} nxt_perl_psgi_io_ctx_t;
static int
nxt_perl_psgi_result_body_fh(PerlInterpreter *my_perl, SV *sv_body,
nxt_unit_request_info_t *req)
{
IO *io;
nxt_unit_read_info_t read_info;
nxt_perl_psgi_io_ctx_t io_ctx;
io = GvIO(SvRV(sv_body));
if (io == NULL) {
return NXT_UNIT_OK;
}
io_ctx.my_perl = my_perl;
io_ctx.fp = IoIFP(io);
read_info.read = nxt_perl_psgi_io_read;
read_info.eof = PerlIO_eof(io_ctx.fp);
read_info.buf_size = 8192;
read_info.data = &io_ctx;
return nxt_unit_response_write_cb(req, &read_info);
}
static ssize_t
nxt_perl_psgi_io_read(nxt_unit_read_info_t *read_info, void *dst, size_t size)
{
ssize_t res;
nxt_perl_psgi_io_ctx_t *ctx;
ctx = read_info->data;
dTHXa(ctx->my_perl);
res = PerlIO_read(ctx->fp, dst, size);
read_info->eof = PerlIO_eof(ctx->fp);
return res;
}
static int
nxt_perl_psgi_result_array(PerlInterpreter *my_perl, SV *result,
nxt_unit_request_info_t *req)
{
AV *array;
SV **sv_temp;
int rc;
long array_len;
nxt_int_t status;
array = (AV *) SvRV(result);
array_len = av_len(array);
if (nxt_slow_path(array_len < 0)) {
nxt_unit_req_error(req,
"PSGI: Invalid result format from Perl Application");
return NXT_UNIT_ERROR;
}
status = nxt_perl_psgi_result_status(my_perl, result);
if (nxt_slow_path(status < 0)) {
nxt_unit_req_error(req,
"PSGI: An unexpected status was received "
"from Perl Application");
return NXT_UNIT_ERROR;
}
if (array_len >= 1) {
sv_temp = av_fetch(array, 1, 0);
if (nxt_slow_path(sv_temp == NULL)) {
nxt_unit_req_error(req, "PSGI: Failed to get head from "
"Perl ARRAY variable");
return NXT_UNIT_ERROR;
}
rc = nxt_perl_psgi_result_head(my_perl, *sv_temp, req, status);
if (nxt_slow_path(rc != NXT_UNIT_OK)) {
return rc;
}
} else {
return nxt_unit_response_init(req, status, 0, 0);
}
if (nxt_fast_path(array_len < 2)) {
return NXT_UNIT_OK;
}
sv_temp = av_fetch(array, 2, 0);
if (nxt_slow_path(sv_temp == NULL || SvROK(*sv_temp) == FALSE)) {
nxt_unit_req_error(req,
"PSGI: Failed to get body from "
"Perl ARRAY variable");
return NXT_UNIT_ERROR;
}
if (SvTYPE(SvRV(*sv_temp)) == SVt_PVAV) {
return nxt_perl_psgi_result_body(my_perl, *sv_temp, req);
}
if (SvTYPE(SvRV(*sv_temp)) == SVt_PVGV) {
return nxt_perl_psgi_result_body_fh(my_perl, *sv_temp, req);
}
return nxt_perl_psgi_result_body_ref(my_perl, *sv_temp, req);
}
static void
nxt_perl_psgi_result_cb(PerlInterpreter *my_perl, SV *result,
nxt_unit_request_info_t *req)
{
nxt_perl_psgi_ctx_t *pctx;
dSP;
pctx = req->ctx->data;
ENTER;
SAVETMPS;
PUSHMARK(sp);
XPUSHs(newRV_noinc((SV*) pctx->cb));
PUTBACK;
call_sv(result, G_EVAL|G_SCALAR);
SPAGAIN;
if (SvTRUE(ERRSV)) {
nxt_unit_error(NULL, "PSGI: Failed to execute result callback: \n%s",
SvPV_nolen(ERRSV));
nxt_perl_psgi_cb_request_done(pctx, NXT_UNIT_ERROR);
}
PUTBACK;
FREETMPS;
LEAVE;
}
static nxt_int_t
nxt_perl_psgi_start(nxt_task_t *task, nxt_process_data_t *data)
{
int rc, pargc;
char **pargv, **penv;
nxt_unit_ctx_t *unit_ctx;
nxt_unit_init_t perl_init;
nxt_perl_psgi_ctx_t pctx;
nxt_perl_app_conf_t *c;
nxt_common_app_conf_t *common_conf;
common_conf = data->app;
c = &common_conf->u.perl;
pargc = 0;
pargv = NULL;
penv = NULL;
PERL_SYS_INIT3(&pargc, &pargv, &penv);
memset(&pctx, 0, sizeof(nxt_perl_psgi_ctx_t));
rc = nxt_perl_psgi_ctx_init(c->script, &pctx);
if (nxt_slow_path(rc != NXT_UNIT_OK)) {
goto fail;
}
rc = nxt_perl_psgi_init_threads(c);
PERL_SET_CONTEXT(pctx.my_perl);
if (nxt_slow_path(rc != NXT_UNIT_OK)) {
goto fail;
}
nxt_unit_default_init(task, &perl_init, common_conf);
perl_init.callbacks.request_handler = nxt_perl_psgi_request_handler;
perl_init.callbacks.ready_handler = nxt_perl_psgi_ready_handler;
perl_init.data = c;
perl_init.ctx_data = &pctx;
unit_ctx = nxt_unit_init(&perl_init);
if (nxt_slow_path(unit_ctx == NULL)) {
goto fail;
}
rc = nxt_unit_run(unit_ctx);
nxt_perl_psgi_join_threads(unit_ctx, c);
nxt_unit_done(unit_ctx);
nxt_perl_psgi_ctx_free(&pctx);
PERL_SYS_TERM();
exit(rc);
return NXT_OK;
fail:
nxt_perl_psgi_join_threads(NULL, c);
nxt_perl_psgi_ctx_free(&pctx);
PERL_SYS_TERM();
return NXT_ERROR;
}
static void
nxt_perl_psgi_request_handler(nxt_unit_request_info_t *req)
{
SV *env, *result;
nxt_int_t rc;
PerlInterpreter *my_perl;
nxt_perl_psgi_ctx_t *pctx;
pctx = req->ctx->data;
my_perl = pctx->my_perl;
pctx->req = req;
/*
* Create environ variable for perl sub "application".
* > sub application {
* > my ($environ) = @_;
*/
env = nxt_perl_psgi_env_create(my_perl, req);
if (nxt_slow_path(env == NULL)) {
nxt_unit_req_error(req,
"PSGI: Failed to create 'env' for Perl Application");
nxt_unit_request_done(req, NXT_UNIT_ERROR);
pctx->req = NULL;
return;
}
/* Call perl sub and get result as SV*. */
result = nxt_perl_psgi_call_var_application(my_perl, env, pctx->app,
req);
if (nxt_fast_path(SvOK(result) != 0 && SvROK(result) != 0)) {
if (SvTYPE(SvRV(result)) == SVt_PVAV) {
rc = nxt_perl_psgi_result_array(my_perl, result, req);
nxt_unit_request_done(req, rc);
pctx->req = NULL;
goto release;
}
if (SvTYPE(SvRV(result)) == SVt_PVCV) {
nxt_perl_psgi_result_cb(my_perl, result, req);
goto release;
}
}
nxt_unit_req_error(req, "PSGI: An unexpected response was received "
"from Perl Application");
nxt_unit_request_done(req, NXT_UNIT_ERROR);
pctx->req = NULL;
release:
SvREFCNT_dec(result);
SvREFCNT_dec(env);
}
static int
nxt_perl_psgi_ready_handler(nxt_unit_ctx_t *ctx)
{
int res;
uint32_t i;
nxt_perl_app_conf_t *c;
nxt_perl_psgi_ctx_t *pctx;
c = ctx->unit->data;
if (c->threads <= 1) {
return NXT_UNIT_OK;
}
for (i = 0; i < c->threads - 1; i++) {
pctx = &nxt_perl_psgi_ctxs[i];
pctx->ctx = ctx;
res = pthread_create(&pctx->thread, nxt_perl_psgi_thread_attr,
nxt_perl_psgi_thread_func, pctx);
if (nxt_fast_path(res == 0)) {
nxt_unit_debug(ctx, "thread #%d created", (int) (i + 1));
} else {
nxt_unit_alert(ctx, "thread #%d create failed: %s (%d)",
(int) (i + 1), strerror(res), res);
return NXT_UNIT_ERROR;
}
}
return NXT_UNIT_OK;
}
static void *
nxt_perl_psgi_thread_func(void *data)
{
nxt_unit_ctx_t *ctx;
nxt_perl_psgi_ctx_t *pctx;
pctx = data;
nxt_unit_debug(pctx->ctx, "worker thread start");
ctx = nxt_unit_ctx_alloc(pctx->ctx, pctx);
if (nxt_slow_path(ctx == NULL)) {
return NULL;
}
pctx->ctx = ctx;
PERL_SET_CONTEXT(pctx->my_perl);
(void) nxt_unit_run(ctx);
nxt_unit_done(ctx);
nxt_unit_debug(NULL, "worker thread end");
return NULL;
}
static int
nxt_perl_psgi_init_threads(nxt_perl_app_conf_t *c)
{
int rc;
uint32_t i;
static pthread_attr_t attr;
if (c->threads <= 1) {
return NXT_UNIT_OK;
}
if (c->thread_stack_size > 0) {
rc = pthread_attr_init(&attr);
if (nxt_slow_path(rc != 0)) {
nxt_unit_alert(NULL, "thread attr init failed: %s (%d)",
strerror(rc), rc);
return NXT_UNIT_ERROR;
}
rc = pthread_attr_setstacksize(&attr, c->thread_stack_size);
if (nxt_slow_path(rc != 0)) {
nxt_unit_alert(NULL, "thread attr set stack size failed: %s (%d)",
strerror(rc), rc);
return NXT_UNIT_ERROR;
}
nxt_perl_psgi_thread_attr = &attr;
}
nxt_perl_psgi_ctxs = nxt_unit_malloc(NULL, sizeof(nxt_perl_psgi_ctx_t)
* (c->threads - 1));
if (nxt_slow_path(nxt_perl_psgi_ctxs == NULL)) {
return NXT_UNIT_ERROR;
}
memset(nxt_perl_psgi_ctxs, 0, sizeof(nxt_perl_psgi_ctx_t)
* (c->threads - 1));
for (i = 0; i < c->threads - 1; i++) {
rc = nxt_perl_psgi_ctx_init(c->script, &nxt_perl_psgi_ctxs[i]);
if (nxt_slow_path(rc != NXT_UNIT_OK)) {
return NXT_UNIT_ERROR;
}
}
return NXT_UNIT_OK;
}
static void
nxt_perl_psgi_join_threads(nxt_unit_ctx_t *ctx, nxt_perl_app_conf_t *c)
{
int res;
uint32_t i;
nxt_perl_psgi_ctx_t *pctx;
if (nxt_perl_psgi_ctxs == NULL) {
return;
}
for (i = 0; i < c->threads - 1; i++) {
pctx = &nxt_perl_psgi_ctxs[i];
res = pthread_join(pctx->thread, NULL);
if (nxt_fast_path(res == 0)) {
nxt_unit_debug(ctx, "thread #%d joined", (int) (i + 1));
} else {
nxt_unit_alert(ctx, "thread #%d join failed: %s (%d)",
(int) (i + 1), strerror(res), res);
}
}
for (i = 0; i < c->threads - 1; i++) {
nxt_perl_psgi_ctx_free(&nxt_perl_psgi_ctxs[i]);
}
nxt_unit_free(NULL, nxt_perl_psgi_ctxs);
}
static void
nxt_perl_psgi_ctx_free(nxt_perl_psgi_ctx_t *pctx)
{
PerlInterpreter *my_perl;
my_perl = pctx->my_perl;
if (nxt_slow_path(my_perl == NULL)) {
return;
}
PERL_SET_CONTEXT(my_perl);
SvREFCNT_dec(pctx->arg_input.rv);
SvREFCNT_dec(pctx->arg_error.rv);
nxt_perl_psgi_io_release(my_perl, &pctx->arg_input);
nxt_perl_psgi_io_release(my_perl, &pctx->arg_error);
perl_destruct(my_perl);
perl_free(my_perl);
}
Reference in New Issue View Git Blame Copy Permalink