MultithreadedPatching.h

Go to the documentation of this file.

// Copyright Epic Games, Inc. All Rights Reserved.
 
#pragma once
 
#include "CoreMinimal.h"
#include "DSP/BufferVectorOperations.h"
#include "DSP/Dsp.h"
 
class FEvent;
 
namespace Audio
{
 
    
    struct FPatchOutput
    {
    public:
        SIGNALPROCESSING_API FPatchOutput(int32 InMaxCapacity, float InGain = 1.0f);
 
        SIGNALPROCESSING_API FPatchOutput();
 
        SIGNALPROCESSING_API virtual ~FPatchOutput();
        
        SIGNALPROCESSING_API int32 PopAudio(float* OutBuffer, int32 NumSamples, bool bUseLatestAudio);
 
        SIGNALPROCESSING_API int32 MixInAudio(float* OutBuffer, int32 NumSamples, bool bUseLatestAudio);
 
        SIGNALPROCESSING_API int32 GetNumSamplesAvailable() const;
 
        SIGNALPROCESSING_API bool WaitUntilNumSamplesAvailable(int32 NumSamples, uint32 TimeOutMilliseconds = MAX_uint32);
 
        SIGNALPROCESSING_API bool IsInputStale() const;
 
        friend class FPatchInput;
        friend class FPatchMixer;
        friend class FPatchSplitter;
    private:
        
        SIGNALPROCESSING_API int32 PushAudioToInternalBuffer(const float* InBuffer, int32 NumSamples);
        
        SIGNALPROCESSING_API void EmptyInternalBuffer();
 
        // Internal buffer.
        TCircularAudioBuffer<float> InternalBuffer;
 
        // For MixInAudio, audio is popped off of InternalBuffer onto here, and then mixed into OutBuffer in MixInAudio.
        FAlignedFloatBuffer MixingBuffer;
 
        // This is applied in PopAudio or MixInAudio.
        TAtomic<float> TargetGain;
        float PreviousGain;
 
        // This is used to breadcrumb the FPatchOutput when we want to delete it.
        int32 PatchID;
 
        // Counter that is incremented/decremented to allow FPatchInput to be copied around safely.
        std::atomic<int32> NumAliveInputs;
 
        // Event to pause the current thread until a given number of samples has been filled
        std::atomic<FEvent*> SamplesPushedEvent;
 
        static SIGNALPROCESSING_API TAtomic<int32> PatchIDCounter;
    };
 
    typedef TSharedPtr<FPatchOutput, ESPMode::ThreadSafe> FPatchOutputStrongPtr;
    typedef TWeakPtr<FPatchOutput, ESPMode::ThreadSafe> FPatchOutputWeakPtr;
 
    class FPatchInput
    {
    public:
        SIGNALPROCESSING_API FPatchInput(const FPatchOutputStrongPtr& InOutput);
        SIGNALPROCESSING_API FPatchInput(const FPatchInput& Other);
        SIGNALPROCESSING_API FPatchInput& operator=(const FPatchInput& Other);
        SIGNALPROCESSING_API FPatchInput& operator=(FPatchInput&& Other);
        SIGNALPROCESSING_API FPatchInput(FPatchInput&& Other);
 
        FPatchInput() = default;
 
        SIGNALPROCESSING_API ~FPatchInput();
 
        SIGNALPROCESSING_API int32 PushAudio(const float* InBuffer, int32 NumSamples);
 
        SIGNALPROCESSING_API void ClearAudio();
 
        SIGNALPROCESSING_API int32 GetNumSamplesAvailable() const;
 
        SIGNALPROCESSING_API void SetGain(float InGain);
 
        SIGNALPROCESSING_API bool IsOutputStillActive() const;
 
        SIGNALPROCESSING_API bool IsValid() const;
 
        SIGNALPROCESSING_API void Reset();
        
        friend class FPatchMixer;
        friend class FPatchSplitter;
 
    private:
        FPatchOutputStrongPtr OutputHandle;
 
        int32 PushCallsCounter = 0;
    };
 
    class FPatchMixer
    {
    public:
        SIGNALPROCESSING_API FPatchInput AddNewInput(int32 MaxLatencyInSamples, float InGain);
 
        SIGNALPROCESSING_API void AddNewInput(const FPatchInput& InPatchInput);
 
        SIGNALPROCESSING_API void RemovePatch(const FPatchInput& InPatchInput);
 
        SIGNALPROCESSING_API int32 PopAudio(float* OutBuffer, int32 OutNumSamples, bool bUseLatestAudio);
 
        SIGNALPROCESSING_API int32 Num();
 
        SIGNALPROCESSING_API int32 MaxNumberOfSamplesThatCanBePopped();
 
        SIGNALPROCESSING_API bool WaitUntilNumSamplesAvailable(int32 NumSamples, uint32 TimeOutMilliseconds = MAX_uint32);
 
        SIGNALPROCESSING_API void DisconnectAllInputs();
 
    private:
        void ConnectNewPatches();
 
        void CleanUpDisconnectedPatches();
 
        TArray<FPatchOutputStrongPtr> PendingNewInputs;
        FCriticalSection PendingNewInputsCriticalSection;
 
        TArray<int32> DisconnectedInputs;
        FCriticalSection InputDeletionCriticalSection;
 
        TArray<FPatchOutputStrongPtr> CurrentInputs;
        FCriticalSection CurrentPatchesCriticalSection;
    };
 
    class FPatchSplitter
    {
    public:
        SIGNALPROCESSING_API FPatchOutputStrongPtr AddNewPatch(int32 MaxLatencyInSamples, float InGain);
 
        SIGNALPROCESSING_API void AddNewPatch(FPatchOutputStrongPtr&& InPatchOutputStrongPtr);
        SIGNALPROCESSING_API void AddNewPatch(const FPatchOutputStrongPtr& InPatchOutputStrongPtr);
 
        SIGNALPROCESSING_API int32 PushAudio(const float* InBuffer, int32 InNumSamples);
 
        SIGNALPROCESSING_API int32 Num();
 
        SIGNALPROCESSING_API int32 MaxNumberOfSamplesThatCanBePushed();
 
    private:
        void AddPendingPatches();
 
        TArray<FPatchInput> PendingOutputs;
        FCriticalSection PendingOutputsCriticalSection;
 
        TArray<FPatchInput> ConnectedOutputs;
        FCriticalSection ConnectedOutputsCriticalSection;
    };
 
    class FPatchMixerSplitter
    {
    public:
        SIGNALPROCESSING_API FPatchOutputStrongPtr AddNewOutput(int32 MaxLatencyInSamples, float InGain);
 
        SIGNALPROCESSING_API void AddNewOutput(const FPatchOutputStrongPtr& InPatchOutputStrongPtr);
 
        SIGNALPROCESSING_API FPatchInput AddNewInput(int32 MaxLatencyInSamples, float InGain);
 
        SIGNALPROCESSING_API void AddNewInput(FPatchInput& InInput);
 
        SIGNALPROCESSING_API void RemovePatch(const FPatchInput& InInput);
 
        SIGNALPROCESSING_API void ProcessAudio();
 
    private:
        FPatchMixer Mixer;
        FPatchSplitter Splitter;
 
        FAlignedFloatBuffer IntermediateBuffer;
    };
}