Replace dispatcher_test with a new demuxer_test unit test.

asio/src/Makefile.am

@@ -3,7 +3,6 @@ AUTOMAKE_OPTIONS = subdir-objects
 SUBDIRS = doc
 
 noinst_PROGRAMS = \
-	tests/dispatcher_test \
 	tests/dgram_echo_client_test \
 	tests/dgram_echo_server_test \
 	tests/echo_client_test \
@@ -16,6 +15,7 @@ noinst_PROGRAMS = \
 	tests/tpc_echo_server_test \
 	tests/performance/client \
 	tests/performance/server \
+	tests/unit/demuxer_test \
 	tests/unit/error_handler_test \
 	tests/unit/fixed_buffer_test \
 	examples/chat/chat_client \
@@ -32,6 +32,7 @@ noinst_PROGRAMS = \
 	examples/tutorial/timer4/timer
 
 TESTS = \
+	tests/unit/demuxer_test \
 	tests/unit/error_handler_test \
 	tests/unit/fixed_buffer_test
 
@@ -41,7 +42,6 @@ noinst_HEADERS = \
 
 AM_CXXFLAGS = -I../include
 
-tests_dispatcher_test_SOURCES = tests/dispatcher_test.cpp
 tests_dgram_echo_server_test_SOURCES = tests/dgram_echo_server_test.cpp
 tests_dgram_echo_client_test_SOURCES = tests/dgram_echo_client_test.cpp
 tests_echo_client_test_SOURCES = tests/echo_client_test.cpp
@@ -54,6 +54,7 @@ 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
+tests_unit_demuxer_test_SOURCES = tests/unit/demuxer_test.cpp
 tests_unit_error_handler_test_SOURCES = tests/unit/error_handler_test.cpp
 tests_unit_fixed_buffer_test_SOURCES = tests/unit/fixed_buffer_test.cpp
 examples_chat_chat_client_SOURCES = examples/chat/chat_client.cpp

asio/src/Makefile.bor

@@ -7,7 +7,6 @@ LDFLAGS = -q -O2 -v -vi -y -a8 -b -Ve- -Vx- -tWM -tWR -tWC
 LIBS = ws2_32.lib
 
 all: \
-	tests\dispatcher_test.exe \
 	tests\dgram_echo_client_test.exe \
 	tests\dgram_echo_server_test.exe \
 	tests\echo_client_test.exe \
@@ -20,6 +19,7 @@ all: \
 	tests\tpc_echo_server_test.exe \
 	tests\performance\client.exe \
 	tests\performance\server.exe \
+	tests\unit\demuxer_test.exe \
 	tests\unit\error_handler_test.exe \
 	tests\unit\fixed_buffer_test.exe \
 	examples\chat\chat_client.exe \

asio/src/Makefile.mgw

@@ -7,7 +7,6 @@ LDFLAGS = -g -O3
 LIBS = -lws2_32
 
 TEST_EXES = \
-	tests/dispatcher_test.exe \
 	tests/dgram_echo_client_test.exe \
 	tests/dgram_echo_server_test.exe \
 	tests/echo_client_test.exe \
@@ -20,6 +19,7 @@ TEST_EXES = \
 	tests/tpc_echo_server_test.exe \
 	tests/performance/client.exe \
 	tests/performance/server.exe \
+	tests/unit/demuxer_test.exe \
 	tests/unit/error_handler_test.exe \
 	tests/unit/fixed_buffer_test.exe
 

asio/src/Makefile.msc

@@ -7,7 +7,6 @@ LDFLAGS = -nologo -O2 -GX -GR -Zi -MD
 LIBS = ws2_32.lib
 
 all: \
-	tests\dispatcher_test.exe \
 	tests\dgram_echo_client_test.exe \
 	tests\dgram_echo_server_test.exe \
 	tests\echo_client_test.exe \
@@ -20,6 +19,7 @@ all: \
 	tests\tpc_echo_server_test.exe \
 	tests\performance\client.exe \
 	tests\performance\server.exe \
+	tests\unit\demuxer_test.exe \
 	tests\unit\error_handler_test.exe \
 	tests\unit\fixed_buffer_test.exe \
 	examples\chat\chat_client.exe \
@@ -35,7 +35,6 @@ all: \
 	examples\tutorial\timer3\timer.exe \
 	examples\tutorial\timer4\timer.exe
 
-tests\dispatcher_test.exe: tests\dispatcher_test.obj
 tests\dgram_echo_client_test.exe: tests\dgram_echo_client_test.obj
 tests\dgram_echo_server_test.exe: tests\dgram_echo_server_test.obj
 tests\echo_client_test.exe: tests\echo_client_test.obj
@@ -48,6 +47,7 @@ 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
+tests\unit\demuxer_test.exe: tests\unit\demuxer_test.obj
 tests\unit\error_handler_test.exe: tests\unit\error_handler_test.obj
 tests\unit\fixed_buffer_test.exe: tests\unit\fixed_buffer_test.obj
 examples\chat\chat_client.exe: examples\chat\chat_client.obj

asio/src/tests/dispatcher_test.cpp

@@ -1,103 +0,0 @@
-#include "asio.hpp"
-#include "asio/detail/mutex.hpp"
-#include "asio/detail/thread.hpp"
-#include <iostream>
-#include <boost/bind.hpp>
-
-using namespace asio;
-
-void print(demuxer& d, int id, int sleep_time, detail::mutex& io_mutex)
-{
-  detail::mutex::scoped_lock lock(io_mutex);
-  std::cout << "Starting " << id << "\n";
-  lock.unlock();
-
-  timer t(d, timer::from_now, 5);
-  t.wait();
-
-  lock.lock();
-  std::cout << "Finished " << id << "\n";
-  lock.unlock();
-}
-
-void inner_print(int id, detail::mutex& io_mutex)
-{
-  detail::mutex::scoped_lock lock(io_mutex);
-  std::cout << "Nested " << id << "\n";
-}
-
-void outer_print(demuxer& d, int id, detail::mutex& io_mutex)
-{
-  detail::mutex::scoped_lock lock(io_mutex);
-  std::cout << "Starting " << id << "\n";
-  lock.unlock();
-
-  d.operation_immediate(boost::bind(inner_print, id, boost::ref(io_mutex)),
-      null_completion_context(), true);
-
-  lock.lock();
-  std::cout << "Finished " << id << "\n";
-  lock.unlock();
-}
-
-void post_events(demuxer& d, counting_completion_context& c1,
-    counting_completion_context& c2, detail::mutex& io_mutex)
-{
-  // Give all threads an opportunity to start.
-  timer t(d, timer::from_now, 2);
-  t.wait();
-
-  // Post a bunch of completions to run across the different threads.
-  d.operation_immediate(boost::bind(print, boost::ref(d), 1, 10,
-        boost::ref(io_mutex)), c1);
-  d.operation_immediate(boost::bind(print, boost::ref(d), 2, 5,
-        boost::ref(io_mutex)), c2);
-  d.operation_immediate(boost::bind(print, boost::ref(d), 3, 5,
-        boost::ref(io_mutex)), c1);
-  d.operation_immediate(boost::bind(print, boost::ref(d), 4, 5,
-        boost::ref(io_mutex)), c2);
-  d.operation_immediate(boost::bind(print, boost::ref(d), 5, 5,
-        boost::ref(io_mutex)), c1);
-  d.operation_immediate(boost::bind(outer_print, boost::ref(d), 6,
-        boost::ref(io_mutex)));
-}
-
-void do_dispatch(demuxer& d)
-{
-  counting_completion_context c1(2);
-  counting_completion_context c2(1);
-  detail::mutex io_mutex;
-
-  d.operation_immediate(boost::bind(post_events, boost::ref(d), boost::ref(c1),
-        boost::ref(c2), boost::ref(io_mutex)));
-
-  // Create more threads than the tasks can use, since they are limited by
-  // their completion_context counts.
-  detail::thread t1(boost::bind(&demuxer::run, &d));
-  detail::thread t2(boost::bind(&demuxer::run, &d));
-  detail::thread t3(boost::bind(&demuxer::run, &d));
-  detail::thread t4(boost::bind(&demuxer::run, &d));
-  detail::thread t5(boost::bind(&demuxer::run, &d));
-  t1.join();
-  t2.join();
-  t3.join();
-  t4.join();
-  t5.join();
-}
-
-int main()
-{
-  try
-  {
-    demuxer d;
-    do_dispatch(d);
-    d.reset();
-    do_dispatch(d);
-  }
-  catch (std::exception& e)
-  {
-    std::cerr << "Exception: " << e.what() << "\n";
-  }
-
-  return 0;
-}

asio/src/tests/unit/demuxer_test.cpp

@@ -0,0 +1,189 @@
+//
+// demuxer_test.hpp
+// ~~~~~~~~~~~~~~~~
+//
+// Copyright (c) 2003, 2004 Christopher M. Kohlhoff (chris@kohlhoff.com)
+//
+// Permission to use, copy, modify, distribute and sell this software and its
+// documentation for any purpose is hereby granted without fee, provided that
+// the above copyright notice appears in all copies and that both the copyright
+// notice and this permission notice appear in supporting documentation. This
+// software is provided "as is" without express or implied warranty, and with
+// no claim as to its suitability for any purpose.
+//
+
+#include <sstream>
+#include <boost/bind.hpp>
+#include "asio.hpp"
+#include "unit_test.hpp"
+
+using namespace asio;
+
+void increment(int* count)
+{
+  ++(*count);
+}
+
+void decrement_to_zero(demuxer* d, int* count)
+{
+  if (*count > 0)
+  {
+    --(*count);
+
+    int before_value = *count;
+    d->operation_immediate(boost::bind(decrement_to_zero, d, count));
+
+    // Completion cannot nest, so count value should remain unchanged.
+    UNIT_TEST_CHECK(*count == before_value);
+  }
+}
+
+void nested_decrement_to_zero(demuxer* d, int* count)
+{
+  if (*count > 0)
+  {
+    --(*count);
+
+    d->operation_immediate(boost::bind(nested_decrement_to_zero, d, count),
+        null_completion_context(), true);
+
+    // Completion is nested, so count value should now be zero.
+    UNIT_TEST_CHECK(*count == 0);
+  }
+}
+
+void sleep_increment(demuxer* d, int* count)
+{
+  timer t(*d, timer::from_now, 2);
+  t.wait();
+
+  ++(*count);
+}
+
+void start_sleep_increments(demuxer* d, int* count)
+{
+  // Give all threads a chance to start.
+  timer t(*d, timer::from_now, 2);
+  t.wait();
+
+  // Start three increments which cannot run in parallel.
+  counting_completion_context ctx1(1);
+  d->operation_immediate(boost::bind(sleep_increment, d, count), ctx1);
+  d->operation_immediate(boost::bind(sleep_increment, d, count), ctx1);
+  d->operation_immediate(boost::bind(sleep_increment, d, count), ctx1);
+}
+
+void demuxer_test()
+{
+  demuxer d;
+  int count = 0;
+
+  d.operation_immediate(boost::bind(increment, &count));
+
+  // No completions can be delivered until run() is called.
+  UNIT_TEST_CHECK(count == 0);
+
+  d.run();
+
+  // The run() call will not return until all operations have finished.
+  UNIT_TEST_CHECK(count == 1);
+
+  count = 0;
+  d.reset();
+  d.operation_immediate(boost::bind(increment, &count));
+  d.operation_immediate(boost::bind(increment, &count));
+  d.operation_immediate(boost::bind(increment, &count));
+  d.operation_immediate(boost::bind(increment, &count));
+  d.operation_immediate(boost::bind(increment, &count));
+
+  // No completions can be delivered until run() is called.
+  UNIT_TEST_CHECK(count == 0);
+
+  d.run();
+
+  // The run() call will not return until all operations have finished.
+  UNIT_TEST_CHECK(count == 5);
+
+  count = 0;
+  d.reset();
+  d.operation_started();
+  d.operation_immediate(boost::bind(&demuxer::interrupt, &d));
+  d.run();
+
+  // The only operation executed should have been to interrupt run().
+  UNIT_TEST_CHECK(count == 0);
+
+  d.reset();
+  d.operation_completed(boost::bind(increment, &count));
+
+  // No completions can be delivered until run() is called.
+  UNIT_TEST_CHECK(count == 0);
+
+  d.run();
+
+  // The run() call will not return until all operations have finished.
+  UNIT_TEST_CHECK(count == 1);
+
+  count = 10;
+  d.reset();
+  d.operation_immediate(boost::bind(decrement_to_zero, &d, &count));
+
+  // No completions can be delivered until run() is called.
+  UNIT_TEST_CHECK(count == 10);
+
+  d.run();
+
+  // The run() call will not return until all operations have finished.
+  UNIT_TEST_CHECK(count == 0);
+
+  count = 10;
+  d.reset();
+  d.operation_immediate(boost::bind(nested_decrement_to_zero, &d, &count));
+
+  // No completions can be delivered until run() is called.
+  UNIT_TEST_CHECK(count == 10);
+
+  d.run();
+
+  // The run() call will not return until all operations have finished.
+  UNIT_TEST_CHECK(count == 0);
+
+  count = 10;
+  d.reset();
+  d.operation_immediate(boost::bind(nested_decrement_to_zero, &d, &count),
+      null_completion_context(), true);
+
+  // No completions can be delivered until run() is called, even though nested
+  // delivery was specifically allowed in the previous call.
+  UNIT_TEST_CHECK(count == 10);
+
+  d.run();
+
+  // The run() call will not return until all operations have finished.
+  UNIT_TEST_CHECK(count == 0);
+
+  count = 0;
+  d.reset();
+  d.operation_immediate(boost::bind(start_sleep_increments, &d, &count));
+  detail::thread thread1(boost::bind(&demuxer::run, &d));
+  detail::thread thread2(boost::bind(&demuxer::run, &d));
+
+  // Check all events run one after another even though there are two threads.
+  timer timer1(d, timer::from_now, 3);
+  timer1.wait();
+  UNIT_TEST_CHECK(count == 0);
+  timer1.set(timer::from_existing, 2);
+  timer1.wait();
+  UNIT_TEST_CHECK(count == 1);
+  timer1.set(timer::from_existing, 2);
+  timer1.wait();
+  UNIT_TEST_CHECK(count == 2);
+
+  thread1.join();
+  thread2.join();
+
+  // The run() calls will not return until all operations have finished.
+  UNIT_TEST_CHECK(count == 3);
+}
+
+UNIT_TEST(demuxer_test)