Merge pull request #1428 from iand/fix/dataraces

Fix data races

cluster.go

@@ -45,9 +45,7 @@ const (
 	maxAlerts           = 1000
 )
 
-var (
-	errFollowerMode = errors.New("this peer is configured to be in follower mode. Write operations are disabled")
-)
+var errFollowerMode = errors.New("this peer is configured to be in follower mode. Write operations are disabled")
 
 // Cluster is the main IPFS cluster component. It provides
 // the go-API for it and orchestrates the components that make up the system.
@@ -322,7 +320,6 @@ func (c *Cluster) pushInformerMetrics(ctx context.Context, informer Informer) {
 		}
 
 		metric, err := c.sendInformerMetric(ctx, informer)
-
 		if err != nil {
 			if (retries % retryWarnMod) == 0 {
 				logger.Errorf("error broadcasting metric: %s", err)
@@ -392,9 +389,8 @@ func (c *Cluster) pushPingMetrics(ctx context.Context) {
 // Alerts returns the last alerts recorded by this cluster peer with the most
 // recent first.
 func (c *Cluster) Alerts() []api.Alert {
-	alerts := make([]api.Alert, len(c.alerts))
-
 	c.alertsMux.Lock()
+	alerts := make([]api.Alert, len(c.alerts))
 	{
 		total := len(alerts)
 		for i, a := range c.alerts {
@@ -833,7 +829,7 @@ func (c *Cluster) ID(ctx context.Context) *api.ID {
 
 	id := &api.ID{
 		ID: c.id,
-		//PublicKey:          c.host.Peerstore().PubKey(c.id),
+		// PublicKey:          c.host.Peerstore().PubKey(c.id),
 		Addresses:             addrs,
 		ClusterPeers:          peers,
 		ClusterPeersAddresses: addresses,
@@ -1148,7 +1144,6 @@ func (c *Cluster) localPinInfoOp(
 	}
 	// return the last pInfo/err, should be the root Cid if everything ok
 	return pInfo, err
-
 }
 
 // RecoverAll triggers a RecoverAllLocal operation on all peers.

informer/disk/disk.go

@@ -5,6 +5,7 @@ package disk
 import (
 	"context"
 	"fmt"
+	"sync"
 
 	"github.com/ipfs/ipfs-cluster/api"
 
@@ -29,7 +30,9 @@ var logger = logging.Logger("diskinfo")
 // Informer is a simple object to implement the ipfscluster.Informer
 // and Component interfaces.
 type Informer struct {
-	config    *Config
+	config *Config // set when created, readonly
+
+	mu        sync.Mutex // guards access to following fields
 	rpcClient *rpc.Client
 }
 
@@ -53,6 +56,8 @@ func (disk *Informer) Name() string {
 // SetClient provides us with an rpc.Client which allows
 // contacting other components in the cluster.
 func (disk *Informer) SetClient(c *rpc.Client) {
+	disk.mu.Lock()
+	defer disk.mu.Unlock()
 	disk.rpcClient = c
 }
 
@@ -62,6 +67,9 @@ func (disk *Informer) Shutdown(ctx context.Context) error {
 	_, span := trace.StartSpan(ctx, "informer/disk/Shutdown")
 	defer span.End()
 
+	disk.mu.Lock()
+	defer disk.mu.Unlock()
+
 	disk.rpcClient = nil
 	return nil
 }
@@ -72,7 +80,11 @@ func (disk *Informer) GetMetric(ctx context.Context) *api.Metric {
 	ctx, span := trace.StartSpan(ctx, "informer/disk/GetMetric")
 	defer span.End()
 
-	if disk.rpcClient == nil {
+	disk.mu.Lock()
+	rpcClient := disk.rpcClient
+	disk.mu.Unlock()
+
+	if rpcClient == nil {
 		return &api.Metric{
 			Name:  disk.Name(),
 			Valid: false,
@@ -84,7 +96,7 @@ func (disk *Informer) GetMetric(ctx context.Context) *api.Metric {
 
 	valid := true
 
-	err := disk.rpcClient.CallContext(
+	err := rpcClient.CallContext(
 		ctx,
 		"",
 		"IPFSConnector",

test/ipfs_mock.go

@@ -40,9 +40,11 @@ type IpfsMock struct {
 	Port       int
 	pinMap     state.State
 	BlockStore map[string][]byte
-	reqCounts  map[string]int
 	reqCounter chan string
 
+	reqCountsMux sync.Mutex // guards access to reqCounts
+	reqCounts    map[string]int
+
 	closeMux sync.Mutex
 	closed   bool
 }
@@ -142,18 +144,20 @@ func NewIpfsMock(t *testing.T) *IpfsMock {
 	m.Port = i
 	m.Addr = h[0]
 	return m
-
 }
 
 func (m *IpfsMock) countRequests() {
 	for str := range m.reqCounter {
+		m.reqCountsMux.Lock()
 		m.reqCounts[str]++
+		m.reqCountsMux.Unlock()
 	}
 }
 
 // GetCount allows to get the number of times and endpoint was called.
-// Do not use concurrently to requests happening.
 func (m *IpfsMock) GetCount(path string) int {
+	m.reqCountsMux.Lock()
+	defer m.reqCountsMux.Unlock()
 	return m.reqCounts[path]
 }
 
@@ -432,7 +436,7 @@ func (m *IpfsMock) handler(w http.ResponseWriter, r *http.Request) {
 		resp := mockRepoStatResp{
 			RepoSize:   uint64(len) * 1000,
 			NumObjects: numObjs,
-			StorageMax: 10000000000, //10 GB
+			StorageMax: 10000000000, // 10 GB
 		}
 		j, _ := json.Marshal(resp)
 		w.Write(j)