plonk_TaskChannel.h

/*
 -------------------------------------------------------------------------------
 This file is part of the Plink, Plonk, Plank libraries
  by Martin Robinson
 
 http://code.google.com/p/pl-nk/
 
 Copyright University of the West of England, Bristol 2011-14
 All rights reserved.
 
 Redistribution and use in source and binary forms, with or without
 modification, are permitted provided that the following conditions are met:
 
 * Redistributions of source code must retain the above copyright
   notice, this list of conditions and the following disclaimer.
 * Redistributions in binary form must reproduce the above copyright
   notice, this list of conditions and the following disclaimer in the
   documentation and/or other materials provided with the distribution.
 * Neither the name of University of the West of England, Bristol nor 
   the names of its contributors may be used to endorse or promote products
   derived from this software without specific prior written permission.
 
 THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
 ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE 
 DISCLAIMED. IN NO EVENT SHALL UNIVERSITY OF THE WEST OF ENGLAND, BRISTOL BE 
 LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
 CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE 
 GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) 
 HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT 
 LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT 
 OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 
 
 This software makes use of third party libraries. For more information see:
 doc/license.txt included in the distribution.
 -------------------------------------------------------------------------------
 */

#ifndef PLONK_THREADEDCHANNEL_H
#define PLONK_THREADEDCHANNEL_H

#include "../channel/plonk_ChannelInternalCore.h"
#include "../plonk_GraphForwardDeclarations.h"

template<class SampleType, Interp::TypeCode InterpTypeCode> class InputTaskChannelInternal;

template<class SampleType,Interp::TypeCode InterpTypeCode>
struct ChannelData< InputTaskChannelInternal<SampleType,InterpTypeCode> >
{
    ChannelInternalCore::Data base;
    int numChannels;
    int numBuffers;
    int priority;
    bool resampleInput;
};      

//------------------------------------------------------------------------------

class TaskMessageInternal : public SmartPointer
{
public:
    TaskMessageInternal() throw()
    {
    }
    
    TaskMessageInternal (Text const& t)
    :   message (t)
    {
    }
    
    TaskMessageInternal (Text const& t, Dynamic const& d)
    :   message (t), payload (d)
    {
    }
    
    friend class TaskMessage;
    
private:
    const Text message;
    const Dynamic payload;
};

class TaskMessage : public SmartPointerContainer<TaskMessageInternal>
{
public:
    TaskMessage() throw()
    :   SmartPointerContainer<TaskMessageInternal> (static_cast<TaskMessageInternal*> (0))
    {
    }
    
    TaskMessage (Text const& t)
    :   SmartPointerContainer<TaskMessageInternal> (new TaskMessageInternal (t, Dynamic::getNull()))
    {
    }
    
    TaskMessage (Text const& t, Dynamic const& d)
    :   SmartPointerContainer<TaskMessageInternal> (new TaskMessageInternal (t, d))
    {
    }
    
    inline const Text& getMessage() const throw()    { return this->getInternal()->message; }
    inline const Dynamic& getPayload() const throw() { return this->getInternal()->payload; }
};

template<class SampleType>
class TaskBufferInternal : public SmartPointer
{
public:
    typedef NumericalArray<SampleType> Buffer;
    typedef LockFreeQueue<TaskMessage> TaskMessages;
    
    TaskBufferInternal (const int size) throw()
    :   buffer (Buffer::newClear (size))
    {
    }
    
    Buffer buffer;
    TaskMessages messages;
};

template<class SampleType>
class TaskBufferBase : public SmartPointerContainer< TaskBufferInternal<SampleType> >
{
public:
    TaskBufferBase() throw()
    :   SmartPointerContainer< TaskBufferInternal<SampleType> > (static_cast< TaskBufferInternal<SampleType>* > (0))
    {
    }
    
    TaskBufferBase (const int size) throw()
    :   SmartPointerContainer< TaskBufferInternal<SampleType> > (new TaskBufferInternal<SampleType> (size))
    {
    }
    
    static const TaskBufferBase& getNull() throw()
    {
        static TaskBufferBase null;
        return null;
    }
};


template<class SampleType, Interp::TypeCode InterpTypeCode>
class InputTaskChannelInternal
//:   public ProxyOwnerChannelInternal<SampleType, PLONK_CHANNELDATA_NAME(InputTaskChannelInternal,SampleType)>
:  public ProxyOwnerChannelInternal<SampleType, ChannelData< InputTaskChannelInternal<SampleType,InterpTypeCode> > >
{
public:
//    typedef PLONK_CHANNELDATA_NAME(InputTaskChannelInternal,SampleType) Data;
   
    typedef InputTaskChannelInternal<SampleType,InterpTypeCode>         TaskInternal;
    typedef ChannelData<TaskInternal>                                   Data;
    typedef ChannelBase<SampleType>                                     ChannelType;
    typedef ObjectArray<ChannelType>                                    ChannelArrayType;
    typedef ProxyOwnerChannelInternal<SampleType,Data>                  Internal;
    typedef UnitBase<SampleType>                                        UnitType;
    typedef InputDictionary                                             Inputs;
    typedef NumericalArray<SampleType>                                  Buffer;
    typedef TaskBufferBase<SampleType>                                  TaskBuffer;
    typedef ResampleUnit<SampleType,InterpTypeCode>                     ResampleType;

    
    //--------------------------------------------------------------------------
    
    class InputTask :  public Threading::Thread, public Channel::Receiver
    {
    public:
        typedef LockFreeQueue<TaskBuffer> BufferQueue;
        
        InputTask (InputTaskChannelInternal* o) throw()
        :   Threading::Thread (Text ("InputTaskChannelInternal::Thread[" + Text (*(LongLong*)&o) + Text ("]"))),
            owner (o),
            info (owner->getProcessInfo()),
            event (Lock::MutexLock)
        {
        }
        
        void changed (ChannelType const& source, Text const& message, Dynamic const& payload) throw()
        {
            TaskMessage tm (message, payload);
            currentTaskBuffer.getInternal()->messages.push (tm);
        }
                
        ResultCode run() throw()
        {
            Data& data = owner->getState();
            setPriority (data.priority);
            
            const int numChannels = owner->getNumChannels();
            const int numBuffers = owner->getState().numBuffers;;
                
            plonk_assert (numBuffers > 0);
            plonk_assert (owner->getBlockSize().getValue() > 0);
            
            const int bufferSize = numChannels * owner->getBlockSize().getValue();
            
            for (int i = 0; i < numBuffers; ++i)
                activeBuffers.push (TaskBuffer (bufferSize));
            
            Threading::yield();
            
            while (!getShouldExit())
            {                     
                const int blockSize = owner->getBlockSize().getValue();
                const int numFreeBuffers = freeBuffers.length();
                                
                if (numFreeBuffers > 0)
                {                    
                    UnitType& inputUnit (owner->getInputAsUnit (IOKey::Generic));
                    plonk_assert (inputUnit.channelsHaveSameBlockSize());
                    
                    currentTaskBuffer = freeBuffers.pop(); // nothing else pops so must be still available
                    Buffer& buffer = currentTaskBuffer.getInternal()->buffer;
                    buffer.setSize (blockSize * numChannels, false);
                    
                    SampleType* bufferSamples = buffer.getArray();

                    for (int channel = 0; channel < numChannels; ++channel)
                    {
                        const Buffer& inputBuffer (inputUnit.process (info, channel));                    
                        const SampleType* inputSamples = inputBuffer.getArray();
                        const int inputBufferLength = inputBuffer.length();
                        
//                        plonk_assert (buffer.length() == (numChannels * inputBufferLength));
                        
                        if (buffer.length() == (numChannels * inputBufferLength))
                        {
                            NumericalArray<SampleType>::copyData (bufferSamples, inputSamples, inputBufferLength);
                            bufferSamples += inputBufferLength;
                        }
                        else
                        {
                            // probably got deleted..
                            buffer.zero();
                            break;
                        }
                    }
                    
                    activeBuffers.push (currentTaskBuffer);
                    currentTaskBuffer = TaskBuffer::getNull();

                    plonk_assert (inputUnit.channelsHaveSameSampleRate());
                    info.offsetTimeStamp (owner->getSampleRate().getSampleDurationInTicks() * blockSize);
                }
                
                event.wait();
            }
            
            freeBuffers.clearAll();
            activeBuffers.clearAll();
            
            return 0;
        }
                
        void end() throw()
        {
            setShouldExit();
            event.signal();
    
            while (isRunning())
                Threading::sleep (0.0001); // will block!
        }
    
        inline bool pop (TaskBuffer& buffer) throw()
        {
            return activeBuffers.pop (buffer);
        }
    
        inline void push (TaskBuffer const& buffer) throw()
        {
            buffer.getInternal()->messages.clear();
            freeBuffers.push (buffer);
            event.signal();
        }
    
    private:
        InputTaskChannelInternal* owner;
        ProcessInfo& info;
        RNG rng;
        BufferQueue activeBuffers;
        BufferQueue freeBuffers;
        TaskBuffer currentTaskBuffer;
        Lock event;
    };
    
    //--------------------------------------------------------------------------
    
    InputTaskChannelInternal (Inputs const& inputs,
                              Data const& data,
                              BlockSize const& blockSize,
                              SampleRate const& sampleRate,
                              ChannelArrayType& channels) throw()
    :   Internal (numChannelsInSource (inputs), 
                  inputs, data, blockSize, sampleRate,
                  channels),
        task (this)
    {
        UnitType& inputUnit (this->getInputAsUnit (IOKey::Generic));
        inputUnit.addReceiverToChannels (&task);
        
        if (data.resampleInput)
            inputUnit = ResampleType::ar (inputUnit, 1, blockSize, sampleRate);
        
        task.start();
    }
    
    ~InputTaskChannelInternal()
    {
        task.end();
        UnitType& inputUnit (this->getInputAsUnit (IOKey::Generic));
        inputUnit.removeReceiverFromChannels (&task);
    }
            
    Text getName() const throw()
    {
        return "Task";
    }       
    
    IntArray getInputKeys() const throw()
    {
        const IntArray keys (IOKey::Generic);
        return keys;
    }    
        
    void initChannel (const int channel) throw()
    {               
        this->initProxyValue (channel, 0);
    }    
    
    void process (ProcessInfo& info, const int /*channel*/) throw()
    {                
        const int numChannels = this->getNumChannels();
        
        TaskBuffer taskBuffer;
        
        if (task.pop (taskBuffer)) 
        {
            // could be smarter in here in case the buffer size changes
            
            Buffer& buffer = taskBuffer.getInternal()->buffer;
            SampleType* bufferSamples = buffer.getArray();
            
            for (int channel = 0; channel < numChannels; ++channel)
            {
                Buffer& outputBuffer = this->getOutputBuffer (channel);
                SampleType* const outputSamples = outputBuffer.getArray();
                const int outputBufferLength = outputBuffer.length();  
                
                plonk_assert (buffer.length() == (numChannels * outputBufferLength));
                
                Buffer::copyData (outputSamples, bufferSamples, outputBufferLength);
                bufferSamples += outputBufferLength;                
            }
            
            buffer.zero();
            
            TaskMessage taskMessage;
            while (taskBuffer.getInternal()->messages.pop (taskMessage))
                this->update (taskMessage.getMessage(), taskMessage.getPayload());
            
            task.push (taskBuffer);
        }
        else
        {            
            // buffer underrun or other error
            for (int channel = 0; channel < numChannels; ++channel)
            {
                Buffer& outputBuffer = this->getOutputBuffer (channel);
                outputBuffer.zero();
            }
        }        
    }
    
    ProcessInfo& getProcessInfo() throw() { return info; }
    
private:
    InputTask task;
    ProcessInfo info; // private info for this object as we're running out of sync with everything else
    
    static inline int numChannelsInSource (Inputs const& inputs) throw()
    {
        return inputs[IOKey::Generic].asUnchecked<UnitType>().getNumChannels();
    }
    
};

//------------------------------------------------------------------------------

template<class SampleType, Interp::TypeCode InterpTypeCode>
class InputTaskUnit
{
public:    
    typedef InputTaskChannelInternal<SampleType,InterpTypeCode>     TaskInternal;
    typedef typename TaskInternal::Data                             Data;
    typedef ChannelBase<SampleType>                                 ChannelType;
    typedef ChannelInternal<SampleType,Data>                        Internal;
    typedef UnitBase<SampleType>                                    UnitType;
    typedef InputDictionary                                         Inputs;
    typedef ResampleUnit<SampleType,InterpTypeCode>                 ResampleType;
    typedef InputTaskUnit<SampleType,Interp::Lagrange3>             HQ;
    
    static inline UnitInfos getInfo() throw()
    {
        const double blockSize = (double)BlockSize::getDefault().getValue();
        const double sampleRate = SampleRate::getDefault().getValue();
        
        return UnitInfo ("InputTask", "Defer a unit's processing to a separate task, thread, process or core.",
                         
                         // output
                         ChannelCount::VariableChannelCount, 
                         IOKey::Generic,            Measure::None,      0.0,        IOLimit::None,
                         IOKey::End,
                         
                         // inputs
                         IOKey::Generic,            Measure::None,
                         IOKey::BufferCount,        Measure::Count,     16.0,       IOLimit::Minimum,   Measure::Count,             1.0,
                         IOKey::Priority,           Measure::Percent,   50.0,       IOLimit::Clipped,   Measure::Percent,           0.0, 100.0,
                         IOKey::Multiply,           Measure::Factor,    1.0,        IOLimit::None,
                         IOKey::Add,                Measure::None,      0.0,        IOLimit::None,
                         IOKey::BlockSize,          Measure::Samples,   blockSize,  IOLimit::Minimum,   Measure::Samples,           1.0,
                         IOKey::SampleRate,         Measure::Hertz,     sampleRate, IOLimit::Minimum,   Measure::Hertz,             0.0,
                         IOKey::End);
    }
    
    static UnitType ar (UnitType const& input,
                        const int numBuffers = 16,
                        const int priority = 50,
                        BlockSize const& preferredBlockSize = BlockSize::noPreference(),
                        SampleRate const& preferredSampleRate = SampleRate::noPreference()) throw()
    {                     
        BlockSize blockSize = BlockSize::decide (input.getBlockSize (0), preferredBlockSize);
        SampleRate sampleRate = SampleRate::decide (input.getSampleRate (0), preferredSampleRate);
        
        // could avoid the resample if we added the function the check if all bs/sr are the same in each channel
        
        Inputs inputs;
//        inputs.put (IOKey::Generic, ResampleType::ar (input, 1, blockSize, sampleRate));
        inputs.put (IOKey::Generic, input);
        
        Data data = { { -1.0, -1.0 }, 0, numBuffers, priority, true };
        
        return UnitType::template proxiesFromInputs<TaskInternal> (inputs, 
                                                                   data, 
                                                                   blockSize, 
                                                                   sampleRate);
    }
    
    static UnitType arNoResample (UnitType const& input,
                                  const int numBuffers = 16,
                                  const int priority = 50,
                                  BlockSize const& preferredBlockSize = BlockSize::noPreference(),
                                  SampleRate const& preferredSampleRate = SampleRate::noPreference()) throw()
    {
        BlockSize blockSize = BlockSize::decide (input.getBlockSize (0), preferredBlockSize);
        SampleRate sampleRate = SampleRate::decide (input.getSampleRate (0), preferredSampleRate);
        
        // could avoid the resample if we added the function the check if all bs/sr are the same in each channel
        
        Inputs inputs;
        inputs.put (IOKey::Generic, input);
        
        Data data = { { -1.0, -1.0 }, 0, numBuffers, priority, false };
        
        return UnitType::template proxiesFromInputs<TaskInternal> (inputs,
                                                                   data,
                                                                   blockSize,
                                                                   sampleRate);
    }

};

typedef InputTaskUnit<PLONK_TYPE_DEFAULT> InputTask;

#endif // PLONK_THREADEDCHANNEL_H