fs/fscache/operation.c

   1 /* FS-Cache worker operation management routines
   2  *
   3  * Copyright (C) 2008 Red Hat, Inc. All Rights Reserved.
   4  * Written by David Howells (dhowells@redhat.com)
   5  *
   6  * This program is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU General Public License
   8  * as published by the Free Software Foundation; either version
   9  * 2 of the License, or (at your option) any later version.
  10  *
  11  * See Documentation/filesystems/caching/operations.txt
  12  */
  13
  14 #define FSCACHE_DEBUG_LEVEL OPERATION
  15 #include <linux/module.h>
  16 #include <linux/seq_file.h>
  17 #include <linux/slab.h>
  18 #include "internal.h"
  19
  20 atomic_t fscache_op_debug_id;
  21 EXPORT_SYMBOL(fscache_op_debug_id);
  22
  23 /**
  24  * fscache_enqueue_operation - Enqueue an operation for processing
  25  * @op: The operation to enqueue
  26  *
  27  * Enqueue an operation for processing by the FS-Cache thread pool.
  28  *
  29  * This will get its own ref on the object.
  30  */
  31 void fscache_enqueue_operation(struct fscache_operation *op)
  32 {
  33         _enter("{OBJ%x OP%x,%u}",
  34                op->object->debug_id, op->debug_id, atomic_read(&op->usage));
  35
  36         ASSERT(list_empty(&op->pend_link));
  37         ASSERT(op->processor != NULL);
  38         ASSERTCMP(op->object->state, >=, FSCACHE_OBJECT_AVAILABLE);
  39         ASSERTCMP(atomic_read(&op->usage), >, 0);
  40         ASSERTCMP(op->state, ==, FSCACHE_OP_ST_IN_PROGRESS);
  41
  42         fscache_stat(&fscache_n_op_enqueue);
  43         switch (op->flags & FSCACHE_OP_TYPE) {
  44         case FSCACHE_OP_ASYNC:
  45                 _debug("queue async");
  46                 atomic_inc(&op->usage);
  47                 if (!queue_work(fscache_op_wq, &op->work))
  48                         fscache_put_operation(op);
  49                 break;
  50         case FSCACHE_OP_MYTHREAD:
  51                 _debug("queue for caller's attention");
  52                 break;
  53         default:
  54                 printk(KERN_ERR "FS-Cache: Unexpected op type %lx",
  55                        op->flags);
  56                 BUG();
  57                 break;
  58         }
  59 }
  60 EXPORT_SYMBOL(fscache_enqueue_operation);
  61
  62 /*
  63  * start an op running
  64  */
  65 static void fscache_run_op(struct fscache_object *object,
  66                            struct fscache_operation *op)
  67 {
  68         ASSERTCMP(op->state, ==, FSCACHE_OP_ST_PENDING);
  69
  70         op->state = FSCACHE_OP_ST_IN_PROGRESS;
  71         object->n_in_progress++;
  72         if (test_and_clear_bit(FSCACHE_OP_WAITING, &op->flags))
  73                 wake_up_bit(&op->flags, FSCACHE_OP_WAITING);
  74         if (op->processor)
  75                 fscache_enqueue_operation(op);
  76         fscache_stat(&fscache_n_op_run);
  77 }
  78
  79 /*
  80  * submit an exclusive operation for an object
  81  * - other ops are excluded from running simultaneously with this one
  82  * - this gets any extra refs it needs on an op
  83  */
  84 int fscache_submit_exclusive_op(struct fscache_object *object,
  85                                 struct fscache_operation *op)
  86 {
  87         int ret;
  88
  89         _enter("{OBJ%x OP%x},", object->debug_id, op->debug_id);
  90
  91         ASSERTCMP(op->state, ==, FSCACHE_OP_ST_INITIALISED);
  92         ASSERTCMP(atomic_read(&op->usage), >, 0);
  93
  94         spin_lock(&object->lock);
  95         ASSERTCMP(object->n_ops, >=, object->n_in_progress);
  96         ASSERTCMP(object->n_ops, >=, object->n_exclusive);
  97         ASSERT(list_empty(&op->pend_link));
  98
  99         op->state = FSCACHE_OP_ST_PENDING;
 100         if (fscache_object_is_active(object)) {
 101                 op->object = object;
 102                 object->n_ops++;
 103                 object->n_exclusive++;  /* reads and writes must wait */
 104
 105                 if (object->n_in_progress > 0) {
 106                         atomic_inc(&op->usage);
 107                         list_add_tail(&op->pend_link, &object->pending_ops);
 108                         fscache_stat(&fscache_n_op_pend);
 109                 } else if (!list_empty(&object->pending_ops)) {
 110                         atomic_inc(&op->usage);
 111                         list_add_tail(&op->pend_link, &object->pending_ops);
 112                         fscache_stat(&fscache_n_op_pend);
 113                         fscache_start_operations(object);
 114                 } else {
 115                         ASSERTCMP(object->n_in_progress, ==, 0);
 116                         fscache_run_op(object, op);
 117                 }
 118
 119                 /* need to issue a new write op after this */
 120                 clear_bit(FSCACHE_OBJECT_PENDING_WRITE, &object->flags);
 121                 ret = 0;
 122         } else if (object->state == FSCACHE_OBJECT_CREATING) {
 123                 op->object = object;
 124                 object->n_ops++;
 125                 object->n_exclusive++;  /* reads and writes must wait */
 126                 atomic_inc(&op->usage);
 127                 list_add_tail(&op->pend_link, &object->pending_ops);
 128                 fscache_stat(&fscache_n_op_pend);
 129                 ret = 0;
 130         } else {
 131                 /* If we're in any other state, there must have been an I/O
 132                  * error of some nature.
 133                  */
 134                 ASSERT(test_bit(FSCACHE_IOERROR, &object->cache->flags));
 135                 ret = -EIO;
 136         }
 137
 138         spin_unlock(&object->lock);
 139         return ret;
 140 }
 141
 142 /*
 143  * report an unexpected submission
 144  */
 145 static void fscache_report_unexpected_submission(struct fscache_object *object,
 146                                                  struct fscache_operation *op,
 147                                                  unsigned long ostate)
 148 {
 149         static bool once_only;
 150         struct fscache_operation *p;
 151         unsigned n;
 152
 153         if (once_only)
 154                 return;
 155         once_only = true;
 156
 157         kdebug("unexpected submission OP%x [OBJ%x %s]",
 158                op->debug_id, object->debug_id,
 159                fscache_object_states[object->state]);
 160         kdebug("objstate=%s [%s]",
 161                fscache_object_states[object->state],
 162                fscache_object_states[ostate]);
 163         kdebug("objflags=%lx", object->flags);
 164         kdebug("objevent=%lx [%lx]", object->events, object->event_mask);
 165         kdebug("ops=%u inp=%u exc=%u",
 166                object->n_ops, object->n_in_progress, object->n_exclusive);
 167
 168         if (!list_empty(&object->pending_ops)) {
 169                 n = 0;
 170                 list_for_each_entry(p, &object->pending_ops, pend_link) {
 171                         ASSERTCMP(p->object, ==, object);
 172                         kdebug("%p %p", op->processor, op->release);
 173                         n++;
 174                 }
 175
 176                 kdebug("n=%u", n);
 177         }
 178
 179         dump_stack();
 180 }
 181
 182 /*
 183  * submit an operation for an object
 184  * - objects may be submitted only in the following states:
 185  *   - during object creation (write ops may be submitted)
 186  *   - whilst the object is active
 187  *   - after an I/O error incurred in one of the two above states (op rejected)
 188  * - this gets any extra refs it needs on an op
 189  */
 190 int fscache_submit_op(struct fscache_object *object,
 191                       struct fscache_operation *op)
 192 {
 193         unsigned long ostate;
 194         int ret;
 195
 196         _enter("{OBJ%x OP%x},{%u}",
 197                object->debug_id, op->debug_id, atomic_read(&op->usage));
 198
 199         ASSERTCMP(op->state, ==, FSCACHE_OP_ST_INITIALISED);
 200         ASSERTCMP(atomic_read(&op->usage), >, 0);
 201
 202         spin_lock(&object->lock);
 203         ASSERTCMP(object->n_ops, >=, object->n_in_progress);
 204         ASSERTCMP(object->n_ops, >=, object->n_exclusive);
 205         ASSERT(list_empty(&op->pend_link));
 206
 207         ostate = object->state;
 208         smp_rmb();
 209
 210         op->state = FSCACHE_OP_ST_PENDING;
 211         if (fscache_object_is_active(object)) {
 212                 op->object = object;
 213                 object->n_ops++;
 214
 215                 if (object->n_exclusive > 0) {
 216                         atomic_inc(&op->usage);
 217                         list_add_tail(&op->pend_link, &object->pending_ops);
 218                         fscache_stat(&fscache_n_op_pend);
 219                 } else if (!list_empty(&object->pending_ops)) {
 220                         atomic_inc(&op->usage);
 221                         list_add_tail(&op->pend_link, &object->pending_ops);
 222                         fscache_stat(&fscache_n_op_pend);
 223                         fscache_start_operations(object);
 224                 } else {
 225                         ASSERTCMP(object->n_exclusive, ==, 0);
 226                         fscache_run_op(object, op);
 227                 }
 228                 ret = 0;
 229         } else if (object->state == FSCACHE_OBJECT_CREATING) {
 230                 op->object = object;
 231                 object->n_ops++;
 232                 atomic_inc(&op->usage);
 233                 list_add_tail(&op->pend_link, &object->pending_ops);
 234                 fscache_stat(&fscache_n_op_pend);
 235                 ret = 0;
 236         } else if (object->state == FSCACHE_OBJECT_DYING ||
 237                    object->state == FSCACHE_OBJECT_LC_DYING ||
 238                    object->state == FSCACHE_OBJECT_WITHDRAWING) {
 239                 fscache_stat(&fscache_n_op_rejected);
 240                 op->state = FSCACHE_OP_ST_CANCELLED;
 241                 ret = -ENOBUFS;
 242         } else if (!test_bit(FSCACHE_IOERROR, &object->cache->flags)) {
 243                 fscache_report_unexpected_submission(object, op, ostate);
 244                 ASSERT(!fscache_object_is_active(object));
 245                 op->state = FSCACHE_OP_ST_CANCELLED;
 246                 ret = -ENOBUFS;
 247         } else {
 248                 op->state = FSCACHE_OP_ST_CANCELLED;
 249                 ret = -ENOBUFS;
 250         }
 251
 252         spin_unlock(&object->lock);
 253         return ret;
 254 }
 255
 256 /*
 257  * queue an object for withdrawal on error, aborting all following asynchronous
 258  * operations
 259  */
 260 void fscache_abort_object(struct fscache_object *object)
 261 {
 262         _enter("{OBJ%x}", object->debug_id);
 263
 264         fscache_raise_event(object, FSCACHE_OBJECT_EV_ERROR);
 265 }
 266
 267 /*
 268  * jump start the operation processing on an object
 269  * - caller must hold object->lock
 270  */
 271 void fscache_start_operations(struct fscache_object *object)
 272 {
 273         struct fscache_operation *op;
 274         bool stop = false;
 275
 276         while (!list_empty(&object->pending_ops) && !stop) {
 277                 op = list_entry(object->pending_ops.next,
 278                                 struct fscache_operation, pend_link);
 279
 280                 if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags)) {
 281                         if (object->n_in_progress > 0)
 282                                 break;
 283                         stop = true;
 284                 }
 285                 list_del_init(&op->pend_link);
 286                 fscache_run_op(object, op);
 287
 288                 /* the pending queue was holding a ref on the object */
 289                 fscache_put_operation(op);
 290         }
 291
 292         ASSERTCMP(object->n_in_progress, <=, object->n_ops);
 293
 294         _debug("woke %d ops on OBJ%x",
 295                object->n_in_progress, object->debug_id);
 296 }
 297
 298 /*
 299  * cancel an operation that's pending on an object
 300  */
 301 int fscache_cancel_op(struct fscache_operation *op,
 302                       void (*do_cancel)(struct fscache_operation *))
 303 {
 304         struct fscache_object *object = op->object;
 305         int ret;
 306
 307         _enter("OBJ%x OP%x}", op->object->debug_id, op->debug_id);
 308
 309         ASSERTCMP(op->state, >=, FSCACHE_OP_ST_PENDING);
 310         ASSERTCMP(op->state, !=, FSCACHE_OP_ST_CANCELLED);
 311         ASSERTCMP(atomic_read(&op->usage), >, 0);
 312
 313         spin_lock(&object->lock);
 314
 315         ret = -EBUSY;
 316         if (op->state == FSCACHE_OP_ST_PENDING) {
 317                 ASSERT(!list_empty(&op->pend_link));
 318                 fscache_stat(&fscache_n_op_cancelled);
 319                 list_del_init(&op->pend_link);
 320                 if (do_cancel)
 321                         do_cancel(op);
 322                 op->state = FSCACHE_OP_ST_CANCELLED;
 323                 if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags))
 324                         object->n_exclusive--;
 325                 if (test_and_clear_bit(FSCACHE_OP_WAITING, &op->flags))
 326                         wake_up_bit(&op->flags, FSCACHE_OP_WAITING);
 327                 fscache_put_operation(op);
 328                 ret = 0;
 329         }
 330
 331         spin_unlock(&object->lock);
 332         _leave(" = %d", ret);
 333         return ret;
 334 }
 335
 336 /*
 337  * Cancel all pending operations on an object
 338  */
 339 void fscache_cancel_all_ops(struct fscache_object *object)
 340 {
 341         struct fscache_operation *op;
 342
 343         _enter("OBJ%x", object->debug_id);
 344
 345         spin_lock(&object->lock);
 346
 347         while (!list_empty(&object->pending_ops)) {
 348                 op = list_entry(object->pending_ops.next,
 349                                 struct fscache_operation, pend_link);
 350                 fscache_stat(&fscache_n_op_cancelled);
 351                 list_del_init(&op->pend_link);
 352
 353                 ASSERTCMP(op->state, ==, FSCACHE_OP_ST_PENDING);
 354                 op->state = FSCACHE_OP_ST_CANCELLED;
 355
 356                 if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags))
 357                         object->n_exclusive--;
 358                 if (test_and_clear_bit(FSCACHE_OP_WAITING, &op->flags))
 359                         wake_up_bit(&op->flags, FSCACHE_OP_WAITING);
 360                 fscache_put_operation(op);
 361                 cond_resched_lock(&object->lock);
 362         }
 363
 364         spin_unlock(&object->lock);
 365         _leave("");
 366 }
 367
 368 /*
 369  * Record the completion or cancellation of an in-progress operation.
 370  */
 371 void fscache_op_complete(struct fscache_operation *op, bool cancelled)
 372 {
 373         struct fscache_object *object = op->object;
 374
 375         _enter("OBJ%x", object->debug_id);
 376
 377         ASSERTCMP(op->state, ==, FSCACHE_OP_ST_IN_PROGRESS);
 378         ASSERTCMP(object->n_in_progress, >, 0);
 379         ASSERTIFCMP(test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags),
 380                     object->n_exclusive, >, 0);
 381         ASSERTIFCMP(test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags),
 382                     object->n_in_progress, ==, 1);
 383
 384         spin_lock(&object->lock);
 385
 386         op->state = cancelled ?
 387                 FSCACHE_OP_ST_CANCELLED : FSCACHE_OP_ST_COMPLETE;
 388
 389         if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags))
 390                 object->n_exclusive--;
 391         object->n_in_progress--;
 392         if (object->n_in_progress == 0)
 393                 fscache_start_operations(object);
 394
 395         spin_unlock(&object->lock);
 396         _leave("");
 397 }
 398 EXPORT_SYMBOL(fscache_op_complete);
 399
 400 /*
 401  * release an operation
 402  * - queues pending ops if this is the last in-progress op
 403  */
 404 void fscache_put_operation(struct fscache_operation *op)
 405 {
 406         struct fscache_object *object;
 407         struct fscache_cache *cache;
 408
 409         _enter("{OBJ%x OP%x,%d}",
 410                op->object->debug_id, op->debug_id, atomic_read(&op->usage));
 411
 412         ASSERTCMP(atomic_read(&op->usage), >, 0);
 413
 414         if (!atomic_dec_and_test(&op->usage))
 415                 return;
 416
 417         _debug("PUT OP");
 418         ASSERTIFCMP(op->state != FSCACHE_OP_ST_COMPLETE,
 419                     op->state, ==, FSCACHE_OP_ST_CANCELLED);
 420         op->state = FSCACHE_OP_ST_DEAD;
 421
 422         fscache_stat(&fscache_n_op_release);
 423
 424         if (op->release) {
 425                 op->release(op);
 426                 op->release = NULL;
 427         }
 428
 429         object = op->object;
 430
 431         if (test_bit(FSCACHE_OP_DEC_READ_CNT, &op->flags)) {
 432                 if (atomic_dec_and_test(&object->n_reads)) {
 433                         clear_bit(FSCACHE_COOKIE_WAITING_ON_READS,
 434                                   &object->cookie->flags);
 435                         wake_up_bit(&object->cookie->flags,
 436                                     FSCACHE_COOKIE_WAITING_ON_READS);
 437                 }
 438         }
 439
 440         /* now... we may get called with the object spinlock held, so we
 441          * complete the cleanup here only if we can immediately acquire the
 442          * lock, and defer it otherwise */
 443         if (!spin_trylock(&object->lock)) {
 444                 _debug("defer put");
 445                 fscache_stat(&fscache_n_op_deferred_release);
 446
 447                 cache = object->cache;
 448                 spin_lock(&cache->op_gc_list_lock);
 449                 list_add_tail(&op->pend_link, &cache->op_gc_list);
 450                 spin_unlock(&cache->op_gc_list_lock);
 451                 schedule_work(&cache->op_gc);
 452                 _leave(" [defer]");
 453                 return;
 454         }
 455
 456         ASSERTCMP(object->n_ops, >, 0);
 457         object->n_ops--;
 458         if (object->n_ops == 0)
 459                 fscache_raise_event(object, FSCACHE_OBJECT_EV_CLEARED);
 460
 461         spin_unlock(&object->lock);
 462
 463         kfree(op);
 464         _leave(" [done]");
 465 }
 466 EXPORT_SYMBOL(fscache_put_operation);
 467
 468 /*
 469  * garbage collect operations that have had their release deferred
 470  */
 471 void fscache_operation_gc(struct work_struct *work)
 472 {
 473         struct fscache_operation *op;
 474         struct fscache_object *object;
 475         struct fscache_cache *cache =
 476                 container_of(work, struct fscache_cache, op_gc);
 477         int count = 0;
 478
 479         _enter("");
 480
 481         do {
 482                 spin_lock(&cache->op_gc_list_lock);
 483                 if (list_empty(&cache->op_gc_list)) {
 484                         spin_unlock(&cache->op_gc_list_lock);
 485                         break;
 486                 }
 487
 488                 op = list_entry(cache->op_gc_list.next,
 489                                 struct fscache_operation, pend_link);
 490                 list_del(&op->pend_link);
 491                 spin_unlock(&cache->op_gc_list_lock);
 492
 493                 object = op->object;
 494                 spin_lock(&object->lock);
 495
 496                 _debug("GC DEFERRED REL OBJ%x OP%x",
 497                        object->debug_id, op->debug_id);
 498                 fscache_stat(&fscache_n_op_gc);
 499
 500                 ASSERTCMP(atomic_read(&op->usage), ==, 0);
 501                 ASSERTCMP(op->state, ==, FSCACHE_OP_ST_DEAD);
 502
 503                 ASSERTCMP(object->n_ops, >, 0);
 504                 object->n_ops--;
 505                 if (object->n_ops == 0)
 506                         fscache_raise_event(object, FSCACHE_OBJECT_EV_CLEARED);
 507
 508                 spin_unlock(&object->lock);
 509                 kfree(op);
 510
 511         } while (count++ < 20);
 512
 513         if (!list_empty(&cache->op_gc_list))
 514                 schedule_work(&cache->op_gc);
 515
 516         _leave("");
 517 }
 518
 519 /*
 520  * execute an operation using fs_op_wq to provide processing context -
 521  * the caller holds a ref to this object, so we don't need to hold one
 522  */
 523 void fscache_op_work_func(struct work_struct *work)
 524 {
 525         struct fscache_operation *op =
 526                 container_of(work, struct fscache_operation, work);
 527         unsigned long start;
 528
 529         _enter("{OBJ%x OP%x,%d}",
 530                op->object->debug_id, op->debug_id, atomic_read(&op->usage));
 531
 532         ASSERT(op->processor != NULL);
 533         start = jiffies;
 534         op->processor(op);
 535         fscache_hist(fscache_ops_histogram, start);
 536         fscache_put_operation(op);
 537
 538         _leave("");
 539 }