ncbi_pipe.cpp

ncbi源码

        }

        // Retry if either blocked or interrupted
        for (;;) {
            // Try to write
            ssize_t bytes_written = write(m_ChildStdIn, buf, count);
            if (bytes_written >= 0) {
                if ( n_written ) {
                    *n_written = (size_t) bytes_written;
                }
                status = eIO_Success;
                break;
            }

            // Peer has closed its end
            if (errno == EPIPE) {
                return eIO_Closed;
            }
            // Blocked -- wait for write readiness;  exit if timeout/error
            if (errno == EAGAIN  ||  errno == EWOULDBLOCK) {
                status = x_Wait(m_ChildStdIn, eIO_Write, timeout);
                if (status != eIO_Success) {
                    break;
                }
                continue;
            }
            // Interrupted write -- restart
            if (errno != EINTR) {
                throw string("Failed to write data into pipe");
            }
        }
    }
    catch (string& what) {
        ERR_POST(s_FormatErrorMessage("Write", what));
    }
    return status;
}


void CPipeHandle::x_Clear(void)
{
    if (m_ChildStdIn  != -1) {
        close(m_ChildStdIn);
        m_ChildStdIn   = -1;
    }
    if (m_ChildStdOut != -1) {
        close(m_ChildStdOut);
        m_ChildStdOut  = -1;
    }
    if (m_ChildStdErr != -1) {
        close(m_ChildStdErr);
        m_ChildStdErr  = -1;
    }
    m_Pid = -1;
}


int CPipeHandle::x_GetHandle(CPipe::EChildIOHandle from_handle) const
{
    switch (from_handle) {
    case CPipe::eStdIn:
        return m_ChildStdIn;
    case CPipe::eStdOut:
        return m_ChildStdOut;
    case CPipe::eStdErr:
        return m_ChildStdErr;
    default:
        break;
    }
    return -1;
}


bool CPipeHandle::x_SetNonBlockingMode(int fd, bool nonblock) const
{
    return fcntl(fd, F_SETFL,
                 nonblock ?
                 fcntl(fd, F_GETFL, 0) | O_NONBLOCK :
                 fcntl(fd, F_GETFL, 0) & (int) ~O_NONBLOCK) != -1;
}


EIO_Status CPipeHandle::x_Wait(int fd, EIO_Event direction,
                               const STimeout* timeout) const
{
    // Wait for the file descriptor to become ready only
    // if timeout is set or infinite
    if (timeout  &&  !timeout->sec  &&  !timeout->usec) {
        return eIO_Timeout;
    }
    for (;;) { // Auto-resume if interrupted by a signal
        struct timeval* tmp;
        struct timeval  tm;

        if ( timeout ) {
            // NB: Timeout has been normalized already
            tm.tv_sec  = timeout->sec;
            tm.tv_usec = timeout->usec;
            tmp = &tm;
        } else
            tmp = 0;
        fd_set rfds;
        fd_set wfds;
        fd_set efds;
        switch (direction) {
        case eIO_Read:
            FD_ZERO(&rfds);
            FD_SET(fd, &rfds);
            break;
        case eIO_Write:
            FD_ZERO(&wfds);
            FD_SET(fd, &wfds);
            break;
        default:
            assert(0);
            return eIO_Unknown;
        }
        FD_ZERO(&efds);
        FD_SET(fd, &efds);

        int n = select(fd + 1,
                       direction == eIO_Read  ? &rfds : 0,
                       direction == eIO_Write ? &wfds : 0,
                       &efds, tmp);
        if (n == 0) {
            return eIO_Timeout;
        } else if (n > 0) {
            switch (direction) {
            case eIO_Read:
                if ( !FD_ISSET(fd, &rfds) ) {
                    assert( FD_ISSET(fd, &efds) );
                    return eIO_Closed;
                }
                break;
            case eIO_Write:
                if ( !FD_ISSET(fd, &wfds) ) {
                    assert( FD_ISSET(fd, &efds) );
                    return eIO_Closed;
                }
                break;
            default:
                assert(0);
                return eIO_Unknown;
            }
            break;
        } else if (errno != EINTR) {
            throw string("Failed select() on pipe");
        }
    }
    return eIO_Success;
}


#endif  /* NCBI_OS_UNIX | NCBI_OS_MSWIN */



//////////////////////////////////////////////////////////////////////////////
//
// CPipe
//

CPipe::CPipe(void)
    : m_PipeHandle(0), m_ReadHandle(eStdOut),
      m_ReadStatus(eIO_Closed), m_WriteStatus(eIO_Closed),
      m_ReadTimeout(0), m_WriteTimeout(0), m_CloseTimeout(0)
 
{
    // Create new OS-specific pipe handle
    m_PipeHandle = new CPipeHandle();
    if ( !m_PipeHandle ) {
        NCBI_THROW(CPipeException, eInit,
                   "Cannot create OS-specific pipe handle");
    }
}


CPipe::CPipe(const string& cmd, const vector<string>& args,
             TCreateFlags create_flags)
    : m_PipeHandle(0), m_ReadHandle(eStdOut),
      m_ReadStatus(eIO_Closed), m_WriteStatus(eIO_Closed),
      m_ReadTimeout(0), m_WriteTimeout(0), m_CloseTimeout(0)
{
    // Create new OS-specific pipe handle
    m_PipeHandle = new CPipeHandle();
    if ( !m_PipeHandle ) {
        NCBI_THROW(CPipeException, eInit,
                   "Cannot create OS-specific pipe handle");
    }
    EIO_Status status = Open(cmd, args, create_flags);
    if (status != eIO_Success) {
        NCBI_THROW(CPipeException, eOpen, "CPipe::Open() failed");
    }
}


CPipe::~CPipe(void)
{
    Close();
    if ( m_PipeHandle ) {
        delete m_PipeHandle;
    }
}


EIO_Status CPipe::Open(const string& cmd, const vector<string>& args,
                       TCreateFlags create_flags)
{
    if ( !m_PipeHandle ) {
        return eIO_Unknown;
    }
    assert(CPipe::fStdIn_Close);


    EIO_Status status = m_PipeHandle->Open(cmd, args, create_flags);
    if (status == eIO_Success) {
        m_ReadStatus  = eIO_Success;
        m_WriteStatus = eIO_Success;
    }
    return status;
}


EIO_Status CPipe::Close(int* exitcode)
{
    if ( !m_PipeHandle ) {
        return eIO_Unknown;
    }
    m_ReadStatus  = eIO_Closed;
    m_WriteStatus = eIO_Closed;

    return m_PipeHandle->Close(exitcode, m_CloseTimeout);
}


EIO_Status CPipe::CloseHandle(EChildIOHandle handle)
{
    if ( !m_PipeHandle ) {
        return eIO_Unknown;
    }
    return m_PipeHandle->CloseHandle(handle);
}


EIO_Status CPipe::Read(void* buf, size_t count, size_t* read,
                       EChildIOHandle from_handle)
{
    if ( read ) {
        *read = 0;
    }
    if (from_handle == eDefault)
        from_handle = m_ReadHandle;
    if (from_handle == eStdIn) {
        return eIO_InvalidArg;
    }
    if (count  &&  !buf) {
        return eIO_InvalidArg;
    }
    if ( !m_PipeHandle ) {
        return eIO_Unknown;
    }
    m_ReadStatus = m_PipeHandle->Read(buf, count, read, from_handle,
                                      m_ReadTimeout);
    return m_ReadStatus;
}


EIO_Status CPipe::SetReadHandle(EChildIOHandle from_handle)
{
    if (from_handle == eStdIn) {
        return eIO_Unknown;
    }
    if (from_handle != eDefault) {
        m_ReadHandle = from_handle;
    }
    return eIO_Success;
}


EIO_Status CPipe::Write(const void* buf, size_t count, size_t* written)
{
    if ( written ) {
        *written = 0;
    }
    if (count  &&  !buf) {
        return eIO_InvalidArg;
    }
    if ( !m_PipeHandle ) {
        return eIO_Unknown;
    }
    m_WriteStatus = m_PipeHandle->Write(buf, count, written,
                                        m_WriteTimeout);
    return m_WriteStatus;
}


EIO_Status CPipe::Status(EIO_Event direction) const
{
    switch ( direction ) {
    case eIO_Read:
        return m_ReadStatus;
    case eIO_Write:
        return m_WriteStatus;
    default:
        break;
    }
    return eIO_InvalidArg;
}


EIO_Status CPipe::SetTimeout(EIO_Event event, const STimeout* timeout)
{
    if (timeout == kDefaultTimeout) {
        return eIO_Success;
    }
    switch ( event ) {
    case eIO_Close:
        m_CloseTimeout = s_SetTimeout(timeout, &m_CloseTimeoutValue);
        break;
    case eIO_Read:
        m_ReadTimeout  = s_SetTimeout(timeout, &m_ReadTimeoutValue);
        break;
    case eIO_Write:
        m_WriteTimeout = s_SetTimeout(timeout, &m_WriteTimeoutValue);
        break;
    case eIO_ReadWrite:
        m_ReadTimeout  = s_SetTimeout(timeout, &m_ReadTimeoutValue);
        m_WriteTimeout = s_SetTimeout(timeout, &m_WriteTimeoutValue);
        break;
    default:
        return eIO_InvalidArg;
    }
    return eIO_Success;
}


const STimeout* CPipe::GetTimeout(EIO_Event event) const
{
    switch ( event ) {
    case eIO_Close:
        return m_CloseTimeout;
    case eIO_Read:
        return m_ReadTimeout;
    case eIO_Write:
        return m_WriteTimeout;
    default:
        ;
    }
    return kDefaultTimeout;
}


TProcessHandle CPipe::GetProcessHandle(void) const
{
    return m_PipeHandle ? m_PipeHandle->GetProcessHandle() : 0;
}


END_NCBI_SCOPE


/*
 * ===========================================================================
 * $Log: ncbi_pipe.cpp,v $
 * Revision 1000.4  2004/06/01 18:45:13  gouriano
 * PRODUCTION: UPGRADED [GCC34_MSVC7] Dev-tree R1.37
 *
 * Revision 1.37  2004/05/17 20:58:13  gorelenk
 * Added include of PCH ncbi_pch.hpp
 *
 * Revision 1.36  2003/12/04 16:28:52  ivanov
 * CPipeHandle::Close(): added handling new create flags f*OnClose.
 * Added GetProcessHandle(). Throw/catch string exceptions instead char*.
 *
 * Revision 1.35  2003/11/14 13:06:01  ucko
 * +<stdio.h> (still needed by source...)
 *
 * Revision 1.34  2003/11/04 03:11:51  lavr
 * x_GetHandle(): default switch case added [for GCC2.95.2 to be happy]
 *
 * Revision 1.33  2003/10/24 16:52:38  lavr
 * Check RW bits before E bits in select()
 *
 * Revision 1.32  2003/09/25 04:40:42  lavr
 * Fixed uninitted member in ctor; few more minor changes
 *
 * Revision 1.31  2003/09/23 21:08:37  lavr
 * PipeStreambuf and special stream removed: now all in ncbi_conn_stream.cpp
 *
 * Revision 1.30  2003/09/16 13:42:36  ivanov
 * Added deleting OS-specific pipe handle in the destructor
 *
 * Revision 1.29  2003/09/09 19:30:33  ivanov
 * Fixed I/O timeout handling in the UNIX CPipeHandle. Cleanup code.
 * Comments and messages changes.
 *
 * Revision 1.28  2003/09/04 13:55:47  kans
 * added include sys/time.h to fix Mac compiler error
 *
 * Revision 1.27  2003/09/03 21:37:05  ivanov
 * Removed Linux ESPIPE workaround
 *
 * Revision 1.26  2003/09/03 20:53:02  ivanov
 * UNIX CPipeHandle::Read(): Added workaround for Linux -- read() returns
 * ESPIPE if a second pipe end is closed.
 * UNIX CPipeHandle::Close(): Implemented close timeout handling.
 *
 * Revision 1.25  2003/09/03 14:35:59  ivanov
 * Fixed previous accidentally commited log message
 *
 * Revision 1.24  2003/09/03 14:29:58  ivanov
 * Set r/w status to eIO_Success in the CPipe::Open()
 *
 * Revision 1.23  2003/09/02 20:32:34  ivanov
 * Moved ncbipipe to CONNECT library from CORELIB.
 * Rewritten CPipe class using I/O timeouts.
 *
 * Revision 1.22  2003/06/18 21:21:48  vakatov
 * Formal code formatting
 *
 * Revision 1.21  2003/06/18 18:59:33  rsmith
 * put in 0,1,2 for fileno(stdin/out/err) for library that does not have
 * fileno().
 *
 * Revision 1.20  2003/05/19 21:21:38  vakatov
 * Get rid of unnecessary unreached statement
 *
 * Revision 1.19  2003/05/08 20:13:35  ivanov
 * Cleanup #include <util/...>
 *
 * Revision 1.18  2003/04/23 20:49:21  ivanov
 * Entirely rewritten the Unix version of the CPipeHandle.
 * Added CPipe/CPipeHandle::CloseHandle().
 * Added flushing a standart output streams before it redirecting.
 *
 * Revision 1.17  2003/04/04 16:02:38  lavr
 * Lines wrapped at 79th column; some minor reformatting
 *
 * Revision 1.16  2003/04/03 14:15:48  rsmith
 * combine pp symbols NCBI_COMPILER_METROWERKS & _MSL_USING_MW_C_HEADERS
 * into NCBI_COMPILER_MW_MSL
 *
 * Revision 1.15  2003/04/02 16:22:34  rsmith
 * clean up metrowerks ifdefs.
 *
 * Revision 1.14  2003/04/02 13:29:53  rsmith
 * include ncbi_mslextras.h when compiling with MSL libs in Codewarrior.
 *
 * Revision 1.13  2003/03/10 18:57:08  kuznets
 * iterate->ITERATE
 *
 * Revision 1.12  2003/03/07 16:19:39  ivanov
 * MSWin CPipeHandle::Close() -- Handling GetExitCodeProcess() returns
 * value 259 (STILL_ACTIVE)
 *
 * Revision 1.11  2003/03/06 21:12:10  ivanov
 * Formal comments rearrangement
 *
 * Revision 1.10  2003/03/03 14:47:20  dicuccio
 * Remplemented CPipe using private platform specific classes.  Remplemented
 * Win32 pipes using CreatePipe() / CreateProcess() - enabled CPipe in windows
 * subsystem
 *
 * Revision 1.9  2002/09/05 18:38:07  vakatov
 * Formal code rearrangement to get rid of comp.warnings;  some nice-ification
 *
 * Revision 1.8  2002/07/15 18:17:24  gouriano
 * renamed CNcbiException and its descendents
 *
 * Revision 1.7  2002/07/11 14:18:27  gouriano
 * exceptions replaced by CNcbiException-type ones
 *
 * Revision 1.6  2002/07/02 16:22:25  ivanov
 * Added closing file descriptors of the local end of the pipe in Close()
 *
 * Revision 1.5  2002/06/12 14:20:44  ivanov
 * Fixed some bugs in CPipeStreambuf class.
 *
 * Revision 1.4  2002/06/12 13:48:56  ivanov
 * Fixed contradiction in types of constructor CPipeStreambuf and its
 * realization
 *
 * Revision 1.3  2002/06/11 19:25:35  ivanov
 * Added class CPipeIOStream, CPipeStreambuf
 *
 * Revision 1.2  2002/06/10 18:35:27  ivanov
 * Changed argument's type of a running child program from char*[]
 * to vector<string>
 *
 * Revision 1.1  2002/06/10 16:59:39  ivanov
 * Initial revision
 *
 * ===========================================================================
 */