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

   1 /*
   2  * Copyright (c) 2023 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 void MessageQueue::EventProcessingFinished()
 152 {
 153   mImpl->processingEvents = false; // called from event thread
 154 }
 155
 156 // Called from event thread
 157 uint32_t* MessageQueue::ReserveMessageSlot(uint32_t requestedSize, bool updateScene)
 158 {
 159   DALI_ASSERT_DEBUG(0 != requestedSize);
 160
 161   if(updateScene)
 162   {
 163     mImpl->sceneUpdateFlag = true;
 164   }
 165
 166   if(!mImpl->currentMessageBuffer)
 167   {
 168     const MessageBufferIter endIter = mImpl->freeQueue.end();
 169
 170     // Find the largest recycled buffer from freeQueue
 171     MessageBufferIter nextBuffer = endIter;
 172     for(MessageBufferIter iter = mImpl->freeQueue.begin(); iter != endIter; ++iter)
 173     {
 174       if(endIter == nextBuffer ||
 175          (*nextBuffer)->GetCapacity() < (*iter)->GetCapacity())
 176       {
 177         nextBuffer = iter;
 178       }
 179     }
 180
 181     if(endIter != nextBuffer)
 182     {
 183       // Reuse a recycled buffer from freeQueue
 184       mImpl->currentMessageBuffer = *nextBuffer;
 185       mImpl->freeQueue.erase(nextBuffer);
 186     }
 187     else
 188     {
 189       mImpl->currentMessageBuffer = new MessageBuffer(INITIAL_BUFFER_SIZE);
 190     }
 191   }
 192
 193   // If we are inside Core::ProcessEvents(), core will automatically flush the queue.
 194   // If we are outside, then we have to request a call to Core::ProcessEvents() on idle.
 195   if(false == mImpl->processingEvents)
 196   {
 197     mImpl->renderController.RequestProcessEventsOnIdle();
 198   }
 199
 200   return mImpl->currentMessageBuffer->ReserveMessageSlot(requestedSize);
 201 }
 202
 203 // Called from event thread
 204 bool MessageQueue::FlushQueue()
 205 {
 206   const bool messagesToProcess = (nullptr != mImpl->currentMessageBuffer);
 207
 208   // If there're messages to flush
 209   if(messagesToProcess)
 210   {
 211     // queueMutex must be locked whilst accessing processQueue or recycleQueue
 212     MessageQueueMutex::ScopedLock lock(mImpl->queueMutex);
 213
 214     mImpl->processQueue.push_back(mImpl->currentMessageBuffer);
 215     mImpl->currentMessageBuffer = nullptr;
 216
 217     // Grab any recycled MessageBuffers
 218     while(!mImpl->recycleQueue.empty())
 219     {
 220       MessageBuffer* recycled = mImpl->recycleQueue.back();
 221       mImpl->recycleQueue.pop_back();
 222
 223       // Guard against excessive message buffer growth
 224       if(MAX_FREE_BUFFER_COUNT < mImpl->freeQueue.size() ||
 225          MAX_BUFFER_CAPACITY < recycled->GetCapacity())
 226       {
 227         delete recycled;
 228       }
 229       else
 230       {
 231         mImpl->freeQueue.push_back(recycled);
 232       }
 233     }
 234
 235     if(mImpl->sceneUpdateFlag)
 236     {
 237       mImpl->sceneUpdate |= 2;
 238       mImpl->sceneUpdateFlag = false;
 239     }
 240   }
 241
 242   return messagesToProcess;
 243 }
 244
 245 bool MessageQueue::ProcessMessages(BufferIndex updateBufferIndex)
 246 {
 247   PERF_MONITOR_START(PerformanceMonitor::PROCESS_MESSAGES);
 248
 249   MessageBufferQueue copiedProcessQueue;
 250   bool               sceneUpdated = false;
 251   {
 252     // queueMutex must be locked whilst accessing queue
 253     MessageQueueMutex::ScopedLock lock(mImpl->queueMutex);
 254
 255     mImpl->sceneUpdate >>= 1;
 256     sceneUpdated = (mImpl->sceneUpdate & 0x01); // if it was previously 2, scene graph was updated.
 257
 258     mImpl->queueWasEmpty = mImpl->processQueue.empty(); // Flag whether we processed anything
 259
 260     copiedProcessQueue = std::move(mImpl->processQueue); // Move message queue
 261   }
 262
 263   for(auto&& buffer : copiedProcessQueue)
 264   {
 265     for(MessageBuffer::Iterator bufferIter = buffer->Begin(); bufferIter.IsValid(); bufferIter.Next())
 266     {
 267       MessageBase* message = reinterpret_cast<MessageBase*>(bufferIter.Get());
 268
 269       message->Process(updateBufferIndex);
 270
 271       // Call virtual destructor explictly; since delete will not be called after placement new
 272       message->~MessageBase();
 273     }
 274     buffer->Reset();
 275   }
 276
 277   // Pass back for use in the event-thread
 278   {
 279     MessageQueueMutex::ScopedLock lock(mImpl->queueMutex);
 280     mImpl->recycleQueue.insert(mImpl->recycleQueue.end(),
 281                                std::make_move_iterator(copiedProcessQueue.begin()),
 282                                std::make_move_iterator(copiedProcessQueue.end()));
 283   }
 284
 285   PERF_MONITOR_END(PerformanceMonitor::PROCESS_MESSAGES);
 286
 287   return sceneUpdated;
 288 }
 289
 290 bool MessageQueue::WasEmpty() const
 291 {
 292   return mImpl->queueWasEmpty;
 293 }
 294
 295 bool MessageQueue::IsSceneUpdateRequired() const
 296 {
 297   return mImpl->sceneUpdate;
 298 }
 299
 300 std::size_t MessageQueue::GetCapacity() const
 301 {
 302   MessageQueueMutex::ScopedLock lock(mImpl->queueMutex);
 303
 304   uint32_t          capacity = 0u;
 305   MessageBufferIter endIter  = mImpl->freeQueue.end();
 306   for(MessageBufferIter iter = mImpl->freeQueue.begin(); iter != endIter; ++iter)
 307   {
 308     capacity += (*iter)->GetCapacity();
 309   }
 310   endIter = mImpl->processQueue.end();
 311   for(MessageBufferIter iter = mImpl->processQueue.begin(); iter != endIter; ++iter)
 312   {
 313     capacity += (*iter)->GetCapacity();
 314   }
 315   if(mImpl->currentMessageBuffer != nullptr)
 316   {
 317     capacity += mImpl->currentMessageBuffer->GetCapacity();
 318   }
 319   return capacity;
 320 }
 321
 322 } // namespace Update
 323
 324 } // namespace Internal
 325
 326 } // namespace Dali