drivers/lightnvm/pblk-gc.c

   1 // SPDX-License-Identifier: GPL-2.0
   2 /*
   3  * Copyright (C) 2016 CNEX Labs
   4  * Initial release: Javier Gonzalez <javier@cnexlabs.com>
   5  *                  Matias Bjorling <matias@cnexlabs.com>
   6  *
   7  * This program is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU General Public License version
   9  * 2 as published by the Free Software Foundation.
  10  *
  11  * This program is distributed in the hope that it will be useful, but
  12  * WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * General Public License for more details.
  15  *
  16  * pblk-gc.c - pblk's garbage collector
  17  */
  18
  19 #include "pblk.h"
  20 #include "pblk-trace.h"
  21 #include <linux/delay.h>
  22
  23
  24 static void pblk_gc_free_gc_rq(struct pblk_gc_rq *gc_rq)
  25 {
  26         if (gc_rq->data)
  27                 vfree(gc_rq->data);
  28         kfree(gc_rq);
  29 }
  30
  31 static int pblk_gc_write(struct pblk *pblk)
  32 {
  33         struct pblk_gc *gc = &pblk->gc;
  34         struct pblk_gc_rq *gc_rq, *tgc_rq;
  35         LIST_HEAD(w_list);
  36
  37         spin_lock(&gc->w_lock);
  38         if (list_empty(&gc->w_list)) {
  39                 spin_unlock(&gc->w_lock);
  40                 return 1;
  41         }
  42
  43         list_cut_position(&w_list, &gc->w_list, gc->w_list.prev);
  44         gc->w_entries = 0;
  45         spin_unlock(&gc->w_lock);
  46
  47         list_for_each_entry_safe(gc_rq, tgc_rq, &w_list, list) {
  48                 pblk_write_gc_to_cache(pblk, gc_rq);
  49                 list_del(&gc_rq->list);
  50                 kref_put(&gc_rq->line->ref, pblk_line_put);
  51                 pblk_gc_free_gc_rq(gc_rq);
  52         }
  53
  54         return 0;
  55 }
  56
  57 static void pblk_gc_writer_kick(struct pblk_gc *gc)
  58 {
  59         wake_up_process(gc->gc_writer_ts);
  60 }
  61
  62 static void pblk_put_line_back(struct pblk *pblk, struct pblk_line *line)
  63 {
  64         struct pblk_line_mgmt *l_mg = &pblk->l_mg;
  65         struct list_head *move_list;
  66
  67         spin_lock(&line->lock);
  68         WARN_ON(line->state != PBLK_LINESTATE_GC);
  69         line->state = PBLK_LINESTATE_CLOSED;
  70         trace_pblk_line_state(pblk_disk_name(pblk), line->id,
  71                                         line->state);
  72         move_list = pblk_line_gc_list(pblk, line);
  73         spin_unlock(&line->lock);
  74
  75         if (move_list) {
  76                 spin_lock(&l_mg->gc_lock);
  77                 list_add_tail(&line->list, move_list);
  78                 spin_unlock(&l_mg->gc_lock);
  79         }
  80 }
  81
  82 static void pblk_gc_line_ws(struct work_struct *work)
  83 {
  84         struct pblk_line_ws *gc_rq_ws = container_of(work,
  85                                                 struct pblk_line_ws, ws);
  86         struct pblk *pblk = gc_rq_ws->pblk;
  87         struct nvm_tgt_dev *dev = pblk->dev;
  88         struct nvm_geo *geo = &dev->geo;
  89         struct pblk_gc *gc = &pblk->gc;
  90         struct pblk_line *line = gc_rq_ws->line;
  91         struct pblk_gc_rq *gc_rq = gc_rq_ws->priv;
  92         int ret;
  93
  94         up(&gc->gc_sem);
  95
  96         gc_rq->data = vmalloc(array_size(gc_rq->nr_secs, geo->csecs));
  97         if (!gc_rq->data) {
  98                 pblk_err(pblk, "could not GC line:%d (%d/%d)\n",
  99                                         line->id, *line->vsc, gc_rq->nr_secs);
 100                 goto out;
 101         }
 102
 103         /* Read from GC victim block */
 104         ret = pblk_submit_read_gc(pblk, gc_rq);
 105         if (ret) {
 106                 pblk_err(pblk, "failed GC read in line:%d (err:%d)\n",
 107                                                                 line->id, ret);
 108                 goto out;
 109         }
 110
 111         if (!gc_rq->secs_to_gc)
 112                 goto out;
 113
 114 retry:
 115         spin_lock(&gc->w_lock);
 116         if (gc->w_entries >= PBLK_GC_RQ_QD) {
 117                 spin_unlock(&gc->w_lock);
 118                 pblk_gc_writer_kick(&pblk->gc);
 119                 usleep_range(128, 256);
 120                 goto retry;
 121         }
 122         gc->w_entries++;
 123         list_add_tail(&gc_rq->list, &gc->w_list);
 124         spin_unlock(&gc->w_lock);
 125
 126         pblk_gc_writer_kick(&pblk->gc);
 127
 128         kfree(gc_rq_ws);
 129         return;
 130
 131 out:
 132         pblk_gc_free_gc_rq(gc_rq);
 133         kref_put(&line->ref, pblk_line_put);
 134         kfree(gc_rq_ws);
 135 }
 136
 137 static __le64 *get_lba_list_from_emeta(struct pblk *pblk,
 138                                        struct pblk_line *line)
 139 {
 140         struct line_emeta *emeta_buf;
 141         struct pblk_line_mgmt *l_mg = &pblk->l_mg;
 142         struct pblk_line_meta *lm = &pblk->lm;
 143         unsigned int lba_list_size = lm->emeta_len[2];
 144         __le64 *lba_list;
 145         int ret;
 146
 147         emeta_buf = pblk_malloc(lm->emeta_len[0],
 148                                 l_mg->emeta_alloc_type, GFP_KERNEL);
 149         if (!emeta_buf)
 150                 return NULL;
 151
 152         ret = pblk_line_emeta_read(pblk, line, emeta_buf);
 153         if (ret) {
 154                 pblk_err(pblk, "line %d read emeta failed (%d)\n",
 155                                 line->id, ret);
 156                 pblk_mfree(emeta_buf, l_mg->emeta_alloc_type);
 157                 return NULL;
 158         }
 159
 160         /* If this read fails, it means that emeta is corrupted.
 161          * For now, leave the line untouched.
 162          * TODO: Implement a recovery routine that scans and moves
 163          * all sectors on the line.
 164          */
 165
 166         ret = pblk_recov_check_emeta(pblk, emeta_buf);
 167         if (ret) {
 168                 pblk_err(pblk, "inconsistent emeta (line %d)\n",
 169                                 line->id);
 170                 pblk_mfree(emeta_buf, l_mg->emeta_alloc_type);
 171                 return NULL;
 172         }
 173
 174         lba_list = pblk_malloc(lba_list_size,
 175                                l_mg->emeta_alloc_type, GFP_KERNEL);
 176         if (lba_list)
 177                 memcpy(lba_list, emeta_to_lbas(pblk, emeta_buf), lba_list_size);
 178
 179         pblk_mfree(emeta_buf, l_mg->emeta_alloc_type);
 180
 181         return lba_list;
 182 }
 183
 184 static void pblk_gc_line_prepare_ws(struct work_struct *work)
 185 {
 186         struct pblk_line_ws *line_ws = container_of(work, struct pblk_line_ws,
 187                                                                         ws);
 188         struct pblk *pblk = line_ws->pblk;
 189         struct pblk_line *line = line_ws->line;
 190         struct pblk_line_mgmt *l_mg = &pblk->l_mg;
 191         struct pblk_line_meta *lm = &pblk->lm;
 192         struct pblk_gc *gc = &pblk->gc;
 193         struct pblk_line_ws *gc_rq_ws;
 194         struct pblk_gc_rq *gc_rq;
 195         __le64 *lba_list;
 196         unsigned long *invalid_bitmap;
 197         int sec_left, nr_secs, bit;
 198
 199         invalid_bitmap = kmalloc(lm->sec_bitmap_len, GFP_KERNEL);
 200         if (!invalid_bitmap)
 201                 goto fail_free_ws;
 202
 203         if (line->w_err_gc->has_write_err) {
 204                 lba_list = line->w_err_gc->lba_list;
 205                 line->w_err_gc->lba_list = NULL;
 206         } else {
 207                 lba_list = get_lba_list_from_emeta(pblk, line);
 208                 if (!lba_list) {
 209                         pblk_err(pblk, "could not interpret emeta (line %d)\n",
 210                                         line->id);
 211                         goto fail_free_invalid_bitmap;
 212                 }
 213         }
 214
 215         spin_lock(&line->lock);
 216         bitmap_copy(invalid_bitmap, line->invalid_bitmap, lm->sec_per_line);
 217         sec_left = pblk_line_vsc(line);
 218         spin_unlock(&line->lock);
 219
 220         if (sec_left < 0) {
 221                 pblk_err(pblk, "corrupted GC line (%d)\n", line->id);
 222                 goto fail_free_lba_list;
 223         }
 224
 225         bit = -1;
 226 next_rq:
 227         gc_rq = kmalloc(sizeof(struct pblk_gc_rq), GFP_KERNEL);
 228         if (!gc_rq)
 229                 goto fail_free_lba_list;
 230
 231         nr_secs = 0;
 232         do {
 233                 bit = find_next_zero_bit(invalid_bitmap, lm->sec_per_line,
 234                                                                 bit + 1);
 235                 if (bit > line->emeta_ssec)
 236                         break;
 237
 238                 gc_rq->paddr_list[nr_secs] = bit;
 239                 gc_rq->lba_list[nr_secs++] = le64_to_cpu(lba_list[bit]);
 240         } while (nr_secs < pblk->max_write_pgs);
 241
 242         if (unlikely(!nr_secs)) {
 243                 kfree(gc_rq);
 244                 goto out;
 245         }
 246
 247         gc_rq->nr_secs = nr_secs;
 248         gc_rq->line = line;
 249
 250         gc_rq_ws = kmalloc(sizeof(struct pblk_line_ws), GFP_KERNEL);
 251         if (!gc_rq_ws)
 252                 goto fail_free_gc_rq;
 253
 254         gc_rq_ws->pblk = pblk;
 255         gc_rq_ws->line = line;
 256         gc_rq_ws->priv = gc_rq;
 257
 258         /* The write GC path can be much slower than the read GC one due to
 259          * the budget imposed by the rate-limiter. Balance in case that we get
 260          * back pressure from the write GC path.
 261          */
 262         while (down_timeout(&gc->gc_sem, msecs_to_jiffies(30000)))
 263                 io_schedule();
 264
 265         kref_get(&line->ref);
 266
 267         INIT_WORK(&gc_rq_ws->ws, pblk_gc_line_ws);
 268         queue_work(gc->gc_line_reader_wq, &gc_rq_ws->ws);
 269
 270         sec_left -= nr_secs;
 271         if (sec_left > 0)
 272                 goto next_rq;
 273
 274 out:
 275         pblk_mfree(lba_list, l_mg->emeta_alloc_type);
 276         kfree(line_ws);
 277         kfree(invalid_bitmap);
 278
 279         kref_put(&line->ref, pblk_line_put);
 280         atomic_dec(&gc->read_inflight_gc);
 281
 282         return;
 283
 284 fail_free_gc_rq:
 285         kfree(gc_rq);
 286 fail_free_lba_list:
 287         pblk_mfree(lba_list, l_mg->emeta_alloc_type);
 288 fail_free_invalid_bitmap:
 289         kfree(invalid_bitmap);
 290 fail_free_ws:
 291         kfree(line_ws);
 292
 293         pblk_put_line_back(pblk, line);
 294         kref_put(&line->ref, pblk_line_put);
 295         atomic_dec(&gc->read_inflight_gc);
 296
 297         pblk_err(pblk, "failed to GC line %d\n", line->id);
 298 }
 299
 300 static int pblk_gc_line(struct pblk *pblk, struct pblk_line *line)
 301 {
 302         struct pblk_gc *gc = &pblk->gc;
 303         struct pblk_line_ws *line_ws;
 304
 305         pblk_debug(pblk, "line '%d' being reclaimed for GC\n", line->id);
 306
 307         line_ws = kmalloc(sizeof(struct pblk_line_ws), GFP_KERNEL);
 308         if (!line_ws)
 309                 return -ENOMEM;
 310
 311         line_ws->pblk = pblk;
 312         line_ws->line = line;
 313
 314         atomic_inc(&gc->pipeline_gc);
 315         INIT_WORK(&line_ws->ws, pblk_gc_line_prepare_ws);
 316         queue_work(gc->gc_reader_wq, &line_ws->ws);
 317
 318         return 0;
 319 }
 320
 321 static void pblk_gc_reader_kick(struct pblk_gc *gc)
 322 {
 323         wake_up_process(gc->gc_reader_ts);
 324 }
 325
 326 static void pblk_gc_kick(struct pblk *pblk)
 327 {
 328         struct pblk_gc *gc = &pblk->gc;
 329
 330         pblk_gc_writer_kick(gc);
 331         pblk_gc_reader_kick(gc);
 332
 333         /* If we're shutting down GC, let's not start it up again */
 334         if (gc->gc_enabled) {
 335                 wake_up_process(gc->gc_ts);
 336                 mod_timer(&gc->gc_timer,
 337                           jiffies + msecs_to_jiffies(GC_TIME_MSECS));
 338         }
 339 }
 340
 341 static int pblk_gc_read(struct pblk *pblk)
 342 {
 343         struct pblk_gc *gc = &pblk->gc;
 344         struct pblk_line *line;
 345
 346         spin_lock(&gc->r_lock);
 347         if (list_empty(&gc->r_list)) {
 348                 spin_unlock(&gc->r_lock);
 349                 return 1;
 350         }
 351
 352         line = list_first_entry(&gc->r_list, struct pblk_line, list);
 353         list_del(&line->list);
 354         spin_unlock(&gc->r_lock);
 355
 356         pblk_gc_kick(pblk);
 357
 358         if (pblk_gc_line(pblk, line))
 359                 pblk_err(pblk, "failed to GC line %d\n", line->id);
 360
 361         return 0;
 362 }
 363
 364 static struct pblk_line *pblk_gc_get_victim_line(struct pblk *pblk,
 365                                                  struct list_head *group_list)
 366 {
 367         struct pblk_line *line, *victim;
 368         int line_vsc, victim_vsc;
 369
 370         victim = list_first_entry(group_list, struct pblk_line, list);
 371         list_for_each_entry(line, group_list, list) {
 372                 line_vsc = le32_to_cpu(*line->vsc);
 373                 victim_vsc = le32_to_cpu(*victim->vsc);
 374                 if (line_vsc < victim_vsc)
 375                         victim = line;
 376         }
 377
 378         return victim;
 379 }
 380
 381 static bool pblk_gc_should_run(struct pblk_gc *gc, struct pblk_rl *rl)
 382 {
 383         unsigned int nr_blocks_free, nr_blocks_need;
 384         unsigned int werr_lines = atomic_read(&rl->werr_lines);
 385
 386         nr_blocks_need = pblk_rl_high_thrs(rl);
 387         nr_blocks_free = pblk_rl_nr_free_blks(rl);
 388
 389         /* This is not critical, no need to take lock here */
 390         return ((werr_lines > 0) ||
 391                 ((gc->gc_active) && (nr_blocks_need > nr_blocks_free)));
 392 }
 393
 394 void pblk_gc_free_full_lines(struct pblk *pblk)
 395 {
 396         struct pblk_line_mgmt *l_mg = &pblk->l_mg;
 397         struct pblk_gc *gc = &pblk->gc;
 398         struct pblk_line *line;
 399
 400         do {
 401                 spin_lock(&l_mg->gc_lock);
 402                 if (list_empty(&l_mg->gc_full_list)) {
 403                         spin_unlock(&l_mg->gc_lock);
 404                         return;
 405                 }
 406
 407                 line = list_first_entry(&l_mg->gc_full_list,
 408                                                         struct pblk_line, list);
 409
 410                 spin_lock(&line->lock);
 411                 WARN_ON(line->state != PBLK_LINESTATE_CLOSED);
 412                 line->state = PBLK_LINESTATE_GC;
 413                 trace_pblk_line_state(pblk_disk_name(pblk), line->id,
 414                                         line->state);
 415                 spin_unlock(&line->lock);
 416
 417                 list_del(&line->list);
 418                 spin_unlock(&l_mg->gc_lock);
 419
 420                 atomic_inc(&gc->pipeline_gc);
 421                 kref_put(&line->ref, pblk_line_put);
 422         } while (1);
 423 }
 424
 425 /*
 426  * Lines with no valid sectors will be returned to the free list immediately. If
 427  * GC is activated - either because the free block count is under the determined
 428  * threshold, or because it is being forced from user space - only lines with a
 429  * high count of invalid sectors will be recycled.
 430  */
 431 static void pblk_gc_run(struct pblk *pblk)
 432 {
 433         struct pblk_line_mgmt *l_mg = &pblk->l_mg;
 434         struct pblk_gc *gc = &pblk->gc;
 435         struct pblk_line *line;
 436         struct list_head *group_list;
 437         bool run_gc;
 438         int read_inflight_gc, gc_group = 0, prev_group = 0;
 439
 440         pblk_gc_free_full_lines(pblk);
 441
 442         run_gc = pblk_gc_should_run(&pblk->gc, &pblk->rl);
 443         if (!run_gc || (atomic_read(&gc->read_inflight_gc) >= PBLK_GC_L_QD))
 444                 return;
 445
 446 next_gc_group:
 447         group_list = l_mg->gc_lists[gc_group++];
 448
 449         do {
 450                 spin_lock(&l_mg->gc_lock);
 451                 if (list_empty(group_list)) {
 452                         spin_unlock(&l_mg->gc_lock);
 453                         break;
 454                 }
 455
 456                 line = pblk_gc_get_victim_line(pblk, group_list);
 457
 458                 spin_lock(&line->lock);
 459                 WARN_ON(line->state != PBLK_LINESTATE_CLOSED);
 460                 line->state = PBLK_LINESTATE_GC;
 461                 trace_pblk_line_state(pblk_disk_name(pblk), line->id,
 462                                         line->state);
 463                 spin_unlock(&line->lock);
 464
 465                 list_del(&line->list);
 466                 spin_unlock(&l_mg->gc_lock);
 467
 468                 spin_lock(&gc->r_lock);
 469                 list_add_tail(&line->list, &gc->r_list);
 470                 spin_unlock(&gc->r_lock);
 471
 472                 read_inflight_gc = atomic_inc_return(&gc->read_inflight_gc);
 473                 pblk_gc_reader_kick(gc);
 474
 475                 prev_group = 1;
 476
 477                 /* No need to queue up more GC lines than we can handle */
 478                 run_gc = pblk_gc_should_run(&pblk->gc, &pblk->rl);
 479                 if (!run_gc || read_inflight_gc >= PBLK_GC_L_QD)
 480                         break;
 481         } while (1);
 482
 483         if (!prev_group && pblk->rl.rb_state > gc_group &&
 484                                                 gc_group < PBLK_GC_NR_LISTS)
 485                 goto next_gc_group;
 486 }
 487
 488 static void pblk_gc_timer(struct timer_list *t)
 489 {
 490         struct pblk *pblk = from_timer(pblk, t, gc.gc_timer);
 491
 492         pblk_gc_kick(pblk);
 493 }
 494
 495 static int pblk_gc_ts(void *data)
 496 {
 497         struct pblk *pblk = data;
 498
 499         while (!kthread_should_stop()) {
 500                 pblk_gc_run(pblk);
 501                 set_current_state(TASK_INTERRUPTIBLE);
 502                 io_schedule();
 503         }
 504
 505         return 0;
 506 }
 507
 508 static int pblk_gc_writer_ts(void *data)
 509 {
 510         struct pblk *pblk = data;
 511
 512         while (!kthread_should_stop()) {
 513                 if (!pblk_gc_write(pblk))
 514                         continue;
 515                 set_current_state(TASK_INTERRUPTIBLE);
 516                 io_schedule();
 517         }
 518
 519         return 0;
 520 }
 521
 522 static int pblk_gc_reader_ts(void *data)
 523 {
 524         struct pblk *pblk = data;
 525         struct pblk_gc *gc = &pblk->gc;
 526
 527         while (!kthread_should_stop()) {
 528                 if (!pblk_gc_read(pblk))
 529                         continue;
 530                 set_current_state(TASK_INTERRUPTIBLE);
 531                 io_schedule();
 532         }
 533
 534 #ifdef CONFIG_NVM_PBLK_DEBUG
 535         pblk_info(pblk, "flushing gc pipeline, %d lines left\n",
 536                 atomic_read(&gc->pipeline_gc));
 537 #endif
 538
 539         do {
 540                 if (!atomic_read(&gc->pipeline_gc))
 541                         break;
 542
 543                 schedule();
 544         } while (1);
 545
 546         return 0;
 547 }
 548
 549 static void pblk_gc_start(struct pblk *pblk)
 550 {
 551         pblk->gc.gc_active = 1;
 552         pblk_debug(pblk, "gc start\n");
 553 }
 554
 555 void pblk_gc_should_start(struct pblk *pblk)
 556 {
 557         struct pblk_gc *gc = &pblk->gc;
 558
 559         if (gc->gc_enabled && !gc->gc_active) {
 560                 pblk_gc_start(pblk);
 561                 pblk_gc_kick(pblk);
 562         }
 563 }
 564
 565 void pblk_gc_should_stop(struct pblk *pblk)
 566 {
 567         struct pblk_gc *gc = &pblk->gc;
 568
 569         if (gc->gc_active && !gc->gc_forced)
 570                 gc->gc_active = 0;
 571 }
 572
 573 void pblk_gc_should_kick(struct pblk *pblk)
 574 {
 575         pblk_rl_update_rates(&pblk->rl);
 576 }
 577
 578 void pblk_gc_sysfs_state_show(struct pblk *pblk, int *gc_enabled,
 579                               int *gc_active)
 580 {
 581         struct pblk_gc *gc = &pblk->gc;
 582
 583         spin_lock(&gc->lock);
 584         *gc_enabled = gc->gc_enabled;
 585         *gc_active = gc->gc_active;
 586         spin_unlock(&gc->lock);
 587 }
 588
 589 int pblk_gc_sysfs_force(struct pblk *pblk, int force)
 590 {
 591         struct pblk_gc *gc = &pblk->gc;
 592
 593         if (force < 0 || force > 1)
 594                 return -EINVAL;
 595
 596         spin_lock(&gc->lock);
 597         gc->gc_forced = force;
 598
 599         if (force)
 600                 gc->gc_enabled = 1;
 601         else
 602                 gc->gc_enabled = 0;
 603         spin_unlock(&gc->lock);
 604
 605         pblk_gc_should_start(pblk);
 606
 607         return 0;
 608 }
 609
 610 int pblk_gc_init(struct pblk *pblk)
 611 {
 612         struct pblk_gc *gc = &pblk->gc;
 613         int ret;
 614
 615         gc->gc_ts = kthread_create(pblk_gc_ts, pblk, "pblk-gc-ts");
 616         if (IS_ERR(gc->gc_ts)) {
 617                 pblk_err(pblk, "could not allocate GC main kthread\n");
 618                 return PTR_ERR(gc->gc_ts);
 619         }
 620
 621         gc->gc_writer_ts = kthread_create(pblk_gc_writer_ts, pblk,
 622                                                         "pblk-gc-writer-ts");
 623         if (IS_ERR(gc->gc_writer_ts)) {
 624                 pblk_err(pblk, "could not allocate GC writer kthread\n");
 625                 ret = PTR_ERR(gc->gc_writer_ts);
 626                 goto fail_free_main_kthread;
 627         }
 628
 629         gc->gc_reader_ts = kthread_create(pblk_gc_reader_ts, pblk,
 630                                                         "pblk-gc-reader-ts");
 631         if (IS_ERR(gc->gc_reader_ts)) {
 632                 pblk_err(pblk, "could not allocate GC reader kthread\n");
 633                 ret = PTR_ERR(gc->gc_reader_ts);
 634                 goto fail_free_writer_kthread;
 635         }
 636
 637         timer_setup(&gc->gc_timer, pblk_gc_timer, 0);
 638         mod_timer(&gc->gc_timer, jiffies + msecs_to_jiffies(GC_TIME_MSECS));
 639
 640         gc->gc_active = 0;
 641         gc->gc_forced = 0;
 642         gc->gc_enabled = 1;
 643         gc->w_entries = 0;
 644         atomic_set(&gc->read_inflight_gc, 0);
 645         atomic_set(&gc->pipeline_gc, 0);
 646
 647         /* Workqueue that reads valid sectors from a line and submit them to the
 648          * GC writer to be recycled.
 649          */
 650         gc->gc_line_reader_wq = alloc_workqueue("pblk-gc-line-reader-wq",
 651                         WQ_MEM_RECLAIM | WQ_UNBOUND, PBLK_GC_MAX_READERS);
 652         if (!gc->gc_line_reader_wq) {
 653                 pblk_err(pblk, "could not allocate GC line reader workqueue\n");
 654                 ret = -ENOMEM;
 655                 goto fail_free_reader_kthread;
 656         }
 657
 658         /* Workqueue that prepare lines for GC */
 659         gc->gc_reader_wq = alloc_workqueue("pblk-gc-line_wq",
 660                                         WQ_MEM_RECLAIM | WQ_UNBOUND, 1);
 661         if (!gc->gc_reader_wq) {
 662                 pblk_err(pblk, "could not allocate GC reader workqueue\n");
 663                 ret = -ENOMEM;
 664                 goto fail_free_reader_line_wq;
 665         }
 666
 667         spin_lock_init(&gc->lock);
 668         spin_lock_init(&gc->w_lock);
 669         spin_lock_init(&gc->r_lock);
 670
 671         sema_init(&gc->gc_sem, PBLK_GC_RQ_QD);
 672
 673         INIT_LIST_HEAD(&gc->w_list);
 674         INIT_LIST_HEAD(&gc->r_list);
 675
 676         return 0;
 677
 678 fail_free_reader_line_wq:
 679         destroy_workqueue(gc->gc_line_reader_wq);
 680 fail_free_reader_kthread:
 681         kthread_stop(gc->gc_reader_ts);
 682 fail_free_writer_kthread:
 683         kthread_stop(gc->gc_writer_ts);
 684 fail_free_main_kthread:
 685         kthread_stop(gc->gc_ts);
 686
 687         return ret;
 688 }
 689
 690 void pblk_gc_exit(struct pblk *pblk, bool graceful)
 691 {
 692         struct pblk_gc *gc = &pblk->gc;
 693
 694         gc->gc_enabled = 0;
 695         del_timer_sync(&gc->gc_timer);
 696         gc->gc_active = 0;
 697
 698         if (gc->gc_ts)
 699                 kthread_stop(gc->gc_ts);
 700
 701         if (gc->gc_reader_ts)
 702                 kthread_stop(gc->gc_reader_ts);
 703
 704         if (graceful) {
 705                 flush_workqueue(gc->gc_reader_wq);
 706                 flush_workqueue(gc->gc_line_reader_wq);
 707         }
 708
 709         destroy_workqueue(gc->gc_reader_wq);
 710         destroy_workqueue(gc->gc_line_reader_wq);
 711
 712         if (gc->gc_writer_ts)
 713                 kthread_stop(gc->gc_writer_ts);
 714 }