lustre/ptlrpc/target.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2012, Whamcloud, Inc.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * Lustre Common Target
  37  * These are common function for MDT and OST recovery-related functionality
  38  *
  39  *   Author: Mikhail Pershin <tappro@sun.com>
  40  */
  41
  42 #include <obd.h>
  43 #include <lustre_fsfilt.h>
  44 #include <obd_class.h>
  45
  46 /**
  47  * Update client data in last_rcvd file. An obd API
  48  */
  49 static int obt_client_data_update(struct obd_export *exp)
  50 {
  51         struct tg_export_data *ted = &exp->exp_target_data;
  52         struct obd_device_target *obt = &exp->exp_obd->u.obt;
  53         struct lu_target *lut = class_exp2tgt(exp);
  54         loff_t off = ted->ted_lr_off;
  55         int rc = 0;
  56
  57         rc = fsfilt_write_record(exp->exp_obd, obt->obt_rcvd_filp,
  58                                  ted->ted_lcd, sizeof(*ted->ted_lcd), &off, 0);
  59
  60         CDEBUG(D_INFO, "update client idx %u last_epoch %#x (%#x)\n",
  61                ted->ted_lr_idx, le32_to_cpu(ted->ted_lcd->lcd_last_epoch),
  62                le32_to_cpu(lut->lut_lsd.lsd_start_epoch));
  63
  64         return rc;
  65 }
  66
  67 /**
  68  * Update server data in last_rcvd file. An obd API
  69  */
  70 int obt_server_data_update(struct lu_target *lut, int force_sync)
  71 {
  72         struct obd_device_target *obt = &lut->lut_obd->u.obt;
  73         loff_t off = 0;
  74         int rc;
  75         ENTRY;
  76
  77         CDEBUG(D_SUPER,
  78                "%s: mount_count is "LPU64", last_transno is "LPU64"\n",
  79                lut->lut_lsd.lsd_uuid,
  80                le64_to_cpu(lut->lut_lsd.lsd_mount_count),
  81                le64_to_cpu(lut->lut_lsd.lsd_last_transno));
  82
  83         rc = fsfilt_write_record(lut->lut_obd, obt->obt_rcvd_filp,
  84                                  &lut->lut_lsd, sizeof(lut->lut_lsd),
  85                                  &off, force_sync);
  86         if (rc)
  87                 CERROR("error writing lr_server_data: rc = %d\n", rc);
  88
  89         RETURN(rc);
  90 }
  91
  92 /**
  93  * Update client epoch with server's one
  94  */
  95 void obt_client_epoch_update(struct obd_export *exp)
  96 {
  97         struct lsd_client_data *lcd = exp->exp_target_data.ted_lcd;
  98         struct lu_target *lut = class_exp2tgt(exp);
  99
 100         /** VBR: set client last_epoch to current epoch */
 101         if (le32_to_cpu(lcd->lcd_last_epoch) >=
 102             le32_to_cpu(lut->lut_lsd.lsd_start_epoch))
 103                 return;
 104         lcd->lcd_last_epoch = lut->lut_lsd.lsd_start_epoch;
 105         obt_client_data_update(exp);
 106 }
 107
 108 /**
 109  * Increment server epoch. An obd API
 110  */
 111 static void obt_boot_epoch_update(struct lu_target *lut)
 112 {
 113         struct obd_device *obd = lut->lut_obd;
 114         __u32 start_epoch;
 115         struct ptlrpc_request *req;
 116         cfs_list_t client_list;
 117
 118         cfs_spin_lock(&lut->lut_translock);
 119         start_epoch = lr_epoch(le64_to_cpu(lut->lut_last_transno)) + 1;
 120         lut->lut_last_transno = cpu_to_le64((__u64)start_epoch <<
 121                                             LR_EPOCH_BITS);
 122         lut->lut_lsd.lsd_start_epoch = cpu_to_le32(start_epoch);
 123         cfs_spin_unlock(&lut->lut_translock);
 124
 125         CFS_INIT_LIST_HEAD(&client_list);
 126         cfs_spin_lock(&obd->obd_recovery_task_lock);
 127         cfs_list_splice_init(&obd->obd_final_req_queue, &client_list);
 128         cfs_spin_unlock(&obd->obd_recovery_task_lock);
 129
 130         /**
 131          * go through list of exports participated in recovery and
 132          * set new epoch for them
 133          */
 134         cfs_list_for_each_entry(req, &client_list, rq_list) {
 135                 LASSERT(!req->rq_export->exp_delayed);
 136                 obt_client_epoch_update(req->rq_export);
 137         }
 138         /** return list back at once */
 139         cfs_spin_lock(&obd->obd_recovery_task_lock);
 140         cfs_list_splice_init(&client_list, &obd->obd_final_req_queue);
 141         cfs_spin_unlock(&obd->obd_recovery_task_lock);
 142         obt_server_data_update(lut, 1);
 143 }
 144
 145 /**
 146  * write data in last_rcvd file.
 147  */
 148 static int lut_last_rcvd_write(const struct lu_env *env, struct lu_target *lut,
 149                                const struct lu_buf *buf, loff_t *off, int sync)
 150 {
 151         struct thandle *th;
 152         int rc;
 153         ENTRY;
 154
 155         th = dt_trans_create(env, lut->lut_bottom);
 156         if (IS_ERR(th))
 157                 RETURN(PTR_ERR(th));
 158
 159         rc = dt_declare_record_write(env, lut->lut_last_rcvd,
 160                                      buf->lb_len, *off, th);
 161         if (rc)
 162                 goto stop;
 163
 164         rc = dt_trans_start_local(env, lut->lut_bottom, th);
 165         if (rc)
 166                 goto stop;
 167
 168         rc = dt_record_write(env, lut->lut_last_rcvd, buf, off, th);
 169
 170 stop:
 171         dt_trans_stop(env, lut->lut_bottom, th);
 172
 173         CDEBUG(D_INFO, "write last_rcvd header rc = %d:\n"
 174                "uuid = %s\nlast_transno = "LPU64"\n",
 175                rc, lut->lut_lsd.lsd_uuid, lut->lut_lsd.lsd_last_transno);
 176
 177         RETURN(rc);
 178 }
 179
 180 /**
 181  * Allocate in-memory data for client slot related to export.
 182  */
 183 int lut_client_alloc(struct obd_export *exp)
 184 {
 185         LASSERT(exp != exp->exp_obd->obd_self_export);
 186
 187         OBD_ALLOC_PTR(exp->exp_target_data.ted_lcd);
 188         if (exp->exp_target_data.ted_lcd == NULL)
 189                 RETURN(-ENOMEM);
 190         /* Mark that slot is not yet valid, 0 doesn't work here */
 191         exp->exp_target_data.ted_lr_idx = -1;
 192         RETURN(0);
 193 }
 194 EXPORT_SYMBOL(lut_client_alloc);
 195
 196 /**
 197  * Free in-memory data for client slot related to export.
 198  */
 199 void lut_client_free(struct obd_export *exp)
 200 {
 201         struct tg_export_data *ted = &exp->exp_target_data;
 202         struct lu_target *lut = class_exp2tgt(exp);
 203
 204         LASSERT(exp != exp->exp_obd->obd_self_export);
 205
 206         OBD_FREE_PTR(ted->ted_lcd);
 207         ted->ted_lcd = NULL;
 208
 209         /* Slot may be not yet assigned */
 210         if (ted->ted_lr_idx < 0)
 211                 return;
 212         /* Clear bit when lcd is freed */
 213         cfs_spin_lock(&lut->lut_client_bitmap_lock);
 214         if (!cfs_test_and_clear_bit(ted->ted_lr_idx, lut->lut_client_bitmap)) {
 215                 CERROR("%s: client %u bit already clear in bitmap\n",
 216                        exp->exp_obd->obd_name, ted->ted_lr_idx);
 217                 LBUG();
 218         }
 219         cfs_spin_unlock(&lut->lut_client_bitmap_lock);
 220 }
 221 EXPORT_SYMBOL(lut_client_free);
 222
 223 /**
 224  * Update client data in last_rcvd
 225  */
 226 int lut_client_data_update(const struct lu_env *env, struct obd_export *exp)
 227 {
 228         struct tg_export_data *ted = &exp->exp_target_data;
 229         struct lu_target *lut = class_exp2tgt(exp);
 230         struct lsd_client_data tmp_lcd;
 231         loff_t tmp_off = ted->ted_lr_off;
 232         struct lu_buf tmp_buf = {
 233                                         .lb_buf = &tmp_lcd,
 234                                         .lb_len = sizeof(tmp_lcd)
 235                                 };
 236         int rc = 0;
 237
 238         lcd_cpu_to_le(ted->ted_lcd, &tmp_lcd);
 239         LASSERT(lut->lut_last_rcvd);
 240         rc = lut_last_rcvd_write(env, lut, &tmp_buf, &tmp_off, 0);
 241
 242         return rc;
 243 }
 244
 245 /**
 246  * Update server data in last_rcvd
 247  */
 248 int lut_server_data_update(const struct lu_env *env,
 249                            struct lu_target *lut, int sync)
 250 {
 251         struct lr_server_data tmp_lsd;
 252         loff_t tmp_off = 0;
 253         struct lu_buf tmp_buf = {
 254                                         .lb_buf = &tmp_lsd,
 255                                         .lb_len = sizeof(tmp_lsd)
 256                                 };
 257         int rc = 0;
 258         ENTRY;
 259
 260         CDEBUG(D_SUPER,
 261                "%s: mount_count is "LPU64", last_transno is "LPU64"\n",
 262                lut->lut_lsd.lsd_uuid, lut->lut_obd->u.obt.obt_mount_count,
 263                lut->lut_last_transno);
 264
 265         cfs_spin_lock(&lut->lut_translock);
 266         lut->lut_lsd.lsd_last_transno = lut->lut_last_transno;
 267         cfs_spin_unlock(&lut->lut_translock);
 268
 269         lsd_cpu_to_le(&lut->lut_lsd, &tmp_lsd);
 270         if (lut->lut_last_rcvd != NULL)
 271                 rc = lut_last_rcvd_write(env, lut, &tmp_buf, &tmp_off, sync);
 272         RETURN(rc);
 273 }
 274
 275 void lut_client_epoch_update(const struct lu_env *env, struct obd_export *exp)
 276 {
 277         struct lsd_client_data *lcd = exp->exp_target_data.ted_lcd;
 278         struct lu_target *lut = class_exp2tgt(exp);
 279
 280         LASSERT(lut->lut_bottom);
 281         /** VBR: set client last_epoch to current epoch */
 282         if (lcd->lcd_last_epoch >= lut->lut_lsd.lsd_start_epoch)
 283                 return;
 284         lcd->lcd_last_epoch = lut->lut_lsd.lsd_start_epoch;
 285         lut_client_data_update(env, exp);
 286 }
 287
 288 /**
 289  * Update boot epoch when recovery ends
 290  */
 291 void lut_boot_epoch_update(struct lu_target *lut)
 292 {
 293         struct lu_env env;
 294         struct ptlrpc_request *req;
 295         __u32 start_epoch;
 296         cfs_list_t client_list;
 297         int rc;
 298
 299         if (lut->lut_obd->obd_stopping)
 300                 return;
 301         /** Increase server epoch after recovery */
 302         if (lut->lut_bottom == NULL)
 303                 return obt_boot_epoch_update(lut);
 304
 305         rc = lu_env_init(&env, LCT_LOCAL);
 306         if (rc) {
 307                 CERROR("Can't initialize environment rc=%d\n", rc);
 308                 return;
 309         }
 310
 311         cfs_spin_lock(&lut->lut_translock);
 312         start_epoch = lr_epoch(lut->lut_last_transno) + 1;
 313         lut->lut_last_transno = (__u64)start_epoch << LR_EPOCH_BITS;
 314         lut->lut_lsd.lsd_start_epoch = start_epoch;
 315         cfs_spin_unlock(&lut->lut_translock);
 316
 317         CFS_INIT_LIST_HEAD(&client_list);
 318         /**
 319          * The recovery is not yet finished and final queue can still be updated
 320          * with resend requests. Move final list to separate one for processing
 321          */
 322         cfs_spin_lock(&lut->lut_obd->obd_recovery_task_lock);
 323         cfs_list_splice_init(&lut->lut_obd->obd_final_req_queue, &client_list);
 324         cfs_spin_unlock(&lut->lut_obd->obd_recovery_task_lock);
 325
 326         /**
 327          * go through list of exports participated in recovery and
 328          * set new epoch for them
 329          */
 330         cfs_list_for_each_entry(req, &client_list, rq_list) {
 331                 LASSERT(!req->rq_export->exp_delayed);
 332                 if (!req->rq_export->exp_vbr_failed)
 333                         lut_client_epoch_update(&env, req->rq_export);
 334         }
 335         /** return list back at once */
 336         cfs_spin_lock(&lut->lut_obd->obd_recovery_task_lock);
 337         cfs_list_splice_init(&client_list, &lut->lut_obd->obd_final_req_queue);
 338         cfs_spin_unlock(&lut->lut_obd->obd_recovery_task_lock);
 339         /** update server epoch */
 340         lut_server_data_update(&env, lut, 1);
 341         lu_env_fini(&env);
 342 }
 343 EXPORT_SYMBOL(lut_boot_epoch_update);
 344
 345 /**
 346  * commit callback, need to update last_commited value
 347  */
 348 struct lut_last_committed_callback {
 349         struct dt_txn_commit_cb llcc_cb;
 350         struct lu_target       *llcc_lut;
 351         struct obd_export      *llcc_exp;
 352         __u64                   llcc_transno;
 353 };
 354
 355 void lut_cb_last_committed(struct lu_env *env, struct thandle *th,
 356                            struct dt_txn_commit_cb *cb, int err)
 357 {
 358         struct lut_last_committed_callback *ccb;
 359
 360         ccb = container_of0(cb, struct lut_last_committed_callback, llcc_cb);
 361
 362         LASSERT(ccb->llcc_exp->exp_obd == ccb->llcc_lut->lut_obd);
 363
 364         cfs_spin_lock(&ccb->llcc_lut->lut_translock);
 365         if (ccb->llcc_transno > ccb->llcc_lut->lut_obd->obd_last_committed)
 366                 ccb->llcc_lut->lut_obd->obd_last_committed = ccb->llcc_transno;
 367
 368         LASSERT(ccb->llcc_exp);
 369         if (ccb->llcc_transno > ccb->llcc_exp->exp_last_committed) {
 370                 ccb->llcc_exp->exp_last_committed = ccb->llcc_transno;
 371                 cfs_spin_unlock(&ccb->llcc_lut->lut_translock);
 372                 ptlrpc_commit_replies(ccb->llcc_exp);
 373         } else {
 374                 cfs_spin_unlock(&ccb->llcc_lut->lut_translock);
 375         }
 376         class_export_cb_put(ccb->llcc_exp);
 377         if (ccb->llcc_transno)
 378                 CDEBUG(D_HA, "%s: transno "LPD64" is committed\n",
 379                        ccb->llcc_lut->lut_obd->obd_name, ccb->llcc_transno);
 380         cfs_list_del(&ccb->llcc_cb.dcb_linkage);
 381         OBD_FREE_PTR(ccb);
 382 }
 383
 384 int lut_last_commit_cb_add(struct thandle *th, struct lu_target *lut,
 385                            struct obd_export *exp, __u64 transno)
 386 {
 387         struct lut_last_committed_callback *ccb;
 388         int rc;
 389
 390         OBD_ALLOC_PTR(ccb);
 391         if (ccb == NULL)
 392                 return -ENOMEM;
 393
 394         ccb->llcc_cb.dcb_func = lut_cb_last_committed;
 395         CFS_INIT_LIST_HEAD(&ccb->llcc_cb.dcb_linkage);
 396         ccb->llcc_lut = lut;
 397         ccb->llcc_exp = class_export_cb_get(exp);
 398         ccb->llcc_transno = transno;
 399
 400         rc = dt_trans_cb_add(th, &ccb->llcc_cb);
 401         if (rc) {
 402                 class_export_cb_put(exp);
 403                 OBD_FREE_PTR(ccb);
 404         }
 405         return rc;
 406 }
 407 EXPORT_SYMBOL(lut_last_commit_cb_add);
 408
 409 struct lut_new_client_callback {
 410         struct dt_txn_commit_cb lncc_cb;
 411         struct obd_export      *lncc_exp;
 412 };
 413
 414 void lut_cb_new_client(struct lu_env *env, struct thandle *th,
 415                        struct dt_txn_commit_cb *cb, int err)
 416 {
 417         struct lut_new_client_callback *ccb;
 418
 419         ccb = container_of0(cb, struct lut_new_client_callback, lncc_cb);
 420
 421         LASSERT(ccb->lncc_exp->exp_obd);
 422
 423         CDEBUG(D_RPCTRACE, "%s: committing for initial connect of %s\n",
 424                ccb->lncc_exp->exp_obd->obd_name,
 425                ccb->lncc_exp->exp_client_uuid.uuid);
 426
 427         cfs_spin_lock(&ccb->lncc_exp->exp_lock);
 428         ccb->lncc_exp->exp_need_sync = 0;
 429         cfs_spin_unlock(&ccb->lncc_exp->exp_lock);
 430         class_export_cb_put(ccb->lncc_exp);
 431
 432         cfs_list_del(&ccb->lncc_cb.dcb_linkage);
 433         OBD_FREE_PTR(ccb);
 434 }
 435
 436 int lut_new_client_cb_add(struct thandle *th, struct obd_export *exp)
 437 {
 438         struct lut_new_client_callback *ccb;
 439         int rc;
 440
 441         OBD_ALLOC_PTR(ccb);
 442         if (ccb == NULL)
 443                 return -ENOMEM;
 444
 445         ccb->lncc_cb.dcb_func = lut_cb_new_client;
 446         CFS_INIT_LIST_HEAD(&ccb->lncc_cb.dcb_linkage);
 447         ccb->lncc_exp = class_export_cb_get(exp);
 448
 449         rc = dt_trans_cb_add(th, &ccb->lncc_cb);
 450         if (rc) {
 451                 class_export_cb_put(exp);
 452                 OBD_FREE_PTR(ccb);
 453         }
 454         return rc;
 455 }
 456 EXPORT_SYMBOL(lut_new_client_cb_add);
 457
 458 int lut_init(const struct lu_env *env, struct lu_target *lut,
 459              struct obd_device *obd, struct dt_device *dt)
 460 {
 461         struct lu_fid fid;
 462         struct dt_object *o;
 463         int rc = 0;
 464         ENTRY;
 465
 466         LASSERT(lut);
 467         LASSERT(obd);
 468         lut->lut_obd = obd;
 469         lut->lut_bottom = dt;
 470         lut->lut_last_rcvd = NULL;
 471         obd->u.obt.obt_lut = lut;
 472
 473         cfs_spin_lock_init(&lut->lut_translock);
 474         cfs_spin_lock_init(&lut->lut_client_bitmap_lock);
 475
 476         OBD_ALLOC(lut->lut_client_bitmap, LR_MAX_CLIENTS >> 3);
 477         if (lut->lut_client_bitmap == NULL)
 478                 RETURN(-ENOMEM);
 479
 480         /** obdfilter has no lu_device stack yet */
 481         if (dt == NULL)
 482                 RETURN(rc);
 483         o = dt_store_open(env, lut->lut_bottom, "", LAST_RCVD, &fid);
 484         if (!IS_ERR(o)) {
 485                 lut->lut_last_rcvd = o;
 486         } else {
 487                 OBD_FREE(lut->lut_client_bitmap, LR_MAX_CLIENTS >> 3);
 488                 lut->lut_client_bitmap = NULL;
 489                 rc = PTR_ERR(o);
 490                 CERROR("cannot open %s: rc = %d\n", LAST_RCVD, rc);
 491         }
 492
 493         RETURN(rc);
 494 }
 495 EXPORT_SYMBOL(lut_init);
 496
 497 void lut_fini(const struct lu_env *env, struct lu_target *lut)
 498 {
 499         ENTRY;
 500
 501         if (lut->lut_client_bitmap) {
 502                 OBD_FREE(lut->lut_client_bitmap, LR_MAX_CLIENTS >> 3);
 503                 lut->lut_client_bitmap = NULL;
 504         }
 505         if (lut->lut_last_rcvd) {
 506                 lu_object_put(env, &lut->lut_last_rcvd->do_lu);
 507                 lut->lut_last_rcvd = NULL;
 508         }
 509         EXIT;
 510 }
 511 EXPORT_SYMBOL(lut_fini);