Much more efficient use of threads to process requests

awwit · awwit · commit 7075307e9032 · 2014-11-15T23:40:35.000+03:00
Fixed: correct termination for Windows
diff --git a/httpserver.pro.user b/httpserver.pro.user
@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!DOCTYPE QtCreatorProject>
-<!-- Written by QtCreator 3.2.1, 2014-11-13T02:24:14. -->
+<!-- Written by QtCreator 3.2.1, 2014-11-15T23:29:46. -->
 <qtcreator>
  <data>
   <variable>EnvironmentId</variable>
diff --git a/httpserver/Server.cpp b/httpserver/Server.cpp
@@ -318,7 +318,7 @@ namespace HttpServer
 
 		auto it_mime = mimes_types.find(file_ext);
 
-		std::string file_mime_type = mimes_types.end() != it_mime ? it_mime->second : "application/octet-stream";
+        std::string file_mime_type = mimes_types.cend() != it_mime ? it_mime->second : "application/octet-stream";
 
 		std::string headers("HTTP/1.1 200 OK\r\n");
 		headers += "Content-Type: " + file_mime_type + "\r\n"
@@ -838,16 +838,57 @@ namespace HttpServer
 		return app_exit_code;
 	}
 
+	void Server::threadRequestCycle(std::queue<Socket> &sockets) const
+	{
+		while (true)
+		{
+			Socket clientSocket;
+
+			this->eventThreadCycle->wait();
+
+			if (false == process_flag)
+			{
+				break;
+			}
+
+			this->sockets_queue_mtx.lock();
+
+			if (sockets.size() )
+			{
+				clientSocket = std::move(sockets.front() );
+				sockets.pop();
+			}
+
+			if (sockets.empty() )
+			{
+				this->eventThreadCycle->reset();
+
+				this->eventNotFullQueue->notify();
+			}
+
+			this->sockets_queue_mtx.unlock();
+
+			if (clientSocket.is_open() )
+			{
+				++this->threads_working_count;
+
+				this->threadRequestProc(clientSocket);
+
+				--this->threads_working_count;
+			}
+		}
+	}
+
 	/**
 	 * Цикл обработки очереди запросов
 	 */
 	int Server::cycleQueue(std::queue<Socket> &sockets)
 	{
-		auto it_option = settings.find("threads_max_count");
+		auto it_option = this->settings.find("threads_max_count");
 
 		size_t threads_max_count = 0;
 
-        if (settings.cend() != it_option)
+		if (this->settings.cend() != it_option)
 		{
 			threads_max_count = std::strtoull(it_option->second.c_str(), nullptr, 10);
 		}
@@ -864,68 +905,40 @@ namespace HttpServer
 			threads_max_count *= 2;
 		}
 
-		std::function<int(Server *, Socket)> serverThreadRequestProc = std::mem_fn(&Server::threadRequestProc);
+		this->threads_working_count = 0;
+		this->eventThreadCycle = new Event(false, true);
+
+		std::function<void(Server *, std::queue<Socket> &)> serverThreadRequestCycle = std::mem_fn(&Server::threadRequestCycle);
 
 		std::vector<std::thread> active_threads;
 		active_threads.reserve(threads_max_count);
 
 		// For update applications modules
 		do
 		{
-			if (eventUpdateModule->notifed() )
+			if (this->eventUpdateModule->notifed() )
 			{
 				updateModules();
 			}
 
 			// Cycle creation threads applications requests
 			do
 			{
-				if (threads_max_count <= active_threads.size() )
-				{
-					size_t i = 0;
-
-					while (false == System::isDoneThread(active_threads[i].native_handle() ) )
-					{
-						if (++i == active_threads.size() )
-						{
-							std::this_thread::yield();
-							i = 0;
-						}
-					}
-				}
-
-				for (size_t i = 0; i != active_threads.size();)
-				{
-					auto th = active_threads.begin() + i;
-
-					if (System::isDoneThread(th->native_handle() ) )
-					{
-						th->join();
-						active_threads.erase(th);
-					}
-					else
-					{
-						++i;
-					}
-				}
-
-				while (active_threads.size() <= threads_max_count && sockets.empty() == false)
+				while (this->threads_working_count == active_threads.size() && active_threads.size() < threads_max_count && sockets.empty() == false)
 				{
-					active_threads.emplace_back(serverThreadRequestProc, this, std::move(sockets.front() ) );
-					sockets.pop();
+					active_threads.emplace_back(serverThreadRequestCycle, this, std::ref(sockets) );
 				}
 
-				if (false == eventNotFullQueue->notifed() )
-				{
-					eventNotFullQueue->notify();
-				}
+				this->eventThreadCycle->notify();
 
-				eventProcessQueue->wait();
+				this->eventProcessQueue->wait();
 			}
-			while (process_flag);
+			while (this->process_flag);
 
 			// Data clear
 
+			this->eventThreadCycle->notify();
+
 			if (false == active_threads.empty() )
 			{
 				// Join threads (wait completion)
@@ -937,9 +950,21 @@ namespace HttpServer
 				active_threads.clear();
 			}
 
-			eventNotFullQueue->notify();
+			this->eventNotFullQueue->notify();
 		}
-		while (eventUpdateModule->notifed() );
+		while (this->eventUpdateModule->notifed() );
+
+		if (false == this->server_sockets.empty() )
+		{
+			for (Socket &s : this->server_sockets)
+			{
+				s.close();
+			}
+
+			this->server_sockets.clear();
+		}
+
+		delete this->eventThreadCycle;
 
 		return 0;
 	}
@@ -1678,13 +1703,11 @@ namespace HttpServer
 		// Applications settings list
 		std::unordered_set<ServerApplicationSettings *> applications;
 		// Get full applications settings list
-		apps_tree.collectApplicationSettings(applications);
+		this->apps_tree.collectApplicationSettings(applications);
 
 		// Bind ports set
 		std::unordered_set<int> ports;
 
-		std::vector<Socket> server_sockets;
-
 		// Open applications sockets
 		for (auto &app : applications)
 		{
@@ -1703,7 +1726,7 @@ namespace HttpServer
 						{
 							sock.nonblock(true);
 
-							server_sockets.emplace_back(std::move(sock) );
+							this->server_sockets.emplace_back(std::move(sock) );
 
 							ports.emplace(port);
 						}
@@ -1724,29 +1747,31 @@ namespace HttpServer
 			}
 		}
 
-		if (server_sockets.empty() )
+		if (this->server_sockets.empty() )
 		{
 			std::cout << "Error: do not open any socket;" << std::endl;
 			return 2;
 		}
 
-		sockets_list.create(server_sockets.size() );
+		SocketList sockets_list;
+
+		sockets_list.create(this->server_sockets.size() );
 
-		for (auto &sock : server_sockets)
+		for (auto &sock : this->server_sockets)
 		{
 			sockets_list.addSocket(sock);
 		}
 
 		std::cout << "Log: start server cycle;" << std::endl << std::endl;
 
 		const size_t queue_max_length = 1024;
-		eventNotFullQueue = new Event(true, true);
-		eventProcessQueue = new Event();
-		eventUpdateModule = new Event(false, true);
+		this->eventNotFullQueue = new Event(true, true);
+		this->eventProcessQueue = new Event();
+		this->eventUpdateModule = new Event(false, true);
 
 		std::queue<Socket> sockets;
 
-		process_flag = true;
+		this->process_flag = true;
 
 		std::function<int(Server *, std::queue<Socket> &)> serverCycleQueue = std::mem_fn(&Server::cycleQueue);
 
@@ -1759,45 +1784,39 @@ namespace HttpServer
 		{
 			if (sockets_list.accept(client_sockets) )
 			{
+				this->sockets_queue_mtx.lock();
+
 				for (Socket &sock : client_sockets)
 				{
 					if (sock.is_open() )
 					{
 						sock.nonblock(true);
 						sockets.emplace(std::move(sock) );
+					}
+				}
 
-						if (sockets.size() <= queue_max_length)
-						{
-							eventNotFullQueue->reset();
-						}
+				this->sockets_queue_mtx.unlock();
 
-						eventProcessQueue->notify();
-					}
+				this->eventProcessQueue->notify();
+
+				if (sockets.size() >= queue_max_length)
+				{
+					this->eventNotFullQueue->reset();
 				}
 
 				client_sockets.clear();
 
-				eventNotFullQueue->wait();
+				this->eventNotFullQueue->wait();
 			}
 		}
-		while (process_flag || eventUpdateModule->notifed() );
+		while (this->process_flag || this->eventUpdateModule->notifed() );
 
-		eventProcessQueue->notify();
+		this->eventProcessQueue->notify();
 
 		threadQueue.join();
 
 		sockets_list.destroy();
 
-		if (server_sockets.size() )
-		{
-			for (Socket &s : server_sockets)
-			{
-				s.close();
-			}
-
-			server_sockets.clear();
-		}
-
 		clear();
 
 		std::cout << "Log: final server cycle;" << std::endl;
diff --git a/httpserver/Server.h b/httpserver/Server.h
@@ -8,6 +8,7 @@
 #include <string>
 #include <unordered_map>
 #include <csignal>
+#include <atomic>
 
 #include "SocketList.h"
 #include "DataVariantAbstract.h"
@@ -30,11 +31,15 @@ namespace HttpServer
 
 		ServerApplicationsTree apps_tree;
 
-		SocketList sockets_list;
+		std::vector<Socket> server_sockets;
 
 		Event *eventNotFullQueue;
 		Event *eventProcessQueue;
 		Event *eventUpdateModule;
+        Event *eventThreadCycle;
+
+        mutable std::atomic_size_t threads_working_count;
+        mutable std::mutex sockets_queue_mtx;
 
 		// Флаг, означающий - активированы ли главные циклы сервера
 		// (с помощью этого флага можно деактивировать циклы, чтобы завершить работу сервера)
@@ -44,7 +49,8 @@ namespace HttpServer
 	protected:
 		int cycleQueue(std::queue<Socket> &);
 		void sendStatus(const Socket &, const std::chrono::milliseconds &, const size_t) const;
-		int threadRequestProc(Socket) const;
+        int threadRequestProc(Socket) const;
+        void threadRequestCycle(std::queue<Socket> &) const;
 		int transferFilePart(const Socket &, const std::chrono::milliseconds &, const std::string &, const time_t, const size_t, const std::string &, const std::string &, const std::string &, const bool) const;
 		int transferFile(const Socket &, const std::chrono::milliseconds &, const std::string &, const std::unordered_map<std::string, std::string> &, const std::map<std::string, std::string> &, const std::string &, const bool) const;
 		bool parseIncomingVars(std::unordered_multimap<std::string, std::string> &, const std::string &) const;
diff --git a/httpserver/SignalsHandles.cpp b/httpserver/SignalsHandles.cpp
@@ -115,9 +115,10 @@ int bindSignalsHandles(HttpServer::Server *server)
 
 	HINSTANCE hInstance = GetModuleHandle(nullptr);
 
-	WNDCLASSEX wcex = {0};
+	WNDCLASSEX wcex = {
+		sizeof(WNDCLASSEX)
+	};
 
-	wcex.cbSize = sizeof(WNDCLASSEX);
 	wcex.lpfnWndProc = WndProc;
 	wcex.hInstance = hInstance;
 	wcex.lpszClassName = myWndClassName;
diff --git a/httpserver/SocketList.cpp b/httpserver/SocketList.cpp
@@ -138,7 +138,7 @@ namespace HttpServer
 		#elif POSIX
 			size_t count = ::epoll_wait(obj_list, epoll_events.data(), epoll_events.size(), ~0);
 
-			if ( (size_t)~0 == count)
+            if (std::numeric_limits<size_t>::max() == count)
 			{
 				return false;
 			}

Original file line number	Diff line number	Diff line change
`@@ -138,7 +138,7 @@ namespace HttpServer`
`138`	`138`	`#elif POSIX`
`139`	`139`	`size_t count = ::epoll_wait(obj_list, epoll_events.data(), epoll_events.size(), ~0);`
`140`	`140`
`141`		`- if ( (size_t)~0 == count)`
	`141`	`+ if (std::numeric_limits<size_t>::max() == count)`
`142`	`142`	`{`
`143`	`143`	`return false;`
`144`	`144`	`}`