From fc231ff80248cff10ef82aed9b343e3d94bbd4fc Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Thu, 8 Nov 2018 17:01:05 +0300 Subject: [PATCH 1/7] Fixes #1731 --- cmd/import.go | 5 ++--- ctl/import.go | 16 +++++++++------- 2 files changed, 11 insertions(+), 10 deletions(-) diff --git a/cmd/import.go b/cmd/import.go index 878aad286..81a12a47f 100644 --- a/cmd/import.go +++ b/cmd/import.go @@ -19,10 +19,8 @@ import ( "io" "github.com/pilosa/pilosa" - - "github.com/spf13/cobra" - "github.com/pilosa/pilosa/ctl" + "github.com/spf13/cobra" ) var Importer *ctl.ImportCommand @@ -55,6 +53,7 @@ omitted. If it is present then its format should be YYYY-MM-DDTHH:MM. flags.StringVarP(&Importer.Field, "field", "f", "", "Field to import into.") flags.BoolVar(&Importer.IndexOptions.Keys, "index-keys", false, "Specify keys=true when creating an index") flags.BoolVar(&Importer.FieldOptions.Keys, "field-keys", false, "Specify keys=true when creating a field") + flags.StringVar(&Importer.FieldOptions.Type, "field-type", "", "Specify the field type when creating a field. One of: set, int, time, bool, mutex") flags.Int64Var(&Importer.FieldOptions.Min, "field-min", 0, "Specify the minimum for an int field on creation") flags.Int64Var(&Importer.FieldOptions.Max, "field-max", 0, "Specify the maximum for an int field on creation") flags.StringVar(&Importer.FieldOptions.CacheType, "field-cache-type", pilosa.CacheTypeRanked, "Specify the cache type for a set field on creation. One of: none, lru, ranked") diff --git a/ctl/import.go b/ctl/import.go index d49b50eb3..ab19064ce 100644 --- a/ctl/import.go +++ b/ctl/import.go @@ -99,13 +99,15 @@ func (cmd *ImportCommand) Run(ctx context.Context) error { cmd.client = client if cmd.CreateSchema { - // set the correct type for the field - if cmd.FieldOptions.TimeQuantum != "" { - cmd.FieldOptions.Type = "time" - } else if cmd.FieldOptions.Min != 0 || cmd.FieldOptions.Max != 0 { - cmd.FieldOptions.Type = "int" - } else { - cmd.FieldOptions.Type = "set" + if cmd.FieldOptions.Type == "" { + // set the correct type for the field + if cmd.FieldOptions.TimeQuantum != "" { + cmd.FieldOptions.Type = "time" + } else if cmd.FieldOptions.Min != 0 || cmd.FieldOptions.Max != 0 { + cmd.FieldOptions.Type = "int" + } else { + cmd.FieldOptions.Type = "set" + } } err := cmd.ensureSchema(ctx) if err != nil { From 27c222f02dda7b681ceddd1caf76670b633ef913 Mon Sep 17 00:00:00 2001 From: Yuce Tekol Date: Thu, 8 Nov 2018 17:02:07 +0300 Subject: [PATCH 2/7] Refactored missing executeRequest bits; check resp is not nil --- http/client.go | 18 +++++++----------- 1 file changed, 7 insertions(+), 11 deletions(-) diff --git a/http/client.go b/http/client.go index 0d6df281f..3d19df227 100644 --- a/http/client.go +++ b/http/client.go @@ -91,9 +91,7 @@ func (c *InternalClient) maxShardByIndex(ctx context.Context) (map[string]uint64 defer resp.Body.Close() var rsp getShardsMaxResponse - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("http: status=%d", resp.StatusCode) - } else if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil { + if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil { return nil, fmt.Errorf("json decode: %s", err) } @@ -152,7 +150,7 @@ func (c *InternalClient) CreateIndex(ctx context.Context, index string, opt pilo // Execute request against the host. resp, err := c.executeRequest(req.WithContext(ctx)) if err != nil { - if resp.StatusCode == http.StatusConflict { + if resp != nil && resp.StatusCode == http.StatusConflict { return pilosa.ErrIndexExists } return err @@ -258,8 +256,6 @@ func (c *InternalClient) QueryNode(ctx context.Context, uri *pilosa.URI, index s body, err := ioutil.ReadAll(resp.Body) if err != nil { return nil, errors.Wrap(err, "reading") - } else if resp.StatusCode != http.StatusOK { - return nil, errors.New(string(body)) } qresp := &pilosa.QueryResponse{} @@ -689,7 +685,7 @@ func (c *InternalClient) backupShardNode(ctx context.Context, index, field strin // Execute request. resp, err := c.executeRequest(req.WithContext(ctx)) if err != nil { - if resp.StatusCode == http.StatusNotFound { + if resp != nil && resp.StatusCode == http.StatusNotFound { return nil, pilosa.ErrFragmentNotFound } return nil, err @@ -746,7 +742,7 @@ func (c *InternalClient) CreateFieldWithOptions(ctx context.Context, index, fiel // Execute request against the host. resp, err := c.executeRequest(req.WithContext(ctx)) if err != nil { - if resp.StatusCode == http.StatusConflict { + if resp != nil && resp.StatusCode == http.StatusConflict { return pilosa.ErrFieldExists } return err @@ -782,7 +778,7 @@ func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, in resp, err := c.executeRequest(req.WithContext(ctx)) if err != nil { // Return the appropriate error. - if resp.StatusCode == http.StatusNotFound { + if resp != nil && resp.StatusCode == http.StatusNotFound { return nil, pilosa.ErrFragmentNotFound } return nil, err @@ -825,7 +821,7 @@ func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index, resp, err := c.executeRequest(req.WithContext(ctx)) if err != nil { - if resp.StatusCode == http.StatusNotFound { + if resp != nil && resp.StatusCode == http.StatusNotFound { return nil, nil, nil } return nil, nil, err @@ -904,7 +900,7 @@ func (c *InternalClient) RowAttrDiff(ctx context.Context, uri *pilosa.URI, index // Execute request. resp, err := c.executeRequest(req.WithContext(ctx)) if err != nil { - if resp.StatusCode == http.StatusNotFound { + if resp != nil && resp.StatusCode == http.StatusNotFound { return nil, pilosa.ErrFieldNotFound } return nil, err From 65f478470f83d52418fa2de33541c6bd0a95b443 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 19 Nov 2018 15:00:09 -0600 Subject: [PATCH 3/7] logging cleanup - start with lowercase unless reporting error or warning --- api.go | 2 +- cluster.go | 19 +++++++------------ fragment.go | 1 - holder.go | 4 ++-- server.go | 6 ++---- server/server.go | 8 ++++---- translate.go | 7 ++----- 7 files changed, 18 insertions(+), 29 deletions(-) diff --git a/api.go b/api.go index 7a95ce76e..a55e23f7b 100644 --- a/api.go +++ b/api.go @@ -957,7 +957,7 @@ func (api *API) validateShardOwnership(indexName string, shard uint64) error { } func (api *API) indexField(indexName string, fieldName string, shard uint64) (*Index, *Field, error) { - api.server.logger.Printf("importing: %v %v %v", indexName, fieldName, shard) + api.server.logger.Debugf("importing: %v %v %v", indexName, fieldName, shard) // Find the Index. index := api.holder.Index(indexName) diff --git a/cluster.go b/cluster.go index e22781a25..fad19de1b 100644 --- a/cluster.go +++ b/cluster.go @@ -347,8 +347,6 @@ func (c *cluster) unprotectedUpdateCoordinator(n *Node) bool { // addNode adds a node to the Cluster and updates and saves the // new topology. unprotected. func (c *cluster) addNode(node *Node) error { - c.logger.Printf("add node %s to cluster on %s", node, c.Node) - // If the node being added is the coordinator, set it for this node. if node.IsCoordinator { c.Coordinator = node.ID @@ -481,7 +479,7 @@ func (c *cluster) setNodeState(state string) error { // nolint: unparam State: state, } - c.logger.Printf("Sending State %s (%s)", state, c.Coordinator) + c.logger.Printf("sending state %s (%s)", state, c.Coordinator) if err := c.sendTo(c.coordinatorNode(), ns); err != nil { return fmt.Errorf("sending node state error: err=%s", err) } @@ -970,7 +968,6 @@ func (c *cluster) close() error { } func (c *cluster) markAsJoined() { - c.logger.Printf("mark node as joined (received coordinator update)") if !c.joined { c.joined = true close(c.joining) @@ -1069,7 +1066,6 @@ func (c *cluster) unprotectedSetStateAndBroadcast(state string) error { } // Broadcast cluster status changes to the cluster. status := c.unprotectedStatus() - c.logger.Printf("broadcasting ClusterStatus: %s", status) return c.broadcaster.SendSync(status) // TODO fix c.Status } @@ -1246,7 +1242,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { return errors.Wrap(err, "merging cluster status") } - c.logger.Printf("MergeClusterStatus done, start goroutine") + c.logger.Printf("done MergeClusterStatus, start goroutine") // The actual resizing runs in a goroutine because we don't want to block // the distribution of other ResizeInstructions to the rest of the cluster. @@ -1266,7 +1262,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { if err := func() error { // Sync the schema received in the resize instruction. - c.logger.Printf("Holder ApplySchema") + c.logger.Debugf("holder applySchema") if err := c.holder.applySchema(instr.Schema); err != nil { return errors.Wrap(err, "applying schema") } @@ -1651,17 +1647,17 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) { switch e.Event { case NodeJoin: - c.logger.Printf("nodeJoin of %s on %s", e.Node.URI, c.Node.URI) + c.logger.Debugf("nodeJoin of %s on %s", e.Node.URI, c.Node.URI) // Ignore the event if this is not the coordinator. if !c.isCoordinator() { return nil } return c.nodeJoin(e.Node) case NodeLeave: - c.logger.Printf("received node leave on %s: %s, uri: %v", c.Node, e.Node, e.Node.URI) c.mu.Lock() defer c.mu.Unlock() if c.unprotectedIsCoordinator() { + c.logger.Printf("received node leave: %v", e.Node) // if removeNodeBasicSorted succeeds, that means that the node was // not already removed by a removeNode request. We treat this as the // host being temporarily unavailable, and expect it to come back @@ -1673,7 +1669,6 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) { err = c.unprotectedSetStateAndBroadcast(c.determineClusterState()) } } - c.logger.Printf("finished node leave on %s: %s, uri: %v", c.Node, e.Node, e.Node.URI) case NodeUpdate: c.logger.Printf("received node update event: id: %v, string: %v, uri: %v", e.Node.ID, e.Node.String(), e.Node.URI) // NodeUpdate is intentionally not implemented. @@ -1686,7 +1681,7 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) { func (c *cluster) nodeJoin(node *Node) error { c.mu.Lock() defer c.mu.Unlock() - c.logger.Printf("NodeJoin event on coordinator, node: %s, id: %s", node.URI, node.ID) + c.logger.Printf("node join event on coordinator, node: %s, id: %s", node.URI, node.ID) if c.needTopologyAgreement() { // A host that is not part of the topology can't be added to the STARTING cluster. if !c.Topology.ContainsID(node.ID) { @@ -1726,7 +1721,7 @@ func (c *cluster) nodeJoin(node *Node) error { // the cluster. if cnode := c.unprotectedNodeByID(node.ID); cnode != nil { if cnode.URI != node.URI { - c.logger.Printf("Node: %v changed URI from %s to %s", cnode.ID, cnode.URI, node.URI) + c.logger.Printf("node: %v changed URI from %s to %s", cnode.ID, cnode.URI, node.URI) cnode.URI = node.URI } return c.unprotectedSetStateAndBroadcast(c.determineClusterState()) diff --git a/fragment.go b/fragment.go index 805112039..a4e4fd72d 100644 --- a/fragment.go +++ b/fragment.go @@ -1734,7 +1734,6 @@ func (f *fragment) snapshot() error { // f.mu must be locked when calling it. func unprotectedWriteToFragment(f *fragment, bm *roaring.Bitmap) error { // nolint: interfacer - f.Logger.Printf("fragment: snapshotting %s/%s/%s/%d", f.index, f.field, f.view, f.shard) completeMessage := fmt.Sprintf("fragment: snapshot complete %s/%s/%s/%d", f.index, f.field, f.view, f.shard) start := time.Now() defer track(start, completeMessage, f.stats, f.Logger) diff --git a/holder.go b/holder.go index 68643915e..ed8365a77 100644 --- a/holder.go +++ b/holder.go @@ -474,7 +474,7 @@ func (h *Holder) flushCaches() { } if err := fragment.FlushCache(); err != nil { - h.Logger.Printf("error flushing cache: err=%s, path=%s", err, fragment.cachePath()) + h.Logger.Printf("ERROR flushing cache: err=%s, path=%s", err, fragment.cachePath()) } } } @@ -535,7 +535,7 @@ func (h *Holder) setFileLimit() { h.Logger.Printf("ERROR checking open file limit: %s", err) } else { if oldLimit.Cur < fileLimit { - h.Logger.Printf("WARNING: Tried to set open file limit to %d, but it is %d. You may consider running \"sudo ulimit -n %d\" before starting Pilosa to avoid \"too many open files\" error. See https://www.pilosa.com/docs/administration/#open-file-limits for more information.", fileLimit, oldLimit.Cur, fileLimit) + h.Logger.Printf("WARNING: Tried to set open file limit to %d, but it is %d. You may consider running \"sudo ulimit -n %d\" before starting Pilosa to avoid \"too many open files\" error. See https://www.pilosa.com/docs/latest/administration/#open-file-limits for more information.", fileLimit, oldLimit.Cur, fileLimit) } } } diff --git a/server.go b/server.go index 24385c4ea..47dc63938 100644 --- a/server.go +++ b/server.go @@ -585,7 +585,6 @@ func (s *Server) SendSync(m Message) error { msg = append([]byte{getMessageType(m)}, msg...) for _, node := range s.cluster.nodes { node := node - s.logger.Printf("SendSync to: %s", node.URI) // Don't forward the message to ourselves. if s.uri == node.URI { continue @@ -606,7 +605,6 @@ func (s *Server) SendAsync(m Message) error { // SendTo represents an implementation of Broadcaster. func (s *Server) SendTo(to *Node, m Message) error { - s.logger.Printf("SendTo: %s", to.URI) msg, err := s.serializer.Marshal(m) if err != nil { return fmt.Errorf("marshaling message: %v", err) @@ -658,7 +656,7 @@ func (s *Server) mergeRemoteStatus(ns *NodeStatus) error { // if we don't know about a field locally, log an error because // fields should be created and synced prior to shard creation if f == nil { - s.logger.Printf("Local Field not found: %s/%s", is.Name, fs.Name) + s.logger.Printf("local field not found: %s/%s", is.Name, fs.Name) continue } if err := f.AddRemoteAvailableShards(fs.AvailableShards); err != nil { @@ -703,7 +701,7 @@ func (s *Server) monitorDiagnostics() { s.diagnostics.CheckVersion() err = s.diagnostics.Flush() if err != nil { - s.logger.Printf("Diagnostics error: %s", err) + s.logger.Printf("diagnostics error: %s", err) } } diff --git a/server/server.go b/server/server.go index 1e140d1f0..de0d3eae0 100644 --- a/server/server.go +++ b/server/server.go @@ -142,7 +142,7 @@ func (m *Command) Start() (err error) { go func() { err := m.Handler.Serve() if err != nil { - m.logger.Printf("Handler serve error: %v", err) + m.logger.Printf("handler serve error: %v", err) } }() @@ -151,7 +151,7 @@ func (m *Command) Start() (err error) { return errors.Wrap(err, "opening server") } - m.logger.Printf("Listening as %s\n", m.API.Node().URI) + m.logger.Printf("listening as %s\n", m.API.Node().URI) return nil } @@ -163,13 +163,13 @@ func (m *Command) Wait() error { signal.Notify(c, os.Interrupt, syscall.SIGTERM) select { case sig := <-c: - m.logger.Printf("Received %s; gracefully shutting down...\n", sig.String()) + m.logger.Printf("received signal '%s', gracefully shutting down...\n", sig.String()) // Second signal causes a hard shutdown. go func() { <-c; os.Exit(1) }() return errors.Wrap(m.Close(), "closing command") case <-m.done: - m.logger.Printf("Server closed externally") + m.logger.Printf("server closed externally") return nil } } diff --git a/translate.go b/translate.go index 669e1e323..2c3125ad9 100644 --- a/translate.go +++ b/translate.go @@ -195,12 +195,11 @@ func (s *TranslateFile) handlePrimaryStoreEvent(ev primaryStoreEvent) error { } // Stop translate store replication. - s.logger.Printf("stop monitor replication") close(s.replicationClosing) s.repWG.Wait() // Set the primary node for translate store replication. - s.logger.Printf("set primary translate store to %s", ev.id) + s.logger.Debugf("set primary translate store to %s", ev.id) s.primaryID = ev.id if ev.id == "" { s.PrimaryTranslateStore = nil @@ -209,7 +208,6 @@ func (s *TranslateFile) handlePrimaryStoreEvent(ev primaryStoreEvent) error { } // Start translate store replication. Stream from primary, if available. - s.logger.Printf("start monitor replication") if s.PrimaryTranslateStore != nil { s.replicationClosing = make(chan struct{}) s.repWG.Add(1) @@ -386,7 +384,6 @@ func (s *TranslateFile) monitorReplication() { // monitorPrimaryStoreEvents is executed in a separate goroutine and listens for changes // to the primary store assignment. func (s *TranslateFile) monitorPrimaryStoreEvents() { - s.logger.Printf("monitor primary store events") // Keep handling events until the store closes. for { select { @@ -404,7 +401,7 @@ func (s *TranslateFile) replicate(ctx context.Context) error { off := s.size() // Connect to remote primary. - s.logger.Printf("pilosa: replicating from offset %d", off) + s.logger.Debugf("pilosa: replicating from offset %d", off) rc, err := s.PrimaryTranslateStore.Reader(ctx, off) if err != nil { return err From 77598c2cc6f069a2acee6b667219051fecdec1c3 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 19 Nov 2018 16:15:33 -0600 Subject: [PATCH 4/7] dup log output onto stderr to catch panics in log file --- server/server.go | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/server/server.go b/server/server.go index de0d3eae0..41349b21d 100644 --- a/server/server.go +++ b/server/server.go @@ -176,14 +176,18 @@ func (m *Command) Wait() error { // setupLogger sets up the logger based on the configuration. func (m *Command) setupLogger() error { - var err error if m.Config.LogPath == "" { m.logOutput = m.Stderr } else { - m.logOutput, err = os.OpenFile(m.Config.LogPath, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600) + f, err := os.OpenFile(m.Config.LogPath, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600) if err != nil { return errors.Wrap(err, "opening file") } + m.logOutput = f + err = syscall.Dup2(int(f.Fd()), int(os.Stderr.Fd())) + if err != nil { + return errors.Wrap(err, "dup2ing stderr onto logfile") + } } if m.Config.Verbose { From 8bc110458568bb0e62dcb37e3a07d3c38ac03032 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Tue, 20 Nov 2018 08:56:12 -0600 Subject: [PATCH 5/7] fix fragment checksums race condition --- fragment.go | 6 +++--- fragment_internal_test.go | 17 +++++++++++++++++ 2 files changed, 20 insertions(+), 3 deletions(-) diff --git a/fragment.go b/fragment.go index a4e4fd72d..bab1f3ec0 100644 --- a/fragment.go +++ b/fragment.go @@ -1492,9 +1492,6 @@ func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64, options *Impor lastRowID = rowID rowSet[rowID] = struct{}{} } - - // Invalidate block checksum. - delete(f.checksums, int(rowID/HashBlockSize)) } f.mu.Lock() @@ -1518,6 +1515,9 @@ func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64, options *Impor // Update cache counts for all affected rows. for rowID := range rowSet { + // Invalidate block checksum. + delete(f.checksums, int(rowID/HashBlockSize)) + n := results.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth) f.cache.BulkAdd(rowID, n) } diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 4e3007957..b36bf3380 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -25,6 +25,8 @@ import ( "testing" "testing/quick" + "golang.org/x/sync/errgroup" + "github.com/davecgh/go-spew/spew" "github.com/pilosa/pilosa/pql" "github.com/pilosa/pilosa/roaring" @@ -1399,6 +1401,21 @@ func TestFragment_ImportSet(t *testing.T) { } } +func TestFragment_ConcurrentImport(t *testing.T) { + t.Run("bulkImportStandard", func(t *testing.T) { + f := mustOpenFragment("i", "f", viewStandard, 0, "") + defer f.Close() + + eg := errgroup.Group{} + eg.Go(func() error { return f.bulkImportStandard([]uint64{1, 2}, []uint64{1, 2}, &ImportOptions{}) }) + eg.Go(func() error { return f.bulkImportStandard([]uint64{3, 4}, []uint64{3, 4}, &ImportOptions{}) }) + err := eg.Wait() + if err != nil { + t.Fatalf("importing data to fragment: %v", err) + } + }) +} + // Ensure a fragment can import mutually exclusive values. func TestFragment_ImportMutex(t *testing.T) { tests := []struct { From 5458eb1656934ecd8cad846425bc6be8b907cb54 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Tue, 20 Nov 2018 13:04:08 -0600 Subject: [PATCH 6/7] fix holder.opened race with absurd lockedChan --- cluster.go | 2 +- executor.go | 2 +- holder.go | 36 ++++++++++++++++++++++++++++++++---- server.go | 2 +- 4 files changed, 35 insertions(+), 7 deletions(-) diff --git a/cluster.go b/cluster.go index fad19de1b..a70bddc77 100644 --- a/cluster.go +++ b/cluster.go @@ -1249,7 +1249,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { go func() { // Make sure the holder has opened. - <-c.holder.opened + c.holder.opened.Recv() // Prepare the return message. complete := &ResizeInstructionComplete{ diff --git a/executor.go b/executor.go index c4153ab61..9062d9176 100644 --- a/executor.go +++ b/executor.go @@ -2055,7 +2055,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64, if !opt.Remote { nodes = Nodes(e.Cluster.nodes).Clone() } else { - nodes = []*Node{e.Cluster.unprotectedNodeByID(e.Node.ID)} + nodes = []*Node{e.Cluster.nodeByID(e.Node.ID)} } // Start mapping across all primary owners. diff --git a/holder.go b/holder.go index ed8365a77..bb9e9394a 100644 --- a/holder.go +++ b/holder.go @@ -57,7 +57,7 @@ type Holder struct { NewPrimaryTranslateStore func(interface{}) TranslateStore // opened channel is closed once Open() completes. - opened chan struct{} + opened lockedChan broadcaster broadcaster @@ -79,13 +79,39 @@ type Holder struct { Logger logger.Logger } +// lockedChan looks a little ridiculous admittedly, but exists for good reason. +// The channel within is used (for example) to signal to other goroutines when +// the Holder has finished opening (via closing the channel). However, it is +// possible for the holder to be closed and then reopened, but a channel which +// is closed cannot be re-opened. We must create a new channel - this creates a +// data race with any goroutine which might be accessing the channel. To ensure +// that there is no data race on the value of the channel itself, we wrap any +// operation on it with an RWMutex so that we can guarantee that nothing is +// trying to listen on it when it gets swapped. +type lockedChan struct { + ch chan struct{} + mu sync.RWMutex +} + +func (lc *lockedChan) Close() { + lc.mu.RLock() + close(lc.ch) + lc.mu.RUnlock() +} + +func (lc *lockedChan) Recv() { + lc.mu.RLock() + <-lc.ch + lc.mu.RUnlock() +} + // NewHolder returns a new instance of Holder. func NewHolder() *Holder { return &Holder{ indexes: make(map[string]*Index), closing: make(chan struct{}), - opened: make(chan struct{}), + opened: lockedChan{ch: make(chan struct{})}, translateFile: NewTranslateFile(), NewPrimaryTranslateStore: newNopTranslateStore, @@ -159,7 +185,7 @@ func (h *Holder) Open() error { h.Stats.Open() - close(h.opened) + h.opened.Close() return nil } @@ -184,7 +210,9 @@ func (h *Holder) Close() error { } // Reset opened in case Holder needs to be reopened. - h.opened = make(chan struct{}) + h.opened.mu.Lock() + h.opened.ch = make(chan struct{}) + h.opened.mu.Unlock() return nil } diff --git a/server.go b/server.go index 47dc63938..7eb62de3e 100644 --- a/server.go +++ b/server.go @@ -628,7 +628,7 @@ func (s *Server) handleRemoteStatus(pb Message) { go func() { // Make sure the holder has opened. - <-s.holder.opened + s.holder.opened.Recv() err := s.mergeRemoteStatus(pb.(*NodeStatus)) if err != nil { From e1adb8ce5f58b1e004da085d58113c6b8222461f Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Tue, 20 Nov 2018 13:20:39 -0600 Subject: [PATCH 7/7] fix view.createFragment race --- executor.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/executor.go b/executor.go index 9062d9176..51cfae201 100644 --- a/executor.go +++ b/executor.go @@ -1664,7 +1664,7 @@ func (e *executor) executeSetRowShard(ctx context.Context, index string, c *pql. if err != nil { return false, errors.Wrap(err, "creating view") } - fragment, err = view.createFragmentIfNotExists(shard) + fragment, err = view.CreateFragmentIfNotExists(shard) if err != nil { return false, errors.Wrapf(err, "creating fragment: %d", shard) }