rf_paritylogging.c

RAIDFrame是个非常好的磁盘阵列RAID仿真工具

    {
      if (i == 0)
	{
	  RF_Calloc(raidPtr->parityLogPool.parityLogs, 1, sizeof(RF_ParityLog_t), (RF_ParityLog_t *));
          if (raidPtr->parityLogPool.parityLogs == NULL) {
            RF_Free(raidPtr->parityLogBufferHeap, raidPtr->numParityLogs * raidPtr->numSectorsPerLog * raidPtr->bytesPerSector);
            return(ENOMEM);
          }
	  l = raidPtr->parityLogPool.parityLogs;
	}
      else
	{
	  RF_Calloc(l->next, 1, sizeof(RF_ParityLog_t), (RF_ParityLog_t *));
          if (l->next == NULL) {
            RF_Free(raidPtr->parityLogBufferHeap, raidPtr->numParityLogs * raidPtr->numSectorsPerLog * raidPtr->bytesPerSector);
            for(l=raidPtr->parityLogPool.parityLogs;l;l=next) {
              next = l->next;
              if (l->records)
                RF_Free(l->records, (raidPtr->numSectorsPerLog * sizeof(RF_ParityLogRecord_t)));
              RF_Free(l, sizeof(RF_ParityLog_t));
            }
            return(ENOMEM);
          }
	  l = l->next;
	}
      l->bufPtr = lHeapPtr;
      lHeapPtr += raidPtr->numSectorsPerLog * raidPtr->bytesPerSector;
      RF_Malloc(l->records, (raidPtr->numSectorsPerLog * sizeof(RF_ParityLogRecord_t)), (RF_ParityLogRecord_t *));
      if (l->records == NULL) {
        RF_Free(raidPtr->parityLogBufferHeap, raidPtr->numParityLogs * raidPtr->numSectorsPerLog * raidPtr->bytesPerSector);
        for(l=raidPtr->parityLogPool.parityLogs;l;l=next) {
          next = l->next;
          if (l->records)
            RF_Free(l->records, (raidPtr->numSectorsPerLog * sizeof(RF_ParityLogRecord_t)));
          RF_Free(l, sizeof(RF_ParityLog_t));
        }
        return(ENOMEM);
      }
    }
  rc = rf_ShutdownCreate(listp, rf_ShutdownParityLoggingPool, raidPtr);
  if (rc) {
    RF_ERRORMSG3("Unable to create shutdown entry file %s line %d rc=%d\n", __FILE__,
      __LINE__, rc);
    rf_ShutdownParityLoggingPool(raidPtr);
    return(rc);
  }

  /* build pool of region buffers */
  rc = rf_mutex_init(&raidPtr->regionBufferPool.mutex);
  if (rc) {
    RF_ERRORMSG3("Unable to init mutex file %s line %d rc=%d\n", __FILE__,
      __LINE__, rc);
    return(ENOMEM);
  }
  rc = rf_cond_init(&raidPtr->regionBufferPool.cond);
  if (rc) {
    RF_ERRORMSG3("Unable to init cond file %s line %d rc=%d\n", __FILE__,
      __LINE__, rc);
    rf_mutex_destroy(&raidPtr->regionBufferPool.mutex);
    return(ENOMEM);
  }
  raidPtr->regionBufferPool.bufferSize = raidPtr->regionLogCapacity * raidPtr->bytesPerSector;
  printf("regionBufferPool.bufferSize %d\n",raidPtr->regionBufferPool.bufferSize);
  raidPtr->regionBufferPool.totalBuffers = 1;  /* for now, only one region at a time may be reintegrated */
  raidPtr->regionBufferPool.availableBuffers = raidPtr->regionBufferPool.totalBuffers;
  raidPtr->regionBufferPool.availBuffersIndex = 0;
  raidPtr->regionBufferPool.emptyBuffersIndex = 0;
  RF_Malloc(raidPtr->regionBufferPool.buffers, raidPtr->regionBufferPool.totalBuffers * sizeof(caddr_t), (caddr_t *));
  if (raidPtr->regionBufferPool.buffers == NULL) {
    rf_mutex_destroy(&raidPtr->regionBufferPool.mutex);
    rf_cond_destroy(&raidPtr->regionBufferPool.cond);
    return(ENOMEM);
  }
  for (i = 0; i < raidPtr->regionBufferPool.totalBuffers; i++) {
    RF_Malloc(raidPtr->regionBufferPool.buffers[i], raidPtr->regionBufferPool.bufferSize * sizeof(char), (caddr_t));
    if (raidPtr->regionBufferPool.buffers == NULL) {
      rf_mutex_destroy(&raidPtr->regionBufferPool.mutex);
      rf_cond_destroy(&raidPtr->regionBufferPool.cond);
      for(j=0;j<i;j++) {
        RF_Free(raidPtr->regionBufferPool.buffers[i], raidPtr->regionBufferPool.bufferSize * sizeof(char));
      }
      RF_Free(raidPtr->regionBufferPool.buffers, raidPtr->regionBufferPool.totalBuffers * sizeof(caddr_t));
      return(ENOMEM);
    }
    printf("raidPtr->regionBufferPool.buffers[%d] = %lx\n", i, raidPtr->regionBufferPool.buffers[i]);
  }
  rc = rf_ShutdownCreate(listp, rf_ShutdownParityLoggingRegionBufferPool, raidPtr);
  if (rc) {
    RF_ERRORMSG3("Unable to create shutdown entry file %s line %d rc=%d\n", __FILE__,
      __LINE__, rc);
    rf_ShutdownParityLoggingRegionBufferPool(raidPtr);
    return(rc);
  }

  /* build pool of parity buffers */
  parityBufferCapacity = maxRegionParityRange;
  rc = rf_mutex_init(&raidPtr->parityBufferPool.mutex);
  if (rc) {
    RF_ERRORMSG3("Unable to init mutex file %s line %d rc=%d\n", __FILE__,
      __LINE__, rc);
    return(rc);
  }
  rc = rf_cond_init(&raidPtr->parityBufferPool.cond);
  if (rc) {
    RF_ERRORMSG3("Unable to init cond file %s line %d rc=%d\n", __FILE__,
      __LINE__, rc);
    rf_mutex_destroy(&raidPtr->parityBufferPool.mutex);
    return(ENOMEM);
  }
  raidPtr->parityBufferPool.bufferSize = parityBufferCapacity * raidPtr->bytesPerSector;
  printf("parityBufferPool.bufferSize %d\n",raidPtr->parityBufferPool.bufferSize);
  raidPtr->parityBufferPool.totalBuffers = 1;  /* for now, only one region at a time may be reintegrated */
  raidPtr->parityBufferPool.availableBuffers = raidPtr->parityBufferPool.totalBuffers;
  raidPtr->parityBufferPool.availBuffersIndex = 0;
  raidPtr->parityBufferPool.emptyBuffersIndex = 0;
  RF_Malloc(raidPtr->parityBufferPool.buffers, raidPtr->parityBufferPool.totalBuffers * sizeof(caddr_t), (caddr_t *));
  if (raidPtr->parityBufferPool.buffers == NULL) {
    rf_mutex_destroy(&raidPtr->parityBufferPool.mutex);
    rf_cond_destroy(&raidPtr->parityBufferPool.cond);
    return(ENOMEM);
  }
  for (i = 0; i < raidPtr->parityBufferPool.totalBuffers; i++) {
    RF_Malloc(raidPtr->parityBufferPool.buffers[i], raidPtr->parityBufferPool.bufferSize * sizeof(char), (caddr_t));
    if (raidPtr->parityBufferPool.buffers == NULL) {
      rf_mutex_destroy(&raidPtr->parityBufferPool.mutex);
      rf_cond_destroy(&raidPtr->parityBufferPool.cond);
      for(j=0;j<i;j++) {
        RF_Free(raidPtr->parityBufferPool.buffers[i], raidPtr->regionBufferPool.bufferSize * sizeof(char));
      }
      RF_Free(raidPtr->parityBufferPool.buffers, raidPtr->regionBufferPool.totalBuffers * sizeof(caddr_t));
      return(ENOMEM);
    }
    printf("parityBufferPool.buffers[%d] = %lx\n", i, raidPtr->parityBufferPool.buffers[i]);
  }
  rc = rf_ShutdownCreate(listp, rf_ShutdownParityLoggingParityBufferPool, raidPtr);
  if (rc) {
    RF_ERRORMSG3("Unable to create shutdown entry file %s line %d rc=%d\n", __FILE__,
      __LINE__, rc);
    rf_ShutdownParityLoggingParityBufferPool(raidPtr);
    return(rc);
  }

  /* initialize parityLogDiskQueue */
  rc = rf_create_managed_mutex(listp, &raidPtr->parityLogDiskQueue.mutex);
  if (rc) {
    RF_ERRORMSG3("Unable to init mutex file %s line %d rc=%d\n", __FILE__,
      __LINE__, rc);
    return(rc);
  }
  rc = rf_create_managed_cond(listp, &raidPtr->parityLogDiskQueue.cond);
  if (rc) {
    RF_ERRORMSG3("Unable to init cond file %s line %d rc=%d\n", __FILE__,
      __LINE__, rc);
    return(rc);
  }
  raidPtr->parityLogDiskQueue.flushQueue = NULL;
  raidPtr->parityLogDiskQueue.reintQueue = NULL;
  raidPtr->parityLogDiskQueue.bufHead = NULL;
  raidPtr->parityLogDiskQueue.bufTail = NULL;
  raidPtr->parityLogDiskQueue.reintHead = NULL;
  raidPtr->parityLogDiskQueue.reintTail = NULL;
  raidPtr->parityLogDiskQueue.logBlockHead = NULL;
  raidPtr->parityLogDiskQueue.logBlockTail = NULL;
  raidPtr->parityLogDiskQueue.reintBlockHead = NULL;
  raidPtr->parityLogDiskQueue.reintBlockTail = NULL;
  raidPtr->parityLogDiskQueue.freeDataList = NULL;
  raidPtr->parityLogDiskQueue.freeCommonList = NULL;

  rc = rf_ShutdownCreate(listp, rf_ShutdownParityLoggingDiskQueue, raidPtr);
  if (rc) {
    RF_ERRORMSG3("Unable to create shutdown entry file %s line %d rc=%d\n", __FILE__,
      __LINE__, rc);
    return(rc);
  }

  for (i = 0; i < rf_numParityRegions; i++)
    {
      rc = rf_mutex_init(&raidPtr->regionInfo[i].mutex);
      if (rc) {
        RF_ERRORMSG3("Unable to init mutex file %s line %d rc=%d\n", __FILE__,
          __LINE__, rc);
        for(j=0;j<i;j++)
          FreeRegionInfo(raidPtr, j);
        RF_Free(raidPtr->regionInfo, (rf_numParityRegions * sizeof(RF_RegionInfo_t)));
        return(ENOMEM);
      }
      rc = rf_mutex_init(&raidPtr->regionInfo[i].reintMutex);
      if (rc) {
        RF_ERRORMSG3("Unable to init mutex file %s line %d rc=%d\n", __FILE__,
          __LINE__, rc);
        rf_mutex_destroy(&raidPtr->regionInfo[i].mutex);
        for(j=0;j<i;j++)
          FreeRegionInfo(raidPtr, j);
        RF_Free(raidPtr->regionInfo, (rf_numParityRegions * sizeof(RF_RegionInfo_t)));
        return(ENOMEM);
      }
      raidPtr->regionInfo[i].reintInProgress = RF_FALSE;
      raidPtr->regionInfo[i].regionStartAddr = raidPtr->regionLogCapacity * i;
      raidPtr->regionInfo[i].parityStartAddr = raidPtr->regionParityRange * i;
      if (i < rf_numParityRegions - 1)
	{
	  raidPtr->regionInfo[i].capacity = raidPtr->regionLogCapacity;
	  raidPtr->regionInfo[i].numSectorsParity = raidPtr->regionParityRange;
	}
      else
	{
	  raidPtr->regionInfo[i].capacity = lastRegionCapacity;
	  raidPtr->regionInfo[i].numSectorsParity = raidPtr->sectorsPerDisk - raidPtr->regionParityRange * i;
	  if (raidPtr->regionInfo[i].numSectorsParity > maxRegionParityRange)
	    maxRegionParityRange = raidPtr->regionInfo[i].numSectorsParity;
	}
      raidPtr->regionInfo[i].diskCount = 0;
      RF_ASSERT(raidPtr->regionInfo[i].capacity + raidPtr->regionInfo[i].regionStartAddr <= totalLogCapacity);
      RF_ASSERT(raidPtr->regionInfo[i].parityStartAddr + raidPtr->regionInfo[i].numSectorsParity <= raidPtr->sectorsPerDisk);
      RF_Malloc(raidPtr->regionInfo[i].diskMap, (raidPtr->regionInfo[i].capacity * sizeof(RF_DiskMap_t)), (RF_DiskMap_t *));
      if (raidPtr->regionInfo[i].diskMap == NULL) {
        rf_mutex_destroy(&raidPtr->regionInfo[i].mutex);
        rf_mutex_destroy(&raidPtr->regionInfo[i].reintMutex);
        for(j=0;j<i;j++)
          FreeRegionInfo(raidPtr, j);
        RF_Free(raidPtr->regionInfo, (rf_numParityRegions * sizeof(RF_RegionInfo_t)));
        return(ENOMEM);
      }
      raidPtr->regionInfo[i].loggingEnabled = RF_FALSE;
      raidPtr->regionInfo[i].coreLog = NULL;
    }
  rc = rf_ShutdownCreate(listp, rf_ShutdownParityLoggingRegionInfo, raidPtr);
  if (rc) {
    RF_ERRORMSG3("Unable to create shutdown entry file %s line %d rc=%d\n", __FILE__,
      __LINE__, rc);
    rf_ShutdownParityLoggingRegionInfo(raidPtr);
    return(rc);
  }

  RF_ASSERT(raidPtr->parityLogDiskQueue.threadState == 0);
  raidPtr->parityLogDiskQueue.threadState = RF_PLOG_CREATED;
  rc = RF_CREATE_THREAD(raidPtr->pLogDiskThreadHandle, (void (*)())rf_ParityLoggingDiskManager, raidPtr);
  if (rc) {
    raidPtr->parityLogDiskQueue.threadState = 0;
    RF_ERRORMSG3("Unable to create parity logging disk thread file %s line %d rc=%d\n",
      __FILE__, __LINE__, rc);
    return(ENOMEM);
  }
  /* wait for thread to start */
  RF_LOCK_MUTEX(raidPtr->parityLogDiskQueue.mutex);
  while(!(raidPtr->parityLogDiskQueue.threadState&RF_PLOG_RUNNING)) {
    RF_WAIT_COND(raidPtr->parityLogDiskQueue.cond, raidPtr->parityLogDiskQueue.mutex);
  }
  RF_UNLOCK_MUTEX(raidPtr->parityLogDiskQueue.mutex);

  rc = rf_ShutdownCreate(listp, rf_ShutdownParityLogging, raidPtr);
  if (rc) {
    RF_ERRORMSG1("Got rc=%d adding parity logging shutdown event\n", rc);
    rf_ShutdownParityLogging(raidPtr);
    return(rc);
  }

  if (rf_parityLogDebug)
    {
      printf("                            size of disk log in sectors: %d\n", totalLogCapacity);
      printf("                            total number of parity regions is %d\n", rf_numParityRegions);
      printf("                            nominal sectors of log per parity region is %d\n", raidPtr->regionLogCapacity);
      printf("                            nominal region fragmentation is %d sectors\n",fragmentation);
      printf("                            total number of parity logs is %d\n", raidPtr->numParityLogs);
      printf("                            parity log size is %d sectors\n", raidPtr->numSectorsPerLog);
      printf("                            total in-core log space is %d bytes\n", rf_totalInCoreLogCapacity);
    }

  rf_EnableParityLogging(raidPtr);

  return(0);
}

static void FreeRegionInfo(
  RF_Raid_t      *raidPtr,
  RF_RegionId_t   regionID)
{
  RF_LOCK_MUTEX(raidPtr->regionInfo[regionID].mutex);
  RF_Free(raidPtr->regionInfo[regionID].diskMap, (raidPtr->regionInfo[regionID].capacity * sizeof(RF_DiskMap_t)));
  if (!rf_forceParityLogReint && raidPtr->regionInfo[regionID].coreLog) {
    rf_ReleaseParityLogs(raidPtr, raidPtr->regionInfo[regionID].coreLog);
    raidPtr->regionInfo[regionID].coreLog = NULL;
  }
  else {
    RF_ASSERT(raidPtr->regionInfo[regionID].coreLog == NULL);
    RF_ASSERT(raidPtr->regionInfo[regionID].diskCount == 0);
  }
  RF_UNLOCK_MUTEX(raidPtr->regionInfo[regionID].mutex);
  rf_mutex_destroy(&raidPtr->regionInfo[regionID].mutex);
  rf_mutex_destroy(&raidPtr->regionInfo[regionID].reintMutex);
}


static void FreeParityLogQueue(
  RF_Raid_t            *raidPtr,
  RF_ParityLogQueue_t  *queue)
{
  RF_ParityLog_t *l1, *l2;

  RF_LOCK_MUTEX(queue->mutex);
  l1 = queue->parityLogs;
  while (l1)
    {
      l2 = l1;
      l1 = l2->next;
      RF_Free(l2->records, (raidPtr->numSectorsPerLog * sizeof(RF_ParityLogRecord_t)));
      RF_Free(l2, sizeof(RF_ParityLog_t));
    }
  RF_UNLOCK_MUTEX(queue->mutex);
  rf_mutex_destroy(&queue->mutex);
}


static void FreeRegionBufferQueue(RF_RegionBufferQueue_t *queue)
{
  int i;

  RF_LOCK_MUTEX(queue->mutex);
  if (queue->availableBuffers != queue->totalBuffers)
    {
      printf("Attempt to free region queue which is still in use!\n");
      RF_ASSERT(0);
    }
  for (i = 0; i < queue->totalBuffers; i++)
    RF_Free(queue->buffers[i], queue->bufferSize);
  RF_Free(queue->buffers, queue->totalBuffers * sizeof(caddr_t));
  RF_UNLOCK_MUTEX(queue->mutex);
  rf_mutex_destroy(&queue->mutex);
}

static void rf_ShutdownParityLoggingRegionInfo(RF_ThreadArg_t arg)
{
  RF_Raid_t *raidPtr;
  RF_RegionId_t i;

  raidPtr = (RF_Raid_t *)arg;
  if (rf_parityLogDebug) {
    int tid;
    rf_get_threadid(tid);
    printf("[%d] ShutdownParityLoggingRegionInfo\n", tid);
  }
  /* free region information structs */
  for (i = 0; i < rf_numParityRegions; i++)
    FreeRegionInfo(raidPtr, i);
  RF_Free(raidPtr->regionInfo, (rf_numParityRegions * sizeof(raidPtr->regionInfo)));
  raidPtr->regionInfo = NULL;
}

static void rf_ShutdownParityLoggingPool(RF_ThreadArg_t arg)
{
  RF_Raid_t *raidPtr;

  raidPtr = (RF_Raid_t *)arg;
  if (rf_parityLogDebug) {
    int tid;
    rf_get_threadid(tid);
    printf("[%d] ShutdownParityLoggingPool\n", tid);
  }
  /* free contents of parityLogPool */
  FreeParityLogQueue(raidPtr, &raidPtr->parityLogPool);
  RF_Free(raidPtr->parityLogBufferHeap, raidPtr->numParityLogs * raidPtr->numSectorsPerLog * raidPtr->bytesPerSector);