ExecStreamBufAccessor.h

Go to the documentation of this file.

/*
// $Id: //open/dev/fennel/exec/ExecStreamBufAccessor.h#20 $
// Fennel is a library of data storage and processing components.
// Copyright (C) 2005-2009 The Eigenbase Project
// Copyright (C) 2005-2009 SQLstream, Inc.
// Copyright (C) 2005-2009 LucidEra, Inc.
// Portions Copyright (C) 2004-2009 John V. Sichi
//
// This program is free software; you can redistribute it and/or modify it
// under the terms of the GNU General Public License as published by the Free
// Software Foundation; either version 2 of the License, or (at your option)
// any later version approved by The Eigenbase Project.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

#ifndef Fennel_ExecStreamBufAccessor_Included
#define Fennel_ExecStreamBufAccessor_Included

#include "fennel/exec/ExecStreamDefs.h"
#include "fennel/tuple/TupleDescriptor.h"
#include "fennel/tuple/TupleFormat.h"
#include "fennel/tuple/TupleAccessor.h"
#include "fennel/tuple/TupleProjectionAccessor.h"
#include "fennel/tuple/TupleOverflowExcn.h"

#include <boost/utility.hpp>

FENNEL_BEGIN_NAMESPACE

class FENNEL_EXEC_EXPORT ExecStreamBufAccessor
    : public boost::noncopyable
{
    PBuffer pBufStart;

    PBuffer pBufEnd;

    PBuffer pProducer;

    PBuffer pConsumer;

    ExecStreamBufProvision provision;

    ExecStreamBufState state;

    bool pendingEOS;

    TupleDescriptor tupleDesc;

    TupleFormat tupleFormat;

    TupleAccessor tupleProductionAccessor;

    TupleAccessor tupleConsumptionAccessor;

    TupleProjectionAccessor tupleProjectionAccessor;

    uint cbBuffer;

    inline void setEOS();

public:
    inline explicit ExecStreamBufAccessor();

    virtual ~ExecStreamBufAccessor()
    {
    }

    inline void setProvision(ExecStreamBufProvision provision);

    inline void setTupleShape(
        TupleDescriptor const &tupleDesc,
        TupleFormat tupleFormat = TUPLE_FORMAT_STANDARD);

    inline void clear();

    inline void provideBufferForProduction(
        PBuffer pStart,
        PBuffer pEnd,
        bool reusable);

    inline void provideBufferForConsumption(
        PConstBuffer pStart,
        PConstBuffer pEnd);

    inline void requestProduction();

    inline void requestConsumption();

    inline bool isProductionPossible() const;

    inline bool isConsumptionPossible() const;

    inline bool demandData();

    inline void markEOS();

    inline void produceData(PBuffer pEnd);

    inline void consumeData(PConstBuffer pEnd);

    inline PConstBuffer getConsumptionStart() const;

    inline PConstBuffer getConsumptionEnd() const;

    inline uint getConsumptionAvailable() const;

    uint getConsumptionAvailableBounded(uint cbLimit);

    inline uint getConsumptionTuplesAvailable();

    inline PConstBuffer spanWholeTuples(PConstBuffer start, uint size);

    inline PBuffer getProductionStart() const;

    inline PBuffer getProductionEnd() const;

    inline uint getProductionAvailable() const;

    inline ExecStreamBufState getState() const;

    inline bool hasPendingEOS() const;

    inline ExecStreamBufProvision getProvision() const;

    inline TupleDescriptor const &getTupleDesc() const;

    inline TupleFormat getTupleFormat() const;

    inline void validateTupleSize(TupleData const &tupleData);

    inline bool produceTuple(TupleData const &tupleData);

    inline TupleAccessor &accessConsumptionTuple();

    inline void unmarshalTuple(TupleData &tupleData, uint iFirstDatum = 0);

    inline void consumeTuple();

    inline bool isTupleConsumptionPending() const;

    inline TupleAccessor &getConsumptionTupleAccessor();

    inline TupleAccessor &getScratchTupleAccessor();

    inline void bindProjection(TupleProjection const &inputProj);

    inline void unmarshalProjectedTuple(TupleData &projTupleData);
};

inline ExecStreamBufAccessor::ExecStreamBufAccessor()
{
    clear();
    provision = BUFPROV_NONE;
    state = EXECBUF_EOS;
    tupleFormat = TUPLE_FORMAT_STANDARD;
    cbBuffer = 0;
}

inline bool ExecStreamBufAccessor::isProductionPossible() const
{
    return !pendingEOS && (state != EXECBUF_EOS) && (state != EXECBUF_OVERFLOW);
}

inline bool ExecStreamBufAccessor::isConsumptionPossible() const
{
    return (state == EXECBUF_OVERFLOW) || (state == EXECBUF_NONEMPTY);
}

inline void ExecStreamBufAccessor::setProvision(
    ExecStreamBufProvision provisionInit)
{
    assert(provision == BUFPROV_NONE);
    provision = provisionInit;
}

inline void ExecStreamBufAccessor::setTupleShape(
    TupleDescriptor const &tupleDescInit,
    TupleFormat tupleFormatInit)
{
    tupleDesc = tupleDescInit;
    tupleFormat = tupleFormatInit;
    tupleProductionAccessor.compute(tupleDesc, tupleFormat);
    tupleConsumptionAccessor.compute(tupleDesc, tupleFormat);
}

inline void ExecStreamBufAccessor::clear()
{
    pBufStart = NULL;
    pBufEnd = NULL;
    pProducer = NULL;
    pConsumer = NULL;
    cbBuffer = 0;
    state = EXECBUF_EMPTY;
    pendingEOS = false;
    tupleProductionAccessor.resetCurrentTupleBuf();
    tupleConsumptionAccessor.resetCurrentTupleBuf();
}

inline void ExecStreamBufAccessor::provideBufferForProduction(
    PBuffer pStart,
    PBuffer pEnd,
    bool reusable)
{
    assert((state == EXECBUF_UNDERFLOW) || (state == EXECBUF_EMPTY));
    assert(provision == BUFPROV_CONSUMER);
    pBufStart = pStart;
    pBufEnd = pEnd;
    pProducer = pStart;
    pConsumer = pStart;
    cbBuffer = pEnd - pStart;
    state = EXECBUF_UNDERFLOW;

    if (!reusable) {
        // indicate that this buffer is not reusable
        pBufStart = NULL;
    }
}

inline void ExecStreamBufAccessor::provideBufferForConsumption(
    PConstBuffer pStart,
    PConstBuffer pEnd)
{
    assert((state == EXECBUF_UNDERFLOW) || (state == EXECBUF_EMPTY));
    assert(provision == BUFPROV_PRODUCER);
    pBufStart = const_cast<PBuffer>(pStart);
    pBufEnd = const_cast<PBuffer>(pEnd);
    pConsumer = pBufStart;
    pProducer = pBufEnd;
    state = EXECBUF_OVERFLOW;

    // indicate that this buffer is not reusable
    pBufStart = NULL;
}

inline void ExecStreamBufAccessor::requestProduction()
{
    assert((state == EXECBUF_UNDERFLOW) || (state == EXECBUF_EMPTY));
    state = EXECBUF_UNDERFLOW;
    pProducer = pBufStart;
    pConsumer = pBufStart;
}

inline void ExecStreamBufAccessor::requestConsumption()
{
    assert((state == EXECBUF_OVERFLOW) || (state == EXECBUF_NONEMPTY));
    state = EXECBUF_OVERFLOW;
}

inline void ExecStreamBufAccessor::markEOS()
{
    if (isConsumptionPossible()) {
        pendingEOS = true;
        return;
    }
    setEOS();
}

inline void ExecStreamBufAccessor::setEOS()
{
    assert(pProducer == pConsumer);
    clear();
    state = EXECBUF_EOS;
}

inline PConstBuffer ExecStreamBufAccessor::getConsumptionStart() const
{
    return pConsumer;
}

inline PConstBuffer ExecStreamBufAccessor::getConsumptionEnd() const
{
    return pProducer;
}

inline uint ExecStreamBufAccessor::getConsumptionAvailable() const
{
    return getConsumptionEnd() - getConsumptionStart();
}

inline uint ExecStreamBufAccessor::getConsumptionTuplesAvailable()
{
    TupleAccessor& acc = getScratchTupleAccessor();
    PConstBuffer p = getConsumptionStart(),
        end = getConsumptionEnd();
    int count = 0;
    while (p < end) {
        acc.setCurrentTupleBuf(p);
        p += acc.getCurrentByteCount();
        ++count;
    }
    return count;
}

inline PConstBuffer ExecStreamBufAccessor::spanWholeTuples(
    PConstBuffer start, uint size)
{
    TupleAccessor& acc = getScratchTupleAccessor();
    assert(size > 0);
    PConstBuffer p = start;
    PConstBuffer pend = start + size;
    for (int ct = 0; ; ct++) {
        assert(p < pend);
        acc.setCurrentTupleBuf(p);
        PConstBuffer q = p;
        p += acc.getCurrentByteCount(); // forward 1 tuple
        if (p >= pend) {
            if (p == pend) {
                // fit the current tuple, but no more
                return p;
            } else {
                // here when p is too far, and the tuple [q, p] did not fit.
                return q;
            }
        }
    }
    assert(false); // not reached
}


inline PBuffer ExecStreamBufAccessor::getProductionStart() const
{
    return pProducer;
}

inline PBuffer ExecStreamBufAccessor::getProductionEnd() const
{
    return pBufEnd;
}

inline uint ExecStreamBufAccessor::getProductionAvailable() const
{
    return pProducer ? (pBufEnd - pProducer) : 0;
}

inline ExecStreamBufState ExecStreamBufAccessor::getState() const
{
    return state;
}

inline bool  ExecStreamBufAccessor::hasPendingEOS() const
{
    return pendingEOS;
}

inline ExecStreamBufProvision ExecStreamBufAccessor::getProvision() const
{
    return provision;
}

inline TupleFormat ExecStreamBufAccessor::getTupleFormat() const
{
    return tupleFormat;
}

inline TupleDescriptor const &ExecStreamBufAccessor::getTupleDesc() const
{
    return tupleDesc;
}

inline void ExecStreamBufAccessor::produceData(PBuffer pEnd)
{
    assert(isProductionPossible());
    assert(pEnd > getProductionStart());
    assert(pEnd <= getProductionEnd());
    pProducer = pEnd;
    state = EXECBUF_NONEMPTY;
}

inline void ExecStreamBufAccessor::consumeData(PConstBuffer pEnd)
{
    assert(isConsumptionPossible());
    assert(pEnd > getConsumptionStart());
    assert(pEnd <= getConsumptionEnd());
    pConsumer = const_cast<PBuffer>(pEnd);
    if (pConsumer == getConsumptionEnd()) {
        if (pendingEOS) {
            setEOS();
        } else {
            state = EXECBUF_EMPTY;
        }
    } else {
        // NOTE jvs 9-Nov-2004:  this is misleading until circular buffering
        // gets implemented, but it isn't incorrect either
        state = EXECBUF_NONEMPTY;
    }
}

inline void ExecStreamBufAccessor::validateTupleSize(
    TupleData const &tupleData)
{
    if (cbBuffer == 0) {
        return;                         // no buffer yet
    }
    if (!tupleProductionAccessor.isBufferSufficient(tupleData,  cbBuffer)) {
        uint cbTuple = tupleProductionAccessor.getByteCount(tupleData);
        throw TupleOverflowExcn(tupleDesc, tupleData, cbTuple, cbBuffer);
    }
}

inline bool ExecStreamBufAccessor::produceTuple(TupleData const &tupleData)
{
    assert(getState() != EXECBUF_EOS);
    assert(!pendingEOS);

    if (tupleProductionAccessor.isBufferSufficient(
            tupleData, getProductionAvailable()))
    {
        tupleProductionAccessor.marshal(tupleData, getProductionStart());
        produceData(
            getProductionStart()
            + tupleProductionAccessor.getCurrentByteCount());
        return true;
    } else {
        validateTupleSize(tupleData);
        if (getState() == EXECBUF_NONEMPTY) {
            requestConsumption();
        }
        return false;
    }
}

inline TupleAccessor &ExecStreamBufAccessor::accessConsumptionTuple()
{
    assert(isConsumptionPossible());
    assert(!tupleConsumptionAccessor.getCurrentTupleBuf());

    tupleConsumptionAccessor.setCurrentTupleBuf(getConsumptionStart());
    return tupleConsumptionAccessor;
}

inline void ExecStreamBufAccessor::unmarshalTuple(
    TupleData &tupleData, uint iFirstDatum)
{
    accessConsumptionTuple();
    tupleConsumptionAccessor.unmarshal(tupleData, iFirstDatum);
}

inline void ExecStreamBufAccessor::consumeTuple()
{
    assert(tupleConsumptionAccessor.getCurrentTupleBuf());

    consumeData(
        getConsumptionStart() + tupleConsumptionAccessor.getCurrentByteCount());
    tupleConsumptionAccessor.resetCurrentTupleBuf();
}

inline bool ExecStreamBufAccessor::isTupleConsumptionPending() const
{
    if (tupleConsumptionAccessor.getCurrentTupleBuf()) {
        return true;
    } else {
        return false;
    }
}

inline TupleAccessor &ExecStreamBufAccessor::getConsumptionTupleAccessor()
{
    return tupleConsumptionAccessor;
}

inline TupleAccessor &ExecStreamBufAccessor::getScratchTupleAccessor()
{
    // this can be used for scratch purposes since we don't need its state
    // across calls to produceTuple
    return tupleProductionAccessor;
}

inline bool ExecStreamBufAccessor::demandData()
{
    if (state == EXECBUF_EOS) {
        return false;
    } else if (isConsumptionPossible()) {
        return true;
    } else {
        requestProduction();
        return false;
    }
}

inline void ExecStreamBufAccessor::bindProjection(
    TupleProjection const &inputProj)
{
    tupleProjectionAccessor.bind(tupleConsumptionAccessor, inputProj);
}

inline void ExecStreamBufAccessor::unmarshalProjectedTuple(
    TupleData &projTupleData)
{
    accessConsumptionTuple();
    tupleProjectionAccessor.unmarshal(projTupleData);
}

FENNEL_END_NAMESPACE

#endif

// End ExecStreamBufAccessor.h

---