1 // SPDX-License-Identifier: GPL-2.0
2 #include <linux/ceph/ceph_debug.h>
3
4 #include <linux/fs.h>
5 #include <linux/kernel.h>
6 #include <linux/sched/signal.h>
7 #include <linux/slab.h>
8 #include <linux/vmalloc.h>
9 #include <linux/wait.h>
10 #include <linux/writeback.h>
11 #include <linux/iversion.h>
12 #include <linux/filelock.h>
13
14 #include "super.h"
15 #include "mds_client.h"
16 #include "cache.h"
17 #include "crypto.h"
18 #include <linux/ceph/decode.h>
19 #include <linux/ceph/messenger.h>
20
21 /*
22 * Capability management
23 *
24 * The Ceph metadata servers control client access to inode metadata
25 * and file data by issuing capabilities, granting clients permission
26 * to read and/or write both inode field and file data to OSDs
27 * (storage nodes). Each capability consists of a set of bits
28 * indicating which operations are allowed.
29 *
30 * If the client holds a *_SHARED cap, the client has a coherent value
31 * that can be safely read from the cached inode.
32 *
33 * In the case of a *_EXCL (exclusive) or FILE_WR capabilities, the
34 * client is allowed to change inode attributes (e.g., file size,
35 * mtime), note its dirty state in the ceph_cap, and asynchronously
36 * flush that metadata change to the MDS.
37 *
38 * In the event of a conflicting operation (perhaps by another
39 * client), the MDS will revoke the conflicting client capabilities.
40 *
41 * In order for a client to cache an inode, it must hold a capability
42 * with at least one MDS server. When inodes are released, release
43 * notifications are batched and periodically sent en masse to the MDS
44 * cluster to release server state.
45 */
46
47 static u64 __get_oldest_flush_tid(struct ceph_mds_client *mdsc);
48 static void __kick_flushing_caps(struct ceph_mds_client *mdsc,
49 struct ceph_mds_session *session,
50 struct ceph_inode_info *ci,
51 u64 oldest_flush_tid);
52
53 /*
54 * Generate readable cap strings for debugging output.
55 */
56 #define MAX_CAP_STR 20
57 static char cap_str[MAX_CAP_STR][40];
58 static DEFINE_SPINLOCK(cap_str_lock);
59 static int last_cap_str;
60
61 static char *gcap_string(char *s, int c)
62 {
63 if (c & CEPH_CAP_GSHARED)
64 *s++ = 's';
65 if (c & CEPH_CAP_GEXCL)
66 *s++ = 'x';
67 if (c & CEPH_CAP_GCACHE)
68 *s++ = 'c';
69 if (c & CEPH_CAP_GRD)
70 *s++ = 'r';
71 if (c & CEPH_CAP_GWR)
72 *s++ = 'w';
73 if (c & CEPH_CAP_GBUFFER)
74 *s++ = 'b';
75 if (c & CEPH_CAP_GWREXTEND)
76 *s++ = 'a';
77 if (c & CEPH_CAP_GLAZYIO)
78 *s++ = 'l';
79 return s;
80 }
81
82 const char *ceph_cap_string(int caps)
83 {
84 int i;
85 char *s;
86 int c;
87
88 spin_lock(&cap_str_lock);
89 i = last_cap_str++;
90 if (last_cap_str == MAX_CAP_STR)
91 last_cap_str = 0;
92 spin_unlock(&cap_str_lock);
93
94 s = cap_str[i];
95
96 if (caps & CEPH_CAP_PIN)
97 *s++ = 'p';
98
99 c = (caps >> CEPH_CAP_SAUTH) & 3;
100 if (c) {
101 *s++ = 'A';
102 s = gcap_string(s, c);
103 }
104
105 c = (caps >> CEPH_CAP_SLINK) & 3;
106 if (c) {
107 *s++ = 'L';
108 s = gcap_string(s, c);
109 }
110
111 c = (caps >> CEPH_CAP_SXATTR) & 3;
112 if (c) {
113 *s++ = 'X';
114 s = gcap_string(s, c);
115 }
116
117 c = caps >> CEPH_CAP_SFILE;
118 if (c) {
119 *s++ = 'F';
120 s = gcap_string(s, c);
121 }
122
123 if (s == cap_str[i])
124 *s++ = '-';
125 *s = 0;
126 return cap_str[i];
127 }
128
129 void ceph_caps_init(struct ceph_mds_client *mdsc)
130 {
131 INIT_LIST_HEAD(&mdsc->caps_list);
132 spin_lock_init(&mdsc->caps_list_lock);
133 }
134
135 void ceph_caps_finalize(struct ceph_mds_client *mdsc)
136 {
137 struct ceph_cap *cap;
138
139 spin_lock(&mdsc->caps_list_lock);
140 while (!list_empty(&mdsc->caps_list)) {
141 cap = list_first_entry(&mdsc->caps_list,
142 struct ceph_cap, caps_item);
143 list_del(&cap->caps_item);
144 kmem_cache_free(ceph_cap_cachep, cap);
145 }
146 mdsc->caps_total_count = 0;
147 mdsc->caps_avail_count = 0;
148 mdsc->caps_use_count = 0;
149 mdsc->caps_reserve_count = 0;
150 mdsc->caps_min_count = 0;
151 spin_unlock(&mdsc->caps_list_lock);
152 }
153
154 void ceph_adjust_caps_max_min(struct ceph_mds_client *mdsc,
155 struct ceph_mount_options *fsopt)
156 {
157 spin_lock(&mdsc->caps_list_lock);
158 mdsc->caps_min_count = fsopt->max_readdir;
159 if (mdsc->caps_min_count < 1024)
160 mdsc->caps_min_count = 1024;
161 mdsc->caps_use_max = fsopt->caps_max;
162 if (mdsc->caps_use_max > 0 &&
163 mdsc->caps_use_max < mdsc->caps_min_count)
164 mdsc->caps_use_max = mdsc->caps_min_count;
165 spin_unlock(&mdsc->caps_list_lock);
166 }
167
168 static void __ceph_unreserve_caps(struct ceph_mds_client *mdsc, int nr_caps)
169 {
170 struct ceph_cap *cap;
171 int i;
172
173 if (nr_caps) {
174 BUG_ON(mdsc->caps_reserve_count < nr_caps);
175 mdsc->caps_reserve_count -= nr_caps;
176 if (mdsc->caps_avail_count >=
177 mdsc->caps_reserve_count + mdsc->caps_min_count) {
178 mdsc->caps_total_count -= nr_caps;
179 for (i = 0; i < nr_caps; i++) {
180 cap = list_first_entry(&mdsc->caps_list,
181 struct ceph_cap, caps_item);
182 list_del(&cap->caps_item);
183 kmem_cache_free(ceph_cap_cachep, cap);
184 }
185 } else {
186 mdsc->caps_avail_count += nr_caps;
187 }
188
189 doutc(mdsc->fsc->client,
190 "caps %d = %d used + %d resv + %d avail\n",
191 mdsc->caps_total_count, mdsc->caps_use_count,
192 mdsc->caps_reserve_count, mdsc->caps_avail_count);
193 BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
194 mdsc->caps_reserve_count +
195 mdsc->caps_avail_count);
196 }
197 }
198
199 /*
200 * Called under mdsc->mutex.
201 */
202 int ceph_reserve_caps(struct ceph_mds_client *mdsc,
203 struct ceph_cap_reservation *ctx, int need)
204 {
205 struct ceph_client *cl = mdsc->fsc->client;
206 int i, j;
207 struct ceph_cap *cap;
208 int have;
209 int alloc = 0;
210 int max_caps;
211 int err = 0;
212 bool trimmed = false;
213 struct ceph_mds_session *s;
214 LIST_HEAD(newcaps);
215
216 doutc(cl, "ctx=%p need=%d\n", ctx, need);
217
218 /* first reserve any caps that are already allocated */
219 spin_lock(&mdsc->caps_list_lock);
220 if (mdsc->caps_avail_count >= need)
221 have = need;
222 else
223 have = mdsc->caps_avail_count;
224 mdsc->caps_avail_count -= have;
225 mdsc->caps_reserve_count += have;
226 BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
227 mdsc->caps_reserve_count +
228 mdsc->caps_avail_count);
229 spin_unlock(&mdsc->caps_list_lock);
230
231 for (i = have; i < need; ) {
232 cap = kmem_cache_alloc(ceph_cap_cachep, GFP_NOFS);
233 if (cap) {
234 list_add(&cap->caps_item, &newcaps);
235 alloc++;
236 i++;
237 continue;
238 }
239
240 if (!trimmed) {
241 for (j = 0; j < mdsc->max_sessions; j++) {
242 s = __ceph_lookup_mds_session(mdsc, j);
243 if (!s)
244 continue;
245 mutex_unlock(&mdsc->mutex);
246
247 mutex_lock(&s->s_mutex);
248 max_caps = s->s_nr_caps - (need - i);
249 ceph_trim_caps(mdsc, s, max_caps);
250 mutex_unlock(&s->s_mutex);
251
252 ceph_put_mds_session(s);
253 mutex_lock(&mdsc->mutex);
254 }
255 trimmed = true;
256
257 spin_lock(&mdsc->caps_list_lock);
258 if (mdsc->caps_avail_count) {
259 int more_have;
260 if (mdsc->caps_avail_count >= need - i)
261 more_have = need - i;
262 else
263 more_have = mdsc->caps_avail_count;
264
265 i += more_have;
266 have += more_have;
267 mdsc->caps_avail_count -= more_have;
268 mdsc->caps_reserve_count += more_have;
269
270 }
271 spin_unlock(&mdsc->caps_list_lock);
272
273 continue;
274 }
275
276 pr_warn_client(cl, "ctx=%p ENOMEM need=%d got=%d\n", ctx, need,
277 have + alloc);
278 err = -ENOMEM;
279 break;
280 }
281
282 if (!err) {
283 BUG_ON(have + alloc != need);
284 ctx->count = need;
285 ctx->used = 0;
286 }
287
288 spin_lock(&mdsc->caps_list_lock);
289 mdsc->caps_total_count += alloc;
290 mdsc->caps_reserve_count += alloc;
291 list_splice(&newcaps, &mdsc->caps_list);
292
293 BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
294 mdsc->caps_reserve_count +
295 mdsc->caps_avail_count);
296
297 if (err)
298 __ceph_unreserve_caps(mdsc, have + alloc);
299
300 spin_unlock(&mdsc->caps_list_lock);
301
302 doutc(cl, "ctx=%p %d = %d used + %d resv + %d avail\n", ctx,
303 mdsc->caps_total_count, mdsc->caps_use_count,
304 mdsc->caps_reserve_count, mdsc->caps_avail_count);
305 return err;
306 }
307
308 void ceph_unreserve_caps(struct ceph_mds_client *mdsc,
309 struct ceph_cap_reservation *ctx)
310 {
311 struct ceph_client *cl = mdsc->fsc->client;
312 bool reclaim = false;
313 if (!ctx->count)
314 return;
315
316 doutc(cl, "ctx=%p count=%d\n", ctx, ctx->count);
317 spin_lock(&mdsc->caps_list_lock);
318 __ceph_unreserve_caps(mdsc, ctx->count);
319 ctx->count = 0;
320
321 if (mdsc->caps_use_max > 0 &&
322 mdsc->caps_use_count > mdsc->caps_use_max)
323 reclaim = true;
324 spin_unlock(&mdsc->caps_list_lock);
325
326 if (reclaim)
327 ceph_reclaim_caps_nr(mdsc, ctx->used);
328 }
329
330 struct ceph_cap *ceph_get_cap(struct ceph_mds_client *mdsc,
331 struct ceph_cap_reservation *ctx)
332 {
333 struct ceph_client *cl = mdsc->fsc->client;
334 struct ceph_cap *cap = NULL;
335
336 /* temporary, until we do something about cap import/export */
337 if (!ctx) {
338 cap = kmem_cache_alloc(ceph_cap_cachep, GFP_NOFS);
339 if (cap) {
340 spin_lock(&mdsc->caps_list_lock);
341 mdsc->caps_use_count++;
342 mdsc->caps_total_count++;
343 spin_unlock(&mdsc->caps_list_lock);
344 } else {
345 spin_lock(&mdsc->caps_list_lock);
346 if (mdsc->caps_avail_count) {
347 BUG_ON(list_empty(&mdsc->caps_list));
348
349 mdsc->caps_avail_count--;
350 mdsc->caps_use_count++;
351 cap = list_first_entry(&mdsc->caps_list,
352 struct ceph_cap, caps_item);
353 list_del(&cap->caps_item);
354
355 BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
356 mdsc->caps_reserve_count + mdsc->caps_avail_count);
357 }
358 spin_unlock(&mdsc->caps_list_lock);
359 }
360
361 return cap;
362 }
363
364 spin_lock(&mdsc->caps_list_lock);
365 doutc(cl, "ctx=%p (%d) %d = %d used + %d resv + %d avail\n", ctx,
366 ctx->count, mdsc->caps_total_count, mdsc->caps_use_count,
367 mdsc->caps_reserve_count, mdsc->caps_avail_count);
368 BUG_ON(!ctx->count);
369 BUG_ON(ctx->count > mdsc->caps_reserve_count);
370 BUG_ON(list_empty(&mdsc->caps_list));
371
372 ctx->count--;
373 ctx->used++;
374 mdsc->caps_reserve_count--;
375 mdsc->caps_use_count++;
376
377 cap = list_first_entry(&mdsc->caps_list, struct ceph_cap, caps_item);
378 list_del(&cap->caps_item);
379
380 BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
381 mdsc->caps_reserve_count + mdsc->caps_avail_count);
382 spin_unlock(&mdsc->caps_list_lock);
383 return cap;
384 }
385
386 void ceph_put_cap(struct ceph_mds_client *mdsc, struct ceph_cap *cap)
387 {
388 struct ceph_client *cl = mdsc->fsc->client;
389
390 spin_lock(&mdsc->caps_list_lock);
391 doutc(cl, "%p %d = %d used + %d resv + %d avail\n", cap,
392 mdsc->caps_total_count, mdsc->caps_use_count,
393 mdsc->caps_reserve_count, mdsc->caps_avail_count);
394 mdsc->caps_use_count--;
395 /*
396 * Keep some preallocated caps around (ceph_min_count), to
397 * avoid lots of free/alloc churn.
398 */
399 if (mdsc->caps_avail_count >= mdsc->caps_reserve_count +
400 mdsc->caps_min_count) {
401 mdsc->caps_total_count--;
402 kmem_cache_free(ceph_cap_cachep, cap);
403 } else {
404 mdsc->caps_avail_count++;
405 list_add(&cap->caps_item, &mdsc->caps_list);
406 }
407
408 BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
409 mdsc->caps_reserve_count + mdsc->caps_avail_count);
410 spin_unlock(&mdsc->caps_list_lock);
411 }
412
413 void ceph_reservation_status(struct ceph_fs_client *fsc,
414 int *total, int *avail, int *used, int *reserved,
415 int *min)
416 {
417 struct ceph_mds_client *mdsc = fsc->mdsc;
418
419 spin_lock(&mdsc->caps_list_lock);
420
421 if (total)
422 *total = mdsc->caps_total_count;
423 if (avail)
424 *avail = mdsc->caps_avail_count;
425 if (used)
426 *used = mdsc->caps_use_count;
427 if (reserved)
428 *reserved = mdsc->caps_reserve_count;
429 if (min)
430 *min = mdsc->caps_min_count;
431
432 spin_unlock(&mdsc->caps_list_lock);
433 }
434
435 /*
436 * Find ceph_cap for given mds, if any.
437 *
438 * Called with i_ceph_lock held.
439 */
440 struct ceph_cap *__get_cap_for_mds(struct ceph_inode_info *ci, int mds)
441 {
442 struct ceph_cap *cap;
443 struct rb_node *n = ci->i_caps.rb_node;
444
445 while (n) {
446 cap = rb_entry(n, struct ceph_cap, ci_node);
447 if (mds < cap->mds)
448 n = n->rb_left;
449 else if (mds > cap->mds)
450 n = n->rb_right;
451 else
452 return cap;
453 }
454 return NULL;
455 }
456
457 struct ceph_cap *ceph_get_cap_for_mds(struct ceph_inode_info *ci, int mds)
458 {
459 struct ceph_cap *cap;
460
461 spin_lock(&ci->i_ceph_lock);
462 cap = __get_cap_for_mds(ci, mds);
463 spin_unlock(&ci->i_ceph_lock);
464 return cap;
465 }
466
467 /*
468 * Called under i_ceph_lock.
469 */
470 static void __insert_cap_node(struct ceph_inode_info *ci,
471 struct ceph_cap *new)
472 {
473 struct rb_node **p = &ci->i_caps.rb_node;
474 struct rb_node *parent = NULL;
475 struct ceph_cap *cap = NULL;
476
477 while (*p) {
478 parent = *p;
479 cap = rb_entry(parent, struct ceph_cap, ci_node);
480 if (new->mds < cap->mds)
481 p = &(*p)->rb_left;
482 else if (new->mds > cap->mds)
483 p = &(*p)->rb_right;
484 else
485 BUG();
486 }
487
488 rb_link_node(&new->ci_node, parent, p);
489 rb_insert_color(&new->ci_node, &ci->i_caps);
490 }
491
492 /*
493 * (re)set cap hold timeouts, which control the delayed release
494 * of unused caps back to the MDS. Should be called on cap use.
495 */
496 static void __cap_set_timeouts(struct ceph_mds_client *mdsc,
497 struct ceph_inode_info *ci)
498 {
499 struct inode *inode = &ci->netfs.inode;
500 struct ceph_mount_options *opt = mdsc->fsc->mount_options;
501
502 ci->i_hold_caps_max = round_jiffies(jiffies +
503 opt->caps_wanted_delay_max * HZ);
504 doutc(mdsc->fsc->client, "%p %llx.%llx %lu\n", inode,
505 ceph_vinop(inode), ci->i_hold_caps_max - jiffies);
506 }
507
508 /*
509 * (Re)queue cap at the end of the delayed cap release list.
510 *
511 * If I_FLUSH is set, leave the inode at the front of the list.
512 *
513 * Caller holds i_ceph_lock
514 * -> we take mdsc->cap_delay_lock
515 */
516 static void __cap_delay_requeue(struct ceph_mds_client *mdsc,
517 struct ceph_inode_info *ci)
518 {
519 struct inode *inode = &ci->netfs.inode;
520
521 doutc(mdsc->fsc->client, "%p %llx.%llx flags 0x%lx at %lu\n",
522 inode, ceph_vinop(inode), ci->i_ceph_flags,
523 ci->i_hold_caps_max);
524 if (!mdsc->stopping) {
525 spin_lock(&mdsc->cap_delay_lock);
526 if (!list_empty(&ci->i_cap_delay_list)) {
527 if (ci->i_ceph_flags & CEPH_I_FLUSH)
528 goto no_change;
529 list_del_init(&ci->i_cap_delay_list);
530 }
531 __cap_set_timeouts(mdsc, ci);
532 list_add_tail(&ci->i_cap_delay_list, &mdsc->cap_delay_list);
533 no_change:
534 spin_unlock(&mdsc->cap_delay_lock);
535 }
536 }
537
538 /*
539 * Queue an inode for immediate writeback. Mark inode with I_FLUSH,
540 * indicating we should send a cap message to flush dirty metadata
541 * asap, and move to the front of the delayed cap list.
542 */
543 static void __cap_delay_requeue_front(struct ceph_mds_client *mdsc,
544 struct ceph_inode_info *ci)
545 {
546 struct inode *inode = &ci->netfs.inode;
547
548 doutc(mdsc->fsc->client, "%p %llx.%llx\n", inode, ceph_vinop(inode));
549 spin_lock(&mdsc->cap_delay_lock);
550 ci->i_ceph_flags |= CEPH_I_FLUSH;
551 if (!list_empty(&ci->i_cap_delay_list))
552 list_del_init(&ci->i_cap_delay_list);
553 list_add(&ci->i_cap_delay_list, &mdsc->cap_delay_list);
554 spin_unlock(&mdsc->cap_delay_lock);
555 }
556
557 /*
558 * Cancel delayed work on cap.
559 *
560 * Caller must hold i_ceph_lock.
561 */
562 static void __cap_delay_cancel(struct ceph_mds_client *mdsc,
563 struct ceph_inode_info *ci)
564 {
565 struct inode *inode = &ci->netfs.inode;
566
567 doutc(mdsc->fsc->client, "%p %llx.%llx\n", inode, ceph_vinop(inode));
568 if (list_empty(&ci->i_cap_delay_list))
569 return;
570 spin_lock(&mdsc->cap_delay_lock);
571 list_del_init(&ci->i_cap_delay_list);
572 spin_unlock(&mdsc->cap_delay_lock);
573 }
574
575 /* Common issue checks for add_cap, handle_cap_grant. */
576 static void __check_cap_issue(struct ceph_inode_info *ci, struct ceph_cap *cap,
577 unsigned issued)
578 {
579 struct inode *inode = &ci->netfs.inode;
580 struct ceph_client *cl = ceph_inode_to_client(inode);
581
582 unsigned had = __ceph_caps_issued(ci, NULL);
583
584 lockdep_assert_held(&ci->i_ceph_lock);
585
586 /*
587 * Each time we receive FILE_CACHE anew, we increment
588 * i_rdcache_gen.
589 */
590 if (S_ISREG(ci->netfs.inode.i_mode) &&
591 (issued & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) &&
592 (had & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) == 0) {
593 ci->i_rdcache_gen++;
594 }
595
596 /*
597 * If FILE_SHARED is newly issued, mark dir not complete. We don't
598 * know what happened to this directory while we didn't have the cap.
599 * If FILE_SHARED is being revoked, also mark dir not complete. It
600 * stops on-going cached readdir.
601 */
602 if ((issued & CEPH_CAP_FILE_SHARED) != (had & CEPH_CAP_FILE_SHARED)) {
603 if (issued & CEPH_CAP_FILE_SHARED)
604 atomic_inc(&ci->i_shared_gen);
605 if (S_ISDIR(ci->netfs.inode.i_mode)) {
606 doutc(cl, " marking %p NOT complete\n", inode);
607 __ceph_dir_clear_complete(ci);
608 }
609 }
610
611 /* Wipe saved layout if we're losing DIR_CREATE caps */
612 if (S_ISDIR(ci->netfs.inode.i_mode) && (had & CEPH_CAP_DIR_CREATE) &&
613 !(issued & CEPH_CAP_DIR_CREATE)) {
614 ceph_put_string(rcu_dereference_raw(ci->i_cached_layout.pool_ns));
615 memset(&ci->i_cached_layout, 0, sizeof(ci->i_cached_layout));
616 }
617 }
618
619 /**
620 * change_auth_cap_ses - move inode to appropriate lists when auth caps change
621 * @ci: inode to be moved
622 * @session: new auth caps session
623 */
624 void change_auth_cap_ses(struct ceph_inode_info *ci,
625 struct ceph_mds_session *session)
626 {
627 lockdep_assert_held(&ci->i_ceph_lock);
628
629 if (list_empty(&ci->i_dirty_item) && list_empty(&ci->i_flushing_item))
630 return;
631
632 spin_lock(&session->s_mdsc->cap_dirty_lock);
633 if (!list_empty(&ci->i_dirty_item))
634 list_move(&ci->i_dirty_item, &session->s_cap_dirty);
635 if (!list_empty(&ci->i_flushing_item))
636 list_move_tail(&ci->i_flushing_item, &session->s_cap_flushing);
637 spin_unlock(&session->s_mdsc->cap_dirty_lock);
638 }
639
640 /*
641 * Add a capability under the given MDS session.
642 *
643 * Caller should hold session snap_rwsem (read) and ci->i_ceph_lock
644 *
645 * @fmode is the open file mode, if we are opening a file, otherwise
646 * it is < 0. (This is so we can atomically add the cap and add an
647 * open file reference to it.)
648 */
649 void ceph_add_cap(struct inode *inode,
650 struct ceph_mds_session *session, u64 cap_id,
651 unsigned issued, unsigned wanted,
652 unsigned seq, unsigned mseq, u64 realmino, int flags,
653 struct ceph_cap **new_cap)
654 {
655 struct ceph_mds_client *mdsc = ceph_inode_to_fs_client(inode)->mdsc;
656 struct ceph_client *cl = ceph_inode_to_client(inode);
657 struct ceph_inode_info *ci = ceph_inode(inode);
658 struct ceph_cap *cap;
659 int mds = session->s_mds;
660 int actual_wanted;
661 u32 gen;
662
663 lockdep_assert_held(&ci->i_ceph_lock);
664
665 doutc(cl, "%p %llx.%llx mds%d cap %llx %s seq %d\n", inode,
666 ceph_vinop(inode), session->s_mds, cap_id,
667 ceph_cap_string(issued), seq);
668
669 gen = atomic_read(&session->s_cap_gen);
670
671 cap = __get_cap_for_mds(ci, mds);
672 if (!cap) {
673 cap = *new_cap;
674 *new_cap = NULL;
675
676 cap->issued = 0;
677 cap->implemented = 0;
678 cap->mds = mds;
679 cap->mds_wanted = 0;
680 cap->mseq = 0;
681
682 cap->ci = ci;
683 __insert_cap_node(ci, cap);
684
685 /* add to session cap list */
686 cap->session = session;
687 spin_lock(&session->s_cap_lock);
688 list_add_tail(&cap->session_caps, &session->s_caps);
689 session->s_nr_caps++;
690 atomic64_inc(&mdsc->metric.total_caps);
691 spin_unlock(&session->s_cap_lock);
692 } else {
693 spin_lock(&session->s_cap_lock);
694 list_move_tail(&cap->session_caps, &session->s_caps);
695 spin_unlock(&session->s_cap_lock);
696
697 if (cap->cap_gen < gen)
698 cap->issued = cap->implemented = CEPH_CAP_PIN;
699
700 /*
701 * auth mds of the inode changed. we received the cap export
702 * message, but still haven't received the cap import message.
703 * handle_cap_export() updated the new auth MDS' cap.
704 *
705 * "ceph_seq_cmp(seq, cap->seq) <= 0" means we are processing
706 * a message that was send before the cap import message. So
707 * don't remove caps.
708 */
709 if (ceph_seq_cmp(seq, cap->seq) <= 0) {
710 WARN_ON(cap != ci->i_auth_cap);
711 WARN_ON(cap->cap_id != cap_id);
712 seq = cap->seq;
713 mseq = cap->mseq;
714 issued |= cap->issued;
715 flags |= CEPH_CAP_FLAG_AUTH;
716 }
717 }
718
719 if (!ci->i_snap_realm ||
720 ((flags & CEPH_CAP_FLAG_AUTH) &&
721 realmino != (u64)-1 && ci->i_snap_realm->ino != realmino)) {
722 /*
723 * add this inode to the appropriate snap realm
724 */
725 struct ceph_snap_realm *realm = ceph_lookup_snap_realm(mdsc,
726 realmino);
727 if (realm)
728 ceph_change_snap_realm(inode, realm);
729 else
730 WARN(1, "%s: couldn't find snap realm 0x%llx (ino 0x%llx oldrealm 0x%llx)\n",
731 __func__, realmino, ci->i_vino.ino,
732 ci->i_snap_realm ? ci->i_snap_realm->ino : 0);
733 }
734
735 __check_cap_issue(ci, cap, issued);
736
737 /*
738 * If we are issued caps we don't want, or the mds' wanted
739 * value appears to be off, queue a check so we'll release
740 * later and/or update the mds wanted value.
741 */
742 actual_wanted = __ceph_caps_wanted(ci);
743 if ((wanted & ~actual_wanted) ||
744 (issued & ~actual_wanted & CEPH_CAP_ANY_WR)) {
745 doutc(cl, "issued %s, mds wanted %s, actual %s, queueing\n",
746 ceph_cap_string(issued), ceph_cap_string(wanted),
747 ceph_cap_string(actual_wanted));
748 __cap_delay_requeue(mdsc, ci);
749 }
750
751 if (flags & CEPH_CAP_FLAG_AUTH) {
752 if (!ci->i_auth_cap ||
753 ceph_seq_cmp(ci->i_auth_cap->mseq, mseq) < 0) {
754 if (ci->i_auth_cap &&
755 ci->i_auth_cap->session != cap->session)
756 change_auth_cap_ses(ci, cap->session);
757 ci->i_auth_cap = cap;
758 cap->mds_wanted = wanted;
759 }
760 } else {
761 WARN_ON(ci->i_auth_cap == cap);
762 }
763
764 doutc(cl, "inode %p %llx.%llx cap %p %s now %s seq %d mds%d\n",
765 inode, ceph_vinop(inode), cap, ceph_cap_string(issued),
766 ceph_cap_string(issued|cap->issued), seq, mds);
767 cap->cap_id = cap_id;
768 cap->issued = issued;
769 cap->implemented |= issued;
770 if (ceph_seq_cmp(mseq, cap->mseq) > 0)
771 cap->mds_wanted = wanted;
772 else
773 cap->mds_wanted |= wanted;
774 cap->seq = seq;
775 cap->issue_seq = seq;
776 cap->mseq = mseq;
777 cap->cap_gen = gen;
778 wake_up_all(&ci->i_cap_wq);
779 }
780
781 /*
782 * Return true if cap has not timed out and belongs to the current
783 * generation of the MDS session (i.e. has not gone 'stale' due to
784 * us losing touch with the mds).
785 */
786 static int __cap_is_valid(struct ceph_cap *cap)
787 {
788 struct inode *inode = &cap->ci->netfs.inode;
789 struct ceph_client *cl = cap->session->s_mdsc->fsc->client;
790 unsigned long ttl;
791 u32 gen;
792
793 gen = atomic_read(&cap->session->s_cap_gen);
794 ttl = cap->session->s_cap_ttl;
795
796 if (cap->cap_gen < gen || time_after_eq(jiffies, ttl)) {
797 doutc(cl, "%p %llx.%llx cap %p issued %s but STALE (gen %u vs %u)\n",
798 inode, ceph_vinop(inode), cap,
799 ceph_cap_string(cap->issued), cap->cap_gen, gen);
800 return 0;
801 }
802
803 return 1;
804 }
805
806 /*
807 * Return set of valid cap bits issued to us. Note that caps time
808 * out, and may be invalidated in bulk if the client session times out
809 * and session->s_cap_gen is bumped.
810 */
811 int __ceph_caps_issued(struct ceph_inode_info *ci, int *implemented)
812 {
813 struct inode *inode = &ci->netfs.inode;
814 struct ceph_client *cl = ceph_inode_to_client(inode);
815 int have = ci->i_snap_caps;
816 struct ceph_cap *cap;
817 struct rb_node *p;
818
819 if (implemented)
820 *implemented = 0;
821 for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
822 cap = rb_entry(p, struct ceph_cap, ci_node);
823 if (!__cap_is_valid(cap))
824 continue;
825 doutc(cl, "%p %llx.%llx cap %p issued %s\n", inode,
826 ceph_vinop(inode), cap, ceph_cap_string(cap->issued));
827 have |= cap->issued;
828 if (implemented)
829 *implemented |= cap->implemented;
830 }
831 /*
832 * exclude caps issued by non-auth MDS, but are been revoking
833 * by the auth MDS. The non-auth MDS should be revoking/exporting
834 * these caps, but the message is delayed.
835 */
836 if (ci->i_auth_cap) {
837 cap = ci->i_auth_cap;
838 have &= ~cap->implemented | cap->issued;
839 }
840 return have;
841 }
842
843 /*
844 * Get cap bits issued by caps other than @ocap
845 */
846 int __ceph_caps_issued_other(struct ceph_inode_info *ci, struct ceph_cap *ocap)
847 {
848 int have = ci->i_snap_caps;
849 struct ceph_cap *cap;
850 struct rb_node *p;
851
852 for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
853 cap = rb_entry(p, struct ceph_cap, ci_node);
854 if (cap == ocap)
855 continue;
856 if (!__cap_is_valid(cap))
857 continue;
858 have |= cap->issued;
859 }
860 return have;
861 }
862
863 /*
864 * Move a cap to the end of the LRU (oldest caps at list head, newest
865 * at list tail).
866 */
867 static void __touch_cap(struct ceph_cap *cap)
868 {
869 struct inode *inode = &cap->ci->netfs.inode;
870 struct ceph_mds_session *s = cap->session;
871 struct ceph_client *cl = s->s_mdsc->fsc->client;
872
873 spin_lock(&s->s_cap_lock);
874 if (!s->s_cap_iterator) {
875 doutc(cl, "%p %llx.%llx cap %p mds%d\n", inode,
876 ceph_vinop(inode), cap, s->s_mds);
877 list_move_tail(&cap->session_caps, &s->s_caps);
878 } else {
879 doutc(cl, "%p %llx.%llx cap %p mds%d NOP, iterating over caps\n",
880 inode, ceph_vinop(inode), cap, s->s_mds);
881 }
882 spin_unlock(&s->s_cap_lock);
883 }
884
885 /*
886 * Check if we hold the given mask. If so, move the cap(s) to the
887 * front of their respective LRUs. (This is the preferred way for
888 * callers to check for caps they want.)
889 */
890 int __ceph_caps_issued_mask(struct ceph_inode_info *ci, int mask, int touch)
891 {
892 struct inode *inode = &ci->netfs.inode;
893 struct ceph_client *cl = ceph_inode_to_client(inode);
894 struct ceph_cap *cap;
895 struct rb_node *p;
896 int have = ci->i_snap_caps;
897
898 if ((have & mask) == mask) {
899 doutc(cl, "mask %p %llx.%llx snap issued %s (mask %s)\n",
900 inode, ceph_vinop(inode), ceph_cap_string(have),
901 ceph_cap_string(mask));
902 return 1;
903 }
904
905 for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
906 cap = rb_entry(p, struct ceph_cap, ci_node);
907 if (!__cap_is_valid(cap))
908 continue;
909 if ((cap->issued & mask) == mask) {
910 doutc(cl, "mask %p %llx.%llx cap %p issued %s (mask %s)\n",
911 inode, ceph_vinop(inode), cap,
912 ceph_cap_string(cap->issued),
913 ceph_cap_string(mask));
914 if (touch)
915 __touch_cap(cap);
916 return 1;
917 }
918
919 /* does a combination of caps satisfy mask? */
920 have |= cap->issued;
921 if ((have & mask) == mask) {
922 doutc(cl, "mask %p %llx.%llx combo issued %s (mask %s)\n",
923 inode, ceph_vinop(inode),
924 ceph_cap_string(cap->issued),
925 ceph_cap_string(mask));
926 if (touch) {
927 struct rb_node *q;
928
929 /* touch this + preceding caps */
930 __touch_cap(cap);
931 for (q = rb_first(&ci->i_caps); q != p;
932 q = rb_next(q)) {
933 cap = rb_entry(q, struct ceph_cap,
934 ci_node);
935 if (!__cap_is_valid(cap))
936 continue;
937 if (cap->issued & mask)
938 __touch_cap(cap);
939 }
940 }
941 return 1;
942 }
943 }
944
945 return 0;
946 }
947
948 int __ceph_caps_issued_mask_metric(struct ceph_inode_info *ci, int mask,
949 int touch)
950 {
951 struct ceph_fs_client *fsc = ceph_sb_to_fs_client(ci->netfs.inode.i_sb);
952 int r;
953
954 r = __ceph_caps_issued_mask(ci, mask, touch);
955 if (r)
956 ceph_update_cap_hit(&fsc->mdsc->metric);
957 else
958 ceph_update_cap_mis(&fsc->mdsc->metric);
959 return r;
960 }
961
962 /*
963 * Return true if mask caps are currently being revoked by an MDS.
964 */
965 int __ceph_caps_revoking_other(struct ceph_inode_info *ci,
966 struct ceph_cap *ocap, int mask)
967 {
968 struct ceph_cap *cap;
969 struct rb_node *p;
970
971 for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
972 cap = rb_entry(p, struct ceph_cap, ci_node);
973 if (cap != ocap &&
974 (cap->implemented & ~cap->issued & mask))
975 return 1;
976 }
977 return 0;
978 }
979
980 int ceph_caps_revoking(struct ceph_inode_info *ci, int mask)
981 {
982 struct inode *inode = &ci->netfs.inode;
983 struct ceph_client *cl = ceph_inode_to_client(inode);
984 int ret;
985
986 spin_lock(&ci->i_ceph_lock);
987 ret = __ceph_caps_revoking_other(ci, NULL, mask);
988 spin_unlock(&ci->i_ceph_lock);
989 doutc(cl, "%p %llx.%llx %s = %d\n", inode, ceph_vinop(inode),
990 ceph_cap_string(mask), ret);
991 return ret;
992 }
993
994 int __ceph_caps_used(struct ceph_inode_info *ci)
995 {
996 int used = 0;
997 if (ci->i_pin_ref)
998 used |= CEPH_CAP_PIN;
999 if (ci->i_rd_ref)
1000 used |= CEPH_CAP_FILE_RD;
1001 if (ci->i_rdcache_ref ||
1002 (S_ISREG(ci->netfs.inode.i_mode) &&
1003 ci->netfs.inode.i_data.nrpages))
1004 used |= CEPH_CAP_FILE_CACHE;
1005 if (ci->i_wr_ref)
1006 used |= CEPH_CAP_FILE_WR;
1007 if (ci->i_wb_ref || ci->i_wrbuffer_ref)
1008 used |= CEPH_CAP_FILE_BUFFER;
1009 if (ci->i_fx_ref)
1010 used |= CEPH_CAP_FILE_EXCL;
1011 return used;
1012 }
1013
1014 #define FMODE_WAIT_BIAS 1000
1015
1016 /*
1017 * wanted, by virtue of open file modes
1018 */
1019 int __ceph_caps_file_wanted(struct ceph_inode_info *ci)
1020 {
1021 const int PIN_SHIFT = ffs(CEPH_FILE_MODE_PIN);
1022 const int RD_SHIFT = ffs(CEPH_FILE_MODE_RD);
1023 const int WR_SHIFT = ffs(CEPH_FILE_MODE_WR);
1024 const int LAZY_SHIFT = ffs(CEPH_FILE_MODE_LAZY);
1025 struct ceph_mount_options *opt =
1026 ceph_inode_to_fs_client(&ci->netfs.inode)->mount_options;
1027 unsigned long used_cutoff = jiffies - opt->caps_wanted_delay_max * HZ;
1028 unsigned long idle_cutoff = jiffies - opt->caps_wanted_delay_min * HZ;
1029
1030 if (S_ISDIR(ci->netfs.inode.i_mode)) {
1031 int want = 0;
1032
1033 /* use used_cutoff here, to keep dir's wanted caps longer */
1034 if (ci->i_nr_by_mode[RD_SHIFT] > 0 ||
1035 time_after(ci->i_last_rd, used_cutoff))
1036 want |= CEPH_CAP_ANY_SHARED;
1037
1038 if (ci->i_nr_by_mode[WR_SHIFT] > 0 ||
1039 time_after(ci->i_last_wr, used_cutoff)) {
1040 want |= CEPH_CAP_ANY_SHARED | CEPH_CAP_FILE_EXCL;
1041 if (opt->flags & CEPH_MOUNT_OPT_ASYNC_DIROPS)
1042 want |= CEPH_CAP_ANY_DIR_OPS;
1043 }
1044
1045 if (want || ci->i_nr_by_mode[PIN_SHIFT] > 0)
1046 want |= CEPH_CAP_PIN;
1047
1048 return want;
1049 } else {
1050 int bits = 0;
1051
1052 if (ci->i_nr_by_mode[RD_SHIFT] > 0) {
1053 if (ci->i_nr_by_mode[RD_SHIFT] >= FMODE_WAIT_BIAS ||
1054 time_after(ci->i_last_rd, used_cutoff))
1055 bits |= 1 << RD_SHIFT;
1056 } else if (time_after(ci->i_last_rd, idle_cutoff)) {
1057 bits |= 1 << RD_SHIFT;
1058 }
1059
1060 if (ci->i_nr_by_mode[WR_SHIFT] > 0) {
1061 if (ci->i_nr_by_mode[WR_SHIFT] >= FMODE_WAIT_BIAS ||
1062 time_after(ci->i_last_wr, used_cutoff))
1063 bits |= 1 << WR_SHIFT;
1064 } else if (time_after(ci->i_last_wr, idle_cutoff)) {
1065 bits |= 1 << WR_SHIFT;
1066 }
1067
1068 /* check lazyio only when read/write is wanted */
1069 if ((bits & (CEPH_FILE_MODE_RDWR << 1)) &&
1070 ci->i_nr_by_mode[LAZY_SHIFT] > 0)
1071 bits |= 1 << LAZY_SHIFT;
1072
1073 return bits ? ceph_caps_for_mode(bits >> 1) : 0;
1074 }
1075 }
1076
1077 /*
1078 * wanted, by virtue of open file modes AND cap refs (buffered/cached data)
1079 */
1080 int __ceph_caps_wanted(struct ceph_inode_info *ci)
1081 {
1082 int w = __ceph_caps_file_wanted(ci) | __ceph_caps_used(ci);
1083 if (S_ISDIR(ci->netfs.inode.i_mode)) {
1084 /* we want EXCL if holding caps of dir ops */
1085 if (w & CEPH_CAP_ANY_DIR_OPS)
1086 w |= CEPH_CAP_FILE_EXCL;
1087 } else {
1088 /* we want EXCL if dirty data */
1089 if (w & CEPH_CAP_FILE_BUFFER)
1090 w |= CEPH_CAP_FILE_EXCL;
1091 }
1092 return w;
1093 }
1094
1095 /*
1096 * Return caps we have registered with the MDS(s) as 'wanted'.
1097 */
1098 int __ceph_caps_mds_wanted(struct ceph_inode_info *ci, bool check)
1099 {
1100 struct ceph_cap *cap;
1101 struct rb_node *p;
1102 int mds_wanted = 0;
1103
1104 for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
1105 cap = rb_entry(p, struct ceph_cap, ci_node);
1106 if (check && !__cap_is_valid(cap))
1107 continue;
1108 if (cap == ci->i_auth_cap)
1109 mds_wanted |= cap->mds_wanted;
1110 else
1111 mds_wanted |= (cap->mds_wanted & ~CEPH_CAP_ANY_FILE_WR);
1112 }
1113 return mds_wanted;
1114 }
1115
1116 int ceph_is_any_caps(struct inode *inode)
1117 {
1118 struct ceph_inode_info *ci = ceph_inode(inode);
1119 int ret;
1120
1121 spin_lock(&ci->i_ceph_lock);
1122 ret = __ceph_is_any_real_caps(ci);
1123 spin_unlock(&ci->i_ceph_lock);
1124
1125 return ret;
1126 }
1127
1128 /*
1129 * Remove a cap. Take steps to deal with a racing iterate_session_caps.
1130 *
1131 * caller should hold i_ceph_lock.
1132 * caller will not hold session s_mutex if called from destroy_inode.
1133 */
1134 void __ceph_remove_cap(struct ceph_cap *cap, bool queue_release)
1135 {
1136 struct ceph_mds_session *session = cap->session;
1137 struct ceph_client *cl = session->s_mdsc->fsc->client;
1138 struct ceph_inode_info *ci = cap->ci;
1139 struct inode *inode = &ci->netfs.inode;
1140 struct ceph_mds_client *mdsc;
1141 int removed = 0;
1142
1143 /* 'ci' being NULL means the remove have already occurred */
1144 if (!ci) {
1145 doutc(cl, "inode is NULL\n");
1146 return;
1147 }
1148
1149 lockdep_assert_held(&ci->i_ceph_lock);
1150
1151 doutc(cl, "%p from %p %llx.%llx\n", cap, inode, ceph_vinop(inode));
1152
1153 mdsc = ceph_inode_to_fs_client(&ci->netfs.inode)->mdsc;
1154
1155 /* remove from inode's cap rbtree, and clear auth cap */
1156 rb_erase(&cap->ci_node, &ci->i_caps);
1157 if (ci->i_auth_cap == cap)
1158 ci->i_auth_cap = NULL;
1159
1160 /* remove from session list */
1161 spin_lock(&session->s_cap_lock);
1162 if (session->s_cap_iterator == cap) {
1163 /* not yet, we are iterating over this very cap */
1164 doutc(cl, "delaying %p removal from session %p\n", cap,
1165 cap->session);
1166 } else {
1167 list_del_init(&cap->session_caps);
1168 session->s_nr_caps--;
1169 atomic64_dec(&mdsc->metric.total_caps);
1170 cap->session = NULL;
1171 removed = 1;
1172 }
1173 /* protect backpointer with s_cap_lock: see iterate_session_caps */
1174 cap->ci = NULL;
1175
1176 /*
1177 * s_cap_reconnect is protected by s_cap_lock. no one changes
1178 * s_cap_gen while session is in the reconnect state.
1179 */
1180 if (queue_release &&
1181 (!session->s_cap_reconnect ||
1182 cap->cap_gen == atomic_read(&session->s_cap_gen))) {
1183 cap->queue_release = 1;
1184 if (removed) {
1185 __ceph_queue_cap_release(session, cap);
1186 removed = 0;
1187 }
1188 } else {
1189 cap->queue_release = 0;
1190 }
1191 cap->cap_ino = ci->i_vino.ino;
1192
1193 spin_unlock(&session->s_cap_lock);
1194
1195 if (removed)
1196 ceph_put_cap(mdsc, cap);
1197
1198 if (!__ceph_is_any_real_caps(ci)) {
1199 /* when reconnect denied, we remove session caps forcibly,
1200 * i_wr_ref can be non-zero. If there are ongoing write,
1201 * keep i_snap_realm.
1202 */
1203 if (ci->i_wr_ref == 0 && ci->i_snap_realm)
1204 ceph_change_snap_realm(&ci->netfs.inode, NULL);
1205
1206 __cap_delay_cancel(mdsc, ci);
1207 }
1208 }
1209
1210 void ceph_remove_cap(struct ceph_mds_client *mdsc, struct ceph_cap *cap,
1211 bool queue_release)
1212 {
1213 struct ceph_inode_info *ci = cap->ci;
1214 struct ceph_fs_client *fsc;
1215
1216 /* 'ci' being NULL means the remove have already occurred */
1217 if (!ci) {
1218 doutc(mdsc->fsc->client, "inode is NULL\n");
1219 return;
1220 }
1221
1222 lockdep_assert_held(&ci->i_ceph_lock);
1223
1224 fsc = ceph_inode_to_fs_client(&ci->netfs.inode);
1225 WARN_ON_ONCE(ci->i_auth_cap == cap &&
1226 !list_empty(&ci->i_dirty_item) &&
1227 !fsc->blocklisted &&
1228 !ceph_inode_is_shutdown(&ci->netfs.inode));
1229
1230 __ceph_remove_cap(cap, queue_release);
1231 }
1232
1233 struct cap_msg_args {
1234 struct ceph_mds_session *session;
1235 u64 ino, cid, follows;
1236 u64 flush_tid, oldest_flush_tid, size, max_size;
1237 u64 xattr_version;
1238 u64 change_attr;
1239 struct ceph_buffer *xattr_buf;
1240 struct ceph_buffer *old_xattr_buf;
1241 struct timespec64 atime, mtime, ctime, btime;
1242 int op, caps, wanted, dirty;
1243 u32 seq, issue_seq, mseq, time_warp_seq;
1244 u32 flags;
1245 kuid_t uid;
1246 kgid_t gid;
1247 umode_t mode;
1248 bool inline_data;
1249 bool wake;
1250 bool encrypted;
1251 u32 fscrypt_auth_len;
1252 u8 fscrypt_auth[sizeof(struct ceph_fscrypt_auth)]; // for context
1253 };
1254
1255 /* Marshal up the cap msg to the MDS */
1256 static void encode_cap_msg(struct ceph_msg *msg, struct cap_msg_args *arg)
1257 {
1258 struct ceph_mds_caps *fc;
1259 void *p;
1260 struct ceph_mds_client *mdsc = arg->session->s_mdsc;
1261 struct ceph_osd_client *osdc = &mdsc->fsc->client->osdc;
1262
1263 doutc(mdsc->fsc->client,
1264 "%s %llx %llx caps %s wanted %s dirty %s seq %u/%u"
1265 " tid %llu/%llu mseq %u follows %lld size %llu/%llu"
1266 " xattr_ver %llu xattr_len %d\n",
1267 ceph_cap_op_name(arg->op), arg->cid, arg->ino,
1268 ceph_cap_string(arg->caps), ceph_cap_string(arg->wanted),
1269 ceph_cap_string(arg->dirty), arg->seq, arg->issue_seq,
1270 arg->flush_tid, arg->oldest_flush_tid, arg->mseq, arg->follows,
1271 arg->size, arg->max_size, arg->xattr_version,
1272 arg->xattr_buf ? (int)arg->xattr_buf->vec.iov_len : 0);
1273
1274 msg->hdr.version = cpu_to_le16(12);
1275 msg->hdr.tid = cpu_to_le64(arg->flush_tid);
1276
1277 fc = msg->front.iov_base;
1278 memset(fc, 0, sizeof(*fc));
1279
1280 fc->cap_id = cpu_to_le64(arg->cid);
1281 fc->op = cpu_to_le32(arg->op);
1282 fc->seq = cpu_to_le32(arg->seq);
1283 fc->issue_seq = cpu_to_le32(arg->issue_seq);
1284 fc->migrate_seq = cpu_to_le32(arg->mseq);
1285 fc->caps = cpu_to_le32(arg->caps);
1286 fc->wanted = cpu_to_le32(arg->wanted);
1287 fc->dirty = cpu_to_le32(arg->dirty);
1288 fc->ino = cpu_to_le64(arg->ino);
1289 fc->snap_follows = cpu_to_le64(arg->follows);
1290
1291 #if IS_ENABLED(CONFIG_FS_ENCRYPTION)
1292 if (arg->encrypted)
1293 fc->size = cpu_to_le64(round_up(arg->size,
1294 CEPH_FSCRYPT_BLOCK_SIZE));
1295 else
1296 #endif
1297 fc->size = cpu_to_le64(arg->size);
1298 fc->max_size = cpu_to_le64(arg->max_size);
1299 ceph_encode_timespec64(&fc->mtime, &arg->mtime);
1300 ceph_encode_timespec64(&fc->atime, &arg->atime);
1301 ceph_encode_timespec64(&fc->ctime, &arg->ctime);
1302 fc->time_warp_seq = cpu_to_le32(arg->time_warp_seq);
1303
1304 fc->uid = cpu_to_le32(from_kuid(&init_user_ns, arg->uid));
1305 fc->gid = cpu_to_le32(from_kgid(&init_user_ns, arg->gid));
1306 fc->mode = cpu_to_le32(arg->mode);
1307
1308 fc->xattr_version = cpu_to_le64(arg->xattr_version);
1309 if (arg->xattr_buf) {
1310 msg->middle = ceph_buffer_get(arg->xattr_buf);
1311 fc->xattr_len = cpu_to_le32(arg->xattr_buf->vec.iov_len);
1312 msg->hdr.middle_len = cpu_to_le32(arg->xattr_buf->vec.iov_len);
1313 }
1314
1315 p = fc + 1;
1316 /* flock buffer size (version 2) */
1317 ceph_encode_32(&p, 0);
1318 /* inline version (version 4) */
1319 ceph_encode_64(&p, arg->inline_data ? 0 : CEPH_INLINE_NONE);
1320 /* inline data size */
1321 ceph_encode_32(&p, 0);
1322 /*
1323 * osd_epoch_barrier (version 5)
1324 * The epoch_barrier is protected osdc->lock, so READ_ONCE here in
1325 * case it was recently changed
1326 */
1327 ceph_encode_32(&p, READ_ONCE(osdc->epoch_barrier));
1328 /* oldest_flush_tid (version 6) */
1329 ceph_encode_64(&p, arg->oldest_flush_tid);
1330
1331 /*
1332 * caller_uid/caller_gid (version 7)
1333 *
1334 * Currently, we don't properly track which caller dirtied the caps
1335 * last, and force a flush of them when there is a conflict. For now,
1336 * just set this to 0:0, to emulate how the MDS has worked up to now.
1337 */
1338 ceph_encode_32(&p, 0);
1339 ceph_encode_32(&p, 0);
1340
1341 /* pool namespace (version 8) (mds always ignores this) */
1342 ceph_encode_32(&p, 0);
1343
1344 /* btime and change_attr (version 9) */
1345 ceph_encode_timespec64(p, &arg->btime);
1346 p += sizeof(struct ceph_timespec);
1347 ceph_encode_64(&p, arg->change_attr);
1348
1349 /* Advisory flags (version 10) */
1350 ceph_encode_32(&p, arg->flags);
1351
1352 /* dirstats (version 11) - these are r/o on the client */
1353 ceph_encode_64(&p, 0);
1354 ceph_encode_64(&p, 0);
1355
1356 #if IS_ENABLED(CONFIG_FS_ENCRYPTION)
1357 /*
1358 * fscrypt_auth and fscrypt_file (version 12)
1359 *
1360 * fscrypt_auth holds the crypto context (if any). fscrypt_file
1361 * tracks the real i_size as an __le64 field (and we use a rounded-up
1362 * i_size in the traditional size field).
1363 */
1364 ceph_encode_32(&p, arg->fscrypt_auth_len);
1365 ceph_encode_copy(&p, arg->fscrypt_auth, arg->fscrypt_auth_len);
1366 ceph_encode_32(&p, sizeof(__le64));
1367 ceph_encode_64(&p, arg->size);
1368 #else /* CONFIG_FS_ENCRYPTION */
1369 ceph_encode_32(&p, 0);
1370 ceph_encode_32(&p, 0);
1371 #endif /* CONFIG_FS_ENCRYPTION */
1372 }
1373
1374 /*
1375 * Queue cap releases when an inode is dropped from our cache.
1376 */
1377 void __ceph_remove_caps(struct ceph_inode_info *ci)
1378 {
1379 struct inode *inode = &ci->netfs.inode;
1380 struct ceph_mds_client *mdsc = ceph_inode_to_fs_client(inode)->mdsc;
1381 struct rb_node *p;
1382
1383 /* lock i_ceph_lock, because ceph_d_revalidate(..., LOOKUP_RCU)
1384 * may call __ceph_caps_issued_mask() on a freeing inode. */
1385 spin_lock(&ci->i_ceph_lock);
1386 p = rb_first(&ci->i_caps);
1387 while (p) {
1388 struct ceph_cap *cap = rb_entry(p, struct ceph_cap, ci_node);
1389 p = rb_next(p);
1390 ceph_remove_cap(mdsc, cap, true);
1391 }
1392 spin_unlock(&ci->i_ceph_lock);
1393 }
1394
1395 /*
1396 * Prepare to send a cap message to an MDS. Update the cap state, and populate
1397 * the arg struct with the parameters that will need to be sent. This should
1398 * be done under the i_ceph_lock to guard against changes to cap state.
1399 *
1400 * Make note of max_size reported/requested from mds, revoked caps
1401 * that have now been implemented.
1402 */
1403 static void __prep_cap(struct cap_msg_args *arg, struct ceph_cap *cap,
1404 int op, int flags, int used, int want, int retain,
1405 int flushing, u64 flush_tid, u64 oldest_flush_tid)
1406 {
1407 struct ceph_inode_info *ci = cap->ci;
1408 struct inode *inode = &ci->netfs.inode;
1409 struct ceph_client *cl = ceph_inode_to_client(inode);
1410 int held, revoking;
1411
1412 lockdep_assert_held(&ci->i_ceph_lock);
1413
1414 held = cap->issued | cap->implemented;
1415 revoking = cap->implemented & ~cap->issued;
1416 retain &= ~revoking;
1417
1418 doutc(cl, "%p %llx.%llx cap %p session %p %s -> %s (revoking %s)\n",
1419 inode, ceph_vinop(inode), cap, cap->session,
1420 ceph_cap_string(held), ceph_cap_string(held & retain),
1421 ceph_cap_string(revoking));
1422 BUG_ON((retain & CEPH_CAP_PIN) == 0);
1423
1424 ci->i_ceph_flags &= ~CEPH_I_FLUSH;
1425
1426 cap->issued &= retain; /* drop bits we don't want */
1427 /*
1428 * Wake up any waiters on wanted -> needed transition. This is due to
1429 * the weird transition from buffered to sync IO... we need to flush
1430 * dirty pages _before_ allowing sync writes to avoid reordering.
1431 */
1432 arg->wake = cap->implemented & ~cap->issued;
1433 cap->implemented &= cap->issued | used;
1434 cap->mds_wanted = want;
1435
1436 arg->session = cap->session;
1437 arg->ino = ceph_vino(inode).ino;
1438 arg->cid = cap->cap_id;
1439 arg->follows = flushing ? ci->i_head_snapc->seq : 0;
1440 arg->flush_tid = flush_tid;
1441 arg->oldest_flush_tid = oldest_flush_tid;
1442 arg->size = i_size_read(inode);
1443 ci->i_reported_size = arg->size;
1444 arg->max_size = ci->i_wanted_max_size;
1445 if (cap == ci->i_auth_cap) {
1446 if (want & CEPH_CAP_ANY_FILE_WR)
1447 ci->i_requested_max_size = arg->max_size;
1448 else
1449 ci->i_requested_max_size = 0;
1450 }
1451
1452 if (flushing & CEPH_CAP_XATTR_EXCL) {
1453 arg->old_xattr_buf = __ceph_build_xattrs_blob(ci);
1454 arg->xattr_version = ci->i_xattrs.version;
1455 arg->xattr_buf = ceph_buffer_get(ci->i_xattrs.blob);
1456 } else {
1457 arg->xattr_buf = NULL;
1458 arg->old_xattr_buf = NULL;
1459 }
1460
1461 arg->mtime = inode_get_mtime(inode);
1462 arg->atime = inode_get_atime(inode);
1463 arg->ctime = inode_get_ctime(inode);
1464 arg->btime = ci->i_btime;
1465 arg->change_attr = inode_peek_iversion_raw(inode);
1466
1467 arg->op = op;
1468 arg->caps = cap->implemented;
1469 arg->wanted = want;
1470 arg->dirty = flushing;
1471
1472 arg->seq = cap->seq;
1473 arg->issue_seq = cap->issue_seq;
1474 arg->mseq = cap->mseq;
1475 arg->time_warp_seq = ci->i_time_warp_seq;
1476
1477 arg->uid = inode->i_uid;
1478 arg->gid = inode->i_gid;
1479 arg->mode = inode->i_mode;
1480
1481 arg->inline_data = ci->i_inline_version != CEPH_INLINE_NONE;
1482 if (!(flags & CEPH_CLIENT_CAPS_PENDING_CAPSNAP) &&
1483 !list_empty(&ci->i_cap_snaps)) {
1484 struct ceph_cap_snap *capsnap;
1485 list_for_each_entry_reverse(capsnap, &ci->i_cap_snaps, ci_item) {
1486 if (capsnap->cap_flush.tid)
1487 break;
1488 if (capsnap->need_flush) {
1489 flags |= CEPH_CLIENT_CAPS_PENDING_CAPSNAP;
1490 break;
1491 }
1492 }
1493 }
1494 arg->flags = flags;
1495 arg->encrypted = IS_ENCRYPTED(inode);
1496 #if IS_ENABLED(CONFIG_FS_ENCRYPTION)
1497 if (ci->fscrypt_auth_len &&
1498 WARN_ON_ONCE(ci->fscrypt_auth_len > sizeof(struct ceph_fscrypt_auth))) {
1499 /* Don't set this if it's too big */
1500 arg->fscrypt_auth_len = 0;
1501 } else {
1502 arg->fscrypt_auth_len = ci->fscrypt_auth_len;
1503 memcpy(arg->fscrypt_auth, ci->fscrypt_auth,
1504 min_t(size_t, ci->fscrypt_auth_len,
1505 sizeof(arg->fscrypt_auth)));
1506 }
1507 #endif /* CONFIG_FS_ENCRYPTION */
1508 }
1509
1510 #if IS_ENABLED(CONFIG_FS_ENCRYPTION)
1511 #define CAP_MSG_FIXED_FIELDS (sizeof(struct ceph_mds_caps) + \
1512 4 + 8 + 4 + 4 + 8 + 4 + 4 + 4 + 8 + 8 + 4 + 8 + 8 + 4 + 4 + 8)
1513
1514 static inline int cap_msg_size(struct cap_msg_args *arg)
1515 {
1516 return CAP_MSG_FIXED_FIELDS + arg->fscrypt_auth_len;
1517 }
1518 #else
1519 #define CAP_MSG_FIXED_FIELDS (sizeof(struct ceph_mds_caps) + \
1520 4 + 8 + 4 + 4 + 8 + 4 + 4 + 4 + 8 + 8 + 4 + 8 + 8 + 4 + 4)
1521
1522 static inline int cap_msg_size(struct cap_msg_args *arg)
1523 {
1524 return CAP_MSG_FIXED_FIELDS;
1525 }
1526 #endif /* CONFIG_FS_ENCRYPTION */
1527
1528 /*
1529 * Send a cap msg on the given inode.
1530 *
1531 * Caller should hold snap_rwsem (read), s_mutex.
1532 */
1533 static void __send_cap(struct cap_msg_args *arg, struct ceph_inode_info *ci)
1534 {
1535 struct ceph_msg *msg;
1536 struct inode *inode = &ci->netfs.inode;
1537 struct ceph_client *cl = ceph_inode_to_client(inode);
1538
1539 msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPS, cap_msg_size(arg), GFP_NOFS,
1540 false);
1541 if (!msg) {
1542 pr_err_client(cl,
1543 "error allocating cap msg: ino (%llx.%llx)"
1544 " flushing %s tid %llu, requeuing cap.\n",
1545 ceph_vinop(inode), ceph_cap_string(arg->dirty),
1546 arg->flush_tid);
1547 spin_lock(&ci->i_ceph_lock);
1548 __cap_delay_requeue(arg->session->s_mdsc, ci);
1549 spin_unlock(&ci->i_ceph_lock);
1550 return;
1551 }
1552
1553 encode_cap_msg(msg, arg);
1554 ceph_con_send(&arg->session->s_con, msg);
1555 ceph_buffer_put(arg->old_xattr_buf);
1556 ceph_buffer_put(arg->xattr_buf);
1557 if (arg->wake)
1558 wake_up_all(&ci->i_cap_wq);
1559 }
1560
1561 static inline int __send_flush_snap(struct inode *inode,
1562 struct ceph_mds_session *session,
1563 struct ceph_cap_snap *capsnap,
1564 u32 mseq, u64 oldest_flush_tid)
1565 {
1566 struct cap_msg_args arg;
1567 struct ceph_msg *msg;
1568
1569 arg.session = session;
1570 arg.ino = ceph_vino(inode).ino;
1571 arg.cid = 0;
1572 arg.follows = capsnap->follows;
1573 arg.flush_tid = capsnap->cap_flush.tid;
1574 arg.oldest_flush_tid = oldest_flush_tid;
1575
1576 arg.size = capsnap->size;
1577 arg.max_size = 0;
1578 arg.xattr_version = capsnap->xattr_version;
1579 arg.xattr_buf = capsnap->xattr_blob;
1580 arg.old_xattr_buf = NULL;
1581
1582 arg.atime = capsnap->atime;
1583 arg.mtime = capsnap->mtime;
1584 arg.ctime = capsnap->ctime;
1585 arg.btime = capsnap->btime;
1586 arg.change_attr = capsnap->change_attr;
1587
1588 arg.op = CEPH_CAP_OP_FLUSHSNAP;
1589 arg.caps = capsnap->issued;
1590 arg.wanted = 0;
1591 arg.dirty = capsnap->dirty;
1592
1593 arg.seq = 0;
1594 arg.issue_seq = 0;
1595 arg.mseq = mseq;
1596 arg.time_warp_seq = capsnap->time_warp_seq;
1597
1598 arg.uid = capsnap->uid;
1599 arg.gid = capsnap->gid;
1600 arg.mode = capsnap->mode;
1601
1602 arg.inline_data = capsnap->inline_data;
1603 arg.flags = 0;
1604 arg.wake = false;
1605 arg.encrypted = IS_ENCRYPTED(inode);
1606
1607 /* No fscrypt_auth changes from a capsnap.*/
1608 arg.fscrypt_auth_len = 0;
1609
1610 msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPS, cap_msg_size(&arg),
1611 GFP_NOFS, false);
1612 if (!msg)
1613 return -ENOMEM;
1614
1615 encode_cap_msg(msg, &arg);
1616 ceph_con_send(&arg.session->s_con, msg);
1617 return 0;
1618 }
1619
1620 /*
1621 * When a snapshot is taken, clients accumulate dirty metadata on
1622 * inodes with capabilities in ceph_cap_snaps to describe the file
1623 * state at the time the snapshot was taken. This must be flushed
1624 * asynchronously back to the MDS once sync writes complete and dirty
1625 * data is written out.
1626 *
1627 * Called under i_ceph_lock.
1628 */
1629 static void __ceph_flush_snaps(struct ceph_inode_info *ci,
1630 struct ceph_mds_session *session)
1631 __releases(ci->i_ceph_lock)
1632 __acquires(ci->i_ceph_lock)
1633 {
1634 struct inode *inode = &ci->netfs.inode;
1635 struct ceph_mds_client *mdsc = session->s_mdsc;
1636 struct ceph_client *cl = mdsc->fsc->client;
1637 struct ceph_cap_snap *capsnap;
1638 u64 oldest_flush_tid = 0;
1639 u64 first_tid = 1, last_tid = 0;
1640
1641 doutc(cl, "%p %llx.%llx session %p\n", inode, ceph_vinop(inode),
1642 session);
1643
1644 list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) {
1645 /*
1646 * we need to wait for sync writes to complete and for dirty
1647 * pages to be written out.
1648 */
1649 if (capsnap->dirty_pages || capsnap->writing)
1650 break;
1651
1652 /* should be removed by ceph_try_drop_cap_snap() */
1653 BUG_ON(!capsnap->need_flush);
1654
1655 /* only flush each capsnap once */
1656 if (capsnap->cap_flush.tid > 0) {
1657 doutc(cl, "already flushed %p, skipping\n", capsnap);
1658 continue;
1659 }
1660
1661 spin_lock(&mdsc->cap_dirty_lock);
1662 capsnap->cap_flush.tid = ++mdsc->last_cap_flush_tid;
1663 list_add_tail(&capsnap->cap_flush.g_list,
1664 &mdsc->cap_flush_list);
1665 if (oldest_flush_tid == 0)
1666 oldest_flush_tid = __get_oldest_flush_tid(mdsc);
1667 if (list_empty(&ci->i_flushing_item)) {
1668 list_add_tail(&ci->i_flushing_item,
1669 &session->s_cap_flushing);
1670 }
1671 spin_unlock(&mdsc->cap_dirty_lock);
1672
1673 list_add_tail(&capsnap->cap_flush.i_list,
1674 &ci->i_cap_flush_list);
1675
1676 if (first_tid == 1)
1677 first_tid = capsnap->cap_flush.tid;
1678 last_tid = capsnap->cap_flush.tid;
1679 }
1680
1681 ci->i_ceph_flags &= ~CEPH_I_FLUSH_SNAPS;
1682
1683 while (first_tid <= last_tid) {
1684 struct ceph_cap *cap = ci->i_auth_cap;
1685 struct ceph_cap_flush *cf = NULL, *iter;
1686 int ret;
1687
1688 if (!(cap && cap->session == session)) {
1689 doutc(cl, "%p %llx.%llx auth cap %p not mds%d, stop\n",
1690 inode, ceph_vinop(inode), cap, session->s_mds);
1691 break;
1692 }
1693
1694 ret = -ENOENT;
1695 list_for_each_entry(iter, &ci->i_cap_flush_list, i_list) {
1696 if (iter->tid >= first_tid) {
1697 cf = iter;
1698 ret = 0;
1699 break;
1700 }
1701 }
1702 if (ret < 0)
1703 break;
1704
1705 first_tid = cf->tid + 1;
1706
1707 capsnap = container_of(cf, struct ceph_cap_snap, cap_flush);
1708 refcount_inc(&capsnap->nref);
1709 spin_unlock(&ci->i_ceph_lock);
1710
1711 doutc(cl, "%p %llx.%llx capsnap %p tid %llu %s\n", inode,
1712 ceph_vinop(inode), capsnap, cf->tid,
1713 ceph_cap_string(capsnap->dirty));
1714
1715 ret = __send_flush_snap(inode, session, capsnap, cap->mseq,
1716 oldest_flush_tid);
1717 if (ret < 0) {
1718 pr_err_client(cl, "error sending cap flushsnap, "
1719 "ino (%llx.%llx) tid %llu follows %llu\n",
1720 ceph_vinop(inode), cf->tid,
1721 capsnap->follows);
1722 }
1723
1724 ceph_put_cap_snap(capsnap);
1725 spin_lock(&ci->i_ceph_lock);
1726 }
1727 }
1728
1729 void ceph_flush_snaps(struct ceph_inode_info *ci,
1730 struct ceph_mds_session **psession)
1731 {
1732 struct inode *inode = &ci->netfs.inode;
1733 struct ceph_mds_client *mdsc = ceph_inode_to_fs_client(inode)->mdsc;
1734 struct ceph_client *cl = ceph_inode_to_client(inode);
1735 struct ceph_mds_session *session = NULL;
1736 bool need_put = false;
1737 int mds;
1738
1739 doutc(cl, "%p %llx.%llx\n", inode, ceph_vinop(inode));
1740 if (psession)
1741 session = *psession;
1742 retry:
1743 spin_lock(&ci->i_ceph_lock);
1744 if (!(ci->i_ceph_flags & CEPH_I_FLUSH_SNAPS)) {
1745 doutc(cl, " no capsnap needs flush, doing nothing\n");
1746 goto out;
1747 }
1748 if (!ci->i_auth_cap) {
1749 doutc(cl, " no auth cap (migrating?), doing nothing\n");
1750 goto out;
1751 }
1752
1753 mds = ci->i_auth_cap->session->s_mds;
1754 if (session && session->s_mds != mds) {
1755 doutc(cl, " oops, wrong session %p mutex\n", session);
1756 ceph_put_mds_session(session);
1757 session = NULL;
1758 }
1759 if (!session) {
1760 spin_unlock(&ci->i_ceph_lock);
1761 mutex_lock(&mdsc->mutex);
1762 session = __ceph_lookup_mds_session(mdsc, mds);
1763 mutex_unlock(&mdsc->mutex);
1764 goto retry;
1765 }
1766
1767 // make sure flushsnap messages are sent in proper order.
1768 if (ci->i_ceph_flags & CEPH_I_KICK_FLUSH)
1769 __kick_flushing_caps(mdsc, session, ci, 0);
1770
1771 __ceph_flush_snaps(ci, session);
1772 out:
1773 spin_unlock(&ci->i_ceph_lock);
1774
1775 if (psession)
1776 *psession = session;
1777 else
1778 ceph_put_mds_session(session);
1779 /* we flushed them all; remove this inode from the queue */
1780 spin_lock(&mdsc->snap_flush_lock);
1781 if (!list_empty(&ci->i_snap_flush_item))
1782 need_put = true;
1783 list_del_init(&ci->i_snap_flush_item);
1784 spin_unlock(&mdsc->snap_flush_lock);
1785
1786 if (need_put)
1787 iput(inode);
1788 }
1789
1790 /*
1791 * Mark caps dirty. If inode is newly dirty, return the dirty flags.
1792 * Caller is then responsible for calling __mark_inode_dirty with the
1793 * returned flags value.
1794 */
1795 int __ceph_mark_dirty_caps(struct ceph_inode_info *ci, int mask,
1796 struct ceph_cap_flush **pcf)
1797 {
1798 struct ceph_mds_client *mdsc =
1799 ceph_sb_to_fs_client(ci->netfs.inode.i_sb)->mdsc;
1800 struct inode *inode = &ci->netfs.inode;
1801 struct ceph_client *cl = ceph_inode_to_client(inode);
1802 int was = ci->i_dirty_caps;
1803 int dirty = 0;
1804
1805 lockdep_assert_held(&ci->i_ceph_lock);
1806
1807 if (!ci->i_auth_cap) {
1808 pr_warn_client(cl, "%p %llx.%llx mask %s, "
1809 "but no auth cap (session was closed?)\n",
1810 inode, ceph_vinop(inode),
1811 ceph_cap_string(mask));
1812 return 0;
1813 }
1814
1815 doutc(cl, "%p %llx.%llx %s dirty %s -> %s\n", inode,
1816 ceph_vinop(inode), ceph_cap_string(mask),
1817 ceph_cap_string(was), ceph_cap_string(was | mask));
1818 ci->i_dirty_caps |= mask;
1819 if (was == 0) {
1820 struct ceph_mds_session *session = ci->i_auth_cap->session;
1821
1822 WARN_ON_ONCE(ci->i_prealloc_cap_flush);
1823 swap(ci->i_prealloc_cap_flush, *pcf);
1824
1825 if (!ci->i_head_snapc) {
1826 WARN_ON_ONCE(!rwsem_is_locked(&mdsc->snap_rwsem));
1827 ci->i_head_snapc = ceph_get_snap_context(
1828 ci->i_snap_realm->cached_context);
1829 }
1830 doutc(cl, "%p %llx.%llx now dirty snapc %p auth cap %p\n",
1831 inode, ceph_vinop(inode), ci->i_head_snapc,
1832 ci->i_auth_cap);
1833 BUG_ON(!list_empty(&ci->i_dirty_item));
1834 spin_lock(&mdsc->cap_dirty_lock);
1835 list_add(&ci->i_dirty_item, &session->s_cap_dirty);
1836 spin_unlock(&mdsc->cap_dirty_lock);
1837 if (ci->i_flushing_caps == 0) {
1838 ihold(inode);
1839 dirty |= I_DIRTY_SYNC;
1840 }
1841 } else {
1842 WARN_ON_ONCE(!ci->i_prealloc_cap_flush);
1843 }
1844 BUG_ON(list_empty(&ci->i_dirty_item));
1845 if (((was | ci->i_flushing_caps) & CEPH_CAP_FILE_BUFFER) &&
1846 (mask & CEPH_CAP_FILE_BUFFER))
1847 dirty |= I_DIRTY_DATASYNC;
1848 __cap_delay_requeue(mdsc, ci);
1849 return dirty;
1850 }
1851
1852 struct ceph_cap_flush *ceph_alloc_cap_flush(void)
1853 {
1854 struct ceph_cap_flush *cf;
1855
1856 cf = kmem_cache_alloc(ceph_cap_flush_cachep, GFP_KERNEL);
1857 if (!cf)
1858 return NULL;
1859
1860 cf->is_capsnap = false;
1861 return cf;
1862 }
1863
1864 void ceph_free_cap_flush(struct ceph_cap_flush *cf)
1865 {
1866 if (cf)
1867 kmem_cache_free(ceph_cap_flush_cachep, cf);
1868 }
1869
1870 static u64 __get_oldest_flush_tid(struct ceph_mds_client *mdsc)
1871 {
1872 if (!list_empty(&mdsc->cap_flush_list)) {
1873 struct ceph_cap_flush *cf =
1874 list_first_entry(&mdsc->cap_flush_list,
1875 struct ceph_cap_flush, g_list);
1876 return cf->tid;
1877 }
1878 return 0;
1879 }
1880
1881 /*
1882 * Remove cap_flush from the mdsc's or inode's flushing cap list.
1883 * Return true if caller needs to wake up flush waiters.
1884 */
1885 static bool __detach_cap_flush_from_mdsc(struct ceph_mds_client *mdsc,
1886 struct ceph_cap_flush *cf)
1887 {
1888 struct ceph_cap_flush *prev;
1889 bool wake = cf->wake;
1890
1891 if (wake && cf->g_list.prev != &mdsc->cap_flush_list) {
1892 prev = list_prev_entry(cf, g_list);
1893 prev->wake = true;
1894 wake = false;
1895 }
1896 list_del_init(&cf->g_list);
1897 return wake;
1898 }
1899
1900 static bool __detach_cap_flush_from_ci(struct ceph_inode_info *ci,
1901 struct ceph_cap_flush *cf)
1902 {
1903 struct ceph_cap_flush *prev;
1904 bool wake = cf->wake;
1905
1906 if (wake && cf->i_list.prev != &ci->i_cap_flush_list) {
1907 prev = list_prev_entry(cf, i_list);
1908 prev->wake = true;
1909 wake = false;
1910 }
1911 list_del_init(&cf->i_list);
1912 return wake;
1913 }
1914
1915 /*
1916 * Add dirty inode to the flushing list. Assigned a seq number so we
1917 * can wait for caps to flush without starving.
1918 *
1919 * Called under i_ceph_lock. Returns the flush tid.
1920 */
1921 static u64 __mark_caps_flushing(struct inode *inode,
1922 struct ceph_mds_session *session, bool wake,
1923 u64 *oldest_flush_tid)
1924 {
1925 struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(inode->i_sb)->mdsc;
1926 struct ceph_client *cl = ceph_inode_to_client(inode);
1927 struct ceph_inode_info *ci = ceph_inode(inode);
1928 struct ceph_cap_flush *cf = NULL;
1929 int flushing;
1930
1931 lockdep_assert_held(&ci->i_ceph_lock);
1932 BUG_ON(ci->i_dirty_caps == 0);
1933 BUG_ON(list_empty(&ci->i_dirty_item));
1934 BUG_ON(!ci->i_prealloc_cap_flush);
1935
1936 flushing = ci->i_dirty_caps;
1937 doutc(cl, "flushing %s, flushing_caps %s -> %s\n",
1938 ceph_cap_string(flushing),
1939 ceph_cap_string(ci->i_flushing_caps),
1940 ceph_cap_string(ci->i_flushing_caps | flushing));
1941 ci->i_flushing_caps |= flushing;
1942 ci->i_dirty_caps = 0;
1943 doutc(cl, "%p %llx.%llx now !dirty\n", inode, ceph_vinop(inode));
1944
1945 swap(cf, ci->i_prealloc_cap_flush);
1946 cf->caps = flushing;
1947 cf->wake = wake;
1948
1949 spin_lock(&mdsc->cap_dirty_lock);
1950 list_del_init(&ci->i_dirty_item);
1951
1952 cf->tid = ++mdsc->last_cap_flush_tid;
1953 list_add_tail(&cf->g_list, &mdsc->cap_flush_list);
1954 *oldest_flush_tid = __get_oldest_flush_tid(mdsc);
1955
1956 if (list_empty(&ci->i_flushing_item)) {
1957 list_add_tail(&ci->i_flushing_item, &session->s_cap_flushing);
1958 mdsc->num_cap_flushing++;
1959 }
1960 spin_unlock(&mdsc->cap_dirty_lock);
1961
1962 list_add_tail(&cf->i_list, &ci->i_cap_flush_list);
1963
1964 return cf->tid;
1965 }
1966
1967 /*
1968 * try to invalidate mapping pages without blocking.
1969 */
1970 static int try_nonblocking_invalidate(struct inode *inode)
1971 __releases(ci->i_ceph_lock)
1972 __acquires(ci->i_ceph_lock)
1973 {
1974 struct ceph_client *cl = ceph_inode_to_client(inode);
1975 struct ceph_inode_info *ci = ceph_inode(inode);
1976 u32 invalidating_gen = ci->i_rdcache_gen;
1977
1978 spin_unlock(&ci->i_ceph_lock);
1979 ceph_fscache_invalidate(inode, false);
1980 invalidate_mapping_pages(&inode->i_data, 0, -1);
1981 spin_lock(&ci->i_ceph_lock);
1982
1983 if (inode->i_data.nrpages == 0 &&
1984 invalidating_gen == ci->i_rdcache_gen) {
1985 /* success. */
1986 doutc(cl, "%p %llx.%llx success\n", inode,
1987 ceph_vinop(inode));
1988 /* save any racing async invalidate some trouble */
1989 ci->i_rdcache_revoking = ci->i_rdcache_gen - 1;
1990 return 0;
1991 }
1992 doutc(cl, "%p %llx.%llx failed\n", inode, ceph_vinop(inode));
1993 return -1;
1994 }
1995
1996 bool __ceph_should_report_size(struct ceph_inode_info *ci)
1997 {
1998 loff_t size = i_size_read(&ci->netfs.inode);
1999 /* mds will adjust max size according to the reported size */
2000 if (ci->i_flushing_caps & CEPH_CAP_FILE_WR)
2001 return false;
2002 if (size >= ci->i_max_size)
2003 return true;
2004 /* half of previous max_size increment has been used */
2005 if (ci->i_max_size > ci->i_reported_size &&
2006 (size << 1) >= ci->i_max_size + ci->i_reported_size)
2007 return true;
2008 return false;
2009 }
2010
2011 /*
2012 * Swiss army knife function to examine currently used and wanted
2013 * versus held caps. Release, flush, ack revoked caps to mds as
2014 * appropriate.
2015 *
2016 * CHECK_CAPS_AUTHONLY - we should only check the auth cap
2017 * CHECK_CAPS_FLUSH - we should flush any dirty caps immediately, without
2018 * further delay.
2019 * CHECK_CAPS_FLUSH_FORCE - we should flush any caps immediately, without
2020 * further delay.
2021 */
2022 void ceph_check_caps(struct ceph_inode_info *ci, int flags)
2023 {
2024 struct inode *inode = &ci->netfs.inode;
2025 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
2026 struct ceph_client *cl = ceph_inode_to_client(inode);
2027 struct ceph_cap *cap;
2028 u64 flush_tid, oldest_flush_tid;
2029 int file_wanted, used, cap_used;
2030 int issued, implemented, want, retain, revoking, flushing = 0;
2031 int mds = -1; /* keep track of how far we've gone through i_caps list
2032 to avoid an infinite loop on retry */
2033 struct rb_node *p;
2034 bool queue_invalidate = false;
2035 bool tried_invalidate = false;
2036 bool queue_writeback = false;
2037 struct ceph_mds_session *session = NULL;
2038
2039 spin_lock(&ci->i_ceph_lock);
2040 if (ci->i_ceph_flags & CEPH_I_ASYNC_CREATE) {
2041 ci->i_ceph_flags |= CEPH_I_ASYNC_CHECK_CAPS;
2042
2043 /* Don't send messages until we get async create reply */
2044 spin_unlock(&ci->i_ceph_lock);
2045 return;
2046 }
2047
2048 if (ci->i_ceph_flags & CEPH_I_FLUSH)
2049 flags |= CHECK_CAPS_FLUSH;
2050 retry:
2051 /* Caps wanted by virtue of active open files. */
2052 file_wanted = __ceph_caps_file_wanted(ci);
2053
2054 /* Caps which have active references against them */
2055 used = __ceph_caps_used(ci);
2056
2057 /*
2058 * "issued" represents the current caps that the MDS wants us to have.
2059 * "implemented" is the set that we have been granted, and includes the
2060 * ones that have not yet been returned to the MDS (the "revoking" set,
2061 * usually because they have outstanding references).
2062 */
2063 issued = __ceph_caps_issued(ci, &implemented);
2064 revoking = implemented & ~issued;
2065
2066 want = file_wanted;
2067
2068 /* The ones we currently want to retain (may be adjusted below) */
2069 retain = file_wanted | used | CEPH_CAP_PIN;
2070 if (!mdsc->stopping && inode->i_nlink > 0) {
2071 if (file_wanted) {
2072 retain |= CEPH_CAP_ANY; /* be greedy */
2073 } else if (S_ISDIR(inode->i_mode) &&
2074 (issued & CEPH_CAP_FILE_SHARED) &&
2075 __ceph_dir_is_complete(ci)) {
2076 /*
2077 * If a directory is complete, we want to keep
2078 * the exclusive cap. So that MDS does not end up
2079 * revoking the shared cap on every create/unlink
2080 * operation.
2081 */
2082 if (IS_RDONLY(inode)) {
2083 want = CEPH_CAP_ANY_SHARED;
2084 } else {
2085 want |= CEPH_CAP_ANY_SHARED | CEPH_CAP_FILE_EXCL;
2086 }
2087 retain |= want;
2088 } else {
2089
2090 retain |= CEPH_CAP_ANY_SHARED;
2091 /*
2092 * keep RD only if we didn't have the file open RW,
2093 * because then the mds would revoke it anyway to
2094 * journal max_size=0.
2095 */
2096 if (ci->i_max_size == 0)
2097 retain |= CEPH_CAP_ANY_RD;
2098 }
2099 }
2100
2101 doutc(cl, "%p %llx.%llx file_want %s used %s dirty %s "
2102 "flushing %s issued %s revoking %s retain %s %s%s%s%s\n",
2103 inode, ceph_vinop(inode), ceph_cap_string(file_wanted),
2104 ceph_cap_string(used), ceph_cap_string(ci->i_dirty_caps),
2105 ceph_cap_string(ci->i_flushing_caps),
2106 ceph_cap_string(issued), ceph_cap_string(revoking),
2107 ceph_cap_string(retain),
2108 (flags & CHECK_CAPS_AUTHONLY) ? " AUTHONLY" : "",
2109 (flags & CHECK_CAPS_FLUSH) ? " FLUSH" : "",
2110 (flags & CHECK_CAPS_NOINVAL) ? " NOINVAL" : "",
2111 (flags & CHECK_CAPS_FLUSH_FORCE) ? " FLUSH_FORCE" : "");
2112
2113 /*
2114 * If we no longer need to hold onto old our caps, and we may
2115 * have cached pages, but don't want them, then try to invalidate.
2116 * If we fail, it's because pages are locked.... try again later.
2117 */
2118 if ((!(flags & CHECK_CAPS_NOINVAL) || mdsc->stopping) &&
2119 S_ISREG(inode->i_mode) &&
2120 !(ci->i_wb_ref || ci->i_wrbuffer_ref) && /* no dirty pages... */
2121 inode->i_data.nrpages && /* have cached pages */
2122 (revoking & (CEPH_CAP_FILE_CACHE|
2123 CEPH_CAP_FILE_LAZYIO)) && /* or revoking cache */
2124 !tried_invalidate) {
2125 doutc(cl, "trying to invalidate on %p %llx.%llx\n",
2126 inode, ceph_vinop(inode));
2127 if (try_nonblocking_invalidate(inode) < 0) {
2128 doutc(cl, "queuing invalidate\n");
2129 queue_invalidate = true;
2130 ci->i_rdcache_revoking = ci->i_rdcache_gen;
2131 }
2132 tried_invalidate = true;
2133 goto retry;
2134 }
2135
2136 for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
2137 int mflags = 0;
2138 struct cap_msg_args arg;
2139
2140 cap = rb_entry(p, struct ceph_cap, ci_node);
2141
2142 /* avoid looping forever */
2143 if (mds >= cap->mds ||
2144 ((flags & CHECK_CAPS_AUTHONLY) && cap != ci->i_auth_cap))
2145 continue;
2146
2147 /*
2148 * If we have an auth cap, we don't need to consider any
2149 * overlapping caps as used.
2150 */
2151 cap_used = used;
2152 if (ci->i_auth_cap && cap != ci->i_auth_cap)
2153 cap_used &= ~ci->i_auth_cap->issued;
2154
2155 revoking = cap->implemented & ~cap->issued;
2156 doutc(cl, " mds%d cap %p used %s issued %s implemented %s revoking %s\n",
2157 cap->mds, cap, ceph_cap_string(cap_used),
2158 ceph_cap_string(cap->issued),
2159 ceph_cap_string(cap->implemented),
2160 ceph_cap_string(revoking));
2161
2162 /* completed revocation? going down and there are no caps? */
2163 if (revoking) {
2164 if ((revoking & cap_used) == 0) {
2165 doutc(cl, "completed revocation of %s\n",
2166 ceph_cap_string(cap->implemented & ~cap->issued));
2167 goto ack;
2168 }
2169
2170 /*
2171 * If the "i_wrbuffer_ref" was increased by mmap or generic
2172 * cache write just before the ceph_check_caps() is called,
2173 * the Fb capability revoking will fail this time. Then we
2174 * must wait for the BDI's delayed work to flush the dirty
2175 * pages and to release the "i_wrbuffer_ref", which will cost
2176 * at most 5 seconds. That means the MDS needs to wait at
2177 * most 5 seconds to finished the Fb capability's revocation.
2178 *
2179 * Let's queue a writeback for it.
2180 */
2181 if (S_ISREG(inode->i_mode) && ci->i_wrbuffer_ref &&
2182 (revoking & CEPH_CAP_FILE_BUFFER))
2183 queue_writeback = true;
2184 }
2185
2186 if (flags & CHECK_CAPS_FLUSH_FORCE) {
2187 doutc(cl, "force to flush caps\n");
2188 goto ack;
2189 }
2190
2191 if (cap == ci->i_auth_cap &&
2192 (cap->issued & CEPH_CAP_FILE_WR)) {
2193 /* request larger max_size from MDS? */
2194 if (ci->i_wanted_max_size > ci->i_max_size &&
2195 ci->i_wanted_max_size > ci->i_requested_max_size) {
2196 doutc(cl, "requesting new max_size\n");
2197 goto ack;
2198 }
2199
2200 /* approaching file_max? */
2201 if (__ceph_should_report_size(ci)) {
2202 doutc(cl, "i_size approaching max_size\n");
2203 goto ack;
2204 }
2205 }
2206 /* flush anything dirty? */
2207 if (cap == ci->i_auth_cap) {
2208 if ((flags & CHECK_CAPS_FLUSH) && ci->i_dirty_caps) {
2209 doutc(cl, "flushing dirty caps\n");
2210 goto ack;
2211 }
2212 if (ci->i_ceph_flags & CEPH_I_FLUSH_SNAPS) {
2213 doutc(cl, "flushing snap caps\n");
2214 goto ack;
2215 }
2216 }
2217
2218 /* want more caps from mds? */
2219 if (want & ~cap->mds_wanted) {
2220 if (want & ~(cap->mds_wanted | cap->issued))
2221 goto ack;
2222 if (!__cap_is_valid(cap))
2223 goto ack;
2224 }
2225
2226 /* things we might delay */
2227 if ((cap->issued & ~retain) == 0)
2228 continue; /* nope, all good */
2229
2230 ack:
2231 ceph_put_mds_session(session);
2232 session = ceph_get_mds_session(cap->session);
2233
2234 /* kick flushing and flush snaps before sending normal
2235 * cap message */
2236 if (cap == ci->i_auth_cap &&
2237 (ci->i_ceph_flags &
2238 (CEPH_I_KICK_FLUSH | CEPH_I_FLUSH_SNAPS))) {
2239 if (ci->i_ceph_flags & CEPH_I_KICK_FLUSH)
2240 __kick_flushing_caps(mdsc, session, ci, 0);
2241 if (ci->i_ceph_flags & CEPH_I_FLUSH_SNAPS)
2242 __ceph_flush_snaps(ci, session);
2243
2244 goto retry;
2245 }
2246
2247 if (cap == ci->i_auth_cap && ci->i_dirty_caps) {
2248 flushing = ci->i_dirty_caps;
2249 flush_tid = __mark_caps_flushing(inode, session, false,
2250 &oldest_flush_tid);
2251 if (flags & CHECK_CAPS_FLUSH &&
2252 list_empty(&session->s_cap_dirty))
2253 mflags |= CEPH_CLIENT_CAPS_SYNC;
2254 } else {
2255 flushing = 0;
2256 flush_tid = 0;
2257 spin_lock(&mdsc->cap_dirty_lock);
2258 oldest_flush_tid = __get_oldest_flush_tid(mdsc);
2259 spin_unlock(&mdsc->cap_dirty_lock);
2260 }
2261
2262 mds = cap->mds; /* remember mds, so we don't repeat */
2263
2264 __prep_cap(&arg, cap, CEPH_CAP_OP_UPDATE, mflags, cap_used,
2265 want, retain, flushing, flush_tid, oldest_flush_tid);
2266
2267 spin_unlock(&ci->i_ceph_lock);
2268 __send_cap(&arg, ci);
2269 spin_lock(&ci->i_ceph_lock);
2270
2271 goto retry; /* retake i_ceph_lock and restart our cap scan. */
2272 }
2273
2274 /* periodically re-calculate caps wanted by open files */
2275 if (__ceph_is_any_real_caps(ci) &&
2276 list_empty(&ci->i_cap_delay_list) &&
2277 (file_wanted & ~CEPH_CAP_PIN) &&
2278 !(used & (CEPH_CAP_FILE_RD | CEPH_CAP_ANY_FILE_WR))) {
2279 __cap_delay_requeue(mdsc, ci);
2280 }
2281
2282 spin_unlock(&ci->i_ceph_lock);
2283
2284 ceph_put_mds_session(session);
2285 if (queue_writeback)
2286 ceph_queue_writeback(inode);
2287 if (queue_invalidate)
2288 ceph_queue_invalidate(inode);
2289 }
2290
2291 /*
2292 * Try to flush dirty caps back to the auth mds.
2293 */
2294 static int try_flush_caps(struct inode *inode, u64 *ptid)
2295 {
2296 struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(inode->i_sb)->mdsc;
2297 struct ceph_inode_info *ci = ceph_inode(inode);
2298 int flushing = 0;
2299 u64 flush_tid = 0, oldest_flush_tid = 0;
2300
2301 spin_lock(&ci->i_ceph_lock);
2302 retry_locked:
2303 if (ci->i_dirty_caps && ci->i_auth_cap) {
2304 struct ceph_cap *cap = ci->i_auth_cap;
2305 struct cap_msg_args arg;
2306 struct ceph_mds_session *session = cap->session;
2307
2308 if (session->s_state < CEPH_MDS_SESSION_OPEN) {
2309 spin_unlock(&ci->i_ceph_lock);
2310 goto out;
2311 }
2312
2313 if (ci->i_ceph_flags &
2314 (CEPH_I_KICK_FLUSH | CEPH_I_FLUSH_SNAPS)) {
2315 if (ci->i_ceph_flags & CEPH_I_KICK_FLUSH)
2316 __kick_flushing_caps(mdsc, session, ci, 0);
2317 if (ci->i_ceph_flags & CEPH_I_FLUSH_SNAPS)
2318 __ceph_flush_snaps(ci, session);
2319 goto retry_locked;
2320 }
2321
2322 flushing = ci->i_dirty_caps;
2323 flush_tid = __mark_caps_flushing(inode, session, true,
2324 &oldest_flush_tid);
2325
2326 __prep_cap(&arg, cap, CEPH_CAP_OP_FLUSH, CEPH_CLIENT_CAPS_SYNC,
2327 __ceph_caps_used(ci), __ceph_caps_wanted(ci),
2328 (cap->issued | cap->implemented),
2329 flushing, flush_tid, oldest_flush_tid);
2330 spin_unlock(&ci->i_ceph_lock);
2331
2332 __send_cap(&arg, ci);
2333 } else {
2334 if (!list_empty(&ci->i_cap_flush_list)) {
2335 struct ceph_cap_flush *cf =
2336 list_last_entry(&ci->i_cap_flush_list,
2337 struct ceph_cap_flush, i_list);
2338 cf->wake = true;
2339 flush_tid = cf->tid;
2340 }
2341 flushing = ci->i_flushing_caps;
2342 spin_unlock(&ci->i_ceph_lock);
2343 }
2344 out:
2345 *ptid = flush_tid;
2346 return flushing;
2347 }
2348
2349 /*
2350 * Return true if we've flushed caps through the given flush_tid.
2351 */
2352 static int caps_are_flushed(struct inode *inode, u64 flush_tid)
2353 {
2354 struct ceph_inode_info *ci = ceph_inode(inode);
2355 int ret = 1;
2356
2357 spin_lock(&ci->i_ceph_lock);
2358 if (!list_empty(&ci->i_cap_flush_list)) {
2359 struct ceph_cap_flush * cf =
2360 list_first_entry(&ci->i_cap_flush_list,
2361 struct ceph_cap_flush, i_list);
2362 if (cf->tid <= flush_tid)
2363 ret = 0;
2364 }
2365 spin_unlock(&ci->i_ceph_lock);
2366 return ret;
2367 }
2368
2369 /*
2370 * flush the mdlog and wait for any unsafe requests to complete.
2371 */
2372 static int flush_mdlog_and_wait_inode_unsafe_requests(struct inode *inode)
2373 {
2374 struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(inode->i_sb)->mdsc;
2375 struct ceph_client *cl = ceph_inode_to_client(inode);
2376 struct ceph_inode_info *ci = ceph_inode(inode);
2377 struct ceph_mds_request *req1 = NULL, *req2 = NULL;
2378 int ret, err = 0;
2379
2380 spin_lock(&ci->i_unsafe_lock);
2381 if (S_ISDIR(inode->i_mode) && !list_empty(&ci->i_unsafe_dirops)) {
2382 req1 = list_last_entry(&ci->i_unsafe_dirops,
2383 struct ceph_mds_request,
2384 r_unsafe_dir_item);
2385 ceph_mdsc_get_request(req1);
2386 }
2387 if (!list_empty(&ci->i_unsafe_iops)) {
2388 req2 = list_last_entry(&ci->i_unsafe_iops,
2389 struct ceph_mds_request,
2390 r_unsafe_target_item);
2391 ceph_mdsc_get_request(req2);
2392 }
2393 spin_unlock(&ci->i_unsafe_lock);
2394
2395 /*
2396 * Trigger to flush the journal logs in all the relevant MDSes
2397 * manually, or in the worst case we must wait at most 5 seconds
2398 * to wait the journal logs to be flushed by the MDSes periodically.
2399 */
2400 if (req1 || req2) {
2401 struct ceph_mds_request *req;
2402 struct ceph_mds_session **sessions;
2403 struct ceph_mds_session *s;
2404 unsigned int max_sessions;
2405 int i;
2406
2407 mutex_lock(&mdsc->mutex);
2408 max_sessions = mdsc->max_sessions;
2409
2410 sessions = kcalloc(max_sessions, sizeof(s), GFP_KERNEL);
2411 if (!sessions) {
2412 mutex_unlock(&mdsc->mutex);
2413 err = -ENOMEM;
2414 goto out;
2415 }
2416
2417 spin_lock(&ci->i_unsafe_lock);
2418 if (req1) {
2419 list_for_each_entry(req, &ci->i_unsafe_dirops,
2420 r_unsafe_dir_item) {
2421 s = req->r_session;
2422 if (!s)
2423 continue;
2424 if (!sessions[s->s_mds]) {
2425 s = ceph_get_mds_session(s);
2426 sessions[s->s_mds] = s;
2427 }
2428 }
2429 }
2430 if (req2) {
2431 list_for_each_entry(req, &ci->i_unsafe_iops,
2432 r_unsafe_target_item) {
2433 s = req->r_session;
2434 if (!s)
2435 continue;
2436 if (!sessions[s->s_mds]) {
2437 s = ceph_get_mds_session(s);
2438 sessions[s->s_mds] = s;
2439 }
2440 }
2441 }
2442 spin_unlock(&ci->i_unsafe_lock);
2443
2444 /* the auth MDS */
2445 spin_lock(&ci->i_ceph_lock);
2446 if (ci->i_auth_cap) {
2447 s = ci->i_auth_cap->session;
2448 if (!sessions[s->s_mds])
2449 sessions[s->s_mds] = ceph_get_mds_session(s);
2450 }
2451 spin_unlock(&ci->i_ceph_lock);
2452 mutex_unlock(&mdsc->mutex);
2453
2454 /* send flush mdlog request to MDSes */
2455 for (i = 0; i < max_sessions; i++) {
2456 s = sessions[i];
2457 if (s) {
2458 send_flush_mdlog(s);
2459 ceph_put_mds_session(s);
2460 }
2461 }
2462 kfree(sessions);
2463 }
2464
2465 doutc(cl, "%p %llx.%llx wait on tid %llu %llu\n", inode,
2466 ceph_vinop(inode), req1 ? req1->r_tid : 0ULL,
2467 req2 ? req2->r_tid : 0ULL);
2468 if (req1) {
2469 ret = !wait_for_completion_timeout(&req1->r_safe_completion,
2470 ceph_timeout_jiffies(req1->r_timeout));
2471 if (ret)
2472 err = -EIO;
2473 }
2474 if (req2) {
2475 ret = !wait_for_completion_timeout(&req2->r_safe_completion,
2476 ceph_timeout_jiffies(req2->r_timeout));
2477 if (ret)
2478 err = -EIO;
2479 }
2480
2481 out:
2482 if (req1)
2483 ceph_mdsc_put_request(req1);
2484 if (req2)
2485 ceph_mdsc_put_request(req2);
2486 return err;
2487 }
2488
2489 int ceph_fsync(struct file *file, loff_t start, loff_t end, int datasync)
2490 {
2491 struct inode *inode = file->f_mapping->host;
2492 struct ceph_inode_info *ci = ceph_inode(inode);
2493 struct ceph_client *cl = ceph_inode_to_client(inode);
2494 u64 flush_tid;
2495 int ret, err;
2496 int dirty;
2497
2498 doutc(cl, "%p %llx.%llx%s\n", inode, ceph_vinop(inode),
2499 datasync ? " datasync" : "");
2500
2501 ret = file_write_and_wait_range(file, start, end);
2502 if (datasync)
2503 goto out;
2504
2505 ret = ceph_wait_on_async_create(inode);
2506 if (ret)
2507 goto out;
2508
2509 dirty = try_flush_caps(inode, &flush_tid);
2510 doutc(cl, "dirty caps are %s\n", ceph_cap_string(dirty));
2511
2512 err = flush_mdlog_and_wait_inode_unsafe_requests(inode);
2513
2514 /*
2515 * only wait on non-file metadata writeback (the mds
2516 * can recover size and mtime, so we don't need to
2517 * wait for that)
2518 */
2519 if (!err && (dirty & ~CEPH_CAP_ANY_FILE_WR)) {
2520 err = wait_event_interruptible(ci->i_cap_wq,
2521 caps_are_flushed(inode, flush_tid));
2522 }
2523
2524 if (err < 0)
2525 ret = err;
2526
2527 err = file_check_and_advance_wb_err(file);
2528 if (err < 0)
2529 ret = err;
2530 out:
2531 doutc(cl, "%p %llx.%llx%s result=%d\n", inode, ceph_vinop(inode),
2532 datasync ? " datasync" : "", ret);
2533 return ret;
2534 }
2535
2536 /*
2537 * Flush any dirty caps back to the mds. If we aren't asked to wait,
2538 * queue inode for flush but don't do so immediately, because we can
2539 * get by with fewer MDS messages if we wait for data writeback to
2540 * complete first.
2541 */
2542 int ceph_write_inode(struct inode *inode, struct writeback_control *wbc)
2543 {
2544 struct ceph_inode_info *ci = ceph_inode(inode);
2545 struct ceph_client *cl = ceph_inode_to_client(inode);
2546 u64 flush_tid;
2547 int err = 0;
2548 int dirty;
2549 int wait = (wbc->sync_mode == WB_SYNC_ALL && !wbc->for_sync);
2550
2551 doutc(cl, "%p %llx.%llx wait=%d\n", inode, ceph_vinop(inode), wait);
2552 ceph_fscache_unpin_writeback(inode, wbc);
2553 if (wait) {
2554 err = ceph_wait_on_async_create(inode);
2555 if (err)
2556 return err;
2557 dirty = try_flush_caps(inode, &flush_tid);
2558 if (dirty)
2559 err = wait_event_interruptible(ci->i_cap_wq,
2560 caps_are_flushed(inode, flush_tid));
2561 } else {
2562 struct ceph_mds_client *mdsc =
2563 ceph_sb_to_fs_client(inode->i_sb)->mdsc;
2564
2565 spin_lock(&ci->i_ceph_lock);
2566 if (__ceph_caps_dirty(ci))
2567 __cap_delay_requeue_front(mdsc, ci);
2568 spin_unlock(&ci->i_ceph_lock);
2569 }
2570 return err;
2571 }
2572
2573 static void __kick_flushing_caps(struct ceph_mds_client *mdsc,
2574 struct ceph_mds_session *session,
2575 struct ceph_inode_info *ci,
2576 u64 oldest_flush_tid)
2577 __releases(ci->i_ceph_lock)
2578 __acquires(ci->i_ceph_lock)
2579 {
2580 struct inode *inode = &ci->netfs.inode;
2581 struct ceph_client *cl = mdsc->fsc->client;
2582 struct ceph_cap *cap;
2583 struct ceph_cap_flush *cf;
2584 int ret;
2585 u64 first_tid = 0;
2586 u64 last_snap_flush = 0;
2587
2588 /* Don't do anything until create reply comes in */
2589 if (ci->i_ceph_flags & CEPH_I_ASYNC_CREATE)
2590 return;
2591
2592 ci->i_ceph_flags &= ~CEPH_I_KICK_FLUSH;
2593
2594 list_for_each_entry_reverse(cf, &ci->i_cap_flush_list, i_list) {
2595 if (cf->is_capsnap) {
2596 last_snap_flush = cf->tid;
2597 break;
2598 }
2599 }
2600
2601 list_for_each_entry(cf, &ci->i_cap_flush_list, i_list) {
2602 if (cf->tid < first_tid)
2603 continue;
2604
2605 cap = ci->i_auth_cap;
2606 if (!(cap && cap->session == session)) {
2607 pr_err_client(cl, "%p auth cap %p not mds%d ???\n",
2608 inode, cap, session->s_mds);
2609 break;
2610 }
2611
2612 first_tid = cf->tid + 1;
2613
2614 if (!cf->is_capsnap) {
2615 struct cap_msg_args arg;
2616
2617 doutc(cl, "%p %llx.%llx cap %p tid %llu %s\n",
2618 inode, ceph_vinop(inode), cap, cf->tid,
2619 ceph_cap_string(cf->caps));
2620 __prep_cap(&arg, cap, CEPH_CAP_OP_FLUSH,
2621 (cf->tid < last_snap_flush ?
2622 CEPH_CLIENT_CAPS_PENDING_CAPSNAP : 0),
2623 __ceph_caps_used(ci),
2624 __ceph_caps_wanted(ci),
2625 (cap->issued | cap->implemented),
2626 cf->caps, cf->tid, oldest_flush_tid);
2627 spin_unlock(&ci->i_ceph_lock);
2628 __send_cap(&arg, ci);
2629 } else {
2630 struct ceph_cap_snap *capsnap =
2631 container_of(cf, struct ceph_cap_snap,
2632 cap_flush);
2633 doutc(cl, "%p %llx.%llx capsnap %p tid %llu %s\n",
2634 inode, ceph_vinop(inode), capsnap, cf->tid,
2635 ceph_cap_string(capsnap->dirty));
2636
2637 refcount_inc(&capsnap->nref);
2638 spin_unlock(&ci->i_ceph_lock);
2639
2640 ret = __send_flush_snap(inode, session, capsnap, cap->mseq,
2641 oldest_flush_tid);
2642 if (ret < 0) {
2643 pr_err_client(cl, "error sending cap flushsnap,"
2644 " %p %llx.%llx tid %llu follows %llu\n",
2645 inode, ceph_vinop(inode), cf->tid,
2646 capsnap->follows);
2647 }
2648
2649 ceph_put_cap_snap(capsnap);
2650 }
2651
2652 spin_lock(&ci->i_ceph_lock);
2653 }
2654 }
2655
2656 void ceph_early_kick_flushing_caps(struct ceph_mds_client *mdsc,
2657 struct ceph_mds_session *session)
2658 {
2659 struct ceph_client *cl = mdsc->fsc->client;
2660 struct ceph_inode_info *ci;
2661 struct ceph_cap *cap;
2662 u64 oldest_flush_tid;
2663
2664 doutc(cl, "mds%d\n", session->s_mds);
2665
2666 spin_lock(&mdsc->cap_dirty_lock);
2667 oldest_flush_tid = __get_oldest_flush_tid(mdsc);
2668 spin_unlock(&mdsc->cap_dirty_lock);
2669
2670 list_for_each_entry(ci, &session->s_cap_flushing, i_flushing_item) {
2671 struct inode *inode = &ci->netfs.inode;
2672
2673 spin_lock(&ci->i_ceph_lock);
2674 cap = ci->i_auth_cap;
2675 if (!(cap && cap->session == session)) {
2676 pr_err_client(cl, "%p %llx.%llx auth cap %p not mds%d ???\n",
2677 inode, ceph_vinop(inode), cap,
2678 session->s_mds);
2679 spin_unlock(&ci->i_ceph_lock);
2680 continue;
2681 }
2682
2683
2684 /*
2685 * if flushing caps were revoked, we re-send the cap flush
2686 * in client reconnect stage. This guarantees MDS * processes
2687 * the cap flush message before issuing the flushing caps to
2688 * other client.
2689 */
2690 if ((cap->issued & ci->i_flushing_caps) !=
2691 ci->i_flushing_caps) {
2692 /* encode_caps_cb() also will reset these sequence
2693 * numbers. make sure sequence numbers in cap flush
2694 * message match later reconnect message */
2695 cap->seq = 0;
2696 cap->issue_seq = 0;
2697 cap->mseq = 0;
2698 __kick_flushing_caps(mdsc, session, ci,
2699 oldest_flush_tid);
2700 } else {
2701 ci->i_ceph_flags |= CEPH_I_KICK_FLUSH;
2702 }
2703
2704 spin_unlock(&ci->i_ceph_lock);
2705 }
2706 }
2707
2708 void ceph_kick_flushing_caps(struct ceph_mds_client *mdsc,
2709 struct ceph_mds_session *session)
2710 {
2711 struct ceph_client *cl = mdsc->fsc->client;
2712 struct ceph_inode_info *ci;
2713 struct ceph_cap *cap;
2714 u64 oldest_flush_tid;
2715
2716 lockdep_assert_held(&session->s_mutex);
2717
2718 doutc(cl, "mds%d\n", session->s_mds);
2719
2720 spin_lock(&mdsc->cap_dirty_lock);
2721 oldest_flush_tid = __get_oldest_flush_tid(mdsc);
2722 spin_unlock(&mdsc->cap_dirty_lock);
2723
2724 list_for_each_entry(ci, &session->s_cap_flushing, i_flushing_item) {
2725 struct inode *inode = &ci->netfs.inode;
2726
2727 spin_lock(&ci->i_ceph_lock);
2728 cap = ci->i_auth_cap;
2729 if (!(cap && cap->session == session)) {
2730 pr_err_client(cl, "%p %llx.%llx auth cap %p not mds%d ???\n",
2731 inode, ceph_vinop(inode), cap,
2732 session->s_mds);
2733 spin_unlock(&ci->i_ceph_lock);
2734 continue;
2735 }
2736 if (ci->i_ceph_flags & CEPH_I_KICK_FLUSH) {
2737 __kick_flushing_caps(mdsc, session, ci,
2738 oldest_flush_tid);
2739 }
2740 spin_unlock(&ci->i_ceph_lock);
2741 }
2742 }
2743
2744 void ceph_kick_flushing_inode_caps(struct ceph_mds_session *session,
2745 struct ceph_inode_info *ci)
2746 {
2747 struct ceph_mds_client *mdsc = session->s_mdsc;
2748 struct ceph_cap *cap = ci->i_auth_cap;
2749 struct inode *inode = &ci->netfs.inode;
2750
2751 lockdep_assert_held(&ci->i_ceph_lock);
2752
2753 doutc(mdsc->fsc->client, "%p %llx.%llx flushing %s\n",
2754 inode, ceph_vinop(inode),
2755 ceph_cap_string(ci->i_flushing_caps));
2756
2757 if (!list_empty(&ci->i_cap_flush_list)) {
2758 u64 oldest_flush_tid;
2759 spin_lock(&mdsc->cap_dirty_lock);
2760 list_move_tail(&ci->i_flushing_item,
2761 &cap->session->s_cap_flushing);
2762 oldest_flush_tid = __get_oldest_flush_tid(mdsc);
2763 spin_unlock(&mdsc->cap_dirty_lock);
2764
2765 __kick_flushing_caps(mdsc, session, ci, oldest_flush_tid);
2766 }
2767 }
2768
2769
2770 /*
2771 * Take references to capabilities we hold, so that we don't release
2772 * them to the MDS prematurely.
2773 */
2774 void ceph_take_cap_refs(struct ceph_inode_info *ci, int got,
2775 bool snap_rwsem_locked)
2776 {
2777 struct inode *inode = &ci->netfs.inode;
2778 struct ceph_client *cl = ceph_inode_to_client(inode);
2779
2780 lockdep_assert_held(&ci->i_ceph_lock);
2781
2782 if (got & CEPH_CAP_PIN)
2783 ci->i_pin_ref++;
2784 if (got & CEPH_CAP_FILE_RD)
2785 ci->i_rd_ref++;
2786 if (got & CEPH_CAP_FILE_CACHE)
2787 ci->i_rdcache_ref++;
2788 if (got & CEPH_CAP_FILE_EXCL)
2789 ci->i_fx_ref++;
2790 if (got & CEPH_CAP_FILE_WR) {
2791 if (ci->i_wr_ref == 0 && !ci->i_head_snapc) {
2792 BUG_ON(!snap_rwsem_locked);
2793 ci->i_head_snapc = ceph_get_snap_context(
2794 ci->i_snap_realm->cached_context);
2795 }
2796 ci->i_wr_ref++;
2797 }
2798 if (got & CEPH_CAP_FILE_BUFFER) {
2799 if (ci->i_wb_ref == 0)
2800 ihold(inode);
2801 ci->i_wb_ref++;
2802 doutc(cl, "%p %llx.%llx wb %d -> %d (?)\n", inode,
2803 ceph_vinop(inode), ci->i_wb_ref-1, ci->i_wb_ref);
2804 }
2805 }
2806
2807 /*
2808 * Try to grab cap references. Specify those refs we @want, and the
2809 * minimal set we @need. Also include the larger offset we are writing
2810 * to (when applicable), and check against max_size here as well.
2811 * Note that caller is responsible for ensuring max_size increases are
2812 * requested from the MDS.
2813 *
2814 * Returns 0 if caps were not able to be acquired (yet), 1 if succeed,
2815 * or a negative error code. There are 3 speical error codes:
2816 * -EAGAIN: need to sleep but non-blocking is specified
2817 * -EFBIG: ask caller to call check_max_size() and try again.
2818 * -EUCLEAN: ask caller to call ceph_renew_caps() and try again.
2819 */
2820 enum {
2821 /* first 8 bits are reserved for CEPH_FILE_MODE_FOO */
2822 NON_BLOCKING = (1 << 8),
2823 CHECK_FILELOCK = (1 << 9),
2824 };
2825
2826 static int try_get_cap_refs(struct inode *inode, int need, int want,
2827 loff_t endoff, int flags, int *got)
2828 {
2829 struct ceph_inode_info *ci = ceph_inode(inode);
2830 struct ceph_mds_client *mdsc = ceph_inode_to_fs_client(inode)->mdsc;
2831 struct ceph_client *cl = ceph_inode_to_client(inode);
2832 int ret = 0;
2833 int have, implemented;
2834 bool snap_rwsem_locked = false;
2835
2836 doutc(cl, "%p %llx.%llx need %s want %s\n", inode,
2837 ceph_vinop(inode), ceph_cap_string(need),
2838 ceph_cap_string(want));
2839
2840 again:
2841 spin_lock(&ci->i_ceph_lock);
2842
2843 if ((flags & CHECK_FILELOCK) &&
2844 (ci->i_ceph_flags & CEPH_I_ERROR_FILELOCK)) {
2845 doutc(cl, "%p %llx.%llx error filelock\n", inode,
2846 ceph_vinop(inode));
2847 ret = -EIO;
2848 goto out_unlock;
2849 }
2850
2851 /* finish pending truncate */
2852 while (ci->i_truncate_pending) {
2853 spin_unlock(&ci->i_ceph_lock);
2854 if (snap_rwsem_locked) {
2855 up_read(&mdsc->snap_rwsem);
2856 snap_rwsem_locked = false;
2857 }
2858 __ceph_do_pending_vmtruncate(inode);
2859 spin_lock(&ci->i_ceph_lock);
2860 }
2861
2862 have = __ceph_caps_issued(ci, &implemented);
2863
2864 if (have & need & CEPH_CAP_FILE_WR) {
2865 if (endoff >= 0 && endoff > (loff_t)ci->i_max_size) {
2866 doutc(cl, "%p %llx.%llx endoff %llu > maxsize %llu\n",
2867 inode, ceph_vinop(inode), endoff, ci->i_max_size);
2868 if (endoff > ci->i_requested_max_size)
2869 ret = ci->i_auth_cap ? -EFBIG : -EUCLEAN;
2870 goto out_unlock;
2871 }
2872 /*
2873 * If a sync write is in progress, we must wait, so that we
2874 * can get a final snapshot value for size+mtime.
2875 */
2876 if (__ceph_have_pending_cap_snap(ci)) {
2877 doutc(cl, "%p %llx.%llx cap_snap_pending\n", inode,
2878 ceph_vinop(inode));
2879 goto out_unlock;
2880 }
2881 }
2882
2883 if ((have & need) == need) {
2884 /*
2885 * Look at (implemented & ~have & not) so that we keep waiting
2886 * on transition from wanted -> needed caps. This is needed
2887 * for WRBUFFER|WR -> WR to avoid a new WR sync write from
2888 * going before a prior buffered writeback happens.
2889 *
2890 * For RDCACHE|RD -> RD, there is not need to wait and we can
2891 * just exclude the revoking caps and force to sync read.
2892 */
2893 int not = want & ~(have & need);
2894 int revoking = implemented & ~have;
2895 int exclude = revoking & not;
2896 doutc(cl, "%p %llx.%llx have %s but not %s (revoking %s)\n",
2897 inode, ceph_vinop(inode), ceph_cap_string(have),
2898 ceph_cap_string(not), ceph_cap_string(revoking));
2899 if (!exclude || !(exclude & CEPH_CAP_FILE_BUFFER)) {
2900 if (!snap_rwsem_locked &&
2901 !ci->i_head_snapc &&
2902 (need & CEPH_CAP_FILE_WR)) {
2903 if (!down_read_trylock(&mdsc->snap_rwsem)) {
2904 /*
2905 * we can not call down_read() when
2906 * task isn't in TASK_RUNNING state
2907 */
2908 if (flags & NON_BLOCKING) {
2909 ret = -EAGAIN;
2910 goto out_unlock;
2911 }
2912
2913 spin_unlock(&ci->i_ceph_lock);
2914 down_read(&mdsc->snap_rwsem);
2915 snap_rwsem_locked = true;
2916 goto again;
2917 }
2918 snap_rwsem_locked = true;
2919 }
2920 if ((have & want) == want)
2921 *got = need | (want & ~exclude);
2922 else
2923 *got = need;
2924 ceph_take_cap_refs(ci, *got, true);
2925 ret = 1;
2926 }
2927 } else {
2928 int session_readonly = false;
2929 int mds_wanted;
2930 if (ci->i_auth_cap &&
2931 (need & (CEPH_CAP_FILE_WR | CEPH_CAP_FILE_EXCL))) {
2932 struct ceph_mds_session *s = ci->i_auth_cap->session;
2933 spin_lock(&s->s_cap_lock);
2934 session_readonly = s->s_readonly;
2935 spin_unlock(&s->s_cap_lock);
2936 }
2937 if (session_readonly) {
2938 doutc(cl, "%p %llx.%llx need %s but mds%d readonly\n",
2939 inode, ceph_vinop(inode), ceph_cap_string(need),
2940 ci->i_auth_cap->mds);
2941 ret = -EROFS;
2942 goto out_unlock;
2943 }
2944
2945 if (ceph_inode_is_shutdown(inode)) {
2946 doutc(cl, "%p %llx.%llx inode is shutdown\n",
2947 inode, ceph_vinop(inode));
2948 ret = -ESTALE;
2949 goto out_unlock;
2950 }
2951 mds_wanted = __ceph_caps_mds_wanted(ci, false);
2952 if (need & ~mds_wanted) {
2953 doutc(cl, "%p %llx.%llx need %s > mds_wanted %s\n",
2954 inode, ceph_vinop(inode), ceph_cap_string(need),
2955 ceph_cap_string(mds_wanted));
2956 ret = -EUCLEAN;
2957 goto out_unlock;
2958 }
2959
2960 doutc(cl, "%p %llx.%llx have %s need %s\n", inode,
2961 ceph_vinop(inode), ceph_cap_string(have),
2962 ceph_cap_string(need));
2963 }
2964 out_unlock:
2965
2966 __ceph_touch_fmode(ci, mdsc, flags);
2967
2968 spin_unlock(&ci->i_ceph_lock);
2969 if (snap_rwsem_locked)
2970 up_read(&mdsc->snap_rwsem);
2971
2972 if (!ret)
2973 ceph_update_cap_mis(&mdsc->metric);
2974 else if (ret == 1)
2975 ceph_update_cap_hit(&mdsc->metric);
2976
2977 doutc(cl, "%p %llx.%llx ret %d got %s\n", inode,
2978 ceph_vinop(inode), ret, ceph_cap_string(*got));
2979 return ret;
2980 }
2981
2982 /*
2983 * Check the offset we are writing up to against our current
2984 * max_size. If necessary, tell the MDS we want to write to
2985 * a larger offset.
2986 */
2987 static void check_max_size(struct inode *inode, loff_t endoff)
2988 {
2989 struct ceph_inode_info *ci = ceph_inode(inode);
2990 struct ceph_client *cl = ceph_inode_to_client(inode);
2991 int check = 0;
2992
2993 /* do we need to explicitly request a larger max_size? */
2994 spin_lock(&ci->i_ceph_lock);
2995 if (endoff >= ci->i_max_size && endoff > ci->i_wanted_max_size) {
2996 doutc(cl, "write %p %llx.%llx at large endoff %llu, req max_size\n",
2997 inode, ceph_vinop(inode), endoff);
2998 ci->i_wanted_max_size = endoff;
2999 }
3000 /* duplicate ceph_check_caps()'s logic */
3001 if (ci->i_auth_cap &&
3002 (ci->i_auth_cap->issued & CEPH_CAP_FILE_WR) &&
3003 ci->i_wanted_max_size > ci->i_max_size &&
3004 ci->i_wanted_max_size > ci->i_requested_max_size)
3005 check = 1;
3006 spin_unlock(&ci->i_ceph_lock);
3007 if (check)
3008 ceph_check_caps(ci, CHECK_CAPS_AUTHONLY);
3009 }
3010
3011 static inline int get_used_fmode(int caps)
3012 {
3013 int fmode = 0;
3014 if (caps & CEPH_CAP_FILE_RD)
3015 fmode |= CEPH_FILE_MODE_RD;
3016 if (caps & CEPH_CAP_FILE_WR)
3017 fmode |= CEPH_FILE_MODE_WR;
3018 return fmode;
3019 }
3020
3021 int ceph_try_get_caps(struct inode *inode, int need, int want,
3022 bool nonblock, int *got)
3023 {
3024 int ret, flags;
3025
3026 BUG_ON(need & ~CEPH_CAP_FILE_RD);
3027 BUG_ON(want & ~(CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO |
3028 CEPH_CAP_FILE_SHARED | CEPH_CAP_FILE_EXCL |
3029 CEPH_CAP_ANY_DIR_OPS));
3030 if (need) {
3031 ret = ceph_pool_perm_check(inode, need);
3032 if (ret < 0)
3033 return ret;
3034 }
3035
3036 flags = get_used_fmode(need | want);
3037 if (nonblock)
3038 flags |= NON_BLOCKING;
3039
3040 ret = try_get_cap_refs(inode, need, want, 0, flags, got);
3041 /* three special error codes */
3042 if (ret == -EAGAIN || ret == -EFBIG || ret == -EUCLEAN)
3043 ret = 0;
3044 return ret;
3045 }
3046
3047 /*
3048 * Wait for caps, and take cap references. If we can't get a WR cap
3049 * due to a small max_size, make sure we check_max_size (and possibly
3050 * ask the mds) so we don't get hung up indefinitely.
3051 */
3052 int __ceph_get_caps(struct inode *inode, struct ceph_file_info *fi, int need,
3053 int want, loff_t endoff, int *got)
3054 {
3055 struct ceph_inode_info *ci = ceph_inode(inode);
3056 struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
3057 int ret, _got, flags;
3058
3059 ret = ceph_pool_perm_check(inode, need);
3060 if (ret < 0)
3061 return ret;
3062
3063 if (fi && (fi->fmode & CEPH_FILE_MODE_WR) &&
3064 fi->filp_gen != READ_ONCE(fsc->filp_gen))
3065 return -EBADF;
3066
3067 flags = get_used_fmode(need | want);
3068
3069 while (true) {
3070 flags &= CEPH_FILE_MODE_MASK;
3071 if (vfs_inode_has_locks(inode))
3072 flags |= CHECK_FILELOCK;
3073 _got = 0;
3074 ret = try_get_cap_refs(inode, need, want, endoff,
3075 flags, &_got);
3076 WARN_ON_ONCE(ret == -EAGAIN);
3077 if (!ret) {
3078 #ifdef CONFIG_DEBUG_FS
3079 struct ceph_mds_client *mdsc = fsc->mdsc;
3080 struct cap_wait cw;
3081 #endif
3082 DEFINE_WAIT_FUNC(wait, woken_wake_function);
3083
3084 #ifdef CONFIG_DEBUG_FS
3085 cw.ino = ceph_ino(inode);
3086 cw.tgid = current->tgid;
3087 cw.need = need;
3088 cw.want = want;
3089
3090 spin_lock(&mdsc->caps_list_lock);
3091 list_add(&cw.list, &mdsc->cap_wait_list);
3092 spin_unlock(&mdsc->caps_list_lock);
3093 #endif
3094
3095 /* make sure used fmode not timeout */
3096 ceph_get_fmode(ci, flags, FMODE_WAIT_BIAS);
3097 add_wait_queue(&ci->i_cap_wq, &wait);
3098
3099 flags |= NON_BLOCKING;
3100 while (!(ret = try_get_cap_refs(inode, need, want,
3101 endoff, flags, &_got))) {
3102 if (signal_pending(current)) {
3103 ret = -ERESTARTSYS;
3104 break;
3105 }
3106 wait_woken(&wait, TASK_INTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);
3107 }
3108
3109 remove_wait_queue(&ci->i_cap_wq, &wait);
3110 ceph_put_fmode(ci, flags, FMODE_WAIT_BIAS);
3111
3112 #ifdef CONFIG_DEBUG_FS
3113 spin_lock(&mdsc->caps_list_lock);
3114 list_del(&cw.list);
3115 spin_unlock(&mdsc->caps_list_lock);
3116 #endif
3117
3118 if (ret == -EAGAIN)
3119 continue;
3120 }
3121
3122 if (fi && (fi->fmode & CEPH_FILE_MODE_WR) &&
3123 fi->filp_gen != READ_ONCE(fsc->filp_gen)) {
3124 if (ret >= 0 && _got)
3125 ceph_put_cap_refs(ci, _got);
3126 return -EBADF;
3127 }
3128
3129 if (ret < 0) {
3130 if (ret == -EFBIG || ret == -EUCLEAN) {
3131 int ret2 = ceph_wait_on_async_create(inode);
3132 if (ret2 < 0)
3133 return ret2;
3134 }
3135 if (ret == -EFBIG) {
3136 check_max_size(inode, endoff);
3137 continue;
3138 }
3139 if (ret == -EUCLEAN) {
3140 /* session was killed, try renew caps */
3141 ret = ceph_renew_caps(inode, flags);
3142 if (ret == 0)
3143 continue;
3144 }
3145 return ret;
3146 }
3147
3148 if (S_ISREG(ci->netfs.inode.i_mode) &&
3149 ceph_has_inline_data(ci) &&
3150 (_got & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) &&
3151 i_size_read(inode) > 0) {
3152 struct page *page =
3153 find_get_page(inode->i_mapping, 0);
3154 if (page) {
3155 bool uptodate = PageUptodate(page);
3156
3157 put_page(page);
3158 if (uptodate)
3159 break;
3160 }
3161 /*
3162 * drop cap refs first because getattr while
3163 * holding * caps refs can cause deadlock.
3164 */
3165 ceph_put_cap_refs(ci, _got);
3166 _got = 0;
3167
3168 /*
3169 * getattr request will bring inline data into
3170 * page cache
3171 */
3172 ret = __ceph_do_getattr(inode, NULL,
3173 CEPH_STAT_CAP_INLINE_DATA,
3174 true);
3175 if (ret < 0)
3176 return ret;
3177 continue;
3178 }
3179 break;
3180 }
3181 *got = _got;
3182 return 0;
3183 }
3184
3185 int ceph_get_caps(struct file *filp, int need, int want, loff_t endoff,
3186 int *got)
3187 {
3188 struct ceph_file_info *fi = filp->private_data;
3189 struct inode *inode = file_inode(filp);
3190
3191 return __ceph_get_caps(inode, fi, need, want, endoff, got);
3192 }
3193
3194 /*
3195 * Take cap refs. Caller must already know we hold at least one ref
3196 * on the caps in question or we don't know this is safe.
3197 */
3198 void ceph_get_cap_refs(struct ceph_inode_info *ci, int caps)
3199 {
3200 spin_lock(&ci->i_ceph_lock);
3201 ceph_take_cap_refs(ci, caps, false);
3202 spin_unlock(&ci->i_ceph_lock);
3203 }
3204
3205
3206 /*
3207 * drop cap_snap that is not associated with any snapshot.
3208 * we don't need to send FLUSHSNAP message for it.
3209 */
3210 static int ceph_try_drop_cap_snap(struct ceph_inode_info *ci,
3211 struct ceph_cap_snap *capsnap)
3212 {
3213 struct inode *inode = &ci->netfs.inode;
3214 struct ceph_client *cl = ceph_inode_to_client(inode);
3215
3216 if (!capsnap->need_flush &&
3217 !capsnap->writing && !capsnap->dirty_pages) {
3218 doutc(cl, "%p follows %llu\n", capsnap, capsnap->follows);
3219 BUG_ON(capsnap->cap_flush.tid > 0);
3220 ceph_put_snap_context(capsnap->context);
3221 if (!list_is_last(&capsnap->ci_item, &ci->i_cap_snaps))
3222 ci->i_ceph_flags |= CEPH_I_FLUSH_SNAPS;
3223
3224 list_del(&capsnap->ci_item);
3225 ceph_put_cap_snap(capsnap);
3226 return 1;
3227 }
3228 return 0;
3229 }
3230
3231 enum put_cap_refs_mode {
3232 PUT_CAP_REFS_SYNC = 0,
3233 PUT_CAP_REFS_ASYNC,
3234 };
3235
3236 /*
3237 * Release cap refs.
3238 *
3239 * If we released the last ref on any given cap, call ceph_check_caps
3240 * to release (or schedule a release).
3241 *
3242 * If we are releasing a WR cap (from a sync write), finalize any affected
3243 * cap_snap, and wake up any waiters.
3244 */
3245 static void __ceph_put_cap_refs(struct ceph_inode_info *ci, int had,
3246 enum put_cap_refs_mode mode)
3247 {
3248 struct inode *inode = &ci->netfs.inode;
3249 struct ceph_client *cl = ceph_inode_to_client(inode);
3250 int last = 0, put = 0, flushsnaps = 0, wake = 0;
3251 bool check_flushsnaps = false;
3252
3253 spin_lock(&ci->i_ceph_lock);
3254 if (had & CEPH_CAP_PIN)
3255 --ci->i_pin_ref;
3256 if (had & CEPH_CAP_FILE_RD)
3257 if (--ci->i_rd_ref == 0)
3258 last++;
3259 if (had & CEPH_CAP_FILE_CACHE)
3260 if (--ci->i_rdcache_ref == 0)
3261 last++;
3262 if (had & CEPH_CAP_FILE_EXCL)
3263 if (--ci->i_fx_ref == 0)
3264 last++;
3265 if (had & CEPH_CAP_FILE_BUFFER) {
3266 if (--ci->i_wb_ref == 0) {
3267 last++;
3268 /* put the ref held by ceph_take_cap_refs() */
3269 put++;
3270 check_flushsnaps = true;
3271 }
3272 doutc(cl, "%p %llx.%llx wb %d -> %d (?)\n", inode,
3273 ceph_vinop(inode), ci->i_wb_ref+1, ci->i_wb_ref);
3274 }
3275 if (had & CEPH_CAP_FILE_WR) {
3276 if (--ci->i_wr_ref == 0) {
3277 /*
3278 * The Fb caps will always be took and released
3279 * together with the Fw caps.
3280 */
3281 WARN_ON_ONCE(ci->i_wb_ref);
3282
3283 last++;
3284 check_flushsnaps = true;
3285 if (ci->i_wrbuffer_ref_head == 0 &&
3286 ci->i_dirty_caps == 0 &&
3287 ci->i_flushing_caps == 0) {
3288 BUG_ON(!ci->i_head_snapc);
3289 ceph_put_snap_context(ci->i_head_snapc);
3290 ci->i_head_snapc = NULL;
3291 }
3292 /* see comment in __ceph_remove_cap() */
3293 if (!__ceph_is_any_real_caps(ci) && ci->i_snap_realm)
3294 ceph_change_snap_realm(inode, NULL);
3295 }
3296 }
3297 if (check_flushsnaps && __ceph_have_pending_cap_snap(ci)) {
3298 struct ceph_cap_snap *capsnap =
3299 list_last_entry(&ci->i_cap_snaps,
3300 struct ceph_cap_snap,
3301 ci_item);
3302
3303 capsnap->writing = 0;
3304 if (ceph_try_drop_cap_snap(ci, capsnap))
3305 /* put the ref held by ceph_queue_cap_snap() */
3306 put++;
3307 else if (__ceph_finish_cap_snap(ci, capsnap))
3308 flushsnaps = 1;
3309 wake = 1;
3310 }
3311 spin_unlock(&ci->i_ceph_lock);
3312
3313 doutc(cl, "%p %llx.%llx had %s%s%s\n", inode, ceph_vinop(inode),
3314 ceph_cap_string(had), last ? " last" : "", put ? " put" : "");
3315
3316 switch (mode) {
3317 case PUT_CAP_REFS_SYNC:
3318 if (last)
3319 ceph_check_caps(ci, 0);
3320 else if (flushsnaps)
3321 ceph_flush_snaps(ci, NULL);
3322 break;
3323 case PUT_CAP_REFS_ASYNC:
3324 if (last)
3325 ceph_queue_check_caps(inode);
3326 else if (flushsnaps)
3327 ceph_queue_flush_snaps(inode);
3328 break;
3329 default:
3330 break;
3331 }
3332 if (wake)
3333 wake_up_all(&ci->i_cap_wq);
3334 while (put-- > 0)
3335 iput(inode);
3336 }
3337
3338 void ceph_put_cap_refs(struct ceph_inode_info *ci, int had)
3339 {
3340 __ceph_put_cap_refs(ci, had, PUT_CAP_REFS_SYNC);
3341 }
3342
3343 void ceph_put_cap_refs_async(struct ceph_inode_info *ci, int had)
3344 {
3345 __ceph_put_cap_refs(ci, had, PUT_CAP_REFS_ASYNC);
3346 }
3347
3348 /*
3349 * Release @nr WRBUFFER refs on dirty pages for the given @snapc snap
3350 * context. Adjust per-snap dirty page accounting as appropriate.
3351 * Once all dirty data for a cap_snap is flushed, flush snapped file
3352 * metadata back to the MDS. If we dropped the last ref, call
3353 * ceph_check_caps.
3354 */
3355 void ceph_put_wrbuffer_cap_refs(struct ceph_inode_info *ci, int nr,
3356 struct ceph_snap_context *snapc)
3357 {
3358 struct inode *inode = &ci->netfs.inode;
3359 struct ceph_client *cl = ceph_inode_to_client(inode);
3360 struct ceph_cap_snap *capsnap = NULL, *iter;
3361 int put = 0;
3362 bool last = false;
3363 bool flush_snaps = false;
3364 bool complete_capsnap = false;
3365
3366 spin_lock(&ci->i_ceph_lock);
3367 ci->i_wrbuffer_ref -= nr;
3368 if (ci->i_wrbuffer_ref == 0) {
3369 last = true;
3370 put++;
3371 }
3372
3373 if (ci->i_head_snapc == snapc) {
3374 ci->i_wrbuffer_ref_head -= nr;
3375 if (ci->i_wrbuffer_ref_head == 0 &&
3376 ci->i_wr_ref == 0 &&
3377 ci->i_dirty_caps == 0 &&
3378 ci->i_flushing_caps == 0) {
3379 BUG_ON(!ci->i_head_snapc);
3380 ceph_put_snap_context(ci->i_head_snapc);
3381 ci->i_head_snapc = NULL;
3382 }
3383 doutc(cl, "on %p %llx.%llx head %d/%d -> %d/%d %s\n",
3384 inode, ceph_vinop(inode), ci->i_wrbuffer_ref+nr,
3385 ci->i_wrbuffer_ref_head+nr, ci->i_wrbuffer_ref,
3386 ci->i_wrbuffer_ref_head, last ? " LAST" : "");
3387 } else {
3388 list_for_each_entry(iter, &ci->i_cap_snaps, ci_item) {
3389 if (iter->context == snapc) {
3390 capsnap = iter;
3391 break;
3392 }
3393 }
3394
3395 if (!capsnap) {
3396 /*
3397 * The capsnap should already be removed when removing
3398 * auth cap in the case of a forced unmount.
3399 */
3400 WARN_ON_ONCE(ci->i_auth_cap);
3401 goto unlock;
3402 }
3403
3404 capsnap->dirty_pages -= nr;
3405 if (capsnap->dirty_pages == 0) {
3406 complete_capsnap = true;
3407 if (!capsnap->writing) {
3408 if (ceph_try_drop_cap_snap(ci, capsnap)) {
3409 put++;
3410 } else {
3411 ci->i_ceph_flags |= CEPH_I_FLUSH_SNAPS;
3412 flush_snaps = true;
3413 }
3414 }
3415 }
3416 doutc(cl, "%p %llx.%llx cap_snap %p snap %lld %d/%d -> %d/%d %s%s\n",
3417 inode, ceph_vinop(inode), capsnap, capsnap->context->seq,
3418 ci->i_wrbuffer_ref+nr, capsnap->dirty_pages + nr,
3419 ci->i_wrbuffer_ref, capsnap->dirty_pages,
3420 last ? " (wrbuffer last)" : "",
3421 complete_capsnap ? " (complete capsnap)" : "");
3422 }
3423
3424 unlock:
3425 spin_unlock(&ci->i_ceph_lock);
3426
3427 if (last) {
3428 ceph_check_caps(ci, 0);
3429 } else if (flush_snaps) {
3430 ceph_flush_snaps(ci, NULL);
3431 }
3432 if (complete_capsnap)
3433 wake_up_all(&ci->i_cap_wq);
3434 while (put-- > 0) {
3435 iput(inode);
3436 }
3437 }
3438
3439 /*
3440 * Invalidate unlinked inode's aliases, so we can drop the inode ASAP.
3441 */
3442 static void invalidate_aliases(struct inode *inode)
3443 {
3444 struct ceph_client *cl = ceph_inode_to_client(inode);
3445 struct dentry *dn, *prev = NULL;
3446
3447 doutc(cl, "%p %llx.%llx\n", inode, ceph_vinop(inode));
3448 d_prune_aliases(inode);
3449 /*
3450 * For non-directory inode, d_find_alias() only returns
3451 * hashed dentry. After calling d_invalidate(), the
3452 * dentry becomes unhashed.
3453 *
3454 * For directory inode, d_find_alias() can return
3455 * unhashed dentry. But directory inode should have
3456 * one alias at most.
3457 */
3458 while ((dn = d_find_alias(inode))) {
3459 if (dn == prev) {
3460 dput(dn);
3461 break;
3462 }
3463 d_invalidate(dn);
3464 if (prev)
3465 dput(prev);
3466 prev = dn;
3467 }
3468 if (prev)
3469 dput(prev);
3470 }
3471
3472 struct cap_extra_info {
3473 struct ceph_string *pool_ns;
3474 /* inline data */
3475 u64 inline_version;
3476 void *inline_data;
3477 u32 inline_len;
3478 /* dirstat */
3479 bool dirstat_valid;
3480 u64 nfiles;
3481 u64 nsubdirs;
3482 u64 change_attr;
3483 /* currently issued */
3484 int issued;
3485 struct timespec64 btime;
3486 u8 *fscrypt_auth;
3487 u32 fscrypt_auth_len;
3488 u64 fscrypt_file_size;
3489 };
3490
3491 /*
3492 * Handle a cap GRANT message from the MDS. (Note that a GRANT may
3493 * actually be a revocation if it specifies a smaller cap set.)
3494 *
3495 * caller holds s_mutex and i_ceph_lock, we drop both.
3496 */
3497 static void handle_cap_grant(struct inode *inode,
3498 struct ceph_mds_session *session,
3499 struct ceph_cap *cap,
3500 struct ceph_mds_caps *grant,
3501 struct ceph_buffer *xattr_buf,
3502 struct cap_extra_info *extra_info)
3503 __releases(ci->i_ceph_lock)
3504 __releases(session->s_mdsc->snap_rwsem)
3505 {
3506 struct ceph_client *cl = ceph_inode_to_client(inode);
3507 struct ceph_inode_info *ci = ceph_inode(inode);
3508 int seq = le32_to_cpu(grant->seq);
3509 int newcaps = le32_to_cpu(grant->caps);
3510 int used, wanted, dirty;
3511 u64 size = le64_to_cpu(grant->size);
3512 u64 max_size = le64_to_cpu(grant->max_size);
3513 unsigned char check_caps = 0;
3514 bool was_stale = cap->cap_gen < atomic_read(&session->s_cap_gen);
3515 bool wake = false;
3516 bool writeback = false;
3517 bool queue_trunc = false;
3518 bool queue_invalidate = false;
3519 bool deleted_inode = false;
3520 bool fill_inline = false;
3521 bool revoke_wait = false;
3522 int flags = 0;
3523
3524 /*
3525 * If there is at least one crypto block then we'll trust
3526 * fscrypt_file_size. If the real length of the file is 0, then
3527 * ignore it (it has probably been truncated down to 0 by the MDS).
3528 */
3529 if (IS_ENCRYPTED(inode) && size)
3530 size = extra_info->fscrypt_file_size;
3531
3532 doutc(cl, "%p %llx.%llx cap %p mds%d seq %d %s\n", inode,
3533 ceph_vinop(inode), cap, session->s_mds, seq,
3534 ceph_cap_string(newcaps));
3535 doutc(cl, " size %llu max_size %llu, i_size %llu\n", size,
3536 max_size, i_size_read(inode));
3537
3538
3539 /*
3540 * If CACHE is being revoked, and we have no dirty buffers,
3541 * try to invalidate (once). (If there are dirty buffers, we
3542 * will invalidate _after_ writeback.)
3543 */
3544 if (S_ISREG(inode->i_mode) && /* don't invalidate readdir cache */
3545 ((cap->issued & ~newcaps) & CEPH_CAP_FILE_CACHE) &&
3546 (newcaps & CEPH_CAP_FILE_LAZYIO) == 0 &&
3547 !(ci->i_wrbuffer_ref || ci->i_wb_ref)) {
3548 if (try_nonblocking_invalidate(inode)) {
3549 /* there were locked pages.. invalidate later
3550 in a separate thread. */
3551 if (ci->i_rdcache_revoking != ci->i_rdcache_gen) {
3552 queue_invalidate = true;
3553 ci->i_rdcache_revoking = ci->i_rdcache_gen;
3554 }
3555 }
3556 }
3557
3558 if (was_stale)
3559 cap->issued = cap->implemented = CEPH_CAP_PIN;
3560
3561 /*
3562 * auth mds of the inode changed. we received the cap export message,
3563 * but still haven't received the cap import message. handle_cap_export
3564 * updated the new auth MDS' cap.
3565 *
3566 * "ceph_seq_cmp(seq, cap->seq) <= 0" means we are processing a message
3567 * that was sent before the cap import message. So don't remove caps.
3568 */
3569 if (ceph_seq_cmp(seq, cap->seq) <= 0) {
3570 WARN_ON(cap != ci->i_auth_cap);
3571 WARN_ON(cap->cap_id != le64_to_cpu(grant->cap_id));
3572 seq = cap->seq;
3573 newcaps |= cap->issued;
3574 }
3575
3576 /* side effects now are allowed */
3577 cap->cap_gen = atomic_read(&session->s_cap_gen);
3578 cap->seq = seq;
3579
3580 __check_cap_issue(ci, cap, newcaps);
3581
3582 inode_set_max_iversion_raw(inode, extra_info->change_attr);
3583
3584 if ((newcaps & CEPH_CAP_AUTH_SHARED) &&
3585 (extra_info->issued & CEPH_CAP_AUTH_EXCL) == 0) {
3586 umode_t mode = le32_to_cpu(grant->mode);
3587
3588 if (inode_wrong_type(inode, mode))
3589 pr_warn_once("inode type changed! (ino %llx.%llx is 0%o, mds says 0%o)\n",
3590 ceph_vinop(inode), inode->i_mode, mode);
3591 else
3592 inode->i_mode = mode;
3593 inode->i_uid = make_kuid(&init_user_ns, le32_to_cpu(grant->uid));
3594 inode->i_gid = make_kgid(&init_user_ns, le32_to_cpu(grant->gid));
3595 ci->i_btime = extra_info->btime;
3596 doutc(cl, "%p %llx.%llx mode 0%o uid.gid %d.%d\n", inode,
3597 ceph_vinop(inode), inode->i_mode,
3598 from_kuid(&init_user_ns, inode->i_uid),
3599 from_kgid(&init_user_ns, inode->i_gid));
3600 #if IS_ENABLED(CONFIG_FS_ENCRYPTION)
3601 if (ci->fscrypt_auth_len != extra_info->fscrypt_auth_len ||
3602 memcmp(ci->fscrypt_auth, extra_info->fscrypt_auth,
3603 ci->fscrypt_auth_len))
3604 pr_warn_ratelimited_client(cl,
3605 "cap grant attempt to change fscrypt_auth on non-I_NEW inode (old len %d new len %d)\n",
3606 ci->fscrypt_auth_len,
3607 extra_info->fscrypt_auth_len);
3608 #endif
3609 }
3610
3611 if ((newcaps & CEPH_CAP_LINK_SHARED) &&
3612 (extra_info->issued & CEPH_CAP_LINK_EXCL) == 0) {
3613 set_nlink(inode, le32_to_cpu(grant->nlink));
3614 if (inode->i_nlink == 0)
3615 deleted_inode = true;
3616 }
3617
3618 if ((extra_info->issued & CEPH_CAP_XATTR_EXCL) == 0 &&
3619 grant->xattr_len) {
3620 int len = le32_to_cpu(grant->xattr_len);
3621 u64 version = le64_to_cpu(grant->xattr_version);
3622
3623 if (version > ci->i_xattrs.version) {
3624 doutc(cl, " got new xattrs v%llu on %p %llx.%llx len %d\n",
3625 version, inode, ceph_vinop(inode), len);
3626 if (ci->i_xattrs.blob)
3627 ceph_buffer_put(ci->i_xattrs.blob);
3628 ci->i_xattrs.blob = ceph_buffer_get(xattr_buf);
3629 ci->i_xattrs.version = version;
3630 ceph_forget_all_cached_acls(inode);
3631 ceph_security_invalidate_secctx(inode);
3632 }
3633 }
3634
3635 if (newcaps & CEPH_CAP_ANY_RD) {
3636 struct timespec64 mtime, atime, ctime;
3637 /* ctime/mtime/atime? */
3638 ceph_decode_timespec64(&mtime, &grant->mtime);
3639 ceph_decode_timespec64(&atime, &grant->atime);
3640 ceph_decode_timespec64(&ctime, &grant->ctime);
3641 ceph_fill_file_time(inode, extra_info->issued,
3642 le32_to_cpu(grant->time_warp_seq),
3643 &ctime, &mtime, &atime);
3644 }
3645
3646 if ((newcaps & CEPH_CAP_FILE_SHARED) && extra_info->dirstat_valid) {
3647 ci->i_files = extra_info->nfiles;
3648 ci->i_subdirs = extra_info->nsubdirs;
3649 }
3650
3651 if (newcaps & (CEPH_CAP_ANY_FILE_RD | CEPH_CAP_ANY_FILE_WR)) {
3652 /* file layout may have changed */
3653 s64 old_pool = ci->i_layout.pool_id;
3654 struct ceph_string *old_ns;
3655
3656 ceph_file_layout_from_legacy(&ci->i_layout, &grant->layout);
3657 old_ns = rcu_dereference_protected(ci->i_layout.pool_ns,
3658 lockdep_is_held(&ci->i_ceph_lock));
3659 rcu_assign_pointer(ci->i_layout.pool_ns, extra_info->pool_ns);
3660
3661 if (ci->i_layout.pool_id != old_pool ||
3662 extra_info->pool_ns != old_ns)
3663 ci->i_ceph_flags &= ~CEPH_I_POOL_PERM;
3664
3665 extra_info->pool_ns = old_ns;
3666
3667 /* size/truncate_seq? */
3668 queue_trunc = ceph_fill_file_size(inode, extra_info->issued,
3669 le32_to_cpu(grant->truncate_seq),
3670 le64_to_cpu(grant->truncate_size),
3671 size);
3672 }
3673
3674 if (ci->i_auth_cap == cap && (newcaps & CEPH_CAP_ANY_FILE_WR)) {
3675 if (max_size != ci->i_max_size) {
3676 doutc(cl, "max_size %lld -> %llu\n", ci->i_max_size,
3677 max_size);
3678 ci->i_max_size = max_size;
3679 if (max_size >= ci->i_wanted_max_size) {
3680 ci->i_wanted_max_size = 0; /* reset */
3681 ci->i_requested_max_size = 0;
3682 }
3683 wake = true;
3684 }
3685 }
3686
3687 /* check cap bits */
3688 wanted = __ceph_caps_wanted(ci);
3689 used = __ceph_caps_used(ci);
3690 dirty = __ceph_caps_dirty(ci);
3691 doutc(cl, " my wanted = %s, used = %s, dirty %s\n",
3692 ceph_cap_string(wanted), ceph_cap_string(used),
3693 ceph_cap_string(dirty));
3694
3695 if ((was_stale || le32_to_cpu(grant->op) == CEPH_CAP_OP_IMPORT) &&
3696 (wanted & ~(cap->mds_wanted | newcaps))) {
3697 /*
3698 * If mds is importing cap, prior cap messages that update
3699 * 'wanted' may get dropped by mds (migrate seq mismatch).
3700 *
3701 * We don't send cap message to update 'wanted' if what we
3702 * want are already issued. If mds revokes caps, cap message
3703 * that releases caps also tells mds what we want. But if
3704 * caps got revoked by mds forcedly (session stale). We may
3705 * haven't told mds what we want.
3706 */
3707 check_caps = 1;
3708 }
3709
3710 /* revocation, grant, or no-op? */
3711 if (cap->issued & ~newcaps) {
3712 int revoking = cap->issued & ~newcaps;
3713
3714 doutc(cl, "revocation: %s -> %s (revoking %s)\n",
3715 ceph_cap_string(cap->issued), ceph_cap_string(newcaps),
3716 ceph_cap_string(revoking));
3717 if (S_ISREG(inode->i_mode) &&
3718 (revoking & used & CEPH_CAP_FILE_BUFFER)) {
3719 writeback = true; /* initiate writeback; will delay ack */
3720 revoke_wait = true;
3721 } else if (queue_invalidate &&
3722 revoking == CEPH_CAP_FILE_CACHE &&
3723 (newcaps & CEPH_CAP_FILE_LAZYIO) == 0) {
3724 revoke_wait = true; /* do nothing yet, invalidation will be queued */
3725 } else if (cap == ci->i_auth_cap) {
3726 check_caps = 1; /* check auth cap only */
3727 } else {
3728 check_caps = 2; /* check all caps */
3729 }
3730 /* If there is new caps, try to wake up the waiters */
3731 if (~cap->issued & newcaps)
3732 wake = true;
3733 cap->issued = newcaps;
3734 cap->implemented |= newcaps;
3735 } else if (cap->issued == newcaps) {
3736 doutc(cl, "caps unchanged: %s -> %s\n",
3737 ceph_cap_string(cap->issued),
3738 ceph_cap_string(newcaps));
3739 } else {
3740 doutc(cl, "grant: %s -> %s\n", ceph_cap_string(cap->issued),
3741 ceph_cap_string(newcaps));
3742 /* non-auth MDS is revoking the newly grant caps ? */
3743 if (cap == ci->i_auth_cap &&
3744 __ceph_caps_revoking_other(ci, cap, newcaps))
3745 check_caps = 2;
3746
3747 cap->issued = newcaps;
3748 cap->implemented |= newcaps; /* add bits only, to
3749 * avoid stepping on a
3750 * pending revocation */
3751 wake = true;
3752 }
3753 BUG_ON(cap->issued & ~cap->implemented);
3754
3755 /* don't let check_caps skip sending a response to MDS for revoke msgs */
3756 if (!revoke_wait && le32_to_cpu(grant->op) == CEPH_CAP_OP_REVOKE) {
3757 cap->mds_wanted = 0;
3758 flags |= CHECK_CAPS_FLUSH_FORCE;
3759 if (cap == ci->i_auth_cap)
3760 check_caps = 1; /* check auth cap only */
3761 else
3762 check_caps = 2; /* check all caps */
3763 }
3764
3765 if (extra_info->inline_version > 0 &&
3766 extra_info->inline_version >= ci->i_inline_version) {
3767 ci->i_inline_version = extra_info->inline_version;
3768 if (ci->i_inline_version != CEPH_INLINE_NONE &&
3769 (newcaps & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)))
3770 fill_inline = true;
3771 }
3772
3773 if (le32_to_cpu(grant->op) == CEPH_CAP_OP_IMPORT) {
3774 if (ci->i_auth_cap == cap) {
3775 if (newcaps & ~extra_info->issued)
3776 wake = true;
3777
3778 if (ci->i_requested_max_size > max_size ||
3779 !(le32_to_cpu(grant->wanted) & CEPH_CAP_ANY_FILE_WR)) {
3780 /* re-request max_size if necessary */
3781 ci->i_requested_max_size = 0;
3782 wake = true;
3783 }
3784
3785 ceph_kick_flushing_inode_caps(session, ci);
3786 }
3787 up_read(&session->s_mdsc->snap_rwsem);
3788 }
3789 spin_unlock(&ci->i_ceph_lock);
3790
3791 if (fill_inline)
3792 ceph_fill_inline_data(inode, NULL, extra_info->inline_data,
3793 extra_info->inline_len);
3794
3795 if (queue_trunc)
3796 ceph_queue_vmtruncate(inode);
3797
3798 if (writeback)
3799 /*
3800 * queue inode for writeback: we can't actually call
3801 * filemap_write_and_wait, etc. from message handler
3802 * context.
3803 */
3804 ceph_queue_writeback(inode);
3805 if (queue_invalidate)
3806 ceph_queue_invalidate(inode);
3807 if (deleted_inode)
3808 invalidate_aliases(inode);
3809 if (wake)
3810 wake_up_all(&ci->i_cap_wq);
3811
3812 mutex_unlock(&session->s_mutex);
3813 if (check_caps == 1)
3814 ceph_check_caps(ci, flags | CHECK_CAPS_AUTHONLY | CHECK_CAPS_NOINVAL);
3815 else if (check_caps == 2)
3816 ceph_check_caps(ci, flags | CHECK_CAPS_NOINVAL);
3817 }
3818
3819 /*
3820 * Handle FLUSH_ACK from MDS, indicating that metadata we sent to the
3821 * MDS has been safely committed.
3822 */
3823 static void handle_cap_flush_ack(struct inode *inode, u64 flush_tid,
3824 struct ceph_mds_caps *m,
3825 struct ceph_mds_session *session,
3826 struct ceph_cap *cap)
3827 __releases(ci->i_ceph_lock)
3828 {
3829 struct ceph_inode_info *ci = ceph_inode(inode);
3830 struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(inode->i_sb)->mdsc;
3831 struct ceph_client *cl = mdsc->fsc->client;
3832 struct ceph_cap_flush *cf, *tmp_cf;
3833 LIST_HEAD(to_remove);
3834 unsigned seq = le32_to_cpu(m->seq);
3835 int dirty = le32_to_cpu(m->dirty);
3836 int cleaned = 0;
3837 bool drop = false;
3838 bool wake_ci = false;
3839 bool wake_mdsc = false;
3840
3841 list_for_each_entry_safe(cf, tmp_cf, &ci->i_cap_flush_list, i_list) {
3842 /* Is this the one that was flushed? */
3843 if (cf->tid == flush_tid)
3844 cleaned = cf->caps;
3845
3846 /* Is this a capsnap? */
3847 if (cf->is_capsnap)
3848 continue;
3849
3850 if (cf->tid <= flush_tid) {
3851 /*
3852 * An earlier or current tid. The FLUSH_ACK should
3853 * represent a superset of this flush's caps.
3854 */
3855 wake_ci |= __detach_cap_flush_from_ci(ci, cf);
3856 list_add_tail(&cf->i_list, &to_remove);
3857 } else {
3858 /*
3859 * This is a later one. Any caps in it are still dirty
3860 * so don't count them as cleaned.
3861 */
3862 cleaned &= ~cf->caps;
3863 if (!cleaned)
3864 break;
3865 }
3866 }
3867
3868 doutc(cl, "%p %llx.%llx mds%d seq %d on %s cleaned %s, flushing %s -> %s\n",
3869 inode, ceph_vinop(inode), session->s_mds, seq,
3870 ceph_cap_string(dirty), ceph_cap_string(cleaned),
3871 ceph_cap_string(ci->i_flushing_caps),
3872 ceph_cap_string(ci->i_flushing_caps & ~cleaned));
3873
3874 if (list_empty(&to_remove) && !cleaned)
3875 goto out;
3876
3877 ci->i_flushing_caps &= ~cleaned;
3878
3879 spin_lock(&mdsc->cap_dirty_lock);
3880
3881 list_for_each_entry(cf, &to_remove, i_list)
3882 wake_mdsc |= __detach_cap_flush_from_mdsc(mdsc, cf);
3883
3884 if (ci->i_flushing_caps == 0) {
3885 if (list_empty(&ci->i_cap_flush_list)) {
3886 list_del_init(&ci->i_flushing_item);
3887 if (!list_empty(&session->s_cap_flushing)) {
3888 struct inode *inode =
3889 &list_first_entry(&session->s_cap_flushing,
3890 struct ceph_inode_info,
3891 i_flushing_item)->netfs.inode;
3892 doutc(cl, " mds%d still flushing cap on %p %llx.%llx\n",
3893 session->s_mds, inode, ceph_vinop(inode));
3894 }
3895 }
3896 mdsc->num_cap_flushing--;
3897 doutc(cl, " %p %llx.%llx now !flushing\n", inode,
3898 ceph_vinop(inode));
3899
3900 if (ci->i_dirty_caps == 0) {
3901 doutc(cl, " %p %llx.%llx now clean\n", inode,
3902 ceph_vinop(inode));
3903 BUG_ON(!list_empty(&ci->i_dirty_item));
3904 drop = true;
3905 if (ci->i_wr_ref == 0 &&
3906 ci->i_wrbuffer_ref_head == 0) {
3907 BUG_ON(!ci->i_head_snapc);
3908 ceph_put_snap_context(ci->i_head_snapc);
3909 ci->i_head_snapc = NULL;
3910 }
3911 } else {
3912 BUG_ON(list_empty(&ci->i_dirty_item));
3913 }
3914 }
3915 spin_unlock(&mdsc->cap_dirty_lock);
3916
3917 out:
3918 spin_unlock(&ci->i_ceph_lock);
3919
3920 while (!list_empty(&to_remove)) {
3921 cf = list_first_entry(&to_remove,
3922 struct ceph_cap_flush, i_list);
3923 list_del_init(&cf->i_list);
3924 if (!cf->is_capsnap)
3925 ceph_free_cap_flush(cf);
3926 }
3927
3928 if (wake_ci)
3929 wake_up_all(&ci->i_cap_wq);
3930 if (wake_mdsc)
3931 wake_up_all(&mdsc->cap_flushing_wq);
3932 if (drop)
3933 iput(inode);
3934 }
3935
3936 void __ceph_remove_capsnap(struct inode *inode, struct ceph_cap_snap *capsnap,
3937 bool *wake_ci, bool *wake_mdsc)
3938 {
3939 struct ceph_inode_info *ci = ceph_inode(inode);
3940 struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(inode->i_sb)->mdsc;
3941 struct ceph_client *cl = mdsc->fsc->client;
3942 bool ret;
3943
3944 lockdep_assert_held(&ci->i_ceph_lock);
3945
3946 doutc(cl, "removing capsnap %p, %p %llx.%llx ci %p\n", capsnap,
3947 inode, ceph_vinop(inode), ci);
3948
3949 list_del_init(&capsnap->ci_item);
3950 ret = __detach_cap_flush_from_ci(ci, &capsnap->cap_flush);
3951 if (wake_ci)
3952 *wake_ci = ret;
3953
3954 spin_lock(&mdsc->cap_dirty_lock);
3955 if (list_empty(&ci->i_cap_flush_list))
3956 list_del_init(&ci->i_flushing_item);
3957
3958 ret = __detach_cap_flush_from_mdsc(mdsc, &capsnap->cap_flush);
3959 if (wake_mdsc)
3960 *wake_mdsc = ret;
3961 spin_unlock(&mdsc->cap_dirty_lock);
3962 }
3963
3964 void ceph_remove_capsnap(struct inode *inode, struct ceph_cap_snap *capsnap,
3965 bool *wake_ci, bool *wake_mdsc)
3966 {
3967 struct ceph_inode_info *ci = ceph_inode(inode);
3968
3969 lockdep_assert_held(&ci->i_ceph_lock);
3970
3971 WARN_ON_ONCE(capsnap->dirty_pages || capsnap->writing);
3972 __ceph_remove_capsnap(inode, capsnap, wake_ci, wake_mdsc);
3973 }
3974
3975 /*
3976 * Handle FLUSHSNAP_ACK. MDS has flushed snap data to disk and we can
3977 * throw away our cap_snap.
3978 *
3979 * Caller hold s_mutex.
3980 */
3981 static void handle_cap_flushsnap_ack(struct inode *inode, u64 flush_tid,
3982 struct ceph_mds_caps *m,
3983 struct ceph_mds_session *session)
3984 {
3985 struct ceph_inode_info *ci = ceph_inode(inode);
3986 struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(inode->i_sb)->mdsc;
3987 struct ceph_client *cl = mdsc->fsc->client;
3988 u64 follows = le64_to_cpu(m->snap_follows);
3989 struct ceph_cap_snap *capsnap = NULL, *iter;
3990 bool wake_ci = false;
3991 bool wake_mdsc = false;
3992
3993 doutc(cl, "%p %llx.%llx ci %p mds%d follows %lld\n", inode,
3994 ceph_vinop(inode), ci, session->s_mds, follows);
3995
3996 spin_lock(&ci->i_ceph_lock);
3997 list_for_each_entry(iter, &ci->i_cap_snaps, ci_item) {
3998 if (iter->follows == follows) {
3999 if (iter->cap_flush.tid != flush_tid) {
4000 doutc(cl, " cap_snap %p follows %lld "
4001 "tid %lld != %lld\n", iter,
4002 follows, flush_tid,
4003 iter->cap_flush.tid);
4004 break;
4005 }
4006 capsnap = iter;
4007 break;
4008 } else {
4009 doutc(cl, " skipping cap_snap %p follows %lld\n",
4010 iter, iter->follows);
4011 }
4012 }
4013 if (capsnap)
4014 ceph_remove_capsnap(inode, capsnap, &wake_ci, &wake_mdsc);
4015 spin_unlock(&ci->i_ceph_lock);
4016
4017 if (capsnap) {
4018 ceph_put_snap_context(capsnap->context);
4019 ceph_put_cap_snap(capsnap);
4020 if (wake_ci)
4021 wake_up_all(&ci->i_cap_wq);
4022 if (wake_mdsc)
4023 wake_up_all(&mdsc->cap_flushing_wq);
4024 iput(inode);
4025 }
4026 }
4027
4028 /*
4029 * Handle TRUNC from MDS, indicating file truncation.
4030 *
4031 * caller hold s_mutex.
4032 */
4033 static bool handle_cap_trunc(struct inode *inode,
4034 struct ceph_mds_caps *trunc,
4035 struct ceph_mds_session *session,
4036 struct cap_extra_info *extra_info)
4037 {
4038 struct ceph_inode_info *ci = ceph_inode(inode);
4039 struct ceph_client *cl = ceph_inode_to_client(inode);
4040 int mds = session->s_mds;
4041 int seq = le32_to_cpu(trunc->seq);
4042 u32 truncate_seq = le32_to_cpu(trunc->truncate_seq);
4043 u64 truncate_size = le64_to_cpu(trunc->truncate_size);
4044 u64 size = le64_to_cpu(trunc->size);
4045 int implemented = 0;
4046 int dirty = __ceph_caps_dirty(ci);
4047 int issued = __ceph_caps_issued(ceph_inode(inode), &implemented);
4048 bool queue_trunc = false;
4049
4050 lockdep_assert_held(&ci->i_ceph_lock);
4051
4052 issued |= implemented | dirty;
4053
4054 /*
4055 * If there is at least one crypto block then we'll trust
4056 * fscrypt_file_size. If the real length of the file is 0, then
4057 * ignore it (it has probably been truncated down to 0 by the MDS).
4058 */
4059 if (IS_ENCRYPTED(inode) && size)
4060 size = extra_info->fscrypt_file_size;
4061
4062 doutc(cl, "%p %llx.%llx mds%d seq %d to %lld truncate seq %d\n",
4063 inode, ceph_vinop(inode), mds, seq, truncate_size, truncate_seq);
4064 queue_trunc = ceph_fill_file_size(inode, issued,
4065 truncate_seq, truncate_size, size);
4066 return queue_trunc;
4067 }
4068
4069 /*
4070 * Handle EXPORT from MDS. Cap is being migrated _from_ this mds to a
4071 * different one. If we are the most recent migration we've seen (as
4072 * indicated by mseq), make note of the migrating cap bits for the
4073 * duration (until we see the corresponding IMPORT).
4074 *
4075 * caller holds s_mutex
4076 */
4077 static void handle_cap_export(struct inode *inode, struct ceph_mds_caps *ex,
4078 struct ceph_mds_cap_peer *ph,
4079 struct ceph_mds_session *session)
4080 {
4081 struct ceph_mds_client *mdsc = ceph_inode_to_fs_client(inode)->mdsc;
4082 struct ceph_client *cl = mdsc->fsc->client;
4083 struct ceph_mds_session *tsession = NULL;
4084 struct ceph_cap *cap, *tcap, *new_cap = NULL;
4085 struct ceph_inode_info *ci = ceph_inode(inode);
4086 u64 t_cap_id;
4087 unsigned mseq = le32_to_cpu(ex->migrate_seq);
4088 unsigned t_seq, t_mseq;
4089 int target, issued;
4090 int mds = session->s_mds;
4091
4092 if (ph) {
4093 t_cap_id = le64_to_cpu(ph->cap_id);
4094 t_seq = le32_to_cpu(ph->seq);
4095 t_mseq = le32_to_cpu(ph->mseq);
4096 target = le32_to_cpu(ph->mds);
4097 } else {
4098 t_cap_id = t_seq = t_mseq = 0;
4099 target = -1;
4100 }
4101
4102 doutc(cl, "%p %llx.%llx ci %p mds%d mseq %d target %d\n",
4103 inode, ceph_vinop(inode), ci, mds, mseq, target);
4104 retry:
4105 down_read(&mdsc->snap_rwsem);
4106 spin_lock(&ci->i_ceph_lock);
4107 cap = __get_cap_for_mds(ci, mds);
4108 if (!cap || cap->cap_id != le64_to_cpu(ex->cap_id))
4109 goto out_unlock;
4110
4111 if (target < 0) {
4112 ceph_remove_cap(mdsc, cap, false);
4113 goto out_unlock;
4114 }
4115
4116 /*
4117 * now we know we haven't received the cap import message yet
4118 * because the exported cap still exist.
4119 */
4120
4121 issued = cap->issued;
4122 if (issued != cap->implemented)
4123 pr_err_ratelimited_client(cl, "issued != implemented: "
4124 "%p %llx.%llx mds%d seq %d mseq %d"
4125 " issued %s implemented %s\n",
4126 inode, ceph_vinop(inode), mds,
4127 cap->seq, cap->mseq,
4128 ceph_cap_string(issued),
4129 ceph_cap_string(cap->implemented));
4130
4131
4132 tcap = __get_cap_for_mds(ci, target);
4133 if (tcap) {
4134 /* already have caps from the target */
4135 if (tcap->cap_id == t_cap_id &&
4136 ceph_seq_cmp(tcap->seq, t_seq) < 0) {
4137 doutc(cl, " updating import cap %p mds%d\n", tcap,
4138 target);
4139 tcap->cap_id = t_cap_id;
4140 tcap->seq = t_seq - 1;
4141 tcap->issue_seq = t_seq - 1;
4142 tcap->issued |= issued;
4143 tcap->implemented |= issued;
4144 if (cap == ci->i_auth_cap) {
4145 ci->i_auth_cap = tcap;
4146 change_auth_cap_ses(ci, tcap->session);
4147 }
4148 }
4149 ceph_remove_cap(mdsc, cap, false);
4150 goto out_unlock;
4151 } else if (tsession) {
4152 /* add placeholder for the export tagert */
4153 int flag = (cap == ci->i_auth_cap) ? CEPH_CAP_FLAG_AUTH : 0;
4154 tcap = new_cap;
4155 ceph_add_cap(inode, tsession, t_cap_id, issued, 0,
4156 t_seq - 1, t_mseq, (u64)-1, flag, &new_cap);
4157
4158 if (!list_empty(&ci->i_cap_flush_list) &&
4159 ci->i_auth_cap == tcap) {
4160 spin_lock(&mdsc->cap_dirty_lock);
4161 list_move_tail(&ci->i_flushing_item,
4162 &tcap->session->s_cap_flushing);
4163 spin_unlock(&mdsc->cap_dirty_lock);
4164 }
4165
4166 ceph_remove_cap(mdsc, cap, false);
4167 goto out_unlock;
4168 }
4169
4170 spin_unlock(&ci->i_ceph_lock);
4171 up_read(&mdsc->snap_rwsem);
4172 mutex_unlock(&session->s_mutex);
4173
4174 /* open target session */
4175 tsession = ceph_mdsc_open_export_target_session(mdsc, target);
4176 if (!IS_ERR(tsession)) {
4177 if (mds > target) {
4178 mutex_lock(&session->s_mutex);
4179 mutex_lock_nested(&tsession->s_mutex,
4180 SINGLE_DEPTH_NESTING);
4181 } else {
4182 mutex_lock(&tsession->s_mutex);
4183 mutex_lock_nested(&session->s_mutex,
4184 SINGLE_DEPTH_NESTING);
4185 }
4186 new_cap = ceph_get_cap(mdsc, NULL);
4187 } else {
4188 WARN_ON(1);
4189 tsession = NULL;
4190 target = -1;
4191 mutex_lock(&session->s_mutex);
4192 }
4193 goto retry;
4194
4195 out_unlock:
4196 spin_unlock(&ci->i_ceph_lock);
4197 up_read(&mdsc->snap_rwsem);
4198 mutex_unlock(&session->s_mutex);
4199 if (tsession) {
4200 mutex_unlock(&tsession->s_mutex);
4201 ceph_put_mds_session(tsession);
4202 }
4203 if (new_cap)
4204 ceph_put_cap(mdsc, new_cap);
4205 }
4206
4207 /*
4208 * Handle cap IMPORT.
4209 *
4210 * caller holds s_mutex. acquires i_ceph_lock
4211 */
4212 static void handle_cap_import(struct ceph_mds_client *mdsc,
4213 struct inode *inode, struct ceph_mds_caps *im,
4214 struct ceph_mds_cap_peer *ph,
4215 struct ceph_mds_session *session,
4216 struct ceph_cap **target_cap, int *old_issued)
4217 {
4218 struct ceph_inode_info *ci = ceph_inode(inode);
4219 struct ceph_client *cl = mdsc->fsc->client;
4220 struct ceph_cap *cap, *ocap, *new_cap = NULL;
4221 int mds = session->s_mds;
4222 int issued;
4223 unsigned caps = le32_to_cpu(im->caps);
4224 unsigned wanted = le32_to_cpu(im->wanted);
4225 unsigned seq = le32_to_cpu(im->seq);
4226 unsigned mseq = le32_to_cpu(im->migrate_seq);
4227 u64 realmino = le64_to_cpu(im->realm);
4228 u64 cap_id = le64_to_cpu(im->cap_id);
4229 u64 p_cap_id;
4230 int peer;
4231
4232 if (ph) {
4233 p_cap_id = le64_to_cpu(ph->cap_id);
4234 peer = le32_to_cpu(ph->mds);
4235 } else {
4236 p_cap_id = 0;
4237 peer = -1;
4238 }
4239
4240 doutc(cl, "%p %llx.%llx ci %p mds%d mseq %d peer %d\n",
4241 inode, ceph_vinop(inode), ci, mds, mseq, peer);
4242 retry:
4243 cap = __get_cap_for_mds(ci, mds);
4244 if (!cap) {
4245 if (!new_cap) {
4246 spin_unlock(&ci->i_ceph_lock);
4247 new_cap = ceph_get_cap(mdsc, NULL);
4248 spin_lock(&ci->i_ceph_lock);
4249 goto retry;
4250 }
4251 cap = new_cap;
4252 } else {
4253 if (new_cap) {
4254 ceph_put_cap(mdsc, new_cap);
4255 new_cap = NULL;
4256 }
4257 }
4258
4259 __ceph_caps_issued(ci, &issued);
4260 issued |= __ceph_caps_dirty(ci);
4261
4262 ceph_add_cap(inode, session, cap_id, caps, wanted, seq, mseq,
4263 realmino, CEPH_CAP_FLAG_AUTH, &new_cap);
4264
4265 ocap = peer >= 0 ? __get_cap_for_mds(ci, peer) : NULL;
4266 if (ocap && ocap->cap_id == p_cap_id) {
4267 doutc(cl, " remove export cap %p mds%d flags %d\n",
4268 ocap, peer, ph->flags);
4269 if ((ph->flags & CEPH_CAP_FLAG_AUTH) &&
4270 (ocap->seq != le32_to_cpu(ph->seq) ||
4271 ocap->mseq != le32_to_cpu(ph->mseq))) {
4272 pr_err_ratelimited_client(cl, "mismatched seq/mseq: "
4273 "%p %llx.%llx mds%d seq %d mseq %d"
4274 " importer mds%d has peer seq %d mseq %d\n",
4275 inode, ceph_vinop(inode), peer,
4276 ocap->seq, ocap->mseq, mds,
4277 le32_to_cpu(ph->seq),
4278 le32_to_cpu(ph->mseq));
4279 }
4280 ceph_remove_cap(mdsc, ocap, (ph->flags & CEPH_CAP_FLAG_RELEASE));
4281 }
4282
4283 *old_issued = issued;
4284 *target_cap = cap;
4285 }
4286
4287 #ifdef CONFIG_FS_ENCRYPTION
4288 static int parse_fscrypt_fields(void **p, void *end,
4289 struct cap_extra_info *extra)
4290 {
4291 u32 len;
4292
4293 ceph_decode_32_safe(p, end, extra->fscrypt_auth_len, bad);
4294 if (extra->fscrypt_auth_len) {
4295 ceph_decode_need(p, end, extra->fscrypt_auth_len, bad);
4296 extra->fscrypt_auth = kmalloc(extra->fscrypt_auth_len,
4297 GFP_KERNEL);
4298 if (!extra->fscrypt_auth)
4299 return -ENOMEM;
4300 ceph_decode_copy_safe(p, end, extra->fscrypt_auth,
4301 extra->fscrypt_auth_len, bad);
4302 }
4303
4304 ceph_decode_32_safe(p, end, len, bad);
4305 if (len >= sizeof(u64)) {
4306 ceph_decode_64_safe(p, end, extra->fscrypt_file_size, bad);
4307 len -= sizeof(u64);
4308 }
4309 ceph_decode_skip_n(p, end, len, bad);
4310 return 0;
4311 bad:
4312 return -EIO;
4313 }
4314 #else
4315 static int parse_fscrypt_fields(void **p, void *end,
4316 struct cap_extra_info *extra)
4317 {
4318 u32 len;
4319
4320 /* Don't care about these fields unless we're encryption-capable */
4321 ceph_decode_32_safe(p, end, len, bad);
4322 if (len)
4323 ceph_decode_skip_n(p, end, len, bad);
4324 ceph_decode_32_safe(p, end, len, bad);
4325 if (len)
4326 ceph_decode_skip_n(p, end, len, bad);
4327 return 0;
4328 bad:
4329 return -EIO;
4330 }
4331 #endif
4332
4333 /*
4334 * Handle a caps message from the MDS.
4335 *
4336 * Identify the appropriate session, inode, and call the right handler
4337 * based on the cap op.
4338 */
4339 void ceph_handle_caps(struct ceph_mds_session *session,
4340 struct ceph_msg *msg)
4341 {
4342 struct ceph_mds_client *mdsc = session->s_mdsc;
4343 struct ceph_client *cl = mdsc->fsc->client;
4344 struct inode *inode;
4345 struct ceph_inode_info *ci;
4346 struct ceph_cap *cap;
4347 struct ceph_mds_caps *h;
4348 struct ceph_mds_cap_peer *peer = NULL;
4349 struct ceph_snap_realm *realm = NULL;
4350 int op;
4351 int msg_version = le16_to_cpu(msg->hdr.version);
4352 u32 seq, mseq;
4353 struct ceph_vino vino;
4354 void *snaptrace;
4355 size_t snaptrace_len;
4356 void *p, *end;
4357 struct cap_extra_info extra_info = {};
4358 bool queue_trunc;
4359 bool close_sessions = false;
4360 bool do_cap_release = false;
4361
4362 doutc(cl, "from mds%d\n", session->s_mds);
4363
4364 if (!ceph_inc_mds_stopping_blocker(mdsc, session))
4365 return;
4366
4367 /* decode */
4368 end = msg->front.iov_base + msg->front.iov_len;
4369 if (msg->front.iov_len < sizeof(*h))
4370 goto bad;
4371 h = msg->front.iov_base;
4372 op = le32_to_cpu(h->op);
4373 vino.ino = le64_to_cpu(h->ino);
4374 vino.snap = CEPH_NOSNAP;
4375 seq = le32_to_cpu(h->seq);
4376 mseq = le32_to_cpu(h->migrate_seq);
4377
4378 snaptrace = h + 1;
4379 snaptrace_len = le32_to_cpu(h->snap_trace_len);
4380 p = snaptrace + snaptrace_len;
4381
4382 if (msg_version >= 2) {
4383 u32 flock_len;
4384 ceph_decode_32_safe(&p, end, flock_len, bad);
4385 if (p + flock_len > end)
4386 goto bad;
4387 p += flock_len;
4388 }
4389
4390 if (msg_version >= 3) {
4391 if (op == CEPH_CAP_OP_IMPORT) {
4392 if (p + sizeof(*peer) > end)
4393 goto bad;
4394 peer = p;
4395 p += sizeof(*peer);
4396 } else if (op == CEPH_CAP_OP_EXPORT) {
4397 /* recorded in unused fields */
4398 peer = (void *)&h->size;
4399 }
4400 }
4401
4402 if (msg_version >= 4) {
4403 ceph_decode_64_safe(&p, end, extra_info.inline_version, bad);
4404 ceph_decode_32_safe(&p, end, extra_info.inline_len, bad);
4405 if (p + extra_info.inline_len > end)
4406 goto bad;
4407 extra_info.inline_data = p;
4408 p += extra_info.inline_len;
4409 }
4410
4411 if (msg_version >= 5) {
4412 struct ceph_osd_client *osdc = &mdsc->fsc->client->osdc;
4413 u32 epoch_barrier;
4414
4415 ceph_decode_32_safe(&p, end, epoch_barrier, bad);
4416 ceph_osdc_update_epoch_barrier(osdc, epoch_barrier);
4417 }
4418
4419 if (msg_version >= 8) {
4420 u32 pool_ns_len;
4421
4422 /* version >= 6 */
4423 ceph_decode_skip_64(&p, end, bad); // flush_tid
4424 /* version >= 7 */
4425 ceph_decode_skip_32(&p, end, bad); // caller_uid
4426 ceph_decode_skip_32(&p, end, bad); // caller_gid
4427 /* version >= 8 */
4428 ceph_decode_32_safe(&p, end, pool_ns_len, bad);
4429 if (pool_ns_len > 0) {
4430 ceph_decode_need(&p, end, pool_ns_len, bad);
4431 extra_info.pool_ns =
4432 ceph_find_or_create_string(p, pool_ns_len);
4433 p += pool_ns_len;
4434 }
4435 }
4436
4437 if (msg_version >= 9) {
4438 struct ceph_timespec *btime;
4439
4440 if (p + sizeof(*btime) > end)
4441 goto bad;
4442 btime = p;
4443 ceph_decode_timespec64(&extra_info.btime, btime);
4444 p += sizeof(*btime);
4445 ceph_decode_64_safe(&p, end, extra_info.change_attr, bad);
4446 }
4447
4448 if (msg_version >= 11) {
4449 /* version >= 10 */
4450 ceph_decode_skip_32(&p, end, bad); // flags
4451 /* version >= 11 */
4452 extra_info.dirstat_valid = true;
4453 ceph_decode_64_safe(&p, end, extra_info.nfiles, bad);
4454 ceph_decode_64_safe(&p, end, extra_info.nsubdirs, bad);
4455 }
4456
4457 if (msg_version >= 12) {
4458 if (parse_fscrypt_fields(&p, end, &extra_info))
4459 goto bad;
4460 }
4461
4462 /* lookup ino */
4463 inode = ceph_find_inode(mdsc->fsc->sb, vino);
4464 doutc(cl, " op %s ino %llx.%llx inode %p\n", ceph_cap_op_name(op),
4465 vino.ino, vino.snap, inode);
4466
4467 mutex_lock(&session->s_mutex);
4468 doutc(cl, " mds%d seq %lld cap seq %u\n", session->s_mds,
4469 session->s_seq, (unsigned)seq);
4470
4471 if (!inode) {
4472 doutc(cl, " i don't have ino %llx\n", vino.ino);
4473
4474 switch (op) {
4475 case CEPH_CAP_OP_IMPORT:
4476 case CEPH_CAP_OP_REVOKE:
4477 case CEPH_CAP_OP_GRANT:
4478 do_cap_release = true;
4479 break;
4480 default:
4481 break;
4482 }
4483 goto flush_cap_releases;
4484 }
4485 ci = ceph_inode(inode);
4486
4487 /* these will work even if we don't have a cap yet */
4488 switch (op) {
4489 case CEPH_CAP_OP_FLUSHSNAP_ACK:
4490 handle_cap_flushsnap_ack(inode, le64_to_cpu(msg->hdr.tid),
4491 h, session);
4492 goto done;
4493
4494 case CEPH_CAP_OP_EXPORT:
4495 handle_cap_export(inode, h, peer, session);
4496 goto done_unlocked;
4497
4498 case CEPH_CAP_OP_IMPORT:
4499 realm = NULL;
4500 if (snaptrace_len) {
4501 down_write(&mdsc->snap_rwsem);
4502 if (ceph_update_snap_trace(mdsc, snaptrace,
4503 snaptrace + snaptrace_len,
4504 false, &realm)) {
4505 up_write(&mdsc->snap_rwsem);
4506 close_sessions = true;
4507 goto done;
4508 }
4509 downgrade_write(&mdsc->snap_rwsem);
4510 } else {
4511 down_read(&mdsc->snap_rwsem);
4512 }
4513 spin_lock(&ci->i_ceph_lock);
4514 handle_cap_import(mdsc, inode, h, peer, session,
4515 &cap, &extra_info.issued);
4516 handle_cap_grant(inode, session, cap,
4517 h, msg->middle, &extra_info);
4518 if (realm)
4519 ceph_put_snap_realm(mdsc, realm);
4520 goto done_unlocked;
4521 }
4522
4523 /* the rest require a cap */
4524 spin_lock(&ci->i_ceph_lock);
4525 cap = __get_cap_for_mds(ceph_inode(inode), session->s_mds);
4526 if (!cap) {
4527 doutc(cl, " no cap on %p ino %llx.%llx from mds%d\n",
4528 inode, ceph_ino(inode), ceph_snap(inode),
4529 session->s_mds);
4530 spin_unlock(&ci->i_ceph_lock);
4531 switch (op) {
4532 case CEPH_CAP_OP_REVOKE:
4533 case CEPH_CAP_OP_GRANT:
4534 do_cap_release = true;
4535 break;
4536 default:
4537 break;
4538 }
4539 goto flush_cap_releases;
4540 }
4541
4542 /* note that each of these drops i_ceph_lock for us */
4543 switch (op) {
4544 case CEPH_CAP_OP_REVOKE:
4545 case CEPH_CAP_OP_GRANT:
4546 __ceph_caps_issued(ci, &extra_info.issued);
4547 extra_info.issued |= __ceph_caps_dirty(ci);
4548 handle_cap_grant(inode, session, cap,
4549 h, msg->middle, &extra_info);
4550 goto done_unlocked;
4551
4552 case CEPH_CAP_OP_FLUSH_ACK:
4553 handle_cap_flush_ack(inode, le64_to_cpu(msg->hdr.tid),
4554 h, session, cap);
4555 break;
4556
4557 case CEPH_CAP_OP_TRUNC:
4558 queue_trunc = handle_cap_trunc(inode, h, session,
4559 &extra_info);
4560 spin_unlock(&ci->i_ceph_lock);
4561 if (queue_trunc)
4562 ceph_queue_vmtruncate(inode);
4563 break;
4564
4565 default:
4566 spin_unlock(&ci->i_ceph_lock);
4567 pr_err_client(cl, "unknown cap op %d %s\n", op,
4568 ceph_cap_op_name(op));
4569 }
4570
4571 done:
4572 mutex_unlock(&session->s_mutex);
4573 done_unlocked:
4574 iput(inode);
4575 out:
4576 ceph_dec_mds_stopping_blocker(mdsc);
4577
4578 ceph_put_string(extra_info.pool_ns);
4579
4580 /* Defer closing the sessions after s_mutex lock being released */
4581 if (close_sessions)
4582 ceph_mdsc_close_sessions(mdsc);
4583
4584 kfree(extra_info.fscrypt_auth);
4585 return;
4586
4587 flush_cap_releases:
4588 /*
4589 * send any cap release message to try to move things
4590 * along for the mds (who clearly thinks we still have this
4591 * cap).
4592 */
4593 if (do_cap_release) {
4594 cap = ceph_get_cap(mdsc, NULL);
4595 cap->cap_ino = vino.ino;
4596 cap->queue_release = 1;
4597 cap->cap_id = le64_to_cpu(h->cap_id);
4598 cap->mseq = mseq;
4599 cap->seq = seq;
4600 cap->issue_seq = seq;
4601 spin_lock(&session->s_cap_lock);
4602 __ceph_queue_cap_release(session, cap);
4603 spin_unlock(&session->s_cap_lock);
4604 }
4605 ceph_flush_cap_releases(mdsc, session);
4606 goto done;
4607
4608 bad:
4609 pr_err_client(cl, "corrupt message\n");
4610 ceph_msg_dump(msg);
4611 goto out;
4612 }
4613
4614 /*
4615 * Delayed work handler to process end of delayed cap release LRU list.
4616 *
4617 * If new caps are added to the list while processing it, these won't get
4618 * processed in this run. In this case, the ci->i_hold_caps_max will be
4619 * returned so that the work can be scheduled accordingly.
4620 */
4621 unsigned long ceph_check_delayed_caps(struct ceph_mds_client *mdsc)
4622 {
4623 struct ceph_client *cl = mdsc->fsc->client;
4624 struct inode *inode;
4625 struct ceph_inode_info *ci;
4626 struct ceph_mount_options *opt = mdsc->fsc->mount_options;
4627 unsigned long delay_max = opt->caps_wanted_delay_max * HZ;
4628 unsigned long loop_start = jiffies;
4629 unsigned long delay = 0;
4630
4631 doutc(cl, "begin\n");
4632 spin_lock(&mdsc->cap_delay_lock);
4633 while (!list_empty(&mdsc->cap_delay_list)) {
4634 ci = list_first_entry(&mdsc->cap_delay_list,
4635 struct ceph_inode_info,
4636 i_cap_delay_list);
4637 if (time_before(loop_start, ci->i_hold_caps_max - delay_max)) {
4638 doutc(cl, "caps added recently. Exiting loop");
4639 delay = ci->i_hold_caps_max;
4640 break;
4641 }
4642 if ((ci->i_ceph_flags & CEPH_I_FLUSH) == 0 &&
4643 time_before(jiffies, ci->i_hold_caps_max))
4644 break;
4645 list_del_init(&ci->i_cap_delay_list);
4646
4647 inode = igrab(&ci->netfs.inode);
4648 if (inode) {
4649 spin_unlock(&mdsc->cap_delay_lock);
4650 doutc(cl, "on %p %llx.%llx\n", inode,
4651 ceph_vinop(inode));
4652 ceph_check_caps(ci, 0);
4653 iput(inode);
4654 spin_lock(&mdsc->cap_delay_lock);
4655 }
4656
4657 /*
4658 * Make sure too many dirty caps or general
4659 * slowness doesn't block mdsc delayed work,
4660 * preventing send_renew_caps() from running.
4661 */
4662 if (jiffies - loop_start >= 5 * HZ)
4663 break;
4664 }
4665 spin_unlock(&mdsc->cap_delay_lock);
4666 doutc(cl, "done\n");
4667
4668 return delay;
4669 }
4670
4671 /*
4672 * Flush all dirty caps to the mds
4673 */
4674 static void flush_dirty_session_caps(struct ceph_mds_session *s)
4675 {
4676 struct ceph_mds_client *mdsc = s->s_mdsc;
4677 struct ceph_client *cl = mdsc->fsc->client;
4678 struct ceph_inode_info *ci;
4679 struct inode *inode;
4680
4681 doutc(cl, "begin\n");
4682 spin_lock(&mdsc->cap_dirty_lock);
4683 while (!list_empty(&s->s_cap_dirty)) {
4684 ci = list_first_entry(&s->s_cap_dirty, struct ceph_inode_info,
4685 i_dirty_item);
4686 inode = &ci->netfs.inode;
4687 ihold(inode);
4688 doutc(cl, "%p %llx.%llx\n", inode, ceph_vinop(inode));
4689 spin_unlock(&mdsc->cap_dirty_lock);
4690 ceph_wait_on_async_create(inode);
4691 ceph_check_caps(ci, CHECK_CAPS_FLUSH);
4692 iput(inode);
4693 spin_lock(&mdsc->cap_dirty_lock);
4694 }
4695 spin_unlock(&mdsc->cap_dirty_lock);
4696 doutc(cl, "done\n");
4697 }
4698
4699 void ceph_flush_dirty_caps(struct ceph_mds_client *mdsc)
4700 {
4701 ceph_mdsc_iterate_sessions(mdsc, flush_dirty_session_caps, true);
4702 }
4703
4704 void __ceph_touch_fmode(struct ceph_inode_info *ci,
4705 struct ceph_mds_client *mdsc, int fmode)
4706 {
4707 unsigned long now = jiffies;
4708 if (fmode & CEPH_FILE_MODE_RD)
4709 ci->i_last_rd = now;
4710 if (fmode & CEPH_FILE_MODE_WR)
4711 ci->i_last_wr = now;
4712 /* queue periodic check */
4713 if (fmode &&
4714 __ceph_is_any_real_caps(ci) &&
4715 list_empty(&ci->i_cap_delay_list))
4716 __cap_delay_requeue(mdsc, ci);
4717 }
4718
4719 void ceph_get_fmode(struct ceph_inode_info *ci, int fmode, int count)
4720 {
4721 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(ci->netfs.inode.i_sb);
4722 int bits = (fmode << 1) | 1;
4723 bool already_opened = false;
4724 int i;
4725
4726 if (count == 1)
4727 atomic64_inc(&mdsc->metric.opened_files);
4728
4729 spin_lock(&ci->i_ceph_lock);
4730 for (i = 0; i < CEPH_FILE_MODE_BITS; i++) {
4731 /*
4732 * If any of the mode ref is larger than 0,
4733 * that means it has been already opened by
4734 * others. Just skip checking the PIN ref.
4735 */
4736 if (i && ci->i_nr_by_mode[i])
4737 already_opened = true;
4738
4739 if (bits & (1 << i))
4740 ci->i_nr_by_mode[i] += count;
4741 }
4742
4743 if (!already_opened)
4744 percpu_counter_inc(&mdsc->metric.opened_inodes);
4745 spin_unlock(&ci->i_ceph_lock);
4746 }
4747
4748 /*
4749 * Drop open file reference. If we were the last open file,
4750 * we may need to release capabilities to the MDS (or schedule
4751 * their delayed release).
4752 */
4753 void ceph_put_fmode(struct ceph_inode_info *ci, int fmode, int count)
4754 {
4755 struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(ci->netfs.inode.i_sb);
4756 int bits = (fmode << 1) | 1;
4757 bool is_closed = true;
4758 int i;
4759
4760 if (count == 1)
4761 atomic64_dec(&mdsc->metric.opened_files);
4762
4763 spin_lock(&ci->i_ceph_lock);
4764 for (i = 0; i < CEPH_FILE_MODE_BITS; i++) {
4765 if (bits & (1 << i)) {
4766 BUG_ON(ci->i_nr_by_mode[i] < count);
4767 ci->i_nr_by_mode[i] -= count;
4768 }
4769
4770 /*
4771 * If any of the mode ref is not 0 after
4772 * decreased, that means it is still opened
4773 * by others. Just skip checking the PIN ref.
4774 */
4775 if (i && ci->i_nr_by_mode[i])
4776 is_closed = false;
4777 }
4778
4779 if (is_closed)
4780 percpu_counter_dec(&mdsc->metric.opened_inodes);
4781 spin_unlock(&ci->i_ceph_lock);
4782 }
4783
4784 /*
4785 * For a soon-to-be unlinked file, drop the LINK caps. If it
4786 * looks like the link count will hit 0, drop any other caps (other
4787 * than PIN) we don't specifically want (due to the file still being
4788 * open).
4789 */
4790 int ceph_drop_caps_for_unlink(struct inode *inode)
4791 {
4792 struct ceph_inode_info *ci = ceph_inode(inode);
4793 int drop = CEPH_CAP_LINK_SHARED | CEPH_CAP_LINK_EXCL;
4794
4795 spin_lock(&ci->i_ceph_lock);
4796 if (inode->i_nlink == 1) {
4797 drop |= ~(__ceph_caps_wanted(ci) | CEPH_CAP_PIN);
4798
4799 if (__ceph_caps_dirty(ci)) {
4800 struct ceph_mds_client *mdsc =
4801 ceph_inode_to_fs_client(inode)->mdsc;
4802
4803 doutc(mdsc->fsc->client, "%p %llx.%llx\n", inode,
4804 ceph_vinop(inode));
4805 spin_lock(&mdsc->cap_delay_lock);
4806 ci->i_ceph_flags |= CEPH_I_FLUSH;
4807 if (!list_empty(&ci->i_cap_delay_list))
4808 list_del_init(&ci->i_cap_delay_list);
4809 list_add_tail(&ci->i_cap_delay_list,
4810 &mdsc->cap_unlink_delay_list);
4811 spin_unlock(&mdsc->cap_delay_lock);
4812
4813 /*
4814 * Fire the work immediately, because the MDS maybe
4815 * waiting for caps release.
4816 */
4817 ceph_queue_cap_unlink_work(mdsc);
4818 }
4819 }
4820 spin_unlock(&ci->i_ceph_lock);
4821 return drop;
4822 }
4823
4824 /*
4825 * Helpers for embedding cap and dentry lease releases into mds
4826 * requests.
4827 *
4828 * @force is used by dentry_release (below) to force inclusion of a
4829 * record for the directory inode, even when there aren't any caps to
4830 * drop.
4831 */
4832 int ceph_encode_inode_release(void **p, struct inode *inode,
4833 int mds, int drop, int unless, int force)
4834 {
4835 struct ceph_inode_info *ci = ceph_inode(inode);
4836 struct ceph_client *cl = ceph_inode_to_client(inode);
4837 struct ceph_cap *cap;
4838 struct ceph_mds_request_release *rel = *p;
4839 int used, dirty;
4840 int ret = 0;
4841
4842 spin_lock(&ci->i_ceph_lock);
4843 used = __ceph_caps_used(ci);
4844 dirty = __ceph_caps_dirty(ci);
4845
4846 doutc(cl, "%p %llx.%llx mds%d used|dirty %s drop %s unless %s\n",
4847 inode, ceph_vinop(inode), mds, ceph_cap_string(used|dirty),
4848 ceph_cap_string(drop), ceph_cap_string(unless));
4849
4850 /* only drop unused, clean caps */
4851 drop &= ~(used | dirty);
4852
4853 cap = __get_cap_for_mds(ci, mds);
4854 if (cap && __cap_is_valid(cap)) {
4855 unless &= cap->issued;
4856 if (unless) {
4857 if (unless & CEPH_CAP_AUTH_EXCL)
4858 drop &= ~CEPH_CAP_AUTH_SHARED;
4859 if (unless & CEPH_CAP_LINK_EXCL)
4860 drop &= ~CEPH_CAP_LINK_SHARED;
4861 if (unless & CEPH_CAP_XATTR_EXCL)
4862 drop &= ~CEPH_CAP_XATTR_SHARED;
4863 if (unless & CEPH_CAP_FILE_EXCL)
4864 drop &= ~CEPH_CAP_FILE_SHARED;
4865 }
4866
4867 if (force || (cap->issued & drop)) {
4868 if (cap->issued & drop) {
4869 int wanted = __ceph_caps_wanted(ci);
4870 doutc(cl, "%p %llx.%llx cap %p %s -> %s, "
4871 "wanted %s -> %s\n", inode,
4872 ceph_vinop(inode), cap,
4873 ceph_cap_string(cap->issued),
4874 ceph_cap_string(cap->issued & ~drop),
4875 ceph_cap_string(cap->mds_wanted),
4876 ceph_cap_string(wanted));
4877
4878 cap->issued &= ~drop;
4879 cap->implemented &= ~drop;
4880 cap->mds_wanted = wanted;
4881 if (cap == ci->i_auth_cap &&
4882 !(wanted & CEPH_CAP_ANY_FILE_WR))
4883 ci->i_requested_max_size = 0;
4884 } else {
4885 doutc(cl, "%p %llx.%llx cap %p %s (force)\n",
4886 inode, ceph_vinop(inode), cap,
4887 ceph_cap_string(cap->issued));
4888 }
4889
4890 rel->ino = cpu_to_le64(ceph_ino(inode));
4891 rel->cap_id = cpu_to_le64(cap->cap_id);
4892 rel->seq = cpu_to_le32(cap->seq);
4893 rel->issue_seq = cpu_to_le32(cap->issue_seq);
4894 rel->mseq = cpu_to_le32(cap->mseq);
4895 rel->caps = cpu_to_le32(cap->implemented);
4896 rel->wanted = cpu_to_le32(cap->mds_wanted);
4897 rel->dname_len = 0;
4898 rel->dname_seq = 0;
4899 *p += sizeof(*rel);
4900 ret = 1;
4901 } else {
4902 doutc(cl, "%p %llx.%llx cap %p %s (noop)\n",
4903 inode, ceph_vinop(inode), cap,
4904 ceph_cap_string(cap->issued));
4905 }
4906 }
4907 spin_unlock(&ci->i_ceph_lock);
4908 return ret;
4909 }
4910
4911 /**
4912 * ceph_encode_dentry_release - encode a dentry release into an outgoing request
4913 * @p: outgoing request buffer
4914 * @dentry: dentry to release
4915 * @dir: dir to release it from
4916 * @mds: mds that we're speaking to
4917 * @drop: caps being dropped
4918 * @unless: unless we have these caps
4919 *
4920 * Encode a dentry release into an outgoing request buffer. Returns 1 if the
4921 * thing was released, or a negative error code otherwise.
4922 */
4923 int ceph_encode_dentry_release(void **p, struct dentry *dentry,
4924 struct inode *dir,
4925 int mds, int drop, int unless)
4926 {
4927 struct ceph_mds_request_release *rel = *p;
4928 struct ceph_dentry_info *di = ceph_dentry(dentry);
4929 struct ceph_client *cl;
4930 int force = 0;
4931 int ret;
4932
4933 /* This shouldn't happen */
4934 BUG_ON(!dir);
4935
4936 /*
4937 * force an record for the directory caps if we have a dentry lease.
4938 * this is racy (can't take i_ceph_lock and d_lock together), but it
4939 * doesn't have to be perfect; the mds will revoke anything we don't
4940 * release.
4941 */
4942 spin_lock(&dentry->d_lock);
4943 if (di->lease_session && di->lease_session->s_mds == mds)
4944 force = 1;
4945 spin_unlock(&dentry->d_lock);
4946
4947 ret = ceph_encode_inode_release(p, dir, mds, drop, unless, force);
4948
4949 cl = ceph_inode_to_client(dir);
4950 spin_lock(&dentry->d_lock);
4951 if (ret && di->lease_session && di->lease_session->s_mds == mds) {
4952 doutc(cl, "%p mds%d seq %d\n", dentry, mds,
4953 (int)di->lease_seq);
4954 rel->dname_seq = cpu_to_le32(di->lease_seq);
4955 __ceph_mdsc_drop_dentry_lease(dentry);
4956 spin_unlock(&dentry->d_lock);
4957 if (IS_ENCRYPTED(dir) && fscrypt_has_encryption_key(dir)) {
4958 int ret2 = ceph_encode_encrypted_fname(dir, dentry, *p);
4959
4960 if (ret2 < 0)
4961 return ret2;
4962
4963 rel->dname_len = cpu_to_le32(ret2);
4964 *p += ret2;
4965 } else {
4966 rel->dname_len = cpu_to_le32(dentry->d_name.len);
4967 memcpy(*p, dentry->d_name.name, dentry->d_name.len);
4968 *p += dentry->d_name.len;
4969 }
4970 } else {
4971 spin_unlock(&dentry->d_lock);
4972 }
4973 return ret;
4974 }
4975
4976 static int remove_capsnaps(struct ceph_mds_client *mdsc, struct inode *inode)
4977 {
4978 struct ceph_inode_info *ci = ceph_inode(inode);
4979 struct ceph_client *cl = mdsc->fsc->client;
4980 struct ceph_cap_snap *capsnap;
4981 int capsnap_release = 0;
4982
4983 lockdep_assert_held(&ci->i_ceph_lock);
4984
4985 doutc(cl, "removing capsnaps, ci is %p, %p %llx.%llx\n",
4986 ci, inode, ceph_vinop(inode));
4987
4988 while (!list_empty(&ci->i_cap_snaps)) {
4989 capsnap = list_first_entry(&ci->i_cap_snaps,
4990 struct ceph_cap_snap, ci_item);
4991 __ceph_remove_capsnap(inode, capsnap, NULL, NULL);
4992 ceph_put_snap_context(capsnap->context);
4993 ceph_put_cap_snap(capsnap);
4994 capsnap_release++;
4995 }
4996 wake_up_all(&ci->i_cap_wq);
4997 wake_up_all(&mdsc->cap_flushing_wq);
4998 return capsnap_release;
4999 }
5000
5001 int ceph_purge_inode_cap(struct inode *inode, struct ceph_cap *cap, bool *invalidate)
5002 {
5003 struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
5004 struct ceph_mds_client *mdsc = fsc->mdsc;
5005 struct ceph_client *cl = fsc->client;
5006 struct ceph_inode_info *ci = ceph_inode(inode);
5007 bool is_auth;
5008 bool dirty_dropped = false;
5009 int iputs = 0;
5010
5011 lockdep_assert_held(&ci->i_ceph_lock);
5012
5013 doutc(cl, "removing cap %p, ci is %p, %p %llx.%llx\n",
5014 cap, ci, inode, ceph_vinop(inode));
5015
5016 is_auth = (cap == ci->i_auth_cap);
5017 __ceph_remove_cap(cap, false);
5018 if (is_auth) {
5019 struct ceph_cap_flush *cf;
5020
5021 if (ceph_inode_is_shutdown(inode)) {
5022 if (inode->i_data.nrpages > 0)
5023 *invalidate = true;
5024 if (ci->i_wrbuffer_ref > 0)
5025 mapping_set_error(&inode->i_data, -EIO);
5026 }
5027
5028 spin_lock(&mdsc->cap_dirty_lock);
5029
5030 /* trash all of the cap flushes for this inode */
5031 while (!list_empty(&ci->i_cap_flush_list)) {
5032 cf = list_first_entry(&ci->i_cap_flush_list,
5033 struct ceph_cap_flush, i_list);
5034 list_del_init(&cf->g_list);
5035 list_del_init(&cf->i_list);
5036 if (!cf->is_capsnap)
5037 ceph_free_cap_flush(cf);
5038 }
5039
5040 if (!list_empty(&ci->i_dirty_item)) {
5041 pr_warn_ratelimited_client(cl,
5042 " dropping dirty %s state for %p %llx.%llx\n",
5043 ceph_cap_string(ci->i_dirty_caps),
5044 inode, ceph_vinop(inode));
5045 ci->i_dirty_caps = 0;
5046 list_del_init(&ci->i_dirty_item);
5047 dirty_dropped = true;
5048 }
5049 if (!list_empty(&ci->i_flushing_item)) {
5050 pr_warn_ratelimited_client(cl,
5051 " dropping dirty+flushing %s state for %p %llx.%llx\n",
5052 ceph_cap_string(ci->i_flushing_caps),
5053 inode, ceph_vinop(inode));
5054 ci->i_flushing_caps = 0;
5055 list_del_init(&ci->i_flushing_item);
5056 mdsc->num_cap_flushing--;
5057 dirty_dropped = true;
5058 }
5059 spin_unlock(&mdsc->cap_dirty_lock);
5060
5061 if (dirty_dropped) {
5062 mapping_set_error(inode->i_mapping, -EIO);
5063
5064 if (ci->i_wrbuffer_ref_head == 0 &&
5065 ci->i_wr_ref == 0 &&
5066 ci->i_dirty_caps == 0 &&
5067 ci->i_flushing_caps == 0) {
5068 ceph_put_snap_context(ci->i_head_snapc);
5069 ci->i_head_snapc = NULL;
5070 }
5071 }
5072
5073 if (atomic_read(&ci->i_filelock_ref) > 0) {
5074 /* make further file lock syscall return -EIO */
5075 ci->i_ceph_flags |= CEPH_I_ERROR_FILELOCK;
5076 pr_warn_ratelimited_client(cl,
5077 " dropping file locks for %p %llx.%llx\n",
5078 inode, ceph_vinop(inode));
5079 }
5080
5081 if (!ci->i_dirty_caps && ci->i_prealloc_cap_flush) {
5082 cf = ci->i_prealloc_cap_flush;
5083 ci->i_prealloc_cap_flush = NULL;
5084 if (!cf->is_capsnap)
5085 ceph_free_cap_flush(cf);
5086 }
5087
5088 if (!list_empty(&ci->i_cap_snaps))
5089 iputs = remove_capsnaps(mdsc, inode);
5090 }
5091 if (dirty_dropped)
5092 ++iputs;
5093 return iputs;
5094 }
5095
Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.