From 51a9e4db9d86682b08a18b60e5b13fe073844032 Mon Sep 17 00:00:00 2001 From: Joris Vink Date: Mon, 1 Jul 2013 11:30:18 +0200 Subject: [PATCH] Implement SPDY WINDOW_UPDATE and SETTINGS. --- includes/kore.h | 3 ++ includes/spdy.h | 13 ++++++ src/connection.c | 1 + src/http.c | 5 ++- src/linux.c | 3 +- src/spdy.c | 113 +++++++++++++++++++++++++++++++++++++++++++---- 6 files changed, 128 insertions(+), 10 deletions(-) diff --git a/includes/kore.h b/includes/kore.h index 8052443..c3c1b0c 100644 --- a/includes/kore.h +++ b/includes/kore.h @@ -74,6 +74,7 @@ struct listener { #define CONN_READ_POSSIBLE 0x01 #define CONN_WRITE_POSSIBLE 0x02 +#define CONN_WRITE_BLOCK 0x04 struct connection { int fd; @@ -88,6 +89,7 @@ struct connection { z_stream z_inflate; u_int8_t deflate_started; z_stream z_deflate; + u_int32_t wsize_initial; TAILQ_HEAD(, netbuf) send_queue; TAILQ_HEAD(, netbuf) recv_queue; @@ -260,6 +262,7 @@ void kore_buf_appendb(struct kore_buf *, struct kore_buf *); struct spdy_header_block *spdy_header_block_create(int); struct spdy_stream *spdy_stream_lookup(struct connection *, u_int32_t); +int spdy_frame_data_done(struct netbuf *); int spdy_stream_get_header(struct spdy_header_block *, char *, char **); diff --git a/includes/spdy.h b/includes/spdy.h index 63c9fd4..e82bca5 100644 --- a/includes/spdy.h +++ b/includes/spdy.h @@ -49,6 +49,7 @@ struct spdy_stream { u_int32_t stream_id; u_int8_t flags; u_int8_t prio; + int32_t wsize; void *httpreq; struct spdy_header_block *hblock; @@ -65,6 +66,7 @@ extern const unsigned char SPDY_dictionary_txt[]; #define SPDY_SYNFRAME_SIZE 10 #define SPDY_ZLIB_DICT_SIZE 1423 #define SPDY_ZLIB_CHUNK 16348 +#define SPDY_INIT_WSIZE 65536 /* control frames */ #define SPDY_CTRL_FRAME_SYN_STREAM 1 @@ -77,6 +79,17 @@ extern const unsigned char SPDY_dictionary_txt[]; /* flags */ #define FLAG_FIN 0x01 #define FLAG_UNIDIRECTIONAL 0x02 +#define SPDY_STREAM_WILLCLOSE 0x04 + +/* settings */ +#define SETTINGS_UPLOAD_BANDWIDTH 1 +#define SETTINGS_DOWNLOAD_BANDWIDTH 2 +#define SETTINGS_ROUND_TRIP_TIME 3 +#define SETTINGS_MAX_CONCURRENT_STREAMS 4 +#define SETTINGS_CURRENT_CWND 5 +#define SETTINGS_DOWNLOAD_RETRANS_RATE 6 +#define SETTINGS_INITIAL_WINDOW_SIZE 7 +#define SETTINGS_CLIENT_CERTIFICATE_VECTOR_SIZE 8 #define SPDY_HBLOCK_NORMAL 0 #define SPDY_HBLOCK_DELAYED_ALLOC 1 diff --git a/src/connection.c b/src/connection.c index 4b4f366..ba3f936 100644 --- a/src/connection.c +++ b/src/connection.c @@ -76,6 +76,7 @@ kore_connection_accept(struct listener *l, struct connection **out) c->client_stream_id = 0; c->proto = CONN_PROTO_UNKNOWN; c->state = CONN_STATE_SSL_SHAKE; + c->wsize_initial = SPDY_INIT_WSIZE; TAILQ_INIT(&(c->send_queue)); TAILQ_INIT(&(c->recv_queue)); diff --git a/src/http.c b/src/http.c index 8e96d2e..f30a1fa 100644 --- a/src/http.c +++ b/src/http.c @@ -213,6 +213,7 @@ http_request_free(struct http_request *req) int http_response(struct http_request *req, int status, u_int8_t *d, u_int32_t len) { + struct netbuf *nb; u_int32_t hlen; struct http_header *hdr; struct kore_buf *buf; @@ -245,7 +246,9 @@ http_response(struct http_request *req, int status, u_int8_t *d, u_int32_t len) if (len > 0) { spdy_frame_send(req->owner, SPDY_DATA_FRAME, 0, len, req->stream, 0); - net_send_queue(req->owner, d, len, 0, NULL, NULL); + net_send_queue(req->owner, d, len, 0, &nb, + spdy_frame_data_done); + nb->extra = req->stream; } spdy_frame_send(req->owner, SPDY_DATA_FRAME, diff --git a/src/linux.c b/src/linux.c index a71cb85..ed9571f 100644 --- a/src/linux.c +++ b/src/linux.c @@ -126,7 +126,8 @@ kore_platform_event_wait(void) c = (struct connection *)events[i].data.ptr; if (events[i].events & EPOLLIN) c->flags |= CONN_READ_POSSIBLE; - if (events[i].events & EPOLLOUT) + if (events[i].events & EPOLLOUT && + !(c->flags & CONN_WRITE_BLOCK)) c->flags |= CONN_WRITE_POSSIBLE; if (!kore_connection_handle(c)) diff --git a/src/spdy.c b/src/spdy.c index a2d7229..08379a0 100644 --- a/src/spdy.c +++ b/src/spdy.c @@ -43,6 +43,9 @@ static int spdy_ctrl_frame_settings(struct netbuf *); static int spdy_ctrl_frame_ping(struct netbuf *); static int spdy_ctrl_frame_window(struct netbuf *); static int spdy_data_frame_recv(struct netbuf *); +static int spdy_frame_send_done(struct netbuf *); +static void spdy_update_wsize(struct connection *, + struct spdy_stream *, u_int32_t); static void spdy_stream_close(struct connection *, struct spdy_stream *); @@ -134,8 +137,9 @@ void spdy_frame_send(struct connection *c, u_int16_t type, u_int8_t flags, u_int32_t len, struct spdy_stream *s, u_int32_t misc) { - u_int8_t nb[12]; - u_int32_t length; + struct netbuf *nnb; + u_int8_t nb[12]; + u_int32_t length; kore_debug("spdy_frame_send(%p, %d, %d, %d, %p, %d)", c, type, flags, len, s, misc); @@ -170,10 +174,17 @@ spdy_frame_send(struct connection *c, u_int16_t type, u_int8_t flags, break; } - net_send_queue(c, nb, length, 0, NULL, NULL); + if (s != NULL && type == SPDY_DATA_FRAME) { + net_send_queue(c, nb, length, 0, &nnb, spdy_frame_send_done); + nnb->extra = s; + } else { + net_send_queue(c, nb, length, 0, NULL, NULL); + } - if ((flags & FLAG_FIN) && (s->flags & FLAG_FIN)) - spdy_stream_close(c, s); + if (s != NULL) { + if ((flags & FLAG_FIN) && (s->flags & FLAG_FIN)) + s->flags |= SPDY_STREAM_WILLCLOSE; + } } struct spdy_stream * @@ -313,6 +324,16 @@ spdy_stream_get_header(struct spdy_header_block *s, char *header, char **out) return (KORE_RESULT_ERROR); } +int +spdy_frame_data_done(struct netbuf *nb) +{ + struct connection *c = (struct connection *)nb->owner; + struct spdy_stream *s = (struct spdy_stream *)nb->extra; + + spdy_update_wsize(c, s, nb->len); + return (KORE_RESULT_OK); +} + static int spdy_ctrl_frame_syn_stream(struct netbuf *nb) { @@ -359,6 +380,7 @@ spdy_ctrl_frame_syn_stream(struct netbuf *nb) s = (struct spdy_stream *)kore_malloc(sizeof(*s)); s->prio = syn.prio; s->flags = ctrl.flags; + s->wsize = c->wsize_initial; s->stream_id = syn.stream_id; s->hblock = spdy_header_block_create(SPDY_HBLOCK_DELAYED_ALLOC); @@ -415,7 +437,8 @@ spdy_ctrl_frame_syn_stream(struct netbuf *nb) c->client_stream_id = s->stream_id; TAILQ_INSERT_TAIL(&(c->spdy_streams), s, list); - kore_debug("SPDY_SYN_STREAM: %d:%d:%d", s->stream_id, s->flags, s->prio); + kore_debug("SPDY_SYN_STREAM: %d:%d:%d", s->stream_id, + s->flags, s->prio); return (KORE_RESULT_OK); } @@ -423,7 +446,31 @@ spdy_ctrl_frame_syn_stream(struct netbuf *nb) static int spdy_ctrl_frame_settings(struct netbuf *nb) { - kore_debug("SPDY_SETTINGS (to be implemented)"); + u_int8_t *buf, flags; + u_int32_t ecount, i, id, val; + struct connection *c = (struct connection *)nb->owner; + + ecount = net_read32(nb->buf + SPDY_FRAME_SIZE); + kore_debug("SPDY_SETTINGS: %d settings present", ecount); + + buf = nb->buf + SPDY_FRAME_SIZE + 4; + for (i = 0; i < ecount; i++) { + flags = *(u_int8_t *)buf; + id = net_read32(buf) & 0xffffff; + val = net_read32(buf + 4); + + switch (id) { + case SETTINGS_INITIAL_WINDOW_SIZE: + c->wsize_initial = val; + break; + default: + kore_debug("no handling for setting %d:%d (%d)", + id, val, flags); + break; + } + + buf += 8; + } return (KORE_RESULT_OK); } @@ -450,12 +497,34 @@ spdy_ctrl_frame_ping(struct netbuf *nb) static int spdy_ctrl_frame_window(struct netbuf *nb) { - u_int32_t stream_id, window_size; + struct spdy_stream *s; + u_int32_t stream_id, window_size; + struct connection *c = (struct connection *)nb->owner; stream_id = net_read32(nb->buf + SPDY_FRAME_SIZE); window_size = net_read32(nb->buf + SPDY_FRAME_SIZE + 4); + if ((s = spdy_stream_lookup(c, stream_id)) == NULL) { + kore_debug("received WINDOW_UPDATE for nonexistant stream"); + kore_debug("stream_id: %d", stream_id); + return (KORE_RESULT_ERROR); + } + + if (s->flags & SPDY_STREAM_WILLCLOSE) { + kore_debug("received WINDOW_UPDATE for FIN stream"); + return (KORE_RESULT_ERROR); + } + kore_debug("SPDY_WINDOW_UPDATE: %d:%d", stream_id, window_size); + s->wsize += window_size; + if (s->wsize > 0) { + c->flags &= ~CONN_WRITE_BLOCK; + c->flags |= CONN_WRITE_POSSIBLE; + + kore_debug("can now send again (%d wsize)", s->wsize); + return (net_send_flush(c)); + } + return (KORE_RESULT_OK); } @@ -512,6 +581,34 @@ spdy_stream_close(struct connection *c, struct spdy_stream *s) kore_mem_free(s); } +static int +spdy_frame_send_done(struct netbuf *nb) +{ + u_int8_t flags; + struct connection *c = (struct connection *)nb->owner; + struct spdy_stream *s = (struct spdy_stream *)nb->extra; + + flags = *(u_int8_t *)(nb->buf + 4); + if ((flags & FLAG_FIN) && (s->flags & FLAG_FIN)) + spdy_stream_close(c, s); + + return (KORE_RESULT_OK); +} + +static void +spdy_update_wsize(struct connection *c, struct spdy_stream *s, u_int32_t len) +{ + s->wsize -= len; + kore_debug("spdy_update_wsize(): stream %d, window size %d", + s->stream_id, s->wsize); + + if (s->wsize <= 0) { + kore_debug("window size <= 0 for stream %d", s->stream_id); + c->flags &= ~CONN_WRITE_POSSIBLE; + c->flags |= CONN_WRITE_BLOCK; + } +} + static int spdy_zlib_inflate(struct connection *c, u_int8_t *src, size_t len, u_int8_t **dst, u_int32_t *olen)