nils/ebusd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

integration from bus into ebusloop continued.

Browse Source
This commit is contained in:
Roland Jax
2014-11-07 20:58:15 +01:00
parent 6db67385b5
commit 6b0f58b291
6 changed files with 435 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ebusd/ebusloop.cpp
+391 -223
View File
@@ -24,43 +24,27 @@
extern LogInstance& L; extern LogInstance& L;
extern Appl& A; extern Appl& A;
EBusLoop::EBusLoop(Commands* commands) : m_commands(commands), m_stop(false) EBusLoop::EBusLoop(Commands* commands)
: m_commands(commands), m_stop(false), m_busLocked(false), m_priorRetry(false)
{ {
m_port = new Port(A.getParam<const char*>("p_device"), A.getParam<bool>("p_nodevicecheck")); m_port = new Port(A.getParam<const char*>("p_device"), A.getParam<bool>("p_nodevicecheck"));
m_port->open(); m_port->open();
if (m_port->isOpen() == false) if (m_port->isOpen() == false)
L.log(bus, error, "can't open %s", A.getParam<const char*>("p_device")); L.log(bus, error, "can't open %s", A.getParam<const char*>("p_device"));
m_dump = new Dump(A.getParam<const char*>("p_dumpfile"), A.getParam<long>("p_dumpsize")); m_dump = new Dump(A.getParam<const char*>("p_dumpfile"), A.getParam<long>("p_dumpsize"));
m_dumpState = A.getParam<bool>("p_dump"); m_dumpState = A.getParam<bool>("p_dump");
m_logRawData = A.getParam<bool>("p_lograwdata"); m_logRawData = A.getParam<bool>("p_lograwdata");
//~ m_deviceName = A.getParam<const char*>("p_device"); m_pollInterval = A.getParam<int>("p_pollinterval");
//~ m_bus = new Bus(m_deviceName, m_recvTimeout = A.getParam<long>("p_recvtimeout");
//~ A.getParam<bool>("p_nodevicecheck"),
//~ A.getParam<long>("p_recvtimeout"),
//~ A.getParam<const char*>("p_dumpfile"),
//~ A.getParam<long>("p_dumpsize"),
//~ A.getParam<bool>("p_dump"));
//~ m_retries = A.getParam<int>("p_retries"); m_sendRetries = A.getParam<int>("p_sendretries");
//~
//~ m_lookbusretries = A.getParam<int>("p_lookbusretries");
//~
//~ m_pollInterval = A.getParam<int>("p_pollinterval");
//~
m_lockRetries = A.getParam<int>("p_lockretries");
//~ m_bus->connect();
//~ if (m_bus->isConnected() == false)
//~ L.log(bus, error, "can't open %s", m_deviceName.c_str());
} }
EBusLoop::~EBusLoop() EBusLoop::~EBusLoop()
@@ -70,23 +54,35 @@ EBusLoop::~EBusLoop()
delete m_port; delete m_port;
delete m_dump; delete m_dump;
//~ m_bus->disconnect();
//~
//~ if (m_bus->isConnected() == true)
//~ L.log(bus, error, "error during disconnect.");
//~
//~ delete m_bus;
} }
void* EBusLoop::run() void* EBusLoop::run()
{ {
bool busLock = false; int sendRetries = 0;
int lockRetries = 0;
// polling
time_t pollStart, pollEnd;
time(&pollStart);
double pollDelta;
for (;;) { for (;;) {
if (m_port->isOpen() == true) { if (m_port->isOpen() == true) {
unsigned char byte;
ssize_t numBytes; ssize_t numBytes;
// add poll command - timer reached
if (m_commands->sizePolDB() > 0) {
// check polling delta
time(&pollEnd);
pollDelta = difftime(pollEnd, pollStart);
// add new polling command to send
if (pollDelta >= m_pollInterval) {
addPollCommand();
time(&pollStart);
}
}
// read device - no timeout needed (AUTO-SYN) // read device - no timeout needed (AUTO-SYN)
numBytes = m_port->recv(0); numBytes = m_port->recv(0);
@@ -95,46 +91,63 @@ void* EBusLoop::run()
continue; continue;
} }
for (int i = 0; i < numBytes; i++) { // cycle bytes
collectCycData(numBytes);
// fetch byte
byte = recvByte();
// collect cycle data
if (byte != SYN)
m_sstr.push_back(byte, true, false);
// unlock bus
if (byte == SYN && busLock == true) {
busLock = false;
L.log(bus, trace, " bus unlocked");
}
// analyse cycle data
if (byte == SYN && m_sstr.size() > 0) {
analyseCycData(m_sstr);
if (m_sstr.size() == 1) {
busLock = true;
L.log(bus, trace, " bus locked");
}
m_sstr.clear();
}
}
// send command // send command
if (m_sstr.size() == 0 && busLock == false if (m_sstr.size() == 0 && m_busLocked == false && m_sendBuffer.size() > 0) {
&& m_sendBuffer.size() > 0) { // acquire Bus
// TODO: sendCommand.... int busResult = acquireBus();
// send bus command
if (busResult == RESULT_BUS_ACQUIRED) {
BusCommand* busCommand = sendCommand();
L.log(bus, trace, " %s", busCommand->getMessageStr().c_str());
if (busCommand->isErrorResult() == true) {
if (sendRetries < m_sendRetries) {
sendRetries++;
L.log(bus, trace, " send retry %d", sendRetries);
busCommand->setResult(std::string(), RESULT_OK);
}
else {
sendRetries = 0;
if (busCommand->isPoll() == true)
delete m_sendBuffer.remove();
else
busCommand->sendSignal();
}
}
else {
sendRetries = 0;
if (busCommand->isPoll() == true) {
m_commands->storePolData(busCommand->getMessageStr().c_str()); // TODO use getResult()
delete busCommand;
}
else
busCommand->sendSignal();
}
}
else {
L.log(bus, trace, " acquire bus failed");
if (lockRetries >= m_lockRetries) {
L.log(bus, event, " lock bus failed");
BusCommand* busCommand = m_sendBuffer.remove();
if (busCommand->isPoll() == true)
delete busCommand;
else
busCommand->sendSignal();
lockRetries = 0;
}
else
lockRetries++;
}
} }
// poll command - timer reached
if (m_sstr.size() == 0 && busLock == false
&& m_sendBuffer.size() > 0) {
// TODO: pollCommand....
}
} }
else { else {
// TODO: define max reopen // TODO: define max reopen
@@ -158,7 +171,7 @@ void* EBusLoop::run()
return NULL; return NULL;
} }
unsigned char EBusLoop::recvByte() unsigned char EBusLoop::fetchByte()
{ {
unsigned char byte; unsigned char byte;
@@ -174,11 +187,48 @@ unsigned char EBusLoop::recvByte()
return byte; return byte;
} }
void EBusLoop::analyseCycData(SymbolString data) const void EBusLoop::collectCycData(const int numRecv)
{ {
L.log(bus, trace, "%s", data.getDataStr().c_str()); // cycle bytes
for (int i = 0; i < numRecv; i++) {
int index = m_commands->storeCycData(data.getDataStr()); // fetch byte
unsigned char byte = fetchByte();
// collect cycle data
if (byte != SYN)
m_sstr.push_back(byte, true, false);
// unlock bus
if (byte == SYN && m_busLocked == true) {
m_busLocked = false;
L.log(bus, trace, " bus unlocked");
}
// analyse cycle data
if (byte == SYN && m_sstr.size() > 0) {
analyseCycData();
if (m_sstr.size() == 1) {
if (m_priorRetry == true)
m_priorRetry = false;
else {
m_busLocked = true;
L.log(bus, trace, " bus locked");
}
}
m_sstr.clear();
}
}
}
void EBusLoop::analyseCycData()
{
L.log(bus, trace, "%s", m_sstr.getDataStr().c_str());
int index = m_commands->storeCycData(m_sstr.getDataStr());
if (index == -1) { if (index == -1) {
L.log(bus, debug, " command not found"); L.log(bus, debug, " command not found");
@@ -198,174 +248,292 @@ void EBusLoop::analyseCycData(SymbolString data) const
} }
} }
/* void EBusLoop::addPollCommand()
void* EBusLoop::run()
{ {
int busResult; int index = m_commands->nextPolCommand();
int retries = 0; if (index < 0) {
int lookbusretries = 0; L.log(bus, error, "polling index out of range");
bool busCommandActive = false; }
else {
// TODO: implement as methode from class commands?
std::string tmp;
tmp += (*m_commands)[index][1];
tmp += " ";
tmp += (*m_commands)[index][2];
L.log(bus, event, " polling [%d] %s", index, tmp.c_str());
// polling std::string ebusCommand(A.getParam<const char*>("p_address"));
time_t pollStart, pollEnd; ebusCommand += m_commands->getEbusCommand(index);
time(&pollStart); std::transform(ebusCommand.begin(), ebusCommand.end(), ebusCommand.begin(), tolower);
double pollDelta = 0.0;
for (;;) { BusCommand* busCommand = new BusCommand(ebusCommand, true);
if (m_bus->isConnected() == true) { L.log(bus, trace, " msg: %s", ebusCommand.c_str());
// work on bus addBusCommand(busCommand);
busResult = m_bus->proceed(); }
}
// new cyc message arrived int EBusLoop::acquireBus()
if (busResult == RESULT_SYN || busResult == RESULT_BUS_LOCKED) { {
SymbolString data = m_bus->getCycData(); unsigned char recvByte, sendByte;
ssize_t numRecv, numSend;
if (data.size() == 0 && m_logAutoSyn == true) sendByte = m_sendBuffer.next()->getCommand()[0];
L.log(bus, trace, "aa");
if (data.size() != 0) { // send QQ
L.log(bus, trace, "%s", data.getDataStr().c_str()); numSend = m_port->send(&sendByte);
if (numSend <= 0) {
L.log(bus, trace, " ERR_SEND: send error");
return RESULT_ERR_SEND;
}
int index = m_commands->storeCycData(data.getDataStr()); // receive 1 byte - must be QQ
numRecv = m_port->recv(0);
if (index == -1) { if (numRecv < 0) {
L.log(bus, debug, " command not found"); L.log(bus, trace, " ERR_DEVICE: generic device error");
return RESULT_ERR_DEVICE;
}
} else if (index == -2) { if (numRecv == 1) {
L.log(bus, debug, " no commands defined"); // fetch byte
recvByte = fetchByte();
} else if (index == -3) { // compare sent and received byte
L.log(bus, debug, " search skipped - string too short"); if (sendByte == recvByte) {
L.log(bus, trace, " bus acquired");
} else { return RESULT_BUS_ACQUIRED;
std::string tmp;
tmp += (*m_commands)[index][1];
tmp += " ";
tmp += (*m_commands)[index][2];
L.log(bus, event, " cycle [%d] %s", index, tmp.c_str());
}
}
if (busResult == RESULT_BUS_LOCKED)
L.log(bus, trace, "bus locked");
}
// add new bus command to send
if (busResult == RESULT_SYN && busCommandActive == false && m_sendBuffer.size() != 0) {
BusCommand* busCommand = m_sendBuffer.remove();
L.log(bus, debug, " msg: %s", busCommand->getCommand().getDataStr().c_str());
m_bus->addCommand(busCommand);
L.log(bus, debug, " addCommand success");
busCommandActive = true;
}
// add new polling command
if (m_commands->sizePolDB() > 0) {
// check polling delta
time(&pollEnd);
pollDelta = difftime(pollEnd, pollStart);
// add new polling command to send
if (busResult == RESULT_SYN && busCommandActive == false && pollDelta >= m_pollInterval) {
L.log(bus, trace, "polling Intervall reached");
int index = m_commands->nextPolCommand();
if (index < 0) {
L.log(bus, error, "polling index out of range");
time(&pollStart);
continue;
}
std::string tmp;
tmp += (*m_commands)[index][1];
tmp += " ";
tmp += (*m_commands)[index][2];
L.log(bus, event, " polling [%d] %s", index, tmp.c_str());
std::string ebusCommand(A.getParam<const char*>("p_address"));
ebusCommand += m_commands->getEbusCommand(index);
std::transform(ebusCommand.begin(), ebusCommand.end(), ebusCommand.begin(), tolower);
BusCommand* busCommand = new BusCommand(ebusCommand, true);
L.log(bus, trace, " msg: %s", ebusCommand.c_str());
m_bus->addCommand(busCommand);
L.log(bus, debug, " addCommand success");
busCommandActive = true;
time(&pollStart);
}
}
// send bus command
if (busResult == RESULT_BUS_ACQUIRED && busCommandActive == true) {
L.log(bus, trace, " getBus success");
lookbusretries = 0;
BusCommand* busCommand = m_bus->sendCommand();
L.log(bus, trace, " %s", busCommand->getMessageStr().c_str());
if (busCommand->isErrorResult() == true && retries < m_retries) {
retries++;
L.log(bus, trace, " retry number: %d", retries);
busCommand->setResult(std::string(), RESULT_OK);
m_bus->addCommand(busCommand);
} else {
retries = 0;
if (busCommand->isPoll() == true) {
// only save correct results
if (busCommand->isErrorResult() == false)
m_commands->storePolData(busCommand->getMessageStr().c_str()); // TODO use getResult()
delete busCommand;
} else {
busCommand->sendSignal();
}
busCommandActive = false;
}
}
// get bus retry
if (busResult == RESULT_BUS_PRIOR_RETRY)
L.log(bus, trace, " getBus prior retry");
if (busResult == RESULT_ERR_BUS_LOST) {
L.log(bus, trace, " getBus failure");
if (lookbusretries >= m_lookbusretries) {
L.log(bus, event, " getBus failed - command deleted");
BusCommand* busCommand = m_bus->delCommand();
if (busCommand->isPoll() == true) {
delete busCommand;
} else {
busCommand->sendSignal();
}
lookbusretries = 0;
busCommandActive = false;
}else {
lookbusretries++;
}
}
if (busResult == RESULT_ERR_SEND)
L.log(bus, event, " getBus send error");
} else {
sleep(10);
m_bus->connect();
if (m_bus->isConnected() == false)
L.log(bus, error, "can't open %s", m_deviceName.c_str());
} }
if (m_stop == true) { // collect cycle data
m_bus->disconnect(); if (recvByte != SYN)
return NULL; m_sstr.push_back(recvByte, true, false);
// compare prior nibble for retry
if ((sendByte & 0x0F) == (recvByte & 0x0F)) {
m_priorRetry = true;
L.log(bus, trace, " bus prior retry");
return RESULT_BUS_PRIOR_RETRY;
}
L.log(bus, trace, " ERR_BUS_LOST: lost bus arbitration");
return RESULT_ERR_BUS_LOST;
}
// cycle bytes
collectCycData(numRecv);
L.log(bus, trace, " ERR_BUS_LOST: lost bus arbitration");
return RESULT_ERR_BUS_LOST;
}
BusCommand* EBusLoop::sendCommand()
{
unsigned char recvByte;
std::string result;
SymbolString slaveData;
int retval = RESULT_OK;
BusCommand* busCommand = m_sendBuffer.next();
// send ZZ PB SB NN Dx CRC
SymbolString command = busCommand->getCommand();
for (size_t i = 1; i < command.size(); i++) {
retval = sendByte(command[i]);
if (retval < 0)
goto on_exit;
}
// BC -> send SYN
if (busCommand->getType() == broadcast) {
sendByte(SYN);
goto on_exit;
}
// receive ACK
retval = recvSlaveAck(recvByte);
if (retval < 0)
goto on_exit;
// is slave ACK negative?
if (recvByte == NAK) {
// send QQ ZZ PB SB NN Dx CRC again
for (size_t i = 0; i < command.size(); i++) {
retval = sendByte(command[i]);
if (retval < 0)
goto on_exit;
}
// receive ACK
retval = recvSlaveAck(recvByte);
if (retval < 0)
goto on_exit;
// is slave ACK negative?
if (recvByte == NAK) {
sendByte(SYN);
L.log(bus, trace, " ERR_NAK: NAK received");
retval = RESULT_ERR_NAK;
goto on_exit;
} }
} }
return NULL; // MM -> send SYN
if (busCommand->getType() == masterMaster) {
sendByte(SYN);
goto on_exit;
}
// receive NN, Dx, CRC
retval = recvSlaveData(slaveData);
// are calculated and received CRC equal?
if (retval == RESULT_ERR_CRC) {
// send NAK
retval = sendByte(NAK);
if (retval < 0)
goto on_exit;
// receive NN, Dx, CRC
slaveData.clear();
retval = recvSlaveData(slaveData);
// are calculated and received CRC equal?
if (retval == RESULT_ERR_CRC) {
// send NAK
retval = sendByte(NAK);
if (retval >= 0)
retval = RESULT_ERR_CRC;
}
}
if (retval < 0)
goto on_exit;
// send ACK
retval = sendByte(ACK);
if (retval == -1) {
L.log(bus, trace, " ERR_ACK: ACK error");
retval = RESULT_ERR_ACK;
goto on_exit;
}
// MS -> send SYN
sendByte(SYN);
on_exit:
// empty receive buffer
while (m_port->size() != 0)
recvByte = fetchByte();
busCommand->setResult(slaveData, retval);
if (retval == RESULT_OK)
return m_sendBuffer.remove();
else
return busCommand;
}
int EBusLoop::sendByte(const unsigned char sendByte)
{
unsigned char recvByte;
ssize_t numRecv, numSend;
numSend = m_port->send(&sendByte);
// receive 1 byte - must be equal
numRecv = m_port->recv(RECV_TIMEOUT);
if (numSend != numRecv) {
L.log(bus, trace, " ERR_EXTRA_DATA: received bytes > sent bytes");
return RESULT_ERR_EXTRA_DATA;
}
recvByte = fetchByte();
if (sendByte != recvByte) {
L.log(bus, trace, " ERR_SEND: send error");
return RESULT_ERR_SEND;
}
return RESULT_OK;
}
int EBusLoop::recvSlaveAck(unsigned char& recvByte)
{
ssize_t numRecv;
// receive ACK
numRecv = m_port->recv(m_recvTimeout);
if (numRecv > 1) {
L.log(bus, trace, " ERR_EXTRA_DATA: received bytes > sent bytes");
return RESULT_ERR_EXTRA_DATA;
}
else if (numRecv < 0) {
L.log(bus, trace, " ERR_TIMEOUT: read timeout");
return RESULT_ERR_TIMEOUT;
}
recvByte = fetchByte();
// is received byte SYN?
if (recvByte == SYN) {
L.log(bus, trace, " ERR_SYN: SYN received");
return RESULT_ERR_SYN;
}
return RESULT_OK;
}
int EBusLoop::recvSlaveData(SymbolString& result)
{
unsigned char recvByte, calcCrc = 0;
ssize_t numRecv;
size_t NN = 0;
bool updateCrc = true;
int retval = 0;
for (size_t i = 0, needed = 1; i < needed; i++) {
numRecv = m_port->recv(RECV_TIMEOUT);
if (numRecv < 0) {
L.log(bus, trace, " ERR_TIMEOUT: read timeout");
return RESULT_ERR_TIMEOUT;
}
recvByte = fetchByte();
retval = result.push_back(recvByte, true, updateCrc);
if (retval < 0)
return retval;
if (retval == RESULT_IN_ESC)
needed++;
else if (result.size() == 1) { // NN received
NN = result[0];
needed += NN;
}
else if (NN > 0 && result.size() == 1+NN) {// all data received
updateCrc = false;
calcCrc = result.getCRC();
needed++;
}
}
if (retval == RESULT_IN_ESC) {
L.log(bus, trace, " ERR_ESC: invalid escape sequence received");
return RESULT_ERR_ESC;
}
if (updateCrc == true || calcCrc != result[result.size()-1]) {
L.log(bus, trace, " ERR_CRC: CRC error");
return RESULT_ERR_CRC;
}
return RESULT_OK;
} }
*/
src/ebusd/ebusloop.h
+20 -12
View File
@@ -20,7 +20,6 @@
#ifndef EBUSLOOP_H_ #ifndef EBUSLOOP_H_
#define EBUSLOOP_H_ #define EBUSLOOP_H_
//~ #include "bus.h"
#include "commands.h" #include "commands.h"
#include "port.h" #include "port.h"
#include "dump.h" #include "dump.h"
@@ -28,6 +27,9 @@
#include "wqueue.h" #include "wqueue.h"
#include "thread.h" #include "thread.h"
/** the maximum time [us] allowed for retrieving a byte from an addressed slave */
#define RECV_TIMEOUT 10000
using namespace libebus; using namespace libebus;
@@ -43,7 +45,6 @@ public:
void addBusCommand(BusCommand* busCommand) { m_sendBuffer.add(busCommand); } void addBusCommand(BusCommand* busCommand) { m_sendBuffer.add(busCommand); }
//~ void dump(const bool dumpState) { m_bus->setDumpState(dumpState); }
void dump(const bool dumpState) { m_dumpState = dumpState; } void dump(const bool dumpState) { m_dumpState = dumpState; }
void newCommands(Commands* commands) { m_commands = commands; } void newCommands(Commands* commands) { m_commands = commands; }
@@ -59,19 +60,26 @@ private:
bool m_stop; bool m_stop;
SymbolString m_sstr; bool m_busLocked;
bool m_priorRetry;
//~ std::string m_deviceName;
//~ bool m_noDeviceCheck;
//~ Bus* m_bus;
WQueue<BusCommand*> m_sendBuffer; WQueue<BusCommand*> m_sendBuffer;
//~ int m_retries; SymbolString m_sstr;
//~ int m_lookbusretries;
//~ double m_pollInterval;
unsigned char recvByte(); double m_pollInterval;
void analyseCycData(SymbolString data) const; long m_recvTimeout;
int m_sendRetries;
int m_lockRetries;
unsigned char fetchByte();
void collectCycData(const int numRecv);
void analyseCycData();
void addPollCommand();
int acquireBus();
BusCommand* sendCommand();
int sendByte(const unsigned char sendByte);
int recvSlaveAck(unsigned char& recvByte);
int recvSlaveData(SymbolString& result);
}; };
src/ebusd/main.cpp
+4 -4
View File
@@ -47,12 +47,12 @@ void define_args()
"disable valid ebus device test\n", "disable valid ebus device test\n",
Appl::type_bool, Appl::opt_none); Appl::type_bool, Appl::opt_none);
A.addItem("p_retries", Appl::Param(2), "r", "retries", A.addItem("p_sendretries", Appl::Param(2), "s", "sendretries",
"\tnumber retries send ebus command (2)", "number retries send ebus command (2)",
Appl::type_int, Appl::opt_mandatory); Appl::type_int, Appl::opt_mandatory);
A.addItem("p_lookbusretries", Appl::Param(2), "", "lookbusretries", A.addItem("p_lockretries", Appl::Param(2), "", "lockretries",
"number retries to look ebus (2)", "number retries to lock ebus (2)",
Appl::type_int, Appl::opt_mandatory); Appl::type_int, Appl::opt_mandatory);
A.addItem("p_recvtimeout", Appl::Param(15000), "", "recvtimeout", A.addItem("p_recvtimeout", Appl::Param(15000), "", "recvtimeout",
src/lib/ebus/buscommand.cpp
+2 -2
View File
@@ -54,9 +54,9 @@ const std::string BusCommand::getMessageStr()
result += "00"; result += "00";
result += m_result.getDataStr(); result += m_result.getDataStr();
result += "00"; result += "00";
} else {
result = "success";
} }
else
result = "success";
} }
else else
result = "error: "+std::string(getResultCodeCStr()); result = "error: "+std::string(getResultCodeCStr());
src/lib/ebus/port.h
+4 -1
View File
@@ -31,6 +31,9 @@ namespace libebus
/** available device types. */ /** available device types. */
enum DeviceType { SERIAL, NETWORK }; enum DeviceType { SERIAL, NETWORK };
/** max bytes write to bus. */
#define MAX_WRITE_SIZE 1
/** max size of receive buffer. */ /** max size of receive buffer. */
#define MAX_READ_SIZE 100 #define MAX_READ_SIZE 100
@@ -218,7 +221,7 @@ public:
* @param nbytes number of bytes to send. * @param nbytes number of bytes to send.
* @return number of written bytes or -1 if an error has occured. * @return number of written bytes or -1 if an error has occured.
*/ */
ssize_t send(const unsigned char* buffer, size_t nbytes) ssize_t send(const unsigned char* buffer, size_t nbytes = MAX_WRITE_SIZE)
{ return m_device->sendBytes(buffer, nbytes); } { return m_device->sendBytes(buffer, nbytes); }
/** /**
src/lib/utils/wqueue.h
+14
View File
@@ -64,6 +64,20 @@ public:
return item; return item;
} }
T next()
{
pthread_mutex_lock(&m_mutex);
while (m_queue.size() == 0)
pthread_cond_wait(&m_cond, &m_mutex);
T item = m_queue.front();
pthread_mutex_unlock(&m_mutex);
return item;
}
int size() int size()
{ {
pthread_mutex_lock(&m_mutex); pthread_mutex_lock(&m_mutex);