Added new tests for measuring performance.

asio/src/Makefile.am

@@ -15,6 +15,8 @@ noinst_PROGRAMS = \
 	tests/timed_dgram_recv_test \
 	tests/timer_test \
 	tests/tpc_echo_server_test \
+	tests/performance/client \
+	tests/performance/server \
 	examples/chat/chat_client \
 	examples/chat/chat_server \
 	examples/echo/async_tcp_echo_server \
@@ -45,6 +47,8 @@ tests_timed_connect_test_SOURCES = tests/timed_connect_test.cpp
 tests_timed_dgram_recv_test_SOURCES = tests/timed_dgram_recv_test.cpp
 tests_timer_test_SOURCES = tests/timer_test.cpp
 tests_tpc_echo_server_test_SOURCES = tests/tpc_echo_server_test.cpp
+tests_performance_client_SOURCES = tests/performance/client.cpp
+tests_performance_server_SOURCES = tests/performance/server.cpp
 examples_chat_chat_client_SOURCES = examples/chat/chat_client.cpp
 examples_chat_chat_server_SOURCES = examples/chat/chat_server.cpp
 examples_echo_async_tcp_echo_server_SOURCES = examples/echo/async_tcp_echo_server.cpp

asio/src/Makefile.bor

@@ -19,6 +19,8 @@ all: \
 	tests\timed_dgram_recv_test.exe \
 	tests\timer_test.exe \
 	tests\tpc_echo_server_test.exe \
+	tests\performance\client.exe \
+	tests\performance\server.exe \
 	examples\chat\chat_client.exe \
 	examples\chat\chat_server.exe \
 	examples\echo\async_tcp_echo_server.exe \

asio/src/Makefile.mgw

@@ -18,7 +18,9 @@ TEST_EXES = \
 	tests/timed_connect_test.exe \
 	tests/timed_dgram_recv_test.exe \
 	tests/timer_test.exe \
-	tests/tpc_echo_server_test.exe
+	tests/tpc_echo_server_test.exe \
+	tests/performance/client.exe \
+	tests/performance/server.exe
 
 EXAMPLE_EXES = \
 	examples\chat\chat_client.exe \

asio/src/Makefile.msc

@@ -19,6 +19,8 @@ all: \
 	tests\timed_dgram_recv_test.exe \
 	tests\timer_test.exe \
 	tests\tpc_echo_server_test.exe \
+	tests\performance\client.exe \
+	tests\performance\server.exe \
 	examples\chat\chat_client.exe \
 	examples\chat\chat_server.exe \
 	examples\echo\async_tcp_echo_server.exe \
@@ -44,6 +46,8 @@ tests\timed_connect_test.exe: tests\timed_connect_test.obj
 tests\timed_dgram_recv_test.exe: tests\timed_dgram_recv_test.obj
 tests\timer_test.exe: tests\timer_test.obj
 tests\tpc_echo_server_test.exe: tests\tpc_echo_server_test.obj
+tests\performance\client.exe: tests\performance\client.obj
+tests\performance\server.exe: tests\performance\server.obj
 examples\chat\chat_client.exe: examples\chat\chat_client.obj
 examples\chat\chat_server.exe: examples\chat\chat_server.obj
 examples\echo\async_tcp_echo_server.exe: examples\echo\async_tcp_echo_server.obj

asio/src/tests/performance/.cvsignore

@@ -0,0 +1,8 @@
+.deps
+.dirstamp
+*.exe
+client
+server
+*.ilk
+*.pdb
+*.tds

asio/src/tests/performance/client.cpp

@@ -0,0 +1,256 @@
+#include "asio.hpp"
+#include <algorithm>
+#include <boost/bind.hpp>
+#include <boost/mem_fn.hpp>
+#include <iostream>
+#include <list>
+#include <string>
+
+using namespace asio;
+
+class stats
+{
+public:
+  stats()
+    : mutex_(),
+      total_bytes_sent_(0),
+      total_bytes_recvd_(0)
+  {
+  }
+
+  void add(size_t bytes_sent, size_t bytes_recvd)
+  {
+    detail::mutex::scoped_lock lock(mutex_);
+    total_bytes_sent_ += bytes_sent;
+    total_bytes_recvd_ += bytes_recvd;
+  }
+
+  void print()
+  {
+    detail::mutex::scoped_lock lock(mutex_);
+    std::cout << total_bytes_sent_ << " total bytes sent\n";
+    std::cout << total_bytes_recvd_ << " total bytes received\n";
+  }
+
+private:
+  detail::mutex mutex_;
+  size_t total_bytes_sent_;
+  size_t total_bytes_recvd_;
+};
+
+class session
+{
+public:
+  session(demuxer& d, size_t block_size, stats& s)
+    : demuxer_(d),
+      context_(1),
+      socket_(d),
+      block_size_(block_size),
+      recv_data_(new char[block_size]),
+      send_data_(new char[block_size]),
+      unsent_count_(0),
+      bytes_sent_(0),
+      bytes_recvd_(0),
+      stats_(s)
+  {
+    for (size_t i = 0; i < block_size_; ++i)
+      send_data_[i] = i % 128;
+  }
+
+  ~session()
+  {
+    stats_.add(bytes_sent_, bytes_recvd_);
+
+    delete[] recv_data_;
+    delete[] send_data_;
+  }
+
+  stream_socket& socket()
+  {
+    return socket_;
+  }
+
+  void start()
+  {
+    ++unsent_count_;
+    async_send_n(socket_, send_data_, block_size_,
+        boost::bind(&session::handle_send, this, _1, _2, _3), context_);
+    socket_.async_recv(recv_data_, block_size_,
+        boost::bind(&session::handle_recv, this, _1, _2), context_);
+  }
+
+  void stop()
+  {
+    demuxer_.operation_immediate(boost::bind(&stream_socket::close, &socket_),
+        context_);
+  }
+
+  void handle_recv(const socket_error& error, size_t length)
+  {
+    if (!error && length > 0)
+    {
+      bytes_recvd_ += length;
+
+      ++unsent_count_;
+      if (unsent_count_ == 1)
+      {
+        std::swap(recv_data_, send_data_);
+        async_send_n(socket_, send_data_, length,
+            boost::bind(&session::handle_send, this, _1, _2, _3), context_);
+        socket_.async_recv(recv_data_, block_size_,
+            boost::bind(&session::handle_recv, this, _1, _2), context_);
+      }
+    }
+  }
+
+  void handle_send(const socket_error& error, size_t length, size_t last_length)
+  {
+    if (!error && last_length > 0)
+    {
+      bytes_sent_ += length;
+
+      --unsent_count_;
+      if (unsent_count_ == 1)
+      {
+        std::swap(recv_data_, send_data_);
+        async_send_n(socket_, send_data_, length,
+            boost::bind(&session::handle_send, this, _1, _2, _3), context_);
+        socket_.async_recv(recv_data_, block_size_,
+            boost::bind(&session::handle_recv, this, _1, _2), context_);
+      }
+    }
+  }
+
+private:
+  demuxer& demuxer_;
+  counting_completion_context context_;
+  stream_socket socket_;
+  size_t block_size_;
+  char* recv_data_;
+  char* send_data_;
+  int unsent_count_;
+  size_t bytes_sent_;
+  size_t bytes_recvd_;
+  stats& stats_;
+};
+
+class client
+{
+public:
+  client(demuxer& d, const char* host, short port, size_t block_size,
+      size_t session_count, int timeout)
+    : demuxer_(d),
+      context_(1),
+      stop_timer_(d, timer::from_now, timeout),
+      connector_(d),
+      server_addr_(port, host),
+      block_size_(block_size),
+      max_session_count_(session_count),
+      sessions_(),
+      stats_()
+  {
+    session* new_session = new session(demuxer_, block_size, stats_);
+    connector_.async_connect(new_session->socket(), server_addr_,
+        boost::bind(&client::handle_connect, this, new_session, _1), context_);
+
+    stop_timer_.async_wait(boost::bind(&client::handle_timeout, this),
+        context_);
+  }
+
+  ~client()
+  {
+    while (!sessions_.empty())
+    {
+      delete sessions_.front();
+      sessions_.pop_front();
+    }
+
+    stats_.print();
+  }
+
+  void handle_timeout()
+  {
+    std::for_each(sessions_.begin(), sessions_.end(),
+        boost::mem_fn(&session::stop));
+  }
+
+  void handle_connect(session* new_session, const socket_error& error)
+  {
+    if (!error)
+    {
+      sessions_.push_back(new_session);
+      new_session->start();
+
+      if (sessions_.size() < max_session_count_)
+      {
+        new_session = new session(demuxer_, block_size_, stats_);
+        connector_.async_connect(new_session->socket(), server_addr_,
+            boost::bind(&client::handle_connect, this, new_session, _1),
+            context_);
+      }
+    }
+    else
+    {
+      delete new_session;
+    }
+  }
+
+private:
+  demuxer& demuxer_;
+  counting_completion_context context_;
+  timer stop_timer_;
+  socket_connector connector_;
+  inet_address_v4 server_addr_;
+  size_t block_size_;
+  size_t max_session_count_;
+  std::list<session*> sessions_;
+  stats stats_;
+};
+
+int main(int argc, char* argv[])
+{
+  try
+  {
+    if (argc != 7)
+    {
+      std::cerr << "Usage: client <host> <port> <threads> <blocksize> ";
+      std::cerr << "<sessions> <time>\n";
+      return 1;
+    }
+
+    using namespace std; // For atoi.
+    const char* host = argv[1];
+    short port = atoi(argv[2]);
+    int thread_count = atoi(argv[3]);
+    size_t block_size = atoi(argv[4]);
+    size_t session_count = atoi(argv[5]);
+    int timeout = atoi(argv[6]);
+
+    demuxer d;
+
+    client c(d, host, port, block_size, session_count, timeout);
+
+    std::list<detail::thread*> threads;
+    while (--thread_count > 0)
+    {
+      detail::thread* new_thread =
+        new detail::thread(boost::bind(&demuxer::run, &d));
+      threads.push_back(new_thread);
+    }
+
+    d.run();
+
+    while (!threads.empty())
+    {
+      threads.front()->join();
+      delete threads.front();
+      threads.pop_front();
+    }
+  }
+  catch (std::exception& e)
+  {
+    std::cerr << "Exception: " << e.what() << "\n";
+  }
+
+  return 0;
+}

asio/src/tests/performance/server.cpp

@@ -0,0 +1,170 @@
+#include "asio.hpp"
+#include <algorithm>
+#include <boost/bind.hpp>
+#include <iostream>
+#include <list>
+
+using namespace asio;
+
+class session
+{
+public:
+  session(demuxer& d, size_t block_size)
+    : context_(1),
+      socket_(d),
+      block_size_(block_size),
+      recv_data_(new char[block_size]),
+      send_data_(new char[block_size]),
+      unsent_count_(0),
+      op_count_(0)
+  {
+  }
+
+  ~session()
+  {
+    delete[] recv_data_;
+    delete[] send_data_;
+  }
+
+  stream_socket& socket()
+  {
+    return socket_;
+  }
+
+  void start()
+  {
+    ++op_count_;
+    socket_.async_recv(recv_data_, block_size_,
+        boost::bind(&session::handle_recv, this, _1, _2), context_);
+  }
+
+  void handle_recv(const socket_error& error, size_t length)
+  {
+    if (!error && length > 0)
+    {
+      ++unsent_count_;
+      if (unsent_count_ == 1)
+      {
+        op_count_ += 2;
+        std::swap(recv_data_, send_data_);
+        async_send_n(socket_, send_data_, length,
+            boost::bind(&session::handle_send, this, _1, _2, _3), context_);
+        socket_.async_recv(recv_data_, block_size_,
+            boost::bind(&session::handle_recv, this, _1, _2), context_);
+      }
+    }
+    else if (--op_count_ == 0)
+    {
+      delete this;
+    }
+  }
+
+  void handle_send(const socket_error& error, size_t length, size_t last_length)
+  {
+    if (!error && last_length > 0)
+    {
+      --unsent_count_;
+      if (unsent_count_ == 1)
+      {
+        op_count_ += 2;
+        std::swap(recv_data_, send_data_);
+        async_send_n(socket_, send_data_, length,
+            boost::bind(&session::handle_send, this, _1, _2, _3), context_);
+        socket_.async_recv(recv_data_, block_size_,
+            boost::bind(&session::handle_recv, this, _1, _2), context_);
+      }
+    }
+    else if (--op_count_ == 0)
+    {
+      delete this;
+    }
+  }
+
+private:
+  counting_completion_context context_;
+  stream_socket socket_;
+  size_t block_size_;
+  char* recv_data_;
+  char* send_data_;
+  int unsent_count_;
+  int op_count_;
+};
+
+class server
+{
+public:
+  server(demuxer& d, short port, size_t block_size)
+    : demuxer_(d),
+      acceptor_(d, inet_address_v4(port)),
+      block_size_(block_size)
+  {
+    session* new_session = new session(demuxer_, block_size_);
+    acceptor_.async_accept(new_session->socket(),
+        boost::bind(&server::handle_accept, this, new_session, _1));
+  }
+
+  void handle_accept(session* new_session, const socket_error& error)
+  {
+    if (!error)
+    {
+      new_session->start();
+      new_session = new session(demuxer_, block_size_);
+      acceptor_.async_accept(new_session->socket(),
+          boost::bind(&server::handle_accept, this, new_session, _1));
+
+    }
+    else
+    {
+      delete new_session;
+    }
+  }
+
+private:
+  demuxer& demuxer_;
+  socket_acceptor acceptor_;
+  size_t block_size_;
+};
+
+int main(int argc, char* argv[])
+{
+  try
+  {
+    if (argc != 4)
+    {
+      std::cerr << "Usage: server <port> <threads> <blocksize>\n";
+      return 1;
+    }
+
+    using namespace std; // For atoi.
+    short port = atoi(argv[1]);
+    int thread_count = atoi(argv[2]);
+    size_t block_size = atoi(argv[3]);
+
+    demuxer d;
+
+    server s(d, port, block_size);
+
+    std::list<detail::thread*> threads;
+    while (--thread_count > 0)
+    {
+      detail::thread* new_thread =
+        new detail::thread(boost::bind(&demuxer::run, &d));
+      threads.push_back(new_thread);
+    }
+
+    d.run();
+
+    while (!threads.empty())
+    {
+      threads.front()->join();
+      delete threads.front();
+      threads.pop_front();
+    }
+  }
+  catch (std::exception& e)
+  {
+    std::cerr << "Exception: " << e.what() << "\n";
+  }
+
+  return 0;
+}