less debug logging by default
[deb_libcec.git] / 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 return thisMsgCode == msgResponse;
169
170 if (!m_message->IsTranmission())
171 {
172 m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_WARNING, "FIXME! not a transmission: %s", msg.ToString().c_str());
173 return false;
174 }
175
176 return ((msgCode == MSGCODE_COMMAND_ACCEPTED || msgCode == MSGCODE_COMMAND_REJECTED) &&
177 (msgResponse == MSGCODE_TRANSMIT_ACK_POLARITY || msgResponse == MSGCODE_TRANSMIT || msgResponse == MSGCODE_TRANSMIT_EOM)) ||
178 msgCode == MSGCODE_TIMEOUT_ERROR ||
179 msgCode == MSGCODE_RECEIVE_FAILED ||
180 msgCode == MSGCODE_TRANSMIT_FAILED_ACK ||
181 msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_DATA ||
182 msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_LINE ||
183 msgCode == MSGCODE_TRANSMIT_SUCCEEDED;
184 }
185
186 const char *CCECAdapterMessageQueueEntry::ToString(void) const
187 {
188 /* CEC transmissions got the 'set ack polarity' msgcode, which doesn't look nice */
189 if (m_message->IsTranmission())
190 return "CEC transmission";
191 else
192 return CCECAdapterMessage::ToString(m_message->Message());
193 }
194
195 bool CCECAdapterMessageQueueEntry::MessageReceivedCommandAccepted(const CCECAdapterMessage &message)
196 {
197 bool bSendSignal(false);
198 bool bHandled(false);
199 {
200 CLockObject lock(m_mutex);
201 if (m_iPacketsLeft > 0)
202 {
203 /* decrease by 1 */
204 m_iPacketsLeft--;
205
206 #ifdef CEC_DEBUGGING
207 /* log this message */
208 CStdString strLog;
209 strLog.Format("%s - command accepted", ToString());
210 if (m_iPacketsLeft > 0)
211 strLog.AppendFormat(" - waiting for %d more", m_iPacketsLeft);
212 m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, strLog);
213 #endif
214
215 /* no more packets left and not a transmission, so we're done */
216 if (!m_message->IsTranmission() && m_iPacketsLeft == 0)
217 {
218 m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
219 m_message->response = message.packet;
220 bSendSignal = true;
221 }
222 bHandled = true;
223 }
224 }
225
226 if (bSendSignal)
227 Signal();
228
229 return bHandled;
230 }
231
232 bool CCECAdapterMessageQueueEntry::MessageReceivedTransmitSucceeded(const CCECAdapterMessage &message)
233 {
234 {
235 CLockObject lock(m_mutex);
236 if (m_iPacketsLeft == 0)
237 {
238 /* transmission succeeded, so we're done */
239 #ifdef CEC_DEBUGGING
240 m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "%s - transmit succeeded", m_message->ToString().c_str());
241 #endif
242 m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
243 m_message->response = message.packet;
244 }
245 else
246 {
247 /* error, we expected more acks
248 since the messages are processed in order, this should not happen, so this is an error situation */
249 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);
250 m_message->state = ADAPTER_MESSAGE_STATE_ERROR;
251 }
252 }
253
254 Signal();
255
256 return true;
257 }
258
259 bool CCECAdapterMessageQueueEntry::MessageReceivedResponse(const CCECAdapterMessage &message)
260 {
261 {
262 CLockObject lock(m_mutex);
263 #ifdef CEC_DEBUGGING
264 m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "%s - received response - %s", ToString(), message.ToString().c_str());
265 #else
266 if (message.IsError())
267 m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "%s - received response - %s", ToString(), message.ToString().c_str());
268 #endif
269 m_message->response = message.packet;
270 if (m_message->IsTranmission())
271 m_message->state = message.Message() == MSGCODE_TRANSMIT_SUCCEEDED ? ADAPTER_MESSAGE_STATE_SENT_ACKED : ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED;
272 else
273 m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
274 }
275
276 Signal();
277
278 return true;
279 }
280
281 bool CCECAdapterMessageQueueEntry::ProvidesExtendedResponse(void)
282 {
283 return m_queue && m_queue->ProvidesExtendedResponse();
284 }
285
286 bool CCECAdapterMessageQueueEntry::TimedOutOrSucceeded(void) const
287 {
288 return m_message->bFireAndForget && (m_bSucceeded || m_queueTimeout.TimeLeft() == 0);
289 }
290
291 CCECAdapterMessageQueue::CCECAdapterMessageQueue(CUSBCECAdapterCommunication *com) :
292 PLATFORM::CThread(),
293 m_com(com),
294 m_iNextMessage(0)
295 {
296 m_incomingAdapterMessage = new CCECAdapterMessage;
297 m_currentCECFrame.Clear();
298 }
299
300 CCECAdapterMessageQueue::~CCECAdapterMessageQueue(void)
301 {
302 StopThread(-1);
303 Clear();
304 StopThread();
305 delete m_incomingAdapterMessage;
306 }
307
308 void CCECAdapterMessageQueue::Clear(void)
309 {
310 StopThread(5);
311 CLockObject lock(m_mutex);
312 m_writeQueue.Clear();
313 m_messages.clear();
314 }
315
316 void *CCECAdapterMessageQueue::Process(void)
317 {
318 CCECAdapterMessageQueueEntry *message(NULL);
319 while (!IsStopped())
320 {
321 /* wait for a new message */
322 if (m_writeQueue.Pop(message, MESSAGE_QUEUE_SIGNAL_WAIT_TIME) && message)
323 {
324 /* write this message */
325 {
326 CLockObject lock(m_mutex);
327 m_com->WriteToDevice(message->m_message);
328 }
329 if (message->m_message->state == ADAPTER_MESSAGE_STATE_ERROR ||
330 message->m_message->Message() == MSGCODE_START_BOOTLOADER)
331 {
332 message->Signal();
333 Clear();
334 break;
335 }
336 }
337
338 CheckTimedOutMessages();
339 }
340 return NULL;
341 }
342
343 void CCECAdapterMessageQueue::CheckTimedOutMessages(void)
344 {
345 CLockObject lock(m_mutex);
346 vector<uint64_t> timedOut;
347 for (map<uint64_t, CCECAdapterMessageQueueEntry *>::iterator it = m_messages.begin(); it != m_messages.end(); it++)
348 {
349 if (it->second->TimedOutOrSucceeded())
350 {
351 timedOut.push_back(it->first);
352 if (!it->second->m_bSucceeded)
353 m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "command '%s' was not acked by the controller", CCECAdapterMessage::ToString(it->second->m_message->Message()));
354 delete it->second->m_message;
355 delete it->second;
356 }
357 }
358
359 for (vector<uint64_t>::iterator it = timedOut.begin(); it != timedOut.end(); it++)
360 {
361 uint64_t iEntryId = *it;
362 m_messages.erase(iEntryId);
363 }
364 }
365
366 void CCECAdapterMessageQueue::MessageReceived(const CCECAdapterMessage &msg)
367 {
368 bool bHandled(false);
369 CLockObject lock(m_mutex);
370 /* send the received message to each entry in the queue until it is handled */
371 for (map<uint64_t, CCECAdapterMessageQueueEntry *>::iterator it = m_messages.begin(); !bHandled && it != m_messages.end(); it++)
372 bHandled = it->second->MessageReceived(msg);
373
374 if (!bHandled)
375 {
376 /* the message wasn't handled */
377 bool bIsError(m_com->HandlePoll(msg));
378 #ifdef CEC_DEBUGGING
379 m_com->m_callback->GetLib()->AddLog(bIsError ? CEC_LOG_WARNING : CEC_LOG_DEBUG, msg.ToString().c_str());
380 #else
381 if (bIsError)
382 m_com->m_callback->GetLib()->AddLog(CEC_LOG_WARNING, msg.ToString().c_str());
383 #endif
384
385 /* push this message to the current frame */
386 if (!bIsError && msg.PushToCecCommand(m_currentCECFrame))
387 {
388 /* and push the current frame back over the callback method when a full command was received */
389 if (m_com->IsInitialised())
390 m_com->m_callback->OnCommandReceived(m_currentCECFrame);
391
392 /* clear the current frame */
393 m_currentCECFrame.Clear();
394 }
395 }
396 }
397
398 void CCECAdapterMessageQueue::AddData(uint8_t *data, size_t iLen)
399 {
400 for (size_t iPtr = 0; iPtr < iLen; iPtr++)
401 {
402 bool bFullMessage(false);
403 {
404 CLockObject lock(m_mutex);
405 bFullMessage = m_incomingAdapterMessage->PushReceivedByte(data[iPtr]);
406 }
407
408 if (bFullMessage)
409 {
410 /* a full message was received */
411 CCECAdapterMessage newMessage;
412 newMessage.packet = m_incomingAdapterMessage->packet;
413 MessageReceived(newMessage);
414
415 /* clear the current message */
416 CLockObject lock(m_mutex);
417 m_incomingAdapterMessage->Clear();
418 }
419 }
420 }
421
422 bool CCECAdapterMessageQueue::Write(CCECAdapterMessage *msg)
423 {
424 msg->state = ADAPTER_MESSAGE_STATE_WAITING_TO_BE_SENT;
425
426 /* set the correct line timeout */
427 if (msg->IsTranmission())
428 {
429 m_com->SetLineTimeout(msg->lineTimeout);
430 }
431
432 CCECAdapterMessageQueueEntry *entry = new CCECAdapterMessageQueueEntry(this, msg);
433 if (!entry)
434 {
435 m_com->m_callback->GetLib()->AddLog(CEC_LOG_ERROR, "couldn't create queue entry for '%s'", CCECAdapterMessage::ToString(msg->Message()));
436 msg->state = ADAPTER_MESSAGE_STATE_ERROR;
437 return false;
438 }
439
440 uint64_t iEntryId(0);
441 /* add to the wait for ack queue */
442 if (msg->Message() != MSGCODE_START_BOOTLOADER)
443 {
444 CLockObject lock(m_mutex);
445 iEntryId = m_iNextMessage++;
446 m_messages.insert(make_pair(iEntryId, entry));
447 }
448
449 /* add the message to the write queue */
450 m_writeQueue.Push(entry);
451
452 bool bReturn(true);
453 if (!msg->bFireAndForget)
454 {
455 if (!entry->Wait(msg->transmit_timeout <= 5 ? CEC_DEFAULT_TRANSMIT_WAIT : msg->transmit_timeout))
456 {
457 m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "command '%s' was not acked by the controller", CCECAdapterMessage::ToString(msg->Message()));
458 msg->state = ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED;
459 bReturn = false;
460 }
461
462 if (msg->Message() != MSGCODE_START_BOOTLOADER)
463 {
464 CLockObject lock(m_mutex);
465 m_messages.erase(iEntryId);
466 }
467
468 if (msg->ReplyIsError() && msg->state != ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED)
469 msg->state = ADAPTER_MESSAGE_STATE_ERROR;
470
471 delete entry;
472 }
473
474 return bReturn;
475 }
476
477 bool CCECAdapterMessageQueue::ProvidesExtendedResponse(void)
478 {
479 return m_com && m_com->ProvidesExtendedResponse();
480 }