【发布时间】：2014-03-02 01:14:04
【问题描述】：
我有一个 std::deque 类型的缓冲区。 有一个线程要写入其中，另一个线程要从中读取，最后一个线程处理缓冲区中要转发的项目的某些条件。
我只想从 3 个线程安全地访问这个缓冲区。是的，我是初学者 :-)
我创建了一个互斥体，每次我访问缓冲区时，我都会用
包装这个访问myMutex. lock() ;
// access here
myMutex. unlock() ;
另外，我使用 std::thread myThread(this, &fn) 创建线程。我经常调用 this_thread::sleep() 来减少 CPU 负载
我的问题是我得到 Exeption 说 abort() 已被调用！当我调用 myThread.join() 时，它会失败 怎么回事！！
编辑：添加代码 这是我的主要发送功能
void UDPStreamSender::SendStream(const char* sendMsg, size_t size)
{
cout << "---- Send Stream starts... ----" << endl;
char* longMsg = new char[size];
memcpy(longMsg, sendMsg, size);
std::thread segThread(&UDPStreamSender::DoSegmentation, this, longMsg, size);
_isRunning = true;
std::thread sendThrad(&UDPStreamSender::SendBuffer, this);
std::thread ackThrad(&UDPStreamSender::AckRecive, this);
std::thread timeOutThread(&UDPStreamSender::ManageTimeout, this);
sendThrad.join();
ackThrad.join();
timeOutThread.join();
cout << "---- Send Stream done! ----" << endl;
}
fn 在工作线程上
void UDPStreamSender::DoSegmentation(const char* longMsg, unsigned int size)
{
Segment* cSeg = new Segment();
cSeg->seqNum = lastSeqNum;
msgLength = size;
msgSegLen = segLength - SEQ_NUM_LEN;
segmentsNumber = (unsigned int)ceil((float)msgLength / (msgSegLen));
for (size_t i = 0; i < segmentsNumber; i++)
{
cSeg->seqNum++;
lastSeqNum = cSeg->seqNum;
cSeg->data = new char[msgSegLen];
int sendMsgSegLen = msgSegLen;
if (i == segmentsNumber - 1)
sendMsgSegLen = msgLength - i*msgSegLen;
memcpy(cSeg->data, longMsg + i*msgSegLen, sendMsgSegLen);
// Add to send buffer
while (sendBuffer->isFull())
{
this_thread::sleep_for(std::chrono::milliseconds(50));
}
cSeg->isSent = false;
bufLock.lock();
std::unique_lock<std::mutex> lock(bufLock);
sendBuffer->Add(cSeg);
bufLock.unlock();
}
cv.notify_all();
}
void UDPStreamSender::SendBuffer()
{
bufLock.lock();
bool hasElms = sendBuffer->hasElems();
bufLock.unlock();
while (_isRunning || hasElms)
{
bufLock.lock();
size_t firstUnsent = sendBuffer->firstUnsent();
size_t buffCount = sendBuffer->count();
bufLock.unlock();
if (firstUnsent == buffCount)
{
this_thread::sleep_for(std::chrono::milliseconds(50));
continue;
}
for (size_t i = firstUnsent; i < buffCount; i++)
{
bufLock.lock();
Segment sSeg = sendBuffer->at(i);
int st = SendSegment(&sSeg);
if (st >= segLength)
{
sSeg.isSent = true;
DWORD j = GetTickCount();
sSeg.timeOutTick = j + timeOutTicks;
sendBuffer->Replace(i, &sSeg);
sendBuffer->sendSeg();
cout << "SEG sent: SeqNum=" << sSeg.seqNum << endl;
}
bufLock.unlock();
}
bufLock.lock();
hasElms = sendBuffer->hasElems();
bufLock.unlock();
}
}
void UDPStreamSender::AckRecive()
{
char* ackMessage;
while (!_allRecived)
{
ackMessage = ackReciver->Recive();
string ackMsg(ackMessage);
if (ackMsg.substr(0, 3).compare("ACK") != 0)
continue;
unsigned short ackSeqNum = 0;
memcpy(&ackSeqNum, ackMessage + 3, 2);
cout << "ACK recieved: seqNum=" << ackSeqNum << endl;
bufLock.lock();
sendBuffer->Ack(ackSeqNum);
_allRecived = !sendBuffer->hasElems();
bufLock.unlock();
}
}
void UDPStreamSender::ManageTimeout()
{
bufLock.lock();
bool hasElms = sendBuffer->hasElems();
bufLock.unlock();
while (hasElms)
{
bufLock.lock();
DWORD segTick = sendBuffer->first().timeOutTick;
DWORD cTick = GetTickCount();
if (sendBuffer->hasElems() && cTick > segTick)
{ // timeout, resend all buffer
sendBuffer->resendAll();
cout << "Timeout: seqNum=" << sendBuffer->first().seqNum << endl;
}
bufLock.unlock();
this_thread::sleep_for(std::chrono::milliseconds(50));
bufLock.lock();
hasElms = sendBuffer->hasElems();
bufLock.unlock();
}
}
我知道那里有很多线程，但这只是一个任务！
【问题讨论】：
-
代码在哪里？您的锁定/解锁 sn-p 并没有说明可能导致问题的实际原因
-
你试过用
-pthread编译吗？ -
@Smac89 我添加了代码。
-
@KerrekSB 我真的不知道什么是“-pthread”！
-
-pthread是一个编译器选项。就像在不支持的编译器中添加-std=gnu++11用于编译 c++11 代码一样，-pthread用于编译 posix 线程代码。
标签： c++ multithreading mutex