lustre/mdc/mdc_lib.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (c) 2003 Cluster File Systems, Inc.
   5  *
   6  *   This file is part of the Lustre file system, http://www.lustre.org
   7  *   Lustre is a trademark of Cluster File Systems, Inc.
   8  *
   9  *   You may have signed or agreed to another license before downloading
  10  *   this software.  If so, you are bound by the terms and conditions
  11  *   of that agreement, and the following does not apply to you.  See the
  12  *   LICENSE file included with this distribution for more information.
  13  *
  14  *   If you did not agree to a different license, then this copy of Lustre
  15  *   is open source software; you can redistribute it and/or modify it
  16  *   under the terms of version 2 of the GNU General Public License as
  17  *   published by the Free Software Foundation.
  18  *
  19  *   In either case, Lustre is distributed in the hope that it will be
  20  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  21  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  22  *   license text for more details.
  23  */
  24
  25 #define DEBUG_SUBSYSTEM S_MDC
  26 #ifndef __KERNEL__
  27 # include <fcntl.h>
  28 # include <liblustre.h>
  29 #endif
  30 #include <lustre/lustre_idl.h>
  31 #include <lustre_net.h>
  32 #include "mdc_internal.h"
  33
  34 #ifndef __KERNEL__
  35 /* some liblustre hackings here */
  36 #ifndef O_DIRECTORY
  37 #define O_DIRECTORY     0
  38 #endif
  39 #endif
  40
  41 static void mdc_pack_body(struct mdt_body *b)
  42 {
  43         LASSERT (b != NULL);
  44
  45         b->fsuid = current->fsuid;
  46         b->fsgid = current->fsgid;
  47         b->capability = current->cap_effective;
  48 }
  49
  50 void mdc_pack_capa(struct ptlrpc_request *req, int offset, struct obd_capa *oc)
  51 {
  52         struct lustre_capa *c;
  53
  54         if (oc == NULL) {
  55                 LASSERT(lustre_msg_buflen(req->rq_reqmsg, offset) == 0);
  56                 return;
  57         }
  58
  59         c = lustre_msg_buf(req->rq_reqmsg, offset, sizeof(*c));
  60         LASSERT(c != NULL);
  61         capa_cpy(c, oc);
  62         DEBUG_CAPA(D_SEC, c, "pack");
  63 }
  64
  65 void mdc_is_subdir_pack(struct ptlrpc_request *req, int offset,
  66                         const struct lu_fid *pfid,
  67                         const struct lu_fid *cfid, int flags)
  68 {
  69         struct mdt_body *b = lustre_msg_buf(req->rq_reqmsg, offset, sizeof(*b));
  70
  71         if (pfid)
  72                 b->fid1 = *pfid;
  73         if (cfid)
  74                 b->fid2 = *cfid;
  75         b->valid = OBD_MD_FLID;
  76         b->flags = flags;
  77 }
  78
  79 void mdc_pack_req_body(struct ptlrpc_request *req, int offset,
  80                        __u64 valid, const struct lu_fid *fid,
  81                        struct obd_capa *oc, int ea_size, int flags)
  82 {
  83         struct mdt_body *b = lustre_msg_buf(req->rq_reqmsg, offset, sizeof(*b));
  84
  85         b->valid = valid;
  86         b->eadatasize = ea_size;
  87         b->flags = flags;
  88         mdc_pack_body(b);
  89         if (fid) {
  90                 b->fid1 = *fid;
  91                 mdc_pack_capa(req, offset + 1, oc);
  92         }
  93 }
  94
  95 void mdc_readdir_pack(struct ptlrpc_request *req, int offset, __u64 pgoff,
  96                       __u32 size, const struct lu_fid *fid, struct obd_capa *oc)
  97 {
  98         struct mdt_body *b;
  99
 100         b = lustre_msg_buf(req->rq_reqmsg, offset, sizeof(*b));
 101         b->fid1 = *fid;
 102         b->size = pgoff;                       /* !! */
 103         b->suppgid = -1;
 104         b->nlink = size;                        /* !! */
 105         mdc_pack_body(b);
 106         mdc_pack_capa(req, offset + 1, oc);
 107 }
 108
 109 /* packing of MDS records */
 110 void mdc_create_pack(struct ptlrpc_request *req, int offset,
 111                      struct md_op_data *op_data, const void *data, int datalen,
 112                      __u32 mode, __u32 uid, __u32 gid, __u32 cap_effective,
 113                      __u64 rdev)
 114 {
 115         struct mdt_rec_create *rec;
 116         char *tmp;
 117
 118         rec = lustre_msg_buf(req->rq_reqmsg, offset, sizeof(*rec));
 119
 120         rec->cr_opcode = REINT_CREATE;
 121         rec->cr_fsuid = uid;
 122         rec->cr_fsgid = gid;
 123         rec->cr_cap = cap_effective;
 124         rec->cr_fid1 = op_data->op_fid1;
 125         rec->cr_fid2 = op_data->op_fid2;
 126         rec->cr_mode = mode;
 127         rec->cr_rdev = rdev;
 128         rec->cr_time = op_data->op_mod_time;
 129         rec->cr_suppgid1 = op_data->op_suppgids[0];
 130         rec->cr_suppgid2 = op_data->op_suppgids[1];
 131         rec->cr_flags = op_data->op_flags;
 132         rec->cr_bias = op_data->op_bias;
 133
 134         mdc_pack_capa(req, offset + 1, op_data->op_capa1);
 135
 136         tmp = lustre_msg_buf(req->rq_reqmsg, offset + 2, op_data->op_namelen + 1);
 137         LOGL0(op_data->op_name, op_data->op_namelen, tmp);
 138
 139         if (data) {
 140                 tmp = lustre_msg_buf(req->rq_reqmsg, offset + 3, datalen);
 141                 memcpy(tmp, data, datalen);
 142         }
 143 }
 144
 145 static __u32 mds_pack_open_flags(__u32 flags)
 146 {
 147         return
 148                 (flags & (FMODE_READ | FMODE_WRITE |
 149                           MDS_OPEN_DELAY_CREATE | MDS_OPEN_HAS_EA |
 150                           MDS_OPEN_HAS_OBJS | MDS_OPEN_OWNEROVERRIDE |
 151                           MDS_OPEN_LOCK)) |
 152                 ((flags & O_CREAT) ? MDS_OPEN_CREAT : 0) |
 153                 ((flags & O_EXCL) ? MDS_OPEN_EXCL : 0) |
 154                 ((flags & O_TRUNC) ? MDS_OPEN_TRUNC : 0) |
 155                 ((flags & O_APPEND) ? MDS_OPEN_APPEND : 0) |
 156                 ((flags & O_SYNC) ? MDS_OPEN_SYNC : 0) |
 157                 ((flags & O_DIRECTORY) ? MDS_OPEN_DIRECTORY : 0) |
 158                 ((flags & O_JOIN_FILE) ? MDS_OPEN_JOIN_FILE : 0) |
 159 #ifdef FMODE_EXEC
 160                 ((flags & FMODE_EXEC) ? MDS_FMODE_EXEC : 0) |
 161 #endif
 162                 0;
 163 }
 164
 165 /* packing of MDS records */
 166 void mdc_join_pack(struct ptlrpc_request *req, int offset,
 167                    struct md_op_data *op_data, __u64 head_size)
 168 {
 169         struct mdt_rec_join *rec;
 170
 171         rec = lustre_msg_buf(req->rq_reqmsg, offset, sizeof(*rec));
 172         LASSERT(rec != NULL);
 173         rec->jr_fid = op_data->op_fid2;
 174         rec->jr_headsize = head_size;
 175 }
 176
 177 void mdc_open_pack(struct ptlrpc_request *req, int offset,
 178                    struct md_op_data *op_data, __u32 mode, __u64 rdev,
 179                    __u32 flags, const void *lmm, int lmmlen)
 180 {
 181         struct mdt_rec_create *rec;
 182         char *tmp;
 183         rec = lustre_msg_buf(req->rq_reqmsg, offset, sizeof (*rec));
 184
 185         /* XXX do something about time, uid, gid */
 186         rec->cr_opcode = REINT_OPEN;
 187         rec->cr_fsuid = current->fsuid;
 188         rec->cr_fsgid = current->fsgid;
 189         rec->cr_cap = current->cap_effective;
 190         if (op_data != NULL) {
 191                 rec->cr_fid1 = op_data->op_fid1;
 192                 rec->cr_fid2 = op_data->op_fid2;
 193         }
 194         rec->cr_mode = mode;
 195         rec->cr_flags = mds_pack_open_flags(flags);
 196         rec->cr_time = op_data->op_mod_time;
 197         rec->cr_rdev = rdev;
 198         rec->cr_suppgid1 = op_data->op_suppgids[0];
 199         rec->cr_suppgid2 = op_data->op_suppgids[1];
 200         rec->cr_bias = op_data->op_bias;
 201
 202         mdc_pack_capa(req, offset + 1, op_data->op_capa1);
 203         /* the next buffer is child capa, which is used for replay,
 204          * will be packed from the data in reply message. */
 205
 206         if (op_data->op_name) {
 207                 tmp = lustre_msg_buf(req->rq_reqmsg, offset + 3,
 208                                      op_data->op_namelen + 1);
 209                 LOGL0(op_data->op_name, op_data->op_namelen, tmp);
 210         }
 211
 212         if (lmm) {
 213                 rec->cr_flags |= MDS_OPEN_HAS_EA;
 214 #ifndef __KERNEL__
 215                 /*XXX a hack for liblustre to set EA (LL_IOC_LOV_SETSTRIPE) */
 216                 rec->cr_fid2 = op_data->op_fid2;
 217 #endif
 218                 tmp = lustre_msg_buf(req->rq_reqmsg, offset + 4, lmmlen);
 219                 memcpy (tmp, lmm, lmmlen);
 220         }
 221 }
 222
 223 static void mdc_setattr_pack_rec(struct mdt_rec_setattr *rec,
 224                                  struct md_op_data *op_data)
 225 {
 226         rec->sa_opcode = REINT_SETATTR;
 227         rec->sa_fsuid = current->fsuid;
 228         rec->sa_fsgid = current->fsgid;
 229         rec->sa_cap = current->cap_effective;
 230         rec->sa_suppgid = -1;
 231
 232         rec->sa_fid = op_data->op_fid1;
 233         rec->sa_valid = op_data->op_attr.ia_valid;
 234         rec->sa_mode = op_data->op_attr.ia_mode;
 235         rec->sa_uid = op_data->op_attr.ia_uid;
 236         rec->sa_gid = op_data->op_attr.ia_gid;
 237         rec->sa_size = op_data->op_attr.ia_size;
 238         rec->sa_blocks = op_data->op_attr_blocks;
 239         rec->sa_atime = LTIME_S(op_data->op_attr.ia_atime);
 240         rec->sa_mtime = LTIME_S(op_data->op_attr.ia_mtime);
 241         rec->sa_ctime = LTIME_S(op_data->op_attr.ia_ctime);
 242         rec->sa_attr_flags = ((struct ll_iattr *)&op_data->op_attr)->ia_attr_flags;
 243         if ((op_data->op_attr.ia_valid & ATTR_GID) &&
 244             in_group_p(op_data->op_attr.ia_gid))
 245                 rec->sa_suppgid = op_data->op_attr.ia_gid;
 246         else
 247                 rec->sa_suppgid = op_data->op_suppgids[0];
 248 }
 249
 250 static void mdc_epoch_pack(struct mdt_epoch *epoch, struct md_op_data *op_data)
 251 {
 252         memcpy(&epoch->handle, &op_data->op_handle, sizeof(epoch->handle));
 253         epoch->ioepoch = op_data->op_ioepoch;
 254         epoch->flags = op_data->op_flags;
 255 }
 256
 257 void mdc_setattr_pack(struct ptlrpc_request *req, int offset,
 258                       struct md_op_data *op_data, void *ea,
 259                       int ealen, void *ea2, int ea2len)
 260 {
 261         struct mdt_rec_setattr *rec;
 262         struct mdt_epoch *epoch;
 263
 264         rec = lustre_msg_buf(req->rq_reqmsg, offset, sizeof (*rec));
 265         mdc_setattr_pack_rec(rec, op_data);
 266
 267         mdc_pack_capa(req, offset + 1, op_data->op_capa1);
 268
 269         if (op_data->op_flags & (MF_SOM_CHANGE | MF_EPOCH_OPEN)) {
 270                 epoch = lustre_msg_buf(req->rq_reqmsg, offset + 2,
 271                                         sizeof(*epoch));
 272                 mdc_epoch_pack(epoch, op_data);
 273         }
 274
 275         if (ealen == 0)
 276                 return;
 277
 278         memcpy(lustre_msg_buf(req->rq_reqmsg, offset + 3, ealen), ea, ealen);
 279
 280         if (ea2len == 0)
 281                 return;
 282
 283         memcpy(lustre_msg_buf(req->rq_reqmsg, offset + 4, ea2len), ea2, ea2len);
 284 }
 285
 286 void mdc_unlink_pack(struct ptlrpc_request *req, int offset,
 287                      struct md_op_data *op_data)
 288 {
 289         struct mdt_rec_unlink *rec;
 290         char *tmp;
 291
 292         rec = lustre_msg_buf(req->rq_reqmsg, offset, sizeof (*rec));
 293         LASSERT (rec != NULL);
 294
 295         rec->ul_opcode = REINT_UNLINK;
 296         rec->ul_fsuid = op_data->op_fsuid;//current->fsuid;
 297         rec->ul_fsgid = op_data->op_fsgid;//current->fsgid;
 298         rec->ul_cap = op_data->op_cap;//current->cap_effective;
 299         rec->ul_mode = op_data->op_mode;
 300         rec->ul_suppgid = op_data->op_suppgids[0];
 301         rec->ul_fid1 = op_data->op_fid1;
 302         rec->ul_fid2 = op_data->op_fid2;
 303         rec->ul_time = op_data->op_mod_time;
 304         rec->ul_bias = op_data->op_bias;
 305
 306         mdc_pack_capa(req, offset + 1, op_data->op_capa1);
 307
 308         tmp = lustre_msg_buf(req->rq_reqmsg, offset + 2, op_data->op_namelen + 1);
 309         LASSERT(tmp != NULL);
 310         LOGL0(op_data->op_name, op_data->op_namelen, tmp);
 311 }
 312
 313 void mdc_link_pack(struct ptlrpc_request *req, int offset,
 314                    struct md_op_data *op_data)
 315 {
 316         struct mdt_rec_link *rec;
 317         char *tmp;
 318
 319         rec = lustre_msg_buf(req->rq_reqmsg, offset, sizeof (*rec));
 320
 321         rec->lk_opcode = REINT_LINK;
 322         rec->lk_fsuid = op_data->op_fsuid;//current->fsuid;
 323         rec->lk_fsgid = op_data->op_fsgid;//current->fsgid;
 324         rec->lk_cap = op_data->op_cap;//current->cap_effective;
 325         rec->lk_suppgid1 = op_data->op_suppgids[0];
 326         rec->lk_suppgid2 = op_data->op_suppgids[1];
 327         rec->lk_fid1 = op_data->op_fid1;
 328         rec->lk_fid2 = op_data->op_fid2;
 329         rec->lk_time = op_data->op_mod_time;
 330         rec->lk_bias = op_data->op_bias;
 331
 332         mdc_pack_capa(req, offset + 1, op_data->op_capa1);
 333         mdc_pack_capa(req, offset + 2, op_data->op_capa2);
 334
 335         tmp = lustre_msg_buf(req->rq_reqmsg, offset + 3, op_data->op_namelen + 1);
 336         LOGL0(op_data->op_name, op_data->op_namelen, tmp);
 337 }
 338
 339 void mdc_rename_pack(struct ptlrpc_request *req, int offset,
 340                      struct md_op_data *op_data,
 341                      const char *old, int oldlen, const char *new, int newlen)
 342 {
 343         struct mdt_rec_rename *rec;
 344         char *tmp;
 345
 346         rec = lustre_msg_buf(req->rq_reqmsg, offset, sizeof (*rec));
 347
 348         /* XXX do something about time, uid, gid */
 349         rec->rn_opcode = REINT_RENAME;
 350         rec->rn_fsuid = op_data->op_fsuid;//current->fsuid;
 351         rec->rn_fsgid = op_data->op_fsgid;//current->fsgid;
 352         rec->rn_cap = op_data->op_cap;//current->cap_effective;
 353         rec->rn_suppgid1 = op_data->op_suppgids[0];
 354         rec->rn_suppgid2 = op_data->op_suppgids[1];
 355         rec->rn_fid1 = op_data->op_fid1;
 356         rec->rn_fid2 = op_data->op_fid2;
 357         rec->rn_time = op_data->op_mod_time;
 358         rec->rn_mode = op_data->op_mode;
 359         rec->rn_bias = op_data->op_bias;
 360
 361         mdc_pack_capa(req, offset + 1, op_data->op_capa1);
 362         mdc_pack_capa(req, offset + 2, op_data->op_capa2);
 363
 364         tmp = lustre_msg_buf(req->rq_reqmsg, offset + 3, oldlen + 1);
 365         LOGL0(old, oldlen, tmp);
 366
 367         if (new) {
 368                 tmp = lustre_msg_buf(req->rq_reqmsg, offset + 4, newlen + 1);
 369                 LOGL0(new, newlen, tmp);
 370         }
 371 }
 372
 373 void mdc_getattr_pack(struct ptlrpc_request *req, int offset, __u64 valid,
 374                       int flags, struct md_op_data *op_data)
 375 {
 376         struct mdt_body *b;
 377         b = lustre_msg_buf(req->rq_reqmsg, offset, sizeof (*b));
 378
 379         b->fsuid = current->fsuid;
 380         b->fsgid = current->fsgid;
 381         b->capability = current->cap_effective;
 382         b->valid = valid;
 383         if (op_data->op_bias & MDS_CHECK_SPLIT)
 384                 b->valid |= OBD_MD_FLCKSPLIT;
 385         if (op_data->op_bias & MDS_CROSS_REF)
 386                 b->valid |= OBD_MD_FLCROSSREF;
 387         b->flags = flags | MDS_BFLAG_EXT_FLAGS;
 388         b->suppgid = op_data->op_suppgids[0];
 389
 390         b->fid1 = op_data->op_fid1;
 391         b->fid2 = op_data->op_fid2;
 392
 393         mdc_pack_capa(req, offset + 1, op_data->op_capa1);
 394
 395         if (op_data->op_name) {
 396                 char *tmp;
 397                 tmp = lustre_msg_buf(req->rq_reqmsg, offset + 2,
 398                                      op_data->op_namelen + 1);
 399                 LOGL0(op_data->op_name, op_data->op_namelen, tmp);
 400         }
 401 }
 402
 403 void mdc_close_pack(struct ptlrpc_request *req, int offset,
 404                     struct md_op_data *op_data)
 405 {
 406         struct mdt_epoch *epoch;
 407         struct mdt_rec_setattr *rec;
 408
 409         epoch = lustre_msg_buf(req->rq_reqmsg, offset, sizeof(*epoch));
 410         rec = lustre_msg_buf(req->rq_reqmsg, offset + 1, sizeof(*rec));
 411
 412         mdc_setattr_pack_rec(rec, op_data);
 413         mdc_pack_capa(req, offset + 2, op_data->op_capa1);
 414         mdc_epoch_pack(epoch, op_data);
 415 }
 416
 417 static int mdc_req_avail(struct client_obd *cli, struct mdc_cache_waiter *mcw)
 418 {
 419         int rc;
 420         ENTRY;
 421         spin_lock(&cli->cl_loi_list_lock);
 422         rc = list_empty(&mcw->mcw_entry);
 423         spin_unlock(&cli->cl_loi_list_lock);
 424         RETURN(rc);
 425 };
 426
 427 /* We record requests in flight in cli->cl_r_in_flight here.
 428  * There is only one write rpc possible in mdc anyway. If this to change
 429  * in the future - the code may need to be revisited. */
 430 void mdc_enter_request(struct client_obd *cli)
 431 {
 432         struct mdc_cache_waiter mcw;
 433         struct l_wait_info lwi = { 0 };
 434
 435         spin_lock(&cli->cl_loi_list_lock);
 436         if (cli->cl_r_in_flight >= cli->cl_max_rpcs_in_flight) {
 437                 list_add_tail(&mcw.mcw_entry, &cli->cl_cache_waiters);
 438                 init_waitqueue_head(&mcw.mcw_waitq);
 439                 spin_unlock(&cli->cl_loi_list_lock);
 440                 l_wait_event(mcw.mcw_waitq, mdc_req_avail(cli, &mcw), &lwi);
 441         } else {
 442                 cli->cl_r_in_flight++;
 443                 spin_unlock(&cli->cl_loi_list_lock);
 444         }
 445 }
 446
 447 void mdc_exit_request(struct client_obd *cli)
 448 {
 449         struct list_head *l, *tmp;
 450         struct mdc_cache_waiter *mcw;
 451
 452         spin_lock(&cli->cl_loi_list_lock);
 453         cli->cl_r_in_flight--;
 454         list_for_each_safe(l, tmp, &cli->cl_cache_waiters) {
 455
 456                 if (cli->cl_r_in_flight >= cli->cl_max_rpcs_in_flight) {
 457                         /* No free request slots anymore */
 458                         break;
 459                 }
 460
 461                 mcw = list_entry(l, struct mdc_cache_waiter, mcw_entry);
 462                 list_del_init(&mcw->mcw_entry);
 463                 cli->cl_r_in_flight++;
 464                 wake_up(&mcw->mcw_waitq);
 465         }
 466         /* Empty waiting list? Decrease reqs in-flight number */
 467
 468         spin_unlock(&cli->cl_loi_list_lock);
 469 }