Add context cancel for epoll

crosbymichael · crosbymichael · commit d50e25360c8e · 2018-08-29T10:44:47.000-04:00
Signed-off-by: Michael Crosby &lt;crosbymichael@gmail.com&gt;
diff --git a/runtime/v2/runc/epoll.go b/runtime/v2/runc/epoll.go
@@ -62,15 +62,21 @@ func (e *epoller) Close() error {
 func (e *epoller) run(ctx context.Context) {
 	var events [128]unix.EpollEvent
 	for {
-		n, err := unix.EpollWait(e.fd, events[:], -1)
-		if err != nil {
-			if err == unix.EINTR {
-				continue
+		select {
+		case <-ctx.Done():
+			e.Close()
+			return
+		default:
+			n, err := unix.EpollWait(e.fd, events[:], -1)
+			if err != nil {
+				if err == unix.EINTR {
+					continue
+				}
+				logrus.WithError(err).Error("cgroups: epoll wait")
+			}
+			for i := 0; i < n; i++ {
+				e.process(ctx, uintptr(events[i].Fd))
 			}
-			logrus.WithError(err).Error("cgroups: epoll wait")
-		}
-		for i := 0; i < n; i++ {
-			e.process(ctx, uintptr(events[i].Fd))
 		}
 	}
 }
diff --git a/runtime/v2/runc/service.go b/runtime/v2/runc/service.go
@@ -68,6 +68,7 @@ func New(ctx context.Context, id string, publisher events.Publisher) (shim.Shim,
 	if err != nil {
 		return nil, err
 	}
+	ctx, cancel := context.WithCancel(ctx)
 	go ep.run(ctx)
 	s := &service{
 		id:        id,
@@ -76,10 +77,12 @@ func New(ctx context.Context, id string, publisher events.Publisher) (shim.Shim,
 		events:    make(chan interface{}, 128),
 		ec:        shim.Default.Subscribe(),
 		ep:        ep,
+		cancel:    cancel,
 	}
 	go s.processExits()
 	runcC.Monitor = shim.Default
 	if err := s.initPlatform(); err != nil {
+		cancel()
 		return nil, errors.Wrap(err, "failed to initialized platform behavior")
 	}
 	go s.forward(publisher)
@@ -101,6 +104,7 @@ type service struct {
 	id     string
 	bundle string
 	cg     cgroups.Cgroup
+	cancel func()
 }
 
 func newCommand(ctx context.Context, containerdBinary, containerdAddress string) (*exec.Cmd, error) {
@@ -579,6 +583,7 @@ func (s *service) Connect(ctx context.Context, r *taskAPI.ConnectRequest) (*task
 }
 
 func (s *service) Shutdown(ctx context.Context, r *taskAPI.ShutdownRequest) (*ptypes.Empty, error) {
+	s.cancel()
 	os.Exit(0)
 	return empty, nil
 }
@@ -698,7 +703,10 @@ func (s *service) getContainerPids(ctx context.Context, id string) ([]uint32, er
 
 func (s *service) forward(publisher events.Publisher) {
 	for e := range s.events {
-		if err := publisher.Publish(s.context, getTopic(s.context, e), e); err != nil {
+		ctx, cancel := context.WithTimeout(s.context, 5*time.Second)
+		err := publisher.Publish(ctx, getTopic(e), e)
+		cancel()
+		if err != nil {
 			logrus.WithError(err).Error("post event")
 		}
 	}
@@ -722,7 +730,7 @@ func (s *service) setCgroup(cg cgroups.Cgroup) {
 	}
 }
 
-func getTopic(ctx context.Context, e interface{}) string {
+func getTopic(e interface{}) string {
 	switch e.(type) {
 	case *eventstypes.TaskCreate:
 		return runtime.TaskCreateEventTopic

Original file line number	Diff line number	Diff line change
`@@ -68,6 +68,7 @@ func New(ctx context.Context, id string, publisher events.Publisher) (shim.Shim,`
`68`	`68`	`if err != nil {`
`69`	`69`	`return nil, err`
`70`	`70`	`}`
	`71`	`+ ctx, cancel := context.WithCancel(ctx)`
`71`	`72`	`go ep.run(ctx)`
`72`	`73`	`s := &service{`
`73`	`74`	`id: id,`
`@@ -76,10 +77,12 @@ func New(ctx context.Context, id string, publisher events.Publisher) (shim.Shim,`
`76`	`77`	`events: make(chan interface{}, 128),`
`77`	`78`	`ec: shim.Default.Subscribe(),`
`78`	`79`	`ep: ep,`
	`80`	`+ cancel: cancel,`
`79`	`81`	`}`
`80`	`82`	`go s.processExits()`
`81`	`83`	`runcC.Monitor = shim.Default`
`82`	`84`	`if err := s.initPlatform(); err != nil {`
	`85`	`+ cancel()`
`83`	`86`	`return nil, errors.Wrap(err, "failed to initialized platform behavior")`
`84`	`87`	`}`
`85`	`88`	`go s.forward(publisher)`
`@@ -101,6 +104,7 @@ type service struct {`
`101`	`104`	`id string`
`102`	`105`	`bundle string`
`103`	`106`	`cg cgroups.Cgroup`
	`107`	`+ cancel func()`
`104`	`108`	`}`
`105`	`109`
`106`	`110`	`func newCommand(ctx context.Context, containerdBinary, containerdAddress string) (*exec.Cmd, error) {`
`@@ -579,6 +583,7 @@ func (s service) Connect(ctx context.Context, r taskAPI.ConnectRequest) (*task`
`579`	`583`	`}`
`580`	`584`
`581`	`585`	`func (s service) Shutdown(ctx context.Context, r taskAPI.ShutdownRequest) (*ptypes.Empty, error) {`
	`586`	`+ s.cancel()`
`582`	`587`	`os.Exit(0)`
`583`	`588`	`return empty, nil`
`584`	`589`	`}`
`@@ -698,7 +703,10 @@ func (s *service) getContainerPids(ctx context.Context, id string) ([]uint32, er`
`698`	`703`
`699`	`704`	`func (s *service) forward(publisher events.Publisher) {`
`700`	`705`	`for e := range s.events {`
`701`		`- if err := publisher.Publish(s.context, getTopic(s.context, e), e); err != nil {`
	`706`	`+ ctx, cancel := context.WithTimeout(s.context, 5*time.Second)`
	`707`	`+ err := publisher.Publish(ctx, getTopic(e), e)`
	`708`	`+ cancel()`
	`709`	`+ if err != nil {`
`702`	`710`	`logrus.WithError(err).Error("post event")`
`703`	`711`	`}`
`704`	`712`	`}`
`@@ -722,7 +730,7 @@ func (s *service) setCgroup(cg cgroups.Cgroup) {`
`722`	`730`	`}`
`723`	`731`	`}`
`724`	`732`
`725`		`-func getTopic(ctx context.Context, e interface{}) string {`
	`733`	`+func getTopic(e interface{}) string {`
`726`	`734`	`switch e.(type) {`
`727`	`735`	`case *eventstypes.TaskCreate:`
`728`	`736`	`return runtime.TaskCreateEventTopic`