1 files changed, 131 insertions, 0 deletions
diff --git a/xlators/features/changelog/src/changelog-barrier.c b/xlators/features/changelog/src/changelog-barrier.c
new file mode 100644
index 00000000000..0fb89ddb127
--- /dev/null
+++ b/xlators/features/changelog/src/changelog-barrier.c
@@ -0,0 +1,131 @@
+/*
+     Copyright (c) 2014 Red Hat, Inc. <http://www.redhat.com>
+     This file is part of GlusterFS.
+
+     This file is licensed to you under your choice of the GNU Lesser
+     General Public License, version 3 or any later version (LGPLv3 or
+     later), or the GNU General Public License, version 2 (GPLv2), in all
+     cases as published by the Free Software Foundation.
+*/
+
+#include "changelog-helpers.h"
+#include "changelog-messages.h"
+#include <glusterfs/call-stub.h>
+
+/* Enqueue a stub*/
+void
+__chlog_barrier_enqueue(xlator_t *this, call_stub_t *stub)
+{
+    changelog_priv_t *priv = NULL;
+
+    priv = this->private;
+    GF_ASSERT(priv);
+
+    list_add_tail(&stub->list, &priv->queue);
+    priv->queue_size++;
+
+    return;
+}
+
+/* Dequeue a stub */
+call_stub_t *
+__chlog_barrier_dequeue(xlator_t *this, struct list_head *queue)
+{
+    call_stub_t *stub = NULL;
+    changelog_priv_t *priv = NULL;
+
+    priv = this->private;
+    GF_ASSERT(priv);
+
+    if (list_empty(queue))
+        goto out;
+
+    stub = list_entry(queue->next, call_stub_t, list);
+    list_del_init(&stub->list);
+
+out:
+    return stub;
+}
+
+/* Dequeue all the stubs and call corresponding resume functions */
+void
+chlog_barrier_dequeue_all(xlator_t *this, struct list_head *queue)
+{
+    call_stub_t *stub = NULL;
+
+    gf_smsg(this->name, GF_LOG_INFO, 0, CHANGELOG_MSG_DEQUEUING_BARRIER_FOPS,
+            NULL);
+
+    while ((stub = __chlog_barrier_dequeue(this, queue)))
+        call_resume(stub);
+
+    gf_smsg(this->name, GF_LOG_INFO, 0,
+            CHANGELOG_MSG_DEQUEUING_BARRIER_FOPS_FINISHED, NULL);
+    return;
+}
+
+/* Function called on changelog barrier timeout */
+void
+chlog_barrier_timeout(void *data)
+{
+    xlator_t *this = NULL;
+    changelog_priv_t *priv = NULL;
+    struct list_head queue = {
+        0,
+    };
+
+    this = data;
+    THIS = this;
+    priv = this->private;
+
+    INIT_LIST_HEAD(&queue);
+
+    gf_smsg(this->name, GF_LOG_ERROR, 0, CHANGELOG_MSG_BARRIER_TIMEOUT, NULL);
+
+    LOCK(&priv->lock);
+    {
+        __chlog_barrier_disable(this, &queue);
+    }
+    UNLOCK(&priv->lock);
+
+    chlog_barrier_dequeue_all(this, &queue);
+
+    return;
+}
+
+/* Disable changelog barrier enable flag */
+void
+__chlog_barrier_disable(xlator_t *this, struct list_head *queue)
+{
+    changelog_priv_t *priv = this->private;
+    GF_ASSERT(priv);
+
+    if (priv->timer) {
+        gf_timer_call_cancel(this->ctx, priv->timer);
+        priv->timer = NULL;
+    }
+
+    list_splice_init(&priv->queue, queue);
+    priv->queue_size = 0;
+    priv->barrier_enabled = _gf_false;
+}
+
+/* Enable chagelog barrier enable with timer */
+int
+__chlog_barrier_enable(xlator_t *this, changelog_priv_t *priv)
+{
+    int ret = -1;
+
+    priv->timer = gf_timer_call_after(this->ctx, priv->timeout,
+                                      chlog_barrier_timeout, (void *)this);
+    if (!priv->timer) {
+        gf_smsg(this->name, GF_LOG_CRITICAL, 0,
+                CHANGELOG_MSG_TIMEOUT_ADD_FAILED, NULL);
+        goto out;
+    }
+
+    priv->barrier_enabled = _gf_true;
+    ret = 0;
+out:
+    return ret;
+}