- Fix race condition when start watching a mailbox and directly afterwards stop watching it. A BLooper was not suitable to synchronise start and stop watching. Wait till the IDLE command is send before returning the SyncAndStartWatchingMailbox method now. That ensures that a later StopWatchingMailbox call find the maibox in an expected watching state.
There is one thread (BLooper) to handle new commands and one watcher thread which is just listening at the server port for updates. The race condition occurred for example when a sync/watching and a fetch body message are send to the looper. The sync message just triggered the IDLE command in the watcher thread. In the meantime the fetch body command send a DONE command, because the IDLE command has not be send at this time the watcher keeps watching. - fix int32 -> ssize_t thanks Axel and Stippi - clean up git-svn-id: file:///srv/svn/repos/haiku/haiku/trunk@40919 a95241bf-73f2-0310-859d-f6bbb57e9c96
This commit is contained in:
@@ -77,39 +77,30 @@ DispatcherIMAPListener::FetchEnd()
|
|||||||
const uint32 kMsgStartWatching = '&StW';
|
const uint32 kMsgStartWatching = '&StW';
|
||||||
|
|
||||||
|
|
||||||
IMAPMailboxThread::IMAPMailboxThread(IMAPInboundProtocol& protocol,
|
int32
|
||||||
IMAPMailbox& mailbox)
|
watch_mailbox(void* data)
|
||||||
:
|
|
||||||
BLooper("IMAPMailboxThread"),
|
|
||||||
|
|
||||||
fProtocol(protocol),
|
|
||||||
fIMAPMailbox(mailbox),
|
|
||||||
|
|
||||||
fIsWatching(false)
|
|
||||||
{
|
{
|
||||||
|
((IMAPMailboxThread*)data)->_Watch();
|
||||||
|
return B_OK;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
void
|
IMAPMailboxThread::IMAPMailboxThread(IMAPInboundProtocol& protocol,
|
||||||
IMAPMailboxThread::MessageReceived(BMessage* message)
|
IMAPMailbox& mailbox)
|
||||||
|
:
|
||||||
|
fProtocol(protocol),
|
||||||
|
fIMAPMailbox(mailbox),
|
||||||
|
|
||||||
|
fIsWatching(false),
|
||||||
|
fThread(-1)
|
||||||
{
|
{
|
||||||
status_t status = B_ERROR;
|
fWatchSyncSem = create_sem(0, "watch sync sem");
|
||||||
|
}
|
||||||
|
|
||||||
switch (message->what) {
|
|
||||||
case kMsgStartWatching:
|
|
||||||
status = fIMAPMailbox.StartWatchingMailbox();
|
|
||||||
if (status != B_OK)
|
|
||||||
fProtocol.Disconnect();
|
|
||||||
|
|
||||||
fLock.Lock();
|
IMAPMailboxThread::~IMAPMailboxThread()
|
||||||
fIsWatching = false;
|
{
|
||||||
fLock.Unlock();
|
delete_sem(fWatchSyncSem);
|
||||||
break;
|
|
||||||
|
|
||||||
default:
|
|
||||||
BLooper::MessageReceived(message);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -131,9 +122,14 @@ IMAPMailboxThread::SyncAndStartWatchingMailbox()
|
|||||||
BAutolock autolock(fLock);
|
BAutolock autolock(fLock);
|
||||||
if (fIsWatching)
|
if (fIsWatching)
|
||||||
return B_OK;
|
return B_OK;
|
||||||
|
fThread = spawn_thread(watch_mailbox, "IMAPMailboxThread",
|
||||||
|
B_LOW_PRIORITY, this);
|
||||||
|
if (resume_thread(fThread) != B_OK) {
|
||||||
|
fThread = -1;
|
||||||
|
return B_ERROR;
|
||||||
|
}
|
||||||
|
acquire_sem(fWatchSyncSem);
|
||||||
fIsWatching = true;
|
fIsWatching = true;
|
||||||
autolock.Unlock();
|
|
||||||
PostMessage(kMsgStartWatching);
|
|
||||||
} else {
|
} else {
|
||||||
status_t status = fIMAPMailbox.CheckMailbox();
|
status_t status = fIMAPMailbox.CheckMailbox();
|
||||||
// if we lost connection reconnect and try again
|
// if we lost connection reconnect and try again
|
||||||
@@ -158,9 +154,23 @@ IMAPMailboxThread::StopWatchingMailbox()
|
|||||||
return status;
|
return status;
|
||||||
|
|
||||||
// wait till watching stopped
|
// wait till watching stopped
|
||||||
const uint32 kMsgNoMeaning = '&NME';
|
status_t exitCode;
|
||||||
BMessage reply;
|
return wait_for_thread(fThread, &exitCode);
|
||||||
return BMessenger(this).SendMessage(kMsgNoMeaning, &reply);
|
}
|
||||||
|
|
||||||
|
|
||||||
|
void
|
||||||
|
IMAPMailboxThread::_Watch()
|
||||||
|
{
|
||||||
|
status_t status = fIMAPMailbox.StartWatchingMailbox(fWatchSyncSem);
|
||||||
|
if (status != B_OK)
|
||||||
|
fProtocol.Disconnect();
|
||||||
|
|
||||||
|
fLock.Lock();
|
||||||
|
fIsWatching = false;
|
||||||
|
fLock.Unlock();
|
||||||
|
|
||||||
|
fThread = -1;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -192,7 +202,6 @@ MailboxWatcher::StartWatching(const char* mailboxDir)
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
#include <stdio.h>
|
|
||||||
void
|
void
|
||||||
MailboxWatcher::MessageReceived(BMessage* message)
|
MailboxWatcher::MessageReceived(BMessage* message)
|
||||||
{
|
{
|
||||||
@@ -206,7 +215,6 @@ MailboxWatcher::MessageReceived(BMessage* message)
|
|||||||
break;
|
break;
|
||||||
switch (opcode) {
|
switch (opcode) {
|
||||||
case B_ENTRY_CREATED:
|
case B_ENTRY_CREATED:
|
||||||
printf("entry created\n");
|
|
||||||
break;
|
break;
|
||||||
message->FindInt32("device", &ref.device);
|
message->FindInt32("device", &ref.device);
|
||||||
message->FindInt64("directory", &ref.directory);
|
message->FindInt64("directory", &ref.directory);
|
||||||
@@ -225,7 +233,6 @@ MailboxWatcher::MessageReceived(BMessage* message)
|
|||||||
|
|
||||||
case B_ENTRY_MOVED:
|
case B_ENTRY_MOVED:
|
||||||
{
|
{
|
||||||
printf("entry moved\n");
|
|
||||||
break;
|
break;
|
||||||
entry_ref from;
|
entry_ref from;
|
||||||
entry_ref to;
|
entry_ref to;
|
||||||
@@ -301,7 +308,6 @@ IMAPInboundProtocol::IMAPInboundProtocol(BMailAccountSettings* settings,
|
|||||||
fIMAPMailbox.SetFetchBodyLimit(bodyLimit);
|
fIMAPMailbox.SetFetchBodyLimit(bodyLimit);
|
||||||
|
|
||||||
fIMAPMailboxThread = new IMAPMailboxThread(*this, fIMAPMailbox);
|
fIMAPMailboxThread = new IMAPMailboxThread(*this, fIMAPMailbox);
|
||||||
fIMAPMailboxThread->Run();
|
|
||||||
|
|
||||||
// set watch directory
|
// set watch directory
|
||||||
fINBOXWatcher = new MailboxWatcher(this);
|
fINBOXWatcher = new MailboxWatcher(this);
|
||||||
@@ -316,8 +322,7 @@ IMAPInboundProtocol::~IMAPInboundProtocol()
|
|||||||
RemoveHandler(fINBOXWatcher);
|
RemoveHandler(fINBOXWatcher);
|
||||||
delete fINBOXWatcher;
|
delete fINBOXWatcher;
|
||||||
|
|
||||||
fIMAPMailboxThread->Lock();
|
delete fIMAPMailboxThread;
|
||||||
fIMAPMailboxThread->Quit();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -456,7 +461,6 @@ IMAPInboundProtocol::UpdateSettings(const BMessage& settings)
|
|||||||
delete[] passwd;
|
delete[] passwd;
|
||||||
}
|
}
|
||||||
|
|
||||||
// restart mailbox's
|
|
||||||
SyncMessages();
|
SyncMessages();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -41,24 +41,34 @@ private:
|
|||||||
|
|
||||||
class IMAPInboundProtocol;
|
class IMAPInboundProtocol;
|
||||||
|
|
||||||
|
|
||||||
|
int32 watch_mailbox(void* data);
|
||||||
|
|
||||||
|
|
||||||
/*! Just wait for a IDLE (watching) IMAP response in this thread. */
|
/*! Just wait for a IDLE (watching) IMAP response in this thread. */
|
||||||
class IMAPMailboxThread : public BLooper {
|
class IMAPMailboxThread {
|
||||||
public:
|
public:
|
||||||
IMAPMailboxThread(IMAPInboundProtocol& protocol,
|
IMAPMailboxThread(IMAPInboundProtocol& protocol,
|
||||||
IMAPMailbox& mailbox);
|
IMAPMailbox& mailbox);
|
||||||
|
~IMAPMailboxThread();
|
||||||
void MessageReceived(BMessage* message);
|
|
||||||
|
|
||||||
bool IsWatching();
|
bool IsWatching();
|
||||||
status_t SyncAndStartWatchingMailbox();
|
status_t SyncAndStartWatchingMailbox();
|
||||||
status_t StopWatchingMailbox();
|
status_t StopWatchingMailbox();
|
||||||
|
|
||||||
private:
|
private:
|
||||||
|
void _Watch();
|
||||||
|
|
||||||
|
friend int32 watch_mailbox(void* data);
|
||||||
|
|
||||||
IMAPInboundProtocol& fProtocol;
|
IMAPInboundProtocol& fProtocol;
|
||||||
IMAPMailbox& fIMAPMailbox;
|
IMAPMailbox& fIMAPMailbox;
|
||||||
|
|
||||||
BLocker fLock;
|
BLocker fLock;
|
||||||
bool fIsWatching;
|
bool fIsWatching;
|
||||||
|
|
||||||
|
thread_id fThread;
|
||||||
|
sem_id fWatchSyncSem;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -42,7 +42,6 @@ IMAPRootInboundProtocol::Connect(const char* server, const char* username,
|
|||||||
if (!folders[i].subscribed || folders[i].folder == "INBOX")
|
if (!folders[i].subscribed || folders[i].folder == "INBOX")
|
||||||
continue;
|
continue;
|
||||||
|
|
||||||
|
|
||||||
IMAPInboundProtocol* inboundProtocol = new IMAPInboundProtocol(
|
IMAPInboundProtocol* inboundProtocol = new IMAPInboundProtocol(
|
||||||
&fAccountSettings, folders[i].folder);
|
&fAccountSettings, folders[i].folder);
|
||||||
inboundProtocol->SetMailNotifier(fMailNotifier->Clone());
|
inboundProtocol->SetMailNotifier(fMailNotifier->Clone());
|
||||||
|
|||||||
@@ -122,11 +122,11 @@ IMAPMailbox::SupportWatching()
|
|||||||
|
|
||||||
|
|
||||||
status_t
|
status_t
|
||||||
IMAPMailbox::StartWatchingMailbox()
|
IMAPMailbox::StartWatchingMailbox(sem_id startedSem)
|
||||||
{
|
{
|
||||||
//TODO set it when we actually watching
|
|
||||||
atomic_set(&fWatching, 1);
|
atomic_set(&fWatching, 1);
|
||||||
|
|
||||||
|
bool firstIDLE = true;
|
||||||
// refresh every 29 min
|
// refresh every 29 min
|
||||||
bigtime_t timeout = 1000 * 1000 * 60 * 29; // 29 min
|
bigtime_t timeout = 1000 * 1000 * 60 * 29; // 29 min
|
||||||
status_t status;
|
status_t status;
|
||||||
@@ -134,8 +134,13 @@ IMAPMailbox::StartWatchingMailbox()
|
|||||||
int32 commandId = NextCommandId();
|
int32 commandId = NextCommandId();
|
||||||
TRACE("IDLE ...\n");
|
TRACE("IDLE ...\n");
|
||||||
status = SendCommand("IDLE", commandId);
|
status = SendCommand("IDLE", commandId);
|
||||||
|
if (firstIDLE) {
|
||||||
|
release_sem(startedSem);
|
||||||
|
firstIDLE = false;
|
||||||
|
}
|
||||||
if (status != B_OK)
|
if (status != B_OK)
|
||||||
break;
|
break;
|
||||||
|
|
||||||
status = HandleResponse(commandId, timeout, false);
|
status = HandleResponse(commandId, timeout, false);
|
||||||
ProcessAfterQuacks(kIMAP4ClientTimeout);
|
ProcessAfterQuacks(kIMAP4ClientTimeout);
|
||||||
|
|
||||||
@@ -157,6 +162,7 @@ IMAPMailbox::StartWatchingMailbox()
|
|||||||
if (status != B_OK)
|
if (status != B_OK)
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
atomic_set(&fWatching, 0);
|
atomic_set(&fWatching, 0);
|
||||||
return status;
|
return status;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -44,7 +44,7 @@ public:
|
|||||||
status_t Sync();
|
status_t Sync();
|
||||||
|
|
||||||
bool SupportWatching();
|
bool SupportWatching();
|
||||||
status_t StartWatchingMailbox();
|
status_t StartWatchingMailbox(sem_id startedSem = -1);
|
||||||
status_t StopWatchingMailbox();
|
status_t StopWatchingMailbox();
|
||||||
|
|
||||||
status_t CheckMailbox();
|
status_t CheckMailbox();
|
||||||
|
|||||||
@@ -95,10 +95,10 @@ status_t
|
|||||||
BRawNetBuffer::ReadString(BString& string)
|
BRawNetBuffer::ReadString(BString& string)
|
||||||
{
|
{
|
||||||
string = "";
|
string = "";
|
||||||
int32 read = _ReadStringAt(string, fReadPosition);
|
ssize_t bytesRead = _ReadStringAt(string, fReadPosition);
|
||||||
if (read < 0)
|
if (bytesRead < 0)
|
||||||
return B_ERROR;
|
return B_ERROR;
|
||||||
fReadPosition += read;
|
fReadPosition += bytesRead;
|
||||||
return B_OK;
|
return B_OK;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -122,13 +122,13 @@ BRawNetBuffer::_Init(const void* buf, size_t size)
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
int32
|
ssize_t
|
||||||
BRawNetBuffer::_ReadStringAt(BString& string, off_t pos)
|
BRawNetBuffer::_ReadStringAt(BString& string, off_t pos)
|
||||||
{
|
{
|
||||||
if (pos >= fBuffer.BufferLength())
|
if (pos >= fBuffer.BufferLength())
|
||||||
return -1;
|
return -1;
|
||||||
|
|
||||||
int32 readed = 0;
|
ssize_t bytesRead = 0;
|
||||||
char* buffer = (char*)fBuffer.Buffer();
|
char* buffer = (char*)fBuffer.Buffer();
|
||||||
buffer = &buffer[pos];
|
buffer = &buffer[pos];
|
||||||
// if the string is compressed we have to follow the links to the
|
// if the string is compressed we have to follow the links to the
|
||||||
@@ -137,17 +137,17 @@ BRawNetBuffer::_ReadStringAt(BString& string, off_t pos)
|
|||||||
if (uint8(*buffer) == 192) {
|
if (uint8(*buffer) == 192) {
|
||||||
// found a pointer mark
|
// found a pointer mark
|
||||||
buffer++;
|
buffer++;
|
||||||
readed++;
|
bytesRead++;
|
||||||
off_t subPos = uint8(*buffer);
|
off_t subPos = uint8(*buffer);
|
||||||
_ReadStringAt(string, subPos);
|
_ReadStringAt(string, subPos);
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
string.Append(buffer, 1);
|
string.Append(buffer, 1);
|
||||||
buffer++;
|
buffer++;
|
||||||
readed++;
|
bytesRead++;
|
||||||
}
|
}
|
||||||
readed++;
|
bytesRead++;
|
||||||
return readed;
|
return bytesRead;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -42,7 +42,7 @@ public:
|
|||||||
|
|
||||||
private:
|
private:
|
||||||
void _Init(const void* buf, size_t size);
|
void _Init(const void* buf, size_t size);
|
||||||
int32 _ReadStringAt(BString& string, off_t pos);
|
ssize_t _ReadStringAt(BString& string, off_t pos);
|
||||||
|
|
||||||
off_t fWritePosition;
|
off_t fWritePosition;
|
||||||
off_t fReadPosition;
|
off_t fReadPosition;
|
||||||
|
|||||||
@@ -197,15 +197,15 @@ MailDaemonApp::RefsReceived(BMessage* message)
|
|||||||
sizeof(account)) < 0)
|
sizeof(account)) < 0)
|
||||||
continue;
|
continue;
|
||||||
|
|
||||||
InboundProtocolThread* protocol = _FindInboundProtocol(account);
|
InboundProtocolThread* protocolThread = _FindInboundProtocol(account);
|
||||||
if (!protocol)
|
if (!protocolThread)
|
||||||
continue;
|
continue;
|
||||||
|
|
||||||
BMessenger target;
|
BMessenger target;
|
||||||
BMessenger* messenger = ⌖
|
BMessenger* messenger = ⌖
|
||||||
if (message->FindMessenger("target", &target) != B_OK)
|
if (message->FindMessenger("target", &target) != B_OK)
|
||||||
messenger = NULL;
|
messenger = NULL;
|
||||||
protocol->FetchBody(ref, messenger);
|
protocolThread->FetchBody(ref, messenger);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user