Make windows terminal keep up with real time

lib/client/terminal/terminal_windows.go

@@ -137,7 +137,7 @@ func New(stdin io.Reader, stdout, stderr io.Writer) (*Terminal, error) {
 	return &term, nil
 }
 
-// InitRaw puts the terminal into raw output mode. If `input` set set, it also
+// InitRaw puts the terminal into raw output mode. If `input` is set, it also
 // begins capturing raw input events from the Windows API, asynchronously
 // writing them to a Pipe emulating a traditional Unix stdin.
 // Note that some implementations may replace one or more streams (particularly
@@ -165,36 +165,23 @@ func (t *Terminal) InitRaw(input bool) error {
 		cleanup()
 	}()
 
-	// Convert input events into a usable io.Reader.
-	pipeRead, pipeWrite := io.Pipe()
+	// emit resize events
 	t.closeWait.Add(1)
 	go func() {
 		defer t.closeWait.Done()
 
-		events := tncon.Subscribe()
+		ch := tncon.SubcribeResizeEvents()
 		for {
 			select {
-			case event := <-events:
-				switch e := event.(type) {
-				case tncon.SequenceEvent:
-					if len(e.Sequence) > 0 {
-						_, err := pipeWrite.Write(e.Sequence)
-						if err != nil {
-							log.Errorf("failed to write input sequence: %+v", err)
-							_ = t.closer.Close()
-							return
-						}
-					}
-				case tncon.ResizeEvent:
-					t.writeEvent(ResizeEvent{})
-				}
+			case <-ch:
+				t.writeEvent(ResizeEvent{})
 			case <-t.closer.C:
 				return
 			}
 		}
 	}()
 
-	t.stdin = pipeRead
+	t.stdin = tncon.SequenceReader()
 	return nil
 }
 

lib/client/tncon/buffer.go

@@ -0,0 +1,73 @@
+// Copyright (c) 2020 Leonid Titov. All rights reserved.
+// MIT licence.
+// Version 2020-12-23
+
+package tncon
+
+import (
+	"io"
+)
+
+// bufferedChannelPipe is a synchronous buffered pipe implemented with a channel. This pipe
+// is much more efficient than the standard io.Pipe, and can keep up with real-time
+// shell output, which is needed for the lib/client/tncon implementation.
+//
+// Derived from https://github.com/latitov/milkthisbuffer/blob/main/milkthisbuffer.go
+type bufferedChannelPipe struct {
+	ch chan byte
+}
+
+func newBufferedChannelPipe(len int) *bufferedChannelPipe {
+	return &bufferedChannelPipe{
+		ch: make(chan byte, len),
+	}
+}
+
+// Write will write all of p to the buffer unless the buffer is closed
+func (b *bufferedChannelPipe) Write(p []byte) (n int, err error) {
+	// Catch write to closed buffer with a recover
+	// https://stackoverflow.com/a/34899098/11729048
+	defer func() {
+		if err2 := recover(); err2 != nil {
+			err = io.ErrClosedPipe
+		}
+	}()
+
+	for n = 0; n < len(p); n++ {
+		// blocking behaviour
+		b.ch <- p[n]
+	}
+	return n, nil
+}
+
+// Read will always read at least one byte from the buffer unless the buffer is closed
+func (b *bufferedChannelPipe) Read(p []byte) (n int, err error) {
+	if len(p) == 0 {
+		return 0, nil
+	}
+
+	// blocking behaviour
+	r, ok := <-b.ch
+	if !ok {
+		return 0, io.EOF
+	}
+	p[n] = r
+
+	for n = 1; n < len(p); n++ {
+		select {
+		case r, ok := <-b.ch:
+			if !ok {
+				return n, io.EOF
+			}
+			p[n] = r
+		default:
+			return n, nil
+		}
+	}
+	return n, nil
+}
+
+func (b *bufferedChannelPipe) Close() error {
+	close(b.ch)
+	return nil
+}

lib/client/tncon/buffer_test.go

@@ -0,0 +1,160 @@
+/*
+Copyright 2022 Gravitational, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package tncon
+
+import (
+	"fmt"
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestBufferedChannelPipeClose(t *testing.T) {
+	buffer := newBufferedChannelPipe(0)
+	buffer.Close()
+
+	// Reading from a closed channel should return EOF
+	n, err := buffer.Read(make([]byte, 1))
+	require.Equal(t, 0, n)
+	require.Error(t, err)
+	require.Equal(t, io.EOF, err)
+
+	// Reading from a closed channel should return ErrClosedPipe
+	n, err = buffer.Write(make([]byte, 1))
+	require.Equal(t, 0, n)
+	require.Error(t, err)
+	require.Equal(t, io.ErrClosedPipe, err)
+}
+
+func TestBufferedChannelPipeWrite(t *testing.T) {
+	// With a sufficient buffer, write should successfully
+	// write to the channel without blocking
+	for _, tc := range []struct {
+		buffer int
+		len    int
+	}{
+		{
+			buffer: 0,
+			len:    0,
+		}, {
+			buffer: 0,
+			len:    10,
+		}, {
+			buffer: 10,
+			len:    10,
+		}, {
+			buffer: 10,
+			len:    100,
+		}, {
+			buffer: 100,
+			len:    100,
+		},
+	} {
+		t.Run(fmt.Sprintf("buffer=%v, len=%v", tc.buffer, tc.len), func(t *testing.T) {
+			buffer := newBufferedChannelPipe(tc.buffer)
+			t.Cleanup(func() { buffer.Close() })
+
+			// drain channel
+			rc := make(chan []byte)
+			go func() {
+				read := make([]byte, tc.len)
+				for i := range read {
+					read[i] = <-buffer.ch
+				}
+				rc <- read
+			}()
+
+			p := make([]byte, tc.len)
+			for n := 0; n < tc.len; n++ {
+				p[n] = byte(n)
+			}
+
+			n, err := buffer.Write(p)
+			require.NoError(t, err)
+			require.Equal(t, tc.len, n)
+			require.Equal(t, p, <-rc)
+		})
+	}
+}
+
+func TestBufferedChannelPipeRead(t *testing.T) {
+	for _, tc := range []struct {
+		desc     string
+		buffer   int
+		writeLen int
+		readLen  int
+		expectN  int
+	}{
+		{
+			desc:     "empty read",
+			buffer:   0,
+			writeLen: 0,
+			readLen:  0,
+			expectN:  0,
+		}, {
+			desc:     "one byte read",
+			buffer:   0,
+			writeLen: 1,
+			readLen:  1,
+			expectN:  1,
+		}, {
+			desc:     "read with sufficient buffer",
+			buffer:   10,
+			writeLen: 10,
+			readLen:  10,
+			expectN:  10,
+		}, {
+			desc:     "large read with sufficient buffer",
+			buffer:   255,
+			writeLen: 255,
+			readLen:  255,
+			expectN:  255,
+		}, {
+			desc:     "read with smaller write",
+			buffer:   100,
+			writeLen: 10,
+			readLen:  100,
+			expectN:  10,
+		}, {
+			desc:     "read with larger write",
+			buffer:   100,
+			writeLen: 100,
+			readLen:  10,
+			expectN:  10,
+		},
+	} {
+		t.Run(tc.desc, func(t *testing.T) {
+			buffer := newBufferedChannelPipe(tc.buffer)
+			t.Cleanup(func() { buffer.Close() })
+
+			write := make([]byte, tc.writeLen)
+			for i := 0; i < tc.writeLen; i++ {
+				write[i] = byte(i)
+			}
+
+			// fill channel
+			go buffer.Write(write)
+
+			p := make([]byte, tc.readLen)
+			n, err := buffer.Read(p)
+			require.NoError(t, err)
+			require.Equal(t, tc.expectN, n)
+			require.Equal(t, write[:n], p[:n])
+		})
+	}
+}

lib/client/tncon/tncon.c

@@ -54,7 +54,7 @@ WriteToBuffer(char* source, size_t len)
 {
 	// NOTE: Modified to emit an event to the Go lib rather than mutate a
 	// global buffer.
-	writeSequenceEvent(source, len);
+	writeSequence(source, len);
 
 	return len;
 }
@@ -99,7 +99,7 @@ ReadConsoleForTermEmul(HANDLE hInput, HANDLE hQuitEvent)
 
 			// NOTE: modified here to emit events directly
 			case WINDOW_BUFFER_SIZE_EVENT:
-				writeResizeEvent(inputRecord.Event.WindowBufferSizeEvent.dwSize);
+				notifyResizeEvent();
 				break;
 			case FOCUS_EVENT:
 				break;