Pipe.cxx

Go to the documentation of this file.

 
#include <sys/select.h>
#include <unistd.h>
#include <fcntl.h>
 
#include "Devtab.hxx"
#include "utils/constants.hxx"
#include "utils/RingBuffer.hxx"
 
const size_t DEFAULT_PIPE_SIZE = 256;
 
class Pipe : public Node
{
public:
    Pipe(const char *name)
        : Node(name)
        , selInfoRd()
        , selInfoWr()
        , ring(NULL)
        , size(DEFAULT_PIPE_SIZE)
    {
        mode_ = S_IFIFO;
    }    
 
    ~Pipe()
    {
        if (ring)
        {
            ring->destroy();
        }
    }
 
    static int pipe(int pipefds[2]);
 
private:
    int close(File *file) OVERRIDE;
 
    ssize_t read(File *file, void *buf, size_t count) OVERRIDE;
 
    ssize_t write(File *file, const void *buf, size_t count) OVERRIDE;
 
    off_t lseek(File* f, off_t offset, int whence) override
    {
        return (off_t)-ESPIPE;
    }
 
    int fcntl(File *file, int cmd, unsigned long data) OVERRIDE;
 
    bool select(File* file, int mode) OVERRIDE;
 
    void enable() OVERRIDE {} 
    void disable() OVERRIDE {}; 
    void flush_buffers() OVERRIDE {}; 
    SelectInfo selInfoRd; 
    SelectInfo selInfoWr; 
    RingBuffer<uint8_t> *ring; 
    size_t size; 
    DISALLOW_COPY_AND_ASSIGN(Pipe);
};
 
ssize_t Pipe::read(File *file, void *buf, size_t count)
{
    if ((file->flags & O_ACCMODE) == O_WRONLY)
    {
        return -EBADF;
    }
 
    uint8_t *data = (uint8_t*)buf;
    ssize_t result = 0;
    
    while (count)
    {
        size_t bytes;
        {
            OSMutexLock locker(&lock_);
            if (ring == NULL)
            {
                ring = RingBuffer<uint8_t>::create(size);
            }
            bytes = ring->get(data, count);
 
            count -= bytes;
            result += bytes;
            data += bytes;
        }
 
        if (bytes)
        {
            select_wakeup(&selInfoWr);
        }
 
        if (count)
        {
            /* no more data to receive */
            if (file->flags & O_NONBLOCK || result > 0)
            {
                break;
            }
            else
            {
                /* blocking mode, wait for writer */
                fd_set rdfds;
                FD_ZERO(&rdfds);
                int fd = fd_lookup(file);
                FD_SET(fd, &rdfds);
                ::select(fd + 1, &rdfds, NULL, NULL, NULL);
            }
        }
    }
    
    return result;
}
 
ssize_t Pipe::write(File *file, const void *buf, size_t count)
{
    if ((file->flags & O_ACCMODE) == O_RDONLY)
    {
        return -EBADF;
    }
 
    const uint8_t *data = (const uint8_t*)buf;
    ssize_t result = 0;
    
    while (count)
    {
        size_t bytes;
        {
            OSMutexLock locker(&lock_);
            if (ring == NULL)
            {
                ring = RingBuffer<uint8_t>::create(size);
            }
            bytes = ring->put(data, count);
 
            count -= bytes;
            result += bytes;
            data += bytes;
        }
 
        if (bytes)
        {
            select_wakeup(&selInfoRd);
        }
 
        if (count)
        {
            /* no more room left */
            if (file->flags & O_NONBLOCK || result > 0)
            {
                break;
            }
            else
            {
                /* blocking mode, wait for reader */
                fd_set wrfds;
                FD_ZERO(&wrfds);
                int fd = fd_lookup(file);
                FD_SET(fd, &wrfds);
                ::select(fd + 1, NULL, &wrfds, NULL, NULL);
            }
        }
    }
    
    return result;
}
 
int Pipe::close(File *file)
{
    mutex.lock();
    if (--references_ == 0)
    {
        mutex.unlock();
        HASSERT(file->device);
        delete static_cast<Device *>(file->dev);
    }
    else
    {
        mutex.unlock();
    }
 
    return 0;
}
 
int Pipe::fcntl(File *file, int cmd, unsigned long data)
{
    switch (cmd)
    {
        default:
            return 0;
        case F_SETPIPE_SZ:
            size = data;
            return 0;
    }
}
 
bool Pipe::select(File* file, int mode)
{
    bool retval = false;
    switch (mode)
    {
        case FREAD:
            portENTER_CRITICAL();
            if (ring->items())
            {
                retval = true;
            }
            else
            {
                select_insert(&selInfoRd);
            }
            portEXIT_CRITICAL();
            break;
        case FWRITE:
            portENTER_CRITICAL();
            if (ring->space())
            {
                retval = true;
            }
            else
            {
                select_insert(&selInfoWr);
            }
            portEXIT_CRITICAL();
            break;
        default:
        case 0:
            /* we don't support any exceptions */
            break;
    }
    return retval;
}
 
 
int Pipe::pipe(int pipefds[2])
{
    Pipe *new_pipe = new Pipe(NULL);
 
    mutex.lock();
    pipefds[0] = fd_alloc();
    if (pipefds[0] < 0)
    {
        mutex.unlock();
        errno = EMFILE;
        return -1;
    }
    pipefds[1] = fd_alloc();
    mutex.unlock();
    if (pipefds[1] < 0)
    {
        fd_free(pipefds[0]);
        errno = EMFILE;
        return -1;
    }
 
    File *files[2] = {file_lookup(pipefds[0]), file_lookup(pipefds[1])};
 
    files[0]->dev = new_pipe;
    files[1]->dev = new_pipe;
    files[0]->flags = O_RDONLY;
    files[1]->flags = O_WRONLY;
 
    new_pipe->references_ = 2;
    new_pipe->enable();
 
    return 0;
}
 
int pipe(int pipefds[2])
{
    return Pipe::pipe(pipefds);
}