/*      $NetBSD: rf_paritylogDiskMgr.c,v 1.31 2021/07/23 00:54:45 oster Exp $   */
/*
 * Copyright (c) 1995 Carnegie-Mellon University.
 * All rights reserved.
 *
 * Author: William V. Courtright II
 *
 * Permission to use, copy, modify and distribute this software and
 * its documentation is hereby granted, provided that both the copyright
 * notice and this permission notice appear in all copies of the
 * software, derivative works or modified versions, and any portions
 * thereof, and that both notices appear in supporting documentation.
 *
 * CARNEGIE MELLON ALLOWS FREE USE OF THIS SOFTWARE IN ITS "AS IS"
 * CONDITION.  CARNEGIE MELLON DISCLAIMS ANY LIABILITY OF ANY KIND
 * FOR ANY DAMAGES WHATSOEVER RESULTING FROM THE USE OF THIS SOFTWARE.
 *
 * Carnegie Mellon requests users of this software to return to
 *
 *  Software Distribution Coordinator  or  Software.Distribution@CS.CMU.EDU
 *  School of Computer Science
 *  Carnegie Mellon University
 *  Pittsburgh PA 15213-3890
 *
 * any improvements or extensions that they make and grant Carnegie the
 * rights to redistribute these changes.
 */
/* Code for flushing and reintegration operations related to parity logging.
 *
 */

#include <sys/cdefs.h>
__KERNEL_RCSID(0, "$NetBSD: rf_paritylogDiskMgr.c,v 1.31 2021/07/23 00:54:45 oster Exp $");

#include "rf_archs.h"

#if RF_INCLUDE_PARITYLOGGING > 0

#include <dev/raidframe/raidframevar.h>

#include "rf_threadstuff.h"
#include "rf_mcpair.h"
#include "rf_raid.h"
#include "rf_dag.h"
#include "rf_dagfuncs.h"
#include "rf_desc.h"
#include "rf_layout.h"
#include "rf_diskqueue.h"
#include "rf_paritylog.h"
#include "rf_general.h"
#include "rf_etimer.h"
#include "rf_paritylogging.h"
#include "rf_engine.h"
#include "rf_dagutils.h"
#include "rf_map.h"
#include "rf_parityscan.h"

#include "rf_paritylogDiskMgr.h"

static void *AcquireReintBuffer(RF_RegionBufferQueue_t *);

static void *
AcquireReintBuffer(RF_RegionBufferQueue_t *pool)
{
        void *bufPtr = NULL;

        /* Return a region buffer from the free list (pool). If the free list
         * is empty, WAIT. BLOCKING */

        rf_lock_mutex2(pool->mutex);
        if (pool->availableBuffers > 0) {
                bufPtr = pool->buffers[pool->availBuffersIndex];
                pool->availableBuffers--;
                pool->availBuffersIndex++;
                if (pool->availBuffersIndex == pool->totalBuffers)
                        pool->availBuffersIndex = 0;
                rf_unlock_mutex2(pool->mutex);
        } else {
                RF_PANIC();     /* should never happen in correct config,
                                 * single reint */
                rf_wait_cond2(pool->cond, pool->mutex);
        }
        return (bufPtr);
}

static void
ReleaseReintBuffer(
    RF_RegionBufferQueue_t * pool,
    void *bufPtr)
{
        /* Insert a region buffer (bufPtr) into the free list (pool).
         * NON-BLOCKING */

        rf_lock_mutex2(pool->mutex);
        pool->availableBuffers++;
        pool->buffers[pool->emptyBuffersIndex] = bufPtr;
        pool->emptyBuffersIndex++;
        if (pool->emptyBuffersIndex == pool->totalBuffers)
                pool->emptyBuffersIndex = 0;
        RF_ASSERT(pool->availableBuffers <= pool->totalBuffers);
        /*
         * XXXmrg this signal goes with the above "shouldn't happen" wait?
         */
        rf_signal_cond2(pool->cond);
        rf_unlock_mutex2(pool->mutex);
}



static void
ReadRegionLog(
    RF_RegionId_t regionID,
    RF_MCPair_t * rrd_mcpair,
    void *regionBuffer,
    RF_Raid_t * raidPtr,
    RF_DagHeader_t ** rrd_dag_h,
    RF_AllocListElem_t ** rrd_alloclist,
    RF_PhysDiskAddr_t ** rrd_pda)
{
        /* Initiate the read a region log from disk.  Once initiated, return
         * to the calling routine.
         *
         * NON-BLOCKING */

        RF_AccTraceEntry_t *tracerec;
        RF_DagNode_t *rrd_rdNode;

        /* create DAG to read region log from disk */
        rf_MakeAllocList(*rrd_alloclist);
        *rrd_dag_h = rf_MakeSimpleDAG(raidPtr, 1, 0, regionBuffer,
                                      rf_DiskReadFunc, rf_DiskReadUndoFunc,
                                      "Rrl", *rrd_alloclist,
                                      RF_DAG_FLAGS_NONE,
                                      RF_IO_NORMAL_PRIORITY);

        /* create and initialize PDA for the core log */
        *rrd_pda = rf_AllocPDAList(raidPtr, 1);
        rf_MapLogParityLogging(raidPtr, regionID, 0,
                               &((*rrd_pda)->col), &((*rrd_pda)->startSector));
        (*rrd_pda)->numSector = raidPtr->regionInfo[regionID].capacity;

        if ((*rrd_pda)->next) {
                (*rrd_pda)->next = NULL;
                printf("set rrd_pda->next to NULL\n");
        }
        /* initialize DAG parameters */
        tracerec = RF_Malloc(sizeof(*tracerec));
        (*rrd_dag_h)->tracerec = tracerec;
        rrd_rdNode = (*rrd_dag_h)->succedents[0]->succedents[0];
        rrd_rdNode->params[0].p = *rrd_pda;
/*  rrd_rdNode->params[1] = regionBuffer; */
        rrd_rdNode->params[2].v = 0;
        rrd_rdNode->params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0);

        /* launch region log read dag */
        rf_DispatchDAG(*rrd_dag_h, (void (*) (void *)) rf_MCPairWakeupFunc,
            (void *) rrd_mcpair);
}



static void
WriteCoreLog(
    RF_ParityLog_t * log,
    RF_MCPair_t * fwr_mcpair,
    RF_Raid_t * raidPtr,
    RF_DagHeader_t ** fwr_dag_h,
    RF_AllocListElem_t ** fwr_alloclist,
    RF_PhysDiskAddr_t ** fwr_pda)
{
        RF_RegionId_t regionID = log->regionID;
        RF_AccTraceEntry_t *tracerec;
        RF_SectorNum_t regionOffset;
        RF_DagNode_t *fwr_wrNode;

        /* Initiate the write of a core log to a region log disk. Once
         * initiated, return to the calling routine.
         *
         * NON-BLOCKING */

        /* create DAG to write a core log to a region log disk */
        rf_MakeAllocList(*fwr_alloclist);
        *fwr_dag_h = rf_MakeSimpleDAG(raidPtr, 1, 0, log->bufPtr,
                                      rf_DiskWriteFunc, rf_DiskWriteUndoFunc,
            "Wcl", *fwr_alloclist, RF_DAG_FLAGS_NONE, RF_IO_NORMAL_PRIORITY);

        *fwr_pda = rf_AllocPDAList(raidPtr, 1);
        regionOffset = log->diskOffset;
        rf_MapLogParityLogging(raidPtr, regionID, regionOffset,
                               &((*fwr_pda)->col),
                               &((*fwr_pda)->startSector));
        (*fwr_pda)->numSector = raidPtr->numSectorsPerLog;

        /* initialize DAG parameters */
        tracerec = RF_Malloc(sizeof(*tracerec));
        (*fwr_dag_h)->tracerec = tracerec;
        fwr_wrNode = (*fwr_dag_h)->succedents[0]->succedents[0];
        fwr_wrNode->params[0].p = *fwr_pda;
/*  fwr_wrNode->params[1] = log->bufPtr; */
        fwr_wrNode->params[2].v = 0;
        fwr_wrNode->params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0);

        /* launch the dag to write the core log to disk */
        rf_DispatchDAG(*fwr_dag_h, (void (*) (void *)) rf_MCPairWakeupFunc,
            (void *) fwr_mcpair);
}


static void
ReadRegionParity(
    RF_RegionId_t regionID,
    RF_MCPair_t * prd_mcpair,
    void *parityBuffer,
    RF_Raid_t * raidPtr,
    RF_DagHeader_t ** prd_dag_h,
    RF_AllocListElem_t ** prd_alloclist,
    RF_PhysDiskAddr_t ** prd_pda)
{
        /* Initiate the read region parity from disk. Once initiated, return
         * to the calling routine.
         *
         * NON-BLOCKING */

        RF_AccTraceEntry_t *tracerec;
        RF_DagNode_t *prd_rdNode;

        /* create DAG to read region parity from disk */
        rf_MakeAllocList(*prd_alloclist);
        *prd_dag_h = rf_MakeSimpleDAG(raidPtr, 1, 0, NULL, rf_DiskReadFunc,
                                      rf_DiskReadUndoFunc, "Rrp",
                                      *prd_alloclist, RF_DAG_FLAGS_NONE,
                                      RF_IO_NORMAL_PRIORITY);

        /* create and initialize PDA for region parity */
        *prd_pda = rf_AllocPDAList(raidPtr, 1);
        rf_MapRegionParity(raidPtr, regionID,
                           &((*prd_pda)->col), &((*prd_pda)->startSector),
                           &((*prd_pda)->numSector));
        if (rf_parityLogDebug)
                printf("[reading %d sectors of parity from region %d]\n",
                    (int) (*prd_pda)->numSector, regionID);
        if ((*prd_pda)->next) {
                (*prd_pda)->next = NULL;
                printf("set prd_pda->next to NULL\n");
        }
        /* initialize DAG parameters */
        tracerec = RF_Malloc(sizeof(*tracerec));
        (*prd_dag_h)->tracerec = tracerec;
        prd_rdNode = (*prd_dag_h)->succedents[0]->succedents[0];
        prd_rdNode->params[0].p = *prd_pda;
        prd_rdNode->params[1].p = parityBuffer;
        prd_rdNode->params[2].v = 0;
        prd_rdNode->params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0);
#if RF_DEBUG_VALIDATE_DAG
        if (rf_validateDAGDebug)
                rf_ValidateDAG(*prd_dag_h);
#endif
        /* launch region parity read dag */
        rf_DispatchDAG(*prd_dag_h, (void (*) (void *)) rf_MCPairWakeupFunc,
            (void *) prd_mcpair);
}

static void
WriteRegionParity(
    RF_RegionId_t regionID,
    RF_MCPair_t * pwr_mcpair,
    void *parityBuffer,
    RF_Raid_t * raidPtr,
    RF_DagHeader_t ** pwr_dag_h,
    RF_AllocListElem_t ** pwr_alloclist,
    RF_PhysDiskAddr_t ** pwr_pda)
{
        /* Initiate the write of region parity to disk. Once initiated, return
         * to the calling routine.
         *
         * NON-BLOCKING */

        RF_AccTraceEntry_t *tracerec;
        RF_DagNode_t *pwr_wrNode;

        /* create DAG to write region log from disk */
        rf_MakeAllocList(*pwr_alloclist);
        *pwr_dag_h = rf_MakeSimpleDAG(raidPtr, 1, 0, parityBuffer,
                                      rf_DiskWriteFunc, rf_DiskWriteUndoFunc,
                                      "Wrp", *pwr_alloclist,
                                      RF_DAG_FLAGS_NONE,
                                      RF_IO_NORMAL_PRIORITY);

        /* create and initialize PDA for region parity */
        *pwr_pda = rf_AllocPDAList(raidPtr, 1);
        rf_MapRegionParity(raidPtr, regionID,
                           &((*pwr_pda)->col), &((*pwr_pda)->startSector),
                           &((*pwr_pda)->numSector));

        /* initialize DAG parameters */
        tracerec = RF_Malloc(sizeof(*tracerec));
        (*pwr_dag_h)->tracerec = tracerec;
        pwr_wrNode = (*pwr_dag_h)->succedents[0]->succedents[0];
        pwr_wrNode->params[0].p = *pwr_pda;
/*  pwr_wrNode->params[1] = parityBuffer; */
        pwr_wrNode->params[2].v = 0;
        pwr_wrNode->params[3].v = RF_CREATE_PARAM3(RF_IO_NORMAL_PRIORITY, 0);

        /* launch the dag to write region parity to disk */
        rf_DispatchDAG(*pwr_dag_h, (void (*) (void *)) rf_MCPairWakeupFunc,
            (void *) pwr_mcpair);
}

static void
FlushLogsToDisk(
    RF_Raid_t * raidPtr,
    RF_ParityLog_t * logList)
{
        /* Flush a linked list of core logs to the log disk. Logs contain the
         * disk location where they should be written.  Logs were written in
         * FIFO order and that order must be preserved.
         *
         * Recommended optimizations: 1) allow multiple flushes to occur
         * simultaneously 2) coalesce contiguous flush operations
         *
         * BLOCKING */

        RF_ParityLog_t *log;
        RF_RegionId_t regionID;
        RF_MCPair_t *fwr_mcpair;
        RF_DagHeader_t *fwr_dag_h;
        RF_AllocListElem_t *fwr_alloclist;
        RF_PhysDiskAddr_t *fwr_pda;

        fwr_mcpair = rf_AllocMCPair(raidPtr);
        RF_LOCK_MCPAIR(fwr_mcpair);

        RF_ASSERT(logList);
        log = logList;
        while (log) {
                regionID = log->regionID;

                /* create and launch a DAG to write the core log */
                if (rf_parityLogDebug)
                        printf("[initiating write of core log for region %d]\n", regionID);
                fwr_mcpair->flag = RF_FALSE;
                WriteCoreLog(log, fwr_mcpair, raidPtr, &fwr_dag_h,
                             &fwr_alloclist, &fwr_pda);

                /* wait for the DAG to complete */
                while (!fwr_mcpair->flag)
                        RF_WAIT_MCPAIR(fwr_mcpair);
                if (fwr_dag_h->status != rf_enable) {
                        RF_ERRORMSG1("Unable to write core log to disk (region %d)\n", regionID);
                        RF_ASSERT(0);
                }
                /* RF_Free(fwr_pda, sizeof(RF_PhysDiskAddr_t)); */
                rf_FreePhysDiskAddr(raidPtr, fwr_pda);
                rf_FreeDAG(fwr_dag_h);
                rf_FreeAllocList(fwr_alloclist);

                log = log->next;
        }
        RF_UNLOCK_MCPAIR(fwr_mcpair);
        rf_FreeMCPair(raidPtr, fwr_mcpair);
        rf_ReleaseParityLogs(raidPtr, logList);
}

static void
ReintegrateRegion(
    RF_Raid_t * raidPtr,
    RF_RegionId_t regionID,
    RF_ParityLog_t * coreLog)
{
        RF_MCPair_t *rrd_mcpair = NULL, *prd_mcpair, *pwr_mcpair;
        RF_DagHeader_t *rrd_dag_h = NULL, *prd_dag_h, *pwr_dag_h;
        RF_AllocListElem_t *rrd_alloclist = NULL, *prd_alloclist, *pwr_alloclist;
        RF_PhysDiskAddr_t *rrd_pda = NULL, *prd_pda, *pwr_pda;
        void *parityBuffer, *regionBuffer = NULL;

        /* Reintegrate a region (regionID).
         *
         * 1. acquire region and parity buffers
         * 2. read log from disk
         * 3. read parity from disk
         * 4. apply log to parity
         * 5. apply core log to parity
         * 6. write new parity to disk
         *
         * BLOCKING */

        if (rf_parityLogDebug)
                printf("[reintegrating region %d]\n", regionID);

        /* initiate read of region parity */
        if (rf_parityLogDebug)
                printf("[initiating read of parity for region %d]\n",regionID);
        parityBuffer = AcquireReintBuffer(&raidPtr->parityBufferPool);
        prd_mcpair = rf_AllocMCPair(raidPtr);
        RF_LOCK_MCPAIR(prd_mcpair);
        prd_mcpair->flag = RF_FALSE;
        ReadRegionParity(regionID, prd_mcpair, parityBuffer, raidPtr,
                         &prd_dag_h, &prd_alloclist, &prd_pda);

        /* if region log nonempty, initiate read */
        if (raidPtr->regionInfo[regionID].diskCount > 0) {
                if (rf_parityLogDebug)
                        printf("[initiating read of disk log for region %d]\n",
                               regionID);
                regionBuffer = AcquireReintBuffer(&raidPtr->regionBufferPool);
                rrd_mcpair = rf_AllocMCPair(raidPtr);
                RF_LOCK_MCPAIR(rrd_mcpair);
                rrd_mcpair->flag = RF_FALSE;
                ReadRegionLog(regionID, rrd_mcpair, regionBuffer, raidPtr,
                              &rrd_dag_h, &rrd_alloclist, &rrd_pda);
        }
        /* wait on read of region parity to complete */
        while (!prd_mcpair->flag) {
                RF_WAIT_MCPAIR(prd_mcpair);
        }
        RF_UNLOCK_MCPAIR(prd_mcpair);
        if (prd_dag_h->status != rf_enable) {
                RF_ERRORMSG("Unable to read parity from disk\n");
                /* add code to fail the parity disk */
                RF_ASSERT(0);
        }
        /* apply core log to parity */
        /* if (coreLog) ApplyLogsToParity(coreLog, parityBuffer); */

        if (raidPtr->regionInfo[regionID].diskCount > 0) {
                /* wait on read of region log to complete */
                while (!rrd_mcpair->flag)
                        RF_WAIT_MCPAIR(rrd_mcpair);
                RF_UNLOCK_MCPAIR(rrd_mcpair);
                if (rrd_dag_h->status != rf_enable) {
                        RF_ERRORMSG("Unable to read region log from disk\n");
                        /* add code to fail the log disk */
                        RF_ASSERT(0);
                }
                /* apply region log to parity */
                /* ApplyRegionToParity(regionID, regionBuffer, parityBuffer); */
                /* release resources associated with region log */
                /* RF_Free(rrd_pda, sizeof(RF_PhysDiskAddr_t)); */
                rf_FreePhysDiskAddr(raidPtr, rrd_pda);
                rf_FreeDAG(rrd_dag_h);
                rf_FreeAllocList(rrd_alloclist);
                rf_FreeMCPair(raidPtr, rrd_mcpair);
                ReleaseReintBuffer(&raidPtr->regionBufferPool, regionBuffer);
        }
        /* write reintegrated parity to disk */
        if (rf_parityLogDebug)
                printf("[initiating write of parity for region %d]\n",
                       regionID);
        pwr_mcpair = rf_AllocMCPair(raidPtr);
        RF_LOCK_MCPAIR(pwr_mcpair);
        pwr_mcpair->flag = RF_FALSE;
        WriteRegionParity(regionID, pwr_mcpair, parityBuffer, raidPtr,
                          &pwr_dag_h, &pwr_alloclist, &pwr_pda);
        while (!pwr_mcpair->flag)
                RF_WAIT_MCPAIR(pwr_mcpair);
        RF_UNLOCK_MCPAIR(pwr_mcpair);
        if (pwr_dag_h->status != rf_enable) {
                RF_ERRORMSG("Unable to write parity to disk\n");
                /* add code to fail the parity disk */
                RF_ASSERT(0);
        }
        /* release resources associated with read of old parity */
        /* RF_Free(prd_pda, sizeof(RF_PhysDiskAddr_t)); */
        rf_FreePhysDiskAddr(raidPtr, prd_pda);
        rf_FreeDAG(prd_dag_h);
        rf_FreeAllocList(prd_alloclist);
        rf_FreeMCPair(raidPtr, prd_mcpair);

        /* release resources associated with write of new parity */
        ReleaseReintBuffer(&raidPtr->parityBufferPool, parityBuffer);
        /* RF_Free(pwr_pda, sizeof(RF_PhysDiskAddr_t)); */
        rf_FreePhysDiskAddr(raidPtr, pwr_pda);
        rf_FreeDAG(pwr_dag_h);
        rf_FreeAllocList(pwr_alloclist);
        rf_FreeMCPair(raidPtr, pwr_mcpair);

        if (rf_parityLogDebug)
                printf("[finished reintegrating region %d]\n", regionID);
}



static void
ReintegrateLogs(
    RF_Raid_t * raidPtr,
    RF_ParityLog_t * logList)
{
        RF_ParityLog_t *log, *freeLogList = NULL;
        RF_ParityLogData_t *logData, *logDataList;
        RF_RegionId_t regionID;

        RF_ASSERT(logList);
        while (logList) {
                log = logList;
                logList = logList->next;
                log->next = NULL;
                regionID = log->regionID;
                ReintegrateRegion(raidPtr, regionID, log);
                log->numRecords = 0;

                /* remove all items which are blocked on reintegration of this
                 * region */
                rf_lock_mutex2(raidPtr->parityLogDiskQueue.mutex);
                logData = rf_SearchAndDequeueParityLogData(raidPtr, regionID,
                           &raidPtr->parityLogDiskQueue.reintBlockHead,
                           &raidPtr->parityLogDiskQueue.reintBlockTail,
                                                           RF_TRUE);
                logDataList = logData;
                while (logData) {
                        logData->next = rf_SearchAndDequeueParityLogData(
                                         raidPtr, regionID,
                                         &raidPtr->parityLogDiskQueue.reintBlockHead,
                                         &raidPtr->parityLogDiskQueue.reintBlockTail,
                                         RF_TRUE);
                        logData = logData->next;
                }
                rf_unlock_mutex2(raidPtr->parityLogDiskQueue.mutex);

                /* process blocked log data and clear reintInProgress flag for
                 * this region */
                if (logDataList)
                        rf_ParityLogAppend(logDataList, RF_TRUE, &log, RF_TRUE);
                else {
                        /* Enable flushing for this region.  Holding both
                         * locks provides a synchronization barrier with
                         * DumpParityLogToDisk */
                        rf_lock_mutex2(raidPtr->regionInfo[regionID].mutex);
                        rf_lock_mutex2(raidPtr->regionInfo[regionID].reintMutex);
                        /* XXXmrg: don't need this? */
                        rf_lock_mutex2(raidPtr->parityLogDiskQueue.mutex);
                        raidPtr->regionInfo[regionID].diskCount = 0;
                        raidPtr->regionInfo[regionID].reintInProgress = RF_FALSE;
                        rf_unlock_mutex2(raidPtr->regionInfo[regionID].mutex);
                        rf_unlock_mutex2(raidPtr->regionInfo[regionID].reintMutex);     /* flushing is now
                                                                                         * enabled */
                        /* XXXmrg: don't need this? */
                        rf_unlock_mutex2(raidPtr->parityLogDiskQueue.mutex);
                }
                /* if log wasn't used, attach it to the list of logs to be
                 * returned */
                if (log) {
                        log->next = freeLogList;
                        freeLogList = log;
                }
        }
        if (freeLogList)
                rf_ReleaseParityLogs(raidPtr, freeLogList);
}

int
rf_ShutdownLogging(RF_Raid_t * raidPtr)
{
        /* shutdown parity logging 1) disable parity logging in all regions 2)
         * reintegrate all regions */

        RF_SectorCount_t diskCount;
        RF_RegionId_t regionID;
        RF_ParityLog_t *log;

        if (rf_parityLogDebug)
                printf("[shutting down parity logging]\n");
        /* Since parity log maps are volatile, we must reintegrate all
         * regions. */
        if (rf_forceParityLogReint) {
                for (regionID = 0; regionID < rf_numParityRegions; regionID++) {
                        rf_lock_mutex2(raidPtr->regionInfo[regionID].mutex);
                        raidPtr->regionInfo[regionID].loggingEnabled =
                                RF_FALSE;
                        log = raidPtr->regionInfo[regionID].coreLog;
                        raidPtr->regionInfo[regionID].coreLog = NULL;
                        diskCount = raidPtr->regionInfo[regionID].diskCount;
                        rf_unlock_mutex2(raidPtr->regionInfo[regionID].mutex);
                        if (diskCount > 0 || log != NULL)
                                ReintegrateRegion(raidPtr, regionID, log);
                        if (log != NULL)
                                rf_ReleaseParityLogs(raidPtr, log);
                }
        }
        if (rf_parityLogDebug) {
                printf("[parity logging disabled]\n");
                printf("[should be done!]\n");
        }
        return (0);
}

void
rf_ParityLoggingDiskManager(void *v)
{
        RF_Raid_t *raidPtr = v;
        RF_ParityLog_t *reintQueue, *flushQueue;
        int     workNeeded, done = RF_FALSE;
        int s;

        /* Main program for parity logging disk thread.  This routine waits
         * for work to appear in either the flush or reintegration queues and
         * is responsible for flushing core logs to the log disk as well as
         * reintegrating parity regions.
         *
         * BLOCKING */

        s = splbio();

        rf_lock_mutex2(raidPtr->parityLogDiskQueue.mutex);

        /*
         * Inform our creator that we're running. Don't bother doing the
         * mutex lock/unlock dance- we locked above, and we'll unlock
         * below with nothing to do, yet.
         */
        raidPtr->parityLogDiskQueue.threadState |= RF_PLOG_RUNNING;
        rf_signal_cond2(raidPtr->parityLogDiskQueue.cond);

        /* empty the work queues */
        flushQueue = raidPtr->parityLogDiskQueue.flushQueue;
        raidPtr->parityLogDiskQueue.flushQueue = NULL;
        reintQueue = raidPtr->parityLogDiskQueue.reintQueue;
        raidPtr->parityLogDiskQueue.reintQueue = NULL;
        workNeeded = (flushQueue || reintQueue);

        while (!done) {
                while (workNeeded) {
                        /* First, flush all logs in the flush queue, freeing
                         * buffers Second, reintegrate all regions which are
                         * reported as full. Third, append queued log data
                         * until blocked.
                         *
                         * Note: Incoming appends (ParityLogAppend) can block on
                         * either 1. empty buffer pool 2. region under
                         * reintegration To preserve a global FIFO ordering of
                         * appends, buffers are not released to the world
                         * until those appends blocked on buffers are removed
                         * from the append queue.  Similarly, regions which
                         * are reintegrated are not opened for general use
                         * until the append queue has been emptied. */

                        rf_unlock_mutex2(raidPtr->parityLogDiskQueue.mutex);

                        /* empty flushQueue, using free'd log buffers to
                         * process bufTail */
                        if (flushQueue)
                               FlushLogsToDisk(raidPtr, flushQueue);

                        /* empty reintQueue, flushing from reintTail as we go */
                        if (reintQueue)
                                ReintegrateLogs(raidPtr, reintQueue);

                        rf_lock_mutex2(raidPtr->parityLogDiskQueue.mutex);
                        flushQueue = raidPtr->parityLogDiskQueue.flushQueue;
                        raidPtr->parityLogDiskQueue.flushQueue = NULL;
                        reintQueue = raidPtr->parityLogDiskQueue.reintQueue;
                        raidPtr->parityLogDiskQueue.reintQueue = NULL;
                        workNeeded = (flushQueue || reintQueue);
                }
                /* no work is needed at this point */
                if (raidPtr->parityLogDiskQueue.threadState & RF_PLOG_TERMINATE) {
                        /* shutdown parity logging 1. disable parity logging
                         * in all regions 2. reintegrate all regions */
                        done = RF_TRUE; /* thread disabled, no work needed */
                        rf_unlock_mutex2(raidPtr->parityLogDiskQueue.mutex);
                        rf_ShutdownLogging(raidPtr);
                }
                if (!done) {
                        /* thread enabled, no work needed, so sleep */
                        if (rf_parityLogDebug)
                                printf("[parity logging disk manager sleeping]\n");
                        rf_wait_cond2(raidPtr->parityLogDiskQueue.cond,
                                      raidPtr->parityLogDiskQueue.mutex);
                        if (rf_parityLogDebug)
                                printf("[parity logging disk manager just woke up]\n");
                        flushQueue = raidPtr->parityLogDiskQueue.flushQueue;
                        raidPtr->parityLogDiskQueue.flushQueue = NULL;
                        reintQueue = raidPtr->parityLogDiskQueue.reintQueue;
                        raidPtr->parityLogDiskQueue.reintQueue = NULL;
                        workNeeded = (flushQueue || reintQueue);
                }
        }
        /*
         * Announce that we're done.
         */
        rf_lock_mutex2(raidPtr->parityLogDiskQueue.mutex);
        raidPtr->parityLogDiskQueue.threadState |= RF_PLOG_SHUTDOWN;
        rf_signal_cond2(raidPtr->parityLogDiskQueue.cond);
        rf_unlock_mutex2(raidPtr->parityLogDiskQueue.mutex);

        splx(s);

        /*
         * In the NetBSD kernel, the thread must exit; returning would
         * cause the proc trampoline to attempt to return to userspace.
         */
        kthread_exit(0);        /* does not return */
}
#endif                          /* RF_INCLUDE_PARITYLOGGING > 0 */