Add retry and non-blocking send for exit events

Signed-off-by: Michael Crosby <crosbymichael@gmail.com>

cmd/containerd-shim/shim_darwin.go

@@ -22,7 +22,7 @@ import (
 	"os"
 	"os/signal"
 
-	"github.com/containerd/containerd/runtime/v1/shim"
+	"github.com/containerd/containerd/sys/reaper"
 	runc "github.com/containerd/go-runc"
 	"github.com/containerd/ttrpc"
 )
@@ -34,7 +34,7 @@ func setupSignals() (chan os.Signal, error) {
 	signal.Notify(signals)
 	// make sure runc is setup to use the monitor
 	// for waiting on processes
-	runc.Monitor = shim.Default
+	runc.Monitor = reaper.Default
 	return signals, nil
 }
 

sys/reaper/reaper_unix.go

@@ -36,19 +36,25 @@ const bufferSize = 2048
 // Reap should be called when the process receives an SIGCHLD.  Reap will reap
 // all exited processes and close their wait channels
 func Reap() error {
-	now := time.Now()
+	var (
+		now     = time.Now()
+		current []chan runc.Exit
+	)
 	exits, err := sys.Reap(false)
+
 	Default.Lock()
 	for c := range Default.subscribers {
-		for _, e := range exits {
-			c <- runc.Exit{
-				Timestamp: now,
-				Pid:       e.Pid,
-				Status:    e.Status,
-			}
-		}
+		current = append(current, c)
 	}
 	Default.Unlock()
+
+	for _, e := range exits {
+		go notify(runc.Exit{
+			Timestamp: now,
+			Pid:       e.Pid,
+			Status:    e.Status,
+		}, current)
+	}
 	return err
 }
 
@@ -107,3 +113,28 @@ func (m *Monitor) Unsubscribe(c chan runc.Exit) {
 	close(c)
 	m.Unlock()
 }
+
+func notify(e runc.Exit, subscribers []chan runc.Exit) {
+	const timeout = 10 * time.Millisecond
+	timer := time.NewTimer(timeout)
+	timer.Stop()
+
+	for i := 0; i < 50; i++ {
+		var failed []chan runc.Exit
+		for _, s := range subscribers {
+			timer.Reset(timeout)
+
+			select {
+			case s <- e:
+			case <-timer.C:
+				failed = append(failed, s)
+			}
+			timer.Stop()
+		}
+		// all subscribers received the message
+		if len(failed) == 0 {
+			return
+		}
+		subscribers = failed
+	}
+}