server-protocol.c

分布式文件系统

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

   GlusterFS is freee 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: whenever inode_search() fails, we need to do dummy_inode() before diverting to lookup()s
 */

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

#include "transport.h"
#include "fnmatch.h"
#include "xlator.h"
#include "protocol.h"
#include "lock.h"
#include "server-protocol.h"
#include <time.h>
#include <sys/uio.h>
#include "call-stub.h"
#include "defaults.h"
#include "list.h"
#include "dict.h"

#include <sys/resource.h>

#if __WORDSIZE == 64
# define F_L64 "%l"
#else
# define F_L64 "%ll"
#endif

#define STATE(frame) ((server_state_t *)frame->root->state)
#define TRANSPORT_OF(frame) ((transport_t *) STATE (frame)->trans)
#define SERVER_PRIV(frame) ((server_proto_priv_t *) TRANSPORT_OF(frame)->xl_private)
#define BOUND_XL(frame) ((xlator_t *) STATE (frame)->bound_xl)

/*
 * str_to_ptr - convert a string to pointer
 * @string: string
 *
 */
static int32_t
server_inode_prune (xlator_t *bound_xl);

static inode_t *
dummy_inode (inode_table_t *table)
{
  inode_t *dummy;

  dummy = calloc (1, sizeof (*dummy));

  dummy->table = table;

  INIT_LIST_HEAD (&dummy->list);
  INIT_LIST_HEAD (&dummy->inode_hash);
  INIT_LIST_HEAD (&dummy->fds);
  INIT_LIST_HEAD (&dummy->dentry.name_hash);
  INIT_LIST_HEAD (&dummy->dentry.inode_list);

  dummy->ref = 1;
  dummy->ctx = get_new_dict ();

  LOCK_INIT (&dummy->lock);
  return dummy;
}

/* 
 * stat_to_str - convert struct stat to a ASCII string
 * @stbuf: struct stat pointer
 *
 * not for external reference
 */
static char *
stat_to_str (struct stat *stbuf)
{
  char *tmp_buf = NULL;

  uint64_t dev = stbuf->st_dev;
  uint64_t ino = stbuf->st_ino;
  uint32_t mode = stbuf->st_mode;
  uint32_t nlink = stbuf->st_nlink;
  uint32_t uid = stbuf->st_uid;
  uint32_t gid = stbuf->st_gid;
  uint64_t rdev = stbuf->st_rdev;
  uint64_t size = stbuf->st_size;
  uint32_t blksize = stbuf->st_blksize;
  uint64_t blocks = stbuf->st_blocks;
  uint32_t atime = stbuf->st_atime;
  uint32_t mtime = stbuf->st_mtime;
  uint32_t ctime = stbuf->st_ctime;

#ifdef HAVE_TV_NSEC
  uint32_t atime_nsec = stbuf->st_atim.tv_nsec;
  uint32_t mtime_nsec = stbuf->st_mtim.tv_nsec;
  uint32_t ctime_nsec = stbuf->st_ctim.tv_nsec;
#else
  uint32_t atime_nsec = 0;
  uint32_t mtime_nsec = 0;
  uint32_t ctime_nsec = 0;
#endif

  asprintf (&tmp_buf,
	    GF_STAT_PRINT_FMT_STR,
	    dev,
	    ino,
	    mode,
	    nlink,
	    uid,
	    gid,
	    rdev,
	    size,
	    blksize,
	    blocks,
	    atime,
	    atime_nsec,
	    mtime,
	    mtime_nsec,
	    ctime,
	    ctime_nsec);
  
  return tmp_buf;
}

/*
 * generic_reply - generic reply, used to send reply packet to client
 * @frame: call frame
 * @type: reply type GF_MOP_REPLY/GF_FOP_REPLY
 * @op: operation to which this reply corresponds to
 * @params: parameter dictionary, actual data of the reply packet
 *
 * not for external reference
 */
static int32_t
generic_reply (call_frame_t *frame,
	       int32_t type,
	       glusterfs_fop_t op,
	       dict_t *params)
{
  gf_block_t *blk;
  transport_t *trans;
  int32_t count, i, ret;
  struct iovec *vector;

  trans = TRANSPORT_OF (frame);
  
  blk = gf_block_new (frame->root->unique);
  blk->data = NULL;
  blk->size = 0;
  blk->type = type;
  blk->op   = op;
  blk->dict = params;

  count = gf_block_iovec_len (blk);
  vector = alloca (count * sizeof (*vector));
  memset (vector, 0, count * sizeof (*vector));

  gf_block_to_iovec (blk, vector, count);
  for (i=0; i<count; i++)
    if (!vector[i].iov_base)
      vector[i].iov_base = alloca (vector[i].iov_len);
  gf_block_to_iovec (blk, vector, count);

  freee (blk);

  ret = trans->ops->writev (trans, vector, count);
  if (ret != 0) {
    gf_log (trans->xl->name, GF_LOG_ERROR,
	    "transport_writev failed");
    transport_except (trans);
  }

  return 0;
}

server_reply_t *
server_reply_dequeue (server_reply_queue_t *queue)
{
  server_reply_t *entry = NULL;

  pthread_mutex_lock (&queue->lock);
  {
    while (list_empty (&queue->list))
      pthread_cond_wait (&queue->cond, &queue->lock);
    
    entry = list_entry (queue->list.next, server_reply_t, list);
    list_del_init (&entry->list);
  }
  pthread_mutex_unlock (&queue->lock);

  return entry;
}


static void
server_reply_queue (server_reply_t *entry,
		    server_reply_queue_t *queue)
{
  pthread_mutex_lock (&queue->lock);
  {
    list_add_tail (&entry->list, &queue->list);
    pthread_cond_broadcast (&queue->cond);
  }
  pthread_mutex_unlock (&queue->lock);
}


static void *
server_reply_proc (void *data)
{
  server_reply_queue_t *queue = data;

  while (1) {
    server_reply_t *entry = NULL;
    server_state_t *state = NULL;
    xlator_t *bound_xl = NULL;

    entry = server_reply_dequeue (queue);

    bound_xl = BOUND_XL (entry->frame);

    generic_reply (entry->frame, entry->type, entry->op, entry->reply);

    server_inode_prune (bound_xl);

    state = STATE (entry->frame);
    {
      if (entry->refs)
	dict_unref (entry->refs);
      dict_destroy (entry->reply);
      STACK_DESTROY (entry->frame->root);
      freee (entry);
    }
    {
      transport_unref (state->trans);
      if (state->inode)
	inode_unref (state->inode);
      if (state->inode2)
	inode_unref (state->inode2);
      freee (state);
    }
  }

  return NULL;
}

static void
server_reply (call_frame_t *frame,
	      int32_t type,
	      glusterfs_fop_t op,
	      dict_t *reply,
	      dict_t *refs)
{
  server_reply_t *entry = NULL;
  transport_t *trans = ((server_private_t *)frame->this->private)->trans;
  server_conf_t *conf = NULL;
  entry = calloc (1, sizeof (*entry));
  entry->frame = frame;
  entry->type = type;
  entry->op = op;
  entry->reply = reply;
  if (refs) {
    switch (entry->op) {
    case GF_FOP_READ:
      entry->refs = dict_ref (refs);
      break;
    }
  }
  conf = trans->xl_private;

#if 1
  /* TODO: This part is removed as it is observed that, with the queuing
   * method, there is a memory leak. Need to investigate further. Till then 
   * this code will be part of #if 0 */
  server_reply_queue (entry, conf->queue);
#else
  server_state_t *state = NULL;
  xlator_t *bound_xl = NULL;
  bound_xl = BOUND_XL (entry->frame);
  
  generic_reply (entry->frame, entry->type, entry->op, entry->reply);
  
  server_inode_prune (bound_xl);
  
  state = STATE (entry->frame);
  {
    if (entry->refs)
      dict_unref (entry->refs);
    dict_destroy (entry->reply);
    STACK_DESTROY (entry->frame->root);
    freee (entry);
  }
  {
    transport_unref (state->trans);
    if (state->inode)
      inode_unref (state->inode);
    if (state->inode2)
      inode_unref (state->inode2);
    freee (state);
  }
#endif
}

/*
 * server_fchmod_cbk
 */
int32_t
server_fchmod_cbk (call_frame_t *frame,
		   void *cookie,
		   xlator_t *this,
		   int32_t op_ret,
		   int32_t op_errno,
		   struct stat *stbuf)
{
  dict_t *reply = get_new_dict ();
  char *stat_str = NULL;

  dict_set (reply, "RET", data_from_uint64 (op_ret));
  dict_set (reply, "ERRNO", data_from_uint64 (op_errno));

  stat_str = stat_to_str (stbuf);
  dict_set (reply, "STAT", data_from_dynstr (stat_str));

  server_reply (frame, GF_OP_TYPE_FOP_REPLY, GF_FOP_FCHMOD, reply,
		frame->root->rsp_refs);

  return 0;
}

/*
 * server_fchmod
 *
 */
int32_t
server_fchmod (call_frame_t *frame,
	       xlator_t *bound_xl,
	       dict_t *params)
{  
  data_t *fd_data = dict_get (params, "FD");
  data_t *mode_data = dict_get (params, "MODE");
  fd_t *fd = NULL;
  mode_t mode = 0;
  server_proto_priv_t *priv = SERVER_PRIV (frame);
  int32_t fd_no = -1;
 
  if (fd_data) {
    fd_no = data_to_int32 (fd_data);
    fd = gf_fd_fdptr_get (priv->fdtable, fd_no);
    if (!fd)
      gf_log (frame->this->name, GF_LOG_ERROR,
	      "unresolved fd %d", fd_no);
  }

  if (!fd || !mode_data) {
    struct stat stbuf = {0,};
    gf_log (frame->this->name, GF_LOG_ERROR, 
	    "not getting enough data, returning EINVAL");
    server_fchmod_cbk (frame, NULL, frame->this, -1, EINVAL, &stbuf);
    return 0;
  }

  mode = data_to_uint64 (mode_data);

  STACK_WIND (frame, 
	      server_fchmod_cbk, 
	      bound_xl,
	      bound_xl->fops->fchmod,
	      fd,
	      mode);
  
  return 0;

}


/*
 * server_fchown_cbk 
 */
int32_t
server_fchown_cbk (call_frame_t *frame,
		   void *cookie,
		   xlator_t *this,
		   int32_t op_ret,
		   int32_t op_errno,
		   struct stat *stbuf)
{
  dict_t *reply = get_new_dict ();
  char *stat_str = NULL;
  dict_set (reply, "RET", data_from_uint64 (op_ret));
  dict_set (reply, "ERRNO", data_from_uint64 (op_errno));

  if (op_ret >= 0) {
    stat_str = stat_to_str (stbuf);
    dict_set (reply, "STAT", data_from_dynstr (stat_str));
  }

  server_reply (frame, GF_OP_TYPE_FOP_REPLY, GF_FOP_FCHOWN,
		reply, frame->root->rsp_refs);
  return 0;

}

/*
 * server_fchown
 *
 */
int32_t
server_fchown (call_frame_t *frame,
	       xlator_t *bound_xl,
	       dict_t *params)
{
  data_t *fd_data = dict_get (params, "FD");
  data_t *uid_data = dict_get (params, "UID");
  data_t *gid_data = dict_get (params, "GID");
  uid_t uid = 0;
  gid_t gid = 0;
  fd_t *fd = NULL;
  server_proto_priv_t *priv = SERVER_PRIV (frame);
  int32_t fd_no = -1;
 
  if (fd_data) {
    fd_no = data_to_int32 (fd_data);
    fd = gf_fd_fdptr_get (priv->fdtable, fd_no);
    if (!fd)
      gf_log (frame->this->name, GF_LOG_ERROR,
	      "unresolved fd %d", fd_no);
  }

  if (!fd || !uid_data || !gid_data) {
    struct stat stbuf = {0,};
    gf_log (frame->this->name, GF_LOG_ERROR, 
	    "not getting enough data, returning EINVAL");
    server_fchown_cbk (frame,
		       NULL,
		       frame->this,
		       -1,
		       EINVAL,
		       &stbuf);
    return 0;
  }

  uid = data_to_uint64 (uid_data);
  gid = data_to_uint64 (gid_data);
  STACK_WIND (frame, 
	      server_fchown_cbk, 
	      bound_xl,
	      bound_xl->fops->fchown,
	      fd,
	      uid,
	      gid);
  
  return 0;

}

/*
 * server_writedir_cbk - writedir callback for server protocol
 * @frame: call frame
 * @cookie: 
 * @this:
 * @op_ret: return value
 * @op_errno: errno
 *
 * not for external reference
 */
int32_t
server_setdents_cbk (call_frame_t *frame,
		     void *cookie,
		     xlator_t *this,
		     int32_t op_ret,
		     int32_t op_errno)
{
  dict_t *reply = get_new_dict ();

  dict_set (reply, "RET", data_from_int32 (op_ret));
  dict_set (reply, "ERRNO", data_from_int32 (op_errno));
  
  server_reply (frame, GF_OP_TYPE_FOP_REPLY, GF_FOP_SETDENTS,
		reply, frame->root->rsp_refs);

  return 0;
}

/*
 * server_lk_cbk - lk callback for server protocol
 * @frame: call frame
 * @cookie:
 * @this:
 * @op_ret:
 * @op_errno:
 * @lock:
 *
 * not for external reference
 */
int32_t
server_lk_cbk (call_frame_t *frame,
	       void *cookie,
	       xlator_t *this,
	       int32_t op_ret,
	       int32_t op_errno,
	       struct flock *lock)
{
  dict_t *reply = get_new_dict ();
  
  dict_set (reply, "RET", data_from_int32 (op_ret));
  dict_set (reply, "ERRNO", data_from_int32 (op_errno));
  
  if (op_ret >= 0) {
    dict_set (reply, "TYPE", data_from_int16 (lock->l_type));
    dict_set (reply, "WHENCE", data_from_int16 (lock->l_whence));
    dict_set (reply, "START", data_from_int64 (lock->l_start));
    dict_set (reply, "LEN", data_from_int64 (lock->l_len));
    dict_set (reply, "PID", data_from_uint64 (lock->l_pid));
  }

  server_reply (frame, GF_OP_TYPE_FOP_REPLY, GF_FOP_LK,
		reply, frame->root->rsp_refs);

  return 0;
}

/*
 * server_access_cbk - access callback for server protocol
 * @frame: call frame
 * @cookie:
 * @this:
 * @op_ret:
 * @op_errno:
 *
 * not for external reference
 */
int32_t
server_access_cbk (call_frame_t *frame,
		   void *cookie,
		   xlator_t *this,
		   int32_t op_ret,
		   int32_t op_errno)
{
  dict_t *reply = get_new_dict ();
  
  dict_set (reply, "RET", data_from_int32 (op_ret));
  dict_set (reply, "ERRNO", data_from_int32 (op_errno));

  server_reply (frame, GF_OP_TYPE_FOP_REPLY, GF_FOP_ACCESS,
		reply, frame->root->rsp_refs);

  return 0;
}

/*
 * server_utimens_cbk - utimens callback for server protocol
 * @frame: call frame
 * @cookie: 
 * @this:
 * @op_ret:
 * @op_errno:
 * @stbuf:
 *
 * not for external reference
 */
int32_t
server_utimens_cbk (call_frame_t *frame,
		    void *cookie,
		    xlator_t *this,
		    int32_t op_ret,
		    int32_t op_errno,
		    struct stat *stbuf)
{
  dict_t *reply = get_new_dict ();
  char *stat_buf = NULL;

  dict_set (reply, "RET", data_from_int32 (op_ret));
  dict_set (reply, "ERRNO", data_from_int32 (op_errno));