Merge branch 'for-linus' of git://git.kernel.org/pub/scm/linux/kernel/git/sage/ceph...
[linux-2.6.git] / fs / ceph / caps.c
1 #include "ceph_debug.h"
2
3 #include <linux/fs.h>
4 #include <linux/kernel.h>
5 #include <linux/sched.h>
6 #include <linux/vmalloc.h>
7 #include <linux/wait.h>
8 #include <linux/writeback.h>
9
10 #include "super.h"
11 #include "decode.h"
12 #include "messenger.h"
13
14 /*
15  * Capability management
16  *
17  * The Ceph metadata servers control client access to inode metadata
18  * and file data by issuing capabilities, granting clients permission
19  * to read and/or write both inode field and file data to OSDs
20  * (storage nodes).  Each capability consists of a set of bits
21  * indicating which operations are allowed.
22  *
23  * If the client holds a *_SHARED cap, the client has a coherent value
24  * that can be safely read from the cached inode.
25  *
26  * In the case of a *_EXCL (exclusive) or FILE_WR capabilities, the
27  * client is allowed to change inode attributes (e.g., file size,
28  * mtime), note its dirty state in the ceph_cap, and asynchronously
29  * flush that metadata change to the MDS.
30  *
31  * In the event of a conflicting operation (perhaps by another
32  * client), the MDS will revoke the conflicting client capabilities.
33  *
34  * In order for a client to cache an inode, it must hold a capability
35  * with at least one MDS server.  When inodes are released, release
36  * notifications are batched and periodically sent en masse to the MDS
37  * cluster to release server state.
38  */
39
40
41 /*
42  * Generate readable cap strings for debugging output.
43  */
44 #define MAX_CAP_STR 20
45 static char cap_str[MAX_CAP_STR][40];
46 static DEFINE_SPINLOCK(cap_str_lock);
47 static int last_cap_str;
48
49 static char *gcap_string(char *s, int c)
50 {
51         if (c & CEPH_CAP_GSHARED)
52                 *s++ = 's';
53         if (c & CEPH_CAP_GEXCL)
54                 *s++ = 'x';
55         if (c & CEPH_CAP_GCACHE)
56                 *s++ = 'c';
57         if (c & CEPH_CAP_GRD)
58                 *s++ = 'r';
59         if (c & CEPH_CAP_GWR)
60                 *s++ = 'w';
61         if (c & CEPH_CAP_GBUFFER)
62                 *s++ = 'b';
63         if (c & CEPH_CAP_GLAZYIO)
64                 *s++ = 'l';
65         return s;
66 }
67
68 const char *ceph_cap_string(int caps)
69 {
70         int i;
71         char *s;
72         int c;
73
74         spin_lock(&cap_str_lock);
75         i = last_cap_str++;
76         if (last_cap_str == MAX_CAP_STR)
77                 last_cap_str = 0;
78         spin_unlock(&cap_str_lock);
79
80         s = cap_str[i];
81
82         if (caps & CEPH_CAP_PIN)
83                 *s++ = 'p';
84
85         c = (caps >> CEPH_CAP_SAUTH) & 3;
86         if (c) {
87                 *s++ = 'A';
88                 s = gcap_string(s, c);
89         }
90
91         c = (caps >> CEPH_CAP_SLINK) & 3;
92         if (c) {
93                 *s++ = 'L';
94                 s = gcap_string(s, c);
95         }
96
97         c = (caps >> CEPH_CAP_SXATTR) & 3;
98         if (c) {
99                 *s++ = 'X';
100                 s = gcap_string(s, c);
101         }
102
103         c = caps >> CEPH_CAP_SFILE;
104         if (c) {
105                 *s++ = 'F';
106                 s = gcap_string(s, c);
107         }
108
109         if (s == cap_str[i])
110                 *s++ = '-';
111         *s = 0;
112         return cap_str[i];
113 }
114
115 /*
116  * Cap reservations
117  *
118  * Maintain a global pool of preallocated struct ceph_caps, referenced
119  * by struct ceph_caps_reservations.  This ensures that we preallocate
120  * memory needed to successfully process an MDS response.  (If an MDS
121  * sends us cap information and we fail to process it, we will have
122  * problems due to the client and MDS being out of sync.)
123  *
124  * Reservations are 'owned' by a ceph_cap_reservation context.
125  */
126 static spinlock_t caps_list_lock;
127 static struct list_head caps_list;  /* unused (reserved or unreserved) */
128 static int caps_total_count;        /* total caps allocated */
129 static int caps_use_count;          /* in use */
130 static int caps_reserve_count;      /* unused, reserved */
131 static int caps_avail_count;        /* unused, unreserved */
132 static int caps_min_count;          /* keep at least this many (unreserved) */
133
134 void __init ceph_caps_init(void)
135 {
136         INIT_LIST_HEAD(&caps_list);
137         spin_lock_init(&caps_list_lock);
138 }
139
140 void ceph_caps_finalize(void)
141 {
142         struct ceph_cap *cap;
143
144         spin_lock(&caps_list_lock);
145         while (!list_empty(&caps_list)) {
146                 cap = list_first_entry(&caps_list, struct ceph_cap, caps_item);
147                 list_del(&cap->caps_item);
148                 kmem_cache_free(ceph_cap_cachep, cap);
149         }
150         caps_total_count = 0;
151         caps_avail_count = 0;
152         caps_use_count = 0;
153         caps_reserve_count = 0;
154         caps_min_count = 0;
155         spin_unlock(&caps_list_lock);
156 }
157
158 void ceph_adjust_min_caps(int delta)
159 {
160         spin_lock(&caps_list_lock);
161         caps_min_count += delta;
162         BUG_ON(caps_min_count < 0);
163         spin_unlock(&caps_list_lock);
164 }
165
166 int ceph_reserve_caps(struct ceph_cap_reservation *ctx, int need)
167 {
168         int i;
169         struct ceph_cap *cap;
170         int have;
171         int alloc = 0;
172         LIST_HEAD(newcaps);
173         int ret = 0;
174
175         dout("reserve caps ctx=%p need=%d\n", ctx, need);
176
177         /* first reserve any caps that are already allocated */
178         spin_lock(&caps_list_lock);
179         if (caps_avail_count >= need)
180                 have = need;
181         else
182                 have = caps_avail_count;
183         caps_avail_count -= have;
184         caps_reserve_count += have;
185         BUG_ON(caps_total_count != caps_use_count + caps_reserve_count +
186                caps_avail_count);
187         spin_unlock(&caps_list_lock);
188
189         for (i = have; i < need; i++) {
190                 cap = kmem_cache_alloc(ceph_cap_cachep, GFP_NOFS);
191                 if (!cap) {
192                         ret = -ENOMEM;
193                         goto out_alloc_count;
194                 }
195                 list_add(&cap->caps_item, &newcaps);
196                 alloc++;
197         }
198         BUG_ON(have + alloc != need);
199
200         spin_lock(&caps_list_lock);
201         caps_total_count += alloc;
202         caps_reserve_count += alloc;
203         list_splice(&newcaps, &caps_list);
204
205         BUG_ON(caps_total_count != caps_use_count + caps_reserve_count +
206                caps_avail_count);
207         spin_unlock(&caps_list_lock);
208
209         ctx->count = need;
210         dout("reserve caps ctx=%p %d = %d used + %d resv + %d avail\n",
211              ctx, caps_total_count, caps_use_count, caps_reserve_count,
212              caps_avail_count);
213         return 0;
214
215 out_alloc_count:
216         /* we didn't manage to reserve as much as we needed */
217         pr_warning("reserve caps ctx=%p ENOMEM need=%d got=%d\n",
218                    ctx, need, have);
219         return ret;
220 }
221
222 int ceph_unreserve_caps(struct ceph_cap_reservation *ctx)
223 {
224         dout("unreserve caps ctx=%p count=%d\n", ctx, ctx->count);
225         if (ctx->count) {
226                 spin_lock(&caps_list_lock);
227                 BUG_ON(caps_reserve_count < ctx->count);
228                 caps_reserve_count -= ctx->count;
229                 caps_avail_count += ctx->count;
230                 ctx->count = 0;
231                 dout("unreserve caps %d = %d used + %d resv + %d avail\n",
232                      caps_total_count, caps_use_count, caps_reserve_count,
233                      caps_avail_count);
234                 BUG_ON(caps_total_count != caps_use_count + caps_reserve_count +
235                        caps_avail_count);
236                 spin_unlock(&caps_list_lock);
237         }
238         return 0;
239 }
240
241 static struct ceph_cap *get_cap(struct ceph_cap_reservation *ctx)
242 {
243         struct ceph_cap *cap = NULL;
244
245         /* temporary, until we do something about cap import/export */
246         if (!ctx)
247                 return kmem_cache_alloc(ceph_cap_cachep, GFP_NOFS);
248
249         spin_lock(&caps_list_lock);
250         dout("get_cap ctx=%p (%d) %d = %d used + %d resv + %d avail\n",
251              ctx, ctx->count, caps_total_count, caps_use_count,
252              caps_reserve_count, caps_avail_count);
253         BUG_ON(!ctx->count);
254         BUG_ON(ctx->count > caps_reserve_count);
255         BUG_ON(list_empty(&caps_list));
256
257         ctx->count--;
258         caps_reserve_count--;
259         caps_use_count++;
260
261         cap = list_first_entry(&caps_list, struct ceph_cap, caps_item);
262         list_del(&cap->caps_item);
263
264         BUG_ON(caps_total_count != caps_use_count + caps_reserve_count +
265                caps_avail_count);
266         spin_unlock(&caps_list_lock);
267         return cap;
268 }
269
270 void ceph_put_cap(struct ceph_cap *cap)
271 {
272         spin_lock(&caps_list_lock);
273         dout("put_cap %p %d = %d used + %d resv + %d avail\n",
274              cap, caps_total_count, caps_use_count,
275              caps_reserve_count, caps_avail_count);
276         caps_use_count--;
277         /*
278          * Keep some preallocated caps around (ceph_min_count), to
279          * avoid lots of free/alloc churn.
280          */
281         if (caps_avail_count >= caps_reserve_count + caps_min_count) {
282                 caps_total_count--;
283                 kmem_cache_free(ceph_cap_cachep, cap);
284         } else {
285                 caps_avail_count++;
286                 list_add(&cap->caps_item, &caps_list);
287         }
288
289         BUG_ON(caps_total_count != caps_use_count + caps_reserve_count +
290                caps_avail_count);
291         spin_unlock(&caps_list_lock);
292 }
293
294 void ceph_reservation_status(struct ceph_client *client,
295                              int *total, int *avail, int *used, int *reserved,
296                              int *min)
297 {
298         if (total)
299                 *total = caps_total_count;
300         if (avail)
301                 *avail = caps_avail_count;
302         if (used)
303                 *used = caps_use_count;
304         if (reserved)
305                 *reserved = caps_reserve_count;
306         if (min)
307                 *min = caps_min_count;
308 }
309
310 /*
311  * Find ceph_cap for given mds, if any.
312  *
313  * Called with i_lock held.
314  */
315 static struct ceph_cap *__get_cap_for_mds(struct ceph_inode_info *ci, int mds)
316 {
317         struct ceph_cap *cap;
318         struct rb_node *n = ci->i_caps.rb_node;
319
320         while (n) {
321                 cap = rb_entry(n, struct ceph_cap, ci_node);
322                 if (mds < cap->mds)
323                         n = n->rb_left;
324                 else if (mds > cap->mds)
325                         n = n->rb_right;
326                 else
327                         return cap;
328         }
329         return NULL;
330 }
331
332 /*
333  * Return id of any MDS with a cap, preferably FILE_WR|WRBUFFER|EXCL, else
334  * -1.
335  */
336 static int __ceph_get_cap_mds(struct ceph_inode_info *ci, u32 *mseq)
337 {
338         struct ceph_cap *cap;
339         int mds = -1;
340         struct rb_node *p;
341
342         /* prefer mds with WR|WRBUFFER|EXCL caps */
343         for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
344                 cap = rb_entry(p, struct ceph_cap, ci_node);
345                 mds = cap->mds;
346                 if (mseq)
347                         *mseq = cap->mseq;
348                 if (cap->issued & (CEPH_CAP_FILE_WR |
349                                    CEPH_CAP_FILE_BUFFER |
350                                    CEPH_CAP_FILE_EXCL))
351                         break;
352         }
353         return mds;
354 }
355
356 int ceph_get_cap_mds(struct inode *inode)
357 {
358         int mds;
359         spin_lock(&inode->i_lock);
360         mds = __ceph_get_cap_mds(ceph_inode(inode), NULL);
361         spin_unlock(&inode->i_lock);
362         return mds;
363 }
364
365 /*
366  * Called under i_lock.
367  */
368 static void __insert_cap_node(struct ceph_inode_info *ci,
369                               struct ceph_cap *new)
370 {
371         struct rb_node **p = &ci->i_caps.rb_node;
372         struct rb_node *parent = NULL;
373         struct ceph_cap *cap = NULL;
374
375         while (*p) {
376                 parent = *p;
377                 cap = rb_entry(parent, struct ceph_cap, ci_node);
378                 if (new->mds < cap->mds)
379                         p = &(*p)->rb_left;
380                 else if (new->mds > cap->mds)
381                         p = &(*p)->rb_right;
382                 else
383                         BUG();
384         }
385
386         rb_link_node(&new->ci_node, parent, p);
387         rb_insert_color(&new->ci_node, &ci->i_caps);
388 }
389
390 /*
391  * (re)set cap hold timeouts, which control the delayed release
392  * of unused caps back to the MDS.  Should be called on cap use.
393  */
394 static void __cap_set_timeouts(struct ceph_mds_client *mdsc,
395                                struct ceph_inode_info *ci)
396 {
397         struct ceph_mount_args *ma = mdsc->client->mount_args;
398
399         ci->i_hold_caps_min = round_jiffies(jiffies +
400                                             ma->caps_wanted_delay_min * HZ);
401         ci->i_hold_caps_max = round_jiffies(jiffies +
402                                             ma->caps_wanted_delay_max * HZ);
403         dout("__cap_set_timeouts %p min %lu max %lu\n", &ci->vfs_inode,
404              ci->i_hold_caps_min - jiffies, ci->i_hold_caps_max - jiffies);
405 }
406
407 /*
408  * (Re)queue cap at the end of the delayed cap release list.
409  *
410  * If I_FLUSH is set, leave the inode at the front of the list.
411  *
412  * Caller holds i_lock
413  *    -> we take mdsc->cap_delay_lock
414  */
415 static void __cap_delay_requeue(struct ceph_mds_client *mdsc,
416                                 struct ceph_inode_info *ci)
417 {
418         __cap_set_timeouts(mdsc, ci);
419         dout("__cap_delay_requeue %p flags %d at %lu\n", &ci->vfs_inode,
420              ci->i_ceph_flags, ci->i_hold_caps_max);
421         if (!mdsc->stopping) {
422                 spin_lock(&mdsc->cap_delay_lock);
423                 if (!list_empty(&ci->i_cap_delay_list)) {
424                         if (ci->i_ceph_flags & CEPH_I_FLUSH)
425                                 goto no_change;
426                         list_del_init(&ci->i_cap_delay_list);
427                 }
428                 list_add_tail(&ci->i_cap_delay_list, &mdsc->cap_delay_list);
429 no_change:
430                 spin_unlock(&mdsc->cap_delay_lock);
431         }
432 }
433
434 /*
435  * Queue an inode for immediate writeback.  Mark inode with I_FLUSH,
436  * indicating we should send a cap message to flush dirty metadata
437  * asap, and move to the front of the delayed cap list.
438  */
439 static void __cap_delay_requeue_front(struct ceph_mds_client *mdsc,
440                                       struct ceph_inode_info *ci)
441 {
442         dout("__cap_delay_requeue_front %p\n", &ci->vfs_inode);
443         spin_lock(&mdsc->cap_delay_lock);
444         ci->i_ceph_flags |= CEPH_I_FLUSH;
445         if (!list_empty(&ci->i_cap_delay_list))
446                 list_del_init(&ci->i_cap_delay_list);
447         list_add(&ci->i_cap_delay_list, &mdsc->cap_delay_list);
448         spin_unlock(&mdsc->cap_delay_lock);
449 }
450
451 /*
452  * Cancel delayed work on cap.
453  *
454  * Caller must hold i_lock.
455  */
456 static void __cap_delay_cancel(struct ceph_mds_client *mdsc,
457                                struct ceph_inode_info *ci)
458 {
459         dout("__cap_delay_cancel %p\n", &ci->vfs_inode);
460         if (list_empty(&ci->i_cap_delay_list))
461                 return;
462         spin_lock(&mdsc->cap_delay_lock);
463         list_del_init(&ci->i_cap_delay_list);
464         spin_unlock(&mdsc->cap_delay_lock);
465 }
466
467 /*
468  * Common issue checks for add_cap, handle_cap_grant.
469  */
470 static void __check_cap_issue(struct ceph_inode_info *ci, struct ceph_cap *cap,
471                               unsigned issued)
472 {
473         unsigned had = __ceph_caps_issued(ci, NULL);
474
475         /*
476          * Each time we receive FILE_CACHE anew, we increment
477          * i_rdcache_gen.
478          */
479         if ((issued & CEPH_CAP_FILE_CACHE) &&
480             (had & CEPH_CAP_FILE_CACHE) == 0)
481                 ci->i_rdcache_gen++;
482
483         /*
484          * if we are newly issued FILE_SHARED, clear I_COMPLETE; we
485          * don't know what happened to this directory while we didn't
486          * have the cap.
487          */
488         if ((issued & CEPH_CAP_FILE_SHARED) &&
489             (had & CEPH_CAP_FILE_SHARED) == 0) {
490                 ci->i_shared_gen++;
491                 if (S_ISDIR(ci->vfs_inode.i_mode)) {
492                         dout(" marking %p NOT complete\n", &ci->vfs_inode);
493                         ci->i_ceph_flags &= ~CEPH_I_COMPLETE;
494                 }
495         }
496 }
497
498 /*
499  * Add a capability under the given MDS session.
500  *
501  * Caller should hold session snap_rwsem (read) and s_mutex.
502  *
503  * @fmode is the open file mode, if we are opening a file, otherwise
504  * it is < 0.  (This is so we can atomically add the cap and add an
505  * open file reference to it.)
506  */
507 int ceph_add_cap(struct inode *inode,
508                  struct ceph_mds_session *session, u64 cap_id,
509                  int fmode, unsigned issued, unsigned wanted,
510                  unsigned seq, unsigned mseq, u64 realmino, int flags,
511                  struct ceph_cap_reservation *caps_reservation)
512 {
513         struct ceph_mds_client *mdsc = &ceph_inode_to_client(inode)->mdsc;
514         struct ceph_inode_info *ci = ceph_inode(inode);
515         struct ceph_cap *new_cap = NULL;
516         struct ceph_cap *cap;
517         int mds = session->s_mds;
518         int actual_wanted;
519
520         dout("add_cap %p mds%d cap %llx %s seq %d\n", inode,
521              session->s_mds, cap_id, ceph_cap_string(issued), seq);
522
523         /*
524          * If we are opening the file, include file mode wanted bits
525          * in wanted.
526          */
527         if (fmode >= 0)
528                 wanted |= ceph_caps_for_mode(fmode);
529
530 retry:
531         spin_lock(&inode->i_lock);
532         cap = __get_cap_for_mds(ci, mds);
533         if (!cap) {
534                 if (new_cap) {
535                         cap = new_cap;
536                         new_cap = NULL;
537                 } else {
538                         spin_unlock(&inode->i_lock);
539                         new_cap = get_cap(caps_reservation);
540                         if (new_cap == NULL)
541                                 return -ENOMEM;
542                         goto retry;
543                 }
544
545                 cap->issued = 0;
546                 cap->implemented = 0;
547                 cap->mds = mds;
548                 cap->mds_wanted = 0;
549
550                 cap->ci = ci;
551                 __insert_cap_node(ci, cap);
552
553                 /* clear out old exporting info?  (i.e. on cap import) */
554                 if (ci->i_cap_exporting_mds == mds) {
555                         ci->i_cap_exporting_issued = 0;
556                         ci->i_cap_exporting_mseq = 0;
557                         ci->i_cap_exporting_mds = -1;
558                 }
559
560                 /* add to session cap list */
561                 cap->session = session;
562                 spin_lock(&session->s_cap_lock);
563                 list_add_tail(&cap->session_caps, &session->s_caps);
564                 session->s_nr_caps++;
565                 spin_unlock(&session->s_cap_lock);
566         }
567
568         if (!ci->i_snap_realm) {
569                 /*
570                  * add this inode to the appropriate snap realm
571                  */
572                 struct ceph_snap_realm *realm = ceph_lookup_snap_realm(mdsc,
573                                                                realmino);
574                 if (realm) {
575                         ceph_get_snap_realm(mdsc, realm);
576                         spin_lock(&realm->inodes_with_caps_lock);
577                         ci->i_snap_realm = realm;
578                         list_add(&ci->i_snap_realm_item,
579                                  &realm->inodes_with_caps);
580                         spin_unlock(&realm->inodes_with_caps_lock);
581                 } else {
582                         pr_err("ceph_add_cap: couldn't find snap realm %llx\n",
583                                realmino);
584                 }
585         }
586
587         __check_cap_issue(ci, cap, issued);
588
589         /*
590          * If we are issued caps we don't want, or the mds' wanted
591          * value appears to be off, queue a check so we'll release
592          * later and/or update the mds wanted value.
593          */
594         actual_wanted = __ceph_caps_wanted(ci);
595         if ((wanted & ~actual_wanted) ||
596             (issued & ~actual_wanted & CEPH_CAP_ANY_WR)) {
597                 dout(" issued %s, mds wanted %s, actual %s, queueing\n",
598                      ceph_cap_string(issued), ceph_cap_string(wanted),
599                      ceph_cap_string(actual_wanted));
600                 __cap_delay_requeue(mdsc, ci);
601         }
602
603         if (flags & CEPH_CAP_FLAG_AUTH)
604                 ci->i_auth_cap = cap;
605         else if (ci->i_auth_cap == cap)
606                 ci->i_auth_cap = NULL;
607
608         dout("add_cap inode %p (%llx.%llx) cap %p %s now %s seq %d mds%d\n",
609              inode, ceph_vinop(inode), cap, ceph_cap_string(issued),
610              ceph_cap_string(issued|cap->issued), seq, mds);
611         cap->cap_id = cap_id;
612         cap->issued = issued;
613         cap->implemented |= issued;
614         cap->mds_wanted |= wanted;
615         cap->seq = seq;
616         cap->issue_seq = seq;
617         cap->mseq = mseq;
618         cap->cap_gen = session->s_cap_gen;
619
620         if (fmode >= 0)
621                 __ceph_get_fmode(ci, fmode);
622         spin_unlock(&inode->i_lock);
623         wake_up(&ci->i_cap_wq);
624         return 0;
625 }
626
627 /*
628  * Return true if cap has not timed out and belongs to the current
629  * generation of the MDS session (i.e. has not gone 'stale' due to
630  * us losing touch with the mds).
631  */
632 static int __cap_is_valid(struct ceph_cap *cap)
633 {
634         unsigned long ttl;
635         u32 gen;
636
637         spin_lock(&cap->session->s_cap_lock);
638         gen = cap->session->s_cap_gen;
639         ttl = cap->session->s_cap_ttl;
640         spin_unlock(&cap->session->s_cap_lock);
641
642         if (cap->cap_gen < gen || time_after_eq(jiffies, ttl)) {
643                 dout("__cap_is_valid %p cap %p issued %s "
644                      "but STALE (gen %u vs %u)\n", &cap->ci->vfs_inode,
645                      cap, ceph_cap_string(cap->issued), cap->cap_gen, gen);
646                 return 0;
647         }
648
649         return 1;
650 }
651
652 /*
653  * Return set of valid cap bits issued to us.  Note that caps time
654  * out, and may be invalidated in bulk if the client session times out
655  * and session->s_cap_gen is bumped.
656  */
657 int __ceph_caps_issued(struct ceph_inode_info *ci, int *implemented)
658 {
659         int have = ci->i_snap_caps | ci->i_cap_exporting_issued;
660         struct ceph_cap *cap;
661         struct rb_node *p;
662
663         if (implemented)
664                 *implemented = 0;
665         for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
666                 cap = rb_entry(p, struct ceph_cap, ci_node);
667                 if (!__cap_is_valid(cap))
668                         continue;
669                 dout("__ceph_caps_issued %p cap %p issued %s\n",
670                      &ci->vfs_inode, cap, ceph_cap_string(cap->issued));
671                 have |= cap->issued;
672                 if (implemented)
673                         *implemented |= cap->implemented;
674         }
675         return have;
676 }
677
678 /*
679  * Get cap bits issued by caps other than @ocap
680  */
681 int __ceph_caps_issued_other(struct ceph_inode_info *ci, struct ceph_cap *ocap)
682 {
683         int have = ci->i_snap_caps;
684         struct ceph_cap *cap;
685         struct rb_node *p;
686
687         for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
688                 cap = rb_entry(p, struct ceph_cap, ci_node);
689                 if (cap == ocap)
690                         continue;
691                 if (!__cap_is_valid(cap))
692                         continue;
693                 have |= cap->issued;
694         }
695         return have;
696 }
697
698 /*
699  * Move a cap to the end of the LRU (oldest caps at list head, newest
700  * at list tail).
701  */
702 static void __touch_cap(struct ceph_cap *cap)
703 {
704         struct ceph_mds_session *s = cap->session;
705
706         spin_lock(&s->s_cap_lock);
707         if (s->s_cap_iterator == NULL) {
708                 dout("__touch_cap %p cap %p mds%d\n", &cap->ci->vfs_inode, cap,
709                      s->s_mds);
710                 list_move_tail(&cap->session_caps, &s->s_caps);
711         } else {
712                 dout("__touch_cap %p cap %p mds%d NOP, iterating over caps\n",
713                      &cap->ci->vfs_inode, cap, s->s_mds);
714         }
715         spin_unlock(&s->s_cap_lock);
716 }
717
718 /*
719  * Check if we hold the given mask.  If so, move the cap(s) to the
720  * front of their respective LRUs.  (This is the preferred way for
721  * callers to check for caps they want.)
722  */
723 int __ceph_caps_issued_mask(struct ceph_inode_info *ci, int mask, int touch)
724 {
725         struct ceph_cap *cap;
726         struct rb_node *p;
727         int have = ci->i_snap_caps;
728
729         if ((have & mask) == mask) {
730                 dout("__ceph_caps_issued_mask %p snap issued %s"
731                      " (mask %s)\n", &ci->vfs_inode,
732                      ceph_cap_string(have),
733                      ceph_cap_string(mask));
734                 return 1;
735         }
736
737         for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
738                 cap = rb_entry(p, struct ceph_cap, ci_node);
739                 if (!__cap_is_valid(cap))
740                         continue;
741                 if ((cap->issued & mask) == mask) {
742                         dout("__ceph_caps_issued_mask %p cap %p issued %s"
743                              " (mask %s)\n", &ci->vfs_inode, cap,
744                              ceph_cap_string(cap->issued),
745                              ceph_cap_string(mask));
746                         if (touch)
747                                 __touch_cap(cap);
748                         return 1;
749                 }
750
751                 /* does a combination of caps satisfy mask? */
752                 have |= cap->issued;
753                 if ((have & mask) == mask) {
754                         dout("__ceph_caps_issued_mask %p combo issued %s"
755                              " (mask %s)\n", &ci->vfs_inode,
756                              ceph_cap_string(cap->issued),
757                              ceph_cap_string(mask));
758                         if (touch) {
759                                 struct rb_node *q;
760
761                                 /* touch this + preceeding caps */
762                                 __touch_cap(cap);
763                                 for (q = rb_first(&ci->i_caps); q != p;
764                                      q = rb_next(q)) {
765                                         cap = rb_entry(q, struct ceph_cap,
766                                                        ci_node);
767                                         if (!__cap_is_valid(cap))
768                                                 continue;
769                                         __touch_cap(cap);
770                                 }
771                         }
772                         return 1;
773                 }
774         }
775
776         return 0;
777 }
778
779 /*
780  * Return true if mask caps are currently being revoked by an MDS.
781  */
782 int ceph_caps_revoking(struct ceph_inode_info *ci, int mask)
783 {
784         struct inode *inode = &ci->vfs_inode;
785         struct ceph_cap *cap;
786         struct rb_node *p;
787         int ret = 0;
788
789         spin_lock(&inode->i_lock);
790         for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
791                 cap = rb_entry(p, struct ceph_cap, ci_node);
792                 if (__cap_is_valid(cap) &&
793                     (cap->implemented & ~cap->issued & mask)) {
794                         ret = 1;
795                         break;
796                 }
797         }
798         spin_unlock(&inode->i_lock);
799         dout("ceph_caps_revoking %p %s = %d\n", inode,
800              ceph_cap_string(mask), ret);
801         return ret;
802 }
803
804 int __ceph_caps_used(struct ceph_inode_info *ci)
805 {
806         int used = 0;
807         if (ci->i_pin_ref)
808                 used |= CEPH_CAP_PIN;
809         if (ci->i_rd_ref)
810                 used |= CEPH_CAP_FILE_RD;
811         if (ci->i_rdcache_ref || ci->i_rdcache_gen)
812                 used |= CEPH_CAP_FILE_CACHE;
813         if (ci->i_wr_ref)
814                 used |= CEPH_CAP_FILE_WR;
815         if (ci->i_wrbuffer_ref)
816                 used |= CEPH_CAP_FILE_BUFFER;
817         return used;
818 }
819
820 /*
821  * wanted, by virtue of open file modes
822  */
823 int __ceph_caps_file_wanted(struct ceph_inode_info *ci)
824 {
825         int want = 0;
826         int mode;
827         for (mode = 0; mode < 4; mode++)
828                 if (ci->i_nr_by_mode[mode])
829                         want |= ceph_caps_for_mode(mode);
830         return want;
831 }
832
833 /*
834  * Return caps we have registered with the MDS(s) as 'wanted'.
835  */
836 int __ceph_caps_mds_wanted(struct ceph_inode_info *ci)
837 {
838         struct ceph_cap *cap;
839         struct rb_node *p;
840         int mds_wanted = 0;
841
842         for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
843                 cap = rb_entry(p, struct ceph_cap, ci_node);
844                 if (!__cap_is_valid(cap))
845                         continue;
846                 mds_wanted |= cap->mds_wanted;
847         }
848         return mds_wanted;
849 }
850
851 /*
852  * called under i_lock
853  */
854 static int __ceph_is_any_caps(struct ceph_inode_info *ci)
855 {
856         return !RB_EMPTY_ROOT(&ci->i_caps) || ci->i_cap_exporting_mds >= 0;
857 }
858
859 /*
860  * caller should hold i_lock.
861  * caller will not hold session s_mutex if called from destroy_inode.
862  */
863 void __ceph_remove_cap(struct ceph_cap *cap)
864 {
865         struct ceph_mds_session *session = cap->session;
866         struct ceph_inode_info *ci = cap->ci;
867         struct ceph_mds_client *mdsc = &ceph_client(ci->vfs_inode.i_sb)->mdsc;
868
869         dout("__ceph_remove_cap %p from %p\n", cap, &ci->vfs_inode);
870
871         /* remove from inode list */
872         rb_erase(&cap->ci_node, &ci->i_caps);
873         cap->ci = NULL;
874         if (ci->i_auth_cap == cap)
875                 ci->i_auth_cap = NULL;
876
877         /* remove from session list */
878         spin_lock(&session->s_cap_lock);
879         if (session->s_cap_iterator == cap) {
880                 /* not yet, we are iterating over this very cap */
881                 dout("__ceph_remove_cap  delaying %p removal from session %p\n",
882                      cap, cap->session);
883         } else {
884                 list_del_init(&cap->session_caps);
885                 session->s_nr_caps--;
886                 cap->session = NULL;
887         }
888         spin_unlock(&session->s_cap_lock);
889
890         if (cap->session == NULL)
891                 ceph_put_cap(cap);
892
893         if (!__ceph_is_any_caps(ci) && ci->i_snap_realm) {
894                 struct ceph_snap_realm *realm = ci->i_snap_realm;
895                 spin_lock(&realm->inodes_with_caps_lock);
896                 list_del_init(&ci->i_snap_realm_item);
897                 ci->i_snap_realm_counter++;
898                 ci->i_snap_realm = NULL;
899                 spin_unlock(&realm->inodes_with_caps_lock);
900                 ceph_put_snap_realm(mdsc, realm);
901         }
902         if (!__ceph_is_any_real_caps(ci))
903                 __cap_delay_cancel(mdsc, ci);
904 }
905
906 /*
907  * Build and send a cap message to the given MDS.
908  *
909  * Caller should be holding s_mutex.
910  */
911 static int send_cap_msg(struct ceph_mds_session *session,
912                         u64 ino, u64 cid, int op,
913                         int caps, int wanted, int dirty,
914                         u32 seq, u64 flush_tid, u32 issue_seq, u32 mseq,
915                         u64 size, u64 max_size,
916                         struct timespec *mtime, struct timespec *atime,
917                         u64 time_warp_seq,
918                         uid_t uid, gid_t gid, mode_t mode,
919                         u64 xattr_version,
920                         struct ceph_buffer *xattrs_buf,
921                         u64 follows)
922 {
923         struct ceph_mds_caps *fc;
924         struct ceph_msg *msg;
925
926         dout("send_cap_msg %s %llx %llx caps %s wanted %s dirty %s"
927              " seq %u/%u mseq %u follows %lld size %llu/%llu"
928              " xattr_ver %llu xattr_len %d\n", ceph_cap_op_name(op),
929              cid, ino, ceph_cap_string(caps), ceph_cap_string(wanted),
930              ceph_cap_string(dirty),
931              seq, issue_seq, mseq, follows, size, max_size,
932              xattr_version, xattrs_buf ? (int)xattrs_buf->vec.iov_len : 0);
933
934         msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPS, sizeof(*fc), 0, 0, NULL);
935         if (IS_ERR(msg))
936                 return PTR_ERR(msg);
937
938         msg->hdr.tid = cpu_to_le64(flush_tid);
939
940         fc = msg->front.iov_base;
941         memset(fc, 0, sizeof(*fc));
942
943         fc->cap_id = cpu_to_le64(cid);
944         fc->op = cpu_to_le32(op);
945         fc->seq = cpu_to_le32(seq);
946         fc->issue_seq = cpu_to_le32(issue_seq);
947         fc->migrate_seq = cpu_to_le32(mseq);
948         fc->caps = cpu_to_le32(caps);
949         fc->wanted = cpu_to_le32(wanted);
950         fc->dirty = cpu_to_le32(dirty);
951         fc->ino = cpu_to_le64(ino);
952         fc->snap_follows = cpu_to_le64(follows);
953
954         fc->size = cpu_to_le64(size);
955         fc->max_size = cpu_to_le64(max_size);
956         if (mtime)
957                 ceph_encode_timespec(&fc->mtime, mtime);
958         if (atime)
959                 ceph_encode_timespec(&fc->atime, atime);
960         fc->time_warp_seq = cpu_to_le32(time_warp_seq);
961
962         fc->uid = cpu_to_le32(uid);
963         fc->gid = cpu_to_le32(gid);
964         fc->mode = cpu_to_le32(mode);
965
966         fc->xattr_version = cpu_to_le64(xattr_version);
967         if (xattrs_buf) {
968                 msg->middle = ceph_buffer_get(xattrs_buf);
969                 fc->xattr_len = cpu_to_le32(xattrs_buf->vec.iov_len);
970                 msg->hdr.middle_len = cpu_to_le32(xattrs_buf->vec.iov_len);
971         }
972
973         ceph_con_send(&session->s_con, msg);
974         return 0;
975 }
976
977 /*
978  * Queue cap releases when an inode is dropped from our cache.  Since
979  * inode is about to be destroyed, there is no need for i_lock.
980  */
981 void ceph_queue_caps_release(struct inode *inode)
982 {
983         struct ceph_inode_info *ci = ceph_inode(inode);
984         struct rb_node *p;
985
986         p = rb_first(&ci->i_caps);
987         while (p) {
988                 struct ceph_cap *cap = rb_entry(p, struct ceph_cap, ci_node);
989                 struct ceph_mds_session *session = cap->session;
990                 struct ceph_msg *msg;
991                 struct ceph_mds_cap_release *head;
992                 struct ceph_mds_cap_item *item;
993
994                 spin_lock(&session->s_cap_lock);
995                 BUG_ON(!session->s_num_cap_releases);
996                 msg = list_first_entry(&session->s_cap_releases,
997                                        struct ceph_msg, list_head);
998
999                 dout(" adding %p release to mds%d msg %p (%d left)\n",
1000                      inode, session->s_mds, msg, session->s_num_cap_releases);
1001
1002                 BUG_ON(msg->front.iov_len + sizeof(*item) > PAGE_CACHE_SIZE);
1003                 head = msg->front.iov_base;
1004                 head->num = cpu_to_le32(le32_to_cpu(head->num) + 1);
1005                 item = msg->front.iov_base + msg->front.iov_len;
1006                 item->ino = cpu_to_le64(ceph_ino(inode));
1007                 item->cap_id = cpu_to_le64(cap->cap_id);
1008                 item->migrate_seq = cpu_to_le32(cap->mseq);
1009                 item->seq = cpu_to_le32(cap->issue_seq);
1010
1011                 session->s_num_cap_releases--;
1012
1013                 msg->front.iov_len += sizeof(*item);
1014                 if (le32_to_cpu(head->num) == CEPH_CAPS_PER_RELEASE) {
1015                         dout(" release msg %p full\n", msg);
1016                         list_move_tail(&msg->list_head,
1017                                        &session->s_cap_releases_done);
1018                 } else {
1019                         dout(" release msg %p at %d/%d (%d)\n", msg,
1020                              (int)le32_to_cpu(head->num),
1021                              (int)CEPH_CAPS_PER_RELEASE,
1022                              (int)msg->front.iov_len);
1023                 }
1024                 spin_unlock(&session->s_cap_lock);
1025                 p = rb_next(p);
1026                 __ceph_remove_cap(cap);
1027         }
1028 }
1029
1030 /*
1031  * Send a cap msg on the given inode.  Update our caps state, then
1032  * drop i_lock and send the message.
1033  *
1034  * Make note of max_size reported/requested from mds, revoked caps
1035  * that have now been implemented.
1036  *
1037  * Make half-hearted attempt ot to invalidate page cache if we are
1038  * dropping RDCACHE.  Note that this will leave behind locked pages
1039  * that we'll then need to deal with elsewhere.
1040  *
1041  * Return non-zero if delayed release, or we experienced an error
1042  * such that the caller should requeue + retry later.
1043  *
1044  * called with i_lock, then drops it.
1045  * caller should hold snap_rwsem (read), s_mutex.
1046  */
1047 static int __send_cap(struct ceph_mds_client *mdsc, struct ceph_cap *cap,
1048                       int op, int used, int want, int retain, int flushing,
1049                       unsigned *pflush_tid)
1050         __releases(cap->ci->vfs_inode->i_lock)
1051 {
1052         struct ceph_inode_info *ci = cap->ci;
1053         struct inode *inode = &ci->vfs_inode;
1054         u64 cap_id = cap->cap_id;
1055         int held, revoking, dropping, keep;
1056         u64 seq, issue_seq, mseq, time_warp_seq, follows;
1057         u64 size, max_size;
1058         struct timespec mtime, atime;
1059         int wake = 0;
1060         mode_t mode;
1061         uid_t uid;
1062         gid_t gid;
1063         struct ceph_mds_session *session;
1064         u64 xattr_version = 0;
1065         int delayed = 0;
1066         u64 flush_tid = 0;
1067         int i;
1068         int ret;
1069
1070         held = cap->issued | cap->implemented;
1071         revoking = cap->implemented & ~cap->issued;
1072         retain &= ~revoking;
1073         dropping = cap->issued & ~retain;
1074
1075         dout("__send_cap %p cap %p session %p %s -> %s (revoking %s)\n",
1076              inode, cap, cap->session,
1077              ceph_cap_string(held), ceph_cap_string(held & retain),
1078              ceph_cap_string(revoking));
1079         BUG_ON((retain & CEPH_CAP_PIN) == 0);
1080
1081         session = cap->session;
1082
1083         /* don't release wanted unless we've waited a bit. */
1084         if ((ci->i_ceph_flags & CEPH_I_NODELAY) == 0 &&
1085             time_before(jiffies, ci->i_hold_caps_min)) {
1086                 dout(" delaying issued %s -> %s, wanted %s -> %s on send\n",
1087                      ceph_cap_string(cap->issued),
1088                      ceph_cap_string(cap->issued & retain),
1089                      ceph_cap_string(cap->mds_wanted),
1090                      ceph_cap_string(want));
1091                 want |= cap->mds_wanted;
1092                 retain |= cap->issued;
1093                 delayed = 1;
1094         }
1095         ci->i_ceph_flags &= ~(CEPH_I_NODELAY | CEPH_I_FLUSH);
1096
1097         cap->issued &= retain;  /* drop bits we don't want */
1098         if (cap->implemented & ~cap->issued) {
1099                 /*
1100                  * Wake up any waiters on wanted -> needed transition.
1101                  * This is due to the weird transition from buffered
1102                  * to sync IO... we need to flush dirty pages _before_
1103                  * allowing sync writes to avoid reordering.
1104                  */
1105                 wake = 1;
1106         }
1107         cap->implemented &= cap->issued | used;
1108         cap->mds_wanted = want;
1109
1110         if (flushing) {
1111                 /*
1112                  * assign a tid for flush operations so we can avoid
1113                  * flush1 -> dirty1 -> flush2 -> flushack1 -> mark
1114                  * clean type races.  track latest tid for every bit
1115                  * so we can handle flush AxFw, flush Fw, and have the
1116                  * first ack clean Ax.
1117                  */
1118                 flush_tid = ++ci->i_cap_flush_last_tid;
1119                 if (pflush_tid)
1120                         *pflush_tid = flush_tid;
1121                 dout(" cap_flush_tid %d\n", (int)flush_tid);
1122                 for (i = 0; i < CEPH_CAP_BITS; i++)
1123                         if (flushing & (1 << i))
1124                                 ci->i_cap_flush_tid[i] = flush_tid;
1125         }
1126
1127         keep = cap->implemented;
1128         seq = cap->seq;
1129         issue_seq = cap->issue_seq;
1130         mseq = cap->mseq;
1131         size = inode->i_size;
1132         ci->i_reported_size = size;
1133         max_size = ci->i_wanted_max_size;
1134         ci->i_requested_max_size = max_size;
1135         mtime = inode->i_mtime;
1136         atime = inode->i_atime;
1137         time_warp_seq = ci->i_time_warp_seq;
1138         follows = ci->i_snap_realm->cached_context->seq;
1139         uid = inode->i_uid;
1140         gid = inode->i_gid;
1141         mode = inode->i_mode;
1142
1143         if (dropping & CEPH_CAP_XATTR_EXCL) {
1144                 __ceph_build_xattrs_blob(ci);
1145                 xattr_version = ci->i_xattrs.version + 1;
1146         }
1147
1148         spin_unlock(&inode->i_lock);
1149
1150         ret = send_cap_msg(session, ceph_vino(inode).ino, cap_id,
1151                 op, keep, want, flushing, seq, flush_tid, issue_seq, mseq,
1152                 size, max_size, &mtime, &atime, time_warp_seq,
1153                 uid, gid, mode,
1154                 xattr_version,
1155                 (flushing & CEPH_CAP_XATTR_EXCL) ? ci->i_xattrs.blob : NULL,
1156                 follows);
1157         if (ret < 0) {
1158                 dout("error sending cap msg, must requeue %p\n", inode);
1159                 delayed = 1;
1160         }
1161
1162         if (wake)
1163                 wake_up(&ci->i_cap_wq);
1164
1165         return delayed;
1166 }
1167
1168 /*
1169  * When a snapshot is taken, clients accumulate dirty metadata on
1170  * inodes with capabilities in ceph_cap_snaps to describe the file
1171  * state at the time the snapshot was taken.  This must be flushed
1172  * asynchronously back to the MDS once sync writes complete and dirty
1173  * data is written out.
1174  *
1175  * Called under i_lock.  Takes s_mutex as needed.
1176  */
1177 void __ceph_flush_snaps(struct ceph_inode_info *ci,
1178                         struct ceph_mds_session **psession)
1179 {
1180         struct inode *inode = &ci->vfs_inode;
1181         int mds;
1182         struct ceph_cap_snap *capsnap;
1183         u32 mseq;
1184         struct ceph_mds_client *mdsc = &ceph_inode_to_client(inode)->mdsc;
1185         struct ceph_mds_session *session = NULL; /* if session != NULL, we hold
1186                                                     session->s_mutex */
1187         u64 next_follows = 0;  /* keep track of how far we've gotten through the
1188                              i_cap_snaps list, and skip these entries next time
1189                              around to avoid an infinite loop */
1190
1191         if (psession)
1192                 session = *psession;
1193
1194         dout("__flush_snaps %p\n", inode);
1195 retry:
1196         list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) {
1197                 /* avoid an infiniute loop after retry */
1198                 if (capsnap->follows < next_follows)
1199                         continue;
1200                 /*
1201                  * we need to wait for sync writes to complete and for dirty
1202                  * pages to be written out.
1203                  */
1204                 if (capsnap->dirty_pages || capsnap->writing)
1205                         continue;
1206
1207                 /* pick mds, take s_mutex */
1208                 mds = __ceph_get_cap_mds(ci, &mseq);
1209                 if (session && session->s_mds != mds) {
1210                         dout("oops, wrong session %p mutex\n", session);
1211                         mutex_unlock(&session->s_mutex);
1212                         ceph_put_mds_session(session);
1213                         session = NULL;
1214                 }
1215                 if (!session) {
1216                         spin_unlock(&inode->i_lock);
1217                         mutex_lock(&mdsc->mutex);
1218                         session = __ceph_lookup_mds_session(mdsc, mds);
1219                         mutex_unlock(&mdsc->mutex);
1220                         if (session) {
1221                                 dout("inverting session/ino locks on %p\n",
1222                                      session);
1223                                 mutex_lock(&session->s_mutex);
1224                         }
1225                         /*
1226                          * if session == NULL, we raced against a cap
1227                          * deletion.  retry, and we'll get a better
1228                          * @mds value next time.
1229                          */
1230                         spin_lock(&inode->i_lock);
1231                         goto retry;
1232                 }
1233
1234                 capsnap->flush_tid = ++ci->i_cap_flush_last_tid;
1235                 atomic_inc(&capsnap->nref);
1236                 if (!list_empty(&capsnap->flushing_item))
1237                         list_del_init(&capsnap->flushing_item);
1238                 list_add_tail(&capsnap->flushing_item,
1239                               &session->s_cap_snaps_flushing);
1240                 spin_unlock(&inode->i_lock);
1241
1242                 dout("flush_snaps %p cap_snap %p follows %lld size %llu\n",
1243                      inode, capsnap, next_follows, capsnap->size);
1244                 send_cap_msg(session, ceph_vino(inode).ino, 0,
1245                              CEPH_CAP_OP_FLUSHSNAP, capsnap->issued, 0,
1246                              capsnap->dirty, 0, capsnap->flush_tid, 0, mseq,
1247                              capsnap->size, 0,
1248                              &capsnap->mtime, &capsnap->atime,
1249                              capsnap->time_warp_seq,
1250                              capsnap->uid, capsnap->gid, capsnap->mode,
1251                              0, NULL,
1252                              capsnap->follows);
1253
1254                 next_follows = capsnap->follows + 1;
1255                 ceph_put_cap_snap(capsnap);
1256
1257                 spin_lock(&inode->i_lock);
1258                 goto retry;
1259         }
1260
1261         /* we flushed them all; remove this inode from the queue */
1262         spin_lock(&mdsc->snap_flush_lock);
1263         list_del_init(&ci->i_snap_flush_item);
1264         spin_unlock(&mdsc->snap_flush_lock);
1265
1266         if (psession)
1267                 *psession = session;
1268         else if (session) {
1269                 mutex_unlock(&session->s_mutex);
1270                 ceph_put_mds_session(session);
1271         }
1272 }
1273
1274 static void ceph_flush_snaps(struct ceph_inode_info *ci)
1275 {
1276         struct inode *inode = &ci->vfs_inode;
1277
1278         spin_lock(&inode->i_lock);
1279         __ceph_flush_snaps(ci, NULL);
1280         spin_unlock(&inode->i_lock);
1281 }
1282
1283 /*
1284  * Mark caps dirty.  If inode is newly dirty, add to the global dirty
1285  * list.
1286  */
1287 void __ceph_mark_dirty_caps(struct ceph_inode_info *ci, int mask)
1288 {
1289         struct ceph_mds_client *mdsc = &ceph_client(ci->vfs_inode.i_sb)->mdsc;
1290         struct inode *inode = &ci->vfs_inode;
1291         int was = ci->i_dirty_caps;
1292         int dirty = 0;
1293
1294         dout("__mark_dirty_caps %p %s dirty %s -> %s\n", &ci->vfs_inode,
1295              ceph_cap_string(mask), ceph_cap_string(was),
1296              ceph_cap_string(was | mask));
1297         ci->i_dirty_caps |= mask;
1298         if (was == 0) {
1299                 dout(" inode %p now dirty\n", &ci->vfs_inode);
1300                 BUG_ON(!list_empty(&ci->i_dirty_item));
1301                 spin_lock(&mdsc->cap_dirty_lock);
1302                 list_add(&ci->i_dirty_item, &mdsc->cap_dirty);
1303                 spin_unlock(&mdsc->cap_dirty_lock);
1304                 if (ci->i_flushing_caps == 0) {
1305                         igrab(inode);
1306                         dirty |= I_DIRTY_SYNC;
1307                 }
1308         }
1309         BUG_ON(list_empty(&ci->i_dirty_item));
1310         if (((was | ci->i_flushing_caps) & CEPH_CAP_FILE_BUFFER) &&
1311             (mask & CEPH_CAP_FILE_BUFFER))
1312                 dirty |= I_DIRTY_DATASYNC;
1313         if (dirty)
1314                 __mark_inode_dirty(inode, dirty);
1315         __cap_delay_requeue(mdsc, ci);
1316 }
1317
1318 /*
1319  * Add dirty inode to the flushing list.  Assigned a seq number so we
1320  * can wait for caps to flush without starving.
1321  *
1322  * Called under i_lock.
1323  */
1324 static int __mark_caps_flushing(struct inode *inode,
1325                                  struct ceph_mds_session *session)
1326 {
1327         struct ceph_mds_client *mdsc = &ceph_client(inode->i_sb)->mdsc;
1328         struct ceph_inode_info *ci = ceph_inode(inode);
1329         int flushing;
1330
1331         BUG_ON(ci->i_dirty_caps == 0);
1332         BUG_ON(list_empty(&ci->i_dirty_item));
1333
1334         flushing = ci->i_dirty_caps;
1335         dout("__mark_caps_flushing flushing %s, flushing_caps %s -> %s\n",
1336              ceph_cap_string(flushing),
1337              ceph_cap_string(ci->i_flushing_caps),
1338              ceph_cap_string(ci->i_flushing_caps | flushing));
1339         ci->i_flushing_caps |= flushing;
1340         ci->i_dirty_caps = 0;
1341         dout(" inode %p now !dirty\n", inode);
1342
1343         spin_lock(&mdsc->cap_dirty_lock);
1344         list_del_init(&ci->i_dirty_item);
1345
1346         ci->i_cap_flush_seq = ++mdsc->cap_flush_seq;
1347         if (list_empty(&ci->i_flushing_item)) {
1348                 list_add_tail(&ci->i_flushing_item, &session->s_cap_flushing);
1349                 mdsc->num_cap_flushing++;
1350                 dout(" inode %p now flushing seq %lld\n", inode,
1351                      ci->i_cap_flush_seq);
1352         } else {
1353                 list_move_tail(&ci->i_flushing_item, &session->s_cap_flushing);
1354                 dout(" inode %p now flushing (more) seq %lld\n", inode,
1355                      ci->i_cap_flush_seq);
1356         }
1357         spin_unlock(&mdsc->cap_dirty_lock);
1358
1359         return flushing;
1360 }
1361
1362 /*
1363  * try to invalidate mapping pages without blocking.
1364  */
1365 static int mapping_is_empty(struct address_space *mapping)
1366 {
1367         struct page *page = find_get_page(mapping, 0);
1368
1369         if (!page)
1370                 return 1;
1371
1372         put_page(page);
1373         return 0;
1374 }
1375
1376 static int try_nonblocking_invalidate(struct inode *inode)
1377 {
1378         struct ceph_inode_info *ci = ceph_inode(inode);
1379         u32 invalidating_gen = ci->i_rdcache_gen;
1380
1381         spin_unlock(&inode->i_lock);
1382         invalidate_mapping_pages(&inode->i_data, 0, -1);
1383         spin_lock(&inode->i_lock);
1384
1385         if (mapping_is_empty(&inode->i_data) &&
1386             invalidating_gen == ci->i_rdcache_gen) {
1387                 /* success. */
1388                 dout("try_nonblocking_invalidate %p success\n", inode);
1389                 ci->i_rdcache_gen = 0;
1390                 ci->i_rdcache_revoking = 0;
1391                 return 0;
1392         }
1393         dout("try_nonblocking_invalidate %p failed\n", inode);
1394         return -1;
1395 }
1396
1397 /*
1398  * Swiss army knife function to examine currently used and wanted
1399  * versus held caps.  Release, flush, ack revoked caps to mds as
1400  * appropriate.
1401  *
1402  *  CHECK_CAPS_NODELAY - caller is delayed work and we should not delay
1403  *    cap release further.
1404  *  CHECK_CAPS_AUTHONLY - we should only check the auth cap
1405  *  CHECK_CAPS_FLUSH - we should flush any dirty caps immediately, without
1406  *    further delay.
1407  */
1408 void ceph_check_caps(struct ceph_inode_info *ci, int flags,
1409                      struct ceph_mds_session *session)
1410         __releases(session->s_mutex)
1411 {
1412         struct ceph_client *client = ceph_inode_to_client(&ci->vfs_inode);
1413         struct ceph_mds_client *mdsc = &client->mdsc;
1414         struct inode *inode = &ci->vfs_inode;
1415         struct ceph_cap *cap;
1416         int file_wanted, used;
1417         int took_snap_rwsem = 0;             /* true if mdsc->snap_rwsem held */
1418         int issued, implemented, want, retain, revoking, flushing = 0;
1419         int mds = -1;   /* keep track of how far we've gone through i_caps list
1420                            to avoid an infinite loop on retry */
1421         struct rb_node *p;
1422         int tried_invalidate = 0;
1423         int delayed = 0, sent = 0, force_requeue = 0, num;
1424         int queue_invalidate = 0;
1425         int is_delayed = flags & CHECK_CAPS_NODELAY;
1426
1427         /* if we are unmounting, flush any unused caps immediately. */
1428         if (mdsc->stopping)
1429                 is_delayed = 1;
1430
1431         spin_lock(&inode->i_lock);
1432
1433         if (ci->i_ceph_flags & CEPH_I_FLUSH)
1434                 flags |= CHECK_CAPS_FLUSH;
1435
1436         /* flush snaps first time around only */
1437         if (!list_empty(&ci->i_cap_snaps))
1438                 __ceph_flush_snaps(ci, &session);
1439         goto retry_locked;
1440 retry:
1441         spin_lock(&inode->i_lock);
1442 retry_locked:
1443         file_wanted = __ceph_caps_file_wanted(ci);
1444         used = __ceph_caps_used(ci);
1445         want = file_wanted | used;
1446         issued = __ceph_caps_issued(ci, &implemented);
1447         revoking = implemented & ~issued;
1448
1449         retain = want | CEPH_CAP_PIN;
1450         if (!mdsc->stopping && inode->i_nlink > 0) {
1451                 if (want) {
1452                         retain |= CEPH_CAP_ANY;       /* be greedy */
1453                 } else {
1454                         retain |= CEPH_CAP_ANY_SHARED;
1455                         /*
1456                          * keep RD only if we didn't have the file open RW,
1457                          * because then the mds would revoke it anyway to
1458                          * journal max_size=0.
1459                          */
1460                         if (ci->i_max_size == 0)
1461                                 retain |= CEPH_CAP_ANY_RD;
1462                 }
1463         }
1464
1465         dout("check_caps %p file_want %s used %s dirty %s flushing %s"
1466              " issued %s revoking %s retain %s %s%s%s\n", inode,
1467              ceph_cap_string(file_wanted),
1468              ceph_cap_string(used), ceph_cap_string(ci->i_dirty_caps),
1469              ceph_cap_string(ci->i_flushing_caps),
1470              ceph_cap_string(issued), ceph_cap_string(revoking),
1471              ceph_cap_string(retain),
1472              (flags & CHECK_CAPS_AUTHONLY) ? " AUTHONLY" : "",
1473              (flags & CHECK_CAPS_NODELAY) ? " NODELAY" : "",
1474              (flags & CHECK_CAPS_FLUSH) ? " FLUSH" : "");
1475
1476         /*
1477          * If we no longer need to hold onto old our caps, and we may
1478          * have cached pages, but don't want them, then try to invalidate.
1479          * If we fail, it's because pages are locked.... try again later.
1480          */
1481         if ((!is_delayed || mdsc->stopping) &&
1482             ci->i_wrbuffer_ref == 0 &&               /* no dirty pages... */
1483             ci->i_rdcache_gen &&                     /* may have cached pages */
1484             (file_wanted == 0 ||                     /* no open files */
1485              (revoking & CEPH_CAP_FILE_CACHE)) &&     /*  or revoking cache */
1486             !tried_invalidate) {
1487                 dout("check_caps trying to invalidate on %p\n", inode);
1488                 if (try_nonblocking_invalidate(inode) < 0) {
1489                         if (revoking & CEPH_CAP_FILE_CACHE) {
1490                                 dout("check_caps queuing invalidate\n");
1491                                 queue_invalidate = 1;
1492                                 ci->i_rdcache_revoking = ci->i_rdcache_gen;
1493                         } else {
1494                                 dout("check_caps failed to invalidate pages\n");
1495                                 /* we failed to invalidate pages.  check these
1496                                    caps again later. */
1497                                 force_requeue = 1;
1498                                 __cap_set_timeouts(mdsc, ci);
1499                         }
1500                 }
1501                 tried_invalidate = 1;
1502                 goto retry_locked;
1503         }
1504
1505         num = 0;
1506         for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
1507                 cap = rb_entry(p, struct ceph_cap, ci_node);
1508                 num++;
1509
1510                 /* avoid looping forever */
1511                 if (mds >= cap->mds ||
1512                     ((flags & CHECK_CAPS_AUTHONLY) && cap != ci->i_auth_cap))
1513                         continue;
1514
1515                 /* NOTE: no side-effects allowed, until we take s_mutex */
1516
1517                 revoking = cap->implemented & ~cap->issued;
1518                 if (revoking)
1519                         dout(" mds%d revoking %s\n", cap->mds,
1520                              ceph_cap_string(revoking));
1521
1522                 if (cap == ci->i_auth_cap &&
1523                     (cap->issued & CEPH_CAP_FILE_WR)) {
1524                         /* request larger max_size from MDS? */
1525                         if (ci->i_wanted_max_size > ci->i_max_size &&
1526                             ci->i_wanted_max_size > ci->i_requested_max_size) {
1527                                 dout("requesting new max_size\n");
1528                                 goto ack;
1529                         }
1530
1531                         /* approaching file_max? */
1532                         if ((inode->i_size << 1) >= ci->i_max_size &&
1533                             (ci->i_reported_size << 1) < ci->i_max_size) {
1534                                 dout("i_size approaching max_size\n");
1535                                 goto ack;
1536                         }
1537                 }
1538                 /* flush anything dirty? */
1539                 if (cap == ci->i_auth_cap && (flags & CHECK_CAPS_FLUSH) &&
1540                     ci->i_dirty_caps) {
1541                         dout("flushing dirty caps\n");
1542                         goto ack;
1543                 }
1544
1545                 /* completed revocation? going down and there are no caps? */
1546                 if (revoking && (revoking & used) == 0) {
1547                         dout("completed revocation of %s\n",
1548                              ceph_cap_string(cap->implemented & ~cap->issued));
1549                         goto ack;
1550                 }
1551
1552                 /* want more caps from mds? */
1553                 if (want & ~(cap->mds_wanted | cap->issued))
1554                         goto ack;
1555
1556                 /* things we might delay */
1557                 if ((cap->issued & ~retain) == 0 &&
1558                     cap->mds_wanted == want)
1559                         continue;     /* nope, all good */
1560
1561                 if (is_delayed)
1562                         goto ack;
1563
1564                 /* delay? */
1565                 if ((ci->i_ceph_flags & CEPH_I_NODELAY) == 0 &&
1566                     time_before(jiffies, ci->i_hold_caps_max)) {
1567                         dout(" delaying issued %s -> %s, wanted %s -> %s\n",
1568                              ceph_cap_string(cap->issued),
1569                              ceph_cap_string(cap->issued & retain),
1570                              ceph_cap_string(cap->mds_wanted),
1571                              ceph_cap_string(want));
1572                         delayed++;
1573                         continue;
1574                 }
1575
1576 ack:
1577                 if (ci->i_ceph_flags & CEPH_I_NOFLUSH) {
1578                         dout(" skipping %p I_NOFLUSH set\n", inode);
1579                         continue;
1580                 }
1581
1582                 if (session && session != cap->session) {
1583                         dout("oops, wrong session %p mutex\n", session);
1584                         mutex_unlock(&session->s_mutex);
1585                         session = NULL;
1586                 }
1587                 if (!session) {
1588                         session = cap->session;
1589                         if (mutex_trylock(&session->s_mutex) == 0) {
1590                                 dout("inverting session/ino locks on %p\n",
1591                                      session);
1592                                 spin_unlock(&inode->i_lock);
1593                                 if (took_snap_rwsem) {
1594                                         up_read(&mdsc->snap_rwsem);
1595                                         took_snap_rwsem = 0;
1596                                 }
1597                                 mutex_lock(&session->s_mutex);
1598                                 goto retry;
1599                         }
1600                 }
1601                 /* take snap_rwsem after session mutex */
1602                 if (!took_snap_rwsem) {
1603                         if (down_read_trylock(&mdsc->snap_rwsem) == 0) {
1604                                 dout("inverting snap/in locks on %p\n",
1605                                      inode);
1606                                 spin_unlock(&inode->i_lock);
1607                                 down_read(&mdsc->snap_rwsem);
1608                                 took_snap_rwsem = 1;
1609                                 goto retry;
1610                         }
1611                         took_snap_rwsem = 1;
1612                 }
1613
1614                 if (cap == ci->i_auth_cap && ci->i_dirty_caps)
1615                         flushing = __mark_caps_flushing(inode, session);
1616
1617                 mds = cap->mds;  /* remember mds, so we don't repeat */
1618                 sent++;
1619
1620                 /* __send_cap drops i_lock */
1621                 delayed += __send_cap(mdsc, cap, CEPH_CAP_OP_UPDATE, used, want,
1622                                       retain, flushing, NULL);
1623                 goto retry; /* retake i_lock and restart our cap scan. */
1624         }
1625
1626         /*
1627          * Reschedule delayed caps release if we delayed anything,
1628          * otherwise cancel.
1629          */
1630         if (delayed && is_delayed)
1631                 force_requeue = 1;   /* __send_cap delayed release; requeue */
1632         if (!delayed && !is_delayed)
1633                 __cap_delay_cancel(mdsc, ci);
1634         else if (!is_delayed || force_requeue)
1635                 __cap_delay_requeue(mdsc, ci);
1636
1637         spin_unlock(&inode->i_lock);
1638
1639         if (queue_invalidate)
1640                 ceph_queue_invalidate(inode);
1641
1642         if (session)
1643                 mutex_unlock(&session->s_mutex);
1644         if (took_snap_rwsem)
1645                 up_read(&mdsc->snap_rwsem);
1646 }
1647
1648 /*
1649  * Try to flush dirty caps back to the auth mds.
1650  */
1651 static int try_flush_caps(struct inode *inode, struct ceph_mds_session *session,
1652                           unsigned *flush_tid)
1653 {
1654         struct ceph_mds_client *mdsc = &ceph_client(inode->i_sb)->mdsc;
1655         struct ceph_inode_info *ci = ceph_inode(inode);
1656         int unlock_session = session ? 0 : 1;
1657         int flushing = 0;
1658
1659 retry:
1660         spin_lock(&inode->i_lock);
1661         if (ci->i_ceph_flags & CEPH_I_NOFLUSH) {
1662                 dout("try_flush_caps skipping %p I_NOFLUSH set\n", inode);
1663                 goto out;
1664         }
1665         if (ci->i_dirty_caps && ci->i_auth_cap) {
1666                 struct ceph_cap *cap = ci->i_auth_cap;
1667                 int used = __ceph_caps_used(ci);
1668                 int want = __ceph_caps_wanted(ci);
1669                 int delayed;
1670
1671                 if (!session) {
1672                         spin_unlock(&inode->i_lock);
1673                         session = cap->session;
1674                         mutex_lock(&session->s_mutex);
1675                         goto retry;
1676                 }
1677                 BUG_ON(session != cap->session);
1678                 if (cap->session->s_state < CEPH_MDS_SESSION_OPEN)
1679                         goto out;
1680
1681                 flushing = __mark_caps_flushing(inode, session);
1682
1683                 /* __send_cap drops i_lock */
1684                 delayed = __send_cap(mdsc, cap, CEPH_CAP_OP_FLUSH, used, want,
1685                                      cap->issued | cap->implemented, flushing,
1686                                      flush_tid);
1687                 if (!delayed)
1688                         goto out_unlocked;
1689
1690                 spin_lock(&inode->i_lock);
1691                 __cap_delay_requeue(mdsc, ci);
1692         }
1693 out:
1694         spin_unlock(&inode->i_lock);
1695 out_unlocked:
1696         if (session && unlock_session)
1697                 mutex_unlock(&session->s_mutex);
1698         return flushing;
1699 }
1700
1701 /*
1702  * Return true if we've flushed caps through the given flush_tid.
1703  */
1704 static int caps_are_flushed(struct inode *inode, unsigned tid)
1705 {
1706         struct ceph_inode_info *ci = ceph_inode(inode);
1707         int dirty, i, ret = 1;
1708
1709         spin_lock(&inode->i_lock);
1710         dirty = __ceph_caps_dirty(ci);
1711         for (i = 0; i < CEPH_CAP_BITS; i++)
1712                 if ((ci->i_flushing_caps & (1 << i)) &&
1713                     ci->i_cap_flush_tid[i] <= tid) {
1714                         /* still flushing this bit */
1715                         ret = 0;
1716                         break;
1717                 }
1718         spin_unlock(&inode->i_lock);
1719         return ret;
1720 }
1721
1722 /*
1723  * Wait on any unsafe replies for the given inode.  First wait on the
1724  * newest request, and make that the upper bound.  Then, if there are
1725  * more requests, keep waiting on the oldest as long as it is still older
1726  * than the original request.
1727  */
1728 static void sync_write_wait(struct inode *inode)
1729 {
1730         struct ceph_inode_info *ci = ceph_inode(inode);
1731         struct list_head *head = &ci->i_unsafe_writes;
1732         struct ceph_osd_request *req;
1733         u64 last_tid;
1734
1735         spin_lock(&ci->i_unsafe_lock);
1736         if (list_empty(head))
1737                 goto out;
1738
1739         /* set upper bound as _last_ entry in chain */
1740         req = list_entry(head->prev, struct ceph_osd_request,
1741                          r_unsafe_item);
1742         last_tid = req->r_tid;
1743
1744         do {
1745                 ceph_osdc_get_request(req);
1746                 spin_unlock(&ci->i_unsafe_lock);
1747                 dout("sync_write_wait on tid %llu (until %llu)\n",
1748                      req->r_tid, last_tid);
1749                 wait_for_completion(&req->r_safe_completion);
1750                 spin_lock(&ci->i_unsafe_lock);
1751                 ceph_osdc_put_request(req);
1752
1753                 /*
1754                  * from here on look at first entry in chain, since we
1755                  * only want to wait for anything older than last_tid
1756                  */
1757                 if (list_empty(head))
1758                         break;
1759                 req = list_entry(head->next, struct ceph_osd_request,
1760                                  r_unsafe_item);
1761         } while (req->r_tid < last_tid);
1762 out:
1763         spin_unlock(&ci->i_unsafe_lock);
1764 }
1765
1766 int ceph_fsync(struct file *file, struct dentry *dentry, int datasync)
1767 {
1768         struct inode *inode = dentry->d_inode;
1769         struct ceph_inode_info *ci = ceph_inode(inode);
1770         unsigned flush_tid;
1771         int ret;
1772         int dirty;
1773
1774         dout("fsync %p%s\n", inode, datasync ? " datasync" : "");
1775         sync_write_wait(inode);
1776
1777         ret = filemap_write_and_wait(inode->i_mapping);
1778         if (ret < 0)
1779                 return ret;
1780
1781         dirty = try_flush_caps(inode, NULL, &flush_tid);
1782         dout("fsync dirty caps are %s\n", ceph_cap_string(dirty));
1783
1784         /*
1785          * only wait on non-file metadata writeback (the mds
1786          * can recover size and mtime, so we don't need to
1787          * wait for that)
1788          */
1789         if (!datasync && (dirty & ~CEPH_CAP_ANY_FILE_WR)) {
1790                 dout("fsync waiting for flush_tid %u\n", flush_tid);
1791                 ret = wait_event_interruptible(ci->i_cap_wq,
1792                                        caps_are_flushed(inode, flush_tid));
1793         }
1794
1795         dout("fsync %p%s done\n", inode, datasync ? " datasync" : "");
1796         return ret;
1797 }
1798
1799 /*
1800  * Flush any dirty caps back to the mds.  If we aren't asked to wait,
1801  * queue inode for flush but don't do so immediately, because we can
1802  * get by with fewer MDS messages if we wait for data writeback to
1803  * complete first.
1804  */
1805 int ceph_write_inode(struct inode *inode, struct writeback_control *wbc)
1806 {
1807         struct ceph_inode_info *ci = ceph_inode(inode);
1808         unsigned flush_tid;
1809         int err = 0;
1810         int dirty;
1811         int wait = wbc->sync_mode == WB_SYNC_ALL;
1812
1813         dout("write_inode %p wait=%d\n", inode, wait);
1814         if (wait) {
1815                 dirty = try_flush_caps(inode, NULL, &flush_tid);
1816                 if (dirty)
1817                         err = wait_event_interruptible(ci->i_cap_wq,
1818                                        caps_are_flushed(inode, flush_tid));
1819         } else {
1820                 struct ceph_mds_client *mdsc = &ceph_client(inode->i_sb)->mdsc;
1821
1822                 spin_lock(&inode->i_lock);
1823                 if (__ceph_caps_dirty(ci))
1824                         __cap_delay_requeue_front(mdsc, ci);
1825                 spin_unlock(&inode->i_lock);
1826         }
1827         return err;
1828 }
1829
1830 /*
1831  * After a recovering MDS goes active, we need to resend any caps
1832  * we were flushing.
1833  *
1834  * Caller holds session->s_mutex.
1835  */
1836 static void kick_flushing_capsnaps(struct ceph_mds_client *mdsc,
1837                                    struct ceph_mds_session *session)
1838 {
1839         struct ceph_cap_snap *capsnap;
1840
1841         dout("kick_flushing_capsnaps mds%d\n", session->s_mds);
1842         list_for_each_entry(capsnap, &session->s_cap_snaps_flushing,
1843                             flushing_item) {
1844                 struct ceph_inode_info *ci = capsnap->ci;
1845                 struct inode *inode = &ci->vfs_inode;
1846                 struct ceph_cap *cap;
1847
1848                 spin_lock(&inode->i_lock);
1849                 cap = ci->i_auth_cap;
1850                 if (cap && cap->session == session) {
1851                         dout("kick_flushing_caps %p cap %p capsnap %p\n", inode,
1852                              cap, capsnap);
1853                         __ceph_flush_snaps(ci, &session);
1854                 } else {
1855                         pr_err("%p auth cap %p not mds%d ???\n", inode,
1856                                cap, session->s_mds);
1857                         spin_unlock(&inode->i_lock);
1858                 }
1859         }
1860 }
1861
1862 void ceph_kick_flushing_caps(struct ceph_mds_client *mdsc,
1863                              struct ceph_mds_session *session)
1864 {
1865         struct ceph_inode_info *ci;
1866
1867         kick_flushing_capsnaps(mdsc, session);
1868
1869         dout("kick_flushing_caps mds%d\n", session->s_mds);
1870         list_for_each_entry(ci, &session->s_cap_flushing, i_flushing_item) {
1871                 struct inode *inode = &ci->vfs_inode;
1872                 struct ceph_cap *cap;
1873                 int delayed = 0;
1874
1875                 spin_lock(&inode->i_lock);
1876                 cap = ci->i_auth_cap;
1877                 if (cap && cap->session == session) {
1878                         dout("kick_flushing_caps %p cap %p %s\n", inode,
1879                              cap, ceph_cap_string(ci->i_flushing_caps));
1880                         delayed = __send_cap(mdsc, cap, CEPH_CAP_OP_FLUSH,
1881                                              __ceph_caps_used(ci),
1882                                              __ceph_caps_wanted(ci),
1883                                              cap->issued | cap->implemented,
1884                                              ci->i_flushing_caps, NULL);
1885                         if (delayed) {
1886                                 spin_lock(&inode->i_lock);
1887                                 __cap_delay_requeue(mdsc, ci);
1888                                 spin_unlock(&inode->i_lock);
1889                         }
1890                 } else {
1891                         pr_err("%p auth cap %p not mds%d ???\n", inode,
1892                                cap, session->s_mds);
1893                         spin_unlock(&inode->i_lock);
1894                 }
1895         }
1896 }
1897
1898
1899 /*
1900  * Take references to capabilities we hold, so that we don't release
1901  * them to the MDS prematurely.
1902  *
1903  * Protected by i_lock.
1904  */
1905 static void __take_cap_refs(struct ceph_inode_info *ci, int got)
1906 {
1907         if (got & CEPH_CAP_PIN)
1908                 ci->i_pin_ref++;
1909         if (got & CEPH_CAP_FILE_RD)
1910                 ci->i_rd_ref++;
1911         if (got & CEPH_CAP_FILE_CACHE)
1912                 ci->i_rdcache_ref++;
1913         if (got & CEPH_CAP_FILE_WR)
1914                 ci->i_wr_ref++;
1915         if (got & CEPH_CAP_FILE_BUFFER) {
1916                 if (ci->i_wrbuffer_ref == 0)
1917                         igrab(&ci->vfs_inode);
1918                 ci->i_wrbuffer_ref++;
1919                 dout("__take_cap_refs %p wrbuffer %d -> %d (?)\n",
1920                      &ci->vfs_inode, ci->i_wrbuffer_ref-1, ci->i_wrbuffer_ref);
1921         }
1922 }
1923
1924 /*
1925  * Try to grab cap references.  Specify those refs we @want, and the
1926  * minimal set we @need.  Also include the larger offset we are writing
1927  * to (when applicable), and check against max_size here as well.
1928  * Note that caller is responsible for ensuring max_size increases are
1929  * requested from the MDS.
1930  */
1931 static int try_get_cap_refs(struct ceph_inode_info *ci, int need, int want,
1932                             int *got, loff_t endoff, int *check_max, int *err)
1933 {
1934         struct inode *inode = &ci->vfs_inode;
1935         int ret = 0;
1936         int have, implemented;
1937         int file_wanted;
1938
1939         dout("get_cap_refs %p need %s want %s\n", inode,
1940              ceph_cap_string(need), ceph_cap_string(want));
1941         spin_lock(&inode->i_lock);
1942
1943         /* make sure file is actually open */
1944         file_wanted = __ceph_caps_file_wanted(ci);
1945         if ((file_wanted & need) == 0) {
1946                 dout("try_get_cap_refs need %s file_wanted %s, EBADF\n",
1947                      ceph_cap_string(need), ceph_cap_string(file_wanted));
1948                 *err = -EBADF;
1949                 ret = 1;
1950                 goto out;
1951         }
1952
1953         if (need & CEPH_CAP_FILE_WR) {
1954                 if (endoff >= 0 && endoff > (loff_t)ci->i_max_size) {
1955                         dout("get_cap_refs %p endoff %llu > maxsize %llu\n",
1956                              inode, endoff, ci->i_max_size);
1957                         if (endoff > ci->i_wanted_max_size) {
1958                                 *check_max = 1;
1959                                 ret = 1;
1960                         }
1961                         goto out;
1962                 }
1963                 /*
1964                  * If a sync write is in progress, we must wait, so that we
1965                  * can get a final snapshot value for size+mtime.
1966                  */
1967                 if (__ceph_have_pending_cap_snap(ci)) {
1968                         dout("get_cap_refs %p cap_snap_pending\n", inode);
1969                         goto out;
1970                 }
1971         }
1972         have = __ceph_caps_issued(ci, &implemented);
1973
1974         /*
1975          * disallow writes while a truncate is pending
1976          */
1977         if (ci->i_truncate_pending)
1978                 have &= ~CEPH_CAP_FILE_WR;
1979
1980         if ((have & need) == need) {
1981                 /*
1982                  * Look at (implemented & ~have & not) so that we keep waiting
1983                  * on transition from wanted -> needed caps.  This is needed
1984                  * for WRBUFFER|WR -> WR to avoid a new WR sync write from
1985                  * going before a prior buffered writeback happens.
1986                  */
1987                 int not = want & ~(have & need);
1988                 int revoking = implemented & ~have;
1989                 dout("get_cap_refs %p have %s but not %s (revoking %s)\n",
1990                      inode, ceph_cap_string(have), ceph_cap_string(not),
1991                      ceph_cap_string(revoking));
1992                 if ((revoking & not) == 0) {
1993                         *got = need | (have & want);
1994                         __take_cap_refs(ci, *got);
1995                         ret = 1;
1996                 }
1997         } else {
1998                 dout("get_cap_refs %p have %s needed %s\n", inode,
1999                      ceph_cap_string(have), ceph_cap_string(need));
2000         }
2001 out:
2002         spin_unlock(&inode->i_lock);
2003         dout("get_cap_refs %p ret %d got %s\n", inode,
2004              ret, ceph_cap_string(*got));
2005         return ret;
2006 }
2007
2008 /*
2009  * Check the offset we are writing up to against our current
2010  * max_size.  If necessary, tell the MDS we want to write to
2011  * a larger offset.
2012  */
2013 static void check_max_size(struct inode *inode, loff_t endoff)
2014 {
2015         struct ceph_inode_info *ci = ceph_inode(inode);
2016         int check = 0;
2017
2018         /* do we need to explicitly request a larger max_size? */
2019         spin_lock(&inode->i_lock);
2020         if ((endoff >= ci->i_max_size ||
2021              endoff > (inode->i_size << 1)) &&
2022             endoff > ci->i_wanted_max_size) {
2023                 dout("write %p at large endoff %llu, req max_size\n",
2024                      inode, endoff);
2025                 ci->i_wanted_max_size = endoff;
2026                 check = 1;
2027         }
2028         spin_unlock(&inode->i_lock);
2029         if (check)
2030                 ceph_check_caps(ci, CHECK_CAPS_AUTHONLY, NULL);
2031 }
2032
2033 /*
2034  * Wait for caps, and take cap references.  If we can't get a WR cap
2035  * due to a small max_size, make sure we check_max_size (and possibly
2036  * ask the mds) so we don't get hung up indefinitely.
2037  */
2038 int ceph_get_caps(struct ceph_inode_info *ci, int need, int want, int *got,
2039                   loff_t endoff)
2040 {
2041         int check_max, ret, err;
2042
2043 retry:
2044         if (endoff > 0)
2045                 check_max_size(&ci->vfs_inode, endoff);
2046         check_max = 0;
2047         err = 0;
2048         ret = wait_event_interruptible(ci->i_cap_wq,
2049                                        try_get_cap_refs(ci, need, want,
2050                                                         got, endoff,
2051                                                         &check_max, &err));
2052         if (err)
2053                 ret = err;
2054         if (check_max)
2055                 goto retry;
2056         return ret;
2057 }
2058
2059 /*
2060  * Take cap refs.  Caller must already know we hold at least one ref
2061  * on the caps in question or we don't know this is safe.
2062  */
2063 void ceph_get_cap_refs(struct ceph_inode_info *ci, int caps)
2064 {
2065         spin_lock(&ci->vfs_inode.i_lock);
2066         __take_cap_refs(ci, caps);
2067         spin_unlock(&ci->vfs_inode.i_lock);
2068 }
2069
2070 /*
2071  * Release cap refs.
2072  *
2073  * If we released the last ref on any given cap, call ceph_check_caps
2074  * to release (or schedule a release).
2075  *
2076  * If we are releasing a WR cap (from a sync write), finalize any affected
2077  * cap_snap, and wake up any waiters.
2078  */
2079 void ceph_put_cap_refs(struct ceph_inode_info *ci, int had)
2080 {
2081         struct inode *inode = &ci->vfs_inode;
2082         int last = 0, put = 0, flushsnaps = 0, wake = 0;
2083         struct ceph_cap_snap *capsnap;
2084
2085         spin_lock(&inode->i_lock);
2086         if (had & CEPH_CAP_PIN)
2087                 --ci->i_pin_ref;
2088         if (had & CEPH_CAP_FILE_RD)
2089                 if (--ci->i_rd_ref == 0)
2090                         last++;
2091         if (had & CEPH_CAP_FILE_CACHE)
2092                 if (--ci->i_rdcache_ref == 0)
2093                         last++;
2094         if (had & CEPH_CAP_FILE_BUFFER) {
2095                 if (--ci->i_wrbuffer_ref == 0) {
2096                         last++;
2097                         put++;
2098                 }
2099                 dout("put_cap_refs %p wrbuffer %d -> %d (?)\n",
2100                      inode, ci->i_wrbuffer_ref+1, ci->i_wrbuffer_ref);
2101         }
2102         if (had & CEPH_CAP_FILE_WR)
2103                 if (--ci->i_wr_ref == 0) {
2104                         last++;
2105                         if (!list_empty(&ci->i_cap_snaps)) {
2106                                 capsnap = list_first_entry(&ci->i_cap_snaps,
2107                                                      struct ceph_cap_snap,
2108                                                      ci_item);
2109                                 if (capsnap->writing) {
2110                                         capsnap->writing = 0;
2111                                         flushsnaps =
2112                                                 __ceph_finish_cap_snap(ci,
2113                                                                        capsnap);
2114                                         wake = 1;
2115                                 }
2116                         }
2117                 }
2118         spin_unlock(&inode->i_lock);
2119
2120         dout("put_cap_refs %p had %s %s\n", inode, ceph_cap_string(had),
2121              last ? "last" : "");
2122
2123         if (last && !flushsnaps)
2124                 ceph_check_caps(ci, 0, NULL);
2125         else if (flushsnaps)
2126                 ceph_flush_snaps(ci);
2127         if (wake)
2128                 wake_up(&ci->i_cap_wq);
2129         if (put)
2130                 iput(inode);
2131 }
2132
2133 /*
2134  * Release @nr WRBUFFER refs on dirty pages for the given @snapc snap
2135  * context.  Adjust per-snap dirty page accounting as appropriate.
2136  * Once all dirty data for a cap_snap is flushed, flush snapped file
2137  * metadata back to the MDS.  If we dropped the last ref, call
2138  * ceph_check_caps.
2139  */
2140 void ceph_put_wrbuffer_cap_refs(struct ceph_inode_info *ci, int nr,
2141                                 struct ceph_snap_context *snapc)
2142 {
2143         struct inode *inode = &ci->vfs_inode;
2144         int last = 0;
2145         int last_snap = 0;
2146         int found = 0;
2147         struct ceph_cap_snap *capsnap = NULL;
2148
2149         spin_lock(&inode->i_lock);
2150         ci->i_wrbuffer_ref -= nr;
2151         last = !ci->i_wrbuffer_ref;
2152
2153         if (ci->i_head_snapc == snapc) {
2154                 ci->i_wrbuffer_ref_head -= nr;
2155                 if (!ci->i_wrbuffer_ref_head) {
2156                         ceph_put_snap_context(ci->i_head_snapc);
2157                         ci->i_head_snapc = NULL;
2158                 }
2159                 dout("put_wrbuffer_cap_refs on %p head %d/%d -> %d/%d %s\n",
2160                      inode,
2161                      ci->i_wrbuffer_ref+nr, ci->i_wrbuffer_ref_head+nr,
2162                      ci->i_wrbuffer_ref, ci->i_wrbuffer_ref_head,
2163                      last ? " LAST" : "");
2164         } else {
2165                 list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) {
2166                         if (capsnap->context == snapc) {
2167                                 found = 1;
2168                                 capsnap->dirty_pages -= nr;
2169                                 last_snap = !capsnap->dirty_pages;
2170                                 break;
2171                         }
2172                 }
2173                 BUG_ON(!found);
2174                 dout("put_wrbuffer_cap_refs on %p cap_snap %p "
2175                      " snap %lld %d/%d -> %d/%d %s%s\n",
2176                      inode, capsnap, capsnap->context->seq,
2177                      ci->i_wrbuffer_ref+nr, capsnap->dirty_pages + nr,
2178                      ci->i_wrbuffer_ref, capsnap->dirty_pages,
2179                      last ? " (wrbuffer last)" : "",
2180                      last_snap ? " (capsnap last)" : "");
2181         }
2182
2183         spin_unlock(&inode->i_lock);
2184
2185         if (last) {
2186                 ceph_check_caps(ci, CHECK_CAPS_AUTHONLY, NULL);
2187                 iput(inode);
2188         } else if (last_snap) {
2189                 ceph_flush_snaps(ci);
2190                 wake_up(&ci->i_cap_wq);
2191         }
2192 }
2193
2194 /*
2195  * Handle a cap GRANT message from the MDS.  (Note that a GRANT may
2196  * actually be a revocation if it specifies a smaller cap set.)
2197  *
2198  * caller holds s_mutex and i_lock, we drop both.
2199  *
2200  * return value:
2201  *  0 - ok
2202  *  1 - check_caps on auth cap only (writeback)
2203  *  2 - check_caps (ack revoke)
2204  */
2205 static void handle_cap_grant(struct inode *inode, struct ceph_mds_caps *grant,
2206                              struct ceph_mds_session *session,
2207                              struct ceph_cap *cap,
2208                              struct ceph_buffer *xattr_buf)
2209         __releases(inode->i_lock)
2210         __releases(session->s_mutex)
2211 {
2212         struct ceph_inode_info *ci = ceph_inode(inode);
2213         int mds = session->s_mds;
2214         int seq = le32_to_cpu(grant->seq);
2215         int newcaps = le32_to_cpu(grant->caps);
2216         int issued, implemented, used, wanted, dirty;
2217         u64 size = le64_to_cpu(grant->size);
2218         u64 max_size = le64_to_cpu(grant->max_size);
2219         struct timespec mtime, atime, ctime;
2220         int check_caps = 0;
2221         int wake = 0;
2222         int writeback = 0;
2223         int revoked_rdcache = 0;
2224         int queue_invalidate = 0;
2225
2226         dout("handle_cap_grant inode %p cap %p mds%d seq %d %s\n",
2227              inode, cap, mds, seq, ceph_cap_string(newcaps));
2228         dout(" size %llu max_size %llu, i_size %llu\n", size, max_size,
2229                 inode->i_size);
2230
2231         /*
2232          * If CACHE is being revoked, and we have no dirty buffers,
2233          * try to invalidate (once).  (If there are dirty buffers, we
2234          * will invalidate _after_ writeback.)
2235          */
2236         if (((cap->issued & ~newcaps) & CEPH_CAP_FILE_CACHE) &&
2237             !ci->i_wrbuffer_ref) {
2238                 if (try_nonblocking_invalidate(inode) == 0) {
2239                         revoked_rdcache = 1;
2240                 } else {
2241                         /* there were locked pages.. invalidate later
2242                            in a separate thread. */
2243                         if (ci->i_rdcache_revoking != ci->i_rdcache_gen) {
2244                                 queue_invalidate = 1;
2245                                 ci->i_rdcache_revoking = ci->i_rdcache_gen;
2246                         }
2247                 }
2248         }
2249
2250         /* side effects now are allowed */
2251
2252         issued = __ceph_caps_issued(ci, &implemented);
2253         issued |= implemented | __ceph_caps_dirty(ci);
2254
2255         cap->cap_gen = session->s_cap_gen;
2256
2257         __check_cap_issue(ci, cap, newcaps);
2258
2259         if ((issued & CEPH_CAP_AUTH_EXCL) == 0) {
2260                 inode->i_mode = le32_to_cpu(grant->mode);
2261                 inode->i_uid = le32_to_cpu(grant->uid);
2262                 inode->i_gid = le32_to_cpu(grant->gid);
2263                 dout("%p mode 0%o uid.gid %d.%d\n", inode, inode->i_mode,
2264                      inode->i_uid, inode->i_gid);
2265         }
2266
2267         if ((issued & CEPH_CAP_LINK_EXCL) == 0)
2268                 inode->i_nlink = le32_to_cpu(grant->nlink);
2269
2270         if ((issued & CEPH_CAP_XATTR_EXCL) == 0 && grant->xattr_len) {
2271                 int len = le32_to_cpu(grant->xattr_len);
2272                 u64 version = le64_to_cpu(grant->xattr_version);
2273
2274                 if (version > ci->i_xattrs.version) {
2275                         dout(" got new xattrs v%llu on %p len %d\n",
2276                              version, inode, len);
2277                         if (ci->i_xattrs.blob)
2278                                 ceph_buffer_put(ci->i_xattrs.blob);
2279                         ci->i_xattrs.blob = ceph_buffer_get(xattr_buf);
2280                         ci->i_xattrs.version = version;
2281                 }
2282         }
2283
2284         /* size/ctime/mtime/atime? */
2285         ceph_fill_file_size(inode, issued,
2286                             le32_to_cpu(grant->truncate_seq),
2287                             le64_to_cpu(grant->truncate_size), size);
2288         ceph_decode_timespec(&mtime, &grant->mtime);
2289         ceph_decode_timespec(&atime, &grant->atime);
2290         ceph_decode_timespec(&ctime, &grant->ctime);
2291         ceph_fill_file_time(inode, issued,
2292                             le32_to_cpu(grant->time_warp_seq), &ctime, &mtime,
2293                             &atime);
2294
2295         /* max size increase? */
2296         if (max_size != ci->i_max_size) {
2297                 dout("max_size %lld -> %llu\n", ci->i_max_size, max_size);
2298                 ci->i_max_size = max_size;
2299                 if (max_size >= ci->i_wanted_max_size) {
2300                         ci->i_wanted_max_size = 0;  /* reset */
2301                         ci->i_requested_max_size = 0;
2302                 }
2303                 wake = 1;
2304         }
2305
2306         /* check cap bits */
2307         wanted = __ceph_caps_wanted(ci);
2308         used = __ceph_caps_used(ci);
2309         dirty = __ceph_caps_dirty(ci);
2310         dout(" my wanted = %s, used = %s, dirty %s\n",
2311              ceph_cap_string(wanted),
2312              ceph_cap_string(used),
2313              ceph_cap_string(dirty));
2314         if (wanted != le32_to_cpu(grant->wanted)) {
2315                 dout("mds wanted %s -> %s\n",
2316                      ceph_cap_string(le32_to_cpu(grant->wanted)),
2317                      ceph_cap_string(wanted));
2318                 grant->wanted = cpu_to_le32(wanted);
2319         }
2320
2321         cap->seq = seq;
2322
2323         /* file layout may have changed */
2324         ci->i_layout = grant->layout;
2325
2326         /* revocation, grant, or no-op? */
2327         if (cap->issued & ~newcaps) {
2328                 dout("revocation: %s -> %s\n", ceph_cap_string(cap->issued),
2329                      ceph_cap_string(newcaps));
2330                 if ((used & ~newcaps) & CEPH_CAP_FILE_BUFFER)
2331                         writeback = 1; /* will delay ack */
2332                 else if (dirty & ~newcaps)
2333                         check_caps = 1;  /* initiate writeback in check_caps */
2334                 else if (((used & ~newcaps) & CEPH_CAP_FILE_CACHE) == 0 ||
2335                            revoked_rdcache)
2336                         check_caps = 2;     /* send revoke ack in check_caps */
2337                 cap->issued = newcaps;
2338                 cap->implemented |= newcaps;
2339         } else if (cap->issued == newcaps) {
2340                 dout("caps unchanged: %s -> %s\n",
2341                      ceph_cap_string(cap->issued), ceph_cap_string(newcaps));
2342         } else {
2343                 dout("grant: %s -> %s\n", ceph_cap_string(cap->issued),
2344                      ceph_cap_string(newcaps));
2345                 cap->issued = newcaps;
2346                 cap->implemented |= newcaps; /* add bits only, to
2347                                               * avoid stepping on a
2348                                               * pending revocation */
2349                 wake = 1;
2350         }
2351         BUG_ON(cap->issued & ~cap->implemented);
2352
2353         spin_unlock(&inode->i_lock);
2354         if (writeback)
2355                 /*
2356                  * queue inode for writeback: we can't actually call
2357                  * filemap_write_and_wait, etc. from message handler
2358                  * context.
2359                  */
2360                 ceph_queue_writeback(inode);
2361         if (queue_invalidate)
2362                 ceph_queue_invalidate(inode);
2363         if (wake)
2364                 wake_up(&ci->i_cap_wq);
2365
2366         if (check_caps == 1)
2367                 ceph_check_caps(ci, CHECK_CAPS_NODELAY|CHECK_CAPS_AUTHONLY,
2368                                 session);
2369         else if (check_caps == 2)
2370                 ceph_check_caps(ci, CHECK_CAPS_NODELAY, session);
2371         else
2372                 mutex_unlock(&session->s_mutex);
2373 }
2374
2375 /*
2376  * Handle FLUSH_ACK from MDS, indicating that metadata we sent to the
2377  * MDS has been safely committed.
2378  */
2379 static void handle_cap_flush_ack(struct inode *inode, u64 flush_tid,
2380                                  struct ceph_mds_caps *m,
2381                                  struct ceph_mds_session *session,
2382                                  struct ceph_cap *cap)
2383         __releases(inode->i_lock)
2384 {
2385         struct ceph_inode_info *ci = ceph_inode(inode);
2386         struct ceph_mds_client *mdsc = &ceph_client(inode->i_sb)->mdsc;
2387         unsigned seq = le32_to_cpu(m->seq);
2388         int dirty = le32_to_cpu(m->dirty);
2389         int cleaned = 0;
2390         int drop = 0;
2391         int i;
2392
2393         for (i = 0; i < CEPH_CAP_BITS; i++)
2394                 if ((dirty & (1 << i)) &&
2395                     flush_tid == ci->i_cap_flush_tid[i])
2396                         cleaned |= 1 << i;
2397
2398         dout("handle_cap_flush_ack inode %p mds%d seq %d on %s cleaned %s,"
2399              " flushing %s -> %s\n",
2400              inode, session->s_mds, seq, ceph_cap_string(dirty),
2401              ceph_cap_string(cleaned), ceph_cap_string(ci->i_flushing_caps),
2402              ceph_cap_string(ci->i_flushing_caps & ~cleaned));
2403
2404         if (ci->i_flushing_caps == (ci->i_flushing_caps & ~cleaned))
2405                 goto out;
2406
2407         ci->i_flushing_caps &= ~cleaned;
2408
2409         spin_lock(&mdsc->cap_dirty_lock);
2410         if (ci->i_flushing_caps == 0) {
2411                 list_del_init(&ci->i_flushing_item);
2412                 if (!list_empty(&session->s_cap_flushing))
2413                         dout(" mds%d still flushing cap on %p\n",
2414                              session->s_mds,
2415                              &list_entry(session->s_cap_flushing.next,
2416                                          struct ceph_inode_info,
2417                                          i_flushing_item)->vfs_inode);
2418                 mdsc->num_cap_flushing--;
2419                 wake_up(&mdsc->cap_flushing_wq);
2420                 dout(" inode %p now !flushing\n", inode);
2421
2422                 if (ci->i_dirty_caps == 0) {
2423                         dout(" inode %p now clean\n", inode);
2424                         BUG_ON(!list_empty(&ci->i_dirty_item));
2425                         drop = 1;
2426                 } else {
2427                         BUG_ON(list_empty(&ci->i_dirty_item));
2428                 }
2429         }
2430         spin_unlock(&mdsc->cap_dirty_lock);
2431         wake_up(&ci->i_cap_wq);
2432
2433 out:
2434         spin_unlock(&inode->i_lock);
2435         if (drop)
2436                 iput(inode);
2437 }
2438
2439 /*
2440  * Handle FLUSHSNAP_ACK.  MDS has flushed snap data to disk and we can
2441  * throw away our cap_snap.
2442  *
2443  * Caller hold s_mutex.
2444  */
2445 static void handle_cap_flushsnap_ack(struct inode *inode, u64 flush_tid,
2446                                      struct ceph_mds_caps *m,
2447                                      struct ceph_mds_session *session)
2448 {
2449         struct ceph_inode_info *ci = ceph_inode(inode);
2450         u64 follows = le64_to_cpu(m->snap_follows);
2451         struct ceph_cap_snap *capsnap;
2452         int drop = 0;
2453
2454         dout("handle_cap_flushsnap_ack inode %p ci %p mds%d follows %lld\n",
2455              inode, ci, session->s_mds, follows);
2456
2457         spin_lock(&inode->i_lock);
2458         list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) {
2459                 if (capsnap->follows == follows) {
2460                         if (capsnap->flush_tid != flush_tid) {
2461                                 dout(" cap_snap %p follows %lld tid %lld !="
2462                                      " %lld\n", capsnap, follows,
2463                                      flush_tid, capsnap->flush_tid);
2464                                 break;
2465                         }
2466                         WARN_ON(capsnap->dirty_pages || capsnap->writing);
2467                         dout(" removing cap_snap %p follows %lld\n",
2468                              capsnap, follows);
2469                         ceph_put_snap_context(capsnap->context);
2470                         list_del(&capsnap->ci_item);
2471                         list_del(&capsnap->flushing_item);
2472                         ceph_put_cap_snap(capsnap);
2473                         drop = 1;
2474                         break;
2475                 } else {
2476                         dout(" skipping cap_snap %p follows %lld\n",
2477                              capsnap, capsnap->follows);
2478                 }
2479         }
2480         spin_unlock(&inode->i_lock);
2481         if (drop)
2482                 iput(inode);
2483 }
2484
2485 /*
2486  * Handle TRUNC from MDS, indicating file truncation.
2487  *
2488  * caller hold s_mutex.
2489  */
2490 static void handle_cap_trunc(struct inode *inode,
2491                              struct ceph_mds_caps *trunc,
2492                              struct ceph_mds_session *session)
2493         __releases(inode->i_lock)
2494 {
2495         struct ceph_inode_info *ci = ceph_inode(inode);
2496         int mds = session->s_mds;
2497         int seq = le32_to_cpu(trunc->seq);
2498         u32 truncate_seq = le32_to_cpu(trunc->truncate_seq);
2499         u64 truncate_size = le64_to_cpu(trunc->truncate_size);
2500         u64 size = le64_to_cpu(trunc->size);
2501         int implemented = 0;
2502         int dirty = __ceph_caps_dirty(ci);
2503         int issued = __ceph_caps_issued(ceph_inode(inode), &implemented);
2504         int queue_trunc = 0;
2505
2506         issued |= implemented | dirty;
2507
2508         dout("handle_cap_trunc inode %p mds%d seq %d to %lld seq %d\n",
2509              inode, mds, seq, truncate_size, truncate_seq);
2510         queue_trunc = ceph_fill_file_size(inode, issued,
2511                                           truncate_seq, truncate_size, size);
2512         spin_unlock(&inode->i_lock);
2513
2514         if (queue_trunc)
2515                 ceph_queue_vmtruncate(inode);
2516 }
2517
2518 /*
2519  * Handle EXPORT from MDS.  Cap is being migrated _from_ this mds to a
2520  * different one.  If we are the most recent migration we've seen (as
2521  * indicated by mseq), make note of the migrating cap bits for the
2522  * duration (until we see the corresponding IMPORT).
2523  *
2524  * caller holds s_mutex
2525  */
2526 static void handle_cap_export(struct inode *inode, struct ceph_mds_caps *ex,
2527                               struct ceph_mds_session *session)
2528 {
2529         struct ceph_inode_info *ci = ceph_inode(inode);
2530         int mds = session->s_mds;
2531         unsigned mseq = le32_to_cpu(ex->migrate_seq);
2532         struct ceph_cap *cap = NULL, *t;
2533         struct rb_node *p;
2534         int remember = 1;
2535
2536         dout("handle_cap_export inode %p ci %p mds%d mseq %d\n",
2537              inode, ci, mds, mseq);
2538
2539         spin_lock(&inode->i_lock);
2540
2541         /* make sure we haven't seen a higher mseq */
2542         for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
2543                 t = rb_entry(p, struct ceph_cap, ci_node);
2544                 if (ceph_seq_cmp(t->mseq, mseq) > 0) {
2545                         dout(" higher mseq on cap from mds%d\n",
2546                              t->session->s_mds);
2547                         remember = 0;
2548                 }
2549                 if (t->session->s_mds == mds)
2550                         cap = t;
2551         }
2552
2553         if (cap) {
2554                 if (remember) {
2555                         /* make note */
2556                         ci->i_cap_exporting_mds = mds;
2557                         ci->i_cap_exporting_mseq = mseq;
2558                         ci->i_cap_exporting_issued = cap->issued;
2559                 }
2560                 __ceph_remove_cap(cap);
2561         }
2562         /* else, we already released it */
2563
2564         spin_unlock(&inode->i_lock);
2565 }
2566
2567 /*
2568  * Handle cap IMPORT.  If there are temp bits from an older EXPORT,
2569  * clean them up.
2570  *
2571  * caller holds s_mutex.
2572  */
2573 static void handle_cap_import(struct ceph_mds_client *mdsc,
2574                               struct inode *inode, struct ceph_mds_caps *im,
2575                               struct ceph_mds_session *session,
2576                               void *snaptrace, int snaptrace_len)
2577 {
2578         struct ceph_inode_info *ci = ceph_inode(inode);
2579         int mds = session->s_mds;
2580         unsigned issued = le32_to_cpu(im->caps);
2581         unsigned wanted = le32_to_cpu(im->wanted);
2582         unsigned seq = le32_to_cpu(im->seq);
2583         unsigned mseq = le32_to_cpu(im->migrate_seq);
2584         u64 realmino = le64_to_cpu(im->realm);
2585         u64 cap_id = le64_to_cpu(im->cap_id);
2586
2587         if (ci->i_cap_exporting_mds >= 0 &&
2588             ceph_seq_cmp(ci->i_cap_exporting_mseq, mseq) < 0) {
2589                 dout("handle_cap_import inode %p ci %p mds%d mseq %d"
2590                      " - cleared exporting from mds%d\n",
2591                      inode, ci, mds, mseq,
2592                      ci->i_cap_exporting_mds);
2593                 ci->i_cap_exporting_issued = 0;
2594                 ci->i_cap_exporting_mseq = 0;
2595                 ci->i_cap_exporting_mds = -1;
2596         } else {
2597                 dout("handle_cap_import inode %p ci %p mds%d mseq %d\n",
2598                      inode, ci, mds, mseq);
2599         }
2600
2601         down_write(&mdsc->snap_rwsem);
2602         ceph_update_snap_trace(mdsc, snaptrace, snaptrace+snaptrace_len,
2603                                false);
2604         downgrade_write(&mdsc->snap_rwsem);
2605         ceph_add_cap(inode, session, cap_id, -1,
2606                      issued, wanted, seq, mseq, realmino, CEPH_CAP_FLAG_AUTH,
2607                      NULL /* no caps context */);
2608         try_flush_caps(inode, session, NULL);
2609         up_read(&mdsc->snap_rwsem);
2610 }
2611
2612 /*
2613  * Handle a caps message from the MDS.
2614  *
2615  * Identify the appropriate session, inode, and call the right handler
2616  * based on the cap op.
2617  */
2618 void ceph_handle_caps(struct ceph_mds_session *session,
2619                       struct ceph_msg *msg)
2620 {
2621         struct ceph_mds_client *mdsc = session->s_mdsc;
2622         struct super_block *sb = mdsc->client->sb;
2623         struct inode *inode;
2624         struct ceph_cap *cap;
2625         struct ceph_mds_caps *h;
2626         int mds = session->s_mds;
2627         int op;
2628         u32 seq;
2629         struct ceph_vino vino;
2630         u64 cap_id;
2631         u64 size, max_size;
2632         u64 tid;
2633         void *snaptrace;
2634
2635         dout("handle_caps from mds%d\n", mds);
2636
2637         /* decode */
2638         tid = le64_to_cpu(msg->hdr.tid);
2639         if (msg->front.iov_len < sizeof(*h))
2640                 goto bad;
2641         h = msg->front.iov_base;
2642         snaptrace = h + 1;
2643         op = le32_to_cpu(h->op);
2644         vino.ino = le64_to_cpu(h->ino);
2645         vino.snap = CEPH_NOSNAP;
2646         cap_id = le64_to_cpu(h->cap_id);
2647         seq = le32_to_cpu(h->seq);
2648         size = le64_to_cpu(h->size);
2649         max_size = le64_to_cpu(h->max_size);
2650
2651         mutex_lock(&session->s_mutex);
2652         session->s_seq++;
2653         dout(" mds%d seq %lld cap seq %u\n", session->s_mds, session->s_seq,
2654              (unsigned)seq);
2655
2656         /* lookup ino */
2657         inode = ceph_find_inode(sb, vino);
2658         dout(" op %s ino %llx.%llx inode %p\n", ceph_cap_op_name(op), vino.ino,
2659              vino.snap, inode);
2660         if (!inode) {
2661                 dout(" i don't have ino %llx\n", vino.ino);
2662                 goto done;
2663         }
2664
2665         /* these will work even if we don't have a cap yet */
2666         switch (op) {
2667         case CEPH_CAP_OP_FLUSHSNAP_ACK:
2668                 handle_cap_flushsnap_ack(inode, tid, h, session);
2669                 goto done;
2670
2671         case CEPH_CAP_OP_EXPORT:
2672                 handle_cap_export(inode, h, session);
2673                 goto done;
2674
2675         case CEPH_CAP_OP_IMPORT:
2676                 handle_cap_import(mdsc, inode, h, session,
2677                                   snaptrace, le32_to_cpu(h->snap_trace_len));
2678                 ceph_check_caps(ceph_inode(inode), CHECK_CAPS_NODELAY,
2679                                 session);
2680                 goto done_unlocked;
2681         }
2682
2683         /* the rest require a cap */
2684         spin_lock(&inode->i_lock);
2685         cap = __get_cap_for_mds(ceph_inode(inode), mds);
2686         if (!cap) {
2687                 dout("no cap on %p ino %llx.%llx from mds%d, releasing\n",
2688                      inode, ceph_ino(inode), ceph_snap(inode), mds);
2689                 spin_unlock(&inode->i_lock);
2690                 goto done;
2691         }
2692
2693         /* note that each of these drops i_lock for us */
2694         switch (op) {
2695         case CEPH_CAP_OP_REVOKE:
2696         case CEPH_CAP_OP_GRANT:
2697                 handle_cap_grant(inode, h, session, cap, msg->middle);
2698                 goto done_unlocked;
2699
2700         case CEPH_CAP_OP_FLUSH_ACK:
2701                 handle_cap_flush_ack(inode, tid, h, session, cap);
2702                 break;
2703
2704         case CEPH_CAP_OP_TRUNC:
2705                 handle_cap_trunc(inode, h, session);
2706                 break;
2707
2708         default:
2709                 spin_unlock(&inode->i_lock);
2710                 pr_err("ceph_handle_caps: unknown cap op %d %s\n", op,
2711                        ceph_cap_op_name(op));
2712         }
2713
2714 done:
2715         mutex_unlock(&session->s_mutex);
2716 done_unlocked:
2717         if (inode)
2718                 iput(inode);
2719         return;
2720
2721 bad:
2722         pr_err("ceph_handle_caps: corrupt message\n");
2723         ceph_msg_dump(msg);
2724         return;
2725 }
2726
2727 /*
2728  * Delayed work handler to process end of delayed cap release LRU list.
2729  */
2730 void ceph_check_delayed_caps(struct ceph_mds_client *mdsc)
2731 {
2732         struct ceph_inode_info *ci;
2733         int flags = CHECK_CAPS_NODELAY;
2734
2735         dout("check_delayed_caps\n");
2736         while (1) {
2737                 spin_lock(&mdsc->cap_delay_lock);
2738                 if (list_empty(&mdsc->cap_delay_list))
2739                         break;
2740                 ci = list_first_entry(&mdsc->cap_delay_list,
2741                                       struct ceph_inode_info,
2742                                       i_cap_delay_list);
2743                 if ((ci->i_ceph_flags & CEPH_I_FLUSH) == 0 &&
2744                     time_before(jiffies, ci->i_hold_caps_max))
2745                         break;
2746                 list_del_init(&ci->i_cap_delay_list);
2747                 spin_unlock(&mdsc->cap_delay_lock);
2748                 dout("check_delayed_caps on %p\n", &ci->vfs_inode);
2749                 ceph_check_caps(ci, flags, NULL);
2750         }
2751         spin_unlock(&mdsc->cap_delay_lock);
2752 }
2753
2754 /*
2755  * Flush all dirty caps to the mds
2756  */
2757 void ceph_flush_dirty_caps(struct ceph_mds_client *mdsc)
2758 {
2759         struct ceph_inode_info *ci, *nci = NULL;
2760         struct inode *inode, *ninode = NULL;
2761         struct list_head *p, *n;
2762
2763         dout("flush_dirty_caps\n");
2764         spin_lock(&mdsc->cap_dirty_lock);
2765         list_for_each_safe(p, n, &mdsc->cap_dirty) {
2766                 if (nci) {
2767                         ci = nci;
2768                         inode = ninode;
2769                         ci->i_ceph_flags &= ~CEPH_I_NOFLUSH;
2770                         dout("flush_dirty_caps inode %p (was next inode)\n",
2771                              inode);
2772                 } else {
2773                         ci = list_entry(p, struct ceph_inode_info,
2774                                         i_dirty_item);
2775                         inode = igrab(&ci->vfs_inode);
2776                         BUG_ON(!inode);
2777                         dout("flush_dirty_caps inode %p\n", inode);
2778                 }
2779                 if (n != &mdsc->cap_dirty) {
2780                         nci = list_entry(n, struct ceph_inode_info,
2781                                          i_dirty_item);
2782                         ninode = igrab(&nci->vfs_inode);
2783                         BUG_ON(!ninode);
2784                         nci->i_ceph_flags |= CEPH_I_NOFLUSH;
2785                         dout("flush_dirty_caps next inode %p, noflush\n",
2786                              ninode);
2787                 } else {
2788                         nci = NULL;
2789                         ninode = NULL;
2790                 }
2791                 spin_unlock(&mdsc->cap_dirty_lock);
2792                 if (inode) {
2793                         ceph_check_caps(ci, CHECK_CAPS_NODELAY|CHECK_CAPS_FLUSH,
2794                                         NULL);
2795                         iput(inode);
2796                 }
2797                 spin_lock(&mdsc->cap_dirty_lock);
2798         }
2799         spin_unlock(&mdsc->cap_dirty_lock);
2800 }
2801
2802 /*
2803  * Drop open file reference.  If we were the last open file,
2804  * we may need to release capabilities to the MDS (or schedule
2805  * their delayed release).
2806  */
2807 void ceph_put_fmode(struct ceph_inode_info *ci, int fmode)
2808 {
2809         struct inode *inode = &ci->vfs_inode;
2810         int last = 0;
2811
2812         spin_lock(&inode->i_lock);
2813         dout("put_fmode %p fmode %d %d -> %d\n", inode, fmode,
2814              ci->i_nr_by_mode[fmode], ci->i_nr_by_mode[fmode]-1);
2815         BUG_ON(ci->i_nr_by_mode[fmode] == 0);
2816         if (--ci->i_nr_by_mode[fmode] == 0)
2817                 last++;
2818         spin_unlock(&inode->i_lock);
2819
2820         if (last && ci->i_vino.snap == CEPH_NOSNAP)
2821                 ceph_check_caps(ci, 0, NULL);
2822 }
2823
2824 /*
2825  * Helpers for embedding cap and dentry lease releases into mds
2826  * requests.
2827  *
2828  * @force is used by dentry_release (below) to force inclusion of a
2829  * record for the directory inode, even when there aren't any caps to
2830  * drop.
2831  */
2832 int ceph_encode_inode_release(void **p, struct inode *inode,
2833                               int mds, int drop, int unless, int force)
2834 {
2835         struct ceph_inode_info *ci = ceph_inode(inode);
2836         struct ceph_cap *cap;
2837         struct ceph_mds_request_release *rel = *p;
2838         int ret = 0;
2839         int used = 0;
2840
2841         spin_lock(&inode->i_lock);
2842         used = __ceph_caps_used(ci);
2843
2844         dout("encode_inode_release %p mds%d used %s drop %s unless %s\n", inode,
2845              mds, ceph_cap_string(used), ceph_cap_string(drop),
2846              ceph_cap_string(unless));
2847
2848         /* only drop unused caps */
2849         drop &= ~used;
2850
2851         cap = __get_cap_for_mds(ci, mds);
2852         if (cap && __cap_is_valid(cap)) {
2853                 if (force ||
2854                     ((cap->issued & drop) &&
2855                      (cap->issued & unless) == 0)) {
2856                         if ((cap->issued & drop) &&
2857                             (cap->issued & unless) == 0) {
2858                                 dout("encode_inode_release %p cap %p %s -> "
2859                                      "%s\n", inode, cap,
2860                                      ceph_cap_string(cap->issued),
2861                                      ceph_cap_string(cap->issued & ~drop));
2862                                 cap->issued &= ~drop;
2863                                 cap->implemented &= ~drop;
2864                                 if (ci->i_ceph_flags & CEPH_I_NODELAY) {
2865                                         int wanted = __ceph_caps_wanted(ci);
2866                                         dout("  wanted %s -> %s (act %s)\n",
2867                                              ceph_cap_string(cap->mds_wanted),
2868                                              ceph_cap_string(cap->mds_wanted &
2869                                                              ~wanted),
2870                                              ceph_cap_string(wanted));
2871                                         cap->mds_wanted &= wanted;
2872                                 }
2873                         } else {
2874                                 dout("encode_inode_release %p cap %p %s"
2875                                      " (force)\n", inode, cap,
2876                                      ceph_cap_string(cap->issued));
2877                         }
2878
2879                         rel->ino = cpu_to_le64(ceph_ino(inode));
2880                         rel->cap_id = cpu_to_le64(cap->cap_id);
2881                         rel->seq = cpu_to_le32(cap->seq);
2882                         rel->issue_seq = cpu_to_le32(cap->issue_seq),
2883                         rel->mseq = cpu_to_le32(cap->mseq);
2884                         rel->caps = cpu_to_le32(cap->issued);
2885                         rel->wanted = cpu_to_le32(cap->mds_wanted);
2886                         rel->dname_len = 0;
2887                         rel->dname_seq = 0;
2888                         *p += sizeof(*rel);
2889                         ret = 1;
2890                 } else {
2891                         dout("encode_inode_release %p cap %p %s\n",
2892                              inode, cap, ceph_cap_string(cap->issued));
2893                 }
2894         }
2895         spin_unlock(&inode->i_lock);
2896         return ret;
2897 }
2898
2899 int ceph_encode_dentry_release(void **p, struct dentry *dentry,
2900                                int mds, int drop, int unless)
2901 {
2902         struct inode *dir = dentry->d_parent->d_inode;
2903         struct ceph_mds_request_release *rel = *p;
2904         struct ceph_dentry_info *di = ceph_dentry(dentry);
2905         int force = 0;
2906         int ret;
2907
2908         /*
2909          * force an record for the directory caps if we have a dentry lease.
2910          * this is racy (can't take i_lock and d_lock together), but it
2911          * doesn't have to be perfect; the mds will revoke anything we don't
2912          * release.
2913          */
2914         spin_lock(&dentry->d_lock);
2915         if (di->lease_session && di->lease_session->s_mds == mds)
2916                 force = 1;
2917         spin_unlock(&dentry->d_lock);
2918
2919         ret = ceph_encode_inode_release(p, dir, mds, drop, unless, force);
2920
2921         spin_lock(&dentry->d_lock);
2922         if (ret && di->lease_session && di->lease_session->s_mds == mds) {
2923                 dout("encode_dentry_release %p mds%d seq %d\n",
2924                      dentry, mds, (int)di->lease_seq);
2925                 rel->dname_len = cpu_to_le32(dentry->d_name.len);
2926                 memcpy(*p, dentry->d_name.name, dentry->d_name.len);
2927                 *p += dentry->d_name.len;
2928                 rel->dname_seq = cpu_to_le32(di->lease_seq);
2929         }
2930         spin_unlock(&dentry->d_lock);
2931         return ret;
2932 }