/*	$NetBSD: rf_paritylogDiskMgr.c,v 1.4 1999/08/13 03:41:57 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 "rf_archs.h"

#if RF_INCLUDE_PARITYLOGGING > 0

#include "rf_types.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_threadid.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 caddr_t AcquireReintBuffer(RF_RegionBufferQueue_t *);

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

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

	RF_LOCK_MUTEX(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_MUTEX(pool->mutex);
	} else {
		RF_PANIC();	/* should never happen in currect config,
				 * single reint */
		RF_WAIT_COND(pool->cond, pool->mutex);
	}
	return (bufPtr);
}

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

	RF_LOCK_MUTEX(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);
	RF_UNLOCK_MUTEX(pool->mutex);
	RF_SIGNAL_COND(pool->cond);
}



static void 
ReadRegionLog(
    RF_RegionId_t regionID,
    RF_MCPair_t * rrd_mcpair,
    caddr_t 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 */
	/* RF_Malloc(*rrd_pda, sizeof(RF_PhysDiskAddr_t), (RF_PhysDiskAddr_t
	 * *)); */
	*rrd_pda = rf_AllocPDAList(1);
	rf_MapLogParityLogging(raidPtr, regionID, 0, &((*rrd_pda)->row), &((*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 */
	bzero((char *) &tracerec, 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, 0, 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);

	/* create and initialize PDA for the region log */
	/* RF_Malloc(*fwr_pda, sizeof(RF_PhysDiskAddr_t), (RF_PhysDiskAddr_t
	 * *)); */
	*fwr_pda = rf_AllocPDAList(1);
	regionOffset = log->diskOffset;
	rf_MapLogParityLogging(raidPtr, regionID, regionOffset, &((*fwr_pda)->row), &((*fwr_pda)->col), &((*fwr_pda)->startSector));
	(*fwr_pda)->numSector = raidPtr->numSectorsPerLog;

	/* initialize DAG parameters */
	bzero((char *) &tracerec, 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, 0, 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,
    caddr_t 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 */
	/* RF_Malloc(*prd_pda, sizeof(RF_PhysDiskAddr_t), (RF_PhysDiskAddr_t
	 * *)); */
	*prd_pda = rf_AllocPDAList(1);
	rf_MapRegionParity(raidPtr, regionID, &((*prd_pda)->row), &((*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 */
	bzero((char *) &tracerec, 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, 0, 0);
	if (rf_validateDAGDebug)
		rf_ValidateDAG(*prd_dag_h);
	/* 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,
    caddr_t 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 */
	/* RF_Malloc(*pwr_pda, sizeof(RF_PhysDiskAddr_t), (RF_PhysDiskAddr_t
	 * *)); */
	*pwr_pda = rf_AllocPDAList(1);
	rf_MapRegionParity(raidPtr, regionID, &((*pwr_pda)->row), &((*pwr_pda)->col), &((*pwr_pda)->startSector), &((*pwr_pda)->numSector));

	/* initialize DAG parameters */
	bzero((char *) &tracerec, 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, 0, 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();
	RF_LOCK_MUTEX(fwr_mcpair->mutex);

	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_COND(fwr_mcpair->cond, fwr_mcpair->mutex);
		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(fwr_pda);
		rf_FreeDAG(fwr_dag_h);
		rf_FreeAllocList(fwr_alloclist);

		log = log->next;
	}
	RF_UNLOCK_MUTEX(fwr_mcpair->mutex);
	rf_FreeMCPair(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, *prd_dag_h, *pwr_dag_h;
	RF_AllocListElem_t *rrd_alloclist, *prd_alloclist, *pwr_alloclist;
	RF_PhysDiskAddr_t *rrd_pda, *prd_pda, *pwr_pda;
	caddr_t 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();
	RF_LOCK_MUTEX(prd_mcpair->mutex);
	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();
		RF_LOCK_MUTEX(rrd_mcpair->mutex);
		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_COND(prd_mcpair->cond, prd_mcpair->mutex);
	}
	RF_UNLOCK_MUTEX(prd_mcpair->mutex);
	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_COND(rrd_mcpair->cond, rrd_mcpair->mutex);
		RF_UNLOCK_MUTEX(rrd_mcpair->mutex);
		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(rrd_pda);
		rf_FreeDAG(rrd_dag_h);
		rf_FreeAllocList(rrd_alloclist);
		rf_FreeMCPair(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();
	RF_LOCK_MUTEX(pwr_mcpair->mutex);
	pwr_mcpair->flag = RF_FALSE;
	WriteRegionParity(regionID, pwr_mcpair, parityBuffer, raidPtr, &pwr_dag_h, &pwr_alloclist, &pwr_pda);
	while (!pwr_mcpair->flag)
		RF_WAIT_COND(pwr_mcpair->cond, pwr_mcpair->mutex);
	RF_UNLOCK_MUTEX(pwr_mcpair->mutex);
	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(prd_pda);
	rf_FreeDAG(prd_dag_h);
	rf_FreeAllocList(prd_alloclist);
	rf_FreeMCPair(prd_mcpair);

	/* release resources associated with write of new parity */
	ReleaseReintBuffer(&raidPtr->parityBufferPool, parityBuffer);
	/* RF_Free(pwr_pda, sizeof(RF_PhysDiskAddr_t)); */
	rf_FreePhysDiskAddr(pwr_pda);
	rf_FreeDAG(pwr_dag_h);
	rf_FreeAllocList(pwr_alloclist);
	rf_FreeMCPair(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_MUTEX(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_MUTEX(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_MUTEX(raidPtr->regionInfo[regionID].mutex);
			RF_LOCK_MUTEX(raidPtr->regionInfo[regionID].reintMutex);
			RF_LOCK_MUTEX(raidPtr->parityLogDiskQueue.mutex);
			raidPtr->regionInfo[regionID].diskCount = 0;
			raidPtr->regionInfo[regionID].reintInProgress = RF_FALSE;
			RF_UNLOCK_MUTEX(raidPtr->regionInfo[regionID].mutex);
			RF_UNLOCK_MUTEX(raidPtr->regionInfo[regionID].reintMutex);	/* flushing is now
											 * enabled */
			RF_UNLOCK_MUTEX(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_MUTEX(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_MUTEX(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);
}

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

	rf_assign_threadid();	/* don't remove this line */

	/* 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 */

	RF_LOCK_MUTEX(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_COND(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_MUTEX(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_MUTEX(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_MUTEX(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_COND(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_MUTEX(raidPtr->parityLogDiskQueue.mutex);
	raidPtr->parityLogDiskQueue.threadState |= RF_PLOG_SHUTDOWN;
	RF_UNLOCK_MUTEX(raidPtr->parityLogDiskQueue.mutex);
	RF_SIGNAL_COND(raidPtr->parityLogDiskQueue.cond);
#if defined(__NetBSD__) && defined(_KERNEL)
	/*
         * 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 */
#else
	return (0);
#endif
}
#endif				/* RF_INCLUDE_PARITYLOGGING > 0 */