Merge pull request #26819 from ripcurld00d/fix_stats_mutex

Hide the mutex lock in formatter.ContainerStats

command/container/stats.go

@@ -163,11 +163,10 @@ func runStats(dockerCli *command.DockerCli, opts *statsOptions) error {
 		var errs []string
 		cStats.mu.Lock()
 		for _, c := range cStats.cs {
-			c.Mu.Lock()
+			cErr := c.GetError()
-			if c.Err != nil {
+			if cErr != nil {
-				errs = append(errs, fmt.Sprintf("%s: %v", c.Name, c.Err))
+				errs = append(errs, fmt.Sprintf("%s: %v", c.Name, cErr))
 			}
-			c.Mu.Unlock()
 		}
 		cStats.mu.Unlock()
 		if len(errs) > 0 {
@@ -186,7 +185,7 @@ func runStats(dockerCli *command.DockerCli, opts *statsOptions) error {
 		Format: formatter.NewStatsFormat(f, daemonOSType),
 	}
 
-	cleanHeader := func() {
+	cleanScreen := func() {
 		if !opts.noStream {
 			fmt.Fprint(dockerCli.Out(), "\033[2J")
 			fmt.Fprint(dockerCli.Out(), "\033[H")
@@ -195,14 +194,17 @@ func runStats(dockerCli *command.DockerCli, opts *statsOptions) error {
 
 	var err error
 	for range time.Tick(500 * time.Millisecond) {
-		cleanHeader()
+		cleanScreen()
-		cStats.mu.RLock()
+		ccstats := []formatter.StatsEntry{}
-		csLen := len(cStats.cs)
+		cStats.mu.Lock()
-		if err = formatter.ContainerStatsWrite(statsCtx, cStats.cs); err != nil {
+		for _, c := range cStats.cs {
+			ccstats = append(ccstats, c.GetStatistics())
+		}
+		cStats.mu.Unlock()
+		if err = formatter.ContainerStatsWrite(statsCtx, ccstats); err != nil {
 			break
 		}
-		cStats.mu.RUnlock()
+		if len(cStats.cs) == 0 && !showAll {
-		if csLen == 0 && !showAll {
 			break
 		}
 		if opts.noStream {

command/container/stats_helpers.go

@@ -17,7 +17,7 @@ import (
 
 type stats struct {
 	ostype string
-	mu     sync.RWMutex
+	mu     sync.Mutex
 	cs     []*formatter.ContainerStats
 }
 
@@ -72,9 +72,7 @@ func collect(s *formatter.ContainerStats, ctx context.Context, cli client.APICli
 
 	response, err := cli.ContainerStats(ctx, s.Name, streamStats)
 	if err != nil {
-		s.Mu.Lock()
+		s.SetError(err)
-		s.Err = err
-		s.Mu.Unlock()
 		return
 	}
 	defer response.Body.Close()
@@ -88,6 +86,9 @@ func collect(s *formatter.ContainerStats, ctx context.Context, cli client.APICli
 				cpuPercent        = 0.0
 				blkRead, blkWrite uint64 // Only used on Linux
 				mem               = 0.0
+				memLimit          = 0.0
+				memPerc           = 0.0
+				pidsStatsCurrent  uint64
 			)
 
 			if err := dec.Decode(&v); err != nil {
@@ -113,26 +114,27 @@ func collect(s *formatter.ContainerStats, ctx context.Context, cli client.APICli
 				cpuPercent = calculateCPUPercentUnix(previousCPU, previousSystem, v)
 				blkRead, blkWrite = calculateBlockIO(v.BlkioStats)
 				mem = float64(v.MemoryStats.Usage)
-
+				memLimit = float64(v.MemoryStats.Limit)
+				memPerc = memPercent
+				pidsStatsCurrent = v.PidsStats.Current
 			} else {
 				cpuPercent = calculateCPUPercentWindows(v)
 				blkRead = v.StorageStats.ReadSizeBytes
 				blkWrite = v.StorageStats.WriteSizeBytes
 				mem = float64(v.MemoryStats.PrivateWorkingSet)
 			}
-
+			netRx, netTx := calculateNetwork(v.Networks)
-			s.Mu.Lock()
+			s.SetStatistics(formatter.StatsEntry{
-			s.CPUPercentage = cpuPercent
+				CPUPercentage:    cpuPercent,
-			s.Memory = mem
+				Memory:           mem,
-			s.NetworkRx, s.NetworkTx = calculateNetwork(v.Networks)
+				MemoryPercentage: memPerc,
-			s.BlockRead = float64(blkRead)
+				MemoryLimit:      memLimit,
-			s.BlockWrite = float64(blkWrite)
+				NetworkRx:        netRx,
-			if daemonOSType != "windows" {
+				NetworkTx:        netTx,
-				s.MemoryLimit = float64(v.MemoryStats.Limit)
+				BlockRead:        float64(blkRead),
-				s.MemoryPercentage = memPercent
+				BlockWrite:       float64(blkWrite),
-				s.PidsCurrent = v.PidsStats.Current
+				PidsCurrent:      pidsStatsCurrent,
-			}
+			})
-			s.Mu.Unlock()
 			u <- nil
 			if !streamStats {
 				return
@@ -144,18 +146,7 @@ func collect(s *formatter.ContainerStats, ctx context.Context, cli client.APICli
 		case <-time.After(2 * time.Second):
 			// zero out the values if we have not received an update within
 			// the specified duration.
-			s.Mu.Lock()
+			s.SetErrorAndReset(errors.New("timeout waiting for stats"))
-			s.CPUPercentage = 0
-			s.Memory = 0
-			s.MemoryPercentage = 0
-			s.MemoryLimit = 0
-			s.NetworkRx = 0
-			s.NetworkTx = 0
-			s.BlockRead = 0
-			s.BlockWrite = 0
-			s.PidsCurrent = 0
-			s.Err = errors.New("timeout waiting for stats")
-			s.Mu.Unlock()
 			// if this is the first stat you get, release WaitGroup
 			if !getFirst {
 				getFirst = true
@@ -163,12 +154,10 @@ func collect(s *formatter.ContainerStats, ctx context.Context, cli client.APICli
 			}
 		case err := <-u:
 			if err != nil {
-				s.Mu.Lock()
+				s.SetError(err)
-				s.Err = err
-				s.Mu.Unlock()
 				continue
 			}
-			s.Err = nil
+			s.SetError(nil)
 			// if this is the first stat you get, release WaitGroup
 			if !getFirst {
 				getFirst = true

command/formatter/stats.go

@@ -4,27 +4,27 @@ import (
 	"fmt"
 	"sync"
 
-	"github.com/docker/go-units"
+	units "src/github.com/docker/go-units"
 )
 
 const (
-	defaultStatsTableFormat    = "table {{.Container}}\t{{.CPUPrec}}\t{{.MemUsage}}\t{{.MemPrec}}\t{{.NetIO}}\t{{.BlockIO}}\t{{.PIDs}}"
+	winOSType                  = "windows"
-	winDefaultStatsTableFormat = "table {{.Container}}\t{{.CPUPrec}}\t{{{.MemUsage}}\t{.NetIO}}\t{{.BlockIO}}"
+	defaultStatsTableFormat    = "table {{.Container}}\t{{.CPUPerc}}\t{{.MemUsage}}\t{{.MemPerc}}\t{{.NetIO}}\t{{.BlockIO}}\t{{.PIDs}}"
+	winDefaultStatsTableFormat = "table {{.Container}}\t{{.CPUPerc}}\t{{{.MemUsage}}\t{.NetIO}}\t{{.BlockIO}}"
 	emptyStatsTableFormat      = "Waiting for statistics..."
 
 	containerHeader  = "CONTAINER"
-	cpuPrecHeader    = "CPU %"
+	cpuPercHeader    = "CPU %"
 	netIOHeader      = "NET I/O"
 	blockIOHeader    = "BLOCK I/O"
-	winMemPrecHeader = "PRIV WORKING SET"  // Used only on Window
+	winMemPercHeader = "PRIV WORKING SET"  // Used only on Window
-	memPrecHeader    = "MEM %"             // Used only on Linux
+	memPercHeader    = "MEM %"             // Used only on Linux
 	memUseHeader     = "MEM USAGE / LIMIT" // Used only on Linux
 	pidsHeader       = "PIDS"              // Used only on Linux
 )
 
-// ContainerStatsAttrs represents the statistics data collected from a container.
+// StatsEntry represents represents the statistics data collected from a container
-type ContainerStatsAttrs struct {
+type StatsEntry struct {
-	Windows          bool
 	Name             string
 	CPUPercentage    float64
 	Memory           float64 // On Windows this is the private working set
@@ -35,19 +35,73 @@ type ContainerStatsAttrs struct {
 	BlockRead        float64
 	BlockWrite       float64
 	PidsCurrent      uint64 // Not used on Windows
+	IsInvalid        bool
+	OSType           string
 }
 
-// ContainerStats represents the containers statistics data.
+// ContainerStats represents an entity to store containers statistics synchronously
 type ContainerStats struct {
-	Mu sync.RWMutex
+	mutex sync.Mutex
-	ContainerStatsAttrs
+	StatsEntry
-	Err error
+	err error
+}
+
+// GetError returns the container statistics error.
+// This is used to determine whether the statistics are valid or not
+func (cs *ContainerStats) GetError() error {
+	cs.mutex.Lock()
+	defer cs.mutex.Unlock()
+	return cs.err
+}
+
+// SetErrorAndReset zeroes all the container statistics and store the error.
+// It is used when receiving time out error during statistics collecting to reduce lock overhead
+func (cs *ContainerStats) SetErrorAndReset(err error) {
+	cs.mutex.Lock()
+	defer cs.mutex.Unlock()
+	cs.CPUPercentage = 0
+	cs.Memory = 0
+	cs.MemoryPercentage = 0
+	cs.MemoryLimit = 0
+	cs.NetworkRx = 0
+	cs.NetworkTx = 0
+	cs.BlockRead = 0
+	cs.BlockWrite = 0
+	cs.PidsCurrent = 0
+	cs.err = err
+	cs.IsInvalid = true
+}
+
+// SetError sets container statistics error
+func (cs *ContainerStats) SetError(err error) {
+	cs.mutex.Lock()
+	defer cs.mutex.Unlock()
+	cs.err = err
+	if err != nil {
+		cs.IsInvalid = true
+	}
+}
+
+// SetStatistics set the container statistics
+func (cs *ContainerStats) SetStatistics(s StatsEntry) {
+	cs.mutex.Lock()
+	defer cs.mutex.Unlock()
+	s.Name = cs.Name
+	s.OSType = cs.OSType
+	cs.StatsEntry = s
+}
+
+// GetStatistics returns container statistics with other meta data such as the container name
+func (cs *ContainerStats) GetStatistics() StatsEntry {
+	cs.mutex.Lock()
+	defer cs.mutex.Unlock()
+	return cs.StatsEntry
 }
 
 // NewStatsFormat returns a format for rendering an CStatsContext
 func NewStatsFormat(source, osType string) Format {
 	if source == TableFormatKey {
-		if osType == "windows" {
+		if osType == winOSType {
 			return Format(winDefaultStatsTableFormat)
 		}
 		return Format(defaultStatsTableFormat)
@@ -58,22 +112,16 @@ func NewStatsFormat(source, osType string) Format {
 // NewContainerStats returns a new ContainerStats entity and sets in it the given name
 func NewContainerStats(name, osType string) *ContainerStats {
 	return &ContainerStats{
-		ContainerStatsAttrs: ContainerStatsAttrs{
+		StatsEntry: StatsEntry{Name: name, OSType: osType},
-			Name:    name,
-			Windows: (osType == "windows"),
-		},
 	}
 }
 
 // ContainerStatsWrite renders the context for a list of containers statistics
-func ContainerStatsWrite(ctx Context, containerStats []*ContainerStats) error {
+func ContainerStatsWrite(ctx Context, containerStats []StatsEntry) error {
 	render := func(format func(subContext subContext) error) error {
 		for _, cstats := range containerStats {
-			cstats.Mu.RLock()
-			cstatsAttrs := cstats.ContainerStatsAttrs
-			cstats.Mu.RUnlock()
 			containerStatsCtx := &containerStatsContext{
-				s: cstatsAttrs,
+				s: cstats,
 			}
 			if err := format(containerStatsCtx); err != nil {
 				return err
@@ -86,7 +134,7 @@ func ContainerStatsWrite(ctx Context, containerStats []*ContainerStats) error {
 
 type containerStatsContext struct {
 	HeaderContext
-	s ContainerStatsAttrs
+	s StatsEntry
 }
 
 func (c *containerStatsContext) Container() string {
@@ -94,42 +142,54 @@ func (c *containerStatsContext) Container() string {
 	return c.s.Name
 }
 
-func (c *containerStatsContext) CPUPrec() string {
+func (c *containerStatsContext) CPUPerc() string {
-	c.AddHeader(cpuPrecHeader)
+	c.AddHeader(cpuPercHeader)
+	if c.s.IsInvalid {
+		return fmt.Sprintf("--")
+	}
 	return fmt.Sprintf("%.2f%%", c.s.CPUPercentage)
 }
 
 func (c *containerStatsContext) MemUsage() string {
 	c.AddHeader(memUseHeader)
-	if !c.s.Windows {
+	if c.s.IsInvalid || c.s.OSType == winOSType {
-		return fmt.Sprintf("%s / %s", units.BytesSize(c.s.Memory), units.BytesSize(c.s.MemoryLimit))
+		return fmt.Sprintf("-- / --")
 	}
-	return fmt.Sprintf("-- / --")
+	return fmt.Sprintf("%s / %s", units.BytesSize(c.s.Memory), units.BytesSize(c.s.MemoryLimit))
 }
 
-func (c *containerStatsContext) MemPrec() string {
+func (c *containerStatsContext) MemPerc() string {
-	header := memPrecHeader
+	header := memPercHeader
-	if c.s.Windows {
+	if c.s.OSType == winOSType {
-		header = winMemPrecHeader
+		header = winMemPercHeader
 	}
 	c.AddHeader(header)
+	if c.s.IsInvalid {
+		return fmt.Sprintf("--")
+	}
 	return fmt.Sprintf("%.2f%%", c.s.MemoryPercentage)
 }
 
 func (c *containerStatsContext) NetIO() string {
 	c.AddHeader(netIOHeader)
+	if c.s.IsInvalid {
+		return fmt.Sprintf("--")
+	}
 	return fmt.Sprintf("%s / %s", units.HumanSizeWithPrecision(c.s.NetworkRx, 3), units.HumanSizeWithPrecision(c.s.NetworkTx, 3))
 }
 
 func (c *containerStatsContext) BlockIO() string {
 	c.AddHeader(blockIOHeader)
+	if c.s.IsInvalid {
+		return fmt.Sprintf("--")
+	}
 	return fmt.Sprintf("%s / %s", units.HumanSizeWithPrecision(c.s.BlockRead, 3), units.HumanSizeWithPrecision(c.s.BlockWrite, 3))
 }
 
 func (c *containerStatsContext) PIDs() string {
 	c.AddHeader(pidsHeader)
-	if !c.s.Windows {
+	if c.s.IsInvalid || c.s.OSType == winOSType {
-		return fmt.Sprintf("%d", c.s.PidsCurrent)
+		return fmt.Sprintf("--")
 	}
-	return fmt.Sprintf("-")
+	return fmt.Sprintf("%d", c.s.PidsCurrent)
 }