nsqd: allow unbuffered memory chan if ephemeral or deferred

In b3b29b7 / #1159 unbuffered
memoryMsgChan were replaced with nil chan for both Topic and Channel
if mem-queue-size=0. This inadvertently made deferred publish never
work (with mem-queue-size=0), because some metadata is lost when
messages go through the "backend" disk queue, instead of immediately
going through the topic memory chan to the channels' deferred pqueues.

The motivation for fully disabling the memoryMsgQueue for
mem-queue-size=0 was to avoid excessively shuffling message order,
which would happen if some messages jump instantly through the
memory-queue while others take the longer way through the disk-queue.
So, only allow using the memory queue in this case if really needed,
for deferred messages, or for ephemeral topic or channel which just
lose all messages if they all go through the no-op backend queue.

Granted, mem-queue-size=0 never worked very well, in many respects.

nsqd/channel.go

@@ -82,9 +82,10 @@ func NewChannel(topicName string, channelName string, nsqd *NSQD,
 		clients:        make(map[int64]Consumer),
 		deleteCallback: deleteCallback,
 		nsqd:           nsqd,
+		ephemeral:      strings.HasSuffix(channelName, "#ephemeral"),
 	}
-	// create mem-queue only if size > 0 (do not use unbuffered chan)
-	if nsqd.getOpts().MemQueueSize > 0 {
+	// avoid mem-queue if size == 0 for more consistent ordering
+	if nsqd.getOpts().MemQueueSize > 0 || c.ephemeral {
 		c.memoryMsgChan = make(chan *Message, nsqd.getOpts().MemQueueSize)
 	}
 	if len(nsqd.getOpts().E2EProcessingLatencyPercentiles) > 0 {
@@ -96,8 +97,7 @@ func NewChannel(topicName string, channelName string, nsqd *NSQD,
 
 	c.initPQ()
 
-	if strings.HasSuffix(channelName, "#ephemeral") {
-		c.ephemeral = true
+	if c.ephemeral {
 		c.backend = newDummyBackendQueue()
 	} else {
 		dqLogf := func(level diskqueue.LogLevel, f string, args ...interface{}) {

nsqd/topic.go

@@ -46,7 +46,7 @@ func NewTopic(topicName string, nsqd *NSQD, deleteCallback func(*Topic)) *Topic
 	t := &Topic{
 		name:              topicName,
 		channelMap:        make(map[string]*Channel),
-		memoryMsgChan:     nil,
+		memoryMsgChan:     make(chan *Message, nsqd.getOpts().MemQueueSize),
 		startChan:         make(chan int, 1),
 		exitChan:          make(chan int),
 		channelUpdateChan: make(chan int),
@@ -56,10 +56,6 @@ func NewTopic(topicName string, nsqd *NSQD, deleteCallback func(*Topic)) *Topic
 		deleteCallback:    deleteCallback,
 		idFactory:         NewGUIDFactory(nsqd.getOpts().ID),
 	}
-	// create mem-queue only if size > 0 (do not use unbuffered chan)
-	if nsqd.getOpts().MemQueueSize > 0 {
-		t.memoryMsgChan = make(chan *Message, nsqd.getOpts().MemQueueSize)
-	}
 	if strings.HasSuffix(topicName, "#ephemeral") {
 		t.ephemeral = true
 		t.backend = newDummyBackendQueue()
@@ -222,18 +218,25 @@ func (t *Topic) PutMessages(msgs []*Message) error {
 }
 
 func (t *Topic) put(m *Message) error {
-	select {
-	case t.memoryMsgChan <- m:
-	default:
-		err := writeMessageToBackend(m, t.backend)
-		t.nsqd.SetHealth(err)
-		if err != nil {
-			t.nsqd.logf(LOG_ERROR,
-				"TOPIC(%s) ERROR: failed to write message to backend - %s",
-				t.name, err)
-			return err
+	// If mem-queue-size == 0, avoid memory chan, for more consistent ordering,
+	// but try to use memory chan for deferred messages (they lose deferred timer
+	// in backend queue) or if topic is ephemeral (there is no backend queue).
+	if cap(t.memoryMsgChan) > 0 || t.ephemeral || m.deferred != 0 {
+		select {
+		case t.memoryMsgChan <- m:
+			return nil
+		default:
+			break // write to backend
 		}
 	}
+	err := writeMessageToBackend(m, t.backend)
+	t.nsqd.SetHealth(err)
+	if err != nil {
+		t.nsqd.logf(LOG_ERROR,
+			"TOPIC(%s) ERROR: failed to write message to backend - %s",
+			t.name, err)
+		return err
+	}
 	return nil
 }