lustre/ofd/ofd_fs.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) 2012, 2013, Intel Corporation.
  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/ofd/ofd_fs.c
  37  *
  38  * Author: Alexey Zhuravlev <bzzz@whamcloud.com>
  39  * Author: Mikhail Pershin <tappro@whamcloud.com>
  40  */
  41
  42 #define DEBUG_SUBSYSTEM S_FILTER
  43
  44 #include "ofd_internal.h"
  45
  46 int ofd_record_write(const struct lu_env *env, struct ofd_device *ofd,
  47                      struct dt_object *dt, struct lu_buf *buf, loff_t *off)
  48 {
  49         struct thandle  *th;
  50         int              rc;
  51
  52         ENTRY;
  53
  54         LASSERT(dt);
  55
  56         th = dt_trans_create(env, ofd->ofd_osd);
  57         if (IS_ERR(th))
  58                 RETURN(PTR_ERR(th));
  59
  60         rc = dt_declare_record_write(env, dt, buf->lb_len, *off, th);
  61         if (rc == 0) {
  62                 rc = dt_trans_start_local(env, ofd->ofd_osd, th);
  63                 if (rc == 0)
  64                         rc = dt_record_write(env, dt, buf, off, th);
  65         }
  66         dt_trans_stop(env, ofd->ofd_osd, th);
  67
  68         RETURN(rc);
  69 }
  70
  71 int ofd_precreate_batch(struct ofd_device *ofd, int batch)
  72 {
  73         int count;
  74
  75         spin_lock(&ofd->ofd_batch_lock);
  76         count = min(ofd->ofd_precreate_batch, batch);
  77         spin_unlock(&ofd->ofd_batch_lock);
  78
  79         return count;
  80 }
  81
  82 struct ofd_seq *ofd_seq_get(struct ofd_device *ofd, obd_seq seq)
  83 {
  84         struct ofd_seq *oseq;
  85
  86         read_lock(&ofd->ofd_seq_list_lock);
  87         cfs_list_for_each_entry(oseq, &ofd->ofd_seq_list, os_list) {
  88                 if (ostid_seq(&oseq->os_oi) == seq) {
  89                         atomic_inc(&oseq->os_refc);
  90                         read_unlock(&ofd->ofd_seq_list_lock);
  91                         return oseq;
  92                 }
  93         }
  94         read_unlock(&ofd->ofd_seq_list_lock);
  95         return NULL;
  96 }
  97
  98 static void ofd_seq_destroy(const struct lu_env *env,
  99                             struct ofd_seq *oseq)
 100 {
 101         LASSERT(cfs_list_empty(&oseq->os_list));
 102         LASSERT(oseq->os_lastid_obj != NULL);
 103         lu_object_put(env, &oseq->os_lastid_obj->do_lu);
 104         OBD_FREE_PTR(oseq);
 105 }
 106
 107 void ofd_seq_put(const struct lu_env *env, struct ofd_seq *oseq)
 108 {
 109         if (atomic_dec_and_test(&oseq->os_refc))
 110                 ofd_seq_destroy(env, oseq);
 111 }
 112
 113 static void ofd_seq_delete(const struct lu_env *env, struct ofd_seq *oseq)
 114 {
 115         cfs_list_del_init(&oseq->os_list);
 116         ofd_seq_put(env, oseq);
 117 }
 118
 119 /**
 120  * Add a new sequence to the OFD device.
 121  *
 122  * \param ofd OFD device
 123  * \param new_seq new sequence to be added
 124  *
 125  * \retval the seq to be added or the existing seq
 126  **/
 127 static struct ofd_seq *ofd_seq_add(const struct lu_env *env,
 128                                    struct ofd_device *ofd,
 129                                    struct ofd_seq *new_seq)
 130 {
 131         struct ofd_seq *os = NULL;
 132
 133         write_lock(&ofd->ofd_seq_list_lock);
 134         cfs_list_for_each_entry(os, &ofd->ofd_seq_list, os_list) {
 135                 if (ostid_seq(&os->os_oi) == ostid_seq(&new_seq->os_oi)) {
 136                         atomic_inc(&os->os_refc);
 137                         write_unlock(&ofd->ofd_seq_list_lock);
 138                         /* The seq has not been added to the list */
 139                         ofd_seq_put(env, new_seq);
 140                         return os;
 141                 }
 142         }
 143         atomic_inc(&new_seq->os_refc);
 144         cfs_list_add_tail(&new_seq->os_list, &ofd->ofd_seq_list);
 145         ofd->ofd_seq_count++;
 146         write_unlock(&ofd->ofd_seq_list_lock);
 147         return new_seq;
 148 }
 149
 150 obd_id ofd_seq_last_oid(struct ofd_seq *oseq)
 151 {
 152         obd_id id;
 153
 154         spin_lock(&oseq->os_last_oid_lock);
 155         id = ostid_id(&oseq->os_oi);
 156         spin_unlock(&oseq->os_last_oid_lock);
 157
 158         return id;
 159 }
 160
 161 void ofd_seq_last_oid_set(struct ofd_seq *oseq, obd_id id)
 162 {
 163         spin_lock(&oseq->os_last_oid_lock);
 164         if (likely(ostid_id(&oseq->os_oi) < id))
 165                 ostid_set_id(&oseq->os_oi, id);
 166         spin_unlock(&oseq->os_last_oid_lock);
 167 }
 168
 169 int ofd_seq_last_oid_write(const struct lu_env *env, struct ofd_device *ofd,
 170                            struct ofd_seq *oseq)
 171 {
 172         struct ofd_thread_info  *info = ofd_info(env);
 173         obd_id                   tmp;
 174         int                      rc;
 175
 176         ENTRY;
 177
 178         tmp = cpu_to_le64(ofd_seq_last_oid(oseq));
 179
 180         info->fti_buf.lb_buf = &tmp;
 181         info->fti_buf.lb_len = sizeof(tmp);
 182         info->fti_off = 0;
 183
 184         rc = ofd_record_write(env, ofd, oseq->os_lastid_obj, &info->fti_buf,
 185                               &info->fti_off);
 186
 187         CDEBUG(D_INODE, "%s: write last_objid "DOSTID": rc = %d\n",
 188                ofd_name(ofd), POSTID(&oseq->os_oi), rc);
 189
 190         RETURN(rc);
 191 }
 192
 193 static void ofd_deregister_seq_exp(struct ofd_device *ofd)
 194 {
 195         struct seq_server_site  *ss = &ofd->ofd_seq_site;
 196
 197         if (ss->ss_client_seq != NULL) {
 198                 lustre_deregister_lwp_item(&ss->ss_client_seq->lcs_exp);
 199                 ss->ss_client_seq->lcs_exp = NULL;
 200         }
 201
 202         if (ss->ss_server_fld != NULL) {
 203                 lustre_deregister_lwp_item(&ss->ss_server_fld->lsf_control_exp);
 204                 ss->ss_server_fld->lsf_control_exp = NULL;
 205         }
 206 }
 207
 208 static int ofd_fld_fini(const struct lu_env *env,
 209                         struct ofd_device *ofd)
 210 {
 211         struct seq_server_site *ss = &ofd->ofd_seq_site;
 212         ENTRY;
 213
 214         if (ss && ss->ss_server_fld) {
 215                 fld_server_fini(env, ss->ss_server_fld);
 216                 OBD_FREE_PTR(ss->ss_server_fld);
 217                 ss->ss_server_fld = NULL;
 218         }
 219
 220         RETURN(0);
 221 }
 222
 223 void ofd_seqs_free(const struct lu_env *env, struct ofd_device *ofd)
 224 {
 225         struct ofd_seq  *oseq;
 226         struct ofd_seq  *tmp;
 227         cfs_list_t       dispose;
 228
 229         CFS_INIT_LIST_HEAD(&dispose);
 230         write_lock(&ofd->ofd_seq_list_lock);
 231         cfs_list_for_each_entry_safe(oseq, tmp, &ofd->ofd_seq_list, os_list) {
 232                 cfs_list_move(&oseq->os_list, &dispose);
 233         }
 234         write_unlock(&ofd->ofd_seq_list_lock);
 235
 236         while (!cfs_list_empty(&dispose)) {
 237                 oseq = container_of0(dispose.next, struct ofd_seq, os_list);
 238                 ofd_seq_delete(env, oseq);
 239         }
 240 }
 241
 242 void ofd_seqs_fini(const struct lu_env *env, struct ofd_device *ofd)
 243 {
 244         int rc;
 245
 246         ofd_deregister_seq_exp(ofd);
 247
 248         rc = ofd_fid_fini(env, ofd);
 249         if (rc != 0)
 250                 CERROR("%s: fid fini error: rc = %d\n", ofd_name(ofd), rc);
 251
 252         rc = ofd_fld_fini(env, ofd);
 253         if (rc != 0)
 254                 CERROR("%s: fld fini error: rc = %d\n", ofd_name(ofd), rc);
 255
 256         ofd_seqs_free(env, ofd);
 257
 258         LASSERT(cfs_list_empty(&ofd->ofd_seq_list));
 259 }
 260
 261 /**
 262  *
 263  * \retval the seq with seq number or errno (never NULL)
 264  */
 265 struct ofd_seq *ofd_seq_load(const struct lu_env *env, struct ofd_device *ofd,
 266                              obd_seq seq)
 267 {
 268         struct ofd_thread_info  *info = ofd_info(env);
 269         struct ofd_seq          *oseq = NULL;
 270         struct dt_object        *dob;
 271         obd_id                   lastid;
 272         int                      rc;
 273
 274         ENTRY;
 275
 276         /* if seq is already initialized */
 277         oseq = ofd_seq_get(ofd, seq);
 278         if (oseq != NULL)
 279                 RETURN(oseq);
 280
 281         OBD_ALLOC_PTR(oseq);
 282         if (oseq == NULL)
 283                 RETURN(ERR_PTR(-ENOMEM));
 284
 285         lu_last_id_fid(&info->fti_fid, seq, ofd->ofd_lut.lut_lsd.lsd_osd_index);
 286         memset(&info->fti_attr, 0, sizeof(info->fti_attr));
 287         info->fti_attr.la_valid = LA_MODE;
 288         info->fti_attr.la_mode = S_IFREG |  S_IRUGO | S_IWUSR;
 289         info->fti_dof.dof_type = dt_mode_to_dft(S_IFREG);
 290
 291         /* create object tracking per-seq last created
 292          * id to be used by orphan recovery mechanism */
 293         dob = dt_find_or_create(env, ofd->ofd_osd, &info->fti_fid,
 294                                 &info->fti_dof, &info->fti_attr);
 295         if (IS_ERR(dob)) {
 296                 OBD_FREE_PTR(oseq);
 297                 RETURN((void *)dob);
 298         }
 299
 300         oseq->os_lastid_obj = dob;
 301
 302         CFS_INIT_LIST_HEAD(&oseq->os_list);
 303         mutex_init(&oseq->os_create_lock);
 304         spin_lock_init(&oseq->os_last_oid_lock);
 305         ostid_set_seq(&oseq->os_oi, seq);
 306
 307         atomic_set(&oseq->os_refc, 1);
 308
 309         rc = dt_attr_get(env, dob, &info->fti_attr, BYPASS_CAPA);
 310         if (rc)
 311                 GOTO(cleanup, rc);
 312
 313         if (info->fti_attr.la_size == 0) {
 314                 /* object is just created, initialize last id */
 315                 ofd_seq_last_oid_set(oseq, OFD_INIT_OBJID);
 316                 ofd_seq_last_oid_write(env, ofd, oseq);
 317         } else if (info->fti_attr.la_size == sizeof(lastid)) {
 318                 info->fti_off = 0;
 319                 info->fti_buf.lb_buf = &lastid;
 320                 info->fti_buf.lb_len = sizeof(lastid);
 321
 322                 rc = dt_record_read(env, dob, &info->fti_buf, &info->fti_off);
 323                 if (rc) {
 324                         CERROR("%s: can't read last_id: rc = %d\n",
 325                                 ofd_name(ofd), rc);
 326                         GOTO(cleanup, rc);
 327                 }
 328                 ofd_seq_last_oid_set(oseq, le64_to_cpu(lastid));
 329         } else {
 330                 CERROR("%s: corrupted size "LPU64" LAST_ID of seq "LPX64"\n",
 331                         ofd_name(ofd), (__u64)info->fti_attr.la_size, seq);
 332                 GOTO(cleanup, rc = -EINVAL);
 333         }
 334
 335         oseq = ofd_seq_add(env, ofd, oseq);
 336         RETURN((oseq != NULL) ? oseq : ERR_PTR(-ENOENT));
 337 cleanup:
 338         ofd_seq_put(env, oseq);
 339         return ERR_PTR(rc);
 340 }
 341
 342 static int ofd_fld_init(const struct lu_env *env, const char *uuid,
 343                         struct ofd_device *ofd)
 344 {
 345         struct seq_server_site *ss = &ofd->ofd_seq_site;
 346         int rc;
 347         ENTRY;
 348
 349         OBD_ALLOC_PTR(ss->ss_server_fld);
 350         if (ss->ss_server_fld == NULL)
 351                 RETURN(rc = -ENOMEM);
 352
 353         rc = fld_server_init(env, ss->ss_server_fld, ofd->ofd_osd, uuid,
 354                              LU_SEQ_RANGE_OST);
 355         if (rc) {
 356                 OBD_FREE_PTR(ss->ss_server_fld);
 357                 ss->ss_server_fld = NULL;
 358                 RETURN(rc);
 359         }
 360         RETURN(0);
 361 }
 362
 363 /**
 364  * It will retrieve its FLDB entries from MDT0, and it only happens
 365  * when upgrading existent FS to 2.6.
 366  **/
 367 static int ofd_register_lwp_callback(void *data)
 368 {
 369         struct lu_env           env;
 370         struct ofd_device       *ofd = data;
 371         struct lu_server_fld    *fld = ofd->ofd_seq_site.ss_server_fld;
 372         int                     rc;
 373         ENTRY;
 374
 375         if (!likely(fld->lsf_new))
 376                 RETURN(0);
 377
 378         rc = lu_env_init(&env, LCT_DT_THREAD);
 379         if (rc) {
 380                 CERROR("%s: cannot init env: rc = %d\n", ofd_name(ofd), rc);
 381                 RETURN(rc);
 382         }
 383
 384         rc = fld_update_from_controller(&env, fld);
 385         if (rc != 0) {
 386                 CERROR("%s: cannot update controller: rc = %d\n",
 387                        ofd_name(ofd), rc);
 388                 GOTO(out, rc);
 389         }
 390 out:
 391         lu_env_fini(&env);
 392         RETURN(rc);
 393 }
 394
 395 static int ofd_register_seq_exp(struct ofd_device *ofd)
 396 {
 397         struct seq_server_site  *ss = &ofd->ofd_seq_site;
 398         char                    *lwp_name = NULL;
 399         int                     rc;
 400
 401         OBD_ALLOC(lwp_name, MAX_OBD_NAME);
 402         if (lwp_name == NULL)
 403                 GOTO(out_free, rc = -ENOMEM);
 404
 405         rc = tgt_name2lwpname(ofd_name(ofd), lwp_name);
 406         if (rc != 0)
 407                 GOTO(out_free, rc);
 408
 409         rc = lustre_register_lwp_item(lwp_name, &ss->ss_client_seq->lcs_exp,
 410                                       NULL, NULL);
 411         if (rc != 0)
 412                 GOTO(out_free, rc);
 413
 414         rc = lustre_register_lwp_item(lwp_name,
 415                                       &ss->ss_server_fld->lsf_control_exp,
 416                                       ofd_register_lwp_callback, ofd);
 417         if (rc != 0) {
 418                 lustre_deregister_lwp_item(&ss->ss_client_seq->lcs_exp);
 419                 ss->ss_client_seq->lcs_exp = NULL;
 420                 GOTO(out_free, rc);
 421         }
 422 out_free:
 423         if (lwp_name != NULL)
 424                 OBD_FREE(lwp_name, MAX_OBD_NAME);
 425
 426         return rc;
 427 }
 428
 429 /* object sequence management */
 430 int ofd_seqs_init(const struct lu_env *env, struct ofd_device *ofd)
 431 {
 432         int rc;
 433
 434         rc = ofd_fid_init(env, ofd);
 435         if (rc != 0) {
 436                 CERROR("%s: fid init error: rc = %d\n", ofd_name(ofd), rc);
 437                 return rc;
 438         }
 439
 440         rc = ofd_fld_init(env, ofd_name(ofd), ofd);
 441         if (rc) {
 442                 CERROR("%s: Can't init fld, rc %d\n", ofd_name(ofd), rc);
 443                 return rc;
 444         }
 445
 446         rc = ofd_register_seq_exp(ofd);
 447         if (rc) {
 448                 CERROR("%s: Can't init seq exp, rc %d\n", ofd_name(ofd), rc);
 449                 return rc;
 450         }
 451
 452         rwlock_init(&ofd->ofd_seq_list_lock);
 453         CFS_INIT_LIST_HEAD(&ofd->ofd_seq_list);
 454         ofd->ofd_seq_count = 0;
 455         return rc;
 456 }
 457
 458 int ofd_fs_setup(const struct lu_env *env, struct ofd_device *ofd,
 459                  struct obd_device *obd)
 460 {
 461         struct ofd_thread_info  *info = ofd_info(env);
 462         struct dt_object        *fo;
 463         int                      rc = 0;
 464
 465         ENTRY;
 466
 467         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_FS_SETUP))
 468                 RETURN (-ENOENT);
 469
 470         lu_local_obj_fid(&info->fti_fid, OFD_HEALTH_CHECK_OID);
 471         memset(&info->fti_attr, 0, sizeof(info->fti_attr));
 472         info->fti_attr.la_valid = LA_MODE;
 473         info->fti_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
 474         info->fti_dof.dof_type = dt_mode_to_dft(S_IFREG);
 475
 476         fo = dt_find_or_create(env, ofd->ofd_osd, &info->fti_fid,
 477                                &info->fti_dof, &info->fti_attr);
 478         if (IS_ERR(fo))
 479                 GOTO(out, rc = PTR_ERR(fo));
 480
 481         ofd->ofd_health_check_file = fo;
 482
 483         rc = ofd_seqs_init(env, ofd);
 484         if (rc)
 485                 GOTO(out_hc, rc);
 486
 487         RETURN(0);
 488 out_hc:
 489         lu_object_put(env, &ofd->ofd_health_check_file->do_lu);
 490 out:
 491         return rc;
 492 }
 493
 494 void ofd_fs_cleanup(const struct lu_env *env, struct ofd_device *ofd)
 495 {
 496         int i;
 497
 498         ENTRY;
 499
 500         ofd_info_init(env, NULL);
 501
 502         ofd_seqs_fini(env, ofd);
 503
 504         i = dt_sync(env, ofd->ofd_osd);
 505         if (i)
 506                 CERROR("can't sync: %d\n", i);
 507
 508         if (ofd->ofd_health_check_file) {
 509                 lu_object_put(env, &ofd->ofd_health_check_file->do_lu);
 510                 ofd->ofd_health_check_file = NULL;
 511         }
 512
 513         EXIT;
 514 }
 515