client-protocol.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/>.
*/

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

#include "glusterfs.h"
#include "client-protocol.h"
#include "compat.h"
#include "dict.h"
#include "protocol.h"
#include "transport.h"
#include "xlator.h"
#include "logging.h"
#include "timer.h"
#include "defaults.h"

#include <sys/resource.h>
#include <inttypes.h>

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

static int32_t client_protocol_interpret (transport_t *trans, gf_block_t *blk);
static int32_t client_protocol_cleanup (transport_t *trans);


typedef int32_t (*gf_op_t) (call_frame_t *frame,
			    dict_t *args);

static gf_op_t gf_fops[];
static gf_op_t gf_mops[];

/* 
 * lookup_frame - lookup call frame corresponding to a given callid
 * @trans: transport object
 * @callid: call id of the frame
 *
 * not for external reference
 */
static call_frame_t *
lookup_frame (transport_t *trans, int64_t callid)
{
  char buf[64];
  call_frame_t *frame = NULL;
  client_proto_priv_t *priv = NULL;

  if (!trans) {
    gf_log ("", GF_LOG_ERROR, "!trans");
    return NULL;
  }

  snprintf (buf, 64, "%"PRId64, callid);
  priv = trans->xl_private;

  pthread_mutex_lock (&priv->lock);
  {
    frame = data_to_bin (dict_get (priv->saved_frames, buf));
    dict_del (priv->saved_frames, buf);
  }
  pthread_mutex_unlock (&priv->lock);

  return frame;
}

/*
 * str_to_stat - convert a ASCII string to a struct stat
 * @buf: string
 *
 * not for external reference
 */
static struct stat *
str_to_stat (char *buf)
{
  struct stat *stbuf = calloc (1, sizeof (*stbuf));

  uint64_t dev;
  uint64_t ino;
  uint32_t mode;
  uint32_t nlink;
  uint32_t uid;
  uint32_t gid;
  uint64_t rdev;
  uint64_t size;
  uint32_t blksize;
  uint64_t blocks;
  uint32_t atime;
  uint32_t atime_nsec;
  uint32_t mtime;
  uint32_t mtime_nsec;
  uint32_t ctime;
  uint32_t ctime_nsec;

  sscanf (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);

  stbuf->st_dev = dev;
  stbuf->st_ino = ino;
  stbuf->st_mode = mode;
  stbuf->st_nlink = nlink;
  stbuf->st_uid = uid;
  stbuf->st_gid = gid;
  stbuf->st_rdev = rdev;
  stbuf->st_size = size;
  stbuf->st_blksize = blksize;
  stbuf->st_blocks = blocks;

  stbuf->st_atime = atime;
  stbuf->st_mtime = mtime;
  stbuf->st_ctime = ctime;

#if HAVE_TV_NSEC
  stbuf->st_atim.tv_nsec = atime_nsec;
  stbuf->st_mtim.tv_nsec = mtime_nsec;
  stbuf->st_ctim.tv_nsec = ctime_nsec;
#endif

  return stbuf;
}

static void
call_bail (void *trans)
{
  client_proto_priv_t *priv = NULL;
  struct timeval current;
  int32_t bail_out = 0;

  if (!trans) {
    gf_log ("", GF_LOG_ERROR, "!trans");
    return;
  }

  priv = ((transport_t *)trans)->xl_private;

  gettimeofday (&current, NULL);
  pthread_mutex_lock (&priv->lock);
  {
    /* Chaining to get call-always functionality from call-once timer */
    if (priv->timer) {
      struct timeval timeout = {0,};
      timeout.tv_sec = 10;
      timeout.tv_usec = 0;
      gf_timer_cbk_t timer_cbk = priv->timer->cbk;
      gf_timer_call_cancel (((transport_t *) trans)->xl->ctx, priv->timer);
      priv->timer = gf_timer_call_after (((transport_t *) trans)->xl->ctx,
					 timeout,
					 timer_cbk,
					 trans);
      if (!priv->timer) {
	gf_log (((transport_t *)trans)->xl->name, GF_LOG_DEBUG,
		"Cannot create timer");
      }
    }

    if ((priv->saved_frames->count > 0)
	&& (((unsigned long long)priv->last_recieved.tv_sec + priv->transport_timeout) < current.tv_sec)
	&& (((unsigned long long)priv->last_sent.tv_sec + priv->transport_timeout ) < current.tv_sec)) {
      struct tm last_sent_tm, last_received_tm;
      char last_sent[32], last_received[32];

      bail_out = 1;
      localtime_r (&priv->last_sent.tv_sec, &last_sent_tm);
      localtime_r (&priv->last_recieved.tv_sec, &last_received_tm);
      strftime (last_sent, 32, "%Y-%m-%d %H:%M:%S", &last_sent_tm);
      strftime (last_received, 32, "%Y-%m-%d %H:%M:%S", &last_received_tm);
      gf_log (((transport_t *)trans)->xl->name, GF_LOG_WARNING,
	      "activating bail-out. pending frames = %d. last sent = %s. last received = %s transport-timeout = %d", 
	      priv->saved_frames->count, last_sent, last_received, 
	      priv->transport_timeout);
    }
  }
  pthread_mutex_unlock (&priv->lock);

  if (bail_out) {
    gf_log (((transport_t *)trans)->xl->name, GF_LOG_CRITICAL,
	  "bailing transport");
    transport_bail (trans);
  }
}

/* 
 * client_protocol_xfer - client protocol transfer routine. called to send 
 *                        request packet to server
 * @frame: call frame
 * @this:
 * @type: operation type
 * @op: operation
 * @request: request data
 *
 * not for external reference
 */
int32_t
client_protocol_xfer (call_frame_t *frame,
		      xlator_t *this,
		      glusterfs_op_type_t type,
		      glusterfs_fop_t op,
		      dict_t *request)
{
  int32_t ret = -1;
  transport_t *trans;
  client_proto_priv_t *proto_priv;

  if (!request) {
    gf_log (this->name, GF_LOG_ERROR, "request is NULL");
    return -1;
  }

  trans = this->private;
  if (!trans) {
    gf_log (this->name, GF_LOG_ERROR, "this->private is NULL");
    return -1;
  }

  proto_priv = trans->xl_private;
  if (!proto_priv) {
    gf_log (this->name, GF_LOG_ERROR, "trans->xl_private is NULL");
    return -1;
  }

  dict_set (request, "CALLER_UID", data_from_uint64 (frame->root->uid));
  dict_set (request, "CALLER_GID", data_from_uint64 (frame->root->gid));
  dict_set (request, "CALLER_PID", data_from_uint64 (frame->root->pid));

  {
    int64_t callid;
    gf_block_t *blk;
    struct iovec *vector = NULL;
    int32_t count = 0;
    int32_t i;
    char connected = 0;
    char buf[64];

    pthread_mutex_lock (&proto_priv->lock);
    {
      callid = proto_priv->callid++;
      connected = proto_priv->connected;
      if (!connected) {
	/* tricky code - taking chances:
	   cause pipelining of handshake packet and this frame */
	connected = (transport_connect (trans) == 0);
	if (connected)
	  gf_log (this->name, GF_LOG_WARNING,
		  "attempting to pipeline request type(%d) op(%d) with handshake",
		  type, op);
      }
      if (connected) {
	snprintf (buf, 64, "%"PRId64, callid);
	frame->op = op;
	frame->type = type;
	dict_set (proto_priv->saved_frames, buf,
		  bin_to_data (frame, sizeof (frame)));
      }
    }
    pthread_mutex_unlock (&proto_priv->lock);

    blk = gf_block_new (callid);
    blk->type = type;
    blk->op = op;
    blk->size = 0;    // obselete
    blk->data = NULL; // obselete
    blk->dict = request;

    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);

    if (connected) {
      client_proto_priv_t *priv = ((transport_t *)this->private)->xl_private;

      ret = trans->ops->writev (trans, vector, count);

      pthread_mutex_lock (&(priv->lock));
      {
	gettimeofday (&(priv->last_sent), NULL);
      }
      pthread_mutex_unlock (&(priv->lock));
    }

    free (blk);

    if (ret != 0) {
      if (connected) {
	gf_log (this->name, GF_LOG_ERROR, "transport_submit failed");
      } else {
	dict_t *reply = get_new_dict ();

	reply->is_locked = 1;
	gf_log (this->name, GF_LOG_WARNING,
		"not connected at the moment to submit frame type(%d) op(%d)",
		type, op);
	frame->root->rsp_refs = dict_ref (reply);
	if (type == GF_OP_TYPE_FOP_REQUEST)
	  gf_fops[op] (frame, reply);
	else
	  gf_mops[op] (frame, reply);
	dict_unref (reply);
      }
      return -1;
    }
  }
  return ret;
}


/**
 * client_create - create function for client protocol
 * @frame: call frame
 * @this: this translator structure
 * @path: complete path to file
 * @flags: create flags
 * @mode: create mode
 *
 * external reference through client_protocol_xlator->fops->create
 */
 
int32_t 
client_create (call_frame_t *frame,
	       xlator_t *this,
	       loc_t *loc,
	       int32_t flags,
	       mode_t mode,
	       fd_t *fd)
{
  int32_t ret = -1;
  dict_t *request = NULL;
  client_local_t *local = NULL;

  local = calloc (1, sizeof (client_local_t));
  local->inode = loc->inode;
  local->fd = fd;
  
  frame->local = local;

  request = get_new_dict ();
  dict_set (request, "PATH", str_to_data ((char *)loc->path));
  dict_set (request, "FLAGS", data_from_int64 (flags));
  dict_set (request, "MODE", data_from_int64 (mode));

  ret = client_protocol_xfer (frame,
			      this,
			      GF_OP_TYPE_FOP_REQUEST,
			      GF_FOP_CREATE,
			      request);
  dict_destroy (request);
  return ret;
}

/**
 * client_open - open function for client protocol
 * @frame: call frame
 * @this: this translator structure
 * @loc: location of file
 * @flags: open flags
 * @mode: open modes
 *
 * external reference through client_protocol_xlator->fops->open
 */

int32_t 
client_open (call_frame_t *frame,
	     xlator_t *this,
	     loc_t *loc,
	     int32_t flags,
	     fd_t *fd)
{
  ino_t ino = 0;
  int32_t ret = -1;
  dict_t *request = NULL;
  data_t *ino_data = NULL;
  client_local_t *local = NULL;

  if (loc && loc->inode && loc->inode->ctx)
    ino_data = dict_get (loc->inode->ctx, this->name);
  
  if (ino_data) {
    ino = data_to_uint64 (ino_data);
  } else {
    TRAP_ON (ino_data == NULL);
    frame->root->rsp_refs = NULL;
    gf_log (this->name, GF_LOG_ERROR, "%s: returning EINVAL", loc->path);
    STACK_UNWIND (frame, -1, EINVAL, fd);
    return 0;
  }

  request = get_new_dict ();
  dict_set (request, "PATH", str_to_data ((char *)loc->path));
  dict_set (request, "INODE", data_from_uint64 (ino));
  dict_set (request, "FLAGS", data_from_int64 (flags));

  local = calloc (1, sizeof (client_local_t));
  local->inode = loc->inode;
  local->fd = fd;
  
  frame->local = local;
  
  ret = client_protocol_xfer (frame, this, GF_OP_TYPE_FOP_REQUEST,
			      GF_FOP_OPEN, request);

  dict_destroy (request);
  return ret;
}


/**
 * client_stat - stat function for client protocol
 * @frame: call frame
 * @this: this translator structure
 * @loc: location
 *
 * external reference through client_protocol_xlator->fops->stat
 */

int32_t 
client_stat (call_frame_t *frame,
	     xlator_t *this,
	     loc_t *loc)
{
  ino_t ino = 0;
  int32_t ret = -1;
  dict_t *request = NULL;
  data_t *ino_data = NULL;

  if (loc && loc->inode && loc->inode->ctx)
    ino_data = dict_get (loc->inode->ctx, this->name);

  if (ino_data) {
    ino = data_to_uint64 (ino_data);
  } else {
    gf_log (this->name, GF_LOG_ERROR, "%s: returning EINVAL", loc->path);
    TRAP_ON (ino_data == NULL);
    frame->root->rsp_refs = NULL;
    STACK_UNWIND (frame, -1, EINVAL, NULL);
    return 0;
  }
  request = get_new_dict ();
  dict_set (request, "PATH", str_to_data ((char *)loc->path));
  dict_set (request, "INODE", data_from_uint64 (ino));
  
  ret = client_protocol_xfer (frame, this, GF_OP_TYPE_FOP_REQUEST,
			      GF_FOP_STAT, request);

  dict_destroy (request);
  return ret;
}


/**
 * client_readlink - readlink function for client protocol
 * @frame: call frame
 * @this: this translator structure
 * @loc: location
 * @size: 
 *
 * external reference through client_protocol_xlator->fops->readlink
 */


int32_t 
client_readlink (call_frame_t *frame,
		 xlator_t *this,
		 loc_t *loc,
		 size_t size)
{
  ino_t ino = 0;
  int32_t ret = -1;
  dict_t *request = NULL;
  data_t *ino_data = NULL;

  if (loc && loc->inode && loc->inode->ctx)
    ino_data = dict_get (loc->inode->ctx, this->name);

  if (ino_data) {
    ino = data_to_uint64 (ino_data);
  } else {
    gf_log (this->name, GF_LOG_ERROR, "%s: returning EINVAL", loc->path);
    TRAP_ON (ino_data == NULL);
    frame->root->rsp_refs = NULL;
    STACK_UNWIND (frame, -1, EINVAL, NULL);
    return 0;
  }

  request = get_new_dict ();
  dict_set (request, "PATH", str_to_data ((char *)loc->path));
  dict_set (request, "INODE", data_from_uint64 (ino));
  dict_set (request, "LEN", data_from_int64 (size));

  ret = client_protocol_xfer (frame, this, GF_OP_TYPE_FOP_REQUEST,
			      GF_FOP_READLINK, request);

  dict_destroy (request);
  return ret;
}


/**
 * client_mknod - mknod function for client protocol
 * @frame: call frame
 * @this: this translator structure
 * @path: pathname of node
 * @mode: 
 * @dev:
 *
 * external reference through client_protocol_xlator->fops->mknod
 */

int32_t 
client_mknod (call_frame_t *frame,
	      xlator_t *this,
	      loc_t *loc,