ARCHIVE/rippled
1
0
Fork 0
mirror of https://github.com/XRPLF/rippled.git synced 2026-04-29 15:37:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into exception_refactor

Browse Source
This commit is contained in:
Peter Thorson
2014-03-25 07:22:54 -05:00
parent 1f73456e25 002d2d0007
commit b92a2b7f61
13 changed files with 120 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
SConstruct
View File

@@ -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

8
changelog.md
View File

@@ -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.

5
examples/broadcast_server/broadcast_server.cpp
View File

@@ -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<mutex> 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();
}

8
websocketpp/common/cpp11.hpp
View File

@@ -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

2
websocketpp/common/network.hpp
View File

@@ -29,7 +29,7 @@
#define WEBSOCKETPP_COMMON_NETWORK_HPP
// For ntohs and htons
#if defined(WIN32)
#if defined(_WIN32)
#include <winsock2.h>
#else
//#include <arpa/inet.h>

2
websocketpp/common/platforms.hpp
View File

@@ -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

2
websocketpp/common/stdint.hpp
View File

@@ -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 <boost/cstdint.hpp>
using boost::int8_t;

8
websocketpp/config/core_client.hpp
View File

@@ -34,7 +34,11 @@
#include <websocketpp/common/stdint.hpp>
// Concurrency
#ifndef _WEBSOCKETPP_NO_THREADING_
#include <websocketpp/concurrency/basic.hpp>
#else
#include <websocketpp/concurrency/none.hpp>
#endif
// Transport
#include <websocketpp/transport/iostream/endpoint.hpp>
@@ -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;

6
websocketpp/connection.hpp
View File

@@ -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<transport::buffer> 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<message_ptr> m_current_msgs;
/// True if there is currently an outstanding transport write
/**

2
websocketpp/http/impl/request.hpp
View File

@@ -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);
}
}

99
websocketpp/impl/connection_impl.hpp
View File

@@ -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<config>::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<message_ptr>::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<config>::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);

31
websocketpp/transport/asio/endpoint.hpp
View File

@@ -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

8
websocketpp/uri.hpp
View File

@@ -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 "";
}
}