fix(state): don't lock global state lock when adding a task log

internals/overlord/servstate/handlers.go

@@ -190,7 +190,7 @@ func (m *ServiceManager) serviceForStart(task *state.Task, config *plan.Service)
 
 	switch service.state {
 	case stateInitial, stateStarting, stateRunning:
-		taskLogf(task, "Service %q already started.", config.Name)
+		task.Logf("Service %q already started.", config.Name)
 		return nil
 	case stateBackoff, stateStopped, stateExited:
 		// Start allowed when service is backing off, was stopped, or has exited.
@@ -204,14 +204,6 @@ func (m *ServiceManager) serviceForStart(task *state.Task, config *plan.Service)
 	}
 }
 
-func taskLogf(task *state.Task, format string, args ...interface{}) {
-	st := task.State()
-	st.Lock()
-	defer st.Unlock()
-
-	task.Logf(format, args...)
-}
-
 func (m *ServiceManager) doStop(task *state.Task, tomb *tomb.Tomb) error {
 	m.state.Lock()
 	request, err := TaskServiceRequest(task)
@@ -258,19 +250,19 @@ func (m *ServiceManager) serviceForStop(task *state.Task, name string) *serviceD
 
 	service := m.services[name]
 	if service == nil {
-		taskLogf(task, "Service %q has never been started.", name)
+		task.Logf("Service %q has never been started.", name)
 		return nil
 	}
 
 	switch service.state {
 	case stateTerminating, stateKilling:
-		taskLogf(task, "Service %q already stopping.", name)
+		task.Logf("Service %q already stopping.", name)
 		return nil
 	case stateStopped:
-		taskLogf(task, "Service %q already stopped.", name)
+		task.Logf("Service %q already stopped.", name)
 		return nil
 	case stateExited:
-		taskLogf(task, "Service %q had already exited.", name)
+		task.Logf("Service %q had already exited.", name)
 		service.transition(stateStopped)
 		return nil
 	default:

internals/overlord/state/state.go

@@ -96,7 +96,7 @@ type State struct {
 
 	noticeCond *sync.Cond
 
-	modified bool
+	modified atomic.Bool
 
 	cache map[interface{}]interface{}
 
@@ -116,19 +116,19 @@ func New(backend Backend) *State {
 		tasks:               make(map[string]*Task),
 		warnings:            make(map[string]*Warning),
 		notices:             make(map[noticeKey]*Notice),
-		modified:            true,
 		cache:               make(map[interface{}]interface{}),
 		pendingChangeByAttr: make(map[string]func(*Change) bool),
 		taskHandlers:        make(map[int]func(t *Task, old Status, new Status)),
 		changeHandlers:      make(map[int]func(chg *Change, old Status, new Status)),
 	}
+	st.modified.Store(true)
 	st.noticeCond = sync.NewCond(st) // use State.Lock and State.Unlock
 	return st
 }
 
 // Modified returns whether the state was modified since the last checkpoint.
 func (s *State) Modified() bool {
-	return s.modified
+	return s.modified.Load()
 }
 
 // Lock acquires the state lock.
@@ -144,7 +144,7 @@ func (s *State) reading() {
 }
 
 func (s *State) writing() {
-	s.modified = true
+	s.modified.Store(true)
 	if atomic.LoadInt32(&s.muC) != 1 {
 		panic("internal error: accessing state without lock")
 	}
@@ -243,7 +243,14 @@ func (s *State) Unlocker() (unlock func() (relock func())) {
 func (s *State) Unlock() {
 	defer s.unlock()
 
-	if !s.modified || s.backend == nil {
+	if s.backend == nil {
+		return
+	}
+
+	// If state hasn't been modified, do nothing, Otherwise, save state to
+	// disk and clear the flag.
+	modified := s.modified.Swap(false)
+	if !modified {
 		return
 	}
 
@@ -252,7 +259,6 @@ func (s *State) Unlock() {
 	start := time.Now()
 	for time.Since(start) <= unlockCheckpointRetryMaxTime {
 		if err = s.backend.Checkpoint(data); err == nil {
-			s.modified = false
 			return
 		}
 		time.Sleep(unlockCheckpointRetryInterval)
@@ -587,7 +593,7 @@ func ReadState(backend Backend, r io.Reader) (*State, error) {
 	}
 	s.backend = backend
 	s.noticeCond = sync.NewCond(s)
-	s.modified = false
+	s.modified.Store(false)
 	s.cache = make(map[interface{}]interface{})
 	s.pendingChangeByAttr = make(map[string]func(*Change) bool)
 	s.changeHandlers = make(map[int]func(chg *Change, old Status, new Status))

internals/overlord/state/task.go

@@ -17,6 +17,7 @@ package state
 import (
 	"encoding/json"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/canonical/pebble/internals/logger"
@@ -47,9 +48,14 @@ type Task struct {
 	waitTasks    []string
 	haltTasks    []string
 	lanes        []int
-	log          []string
 	change       string
 
+	// Use a fine-grained lock for the logs to avoid holding the global state
+	// lock when just adding a log -- that easily caused deadlocks. See:
+	// https://github.com/canonical/pebble/issues/314
+	logLock sync.Mutex
+	log     []string
+
 	spawnTime time.Time
 	readyTime time.Time
 
@@ -121,7 +127,7 @@ func (t *Task) MarshalJSON() ([]byte, error) {
 		WaitTasks:    t.waitTasks,
 		HaltTasks:    t.haltTasks,
 		Lanes:        t.lanes,
-		Log:          t.log,
+		Log:          t.Log(),
 		Change:       t.change,
 
 		SpawnTime: t.spawnTime,
@@ -165,7 +171,11 @@ func (t *Task) UnmarshalJSON(data []byte) error {
 	t.waitTasks = unmarshalled.WaitTasks
 	t.haltTasks = unmarshalled.HaltTasks
 	t.lanes = unmarshalled.Lanes
-	t.log = unmarshalled.Log
+	func() {
+		t.logLock.Lock()
+		defer t.logLock.Unlock()
+		t.log = unmarshalled.Log
+	}()
 	t.change = unmarshalled.Change
 	t.spawnTime = unmarshalled.SpawnTime
 	if unmarshalled.ReadyTime != nil {
@@ -416,42 +426,56 @@ func FakeTime(now time.Time) (restore func()) {
 }
 
 func (t *Task) addLog(kind, format string, args []interface{}) {
-	if len(t.log) > 9 {
-		copy(t.log, t.log[len(t.log)-9:])
-		t.log = t.log[:9]
-	}
+	var msg string
+	func() {
+		t.logLock.Lock()
+		defer t.logLock.Unlock()
+
+		if len(t.log) > 9 {
+			copy(t.log, t.log[len(t.log)-9:])
+			t.log = t.log[:9]
+		}
+
+		tstr := timeNow().Format(time.RFC3339)
+		msg = tstr + " " + kind + " " + fmt.Sprintf(format, args...)
+		t.log = append(t.log, msg)
+	}()
+
+	t.state.modified.Store(true)
 
-	tstr := timeNow().Format(time.RFC3339)
-	msg := tstr + " " + kind + " " + fmt.Sprintf(format, args...)
-	t.log = append(t.log, msg)
 	logger.Debugf(msg)
 }
 
-// Log returns the most recent messages logged into the task.
+// Log returns a copy of the most recent messages logged into the task.
 //
 // Only the most recent entries logged are returned, potentially with
 // different behavior for different task statuses. How many entries
 // are returned is an implementation detail and may change over time.
 //
 // Messages are prefixed with one of the known message kinds.
 // See details about LogInfo and LogError.
-//
-// The returned slice should not be read from without the
-// state lock held, and should not be written to.
 func (t *Task) Log() []string {
-	t.state.reading()
-	return t.log
+	t.logLock.Lock()
+	defer t.logLock.Unlock()
+
+	logCopy := make([]string, len(t.log))
+	for i, s := range t.log {
+		logCopy[i] = s
+	}
+	return logCopy
 }
 
 // Logf logs information about the progress of the task.
+//
+// The state lock does not need to be held when calling this method.
 func (t *Task) Logf(format string, args ...interface{}) {
-	t.state.writing()
 	t.addLog(LogInfo, format, args)
 }
 
 // Errorf logs error information about the progress of the task.
+//
+// The state lock does not need to be held when calling this method.
 func (t *Task) Errorf(format string, args ...interface{}) {
-	t.state.writing()
 	t.addLog(LogError, format, args)
 }
 

internals/overlord/state/task_test.go

@@ -454,8 +454,6 @@ func (cs *taskSuite) TestMethodEntrance(c *C) {
 		func() { t1.Set("a", 1) },
 		func() { t2.WaitFor(t1) },
 		func() { t1.SetProgress("", 2, 2) },
-		func() { t1.Logf("") },
-		func() { t1.Errorf("") },
 		func() { t1.UnmarshalJSON(nil) },
 		func() { t1.SetProgress("", 1, 1) },
 		func() { t1.JoinLane(1) },
@@ -470,7 +468,6 @@ func (cs *taskSuite) TestMethodEntrance(c *C) {
 		func() { t1.WaitTasks() },
 		func() { t1.HaltTasks() },
 		func() { t1.Progress() },
-		func() { t1.Log() },
 		func() { t1.MarshalJSON() },
 		func() { t1.Progress() },
 		func() { t1.SetProgress("", 0, 1) },