"Connection was broken" error with UDT (UDP-based data transfer protocol)

Question

I am programming a real-time game in which I need reliable UDP, so I've chosen to work with UDT (UDP-based data transfer protocol - http://sourceforge.net/projects/udt/).

The clients (on browsers) send real-time messages to my server via CGI scripts. The problem is that there are some messages that are being lost, and I don't know why because the server says that it sent all the messages successfully to the corresponding clients, but sometimes the client doesn't receive the message.

In my debug file, I've found that when a message is not received by the client, its script says:

error in recv();
recv: Connection was broken.

I would like to get some help on how the server shall know if the client got its message; should I send a NACK or something from the client side? I thought that UDT should do that for me. Can someone clarify this situation?

The relevant sections of the communication parts of my code are bellow, with some comments:

server's relevant code:

//...

void send_msg_in(player cur, char* xml){
/*this function stores the current message, xml, in a queue if xml!=NULL, and sends the 1st message of the queue to the client*/
/*this function is called when the player connects with the entering xml=NULL to get the 1st message of the queue,
or with xml!=NULL when a new message arrives: in this case the message is stored in the queue, and then the message will be sent in the appropriate time, i.e. the messages are ordered.*/
    char* msg_ptr=NULL;

    if (xml!=NULL){         //add the message to a queue (FIFO), the cur.xml_msgs
        msg_ptr=(char*) calloc(strlen(xml)+1, sizeof(char));
        strcpy(msg_ptr, xml);
        (*(cur.xml_msgs)).push(msg_ptr);
    }                       //get the 1st message of the queue
    if (!(*(cur.xml_msgs)).empty()){
        xml=(*(cur.xml_msgs)).front();
    }

    if (cur.get_udt_socket_in()!=NULL){
        UDTSOCKET cur_udt = *(cur.get_udt_socket_in());
//      cout << "send_msg_in(), cur_udt: " << cur_udt << endl;

        //send the "xml", i.e. the 1st message of the queue...
        if (UDT::ERROR == UDT::send(cur_udt, xml, strlen(xml)+1, 0)){
            UDT::close(cur_udt);
            cur.set_udt_socket_in(NULL);
        }
        else{    //if no error this else is reached
            cout << "TO client:\n" << xml << "\n";    /*if there is no error,
                             i.e. on success, the server prints the message that was sent.*/
                             //  / \
                             // /_!_\
                             /*the problem is that
                             the messages that are lost don't appear on the client side,
                             but they appear here on the server! */
            if (((string) xml).find("<ack.>")==string::npos){
                UDT::close(cur_udt);
                cur.set_udt_socket_in(NULL);          //close the socket
            }
            (*(cur.xml_msgs)).pop();
        }
    }
}

//...

client's relevant code:

//...
#define MSGBUFSIZE 1024
char msgbuf[MSGBUFSIZE];
UDTSOCKET client;
ofstream myfile;

//...

main(int argc, char *argv[]){
    //...

    // connect to the server, implict bind
    if (UDT::ERROR == UDT::connect(client, (sockaddr*)&serv_addr, sizeof(serv_addr))){
        cout << "error in connect();" << endl;
        return 0;
    }

    myfile.open("./log.txt", ios::app);

    send(xml);
    char* cur_xml;

    do{
        cur_xml = receive();                //wait for an ACK or a new message...
        myfile << cur_xml << endl << endl;  //  / \
                                            /* /_!_\ the lost messages don't appear on the website
                                            neither on this log file.*/
    } while (((string) cur_xml).find("<ack.>")!=string::npos);
    cout << cur_xml << endl;

    myfile.close();
    UDT::close(client);
    return 0;
}


char* receive(){
    if (UDT::ERROR == UDT::recv(client, msgbuf, MSGBUFSIZE, 0)){
                    //  / \
                    /* /_!_\ when a message is not well received
                    this code is usually reached, and an error is printed.*/
        cout << "error in recv();" << endl;
        myfile << "error in recv();" << endl;
        myfile << "recv: " << UDT::getlasterror().getErrorMessage() << endl << endl;
        return 0;
    }
    return msgbuf;
}


void* send(string xml){
    if (UDT::ERROR == UDT::send(client, xml.c_str(), strlen(xml.c_str())+1, 0)){
        cout << "error in send();" << endl;
        myfile << "error in send();" << endl;
        myfile << "send: " << UDT::getlasterror().getErrorMessage() << endl << endl;
        return 0;
    }
}

Thank you for any help!

PS. I tried to increase the linger time on close(), after finding the link http://udt.sourceforge.net/udt4/doc/opt.htm, adding the following to the server's code:

    struct linger l;
    l.l_onoff  = 1;
    l.l_linger = ...;  //a huge value in seconds...
    UDT::setsockopt(*udt_socket_ptr, 0, UDT_LINGER, &l, sizeof(l));

but the problem is still the same...

PPS. the other parts of the communication in the server side are: (note: it seams for me that they are not so relevant)

main(int argc, char *argv[]){
    char msgbuf[MSGBUFSIZE];

    UDTSOCKET serv = UDT::socket(AF_INET, SOCK_STREAM, 0);

    sockaddr_in my_addr;
    my_addr.sin_family = AF_INET;
    my_addr.sin_port = htons(PORT);
    my_addr.sin_addr.s_addr = INADDR_ANY;
    memset(&(my_addr.sin_zero), '\0', sizeof(my_addr.sin_zero));

    if (UDT::ERROR == UDT::bind(serv, (sockaddr*)&my_addr, sizeof(my_addr))){
        cout << "error in bind();";
        return 0;
    }

    UDT::listen(serv, 1);

    int namelen;
    sockaddr_in their_addr;

    while (true){
        UDTSOCKET recver = UDT::accept(serv, (sockaddr*)&their_addr, &namelen);
        if (UDT::ERROR == UDT::recv(recver, msgbuf, MSGBUFSIZE, 0)){
        //this recv() function is called only once for each aqccept(), because the clients call CGI scripts via a browser, they need to call a new CGI script with a new UDT socket for each request (this in in agreement to the clients' code presented before).
            cout << "error in recv();" << endl;
        }

        char* player_xml = (char*) &msgbuf;
        cur_result = process_request((char*) &msgbuf, &recver, verbose);    //ACK
    }
}


struct result process_request(char* xml, UDTSOCKET* udt_socket_ptr, bool verbose){
    //parse the XML...
    //...

    player* cur_ptr = get_player(me);   //searches in a vector of player, according to the string "me" of the XML parsing.

    UDTSOCKET* udt_ptr = (UDTSOCKET*) calloc(1, sizeof(UDTSOCKET));
    memcpy(udt_ptr, udt_socket_ptr, sizeof(UDTSOCKET));

    if (cur_ptr==NULL){
        //register the player:
        player* this_player = (player*) calloc(1, sizeof(player));
        //...
        }
    }
    else if (strcmp(request_type.c_str(), "info_waitformsg")==0){
        if (udt_ptr!=NULL){
            cur_ptr->set_udt_socket_in(udt_ptr);

            if (!(*(cur_ptr->xml_msgs)).empty()){
                send_msg_in(*cur_ptr, NULL, true);
            }
        }
    }
    else{           //messages that get instant response from the server.
        if (udt_ptr!=NULL){
            cur_ptr->set_udt_socket_out(udt_ptr);
        }
        if (strcmp(request_type.c_str(), "info_chat")==0){
            info_chat cur_info;
            to_object(&cur_info, me, request_type, msg_ptr);    //convert the XML string values to a struct
            process_chat_msg(cur_info, xml);
        }
    /*  else if (...){  //other types of messages...
        }*/
    }
}


void process_chat_msg(info_chat cur_info, char* xml_in){
    player* player_ptr=get_player(cur_info.me);

    if (player_ptr){
        int i=search_in_matches(matches, cur_info.match_ID);
        if (i>=0){
            match* cur_match=matches[i];
            vector<player*> players_in = cur_match->followers;
            int n=players_in.size();
            for (int i=0; i<n; i++){
                if (players_in[i]!=msg_owner){
                    send_msg_in(*(players_in[i]), xml, flag);
                }
            }
        }
    }
}

Answer 1

Looking at the UDT source code at http://sourceforge.net/p/udt/git/ci/master/tree/udt4/src/core.cpp, the error message "Connection was broken" is produced when either of the Boolean flags m_bBroken or m_bClosing is true and there is no data in the receive buffer.

Those flags are set in just a few cases:

• In sections of code marked "should not happen; attack or bug" (unlikely)
• In deliberate close or shutdown actions (don't see this happening in your code)
• In expiration of a timer that checks for peer activity (the likely culprit)

In that source file at line 2593 it says:

     // Connection is broken. 
     // UDT does not signal any information about this instead of to stop quietly.
     // Application will detect this when it calls any UDT methods next time.
     //
     m_bClosing = true;
     m_bBroken = true;

     // ...[code omitted]...

     // app can call any UDT API to learn the connection_broken error

Looking at the send() call, I don't see anywhere that it waits for an ACK or NAK from the peer before returning, so I don't think a successful return from send() on the server side is indicative of successful receipt of the message by the client.

You didn't show the code on the server side that binds to the socket and listens for responses from the client; if the problem is there then the server might be happily sending messages and never listening to the client that is trying to respond.

Answer 2

UDP is not a guaranteed-transmission protocol. A host will send a message, but if the recipient does not receive it, or if it is not received properly, no error will be raised. Therefore, it is commonly used in applications that require speed over perfect delivery, such as games. TCP does guarantee delivery, because it requires that a connection be set up first, and each message is acknowledged by the client.

I would encourage you to think about whether you actually need guaranteed receipt of that data, and, if you do, consider using TCP.