ec: Implement ipc fop

The ipc will be wound to all the bricks, but for it to be
successfull, the fop should succeed on minimum number of bricks.

Change-Id: I3f8cb6a349e87bafd0773583def9d4e3765aa140
BUG: 1211863
Signed-off-by: Poornima G <pgurusid@redhat.com>
Reviewed-on: http://review.gluster.org/15387
NetBSD-regression: NetBSD Build System <jenkins@build.gluster.org>
Smoke: Gluster Build System <jenkins@build.gluster.org>
Reviewed-by: Ashish Pandey <aspandey@redhat.com>
CentOS-regression: Gluster Build System <jenkins@build.gluster.org>
Reviewed-by: Pranith Kumar Karampuri <pkarampu@redhat.com>

4 files changed, 150 insertions, 1 deletions

diff --git a/xlators/cluster/ec/src/ec-fops.h b/xlators/cluster/ec/src/ec-fops.h
index 13f419b5a30..395641cde08 100644
--- a/xlators/cluster/ec/src/ec-fops.h
+++ b/xlators/cluster/ec/src/ec-fops.h
@@ -199,4 +199,8 @@ void ec_seek(call_frame_t *frame, xlator_t *this, uintptr_t target,
              int32_t minimum, fop_seek_cbk_t func, void *data, fd_t *fd,
              off_t offset, gf_seek_what_t what, dict_t *xdata);
 
+void ec_ipc(call_frame_t *frame, xlator_t *this, uintptr_t target,
+            int32_t minimum, fop_ipc_cbk_t func, void *data, int32_t op,
+            dict_t *xdata);
+
 #endif /* __EC_FOPS_H__ */
diff --git a/xlators/cluster/ec/src/ec-generic.c b/xlators/cluster/ec/src/ec-generic.c
index d67420469a8..3ce3c2ab02a 100644
--- a/xlators/cluster/ec/src/ec-generic.c
+++ b/xlators/cluster/ec/src/ec-generic.c
@@ -1447,3 +1447,139 @@ out:
         func(frame, NULL, this, -1, error, NULL, NULL);
     }
 }
+
+/* FOP: IPC */
+
+int32_t ec_ipc_cbk(call_frame_t * frame, void * cookie, xlator_t * this,
+                   int32_t op_ret, int32_t op_errno, dict_t * xdata)
+{
+    ec_fop_data_t * fop = NULL;
+    ec_cbk_data_t * cbk = NULL;
+    int32_t idx = (int32_t)(uintptr_t)cookie;
+
+    VALIDATE_OR_GOTO(this, out);
+    GF_VALIDATE_OR_GOTO(this->name, frame, out);
+    GF_VALIDATE_OR_GOTO(this->name, frame->local, out);
+    GF_VALIDATE_OR_GOTO(this->name, this->private, out);
+
+    fop = frame->local;
+
+    ec_trace("CBK", fop, "idx=%d, frame=%p, op_ret=%d, op_errno=%d", idx,
+             frame, op_ret, op_errno);
+
+    cbk = ec_cbk_data_allocate(frame, this, fop, GF_FOP_IPC, idx, op_ret,
+                               op_errno);
+
+    if (cbk != NULL)
+    {
+        if (xdata != NULL)
+        {
+            cbk->xdata = dict_ref(xdata);
+        }
+
+        ec_combine(cbk, NULL);
+    }
+
+out:
+    if (fop != NULL)
+    {
+        ec_complete(fop);
+    }
+
+    return 0;
+}
+
+void ec_wind_ipc(ec_t * ec, ec_fop_data_t * fop, int32_t idx)
+{
+    ec_trace("WIND", fop, "idx=%d", idx);
+
+    STACK_WIND_COOKIE(fop->frame, ec_ipc_cbk, (void *)(uintptr_t)idx,
+                      ec->xl_list[idx], ec->xl_list[idx]->fops->ipc,
+                      fop->int32, fop->xdata);
+}
+
+int32_t ec_manager_ipc(ec_fop_data_t *fop, int32_t state)
+{
+    ec_cbk_data_t * cbk;
+
+    switch (state)
+    {
+        case EC_STATE_INIT:
+        case EC_STATE_DISPATCH:
+            ec_dispatch_all(fop);
+
+            return EC_STATE_PREPARE_ANSWER;
+
+        case EC_STATE_PREPARE_ANSWER:
+            ec_fop_prepare_answer(fop, _gf_true);
+
+            return EC_STATE_REPORT;
+
+        case EC_STATE_REPORT:
+            cbk = fop->answer;
+
+            GF_ASSERT(cbk != NULL);
+            if (fop->cbks.ipc != NULL)
+            {
+                fop->cbks.ipc(fop->req_frame, fop, fop->xl, cbk->op_ret,
+                               cbk->op_errno, cbk->xdata);
+            }
+
+            return EC_STATE_END;
+
+        case -EC_STATE_INIT:
+        case -EC_STATE_DISPATCH:
+        case -EC_STATE_PREPARE_ANSWER:
+        case -EC_STATE_REPORT:
+            GF_ASSERT(fop->error != 0);
+
+            if (fop->cbks.ipc != NULL)
+            {
+                fop->cbks.ipc(fop->req_frame, fop, fop->xl, -1, fop->error,
+                              NULL);
+            }
+
+            return EC_STATE_END;
+
+        default:
+            gf_msg (fop->xl->name, GF_LOG_ERROR, EINVAL,
+                    EC_MSG_UNHANDLED_STATE, "Unhandled state %d for %s",
+                    state, ec_fop_name(fop->id));
+
+            return EC_STATE_END;
+    }
+}
+
+void ec_ipc(call_frame_t *frame, xlator_t *this, uintptr_t target,
+            int32_t minimum, fop_ipc_cbk_t func, void *data, int32_t op,
+            dict_t *xdata)
+{
+    ec_cbk_t callback = { .ipc = func };
+    ec_fop_data_t * fop = NULL;
+    int32_t error = ENOMEM;
+
+    gf_msg_trace ("ec", 0, "EC(IPC) %p", frame);
+
+    VALIDATE_OR_GOTO(this, out);
+    GF_VALIDATE_OR_GOTO(this->name, frame, out);
+    GF_VALIDATE_OR_GOTO(this->name, this->private, out);
+
+    fop = ec_fop_data_allocate(frame, this, GF_FOP_IPC, 0, target, minimum,
+                               ec_wind_ipc, ec_manager_ipc, callback, data);
+    if (fop == NULL) {
+        goto out;
+    }
+    if (xdata != NULL) {
+        fop->xdata = dict_ref(xdata);
+    }
+    fop->int32 = op;
+
+    error = 0;
+
+out:
+    if (fop != NULL) {
+        ec_manager(fop, error);
+    } else {
+        func(frame, NULL, this, -1, error, NULL);
+    }
+}
diff --git a/xlators/cluster/ec/src/ec-types.h b/xlators/cluster/ec/src/ec-types.h
index 5249ca980c2..de13b2562f1 100644
--- a/xlators/cluster/ec/src/ec-types.h
+++ b/xlators/cluster/ec/src/ec-types.h
@@ -201,6 +201,7 @@ union _ec_cbk {
     fop_fxattrop_cbk_t     fxattrop;
     fop_zerofill_cbk_t     zerofill;
     fop_seek_cbk_t         seek;
+    fop_ipc_cbk_t          ipc;
 };
 
 struct _ec_lock {
diff --git a/xlators/cluster/ec/src/ec.c b/xlators/cluster/ec/src/ec.c
index 2aff4374b82..dff5a784b2b 100644
--- a/xlators/cluster/ec/src/ec.c
+++ b/xlators/cluster/ec/src/ec.c
@@ -1212,6 +1212,13 @@ int32_t ec_gf_seek(call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
     return 0;
 }
 
+int32_t ec_gf_ipc(call_frame_t *frame, xlator_t *this, int32_t op,
+                  dict_t *xdata)
+{
+    ec_ipc(frame, this, -1, EC_MINIMUM_MIN, default_ipc_cbk, NULL, op, xdata);
+    return 0;
+}
+
 int32_t ec_gf_forget(xlator_t * this, inode_t * inode)
 {
     uint64_t value = 0;
@@ -1327,7 +1334,8 @@ struct xlator_fops fops =
     .fallocate    = ec_gf_fallocate,
     .discard      = ec_gf_discard,
     .zerofill     = ec_gf_zerofill,
-    .seek         = ec_gf_seek
+    .seek         = ec_gf_seek,
+    .ipc          = ec_gf_ipc
 };
 
 struct xlator_cbks cbks =