features/changelog: Change barrier notification mechanism

Backport of http://review.gluster.org/#/c/14272/

The barrier notification mechanism was fd based and 'select'
was being used. 'select' breaks when number of fds opened
by brick process exceeds 1024. To avoid this and also
the maintainance of pipe between notify and 'changelog_rollover',
the pipe has been replaced with pthread condition signal
and timed wait mechanism.

> Change-Id: I530ea90d9a06953f8b23b4e12d122872ee1925de
> BUG: 1334314
> Signed-off-by: Kotresh HR <khiremat@redhat.com>
> Reviewed-on: http://review.gluster.org/14272
(cherry picked from commit be00012e7be55d25870411f3e975db9a8e19c70a)

Change-Id: I530ea90d9a06953f8b23b4e12d122872ee1925de
BUG: 1342083
Signed-off-by: Kotresh HR <khiremat@redhat.com>
Reviewed-on: http://review.gluster.org/14614
NetBSD-regression: NetBSD Build System <jenkins@build.gluster.org>
CentOS-regression: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Niels de Vos <ndevos@redhat.com>
Smoke: Gluster Build System <jenkins@build.gluster.com>

3 files changed, 69 insertions, 80 deletions

diff --git a/xlators/features/changelog/src/changelog-helpers.c b/xlators/features/changelog/src/changelog-helpers.c
index 6d3b99c6cdc..0cb68587e57 100644
--- a/xlators/features/changelog/src/changelog-helpers.c
+++ b/xlators/features/changelog/src/changelog-helpers.c
@@ -1259,15 +1259,10 @@ changelog_rollover (void *data)
 {
         int                     ret             = 0;
         xlator_t               *this            = NULL;
-        struct timeval          tv              = {0,};
+        struct timespec         tv              = {0,};
         changelog_log_data_t    cld             = {0,};
         changelog_time_slice_t *slice           = NULL;
         changelog_priv_t       *priv            = data;
-        int                     max_fd          = 0;
-        char                    buf[1]          = {0};
-        int                     len             = 0;
-
-        fd_set                  rset;
 
         this = priv->cr.this;
         slice = &priv->slice;
@@ -1275,48 +1270,42 @@ changelog_rollover (void *data)
         while (1) {
                 (void) pthread_testcancel();
 
-                tv.tv_sec  = priv->rollover_time;
-                tv.tv_usec = 0;
-                FD_ZERO(&rset);
-                FD_SET(priv->cr.rfd, &rset);
-                max_fd = priv->cr.rfd;
-                max_fd = max_fd + 1;
-
-               /* It seems there is a race between actual rollover and explicit
-                * rollover. But it is handled. If actual rollover is being
-                * done and the explicit rollover event comes, the event is
-                * not missed. The next select will immediately wakeup to
-                * handle explicit wakeup.
+                tv.tv_sec  = time (NULL) + priv->rollover_time;
+                tv.tv_nsec = 0;
+                ret = 0; /* Reset ret to zero */
+
+               /* The race between actual rollover and explicit rollover is
+                * handled. If actual rollover is being done and the
+                * explicit rollover event comes, the event is not missed.
+                * Since explicit rollover sets 'cr.notify' to true, this
+                * thread doesn't wait on 'pthread_cond_timedwait'.
                 */
+                pthread_cleanup_push (changelog_cleanup_free_mutex,
+                                      &priv->cr.lock);
+                pthread_mutex_lock (&priv->cr.lock);
+                {
+                        while (ret == 0 && !priv->cr.notify)
+                                ret = pthread_cond_timedwait (&priv->cr.cond,
+                                                              &priv->cr.lock,
+                                                              &tv);
+                        if (ret == 0)
+                                priv->cr.notify = _gf_false;
+                }
+                pthread_mutex_unlock (&priv->cr.lock);
+                pthread_cleanup_pop (0);
 
-                ret = select (max_fd, &rset, NULL, NULL, &tv);
-                if (ret == -1) {
-                        gf_msg (this->name, GF_LOG_ERROR, errno,
-                                CHANGELOG_MSG_SELECT_FAILED,
-                                "select failed");
-                        continue;
-                } else if (ret && FD_ISSET(priv->cr.rfd, &rset)) {
+                if (ret == 0) {
                         gf_msg (this->name, GF_LOG_INFO, 0,
                                 CHANGELOG_MSG_BARRIER_INFO,
-                                "Explicit wakeup of select on barrier notify");
-                        len = sys_read (priv->cr.rfd, buf, 1);
-                        if (len == 0) {
-                                gf_msg (this->name, GF_LOG_ERROR, errno,
-                                        CHANGELOG_MSG_READ_ERROR, "BUG: Got EOF"
-                                        " from reconfigure notification pipe");
-                                continue;
-                        }
-                        if (len < 0) {
-                                gf_msg (this->name, GF_LOG_ERROR, 0,
-                                        CHANGELOG_MSG_READ_ERROR,
-                                        "Failed to read wakeup data");
-                                continue;
-                        }
-                        /* Lock is not required as same thread is modifying.*/
+                                "Explicit wakeup on barrier notify");
                         priv->explicit_rollover = _gf_true;
-                } else {
-                        gf_msg_debug (this->name, 0,
-                                      "select wokeup on timeout");
+                } else if (ret && ret != ETIMEDOUT) {
+                        gf_msg (this->name, GF_LOG_ERROR, errno,
+                                CHANGELOG_MSG_SELECT_FAILED,
+                                "pthread_cond_timedwait failed");
+                        continue;
+                } else if (ret && ret == ETIMEDOUT) {
+                        gf_msg_debug (this->name, 0, "Wokeup on timeout");
                 }
 
                /* Reading curent_color without lock is fine here
@@ -1783,9 +1772,12 @@ changelog_barrier_notify (changelog_priv_t *priv, char *buf)
 {
         int ret = 0;
 
-        LOCK(&priv->lock);
-                ret = changelog_write (priv->cr_wfd, buf, 1);
-        UNLOCK(&priv->lock);
+        pthread_mutex_lock (&priv->cr.lock);
+        {
+                ret = pthread_cond_signal (&priv->cr.cond);
+                priv->cr.notify = _gf_true;
+        }
+        pthread_mutex_unlock (&priv->cr.lock);
         return ret;
 }
 
diff --git a/xlators/features/changelog/src/changelog-helpers.h b/xlators/features/changelog/src/changelog-helpers.h
index 70e6f60f30c..4fdba244aa1 100644
--- a/xlators/features/changelog/src/changelog-helpers.h
+++ b/xlators/features/changelog/src/changelog-helpers.h
@@ -116,8 +116,9 @@ typedef struct changelog_rollover {
 
         xlator_t *this;
 
-        /* read end of pipe used as event from barrier on snapshot */
-        int rfd;
+        pthread_mutex_t lock;
+        pthread_cond_t cond;
+        gf_boolean_t notify;
 } changelog_rollover_t;
 
 typedef struct changelog_fsync {
@@ -265,9 +266,6 @@ struct changelog_priv {
         /* Represents the active color. Initially by default black */
         chlog_fop_color_t current_color;
 
-        /* write end of pipe to do explicit rollover on barrier during snap */
-        int cr_wfd;
-
         /* flag to determine explicit rollover is triggered */
         gf_boolean_t explicit_rollover;
 
diff --git a/xlators/features/changelog/src/changelog.c b/xlators/features/changelog/src/changelog.c
index cbfc307eb3c..f8f95cf0e81 100644
--- a/xlators/features/changelog/src/changelog.c
+++ b/xlators/features/changelog/src/changelog.c
@@ -2046,11 +2046,6 @@ changelog_cleanup_helper_threads (xlator_t *this, changelog_priv_t *priv)
         if (priv->cr.rollover_th) {
                 (void) changelog_thread_cleanup (this, priv->cr.rollover_th);
                 priv->cr.rollover_th = 0;
-                ret = sys_close (priv->cr_wfd);
-                if (ret)
-                        gf_msg (this->name, GF_LOG_ERROR, errno,
-                                CHANGELOG_MSG_CLOSE_ERROR,
-                                "error closing write end of rollover pipe");
         }
 
         if (priv->cf.fsync_th) {
@@ -2065,7 +2060,6 @@ changelog_spawn_helper_threads (xlator_t *this, changelog_priv_t *priv)
 {
         int ret = 0;
         int flags = 0;
-        int pipe_fd[2] = {0, 0};
 
         /* Geo-Rep snapshot dependency:
          *
@@ -2079,29 +2073,7 @@ changelog_spawn_helper_threads (xlator_t *this, changelog_priv_t *priv)
          * the write end of the pipe in 'reconfigure'.
          */
 
-        ret = pipe (pipe_fd);
-        if (ret == -1) {
-                gf_msg (this->name, GF_LOG_ERROR,
-                        errno, CHANGELOG_MSG_PIPE_CREATION_ERROR,
-                        "Cannot create pipe");
-                goto out;
-        }
-
-        /* writer is non-blocking */
-        flags = fcntl (pipe_fd[1], F_GETFL);
-        flags |= O_NONBLOCK;
-
-        ret = fcntl (pipe_fd[1], F_SETFL, flags);
-        if (ret) {
-                gf_msg (this->name, GF_LOG_ERROR, errno,
-                        CHANGELOG_MSG_FCNTL_FAILED,
-                        "failed to set O_NONBLOCK flag");
-                goto out;
-        }
-
-        priv->cr_wfd = pipe_fd[1];
-        priv->cr.rfd = pipe_fd[0];
-
+        priv->cr.notify = _gf_false;
         priv->cr.this = this;
         ret = gf_thread_create (&priv->cr.rollover_th,
 				NULL, changelog_rollover, priv);
@@ -2419,6 +2391,8 @@ changelog_barrier_pthread_init (xlator_t *this, changelog_priv_t *priv)
         gf_boolean_t    dm_cond_black_init    = _gf_false;
         gf_boolean_t    dm_mutex_white_init   = _gf_false;
         gf_boolean_t    dm_cond_white_init    = _gf_false;
+        gf_boolean_t    cr_mutex_init         = _gf_false;
+        gf_boolean_t    cr_cond_init          = _gf_false;
         int             ret                   = 0;
 
         if ((ret = pthread_mutex_init(&priv->bn.bnotify_mutex, NULL)) != 0) {
@@ -2476,6 +2450,24 @@ changelog_barrier_pthread_init (xlator_t *this, changelog_priv_t *priv)
                 goto out;
         }
         dm_cond_white_init = _gf_true;
+
+        if ((pthread_mutex_init(&priv->cr.lock, NULL)) != 0) {
+                gf_msg (this->name, GF_LOG_ERROR, errno,
+                        CHANGELOG_MSG_PTHREAD_MUTEX_INIT_FAILED,
+                        "changelog_rollover lock init failed (%d)", ret);
+                ret = -1;
+                goto out;
+        }
+        cr_mutex_init = _gf_true;
+
+        if ((pthread_cond_init(&priv->cr.cond, NULL)) != 0) {
+                gf_msg (this->name, GF_LOG_ERROR, errno,
+                        CHANGELOG_MSG_PTHREAD_COND_INIT_FAILED,
+                        "changelog_rollover cond init failed (%d)", ret);
+                ret = -1;
+                goto out;
+        }
+        cr_cond_init = _gf_true;
  out:
         if (ret) {
                 if (bn_mutex_init)
@@ -2490,6 +2482,10 @@ changelog_barrier_pthread_init (xlator_t *this, changelog_priv_t *priv)
                         pthread_mutex_destroy(&priv->dm.drain_white_mutex);
                 if (dm_cond_white_init)
                         pthread_cond_destroy (&priv->dm.drain_white_cond);
+                if (cr_mutex_init)
+                        pthread_mutex_destroy(&priv->cr.lock);
+                if (cr_cond_init)
+                        pthread_cond_destroy (&priv->cr.cond);
         }
         return ret;
 }
@@ -2504,6 +2500,8 @@ changelog_barrier_pthread_destroy (changelog_priv_t *priv)
         pthread_cond_destroy (&priv->dm.drain_black_cond);
         pthread_mutex_destroy (&priv->dm.drain_white_mutex);
         pthread_cond_destroy (&priv->dm.drain_white_cond);
+        pthread_mutex_destroy(&priv->cr.lock);
+        pthread_cond_destroy (&priv->cr.cond);
         LOCK_DESTROY (&priv->bflags.lock);
 }
 
@@ -2834,6 +2832,7 @@ init (xlator_t *this)
         priv->current_color = FOP_COLOR_BLACK;
         priv->explicit_rollover = _gf_false;
 
+        priv->cr.notify = _gf_false;
         /* Mutex is not needed as threads are not spawned yet */
         priv->bn.bnotify = _gf_false;
         priv->bn.bnotify_error = _gf_false;