nghttp2/lib/nghttp2_stream.c

847 lines
21 KiB
C

/*
* nghttp2 - HTTP/2 C Library
*
* Copyright (c) 2012 Tatsuhiro Tsujikawa
*
* Permission is hereby granted, free of charge, to any person obtaining
* a copy of this software and associated documentation files (the
* "Software"), to deal in the Software without restriction, including
* without limitation the rights to use, copy, modify, merge, publish,
* distribute, sublicense, and/or sell copies of the Software, and to
* permit persons to whom the Software is furnished to do so, subject to
* the following conditions:
*
* The above copyright notice and this permission notice shall be
* included in all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
* LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
* OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include "nghttp2_stream.h"
#include <assert.h>
#include <stdio.h>
#include "nghttp2_helper.h"
void nghttp2_stream_init(nghttp2_stream *stream, int32_t stream_id,
uint8_t flags,
nghttp2_stream_state initial_state,
int32_t remote_initial_window_size,
int32_t local_initial_window_size,
void *stream_user_data)
{
nghttp2_map_entry_init(&stream->map_entry, stream_id);
stream->stream_id = stream_id;
stream->flags = flags;
stream->state = initial_state;
stream->shut_flags = NGHTTP2_SHUT_NONE;
stream->stream_user_data = stream_user_data;
stream->data = NULL;
stream->deferred_data = NULL;
stream->deferred_flags = NGHTTP2_DEFERRED_NONE;
stream->remote_window_size = remote_initial_window_size;
stream->local_window_size = local_initial_window_size;
stream->recv_window_size = 0;
stream->recv_reduction = 0;
stream->dep_prev = NULL;
stream->dep_next = NULL;
stream->sib_prev = NULL;
stream->sib_next = NULL;
stream->closed_next = NULL;
stream->stream_group = NULL;
stream->dpri = NGHTTP2_STREAM_DPRI_NO_DATA;
stream->num_substreams = 1;
stream->num_subtop = 0;
}
void nghttp2_stream_free(nghttp2_stream *stream)
{
nghttp2_outbound_item_free(stream->deferred_data);
free(stream->deferred_data);
/* We don't free stream->data. */
}
void nghttp2_stream_shutdown(nghttp2_stream *stream, nghttp2_shut_flag flag)
{
stream->shut_flags |= flag;
}
static int stream_push_data(nghttp2_stream *stream, nghttp2_pq *pq)
{
int rv;
ssize_t weight;
assert(stream->data);
assert(stream->data->queued == 0);
weight = nghttp2_stream_group_shared_wait(stream->stream_group);
if(stream->data->weight > weight) {
stream->data->weight = weight;
}
rv = nghttp2_pq_push(pq, stream->data);
if(rv != 0) {
return rv;
}
stream->data->queued = 1;
return 0;
}
static nghttp2_stream* stream_first_sib(nghttp2_stream *stream)
{
for(; stream->sib_prev; stream = stream->sib_prev);
return stream;
}
static nghttp2_stream* stream_last_sib(nghttp2_stream *stream)
{
for(; stream->sib_next; stream = stream->sib_next);
return stream;
}
static nghttp2_stream* stream_update_dep_length(nghttp2_stream *stream,
ssize_t delta)
{
stream->num_substreams += delta;
stream = stream_first_sib(stream);
if(stream->dep_prev) {
return stream_update_dep_length(stream->dep_prev, delta);
}
return stream;
}
static nghttp2_stream* stream_update_dep_both_length(nghttp2_stream *stream,
ssize_t delta_stream,
ssize_t delta_top)
{
stream->num_substreams += delta_stream;
stream->num_subtop += delta_top;
stream = stream_first_sib(stream);
if(stream->dep_prev) {
return stream_update_dep_both_length(stream->dep_prev, delta_stream,
delta_top);
}
return stream;
}
static void stream_update_dep_set_rest_stream_group
(nghttp2_stream *stream, nghttp2_stream_group *stream_group)
{
if(stream == NULL) {
return;
}
nghttp2_stream_group_remove_stream(stream->stream_group, stream);
nghttp2_stream_group_add_stream(stream_group, stream);
if(stream->dpri == NGHTTP2_STREAM_DPRI_TOP) {
stream->dpri = NGHTTP2_STREAM_DPRI_REST;
}
stream_update_dep_set_rest_stream_group(stream->sib_next, stream_group);
stream_update_dep_set_rest_stream_group(stream->dep_next, stream_group);
}
static void stream_update_dep_set_rest(nghttp2_stream *stream)
{
if(stream == NULL) {
return;
}
if(stream->dpri == NGHTTP2_STREAM_DPRI_REST) {
return;
}
stream->num_subtop = 0;
if(stream->dpri == NGHTTP2_STREAM_DPRI_TOP) {
stream->dpri = NGHTTP2_STREAM_DPRI_REST;
stream_update_dep_set_rest(stream->sib_next);
return;
}
stream_update_dep_set_rest(stream->sib_next);
stream_update_dep_set_rest(stream->dep_next);
}
/*
* Performs dfs starting |stream|, search stream which can become
* NGHTTP2_STREAM_DPRI_TOP and queues its data.
*
* This function returns the number of stream marked as
* NGHTTP2_STREAM_DPRI_TOP (including already marked as such) if it
* succeeds, or one of the following negative error codes:
*
* NGHTTP2_ERR_NOMEM
* Out of memory.
*/
static ssize_t stream_update_dep_set_top(nghttp2_stream *stream,
nghttp2_pq *pq)
{
ssize_t rv;
ssize_t num_top;
if(stream == NULL) {
return 0;
}
if(stream->dpri == NGHTTP2_STREAM_DPRI_TOP) {
rv = stream_update_dep_set_top(stream->sib_next, pq);
if(rv < 0) {
return rv;
}
stream->num_subtop = 1;
return stream->num_subtop + rv;
}
if(stream->dpri == NGHTTP2_STREAM_DPRI_REST) {
DEBUGF(fprintf(stderr, "stream: stream=%d data is top\n",
stream->stream_id));
if(!stream->data->queued) {
rv = stream_push_data(stream, pq);
if(rv != 0) {
return rv;
}
}
stream->dpri = NGHTTP2_STREAM_DPRI_TOP;
rv = stream_update_dep_set_top(stream->sib_next, pq);
if(rv < 0) {
return rv;
}
stream->num_subtop = 1;
return stream->num_subtop + rv;
}
assert(stream->dpri == NGHTTP2_STREAM_DPRI_NO_DATA);
rv = stream_update_dep_set_top(stream->sib_next, pq);
if(rv < 0) {
return rv;
}
num_top = rv;
rv = stream_update_dep_set_top(stream->dep_next, pq);
if(rv < 0) {
return rv;
}
stream->num_subtop = rv;
return stream->num_subtop + num_top;
}
static ssize_t stream_update_dep_on_attach_data(nghttp2_stream *stream,
nghttp2_pq *pq)
{
ssize_t rv;
nghttp2_stream *root_stream;
ssize_t old_num_subtop;
stream->dpri = NGHTTP2_STREAM_DPRI_REST;
stream_update_dep_set_rest(stream->dep_next);
root_stream = nghttp2_stream_get_dep_root(stream);
DEBUGF(fprintf(stderr, "root=%p, stream=%p\n", root_stream, stream));
old_num_subtop = root_stream->num_subtop;
rv = stream_update_dep_set_top(root_stream, pq);
if(rv < 0) {
return rv;
}
nghttp2_stream_group_update_num_top
(root_stream->stream_group, root_stream->num_subtop - old_num_subtop);
return 0;
}
static int stream_update_dep_on_detach_data(nghttp2_stream *stream,
nghttp2_pq *pq)
{
ssize_t rv;
nghttp2_stream *root_stream;
ssize_t old_num_subtop;
if(stream->dpri != NGHTTP2_STREAM_DPRI_TOP) {
stream->dpri = NGHTTP2_STREAM_DPRI_NO_DATA;
return 0;
}
stream->dpri = NGHTTP2_STREAM_DPRI_NO_DATA;
root_stream = nghttp2_stream_get_dep_root(stream);
old_num_subtop = root_stream->num_subtop;
rv = stream_update_dep_set_top(root_stream, pq);
if(rv < 0) {
return rv;
}
nghttp2_stream_group_update_num_top
(root_stream->stream_group, root_stream->num_subtop - old_num_subtop);
return 0;
}
int nghttp2_stream_attach_data(nghttp2_stream *stream,
nghttp2_outbound_item *data,
nghttp2_pq *pq)
{
assert(stream->data == NULL);
assert(stream->deferred_data == NULL);
stream->data = data;
DEBUGF(fprintf(stderr, "stream: stream=%d attach data=%p\n",
stream->stream_id, data));
return stream_update_dep_on_attach_data(stream, pq);
}
int nghttp2_stream_detach_data(nghttp2_stream *stream, nghttp2_pq *pq)
{
DEBUGF(fprintf(stderr, "stream: stream=%d detach data=%p\n",
stream->stream_id, stream->data));
stream->data = NULL;
return stream_update_dep_on_detach_data(stream, pq);
}
void nghttp2_stream_defer_data(nghttp2_stream *stream,
nghttp2_outbound_item *data,
uint8_t flags)
{
assert(stream->data);
assert(stream->data == data);
assert(stream->deferred_data == NULL);
stream->deferred_data = data;
stream->deferred_flags = flags;
stream->data = NULL;
}
int nghttp2_stream_detach_deferred_data(nghttp2_stream *stream,
nghttp2_pq *pq)
{
nghttp2_outbound_item *data;
assert(stream->data == NULL);
assert(stream->deferred_data);
data = stream->deferred_data;
stream->deferred_data = NULL;
stream->deferred_flags = NGHTTP2_DEFERRED_NONE;
return nghttp2_stream_attach_data(stream, data, pq);
}
static int update_initial_window_size
(int32_t *window_size_ptr,
int32_t new_initial_window_size,
int32_t old_initial_window_size)
{
int64_t new_window_size = (int64_t)(*window_size_ptr) +
new_initial_window_size - old_initial_window_size;
if(INT32_MIN > new_window_size ||
new_window_size > NGHTTP2_MAX_WINDOW_SIZE) {
return -1;
}
*window_size_ptr = new_window_size;
return 0;
}
int nghttp2_stream_update_remote_initial_window_size
(nghttp2_stream *stream,
int32_t new_initial_window_size,
int32_t old_initial_window_size)
{
return update_initial_window_size(&stream->remote_window_size,
new_initial_window_size,
old_initial_window_size);
}
int nghttp2_stream_update_local_initial_window_size
(nghttp2_stream *stream,
int32_t new_initial_window_size,
int32_t old_initial_window_size)
{
return update_initial_window_size(&stream->local_window_size,
new_initial_window_size,
old_initial_window_size);
}
void nghttp2_stream_promise_fulfilled(nghttp2_stream *stream)
{
stream->state = NGHTTP2_STREAM_OPENED;
}
nghttp2_stream* nghttp2_stream_get_dep_root(nghttp2_stream *stream)
{
for(;;) {
if(stream->sib_prev) {
stream = stream->sib_prev;
continue;
}
if(stream->dep_prev) {
stream = stream->dep_prev;
continue;
}
break;
}
return stream;
}
int nghttp2_stream_dep_subtree_find(nghttp2_stream *stream,
nghttp2_stream *target)
{
if(stream == NULL) {
return 0;
}
if(stream == target) {
return 1;
}
if(nghttp2_stream_dep_subtree_find(stream->sib_next, target)) {
return 1;
}
return nghttp2_stream_dep_subtree_find(stream->dep_next, target);
}
void nghttp2_stream_dep_insert(nghttp2_stream *dep_stream,
nghttp2_stream *stream)
{
nghttp2_stream *si;
assert(stream->data == NULL);
DEBUGF(fprintf(stderr,
"stream: dep_insert dep_stream(%p)=%d, stream(%p)=%d\n",
dep_stream, dep_stream->stream_id,
stream, stream->stream_id));
if(dep_stream->dep_next) {
for(si = dep_stream->dep_next; si; si = si->sib_next) {
stream->num_substreams += si->num_substreams;
}
stream->dep_next = dep_stream->dep_next;
stream->dep_next->dep_prev = stream;
}
dep_stream->dep_next = stream;
stream->dep_prev = dep_stream;
stream_update_dep_length(dep_stream, 1);
}
void nghttp2_stream_dep_add(nghttp2_stream *dep_stream,
nghttp2_stream *stream)
{
nghttp2_stream *last_sib;
assert(stream->data == NULL);
DEBUGF(fprintf(stderr,
"stream: dep_add dep_stream(%p)=%d, stream(%p)=%d\n",
dep_stream, dep_stream->stream_id,
stream, stream->stream_id));
stream_update_dep_length(dep_stream, 1);
if(dep_stream->dep_next == NULL) {
dep_stream->dep_next = stream;
stream->dep_prev = dep_stream;
return;
}
last_sib = stream_last_sib(dep_stream->dep_next);
last_sib->sib_next = stream;
stream->sib_prev = last_sib;
}
void nghttp2_stream_dep_remove(nghttp2_stream *stream)
{
nghttp2_stream *prev, *next, *dep_next;
DEBUGF(fprintf(stderr, "stream: dep_remove stream(%p)=%d\n",
stream, stream->stream_id));
prev = stream_first_sib(stream);
if(prev->dep_prev) {
stream_update_dep_length(prev->dep_prev, -1);
}
if(stream->sib_prev) {
prev = stream->sib_prev;
dep_next = stream->dep_next;
if(dep_next) {
dep_next->dep_prev = NULL;
prev->sib_next = dep_next;
dep_next->sib_prev = prev;
} else {
next = stream->sib_next;
prev->sib_next = next;
if(next) {
next->sib_prev = prev;
}
}
} else if(stream->dep_prev) {
prev = stream->dep_prev;
dep_next = stream->dep_next;
if(dep_next) {
prev->dep_next = dep_next;
dep_next->dep_prev = prev;
} else if(stream->sib_next) {
next = stream->sib_next;
prev->dep_next = next;
next->dep_prev = prev;
next->sib_prev = NULL;
} else {
prev->dep_next = NULL;
dep_next = NULL;
}
} else {
nghttp2_stream *si;
dep_next = NULL;
/* stream is a root of tree. Removing stream makes its
descendants a root of its own subtree. */
for(si = stream->dep_next; si;) {
next = si->sib_next;
si->dep_prev = NULL;
si->sib_prev = NULL;
si->sib_next = NULL;
si = next;
}
}
if(dep_next && stream->sib_next) {
prev = stream_last_sib(dep_next);
next = stream->sib_next;
prev->sib_next = next;
next->sib_prev = prev;
}
stream->num_substreams = 1;
stream->dep_prev = NULL;
stream->dep_next = NULL;
stream->sib_prev = NULL;
stream->sib_next = NULL;
}
int nghttp2_stream_dep_insert_subtree(nghttp2_stream *dep_stream,
nghttp2_stream *stream,
nghttp2_pq *pq)
{
nghttp2_stream *last_sib;
nghttp2_stream *dep_next;
nghttp2_stream *root_stream;
nghttp2_stream *si;
size_t delta_substreams;
ssize_t old_num_subtop;
ssize_t rv;
DEBUGF(fprintf(stderr, "stream: dep_insert_subtree dep_stream(%p)=%d "
"stream(%p)=%d\n",
dep_stream, dep_stream->stream_id,
stream, stream->stream_id));
delta_substreams = stream->num_substreams;
nghttp2_stream_group_update_num_top
(stream->stream_group, -stream->num_subtop);
stream_update_dep_set_rest_stream_group(stream, dep_stream->stream_group);
if(dep_stream->dep_next) {
dep_next = dep_stream->dep_next;
for(si = dep_stream->dep_next; si; si = si->sib_next) {
stream->num_substreams += si->num_substreams;
}
stream_update_dep_set_rest(dep_next);
dep_stream->dep_next = stream;
stream->dep_prev = dep_stream;
if(stream->dep_next) {
last_sib = stream_last_sib(stream->dep_next);
last_sib->sib_next = dep_next;
dep_next->sib_prev = last_sib;
dep_next->dep_prev = NULL;
} else {
stream->dep_next = dep_next;
dep_next->dep_prev = stream;
}
} else {
dep_stream->dep_next = stream;
stream->dep_prev = dep_stream;
}
root_stream = stream_update_dep_length(dep_stream, delta_substreams);
old_num_subtop = root_stream->num_subtop;
rv = stream_update_dep_set_top(root_stream, pq);
if(rv < 0) {
return rv;
}
nghttp2_stream_group_update_num_top
(root_stream->stream_group, root_stream->num_subtop - old_num_subtop);
return 0;
}
int nghttp2_stream_dep_add_subtree(nghttp2_stream *dep_stream,
nghttp2_stream *stream,
nghttp2_pq *pq)
{
nghttp2_stream *last_sib;
nghttp2_stream *root_stream;
ssize_t old_num_subtop;
ssize_t rv;
DEBUGF(fprintf(stderr, "stream: dep_add_subtree dep_stream(%p)=%d "
"stream(%p)=%d\n",
dep_stream, dep_stream->stream_id,
stream, stream->stream_id));
nghttp2_stream_group_update_num_top
(stream->stream_group, -stream->num_subtop);
stream_update_dep_set_rest_stream_group(stream, dep_stream->stream_group);
if(dep_stream->dep_next) {
last_sib = stream_last_sib(dep_stream->dep_next);
last_sib->sib_next = stream;
stream->sib_prev = last_sib;
} else {
dep_stream->dep_next = stream;
stream->dep_prev = dep_stream;
}
root_stream = stream_update_dep_length(dep_stream, stream->num_substreams);
old_num_subtop = root_stream->num_subtop;
rv = stream_update_dep_set_top(root_stream, pq);
if(rv < 0) {
return rv;
}
nghttp2_stream_group_update_num_top
(root_stream->stream_group, root_stream->num_subtop - old_num_subtop);
return 0;
}
void nghttp2_stream_dep_remove_subtree(nghttp2_stream *stream)
{
nghttp2_stream *prev, *next;
DEBUGF(fprintf(stderr, "stream: dep_remove_subtree stream(%p)=%d\n",
stream, stream->stream_id));
/* Removing subtree does not change stream_group->num_top */
if(stream->sib_prev) {
prev = stream->sib_prev;
prev->sib_next = stream->sib_next;
if(prev->sib_next) {
prev->sib_next->sib_prev = prev;
}
prev = stream_first_sib(prev);
if(prev->dep_prev) {
stream_update_dep_both_length(prev->dep_prev, -stream->num_substreams,
-stream->num_subtop);
}
} else if(stream->dep_prev) {
prev = stream->dep_prev;
next = stream->sib_next;
prev->dep_next = next;
if(next) {
next->dep_prev = prev;
next->sib_prev = NULL;
}
stream_update_dep_both_length(prev, -stream->num_substreams,
-stream->num_subtop);
}
stream->sib_prev = NULL;
stream->sib_next = NULL;
stream->dep_prev = NULL;
}
int nghttp2_stream_dep_make_root(nghttp2_stream_group *stream_group,
nghttp2_stream *stream,
nghttp2_pq *pq)
{
ssize_t rv;
DEBUGF(fprintf(stderr, "stream: dep_make_root new_stream_group(%p)=%d, "
"old_stream_group(%p)=%d, stream(%p)=%d\n",
stream_group, stream_group->pri_group_id,
stream->stream_group, stream->stream_group->pri_group_id,
stream, stream->stream_id));
/* First update num_top of old stream_group */
nghttp2_stream_group_update_num_top
(stream->stream_group, -stream->num_subtop);
stream_update_dep_set_rest_stream_group(stream, stream_group);
rv = stream_update_dep_set_top(stream, pq);
if(rv < 0) {
return rv;
}
nghttp2_stream_group_update_num_top(stream_group, stream->num_subtop);
return 0;
}
void nghttp2_stream_group_init(nghttp2_stream_group *stream_group,
int32_t pri_group_id,
int32_t weight)
{
nghttp2_map_entry_init(&stream_group->map_entry, pri_group_id);
stream_group->num_streams = 0;
stream_group->num_top = 0;
stream_group->pri_group_id = pri_group_id;
stream_group->weight = weight;
}
void nghttp2_stream_group_free(nghttp2_stream_group *stream_group)
{}
void nghttp2_stream_group_add_stream(nghttp2_stream_group *stream_group,
nghttp2_stream *stream)
{
DEBUGF(fprintf(stderr, "stream_group: stream_group(%p)=%d "
"add stream(%p)=%d\n",
stream_group, stream_group->pri_group_id,
stream, stream->stream_id));
stream->stream_group = stream_group;
++stream_group->num_streams;
}
void nghttp2_stream_group_remove_stream(nghttp2_stream_group *stream_group,
nghttp2_stream *stream)
{
DEBUGF(fprintf(stderr, "stream_group: stream_group(%p)=%d "
"remove stream(%p)=%d\n",
stream_group, stream_group->pri_group_id,
stream, stream->stream_id));
stream->stream_group = NULL;
--stream_group->num_streams;
}
void nghttp2_stream_group_update_num_top(nghttp2_stream_group *stream_group,
ssize_t delta)
{
DEBUGF(fprintf(stderr, "stream_group: stream_group(%p)=%d "
"update num_top current=%zd, delta=%zd, after=%zd\n",
stream_group, stream_group->pri_group_id,
stream_group->num_top, delta, stream_group->num_top + delta));
stream_group->num_top += delta;
assert(stream_group->num_top >= 0);
}
size_t nghttp2_stream_group_shared_wait(nghttp2_stream_group *stream_group)
{
if(stream_group->num_top == 0) {
return 1;
}
return nghttp2_max(1, stream_group->weight / stream_group->num_top);
}