dali/internal/update/queue/update-message-queue.cpp

   1 /*
   2  * Copyright (c) 2022 Samsung Electronics Co., Ltd.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  *
  16  */
  17
  18 // CLASS HEADER
  19 #include <dali/internal/update/queue/update-message-queue.h>
  20
  21 // INTERNAL INCLUDES
  22 #include <dali/devel-api/threading/mutex.h>
  23 #include <dali/integration-api/render-controller.h>
  24 #include <dali/internal/common/message-buffer.h>
  25 #include <dali/internal/common/message.h>
  26 #include <dali/internal/render/common/performance-monitor.h>
  27 #include <dali/public-api/common/vector-wrapper.h>
  28
  29 using std::vector;
  30
  31 using Dali::Integration::RenderController;
  32 using Dali::Internal::SceneGraph::SceneGraphBuffers;
  33
  34 namespace Dali
  35 {
  36 namespace Internal
  37 {
  38 namespace // unnamed namespace
  39 {
  40 // A message to set Actor::SIZE is 72 bytes on 32bit device
  41 // A buffer of size 32768 would store (32768 - 4) / (72 + 4) = 431 of those messages
  42 static const std::size_t INITIAL_BUFFER_SIZE   = 32768;
  43 static const std::size_t MAX_BUFFER_CAPACITY   = 73728; // Avoid keeping buffers which exceed this
  44 static const std::size_t MAX_FREE_BUFFER_COUNT = 3;     // Allow this number of buffers to be recycled
  45
  46 // A queue of message buffers
  47 typedef vector<MessageBuffer*> MessageBufferQueue;
  48 using MessageBufferIter = MessageBufferQueue::iterator;
  49
  50 using MessageQueueMutex = Dali::Mutex;
  51
  52 } // unnamed namespace
  53
  54 namespace Update
  55 {
  56 /**
  57  * Private MessageQueue data
  58  */
  59 struct MessageQueue::Impl
  60 {
  61   Impl(RenderController& controller, const SceneGraphBuffers& buffers)
  62   : renderController(controller),
  63     sceneGraphBuffers(buffers),
  64     processingEvents(false),
  65     queueWasEmpty(true),
  66     sceneUpdateFlag(false),
  67     sceneUpdate(0),
  68     currentMessageBuffer(nullptr)
  69   {
  70   }
  71
  72   ~Impl()
  73   {
  74     // Delete the current buffer
  75     if(currentMessageBuffer)
  76     {
  77       DeleteBufferContents(currentMessageBuffer);
  78       delete currentMessageBuffer;
  79     }
  80
  81     // Delete the unprocessed buffers
  82     const MessageBufferIter processQueueEndIter = processQueue.end();
  83     for(MessageBufferIter iter = processQueue.begin(); iter != processQueueEndIter; ++iter)
  84     {
  85       MessageBuffer* unprocessedBuffer = *iter;
  86       DeleteBufferContents(unprocessedBuffer);
  87       delete unprocessedBuffer;
  88     }
  89
  90     // Delete the recycled buffers
  91     const MessageBufferIter recycleQueueEndIter = recycleQueue.end();
  92     for(MessageBufferIter iter = recycleQueue.begin(); iter != recycleQueueEndIter; ++iter)
  93     {
  94       MessageBuffer* recycledBuffer = *iter;
  95       DeleteBufferContents(recycledBuffer);
  96       delete recycledBuffer;
  97     }
  98
  99     const MessageBufferIter freeQueueEndIter = freeQueue.end();
 100     for(MessageBufferIter iter = freeQueue.begin(); iter != freeQueueEndIter; ++iter)
 101     {
 102       MessageBuffer* freeBuffer = *iter;
 103       DeleteBufferContents(freeBuffer);
 104       delete freeBuffer;
 105     }
 106   }
 107
 108   void DeleteBufferContents(MessageBuffer* buffer)
 109   {
 110     for(MessageBuffer::Iterator iter = buffer->Begin(); iter.IsValid(); iter.Next())
 111     {
 112       MessageBase* message = reinterpret_cast<MessageBase*>(iter.Get());
 113
 114       // Call virtual destructor explictly; since delete will not be called after placement new
 115       message->~MessageBase();
 116     }
 117   }
 118
 119   RenderController&        renderController;  ///< render controller
 120   const SceneGraphBuffers& sceneGraphBuffers; ///< Used to keep track of which buffers are being written or read.
 121
 122   bool processingEvents; ///< Whether messages queued will be flushed by core
 123   bool queueWasEmpty;    ///< Flag whether the queue was empty during the Update()
 124   bool sceneUpdateFlag;  ///< true when there is a new message that requires a scene-graph node tree update
 125   int  sceneUpdate;      ///< Non zero when there is a message in the queue requiring a scene-graph node tree update
 126
 127   MessageQueueMutex  queueMutex;   ///< queueMutex must be locked whilst accessing processQueue or recycleQueue
 128   MessageBufferQueue processQueue; ///< to process in the next update
 129   MessageBufferQueue recycleQueue; ///< to recycle MessageBuffers after the messages have been processed
 130
 131   MessageBuffer*     currentMessageBuffer; ///< can be used without locking
 132   MessageBufferQueue freeQueue;            ///< buffers from the recycleQueue; can be used without locking
 133 };
 134
 135 MessageQueue::MessageQueue(Integration::RenderController& controller, const SceneGraph::SceneGraphBuffers& buffers)
 136 : mImpl(nullptr)
 137 {
 138   mImpl = new Impl(controller, buffers);
 139 }
 140
 141 MessageQueue::~MessageQueue()
 142 {
 143   delete mImpl;
 144 }
 145
 146 void MessageQueue::EventProcessingStarted()
 147 {
 148   mImpl->processingEvents = true; // called from event thread
 149 }
 150
 151 // Called from event thread
 152 uint32_t* MessageQueue::ReserveMessageSlot(uint32_t requestedSize, bool updateScene)
 153 {
 154   DALI_ASSERT_DEBUG(0 != requestedSize);
 155
 156   if(updateScene)
 157   {
 158     mImpl->sceneUpdateFlag = true;
 159   }
 160
 161   if(!mImpl->currentMessageBuffer)
 162   {
 163     const MessageBufferIter endIter = mImpl->freeQueue.end();
 164
 165     // Find the largest recycled buffer from freeQueue
 166     MessageBufferIter nextBuffer = endIter;
 167     for(MessageBufferIter iter = mImpl->freeQueue.begin(); iter != endIter; ++iter)
 168     {
 169       if(endIter == nextBuffer ||
 170          (*nextBuffer)->GetCapacity() < (*iter)->GetCapacity())
 171       {
 172         nextBuffer = iter;
 173       }
 174     }
 175
 176     if(endIter != nextBuffer)
 177     {
 178       // Reuse a recycled buffer from freeQueue
 179       mImpl->currentMessageBuffer = *nextBuffer;
 180       mImpl->freeQueue.erase(nextBuffer);
 181     }
 182     else
 183     {
 184       mImpl->currentMessageBuffer = new MessageBuffer(INITIAL_BUFFER_SIZE);
 185     }
 186   }
 187
 188   // If we are inside Core::ProcessEvents(), core will automatically flush the queue.
 189   // If we are outside, then we have to request a call to Core::ProcessEvents() on idle.
 190   if(false == mImpl->processingEvents)
 191   {
 192     mImpl->renderController.RequestProcessEventsOnIdle(false);
 193   }
 194
 195   return mImpl->currentMessageBuffer->ReserveMessageSlot(requestedSize);
 196 }
 197
 198 // Called from event thread
 199 bool MessageQueue::FlushQueue()
 200 {
 201   const bool messagesToProcess = (nullptr != mImpl->currentMessageBuffer);
 202
 203   // If there're messages to flush
 204   if(messagesToProcess)
 205   {
 206     // queueMutex must be locked whilst accessing processQueue or recycleQueue
 207     MessageQueueMutex::ScopedLock lock(mImpl->queueMutex);
 208
 209     mImpl->processQueue.push_back(mImpl->currentMessageBuffer);
 210     mImpl->currentMessageBuffer = nullptr;
 211
 212     // Grab any recycled MessageBuffers
 213     while(!mImpl->recycleQueue.empty())
 214     {
 215       MessageBuffer* recycled = mImpl->recycleQueue.back();
 216       mImpl->recycleQueue.pop_back();
 217
 218       // Guard against excessive message buffer growth
 219       if(MAX_FREE_BUFFER_COUNT < mImpl->freeQueue.size() ||
 220          MAX_BUFFER_CAPACITY < recycled->GetCapacity())
 221       {
 222         delete recycled;
 223       }
 224       else
 225       {
 226         mImpl->freeQueue.push_back(recycled);
 227       }
 228     }
 229
 230     if(mImpl->sceneUpdateFlag)
 231     {
 232       mImpl->sceneUpdate |= 2;
 233       mImpl->sceneUpdateFlag = false;
 234     }
 235   }
 236
 237   mImpl->processingEvents = false;
 238
 239   return messagesToProcess;
 240 }
 241
 242 bool MessageQueue::ProcessMessages(BufferIndex updateBufferIndex)
 243 {
 244   PERF_MONITOR_START(PerformanceMonitor::PROCESS_MESSAGES);
 245
 246   MessageBufferQueue copiedProcessQueue;
 247   bool               sceneUpdated = false;
 248   {
 249     // queueMutex must be locked whilst accessing queue
 250     MessageQueueMutex::ScopedLock lock(mImpl->queueMutex);
 251     copiedProcessQueue = mImpl->processQueue;
 252
 253     mImpl->sceneUpdate >>= 1;
 254
 255     sceneUpdated = (mImpl->sceneUpdate & 0x01); // if it was previously 2, scene graph was updated.
 256
 257     mImpl->queueWasEmpty = mImpl->processQueue.empty(); // Flag whether we processed anything
 258
 259     mImpl->processQueue.clear();
 260   }
 261
 262   for(auto&& buffer : copiedProcessQueue)
 263   {
 264     for(MessageBuffer::Iterator bufferIter = buffer->Begin(); bufferIter.IsValid(); bufferIter.Next())
 265     {
 266       MessageBase* message = reinterpret_cast<MessageBase*>(bufferIter.Get());
 267
 268       message->Process(updateBufferIndex);
 269
 270       // Call virtual destructor explictly; since delete will not be called after placement new
 271       message->~MessageBase();
 272     }
 273     buffer->Reset();
 274   }
 275
 276   // Pass back for use in the event-thread
 277   {
 278     MessageQueueMutex::ScopedLock lock(mImpl->queueMutex);
 279     mImpl->recycleQueue.insert(mImpl->recycleQueue.end(),
 280                                std::make_move_iterator(copiedProcessQueue.begin()),
 281                                std::make_move_iterator(copiedProcessQueue.end()));
 282   }
 283
 284   PERF_MONITOR_END(PerformanceMonitor::PROCESS_MESSAGES);
 285
 286   return sceneUpdated;
 287 }
 288
 289 bool MessageQueue::WasEmpty() const
 290 {
 291   return mImpl->queueWasEmpty;
 292 }
 293
 294 bool MessageQueue::IsSceneUpdateRequired() const
 295 {
 296   return mImpl->sceneUpdate;
 297 }
 298
 299 std::size_t MessageQueue::GetCapacity() const
 300 {
 301   MessageQueueMutex::ScopedLock lock(mImpl->queueMutex);
 302
 303   uint32_t          capacity = 0u;
 304   MessageBufferIter endIter  = mImpl->freeQueue.end();
 305   for(MessageBufferIter iter = mImpl->freeQueue.begin(); iter != endIter; ++iter)
 306   {
 307     capacity += (*iter)->GetCapacity();
 308   }
 309   endIter = mImpl->processQueue.end();
 310   for(MessageBufferIter iter = mImpl->processQueue.begin(); iter != endIter; ++iter)
 311   {
 312     capacity += (*iter)->GetCapacity();
 313   }
 314   if(mImpl->currentMessageBuffer != nullptr)
 315   {
 316     capacity += mImpl->currentMessageBuffer->GetCapacity();
 317   }
 318   return capacity;
 319 }
 320
 321 } // namespace Update
 322
 323 } // namespace Internal
 324
 325 } // namespace Dali