ws: make pong sending lazy
No binary (build failed)
—
(log yok)
| @@ -632,6 +632,7 @@ static CURLcode ws_enc_add_cntrl(struct Curl_easy *data, | |
| size_t plen, | |
| unsigned int frame_type) | |
| { | |
| + | (void)data; |
| DEBUGASSERT(plen <= WS_MAX_CNTRL_LEN); | |
| if(plen > WS_MAX_CNTRL_LEN) | |
| return CURLE_BAD_FUNCTION_ARGUMENT; | |
| @@ -641,13 +642,6 @@ static CURLcode ws_enc_add_cntrl(struct Curl_easy *data, | |
| ws->pending.type = frame_type; | |
| ws->pending.payload_len = plen; | |
| memcpy(ws->pending.payload, payload, plen); | |
| - | |
| - | if(!ws->enc.payload_remain) { /* not in the middle of another frame */ |
| - | CURLcode result = ws_enc_add_pending(data, ws); |
| - | if(!result) |
| - | (void)ws_flush(data, ws, Curl_is_in_callback(data)); |
| - | return result; |
| - | } |
| return CURLE_OK; | |
| } | |
| @@ -716,7 +710,7 @@ static CURLcode ws_cw_write(struct Curl_easy *data, | |
| { | |
| struct ws_cw_ctx *ctx = writer->ctx; | |
| struct websocket *ws; | |
| - | CURLcode result; |
| + | CURLcode result = CURLE_OK; |
| CURL_TRC_WRITE(data, "ws_cw_write(len=%zu, type=%d)", nbytes, type); | |
| if(!(type & CLIENTWRITE_BODY) || data->set.ws_raw_mode) | |
| @@ -749,7 +743,8 @@ static CURLcode ws_cw_write(struct Curl_easy *data, | |
| if(result == CURLE_AGAIN) { | |
| /* insufficient amount of data, keep it for later. | |
| * we pretend to have written all since we have a copy */ | |
| - | return CURLE_OK; |
| + | result = CURLE_OK; |
| + | goto out; |
| } | |
| else if(result) { | |
| failf(data, "[WS] decode payload error %d", (int)result); | |
| @@ -760,10 +755,16 @@ static CURLcode ws_cw_write(struct Curl_easy *data, | |
| if((type & CLIENTWRITE_EOS) && !Curl_bufq_is_empty(&ctx->buf)) { | |
| failf(data, "[WS] decode ending with %zu frame bytes remaining", | |
| Curl_bufq_len(&ctx->buf)); | |
| - | return CURLE_RECV_ERROR; |
| + | result = CURLE_RECV_ERROR; |
| } | |
| - | return CURLE_OK; |
| + | out: |
| + | if(!result) { |
| + | result = ws_flush(data, ws, Curl_is_in_callback(data)); |
| + | if(result == CURLE_AGAIN) |
| + | result = CURLE_OK; |
| + | } |
| + | return result; |
| } | |
| /* WebSocket payload decoding client writer. */ | |
| @@ -1627,8 +1628,16 @@ CURLcode curl_ws_recv(CURL *curl, void *buffer, | |
| static CURLcode ws_flush(struct Curl_easy *data, struct websocket *ws, | |
| bool blocking) | |
| { | |
| + | CURLcode result; |
| + | |
| + | /* If there is space, add any pending control frame */ |
| + | if(Curl_bufq_len(&ws->sendbuf) < ws->sendbuf.chunk_size) { |
| + | result = ws_enc_add_pending(data, ws); |
| + | if(result && (result != CURLE_AGAIN)) |
| + | return result; |
| + | } |
| + | |
| if(!Curl_bufq_is_empty(&ws->sendbuf)) { | |
| - | CURLcode result; |
| const uint8_t *out; | |
| size_t outlen, n; | |
| #ifdef DEBUGBUILD | |
| @@ -24,11 +24,15 @@ | |
| # | |
| ########################################################################### | |
| # | |
| + | import base64 |
| + | import hashlib |
| import logging | |
| import os | |
| + | import re |
| import shutil | |
| import socket | |
| import subprocess | |
| + | import threading |
| import time | |
| from datetime import datetime, timedelta | |
| from typing import Dict | |
| @@ -220,3 +224,77 @@ class TestWebsockets: | |
| # The CONNECT through the proxy fails as it does not allow it | |
| r.check_exit_code(7) # CURLE_COULDNT_CONNECT | |
| assert r.stats[0]['http_connect'] == 403, f'{r}' | |
| + | |
| + | def test_20_11_crazy_pings(self, env: Env): |
| + | st = {} |
| + | send_rounds = 1 |
| + | |
| + | def srv(): |
| + | try: |
| + | with socket.socket() as s: |
| + | s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| + | s.bind(("127.0.0.1", 0)) |
| + | s.listen(1) |
| + | st["p"] = s.getsockname()[1] |
| + | |
| + | c, _ = s.accept() |
| + | c.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4096) |
| + | c.settimeout(Env.SERVER_TIMEOUT) |
| + | req = b"" |
| + | while b"\r\n\r\n" not in req: |
| + | req += c.recv(4096) |
| + | |
| + | k = re.search(rb"(?im)^Sec-WebSocket-Key:\s*(\S+)", req).group(1) |
| + | a = base64.b64encode( |
| + | hashlib.sha1(k + b"258EAFA5-E914-47DA-95CA-C5AB0DC85B11").digest() |
| + | ).decode() |
| + | c.sendall( |
| + | ( |
| + | "HTTP/1.1 101 Switching Protocols\r\n" |
| + | "Upgrade: websocket\r\n" |
| + | "Connection: Upgrade\r\n" |
| + | f"Sec-WebSocket-Accept: {a}\r\n\r\n" |
| + | ).encode() |
| + | ) |
| + | |
| + | f = b"\x89\x00" * 65536 # PING frames, many |
| + | try: |
| + | for _ in range(send_rounds): |
| + | c.sendall(f) |
| + | f = b"\x88\x00" # CLOSE frame |
| + | c.sendall(f) |
| + | except OSError: |
| + | pass |
| + | time.sleep(1) |
| + | c.close() |
| + | except OSError as e: |
| + | st["err"] = e |
| + | |
| + | curl = CurlClient(env=env) |
| + | send_rounds = 2 |
| + | threading.Thread(target=srv, daemon=True).start() |
| + | while "p" not in st and "err" not in st: |
| + | time.sleep(0.01) |
| + | assert "err" not in st, f'ws-ping server failed to start: {st["err"]}' |
| + | |
| + | url = f'ws://127.0.0.1:{st["p"]}/' |
| + | r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, |
| + | with_profile=True) |
| + | assert r.exit_code in [55, 56], f'{r.dump_logs()}' # SEND/RECV_ERROR |
| + | assert r.profile, f'{r}' |
| + | rss1 = r.profile.stats['rss'] / (1024 * 1024) |
| + | |
| + | st.clear() |
| + | send_rounds = 10 |
| + | threading.Thread(target=srv, daemon=True).start() |
| + | while "p" not in st and "err" not in st: |
| + | time.sleep(0.01) |
| + | assert "err" not in st, f'ws-ping server failed to start: {st["err"]}' |
| + | |
| + | url = f'ws://127.0.0.1:{st["p"]}/' |
| + | r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, |
| + | with_profile=True) |
| + | assert r.exit_code in [55, 56], f'{r.dump_logs()}' # SEND/RECV_ERROR |
| + | assert r.profile, f'{r}' |
| + | rss2 = r.profile.stats['rss'] / (1024 * 1024) |
| + | assert (rss1 * 1.1) >= rss2, 'bad memory increase' |