Documentation
¶
Index ¶
Constants ¶
const ( // DefaultGuestBufferSize bounds how much output may sit in pending before a // guest is dropped; the worst case for undelivered output is this plus one // in-flight maxDrainChunk. A guest merely slower than the host's terminal, // rather than stalled, is throttled only by the host's own ingest rate, so // backlog accrues at the difference between that rate and the guest's // drain rate, and a sustained mismatch exhausts any finite cap: this // number sets how long a slow guest survives, not whether. Total slack is // roughly 5 MiB, the cap plus a 2 MiB SSH window on each of the two legs // between host and guest. The buffer exists to decouple the fan-out from a // blocking write, not to store the session. DefaultGuestBufferSize = 1 << 20 // 1 MiB )
Variables ¶
var ( // ErrOverflow is returned once a sink has passed its cap. A guest that // cannot receive output is not attached in any useful sense, so the sink // fails permanently rather than dropping bytes out of the middle of a // terminal stream, which the guest could not detect. ErrOverflow = errors.New("asyncwriter: buffer overflow") // ErrWriterClosed is returned by Write after Close. ErrWriterClosed = errors.New("asyncwriter: closed") )
var ErrClosed = errors.New("multiwriter: closed to new writers")
ErrClosed is returned by Append once Shutdown has run. A guest that reaches the door as the session is ending is refused rather than attached to a fan-out nothing will flush again.
Functions ¶
Types ¶
type AsyncWriter ¶ added in v0.27.0
type AsyncWriter struct {
// contains filtered or unexported fields
}
AsyncWriter delivers to one writer from one goroutine, so writing to it never blocks on that writer's I/O.
It exists because the host fans its pty output out to every guest serially: a guest that stops reading its SSH channel blocks in Write once the channel window fills, and holds up every writer behind it, including the host's own terminal. Handing off into a bounded buffer means the slowest viewer no longer sets the pace. See owenthereal/upterm#524.
The buffer is bounded in bytes rather than in writes, because pty writes run from one byte to io.Copy's 32 KiB and a count of writes is therefore not a bound on memory at all.
func NewAsyncWriter ¶ added in v0.27.0
func NewAsyncWriter(w io.Writer, max int, onDrop func(error)) *AsyncWriter
NewAsyncWriter starts delivery to w immediately. max bounds the payload held undelivered; onDrop, which may be nil, is called once if the sink dies, always on its own goroutine so the caller of Write is never held up by it.
The caller owns w. Close stops the goroutine but does not close w.
func (*AsyncWriter) Close ¶ added in v0.27.0
func (a *AsyncWriter) Close() error
Close stops accepting writes and signals the drain. It deliberately does not wait for the goroutine to exit: in the case this type exists for, that goroutine is blocked in a write that only session teardown will release, and waiting here would deadlock exactly then. The goroutine exits when its in-flight write returns.
Close always returns nil and is safe to call repeatedly. Pending bytes are discarded; MultiWriter.Shutdown is the path that delivers a tail.
func (*AsyncWriter) Flush ¶ added in v0.27.0
func (a *AsyncWriter) Flush(ctx context.Context) error
Flush waits until everything written so far has been delivered, or until ctx is done.
It waits on the idle signal the drain publishes once it has emptied the buffer and its last write has returned, rather than polling, and it honours ctx so that one stuck guest cannot hold up the host's exit. A dead or closed sink flushes to nil: there is nothing left to deliver and a guest that is already gone is not a shutdown error.
func (*AsyncWriter) Write ¶ added in v0.27.0
func (a *AsyncWriter) Write(p []byte) (int, error)
Write copies p into the pending buffer and returns. It never performs I/O and never blocks on the underlying writer.
It reports overflow as an error so that MultiWriter, which already drops a writer that fails, removes this one without needing to know anything about buffering.
type Flusher ¶ added in v0.27.0
Flusher is implemented by attached writers that deliver asynchronously and so can still be holding output when the producer stops.
type MultiWriter ¶
type MultiWriter struct {
// contains filtered or unexported fields
}
MultiWriter is a concurrent safe writer that allows appending/removing writers. Newly appended writers get the last write to preserve last output.
Attached writers are tracked by identity, so they must be comparable; see checkRemovable, which is how Append enforces it. NewMultiWriter does not, because it cannot report the error, so prefer Append.
func NewMultiWriter ¶
func NewMultiWriter(bufferSize int, writers ...io.Writer) *MultiWriter
func (*MultiWriter) Append ¶
func (t *MultiWriter) Append(writers ...io.Writer) error
Append attaches writers, handing each the replay buffer first so it starts from recent output rather than mid-screen.
Both steps happen under writeMu, the fan-out lock, so attaching is atomic with respect to a Write: a joining writer gets the replay and every write after it, never a write that is also in its replay and never a gap between them. Holding the fan-out lock here is only safe because attached guest writers no longer block on I/O; under the old design it would have reintroduced the deadlock #523 removed.
func (*MultiWriter) Remove ¶
func (t *MultiWriter) Remove(writers ...io.Writer)
func (*MultiWriter) Shutdown ¶ added in v0.27.0
func (t *MultiWriter) Shutdown(ctx context.Context) error
Shutdown closes the fan-out to new writers and then waits for everything already accepted to be delivered.
The two halves are inseparable. Flushing a snapshot alone would leave a window: the SSH server keeps serving while the flush runs, so a guest attaching after the snapshot has its replay queued into a sink this call will never flush, and teardown closes it before delivery. Quiescing under writeMu, the same lock Append takes, leaves no such window — an attach is either inside the snapshot or refused.
Members that do not buffer are skipped: a synchronous writer is delivered by definition. A sink that has already failed flushes to nil, because a guest that is already gone is not a shutdown error.
func (*MultiWriter) Write ¶
func (t *MultiWriter) Write(p []byte) (int, error)
Write fans p out to every attached writer. It always reports success.
Two things here are deliberate, and both were bugs.
The membership lock is released before any writer is written to. Holding one lock for both jobs made Append and Remove wait on whatever the slowest attached writer was doing, and a guest that stops reading its SSH channel blocks in Write indefinitely once the channel window fills. That wedged the host: HandleSession removes its writer on the way out, so Remove blocked, HandleSession never returned, and the client-left event it emits on the way out was never sent. Writes are still serialized among themselves, by writeMu, because concurrent producers must not interleave in a writer or race one that is not concurrency safe.
A failing writer is dropped rather than reported. This is a broadcast to whoever is attached, and the producer is the host's pty: returning an error aborted the io.Copy feeding it, which ended the command and tore down the whole session. One guest losing its connection at the wrong moment must not take the session with it. Errors used to abandon the rest of the slice too, so a broken guest silenced everyone attached after it.
A writer removed between the snapshot and the write still receives this one write. That is harmless: it is a session on its way out.
Writers that buffer are what keep this serial loop honest: each attached guest is an AsyncWriter, so its Write is a copy and a signal rather than SSH I/O, and a guest that cannot keep up overflows and is dropped instead of pacing everyone else. The host's own stdout is attached unwrapped and so is written inline here, which is the one place back-pressure belongs: the pty should not run ahead of the terminal that owns it.
type TerminalQueryFilter ¶ added in v0.21.0
type TerminalQueryFilter struct {
// contains filtered or unexported fields
}
TerminalQueryFilter wraps an io.Writer and filters out terminal query sequences from the output. This prevents queries sent by the host's shell from reaching connected clients, whose terminals would otherwise respond and pollute the PTY input.
Filtered queries include:
- OSC 10/11/12 queries (foreground/background/cursor color): ESC ] N ; ? BEL/ST
- CSI 5 n (device status request)
- CSI 6 n (cursor position request)
- CSI 14 t / CSI 18 t (text area size in pixels / characters)
- CSI c / CSI 0 c (primary device attributes)
- CSI > c / CSI > 0 c (secondary device attributes)
- CSI = c / CSI = 0 c (tertiary device attributes)
- CSI > q / CSI > 0 q (terminal name and version)
func NewTerminalQueryFilter ¶ added in v0.21.0
func NewTerminalQueryFilter(w io.Writer) *TerminalQueryFilter
NewTerminalQueryFilter creates a filter that removes terminal query sequences from output before writing to the underlying writer.
func (*TerminalQueryFilter) Write ¶ added in v0.21.0
func (f *TerminalQueryFilter) Write(p []byte) (int, error)
Write filters terminal query sequences from p and writes the result to the underlying writer. Returns len(p) on success to indicate all input bytes were processed. On error, returns 0 because the filtered output is written atomically (all or nothing) and input bytes don't map 1:1 to output bytes due to filtering.
Source Files