Discussion :

[ramislebob]

Bonjour,
je compte faire un server tcp synchrone multi-thread en utilisant boost asio.
J'ai essayer d'adapter l'exemple simple de boost de l'echo server (http://www.boost.org/doc/libs/1_55_0...cho_server.cpp) en utilisant un functor et j'obtiens un comportement bizarre...
Je crée mon functor en l'initialisant avec la socket puis je le passe dans un thread et le detache.

Voici mon code:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
 
// g++ -std=c++11 -lboost_system -pthread blocking_echo_serv.cpp -o echo_serv
 
#include <cstdlib>
#include <iostream>
#include <thread>
#include <utility>
#include <boost/asio.hpp>
#include <atomic>
#include <vector>
 
using boost::asio::ip::tcp;
 
const int max_length = 1024;
 
class Session
{
	public:
		Session(tcp::socket *sock):
			iSock(sock), iId(sIds)
		{
			sIds++;
			std::cout << "Session[" << iId << "] created\n";
			std::cout << "sIds: " << sIds << std::endl;
		}
 
		~Session()
		{
			delete iSock;
			std::cout << "Session[" << iId << "] deleted\n";
		}
 
		tcp::socket *sock(){return iSock;}
 
		void operator() (){
			std::cout << "Session[" << iId << "] starting\n";
			try
			{
				for (;;)
				{
					char data[max_length];
 
					boost::system::error_code error;
					size_t length = iSock->read_some(boost::asio::buffer(data), error);
					if (error == boost::asio::error::eof)
						break; // Connection closed cleanly by peer.
					else if (error)
						throw boost::system::system_error(error); // Some other error.
 
					boost::asio::write(*iSock, boost::asio::buffer(data, length));
				}
			}
			catch (std::exception& e)
			{
				std::cerr << "Exception in thread: " << e.what() << "\n";
			}
 
			std::cout << "Session[" << iId << "]  finished\n";
		}
 
	private:
		tcp::socket *iSock;
		int iId;
 
		static std::atomic<int> sIds;
};
 
std::atomic<int> Session::sIds{0};
 
 
void server(boost::asio::io_service& io_service, unsigned short port)
{
	std::vector<Session *> vSessions{};
	tcp::acceptor a(io_service, tcp::endpoint(tcp::v4(), port));
	for (;;)
	{
		tcp::socket *sock = new tcp::socket(io_service);
		a.accept(*sock);
		Session *session = new Session(sock);
		vSessions.push_back(session);
		std::thread t(*session);
		t.detach();
		std::cout << "Thread detached\n";
 
	}
}
 
int main(int argc, char* argv[])
{
	if (argc != 2)
	{
		std::cerr << "Usage: echo_serv <port>\n";
		return 1;
	}
 
	try
	{
		boost::asio::io_service io_service;
 
		server(io_service, std::atoi(argv[1]));
	}
	catch (std::exception& e)
	{
		std::cerr << "Exception: " << e.what() << "\n";
	}
 
	return 0;
}

Lorsque je lance un client (telnet ou http://www.boost.org/doc/libs/1_55_0...cho_client.cpp) voici l'output:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
 
$ ./echo_serv 1111
Session[0] created
sIds: 1
Session[0] starting
Session[0] deleted
Thread detached
Exception in thread: Socket operation on non-socket
Session[0]  finished
Session[0] deleted
 
 
 
Exception: accept: Already open

Les premières lignes sont le resultat d'un telnet, la dernière avec l'exception est quand j'en relance un deuxième.

Je ne comprends pas pourquoi mon Functor est supprimer avant/durant l'exécution dans le thread. Je ne comprends pas non plus pourquoi il est supprimé aussi à la fin.

Comment puis je faire pour initialiser un objet et ensuite lancer une de ses functions dans un thread? Ce n'est pas comme cela en utilisant un Functor?

[Iradrille]

Hello,

Voici mon code:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

std::thread t(*session);

Tu provoques une copie ici car Session n'est pas movable ce qui génère un appel au destructeur.

Tu peux corriger ça soit en copiant le pointeur, soit en rendant Session moveable.

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
std::thread t([session]() { (*session)(); });
 
// ou
class Session {
   Session(Session &&other):
      iSock(other.iSock), iId(other.iId)
   {
       other.iSock = nullptr;
       other.iId = -1;
   }
   // ...
};

edit : Si tu rend Session moveable, tu auras std::thread t(std::move(*session));. Ton pointeur ne sera plus valide et tu ne pourra donc pas garder un pointeur sur ta Session.

[ramislebob]

Merci pour ta réponse Iradrille.
Je débute avec C++11 et boost, je ne suis pas encore familier avec les move et les thread.
Je comprends maintenant ce qu'il se passe.
J'ai changé d'optique car je souhaite pouvoir garder un handle dans mon vector. (Au lieu d'un simple vector, j'utiliserai une factory pour créer les Session et les détruire)
Du coup, au lieu de lancer le thread depuis le main, mon objet session l'a en tant que membre et le lance lui même.

Un truc comme ça:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
 
#include <cstdlib>
#include <iostream>
#include <thread>
#include <utility>
#include <boost/asio.hpp>
#include <atomic>
#include <vector>
#include <memory>
 
using boost::asio::ip::tcp;
 
const int max_length = 1024;
 
class Session
{
	public:
		Session(tcp::socket *sock):
			iSock(sock), iId(sIds), iStopSession(false)
		{
			sIds++;
			std::cout << "Session[" << iId << "] created\n";
		}
 
		~Session() {
			iStopSession = true;
			iThread.join();
 
			if (iSock) {
				delete iSock;
				iSock = nullptr;
			}
			std::cout << "Session[" << iId << "] deleted\n";
		}
 
		tcp::socket *sock(){return iSock;}
 
 
		void startThread() {
			iThread = std::thread(&Session::run, this);
		}
 
	private:
		static std::atomic<unsigned int> sIds;
 
		tcp::socket *iSock;
 
		unsigned int iId;
 
		std::thread iThread;
 
		bool iStopSession;
 
 
		void run(){
			std::cout << "Session[" << iId << "] starting\n";
 
			do{
				char data[max_length];
				boost::system::error_code error;
 
				// Read on the socket
				size_t length = iSock->read_some(boost::asio::buffer(data), error);
 
				// Connection closed cleanly by peer.
				if (error == boost::asio::error::eof) {
					std::cout << "Session[" << iId << "] client closed socket\n";
					iStopSession = true;
				}
				// Error during reading
				else if (error) {
					std::cerr << "Session[" << iId << "] error read socket:" << error.message() << std::endl;
					iStopSession = true;
				}
 
				// Reading OK
				else {
 
					boost::asio::write(*iSock, boost::asio::buffer(data, length), error);
					if (error) {
						std::cerr << "Session[" << iId << "] error writting socket:" << error.message() << std::endl;
						iStopSession = true;
					}
				}
 
			} while (!iStopSession);
 
			std::cout << "Session[" << iId << "]  Finished\n";
		}
};
 
std::atomic<unsigned int> Session::sIds{0};
 
 
std::vector<std::shared_ptr<Session> > vSessions{};
 
void signal_handler(
    const boost::system::error_code& error,
    int signal_number)
{
	std::cout << "signal_handler " << signal_number << std::endl;
	if (!error)
	{
		vSessions.clear();
		return;
	}
	std::cerr<<"[signal_handler] error: " << error.message() << std::endl;
}
 
 
void intercept_signals(){
	boost::asio::io_service io_service;
	boost::asio::signal_set sig(io_service, SIGINT, SIGTERM);
	sig.async_wait(signal_handler);
	io_service.run();
}
 
 
int main(int argc, char* argv[])
{
	if (argc != 2)
	{
		std::cerr << "Usage: echo_serv <port>\n";
		return 1;
	}
 
	// Proper cleaning on SIGTERM or SIGINT
	std::thread sig(intercept_signals);
 
 
	unsigned short port = std::atoi(argv[1]);
	boost::asio::io_service io_service;
 
	try
	{
		tcp::acceptor a(io_service, tcp::endpoint(tcp::v4(), port));
 
		for (;;)
		{
			tcp::socket *sock = new tcp::socket(io_service);
			a.accept(*sock);
			std::shared_ptr<Session> session(new Session(sock));
			vSessions.push_back(session);
			session->startThread();
		}
 
	}
	catch (std::exception& e)
	{
		std::cerr << "Exception: " << e.what() << "\n";
	}
 
	io_service.run();
	sig.join();
 
	return 0;
}