Changes to NSR reconciliation code.

Following is list of changes:
1) Simulation of etcd using a file as a registry protected using locks.
2) Implement notifications for child up and child down.
3) Join a new brick into quorum.
4) add support for proper fencing and draining of IO required for reconciliaiton
5) misc changes and addressed review comments.

Change-Id: Iddd1137ad6205252ed03301888bb1e83fa2221e0
Signed-off-by: Raghavan P <rpichai@redhat.com>

4 files changed, 181 insertions, 66 deletions

diff --git a/xlators/cluster/nsr-recon/src/recon_driver.c b/xlators/cluster/nsr-recon/src/recon_driver.c
index 1328d52dc..2e2299ad1 100644
--- a/xlators/cluster/nsr-recon/src/recon_driver.c
+++ b/xlators/cluster/nsr-recon/src/recon_driver.c
@@ -495,7 +495,7 @@ control_worker_func_0(nsr_per_node_worker_t *ctx,
                                         nsr_worker_log(this->name, GF_LOG_INFO,
                                                        "out of get term info for term %d. got ops %d with first %d and last %d \n",
                                                        recon_info->last_term, recon_info->commited_ops,
-                                                       recon_info->last_index, recon_info->first_index);
+                                                       recon_info->first_index, recon_info->last_index);
 
                                         break;
                                 }
@@ -538,12 +538,14 @@ control_worker_func_0(nsr_per_node_worker_t *ctx,
                                                        "trying to get reconciliation window records for node %d for term %d with first %d last %d\n",
                                                        index, recon_info->last_term, recon_info->first_index, recon_info->last_index);
 
-                                        GF_ASSERT(num <= MAX_RECONCILIATION_WINDOW_SIZE);
 
                                         // TBD - handle buffer allocation errors
                                         rd = GF_CALLOC(num,
                                                        sizeof(nsr_recon_record_details_t),
                                                        gf_mt_recon_private_t);
+                                        recon_info->records = GF_CALLOC(num,
+                                                       sizeof(nsr_reconciliation_record_t),
+                                                       gf_mt_recon_private_t);
 
                                         // TBD - handle errors
                                         nsr_recon_libchangelog_get_records(this, priv->changelog_base_path,
@@ -684,7 +686,7 @@ control_worker_func(nsr_per_node_worker_t *ctx,
                                         nsr_worker_log(this->name, GF_LOG_INFO,
                                                        "out of get last term info with current term %d. got ops %d with first %d and last %d \n",
                                                        recon_info->last_term, recon_info->commited_ops,
-                                                       recon_info->last_index, recon_info->first_index);
+                                                       recon_info->first_index, recon_info->last_index);
 
                                         break;
                                 }
@@ -711,7 +713,7 @@ control_worker_func(nsr_per_node_worker_t *ctx,
                                         nsr_worker_log(this->name, GF_LOG_INFO,
                                                        "out of get term info for term %d. got ops %d with first %d and last %d \n",
                                                        recon_info->last_term, recon_info->commited_ops,
-                                                       recon_info->last_index, recon_info->first_index);
+                                                       recon_info->first_index, recon_info->last_index);
 
                                         break;
                                 }
@@ -848,7 +850,6 @@ control_worker_func(nsr_per_node_worker_t *ctx,
                                                        "trying to get reconciliation window records for node %d for term %d with first %d last %d\n",
                                                        index, recon_info->last_term, recon_info->first_index, recon_info->last_index);
 
-                                        GF_ASSERT(num <= MAX_RECONCILIATION_WINDOW_SIZE);
 
                                         // TBD - error handling for all the glfs APIs
                                         glfs_lseek(ctx->aux_fd, nsr_recon_xlator_sector_2, SEEK_SET);
@@ -864,6 +865,10 @@ control_worker_func(nsr_per_node_worker_t *ctx,
                                         rd = GF_CALLOC(num,
                                                        sizeof(nsr_recon_record_details_t),
                                                        gf_mt_recon_private_t);
+                                        recon_info->records = GF_CALLOC(num,
+                                                       sizeof(nsr_reconciliation_record_t),
+                                                       gf_mt_recon_private_t);
+
                                         glfs_read(ctx->aux_fd, rd, num * sizeof(nsr_recon_record_details_t), 0);
                                         for (i=0; i < num; i++) {
                                                 ENDIAN_CONVERSION_RD(rd[i], _gf_true); //ntohl
@@ -875,7 +880,7 @@ control_worker_func(nsr_per_node_worker_t *ctx,
                                                                recon_info->records[i].rec.type,
                                                                i + recon_info->first_index);
                                         }
-                                        free(rd);
+                                        GF_FREE(rd);
 
                                         nsr_worker_log(this->name, GF_LOG_INFO,
                                                        "got reconciliation window records for node %d for term %d \n",
@@ -1003,6 +1008,28 @@ nsr_recon_in_use(nsr_recon_driver_ctx_t *ctx,
  * Write the role and associated information to the node.
  * This gets called from recon xlator indicating node is either
  * leader, reconciliator or should do resolution.
+ */
+gf_boolean_t
+nsr_recon_driver_set_role(nsr_recon_driver_ctx_t *ctx,
+                          nsr_recon_role_t *rr,
+                          uint32_t txn_id)
+{
+        nsr_role_work_t *rw;
+
+        nsr_driver_log(this->name, GF_LOG_INFO, "set role called \n");
+        rw = GF_CALLOC(1, sizeof (nsr_role_work_t), 0);
+        memcpy(&rw->role, rr, sizeof(nsr_recon_role_t));
+        rw->txn_id = txn_id;
+        INIT_LIST_HEAD(&(rw->list));
+        pthread_mutex_lock(&(ctx->mutex));
+        list_add_tail(&rw->list, &ctx->role_head.list);
+        pthread_cond_signal(&(ctx->cv));
+        pthread_mutex_unlock(&(ctx->mutex));
+        nsr_driver_log(this->name, GF_LOG_INFO, "set role returns \n");
+        return _gf_true;
+}
+
+/*
  * First we undo the last role to make sure we clean up.
  *
  * Input arguments
@@ -1023,19 +1050,18 @@ nsr_recon_in_use(nsr_recon_driver_ctx_t *ctx,
  *      needs to be returned back after the actual reconciliation is done.
  *      For that we store the frame against this id which acts as a key.
  */
-gf_boolean_t
-nsr_recon_driver_set_role(nsr_recon_driver_ctx_t *ctx,
-                          nsr_recon_role_t *rr,
-                          uint32_t txn_id)
+nsr_recon_driver_state_t
+nsr_recon_driver_get_role(nsr_recon_driver_ctx_t *ctx,
+                          nsr_role_work_t *rw)
 {
         uint8_t i=0, j=0;
-        pthread_mutex_lock(&(ctx->mutex));
-        ctx->state = rr->role;
+        nsr_recon_role_t *rr = &(rw->role);
+
         // First make all the threads uninitialise
         for (i = 0; i < ctx->replica_group_size; i++) {
                 nsr_recon_in_use(ctx, i, _gf_false);
         }
-        if (rr->role == leader) {
+        if ((rr->role == leader) || (rr->role == joiner)) {
 
                 // First set info this node
                 nsr_recon_in_use(ctx, 0, _gf_true);
@@ -1051,10 +1077,11 @@ nsr_recon_driver_set_role(nsr_recon_driver_ctx_t *ctx,
                 for (i=1; i < ctx->replica_group_size; i++) {
                         for (j=0 ; j <  rr->num; j++) {
                                 // TBD - make this strcmp later when etcd servers set properly
-                                //if (!strcmp(ctx->workers[i].name, rr->info[j].name)) {
-                                if (strstr(ctx->workers[i].name, rr->info[j].name)) {
+                                if (!strcmp(ctx->workers[i].name, rr->info[j].name)) {
+                                //if (strstr(ctx->workers[i].name, rr->info[j].name)) {
                                         nsr_driver_log(this->name, GF_LOG_INFO,
-                                                       "nsr_recon_driver_set_role: this as leader. found other server %s\n",
+                                                       "nsr_recon_driver_set_role: this as %s. found other server %s\n",
+                                                       (rr->role == leader) ? "leader" : "joiner",
                                                        ctx->workers[i].name);
 
                                         nsr_recon_in_use(ctx, i, _gf_true);
@@ -1071,14 +1098,19 @@ nsr_recon_driver_set_role(nsr_recon_driver_ctx_t *ctx,
                                 }
                         }
                 }
-                ctx->reconciliator_index = -1;
+                // If leader, reconciliator has to be chosen.
+                // If joiner, we are the reconciliator.
+                if (rr->role == leader)
+                        ctx->reconciliator_index = -1;
+                else
+                        ctx->reconciliator_index = 0;
         } else if (rr->role == reconciliator) {
                 ctx->reconciliator_index = 0;
                 // Copy information about all the other members which had the same term
                 for (i=0; i < rr->num; i++) {
                         for (j=0; j < ctx->replica_group_size; j++) {
-                                //if (!strcmp(rr->info[i].name, ctx->workers[j].name)) {
-                                if (strstr(ctx->workers[j].name, rr->info[i].name)) {
+                                if (!strcmp(rr->info[i].name, ctx->workers[j].name)) {
+                                //if (strstr(ctx->workers[j].name, rr->info[i].name)) {
                                         nsr_driver_log(this->name, GF_LOG_INFO,
                                                        "nsr_recon_driver_set_role: this as reconciliator. found other server %s\n",
                                                        ctx->workers[j].name);
@@ -1104,8 +1136,8 @@ nsr_recon_driver_set_role(nsr_recon_driver_ctx_t *ctx,
         } else if (rr->role == resolutor) {
                 for (j=0; j < ctx->replica_group_size; j++) {
                         // info[1] has the information regarding the reconciliator
-                        if (strstr(ctx->workers[j].name, rr->info[1].name)) {
-                        //if (!strcmp(rr->info[1].name, ctx->workers[j].name)) {
+                        //if (strstr(ctx->workers[j].name, rr->info[1].name)) {
+                        if (!strcmp(rr->info[1].name, ctx->workers[j].name)) {
                                 nsr_driver_log(this->name, GF_LOG_INFO,
                                                "nsr_recon_driver_set_role: this as resolutor. found other server %s as reconciliator\n",
                                                ctx->workers[1].name);
@@ -1143,11 +1175,9 @@ nsr_recon_driver_set_role(nsr_recon_driver_ctx_t *ctx,
                  nsr_recon_in_use(ctx, 0, _gf_true);
         }
 
-        ctx->txn_id = txn_id;
-        // Signal the main driver thread
-        pthread_cond_signal(&(ctx->cv));
-        pthread_mutex_unlock(&(ctx->mutex));
-        return _gf_true;
+        ctx->txn_id = rw->txn_id;
+
+        return rr->role;
 }
 
 
@@ -1171,6 +1201,15 @@ compute_resolution_work(nsr_recon_driver_ctx_t *ctx,
         uint32_t i=0;
         uint32_t num = (my_info->last_index - my_info->first_index + 1);
 
+        if (invalidate) {
+                if (my_info->records) {
+                        GF_FREE(my_info->records);
+                        my_info->records = GF_CALLOC(num,
+                                                     sizeof(nsr_reconciliation_record_t),
+                                                     gf_mt_recon_private_t);
+                }
+        }
+
         for (i=0; i < num; i++) {
                 nsr_log_type_t orig, new;
                 nsr_recon_work_type_t tw = NSR_RECON_WORK_NONE;
@@ -1841,7 +1880,6 @@ read_out:
                                         }
                                         apply_record(ctx, ri, dict);
 commit_out:
-                                        dict_unref (dict);
                                         nsr_worker_log(this->name, GF_LOG_INFO,
                                                         "finished recon commit for gfid %s \n",
                                                         rd->gfid);
@@ -2107,6 +2145,10 @@ nsr_recon_in_use(nsr_recon_driver_ctx_t *ctx,
                 send_and_wait(bm, ctx->replica_group_size, ctx,
                       (in_use == _gf_true) ? NSR_WORK_ID_INI : NSR_WORK_ID_FINI,
                       NSR_RECON_QUEUE_TO_DATA, -1);
+                if (in_use == _gf_false) {
+                        //GF_FREE(ctx->workers[i].recon_info->records);
+                        GF_FREE(ctx->workers[i].recon_info);
+                }
 	}
 #endif
 }
@@ -2139,6 +2181,7 @@ nsr_reconciliation_driver(void *arg)
                 nsr_driver_log (this->name, GF_LOG_ERROR, "mutex init error \n");
                 return NULL;
         }
+        INIT_LIST_HEAD(&(ctx->role_head.list));
 
         ctx->workers = GF_CALLOC (replica_group_size,
                                   sizeof(nsr_replica_worker_t),
@@ -2188,24 +2231,20 @@ nsr_reconciliation_driver(void *arg)
 
         while (1) {
 
-                nsr_driver_log (this->name, GF_LOG_INFO, "waiting for state change \n");
-                pthread_mutex_lock(&(ctx->mutex));
-                while ((*driver_ctx)->state == 0) {
-                        pthread_cond_wait(&(ctx->cv), &(ctx->mutex));
-                }
-                pthread_mutex_unlock(&(ctx->mutex));
+            nsr_role_work_t *rr;
 
-                nsr_driver_log (this->name, GF_LOG_INFO, " state changed to %d \n", ctx->state);
-#if 0
-		for (i=0; i < replica_group_size; i++) {
-			if (ctx->workers[i].in_use) {
-				nsr_recon_start_work(ctx->workers[i].control_worker, _gf_true);
-				nsr_recon_start_work(ctx->workers[i].data_worker, _gf_false);
-			}
-		}
-#endif
+            nsr_driver_log (this->name, GF_LOG_INFO, "waiting for role to be queued \n");
+            pthread_mutex_lock(&(ctx->mutex));
+            while (list_empty(&(ctx->role_head.list))) {
+                  pthread_cond_wait(&(ctx->cv), &(ctx->mutex));
+            }
+            pthread_mutex_unlock(&(ctx->mutex));
+
+            list_for_each_entry(rr, &(ctx->role_head.list), list) {
+                nsr_recon_driver_state_t state;
+                state = nsr_recon_driver_get_role(ctx, rr);
 
-                if (ctx->state == leader) {
+                if (state == leader) {
 
                         int32_t chosen = -1;
                         int32_t last_term = -1, last_ops = -1;
@@ -2278,13 +2317,13 @@ nsr_reconciliation_driver(void *arg)
                                  * file almost unreadable.
                                  */
 				if (!setjmp(*(ctx->env))) {
-					ctx->state = reconciliator;
+					state = reconciliator;
 					goto i_am_reconciliator;
 				} else {
                                         nsr_driver_log (this->name, GF_LOG_INFO, "long jmp return to leader\n");
 					free(ctx->env);
 					ctx->env = NULL;
-					ctx->state = leader;
+					state = leader;
 				}
 			}
 
@@ -2294,13 +2333,13 @@ nsr_reconciliation_driver(void *arg)
                                 nsr_driver_log (this->name, GF_LOG_INFO, "local node resolution needs to be done. before set jmp\n");
 				ctx->env = calloc(1,sizeof(jmp_buf));
 				if (!setjmp(*(ctx->env))) {
-					ctx->state = resolutor;
+					state = resolutor;
 					goto i_am_resolutor;
 				} else {
                                         nsr_driver_log (this->name, GF_LOG_INFO, "long jmp return to leader\n");
 					free(ctx->env);
 					ctx->env = NULL;
-					ctx->state = leader;
+					state = leader;
 				}
                         }
 
@@ -2316,7 +2355,7 @@ nsr_reconciliation_driver(void *arg)
 
                 }
 i_am_reconciliator:
-		if (ctx->state == reconciliator) {
+		if (state == reconciliator) {
                         gf_boolean_t do_recon = _gf_false;
                         uint32_t start_index = ctx->workers[0].recon_info->first_index;
                         uint32_t end_index = ctx->workers[0].recon_info->last_index;
@@ -2331,7 +2370,7 @@ i_am_reconciliator:
                                    (ctx->workers[0].recon_info->last_term == ctx->workers[i].recon_info->last_term)) {
                                         ctx->workers[i].recon_info->last_index = end_index;
                                         ctx->workers[i].recon_info->first_index = start_index;
-                                        bm = (1 << i);
+                                        bm |= (1 << i);
                                         do_recon = _gf_true;
                                 }
                         }
@@ -2445,7 +2484,7 @@ i_am_reconciliator:
 
                 }
 i_am_resolutor:
-		if (ctx->state == resolutor) {
+		if (state == resolutor) {
 
                         // This node's last term is filled when it gets a message
                         // from the leader to act as a reconciliator.
@@ -2600,6 +2639,36 @@ i_am_resolutor:
 
                 }
 
+                if (state == joiner) {
+
+                        int32_t chosen = -1;
+                        int32_t last_term = -1, last_ops = -1;
+
+                        nsr_driver_log (this->name, GF_LOG_INFO, "getting last term info from all members of this group\n");
+                        // Get last term info from all members for this group
+                        // which will be the leader(this node) and the node that wants to join.
+                        send_and_wait(-1,
+                                      replica_group_size,
+                                      ctx,
+                                      NSR_WORK_ID_GET_LAST_TERM_INFO,
+                                      NSR_RECON_QUEUE_TO_CONTROL, ctx->current_term);
+
+
+                        // send message to other node that just joined to sync up with this node which is also the leader
+                        nsr_driver_log (this->name, GF_LOG_INFO, "sending resolution work to all nodes except this\n");
+                        bm = ~(1);
+                        send_and_wait(bm,
+                                      replica_group_size,
+                                      ctx,
+                                      NSR_WORK_ID_RESOLUTION_DO_WORK,
+                                      NSR_RECON_QUEUE_TO_CONTROL, -1);
+
+                        nsr_driver_log (this->name, GF_LOG_INFO,
+                                        "finished recon work as joiner \n");
+
+                }
+
+
                 // free the asasociated recon_info contexts created as part of this role
 
 out:
@@ -2617,7 +2686,8 @@ out:
 #endif
                 nsr_driver_log (this->name, GF_LOG_INFO,
                                "finished sending end of reconciliation message \n");
-		ctx->state = 0;
+           }
+           list_del_init (&rr->list);
         }
 
         return NULL;
diff --git a/xlators/cluster/nsr-recon/src/recon_driver.h b/xlators/cluster/nsr-recon/src/recon_driver.h
index 67f4d6014..a9a9a9182 100644
--- a/xlators/cluster/nsr-recon/src/recon_driver.h
+++ b/xlators/cluster/nsr-recon/src/recon_driver.h
@@ -78,6 +78,7 @@ typedef enum nsr_recon_driver_state_t {
         leader = 1,
         reconciliator = 2,
         resolutor = 3,
+        joiner = 4,
 } nsr_recon_driver_state_t;
 
 // role structure
@@ -177,6 +178,12 @@ typedef struct nsr_recon_record_details_s {
         rd.len = f(rd.len); \
 }
 
+typedef struct _nsr_role_work_s {
+        nsr_recon_role_t role;
+        uint32_t txn_id;
+        struct list_head list;
+} nsr_role_work_t;
+        
 typedef struct _nsr_recon_work_s {
 	gf_boolean_t in_use;
         uint32_t index;
@@ -206,7 +213,8 @@ typedef struct _nsr_reconciliator_info {
         int32_t commited_ops;
         uint32_t last_index;
         uint32_t first_index;
-        nsr_reconciliation_record_t records[MAX_RECONCILIATION_WINDOW_SIZE];
+        //nsr_reconciliation_record_t records[MAX_RECONCILIATION_WINDOW_SIZE];
+        nsr_reconciliation_record_t *records;
 } nsr_reconciliator_info_t;
 
 typedef struct _nsr_per_node_worker_s {
@@ -221,7 +229,7 @@ typedef struct _nsr_per_node_worker_s {
        char local; // local data worker
        //struct list_head  list; //list of work items
        nsr_recon_work_t head;
-       pthread_mutex_t mutex; //mutex to gaurd the above list
+       pthread_mutex_t mutex; //mutex to guard the state
        pthread_cond_t cv; //condition variable for signaling the worker thread
        gf_boolean_t is_control;
 #ifdef  NSR_DEBUG
@@ -242,9 +250,9 @@ typedef struct _nsr_recon_driver_ctxt {
        uint32_t replica_group_size; // number of static members of replica group
        nsr_replica_worker_t *workers; // worker info
        int32_t reconciliator;
-       pthread_mutex_t mutex; //mutex to gaurd the state
-       pthread_cond_t cv; //condition variable for signaling the driver thread
-       uint32_t state; //driver state
+       pthread_mutex_t mutex; 
+       pthread_cond_t cv;
+       nsr_role_work_t role_head;
        volatile int32_t outstanding;
        uint32_t reconciliator_index;
        uint32_t txn_id;
@@ -278,8 +286,20 @@ nsr_recon_driver_set_role(nsr_recon_driver_ctx_t *ctx, nsr_recon_role_t *rr, uin
         { \
                 char c[255]; \
                 if (!ctx->driver_log_fd) { \
+                        char str[255], b[255]; \
+                        char *ptr; \
+                        nsr_recon_private_t *priv = ctx->this->private; \
+                        strcpy(b, priv->replica_group_members[0]); \
+                        ptr = strchr (b, '/'); \
+                        while (ptr) { \
+                                *ptr = '-'; \
+                                ptr = strchr (b, '/'); \
+                        } \
+                        sprintf(str,"/tmp/nsr-logs/%s",b); \
                         mkdir("/tmp/nsr-logs/", 0777);  \
-                        ctx->driver_log_fd = open("/tmp/nsr-logs/nsr-driver-log", O_RDWR|O_CREAT|O_TRUNC); \
+                        mkdir(str, 0777);  \
+                        sprintf(str,"/tmp/nsr-logs/%s/nsr-driver-log",b); \
+                        ctx->driver_log_fd = open(str, O_RDWR|O_CREAT|O_TRUNC); \
                 } \
                 sprintf(c, fmt); \
                 write(ctx->driver_log_fd, c, strlen(c));  \
@@ -293,9 +313,19 @@ nsr_recon_driver_set_role(nsr_recon_driver_ctx_t *ctx, nsr_recon_role_t *rr, uin
         { \
                 char c[255]; \
                 if (!ctx->worker_log_fd) { \
-                        char str[255]; \
-                        sprintf(str,"/tmp/nsr-logs/%s-%d",ctx->is_control? "con" : "data",ctx->index); \
+                        char str[255], b[255]; \
+                        char *ptr; \
+                        nsr_recon_private_t *priv = ctx->driver_ctx->this->private; \
+                        strcpy(b, priv->replica_group_members[0]); \
+                        ptr = strchr (b, '/'); \
+                        while (ptr) { \
+                                *ptr = '-'; \
+                                ptr = strchr (b, '/'); \
+                        } \
+                        sprintf(str,"/tmp/nsr-logs/%s",b); \
                         mkdir("/tmp/nsr-logs/", 0777);  \
+                        mkdir(str, 0777);  \
+                        sprintf(str,"/tmp/nsr-logs/%s/%s-%d",b,ctx->is_control?"con":"data",ctx->index); \
                         ctx->worker_log_fd = open(str, O_RDWR|O_CREAT|O_TRUNC); \
                 } \
                 sprintf(c, fmt); \
diff --git a/xlators/cluster/nsr-recon/src/recon_xlator.c b/xlators/cluster/nsr-recon/src/recon_xlator.c
index 62583d526..5f63f6671 100644
--- a/xlators/cluster/nsr-recon/src/recon_xlator.c
+++ b/xlators/cluster/nsr-recon/src/recon_xlator.c
@@ -196,8 +196,7 @@ void nsr_recon_libchangelog_get_records(xlator_t *this, char *bp, int32_t term,
         // do a mmap; seek into the first and read all records till last.
         // TBD - right now all records are pseudo holes but mark them as fills.
         // TBD - pseudo hole to be implemented when actual fsync gets done on data.
-        char  read_buf[((last - first) + 1) * 128];
-        char *rb = &(read_buf[0]);
+        char *rb = NULL, *orig = NULL;
         char path[PATH_MAX];
         int fd;
         uint32_t index = 0;
@@ -206,11 +205,14 @@ void nsr_recon_libchangelog_get_records(xlator_t *this, char *bp, int32_t term,
                         "libchangelog_get_records called for term %d  index from %d to %d \n",
                         term, first, last );
 
+        orig = rb = GF_CALLOC(128, ((last - first) + 1), gf_mt_recon_private_t);
+
         sprintf(path,"%s/%s%d",bp,"TERM.",term);
         fd = open(path, O_RDONLY);
         if (fd != -1) {
                 char *start = NULL;
                 nsr_recon_record_details_t * rec = (nsr_recon_record_details_t *)buf;
+
                 if (first == 0)
                         lseek(fd, 128, SEEK_SET);
                 else
@@ -407,6 +409,7 @@ finish:
                         rec++;
                 } while(1);
         }
+        GF_FREE(orig);
         close(fd);
 
         recon_main_log (this->name, GF_LOG_INFO,
@@ -484,7 +487,8 @@ nsr_recon_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
                         recon_main_log (this->name, GF_LOG_INFO, "nsr_recon_writev called to set role %d\n", rr.role);
                         if ((rr.role != leader) &&
                             (rr.role != reconciliator) &&
-                            (rr.role != resolutor)) {
+                            (rr.role != resolutor) &&
+                            (rr.role != joiner)) {
                                 recon_main_log (this->name, GF_LOG_ERROR,
                                                 "EIII---nsr_recon_writev cannot set state \n");
                                 STACK_UNWIND_STRICT (writev, frame, -1, op_errno,
@@ -577,7 +581,7 @@ nsr_recon_readv (call_frame_t *frame, xlator_t *this,
         int32_t ret = -1;
         nsr_recon_private_t *priv = NULL;
 
-        iobuf = iobuf_get2 (this->ctx->iobuf_pool, op_ret);
+        iobuf = iobuf_get2 (this->ctx->iobuf_pool, size);
         if (!iobuf) {
                 op_errno = ENOMEM;
                 goto out;
@@ -623,6 +627,7 @@ nsr_recon_readv (call_frame_t *frame, xlator_t *this,
                                         (num * sizeof(nsr_recon_record_details_t)), size);
 
                         GF_ASSERT(size == (num * sizeof(nsr_recon_record_details_t)));
+                        bzero(iobuf->ptr, size);
                         recon_main_log (this->name, GF_LOG_INFO,
                                         "nsr_recon_readv - getting records for term=%d from %d to %d\n",
                                         rfd->term, rfd->first_index, rfd->last_index);
diff --git a/xlators/cluster/nsr-recon/src/recon_xlator.h b/xlators/cluster/nsr-recon/src/recon_xlator.h
index c0f1e2145..168db518b 100644
--- a/xlators/cluster/nsr-recon/src/recon_xlator.h
+++ b/xlators/cluster/nsr-recon/src/recon_xlator.h
@@ -55,11 +55,22 @@ typedef struct _nsr_recon_private_s {
 #ifdef NSR_DEBUG
 #define recon_main_log(dom, levl, fmt...) \
         { \
-                nsr_recon_private_t   *priv   = this->private;  \
                 char c[255]; \
+                nsr_recon_private_t   *priv   = this->private;  \
                 if (!priv->recon_main_log_fd) { \
+                        char str[255], b[255]; \
+                        char *ptr; \
+                        strcpy(b, priv->replica_group_members[0]); \
+                        ptr = strchr (b, '/'); \
+                        while (ptr) { \
+                                *ptr = '-'; \
+                                ptr = strchr (b, '/'); \
+                        } \
+                        sprintf(str,"/tmp/nsr-logs/%s",b); \
                         mkdir("/tmp/nsr-logs/", 0777);  \
-                        priv->recon_main_log_fd = open("/tmp/nsr-logs/recon-main-log", O_RDWR|O_CREAT|O_TRUNC); \
+                        mkdir(str, 0777);  \
+                        sprintf(str,"/tmp/nsr-logs/%s/recon-main-log",b); \
+                        priv->recon_main_log_fd = open(str, O_RDWR|O_CREAT|O_TRUNC); \
                 } \
                 sprintf(c, fmt); \
                 write(priv->recon_main_log_fd, c, strlen(c));  \
@@ -68,7 +79,6 @@ typedef struct _nsr_recon_private_s {
 #define recon_main_log(dom, levl, fmt...) gf_log(dom, levl, fmt)
 #endif
 
-
 void nsr_recon_libchangelog_get_this_term_info(xlator_t *this, char *bp, int32_t term, nsr_recon_last_term_info_t *lt);
 void nsr_recon_libchangelog_get_last_term_info(xlator_t *this, char *bp, int32_t term, nsr_recon_last_term_info_t *lt);
 void nsr_recon_return_back(nsr_recon_private_t *priv, uint32_t term_id);