cvd/readaheadvideobuffer.h

/*                       
    This file is part of the CVD Library.

    Copyright (C) 2005 The Authors

    This library is free software; you can redistribute it and/or
    modify it under the terms of the GNU Lesser General Public
    License as published by the Free Software Foundation; either
    version 2.1 of the License, or (at your option) any later version.

    This library 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
    Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public
    License along with this library; if not, write to the Free Software
    Foundation, Inc., 
    51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
*/

#ifndef CVD_READAHEADVIDEOBUFFER_H
#define CVD_READAHEADVIDEOBUFFER_H

#include <cvd/config.h>
#include <cvd/videobuffer.h>

#ifdef CVD_HAVE_PTHREAD
#include <deque>
#include <cvd/thread.h>
#include <cvd/synchronized.h>
#include <cvd/eventobject.h>
#endif

namespace CVD {
    #ifndef CVD_HAVE_PTHREAD
    #ifndef _WIN32
    #warning ReadAheadVideoBuffer will not do any read-ahead because threads are not supported in this build
    #endif
    template <class T> 
    class ReadAheadVideoBuffer : public VideoBuffer<T>
    {
    private:
    VideoBuffer<T>& vbuffer;
    public:
    virtual ~ReadAheadVideoBuffer() {}
    ReadAheadVideoBuffer(VideoBuffer<T>& vb, size_t maxReadAhead=10) 
    : VideoBuffer<T>(vb.type()),vbuffer(vb) {}
    ImageRef size() { return vbuffer.size(); }
    VideoFrame<T>* get_frame() { return vbuffer.get_frame(); }
    void put_frame(VideoFrame<T>* f) { vbuffer.put_frame(f); }
    bool frame_pending() { return vbuffer.frame_pending(); }
    double frame_rate() { return vbuffer.frame_rate(); }
    void seek_to(double t){ vbuffer.seek_to(t); }
    };
#else    






    template <class T> 
    class ReadAheadVideoBuffer : public VideoBuffer<T>, public Runnable
    {
    private:
    VideoBuffer<T>& vbuffer;
    size_t maxRA;
    std::deque<VideoFrame<T>*> q, putq;
    Synchronized vblock;
    EventObject qevent;
    Thread thread;


    static VideoBufferType::Type type_update(VideoBufferType::Type t)
    {
        if(t== VideoBufferType::NotLive)
            return t; 
        else
            return VideoBufferType::Flushable;
    }

    public:
    virtual ~ReadAheadVideoBuffer() {
        qevent.lock();
        thread.stop();
        qevent.trigger();
        qevent.unlock();
        thread.join();      
    }
    ReadAheadVideoBuffer(VideoBuffer<T>& vb, size_t maxReadAhead=10) 
    : VideoBuffer<T>(type_update(vb.type())), vbuffer(vb), maxRA(maxReadAhead) {
        thread.start(this);     
    }

    void run() {
        while (!thread.shouldStop()) 
        {
        // lock queue
        { Lock l(qevent);
            for (size_t i=0; i<putq.size(); i++)
            vbuffer.put_frame(putq[i]);
            putq.resize(0);
            while (q.size() >= maxRA) {         
            qevent.wait();
            if (thread.shouldStop())
                return;
            }
        }
        // lock videobuffer
        { Lock l(vblock);
            VideoFrame<T>* frame = vbuffer.get_frame();
            // lock queue
            { Lock l(qevent);
            q.push_back(frame);
            qevent.trigger();
            }
        }
        }
    }
    ImageRef size() { return vbuffer.size(); }
    VideoFrame<T>* get_frame() {
        VideoFrame<T>* frame = 0;
        Lock l(qevent);
        while (q.empty())
        qevent.wait();
        frame = q.front();
        q.pop_front();
        qevent.trigger(); 
        return frame;
    }
    void put_frame(VideoFrame<T>* f) { Lock l(qevent); putq.push_back(f); }
    bool frame_pending() {
        return !q.empty();
    }
    double frame_rate() { return vbuffer.frame_rate(); }
    void seek_to(double t){
        Lock vbl(vblock);
        Lock ql(qevent);
        for (size_t i=0; i<q.size(); i++)
        putq.push_back(q[i]);
        q.resize(0);
        qevent.triggerAll();
        vbuffer.seek_to(t);
    }
    };
    #endif
}

#endif