/*------------------------------------------------------------------------- * drawElements C++ Base Library * ----------------------------- * * Copyright 2014 The Android Open Source Project * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. * *//*! * \file * \brief Thread-safe ring buffer template. *//*--------------------------------------------------------------------*/ #include "deThreadSafeRingBuffer.hpp" #include "deRandom.hpp" #include "deThread.hpp" #include <vector> using std::vector; namespace de { namespace { struct Message { deUint32 data; Message (deUint16 threadId, deUint16 payload) : data((threadId << 16) | payload) { } Message (void) : data(0) { } deUint16 getThreadId (void) const { return (deUint16)(data >> 16); } deUint16 getPayload (void) const { return (deUint16)(data & 0xffff); } }; class Consumer : public Thread { public: Consumer (ThreadSafeRingBuffer<Message>& buffer, int numProducers) : m_buffer (buffer) { m_lastPayload.resize(numProducers, 0); m_payloadSum.resize(numProducers, 0); } void run (void) { for (;;) { Message msg = m_buffer.popBack(); deUint16 threadId = msg.getThreadId(); if (threadId == 0xffff) break; DE_TEST_ASSERT(de::inBounds<int>(threadId, 0, (int)m_lastPayload.size())); DE_TEST_ASSERT((m_lastPayload[threadId] == 0 && msg.getPayload() == 0) || m_lastPayload[threadId] < msg.getPayload()); m_lastPayload[threadId] = msg.getPayload(); m_payloadSum[threadId] += (deUint32)msg.getPayload(); } } deUint32 getPayloadSum (deUint16 threadId) const { return m_payloadSum[threadId]; } private: ThreadSafeRingBuffer<Message>& m_buffer; vector<deUint16> m_lastPayload; vector<deUint32> m_payloadSum; }; class Producer : public Thread { public: Producer (ThreadSafeRingBuffer<Message>& buffer, deUint16 threadId, int dataSize) : m_buffer (buffer) , m_threadId (threadId) , m_dataSize (dataSize) { } void run (void) { // Yield to give main thread chance to start other producers. deSleep(1); for (int ndx = 0; ndx < m_dataSize; ndx++) m_buffer.pushFront(Message(m_threadId, (deUint16)ndx)); } private: ThreadSafeRingBuffer<Message>& m_buffer; deUint16 m_threadId; int m_dataSize; }; } // anonymous void ThreadSafeRingBuffer_selfTest (void) { const int numIterations = 16; for (int iterNdx = 0; iterNdx < numIterations; iterNdx++) { Random rnd (iterNdx); int bufSize = rnd.getInt(1, 2048); int numProducers = rnd.getInt(1, 16); int numConsumers = rnd.getInt(1, 16); int dataSize = rnd.getInt(1000, 10000); ThreadSafeRingBuffer<Message> buffer (bufSize); vector<Producer*> producers; vector<Consumer*> consumers; for (int i = 0; i < numProducers; i++) producers.push_back(new Producer(buffer, (deUint16)i, dataSize)); for (int i = 0; i < numConsumers; i++) consumers.push_back(new Consumer(buffer, numProducers)); // Start consumers. for (vector<Consumer*>::iterator i = consumers.begin(); i != consumers.end(); i++) (*i)->start(); // Start producers. for (vector<Producer*>::iterator i = producers.begin(); i != producers.end(); i++) (*i)->start(); // Wait for producers. for (vector<Producer*>::iterator i = producers.begin(); i != producers.end(); i++) (*i)->join(); // Write end messages for consumers. for (int i = 0; i < numConsumers; i++) buffer.pushFront(Message(0xffff, 0)); // Wait for consumers. for (vector<Consumer*>::iterator i = consumers.begin(); i != consumers.end(); i++) (*i)->join(); // Verify payload sums. deUint32 refSum = 0; for (int i = 0; i < dataSize; i++) refSum += (deUint32)(deUint16)i; for (int i = 0; i < numProducers; i++) { deUint32 cmpSum = 0; for (int j = 0; j < numConsumers; j++) cmpSum += consumers[j]->getPayloadSum((deUint16)i); DE_TEST_ASSERT(refSum == cmpSum); } // Free resources. for (vector<Producer*>::iterator i = producers.begin(); i != producers.end(); i++) delete *i; for (vector<Consumer*>::iterator i = consumers.begin(); i != consumers.end(); i++) delete *i; } } } // de