new repo

[bytom/vapor.git] / vendor / golang.org / x / net / http2 / writesched_priority.go

// Copyright 2016 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package http2

import (
        "fmt"
        "math"
        "sort"
)

// RFC 7540, Section 5.3.5: the default weight is 16.
const priorityDefaultWeight = 15 // 16 = 15 + 1

// PriorityWriteSchedulerConfig configures a priorityWriteScheduler.
type PriorityWriteSchedulerConfig struct {
        // MaxClosedNodesInTree controls the maximum number of closed streams to
        // retain in the priority tree. Setting this to zero saves a small amount
        // of memory at the cost of performance.
        //
        // See RFC 7540, Section 5.3.4:
        //   "It is possible for a stream to become closed while prioritization
        //   information ... is in transit. ... This potentially creates suboptimal
        //   prioritization, since the stream could be given a priority that is
        //   different from what is intended. To avoid these problems, an endpoint
        //   SHOULD retain stream prioritization state for a period after streams
        //   become closed. The longer state is retained, the lower the chance that
        //   streams are assigned incorrect or default priority values."
        MaxClosedNodesInTree int

        // MaxIdleNodesInTree controls the maximum number of idle streams to
        // retain in the priority tree. Setting this to zero saves a small amount
        // of memory at the cost of performance.
        //
        // See RFC 7540, Section 5.3.4:
        //   Similarly, streams that are in the "idle" state can be assigned
        //   priority or become a parent of other streams. This allows for the
        //   creation of a grouping node in the dependency tree, which enables
        //   more flexible expressions of priority. Idle streams begin with a
        //   default priority (Section 5.3.5).
        MaxIdleNodesInTree int

        // ThrottleOutOfOrderWrites enables write throttling to help ensure that
        // data is delivered in priority order. This works around a race where
        // stream B depends on stream A and both streams are about to call Write
        // to queue DATA frames. If B wins the race, a naive scheduler would eagerly
        // write as much data from B as possible, but this is suboptimal because A
        // is a higher-priority stream. With throttling enabled, we write a small
        // amount of data from B to minimize the amount of bandwidth that B can
        // steal from A.
        ThrottleOutOfOrderWrites bool
}

// NewPriorityWriteScheduler constructs a WriteScheduler that schedules
// frames by following HTTP/2 priorities as described in RFC 7540 Section 5.3.
// If cfg is nil, default options are used.
func NewPriorityWriteScheduler(cfg *PriorityWriteSchedulerConfig) WriteScheduler {
        if cfg == nil {
                // For justification of these defaults, see:
                // https://docs.google.com/document/d/1oLhNg1skaWD4_DtaoCxdSRN5erEXrH-KnLrMwEpOtFY
                cfg = &PriorityWriteSchedulerConfig{
                        MaxClosedNodesInTree:     10,
                        MaxIdleNodesInTree:       10,
                        ThrottleOutOfOrderWrites: false,
                }
        }

        ws := &priorityWriteScheduler{
                nodes:                make(map[uint32]*priorityNode),
                maxClosedNodesInTree: cfg.MaxClosedNodesInTree,
                maxIdleNodesInTree:   cfg.MaxIdleNodesInTree,
                enableWriteThrottle:  cfg.ThrottleOutOfOrderWrites,
        }
        ws.nodes[0] = &ws.root
        if cfg.ThrottleOutOfOrderWrites {
                ws.writeThrottleLimit = 1024
        } else {
                ws.writeThrottleLimit = math.MaxInt32
        }
        return ws
}

type priorityNodeState int

const (
        priorityNodeOpen priorityNodeState = iota
        priorityNodeClosed
        priorityNodeIdle
)

// priorityNode is a node in an HTTP/2 priority tree.
// Each node is associated with a single stream ID.
// See RFC 7540, Section 5.3.
type priorityNode struct {
        q            writeQueue        // queue of pending frames to write
        id           uint32            // id of the stream, or 0 for the root of the tree
        weight       uint8             // the actual weight is weight+1, so the value is in [1,256]
        state        priorityNodeState // open | closed | idle
        bytes        int64             // number of bytes written by this node, or 0 if closed
        subtreeBytes int64             // sum(node.bytes) of all nodes in this subtree

        // These links form the priority tree.
        parent     *priorityNode
        kids       *priorityNode // start of the kids list
        prev, next *priorityNode // doubly-linked list of siblings
}

func (n *priorityNode) setParent(parent *priorityNode) {
        if n == parent {
                panic("setParent to self")
        }
        if n.parent == parent {
                return
        }
        // Unlink from current parent.
        if parent := n.parent; parent != nil {
                if n.prev == nil {
                        parent.kids = n.next
                } else {
                        n.prev.next = n.next
                }
                if n.next != nil {
                        n.next.prev = n.prev
                }
        }
        // Link to new parent.
        // If parent=nil, remove n from the tree.
        // Always insert at the head of parent.kids (this is assumed by walkReadyInOrder).
        n.parent = parent
        if parent == nil {
                n.next = nil
                n.prev = nil
        } else {
                n.next = parent.kids
                n.prev = nil
                if n.next != nil {
                        n.next.prev = n
                }
                parent.kids = n
        }
}

func (n *priorityNode) addBytes(b int64) {
        n.bytes += b
        for ; n != nil; n = n.parent {
                n.subtreeBytes += b
        }
}

// walkReadyInOrder iterates over the tree in priority order, calling f for each node
// with a non-empty write queue. When f returns true, this funcion returns true and the
// walk halts. tmp is used as scratch space for sorting.
//
// f(n, openParent) takes two arguments: the node to visit, n, and a bool that is true
// if any ancestor p of n is still open (ignoring the root node).
func (n *priorityNode) walkReadyInOrder(openParent bool, tmp *[]*priorityNode, f func(*priorityNode, bool) bool) bool {
        if !n.q.empty() && f(n, openParent) {
                return true
        }
        if n.kids == nil {
                return false
        }

        // Don't consider the root "open" when updating openParent since
        // we can't send data frames on the root stream (only control frames).
        if n.id != 0 {
                openParent = openParent || (n.state == priorityNodeOpen)
        }

        // Common case: only one kid or all kids have the same weight.
        // Some clients don't use weights; other clients (like web browsers)
        // use mostly-linear priority trees.
        w := n.kids.weight
        needSort := false
        for k := n.kids.next; k != nil; k = k.next {
                if k.weight != w {
                        needSort = true
                        break
                }
        }
        if !needSort {
                for k := n.kids; k != nil; k = k.next {
                        if k.walkReadyInOrder(openParent, tmp, f) {
                                return true
                        }
                }
                return false
        }

        // Uncommon case: sort the child nodes. We remove the kids from the parent,
        // then re-insert after sorting so we can reuse tmp for future sort calls.
        *tmp = (*tmp)[:0]
        for n.kids != nil {
                *tmp = append(*tmp, n.kids)
                n.kids.setParent(nil)
        }
        sort.Sort(sortPriorityNodeSiblings(*tmp))
        for i := len(*tmp) - 1; i >= 0; i-- {
                (*tmp)[i].setParent(n) // setParent inserts at the head of n.kids
        }
        for k := n.kids; k != nil; k = k.next {
                if k.walkReadyInOrder(openParent, tmp, f) {
                        return true
                }
        }
        return false
}

type sortPriorityNodeSiblings []*priorityNode

func (z sortPriorityNodeSiblings) Len() int      { return len(z) }
func (z sortPriorityNodeSiblings) Swap(i, k int) { z[i], z[k] = z[k], z[i] }
func (z sortPriorityNodeSiblings) Less(i, k int) bool {
        // Prefer the subtree that has sent fewer bytes relative to its weight.
        // See sections 5.3.2 and 5.3.4.
        wi, bi := float64(z[i].weight+1), float64(z[i].subtreeBytes)
        wk, bk := float64(z[k].weight+1), float64(z[k].subtreeBytes)
        if bi == 0 && bk == 0 {
                return wi >= wk
        }
        if bk == 0 {
                return false
        }
        return bi/bk <= wi/wk
}

type priorityWriteScheduler struct {
        // root is the root of the priority tree, where root.id = 0.
        // The root queues control frames that are not associated with any stream.
        root priorityNode

        // nodes maps stream ids to priority tree nodes.
        nodes map[uint32]*priorityNode

        // maxID is the maximum stream id in nodes.
        maxID uint32

        // lists of nodes that have been closed or are idle, but are kept in
        // the tree for improved prioritization. When the lengths exceed either
        // maxClosedNodesInTree or maxIdleNodesInTree, old nodes are discarded.
        closedNodes, idleNodes []*priorityNode

        // From the config.
        maxClosedNodesInTree int
        maxIdleNodesInTree   int
        writeThrottleLimit   int32
        enableWriteThrottle  bool

        // tmp is scratch space for priorityNode.walkReadyInOrder to reduce allocations.
        tmp []*priorityNode

        // pool of empty queues for reuse.
        queuePool writeQueuePool
}

func (ws *priorityWriteScheduler) OpenStream(streamID uint32, options OpenStreamOptions) {
        // The stream may be currently idle but cannot be opened or closed.
        if curr := ws.nodes[streamID]; curr != nil {
                if curr.state != priorityNodeIdle {
                        panic(fmt.Sprintf("stream %d already opened", streamID))
                }
                curr.state = priorityNodeOpen
                return
        }

        // RFC 7540, Section 5.3.5:
        //  "All streams are initially assigned a non-exclusive dependency on stream 0x0.
        //  Pushed streams initially depend on their associated stream. In both cases,
        //  streams are assigned a default weight of 16."
        parent := ws.nodes[options.PusherID]
        if parent == nil {
                parent = &ws.root
        }
        n := &priorityNode{
                q:      *ws.queuePool.get(),
                id:     streamID,
                weight: priorityDefaultWeight,
                state:  priorityNodeOpen,
        }
        n.setParent(parent)
        ws.nodes[streamID] = n
        if streamID > ws.maxID {
                ws.maxID = streamID
        }
}

func (ws *priorityWriteScheduler) CloseStream(streamID uint32) {
        if streamID == 0 {
                panic("violation of WriteScheduler interface: cannot close stream 0")
        }
        if ws.nodes[streamID] == nil {
                panic(fmt.Sprintf("violation of WriteScheduler interface: unknown stream %d", streamID))
        }
        if ws.nodes[streamID].state != priorityNodeOpen {
                panic(fmt.Sprintf("violation of WriteScheduler interface: stream %d already closed", streamID))
        }

        n := ws.nodes[streamID]
        n.state = priorityNodeClosed
        n.addBytes(-n.bytes)

        q := n.q
        ws.queuePool.put(&q)
        n.q.s = nil
        if ws.maxClosedNodesInTree > 0 {
                ws.addClosedOrIdleNode(&ws.closedNodes, ws.maxClosedNodesInTree, n)
        } else {
                ws.removeNode(n)
        }
}

func (ws *priorityWriteScheduler) AdjustStream(streamID uint32, priority PriorityParam) {
        if streamID == 0 {
                panic("adjustPriority on root")
        }

        // If streamID does not exist, there are two cases:
        // - A closed stream that has been removed (this will have ID <= maxID)
        // - An idle stream that is being used for "grouping" (this will have ID > maxID)
        n := ws.nodes[streamID]
        if n == nil {
                if streamID <= ws.maxID || ws.maxIdleNodesInTree == 0 {
                        return
                }
                ws.maxID = streamID
                n = &priorityNode{
                        q:      *ws.queuePool.get(),
                        id:     streamID,
                        weight: priorityDefaultWeight,
                        state:  priorityNodeIdle,
                }
                n.setParent(&ws.root)
                ws.nodes[streamID] = n
                ws.addClosedOrIdleNode(&ws.idleNodes, ws.maxIdleNodesInTree, n)
        }

        // Section 5.3.1: A dependency on a stream that is not currently in the tree
        // results in that stream being given a default priority (Section 5.3.5).
        parent := ws.nodes[priority.StreamDep]
        if parent == nil {
                n.setParent(&ws.root)
                n.weight = priorityDefaultWeight
                return
        }

        // Ignore if the client tries to make a node its own parent.
        if n == parent {
                return
        }

        // Section 5.3.3:
        //   "If a stream is made dependent on one of its own dependencies, the
        //   formerly dependent stream is first moved to be dependent on the
        //   reprioritized stream's previous parent. The moved dependency retains
        //   its weight."
        //
        // That is: if parent depends on n, move parent to depend on n.parent.
        for x := parent.parent; x != nil; x = x.parent {
                if x == n {
                        parent.setParent(n.parent)
                        break
                }
        }

        // Section 5.3.3: The exclusive flag causes the stream to become the sole
        // dependency of its parent stream, causing other dependencies to become
        // dependent on the exclusive stream.
        if priority.Exclusive {
                k := parent.kids
                for k != nil {
                        next := k.next
                        if k != n {
                                k.setParent(n)
                        }
                        k = next
                }
        }

        n.setParent(parent)
        n.weight = priority.Weight
}

func (ws *priorityWriteScheduler) Push(wr FrameWriteRequest) {
        var n *priorityNode
        if id := wr.StreamID(); id == 0 {
                n = &ws.root
        } else {
                n = ws.nodes[id]
                if n == nil {
                        // id is an idle or closed stream. wr should not be a HEADERS or
                        // DATA frame. However, wr can be a RST_STREAM. In this case, we
                        // push wr onto the root, rather than creating a new priorityNode,
                        // since RST_STREAM is tiny and the stream's priority is unknown
                        // anyway. See issue #17919.
                        if wr.DataSize() > 0 {
                                panic("add DATA on non-open stream")
                        }
                        n = &ws.root
                }
        }
        n.q.push(wr)
}

func (ws *priorityWriteScheduler) Pop() (wr FrameWriteRequest, ok bool) {
        ws.root.walkReadyInOrder(false, &ws.tmp, func(n *priorityNode, openParent bool) bool {
                limit := int32(math.MaxInt32)
                if openParent {
                        limit = ws.writeThrottleLimit
                }
                wr, ok = n.q.consume(limit)
                if !ok {
                        return false
                }
                n.addBytes(int64(wr.DataSize()))
                // If B depends on A and B continuously has data available but A
                // does not, gradually increase the throttling limit to allow B to
                // steal more and more bandwidth from A.
                if openParent {
                        ws.writeThrottleLimit += 1024
                        if ws.writeThrottleLimit < 0 {
                                ws.writeThrottleLimit = math.MaxInt32
                        }
                } else if ws.enableWriteThrottle {
                        ws.writeThrottleLimit = 1024
                }
                return true
        })
        return wr, ok
}

func (ws *priorityWriteScheduler) addClosedOrIdleNode(list *[]*priorityNode, maxSize int, n *priorityNode) {
        if maxSize == 0 {
                return
        }
        if len(*list) == maxSize {
                // Remove the oldest node, then shift left.
                ws.removeNode((*list)[0])
                x := (*list)[1:]
                copy(*list, x)
                *list = (*list)[:len(x)]
        }
        *list = append(*list, n)
}

func (ws *priorityWriteScheduler) removeNode(n *priorityNode) {
        for k := n.kids; k != nil; k = k.next {
                k.setParent(n.parent)
        }
        n.setParent(nil)
        delete(ws.nodes, n.id)
}