pipe.h
Go to the documentation of this file.
// -*- C++ -*- (c) 2008 Petr Rockai <me@mornfall.net>
#include <wibble/sys/macros.h>
#ifdef POSIX
#include <fcntl.h>
#include <sys/select.h>
#endif
#include <unistd.h>
#include <deque>
#include <cerrno>
#include <wibble/exception.h>
#include <wibble/sys/thread.h>
#include <wibble/sys/mutex.h>
#include <wibble/sys/exec.h>
#ifndef WIBBLE_SYS_PIPE_H
#define WIBBLE_SYS_PIPE_H
namespace wibble {
namespace sys {
namespace wexcept = wibble::exception;
struct Pipe {
struct Writer : wibble::sys::Thread {
int fd;
bool close;
std::string data;
bool running;
bool closed;
wibble::sys::Mutex mutex;
Writer() : fd( -1 ), close( false ), running( false ) {}
void *main() {
do {
int wrote = 0;
{
wibble::sys::MutexLock __l( mutex );
wrote = ::write( fd, data.c_str(), data.length() );
if ( wrote > 0 )
data.erase( data.begin(), data.begin() + wrote );
}
if ( wrote == -1 ) {
if ( blocking( errno ) )
#ifdef POSIX
sched_yield();
#else
;
#endif
else
throw wexcept::System( "writing to pipe" );
}
} while ( !done() );
wibble::sys::MutexLock __l( mutex );
running = false;
if ( close )
::close( fd );
return 0;
}
bool done() {
wibble::sys::MutexLock __l( mutex );
if ( data.empty() )
running = false;
return !running;
}
void run( int _fd, std::string what ) {
wibble::sys::MutexLock __l( mutex );
if ( running )
assert_eq( _fd, fd );
fd = _fd;
assert_neq( fd, -1 );
data += what;
if ( running )
return;
running = true;
start();
}
};
typedef std::deque< char > Buffer;
Buffer buffer;
int fd;
bool _eof;
Writer writer;
Pipe( int p ) : fd( p ), _eof( false )
{
if ( p == -1 )
return;
#ifdef POSIX
if ( fcntl( fd, F_SETFL, O_NONBLOCK ) == -1 )
throw wexcept::System( "fcntl on a pipe" );
#endif
}
Pipe() : fd( -1 ), _eof( false ) {}
/* Writes data to the pipe, asynchronously. */
void write( std::string what ) {
writer.run( fd, what );
}
void close() {
wibble::sys::MutexLock __l( writer.mutex );
writer.close = true;
if ( !writer.running )
::close( fd );
}
bool valid() {
return fd != -1;
}
bool active() {
return valid() && !eof();
}
bool eof() {
return _eof;
}
static bool blocking( int err ) {
#ifdef POSIX
return err == EAGAIN || err == EWOULDBLOCK;
#else
return err == EAGAIN;
#endif
}
int readMore() {
assert( valid() );
char _buffer[1024];
int r = ::read( fd, _buffer, 1023 );
if ( r == -1 && !blocking( errno ) )
throw wexcept::System( "reading from pipe" );
else if ( r == -1 )
return 0;
if ( r == 0 )
_eof = true;
else
std::copy( _buffer, _buffer + r, std::back_inserter( buffer ) );
return r;
}
std::string nextChunk() {
std::string line( buffer.begin(), buffer.end() );
buffer.clear();
return line;
}
std::string nextLine() {
assert( valid() );
Buffer::iterator nl =
std::find( buffer.begin(), buffer.end(), '\n' );
while ( nl == buffer.end() ) {
if ( !readMore() )
return ""; // would block, so give up
nl = std::find( buffer.begin(), buffer.end(), '\n' );
}
std::string line( buffer.begin(), nl );
if ( nl != buffer.end() )
++ nl;
buffer.erase( buffer.begin(), nl );
return line;
}
/* Only returns on eof() or when data is buffered. */
void wait() {
assert( valid() );
#ifdef POSIX
fd_set fds;
FD_ZERO( &fds );
#endif
while ( buffer.empty() && !eof() ) {
if ( readMore() )
return;
if ( eof() )
return;
#ifdef POSIX
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wold-style-cast"
FD_SET( fd, &fds );
select( fd + 1, &fds, 0, 0, 0 );
#pragma GCC diagnostic pop
#else
sleep( 1 );
#endif
}
}
std::string nextLineBlocking() {
assert( valid() );
std::string l;
while ( !eof() ) {
l = nextLine();
if ( !l.empty() )
return l;
if ( eof() )
return std::string( buffer.begin(), buffer.end() );
wait();
}
return l;
}
};
struct PipeThrough
{
std::string cmd;
PipeThrough( const std::string& _cmd ) : cmd( _cmd ) {}
std::string run( std::string data ) {
int _in, _out;
#ifdef _WIN32
Exec exec(cmd);
#elif defined POSIX
ShellCommand exec(cmd);
#endif
exec.setupRedirects( &_in, &_out, 0 );
exec.fork();
Pipe in( _in ), out( _out );
in.write( data );
in.close();
std::string ret;
while ( !out.eof() ) {
out.wait();
ret += out.nextChunk();
}
return ret;
}
};
}
}
#endif
wibble::exception::System
Base class for system exceptions.
Definition: exception.h:397
wibble::sys::ChildProcess::fork
pid_t fork()
For a subprocess to run proc.
wibble::sys::ChildProcess::setupRedirects
void setupRedirects(int *stdinfd=0, int *stdoutfd=0, int *stderrfd=0)
Definition: childprocess.cpp:145
wibble::sys::Exec
Execute external commands, either forked as a ChildProcess or directly using exec().
Definition: exec.h:34
wibble::sys::MutexLockT
Acquire a mutex lock, RAII-style.
Definition: mutex.h:201
wibble::sys::Mutex
pthread mutex wrapper; WARNING: the class allows copying and assignment, but this is not always safe.
Definition: mutex.h:48
wibble::sys::ShellCommand
Execute a shell command using /bin/sh -c.
Definition: exec.h:98
wibble::sys::Thread
Encapsulates a thread.
Definition: thread.h:84
wibble::sys::Thread::start
void start()
Start the thread.
Definition: thread.cpp:70
exception.h
exec.h
macros.h
mutex.h
wibble::exception
Definition: core.h:11
wibble::sys::sleep
void sleep(int secs)
Portable version of sleep.
Definition: thread.cpp:31
wibble
Definition: amorph.h:17
wibble::iterator
Iterator< typename I::value_type > iterator(I i)
Definition: iterator.h:123
wibble::sys::PipeThrough
Definition: pipe.h:221
wibble::sys::PipeThrough::cmd
std::string cmd
Definition: pipe.h:222
wibble::sys::PipeThrough::run
std::string run(std::string data)
Definition: pipe.h:226
wibble::sys::PipeThrough::PipeThrough
PipeThrough(const std::string &_cmd)
Definition: pipe.h:224
wibble::sys::Pipe::Writer
Definition: pipe.h:29
wibble::sys::Pipe::Writer::Writer
Writer()
Definition: pipe.h:37
wibble::sys::Pipe::Writer::mutex
wibble::sys::Mutex mutex
Definition: pipe.h:35
wibble::sys::Pipe::Writer::done
bool done()
Definition: pipe.h:70
wibble::sys::Pipe::Writer::run
void run(int _fd, std::string what)
Definition: pipe.h:77
wibble::sys::Pipe::Writer::running
bool running
Definition: pipe.h:33
wibble::sys::Pipe::Writer::data
std::string data
Definition: pipe.h:32
wibble::sys::Pipe::Writer::main
void * main()
Main thread function, executed in the new thread after creation.
Definition: pipe.h:39
wibble::sys::Pipe::Writer::close
bool close
Definition: pipe.h:31
wibble::sys::Pipe::Writer::fd
int fd
Definition: pipe.h:30
wibble::sys::Pipe::Writer::closed
bool closed
Definition: pipe.h:34
wibble::sys::Pipe
Definition: pipe.h:27
wibble::sys::Pipe::write
void write(std::string what)
Definition: pipe.h:111
wibble::sys::Pipe::blocking
static bool blocking(int err)
Definition: pipe.h:134
wibble::sys::Pipe::buffer
Buffer buffer
Definition: pipe.h:94
wibble::sys::Pipe::nextLineBlocking
std::string nextLineBlocking()
Definition: pipe.h:204
wibble::sys::Pipe::_eof
bool _eof
Definition: pipe.h:96
wibble::sys::Pipe::close
void close()
Definition: pipe.h:115
wibble::sys::Pipe::eof
bool eof()
Definition: pipe.h:130
wibble::sys::Pipe::readMore
int readMore()
Definition: pipe.h:142
wibble::sys::Pipe::nextLine
std::string nextLine()
Definition: pipe.h:163
wibble::sys::Pipe::fd
int fd
Definition: pipe.h:95
wibble::sys::Pipe::valid
bool valid()
Definition: pipe.h:122
wibble::sys::Pipe::writer
Writer writer
Definition: pipe.h:97
wibble::sys::Pipe::active
bool active()
Definition: pipe.h:126
wibble::sys::Pipe::Pipe
Pipe()
Definition: pipe.h:108
wibble::sys::Pipe::Pipe
Pipe(int p)
Definition: pipe.h:99
wibble::sys::Pipe::wait
void wait()
Definition: pipe.h:182
wibble::sys::Pipe::nextChunk
std::string nextChunk()
Definition: pipe.h:157
wibble::sys::Pipe::Buffer
std::deque< char > Buffer
Definition: pipe.h:93
assert_eq
#define assert_eq(x, y)
Definition: test.h:33
assert
#define assert(x)
Definition: test.h:30
assert_neq
#define assert_neq(x, y)
Definition: test.h:36
thread.h