IMAP: sync fixes
This patch should fix: - deadlocks while manual syncing/shutdown; - crashes while manual syncing/shutdown. Change-Id: I0962ff5580c19cbf740d002b6f62681ed5f558ab Reviewed-on: https://review.haiku-os.org/758 Reviewed-by: Stephan Aßmus <[email protected]>
This commit is contained in:
committed by
Stephan Aßmus
parent
e7e7a55250
commit
6fa279737e
@@ -618,10 +618,9 @@ void
|
|||||||
IMAPConnectionWorker::Quit()
|
IMAPConnectionWorker::Quit()
|
||||||
{
|
{
|
||||||
printf("IMAP: worker %p: enqueue quit\n", this);
|
printf("IMAP: worker %p: enqueue quit\n", this);
|
||||||
BAutolock locker(fLocker);
|
BAutolock qlocker(fQueueLocker);
|
||||||
while (!fPendingCommands.IsEmpty())
|
while (!fPendingCommands.IsEmpty())
|
||||||
delete(fPendingCommands.RemoveItemAt(0));
|
delete(fPendingCommands.RemoveItemAt(0));
|
||||||
locker.Unlock();
|
|
||||||
_EnqueueCommand(new QuitCommand());
|
_EnqueueCommand(new QuitCommand());
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -720,12 +719,12 @@ IMAPConnectionWorker::_Worker()
|
|||||||
status_t status = B_OK;
|
status_t status = B_OK;
|
||||||
|
|
||||||
while (!fStopped) {
|
while (!fStopped) {
|
||||||
BAutolock locker(fLocker);
|
BAutolock qlocker(fQueueLocker);
|
||||||
|
|
||||||
if (fPendingCommands.IsEmpty()) {
|
if (fPendingCommands.IsEmpty()) {
|
||||||
if (!fIdle)
|
if (!fIdle)
|
||||||
_Disconnect();
|
_Disconnect();
|
||||||
locker.Unlock();
|
qlocker.Unlock();
|
||||||
|
|
||||||
// TODO: in idle mode, we'd need to parse any incoming message here
|
// TODO: in idle mode, we'd need to parse any incoming message here
|
||||||
_WaitForCommands();
|
_WaitForCommands();
|
||||||
@@ -736,6 +735,8 @@ IMAPConnectionWorker::_Worker()
|
|||||||
if (command == NULL)
|
if (command == NULL)
|
||||||
continue;
|
continue;
|
||||||
|
|
||||||
|
qlocker.Unlock();
|
||||||
|
BAutolock locker(fLocker);
|
||||||
CommandDeleter deleter(*this, command);
|
CommandDeleter deleter(*this, command);
|
||||||
|
|
||||||
if (dynamic_cast<QuitCommand*>(command) == NULL) { // do not connect on QuitCommand
|
if (dynamic_cast<QuitCommand*>(command) == NULL) { // do not connect on QuitCommand
|
||||||
@@ -751,6 +752,7 @@ IMAPConnectionWorker::_Worker()
|
|||||||
if (!command->IsDone()) {
|
if (!command->IsDone()) {
|
||||||
deleter.Detach();
|
deleter.Detach();
|
||||||
command->SetContinuation();
|
command->SetContinuation();
|
||||||
|
locker.Unlock();
|
||||||
_EnqueueCommand(command);
|
_EnqueueCommand(command);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -766,7 +768,7 @@ IMAPConnectionWorker::_Worker()
|
|||||||
status_t
|
status_t
|
||||||
IMAPConnectionWorker::_EnqueueCommand(WorkerCommand* command)
|
IMAPConnectionWorker::_EnqueueCommand(WorkerCommand* command)
|
||||||
{
|
{
|
||||||
BAutolock locker(fLocker);
|
BAutolock qlocker(fQueueLocker);
|
||||||
|
|
||||||
if (!fPendingCommands.AddItem(command)) {
|
if (!fPendingCommands.AddItem(command)) {
|
||||||
delete command;
|
delete command;
|
||||||
@@ -777,7 +779,7 @@ IMAPConnectionWorker::_EnqueueCommand(WorkerCommand* command)
|
|||||||
&& !command->IsContinuation())
|
&& !command->IsContinuation())
|
||||||
fSyncPending++;
|
fSyncPending++;
|
||||||
|
|
||||||
locker.Unlock();
|
qlocker.Unlock();
|
||||||
release_sem(fPendingCommandsSemaphore);
|
release_sem(fPendingCommandsSemaphore);
|
||||||
return B_OK;
|
return B_OK;
|
||||||
}
|
}
|
||||||
@@ -840,22 +842,33 @@ IMAPConnectionWorker::_SyncCommandDone()
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
bool
|
||||||
|
IMAPConnectionWorker::_IsQuitPending()
|
||||||
|
{
|
||||||
|
BAutolock locker(fQueueLocker);
|
||||||
|
WorkerCommand* nextCommand = fPendingCommands.ItemAt(0);
|
||||||
|
return dynamic_cast<QuitCommand*>(nextCommand) != NULL;
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
status_t
|
status_t
|
||||||
IMAPConnectionWorker::_Connect()
|
IMAPConnectionWorker::_Connect()
|
||||||
{
|
{
|
||||||
if (fProtocol.IsConnected())
|
if (fProtocol.IsConnected())
|
||||||
return B_OK;
|
return B_OK;
|
||||||
|
|
||||||
status_t status;
|
status_t status = B_INTERRUPTED;
|
||||||
int tries = 6;
|
int tries = 10;
|
||||||
while (tries-- > 0) {
|
while (tries-- > 0) {
|
||||||
|
if (_IsQuitPending())
|
||||||
|
break;
|
||||||
status = fProtocol.Connect(fSettings.ServerAddress(),
|
status = fProtocol.Connect(fSettings.ServerAddress(),
|
||||||
fSettings.Username(), fSettings.Password(), fSettings.UseSSL());
|
fSettings.Username(), fSettings.Password(), fSettings.UseSSL());
|
||||||
if (status == B_OK)
|
if (status == B_OK)
|
||||||
break;
|
break;
|
||||||
|
|
||||||
// Wait for 10 seconds, and try again
|
// Wait for 1 second, and try again
|
||||||
snooze(10000000);
|
snooze(1000000);
|
||||||
}
|
}
|
||||||
// TODO: if other workers are connected, but it fails for us, we need to
|
// TODO: if other workers are connected, but it fails for us, we need to
|
||||||
// remove this worker, and reduce the number of concurrent connections
|
// remove this worker, and reduce the number of concurrent connections
|
||||||
|
|||||||
@@ -40,6 +40,7 @@ public:
|
|||||||
bool UsesIdle() const { return fIdle; }
|
bool UsesIdle() const { return fIdle; }
|
||||||
|
|
||||||
status_t Run();
|
status_t Run();
|
||||||
|
thread_id Thread() const { return fThread; }
|
||||||
void Quit();
|
void Quit();
|
||||||
|
|
||||||
status_t EnqueueCheckSubscribedFolders();
|
status_t EnqueueCheckSubscribedFolders();
|
||||||
@@ -65,6 +66,7 @@ private:
|
|||||||
void _SyncCommandDone();
|
void _SyncCommandDone();
|
||||||
uint32 _MessagesExist() const
|
uint32 _MessagesExist() const
|
||||||
{ return fMessagesExist; }
|
{ return fMessagesExist; }
|
||||||
|
bool _IsQuitPending();
|
||||||
|
|
||||||
status_t _Connect();
|
status_t _Connect();
|
||||||
void _Disconnect();
|
void _Disconnect();
|
||||||
@@ -91,6 +93,7 @@ private:
|
|||||||
uint32 fMessagesExist;
|
uint32 fMessagesExist;
|
||||||
|
|
||||||
BLocker fLocker;
|
BLocker fLocker;
|
||||||
|
BLocker fQueueLocker;
|
||||||
thread_id fThread;
|
thread_id fThread;
|
||||||
bool fMain;
|
bool fMain;
|
||||||
bool fStopped;
|
bool fStopped;
|
||||||
|
|||||||
@@ -104,8 +104,10 @@ IMAPFolder::IMAPFolder(IMAPProtocol& protocol, const BString& mailboxName,
|
|||||||
|
|
||||||
IMAPFolder::~IMAPFolder()
|
IMAPFolder::~IMAPFolder()
|
||||||
{
|
{
|
||||||
if (!fFolderStateInitialized) {
|
MutexLocker locker(fLock);
|
||||||
|
if (!fFolderStateInitialized && fListener != NULL) {
|
||||||
fQuitFolderState = true;
|
fQuitFolderState = true;
|
||||||
|
locker.Unlock();
|
||||||
wait_for_thread(fReadFolderStateThread, NULL);
|
wait_for_thread(fReadFolderStateThread, NULL);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -38,8 +38,23 @@ IMAPProtocol::IMAPProtocol(const BMailAccountSettings& settings)
|
|||||||
|
|
||||||
IMAPProtocol::~IMAPProtocol()
|
IMAPProtocol::~IMAPProtocol()
|
||||||
{
|
{
|
||||||
}
|
MutexLocker locker(fWorkerLock);
|
||||||
|
std::vector<thread_id> threads;
|
||||||
|
for (int32 i = 0; i < fWorkers.CountItems(); i++) {
|
||||||
|
threads.push_back(fWorkers.ItemAt(i)->Thread());
|
||||||
|
fWorkers.ItemAt(i)->Quit();
|
||||||
|
}
|
||||||
|
locker.Unlock();
|
||||||
|
|
||||||
|
for (uint32 i = 0; i < threads.size(); i++)
|
||||||
|
wait_for_thread(threads[i], NULL);
|
||||||
|
|
||||||
|
FolderMap::iterator iterator = fFolders.begin();
|
||||||
|
for (; iterator != fFolders.end(); iterator++) {
|
||||||
|
IMAPFolder* folder = iterator->second;
|
||||||
|
delete folder; // to stop thread
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
status_t
|
status_t
|
||||||
IMAPProtocol::CheckSubscribedFolders(IMAP::Protocol& protocol, bool idle)
|
IMAPProtocol::CheckSubscribedFolders(IMAP::Protocol& protocol, bool idle)
|
||||||
@@ -70,7 +85,7 @@ IMAPProtocol::CheckSubscribedFolders(IMAP::Protocol& protocol, bool idle)
|
|||||||
|
|
||||||
if (newFolders.IsEmpty() && fWorkers.CountItems() == workersWanted) {
|
if (newFolders.IsEmpty() && fWorkers.CountItems() == workersWanted) {
|
||||||
// Nothing to do - we've already distributed everything
|
// Nothing to do - we've already distributed everything
|
||||||
return B_OK;
|
return _EnqueueCheckMailboxes();
|
||||||
}
|
}
|
||||||
|
|
||||||
// Remove mailboxes from workers
|
// Remove mailboxes from workers
|
||||||
@@ -187,8 +202,8 @@ IMAPProtocol::SyncMessages()
|
|||||||
worker->EnqueueCheckSubscribedFolders();
|
worker->EnqueueCheckSubscribedFolders();
|
||||||
return worker->Run();
|
return worker->Run();
|
||||||
}
|
}
|
||||||
|
fWorkers.ItemAt(0)->EnqueueCheckSubscribedFolders();
|
||||||
return _EnqueueCheckMailboxes();
|
return B_OK;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -49,7 +49,7 @@ public:
|
|||||||
read_flags flags = B_READ);
|
read_flags flags = B_READ);
|
||||||
|
|
||||||
virtual void MessageReceived(BMessage* message);
|
virtual void MessageReceived(BMessage* message);
|
||||||
const ::Settings* Settings() const { return &fSettings; };
|
const ::Settings* Settings() const { return &fSettings; }
|
||||||
|
|
||||||
protected:
|
protected:
|
||||||
virtual status_t HandleFetchBody(const entry_ref& ref,
|
virtual status_t HandleFetchBody(const entry_ref& ref,
|
||||||
|
|||||||
Reference in New Issue
Block a user