Changeset 759 for Whitix/branches

Timestamp:

07/09/08 15:26:36 (5 months ago)

Author:

mwhitworth

Message:

Add to journal code. Support for per-journal threads, flushing of metadata, and start of commit code.

Location:

Whitix/branches/fs/fs/journal

Files:

• init.c (modified) (2 diffs)
• journals.c (modified) (3 diffs)
• transaction.c (modified) (4 diffs)

Whitix/branches/fs/fs/journal/init.c

@@ -28 +28 @@
 struct Cache* journalCache=NULL, *journHandleCache=NULL, *journTransCache=NULL;
 
-static void JournalCommitter()
-{
-        ThrSetPriority(currThread, 1);
-
-        while (1)
-        {
-                ThrSuspendThread(currThread);
-                ThrSchedule();
-
-                KePrint("JournalCommitter was woken up\n");
-        }
-}
-
 int JournalInit()
 {
@@ -50 +37 @@
                 sizeof(struct JournalTrans), NULL, NULL, 0);
 
-        journalCommitter=ThrCreateKernelThread(JournalCommitter);
-
         return 0;
 }

Whitix/branches/fs/fs/journal/journals.c

@@ -44 +44 @@
 }
 
+struct JournalHead* JournalAllocHeader()
+{
+        return malloc(sizeof(struct JournalHead));
+}
+
+struct JournalHead* JournalAddHeader(struct Buffer* buffer)
+{
+        struct JournalHead* ret;
+
+        if (BufferJournal(buffer))
+        {
+                ret=BufferToJournHead(buffer);
+        }else{
+                ret=JournalAllocHeader();
+
+                buffer->flags |= 1 << BUFFER_JOURNAL;
+                buffer->privData=ret;
+                ret->buffer=buffer;
+                BufferGet(buffer);
+        }
+
+        return buffer->privData;
+}
+
+extern void SleepWakeup(void* data);
+
+void JournalCommitSetTimer(struct Timer* commitTimer, struct Journal* journal)
+{
+        /* Set up timer. */
+        commitTimer->func=SleepWakeup;
+        commitTimer->expires=journal->commitInterval*100000;
+        commitTimer->data=currThread;
+
+        TimerAdd(commitTimer);
+}
+
+/* Journal commit thread. */
+
+/* TODO: Move to commit.c */
+void JournalCommitter(void* arg)
+{
+        struct Journal* journal=(struct Journal*)arg;
+        struct Timer commitTimer;
+        INIT_WAITQUEUE_ENTRY(commitWait);
+
+        INIT_WAITQUEUE_HEAD(&journal->commitWait);
+        INIT_WAITQUEUE_HEAD(&journal->commitWaitDone);
+
+        KePrint(KERN_INFO "JournalCommitter: started journal with commit interval of %d"
+                " seconds\n", journal->commitInterval);
+
+        while (1)
+        {
+                /* There should be a transaction to commit. */
+                if (journal->commitSequence != journal->commitRequest)
+                {
+                        TimerRemove(&commitTimer);
+
+                        JournalCommitTransaction(journal);
+                }
+
+                /* May not always want timer running, esp. if we are waiting for commits. */
+                JournalCommitSetTimer(&commitTimer, journal);
+
+                WaitAddToQueue(&journal->commitWait, &commitWait);
+
+                ThrSuspendThread(currThread);
+                ThrSchedule(); /* Will return here when the thread is scheduled again */
+                WaitRemoveFromQueue(&journal->commitWait, &commitWait);
+                
+                /* Get the running transaction, if we timed out, and commit that. Check expires. */
+                if (journal->currTransaction && currTime.seconds >= journal->currTransaction->expires)
+                        journal->commitRequest=journal->currTransaction->transId;
+        }
+}
+
+int JournalNextLogBlock(struct Journal* journal, DWORD* ret)
+{
+        DWORD blockNum;
+
+        blockNum=journal->head;
+
+        journal->head++;
+        journal->free--;
+        
+        /* Check for wraparound! */
+
+        return JournalBlockMap(journal, blockNum, ret);
+}
+
+struct JournalHead* JournalGetDescriptorBuffer(struct Journal* journal)
+{
+        int err;
+        DWORD blockNum;
+        struct Buffer* buffer;
+
+        err=JournalNextLogBlock(journal, &blockNum);
+
+        if (err)
+                return NULL;
+
+        buffer=JournalBlockRead(journal, blockNum);
+
+        if (!buffer)
+                return NULL;
+
+        memset(buffer->data, 0, journal->sectorSize);
+
+        return JournalAddHeader(buffer);
+}
+
 struct Journal* JournalCreate(struct VNode* vNode)
 {
-        struct Journal* journal=JournalAllocate();
+        struct Journal* journal;
         DWORD blockNo;
         struct Buffer* headerBuf;
@@ -52 +163 @@
         struct VfsSuperBlock* superBlock=vNode->superBlock;
 
+        journal=JournalAllocate();
+
         if (!journal)
                 return NULL;
@@ -100 +213 @@
                 journal->version, journal->maxLength);
 
+        /* Start the per-journal commit kernel thread. */
+        journal->commitThread=ThrCreateKernelThreadArg(JournalCommitter, journal);
+        journal->commitInterval=JOURN_COMMIT_INTERVAL;
+
+        JournalRecover(journal, jSb);
+
+        /* Set up transaction sequence numbers. */
+        journal->commitSequence=journal->transSequence-1;
+        journal->commitRequest=journal->commitSequence;
+
+        journal->head=BeToCpu32(jSb->first);
+        journal->free=BeToCpu32(jSb->maxLen)-journal->head;
+
+        ThrStartThread(journal->commitThread);
+
         return journal;
 }

Whitix/branches/fs/fs/journal/transaction.c

@@ -18 +18 @@
 
 #include <fs/journal.h>
+#include <fs/vfs.h>
 #include <module.h>
 #include <print.h>
@@ -51 +52 @@
         ret->state=JTRANS_RUNNING;
         ret->transId=journal->transSequence++;
+        ret->expires=currTime.seconds+JOURN_COMMIT_INTERVAL;
+
+        /* TODO: Move to constructor. */
+        INIT_LIST_HEAD(&ret->reservedList);
+        INIT_LIST_HEAD(&ret->metaDataList);
 
         /* TODO: Set up timer. */
@@ -148 +154 @@
 SYMBOL_EXPORT(JournalStart);
 
+int JournalStop(struct JournalHandle* handle)
+{
+        struct JournalTrans* trans=handle->transaction;
+        struct Journal* journal=trans->journal;
+
+        currThread->currHandle=NULL;
+
+        trans->updates--;
+        trans->outstandingBlocks-=handle->numBlocks;
+
+        if (!trans->updates)
+        {
+                /* Wake up waiters. */
+        }
+
+        if (handle->sync == 1 || currTime.seconds >= trans->expires)
+        {
+                int id=trans->transId;
+
+                JournalStartCommit(journal, trans);
+
+                if (handle->sync == 1)
+                        JournalWaitCommit(journal, id);
+        }
+
+        MemCacheFree(journHandleCache, handle);
+
+        return 0;
+}
+
+SYMBOL_EXPORT(JournalStop);
+
+void JournalFileBuffer(struct JournalHead* head, struct JournalTrans* trans, int type)
+{
+        struct ListHead* list;
+
+        /* Already filed? */
+        if (head->transaction && head->list == type)
+                return;
+
+        /* 
+         * It's already in this transaction, but on a different list, so we remove it
+         * before adding it to the other list.
+         */
+
+        if (head->transaction == trans && head->list > 0 && head->list != type)
+                ListRemove(&head->next);
+
+        switch (type)
+        {
+                case JOURN_RESERVED:
+                        list=&trans->reservedList;
+                        break;
+
+                case JOURN_METADATA:
+                        list=&trans->metaDataList;
+                        break;
+
+                default:
+                        KePrint("FileBuffer: handle %d!\n", type);
+                        return;
+        }
+
+        ListAddTail(&head->next, list);
+        head->list=type;
+}
+
 /* Intent to modify a buffer for metadata update. */
 
@@ -158 +231 @@
         head=JournalAddHeader(buffer);
 
+        JournalLock(journal);
+
+        /* Check if buffer is locked. */
+
+        if (!head->transaction)
+        {
+                head->transaction=trans;
+                JournalFileBuffer(head, trans, JOURN_RESERVED);
+        }
+
+        JournalUnlock(journal);
+
         return 0;
 }
 
 SYMBOL_EXPORT(JournalGetWriteAccess);
+
+int JournalDirtyMetadata(struct JournalHandle* handle, struct Buffer* buffer)
+{
+        struct JournalTrans* trans=handle->transaction;
+        struct Journal* journal=trans->journal;
+        struct JournalHead* head=BufferToJournHead(buffer);
+
+        JournalLock(journal);
+
+        JournalFileBuffer(head, trans, JOURN_METADATA);
+
+        JournalUnlock(journal);
+
+        return 0;
+}
+
+SYMBOL_EXPORT(JournalDirtyMetadata);
+
+int JournalForceCommit(struct Journal* journal)
+{
+        struct JournalHandle* handle;
+
+        handle=JournalStart(journal, 1);
+
+        handle->sync=1;
+
+        /* The sync forces the commit to disk. */
+        JournalStop(handle);
+
+        return 0;
+}
+
+SYMBOL_EXPORT(JournalForceCommit);