agent: wait serial channel to be ready before reading

Otherwise grpc.Serv() will return directly with io.EOF error
if there is no serial connection peer point yet, because there is nothing
to read yet. We haven't seen it yet simply because kata-proxy is always
started before qemu boots up.

This also allows kata runtime to reconnect after dropping the
serial port connection.

Fixes: #213

Signed-off-by: Peng Tao <[email protected]>

agent.go

@@ -464,6 +464,12 @@ func (s *sandbox) startGRPC() {
 				return
 			}
 
+			err = s.channel.wait()
+			if err != nil {
+				agentLog.WithError(err).Warn("Failed to wait agent grpc channel ready")
+				return
+			}
+
 			var l net.Listener
 			l, err = s.channel.listen()
 			if err != nil {

channel.go

@@ -7,11 +7,13 @@
 package main
 
 import (
+	"fmt"
 	"io/ioutil"
 	"net"
 	"os"
 	"path/filepath"
 	"strings"
+	"syscall"
 
 	"github.com/hashicorp/yamux"
 	"github.com/mdlayher/vsock"
@@ -22,6 +24,7 @@ import (
 
 type channel interface {
 	setup() error
+	wait() error
 	listen() (net.Listener, error)
 	teardown() error
 }
@@ -57,6 +60,10 @@ func (c *vSockChannel) setup() error {
 	return nil
 }
 
+func (c *vSockChannel) wait() error {
+	return nil
+}
+
 func (c *vSockChannel) listen() (net.Listener, error) {
 	l, err := vsock.Listen(vSockPort)
 	if err != nil {
@@ -86,6 +93,57 @@ func (c *serialChannel) setup() error {
 	return nil
 }
 
+func (c *serialChannel) wait() error {
+	var event syscall.EpollEvent
+	var events [1]syscall.EpollEvent
+
+	fd := c.serialConn.Fd()
+	if fd <= 0 {
+		return fmt.Errorf("serial port IO closed")
+	}
+
+	epfd, err := syscall.EpollCreate1(syscall.EPOLL_CLOEXEC)
+	if err != nil {
+		return err
+	}
+	defer syscall.Close(epfd)
+
+	// EPOLLOUT: Writable when there is a connection
+	// EPOLLET: Edge trigger as EPOLLHUP is always on when there is no connection
+	// 0xffffffff: EPOLLET is negative and cannot fit in uint32 in golang
+	event.Events = syscall.EPOLLOUT | syscall.EPOLLET&0xffffffff
+	event.Fd = int32(fd)
+	if err = syscall.EpollCtl(epfd, syscall.EPOLL_CTL_ADD, int(fd), &event); err != nil {
+		return err
+	}
+	defer syscall.EpollCtl(epfd, syscall.EPOLL_CTL_DEL, int(fd), nil)
+
+	for {
+		nev, err := syscall.EpollWait(epfd, events[:], -1)
+		if err != nil {
+			return err
+		}
+
+		for i := 0; i < nev; i++ {
+			ev := events[i]
+			if ev.Fd == int32(fd) {
+				agentLog.WithField("events", ev.Events).Debug("New serial channel event")
+				if ev.Events&syscall.EPOLLOUT != 0 {
+					return nil
+				}
+				if ev.Events&syscall.EPOLLERR != 0 {
+					return fmt.Errorf("serial port IO failure")
+				}
+				if ev.Events&syscall.EPOLLHUP != 0 {
+					continue
+				}
+			}
+		}
+	}
+
+	// Never reach here
+}
+
 func (c *serialChannel) listen() (net.Listener, error) {
 	// Initialize Yamux server.
 	session, err := yamux.Server(c.serialConn, nil)