lustre/lov/lov_request.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * Copyright (C) 2002, 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 #ifndef EXPORT_SYMTAB
  26 # define EXPORT_SYMTAB
  27 #endif
  28 #define DEBUG_SUBSYSTEM S_LOV
  29
  30 #ifdef __KERNEL__
  31 #include <libcfs/libcfs.h>
  32 #else
  33 #include <liblustre.h>
  34 #endif
  35
  36 #include <obd_class.h>
  37 #include <obd_lov.h>
  38 #include <lustre/lustre_idl.h>
  39
  40 #include "lov_internal.h"
  41
  42 static void lov_init_set(struct lov_request_set *set)
  43 {
  44         set->set_count = 0;
  45         set->set_completes = 0;
  46         set->set_success = 0;
  47         set->set_cookies = 0;
  48         CFS_INIT_LIST_HEAD(&set->set_list);
  49         atomic_set(&set->set_refcount, 1);
  50 }
  51
  52 static void lov_finish_set(struct lov_request_set *set)
  53 {
  54         struct list_head *pos, *n;
  55         ENTRY;
  56
  57         LASSERT(set);
  58         list_for_each_safe(pos, n, &set->set_list) {
  59                 struct lov_request *req = list_entry(pos, struct lov_request,
  60                                                      rq_link);
  61                 list_del_init(&req->rq_link);
  62
  63                 if (req->rq_oi.oi_oa)
  64                         OBDO_FREE(req->rq_oi.oi_oa);
  65                 if (req->rq_oi.oi_md)
  66                         OBD_FREE(req->rq_oi.oi_md, req->rq_buflen);
  67                 if (req->rq_oi.oi_osfs)
  68                         OBD_FREE(req->rq_oi.oi_osfs,
  69                                  sizeof(*req->rq_oi.oi_osfs));
  70                 OBD_FREE(req, sizeof(*req));
  71         }
  72
  73         if (set->set_pga) {
  74                 int len = set->set_oabufs * sizeof(*set->set_pga);
  75                 OBD_FREE(set->set_pga, len);
  76         }
  77         if (set->set_lockh)
  78                 lov_llh_put(set->set_lockh);
  79
  80         OBD_FREE(set, sizeof(*set));
  81         EXIT;
  82 }
  83
  84 void lov_update_set(struct lov_request_set *set,
  85                     struct lov_request *req, int rc)
  86 {
  87         req->rq_complete = 1;
  88         req->rq_rc = rc;
  89
  90         set->set_completes++;
  91         if (rc == 0)
  92                 set->set_success++;
  93 }
  94
  95 int lov_update_common_set(struct lov_request_set *set,
  96                           struct lov_request *req, int rc)
  97 {
  98         struct lov_obd *lov = &set->set_exp->exp_obd->u.lov;
  99         ENTRY;
 100
 101         lov_update_set(set, req, rc);
 102
 103         /* grace error on inactive ost */
 104         if (rc && !(lov->lov_tgts[req->rq_idx] &&
 105                     lov->lov_tgts[req->rq_idx]->ltd_active))
 106                 rc = 0;
 107
 108         /* FIXME in raid1 regime, should return 0 */
 109         RETURN(rc);
 110 }
 111
 112 void lov_set_add_req(struct lov_request *req, struct lov_request_set *set)
 113 {
 114         list_add_tail(&req->rq_link, &set->set_list);
 115         set->set_count++;
 116 }
 117
 118 int lov_update_enqueue_set(struct lov_request *req, __u32 mode, int rc)
 119 {
 120         struct lov_request_set *set = req->rq_rqset;
 121         struct lustre_handle *lov_lockhp;
 122         struct lov_oinfo *loi;
 123         ENTRY;
 124
 125         LASSERT(set != NULL);
 126         LASSERT(set->set_oi != NULL);
 127
 128         lov_lockhp = set->set_lockh->llh_handles + req->rq_stripe;
 129         loi = set->set_oi->oi_md->lsm_oinfo[req->rq_stripe];
 130
 131         /* XXX LOV STACKING: OSC gets a copy, created in lov_prep_enqueue_set
 132          * and that copy can be arbitrarily out of date.
 133          *
 134          * The LOV API is due for a serious rewriting anyways, and this
 135          * can be addressed then. */
 136
 137         if (rc == ELDLM_OK) {
 138                 struct ldlm_lock *lock = ldlm_handle2lock(lov_lockhp);
 139                 __u64 tmp;
 140
 141                 LASSERT(lock != NULL);
 142                 lov_stripe_lock(set->set_oi->oi_md);
 143                 loi->loi_lvb = req->rq_oi.oi_md->lsm_oinfo[0]->loi_lvb;
 144                 tmp = loi->loi_lvb.lvb_size;
 145                 /* Extend KMS up to the end of this lock and no further
 146                  * A lock on [x,y] means a KMS of up to y + 1 bytes! */
 147                 if (tmp > lock->l_policy_data.l_extent.end)
 148                         tmp = lock->l_policy_data.l_extent.end + 1;
 149                 if (tmp >= loi->loi_kms) {
 150                         LDLM_DEBUG(lock, "lock acquired, setting rss="LPU64
 151                                    ", kms="LPU64, loi->loi_lvb.lvb_size, tmp);
 152                         loi->loi_kms = tmp;
 153                         loi->loi_kms_valid = 1;
 154                 } else {
 155                         LDLM_DEBUG(lock, "lock acquired, setting rss="
 156                                    LPU64"; leaving kms="LPU64", end="LPU64,
 157                                    loi->loi_lvb.lvb_size, loi->loi_kms,
 158                                    lock->l_policy_data.l_extent.end);
 159                 }
 160                 lov_stripe_unlock(set->set_oi->oi_md);
 161                 ldlm_lock_allow_match(lock);
 162                 LDLM_LOCK_PUT(lock);
 163         } else if ((rc == ELDLM_LOCK_ABORTED) &&
 164                    (set->set_oi->oi_flags & LDLM_FL_HAS_INTENT)) {
 165                 memset(lov_lockhp, 0, sizeof(*lov_lockhp));
 166                 lov_stripe_lock(set->set_oi->oi_md);
 167                 loi->loi_lvb = req->rq_oi.oi_md->lsm_oinfo[0]->loi_lvb;
 168                 lov_stripe_unlock(set->set_oi->oi_md);
 169                 CDEBUG(D_INODE, "glimpsed, setting rss="LPU64"; leaving"
 170                        " kms="LPU64"\n", loi->loi_lvb.lvb_size, loi->loi_kms);
 171                 rc = ELDLM_OK;
 172         } else {
 173                 struct obd_export *exp = set->set_exp;
 174                 struct lov_obd *lov = &exp->exp_obd->u.lov;
 175
 176                 memset(lov_lockhp, 0, sizeof(*lov_lockhp));
 177                 if (lov->lov_tgts[req->rq_idx] &&
 178                     lov->lov_tgts[req->rq_idx]->ltd_active) {
 179                         if (rc != -EINTR)
 180                                 CERROR("enqueue objid "LPX64" subobj "
 181                                        LPX64" on OST idx %d: rc %d\n",
 182                                        set->set_oi->oi_md->lsm_object_id,
 183                                        loi->loi_id, loi->loi_ost_idx, rc);
 184                 } else {
 185                         rc = ELDLM_OK;
 186                 }
 187         }
 188         lov_update_set(set, req, rc);
 189         RETURN(rc);
 190 }
 191
 192 /* The callback for osc_enqueue that updates lov info for every OSC request. */
 193 static int cb_update_enqueue(struct obd_info *oinfo, int rc)
 194 {
 195         struct ldlm_enqueue_info *einfo;
 196         struct lov_request *lovreq;
 197
 198         lovreq = container_of(oinfo, struct lov_request, rq_oi);
 199         einfo = lovreq->rq_rqset->set_ei;
 200         return lov_update_enqueue_set(lovreq, einfo->ei_mode, rc);
 201 }
 202
 203 static int enqueue_done(struct lov_request_set *set, __u32 mode)
 204 {
 205         struct lov_request *req;
 206         struct lov_obd *lov = &set->set_exp->exp_obd->u.lov;
 207         int rc = 0;
 208         ENTRY;
 209
 210         /* enqueue/match success, just return */
 211         if (set->set_completes && set->set_completes == set->set_success)
 212                 RETURN(0);
 213
 214         /* cancel enqueued/matched locks */
 215         list_for_each_entry(req, &set->set_list, rq_link) {
 216                 struct lustre_handle *lov_lockhp;
 217
 218                 if (!req->rq_complete || req->rq_rc)
 219                         continue;
 220
 221                 lov_lockhp = set->set_lockh->llh_handles + req->rq_stripe;
 222                 LASSERT(lov_lockhp);
 223                 if (!lustre_handle_is_used(lov_lockhp))
 224                         continue;
 225
 226                 rc = obd_cancel(lov->lov_tgts[req->rq_idx]->ltd_exp,
 227                                 req->rq_oi.oi_md, mode, lov_lockhp);
 228                 if (rc && lov->lov_tgts[req->rq_idx] &&
 229                     lov->lov_tgts[req->rq_idx]->ltd_active)
 230                         CERROR("cancelling obdjid "LPX64" on OST "
 231                                "idx %d error: rc = %d\n",
 232                                req->rq_oi.oi_md->lsm_object_id,
 233                                req->rq_idx, rc);
 234         }
 235         if (set->set_lockh)
 236                 lov_llh_put(set->set_lockh);
 237         RETURN(rc);
 238 }
 239
 240 int lov_fini_enqueue_set(struct lov_request_set *set, __u32 mode, int rc,
 241                          struct ptlrpc_request_set *rqset)
 242 {
 243         int ret = 0;
 244         ENTRY;
 245
 246         if (set == NULL)
 247                 RETURN(0);
 248         LASSERT(set->set_exp);
 249         /* Do enqueue_done only for sync requests and if any request
 250          * succeeded. */
 251         if (!rqset) {
 252                 if (rc)
 253                         set->set_completes = 0;
 254                 ret = enqueue_done(set, mode);
 255         } else if (set->set_lockh)
 256                 lov_llh_put(set->set_lockh);
 257
 258         if (atomic_dec_and_test(&set->set_refcount))
 259                 lov_finish_set(set);
 260
 261         RETURN(rc ? rc : ret);
 262 }
 263
 264 int lov_prep_enqueue_set(struct obd_export *exp, struct obd_info *oinfo,
 265                          struct ldlm_enqueue_info *einfo,
 266                          struct lov_request_set **reqset)
 267 {
 268         struct lov_obd *lov = &exp->exp_obd->u.lov;
 269         struct lov_request_set *set;
 270         int i, rc = 0;
 271         ENTRY;
 272
 273         OBD_ALLOC(set, sizeof(*set));
 274         if (set == NULL)
 275                 RETURN(-ENOMEM);
 276         lov_init_set(set);
 277
 278         set->set_exp = exp;
 279         set->set_oi = oinfo;
 280         set->set_ei = einfo;
 281         set->set_lockh = lov_llh_new(oinfo->oi_md);
 282         if (set->set_lockh == NULL)
 283                 GOTO(out_set, rc = -ENOMEM);
 284         oinfo->oi_lockh->cookie = set->set_lockh->llh_handle.h_cookie;
 285
 286         for (i = 0; i < oinfo->oi_md->lsm_stripe_count; i++) {
 287                 struct lov_oinfo *loi;
 288                 struct lov_request *req;
 289                 obd_off start, end;
 290
 291                 loi = oinfo->oi_md->lsm_oinfo[i];
 292                 if (!lov_stripe_intersects(oinfo->oi_md, i,
 293                                            oinfo->oi_policy.l_extent.start,
 294                                            oinfo->oi_policy.l_extent.end,
 295                                            &start, &end))
 296                         continue;
 297
 298                 if (!lov->lov_tgts[loi->loi_ost_idx] ||
 299                     !lov->lov_tgts[loi->loi_ost_idx]->ltd_active) {
 300                         CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
 301                         continue;
 302                 }
 303
 304                 OBD_ALLOC(req, sizeof(*req));
 305                 if (req == NULL)
 306                         GOTO(out_set, rc = -ENOMEM);
 307
 308                 req->rq_buflen = sizeof(*req->rq_oi.oi_md) +
 309                         sizeof(struct lov_oinfo *) +
 310                         sizeof(struct lov_oinfo);
 311                 OBD_ALLOC(req->rq_oi.oi_md, req->rq_buflen);
 312                 if (req->rq_oi.oi_md == NULL) {
 313                         OBD_FREE(req, sizeof(*req));
 314                         GOTO(out_set, rc = -ENOMEM);
 315                 }
 316                 req->rq_oi.oi_md->lsm_oinfo[0] =
 317                         ((void *)req->rq_oi.oi_md) + sizeof(*req->rq_oi.oi_md) +
 318                         sizeof(struct lov_oinfo *);
 319
 320
 321                 req->rq_rqset = set;
 322                 /* Set lov request specific parameters. */
 323                 req->rq_oi.oi_lockh = set->set_lockh->llh_handles + i;
 324                 req->rq_oi.oi_cb_up = cb_update_enqueue;
 325                 req->rq_oi.oi_flags = oinfo->oi_flags;
 326
 327                 LASSERT(req->rq_oi.oi_lockh);
 328
 329                 req->rq_oi.oi_policy.l_extent.gid =
 330                         oinfo->oi_policy.l_extent.gid;
 331                 req->rq_oi.oi_policy.l_extent.start = start;
 332                 req->rq_oi.oi_policy.l_extent.end = end;
 333
 334                 req->rq_idx = loi->loi_ost_idx;
 335                 req->rq_stripe = i;
 336
 337                 /* XXX LOV STACKING: submd should be from the subobj */
 338                 req->rq_oi.oi_md->lsm_object_id = loi->loi_id;
 339                 req->rq_oi.oi_md->lsm_object_gr = oinfo->oi_md->lsm_object_gr;
 340                 req->rq_oi.oi_md->lsm_stripe_count = 0;
 341                 req->rq_oi.oi_md->lsm_oinfo[0]->loi_kms_valid =
 342                         loi->loi_kms_valid;
 343                 req->rq_oi.oi_md->lsm_oinfo[0]->loi_kms = loi->loi_kms;
 344                 req->rq_oi.oi_md->lsm_oinfo[0]->loi_lvb = loi->loi_lvb;
 345
 346                 lov_set_add_req(req, set);
 347         }
 348         if (!set->set_count)
 349                 GOTO(out_set, rc = -EIO);
 350         *reqset = set;
 351         RETURN(0);
 352 out_set:
 353         lov_fini_enqueue_set(set, einfo->ei_mode, rc, NULL);
 354         RETURN(rc);
 355 }
 356
 357 int lov_update_match_set(struct lov_request_set *set, struct lov_request *req,
 358                          int rc)
 359 {
 360         int ret = rc;
 361         ENTRY;
 362
 363         if (rc > 0)
 364                 ret = 0;
 365         else if (rc == 0)
 366                 ret = 1;
 367         lov_update_set(set, req, ret);
 368         RETURN(rc);
 369 }
 370
 371 int lov_fini_match_set(struct lov_request_set *set, __u32 mode, int flags)
 372 {
 373         int rc = 0;
 374         ENTRY;
 375
 376         if (set == NULL)
 377                 RETURN(0);
 378         LASSERT(set->set_exp);
 379         rc = enqueue_done(set, mode);
 380         if ((set->set_count == set->set_success) &&
 381             (flags & LDLM_FL_TEST_LOCK))
 382                 lov_llh_put(set->set_lockh);
 383
 384         if (atomic_dec_and_test(&set->set_refcount))
 385                 lov_finish_set(set);
 386
 387         RETURN(rc);
 388 }
 389
 390 int lov_prep_match_set(struct obd_export *exp, struct obd_info *oinfo,
 391                        struct lov_stripe_md *lsm, ldlm_policy_data_t *policy,
 392                        __u32 mode, struct lustre_handle *lockh,
 393                        struct lov_request_set **reqset)
 394 {
 395         struct lov_obd *lov = &exp->exp_obd->u.lov;
 396         struct lov_request_set *set;
 397         int i, rc = 0;
 398         ENTRY;
 399
 400         OBD_ALLOC(set, sizeof(*set));
 401         if (set == NULL)
 402                 RETURN(-ENOMEM);
 403         lov_init_set(set);
 404
 405         set->set_exp = exp;
 406         set->set_oi = oinfo;
 407         set->set_oi->oi_md = lsm;
 408         set->set_lockh = lov_llh_new(lsm);
 409         if (set->set_lockh == NULL)
 410                 GOTO(out_set, rc = -ENOMEM);
 411         lockh->cookie = set->set_lockh->llh_handle.h_cookie;
 412
 413         for (i = 0; i < lsm->lsm_stripe_count; i++){
 414                 struct lov_oinfo *loi;
 415                 struct lov_request *req;
 416                 obd_off start, end;
 417
 418                 loi = lsm->lsm_oinfo[i];
 419                 if (!lov_stripe_intersects(lsm, i, policy->l_extent.start,
 420                                            policy->l_extent.end, &start, &end))
 421                         continue;
 422
 423                 /* FIXME raid1 should grace this error */
 424                 if (!lov->lov_tgts[loi->loi_ost_idx] ||
 425                     !lov->lov_tgts[loi->loi_ost_idx]->ltd_active) {
 426                         CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
 427                         GOTO(out_set, rc = -EIO);
 428                 }
 429
 430                 OBD_ALLOC(req, sizeof(*req));
 431                 if (req == NULL)
 432                         GOTO(out_set, rc = -ENOMEM);
 433
 434                 req->rq_buflen = sizeof(*req->rq_oi.oi_md);
 435                 OBD_ALLOC(req->rq_oi.oi_md, req->rq_buflen);
 436                 if (req->rq_oi.oi_md == NULL) {
 437                         OBD_FREE(req, sizeof(*req));
 438                         GOTO(out_set, rc = -ENOMEM);
 439                 }
 440
 441                 req->rq_oi.oi_policy.l_extent.start = start;
 442                 req->rq_oi.oi_policy.l_extent.end = end;
 443                 req->rq_oi.oi_policy.l_extent.gid = policy->l_extent.gid;
 444
 445                 req->rq_idx = loi->loi_ost_idx;
 446                 req->rq_stripe = i;
 447
 448                 /* XXX LOV STACKING: submd should be from the subobj */
 449                 req->rq_oi.oi_md->lsm_object_id = loi->loi_id;
 450                 req->rq_oi.oi_md->lsm_object_gr = lsm->lsm_object_gr;
 451                 req->rq_oi.oi_md->lsm_stripe_count = 0;
 452
 453                 lov_set_add_req(req, set);
 454         }
 455         if (!set->set_count)
 456                 GOTO(out_set, rc = -EIO);
 457         *reqset = set;
 458         RETURN(rc);
 459 out_set:
 460         lov_fini_match_set(set, mode, 0);
 461         RETURN(rc);
 462 }
 463
 464 int lov_fini_cancel_set(struct lov_request_set *set)
 465 {
 466         int rc = 0;
 467         ENTRY;
 468
 469         if (set == NULL)
 470                 RETURN(0);
 471
 472         LASSERT(set->set_exp);
 473         if (set->set_lockh)
 474                 lov_llh_put(set->set_lockh);
 475
 476         if (atomic_dec_and_test(&set->set_refcount))
 477                 lov_finish_set(set);
 478
 479         RETURN(rc);
 480 }
 481
 482 int lov_prep_cancel_set(struct obd_export *exp, struct obd_info *oinfo,
 483                         struct lov_stripe_md *lsm, __u32 mode,
 484                         struct lustre_handle *lockh,
 485                         struct lov_request_set **reqset)
 486 {
 487         struct lov_request_set *set;
 488         int i, rc = 0;
 489         ENTRY;
 490
 491         OBD_ALLOC(set, sizeof(*set));
 492         if (set == NULL)
 493                 RETURN(-ENOMEM);
 494         lov_init_set(set);
 495
 496         set->set_exp = exp;
 497         set->set_oi = oinfo;
 498         set->set_oi->oi_md = lsm;
 499         set->set_lockh = lov_handle2llh(lockh);
 500         if (set->set_lockh == NULL) {
 501                 CERROR("LOV: invalid lov lock handle %p\n", lockh);
 502                 GOTO(out_set, rc = -EINVAL);
 503         }
 504         lockh->cookie = set->set_lockh->llh_handle.h_cookie;
 505
 506         for (i = 0; i < lsm->lsm_stripe_count; i++){
 507                 struct lov_request *req;
 508                 struct lustre_handle *lov_lockhp;
 509                 struct lov_oinfo *loi = lsm->lsm_oinfo[i];
 510
 511                 lov_lockhp = set->set_lockh->llh_handles + i;
 512                 if (!lustre_handle_is_used(lov_lockhp)) {
 513                         CDEBUG(D_RPCTRACE,"lov idx %d subobj "LPX64" no lock\n",
 514                                loi->loi_ost_idx, loi->loi_id);
 515                         continue;
 516                 }
 517
 518                 OBD_ALLOC(req, sizeof(*req));
 519                 if (req == NULL)
 520                         GOTO(out_set, rc = -ENOMEM);
 521
 522                 req->rq_buflen = sizeof(*req->rq_oi.oi_md);
 523                 OBD_ALLOC(req->rq_oi.oi_md, req->rq_buflen);
 524                 if (req->rq_oi.oi_md == NULL) {
 525                         OBD_FREE(req, sizeof(*req));
 526                         GOTO(out_set, rc = -ENOMEM);
 527                 }
 528
 529                 req->rq_idx = loi->loi_ost_idx;
 530                 req->rq_stripe = i;
 531
 532                 /* XXX LOV STACKING: submd should be from the subobj */
 533                 req->rq_oi.oi_md->lsm_object_id = loi->loi_id;
 534                 req->rq_oi.oi_md->lsm_object_gr = lsm->lsm_object_gr;
 535                 req->rq_oi.oi_md->lsm_stripe_count = 0;
 536
 537                 lov_set_add_req(req, set);
 538         }
 539         if (!set->set_count)
 540                 GOTO(out_set, rc = -EIO);
 541         *reqset = set;
 542         RETURN(rc);
 543 out_set:
 544         lov_fini_cancel_set(set);
 545         RETURN(rc);
 546 }
 547
 548 static int create_done(struct obd_export *exp, struct lov_request_set *set,
 549                        struct lov_stripe_md **lsmp)
 550 {
 551         struct lov_obd *lov = &exp->exp_obd->u.lov;
 552         struct obd_trans_info *oti = set->set_oti;
 553         struct obdo *src_oa = set->set_oi->oi_oa;
 554         struct lov_request *req;
 555         struct obdo *ret_oa = NULL;
 556         int attrset = 0, rc = 0;
 557         ENTRY;
 558
 559         LASSERT(set->set_completes);
 560
 561         /* try alloc objects on other osts if osc_create fails for
 562          * exceptions: RPC failure, ENOSPC, etc */
 563         if (set->set_count != set->set_success) {
 564                 list_for_each_entry (req, &set->set_list, rq_link) {
 565                         if (req->rq_rc == 0)
 566                                 continue;
 567
 568                         set->set_completes--;
 569                         req->rq_complete = 0;
 570
 571                         rc = qos_remedy_create(set, req);
 572                         lov_update_create_set(set, req, rc);
 573
 574                         if (rc)
 575                                 break;
 576                 }
 577         }
 578
 579         /* no successful creates */
 580         if (set->set_success == 0)
 581                 GOTO(cleanup, rc);
 582
 583         /* If there was an explicit stripe set, fail.  Otherwise, we
 584          * got some objects and that's not bad. */
 585         if (set->set_count != set->set_success) {
 586                 if (*lsmp)
 587                         GOTO(cleanup, rc);
 588                 set->set_count = set->set_success;
 589                 qos_shrink_lsm(set);
 590         }
 591
 592         OBDO_ALLOC(ret_oa);
 593         if (ret_oa == NULL)
 594                 GOTO(cleanup, rc = -ENOMEM);
 595
 596         list_for_each_entry(req, &set->set_list, rq_link) {
 597                 if (!req->rq_complete || req->rq_rc)
 598                         continue;
 599                 lov_merge_attrs(ret_oa, req->rq_oi.oi_oa,
 600                                 req->rq_oi.oi_oa->o_valid, set->set_oi->oi_md,
 601                                 req->rq_stripe, &attrset);
 602         }
 603         if (src_oa->o_valid & OBD_MD_FLSIZE &&
 604             ret_oa->o_size != src_oa->o_size) {
 605                 CERROR("original size "LPU64" isn't new object size "LPU64"\n",
 606                        src_oa->o_size, ret_oa->o_size);
 607                 LBUG();
 608         }
 609         ret_oa->o_id = src_oa->o_id;
 610         ret_oa->o_gr = src_oa->o_gr;
 611         ret_oa->o_valid |= OBD_MD_FLID | OBD_MD_FLGROUP;
 612         memcpy(src_oa, ret_oa, sizeof(*src_oa));
 613         OBDO_FREE(ret_oa);
 614
 615         *lsmp = set->set_oi->oi_md;
 616         GOTO(done, rc = 0);
 617
 618 cleanup:
 619         list_for_each_entry(req, &set->set_list, rq_link) {
 620                 struct obd_export *sub_exp;
 621                 int err = 0;
 622
 623                 if (!req->rq_complete || req->rq_rc)
 624                         continue;
 625
 626                 sub_exp = lov->lov_tgts[req->rq_idx]->ltd_exp;
 627                 err = obd_destroy(sub_exp, req->rq_oi.oi_oa, NULL, oti, NULL);
 628                 if (err)
 629                         CERROR("Failed to uncreate objid "LPX64" subobj "
 630                                LPX64" on OST idx %d: rc = %d\n",
 631                                src_oa->o_id, req->rq_oi.oi_oa->o_id,
 632                                req->rq_idx, rc);
 633         }
 634         if (*lsmp == NULL)
 635                 obd_free_memmd(exp, &set->set_oi->oi_md);
 636 done:
 637         if (oti && set->set_cookies) {
 638                 oti->oti_logcookies = set->set_cookies;
 639                 if (!set->set_cookie_sent) {
 640                         oti_free_cookies(oti);
 641                         src_oa->o_valid &= ~OBD_MD_FLCOOKIE;
 642                 } else {
 643                         src_oa->o_valid |= OBD_MD_FLCOOKIE;
 644                 }
 645         }
 646         RETURN(rc);
 647 }
 648
 649 int lov_fini_create_set(struct lov_request_set *set,struct lov_stripe_md **lsmp)
 650 {
 651         int rc = 0;
 652         ENTRY;
 653
 654         if (set == NULL)
 655                 RETURN(0);
 656         LASSERT(set->set_exp);
 657         if (set->set_completes)
 658                 rc = create_done(set->set_exp, set, lsmp);
 659
 660         if (atomic_dec_and_test(&set->set_refcount))
 661                 lov_finish_set(set);
 662
 663         RETURN(rc);
 664 }
 665
 666 int lov_update_create_set(struct lov_request_set *set,
 667                           struct lov_request *req, int rc)
 668 {
 669         struct obd_trans_info *oti = set->set_oti;
 670         struct lov_stripe_md *lsm = set->set_oi->oi_md;
 671         struct lov_oinfo *loi;
 672         struct lov_obd *lov = &set->set_exp->exp_obd->u.lov;
 673         ENTRY;
 674
 675         req->rq_stripe = set->set_success;
 676         loi = lsm->lsm_oinfo[req->rq_stripe];
 677
 678         if (rc && lov->lov_tgts[req->rq_idx] &&
 679             lov->lov_tgts[req->rq_idx]->ltd_active) {
 680                 CERROR("error creating fid "LPX64" sub-object"
 681                        " on OST idx %d/%d: rc = %d\n",
 682                        set->set_oi->oi_oa->o_id, req->rq_idx,
 683                        lsm->lsm_stripe_count, rc);
 684                 if (rc > 0) {
 685                         CERROR("obd_create returned invalid err %d\n", rc);
 686                         rc = -EIO;
 687                 }
 688         }
 689         lov_update_set(set, req, rc);
 690         if (rc)
 691                 RETURN(rc);
 692
 693         loi->loi_id = req->rq_oi.oi_oa->o_id;
 694         loi->loi_gr = req->rq_oi.oi_oa->o_gr;
 695         loi->loi_ost_idx = req->rq_idx;
 696         CDEBUG(D_INODE, "objid "LPX64" has subobj "LPX64"/"LPU64" at idx %d\n",
 697                lsm->lsm_object_id, loi->loi_id, loi->loi_id, req->rq_idx);
 698         loi_init(loi);
 699
 700         if (oti && set->set_cookies)
 701                 ++oti->oti_logcookies;
 702         if (req->rq_oi.oi_oa->o_valid & OBD_MD_FLCOOKIE)
 703                 set->set_cookie_sent++;
 704
 705         RETURN(0);
 706 }
 707
 708 int lov_prep_create_set(struct obd_export *exp, struct obd_info *oinfo,
 709                         struct lov_stripe_md **lsmp, struct obdo *src_oa,
 710                         struct obd_trans_info *oti,
 711                         struct lov_request_set **reqset)
 712 {
 713         struct lov_request_set *set;
 714         int rc = 0;
 715         ENTRY;
 716
 717         OBD_ALLOC(set, sizeof(*set));
 718         if (set == NULL)
 719                 RETURN(-ENOMEM);
 720         lov_init_set(set);
 721
 722         set->set_exp = exp;
 723         set->set_oi = oinfo;
 724         set->set_oi->oi_md = *lsmp;
 725         set->set_oi->oi_oa = src_oa;
 726         set->set_oti = oti;
 727
 728         rc = qos_prep_create(exp, set);
 729         if (rc)
 730                 lov_fini_create_set(set, lsmp);
 731         else
 732                 *reqset = set;
 733         RETURN(rc);
 734 }
 735
 736 static int common_attr_done(struct lov_request_set *set)
 737 {
 738         struct list_head *pos;
 739         struct lov_request *req;
 740         struct obdo *tmp_oa;
 741         int rc = 0, attrset = 0;
 742         ENTRY;
 743
 744         LASSERT(set->set_oi != NULL);
 745
 746         if (set->set_oi->oi_oa == NULL)
 747                 RETURN(0);
 748
 749         if (!set->set_success)
 750                 RETURN(-EIO);
 751
 752         OBDO_ALLOC(tmp_oa);
 753         if (tmp_oa == NULL)
 754                 GOTO(out, rc = -ENOMEM);
 755
 756         list_for_each (pos, &set->set_list) {
 757                 req = list_entry(pos, struct lov_request, rq_link);
 758
 759                 if (!req->rq_complete || req->rq_rc)
 760                         continue;
 761                 if (req->rq_oi.oi_oa->o_valid == 0)   /* inactive stripe */
 762                         continue;
 763                 lov_merge_attrs(tmp_oa, req->rq_oi.oi_oa,
 764                                 req->rq_oi.oi_oa->o_valid,
 765                                 set->set_oi->oi_md, req->rq_stripe, &attrset);
 766         }
 767         if (!attrset) {
 768                 CERROR("No stripes had valid attrs\n");
 769                 rc = -EIO;
 770         }
 771         tmp_oa->o_id = set->set_oi->oi_oa->o_id;
 772         memcpy(set->set_oi->oi_oa, tmp_oa, sizeof(*set->set_oi->oi_oa));
 773 out:
 774         if (tmp_oa)
 775                 OBDO_FREE(tmp_oa);
 776         RETURN(rc);
 777
 778 }
 779
 780 static int brw_done(struct lov_request_set *set)
 781 {
 782         struct lov_stripe_md *lsm = set->set_oi->oi_md;
 783         struct lov_oinfo     *loi = NULL;
 784         struct list_head *pos;
 785         struct lov_request *req;
 786         ENTRY;
 787
 788         list_for_each (pos, &set->set_list) {
 789                 req = list_entry(pos, struct lov_request, rq_link);
 790
 791                 if (!req->rq_complete || req->rq_rc)
 792                         continue;
 793
 794                 loi = lsm->lsm_oinfo[req->rq_stripe];
 795
 796                 if (req->rq_oi.oi_oa->o_valid & OBD_MD_FLBLOCKS)
 797                         loi->loi_lvb.lvb_blocks = req->rq_oi.oi_oa->o_blocks;
 798         }
 799
 800         RETURN(0);
 801 }
 802
 803 int lov_fini_brw_set(struct lov_request_set *set)
 804 {
 805         int rc = 0;
 806         ENTRY;
 807
 808         if (set == NULL)
 809                 RETURN(0);
 810         LASSERT(set->set_exp);
 811         if (set->set_completes) {
 812                 rc = brw_done(set);
 813                 /* FIXME update qos data here */
 814         }
 815         if (atomic_dec_and_test(&set->set_refcount))
 816                 lov_finish_set(set);
 817
 818         RETURN(rc);
 819 }
 820
 821 int lov_prep_brw_set(struct obd_export *exp, struct obd_info *oinfo,
 822                      obd_count oa_bufs, struct brw_page *pga,
 823                      struct obd_trans_info *oti,
 824                      struct lov_request_set **reqset)
 825 {
 826         struct {
 827                 obd_count       index;
 828                 obd_count       count;
 829                 obd_count       off;
 830         } *info = NULL;
 831         struct lov_request_set *set;
 832         struct lov_obd *lov = &exp->exp_obd->u.lov;
 833         int rc = 0, i, shift;
 834         ENTRY;
 835
 836         OBD_ALLOC(set, sizeof(*set));
 837         if (set == NULL)
 838                 RETURN(-ENOMEM);
 839         lov_init_set(set);
 840
 841         set->set_exp = exp;
 842         set->set_oti = oti;
 843         set->set_oi = oinfo;
 844         set->set_oabufs = oa_bufs;
 845         OBD_ALLOC(set->set_pga, oa_bufs * sizeof(*set->set_pga));
 846         if (!set->set_pga)
 847                 GOTO(out, rc = -ENOMEM);
 848
 849         OBD_ALLOC(info, sizeof(*info) * oinfo->oi_md->lsm_stripe_count);
 850         if (!info)
 851                 GOTO(out, rc = -ENOMEM);
 852
 853         /* calculate the page count for each stripe */
 854         for (i = 0; i < oa_bufs; i++) {
 855                 int stripe = lov_stripe_number(oinfo->oi_md, pga[i].off);
 856                 info[stripe].count++;
 857         }
 858
 859         /* alloc and initialize lov request */
 860         shift = 0;
 861         for (i = 0; i < oinfo->oi_md->lsm_stripe_count; i++){
 862                 struct lov_oinfo *loi = NULL;
 863                 struct lov_request *req;
 864
 865                 if (info[i].count == 0)
 866                         continue;
 867
 868                 loi = oinfo->oi_md->lsm_oinfo[i];
 869                 if (!lov->lov_tgts[loi->loi_ost_idx] ||
 870                     !lov->lov_tgts[loi->loi_ost_idx]->ltd_active) {
 871                         CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
 872                         GOTO(out, rc = -EIO);
 873                 }
 874
 875                 OBD_ALLOC(req, sizeof(*req));
 876                 if (req == NULL)
 877                         GOTO(out, rc = -ENOMEM);
 878
 879                 OBDO_ALLOC(req->rq_oi.oi_oa);
 880                 if (req->rq_oi.oi_oa == NULL) {
 881                         OBD_FREE(req, sizeof(*req));
 882                         GOTO(out, rc = -ENOMEM);
 883                 }
 884
 885                 if (oinfo->oi_oa) {
 886                         memcpy(req->rq_oi.oi_oa, oinfo->oi_oa,
 887                                sizeof(*req->rq_oi.oi_oa));
 888                 }
 889                 req->rq_oi.oi_oa->o_id = loi->loi_id;
 890                 req->rq_oi.oi_oa->o_stripe_idx = i;
 891
 892                 req->rq_buflen = sizeof(*req->rq_oi.oi_md);
 893                 OBD_ALLOC(req->rq_oi.oi_md, req->rq_buflen);
 894                 if (req->rq_oi.oi_md == NULL) {
 895                         OBDO_FREE(req->rq_oi.oi_oa);
 896                         OBD_FREE(req, sizeof(*req));
 897                         GOTO(out, rc = -ENOMEM);
 898                 }
 899
 900                 req->rq_idx = loi->loi_ost_idx;
 901                 req->rq_stripe = i;
 902
 903                 /* XXX LOV STACKING */
 904                 req->rq_oi.oi_md->lsm_object_id = loi->loi_id;
 905                 req->rq_oi.oi_md->lsm_object_gr = oinfo->oi_md->lsm_object_gr;
 906                 req->rq_oabufs = info[i].count;
 907                 req->rq_pgaidx = shift;
 908                 shift += req->rq_oabufs;
 909
 910                 /* remember the index for sort brw_page array */
 911                 info[i].index = req->rq_pgaidx;
 912
 913                 req->rq_oi.oi_capa = oinfo->oi_capa;
 914
 915                 lov_set_add_req(req, set);
 916         }
 917         if (!set->set_count)
 918                 GOTO(out, rc = -EIO);
 919
 920         /* rotate & sort the brw_page array */
 921         for (i = 0; i < oa_bufs; i++) {
 922                 int stripe = lov_stripe_number(oinfo->oi_md, pga[i].off);
 923
 924                 shift = info[stripe].index + info[stripe].off;
 925                 LASSERT(shift < oa_bufs);
 926                 set->set_pga[shift] = pga[i];
 927                 lov_stripe_offset(oinfo->oi_md, pga[i].off, stripe,
 928                                   &set->set_pga[shift].off);
 929                 info[stripe].off++;
 930         }
 931 out:
 932         if (info)
 933                 OBD_FREE(info, sizeof(*info) * oinfo->oi_md->lsm_stripe_count);
 934
 935         if (rc == 0)
 936                 *reqset = set;
 937         else
 938                 lov_fini_brw_set(set);
 939
 940         RETURN(rc);
 941 }
 942
 943 int lov_fini_getattr_set(struct lov_request_set *set)
 944 {
 945         int rc = 0;
 946         ENTRY;
 947
 948         if (set == NULL)
 949                 RETURN(0);
 950         LASSERT(set->set_exp);
 951         if (set->set_completes)
 952                 rc = common_attr_done(set);
 953
 954         if (atomic_dec_and_test(&set->set_refcount))
 955                 lov_finish_set(set);
 956
 957         RETURN(rc);
 958 }
 959
 960 /* The callback for osc_getattr_async that finilizes a request info when a
 961  * response is recieved. */
 962 static int cb_getattr_update(struct obd_info *oinfo, int rc)
 963 {
 964         struct lov_request *lovreq;
 965         lovreq = container_of(oinfo, struct lov_request, rq_oi);
 966         return lov_update_common_set(lovreq->rq_rqset, lovreq, rc);
 967 }
 968
 969 int lov_prep_getattr_set(struct obd_export *exp, struct obd_info *oinfo,
 970                          struct lov_request_set **reqset)
 971 {
 972         struct lov_request_set *set;
 973         struct lov_obd *lov = &exp->exp_obd->u.lov;
 974         int rc = 0, i;
 975         ENTRY;
 976
 977         OBD_ALLOC(set, sizeof(*set));
 978         if (set == NULL)
 979                 RETURN(-ENOMEM);
 980         lov_init_set(set);
 981
 982         set->set_exp = exp;
 983         set->set_oi = oinfo;
 984
 985         for (i = 0; i < oinfo->oi_md->lsm_stripe_count; i++) {
 986                 struct lov_oinfo *loi;
 987                 struct lov_request *req;
 988
 989                 loi = oinfo->oi_md->lsm_oinfo[i];
 990                 if (!lov->lov_tgts[loi->loi_ost_idx] ||
 991                     !lov->lov_tgts[loi->loi_ost_idx]->ltd_active) {
 992                         CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
 993                         continue;
 994                 }
 995
 996                 OBD_ALLOC(req, sizeof(*req));
 997                 if (req == NULL)
 998                         GOTO(out_set, rc = -ENOMEM);
 999
1000                 req->rq_stripe = i;
1001                 req->rq_idx = loi->loi_ost_idx;
1002
1003                 OBDO_ALLOC(req->rq_oi.oi_oa);
1004                 if (req->rq_oi.oi_oa == NULL) {
1005                         OBD_FREE(req, sizeof(*req));
1006                         GOTO(out_set, rc = -ENOMEM);
1007                 }
1008                 memcpy(req->rq_oi.oi_oa, oinfo->oi_oa,
1009                        sizeof(*req->rq_oi.oi_oa));
1010                 req->rq_oi.oi_oa->o_id = loi->loi_id;
1011                 req->rq_oi.oi_cb_up = cb_getattr_update;
1012                 req->rq_oi.oi_capa = oinfo->oi_capa;
1013                 req->rq_rqset = set;
1014
1015                 lov_set_add_req(req, set);
1016         }
1017         if (!set->set_count)
1018                 GOTO(out_set, rc = -EIO);
1019         *reqset = set;
1020         RETURN(rc);
1021 out_set:
1022         lov_fini_getattr_set(set);
1023         RETURN(rc);
1024 }
1025
1026 int lov_fini_destroy_set(struct lov_request_set *set)
1027 {
1028         ENTRY;
1029
1030         if (set == NULL)
1031                 RETURN(0);
1032         LASSERT(set->set_exp);
1033         if (set->set_completes) {
1034                 /* FIXME update qos data here */
1035         }
1036
1037         if (atomic_dec_and_test(&set->set_refcount))
1038                 lov_finish_set(set);
1039
1040         RETURN(0);
1041 }
1042
1043 int lov_prep_destroy_set(struct obd_export *exp, struct obd_info *oinfo,
1044                          struct obdo *src_oa, struct lov_stripe_md *lsm,
1045                          struct obd_trans_info *oti,
1046                          struct lov_request_set **reqset)
1047 {
1048         struct lov_request_set *set;
1049         struct lov_obd *lov = &exp->exp_obd->u.lov;
1050         int rc = 0, i;
1051         ENTRY;
1052
1053         OBD_ALLOC(set, sizeof(*set));
1054         if (set == NULL)
1055                 RETURN(-ENOMEM);
1056         lov_init_set(set);
1057
1058         set->set_exp = exp;
1059         set->set_oi = oinfo;
1060         set->set_oi->oi_md = lsm;
1061         set->set_oi->oi_oa = src_oa;
1062         set->set_oti = oti;
1063         if (oti != NULL && src_oa->o_valid & OBD_MD_FLCOOKIE)
1064                 set->set_cookies = oti->oti_logcookies;
1065
1066         for (i = 0; i < lsm->lsm_stripe_count; i++) {
1067                 struct lov_oinfo *loi;
1068                 struct lov_request *req;
1069
1070                 loi = lsm->lsm_oinfo[i];
1071                 if (!lov->lov_tgts[loi->loi_ost_idx] ||
1072                     !lov->lov_tgts[loi->loi_ost_idx]->ltd_active) {
1073                         CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
1074                         continue;
1075                 }
1076
1077                 OBD_ALLOC(req, sizeof(*req));
1078                 if (req == NULL)
1079                         GOTO(out_set, rc = -ENOMEM);
1080
1081                 req->rq_stripe = i;
1082                 req->rq_idx = loi->loi_ost_idx;
1083
1084                 OBDO_ALLOC(req->rq_oi.oi_oa);
1085                 if (req->rq_oi.oi_oa == NULL) {
1086                         OBD_FREE(req, sizeof(*req));
1087                         GOTO(out_set, rc = -ENOMEM);
1088                 }
1089                 memcpy(req->rq_oi.oi_oa, src_oa, sizeof(*req->rq_oi.oi_oa));
1090                 req->rq_oi.oi_oa->o_id = loi->loi_id;
1091                 lov_set_add_req(req, set);
1092         }
1093         if (!set->set_count)
1094                 GOTO(out_set, rc = -EIO);
1095         *reqset = set;
1096         RETURN(rc);
1097 out_set:
1098         lov_fini_destroy_set(set);
1099         RETURN(rc);
1100 }
1101
1102 int lov_fini_setattr_set(struct lov_request_set *set)
1103 {
1104         int rc = 0;
1105         ENTRY;
1106
1107         if (set == NULL)
1108                 RETURN(0);
1109         LASSERT(set->set_exp);
1110         if (set->set_completes) {
1111                 rc = common_attr_done(set);
1112                 /* FIXME update qos data here */
1113         }
1114
1115         if (atomic_dec_and_test(&set->set_refcount))
1116                 lov_finish_set(set);
1117         RETURN(rc);
1118 }
1119
1120 int lov_update_setattr_set(struct lov_request_set *set,
1121                            struct lov_request *req, int rc)
1122 {
1123         struct lov_obd *lov = &req->rq_rqset->set_exp->exp_obd->u.lov;
1124         struct lov_stripe_md *lsm = req->rq_rqset->set_oi->oi_md;
1125         ENTRY;
1126
1127         lov_update_set(set, req, rc);
1128
1129         /* grace error on inactive ost */
1130         if (rc && !(lov->lov_tgts[req->rq_idx] &&
1131                     lov->lov_tgts[req->rq_idx]->ltd_active))
1132                 rc = 0;
1133
1134         if (rc == 0) {
1135                 if (req->rq_oi.oi_oa->o_valid & OBD_MD_FLCTIME)
1136                         lsm->lsm_oinfo[req->rq_stripe]->loi_lvb.lvb_ctime =
1137                                 req->rq_oi.oi_oa->o_ctime;
1138                 if (req->rq_oi.oi_oa->o_valid & OBD_MD_FLMTIME)
1139                         lsm->lsm_oinfo[req->rq_stripe]->loi_lvb.lvb_mtime =
1140                                 req->rq_oi.oi_oa->o_mtime;
1141                 if (req->rq_oi.oi_oa->o_valid & OBD_MD_FLATIME)
1142                         lsm->lsm_oinfo[req->rq_stripe]->loi_lvb.lvb_atime =
1143                                 req->rq_oi.oi_oa->o_atime;
1144         }
1145
1146         RETURN(rc);
1147 }
1148
1149 /* The callback for osc_setattr_async that finilizes a request info when a
1150  * response is recieved. */
1151 static int cb_setattr_update(struct obd_info *oinfo, int rc)
1152 {
1153         struct lov_request *lovreq;
1154         lovreq = container_of(oinfo, struct lov_request, rq_oi);
1155         return lov_update_setattr_set(lovreq->rq_rqset, lovreq, rc);
1156 }
1157
1158 int lov_prep_setattr_set(struct obd_export *exp, struct obd_info *oinfo,
1159                          struct obd_trans_info *oti,
1160                          struct lov_request_set **reqset)
1161 {
1162         struct lov_request_set *set;
1163         struct lov_obd *lov = &exp->exp_obd->u.lov;
1164         int rc = 0, i;
1165         ENTRY;
1166
1167         OBD_ALLOC(set, sizeof(*set));
1168         if (set == NULL)
1169                 RETURN(-ENOMEM);
1170         lov_init_set(set);
1171
1172         set->set_exp = exp;
1173         set->set_oti = oti;
1174         set->set_oi = oinfo;
1175         if (oti != NULL && oinfo->oi_oa->o_valid & OBD_MD_FLCOOKIE)
1176                 set->set_cookies = oti->oti_logcookies;
1177
1178         for (i = 0; i < oinfo->oi_md->lsm_stripe_count; i++) {
1179                 struct lov_oinfo *loi = oinfo->oi_md->lsm_oinfo[i];
1180                 struct lov_request *req;
1181
1182                 if (!lov->lov_tgts[loi->loi_ost_idx] ||
1183                     !lov->lov_tgts[loi->loi_ost_idx]->ltd_active) {
1184                         CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
1185                         continue;
1186                 }
1187
1188                 OBD_ALLOC(req, sizeof(*req));
1189                 if (req == NULL)
1190                         GOTO(out_set, rc = -ENOMEM);
1191                 req->rq_stripe = i;
1192                 req->rq_idx = loi->loi_ost_idx;
1193
1194                 OBDO_ALLOC(req->rq_oi.oi_oa);
1195                 if (req->rq_oi.oi_oa == NULL) {
1196                         OBD_FREE(req, sizeof(*req));
1197                         GOTO(out_set, rc = -ENOMEM);
1198                 }
1199                 memcpy(req->rq_oi.oi_oa, oinfo->oi_oa,
1200                        sizeof(*req->rq_oi.oi_oa));
1201                 req->rq_oi.oi_oa->o_id = loi->loi_id;
1202                 LASSERT(!(req->rq_oi.oi_oa->o_valid & OBD_MD_FLGROUP)
1203                                 || req->rq_oi.oi_oa->o_gr>0);
1204                 req->rq_oi.oi_oa->o_stripe_idx = i;
1205                 req->rq_oi.oi_cb_up = cb_setattr_update;
1206                 req->rq_oi.oi_capa = oinfo->oi_capa;
1207                 req->rq_rqset = set;
1208
1209                 if (oinfo->oi_oa->o_valid & OBD_MD_FLSIZE) {
1210                         int off = lov_stripe_offset(oinfo->oi_md,
1211                                                     oinfo->oi_oa->o_size, i,
1212                                                     &req->rq_oi.oi_oa->o_size);
1213
1214                         if (off < 0 && req->rq_oi.oi_oa->o_size)
1215                                 req->rq_oi.oi_oa->o_size--;
1216
1217                         CDEBUG(D_INODE, "stripe %d has size "LPU64"/"LPU64"\n",
1218                                i, req->rq_oi.oi_oa->o_size,
1219                                oinfo->oi_oa->o_size);
1220                 }
1221                 lov_set_add_req(req, set);
1222         }
1223         if (!set->set_count)
1224                 GOTO(out_set, rc = -EIO);
1225         *reqset = set;
1226         RETURN(rc);
1227 out_set:
1228         lov_fini_setattr_set(set);
1229         RETURN(rc);
1230 }
1231
1232 int lov_fini_punch_set(struct lov_request_set *set)
1233 {
1234         int rc = 0;
1235         ENTRY;
1236
1237         if (set == NULL)
1238                 RETURN(0);
1239         LASSERT(set->set_exp);
1240         if (set->set_completes) {
1241                 rc = -EIO;
1242                 /* FIXME update qos data here */
1243                 if (set->set_success)
1244                         rc = common_attr_done(set);
1245         }
1246
1247         if (atomic_dec_and_test(&set->set_refcount))
1248                 lov_finish_set(set);
1249
1250         RETURN(rc);
1251 }
1252
1253 int lov_update_punch_set(struct lov_request_set *set,
1254                          struct lov_request *req, int rc)
1255 {
1256         struct lov_obd *lov = &req->rq_rqset->set_exp->exp_obd->u.lov;
1257         struct lov_stripe_md *lsm = req->rq_rqset->set_oi->oi_md;
1258         ENTRY;
1259
1260         lov_update_set(set, req, rc);
1261
1262         /* grace error on inactive ost */
1263         if (rc && !lov->lov_tgts[req->rq_idx]->ltd_active)
1264                 rc = 0;
1265
1266         if (rc == 0) {
1267                 lov_stripe_lock(lsm);
1268                 if (req->rq_oi.oi_oa->o_valid & OBD_MD_FLBLOCKS) {
1269                         lsm->lsm_oinfo[req->rq_stripe]->loi_lvb.lvb_blocks =
1270                                 req->rq_oi.oi_oa->o_blocks;
1271                 }
1272
1273                 /* Do we need to update lvb_size here? It needn't because
1274                  * it have been done in ll_truncate(). -jay */
1275                 lov_stripe_unlock(lsm);
1276         }
1277
1278         RETURN(rc);
1279 }
1280
1281 /* The callback for osc_punch that finilizes a request info when a response
1282  * is recieved. */
1283 static int cb_update_punch(struct obd_info *oinfo, int rc)
1284 {
1285         struct lov_request *lovreq;
1286         lovreq = container_of(oinfo, struct lov_request, rq_oi);
1287         return lov_update_punch_set(lovreq->rq_rqset, lovreq, rc);
1288 }
1289
1290 int lov_prep_punch_set(struct obd_export *exp, struct obd_info *oinfo,
1291                        struct obd_trans_info *oti,
1292                        struct lov_request_set **reqset)
1293 {
1294         struct lov_request_set *set;
1295         struct lov_obd *lov = &exp->exp_obd->u.lov;
1296         int rc = 0, i;
1297         ENTRY;
1298
1299         OBD_ALLOC(set, sizeof(*set));
1300         if (set == NULL)
1301                 RETURN(-ENOMEM);
1302         lov_init_set(set);
1303
1304         set->set_oi = oinfo;
1305         set->set_exp = exp;
1306
1307         for (i = 0; i < oinfo->oi_md->lsm_stripe_count; i++) {
1308                 struct lov_oinfo *loi = oinfo->oi_md->lsm_oinfo[i];
1309                 struct lov_request *req;
1310                 obd_off rs, re;
1311
1312                 if (!lov->lov_tgts[loi->loi_ost_idx] ||
1313                     !lov->lov_tgts[loi->loi_ost_idx]->ltd_active) {
1314                         CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
1315                         continue;
1316                 }
1317
1318                 if (!lov_stripe_intersects(oinfo->oi_md, i,
1319                                            oinfo->oi_policy.l_extent.start,
1320                                            oinfo->oi_policy.l_extent.end,
1321                                            &rs, &re))
1322                         continue;
1323
1324                 OBD_ALLOC(req, sizeof(*req));
1325                 if (req == NULL)
1326                         GOTO(out_set, rc = -ENOMEM);
1327                 req->rq_stripe = i;
1328                 req->rq_idx = loi->loi_ost_idx;
1329
1330                 OBDO_ALLOC(req->rq_oi.oi_oa);
1331                 if (req->rq_oi.oi_oa == NULL) {
1332                         OBD_FREE(req, sizeof(*req));
1333                         GOTO(out_set, rc = -ENOMEM);
1334                 }
1335                 memcpy(req->rq_oi.oi_oa, oinfo->oi_oa,
1336                        sizeof(*req->rq_oi.oi_oa));
1337                 req->rq_oi.oi_oa->o_id = loi->loi_id;
1338                 req->rq_oi.oi_oa->o_gr = loi->loi_gr;
1339                 req->rq_oi.oi_oa->o_valid |= OBD_MD_FLGROUP;
1340
1341                 req->rq_oi.oi_oa->o_stripe_idx = i;
1342                 req->rq_oi.oi_cb_up = cb_update_punch;
1343                 req->rq_rqset = set;
1344
1345                 req->rq_oi.oi_policy.l_extent.start = rs;
1346                 req->rq_oi.oi_policy.l_extent.end = re;
1347                 req->rq_oi.oi_policy.l_extent.gid = -1;
1348
1349                 req->rq_oi.oi_capa = oinfo->oi_capa;
1350
1351                 lov_set_add_req(req, set);
1352         }
1353         if (!set->set_count)
1354                 GOTO(out_set, rc = -EIO);
1355         *reqset = set;
1356         RETURN(rc);
1357 out_set:
1358         lov_fini_punch_set(set);
1359         RETURN(rc);
1360 }
1361
1362 int lov_fini_sync_set(struct lov_request_set *set)
1363 {
1364         int rc = 0;
1365         ENTRY;
1366
1367         if (set == NULL)
1368                 RETURN(0);
1369         LASSERT(set->set_exp);
1370         if (set->set_completes) {
1371                 if (!set->set_success)
1372                         rc = -EIO;
1373                 /* FIXME update qos data here */
1374         }
1375
1376         if (atomic_dec_and_test(&set->set_refcount))
1377                 lov_finish_set(set);
1378
1379         RETURN(rc);
1380 }
1381
1382 int lov_prep_sync_set(struct obd_export *exp, struct obd_info *oinfo,
1383                       struct obdo *src_oa, struct lov_stripe_md *lsm,
1384                       obd_off start, obd_off end,
1385                       struct lov_request_set **reqset)
1386 {
1387         struct lov_request_set *set;
1388         struct lov_obd *lov = &exp->exp_obd->u.lov;
1389         int rc = 0, i;
1390         ENTRY;
1391
1392         OBD_ALLOC(set, sizeof(*set));
1393         if (set == NULL)
1394                 RETURN(-ENOMEM);
1395         lov_init_set(set);
1396
1397         set->set_exp = exp;
1398         set->set_oi = oinfo;
1399         set->set_oi->oi_md = lsm;
1400         set->set_oi->oi_oa = src_oa;
1401
1402         for (i = 0; i < lsm->lsm_stripe_count; i++) {
1403                 struct lov_oinfo *loi = lsm->lsm_oinfo[i];
1404                 struct lov_request *req;
1405                 obd_off rs, re;
1406
1407                 if (!lov->lov_tgts[loi->loi_ost_idx] ||
1408                     !lov->lov_tgts[loi->loi_ost_idx]->ltd_active) {
1409                         CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
1410                         continue;
1411                 }
1412
1413                 if (!lov_stripe_intersects(lsm, i, start, end, &rs, &re))
1414                         continue;
1415
1416                 OBD_ALLOC(req, sizeof(*req));
1417                 if (req == NULL)
1418                         GOTO(out_set, rc = -ENOMEM);
1419                 req->rq_stripe = i;
1420                 req->rq_idx = loi->loi_ost_idx;
1421
1422                 OBDO_ALLOC(req->rq_oi.oi_oa);
1423                 if (req->rq_oi.oi_oa == NULL) {
1424                         OBD_FREE(req, sizeof(*req));
1425                         GOTO(out_set, rc = -ENOMEM);
1426                 }
1427                 memcpy(req->rq_oi.oi_oa, src_oa, sizeof(*req->rq_oi.oi_oa));
1428                 req->rq_oi.oi_oa->o_id = loi->loi_id;
1429                 req->rq_oi.oi_oa->o_stripe_idx = i;
1430
1431                 req->rq_oi.oi_policy.l_extent.start = rs;
1432                 req->rq_oi.oi_policy.l_extent.end = re;
1433                 req->rq_oi.oi_policy.l_extent.gid = -1;
1434
1435                 lov_set_add_req(req, set);
1436         }
1437         if (!set->set_count)
1438                 GOTO(out_set, rc = -EIO);
1439         *reqset = set;
1440         RETURN(rc);
1441 out_set:
1442         lov_fini_sync_set(set);
1443         RETURN(rc);
1444 }
1445
1446 #define LOV_U64_MAX ((__u64)~0ULL)
1447 #define LOV_SUM_MAX(tot, add)                                           \
1448         do {                                                            \
1449                 if ((tot) + (add) < (tot))                              \
1450                         (tot) = LOV_U64_MAX;                            \
1451                 else                                                    \
1452                         (tot) += (add);                                 \
1453         } while(0)
1454
1455 int lov_fini_statfs(struct obd_device *obd, struct obd_statfs *osfs,int success)
1456 {
1457         ENTRY;
1458
1459         if (success) {
1460                 __u32 expected_stripes = lov_get_stripecnt(&obd->u.lov, 0);
1461
1462                 if (osfs->os_files != LOV_U64_MAX)
1463                         do_div(osfs->os_files, expected_stripes);
1464                 if (osfs->os_ffree != LOV_U64_MAX)
1465                         do_div(osfs->os_ffree, expected_stripes);
1466
1467                 spin_lock(&obd->obd_osfs_lock);
1468                 memcpy(&obd->obd_osfs, osfs, sizeof(*osfs));
1469                 obd->obd_osfs_age = get_jiffies_64();
1470                 spin_unlock(&obd->obd_osfs_lock);
1471                 RETURN(0);
1472         }
1473
1474         RETURN(-EIO);
1475 }
1476
1477 int lov_fini_statfs_set(struct lov_request_set *set)
1478 {
1479         int rc = 0;
1480         ENTRY;
1481
1482         if (set == NULL)
1483                 RETURN(0);
1484
1485         if (set->set_completes) {
1486                 rc = lov_fini_statfs(set->set_obd, set->set_oi->oi_osfs,
1487                                      set->set_success);
1488         }
1489
1490         if (atomic_dec_and_test(&set->set_refcount))
1491                 lov_finish_set(set);
1492
1493         RETURN(rc);
1494 }
1495
1496 void lov_update_statfs(struct obd_device *obd, struct obd_statfs *osfs,
1497                        struct obd_statfs *lov_sfs, int success)
1498 {
1499         int shift = 0, quit = 0;
1500         __u64 tmp;
1501         spin_lock(&obd->obd_osfs_lock);
1502         memcpy(&obd->obd_osfs, lov_sfs, sizeof(*lov_sfs));
1503         obd->obd_osfs_age = get_jiffies_64();
1504         spin_unlock(&obd->obd_osfs_lock);
1505
1506         if (success == 0) {
1507                 memcpy(osfs, lov_sfs, sizeof(*lov_sfs));
1508         } else {
1509                 if (osfs->os_bsize != lov_sfs->os_bsize) {
1510                         /* assume all block sizes are always powers of 2 */
1511                         /* get the bits difference */
1512                         tmp = osfs->os_bsize | lov_sfs->os_bsize;
1513                         for (shift = 0; shift <= 64; ++shift) {
1514                                 if (tmp & 1) {
1515                                         if (quit)
1516                                                 break;
1517                                         else
1518                                                 quit = 1;
1519                                         shift = 0;
1520                                 }
1521                                 tmp >>= 1;
1522                         }
1523                 }
1524
1525                 if (osfs->os_bsize < lov_sfs->os_bsize) {
1526                         osfs->os_bsize = lov_sfs->os_bsize;
1527
1528                         osfs->os_bfree  >>= shift;
1529                         osfs->os_bavail >>= shift;
1530                         osfs->os_blocks >>= shift;
1531                 } else if (shift != 0) {
1532                         lov_sfs->os_bfree  >>= shift;
1533                         lov_sfs->os_bavail >>= shift;
1534                         lov_sfs->os_blocks >>= shift;
1535                 }
1536 #ifdef MIN_DF
1537                 /* Sandia requested that df (and so, statfs) only
1538                    returned minimal available space on
1539                    a single OST, so people would be able to
1540                    write this much data guaranteed. */
1541                 if (osfs->os_bavail > lov_sfs->os_bavail) {
1542                         /* Presumably if new bavail is smaller,
1543                            new bfree is bigger as well */
1544                         osfs->os_bfree = lov_sfs->os_bfree;
1545                         osfs->os_bavail = lov_sfs->os_bavail;
1546                 }
1547 #else
1548                 osfs->os_bfree += lov_sfs->os_bfree;
1549                 osfs->os_bavail += lov_sfs->os_bavail;
1550 #endif
1551                 osfs->os_blocks += lov_sfs->os_blocks;
1552                 /* XXX not sure about this one - depends on policy.
1553                  *   - could be minimum if we always stripe on all OBDs
1554                  *     (but that would be wrong for any other policy,
1555                  *     if one of the OBDs has no more objects left)
1556                  *   - could be sum if we stripe whole objects
1557                  *   - could be average, just to give a nice number
1558                  *
1559                  * To give a "reasonable" (if not wholly accurate)
1560                  * number, we divide the total number of free objects
1561                  * by expected stripe count (watch out for overflow).
1562                  */
1563                 LOV_SUM_MAX(osfs->os_files, lov_sfs->os_files);
1564                 LOV_SUM_MAX(osfs->os_ffree, lov_sfs->os_ffree);
1565         }
1566 }
1567
1568 /* The callback for osc_statfs_async that finilizes a request info when a
1569  * response is recieved. */
1570 static int cb_statfs_update(struct obd_info *oinfo, int rc)
1571 {
1572         struct lov_request *lovreq;
1573         struct obd_statfs *osfs, *lov_sfs;
1574         struct obd_device *obd;
1575         struct lov_obd *lov;
1576         int success;
1577         ENTRY;
1578
1579         lovreq = container_of(oinfo, struct lov_request, rq_oi);
1580         lov = &lovreq->rq_rqset->set_obd->u.lov;
1581         obd = class_exp2obd(lov->lov_tgts[lovreq->rq_idx]->ltd_exp);
1582
1583         osfs = lovreq->rq_rqset->set_oi->oi_osfs;
1584         lov_sfs = oinfo->oi_osfs;
1585
1586         success = lovreq->rq_rqset->set_success;
1587
1588         /* XXX: the same is done in lov_update_common_set, however
1589            lovset->set_exp is not initialized. */
1590         lov_update_set(lovreq->rq_rqset, lovreq, rc);
1591         if (rc) {
1592                 if (rc && !(lov->lov_tgts[lovreq->rq_idx] &&
1593                             lov->lov_tgts[lovreq->rq_idx]->ltd_active))
1594                         rc = 0;
1595                 RETURN(rc);
1596         }
1597
1598         lov_update_statfs(obd, osfs, lov_sfs, success);
1599         qos_update(lov);
1600
1601         RETURN(0);
1602 }
1603
1604 int lov_prep_statfs_set(struct obd_device *obd, struct obd_info *oinfo,
1605                         struct lov_request_set **reqset)
1606 {
1607         struct lov_request_set *set;
1608         struct lov_obd *lov = &obd->u.lov;
1609         int rc = 0, i;
1610         ENTRY;
1611
1612         OBD_ALLOC(set, sizeof(*set));
1613         if (set == NULL)
1614                 RETURN(-ENOMEM);
1615         lov_init_set(set);
1616
1617         set->set_obd = obd;
1618         set->set_oi = oinfo;
1619
1620         /* We only get block data from the OBD */
1621         for (i = 0; i < lov->desc.ld_tgt_count; i++) {
1622                 struct lov_request *req;
1623
1624                 if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_active) {
1625                         CDEBUG(D_HA, "lov idx %d inactive\n", i);
1626                         continue;
1627                 }
1628
1629                 OBD_ALLOC(req, sizeof(*req));
1630                 if (req == NULL)
1631                         GOTO(out_set, rc = -ENOMEM);
1632
1633                 OBD_ALLOC(req->rq_oi.oi_osfs, sizeof(*req->rq_oi.oi_osfs));
1634                 if (req->rq_oi.oi_osfs == NULL) {
1635                         OBD_FREE(req, sizeof(*req));
1636                         GOTO(out_set, rc = -ENOMEM);
1637                 }
1638
1639                 req->rq_idx = i;
1640                 req->rq_oi.oi_cb_up = cb_statfs_update;
1641                 req->rq_rqset = set;
1642
1643                 lov_set_add_req(req, set);
1644         }
1645         if (!set->set_count)
1646                 GOTO(out_set, rc = -EIO);
1647         *reqset = set;
1648         RETURN(rc);
1649 out_set:
1650         lov_fini_statfs_set(set);
1651         RETURN(rc);
1652 }