andy/glusterfs
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

cluster/afr: [Feature] Command implementation to get heal-count

Browse Source 
Currently to know the number of files to be healed, either user
has to go to backend and check the number of entries present in
indices/xattrop directory. But if a volume consists of large
number of bricks, going to each backend and counting the number
of entries is a time-taking task. Otherwise user can give
gluster volume heal vol-name info command but with this
approach if no. of entries are very hugh in the indices/
xattrop directory, it will comsume time.

So as a feature, new command is implemented.

Command 1: gluster volume heal vn statistics heal-count
This command will get the number of entries present in
every brick of a volume. The output displays only entries
count.

Command 2: gluster volume heal vn statistics heal-count
           replica 192.168.122.1:/home/user/brickname

           Here if we are concerned with just one replica.
So providing any one of the brick of a replica will get
the number of entries to be healed for that replica only.

Example:
Replicate volume with replica count 2.

Backend status:
--------------
[root@dhcp-0-17 xattrop]# ls -lia | wc -l
1918

NOTE: Out of 1918, 2 entries are <xattrop-gfid> dummy
entries so actual no. of entries to be healed are
1916.

[root@dhcp-0-17 xattrop]# pwd
/home/user/2ty/.glusterfs/indices/xattrop

Command output:
--------------
Gathering count of entries to be healed on volume volume3 has been successful

Brick 192.168.122.1:/home/user/22iu
Status: Brick is Not connected
Entries count is not available

Brick 192.168.122.1:/home/user/2ty
Number of entries: 1916

Change-Id: I72452f3de50502dc898076ec74d434d9e77fd290
BUG: 1015990
Signed-off-by: Venkatesh Somyajulu <vsomyaju@redhat.com>
Reviewed-on: http://review.gluster.org/6044
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Anand Avati <avati@redhat.com>
This commit is contained in:
Venkatesh Somyajulu 2013-10-07 13:47:47 +05:30 committed by Anand Avati
parent 047882750e
commit 75caba6371
15 changed files with 1009 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

157
cli/src/cli-cmd-parser.c
View File

@ -2482,12 +2482,103 @@ out:
return ret;
}
static int
extract_hostname_path_from_token (const char *tmp_words, char **hostname,
char **path)
{
int ret = 0;
char *delimiter = NULL;
char *tmp_host = NULL;
char *host_name = NULL;
char *words = NULL;
*hostname = NULL;
*path = NULL;
words = GF_CALLOC (1, strlen (tmp_words) + 1, gf_common_mt_char);
if (!words){
ret = -1;
goto out;
}
strncpy (words, tmp_words, strlen (tmp_words) + 1);
if (validate_brick_name (words)) {
cli_err ("Wrong brick type: %s, use <HOSTNAME>:"
"<export-dir-abs-path>", words);
ret = -1;
goto out;
} else {
delimiter = strrchr (words, ':');
ret = gf_canonicalize_path (delimiter + 1);
if (ret) {
goto out;
} else {
*path = GF_CALLOC (1, strlen (delimiter+1) +1,
gf_common_mt_char);
if (!*path) {
ret = -1;
goto out;
}
strncpy (*path, delimiter +1,
strlen(delimiter + 1) + 1);
}
}
tmp_host = gf_strdup (words);
if (!tmp_host) {
gf_log ("cli", GF_LOG_ERROR, "Out of memory");
ret = -1;
goto out;
}
get_host_name (tmp_host, &host_name);
if (!host_name) {
ret = -1;
gf_log("cli",GF_LOG_ERROR, "Unable to allocate "
"memory");
goto out;
}
if (!(strcmp (host_name, "localhost") &&
strcmp (host_name, "127.0.0.1") &&
strncmp (host_name, "0.", 2))) {
cli_err ("Please provide a valid hostname/ip other "
"than localhost, 127.0.0.1 or loopback "
"address (0.0.0.0 to 0.255.255.255).");
ret = -1;
goto out;
}
if (!valid_internet_address (host_name, _gf_false)) {
cli_err ("internet address '%s' does not conform to "
"standards", host_name);
ret = -1;
goto out;
}
*hostname = GF_CALLOC (1, strlen (host_name) + 1,
gf_common_mt_char);
if (!*hostname) {
ret = -1;
goto out;
}
strncpy (*hostname, host_name, strlen (host_name) + 1);
ret = 0;
out:
GF_FREE (words);
GF_FREE (tmp_host);
return ret;
}
int
cli_cmd_volume_heal_options_parse (const char **words, int wordcount,
dict_t **options)
{
int ret = 0;
dict_t *dict = NULL;
char *hostname = NULL;
char *path = NULL;
dict = dict_new ();
if (!dict)
@ -2524,28 +2615,66 @@ cli_cmd_volume_heal_options_parse (const char **words, int wordcount,
}
}
if (wordcount == 5) {
if (strcmp (words[3], "info")) {
if (strcmp (words[3], "info") &&
strcmp (words[3], "statistics")) {
ret = -1;
goto out;
}
if (!strcmp (words[4], "healed")) {
ret = dict_set_int32 (dict, "heal-op",
GF_AFR_OP_HEALED_FILES);
goto done;
if (!strcmp (words[3], "info")) {
if (!strcmp (words[4], "healed")) {
ret = dict_set_int32 (dict, "heal-op",
GF_AFR_OP_HEALED_FILES);
goto done;
}
if (!strcmp (words[4], "heal-failed")) {
ret = dict_set_int32 (dict, "heal-op",
GF_AFR_OP_HEAL_FAILED_FILES);
goto done;
}
if (!strcmp (words[4], "split-brain")) {
ret = dict_set_int32 (dict, "heal-op",
GF_AFR_OP_SPLIT_BRAIN_FILES);
goto done;
}
}
if (!strcmp (words[4], "heal-failed")) {
ret = dict_set_int32 (dict, "heal-op",
GF_AFR_OP_HEAL_FAILED_FILES);
goto done;
}
if (!strcmp (words[4], "split-brain")) {
ret = dict_set_int32 (dict, "heal-op",
GF_AFR_OP_SPLIT_BRAIN_FILES);
goto done;
if (!strcmp (words[3], "statistics")) {
if (!strcmp (words[4], "heal-count")) {
ret = dict_set_int32 (dict, "heal-op",
GF_AFR_OP_STATISTICS_HEAL_COUNT);
goto done;
}
}
ret = -1;
goto out;
}
if (wordcount == 7) {
if (!strcmp (words[3], "statistics")
&& !strcmp (words[4], "heal-count")
&& !strcmp (words[5], "replica")) {
ret = dict_set_int32 (dict, "heal-op",
GF_AFR_OP_STATISTICS_HEAL_COUNT_PER_REPLICA);
if (ret)
goto out;
ret = extract_hostname_path_from_token (words[6],
&hostname, &path);
if (ret)
goto out;
ret = dict_set_dynstr (dict, "per-replica-cmd-hostname",
hostname);
if (ret)
goto out;
ret = dict_set_dynstr (dict, "per-replica-cmd-path",
path);
if (ret)
goto out;
else
goto done;
}
}
ret = -1;
goto out;
done:

2
cli/src/cli-cmd-volume.c
View File

@ -1918,7 +1918,7 @@ struct cli_cmd volume_cmds[] = {
cli_cmd_volume_status_cbk,
"display status of all or specified volume(s)/brick"},
{ "volume heal <VOLNAME> [{full | statistics |info {healed | heal-failed | split-brain}}]",
{ "volume heal <VOLNAME> [{full | statistics {heal-count {replica <hostname:brickname>}} |info {healed | heal-failed | split-brain}}]",
cli_cmd_volume_heal_cbk,
"self-heal commands on volume specified by <VOLNAME>"},

59
cli/src/cli-rpc-ops.c
View File

@ -7109,6 +7109,53 @@ out:
return;
}
void
cmd_heal_volume_statistics_heal_count_out (dict_t *dict, int brick)
{
uint64_t num_entries = 0;
int ret = 0;
char key[256] = {0};
char *hostname = NULL;
char *path = NULL;
char *status = NULL;
char *shd_status = NULL;
snprintf (key, sizeof key, "%d-hostname", brick);
ret = dict_get_str (dict, key, &hostname);
if (ret)
goto out;
snprintf (key, sizeof key, "%d-path", brick);
ret = dict_get_str (dict, key, &path);
if (ret)
goto out;
cli_out ("\nBrick %s:%s", hostname, path);
snprintf (key, sizeof key, "%d-status", brick);
ret = dict_get_str (dict, key, &status);
if (status && strlen (status))
cli_out ("Status: %s", status);
snprintf (key, sizeof key, "%d-shd-status",brick);
ret = dict_get_str (dict, key, &shd_status);
if(!shd_status)
{
snprintf (key, sizeof key, "%d-hardlinks", brick);
ret = dict_get_uint64 (dict, key, &num_entries);
if (ret)
cli_out ("No gathered input for this brick");
else
cli_out ("Number of entries: %"PRIu64, num_entries);
}
out:
return;
}
int
gf_cli_heal_volume_cbk (struct rpc_req *req, struct iovec *iov,
int count, void *myframe)
@ -7190,6 +7237,12 @@ gf_cli_heal_volume_cbk (struct rpc_req *req, struct iovec *iov,
case GF_AFR_OP_STATISTICS:
heal_op_str = "crawl statistics";
break;
case GF_AFR_OP_STATISTICS_HEAL_COUNT:
heal_op_str = "count of entries to be healed";
break;
case GF_AFR_OP_STATISTICS_HEAL_COUNT_PER_REPLICA:
heal_op_str = "count of entries to be healed per replica";
break;
case GF_AFR_OP_INVALID:
heal_op_str = "invalid heal op";
break;
@ -7255,6 +7308,12 @@ gf_cli_heal_volume_cbk (struct rpc_req *req, struct iovec *iov,
for (i = 0; i < brick_count; i++)
cmd_heal_volume_statistics_out (dict, i);
break;
case GF_AFR_OP_STATISTICS_HEAL_COUNT:
case GF_AFR_OP_STATISTICS_HEAL_COUNT_PER_REPLICA:
for (i = 0; i < brick_count; i++)
cmd_heal_volume_statistics_heal_count_out (dict,
i);
break;
case GF_AFR_OP_INDEX_SUMMARY:
case GF_AFR_OP_HEALED_FILES:
case GF_AFR_OP_HEAL_FAILED_FILES:

1
libglusterfs/src/glusterfs.h
View File

@ -119,6 +119,7 @@
/* Index xlator related */
#define GF_XATTROP_INDEX_GFID "glusterfs.xattrop_index_gfid"
#define GF_BASE_INDICES_HOLDER_GFID "glusterfs.base_indicies_holder_gfid"
#define GF_GFIDLESS_LOOKUP "gfidless-lookup"
/* replace-brick and pump related internal xattrs */

2
rpc/rpc-lib/src/protocol-common.h
View File

@ -211,6 +211,8 @@ typedef enum {
GF_AFR_OP_HEAL_FAILED_FILES,
GF_AFR_OP_SPLIT_BRAIN_FILES,
GF_AFR_OP_STATISTICS,
GF_AFR_OP_STATISTICS_HEAL_COUNT,
GF_AFR_OP_STATISTICS_HEAL_COUNT_PER_REPLICA,
} gf_xl_afr_op_t ;
typedef enum {

81
tests/bugs/bug-1015990-rep.t Executable file
View File

@ -0,0 +1,81 @@
#!/bin/bash
. $(dirname $0)/../include.rc
. $(dirname $0)/../volume.rc
. $(dirname $0)/../afr.rc
cleanup;
## Start and create a volume
TEST glusterd;
TEST pidof glusterd;
TEST $CLI volume info;
TEST $CLI volume create $V0 replica 2 $H0:$B0/${V0}{1,2,3,4};
## Verify volume is is created
EXPECT "$V0" volinfo_field $V0 'Volume Name';
EXPECT 'Created' volinfo_field $V0 'Status';
## Start volume and verify
TEST $CLI volume start $V0;
EXPECT 'Started' volinfo_field $V0 'Status';
TEST glusterfs --volfile-id=/$V0 --volfile-server=$H0 $M0 --attribute-timeout=0 --entry-timeout=0
TEST kill_brick $V0 $H0 $B0/$V0"1"
sleep 5
TEST kill_brick $V0 $H0 $B0/$V0"3"
sleep 5
for i in {1..100}; do echo "STRING" > $M0/File$i; done
brick_2_sh_entries=$(count_sh_entries $B0/$V0"2")
brick_4_sh_entries=$(count_sh_entries $B0/$V0"4")
command_output=$(gluster volume heal $V0 statistics heal-count replica $H0:$B0/$V0"1")
substring="Number of entries:"
count=0
while read -r line;
do
if [[ "$line" == *$substring* ]]
then
value=$(echo $line | cut -f 2 -d :)
count=$(($count + $value))
fi
done <<< "$command_output"
brick_2_entries_count=$(($count-$value))
EXPECT "0" echo $brick_2_entries_count
brick_2_entries_count=$count
xattrop_count_brick_2=$(count_sh_entries $B0/$V0"2")
##Remove the count of the xattrop-gfid entry count as it does not contribute
##to the number of files to be healed
sub_val=1
xattrop_count_brick_2=$(($xattrop_count_brick_2-$sub_val))
ret=0
if [ "$xattrop_count_brick_2" -eq "$brick_2_entries_count" ]
then
ret=$(($ret + $sub_val))
fi
EXPECT "1" echo $ret
## Finish up
TEST $CLI volume stop $V0;
EXPECT 'Stopped' volinfo_field $V0 'Status';
TEST $CLI volume delete $V0;
TEST ! $CLI volume info $V0
cleanup;

95
tests/bugs/bug-1015990.t Executable file
View File

@ -0,0 +1,95 @@
#!/bin/bash
. $(dirname $0)/../include.rc
. $(dirname $0)/../volume.rc
. $(dirname $0)/../afr.rc
cleanup;
## Start and create a volume
TEST glusterd;
TEST pidof glusterd;
TEST $CLI volume info;
TEST $CLI volume create $V0 replica 2 $H0:$B0/${V0}{1,2,3,4};
## Verify volume is is created
EXPECT "$V0" volinfo_field $V0 'Volume Name';
EXPECT 'Created' volinfo_field $V0 'Status';
## Start volume and verify
TEST $CLI volume start $V0;
EXPECT 'Started' volinfo_field $V0 'Status';
TEST glusterfs --volfile-id=/$V0 --volfile-server=$H0 $M0 --attribute-timeout=0 --entry-timeout=0
TEST kill_brick $V0 $H0 $B0/$V0"1"
sleep 5
TEST kill_brick $V0 $H0 $B0/$V0"3"
sleep 5
for i in {1..100}; do echo "STRING" > $M0/File$i; done
brick_2_sh_entries=$(count_sh_entries $B0/$V0"2")
brick_4_sh_entries=$(count_sh_entries $B0/$V0"4")
command_output=$(gluster volume heal $V0 statistics heal-count)
substring="Number of entries:"
count=0
while read -r line;
do
if [[ "$line" == *$substring* ]]
then
value=$(echo $line | cut -f 2 -d :)
count=$(($count + $value))
fi
done <<< "$command_output"
brick_2_entries_count=$(($count-$value))
brick_4_entries_count=$value
xattrop_count_brick_2=$(count_sh_entries $B0/$V0"2")
##Remove the count of the xattrop-gfid entry count as it does not contribute
##to the number of files to be healed
sub_val=1
xattrop_count_brick_2=$(($xattrop_count_brick_2-$sub_val))
xattrop_count_brick_4=$(count_sh_entries $B0/$V0"4")
##Remove xattrop-gfid entry count
xattrop_count_brick_4=$(($xattrop_count_brick_4-$sub_val))
ret=0
if [ "$xattrop_count_brick_2" -eq "$brick_2_entries_count" ]
then
ret=$(($ret + $sub_val))
fi
EXPECT "1" echo $ret
ret=0
if [ "$xattrop_count_brick_4" -eq "$brick_4_entries_count" ]
then
ret=$(($ret + $sub_val))
fi
EXPECT "1" echo $ret
## Finish up
TEST $CLI volume stop $V0;
EXPECT 'Stopped' volinfo_field $V0 'Status';
TEST $CLI volume delete $V0;
TEST ! $CLI volume info $V0
cleanup;

138
xlators/cluster/afr/src/afr-self-heald.c
View File

@ -25,7 +25,8 @@ typedef enum {
typedef enum {
HEAL = 1,
INFO
INFO,
STATISTICS_TO_BE_HEALED,
} shd_crawl_op;
typedef struct shd_dump {
@ -446,6 +447,46 @@ out:
return;
}
int
_count_hard_links_under_base_indices_dir (xlator_t *this,
afr_crawl_data_t *crawl_data,
gf_dirent_t *entry, loc_t *childloc,
loc_t *parentloc, struct iatt *iattr)
{
xlator_t *readdir_xl = crawl_data->readdir_xl;
struct iatt parent = {0};
int ret = 0;
dict_t *output = NULL;
int xl_id = 0;
char key[256] = {0};
int child = -1;
uint64_t hardlinks = 0;
output = crawl_data->op_data;
child = crawl_data->child;
ret = syncop_lookup (readdir_xl, childloc, NULL, iattr, NULL, &parent);
if (ret)
goto out;
ret = dict_get_int32 (output, this->name, &xl_id);
if (ret)
goto out;
snprintf (key, sizeof (key), "%d-%d-hardlinks", xl_id, child);
ret = dict_get_uint64 (output, key, &hardlinks);
/*Removing the count of base_entry under indices/base_indicies and
* entry under indices/xattrop */
hardlinks = hardlinks + iattr->ia_nlink - 2;
ret = dict_set_uint64 (output, key, hardlinks);
if (ret)
goto out;
out:
return ret;
}
int
_add_summary_to_dict (xlator_t *this, afr_crawl_data_t *crawl_data,
gf_dirent_t *entry,
@ -724,12 +765,20 @@ _do_crawl_op_on_local_subvols (xlator_t *this, afr_crawl_type_t crawl,
status = "Started self-heal";
_do_self_heal_on_subvol (this, i,
crawl);
} else if (output) {
} else if (output && (op == INFO)) {
status = "";
afr_start_crawl (this, i, INDEX,
_add_summary_to_dict,
output, _gf_false, 0,
NULL);
} else if (output &&
(op == STATISTICS_TO_BE_HEALED)) {
status = "";
afr_start_crawl (this, i,
INDEX_TO_BE_HEALED,
_count_hard_links_under_base_indices_dir,
output, _gf_false,
0, NULL);
}
}
if (output) {
@ -922,6 +971,12 @@ afr_xl_op (xlator_t *this, dict_t *input, dict_t *output)
case GF_AFR_OP_STATISTICS:
ret = _add_local_subvols_crawl_statistics_to_dict (this, output);
break;
case GF_AFR_OP_STATISTICS_HEAL_COUNT:
case GF_AFR_OP_STATISTICS_HEAL_COUNT_PER_REPLICA:
ret = _do_crawl_op_on_local_subvols (this, INDEX_TO_BE_HEALED,
STATISTICS_TO_BE_HEALED,
output);
break;
default:
gf_log (this->name, GF_LOG_ERROR, "Unknown set op %d", op);
break;
@ -1094,6 +1149,7 @@ afr_crawl_build_start_loc (xlator_t *this, afr_crawl_data_t *crawl_data,
afr_private_t *priv = NULL;
dict_t *xattr = NULL;
void *index_gfid = NULL;
void *base_indices_holder_vgfid = NULL;
loc_t rootloc = {0};
struct iatt iattr = {0};
struct iatt parent = {0};
@ -1103,7 +1159,7 @@ afr_crawl_build_start_loc (xlator_t *this, afr_crawl_data_t *crawl_data,
priv = this->private;
if (crawl_data->crawl == FULL) {
afr_build_root_loc (this, dirloc);
} else {
} else if (crawl_data->crawl == INDEX) {
afr_build_root_loc (this, &rootloc);
ret = syncop_getxattr (readdir_xl, &rootloc, &xattr,
GF_XATTROP_INDEX_GFID);
@ -1137,6 +1193,47 @@ afr_crawl_build_start_loc (xlator_t *this, afr_crawl_data_t *crawl_data,
ret = _link_inode_update_loc (this, dirloc, &iattr);
if (ret)
goto out;
} else if (crawl_data->crawl == INDEX_TO_BE_HEALED) {
afr_build_root_loc (this, &rootloc);
ret = syncop_getxattr (readdir_xl, &rootloc, &xattr,
GF_BASE_INDICES_HOLDER_GFID);
if (ret < 0)
goto out;
ret = dict_get_ptr (xattr, GF_BASE_INDICES_HOLDER_GFID,
&base_indices_holder_vgfid);
if (ret < 0) {
gf_log (this->name, GF_LOG_ERROR, "index gfid empty "
"on %s", readdir_xl->name);
ret = -1;
goto out;
}
if (!base_indices_holder_vgfid) {
gf_log (this->name, GF_LOG_ERROR, "Base indices holder"
"virtual gfid is null on %s", readdir_xl->name);
ret = -1;
goto out;
}
uuid_copy (dirloc->gfid, base_indices_holder_vgfid);
dirloc->path = "";
dirloc->inode = inode_new (priv->root_inode->table);
ret = syncop_lookup (readdir_xl, dirloc, NULL, &iattr, NULL,
&parent);
if (ret < 0) {
if (errno != ENOENT) {
gf_log (this->name, GF_LOG_ERROR, "lookup "
"failed for base_indices_holder dir"
" on %s - (%s)", readdir_xl->name,
strerror (errno));
} else {
gf_log (this->name, GF_LOG_ERROR, "base_indices"
"_holder is not yet created.");
}
goto out;
}
ret = _link_inode_update_loc (this, dirloc, &iattr);
if (ret)
goto out;
}
ret = 0;
out:
@ -1201,6 +1298,16 @@ afr_crawl_build_child_loc (xlator_t *this, loc_t *child, loc_t *parent,
priv = this->private;
if (crawl_data->crawl == FULL) {
ret = afr_build_child_loc (this, child, parent, entry->d_name);
} else if (crawl_data->crawl == INDEX_TO_BE_HEALED) {
ret = _build_index_loc (this, child, entry->d_name, parent);
if (ret)
goto out;
child->inode = inode_new (priv->root_inode->table);
if (!child->inode) {
ret = -1;
goto out;
}
child->path = NULL;
} else {
child->inode = inode_new (priv->root_inode->table);
if (!child->inode)
@ -1250,10 +1357,14 @@ _process_entries (xlator_t *this, loc_t *parentloc, gf_dirent_t *entries,
ret = crawl_data->process_entry (this, crawl_data, entry,
&entry_loc, parentloc, &iattr);
if (ret)
if (crawl_data->crawl == INDEX_TO_BE_HEALED && ret) {
goto out;
} else if (ret) {
continue;
}
if (crawl_data->crawl == INDEX)
if ((crawl_data->crawl == INDEX) ||
(crawl_data->crawl == INDEX_TO_BE_HEALED))
continue;
if (!IA_ISDIR (iattr.ia_type))
@ -1268,6 +1379,10 @@ _process_entries (xlator_t *this, loc_t *parentloc, gf_dirent_t *entries,
}
ret = 0;
out:
if ((crawl_data->crawl == INDEX_TO_BE_HEALED) && ret) {
gf_log (this->name, GF_LOG_ERROR,"Failed to get the hardlink "
"count");
}
loc_wipe (&entry_loc);
return ret;
}
@ -1315,6 +1430,9 @@ _crawl_directory (fd_t *fd, loc_t *loc, afr_crawl_data_t *crawl_data)
ret = _process_entries (this, loc, &entries, &offset,
crawl_data);
if ((ret < 0) && (crawl_data->crawl == INDEX_TO_BE_HEALED)) {
goto out;
}
gf_dirent_free (&entries);
free_entries = _gf_false;
}
@ -1420,8 +1538,13 @@ afr_dir_crawl (void *data)
goto out;
ret = afr_crawl_opendir (this, crawl_data, &fd, &dirloc);
if (ret)
if (ret) {
if (crawl_data->crawl == INDEX_TO_BE_HEALED) {
gf_log (this->name, GF_LOG_ERROR, "Failed to open base_"
"indices_holder");
}
goto out;
}
ret = _crawl_directory (fd, &dirloc, crawl_data);
if (ret)
@ -1435,7 +1558,8 @@ afr_dir_crawl (void *data)
out:
if (fd)
fd_unref (fd);
if (crawl_data->crawl == INDEX)
if ((crawl_data->crawl == INDEX) ||
(crawl_data->crawl == INDEX_TO_BE_HEALED ))
dirloc.path = NULL;
loc_wipe (&dirloc);
return ret;

1
xlators/cluster/afr/src/afr.h
View File

@ -94,6 +94,7 @@ typedef struct afr_inode_ctx_ {
typedef enum {
NONE,
INDEX,
INDEX_TO_BE_HEALED,
FULL,
} afr_crawl_type_t;

315
xlators/features/index/src/index.c
View File

@ -15,8 +15,10 @@
#include "index.h"
#include "options.h"
#include "glusterfs3-xdr.h"
#include "syncop.h"
#define XATTROP_SUBDIR "xattrop"
#define BASE_INDICES_HOLDER_SUBDIR "base_indices_holder"
call_stub_t *
__index_dequeue (struct list_head *callstubs)
@ -242,20 +244,40 @@ check_delete_stale_index_file (xlator_t *this, char *filename)
{
int ret = 0;
struct stat st = {0};
struct stat base_index_st = {0};
char filepath[PATH_MAX] = {0};
char filepath_under_base_indices_holder[PATH_MAX] = {0};
index_priv_t *priv = NULL;
priv = this->private;
if (priv->to_be_healed_states != synced_state)
return;
make_file_path (priv->index_basepath, XATTROP_SUBDIR,
filename, filepath, sizeof (filepath));
make_file_path (priv->index_basepath, BASE_INDICES_HOLDER_SUBDIR,
filename, filepath_under_base_indices_holder,
sizeof (filepath_under_base_indices_holder));
ret = stat (filepath_under_base_indices_holder, &base_index_st);
if (ret) {
gf_log (THIS->name, GF_LOG_ERROR, "Base index is not created"
"under index/base_indices_holder");
return;
}
ret = stat (filepath, &st);
if (!ret && st.st_nlink == 1)
if (!ret && st.st_nlink == 2) {
unlink (filepath);
unlink (filepath_under_base_indices_holder);
}
}
static int
index_fill_readdir (fd_t *fd, DIR *dir, off_t off,
size_t size, gf_dirent_t *entries)
size_t size, gf_dirent_t *entries, readdir_directory type)
{
off_t in_case = -1;
size_t filled = 0;
@ -298,7 +320,8 @@ index_fill_readdir (fd_t *fd, DIR *dir, off_t off,
}
if (!strncmp (entry->d_name, XATTROP_SUBDIR"-",
strlen (XATTROP_SUBDIR"-"))) {
strlen (XATTROP_SUBDIR"-")) &&
(type == INDEX_XATTROP)) {
check_delete_stale_index_file (this, entry->d_name);
continue;
}
@ -336,17 +359,176 @@ out:
return count;
}
int
sync_base_indices (void *index_priv)
{
index_priv_t *priv = NULL;
DIR *dir_base_holder = NULL;
DIR *xattrop_dir = NULL;
struct dirent *entry = NULL;
char base_indices_holder[PATH_MAX] = {0};
char xattrop_directory[PATH_MAX] = {0};
char base_index_path[PATH_MAX] = {0};
char xattrop_index_path[PATH_MAX] = {0};
int ret = 0;
priv = index_priv;
snprintf (base_indices_holder, PATH_MAX, "%s/%s", priv->index_basepath,
BASE_INDICES_HOLDER_SUBDIR);
snprintf (xattrop_directory, PATH_MAX, "%s/%s", priv->index_basepath,
XATTROP_SUBDIR);
if ((dir_base_holder = opendir(base_indices_holder)) == NULL) {
ret = -1;
goto out;
}
if ((xattrop_dir = opendir (xattrop_directory)) == NULL) {
ret = -1;
goto out;
}
priv->to_be_healed_states = sync_started;
while ((entry = readdir(xattrop_dir)) != NULL) {
if (!strcmp (entry->d_name, ".") ||
!strcmp (entry->d_name, "..")) {
continue;
}
if (strncmp (entry->d_name, XATTROP_SUBDIR"-",
strlen (XATTROP_SUBDIR"-"))) {
continue;
}
if (!strncmp (entry->d_name, XATTROP_SUBDIR"-",
strlen (XATTROP_SUBDIR"-"))) {
snprintf (xattrop_index_path, PATH_MAX, "%s/%s",
xattrop_directory, entry->d_name);
snprintf (base_index_path, PATH_MAX, "%s/%s",
base_indices_holder, entry->d_name);
ret = link (xattrop_index_path, base_index_path);
if (ret && errno != EEXIST)
goto out;
}
}
ret = closedir (xattrop_dir);
if (ret)
goto out;
ret = closedir (dir_base_holder);
if (ret)
goto out;
ret = 0;
out:
return ret;
}
int
base_indices_syncing_done (int ret, call_frame_t *frame, void *data)
{
index_priv_t *priv = NULL;
priv = data;
if (!priv)
goto out;
if (ret) {
priv->to_be_healed_states = sync_not_started;
} else {
priv->to_be_healed_states = synced_state;
}
STACK_DESTROY (frame->root);
out:
return 0;
}
int
sync_base_indices_from_xattrop (xlator_t *this)
{
index_priv_t *priv = NULL;
char base_indices_holder[PATH_MAX] = {0};
int ret = 0;
struct stat st = {0};
DIR *dir = NULL;
struct dirent *entry = NULL;
call_frame_t *frame = NULL;
priv = this->private;
if (priv->to_be_healed_states != sync_not_started) {
ret = -1;
goto out;
}
snprintf (base_indices_holder, PATH_MAX, "%s/%s", priv->index_basepath,
BASE_INDICES_HOLDER_SUBDIR);
ret = stat (base_indices_holder, &st);
if (ret && (errno != ENOENT)) {
goto out;
} else if (errno == ENOENT) {
ret = index_dir_create (this, BASE_INDICES_HOLDER_SUBDIR);
if (ret)
goto out;
} else {
if ((dir = opendir (base_indices_holder)) == NULL) {
ret = -1;
goto out;
}
while ((entry = readdir (dir)) != NULL) {
if (!strcmp (entry->d_name, ".") ||
!strcmp (entry->d_name,"..")) {
continue;
}
ret = unlink (entry->d_name);
if (ret)
goto out;
}
closedir (dir);
}
/*At this point of time we have index/base_indicies_holder directory
*is with no entries*/
frame = create_frame (this, this->ctx->pool);
if (!frame) {
ret = -1;
goto out;
}
set_lk_owner_from_ptr (&frame->root->lk_owner, frame->root);
frame->root->pid = LOW_PRIO_PROC_PID;
ret = synctask_new (this->ctx->env, sync_base_indices,
base_indices_syncing_done,frame, priv);
out:
return ret;
}
int
index_add (xlator_t *this, uuid_t gfid, const char *subdir)
{
int32_t op_errno = 0;
char gfid_path[PATH_MAX] = {0};
char index_path[PATH_MAX] = {0};
char base_path[PATH_MAX] = {0};
int ret = 0;
uuid_t index = {0};
index_priv_t *priv = NULL;
struct stat st = {0};
int fd = 0;
int index_created = 0;
priv = this->private;
GF_ASSERT_AND_GOTO_WITH_ERROR (this->name, !uuid_is_null (gfid),
@ -364,9 +546,11 @@ index_add (xlator_t *this, uuid_t gfid, const char *subdir)
ret = link (index_path, gfid_path);
if (!ret || (errno == EEXIST)) {
ret = 0;
index_created = 1;
goto out;
}
op_errno = errno;
if (op_errno == ENOENT) {
ret = index_dir_create (this, subdir);
@ -398,10 +582,36 @@ index_add (xlator_t *this, uuid_t gfid, const char *subdir)
"add to index (%s)", uuid_utoa (gfid),
strerror (errno));
goto out;
} else {
index_created = 1;
}
if (priv->to_be_healed_states != sync_not_started) {
make_index_path (priv->index_basepath,
GF_BASE_INDICES_HOLDER_GFID,
index, base_path, sizeof (base_path));
ret = link (index_path, base_path);
if (ret)
goto out;
}
ret = 0;
out:
/*If base_indices_holder is not created: create and sync
*If directory is present: delete contents and start syncing
*If syncing is in progress :No need to do any thing
*If syncing is done: No need to do anything*/
if (!ret) {
switch (priv->to_be_healed_states) {
case sync_not_started:
ret = sync_base_indices_from_xattrop (this);
break;
case sync_started:
case synced_state:
/*No need to do anything*/
break;
}
}
return ret;
}
@ -737,8 +947,18 @@ index_getxattr_wrapper (call_frame_t *frame, xlator_t *this,
goto done;
}
ret = dict_set_static_bin (xattr, (char*)name, priv->xattrop_vgfid,
sizeof (priv->xattrop_vgfid));
if (!strcmp (name, GF_XATTROP_INDEX_GFID)) {
ret = dict_set_static_bin (xattr, (char*)name,
priv->xattrop_vgfid,
sizeof (priv->xattrop_vgfid));
} else if (!strcmp (name, GF_BASE_INDICES_HOLDER_GFID)) {
ret = dict_set_static_bin (xattr, (char*)name,
priv->base_indices_holder_vgfid,
sizeof (priv->base_indices_holder_vgfid));
}
if (ret) {
ret = -ENOMEM;
gf_log (THIS->name, GF_LOG_ERROR, "xattrop index "
@ -782,6 +1002,15 @@ index_lookup_wrapper (call_frame_t *frame, xlator_t *this,
} else if (!uuid_compare (loc->pargfid, priv->xattrop_vgfid)) {
make_file_path (priv->index_basepath, XATTROP_SUBDIR,
loc->name, path, sizeof (path));
} else if (!uuid_compare (loc->gfid,priv->base_indices_holder_vgfid)){
make_index_dir_path (priv->index_basepath,
BASE_INDICES_HOLDER_SUBDIR, path,
sizeof (path));
is_dir = _gf_true;
} else if (!uuid_compare (loc->pargfid, priv->base_indices_holder_vgfid)) {
make_file_path (priv->index_basepath,
BASE_INDICES_HOLDER_SUBDIR,loc->name, path,
sizeof (path));
}
ret = lstat (path, &lstatbuf);
@ -803,10 +1032,14 @@ index_lookup_wrapper (call_frame_t *frame, xlator_t *this,
}
iatt_from_stat (&stbuf, &lstatbuf);
if (is_dir)
if (is_dir && !uuid_compare (loc->gfid, priv->xattrop_vgfid)) {
uuid_copy (stbuf.ia_gfid, priv->xattrop_vgfid);
else
} else if (is_dir &&
!uuid_compare (loc->gfid, priv->base_indices_holder_vgfid)) {
uuid_copy (stbuf.ia_gfid, priv->base_indices_holder_vgfid);
} else {
uuid_generate (stbuf.ia_gfid);
}
stbuf.ia_ino = -1;
op_ret = 0;
done:
@ -817,6 +1050,44 @@ done:
return 0;
}
int32_t
base_indices_readdir_wrapper (call_frame_t *frame, xlator_t *this,
fd_t *fd, size_t size, off_t off, dict_t *xdata)
{
index_priv_t *priv = NULL;
char base_indices_holder[PATH_MAX] = {0};
DIR *dir = NULL;
int32_t op_ret = -1;
int32_t op_errno = 0;
int count = 0;
gf_dirent_t entries;
priv = this->private;
make_index_dir_path (priv->index_basepath, BASE_INDICES_HOLDER_SUBDIR,
base_indices_holder, sizeof (base_indices_holder));
dir = opendir (base_indices_holder);
if (!dir) {
op_errno = EINVAL;
goto done;
}
INIT_LIST_HEAD (&entries.list);
count = index_fill_readdir (fd, dir, off, size, &entries,
BASE_INDICES_HOLDER);
/* pick ENOENT to indicate EOF */
op_errno = errno;
op_ret = count;
closedir (dir);
done:
STACK_UNWIND_STRICT (readdir, frame, op_ret, op_errno, &entries, xdata);
gf_dirent_free (&entries);
return 0;
}
int32_t
index_readdir_wrapper (call_frame_t *frame, xlator_t *this,
fd_t *fd, size_t size, off_t off, dict_t *xdata)
@ -848,7 +1119,8 @@ index_readdir_wrapper (call_frame_t *frame, xlator_t *this,
goto done;
}
count = index_fill_readdir (fd, dir, off, size, &entries);
count = index_fill_readdir (fd, dir, off, size, &entries,
INDEX_XATTROP);
/* pick ENOENT to indicate EOF */
op_errno = errno;
@ -915,7 +1187,10 @@ index_getxattr (call_frame_t *frame, xlator_t *this,
{
call_stub_t *stub = NULL;
if (!name || strcmp (GF_XATTROP_INDEX_GFID, name))
if (!name)
goto out;
if (strcmp (GF_XATTROP_INDEX_GFID, name) &&
strcmp (GF_BASE_INDICES_HOLDER_GFID, name))
goto out;
stub = fop_getxattr_stub (frame, index_getxattr_wrapper, loc, name,
@ -942,7 +1217,9 @@ index_lookup (call_frame_t *frame, xlator_t *this,
priv = this->private;
if (uuid_compare (loc->gfid, priv->xattrop_vgfid) &&
uuid_compare (loc->pargfid, priv->xattrop_vgfid))
uuid_compare (loc->pargfid, priv->xattrop_vgfid) &&
uuid_compare (loc->gfid, priv->base_indices_holder_vgfid) &&
uuid_compare (loc->pargfid, priv->base_indices_holder_vgfid))
goto normal;
stub = fop_lookup_stub (frame, index_lookup_wrapper, loc, xattr_req);
@ -968,10 +1245,19 @@ index_readdir (call_frame_t *frame, xlator_t *this,
index_priv_t *priv = NULL;
priv = this->private;
if (uuid_compare (fd->inode->gfid, priv->xattrop_vgfid))
if (uuid_compare (fd->inode->gfid, priv->xattrop_vgfid) &&
uuid_compare (fd->inode->gfid, priv->base_indices_holder_vgfid))
goto out;
stub = fop_readdir_stub (frame, index_readdir_wrapper, fd, size, off,
xdata);
if (!uuid_compare (fd->inode->gfid, priv->xattrop_vgfid)) {
stub = fop_readdir_stub (frame, index_readdir_wrapper, fd, size,
off, xdata);
} else if (!uuid_compare (fd->inode->gfid,
priv->base_indices_holder_vgfid)) {
stub = fop_readdir_stub (frame, base_indices_readdir_wrapper,
fd, size, off, xdata);
}
if (!stub) {
STACK_UNWIND_STRICT (readdir, frame, -1, ENOMEM, NULL, NULL);
return 0;
@ -1075,6 +1361,9 @@ init (xlator_t *this)
GF_OPTION_INIT ("index-base", priv->index_basepath, path, out);
uuid_generate (priv->index);
uuid_generate (priv->xattrop_vgfid);
/*base_indices_holder is a directory which contains hard links to
* all base indices inside indices/xattrop directory*/
uuid_generate (priv->base_indices_holder_vgfid);
INIT_LIST_HEAD (&priv->callstubs);
this->private = priv;

14
xlators/features/index/src/index.h
View File

@ -36,14 +36,28 @@ typedef struct index_fd_ctx {
DIR *dir;
} index_fd_ctx_t;
typedef enum {
sync_not_started,
sync_started,
synced_state,
} to_be_healed_states_t;
typedef enum {
INDEX_XATTROP,
BASE_INDICES_HOLDER,
} readdir_directory;
typedef struct index_priv {
char *index_basepath;
uuid_t index;
gf_lock_t lock;
uuid_t xattrop_vgfid;//virtual gfid of the xattrop index dir
uuid_t base_indices_holder_vgfid; //virtual gfid of the
//to_be_healed_xattrop directory
struct list_head callstubs;
pthread_mutex_t mutex;
pthread_cond_t cond;
to_be_healed_states_t to_be_healed_states;
} index_priv_t;
#define INDEX_STACK_UNWIND(fop, frame, params ...) \

174
xlators/mgmt/glusterd/src/glusterd-op-sm.c
View File

@ -4531,25 +4531,96 @@ out:
return ret;
}
int
get_replica_index_for_per_replica_cmd (glusterd_volinfo_t *volinfo,
dict_t *dict) {
int ret = 0;
char *hostname = NULL;
char *path = NULL;
int index = 0;
glusterd_brickinfo_t *brickinfo = NULL;
int cmd_replica_index = -1;
int replica_count = -1;
if (!dict) {
ret = -1;
goto out;
}
ret = dict_get_str (dict, "per-replica-cmd-hostname", &hostname);
if (ret)
goto out;
ret = dict_get_str (dict, "per-replica-cmd-path", &path);
if (ret)
goto out;
replica_count = volinfo->replica_count;
list_for_each_entry (brickinfo, &volinfo->bricks, brick_list) {
if (uuid_is_null (brickinfo->uuid))
(void)glusterd_resolve_brick (brickinfo);
if (!strcmp (brickinfo->path, path) &&
!strcmp (brickinfo->hostname, hostname)) {
cmd_replica_index = index/(replica_count);
goto out;
}
index++;
}
out:
if (ret)
cmd_replica_index = -1;
return cmd_replica_index;
}
int
_select_rxlators_with_local_bricks (xlator_t *this, glusterd_volinfo_t *volinfo,
dict_t *dict)
dict_t *dict, cli_cmd_type type)
{
glusterd_brickinfo_t *brickinfo = NULL;
glusterd_conf_t *priv = NULL;
int index = 1;
int index = 0;
int rxlator_count = 0;
int replica_count = 0;
gf_boolean_t add = _gf_false;
int ret = 0;
int cmd_replica_index = -1;
priv = this->private;
replica_count = volinfo->replica_count;
if (type == PER_REPLICA) {
cmd_replica_index = get_replica_index_for_per_replica_cmd
(volinfo, dict);
if (cmd_replica_index == -1) {
ret = -1;
goto err;
}
}
index = 1;
list_for_each_entry (brickinfo, &volinfo->bricks, brick_list) {
if (uuid_is_null (brickinfo->uuid))
(void)glusterd_resolve_brick (brickinfo);
if (!uuid_compare (MY_UUID, brickinfo->uuid))
add = _gf_true;
switch (type) {
case ALL_REPLICA:
if (!uuid_compare (MY_UUID, brickinfo->uuid))
add = _gf_true;
break;
case PER_REPLICA:
if (!uuid_compare (MY_UUID, brickinfo->uuid) &&
((index-1)/replica_count == cmd_replica_index))
add = _gf_true;
break;
}
if (index % replica_count == 0) {
if (add) {
_add_rxlator_to_dict (dict, volinfo->volname,
@ -4562,6 +4633,10 @@ _select_rxlators_with_local_bricks (xlator_t *this, glusterd_volinfo_t *volinfo,
index++;
}
err:
if (ret)
rxlator_count = -1;
return rxlator_count;
}
@ -4659,7 +4734,8 @@ out:
#endif
static int
fill_shd_status_for_local_bricks (dict_t *dict, glusterd_volinfo_t *volinfo)
fill_shd_status_for_local_bricks (dict_t *dict, glusterd_volinfo_t *volinfo,
cli_cmd_type type, dict_t *req_dict)
{
glusterd_brickinfo_t *brickinfo = NULL;
char msg[1024] = {0,};
@ -4668,10 +4744,22 @@ fill_shd_status_for_local_bricks (dict_t *dict, glusterd_volinfo_t *volinfo)
int index = 0;
int ret = 0;
xlator_t *this = NULL;
int cmd_replica_index = -1;
this = THIS;
snprintf (msg, sizeof (msg), "self-heal-daemon is not running on");
if (type == PER_REPLICA) {
cmd_replica_index = get_replica_index_for_per_replica_cmd
(volinfo, req_dict);
if (cmd_replica_index == -1) {
gf_log (THIS->name, GF_LOG_ERROR, "Could not find the "
"replica index for per replica type command");
ret = -1;
goto out;
}
}
list_for_each_entry (brickinfo, &volinfo->bricks, brick_list) {
if (uuid_is_null (brickinfo->uuid))
(void)glusterd_resolve_brick (brickinfo);
@ -4680,6 +4768,14 @@ fill_shd_status_for_local_bricks (dict_t *dict, glusterd_volinfo_t *volinfo)
index++;
continue;
}
if (type == PER_REPLICA) {
if (cmd_replica_index != (index/volinfo->replica_count)) {
index++;
continue;
}
}
snprintf (key, sizeof (key), "%d-status",index);
snprintf (value, sizeof (value), "%s %s",msg,
uuid_utoa(MY_UUID));
@ -4748,21 +4844,49 @@ glusterd_bricks_select_heal_volume (dict_t *dict, char **op_errstr,
goto out;
}
switch (heal_op) {
case GF_AFR_OP_INDEX_SUMMARY:
case GF_AFR_OP_STATISTICS_HEAL_COUNT:
if (!glusterd_is_nodesvc_online ("glustershd")) {
if (!rsp_dict) {
gf_log (this->name, GF_LOG_ERROR, "Received "
"empty ctx.");
goto out;
}
if (!glusterd_is_nodesvc_online ("glustershd") &&
(heal_op == GF_AFR_OP_INDEX_SUMMARY)) {
if (!rsp_dict) {
gf_log (this->name, GF_LOG_ERROR, "Received empty "
"ctx.");
ret = fill_shd_status_for_local_bricks (rsp_dict,
volinfo,
ALL_REPLICA,
dict);
if (ret)
gf_log (this->name, GF_LOG_ERROR, "Unable to "
"fill the shd status for the local "
"bricks");
goto out;
}
ret = fill_shd_status_for_local_bricks (rsp_dict, volinfo);
if (ret)
gf_log (this->name, GF_LOG_ERROR, "Unable to fill the shd"
" status for the local bricks");
goto out;
}
break;
case GF_AFR_OP_STATISTICS_HEAL_COUNT_PER_REPLICA:
if (!glusterd_is_nodesvc_online ("glustershd")) {
if (!rsp_dict) {
gf_log (this->name, GF_LOG_ERROR, "Received "
"empty ctx.");
goto out;
}
ret = fill_shd_status_for_local_bricks (rsp_dict,
volinfo,
PER_REPLICA,
dict);
if (ret)
gf_log (this->name, GF_LOG_ERROR, "Unable to "
"fill the shd status for the local"
" bricks.");
goto out;
}
break;
default:
break;
}
@ -4772,14 +4896,28 @@ glusterd_bricks_select_heal_volume (dict_t *dict, char **op_errstr,
volinfo,
dict);
break;
case GF_AFR_OP_STATISTICS_HEAL_COUNT_PER_REPLICA:
rxlator_count = _select_rxlators_with_local_bricks (this,
volinfo,
dict,
PER_REPLICA);
break;
default:
rxlator_count = _select_rxlators_with_local_bricks (this,
volinfo,
dict);
dict,
ALL_REPLICA);
break;
}
if (!rxlator_count)
goto out;
if (rxlator_count == -1){
gf_log (this->name, GF_LOG_ERROR, "Could not determine the"
"translator count");
ret = -1;
goto out;
}
ret = dict_set_int32 (dict, "count", rxlator_count);
if (ret)
goto out;

6
xlators/mgmt/glusterd/src/glusterd-op-sm.h
View File

@ -162,6 +162,12 @@ typedef struct glusterd_gsync_status_temp {
glusterd_volinfo_t *volinfo;
char *node;
}glusterd_gsync_status_temp_t;
typedef enum cli_cmd_type_ {
PER_REPLICA,
ALL_REPLICA,
} cli_cmd_type;
int
glusterd_op_sm_new_event (glusterd_op_sm_event_type_t event_type,
glusterd_op_sm_event_t **new_event);

1
xlators/mgmt/glusterd/src/glusterd-utils.c
View File

@ -7390,6 +7390,7 @@ glusterd_heal_volume_brick_rsp (dict_t *req_dict, dict_t *rsp_dict,
else
dict_foreach (rsp_dict, _heal_volume_add_shd_rsp, &rsp_ctx);
out:
return ret;
}

24
xlators/mgmt/glusterd/src/glusterd-volume-ops.c
View File

@ -1211,14 +1211,22 @@ glusterd_op_stage_heal_volume (dict_t *dict, char **op_errstr)
goto out;
}
if ((heal_op != GF_AFR_OP_INDEX_SUMMARY) &&
!glusterd_is_nodesvc_online ("glustershd")) {
ret = -1;
*op_errstr = gf_strdup ("Self-heal daemon is not running."
" Check self-heal daemon log file.");
gf_log (this->name, GF_LOG_WARNING, "%s", "Self-heal daemon is "
"not running. Check self-heal daemon log file.");
goto out;
switch (heal_op) {
case GF_AFR_OP_INDEX_SUMMARY:
case GF_AFR_OP_STATISTICS_HEAL_COUNT:
case GF_AFR_OP_STATISTICS_HEAL_COUNT_PER_REPLICA:
break;
default:
if (!glusterd_is_nodesvc_online("glustershd")){
ret = -1;
*op_errstr = gf_strdup ("Self-heal daemon is "
"not running. Check self-heal "
"daemon log file.");
gf_log (this->name, GF_LOG_WARNING, "%s",
"Self-heal daemon is not running."
"Check self-heal daemon log file.");
goto out;
}
}
ret = 0;