[WIP] Stability: Add stability test example and start multithreaded fixes

src/examples/CMakeLists.txt

@@ -28,3 +28,4 @@ ADD_SUBDIRECTORY (query_params)
 ADD_SUBDIRECTORY (list_endpoints)
 ADD_SUBDIRECTORY (request_logger_plugin)
 ADD_SUBDIRECTORY (rest_resource)
+ADD_SUBDIRECTORY (stability)

src/examples/stability/CMakeLists.txt

@@ -0,0 +1,36 @@
+# Copyright (C) 2014 MediaSift Ltd.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+#
+# Locate project sources
+#
+FILE (GLOB_RECURSE eg_stability_SRCS RELATIVE ${CMAKE_CURRENT_SOURCE_DIR} ${CMAKE_CURRENT_SOURCE_DIR}/*.cpp ${CMAKE_CURRENT_SOURCE_DIR}/*.hpp)
+
+#
+# Configure common project settings
+#
+SET (eg_stability_LIBS ${PROJECT_NAME} ${Boost_LIBRARIES} ${CMAKE_THREAD_LIBS_INIT})
+
+#
+# Executable build rules
+#
+ADD_EXECUTABLE (eg_stability ${eg_stability_SRCS})
+TARGET_LINK_LIBRARIES (eg_stability ${eg_stability_LIBS})
+INSTALL (TARGETS eg_stability DESTINATION bin)

src/examples/stability/main.cpp

@@ -0,0 +1,70 @@
+/*
+ * Copyright (C) 2014 MediaSift Ltd.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+#include <served/served.hpp>
+#include <unistd.h>
+
+/* stability example
+ *
+ * This example is a basic stability test of served run in non blocking way.
+ */
+void test(bool stop)
+{
+	served::multiplexer mux;
+
+	mux.handle("/hello")
+		.get([](served::response & res, const served::request &) {
+			res << "Hello world";
+		});
+
+	std::cout << "Try this example with:" << std::endl;
+	std::cout << " curl http://localhost:8123/hello" << std::endl;
+
+	served::net::server server("127.0.0.1", "8123", mux, false);
+	server.run(10, false); // Run with a pool of 10 threads (not blocking)
+
+	std::cout << "Time to stop the server" << std::endl;
+	if (stop) {
+		server.stop();
+	}
+}
+
+int main(int, char const**)
+{
+	for (size_t i = 0; i < 10000; ++i) {
+		std::cout << std::endl;
+		std::cout << "Performing test " << i << " (with stop()) :" << std::endl;
+		std::cout << std::endl;
+		test(true);
+	}
+
+	for (size_t i = 0; i < 10000; ++i) {
+		std::cout << std::endl;
+		std::cout << "Performing test " << i << " (without stop()) :" << std::endl;
+		std::cout << std::endl;
+		test(false);
+	}
+
+	std::cout << "Successfully performed the stability tests" << std::endl;
+	std::cout << std::endl;
+	return 0;
+}

src/served/net/server.cpp

@@ -22,7 +22,6 @@
 
 #include <signal.h>
 #include <utility>
-#include <thread>
 
 #include <served/net/server.hpp>
 
@@ -71,6 +70,30 @@ server::server( const std::string & address
 	do_accept();
 }
 
+server::~server()
+{
+	_acceptor.close();
+	_connection_manager.stop_all();
+
+	if ( ! _io_service.stopped() )
+	{
+		_io_service.stop();
+	}
+
+	if ( _threads.size() > 0 ) {
+		for ( auto & thread : _threads )
+		{
+			if ( thread.joinable() )
+			{
+				thread.join();
+			}
+		}
+
+		std::this_thread::sleep_for(std::chrono::milliseconds(1));
+		_threads.clear();
+	}
+}
+
 void
 server::run(int n_threads /* = 1 */, bool block /* = true */)
 {
@@ -82,14 +105,13 @@ server::run(int n_threads /* = 1 */, bool block /* = true */)
 	 */
 	if ( n_threads > 1 )
 	{
-		std::vector<std::thread> v_threads;
-		for ( int i = 0; i < n_threads; i++ )
+		for ( int i = 0; i < 1; i++ )
 		{
-			v_threads.push_back(std::thread([this](){
-				_io_service.run();
-			}));
+			_threads.push_back(std::thread(
+					std::bind(static_cast<size_t(boost::asio::io_service::*)()>(
+						&boost::asio::io_service::run), std::ref(_io_service))));
 		}
-		for ( auto & thread : v_threads )
+		for ( auto & thread : _threads )
 		{
 			if ( block )
 			{

src/served/net/server.hpp

@@ -25,6 +25,7 @@
 
 #include <boost/asio.hpp>
 #include <string>
+#include <thread>
 #include <served/net/connection_manager.hpp>
 #include <served/multiplexer.hpp>
 
@@ -49,6 +50,7 @@ class server
 	int                            _read_timeout;
 	int                            _write_timeout;
 	size_t                         _req_max_bytes;
+	std::vector<std::thread>       _threads;
 
 public:
 	server(const server&) = delete;
@@ -70,14 +72,19 @@ class server
 	               , multiplexer       & mux
 	               , bool              register_signals = true );
 
+	/*
+	 * Destroys the server.
+	 */
+	~server();
+
 	/*
 	 * A call that prompts the server into listening for HTTP requests.
 	 *
 	 * This call accepts a value for how large the thread pool should be for distributing requests
 	 * and another param which defines the blocking nature.
 	 *
 	 * @param n_threads the number of threads to pool for request handling
-	 * @param block if n_threads > 0, defines whether this operation is blocking or not
+	 * @param block if n_threads > 1, defines whether this operation is blocking or not
 	 */
 	void run(int n_threads = 1, bool block = true);