/*
* All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
* its licensors.
*
* For complete copyright and license terms please see the LICENSE at the root of this
* distribution (the "License"). All use of this software is governed by the License,
* or, if provided, by the license below or the license accompanying this file. Do not
* remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
*
*/
// Original file Copyright Crytek GMBH or its affiliates, used under license.

// Description : Queue class that can be used to send data from a single
//               writer thread to a single reader thread efficiently.


#ifndef CRYINCLUDE_CRYCOMMON_PIPE_H
#define CRYINCLUDE_CRYCOMMON_PIPE_H
#pragma once


namespace CryMT
{
    //--------------------------------------------------------------------------
    // Pipe
    //
    // This class is a templated, dynamically-resizing, lock-free queue class
    // that can be used to convey data from a single writer thread to a single
    // reader thread. It does *NOT* handle the case where there are multiple
    // readers or writers. It is designed for maximum efficiency in its
    // intended use-case.
    //
    // It differs from standard lock-free queue implementations in that it
    // uses a circular buffer approach, rather than a linked-list approach.
    // This is more memory efficient, and also should be more efficient in
    // time in all cases.
    //
    // In addition there is zero contention overhead. A normal lock-free
    // implementation makes use of atomic primitives such as Compare and Swap
    // (CAS). The drawback of this is that if there is contention, the code
    // must loop until it successfully updates the variable, whereas the time
    // taken here is basically constant (in the equlibrium case where the
    // buffer has grown to a size that is large enough to handle all the data,
    // which should happen very quickly).
    //
    //
    // Usage Notes:
    //
    // The class implements the following members:
    //
    // Push(item) -          MUST ONLY BE CALLED FROM WRITER THREAD. Push a
    //                       single value onto the queue. Constant complexity
    //                       in equilibrium case.
    //
    // bool Pop(item) -      MUST ONLY BE CALLED FROM READER THREAD. Retrieve
    //                       the next value from the queue. The return value
    //                       indicates whether there was anything to retrieve.
    //                       Constant complexity in equilibrium case.
    //
    // Clear() -             MUST ONLY BE CALLED FROM WRITER THREAD. Remove all
    //                       items from the queue. Constant complexity in
    //                       equilibrium case.
    //
    // Replace(item, repl) - MUST ONLY BE CALLED FROM WRITER THREAD. Replace
    //                       all occurrences of item with repl. Linear
    //                       complexity with respect to the size of the queue.
    //                       NOTE that this is *only* supported for data types
    //                       that can be written atomically - ints, pointers,
    //                       etc.
    //
    // Swap(other) -         MUST ONLY BE CALLED FROM WRITER THREAD. Replace
    //                       the contents of the queue with those of another
    //                       instance. This is *only* safe if the other queue
    //                       is not referred to by other threads!
    //
    //
    // Implementation Overview:
    //
    // The class basically contains a queue of queues. This is required to
    // support dynamic resizing. Each sub-queue is called a page. Values are
    // only ever pushed onto the last page - the earlier pages are simply kept
    // around until all items on them have been read, at which point they are
    // deleted.
    //
    // When the last page is full, a new page is allocated that is twice the
    // size of the last page. This means that eventually we will reach the
    // situation where we have a single page that is large enough to handle
    // the traffic in the current situation - in this situation performance is
    // optimal.
    //
    //--------------------------------------------------------------------------

#if defined(PROFILE_PIPE_SYS)
#define PIPE_FUNCTION_PROFILER(sys) FUNCTION_PROFILER_SYS(sys)
#else //defined(PROFILE_PIPE_SYS)
#define PROFILE_PIPE_SYS
#define PIPE_FUNCTION_PROFILER(sys)
#endif //defined(PROFILE_PIPE_SYS)

#if defined(PIPE_TESTING)
    namespace
    {
        FILE* f = 0;
        const char* indentation = "                                          ";
    }
#endif //defined(PIPE_TESTING)

    // Metafunction to decide whether replacing should be supported - replacing can only
    // be performed if the data type can be written atomically with no special
    // synching/primitives, ie datatype is one word or less and has no copy constructor/
    // assignment operator.
    template <typename _T>
    struct Pipe_ReplaceSupported
    {
        enum
        {
            Value = false
        };
    };

    // Replacee is supported for these simple types.
#define PIPE_REPLACE_SUPPORTED_FOR(type)    template <> \
    struct Pipe_ReplaceSupported<type> {enum {Value = true}; }
    PIPE_REPLACE_SUPPORTED_FOR(int);
    PIPE_REPLACE_SUPPORTED_FOR(unsigned int);
    PIPE_REPLACE_SUPPORTED_FOR(char);
    PIPE_REPLACE_SUPPORTED_FOR(unsigned char);
    PIPE_REPLACE_SUPPORTED_FOR(float);
#undef REPLACE_SUPPORTED_FOR

    // Replace is supported for pointer types.
    template <typename _V>
    struct Pipe_ReplaceSupported<_V*>
    {
        enum
        {
            Value = true
        };
    };

    template <typename T>
    class Pipe
    {
    public:
        typedef T Item;

        // Non-thread-safe operations.
        Pipe();
        ~Pipe();

        // Operations that can be called from the writer thread.
        void Push(const Item& item);
        void Replace(const Item& item, const Item& replacement);
        void Swap(Pipe& other);
        void Clear();

        // Operations that can be called from the reader thread.
        bool Pop(Item& item);

#if defined(PIPE_TESTING)
        int lastHead, lastTail, lastSize, lastPageCount, lastPagesExpired;
#endif //defined(PIPE_TESTING)

    private:
        enum
        {
            DEFAULT_INITIAL_SIZE = 512
        };
        enum
        {
            SIZE_INCREASE_FACTOR = 2
        };
        enum
        {
            MAX_PAGES = 32
        };

        struct Page
        {
            int size;
            int volatile head;
            int volatile tail;
            unsigned int volatile skipSend;
            unsigned int skipReceive;
            int lastSkipHeadSend;
            int lastSkipHeadReceive;

#if defined(PIPE_TESTING)
            int dbgSkipDistance, dbgClearSize, dbgClearTail;
#endif //defined(PIPE_TESTING)

            Item volatile elements[1];
        };

        struct Data
        {
            int expiredHead;
            int volatile head;
            int volatile tail;
            Page* volatile pages[MAX_PAGES];
        };

        Data* volatile m_data;
    };

    template <typename T>
    inline Pipe<T>::Pipe()
        :   m_data(0)
    {
    }

    template <typename T>
    inline Pipe<T>::~Pipe()
    {
        // Loop through all the pages and delete them.
        if (Data* data = m_data)
        {
            int head = data->head;
            int tail = data->tail;
            for (int index = head; index != tail; index = (index < MAX_PAGES - 1 ? index + 1 : 0))
            {
                delete [] reinterpret_cast<char*>(data->pages[index]);
            }
        }
    }

    template <typename T>
    inline void Pipe<T>::Push(const Item& item)
    {
        PIPE_FUNCTION_PROFILER(PROFILE_PIPE_SYS);

        // First we need to check whether the class is empty - this means we have to initialize the object.
        Data* data = m_data;
        if (!data)
        {
            data = new Data;
            data->expiredHead = 0;
            data->head = 0;
            data->tail = 0;
            m_data = data;
        }

        // Now we should clean up any pages in the expired queue. When a page is emptied, and there is a larger page
        // to use, the reader thread increments the head pointer to point to the next page. The expiredHead member
        // lags behind, so we can use it to see if there are any pages that *both* the reader and writer threads are
        // done with. These can then be safely deleted.
        int expiredHead = data->expiredHead;
        int pageHead = data->head;
        int pageTail = data->tail;
        for (; expiredHead != pageHead; expiredHead = (expiredHead < MAX_PAGES - 1 ? expiredHead + 1 : 0))
        {
#if defined(PIPE_TESTING)
            int pageCount = pageTail - expiredHead;
            if (pageCount < 0)
            {
                pageCount += MAX_PAGES;
            }
            if (f)
            {
                fprintf(f, "Deleting page %d of size %d, %d pages remaining\n", expiredHead, data->pages[expiredHead]->size, pageCount - 1);
            }
#endif //defined(PIPE_TESTING)

            delete [] reinterpret_cast<char*>(data->pages[expiredHead]);
        }
        data->expiredHead = expiredHead;

        // In order to guarantee delivery order, we can only add to the last page. If there is no last page or there
        // is no space in the last page, then we must allocate a new page.
        Page* page = 0;
        int head = 0;
        int tail = -1; // This makes the calculations think the queue is full if there is no page.
        int size = 0;
        if (pageHead != pageTail)
        {
            int lastPageIndex = pageTail - 1;
            if (lastPageIndex < 0)
            {
                lastPageIndex += MAX_PAGES;
            }
            page = data->pages[lastPageIndex];
            head = page->head;
            tail = page->tail;
            size = page->size;
        }

        // Figure out how many free elements are in this page.
        int freeCount = head - tail;
        if (freeCount < 0)
        {
            freeCount += size;
        }

        // If there is no free page, create a new one now.
        if (!page || freeCount == 1)
        {
            // Calculate the size of the new page - we start at 16 elements and double the size each time after that.
            int newSize = size * SIZE_INCREASE_FACTOR;
            if (newSize == 0)
            {
                newSize = DEFAULT_INITIAL_SIZE;
            }
            int newSizeInBytes = sizeof(Page) + (newSize - 1) * sizeof(Item); // The -1 is because the Page structure has a 1-element array member.

            // Create and initialize the page.
            page = reinterpret_cast<Page*>(new char[newSizeInBytes]);
            page->size = size = newSize;
            page->head = head = 0;
            page->tail = tail = 0;

            page->skipSend = 0;
            page->skipReceive = 0;
            page->lastSkipHeadSend = 0;
            page->lastSkipHeadReceive = 0;

            // Add the page to the page queue. We have made the page queue big enough that it is inconceivable that it
            // could ever be full (if the page queue became full, it would mean that the writer thread added more than
            // 68bn items without any being read by the reader thread).
            data->pages[pageTail] = page;
            ++pageTail;
            if (pageTail >= MAX_PAGES)
            {
                pageTail = 0;
            }
            data->tail = pageTail;

#if defined(PIPE_TESTING)
            int pageCount = pageTail - expiredHead;
            if (pageCount < 0)
            {
                pageCount += MAX_PAGES;
            }
            if (f)
            {
                fprintf(f, "Creating page of size %d, %d pages exist\n", newSize, pageCount);
            }
#endif //defined(PIPE_TESTING)
        }

        // Add the item to the selected page. We can now safely assume that the page is not full.
        page->elements[tail] = item;
        ++tail;
        if (tail >= size)
        {
            tail = 0;
        }
        page->tail = tail;
    }

    template <typename T>
    inline void Pipe<T>::Swap(Pipe& other)
    {
        // Basically this operation simply exchanges the m_data pointers of both instances. Note that
        // it is not safe to destruct either of the instances immediately, as the reader thread may
        // still be referring to the m_data object. Both instances must be destructed when only one thread
        // is in operation. This means that code like Pipe().Swap(myPipe) is a very bad idea (this is
        // an stl idiom that some people might be tempted to try).

        // NB that it is NOT safe to do a swap if BOTH instances are currently being read by other threads.
        // There is no way to guarantee that the right items go to the right threads in this case. It is
        // only safe to call Swap() on the instance that is being actively read, passing it the inactive
        // instance as the parameter.
        Data* data = m_data;
        m_data = other.m_data; // At this point both instances refer to the same data.
        other.m_data = data;
    }

    template <typename T>
    inline void Pipe<T>::Replace(const Item& item, const Item& replacement)
    {
        PIPE_FUNCTION_PROFILER(PROFILE_PIPE_SYS);

        Data* data = m_data;
        if (!data)
        {
            return;
        }

        // In the code following, it is possible that we may end up searching through elements of the pages that
        // the reader thread has already read. However, they should still be in a valid state, since the reader thread
        // does not alter the items in the queue, and overwriting them should have no impact. Note that no new items will
        // be added to any pages, since we are the only writer thread.

        // Loop through all the pages that are still active, searching for any items to replace.
        int iterations = 0;
        for (int pageIndex = data->head, pageTail = data->tail; pageIndex != pageTail; pageIndex = (pageIndex < MAX_PAGES - 1 ? pageIndex + 1 : 0))
        {
            Page* page = data->pages[pageIndex];
            int head = page->head;
            int tail = page->tail;
            int size = page->size;

            for (int index = head; index != tail; index = (index < size - 1 ? index + 1 : 0))
            {
                ++iterations;
                if (page->elements[index] == item)
                {
                    page->elements[index] = replacement;
                }
            }
        }
    }

    template <typename T>
    inline void Pipe<T>::Clear()
    {
        PIPE_FUNCTION_PROFILER(PROFILE_PIPE_SYS);

        Data* data = m_data;
        if (!data)
        {
            return;
        }

        // It is not safe to clear the data directly, and in particular it's not safe to delete any pages. A simple
        // way to clear the queue would be to move the head index down to the tail. However, this cannot be done
        // safely directly. Therefore we need another variable which contains the desired head, which the reader
        // thread can copy into the head. Yet this too is unsafe, since the reader thread wont know whether a skip
        // value is new or whether it is the same one it read last time. Therefore we use a monotonically increasing variable
        // to convey the desired skip position to the reader thread.
        for (int pageIndex = data->head, pageTail = data->tail; pageIndex != pageTail; pageIndex = (pageIndex < MAX_PAGES - 1 ? pageIndex + 1 : 0))
        {
            Page* page = data->pages[pageIndex];
            //int head = page->head;
            int tail = page->tail;
            int size = page->size;

            // Work out how far to increment the skip position. We want to move the head pointer to the tail pointer.
            // We cannot safely refer directly to the head variable, since it is volatile, so we refer to the last head
            // value we conveyed to the reader thread.
            int skipDistance = tail - page->lastSkipHeadSend;
            page->lastSkipHeadSend = tail;
            if (skipDistance < 0)
            {
                skipDistance += size;
            }

#if defined(PIPE_TESTING)
            int skipReceive = page->skipReceive;
            int oldDistance = page->skipSend - skipReceive;
            if (oldDistance + skipDistance >= size)
            {
                static int s_size;
                static int s_skipDistance;
                static int s_tail;
                static int s_pageIndex;
                static int s_lastSkipHeadSend;
                static int s_oldDistance;
                static int s_skipReceive;
                s_size = size;
                s_skipDistance = skipDistance;
                s_tail = tail;
                s_pageIndex = pageIndex;
                s_lastSkipHeadSend = page->lastSkipHeadSend;
                s_oldDistance = oldDistance;
                s_skipReceive = skipReceive;

                //DebugBreak();
            }
#endif //defined(PIPE_TESTING)

            // Convey this new skip value to the reader thread. We do this by adding the skip distance to the send variable
            // - the reader thread then adds this difference to its own copy of the forced head, and can then update the
            // variable accordingly. This is safe to do multiple times before the reader thread reacts, because the send
            // variable is monotonic, and so does not throw away information. Even if the variable wraps, the mechanism
            // works with differences, so it should be fine unless we want to skip more than 2bn elements before the reader
            // reacts.
            page->skipSend += skipDistance;

#if defined(PIPE_TESTING)
            page->dbgSkipDistance = skipDistance;
            page->dbgClearSize = size;
            page->dbgClearTail = tail;
#endif //defined(PIPE_TESTING)
        }
    }

    template <typename T>
    inline bool Pipe<T>::Pop(Item& item)
    {
        PIPE_FUNCTION_PROFILER(PROFILE_PIPE_SYS);

        // Check whether we have been initialized.
        Data* data = m_data;
        if (!data)
        {
            return false;
        }

        // Find out how many pages exist.
        int pageHead = data->head;
        int pageTail = data->tail;
        int pageCount = pageTail - pageHead;
        if (pageCount < 0)
        {
            pageCount += MAX_PAGES;
        }

        // Loop through the pages until we find a non-empty one. If there are any empty ones,
        // move the head index past them. This will indicate to the writer thread that it can
        // safely delete those pages.
        Page* page = 0;
        int head = -1;
        int tail = -1;
        int size = 0;
        if (pageHead != pageTail)
        {
            for (;; )
            {
                Page* p = data->pages[pageHead];
                head = p->head;
                tail = p->tail;
                size = p->size;

                // It's possible that the writer thread wants us to skip some elements - this happens
                // when the queue is cleared, for example. The writer thread conveys the desired head
                // position to us by increasing the value of a variable - the difference is the amount
                // the head variable should be moved (based on the last time we were told to skip forward).
                unsigned int skipSend = p->skipSend;
                unsigned int skipDistance = skipSend - p->skipReceive;
                if (skipDistance)
                {
                    p->skipReceive = skipSend;
                    p->lastSkipHeadReceive = (p->lastSkipHeadReceive + skipDistance) % size;

                    if (p->lastSkipHeadReceive >= size)
                    {
                        static unsigned int s_skipSend;
                        static unsigned int s_skipDistance;
                        static unsigned int s_head;
                        static unsigned int s_size;
                        static unsigned int s_tail;
                        s_skipSend = skipSend;
                        s_skipDistance = skipDistance;
                        s_head = head;
                        s_size = size;
                        s_tail = tail;

                        //DebugBreak();
                    }

                    p->head = head = p->lastSkipHeadReceive; // I'm pretty sure it's impossible for our head to be ahead of this value...
                }

                if (head != tail)
                {
                    page = p; // We have found a non-empty page.
                    break;
                }

                // Don't skip the last page, otherwise it will be deleted and we will have to reallocate
                // it next push.
                int nextPageHead = (pageHead < MAX_PAGES - 1 ? pageHead + 1 : 0);
                if (nextPageHead == pageTail)
                {
                    break;
                }
                pageHead = nextPageHead;
            }
        }
#if defined(PIPE_TESTING)
        lastPagesExpired = pageHead - data->head;
#endif //defined(PIPE_TESTING)
        data->head = pageHead;

        // If we found no non-empty page, then the queue is empty and we return nothing.
        if (!page)
        {
            return false;
        }

#if defined(PIPE_TESTING)
        lastHead = head;
        lastTail = tail;
        lastSize = size;
        lastPageCount = pageCount;
#endif //defined(PIPE_TESTING)

        // Get the first item of the selected page. It is now safe to assume that the page is not empty.
        item = page->elements[head];
        ++head;
        if (head >= size)
        {
            head = 0;
        }
        page->head = head;
        return true;
    }

#if defined(PIPE_TESTING)

    namespace PipeTest
    {
        inline void TestPipeSingleThread()
        {
            static const int ITERATION_COUNT = 100;
            static const int PUSH_COUNT = 100000;

            Pipe<int> pipe;
            fopen_s(&f, "c:\\pipetestsinglethread.txt", "w");
            int currentValue = 0;
            for (int iteration = 0; iteration < ITERATION_COUNT; ++iteration)
            {
                fprintf(f, "\n******************************   ITERATION %d   **************************************\n", iteration);
                int pushCount = cry_random(0, PUSH_COUNT - 1);
                int popCount = cry_random(0, PUSH_COUNT - 1);

                for (int i = 0; i < pushCount; ++i)
                {
                    fprintf(f, "Pushing: %d\n", currentValue);
                    pipe.Push(currentValue++);
                }

                for (int i = 0; i < popCount; ++i)
                {
                    int value;
                    if (pipe.Pop(value))
                    {
                        fprintf(f, "%sPopping: %d\n", indentation, value);
                    }
                    else
                    {
                        fprintf(f, "%sPopping: -\n", indentation);
                    }
                }

                fprintf(f, "%sClearing:\n", indentation);
                pipe.Clear();
                for (int i = 0; i < 2; ++i)
                {
                    int value;
                    if (pipe.Pop(value))
                    {
                        fprintf(f, "%sPopping: %d\n", indentation, value);
                    }
                    else
                    {
                        fprintf(f, "%sPopping: -\n", indentation);
                    }
                }
            }
            fclose(f);
            f = 0;
        }

        struct TestPipeThreadData
        {
            Pipe<int>* pipe;
        };

        inline DWORD WINAPI TestPipeWriterThread(void* userData)
        {
            static const int MAX_VALUES = 100000000;

            if (f)
            {
                fprintf(f, "Writer thread beginning\n");
            }

            TestPipeThreadData* threadData = static_cast<TestPipeThreadData*>(userData);
            Pipe<int>& pipe = *threadData->pipe;
            for (int i = 0; i < MAX_VALUES; ++i)
            {
                if (i % 100000 == 0)
                {
                    if (f)
                    {
                        fprintf(f, "Clearing queue\n");
                    }
                    pipe.Clear();
                }
                //if (i % 100000 == 10)
                //{
                //  int replaceIndex = i - 9;
                //  if (f)
                //      fprintf(f, "Replacing %d\n", replaceIndex);
                //  pipe.Replace(replaceIndex, 0);
                //}

                pipe.Push(i);
            }

            if (f)
            {
                fprintf(f, "Pushing -1\n");
            }
            pipe.Push(-1);

            if (f)
            {
                fprintf(f, "Writer thread exitting\n");
            }

            return 0;
        }

        inline DWORD WINAPI TestPipeReaderThread(void* userData)
        {
            TestPipeThreadData* threadData = static_cast<TestPipeThreadData*>(userData);
            Pipe<int>& pipe = *threadData->pipe;

            if (f)
            {
                fprintf(f, "%sReader thread beginning\n", indentation);
            }

            static Pipe<int>* TEST;
            TEST = &pipe;

            int lastReceivedValue = -1;
            for (;; )
            {
                int value;
                if (pipe.Pop(value))
                {
                    if (value == -1)
                    {
                        if (f)
                        {
                            fprintf(f, "%sReceived -1\n", indentation);
                        }
                        break;
                    }

                    if (value - lastReceivedValue != 1)
                    {
                        if (pipe.lastPagesExpired)
                        {
                            pipe.lastPagesExpired = pipe.lastPagesExpired;
                        }

                        if (f)
                        {
                            fprintf(f, "%sMissed %d items starting with %d (lastHead = %d, lastTail = %d, lastSize = %d, lastPageCount = %d, lastPagesExpired = %d)\n",
                                indentation, value - lastReceivedValue - 1, lastReceivedValue + 1, pipe.lastHead, pipe.lastTail, pipe.lastSize, pipe.lastPageCount, pipe.lastPagesExpired);
                        }
                    }

                    lastReceivedValue = value;
                }
            }

            if (f)
            {
                fprintf(f, "%sReader thread exitting\n", indentation);
            }

            return 0;
        }

        inline void TestPipeMultiThread()
        {
            TestPipeThreadData threadData;
            Pipe<int> pipe;
            threadData.pipe = &pipe;

            fopen_s(&f, "c:\\pipetestmultithread.txt", "w");

            HANDLE hWriterThread = CreateThread(0, 0, TestPipeWriterThread, &threadData, 0, 0);
            HANDLE hReaderThread = CreateThread(0, 0, TestPipeReaderThread, &threadData, 0, 0);

            HANDLE threads[] = {hWriterThread, hReaderThread};
            WaitForMultipleObjects(sizeof(threads) / sizeof(threads[0]), threads, TRUE, INFINITE);

            fclose(f);
            f = 0;
        }

        inline void TestPipe()
        {
            //TestPipeSingleThread();
            TestPipeMultiThread();
        }
    }

#endif //defined(PIPE_TESTING)
} // namespace CryMT

#endif // CRYINCLUDE_CRYCOMMON_PIPE_H