vendor/github.com/containerd/containerd/supervisor/monitor_linux.go

   1 package supervisor
   2
   3 import (
   4         "sync"
   5         "syscall"
   6
   7         "github.com/Sirupsen/logrus"
   8         "github.com/containerd/containerd/archutils"
   9         "github.com/containerd/containerd/runtime"
  10 )
  11
  12 // NewMonitor starts a new process monitor and returns it
  13 func NewMonitor() (*Monitor, error) {
  14         m := &Monitor{
  15                 receivers: make(map[int]interface{}),
  16                 exits:     make(chan runtime.Process, 1024),
  17                 ooms:      make(chan string, 1024),
  18         }
  19         fd, err := archutils.EpollCreate1(syscall.EPOLL_CLOEXEC)
  20         if err != nil {
  21                 return nil, err
  22         }
  23         m.epollFd = fd
  24         go m.start()
  25         return m, nil
  26 }
  27
  28 // Monitor represents a runtime.Process monitor
  29 type Monitor struct {
  30         m         sync.Mutex
  31         receivers map[int]interface{}
  32         exits     chan runtime.Process
  33         ooms      chan string
  34         epollFd   int
  35 }
  36
  37 // Exits returns the channel used to notify of a process exit
  38 func (m *Monitor) Exits() chan runtime.Process {
  39         return m.exits
  40 }
  41
  42 // OOMs returns the channel used to notify of a container exit due to OOM
  43 func (m *Monitor) OOMs() chan string {
  44         return m.ooms
  45 }
  46
  47 // Monitor adds a process to the list of the one being monitored
  48 func (m *Monitor) Monitor(p runtime.Process) error {
  49         m.m.Lock()
  50         defer m.m.Unlock()
  51         fd := p.ExitFD()
  52         event := syscall.EpollEvent{
  53                 Fd:     int32(fd),
  54                 Events: syscall.EPOLLHUP,
  55         }
  56         if err := archutils.EpollCtl(m.epollFd, syscall.EPOLL_CTL_ADD, fd, &event); err != nil {
  57                 return err
  58         }
  59         EpollFdCounter.Inc(1)
  60         m.receivers[fd] = p
  61         return nil
  62 }
  63
  64 // MonitorOOM adds a container to the list of the ones monitored for OOM
  65 func (m *Monitor) MonitorOOM(c runtime.Container) error {
  66         m.m.Lock()
  67         defer m.m.Unlock()
  68         o, err := c.OOM()
  69         if err != nil {
  70                 return err
  71         }
  72         fd := o.FD()
  73         event := syscall.EpollEvent{
  74                 Fd:     int32(fd),
  75                 Events: syscall.EPOLLHUP | syscall.EPOLLIN,
  76         }
  77         if err := archutils.EpollCtl(m.epollFd, syscall.EPOLL_CTL_ADD, fd, &event); err != nil {
  78                 return err
  79         }
  80         EpollFdCounter.Inc(1)
  81         m.receivers[fd] = o
  82         return nil
  83 }
  84
  85 // Close cleans up resources allocated by NewMonitor()
  86 func (m *Monitor) Close() error {
  87         return syscall.Close(m.epollFd)
  88 }
  89
  90 func (m *Monitor) processEvent(fd int, event uint32) {
  91         m.m.Lock()
  92         r := m.receivers[fd]
  93         switch t := r.(type) {
  94         case runtime.Process:
  95                 if event == syscall.EPOLLHUP {
  96                         delete(m.receivers, fd)
  97                         if err := syscall.EpollCtl(m.epollFd, syscall.EPOLL_CTL_DEL, fd, &syscall.EpollEvent{
  98                                 Events: syscall.EPOLLHUP,
  99                                 Fd:     int32(fd),
 100                         }); err != nil {
 101                                 logrus.WithField("error", err).Error("containerd: epoll remove fd")
 102                         }
 103                         if err := t.Close(); err != nil {
 104                                 logrus.WithField("error", err).Error("containerd: close process IO")
 105                         }
 106                         EpollFdCounter.Dec(1)
 107                         // defer until lock is released
 108                         defer func() {
 109                                 m.exits <- t
 110                         }()
 111                 }
 112         case runtime.OOM:
 113                 // always flush the event fd
 114                 t.Flush()
 115                 if t.Removed() {
 116                         delete(m.receivers, fd)
 117                         // epoll will remove the fd from its set after it has been closed
 118                         t.Close()
 119                         EpollFdCounter.Dec(1)
 120                 } else {
 121                         // defer until lock is released
 122                         defer func() {
 123                                 m.ooms <- t.ContainerID()
 124                         }()
 125                 }
 126         }
 127         // This cannot be a defer to avoid a deadlock in case the channels
 128         // above get full
 129         m.m.Unlock()
 130 }
 131
 132 func (m *Monitor) start() {
 133         var events [128]syscall.EpollEvent
 134         for {
 135                 n, err := archutils.EpollWait(m.epollFd, events[:], -1)
 136                 if err != nil {
 137                         if err == syscall.EINTR {
 138                                 continue
 139                         }
 140                         logrus.WithField("error", err).Fatal("containerd: epoll wait")
 141                 }
 142                 // process events
 143                 for i := 0; i < n; i++ {
 144                         m.processEvent(int(events[i].Fd), events[i].Events)
 145                 }
 146         }
 147 }