[C++] Multithreaded programmeren (bijv. chat server)

[vdvleon]

Hallo iedereen. Ik ben bezig (al weer) om multithreaded te programmeren. Ik wil als oefening een chat server maken (en de clients). Om mij zelf wat werk uit handen te nemen ben ik sinds kort gaan kijken naar de Qt Library.

Een multithreaded server opzetten is opzich het probleem niet. Maar nu komt het probleem. Ik namelijk dat clientX naar clientY iets kan versturen. Dus moet de server alles connectie in een grote lijst opslaan? Maar om je dus vanuit verschillende threads dan deze lijst zou moeten uitlezen geeft dan problemen. Ik zou dan met mutexen moeten gaan werken ? Maar dan gaat er iets fout. Namelijk, ik elke thread zit een loop die gewoon wacht tot de client een regel stuurd en handelt deze dan af. Of te wel, de client socket is continue in gebruik. Hoe kan ik dan toch als deze client socket in gebruik is tegelijkertijd vanuit een andere thread via die list iets sturen naar die client socket?

Ik kom er niet echt uit. Ik heb dit eigenlijk al tig keren geprobeerd, maar dit lukt me nooit. En als ik dan een systeem heb met mutexen etc. kan je in de laatste gemaakt thread wel alle client zien in de list, maar met een oudere thread zie je de nieuwste clients niet... etc. etc.

[Woy]

Zowiezo zou ik het afraden om voor applicaties die veel verbindingen tegelijk open hebben, voor elke verbinding een thread te starten.

Ik zou eens kijken naar ASynchrone communicatie. Het voordeel is dat je dan geen blocking calls hebt, en dus niet voor elke connectie een thread hoeft te starten.

Verder snap ik je probleem met betrekking tot die list niet zo?

Ik zou voor elke connectie een instance van een class maken die de benodigde data bevat.

code:

class MyChatConnection
{
        TcpConnection myConnection;
        
         SendData( data ){  myConnection.BeginSendData( data, EndSendData ); }
         SendDataFinished() { /* Handle send finished*/ }

         BeginReceiveData(){ myConnection.BeginReceive( buffer, ReceivedData ); }
         ReceivedData()
         {
              //Handle data
              BeginReceiveData();
         }
}

Hierboven even een idee hoe het ongeveer zou kunnen.

[danslo]

Gooi gewoon je socket handles (of pointers naar instances van een socket class) in een std::vector, je moet er alleen iets omheen te wrappen om het thread-safe te maken (STL heeft dat niet), daar is genoeg over te vinden op het internet.

[ Voor 11% gewijzigd door danslo op 16-03-2009 12:33 ]

[VyperX]

vdvleon schreef op maandag 16 maart 2009 @ 12:09:
Hallo iedereen. Ik ben bezig (al weer) om multithreaded te programmeren. Ik wil als oefening een chat server maken (en de clients). Om mij zelf wat werk uit handen te nemen ben ik sinds kort gaan kijken naar de Qt Library.

...

Werk je nu al met Qt of nog niet? Als ik het me goed herinner heb je in Qt zowiezo al asynchrone communicatie beschikbaar... (Je sockets geven signals af als er nieuwe data beschikbaar is. Op dat signal kan je weer reageren...) Zolang je een Qt eventloop in je thread laat lopen kan je alles met signals/slots afhandelen.

Als het je echt gaat om het leren omgaan met multithreaded applicaties, dan kan je in elke thread een eventloop starten. Qt handelt het zelf goed af als signals worden gestuurd van objecten in de ene thread naar objecten in een andere thread.

cls schreef op maandag 16 maart 2009 @ 12:32:
Gooi gewoon je socket handles (of pointers naar instances van een socket class) in een std::vector, je moet er alleen iets omheen te wrappen om het thread-safe te maken (STL heeft dat niet), daar is genoeg over te vinden op het internet.

Qt heeft zijn eigen QVectoren die (iirc) wel thread-safe zijn.

[Woy]

VyperX schreef op maandag 16 maart 2009 @ 12:42:
[...]
Qt heeft zijn eigen QVectoren die (iirc) wel thread-safe zijn.

Ook al is de implementatie van de Vector wel thread-safe, dan betekend dat natuurlijk nog niet dat je niet meer na hoeft te denken voordat je ze in een multithreaded omgeven gaat gebruiken.

[Matis]

vdvleon schreef op maandag 16 maart 2009 @ 12:09:
Ik kom er niet echt uit. Ik heb dit eigenlijk al tig keren geprobeerd, maar dit lukt me nooit. En als ik dan een systeem heb met mutexen etc. kan je in de laatste gemaakt thread wel alle client zien in de list, maar met een oudere thread zie je de nieuwste clients niet... etc. etc.

Ik kan uit je tekst aflezen dat je dit probleem al meerdere keren hebt gehad en dat je niet uit je source-code komt.

Misschien wat (pseudo) code posten waaruit wij kunnen opmaken waar het fout gaat...

Heb je in deze thread al een keer geprobeerd je problemen voor je op te laten lossen? :
Voorbeeld Threaded Server

Edit:
http://www.beej.us/guide/...html/multipage/index.html

[ Voor 5% gewijzigd door Matis op 16-03-2009 13:29 . Reden: Extra tekst, Typo's ]

[roy-t]

Lees anders is na over een Threadpool, je zat dan taken in een queue, een set worker threads dat al aangemaakt is (geen thread creation overhead) gaat dan bezig met die taken. Je kunt vaak instellen hoeveel threads je wil (Een leuke is misschien wel 4x aantal cores target machine, maar verschilt natuurlijk ontzettend met hoeveel resources een thread gebruikt en hoeveel idle time een thread heeft).

In C# zit dat patroon standaard ingebakken (Threadpool) maar ik weet niet of dat in een van de C++ libs ook standaard zit.

[Soultaker]

vdvleon schreef op maandag 16 maart 2009 @ 12:09:
Namelijk, in elke thread zit een loop die gewoon wacht tot de client een regel stuurt en handelt deze dan af. Of te wel, de client socket is continue in gebruik.

Het is dus zaak om ervoor te zorgen dat terwijl je aan het wachten bent op invoer, je de connectie niet gelockt hebt en dat ondertussen andere threads dus wel kunnen zenden. Normaalgesproken (met native sockets, ik weet niet precies wat Qt daar allemaal omheen verzint) kun je prima data verzenden door een socket terwijl een andere thread aan het wachten is op binnenkomende data op die socket. Natuurlijk moeten niet twee threads tegelijk door elkaar zenden (hoewel dat soms ook nog wel kan, omdat sommige system calls atomair worden uitgevoerd, hoewel dat nogal eens verschilt tussen besturingssystemen) maar daarvoor kun je prima een mutex gebruiken.

Stel dus dat je per connectie een mutex hebt, en je hebt per connectie een thread die wacht tot de socket readable wordt. Zodra dat gebeurt, lock je de connectie mutex en handel je de binnengekomen data af (wellicht stuur je ook data terug dan) en daarna geef je de mutex weer vrij en ga je opnieuw wachten. Als er maar één thread van de socket leest (de connectie-thread) heb je daarvoor de mutex niet eens nodig, en hoef je die pas te locken als je data terug wil sturen.

Overigens moet je nu wel weer oppassen met deadlock vanwege lock inversion. Een goede manier om dat te voorkomen is ervoor te zorgen dat een thread op enig moment hooguit één mutex gelockt houdt.

[Soultaker]

Woy schreef op maandag 16 maart 2009 @ 12:31:
Zowiezo zou ik het afraden om voor applicaties die veel verbindingen tegelijk open hebben, voor elke verbinding een thread te starten. [..] Ik zou eens kijken naar ASynchrone communicatie. Het voordeel is dat je dan geen blocking calls hebt, en dus niet voor elke connectie een thread hoeft te starten.

Hier geef ik meestal ook de voorkeur aan, niet alleen vanwege de betere schaalbaarheid (hoewel moderne besturingssystemen best een stuk of honderd threads kunnen managen, zeker als die 99% van de tijd staan te wachten op sockets) maar meer omdat je dan géén threads hoeft te gebruiken en dus ook niet na hoeft te denken over locking of concurrente wijzigingen van globale data. Dat is een stuk overzichtelijker, maar heeft weer als nadeel dat het niet (eenvoudig) schaalt over meerdere cores.

Toch zou ik de TS niet aanraden daar nu op over te stappen. Op zichzelf is een thread-gebaseerde aanpak niet verkeerd, en het lijkt me zinniger om te leren om met multithreaded code om te gaan dan om alle threads af te schaffen. Dan omzeil je het probleem wel, maar leer je er ook niets van.

[Woy]

Soultaker schreef op maandag 16 maart 2009 @ 14:30:
[...]
Toch zou ik de TS niet aanraden daar nu op over te stappen. Op zichzelf is een thread-gebaseerde aanpak niet verkeerd, en het lijkt me zinniger om te leren om met multithreaded code te leren omgaan dan om alle threads af te schaffen. Dan omzeil je het probleem wel, maar leer je er ook niets van.

Natuurlijk is het ook belangrijk om met multithreaded code om te leren gaan. Maar juist voor netwerk code vind ik het vaak niet de beste/makkelijkste oplossing.

Voor reken intensieve taken is het zeker aan te raden om wel multi-treaded te gaan werken. Netwerk code zal vaak niet CPU limited zijn. De verwerking van de data die je binnen krijgt kun je natuurlijk eventueel wel weer multi-threaded verwerken.

[vdvleon]

Ik ga maar eens met de Qt signals proberen. Ik begrijp wel ongeveer hoe die werken. Misschien kan ik dan iets leuks in elkaar fabriceren. Bedankt voor jullie hulp. Ik laat wel horen of het gelukt is.

Multithreaded is handig en leuk, maar een stuk lastiger te programeren, maar wel een leuke uitdaging

[unclero]

vdvleon schreef op maandag 16 maart 2009 @ 12:09:
Hallo iedereen. Ik ben bezig (al weer) om multithreaded te programmeren. Ik wil als oefening een chat server maken (en de clients). Om mij zelf wat werk uit handen te nemen ben ik sinds kort gaan kijken naar de Qt Library.

Ik weet niet of je een specifiek protocol voor ogen hebt, maar als je echt lui bent zou je gSOAP kunnen overwegen . Gebruiken een aantal grote jongens ook, en ik heb het ook met veel plezier ingezet bij een heavy-load client-server toepassing.
Ondersteunt prima pthreads.

[vdvleon]

Ik had gewoon het idee om mijn eigen protocol te schrijven. Het word een chat systeem. Ik verzin wel iets leuks en wil zelf graag bezig met multithreaded programmeren. Ik gebruik de Qt Library wel... Thnx anyway

[vdvleon]

Ik heb nu een werkend geheel. Dus een signal driven threaded server. Ook kan ik alle connections opslaan, en benaderen vanuit elke thread. Er is alleen 1 probleem. Heel soms crasht de server opeens. Vaak als net een nieuwe client connect (na tig keer meerdere clients te laten connecten, disconnect, etc.).

Volgens mij is alles toch echt threaded geprogrammeerd, maar daar ben ik ook nog niet zo ervaren in. Misschien zien jullie het:

De connection holder class:

C++:

class vdChat_client_holder{
    public:
        // Constructor
        vdChat_client_holder() : sock(NULL), sm_disconnected(NULL), sm_on_data(NULL), name(":"), in_use(false){}
    
        // Get name
        QString& getName(){
            // Return name
            return name;
        }
        
        // In use
        bool& is_in_use(){
            // Return in_use
            return in_use;
        }
        
        // Get sock
        QTcpSocket* getSock(){
            // Return sock
            return sock;
        }
        QTcpSocket* operator()(){
            // Return sock
            return sock;
        }
        
        // Clear
        void clear(){
            // Clear sock
            if(sock!=NULL){
                sock->abort();
                delete sock;
                sock = NULL;
            }
            
            // Clear disconnected signal mapper
            if(sm_disconnected!=NULL){
                delete sm_disconnected;
                sm_disconnected = NULL;
            }
            
            // Clear on_data signal mapper
            if(sm_on_data==NULL){
                delete sm_on_data;
                sm_on_data = NULL;
            }
            
            // Clear name
            name.clear();
            
            // Set not in use
            in_use = false;
        }
    
    private:
        // Vars
        QTcpSocket* sock;
        QSignalMapper* sm_disconnected;
        QSignalMapper* sm_on_data;
        QString name;
        bool in_use;
    
    // Define friend class
    friend class vdChat_server;
};

De Server class (header):

C++:

class vdChat_server : public QTcpServer{
    Q_OBJECT
    
    public:
        vdChat_server(QHostAddress listen_addr, int port);
    
    protected:
        void on_connect(int id);
        void on_disconnect(int id);
        void on_data(int id);
    
    private:
        QVector<vdChat_client_holder> _clients;
        int _clients_count;
        mutable QMutex* _client_count_mutex;
        
        int& get_clients_count();
        int new_id();
        void send_all(QString line);
        
    protected slots:
        void slot_new_connection();
        void slot_on_disconnect(int id);
        void slot_on_data(int id);
};

De connection class (body):

C++:

vdChat_server::vdChat_server(QHostAddress listen_addr, int port){
    // Init
    _client_count_mutex = new QMutex;
    _clients.resize(MAX_CONNECTIONS);
    _clients_count = 0;
    
    // Listen to
    if(!listen(listen_addr, port) || !isListening()){
        // Error
        std::cerr << "Couldn't start server.\n";
        exit(1);
    }
    
    // Set signals
    connect(this, SIGNAL(newConnection()), this, SLOT(slot_new_connection()));
    
    // Server started
    std::cout << "Server started on " << serverAddress().toString().toStdString() << " on port " << serverPort() << "\n";
}

void vdChat_server::on_connect(int id){
    // Debug
    std::cout << "Client connected: " << _clients[id].getName().toStdString() << "\n";
}
        
void vdChat_server::on_disconnect(int id){
    // Debug
    std::cout << "Client disconnected: " << _clients[id].getName().toStdString() << "\n";
    
    // Send disconnect message
    send_all(QString("Client disconnected: %1").arg(_clients[id].getName()));
}

void vdChat_server::on_data(int id){
    // Debug
    std::cout << "[" << _clients[id].getName().toStdString() << "] " << _clients[id].getSock()->readLine().data() << "\n";
    
    // Send to all
    send_all(QString("[%1] %2\n").arg(_clients[id].getName()).arg(_clients[id].getSock()->readLine().data()));
}

int& vdChat_server::get_clients_count(){
    // Thread safety
    QMutexLocker locker(_client_count_mutex);
    
    // Return
    return _clients_count;
}

int vdChat_server::new_id(){
    // Search for empty position
    for(int i=0; i<MAX_CONNECTIONS; i++){
        if(!_clients[i].is_in_use()){
            // Return empty position
            return i;
        }
    }
    
    // Error
    return -1;
}

void vdChat_server::send_all(QString line){
    // Vars
    int i;
    int j;
    
    // For each client
    for(i=0, j=0; i<MAX_CONNECTIONS && j<get_clients_count(); i++){
        // Found client?
        if(_clients[i].is_in_use()){
            // Send line
            _clients[i].getSock()->write(line.toAscii());
            j++;
        }
    }
}

// Slots
void vdChat_server::slot_new_connection(){
    // Vars
    int id;
    
    // Get client
    QTcpSocket* client = nextPendingConnection();

    // Max connections reached?
    if(get_clients_count()==MAX_CONNECTIONS){
        // Close connection
        client->abort();
        delete client;
        client = NULL;
        return;
    }
            
    // Get new client id
    id = new_id();
    if(id<0){
        // Error
        std::cerr << "Error, some client connection not correctly closed\n";
        exit(1);
    }
    
    // Add client to _clients
    _clients[id].sock               = client;
    _clients[id].name               = QString("%1:%2").arg(client->peerAddress().toString()).arg(client->peerPort());
    _clients[id].in_use             = true;
    _clients[id].sm_disconnected    = new QSignalMapper(this);
    _clients[id].sm_on_data         = new QSignalMapper(this);
    get_clients_count()++;
    
    // Set disconnected signal
    connect(_clients[id].getSock(), SIGNAL(disconnected()), _clients[id].sm_disconnected, SLOT(map()));
    _clients[id].sm_disconnected->setMapping(_clients[id].getSock(), id);
    connect(_clients[id].sm_disconnected, SIGNAL(mapped(int)), this, SLOT(slot_on_disconnect(int)));
    
    // Set on_data signal
    connect(_clients[id].getSock(), SIGNAL(readyRead()), _clients[id].sm_on_data, SLOT(map()));
    _clients[id].sm_on_data->setMapping(_clients[id].getSock(), id);
    connect(_clients[id].sm_on_data, SIGNAL(mapped(int)), this, SLOT(slot_on_data(int)));
    
    // Run on_connect event
    on_connect(id);
}

void vdChat_server::slot_on_disconnect(int id){
    // On disconnect
    on_disconnect(id);
    
    // Remove from list
    _clients[id].clear();
    get_clients_count()--;
}

void vdChat_server::slot_on_data(int id){
    // On data
    on_data(id);
}

[Woy]

Op deze manier is het dus niet de bedoeling. Je zult zelf echt wat meer inzet moeten tonen.

Gewoon 200+ regels code plaatsen en verwachten dat wij voor jou gaan uitpluizen wat er mis is, dat doen we niet.

Lees ook even Tips bij het debuggen door. Als je eerst zelf probeert te debuggen, en dan een concreet probleem hebt, kun je een nieuw topic openen, met daarin alleen relevante code. Zet er dan wel goed bij wat je zelf al geprobeerd hebt, en wat daar niet aan lukt.