src/lib/adapter/Pulse-Eight/USBCECAdapterMessageQueue.cpp

   1 /*
   2  * This file is part of the libCEC(R) library.
   3  *
   4  * libCEC(R) is Copyright (C) 2011-2012 Pulse-Eight Limited.  All rights reserved.
   5  * libCEC(R) is an original work, containing original code.
   6  *
   7  * libCEC(R) is a trademark of Pulse-Eight Limited.
   8  *
   9  * This program is dual-licensed; you can redistribute it and/or modify
  10  * it under the terms of the GNU General Public License as published by
  11  * the Free Software Foundation; either version 2 of the License, or
  12  * (at your option) any later version.
  13  *
  14  * This program is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17  * GNU General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU General Public License
  20  * along with this program; if not, write to the Free Software
  21  * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
  22  *
  23  *
  24  * Alternatively, you can license this library under a commercial license,
  25  * please contact Pulse-Eight Licensing for more information.
  26  *
  27  * For more information contact:
  28  * Pulse-Eight Licensing       <license@pulse-eight.com>
  29  *     http://www.pulse-eight.com/
  30  *     http://www.pulse-eight.net/
  31  */
  32
  33 #include "env.h"
  34 #include "USBCECAdapterMessageQueue.h"
  35
  36 #include "USBCECAdapterCommunication.h"
  37 #include "USBCECAdapterMessage.h"
  38 #include "lib/platform/sockets/socket.h"
  39 #include "lib/LibCEC.h"
  40 #include "lib/platform/util/StdString.h"
  41
  42 using namespace CEC;
  43 using namespace PLATFORM;
  44 using namespace std;
  45
  46 #define MESSAGE_QUEUE_SIGNAL_WAIT_TIME 1000
  47
  48 CCECAdapterMessageQueueEntry::CCECAdapterMessageQueueEntry(CCECAdapterMessageQueue *queue, CCECAdapterMessage *message) :
  49     m_queue(queue),
  50     m_message(message),
  51     m_iPacketsLeft(message->IsTranmission() ? message->Size() / 4 : 1),
  52     m_bSucceeded(false),
  53     m_bWaiting(true),
  54     m_queueTimeout(message->transmit_timeout) {}
  55
  56 CCECAdapterMessageQueueEntry::~CCECAdapterMessageQueueEntry(void) { }
  57
  58 void CCECAdapterMessageQueueEntry::Broadcast(void)
  59 {
  60   CLockObject lock(m_mutex);
  61   m_condition.Broadcast();
  62 }
  63
  64 bool CCECAdapterMessageQueueEntry::MessageReceived(const CCECAdapterMessage &message)
  65 {
  66   bool bHandled(false);
  67
  68   if (IsResponse(message))
  69   {
  70     switch (message.Message())
  71     {
  72     case MSGCODE_COMMAND_ACCEPTED:
  73       bHandled = MessageReceivedCommandAccepted(message);
  74       break;
  75     case MSGCODE_TRANSMIT_SUCCEEDED:
  76       bHandled = MessageReceivedTransmitSucceeded(message);
  77       break;
  78     default:
  79       bHandled = MessageReceivedResponse(message);
  80       break;
  81     }
  82   }
  83
  84   return bHandled;
  85 }
  86
  87 void CCECAdapterMessageQueueEntry::Signal(void)
  88 {
  89   CLockObject lock(m_mutex);
  90   m_bSucceeded = true;
  91   m_condition.Signal();
  92 }
  93
  94 bool CCECAdapterMessageQueueEntry::Wait(uint32_t iTimeout)
  95 {
  96   bool bReturn(false);
  97   /* wait until we receive a signal when the tranmission succeeded */
  98   {
  99     CLockObject lock(m_mutex);
 100     bReturn = m_bSucceeded ? true : m_condition.Wait(m_mutex, m_bSucceeded, iTimeout);
 101     m_bWaiting = false;
 102   }
 103   return bReturn;
 104 }
 105
 106 bool CCECAdapterMessageQueueEntry::IsWaiting(void)
 107 {
 108   CLockObject lock(m_mutex);
 109   return m_bWaiting;
 110 }
 111
 112 cec_adapter_messagecode CCECAdapterMessageQueueEntry::MessageCode(void)
 113 {
 114   return m_message->Message();
 115 }
 116
 117 bool CCECAdapterMessageQueueEntry::IsResponseOld(const CCECAdapterMessage &msg)
 118 {
 119   cec_adapter_messagecode msgCode = msg.Message();
 120
 121   return msgCode == MessageCode() ||
 122          msgCode == MSGCODE_COMMAND_ACCEPTED ||
 123          msgCode == MSGCODE_COMMAND_REJECTED ||
 124          (m_message->IsTranmission() && (msgCode == MSGCODE_TIMEOUT_ERROR ||
 125              msgCode == MSGCODE_HIGH_ERROR ||
 126              msgCode == MSGCODE_LOW_ERROR ||
 127              msgCode == MSGCODE_RECEIVE_FAILED ||
 128              msgCode == MSGCODE_TRANSMIT_FAILED_LINE ||
 129              msgCode == MSGCODE_TRANSMIT_FAILED_ACK ||
 130              msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_DATA ||
 131              msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_LINE ||
 132              msgCode == MSGCODE_TRANSMIT_SUCCEEDED));
 133 }
 134
 135 bool CCECAdapterMessageQueueEntry::IsResponse(const CCECAdapterMessage &msg)
 136 {
 137   if (m_message->state == ADAPTER_MESSAGE_STATE_SENT_ACKED)
 138     return false;
 139
 140   cec_adapter_messagecode thisMsgCode = m_message->Message();
 141   cec_adapter_messagecode msgCode = msg.Message();
 142   cec_adapter_messagecode msgResponse = msg.ResponseTo();
 143
 144   // msgcode matches, always a response
 145   if (msgCode == MessageCode())
 146     return true;
 147
 148   if (!ProvidesExtendedResponse())
 149     return IsResponseOld(msg);
 150
 151   // response without a msgcode
 152   if (msgResponse == MSGCODE_NOTHING)
 153     return false;
 154
 155   // commands that only repond with accepted/rejected
 156   if (thisMsgCode == MSGCODE_PING ||
 157       thisMsgCode == MSGCODE_SET_ACK_MASK ||
 158       thisMsgCode == MSGCODE_SET_CONTROLLED ||
 159       thisMsgCode == MSGCODE_SET_AUTO_ENABLED ||
 160       thisMsgCode == MSGCODE_SET_DEFAULT_LOGICAL_ADDRESS ||
 161       thisMsgCode == MSGCODE_SET_LOGICAL_ADDRESS_MASK ||
 162       thisMsgCode == MSGCODE_SET_PHYSICAL_ADDRESS ||
 163       thisMsgCode == MSGCODE_SET_DEVICE_TYPE ||
 164       thisMsgCode == MSGCODE_SET_HDMI_VERSION ||
 165       thisMsgCode == MSGCODE_SET_OSD_NAME ||
 166       thisMsgCode == MSGCODE_WRITE_EEPROM ||
 167       thisMsgCode == MSGCODE_TRANSMIT_IDLETIME ||
 168       thisMsgCode == MSGCODE_SET_ACTIVE_SOURCE)
 169     return thisMsgCode == msgResponse;
 170
 171   if (!m_message->IsTranmission())
 172   {
 173     m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_WARNING, "FIXME! not a transmission: %s", msg.ToString().c_str());
 174     return false;
 175   }
 176
 177   return ((msgCode == MSGCODE_COMMAND_ACCEPTED || msgCode == MSGCODE_COMMAND_REJECTED) &&
 178       (msgResponse == MSGCODE_TRANSMIT_ACK_POLARITY || msgResponse == MSGCODE_TRANSMIT || msgResponse == MSGCODE_TRANSMIT_EOM)) ||
 179       msgCode == MSGCODE_TIMEOUT_ERROR ||
 180       msgCode == MSGCODE_RECEIVE_FAILED ||
 181       msgCode == MSGCODE_TRANSMIT_FAILED_ACK ||
 182       msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_DATA ||
 183       msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_LINE ||
 184       msgCode == MSGCODE_TRANSMIT_SUCCEEDED;
 185 }
 186
 187 const char *CCECAdapterMessageQueueEntry::ToString(void) const
 188 {
 189   /* CEC transmissions got the 'set ack polarity' msgcode, which doesn't look nice */
 190   if (m_message->IsTranmission())
 191     return "CEC transmission";
 192   else
 193     return CCECAdapterMessage::ToString(m_message->Message());
 194 }
 195
 196 bool CCECAdapterMessageQueueEntry::MessageReceivedCommandAccepted(const CCECAdapterMessage &message)
 197 {
 198   bool bSendSignal(false);
 199   bool bHandled(false);
 200   {
 201     CLockObject lock(m_mutex);
 202     if (m_iPacketsLeft > 0)
 203     {
 204       /* decrease by 1 */
 205       m_iPacketsLeft--;
 206
 207 #ifdef CEC_DEBUGGING
 208       /* log this message */
 209       CStdString strLog;
 210       strLog.Format("%s - command accepted", ToString());
 211       if (m_iPacketsLeft > 0)
 212         strLog.AppendFormat(" - waiting for %d more", m_iPacketsLeft);
 213       m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, strLog);
 214 #endif
 215
 216       /* no more packets left and not a transmission, so we're done */
 217       if (!m_message->IsTranmission() && m_iPacketsLeft == 0)
 218       {
 219         m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
 220         m_message->response = message.packet;
 221         bSendSignal = true;
 222       }
 223       bHandled = true;
 224     }
 225   }
 226
 227   if (bSendSignal)
 228     Signal();
 229
 230   return bHandled;
 231 }
 232
 233 bool CCECAdapterMessageQueueEntry::MessageReceivedTransmitSucceeded(const CCECAdapterMessage &message)
 234 {
 235   {
 236     CLockObject lock(m_mutex);
 237     if (m_iPacketsLeft == 0)
 238     {
 239       /* transmission succeeded, so we're done */
 240 #ifdef CEC_DEBUGGING
 241       m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "%s - transmit succeeded", m_message->ToString().c_str());
 242 #endif
 243       m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
 244       m_message->response = message.packet;
 245     }
 246     else
 247     {
 248       /* error, we expected more acks
 249          since the messages are processed in order, this should not happen, so this is an error situation */
 250       m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_WARNING, "%s - received 'transmit succeeded' but not enough 'command accepted' messages (%d left)", ToString(), m_iPacketsLeft);
 251       m_message->state = ADAPTER_MESSAGE_STATE_ERROR;
 252     }
 253   }
 254
 255   Signal();
 256
 257   return true;
 258 }
 259
 260 bool CCECAdapterMessageQueueEntry::MessageReceivedResponse(const CCECAdapterMessage &message)
 261 {
 262   {
 263     CLockObject lock(m_mutex);
 264 #ifdef CEC_DEBUGGING
 265     m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "%s - received response - %s", ToString(), message.ToString().c_str());
 266 #else
 267     if (message.IsError())
 268       m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "%s - received response - %s", ToString(), message.ToString().c_str());
 269 #endif
 270     m_message->response = message.packet;
 271     if (m_message->IsTranmission())
 272       m_message->state = message.Message() == MSGCODE_TRANSMIT_SUCCEEDED ? ADAPTER_MESSAGE_STATE_SENT_ACKED : ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED;
 273     else
 274       m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
 275   }
 276
 277   Signal();
 278
 279   return true;
 280 }
 281
 282 bool CCECAdapterMessageQueueEntry::ProvidesExtendedResponse(void)
 283 {
 284   return m_queue && m_queue->ProvidesExtendedResponse();
 285 }
 286
 287 bool CCECAdapterMessageQueueEntry::TimedOutOrSucceeded(void) const
 288 {
 289   return m_message->bFireAndForget && (m_bSucceeded || m_queueTimeout.TimeLeft() == 0);
 290 }
 291
 292 CCECAdapterMessageQueue::CCECAdapterMessageQueue(CUSBCECAdapterCommunication *com) :
 293   PLATFORM::CThread(),
 294   m_com(com),
 295   m_iNextMessage(0)
 296 {
 297   m_incomingAdapterMessage = new CCECAdapterMessage;
 298   m_currentCECFrame.Clear();
 299 }
 300
 301 CCECAdapterMessageQueue::~CCECAdapterMessageQueue(void)
 302 {
 303   StopThread(-1);
 304   Clear();
 305   StopThread();
 306   delete m_incomingAdapterMessage;
 307 }
 308
 309 void CCECAdapterMessageQueue::Clear(void)
 310 {
 311   StopThread(5);
 312   CLockObject lock(m_mutex);
 313   m_writeQueue.Clear();
 314   m_messages.clear();
 315 }
 316
 317 void *CCECAdapterMessageQueue::Process(void)
 318 {
 319   CCECAdapterMessageQueueEntry *message(NULL);
 320   while (!IsStopped())
 321   {
 322     /* wait for a new message */
 323     if (m_writeQueue.Pop(message, MESSAGE_QUEUE_SIGNAL_WAIT_TIME) && message)
 324     {
 325       /* write this message */
 326       {
 327         CLockObject lock(m_mutex);
 328         m_com->WriteToDevice(message->m_message);
 329       }
 330       if (message->m_message->state == ADAPTER_MESSAGE_STATE_ERROR ||
 331           message->m_message->Message() == MSGCODE_START_BOOTLOADER)
 332       {
 333         message->Signal();
 334         Clear();
 335         break;
 336       }
 337     }
 338
 339     CheckTimedOutMessages();
 340   }
 341   return NULL;
 342 }
 343
 344 void CCECAdapterMessageQueue::CheckTimedOutMessages(void)
 345 {
 346   CLockObject lock(m_mutex);
 347   vector<uint64_t> timedOut;
 348   for (map<uint64_t, CCECAdapterMessageQueueEntry *>::iterator it = m_messages.begin(); it != m_messages.end(); it++)
 349   {
 350     if (it->second->TimedOutOrSucceeded())
 351     {
 352       timedOut.push_back(it->first);
 353       if (!it->second->m_bSucceeded)
 354         m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "command '%s' was not acked by the controller", CCECAdapterMessage::ToString(it->second->m_message->Message()));
 355       delete it->second->m_message;
 356       delete it->second;
 357     }
 358   }
 359
 360   for (vector<uint64_t>::iterator it = timedOut.begin(); it != timedOut.end(); it++)
 361   {
 362     uint64_t iEntryId = *it;
 363     m_messages.erase(iEntryId);
 364   }
 365 }
 366
 367 void CCECAdapterMessageQueue::MessageReceived(const CCECAdapterMessage &msg)
 368 {
 369   bool bHandled(false);
 370   CLockObject lock(m_mutex);
 371   /* send the received message to each entry in the queue until it is handled */
 372   for (map<uint64_t, CCECAdapterMessageQueueEntry *>::iterator it = m_messages.begin(); !bHandled && it != m_messages.end(); it++)
 373     bHandled = it->second->MessageReceived(msg);
 374
 375   if (!bHandled)
 376   {
 377     /* the message wasn't handled */
 378     bool bIsError(m_com->HandlePoll(msg));
 379 #ifdef CEC_DEBUGGING
 380     m_com->m_callback->GetLib()->AddLog(bIsError ? CEC_LOG_WARNING : CEC_LOG_DEBUG, msg.ToString().c_str());
 381 #else
 382     if (bIsError)
 383       m_com->m_callback->GetLib()->AddLog(CEC_LOG_WARNING, msg.ToString().c_str());
 384 #endif
 385
 386     /* push this message to the current frame */
 387     if (!bIsError && msg.PushToCecCommand(m_currentCECFrame))
 388     {
 389       /* and push the current frame back over the callback method when a full command was received */
 390       if (m_com->IsInitialised())
 391         m_com->m_callback->OnCommandReceived(m_currentCECFrame);
 392
 393       /* clear the current frame */
 394       m_currentCECFrame.Clear();
 395     }
 396   }
 397 }
 398
 399 void CCECAdapterMessageQueue::AddData(uint8_t *data, size_t iLen)
 400 {
 401   for (size_t iPtr = 0; iPtr < iLen; iPtr++)
 402   {
 403     bool bFullMessage(false);
 404     {
 405       CLockObject lock(m_mutex);
 406       bFullMessage = m_incomingAdapterMessage->PushReceivedByte(data[iPtr]);
 407     }
 408
 409     if (bFullMessage)
 410     {
 411       /* a full message was received */
 412       CCECAdapterMessage newMessage;
 413       newMessage.packet = m_incomingAdapterMessage->packet;
 414       MessageReceived(newMessage);
 415
 416       /* clear the current message */
 417       CLockObject lock(m_mutex);
 418       m_incomingAdapterMessage->Clear();
 419     }
 420   }
 421 }
 422
 423 bool CCECAdapterMessageQueue::Write(CCECAdapterMessage *msg)
 424 {
 425   msg->state = ADAPTER_MESSAGE_STATE_WAITING_TO_BE_SENT;
 426
 427   /* set the correct line timeout */
 428   if (msg->IsTranmission())
 429   {
 430     m_com->SetLineTimeout(msg->lineTimeout);
 431   }
 432
 433   CCECAdapterMessageQueueEntry *entry = new CCECAdapterMessageQueueEntry(this, msg);
 434   if (!entry)
 435   {
 436     m_com->m_callback->GetLib()->AddLog(CEC_LOG_ERROR, "couldn't create queue entry for '%s'", CCECAdapterMessage::ToString(msg->Message()));
 437     msg->state = ADAPTER_MESSAGE_STATE_ERROR;
 438     return false;
 439   }
 440
 441   uint64_t iEntryId(0);
 442   /* add to the wait for ack queue */
 443   if (msg->Message() != MSGCODE_START_BOOTLOADER)
 444   {
 445     CLockObject lock(m_mutex);
 446     iEntryId = m_iNextMessage++;
 447     m_messages.insert(make_pair(iEntryId, entry));
 448   }
 449
 450   /* add the message to the write queue */
 451   m_writeQueue.Push(entry);
 452
 453   bool bReturn(true);
 454   if (!msg->bFireAndForget)
 455   {
 456     if (!entry->Wait(msg->transmit_timeout <= 5 ? CEC_DEFAULT_TRANSMIT_WAIT : msg->transmit_timeout))
 457     {
 458       m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "command '%s' was not acked by the controller", CCECAdapterMessage::ToString(msg->Message()));
 459       msg->state = ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED;
 460       bReturn = false;
 461     }
 462
 463     if (msg->Message() != MSGCODE_START_BOOTLOADER)
 464     {
 465       CLockObject lock(m_mutex);
 466       m_messages.erase(iEntryId);
 467     }
 468
 469     if (msg->ReplyIsError() && msg->state != ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED)
 470       msg->state = ADAPTER_MESSAGE_STATE_ERROR;
 471
 472     delete entry;
 473   }
 474
 475   return bReturn;
 476 }
 477
 478 bool CCECAdapterMessageQueue::ProvidesExtendedResponse(void)
 479 {
 480   return m_com && m_com->ProvidesExtendedResponse();
 481 }