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 )
const DefaultReplayBytes = 256 << 10
DefaultReplayBytes bounds the replay ring handed to a joining writer.
It must stay well under DefaultGuestBufferSize: Append replays into a joining writer before attaching it, so a ring at or above a guest's sink size would overflow that guest with its own replay and drop it at the door.
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) Closed ¶ added in v0.30.0
func (a *AsyncWriter) Closed() bool
Closed reports whether Close has been called.
Like Err, it is the other half of what Flush does not say: Flush returns nil for a closed sink too, and a caller that needs to know whether what it wrote was delivered rather than discarded has to ask.
func (*AsyncWriter) Drained ¶ added in v0.30.0
func (a *AsyncWriter) Drained() (<-chan struct{}, bool)
Drained returns a channel closed the next time delivery catches up: the buffer empties and the last write returns. It is Flush for a caller that wants to be told rather than to wait — one holding a lock, or one with nothing to do until the news arrives.
The second return is false for a writer that has already failed or been closed, and there is nothing else it could be: the signal those publish is the one the writer had at the time, and the channel that replaces it is never closed by anything. Answered under the lock the two of them take, so a caller that is told true holds a channel that failing and closing will still close.
func (*AsyncWriter) Err ¶ added in v0.30.0
func (a *AsyncWriter) Err() error
Err reports the error that ended delivery, if any.
Flush cannot answer this: it reports nil for a sink that has already failed, because a guest that is gone is not a shutdown error. A caller that needs to know whether everything written was actually delivered — rather than merely that nothing more is coming — has to ask here as well.
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 ModeTracker ¶ added in v0.30.0
type ModeTracker struct {
// contains filtered or unexported fields
}
ModeTracker watches terminal output and remembers the modes set on it, so a reader joining later can be put into the same modes before it is handed the replay ring.
It is deliberately not a terminal emulator: it reconstructs no screen and no cursor position. Anything it does not recognise passes through unrecorded.
Not safe for concurrent use. MultiWriter owns one and only touches it under writeMu.
func NewModeTracker ¶ added in v0.30.0
func NewModeTracker() *ModeTracker
func (*ModeTracker) Restore ¶ added in v0.30.0
func (m *ModeTracker) Restore() []byte
Restore returns the bytes that put a terminal this tracker has been watching back where it started. It is Snapshot's opposite: Snapshot is for the terminal joining a session, this is for the terminal leaving one.
A terminal is left in whatever modes the session put it in, and restoring the termios settings says nothing about them: a full-screen program that was on screen when its viewer detached leaves that viewer's shell drawing on the alternate screen, with no cursor, with the mouse reporting clicks as input, and confined to the rows the program had chosen for itself. None of that is the shell's doing and none of it goes away on its own.
Only what was actually changed is undone, so a session that never touched a mode produces nothing at all — the same property Snapshot has, and the reason this is derived from the stream rather than being a fixed list of resets sent on the way out. Attributes are not tracked and so are not reset here.
"Where it started" means the terminal's power-on defaults, not the state the caller's own terminal was in before the attachment. That state is not knowable without asking the terminal for it — a DECRQM round-trip per mode, before attaching, racing the user's keystrokes on the same stdin, and unanswered by terminals that do not implement it. tmux, which has the same problem on detach, resets to defaults too rather than asking.
It costs less than it sounds like. At a shell prompt — which is what launches a CLI — every mode in restorable is already at its default except the ones the shell re-asserts on every prompt: measured here, zsh holds DECCKM and bracketed paste and emits both again at the next prompt, bash holds bracketed paste and does the same. So for the modes a caller actually holds, a reset is undone within one keystroke, and for the rest the default is what the caller had. What is left is a caller that is not a shell and keeps the mouse or the alternate screen while running a child, which is not how a program that shells out behaves.
The order is the order a terminal has to receive it in: leave the alternate screen first, through the mode that entered it, since everything after it applies to the screen the terminal is going back to.
func (*ModeTracker) Snapshot ¶ added in v0.30.0
func (m *ModeTracker) Snapshot() []byte
Snapshot returns the bytes that put a fresh terminal into the recorded modes. State already at the terminal's default is left out, so a session that never changed anything replays nothing. Its length is bounded by len(restorable) plus the screen switch, the three verbatim sequences and one partial sequence. Only the partial is sized by the stream rather than by this file, and its own bound is maxSequenceBytes for a mode sequence or maxStringBytes for a string: the ceiling is a few kilobytes, not the tens of bytes everything else comes to.
The order is the order the stream would have had to use to reach this state: the normal screen's margins, then the modes, then the switch to the alternate screen, then that screen's own margins, and only while it is the one showing.
The state as of the ring's first byte includes being partway through a sequence, so the partial goes last, after the charset. The ring's first bytes then complete it on the joiner's terminal instead of printing as text -- which is what they did when the trim boundary fell inside a sequence and the snapshot said nothing about it.
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(replayBytes int, writers ...io.Writer) *MultiWriter
NewMultiWriter returns a fan-out whose replay ring holds the most recent replayBytes bytes of output.
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 only when it is a terminal, which is the one place back-pressure belongs: the pty should not run ahead of the screen that owns it. A non-terminal stdout is wrapped, because a pipe nobody drains would otherwise block here with writeMu held and wedge the whole session.
type StallWriter ¶ added in v0.30.0
type StallWriter struct {
// contains filtered or unexported fields
}
StallWriter writes synchronously to w in chunks, stamping progress after each one, and calls onStall once if a write has been outstanding with no chunk completed for timeout.
It exists for the primary client's pacing. The primary's writer is attached to the fan-out synchronously, so the command cannot outrun the terminal, and MultiWriter.Write holds writeMu across it: a primary that stops reading its SSH channel therefore blocks every other writer and every new attach. The bound has to measure progress rather than duration — "a write took 5 s" and "no chunk has completed for 5 s" treat a slow-but-working terminal differently, and only the second is worth acting on — and it has to be armed only while a write is outstanding, or a healthy shell at a prompt would be judged stalled for producing nothing.
A single blocking Write is unobservable from outside, which is why the writer chunks: the chunk is the unit of progress. The progress clock starts with each write, so the first chunk after a quiet period is not judged against a stale stamp.
onStall runs on the watchdog's goroutine and must not take writeMu: the lock is held by the write it is there to release. Closing the client's SSH connection is what it is for; see host/internal.
func NewStallWriter ¶ added in v0.30.0
NewStallWriter starts the watchdog. chunk bounds the bytes handed to w per call; onStall may be nil.
func (*StallWriter) Close ¶ added in v0.30.0
func (s *StallWriter) Close() error
Close stops the watchdog. It does not close w and does not release a write already blocked in it; only onStall's action can do that.
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) Pending ¶ added in v0.30.0
func (f *TerminalQueryFilter) Pending() []byte
Pending returns a copy of the bytes of an escape sequence the filter is still holding because it has not yet seen the sequence's end. They are not in the underlying writer yet. The result never exceeds maxPendingBytes.
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