engines/common/DiskThreadBase.h

/***************************************************************************
 *                                                                         *
 *   LinuxSampler - modular, streaming capable sampler                     *
 *                                                                         *
 *   Copyright (C) 2003, 2004 by Benno Senoner and Christian Schoenebeck   *
 *   Copyright (C) 2005 - 2008 Christian Schoenebeck                       *
 *   Copyright (C) 2009 - 2011 Christian Schoenebeck and Grigor Iliev      *
 *                                                                         *
 *   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.                                   *
 *                                                                         *
 *   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 __LS_DISKTHREADBASE_H__
#define __LS_DISKTHREADBASE_H__

#include "StreamBase.h"
#include "../EngineChannel.h"
#include "../InstrumentManagerBase.h"

#include "../../common/global_private.h"

#include "../../common/Thread.h"
#include "../../common/RingBuffer.h"
#include "../../common/atomic.h"

namespace LinuxSampler {

    int CompareStreamWriteSpace(const void* A, const void* B);

    /** @brief Disk Reader Thread
     *
     * The disk reader thread is responsible for periodically refilling
     * disk streams in parallel to the audio thread's rendering process.
     *
     * There is also a function for releasing parts of instruments not
     * in use anymore (as this is not real time safe, the audio thread
     * cannot do it directly).
     */
    template <class R /* Resource */, class IM /* Instrument Manager */>
    class DiskThreadBase : public Thread {
        private:
            // Private Types
            struct create_command_t {
                Stream::OrderID_t    OrderID;
                Stream::Handle       hStream;
                Stream::reference_t* pStreamRef;
                R*                   pRegion;
                unsigned long        SampleOffset;
                bool                 DoLoop;
            };
            struct delete_command_t {
                Stream*           pStream;
                Stream::Handle    hStream;
                Stream::OrderID_t OrderID;
                bool              bNotify;
            };
            struct program_change_command_t {
                uint32_t Program;
                EngineChannel* pEngineChannel;
            };
            // Attributes
            bool                           IsIdle;
            uint                           Streams;
            RingBuffer<create_command_t,false>* CreationQueue;                      ///< Contains commands to create streams
            RingBuffer<delete_command_t,false>* DeletionQueue;                      ///< Contains commands to delete streams
            RingBuffer<delete_command_t,false>* GhostQueue;                         ///< Contains handles to streams that are not used anymore and weren't deletable immediately
            RingBuffer<Stream::Handle,false>    DeletionNotificationQueue;          ///< In case the original sender requested a notification for its stream deletion order, this queue will receive the handle of the respective stream once actually be deleted by the disk thread.
            RingBuffer<R*,false>*               DeleteRegionQueue;          ///< Contains dimension regions that are not used anymore and should be handed back to the instrument resource manager
            RingBuffer<program_change_command_t,false> ProgramChangeQueue;          ///< Contains requests for MIDI program change
            unsigned int                   RefillStreamsPerRun;                    ///< How many streams should be refilled in each loop run
            Stream**                       pStreams; ///< Contains all disk streams (whether used or unused)
            Stream**                       pCreatedStreams; ///< This is where the voice (audio thread) picks up it's meanwhile hopefully created disk stream.
            static Stream*                 SLOT_RESERVED;                          ///< This value is used to mark an entry in pCreatedStreams[] as reserved.

            // Methods

            void CreateStream(create_command_t& Command) {
                // search for unused stream
                Stream* newstream = NULL;
                for (int i = Streams - 1; i >= 0; i--) {
                    if (pStreams[i]->GetState() == Stream::state_unused) {
                        newstream = pStreams[i];
                        break;
                    }
                }
                if (!newstream) {
                    std::cerr << "No unused stream found (OrderID:" << Command.OrderID;
                    std::cerr << ") - report if this happens, this is a bug!\n" << std::flush;
                    return;
                }
                LaunchStream(newstream, Command.hStream, Command.pStreamRef, Command.pRegion, Command.SampleOffset, Command.DoLoop);
                dmsg(4,("new Stream launched by disk thread (OrderID:%d,StreamHandle:%d)\n", Command.OrderID, Command.hStream));
                if (pCreatedStreams[Command.OrderID] != SLOT_RESERVED) {
                    std::cerr << "DiskThread: Slot " << Command.OrderID << " already occupied! Please report this!\n" << std::flush;
                    newstream->Kill();
                    return;
                }
                pCreatedStreams[Command.OrderID] = newstream;
            }

            void DeleteStream(delete_command_t& Command) {
                if (Command.pStream) {
                    Command.pStream->Kill();
                    if (Command.bNotify) DeletionNotificationQueue.push(&Command.hStream);
                }
                else { // the stream wasn't created by disk thread or picked up by audio thread yet

                    // if stream was created but not picked up yet
                    Stream* pStream = pCreatedStreams[Command.OrderID];
                    if (pStream && pStream != SLOT_RESERVED) {
                        pStream->Kill();
                        pCreatedStreams[Command.OrderID] = NULL; // free slot for new order
                        // if original sender requested a notification, let him know now
                        if (Command.bNotify)
                            DeletionNotificationQueue.push(&Command.hStream);
                        return;
                    }

                    // the stream was not created yet
                    if (GhostQueue->write_space() > 0) {
                        GhostQueue->push(&Command);
                    } else { // error, queue full
                        if (Command.bNotify) {
                            dmsg(1,("DiskThread: GhostQueue full! (might lead to dead lock with instrument editor!)\n"));
                        } else {
                            dmsg(1,("DiskThread: GhostQueue full!\n"));
                        }
                    }
                }
            }

            void RefillStreams() {
                // sort the streams by most empty stream
                qsort(pStreams, Streams, sizeof(Stream*), CompareStreamWriteSpace);

                // refill the most empty streams
                for (uint i = 0; i < RefillStreamsPerRun; i++) {
                    if (pStreams[i]->GetState() == Stream::state_active) {

                        //float filledpercentage = (float) pStreams[i]->GetReadSpace() / 131072.0 * 100.0;
                        //dmsg(("\nbuffer fill: %.1f%\n", filledpercentage));

                        int writespace = pStreams[i]->GetWriteSpaceToEnd();
                        if (writespace == 0) break;

                        int capped_writespace = writespace;
                        // if there is too much buffer space available then cut the read/write
                        // size to CONFIG_STREAM_MAX_REFILL_SIZE which is by default 65536 samples = 256KBytes
                        if (writespace > CONFIG_STREAM_MAX_REFILL_SIZE) capped_writespace = CONFIG_STREAM_MAX_REFILL_SIZE;

                        // adjust the amount to read in order to ensure that the buffer wraps correctly
                        int read_amount = pStreams[i]->AdjustWriteSpaceToAvoidBoundary(writespace, capped_writespace);
                        // if we wasn't able to refill one of the stream buffers by more than
                        // CONFIG_STREAM_MIN_REFILL_SIZE we'll send the disk thread to sleep later
                        if (pStreams[i]->ReadAhead(read_amount) > CONFIG_STREAM_MIN_REFILL_SIZE) this->IsIdle = false;
                    }
                }
            }

            Stream::Handle CreateHandle() {
                static uint32_t counter = 0;
                if (counter == 0xffffffff) counter = 1; // we use '0' as 'invalid handle' only, so we skip 0
                else                       counter++;
                return counter;
            }

            Stream::OrderID_t CreateOrderID() {
                static Stream::OrderID_t counter(0);
                for (int i = 0; i < Streams; i++) {
                    if (counter == Streams) counter = 1; // we use '0' as 'invalid order' only, so we skip 0
                    else                    counter++;
                    if (!pCreatedStreams[counter]) {
                        pCreatedStreams[counter] = SLOT_RESERVED; // mark this slot as reserved
                        return counter;                           // found empty slot
                    }
                }
                return 0; // no free slot
            }

            atomic_t ActiveStreamCount;
        public:
            // Methods
            DiskThreadBase(int MaxStreams, uint BufferWrapElements, IM* pInstruments) :
                Thread(true, false, 1, -2),
                pInstruments(pInstruments),
                DeletionNotificationQueue(4*MaxStreams),
                ProgramChangeQueue(512)
            {
                CreationQueue       = new RingBuffer<create_command_t,false>(4*MaxStreams);
                DeletionQueue       = new RingBuffer<delete_command_t,false>(4*MaxStreams);
                GhostQueue          = new RingBuffer<delete_command_t,false>(MaxStreams);
                DeleteRegionQueue   = new RingBuffer<R*,false>(4*MaxStreams);
                pStreams            = new Stream*[MaxStreams];
                pCreatedStreams     = new Stream*[MaxStreams + 1];
                Streams             = MaxStreams;
                RefillStreamsPerRun = CONFIG_REFILL_STREAMS_PER_RUN;

                for (int i = 1; i <= MaxStreams; i++) {
                    pCreatedStreams[i] = NULL;
                }
                ActiveStreamCountMax = 0;
            }

            virtual ~DiskThreadBase() {
                for (int i = 0; i < Streams; i++) {
                    if (pStreams[i]) delete pStreams[i];
                }
                if (CreationQueue) delete CreationQueue;
                if (DeletionQueue) delete DeletionQueue;
                if (GhostQueue)    delete GhostQueue;
                if (DeleteRegionQueue) delete DeleteRegionQueue;
                if (pStreams)        delete[] pStreams;
                if (pCreatedStreams) delete[] pCreatedStreams;
            }


        // #########################################################################
        // # Foreign Thread Section
        // #         (following code intended to be interface for audio thread)

            /**
             * Suspend disk thread, kill all active streams, clear all queues and the
             * pickup array and reset all streams. Call this method to bring everything
             * in the disk thread to day one. If the disk thread was running, it will be
             * respawned right after everything was reset.
             */
            void Reset() {
                bool running = this->IsRunning();
                if (running) this->StopThread();
                for (int i = 0; i < Streams; i++) {
                    pStreams[i]->Kill();
                }
                for (int i = 1; i <= Streams; i++) {
                    pCreatedStreams[i] = NULL;
                }
                GhostQueue->init();
                CreationQueue->init();
                DeletionQueue->init();
                DeletionNotificationQueue.init();

                // make sure that all DimensionRegions are released
                while (DeleteRegionQueue->read_space() > 0) {
                    R* pRgn;
                    DeleteRegionQueue->pop(&pRgn);
                    pInstruments->HandBackRegion(pRgn);
                }
                DeleteRegionQueue->init();
                SetActiveStreamCount(0);
                ActiveStreamCountMax = 0;
                if (running) this->StartThread(); // start thread only if it was running before
            }

            String GetBufferFillBytes() {
                bool activestreams = false;
                std::stringstream ss;
                for (uint i = 0; i < this->Streams; i++) {
                    if (pStreams[i]->GetState() == Stream::state_unused) continue;
                    uint bufferfill = pStreams[i]->GetReadSpace() * sizeof(sample_t);
                    uint streamid   = (uint) pStreams[i]->GetHandle();
                    if (!streamid) continue;

                    if (activestreams) ss << ",[" << streamid << ']' << bufferfill;
                    else {
                        ss << '[' << streamid << ']' << bufferfill;
                        activestreams = true;
                    }
                }
                return ss.str();
            }

            String GetBufferFillPercentage() {
                bool activestreams = false;
                std::stringstream ss;
                for (uint i = 0; i < this->Streams; i++) {
                    if (pStreams[i]->GetState() == Stream::state_unused) continue;
                    uint bufferfill = (uint) ((float) pStreams[i]->GetReadSpace() / (float) CONFIG_STREAM_BUFFER_SIZE * 100);
                    uint streamid   = (uint) pStreams[i]->GetHandle();
                    if (!streamid) continue;

                    if (activestreams) ss << ",[" << streamid << ']' << bufferfill << '%';
                    else {
                        ss << '[' << streamid << ']' << bufferfill;
                        activestreams = true;
                    }
                }
                return ss.str();
            }

            /**
             * Returns -1 if command queue or pickup pool is full, 0 on success (will be
             * called by audio thread within the voice class).
             */
            int OrderNewStream(Stream::reference_t* pStreamRef, R* pRegion, unsigned long SampleOffset, bool DoLoop) {
                dmsg(4,("Disk Thread: new stream ordered\n"));
                if (CreationQueue->write_space() < 1) {
                    dmsg(1,("DiskThread: Order queue full!\n"));
                    return -1;
                }

                const Stream::OrderID_t newOrder = CreateOrderID();
                if (!newOrder) {
                        dmsg(1,("DiskThread: there was no free slot\n"));
                        return -1; // there was no free slot
                }

                pStreamRef->State   = Stream::state_active;
                pStreamRef->OrderID = newOrder;
                pStreamRef->hStream = CreateHandle();
                pStreamRef->pStream = NULL; // a stream has to be activated by the disk thread first

                create_command_t cmd;
                cmd.OrderID      = pStreamRef->OrderID;
                cmd.hStream      = pStreamRef->hStream;
                cmd.pStreamRef   = pStreamRef;
                cmd.pRegion      = pRegion;
                cmd.SampleOffset = SampleOffset;
                cmd.DoLoop       = DoLoop;

                CreationQueue->push(&cmd);
                return 0;
            }

            /**
             * Request the disk thread to delete the given disk stream. This method
             * will return immediately, thus it won't block until the respective voice
             * was actually deleted. (Called by audio thread within the Voice class).
             *
             * @param pStreamRef           - stream that shall be deleted
             * @param bRequestNotification - set to true in case you want to receive a
             *                               notification once the stream has actually
             *                               been deleted
             * @returns 0 on success, -1 if command queue is full
             * @see AskForDeletedStream()
             */
            int OrderDeletionOfStream(Stream::reference_t* pStreamRef, bool bRequestNotification = false) {
                dmsg(4,("Disk Thread: stream deletion ordered\n"));
                if (DeletionQueue->write_space() < 1) {
                    dmsg(1,("DiskThread: Deletion queue full!\n"));
                    return -1;
                }

                delete_command_t cmd;
                cmd.pStream = pStreamRef->pStream;
                cmd.hStream = pStreamRef->hStream;
                cmd.OrderID = pStreamRef->OrderID;
                cmd.bNotify = bRequestNotification;

                DeletionQueue->push(&cmd);
                return 0;
            }

            /**
             * Tell the disk thread to release a dimension region that belongs
             * to an instrument which isn't loaded anymore. The disk thread
             * will hand back the dimension region to the instrument resource
             * manager. (OrderDeletionOfDimreg is called from the audio thread
             * when a voice dies.)
             */
            int OrderDeletionOfRegion(R* pReg) {
                dmsg(4,("Disk Thread: dimreg deletion ordered\n"));
                if (DeleteRegionQueue->write_space() < 1) {
                    dmsg(1,("DiskThread: DeleteRegion queue full!\n"));
                    return -1;
                }
                DeleteRegionQueue->push(&pReg);
                return 0;
            }

            /**
             * Tell the disk thread to do a program change on the specified
             * EngineChannel.
             */
            int OrderProgramChange(uint32_t Program, EngineChannel* pEngineChannel) {
                program_change_command_t cmd;
                cmd.Program = Program;
                cmd.pEngineChannel = pEngineChannel;

                dmsg(4,("Disk Thread: program change ordered\n"));
                if (ProgramChangeQueue.write_space() < 1) {
                    dmsg(1,("DiskThread: ProgramChange queue full!\n"));
                    return -1;
                }
                ProgramChangeQueue.push(&cmd);
                return 0;
            }

            /**
             * Returns the pointer to a disk stream if the ordered disk stream
             * represented by the \a StreamOrderID was already activated by the disk
             * thread, returns NULL otherwise. If the call was successful, thus if it
             * returned a valid stream pointer, the caller has to the store that pointer
             * by himself, because it's not possible to call this method again with the
             * same used order ID; this method is just intended for picking up an ordered
             * disk stream. This method will usually be called by the voice class (within
             * the audio thread).
             *
             * @param StreamOrderID - ID previously returned by OrderNewStream()
             * @returns               pointer to created stream object, NULL otherwise
             */
            Stream* AskForCreatedStream(Stream::OrderID_t StreamOrderID) {
                dmsg(4,("Disk Thread: been asked if stream already created, OrderID=%x ", StreamOrderID));
                Stream* pStream = pCreatedStreams[StreamOrderID];
                if (pStream && pStream != SLOT_RESERVED) {
                    dmsg(4,("(yes created)\n"));
                    pCreatedStreams[StreamOrderID] = NULL; // free the slot for a new order
                    return pStream;
                }
                dmsg(4,("(no not yet created)\n"));
                return NULL;
            }

            /**
             * In case the original sender requested a notification with his stream
             * deletion order, he can use this method to poll if the respective stream
             * has actually been deleted.
             *
             * @returns handle / identifier of the deleted stream, or
             *          Stream::INVALID_HANDLE if no notification present
             */
            Stream::Handle AskForDeletedStream() {
                if (DeletionNotificationQueue.read_space()) {
                    Stream::Handle hStream;
                    DeletionNotificationQueue.pop(&hStream);
                    return hStream;
                } else return Stream::INVALID_HANDLE; // no notification received yet
            }

            // the number of streams currently in usage
            // printed on the console the main thread (along with the active voice count)
            uint GetActiveStreamCount() { return atomic_read(&ActiveStreamCount); }
            void SetActiveStreamCount(uint Streams) { atomic_set(&ActiveStreamCount, Streams); }
            int ActiveStreamCountMax;

        protected:
            IM* pInstruments;   ///< The instrument resource manager of the engine that is using this disk thread. Used by the dimension region deletion feature.

        // #########################################################################
        // # Disk Thread Only Section
        // #         (following code should only be executed by the disk thread)

            // Implementation of virtual method from class Thread
            int Main() {
                dmsg(3,("Disk thread running\n"));
                while (true) {
                    #if !defined(WIN32)
                    pthread_testcancel(); // mandatory for OSX
                    #endif
                    #if CONFIG_PTHREAD_TESTCANCEL
                    TestCancel();
                    #endif
                    IsIdle = true; // will be set to false if a stream got filled

                    // if there are ghost streams, delete them
                    for (int i = 0; i < GhostQueue->read_space(); i++) { //FIXME: unefficient
                        delete_command_t ghostStream;
                        GhostQueue->pop(&ghostStream);
                        bool found = false;
                        for (int i = 0; i < this->Streams; i++) {
                            if (pStreams[i]->GetHandle() == ghostStream.hStream) {
                                pStreams[i]->Kill();
                                found = true;
                                // if original sender requested a notification, let him know now
                                if (ghostStream.bNotify)
                                    DeletionNotificationQueue.push(&ghostStream.hStream);
                                break;
                            }
                        }
                        if (!found) GhostQueue->push(&ghostStream); // put ghost stream handle back to the queue
                    }

                    // if there are creation commands, create new streams
                    while (Stream::UnusedStreams > 0 && CreationQueue->read_space() > 0) {
                        create_command_t command;
                        CreationQueue->pop(&command);
                        CreateStream(command);
                    }

                    // if there are deletion commands, delete those streams
                    while (Stream::UnusedStreams < Stream::TotalStreams && DeletionQueue->read_space() > 0) {
                        delete_command_t command;
                        DeletionQueue->pop(&command);
                        DeleteStream(command);
                    }

                    // release DimensionRegions that belong to instruments
                    // that are no longer loaded
                    while (DeleteRegionQueue->read_space() > 0) {
                        R* pRgn;
                        DeleteRegionQueue->pop(&pRgn);
                        pInstruments->HandBackRegion(pRgn);
                    }

                    // perform MIDI program change commands
                    while (ProgramChangeQueue.read_space() > 0) {
                        program_change_command_t cmd;
                        ProgramChangeQueue.pop(&cmd);
                        cmd.pEngineChannel->ExecuteProgramChange(cmd.Program);
                    }
                    RefillStreams(); // refill the most empty streams

                    // if nothing was done during this iteration (eg no streambuffer
                    // filled with data) then sleep for 30ms
                    if (IsIdle) usleep(30000);

                    int streamsInUsage = 0;
                    for (int i = Streams - 1; i >= 0; i--) {
                        if (pStreams[i]->GetState() != Stream::state_unused) streamsInUsage++;
                    }
                    SetActiveStreamCount(streamsInUsage);
                    if (streamsInUsage > ActiveStreamCountMax) ActiveStreamCountMax = streamsInUsage;
                }

                return EXIT_FAILURE;
            }

            virtual Stream* CreateStream(long BufferSize, uint BufferWrapElements) = 0;

            void CreateAllStreams(int MaxStreams, uint BufferWrapElements) {
                for (int i = 0; i < MaxStreams; i++) {
                    pStreams[i] = CreateStream(CONFIG_STREAM_BUFFER_SIZE, BufferWrapElements);
                }
            }

            virtual void LaunchStream (
                Stream*               pStream,
                Stream::Handle        hStream,
                Stream::reference_t*  pExportReference,
                R*                    pRgn,
                unsigned long         SampleOffset,
                bool                  DoLoop
            ) = 0;

            friend class Stream;
    };
} // namespace LinuxSampler

#endif // __LS_DISKTHREADBASE_H__
1	/***************************************************************************
2	* *
3	* LinuxSampler - modular, streaming capable sampler *
4	* *
5	* Copyright (C) 2003, 2004 by Benno Senoner and Christian Schoenebeck *
6	* Copyright (C) 2005 - 2008 Christian Schoenebeck *
7	* Copyright (C) 2009 - 2011 Christian Schoenebeck and Grigor Iliev *
8	* *
9	* This program is free software; you can redistribute it and/or modify *
10	* it under the terms of the GNU General Public License as published by *
11	* the Free Software Foundation; either version 2 of the License, or *
12	* (at your option) any later version. *
13	* *
14	* This program is distributed in the hope that it will be useful, *
15	* but WITHOUT ANY WARRANTY; without even the implied warranty of *
16	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
17	* GNU General Public License for more details. *
18	* *
19	* You should have received a copy of the GNU General Public License *
20	* along with this program; if not, write to the Free Software *
21	* Foundation, Inc., 59 Temple Place, Suite 330, Boston, *
22	* MA 02111-1307 USA *
23	***************************************************************************/
24
25	#ifndef __LS_DISKTHREADBASE_H__
26	#define __LS_DISKTHREADBASE_H__
27
28	#include "StreamBase.h"
29	#include "../EngineChannel.h"
30	#include "../InstrumentManagerBase.h"
31
32	#include "../../common/global_private.h"
33
34	#include "../../common/Thread.h"
35	#include "../../common/RingBuffer.h"
36	#include "../../common/atomic.h"
37
38	namespace LinuxSampler {
39
40	int CompareStreamWriteSpace(const void* A, const void* B);
41
42	/** @brief Disk Reader Thread
43	*
44	* The disk reader thread is responsible for periodically refilling
45	* disk streams in parallel to the audio thread's rendering process.
46	*
47	* There is also a function for releasing parts of instruments not
48	* in use anymore (as this is not real time safe, the audio thread
49	* cannot do it directly).
50	*/
51	template <class R /* Resource /, class IM / Instrument Manager */>
52	class DiskThreadBase : public Thread {
53	private:
54	// Private Types
55	struct create_command_t {
56	Stream::OrderID_t OrderID;
57	Stream::Handle hStream;
58	Stream::reference_t* pStreamRef;
59	R* pRegion;
60	unsigned long SampleOffset;
61	bool DoLoop;
62	};
63	struct delete_command_t {
64	Stream* pStream;
65	Stream::Handle hStream;
66	Stream::OrderID_t OrderID;
67	bool bNotify;
68	};
69	struct program_change_command_t {
70	uint32_t Program;
71	EngineChannel* pEngineChannel;
72	};
73	// Attributes
74	bool IsIdle;
75	uint Streams;
76	RingBuffer<create_command_t,false>* CreationQueue; ///< Contains commands to create streams
77	RingBuffer<delete_command_t,false>* DeletionQueue; ///< Contains commands to delete streams
78	RingBuffer<delete_command_t,false>* GhostQueue; ///< Contains handles to streams that are not used anymore and weren't deletable immediately
79	RingBuffer<Stream::Handle,false> DeletionNotificationQueue; ///< In case the original sender requested a notification for its stream deletion order, this queue will receive the handle of the respective stream once actually be deleted by the disk thread.
80	RingBuffer<R,false> DeleteRegionQueue; ///< Contains dimension regions that are not used anymore and should be handed back to the instrument resource manager
81	RingBuffer<program_change_command_t,false> ProgramChangeQueue; ///< Contains requests for MIDI program change
82	unsigned int RefillStreamsPerRun; ///< How many streams should be refilled in each loop run
83	Stream** pStreams; ///< Contains all disk streams (whether used or unused)
84	Stream** pCreatedStreams; ///< This is where the voice (audio thread) picks up it's meanwhile hopefully created disk stream.
85	static Stream* SLOT_RESERVED; ///< This value is used to mark an entry in pCreatedStreams[] as reserved.
86
87	// Methods
88
89	void CreateStream(create_command_t& Command) {
90	// search for unused stream
91	Stream* newstream = NULL;
92	for (int i = Streams - 1; i >= 0; i--) {
93	if (pStreams[i]->GetState() == Stream::state_unused) {
94	newstream = pStreams[i];
95	break;
96	}
97	}
98	if (!newstream) {
99	std::cerr << "No unused stream found (OrderID:" << Command.OrderID;
100	std::cerr << ") - report if this happens, this is a bug!\n" << std::flush;
101	return;
102	}
103	LaunchStream(newstream, Command.hStream, Command.pStreamRef, Command.pRegion, Command.SampleOffset, Command.DoLoop);
104	dmsg(4,("new Stream launched by disk thread (OrderID:%d,StreamHandle:%d)\n", Command.OrderID, Command.hStream));
105	if (pCreatedStreams[Command.OrderID] != SLOT_RESERVED) {
106	std::cerr << "DiskThread: Slot " << Command.OrderID << " already occupied! Please report this!\n" << std::flush;
107	newstream->Kill();
108	return;
109	}
110	pCreatedStreams[Command.OrderID] = newstream;
111	}
112
113	void DeleteStream(delete_command_t& Command) {
114	if (Command.pStream) {
115	Command.pStream->Kill();
116	if (Command.bNotify) DeletionNotificationQueue.push(&Command.hStream);
117	}
118	else { // the stream wasn't created by disk thread or picked up by audio thread yet
119
120	// if stream was created but not picked up yet
121	Stream* pStream = pCreatedStreams[Command.OrderID];
122	if (pStream && pStream != SLOT_RESERVED) {
123	pStream->Kill();
124	pCreatedStreams[Command.OrderID] = NULL; // free slot for new order
125	// if original sender requested a notification, let him know now
126	if (Command.bNotify)
127	DeletionNotificationQueue.push(&Command.hStream);
128	return;
129	}
130
131	// the stream was not created yet
132	if (GhostQueue->write_space() > 0) {
133	GhostQueue->push(&Command);
134	} else { // error, queue full
135	if (Command.bNotify) {
136	dmsg(1,("DiskThread: GhostQueue full! (might lead to dead lock with instrument editor!)\n"));
137	} else {
138	dmsg(1,("DiskThread: GhostQueue full!\n"));
139	}
140	}
141	}
142	}
143
144	void RefillStreams() {
145	// sort the streams by most empty stream
146	qsort(pStreams, Streams, sizeof(Stream*), CompareStreamWriteSpace);
147
148	// refill the most empty streams
149	for (uint i = 0; i < RefillStreamsPerRun; i++) {
150	if (pStreams[i]->GetState() == Stream::state_active) {
151
152	//float filledpercentage = (float) pStreams[i]->GetReadSpace() / 131072.0 * 100.0;
153	//dmsg(("\nbuffer fill: %.1f%\n", filledpercentage));
154
155	int writespace = pStreams[i]->GetWriteSpaceToEnd();
156	if (writespace == 0) break;
157
158	int capped_writespace = writespace;
159	// if there is too much buffer space available then cut the read/write
160	// size to CONFIG_STREAM_MAX_REFILL_SIZE which is by default 65536 samples = 256KBytes
161	if (writespace > CONFIG_STREAM_MAX_REFILL_SIZE) capped_writespace = CONFIG_STREAM_MAX_REFILL_SIZE;
162
163	// adjust the amount to read in order to ensure that the buffer wraps correctly
164	int read_amount = pStreams[i]->AdjustWriteSpaceToAvoidBoundary(writespace, capped_writespace);
165	// if we wasn't able to refill one of the stream buffers by more than
166	// CONFIG_STREAM_MIN_REFILL_SIZE we'll send the disk thread to sleep later
167	if (pStreams[i]->ReadAhead(read_amount) > CONFIG_STREAM_MIN_REFILL_SIZE) this->IsIdle = false;
168	}
169	}
170	}
171
172	Stream::Handle CreateHandle() {
173	static uint32_t counter = 0;
174	if (counter == 0xffffffff) counter = 1; // we use '0' as 'invalid handle' only, so we skip 0
175	else counter++;
176	return counter;
177	}
178
179	Stream::OrderID_t CreateOrderID() {
180	static Stream::OrderID_t counter(0);
181	for (int i = 0; i < Streams; i++) {
182	if (counter == Streams) counter = 1; // we use '0' as 'invalid order' only, so we skip 0
183	else counter++;
184	if (!pCreatedStreams[counter]) {
185	pCreatedStreams[counter] = SLOT_RESERVED; // mark this slot as reserved
186	return counter; // found empty slot
187	}
188	}
189	return 0; // no free slot
190	}
191
192	atomic_t ActiveStreamCount;
193	public:
194	// Methods
195	DiskThreadBase(int MaxStreams, uint BufferWrapElements, IM* pInstruments) :
196	Thread(true, false, 1, -2),
197	pInstruments(pInstruments),
198	DeletionNotificationQueue(4*MaxStreams),
199	ProgramChangeQueue(512)
200	{
201	CreationQueue = new RingBuffer<create_command_t,false>(4*MaxStreams);
202	DeletionQueue = new RingBuffer<delete_command_t,false>(4*MaxStreams);
203	GhostQueue = new RingBuffer<delete_command_t,false>(MaxStreams);
204	DeleteRegionQueue = new RingBuffer<R,false>(4MaxStreams);
205	pStreams = new Stream*[MaxStreams];
206	pCreatedStreams = new Stream*[MaxStreams + 1];
207	Streams = MaxStreams;
208	RefillStreamsPerRun = CONFIG_REFILL_STREAMS_PER_RUN;
209
210	for (int i = 1; i <= MaxStreams; i++) {
211	pCreatedStreams[i] = NULL;
212	}
213	ActiveStreamCountMax = 0;
214	}
215
216	virtual ~DiskThreadBase() {
217	for (int i = 0; i < Streams; i++) {
218	if (pStreams[i]) delete pStreams[i];
219	}
220	if (CreationQueue) delete CreationQueue;
221	if (DeletionQueue) delete DeletionQueue;
222	if (GhostQueue) delete GhostQueue;
223	if (DeleteRegionQueue) delete DeleteRegionQueue;
224	if (pStreams) delete[] pStreams;
225	if (pCreatedStreams) delete[] pCreatedStreams;
226	}
227
228
229	// #########################################################################
230	// # Foreign Thread Section
231	// # (following code intended to be interface for audio thread)
232
233	/**
234	* Suspend disk thread, kill all active streams, clear all queues and the
235	* pickup array and reset all streams. Call this method to bring everything
236	* in the disk thread to day one. If the disk thread was running, it will be
237	* respawned right after everything was reset.
238	*/
239	void Reset() {
240	bool running = this->IsRunning();
241	if (running) this->StopThread();
242	for (int i = 0; i < Streams; i++) {
243	pStreams[i]->Kill();
244	}
245	for (int i = 1; i <= Streams; i++) {
246	pCreatedStreams[i] = NULL;
247	}
248	GhostQueue->init();
249	CreationQueue->init();
250	DeletionQueue->init();
251	DeletionNotificationQueue.init();
252
253	// make sure that all DimensionRegions are released
254	while (DeleteRegionQueue->read_space() > 0) {
255	R* pRgn;
256	DeleteRegionQueue->pop(&pRgn);
257	pInstruments->HandBackRegion(pRgn);
258	}
259	DeleteRegionQueue->init();
260	SetActiveStreamCount(0);
261	ActiveStreamCountMax = 0;
262	if (running) this->StartThread(); // start thread only if it was running before
263	}
264
265	String GetBufferFillBytes() {
266	bool activestreams = false;
267	std::stringstream ss;
268	for (uint i = 0; i < this->Streams; i++) {
269	if (pStreams[i]->GetState() == Stream::state_unused) continue;
270	uint bufferfill = pStreams[i]->GetReadSpace() * sizeof(sample_t);
271	uint streamid = (uint) pStreams[i]->GetHandle();
272	if (!streamid) continue;
273
274	if (activestreams) ss << ",[" << streamid << ']' << bufferfill;
275	else {
276	ss << '[' << streamid << ']' << bufferfill;
277	activestreams = true;
278	}
279	}
280	return ss.str();
281	}
282
283	String GetBufferFillPercentage() {
284	bool activestreams = false;
285	std::stringstream ss;
286	for (uint i = 0; i < this->Streams; i++) {
287	if (pStreams[i]->GetState() == Stream::state_unused) continue;
288	uint bufferfill = (uint) ((float) pStreams[i]->GetReadSpace() / (float) CONFIG_STREAM_BUFFER_SIZE * 100);
289	uint streamid = (uint) pStreams[i]->GetHandle();
290	if (!streamid) continue;
291
292	if (activestreams) ss << ",[" << streamid << ']' << bufferfill << '%';
293	else {
294	ss << '[' << streamid << ']' << bufferfill;
295	activestreams = true;
296	}
297	}
298	return ss.str();
299	}
300
301	/**
302	* Returns -1 if command queue or pickup pool is full, 0 on success (will be
303	* called by audio thread within the voice class).
304	*/
305	int OrderNewStream(Stream::reference_t* pStreamRef, R* pRegion, unsigned long SampleOffset, bool DoLoop) {
306	dmsg(4,("Disk Thread: new stream ordered\n"));
307	if (CreationQueue->write_space() < 1) {
308	dmsg(1,("DiskThread: Order queue full!\n"));
309	return -1;
310	}
311
312	const Stream::OrderID_t newOrder = CreateOrderID();
313	if (!newOrder) {
314	dmsg(1,("DiskThread: there was no free slot\n"));
315	return -1; // there was no free slot
316	}
317
318	pStreamRef->State = Stream::state_active;
319	pStreamRef->OrderID = newOrder;
320	pStreamRef->hStream = CreateHandle();
321	pStreamRef->pStream = NULL; // a stream has to be activated by the disk thread first
322
323	create_command_t cmd;
324	cmd.OrderID = pStreamRef->OrderID;
325	cmd.hStream = pStreamRef->hStream;
326	cmd.pStreamRef = pStreamRef;
327	cmd.pRegion = pRegion;
328	cmd.SampleOffset = SampleOffset;
329	cmd.DoLoop = DoLoop;
330
331	CreationQueue->push(&cmd);
332	return 0;
333	}
334
335	/**
336	* Request the disk thread to delete the given disk stream. This method
337	* will return immediately, thus it won't block until the respective voice
338	* was actually deleted. (Called by audio thread within the Voice class).
339	*
340	* @param pStreamRef - stream that shall be deleted
341	* @param bRequestNotification - set to true in case you want to receive a
342	* notification once the stream has actually
343	* been deleted
344	* @returns 0 on success, -1 if command queue is full
345	* @see AskForDeletedStream()
346	*/
347	int OrderDeletionOfStream(Stream::reference_t* pStreamRef, bool bRequestNotification = false) {
348	dmsg(4,("Disk Thread: stream deletion ordered\n"));
349	if (DeletionQueue->write_space() < 1) {
350	dmsg(1,("DiskThread: Deletion queue full!\n"));
351	return -1;
352	}
353
354	delete_command_t cmd;
355	cmd.pStream = pStreamRef->pStream;
356	cmd.hStream = pStreamRef->hStream;
357	cmd.OrderID = pStreamRef->OrderID;
358	cmd.bNotify = bRequestNotification;
359
360	DeletionQueue->push(&cmd);
361	return 0;
362	}
363
364	/**
365	* Tell the disk thread to release a dimension region that belongs
366	* to an instrument which isn't loaded anymore. The disk thread
367	* will hand back the dimension region to the instrument resource
368	* manager. (OrderDeletionOfDimreg is called from the audio thread
369	* when a voice dies.)
370	*/
371	int OrderDeletionOfRegion(R* pReg) {
372	dmsg(4,("Disk Thread: dimreg deletion ordered\n"));
373	if (DeleteRegionQueue->write_space() < 1) {
374	dmsg(1,("DiskThread: DeleteRegion queue full!\n"));
375	return -1;
376	}
377	DeleteRegionQueue->push(&pReg);
378	return 0;
379	}
380
381	/**
382	* Tell the disk thread to do a program change on the specified
383	* EngineChannel.
384	*/
385	int OrderProgramChange(uint32_t Program, EngineChannel* pEngineChannel) {
386	program_change_command_t cmd;
387	cmd.Program = Program;
388	cmd.pEngineChannel = pEngineChannel;
389
390	dmsg(4,("Disk Thread: program change ordered\n"));
391	if (ProgramChangeQueue.write_space() < 1) {
392	dmsg(1,("DiskThread: ProgramChange queue full!\n"));
393	return -1;
394	}
395	ProgramChangeQueue.push(&cmd);
396	return 0;
397	}
398
399	/**
400	* Returns the pointer to a disk stream if the ordered disk stream
401	* represented by the \a StreamOrderID was already activated by the disk
402	* thread, returns NULL otherwise. If the call was successful, thus if it
403	* returned a valid stream pointer, the caller has to the store that pointer
404	* by himself, because it's not possible to call this method again with the
405	* same used order ID; this method is just intended for picking up an ordered
406	* disk stream. This method will usually be called by the voice class (within
407	* the audio thread).
408	*
409	* @param StreamOrderID - ID previously returned by OrderNewStream()
410	* @returns pointer to created stream object, NULL otherwise
411	*/
412	Stream* AskForCreatedStream(Stream::OrderID_t StreamOrderID) {
413	dmsg(4,("Disk Thread: been asked if stream already created, OrderID=%x ", StreamOrderID));
414	Stream* pStream = pCreatedStreams[StreamOrderID];
415	if (pStream && pStream != SLOT_RESERVED) {
416	dmsg(4,("(yes created)\n"));
417	pCreatedStreams[StreamOrderID] = NULL; // free the slot for a new order
418	return pStream;
419	}
420	dmsg(4,("(no not yet created)\n"));
421	return NULL;
422	}
423
424	/**
425	* In case the original sender requested a notification with his stream
426	* deletion order, he can use this method to poll if the respective stream
427	* has actually been deleted.
428	*
429	* @returns handle / identifier of the deleted stream, or
430	* Stream::INVALID_HANDLE if no notification present
431	*/
432	Stream::Handle AskForDeletedStream() {
433	if (DeletionNotificationQueue.read_space()) {
434	Stream::Handle hStream;
435	DeletionNotificationQueue.pop(&hStream);
436	return hStream;
437	} else return Stream::INVALID_HANDLE; // no notification received yet
438	}
439
440	// the number of streams currently in usage
441	// printed on the console the main thread (along with the active voice count)
442	uint GetActiveStreamCount() { return atomic_read(&ActiveStreamCount); }
443	void SetActiveStreamCount(uint Streams) { atomic_set(&ActiveStreamCount, Streams); }
444	int ActiveStreamCountMax;
445
446	protected:
447	IM* pInstruments; ///< The instrument resource manager of the engine that is using this disk thread. Used by the dimension region deletion feature.
448
449	// #########################################################################
450	// # Disk Thread Only Section
451	// # (following code should only be executed by the disk thread)
452
453	// Implementation of virtual method from class Thread
454	int Main() {
455	dmsg(3,("Disk thread running\n"));
456	while (true) {
457	#if !defined(WIN32)
458	pthread_testcancel(); // mandatory for OSX
459	#endif
460	#if CONFIG_PTHREAD_TESTCANCEL
461	TestCancel();
462	#endif
463	IsIdle = true; // will be set to false if a stream got filled
464
465	// if there are ghost streams, delete them
466	for (int i = 0; i < GhostQueue->read_space(); i++) { //FIXME: unefficient
467	delete_command_t ghostStream;
468	GhostQueue->pop(&ghostStream);
469	bool found = false;
470	for (int i = 0; i < this->Streams; i++) {
471	if (pStreams[i]->GetHandle() == ghostStream.hStream) {
472	pStreams[i]->Kill();
473	found = true;
474	// if original sender requested a notification, let him know now
475	if (ghostStream.bNotify)
476	DeletionNotificationQueue.push(&ghostStream.hStream);
477	break;
478	}
479	}
480	if (!found) GhostQueue->push(&ghostStream); // put ghost stream handle back to the queue
481	}
482
483	// if there are creation commands, create new streams
484	while (Stream::UnusedStreams > 0 && CreationQueue->read_space() > 0) {
485	create_command_t command;
486	CreationQueue->pop(&command);
487	CreateStream(command);
488	}
489
490	// if there are deletion commands, delete those streams
491	while (Stream::UnusedStreams < Stream::TotalStreams && DeletionQueue->read_space() > 0) {
492	delete_command_t command;
493	DeletionQueue->pop(&command);
494	DeleteStream(command);
495	}
496
497	// release DimensionRegions that belong to instruments
498	// that are no longer loaded
499	while (DeleteRegionQueue->read_space() > 0) {
500	R* pRgn;
501	DeleteRegionQueue->pop(&pRgn);
502	pInstruments->HandBackRegion(pRgn);
503	}
504
505	// perform MIDI program change commands
506	while (ProgramChangeQueue.read_space() > 0) {
507	program_change_command_t cmd;
508	ProgramChangeQueue.pop(&cmd);
509	cmd.pEngineChannel->ExecuteProgramChange(cmd.Program);
510	}
511	RefillStreams(); // refill the most empty streams
512
513	// if nothing was done during this iteration (eg no streambuffer
514	// filled with data) then sleep for 30ms
515	if (IsIdle) usleep(30000);
516
517	int streamsInUsage = 0;
518	for (int i = Streams - 1; i >= 0; i--) {
519	if (pStreams[i]->GetState() != Stream::state_unused) streamsInUsage++;
520	}
521	SetActiveStreamCount(streamsInUsage);
522	if (streamsInUsage > ActiveStreamCountMax) ActiveStreamCountMax = streamsInUsage;
523	}
524
525	return EXIT_FAILURE;
526	}
527
528	virtual Stream* CreateStream(long BufferSize, uint BufferWrapElements) = 0;
529
530	void CreateAllStreams(int MaxStreams, uint BufferWrapElements) {
531	for (int i = 0; i < MaxStreams; i++) {
532	pStreams[i] = CreateStream(CONFIG_STREAM_BUFFER_SIZE, BufferWrapElements);
533	}
534	}
535
536	virtual void LaunchStream (
537	Stream* pStream,
538	Stream::Handle hStream,
539	Stream::reference_t* pExportReference,
540	R* pRgn,
541	unsigned long SampleOffset,
542	bool DoLoop
543	) = 0;
544
545	friend class Stream;
546	};
547	} // namespace LinuxSampler
548
549	#endif // __LS_DISKTHREADBASE_H__