executor/xeCallQueue.cpp

   1 /*-------------------------------------------------------------------------
   2  * drawElements Quality Program Test Executor
   3  * ------------------------------------------
   4  *
   5  * Copyright 2014 The Android Open Source Project
   6  *
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  *
  19  *//*!
  20  * \file
  21  * \brief Cross-thread function call dispatcher.
  22  *//*--------------------------------------------------------------------*/
  23
  24 #include "xeCallQueue.hpp"
  25 #include "deInt32.h"
  26 #include "deMemory.h"
  27
  28 using std::vector;
  29
  30 static inline int getNextQueueSize (int curSize, int minNewSize)
  31 {
  32         return de::max(curSize*2, 1<<deLog2Ceil32(minNewSize));
  33 }
  34
  35 namespace xe
  36 {
  37
  38 // CallQueue
  39
  40 CallQueue::CallQueue (void)
  41         : m_canceled    (false)
  42         , m_callSem             (0)
  43         , m_callQueue   (64)
  44 {
  45 }
  46
  47 CallQueue::~CallQueue (void)
  48 {
  49         // Destroy all calls.
  50         for (vector<Call*>::iterator i = m_calls.begin(); i != m_calls.end(); i++)
  51                 delete *i;
  52 }
  53
  54 void CallQueue::cancel (void)
  55 {
  56         m_canceled = true;
  57         m_callSem.increment();
  58 }
  59
  60 void CallQueue::callNext (void)
  61 {
  62         Call* call = DE_NULL;
  63
  64         // Wait for a call.
  65         m_callSem.decrement();
  66
  67         if (m_canceled)
  68                 return;
  69
  70         // Acquire call from buffer.
  71         {
  72                 de::ScopedLock lock(m_lock);
  73                 call = m_callQueue.popBack();
  74         }
  75
  76         try
  77         {
  78                 // \note Enqueue lock is not held during call so it is possible to enqueue more work from dispatched call.
  79                 CallReader reader(call);
  80
  81                 call->getFunction()(reader);
  82
  83                 // check callee consumed all
  84                 DE_ASSERT(reader.isDataConsumed());
  85                 call->clear();
  86         }
  87         catch (const std::exception&)
  88         {
  89                 try
  90                 {
  91                         // Try to push call into free calls list.
  92                         de::ScopedLock lock(m_lock);
  93                         m_freeCalls.push_back(call);
  94                 }
  95                 catch (const std::exception&)
  96                 {
  97                         // We can't do anything but ignore this.
  98                 }
  99
 100                 throw;
 101         }
 102
 103         // Push back to free calls list.
 104         {
 105                 de::ScopedLock lock(m_lock);
 106                 m_freeCalls.push_back(call);
 107         }
 108 }
 109
 110 Call* CallQueue::getEmptyCall (void)
 111 {
 112         de::ScopedLock  lock    (m_lock);
 113         Call*                   call    = DE_NULL;
 114
 115         // Try to get from free calls list.
 116         if (!m_freeCalls.empty())
 117         {
 118                 call = m_freeCalls.back();
 119                 m_freeCalls.pop_back();
 120         }
 121
 122         // If no free calls were available, create a new.
 123         if (!call)
 124         {
 125                 m_calls.reserve(m_calls.size()+1);
 126                 call = new Call();
 127                 m_calls.push_back(call);
 128         }
 129
 130         return call;
 131 }
 132
 133 void CallQueue::enqueue (Call* call)
 134 {
 135         de::ScopedLock lock(m_lock);
 136
 137         if (m_callQueue.getNumFree() == 0)
 138         {
 139                 // Call queue must be grown.
 140                 m_callQueue.resize(getNextQueueSize(m_callQueue.getSize(), m_callQueue.getSize()+1));
 141         }
 142
 143         m_callQueue.pushFront(call);
 144         m_callSem.increment();
 145 }
 146
 147 void CallQueue::freeCall (Call* call)
 148 {
 149         de::ScopedLock lock(m_lock);
 150         m_freeCalls.push_back(call);
 151 }
 152
 153 // Call
 154
 155 Call::Call (void)
 156         : m_func(DE_NULL)
 157 {
 158 }
 159
 160 Call::~Call (void)
 161 {
 162 }
 163
 164 void Call::clear (void)
 165 {
 166         m_func = DE_NULL;
 167         m_data.clear();
 168 }
 169
 170 // CallReader
 171
 172 CallReader::CallReader (Call* call)
 173         : m_call        (call)
 174         , m_curPos      (0)
 175 {
 176 }
 177
 178 void CallReader::read (deUint8* bytes, size_t numBytes)
 179 {
 180         DE_ASSERT(m_curPos + numBytes <= m_call->getDataSize());
 181         deMemcpy(bytes, m_call->getData()+m_curPos, numBytes);
 182         m_curPos += numBytes;
 183 }
 184
 185 const deUint8* CallReader::getDataBlock (size_t numBytes)
 186 {
 187         DE_ASSERT(m_curPos + numBytes <= m_call->getDataSize());
 188
 189         const deUint8* ptr = m_call->getData()+m_curPos;
 190         m_curPos += numBytes;
 191
 192         return ptr;
 193 }
 194
 195 bool CallReader::isDataConsumed (void) const
 196 {
 197         return m_curPos == m_call->getDataSize();
 198 }
 199
 200 CallReader& operator>> (CallReader& reader, std::string& value)
 201 {
 202         value.clear();
 203         for (;;)
 204         {
 205                 char c;
 206                 reader.read((deUint8*)&c, sizeof(char));
 207                 if (c != 0)
 208                         value.push_back(c);
 209                 else
 210                         break;
 211         }
 212
 213         return reader;
 214 }
 215
 216 // CallWriter
 217
 218 CallWriter::CallWriter (CallQueue* queue, Call::Function function)
 219         : m_queue               (queue)
 220         , m_call                (queue->getEmptyCall())
 221         , m_enqueued    (false)
 222 {
 223         m_call->setFunction(function);
 224 }
 225
 226 CallWriter::~CallWriter (void)
 227 {
 228         if (!m_enqueued)
 229                 m_queue->freeCall(m_call);
 230 }
 231
 232 void CallWriter::write (const deUint8* bytes, size_t numBytes)
 233 {
 234         DE_ASSERT(!m_enqueued);
 235         size_t curPos = m_call->getDataSize();
 236         m_call->setDataSize(curPos+numBytes);
 237         deMemcpy(m_call->getData()+curPos, bytes, numBytes);
 238 }
 239
 240 void CallWriter::enqueue (void)
 241 {
 242         DE_ASSERT(!m_enqueued);
 243         m_queue->enqueue(m_call);
 244         m_enqueued = true;
 245 }
 246
 247 CallWriter& operator<< (CallWriter& writer, const char* str)
 248 {
 249         int pos = 0;
 250         for (;;)
 251         {
 252                 writer.write((const deUint8*)str + pos, sizeof(char));
 253                 if (str[pos] == 0)
 254                         break;
 255                 pos += 1;
 256         }
 257
 258         return writer;
 259 }
 260
 261 } // xe