Move rpc to top-level

Signed-off-by: Vijay Bellur <vijay@gluster.com>
Signed-off-by: Anand V. Avati <avati@dev.gluster.com>

BUG: 875 (Implement a new protocol to provide proper backward/forward compatibility)
URL: http://bugs.gluster.com/cgi-bin/bugzilla3/show_bug.cgi?id=875

1 files changed, 1301 insertions, 0 deletions

diff --git a/rpc/rpc-lib/src/rpc-transport.c b/rpc/rpc-lib/src/rpc-transport.c
new file mode 100644
index 00000000000..82ea9a74bfd
--- /dev/null
+++ b/rpc/rpc-lib/src/rpc-transport.c
@@ -0,0 +1,1301 @@
+/*
+  Copyright (c) 2010 Gluster, Inc. <http://www.gluster.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/>.
+*/
+
+#include <dlfcn.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <sys/poll.h>
+#include <fnmatch.h>
+#include <stdint.h>
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "logging.h"
+#include "rpc-transport.h"
+#include "glusterfs.h"
+/* FIXME: xlator.h is needed for volume_option_t, need to define the datatype
+ * in some other header
+ */
+#include "xlator.h"
+#include "list.h"
+
+#ifndef GF_OPTION_LIST_EMPTY
+#define GF_OPTION_LIST_EMPTY(_opt) (_opt->value[0] == NULL)
+#endif
+
+/* RFC 1123 & 952 */
+static char
+valid_host_name (char *address, int length)
+{
+        int i = 0;
+        char ret = 1;
+
+        if ((length > 75) || (length == 1)) {
+                ret = 0;
+                goto out;
+        }
+
+        if (!isalnum (address[length - 1])) {
+                ret = 0;
+                goto out;
+        }
+
+        for (i = 0; i < length; i++) {
+                if (!isalnum (address[i]) && (address[i] != '.')
+                    && (address[i] != '-')) {
+                        ret = 0;
+                        goto out;
+                }
+        }
+
+out:
+        return ret;
+}
+
+static char
+valid_ipv4_address (char *address, int length)
+{
+        int octets = 0;
+        int value = 0;
+        char *tmp = NULL, *ptr = NULL, *prev = NULL, *endptr = NULL;
+        char ret = 1;
+
+        prev = tmp = gf_strdup (address);
+        prev = strtok_r (tmp, ".", &ptr);
+
+        while (prev != NULL)
+        {
+                octets++;
+                value = strtol (prev, &endptr, 10);
+                if ((value > 255) || (value < 0) || (endptr != NULL)) {
+                        ret = 0;
+                        goto out;
+                }
+
+                prev = strtok_r (NULL, ".", &ptr);
+        }
+
+        if (octets != 4) {
+                ret = 0;
+        }
+
+out:
+        GF_FREE (tmp);
+        return ret;
+}
+
+
+static char
+valid_ipv6_address (char *address, int length)
+{
+        int hex_numbers = 0;
+        int value = 0;
+        char *tmp = NULL, *ptr = NULL, *prev = NULL, *endptr = NULL;
+        char ret = 1;
+
+        tmp = gf_strdup (address);
+        prev = strtok_r (tmp, ":", &ptr);
+
+        while (prev != NULL)
+        {
+                hex_numbers++;
+                value = strtol (prev, &endptr, 16);
+                if ((value > 0xffff) || (value < 0)
+                    || (endptr != NULL && *endptr != '\0')) {
+                        ret = 0;
+                        goto out;
+                }
+
+                prev = strtok_r (NULL, ":", &ptr);
+        }
+
+        if (hex_numbers > 8) {
+                ret = 0;
+        }
+
+out:
+        GF_FREE (tmp);
+        return ret;
+}
+
+
+static char
+valid_internet_address (char *address)
+{
+        char ret = 0;
+        int length = 0;
+
+        if (address == NULL) {
+                goto out;
+        }
+
+        length = strlen (address);
+        if (length == 0) {
+                goto out;
+        }
+
+        if (valid_ipv4_address (address, length)
+            || valid_ipv6_address (address, length)
+            || valid_host_name (address, length)) {
+                ret = 1;
+        }
+
+out:
+        return ret;
+}
+
+
+int
+__volume_option_value_validate (char *name,
+                                data_pair_t *pair,
+                                volume_option_t *opt)
+{
+	int       i = 0;
+	int       ret = -1;
+	uint64_t  input_size = 0;
+	long long inputll = 0;
+
+	/* Key is valid, validate the option */
+	switch (opt->type) {
+        case GF_OPTION_TYPE_XLATOR:
+                break;
+
+	case GF_OPTION_TYPE_PATH:
+	{
+                if (strstr (pair->value->data, "../")) {
+                        gf_log (name, GF_LOG_ERROR,
+                                "invalid path given '%s'",
+                                pair->value->data);
+                        ret = -1;
+                        goto out;
+                }
+
+                /* Make sure the given path is valid */
+		if (pair->value->data[0] != '/') {
+			gf_log (name, GF_LOG_WARNING,
+				"option %s %s: '%s' is not an "
+				"absolute path name",
+				pair->key, pair->value->data,
+				pair->value->data);
+		}
+		ret = 0;
+	}
+	break;
+	case GF_OPTION_TYPE_INT:
+	{
+		/* Check the range */
+		if (gf_string2longlong (pair->value->data,
+					&inputll) != 0) {
+			gf_log (name, GF_LOG_ERROR,
+				"invalid number format \"%s\" in "
+				"\"option %s\"",
+				pair->value->data, pair->key);
+			goto out;
+		}
+
+		if ((opt->min == 0) && (opt->max == 0)) {
+			gf_log (name, GF_LOG_DEBUG,
+				"no range check required for "
+				"'option %s %s'",
+				pair->key, pair->value->data);
+			ret = 0;
+			break;
+		}
+		if ((inputll < opt->min) ||
+		    (inputll > opt->max)) {
+			gf_log (name, GF_LOG_WARNING,
+				"'%lld' in 'option %s %s' is out of "
+				"range [%"PRId64" - %"PRId64"]",
+				inputll, pair->key,
+				pair->value->data,
+				opt->min, opt->max);
+		}
+		ret = 0;
+	}
+	break;
+	case GF_OPTION_TYPE_SIZET:
+	{
+		/* Check the range */
+		if (gf_string2bytesize (pair->value->data,
+					&input_size) != 0) {
+			gf_log (name, GF_LOG_ERROR,
+				"invalid size format \"%s\" in "
+				"\"option %s\"",
+				pair->value->data, pair->key);
+			goto out;
+		}
+
+		if ((opt->min == 0) && (opt->max == 0)) {
+			gf_log (name, GF_LOG_DEBUG,
+				"no range check required for "
+				"'option %s %s'",
+				pair->key, pair->value->data);
+			ret = 0;
+			break;
+		}
+		if ((input_size < opt->min) ||
+		    (input_size > opt->max)) {
+			gf_log (name, GF_LOG_ERROR,
+				"'%"PRId64"' in 'option %s %s' is "
+				"out of range [%"PRId64" - %"PRId64"]",
+				input_size, pair->key,
+				pair->value->data,
+				opt->min, opt->max);
+		}
+		ret = 0;
+	}
+	break;
+	case GF_OPTION_TYPE_BOOL:
+	{
+		/* Check if the value is one of
+		   '0|1|on|off|no|yes|true|false|enable|disable' */
+		gf_boolean_t bool_value;
+		if (gf_string2boolean (pair->value->data,
+				       &bool_value) != 0) {
+			gf_log (name, GF_LOG_ERROR,
+				"option %s %s: '%s' is not a valid "
+				"boolean value",
+				pair->key, pair->value->data,
+				pair->value->data);
+			goto out;
+		}
+		ret = 0;
+	}
+	break;
+	case GF_OPTION_TYPE_STR:
+	{
+		/* Check if the '*str' is valid */
+                if (GF_OPTION_LIST_EMPTY(opt)) {
+                        ret = 0;
+                        goto out;
+                }
+
+		for (i = 0; (i < ZR_OPTION_MAX_ARRAY_SIZE) &&
+			     opt->value[i]; i++) {
+			if (strcasecmp (opt->value[i],
+					pair->value->data) == 0) {
+				ret = 0;
+				break;
+			}
+		}
+
+		if ((i == ZR_OPTION_MAX_ARRAY_SIZE)
+		    || ((i < ZR_OPTION_MAX_ARRAY_SIZE)
+			&& (!opt->value[i]))) {
+			/* enter here only if
+			 * 1. reached end of opt->value array and haven't
+                         *    validated input
+			 *                      OR
+			 * 2. valid input list is less than
+                         *    ZR_OPTION_MAX_ARRAY_SIZE and input has not
+                         *    matched all possible input values.
+			 */
+			char given_array[4096] = {0,};
+			for (i = 0; (i < ZR_OPTION_MAX_ARRAY_SIZE) &&
+				     opt->value[i]; i++) {
+				strcat (given_array, opt->value[i]);
+				strcat (given_array, ", ");
+			}
+
+			gf_log (name, GF_LOG_ERROR,
+				"option %s %s: '%s' is not valid "
+				"(possible options are %s)",
+				pair->key, pair->value->data,
+				pair->value->data, given_array);
+
+			goto out;
+		}
+	}
+	break;
+	case GF_OPTION_TYPE_PERCENT:
+	{
+		uint32_t percent = 0;
+
+
+		/* Check if the value is valid percentage */
+		if (gf_string2percent (pair->value->data,
+				       &percent) != 0) {
+			gf_log (name, GF_LOG_ERROR,
+				"invalid percent format \"%s\" "
+				"in \"option %s\"",
+				pair->value->data, pair->key);
+			goto out;
+		}
+
+		if ((percent < 0) || (percent > 100)) {
+			gf_log (name, GF_LOG_ERROR,
+				"'%d' in 'option %s %s' is out of "
+				"range [0 - 100]",
+				percent, pair->key,
+				pair->value->data);
+		}
+		ret = 0;
+	}
+	break;
+	case GF_OPTION_TYPE_PERCENT_OR_SIZET:
+	{
+		uint32_t percent = 0;
+		uint64_t input_size = 0;
+
+		/* Check if the value is valid percentage */
+		if (gf_string2percent (pair->value->data,
+				       &percent) == 0) {
+			if (percent > 100) {
+				gf_log (name, GF_LOG_DEBUG,
+					"value given was greater than 100, "
+					"assuming this is actually a size");
+		        if (gf_string2bytesize (pair->value->data,
+				                &input_size) == 0) {
+				        /* Check the range */
+				if ((opt->min == 0) &&
+                                            (opt->max == 0)) {
+				        gf_log (name, GF_LOG_DEBUG,
+				        "no range check "
+                                                        "required for "
+					"'option %s %s'",
+				pair->key,
+                                                        pair->value->data);
+						// It is a size
+			                        ret = 0;
+				                goto out;
+				}
+			if ((input_size < opt->min) ||
+				            (input_size > opt->max)) {
+				        gf_log (name, GF_LOG_ERROR,
+				        "'%"PRId64"' in "
+                                                        "'option %s %s' is out"
+					" of range [%"PRId64""
+                                                        "- %"PRId64"]",
+				input_size, pair->key,
+				pair->value->data,
+				                opt->min, opt->max);
+				}
+					// It is a size
+					ret = 0;
+					goto out;
+				} else {
+					// It's not a percent or size
+					gf_log (name, GF_LOG_ERROR,
+					"invalid number format \"%s\" "
+					"in \"option %s\"",
+					pair->value->data, pair->key);
+				}
+
+			}
+			// It is a percent
+			ret = 0;
+			goto out;
+		} else {
+		        if (gf_string2bytesize (pair->value->data,
+				        &input_size) == 0) {
+			        /* Check the range */
+			if ((opt->min == 0) && (opt->max == 0)) {
+			        gf_log (name, GF_LOG_DEBUG,
+				        "no range check required for "
+					"'option %s %s'",
+			pair->key, pair->value->data);
+					// It is a size
+		                        ret = 0;
+				        goto out;
+			}
+		        if ((input_size < opt->min) ||
+			            (input_size > opt->max)) {
+					gf_log (name, GF_LOG_ERROR,
+				        "'%"PRId64"' in 'option %s %s'"
+                                                " is out of range [%"PRId64" -"
+                                                " %"PRId64"]",
+			input_size, pair->key,
+			pair->value->data,
+			                opt->min, opt->max);
+				}
+			} else {
+				// It's not a percent or size
+				gf_log (name, GF_LOG_ERROR,
+					"invalid number format \"%s\" "
+					"in \"option %s\"",
+					pair->value->data, pair->key);
+			}
+			//It is a size
+                        ret = 0;
+		        goto out;
+		}
+
+	}
+	break;
+	case GF_OPTION_TYPE_TIME:
+	{
+		uint32_t input_time = 0;
+
+		/* Check if the value is valid percentage */
+		if (gf_string2time (pair->value->data,
+				    &input_time) != 0) {
+			gf_log (name,
+				GF_LOG_ERROR,
+				"invalid time format \"%s\" in "
+				"\"option %s\"",
+				pair->value->data, pair->key);
+			goto out;
+		}
+
+		if ((opt->min == 0) && (opt->max == 0)) {
+			gf_log (name, GF_LOG_DEBUG,
+				"no range check required for "
+				"'option %s %s'",
+				pair->key, pair->value->data);
+			ret = 0;
+			goto out;
+		}
+		if ((input_time < opt->min) ||
+		    (input_time > opt->max)) {
+			gf_log (name, GF_LOG_ERROR,
+				"'%"PRIu32"' in 'option %s %s' is "
+				"out of range [%"PRId64" - %"PRId64"]",
+				input_time, pair->key,
+				pair->value->data,
+				opt->min, opt->max);
+		}
+		ret = 0;
+	}
+	break;
+	case GF_OPTION_TYPE_DOUBLE:
+	{
+		double input_time = 0.0;
+
+		/* Check if the value is valid double */
+		if (gf_string2double (pair->value->data,
+				      &input_time) != 0) {
+			gf_log (name,
+				GF_LOG_ERROR,
+				"invalid time format \"%s\" in \"option %s\"",
+				pair->value->data, pair->key);
+			goto out;
+		}
+
+		if (input_time < 0.0) {
+			gf_log (name,
+				GF_LOG_ERROR,
+				"invalid time format \"%s\" in \"option %s\"",
+				pair->value->data, pair->key);
+			goto out;
+		}
+
+		if ((opt->min == 0) && (opt->max == 0)) {
+			gf_log (name, GF_LOG_DEBUG,
+				"no range check required for 'option %s %s'",
+				pair->key, pair->value->data);
+			ret = 0;
+			goto out;
+		}
+		ret = 0;
+	}
+	break;
+        case GF_OPTION_TYPE_INTERNET_ADDRESS:
+        {
+                if (valid_internet_address (pair->value->data)) {
+                        ret = 0;
+                }
+	}
+        break;
+	case GF_OPTION_TYPE_ANY:
+		/* NO CHECK */
+		ret = 0;
+		break;
+	}
+
+out:
+	return ret;
+}
+
+/* FIXME: this procedure should be removed from transport */
+int
+validate_volume_options (char *name, dict_t *options, volume_option_t *opt)
+{
+	int i = 0;
+	int ret = -1;
+	int index = 0;
+	volume_option_t *trav  = NULL;
+	data_pair_t     *pairs = NULL;
+
+	if (!opt) {
+		ret = 0;
+		goto out;
+	}
+
+	/* First search for not supported options, if any report error */
+	pairs = options->members_list;
+	while (pairs) {
+		ret = -1;
+		for (index = 0;
+		     opt[index].key && opt[index].key[0] ; index++) {
+			trav = &(opt[index]);
+			for (i = 0 ;
+			     (i < ZR_VOLUME_MAX_NUM_KEY) &&
+				     trav->key[i]; i++) {
+				/* Check if the key is valid */
+				if (fnmatch (trav->key[i],
+					     pairs->key, FNM_NOESCAPE) == 0) {
+					ret = 0;
+					break;
+				}
+			}
+			if (!ret) {
+				if (i) {
+					gf_log (name, GF_LOG_WARNING,
+						"option '%s' is deprecated, "
+						"preferred is '%s', continuing"
+						" with correction",
+						trav->key[i], trav->key[0]);
+					/* TODO: some bytes lost */
+					pairs->key = gf_strdup (trav->key[0]);
+				}
+				break;
+			}
+		}
+		if (!ret) {
+			ret = __volume_option_value_validate (name, pairs, trav);
+			if (-1 == ret) {
+				goto out;
+			}
+		}
+
+		pairs = pairs->next;
+	}
+
+	ret = 0;
+ out:
+	return ret;
+}
+
+int32_t
+rpc_transport_get_myaddr (rpc_transport_t *this, char *peeraddr, int addrlen,
+                          struct sockaddr *sa, size_t salen)
+{
+        if (!this)
+                return -1;
+
+        return this->ops->get_myaddr (this, peeraddr, addrlen, sa, salen);
+}
+
+int32_t
+rpc_transport_get_myname (rpc_transport_t *this, char *hostname, int hostlen)
+{
+        if (!this)
+                return -1;
+
+        return this->ops->get_myname (this, hostname, hostlen);
+}
+
+int32_t
+rpc_transport_get_peername (rpc_transport_t *this, char *hostname, int hostlen)
+{
+        if (!this)
+                return -1;
+        return this->ops->get_peername (this, hostname, hostlen);
+}
+
+int32_t
+rpc_transport_get_peeraddr (rpc_transport_t *this, char *peeraddr, int addrlen,
+                            struct sockaddr *sa, size_t salen)
+{
+        if (!this)
+                return -1;
+        return this->ops->get_peeraddr (this, peeraddr, addrlen, sa, salen);
+}
+
+void
+rpc_transport_pollin_destroy (rpc_transport_pollin_t *pollin)
+{
+        if (!pollin) {
+                goto out;
+        }
+
+        if (pollin->vectored) {
+                if (pollin->data.vector.iobuf1) {
+                        iobuf_unref (pollin->data.vector.iobuf1);
+                }
+
+                if (pollin->data.vector.iobuf2) {
+                        iobuf_unref (pollin->data.vector.iobuf2);
+                }
+        } else {
+                if (pollin->data.simple.iobuf) {
+                        iobuf_unref (pollin->data.simple.iobuf);
+                }
+        }
+
+        if (pollin->private) {
+                /* */
+                GF_FREE (pollin->private);
+        }
+
+        GF_FREE (pollin);
+out:
+        return;
+}
+
+
+rpc_transport_pollin_t *
+rpc_transport_pollin_alloc (rpc_transport_t *this, struct iobuf *iobuf,
+                            size_t size, struct iobuf *vectored_buf,
+                            size_t vectored_size, void *private)
+{
+        rpc_transport_pollin_t *msg = NULL;
+        msg = GF_CALLOC (1, sizeof (*msg), gf_common_mt_rpc_trans_pollin_t);
+        if (!msg) {
+                gf_log ("rpc-transport", GF_LOG_ERROR, "out of memory");
+                goto out;
+        }
+
+        if (vectored_buf) {
+                msg->vectored = 1;
+                msg->data.vector.iobuf1 = iobuf_ref (iobuf);
+                msg->data.vector.size1  = size;
+
+                msg->data.vector.iobuf2 = iobuf_ref (vectored_buf);
+                msg->data.vector.size2  = vectored_size;
+        } else {
+                msg->data.simple.iobuf = iobuf_ref (iobuf);
+                msg->data.simple.size = size;
+        }
+
+        msg->private = private;
+out:
+        return msg;
+}
+
+
+rpc_transport_pollin_t *
+rpc_transport_same_process_pollin_alloc (rpc_transport_t *this,
+                                         struct iovec *rpchdr, int rpchdrcount,
+                                         struct iovec *proghdr,
+                                         int proghdrcount,
+                                         struct iovec *progpayload,
+                                         int progpayloadcount,
+                                         rpc_transport_rsp_t *rsp,
+                                         char is_request)
+{
+        rpc_transport_pollin_t *msg            = NULL;
+        int                     rpchdrlen      = 0, proghdrlen = 0;
+        int                     progpayloadlen = 0;
+        char                    vectored       = 0;
+        char                   *hdr            = NULL, *progpayloadbuf = NULL;
+
+        if (!rpchdr || !proghdr) {
+                goto err;
+        }
+
+        msg = GF_CALLOC (1, sizeof (*msg), gf_common_mt_rpc_trans_pollin_t);
+        if (!msg) {
+                gf_log ("rpc-transport", GF_LOG_ERROR, "out of memory");
+                goto err;
+        }
+
+        rpchdrlen = iov_length (rpchdr, rpchdrcount);
+        proghdrlen = iov_length (proghdr, proghdrcount);
+
+        if (progpayload) {
+                vectored = 1;
+                progpayloadlen = iov_length (progpayload, progpayloadcount);
+        }
+
+        /* FIXME: we are assuming rpchdr and proghdr will fit into
+         * an iobuf (128KB)
+         */
+        if ((rpchdrlen + proghdrlen) > this->ctx->page_size) {
+                gf_log ("rpc_transport", GF_LOG_DEBUG, "program hdr and rpc"
+                        " hdr together combined (%d) is bigger than "
+                        "iobuf size (%zu)", (rpchdrlen + proghdrlen),
+                        this->ctx->page_size);
+                goto err;
+        }
+
+        if (vectored) {
+                msg->data.vector.iobuf1 = iobuf_get (this->ctx->iobuf_pool);
+                if (!msg->data.vector.iobuf1) {
+                        gf_log ("rpc_transport", GF_LOG_ERROR,
+                                "out of memory");
+                        goto err;
+                }
+
+                msg->data.vector.size1 = rpchdrlen + proghdrlen;
+                hdr = iobuf_ptr (msg->data.vector.iobuf1);
+
+                if (!is_request && rsp) {
+                        msg->data.vector.iobuf2 = rsp->rspbuf;
+                        progpayloadbuf = rsp->rspvec->iov_base;
+                } else {
+                        msg->data.vector.iobuf2 = iobuf_get (this->ctx->iobuf_pool);
+                        if (!msg->data.vector.iobuf2) {
+                                gf_log ("rpc_transport", GF_LOG_ERROR,
+                                        "out of memory");
+                                goto err;
+                        }
+
+                        progpayloadbuf = iobuf_ptr (msg->data.vector.iobuf2);
+                }
+                msg->data.vector.size2 = progpayloadlen;
+        } else {
+                if (!is_request && rsp) {
+                        /* FIXME: Assuming rspvec contains only one vector */
+                        hdr = rsp->rspvec->iov_base;
+                        msg->data.simple.iobuf = rsp->rspbuf;
+                } else {
+                        msg->data.simple.iobuf = iobuf_get (this->ctx->iobuf_pool);
+                        if (!msg->data.simple.iobuf) {
+                                gf_log ("rpc_transport", GF_LOG_ERROR,
+                                        "out of memory");
+                                goto err;
+                        }
+
+                        hdr = iobuf_ptr (msg->data.simple.iobuf);
+                }
+
+                msg->data.simple.size = rpchdrlen + proghdrlen;
+        }
+
+        iov_unload (hdr, rpchdr, rpchdrcount);
+        hdr += rpchdrlen;
+        iov_unload (hdr, proghdr, proghdrcount);
+
+        if (progpayload) {
+                iov_unload (progpayloadbuf, progpayload,
+                            progpayloadcount);
+        }
+
+        if (is_request) {
+                msg->private = rsp;
+        }
+        return msg;
+err:
+        if (msg) {
+                rpc_transport_pollin_destroy (msg);
+        }
+
+        return NULL;
+}
+
+
+rpc_transport_handover_t *
+rpc_transport_handover_alloc (rpc_transport_pollin_t *pollin)
+{
+        rpc_transport_handover_t *msg = NULL;
+
+        msg = GF_CALLOC (1, sizeof (*msg), gf_common_mt_rpc_trans_handover_t);
+        if (!msg) {
+                gf_log ("rpc_transport", GF_LOG_ERROR, "out of memory");
+                goto out;
+        }
+
+        msg->pollin = pollin;
+        INIT_LIST_HEAD (&msg->list);
+out:
+        return msg;
+}
+
+
+void
+rpc_transport_handover_destroy (rpc_transport_handover_t *msg)
+{
+        if (!msg) {
+                goto out;
+        }
+
+        if (msg->pollin) {
+                rpc_transport_pollin_destroy (msg->pollin);
+        }
+
+        GF_FREE (msg);
+
+out:
+        return;
+}
+
+
+rpc_transport_t *
+rpc_transport_load (glusterfs_ctx_t *ctx, dict_t *options, char *trans_name)
+{
+	struct rpc_transport *trans = NULL, *return_trans = NULL;
+	char *name = NULL;
+	void *handle = NULL;
+	char *type = NULL;
+	char str[] = "ERROR";
+	int32_t ret = -1;
+	int8_t is_tcp = 0, is_unix = 0, is_ibsdp = 0;
+	volume_opt_list_t *vol_opt = NULL;
+
+	GF_VALIDATE_OR_GOTO("rpc-transport", options, fail);
+	GF_VALIDATE_OR_GOTO("rpc-transport", ctx, fail);
+	GF_VALIDATE_OR_GOTO("rpc-transport", trans_name, fail);
+
+	trans = GF_CALLOC (1, sizeof (struct rpc_transport), gf_common_mt_rpc_trans_t);
+	GF_VALIDATE_OR_GOTO("rpc-transport", trans, fail);
+
+        trans->name = gf_strdup (trans_name);
+        GF_VALIDATE_OR_GOTO ("rpc-transport", trans->name, fail);
+
+	trans->ctx = ctx;
+	type = str;
+
+	/* Backward compatibility */
+	ret = dict_get_str (options, "transport-type", &type);
+	if (ret < 0) {
+		ret = dict_set_str (options, "transport-type", "socket");
+		if (ret < 0)
+			gf_log ("dict", GF_LOG_DEBUG,
+				"setting transport-type failed");
+		gf_log ("rpc-transport", GF_LOG_WARNING,
+			"missing 'option transport-type'. defaulting to "
+			"\"socket\"");
+	} else {
+		{
+			/* Backword compatibility to handle * /client,
+			 * * /server.
+			 */
+			char *tmp = strchr (type, '/');
+			if (tmp)
+				*tmp = '\0';
+		}
+
+		is_tcp = strcmp (type, "tcp");
+		is_unix = strcmp (type, "unix");
+		is_ibsdp = strcmp (type, "ib-sdp");
+		if ((is_tcp == 0) ||
+		    (is_unix == 0) ||
+		    (is_ibsdp == 0)) {
+			if (is_unix == 0)
+				ret = dict_set_str (options,
+						    "transport.address-family",
+						    "unix");
+			if (is_ibsdp == 0)
+				ret = dict_set_str (options,
+						    "transport.address-family",
+						    "inet-sdp");
+
+			if (ret < 0)
+				gf_log ("dict", GF_LOG_DEBUG,
+					"setting address-family failed");
+
+			ret = dict_set_str (options,
+					    "transport-type", "socket");
+			if (ret < 0)
+				gf_log ("dict", GF_LOG_DEBUG,
+					"setting transport-type failed");
+		}
+	}
+
+	ret = dict_get_str (options, "transport-type", &type);
+	if (ret < 0) {
+		gf_log ("rpc-transport", GF_LOG_ERROR,
+			"'option transport-type <xx>' missing in volume '%s'",
+			trans_name);
+		goto fail;
+	}
+
+	ret = gf_asprintf (&name, "%s/%s.so", RPC_TRANSPORTDIR, type);
+        if (-1 == ret) {
+                gf_log ("rpc-transport", GF_LOG_ERROR, "asprintf failed");
+                goto fail;
+        }
+	gf_log ("rpc-transport", GF_LOG_DEBUG,
+		"attempt to load file %s", name);
+
+	handle = dlopen (name, RTLD_NOW|RTLD_GLOBAL);
+	if (handle == NULL) {
+		gf_log ("rpc-transport", GF_LOG_ERROR, "%s", dlerror ());
+		gf_log ("rpc-transport", GF_LOG_ERROR,
+			"volume '%s': transport-type '%s' is not valid or "
+			"not found on this machine",
+			trans_name, type);
+		goto fail;
+	}
+
+	trans->ops = dlsym (handle, "tops");
+	if (trans->ops == NULL) {
+		gf_log ("rpc-transport", GF_LOG_ERROR,
+			"dlsym (rpc_transport_ops) on %s", dlerror ());
+		goto fail;
+	}
+
+	trans->init = dlsym (handle, "init");
+	if (trans->init == NULL) {
+		gf_log ("rpc-transport", GF_LOG_ERROR,
+			"dlsym (gf_rpc_transport_init) on %s", dlerror ());
+		goto fail;
+	}
+
+	trans->fini = dlsym (handle, "fini");
+	if (trans->fini == NULL) {
+		gf_log ("rpc-transport", GF_LOG_ERROR,
+			"dlsym (gf_rpc_transport_fini) on %s", dlerror ());
+		goto fail;
+	}
+
+	vol_opt = GF_CALLOC (1, sizeof (volume_opt_list_t),
+                             gf_common_mt_volume_opt_list_t);
+        if (!vol_opt) {
+                gf_log (trans_name, GF_LOG_ERROR, "out of memory");
+                goto fail;
+        }
+
+	vol_opt->given_opt = dlsym (handle, "options");
+	if (vol_opt->given_opt == NULL) {
+		gf_log ("rpc-transport", GF_LOG_DEBUG,
+			"volume option validation not specified");
+	} else {
+                /* FIXME: is adding really needed? */
+		/* list_add_tail (&vol_opt->list, &xl->volume_options); */
+		if (-1 ==
+		    validate_volume_options (trans_name, options,
+                                             vol_opt->given_opt)) {
+			gf_log ("rpc-transport", GF_LOG_ERROR,
+				"volume option validation failed");
+			goto fail;
+		}
+	}
+
+	ret = trans->init (trans);
+	if (ret != 0) {
+		gf_log ("rpc-transport", GF_LOG_ERROR,
+			"'%s' initialization failed", type);
+		goto fail;
+	}
+
+        trans->options = options;
+
+	pthread_mutex_init (&trans->lock, NULL);
+	return_trans = trans;
+	return return_trans;
+
+fail:
+        if (trans) {
+                if (trans->name) {
+                        GF_FREE (trans->name);
+                }
+
+                GF_FREE (trans);
+        }
+
+        if (name) {
+                GF_FREE (name);
+        }
+
+        if (vol_opt) {
+                GF_FREE (vol_opt);
+        }
+
+        return NULL;
+}
+
+
+int32_t
+rpc_transport_submit_request (rpc_transport_t *this, rpc_transport_req_t *req)
+{
+	int32_t                       ret          = -1;
+        rpc_transport_t              *peer_trans   = NULL;
+        rpc_transport_pollin_t       *pollin       = NULL;
+        rpc_transport_handover_t     *handover_msg = NULL;
+        rpc_transport_rsp_t          *rsp          = NULL;
+
+        if (this->peer_trans) {
+                peer_trans = this->peer_trans;
+
+                rsp = GF_CALLOC (1, sizeof (*rsp), gf_common_mt_rpc_trans_rsp_t);
+                if (!rsp) {
+                        ret = -ENOMEM;
+                        goto fail;
+                }
+
+                *rsp = req->rsp;
+
+                pollin = rpc_transport_same_process_pollin_alloc (this, req->msg.rpchdr,
+                                                                  req->msg.rpchdrcount,
+                                                                  req->msg.proghdr,
+                                                                  req->msg.proghdrcount,
+                                                                  req->msg.progpayload,
+                                                                  req->msg.progpayloadcount,
+                                                                  rsp, 1);
+                if (!pollin) {
+                        GF_FREE (rsp);
+                        ret = -ENOMEM;
+                        goto fail;
+                }
+
+                handover_msg = rpc_transport_handover_alloc (pollin);
+                if (!handover_msg) {
+                        rpc_transport_pollin_destroy (pollin);
+                        ret = -ENOMEM;
+                        goto fail;
+                }
+
+                pthread_mutex_lock (&peer_trans->handover.mutex);
+                {
+                        list_add_tail (&handover_msg->list,
+                                       &peer_trans->handover.msgs);
+                        pthread_cond_broadcast (&peer_trans->handover.cond);
+                }
+                pthread_mutex_unlock (&peer_trans->handover.mutex);
+
+                return 0;
+        }
+
+	GF_VALIDATE_OR_GOTO("rpc_transport", this, fail);
+	GF_VALIDATE_OR_GOTO("rpc_transport", this->ops, fail);
+
+	ret = this->ops->submit_request (this, req);
+fail:
+	return ret;
+}
+
+
+int32_t
+rpc_transport_submit_reply (rpc_transport_t *this, rpc_transport_reply_t *reply)
+{
+	int32_t                   ret          = -1;
+        rpc_transport_t              *peer_trans   = NULL;
+        rpc_transport_pollin_t       *pollin       = NULL;
+        rpc_transport_handover_t     *handover_msg = NULL;
+
+        if (this->peer_trans) {
+                peer_trans = this->peer_trans;
+
+                pollin = rpc_transport_same_process_pollin_alloc (this, reply->msg.rpchdr,
+                                                                  reply->msg.rpchdrcount,
+                                                                  reply->msg.proghdr,
+                                                                  reply->msg.proghdrcount,
+                                                                  reply->msg.progpayload,
+                                                                  reply->msg.progpayloadcount,
+                                                                  reply->private, 0);
+                if (!pollin) {
+                        return -ENOMEM;
+                }
+
+                handover_msg = rpc_transport_handover_alloc (pollin);
+                if (!handover_msg) {
+                        rpc_transport_pollin_destroy (pollin);
+                        return -ENOMEM;
+                }
+
+                pthread_mutex_lock (&peer_trans->handover.mutex);
+                {
+                        list_add_tail (&handover_msg->list,
+                                       &peer_trans->handover.msgs);
+                        pthread_cond_broadcast (&peer_trans->handover.cond);
+                }
+                pthread_mutex_unlock (&peer_trans->handover.mutex);
+
+                return 0;
+        }
+
+	GF_VALIDATE_OR_GOTO("rpc_transport", this, fail);
+	GF_VALIDATE_OR_GOTO("rpc_transport", this->ops, fail);
+
+	ret = this->ops->submit_reply (this, reply);
+fail:
+	return ret;
+}
+
+
+int32_t
+rpc_transport_connect (rpc_transport_t *this)
+{
+	int ret = -1;
+
+	GF_VALIDATE_OR_GOTO("rpc_transport", this, fail);
+
+	ret = this->ops->connect (this);
+fail:
+	return ret;
+}
+
+
+int32_t
+rpc_transport_listen (rpc_transport_t *this)
+{
+	int ret = -1;
+
+	GF_VALIDATE_OR_GOTO("rpc_transport", this, fail);
+
+	ret = this->ops->listen (this);
+fail:
+	return ret;
+}
+
+
+int32_t
+rpc_transport_disconnect (rpc_transport_t *this)
+{
+	int32_t ret = -1;
+
+	GF_VALIDATE_OR_GOTO("rpc_transport", this, fail);
+
+	ret = this->ops->disconnect (this);
+fail:
+	return ret;
+}
+
+
+int32_t
+rpc_transport_destroy (rpc_transport_t *this)
+{
+	int32_t ret = -1;
+
+	GF_VALIDATE_OR_GOTO("rpc_transport", this, fail);
+
+	if (this->fini)
+		this->fini (this);
+	pthread_mutex_destroy (&this->lock);
+	GF_FREE (this);
+fail:
+	return ret;
+}
+
+
+rpc_transport_t *
+rpc_transport_ref (rpc_transport_t *this)
+{
+	rpc_transport_t *return_this = NULL;
+
+	GF_VALIDATE_OR_GOTO("rpc_transport", this, fail);
+
+	pthread_mutex_lock (&this->lock);
+	{
+		this->refcount ++;
+	}
+	pthread_mutex_unlock (&this->lock);
+
+	return_this = this;
+fail:
+	return return_this;
+}
+
+
+int32_t
+rpc_transport_unref (rpc_transport_t *this)
+{
+	int32_t refcount = 0;
+	int32_t ret = -1;
+
+	GF_VALIDATE_OR_GOTO("rpc_transport", this, fail);
+
+	pthread_mutex_lock (&this->lock);
+	{
+		refcount = --this->refcount;
+	}
+	pthread_mutex_unlock (&this->lock);
+
+	if (refcount == 0) {
+		/* xlator_notify (this->xl, GF_EVENT_RPC_TRANSPORT_CLEANUP,
+                   this); */
+		rpc_transport_destroy (this);
+	}
+
+	ret = 0;
+fail:
+	return ret;
+}
+
+
+int32_t
+rpc_transport_notify (rpc_transport_t *this, rpc_transport_event_t event,
+                      void *data, ...)
+{
+        int32_t ret = -1;
+
+        if (this == NULL) {
+                goto out;
+        }
+
+        //ret = this->notify (this, this->notify_data, event, data);
+        ret = this->notify (this, this->mydata, event, data);
+out:
+        return ret;
+}
+
+
+void *
+rpc_transport_peerproc (void *trans_data)
+{
+        rpc_transport_t                     *trans = NULL;
+        rpc_transport_handover_t            *msg   = NULL;
+
+        trans = trans_data;
+
+        while (1) {
+                pthread_mutex_lock (&trans->handover.mutex);
+                {
+                        while (list_empty (&trans->handover.msgs))
+                                pthread_cond_wait (&trans->handover.cond,
+                                                   &trans->handover.mutex);
+
+                        msg = list_entry (trans->handover.msgs.next,
+                                          rpc_transport_handover_t, list);
+
+                        list_del_init (&msg->list);
+                }
+                pthread_mutex_unlock (&trans->handover.mutex);
+
+                rpc_transport_notify (trans, RPC_TRANSPORT_MSG_RECEIVED, msg->pollin);
+                rpc_transport_handover_destroy (msg);
+        }
+}
+
+
+int
+rpc_transport_setpeer (rpc_transport_t *trans, rpc_transport_t *peer_trans)
+{
+        trans->peer_trans = rpc_transport_ref (peer_trans);
+
+        INIT_LIST_HEAD (&trans->handover.msgs);
+        pthread_cond_init (&trans->handover.cond, NULL);
+        pthread_mutex_init (&trans->handover.mutex, NULL);
+        pthread_create (&trans->handover.thread, NULL,
+                        rpc_transport_peerproc, trans);
+
+        peer_trans->peer_trans = rpc_transport_ref (trans);
+
+        INIT_LIST_HEAD (&peer_trans->handover.msgs);
+        pthread_cond_init (&peer_trans->handover.cond, NULL);
+        pthread_mutex_init (&peer_trans->handover.mutex, NULL);
+        pthread_create (&peer_trans->handover.thread, NULL,
+                        rpc_transport_peerproc, peer_trans);
+
+        return 0;
+}
+
+
+inline int
+rpc_transport_register_notify (rpc_transport_t *trans,
+                               rpc_transport_notify_t notify, void *mydata)
+{
+        int ret = -1;
+
+        if (trans == NULL) {
+                goto out;
+        }
+
+        trans->notify = notify;
+        trans->mydata = mydata;
+
+        ret = 0;
+out:
+        return ret;
+}