---

FlowManager.cpp

#define TRACE_NAME "FlowManager"
#include "FlowManager.h"
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <errno.h>
#include <stdio.h>
#include <sys/time.h>
#include <unistd.h>
#include <cassert>


//
//  Standard Member Functions
//

FlowManager FlowManager::main;

FlowManager::FlowManager()
    : streams(16), runOnlyPending(false), maxfd(-1)
{
    for (int i = 0; i < 16; i++) streams[i] = 0;
}

#if 0  /* omit definitions until they are implemented */

FlowManager::FlowManager(const FlowManager& other)
{
    NOT_IMPLEMENTED
}

const FlowManager& FlowManager::operator=(const FlowManager& other)
{
    NOT_IMPLEMENTED
}

bool FlowManager::operator==(const FlowManager& other) const
{
    NOT_IMPLEMENTED
}

bool FlowManager::operator<(const FlowManager& other) const
{
    NOT_IMPLEMENTED
}

size_t FlowManager::hash() const
{
    NOT_IMPLEMENTED
}

std::ostream& FlowManager::print_to(std::ostream& stream) const
{
    NOT_IMPLEMENTED
}

#endif /* omit definitions until they are implemented */
    
FlowManager::~FlowManager()
{
    // blank default    
}

//
//  Additional Public Member Functions
//

static bool absoluteURI(const char* s) 
{
    for (const char* p=s; *p; p++) {
    switch (*p) {
    case ':':
        if (p==s) {
        return false;
        } else {
        return true;
        }
    case 'a':   case 'b':   case 'c':   case 'd':
    case 'e':   case 'f':   case 'g':   case 'h':
    case 'i':   case 'j':   case 'k':   case 'l':
    case 'm':   case 'n':   case 'o':   case 'p':
    case 'q':   case 'r':   case 's':   case 't':
    case 'u':   case 'v':   case 'w':   case 'x':
    case 'y':   case 'z':
    case 'A':   case 'B':   case 'C':   case 'D':
    case 'E':   case 'F':   case 'G':   case 'H':
    case 'I':   case 'J':   case 'K':   case 'L':
    case 'M':   case 'N':   case 'O':   case 'P':
    case 'Q':   case 'R':   case 'S':   case 'T':
    case 'U':   case 'V':   case 'W':   case 'X':
    case 'Y':   case 'Z':
    case '-':
        // so far it might be; let's keep looking...
        break;
    default:
        // it can't be a URI-Scheme, so this must be relative
        return false;
    }
    }
    return false;
}

FlowManager::Stream::Stream(const char* address, 
                ByteSink* streamTo, 
                Direction direction,
                bool returnWhenDone)
    : 
    address(address), 
    streamTo(streamTo), 
    direction(direction), 
    fd(-1),
    wfd(-1),
    returnWhenDone(returnWhenDone)
{

    if (!strncmp(address, "-", 1)) {
    switch (direction) {
    case FlowManager::READ:
        fd = 0;
        break;
    case FlowManager::WRITE:
        fd = 1;
        break;
    case FlowManager::READWRITE:
        fd = 0;
        break;
    }
    return;
    }

    if (absoluteURI(address)) {
    
    if (!strcmp(address, "x-shell:")) {

        TRACE "forking a subprocess!\n";

        int toChild[2];
        int fromChild[2];
        assert(0 == pipe(toChild));
        assert(0 == pipe(fromChild));

        pid_t pid = fork();

        if (pid) {
        // in parent
        close(toChild[0]);
        close(fromChild[1]);
        fd = fromChild[0];
        wfd = toChild[1];
        return;
        } else {
        // in child
        close(toChild[1]);
        close(fromChild[0]);
        assert(-1 != dup2(toChild[0], 0));
        assert(-1 != dup2(fromChild[1], 1));
        // leave stderr as is....
        // or close it...?
        // or make it a file?
        char* s = strdup("/tmp/blindfold.stderr.XXXXXX");
        int efd = mkstemp(s);
        assert(-1 != dup2(efd, 2));
        free(s);
        assert(-1 != execlp("sh", "sh", 0));
        }
    } else {

        // should do full parsing, use http (etc) subsystems

        // IMPLEMENT   "x-shell:xsb%20"  :-)
        //    as fork & set async
        //    have a timer-based run-queue thing, so we 
        //    CAN set timeouts.   And feed those into
        //    the stream from a process.   And how to kill?
        //     .drop(Stream) ...?
        //

        std::cerr << "Can't open \"" << address << "\": ";
        std::cerr << "Absolute-URI addresses are not implemented." << std::endl;
        exit(1);

    }
    }
    int flags = 0;
    switch (direction) {
    case FlowManager::READ:
    flags = O_RDONLY;
    break;
    case FlowManager::WRITE:
    flags = O_CREAT | O_WRONLY | O_TRUNC;
    break;
    case FlowManager::READWRITE:
    flags = O_RDWR | O_APPEND;
    break;
    }
    flags |= O_NDELAY;

    fd = open(address, flags, 0600);

    if (fd == -1) {
    perror(address);
    // should throw error
    exit(1);
    }
}


void FlowManager::runPending()
{
    runOnlyPending = true;
    run();
    runOnlyPending = false;
}

void FlowManager::run()
{
    fd_set readfds, writefds, exceptfds;
    struct timeval tv;
    int retval;
    char buf[8192];
    ssize_t bytes;

    keepRunning = true;
    while (keepRunning) {

    //
    // Set flags for select
    //

    FD_ZERO(&readfds);
    FD_ZERO(&writefds);
    FD_ZERO(&exceptfds);

    bool work_to_do = false;
    for (int i = 0; i <= maxfd; i++) {
        if (streams[i] == 0) continue;
        TRACE "  stream " << i << ": " << streams[i] << std::endl;
        switch (streams[i]->direction) {
        case READ:
        FD_SET(i, &readfds);
        work_to_do = true;
        TRACE "  (want to read)" << std::endl;
        break;
        case WRITE:
        if (streams[i]->streamFrom.str().size()) {
            if (streams[i]->wfd == -1) {
            FD_SET(i, &writefds);
            TRACE "  (want to write on normal fd)" << std::endl;
            } else {
            FD_SET(streams[i]->wfd, &writefds);
            TRACE "  (want to write on " << streams[i]->wfd << std::endl;
            }
            work_to_do = true;
            TRACE "  ... with some bytes to write" << std::endl;
        }
        break;
        case READWRITE:
        FD_SET(i, &readfds);
        work_to_do = true;
        if (streams[i]->streamFrom.str().size()) {
            if (streams[i]->wfd == -1) {
            FD_SET(i, &writefds);
            TRACE "  (want to write on normal fd)" << std::endl;
            } else {
            FD_SET(streams[i]->wfd, &writefds);
            TRACE "  (want to write on " << streams[i]->wfd << std::endl;
            }
            work_to_do = true;
            TRACE "  ... with some bytes to write" << std::endl;
        }
        break;
        }
        FD_SET(i, &exceptfds);
    }
    
    if (runOnlyPending && !work_to_do) {
        TRACE "Returning from runPending -- no more work to do." << std::endl;
        return;
    }

    /* Wait up to 0.05 seconds (~human response time)  */
    tv.tv_sec = 0;
    tv.tv_usec = 50000;

    //
    // Call Select
    //
    retval = select(maxfd+1, &readfds, &writefds, &exceptfds, &tv);

    //
    // Check & act on flags from select
    //
    if (retval) {
        TRACE "select returned " << retval << std::endl;
        for (int i = 0; i <= maxfd; i++) {
        Stream* s = streams[i];
        if (!s) continue;
        if (FD_ISSET(i, &readfds)) {
            bytes = read(i, &buf, sizeof(buf));
            TRACE "readable fd " << i << ", " << bytes << " read\n";
            if (bytes < 0) { 
            if (errno == EINTR || errno == EAGAIN) {
                // do nothing; we'll just try again
            } else {
                perror("read"); 
                exit(1);
            }
            } else {
            if (bytes != 0) {
                s->streamTo->write(&buf, bytes);
            } else {
                s->streamTo->close();
                if (s->direction == READ) {
                bool r = streams[i]->returnWhenDone;
                delete(s);
                streams[i] = 0;
                if (r) keepRunning = false;
                } else {
                // BUG remove read portion, but what
                // about write?
                bool r = streams[i]->returnWhenDone;
                delete(s);
                streams[i] = 0;
                if (r) keepRunning = false;
                }
            }
            }
        }
        int wfd = i;
        if (s->wfd != -1) wfd = s->wfd;
        if (FD_ISSET(wfd, &writefds)) {
            
            int bytes_avail = s->streamFrom.str().size();
            bytes = write(wfd, s->streamFrom.str().c_str(), bytes_avail);
            TRACE "writable fd " << i << "/" << wfd << ", " << bytes << " written (of "
                  << bytes_avail << "queued)" << std::endl;
            if (bytes >= 0) {
            //s->streamFrom.ignore(bytes, '\07ff');   not working
            std::string leftover=s->streamFrom.str().substr(bytes);
            s->streamFrom.str(leftover);
            bytes_avail = s->streamFrom.str().size();
            TRACE "  queued is now "  << bytes_avail << std::endl;
            if (bytes == bytes_avail) {
                // eof & fail flags probably just got set.  clear them!
                s->streamFrom.clear(0);
            }
            } else {
            if (errno == EINTR || errno == EAGAIN) {
                // do nothing; we'll just try again
            } else {
                perror("write"); 
                exit(1);
            }
            }
        }
        if (FD_ISSET(i, &exceptfds)) {
            std::cerr << "ignoring exception on fd " << i << std::endl;
        }
        if (wfd!=i && FD_ISSET(wfd, &exceptfds)) {
            std::cerr << "ignoring exception on fd " << i << std::endl;
        }
        }
    } else {
        TRACE "Select timed out.\n";
        if (runOnlyPending) {
        TRACE "Returning from runPending, because of timeout, even with work to do.   This may be a bug." << std::endl;
        return;
    }

    }
    
    // try runnables...

    }

    // select

    // try to read FD's we can

    // look at our output FIFOs and write anything we can
}



//
//  Additional Private Member Functions
//

//    int open(char* address, Direction direction);

void FlowManager::addStream(Stream* s) 
{
    TRACE "new fd is " << s->fd << std::endl;
    if (s->fd > maxfd) {
    maxfd=s->fd;
    TRACE "maxfd upped to " << maxfd << std::endl;
    } else {
    TRACE "which is <= " << maxfd << std::endl;
    }
    if (s->fd+1 > (int) streams.size()) {
    streams.resize(s->fd+1);
    TRACE "streams resized to " << s->fd+1 << std::endl;
    }
    streams[s->fd] = s;
}
#undef TRACE_NAME

---

Home to blindfold. This page generated via doxygen 1.2.11.1 Wed Oct 10 16:40:33 2001.