[Bug] Randomly error "incorrect protocol state/Operation Canceled" when concurrent Req/Resp

[toni-moreno]

related with #192 and #189 .

Everything ok while working with only 1 worker sending Request to the mangos Socket and waiting for its responses. When added more than one worker

In the master

func SendWaitReq(msg []byte) ([]byte, error) {
	var err error
	var resp []byte
	if err = sock.Send(msg); err != nil {
		return nil, err
	}
	if resp, err = sock.Recv(); err != nil {
		return nil, err
	}

	return resp, nil
}


const NumWorkers = 12

// Init begins workers processing jobs
func Init() {
	queue = NewQueue()
	//Beggin workers.
	for i := 0; i < NumWorkers; i++ {
		log.Infof("TRIGGER WORKER[%d] Init", i)

		go func(wid int) {
			for {
				item := queue.Pop()
				tid := item.(*TriggerID)
                                data=processSomeData(tid)
				resp,err:= SendWaitReq(data)
		                if err != nil {
			             log.Errorf("TRIGGER WORKER[%d] Error on get Queue info: %s", wid, err)
		                }
                                 doSomethingWithResp(resp)
			}
		}(i)
	}

}

In other goroutine someone is writing lots of TriggerID in the Internal Queue. ( queue.Push())

This is the output log randomly

time="2020-03-20 19:52:33" level=error msg="TRIGGER WORKER[10] Error on get Queue info: incorrect protocol state"
time="2020-03-20 19:52:33" level=error msg="TRIGGER WORKER[0] Error on get Queue info: operation canceled"

No data is returned....

[gdamore]

Does each worker either have it's own socket, or using a context? (see socket.OpenContext().)

I suspect you're trying to reuse the single socket across multiple workers, without using an intermediate context, and that is the problem.

[toni-moreno]

As you said, I'm working on the master with only one socket and multiple workers, I didn0t know anything about context when working with multiple workers.

I'm looking for examples on how to use socket.Context in my use case but I can not find any.

Could you please show me how should I use them in the below code ? or any link to watch for examples?

[toni-moreno]

Hello @gdamore we have opened a new context on each msg send/receive and it seems to work.

func SendWaitReq(msg []byte) ([]byte, error) {
	var err error
	var resp []byte
	sctx, err := sock.OpenContext()
	defer sctx.Close()
	if err != nil {
		log.Errorf("Can't open a socket context")
                return resp, err
	}
	if err = sctx.Send(msg); err != nil {
		return nil, err
	}
	if resp, err = sctx.Recv(); err != nil {
		return nil, err
	}

	return resp, nil
}

Thank you very much!!!!

[toni-moreno]

Hello @gdamore , a single question.
could you tell me what is better regarding resource usage , Create a new context per message or one per Worker ?

[gdamore]

You shouldn't have reopened this if it's not a bug. :-) Note that we have a mailing list and a discord chat room (https://discord.gg/nanomsg) where you can ask questions.

Having said that, you want a context per worker, each worker doing "recv a request, ... do work... send reply". This is the use case contexts were designed for.

Creating new contexts is very cheap, but there is no real reason to do it for each new request.