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