[Boost-commit] svn:boost r64210 - in sandbox/SOC/2010/process: boost/process boost/process/detail libs/process/example

Subject: [Boost-commit] svn:boost r64210 - in sandbox/SOC/2010/process: boost/process boost/process/detail libs/process/example
From: boris_at_[hidden]
Date: 2010-07-20 20:37:11

Author: bschaeling
Date: 2010-07-20 20:37:10 EDT (Tue, 20 Jul 2010)
New Revision: 64210
URL: http://svn.boost.org/trac/boost/changeset/64210

Log:
Added support for asynchronous waiting on POSIX platforms
Text files modified: 
   sandbox/SOC/2010/process/boost/process/detail/basic_status_service.hpp |   130 ++++++++++++++++++++++++--------------- 
   sandbox/SOC/2010/process/boost/process/detail/posix_helpers.hpp        |     5 -                                       
   sandbox/SOC/2010/process/boost/process/detail/status_impl.hpp          |    91 +++++++++++++++++++--------             
   sandbox/SOC/2010/process/boost/process/operations.hpp                  |    26 ++++---                                 
   sandbox/SOC/2010/process/boost/process/process.hpp                     |     4                                         
   sandbox/SOC/2010/process/libs/process/example/read_from_child.cpp      |     8 -                                       
   sandbox/SOC/2010/process/libs/process/example/wait_child.cpp           |     6 -                                       
   7 files changed, 164 insertions(+), 106 deletions(-)

Modified: sandbox/SOC/2010/process/boost/process/detail/basic_status_service.hpp
==============================================================================
--- sandbox/SOC/2010/process/boost/process/detail/basic_status_service.hpp	(original)
+++ sandbox/SOC/2010/process/boost/process/detail/basic_status_service.hpp	2010-07-20 20:37:10 EDT (Tue, 20 Jul 2010)
@@ -34,6 +34,8 @@
 #include <algorithm>
 
 #if defined(BOOST_POSIX_API)
+#   include <sys/types.h>
+#   include <sys/wait.h>
 #elif defined(BOOST_WINDOWS_API)
 #   include <windows.h>
 #else
@@ -49,17 +51,18 @@
     : public boost::asio::detail::service_base<StatusImplementation>
 {
 public:
-
-
-    typedef boost::shared_ptr<StatusImplementation> implementation_type;
-
     explicit basic_status_service(boost::asio::io_service &io_service)
         : boost::asio::detail::service_base<StatusImplementation>(io_service),
-        run_(true),
         work_thread_(&basic_status_service<StatusImplementation>::work_thread, this)
+#if defined(BOOST_POSIX_API)
+        , interrupt_pid_(-1),
+        pids_(0)
+#elif defined(BOOST_WINDOWS_API)
+        , run_(true)
+#endif
     {
 #if defined(BOOST_WINDOWS_API)
-        handles_.push_back(::CreateEvent(NULL, FALSE, FALSE, NULL));
+        handles_.push_back(CreateEvent(NULL, FALSE, FALSE, NULL));
         if (handles_[0] == NULL)
             BOOST_PROCESS_THROW_LAST_SYSTEM_ERROR("CreateEvent() failed");
 #endif
@@ -70,14 +73,11 @@
         stop_work_thread();
         work_thread_.join();
 #if defined(BOOST_WINDOWS_API)
-        if (!::CloseHandle(handles_[0]))
-            BOOST_PROCESS_THROW_LAST_SYSTEM_ERROR("CloseHandle() failed");
-#elif defined(BOOST_POSIX_API)
-        if (::close(handles_[0]) == -1)
-            BOOST_PROCESS_THROW_LAST_SYSTEM_ERROR("close() failed");
+        CloseHandle(handles_[0]);
 #endif
     }
 
+    typedef boost::shared_ptr<StatusImplementation> implementation_type;
 
     void construct(implementation_type &impl)
     {
@@ -88,43 +88,45 @@
 
     void destroy(implementation_type &impl)
     {
-        // impl->destroy();
+        boost::unique_lock<boost::mutex> lock(work_thread_mutex_);
+        std::vector<implementation_type>::iterator it = std::find(impls_.begin(), impls_.end(), impl);
+        if (it != impls_.end())
+            impls_.erase(it);
         impl.reset();
-        // remove impl from impls_
     }
 
     int wait(implementation_type &impl, pid_type pid)
     {
-        /*
         boost::system::error_code ec;
-        int exit_code = impl->wait(pid, ec);
+        int status = impl.wait(pid, ec);
         boost::asio::detail::throw_error(ec);
-        */
-        int exit_code = 0;
-        return exit_code;
+        return status;
     }
 
     template <typename Handler>
     void async_wait(implementation_type &impl, pid_type pid, Handler handler)
     {
-#if defined(BOOST_WINDOWS_API)
-        HANDLE handle = ::OpenProcess(SYNCHRONIZE | PROCESS_QUERY_INFORMATION, FALSE, pid);
+#if defined(BOOST_POSIX_API)
+        boost::unique_lock<boost::mutex> lock(work_thread_mutex_);
+        if (!work_)
+            work_.reset(new boost::asio::io_service::work(this->get_io_service()));
+        ++pids_;
+        impl->async_wait(pid, this->get_io_service().wrap(handler));
+#elif defined(BOOST_WINDOWS_API)
+        HANDLE handle = OpenProcess(SYNCHRONIZE | PROCESS_QUERY_INFORMATION, FALSE, pid);
         if (handle == NULL)
             BOOST_PROCESS_THROW_LAST_SYSTEM_ERROR("OpenProcess() failed");
-#elif defined(BOOST_POSIX_API)
-        behavior::stream::native_type handle = pid;
-#endif
         boost::unique_lock<boost::mutex> lock(work_thread_mutex_);
-        interrupt_work_thread();
-        work_thread_cond_.wait(work_thread_mutex_);
         if (!work_)
             work_.reset(new boost::asio::io_service::work(this->get_io_service()));
+        interrupt_work_thread();
+        work_thread_cond_.wait(work_thread_mutex_);
         handles_.push_back(handle);
         impl->async_wait(handle, this->get_io_service().wrap(handler));
         work_thread_cond_.notify_all();
+#endif
     }
 
-
 private:
     void shutdown_service()
     {
@@ -132,10 +134,31 @@
 
     void work_thread()
     {
-        while (running())
+        for (;;)
         {
-#if defined(BOOST_WINDOWS_API)
-            DWORD res = ::WaitForMultipleObjects(handles_.size(), &handles_[0], FALSE, INFINITE);
+#if defined(BOOST_POSIX_API)
+            int status;
+            pid_t pid = wait(&status);
+            if (pid == -1)
+            {
+                if (errno != EINTR)
+                    BOOST_PROCESS_THROW_LAST_SYSTEM_ERROR("wait(2) failed");
+            }
+            else if (interrupted(pid))
+            {
+                // On POSIX the only reason to interrupt is to break out.
+                break;
+            }
+            else
+            {
+                boost::unique_lock<boost::mutex> lock(work_thread_mutex_);
+                for (std::vector<implementation_type>::iterator it = impls_.begin(); it != impls_.end(); ++it)
+                    (*it)->complete(pid, status);
+                if (--pids_ == 0)
+                    work_.reset();
+            }
+#elif defined(BOOST_WINDOWS_API)
+            DWORD res = WaitForMultipleObjects(handles_.size(), &handles_[0], FALSE, INFINITE);
             if (res == WAIT_FAILED)
                 BOOST_PROCESS_THROW_LAST_SYSTEM_ERROR("WaitForMultipleObjects() failed");
             else if (res - WAIT_OBJECT_0 == 0)
@@ -150,7 +173,7 @@
             {
                 HANDLE handle = handles_[res - WAIT_OBJECT_0];
                 DWORD exit_code;
-                if (!::GetExitCodeProcess(handle, &exit_code))
+                if (!GetExitCodeProcess(handle, &exit_code))
                     BOOST_PROCESS_THROW_LAST_SYSTEM_ERROR("GetExitCodeProcess() failed");
                 boost::unique_lock<boost::mutex> lock(work_thread_mutex_);
                 for (std::vector<implementation_type>::iterator it = impls_.begin(); it != impls_.end(); ++it)
@@ -161,53 +184,58 @@
                 if (handles_.size() == 1)
                     work_.reset();
             }
-#elif defined(BOOST_POSIX_API)
-            //linux here
 #endif
         }
     }
 
-    bool running()
-    {
-        // Access to run_ is sychronized with stop_work_thread().
-        boost::mutex::scoped_lock lock(work_thread_mutex_);
-        return run_;
-    }
-
     void interrupt_work_thread()
     {
+#if defined(BOOST_POSIX_API)
+        // By creating a child process which immediately exits we interrupt wait().
+        interrupt_pid_ = create_child("/usr/sh").get_id();
+#elif defined(BOOST_WINDOWS_API)
         // By signaling the event in the first slot WaitForMultipleObjects() will return.
         // The work thread won't do anything except checking if it should continue to run.
-#if defined(BOOST_WINDOWS_API)
-        if (!::SetEvent(handles_[0]))
+        if (!SetEvent(handles_[0]))
             BOOST_PROCESS_THROW_LAST_SYSTEM_ERROR("SetEvent() failed");
-#elif defined(BOOST_POSIX_API)
-            //linux here
 #endif
     }
 
+#if defined(BOOST_POSIX_API)
+    bool interrupted(pid_t pid)
+    {
+        boost::mutex::scoped_lock lock(work_thread_mutex_);
+        return interrupt_pid_ == pid;
+    }
+#endif
+
     void stop_work_thread()
     {
-        // Access to run_ is sychronized with running().
         boost::mutex::scoped_lock lock(work_thread_mutex_);
+#if defined(BOOST_WINDOWS_API)
+        // Access to run_ must be sychronized with running().
         run_ = false;
+#endif
+        // Access to interrupt_pid_ must be sychronized with interrupted().
         interrupt_work_thread();
     }
 
     boost::scoped_ptr<boost::asio::io_service::work> work_;
     std::vector<implementation_type> impls_;
     boost::mutex work_thread_mutex_;
-    boost::condition_variable_any work_thread_cond_;
-    bool run_;
     boost::thread work_thread_;
-    std::vector<behavior::stream::native_type> handles_;
-
-
-
+#if defined(BOOST_POSIX_API)
+    pid_t interrupt_pid_;
+    int pids_;
+#elif defined(BOOST_WINDOWS_API)
+    bool run_;
+    boost::condition_variable_any work_thread_cond_;
+    std::vector<HANDLE> handles_;
+#endif
 };
 
 }
 }
 }
 
-#endif 
+#endif

Modified: sandbox/SOC/2010/process/boost/process/detail/posix_helpers.hpp
==============================================================================
--- sandbox/SOC/2010/process/boost/process/detail/posix_helpers.hpp	(original)
+++ sandbox/SOC/2010/process/boost/process/detail/posix_helpers.hpp	2010-07-20 20:37:10 EDT (Tue, 20 Jul 2010)
@@ -99,9 +99,4 @@
 }
 }
 
-void posix_remap(int native_handle, int new_handle)
-{
-    if (::dup2(new_handle, native_handle) == -1)
-        BOOST_PROCESS_THROW_LAST_SYSTEM_ERROR("dup2() failed"); 
-}
 #endif

Modified: sandbox/SOC/2010/process/boost/process/detail/status_impl.hpp
==============================================================================
--- sandbox/SOC/2010/process/boost/process/detail/status_impl.hpp	(original)
+++ sandbox/SOC/2010/process/boost/process/detail/status_impl.hpp	2010-07-20 20:37:10 EDT (Tue, 20 Jul 2010)
@@ -22,11 +22,11 @@
 #include <boost/process/config.hpp>
 #include <boost/system/error_code.hpp>
 #include <boost/ptr_container/ptr_unordered_map.hpp>
-#include <boost/bind.hpp>
-#include <boost/process/stream_behavior.hpp>
 #include <algorithm>
 
 #if defined(BOOST_POSIX_API)
+#   include <sys/types.h>
+#   include <sys/wait.h>
 #elif defined(BOOST_WINDOWS_API)
 #   include <windows.h>
 #else
@@ -37,13 +37,15 @@
 namespace process {
 namespace detail {
 
+#elif defined(BOOST_POSIX_API)
+typedef pid_t phandle;
+#elif defined(BOOST_WINDOWS_API)
+typedef HANDLE phandle;
+#endif
+
 struct operation
 {
-    #if defined(BOOST_WINDOWS_API)
-    virtual void operator()(DWORD exit_code)
-    #elif defined BOOST_POSIX_API
-    virtual void operator()(unsigned int exit_code)
-    #endif
+    virtual void operator()(int exit_code)
     {
     }
 };
@@ -57,11 +59,7 @@
     {
     }
 
-    #if defined(BOOST_WINDOWS_API)
-    void operator()(DWORD exit_code)
-    #elif defined BOOST_POSIX_API
-    void operator()(unsigned int exit_code)
-    #endif
+    void operator()(int exit_code)
     {
         handler_(boost::system::error_code(), exit_code);
     }
@@ -73,33 +71,72 @@
 class status_impl
 {
 public:
+    int wait(pid_type pid, boost::system::error_code ec)
+    {
+#if defined(BOOST_POSIX_API)
+        pid_t p;
+        int status;
+        do
+        {
+            p = waitpid(pid, &status, 0);
+        } while (p == -1 && errno == EINTR);
+        if (p == -1)
+        {
+            ec = boost::system::system_error(boost::system::error_code(errno, boost::system::get_system_category()), BOOST_PROCESS_SOURCE_LOCATION "waitpid(2) failed");
+            return -1;
+        }
+        return status;
+#elif defined(BOOST_WINDOWS_API)
+        HANDLE h = OpenProcess(PROCESS_QUERY_INFORMATION | SYNCHRONIZE, FALSE, pid);
+        if (h == NULL)
+        {
+            ec = boost::system::system_error(boost::system::error_code(errno, boost::system::get_system_category()), BOOST_PROCESS_SOURCE_LOCATION "OpenProcess() failed");
+            return -1;
+        }
+
+        if (WaitForSingleObject(h, INFINITE) == WAIT_FAILED)
+        {
+            CloseHandle(h);
+            ec = boost::system::system_error(boost::system::error_code(errno, boost::system::get_system_category()), BOOST_PROCESS_SOURCE_LOCATION "WaitForSingleObject() failed");
+            return -1;
+        }
+
+        DWORD exit_code;
+        if (!GetExitCodeProcess(h, &exit_code))
+        {
+            CloseHandle(h);
+            ec = boost::system::system_error(boost::system::error_code(errno, boost::system::get_system_category()), BOOST_PROCESS_SOURCE_LOCATION "GetExitCodeProcess() failed");
+            return -1;
+        }
+        if (!CloseHandle(h))
+        {
+            ec = boost::system::system_error(boost::system::error_code(errno, boost::system::get_system_category()), BOOST_PROCESS_SOURCE_LOCATION "CloseHandle() failed");
+            return -1;
+        }
+        return exit_code;
+#endif
+    }
+
     template <typename Handler>
-    void async_wait(behavior::stream::native_type handle, Handler handler)
+    void async_wait(phandle ph, Handler handler)
     {
-        ops_.insert(handle, new wrapped_handler<Handler>(handler));
+        ops_.insert(ph, new wrapped_handler<Handler>(handler));
     }
 
-    #if defined(BOOST_WINDOWS_API)
-    void complete(behavior::stream::native_type handle, DWORD exit_code)
-    #elif defined(BOOST_POSIX_API)
-    void complete(behavior::stream::native_type handle, unsigned int exit_code)
-    #endif
+    void complete(phandle ph, int exit_code)
     {
-        boost::iterator_range<operations_type::iterator> r = ops_.equal_range(handle);
+        boost::iterator_range<operations_type::iterator> r = ops_.equal_range(ph);
         for (operations_type::iterator it = r.begin(); it != r.end(); ++it)
             (*it->second)(exit_code);
         ops_.erase(r.begin(), r.end());
-        #if defined(BOOST_WINDOWS_API)
-        if (!::CloseHandle(handle))
+#if defined(BOOST_WINDOWS_API)
+        if (!CloseHandle(ph))
             BOOST_PROCESS_THROW_LAST_SYSTEM_ERROR("CloseHandle() failed");
-        #elif defined(BOOST_POSIX_API)
-        if (::close(handle) == -1)
-            BOOST_PROCESS_THROW_LAST_SYSTEM_ERROR("close() failed");
-        #endif
+#endif
     }
 
 private:
-    typedef boost::ptr_unordered_multimap<behavior::stream::native_type, operation> operations_type;
+    typedef boost::ptr_unordered_multimap<phandle, operation> operations_type;
     operations_type ops_;
 };
 

Modified: sandbox/SOC/2010/process/boost/process/operations.hpp
==============================================================================
--- sandbox/SOC/2010/process/boost/process/operations.hpp	(original)
+++ sandbox/SOC/2010/process/boost/process/operations.hpp	2010-07-20 20:37:10 EDT (Tue, 20 Jul 2010)
@@ -200,29 +200,33 @@
 
             int stdin_fd = ctx.stdin_behavior->get_child_end();
             if (stdin_fd != -1 && stdin_fd < maxdescs)
-                closeflags[stdin_fd] = false;
+            {
+                if (::dup2(stdin_fd, STDIN_FILENO) == -1)
+                    BOOST_PROCESS_THROW_LAST_SYSTEM_ERROR("dup2() failed");
+                closeflags[STDIN_FILENO] = false;
+            }
 
             int stdout_fd = ctx.stdout_behavior->get_child_end();
             if (stdout_fd != -1 && stdout_fd < maxdescs)
-                closeflags[stdout_fd] = false;
+            {
+                if (::dup2(stdout_fd, STDOUT_FILENO) == -1)
+                    BOOST_PROCESS_THROW_LAST_SYSTEM_ERROR("dup2() failed");
+                closeflags[STDOUT_FILENO] = false;
+            }
 
             int stderr_fd = ctx.stderr_behavior->get_child_end();
             if (stderr_fd != -1 && stderr_fd < maxdescs)
-                closeflags[stderr_fd] = false;
+            {
+                if (::dup2(stderr_fd, STDERR_FILENO) == -1)
+                    BOOST_PROCESS_THROW_LAST_SYSTEM_ERROR("dup2() failed");
+                closeflags[STDERR_FILENO] = false;
+            }
 
             for (int i = 0; i < maxdescs; ++i)
             {
                 if (closeflags[i])
                     ::close(i);
             }
-
-            if(closeflags[stdin_fd] == false)
-                posix_remap(STDIN_FILENO, stdin_fd);
-            if(closeflags[stdout_fd] == false)
-                posix_remap(STDOUT_FILENO, stdout_fd);
-            if(closeflags[stderr_fd] == false)
-                posix_remap(STDERR_FILENO, stderr_fd);
-
         }
         catch (const boost::system::system_error &e)
         {

Modified: sandbox/SOC/2010/process/boost/process/process.hpp
==============================================================================
--- sandbox/SOC/2010/process/boost/process/process.hpp	(original)
+++ sandbox/SOC/2010/process/boost/process/process.hpp	2010-07-20 20:37:10 EDT (Tue, 20 Jul 2010)
@@ -137,8 +137,7 @@
             BOOST_PROCESS_THROW_LAST_SYSTEM_ERROR("waitpid(2) failed"); 
         return s;
 #elif defined(BOOST_WINDOWS_API)
-
-        HANDLE h = ::OpenProcess(PROCESS_QUERY_INFORMATION |SYNCHRONIZE    , FALSE, id_); 
+        HANDLE h = ::OpenProcess(PROCESS_QUERY_INFORMATION | SYNCHRONIZE, FALSE, id_); 
         if (h == NULL) 
             BOOST_PROCESS_THROW_LAST_SYSTEM_ERROR("OpenProcess() failed"); 
 
@@ -149,7 +148,6 @@
         } 
 
         DWORD exit_code; 
-
         if (!::GetExitCodeProcess(h, &exit_code)) 
         { 
             ::CloseHandle(h); 

Modified: sandbox/SOC/2010/process/libs/process/example/read_from_child.cpp
==============================================================================
--- sandbox/SOC/2010/process/libs/process/example/read_from_child.cpp	(original)
+++ sandbox/SOC/2010/process/libs/process/example/read_from_child.cpp	2010-07-20 20:37:10 EDT (Tue, 20 Jul 2010)
@@ -18,13 +18,11 @@
 
 int main() 
 { 
-
     std::string exe = find_executable_in_path("hostname"); 
-	context ctx; 
-	ctx.stdout_behavior = behavior::pipe::def(behavior::pipe::stream_type::output_stream); 
-
+    context ctx; 
+    ctx.stdout_behavior = behavior::pipe::def(behavior::pipe::stream_type::output_stream); 
     child c = create_child(exe,ctx); 
     pistream &is = c.get_stdout(); 
     std::cout << is.rdbuf(); 
-	c.wait();
+    c.wait();
 } 

Modified: sandbox/SOC/2010/process/libs/process/example/wait_child.cpp
==============================================================================
--- sandbox/SOC/2010/process/libs/process/example/wait_child.cpp	(original)
+++ sandbox/SOC/2010/process/libs/process/example/wait_child.cpp	2010-07-20 20:37:10 EDT (Tue, 20 Jul 2010)
@@ -20,8 +20,6 @@
 { 
     std::string exe = find_executable_in_path("hostname"); 
     child c = create_child(exe); 
-
-	int exit_c = c.wait(); 
-	std::cout << exit_c << std::endl; 
-	
+    int exit_code = c.wait(); 
+    std::cout << exit_code << std::endl; 
 }