Dryad/DryadVertex/VertexHost/system/channel/include/channelbuffer.h

/*
Copyright (c) Microsoft Corporation

All rights reserved.

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in 
compliance with the License.  You may obtain a copy of the License 
at http://www.apache.org/licenses/LICENSE-2.0   


THIS CODE IS PROVIDED *AS IS* BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, EITHER 
EXPRESS OR IMPLIED, INCLUDING WITHOUT LIMITATION ANY IMPLIED WARRANTIES OR CONDITIONS OF 
TITLE, FITNESS FOR A PARTICULAR PURPOSE, MERCHANTABLITY OR NON-INFRINGEMENT.  


See the Apache Version 2.0 License for specific language governing permissions and 
limitations under the License. 

*/

#pragma once

#include <DrCommon.h>
#include <channelitem.h>
#include <channelmemorybuffers.h>
#include <dvertexcommand.h>

class RChannelBufferReader;

#define RCHANNEL_BUFFER_OFFSET_UNDEFINED ((UInt64) -1)

/* this is a dummy class which may be extended in future to allow item
   parsers to communicate with the buffer reader to influence i/o
   buffer sizes, prefetching strategies, etc. Nobody uses it yet and
   NULL should be passed wherever there is a prefetchCookie argument
   for now. */
class RChannelBufferPrefetchInfo
{
public:
    virtual ~RChannelBufferPrefetchInfo();
};


/* buffers read from a channel are typed with one of the following: */
enum RChannelBufferType {
    /* RChannelBuffer_Data: this is a data buffer which can be cast to
       an RChannelBufferData object to retrieve the payload. */
    RChannelBuffer_Data,

    /* the following are all marker buffers which can be cast to an
       RChannelBufferMarker object to retrieve the item describing the
       marker. */

    /* RChannelBuffer_Hole: this describes a "hole" in the
         underlying data stream. */
    RChannelBuffer_Hole,

    /* RChannelBuffer_EndOfStream: this is a marker indicating that
       the end of the stream has been succesfully reached. */
    RChannelBuffer_EndOfStream,

    /* RChannelBuffer_Restart: this is a marker indicating that the
       remote end has requested a restart. */
    RChannelBuffer_Restart,

    /* RChannelBuffer_Abort: this is a marker indicating that the
       underlying stream has suffered an unrecoverable error, restart
       is impossible, and no more data is forthcoming. */
    RChannelBuffer_Abort
};

/* base class for data buffers generated by the byte-oriented read
   layer of a channel */
class RChannelBuffer : public DrRefCounter
{
public:
    static bool IsTerminationBuffer(RChannelBufferType type);

    /* When a consumer has finished using the data in a buffer it
       should call this completion handler returning the
       buffer.

       prefetchCookie is an optional hint which may be used to
       influence subsequent read buffer sizes or prefetching
       behaviour. It is dependent on the implementation of the
       underlying buffer-oriented i/o class and should be NULL for
       now.

       The completion callback mechanism is used to implement flow
       control, as the buffer-oriented i/o will only allow the
       consumer to hold a bounded number of outstanding buffers before
       blocking further reads.
    */
    virtual void
        ProcessingComplete(RChannelBufferPrefetchInfo* prefetchCookie) = 0;

    /* get the buffer's type.
     */
    RChannelBufferType GetType();

    /* this returns a description of the buffer, used for debugging
       and monitoring purposes.
    */
    DryadMetaData* GetMetaData();

    /* this replaces the item's current metadata with a new object */
    void ReplaceMetaData(DryadMetaData* metaData);

protected:
    RChannelBuffer(RChannelBufferType type);
    virtual ~RChannelBuffer();

private:
    RChannelBufferType    m_type;
    DryadMetaDataRef      m_metaData;
    DrBListEntry          m_listPtr;
    friend class DryadBList<RChannelBuffer>;
};

typedef class DryadBList<RChannelBuffer> ChannelBufferList;

/* this is a default completion handler used to return buffers of type
   RChannelBufferDataDefault and RChannelBufferMarkerDefault. It is
   called from the ProcessingComplete methods of each of those
   classes. */
class RChannelBufferDefaultHandler
{
public:
    /* the callee owns a reference to buffer after this call */
    virtual void ReturnBuffer(RChannelBuffer* buffer) = 0;
};

class ChannelDataBufferList;

class RChannelBufferData : public RChannelBuffer
{
public:
    /* The payload of a data buffer is a DryadLockedMemoryBuffer.
       This cannot be grown, nor can its available size be
       modified. It is derived from a DryadFixedMemoryBuffer and hence
       it is guaranteed to wrap a single contiguous memory region. It
       has the same thread-safety properties as the underlying memory
       region, e.g. it can be read concurrently from multiple threads,
       and if the threads co-operate they can safely concurrently
       write to non-overlapping regions.

       The caller must increment the reference count of the returned
       DryadLockedMemoryBuffer if it wishes to use it after the
       RChannelBufferData goes out of scope.
    */
    virtual DryadLockedMemoryBuffer* GetData() = 0;

    /* this returns a metadata description of an offset in the buffer
       (which may include a description of the buffer, along with
       e.g. the offset's overall position in the stream), used for
       debugging and monitoring purposes. The caller owns a reference
       to the returned metadata. If isStart is true the returned
       metadata writes the offset using the ItemStreamStartOffset and
       ItemBufferStartOffset elements, otherwise it uses the
       ItemStreamEndOffset and ItemBufferEndOffset elements.
     */
    virtual void GetOffsetMetaData(bool isStart,
                                   UInt64 offset,
                                   DryadMetaDataRef* dstMetaData) = 0;

protected:
    RChannelBufferData();
    virtual ~RChannelBufferData();
    DrBListEntry   m_dataListPtr;
    friend class ChannelDataBufferList;
};

class ChannelDataBufferList : public DrBList
{
public:
    static RChannelBufferData* CastOut(DrBListEntry* item);
    static DrBListEntry* CastIn(RChannelBufferData* item);
};

class RChannelBufferDataDefault : public RChannelBufferData
{
public:
    static RChannelBufferDataDefault*
        Create(DryadLockedMemoryBuffer* dataBuffer,
               UInt64 startOffset,
               RChannelBufferDefaultHandler* parent);

    /* The payload of a data buffer is a DryadLockedMemoryBuffer.
       This cannot be grown, nor can its available size be
       modified. It is derived from a DryadFixedMemoryBuffer and hence
       it is guaranteed to wrap a single contiguous memory region. It
       has the same thread-safety properties as the underlying memory
       region, e.g. it can be read concurrently from multiple threads,
       and if the threads co-operate they can safely concurrently
       write to non-overlapping regions.

       The caller must increment the reference count of the returned
       DryadLockedMemoryBuffer if it wishes to use it after the
       RChannelBufferData goes out of scope.
    */
    DryadLockedMemoryBuffer* GetData();

    /* this returns a metadata description of an offset in the buffer
       (which may include a description of the buffer, along with
       e.g. the offset's overall position in the stream), used for
       debugging and monitoring purposes. The caller owns a reference
       to the returned metadata. If isStart is true the returned
       metadata writes the offset using the ItemStreamStartOffset and
       ItemBufferStartOffset elements, otherwise it uses the
       ItemStreamEndOffset and ItemBufferEndOffset elements.
     */
    void GetOffsetMetaData(bool isStart, UInt64 offset,
                           DryadMetaDataRef* dstMetaData);

    void ProcessingComplete(RChannelBufferPrefetchInfo* prefetchCookie);

protected:
    RChannelBufferDataDefault(DryadLockedMemoryBuffer* dataBuffer,
                              UInt64 startOffset,
                              RChannelBufferDefaultHandler* parent);
    virtual ~RChannelBufferDataDefault();

    UInt64                          m_startOffset;

private:
    DryadLockedMemoryBuffer*        m_dataBuffer;
    RChannelBufferDefaultHandler*   m_parent;
};

class RChannelBufferDataSettableOffset : public RChannelBufferDataDefault
{
public:
    static RChannelBufferDataSettableOffset*
        Create(DryadLockedMemoryBuffer* dataBuffer,
               RChannelBufferDefaultHandler* parent);

    UInt64 GetOffset();
    void SetOffset(UInt64 offset);

protected:
    RChannelBufferDataSettableOffset(DryadLockedMemoryBuffer* dataBuffer,
                                 RChannelBufferDefaultHandler* parent);
    virtual ~RChannelBufferDataSettableOffset();
};

class RChannelBufferMarker : public RChannelBuffer
{
public:
    /* the caller's referenece to item is transferred to the newly
       created buffer */
    static RChannelBufferMarker*
        Create(RChannelBufferType type,
               RChannelItem* item);

    /* Get an item describing the marker event signified by this
       buffer, which can be passed to the application. The reference
       count of the item is increased before it is returned, so the
       caller must decrease it before letting the item go out of
       scope.
     */
    RChannelItem* GetItem();

protected:
    RChannelBufferMarker(RChannelBufferType type,
                         RChannelItem* item);
    virtual ~RChannelBufferMarker();

private:
    RChannelItem*                   m_item;
};

class RChannelBufferMarkerDefault : public RChannelBufferMarker
{
public:
    /* the caller's referenece to item is transferred to the newly
       created buffer */
    static RChannelBufferMarkerDefault*
        Create(RChannelBufferType type,
               RChannelItem* item,
               RChannelBufferDefaultHandler* parent);

    void ProcessingComplete(RChannelBufferPrefetchInfo* prefetchCookie);

protected:
    RChannelBufferMarkerDefault(RChannelBufferType type,
                                RChannelItem* item,
                                RChannelBufferDefaultHandler* parent);
    ~RChannelBufferMarkerDefault();

private:
    RChannelBufferDefaultHandler*   m_parent;
};