lsquic_qdec_hdl.c revision 5392f7a3
1/* Copyright (c) 2017 - 2019 LiteSpeed Technologies Inc.  See LICENSE. */
2/*
3 * lsquic_qdec_hdl.c -- QPACK decoder streams handler
4 */
5
6#include <assert.h>
7#include <errno.h>
8#include <inttypes.h>
9#include <stdlib.h>
10#include <string.h>
11#include <sys/queue.h>
12
13#include "lsquic.h"
14#include "lsquic_types.h"
15#include "lsquic_int_types.h"
16#include "lsquic_sfcw.h"
17#include "lsquic_varint.h"
18#include "lsquic_hq.h"
19#include "lsquic_hash.h"
20#include "lsquic_stream.h"
21#include "lsquic_frab_list.h"
22#include "lsqpack.h"
23#include "lsquic_http1x_if.h"
24#include "lsquic_qdec_hdl.h"
25#include "lsquic_mm.h"
26#include "lsquic_engine_public.h"
27#include "lsquic_headers.h"
28#include "lsquic_http1x_if.h"
29#include "lsquic_conn.h"
30
31#define LSQUIC_LOGGER_MODULE LSQLM_QDEC_HDL
32#define LSQUIC_LOG_CONN_ID lsquic_conn_log_cid(qdh->qdh_conn)
33#include "lsquic_logger.h"
34
35static void
36qdh_hblock_unblocked (void *);
37
38
39static int
40qdh_write_decoder (struct qpack_dec_hdl *qdh, const unsigned char *buf,
41                                                                size_t sz)
42{
43    ssize_t nw;
44
45    if (!(qdh->qdh_dec_sm_out && lsquic_frab_list_empty(&qdh->qdh_fral)))
46    {
47  write_to_frab:
48        if (0 == lsquic_frab_list_write(&qdh->qdh_fral,
49                                                (unsigned char *) buf, sz))
50        {
51            LSQ_DEBUG("wrote %zu bytes to frab list", sz);
52            lsquic_stream_wantwrite(qdh->qdh_dec_sm_out, 1);
53            return 0;
54        }
55        else
56        {
57            LSQ_INFO("error writing to frab list");
58            return -1;
59        }
60    }
61
62    nw = lsquic_stream_write(qdh->qdh_dec_sm_out, buf, sz);
63    if (nw < 0)
64    {
65        LSQ_INFO("error writing to outgoing QPACK decoder stream: %s",
66                                                        strerror(errno));
67        return -1;
68    }
69    LSQ_DEBUG("wrote %zd bytes to outgoing QPACK decoder stream", nw);
70
71    if ((size_t) nw == sz)
72        return 0;
73
74    buf = buf + nw;
75    sz -= (size_t) nw;
76    goto write_to_frab;
77}
78
79
80static int
81qdh_write_type (struct qpack_dec_hdl *qdh)
82{
83    int s;
84
85#ifndef NDEBUG
86    const char *env = getenv("LSQUIC_RND_VARINT_LEN");
87    if (env && atoi(env))
88    {
89        s = rand() & 3;
90        LSQ_DEBUG("writing %d-byte stream type", 1 << s);
91    }
92    else
93#endif
94        s = 0;
95
96    switch (s)
97    {
98    case 0:
99        return qdh_write_decoder(qdh,
100                                (unsigned char []) { HQUST_QPACK_DEC }, 1);
101    case 1:
102        return qdh_write_decoder(qdh,
103                            (unsigned char []) { 0x40, HQUST_QPACK_DEC }, 2);
104    case 2:
105        return qdh_write_decoder(qdh,
106                (unsigned char []) { 0x80, 0x00, 0x00, HQUST_QPACK_DEC }, 4);
107    default:
108        return qdh_write_decoder(qdh,
109                (unsigned char []) { 0xC0, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
110                                                        HQUST_QPACK_DEC }, 8);
111    }
112}
113
114
115static void
116qdh_begin_out (struct qpack_dec_hdl *qdh)
117{
118    if (0 != qdh_write_type(qdh))
119    {
120        LSQ_WARN("%s: could not write to decoder", __func__);
121        qdh->qdh_conn->cn_if->ci_internal_error(qdh->qdh_conn,
122                                        "cannot write to decoder stream");
123    }
124}
125
126
127int
128lsquic_qdh_init (struct qpack_dec_hdl *qdh, struct lsquic_conn *conn,
129                    int is_server, const struct lsquic_engine_public *enpub,
130                    unsigned dyn_table_size, unsigned max_risked_streams)
131{
132    qdh->qdh_conn = conn;
133    lsquic_frab_list_init(&qdh->qdh_fral, 0x400, NULL, NULL, NULL);
134    lsqpack_dec_init(&qdh->qdh_decoder, (void *) conn, dyn_table_size,
135                        max_risked_streams, qdh_hblock_unblocked);
136    qdh->qdh_flags |= QDH_INITIALIZED;
137    qdh->qdh_enpub = enpub;
138    if (qdh->qdh_enpub->enp_hsi_if == lsquic_http1x_if)
139    {
140        qdh->qdh_h1x_ctor_ctx = (struct http1x_ctor_ctx) {
141            .conn           = conn,
142            .max_headers_sz = 0x10000,  /* XXX */
143            .is_server      = is_server,
144        };
145        qdh->qdh_hsi_ctx = &qdh->qdh_h1x_ctor_ctx;
146    }
147    else
148        qdh->qdh_hsi_ctx = qdh->qdh_enpub->enp_hsi_ctx;
149    if (qdh->qdh_dec_sm_out)
150        qdh_begin_out(qdh);
151    if (qdh->qdh_enc_sm_in)
152        lsquic_stream_wantread(qdh->qdh_enc_sm_in, 1);
153    LSQ_DEBUG("initialized");
154    return 0;
155}
156
157
158void
159lsquic_qdh_cleanup (struct qpack_dec_hdl *qdh)
160{
161    if (qdh->qdh_flags & QDH_INITIALIZED)
162    {
163        LSQ_DEBUG("cleanup");
164        lsqpack_dec_cleanup(&qdh->qdh_decoder);
165        lsquic_frab_list_cleanup(&qdh->qdh_fral);
166        qdh->qdh_flags &= ~QDH_INITIALIZED;
167    }
168}
169
170static lsquic_stream_ctx_t *
171qdh_out_on_new (void *stream_if_ctx, struct lsquic_stream *stream)
172{
173    struct qpack_dec_hdl *const qdh = stream_if_ctx;
174    qdh->qdh_dec_sm_out = stream;
175    if (qdh->qdh_flags & QDH_INITIALIZED)
176        qdh_begin_out(qdh);
177    LSQ_DEBUG("initialized outgoing decoder stream");
178    return (void *) qdh;
179}
180
181
182static void
183qdh_out_on_write (struct lsquic_stream *stream, lsquic_stream_ctx_t *ctx)
184{
185    struct qpack_dec_hdl *const qdh = (void *) ctx;
186    struct lsquic_reader reader;
187    ssize_t nw;
188    unsigned char buf[LSQPACK_LONGEST_ICI];
189
190    if (lsqpack_dec_ici_pending(&qdh->qdh_decoder))
191    {
192        nw = lsqpack_dec_write_ici(&qdh->qdh_decoder, buf, sizeof(buf));
193        if (nw > 0)
194        {
195            if (0 == qdh_write_decoder(qdh, buf, nw))
196                LSQ_DEBUG("wrote %zd-byte TSS instruction", nw);
197            else
198                goto err;
199        }
200        else if (nw < 0)
201        {
202            LSQ_WARN("could not generate TSS instruction");
203            goto err;
204        }
205    }
206
207    if (lsquic_frab_list_empty(&qdh->qdh_fral))
208    {
209        LSQ_DEBUG("%s: nothing to write", __func__);
210        lsquic_stream_wantwrite(stream, 0);
211        return;
212    }
213
214    reader = (struct lsquic_reader) {
215        .lsqr_read  = lsquic_frab_list_read,
216        .lsqr_size  = lsquic_frab_list_size,
217        .lsqr_ctx   = &qdh->qdh_fral,
218    };
219
220    nw = lsquic_stream_writef(stream, &reader);
221    if (nw >= 0)
222    {
223        LSQ_DEBUG("wrote %zd bytes to stream", nw);
224        (void) lsquic_stream_flush(stream);
225        if (lsquic_frab_list_empty(&qdh->qdh_fral))
226            lsquic_stream_wantwrite(stream, 0);
227    }
228    else
229    {
230        LSQ_WARN("cannot write to stream: %s", strerror(errno));
231  err:
232        lsquic_stream_wantwrite(stream, 0);
233        qdh->qdh_conn->cn_if->ci_internal_error(qdh->qdh_conn,
234                                        "cannot write to stream");
235    }
236}
237
238
239static void
240qdh_out_on_close (struct lsquic_stream *stream, lsquic_stream_ctx_t *ctx)
241{
242    struct qpack_dec_hdl *const qdh = (void *) ctx;
243    qdh->qdh_dec_sm_out = NULL;
244    LSQ_DEBUG("closed outgoing decoder stream");
245}
246
247
248static void
249qdh_out_on_read (struct lsquic_stream *stream, lsquic_stream_ctx_t *ctx)
250{
251    assert(0);
252}
253
254
255static const struct lsquic_stream_if qdh_dec_sm_out_if =
256{
257    .on_new_stream  = qdh_out_on_new,
258    .on_read        = qdh_out_on_read,
259    .on_write       = qdh_out_on_write,
260    .on_close       = qdh_out_on_close,
261};
262const struct lsquic_stream_if *const lsquic_qdh_dec_sm_out_if =
263                                                    &qdh_dec_sm_out_if;
264
265
266static lsquic_stream_ctx_t *
267qdh_in_on_new (void *stream_if_ctx, struct lsquic_stream *stream)
268{
269    struct qpack_dec_hdl *const qdh = stream_if_ctx;
270    qdh->qdh_enc_sm_in = stream;
271    if (qdh->qdh_flags & QDH_INITIALIZED)
272        lsquic_stream_wantread(qdh->qdh_enc_sm_in, 1);
273    LSQ_DEBUG("initialized incoming encoder stream");
274    return (void *) qdh;
275}
276
277
278static size_t
279qdh_read_encoder_stream (void *ctx, const unsigned char *buf, size_t sz,
280                                                                    int fin)
281{
282    struct qpack_dec_hdl *const qdh = (void *) ctx;
283    const struct lsqpack_dec_err *qerr;
284    int s;
285
286    if (fin)
287    {
288        LSQ_INFO("encoder stream is closed");
289        qdh->qdh_conn->cn_if->ci_abort_error(qdh->qdh_conn, 1,
290            HEC_CLOSED_CRITICAL_STREAM, "Peer closed QPACK encoder stream");
291        goto end;
292    }
293
294    s = lsqpack_dec_enc_in(&qdh->qdh_decoder, buf, sz);
295    if (s != 0)
296    {
297        LSQ_INFO("error reading decoder stream");
298        qerr = lsqpack_dec_get_err_info(&qdh->qdh_decoder);
299        qdh->qdh_conn->cn_if->ci_abort_error(qdh->qdh_conn, 1,
300            HEC_QPACK_DECODER_STREAM_ERROR, "Error interpreting QPACK encoder "
301            "stream; offset %"PRIu64", line %d", qerr->off, qerr->line);
302        goto end;
303    }
304    if (qdh->qdh_dec_sm_out
305                    && lsqpack_dec_ici_pending(&qdh->qdh_decoder))
306        lsquic_stream_wantwrite(qdh->qdh_dec_sm_out, 1);
307
308    LSQ_DEBUG("successfully fed %zu bytes to QPACK decoder", sz);
309
310  end:
311    return sz;
312}
313
314
315static void
316qdh_in_on_read (struct lsquic_stream *stream, lsquic_stream_ctx_t *ctx)
317{
318    struct qpack_dec_hdl *const qdh = (void *) ctx;
319    ssize_t nread;
320
321    nread = lsquic_stream_readf(stream, qdh_read_encoder_stream, qdh);
322    if (nread <= 0)
323    {
324        if (nread < 0)
325        {
326            LSQ_WARN("cannot read from encoder stream: %s", strerror(errno));
327            qdh->qdh_conn->cn_if->ci_internal_error(qdh->qdh_conn,
328                                        "cannot read from encoder stream");
329        }
330        else
331        {
332            LSQ_INFO("encoder stream closed by peer: abort connection");
333            qdh->qdh_conn->cn_if->ci_abort_error(qdh->qdh_conn, 1,
334                HEC_CLOSED_CRITICAL_STREAM, "encoder stream closed");
335        }
336        lsquic_stream_wantread(stream, 0);
337    }
338}
339
340
341static void
342qdh_in_on_close (struct lsquic_stream *stream, lsquic_stream_ctx_t *ctx)
343{
344    struct qpack_dec_hdl *const qdh = (void *) ctx;
345    LSQ_DEBUG("closed incoming encoder stream");
346    qdh->qdh_enc_sm_in = NULL;
347}
348
349
350static void
351qdh_in_on_write (struct lsquic_stream *stream, lsquic_stream_ctx_t *ctx)
352{
353    assert(0);
354}
355
356
357static const struct lsquic_stream_if qdh_enc_sm_in_if =
358{
359    .on_new_stream  = qdh_in_on_new,
360    .on_read        = qdh_in_on_read,
361    .on_write       = qdh_in_on_write,
362    .on_close       = qdh_in_on_close,
363};
364const struct lsquic_stream_if *const lsquic_qdh_enc_sm_in_if =
365                                                    &qdh_enc_sm_in_if;
366
367
368static void
369qdh_hblock_unblocked (void *stream_p)
370{
371    struct lsquic_stream *const stream = stream_p;
372    struct qpack_dec_hdl *const qdh = lsquic_stream_get_qdh(stream);
373
374    LSQ_DEBUG("header block for stream %"PRIu64" unblocked", stream->id);
375    lsquic_stream_qdec_unblocked(stream);
376}
377
378
379static int
380qdh_supply_hset_to_stream (struct qpack_dec_hdl *qdh,
381            struct lsquic_stream *stream, struct lsqpack_header_list *qlist)
382{
383    const struct lsquic_hset_if *const hset_if = qdh->qdh_enpub->enp_hsi_if;
384    struct uncompressed_headers *uh = NULL;
385    const struct lsqpack_header *header;
386    enum lsquic_header_status st;
387    unsigned i;
388    void *hset;
389
390    hset = hset_if->hsi_create_header_set(qdh->qdh_hsi_ctx, 0);
391    if (!hset)
392    {
393        LSQ_INFO("call to hsi_create_header_set failed");
394        return -1;
395    }
396
397    LSQ_DEBUG("got header set for stream %"PRIu64, stream->id);
398
399    for (i = 0; i < qlist->qhl_count; ++i)
400    {
401        header = qlist->qhl_headers[i];
402        LSQ_DEBUG("%.*s: %.*s", header->qh_name_len, header->qh_name,
403                                        header->qh_value_len, header->qh_value);
404        st = hset_if->hsi_process_header(hset,
405                    header->qh_flags & QH_ID_SET ? 62 /* XXX: 62 */ + header->qh_static_id : 0,
406                    header->qh_name, header->qh_name_len,
407                    header->qh_value, header->qh_value_len);
408        if (st != LSQUIC_HDR_OK)
409            goto err;
410    }
411
412    lsqpack_dec_destroy_header_list(qlist);
413    st = hset_if->hsi_process_header(hset, 0, 0, 0, 0, 0);
414    if (st != LSQUIC_HDR_OK)
415        goto err;
416
417    uh = calloc(1, sizeof(*uh));
418    if (!uh)
419        goto err;
420    uh->uh_stream_id = stream->id;
421    uh->uh_oth_stream_id = 0;
422    uh->uh_weight = 0;
423    uh->uh_exclusive = -1;
424    if (hset_if == lsquic_http1x_if)
425        uh->uh_flags    |= UH_H1H;
426    uh->uh_hset = hset;
427    if (0 != lsquic_stream_uh_in(stream, uh))
428        goto err;
429    LSQ_DEBUG("converted qlist to hset and gave it to stream %"PRIu64,
430                                                                stream->id);
431    return 0;
432
433  err:
434    lsqpack_dec_destroy_header_list(qlist);
435    hset_if->hsi_discard_header_set(hset);
436    free(uh);
437    return -1;
438}
439
440
441static enum lsqpack_read_header_status
442qdh_header_read_results (struct qpack_dec_hdl *qdh,
443        struct lsquic_stream *stream, enum lsqpack_read_header_status rhs,
444        struct lsqpack_header_list *qlist, const unsigned char *dec_buf,
445        size_t dec_buf_sz)
446{
447    const struct lsqpack_dec_err *qerr;
448
449    if (rhs == LQRHS_DONE)
450    {
451        if (qlist)
452        {
453            if (0 != qdh_supply_hset_to_stream(qdh, stream, qlist))
454                return LQRHS_ERROR;
455            if (qdh->qdh_dec_sm_out)
456            {
457                if (dec_buf_sz
458                    && 0 != qdh_write_decoder(qdh, dec_buf, dec_buf_sz))
459                {
460                    return LQRHS_ERROR;
461                }
462                if (dec_buf_sz || lsqpack_dec_ici_pending(&qdh->qdh_decoder))
463                    lsquic_stream_wantwrite(qdh->qdh_dec_sm_out, 1);
464            }
465        }
466        else
467        {
468            assert(0);  /* XXX TODO What do we do here? */
469            return LQRHS_ERROR;
470        }
471    }
472    else if (rhs == LQRHS_ERROR)
473    {
474        qerr = lsqpack_dec_get_err_info(&qdh->qdh_decoder);
475        qdh->qdh_conn->cn_if->ci_abort_error(qdh->qdh_conn, 1,
476            HEC_QPACK_DECOMPRESSION_FAILED, "QPACK decompression error; "
477            "stream %"PRIu64", offset %"PRIu64", line %d", qerr->stream_id,
478            qerr->off, qerr->line);
479    }
480
481    return rhs;
482}
483
484
485enum lsqpack_read_header_status
486lsquic_qdh_header_in_begin (struct qpack_dec_hdl *qdh,
487                        struct lsquic_stream *stream, uint64_t header_size,
488                        const unsigned char **buf, size_t bufsz)
489{
490    enum lsqpack_read_header_status rhs;
491    struct lsqpack_header_list *qlist;
492    size_t dec_buf_sz;
493    unsigned char dec_buf[LSQPACK_LONGEST_HEADER_ACK];
494
495    if (qdh->qdh_flags & QDH_INITIALIZED)
496    {
497        dec_buf_sz = sizeof(dec_buf);
498        rhs = lsqpack_dec_header_in(&qdh->qdh_decoder, stream, stream->id,
499                        header_size, buf, bufsz, &qlist, dec_buf, &dec_buf_sz);
500        return qdh_header_read_results(qdh, stream, rhs, qlist, dec_buf,
501                                                                dec_buf_sz);
502    }
503    else
504    {
505        LSQ_WARN("not initialized: cannot process header block");
506        return LQRHS_ERROR;
507    }
508
509}
510
511
512enum lsqpack_read_header_status
513lsquic_qdh_header_in_continue (struct qpack_dec_hdl *qdh,
514        struct lsquic_stream *stream, const unsigned char **buf, size_t bufsz)
515{
516    enum lsqpack_read_header_status rhs;
517    struct lsqpack_header_list *qlist;
518    size_t dec_buf_sz;
519    unsigned char dec_buf[LSQPACK_LONGEST_HEADER_ACK];
520
521    if (qdh->qdh_flags & QDH_INITIALIZED)
522    {
523        dec_buf_sz = sizeof(dec_buf);
524        rhs = lsqpack_dec_header_read(&qdh->qdh_decoder, stream,
525                                    buf, bufsz, &qlist, dec_buf, &dec_buf_sz);
526        return qdh_header_read_results(qdh, stream, rhs, qlist, dec_buf,
527                                                                dec_buf_sz);
528    }
529    else
530    {
531        LSQ_WARN("not initialized: cannot process header block");
532        return LQRHS_ERROR;
533    }
534}
535
536
537void
538lsquic_qdh_unref_stream (struct qpack_dec_hdl *qdh,
539                                                struct lsquic_stream *stream)
540{
541    if (0 == lsqpack_dec_unref_stream(&qdh->qdh_decoder, stream))
542        LSQ_DEBUG("unreffed stream %"PRIu64, stream->id);
543    else
544        LSQ_WARN("cannot unref stream %"PRIu64, stream->id);
545}
546
547
548void
549lsquic_qdh_cancel_stream (struct qpack_dec_hdl *qdh,
550                                                struct lsquic_stream *stream)
551{
552    ssize_t nw;
553    unsigned char buf[LSQPACK_LONGEST_CANCEL];
554
555    nw = lsqpack_dec_cancel_stream(&qdh->qdh_decoder, stream, buf, sizeof(buf));
556    if (nw > 0)
557    {
558        if (0 == qdh_write_decoder(qdh, buf, nw))
559            LSQ_DEBUG("cancelled stream %"PRIu64" and wrote %zd-byte Cancel "
560                "Stream instruction to the decoder stream", stream->id, nw);
561    }
562    else if (nw == 0)
563        LSQ_WARN("cannot cancel stream %"PRIu64" -- not found", stream->id);
564    else
565    {
566        LSQ_WARN("cannot cancel stream %"PRIu64" -- not enough buffer space "
567            "to encode Cancel Stream instructin", stream->id);
568        lsquic_qdh_unref_stream(qdh, stream);
569    }
570}
571