c# - 应该调用 C# .BeginReceive() 多少次以接收发送的 3 个 block ，即通过 TCP 对 Linux C write() 的 3 次调用？

Question

通过 TCP 对 Linux 'C' write() 的 3 次调用发送的 3 个数据 block ，通过 Windows C# .BeginReceive() 接收相同的 3 个数据 block ，或者一个连续的数据 block ，或者不管有多少已经被接收在调用 .BeginReceived 时收到？

Linux 上的“C”应用程序通过 3 次 write() 调用通过 TCP 连接将消息发送到 Windows C# 应用程序，该应用程序使用 BeginReceive() 进行接收。

是否需要调用 BeginReceive() 三次，以接收 write() 发送的三个 block 中的每一个？还是 BeginReceive() 接收到的大小等于调用 BeginReceive() 时 Windows 接收到的大小？这可能是 3 个 writes() 发送的所有字节，也可能是部分字节，因此应该调用 .BeginReceive() 直到收到所有字节？

Linux C 应用程序在嵌入式 TI ARM 上运行，Windows C# 应用程序在同一个盒子内运行单板计算机。 ARM 具有与 SBC 的直接以太网连接。

ARM 和 SBC 之间的通信有时无法在启动时启动，我正在对源代码进行逆向工程以检查是否存在不良设计。

ARM端为TCP监听器，Windows客户端发起TCP连接。

使用 MontaVista(R) Linux(R) 专业版 5.0.0 (0702774) 和 Windows-7 Visual-Studio 2010 Visual-C#。

这里是ARM发送软件，Windows接收软件......................................................................

LINX 'C'

        char msgBuffer[64];
        sprintf(msgBuffer, START_MSG_ENVELOPE, msgId++, ack);
        write(connection->fd, msgBuffer, strlen(msgBuffer));
        write(connection->fd, msg, strlen(msg));
        write(connection->fd, END_MSG_ENVELOPE, strlen(END_MSG_ENVELOPE));

这是它的 WINDOWS C# 端...................................... ……

        private static void makeConnection(Socket clientSocket, int iPortNo)
    {
        TimeoutObject.Reset();
        socketexception = null;

        IPAddress ip;
        //create the end point 
        IPEndPoint ipEndPoint;

        ip = IPAddress.Parse(ipAddress);

        try
        {
            ipEndPoint = new IPEndPoint(ip, iPortNo);

            //connect to the remote host...
            clientSocket.BeginConnect(ip, iPortNo, new AsyncCallback(CallBackMethod), clientSocket);

            if (TimeoutObject.WaitOne(5 * 1000, false))    //5 secs connection timeout
            {
                if (!IsConnectionSuccessful)
                {
                    string msg = VNResourceManager.Instance.GetString(VNMessages.DAM_NOT_FOUND);
                    if (socketexception != null)
                        msg += ": " + socketexception.Message;
                    throw new Exception(msg);
                }
            }
            else
            {
                clientSocket.Close();
                throw new TimeoutException(VNResourceManager.Instance.GetString(VNMessages.CONNECTION_TO_DAM_TIMED_OUT));
            }
            //watch for data ( asynchronously )...
            WaitForData();
        }
        catch (SocketException e)
        {
            socketexception = e;
            throw;
        }
    }

    private static void CallBackMethod(IAsyncResult asyncresult)
    {
        try
        {
            IsConnectionSuccessful = false;
            Socket socket = asyncresult.AsyncState as Socket;

            if (socket.Connected)
            {
                socket.EndConnect(asyncresult);
                IsConnectionSuccessful = true;
            }
        }
        catch (Exception ex)
        {
            IsConnectionSuccessful = false;
            socketexception = ex;
        }
        finally
        {
            TimeoutObject.Set();
        }
    }

    public static void WaitForData()
    {
        try
        {
            if (asyncCallBack == null)
            {
                asyncCallBack = new AsyncCallback(OnDataReceived);
            }
            CSocketPacket theSocPkt = new CSocketPacket();
            theSocPkt.thisSocket = clientSocket;
            asyncResult = clientSocket.BeginReceive(theSocPkt.dataBuffer, 0, theSocPkt.dataBuffer.Length, SocketFlags.None, asyncCallBack, theSocPkt);
        }
        catch (SocketException se)
        {
            notifyErrorEventSubscribers(se);
        }
    }

    public static void send(string message)
    {
        try
        {
            byte[] byData = System.Text.Encoding.ASCII.GetBytes(message);
            clientSocket.Send(byData);
        }
        catch (SocketException se)
        {
            notifyErrorEventSubscribers(se);
            throw;
        }
    }

    //[MethodImpl(MethodImplOptions.Synchronized)]
    public static void OnDataReceived(IAsyncResult result)
    {
        try
        {
            CSocketPacket theSockId = (CSocketPacket)result.AsyncState;

            //end receive...
            int messageSize = 0;

            messageSize = theSockId.thisSocket.EndReceive(result);

            Console.WriteLine(">>>>>>>>>  messageSize = " + messageSize); // !!!

            char[] chars = new char[messageSize + 1];

            System.Text.Decoder d = System.Text.Encoding.ASCII.GetDecoder();
            int charLen = d.GetChars(theSockId.dataBuffer, 0, messageSize, chars, 0);

            string replyMessage = new System.String(chars);

            lock (syncLock)  //LastIndexOf function accesses the current culture info and we clear it in WM_TIMECHANGE handler (protecting from that race condition here)
            {
                if (replyMessage.LastIndexOf("\0") > 0)
                    replyMessage = replyMessage.Remove(replyMessage.LastIndexOf("\0"), 1);
                if (replyMessage.LastIndexOf(Terminator) > 0)
                    replyMessage = replyMessage.Remove(replyMessage.LastIndexOf(Terminator), 1);
            }


            // Continue the waiting for data on the Socket
            WaitForData();

            receivedMsg += replyMessage;


            // only serialize when we feel we have a message or we have reached the message line limit  
            if (((receivedMsg.Contains("message") && receivedMsg.Contains("/>")) || receivedMsg.Contains("</message>")) /* || (mRecvdMsgLineCount == Message.kMaxLines) */ )
            {
                List<XmlMessage> msgList = new List<XmlMessage>();

                int index = -1;
                do
                {
                    index = receivedMsg.IndexOf("</message>");

                    if (index != -1)
                    {
                        XmlMessage message;
                        string strMessage = receivedMsg.Substring(0, index + "</message>".Length);
                        //MessageBox.Show(strMessage);
                        strMessage = strMessage.TrimStart(new char[] { '\r', '\n' });

                        receivedMsg = receivedMsg.Remove(0, index + "</message>".Length);

                        try
                        {
                            message = (XmlMessage)XmlMessage.GetXmlSerializer().Deserialize(XmlReader.Create(new StringReader(strMessage)));
                        }
                        catch (InvalidOperationException error)
                        {
                            string strErrorMessage = error.Message;
                            if (error.InnerException != null)
                                strErrorMessage += "\r\n" + error.InnerException.Message;

                            notifyErrorEventSubscribers(new Exception(strErrorMessage + "\r\n-----------------------------------------------------------------\r\n" + strMessage));

                            return;
                        }

                        msgList.Add(message);
                    }
                } while (index != -1);

                StringWriter sw = new StringWriter();
                string serializedXml = string.Empty;
                string strXmlMessage = string.Empty;

                foreach (XmlMessage message in msgList)
                {
                    if (message.ack_required && (message.update == null))
                    {
                        XmlMessage messageAcknowledge = new XmlMessage();
                        messageAcknowledge.ack_required = false;
                        messageAcknowledge.ack = new ack();
                        messageAcknowledge.ack.success = true;
                        messageAcknowledge.ack.id = message.id;

                        try
                        {
                            sendMessage(messageAcknowledge);
                        }
                        catch(Exception ex)
                        {
                            Logger.Log(EnumLoggingLevel.Error, "SocketCommunicationXMl.OnDataReceived", ex.Message);
                        }
                    }

                    if (dataReceivedEvent != null)
                    {
                        dataReceivedEvent(new object(), new DataReceivedEventArgs(message));
                    }

                    if ((ackRequiredMsg != null) && (message.ack != null))
                    {
                         if ((message.ack.id == ackRequiredMsg.id) && message.ack.success)
                         {
                             eventWaitForAck.Set();
                         }
                    }
                }
            }
        }
        catch (ObjectDisposedException objectDisposedException)
        {
            //              Dispatcher.dispatchDebug(Debug.Level_3,"Socket has been closed", this);
            notifyErrorEventSubscribers(objectDisposedException);
        }
        catch (SocketException se)
        {
            if (se.ErrorCode == 10054)
            {
                /*
                for (int i = 0; i < 50; i++)
                {
                    Thread.Sleep(1000);
                }

                try
                {
                    SocketCommunicationDaq.Reconnect();
                }
                catch(Exception ex)
                {
                    VNMsgBox.Show(ex.Message, MsgButtonType.OkOnly, MsgIconType.Error);
                    return;
                }

                clientSocket.Shutdown(SocketShutdown.Both);
                clientSocket.Close();

                for (int i = 0; i < 3; i++)
                {
                    try
                    {
                        connect();
                        break;
                    }
                    catch (Exception ex)
                    {
                        System.Threading.Thread.Sleep(5000);
                    }
                }
                */
                Logger.Log(EnumLoggingLevel.Error, "OnDataReceived: ", se.ToString());
            }
            else
            {
                notifyErrorEventSubscribers(se);
            }
        }
    }

Answer 1

正如其他人所提到的，TCP 是一种流式传输协议(protocol)，因此您永远无法知道要获取全部 100 个字节需要多少次 DataReceived 回调。可以是 1，也可以是 100。

接收代码相当复杂，性能有待提高(字符串操作太多)。很难判断是否存在控制流问题。我建议分解 DataReceived 方法以进行简化。这是一个合理的框架:

    public static void OnDataReceived(IAsyncResult result)
    {
        //1) copy all data received into a buffer of some sort, like MemoryStream

        //2) Dispatch any/all full messages that have been received
          // to a queue or handler method (maybe handle on another thread)
          //(hold onto any leftover bytes received that are part of the next message)

        //Call BeginReceive() again
    }

此外，如果您使用长度前缀消息格式，它还可以帮助简化框架。