rio/everywhere: add icreate/namelink fop

icreate creates inode, while namelink links the basename to it's
parent gfid.

For now mkdir is the primary user of these fops. Better distribution is
acheived by creating the inode on ,(say) mds1 and linking the basename to it's
parent gfid on mds2. The inode serves readdirp, stat etc.

More details about the fops are present at:
https://review.gluster.org/#/c/13395/3/design/DHT2/DHT2_Icreate_Namelink_Notes.md

This backport of three patches from experimental branch.
1- https://review.gluster.org/#/c/18085/
2- https://review.gluster.org/#/c/18086/
3- https://review.gluster.org/#/c/18094/

Updates gluster/glusterfs#243
Change-Id: I1bd3d5a441a3cfab1acfeb52f15c6c867d362592
Signed-off-by: Susant Palai <spalai@redhat.com>

3 files changed, 488 insertions, 0 deletions

diff --git a/xlators/protocol/client/src/client-rpc-fops.c b/xlators/protocol/client/src/client-rpc-fops.c
index 9f0a752ccf6..30e0699d446 100644
--- a/xlators/protocol/client/src/client-rpc-fops.c
+++ b/xlators/protocol/client/src/client-rpc-fops.c
@@ -3150,6 +3150,102 @@ out:
         return 0;
 }
 
+int32_t
+client3_3_namelink_cbk (struct rpc_req *req,
+                        struct iovec *iov, int count, void *myframe)
+{
+        int32_t            ret     = 0;
+        xlator_t          *this    = NULL;
+        struct iatt        prebuf  = {0,};
+        struct iatt        postbuf = {0,};
+        dict_t            *xdata   = NULL;
+        call_frame_t      *frame   = NULL;
+        gfs4_namelink_rsp  rsp     = {0,};
+
+        this = THIS;
+        frame = myframe;
+
+        if (req->rpc_status == -1) {
+                rsp.op_ret = -1;
+                rsp.op_errno = ENOTCONN;
+                goto out;
+        }
+
+        ret = xdr_to_generic (*iov, &rsp, (xdrproc_t)xdr_gfs4_namelink_rsp);
+        if (ret < 0) {
+                rsp.op_ret   = -1;
+                rsp.op_errno = EINVAL;
+                goto out;
+        }
+
+        if (rsp.op_ret != -1) {
+                gf_stat_to_iatt (&rsp.preparent, &prebuf);
+                gf_stat_to_iatt (&rsp.postparent, &postbuf);
+        }
+
+        GF_PROTOCOL_DICT_UNSERIALIZE (this, xdata, (rsp.xdata.xdata_val),
+                                      (rsp.xdata.xdata_len), ret,
+                                      rsp.op_errno, out);
+
+ out:
+        CLIENT_STACK_UNWIND (namelink, frame, rsp.op_ret,
+                             gf_error_to_errno (rsp.op_errno),
+                             &prebuf, &postbuf, xdata);
+        free (rsp.xdata.xdata_val);
+        if (xdata)
+                dict_unref (xdata);
+        return 0;
+}
+
+int32_t
+client3_3_icreate_cbk (struct rpc_req *req,
+                       struct iovec *iov, int count, void *myframe)
+{
+        int32_t           ret      = 0;
+        inode_t          *inode    = NULL;
+        clnt_local_t     *local    = NULL;
+        xlator_t         *this     = NULL;
+        struct iatt       stbuf    = {0,};
+        dict_t           *xdata    = NULL;
+        call_frame_t     *frame    = NULL;
+        gfs4_icreate_rsp  rsp      = {0,};
+
+        this = THIS;
+        frame = myframe;
+        local = frame->local;
+
+        inode = local->loc.inode;
+
+        if (req->rpc_status == -1) {
+                rsp.op_ret = -1;
+                rsp.op_errno = ENOTCONN;
+                goto out;
+        }
+
+        ret = xdr_to_generic (*iov, &rsp, (xdrproc_t)xdr_gfs4_icreate_rsp);
+        if (ret < 0) {
+                rsp.op_ret   = -1;
+                rsp.op_errno = EINVAL;
+                goto out;
+        }
+
+        if (rsp.op_ret != -1)
+                gf_stat_to_iatt (&rsp.stat, &stbuf);
+
+        GF_PROTOCOL_DICT_UNSERIALIZE (this, xdata, (rsp.xdata.xdata_val),
+                                      (rsp.xdata.xdata_len), ret,
+                                      rsp.op_errno, out);
+
+ out:
+        CLIENT_STACK_UNWIND (icreate, frame, rsp.op_ret,
+                             gf_error_to_errno (rsp.op_errno),
+                             inode, &stbuf, xdata);
+        free (rsp.xdata.xdata_val);
+        if (xdata)
+                dict_unref (xdata);
+        return 0;
+}
+
 int
 client_fdctx_destroy (xlator_t *this, clnt_fd_ctx_t *fdctx)
 {
@@ -6250,6 +6346,104 @@ unwind:
 }
 
 int32_t
+client3_3_namelink (call_frame_t *frame, xlator_t *this, void *data)
+{
+        int32_t ret = 0;
+        int32_t op_errno = EINVAL;
+        clnt_conf_t *conf = NULL;
+        clnt_args_t *args = NULL;
+        gfs4_namelink_req req = {{0,},};
+
+        GF_ASSERT (frame);
+
+        args = data;
+        conf = this->private;
+
+        if (!(args->loc && args->loc->parent))
+                goto unwind;
+
+        if (!gf_uuid_is_null (args->loc->parent->gfid))
+                memcpy (req.pargfid, args->loc->parent->gfid, sizeof (uuid_t));
+        else
+                memcpy (req.pargfid, args->loc->pargfid, sizeof (uuid_t));
+
+        GF_ASSERT_AND_GOTO_WITH_ERROR (this->name,
+                                       !gf_uuid_is_null (*((uuid_t *)req.pargfid)),
+                                       unwind, op_errno, EINVAL);
+
+        req.bname = (char *)args->loc->name;
+
+        GF_PROTOCOL_DICT_SERIALIZE (this, args->xdata, (&req.xdata.xdata_val),
+                                    req.xdata.xdata_len, op_errno, unwind);
+
+        ret = client_submit_request (this, &req, frame, conf->fops,
+                                     GFS3_OP_NAMELINK, client3_3_namelink_cbk,
+                                     NULL, NULL, 0, NULL, 0, NULL,
+                                     (xdrproc_t)xdr_gfs4_namelink_req);
+        if (ret) {
+                gf_msg (this->name, GF_LOG_WARNING, 0, PC_MSG_FOP_SEND_FAILED,
+                        "failed to send the fop");
+        }
+
+        GF_FREE (req.xdata.xdata_val);
+        return 0;
+
+ unwind:
+        CLIENT_STACK_UNWIND (namelink, frame, -1, op_errno, NULL, NULL, NULL);
+        return 0;
+}
+
+int32_t
+client3_3_icreate (call_frame_t *frame, xlator_t *this, void *data)
+{
+        int32_t ret = 0;
+        int32_t op_errno = EINVAL;
+        clnt_conf_t *conf = NULL;
+        clnt_args_t *args = NULL;
+        clnt_local_t *local = NULL;
+        gfs4_icreate_req req = {{0,},};
+
+        GF_ASSERT (frame);
+
+        args = data;
+        conf = this->private;
+
+        if (!(args->loc && args->loc->inode))
+                goto unwind;
+
+        local = mem_get0 (this->local_pool);
+        if (!local) {
+                op_errno = ENOMEM;
+                goto unwind;
+        }
+        frame->local = local;
+
+        loc_copy (&local->loc, args->loc);
+
+        req.mode = args->mode;
+        memcpy (req.gfid, args->loc->gfid, sizeof (uuid_t));
+
+        op_errno = ESTALE;
+        GF_PROTOCOL_DICT_SERIALIZE (this, args->xdata, (&req.xdata.xdata_val),
+                                    req.xdata.xdata_len, op_errno, unwind);
+        ret = client_submit_request (this, &req, frame, conf->fops,
+                                     GFS3_OP_ICREATE, client3_3_icreate_cbk,
+                                     NULL, NULL, 0, NULL, 0, NULL,
+                                     (xdrproc_t) xdr_gfs4_icreate_req);
+        if (ret)
+                goto free_reqdata;
+        GF_FREE (req.xdata.xdata_val);
+        return 0;
+
+ free_reqdata:
+        GF_FREE (req.xdata.xdata_val);
+ unwind:
+        CLIENT_STACK_UNWIND (icreate, frame,
+                             -1, op_errno, NULL, NULL, NULL);
+        return 0;
+}
+
+int32_t
 client4_0_fsetattr (call_frame_t *frame, xlator_t *this, void *data)
 {
         clnt_args_t          *args      = NULL;
@@ -6456,6 +6650,8 @@ char *clnt3_3_fop_names[GFS3_OP_MAXVALUE] = {
         [GFS3_OP_GETACTIVELK] = "GETACTIVELK",
         [GFS3_OP_SETACTIVELK] = "SETACTIVELK",
         [GFS3_OP_COMPOUND]    = "COMPOUND",
+        [GFS3_OP_ICREATE]     = "ICREATE",
+        [GFS3_OP_NAMELINK]    = "NAMELINK",
 };
 
 rpc_clnt_prog_t clnt3_3_fop_prog = {
@@ -6522,6 +6718,8 @@ rpc_clnt_procedure_t clnt4_0_fop_actors[GF_FOP_MAXVALUE] = {
         [GF_FOP_GETACTIVELK]  = { "GETACTIVELK", client3_3_getactivelk},
         [GF_FOP_SETACTIVELK]  = { "SETACTIVELK", client3_3_setactivelk},
         [GF_FOP_COMPOUND]     = { "COMPOUND",     client3_3_compound },
+        [GF_FOP_ICREATE]     = { "ICREATE",     client3_3_icreate},
+        [GF_FOP_NAMELINK]    = { "NAMELINK",    client3_3_namelink},
 };
 
 
diff --git a/xlators/protocol/client/src/client.c b/xlators/protocol/client/src/client.c
index b8b0ddbbfe9..8df697ed883 100644
--- a/xlators/protocol/client/src/client.c
+++ b/xlators/protocol/client/src/client.c
@@ -2181,6 +2181,60 @@ out:
 	return 0;
 }
 
+int32_t
+client_namelink (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
+{
+        int32_t               ret  = -1;
+        clnt_conf_t          *conf = NULL;
+        clnt_args_t           args = {0,};
+        rpc_clnt_procedure_t *proc = NULL;
+
+        conf = this->private;
+        if (!conf || !conf->fops || !conf->handshake)
+                goto out;
+
+        args.loc = loc;
+        args.xdata = xdata;
+
+        proc = &conf->fops->proctable[GF_FOP_NAMELINK];
+        if (proc->fn)
+                ret = proc->fn (frame, this, &args);
+
+ out:
+        if (ret)
+                STACK_UNWIND_STRICT (namelink, frame,
+                                     -1, EINVAL, NULL, NULL, NULL);
+        return 0;
+}
+
+int32_t
+client_icreate (call_frame_t *frame,
+                xlator_t *this, loc_t *loc, mode_t mode, dict_t *xdata)
+{
+        int32_t               ret  = -1;
+        clnt_conf_t          *conf = NULL;
+        clnt_args_t           args = {0,};
+        rpc_clnt_procedure_t *proc = NULL;
+
+        conf = this->private;
+        if (!conf || !conf->fops || !conf->handshake)
+                goto out;
+
+        args.loc   = loc;
+        args.mode  = mode;
+        args.xdata = xdata;
+
+        proc = &conf->fops->proctable[GF_FOP_ICREATE];
+        if (proc->fn)
+                ret = proc->fn (frame, this, &args);
+
+ out:
+        if (ret)
+                STACK_UNWIND_STRICT (icreate, frame,
+                                     -1, EINVAL, NULL, NULL, NULL);
+        return 0;
+}
+
 int
 client_mark_fd_bad (xlator_t *this)
 {
@@ -2960,6 +3014,8 @@ struct xlator_fops fops = {
         .compound    = client_compound,
         .getactivelk = client_getactivelk,
         .setactivelk = client_setactivelk,
+        .icreate      = client_icreate,
+        .namelink     = client_namelink,
 };
 
 
diff --git a/xlators/protocol/server/src/server-rpc-fops.c b/xlators/protocol/server/src/server-rpc-fops.c
index 1adb262af7f..1b87abaafda 100644
--- a/xlators/protocol/server/src/server-rpc-fops.c
+++ b/xlators/protocol/server/src/server-rpc-fops.c
@@ -2207,6 +2207,97 @@ out:
         return 0;
 }
 
+int
+server_namelink_cbk (call_frame_t *frame,
+                     void *cookie, xlator_t *this,
+                     int32_t op_ret, int32_t op_errno,
+                     struct iatt *prebuf, struct iatt *postbuf, dict_t *xdata)
+{
+        gfs4_namelink_rsp       rsp        = {0,};
+        rpcsvc_request_t    *req        = NULL;
+
+        GF_PROTOCOL_DICT_SERIALIZE (this, xdata, &rsp.xdata.xdata_val,
+                                    rsp.xdata.xdata_len, op_errno, out);
+
+        if (op_ret < 0)
+                goto out;
+
+        gf_stat_from_iatt (&rsp.preparent, prebuf);
+        gf_stat_from_iatt (&rsp.postparent, postbuf);
+
+        /**
+         * no point in linking inode here -- there's no stbuf anyway and a
+         * lookup() for this name entry would be a negative lookup.
+         */
+
+out:
+        rsp.op_ret    = op_ret;
+        rsp.op_errno  = gf_errno_to_error (op_errno);
+
+        req = frame->local;
+        server_submit_reply (frame, req, &rsp, NULL, 0, NULL,
+                             (xdrproc_t)xdr_gfs4_namelink_rsp);
+
+        GF_FREE (rsp.xdata.xdata_val);
+
+        return 0;
+}
+
+int
+server_icreate_cbk (call_frame_t *frame,
+                    void *cookie, xlator_t *this,
+                    int32_t op_ret, int32_t op_errno,
+                    inode_t *inode, struct iatt *stbuf, dict_t *xdata)
+{
+        server_state_t      *state      = NULL;
+        inode_t             *link_inode = NULL;
+        rpcsvc_request_t    *req        = NULL;
+        gfs3_create_rsp      rsp        = {0,};
+
+        GF_PROTOCOL_DICT_SERIALIZE (this, xdata, &rsp.xdata.xdata_val,
+                                    rsp.xdata.xdata_len, op_errno, out);
+
+        state = CALL_STATE (frame);
+
+        if (op_ret < 0) {
+                gf_msg (this->name, GF_LOG_INFO, op_errno, PS_MSG_CREATE_INFO,
+                        "%"PRId64": ICREATE [%s] ==> (%s)",
+                        frame->root->unique, uuid_utoa (state->resolve.gfid),
+                        strerror (op_errno));
+                goto out;
+        }
+
+        gf_msg_trace (frame->root->client->bound_xl->name, 0, "%"PRId64": "
+                      "ICREATE [%s]", frame->root->unique,
+                      uuid_utoa (stbuf->ia_gfid));
+
+        link_inode = inode_link (inode,
+                                 state->loc.parent, state->loc.name, stbuf);
+
+        if (!link_inode) {
+                op_ret = -1;
+                op_errno = ENOENT;
+                goto out;
+        }
+
+        inode_lookup (link_inode);
+        inode_unref (link_inode);
+
+        gf_stat_from_iatt (&rsp.stat, stbuf);
+
+out:
+        rsp.op_ret    = op_ret;
+        rsp.op_errno  = gf_errno_to_error (op_errno);
+
+        req = frame->local;
+        server_submit_reply (frame, req, &rsp, NULL, 0, NULL,
+                             (xdrproc_t)xdr_gfs4_icreate_rsp);
+
+        GF_FREE (rsp.xdata.xdata_val);
+
+        return 0;
+}
+
 /* Resume function section */
 
 int
@@ -3452,6 +3543,53 @@ err:
         return ret;
 }
 
+int
+server_namelink_resume (call_frame_t *frame, xlator_t *bound_xl)
+{
+        server_state_t *state = NULL;
+
+        state = CALL_STATE (frame);
+
+        if (state->resolve.op_ret != 0)
+                goto err;
+
+        state->loc.inode = inode_new (state->itable);
+
+        STACK_WIND (frame, server_namelink_cbk,
+                    bound_xl, bound_xl->fops->namelink,
+                    &(state->loc), state->xdata);
+        return 0;
+
+ err:
+        server_namelink_cbk (frame, NULL,
+                             frame->this,
+                             state->resolve.op_ret,
+                             state->resolve.op_errno, NULL, NULL, NULL);
+        return 0;
+}
+
+int
+server_icreate_resume (call_frame_t *frame, xlator_t *bound_xl)
+{
+        server_state_t *state = NULL;
+
+        state = CALL_STATE (frame);
+
+        if (state->resolve.op_ret != 0)
+                goto err;
+
+        state->loc.inode = inode_new (state->itable);
+
+        STACK_WIND (frame, server_icreate_cbk,
+                    bound_xl, bound_xl->fops->icreate,
+                    &(state->loc), state->mode, state->xdata);
+
+        return 0;
+err:
+        server_icreate_cbk (frame, NULL, frame->this, state->resolve.op_ret,
+                            state->resolve.op_errno, NULL, NULL, NULL);
+        return 0;
+}
 
 /* Fop section */
 static inline int
@@ -6091,6 +6229,100 @@ out:
 }
 
 int
+server3_3_namelink (rpcsvc_request_t *req)
+{
+        server_state_t    *state    = NULL;
+        call_frame_t      *frame    = NULL;
+        gfs4_namelink_req  args     = {{0,},};
+        int                ret      = -1;
+        int                op_errno = 0;
+
+        if (!req)
+                return ret;
+
+        ret = rpc_receive_common (req, &frame, &state, NULL, &args,
+                                  xdr_gfs4_namelink_req, GF_FOP_NAMELINK);
+
+        if (ret != 0)
+                goto out;
+
+        state->resolve.bname = gf_strdup (args.bname);
+        memcpy (state->resolve.pargfid, args.pargfid, sizeof (uuid_t));
+
+        state->resolve.type = RESOLVE_NOT;
+
+        /* TODO: can do alloca for xdata field instead of stdalloc */
+        GF_PROTOCOL_DICT_UNSERIALIZE (frame->root->client->bound_xl,
+                                      state->xdata,
+                                      args.xdata.xdata_val,
+                                      args.xdata.xdata_len, ret,
+                                      op_errno, out);
+
+        ret = 0;
+        resolve_and_resume (frame, server_namelink_resume);
+
+out:
+        /* memory allocated by libc, don't use GF_FREE */
+        free (args.xdata.xdata_val);
+
+        if (op_errno)
+                SERVER_REQ_SET_ERROR (req, ret);
+
+        return ret;
+
+}
+
+int
+server3_3_icreate (rpcsvc_request_t *req)
+{
+        server_state_t   *state    = NULL;
+        call_frame_t     *frame    = NULL;
+        gfs4_icreate_req  args     = {{0,},};
+        int               ret      = -1;
+        int               op_errno = 0;
+        uuid_t            gfid     = {0,};
+
+        if (!req)
+                return ret;
+
+        ret = rpc_receive_common (req, &frame, &state, NULL, &args,
+                                  xdr_gfs4_icreate_req, GF_FOP_ICREATE);
+
+        if (ret != 0)
+                goto out;
+
+        memcpy (gfid, args.gfid, sizeof (uuid_t));
+
+        state->mode  = args.mode;
+        gf_asprintf (&state->resolve.bname, INODE_PATH_FMT, uuid_utoa (gfid));
+
+        /* parent is an auxillary inode number */
+        memset (state->resolve.pargfid, 0, sizeof (uuid_t));
+        state->resolve.pargfid[15] = GF_AUXILLARY_PARGFID;
+
+        state->resolve.type = RESOLVE_NOT;
+
+        /* TODO: can do alloca for xdata field instead of stdalloc */
+        GF_PROTOCOL_DICT_UNSERIALIZE (frame->root->client->bound_xl,
+                                      state->xdata,
+                                      args.xdata.xdata_val,
+                                      args.xdata.xdata_len, ret,
+                                      op_errno, out);
+
+        ret = 0;
+        resolve_and_resume (frame, server_icreate_resume);
+
+out:
+        /* memory allocated by libc, don't use GF_FREE */
+        free (args.xdata.xdata_val);
+
+        if (op_errno)
+                SERVER_REQ_SET_ERROR (req, ret);
+
+        return ret;
+}
+
+int
 server4_0_fsetattr (rpcsvc_request_t *req)
 {
         server_state_t       *state    = NULL;
@@ -6297,6 +6529,8 @@ rpcsvc_actor_t glusterfs4_0_fop_actors[] = {
         [GFS3_OP_GETACTIVELK]  = {"GETACTIVELK",  GFS3_OP_GETACTIVELK,  server3_3_getactivelk,  NULL, 0, DRC_NA},
         [GFS3_OP_SETACTIVELK]  = {"SETACTIVELK",  GFS3_OP_SETACTIVELK,  server3_3_setactivelk,  NULL, 0, DRC_NA},
         [GFS3_OP_COMPOUND]     = {"COMPOUND",     GFS3_OP_COMPOUND,     server3_3_compound,     NULL, 0, DRC_NA},
+        [GFS3_OP_ICREATE]     =  {"ICREATE",      GFS3_OP_ICREATE,      server3_3_icreate,      NULL, 0, DRC_NA},
+        [GFS3_OP_NAMELINK]    =  {"NAMELINK",     GFS3_OP_NAMELINK,     server3_3_namelink,     NULL, 0, DRC_NA},
 };