Perl: creating input and error streams if closed.

Application handler can do anything with a stream object (including close it).
Once the stream is closed, Unit creates a new stream.

This closes #616 issue on GitHub.

docs/changes.xml

@@ -31,6 +31,12 @@ NGINX Unit updated to 1.27.0.
          date="" time=""
          packager="Andrei Belov <defan@nginx.com>">
 
+<change type="bugfix">
+<para>
+some Perl applications failed to process the request body, notably with Plack.
+</para>
+</change>
+
 </changes>
 
 

src/perl/nxt_perl_psgi.c

@@ -28,19 +28,15 @@ typedef struct {
 } nxt_perl_psgi_ctx_t;
 
 
-static long nxt_perl_psgi_io_input_read(PerlInterpreter *my_perl,
+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 long nxt_perl_psgi_io_input_write(PerlInterpreter *my_perl,
+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 long nxt_perl_psgi_io_input_flush(PerlInterpreter *my_perl,
-    nxt_perl_psgi_io_arg_t *arg);
 
-static long nxt_perl_psgi_io_error_read(PerlInterpreter *my_perl,
+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 long nxt_perl_psgi_io_error_write(PerlInterpreter *my_perl,
+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 long nxt_perl_psgi_io_error_flush(PerlInterpreter *my_perl,
-    nxt_perl_psgi_io_arg_t *arg);
 
 /*
 static void nxt_perl_psgi_xs_core_global_changes(PerlInterpreter *my_perl,
@@ -57,10 +53,8 @@ static SV *nxt_perl_psgi_call_method(PerlInterpreter *my_perl, SV *obj,
 /* For currect load XS modules */
 EXTERN_C void boot_DynaLoader(pTHX_ CV *cv);
 
-static nxt_int_t nxt_perl_psgi_io_input_init(PerlInterpreter *my_perl,
+static int nxt_perl_psgi_io_init(PerlInterpreter *my_perl,
-    nxt_perl_psgi_io_arg_t *arg);
+    nxt_perl_psgi_io_arg_t *arg, const char *mode, void *req);
-static nxt_int_t nxt_perl_psgi_io_error_init(PerlInterpreter *my_perl,
-    nxt_perl_psgi_io_arg_t *arg);
 
 static int nxt_perl_psgi_ctx_init(const char *script,
     nxt_perl_psgi_ctx_t *pctx);
@@ -125,20 +119,26 @@ NXT_EXPORT nxt_app_module_t  nxt_app_module = {
     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,
+};
 
-static long
+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)
 {
-    nxt_perl_psgi_ctx_t  *pctx;
+    return nxt_unit_request_read(arg->req, vbuf, length);
-
-    pctx = arg->pctx;
-
-    return nxt_unit_request_read(pctx->req, vbuf, length);
 }
 
 
-static long
+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)
 {
@@ -146,15 +146,7 @@ nxt_perl_psgi_io_input_write(PerlInterpreter *my_perl,
 }
 
 
-static long
+static SSize_t
-nxt_perl_psgi_io_input_flush(PerlInterpreter *my_perl,
-    nxt_perl_psgi_io_arg_t *arg)
-{
-    return 0;
-}
-
-
-static long
 nxt_perl_psgi_io_error_read(PerlInterpreter *my_perl,
     nxt_perl_psgi_io_arg_t *arg, void *vbuf, size_t length)
 {
@@ -162,25 +154,13 @@ nxt_perl_psgi_io_error_read(PerlInterpreter *my_perl,
 }
 
 
-static long
+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_perl_psgi_ctx_t  *pctx;
+    nxt_unit_req_error(arg->req, "Perl: %s", (const char*) vbuf);
 
-    pctx = arg->pctx;
+    return (SSize_t) length;
-
-    nxt_unit_req_error(pctx->req, "Perl: %s", (const char*) vbuf);
-
-    return (long) length;
-}
-
-
-static long
-nxt_perl_psgi_io_error_flush(PerlInterpreter *my_perl,
-    nxt_perl_psgi_io_arg_t *arg)
-{
-    return 0;
 }
 
 
@@ -461,70 +441,49 @@ nxt_perl_psgi_module_create(const char *script)
 }
 
 
-static nxt_int_t
+static int
-nxt_perl_psgi_io_input_init(PerlInterpreter *my_perl,
+nxt_perl_psgi_io_init(PerlInterpreter *my_perl,
-    nxt_perl_psgi_io_arg_t *arg)
+    nxt_perl_psgi_io_arg_t *arg, const char *mode, void *req)
 {
     SV      *io;
     PerlIO  *fp;
 
-    fp = nxt_perl_psgi_layer_stream_fp_create(aTHX_ arg, "r");
+    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;
+        }
 
-    if (nxt_slow_path(fp == NULL)) {
+        io = nxt_perl_psgi_layer_stream_io_create(aTHX_ fp);
-        return NXT_ERROR;
+        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;
     }
 
-    io = nxt_perl_psgi_layer_stream_io_create(aTHX_ fp);
+    arg->req = req;
 
-    if (nxt_slow_path(io == NULL)) {
+    return NXT_UNIT_OK;
-        nxt_perl_psgi_layer_stream_fp_destroy(aTHX_ fp);
-        return NXT_ERROR;
-    }
-
-    arg->io = io;
-    arg->fp = fp;
-    arg->flush = nxt_perl_psgi_io_input_flush;
-    arg->read = nxt_perl_psgi_io_input_read;
-    arg->write = nxt_perl_psgi_io_input_write;
-
-    return NXT_OK;
 }
 
 
-static nxt_int_t
+static void
-nxt_perl_psgi_io_error_init(PerlInterpreter *my_perl,
+nxt_perl_psgi_io_release(PerlInterpreter *my_perl, nxt_perl_psgi_io_arg_t *arg)
-    nxt_perl_psgi_io_arg_t *arg)
 {
-    SV      *io;
+    if (arg->io != NULL) {
-    PerlIO  *fp;
+        SvREFCNT_dec(arg->io);
-
+        arg->io = NULL;
-    fp = nxt_perl_psgi_layer_stream_fp_create(aTHX_ arg, "w");
-
-    if (nxt_slow_path(fp == NULL)) {
-        return NXT_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_ERROR;
-    }
-
-    arg->io = io;
-    arg->fp = fp;
-    arg->flush = nxt_perl_psgi_io_error_flush;
-    arg->read = nxt_perl_psgi_io_error_read;
-    arg->write = nxt_perl_psgi_io_error_write;
-
-    return NXT_OK;
 }
 
 
 static int
 nxt_perl_psgi_ctx_init(const char *script, nxt_perl_psgi_ctx_t *pctx)
 {
-    int              status;
+    int              status, res;
     char             *run_module;
     PerlInterpreter  *my_perl;
 
@@ -577,19 +536,27 @@ nxt_perl_psgi_ctx_init(const char *script, nxt_perl_psgi_ctx_t *pctx)
         goto fail;
     }
 
-    pctx->arg_input.pctx = pctx;
+    pctx->arg_input.rv = newSV_type(SVt_RV);
+    sv_setptrref(pctx->arg_input.rv, &pctx->arg_input);
+    SvSETMAGIC(pctx->arg_input.rv);
 
-    status = nxt_perl_psgi_io_input_init(my_perl, &pctx->arg_input);
+    pctx->arg_input.io_tab = &nxt_perl_psgi_io_tab_input;
-    if (nxt_slow_path(status != NXT_OK)) {
+
+    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.pctx = pctx;
+    pctx->arg_error.rv = newSV_type(SVt_RV);
+    sv_setptrref(pctx->arg_error.rv, &pctx->arg_error);
+    SvSETMAGIC(pctx->arg_error.rv);
 
-    status = nxt_perl_psgi_io_error_init(my_perl, &pctx->arg_error);
+    pctx->arg_error.io_tab = &nxt_perl_psgi_io_tab_error;
-    if (nxt_slow_path(status != NXT_OK)) {
+
-        nxt_unit_alert(NULL, "PSGI: Failed to init io.psgi.errors");
+    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;
     }
 
@@ -607,6 +574,9 @@ nxt_perl_psgi_ctx_init(const char *script, nxt_perl_psgi_ctx_t *pctx)
 
 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);
     }
@@ -614,6 +584,8 @@ fail:
     perl_destruct(my_perl);
     perl_free(my_perl);
 
+    pctx->my_perl = NULL;
+
     return NXT_UNIT_ERROR;
 }
 
@@ -672,21 +644,25 @@ nxt_perl_psgi_env_create(PerlInterpreter *my_perl,
                                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)));
+                               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)));
+                               SvREFCNT_inc(pctx->arg_error.io)));
+
     RC(nxt_perl_psgi_add_value(my_perl, hash_env, NL("psgi.multithread"),
-                                nxt_perl_psgi_ctxs != NULL
+                               nxt_perl_psgi_ctxs != NULL
-                                    ? &PL_sv_yes : &PL_sv_no));
+                                   ? &PL_sv_yes : &PL_sv_no));
     RC(nxt_perl_psgi_add_value(my_perl, hash_env, NL("psgi.multiprocess"),
-                                &PL_sv_yes));
+                               &PL_sv_yes));
     RC(nxt_perl_psgi_add_value(my_perl, hash_env, NL("psgi.run_once"),
-                                &PL_sv_no));
+                               &PL_sv_no));
     RC(nxt_perl_psgi_add_value(my_perl, hash_env, NL("psgi.nonblocking"),
-                                &PL_sv_no));
+                               &PL_sv_no));
     RC(nxt_perl_psgi_add_value(my_perl, hash_env, NL("psgi.streaming"),
-                                &PL_sv_yes));
+                               &PL_sv_yes));
 
     RC(nxt_perl_psgi_add_sptr(my_perl, hash_env, NL("QUERY_STRING"),
                               &r->query, r->query_length));
@@ -1447,11 +1423,11 @@ nxt_perl_psgi_ctx_free(nxt_perl_psgi_ctx_t *pctx)
 
     PERL_SET_CONTEXT(my_perl);
 
-    nxt_perl_psgi_layer_stream_io_destroy(aTHX_ pctx->arg_input.io);
+    SvREFCNT_dec(pctx->arg_input.rv);
-    nxt_perl_psgi_layer_stream_fp_destroy(aTHX_ pctx->arg_input.fp);
+    SvREFCNT_dec(pctx->arg_error.rv);
 
-    nxt_perl_psgi_layer_stream_io_destroy(aTHX_ pctx->arg_error.io);
+    nxt_perl_psgi_io_release(my_perl, &pctx->arg_input);
-    nxt_perl_psgi_layer_stream_fp_destroy(aTHX_ pctx->arg_error.fp);
+    nxt_perl_psgi_io_release(my_perl, &pctx->arg_error);
 
     perl_destruct(my_perl);
     perl_free(my_perl);

src/perl/nxt_perl_psgi_layer.c

@@ -93,11 +93,9 @@ nxt_perl_psgi_layer_stream_pushed(pTHX_ PerlIO *f, const char *mode, SV *arg,
     unit_stream = PerlIOSelf(f, nxt_perl_psgi_layer_stream_t);
 
     if (arg != NULL && SvOK(arg)) {
-        unit_stream->var = arg;
+        unit_stream->var = SvREFCNT_inc(arg);
     }
 
-    SvSETMAGIC(unit_stream->var);
-
     return PerlIOBase_pushed(aTHX_ f, mode, Nullsv, tab);
 }
 
@@ -105,11 +103,17 @@ nxt_perl_psgi_layer_stream_pushed(pTHX_ PerlIO *f, const char *mode, SV *arg,
 static IV
 nxt_perl_psgi_layer_stream_popped(pTHX_ PerlIO *f)
 {
+    nxt_perl_psgi_io_arg_t        *arg;
     nxt_perl_psgi_layer_stream_t  *unit_stream;
 
     unit_stream = PerlIOSelf(f, nxt_perl_psgi_layer_stream_t);
 
     if (unit_stream->var != NULL) {
+        arg = (void *) (intptr_t) SvIV(SvRV(unit_stream->var));
+
+        arg->io = NULL;
+        arg->fp = NULL;
+
         SvREFCNT_dec(unit_stream->var);
         unit_stream->var = Nullsv;
     }
@@ -181,9 +185,6 @@ nxt_perl_psgi_layer_stream_read(pTHX_ PerlIO *f, void *vbuf, Size_t count)
         return 0;
     }
 
-    unit_stream = PerlIOSelf(f, nxt_perl_psgi_layer_stream_t);
-    arg = (nxt_perl_psgi_io_arg_t *) (intptr_t) SvIV(SvRV(unit_stream->var));
-
     if ((PerlIOBase(f)->flags & PERLIO_F_CANREAD) == 0) {
         PerlIOBase(f)->flags |= PERLIO_F_ERROR;
 
@@ -192,7 +193,10 @@ nxt_perl_psgi_layer_stream_read(pTHX_ PerlIO *f, void *vbuf, Size_t count)
         return 0;
     }
 
-    return (SSize_t) arg->read(PERL_GET_CONTEXT, arg, vbuf, count);
+    unit_stream = PerlIOSelf(f, nxt_perl_psgi_layer_stream_t);
+    arg = (void *) (intptr_t) SvIV(SvRV(unit_stream->var));
+
+    return arg->io_tab->read(PERL_GET_CONTEXT, arg, vbuf, count);
 }
 
 
@@ -204,13 +208,10 @@ nxt_perl_psgi_layer_stream_write(pTHX_ PerlIO *f,
     nxt_perl_psgi_layer_stream_t  *unit_stream;
 
     if (PerlIOBase(f)->flags & PERLIO_F_CANWRITE) {
-
         unit_stream = PerlIOSelf(f, nxt_perl_psgi_layer_stream_t);
+        arg = (void *) (intptr_t) SvIV(SvRV(unit_stream->var));
 
-        arg = (nxt_perl_psgi_io_arg_t *)
+        return arg->io_tab->write(PERL_GET_CONTEXT, arg, vbuf, count);
-            (intptr_t) SvIV(SvRV(unit_stream->var));
-
-        return (SSize_t) arg->write(PERL_GET_CONTEXT, arg, vbuf, count);
     }
 
     return 0;
@@ -244,13 +245,7 @@ nxt_perl_psgi_layer_stream_fill(pTHX_ PerlIO *f)
 static IV
 nxt_perl_psgi_layer_stream_flush(pTHX_ PerlIO *f)
 {
-    nxt_perl_psgi_io_arg_t        *arg;
+    return 0;
-    nxt_perl_psgi_layer_stream_t  *unit_stream;
-
-    unit_stream = PerlIOSelf(f, nxt_perl_psgi_layer_stream_t);
-    arg = (nxt_perl_psgi_io_arg_t *) (intptr_t) SvIV(SvRV(unit_stream->var));
-
-    return (IV) arg->flush(PERL_GET_CONTEXT, arg);
 }
 
 
@@ -346,29 +341,11 @@ nxt_perl_psgi_layer_stream_init(pTHX)
 
 
 PerlIO *
-nxt_perl_psgi_layer_stream_fp_create(pTHX_ nxt_perl_psgi_io_arg_t *arg,
+nxt_perl_psgi_layer_stream_fp_create(pTHX_ SV *arg_rv,
     const char *mode)
 {
-    SV      *arg_rv;
+    return PerlIO_openn(aTHX_ "NGINX_Unit_PSGI_Layer_Stream",
-    PerlIO  *fp;
+                        mode, 0, 0, 0, NULL, 1, &arg_rv);
-
-    arg_rv = newSV_type(SVt_RV);
-
-    if (arg_rv == NULL) {
-        return NULL;
-    }
-
-    sv_setptrref(arg_rv, arg);
-
-    fp = PerlIO_openn(aTHX_ "NGINX_Unit_PSGI_Layer_Stream",
-                      mode, 0, 0, 0, NULL, 1, &arg_rv);
-
-    if (fp == NULL) {
-        SvREFCNT_dec(arg_rv);
-        return NULL;
-    }
-
-    return fp;
 }
 
 
@@ -403,10 +380,3 @@ nxt_perl_psgi_layer_stream_io_create(pTHX_ PerlIO *fp)
 
     return rvio;
 }
-
-
-void
-nxt_perl_psgi_layer_stream_io_destroy(pTHX_ SV *rvio)
-{
-    SvREFCNT_dec(rvio);
-}

src/perl/nxt_perl_psgi_layer.h

@@ -14,35 +14,35 @@
 #include <perliol.h>
 
 
+typedef struct nxt_perl_psgi_io_tab_s nxt_perl_psgi_io_tab_t;
 typedef struct nxt_perl_psgi_io_arg_s nxt_perl_psgi_io_arg_t;
 
-typedef long (*nxt_perl_psgi_io_read_f)(PerlInterpreter *my_perl,
+
-    nxt_perl_psgi_io_arg_t *arg, void *vbuf, size_t length);
+struct nxt_perl_psgi_io_tab_s {
-typedef long (*nxt_perl_psgi_io_write_f)(PerlInterpreter *my_perl,
+    SSize_t (*read)(PerlInterpreter *my_perl,
-    nxt_perl_psgi_io_arg_t *arg, const void *vbuf, size_t length);
+        nxt_perl_psgi_io_arg_t *arg, void *vbuf, size_t length);
-typedef long (*nxt_perl_psgi_io_arg_f)(PerlInterpreter *my_perl,
+    SSize_t (*write)(PerlInterpreter *my_perl,
-    nxt_perl_psgi_io_arg_t *arg);
+        nxt_perl_psgi_io_arg_t *arg, const void *vbuf, size_t length);
+};
 
 
 struct nxt_perl_psgi_io_arg_s {
-    SV                        *io;
+    SV                            *rv;
-    PerlIO                    *fp;
+    SV                            *io;
+    PerlIO                        *fp;
 
-    nxt_perl_psgi_io_arg_f    flush;
+    const nxt_perl_psgi_io_tab_t  *io_tab;
-    nxt_perl_psgi_io_read_f   read;
-    nxt_perl_psgi_io_write_f  write;
 
-    void                      *pctx;
+    void                          *req;
 };
 
 
 void nxt_perl_psgi_layer_stream_init(pTHX);
 
-PerlIO *nxt_perl_psgi_layer_stream_fp_create(pTHX_ nxt_perl_psgi_io_arg_t *arg,
+PerlIO *nxt_perl_psgi_layer_stream_fp_create(pTHX_ SV *arg_rv,
     const char *mode);
 void nxt_perl_psgi_layer_stream_fp_destroy(pTHX_ PerlIO *io);
 
 SV *nxt_perl_psgi_layer_stream_io_create(pTHX_ PerlIO *fp);
-void nxt_perl_psgi_layer_stream_io_destroy(pTHX_ SV *rvio);
 
 #endif /* _NXT_PERL_PSGI_LAYER_H_INCLUDED_ */

test/perl/input_buffered_read/psgi.pl

@@ -0,0 +1,17 @@
+use FileHandle;
+
+my $app = sub {
+    my ($environ) = @_;
+
+    $environ->{'psgi.input'}->read(my $body, 1024);
+
+    open my $io, "<", \$body;
+
+    # This makes $io work as FileHandle under 5.8, .10 and .11.
+    bless $io, 'FileHandle';
+
+    $environ->{'psgix.input.buffered'} = 1;
+    $environ->{'psgi.input'} = $io;
+
+    return ['200', ['Content-Length' => length $body], [$body]];
+};

test/perl/input_close/psgi.pl

@@ -0,0 +1,8 @@
+my $app = sub {
+    my ($environ) = @_;
+
+    $environ->{'psgi.input'}->read(my $body, 1024);
+    $environ->{'psgi.input'}->close();
+
+    return ['200', ['Content-Length' => length $body], [$body]];
+};

test/test_perl_application.py

@@ -100,6 +100,28 @@ class TestPerlApplication(TestApplicationPerl):
             self.post(body='0123456789')['body'] == '0123456789'
         ), 'input read parts'
 
+    def test_perl_application_input_buffered_read(self):
+        self.load('input_buffered_read')
+
+        assert (
+            self.post(body='012345')['body'] == '012345'
+        ), 'buffered read #1'
+
+        assert (
+            self.post(body='9876543210')['body'] == '9876543210'
+        ), 'buffered read #2'
+
+    def test_perl_application_input_close(self):
+        self.load('input_close')
+
+        assert (
+            self.post(body='012345')['body'] == '012345'
+        ), 'input close #1'
+
+        assert (
+            self.post(body='9876543210')['body'] == '9876543210'
+        ), 'input close #2'
+
     @pytest.mark.skip('not yet')
     def test_perl_application_input_read_offset(self):
         self.load('input_read_offset')