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