diff options
| -rw-r--r-- | butl/fdstream | 34 | ||||
| -rw-r--r-- | butl/fdstream.cxx | 172 | ||||
| -rw-r--r-- | tests/fdstream/driver.cxx | 93 |
3 files changed, 266 insertions, 33 deletions
diff --git a/butl/fdstream b/butl/fdstream index 49b3d4d..2d7fae5 100644 --- a/butl/fdstream +++ b/butl/fdstream @@ -34,11 +34,15 @@ namespace butl // - char only // - input or output but not both // - no support for put back + // - non-blocking file descriptor is supported only by showmanyc() function + // and only on POSIX // - throws ios::failure in case of open()/read()/write()/close() errors // - exception mask has at least badbit // - after catching an exception caused by badbit the stream is no longer // used // - not movable, though can be easily supported + // - passing to constructor -1 file descriptor is valid and results in the + // creation of an unopened object // class LIBBUTL_EXPORT fdbuf: public std::basic_streambuf<char> { @@ -46,7 +50,7 @@ namespace butl virtual ~fdbuf (); fdbuf () = default; - fdbuf (int fd) {open (fd);} + fdbuf (int fd) {if (fd != -1) open (fd);} fdbuf (const fdbuf&) = delete; fdbuf& operator= (const fdbuf&) = delete; @@ -60,6 +64,9 @@ namespace butl bool is_open () const {return fd_ != -1;} + int + fd () const {return fd_;} + public: using int_type = std::basic_streambuf<char>::int_type; using traits_type = std::basic_streambuf<char>::traits_type; @@ -96,6 +103,7 @@ namespace butl private: int fd_ = -1; char buf_[8192]; + bool non_blocking_ = false; }; // File stream mode. @@ -109,11 +117,21 @@ namespace butl // you may want not to "offend" the other end by closing your end before // reading all the data. // + // The blocking/non_blocking flags determine whether the IO operation should + // block or return control if currently there is no data to read or no room + // to write. Only the istream::readsome() function supports the semantics of + // non-blocking operations. We also only support this on POSIX (Windows does + // not provide means for the non-blocking reading from a file descriptor so + // these flags are noop there). IO stream operations other than readsome() + // are illegal for non_blocking mode and result in the badbit being set. + // enum class fdstream_mode: std::uint16_t { - text = 0x01, - binary = 0x02, - skip = 0x04 + text = 0x01, + binary = 0x02, + skip = 0x04, + blocking = 0x08, + non_blocking = 0x10 }; inline fdstream_mode operator& (fdstream_mode, fdstream_mode); @@ -149,6 +167,10 @@ namespace butl fdstream_base (int fd): buf_ (fd) {} fdstream_base (int, fdstream_mode); + public: + int + fd () const {return buf_.fd ();} + protected: fdbuf buf_; }; @@ -243,7 +265,7 @@ namespace butl // // throw failed (); // - class LIBBUTL_EXPORT ifdstream: fdstream_base, public std::istream + class LIBBUTL_EXPORT ifdstream: public fdstream_base, public std::istream { public: // Create an unopened object with iostate = badbit | failbit (we cannot @@ -323,7 +345,7 @@ namespace butl // (std::uncaught_exception() == true). This is enforced with assert() in // the ofdstream destructor. // - class LIBBUTL_EXPORT ofdstream: fdstream_base, public std::ostream + class LIBBUTL_EXPORT ofdstream: public fdstream_base, public std::ostream { public: // Create an unopened object with iostate = badbit | failbit (we cannot diff --git a/butl/fdstream.cxx b/butl/fdstream.cxx index ac5aa79..8e6f6c9 100644 --- a/butl/fdstream.cxx +++ b/butl/fdstream.cxx @@ -5,8 +5,9 @@ #include <butl/fdstream> #ifndef _WIN32 -# include <fcntl.h> // open(), O_* -# include <unistd.h> // close(), read(), write(), lseek(), ssize_t +# include <fcntl.h> // open(), O_*, fcntl() +# include <unistd.h> // close(), read(), write(), lseek(), ssize_t, + // STD*_FILENO # include <sys/uio.h> // writev(), iovec # include <sys/stat.h> // S_I* # include <sys/types.h> // off_t @@ -72,6 +73,16 @@ namespace butl open (int fd) { close (); + +#ifndef _WIN32 + int flags (fcntl (fd, F_GETFL)); + + if (flags == -1) + throw_ios_failure (errno); + + non_blocking_ = (flags & O_NONBLOCK) == O_NONBLOCK; +#endif + fd_ = fd; setg (buf_, buf_, buf_); setp (buf_, buf_ + sizeof (buf_) - 1); // Keep space for overflow's char. @@ -92,7 +103,36 @@ namespace butl streamsize fdbuf:: showmanyc () { - return is_open () ? static_cast<streamsize> (egptr () - gptr ()) : -1; + if (!is_open ()) + return -1; + + streamsize n (egptr () - gptr ()); + + if (n > 0) + return n; + +#ifndef _WIN32 + if (non_blocking_) + { + ssize_t n (read (fd_, buf_, sizeof (buf_))); + + if (n == -1) + { + if (errno == EAGAIN || errno == EINTR) + return 0; + + throw_ios_failure (errno); + } + + if (n == 0) // EOF. + return -1; + + setg (buf_, buf_, buf_ + n); + return n; + } +#endif + + return 0; } fdbuf::int_type fdbuf:: @@ -102,6 +142,14 @@ namespace butl if (is_open ()) { + // The underflow() function interface doesn't support the non-blocking + // semantics as it must return either the next character or EOF. In the + // future we may implement the blocking behavior for a non-blocking file + // descriptor. + // + if (non_blocking_) + throw_ios_failure (ENOTSUP); + if (gptr () < egptr () || load ()) r = traits_type::to_int_type (*gptr ()); } @@ -112,6 +160,10 @@ namespace butl bool fdbuf:: load () { + // Doesn't handle blocking mode and so should not be called. + // + assert (!non_blocking_); + #ifndef _WIN32 ssize_t n (read (fd_, buf_, sizeof (buf_))); #else @@ -132,6 +184,14 @@ namespace butl if (is_open () && c != traits_type::eof ()) { + // The overflow() function interface doesn't support the non-blocking + // semantics since being unable to serialize the character is supposed + // to be an error. In the future we may implement the blocking behavior + // for a non-blocking file descriptor. + // + if (non_blocking_) + throw_ios_failure (ENOTSUP); + // Store last character in the space we reserved in open(). Note // that pbump() doesn't do any checks. // @@ -148,7 +208,18 @@ namespace butl int fdbuf:: sync () { - return is_open () && save () ? 0 : -1; + if (!is_open ()) + return -1; + + // The sync() function interface doesn't support the non-blocking + // semantics since it should either completely sync the data or fail. In + // the future we may implement the blocking behavior for a non-blocking + // file descriptor. + // + if (non_blocking_) + throw_ios_failure (ENOTSUP); + + return save () ? 0 : -1; } bool fdbuf:: @@ -183,6 +254,16 @@ namespace butl streamsize fdbuf:: xsputn (const char_type* s, streamsize sn) { + // The xsputn() function interface doesn't support the non-blocking + // semantics since the only excuse not to fully serialize the data is + // encountering EOF (the default behaviour is defined as a sequence of + // sputc() calls which stops when either sn characters are written or a + // call would have returned EOF). In the future we may implement the + // blocking behavior for a non-blocking file descriptor. + // + if (non_blocking_) + throw_ios_failure (ENOTSUP); + // To avoid futher 'signed/unsigned comparison' compiler warnings. // size_t n (static_cast<size_t> (sn)); @@ -298,19 +379,31 @@ namespace butl #endif } + inline static bool + flag (fdstream_mode m, fdstream_mode flag) + { + return (m & flag) == flag; + } + + inline static int + mode (int fd, fdstream_mode m) + { + if (fd != -1 && + (flag (m, fdstream_mode::text) || + flag (m, fdstream_mode::binary) || + flag (m, fdstream_mode::blocking) || + flag (m, fdstream_mode::non_blocking))) + fdmode (fd, m); + + return fd; + } + // fdstream_base // fdstream_base:: fdstream_base (int fd, fdstream_mode m) - : fdstream_base (fd) // Delegate. + : fdstream_base (mode (fd, m)) // Delegate. { - // Note that here we rely on fdstream_base() (and fdbuf() which it calls) - // to not read from the file. - // - if (fd != -1 && - ((m & fdstream_mode::text) == fdstream_mode::text || - (m & fdstream_mode::binary) == fdstream_mode::binary)) - fdmode (fd, m); } static fdopen_mode @@ -631,27 +724,54 @@ namespace butl } fdstream_mode - fdmode (int, fdstream_mode) + fdmode (int fd, fdstream_mode m) { - return fdstream_mode::binary; + int flags (fcntl (fd, F_GETFL)); + + if (flags == -1) + throw_ios_failure (errno); + + if (flag (m, fdstream_mode::blocking) || + flag (m, fdstream_mode::non_blocking)) + { + m &= fdstream_mode::blocking | fdstream_mode::non_blocking; + + // Should be exactly one blocking mode flag specified. + // + if (m != fdstream_mode::blocking && m != fdstream_mode::non_blocking) + throw invalid_argument ("invalid blocking mode"); + + int new_flags ( + m == fdstream_mode::non_blocking + ? flags | O_NONBLOCK + : flags & ~O_NONBLOCK); + + if (fcntl (fd, F_SETFL, new_flags) == -1) + throw_ios_failure (errno); + } + + return fdstream_mode::binary | + ((flags & O_NONBLOCK) == O_NONBLOCK + ? fdstream_mode::non_blocking + : fdstream_mode::blocking); } fdstream_mode - stdin_fdmode (fdstream_mode) + stdin_fdmode (fdstream_mode m) { - return fdstream_mode::binary; + return fdmode (STDIN_FILENO, m); } fdstream_mode - stdout_fdmode (fdstream_mode) + stdout_fdmode (fdstream_mode m) { - return fdstream_mode::binary; + return fdmode (STDOUT_FILENO, m); } fdstream_mode - stderr_fdmode (fdstream_mode) + stderr_fdmode (fdstream_mode m) { - return fdstream_mode::binary; + return fdmode (STDERR_FILENO, m); } #else @@ -704,6 +824,11 @@ namespace butl // Should be exactly one translation flag specified. // + // It would have been natural not to change translation mode if none of + // text or binary flags are passed. Unfortunatelly there is no (easy) way + // to obtain the current mode for the file descriptor without setting a + // new one. This is why not specifying one of the modes is an error. + // if (m != fdstream_mode::binary && m != fdstream_mode::text) throw invalid_argument ("invalid translation mode"); @@ -711,9 +836,10 @@ namespace butl if (r == -1) throw_ios_failure (errno); - return (r & _O_BINARY) == _O_BINARY - ? fdstream_mode::binary - : fdstream_mode::text; + return fdstream_mode::blocking | + ((r & _O_BINARY) == _O_BINARY + ? fdstream_mode::binary + : fdstream_mode::text); } fdstream_mode diff --git a/tests/fdstream/driver.cxx b/tests/fdstream/driver.cxx index 2e3bac5..f77e2d5 100644 --- a/tests/fdstream/driver.cxx +++ b/tests/fdstream/driver.cxx @@ -2,6 +2,11 @@ // copyright : Copyright (c) 2014-2016 Code Synthesis Ltd // license : MIT; see accompanying LICENSE file +#ifndef _WIN32 +# include <chrono> +# include <thread> // this_thread::sleep_for() +#endif + #include <ios> #include <string> #include <vector> @@ -13,6 +18,7 @@ #include <exception> #include <butl/path> +#include <butl/process> #include <butl/fdstream> #include <butl/timestamp> #include <butl/filesystem> @@ -90,13 +96,55 @@ read_time (const path& p, const T& s, size_t n) int main (int argc, const char* argv[]) { - if (!(argc == 1 || (argc == 2 && argv[1] == string ("-v")))) + bool v (false); + bool child (false); + + int i (1); + for (; i != argc; ++i) { - cerr << "usage: " << argv[0] << " [-v]" << endl; - return 1; + string a (argv[i]); + if (a == "-c") + child = true; + else if (a == "-v") + v = true; + else + { + cerr << "usage: " << argv[0] << " [-v] [-c]" << endl; + return 1; + } + } + + // To test non-blocking reading from ifdstream the test program launches + // itself as a child process with -c option and roundtrips a string through + // it. The child must write the string in chunks with some delays to make + // sure the parent reads in chunks as well. + // + if (child) + { + cin.exceptions (ios_base::badbit); + cout.exceptions (ios_base::failbit | ios_base::badbit | ios_base::eofbit); + + string s; + getline (cin, s, '\0'); + + size_t n (10); + for (size_t i (0); i < s.size (); i += n) + { + cout.write (s.c_str () + i, min (n, s.size () - i)); + cout.flush (); + + // @@ MINGW GCC 4.9 doesn't implement this_thread. If ifdstream + // non-blocking read will ever be implemented use Win32 Sleep() + // instead. + // +#ifndef _WIN32 + this_thread::sleep_for (chrono::milliseconds (50)); +#endif + } + + return 0; } - bool v (argc == 2); dir_path td (dir_path::temp_directory () / dir_path ("butl-fdstream")); // Recreate the temporary directory (that possibly exists from the previous @@ -351,6 +399,43 @@ main (int argc, const char* argv[]) { } + // Test non-blocking reading. + // + try + { + const char* args[] = {argv[0], "-c", nullptr}; + process pr (args, -1, -1); + + ofdstream os (pr.out_fd); + ifdstream is (pr.in_ofd, fdstream_mode::non_blocking); + + const string s ( + "0123456789\nABCDEFGHIJKLMNOPQRSTUVWXYZ\nabcdefghijklmnopqrstuvwxyz"); + + os << s; + os.close (); + + string r; + char buf[3]; + while (!is.eof ()) + { + streamsize n (is.readsome (buf, sizeof (buf))); + r.append (buf, n); + } + + is.close (); + + assert (r == s); + } + catch (const ios::failure&) + { + assert (false); + } + catch (const process_error&) + { + assert (false); + } + #else // Check translation modes. |