Exception: "A non-blocking socket operation could not be completed immediately"

[Dave Nay]

I have been trying to track down the cause of an exception that I am seeing randomly in my NetMQ server application.

Windows 7 x64

.NET 4.5 target

NetMQ version 3.3.0.13-alpha622 (also happens with 3.3.0.11 and all versions in between)

"An exception occured on 2015.08.06 07:18:13:384

Message: A non-blocking socket operation could not be completed immediately

Source: System

Version: Manual Build

StackTrace:

   at System.Net.Sockets.Socket.Receive(Byte[] buffer, Int32 offset, Int32 size, SocketFlags socketFlags)

   at NetMQ.Core.Mailbox.TryRecv(Int32 timeout, Command& command)

   at NetMQ.Core.SocketBase.ProcessCommands(Int32 timeout, Boolean throttle)

   at NetMQ.Core.SocketBase.GetSocketOption(ZmqSocketOption option)

   at NetMQ.Core.Utils.Selector.Select(SelectItem[] items, Int32 itemsCount, Int64 timeout)

   at NetMQ.Poller.PollWhile(Func`1 condition)

   at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)

   at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)

   at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state)

   at System.Threading.ThreadHelper.ThreadStart()

"

This exception is being randomly thrown from a NetMQ Poller object. The poller is polling on a single router socket. Currently, the code uses the new Poller API (PollTillCancelledNonBlocking()) but previously I was using my own background thread to run the poller.

Looking through the NetMQ code, it would appear that this is being caused by some internal communications and is not obviously related to my actual client/server communications.

I am asking for any assistance in helping me identify what I might be doing wrong in my user code, or perhaps I have identified a NetMQ bug. Either way, this crash is serious to me since we are in final testing prior to release and this is definitely a stop work issue.

Unfortunately, I have not been able to isolate a test case.

Thank you very much,

Dave

[Drew Noakes]

Without a minimal test case it's hard to know what to suggest. You could share your messaging code, but unless it independently reproduces the bug then its not likely to reveal much.

The stack trace includes some code (TryRecv) that was changed in the latest build, but you say it works in older builds too, so it can't be that.

Does the error occur on multiple machines?

> This exception is being randomly thrown from a NetMQ Poller object. The poller is polling on a single router socket. Currently, the code uses the new Poller API (PollTillCancelledNonBlocking()) but previously I was using my own background thread to run the poller.

If you run the poller with your own thread does the error go away?

--
You received this message because you are subscribed to the Google Groups "netmq-dev" group.
To unsubscribe from this group and stop receiving emails from it, send an email to netmq-dev+...@googlegroups.com.
To post to this group, send email to netm...@googlegroups.com.
Visit this group at http://groups.google.com/group/netmq-dev.
For more options, visit https://groups.google.com/d/optout.

[BDeus]

I have the same issue with Poller when sending thousands of messages.

I join a sln of the issue if you have some time.

[BDeus]

I have the same issue with Poller.

I join some code with it, if you have some time to lookup.

Thanks

Le jeudi 6 août 2015 15:54:51 UTC+2, Drew Noakes a écrit :

[Johan Lindberg]

We are getting the same issue on Socket.Send in our load tests. 
~1000 clients, a heartbeat is sent from the server to all clients with a 3 second interval. Doesn't take long before the error occurs.

Version : 3.3.1.15040

Following code.

 Task.Factory.StartNew(() =>

            {

                Thread.CurrentThread.Name = "NetMQ - Publish/Subscribe";

                using (var server = _Context.CreateXPublisherSocket())

                {

                    server.Options.SendBuffer = ZMQConstants.MegaByte * 100; // 100 megabyte

                    server.Options.SendHighWatermark = 10000;

                    server.Options.ReceiveHighWatermark = 10000;

                    server.Options.Linger = TimeSpan.FromSeconds(5);

                    server.Bind(Address);

                    server.ReceiveReady += server_ReceiveReady;

                    Task.Factory.StartNew(() =>

                    {

                        Poller poller = new Poller();

                        poller.AddSocket(server);

                        poller.PollTillCancelled();

                    }).ContinueWith(task => { ConnectionManager.Log(task.Exception); }, TaskContinuationOptions.OnlyOnFaulted);

                    while (true)

                    {

                        //MeasgeQueue is just a System.Collections.Concurrent.BlockingCollection<T> , in the tests a heartbeat is pushed in every third second.

                        var m = MessageQueue.Dequeue();

                        server.SendMore(m.Key);

                        server.Send(m.AsBytes); // <- Here i get the "A non-blocking socket operation could not be completed immediately"

                    }

                }

            });

[Doron Somech]

you are using the server sockets from two different threads which is not allowed. Once from the main thread to send and one from the Task to receive.

[Johan Lindberg]

Thanks,

I changed it to the following,

 Task.Factory.StartNew(() =>
            {
                Thread.CurrentThread.Name = "NetMQ - Publish/Subscribe";


                using (var server = _Context.CreateXPublisherSocket())

                using (var poller = new Poller())

                {
                    server.Options.SendBuffer = ZMQConstants.MegaByte * 100; // 100 megabyte
                    server.Options.SendHighWatermark = 10000;
                    server.Options.ReceiveHighWatermark = 10000;
                    server.Options.Linger = TimeSpan.FromSeconds(5);
                    server.Bind(Address);
                    server.ReceiveReady += server_ReceiveReady;

                    poller.AddSocket(server);
                    poller.PollTillCancelledNonBlocking();
                    while (true)
                    {

                        var m = MessageQueue.Dequeue();
                        server.SendMore(m.Key);
                        server.Send(m.AsBytes);
                    }
                }

            });

Is this correct implemented? If i check the current thread in server_ReceiveReady() it's still the NetMQPollerThread doing the e.Socket.ReceiveFrameString(). 

Or is it the AddSocket that is important?