var updateHeaderTblSize … type itemNode … type itemList … func (il *itemList) enqueue(i any) { … } // peek returns the first item in the list without removing it from the // list. func (il *itemList) peek() any { … } func (il *itemList) dequeue() any { … } func (il *itemList) dequeueAll() *itemNode { … } func (il *itemList) isEmpty() bool { … } const maxQueuedTransportResponseFrames … type cbItem … type registerStream … func (*registerStream) isTransportResponseFrame() bool { … } type headerFrame … func (h *headerFrame) isTransportResponseFrame() bool { … } type cleanupStream … func (c *cleanupStream) isTransportResponseFrame() bool { … } type earlyAbortStream … func (*earlyAbortStream) isTransportResponseFrame() bool { … } type dataFrame … func (*dataFrame) isTransportResponseFrame() bool { … } type incomingWindowUpdate … func (*incomingWindowUpdate) isTransportResponseFrame() bool { … } type outgoingWindowUpdate … func (*outgoingWindowUpdate) isTransportResponseFrame() bool { … } type incomingSettings … func (*incomingSettings) isTransportResponseFrame() bool { … } type outgoingSettings … func (*outgoingSettings) isTransportResponseFrame() bool { … } type incomingGoAway … func (*incomingGoAway) isTransportResponseFrame() bool { … } type goAway … func (*goAway) isTransportResponseFrame() bool { … } type ping … func (*ping) isTransportResponseFrame() bool { … } type outFlowControlSizeRequest … func (*outFlowControlSizeRequest) isTransportResponseFrame() bool { … } type closeConnection … func (closeConnection) isTransportResponseFrame() bool { … } type outStreamState … const active … const empty … const waitingOnStreamQuota … type outStream … func (s *outStream) deleteSelf() { … } type outStreamList … func newOutStreamList() *outStreamList { … } func (l *outStreamList) enqueue(s *outStream) { … } // remove from the beginning of the list. func (l *outStreamList) dequeue() *outStream { … } type controlBuffer … func newControlBuffer(done <-chan struct{ … } // throttle blocks if there are too many incomingSettings/cleanupStreams in the // controlbuf. func (c *controlBuffer) throttle() { … } func (c *controlBuffer) put(it cbItem) error { … } func (c *controlBuffer) executeAndPut(f func() bool, it cbItem) (bool, error) { … } // Note argument f should never be nil. func (c *controlBuffer) execute(f func(it any) bool, it any) (bool, error) { … } func (c *controlBuffer) get(block bool) (any, error) { … } func (c *controlBuffer) finish() { … } type side … const clientSide … const serverSide … type loopyWriter … func newLoopyWriter(s side, fr *framer, cbuf *controlBuffer, bdpEst *bdpEstimator, conn net.Conn, logger *grpclog.PrefixLogger, goAwayHandler func(*goAway) (bool, error)) *loopyWriter { … } const minBatchSize … // run should be run in a separate goroutine. // It reads control frames from controlBuf and processes them by: // 1. Updating loopy's internal state, or/and // 2. Writing out HTTP2 frames on the wire. // // Loopy keeps all active streams with data to send in a linked-list. // All streams in the activeStreams linked-list must have both: // 1. Data to send, and // 2. Stream level flow control quota available. // // In each iteration of run loop, other than processing the incoming control // frame, loopy calls processData, which processes one node from the // activeStreams linked-list. This results in writing of HTTP2 frames into an // underlying write buffer. When there's no more control frames to read from // controlBuf, loopy flushes the write buffer. As an optimization, to increase // the batch size for each flush, loopy yields the processor, once if the batch // size is too low to give stream goroutines a chance to fill it up. // // Upon exiting, if the error causing the exit is not an I/O error, run() // flushes the underlying connection. The connection is always left open to // allow different closing behavior on the client and server. func (l *loopyWriter) run() (err error) { … } func (l *loopyWriter) outgoingWindowUpdateHandler(w *outgoingWindowUpdate) error { … } func (l *loopyWriter) incomingWindowUpdateHandler(w *incomingWindowUpdate) { … } func (l *loopyWriter) outgoingSettingsHandler(s *outgoingSettings) error { … } func (l *loopyWriter) incomingSettingsHandler(s *incomingSettings) error { … } func (l *loopyWriter) registerStreamHandler(h *registerStream) { … } func (l *loopyWriter) headerHandler(h *headerFrame) error { … } func (l *loopyWriter) originateStream(str *outStream, hdr *headerFrame) error { … } func (l *loopyWriter) writeHeader(streamID uint32, endStream bool, hf []hpack.HeaderField, onWrite func()) error { … } func (l *loopyWriter) preprocessData(df *dataFrame) { … } func (l *loopyWriter) pingHandler(p *ping) error { … } func (l *loopyWriter) outFlowControlSizeRequestHandler(o *outFlowControlSizeRequest) { … } func (l *loopyWriter) cleanupStreamHandler(c *cleanupStream) error { … } func (l *loopyWriter) earlyAbortStreamHandler(eas *earlyAbortStream) error { … } func (l *loopyWriter) incomingGoAwayHandler(*incomingGoAway) error { … } func (l *loopyWriter) goAwayHandler(g *goAway) error { … } func (l *loopyWriter) handle(i any) error { … } func (l *loopyWriter) applySettings(ss []http2.Setting) { … } // processData removes the first stream from active streams, writes out at most 16KB // of its data and then puts it at the end of activeStreams if there's still more data // to be sent and stream has some stream-level flow control. func (l *loopyWriter) processData() (bool, error) { … } func min(a, b int) int { … }
Codebase Browser
kubernetes