From 79c45bdc7ad8c40e34b8ae9de897c452bf16036f Mon Sep 17 00:00:00 2001
From: Jed Denlea <jed@fastly.com>
Date: Tue, 7 Apr 2015 19:26:55 -0700
Subject: [PATCH] producer: fix connection race

Prior to this change, it is possible for more than one goroutine to
check:

	if atomic.LoadInt32(&w.state) != StateConnected {

...thereafter calling w.connect.  w.connect first grabs a lock on
w.guard, and then asserts that it be the only goroutine attempting a
connection:

	if !atomic.CompareAndSwapInt32(&w.state, StateInit, StateConnected) {
		return ErrNotConnected
	}

This means if there are two parallel actions which attempt to initiate
the connection, one of them will return ErrNotConnected if the first
succesfully connected.
---
 producer.go | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/producer.go b/producer.go
index 1018d26a..517c14e8 100644
--- a/producer.go
+++ b/producer.go
@@ -243,7 +243,11 @@ func (w *Producer) connect() error {
 		return ErrStopped
 	}
 
-	if !atomic.CompareAndSwapInt32(&w.state, StateInit, StateConnected) {
+	switch state := atomic.LoadInt32(&w.state); state {
+	case StateInit:
+	case StateConnected:
+		return nil
+	default:
 		return ErrNotConnected
 	}
 
@@ -258,9 +262,9 @@ func (w *Producer) connect() error {
 	if err != nil {
 		w.conn.Close()
 		w.log(LogLevelError, "(%s) error connecting to nsqd - %s", w.addr, err)
-		atomic.StoreInt32(&w.state, StateInit)
 		return err
 	}
+	atomic.StoreInt32(&w.state, StateConnected)
 	w.closeChan = make(chan int)
 	w.wg.Add(1)
 	go w.router()