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

+ +

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

+ + + + + + + + + + + +

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

+ + +++ + + + + + +
+ +

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

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

+ + +++ + + + + + +
+ +

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