Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fully lock adding node queues during hinted handoff #4353

Merged
merged 1 commit into from
Oct 7, 2015
Merged

Fully lock adding node queues during hinted handoff #4353

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 20 additions & 5 deletions services/hh/processor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,10 +103,9 @@ func (p *Processor) loadQueues() error {
return nil
}

// addQueue adds a hinted-handoff queue for the given node. This function is not thread-safe
// and the caller must ensure this function is not called concurrently.
func (p *Processor) addQueue(nodeID uint64) (*queue, error) {
p.mu.Lock()
defer p.mu.Unlock()

path := filepath.Join(p.dir, strconv.FormatUint(nodeID, 10))
if err := os.MkdirAll(path, 0700); err != nil {
return nil, err
Expand All @@ -128,11 +127,27 @@ func (p *Processor) addQueue(nodeID uint64) (*queue, error) {
return queue, nil
}

// WriteShard writes hinted-handoff data for the given shard and node. Since it may manipulate
// hinted-handoff queues, and be called concurrently, it takes a lock during queue access.
func (p *Processor) WriteShard(shardID, ownerID uint64, points []models.Point) error {
p.mu.RLock()
queue, ok := p.queues[ownerID]
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the root cause. p.queues was not being locked when checked for an entry for ownerID.

p.mu.RUnlock()
if !ok {
var err error
if queue, err = p.addQueue(ownerID); err != nil {
if err := func() error {
// Check again under write-lock.
p.mu.Lock()
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This locking pattern is the same as used by the expvar package in the standard libary.

defer p.mu.Unlock()

queue, ok = p.queues[ownerID]
if !ok {
var err error
if queue, err = p.addQueue(ownerID); err != nil {
return err
}
}
return nil
}(); err != nil {
return err
}
}
Expand Down