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 "internal.h"
  18
  19 atomic_t fscache_op_debug_id;
  20 EXPORT_SYMBOL(fscache_op_debug_id);
  21
  22 /**
  23  * fscache_enqueue_operation - Enqueue an operation for processing
  24  * @op: The operation to enqueue
  25  *
  26  * Enqueue an operation for processing by the FS-Cache thread pool.
  27  *
  28  * This will get its own ref on the object.
  29  */
  30 void fscache_enqueue_operation(struct fscache_operation *op)
  31 {
  32         _enter("{OBJ%x OP%x,%u}",
  33                op->object->debug_id, op->debug_id, atomic_read(&op->usage));
  34
  35         fscache_set_op_state(op, "EnQ");
  36
  37         ASSERT(op->processor != NULL);
  38         ASSERTCMP(op->object->state, >=, FSCACHE_OBJECT_AVAILABLE);
  39         ASSERTCMP(atomic_read(&op->usage), >, 0);
  40
  41         if (list_empty(&op->pend_link)) {
  42                 switch (op->flags & FSCACHE_OP_TYPE) {
  43                 case FSCACHE_OP_FAST:
  44                         _debug("queue fast");
  45                         atomic_inc(&op->usage);
  46                         if (!schedule_work(&op->fast_work))
  47                                 fscache_put_operation(op);
  48                         break;
  49                 case FSCACHE_OP_SLOW:
  50                         _debug("queue slow");
  51                         slow_work_enqueue(&op->slow_work);
  52                         break;
  53                 case FSCACHE_OP_MYTHREAD:
  54                         _debug("queue for caller's attention");
  55                         break;
  56                 default:
  57                         printk(KERN_ERR "FS-Cache: Unexpected op type %lx",
  58                                op->flags);
  59                         BUG();
  60                         break;
  61                 }
  62                 fscache_stat(&fscache_n_op_enqueue);
  63         }
  64 }
  65 EXPORT_SYMBOL(fscache_enqueue_operation);
  66
  67 /*
  68  * start an op running
  69  */
  70 static void fscache_run_op(struct fscache_object *object,
  71                            struct fscache_operation *op)
  72 {
  73         fscache_set_op_state(op, "Run");
  74
  75         object->n_in_progress++;
  76         if (test_and_clear_bit(FSCACHE_OP_WAITING, &op->flags))
  77                 wake_up_bit(&op->flags, FSCACHE_OP_WAITING);
  78         if (op->processor)
  79                 fscache_enqueue_operation(op);
  80         fscache_stat(&fscache_n_op_run);
  81 }
  82
  83 /*
  84  * submit an exclusive operation for an object
  85  * - other ops are excluded from running simultaneously with this one
  86  * - this gets any extra refs it needs on an op
  87  */
  88 int fscache_submit_exclusive_op(struct fscache_object *object,
  89                                 struct fscache_operation *op)
  90 {
  91         int ret;
  92
  93         _enter("{OBJ%x OP%x},", object->debug_id, op->debug_id);
  94
  95         fscache_set_op_state(op, "SubmitX");
  96
  97         spin_lock(&object->lock);
  98         ASSERTCMP(object->n_ops, >=, object->n_in_progress);
  99         ASSERTCMP(object->n_ops, >=, object->n_exclusive);
 100
 101         ret = -ENOBUFS;
 102         if (fscache_object_is_active(object)) {
 103                 op->object = object;
 104                 object->n_ops++;
 105                 object->n_exclusive++;  /* reads and writes must wait */
 106
 107                 if (object->n_ops > 0) {
 108                         atomic_inc(&op->usage);
 109                         list_add_tail(&op->pend_link, &object->pending_ops);
 110                         fscache_stat(&fscache_n_op_pend);
 111                 } else if (!list_empty(&object->pending_ops)) {
 112                         atomic_inc(&op->usage);
 113                         list_add_tail(&op->pend_link, &object->pending_ops);
 114                         fscache_stat(&fscache_n_op_pend);
 115                         fscache_start_operations(object);
 116                 } else {
 117                         ASSERTCMP(object->n_in_progress, ==, 0);
 118                         fscache_run_op(object, op);
 119                 }
 120
 121                 /* need to issue a new write op after this */
 122                 clear_bit(FSCACHE_OBJECT_PENDING_WRITE, &object->flags);
 123                 ret = 0;
 124         } else if (object->state == FSCACHE_OBJECT_CREATING) {
 125                 op->object = object;
 126                 object->n_ops++;
 127                 object->n_exclusive++;  /* reads and writes must wait */
 128                 atomic_inc(&op->usage);
 129                 list_add_tail(&op->pend_link, &object->pending_ops);
 130                 fscache_stat(&fscache_n_op_pend);
 131                 ret = 0;
 132         } else {
 133                 /* not allowed to submit ops in any other state */
 134                 BUG();
 135         }
 136
 137         spin_unlock(&object->lock);
 138         return ret;
 139 }
 140
 141 /*
 142  * report an unexpected submission
 143  */
 144 static void fscache_report_unexpected_submission(struct fscache_object *object,
 145                                                  struct fscache_operation *op,
 146                                                  unsigned long ostate)
 147 {
 148         static bool once_only;
 149         struct fscache_operation *p;
 150         unsigned n;
 151
 152         if (once_only)
 153                 return;
 154         once_only = true;
 155
 156         kdebug("unexpected submission OP%x [OBJ%x %s]",
 157                op->debug_id, object->debug_id,
 158                fscache_object_states[object->state]);
 159         kdebug("objstate=%s [%s]",
 160                fscache_object_states[object->state],
 161                fscache_object_states[ostate]);
 162         kdebug("objflags=%lx", object->flags);
 163         kdebug("objevent=%lx [%lx]", object->events, object->event_mask);
 164         kdebug("ops=%u inp=%u exc=%u",
 165                object->n_ops, object->n_in_progress, object->n_exclusive);
 166
 167         if (!list_empty(&object->pending_ops)) {
 168                 n = 0;
 169                 list_for_each_entry(p, &object->pending_ops, pend_link) {
 170                         ASSERTCMP(p->object, ==, object);
 171                         kdebug("%p %p", op->processor, op->release);
 172                         n++;
 173                 }
 174
 175                 kdebug("n=%u", n);
 176         }
 177
 178         dump_stack();
 179 }
 180
 181 /*
 182  * submit an operation for an object
 183  * - objects may be submitted only in the following states:
 184  *   - during object creation (write ops may be submitted)
 185  *   - whilst the object is active
 186  *   - after an I/O error incurred in one of the two above states (op rejected)
 187  * - this gets any extra refs it needs on an op
 188  */
 189 int fscache_submit_op(struct fscache_object *object,
 190                       struct fscache_operation *op)
 191 {
 192         unsigned long ostate;
 193         int ret;
 194
 195         _enter("{OBJ%x OP%x},{%u}",
 196                object->debug_id, op->debug_id, atomic_read(&op->usage));
 197
 198         ASSERTCMP(atomic_read(&op->usage), >, 0);
 199
 200         fscache_set_op_state(op, "Submit");
 201
 202         spin_lock(&object->lock);
 203         ASSERTCMP(object->n_ops, >=, object->n_in_progress);
 204         ASSERTCMP(object->n_ops, >=, object->n_exclusive);
 205
 206         ostate = object->state;
 207         smp_rmb();
 208
 209         if (fscache_object_is_active(object)) {
 210                 op->object = object;
 211                 object->n_ops++;
 212
 213                 if (object->n_exclusive > 0) {
 214                         atomic_inc(&op->usage);
 215                         list_add_tail(&op->pend_link, &object->pending_ops);
 216                         fscache_stat(&fscache_n_op_pend);
 217                 } else if (!list_empty(&object->pending_ops)) {
 218                         atomic_inc(&op->usage);
 219                         list_add_tail(&op->pend_link, &object->pending_ops);
 220                         fscache_stat(&fscache_n_op_pend);
 221                         fscache_start_operations(object);
 222                 } else {
 223                         ASSERTCMP(object->n_exclusive, ==, 0);
 224                         fscache_run_op(object, op);
 225                 }
 226                 ret = 0;
 227         } else if (object->state == FSCACHE_OBJECT_CREATING) {
 228                 op->object = object;
 229                 object->n_ops++;
 230                 atomic_inc(&op->usage);
 231                 list_add_tail(&op->pend_link, &object->pending_ops);
 232                 fscache_stat(&fscache_n_op_pend);
 233                 ret = 0;
 234         } else if (!test_bit(FSCACHE_IOERROR, &object->cache->flags)) {
 235                 fscache_report_unexpected_submission(object, op, ostate);
 236                 ASSERT(!fscache_object_is_active(object));
 237                 ret = -ENOBUFS;
 238         } else {
 239                 ret = -ENOBUFS;
 240         }
 241
 242         spin_unlock(&object->lock);
 243         return ret;
 244 }
 245
 246 /*
 247  * queue an object for withdrawal on error, aborting all following asynchronous
 248  * operations
 249  */
 250 void fscache_abort_object(struct fscache_object *object)
 251 {
 252         _enter("{OBJ%x}", object->debug_id);
 253
 254         fscache_raise_event(object, FSCACHE_OBJECT_EV_ERROR);
 255 }
 256
 257 /*
 258  * jump start the operation processing on an object
 259  * - caller must hold object->lock
 260  */
 261 void fscache_start_operations(struct fscache_object *object)
 262 {
 263         struct fscache_operation *op;
 264         bool stop = false;
 265
 266         while (!list_empty(&object->pending_ops) && !stop) {
 267                 op = list_entry(object->pending_ops.next,
 268                                 struct fscache_operation, pend_link);
 269
 270                 if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags)) {
 271                         if (object->n_in_progress > 0)
 272                                 break;
 273                         stop = true;
 274                 }
 275                 list_del_init(&op->pend_link);
 276                 object->n_in_progress++;
 277
 278                 if (test_and_clear_bit(FSCACHE_OP_WAITING, &op->flags))
 279                         wake_up_bit(&op->flags, FSCACHE_OP_WAITING);
 280                 if (op->processor)
 281                         fscache_enqueue_operation(op);
 282
 283                 /* the pending queue was holding a ref on the object */
 284                 fscache_put_operation(op);
 285         }
 286
 287         ASSERTCMP(object->n_in_progress, <=, object->n_ops);
 288
 289         _debug("woke %d ops on OBJ%x",
 290                object->n_in_progress, object->debug_id);
 291 }
 292
 293 /*
 294  * release an operation
 295  * - queues pending ops if this is the last in-progress op
 296  */
 297 void fscache_put_operation(struct fscache_operation *op)
 298 {
 299         struct fscache_object *object;
 300         struct fscache_cache *cache;
 301
 302         _enter("{OBJ%x OP%x,%d}",
 303                op->object->debug_id, op->debug_id, atomic_read(&op->usage));
 304
 305         ASSERTCMP(atomic_read(&op->usage), >, 0);
 306
 307         if (!atomic_dec_and_test(&op->usage))
 308                 return;
 309
 310         fscache_set_op_state(op, "Put");
 311
 312         _debug("PUT OP");
 313         if (test_and_set_bit(FSCACHE_OP_DEAD, &op->flags))
 314                 BUG();
 315
 316         fscache_stat(&fscache_n_op_release);
 317
 318         if (op->release) {
 319                 op->release(op);
 320                 op->release = NULL;
 321         }
 322
 323         object = op->object;
 324
 325         /* now... we may get called with the object spinlock held, so we
 326          * complete the cleanup here only if we can immediately acquire the
 327          * lock, and defer it otherwise */
 328         if (!spin_trylock(&object->lock)) {
 329                 _debug("defer put");
 330                 fscache_stat(&fscache_n_op_deferred_release);
 331
 332                 cache = object->cache;
 333                 spin_lock(&cache->op_gc_list_lock);
 334                 list_add_tail(&op->pend_link, &cache->op_gc_list);
 335                 spin_unlock(&cache->op_gc_list_lock);
 336                 schedule_work(&cache->op_gc);
 337                 _leave(" [defer]");
 338                 return;
 339         }
 340
 341         if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags)) {
 342                 ASSERTCMP(object->n_exclusive, >, 0);
 343                 object->n_exclusive--;
 344         }
 345
 346         ASSERTCMP(object->n_in_progress, >, 0);
 347         object->n_in_progress--;
 348         if (object->n_in_progress == 0)
 349                 fscache_start_operations(object);
 350
 351         ASSERTCMP(object->n_ops, >, 0);
 352         object->n_ops--;
 353         if (object->n_ops == 0)
 354                 fscache_raise_event(object, FSCACHE_OBJECT_EV_CLEARED);
 355
 356         spin_unlock(&object->lock);
 357
 358         kfree(op);
 359         _leave(" [done]");
 360 }
 361 EXPORT_SYMBOL(fscache_put_operation);
 362
 363 /*
 364  * garbage collect operations that have had their release deferred
 365  */
 366 void fscache_operation_gc(struct work_struct *work)
 367 {
 368         struct fscache_operation *op;
 369         struct fscache_object *object;
 370         struct fscache_cache *cache =
 371                 container_of(work, struct fscache_cache, op_gc);
 372         int count = 0;
 373
 374         _enter("");
 375
 376         do {
 377                 spin_lock(&cache->op_gc_list_lock);
 378                 if (list_empty(&cache->op_gc_list)) {
 379                         spin_unlock(&cache->op_gc_list_lock);
 380                         break;
 381                 }
 382
 383                 op = list_entry(cache->op_gc_list.next,
 384                                 struct fscache_operation, pend_link);
 385                 list_del(&op->pend_link);
 386                 spin_unlock(&cache->op_gc_list_lock);
 387
 388                 object = op->object;
 389
 390                 _debug("GC DEFERRED REL OBJ%x OP%x",
 391                        object->debug_id, op->debug_id);
 392                 fscache_stat(&fscache_n_op_gc);
 393
 394                 ASSERTCMP(atomic_read(&op->usage), ==, 0);
 395
 396                 spin_lock(&object->lock);
 397                 if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags)) {
 398                         ASSERTCMP(object->n_exclusive, >, 0);
 399                         object->n_exclusive--;
 400                 }
 401
 402                 ASSERTCMP(object->n_in_progress, >, 0);
 403                 object->n_in_progress--;
 404                 if (object->n_in_progress == 0)
 405                         fscache_start_operations(object);
 406
 407                 ASSERTCMP(object->n_ops, >, 0);
 408                 object->n_ops--;
 409                 if (object->n_ops == 0)
 410                         fscache_raise_event(object, FSCACHE_OBJECT_EV_CLEARED);
 411
 412                 spin_unlock(&object->lock);
 413
 414         } while (count++ < 20);
 415
 416         if (!list_empty(&cache->op_gc_list))
 417                 schedule_work(&cache->op_gc);
 418
 419         _leave("");
 420 }
 421
 422 /*
 423  * allow the slow work item processor to get a ref on an operation
 424  */
 425 static int fscache_op_get_ref(struct slow_work *work)
 426 {
 427         struct fscache_operation *op =
 428                 container_of(work, struct fscache_operation, slow_work);
 429
 430         atomic_inc(&op->usage);
 431         return 0;
 432 }
 433
 434 /*
 435  * allow the slow work item processor to discard a ref on an operation
 436  */
 437 static void fscache_op_put_ref(struct slow_work *work)
 438 {
 439         struct fscache_operation *op =
 440                 container_of(work, struct fscache_operation, slow_work);
 441
 442         fscache_put_operation(op);
 443 }
 444
 445 /*
 446  * execute an operation using the slow thread pool to provide processing context
 447  * - the caller holds a ref to this object, so we don't need to hold one
 448  */
 449 static void fscache_op_execute(struct slow_work *work)
 450 {
 451         struct fscache_operation *op =
 452                 container_of(work, struct fscache_operation, slow_work);
 453         unsigned long start;
 454
 455         _enter("{OBJ%x OP%x,%d}",
 456                op->object->debug_id, op->debug_id, atomic_read(&op->usage));
 457
 458         ASSERT(op->processor != NULL);
 459         start = jiffies;
 460         op->processor(op);
 461         fscache_hist(fscache_ops_histogram, start);
 462
 463         _leave("");
 464 }
 465
 466 /*
 467  * describe an operation for slow-work debugging
 468  */
 469 #ifdef CONFIG_SLOW_WORK_PROC
 470 static void fscache_op_desc(struct slow_work *work, struct seq_file *m)
 471 {
 472         struct fscache_operation *op =
 473                 container_of(work, struct fscache_operation, slow_work);
 474
 475         seq_printf(m, "FSC: OBJ%x OP%x: %s/%s fl=%lx",
 476                    op->object->debug_id, op->debug_id,
 477                    op->name, op->state, op->flags);
 478 }
 479 #endif
 480
 481 const struct slow_work_ops fscache_op_slow_work_ops = {
 482         .owner          = THIS_MODULE,
 483         .get_ref        = fscache_op_get_ref,
 484         .put_ref        = fscache_op_put_ref,
 485         .execute        = fscache_op_execute,
 486 #ifdef CONFIG_SLOW_WORK_PROC
 487         .desc           = fscache_op_desc,
 488 #endif
 489 };