lustre/quota/qsd_writeback.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, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 021110-1307, USA
  20  *
  21  * GPL HEADER END
  22  */
  23 /*
  24  * Copyright (c) 2011, 2012, Intel, Inc.
  25  * Use is subject to license terms.
  26  *
  27  * Author: Johann Lombardi <johann.lombardi@intel.com>
  28  * Author: Niu    Yawei    <yawei.niu@intel.com>
  29  */
  30
  31 #ifndef EXPORT_SYMTAB
  32 # define EXPORT_SYMTAB
  33 #endif
  34
  35 #define DEBUG_SUBSYSTEM S_LQUOTA
  36
  37 #include "qsd_internal.h"
  38
  39 extern cfs_mem_cache_t *upd_kmem;
  40
  41 /*
  42  * Allocate and fill an qsd_upd_rec structure to be processed by the writeback
  43  * thread.
  44  *
  45  * \param qqi - is the qsd_qtype_info structure relevant to the update
  46  * \param lqe - is the lquota entry subject to the update
  47  * \param qid - is the identifier subject to the update
  48  * \param rec - is the record storing the new quota settings
  49  * \param ver - is the version associated with the update
  50  * \param global - is a boolean set to true if this is an update of the global
  51  *                 index and false for a slave index.
  52  */
  53 static struct qsd_upd_rec *qsd_upd_alloc(struct qsd_qtype_info *qqi,
  54                                          struct lquota_entry *lqe,
  55                                          union lquota_id *qid,
  56                                          union lquota_rec *rec, __u64 ver,
  57                                          bool global)
  58 {
  59         struct qsd_upd_rec      *upd;
  60
  61         OBD_SLAB_ALLOC_PTR_GFP(upd, upd_kmem, CFS_ALLOC_IO);
  62         if (upd == NULL) {
  63                 CERROR("Failed to allocate upd");
  64                 return NULL;
  65         }
  66
  67         /* fill it */
  68         CFS_INIT_LIST_HEAD(&upd->qur_link);
  69         upd->qur_qqi = qqi;
  70         upd->qur_lqe = lqe;
  71         if (lqe)
  72                 lqe_getref(lqe);
  73         upd->qur_qid    = *qid;
  74         upd->qur_rec    = *rec;
  75         upd->qur_ver    = ver;
  76         upd->qur_global = global;
  77
  78         return upd;
  79 }
  80
  81 static void qsd_upd_free(struct qsd_upd_rec *upd)
  82 {
  83         if (upd->qur_lqe)
  84                 lqe_putref(upd->qur_lqe);
  85         OBD_SLAB_FREE_PTR(upd, upd_kmem);
  86 }
  87
  88 /* must hold the qsd_lock */
  89 static void qsd_upd_add(struct qsd_instance *qsd, struct qsd_upd_rec *upd)
  90 {
  91         if (!qsd->qsd_stopping) {
  92                 list_add_tail(&upd->qur_link, &qsd->qsd_upd_list);
  93                 /* wake up the upd thread */
  94                 cfs_waitq_signal(&qsd->qsd_upd_thread.t_ctl_waitq);
  95         } else {
  96                 CWARN("%s: discard deferred update.\n", qsd->qsd_svname);
  97                 if (upd->qur_lqe)
  98                         LQUOTA_WARN(upd->qur_lqe, "discard deferred update.");
  99                 qsd_upd_free(upd);
 100         }
 101 }
 102
 103 /* must hold the qsd_lock */
 104 static void qsd_add_deferred(cfs_list_t *list, struct qsd_upd_rec *upd)
 105 {
 106         struct qsd_upd_rec      *tmp;
 107
 108         /* Sort the updates in ascending order */
 109         cfs_list_for_each_entry_reverse(tmp, list, qur_link) {
 110
 111                 LASSERTF(upd->qur_ver != tmp->qur_ver, "ver:"LPU64"\n",
 112                          upd->qur_ver);
 113
 114                 if (upd->qur_ver < tmp->qur_ver) {
 115                         continue;
 116                 } else {
 117                         cfs_list_add_tail(&upd->qur_link, &tmp->qur_link);
 118                         return;
 119                 }
 120         }
 121         cfs_list_add(&upd->qur_link, list);
 122 }
 123
 124 /* must hold the qsd_lock */
 125 static void qsd_kickoff_deferred(struct qsd_qtype_info *qqi, cfs_list_t *list,
 126                                  __u64 ver)
 127 {
 128         struct qsd_upd_rec      *upd, *tmp;
 129         ENTRY;
 130
 131         /* Get the first update record in the list, which has the smallest
 132          * version, discard all records with versions smaller than the current
 133          * one */
 134         cfs_list_for_each_entry_safe(upd, tmp, list, qur_link) {
 135                 if (upd->qur_ver <= ver) {
 136                         /* drop this update */
 137                         cfs_list_del_init(&upd->qur_link);
 138                         CDEBUG(D_QUOTA, "%s: skipping deferred update ver:"
 139                                LPU64"/"LPU64", global:%d, qid:"LPU64"\n",
 140                                qqi->qqi_qsd->qsd_svname, upd->qur_ver, ver,
 141                                upd->qur_global, upd->qur_qid.qid_uid);
 142                         qsd_upd_free(upd);
 143                 } else {
 144                         break;
 145                 }
 146         }
 147
 148         /* No remaining deferred update */
 149         if (cfs_list_empty(list))
 150                 RETURN_EXIT;
 151
 152         CDEBUG(D_QUOTA, "%s: found deferred update record. "
 153                "version:"LPU64"/"LPU64", global:%d, qid:"LPU64"\n",
 154                qqi->qqi_qsd->qsd_svname, upd->qur_ver, ver,
 155                upd->qur_global, upd->qur_qid.qid_uid);
 156
 157         LASSERTF(upd->qur_ver > ver, "lur_ver:"LPU64", cur_ver:"LPU64"\n",
 158                  upd->qur_ver, ver);
 159
 160         /* Kick off the deferred udpate */
 161         if (upd->qur_ver == ver + 1) {
 162                 list_del_init(&upd->qur_link);
 163                 qsd_upd_add(qqi->qqi_qsd, upd);
 164         }
 165         EXIT;
 166 }
 167
 168 /* Bump version of global or slave index copy
 169  *
 170  * \param qqi    - qsd_qtype_info
 171  * \param ver    - version to be bumped to
 172  * \param global - global or slave index copy?
 173  */
 174 void qsd_bump_version(struct qsd_qtype_info *qqi, __u64 ver, bool global)
 175 {
 176         cfs_list_t      *list;
 177         __u64           *idx_ver;
 178
 179         idx_ver = global ? &qqi->qqi_glb_ver : &qqi->qqi_slv_ver;
 180         list    = global ? &qqi->qqi_deferred_glb : &qqi->qqi_deferred_slv;
 181
 182         cfs_write_lock(&qqi->qqi_qsd->qsd_lock);
 183         *idx_ver = ver;
 184         if (global)
 185                 qqi->qqi_glb_uptodate = 1;
 186         else
 187                 qqi->qqi_slv_uptodate = 1;
 188         qsd_kickoff_deferred(qqi, list, ver);
 189         cfs_write_unlock(&qqi->qqi_qsd->qsd_lock);
 190 }
 191
 192 /*
 193  * Schedule a commit of a lquota entry
 194  *
 195  * \param  qqi   - qsd_qtype_info
 196  * \param  lqe   - lquota_entry
 197  * \param  qid   - quota id
 198  * \param  rec   - global or slave record to be updated to disk
 199  * \param  ver   - new index file version
 200  * \param  global- ture : master record; false : slave record
 201  */
 202 void qsd_upd_schedule(struct qsd_qtype_info *qqi, struct lquota_entry *lqe,
 203                       union lquota_id *qid, union lquota_rec *rec, __u64 ver,
 204                       bool global)
 205 {
 206         struct qsd_upd_rec      *upd;
 207         struct qsd_instance     *qsd = qqi->qqi_qsd;
 208         __u64                    cur_ver;
 209         ENTRY;
 210
 211         CDEBUG(D_QUOTA, "%s: schedule update. global:%s, version:"LPU64"\n",
 212                qsd->qsd_svname, global ? "true" : "false", ver);
 213
 214         upd = qsd_upd_alloc(qqi, lqe, qid, rec, ver, global);
 215         if (upd == NULL)
 216                 RETURN_EXIT;
 217
 218         /* If we don't want update index version, no need to sort the
 219          * records in version order, just schedule the updates instantly. */
 220         if (ver == 0) {
 221                 cfs_write_lock(&qsd->qsd_lock);
 222                 qsd_upd_add(qsd, upd);
 223                 cfs_write_unlock(&qsd->qsd_lock);
 224                 RETURN_EXIT;
 225         }
 226
 227         cfs_write_lock(&qsd->qsd_lock);
 228
 229         cur_ver = global ? qqi->qqi_glb_ver : qqi->qqi_slv_ver;
 230
 231         if (ver <= cur_ver) {
 232                 if (global)
 233                         /* legitimate race between glimpse AST and
 234                          * reintegration */
 235                         CDEBUG(D_QUOTA, "%s: discarding glb update from glimpse"
 236                                " ver:"LPU64" local ver:"LPU64"\n",
 237                                qsd->qsd_svname, ver, cur_ver);
 238                 else
 239                         CERROR("%s: discard slv update, ver:"LPU64" local ver:"
 240                                LPU64"\n", qsd->qsd_svname, ver, cur_ver);
 241                 qsd_upd_free(upd);
 242         } else if ((ver == cur_ver + 1) && qqi->qqi_glb_uptodate &&
 243                    qqi->qqi_slv_uptodate) {
 244                 /* In order update, and reintegration has been done. */
 245                 qsd_upd_add(qsd, upd);
 246         } else {
 247                 /* Out of order update (the one with smaller version hasn't
 248                  * reached slave or hasn't been flushed to disk yet), or
 249                  * the reintegration is in progress. Defer the update. */
 250                 cfs_list_t *list = global ? &qqi->qqi_deferred_glb :
 251                                             &qqi->qqi_deferred_slv;
 252                 qsd_add_deferred(list, upd);
 253         }
 254
 255         cfs_write_unlock(&qsd->qsd_lock);
 256
 257         EXIT;
 258 }
 259
 260 static int qsd_process_upd(const struct lu_env *env, struct qsd_upd_rec *upd)
 261 {
 262         struct lquota_entry     *lqe = upd->qur_lqe;
 263         struct qsd_qtype_info   *qqi = upd->qur_qqi;
 264         int                      rc;
 265         ENTRY;
 266
 267         if (lqe == NULL) {
 268                 lqe = lqe_locate(env, qqi->qqi_site, &upd->qur_qid);
 269                 if (IS_ERR(lqe))
 270                         GOTO(out, rc = PTR_ERR(lqe));
 271         }
 272
 273         /* The in-memory lqe update for slave index copy isn't deferred,
 274          * we shouldn't touch it here. */
 275         if (upd->qur_global) {
 276                 rc = qsd_update_lqe(env, lqe, upd->qur_global, &upd->qur_rec);
 277                 if (rc)
 278                         GOTO(out, rc);
 279                 /* Report usage asynchronously */
 280                 if (lqe->lqe_enforced &&
 281                     !qsd_refresh_usage(env, lqe)) {
 282                         rc = qsd_dqacq(env, lqe, QSD_REP);
 283                         LQUOTA_DEBUG(lqe, "Report usage. rc:%d", rc);
 284                 }
 285         }
 286
 287         rc = qsd_update_index(env, qqi, &upd->qur_qid, upd->qur_global,
 288                               upd->qur_ver, &upd->qur_rec);
 289 out:
 290         if (lqe && !IS_ERR(lqe)) {
 291                 lqe_putref(lqe);
 292                 upd->qur_lqe = NULL;
 293         }
 294         RETURN(rc);
 295 }
 296
 297 void qsd_adjust_schedule(struct lquota_entry *lqe, bool defer, bool cancel)
 298 {
 299         struct qsd_instance     *qsd = lqe2qqi(lqe)->qqi_qsd;
 300         bool                     added = false;
 301
 302         lqe_getref(lqe);
 303         cfs_spin_lock(&qsd->qsd_adjust_lock);
 304
 305         /* the lqe is being queued for the per-ID lock cancel, we should
 306          * cancel the lock cancel and re-add it for quota adjust */
 307         if (!cfs_list_empty(&lqe->lqe_link) &&
 308             lqe->lqe_adjust_time == 0) {
 309                 cfs_list_del_init(&lqe->lqe_link);
 310                 lqe_putref(lqe);
 311         }
 312
 313         if (cfs_list_empty(&lqe->lqe_link)) {
 314                 if (cancel)
 315                         lqe->lqe_adjust_time = 0;
 316                 else
 317                         lqe->lqe_adjust_time = defer ?
 318                                 cfs_time_shift_64(QSD_WB_INTERVAL) :
 319                                 cfs_time_current_64();
 320                 /* lqe reference transfered to list */
 321                 if (defer)
 322                         cfs_list_add_tail(&lqe->lqe_link,
 323                                           &qsd->qsd_adjust_list);
 324                 else
 325                         cfs_list_add(&lqe->lqe_link, &qsd->qsd_adjust_list);
 326                 added = true;
 327         }
 328         cfs_spin_unlock(&qsd->qsd_adjust_lock);
 329
 330         if (added)
 331                 cfs_waitq_signal(&qsd->qsd_upd_thread.t_ctl_waitq);
 332         else
 333                 lqe_putref(lqe);
 334 }
 335
 336 /* return true if there is pending writeback records or the pending
 337  * adjust requests */
 338 static bool qsd_job_pending(struct qsd_instance *qsd, cfs_list_t *upd,
 339                             bool *uptodate)
 340 {
 341         bool    job_pending = false;
 342         int     qtype;
 343
 344         LASSERT(cfs_list_empty(upd));
 345         *uptodate = true;
 346
 347         cfs_spin_lock(&qsd->qsd_adjust_lock);
 348         if (!cfs_list_empty(&qsd->qsd_adjust_list)) {
 349                 struct lquota_entry *lqe;
 350                 lqe = cfs_list_entry(qsd->qsd_adjust_list.next,
 351                                      struct lquota_entry, lqe_link);
 352                 if (cfs_time_beforeq_64(lqe->lqe_adjust_time,
 353                                         cfs_time_current_64()))
 354                         job_pending = true;
 355         }
 356         cfs_spin_unlock(&qsd->qsd_adjust_lock);
 357
 358         cfs_write_lock(&qsd->qsd_lock);
 359         if (!cfs_list_empty(&qsd->qsd_upd_list)) {
 360                 cfs_list_splice_init(&qsd->qsd_upd_list, upd);
 361                 job_pending = true;
 362         }
 363
 364         for (qtype = USRQUOTA; qtype < MAXQUOTAS; qtype++) {
 365                 struct qsd_qtype_info *qqi = qsd->qsd_type_array[qtype];
 366
 367                 if (!qsd_type_enabled(qsd, qtype))
 368                         continue;
 369
 370                 if ((!qqi->qqi_glb_uptodate || !qqi->qqi_slv_uptodate) &&
 371                      !qqi->qqi_reint)
 372                         /* global or slave index not up to date and reint
 373                          * thread not running */
 374                         *uptodate = false;
 375         }
 376
 377         cfs_write_unlock(&qsd->qsd_lock);
 378         return job_pending;
 379 }
 380
 381 static int qsd_upd_thread(void *arg)
 382 {
 383         struct qsd_instance     *qsd = (struct qsd_instance *)arg;
 384         struct ptlrpc_thread    *thread = &qsd->qsd_upd_thread;
 385         struct l_wait_info       lwi;
 386         cfs_list_t               queue;
 387         struct qsd_upd_rec      *upd, *n;
 388         char                     pname[MTI_NAME_MAXLEN];
 389         struct lu_env           *env;
 390         int                      qtype, rc = 0;
 391         bool                     uptodate;
 392         struct lquota_entry     *lqe, *tmp;
 393         __u64                    cur_time;
 394         ENTRY;
 395
 396         OBD_ALLOC_PTR(env);
 397         if (env == NULL)
 398                 RETURN(-ENOMEM);
 399
 400         rc = lu_env_init(env, LCT_DT_THREAD);
 401         if (rc) {
 402                 CERROR("%s: Fail to init env.", qsd->qsd_svname);
 403                 OBD_FREE_PTR(env);
 404                 RETURN(rc);
 405         }
 406
 407         snprintf(pname, MTI_NAME_MAXLEN, "lquota_wb_%s", qsd->qsd_svname);
 408         cfs_daemonize(pname);
 409
 410         thread_set_flags(thread, SVC_RUNNING);
 411         cfs_waitq_signal(&thread->t_ctl_waitq);
 412
 413         CFS_INIT_LIST_HEAD(&queue);
 414         lwi = LWI_TIMEOUT(cfs_time_seconds(QSD_WB_INTERVAL), NULL, NULL);
 415         while (1) {
 416                 l_wait_event(thread->t_ctl_waitq,
 417                              qsd_job_pending(qsd, &queue, &uptodate) ||
 418                              !thread_is_running(thread), &lwi);
 419
 420                 cfs_list_for_each_entry_safe(upd, n, &queue, qur_link) {
 421                         cfs_list_del_init(&upd->qur_link);
 422                         qsd_process_upd(env, upd);
 423                         qsd_upd_free(upd);
 424                 }
 425
 426                 cfs_spin_lock(&qsd->qsd_adjust_lock);
 427                 cur_time = cfs_time_current_64();
 428                 cfs_list_for_each_entry_safe(lqe, tmp, &qsd->qsd_adjust_list,
 429                                              lqe_link) {
 430                         /* deferred items are sorted by time */
 431                         if (!cfs_time_beforeq_64(lqe->lqe_adjust_time,
 432                                                  cur_time))
 433                                 break;
 434
 435                         cfs_list_del_init(&lqe->lqe_link);
 436                         cfs_spin_unlock(&qsd->qsd_adjust_lock);
 437
 438                         if (thread_is_running(thread) && uptodate) {
 439                                 qsd_refresh_usage(env, lqe);
 440                                 if (lqe->lqe_adjust_time == 0)
 441                                         qsd_id_lock_cancel(env, lqe);
 442                                 else
 443                                         qsd_dqacq(env, lqe, QSD_ADJ);
 444                         }
 445
 446                         lqe_putref(lqe);
 447                         cfs_spin_lock(&qsd->qsd_adjust_lock);
 448                 }
 449                 cfs_spin_unlock(&qsd->qsd_adjust_lock);
 450
 451                 if (!thread_is_running(thread))
 452                         break;
 453
 454                 if (uptodate)
 455                         continue;
 456
 457                 for (qtype = USRQUOTA; qtype < MAXQUOTAS; qtype++)
 458                         qsd_start_reint_thread(qsd->qsd_type_array[qtype]);
 459         }
 460         lu_env_fini(env);
 461         OBD_FREE_PTR(env);
 462         thread_set_flags(thread, SVC_STOPPED);
 463         cfs_waitq_signal(&thread->t_ctl_waitq);
 464         RETURN(rc);
 465 }
 466
 467 int qsd_start_upd_thread(struct qsd_instance *qsd)
 468 {
 469         struct ptlrpc_thread    *thread = &qsd->qsd_upd_thread;
 470         struct l_wait_info       lwi = { 0 };
 471         int                      rc;
 472         ENTRY;
 473
 474         rc = cfs_create_thread(qsd_upd_thread, (void *)qsd, 0);
 475         if (rc < 0) {
 476                 CERROR("Fail to start quota update thread. rc: %d\n", rc);
 477                 thread_set_flags(thread, SVC_STOPPED);
 478                 RETURN(rc);
 479         }
 480
 481         l_wait_event(thread->t_ctl_waitq,
 482                      thread_is_running(thread) || thread_is_stopped(thread),
 483                      &lwi);
 484         RETURN(0);
 485 }
 486
 487 static void qsd_cleanup_deferred(struct qsd_instance *qsd)
 488 {
 489         int     qtype;
 490
 491         for (qtype = USRQUOTA; qtype < MAXQUOTAS; qtype++) {
 492                 struct qsd_upd_rec      *upd, *tmp;
 493                 struct qsd_qtype_info   *qqi = qsd->qsd_type_array[qtype];
 494
 495                 if (qqi == NULL)
 496                         continue;
 497
 498                 cfs_write_lock(&qsd->qsd_lock);
 499                 cfs_list_for_each_entry_safe(upd, tmp, &qqi->qqi_deferred_glb,
 500                                              qur_link) {
 501                         CWARN("%s: Free global deferred upd: ID:"LPU64", "
 502                               "ver:"LPU64"/"LPU64"\n", qsd->qsd_svname,
 503                               upd->qur_qid.qid_uid, upd->qur_ver,
 504                               qqi->qqi_glb_ver);
 505                         list_del_init(&upd->qur_link);
 506                         qsd_upd_free(upd);
 507                 }
 508                 cfs_list_for_each_entry_safe(upd, tmp, &qqi->qqi_deferred_slv,
 509                                              qur_link) {
 510                         CWARN("%s: Free slave deferred upd: ID:"LPU64", "
 511                               "ver:"LPU64"/"LPU64"\n", qsd->qsd_svname,
 512                               upd->qur_qid.qid_uid, upd->qur_ver,
 513                               qqi->qqi_slv_ver);
 514                         list_del_init(&upd->qur_link);
 515                         qsd_upd_free(upd);
 516                 }
 517                 cfs_write_unlock(&qsd->qsd_lock);
 518         }
 519 }
 520
 521 static void qsd_cleanup_adjust(struct qsd_instance *qsd)
 522 {
 523         struct lquota_entry     *lqe;
 524
 525         cfs_spin_lock(&qsd->qsd_adjust_lock);
 526         while (!cfs_list_empty(&qsd->qsd_adjust_list)) {
 527                 lqe = cfs_list_entry(qsd->qsd_adjust_list.next,
 528                                      struct lquota_entry, lqe_link);
 529                 cfs_list_del_init(&lqe->lqe_link);
 530                 lqe_putref(lqe);
 531         }
 532         cfs_spin_unlock(&qsd->qsd_adjust_lock);
 533 }
 534
 535 void qsd_stop_upd_thread(struct qsd_instance *qsd)
 536 {
 537         struct ptlrpc_thread    *thread = &qsd->qsd_upd_thread;
 538         struct l_wait_info       lwi    = { 0 };
 539
 540         if (!thread_is_stopped(thread)) {
 541                 thread_set_flags(thread, SVC_STOPPING);
 542                 cfs_waitq_signal(&thread->t_ctl_waitq);
 543
 544                 l_wait_event(thread->t_ctl_waitq, thread_is_stopped(thread),
 545                              &lwi);
 546         }
 547         qsd_cleanup_deferred(qsd);
 548         qsd_cleanup_adjust(qsd);
 549 }