geos_one/bongo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

-- initial attempt at a new update system for long processes. doesn't quite work right and it is missing sections like cleanup.

Browse Source
This commit is contained in:
pfelt
2008-03-16 04:55:32 +00:00
parent 4778f84e82
commit d8f4a2c9c6
5 changed files with 88 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agents/imap/copy.c
+3 -3
View File
@@ -39,7 +39,6 @@ CopyMessageRangeToTarget(ImapSession *session, MessageInformation *message,
count--;
if (count > 0) {
message++;
DoProgressUpdate(session);
continue;
}
@@ -92,8 +91,9 @@ HandleCopy(ImapSession *session, BOOL ByUID)
FolderPath targetPath;
long ccode;
char *ptr2;
void *BusyHandle;
StartProgressUpdate(session, "Copied UID range");
BusyHandle = StartBusy(session);
if ((ccode = CheckState(session, STATE_SELECTED)) == STATUS_CONTINUE) {
if ((ccode = EventsSend(session, STORE_EVENT_ALL)) == STATUS_CONTINUE) {
@@ -112,7 +112,7 @@ HandleCopy(ImapSession *session, BOOL ByUID)
}
}
}
StopProgressUpdate(session);
StopBusy(BusyHandle);
return(ccode);
}
src/agents/imap/imapd.c
+25 -7
View File
@@ -2397,12 +2397,13 @@ ImapCommandClose(void *param)
{
ImapSession *session = (ImapSession *)param;
long ccode;
void *BusyHandle;
StartProgressUpdate(session, "Purging deleted messsages");
BusyHandle = StartBusy(session);
if ((ccode = CheckState(session, STATE_SELECTED)) == STATUS_CONTINUE) {
if (session->folder.selected.readOnly) {
FolderDeselect(session);
StopProgressUpdate(session);
StopBusy(BusyHandle);
return(SendOk(session, "CLOSE"));
}
@@ -2410,11 +2411,11 @@ ImapCommandClose(void *param)
FolderDeselect(session);
if (ccode == STATUS_CONTINUE) {
StopProgressUpdate(session);
StopBusy(BusyHandle);
return(SendOk(session, "CLOSE"));
}
}
StopProgressUpdate(session);
StopBusy(BusyHandle);
return(SendError(session->client.conn, session->command.tag, "CLOSE", ccode));
}
@@ -2424,8 +2425,9 @@ ImapCommandExpunge(void *param)
ImapSession *session = (ImapSession *)param;
OpenedFolder *selected = &session->folder.selected;
long ccode;
void *BusyHandle;
StartProgressUpdate(session, NULL);
BusyHandle = StartBusy(session);
if ((ccode = CheckState(session, STATE_SELECTED)) == STATUS_CONTINUE) {
if ((ccode = EventsSend(session, STORE_EVENT_ALL)) == STATUS_CONTINUE) {
@@ -2433,14 +2435,14 @@ ImapCommandExpunge(void *param)
if (!(selected->readOnly)) {
if ((ccode = PurgeDeletedMessages(session, TRUE, &(selected->message[selected->messageCount - 1]), selected->messageCount)) == STATUS_CONTINUE) {
if ((ccode = MessageListLoad(session->store.conn, selected)) == STATUS_CONTINUE) {
StopProgressUpdate(session);
StopBusy(BusyHandle);
return(SendOk(session, "EXPUNGE"));
}
}
}
}
}
StopProgressUpdate(session);
StopBusy(BusyHandle);
return(SendError(session->client.conn, session->command.tag, "EXPUNGE", ccode));
}
@@ -3413,6 +3415,10 @@ InitializeImapGlobals()
Imap.exiting = FALSE;
Imap.logHandle = NULL;
/* Initialize the Busy List and Semaphore */
Imap.list_Busy = NULL;
XplOpenLocalSemaphore(Imap.sem_Busy, 1);
/* Global allocations */
BongoKeywordIndexCreateFromTable(Imap.command.index, ImapProtocolCommands, .name, TRUE);
if (Imap.command.index) {
@@ -3600,6 +3606,15 @@ IMAPServer(void *unused)
return;
}
static void
BusyThread(void *unused)
{
while (!Imap.exiting) {
DoUpdate();
XplDelay(10000);
}
}
static void
IMAPSSLServer(void *unused)
{
@@ -3817,6 +3832,9 @@ XplServiceMain(int argc, char *argv[])
return 1;
}
/* start the busy thread */
XplBeginCountedThread(&id, BusyThread, IMAP_STACK_SIZE, NULL, ccode, Imap.server.active);
XplStartMainThread(PRODUCT_SHORT_NAME, &id, IMAPServer, 8192, NULL, ccode);
XplUnloadApp(XplGetThreadID());
src/agents/imap/imapd.h
+6 -3
View File
@@ -383,6 +383,9 @@ typedef struct {
char postmaster[MAXEMAILNAMESIZE + 1];
} server;
BongoList *list_Busy; /* Singly linked list of sessions that we should update every 10 seconds */
XplSemaphore sem_Busy; /* Semaphore protecting the busy list */
void *logHandle;
BOOL exiting;
} ImapGlobal;
@@ -407,9 +410,9 @@ long FolderListInitialize(ImapSession *session);
long MessageListLoad(Connection *conn, OpenedFolder *selected);
/* progress.c */
void DoProgressUpdate(ImapSession *session);
void StartProgressUpdate(ImapSession *session, char *message);
void StopProgressUpdate(ImapSession *session);
void DoUpdate(void);
void *StartBusy(ImapSession *session);
void StopBusy(void *handle);
void CommandFetchCleanup(void);
BOOL CommandFetchInit(void);
src/agents/imap/progress.c
+46 -34
View File
@@ -1,47 +1,59 @@
#include "imapd.h"
#include <time.h>
// Update our internal status and warn the client if it has been waiting
// for too long since it last heard from us.
/* traverse the global list and send oks to everyone */
void
DoProgressUpdate(ImapSession *session)
DoUpdate(void)
{
ProgressUpdate *status = session->progress;
time_t now = time(0);
if (status == NULL) return;
status->messages_processed++;
if ((now - status->last_update) > 10) {
// update the client - if we have no message, just Flush the output
if (status->message) {
ConnWriteF(session->client.conn,
"* OK - %s (processed %d since last update)\n",
status->message, status->messages_processed);
}
ConnFlush(session->client.conn);
status->last_update = now;
status->messages_processed = 0;
}
BongoList *cur = Imap.list_Busy;
ImapSession *session;
XplWaitOnLocalSemaphore(Imap.sem_Busy);
while(cur) {
session = (ImapSession *)cur->data;
ConnWriteStr(session->client.conn, "* OK - Still Busy\r\n");
ConnFlush(session->client.conn);
cur = cur->next;
}
XplSignalLocalSemaphore(Imap.sem_Busy);
}
void
StartProgressUpdate(ImapSession *session, char *message)
void *
StartBusy(ImapSession *session)
{
ProgressUpdate *p = MemMalloc(sizeof(ProgressUpdate));
if (p == NULL) return;
p->last_update = time(0);
p->messages_processed = 0;
p->message = message;
session->progress = p;
XplWaitOnLocalSemaphore(Imap.sem_Busy);
/* add ourselves to the list and return the pointer */
Imap.list_Busy = BongoListPrepend(Imap.list_Busy, session);
XplSignalLocalSemaphore(Imap.sem_Busy);
return Imap.list_Busy;
}
/* this function takes the handle returned from StartProgressUpdate */
void
StopProgressUpdate(ImapSession *session)
StopBusy(void *handle)
{
if (session->progress == NULL) return;
MemFree(session->progress);
session->progress = NULL;
BongoList *me = (BongoList *)handle;
XplWaitOnLocalSemaphore(Imap.sem_Busy);
/* unlink me */
if (me->prev || me->next) {
if (me->prev) {
me->prev->next = me->next;
}
if (me->next) {
me->next->prev = me->prev;
}
} else {
/* i was the last item in the array. null it */
Imap.list_Busy = NULL;
}
/* free me */
MemFree(me);
XplSignalLocalSemaphore(Imap.sem_Busy);
return;
}
src/agents/imap/store.c
+8 -7
View File
@@ -123,7 +123,6 @@ DoStoreForMessageRange(ImapSession *session, MessageInformation *message, unsign
if ((ccode = DoStoreForMessage(session, message, messageId, actionString, flags, silent, byUid)) == STATUS_CONTINUE) {
count--;
if (count > 0) {
DoProgressUpdate(session);
message++;
messageId++;
continue;
@@ -214,17 +213,18 @@ ImapCommandStore(void *param)
long ccode;
ImapSession *session = (ImapSession *)param;
BOOL purgedMessage = FALSE;
void *BusyHandle;
StartProgressUpdate(session, NULL);
BusyHandle = StartBusy(session);
if ((ccode = HandleStore(session, FALSE, &purgedMessage)) == STATUS_CONTINUE) {
if (!purgedMessage) {
StopProgressUpdate(session);
StopBusy(BusyHandle);
return(SendOk(session, "STORE"));
}
ccode = STATUS_REQUESTED_MESSAGE_NO_LONGER_EXISTS;
}
StopProgressUpdate(session);
StopBusy(BusyHandle);
return(SendError(session->client.conn, session->command.tag, "STORE", ccode));
}
@@ -234,17 +234,18 @@ ImapCommandUidStore(void *param)
long ccode;
ImapSession *session = (ImapSession *)param;
BOOL purgedMessage = FALSE;
void *BusyHandle;
StartProgressUpdate(session, NULL);
BusyHandle = StartBusy(session);
memmove(session->command.buffer, session->command.buffer + strlen("UID "), strlen(session->command.buffer + strlen("UID ")) + 1);
if ((ccode = HandleStore(session, TRUE, &purgedMessage)) == STATUS_CONTINUE) {
if (!purgedMessage) {
StopProgressUpdate(session);
StopBusy(BusyHandle);
return(SendOk(session, "UID STORE"));
}
ccode = STATUS_REQUESTED_MESSAGE_NO_LONGER_EXISTS;
}
StopProgressUpdate(session);
StopBusy(BusyHandle);
return(SendError(session->client.conn, session->command.tag, "UID STORE", ccode));
}