rf_diskqueue.c

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

  }
  SIGNAL_DISK_QUEUE( queue, "DiskIOEnqueue");
  RF_UNLOCK_QUEUE_MUTEX( queue, "DiskIOEnqueue" );
#endif /* KERNEL */
}
    
#if !defined(KERNEL) && !defined(SIMULATE)
/* user-level only: tell all threads to wake up & recheck the queue */
void rf_BroadcastOnQueue(queue)
  RF_DiskQueue_t *queue;
{
  int i;

  if (queue->maxOutstanding > 1) for (i=0; i<queue->maxOutstanding; i++) {
    SIGNAL_DISK_QUEUE(queue, "BroadcastOnQueue" );
  }
}
#endif /* !KERNEL && !SIMULATE */

#ifndef KERNEL /* not used in kernel */

RF_DiskQueueData_t *rf_DiskIODequeue(queue)
  RF_DiskQueue_t *queue;
{
  RF_DiskQueueData_t *p, *headItem;
  int tid;

  rf_get_threadid(tid);
  RF_LOCK_QUEUE_MUTEX( queue, "DiskIODequeue" );
  for (p=NULL; !p; ) {
    if (queue->unlockingOp) {
      /* unlocking request */
      RF_ASSERT(RF_QUEUE_LOCKED(queue));
      p = queue->unlockingOp;
      queue->unlockingOp = NULL;
      Dprintf4("[%d] dequeueing pri %d unlocking op r %d c %d\n", tid, p->priority, queue->row,queue->col);
    }
    else {
      headItem = (queue->qPtr->Peek)(queue->qHdr);
      if (headItem) {
        if (RF_LOCKING_REQ(headItem)) {
          /* locking request */
          if (!RF_QUEUE_LOCKED(queue)) {
            /* queue isn't locked, so dequeue the request & lock the queue */
            p = (queue->qPtr->Dequeue)( queue->qHdr );
            if (p)
              Dprintf4("[%d] dequeueing pri %d locking op r %d c %d\n", tid, p->priority, queue->row, queue->col);
            else
              Dprintf3("[%d] no dequeue -- raw queue empty r %d c %d\n", tid, queue->row, queue->col);
          }
          else {
            /* queue already locked, no dequeue occurs */
            Dprintf3("[%d] no dequeue -- queue is locked r %d c %d\n", tid, queue->row, queue->col);
            p = NULL;
          }
        }
        else {
          /* normal request, always dequeue and assume caller already has lock (if needed) */
          p = (queue->qPtr->Dequeue)( queue->qHdr );
          if (p)
            Dprintf4("[%d] dequeueing pri %d regular op r %d c %d\n", tid, p->priority, queue->row, queue->col);
          else
            Dprintf3("[%d] no dequeue -- raw queue empty r %d c %d\n", tid, queue->row, queue->col);
        }
      }
      else {
        Dprintf3("[%d] no dequeue -- raw queue empty r %d c %d\n", tid, queue->row, queue->col);
      }
    }

    if (queue->raidPtr->terminate_disk_queues) {
      p = NULL;
      break;
    }
#ifdef SIMULATE
    break;		/* in simulator, return NULL on empty queue instead of blocking */
#else /* SIMULATE */
    if (!p) {
      Dprintf3("[%d] nothing to dequeue: waiting r %d c %d\n", tid, queue->row, queue->col);
      WAIT_DISK_QUEUE( queue, "DiskIODequeue" );
    }
#endif /* SIMULATE */
  }

  if (p) {
    queue->queueLength--;  /* decrement count of number of requests waiting in this queue */
    RF_ASSERT(queue->queueLength >= 0);
    queue->numOutstanding++;
    queue->last_deq_sector = p->sectorOffset;
    /* record the amount of time this request spent in the disk queue */
    RF_ETIMER_STOP(p->qtime);
    RF_ETIMER_EVAL(p->qtime);
    if (p->tracerec)
      p->tracerec->diskqueue_us += RF_ETIMER_VAL_US(p->qtime);
  }

  if (p && RF_LOCKING_REQ(p)) {
    RF_ASSERT(!RF_QUEUE_LOCKED(queue));
    Dprintf3("[%d] locking queue r %d c %d\n",tid,queue->row,queue->col);
    RF_LOCK_QUEUE(queue);
  }
  RF_UNLOCK_QUEUE_MUTEX( queue, "DiskIODequeue" );
  
  return(p);
}

#else /* !KERNEL */

/* get the next set of I/Os started, kernel version only */
void rf_DiskIOComplete(queue, req, status)
  RF_DiskQueue_t      *queue;
  RF_DiskQueueData_t  *req;
  int                  status;
{
  int done=0;

  RF_LOCK_QUEUE_MUTEX( queue, "DiskIOComplete" );

  /* unlock the queue:
     (1) after an unlocking req completes
     (2) after a locking req fails
  */
  if (RF_UNLOCKING_REQ(req) || (RF_LOCKING_REQ(req) && status)) {
    Dprintf2("DiskIOComplete: unlocking queue at r %d c %d\n", queue->row, queue->col);
    RF_ASSERT(RF_QUEUE_LOCKED(queue) && (queue->unlockingOp == NULL));
    RF_UNLOCK_QUEUE(queue);
  }

  queue->numOutstanding--;
  RF_ASSERT(queue->numOutstanding >= 0);

  /* dispatch requests to the disk until we find one that we can't. */
  /* no reason to continue once we've filled up the queue */
  /* no reason to even start if the queue is locked */
  
  while (!done && !RF_QUEUE_FULL(queue) && !RF_QUEUE_LOCKED(queue)) {
    if (queue->nextLockingOp) {
      req = queue->nextLockingOp; queue->nextLockingOp = NULL;
      Dprintf3("DiskIOComplete: a pri %d locking req was pending at r %d c %d\n",req->priority,queue->row, queue->col);
    } else {
      req = (queue->qPtr->Dequeue)( queue->qHdr );
      Dprintf3("DiskIOComplete: extracting pri %d req from queue at r %d c %d\n",req->priority,queue->row, queue->col);
    }
    if (req) {
	queue->queueLength--;  /* decrement count of number of requests waiting in this queue */
	RF_ASSERT(queue->queueLength >= 0);
    }
    if (!req) done=1;
    else if (RF_LOCKING_REQ(req)) {
      if (RF_QUEUE_EMPTY(queue)) {                   					/* dispatch it */
	Dprintf3("DiskIOComplete: dispatching pri %d locking req to r %d c %d (queue empty)\n",req->priority,queue->row, queue->col);
	RF_LOCK_QUEUE(queue);
	rf_DispatchKernelIO(queue, req);
	done = 1;
      } else {                         		           /* put it aside to wait for the queue to drain */
	Dprintf3("DiskIOComplete: postponing pri %d locking req to r %d c %d\n",req->priority,queue->row, queue->col);
	RF_ASSERT(queue->nextLockingOp == NULL);
	queue->nextLockingOp = req;
	done = 1;
      }
    } else if (RF_UNLOCKING_REQ(req)) {      	/* should not happen: unlocking ops should not get queued */
      RF_ASSERT(RF_QUEUE_LOCKED(queue)); 			               /* support it anyway for the future */
      Dprintf3("DiskIOComplete: dispatching pri %d unl req to r %d c %d (SHOULD NOT SEE THIS)\n",req->priority,queue->row, queue->col);
      rf_DispatchKernelIO(queue, req);
      done = 1;
    } else if (RF_OK_TO_DISPATCH(queue, req)) {
      Dprintf3("DiskIOComplete: dispatching pri %d regular req to r %d c %d (ok to dispatch)\n",req->priority,queue->row, queue->col);
      rf_DispatchKernelIO(queue, req);
    } else {                                   		  /* we can't dispatch it, so just re-enqueue it.  */
      /* potential trouble here if disk queues batch reqs */
      Dprintf3("DiskIOComplete: re-enqueueing pri %d regular req to r %d c %d\n",req->priority,queue->row, queue->col);
      (queue->qPtr->Enqueue)(queue->qHdr, req, req->priority);
      done = 1;
    }
  }
  
  RF_UNLOCK_QUEUE_MUTEX( queue, "DiskIOComplete" );
}
#endif /* !KERNEL */

/* promotes accesses tagged with the given parityStripeID from low priority
 * to normal priority.  This promotion is optional, meaning that a queue
 * need not implement it.  If there is no promotion routine associated with
 * a queue, this routine does nothing and returns -1.
 */
int rf_DiskIOPromote(queue, parityStripeID, which_ru)
  RF_DiskQueue_t     *queue;
  RF_StripeNum_t      parityStripeID;
  RF_ReconUnitNum_t   which_ru;
{
  int retval;
  
  if (!queue->qPtr->Promote)
    return(-1);
  RF_LOCK_QUEUE_MUTEX( queue, "DiskIOPromote" );
  retval = (queue->qPtr->Promote)( queue->qHdr, parityStripeID, which_ru );
  RF_UNLOCK_QUEUE_MUTEX( queue, "DiskIOPromote" );
  return(retval);
}

RF_DiskQueueData_t *rf_CreateDiskQueueData(
  RF_IoType_t                typ,
  RF_SectorNum_t             ssect,
  RF_SectorCount_t           nsect,
  caddr_t                    buf,
  RF_StripeNum_t             parityStripeID,
  RF_ReconUnitNum_t          which_ru,
  int                      (*wakeF)(),
  void                      *arg,
  RF_DiskQueueData_t        *next,
  RF_AccTraceEntry_t        *tracerec,
  void                      *raidPtr,
  RF_DiskQueueDataFlags_t    flags,
  void                      *kb_proc)
{
  RF_DiskQueueData_t *p;

  RF_FREELIST_GET_INIT(rf_dqd_freelist,p,next,(RF_DiskQueueData_t *),init_dqd);

  p->sectorOffset  = ssect + rf_protectedSectors;
  p->numSector     = nsect;
  p->type          = typ;
  p->buf           = buf;
  p->parityStripeID= parityStripeID;
  p->which_ru      = which_ru;
  p->CompleteFunc  = wakeF;
  p->argument      = arg;
  p->next          = next;
  p->tracerec      = tracerec;
  p->priority      = RF_IO_NORMAL_PRIORITY;
  p->AuxFunc       = NULL;
  p->buf2          = NULL;
#ifdef SIMULATE
  p->owner         = rf_GetCurrentOwner();
#endif /* SIMULATE */
  p->raidPtr       = raidPtr;
  p->flags         = flags;
#ifdef KERNEL
  p->b_proc        = kb_proc;
#endif /* KERNEL */
  return(p);
}

RF_DiskQueueData_t *rf_CreateDiskQueueDataFull(
  RF_IoType_t                typ,
  RF_SectorNum_t             ssect,
  RF_SectorCount_t           nsect,
  caddr_t                    buf,
  RF_StripeNum_t             parityStripeID,
  RF_ReconUnitNum_t          which_ru,
  int                      (*wakeF)(),
  void                      *arg,
  RF_DiskQueueData_t        *next,
  RF_AccTraceEntry_t        *tracerec,
  int                        priority,
  int                      (*AuxFunc)(),
  caddr_t                    buf2,
  void                      *raidPtr,
  RF_DiskQueueDataFlags_t    flags,
  void                      *kb_proc)
{
  RF_DiskQueueData_t *p;

  RF_FREELIST_GET_INIT(rf_dqd_freelist,p,next,(RF_DiskQueueData_t *),init_dqd);

  p->sectorOffset  = ssect + rf_protectedSectors;
  p->numSector     = nsect;
  p->type          = typ;
  p->buf           = buf;
  p->parityStripeID= parityStripeID;
  p->which_ru      = which_ru;
  p->CompleteFunc  = wakeF;
  p->argument      = arg;
  p->next          = next;
  p->tracerec      = tracerec;
  p->priority      = priority;
  p->AuxFunc       = AuxFunc;
  p->buf2          = buf2;
#ifdef SIMULATE
  p->owner         = rf_GetCurrentOwner();
#endif /* SIMULATE */
  p->raidPtr       = raidPtr;
  p->flags         = flags;
#ifdef KERNEL
  p->b_proc        = kb_proc;
#endif /* KERNEL */
  return(p);
}

void rf_FreeDiskQueueData(p)
  RF_DiskQueueData_t  *p;
{
	RF_FREELIST_FREE_CLEAN(rf_dqd_freelist,p,next,clean_dqd);
}