nxio-read.c

MANTIS是由科罗拉多大学开发的传感器网络嵌入式操作系统。 这是mantis的0.9.5版本的源码。

#include <errno.h>
#include <string.h>
#include <unistd.h>
#include <sys/time.h>
#include <netinet/in.h>

#include <glib.h>

#include "bionet-util.h"
#include "bionet-nxio.h"




// 
// The 'acceptable' argument is an array of "const char **", the last one
// of which is NULL.  The function keeps reading messages until the timeout
// expires or it receives one of the acceptable messages.  If any messages
// are received other than the acceptable ones, they are put on the queue.
//

int bionet_nxio_read_acceptable(bionet_nxio_t *nxio, struct timeval *timeout, xmlDoc **xml, const char **acceptable, GSList **queue) {
    int i;
    int r;

    struct timeval start, timeout_remaining;


    timeout_remaining.tv_sec = timeout->tv_sec;
    timeout_remaining.tv_usec = timeout->tv_usec;


    *xml = NULL;


    r = gettimeofday(&start, NULL);
    if (r < 0) {
        g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_WARNING, "error getting time of day: %s", strerror(errno));
        return -1;
    }


    do {
        int r;
        xmlNode *reply;


        r = bionet_nxio_read(nxio, &timeout_remaining, xml);
        if (r < 0) {
            return -1;
        }

        if (*xml == NULL) {
            // timeout
            return 0;
        }

        reply = xmlDocGetRootElement(*xml);

        for (i = 0; acceptable[i] != NULL; i ++) {
            if (strcmp(reply->name, acceptable[i]) == 0) {
                // got an acceptable message, return it to caller
                return 0;
            }
        }

        // not an acceptable message, queue it
        *queue = g_slist_append(*queue, *xml);


        *xml = NULL;


        {
            struct timeval now, waited_so_far;
            int r;

            r = gettimeofday(&now, NULL);
            if (r < 0) {
                g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_WARNING, "error getting time of day: %s", strerror(errno));
                return -1;
            }

            waited_so_far.tv_sec = (now.tv_sec - start.tv_sec);
            waited_so_far.tv_usec = (now.tv_usec - start.tv_usec);
            if (waited_so_far.tv_usec < 0) {
                waited_so_far.tv_sec --;
                waited_so_far.tv_usec += 1000000;
            }

            timeout_remaining.tv_sec = timeout->tv_sec - waited_so_far.tv_sec;
            timeout_remaining.tv_usec = timeout->tv_usec - waited_so_far.tv_sec;
            if (timeout_remaining.tv_usec < 0) {
                timeout_remaining.tv_sec --;
                timeout_remaining.tv_usec += 1000000;
            }

            if (timeout_remaining.tv_sec < 0) {
                return 0;
            }
        }
    } while (1);

    // NOT REACHED
    return 0;
}




//
//  reads one XML message, unless it times out first
//
//  Returns 0 if all went well (*xml will point to the XML document that was received, or will be NULL if it was a timeout).
//
//  Returns -1 on error.
//

int bionet_nxio_read(bionet_nxio_t *nxio, struct timeval *timeout, xmlDoc **xml) {
    int bytes_to_read;
    uint32_t expected_packet_size;

    struct timeval start, timeout_remaining;

    int r;


    if (timeout == NULL) {
        g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_WARNING, "bionet_nxio_read(): NULL timeout passed in");
        return NXIO_CLOSE;
    }

    timeout_remaining.tv_sec = timeout->tv_sec;
    timeout_remaining.tv_usec = timeout->tv_usec;


    *xml = (xmlDoc *)NULL;


#ifdef NXIO_DEBUG
    // FIXME
    g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_INFO, "on entry to nxio_read, index=%d", nxio->in_index);
    log_hexdump(NXIO_LOG_DOMAIN, G_LOG_LEVEL_INFO, nxio->in_buffer, nxio->in_index);
#endif

    r = gettimeofday(&start, NULL);
    if (r < 0) {
        g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_WARNING, "bionet_nxio_read(): error getting time of day: %s", strerror(errno));
        return NXIO_CLOSE;
    }


    do {
        struct timeval now, waited_so_far;
        int r;

        // wait for the socket to become readable
        {
            fd_set read_fds;
            int r;

#ifdef NXIO_DEBUG
            g_log(
                NXIO_LOG_DOMAIN,
                G_LOG_LEVEL_WARNING,
                "nxio_read(): waiting to read, started=%d.%06d, timeout_remaining=%d.%06d",
                (int)start.tv_sec, (int)start.tv_usec,
                (int)timeout_remaining.tv_sec, (int)timeout_remaining.tv_usec
            );
#endif

            FD_ZERO(&read_fds);
            FD_SET(nxio->socket, &read_fds);

            r = select(nxio->socket + 1, &read_fds, NULL, NULL, &timeout_remaining);

            if (r < 0) {
                // error
                g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_WARNING, "bionet_nxio_read(): error selecting on socket: %s", strerror(errno));
                return NXIO_CLOSE;
            }

            if (r == 0) {
                // timeout
                return 0;
            }
        }


        //
        // if we get here, the nxio socket is readable
        //


        // 
        // make sure we've read the framing header before trying to parse it
        // unfortunately this means it takes at least two read(2) calls to read each packet...
        //

        if (nxio->in_index < NXIO_HEADER_SIZE) {
            // we dont have the framing header yet, so try to read it

            int r;

            r = read(nxio->socket, &nxio->in_buffer[nxio->in_index], NXIO_HEADER_SIZE - nxio->in_index);

            if (r == -1) {
                if (errno == EAGAIN) {
                    // tried to read on an empty non-blocking socket
                    // this should never happen if the caller behaves, but hey
                    return 0;
                }
                // read error
                g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_WARNING, "bionet_nxio_read(): error reading from socket: %s", strerror(errno));
                return NXIO_CLOSE;
            }

            if (r == 0) {
                // peer closed connection
                g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_WARNING, "bionet_nxio_read(): socket closed on the far side");
                return NXIO_CLOSE;
            }


            nxio->in_index += r;

#ifdef NXIO_DEBUG
            // FIXME
            g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_INFO, "read %d bytes, now have %d bytes", r, nxio->in_index);
            log_hexdump(NXIO_LOG_DOMAIN, G_LOG_LEVEL_INFO, nxio->in_buffer, nxio->in_index);
#endif

            if (nxio->in_index < NXIO_HEADER_SIZE) {
                // we read all there was, which was not enough, keep waiting
                goto keep_waiting;
            }
        }


        //
        // if we get here, we have the incoming packet header
        //


        expected_packet_size = ntohl(*(uint32_t*)nxio->in_buffer);
        if (expected_packet_size > NXIO_MAX_MESSAGE_SIZE) {
            g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_WARNING, "frame size %u exceeds max size %d!", expected_packet_size, NXIO_MAX_MESSAGE_SIZE);
            return NXIO_CLOSE;
        }

        bytes_to_read = expected_packet_size - (nxio->in_index - NXIO_HEADER_SIZE);

        if (bytes_to_read > 0) {
            int r;

            r = read(nxio->socket, &nxio->in_buffer[nxio->in_index], bytes_to_read);

            if (r == -1) {
                if (errno == EAGAIN) {
                    // tried to read on an empty non-blocking socket
                    // this should never happen if the caller behaves, but hey
                    return 0;
                }
                // read error
                g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_WARNING, "bionet_nxio_read(): error reading from socket: %s", strerror(errno));
                return NXIO_CLOSE;
            }

            if (r == 0) {
                // peer closed connection
                g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_WARNING, "bionet_nxio_read(): socket closed on the far side");
                return NXIO_CLOSE;
            }


            nxio->in_index += r;


#ifdef NXIO_DEBUG
            // FIXME
            g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_INFO, "read %d bytes, now have %d bytes", r, nxio->in_index);
            log_hexdump(NXIO_LOG_DOMAIN, G_LOG_LEVEL_INFO, nxio->in_buffer, nxio->in_index);
#endif

            if (r < bytes_to_read) {
                // we read all there was, which was not enough, keep waiting
                goto keep_waiting;
            }

            break;
        }


keep_waiting:
        r = gettimeofday(&now, NULL);
        if (r < 0) {
            g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_WARNING, "error getting time of day: %s", strerror(errno));
            return NXIO_CLOSE;
        }

        waited_so_far.tv_sec = (now.tv_sec - start.tv_sec);
        waited_so_far.tv_usec = (now.tv_usec - start.tv_usec);
        if (waited_so_far.tv_usec < 0) {
            waited_so_far.tv_sec --;
            waited_so_far.tv_usec += 1000000;
        }

        timeout_remaining.tv_sec = timeout->tv_sec - waited_so_far.tv_sec;
        timeout_remaining.tv_usec = timeout->tv_usec - waited_so_far.tv_usec;
        if (timeout_remaining.tv_usec < 0) {
            timeout_remaining.tv_sec --;
            timeout_remaining.tv_usec += 1000000;
        }

#ifdef NXIO_DEBUG
        g_log(
            NXIO_LOG_DOMAIN,
            G_LOG_LEVEL_WARNING,
            "nxio_read(): read some but didnt finish, started=%d.%06d, now=%d.%06d, waited_so_far=%d.%06d, timeout_remaining=%d.%06d",
            (int)start.tv_sec, (int)start.tv_usec,
            (int)now.tv_sec, (int)now.tv_usec,
            (int)waited_so_far.tv_sec, (int)waited_so_far.tv_usec,
            (int)timeout_remaining.tv_sec, (int)timeout_remaining.tv_usec
        );
#endif

        if (timeout_remaining.tv_sec < 0) {
            return 0;
        }
    } while (1);


    //
    // if we get here, we have the whole packet
    //


    // is it a packet we know how to deal with?
    // 0x00: XML message
    if (nxio->in_buffer[4] != 0x00) {
        g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_WARNING, "unknown message type 0x%02hhX, skipping", nxio->in_buffer[4]);
        nxio->in_index = 0;
        return 0;
    }



    {
        xmlDoc *xml_doc;

        xml_doc = xmlParseMemory(&nxio->in_buffer[NXIO_HEADER_SIZE], expected_packet_size);
        if (xml_doc == NULL) {
            g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_WARNING, "failed to parse XML message:");
            // FIXME
            // clogprint_hex(NXIO_LPC, G_LOG_LEVEL_WARNING, "    ", &nxio->in_buffer[2], nxio->in_index - 2);
            nxio->in_index = 0;
            return NXIO_INVALID_XML;
        }

        if ((nxio->in_dtd != NULL) && (nxio->in_validation_context != NULL)) {
            int r;
            r = xmlValidateDtd(nxio->in_validation_context, xml_doc, nxio->in_dtd);
            if (r == 0) {
                g_log(NXIO_LOG_DOMAIN, G_LOG_LEVEL_WARNING, "XML message fails DTD validation:");
                // FIXME
                // clogprint_hex(NXIO_LPC, G_LOG_LEVEL_WARNING, "    ", &nxio->in_buffer[2], nxio->in_index - 2);
                nxio->in_index = 0;
                xmlFreeDoc(xml_doc);
                return NXIO_INVALID_XML;
            }
        }

        nxio->in_index = 0;
        *xml = xml_doc;
    }

    return 0;
}