page.c

分布式文件系统

  if (page) {
    if (page->size) {
      ioc_table_lock (table);
      table->cache_used += page->size;
      ioc_table_unlock (table);
    }
  }

  if (destroy_size) {
    ioc_table_lock (table);
    table->cache_used -= destroy_size;
    ioc_table_unlock (table);
  }

  if (ioc_need_prune (ioc_inode->table)) {
    ioc_prune (ioc_inode->table);
  }

  gf_log (this->name, GF_LOG_DEBUG, "fault frame %p returned", frame);
  pthread_mutex_destroy (&local->local_lock);

  STACK_DESTROY (frame->root);
  return 0;
}

/*
 * ioc_page_fault -
 * 
 * @ioc_inode:
 * @frame:
 * @fd:
 * @offset:
 *
 */
void
ioc_page_fault (ioc_inode_t *ioc_inode,
		call_frame_t *frame,
		fd_t *fd,
		off_t offset)
{
  ioc_table_t *table = ioc_inode->table;
  call_frame_t *fault_frame = copy_frame (frame);
  ioc_local_t *fault_local = calloc (1, sizeof (ioc_local_t));

  fault_frame->local = fault_local;
  pthread_mutex_init (&fault_local->local_lock, NULL);

  INIT_LIST_HEAD (&fault_local->fill_list);
  fault_local->pending_offset = offset;
  fault_local->pending_size = table->page_size;
  fault_local->inode = ioc_inode;

  gf_log (frame->this->name, GF_LOG_DEBUG,
	  "stack winding page fault for offset = %lld with frame %p",
	  offset, fault_frame);

  STACK_WIND (fault_frame, ioc_fault_cbk,
	      FIRST_CHILD(fault_frame->this),
	      FIRST_CHILD(fault_frame->this)->fops->readv,
	      fd, table->page_size, offset);
  return;
}

void
ioc_frame_fill (ioc_page_t *page,
		call_frame_t *frame,
		off_t offset,
		size_t size)
{
  ioc_local_t *local = frame->local;
  ioc_fill_t *fill = NULL;
  off_t src_offset = 0;
  off_t dst_offset = 0;
  ssize_t copy_size = 0;
  ioc_inode_t *ioc_inode = page->inode;
  
  gf_log (frame->this->name, GF_LOG_DEBUG,
	  "frame (%p) offset = %lld && size = %d && page->size = %d && wait_count = %d", 
	  frame, offset, size, page->size, local->wait_count);

  /* immediately move this page to the end of the page_lru list */
  list_move_tail (&page->page_lru, &ioc_inode->page_lru);
  /* fill from local->pending_offset to local->pending_size */
  if (local->op_ret != -1 && page->size) {
    if (offset > page->offset)
      /* offset is offset in file, convert it to offset in 
       * page */
      src_offset = offset - page->offset;
    else
      /* local->pending_offset is in previous page. do not fill until we
       * have filled all previous pages 
       */
      dst_offset = page->offset - offset;

    /* we have to copy from offset to either end of this page or till the 
     * requested size */
    copy_size = min (page->size - src_offset,
		     size - dst_offset);

    if (copy_size < 0) {
      /* if page contains fewer bytes and the required offset
	 is beyond the page size in the page */
      copy_size = src_offset = 0;
    }
    
    gf_log (page->inode->table->xl->name, GF_LOG_DEBUG,
	    "copy_size = %d && src_offset = %lld && dst_offset = %lld",
	    copy_size, src_offset, dst_offset);

    {
      ioc_fill_t *new = calloc (1, sizeof (*new));
      new->offset = page->offset;
      new->size = copy_size;
      new->refs = dict_ref (page->ref);
      new->count = iov_subset (page->vector,
			       page->count,
			       src_offset,
			       src_offset + copy_size,
			       NULL);
      new->vector = calloc (new->count, sizeof (struct iovec));
      new->count = iov_subset (page->vector,
			       page->count,
			       src_offset,
			       src_offset + copy_size,
			       new->vector);

      /* add the ioc_fill to fill_list for this frame */
      if (list_empty (&local->fill_list)) {
	/* if list is empty, then this is the first time we are filling 
	 * frame, add the ioc_fill_t to the end of list */
	list_add_tail (&new->list, &local->fill_list);
      } else {
	int8_t found = 0;
	/* list is not empty, we need to look for where this offset fits in 
	 * list */
	list_for_each_entry (fill, &local->fill_list, list) {
	    if (fill->offset > new->offset) {
	      found = 1;
	      break;
	    }
	}

	if (found) {
	  found = 0;
	  list_add_tail (&new->list, &fill->list);
	} else {
	  list_add_tail (&new->list, &local->fill_list);
	}
      }
    }
    local->op_ret += copy_size;
  }
}

/*
 * ioc_frame_unwind - frame unwinds only from here 
 *
 * @frame: call frame to unwind
 *
 * to be used only by ioc_frame_return(), when a frame has finished waiting on all pages, required
 *
 */
static void
ioc_frame_unwind (call_frame_t *frame)
{
  ioc_local_t *local = frame->local;
  ioc_fill_t *fill = NULL, *next = NULL;
  int32_t count = 0;
  struct iovec *vector = NULL;
  int32_t copied = 0;
  dict_t *refs = NULL;
  struct stat stbuf = {0,};
  int32_t op_ret = 0;

  //  ioc_local_lock (local);
  refs = get_new_dict ();
  refs->is_locked = 1;

  frame->local = NULL;

  if (list_empty (&local->fill_list)) {
    gf_log (frame->this->name, GF_LOG_DEBUG,
	    "frame(%p) has 0 entries in local->fill_list (offset = %lld && size = %d)",
	    frame, local->offset, local->size);
  }

  list_for_each_entry (fill, &local->fill_list, list) {
    count += fill->count;
  }

  vector = calloc (count, sizeof (*vector));
  
  list_for_each_entry_safe (fill, next, &local->fill_list, list) {
    memcpy (((char *)vector) + copied,
	    fill->vector,
	    fill->count * sizeof (*vector));
    
    copied += (fill->count * sizeof (*vector));

    dict_copy (fill->refs, refs);

    list_del (&fill->list);
    dict_unref (fill->refs);
    free (fill->vector);
    free (fill);
  }
  
  frame->root->rsp_refs = dict_ref (refs);
  
  op_ret = iov_length (vector, count);
  gf_log (frame->this->name, GF_LOG_DEBUG,
	  "frame(%p) unwinding with op_ret=%d", frame, op_ret);

  //  ioc_local_unlock (local);

  STACK_UNWIND (frame,
		op_ret,
		local->op_errno,
		vector,
		count,
		&stbuf);

  dict_unref (refs);
    
  pthread_mutex_destroy (&local->local_lock);
  free (local);
  free (vector);

  return;
}

/*
 * ioc_frame_return -
 * @frame:
 *
 * to be called only when a frame is waiting on an in-transit page
 */
void
ioc_frame_return (call_frame_t *frame)
{
  ioc_local_t *local = frame->local;
  int32_t wait_count;
  assert (local->wait_count > 0);

  ioc_local_lock (local);
  wait_count = --local->wait_count;
  ioc_local_unlock (local);

  if (!wait_count) {
    ioc_frame_unwind (frame);
  } 

  return;
}

/* 
 * ioc_page_wakeup -
 * @page:
 *
 * to be called only when a frame is waiting on an in-transit page
 */
void
ioc_page_wakeup (ioc_page_t *page)
{
  ioc_waitq_t *waitq = NULL, *trav = NULL;
  call_frame_t *frame = NULL;

  waitq = page->waitq;
  page->waitq = NULL;

  trav = waitq;
  page->ready = 1;
  gf_log (page->inode->table->xl->name, GF_LOG_DEBUG,
	  "page is %p && waitq = %p", page, waitq);
  
  for (trav = waitq; trav; trav = trav->next) {
    frame = trav->data; 
    ioc_frame_fill (page, frame, trav->pending_offset, trav->pending_size);
    /* we return to the frame, rest is left to frame to decide, 
     * whether to unwind or to wait for rest
     * of the region to be available */
    ioc_frame_return (frame);
  }

  for (trav = waitq; trav;) {
    ioc_waitq_t *next = trav->next;
    free (trav);
    trav = next;
  }
}


/*
 * ioc_page_error -
 * @page:
 * @op_ret:
 * @op_errno:
 *
 */
void
ioc_page_error (ioc_page_t *page,
		int32_t op_ret,
		int32_t op_errno)
{
  ioc_waitq_t *waitq = NULL, *trav = NULL;
  call_frame_t *frame = NULL;
  int64_t ret = 0;
  ioc_table_t *table = NULL;

  waitq = page->waitq;
  page->waitq = NULL;
  
  gf_log (page->inode->table->xl->name, GF_LOG_DEBUG,
	  "page error for page = %p & waitq = %p", page, waitq);

  for (trav = waitq; trav; trav = trav->next) {
    ioc_local_t *local = NULL;

    frame = trav->data;

    local = frame->local;
    ioc_local_lock (local);

    if (local->op_ret != -1) {
      local->op_ret = op_ret;
      local->op_errno = op_errno;
    }
    ioc_local_unlock (local);
    ioc_frame_return (frame);
  }

  for (trav = waitq; trav;) {
    ioc_waitq_t *next = trav->next;
    free (trav);
    trav = next;
  }

  table = page->inode->table;
  ret = ioc_page_destroy (page);

  if (ret != -1) {
    table->cache_used -= ret;
  }
}