omx/omx.cc

   1 /*
   2     Copyright 2021 Chris Tallon
   3
   4     This file is part of VOMP.
   5
   6     VOMP is free software; you can redistribute it and/or modify
   7     it under the terms of the GNU General Public License as published by
   8     the Free Software Foundation; either version 2 of the License, or
   9     (at your option) any later version.
  10
  11     VOMP is distributed in the hope that it will be useful,
  12     but WITHOUT ANY WARRANTY; without even the implied warranty of
  13     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14     GNU General Public License for more details.
  15
  16     You should have received a copy of the GNU General Public License
  17     along with VOMP.  If not, see <https://www.gnu.org/licenses/>.
  18 */
  19
  20 #include "../log.h"
  21 #include "omximagedecode.h"
  22 #include "omxeglrender.h"
  23
  24 #include "omx.h"
  25
  26 static const char* TAG = "OMX";
  27
  28
  29 // Static variable storage
  30 OMX_Image_Decode* OMX::omx_image_decode{};
  31 OMX_EGL_Render*   OMX::omx_egl_render{};
  32
  33 OMX_HANDLETYPE OMX::handle_image_decode{};
  34 OMX_HANDLETYPE OMX::handle_egl_render{};
  35
  36 std::thread OMX::eventsProcessorThread;
  37 std::mutex OMX::eventsProcessorMutex;
  38 std::condition_variable OMX::eventsProcessorCond;
  39 std::mutex OMX::eventsMutex;
  40 Events OMX::events;
  41 bool OMX::eventsProcessorWake{};
  42 bool OMX::eventsProcessorShutdownNow{};
  43 int OMX::eventsProcessorUsageCount{};
  44 EventWaiters OMX::eventWaiters;
  45 std::mutex OMX::eventWaitersMutex;
  46
  47
  48 // Main static callbacks
  49
  50 OMX_ERRORTYPE OMX::scb_EventHandler(
  51   OMX_IN OMX_HANDLETYPE handle, OMX_IN OMX_PTR appdata, OMX_IN OMX_EVENTTYPE event_type,
  52   OMX_IN OMX_U32 data1, OMX_IN OMX_U32 data2, OMX_IN OMX_PTR event_data)
  53 {
  54   LogNT::getInstance()->debug("OMXSCB", "EventHandler {} {} {} {}", static_cast<void*>(handle_image_decode), static_cast<void*>(omx_image_decode), static_cast<void*>(handle_egl_render), static_cast<void*>(omx_egl_render));
  55   if (handle == handle_image_decode)
  56     return omx_image_decode->cb_EventHandler(handle, appdata, event_type, data1, data2, event_data);
  57   else if (handle == handle_egl_render)
  58     return omx_egl_render->cb_EventHandler(handle, appdata, event_type, data1, data2, event_data);
  59
  60   return OMX_ErrorNone;
  61 }
  62
  63 OMX_ERRORTYPE OMX::scb_EmptyBufferDone(OMX_IN OMX_HANDLETYPE handle, OMX_IN OMX_PTR appdata, OMX_IN OMX_BUFFERHEADERTYPE* buffer)
  64 {
  65   LogNT::getInstance()->debug("OMXSCB", "EmptyBufferDone {} {} {} {}", static_cast<void*>(handle_image_decode), static_cast<void*>(omx_image_decode), static_cast<void*>(handle_egl_render), static_cast<void*>(omx_egl_render));
  66   if (handle == handle_image_decode)
  67     return omx_image_decode->cb_EmptyBufferDone(handle, appdata, buffer);
  68   else if (handle == handle_egl_render)
  69     return omx_egl_render->cb_EmptyBufferDone(handle, appdata, buffer);
  70
  71   return OMX_ErrorNone;
  72 }
  73
  74 OMX_ERRORTYPE OMX::scb_FillBufferDone(OMX_IN OMX_HANDLETYPE handle, OMX_IN OMX_PTR appdata, OMX_IN OMX_BUFFERHEADERTYPE* buffer)
  75 {
  76   LogNT::getInstance()->debug("OMXSCB", "FillBufferDone {} {} {} {}", static_cast<void*>(handle_image_decode), static_cast<void*>(omx_image_decode), static_cast<void*>(handle_egl_render), static_cast<void*>(omx_egl_render));
  77   if (handle == handle_image_decode)
  78     return omx_image_decode->cb_FillBufferDone(handle, appdata, buffer);
  79   else if (handle == handle_egl_render)
  80     return omx_egl_render->cb_FillBufferDone(handle, appdata, buffer);
  81   return OMX_ErrorNone;
  82 }
  83
  84 // ---
  85
  86 OMX::OMX()
  87 {
  88   log = LogNT::getInstance();
  89 }
  90
  91 OMX::~OMX()
  92 {
  93 }
  94
  95 // bool OMX::OMX_Master_Init()
  96 // {
  97 //   OMX_ERRORTYPE error;
  98 //   error = OMX_Init();
  99 //   if (error != OMX_ErrorNone) return false;
 100 //   return true;
 101 // }
 102
 103 OMX_ERRORTYPE OMX::cb_EventHandler(OMX_IN OMX_HANDLETYPE handle, OMX_IN OMX_PTR appdata, OMX_IN OMX_EVENTTYPE event_type, OMX_IN OMX_U32 data1, OMX_IN OMX_U32 data2, OMX_IN OMX_PTR event_data)
 104 {
 105   //Log* l = Log::getInstance();
 106   log->debug(TAG, "CB: eventHandler {} {} {:#x} {:#x} {}", static_cast<void*>(handle), event_type, data1, data2, static_cast<void*>(event_data));
 107
 108   eventsMutex.lock();
 109   Event* incomingEvent = new Event{.appdata = appdata, .eventType = event_type, .data1 = data1, .data2 = data2, .event_data = event_data};
 110   events.push_back(incomingEvent);
 111   eventsMutex.unlock();
 112
 113   eventsProcessorMutex.lock();
 114   eventsProcessorWake = true;
 115   eventsProcessorMutex.unlock();
 116   eventsProcessorCond.notify_one();
 117
 118   log->debug(TAG, "eventHandler end");
 119   return OMX_ErrorNone;
 120 }
 121
 122 void OMX::initEventsProcessing()
 123 {
 124   eventsProcessorMutex.lock();
 125
 126   if (eventsProcessorUsageCount == 0)
 127   {
 128     eventsProcessorUsageCount = 1;
 129
 130     eventsProcessorThread = std::thread( [&]
 131     {
 132       eventsProcessorMutex.lock();
 133       eventsProcessorMutex.unlock();
 134       eventsProcessorLoop();
 135     });
 136   }
 137   else
 138   {
 139     ++eventsProcessorUsageCount;
 140   }
 141
 142   eventsProcessorMutex.unlock();
 143 }
 144
 145 void OMX::stopEventsProcessing()
 146 {
 147   eventsProcessorMutex.lock();
 148
 149   --eventsProcessorUsageCount;
 150
 151   if (eventsProcessorUsageCount == 0)
 152   {
 153     // stop event loop
 154     eventsProcessorWake = true;
 155     eventsProcessorShutdownNow = true;
 156     eventsProcessorCond.notify_one();
 157     eventsProcessorMutex.unlock();
 158     eventsProcessorThread.join();
 159   }
 160   else
 161   {
 162     eventsProcessorMutex.unlock();
 163   }
 164 }
 165
 166 void OMX::eventsProcessorLoop()
 167 {
 168   LogNT* log = LogNT::getInstance();
 169   std::unique_lock<std::mutex> ul(eventsProcessorMutex); // locks
 170
 171   while(1)
 172   {
 173     // locked
 174     if (eventsProcessorWake)
 175     {
 176       eventsProcessorWake = false;
 177       ul.unlock();
 178
 179       if (eventsProcessorShutdownNow) return;
 180
 181       eventsMutex.lock();
 182       for (Events::iterator eventIterator = events.begin(); eventIterator != events.end(); )  // for each event received from OMX
 183       {
 184         Event* event = *eventIterator;
 185
 186         if (event->isNew)
 187         {
 188           event->isNew = false;
 189
 190           log->debug(TAG, "Event Processor Loop - event:");
 191
 192           if (event->eventType == 0)
 193             log->debug(TAG, "* eventType: OMX_EventCmdComplete");
 194           else if (event->eventType == 1)
 195             log->debug(TAG, "* eventType: OMX_EventError");
 196           else if (event->eventType == 2)
 197             log->debug(TAG, "* eventType: OMX_EventMark");
 198           else if (event->eventType == 3)
 199             log->debug(TAG, "* eventType: OMX_EventPortSettingsChanged");
 200           else if (event->eventType == 4)
 201             log->debug(TAG, "* eventType: OMX_EventBufferFlag");
 202           else
 203             log->debug(TAG, "* eventType: {:#x}", event->eventType);
 204
 205           if (event->eventType == OMX_EventCmdComplete)
 206           {
 207             if (event->data1 == OMX_CommandStateSet)
 208               log->debug(TAG, "* OMX_COMMANDTYPE: OMX_CommandStateSet");
 209             if (event->data1 == OMX_CommandFlush)
 210               log->debug(TAG, "* OMX_COMMANDTYPE: OMX_CommandFlush");
 211             if (event->data1 == OMX_CommandPortDisable)
 212               log->debug(TAG, "* OMX_COMMANDTYPE: OMX_CommandPortDisable");
 213             if (event->data1 == OMX_CommandPortEnable)
 214               log->debug(TAG, "* OMX_COMMANDTYPE: OMX_CommandPortEnable");
 215             if (event->data1 == OMX_CommandMarkBuffer)
 216               log->debug(TAG, "* OMX_COMMANDTYPE: OMX_CommandMarkBuffer");
 217
 218             if (event->data1 == OMX_CommandStateSet)
 219             {
 220               log->debug(TAG, "* new state: {}", event->data2);
 221             }
 222
 223             if ( (event->data1 == OMX_CommandPortEnable) ||  (event->data1 == OMX_CommandPortDisable) )
 224             {
 225               log->debug(TAG, "* port en/disabled: {}", event->data2);
 226             }
 227           }
 228           else if (event->eventType == OMX_EventBufferFlag)
 229           {
 230             log->debug(TAG, "* Port: {:#x} {}", event->data1, event->data1);
 231           }
 232           else if (event->eventType == OMX_EventError)
 233           {
 234             log->debug(TAG, "* Error code: {:#x}", event->data1);
 235
 236             if (event->data1 == OMX_ErrorPortUnpopulated)
 237               log->debug(TAG, "* = : OMX_ErrorPortUnpopulated");
 238             if (event->data1 == OMX_ErrorInsufficientResources)
 239               log->debug(TAG, "* =: OMX_ErrorInsufficientResources");
 240           }
 241         } // end if event is new (then print it out)
 242
 243
 244         // Find matching EventWaiter
 245
 246         eventWaitersMutex.lock();
 247
 248         // Now have EventsMutex and eventWaitersMutex
 249
 250         EventWaiters::iterator eventWaiterIterator;
 251         EventWaiter* eventWaiter{};
 252
 253         for (eventWaiterIterator = eventWaiters.begin(); eventWaiterIterator < eventWaiters.end(); eventWaiterIterator++)
 254         {
 255           // structure here not great yet
 256           // not looking at eventWaiter->eventType??
 257
 258           eventWaiter = *eventWaiterIterator;
 259
 260           if (event->eventType == OMX_EventCmdComplete)
 261           {
 262             if (    ((event->data1 == OMX_CommandPortEnable) && (eventWaiter->command == OMX_CommandPortEnable))
 263                 || ((event->data1 == OMX_CommandPortDisable) && (eventWaiter->command == OMX_CommandPortDisable))  )
 264             {
 265               if (event->data2 == eventWaiter->port)
 266               {
 267                 // Found!
 268                 eventWaiter->receivedEvent = event;
 269                 break;
 270               }
 271             }
 272             else if ((event->data1 == OMX_CommandStateSet) && (eventWaiter->command == OMX_CommandStateSet))
 273             {
 274               if (event->data2 == eventWaiter->newState)
 275               {
 276                 // Found!
 277                 eventWaiter->receivedEvent = event;
 278                 break;
 279               }
 280             }
 281             else if ((event->data1 == OMX_CommandFlush) && (eventWaiter->command == OMX_CommandFlush))
 282             {
 283               if (event->data2 == eventWaiter->flushPort)
 284               {
 285                 // Found!
 286                 eventWaiter->receivedEvent = event;
 287                 break;
 288               }
 289             }
 290           }
 291           else if (event->eventType == OMX_EventPortSettingsChanged)
 292           {
 293             if ( (eventWaiter->eventType == OMX_EventPortSettingsChanged)
 294                 && (event->data1 == eventWaiter->port) )
 295             {
 296               // Found!
 297               eventWaiter->receivedEvent = event;
 298               break;
 299             }
 300           }
 301         } // end of each eventWaiter loop
 302
 303
 304         // Now obv we have an event, but do we have an eventWaiter? If not, advance and continue
 305         if (!eventWaiter || (eventWaiterIterator == eventWaiters.end())) // A) none in the loop at all, B) none matched
 306         {
 307           eventWaitersMutex.unlock();
 308
 309           eventIterator++;
 310           continue;
 311         }
 312
 313         // Now we have an event (event, eventIterator) and a found eventWaiter (eventWaiter, eventWaiterIterator)
 314
 315         // Remove the eventWaiter from the deque
 316         eventWaiters.erase(eventWaiterIterator);
 317         eventWaitersMutex.unlock();
 318
 319         // Remove the event from the deque
 320         eventIterator = events.erase(eventIterator);
 321
 322         // Iterators are now dead.
 323         // Now we have an event and an eventWaiter
 324         // Still have the EventsMutex but we have released the eventWaitersMutex.
 325
 326         log->debug(TAG, "This event {} found an event waiter for it {}", static_cast<void*>(event), static_cast<void*>(eventWaiter));
 327
 328
 329         if (eventWaiter->waiting)
 330         {
 331           // FIXME look at whether eventWaiter could miss this signal
 332           log->debug(TAG, "Notifying eventWaiter: {} on cond {}", static_cast<void*>(eventWaiter), static_cast<void*>(&eventWaiter->cond));
 333           eventWaiter->cond.notify_one();
 334         }
 335         else
 336         {
 337           log->debug(TAG, "EventWaiter not waiting. Deleting {}", static_cast<void*>(eventWaiter));
 338           delete event;
 339           delete eventWaiter;
 340         }
 341
 342         log->debug(TAG, "Done processing eventWaiter, goaround events loop");
 343
 344
 345       } // end for each received event
 346
 347       // FIXME If there are events still in events that have not been matched up and sent
 348       // to an eventWaiter, or that were not marked as waiting so we deleted it here, then this is a bug
 349       eventsMutex.unlock();
 350
 351       // Reacquire eventsProcessor lock
 352       ul.lock();
 353     } // end if event processor wake
 354
 355     // locked
 356
 357     eventsProcessorCond.wait(ul, [&] { return eventsProcessorWake; });
 358
 359     // relocked, loop
 360   } // while(1)
 361 }
 362
 363 void OMX::enablePort(OMX_U32 port, bool enable, bool wait)
 364 {
 365   OMX_ERRORTYPE error;
 366
 367   OMX_PARAM_PORTDEFINITIONTYPE pdt;
 368   memset(&pdt, 0, sizeof(pdt));
 369   pdt.nSize = sizeof(pdt);
 370   pdt.nVersion.nVersion = OMX_VERSION;
 371   pdt.nPortIndex = port;
 372   error = OMX_GetParameter(componentHandle, OMX_IndexParamPortDefinition, &pdt);
 373   if (error != OMX_ErrorNone) throw OMX_Exception("OMX_GetParameter in OMX::enablePort()", error);
 374
 375   if (pdt.bEnabled == enable) return; // already in requested state
 376
 377   std::unique_lock<std::mutex> ul(eventWaitersMutex);
 378
 379   OMX_COMMANDTYPE command = enable ? OMX_CommandPortEnable : OMX_CommandPortDisable;
 380
 381   log->debug(TAG, "en/disablePort: port: {:#x}, command: {:#x}", port, command);
 382   error = OMX_SendCommand(componentHandle, command, port, 0);
 383   if (error != OMX_ErrorNone) throw OMX_Exception("OMX_SendCommand in OMX::enablePort()", error);
 384   log->debug(TAG, "en/disablePort: port: {:#x}, command: {:#x} done", port, command);
 385
 386   EventWaiter* eventWaiter = new EventWaiter();
 387   eventWaiter->waiting = wait;
 388   eventWaiter->command = command;
 389   eventWaiter->port = port;
 390
 391   eventWaiters.push_back(eventWaiter);
 392
 393   if (!wait) return; // && unlock. eventWaiter becomes owned by eventProcessor thread
 394
 395   log->debug(TAG, "en/disablePort: Going to wait on cond {}", static_cast<void*>(&eventWaiter->cond));
 396   eventWaiter->cond.wait(ul);  // sleep this thread
 397   log->debug(TAG, "en/disablePort: Back from wait on cond {}", static_cast<void*>(&eventWaiter->cond));
 398   ul.unlock();
 399
 400
 401   if (eventWaiter->receivedEvent)
 402   {
 403     // The event processor thread received an event and saved it here
 404     log->debug(TAG, "received event!");
 405     delete eventWaiter->receivedEvent;
 406     delete eventWaiter;
 407   }
 408   else
 409   {
 410     delete eventWaiter; // This is ours, at least we can clean this up
 411     throw OMX_Exception("Failed to receive event!", 0);
 412   }
 413 }
 414
 415 void OMX::changeState(OMX_STATETYPE reqState, bool wait)
 416 {
 417   OMX_STATETYPE currentState;
 418   OMX_GetState(componentHandle, &currentState);
 419   log->debug(TAG, "Current state: {}", currentState);
 420   if (currentState == reqState)
 421   {
 422     log->debug(TAG, "changeState return immediately, already in reqState");
 423     return;
 424   }
 425
 426   std::unique_lock<std::mutex> ul(eventWaitersMutex);
 427   OMX_ERRORTYPE error = OMX_SendCommand(componentHandle, OMX_CommandStateSet, reqState, 0);
 428   if (error != OMX_ErrorNone) throw OMX_Exception("OMX_SendCommand in OMX::changeState()", error);
 429
 430   EventWaiter* eventWaiter = new EventWaiter();
 431   eventWaiter->waiting = wait;
 432   eventWaiter->command = OMX_CommandStateSet;
 433   eventWaiter->newState = reqState;
 434
 435   eventWaiters.push_back(eventWaiter);
 436
 437   if (!wait) return; // && unlock. eventWaiter becomes owned by eventProcessor thread
 438
 439   log->debug(TAG, "changeState: Going to wait on cond {}", static_cast<void*>(&eventWaiter->cond));
 440   eventWaiter->cond.wait(ul);  // sleep this thread
 441   log->debug(TAG, "changeState: Back from wait on cond {}", static_cast<void*>(&eventWaiter->cond));
 442   ul.unlock();
 443
 444   if (eventWaiter->receivedEvent)
 445   {
 446     // The event processor thread received an event and saved it here
 447     log->debug(TAG, "received event!");
 448     delete eventWaiter->receivedEvent;
 449     delete eventWaiter;
 450   }
 451   else
 452   {
 453     delete eventWaiter; // This is ours, at least we can clean this up
 454     throw OMX_Exception("Failed to receive event!", 0);
 455   }
 456 }
 457
 458 void OMX::flushCommands(OMX_U32 port, bool wait)
 459 {
 460   std::unique_lock<std::mutex> ul(eventWaitersMutex);
 461
 462   OMX_ERRORTYPE error = OMX_SendCommand(componentHandle, OMX_CommandFlush, port, NULL);
 463   if (error != OMX_ErrorNone) throw OMX_Exception("OMX_SendCommand in OMX::flushCommands()", error);
 464
 465   EventWaiter* eventWaiter = new EventWaiter();
 466   eventWaiter->waiting = wait;
 467   eventWaiter->command = OMX_CommandFlush;
 468   eventWaiter->flushPort = port;
 469
 470   eventWaiters.push_back(eventWaiter);
 471
 472   if (!wait) return; // && unlock. eventWaiter becomes owned by eventProcessor thread
 473
 474   log->debug(TAG, "flushCommands: Going to wait on cond {}", static_cast<void*>(&eventWaiter->cond));
 475   eventWaiter->cond.wait(ul);  // sleep this thread
 476   log->debug(TAG, "flushCommands: Back from wait on cond {}", static_cast<void*>(&eventWaiter->cond));
 477   ul.unlock();
 478
 479   if (!eventWaiter->receivedEvent)
 480   {
 481     delete eventWaiter;
 482     throw OMX_Exception("Error waiting for event in flushCommands", 0);
 483   }
 484
 485   // The event processor thread received an event and saved it here
 486   log->debug(TAG, "received event!");
 487   delete eventWaiter->receivedEvent;
 488   delete eventWaiter;
 489 }