diff --git a/SConstruct b/SConstruct index fb8df135a0..5bd61c172f 100644 --- a/SConstruct +++ b/SConstruct @@ -146,6 +146,7 @@ if env_cpp11['CXX'].startswith('g++'): print "C++11 build environment is not supported on this version of G++" elif env_cpp11['CXX'].startswith('clang++'): print "C++11 build environment enabled" + env.Append(CXXFLANGS = ['-stdlib=libc++'],LINKFLAGS=['-stdlib=libc++']) env_cpp11.Append(WSPP_CPP11_ENABLED = "true",CXXFLAGS = ['-std=c++0x','-stdlib=libc++'],LINKFLAGS = ['-stdlib=libc++'],TOOLSET = ['clang++'],CPPDEFINES = ['_WEBSOCKETPP_CPP11_STL_']) # look for optional second boostroot compiled with clang's libc++ STL library diff --git a/changelog.md b/changelog.md index 9703a31662..fdd91a5cfa 100644 --- a/changelog.md +++ b/changelog.md @@ -18,7 +18,7 @@ HEAD will not read more data from their socket, allowing TCP flow control to work without blocking the main thread. - Feature: Adds the ability to specify whether or not to use the `SO_REUSEADDR` - TCP socket option. The default for this value has been changed from `true` to + TCP socket option. The default for this value has been changed from `true` to `false`. - Feature: Adds the ability to specify a maximum message size. - Feature: Adds `close::status::get_string(...)` method to look up a human @@ -36,8 +36,10 @@ HEAD short reads and quasi-expected socket shutdown related errors will no longer be reported as unclean WebSocket shutdowns to the application. Information about them will remain in the info error channel for debugging purposes. -- Improvement: `start_accept` errors are now reported to the caller either via - an exception or an ec parameter. +- Improvement: `start_accept` and `listen` errors are now reported to the caller + either via an exception or an ec parameter. +- Improvement: Outgoing writes are now batched for improved message throughput + and reduced system call and TCP frame overhead. - Bug: Fix some cases of calls to empty lib::function objects. - Bug: Fix memory leak of connection objects due to cached handlers holding on to reference counted pointers. #310 Thank you otaras for reporting. diff --git a/examples/broadcast_server/broadcast_server.cpp b/examples/broadcast_server/broadcast_server.cpp index 29c18b6aa0..bdb01fe444 100644 --- a/examples/broadcast_server/broadcast_server.cpp +++ b/examples/broadcast_server/broadcast_server.cpp @@ -34,7 +34,8 @@ enum action_type { struct action { action(action_type t, connection_hdl h) : type(t), hdl(h) {} - action(action_type t, server::message_ptr m) : type(t), msg(m) {} + action(action_type t, connection_hdl h, server::message_ptr m) + : type(t), hdl(h), msg(m) {} action_type type; websocketpp::connection_hdl hdl; @@ -92,7 +93,7 @@ public: // queue message up for sending by processing thread unique_lock lock(m_action_lock); //std::cout << "on_message" << std::endl; - m_actions.push(action(MESSAGE,msg)); + m_actions.push(action(MESSAGE,hdl,msg)); lock.unlock(); m_action_cond.notify_one(); } diff --git a/websocketpp/common/cpp11.hpp b/websocketpp/common/cpp11.hpp index 7c1ac43388..4b980efdd3 100644 --- a/websocketpp/common/cpp11.hpp +++ b/websocketpp/common/cpp11.hpp @@ -41,9 +41,11 @@ #endif -#ifdef _WEBSOCKETPP_CPP11_STL_ - // This flag indicates that all of the C++11 language features are available - // to us. +#if defined(_WEBSOCKETPP_CPP11_STL_) || __cplusplus >= 201103L + // _WEBSOCKETPP_CPP11_STL_ is a flag from the build system that forces + // WebSocket++ into C++11 mode. __cplusplus is a define set by the compiler + // if it has full support for C++11 language features. If either are set use + // C++11 language features #ifndef _WEBSOCKETPP_NOEXCEPT_TOKEN_ #define _WEBSOCKETPP_NOEXCEPT_TOKEN_ noexcept #endif diff --git a/websocketpp/common/network.hpp b/websocketpp/common/network.hpp index 6a01f140cb..9ba8b538bc 100644 --- a/websocketpp/common/network.hpp +++ b/websocketpp/common/network.hpp @@ -29,7 +29,7 @@ #define WEBSOCKETPP_COMMON_NETWORK_HPP // For ntohs and htons -#if defined(WIN32) +#if defined(_WIN32) #include #else //#include diff --git a/websocketpp/common/platforms.hpp b/websocketpp/common/platforms.hpp index 4e031aa345..1196847b5a 100644 --- a/websocketpp/common/platforms.hpp +++ b/websocketpp/common/platforms.hpp @@ -33,7 +33,7 @@ * don't fit somewhere else better. */ -#if defined(WIN32) && !defined(NOMINMAX) +#if defined(_WIN32) && !defined(NOMINMAX) // don't define min and max macros that conflict with std::min and std::max #define NOMINMAX #endif diff --git a/websocketpp/common/stdint.hpp b/websocketpp/common/stdint.hpp index 233e36f5bc..a8ca1e3c52 100644 --- a/websocketpp/common/stdint.hpp +++ b/websocketpp/common/stdint.hpp @@ -32,7 +32,7 @@ #define __STDC_LIMIT_MACROS 1 #endif -#if WIN32 && (_MSC_VER < 1600) +#if defined (_WIN32) && defined (_MSC_VER) && (_MSC_VER < 1600) #include using boost::int8_t; diff --git a/websocketpp/config/core_client.hpp b/websocketpp/config/core_client.hpp index 51cfa85c79..1b2fae44aa 100644 --- a/websocketpp/config/core_client.hpp +++ b/websocketpp/config/core_client.hpp @@ -34,7 +34,11 @@ #include // Concurrency +#ifndef _WEBSOCKETPP_NO_THREADING_ #include +#else +#include +#endif // Transport #include @@ -68,7 +72,11 @@ struct core_client { typedef core_client type; // Concurrency policy +#ifndef _WEBSOCKETPP_NO_THREADING_ typedef websocketpp::concurrency::basic concurrency_type; +#else + typedef websocketpp::concurrency::none concurrency_type; +#endif // HTTP Parser Policies typedef http::parser::request request_type; diff --git a/websocketpp/connection.hpp b/websocketpp/connection.hpp index a3de8459ee..e2ebcd6ab8 100644 --- a/websocketpp/connection.hpp +++ b/websocketpp/connection.hpp @@ -1,5 +1,5 @@ /* - * Copyright (c) 2013, Peter Thorson. All rights reserved. + * Copyright (c) 2014, Peter Thorson. All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are met: @@ -1451,9 +1451,9 @@ private: */ std::vector m_send_buffer; - /// a pointer to hold on to the current message being written to keep it + /// a list of pointers to hold on to the messages being written to keep them /// from going out of scope before the write is complete. - message_ptr m_current_msg; + std::vector m_current_msgs; /// True if there is currently an outstanding transport write /** diff --git a/websocketpp/http/impl/request.hpp b/websocketpp/http/impl/request.hpp index 19f91d853a..4557df4b75 100644 --- a/websocketpp/http/impl/request.hpp +++ b/websocketpp/http/impl/request.hpp @@ -127,7 +127,7 @@ inline size_t request::consume(const char *buf, size_t len) { } } - begin = end+sizeof(header_delimiter)-1; + begin = end+(sizeof(header_delimiter)-1); } } diff --git a/websocketpp/impl/connection_impl.hpp b/websocketpp/impl/connection_impl.hpp index af4c74aef3..b5e24b710f 100644 --- a/websocketpp/impl/connection_impl.hpp +++ b/websocketpp/impl/connection_impl.hpp @@ -1,5 +1,5 @@ /* - * Copyright (c) 2013, Peter Thorson. All rights reserved. + * Copyright (c) 2014, Peter Thorson. All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are met: @@ -1565,43 +1565,77 @@ void connection::write_frame() { return; } - // Get the next message in the queue. This will return an empty - // message if the queue was empty. - m_current_msg = write_pop(); - - if (!m_current_msg) { - return; + // pull off all the messages that are ready to write. + // stop if we get a message marked terminal + message_ptr next_message = write_pop(); + while (next_message) { + m_current_msgs.push_back(next_message); + if (!next_message->get_terminal()) { + next_message = write_pop(); + } else { + next_message = message_ptr(); + } + } + + if (m_current_msgs.empty()) { + // there was nothing to send + return; + } else { + // At this point we own the next messages to be sent and are + // responsible for holding the write flag until they are + // successfully sent or there is some error + m_write_flag = true; } - - // At this point we own the next message to be sent and are - // responsible for holding the write flag until it is successfully - // sent or there is some error - m_write_flag = true; } - std::string const & header = m_current_msg->get_header(); - std::string const & payload = m_current_msg->get_payload(); - - m_send_buffer.push_back(transport::buffer(header.c_str(),header.size())); - m_send_buffer.push_back(transport::buffer(payload.c_str(),payload.size())); + typename std::vector::iterator it; + for (it = m_current_msgs.begin(); it != m_current_msgs.end(); ++it) { + std::string const & header = (*it)->get_header(); + std::string const & payload = (*it)->get_payload(); + m_send_buffer.push_back(transport::buffer(header.c_str(),header.size())); + m_send_buffer.push_back(transport::buffer(payload.c_str(),payload.size())); + } + // Print detailed send stats if those log levels are enabled if (m_alog.static_test(log::alevel::frame_header)) { if (m_alog.dynamic_test(log::alevel::frame_header)) { - std::stringstream s; - s << "Dispatching write with " << header.size() - << " header bytes and " << payload.size() - << " payload bytes" << std::endl; - m_alog.write(log::alevel::frame_header,s.str()); - m_alog.write(log::alevel::frame_header,"Header: "+utility::to_hex(header)); - } - } - if (m_alog.static_test(log::alevel::frame_payload)) { - if (m_alog.dynamic_test(log::alevel::frame_payload)) { - m_alog.write(log::alevel::frame_payload,"Payload: "+utility::to_hex(payload)); - } - } + std::stringstream general,header,payload; + + general << "Dispatching write containing " << m_current_msgs.size() + <<" message(s) containing "; + header << "Header Bytes: \n"; + payload << "Payload Bytes: \n"; + + size_t hbytes = 0; + size_t pbytes = 0; + + for (size_t i = 0; i < m_current_msgs.size(); i++) { + hbytes += m_current_msgs[i]->get_header().size(); + pbytes += m_current_msgs[i]->get_payload().size(); + + header << "[" << i << "] (" + << m_current_msgs[i]->get_header().size() << ") " + << utility::to_hex(m_current_msgs[i]->get_header()) << "\n"; + + if (m_alog.static_test(log::alevel::frame_payload)) { + if (m_alog.dynamic_test(log::alevel::frame_payload)) { + payload << "[" << i << "] (" + << m_current_msgs[i]->get_payload().size() << ") " + << utility::to_hex(m_current_msgs[i]->get_payload()) + << "\n"; + } + } + } + + general << hbytes << " header bytes and " << pbytes << " payload bytes"; + + m_alog.write(log::alevel::frame_header,general.str()); + m_alog.write(log::alevel::frame_header,header.str()); + m_alog.write(log::alevel::frame_payload,payload.str()); + } + } transport_con_type::async_write( m_send_buffer, @@ -1616,10 +1650,11 @@ void connection::handle_write_frame(lib::error_code const & ec) m_alog.write(log::alevel::devel,"connection handle_write_frame"); } - bool terminal = m_current_msg->get_terminal(); + bool terminal = m_current_msgs.back()->get_terminal(); m_send_buffer.clear(); - m_current_msg.reset(); + m_current_msgs.clear(); + // TODO: recycle instead of deleting if (ec) { log_err(log::elevel::fatal,"handle_write_frame",ec); diff --git a/websocketpp/transport/asio/endpoint.hpp b/websocketpp/transport/asio/endpoint.hpp index 51b1f5edf3..b3cad3a3b2 100644 --- a/websocketpp/transport/asio/endpoint.hpp +++ b/websocketpp/transport/asio/endpoint.hpp @@ -123,7 +123,7 @@ public: : m_io_service(src.m_io_service) , m_external_io_service(src.m_external_io_service) , m_acceptor(src.m_acceptor) - , m_listen_backlog(0) + , m_listen_backlog(boost::asio::socket_base::max_connections) , m_reuse_addr(src.m_reuse_addr) , m_state(src.m_state) { @@ -145,7 +145,7 @@ public: rhs.m_io_service = NULL; rhs.m_external_io_service = false; rhs.m_acceptor = NULL; - rhs.m_listen_backlog = 0; + rhs.m_listen_backlog = boost::asio::socket_base::max_connections; rhs.m_state = UNINITIALIZED; } return *this; @@ -342,16 +342,25 @@ public: m_alog->write(log::alevel::devel,"asio::listen"); - m_acceptor->open(ep.protocol()); - m_acceptor->set_option(boost::asio::socket_base::reuse_address(m_reuse_addr)); - m_acceptor->bind(ep); - if (m_listen_backlog == 0) { - m_acceptor->listen(); - } else { - m_acceptor->listen(m_listen_backlog); + boost::system::error_code bec; + + m_acceptor->open(ep.protocol(),bec); + if (!bec) { + m_acceptor->set_option(boost::asio::socket_base::reuse_address(m_reuse_addr),bec); + } + if (!bec) { + m_acceptor->bind(ep,bec); + } + if (!bec) { + m_acceptor->listen(m_listen_backlog,bec); + } + if (bec) { + log_err(log::elevel::info,"asio listen",bec); + ec = make_error_code(error::pass_through); + } else { + m_state = LISTENING; + ec = lib::error_code(); } - m_state = LISTENING; - ec = lib::error_code(); } /// Set up endpoint for listening manually diff --git a/websocketpp/uri.hpp b/websocketpp/uri.hpp index 07ed32ebf2..b0895d581b 100644 --- a/websocketpp/uri.hpp +++ b/websocketpp/uri.hpp @@ -290,10 +290,10 @@ public: * @return query portion of the URI. */ std::string get_query() const { - std::size_t found = m_resource.find('?'); - if (found != std::string::npos) { - return m_resource.substr(found + 1); - } else { + std::size_t found = m_resource.find('?'); + if (found != std::string::npos) { + return m_resource.substr(found + 1); + } else { return ""; } }