read-ahead.c

分布式文件系统

/*
   Copyright (c) 2006, 2007, 2008 Z RESEARCH, Inc. <http://www.zresearch.com>
   This file is part of GlusterFS.

   GlusterFS is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published
   by the Free Software Foundation; either version 3 of the License,
   or (at your option) any later version.

   GlusterFS is distributed in the hope that it will be useful, but
   WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
   General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program.  If not, see
   <http://www.gnu.org/licenses/>.
*/

/* 
   TODO:
   - handle O_DIRECT
   - maintain offset, flush on lseek
   - ensure efficient memory managment in case of random seek
*/

#ifndef _CONFIG_H
#define _CONFIG_H
#include "config.h"
#endif

#include "glusterfs.h"
#include "logging.h"
#include "dict.h"
#include "xlator.h"
#include "read-ahead.h"
#include <assert.h>
#include <sys/time.h>


static void
read_ahead (call_frame_t *frame,
            ra_file_t *file);


int32_t
ra_open_cbk (call_frame_t *frame,
             void *cookie,
             xlator_t *this,
             int32_t op_ret,
             int32_t op_errno,
             fd_t *fd)
{
  ra_local_t *local = frame->local;
  ra_conf_t *conf = this->private;

  if (op_ret != -1) {
    ra_file_t *file = calloc (1, sizeof (*file));

    file = ra_file_ref (file);
    file->fd = fd;
    dict_set (fd->ctx, this->name,
              data_from_static_ptr (file));

    /* If mandatory locking has been enabled on this file,
       we disable caching on it */

    if ((fd->inode->st_mode & S_ISGID) && !(fd->inode->st_mode & S_IXGRP))
      file->disabled = 1;

    /* If O_DIRECT open, we disable caching on it */

    if ((local->flags & O_DIRECT) || (local->flags & O_WRONLY))
      file->disabled = 1;

    file->offset = (unsigned long long) 0;
    //    file->size = fd->inode->buf.st_size;
    file->conf = conf;
    file->pages.next = &file->pages;
    file->pages.prev = &file->pages;
    file->pages.offset = (unsigned long long) 0;
    file->pages.file = file;

    ra_conf_lock (conf);
    {
      file->next = conf->files.next;
      conf->files.next = file;
      file->next->prev = file;
      file->prev = &conf->files;
    }
    ra_conf_unlock (conf);

    file->page_count = conf->page_count;
    file->page_size = conf->page_size;
    pthread_mutex_init (&file->file_lock, NULL);

    if (!file->disabled) {
      file->page_count = 1;
      read_ahead (frame, file);
    }
  }

  freee (local->file_loc.path);
  freee (local);
  frame->local = NULL;

  STACK_UNWIND (frame, op_ret, op_errno, fd);

  return 0;
}

int32_t
ra_create_cbk (call_frame_t *frame,
               void *cookie,
               xlator_t *this,
               int32_t op_ret,
               int32_t op_errno,
               fd_t *fd,
               inode_t *inode,
               struct stat *buf)
{
  ra_local_t *local = frame->local;
  ra_conf_t *conf = this->private;

  if (op_ret != -1) {
    ra_file_t *file = calloc (1, sizeof (*file));
    file = ra_file_ref (file);

    file->fd = fd;
    dict_set (fd->ctx, this->name,
              data_from_static_ptr (file));

    /* If mandatory locking has been enabled on this file,
       we disable caching on it */

    if ((fd->inode->st_mode & S_ISGID) && !(fd->inode->st_mode & S_IXGRP))
      file->disabled = 1;

    /* If O_DIRECT open, we disable caching on it */

    if ((local->flags & O_DIRECT) || (local->flags & O_WRONLY))
      file->disabled = 1;

    file->offset = (unsigned long long) 0;
    //file->size = fd->inode->buf.st_size;
    file->conf = conf;
    file->pages.next = &file->pages;
    file->pages.prev = &file->pages;
    file->pages.offset = (unsigned long long) 0;
    file->pages.file = file;

    ra_conf_lock (conf);
    {
      file->next = conf->files.next;
      conf->files.next = file;
      file->next->prev = file;
      file->prev = &conf->files;
    }
    ra_conf_unlock (conf);

    file->page_count = conf->page_count;
    file->page_size = conf->page_size;
    pthread_mutex_init (&file->file_lock, NULL);
  }

  freee (local->file_loc.path);
  freee (local);
  frame->local = NULL;

  STACK_UNWIND (frame, op_ret, op_errno, fd, inode, buf);

  return 0;
}

int32_t
ra_open (call_frame_t *frame,
         xlator_t *this,
         loc_t *loc,
         int32_t flags,
	 fd_t *fd)
{
  ra_local_t *local = calloc (1, sizeof (*local));

  local->file_loc.inode = loc->inode;
  local->file_loc.path = strdup (loc->path);

  local->flags = flags;

  frame->local = local;

  STACK_WIND (frame,
              ra_open_cbk,
              FIRST_CHILD(this),
              FIRST_CHILD(this)->fops->open,
              loc,
              flags,
	      fd);

  return 0;
}

int32_t
ra_create (call_frame_t *frame,
           xlator_t *this,
	   loc_t *loc,
           int32_t flags,
           mode_t mode,
	   fd_t *fd)
{
  ra_local_t *local = calloc (1, sizeof (*local));


  local->file_loc.inode = loc->inode;
  local->file_loc.path = strdup (loc->path);

  local->mode = mode;
  local->flags = 0;
  frame->local = local;

  STACK_WIND (frame,
              ra_create_cbk,
              FIRST_CHILD(this),
              FIRST_CHILD(this)->fops->create,
	      loc, flags, mode, fd);

  return 0;
}

/* freee cache pages between offset and offset+size,
   does not touch pages with frames waiting on it
*/

static void
flush_region (call_frame_t *frame,
              ra_file_t *file,
              off_t offset,
              off_t size)
{
  ra_page_t *trav;

  ra_file_lock (file);

  trav = file->pages.next;
  while (trav != &file->pages && trav->offset < (offset + size)) {
    ra_page_t *next = trav->next;
    if (trav->offset >= offset && !trav->waitq) {

      if (!trav->ready) {
	gf_log (frame->this->name, GF_LOG_DEBUG,
		"killing featus, file=%p, offset=%lld, de=%lld, a=%lld",
		file, trav->offset, offset, size);
      }
      ra_page_purge (trav);
    }
    trav = next;
  }

  ra_file_unlock (file);
}


int32_t
ra_close_cbk (call_frame_t *frame,
              void *cookie,
              xlator_t *this,
              int32_t op_ret,
              int32_t op_errno)
{
  frame->local = NULL;

  STACK_UNWIND (frame, op_ret, op_errno);
  return 0;
}

int32_t
ra_close (call_frame_t *frame,
          xlator_t *this,
          fd_t *fd)
{
  data_t *file_data = dict_get (fd->ctx, this->name);
  ra_file_t *file = NULL;

  if (file_data) {
    file = data_to_ptr (file_data);
    
    flush_region (frame, file, 0, file->pages.prev->offset+1);
    dict_del (fd->ctx, this->name);
    
    file->fd = NULL;
    ra_file_unref (file);
  }

  STACK_WIND (frame,
              ra_close_cbk,
              FIRST_CHILD(this),
              FIRST_CHILD(this)->fops->close,
              fd);
  return 0;
}


void
read_ahead (call_frame_t *frame,
            ra_file_t *file)
{
  off_t ra_offset;
  size_t ra_size;
  off_t trav_offset;
  ra_page_t *trav = NULL;
  off_t cap = file->size;

  if (!file->page_count)
    return;

  ra_size = file->page_size * file->page_count;
  ra_offset = floor (file->offset, file->page_size);
  cap = file->size ? file->size : file->offset + ra_size;

  while (ra_offset < min (file->offset + ra_size, cap)) {
    ra_file_lock (file);
    trav = ra_page_get (file, ra_offset);
    ra_file_unlock (file);
    if (!trav)
      break;
    ra_offset += file->page_size;
  }

  if (trav)
    /* comfortable enough */
    return;

  trav_offset = ra_offset;

  trav = file->pages.next;
  cap = file->size ? file->size : ra_offset + ra_size;
  while (trav_offset < min(ra_offset + ra_size, cap)) {
    char fault = 0;
    ra_file_lock (file);
    trav = ra_page_get (file, trav_offset);
    if (!trav) {
      fault = 1;
      trav = ra_page_create (file, trav_offset);
      trav->dirty = 1;
    }
    ra_file_unlock (file);

    if (fault) {
      gf_log (frame->this->name, GF_LOG_DEBUG,
	      "RA at offset=%"PRId64, trav_offset);
      ra_page_fault (file, frame, trav_offset);
    }
    trav_offset += file->page_size;
  }
  return ;
}

int32_t
ra_need_atime_cbk (call_frame_t *frame,
                   void *cookie,
                   xlator_t *this,
                   int32_t op_ret,
                   int32_t op_errno,
                   struct iovec *vector,
                   int32_t count,
		   struct stat *stbuf)
{
  STACK_DESTROY (frame->root);
  return 0;
}

static void
dispatch_requests (call_frame_t *frame,
                   ra_file_t *file)
{
  ra_local_t *local = frame->local;
  ra_conf_t *conf = file->conf;
  off_t rounded_offset;
  off_t rounded_end;
  off_t trav_offset;
  ra_page_t *trav;
  call_frame_t *ra_frame;
  char need_atime_update = 1;

  rounded_offset = floor (local->offset, file->page_size);
  rounded_end = roof (local->offset + local->size, file->page_size);

  trav_offset = rounded_offset;
  trav = file->pages.next;

  while (trav_offset < rounded_end) {
    char fault = 0;

    ra_file_lock (file);
    trav = ra_page_get (file, trav_offset);
    if (!trav) {
      trav = ra_page_create (file, trav_offset);
      fault = 1;
      need_atime_update = 0;
    } 

    if (trav->ready) {
      gf_log (frame->this->name, GF_LOG_DEBUG,
	      "HIT at offset=%"PRId64".",
	      trav_offset);
      ra_frame_fill (trav, frame);
    } else {
      gf_log (frame->this->name, GF_LOG_DEBUG,
	      "IN-TRANSIT at offset=%"PRId64".",
	      trav_offset);
      ra_wait_on_page (trav, frame);
      need_atime_update = 0;
    }
    ra_file_unlock (file);

    if (fault) {
      gf_log (frame->this->name, GF_LOG_DEBUG,
	      "MISS at offset=%"PRId64".",
	      trav_offset);
      ra_page_fault (file, frame, trav_offset);
    }

    trav_offset += file->page_size;
  }

  if (need_atime_update && conf->force_atime_update) {
    /* TODO: use untimens() since readv() can confuse underlying
       io-cache and others */
    ra_frame = copy_frame (frame);
    STACK_WIND (ra_frame, 
                ra_need_atime_cbk,
                FIRST_CHILD (frame->this), 
                FIRST_CHILD (frame->this)->fops->readv,
                file->fd, 1, 1);
  }

  return ;
}

int32_t
ra_readv_disabled_cbk (call_frame_t *frame, 
                       void *cookie,
                       xlator_t *this,
                       int32_t op_ret,
                       int32_t op_errno,
                       struct iovec *vector,
                       int32_t count,
		       struct stat *stbuf)
{
  GF_ERROR_IF_NULL (this);
  GF_ERROR_IF_NULL (vector);

  STACK_UNWIND (frame, op_ret, op_errno, vector, count, stbuf);
  return 0;
}

int32_t
ra_readv (call_frame_t *frame,