rf_diskqueue.c

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

	rf_SstfPromote},

	{"scan", /* SCAN (two-way elevator) */
	rf_ScanCreate,
	rf_SstfEnqueue,
	rf_ScanDequeue,
	rf_ScanPeek,
	rf_SstfPromote},

	{"cscan", /* CSCAN (one-way elevator) */
	rf_CscanCreate,
	rf_SstfEnqueue,
	rf_CscanDequeue,
	rf_CscanPeek,
	rf_SstfPromote},

#if !defined(KERNEL) && RF_INCLUDE_QUEUE_RANDOM > 0
	/* to make a point to Chris :-> */
	{"random", /* random */
	rf_FifoCreate,
	rf_FifoEnqueue,
	rf_RandomDequeue,
	rf_RandomPeek,
	rf_FifoPromote},
#endif /* !KERNEL && RF_INCLUDE_QUEUE_RANDOM > 0 */
};
#define NUM_DISK_QUEUE_TYPES (sizeof(diskqueuesw)/sizeof(RF_DiskQueueSW_t))

static RF_FreeList_t *rf_dqd_freelist;

#define RF_MAX_FREE_DQD 256
#define RF_DQD_INC       16
#define RF_DQD_INITIAL   64

static int init_dqd(dqd)
  RF_DiskQueueData_t  *dqd;
{
#ifdef KERNEL
	dqd->bp = ubc_bufget();
	if (dqd->bp == NULL) {
		return(ENOMEM);
	}
#endif /* KERNEL */
	return(0);
}

static void clean_dqd(dqd)
  RF_DiskQueueData_t  *dqd;
{
#ifdef KERNEL
    ubc_buffree(dqd->bp);
#endif /* KERNEL */
}

/* configures a single disk queue */
static int config_disk_queue(
  RF_Raid_t            *raidPtr,
  RF_DiskQueue_t       *diskqueue,
  RF_RowCol_t           r, /* row & col -- debug only */
  RF_RowCol_t           c,
  RF_DiskQueueSW_t     *p,
  RF_SectorCount_t      sectPerDisk,
  dev_t                 dev,
  int                   maxOutstanding,
  RF_ShutdownList_t   **listp,
  RF_AllocListElem_t   *clList)
{
  int rc;

  diskqueue->row = r;
  diskqueue->col = c;
  diskqueue->qPtr = p;
  diskqueue->qHdr = (p->Create)(sectPerDisk, clList, listp);
  diskqueue->dev  = dev;
  diskqueue->numOutstanding = 0;
  diskqueue->queueLength = 0;
  diskqueue->maxOutstanding = maxOutstanding;
  diskqueue->curPriority    = RF_IO_NORMAL_PRIORITY;
  diskqueue->nextLockingOp  = NULL;
  diskqueue->unlockingOp    = NULL;
  diskqueue->numWaiting=0;
  diskqueue->flags = 0;
  diskqueue->raidPtr = raidPtr;
  rc = rf_create_managed_mutex(listp, &diskqueue->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, &diskqueue->cond);
  if (rc) {
    RF_ERRORMSG3("Unable to init cond file %s line %d rc=%d\n", __FILE__,
      __LINE__, rc);
    return(rc);
  }
  return(0);
}

static void rf_ShutdownDiskQueueSystem(ignored)
  void  *ignored;
{
  RF_FREELIST_DESTROY_CLEAN(rf_dqd_freelist,next,(RF_DiskQueueData_t *),clean_dqd);
}

int rf_ConfigureDiskQueueSystem(listp)
  RF_ShutdownList_t  **listp;
{
  int rc;

  RF_FREELIST_CREATE(rf_dqd_freelist, RF_MAX_FREE_DQD,
    RF_DQD_INC, sizeof(RF_DiskQueueData_t));
  if (rf_dqd_freelist == NULL)
    return(ENOMEM);
  rc = rf_ShutdownCreate(listp, rf_ShutdownDiskQueueSystem, NULL);
  if (rc) {
    RF_ERRORMSG3("Unable to add to shutdown list file %s line %d rc=%d\n",
      __FILE__, __LINE__, rc);
    rf_ShutdownDiskQueueSystem(NULL);
    return(rc);
  }
  RF_FREELIST_PRIME_INIT(rf_dqd_freelist, RF_DQD_INITIAL,next,
    (RF_DiskQueueData_t *),init_dqd);
  return(0);
}

#ifndef KERNEL
/* this is called prior to shutdown to wakeup everyone waiting on a disk queue
 * and tell them to exit
 */
void rf_TerminateDiskQueues(raidPtr)
  RF_Raid_t  *raidPtr;
{
  RF_RowCol_t r, c;

  raidPtr->terminate_disk_queues = 1;
  for (r=0; r<raidPtr->numRow; r++) {
    for (c=0; c<raidPtr->numCol + ((r==0) ? raidPtr->numSpare : 0); c++) {
      RF_LOCK_QUEUE_MUTEX(&raidPtr->Queues[r][c], "TerminateDiskQueues");
      RF_BROADCAST_COND(raidPtr->Queues[r][c].cond);
      RF_UNLOCK_QUEUE_MUTEX(&raidPtr->Queues[r][c], "TerminateDiskQueues");
    }
  }
}
#endif /* !KERNEL */

int rf_ConfigureDiskQueues(
  RF_ShutdownList_t  **listp,
  RF_Raid_t           *raidPtr,
  RF_Config_t         *cfgPtr)
{
  RF_DiskQueue_t **diskQueues, *spareQueues;
  RF_DiskQueueSW_t *p;
  RF_RowCol_t r, c;
  int rc, i;

  raidPtr->maxQueueDepth = cfgPtr->maxOutstandingDiskReqs;

  for(p=NULL,i=0;i<NUM_DISK_QUEUE_TYPES;i++) {
    if (!strcmp(diskqueuesw[i].queueType, cfgPtr->diskQueueType)) {
      p = &diskqueuesw[i];
      break;
    }
  }
  if (p == NULL) {
    RF_ERRORMSG2("Unknown queue type \"%s\".  Using %s\n",cfgPtr->diskQueueType, diskqueuesw[0].queueType);
    p = &diskqueuesw[0];
  }

  RF_CallocAndAdd(diskQueues, raidPtr->numRow, sizeof(RF_DiskQueue_t *), (RF_DiskQueue_t **), raidPtr->cleanupList);
  if (diskQueues == NULL) {
    return(ENOMEM);
  }
  raidPtr->Queues = diskQueues;
  for (r=0; r<raidPtr->numRow; r++) {
    RF_CallocAndAdd(diskQueues[r], raidPtr->numCol + ((r==0) ? raidPtr->numSpare : 0), sizeof(RF_DiskQueue_t), (RF_DiskQueue_t *), raidPtr->cleanupList);
    if (diskQueues[r] == NULL)
      return(ENOMEM);
    for (c=0; c<raidPtr->numCol; c++) {
      rc = config_disk_queue(raidPtr, &diskQueues[r][c], r, c, p,
        raidPtr->sectorsPerDisk, raidPtr->Disks[r][c].dev,
        cfgPtr->maxOutstandingDiskReqs, listp, raidPtr->cleanupList);
      if (rc)
        return(rc);
    }
  }

  spareQueues = &raidPtr->Queues[0][raidPtr->numCol];
  for (r=0; r<raidPtr->numSpare; r++) {
    rc = config_disk_queue(raidPtr, &spareQueues[r], 0, r, p,
      raidPtr->sectorsPerDisk, raidPtr->Disks[0][raidPtr->numCol+r].dev,
      cfgPtr->maxOutstandingDiskReqs, listp, raidPtr->cleanupList);
    if (rc)
      return(rc);
  }
  return(0);
}

/* Enqueue a disk I/O
 *
 * Unfortunately, we have to do things differently in the different
 * environments (simulator, user-level, kernel).
 * At user level, all I/O is blocking, so we have 1 or more threads/disk
 * and the thread that enqueues is different from the thread that dequeues.
 * In the kernel, I/O is non-blocking and so we'd like to have multiple
 * I/Os outstanding on the physical disks when possible.
 *
 * when any request arrives at a queue, we have two choices:
 *    dispatch it to the lower levels
 *    queue it up
 *
 * kernel rules for when to do what:
 *    locking request:  queue empty => dispatch and lock queue,
 *                      else queue it
 *    unlocking req  :  always dispatch it
 *    normal req     :  queue empty => dispatch it & set priority
 *                      queue not full & priority is ok => dispatch it
 *                      else queue it
 *
 * user-level rules:
 *    always enqueue.  In the special case of an unlocking op, enqueue
 *    in a special way that will cause the unlocking op to be the next
 *    thing dequeued.
 *
 * simulator rules:
 *    Do the same as at user level, with the sleeps and wakeups suppressed.
 */
void rf_DiskIOEnqueue(queue, req, pri)
  RF_DiskQueue_t      *queue;
  RF_DiskQueueData_t  *req;
  int                  pri;
{
  int tid;

  RF_ETIMER_START(req->qtime);
  rf_get_threadid(tid);
  RF_ASSERT(req->type == RF_IO_TYPE_NOP || req->numSector);
  req->priority = pri;

  if (rf_queueDebug && (req->numSector == 0)) {
    printf("Warning: Enqueueing zero-sector access\n");
  }
  
#ifdef KERNEL
  /*
   * kernel
   */
  RF_LOCK_QUEUE_MUTEX( queue, "DiskIOEnqueue" );
  /* locking request */
  if (RF_LOCKING_REQ(req)) {
    if (RF_QUEUE_EMPTY(queue)) {
      Dprintf3("Dispatching pri %d locking op to r %d c %d (queue empty)\n",pri,queue->row, queue->col);
      RF_LOCK_QUEUE(queue);
      rf_DispatchKernelIO(queue, req);
    } else {
      queue->queueLength++;  /* increment count of number of requests waiting in this queue */
      Dprintf3("Enqueueing pri %d locking op to r %d c %d (queue not empty)\n",pri,queue->row, queue->col);
      req->queue = (void *)queue;
      (queue->qPtr->Enqueue)(queue->qHdr, req, pri);
    }
  }
  /* unlocking request */
  else if (RF_UNLOCKING_REQ(req)) {           /* we'll do the actual unlock when this I/O completes */
    Dprintf3("Dispatching pri %d unlocking op to r %d c %d\n",pri,queue->row, queue->col);
    RF_ASSERT(RF_QUEUE_LOCKED(queue));
    rf_DispatchKernelIO(queue, req);
  }
  /* normal request */
  else if (RF_OK_TO_DISPATCH(queue, req)) {
    Dprintf3("Dispatching pri %d regular op to r %d c %d (ok to dispatch)\n",pri,queue->row, queue->col);
    rf_DispatchKernelIO(queue, req);
  } else {
    queue->queueLength++;  /* increment count of number of requests waiting in this queue */
    Dprintf3("Enqueueing pri %d regular op to r %d c %d (not ok to dispatch)\n",pri,queue->row, queue->col);
    req->queue = (void *)queue;
    (queue->qPtr->Enqueue)(queue->qHdr, req, pri);
  }
  RF_UNLOCK_QUEUE_MUTEX( queue, "DiskIOEnqueue" );
  
#else /* KERNEL */
  /*
   * user-level
   */
  RF_LOCK_QUEUE_MUTEX( queue, "DiskIOEnqueue" );
  queue->queueLength++;  /* increment count of number of requests waiting in this queue */
  /* unlocking request */
  if (RF_UNLOCKING_REQ(req)) {
    Dprintf4("[%d] enqueueing pri %d unlocking op & signalling r %d c %d\n", tid, pri, queue->row, queue->col);
    RF_ASSERT(RF_QUEUE_LOCKED(queue) && queue->unlockingOp == NULL);
    queue->unlockingOp = req;
  }
  /* locking and normal requests */
  else {
    req->queue = (void *)queue;
    Dprintf5("[%d] enqueueing pri %d %s op & signalling r %d c %d\n", tid, pri,
	     (RF_LOCKING_REQ(req)) ? "locking" : "regular",queue->row,queue->col);
    (queue->qPtr->Enqueue)(queue->qHdr, req, pri);