Is it bug: A non blocking socket operation could not be completed immediately?

[my20050804]

2016-01-20 12:15:17,289 System.Net.Sockets.SocketException (0x80004005): 无法立即完成一个非阻止性套接字操作。
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 NetMQ.Poller.PollTillCancelled()
at System.Threading.ThreadHelper.ThreadStart_Context(Object state)
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()

[somdoron]

It is usually a sign you are using the sockets from multiple threads. Can
you provide code that reproduce?
On Jan 20, 2016 8:16 AM, "my20050804" notifications@github.com wrote:

2016-01-20 12:15:17,289 SystemNetSocketsSocketException (0x80004005):
无法立即完成一个非阻止性套接字操作。
at SystemNetSocketsSocketReceive(Byte[] buffer, Int32 offset, Int32 size,
SocketFlags socketFlags)
at NetMQCoreMailboxTryRecv(Int32 timeout, Command& command)
at NetMQCoreSocketBaseProcessCommands(Int32 timeout, Boolean throttle)
at NetMQCoreSocketBaseGetSocketOption(ZmqSocketOption option)
at NetMQCoreUtilsSelectorSelect(SelectItem[] items, Int32 itemsCount,
Int64 timeout)
at NetMQPollerPollWhile(Func`1 condition)
at NetMQPollerPollTillCancelled()
at SystemThreadingThreadHelperThreadStart_Context(Object state)
at SystemThreadingExecutionContextRunInternal(ExecutionContext
executionContext, ContextCallback callback, Object state, Boolean
preserveSyncCtx)
at SystemThreadingExecutionContextRun(ExecutionContext executionContext,
ContextCallback callback, Object state, Boolean preserveSyncCtx)
at SystemThreadingExecutionContextRun(ExecutionContext executionContext,
ContextCallback callback, Object state)
at SystemThreadingThreadHelperThreadStart()

—
Reply to this email directly or view it on GitHub
#441.

[greendimka]

I have the same problem. But I do not use Router socket from multiple threads. I have a thread-safe queue which I fill with NetMqMessage objects. And I have a separate thread which picks up messages from that queue and sends them via single router socket. As soon as it is ~20 messages per second - Router socket crashes with SocketException code 10035.

[tomscharf]

I also have this exact problem. I re-wrote the code to ensure it is a single thread operation. I use incoming and outgoing synchronized queue's for all data transfer. The zmq object is not even visible to any other threads.

Curiously this bug occurs much more frequently when running in release mode vs debug mode.

Router / Dealer. This problem occurs in both v3 and v4. It would be useful if a complete example was provided that is known to be problem free. Running in x64 mode. Perhaps the synchronized queue is causing the problem?

System.Net.Sockets.SocketException was unhandled
Message: An unhandled exception of type 'System.Net.Sockets.SocketException' occurred in System.dll
Additional information: A non-blocking socket operation could not be completed immediately

[somdoron]

Can you make code that reproduce the issue?

[tomscharf]

Unfortunately that won't be easy. Communicating with a Linux host. Note that it must be coming from the poller since I can't even easily trap this exception and it crashes the entire app, which makes it even worse.

Here is a breakdown of the exception:

System.Net.Sockets.SocketException (0x80004005): A non-blocking socket operation could not be completed immediately
at System.Net.Sockets.Socket.Receive(Byte[] buffer, Int32 offset, Int32 size, SocketFlags socketFlags)
at System.Net.Sockets.Socket.Receive(Byte[] buffer)
at NetMQ.Core.Utils.Signaler.Recv()
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.NetMQPoller.Run()
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()

I've been trying to work around this problem, but I can't live with it. I'll probably have to use clrmq4.

[tomscharf]

Here is the core code if that helps

    //-------------------------------------------------
    //-------------------------------------------------
    // Main Messaging Loop
    //-------------------------------------------------
    //-------------------------------------------------
    private void WorkerZmqMain()
    {
      try
      {
        // Queue to GUI thread
        WriteToZmqQueue = Queue.Synchronized(WriteToZmqQueue_NoSync);
        ReadFromZmqQueue = Queue.Synchronized(ReadFromZmqQueue_NoSync);

        // Setup the protocol handler
        ProtocolHandler = new OeiDecode(DevInterfaceQ);

        // New NetMQ connection
        DevInterfaceQ.DebugLog.Write(DxLogLevel.Info, "StartNetMq - Device: " + zClass.szName + "(" + zClass.szNameIp + ")");

        zmq_client = new DealerSocket();
        if (zmq_client == null)
        {
          DevInterfaceQ.DebugLog.Write(DxLogLevel.Error, "StartNetMq Failed - Device: " + zClass.szName + "(" + zClass.szNameIp + ")");
        }
        else
        {
          // zClass.client.OptionzClass.Identity = 
          zClass.DeviceIsValid = true;

          zmq_client.Connect($"tcp://" + zClass.szNameIp + ":5555");
          zmq_client.ReceiveReady += Client_ReceiveReady;
          zmq_poller = new NetMQPoller { zmq_client };

          // This thread block here
          zmq_poller.RunAsync();
        }

        // Main execution loop
        while(true)
        {
          // Look at outgoing queue for xmt messages
          if (HandleZmqWrQueue() == false)
          {
            // Behave nicely
            Thread.Sleep(10);
          }

          if (bShutdown == true)
          {
            CloseAfterThreadExit();
            bShutdownComplete = true;
            break;
          }
        }
      }
      catch (Exception exc)
      {
        DevInterfaceQ.DebugLog.Write(DxLogLevel.Error, "StartNetMq Exception - Device: " + zClass.szName + "(" + zClass.szNameIp + ") " + " Error: " + exc.Message);
        return;
      }
    }

[somdoron]

What does HandleZmqWrQueue do?
Can you include the code of ReceiveReady?

How do you send on the queue?

Poller.RunAsync create a new thread, the dealer is belong to the poller and cannot be used out of the poller thread (ReceiveReady), I suspect you are using the dealer to send some where in that class which is not the poller thread.

To fix it you can use NetMQQueue, add it to the poller and send from the receive ready of the netmqqueue. It is safe to send from multiple threads

[somdoron]

Safw to send from myltiples threads with netmqqueue only.

[tomscharf]

I wondered about that. I figured since it is known to be illegal to work from different threads, one wouldn't have an API call that sets up that exact situation, ha ha. I thought an in thread timer was being created, but I didn't look at the source code.

I think the documentation should make this more clear, or I would remove the Async call completely because it's just asking for trouble. Maybe I missed it.

So I think I can just remove the Async call, and poll myself in the main loop and the problem will be resolved?

[tomscharf]

Possibly rename the call from RunAsync() to RunAsyncButRememberThatYouCanOnlyTransmitFromThePollerThread(), ha ha.

[somdoron]

So the call will be blocking, you want to use NetMQQueue instead of the
writeToZmqQueue. You can add it to the poller, use it ReiveReady to know
when something is ready on the queue and send it on the dealer.

On Thu, Oct 27, 2016, 19:10 Tom Scharf notifications@github.com wrote:

I wondered about that. I figured since it is known to be illegal to work
from different threads, one wouldn't have an API call that sets up that
exact situation, ha ha. I thought an in thread timer was being created, but
I didn't look at the source code.

I think the documentation should make this more clear, or I would remove
the Async call completely because it's just asking for trouble. Maybe I
missed it.

So I think I can just remove the Async call, and poll myself in the main
loop and the problem will be resolved?

—
You are receiving this because you modified the open/close state.
Reply to this email directly, view it on GitHub
#441 (comment), or mute
the thread
https://github.com/notifications/unsubscribe-auth/AClv9jzwBiuomyydqxTg7kBz4Q-xXz27ks5q4M0KgaJpZM4HIWjx
.

[tomscharf]

Right, it was the blocking that led me to the magical RunAsync originally. I think all the pieces are there in the documentation, but what was missing for me was a simple end to end code sample that implements both xmt and rcv without blocking and doesn't break the rules. That would be helpful.

Thanks for the prompt response, it was very helpful. I will go fix it.

[tomscharf]

Here is a simple example that I believe works:

    //-------------------------------------------------
    //-------------------------------------------------
    // Receive Incoming Zmq Message
    //-------------------------------------------------
    //-------------------------------------------------
    private void IncomingData(object sender, NetMQSocketEventArgs e)
    {
      // This receive operation occurs in the same thread that transmits (OutgoingData)
      byte[] bytes = e.Socket.ReceiveFrameBytes();

      // Process the message ...
    }

    //-------------------------------------------------
    //-------------------------------------------------
    // Transmit Outgoing Zmq Data
    //-------------------------------------------------
    //-------------------------------------------------
    private void OutgoingData(byte[] BytesToSendOut)
    {
      // This transmit operation occurs in the same thread that receives (IncomingData)
      net_mq.SendFrame(BytesToSendOut);
    }

    //-------------------------------------------------
    //-------------------------------------------------
    // Main Messaging Loop
    //-------------------------------------------------
    //-------------------------------------------------
    private void WorkerZmqMain()
    {
      DealerSocket net_mq = new DealerSocket();
      net_mq.Connect($"tcp://192.168.1.100:5555");

      using (var OutgoingQueue = new NetMQQueue<byte[]>())
      {
        using (var AsyncPoller = new NetMQPoller { net_mq, OutgoingQueue })
        {
          // Setup the transmit and receive routines
          net_mq.ReceiveReady += IncomingData;
          OutgoingQueue.ReceiveReady += (sender, args) => OutgoingData(OutgoingQueue.Dequeue());

          AsyncPoller.RunAsync();

          // Main execution loop
          byte[] OutgoingBytes = new byte[10];
          OutgoingQueue.Enqueue(OutgoingBytes);

          // ...
        }
      }
    }