rf_diskthreads.c

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

    int tid;
    rf_get_threadid(tid);
    printf("[%d] done waiting for diskthreads to complete\n", tid);
  }  
  return(0);
}

int rf_ConfigureDiskThreads(
  RF_ShutdownList_t  **listp,
  RF_Raid_t           *raidPtr,
  RF_Config_t         *cfgPtr)
{
  RF_RowCol_t r, c;
  int rc;
  RF_DiskQueue_t **diskQueues     = raidPtr->Queues;
  RF_DiskQueue_t *spareDiskQueues = &raidPtr->Queues[0][raidPtr->numCol];
  RF_RaidDisk_t **disks           = raidPtr->Disks;
  RF_RaidDisk_t *spareDisks       = &raidPtr->Disks[0][raidPtr->numCol];
#ifndef SIMULATE
  int t, threadsPerDisk         = raidPtr->maxQueueDepth;
#endif /* !SIMULATE */

#ifndef SIMULATE
  rc = rf_create_managed_mutex(listp, &raidPtr->diskthread_count_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->diskthread_count_cond);
  if (rc) {
    RF_ERRORMSG3("Unable to init cond file %s line %d rc=%d\n",
      __FILE__, __LINE__, rc);
    return(rc);
  }
#endif /* !SIMULATE */

  /* create threads for the disks */
  RF_CallocAndAdd(raidPtr->diskids, raidPtr->numRow, sizeof(RF_DiskId_t *), (RF_DiskId_t **), raidPtr->cleanupList);
  if (raidPtr->diskids == NULL)
    return(ENOMEM);
#ifndef SIMULATE
  RF_CallocAndAdd(raidPtr->diskthreads, raidPtr->numRow, sizeof(RF_Thread_t *), (RF_Thread_t **), raidPtr->cleanupList);
  if (raidPtr->diskthreads == NULL)
    return(ENOMEM);
#endif /* !SIMULATE */
  for (r=0; r<raidPtr->numRow; r++) {
    RF_CallocAndAdd(raidPtr->diskids[r], raidPtr->numCol, sizeof(struct RF_DiskId_s), (struct RF_DiskId_s *), raidPtr->cleanupList);
    if (raidPtr->diskids[r] == NULL)
      return(ENOMEM);
#ifndef SIMULATE
    RF_CallocAndAdd(raidPtr->diskthreads[r], raidPtr->numCol, sizeof(RF_Thread_t), (RF_Thread_t *), raidPtr->cleanupList);
    if (raidPtr->diskthreads[r] == NULL)
      return(ENOMEM);
#endif /* !SIMULATE */
    for (c=0; c<raidPtr->numCol; c++) {
      raidPtr->diskids[r][c].queue = &(diskQueues[r][c]);
      raidPtr->diskids[r][c].disk  = &(disks[r][c]);
      raidPtr->diskids[r][c].row   = r;
      raidPtr->diskids[r][c].col   = c;
      raidPtr->diskids[r][c].raidPtr = raidPtr;
#ifdef SIMULATE
      raidPtr->diskids[r][c].state   = RF_IDLE;
#else /* SIMULATE */
      for (t=0; t<threadsPerDisk; t++) {
        if (RF_CREATE_THREAD(raidPtr->diskthreads[r][c], disk_rw_func, &raidPtr->diskids[r][c]))
          RF_PANIC();
        RF_LOCK_MUTEX(raidPtr->diskthread_count_mutex);
        raidPtr->diskthreads_created++;
        RF_UNLOCK_MUTEX(raidPtr->diskthread_count_mutex);
      }
#endif /* SIMULATE */
    }
  }

/* create spare drives */

  /* create threads for the spare disks */
  if (raidPtr->numSpare) {
    RF_CallocAndAdd(raidPtr->sparediskids, raidPtr->numSpare, sizeof(RF_DiskId_t), (RF_DiskId_t *), raidPtr->cleanupList);
    if (raidPtr->sparediskids == NULL)
      return(ENOMEM);
#ifndef SIMULATE
    RF_CallocAndAdd(raidPtr->sparediskthreads, raidPtr->numSpare, sizeof(RF_Thread_t), (RF_Thread_t *), raidPtr->cleanupList);
    if (raidPtr->sparediskthreads == NULL)
      return(ENOMEM);
#endif /* !SIMULATE */
    for (r=0; r<raidPtr->numSpare; r++) {
      raidPtr->sparediskids[r].queue = &(spareDiskQueues[r]);
      raidPtr->sparediskids[r].disk  = &(spareDisks[r]);
      raidPtr->sparediskids[r].row   = 0;
      raidPtr->sparediskids[r].col   = r+raidPtr->numCol;
      raidPtr->sparediskids[r].raidPtr = raidPtr;
#ifdef SIMULATE
      raidPtr->sparediskids[r].state   = RF_IDLE;
#else /* SIMULATE */
      for (t=0; t<threadsPerDisk; t++) {
        if (RF_CREATE_THREAD(raidPtr->sparediskthreads[r], disk_rw_func, &raidPtr->sparediskids[r]))
          RF_PANIC();
        RF_LOCK_MUTEX(raidPtr->diskthread_count_mutex);
        raidPtr->diskthreads_created++;
        RF_UNLOCK_MUTEX(raidPtr->diskthread_count_mutex);
      }
#endif /* SIMULATE */
    }
  }

#ifndef SIMULATE
  if (rf_dtDebug) {
    int tid;
    rf_get_threadid(tid);
    printf("[%d] waiting for diskthreads to start\n", tid);
  }
  RF_LOCK_MUTEX(raidPtr->diskthread_count_mutex);
  while(raidPtr->diskthreads_running < raidPtr->diskthreads_created) {
    RF_WAIT_COND(raidPtr->diskthread_count_cond, raidPtr->diskthread_count_mutex);
  }
  RF_UNLOCK_MUTEX(raidPtr->diskthread_count_mutex);
  if (rf_dtDebug) {
    int tid;
    rf_get_threadid(tid);
    printf("[%d] done waiting for diskthreads to start\n", tid);
  }
#endif /* !SIMULATE */

  return(0);
}

#ifndef SIMULATE
static int open_raw_disk_file(fname)
  char  *fname;
{
  int fd;
  if ((fd = open(fname, O_RDWR, 0)) < 0) {
    RF_ERRORMSG1("Unable to open device file \"%s\"\n",fname);
    perror("open_raw_disk_file");
    exit(1);
  }
  return(fd);
}
#endif /* !SIMULATE */

#ifdef SIMULATE

int rf_SetDiskIdle(raidPtr, r, c)
  RF_Raid_t    *raidPtr;
  RF_RowCol_t   r;
  RF_RowCol_t   c;
{
  if((r==0) && (c>=raidPtr->numCol)){raidPtr->sparediskids[c-raidPtr->numCol].state = RF_IDLE;}
  else{raidPtr->diskids[r][c].state = RF_IDLE;}
  return(0);
}

int rf_ScanDiskQueues(raidPtr)
  RF_Raid_t  *raidPtr;
{
  RF_RowCol_t r, c;
  int t;
   
  t=0;
       
  for (r=0; r<raidPtr->numRow; r++) 

    for (c=0; c<raidPtr->numCol; c++) 
      if (raidPtr->diskids[r][c].state == RF_IDLE) t+=disk_rw_func(&raidPtr->diskids[r][c]);
		     
		       
  for (r=0; r<raidPtr->numSpare; r++) {
    if (raidPtr->sparediskids[r].state == RF_IDLE) t+=disk_rw_func(&raidPtr->sparediskids[r]);
  }
  return (t);

}

void rf_simulator_complete_io(id)
  RF_DiskId_t  *id;
{
	id->queue->numOutstanding--;
	RF_ASSERT(id->queue->numOutstanding >= 0);
}

static int disk_rw_func(id)
  RF_DiskId_t  *id;
{
  RF_DiskQueueData_t *req;
  RF_Raid_t *raidPtr;
  int numbytes, status = 0;
  RF_uint64 byteoffs;
#ifndef SIMULATE
  int i, numactual;
#endif /* !SIMULATE */
 
  double access_time;
  double total_access_time;
  double media_done_time;
  double wake_up_time;

  raidPtr = id->raidPtr;

  req = rf_DiskIODequeue(id->queue);       /* blocks thread until something available.  Can also terminate thread. */
  if (!req) {
	  /* there is nothing to do for now */
	  return(0);
  }

  /* reject the request if the disk is in any of the dead states */
  if (RF_DEAD_DISK(id->disk->status)) {
      status = -1; wake_up_time=rf_cur_time; goto out;
  }

  byteoffs = req->sectorOffset * id->disk->blockSize;
  numbytes = req->numSector * id->disk->blockSize;
  /* the mapping code must guarantee this property */
  RF_ASSERT(req->sectorOffset + req->numSector <= id->disk->numBlocks + rf_protectedSectors);
  if (req->type != RF_IO_TYPE_NOP && numbytes == 0) {
    if (id->row >= 0) printf("WARNING: zero-byte transfer to disk row %d col %d\n",id->row,id->col);
    else printf("WARNING: zero-byte transfer to spare disk %d\n",id->col);
  }


#ifndef SIMULATE
  id->queue->numOutstanding--;
  RF_ASSERT(id->queue->numOutstanding >= 0);
#endif /* !SIMULATE */

  switch (req->type)
    {
    case RF_IO_TYPE_NOP:         /* used primarily to unlock a locked queue */
      break;
    case RF_IO_TYPE_READ:
    case RF_IO_TYPE_WRITE:
      rf_Access_time(&access_time,rf_cur_time, req->sectorOffset, req->numSector, &id->disk->diskState, &media_done_time, 1 );
#if RF_KEEP_DISKSTATS > 0
      if (req->type == RF_IO_TYPE_READ)
        id->disk->nreads++;
      else
        id->disk->nwrites++;
#endif /* RF_KEEP_DISKSTATS > 0 */
      wake_up_time=rf_cur_time+access_time; /* should add previos computation time too */
      total_access_time= access_time;
      /* create an event and ret to main */
      break;
    default:
      RF_ERRORMSG1("Error: unknown access type '%c' in disk_rw_func\n",req->type);
      status = -1;
      wake_up_time=rf_cur_time;
      goto out;
      
      break;
    } /* end case */

  id->state = RF_BUSY; 

out:

#ifdef SIMULATE
  rf_DDEventRequest(wake_up_time, req->CompleteFunc,req->argument, req->owner, id->row, id->col, (RF_Raid_t *) req->raidPtr, (void *)id);
#else /* SIMULATE */
  rf_DDEventRequest(wake_up_time, req->CompleteFunc,req->argument, req->owner, id->row, id->col, (RF_Raid_t *) req->raidPtr, (void *)NULL);
#endif /* SIMULATE */

#if RF_DEMO > 0
  if (rf_demoMode)
    rf_update_disk_iops(1);
#endif /* RF_DEMO > 0 */

  /* unlock the disk queue if:
     (1) this is an unlocking req, independent of failure status of I/O
     (2) this is a locking req that failed
  */
  RF_LOCK_MUTEX(req->queue->mutex);
  if (RF_UNLOCKING_REQ(req)) {
    if (rf_queueDebug) printf("unlocking queue after pri %d req to r %d c %d\n",req->priority,id->queue->row, id->queue->col);
    RF_ASSERT(RF_QUEUE_LOCKED(id->queue));
    RF_UNLOCK_QUEUE(id->queue);
    id->queue->nextLockingOp = id->queue->unlockingOp = NULL;
  } else if (RF_LOCKING_REQ(req) && (status)) {
    RF_UNLOCK_QUEUE(id->queue);
  } else if (RF_LOCKING_REQ(req)) {
    if (rf_queueDebug) printf("completed pri %d locking req on r %d c %d\n",req->priority, id->queue->row, id->queue->col);
  } else if (rf_queueDebug) printf("completed pri %d regular req on r %d c %d\n",req->priority,id->queue->row, id->queue->col);
  RF_UNLOCK_MUTEX(req->queue->mutex);

  if (rf_dtDebug > 1) {
    printf("Disk op (r%d c%d) %c %ld %d buf 0x%lx stat %d ",
	   id->row,id->col,req->type,req->sectorOffset, req->numSector, req->buf, status);
  }
  
  if (status == 0) return (1);
  else return (0);
}

#else  /* SIMULATE */

/* this routine is executed by each of the disk threads.  It just grabs the next request off of
 * the disk queue and sends it down.
 */
static void disk_rw_func(id)
  RF_DiskId_t  *id;
{
  RF_DiskQueueData_t *req;
  RF_Raid_t *raidPtr;
  int numactual, numbytes, status, i;
  RF_uint64 byteoffs, sectOffset;
  int fd, tid;
  long nReads = 0, nWrites = 0, rdRtSum = 0, wrRtSum = 0, util;
  RF_Etimer_t timer;
  struct timeval st_time, end_time, elpsd;
  int media_time_ms;
  int initialized = 0, bus, target, lun;
  void *r_handle, *w_handle;        /* these are structures via which we do I/Os through the cam layer */

  raidPtr = id->raidPtr;