windows-head-tail-queue-removal-20060525

while investigating the cause of the delayed write errors it was observed
that all of the sleep queues are LIFO.  This has the side effect of
encouraging starvation.  Changing the queues to FIFOs revealed a serious
problem affecting the use of all queues which use both head and tail
pointers.  The removal function osi_QRemove does not take a tail pointer
and therefore the pointer is always left hanging.  If the number of elements
ever drops to zero the queue becomes corrupted.

Added osi_QRemoveHT to be used whenever head and tail pointers are used.
Updated all callers in afsd.

diff --git a/src/WINNT/afsd/cm_aclent.c b/src/WINNT/afsd/cm_aclent.c
index 46799cdfca1d21d1768fd506d1f162aba8c2b681..f7cb725a86e0a368cb6671b52edd65e6814cfc67 100644 (file)
--- a/src/WINNT/afsd/cm_aclent.c
+++ b/src/WINNT/afsd/cm_aclent.c
@@ -82,7 +82,7 @@ long cm_FindACLCache(cm_scache_t *scp, cm_user_t *userp, afs_uint32 *rightsp)
         if (aclp->userp == userp) {
             if (aclp->tgtLifetime && aclp->tgtLifetime <= osi_Time()) {
                 /* ticket expired */
-                osi_QRemove((osi_queue_t **) &cm_data.aclLRUp, &aclp->q);
+                osi_QRemoveHT((osi_queue_t **) &cm_data.aclLRUp, (osi_queue_t **) &cm_data.aclLRUEndp, &aclp->q);
                 CleanupACLEnt(aclp);
 
                 /* move to the tail of the LRU queue */
@@ -96,7 +96,7 @@ long cm_FindACLCache(cm_scache_t *scp, cm_user_t *userp, afs_uint32 *rightsp)
                        cm_data.aclLRUEndp = (cm_aclent_t *) osi_QPrev(&aclp->q);
 
                    /* move to the head of the LRU queue */
-                   osi_QRemove((osi_queue_t **) &cm_data.aclLRUp, &aclp->q);
+                   osi_QRemoveHT((osi_queue_t **) &cm_data.aclLRUp, (osi_queue_t **) &cm_data.aclLRUEndp, &aclp->q);
                    osi_QAddH((osi_queue_t **) &cm_data.aclLRUp,
                               (osi_queue_t **) &cm_data.aclLRUEndp,
                               &aclp->q);
@@ -125,7 +125,7 @@ static cm_aclent_t *GetFreeACLEnt(cm_scache_t * scp)
 
     aclp = cm_data.aclLRUEndp;
     cm_data.aclLRUEndp = (cm_aclent_t *) osi_QPrev(&aclp->q);
-    osi_QRemove((osi_queue_t **) &cm_data.aclLRUp, &aclp->q);
+    osi_QRemoveHT((osi_queue_t **) &cm_data.aclLRUp, (osi_queue_t **) &cm_data.aclLRUEndp, &aclp->q);
 
     if (aclp->backp && scp != aclp->backp) {
         ascp = aclp->backp;

diff --git a/src/WINNT/afsd/cm_scache.c b/src/WINNT/afsd/cm_scache.c
index c2826926a264a6873a5e002d4ef731b086714589..80d49aced289576f368e111c3929ca5a38ec3087 100644 (file)
--- a/src/WINNT/afsd/cm_scache.c
+++ b/src/WINNT/afsd/cm_scache.c
@@ -46,7 +46,7 @@ void cm_AdjustLRU(cm_scache_t *scp)
 {
     if (scp == cm_data.scacheLRULastp)
         cm_data.scacheLRULastp = (cm_scache_t *) osi_QPrev(&scp->q);
-    osi_QRemove((osi_queue_t **) &cm_data.scacheLRUFirstp, &scp->q);
+    osi_QRemoveHT((osi_queue_t **) &cm_data.scacheLRUFirstp, (osi_queue_t **) &cm_data.scacheLRULastp, &scp->q);
     osi_QAdd((osi_queue_t **) &cm_data.scacheLRUFirstp, &scp->q);
     if (!cm_data.scacheLRULastp) 
         cm_data.scacheLRULastp = scp;

diff --git a/src/WINNT/afsd/cm_vnodeops.c b/src/WINNT/afsd/cm_vnodeops.c
index 65b3b228aec25ce54382d227e963312649cc3afb..4969b4f9aae41b2bfa849866d8d372291776e16f 100644 (file)
--- a/src/WINNT/afsd/cm_vnodeops.c
+++ b/src/WINNT/afsd/cm_vnodeops.c
@@ -3832,7 +3832,7 @@ long cm_UnlockByKey(cm_scache_t * scp,
 
             if (scp->fileLocksT == q)
                 scp->fileLocksT = osi_QPrev(q);
-            osi_QRemove(&scp->fileLocksH,q);
+            osi_QRemoveHT(&scp->fileLocksH, &scp->fileLocksT, q);
 
             if (IS_LOCK_CLIENTONLY(fileLock)) {
                 scp->clientLocks--;
@@ -4020,7 +4020,7 @@ long cm_Unlock(cm_scache_t *scp,
     lock_ObtainWrite(&cm_scacheLock);
     if (scp->fileLocksT == q)
         scp->fileLocksT = osi_QPrev(q);
-    osi_QRemove(&scp->fileLocksH, q);
+    osi_QRemoveHT(&scp->fileLocksH, &scp->fileLocksT, q);
 
     /*
      * Don't delete it here; let the daemon delete it, to simplify
@@ -4605,7 +4605,7 @@ long cm_RetryLock(cm_file_lock_t *oldFileLock, int client_is_dead)
        lock_ObtainWrite(&cm_scacheLock);
         if (scp->fileLocksT == &oldFileLock->fileq)
             scp->fileLocksT = osi_QPrev(&oldFileLock->fileq);
-        osi_QRemove(&scp->fileLocksH, &oldFileLock->fileq);
+        osi_QRemoveHT(&scp->fileLocksH, &scp->fileLocksT, &oldFileLock->fileq);
        lock_ReleaseWrite(&cm_scacheLock);
     } else if (code == 0 && IS_LOCK_WAITLOCK(oldFileLock)) {
         scp->serverLock = newLock;

diff --git a/src/WINNT/client_osi/libosi.def b/src/WINNT/client_osi/libosi.def
index d68cace828edf64e82b06fc39584b4fbf0876db7..8def9d03318d865aa1b14f2a01ccfb22ed774a2c 100644 (file)
--- a/src/WINNT/client_osi/libosi.def
+++ b/src/WINNT/client_osi/libosi.def
@@ -65,7 +65,8 @@ EXPORTS
        osi_LogPrint            @58
        osi_LogSaveString       @59
        osi_InitPanic           @60
-       osi_InitTraceOption @61
+       osi_InitTraceOption     @61
        osi_LogEvent0           @62
        osi_LogEvent            @63
-    osi_HexifyString    @64
+        osi_HexifyString        @64
+        osi_QRemoveHT           @65

diff --git a/src/WINNT/client_osi/osiqueue.c b/src/WINNT/client_osi/osiqueue.c
index 8963d2e9c31a3e0b219cf5babc26fc5e2bedfaf9..b4bce198d998652768519cca257e2afb28406255 100644 (file)
--- a/src/WINNT/client_osi/osiqueue.c
+++ b/src/WINNT/client_osi/osiqueue.c
@@ -93,19 +93,53 @@ void osi_QAddT(osi_queue_t **headpp, osi_queue_t **tailpp, osi_queue_t *eltp)
 
 void osi_QRemove(osi_queue_t **headpp, osi_queue_t *eltp)
 {
-       osi_queue_t *np;        /* next dude */
-
-       np = eltp->nextp;       /* useful for both paths */
+    osi_queue_t *np = eltp->nextp;     /* next dude */
+    osi_queue_t *pp = eltp->prevp;     /* prev dude */
+
+    if (eltp == *headpp) {
+       /* we're the first element in the list */
+       *headpp = np;
+       if (np) 
+           np->prevp = NULL;
+    }
+    else {
+       pp->nextp = np;
+       if (np) 
+           np->prevp = pp;
+    }
+    eltp->prevp = NULL;
+    eltp->nextp = NULL;
+}
 
-       if (eltp == *headpp) {
-               /* we're the first element in the list */
-               *headpp = np;
-               if (np) np->prevp = NULL;
-       }
-       else {
-               eltp->prevp->nextp = np;
-               if (np) np->prevp = eltp->prevp;
-       }
+void osi_QRemoveHT(osi_queue_t **headpp, osi_queue_t **tailpp, osi_queue_t *eltp)
+{
+    osi_queue_t *np = eltp->nextp;     /* next dude */
+    osi_queue_t *pp = eltp->prevp;     /* prev dude */
+
+    if (eltp == *headpp && eltp == *tailpp) 
+    {
+       *headpp = *tailpp = NULL;
+    }
+    else if (eltp == *headpp) {
+       /* we're the first element in the list */
+       *headpp = np;
+       if (np) 
+           np->prevp = NULL;
+    }  
+    else if (eltp == *tailpp) {
+       /* we're the last element in the list */
+       *tailpp = pp;
+       if (pp) 
+           pp->nextp = NULL;
+    }  
+    else {
+       if (pp)
+               pp->nextp = np;
+       if (np)
+               np->prevp = pp;
+    }
+    eltp->prevp = NULL; 
+    eltp->nextp = NULL;
 }
 
 void osi_InitQueue(void)

diff --git a/src/WINNT/client_osi/osiqueue.h b/src/WINNT/client_osi/osiqueue.h
index a6c3ea9847a77d2f16214391907cffb9795629ad..ab43f230d70e9606ac3e70c4990876b02177b66c 100644 (file)
--- a/src/WINNT/client_osi/osiqueue.h
+++ b/src/WINNT/client_osi/osiqueue.h
@@ -59,6 +59,11 @@ extern void osi_QAddH(osi_queue_t **headpp, osi_queue_t **tailpp, osi_queue_t *e
  */
 extern void osi_QRemove(osi_queue_t **headpp, osi_queue_t *eltp);
 
+/* remove an element from a queue with both head and tail pointers; 
+ * takes address of head and tail lists, and element to remove as parameters.
+ */
+extern void osi_QRemoveHT(osi_queue_t **headpp, osi_queue_t **tailpp, osi_queue_t *eltp);
+
 /* initialize the queue package */
 extern void osi_InitQueue(void);
 

diff --git a/src/WINNT/client_osi/osisleep.c b/src/WINNT/client_osi/osisleep.c
index b8ca754a8b5fed69f0716052566e9d464b57a870..b82dc6d20b3ea8873543a45e2df1c3c136154326 100644 (file)
--- a/src/WINNT/client_osi/osisleep.c
+++ b/src/WINNT/client_osi/osisleep.c
@@ -56,6 +56,7 @@ static CRITICAL_SECTION osi_critSec[OSI_SLEEPHASHSIZE];
  * should be ignored.
  */
 static osi_sleepInfo_t *osi_sleepers[OSI_SLEEPHASHSIZE];
+static osi_sleepInfo_t *osi_sleepersEnd[OSI_SLEEPHASHSIZE];
 
 /* allocate space for lock operations */
 osi_lockOps_t *osi_lockOps[OSI_NLOCKTYPES];
@@ -109,7 +110,7 @@ void osi_FreeSleepInfo(osi_sleepInfo_t *ap)
     if (ap->states & OSI_SLEEPINFO_INHASH) {
        ap->states &= ~OSI_SLEEPINFO_INHASH;
        idx = osi_SLEEPHASH(ap->value);
-       osi_QRemove((osi_queue_t **) &osi_sleepers[idx], &ap->q);
+       osi_QRemoveHT((osi_queue_t **) &osi_sleepers[idx], (osi_queue_t **) &osi_sleepersEnd[idx], &ap->q);
     }
 
     if (ap->states & OSI_SLEEPINFO_DELETED) {
@@ -226,6 +227,7 @@ void osi_Init(void)
        for(i=0;i<OSI_SLEEPHASHSIZE; i++) {
                InitializeCriticalSection(&osi_critSec[i]);
                osi_sleepers[i] = (osi_sleepInfo_t *) NULL;
+               osi_sleepersEnd[i] = (osi_sleepInfo_t *) NULL;
        }
 
        /* free list CS */
@@ -257,22 +259,24 @@ void osi_Init(void)
 
 void osi_TWait(osi_turnstile_t *turnp, int waitFor, void *patchp, CRITICAL_SECTION *releasep)
 {
-       osi_sleepInfo_t *sp;
-        unsigned int code;
+    osi_sleepInfo_t *sp;
+    unsigned int code;
 
-       sp = TlsGetValue(osi_SleepSlot);
-       if (sp == NULL) {
-               sp = osi_AllocSleepInfo();
-               TlsSetValue(osi_SleepSlot, sp);
-       }
-       else
-               sp->states = 0;
-       sp->refCount = 0;
-        sp->waitFor = waitFor;
-        sp->value = (LONG_PTR) patchp;
-        osi_QAdd((osi_queue_t **) &turnp->firstp, &sp->q);
-        if (!turnp->lastp) turnp->lastp = sp;
-        LeaveCriticalSection(releasep);
+    sp = TlsGetValue(osi_SleepSlot);
+    if (sp == NULL) {
+       sp = osi_AllocSleepInfo();
+       TlsSetValue(osi_SleepSlot, sp);
+    }
+    else {
+       sp->states = 0;
+    }
+    sp->refCount = 0;
+    sp->waitFor = waitFor;
+    sp->value = (LONG_PTR) patchp;
+    osi_QAddT((osi_queue_t **) &turnp->firstp, (osi_queue_t **) &turnp->lastp, &sp->q);
+    if (!turnp->lastp) 
+       turnp->lastp = sp;
+    LeaveCriticalSection(releasep);
 
        /* now wait for the signal */
        while(1) {
@@ -309,11 +313,12 @@ void osi_TSignal(osi_turnstile_t *turnp)
 {
        osi_sleepInfo_t *sp;
         
-       if (!turnp->lastp) return;
+       if (!turnp->lastp) 
+           return;
         
         sp = turnp->lastp;
        turnp->lastp = (osi_sleepInfo_t *) osi_QPrev(&sp->q);
-        osi_QRemove((osi_queue_t **) &turnp->firstp, &sp->q);
+        osi_QRemoveHT((osi_queue_t **) &turnp->firstp, (osi_queue_t **) &turnp->lastp, &sp->q);
         sp->states |= OSI_SLEEPINFO_SIGNALLED;
         ReleaseSemaphore(sp->sema, 1, (long *) 0);
 }
@@ -325,7 +330,7 @@ void osi_TBroadcast(osi_turnstile_t *turnp)
         
         while(sp = turnp->lastp) {
                turnp->lastp = (osi_sleepInfo_t *) osi_QPrev(&sp->q);
-               osi_QRemove((osi_queue_t **) &turnp->firstp, &sp->q);
+               osi_QRemoveHT((osi_queue_t **) &turnp->firstp, (osi_queue_t **) &turnp->lastp, &sp->q);
                sp->states |= OSI_SLEEPINFO_SIGNALLED;
                ReleaseSemaphore(sp->sema, 1, (long *) 0);
        }       /* while someone's still asleep */
@@ -370,7 +375,7 @@ void osi_TSignalForMLs(osi_turnstile_t *turnp, int stillHaveReaders, CRITICAL_SE
                  * the crit sec.
                  */
                turnp->lastp = (osi_sleepInfo_t *) osi_QPrev(&tsp->q);
-               osi_QRemove((osi_queue_t **) &turnp->firstp, &tsp->q);
+               osi_QRemoveHT((osi_queue_t **) &turnp->firstp, (osi_queue_t **) &turnp->lastp, &tsp->q);
                 
                /* do the patching required for lock obtaining */
                 if (tsp->waitFor & OSI_SLEEPINFO_W4WRITE) {
@@ -425,14 +430,15 @@ void osi_SleepSpin(LONG_PTR sleepValue, CRITICAL_SECTION *releasep)
        sp = osi_AllocSleepInfo();
        TlsSetValue(osi_SleepSlot, sp);
     }
-    else
+    else {
        sp->states = 0;
+    }
     sp->refCount = 0;
     sp->value = sleepValue;
     idx = osi_SLEEPHASH(sleepValue);
     csp = &osi_critSec[idx];
     EnterCriticalSection(csp);
-    osi_QAdd((osi_queue_t **) &osi_sleepers[idx], &sp->q);
+    osi_QAddT((osi_queue_t **) &osi_sleepers[idx], (osi_queue_t **) &osi_sleepersEnd[idx], &sp->q);
     sp->states |= OSI_SLEEPINFO_INHASH;
     LeaveCriticalSection(releasep);
     LeaveCriticalSection(csp);