How to check if there is a message to fetch from server without actualy fetching it

[rkintzi]

I have pool of goroutines to handle messages from NATS. I only want to fetch a message from NATS is there is a free goroutine in the pool.

In the code this may look like (pseudocode):

for {
    worker := waitForWorker() // blocks if there is no free worker in pool
    msgs := natsSub.Fetch(1)
    worker.Handle(msgs[0]) // passes message to worker's goroutine and returns immediately 
}

This works if there is only one subscription. If there is more than one I need to select subscription that has undelivered message:

for {
    worker := waitForWorker() // blocks is there is no free worker in pool
    natsSub := selectNatsSubWithAwaitingMessage() // wait until there is subscription with undelivered message
    msgs := natsSub.Fetch(1)
    worker.Handle(msgs[0]) // passes message to worker's goroutine and returns immediately 
}

So the question is how to implement selectNatsSubWithAwaitingMessage() function.

[derekcollison]

I would suggest using a chan subscriber, then you can select on them all.

https://github.com/nats-io/nats.go/blob/main/test/sub_test.go#L799

[rkintzi]

Hi Derek,

I unmarked your answer because ChanSubscribe can only be used with push based consumers and I prefer to use pull based consumers. I wasn't precise enough.

[derekcollison]

We could expose the internal channel we use potentially. /cc @wallyqs

[rkintzi]

Do you mean mch in Subscription struct?

[derekcollison]

Possibly but would need to look into it more. The approach @ripienaar was mentioning might be worth trying.

[rkintzi]

I don’t think that exposing the channel would help solve the problem, as, in my opinion, it lies not in the library API but in how the client (the library) interacts with the server. Before I explain what I mean, let me describe currently possible solutions and their drawbacks, so we can better understand why we need to change the library (which, I believe we can do without affecting existing clients). I'll do this in a separate message.

[rkintzi]

To clarify what I want to achieve I prepared the simplest worker pool implementation I can come up with (see the bottom of this message; please note, that although it can be used for simple tests, I omitted error handling and graceful shutdown; the real-live implementation would dispatch messages to handlers based on messages’ subjects).

Like I said, this solution works well if it is ok to create a separate pool for each subscription. And while in many cases this is in fact ok, sometimes that can lead to non optimal resource utilization.

I dug a bit in nats internals and I think the fetching from multiple subscription could be implemented as follows (the example assumes two subscription, two available workers (hence batch:2 and UNSUB 1 2) and 1s timeout):

SUB _INBOX.hZsmVxmuc3ClIO9cxszjfW.CO3fJmUq  1
UNSUB 1 2
PUB $JS.API.CONSUMER.MSG.NEXT.teststream.testcons1 _INBOX.hZsmVxmuc3ClIO9cxszjfW.CO3fJmUq 34
{"batch": 2, "expires":1000000000}
PUB $JS.API.CONSUMER.MSG.NEXT.teststream.testcons2 _INBOX.hZsmVxmuc3ClIO9cxszjfW.CO3fJmUq 34
{"batch": 2, "expires":1000000000}

I haven’t yet implemented this in Go, but from what I have observed using telnet I can tell that this works as expected (I am mostly worried about pull requests being accumulated on the server but this seems not to be happening).

Can anyone confirm if this is the right approach?

Worker pool implementation

package main

import (
	"time"

	"github.com/nats-io/nats.go"
)

type (
	Worker chan *nats.Msg
	Pool   chan Worker
)

func (w Worker) run(p Pool) {
	p <- w
	for msg := range w {
		w.handleMessage(msg)
		p <- w // return worker to the pool once message is handled
	}
}

func (w Worker) handleMessage(m *nats.Msg) {
	m.Ack() // real worker would select a handler based on m.Subject
}

func (p Pool) run(sub *nats.Subscription, nWorkers int) {
	for ; nWorkers > 0; nWorkers-- {
		go make(Worker).run(p)
	}
	var ws []Worker
	var i int
	for {
		select {
		case w := <-p:
			ws = ws[:0]
			ws = append(ws, w)
		L:
			for { // collect free workers so we can fetch batch of messages
				select {
				case w := <-p:
					ws = append(ws, w)
				default:
					break L
				}
			}
			msgs, _ := sub.Fetch(len(ws), nats.MaxWait(time.Second))
			for i = 0; i < len(msgs); i++ {
				ws[i] <- msgs[i]
			}
			for ; i < len(ws); i++ { // return unused workes to the pool
				p <- ws[i]
			}
		}
	}
}

func main() {
	n := 10
	c, _ := nats.Connect(nats.DefaultURL)
	js, _ := c.JetStream()
	sub, _ := js.PullSubscribe("testsubject.cons2.>", "", nats.Bind("teststream", "testcons2"))
	p := make(Pool, n)
	p.run(sub, n)
}

[derekcollison]

Is the work cpu or io bound? How long does it take to process? Are you trying to optimize using all cores and minimizing latency?

[rkintzi]

The main goal is to distribute CPU-intensive tasks across multiple machines and multiple cores. I run multiple instances of the same consumer to distribute workload across multiple nodes in a cluster. Nodes should pick tasks from a stream only when they have the resources to process them. A pool of N workers (where N is more or less equal to the number of CPU cores) solves the problem.

The other use case is to limit the number of simultaneous execution of a given task. Consider an API to which you can send no more than M simultaneous requests. You can run N instances of a consumer, each with M/N workers in the pool.

I also think that the worker pool could automatically scale up and down the number of goroutines and maybe even spawn new nodes in the cluster when it discovers that resources are over-utilized.

[ripienaar]

I do this by setting a stream to reject new and max per subject equal to the maximum concurrent jobs.

To start a job you need to be able to put a message into the stream, the pub ack tells you the sequence. When you are done you delete your sequence so the slot is free for the next worker.

To prevent workers crashing leaving stale reservations I set a stream TTL bigger than the longest time my work will realistically take (maybe in sync with what kind of timeout contexts you make, assuming go).

This way you can coordinate a distributed set of workers against a hard cap of how many may hit a resource constraint central component.

[rkintzi]

Thanks @ripienaar for sharing an interesting idea. From what I understand, you suggest that before the worker picks a task from the queue, it must reserve a slot in a dedicated stream, so there is always a limited number of active workers (equal to the number of available slots). That could be a good (maybe even better) solution for the second problem I described.

I don’t see how this can help with the main problem, where tasks should be distributed across nodes in the cluster and CPU cores for the fastest execution possible.

[ripienaar]

A worker will only obtain a slot when its ready and able to do work, so assuming there is more work than available workers - there will always be busy workers up to your cap.

This means the work will automatically distribute to the most appropriate place - where a worker is available. No worker will get work he cant do and all your available cores will be busy doing one thing each (capped by the limiter)

This is also compatible with scaling up and down since you can stream edit the limiter stream to whatever max per subject you desire at any given time, do that in tandem with starting/stopping workers and you will have all the controls you seem to desire here.

The model performs well and has proven to be very stable for my use, here I run 1000 jobs with a max concurrency of 100 and the job takes 50ms to complete:

=== RUN   TestJsGovernor
2023/01/25 09:42:06 ran: 1000, max concurrent: 100
--- PASS: TestJsGovernor (0.74s)

(time includes starting the jetstream server, setting up the stream and creating all the 1000 worker routines)

[rkintzi]

I am sure that your approach do the job. But I don't see how it solves my problem which is not how cap the number of concurrent executions (this is easy - you simply avoid spawning too many workers and worker pool works great here). Instead the problem is how to efficiently fetch from multiple subscriptions (aka consumers). I elaborate more on the problem in the message below.

[rkintzi]

In this message I elaborate on the original problem of how to efficiently fetch messages from more than one subscription. I present possible solutions and discuss their drawbacks. I build on the example of a worker pool from one of my previous messages. @derekcollison

The problem with fetching from multiple subscriptions is that we don't know upfront which of them has messages to be consumed. We need to fetch from subscriptions to find out. We can do this concurrently (as in the first example implementation below) or sequentially (second example). Both approaches have their drawbacks which I point out below.

As a reminder, the worker pool has a run method responsible for spawning workers and fetching tasks for them:

func (p Pool) run(sub *nats.Subscription, nWorkers int) {
    // ...
    sub.Fetch(nIdleWorkers, nats.MaxWait(timeout))
   //
}

Lets modify it so that it operates not on a single subscription, but on a group of subscriptions (this is the only modification needed when it comes to worker pool implementation):

func (p Pool) run(grp *SubscriptionGroup, nWorkers int) {
    // ...
    grp.Fetch(nIdleWorkers, nats.MaxWait(timeout))
   //
}

Note that, when the pool calls the Fetch method it passes it the number of idle workers as a batch argument. Thus the method should fetch at most batch messages. One of possible implementation is as follows (cancellation and error handling omitted, for clarity):

type SubscriptionGroup struct {
	subs []*nats.Subscription
}

func (g *SubscriptionGroup) Fetch(batch int, opts ...nats.PullOpt) ([]*nats.Msg, error) {
	ch := make(chan []*nats.Msg, len(g.subs)*batch)
	for _, s := range g.subs {
		go func(s *nats.Subscription) {
			msgs, _ := s.Fetch(batch, opts...)
			ch <- msgs
		}(s)
	}

	msgs := make([]*nats.Msg, 0, len(g.subs)*batch)
	for i := 0; i < len(g.subs); i++ {
		ms := <-ch
		msgs = append(msgs, ms...)
	}
	if len(msgs) <= batch {
		return msgs, nil
	}
	for i := batch; i < len(msgs); i++ { // we push back messages for which we have no workers
		msgs[i].Nak()
	}
	return msgs[:batch], nil
}

In the above example we fetch messages concurrently, which may lead to the situation where we have fetched more messages than we can process, so we need to push them back (by Nak-ing them). This works correctly but puts unnecessary load on a network and nats server.

Bellow is the alternative implementation where subscriptions are fetched sequentially:

type SubscriptionGroup struct {
	next int
	subs []*nats.Subscription
}

func (g *SubscriptionGroup) Fetch(batch int, opts ...nats.PullOpt) ([]*nats.Msg, error) {
	for {
		ms, _ := g.subs[g.next].Fetch(batch, opts...)
		g.next = (g.next + 1) % len(g.subs)
		if len(ms) > 0 {
			return ms, nil
		}
	}
}

The problem here is that in the worst scenario we need to wait n * t for messages, where n = len(g.subs) and t is timeout passed to Fetch method (g.Fetch(batch, nats.MaxWait(t))). We waste CPU cycles.

[ripienaar]

I think the model where the pool fetches for the pool size is the root of the problem. Each worker should do its own fetching when its ready and wait for work to be available.

Then you never need to check if there are messages in the stream in a pool manager or deal with a pool manager somehow having to know how many in the pool are free at any one time. And you have no messages sitting idle spending their max ack time budget while in the pool manager channel.

When a worker is free it pulls for itself (with optional limiter awareness, which in this case you can also do with max outstanding pulls setting for the consumer).

[rkintzi]

So the worker needs one message. It can call the grp.Fetch(1) - how this solves the problem?

[ripienaar]

The worker has the consumer subscription (each worker their own against the same consumer), and when its ready to work it fetches a work item.

You then do not need to fetch any multiple messages or get caught up when there are fewer stalling the entire group etc, just have each worker fetch when its ready.

I suggest you write some code to try it.

[rkintzi]

Let's be more concrete. I have 5 servers with 64 CPU cores each. They should be utilized as efficiently as possible. That's why I want to spawn 64 goroutines on each server and provide them with tasks to do. I have around 10 different tasks that I want to execute to these servers.

Each task is sent as a message to the NATS stream with the subject TASKS.task_type1 (stream catches everything that matches TASKS.>), In my current implementation I have only one consumer that catches all tasks. So I have only one subscription on each server (each server subscribes using the same consumer). I have zero problems with this implementation that is based on the worker pool example above.

Like I said, the above works perfectly fine, but if I could have separate consumers for different task kinds, it would help me a bit with scaling when the number of tasks suddenly grows. Without separate consumers I can not dedicate a separate server just for one specific task. But with separate consumers, servers (general purpose servers, not dedicated ones) need to maintain more than one subscription and I end up with the problem described above.

From what you are writing in your solution there is only one consumer. In such a case I have no problem and I don't see how I could benefit from your solution. I also don't see how your solution could help me with my real problem.

But you suggested me to write some code. Challenge accepted.

I see two possible implementations. First one goes like this:

type worker struct {
   sub *nats.Subscription
   slots *Slots
}

func (w *worker) run() {
   for {
        w.slots.Reserve()
        msg:=w.sub.Fetch(1, nats.MaxWait(time.Second))
        if msg != nil {
            w.doTheJob(msg)
       }
       w.slots.Free()

   }
}

Let’s assume that w.slots.Reserve() blocks when there is no available slot so there is no active waiting. Let’s also assume that we have two workers each of which has its own subscription/consumer (but they share slots).

Consider this. First worker reserves the slot and then calls Fetch() only to find out (after a second) that there is no message available. In the meantime the second worker waits for a slot, while it could already execute its task (assuming the message is available in its queue). This wastes CPU cycles (as in my second example above).

Alternative implementation:

func (w *worker) run() {
   for {
        msg:=w.sub.Fetch(1, nats.MaxWait(time.Second))
        if msg != nil {
            w.slots.Reserve()
            w.doTheJob(msg)
       }
       w.slots.Free()

   }
}

Here workers concurrently fetch messages from their subscriptions. When a message arrives, the worker reserves a spot and starts execution. Let’s say both workers received a message from their consumers so they both try to reserve the spot. But there is only one idle CPU core available on this server, so only one worker succeeds and executes.

What if there is another idle CPU core on another server? The worker there can not fetch the message from the consumer, because it is held by the waiting worker on the first server. Of course you may argue that the waiting worker should push back messages as it is unable to execute it, but how does this differ from the first solution I presented above?

I appreciate your engagement. I am sure your solution works great in your setup. But mine seems different. Maybe you can share some code so I can evaluate it before I invest time to pursue a solution that might not be the best suited for my needs. Otherwise I prefer to focus on discussing a solution I have in mind and maybe even on contributing it to the library.

Once again I want to thank you for sharing your idea. I have never thought before that I can use a stream as a sort of distributed semaphore. I learned from you.

[ripienaar]

What if there is another idle CPU core on another server? The worker there can not fetch the message from the consumer, because it is held by the waiting worker on the first server. Of course you may argue that the waiting worker should push back messages as it is unable to execute it, but how does this differ from the first solution I presented above?

It could indeed NAK the message for redelivery to another worker.

It's inherently the case that if you want to cap concurrent workers some will idle, I can think of solutions around them also holding messages (you could cap around how many pull requests the servers will service concurrently) but I am not sure this forum is really the most productive.

[rkintzi]

Intro

With this message I start a discussion on how fetching from multiple subscriptions/consumers can be implemented in the library. First I propose two possible APIs extensions that will enable users to request new behavior. Then I describe how pulling from a consumer actually works and what changes are required to implement the discussed functionality. I focus on the protocol and messages exchanged between the library and the server, and on showing how we can use what the server and protocol already provide to achieve the goal.

On required API change

We clearly need a notion of a subscription group in the API. Otherwise how do we enable the user to express the will of fetching from multiple subscriptions? I see two possible approaches:

Option 1 - new SubOpt

We can provide a new SubOpt that when passed to js.PullSubscribe makes the new subscription part of a group, eg:

sub1, _ := js.PullSubscribe(subj1, dur1, nats.SubGroup(“group1”))
sub2, _ := js.PullSubscribe(subj2, dur2, nats.SubGroup(“group1”))

From now on every call to a Fetch method on any of the subscriptions fetches messages from both of them.

The drawback here is that users may be a bit confused after receiving messages from sub2 after a call to sub1.Fetch(). On the other hand, this may happen only after a programer conscious decision to use the new option.

The huge pros is that this seems to be the simplest possible API change. But, of course, there is more to unpack, here. For example we need to decide what sub1.Drain() or sub2.AutoUnsubscribe() should do or if these calls should be allowed at all (fortunately ErrTypeSubscription is already part of the library interface).

Option 2 - new subscription type.

We can introduce new JetsonContext method, eg:

grp, _ := js.GroupSubscription()
sub1, _ := grp.PullSubscribe(subj1, dur1)
sub2, _ := grp.PullSubscribe(subj2, dur2)
grp.Fetch(n)

In my opinion, while this seems less confusing from a user perspective, there are more corner cases here. For example, what should happen when grp.Fetch() is called simultaneously with sub1.Fetch().

Current implementation

Moving to implementation details. Let’s remember what happens when a new pull subscription is created and the Fetch method called on it. Please correct me if I mess something up or I miss some details.

1. When the PullSubscribe() method is called, a new subscription with a random subject is created. On a protocol level this is done by sending following message to the server:

   SUB _INBOX.hZsmVxmuc3ClIO9cxszjfW.CO3fJmUq  1
   

2. When a Fetch() is called, the library checks the local cache (mch channel associated with the subscription) and if there are any messages they are collected. From where these messages may come from, I am not entirely sure (besides that they are artifacts from the previous Fetch() call).

3. After collecting messages from the cache library sends the server a message similar to:

   PUB $JS.API.CONSUMER.MSG.NEXT.teststream.testcons1 _INBOX.hZsmVxmuc3ClIO9cxszjfW.CO3fJmUq 28
   {"batch": 2, "no_wait":true}
   

   This asks the server to send up to batch messages from a specified consumer to the subscription created in pt.1 above. The library asks for batch - N messages, where bach is the number the user requested and N is the number of messages collected in the previous step. The server is requested not to wait for messages. Instead if there are no messages the server responds with “NATS/1.0 404 No Messages” message.

   If the server responds with messages they are appended to messages collected in point 2. If there are any messages collected so far they are returned from the Fetch method to the caller.

4. When on the other hand there are no messages, the library sends it another request, this time to wait for messages for a specified time:

   PUB $JS.API.CONSUMER.MSG.NEXT.teststream.testcons1 _INBOX.hZsmVxmuc3ClIO9cxszjfW.CO3fJmUq 34
   {"batch": 2, "expires":1000000000}
   

   The library collects messages from the server until the timeout expires (as signaled by the server) and returns them to the client. If there were no messages, this step repeats.

Remarks:

5. Note that the server always sends messages to the subscription created in point 1 above.
6. I wonder what has to happen for the messages to appear in the cache (in the channel) at the beginning of the Fetch call. Anyway, the fact that they may, indicates that the library client code should process messages asynchronously. Otherwise some messages may stay in the library cache and expire there. Thus the worker pool pattern I presented in one of my previous messages seems well suited for handling pull subscriptions (on the other hand, examples in the documentation suggest otherwise).

Suggested changes

Let’s now discuss how to change the above to allow pulling from more than one subscriptions/consumer. Note that following changes may apply only to the group subscriptions and their members or - to avoid branching the code - to all subscriptions, as they seem not possible to observe from a library user perspective (this should be further investigated).

1. For reasons that will become apparent in a moment, we should avoid sending the SUB message in the PullSubscribe() method. We should postpone it until the Fetch method is called. Because this method returns a subscription object, I suggest creating such objects without contacting a server. We should somehow mark such "unsent" subscriptions so we can later distinguish them. This applies only to subscriptions associated with pull subscribers. Other subscriptions should be created as usual.
2. The Fetch() method should collect messages from the cache (like in current implementation).
3. Then the method should send a SUB message to the server.
4. For every subscription in a group it should sequentially send a no_wait request as described in point 3 in the above list. It should stop after receiving first messages or when all subscriptions return 404. If this ends up with messages they should be returned to the caller.
5. Otherwise, the method should send the server UNSUB 0 N message (where N is the size of the batch requested by the user + the number of messages exchanged since SUB message). Then it should send requests for messages to all the consumers in the group this time with the “expires'' parameter (as in point 5 above). Requests can be sent sequentially without waiting for responses. Thanks to the UNSUB 0 N trick, requests will expire after receiving the number of messages requested by the user or after specified timeout. This trick is a reason why we should postpone sending a SUB message until the call to the Fetch method.
6. Method should process received messages as in the pt.4 of the first list.

Remarks:

7. Note that the UNSUB trick is suggested by ADR-13.

The above lacks lots of details (that I have intentionally skipped or I am unaware of). Nevertheless I believe it's a good start for further discussion. Please, share your thoughts and remarks.

My plan is to start prototyping a solution. But first I want to hear from the community if such changes have a chance to be accepted. I will wait a day or two for your comments.

@derekcollison @ripienaar - pls invite other community members so we can have valuable discussion.

[jnmoyne]

The ability to have more than one subject filter in a consumer is coming soon, once you have that then you can use just one stream that captures all the subjects and can create durable pull consumers for each 'sub group'.

[derekcollison]

Why not flow controlled push with the callback putting all messages in a single channel?

[rkintzi]

Maybe I am missing something, but for me the push consumers never worked as efficiently as pull consumers. The only way I achieve the same performance with push consumers is to set MaxAckPending to the number of workers (essentially the number of CPU cores). What I observe is that the server pushes this number (MaxAckPending) of messages across all instances of the consumer. When this number is greater the number of workers then messages are buffered in the library cache where they expire and then are then redelivered.

Here are situations where push consumers are hard to be configured properly:

1. The number of CPU cores is not always fixed. Sometimes servers are started when the task queue grows and shut down when it shrinks. Then MaxAckPending has to be adjusted accordingly which is possible but complicates the solution.

2. When the new server joins the cluster it sometimes needs to wait for messages buffered in other instances to expire. This is a rare case as most often there are messages waiting in the queue.

3. Servers are not all the same. Some have more CPU cores than others. Push consumers don't know that so they can't properly distribute messages to servers. Some servers will receive more messages than they can process at the time.

4. Also, some servers have more performant CPUs than others. This is not a problem if servers pull messages, but push consumers have no clue where to push more.

5. Things get even worse when there are multiple consumers, as MaxAckPending is a parameter of a consumer not a stream.

To me it seems that when push consumers are in use it is impossible to avoid redelivered messages. This impacts overall system performance and adds burden to the network and nats server. But I have to admit that I have no great experience with push consumers, as I never had time to tweak them to work as efficiently as pull consumers. It might be that with proper configuration negative effects are neglectable.

Pull consumers have no such issues. They are easier to configure and they are even advised in the documentation when horizontal scalability is important.

Let me clarify one more thing. I have absolutely no pressure to develop a described solution. The system I built (with the team of great engineers) works really well. To the extent that I decided that my role in the project has ended. I thought that before I jump into the next one I can give back to the community and implement the feature I was missing.