diff --git a/lib/cw-out.c b/lib/cw-out.c index ee7dc65dff..c2c2f49c03 100644 --- a/lib/cw-out.c +++ b/lib/cw-out.c @@ -74,6 +74,7 @@ typedef enum { CW_OUT_NONE, CW_OUT_BODY, + CW_OUT_BODY_0LEN, CW_OUT_HDS } cw_out_type; @@ -170,6 +171,7 @@ static void cw_get_writefunc(struct Curl_easy *data, cw_out_type otype, { switch(otype) { case CW_OUT_BODY: + case CW_OUT_BODY_0LEN: *pwcb = data->set.fwrite_func; *pwcb_data = data->set.out; *pmax_write = CURL_MAX_WRITE_SIZE; @@ -217,40 +219,50 @@ static CURLcode cw_out_ptr_flush(struct cw_out_ctx *ctx, } *pconsumed = 0; - while(blen && !ctx->paused) { - if(!flush_all && blen < min_write) - break; - wlen = max_write ? CURLMIN(blen, max_write) : blen; + if(otype == CW_OUT_BODY_0LEN) { + DEBUGASSERT(!blen); Curl_set_in_callback(data, TRUE); - nwritten = wcb((char *)CURL_UNCONST(buf), 1, wlen, wcb_data); + nwritten = wcb((char *)CURL_UNCONST(buf), 1, blen, wcb_data); Curl_set_in_callback(data, FALSE); - CURL_TRC_WRITE(data, "[OUT] wrote %zu %s bytes -> %zu", - wlen, (otype == CW_OUT_BODY) ? "body" : "header", - nwritten); - if(CURL_WRITEFUNC_PAUSE == nwritten) { - if(data->conn && data->conn->handler->flags & PROTOPT_NONETWORK) { - /* Protocols that work without network cannot be paused. This is - actually only FILE:// just now, and it cannot pause since the - transfer is not done using the "normal" procedure. */ - failf(data, "Write callback asked for PAUSE when not supported"); + CURL_TRC_WRITE(data, "[OUT] wrote %zu BODY bytes -> %zu", + blen, nwritten); + } + else { + while(blen && !ctx->paused) { + if(!flush_all && blen < min_write) + break; + wlen = max_write ? CURLMIN(blen, max_write) : blen; + Curl_set_in_callback(data, TRUE); + nwritten = wcb((char *)CURL_UNCONST(buf), 1, wlen, wcb_data); + Curl_set_in_callback(data, FALSE); + CURL_TRC_WRITE(data, "[OUT] wrote %zu %s bytes -> %zu", + wlen, (otype == CW_OUT_BODY) ? "body" : "header", + nwritten); + if(CURL_WRITEFUNC_PAUSE == nwritten) { + if(data->conn && data->conn->handler->flags & PROTOPT_NONETWORK) { + /* Protocols that work without network cannot be paused. This is + actually only FILE:// just now, and it cannot pause since the + transfer is not done using the "normal" procedure. */ + failf(data, "Write callback asked for PAUSE when not supported"); + return CURLE_WRITE_ERROR; + } + ctx->paused = TRUE; + CURL_TRC_WRITE(data, "[OUT] PAUSE requested by client"); + return Curl_xfer_pause_recv(data, TRUE); + } + else if(CURL_WRITEFUNC_ERROR == nwritten) { + failf(data, "client returned ERROR on write of %zu bytes", wlen); return CURLE_WRITE_ERROR; } - ctx->paused = TRUE; - CURL_TRC_WRITE(data, "[OUT] PAUSE requested by client"); - return Curl_xfer_pause_recv(data, TRUE); + else if(nwritten != wlen) { + failf(data, "Failure writing output to destination, " + "passed %zu returned %zd", wlen, nwritten); + return CURLE_WRITE_ERROR; + } + *pconsumed += nwritten; + blen -= nwritten; + buf += nwritten; } - else if(CURL_WRITEFUNC_ERROR == nwritten) { - failf(data, "client returned ERROR on write of %zu bytes", wlen); - return CURLE_WRITE_ERROR; - } - else if(nwritten != wlen) { - failf(data, "Failure writing output to destination, " - "passed %zu returned %zd", wlen, nwritten); - return CURLE_WRITE_ERROR; - } - *pconsumed += nwritten; - blen -= nwritten; - buf += nwritten; } return CURLE_OK; } @@ -413,7 +425,9 @@ static CURLcode cw_out_write(struct Curl_easy *data, if((type & CLIENTWRITE_BODY) || ((type & CLIENTWRITE_HEADER) && data->set.include_header)) { - result = cw_out_do_write(ctx, data, CW_OUT_BODY, flush_all, buf, blen); + cw_out_type otype = (!blen && (type & CLIENTWRITE_0LEN)) ? + CW_OUT_BODY_0LEN : CW_OUT_BODY; + result = cw_out_do_write(ctx, data, otype, flush_all, buf, blen); if(result) return result; } diff --git a/lib/sendf.h b/lib/sendf.h index e5cc600bfe..b3b8ef475d 100644 --- a/lib/sendf.h +++ b/lib/sendf.h @@ -50,6 +50,7 @@ #define CLIENTWRITE_1XX (1<<5) /* a 1xx response related HEADER */ #define CLIENTWRITE_TRAILER (1<<6) /* a trailer HEADER */ #define CLIENTWRITE_EOS (1<<7) /* End Of transfer download Stream */ +#define CLIENTWRITE_0LEN (1<<8) /* write even 0-length buffers */ /** * Write `len` bytes at `prt` to the client. `type` indicates what diff --git a/lib/ws.c b/lib/ws.c index 82b8a2c220..8abc0880ec 100644 --- a/lib/ws.c +++ b/lib/ws.c @@ -474,7 +474,7 @@ static CURLcode ws_dec_read_head(struct ws_decoder *dec, static CURLcode ws_dec_pass_payload(struct ws_decoder *dec, struct Curl_easy *data, struct bufq *inraw, - ws_write_payload *write_payload, + ws_write_payload *write_cb, void *write_ctx) { const unsigned char *inbuf; @@ -487,9 +487,9 @@ static CURLcode ws_dec_pass_payload(struct ws_decoder *dec, while(remain && Curl_bufq_peek(inraw, &inbuf, &inlen)) { if((curl_off_t)inlen > remain) inlen = (size_t)remain; - nwritten = write_payload(inbuf, inlen, dec->frame_age, dec->frame_flags, - dec->payload_offset, dec->payload_len, - write_ctx, &result); + nwritten = write_cb(inbuf, inlen, dec->frame_age, dec->frame_flags, + dec->payload_offset, dec->payload_len, + write_ctx, &result); if(nwritten < 0) return result; Curl_bufq_skip(inraw, (size_t)nwritten); @@ -505,7 +505,7 @@ static CURLcode ws_dec_pass_payload(struct ws_decoder *dec, static CURLcode ws_dec_pass(struct ws_decoder *dec, struct Curl_easy *data, struct bufq *inraw, - ws_write_payload *write_payload, + ws_write_payload *write_cb, void *write_ctx) { CURLcode result; @@ -535,8 +535,8 @@ static CURLcode ws_dec_pass(struct ws_decoder *dec, ssize_t nwritten; const unsigned char tmp = '\0'; /* special case of a 0 length frame, need to write once */ - nwritten = write_payload(&tmp, 0, dec->frame_age, dec->frame_flags, - 0, 0, write_ctx, &result); + nwritten = write_cb(&tmp, 0, dec->frame_age, dec->frame_flags, + 0, 0, write_ctx, &result); if(nwritten < 0) return result; dec->state = WS_DEC_INIT; @@ -544,7 +544,7 @@ static CURLcode ws_dec_pass(struct ws_decoder *dec, } FALLTHROUGH(); case WS_DEC_PAYLOAD: - result = ws_dec_pass_payload(dec, data, inraw, write_payload, write_ctx); + result = ws_dec_pass_payload(dec, data, inraw, write_cb, write_ctx); ws_dec_info(dec, data, "passing"); if(result) return result; @@ -631,7 +631,8 @@ static ssize_t ws_cw_dec_next(const unsigned char *buf, size_t buflen, update_meta(ws, frame_age, frame_flags, payload_offset, payload_len, buflen); - *err = Curl_cwriter_write(data, ctx->next_writer, ctx->cw_type, + *err = Curl_cwriter_write(data, ctx->next_writer, + (ctx->cw_type | CLIENTWRITE_0LEN), (const char *)buf, buflen); if(*err) return -1; @@ -943,7 +944,11 @@ static CURLcode cr_ws_read(struct Curl_easy *data, return result; ctx->read_eos = eos; - if(!nread) { + if(!Curl_bufq_is_empty(&ws->sendbuf)) { + /* client_read started a new frame, we disregard any eos reported */ + ctx->read_eos = FALSE; + } + else if(!nread) { /* nothing to convert, return this right away */ if(ctx->read_eos) ctx->eos = TRUE; @@ -952,7 +957,7 @@ static CURLcode cr_ws_read(struct Curl_easy *data, goto out; } - if(!ws->enc.payload_remain) { + if(!ws->enc.payload_remain && Curl_bufq_is_empty(&ws->sendbuf)) { /* encode the data as a new BINARY frame */ result = ws_enc_write_head(data, &ws->enc, CURLWS_BINARY, nread, &ws->sendbuf); @@ -1732,7 +1737,7 @@ CURL_EXTERN CURLcode curl_ws_start_frame(CURL *d, return CURLE_FAILED_INIT; } - CURL_TRC_WS(data, "curl_start_frame(flags=%x, frame_len=%" FMT_OFF_T, + CURL_TRC_WS(data, "curl_ws_start_frame(flags=%x, frame_len=%" FMT_OFF_T, flags, frame_len); if(!data->conn) { diff --git a/tests/http/test_20_websockets.py b/tests/http/test_20_websockets.py index c28b610f58..2612b0afe1 100644 --- a/tests/http/test_20_websockets.py +++ b/tests/http/test_20_websockets.py @@ -135,7 +135,7 @@ class TestWebsockets: if not client.exists(): pytest.skip(f'example client not built: {client.name}') url = f'ws://localhost:{env.ws_port}/' - r = client.run(args=[f'-{model}', '-m', str(0), '-M', str(10), url]) + r = client.run(args=[f'-{model}', '-m', str(1), '-M', str(10), url]) r.check_exit_code(0) @pytest.mark.parametrize("model", [ @@ -193,3 +193,17 @@ class TestWebsockets: large = 20000 r = client.run(args=[f'-{model}', '-c', str(count), '-m', str(large), url]) r.check_exit_code(0) + + @pytest.mark.parametrize("model", [ + pytest.param(1, id='multi_perform'), + pytest.param(2, id='curl_ws_send+recv'), + ]) + def test_20_09_data_empty(self, env: Env, ws_echo, model): + client = LocalClient(env=env, name='cli_ws_data') + if not client.exists(): + pytest.skip(f'example client not built: {client.name}') + url = f'ws://localhost:{env.ws_port}/' + count = 10 + large = 0 + r = client.run(args=[f'-{model}', '-c', str(count), '-m', str(large), url]) + r.check_exit_code(0) diff --git a/tests/libtest/cli_ws_data.c b/tests/libtest/cli_ws_data.c index 501c01eae2..f5cfc30afa 100644 --- a/tests/libtest/cli_ws_data.c +++ b/tests/libtest/cli_ws_data.c @@ -195,6 +195,8 @@ struct test_ws_m1_ctx { char *recv_buf; size_t send_len, nsent; size_t recv_len, nrcvd; + int read_calls; + int write_calls; }; static size_t test_ws_data_m1_read(char *buf, size_t nitems, size_t buflen, @@ -204,14 +206,21 @@ static size_t test_ws_data_m1_read(char *buf, size_t nitems, size_t buflen, size_t len = nitems * buflen; size_t left = ctx->send_len - ctx->nsent; - curl_mfprintf(stderr, "m1_read(len=%zu, left=%zu)\n", len, left); - if(left) { + ctx->read_calls++; + curl_mfprintf(stderr, "m1_read(len=%zu, left=%zu, call #%d)\n", + len, left, ctx->read_calls); + if(ctx->read_calls == 1) + curl_ws_start_frame(ctx->easy, CURLWS_BINARY, ctx->send_len); + + if(left || (ctx->read_calls == 1)) { if(left > len) left = len; memcpy(buf, ctx->send_buf + ctx->nsent, left); ctx->nsent += left; return left; } + curl_mfprintf(stderr, "m1_read, pausing, call #%d\n", + ctx->read_calls); return CURL_READFUNC_PAUSE; } @@ -221,7 +230,9 @@ static size_t test_ws_data_m1_write(char *buf, size_t nitems, size_t buflen, struct test_ws_m1_ctx *ctx = userdata; size_t len = nitems * buflen; - curl_mfprintf(stderr, "m1_write(len=%zu)\n", len); + ctx->write_calls++; + curl_mfprintf(stderr, "m1_write(len=%zu, call +%d)\n", + len, ctx->write_calls); if(len > (ctx->recv_len - ctx->nrcvd)) return CURL_WRITEFUNC_ERROR; memcpy(ctx->recv_buf + ctx->nrcvd, buf, len); @@ -240,6 +251,8 @@ static CURLcode test_ws_data_m1_echo(const char *url, struct test_ws_m1_ctx m1_ctx; size_t i, len; + curl_mfprintf(stderr, "test_ws_data_m1_echo(min=%zu, max=%zu)\n", + plen_min, plen_max); memset(&m1_ctx, 0, sizeof(m1_ctx)); m1_ctx.send_buf = calloc(1, plen_max + 1); m1_ctx.recv_buf = calloc(1, plen_max + 1); @@ -278,19 +291,25 @@ static CURLcode test_ws_data_m1_echo(const char *url, for(len = plen_min; len <= plen_max; ++len) { /* init what we want to send and expect to receive */ + curl_mfprintf(stderr, "m1_echo, iter len=%zu\n", len); m1_ctx.send_len = len; m1_ctx.nsent = 0; m1_ctx.recv_len = len; m1_ctx.nrcvd = 0; + m1_ctx.read_calls = 0; + m1_ctx.write_calls = 0; memset(m1_ctx.recv_buf, 0, plen_max); curl_easy_pause(m1_ctx.easy, CURLPAUSE_CONT); for(i = 0; i < count; ++i) { + curl_mfprintf(stderr, "m1_echo, iter len=%zu, count=%zu\n", len, i); while(1) { int still_running; /* keep number of running handles */ CURLMcode mc = curl_multi_perform(multi, &still_running); - if(!still_running || (m1_ctx.nrcvd == m1_ctx.recv_len)) { + if(!still_running || + (m1_ctx.read_calls && m1_ctx.write_calls && + m1_ctx.nrcvd == m1_ctx.recv_len)) { /* got the full echo back or failed */ break; }