pipe.go RawHistoryFolder

1package brokeripc
2
3import (
4	"io"
5	"sync"
6)
7
8// streamPipe is the private ordered, bounded byte channel between one broker
9// wire session and the sessionwire framing of one logical stream.
10//
11// The wire reader never blocks on a slow consumer: a chunk the peer sent is
12// appended to a bounded queue and refused with ErrStreamBackpressure once the
13// bound is reached, so only that stream is settled and every sibling and the
14// physical connection stay untouched. The consumer never blocks the writer:
15// Read serves already-queued chunks or parks on the wake channel, and Close
16// releases both ends promptly. Write hands each byte slice to the injected
17// outbound hook, which owns chunking it under the negotiated stream chunk
18// ceiling and enqueueing the frames on the connection.
19type streamPipe struct {
20	mu       sync.Mutex
21	queue    [][]byte
22	queued   int
23	closed   bool
24	err      error
25	maxChunk int
26	maxBytes int
27
28	wake     chan struct{}
29	closedCh chan struct{}
30
31	outbound func([]byte) error
32}
33
34// newStreamPipe builds one bounded stream carriage. outbound must not be nil.
35func newStreamPipe(maxChunks int, maxBytes uint64, outbound func([]byte) error) *streamPipe {
36	if maxChunks <= 0 {
37		maxChunks = DefaultStreamInboundChunks
38	}
39	if maxBytes == 0 {
40		maxBytes = DefaultStreamInboundBytes
41	}
42	return &streamPipe{
43		maxChunk: maxChunks,
44		maxBytes: int(maxBytes),
45		wake:     make(chan struct{}, 1),
46		closedCh: make(chan struct{}),
47		outbound: outbound,
48	}
49}
50
51// Read serves the next queued chunk, blocking until one arrives or the pipe
52// closes. A partial copy retains the unread remainder for the next call.
53func (p *streamPipe) Read(b []byte) (int, error) {
54	if len(b) == 0 {
55		return 0, nil
56	}
57	for {
58		p.mu.Lock()
59		if len(p.queue) > 0 {
60			chunk := p.queue[0]
61			n := copy(b, chunk)
62			if n == len(chunk) {
63				p.queue[0] = nil
64				p.queue = p.queue[1:]
65			} else {
66				p.queue[0] = chunk[n:]
67			}
68			p.queued -= n
69			p.mu.Unlock()
70			return n, nil
71		}
72		if p.closed {
73			err := p.err
74			p.mu.Unlock()
75			if err != nil {
76				return 0, err
77			}
78			return 0, io.EOF
79		}
80		p.mu.Unlock()
81		select {
82		case <-p.wake:
83		case <-p.closedCh:
84		}
85	}
86}
87
88// Write hands one byte slice to the outbound hook. It returns ErrClosed once the
89// pipe reached its terminal outcome.
90func (p *streamPipe) Write(b []byte) (int, error) {
91	p.mu.Lock()
92	closed := p.closed
93	p.mu.Unlock()
94	if closed {
95		return 0, ErrStreamGone
96	}
97	if err := p.outbound(b); err != nil {
98		return 0, err
99	}
100	return len(b), nil
101}
102
103// deliver accepts one inbound chunk from the connection reader. It returns
104// ErrStreamBackpressure when the bound is reached and ErrStreamGone once the
105// pipe is closed; the caller settles the affected stream in both cases.
106func (p *streamPipe) deliver(chunk []byte) error {
107	if len(chunk) == 0 {
108		return nil
109	}
110	p.mu.Lock()
111	if p.closed {
112		p.mu.Unlock()
113		return ErrStreamGone
114	}
115	if len(p.queue) >= p.maxChunk || p.queued+len(chunk) > p.maxBytes {
116		p.mu.Unlock()
117		return ErrStreamBackpressure
118	}
119	owned := append([]byte(nil), chunk...)
120	p.queue = append(p.queue, owned)
121	p.queued += len(owned)
122	p.mu.Unlock()
123	select {
124	case p.wake <- struct{}{}:
125	default:
126	}
127	return nil
128}
129
130// closeWith settles the pipe exactly once with cause. A nil cause is an orderly
131// close; every blocked Read observes it promptly. On a nil cause, any chunks
132// already queued are kept: Read serves them first and only returns io.EOF once
133// the queue is drained, so a reply that arrived just before an orderly peer
134// close is never discarded underneath its reader. An error cause is a failure:
135// the queue is discarded, since its content can no longer be trusted to reach
136// a consumer that is being torn down for cause.
137func (p *streamPipe) closeWith(cause error) {
138	p.mu.Lock()
139	if p.closed {
140		p.mu.Unlock()
141		return
142	}
143	p.closed = true
144	p.err = cause
145	if cause != nil {
146		p.queue = nil
147		p.queued = 0
148	}
149	p.mu.Unlock()
150	close(p.closedCh)
151	select {
152	case p.wake <- struct{}{}:
153	default:
154	}
155}
156
157// Close settles the pipe as an orderly close.
158func (p *streamPipe) Close() error {
159	p.closeWith(nil)
160	return nil
161}