merged 2008-03-30T18:02:36Z!davewx7@gmail.com from trunk...

...(added 'netstats' query command to wesnothd to query for current
network transfer info)

src/network.cpp

@@ -223,6 +223,11 @@ void error::disconnect()
 	if(socket) network::disconnect(socket);
 }
 
+pending_statistics get_pending_stats()
+{
+	return network_worker_pool::get_pending_stats();
+}
+
 manager::manager(size_t min_threads, size_t max_threads) : free_(true)
 {
 	// If the network is already being managed

src/network.hpp

@@ -34,6 +34,13 @@ namespace threading
 
 namespace network {
 
+struct pending_statistics {
+	int npending_sends;
+	int nbytes_pending_sends;
+};
+
+pending_statistics get_pending_stats();
+
 // A network manager must be created before networking can be used.
 // It must be destroyed only after all networking activity stops.
 

src/network_worker.cpp

@@ -101,7 +101,6 @@ unsigned int waiting_threads[NUM_SHARDS];
 size_t min_threads = 0;
 size_t max_threads = 0;
 
-
 int get_shard(TCPsocket sock) { return intptr_t(sock)%NUM_SHARDS; }
 
 struct buffer {
@@ -130,7 +129,7 @@ struct buffer {
 
 bool managed = false, raw_data_only = false;
 typedef std::vector< buffer* > buffer_set;
-buffer_set bufs[NUM_SHARDS];
+buffer_set outgoing_bufs[NUM_SHARDS];
 
 struct schema_pair
 {
@@ -439,7 +438,7 @@ static int process_queue(void* shard_num)
 			waiting_threads[shard]++;
 			for(;;) {
 
-				buffer_set::iterator itor = bufs[shard].begin(), itor_end = bufs[shard].end();
+				buffer_set::iterator itor = outgoing_bufs[shard].begin(), itor_end = outgoing_bufs[shard].end();
 				for(; itor != itor_end; ++itor) {
 					socket_state_map::iterator lock_it = sockets_locked[shard].find((*itor)->sock);
 					assert(lock_it != sockets_locked[shard].end());
@@ -447,7 +446,7 @@ static int process_queue(void* shard_num)
 						lock_it->second = SOCKET_LOCKED;
 						sent_buf = *itor;
 						sock = sent_buf->sock;
-						bufs[shard].erase(itor);
+						outgoing_bufs[shard].erase(itor);
 						break;
 					}
 				}
@@ -625,6 +624,22 @@ manager::~manager()
 	}
 }
 
+network::pending_statistics get_pending_stats()
+{
+	network::pending_statistics stats;
+	stats.npending_sends = 0;
+	stats.nbytes_pending_sends = 0;
+	for(int shard = 0; shard != NUM_SHARDS; ++shard) {
+		const threading::lock lock(*shard_mutexes[shard]);
+		stats.npending_sends += outgoing_bufs[shard].size();
+		for(buffer_set::const_iterator i = outgoing_bufs[shard].begin(); i != outgoing_bufs[shard].end(); ++i) {
+			stats.nbytes_pending_sends += (*i)->raw_buffer.size();
+		}
+	}
+
+	return stats;
+}
+
 void set_raw_data_only()
 {
 	raw_data_only = true;
@@ -698,7 +713,7 @@ void queue_raw_data(TCPsocket sock, const char* buf, int len)
 	make_network_buffer(buf, len, queued_buf->raw_buffer);
 	const int shard = get_shard(sock);
 	const threading::lock lock(*shard_mutexes[shard]);
-	bufs[shard].push_back(queued_buf);
+	outgoing_bufs[shard].push_back(queued_buf);
 	socket_state_map::const_iterator i = sockets_locked[shard].insert(std::pair<TCPsocket,SOCKET_STATE>(sock,SOCKET_READY)).first;
 	if(i->second == SOCKET_READY || i->second == SOCKET_ERRORED) {
 		cond[shard]->notify_one();
@@ -717,7 +732,7 @@ void queue_data(TCPsocket sock,const  config& buf, const bool gzipped)
 		const int shard = get_shard(sock);
 		const threading::lock lock(*shard_mutexes[shard]);
 
-		bufs[shard].push_back(queued_buf);
+		outgoing_bufs[shard].push_back(queued_buf);
 
 		socket_state_map::const_iterator i = sockets_locked[shard].insert(std::pair<TCPsocket,SOCKET_STATE>(sock,SOCKET_READY)).first;
 		if(i->second == SOCKET_READY || i->second == SOCKET_ERRORED) {
@@ -735,11 +750,11 @@ void remove_buffers(TCPsocket sock)
 {
 	{
 		const int shard = get_shard(sock);
-		for(buffer_set::iterator i = bufs[shard].begin(); i != bufs[shard].end();) {
+		for(buffer_set::iterator i = outgoing_bufs[shard].begin(); i != outgoing_bufs[shard].end();) {
 			if ((*i)->sock == sock)
 			{
 				buffer* buf = *i;
-				i = bufs[shard].erase(i);
+				i = outgoing_bufs[shard].erase(i);
 				delete buf;
 			}
 			else

src/network_worker.hpp

@@ -36,6 +36,8 @@ private:
 	bool active_;
 };
 
+network::pending_statistics get_pending_stats();
+
 void set_raw_data_only();
 
 //! Function to asynchronously received data to the given socket.

src/server/server.cpp

@@ -816,7 +816,7 @@ void server::process_query(const network::connection sock,
 	} else if (command == "status") {
 		response << process_command(command.to_string() + " " + pl->second.name());
 	} else if (command == "status " + pl->second.name() || command == "metrics"
-	|| command == "motd" || command == "wml") {
+	|| command == "motd" || command == "wml" || command == "netstats") {
 		response << process_command(command.to_string());
 	} else if (command == admin_passwd_) {
 		LOG_SERVER << "New Admin recognized:" << "\tIP: "
@@ -842,7 +842,7 @@ std::string server::process_command(const std::string& query) {
 	std::string parameters = (i == query.end() ? "" : std::string(i+1,query.end()));
 	utils::strip(parameters);
 	const std::string& help_msg = "Available commands are: (k)ban(s) [<mask>],"
-			"kick <mask>, help, metrics, (lobby)msg <message>, motd [<message>],"
+			"kick <mask>, help, metrics, netstats, (lobby)msg <message>, motd [<message>],"
 			"status [<mask>], unban <ipmask>";
 	if (command == "shut_down") {
 		throw network::error("shut down");
@@ -854,6 +854,11 @@ std::string server::process_command(const std::string& query) {
 		"Number of users in the lobby = " << lobby_.nobservers() << "\n";
 	} else if (command == "wml") {
 		out << simple_wml::document::stats();
+	} else if (command == "netstats") {
+		network::pending_statistics stats = network::get_pending_stats();
+		out << "Network stats:\nPending send buffers: "
+		    << stats.npending_sends << "\nBytes in buffers: "
+			<< stats.nbytes_pending_sends << "\n";
 	} else if (command == "msg" || command == "lobbymsg") {
 		if (parameters == "") {
 			return "You must type a message.";