qunixsocket.cpp

奇趣公司比较新的qt/emd版本

  \sa QUnixSocket::rightsBufferSize()
  */
void QUnixSocket::setRightsBufferSize(qint64 size)
{
    Q_ASSERT(size >= 0);

    if((size == d->ancillaryBufferCount || d->dataBufferLength) &&
            d->ancillaryBuffer)
        return;

    qint64 byteSize = CMSG_SPACE(sizeof(::ucred)) +
                      CMSG_SPACE(size * sizeof(int));

    if(d->ancillaryBuffer) delete [] d->ancillaryBuffer;
    d->ancillaryBuffer = new char[byteSize];
    d->ancillaryBufferCount = size;
}

/*!
  \overload

  Writes \a socketdata to the socket.  In addition to failing if the socket
  is not in the Connected state, writing will fail if \a socketdata is
  \l {QUnixSocketMessage::isValid()}{invalid. }

  Writes through the QUnixSocket class are asynchronous.  Rather than being
  written immediately, data is enqueued and written once the application
  reenters the Qt event loop and the socket becomes available for writing.
  Thus, this method will only fail if the socket is not in the Connected state
  - it is illegal to attempt a write on a Unconnected or Closing socket.

  Applications can monitor the progress of data writes through the
  \l QUnixSocket::bytesWritten() signal and \l QUnixSocket::bytesToWrite()
  method.

  \sa QUnixSocketMessage
  */
qint64 QUnixSocket::write(const QUnixSocketMessage & socketdata)
{
    if(ConnectedState != state() || !socketdata.isValid()) return -1;
    if(socketdata.d->size() == 0) return 0;

    d->writeQueue.enqueue(socketdata);
    d->writeQueueBytes += socketdata.d->size();
    d->writeNotifier->setEnabled(true);

    return socketdata.d->size();
}

/*!
  Return the next available message, or an empty message if none is available.

  To avoid retrieving empty messages, applications should connect to the
  \l QUnixSocket::readyRead() signal to be notified when new messages are
  available or periodically poll the \l QUnixSocket::bytesAvailable() method.

  \sa QUnixSocket::readyRead() QUnixSocket::bytesAvailable()
  */
QUnixSocketMessage QUnixSocket::read()
{
    QUnixSocketMessage data;
    if(!d->dataBufferLength)
        return data;

    data.d->state = QUnixSocketMessagePrivate::Credential;

    // Bytes are easy
    data.setBytes(QByteArray(d->dataBuffer, d->dataBufferLength));

    // Extract ancillary data
    QList<QUnixSocketRights> a;

    ::cmsghdr * h = (::cmsghdr *)CMSG_FIRSTHDR(&(d->message));
    while(h) {

        if(SCM_CREDENTIALS == h->cmsg_type) {
            ::ucred * cred = (::ucred *)CMSG_DATA(h);
#ifdef QUNIXSOCKET_DEBUG
            qDebug( "Credentials recd: pid %lu - gid %lu - uid %lu",
                    cred->pid, cred->gid, cred->uid );
#endif
            data.d->pid = cred->pid;
            data.d->gid = cred->gid;
            data.d->uid = cred->uid;

        } else if(SCM_RIGHTS == h->cmsg_type) {

            int * fds = (int *)CMSG_DATA(h);
            int numFds = (h->cmsg_len - CMSG_LEN(0)) / sizeof(int);

            for(int ii = 0; ii < numFds; ++ii) {
                QUnixSocketRights qusr(fds[ii], 0);
                a.append(qusr);
            }

        } else {

#ifdef QUNIXSOCKET_DEBUG
            qFatal("QUnixSocket: Unknown ancillary data type (%d) received.",
                   h->cmsg_type);
#endif

        }

        h = (::cmsghdr *)CMSG_NXTHDR(&(d->message), h);
    }

    if(d->message.msg_flags & MSG_CTRUNC) {
        data.d->state = (QUnixSocketMessagePrivate::AncillaryDataState)(QUnixSocketMessagePrivate::Truncated |
                               QUnixSocketMessagePrivate::Credential );
    }

    if(!a.isEmpty())
        data.d->rights = a;

    d->dataBufferLength = 0;
    d->messageValid = false;
    d->readNotifier->setEnabled(true);

    return data;
}

/*! \internal */
bool QUnixSocket::isSequential() const
{
    return true;
}

/*! \internal */
bool QUnixSocket::waitForReadyRead(int msecs)
{
    if(UnconnectedState == d->state)
        return false;

    if(d->messageValid) {
        return true;
    }

    Q_ASSERT(-1 != d->fd);

    int     timeout = msecs;
    struct  timeval tv;
    struct  timeval *ptrTv = 0;
    QTime   stopWatch;

    stopWatch.start();

    do
    {
        fd_set readset;

        FD_ZERO(&readset);
        FD_SET(d->fd, &readset);

        if(-1 != msecs) {
            tv.tv_sec = timeout / 1000;
            tv.tv_usec = (timeout % 1000) * 1000;
            ptrTv = &tv;
        }

        int rv = ::select(d->fd + 1, &readset, 0, 0, ptrTv);
        switch(rv) {
            case 0:
                // timeout
                return false;
            case 1:
                // ok
                d->readActivated();
                return true;
            default:
                if (errno != EINTR)
                    abort();    // error
                break;
        }

        timeout = msecs - stopWatch.elapsed();
    }
    while (timeout > 0);

    return false;
}

bool QUnixSocket::waitForBytesWritten(int msecs)
{
    if(UnconnectedState == d->state)
        return false;

    Q_ASSERT(-1 != d->fd);

    if ( d->writeQueue.isEmpty() )
        return true;

    QTime stopWatch;
    stopWatch.start();

    while ( true )
    {
        fd_set fdwrite;
        FD_ZERO(&fdwrite);
        FD_SET(d->fd, &fdwrite);
        int timeout = msecs < 0 ? 0 : msecs - stopWatch.elapsed();
        struct timeval tv;
        struct timeval *ptrTv = 0;
        if ( -1 != msecs )
        {
            tv.tv_sec = timeout / 1000;
            tv.tv_usec = (timeout % 1000) * 1000;
            ptrTv = &tv;
        }

        int rv = ::select(d->fd + 1, 0, &fdwrite, 0, ptrTv);
        switch ( rv )
        {
            case 0:
                // timeout
                return false;
            case 1:
                // ok
                d->writeActivated();
                return true;
            default:
                // error - or an uncaught signal!!!!!!!!!
                if ( rv == EINTR )
                    continue;
                abort();
                return false;
        }
    }
    return false; // fix warnings
}

/*! \internal */
bool QUnixSocket::canReadLine() const
{
    for(unsigned int ii = 0; ii < d->dataBufferLength; ++ii)
        if(d->dataBuffer[ii] == '\n') return true;
    return false;
}

/*! \internal */
qint64 QUnixSocket::readData(char * data, qint64 maxSize)
{
    Q_ASSERT(data);
    if(0 >= maxSize) return 0;
    if(!d->dataBufferLength) return 0;

    // Read data
    unsigned int size = d->dataBufferLength>maxSize?maxSize:d->dataBufferLength;
    memcpy(data, d->dataBuffer, size);
    if(size == d->dataBufferLength) {
        d->dataBufferLength = 0;
    } else {
        memmove(d->dataBuffer, d->dataBuffer + size, d->dataBufferLength - size);
        d->dataBufferLength -= size;
    }


    // Flush ancillary
    d->flushAncillary();

    if(0 == d->dataBufferLength)
        d->readNotifier->setEnabled(true);

    return size;
}

/*! \internal */
qint64 QUnixSocket::writeData (const char * data, qint64 maxSize)
{
    return write(QUnixSocketMessage(QByteArray(data, maxSize)));
}

void QUnixSocketPrivate::writeActivated()
{
    writeNotifier->setEnabled(false);

    QUnixSocketMessage & m = writeQueue.head();
    const QList<QUnixSocketRights> & a = m.rights();

    //
    // Construct the message
    //
    ::iovec vec;
    if ( !m.d->vec ) // message does not already have an iovec
    {
        vec.iov_base = (void *)m.bytes().constData();
        vec.iov_len = m.bytes().size();
    }

    // Allocate the control buffer
    ::msghdr sendmessage;
    ::bzero(&sendmessage, sizeof(::msghdr));
    if ( m.d->vec )
    {
        sendmessage.msg_iov = m.d->vec;
        sendmessage.msg_iovlen = m.d->iovecLen;
    }
    else
    {
        sendmessage.msg_iov = &vec;
        sendmessage.msg_iovlen = 1;
    }
    unsigned int required = CMSG_SPACE(sizeof(::ucred)) +
                            a.size() * CMSG_SPACE(sizeof(int));
    sendmessage.msg_control = new char[required];
    ::bzero(sendmessage.msg_control, required);
    sendmessage.msg_controllen = required;

    // Create ancillary buffer
    ::cmsghdr * h = CMSG_FIRSTHDR(&sendmessage);

    if(m.d->state & QUnixSocketMessagePrivate::Credential) {
        h->cmsg_len = CMSG_LEN(sizeof(::ucred));
        h->cmsg_level = SOL_SOCKET;
        h->cmsg_type = SCM_CREDENTIALS;
        ((::ucred *)CMSG_DATA(h))->pid = m.d->pid;
        ((::ucred *)CMSG_DATA(h))->gid = m.d->gid;
        ((::ucred *)CMSG_DATA(h))->uid = m.d->uid;
        h = CMSG_NXTHDR(&sendmessage, h);
    } else {
        sendmessage.msg_controllen -= CMSG_SPACE(sizeof(::ucred));
    }

    for(int ii = 0; ii < a.count(); ++ii) {
        const QUnixSocketRights & r = a.at(ii);

        if(r.isValid()) {
            h->cmsg_len = CMSG_LEN(sizeof(int));
            h->cmsg_level = SOL_SOCKET;
            h->cmsg_type = SCM_RIGHTS;
            *((int *)CMSG_DATA(h)) = r.peekFd();
            h = CMSG_NXTHDR(&sendmessage, h);
        } else {
            sendmessage.msg_controllen -= CMSG_SPACE(sizeof(int));
        }
    }

#ifdef QUNIXSOCKET_DEBUG
    qDebug() << "QUnixSocket: Transmitting message (length" << m.d->size() << ")";
#endif
    ::ssize_t s = ::sendmsg(fd, &sendmessage, MSG_DONTWAIT | MSG_NOSIGNAL);
#ifdef QUNIXSOCKET_DEBUG
    qDebug() << "QUnixSocket: Transmitted message (" << s << ")";
#endif

    if(-1 == s) {
        if(EAGAIN == errno || EWOULDBLOCK == errno) {
            writeNotifier->setEnabled(true);
        } else if(EPIPE == errno) {
#ifdef QUNIXSOCKET_DEBUG
            qDebug() << "QUnixSocket: Remote side disconnected during transmit "
                        "(" << ::strerror(errno) << ")";
#endif
            me->abort();
        } else {
#ifdef QUNIXSOCKET_DEBUG
            qDebug() << "QUnixSocket: Unable to transmit data ("
                     << ::strerror(errno) << ")";
#endif
            error = (QUnixSocket::SocketError)(QUnixSocket::WriteFailure |
                    CausedAbort);
            me->abort();
        }
    } else if(s != m.d->size()) {

        // A partial transmission
        writeNotifier->setEnabled(true);
        delete [] (char *)sendmessage.msg_control;
        m.d->rights = QList<QUnixSocketRights>();
        m.d->removeBytes( s );
        writeQueueBytes -= s;
        emit bytesWritten(s);
        return;

    } else {

        // Success!
        writeQueue.dequeue();
        Q_ASSERT(writeQueueBytes >= (unsigned)s);
        writeQueueBytes -= s;
        emit bytesWritten(s);

    }

    delete [] (char *)sendmessage.msg_control;
    if(-1 != s && !writeQueue.isEmpty())
        writeActivated();
    else if(QUnixSocket::ClosingState == me->state() && writeQueue.isEmpty())
        me->abort();
}

void QUnixSocketPrivate::readActivated()
{
#ifdef QUNIXSOCKET_DEBUG
    qDebug() << "QUnixSocket: readActivated";
#endif
    readNotifier->setEnabled(false);

    ::iovec vec;
    vec.iov_base = dataBuffer;
    vec.iov_len = dataBufferCapacity;

    bzero(&message, sizeof(::msghdr));
    message.msg_iov = &vec;
    message.msg_iovlen = 1;
    message.msg_controllen = ancillaryBufferCapacity();
    message.msg_control = ancillaryBuffer;

    int recvrv = ::recvmsg(fd, &message, 0);
#ifdef QUNIXSOCKET_DEBUG
    qDebug() << "QUnixSocket: Received message (" << recvrv << ")";
#endif
    if(-1 == recvrv) {
#ifdef QUNIXSOCKET_DEBUG
        qDebug() << "QUnixSocket: Unable to receive data ("
                 << ::strerror(errno) << ")";
#endif
        error = (QUnixSocket::SocketError)(QUnixSocket::ReadFailure |
                                           CausedAbort);
        me->abort();
    } else if(0 == recvrv) {
        me->abort();
    } else {
        Q_ASSERT(recvrv);
        Q_ASSERT((unsigned)recvrv <= dataBufferCapacity);
        dataBufferLength = recvrv;
        messageValid = true;

#ifdef QUNIXSOCKET_DEBUG
        qDebug() << "QUnixSocket: readyRead() " << dataBufferLength;
#endif
        emit readyRead();
    }
}

#include "qunixsocket.moc"