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>

1 files changed, 222 insertions, 0 deletions

diff --git a/xlators/cluster/nsr-server/src/etcd-sim.c b/xlators/cluster/nsr-server/src/etcd-sim.c
new file mode 100644
index 000000000..5c5cdcec0
--- /dev/null
+++ b/xlators/cluster/nsr-server/src/etcd-sim.c
@@ -0,0 +1,222 @@
+/*
+ * Copyright (c) 2013, Red Hat
+ * All rights reserved.
+
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+
+ * Redistributions of source code must retain the above copyright notice, this
+ * list of conditions and the following disclaimer.  Redistributions in binary
+ * form must reproduce the above copyright notice, this list of conditions and
+ * the following disclaimer in the documentation and/or other materials
+ * provided with the distribution.
+
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+
+#include "call-stub.h"
+#include "defaults.h"
+#include "xlator.h"
+#include "api/src/glfs.h"
+#include "api/src/glfs-internal.h"
+#include "run.h"
+
+
+/*
+ * Mock implementation of etcd
+ * The etcd file is simulated in /tmp/<server-names>
+ * Writes from Multiple writers are protected using file lock.
+*/
+
+#include "etcd-api.h"
+#define T_FORMAT "%d-%b-%Y,%H:%M:%S"
+#define MAX_KEY_LEN 64
+#define MAX_VALUE_LEN 64
+#define MAX_TTL_LEN 12
+
+etcd_session
+etcd_open (etcd_server *server_list)
+{
+        return NULL;
+}
+
+typedef struct _etcd_sim_s {
+        int fd;
+        FILE *stream;
+} etcd_sim_t;
+
+void
+etcd_close (etcd_session this)
+{
+        etcd_sim_t *sim = (etcd_sim_t *)this;
+        fflush(sim->stream);
+        fclose(sim->stream);
+        close(sim->fd);
+        free(this);
+}
+
+
+char *
+etcd_get (etcd_session this, char *key)
+{
+        char *str = NULL;
+        size_t len;
+        etcd_sim_t *sim = (etcd_sim_t *)this;
+        struct tm tm;
+        time_t old, new;
+        lockf(sim->fd, F_LOCK, 0 );
+        if (fseek(sim->stream, 0, SEEK_SET) == -1) {
+                lockf(sim->fd, F_ULOCK, 0 );
+                return NULL;
+        }
+        // Read the file
+        while(1) {
+                if(str) {
+                        free(str);
+                        str = NULL;
+                }
+                if (getline((char **)&str, &len,sim->stream) == -1) {
+                        break;
+                }
+                if (!strncmp(str, key, strlen(key))) {
+                       char k[256], s[256], *ret, past[256];
+                       unsigned int ttl;
+                       double delta;
+                       sscanf(str,"%s %s %d %s",k, s, &ttl, past);
+                       strptime(past, T_FORMAT, &tm);
+                       old = mktime(&tm);
+                       new = time(NULL);
+                       delta = difftime(new, old);
+                       // check if key is expired.
+                       // If ttl is 0, it means key has infinite ttk=l.
+                       if ((!ttl) || ((delta >= 0) && (delta < ttl))) {
+                               ret = calloc(1, strlen(s) + 1);
+                               strcpy(ret,s);
+                               free(str);
+                               lockf(sim->fd, F_ULOCK, 0 );
+                               return(ret);
+                       }
+                }
+        }
+        lockf(sim->fd, F_ULOCK, 0 );
+        return NULL;
+}
+
+
+etcd_result
+etcd_set (etcd_session this, char *key, char *value,
+          char *precond, unsigned int ttl)
+{
+        char *str = NULL;
+        char buf[255];
+        char tp[255];
+        char s[255];
+        size_t len;
+        etcd_sim_t *sim = (etcd_sim_t *)this;
+        struct tm tm;
+        time_t old, new;
+        lockf(sim->fd, F_LOCK, 0 );
+        if (fseek(sim->stream, 0, SEEK_SET) == -1) {
+                lockf(sim->fd, F_ULOCK, 0 );
+                return ETCD_WTF;
+        }
+        while(1) {
+                if(str) {
+                        free(str);
+                        str = NULL;
+                }
+                if (getline((char **)&str, &len,sim->stream) == -1) {
+                        break;
+                }
+                if (!strncmp(str, key, strlen(key))) {
+                        char k[256], s[256], past[256];
+                        unsigned int ttl;
+                        double delta;
+                        sscanf(str,"%s %s %d %s",k, s, &ttl, past);
+                        strptime(past, T_FORMAT, &tm);
+                        old = mktime(&tm);
+                        new = time(NULL);
+                        delta = difftime(new, old);
+                        // check if the present key is expired
+                        if ( (!ttl) || ((delta >= 0) && (delta < ttl))) {
+                                // present key not expired. In case of precondition,
+                                // check if it matches. If not return with error
+                                // In case of no precond, return error since
+                                // present key not yet expired.
+                                if ((!precond) || (strcmp(precond, s))) {
+                                        free(str);
+                                        lockf(sim->fd, F_ULOCK, 0 );
+                                        return ETCD_WTF;
+                                }
+                        }
+                        fseek(sim->stream, -strlen(str), SEEK_CUR);               
+                        free(str);
+                        goto here;
+                }
+        }
+here:
+        memset(tp, 0, 255);
+        new = time(NULL);
+        memcpy(&tm, localtime(&new), sizeof(struct tm));
+        strftime(buf, sizeof(buf), T_FORMAT, &tm);
+        // what we want to print in the file is something like this
+        // key value(at offset of 64) ttl(offset to 128) time(left offset to 140)
+        // hence we would want to create a format buf as follows:
+        // "%-64s%-64s%-16d%-18s"
+        // Hence we construct this first (in string s) and use that to print into tp
+        // which gets written to the registry file.
+        sprintf(s,"%%-%ds%%-%ds%%-%dd%%s\n",
+                MAX_KEY_LEN, MAX_VALUE_LEN, MAX_TTL_LEN);
+        sprintf(tp,s,key, value, ttl, buf);
+        if (fwrite(tp, 1,strlen(tp), sim->stream) != strlen(tp)) {
+                lockf(sim->fd, F_ULOCK, 0 );
+                return ETCD_WTF;
+        }
+        fflush(sim->stream);
+        lockf(sim->fd, F_ULOCK, 0 );
+        return ETCD_OK;
+}
+
+
+
+etcd_session
+etcd_open_str (char *server_names)
+{
+        etcd_sim_t *sim;
+        char name[256];
+
+        sim = calloc(1, sizeof(etcd_sim_t));
+        sprintf(name, "/tmp/%s", server_names);
+        sim->fd = open(name, O_RDWR | O_CREAT);
+        if (sim->fd == -1)
+                return NULL;
+        sim->stream = fopen(name, "r+");
+        if (sim->stream == NULL) 
+                return NULL;
+
+        return ((void *)sim);
+}
+
+
+void
+etcd_close_str (etcd_session this)
+{
+        etcd_close(this);
+}