Method FlushPackets

library/cpp/netliba/v12/udp_host.cpp:878–937

ui8 - error bitmap FPR_OVERFLOW - there were non-sent packets because of udp/kernel buffer overflow. FPR_OUT_TRANSFERS_CHANGED - some packets were not send because NO_ROUTE_TO_HOST Informs all transfers about problems ("failed transfer" or "add packet to resend"). Clears TUdpSocket packet queue.

    // Informs all transfers about problems ("failed transfer" or "add packet to resend").
    // Clears TUdpSocket packet queue.
    ui8 TUdpHost::FlushPackets() {
        NewPacketsAfterLastFlush = 0;
        //The logic is we must inform caller about errors, but it is not good idea
        //to break flushing in case of NO_ROUTE_TO_HOST error. So, we will accumulate errors
        //as bitmask and pass it to caller
        ui8 errRet = TUdpHost::EFlashPacketResult::FPR_OK;
        for (;;) {
            size_t numSentPackets;
            TVector<std::pair<char*, size_t>> failedPackets;
            TUdpSocket::ESendError err = S.FlushPackets(&numSentPackets, &failedPackets);

            if (err == TUdpSocket::SEND_OK) {
                Y_ASSERT(S.IsPacketsQueueEmpty());
                return errRet;

                // most probably out of send buffer space (or something terrible has happened)
            } else if (err == TUdpSocket::SEND_BUFFER_OVERFLOW) {
                TVector<std::pair<char*, size_t>> notSentPackets;
                S.GetPacketsInQueue(&notSentPackets);

                for (size_t i = 0; i != notSentPackets.size(); ++i) {
                    const std::pair<char*, size_t>& notSentPacket = notSentPackets[i];
                    TTransfer transfer;
                    int packetId;
                    if (ParseDataPacketHeader(notSentPacket.first, notSentPacket.first + notSentPacket.second, &transfer, &packetId)) {
                        TConnection* connection = CheckedCast<TConnection*>(transfer.Connection.Get());
                        connection->GetSendQueue().Get(transfer.Id)->AckTracker.AddToResend(packetId);
                        //fprintf(stderr, "Failed send\n");
                    }
                }
                MaxWaitTime = 0;
                S.ClearPacketsQueue(); // MUST be called only after flush (to simulate 100% flush in local connections)
                Y_ASSERT(S.IsPacketsQueueEmpty());
                errRet |= TUdpHost::EFlashPacketResult::FPR_OVERFLOW;
                return errRet;

            } else if (err == TUdpSocket::SEND_NO_ROUTE_TO_HOST || err == TUdpSocket::SEND_EINVAL) {
                const char* errText = (err == TUdpSocket::SEND_NO_ROUTE_TO_HOST) ? "No route to host" : "Error in value";
                for (size_t i = 0; i != failedPackets.size(); ++i) {
                    const std::pair<char*, size_t>& failedPacket = failedPackets[i];
                    TTransfer transfer;
                    int packetId;
                    if (ParseDataPacketHeader(failedPacket.first, failedPacket.first + failedPacket.second, &transfer, &packetId)) {
                        FailedSend(transfer);
                        fprintf(stderr, "%s, transfer: %" PRIu64 " failed, packetId: %" PRIi32 "\n",
                                errText, transfer.Id, packetId);
                    }
                    // packet is already dropped by socket queue
                }
                errRet |= TUdpHost::EFlashPacketResult::FPR_OUT_TRANSFERS_CHANGED;
                continue;

            } else {
                Y_ASSERT(false);
                break;
            }
        }
        Y_ASSERT(false); // unreachable