diff --git a/containerd/main.go b/containerd/main.go index 353efe8..cce16a9 100644 --- a/containerd/main.go +++ b/containerd/main.go @@ -89,10 +89,9 @@ func daemon(stateDir string, concurrency, bufferSize int) error { w := containerd.NewWorker(supervisor, wg) go w.Start() } - events := make(chan *containerd.Event, bufferSize) // start the signal handler in the background. go startSignalHandler(supervisor, bufferSize) - if err := supervisor.Start(events); err != nil { + if err := supervisor.Start(); err != nil { return err } server := v1.NewServer(supervisor) diff --git a/supervisor.go b/supervisor.go index 532ad3f..30a053b 100644 --- a/supervisor.go +++ b/supervisor.go @@ -31,6 +31,7 @@ func NewSupervisor(stateDir string, tasks chan *StartTask) (*Supervisor, error) runtime: r, journal: j, tasks: tasks, + events: make(chan *Event, 2048), } // register default event handlers s.handlers = map[EventType]Handler{ @@ -87,16 +88,12 @@ func (s *Supervisor) NotifySubscribers(e *Event) { // executing new containers. // // This event loop is the only thing that is allowed to modify state of containers and processes. -func (s *Supervisor) Start(events chan *Event) error { - if events == nil { - return ErrEventChanNil - } - s.events = events +func (s *Supervisor) Start() error { go func() { // allocate an entire thread to this goroutine for the main event loop // so that nothing else is scheduled over the top of it. goruntime.LockOSThread() - for e := range events { + for e := range s.events { EventsCounter.Inc(1) s.journal.write(e) h, ok := s.handlers[e.Type]