第二人生的源码分析(三十二)消息解包的实现

从UDP接收到数据后，就会组装成一个完整的数据包，然后检验整个数据包是否有效，并且还处理收到回应的UDP包标识，这样构造一个完整的可靠性连接。具体处理代码如下：
#001 BOOL LLMessageSystem::checkMessages( S64 frame_count )
#002 {
#003 // Pump
#004 BOOL valid_packet = FALSE;
#005 mMessageReader = mTemplateMessageReader;
#006
#007 LLTransferTargetVFile::updateQueue();
#008

下面保存第一次收到消息的时间。
#009 if (!mNumMessageCounts)
#010 {
#011 // This is the first message being handled after a resetReceiveCounts,
#012 // we must be starting the message processing loop. Reset the timers.
#013 mCurrentMessageTimeSeconds = totalTime() * SEC_PER_USEC;
#014 mMessageCountTime = getMessageTimeSeconds();
#015 }
#016

下面开始循环处理所有收到的消息包。
#017 // loop until either no packets or a valid packet
#018 // i.e., burn through packets from unregistered circuits
#019 S32 receive_size = 0;
#020 do
#021 {
#022 clearReceiveState();
#023
#024 BOOL recv_reliable = FALSE;
#025 BOOL recv_resent = FALSE;
#026 S32 acks = 0;
#027 S32 true_rcv_size = 0;
#028
#029 U8* buffer = mTrueReceiveBuffer;
#030

从底层环路里接收数据包。
#031 mTrueReceiveSize = mPacketRing.receivePacket(mSocket, (char *)mTrueReceiveBuffer);
#032 // If you want to dump all received packets into SecondLife.log, uncomment this
#033 //dumpPacketToLog();
#034

获取数据包的大小和发送的服务器地址。
#035 receive_size = mTrueReceiveSize;
#036 mLastSender = mPacketRing.getLastSender();
#037
下面检验数据包的长度是否有效，如果无效就处理下一个数据包。
#038 if (receive_size < (S32) LL_MINIMUM_VALID_PACKET_SIZE)
#039 {
#040 // A receive size of zero is OK, that means that there are no more packets available.
#041 // Ones that are non-zero but below the minimum packet size are worrisome.
#042 if (receive_size > 0)
#043 {
#044 llwarns << "Invalid (too short) packet discarded " << receive_size << llendl;
#045 callExceptionFunc(MX_PACKET_TOO_SHORT);
#046 }
#047 // no data in packet receive buffer
#048 valid_packet = FALSE;
#049 }
#050 else
#051 {
#052 LLHost host;
#053 LLCircuitData* cdp;
#054

判断这个数据包后面是否包含发送数据包的回应ID，这样可以处理可靠性的包发送，不像包丢失。
#055 // note if packet acks are appended.
#056 if(buffer[0] & LL_ACK_FLAG)
#057 {
#058 acks += buffer[--receive_size];
#059 true_rcv_size = receive_size;
#060 if(receive_size >= ((S32)(acks * sizeof(TPACKETID) + LL_MINIMUM_VALID_PACKET_SIZE)))
#061 {
#062 receive_size -= acks * sizeof(TPACKETID);
#063 }
#064 else
#065 {
#066 // mal-formed packet. ignore it and continue with
#067 // the next one
#068 llwarns << "Malformed packet received. Packet size "
#069 << receive_size << " with invalid no. of acks " << acks
#070 << llendl;
#071 valid_packet = FALSE;
#072 continue;
#073 }
#074 }
#075

下面解压缩包数据。
#076 // process the message as normal
#077 mIncomingCompressedSize = zeroCodeExpand(&buffer, &receive_size);
#078 mCurrentRecvPacketID = ntohl(*((U32*)(&buffer[1])));
#079 host = getSender();
#080

找到这个HOST的环路，然后让这个环路管理类处理。
#081 const bool resetPacketId = true;
#082 cdp = findCircuit(host, resetPacketId);
#083
#084 // At this point, cdp is now a pointer to the circuit that
#085 // this message came in on if it's valid, and NULL if the
#086 // circuit was bogus.
#087

处理那些服务器已经收到回应的ID。
#088 if(cdp && (acks > 0) && ((S32)(acks * sizeof(TPACKETID)) < (true_rcv_size)))
#089 {
#090 TPACKETID packet_id;
#091 U32 mem_id=0;
#092 for(S32 i = 0; i < acks; ++i)
#093 {
#094 true_rcv_size -= sizeof(TPACKETID);
#095 memcpy(&mem_id, &mTrueReceiveBuffer[true_rcv_size], /* Flawfinder: ignore*/
#096 sizeof(TPACKETID));
#097 packet_id = ntohl(mem_id);
#098 //llinfos << "got ack: " << packet_id << llendl;
#099 cdp->ackReliablePacket(packet_id);
#100 }
#101 if (!cdp->getUnackedPacketCount())
#102 {
#103 // Remove this circuit from the list of circuits with unacked packets
#104 mCircuitInfo.mUnackedCircuitMap.erase(cdp->mHost);
#105 }
#106 }
#107

查看这个包是否需要确认的处理的。
#108 if (buffer[0] & LL_RELIABLE_FLAG)
#109 {
#110 recv_reliable = TRUE;
#111 }
#112 if (buffer[0] & LL_RESENT_FLAG)
#113 {
#114 recv_resent = TRUE;
#115 if (cdp && cdp->isDuplicateResend(mCurrentRecvPacketID))
#116 {
#117 // We need to ACK here to suppress
#118 // further resends of packets we've
#119 // already seen.
#120 if (recv_reliable)
#121 {
#122 //mAckList.addData(new LLPacketAck(host, mCurrentRecvPacketID));
#123 // ***************************************
#124 // TESTING CODE
#125 //if(mCircuitInfo.mCurrentCircuit->mHost != host)
#126 //{
#127 // llwarns << "DISCARDED PACKET HOST MISMATCH! HOST: "
#128 // << host << " CIRCUIT: "
#129 // << mCircuitInfo.mCurrentCircuit-
#130 >mHost
#131 // << llendl;
#132 //}
#133 // ***************************************
#134 //mCircuitInfo.mCurrentCircuit->mAcks.put(mCurrentRecvPacketID);
#135 cdp->collectRAck(mCurrentRecvPacketID);
#136 }
上面把收到服务器的包ID也保存到下一次回应包队列里，准备返回去给服务器确认。

#137
#138 //llinfos << "Discarding duplicate resend from " << host << llendl;
#139 if(mVerboseLog)
#140 {
#141 std::ostringstream str;
#142 str << "MSG: <- " << host;
#143 char buffer[MAX_STRING]; /* Flawfinder: ignore*/
#144 snprintf(buffer, MAX_STRING, "/t%6d/t%6d/t%6d ", receive_size,
#145 (mIncomingCompressedSize ? mIncomingCompressedSize : receive_size), mCurrentRecvPacketID); /* Flawfinder: ignore */
#146 str << buffer << "(unknown)"
#147 << (recv_reliable ? " reliable" : "")
#148 << " resent "
#149 << ((acks > 0) ? "acks" : "")
#150 << " DISCARD DUPLICATE";
#151 llinfos << str.str() << llendl;
#152 }
#153 mPacketsIn++;
#154 valid_packet = FALSE;
#155 continue;
#156 }
#157 }
#158

通过上面这段代码的学习，可以看到UDP的可靠性，就是通过自己的包ID来作确认的。主要通过包的长度来查看包是否完整，然后通过包的标识来查看这个数据包后面是否有服务器收到的包ID发回来确认。因为有些数据是经过压缩的，所以在这里也调用函数zeroCodeExpand来解压。