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           /*
 191           log->debug(TAG, "Event Processor Loop - event:");
 192
 193           if (event->eventType == 0)
 194             log->debug(TAG, "* eventType: OMX_EventCmdComplete");
 195           else if (event->eventType == 1)
 196             log->debug(TAG, "* eventType: OMX_EventError");
 197           else if (event->eventType == 2)
 198             log->debug(TAG, "* eventType: OMX_EventMark");
 199           else if (event->eventType == 3)
 200             log->debug(TAG, "* eventType: OMX_EventPortSettingsChanged");
 201           else if (event->eventType == 4)
 202             log->debug(TAG, "* eventType: OMX_EventBufferFlag");
 203           else
 204             log->debug(TAG, "* eventType: {:#x}", event->eventType);
 205
 206           if (event->eventType == OMX_EventCmdComplete)
 207           {
 208             if (event->data1 == OMX_CommandStateSet)
 209               log->debug(TAG, "* OMX_COMMANDTYPE: OMX_CommandStateSet");
 210             if (event->data1 == OMX_CommandFlush)
 211               log->debug(TAG, "* OMX_COMMANDTYPE: OMX_CommandFlush");
 212             if (event->data1 == OMX_CommandPortDisable)
 213               log->debug(TAG, "* OMX_COMMANDTYPE: OMX_CommandPortDisable");
 214             if (event->data1 == OMX_CommandPortEnable)
 215               log->debug(TAG, "* OMX_COMMANDTYPE: OMX_CommandPortEnable");
 216             if (event->data1 == OMX_CommandMarkBuffer)
 217               log->debug(TAG, "* OMX_COMMANDTYPE: OMX_CommandMarkBuffer");
 218
 219             if (event->data1 == OMX_CommandStateSet)
 220             {
 221               log->debug(TAG, "* new state: {}", event->data2);
 222             }
 223
 224             if ( (event->data1 == OMX_CommandPortEnable) ||  (event->data1 == OMX_CommandPortDisable) )
 225             {
 226               log->debug(TAG, "* port en/disabled: {}", event->data2);
 227             }
 228           }
 229           else if (event->eventType == OMX_EventBufferFlag)
 230           {
 231             log->debug(TAG, "* Port: {:#x} {}", event->data1, event->data1);
 232           }
 233           else if (event->eventType == OMX_EventError)
 234           {
 235             log->debug(TAG, "* Error code: {:#x}", event->data1);
 236
 237             if (event->data1 == OMX_ErrorPortUnpopulated)
 238               log->debug(TAG, "* = : OMX_ErrorPortUnpopulated");
 239             if (event->data1 == OMX_ErrorInsufficientResources)
 240               log->debug(TAG, "* =: OMX_ErrorInsufficientResources");
 241           }
 242           */
 243
 244         } // end if event is new (then print it out)
 245
 246
 247         // Find matching EventWaiter
 248
 249         eventWaitersMutex.lock();
 250
 251         // Now have EventsMutex and eventWaitersMutex
 252
 253         EventWaiters::iterator eventWaiterIterator;
 254         EventWaiter* eventWaiter{};
 255
 256         for (eventWaiterIterator = eventWaiters.begin(); eventWaiterIterator < eventWaiters.end(); eventWaiterIterator++)
 257         {
 258           // structure here not great yet
 259           // not looking at eventWaiter->eventType??
 260
 261           eventWaiter = *eventWaiterIterator;
 262
 263           if (event->eventType == OMX_EventCmdComplete)
 264           {
 265             if (    ((event->data1 == OMX_CommandPortEnable) && (eventWaiter->command == OMX_CommandPortEnable))
 266                 || ((event->data1 == OMX_CommandPortDisable) && (eventWaiter->command == OMX_CommandPortDisable))  )
 267             {
 268               if (event->data2 == eventWaiter->port)
 269               {
 270                 // Found!
 271                 eventWaiter->receivedEvent = event;
 272                 break;
 273               }
 274             }
 275             else if ((event->data1 == OMX_CommandStateSet) && (eventWaiter->command == OMX_CommandStateSet))
 276             {
 277               if (event->data2 == eventWaiter->newState)
 278               {
 279                 // Found!
 280                 eventWaiter->receivedEvent = event;
 281                 break;
 282               }
 283             }
 284             else if ((event->data1 == OMX_CommandFlush) && (eventWaiter->command == OMX_CommandFlush))
 285             {
 286               if (event->data2 == eventWaiter->flushPort)
 287               {
 288                 // Found!
 289                 eventWaiter->receivedEvent = event;
 290                 break;
 291               }
 292             }
 293           }
 294           else if (event->eventType == OMX_EventPortSettingsChanged)
 295           {
 296             if ( (eventWaiter->eventType == OMX_EventPortSettingsChanged)
 297                 && (event->data1 == eventWaiter->port) )
 298             {
 299               // Found!
 300               eventWaiter->receivedEvent = event;
 301               break;
 302             }
 303           }
 304         } // end of each eventWaiter loop
 305
 306
 307         // Now obv we have an event, but do we have an eventWaiter? If not, advance and continue
 308         if (!eventWaiter || (eventWaiterIterator == eventWaiters.end())) // A) none in the loop at all, B) none matched
 309         {
 310           eventWaitersMutex.unlock();
 311
 312           eventIterator++;
 313           continue;
 314         }
 315
 316         // Now we have an event (event, eventIterator) and a found eventWaiter (eventWaiter, eventWaiterIterator)
 317
 318         // Remove the eventWaiter from the deque
 319         eventWaiters.erase(eventWaiterIterator);
 320         eventWaitersMutex.unlock();
 321
 322         // Remove the event from the deque
 323         eventIterator = events.erase(eventIterator);
 324
 325         // Iterators are now dead.
 326         // Now we have an event and an eventWaiter
 327         // Still have the EventsMutex but we have released the eventWaitersMutex.
 328
 329 //         log->debug(TAG, "This event {} found an event waiter for it {}", static_cast<void*>(event), static_cast<void*>(eventWaiter));
 330
 331
 332         if (eventWaiter->waiting)
 333         {
 334           // FIXME look at whether eventWaiter could miss this signal
 335 //           log->debug(TAG, "Notifying eventWaiter: {} on cond {}", static_cast<void*>(eventWaiter), static_cast<void*>(&eventWaiter->cond));
 336           eventWaiter->cond.notify_one();
 337         }
 338         else
 339         {
 340 //           log->debug(TAG, "EventWaiter not waiting. Deleting {}", static_cast<void*>(eventWaiter));
 341           delete event;
 342           delete eventWaiter;
 343         }
 344
 345 //         log->debug(TAG, "Done processing eventWaiter, goaround events loop");
 346
 347
 348       } // end for each received event
 349
 350       // FIXME If there are events still in events that have not been matched up and sent
 351       // to an eventWaiter, or that were not marked as waiting so we deleted it here, then this is a bug
 352       eventsMutex.unlock();
 353
 354       // Reacquire eventsProcessor lock
 355       ul.lock();
 356     } // end if event processor wake
 357
 358     // locked
 359
 360     eventsProcessorCond.wait(ul, [&] { return eventsProcessorWake; });
 361
 362     // relocked, loop
 363   } // while(1)
 364 }
 365
 366 void OMX::enablePort(OMX_U32 port, bool enable, bool wait)
 367 {
 368   OMX_ERRORTYPE error;
 369
 370   OMX_PARAM_PORTDEFINITIONTYPE pdt;
 371   memset(&pdt, 0, sizeof(pdt));
 372   pdt.nSize = sizeof(pdt);
 373   pdt.nVersion.nVersion = OMX_VERSION;
 374   pdt.nPortIndex = port;
 375   error = OMX_GetParameter(componentHandle, OMX_IndexParamPortDefinition, &pdt);
 376   if (error != OMX_ErrorNone) throw OMX_Exception("OMX_GetParameter in OMX::enablePort()", error);
 377
 378   if (pdt.bEnabled == enable) return; // already in requested state
 379
 380   std::unique_lock<std::mutex> ul(eventWaitersMutex);
 381
 382   OMX_COMMANDTYPE command = enable ? OMX_CommandPortEnable : OMX_CommandPortDisable;
 383
 384 //   log->debug(TAG, "en/disablePort: port: {:#x}, command: {:#x}", port, command);
 385   error = OMX_SendCommand(componentHandle, command, port, 0);
 386   if (error != OMX_ErrorNone) throw OMX_Exception("OMX_SendCommand in OMX::enablePort()", error);
 387 //   log->debug(TAG, "en/disablePort: port: {:#x}, command: {:#x} done", port, command);
 388
 389   EventWaiter* eventWaiter = new EventWaiter();
 390   eventWaiter->waiting = wait;
 391   eventWaiter->command = command;
 392   eventWaiter->port = port;
 393
 394   eventWaiters.push_back(eventWaiter);
 395
 396   if (!wait) return; // && unlock. eventWaiter becomes owned by eventProcessor thread
 397
 398 //   log->debug(TAG, "en/disablePort: Going to wait on cond {}", static_cast<void*>(&eventWaiter->cond));
 399   eventWaiter->cond.wait(ul);  // sleep this thread
 400 //   log->debug(TAG, "en/disablePort: Back from wait on cond {}", static_cast<void*>(&eventWaiter->cond));
 401   ul.unlock();
 402
 403
 404   if (eventWaiter->receivedEvent)
 405   {
 406     // The event processor thread received an event and saved it here
 407 //     log->debug(TAG, "received event!");
 408     delete eventWaiter->receivedEvent;
 409     delete eventWaiter;
 410   }
 411   else
 412   {
 413     delete eventWaiter; // This is ours, at least we can clean this up
 414     throw OMX_Exception("Failed to receive event!", 0);
 415   }
 416 }
 417
 418 void OMX::changeState(OMX_STATETYPE reqState, bool wait)
 419 {
 420   OMX_STATETYPE currentState;
 421   OMX_GetState(componentHandle, &currentState);
 422 //   log->debug(TAG, "Current state: {}", currentState);
 423   if (currentState == reqState)
 424   {
 425 //     log->debug(TAG, "changeState return immediately, already in reqState");
 426     return;
 427   }
 428
 429   std::unique_lock<std::mutex> ul(eventWaitersMutex);
 430   OMX_ERRORTYPE error = OMX_SendCommand(componentHandle, OMX_CommandStateSet, reqState, 0);
 431   if (error != OMX_ErrorNone) throw OMX_Exception("OMX_SendCommand in OMX::changeState()", error);
 432
 433   EventWaiter* eventWaiter = new EventWaiter();
 434   eventWaiter->waiting = wait;
 435   eventWaiter->command = OMX_CommandStateSet;
 436   eventWaiter->newState = reqState;
 437
 438   eventWaiters.push_back(eventWaiter);
 439
 440   if (!wait) return; // && unlock. eventWaiter becomes owned by eventProcessor thread
 441
 442 //   log->debug(TAG, "changeState: Going to wait on cond {}", static_cast<void*>(&eventWaiter->cond));
 443   eventWaiter->cond.wait(ul);  // sleep this thread
 444 //   log->debug(TAG, "changeState: Back from wait on cond {}", static_cast<void*>(&eventWaiter->cond));
 445   ul.unlock();
 446
 447   if (eventWaiter->receivedEvent)
 448   {
 449     // The event processor thread received an event and saved it here
 450 //     log->debug(TAG, "received event!");
 451     delete eventWaiter->receivedEvent;
 452     delete eventWaiter;
 453   }
 454   else
 455   {
 456     delete eventWaiter; // This is ours, at least we can clean this up
 457     throw OMX_Exception("Failed to receive event!", 0);
 458   }
 459 }
 460
 461 void OMX::flushCommands(OMX_U32 port, bool wait)
 462 {
 463   std::unique_lock<std::mutex> ul(eventWaitersMutex);
 464
 465   OMX_ERRORTYPE error = OMX_SendCommand(componentHandle, OMX_CommandFlush, port, NULL);
 466   if (error != OMX_ErrorNone) throw OMX_Exception("OMX_SendCommand in OMX::flushCommands()", error);
 467
 468   EventWaiter* eventWaiter = new EventWaiter();
 469   eventWaiter->waiting = wait;
 470   eventWaiter->command = OMX_CommandFlush;
 471   eventWaiter->flushPort = port;
 472
 473   eventWaiters.push_back(eventWaiter);
 474
 475   if (!wait) return; // && unlock. eventWaiter becomes owned by eventProcessor thread
 476
 477 //   log->debug(TAG, "flushCommands: Going to wait on cond {}", static_cast<void*>(&eventWaiter->cond));
 478   eventWaiter->cond.wait(ul);  // sleep this thread
 479 //   log->debug(TAG, "flushCommands: Back from wait on cond {}", static_cast<void*>(&eventWaiter->cond));
 480   ul.unlock();
 481
 482   if (!eventWaiter->receivedEvent)
 483   {
 484     delete eventWaiter;
 485     throw OMX_Exception("Error waiting for event in flushCommands", 0);
 486   }
 487
 488   // The event processor thread received an event and saved it here
 489 //   log->debug(TAG, "received event!");
 490   delete eventWaiter->receivedEvent;
 491   delete eventWaiter;
 492 }