From 82d467944fd049412942adef9c6f9602ac6eb23a Mon Sep 17 00:00:00 2001 From: Ray Speth Date: Fri, 9 Mar 2012 22:59:21 +0000 Subject: [PATCH] Eliminated temporary files from the ct2ctml conversion process Added libstringstream, which maps stdin and stdout to std::iostream. Using this library means we no longer create either the temporary .py file or the ct2ctml.log file. This also eliminates the synchronization problems that motivated the sleep commands around the system call, so cti to ctml conversions are significantly faster now. --- SConstruct | 12 +- ext/SConscript | 2 +- ext/libexecstream/README | 42 + ext/libexecstream/doc/index.html | 163 +++ ext/libexecstream/doc/libexecstream.css | 137 +++ ext/libexecstream/doc/license.html | 73 ++ ext/libexecstream/doc/news.html | 41 + ext/libexecstream/doc/reference.html | 175 ++++ ext/libexecstream/exec-stream.cpp | 462 +++++++++ ext/libexecstream/exec-stream.h | 163 +++ .../posix/exec-stream-helpers.cpp | 842 ++++++++++++++++ ext/libexecstream/posix/exec-stream-helpers.h | 239 +++++ ext/libexecstream/posix/exec-stream-impl.cpp | 386 +++++++ ext/libexecstream/test/Makefile.gcc | 21 + ext/libexecstream/test/Makefile.msvc | 19 + ext/libexecstream/test/exec-stream-test.cpp | 944 ++++++++++++++++++ ext/libexecstream/win/exec-stream-helpers.cpp | 727 ++++++++++++++ ext/libexecstream/win/exec-stream-helpers.h | 183 ++++ ext/libexecstream/win/exec-stream-impl.cpp | 315 ++++++ interfaces/python/ctml_writer.py | 13 +- src/base/ct2ctml.cpp | 134 +-- 21 files changed, 4992 insertions(+), 101 deletions(-) create mode 100644 ext/libexecstream/README create mode 100644 ext/libexecstream/doc/index.html create mode 100644 ext/libexecstream/doc/libexecstream.css create mode 100644 ext/libexecstream/doc/license.html create mode 100644 ext/libexecstream/doc/news.html create mode 100644 ext/libexecstream/doc/reference.html create mode 100644 ext/libexecstream/exec-stream.cpp create mode 100644 ext/libexecstream/exec-stream.h create mode 100644 ext/libexecstream/posix/exec-stream-helpers.cpp create mode 100644 ext/libexecstream/posix/exec-stream-helpers.h create mode 100644 ext/libexecstream/posix/exec-stream-impl.cpp create mode 100644 ext/libexecstream/test/Makefile.gcc create mode 100644 ext/libexecstream/test/Makefile.msvc create mode 100644 ext/libexecstream/test/exec-stream-test.cpp create mode 100644 ext/libexecstream/win/exec-stream-helpers.cpp create mode 100644 ext/libexecstream/win/exec-stream-helpers.h create mode 100644 ext/libexecstream/win/exec-stream-impl.cpp diff --git a/SConstruct b/SConstruct index 834ca984a..d81db1408 100644 --- a/SConstruct +++ b/SConstruct @@ -196,6 +196,8 @@ elif env['CC'] == 'clang': else: print "WARNING: Unrecognized C compiler '%s'" % env['CC'] +defaults.threadFlags = '-pthread' if os.name != 'nt' else '' + # TODO: Once deprecated functions have been removed, remove the # compiler options: -Wno-deprecated-declarations and /wd4996 @@ -468,6 +470,9 @@ opts.AddVariables( ('cc_flags', 'Compiler flags passed to both the C and C++ compilers, regardless of optimization level', defaults.ccFlags), + ('thread_flags', + 'Compiler and linker flags for POSIX multithreading support', + defaults.threadFlags), BoolVariable( 'optimize', """Enable extra compiler optimizations specified by the "optimize_flags" variable, @@ -831,11 +836,13 @@ env['inst_mandir'] = pjoin(instRoot, 'man1') env['inst_matlab_dir'] = pjoin(instRoot, 'matlab', 'toolbox') env['CXXFLAGS'] = listify(env['cxx_flags']) +env['CCFLAGS'] = listify(env['cc_flags']) + listify(env['thread_flags']) +env['LINKFLAGS'] += listify(env['thread_flags']) if env['optimize']: - env['CCFLAGS'] = listify(env['cc_flags']) + listify(env['optimize_flags']) + env['CCFLAGS'] += listify(env['optimize_flags']) else: - env['CCFLAGS'] = listify(env['cc_flags']) + listify(env['no_optimize_flags']) + env['CCFLAGS'] += listify(env['no_optimize_flags']) if env['debug']: env['CCFLAGS'] += listify(env['debug_flags']) @@ -844,7 +851,6 @@ else: env['CCFLAGS'] += listify(env['no_debug_flags']) env['LINKFLAGS'] += listify(env['no_debug_linker_flags']) - if env['coverage']: if env['CC'] == 'gcc': env.Append(CCFLAGS=['-fprofile-arcs', '-ftest-coverage']) diff --git a/ext/SConscript b/ext/SConscript index 9611ce634..b5dab2ca8 100644 --- a/ext/SConscript +++ b/ext/SConscript @@ -53,7 +53,7 @@ def prep_gtest(env): return localenv # (subdir, (file extensions), prepfunction) -libs = [] +libs = [('libexecstream', ['cpp'], prep_default)] if env['build_with_f2c']: libs.append(('f2c_math', ['cpp','c'], prep_f2c)) diff --git a/ext/libexecstream/README b/ext/libexecstream/README new file mode 100644 index 000000000..e07773e9a --- /dev/null +++ b/ext/libexecstream/README @@ -0,0 +1,42 @@ +This is version 0.3 of libexecstream, a C++ library +that allows you to run a child process and have its input, +output and error avaliable as standard C++ streams. + +Copyright (c) 2004 Artem Khodush +Libexecstream is distributed under the BSD-style license, +see doc/license.html for the details. + +Documentation: + doc/index.html + http://libexecstream/sourceforge.net/ + +Features: + Works on Linux and Windows + Uses threads + Does not depend on any other non-standard library + Distributed as source code only, requires you to compile and link + one file into your program + +Installaion: + +Libexecstream is provided in source code form only. +In order to use it, you need to compile and link one file, exec-stream.cpp, +into your program. + +Header file exec-stream.h defines interface of the library and uses +only standard C++. It does not include any platform-specific header files. + +On Linux, libexecstream was tested on Red Hat 9 with gcc compiler. +Versions of gcc prior to 3.0 will not work. Make sure that exec-stream.h +is found somewhere on the include path, compile exec-stream.cpp as usual, +link your program with -lpthread. GCC must be configured with --enable-threads, +which is by default on most Linux distributions. + +On Windows, libexecstream was tested on XP and 95 flavors with VC++ 7 compiler. +VC++ 6 will not work. Make sure that exec-stream.h is found somewhere +on the include path, compile exec-stream.cpp as usual, link you program +with multi-threaded runtime. + +Example makefiles for Windows and Linux (used to build the testsute) +are provided in the test subdirectory. + diff --git a/ext/libexecstream/doc/index.html b/ext/libexecstream/doc/index.html new file mode 100644 index 000000000..eee86f261 --- /dev/null +++ b/ext/libexecstream/doc/index.html @@ -0,0 +1,163 @@ + + + + + +libexecstream home page + + + + + +

The libexecstream library

+ + +++ + + + + + +
+
Overview
+
Download
+
Installation
+
Reference
+
News
+
License
+ +		 + + + +

Overview

+ +

Libexecstream is a C++ library that allows you to run a child process and have its input, output and error +avaliable as standard C++ streams. +

+

Like this:

+ +
+#include <exec-stream.h>
+#include <string>
+...
+try {
+    exec_stream_t es( "perl", "" ); // run perl without any arguments 
+    es.in() << "print \"hello world\";"; // and make it print "hello world" 
+    es.close_in();                        // after the input was closed 
+    std::string hello, world;
+    es.out() >> hello; // read the first word of output 
+    es.out() >> world; // read the second word 
+}catch( std::exception const & e ) {
+    std::cerr << "error: "  <<  e.what()  <<  "\n";
+}
+
+ +

Features: +

    +
  • Works on Linux and Windows
    • +
  • Uses threads
    • +
  • Does not depend on any other non-standard library
    • +
  • Distributed as source code only, requires you to compile and link one file into your program
    • +
  • BSD-style license
    • +
+

+ +

Another example: +

+
+#include <exec-stream.h>
+...
+exec_stream_t es;
+try {
+    // run command to print network configuration, depending on the operating system
+    #ifdef _WIN32
+        es.start( "ipconfig", "/all" );
+    #else
+        es.start( "ifconfig", "-a" );
+    #endif
+    
+    std::string s;
+    while( std::getline( es.out(), s ).good() ) {
+        // do something with s
+    }
+}catch( std::exception const & e ) {
+    std::cerr << "error: "  <<  e.what()  <<  "\n";
+}
+
+ +

For more examples see the file test/exec-stream-test.cpp in the source distribution. +The interface provided by the library is documented in the reference. +

+ + + +

Download

+ +
+Latest source release from SourceForge: + +follow this link for downloading libexecstream-0.3.tar.gz +
+ +
+Get source from the CVS: +
(password is empty) +		 +cvs -d:pserver:anonymous@cvs.sourceforge.net:/cvsroot/libexecstream login +
cvs -z3 -d:pserver:anonymous@cvs.sourceforge.net:/cvsroot/libexecstream co libexecstream +
+ +
+View source in the CVS: + +http://cvs.sourceforge.net/viewcvs.py/libexecstream +
+ +
+Link to the SourceForge project page: + +http://sourceforge.net/projects/libexecstream/ +
+ + + +

Installation

+ +

Libexecstream is provided in source code form only. In order to use it, you need to compile and link +one file, exec-stream.cpp, into your program. +

+ +

On Linux, libexecstream was tested on Red Hat 9 with gcc compiler. Versions of gcc prior to 3.0 will not work. +Make sure that exec-stream.h is found somewhere on the include path, +compile exec-stream.cpp as usual, link your program with -lpthread. +GCC must be configured with --enable-threads, which is by default on most Linux distributions. +

+ +

On Windows, libexecstream was tested on XP and 95 flavors with VC++ 7 compiler. VC++ 6 will not work. +Make sure that exec-stream.h is found somewhere on the include path, +compile exec-stream.cpp as usual, link you program with multi-threaded runtime. +

+ +

Example makefiles for Windows and Linux (used to build the testsute) are provided in the test directory +of the source distribution. +

+ +

The exec-stream.cpp file includes several platform-dependent +implementation files. Selection of platform-specific implementation is done at compile time: when _WIN32 +macro is defined (usually by windows compiler) win32 implementation is included, when that macro is not defined, +posix implementation is included. +

+ +

Header file exec-stream.h defines interface of the library and uses only standard C++. +It does not include any platform-specific header files. +

+ +
+ + + diff --git a/ext/libexecstream/doc/libexecstream.css b/ext/libexecstream/doc/libexecstream.css new file mode 100644 index 000000000..2e1014b02 --- /dev/null +++ b/ext/libexecstream/doc/libexecstream.css @@ -0,0 +1,137 @@ + +body { + background-color: #ffffff; +} + +body, table, td, p { + color: #000000; +} + +a { + color: #1010d0; +} + +h1 { + font-size: 120%; + font-weight: bold; +} + +h2 { + font-size: 100%; + font-weight: bold; + background-color: #f4f4ff; + padding-top: 2px; + padding-bottom: 2px; + padding-left: 1em; +} + +p { + text-indent: 3em; +} + +col.linksbar { + width: 1em; +} + +table.maintable { + width: 100%; + margin-top: 1em; + border-collapse: collapse; +} + +td.linksbar { + vertical-align: top; + text-align: right; +} + +div.linksbar { + margin-bottom: 8px; + padding-top: 2px; + padding-bottom: 2px; + padding-left: 1em; + padding-right: 1em; + background-color: #e0e0ff; +} + +div.linksbar a:link { + text-decoration: none; + color: #0030d0; +} +div.linksbar a:active { + text-decoration: none; + color: #0030d0; +} +div.linksbar a:visited { + text-decoration: none; + color: #0060d0; +} + +td.body { + vertical-align: top; + text-align: left; + padding-left: 1em; +} + +pre { + color: #001090; +} + +.comment { + color: #3540f0; +} + +table.downloadlinks a { + margin-left: 1em; + margin-bottom: 5px; +} + +div.membersdesc { + padding-left: 3em; +} +div.memberslink { + padding-top: 5px; + color: #001090; +} +div.memberslink a:link { + color: #001090; + text-decoration: none; +} +div.memberslink a:visited { + color: #001090; + text-decoration: none; +} +div.memberslink a:active { + color: #001090; + text-decoration: none; +} +div.memberslink a .link { + text-decoration: underline; +} + +td.reference { + vertical-align: top; + text-align: left; + padding-left: 1em; +} + +td.reference p { + font-family: verdana, arial, sans; + font-size: smaller; +} + +td.reference h2 { + color: #001090; + padding-top: 2px; + padding-bottom: 2px; + padding-left: 1em; + margin-top: 40px; + font-weight: normal; + font-family: courier new, fixed, courier; + font-size: smaller; +} + +table.downloadlinks td { + vertical-align: top; + padding-top: 8px; + padding-left: 0.5em; +} \ No newline at end of file diff --git a/ext/libexecstream/doc/license.html b/ext/libexecstream/doc/license.html new file mode 100644 index 000000000..19f9eddef --- /dev/null +++ b/ext/libexecstream/doc/license.html @@ -0,0 +1,73 @@ + + + + + +libexecstream license + + + + + +

The libexecstream library license

+ + +++ + + + + + +
+
Overview
+
Download
+
Installation
+
Reference
+
News
+
License
+ +		 + +

Libexecsteam copyright Artem Khodush, 2004. +

+ +

License

+ +

Redistribution and use in source and binary forms, with or without modification, +are permitted provided that the following conditions are met: +

+ +

1. Redistributions of source code must retain the above copyright notice, +this list of conditions and the following disclaimer. +

+ +

2. Redistributions in binary form must reproduce the above copyright notice, +this list of conditions and the following disclaimer in the documentation +and/or other materials provided with the distribution. +

+ +

3. The name of the author may not be used to endorse or promote products +derived from this software without specific prior written permission. +

+ +

+THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED +WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES +OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. +IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; +OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, +WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR +OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, +EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +

+ +
+ + + diff --git a/ext/libexecstream/doc/news.html b/ext/libexecstream/doc/news.html new file mode 100644 index 000000000..3617c2b56 --- /dev/null +++ b/ext/libexecstream/doc/news.html @@ -0,0 +1,41 @@ + + + + + +libexecstream news + + + + + +

The libexecstream library news

+ + +++ + + + + + +
+
Overview
+
Download
+
Installation
+
Reference
+
News
+
License
+ +		 + +April 12, 2004. +Version 0.3 - first public release. + +
+ + + diff --git a/ext/libexecstream/doc/reference.html b/ext/libexecstream/doc/reference.html new file mode 100644 index 000000000..726077636 --- /dev/null +++ b/ext/libexecstream/doc/reference.html @@ -0,0 +1,175 @@ + + + + + +libexecstream reference + + + + + +

The libexecstream library reference

+ + +++ + + + + + +
+
Overview
+
Download
+
Installation
+
Reference
+
News
+
License
+ +		 + +

Libexecstream provides one class, exec_stream_t, which has the following members: +

+ +

class error_t : public std::exception

+

Exceptions thrown from exec_stream_t members are derived from error_t. error_t is derived from std::exception and has no additional +public members besides constructors. Exceptions may be thrown from any exec_stream_t member function except destructor and accessors: in(), out() and err(). +Writing to in() and reading out() and err() will also throw exceptions when errors occur. +

+ +

exec_stream_t()

+

Constructs exec_stream_t in the default state. You may change timeouts, buffer limits and text or binary modes +of the streams before starting child process (see set_buffer_limit, +set_binary_mode, set_text_mode, +set_wait_timeout). In the default state, amount of data buffered for writing to child's stdin, +and amount of data read in advance from child's stdout and stderr is unlimited. On windows, all streams are in the text mode. +

+ +

exec_stream_t( std::string const & program, std::string const & arguments )

+

Constructs exec_stream_t in the default state, then starts program with arguments. Arguments containing space should be +included in double quotation marks, and double quote in such arguments should be escaped with backslash. +

+ +

template< class iterator > exec_stream_t( std::string const & program, iterator args_begin, iterator args_end )

+

Constructs exec_stream_t in the default state, then starts program with arguments specified by the range args_begin, args_end. +args_begin should be an input iterator that when dereferenced gives value assignable to std::string. +Spaces and double quotes in arguments need not to be escaped. +

+ +

~exec_stream_t()

+

Writes (with timeout) all pending data to child stdin, +closes streams and waits (with timeout) for child process to stop. +

+ +

std::ostream & in()

+

Returns output stream for writing to child's stdin. +

+ +

std::istream & out()

+

Returns input stream for reading child's stdout. +

+ +

std::istream & err()

+

Returns input stream for reading child's stderr. +

+ +

bool close_in()

+

Closes child's standard input after writing (with timeout) all pending data to it. +

+ +

void start( std::string const & program, std::string const & arguments )

+

Starts program with arguments. Arguments are space-separated. Arguments containing space should be +included in double quotation marks, and double quote in such arguments should be escaped with backslash. +

+ +

template< class iterator > void start( std::string const & program, iterator args_begin, iterator args_end )

+

Starts program with arguments specified by the range args_begin, args_end. args_begin should be an input iterator that when dereferenced +gives value assignable to std::string. Spaces and double quotes in arguments need not to be escaped. +

+ +

enum stream_kind_t { s_in=1, s_out=2, s_err=4, s_all=s_in|s_out|s_err, s_child=8 }

+

Used for the first argument to set_buffer_limit, set_wait_timeout, +set_binary_mode, set_text_mode for selecting stream to operate upon. +

+ +

void set_buffer_limit( int stream_kind, std::size_t size )

+

For out() and err() streams (when exec_stream_t::s_out +or exec_stream_t::s_err is set in the stream_kind), sets maximum amount of data to read from child process +before it will be consumed by reading from out() or err(). +

+

For in() stream (when exec_stream_t::s_in is set in the stream_kind) +sets maximum amount of data to store as result of writing to in() before it will be consumed by child process. +

+

Setting limit for both input and output streams may cause deadlock in situations when both your program and child process +are writing data to each other without reading it. Such deadlock will cause the timeout to expire while +writing to in(). +

+

When size argument to set_buffer_limit is 0, buffers are considered unlimited, and will grow unlimited if one side produce data that the other side does not consume. +This is the default state after exec_stream_t creation. +

+

set_buffer_limit will throw exception when called while child process is running. +

+ +

typedef unsigned long timeout_t

+

Type of second argument to set_wait_timeout - timeout in milliseconds. +

+ +

void set_wait_timeout( int stream_kind, timeout_t milliseconds )

+

For out() and err() streams (when exec_stream_t::s_out +or exec_stream_t::s_err is set in the stream_kind), sets maximum amount of time to wait for a +child process to produce data when reading out() and err() respectively. +

+

For in() stream (when exec_stream_t::s_in is set in the stream_kind), +sets maximum amount of time to wait for a child process to consume data that were written to in(). +Note that when buffer limit for in() is not set, writing to in() always writes to buffer and does not wait for child at all. +

+

If that amount of time is exceeded while reading in() or writing to out() and err(), exception is thrown. +

+

When exec_stream_t::s_child is set in the stream kind, set_wait_timeout sets the maximum amount of time to wait +for a child process to terminate when close is called. If that amount of time is exceeded, close() will return false. +

+

set_wait_timeout will throw exception when called while child process is running. +

+ +

void set_text_mode( int stream_kind )

+

+sets stream specified by stream_kind to text mode. In text mode, in the data written to child's stdin, +\n are replaced by \r\n; and in the data read from child's stdout and stderr, \r\n are replaced by \n. Text mode is the default on Windows. +set_text_mode has no effect on Linux. +

+

set_text_mode will throw exception when called while child process is running. +

+ +

void set_binary_mode( int stream_kind )

+

+sets stream specified by stream_kind to binary mode. All data written or read from streams are passed unchanged. +set_binary_mode has no effect on Linux. +

+

set_binary_mode will throw exception when called while child process is running. +

+ +

bool close()

+

Writes (with timeout) all pending data to child stdin, +closes streams and waits (with timeout) for child process to stop. +If timeout expires while waiting for child to stop, returns false. Otherwise, returns true. +

+ +

void kill()

+

+Terminates child process, without giving it a chance of proper shutdown. +

+ +

int exit_code()

+

Returns exit code from child process. Exit code usually is available only after close. Exception is thrown if chid process +has not yet terminated. Exit code has indeterminable value after kill. +

+ + +
+ + + \ No newline at end of file diff --git a/ext/libexecstream/exec-stream.cpp b/ext/libexecstream/exec-stream.cpp new file mode 100644 index 000000000..f4c32e31d --- /dev/null +++ b/ext/libexecstream/exec-stream.cpp @@ -0,0 +1,462 @@ +/* +Copyright (C) 2004 Artem Khodush + +Redistribution and use in source and binary forms, with or without modification, +are permitted provided that the following conditions are met: + +1. Redistributions of source code must retain the above copyright notice, +this list of conditions and the following disclaimer. + +2. Redistributions in binary form must reproduce the above copyright notice, +this list of conditions and the following disclaimer in the documentation +and/or other materials provided with the distribution. + +3. The name of the author may not be used to endorse or promote products +derived from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED +WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES +OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. +IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; +OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, +WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR +OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, +EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ + +#include "exec-stream.h" + +#include +#include +#include +#include + +#ifdef _WIN32 + +#define NOMINMAX +#include + +#define HELPERS_H "win/exec-stream-helpers.h" +#define HELPERS_CPP "win/exec-stream-helpers.cpp" +#define IMPL_CPP "win/exec-stream-impl.cpp" + +#else + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#define HELPERS_H "posix/exec-stream-helpers.h" +#define HELPERS_CPP "posix/exec-stream-helpers.cpp" +#define IMPL_CPP "posix/exec-stream-impl.cpp" + +#endif + +// helper classes +namespace { + +class buffer_list_t { +public: + struct buffer_t { + std::size_t size; + char * data; + }; + + buffer_list_t(); + ~buffer_list_t(); + + void get( char * dst, std::size_t & size ); + void get_translate_crlf( char * dst, std::size_t & size ); + void put( char * const src, std::size_t size ); + void put_translate_crlf( char * const src, std::size_t size ); + buffer_t detach(); + + bool empty(); + bool full( std::size_t limit ); // limit==0 -> no limit + + void clear(); + +private: + typedef std::list< buffer_t > buffers_t; + buffers_t m_buffers; + std::size_t m_read_offset; // offset into the first buffer + std::size_t m_total_size; +}; + +buffer_list_t::buffer_list_t() +{ + m_total_size=0; + m_read_offset=0; +} + +buffer_list_t::~buffer_list_t() +{ + clear(); +} + +void buffer_list_t::get( char * dst, std::size_t & size ) +{ + std::size_t written_size=0; + while( size>0 && m_total_size>0 ) { + std::size_t portion_size=std::min( size, m_buffers.front().size-m_read_offset ); + std::char_traits< char >::copy( dst, m_buffers.front().data+m_read_offset, portion_size ); + dst+=portion_size; + size-=portion_size; + m_total_size-=portion_size; + m_read_offset+=portion_size; + written_size+=portion_size; + if( m_read_offset==m_buffers.front().size ) { + delete[] m_buffers.front().data; + m_buffers.pop_front(); + m_read_offset=0; + } + } + size=written_size; +} + +void buffer_list_t::get_translate_crlf( char * dst, std::size_t & size ) +{ + std::size_t written_size=0; + while( written_size!=size && m_total_size>0 ) { + while( written_size!=size && m_read_offset!=m_buffers.front().size ) { + char c=m_buffers.front().data[m_read_offset]; + if( c!='\r' ) { // MISFEATURE: single \r in the buffer will cause end of file + *dst++=c; + ++written_size; + } + --m_total_size; + ++m_read_offset; + } + if( m_read_offset==m_buffers.front().size ) { + delete[] m_buffers.front().data; + m_buffers.pop_front(); + m_read_offset=0; + } + } + size=written_size; +} + +void buffer_list_t::put( char * const src, std::size_t size ) +{ + buffer_t buffer; + buffer.data=new char[size]; + buffer.size=size; + std::char_traits< char >::copy( buffer.data, src, size ); + m_buffers.push_back( buffer ); + m_total_size+=buffer.size; +} + +void buffer_list_t::put_translate_crlf( char * const src, std::size_t size ) +{ + char const * p=src; + std::size_t lf_count=0; + while( p!=src+size ) { + if( *p=='\n' ) { + ++lf_count; + } + ++p; + } + buffer_t buffer; + buffer.data=new char[size+lf_count]; + buffer.size=size+lf_count; + p=src; + char * dst=buffer.data; + while( p!=src+size ) { + if( *p=='\n' ) { + *dst++='\r'; + } + *dst++=*p; + ++p; + } + m_buffers.push_back( buffer ); + m_total_size+=buffer.size; +} + +buffer_list_t::buffer_t buffer_list_t::detach() +{ + buffer_t buffer=m_buffers.front(); + m_buffers.pop_front(); + m_total_size-=buffer.size; + return buffer; +} + +bool buffer_list_t::empty() +{ + return m_total_size==0; +} + +bool buffer_list_t::full( std::size_t limit ) +{ + return limit!=0 && m_total_size>=limit; +} + +void buffer_list_t::clear() +{ + for( buffers_t::iterator i=m_buffers.begin(); i!=m_buffers.end(); ++i ) { + delete[] i->data; + } + m_buffers.clear(); + m_read_offset=0; + m_total_size=0; +} + +} + +// platform-dependent helpers + +namespace { + +#include HELPERS_H +#include HELPERS_CPP + +} + +// stream buffer class +namespace { + +class exec_stream_buffer_t : public std::streambuf { +public: + exec_stream_buffer_t( exec_stream_t::stream_kind_t kind, thread_buffer_t & thread_buffer ); + virtual ~exec_stream_buffer_t(); + + void clear(); + +protected: + virtual int_type underflow(); + virtual int_type overflow( int_type c ); + virtual int sync(); + +private: + bool send_buffer(); + bool send_char( char c ); + + exec_stream_t::stream_kind_t m_kind; + thread_buffer_t & m_thread_buffer; + char * m_stream_buffer; +}; + +const std::size_t STREAM_BUFFER_SIZE=4096; + +exec_stream_buffer_t::exec_stream_buffer_t( exec_stream_t::stream_kind_t kind, thread_buffer_t & thread_buffer ) +: m_kind( kind ), m_thread_buffer( thread_buffer ) +{ + m_stream_buffer=new char[STREAM_BUFFER_SIZE]; + clear(); +} + +exec_stream_buffer_t::~exec_stream_buffer_t() +{ + delete[] m_stream_buffer; +} + +void exec_stream_buffer_t::clear() +{ + if( m_kind==exec_stream_t::s_in ) { + setp( m_stream_buffer, m_stream_buffer+STREAM_BUFFER_SIZE ); + }else { + setg( m_stream_buffer, m_stream_buffer+STREAM_BUFFER_SIZE, m_stream_buffer+STREAM_BUFFER_SIZE ); + } +} + +exec_stream_buffer_t::int_type exec_stream_buffer_t::underflow() +{ + if( gptr()==egptr() ) { + std::size_t read_size=STREAM_BUFFER_SIZE; + bool no_more; + m_thread_buffer.get( m_kind, m_stream_buffer, read_size, no_more ); + if( no_more || read_size==0 ) { // there is no way for underflow to return something other than eof when 0 bytes are read + return traits_type::eof(); + }else { + setg( m_stream_buffer, m_stream_buffer, m_stream_buffer+read_size ); + } + } + return traits_type::to_int_type( *eback() ); +} + +bool exec_stream_buffer_t::send_buffer() +{ + if( pbase()!=pptr() ) { + std::size_t write_size=pptr()-pbase(); + std::size_t n=write_size; + bool no_more; + m_thread_buffer.put( pbase(), n, no_more ); + if( no_more || n!=write_size ) { + return false; + }else { + setp( m_stream_buffer, m_stream_buffer+STREAM_BUFFER_SIZE ); + } + } + return true; +} + +bool exec_stream_buffer_t::send_char( char c ) +{ + std::size_t write_size=1; + bool no_more; + m_thread_buffer.put( &c, write_size, no_more ); + return write_size==1 && !no_more; +} + +exec_stream_buffer_t::int_type exec_stream_buffer_t::overflow( exec_stream_buffer_t::int_type c ) +{ + if( !send_buffer() ) { + return traits_type::eof(); + } + if( c!=traits_type::eof() ) { + if( pbase()==epptr() ) { + if( !send_char( c ) ) { + return traits_type::eof(); + } + }else { + sputc( c ); + } + } + return traits_type::not_eof( c ); +} + +int exec_stream_buffer_t::sync() +{ + if( !send_buffer() ) { + return -1; + } + return 0; +} + +// stream classes + +class exec_istream_t : public std::istream { +public: + exec_istream_t( exec_stream_buffer_t & buf ) + : std::istream( &buf ) { + } +}; + + +class exec_ostream_t : public std::ostream { +public: + exec_ostream_t( exec_stream_buffer_t & buf ) + : std::ostream( &buf ){ + } +}; + +} + +// platform-dependent implementation +#include IMPL_CPP + + +//platform-independent exec_stream_t member functions +exec_stream_t::exec_stream_t() +{ + m_impl=new impl_t; + exceptions( true ); +} + +exec_stream_t::exec_stream_t( std::string const & program, std::string const & arguments ) +{ + m_impl=new impl_t; + exceptions( true ); + start( program, arguments ); +} + +void exec_stream_t::new_impl() +{ + m_impl=new impl_t; +} + +exec_stream_t::~exec_stream_t() +{ + try { + close(); + }catch( ... ) { + } + delete m_impl; +} + +std::ostream & exec_stream_t::in() +{ + return m_impl->m_in; +} + +std::istream & exec_stream_t::out() +{ + return m_impl->m_out; +} + +std::istream & exec_stream_t::err() +{ + return m_impl->m_err; +} + +void exec_stream_t::exceptions( bool enable ) +{ + if( enable ) { + // getline sets failbit on eof, so we should enable badbit and badbit _only_ to propagate our exceptions through iostream code. + m_impl->m_in.exceptions( std::ios_base::badbit ); + m_impl->m_out.exceptions( std::ios_base::badbit ); + m_impl->m_err.exceptions( std::ios_base::badbit ); + }else { + m_impl->m_in.exceptions( std::ios_base::goodbit ); + m_impl->m_out.exceptions( std::ios_base::goodbit ); + m_impl->m_err.exceptions( std::ios_base::goodbit ); + } +} + +// exec_stream_t::error_t +namespace { + +std::string int2str( unsigned long i, int base, std::size_t width ) +{ + std::string s; + s.reserve(4); + while( i!=0 ) { + s="0123456789abcdef"[i%base]+s; + i/=base; + } + if( width!=0 ) { + while( s.size() +#include +#include +#include +#include + +class exec_stream_t { +public: + exec_stream_t(); + exec_stream_t( std::string const & program, std::string const & arguments ); + template< class iterator > exec_stream_t( std::string const & program, iterator args_begin, iterator args_end ); + + ~exec_stream_t(); + + enum stream_kind_t { s_in=1, s_out=2, s_err=4, s_all=s_in|s_out|s_err, s_child=8 }; + + void set_buffer_limit( int stream_kind, std::size_t size ); + + typedef unsigned long timeout_t; + void set_wait_timeout( int stream_kind, timeout_t milliseconds ); + + void set_binary_mode( int stream_kind ); + void set_text_mode( int stream_kind ); + + void start( std::string const & program, std::string const & arguments ); + template< class iterator > void start( std::string const & program, iterator args_begin, iterator args_end ); + void start( std::string const & program, char const * arg1, char const * arg2 ); // to compensate for damage from the previous one + void start( std::string const & program, char * arg1, char * arg2 ); + + bool close_in(); + bool close(); + void kill(); + int exit_code(); + + std::ostream & in(); + std::istream & out(); + std::istream & err(); + + typedef unsigned long error_code_t; + + class error_t : public std::exception { + public: + error_t( std::string const & msg ); + error_t( std::string const & msg, error_code_t code ); + ~error_t() throw(); + virtual char const * what() const throw(); + protected: + error_t(); + void compose( std::string const & msg, error_code_t code ); + + std::string m_msg; + }; + +private: + exec_stream_t( exec_stream_t const & ); + exec_stream_t & operator=( exec_stream_t const & ); + + struct impl_t; + friend struct impl_t; + impl_t * m_impl; + + void exceptions( bool enable ); + +// helpers for template member functions + void new_impl(); + + class next_arg_t { + public: + virtual ~next_arg_t() + { + } + + virtual std::string const * next()=0; + }; + + template< class iterator > class next_arg_impl_t : public next_arg_t { + public: + next_arg_impl_t( iterator args_begin, iterator args_end ) + : m_args_i( args_begin ), m_args_end( args_end ) + { + } + + virtual std::string const * next() + { + if( m_args_i==m_args_end ) { + return 0; + }else { + m_arg=*m_args_i; + ++m_args_i; + return &m_arg; + } + } + + private: + iterator m_args_i; + iterator m_args_end; + std::string m_arg; + }; + + void start( std::string const & program, next_arg_t & next_arg ); +}; + +template< class iterator > inline exec_stream_t::exec_stream_t( std::string const & program, iterator args_begin, iterator args_end ) +{ + new_impl(); + exceptions( true ); + start( program, args_begin, args_end ); +} + +template< class iterator > inline void exec_stream_t::start( std::string const & program, iterator args_begin, iterator args_end ) +{ + exec_stream_t::next_arg_impl_t< iterator > next_arg( args_begin, args_end ); + start( program, next_arg ); +} + +inline void exec_stream_t::start( std::string const & program, char const * arg1, char const * arg2 ) +{ + std::vector< std::string > args; + args.push_back( std::string( arg1 ) ); + args.push_back( std::string( arg2 ) ); + start( program, args.begin(), args.end() ); +} + +inline void exec_stream_t::start( std::string const & program, char * arg1, char * arg2 ) +{ + std::vector< std::string > args; + args.push_back( std::string( arg1 ) ); + args.push_back( std::string( arg2 ) ); + start( program, args.begin(), args.end() ); +} + +#endif diff --git a/ext/libexecstream/posix/exec-stream-helpers.cpp b/ext/libexecstream/posix/exec-stream-helpers.cpp new file mode 100644 index 000000000..a1c07b766 --- /dev/null +++ b/ext/libexecstream/posix/exec-stream-helpers.cpp @@ -0,0 +1,842 @@ +/* +Copyright (C) 2004 Artem Khodush + +Redistribution and use in source and binary forms, with or without modification, +are permitted provided that the following conditions are met: + +1. Redistributions of source code must retain the above copyright notice, +this list of conditions and the following disclaimer. + +2. Redistributions in binary form must reproduce the above copyright notice, +this list of conditions and the following disclaimer in the documentation +and/or other materials provided with the distribution. + +3. The name of the author may not be used to endorse or promote products +derived from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED +WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES +OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. +IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; +OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, +WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR +OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, +EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ + +// os_error_t +os_error_t::os_error_t( std::string const & msg ) +{ + compose( msg, errno ); +} + +os_error_t::os_error_t( std::string const & msg, exec_stream_t::error_code_t code ) +{ + compose( msg, code ); +} + +void os_error_t::compose( std::string const & msg, exec_stream_t::error_code_t code ) +{ + std::string s( msg ); + s+='\n'; + errno=0; + char const * x=strerror( code ); + if( errno!=0 ) { + s+="[unable to retrieve error description]"; + }else { + s+=x; + } + exec_stream_t::error_t::compose( s, code ); +} + +// pipe_t +pipe_t::pipe_t() +: m_direction( closed ) +{ + m_fds[0]=-1; + m_fds[1]=-1; +} + +pipe_t::~pipe_t() +{ + try { + close(); + }catch(...) { + } +} + +int pipe_t::r() const +{ + return m_fds[0]; +} + +int pipe_t::w() const +{ + return m_fds[1]; +} + +void pipe_t::close_r() +{ + if( m_direction==both || m_direction==read ) { + if( ::close( m_fds[0] )==-1 ) { + throw os_error_t( "pipe_t::close_r: close failed" ); + } + m_direction= m_direction==both ? write : closed; + } +} + +void pipe_t::close_w() +{ + if( m_direction==both || m_direction==write ) { + if( ::close( m_fds[1] )==-1 ) { + throw os_error_t( "pipe_t::close_w: close failed" ); + } + m_direction= m_direction==both ? read : closed; + } +} + +void pipe_t::close() +{ + close_r(); + close_w(); +} + +void pipe_t::open() +{ + close(); + if( pipe( m_fds )==-1 ) { + throw os_error_t( "pipe_t::open(): pipe() failed" ); + } + m_direction=both; +} + + +// mutex_t +mutex_t::mutex_t() +{ + if( int code=pthread_mutex_init( &m_mutex, 0 ) ) { + throw os_error_t( "mutex_t::mutex_t: pthread_mutex_init failed", code ); + } +} + +mutex_t::~mutex_t() +{ + pthread_mutex_destroy( &m_mutex ); +} + + +// grab_mutex_t +grab_mutex_t::grab_mutex_t( mutex_t & mutex, mutex_registrator_t * mutex_registrator ) +{ + m_mutex=&mutex.m_mutex; + m_error_code=pthread_mutex_lock( m_mutex ); + m_grabbed=ok(); + m_mutex_registrator=mutex_registrator; + if( m_mutex_registrator ) { + m_mutex_registrator->add( this ); + } +} + +grab_mutex_t::~grab_mutex_t() +{ + release(); + if( m_mutex_registrator ) { + m_mutex_registrator->remove( this ); + } +} + +int grab_mutex_t::release() +{ + int code=0; + if( m_grabbed ) { + code=pthread_mutex_unlock( m_mutex ); + m_grabbed=false; + } + return code; +} + +bool grab_mutex_t::ok() +{ + return m_error_code==0; +} + +int grab_mutex_t::error_code() +{ + return m_error_code; +} + +// mutex_registrator_t +mutex_registrator_t::~mutex_registrator_t() +{ + for( mutexes_t::iterator i=m_mutexes.begin(); i!=m_mutexes.end(); ++i ) { + (*i)->m_mutex_registrator=0; + } +} + +void mutex_registrator_t::add( grab_mutex_t * g ) +{ + m_mutexes.insert( m_mutexes.end(), g ); +} + +void mutex_registrator_t::remove( grab_mutex_t * g ) +{ + m_mutexes.erase( std::find( m_mutexes.begin(), m_mutexes.end(), g ) ); +} + +void mutex_registrator_t::release_all() +{ + for( mutexes_t::iterator i=m_mutexes.begin(); i!=m_mutexes.end(); ++i ) { + (*i)->release(); + } +} + +// wait_result_t +wait_result_t::wait_result_t( unsigned signaled_state, int error_code, bool timed_out ) +{ + m_timed_out=timed_out; + m_error_code=error_code; + m_signaled_state= error_code==0 ? signaled_state : 0; +} + +bool wait_result_t::ok() +{ + return m_error_code==0; +} + +bool wait_result_t::is_signaled( int state ) +{ + return m_signaled_state&state; +} + +int wait_result_t::error_code() +{ + return m_error_code; +} + +bool wait_result_t::timed_out() +{ + return m_timed_out; +} + + +// event_t +event_t::event_t() +{ + if( int code=pthread_cond_init( &m_cond, 0 ) ) { + throw os_error_t( "event_t::event_t: pthread_cond_init failed", code ); + } + m_state=0; +} + +event_t::~event_t() +{ + pthread_cond_destroy( &m_cond ); +} + +int event_t::set( unsigned bits, mutex_registrator_t * mutex_registrator ) +{ + grab_mutex_t grab_mutex( m_mutex, mutex_registrator ); + if( !grab_mutex.ok() ) { + return grab_mutex.error_code(); + } + + int code=0; + if( bits&~m_state ) { + m_state|=bits; + code=pthread_cond_broadcast( &m_cond ); + } + + int release_code=grab_mutex.release(); + if( code==0 ) { + code=release_code; + } + return code; +} + +int event_t::reset( unsigned bits, mutex_registrator_t * mutex_registrator ) +{ + grab_mutex_t grab_mutex( m_mutex, mutex_registrator ); + if( !grab_mutex.ok() ) { + return grab_mutex.error_code(); + } + m_state&=~bits; + return grab_mutex.release(); +} + +wait_result_t event_t::wait( unsigned any_bits, unsigned long timeout, mutex_registrator_t * mutex_registrator ) +{ + if( any_bits==0 ) { + // we ain't waiting for anything + return wait_result_t( 0, 0, false ); + } + + grab_mutex_t grab_mutex( m_mutex, mutex_registrator ); + if( !grab_mutex.ok() ) { + return wait_result_t( 0, grab_mutex.error_code(), false ); + } + + struct timeval time_val_limit; + gettimeofday( &time_val_limit, 0 ); + struct timespec time_limit; + time_limit.tv_sec=time_val_limit.tv_sec+timeout/1000; + time_limit.tv_nsec=1000*(time_val_limit.tv_usec+1000*(timeout%1000)); + int code=0; + while( code==0 && (m_state&any_bits)==0 ) { + code=pthread_cond_timedwait( &m_cond, &m_mutex.m_mutex, &time_limit ); + } + + unsigned state=m_state; + int release_code=grab_mutex.release(); + if( code==0 ) { + code=release_code; + } + return wait_result_t( state, code, code==ETIMEDOUT ); +} + +// thread_buffer_t +thread_buffer_t::thread_buffer_t( pipe_t & in_pipe, pipe_t & out_pipe, pipe_t & err_pipe, std::ostream & in ) +: m_in_pipe( in_pipe ), m_out_pipe( out_pipe ), m_err_pipe( err_pipe ), m_in( in ) +{ + m_in_bad=false; + m_error_prefix=""; + m_error_code=0; + m_in_wait_timeout=2000; + m_out_wait_timeout=2000; + m_err_wait_timeout=2000; + m_thread_termination_timeout=1000; + m_in_buffer_limit=0; + m_out_buffer_limit=0; + m_err_buffer_limit=0; + m_out_read_buffer_size=4096; + m_err_read_buffer_size=4096; + m_thread_started=false; + m_in_closed=false; +} + +thread_buffer_t::~thread_buffer_t() +{ + bool stopped=false; + try { + stopped=stop_thread(); + }catch( ... ) { + } + if( !stopped ) { + try { + stopped=abort_thread(); + }catch( ... ) { + } + } + if( !stopped ) { + std::terminate(); + } +} + +void thread_buffer_t::set_wait_timeout( int stream_kind, unsigned long milliseconds ) +{ + if( m_thread_started ) { + throw exec_stream_t::error_t( "thread_buffer_t::set_wait_timeout: thread already started" ); + } + if( stream_kind&exec_stream_t::s_in ) { + m_in_wait_timeout=milliseconds; + } + if( stream_kind&exec_stream_t::s_out ) { + m_out_wait_timeout=milliseconds; + } + if( stream_kind&exec_stream_t::s_err ) { + m_err_wait_timeout=milliseconds; + } + if( stream_kind&exec_stream_t::s_child ) { + m_thread_termination_timeout=milliseconds; + } +} + +void thread_buffer_t::set_buffer_limit( int stream_kind, std::size_t limit ) +{ + if( m_thread_started ) { + throw exec_stream_t::error_t( "thread_buffer_t::set_buffer_limit: thread already started" ); + } + if( stream_kind&exec_stream_t::s_in ) { + m_in_buffer_limit=limit; + } + if( stream_kind&exec_stream_t::s_out ) { + m_out_buffer_limit=limit; + } + if( stream_kind&exec_stream_t::s_err ) { + m_err_buffer_limit=limit; + } +} + +void thread_buffer_t::set_read_buffer_size( int stream_kind, std::size_t size ) +{ + if( m_thread_started ) { + throw exec_stream_t::error_t( "thread_buffer_t::set_read_buffer_size: thread already started" ); + } + if( stream_kind&exec_stream_t::s_out ) { + m_out_read_buffer_size=size; + } + if( stream_kind&exec_stream_t::s_err ) { + m_err_read_buffer_size=size; + } +} + +void thread_buffer_t::start() +{ + if( m_thread_started ) { + throw exec_stream_t::error_t( "thread_buffer_t::start: thread already started" ); + } + m_in_buffer.clear(); + m_out_buffer.clear(); + m_err_buffer.clear(); + + int code; + if( (code=m_thread_control.reset( ~0u, 0 )) || (code=m_thread_control.set( exec_stream_t::s_out|exec_stream_t::s_err, 0 ) ) ) { + throw os_error_t( "thread_buffer_t::start: unable to initialize m_thread_control event", code ); + } + if( (code=m_thread_responce.reset( ~0u, 0 )) || (code=m_thread_responce.set( exec_stream_t::s_in, 0 )) ) { + throw os_error_t( "thread_buffer_t::start: unable to initialize m_thread_responce event", code ); + } + + m_error_prefix=""; + m_error_code=0; + + if( int code=pthread_create( &m_thread, 0, &thread_func, this ) ) { + throw os_error_t( "exec_stream_therad_t::start: pthread_create failed", code ); + } + m_thread_started=true; + m_in_closed=false; + m_in_bad=false; +} + +bool thread_buffer_t::stop_thread() +{ + if( m_thread_started ) { + if( int code=m_thread_control.set( exec_stream_t::s_child, 0 ) ) { + throw os_error_t( "thread_buffer_t::stop_thread: unable to set thread termination event", code ); + } + wait_result_t wait_result=m_thread_responce.wait( exec_stream_t::s_child, m_thread_termination_timeout, 0 ); + if( !wait_result.ok() && !wait_result.timed_out() ) { + throw os_error_t( "thread_buffer_t::stop_thread: wait for m_thread_stopped failed", wait_result.error_code() ); + } + if( wait_result.ok() ) { + void * thread_result; + if( int code=pthread_join( m_thread, &thread_result ) ) { + throw os_error_t( "thread_buffer_t::stop_thread: pthread_join failed", code ); + } + m_thread_started=false; + // check for any errors encountered in the thread + if( m_error_code!=0 ) { + throw os_error_t( m_error_prefix, m_error_code ); + } + return true; + }else { + return false; + } + } + return true; +} + +bool thread_buffer_t::abort_thread() +{ + if( m_thread_started ) { + if( int code=pthread_cancel( m_thread ) ) { + throw os_error_t( "thread_buffer_t::abort_thread: pthread_cancel failed", code ); + } + void * thread_result; + if( int code=pthread_join( m_thread, &thread_result ) ) { + throw os_error_t( "thread_buffer_t::stop_thread: pthread_join failed", code ); + } + m_thread_started=false; + } + return true; +} + +const int s_in_eof=16; +const int s_out_eof=32; +const int s_err_eof=64; + +void thread_buffer_t::get( exec_stream_t::stream_kind_t kind, char * dst, std::size_t & size, bool & no_more ) +{ + if( !m_thread_started ) { + throw exec_stream_t::error_t( "thread_buffer_t::get: thread was not started" ); + } + unsigned long timeout= kind==exec_stream_t::s_out ? m_out_wait_timeout : m_err_wait_timeout; + int eof_kind= kind==exec_stream_t::s_out ? s_out_eof : s_err_eof; + buffer_list_t & buffer= kind==exec_stream_t::s_out ? m_out_buffer : m_err_buffer; + + wait_result_t wait_result=m_thread_responce.wait( kind|exec_stream_t::s_child|eof_kind, timeout, 0 ); + if( !wait_result.ok() ) { + throw os_error_t( "thread_buffer_t::get: wait for got_data failed", wait_result.error_code() ); + } + + if( wait_result.is_signaled( exec_stream_t::s_child ) ) { + // thread stopped - no need to synchronize + if( !buffer.empty() ) { + // we have data - deliver it first + // when thread terminated, there is no need to synchronize + buffer.get( dst, size ); + no_more=false; + }else { + // thread terminated and we have no more data to return - report errors, if any + if( m_error_code!=0 ) { + throw os_error_t( m_error_prefix, m_error_code ); + } + // if terminated without error - signal eof + size=0; + no_more=true; + } + }else if( wait_result.is_signaled( kind|eof_kind ) ) { + // thread got some data for us - grab them + grab_mutex_t grab_mutex( m_mutex, 0 ); + if( !grab_mutex.ok() ) { + throw os_error_t( "thread_buffer_t::get: wait for mutex failed", grab_mutex.error_code() ); + } + + if( !buffer.empty() ) { + buffer.get( dst, size ); + no_more=false; + }else { + size=0; + no_more=wait_result.is_signaled( eof_kind ); + } + // if no data left - make the next get() wait until it arrives + if( buffer.empty() ) { + if( int code=m_thread_responce.reset( kind, 0 ) ) { + throw os_error_t( "thread_buffer_t::get: unable to reset got_data event", code ); + } + } + // if buffer is not too long tell the thread we want more data + std::size_t buffer_limit= kind==exec_stream_t::s_out ? m_out_buffer_limit : m_err_buffer_limit; + if( !buffer.full( buffer_limit ) ) { + if( int code=m_thread_control.set( kind, 0 ) ) { + throw os_error_t( "thread_buffer_t::get: unable to set want_data event", code ); + } + } + } +} + +void thread_buffer_t::put( char * src, std::size_t & size, bool & no_more ) +{ + if( !m_thread_started ) { + throw exec_stream_t::error_t( "thread_buffer_t::put: thread was not started" ); + } + if( m_in_closed || m_in_bad ) { + size=0; + no_more=true; + return; + } + // wait for both m_want_data and m_mutex + wait_result_t wait_result=m_thread_responce.wait( exec_stream_t::s_in|exec_stream_t::s_child, m_in_wait_timeout, 0 ); + if( !wait_result.ok() ) { + // workaround for versions of libstdc++ (at least in gcc 3.1 pre) that do not intercept exceptions in operator<<( std::ostream, std::string ) + m_in_bad=true; + if( m_in.exceptions()&std::ios_base::badbit ) { + throw os_error_t( "thread_buffer_t::put: wait for want_data failed", wait_result.error_code() ); + }else { + m_in.setstate( std::ios_base::badbit ); + size=0; + no_more=true; + return; + } + } + if( wait_result.is_signaled( exec_stream_t::s_child ) ) { + // thread stopped - check for errors + if( m_error_code!=0 ) { + throw os_error_t( m_error_prefix, m_error_code ); + } + // if terminated without error - signal eof, since no one will ever write our data + size=0; + no_more=true; + }else if( wait_result.is_signaled( exec_stream_t::s_in ) ) { + // thread wants some data from us - stuff them + grab_mutex_t grab_mutex( m_mutex, 0 ); + if( !grab_mutex.ok() ) { + throw os_error_t( "thread_buffer_t::put: wait for mutex failed", grab_mutex.error_code() ); + } + + no_more=false; + m_in_buffer.put( src, size ); + + // if the buffer is too long - make the next put() wait until it shrinks + if( m_in_buffer.full( m_in_buffer_limit ) ) { + if( int code=m_thread_responce.reset( exec_stream_t::s_in, 0 ) ) { + throw os_error_t( "thread_buffer_t::put: unable to reset want_data event", code ); + } + } + // tell the thread we got data + if( !m_in_buffer.empty() ) { + if( int code=m_thread_control.set( exec_stream_t::s_in, 0 ) ) { + throw os_error_t( "thread_buffer_t::put: unable to set got_data event", code ); + } + } + } +} + +void thread_buffer_t::close_in() +{ + if( !m_in_bad ) { + m_in.flush(); + } + if( m_thread_started ) { + if( int code=m_thread_control.set( s_in_eof, 0 ) ) { + throw os_error_t( "thread_buffer_t::close_in: unable to set in_got_data event", code ); + } + m_in_closed=true; + } +} + +void mutex_cleanup( void * p ) +{ + static_cast< mutex_registrator_t * >( p )->release_all(); +} + +void * thread_buffer_t::thread_func( void * param ) +{ + thread_buffer_t * p=static_cast< thread_buffer_t * >( param ); + // accessing p anywhere here is safe because thread_buffer_t destructor + // ensures the thread is terminated before p get destroyed + char * out_read_buffer=0; + char * err_read_buffer=0; + bool in_eof=false; + bool in_closed=false; + bool out_eof=false; + bool err_eof=false; + + mutex_registrator_t mutex_registrator; + pthread_cleanup_push( mutex_cleanup, &mutex_registrator ); + + try { + out_read_buffer=new char[p->m_out_read_buffer_size]; + err_read_buffer=new char[p->m_err_read_buffer_size]; + + buffer_list_t::buffer_t write_buffer; + write_buffer.data=0; + write_buffer.size=0; + std::size_t write_buffer_offset=0; + + unsigned long timeout=std::max( p->m_in_wait_timeout, std::max( p->m_out_wait_timeout, p->m_err_wait_timeout ) ); + + fd_set read_fds; + FD_ZERO( &read_fds ); + fd_set write_fds; + FD_ZERO( &write_fds ); + + while( true ) { + unsigned wait_for=exec_stream_t::s_child; + if( !in_eof && write_buffer.data==0 ) { + wait_for|=exec_stream_t::s_in|s_in_eof; + } + if( !out_eof ) { + wait_for|=exec_stream_t::s_out; + } + if( !err_eof ) { + wait_for|=exec_stream_t::s_err; + } + + wait_result_t wait_result=p->m_thread_control.wait( wait_for, timeout, &mutex_registrator ); + if( !wait_result.ok() && !wait_result.timed_out() ) { + p->m_error_code=wait_result.error_code(); + p->m_error_prefix="thread_buffer_t::thread_func: wait for thread_event failed"; + break; + } + + // we need more data - get from p->m_buffers + if( write_buffer.data==0 && wait_result.is_signaled( exec_stream_t::s_in|s_in_eof ) ) { + grab_mutex_t grab_mutex( p->m_mutex, &mutex_registrator ); + if( !grab_mutex.ok() ) { + p->m_error_code=grab_mutex.error_code(); + p->m_error_prefix="thread_buffer_t::thread_func: wait for mutex failed"; + break; + } + + if( p->m_in_buffer.empty() ) { + // we have empty write_buffer, empty p->m_in_buffer and we are told it will stay so - time to close child's stdin + if( wait_result.is_signaled( s_in_eof ) ) { + in_eof=true; + } + } + if( !p->m_in_buffer.empty() ) { + // we've got buffer - detach it + write_buffer=p->m_in_buffer.detach(); + write_buffer_offset=0; + } + // if no data left in p->m_in_buffer - wait until it arrives + if( p->m_in_buffer.empty() ) { + // if no data for us - stop trying to get it until we are told it arrived + if( int code=p->m_thread_control.reset( exec_stream_t::s_in, &mutex_registrator ) ) { + p->m_error_code=code; + p->m_error_prefix="thread_buffer_t::thread_func: unable to reset thread_event (s_in)"; + break; + } + } + + // if buffer is not too long - tell put() it can proceed + if( !p->m_in_buffer.full( p->m_in_buffer_limit ) ) { + if( int code=p->m_thread_responce.set( exec_stream_t::s_in, &mutex_registrator ) ) { + p->m_error_code=code; + p->m_error_prefix="thread_buffer_t::thread_func: unable to set in_want_data event"; + break; + } + } + } + + if( in_eof && write_buffer.data==0 ) { + p->m_in_pipe.close(); + in_closed=true; + } + + // see if they want us to stop, but only when there is nothing more to write + if( write_buffer.data==0 && wait_result.is_signaled( exec_stream_t::s_child ) ) { + break; + } + + // determine whether we want something + if( write_buffer.data!=0 ) { + FD_SET( p->m_in_pipe.w(), &write_fds ); + }else { + FD_CLR( p->m_in_pipe.w(), &write_fds ); + } + if( !out_eof && wait_result.is_signaled( exec_stream_t::s_out ) ) { + FD_SET( p->m_out_pipe.r(), &read_fds ); + }else { + FD_CLR( p->m_out_pipe.r(), &read_fds ); + } + if( !err_eof && wait_result.is_signaled( exec_stream_t::s_err ) ) { + FD_SET( p->m_err_pipe.r(), &read_fds ); + }else { + FD_CLR( p->m_err_pipe.r(), &read_fds ); + } + + if( FD_ISSET( p->m_in_pipe.w(), &write_fds ) || FD_ISSET( p->m_out_pipe.r(), &read_fds ) || FD_ISSET( p->m_err_pipe.r(), &read_fds ) ) { + // we want something - get it + struct timeval select_timeout; + select_timeout.tv_sec=0; + select_timeout.tv_usec=100000; + int nfds=std::max( p->m_in_pipe.w(), std::max( p->m_out_pipe.r(), p->m_err_pipe.r() ) )+1; + if( select( nfds, &read_fds, &write_fds, 0, &select_timeout )==-1 ) { + p->m_error_code=errno; + p->m_error_prefix="thread_buffer_t::thread_func: select failed"; + break; + } + } + + // determine what we got + + if( FD_ISSET( p->m_in_pipe.w(), &write_fds ) ) { + // it seems we may write to child's stdin + int n_written=write( p->m_in_pipe.w(), write_buffer.data+write_buffer_offset, write_buffer.size-write_buffer_offset ); + if( n_written==-1 ) { + if( errno!=EAGAIN ) { + p->m_error_code=errno; + p->m_error_prefix="thread_buffer_t::thread_func: write to child stdin failed"; + break; + } + }else { + write_buffer_offset+=n_written; + if( write_buffer_offset==write_buffer.size ) { + delete[] write_buffer.data; + write_buffer.data=0; + write_buffer.size=0; + } + } + } + + if( FD_ISSET( p->m_out_pipe.r(), &read_fds ) ) { + // it seems we may read child's stdout + int n_out_read=read( p->m_out_pipe.r(), out_read_buffer, p->m_out_read_buffer_size ); + if( n_out_read==-1 ) { + if( errno!=EAGAIN ) { + p->m_error_code=errno; + p->m_error_prefix="exec_stream_t::thread_func: read from child stdout failed"; + break; + } + }else { + grab_mutex_t grab_mutex( p->m_mutex, &mutex_registrator ); + if( n_out_read!=0 ) { + p->m_out_buffer.put( out_read_buffer, n_out_read ); + // if buffer is full - stop reading + if( p->m_out_buffer.full( p->m_out_buffer_limit ) ) { + if( int code=p->m_thread_control.reset( exec_stream_t::s_out, &mutex_registrator ) ) { + p->m_error_code=code; + p->m_error_prefix="exec_stream_t::thread_func: unable to reset m_out_want_data event"; + break; + } + } + } + unsigned responce=exec_stream_t::s_out; + if( n_out_read==0 ) { // EOF when read 0 bytes while select told that it's ready + out_eof=true; + responce|=s_out_eof; + } + // we got either data or eof - tell always + if( int code=p->m_thread_responce.set( responce, &mutex_registrator ) ) { + p->m_error_code=code; + p->m_error_prefix="exec_stream_t::thread_func: unable to set out_got_data event"; + break; + } + } + } + + if( FD_ISSET( p->m_err_pipe.r(), &read_fds ) ) { + // it seemds we may read child's stderr + int n_err_read=read( p->m_err_pipe.r(), err_read_buffer, p->m_err_read_buffer_size ); + if( n_err_read==-1 ) { + if( errno!=EAGAIN ) { + p->m_error_code=errno; + p->m_error_prefix="exec_stream_t::thread_func: read from child stderr failed"; + break; + } + }else { + grab_mutex_t grab_mutex( p->m_mutex, &mutex_registrator ); + if( n_err_read!=0 ) { + p->m_err_buffer.put( err_read_buffer, n_err_read ); + // if buffer is full - stop reading + if( p->m_err_buffer.full( p->m_err_buffer_limit ) ) { + if( int code=p->m_thread_control.reset( exec_stream_t::s_err, &mutex_registrator ) ) { + p->m_error_code=code; + p->m_error_prefix="exec_stream_t::thread_func: unable to reset m_err_want_data event"; + break; + } + } + } + unsigned responce=exec_stream_t::s_err; + if( n_err_read==0 ) { + err_eof=true; + responce|=s_err_eof; + } + // we got either data or eof - tell always + if( int code=p->m_thread_responce.set( responce, &mutex_registrator ) ) { + p->m_error_code=code; + p->m_error_prefix="exec_stream_t::thread_func: unable to set err_got_data event"; + break; + } + } + } + + if( in_closed && out_eof && err_eof ) { + // have nothing more to do + break; + } + } + + delete[] write_buffer.data; + + }catch( ... ) { + // might only be std::bad_alloc + p->m_error_code=0; + p->m_error_prefix="thread_buffer_t::writer_thread: exception caught"; + } + + delete[] out_read_buffer; + delete[] err_read_buffer; + + // tell everyone that we've stopped, so that get() and put() will be unblocked + if( int code=p->m_thread_responce.set( exec_stream_t::s_child, &mutex_registrator ) ) { + p->m_error_code=code; + p->m_error_prefix="exec_stream_t::thread_func: unable to set thread_stopped event"; + } + + pthread_cleanup_pop( 0 ); + return 0; +} diff --git a/ext/libexecstream/posix/exec-stream-helpers.h b/ext/libexecstream/posix/exec-stream-helpers.h new file mode 100644 index 000000000..13f1eaae5 --- /dev/null +++ b/ext/libexecstream/posix/exec-stream-helpers.h @@ -0,0 +1,239 @@ +/* +Copyright (C) 2004 Artem Khodush + +Redistribution and use in source and binary forms, with or without modification, +are permitted provided that the following conditions are met: + +1. Redistributions of source code must retain the above copyright notice, +this list of conditions and the following disclaimer. + +2. Redistributions in binary form must reproduce the above copyright notice, +this list of conditions and the following disclaimer in the documentation +and/or other materials provided with the distribution. + +3. The name of the author may not be used to endorse or promote products +derived from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED +WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES +OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. +IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; +OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, +WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR +OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, +EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ + +class os_error_t : public exec_stream_t::error_t { +public: + os_error_t( std::string const & msg ); + os_error_t( std::string const & msg, exec_stream_t::error_code_t code ); +private: + void compose( std::string const & msg, exec_stream_t::error_code_t code ); +}; + + +template< class T > class buf_t { +public: + typedef T data_t; + + buf_t() + { + m_buf=0; + m_size=0; + } + + ~buf_t() + { + delete [] m_buf; + } + + data_t * new_data( std::size_t size ) + { + m_buf=new T[size]; + m_size=size; + return m_buf; + } + + void append_data( data_t const * data, std::size_t size ) + { + buf_t new_buf; + new_buf.new_data( m_size+size ); + std::char_traits< data_t >::copy( new_buf.m_buf, m_buf, m_size ); + std::char_traits< data_t >::copy( new_buf.m_buf+m_size, data, size ); + std::swap( this->m_buf, new_buf.m_buf ); + std::swap( this->m_size, new_buf.m_size ); + } + + data_t * data() + { + return m_buf; + } + + std::size_t size() + { + return m_size; + } + +private: + buf_t( buf_t const & ); + buf_t & operator=( buf_t const & ); + + data_t * m_buf; + std::size_t m_size; +}; + + +class pipe_t { +public: + pipe_t(); + ~pipe_t(); + int r() const; + int w() const; + void close_r(); + void close_w(); + void close(); + void open(); +private: + enum direction_t{ closed, read, write, both }; + direction_t m_direction; + int m_fds[2]; +}; + + +class mutex_t { +public: + mutex_t(); + ~mutex_t(); + +private: + pthread_mutex_t m_mutex; + + friend class event_t; + friend class grab_mutex_t; +}; + + +class grab_mutex_t { +public: + grab_mutex_t( mutex_t & mutex, class mutex_registrator_t * mutex_registrator ); + ~grab_mutex_t(); + + int release(); + bool ok(); + int error_code(); + +private: + pthread_mutex_t * m_mutex; + int m_error_code; + bool m_grabbed; + class mutex_registrator_t * m_mutex_registrator; + + friend class mutex_registrator_t; +}; + +class mutex_registrator_t { +public: + ~mutex_registrator_t(); + void add( grab_mutex_t * g ); + void remove( grab_mutex_t * g ); + void release_all(); +private: + typedef std::list< grab_mutex_t * > mutexes_t; + mutexes_t m_mutexes; +}; + + +class wait_result_t { +public: + wait_result_t( unsigned signaled_state, int error_code, bool timed_out ); + + bool ok(); + bool is_signaled( int state ); + int error_code(); + bool timed_out(); + +private: + unsigned m_signaled_state; + int m_error_code; + bool m_timed_out; +}; + + +class event_t { +public: + event_t(); + ~event_t(); + + int set( unsigned bits, mutex_registrator_t * mutex_registrator ); + int reset( unsigned bits, mutex_registrator_t * mutex_registrator ); + + wait_result_t wait( unsigned any_bits, unsigned long timeout, mutex_registrator_t * mutex_registrator ); + +private: + mutex_t m_mutex; + pthread_cond_t m_cond; + unsigned volatile m_state; +}; + + +class thread_buffer_t { +public: + thread_buffer_t( pipe_t & in_pipe, pipe_t & out_pipe, pipe_t & err_pipe, std::ostream & in ); + ~thread_buffer_t(); + + void set_wait_timeout( int stream_kind, unsigned long milliseconds ); + void set_buffer_limit( int stream_kind, std::size_t limit ); + void set_read_buffer_size( int stream_kind, std::size_t size ); + + void start(); + + void get( exec_stream_t::stream_kind_t kind, char * dst, std::size_t & size, bool & no_more ); + void put( char * src, std::size_t & size, bool & no_more ); + + void close_in(); + bool stop_thread(); + bool abort_thread(); + +private: + static void * thread_func( void * param ); + + pthread_t m_thread; + mutex_t m_mutex; // protecting m_in_buffer, m_out_buffer, m_err_buffer + + buffer_list_t m_in_buffer; + buffer_list_t m_out_buffer; + buffer_list_t m_err_buffer; + + event_t m_thread_control; // s_in : in got_data; s_out: out want data; s_err: err want data; s_child: stop thread + event_t m_thread_responce; // s_in : in want data; s_out: out got data; s_err: err got data; s_child: thread stopped + + char const * m_error_prefix; + int m_error_code; + + bool m_thread_started; // set in start(), checked in set_xxx(), get() and put() + bool m_in_closed; // set in close_in(), checked in put() + + pipe_t & m_in_pipe; + pipe_t & m_out_pipe; + pipe_t & m_err_pipe; + + unsigned long m_in_wait_timeout; + unsigned long m_out_wait_timeout; + unsigned long m_err_wait_timeout; + + unsigned long m_thread_termination_timeout; + + std::size_t m_in_buffer_limit; + std::size_t m_out_buffer_limit; + std::size_t m_err_buffer_limit; + + std::size_t m_out_read_buffer_size; + std::size_t m_err_read_buffer_size; + + // workaround for not-quite-conformant libstdc++ (see put()) + std::ostream & m_in; + bool m_in_bad; +}; diff --git a/ext/libexecstream/posix/exec-stream-impl.cpp b/ext/libexecstream/posix/exec-stream-impl.cpp new file mode 100644 index 000000000..c45253398 --- /dev/null +++ b/ext/libexecstream/posix/exec-stream-impl.cpp @@ -0,0 +1,386 @@ +/* +Copyright (C) 2004 Artem Khodush + +Redistribution and use in source and binary forms, with or without modification, +are permitted provided that the following conditions are met: + +1. Redistributions of source code must retain the above copyright notice, +this list of conditions and the following disclaimer. + +2. Redistributions in binary form must reproduce the above copyright notice, +this list of conditions and the following disclaimer in the documentation +and/or other materials provided with the distribution. + +3. The name of the author may not be used to endorse or promote products +derived from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED +WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES +OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. +IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; +OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, +WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR +OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, +EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ + +// exec_stream_t::impl_t +struct exec_stream_t::impl_t { + impl_t(); + ~impl_t(); + + void split_args( std::string const & program, std::string const & arguments ); + void split_args( std::string const & program, exec_stream_t::next_arg_t & next_arg ); + void start( std::string const & program ); + + pid_t m_child_pid; + int m_exit_code; + unsigned long m_child_timeout; + + buf_t< char > m_child_args; + buf_t< char * > m_child_argp; + + pipe_t m_in_pipe; + pipe_t m_out_pipe; + pipe_t m_err_pipe; + + thread_buffer_t m_thread; + + exec_stream_buffer_t m_in_buffer; + exec_stream_buffer_t m_out_buffer; + exec_stream_buffer_t m_err_buffer; + + exec_ostream_t m_in; + exec_istream_t m_out; + exec_istream_t m_err; + + void (*m_old_sigpipe_handler)(int); +}; + +exec_stream_t::impl_t::impl_t() +: m_thread( m_in_pipe, m_out_pipe, m_err_pipe, m_in ), /* m_in here is not initialized, but its ok */ + m_in_buffer( exec_stream_t::s_in, m_thread ), m_out_buffer( exec_stream_t::s_out, m_thread ), m_err_buffer( exec_stream_t::s_err, m_thread ), + m_in( m_in_buffer ), m_out( m_out_buffer ), m_err( m_err_buffer ) +{ + m_out.tie( &m_in ); + m_err.tie( &m_in ); + m_child_timeout=1000; + m_child_pid=-1; + m_old_sigpipe_handler=signal( SIGPIPE, SIG_IGN ); +} + +exec_stream_t::impl_t::~impl_t() +{ + signal( SIGPIPE, m_old_sigpipe_handler ); +} + +void exec_stream_t::impl_t::split_args( std::string const & program, std::string const & arguments ) +{ + char * args_end=m_child_args.new_data( program.size()+1+arguments.size()+1 ); + int argc=1; + + std::string::traits_type::copy( args_end, program.data(), program.size() ); + args_end+=program.size(); + *args_end++=0; + + std::string whitespace=" \t\r\n\v"; + + std::string::size_type arg_start=arguments.find_first_not_of( whitespace ); + while( arg_start!=std::string::npos ) { + ++argc; + std::string::size_type arg_stop; + if( arguments[arg_start]!='"' ) { + arg_stop=arguments.find_first_of( whitespace, arg_start ); + if( arg_stop==std::string::npos ) { + arg_stop=arguments.size(); + } + std::string::traits_type::copy( args_end, arguments.data()+arg_start, arg_stop-arg_start ); + args_end+=arg_stop-arg_start; + }else { + std::string::size_type cur=arg_start+1; + while( true ) { + std::string::size_type next=arguments.find( '"', cur ); + if( next==std::string::npos || arguments[next-1]!='\\' ) { + if( next==std::string::npos ) { + next=arguments.size(); + arg_stop=next; + }else { + arg_stop=next+1; + } + std::string::traits_type::copy( args_end, arguments.data()+cur, next-cur ); + args_end+=next-cur; + break; + }else { + std::string::traits_type::copy( args_end, arguments.data()+cur, next-1-cur ); + args_end+=next-1-cur; + *args_end++='"'; + cur=next+1; + } + } + } + *args_end++=0; + arg_start=arguments.find_first_not_of( whitespace, arg_stop ); + } + + char ** argp_end=m_child_argp.new_data( argc+1 ); + char * args=m_child_args.data(); + while( args!=args_end ) { + *argp_end=args; + args+=std::string::traits_type::length( args )+1; + ++argp_end; + } + *argp_end=0; +} + +void exec_stream_t::impl_t::split_args( std::string const & program, exec_stream_t::next_arg_t & next_arg ) +{ + typedef std::vector< std::size_t > arg_sizes_t; + arg_sizes_t arg_sizes; + + m_child_args.new_data( program.size()+1 ); + std::string::traits_type::copy( m_child_args.data(), program.c_str(), program.size()+1 ); + arg_sizes.push_back( program.size()+1 ); + + while( std::string const * s=next_arg.next() ) { + m_child_args.append_data( s->c_str(), s->size()+1 ); + arg_sizes.push_back( s->size()+1 ); + } + + char ** argp_end=m_child_argp.new_data( arg_sizes.size()+1 ); + char * argp=m_child_args.data(); + for( arg_sizes_t::iterator i=arg_sizes.begin(); i!=arg_sizes.end(); ++i ) { + *argp_end=argp; + argp+=*i; + ++argp_end; + } + *argp_end=0; +} + +void exec_stream_t::set_buffer_limit( int stream_kind, std::size_t size ) +{ + m_impl->m_thread.set_buffer_limit( stream_kind, size ); +} + +void exec_stream_t::set_wait_timeout( int stream_kind, timeout_t milliseconds ) +{ + m_impl->m_thread.set_wait_timeout( stream_kind, milliseconds ); + if( stream_kind&exec_stream_t::s_child ) { + m_impl->m_child_timeout=milliseconds; + } +} + +void exec_stream_t::start( std::string const & program, std::string const & arguments ) +{ + if( !close() ) { + throw exec_stream_t::error_t( "exec_stream_t::start: previous child process has not yet terminated" ); + } + + m_impl->split_args( program, arguments ); + m_impl->start( program ); +} + +void exec_stream_t::start( std::string const & program, exec_stream_t::next_arg_t & next_arg ) +{ + if( !close() ) { + throw exec_stream_t::error_t( "exec_stream_t::start: previous child process has not yet terminated" ); + } + + m_impl->split_args( program, next_arg ); + m_impl->start( program ); +} + +void exec_stream_t::impl_t::start( std::string const & program ) +{ + m_in_pipe.open(); + m_out_pipe.open(); + m_err_pipe.open(); + + pipe_t status_pipe; + status_pipe.open(); + + pid_t pid=fork(); + if( pid==-1 ) { + throw os_error_t( "exec_stream_t::start: fork failed" ); + }else if( pid==0 ) { + try { + status_pipe.close_r(); + if( fcntl( status_pipe.w(), F_SETFD, FD_CLOEXEC )==-1 ) { + throw os_error_t( "exec_stream_t::start: unable to fcnth( status_pipe, F_SETFD, FD_CLOEXEC ) in child process" ); + } + m_in_pipe.close_w(); + m_out_pipe.close_r(); + m_err_pipe.close_r(); + if( ::close( 0 )==-1 ) { + throw os_error_t( "exec_stream_t::start: unable to close( 0 ) in child process" ); + } + if( fcntl( m_in_pipe.r(), F_DUPFD, 0 )==-1 ) { + throw os_error_t( "exec_stream_t::start: unable to fcntl( .., F_DUPFD, 0 ) in child process" ); + } + if( ::close( 1 )==-1 ) { + throw os_error_t( "exec_stream_t::start: unable to close( 1 ) in child process" ); + } + if( fcntl( m_out_pipe.w(), F_DUPFD, 1 )==-1 ) { + throw os_error_t( "exec_stream_t::start: unable to fcntl( .., F_DUPFD, 1 ) in child process" ); + } + if( ::close( 2 )==-1 ) { + throw os_error_t( "exec_stream_t::start: unable to close( 2 ) in child process" ); + } + if( fcntl( m_err_pipe.w(), F_DUPFD, 2 )==-1 ) { + throw os_error_t( "exec_stream_t::start: unable to fcntl( .., F_DUPFD, 2 ) in child process" ); + } + m_in_pipe.close_r(); + m_out_pipe.close_w(); + m_err_pipe.close_w(); + if( execvp( m_child_args.data(), m_child_argp.data() )==-1 ) { + throw os_error_t( "exec_stream_t::start: exec in child process failed. "+program ); + } + throw exec_stream_t::error_t( "exec_stream_t::start: exec in child process returned" ); + }catch( std::exception const & e ) { + const char * msg=e.what(); + std::size_t len=strlen( msg ); + write( status_pipe.w(), &len, sizeof( len ) ); + write( status_pipe.w(), msg, len ); + _exit( -1 ); + }catch( ... ) { + char * msg="exec_stream_t::start: unknown exception in child process"; + std::size_t len=strlen( msg ); + write( status_pipe.w(), &len, sizeof( len ) ); + write( status_pipe.w(), msg, len ); + _exit( 1 ); + } + }else { + m_child_pid=pid; + status_pipe.close_w(); + fd_set status_fds; + FD_ZERO( &status_fds ); + FD_SET( status_pipe.r(), &status_fds ); + struct timeval timeout; + timeout.tv_sec=3; + timeout.tv_usec=0; + if( select( status_pipe.r()+1, &status_fds, 0, 0, &timeout )==-1 ) { + throw os_error_t( "exec_stream_t::start: select on status_pipe failed" ); + } + if( !FD_ISSET( status_pipe.r(), &status_fds ) ) { + throw os_error_t( "exec_stream_t::start: timeout while waiting for child to report via status_pipe" ); + } + std::size_t status_len; + int status_nread=read( status_pipe.r(), &status_len, sizeof( status_len ) ); + // when all ok, status_pipe is closed on child's exec, and nothing is written to it + if( status_nread!=0 ) { + // otherwize, check what went wrong. + if( status_nread==-1 ) { + throw os_error_t( "exec_stream_t::start: read from status pipe failed" ); + }else if( status_nread!=sizeof( status_len ) ) { + throw os_error_t( "exec_stream_t::start: unable to read length of status message from status_pipe" ); + } + std::string status_msg; + if( status_len!=0 ) { + buf_t< char > status_buf; + status_buf.new_data( status_len ); + status_nread=read( status_pipe.r(), status_buf.data(), status_len ); + if( status_nread==-1 ) { + throw os_error_t( "exec_stream_t::start: readof status message from status pipe failed" ); + } + status_msg.assign( status_buf.data(), status_len ); + } + throw exec_stream_t::error_t( "exec_stream_t::start: error in child process."+status_msg ); + } + status_pipe.close_r(); + + m_in_pipe.close_r(); + m_out_pipe.close_w(); + m_err_pipe.close_w(); + + if( fcntl( m_in_pipe.w(), F_SETFL, O_NONBLOCK )==-1 ) { + throw os_error_t( "exec_stream_t::start: fcntl( in_pipe, F_SETFL, O_NONBLOCK ) failed" ); + } + + m_in_buffer.clear(); + m_out_buffer.clear(); + m_err_buffer.clear(); + + m_in.clear(); + m_out.clear(); + m_err.clear(); + + m_thread.set_read_buffer_size( exec_stream_t::s_out, STREAM_BUFFER_SIZE ); + m_thread.set_read_buffer_size( exec_stream_t::s_err, STREAM_BUFFER_SIZE ); + m_thread.start(); + } +} + +bool exec_stream_t::close_in() +{ + m_impl->m_thread.close_in(); + return true; +} + +bool exec_stream_t::close() +{ + close_in(); + if( !m_impl->m_thread.stop_thread() ) { + m_impl->m_thread.abort_thread(); + } + m_impl->m_in_pipe.close(); + m_impl->m_out_pipe.close(); + m_impl->m_err_pipe.close(); + + if( m_impl->m_child_pid!=-1 ) { + pid_t code=waitpid( m_impl->m_child_pid, &m_impl->m_exit_code, WNOHANG ); + if( code==-1 ) { + throw os_error_t( "exec_stream_t::close: first waitpid failed" ); + }else if( code==0 ) { + + struct timeval select_timeout; + select_timeout.tv_sec=m_impl->m_child_timeout/1000; + select_timeout.tv_usec=(m_impl->m_child_timeout%1000)*1000; + if( (code=select( 0, 0, 0, 0, &select_timeout ))==-1 ) { + throw os_error_t( "exec_stream_t::close: select failed" ); + } + + code=waitpid( m_impl->m_child_pid, &m_impl->m_exit_code, WNOHANG ); + if( code==-1 ) { + throw os_error_t( "exec_stream_t::close: second waitpid failed" ); + }else if( code==0 ) { + return false; + }else { + m_impl->m_child_pid=-1; + return true; + } + + }else { + m_impl->m_child_pid=-1; + return true; + } + } + return true; +} + +void exec_stream_t::kill() +{ + if( m_impl->m_child_pid!=-1 ) { + if( ::kill( m_impl->m_child_pid, SIGKILL )==-1 ) { + throw os_error_t( "exec_stream_t::kill: kill failed" ); + } + m_impl->m_child_pid=-1; + m_impl->m_exit_code=0; + } +} + +int exec_stream_t::exit_code() +{ + if( m_impl->m_child_pid!=-1 ) { + throw exec_stream_t::error_t( "exec_stream_t::exit_code: child process still running" ); + } + return WEXITSTATUS( m_impl->m_exit_code ); +} + +void exec_stream_t::set_binary_mode( int ) +{ +} + +void exec_stream_t::set_text_mode( int ) +{ +} diff --git a/ext/libexecstream/test/Makefile.gcc b/ext/libexecstream/test/Makefile.gcc new file mode 100644 index 000000000..b686f4075 --- /dev/null +++ b/ext/libexecstream/test/Makefile.gcc @@ -0,0 +1,21 @@ + +builddir=build +srcdir=.. + +CC=g++ + +CDEBUGFLAGS= -g -D _DEBUG + +CFLAGS= -I .. -Wall $(CDEBUGFLAGS) + +all : $(builddir)/exec-stream-test + +$(builddir)/exec-stream.o : $(srcdir)/exec-stream.cpp $(srcdir)/exec-stream.h $(srcdir)/posix/exec-stream-helpers.h $(srcdir)/posix/exec-stream-helpers.cpp $(srcdir)/posix/exec-stream-impl.cpp + $(CC) -c $(CFLAGS) -o $(builddir)/exec-stream.o $(srcdir)/exec-stream.cpp + +$(builddir)/exec-stream-test.o : ./exec-stream-test.cpp $(srcdir)/exec-stream.h + $(CC) -c $(CFLAGS) -o $(builddir)/exec-stream-test.o ./exec-stream-test.cpp + +$(builddir)/exec-stream-test : $(builddir)/exec-stream.o $(builddir)/exec-stream-test.o + $(CC) $(CFLAGS) -o $(builddir)/exec-stream-test $(builddir)/exec-stream-test.o $(builddir)/exec-stream.o -lpthread + diff --git a/ext/libexecstream/test/Makefile.msvc b/ext/libexecstream/test/Makefile.msvc new file mode 100644 index 000000000..57a8b15b7 --- /dev/null +++ b/ext/libexecstream/test/Makefile.msvc @@ -0,0 +1,19 @@ + +builddir=build +srcdir=.. + +CLDEBUGFLAG=/Zi /D "_DEBUG" +LIBDEBUGFLAG=d +CLFLAGS=/nologo /I .. /W3 /EHsc /MD$(LIBDEBUGFLAG) $(CLDEBUGFLAG) /Fo"$(builddir)/" /Fe"$(builddir)/" /Fd"$(builddir)/exec-stream-test.pdb" + +all : $(builddir)/exec-stream-test.exe + +$(builddir)/exec-stream.obj : $(srcdir)/exec-stream.cpp $(srcdir)/exec-stream.h $(srcdir)/win/exec-stream-helpers.h $(srcdir)/win/exec-stream-helpers.cpp $(srcdir)/win/exec-stream-impl.cpp + cl /c $(CLFLAGS) $(srcdir)/exec-stream.cpp + +$(builddir)/exec-stream-test.obj : ./exec-stream-test.cpp $(srcdir)/exec-stream.h + cl /c $(CLFLAGS) ./exec-stream-test.cpp + +$(builddir)/exec-stream-test.exe : $(builddir)/exec-stream.obj $(builddir)/exec-stream-test.obj + cl $(CLFLAGS) $(builddir)/exec-stream-test.obj $(builddir)/exec-stream.obj /link /subsystem:console + diff --git a/ext/libexecstream/test/exec-stream-test.cpp b/ext/libexecstream/test/exec-stream-test.cpp new file mode 100644 index 000000000..050cd77ba --- /dev/null +++ b/ext/libexecstream/test/exec-stream-test.cpp @@ -0,0 +1,944 @@ +/* +Copyright (C) 2004 Artem Khodush + +Redistribution and use in source and binary forms, with or without modification, +are permitted provided that the following conditions are met: + +1. Redistributions of source code must retain the above copyright notice, +this list of conditions and the following disclaimer. + +2. Redistributions in binary form must reproduce the above copyright notice, +this list of conditions and the following disclaimer in the documentation +and/or other materials provided with the distribution. + +3. The name of the author may not be used to endorse or promote products +derived from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED +WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES +OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. +IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; +OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, +WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR +OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, +EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ + +#include "exec-stream.h" + +#include +#include +#include +#include +#include +#include + +#include +#include + +#ifdef _WIN32 + +#include + +void sleep( int seconds ) +{ + Sleep( seconds*1000 ); +} + +#else + +#include + +#endif + +// class for collecting and printing test results + +class test_results_t { +public: + static void add_test( std::string const & name ); + static bool register_failure( std::string const & assertion, std::string const & file, int line ); + static int print( std::ostream & o ); + + class error_t : public std::exception { + public: + error_t( std::string const & msg ) + : m_msg( msg ) + { + } + + ~error_t() throw() + { + } + + virtual char const * what() const throw() + { + return m_msg.c_str(); + } + + private: + std::string m_msg; + }; + +private: + struct failure_t { + std::string assertion; + std::string file; + int line; + }; + typedef std::vector< failure_t > failures_t; + struct result_t { + std::string test_name; + failures_t failures; + }; + typedef std::vector< result_t > results_t; + + static results_t m_results; + static bool m_printed; + + struct force_print_t { + ~force_print_t(); + }; + friend struct force_print_t; + + static force_print_t m_force_print; +}; + +test_results_t::results_t test_results_t::m_results; +bool test_results_t::m_printed; +test_results_t::force_print_t test_results_t::m_force_print; + +void test_results_t::add_test( std::string const & name ) +{ + results_t::iterator i=m_results.begin(); + while( i!=m_results.end() && i->test_name!=name ) { + ++i; + } + if( i!=m_results.end() ) { + throw error_t( "test_results_t::add_test: duplicate test name: "+name ); + } + result_t & r=*m_results.insert( m_results.end(), result_t() ); + r.test_name=name; +} + +bool test_results_t::register_failure( std::string const & assertion, std::string const & file, int line ) +{ + if( m_results.empty() ) { + throw error_t( "test_results_t::register_failure: called without add_test() first" ); + } + failures_t & failures=m_results.back().failures; + failure_t & failure=*failures.insert( failures.end(), failure_t() ); + failure.assertion=assertion; + failure.file=file; + failure.line=line; + return true; +} + +int test_results_t::print( std::ostream & o ) +{ + int n_ok=0; + int n_failed=0; + for( results_t::iterator res_i=m_results.begin(); res_i!=m_results.end(); ++res_i ) { + if( res_i->failures.empty() ) { + ++n_ok; + }else { + o<<"FAILED TEST: "<test_name<<"\n"; + for( failures_t::iterator fail_i=res_i->failures.begin(); fail_i!=res_i->failures.end(); ++fail_i ) { + o<<" "<assertion<<" [at file "<file<<" line "<line<<"]\n"; + } + ++n_failed; + } + } + o<<"\nOK: "<>s; + if( s==random_string( 20000 ) ) { + std::cerr<<"OK"; + }else { + std::cerr<<"ERROR"; + } + return 0; +} + +int dont_read() +{ + return 0; +} + +int dont_stop() +{ + std::string s=random_string( 100 )+"\n"; + while( true ) { + std::cout<< s; + } + return 0; +} + +int echo_size() { + std::string s; + while( std::getline( std::cin, s ).good() ) { + std::cout<0 && buf[len-1]=='\n' ) { + --len; + } + out_s.assign( buf, len ); + if( out_s!=in_s ) { + fputs( "ERROR", stderr ); + return 0; + } + ++n; + } + if( n==1200 ) { + fputs( "OK", stderr ); + }else { + fputs( "ERROR", stderr ); + } + return 0; +} + +int long_out_line() +{ + std::cout<>s; + ss.str( s ); + int n; + ss>>n; + return n; +} + +// selection of child functions + +typedef int(*child_func_t)(); + +child_func_t find_child_func( std::string const & name ) +{ + typedef std::map< std::string, child_func_t > child_funcs_t; + static child_funcs_t funcs; + if( funcs.size()==0 ) { + funcs["hello"]=hello; + funcs["helloworld"]=helloworld; + funcs["hello-o-world-e"]=hello_o_world_e; + funcs["with space"]=with_space; + funcs["write-after-pause"]=write_after_pause; + funcs["read-after-pause"]=read_after_pause; + funcs["dont-read"]=dont_read; + funcs["dont-stop"]=dont_stop; + funcs["echo-size"]=echo_size; + funcs["echo"]=echo; + funcs["echo-with-err"]=echo_with_err; + funcs["echo-picky"]=echo_picky; + funcs["pathologic"]=pathologic; + funcs["long-out-line"]=long_out_line; + funcs["exit-code"]=exit_code; + } + child_funcs_t::iterator i=funcs.find( name ); + return i==funcs.end() ? 0 : i->second; +} + +void pathologic_one( exec_stream_t & exec_stream, bool expect_ok=true ) +{ + std::string in_s=random_string( 1000 ); + int cnt=1200; + for( int i=0; i>out_s; + if( expect_ok ) { + TEST( out_s=="OK" ); + }else { + TEST( out_s=="ERROR" ); + } +} + +int main( int argc, char ** argv ) +{ + if( argc>=3 ) { + if( std::string( argv[1] )=="child" ) { + if( child_func_t func=find_child_func( argv[2] ) ) { + return func(); + } + }else if( std::string( argv[1] )=="args" ) { + for( int i=2; iout world->err" ); + exec_stream_t exec_stream( program, "child hello-o-world-e" ); + TEST( read_all( exec_stream.out() )=="hello" ); + TEST( read_all( exec_stream.err() )=="world" ); + } + + { + TEST_NAME( "hello->out world->err read reversed" ); + exec_stream_t exec_stream( program, "child hello-o-world-e" ); + TEST( read_all( exec_stream.err() )=="world" ); + TEST( read_all( exec_stream.out() )=="hello" ); + } + + { + TEST_NAME( "hello 5 times" ); + exec_stream_t exec_stream; + + exec_stream.start( program, "child helloworld" ); + TEST( read_all( exec_stream.out() )=="hello\nworld" ); + TEST( read_all( exec_stream.err() )=="" ); + + exec_stream.start( program, "child helloworld" ); + // do not read, leave all in buffers + + exec_stream.start( program, "child hello-o-world-e" ); + + TEST( read_all( exec_stream.out() )=="hello" ); + TEST( read_all( exec_stream.err() )=="world" ); + + exec_stream.start( program, "child helloworld" ); + // do not read all, leave something in buffers + std::string s; + std::getline( exec_stream.out(), s ); + + TEST( s=="hello" ); + + exec_stream.start( program, "child hello-o-world-e" ); + + TEST( read_all( exec_stream.out() )=="hello" ); + TEST( read_all( exec_stream.err() )=="world" ); + + TEST( exec_stream.close() ); + } + + { + TEST_NAME( "with space" ); + std::vector< std::string > args; + args.push_back( "child" ); + args.push_back( "with space" ); + exec_stream_t exec_stream( program, args.begin(), args.end() ); + std::string s; + std::getline( exec_stream.out(), s ); + TEST( s=="with space ok" ); + TEST( exec_stream.close() ); + } + + { + TEST_NAME( "child write after pause" ); + exec_stream_t exec_stream; + exec_stream.start( program, "child write-after-pause" ); + try { + std::string s; + std::getline( exec_stream.out(), s ); + TEST( 0=="unreached" ); + }catch( std::exception const & e ) { + std::string m=e.what(); + if( english_error_messages ) { + TEST( m.find( "timeout" )!=std::string::npos || m.find( "timed out" )!=std::string::npos ); + } + } + TEST( !exec_stream.close() ); + exec_stream.kill(); + + exec_stream.set_wait_timeout( exec_stream_t::s_out, 12000 ); + exec_stream.start( program, "child write-after-pause" ); + std::string s; + std::getline( exec_stream.out(), s ); + TEST( s=="after pause" ); + + TEST( exec_stream.close() ); + } + + { + TEST_NAME( "child read after pause" ); + exec_stream_t exec_stream; + exec_stream.set_wait_timeout( exec_stream_t::s_err, 15000 ); + std::string in_s=random_string( 20000 )+"\n"; + + exec_stream.start( program, "child read-after-pause" ); + exec_stream.in()<>s; + TEST( s=="OK" ); + TEST( exec_stream.close() ); + + exec_stream.set_buffer_limit( exec_stream_t::s_in, 1000 ); + exec_stream.start( program, "child read-after-pause" ); + try { + exec_stream.in()<>s; + TEST( s=="OK" ); + TEST( exec_stream.close() ); + } + + { + TEST_NAME( "dont read" ); + exec_stream_t exec_stream; + exec_stream.start( program, "child dont-read" ); + try { + exec_stream.in()<>out; + TEST( out=="4" ); + exec_stream.in()<<"\n"; + exec_stream.out()>>out; + TEST( out=="0" ); + TEST( exec_stream.close() ); + } + + { + TEST_NAME( "echo" ); + exec_stream_t exec_stream; + exec_stream.start( program, "child echo" ); + int sizes[]={ 4096-1, 4096+1, 4096+1, 4096-1, 4096*2, 3 }; + int cnt=sizeof( sizes )/sizeof( sizes[0] ); + for( int i=0; i>err; + TEST( err==random_string( 500000 ) ); + TEST( exec_stream.close() ); + } + + { + TEST_NAME( "echo-picky" ); + exec_stream_t exec_stream; + + // disable exceptions while reading/writing + exec_stream.in().exceptions( std::ios_base::goodbit ); + exec_stream.out().exceptions( std::ios_base::goodbit ); + exec_stream.err().exceptions( std::ios_base::goodbit ); + + exec_stream.start( program, "child echo-picky" ); + std::string in_s; + std::string out_s; + + in_s=random_string( 4096 ); + exec_stream.in()<>s; + TEST( s==random_string( 2000000 ) ); + TEST( exec_stream.close() ); + } + + } + + { + TEST_NAME( "exit code" ); + exec_stream_t exec_stream; + + exec_stream.start( program, "child exit-code" ); + exec_stream.in()<<"2\n"; + TEST( exec_stream.close() ); + TEST( exec_stream.exit_code()==2 ); + + exec_stream.start( program, "child exit-code" ); + exec_stream.in()<<"42\n"; + TEST( exec_stream.close() ); + TEST( exec_stream.exit_code()==42 ); + } + + { + TEST_NAME( "args" ); + exec_stream_t exec_stream; + std::string s; + + exec_stream.start( program, "args \"one two\" three" ); + TEST( std::getline( exec_stream.out(), s ).good() ); + TEST( s=="one two" ); + TEST( std::getline( exec_stream.out(), s ).good() ); + TEST( s=="three" ); + TEST( !std::getline( exec_stream.out(), s ).good() ); + + exec_stream.start( program, "args -e \"print \\\"hello world\\\";\"" ); + TEST( std::getline( exec_stream.out(), s ).good() ); + TEST( s=="-e" ); + TEST( std::getline( exec_stream.out(), s ).good() ); + TEST( s=="print \"hello world\";" ); + TEST( !std::getline( exec_stream.out(), s ).good() ); + + char const * args[]={ "args", "one two", "three" }; + exec_stream.start( program, &args[0], &args[3] ); + TEST( std::getline( exec_stream.out(), s ).good() ); + TEST( s=="one two" ); + TEST( std::getline( exec_stream.out(), s ).good() ); + TEST( s=="three" ); + TEST( !std::getline( exec_stream.out(), s ).good() ); + + std::vector< std::string > args2; + args2.push_back( "args" ); + args2.push_back( "-e" ); + args2.push_back( "print \"hello world\";" ); + exec_stream.start( program, args2.begin(), args2.end() ); + TEST( std::getline( exec_stream.out(), s ).good() ); + TEST( s=="-e" ); + TEST( std::getline( exec_stream.out(), s ).good() ); + TEST( s=="print \"hello world\";" ); + TEST( !std::getline( exec_stream.out(), s ).good() ); + } + + { + TEST_NAME( "kinky args" ); + exec_stream_t exec_stream( program, "args", "zzzz" ); + std::string s; + TEST( std::getline( exec_stream.out(), s ).good() ); + TEST( s=="zzzz" ); + TEST( !std::getline( exec_stream.out(), s ).good() ); + } + + { + TEST_NAME( "run unexistent program" ); + std::string prog="don't know what"; + try { + exec_stream_t exec_stream; + exec_stream.start( prog, "" ); + TEST( 0=="unreachable" ); + }catch( exec_stream_t::error_t & e ) { + std::string msg=e.what(); + TEST( msg.find( prog )!=std::string::npos ); + }catch(...) { + TEST( 0=="unexpected" ); + } + } + + n_failed=test_results_t::print( std::cout ); + + }catch( std::exception const & e ) { + std::cerr<<"exception:"<0 && str_buf[buf_len-1]=='\n' ) { + --buf_len; + str_buf[buf_len]=0; + } + s+=(LPTSTR)buf; + LocalFree( buf ); + } + exec_stream_t::error_t::compose( s, code ); +} + +// pipe_t +pipe_t::pipe_t() +: m_direction( closed ), m_r( INVALID_HANDLE_VALUE ), m_w( INVALID_HANDLE_VALUE ) +{ + open(); +} + +pipe_t::~pipe_t() +{ + close(); +} + +void pipe_t::close_r() +{ + if( m_direction==both || m_direction==read ) { + if( !CloseHandle( m_r ) ) { + throw os_error_t( "pipe_t::close_r: CloseHandle failed" ); + } + m_direction= m_direction==both ? write : closed; + } +} + +void pipe_t::close_w() +{ + if( m_direction==both || m_direction==write ) { + if( !CloseHandle( m_w ) ) { + throw os_error_t( "pipe_t::close_w: CloseHandle failed" ); + } + m_direction= m_direction==both ? read : closed; + } +} + +void pipe_t::close() +{ + close_r(); + close_w(); +} + +void pipe_t::open() +{ + close(); + SECURITY_ATTRIBUTES sa; + sa.nLength=sizeof( sa ); + sa.bInheritHandle=true; + sa.lpSecurityDescriptor=0; + if( !CreatePipe( &m_r, &m_w, &sa, 0 ) ) + throw os_error_t( "pipe_t::pipe_t: CreatePipe failed" ); + m_direction=both; +} + +HANDLE pipe_t::r() const +{ + return m_r; +} + +HANDLE pipe_t::w() const +{ + return m_w; +} + +// set_stdhandle_t +set_stdhandle_t::set_stdhandle_t( DWORD kind, HANDLE handle ) +: m_kind( kind ), m_save_handle( GetStdHandle( kind ) ) +{ + if( m_save_handle==INVALID_HANDLE_VALUE ) + throw os_error_t( "set_stdhandle_t::set_stdhandle_t: GetStdHandle() failed" ); + if( !SetStdHandle( kind, handle ) ) + throw os_error_t( "set_stdhandle_t::set_stdhandle_t: SetStdHandle() failed" ); +} + +set_stdhandle_t::~set_stdhandle_t() +{ + SetStdHandle( m_kind, m_save_handle ); +} + +//wait_result_t +wait_result_t::wait_result_t() +{ + m_signaled_object=INVALID_HANDLE_VALUE; + m_timed_out=false; + m_error_code=ERROR_SUCCESS; + m_error_message=""; +} + +wait_result_t::wait_result_t( DWORD wait_result, int objects_count, HANDLE const * objects ) +{ + m_signaled_object=INVALID_HANDLE_VALUE; + m_timed_out=false; + m_error_code=ERROR_SUCCESS; + m_error_message=""; + if( wait_result>=WAIT_OBJECT_0 && wait_result=WAIT_ABANDONED_0 && wait_result( param ); + // accessing p anywhere here is safe because thread_buffer_t destructor + // ensures the thread is terminated before p get destroyed + char * read_buffer=0; + try { + read_buffer=new char[p->m_read_buffer_size]; + + while( true ) { + // see if get() wants more data, or if someone wants to stop the thread + wait_result_t wait_result=wait( p->m_stop_thread, p->m_want_data, p->m_wait_timeout ); + if( !wait_result.ok() && !wait_result.timed_out() ) { + p->note_thread_error( "thread_buffer_t::reader_thread: wait for want_data, destruction failed", wait_result.error_code(), wait_result.error_message() ); + break; + } + + if( wait_result.is_signaled( p->m_stop_thread ) ) { + // they want us to stop + break; + } + + if( wait_result.is_signaled( p->m_want_data ) ) { + // they want more data - read the file + DWORD read_size=0; + DWORD read_status=ERROR_SUCCESS; + if( !ReadFile( p->m_pipe, read_buffer, p->m_read_buffer_size, &read_size, 0 ) ) { + read_status=GetLastError(); + if( read_status!=ERROR_BROKEN_PIPE ) { + p->note_thread_error( "thread_buffer_t::reader_thread: ReadFile failed", read_status, "" ); + break; + } + } + + // read something - append to p->m_buffers + if( read_size!=0 ) { + grab_mutex_t grab_mutex( p->m_mutex, p->m_wait_timeout ); + if( !grab_mutex.ok() ) { + p->note_thread_error( "thread_buffer_t::reader_thread: wait for mutex failed", grab_mutex.error_code(), grab_mutex.error_message() ); + break; + } + + p->m_buffer_list.put( read_buffer, read_size ); + + // if buffer is too long - do not read any more until it shrinks + if( p->m_buffer_list.full( p->m_buffer_limit ) ) { + if( !p->m_want_data.reset() ) { + p->note_thread_error( "thread_buffer_t::reader_thread: unable to reset m_want_data event", GetLastError(), "" ); + break; + } + } + // tell get() we got some data + if( !p->m_got_data.set() ) { + p->note_thread_error( "thread_buffer_t::reader_thread: unable to set m_got_data event", GetLastError(), "" ); + break; + } + } + // pipe broken - quit thread, which will be seen by get() as eof. + if( read_status==ERROR_BROKEN_PIPE ) { + break; + } + } + } + }catch( ... ) { + // might only be std::bad_alloc + p->note_thread_error( "", ERROR_SUCCESS, "thread_buffer_t::reader_thread: unknown exception caught" ); + } + + delete[] read_buffer; + + // ensure that get() is not left waiting on got_data + p->m_got_data.set(); + return 0; +} + +void thread_buffer_t::put( char * const src, std::size_t & size, bool & no_more ) +{ + if( m_direction!=dir_write ) { + throw exec_stream_t::error_t( "thread_buffer_t::put: thread not started or started for reading" ); + } + // check thread status + DWORD thread_exit_code; + if( !GetExitCodeThread( m_thread, &thread_exit_code ) ) { + throw os_error_t( "thread_buffer_t::get: GetExitCodeThread failed" ); + } + + if( thread_exit_code!=STILL_ACTIVE ) { + // thread terminated - check for errors + check_error( m_message_prefix, m_error_code, m_error_message ); + // if terminated without error - signal eof, since no one will ever write our data + size=0; + no_more=true; + }else { + // wait for both m_want_data and m_mutex + wait_result_t wait_result=wait( m_want_data, m_wait_timeout ); + if( !wait_result.ok() ) { + check_error( "thread_buffer_t::put: wait for want_data failed", wait_result.error_code(), wait_result.error_message() ); + } + grab_mutex_t grab_mutex( m_mutex, m_wait_timeout ); + if( !grab_mutex.ok() ) { + check_error( "thread_buffer_t::put: wait for mutex failed", grab_mutex.error_code(), grab_mutex.error_message() ); + } + + // got them - put data + no_more=false; + if( m_translate_crlf ) { + m_buffer_list.put_translate_crlf( src, size ); + }else { + m_buffer_list.put( src, size ); + } + + // if the buffer is too long - make the next put() wait until it shrinks + if( m_buffer_list.full( m_buffer_limit ) ) { + if( !m_want_data.reset() ) { + throw os_error_t( "thread_buffer_t::put: unable to reset m_want_data event" ); + } + } + // tell the thread we got data + if( !m_buffer_list.empty() ) { + if( !m_got_data.set() ) { + throw os_error_t( "thread_buffer_t::put: unable to set m_got_data event" ); + } + } + } +} + +DWORD WINAPI thread_buffer_t::writer_thread( LPVOID param ) +{ + thread_buffer_t * p=static_cast< thread_buffer_t * >( param ); + // accessing p anywhere here is safe because thread_buffer_t destructor + // ensures the thread is terminated before p get destroyed + try { + buffer_list_t::buffer_t buffer; + buffer.data=0; + buffer.size=0; + std::size_t buffer_offset=0; + + while( true ) { + // wait for got_data or destruction, ignore timeout errors + // for destruction the timeout is normally expected, + // for got data the timeout is not normally expected but tolerable (no one wants to write) + wait_result_t wait_result=wait( p->m_got_data, p->m_stop_thread, p->m_wait_timeout ); + + if( !wait_result.ok() && !wait_result.timed_out() ) { + p->note_thread_error( "thread_buffer_t::writer_thread: wait for got_data, destruction failed", wait_result.error_code(), wait_result.error_message() ); + break; + } + + // if no data in local buffer to write - get from p->m_buffers + if( buffer.data==0 && wait_result.is_signaled( p->m_got_data ) ) { + grab_mutex_t grab_mutex( p->m_mutex, p->m_wait_timeout ); + if( !grab_mutex.ok() ) { + p->note_thread_error( "thread_buffer_t::writer_thread: wait for mutex failed", grab_mutex.error_code(), grab_mutex.error_message() ); + break; + } + if( !p->m_buffer_list.empty() ) { + // we've got buffer - detach it + buffer=p->m_buffer_list.detach(); + buffer_offset=0; + } + // if no data left in p->m_buffers - wait until it arrives + if( p->m_buffer_list.empty() ) { + if( !p->m_got_data.reset() ) { + p->note_thread_error( "thread_buffer_t::writer_thread: unable to reset m_got_data event", GetLastError(), "" ); + break; + } + } + // if buffer is not too long - tell put() it can proceed + if( !p->m_buffer_list.full( p->m_buffer_limit ) ) { + if( !p->m_want_data.set() ) { + p->note_thread_error( "thread_buffer_t::writer_thread: unable to set m_want_data event", GetLastError(), "" ); + break; + } + } + } + + // see if they want us to stop, but only when all is written + if( buffer.data==0 && wait_result.is_signaled( p->m_stop_thread ) ) { + break; + } + + if( buffer.data!=0 ) { + // we have buffer - write it + DWORD written_size; + if( !WriteFile( p->m_pipe, buffer.data+buffer_offset, buffer.size-buffer_offset, &written_size, 0 ) ) { + p->note_thread_error( "thread_buffer_t::writer_thread: WriteFile failed", GetLastError(), "" ); + break; + } + buffer_offset+=written_size; + if( buffer_offset==buffer.size ) { + delete[] buffer.data; + buffer.data=0; + } + } + + } + + // we won't be writing any more - close child's stdin + CloseHandle( p->m_pipe ); + + // buffer may be left astray - clean up + delete[] buffer.data; + + }catch( ... ) { + // unreachable code. really. + p->note_thread_error( "", ERROR_SUCCESS, "thread_buffer_t::writer_thread: unknown exception caught" ); + } + // ensure that put() is not left waiting on m_want_data + p->m_want_data.set(); + return 0; +} + +void thread_buffer_t::check_error( std::string const & message_prefix, DWORD error_code, std::string const & error_message ) +{ + if( !error_message.empty() ) { + throw exec_stream_t::error_t( message_prefix+"\n"+error_message, error_code ); + }else if( error_code!=ERROR_SUCCESS ) { + throw os_error_t( message_prefix, error_code ); + } +} + +void thread_buffer_t::note_thread_error( char const * message_prefix, DWORD error_code, char const * error_message ) +{ + m_message_prefix=message_prefix; + m_error_code=error_code; + m_error_message=error_message; +} + diff --git a/ext/libexecstream/win/exec-stream-helpers.h b/ext/libexecstream/win/exec-stream-helpers.h new file mode 100644 index 000000000..3cd175938 --- /dev/null +++ b/ext/libexecstream/win/exec-stream-helpers.h @@ -0,0 +1,183 @@ +/* +Copyright (C) 2004 Artem Khodush + +Redistribution and use in source and binary forms, with or without modification, +are permitted provided that the following conditions are met: + +1. Redistributions of source code must retain the above copyright notice, +this list of conditions and the following disclaimer. + +2. Redistributions in binary form must reproduce the above copyright notice, +this list of conditions and the following disclaimer in the documentation +and/or other materials provided with the distribution. + +3. The name of the author may not be used to endorse or promote products +derived from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED +WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES +OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. +IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; +OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, +WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR +OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, +EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ + +class os_error_t : public exec_stream_t::error_t { +public: + os_error_t( std::string const & msg ); + os_error_t( std::string const & msg, exec_stream_t::error_code_t code ); +private: + void compose( std::string const & msg, exec_stream_t::error_code_t code ); +}; + + +class pipe_t { +public: + pipe_t(); + ~pipe_t(); + HANDLE r() const; + HANDLE w() const; + void close_r(); + void close_w(); + void close(); + void open(); +private: + enum direction_t{ closed, read, write, both }; + direction_t m_direction; + HANDLE m_r; + HANDLE m_w; +}; + + +class set_stdhandle_t { +public: + set_stdhandle_t( DWORD kind, HANDLE handle ); + ~set_stdhandle_t(); +private: + DWORD m_kind; + HANDLE m_save_handle; +}; + + +class wait_result_t { +public: + wait_result_t(); + wait_result_t( DWORD wait_result, int objects_count, HANDLE const * objects ); + + bool ok(); + bool is_signaled( class event_t & event ); + bool timed_out(); + DWORD error_code(); + char const * error_message(); + +private: + + HANDLE m_signaled_object; + bool m_timed_out; + DWORD m_error_code; + char const * m_error_message; +}; + + +class event_t { +public: + event_t(); + ~event_t(); + bool set(); + bool reset(); + +private: + HANDLE m_handle; + + friend wait_result_t wait( event_t & e, DWORD timeout ); + friend wait_result_t wait( event_t & e1, event_t & e2, DWORD timeout ); + friend class wait_result_t; +}; + +wait_result_t wait( HANDLE e, DWORD timeout ); +wait_result_t wait( event_t & e, DWORD timeout ); +wait_result_t wait( event_t & e1, event_t & e2, DWORD timeout ); // waits for any one of e1, e2 + +class mutex_t { +public: + mutex_t(); + ~mutex_t(); + +private: + HANDLE m_handle; + friend class grab_mutex_t; +}; + +class grab_mutex_t { +public: + grab_mutex_t( mutex_t & mutex, DWORD timeout ); + ~grab_mutex_t(); + + bool ok(); + DWORD error_code(); + char const * error_message(); + +private: + HANDLE m_mutex; + wait_result_t m_wait_result; +}; + + +class thread_buffer_t { +public: + thread_buffer_t(); + ~thread_buffer_t(); + + // those three may be called only before the thread is started + void set_wait_timeout( DWORD milliseconds ); + void set_thread_termination_timeout( DWORD milliseconds ); + void set_buffer_limit( std::size_t limit ); + void set_read_buffer_size( std::size_t size ); + void set_binary_mode(); + void set_text_mode(); + + void start_reader_thread( HANDLE pipe ); + void start_writer_thread( HANDLE pipe); + + void get( exec_stream_t::stream_kind_t kind, char * dst, std::size_t & size, bool & no_more ); // may be called only after start_reader_thread + void put( char * const src, std::size_t & size, bool & no_more );// may be called only after start_writer_thread + + bool stop_thread(); + bool abort_thread(); + +private: + enum direction_t { dir_none, dir_read, dir_write }; + direction_t m_direction; // set by start_thread + + buffer_list_t m_buffer_list; + mutex_t m_mutex; // protecting m_buffer_list + + char const * m_message_prefix; // error occured in the thread, if any + DWORD m_error_code; // they are examined only after the thread has terminated + char const * m_error_message; // so setting them anywhere in the thread is safe + + DWORD m_wait_timeout; // parameters used in thread + std::size_t m_buffer_limit; // they are set before the thread is started, + std::size_t m_read_buffer_size; // so accessing them anywhere in the thread is safe + + HANDLE m_thread; + event_t m_want_data; // for synchronisation between get and reader_thread + event_t m_got_data; // or between put and writer_thread + event_t m_stop_thread; + HANDLE m_pipe; + DWORD m_thread_termination_timeout; + bool m_translate_crlf; + + void start_thread( HANDLE pipe, direction_t direction ); + static DWORD WINAPI reader_thread( LPVOID param ); + static DWORD WINAPI writer_thread( LPVOID param ); + + void check_error( std::string const & message_prefix, DWORD error_code, std::string const & error_message ); + void note_thread_error( char const * message_prefix, DWORD error_code, char const * error_message ); + bool check_thread_stopped(); +}; + diff --git a/ext/libexecstream/win/exec-stream-impl.cpp b/ext/libexecstream/win/exec-stream-impl.cpp new file mode 100644 index 000000000..42b8805c0 --- /dev/null +++ b/ext/libexecstream/win/exec-stream-impl.cpp @@ -0,0 +1,315 @@ +/* +Copyright (C) 2004 Artem Khodush + +Redistribution and use in source and binary forms, with or without modification, +are permitted provided that the following conditions are met: + +1. Redistributions of source code must retain the above copyright notice, +this list of conditions and the following disclaimer. + +2. Redistributions in binary form must reproduce the above copyright notice, +this list of conditions and the following disclaimer in the documentation +and/or other materials provided with the distribution. + +3. The name of the author may not be used to endorse or promote products +derived from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED +WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES +OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. +IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, +PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; +OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, +WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR +OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, +EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ + +// exec_stream_t::impl_t +struct exec_stream_t::impl_t { + impl_t(); + + HANDLE m_child_process; + + HANDLE m_in_pipe; + HANDLE m_out_pipe; + HANDLE m_err_pipe; + + thread_buffer_t m_in_thread; + thread_buffer_t m_out_thread; + thread_buffer_t m_err_thread; + + exec_stream_buffer_t m_in_buffer; + exec_stream_buffer_t m_out_buffer; + exec_stream_buffer_t m_err_buffer; + + exec_ostream_t m_in; + exec_istream_t m_out; + exec_istream_t m_err; + + DWORD m_child_timeout; + int m_exit_code; +}; + +exec_stream_t::impl_t::impl_t() +: m_in_buffer( exec_stream_t::s_in, m_in_thread ), m_out_buffer( exec_stream_t::s_out, m_out_thread ), m_err_buffer( exec_stream_t::s_err, m_err_thread ), + m_in( m_in_buffer ), m_out( m_out_buffer ), m_err( m_err_buffer ) +{ + m_out.tie( &m_in ); + m_err.tie( &m_in ); + m_child_process=0; + m_in_pipe=0; + m_out_pipe=0; + m_err_pipe=0; + m_child_timeout=500; + m_exit_code=0; +} + + +void exec_stream_t::set_buffer_limit( int stream_kind, std::size_t size ) +{ + if( stream_kind&s_in ) { + m_impl->m_in_thread.set_buffer_limit( size ); + } + if( stream_kind&s_out ) { + m_impl->m_out_thread.set_buffer_limit( size ); + } + if( stream_kind&s_err ) { + m_impl->m_err_thread.set_buffer_limit( size ); + } +} + +void exec_stream_t::set_wait_timeout( int stream_kind, exec_stream_t::timeout_t milliseconds ) +{ + if( stream_kind&s_in ) { + m_impl->m_in_thread.set_wait_timeout( milliseconds ); + } + if( stream_kind&s_out ) { + m_impl->m_out_thread.set_wait_timeout( milliseconds ); + } + if( stream_kind&s_err ) { + m_impl->m_err_thread.set_wait_timeout( milliseconds ); + } + if( stream_kind&s_child ) { + m_impl->m_child_timeout=milliseconds; + m_impl->m_in_thread.set_thread_termination_timeout( milliseconds ); + m_impl->m_out_thread.set_thread_termination_timeout( milliseconds ); + m_impl->m_err_thread.set_thread_termination_timeout( milliseconds ); + } +} + +void exec_stream_t::set_binary_mode( int stream_kind ) +{ + if( stream_kind&s_in ) { + m_impl->m_in_thread.set_binary_mode(); + } + if( stream_kind&s_out ) { + m_impl->m_out_thread.set_binary_mode(); + } + if( stream_kind&s_err ) { + m_impl->m_err_thread.set_binary_mode(); + } +} + +void exec_stream_t::set_text_mode( int stream_kind ) +{ + if( stream_kind&s_in ) { + m_impl->m_in_thread.set_text_mode(); + } + if( stream_kind&s_out ) { + m_impl->m_out_thread.set_text_mode(); + } + if( stream_kind&s_err ) { + m_impl->m_err_thread.set_text_mode(); + } +} + +void exec_stream_t::start( std::string const & program, std::string const & arguments ) +{ + if( !close() ) { + throw exec_stream_t::error_t( "exec_stream_t::start: previous child process has not yet terminated" ); + } + + pipe_t in; + pipe_t out; + pipe_t err; + set_stdhandle_t set_in( STD_INPUT_HANDLE, in.r() ); + set_stdhandle_t set_out( STD_OUTPUT_HANDLE, out.w() ); + set_stdhandle_t set_err( STD_ERROR_HANDLE, err.w() ); + HANDLE cp=GetCurrentProcess(); + if( !DuplicateHandle( cp, in.w(), cp, &m_impl->m_in_pipe, 0, FALSE, DUPLICATE_SAME_ACCESS ) ) { + throw os_error_t( "exec_stream_t::start: unable to duplicate in handle" ); + } + in.close_w(); + if( !DuplicateHandle( cp, out.r(), cp, &m_impl->m_out_pipe, 0, FALSE, DUPLICATE_SAME_ACCESS ) ) { + throw os_error_t( "exec_stream_t::start: unable to duplicate out handle" ); + } + out.close_r(); + if( !DuplicateHandle( cp, err.r(), cp, &m_impl->m_err_pipe, 0, FALSE, DUPLICATE_SAME_ACCESS ) ) { + throw os_error_t( "exec_stream_t::start: unable to duplicate err handle" ); + } + err.close_r(); + + std::string command; + command.reserve( program.size()+arguments.size()+3 ); + if( program.find_first_of( " \t" )!=std::string::npos ) { + command+='"'; + command+=program; + command+='"'; + }else + command=program; + if( arguments.size()!=0 ) { + command+=' '; + command+=arguments; + } + STARTUPINFO si; + ZeroMemory( &si, sizeof( si ) ); + si.cb=sizeof( si ); + PROCESS_INFORMATION pi; + ZeroMemory( &pi, sizeof( pi ) ); + if( !CreateProcess( 0, const_cast< char * >( command.c_str() ), 0, 0, TRUE, 0, 0, 0, &si, &pi ) ) { + throw os_error_t( "exec_stream_t::start: CreateProcess failed.\n command line was: "+command ); + } + + m_impl->m_child_process=pi.hProcess; + + m_impl->m_in_buffer.clear(); + m_impl->m_out_buffer.clear(); + m_impl->m_err_buffer.clear(); + + m_impl->m_in.clear(); + m_impl->m_out.clear(); + m_impl->m_err.clear(); + + m_impl->m_out_thread.set_read_buffer_size( STREAM_BUFFER_SIZE ); + m_impl->m_out_thread.start_reader_thread( m_impl->m_out_pipe ); + + m_impl->m_err_thread.set_read_buffer_size( STREAM_BUFFER_SIZE ); + m_impl->m_err_thread.start_reader_thread( m_impl->m_err_pipe ); + + m_impl->m_in_thread.start_writer_thread( m_impl->m_in_pipe ); +} + +void exec_stream_t::start( std::string const & program, exec_stream_t::next_arg_t & next_arg ) +{ + std::string arguments; + while( std::string const * arg=next_arg.next() ) { + if( arg->find_first_of( " \t\"" )!=std::string::npos ) { + arguments+=" \""; + std::string::size_type cur=0; + while( cursize() ) { + std::string::size_type next=arg->find( '"', cur ); + if( next==std::string::npos ) { + next=arg->size(); + arguments.append( *arg, cur, next-cur ); + cur=next; + }else { + arguments.append( *arg, cur, next-cur ); + arguments+="\\\""; + cur=next+1; + } + } + arguments+="\""; + }else { + arguments+=" "+*arg; + } + } + start( program, arguments ); +} + +bool exec_stream_t::close_in() +{ + if( m_impl->m_in_pipe!=0 ) { + m_impl->m_in.flush(); + // stop writer thread before closing the handle it writes to, + // the thread will attempt to write anything it can and close child's stdin + // before thread_termination_timeout elapses + if( m_impl->m_in_thread.stop_thread() ) { + m_impl->m_in_pipe=0; + return true; + }else { + return false; + } + }else { + return true; + } +} + +bool exec_stream_t::close() +{ + if( !close_in() ) { + // need to close child's stdin no matter what, because otherwise "usual" child will run forever + // And before closing child's stdin the writer thread should be stopped no matter what, + // because it may be blocked on Write to m_in_pipe, and in that case closing m_in_pipe may block. + if( !m_impl->m_in_thread.abort_thread() ) { + throw exec_stream_t::error_t( "exec_stream_t::close: waiting till in_thread stops exceeded timeout" ); + } + // when thread is terminated abnormally, it may left child's stdin open + // try to close it here + CloseHandle( m_impl->m_in_pipe ); + m_impl->m_in_pipe=0; + } + if( !m_impl->m_out_thread.stop_thread() ) { + if( !m_impl->m_out_thread.abort_thread() ) { + throw exec_stream_t::error_t( "exec_stream_t::close: waiting till out_thread stops exceeded timeout" ); + } + } + if( !m_impl->m_err_thread.stop_thread() ) { + if( !m_impl->m_err_thread.abort_thread() ) { + throw exec_stream_t::error_t( "exec_stream_t::close: waiting till err_thread stops exceeded timeout" ); + } + } + if( m_impl->m_out_pipe!=0 ) { + if( !CloseHandle( m_impl->m_out_pipe ) ) { + throw os_error_t( "exec_stream_t::close: unable to close out_pipe handle" ); + } + m_impl->m_out_pipe=0; + } + if( m_impl->m_err_pipe!=0 ) { + if( !CloseHandle( m_impl->m_err_pipe ) ) { + throw os_error_t( "exec_stream_t::close: unable to close err_pipe handle" ); + } + m_impl->m_err_pipe=0; + } + if( m_impl->m_child_process!=0 ) { + wait_result_t wait_result=wait( m_impl->m_child_process, m_impl->m_child_timeout ); + if( !wait_result.ok() & !wait_result.timed_out() ) { + throw os_error_t( std::string( "exec_stream_t::close: wait for child process failed. " )+wait_result.error_message() ); + } + if( wait_result.ok() ) { + DWORD exit_code; + if( !GetExitCodeProcess( m_impl->m_child_process, &exit_code ) ) { + throw os_error_t( "exec_stream_t::close: unable to get process exit code" ); + } + m_impl->m_exit_code=exit_code; + if( !CloseHandle( m_impl->m_child_process ) ) { + throw os_error_t( "exec_stream_t::close: unable to close child process handle" ); + } + m_impl->m_child_process=0; + } + } + return m_impl->m_child_process==0; +} + +void exec_stream_t::kill() +{ + if( m_impl->m_child_process!=0 ) { + if( !TerminateProcess( m_impl->m_child_process, 0 ) ) { + throw os_error_t( "exec_stream_t::kill: unable to terminate child process" ); + } + m_impl->m_exit_code=0; + if( !CloseHandle( m_impl->m_child_process ) ) { + throw os_error_t( "exec_stream_t::close: unable to close child process handle" ); + } + m_impl->m_child_process=0; + } +} + +int exec_stream_t::exit_code() +{ + if( m_impl->m_child_process!=0 ) { + throw exec_stream_t::error_t( "exec_stream_t:exit_code: child process still running" ); + } + return m_impl->m_exit_code; +} diff --git a/interfaces/python/ctml_writer.py b/interfaces/python/ctml_writer.py index a5222b2c5..c4528c806 100644 --- a/interfaces/python/ctml_writer.py +++ b/interfaces/python/ctml_writer.py @@ -1998,11 +1998,14 @@ class Lindemann: #get_atomic_wts() validate() -if __name__ == "__main__": - import sys, os - file = sys.argv[1] - base = os.path.basename(file) +def convert(filename): + import os + base = os.path.basename(filename) root, ext = os.path.splitext(base) dataset(root) - execfile(file) + execfile(filename) write() + +if __name__ == "__main__": + import sys + convert(sys.argv[1]) diff --git a/src/base/ct2ctml.cpp b/src/base/ct2ctml.cpp index 970b75f2b..212ba0a41 100644 --- a/src/base/ct2ctml.cpp +++ b/src/base/ct2ctml.cpp @@ -10,6 +10,7 @@ #include "cantera/base/ctml.h" #include "cantera/base/global.h" #include "cantera/base/stringUtils.h" +#include "../../ext/libexecstream/exec-stream.h" #include #include @@ -70,7 +71,6 @@ static string pypath() */ void ct2ctml(const char* file, const int debug) { - #ifdef HAS_NO_PYTHON /* * Section to bomb out if python is not @@ -82,107 +82,57 @@ void ct2ctml(const char* file, const int debug) ", but not available in this computational environment"); #endif - time_t aclock; - time(&aclock); - int ia = static_cast(aclock); - string path = tmpDir()+"/.cttmp"+int2str(ia)+".pyw"; - ofstream f(path.c_str()); - if (!f) { - throw CanteraError("ct2ctml","cannot open "+path+" for writing."); - } - - f << "from ctml_writer import *\n" - << "import sys, os, os.path\n" - << "file = \"" << file << "\"\n" - << "base = os.path.basename(file)\n" - << "root, ext = os.path.splitext(base)\n" - << "dataset(root)\n" - << "execfile(file)\n" - << "write()\n"; - f.close(); - string logfile = tmpDir()+"/ct2ctml.log"; -#ifdef _WIN32 - string cmd = pypath() + " " + "\"" + path + "\"" + "> " + logfile + " 2>&1"; -#else - string cmd = "sleep " + sleep() + "; " + "\"" + pypath() + "\"" + - " " + "\"" + path + "\"" + " &> " + logfile; -#endif - if (debug > 0) { - writelog("ct2ctml: executing the command " + cmd + "\n"); - writelog("ct2ctml: the Python command is: " + pypath() + "\n"); - } - - int ierr = 0; + string python_output; + int python_exit_code; try { - ierr = system(cmd.c_str()); - } catch (...) { - ierr = -10; - if (debug > 0) { - writelog("ct2ctml: command execution failed.\n"); + exec_stream_t python; + python.set_wait_timeout(exec_stream_t::s_child, 10000); + python.start(pypath(), "-i"); + stringstream output_stream; + python.in() << + "if True:\n" << // Use this so that the rest is a single block + " import sys\n" << + " sys.stderr = sys.stdout\n" << + " import ctml_writer\n" << + " ctml_writer.convert(r'" << file << "')\n"; + python.close_in(); + std::string line; + while (std::getline(python.out(), line).good()) { + output_stream << line << std::endl;; } + python.close(); + python_exit_code = python.exit_code(); + python_output = stripws(output_stream.str()); + } catch (std::exception& err) { + // Report failure to execute the python + stringstream message; + message << "Error executing python while converting input file:\n"; + message << "Python command was: '" << pypath() << "'\n"; + message << err.what() << std::endl; + throw CanteraError("ct2ctml", message.str()); } - /* - * This next section may seem a bit weird. However, it is in - * response to an issue that arises when running cantera with - * cygwin, using cygwin's python intepreter. Basically, the - * xml file is written to the local directory by the last - * system command. Then, the xml file is read immediately - * after by an ifstream() c++ command. Unfortunately, it seems - * that the directory info is not being synched fast enough so - * that the ifstream() read fails, even though the file is - * actually there. Putting in a sleep system call here fixes - * this problem. Also, having the xml file pre-existing fixes - * the problem as well. There may be more direct ways to fix - * this bug; however, I am not aware of them. - * HKM -> During the solaris port, I found the same thing. - * It probably has to do with NFS syncing problems. - * 3/3/06 - */ -#ifndef _WIN32 - string sss = sleep(); - if (debug > 0) { - writelog("sleeping for " + sss + " secs+\n"); - } - cmd = "sleep " + sss; - try { - ierr = system(cmd.c_str()); - } catch (...) { - ierr = -10; - writelog("ct2ctml: command execution failed.\n"); - } -#else - // This command works on windows machines if Windows.h and Winbase.h are included - // Sleep(5000); -#endif - - if (ierr != 0) { - // Generate an error message that includes the contents of the - // ct2ctml log file. + if (python_exit_code != 0) { + // Report a failure in the conversion process stringstream message; message << "Error converting input file \"" << file << "\" to CTML.\n"; - message << "Command was:\n\n"; - message << cmd << "\n\n"; - ifstream ferr(logfile.c_str()); - if (ferr) { - message << "-------------- start of ct2ctml.log --------------\n"; - message << ferr.rdbuf(); - message << "--------------- end of ct2ctml.log ---------------"; - } else { - message << "Additionally, the contents of ct2ctml.log" - "could not be read"; + message << "Python command was: '" << pypath() << "'\n"; + if (python_output.size() > 0) { + message << "-------------- start of converter log --------------\n"; + message << python_output << std::endl; + message << "--------------- end of converter log ---------------"; } throw CanteraError("ct2ctml", message.str()); } - // If the conversion succeeded and no debugging information is needed, - // clean up by deleting the temporary Python file and the log file. - if (debug == 0) { - remove(path.c_str()); - remove(logfile.c_str()); - } else { - writelog("ct2ctml: retaining temporary file "+path+"\n"); - writelog("ct2ctml: retaining temporary file "+logfile+"\n"); + if (python_output.size() > 0) { + // Warn if there was any output from the conversion process + stringstream message; + message << "Warning: Unexpected output from CTI converter\n"; + message << "-------------- start of converter log --------------\n"; + message << python_output << std::endl; + message << "--------------- end of converter log ---------------\n"; + writelog(message.str()); } }