kernel/fs/ceph/mds_client.c

   1 #include <linux/ceph/ceph_debug.h>
   2
   3 #include <linux/fs.h>
   4 #include <linux/wait.h>
   5 #include <linux/slab.h>
   6 #include <linux/gfp.h>
   7 #include <linux/sched.h>
   8 #include <linux/debugfs.h>
   9 #include <linux/seq_file.h>
  10 #include <linux/utsname.h>
  11 #include <linux/ratelimit.h>
  12
  13 #include "super.h"
  14 #include "mds_client.h"
  15
  16 #include <linux/ceph/ceph_features.h>
  17 #include <linux/ceph/messenger.h>
  18 #include <linux/ceph/decode.h>
  19 #include <linux/ceph/pagelist.h>
  20 #include <linux/ceph/auth.h>
  21 #include <linux/ceph/debugfs.h>
  22
  23 /*
  24  * A cluster of MDS (metadata server) daemons is responsible for
  25  * managing the file system namespace (the directory hierarchy and
  26  * inodes) and for coordinating shared access to storage.  Metadata is
  27  * partitioning hierarchically across a number of servers, and that
  28  * partition varies over time as the cluster adjusts the distribution
  29  * in order to balance load.
  30  *
  31  * The MDS client is primarily responsible to managing synchronous
  32  * metadata requests for operations like open, unlink, and so forth.
  33  * If there is a MDS failure, we find out about it when we (possibly
  34  * request and) receive a new MDS map, and can resubmit affected
  35  * requests.
  36  *
  37  * For the most part, though, we take advantage of a lossless
  38  * communications channel to the MDS, and do not need to worry about
  39  * timing out or resubmitting requests.
  40  *
  41  * We maintain a stateful "session" with each MDS we interact with.
  42  * Within each session, we sent periodic heartbeat messages to ensure
  43  * any capabilities or leases we have been issues remain valid.  If
  44  * the session times out and goes stale, our leases and capabilities
  45  * are no longer valid.
  46  */
  47
  48 struct ceph_reconnect_state {
  49         int nr_caps;
  50         struct ceph_pagelist *pagelist;
  51         bool flock;
  52 };
  53
  54 static void __wake_requests(struct ceph_mds_client *mdsc,
  55                             struct list_head *head);
  56
  57 static const struct ceph_connection_operations mds_con_ops;
  58
  59
  60 /*
  61  * mds reply parsing
  62  */
  63
  64 /*
  65  * parse individual inode info
  66  */
  67 static int parse_reply_info_in(void **p, void *end,
  68                                struct ceph_mds_reply_info_in *info,
  69                                u64 features)
  70 {
  71         int err = -EIO;
  72
  73         info->in = *p;
  74         *p += sizeof(struct ceph_mds_reply_inode) +
  75                 sizeof(*info->in->fragtree.splits) *
  76                 le32_to_cpu(info->in->fragtree.nsplits);
  77
  78         ceph_decode_32_safe(p, end, info->symlink_len, bad);
  79         ceph_decode_need(p, end, info->symlink_len, bad);
  80         info->symlink = *p;
  81         *p += info->symlink_len;
  82
  83         if (features & CEPH_FEATURE_DIRLAYOUTHASH)
  84                 ceph_decode_copy_safe(p, end, &info->dir_layout,
  85                                       sizeof(info->dir_layout), bad);
  86         else
  87                 memset(&info->dir_layout, 0, sizeof(info->dir_layout));
  88
  89         ceph_decode_32_safe(p, end, info->xattr_len, bad);
  90         ceph_decode_need(p, end, info->xattr_len, bad);
  91         info->xattr_data = *p;
  92         *p += info->xattr_len;
  93
  94         if (features & CEPH_FEATURE_MDS_INLINE_DATA) {
  95                 ceph_decode_64_safe(p, end, info->inline_version, bad);
  96                 ceph_decode_32_safe(p, end, info->inline_len, bad);
  97                 ceph_decode_need(p, end, info->inline_len, bad);
  98                 info->inline_data = *p;
  99                 *p += info->inline_len;
 100         } else
 101                 info->inline_version = CEPH_INLINE_NONE;
 102
 103         return 0;
 104 bad:
 105         return err;
 106 }
 107
 108 /*
 109  * parse a normal reply, which may contain a (dir+)dentry and/or a
 110  * target inode.
 111  */
 112 static int parse_reply_info_trace(void **p, void *end,
 113                                   struct ceph_mds_reply_info_parsed *info,
 114                                   u64 features)
 115 {
 116         int err;
 117
 118         if (info->head->is_dentry) {
 119                 err = parse_reply_info_in(p, end, &info->diri, features);
 120                 if (err < 0)
 121                         goto out_bad;
 122
 123                 if (unlikely(*p + sizeof(*info->dirfrag) > end))
 124                         goto bad;
 125                 info->dirfrag = *p;
 126                 *p += sizeof(*info->dirfrag) +
 127                         sizeof(u32)*le32_to_cpu(info->dirfrag->ndist);
 128                 if (unlikely(*p > end))
 129                         goto bad;
 130
 131                 ceph_decode_32_safe(p, end, info->dname_len, bad);
 132                 ceph_decode_need(p, end, info->dname_len, bad);
 133                 info->dname = *p;
 134                 *p += info->dname_len;
 135                 info->dlease = *p;
 136                 *p += sizeof(*info->dlease);
 137         }
 138
 139         if (info->head->is_target) {
 140                 err = parse_reply_info_in(p, end, &info->targeti, features);
 141                 if (err < 0)
 142                         goto out_bad;
 143         }
 144
 145         if (unlikely(*p != end))
 146                 goto bad;
 147         return 0;
 148
 149 bad:
 150         err = -EIO;
 151 out_bad:
 152         pr_err("problem parsing mds trace %d\n", err);
 153         return err;
 154 }
 155
 156 /*
 157  * parse readdir results
 158  */
 159 static int parse_reply_info_dir(void **p, void *end,
 160                                 struct ceph_mds_reply_info_parsed *info,
 161                                 u64 features)
 162 {
 163         u32 num, i = 0;
 164         int err;
 165
 166         info->dir_dir = *p;
 167         if (*p + sizeof(*info->dir_dir) > end)
 168                 goto bad;
 169         *p += sizeof(*info->dir_dir) +
 170                 sizeof(u32)*le32_to_cpu(info->dir_dir->ndist);
 171         if (*p > end)
 172                 goto bad;
 173
 174         ceph_decode_need(p, end, sizeof(num) + 2, bad);
 175         num = ceph_decode_32(p);
 176         info->dir_end = ceph_decode_8(p);
 177         info->dir_complete = ceph_decode_8(p);
 178         if (num == 0)
 179                 goto done;
 180
 181         BUG_ON(!info->dir_in);
 182         info->dir_dname = (void *)(info->dir_in + num);
 183         info->dir_dname_len = (void *)(info->dir_dname + num);
 184         info->dir_dlease = (void *)(info->dir_dname_len + num);
 185         if ((unsigned long)(info->dir_dlease + num) >
 186             (unsigned long)info->dir_in + info->dir_buf_size) {
 187                 pr_err("dir contents are larger than expected\n");
 188                 WARN_ON(1);
 189                 goto bad;
 190         }
 191
 192         info->dir_nr = num;
 193         while (num) {
 194                 /* dentry */
 195                 ceph_decode_need(p, end, sizeof(u32)*2, bad);
 196                 info->dir_dname_len[i] = ceph_decode_32(p);
 197                 ceph_decode_need(p, end, info->dir_dname_len[i], bad);
 198                 info->dir_dname[i] = *p;
 199                 *p += info->dir_dname_len[i];
 200                 dout("parsed dir dname '%.*s'\n", info->dir_dname_len[i],
 201                      info->dir_dname[i]);
 202                 info->dir_dlease[i] = *p;
 203                 *p += sizeof(struct ceph_mds_reply_lease);
 204
 205                 /* inode */
 206                 err = parse_reply_info_in(p, end, &info->dir_in[i], features);
 207                 if (err < 0)
 208                         goto out_bad;
 209                 i++;
 210                 num--;
 211         }
 212
 213 done:
 214         if (*p != end)
 215                 goto bad;
 216         return 0;
 217
 218 bad:
 219         err = -EIO;
 220 out_bad:
 221         pr_err("problem parsing dir contents %d\n", err);
 222         return err;
 223 }
 224
 225 /*
 226  * parse fcntl F_GETLK results
 227  */
 228 static int parse_reply_info_filelock(void **p, void *end,
 229                                      struct ceph_mds_reply_info_parsed *info,
 230                                      u64 features)
 231 {
 232         if (*p + sizeof(*info->filelock_reply) > end)
 233                 goto bad;
 234
 235         info->filelock_reply = *p;
 236         *p += sizeof(*info->filelock_reply);
 237
 238         if (unlikely(*p != end))
 239                 goto bad;
 240         return 0;
 241
 242 bad:
 243         return -EIO;
 244 }
 245
 246 /*
 247  * parse create results
 248  */
 249 static int parse_reply_info_create(void **p, void *end,
 250                                   struct ceph_mds_reply_info_parsed *info,
 251                                   u64 features)
 252 {
 253         if (features & CEPH_FEATURE_REPLY_CREATE_INODE) {
 254                 if (*p == end) {
 255                         info->has_create_ino = false;
 256                 } else {
 257                         info->has_create_ino = true;
 258                         info->ino = ceph_decode_64(p);
 259                 }
 260         }
 261
 262         if (unlikely(*p != end))
 263                 goto bad;
 264         return 0;
 265
 266 bad:
 267         return -EIO;
 268 }
 269
 270 /*
 271  * parse extra results
 272  */
 273 static int parse_reply_info_extra(void **p, void *end,
 274                                   struct ceph_mds_reply_info_parsed *info,
 275                                   u64 features)
 276 {
 277         u32 op = le32_to_cpu(info->head->op);
 278
 279         if (op == CEPH_MDS_OP_GETFILELOCK)
 280                 return parse_reply_info_filelock(p, end, info, features);
 281         else if (op == CEPH_MDS_OP_READDIR || op == CEPH_MDS_OP_LSSNAP)
 282                 return parse_reply_info_dir(p, end, info, features);
 283         else if (op == CEPH_MDS_OP_CREATE)
 284                 return parse_reply_info_create(p, end, info, features);
 285         else
 286                 return -EIO;
 287 }
 288
 289 /*
 290  * parse entire mds reply
 291  */
 292 static int parse_reply_info(struct ceph_msg *msg,
 293                             struct ceph_mds_reply_info_parsed *info,
 294                             u64 features)
 295 {
 296         void *p, *end;
 297         u32 len;
 298         int err;
 299
 300         info->head = msg->front.iov_base;
 301         p = msg->front.iov_base + sizeof(struct ceph_mds_reply_head);
 302         end = p + msg->front.iov_len - sizeof(struct ceph_mds_reply_head);
 303
 304         /* trace */
 305         ceph_decode_32_safe(&p, end, len, bad);
 306         if (len > 0) {
 307                 ceph_decode_need(&p, end, len, bad);
 308                 err = parse_reply_info_trace(&p, p+len, info, features);
 309                 if (err < 0)
 310                         goto out_bad;
 311         }
 312
 313         /* extra */
 314         ceph_decode_32_safe(&p, end, len, bad);
 315         if (len > 0) {
 316                 ceph_decode_need(&p, end, len, bad);
 317                 err = parse_reply_info_extra(&p, p+len, info, features);
 318                 if (err < 0)
 319                         goto out_bad;
 320         }
 321
 322         /* snap blob */
 323         ceph_decode_32_safe(&p, end, len, bad);
 324         info->snapblob_len = len;
 325         info->snapblob = p;
 326         p += len;
 327
 328         if (p != end)
 329                 goto bad;
 330         return 0;
 331
 332 bad:
 333         err = -EIO;
 334 out_bad:
 335         pr_err("mds parse_reply err %d\n", err);
 336         return err;
 337 }
 338
 339 static void destroy_reply_info(struct ceph_mds_reply_info_parsed *info)
 340 {
 341         if (!info->dir_in)
 342                 return;
 343         free_pages((unsigned long)info->dir_in, get_order(info->dir_buf_size));
 344 }
 345
 346
 347 /*
 348  * sessions
 349  */
 350 const char *ceph_session_state_name(int s)
 351 {
 352         switch (s) {
 353         case CEPH_MDS_SESSION_NEW: return "new";
 354         case CEPH_MDS_SESSION_OPENING: return "opening";
 355         case CEPH_MDS_SESSION_OPEN: return "open";
 356         case CEPH_MDS_SESSION_HUNG: return "hung";
 357         case CEPH_MDS_SESSION_CLOSING: return "closing";
 358         case CEPH_MDS_SESSION_RESTARTING: return "restarting";
 359         case CEPH_MDS_SESSION_RECONNECTING: return "reconnecting";
 360         default: return "???";
 361         }
 362 }
 363
 364 static struct ceph_mds_session *get_session(struct ceph_mds_session *s)
 365 {
 366         if (atomic_inc_not_zero(&s->s_ref)) {
 367                 dout("mdsc get_session %p %d -> %d\n", s,
 368                      atomic_read(&s->s_ref)-1, atomic_read(&s->s_ref));
 369                 return s;
 370         } else {
 371                 dout("mdsc get_session %p 0 -- FAIL", s);
 372                 return NULL;
 373         }
 374 }
 375
 376 void ceph_put_mds_session(struct ceph_mds_session *s)
 377 {
 378         dout("mdsc put_session %p %d -> %d\n", s,
 379              atomic_read(&s->s_ref), atomic_read(&s->s_ref)-1);
 380         if (atomic_dec_and_test(&s->s_ref)) {
 381                 if (s->s_auth.authorizer)
 382                         ceph_auth_destroy_authorizer(
 383                                 s->s_mdsc->fsc->client->monc.auth,
 384                                 s->s_auth.authorizer);
 385                 kfree(s);
 386         }
 387 }
 388
 389 /*
 390  * called under mdsc->mutex
 391  */
 392 struct ceph_mds_session *__ceph_lookup_mds_session(struct ceph_mds_client *mdsc,
 393                                                    int mds)
 394 {
 395         struct ceph_mds_session *session;
 396
 397         if (mds >= mdsc->max_sessions || mdsc->sessions[mds] == NULL)
 398                 return NULL;
 399         session = mdsc->sessions[mds];
 400         dout("lookup_mds_session %p %d\n", session,
 401              atomic_read(&session->s_ref));
 402         get_session(session);
 403         return session;
 404 }
 405
 406 static bool __have_session(struct ceph_mds_client *mdsc, int mds)
 407 {
 408         if (mds >= mdsc->max_sessions)
 409                 return false;
 410         return mdsc->sessions[mds];
 411 }
 412
 413 static int __verify_registered_session(struct ceph_mds_client *mdsc,
 414                                        struct ceph_mds_session *s)
 415 {
 416         if (s->s_mds >= mdsc->max_sessions ||
 417             mdsc->sessions[s->s_mds] != s)
 418                 return -ENOENT;
 419         return 0;
 420 }
 421
 422 /*
 423  * create+register a new session for given mds.
 424  * called under mdsc->mutex.
 425  */
 426 static struct ceph_mds_session *register_session(struct ceph_mds_client *mdsc,
 427                                                  int mds)
 428 {
 429         struct ceph_mds_session *s;
 430
 431         if (mds >= mdsc->mdsmap->m_max_mds)
 432                 return ERR_PTR(-EINVAL);
 433
 434         s = kzalloc(sizeof(*s), GFP_NOFS);
 435         if (!s)
 436                 return ERR_PTR(-ENOMEM);
 437         s->s_mdsc = mdsc;
 438         s->s_mds = mds;
 439         s->s_state = CEPH_MDS_SESSION_NEW;
 440         s->s_ttl = 0;
 441         s->s_seq = 0;
 442         mutex_init(&s->s_mutex);
 443
 444         ceph_con_init(&s->s_con, s, &mds_con_ops, &mdsc->fsc->client->msgr);
 445
 446         spin_lock_init(&s->s_gen_ttl_lock);
 447         s->s_cap_gen = 0;
 448         s->s_cap_ttl = jiffies - 1;
 449
 450         spin_lock_init(&s->s_cap_lock);
 451         s->s_renew_requested = 0;
 452         s->s_renew_seq = 0;
 453         INIT_LIST_HEAD(&s->s_caps);
 454         s->s_nr_caps = 0;
 455         s->s_trim_caps = 0;
 456         atomic_set(&s->s_ref, 1);
 457         INIT_LIST_HEAD(&s->s_waiting);
 458         INIT_LIST_HEAD(&s->s_unsafe);
 459         s->s_num_cap_releases = 0;
 460         s->s_cap_reconnect = 0;
 461         s->s_cap_iterator = NULL;
 462         INIT_LIST_HEAD(&s->s_cap_releases);
 463         INIT_LIST_HEAD(&s->s_cap_flushing);
 464         INIT_LIST_HEAD(&s->s_cap_snaps_flushing);
 465
 466         dout("register_session mds%d\n", mds);
 467         if (mds >= mdsc->max_sessions) {
 468                 int newmax = 1 << get_count_order(mds+1);
 469                 struct ceph_mds_session **sa;
 470
 471                 dout("register_session realloc to %d\n", newmax);
 472                 sa = kcalloc(newmax, sizeof(void *), GFP_NOFS);
 473                 if (sa == NULL)
 474                         goto fail_realloc;
 475                 if (mdsc->sessions) {
 476                         memcpy(sa, mdsc->sessions,
 477                                mdsc->max_sessions * sizeof(void *));
 478                         kfree(mdsc->sessions);
 479                 }
 480                 mdsc->sessions = sa;
 481                 mdsc->max_sessions = newmax;
 482         }
 483         mdsc->sessions[mds] = s;
 484         atomic_inc(&mdsc->num_sessions);
 485         atomic_inc(&s->s_ref);  /* one ref to sessions[], one to caller */
 486
 487         ceph_con_open(&s->s_con, CEPH_ENTITY_TYPE_MDS, mds,
 488                       ceph_mdsmap_get_addr(mdsc->mdsmap, mds));
 489
 490         return s;
 491
 492 fail_realloc:
 493         kfree(s);
 494         return ERR_PTR(-ENOMEM);
 495 }
 496
 497 /*
 498  * called under mdsc->mutex
 499  */
 500 static void __unregister_session(struct ceph_mds_client *mdsc,
 501                                struct ceph_mds_session *s)
 502 {
 503         dout("__unregister_session mds%d %p\n", s->s_mds, s);
 504         BUG_ON(mdsc->sessions[s->s_mds] != s);
 505         mdsc->sessions[s->s_mds] = NULL;
 506         ceph_con_close(&s->s_con);
 507         ceph_put_mds_session(s);
 508         atomic_dec(&mdsc->num_sessions);
 509 }
 510
 511 /*
 512  * drop session refs in request.
 513  *
 514  * should be last request ref, or hold mdsc->mutex
 515  */
 516 static void put_request_session(struct ceph_mds_request *req)
 517 {
 518         if (req->r_session) {
 519                 ceph_put_mds_session(req->r_session);
 520                 req->r_session = NULL;
 521         }
 522 }
 523
 524 void ceph_mdsc_release_request(struct kref *kref)
 525 {
 526         struct ceph_mds_request *req = container_of(kref,
 527                                                     struct ceph_mds_request,
 528                                                     r_kref);
 529         destroy_reply_info(&req->r_reply_info);
 530         if (req->r_request)
 531                 ceph_msg_put(req->r_request);
 532         if (req->r_reply)
 533                 ceph_msg_put(req->r_reply);
 534         if (req->r_inode) {
 535                 ceph_put_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
 536                 iput(req->r_inode);
 537         }
 538         if (req->r_locked_dir)
 539                 ceph_put_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN);
 540         iput(req->r_target_inode);
 541         if (req->r_dentry)
 542                 dput(req->r_dentry);
 543         if (req->r_old_dentry)
 544                 dput(req->r_old_dentry);
 545         if (req->r_old_dentry_dir) {
 546                 /*
 547                  * track (and drop pins for) r_old_dentry_dir
 548                  * separately, since r_old_dentry's d_parent may have
 549                  * changed between the dir mutex being dropped and
 550                  * this request being freed.
 551                  */
 552                 ceph_put_cap_refs(ceph_inode(req->r_old_dentry_dir),
 553                                   CEPH_CAP_PIN);
 554                 iput(req->r_old_dentry_dir);
 555         }
 556         kfree(req->r_path1);
 557         kfree(req->r_path2);
 558         if (req->r_pagelist)
 559                 ceph_pagelist_release(req->r_pagelist);
 560         put_request_session(req);
 561         ceph_unreserve_caps(req->r_mdsc, &req->r_caps_reservation);
 562         kfree(req);
 563 }
 564
 565 /*
 566  * lookup session, bump ref if found.
 567  *
 568  * called under mdsc->mutex.
 569  */
 570 static struct ceph_mds_request *__lookup_request(struct ceph_mds_client *mdsc,
 571                                              u64 tid)
 572 {
 573         struct ceph_mds_request *req;
 574         struct rb_node *n = mdsc->request_tree.rb_node;
 575
 576         while (n) {
 577                 req = rb_entry(n, struct ceph_mds_request, r_node);
 578                 if (tid < req->r_tid)
 579                         n = n->rb_left;
 580                 else if (tid > req->r_tid)
 581                         n = n->rb_right;
 582                 else {
 583                         ceph_mdsc_get_request(req);
 584                         return req;
 585                 }
 586         }
 587         return NULL;
 588 }
 589
 590 static void __insert_request(struct ceph_mds_client *mdsc,
 591                              struct ceph_mds_request *new)
 592 {
 593         struct rb_node **p = &mdsc->request_tree.rb_node;
 594         struct rb_node *parent = NULL;
 595         struct ceph_mds_request *req = NULL;
 596
 597         while (*p) {
 598                 parent = *p;
 599                 req = rb_entry(parent, struct ceph_mds_request, r_node);
 600                 if (new->r_tid < req->r_tid)
 601                         p = &(*p)->rb_left;
 602                 else if (new->r_tid > req->r_tid)
 603                         p = &(*p)->rb_right;
 604                 else
 605                         BUG();
 606         }
 607
 608         rb_link_node(&new->r_node, parent, p);
 609         rb_insert_color(&new->r_node, &mdsc->request_tree);
 610 }
 611
 612 /*
 613  * Register an in-flight request, and assign a tid.  Link to directory
 614  * are modifying (if any).
 615  *
 616  * Called under mdsc->mutex.
 617  */
 618 static void __register_request(struct ceph_mds_client *mdsc,
 619                                struct ceph_mds_request *req,
 620                                struct inode *dir)
 621 {
 622         req->r_tid = ++mdsc->last_tid;
 623         if (req->r_num_caps)
 624                 ceph_reserve_caps(mdsc, &req->r_caps_reservation,
 625                                   req->r_num_caps);
 626         dout("__register_request %p tid %lld\n", req, req->r_tid);
 627         ceph_mdsc_get_request(req);
 628         __insert_request(mdsc, req);
 629
 630         req->r_uid = current_fsuid();
 631         req->r_gid = current_fsgid();
 632
 633         if (mdsc->oldest_tid == 0 && req->r_op != CEPH_MDS_OP_SETFILELOCK)
 634                 mdsc->oldest_tid = req->r_tid;
 635
 636         if (dir) {
 637                 ihold(dir);
 638                 req->r_unsafe_dir = dir;
 639         }
 640 }
 641
 642 static void __unregister_request(struct ceph_mds_client *mdsc,
 643                                  struct ceph_mds_request *req)
 644 {
 645         dout("__unregister_request %p tid %lld\n", req, req->r_tid);
 646
 647         if (req->r_tid == mdsc->oldest_tid) {
 648                 struct rb_node *p = rb_next(&req->r_node);
 649                 mdsc->oldest_tid = 0;
 650                 while (p) {
 651                         struct ceph_mds_request *next_req =
 652                                 rb_entry(p, struct ceph_mds_request, r_node);
 653                         if (next_req->r_op != CEPH_MDS_OP_SETFILELOCK) {
 654                                 mdsc->oldest_tid = next_req->r_tid;
 655                                 break;
 656                         }
 657                         p = rb_next(p);
 658                 }
 659         }
 660
 661         rb_erase(&req->r_node, &mdsc->request_tree);
 662         RB_CLEAR_NODE(&req->r_node);
 663
 664         if (req->r_unsafe_dir && req->r_got_unsafe) {
 665                 struct ceph_inode_info *ci = ceph_inode(req->r_unsafe_dir);
 666                 spin_lock(&ci->i_unsafe_lock);
 667                 list_del_init(&req->r_unsafe_dir_item);
 668                 spin_unlock(&ci->i_unsafe_lock);
 669         }
 670         if (req->r_target_inode && req->r_got_unsafe) {
 671                 struct ceph_inode_info *ci = ceph_inode(req->r_target_inode);
 672                 spin_lock(&ci->i_unsafe_lock);
 673                 list_del_init(&req->r_unsafe_target_item);
 674                 spin_unlock(&ci->i_unsafe_lock);
 675         }
 676
 677         if (req->r_unsafe_dir) {
 678                 iput(req->r_unsafe_dir);
 679                 req->r_unsafe_dir = NULL;
 680         }
 681
 682         complete_all(&req->r_safe_completion);
 683
 684         ceph_mdsc_put_request(req);
 685 }
 686
 687 /*
 688  * Choose mds to send request to next.  If there is a hint set in the
 689  * request (e.g., due to a prior forward hint from the mds), use that.
 690  * Otherwise, consult frag tree and/or caps to identify the
 691  * appropriate mds.  If all else fails, choose randomly.
 692  *
 693  * Called under mdsc->mutex.
 694  */
 695 static struct dentry *get_nonsnap_parent(struct dentry *dentry)
 696 {
 697         /*
 698          * we don't need to worry about protecting the d_parent access
 699          * here because we never renaming inside the snapped namespace
 700          * except to resplice to another snapdir, and either the old or new
 701          * result is a valid result.
 702          */
 703         while (!IS_ROOT(dentry) && ceph_snap(d_inode(dentry)) != CEPH_NOSNAP)
 704                 dentry = dentry->d_parent;
 705         return dentry;
 706 }
 707
 708 static int __choose_mds(struct ceph_mds_client *mdsc,
 709                         struct ceph_mds_request *req)
 710 {
 711         struct inode *inode;
 712         struct ceph_inode_info *ci;
 713         struct ceph_cap *cap;
 714         int mode = req->r_direct_mode;
 715         int mds = -1;
 716         u32 hash = req->r_direct_hash;
 717         bool is_hash = req->r_direct_is_hash;
 718
 719         /*
 720          * is there a specific mds we should try?  ignore hint if we have
 721          * no session and the mds is not up (active or recovering).
 722          */
 723         if (req->r_resend_mds >= 0 &&
 724             (__have_session(mdsc, req->r_resend_mds) ||
 725              ceph_mdsmap_get_state(mdsc->mdsmap, req->r_resend_mds) > 0)) {
 726                 dout("choose_mds using resend_mds mds%d\n",
 727                      req->r_resend_mds);
 728                 return req->r_resend_mds;
 729         }
 730
 731         if (mode == USE_RANDOM_MDS)
 732                 goto random;
 733
 734         inode = NULL;
 735         if (req->r_inode) {
 736                 inode = req->r_inode;
 737         } else if (req->r_dentry) {
 738                 /* ignore race with rename; old or new d_parent is okay */
 739                 struct dentry *parent = req->r_dentry->d_parent;
 740                 struct inode *dir = d_inode(parent);
 741
 742                 if (dir->i_sb != mdsc->fsc->sb) {
 743                         /* not this fs! */
 744                         inode = d_inode(req->r_dentry);
 745                 } else if (ceph_snap(dir) != CEPH_NOSNAP) {
 746                         /* direct snapped/virtual snapdir requests
 747                          * based on parent dir inode */
 748                         struct dentry *dn = get_nonsnap_parent(parent);
 749                         inode = d_inode(dn);
 750                         dout("__choose_mds using nonsnap parent %p\n", inode);
 751                 } else {
 752                         /* dentry target */
 753                         inode = d_inode(req->r_dentry);
 754                         if (!inode || mode == USE_AUTH_MDS) {
 755                                 /* dir + name */
 756                                 inode = dir;
 757                                 hash = ceph_dentry_hash(dir, req->r_dentry);
 758                                 is_hash = true;
 759                         }
 760                 }
 761         }
 762
 763         dout("__choose_mds %p is_hash=%d (%d) mode %d\n", inode, (int)is_hash,
 764              (int)hash, mode);
 765         if (!inode)
 766                 goto random;
 767         ci = ceph_inode(inode);
 768
 769         if (is_hash && S_ISDIR(inode->i_mode)) {
 770                 struct ceph_inode_frag frag;
 771                 int found;
 772
 773                 ceph_choose_frag(ci, hash, &frag, &found);
 774                 if (found) {
 775                         if (mode == USE_ANY_MDS && frag.ndist > 0) {
 776                                 u8 r;
 777
 778                                 /* choose a random replica */
 779                                 get_random_bytes(&r, 1);
 780                                 r %= frag.ndist;
 781                                 mds = frag.dist[r];
 782                                 dout("choose_mds %p %llx.%llx "
 783                                      "frag %u mds%d (%d/%d)\n",
 784                                      inode, ceph_vinop(inode),
 785                                      frag.frag, mds,
 786                                      (int)r, frag.ndist);
 787                                 if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=
 788                                     CEPH_MDS_STATE_ACTIVE)
 789                                         return mds;
 790                         }
 791
 792                         /* since this file/dir wasn't known to be
 793                          * replicated, then we want to look for the
 794                          * authoritative mds. */
 795                         mode = USE_AUTH_MDS;
 796                         if (frag.mds >= 0) {
 797                                 /* choose auth mds */
 798                                 mds = frag.mds;
 799                                 dout("choose_mds %p %llx.%llx "
 800                                      "frag %u mds%d (auth)\n",
 801                                      inode, ceph_vinop(inode), frag.frag, mds);
 802                                 if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=
 803                                     CEPH_MDS_STATE_ACTIVE)
 804                                         return mds;
 805                         }
 806                 }
 807         }
 808
 809         spin_lock(&ci->i_ceph_lock);
 810         cap = NULL;
 811         if (mode == USE_AUTH_MDS)
 812                 cap = ci->i_auth_cap;
 813         if (!cap && !RB_EMPTY_ROOT(&ci->i_caps))
 814                 cap = rb_entry(rb_first(&ci->i_caps), struct ceph_cap, ci_node);
 815         if (!cap) {
 816                 spin_unlock(&ci->i_ceph_lock);
 817                 goto random;
 818         }
 819         mds = cap->session->s_mds;
 820         dout("choose_mds %p %llx.%llx mds%d (%scap %p)\n",
 821              inode, ceph_vinop(inode), mds,
 822              cap == ci->i_auth_cap ? "auth " : "", cap);
 823         spin_unlock(&ci->i_ceph_lock);
 824         return mds;
 825
 826 random:
 827         mds = ceph_mdsmap_get_random_mds(mdsc->mdsmap);
 828         dout("choose_mds chose random mds%d\n", mds);
 829         return mds;
 830 }
 831
 832
 833 /*
 834  * session messages
 835  */
 836 static struct ceph_msg *create_session_msg(u32 op, u64 seq)
 837 {
 838         struct ceph_msg *msg;
 839         struct ceph_mds_session_head *h;
 840
 841         msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h), GFP_NOFS,
 842                            false);
 843         if (!msg) {
 844                 pr_err("create_session_msg ENOMEM creating msg\n");
 845                 return NULL;
 846         }
 847         h = msg->front.iov_base;
 848         h->op = cpu_to_le32(op);
 849         h->seq = cpu_to_le64(seq);
 850
 851         return msg;
 852 }
 853
 854 /*
 855  * session message, specialization for CEPH_SESSION_REQUEST_OPEN
 856  * to include additional client metadata fields.
 857  */
 858 static struct ceph_msg *create_session_open_msg(struct ceph_mds_client *mdsc, u64 seq)
 859 {
 860         struct ceph_msg *msg;
 861         struct ceph_mds_session_head *h;
 862         int i = -1;
 863         int metadata_bytes = 0;
 864         int metadata_key_count = 0;
 865         struct ceph_options *opt = mdsc->fsc->client->options;
 866         void *p;
 867
 868         const char* metadata[][2] = {
 869                 {"hostname", utsname()->nodename},
 870                 {"kernel_version", utsname()->release},
 871                 {"entity_id", opt->name ? opt->name : ""},
 872                 {NULL, NULL}
 873         };
 874
 875         /* Calculate serialized length of metadata */
 876         metadata_bytes = 4;  /* map length */
 877         for (i = 0; metadata[i][0] != NULL; ++i) {
 878                 metadata_bytes += 8 + strlen(metadata[i][0]) +
 879                         strlen(metadata[i][1]);
 880                 metadata_key_count++;
 881         }
 882
 883         /* Allocate the message */
 884         msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h) + metadata_bytes,
 885                            GFP_NOFS, false);
 886         if (!msg) {
 887                 pr_err("create_session_msg ENOMEM creating msg\n");
 888                 return NULL;
 889         }
 890         h = msg->front.iov_base;
 891         h->op = cpu_to_le32(CEPH_SESSION_REQUEST_OPEN);
 892         h->seq = cpu_to_le64(seq);
 893
 894         /*
 895          * Serialize client metadata into waiting buffer space, using
 896          * the format that userspace expects for map<string, string>
 897          *
 898          * ClientSession messages with metadata are v2
 899          */
 900         msg->hdr.version = cpu_to_le16(2);
 901         msg->hdr.compat_version = cpu_to_le16(1);
 902
 903         /* The write pointer, following the session_head structure */
 904         p = msg->front.iov_base + sizeof(*h);
 905
 906         /* Number of entries in the map */
 907         ceph_encode_32(&p, metadata_key_count);
 908
 909         /* Two length-prefixed strings for each entry in the map */
 910         for (i = 0; metadata[i][0] != NULL; ++i) {
 911                 size_t const key_len = strlen(metadata[i][0]);
 912                 size_t const val_len = strlen(metadata[i][1]);
 913
 914                 ceph_encode_32(&p, key_len);
 915                 memcpy(p, metadata[i][0], key_len);
 916                 p += key_len;
 917                 ceph_encode_32(&p, val_len);
 918                 memcpy(p, metadata[i][1], val_len);
 919                 p += val_len;
 920         }
 921
 922         return msg;
 923 }
 924
 925 /*
 926  * send session open request.
 927  *
 928  * called under mdsc->mutex
 929  */
 930 static int __open_session(struct ceph_mds_client *mdsc,
 931                           struct ceph_mds_session *session)
 932 {
 933         struct ceph_msg *msg;
 934         int mstate;
 935         int mds = session->s_mds;
 936
 937         /* wait for mds to go active? */
 938         mstate = ceph_mdsmap_get_state(mdsc->mdsmap, mds);
 939         dout("open_session to mds%d (%s)\n", mds,
 940              ceph_mds_state_name(mstate));
 941         session->s_state = CEPH_MDS_SESSION_OPENING;
 942         session->s_renew_requested = jiffies;
 943
 944         /* send connect message */
 945         msg = create_session_open_msg(mdsc, session->s_seq);
 946         if (!msg)
 947                 return -ENOMEM;
 948         ceph_con_send(&session->s_con, msg);
 949         return 0;
 950 }
 951
 952 /*
 953  * open sessions for any export targets for the given mds
 954  *
 955  * called under mdsc->mutex
 956  */
 957 static struct ceph_mds_session *
 958 __open_export_target_session(struct ceph_mds_client *mdsc, int target)
 959 {
 960         struct ceph_mds_session *session;
 961
 962         session = __ceph_lookup_mds_session(mdsc, target);
 963         if (!session) {
 964                 session = register_session(mdsc, target);
 965                 if (IS_ERR(session))
 966                         return session;
 967         }
 968         if (session->s_state == CEPH_MDS_SESSION_NEW ||
 969             session->s_state == CEPH_MDS_SESSION_CLOSING)
 970                 __open_session(mdsc, session);
 971
 972         return session;
 973 }
 974
 975 struct ceph_mds_session *
 976 ceph_mdsc_open_export_target_session(struct ceph_mds_client *mdsc, int target)
 977 {
 978         struct ceph_mds_session *session;
 979
 980         dout("open_export_target_session to mds%d\n", target);
 981
 982         mutex_lock(&mdsc->mutex);
 983         session = __open_export_target_session(mdsc, target);
 984         mutex_unlock(&mdsc->mutex);
 985
 986         return session;
 987 }
 988
 989 static void __open_export_target_sessions(struct ceph_mds_client *mdsc,
 990                                           struct ceph_mds_session *session)
 991 {
 992         struct ceph_mds_info *mi;
 993         struct ceph_mds_session *ts;
 994         int i, mds = session->s_mds;
 995
 996         if (mds >= mdsc->mdsmap->m_max_mds)
 997                 return;
 998
 999         mi = &mdsc->mdsmap->m_info[mds];
1000         dout("open_export_target_sessions for mds%d (%d targets)\n",
1001              session->s_mds, mi->num_export_targets);
1002
1003         for (i = 0; i < mi->num_export_targets; i++) {
1004                 ts = __open_export_target_session(mdsc, mi->export_targets[i]);
1005                 if (!IS_ERR(ts))
1006                         ceph_put_mds_session(ts);
1007         }
1008 }
1009
1010 void ceph_mdsc_open_export_target_sessions(struct ceph_mds_client *mdsc,
1011                                            struct ceph_mds_session *session)
1012 {
1013         mutex_lock(&mdsc->mutex);
1014         __open_export_target_sessions(mdsc, session);
1015         mutex_unlock(&mdsc->mutex);
1016 }
1017
1018 /*
1019  * session caps
1020  */
1021
1022 /* caller holds s_cap_lock, we drop it */
1023 static void cleanup_cap_releases(struct ceph_mds_client *mdsc,
1024                                  struct ceph_mds_session *session)
1025         __releases(session->s_cap_lock)
1026 {
1027         LIST_HEAD(tmp_list);
1028         list_splice_init(&session->s_cap_releases, &tmp_list);
1029         session->s_num_cap_releases = 0;
1030         spin_unlock(&session->s_cap_lock);
1031
1032         dout("cleanup_cap_releases mds%d\n", session->s_mds);
1033         while (!list_empty(&tmp_list)) {
1034                 struct ceph_cap *cap;
1035                 /* zero out the in-progress message */
1036                 cap = list_first_entry(&tmp_list,
1037                                         struct ceph_cap, session_caps);
1038                 list_del(&cap->session_caps);
1039                 ceph_put_cap(mdsc, cap);
1040         }
1041 }
1042
1043 static void cleanup_session_requests(struct ceph_mds_client *mdsc,
1044                                      struct ceph_mds_session *session)
1045 {
1046         struct ceph_mds_request *req;
1047         struct rb_node *p;
1048
1049         dout("cleanup_session_requests mds%d\n", session->s_mds);
1050         mutex_lock(&mdsc->mutex);
1051         while (!list_empty(&session->s_unsafe)) {
1052                 req = list_first_entry(&session->s_unsafe,
1053                                        struct ceph_mds_request, r_unsafe_item);
1054                 list_del_init(&req->r_unsafe_item);
1055                 pr_warn_ratelimited(" dropping unsafe request %llu\n",
1056                                     req->r_tid);
1057                 __unregister_request(mdsc, req);
1058         }
1059         /* zero r_attempts, so kick_requests() will re-send requests */
1060         p = rb_first(&mdsc->request_tree);
1061         while (p) {
1062                 req = rb_entry(p, struct ceph_mds_request, r_node);
1063                 p = rb_next(p);
1064                 if (req->r_session &&
1065                     req->r_session->s_mds == session->s_mds)
1066                         req->r_attempts = 0;
1067         }
1068         mutex_unlock(&mdsc->mutex);
1069 }
1070
1071 /*
1072  * Helper to safely iterate over all caps associated with a session, with
1073  * special care taken to handle a racing __ceph_remove_cap().
1074  *
1075  * Caller must hold session s_mutex.
1076  */
1077 static int iterate_session_caps(struct ceph_mds_session *session,
1078                                  int (*cb)(struct inode *, struct ceph_cap *,
1079                                             void *), void *arg)
1080 {
1081         struct list_head *p;
1082         struct ceph_cap *cap;
1083         struct inode *inode, *last_inode = NULL;
1084         struct ceph_cap *old_cap = NULL;
1085         int ret;
1086
1087         dout("iterate_session_caps %p mds%d\n", session, session->s_mds);
1088         spin_lock(&session->s_cap_lock);
1089         p = session->s_caps.next;
1090         while (p != &session->s_caps) {
1091                 cap = list_entry(p, struct ceph_cap, session_caps);
1092                 inode = igrab(&cap->ci->vfs_inode);
1093                 if (!inode) {
1094                         p = p->next;
1095                         continue;
1096                 }
1097                 session->s_cap_iterator = cap;
1098                 spin_unlock(&session->s_cap_lock);
1099
1100                 if (last_inode) {
1101                         iput(last_inode);
1102                         last_inode = NULL;
1103                 }
1104                 if (old_cap) {
1105                         ceph_put_cap(session->s_mdsc, old_cap);
1106                         old_cap = NULL;
1107                 }
1108
1109                 ret = cb(inode, cap, arg);
1110                 last_inode = inode;
1111
1112                 spin_lock(&session->s_cap_lock);
1113                 p = p->next;
1114                 if (cap->ci == NULL) {
1115                         dout("iterate_session_caps  finishing cap %p removal\n",
1116                              cap);
1117                         BUG_ON(cap->session != session);
1118                         cap->session = NULL;
1119                         list_del_init(&cap->session_caps);
1120                         session->s_nr_caps--;
1121                         if (cap->queue_release) {
1122                                 list_add_tail(&cap->session_caps,
1123                                               &session->s_cap_releases);
1124                                 session->s_num_cap_releases++;
1125                         } else {
1126                                 old_cap = cap;  /* put_cap it w/o locks held */
1127                         }
1128                 }
1129                 if (ret < 0)
1130                         goto out;
1131         }
1132         ret = 0;
1133 out:
1134         session->s_cap_iterator = NULL;
1135         spin_unlock(&session->s_cap_lock);
1136
1137         iput(last_inode);
1138         if (old_cap)
1139                 ceph_put_cap(session->s_mdsc, old_cap);
1140
1141         return ret;
1142 }
1143
1144 static int remove_session_caps_cb(struct inode *inode, struct ceph_cap *cap,
1145                                   void *arg)
1146 {
1147         struct ceph_inode_info *ci = ceph_inode(inode);
1148         LIST_HEAD(to_remove);
1149         int drop = 0;
1150
1151         dout("removing cap %p, ci is %p, inode is %p\n",
1152              cap, ci, &ci->vfs_inode);
1153         spin_lock(&ci->i_ceph_lock);
1154         __ceph_remove_cap(cap, false);
1155         if (!ci->i_auth_cap) {
1156                 struct ceph_cap_flush *cf;
1157                 struct ceph_mds_client *mdsc =
1158                         ceph_sb_to_client(inode->i_sb)->mdsc;
1159
1160                 while (true) {
1161                         struct rb_node *n = rb_first(&ci->i_cap_flush_tree);
1162                         if (!n)
1163                                 break;
1164                         cf = rb_entry(n, struct ceph_cap_flush, i_node);
1165                         rb_erase(&cf->i_node, &ci->i_cap_flush_tree);
1166                         list_add(&cf->list, &to_remove);
1167                 }
1168
1169                 spin_lock(&mdsc->cap_dirty_lock);
1170
1171                 list_for_each_entry(cf, &to_remove, list)
1172                         rb_erase(&cf->g_node, &mdsc->cap_flush_tree);
1173
1174                 if (!list_empty(&ci->i_dirty_item)) {
1175                         pr_warn_ratelimited(
1176                                 " dropping dirty %s state for %p %lld\n",
1177                                 ceph_cap_string(ci->i_dirty_caps),
1178                                 inode, ceph_ino(inode));
1179                         ci->i_dirty_caps = 0;
1180                         list_del_init(&ci->i_dirty_item);
1181                         drop = 1;
1182                 }
1183                 if (!list_empty(&ci->i_flushing_item)) {
1184                         pr_warn_ratelimited(
1185                                 " dropping dirty+flushing %s state for %p %lld\n",
1186                                 ceph_cap_string(ci->i_flushing_caps),
1187                                 inode, ceph_ino(inode));
1188                         ci->i_flushing_caps = 0;
1189                         list_del_init(&ci->i_flushing_item);
1190                         mdsc->num_cap_flushing--;
1191                         drop = 1;
1192                 }
1193                 spin_unlock(&mdsc->cap_dirty_lock);
1194
1195                 if (!ci->i_dirty_caps && ci->i_prealloc_cap_flush) {
1196                         list_add(&ci->i_prealloc_cap_flush->list, &to_remove);
1197                         ci->i_prealloc_cap_flush = NULL;
1198                 }
1199         }
1200         spin_unlock(&ci->i_ceph_lock);
1201         while (!list_empty(&to_remove)) {
1202                 struct ceph_cap_flush *cf;
1203                 cf = list_first_entry(&to_remove,
1204                                       struct ceph_cap_flush, list);
1205                 list_del(&cf->list);
1206                 ceph_free_cap_flush(cf);
1207         }
1208         while (drop--)
1209                 iput(inode);
1210         return 0;
1211 }
1212
1213 /*
1214  * caller must hold session s_mutex
1215  */
1216 static void remove_session_caps(struct ceph_mds_session *session)
1217 {
1218         dout("remove_session_caps on %p\n", session);
1219         iterate_session_caps(session, remove_session_caps_cb, NULL);
1220
1221         spin_lock(&session->s_cap_lock);
1222         if (session->s_nr_caps > 0) {
1223                 struct super_block *sb = session->s_mdsc->fsc->sb;
1224                 struct inode *inode;
1225                 struct ceph_cap *cap, *prev = NULL;
1226                 struct ceph_vino vino;
1227                 /*
1228                  * iterate_session_caps() skips inodes that are being
1229                  * deleted, we need to wait until deletions are complete.
1230                  * __wait_on_freeing_inode() is designed for the job,
1231                  * but it is not exported, so use lookup inode function
1232                  * to access it.
1233                  */
1234                 while (!list_empty(&session->s_caps)) {
1235                         cap = list_entry(session->s_caps.next,
1236                                          struct ceph_cap, session_caps);
1237                         if (cap == prev)
1238                                 break;
1239                         prev = cap;
1240                         vino = cap->ci->i_vino;
1241                         spin_unlock(&session->s_cap_lock);
1242
1243                         inode = ceph_find_inode(sb, vino);
1244                         iput(inode);
1245
1246                         spin_lock(&session->s_cap_lock);
1247                 }
1248         }
1249
1250         // drop cap expires and unlock s_cap_lock
1251         cleanup_cap_releases(session->s_mdsc, session);
1252
1253         BUG_ON(session->s_nr_caps > 0);
1254         BUG_ON(!list_empty(&session->s_cap_flushing));
1255 }
1256
1257 /*
1258  * wake up any threads waiting on this session's caps.  if the cap is
1259  * old (didn't get renewed on the client reconnect), remove it now.
1260  *
1261  * caller must hold s_mutex.
1262  */
1263 static int wake_up_session_cb(struct inode *inode, struct ceph_cap *cap,
1264                               void *arg)
1265 {
1266         struct ceph_inode_info *ci = ceph_inode(inode);
1267
1268         wake_up_all(&ci->i_cap_wq);
1269         if (arg) {
1270                 spin_lock(&ci->i_ceph_lock);
1271                 ci->i_wanted_max_size = 0;
1272                 ci->i_requested_max_size = 0;
1273                 spin_unlock(&ci->i_ceph_lock);
1274         }
1275         return 0;
1276 }
1277
1278 static void wake_up_session_caps(struct ceph_mds_session *session,
1279                                  int reconnect)
1280 {
1281         dout("wake_up_session_caps %p mds%d\n", session, session->s_mds);
1282         iterate_session_caps(session, wake_up_session_cb,
1283                              (void *)(unsigned long)reconnect);
1284 }
1285
1286 /*
1287  * Send periodic message to MDS renewing all currently held caps.  The
1288  * ack will reset the expiration for all caps from this session.
1289  *
1290  * caller holds s_mutex
1291  */
1292 static int send_renew_caps(struct ceph_mds_client *mdsc,
1293                            struct ceph_mds_session *session)
1294 {
1295         struct ceph_msg *msg;
1296         int state;
1297
1298         if (time_after_eq(jiffies, session->s_cap_ttl) &&
1299             time_after_eq(session->s_cap_ttl, session->s_renew_requested))
1300                 pr_info("mds%d caps stale\n", session->s_mds);
1301         session->s_renew_requested = jiffies;
1302
1303         /* do not try to renew caps until a recovering mds has reconnected
1304          * with its clients. */
1305         state = ceph_mdsmap_get_state(mdsc->mdsmap, session->s_mds);
1306         if (state < CEPH_MDS_STATE_RECONNECT) {
1307                 dout("send_renew_caps ignoring mds%d (%s)\n",
1308                      session->s_mds, ceph_mds_state_name(state));
1309                 return 0;
1310         }
1311
1312         dout("send_renew_caps to mds%d (%s)\n", session->s_mds,
1313                 ceph_mds_state_name(state));
1314         msg = create_session_msg(CEPH_SESSION_REQUEST_RENEWCAPS,
1315                                  ++session->s_renew_seq);
1316         if (!msg)
1317                 return -ENOMEM;
1318         ceph_con_send(&session->s_con, msg);
1319         return 0;
1320 }
1321
1322 static int send_flushmsg_ack(struct ceph_mds_client *mdsc,
1323                              struct ceph_mds_session *session, u64 seq)
1324 {
1325         struct ceph_msg *msg;
1326
1327         dout("send_flushmsg_ack to mds%d (%s)s seq %lld\n",
1328              session->s_mds, ceph_session_state_name(session->s_state), seq);
1329         msg = create_session_msg(CEPH_SESSION_FLUSHMSG_ACK, seq);
1330         if (!msg)
1331                 return -ENOMEM;
1332         ceph_con_send(&session->s_con, msg);
1333         return 0;
1334 }
1335
1336
1337 /*
1338  * Note new cap ttl, and any transition from stale -> not stale (fresh?).
1339  *
1340  * Called under session->s_mutex
1341  */
1342 static void renewed_caps(struct ceph_mds_client *mdsc,
1343                          struct ceph_mds_session *session, int is_renew)
1344 {
1345         int was_stale;
1346         int wake = 0;
1347
1348         spin_lock(&session->s_cap_lock);
1349         was_stale = is_renew && time_after_eq(jiffies, session->s_cap_ttl);
1350
1351         session->s_cap_ttl = session->s_renew_requested +
1352                 mdsc->mdsmap->m_session_timeout*HZ;
1353
1354         if (was_stale) {
1355                 if (time_before(jiffies, session->s_cap_ttl)) {
1356                         pr_info("mds%d caps renewed\n", session->s_mds);
1357                         wake = 1;
1358                 } else {
1359                         pr_info("mds%d caps still stale\n", session->s_mds);
1360                 }
1361         }
1362         dout("renewed_caps mds%d ttl now %lu, was %s, now %s\n",
1363              session->s_mds, session->s_cap_ttl, was_stale ? "stale" : "fresh",
1364              time_before(jiffies, session->s_cap_ttl) ? "stale" : "fresh");
1365         spin_unlock(&session->s_cap_lock);
1366
1367         if (wake)
1368                 wake_up_session_caps(session, 0);
1369 }
1370
1371 /*
1372  * send a session close request
1373  */
1374 static int request_close_session(struct ceph_mds_client *mdsc,
1375                                  struct ceph_mds_session *session)
1376 {
1377         struct ceph_msg *msg;
1378
1379         dout("request_close_session mds%d state %s seq %lld\n",
1380              session->s_mds, ceph_session_state_name(session->s_state),
1381              session->s_seq);
1382         msg = create_session_msg(CEPH_SESSION_REQUEST_CLOSE, session->s_seq);
1383         if (!msg)
1384                 return -ENOMEM;
1385         ceph_con_send(&session->s_con, msg);
1386         return 0;
1387 }
1388
1389 /*
1390  * Called with s_mutex held.
1391  */
1392 static int __close_session(struct ceph_mds_client *mdsc,
1393                          struct ceph_mds_session *session)
1394 {
1395         if (session->s_state >= CEPH_MDS_SESSION_CLOSING)
1396                 return 0;
1397         session->s_state = CEPH_MDS_SESSION_CLOSING;
1398         return request_close_session(mdsc, session);
1399 }
1400
1401 /*
1402  * Trim old(er) caps.
1403  *
1404  * Because we can't cache an inode without one or more caps, we do
1405  * this indirectly: if a cap is unused, we prune its aliases, at which
1406  * point the inode will hopefully get dropped to.
1407  *
1408  * Yes, this is a bit sloppy.  Our only real goal here is to respond to
1409  * memory pressure from the MDS, though, so it needn't be perfect.
1410  */
1411 static int trim_caps_cb(struct inode *inode, struct ceph_cap *cap, void *arg)
1412 {
1413         struct ceph_mds_session *session = arg;
1414         struct ceph_inode_info *ci = ceph_inode(inode);
1415         int used, wanted, oissued, mine;
1416
1417         if (session->s_trim_caps <= 0)
1418                 return -1;
1419
1420         spin_lock(&ci->i_ceph_lock);
1421         mine = cap->issued | cap->implemented;
1422         used = __ceph_caps_used(ci);
1423         wanted = __ceph_caps_file_wanted(ci);
1424         oissued = __ceph_caps_issued_other(ci, cap);
1425
1426         dout("trim_caps_cb %p cap %p mine %s oissued %s used %s wanted %s\n",
1427              inode, cap, ceph_cap_string(mine), ceph_cap_string(oissued),
1428              ceph_cap_string(used), ceph_cap_string(wanted));
1429         if (cap == ci->i_auth_cap) {
1430                 if (ci->i_dirty_caps || ci->i_flushing_caps ||
1431                     !list_empty(&ci->i_cap_snaps))
1432                         goto out;
1433                 if ((used | wanted) & CEPH_CAP_ANY_WR)
1434                         goto out;
1435         }
1436         /* The inode has cached pages, but it's no longer used.
1437          * we can safely drop it */
1438         if (wanted == 0 && used == CEPH_CAP_FILE_CACHE &&
1439             !(oissued & CEPH_CAP_FILE_CACHE)) {
1440           used = 0;
1441           oissued = 0;
1442         }
1443         if ((used | wanted) & ~oissued & mine)
1444                 goto out;   /* we need these caps */
1445
1446         session->s_trim_caps--;
1447         if (oissued) {
1448                 /* we aren't the only cap.. just remove us */
1449                 __ceph_remove_cap(cap, true);
1450         } else {
1451                 /* try dropping referring dentries */
1452                 spin_unlock(&ci->i_ceph_lock);
1453                 d_prune_aliases(inode);
1454                 dout("trim_caps_cb %p cap %p  pruned, count now %d\n",
1455                      inode, cap, atomic_read(&inode->i_count));
1456                 return 0;
1457         }
1458
1459 out:
1460         spin_unlock(&ci->i_ceph_lock);
1461         return 0;
1462 }
1463
1464 /*
1465  * Trim session cap count down to some max number.
1466  */
1467 static int trim_caps(struct ceph_mds_client *mdsc,
1468                      struct ceph_mds_session *session,
1469                      int max_caps)
1470 {
1471         int trim_caps = session->s_nr_caps - max_caps;
1472
1473         dout("trim_caps mds%d start: %d / %d, trim %d\n",
1474              session->s_mds, session->s_nr_caps, max_caps, trim_caps);
1475         if (trim_caps > 0) {
1476                 session->s_trim_caps = trim_caps;
1477                 iterate_session_caps(session, trim_caps_cb, session);
1478                 dout("trim_caps mds%d done: %d / %d, trimmed %d\n",
1479                      session->s_mds, session->s_nr_caps, max_caps,
1480                         trim_caps - session->s_trim_caps);
1481                 session->s_trim_caps = 0;
1482         }
1483
1484         ceph_send_cap_releases(mdsc, session);
1485         return 0;
1486 }
1487
1488 static int check_capsnap_flush(struct ceph_inode_info *ci,
1489                                u64 want_snap_seq)
1490 {
1491         int ret = 1;
1492         spin_lock(&ci->i_ceph_lock);
1493         if (want_snap_seq > 0 && !list_empty(&ci->i_cap_snaps)) {
1494                 struct ceph_cap_snap *capsnap =
1495                         list_first_entry(&ci->i_cap_snaps,
1496                                          struct ceph_cap_snap, ci_item);
1497                 ret = capsnap->follows >= want_snap_seq;
1498         }
1499         spin_unlock(&ci->i_ceph_lock);
1500         return ret;
1501 }
1502
1503 static int check_caps_flush(struct ceph_mds_client *mdsc,
1504                             u64 want_flush_tid)
1505 {
1506         struct rb_node *n;
1507         struct ceph_cap_flush *cf;
1508         int ret = 1;
1509
1510         spin_lock(&mdsc->cap_dirty_lock);
1511         n = rb_first(&mdsc->cap_flush_tree);
1512         cf = n ? rb_entry(n, struct ceph_cap_flush, g_node) : NULL;
1513         if (cf && cf->tid <= want_flush_tid) {
1514                 dout("check_caps_flush still flushing tid %llu <= %llu\n",
1515                      cf->tid, want_flush_tid);
1516                 ret = 0;
1517         }
1518         spin_unlock(&mdsc->cap_dirty_lock);
1519         return ret;
1520 }
1521
1522 /*
1523  * flush all dirty inode data to disk.
1524  *
1525  * returns true if we've flushed through want_flush_tid
1526  */
1527 static void wait_caps_flush(struct ceph_mds_client *mdsc,
1528                             u64 want_flush_tid, u64 want_snap_seq)
1529 {
1530         int mds;
1531
1532         dout("check_caps_flush want %llu snap want %llu\n",
1533              want_flush_tid, want_snap_seq);
1534         mutex_lock(&mdsc->mutex);
1535         for (mds = 0; mds < mdsc->max_sessions; ) {
1536                 struct ceph_mds_session *session = mdsc->sessions[mds];
1537                 struct inode *inode = NULL;
1538
1539                 if (!session) {
1540                         mds++;
1541                         continue;
1542                 }
1543                 get_session(session);
1544                 mutex_unlock(&mdsc->mutex);
1545
1546                 mutex_lock(&session->s_mutex);
1547                 if (!list_empty(&session->s_cap_snaps_flushing)) {
1548                         struct ceph_cap_snap *capsnap =
1549                                 list_first_entry(&session->s_cap_snaps_flushing,
1550                                                  struct ceph_cap_snap,
1551                                                  flushing_item);
1552                         struct ceph_inode_info *ci = capsnap->ci;
1553                         if (!check_capsnap_flush(ci, want_snap_seq)) {
1554                                 dout("check_cap_flush still flushing snap %p "
1555                                      "follows %lld <= %lld to mds%d\n",
1556                                      &ci->vfs_inode, capsnap->follows,
1557                                      want_snap_seq, mds);
1558                                 inode = igrab(&ci->vfs_inode);
1559                         }
1560                 }
1561                 mutex_unlock(&session->s_mutex);
1562                 ceph_put_mds_session(session);
1563
1564                 if (inode) {
1565                         wait_event(mdsc->cap_flushing_wq,
1566                                    check_capsnap_flush(ceph_inode(inode),
1567                                                        want_snap_seq));
1568                         iput(inode);
1569                 } else {
1570                         mds++;
1571                 }
1572
1573                 mutex_lock(&mdsc->mutex);
1574         }
1575         mutex_unlock(&mdsc->mutex);
1576
1577         wait_event(mdsc->cap_flushing_wq,
1578                    check_caps_flush(mdsc, want_flush_tid));
1579
1580         dout("check_caps_flush ok, flushed thru %llu\n", want_flush_tid);
1581 }
1582
1583 /*
1584  * called under s_mutex
1585  */
1586 void ceph_send_cap_releases(struct ceph_mds_client *mdsc,
1587                             struct ceph_mds_session *session)
1588 {
1589         struct ceph_msg *msg = NULL;
1590         struct ceph_mds_cap_release *head;
1591         struct ceph_mds_cap_item *item;
1592         struct ceph_cap *cap;
1593         LIST_HEAD(tmp_list);
1594         int num_cap_releases;
1595
1596         spin_lock(&session->s_cap_lock);
1597 again:
1598         list_splice_init(&session->s_cap_releases, &tmp_list);
1599         num_cap_releases = session->s_num_cap_releases;
1600         session->s_num_cap_releases = 0;
1601         spin_unlock(&session->s_cap_lock);
1602
1603         while (!list_empty(&tmp_list)) {
1604                 if (!msg) {
1605                         msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPRELEASE,
1606                                         PAGE_CACHE_SIZE, GFP_NOFS, false);
1607                         if (!msg)
1608                                 goto out_err;
1609                         head = msg->front.iov_base;
1610                         head->num = cpu_to_le32(0);
1611                         msg->front.iov_len = sizeof(*head);
1612                 }
1613                 cap = list_first_entry(&tmp_list, struct ceph_cap,
1614                                         session_caps);
1615                 list_del(&cap->session_caps);
1616                 num_cap_releases--;
1617
1618                 head = msg->front.iov_base;
1619                 le32_add_cpu(&head->num, 1);
1620                 item = msg->front.iov_base + msg->front.iov_len;
1621                 item->ino = cpu_to_le64(cap->cap_ino);
1622                 item->cap_id = cpu_to_le64(cap->cap_id);
1623                 item->migrate_seq = cpu_to_le32(cap->mseq);
1624                 item->seq = cpu_to_le32(cap->issue_seq);
1625                 msg->front.iov_len += sizeof(*item);
1626
1627                 ceph_put_cap(mdsc, cap);
1628
1629                 if (le32_to_cpu(head->num) == CEPH_CAPS_PER_RELEASE) {
1630                         msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
1631                         dout("send_cap_releases mds%d %p\n", session->s_mds, msg);
1632                         ceph_con_send(&session->s_con, msg);
1633                         msg = NULL;
1634                 }
1635         }
1636
1637         BUG_ON(num_cap_releases != 0);
1638
1639         spin_lock(&session->s_cap_lock);
1640         if (!list_empty(&session->s_cap_releases))
1641                 goto again;
1642         spin_unlock(&session->s_cap_lock);
1643
1644         if (msg) {
1645                 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
1646                 dout("send_cap_releases mds%d %p\n", session->s_mds, msg);
1647                 ceph_con_send(&session->s_con, msg);
1648         }
1649         return;
1650 out_err:
1651         pr_err("send_cap_releases mds%d, failed to allocate message\n",
1652                 session->s_mds);
1653         spin_lock(&session->s_cap_lock);
1654         list_splice(&tmp_list, &session->s_cap_releases);
1655         session->s_num_cap_releases += num_cap_releases;
1656         spin_unlock(&session->s_cap_lock);
1657 }
1658
1659 /*
1660  * requests
1661  */
1662
1663 int ceph_alloc_readdir_reply_buffer(struct ceph_mds_request *req,
1664                                     struct inode *dir)
1665 {
1666         struct ceph_inode_info *ci = ceph_inode(dir);
1667         struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
1668         struct ceph_mount_options *opt = req->r_mdsc->fsc->mount_options;
1669         size_t size = sizeof(*rinfo->dir_in) + sizeof(*rinfo->dir_dname_len) +
1670                       sizeof(*rinfo->dir_dname) + sizeof(*rinfo->dir_dlease);
1671         int order, num_entries;
1672
1673         spin_lock(&ci->i_ceph_lock);
1674         num_entries = ci->i_files + ci->i_subdirs;
1675         spin_unlock(&ci->i_ceph_lock);
1676         num_entries = max(num_entries, 1);
1677         num_entries = min(num_entries, opt->max_readdir);
1678
1679         order = get_order(size * num_entries);
1680         while (order >= 0) {
1681                 rinfo->dir_in = (void*)__get_free_pages(GFP_KERNEL |
1682                                                         __GFP_NOWARN,
1683                                                         order);
1684                 if (rinfo->dir_in)
1685                         break;
1686                 order--;
1687         }
1688         if (!rinfo->dir_in)
1689                 return -ENOMEM;
1690
1691         num_entries = (PAGE_SIZE << order) / size;
1692         num_entries = min(num_entries, opt->max_readdir);
1693
1694         rinfo->dir_buf_size = PAGE_SIZE << order;
1695         req->r_num_caps = num_entries + 1;
1696         req->r_args.readdir.max_entries = cpu_to_le32(num_entries);
1697         req->r_args.readdir.max_bytes = cpu_to_le32(opt->max_readdir_bytes);
1698         return 0;
1699 }
1700
1701 /*
1702  * Create an mds request.
1703  */
1704 struct ceph_mds_request *
1705 ceph_mdsc_create_request(struct ceph_mds_client *mdsc, int op, int mode)
1706 {
1707         struct ceph_mds_request *req = kzalloc(sizeof(*req), GFP_NOFS);
1708
1709         if (!req)
1710                 return ERR_PTR(-ENOMEM);
1711
1712         mutex_init(&req->r_fill_mutex);
1713         req->r_mdsc = mdsc;
1714         req->r_started = jiffies;
1715         req->r_resend_mds = -1;
1716         INIT_LIST_HEAD(&req->r_unsafe_dir_item);
1717         INIT_LIST_HEAD(&req->r_unsafe_target_item);
1718         req->r_fmode = -1;
1719         kref_init(&req->r_kref);
1720         INIT_LIST_HEAD(&req->r_wait);
1721         init_completion(&req->r_completion);
1722         init_completion(&req->r_safe_completion);
1723         INIT_LIST_HEAD(&req->r_unsafe_item);
1724
1725         req->r_stamp = CURRENT_TIME;
1726
1727         req->r_op = op;
1728         req->r_direct_mode = mode;
1729         return req;
1730 }
1731
1732 /*
1733  * return oldest (lowest) request, tid in request tree, 0 if none.
1734  *
1735  * called under mdsc->mutex.
1736  */
1737 static struct ceph_mds_request *__get_oldest_req(struct ceph_mds_client *mdsc)
1738 {
1739         if (RB_EMPTY_ROOT(&mdsc->request_tree))
1740                 return NULL;
1741         return rb_entry(rb_first(&mdsc->request_tree),
1742                         struct ceph_mds_request, r_node);
1743 }
1744
1745 static inline  u64 __get_oldest_tid(struct ceph_mds_client *mdsc)
1746 {
1747         return mdsc->oldest_tid;
1748 }
1749
1750 /*
1751  * Build a dentry's path.  Allocate on heap; caller must kfree.  Based
1752  * on build_path_from_dentry in fs/cifs/dir.c.
1753  *
1754  * If @stop_on_nosnap, generate path relative to the first non-snapped
1755  * inode.
1756  *
1757  * Encode hidden .snap dirs as a double /, i.e.
1758  *   foo/.snap/bar -> foo//bar
1759  */
1760 char *ceph_mdsc_build_path(struct dentry *dentry, int *plen, u64 *base,
1761                            int stop_on_nosnap)
1762 {
1763         struct dentry *temp;
1764         char *path;
1765         int len, pos;
1766         unsigned seq;
1767
1768         if (dentry == NULL)
1769                 return ERR_PTR(-EINVAL);
1770
1771 retry:
1772         len = 0;
1773         seq = read_seqbegin(&rename_lock);
1774         rcu_read_lock();
1775         for (temp = dentry; !IS_ROOT(temp);) {
1776                 struct inode *inode = d_inode(temp);
1777                 if (inode && ceph_snap(inode) == CEPH_SNAPDIR)
1778                         len++;  /* slash only */
1779                 else if (stop_on_nosnap && inode &&
1780                          ceph_snap(inode) == CEPH_NOSNAP)
1781                         break;
1782                 else
1783                         len += 1 + temp->d_name.len;
1784                 temp = temp->d_parent;
1785         }
1786         rcu_read_unlock();
1787         if (len)
1788                 len--;  /* no leading '/' */
1789
1790         path = kmalloc(len+1, GFP_NOFS);
1791         if (path == NULL)
1792                 return ERR_PTR(-ENOMEM);
1793         pos = len;
1794         path[pos] = 0;  /* trailing null */
1795         rcu_read_lock();
1796         for (temp = dentry; !IS_ROOT(temp) && pos != 0; ) {
1797                 struct inode *inode;
1798
1799                 spin_lock(&temp->d_lock);
1800                 inode = d_inode(temp);
1801                 if (inode && ceph_snap(inode) == CEPH_SNAPDIR) {
1802                         dout("build_path path+%d: %p SNAPDIR\n",
1803                              pos, temp);
1804                 } else if (stop_on_nosnap && inode &&
1805                            ceph_snap(inode) == CEPH_NOSNAP) {
1806                         spin_unlock(&temp->d_lock);
1807                         break;
1808                 } else {
1809                         pos -= temp->d_name.len;
1810                         if (pos < 0) {
1811                                 spin_unlock(&temp->d_lock);
1812                                 break;
1813                         }
1814                         strncpy(path + pos, temp->d_name.name,
1815                                 temp->d_name.len);
1816                 }
1817                 spin_unlock(&temp->d_lock);
1818                 if (pos)
1819                         path[--pos] = '/';
1820                 temp = temp->d_parent;
1821         }
1822         rcu_read_unlock();
1823         if (pos != 0 || read_seqretry(&rename_lock, seq)) {
1824                 pr_err("build_path did not end path lookup where "
1825                        "expected, namelen is %d, pos is %d\n", len, pos);
1826                 /* presumably this is only possible if racing with a
1827                    rename of one of the parent directories (we can not
1828                    lock the dentries above us to prevent this, but
1829                    retrying should be harmless) */
1830                 kfree(path);
1831                 goto retry;
1832         }
1833
1834         *base = ceph_ino(d_inode(temp));
1835         *plen = len;
1836         dout("build_path on %p %d built %llx '%.*s'\n",
1837              dentry, d_count(dentry), *base, len, path);
1838         return path;
1839 }
1840
1841 static int build_dentry_path(struct dentry *dentry,
1842                              const char **ppath, int *ppathlen, u64 *pino,
1843                              int *pfreepath)
1844 {
1845         char *path;
1846
1847         if (ceph_snap(d_inode(dentry->d_parent)) == CEPH_NOSNAP) {
1848                 *pino = ceph_ino(d_inode(dentry->d_parent));
1849                 *ppath = dentry->d_name.name;
1850                 *ppathlen = dentry->d_name.len;
1851                 return 0;
1852         }
1853         path = ceph_mdsc_build_path(dentry, ppathlen, pino, 1);
1854         if (IS_ERR(path))
1855                 return PTR_ERR(path);
1856         *ppath = path;
1857         *pfreepath = 1;
1858         return 0;
1859 }
1860
1861 static int build_inode_path(struct inode *inode,
1862                             const char **ppath, int *ppathlen, u64 *pino,
1863                             int *pfreepath)
1864 {
1865         struct dentry *dentry;
1866         char *path;
1867
1868         if (ceph_snap(inode) == CEPH_NOSNAP) {
1869                 *pino = ceph_ino(inode);
1870                 *ppathlen = 0;
1871                 return 0;
1872         }
1873         dentry = d_find_alias(inode);
1874         path = ceph_mdsc_build_path(dentry, ppathlen, pino, 1);
1875         dput(dentry);
1876         if (IS_ERR(path))
1877                 return PTR_ERR(path);
1878         *ppath = path;
1879         *pfreepath = 1;
1880         return 0;
1881 }
1882
1883 /*
1884  * request arguments may be specified via an inode *, a dentry *, or
1885  * an explicit ino+path.
1886  */
1887 static int set_request_path_attr(struct inode *rinode, struct dentry *rdentry,
1888                                   const char *rpath, u64 rino,
1889                                   const char **ppath, int *pathlen,
1890                                   u64 *ino, int *freepath)
1891 {
1892         int r = 0;
1893
1894         if (rinode) {
1895                 r = build_inode_path(rinode, ppath, pathlen, ino, freepath);
1896                 dout(" inode %p %llx.%llx\n", rinode, ceph_ino(rinode),
1897                      ceph_snap(rinode));
1898         } else if (rdentry) {
1899                 r = build_dentry_path(rdentry, ppath, pathlen, ino, freepath);
1900                 dout(" dentry %p %llx/%.*s\n", rdentry, *ino, *pathlen,
1901                      *ppath);
1902         } else if (rpath || rino) {
1903                 *ino = rino;
1904                 *ppath = rpath;
1905                 *pathlen = rpath ? strlen(rpath) : 0;
1906                 dout(" path %.*s\n", *pathlen, rpath);
1907         }
1908
1909         return r;
1910 }
1911
1912 /*
1913  * called under mdsc->mutex
1914  */
1915 static struct ceph_msg *create_request_message(struct ceph_mds_client *mdsc,
1916                                                struct ceph_mds_request *req,
1917                                                int mds, bool drop_cap_releases)
1918 {
1919         struct ceph_msg *msg;
1920         struct ceph_mds_request_head *head;
1921         const char *path1 = NULL;
1922         const char *path2 = NULL;
1923         u64 ino1 = 0, ino2 = 0;
1924         int pathlen1 = 0, pathlen2 = 0;
1925         int freepath1 = 0, freepath2 = 0;
1926         int len;
1927         u16 releases;
1928         void *p, *end;
1929         int ret;
1930
1931         ret = set_request_path_attr(req->r_inode, req->r_dentry,
1932                               req->r_path1, req->r_ino1.ino,
1933                               &path1, &pathlen1, &ino1, &freepath1);
1934         if (ret < 0) {
1935                 msg = ERR_PTR(ret);
1936                 goto out;
1937         }
1938
1939         ret = set_request_path_attr(NULL, req->r_old_dentry,
1940                               req->r_path2, req->r_ino2.ino,
1941                               &path2, &pathlen2, &ino2, &freepath2);
1942         if (ret < 0) {
1943                 msg = ERR_PTR(ret);
1944                 goto out_free1;
1945         }
1946
1947         len = sizeof(*head) +
1948                 pathlen1 + pathlen2 + 2*(1 + sizeof(u32) + sizeof(u64)) +
1949                 sizeof(struct ceph_timespec);
1950
1951         /* calculate (max) length for cap releases */
1952         len += sizeof(struct ceph_mds_request_release) *
1953                 (!!req->r_inode_drop + !!req->r_dentry_drop +
1954                  !!req->r_old_inode_drop + !!req->r_old_dentry_drop);
1955         if (req->r_dentry_drop)
1956                 len += req->r_dentry->d_name.len;
1957         if (req->r_old_dentry_drop)
1958                 len += req->r_old_dentry->d_name.len;
1959
1960         msg = ceph_msg_new(CEPH_MSG_CLIENT_REQUEST, len, GFP_NOFS, false);
1961         if (!msg) {
1962                 msg = ERR_PTR(-ENOMEM);
1963                 goto out_free2;
1964         }
1965
1966         msg->hdr.version = cpu_to_le16(2);
1967         msg->hdr.tid = cpu_to_le64(req->r_tid);
1968
1969         head = msg->front.iov_base;
1970         p = msg->front.iov_base + sizeof(*head);
1971         end = msg->front.iov_base + msg->front.iov_len;
1972
1973         head->mdsmap_epoch = cpu_to_le32(mdsc->mdsmap->m_epoch);
1974         head->op = cpu_to_le32(req->r_op);
1975         head->caller_uid = cpu_to_le32(from_kuid(&init_user_ns, req->r_uid));
1976         head->caller_gid = cpu_to_le32(from_kgid(&init_user_ns, req->r_gid));
1977         head->args = req->r_args;
1978
1979         ceph_encode_filepath(&p, end, ino1, path1);
1980         ceph_encode_filepath(&p, end, ino2, path2);
1981
1982         /* make note of release offset, in case we need to replay */
1983         req->r_request_release_offset = p - msg->front.iov_base;
1984
1985         /* cap releases */
1986         releases = 0;
1987         if (req->r_inode_drop)
1988                 releases += ceph_encode_inode_release(&p,
1989                       req->r_inode ? req->r_inode : d_inode(req->r_dentry),
1990                       mds, req->r_inode_drop, req->r_inode_unless, 0);
1991         if (req->r_dentry_drop)
1992                 releases += ceph_encode_dentry_release(&p, req->r_dentry,
1993                        mds, req->r_dentry_drop, req->r_dentry_unless);
1994         if (req->r_old_dentry_drop)
1995                 releases += ceph_encode_dentry_release(&p, req->r_old_dentry,
1996                        mds, req->r_old_dentry_drop, req->r_old_dentry_unless);
1997         if (req->r_old_inode_drop)
1998                 releases += ceph_encode_inode_release(&p,
1999                       d_inode(req->r_old_dentry),
2000                       mds, req->r_old_inode_drop, req->r_old_inode_unless, 0);
2001
2002         if (drop_cap_releases) {
2003                 releases = 0;
2004                 p = msg->front.iov_base + req->r_request_release_offset;
2005         }
2006
2007         head->num_releases = cpu_to_le16(releases);
2008
2009         /* time stamp */
2010         {
2011                 struct ceph_timespec ts;
2012                 ceph_encode_timespec(&ts, &req->r_stamp);
2013                 ceph_encode_copy(&p, &ts, sizeof(ts));
2014         }
2015
2016         BUG_ON(p > end);
2017         msg->front.iov_len = p - msg->front.iov_base;
2018         msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
2019
2020         if (req->r_pagelist) {
2021                 struct ceph_pagelist *pagelist = req->r_pagelist;
2022                 atomic_inc(&pagelist->refcnt);
2023                 ceph_msg_data_add_pagelist(msg, pagelist);
2024                 msg->hdr.data_len = cpu_to_le32(pagelist->length);
2025         } else {
2026                 msg->hdr.data_len = 0;
2027         }
2028
2029         msg->hdr.data_off = cpu_to_le16(0);
2030
2031 out_free2:
2032         if (freepath2)
2033                 kfree((char *)path2);
2034 out_free1:
2035         if (freepath1)
2036                 kfree((char *)path1);
2037 out:
2038         return msg;
2039 }
2040
2041 /*
2042  * called under mdsc->mutex if error, under no mutex if
2043  * success.
2044  */
2045 static void complete_request(struct ceph_mds_client *mdsc,
2046                              struct ceph_mds_request *req)
2047 {
2048         if (req->r_callback)
2049                 req->r_callback(mdsc, req);
2050         else
2051                 complete_all(&req->r_completion);
2052 }
2053
2054 /*
2055  * called under mdsc->mutex
2056  */
2057 static int __prepare_send_request(struct ceph_mds_client *mdsc,
2058                                   struct ceph_mds_request *req,
2059                                   int mds, bool drop_cap_releases)
2060 {
2061         struct ceph_mds_request_head *rhead;
2062         struct ceph_msg *msg;
2063         int flags = 0;
2064
2065         req->r_attempts++;
2066         if (req->r_inode) {
2067                 struct ceph_cap *cap =
2068                         ceph_get_cap_for_mds(ceph_inode(req->r_inode), mds);
2069
2070                 if (cap)
2071                         req->r_sent_on_mseq = cap->mseq;
2072                 else
2073                         req->r_sent_on_mseq = -1;
2074         }
2075         dout("prepare_send_request %p tid %lld %s (attempt %d)\n", req,
2076              req->r_tid, ceph_mds_op_name(req->r_op), req->r_attempts);
2077
2078         if (req->r_got_unsafe) {
2079                 void *p;
2080                 /*
2081                  * Replay.  Do not regenerate message (and rebuild
2082                  * paths, etc.); just use the original message.
2083                  * Rebuilding paths will break for renames because
2084                  * d_move mangles the src name.
2085                  */
2086                 msg = req->r_request;
2087                 rhead = msg->front.iov_base;
2088
2089                 flags = le32_to_cpu(rhead->flags);
2090                 flags |= CEPH_MDS_FLAG_REPLAY;
2091                 rhead->flags = cpu_to_le32(flags);
2092
2093                 if (req->r_target_inode)
2094                         rhead->ino = cpu_to_le64(ceph_ino(req->r_target_inode));
2095
2096                 rhead->num_retry = req->r_attempts - 1;
2097
2098                 /* remove cap/dentry releases from message */
2099                 rhead->num_releases = 0;
2100
2101                 /* time stamp */
2102                 p = msg->front.iov_base + req->r_request_release_offset;
2103                 {
2104                         struct ceph_timespec ts;
2105                         ceph_encode_timespec(&ts, &req->r_stamp);
2106                         ceph_encode_copy(&p, &ts, sizeof(ts));
2107                 }
2108
2109                 msg->front.iov_len = p - msg->front.iov_base;
2110                 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
2111                 return 0;
2112         }
2113
2114         if (req->r_request) {
2115                 ceph_msg_put(req->r_request);
2116                 req->r_request = NULL;
2117         }
2118         msg = create_request_message(mdsc, req, mds, drop_cap_releases);
2119         if (IS_ERR(msg)) {
2120                 req->r_err = PTR_ERR(msg);
2121                 return PTR_ERR(msg);
2122         }
2123         req->r_request = msg;
2124
2125         rhead = msg->front.iov_base;
2126         rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
2127         if (req->r_got_unsafe)
2128                 flags |= CEPH_MDS_FLAG_REPLAY;
2129         if (req->r_locked_dir)
2130                 flags |= CEPH_MDS_FLAG_WANT_DENTRY;
2131         rhead->flags = cpu_to_le32(flags);
2132         rhead->num_fwd = req->r_num_fwd;
2133         rhead->num_retry = req->r_attempts - 1;
2134         rhead->ino = 0;
2135
2136         dout(" r_locked_dir = %p\n", req->r_locked_dir);
2137         return 0;
2138 }
2139
2140 /*
2141  * send request, or put it on the appropriate wait list.
2142  */
2143 static int __do_request(struct ceph_mds_client *mdsc,
2144                         struct ceph_mds_request *req)
2145 {
2146         struct ceph_mds_session *session = NULL;
2147         int mds = -1;
2148         int err = 0;
2149
2150         if (req->r_err || req->r_got_result) {
2151                 if (req->r_aborted)
2152                         __unregister_request(mdsc, req);
2153                 goto out;
2154         }
2155
2156         if (req->r_timeout &&
2157             time_after_eq(jiffies, req->r_started + req->r_timeout)) {
2158                 dout("do_request timed out\n");
2159                 err = -EIO;
2160                 goto finish;
2161         }
2162         if (ACCESS_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) {
2163                 dout("do_request forced umount\n");
2164                 err = -EIO;
2165                 goto finish;
2166         }
2167
2168         put_request_session(req);
2169
2170         mds = __choose_mds(mdsc, req);
2171         if (mds < 0 ||
2172             ceph_mdsmap_get_state(mdsc->mdsmap, mds) < CEPH_MDS_STATE_ACTIVE) {
2173                 dout("do_request no mds or not active, waiting for map\n");
2174                 list_add(&req->r_wait, &mdsc->waiting_for_map);
2175                 goto out;
2176         }
2177
2178         /* get, open session */
2179         session = __ceph_lookup_mds_session(mdsc, mds);
2180         if (!session) {
2181                 session = register_session(mdsc, mds);
2182                 if (IS_ERR(session)) {
2183                         err = PTR_ERR(session);
2184                         goto finish;
2185                 }
2186         }
2187         req->r_session = get_session(session);
2188
2189         dout("do_request mds%d session %p state %s\n", mds, session,
2190              ceph_session_state_name(session->s_state));
2191         if (session->s_state != CEPH_MDS_SESSION_OPEN &&
2192             session->s_state != CEPH_MDS_SESSION_HUNG) {
2193                 if (session->s_state == CEPH_MDS_SESSION_NEW ||
2194                     session->s_state == CEPH_MDS_SESSION_CLOSING)
2195                         __open_session(mdsc, session);
2196                 list_add(&req->r_wait, &session->s_waiting);
2197                 goto out_session;
2198         }
2199
2200         /* send request */
2201         req->r_resend_mds = -1;   /* forget any previous mds hint */
2202
2203         if (req->r_request_started == 0)   /* note request start time */
2204                 req->r_request_started = jiffies;
2205
2206         err = __prepare_send_request(mdsc, req, mds, false);
2207         if (!err) {
2208                 ceph_msg_get(req->r_request);
2209                 ceph_con_send(&session->s_con, req->r_request);
2210         }
2211
2212 out_session:
2213         ceph_put_mds_session(session);
2214 finish:
2215         if (err) {
2216                 dout("__do_request early error %d\n", err);
2217                 req->r_err = err;
2218                 complete_request(mdsc, req);
2219                 __unregister_request(mdsc, req);
2220         }
2221 out:
2222         return err;
2223 }
2224
2225 /*
2226  * called under mdsc->mutex
2227  */
2228 static void __wake_requests(struct ceph_mds_client *mdsc,
2229                             struct list_head *head)
2230 {
2231         struct ceph_mds_request *req;
2232         LIST_HEAD(tmp_list);
2233
2234         list_splice_init(head, &tmp_list);
2235
2236         while (!list_empty(&tmp_list)) {
2237                 req = list_entry(tmp_list.next,
2238                                  struct ceph_mds_request, r_wait);
2239                 list_del_init(&req->r_wait);
2240                 dout(" wake request %p tid %llu\n", req, req->r_tid);
2241                 __do_request(mdsc, req);
2242         }
2243 }
2244
2245 /*
2246  * Wake up threads with requests pending for @mds, so that they can
2247  * resubmit their requests to a possibly different mds.
2248  */
2249 static void kick_requests(struct ceph_mds_client *mdsc, int mds)
2250 {
2251         struct ceph_mds_request *req;
2252         struct rb_node *p = rb_first(&mdsc->request_tree);
2253
2254         dout("kick_requests mds%d\n", mds);
2255         while (p) {
2256                 req = rb_entry(p, struct ceph_mds_request, r_node);
2257                 p = rb_next(p);
2258                 if (req->r_got_unsafe)
2259                         continue;
2260                 if (req->r_attempts > 0)
2261                         continue; /* only new requests */
2262                 if (req->r_session &&
2263                     req->r_session->s_mds == mds) {
2264                         dout(" kicking tid %llu\n", req->r_tid);
2265                         list_del_init(&req->r_wait);
2266                         __do_request(mdsc, req);
2267                 }
2268         }
2269 }
2270
2271 void ceph_mdsc_submit_request(struct ceph_mds_client *mdsc,
2272                               struct ceph_mds_request *req)
2273 {
2274         dout("submit_request on %p\n", req);
2275         mutex_lock(&mdsc->mutex);
2276         __register_request(mdsc, req, NULL);
2277         __do_request(mdsc, req);
2278         mutex_unlock(&mdsc->mutex);
2279 }
2280
2281 /*
2282  * Synchrously perform an mds request.  Take care of all of the
2283  * session setup, forwarding, retry details.
2284  */
2285 int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
2286                          struct inode *dir,
2287                          struct ceph_mds_request *req)
2288 {
2289         int err;
2290
2291         dout("do_request on %p\n", req);
2292
2293         /* take CAP_PIN refs for r_inode, r_locked_dir, r_old_dentry */
2294         if (req->r_inode)
2295                 ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
2296         if (req->r_locked_dir)
2297                 ceph_get_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN);
2298         if (req->r_old_dentry_dir)
2299                 ceph_get_cap_refs(ceph_inode(req->r_old_dentry_dir),
2300                                   CEPH_CAP_PIN);
2301
2302         /* issue */
2303         mutex_lock(&mdsc->mutex);
2304         __register_request(mdsc, req, dir);
2305         __do_request(mdsc, req);
2306
2307         if (req->r_err) {
2308                 err = req->r_err;
2309                 goto out;
2310         }
2311
2312         /* wait */
2313         mutex_unlock(&mdsc->mutex);
2314         dout("do_request waiting\n");
2315         if (!req->r_timeout && req->r_wait_for_completion) {
2316                 err = req->r_wait_for_completion(mdsc, req);
2317         } else {
2318                 long timeleft = wait_for_completion_killable_timeout(
2319                                         &req->r_completion,
2320                                         ceph_timeout_jiffies(req->r_timeout));
2321                 if (timeleft > 0)
2322                         err = 0;
2323                 else if (!timeleft)
2324                         err = -EIO;  /* timed out */
2325                 else
2326                         err = timeleft;  /* killed */
2327         }
2328         dout("do_request waited, got %d\n", err);
2329         mutex_lock(&mdsc->mutex);
2330
2331         /* only abort if we didn't race with a real reply */
2332         if (req->r_got_result) {
2333                 err = le32_to_cpu(req->r_reply_info.head->result);
2334         } else if (err < 0) {
2335                 dout("aborted request %lld with %d\n", req->r_tid, err);
2336
2337                 /*
2338                  * ensure we aren't running concurrently with
2339                  * ceph_fill_trace or ceph_readdir_prepopulate, which
2340                  * rely on locks (dir mutex) held by our caller.
2341                  */
2342                 mutex_lock(&req->r_fill_mutex);
2343                 req->r_err = err;
2344                 req->r_aborted = true;
2345                 mutex_unlock(&req->r_fill_mutex);
2346
2347                 if (req->r_locked_dir &&
2348                     (req->r_op & CEPH_MDS_OP_WRITE))
2349                         ceph_invalidate_dir_request(req);
2350         } else {
2351                 err = req->r_err;
2352         }
2353
2354 out:
2355         mutex_unlock(&mdsc->mutex);
2356         dout("do_request %p done, result %d\n", req, err);
2357         return err;
2358 }
2359
2360 /*
2361  * Invalidate dir's completeness, dentry lease state on an aborted MDS
2362  * namespace request.
2363  */
2364 void ceph_invalidate_dir_request(struct ceph_mds_request *req)
2365 {
2366         struct inode *inode = req->r_locked_dir;
2367
2368         dout("invalidate_dir_request %p (complete, lease(s))\n", inode);
2369
2370         ceph_dir_clear_complete(inode);
2371         if (req->r_dentry)
2372                 ceph_invalidate_dentry_lease(req->r_dentry);
2373         if (req->r_old_dentry)
2374                 ceph_invalidate_dentry_lease(req->r_old_dentry);
2375 }
2376
2377 /*
2378  * Handle mds reply.
2379  *
2380  * We take the session mutex and parse and process the reply immediately.
2381  * This preserves the logical ordering of replies, capabilities, etc., sent
2382  * by the MDS as they are applied to our local cache.
2383  */
2384 static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
2385 {
2386         struct ceph_mds_client *mdsc = session->s_mdsc;
2387         struct ceph_mds_request *req;
2388         struct ceph_mds_reply_head *head = msg->front.iov_base;
2389         struct ceph_mds_reply_info_parsed *rinfo;  /* parsed reply info */
2390         struct ceph_snap_realm *realm;
2391         u64 tid;
2392         int err, result;
2393         int mds = session->s_mds;
2394
2395         if (msg->front.iov_len < sizeof(*head)) {
2396                 pr_err("mdsc_handle_reply got corrupt (short) reply\n");
2397                 ceph_msg_dump(msg);
2398                 return;
2399         }
2400
2401         /* get request, session */
2402         tid = le64_to_cpu(msg->hdr.tid);
2403         mutex_lock(&mdsc->mutex);
2404         req = __lookup_request(mdsc, tid);
2405         if (!req) {
2406                 dout("handle_reply on unknown tid %llu\n", tid);
2407                 mutex_unlock(&mdsc->mutex);
2408                 return;
2409         }
2410         dout("handle_reply %p\n", req);
2411
2412         /* correct session? */
2413         if (req->r_session != session) {
2414                 pr_err("mdsc_handle_reply got %llu on session mds%d"
2415                        " not mds%d\n", tid, session->s_mds,
2416                        req->r_session ? req->r_session->s_mds : -1);
2417                 mutex_unlock(&mdsc->mutex);
2418                 goto out;
2419         }
2420
2421         /* dup? */
2422         if ((req->r_got_unsafe && !head->safe) ||
2423             (req->r_got_safe && head->safe)) {
2424                 pr_warn("got a dup %s reply on %llu from mds%d\n",
2425                            head->safe ? "safe" : "unsafe", tid, mds);
2426                 mutex_unlock(&mdsc->mutex);
2427                 goto out;
2428         }
2429         if (req->r_got_safe) {
2430                 pr_warn("got unsafe after safe on %llu from mds%d\n",
2431                            tid, mds);
2432                 mutex_unlock(&mdsc->mutex);
2433                 goto out;
2434         }
2435
2436         result = le32_to_cpu(head->result);
2437
2438         /*
2439          * Handle an ESTALE
2440          * if we're not talking to the authority, send to them
2441          * if the authority has changed while we weren't looking,
2442          * send to new authority
2443          * Otherwise we just have to return an ESTALE
2444          */
2445         if (result == -ESTALE) {
2446                 dout("got ESTALE on request %llu", req->r_tid);
2447                 req->r_resend_mds = -1;
2448                 if (req->r_direct_mode != USE_AUTH_MDS) {
2449                         dout("not using auth, setting for that now");
2450                         req->r_direct_mode = USE_AUTH_MDS;
2451                         __do_request(mdsc, req);
2452                         mutex_unlock(&mdsc->mutex);
2453                         goto out;
2454                 } else  {
2455                         int mds = __choose_mds(mdsc, req);
2456                         if (mds >= 0 && mds != req->r_session->s_mds) {
2457                                 dout("but auth changed, so resending");
2458                                 __do_request(mdsc, req);
2459                                 mutex_unlock(&mdsc->mutex);
2460                                 goto out;
2461                         }
2462                 }
2463                 dout("have to return ESTALE on request %llu", req->r_tid);
2464         }
2465
2466
2467         if (head->safe) {
2468                 req->r_got_safe = true;
2469                 __unregister_request(mdsc, req);
2470
2471                 if (req->r_got_unsafe) {
2472                         /*
2473                          * We already handled the unsafe response, now do the
2474                          * cleanup.  No need to examine the response; the MDS
2475                          * doesn't include any result info in the safe
2476                          * response.  And even if it did, there is nothing
2477                          * useful we could do with a revised return value.
2478                          */
2479                         dout("got safe reply %llu, mds%d\n", tid, mds);
2480                         list_del_init(&req->r_unsafe_item);
2481
2482                         /* last unsafe request during umount? */
2483                         if (mdsc->stopping && !__get_oldest_req(mdsc))
2484                                 complete_all(&mdsc->safe_umount_waiters);
2485                         mutex_unlock(&mdsc->mutex);
2486                         goto out;
2487                 }
2488         } else {
2489                 req->r_got_unsafe = true;
2490                 list_add_tail(&req->r_unsafe_item, &req->r_session->s_unsafe);
2491                 if (req->r_unsafe_dir) {
2492                         struct ceph_inode_info *ci =
2493                                         ceph_inode(req->r_unsafe_dir);
2494                         spin_lock(&ci->i_unsafe_lock);
2495                         list_add_tail(&req->r_unsafe_dir_item,
2496                                       &ci->i_unsafe_dirops);
2497                         spin_unlock(&ci->i_unsafe_lock);
2498                 }
2499         }
2500
2501         dout("handle_reply tid %lld result %d\n", tid, result);
2502         rinfo = &req->r_reply_info;
2503         err = parse_reply_info(msg, rinfo, session->s_con.peer_features);
2504         mutex_unlock(&mdsc->mutex);
2505
2506         mutex_lock(&session->s_mutex);
2507         if (err < 0) {
2508                 pr_err("mdsc_handle_reply got corrupt reply mds%d(tid:%lld)\n", mds, tid);
2509                 ceph_msg_dump(msg);
2510                 goto out_err;
2511         }
2512
2513         /* snap trace */
2514         realm = NULL;
2515         if (rinfo->snapblob_len) {
2516                 down_write(&mdsc->snap_rwsem);
2517                 ceph_update_snap_trace(mdsc, rinfo->snapblob,
2518                                 rinfo->snapblob + rinfo->snapblob_len,
2519                                 le32_to_cpu(head->op) == CEPH_MDS_OP_RMSNAP,
2520                                 &realm);
2521                 downgrade_write(&mdsc->snap_rwsem);
2522         } else {
2523                 down_read(&mdsc->snap_rwsem);
2524         }
2525
2526         /* insert trace into our cache */
2527         mutex_lock(&req->r_fill_mutex);
2528         err = ceph_fill_trace(mdsc->fsc->sb, req, req->r_session);
2529         if (err == 0) {
2530                 if (result == 0 && (req->r_op == CEPH_MDS_OP_READDIR ||
2531                                     req->r_op == CEPH_MDS_OP_LSSNAP))
2532                         ceph_readdir_prepopulate(req, req->r_session);
2533                 ceph_unreserve_caps(mdsc, &req->r_caps_reservation);
2534         }
2535         mutex_unlock(&req->r_fill_mutex);
2536
2537         up_read(&mdsc->snap_rwsem);
2538         if (realm)
2539                 ceph_put_snap_realm(mdsc, realm);
2540
2541         if (err == 0 && req->r_got_unsafe && req->r_target_inode) {
2542                 struct ceph_inode_info *ci = ceph_inode(req->r_target_inode);
2543                 spin_lock(&ci->i_unsafe_lock);
2544                 list_add_tail(&req->r_unsafe_target_item, &ci->i_unsafe_iops);
2545                 spin_unlock(&ci->i_unsafe_lock);
2546         }
2547 out_err:
2548         mutex_lock(&mdsc->mutex);
2549         if (!req->r_aborted) {
2550                 if (err) {
2551                         req->r_err = err;
2552                 } else {
2553                         req->r_reply =  ceph_msg_get(msg);
2554                         req->r_got_result = true;
2555                 }
2556         } else {
2557                 dout("reply arrived after request %lld was aborted\n", tid);
2558         }
2559         mutex_unlock(&mdsc->mutex);
2560
2561         mutex_unlock(&session->s_mutex);
2562
2563         /* kick calling process */
2564         complete_request(mdsc, req);
2565 out:
2566         ceph_mdsc_put_request(req);
2567         return;
2568 }
2569
2570
2571
2572 /*
2573  * handle mds notification that our request has been forwarded.
2574  */
2575 static void handle_forward(struct ceph_mds_client *mdsc,
2576                            struct ceph_mds_session *session,
2577                            struct ceph_msg *msg)
2578 {
2579         struct ceph_mds_request *req;
2580         u64 tid = le64_to_cpu(msg->hdr.tid);
2581         u32 next_mds;
2582         u32 fwd_seq;
2583         int err = -EINVAL;
2584         void *p = msg->front.iov_base;
2585         void *end = p + msg->front.iov_len;
2586
2587         ceph_decode_need(&p, end, 2*sizeof(u32), bad);
2588         next_mds = ceph_decode_32(&p);
2589         fwd_seq = ceph_decode_32(&p);
2590
2591         mutex_lock(&mdsc->mutex);
2592         req = __lookup_request(mdsc, tid);
2593         if (!req) {
2594                 dout("forward tid %llu to mds%d - req dne\n", tid, next_mds);
2595                 goto out;  /* dup reply? */
2596         }
2597
2598         if (req->r_aborted) {
2599                 dout("forward tid %llu aborted, unregistering\n", tid);
2600                 __unregister_request(mdsc, req);
2601         } else if (fwd_seq <= req->r_num_fwd) {
2602                 dout("forward tid %llu to mds%d - old seq %d <= %d\n",
2603                      tid, next_mds, req->r_num_fwd, fwd_seq);
2604         } else {
2605                 /* resend. forward race not possible; mds would drop */
2606                 dout("forward tid %llu to mds%d (we resend)\n", tid, next_mds);
2607                 BUG_ON(req->r_err);
2608                 BUG_ON(req->r_got_result);
2609                 req->r_attempts = 0;
2610                 req->r_num_fwd = fwd_seq;
2611                 req->r_resend_mds = next_mds;
2612                 put_request_session(req);
2613                 __do_request(mdsc, req);
2614         }
2615         ceph_mdsc_put_request(req);
2616 out:
2617         mutex_unlock(&mdsc->mutex);
2618         return;
2619
2620 bad:
2621         pr_err("mdsc_handle_forward decode error err=%d\n", err);
2622 }
2623
2624 /*
2625  * handle a mds session control message
2626  */
2627 static void handle_session(struct ceph_mds_session *session,
2628                            struct ceph_msg *msg)
2629 {
2630         struct ceph_mds_client *mdsc = session->s_mdsc;
2631         u32 op;
2632         u64 seq;
2633         int mds = session->s_mds;
2634         struct ceph_mds_session_head *h = msg->front.iov_base;
2635         int wake = 0;
2636
2637         /* decode */
2638         if (msg->front.iov_len != sizeof(*h))
2639                 goto bad;
2640         op = le32_to_cpu(h->op);
2641         seq = le64_to_cpu(h->seq);
2642
2643         mutex_lock(&mdsc->mutex);
2644         if (op == CEPH_SESSION_CLOSE)
2645                 __unregister_session(mdsc, session);
2646         /* FIXME: this ttl calculation is generous */
2647         session->s_ttl = jiffies + HZ*mdsc->mdsmap->m_session_autoclose;
2648         mutex_unlock(&mdsc->mutex);
2649
2650         mutex_lock(&session->s_mutex);
2651
2652         dout("handle_session mds%d %s %p state %s seq %llu\n",
2653              mds, ceph_session_op_name(op), session,
2654              ceph_session_state_name(session->s_state), seq);
2655
2656         if (session->s_state == CEPH_MDS_SESSION_HUNG) {
2657                 session->s_state = CEPH_MDS_SESSION_OPEN;
2658                 pr_info("mds%d came back\n", session->s_mds);
2659         }
2660
2661         switch (op) {
2662         case CEPH_SESSION_OPEN:
2663                 if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
2664                         pr_info("mds%d reconnect success\n", session->s_mds);
2665                 session->s_state = CEPH_MDS_SESSION_OPEN;
2666                 renewed_caps(mdsc, session, 0);
2667                 wake = 1;
2668                 if (mdsc->stopping)
2669                         __close_session(mdsc, session);
2670                 break;
2671
2672         case CEPH_SESSION_RENEWCAPS:
2673                 if (session->s_renew_seq == seq)
2674                         renewed_caps(mdsc, session, 1);
2675                 break;
2676
2677         case CEPH_SESSION_CLOSE:
2678                 if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
2679                         pr_info("mds%d reconnect denied\n", session->s_mds);
2680                 cleanup_session_requests(mdsc, session);
2681                 remove_session_caps(session);
2682                 wake = 2; /* for good measure */
2683                 wake_up_all(&mdsc->session_close_wq);
2684                 break;
2685
2686         case CEPH_SESSION_STALE:
2687                 pr_info("mds%d caps went stale, renewing\n",
2688                         session->s_mds);
2689                 spin_lock(&session->s_gen_ttl_lock);
2690                 session->s_cap_gen++;
2691                 session->s_cap_ttl = jiffies - 1;
2692                 spin_unlock(&session->s_gen_ttl_lock);
2693                 send_renew_caps(mdsc, session);
2694                 break;
2695
2696         case CEPH_SESSION_RECALL_STATE:
2697                 trim_caps(mdsc, session, le32_to_cpu(h->max_caps));
2698                 break;
2699
2700         case CEPH_SESSION_FLUSHMSG:
2701                 send_flushmsg_ack(mdsc, session, seq);
2702                 break;
2703
2704         case CEPH_SESSION_FORCE_RO:
2705                 dout("force_session_readonly %p\n", session);
2706                 spin_lock(&session->s_cap_lock);
2707                 session->s_readonly = true;
2708                 spin_unlock(&session->s_cap_lock);
2709                 wake_up_session_caps(session, 0);
2710                 break;
2711
2712         default:
2713                 pr_err("mdsc_handle_session bad op %d mds%d\n", op, mds);
2714                 WARN_ON(1);
2715         }
2716
2717         mutex_unlock(&session->s_mutex);
2718         if (wake) {
2719                 mutex_lock(&mdsc->mutex);
2720                 __wake_requests(mdsc, &session->s_waiting);
2721                 if (wake == 2)
2722                         kick_requests(mdsc, mds);
2723                 mutex_unlock(&mdsc->mutex);
2724         }
2725         return;
2726
2727 bad:
2728         pr_err("mdsc_handle_session corrupt message mds%d len %d\n", mds,
2729                (int)msg->front.iov_len);
2730         ceph_msg_dump(msg);
2731         return;
2732 }
2733
2734
2735 /*
2736  * called under session->mutex.
2737  */
2738 static void replay_unsafe_requests(struct ceph_mds_client *mdsc,
2739                                    struct ceph_mds_session *session)
2740 {
2741         struct ceph_mds_request *req, *nreq;
2742         struct rb_node *p;
2743         int err;
2744
2745         dout("replay_unsafe_requests mds%d\n", session->s_mds);
2746
2747         mutex_lock(&mdsc->mutex);
2748         list_for_each_entry_safe(req, nreq, &session->s_unsafe, r_unsafe_item) {
2749                 err = __prepare_send_request(mdsc, req, session->s_mds, true);
2750                 if (!err) {
2751                         ceph_msg_get(req->r_request);
2752                         ceph_con_send(&session->s_con, req->r_request);
2753                 }
2754         }
2755
2756         /*
2757          * also re-send old requests when MDS enters reconnect stage. So that MDS
2758          * can process completed request in clientreplay stage.
2759          */
2760         p = rb_first(&mdsc->request_tree);
2761         while (p) {
2762                 req = rb_entry(p, struct ceph_mds_request, r_node);
2763                 p = rb_next(p);
2764                 if (req->r_got_unsafe)
2765                         continue;
2766                 if (req->r_attempts == 0)
2767                         continue; /* only old requests */
2768                 if (req->r_session &&
2769                     req->r_session->s_mds == session->s_mds) {
2770                         err = __prepare_send_request(mdsc, req,
2771                                                      session->s_mds, true);
2772                         if (!err) {
2773                                 ceph_msg_get(req->r_request);
2774                                 ceph_con_send(&session->s_con, req->r_request);
2775                         }
2776                 }
2777         }
2778         mutex_unlock(&mdsc->mutex);
2779 }
2780
2781 /*
2782  * Encode information about a cap for a reconnect with the MDS.
2783  */
2784 static int encode_caps_cb(struct inode *inode, struct ceph_cap *cap,
2785                           void *arg)
2786 {
2787         union {
2788                 struct ceph_mds_cap_reconnect v2;
2789                 struct ceph_mds_cap_reconnect_v1 v1;
2790         } rec;
2791         size_t reclen;
2792         struct ceph_inode_info *ci;
2793         struct ceph_reconnect_state *recon_state = arg;
2794         struct ceph_pagelist *pagelist = recon_state->pagelist;
2795         char *path;
2796         int pathlen, err;
2797         u64 pathbase;
2798         struct dentry *dentry;
2799
2800         ci = cap->ci;
2801
2802         dout(" adding %p ino %llx.%llx cap %p %lld %s\n",
2803              inode, ceph_vinop(inode), cap, cap->cap_id,
2804              ceph_cap_string(cap->issued));
2805         err = ceph_pagelist_encode_64(pagelist, ceph_ino(inode));
2806         if (err)
2807                 return err;
2808
2809         dentry = d_find_alias(inode);
2810         if (dentry) {
2811                 path = ceph_mdsc_build_path(dentry, &pathlen, &pathbase, 0);
2812                 if (IS_ERR(path)) {
2813                         err = PTR_ERR(path);
2814                         goto out_dput;
2815                 }
2816         } else {
2817                 path = NULL;
2818                 pathlen = 0;
2819         }
2820         err = ceph_pagelist_encode_string(pagelist, path, pathlen);
2821         if (err)
2822                 goto out_free;
2823
2824         spin_lock(&ci->i_ceph_lock);
2825         cap->seq = 0;        /* reset cap seq */
2826         cap->issue_seq = 0;  /* and issue_seq */
2827         cap->mseq = 0;       /* and migrate_seq */
2828         cap->cap_gen = cap->session->s_cap_gen;
2829
2830         if (recon_state->flock) {
2831                 rec.v2.cap_id = cpu_to_le64(cap->cap_id);
2832                 rec.v2.wanted = cpu_to_le32(__ceph_caps_wanted(ci));
2833                 rec.v2.issued = cpu_to_le32(cap->issued);
2834                 rec.v2.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
2835                 rec.v2.pathbase = cpu_to_le64(pathbase);
2836                 rec.v2.flock_len = 0;
2837                 reclen = sizeof(rec.v2);
2838         } else {
2839                 rec.v1.cap_id = cpu_to_le64(cap->cap_id);
2840                 rec.v1.wanted = cpu_to_le32(__ceph_caps_wanted(ci));
2841                 rec.v1.issued = cpu_to_le32(cap->issued);
2842                 rec.v1.size = cpu_to_le64(inode->i_size);
2843                 ceph_encode_timespec(&rec.v1.mtime, &inode->i_mtime);
2844                 ceph_encode_timespec(&rec.v1.atime, &inode->i_atime);
2845                 rec.v1.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
2846                 rec.v1.pathbase = cpu_to_le64(pathbase);
2847                 reclen = sizeof(rec.v1);
2848         }
2849         spin_unlock(&ci->i_ceph_lock);
2850
2851         if (recon_state->flock) {
2852                 int num_fcntl_locks, num_flock_locks;
2853                 struct ceph_filelock *flocks;
2854
2855 encode_again:
2856                 ceph_count_locks(inode, &num_fcntl_locks, &num_flock_locks);
2857                 flocks = kmalloc((num_fcntl_locks+num_flock_locks) *
2858                                  sizeof(struct ceph_filelock), GFP_NOFS);
2859                 if (!flocks) {
2860                         err = -ENOMEM;
2861                         goto out_free;
2862                 }
2863                 err = ceph_encode_locks_to_buffer(inode, flocks,
2864                                                   num_fcntl_locks,
2865                                                   num_flock_locks);
2866                 if (err) {
2867                         kfree(flocks);
2868                         if (err == -ENOSPC)
2869                                 goto encode_again;
2870                         goto out_free;
2871                 }
2872                 /*
2873                  * number of encoded locks is stable, so copy to pagelist
2874                  */
2875                 rec.v2.flock_len = cpu_to_le32(2*sizeof(u32) +
2876                                     (num_fcntl_locks+num_flock_locks) *
2877                                     sizeof(struct ceph_filelock));
2878                 err = ceph_pagelist_append(pagelist, &rec, reclen);
2879                 if (!err)
2880                         err = ceph_locks_to_pagelist(flocks, pagelist,
2881                                                      num_fcntl_locks,
2882                                                      num_flock_locks);
2883                 kfree(flocks);
2884         } else {
2885                 err = ceph_pagelist_append(pagelist, &rec, reclen);
2886         }
2887
2888         recon_state->nr_caps++;
2889 out_free:
2890         kfree(path);
2891 out_dput:
2892         dput(dentry);
2893         return err;
2894 }
2895
2896
2897 /*
2898  * If an MDS fails and recovers, clients need to reconnect in order to
2899  * reestablish shared state.  This includes all caps issued through
2900  * this session _and_ the snap_realm hierarchy.  Because it's not
2901  * clear which snap realms the mds cares about, we send everything we
2902  * know about.. that ensures we'll then get any new info the
2903  * recovering MDS might have.
2904  *
2905  * This is a relatively heavyweight operation, but it's rare.
2906  *
2907  * called with mdsc->mutex held.
2908  */
2909 static void send_mds_reconnect(struct ceph_mds_client *mdsc,
2910                                struct ceph_mds_session *session)
2911 {
2912         struct ceph_msg *reply;
2913         struct rb_node *p;
2914         int mds = session->s_mds;
2915         int err = -ENOMEM;
2916         int s_nr_caps;
2917         struct ceph_pagelist *pagelist;
2918         struct ceph_reconnect_state recon_state;
2919
2920         pr_info("mds%d reconnect start\n", mds);
2921
2922         pagelist = kmalloc(sizeof(*pagelist), GFP_NOFS);
2923         if (!pagelist)
2924                 goto fail_nopagelist;
2925         ceph_pagelist_init(pagelist);
2926
2927         reply = ceph_msg_new(CEPH_MSG_CLIENT_RECONNECT, 0, GFP_NOFS, false);
2928         if (!reply)
2929                 goto fail_nomsg;
2930
2931         mutex_lock(&session->s_mutex);
2932         session->s_state = CEPH_MDS_SESSION_RECONNECTING;
2933         session->s_seq = 0;
2934
2935         dout("session %p state %s\n", session,
2936              ceph_session_state_name(session->s_state));
2937
2938         spin_lock(&session->s_gen_ttl_lock);
2939         session->s_cap_gen++;
2940         spin_unlock(&session->s_gen_ttl_lock);
2941
2942         spin_lock(&session->s_cap_lock);
2943         /* don't know if session is readonly */
2944         session->s_readonly = 0;
2945         /*
2946          * notify __ceph_remove_cap() that we are composing cap reconnect.
2947          * If a cap get released before being added to the cap reconnect,
2948          * __ceph_remove_cap() should skip queuing cap release.
2949          */
2950         session->s_cap_reconnect = 1;
2951         /* drop old cap expires; we're about to reestablish that state */
2952         cleanup_cap_releases(mdsc, session);
2953
2954         /* trim unused caps to reduce MDS's cache rejoin time */
2955         if (mdsc->fsc->sb->s_root)
2956                 shrink_dcache_parent(mdsc->fsc->sb->s_root);
2957
2958         ceph_con_close(&session->s_con);
2959         ceph_con_open(&session->s_con,
2960                       CEPH_ENTITY_TYPE_MDS, mds,
2961                       ceph_mdsmap_get_addr(mdsc->mdsmap, mds));
2962
2963         /* replay unsafe requests */
2964         replay_unsafe_requests(mdsc, session);
2965
2966         down_read(&mdsc->snap_rwsem);
2967
2968         /* traverse this session's caps */
2969         s_nr_caps = session->s_nr_caps;
2970         err = ceph_pagelist_encode_32(pagelist, s_nr_caps);
2971         if (err)
2972                 goto fail;
2973
2974         recon_state.nr_caps = 0;
2975         recon_state.pagelist = pagelist;
2976         recon_state.flock = session->s_con.peer_features & CEPH_FEATURE_FLOCK;
2977         err = iterate_session_caps(session, encode_caps_cb, &recon_state);
2978         if (err < 0)
2979                 goto fail;
2980
2981         spin_lock(&session->s_cap_lock);
2982         session->s_cap_reconnect = 0;
2983         spin_unlock(&session->s_cap_lock);
2984
2985         /*
2986          * snaprealms.  we provide mds with the ino, seq (version), and
2987          * parent for all of our realms.  If the mds has any newer info,
2988          * it will tell us.
2989          */
2990         for (p = rb_first(&mdsc->snap_realms); p; p = rb_next(p)) {
2991                 struct ceph_snap_realm *realm =
2992                         rb_entry(p, struct ceph_snap_realm, node);
2993                 struct ceph_mds_snaprealm_reconnect sr_rec;
2994
2995                 dout(" adding snap realm %llx seq %lld parent %llx\n",
2996                      realm->ino, realm->seq, realm->parent_ino);
2997                 sr_rec.ino = cpu_to_le64(realm->ino);
2998                 sr_rec.seq = cpu_to_le64(realm->seq);
2999                 sr_rec.parent = cpu_to_le64(realm->parent_ino);
3000                 err = ceph_pagelist_append(pagelist, &sr_rec, sizeof(sr_rec));
3001                 if (err)
3002                         goto fail;
3003         }
3004
3005         if (recon_state.flock)
3006                 reply->hdr.version = cpu_to_le16(2);
3007
3008         /* raced with cap release? */
3009         if (s_nr_caps != recon_state.nr_caps) {
3010                 struct page *page = list_first_entry(&pagelist->head,
3011                                                      struct page, lru);
3012                 __le32 *addr = kmap_atomic(page);
3013                 *addr = cpu_to_le32(recon_state.nr_caps);
3014                 kunmap_atomic(addr);
3015         }
3016
3017         reply->hdr.data_len = cpu_to_le32(pagelist->length);
3018         ceph_msg_data_add_pagelist(reply, pagelist);
3019
3020         ceph_early_kick_flushing_caps(mdsc, session);
3021
3022         ceph_con_send(&session->s_con, reply);
3023
3024         mutex_unlock(&session->s_mutex);
3025
3026         mutex_lock(&mdsc->mutex);
3027         __wake_requests(mdsc, &session->s_waiting);
3028         mutex_unlock(&mdsc->mutex);
3029
3030         up_read(&mdsc->snap_rwsem);
3031         return;
3032
3033 fail:
3034         ceph_msg_put(reply);
3035         up_read(&mdsc->snap_rwsem);
3036         mutex_unlock(&session->s_mutex);
3037 fail_nomsg:
3038         ceph_pagelist_release(pagelist);
3039 fail_nopagelist:
3040         pr_err("error %d preparing reconnect for mds%d\n", err, mds);
3041         return;
3042 }
3043
3044
3045 /*
3046  * compare old and new mdsmaps, kicking requests
3047  * and closing out old connections as necessary
3048  *
3049  * called under mdsc->mutex.
3050  */
3051 static void check_new_map(struct ceph_mds_client *mdsc,
3052                           struct ceph_mdsmap *newmap,
3053                           struct ceph_mdsmap *oldmap)
3054 {
3055         int i;
3056         int oldstate, newstate;
3057         struct ceph_mds_session *s;
3058
3059         dout("check_new_map new %u old %u\n",
3060              newmap->m_epoch, oldmap->m_epoch);
3061
3062         for (i = 0; i < oldmap->m_max_mds && i < mdsc->max_sessions; i++) {
3063                 if (mdsc->sessions[i] == NULL)
3064                         continue;
3065                 s = mdsc->sessions[i];
3066                 oldstate = ceph_mdsmap_get_state(oldmap, i);
3067                 newstate = ceph_mdsmap_get_state(newmap, i);
3068
3069                 dout("check_new_map mds%d state %s%s -> %s%s (session %s)\n",
3070                      i, ceph_mds_state_name(oldstate),
3071                      ceph_mdsmap_is_laggy(oldmap, i) ? " (laggy)" : "",
3072                      ceph_mds_state_name(newstate),
3073                      ceph_mdsmap_is_laggy(newmap, i) ? " (laggy)" : "",
3074                      ceph_session_state_name(s->s_state));
3075
3076                 if (i >= newmap->m_max_mds ||
3077                     memcmp(ceph_mdsmap_get_addr(oldmap, i),
3078                            ceph_mdsmap_get_addr(newmap, i),
3079                            sizeof(struct ceph_entity_addr))) {
3080                         if (s->s_state == CEPH_MDS_SESSION_OPENING) {
3081                                 /* the session never opened, just close it
3082                                  * out now */
3083                                 __wake_requests(mdsc, &s->s_waiting);
3084                                 __unregister_session(mdsc, s);
3085                         } else {
3086                                 /* just close it */
3087                                 mutex_unlock(&mdsc->mutex);
3088                                 mutex_lock(&s->s_mutex);
3089                                 mutex_lock(&mdsc->mutex);
3090                                 ceph_con_close(&s->s_con);
3091                                 mutex_unlock(&s->s_mutex);
3092                                 s->s_state = CEPH_MDS_SESSION_RESTARTING;
3093                         }
3094                 } else if (oldstate == newstate) {
3095                         continue;  /* nothing new with this mds */
3096                 }
3097
3098                 /*
3099                  * send reconnect?
3100                  */
3101                 if (s->s_state == CEPH_MDS_SESSION_RESTARTING &&
3102                     newstate >= CEPH_MDS_STATE_RECONNECT) {
3103                         mutex_unlock(&mdsc->mutex);
3104                         send_mds_reconnect(mdsc, s);
3105                         mutex_lock(&mdsc->mutex);
3106                 }
3107
3108                 /*
3109                  * kick request on any mds that has gone active.
3110                  */
3111                 if (oldstate < CEPH_MDS_STATE_ACTIVE &&
3112                     newstate >= CEPH_MDS_STATE_ACTIVE) {
3113                         if (oldstate != CEPH_MDS_STATE_CREATING &&
3114                             oldstate != CEPH_MDS_STATE_STARTING)
3115                                 pr_info("mds%d recovery completed\n", s->s_mds);
3116                         kick_requests(mdsc, i);
3117                         ceph_kick_flushing_caps(mdsc, s);
3118                         wake_up_session_caps(s, 1);
3119                 }
3120         }
3121
3122         for (i = 0; i < newmap->m_max_mds && i < mdsc->max_sessions; i++) {
3123                 s = mdsc->sessions[i];
3124                 if (!s)
3125                         continue;
3126                 if (!ceph_mdsmap_is_laggy(newmap, i))
3127                         continue;
3128                 if (s->s_state == CEPH_MDS_SESSION_OPEN ||
3129                     s->s_state == CEPH_MDS_SESSION_HUNG ||
3130                     s->s_state == CEPH_MDS_SESSION_CLOSING) {
3131                         dout(" connecting to export targets of laggy mds%d\n",
3132                              i);
3133                         __open_export_target_sessions(mdsc, s);
3134                 }
3135         }
3136 }
3137
3138
3139
3140 /*
3141  * leases
3142  */
3143
3144 /*
3145  * caller must hold session s_mutex, dentry->d_lock
3146  */
3147 void __ceph_mdsc_drop_dentry_lease(struct dentry *dentry)
3148 {
3149         struct ceph_dentry_info *di = ceph_dentry(dentry);
3150
3151         ceph_put_mds_session(di->lease_session);
3152         di->lease_session = NULL;
3153 }
3154
3155 static void handle_lease(struct ceph_mds_client *mdsc,
3156                          struct ceph_mds_session *session,
3157                          struct ceph_msg *msg)
3158 {
3159         struct super_block *sb = mdsc->fsc->sb;
3160         struct inode *inode;
3161         struct dentry *parent, *dentry;
3162         struct ceph_dentry_info *di;
3163         int mds = session->s_mds;
3164         struct ceph_mds_lease *h = msg->front.iov_base;
3165         u32 seq;
3166         struct ceph_vino vino;
3167         struct qstr dname;
3168         int release = 0;
3169
3170         dout("handle_lease from mds%d\n", mds);
3171
3172         /* decode */
3173         if (msg->front.iov_len < sizeof(*h) + sizeof(u32))
3174                 goto bad;
3175         vino.ino = le64_to_cpu(h->ino);
3176         vino.snap = CEPH_NOSNAP;
3177         seq = le32_to_cpu(h->seq);
3178         dname.name = (void *)h + sizeof(*h) + sizeof(u32);
3179         dname.len = msg->front.iov_len - sizeof(*h) - sizeof(u32);
3180         if (dname.len != get_unaligned_le32(h+1))
3181                 goto bad;
3182
3183         /* lookup inode */
3184         inode = ceph_find_inode(sb, vino);
3185         dout("handle_lease %s, ino %llx %p %.*s\n",
3186              ceph_lease_op_name(h->action), vino.ino, inode,
3187              dname.len, dname.name);
3188
3189         mutex_lock(&session->s_mutex);
3190         session->s_seq++;
3191
3192         if (inode == NULL) {
3193                 dout("handle_lease no inode %llx\n", vino.ino);
3194                 goto release;
3195         }
3196
3197         /* dentry */
3198         parent = d_find_alias(inode);
3199         if (!parent) {
3200                 dout("no parent dentry on inode %p\n", inode);
3201                 WARN_ON(1);
3202                 goto release;  /* hrm... */
3203         }
3204         dname.hash = full_name_hash(dname.name, dname.len);
3205         dentry = d_lookup(parent, &dname);
3206         dput(parent);
3207         if (!dentry)
3208                 goto release;
3209
3210         spin_lock(&dentry->d_lock);
3211         di = ceph_dentry(dentry);
3212         switch (h->action) {
3213         case CEPH_MDS_LEASE_REVOKE:
3214                 if (di->lease_session == session) {
3215                         if (ceph_seq_cmp(di->lease_seq, seq) > 0)
3216                                 h->seq = cpu_to_le32(di->lease_seq);
3217                         __ceph_mdsc_drop_dentry_lease(dentry);
3218                 }
3219                 release = 1;
3220                 break;
3221
3222         case CEPH_MDS_LEASE_RENEW:
3223                 if (di->lease_session == session &&
3224                     di->lease_gen == session->s_cap_gen &&
3225                     di->lease_renew_from &&
3226                     di->lease_renew_after == 0) {
3227                         unsigned long duration =
3228                                 msecs_to_jiffies(le32_to_cpu(h->duration_ms));
3229
3230                         di->lease_seq = seq;
3231                         dentry->d_time = di->lease_renew_from + duration;
3232                         di->lease_renew_after = di->lease_renew_from +
3233                                 (duration >> 1);
3234                         di->lease_renew_from = 0;
3235                 }
3236                 break;
3237         }
3238         spin_unlock(&dentry->d_lock);
3239         dput(dentry);
3240
3241         if (!release)
3242                 goto out;
3243
3244 release:
3245         /* let's just reuse the same message */
3246         h->action = CEPH_MDS_LEASE_REVOKE_ACK;
3247         ceph_msg_get(msg);
3248         ceph_con_send(&session->s_con, msg);
3249
3250 out:
3251         iput(inode);
3252         mutex_unlock(&session->s_mutex);
3253         return;
3254
3255 bad:
3256         pr_err("corrupt lease message\n");
3257         ceph_msg_dump(msg);
3258 }
3259
3260 void ceph_mdsc_lease_send_msg(struct ceph_mds_session *session,
3261                               struct inode *inode,
3262                               struct dentry *dentry, char action,
3263                               u32 seq)
3264 {
3265         struct ceph_msg *msg;
3266         struct ceph_mds_lease *lease;
3267         int len = sizeof(*lease) + sizeof(u32);
3268         int dnamelen = 0;
3269
3270         dout("lease_send_msg inode %p dentry %p %s to mds%d\n",
3271              inode, dentry, ceph_lease_op_name(action), session->s_mds);
3272         dnamelen = dentry->d_name.len;
3273         len += dnamelen;
3274
3275         msg = ceph_msg_new(CEPH_MSG_CLIENT_LEASE, len, GFP_NOFS, false);
3276         if (!msg)
3277                 return;
3278         lease = msg->front.iov_base;
3279         lease->action = action;
3280         lease->ino = cpu_to_le64(ceph_vino(inode).ino);
3281         lease->first = lease->last = cpu_to_le64(ceph_vino(inode).snap);
3282         lease->seq = cpu_to_le32(seq);
3283         put_unaligned_le32(dnamelen, lease + 1);
3284         memcpy((void *)(lease + 1) + 4, dentry->d_name.name, dnamelen);
3285
3286         /*
3287          * if this is a preemptive lease RELEASE, no need to
3288          * flush request stream, since the actual request will
3289          * soon follow.
3290          */
3291         msg->more_to_follow = (action == CEPH_MDS_LEASE_RELEASE);
3292
3293         ceph_con_send(&session->s_con, msg);
3294 }
3295
3296 /*
3297  * Preemptively release a lease we expect to invalidate anyway.
3298  * Pass @inode always, @dentry is optional.
3299  */
3300 void ceph_mdsc_lease_release(struct ceph_mds_client *mdsc, struct inode *inode,
3301                              struct dentry *dentry)
3302 {
3303         struct ceph_dentry_info *di;
3304         struct ceph_mds_session *session;
3305         u32 seq;
3306
3307         BUG_ON(inode == NULL);
3308         BUG_ON(dentry == NULL);
3309
3310         /* is dentry lease valid? */
3311         spin_lock(&dentry->d_lock);
3312         di = ceph_dentry(dentry);
3313         if (!di || !di->lease_session ||
3314             di->lease_session->s_mds < 0 ||
3315             di->lease_gen != di->lease_session->s_cap_gen ||
3316             !time_before(jiffies, dentry->d_time)) {
3317                 dout("lease_release inode %p dentry %p -- "
3318                      "no lease\n",
3319                      inode, dentry);
3320                 spin_unlock(&dentry->d_lock);
3321                 return;
3322         }
3323
3324         /* we do have a lease on this dentry; note mds and seq */
3325         session = ceph_get_mds_session(di->lease_session);
3326         seq = di->lease_seq;
3327         __ceph_mdsc_drop_dentry_lease(dentry);
3328         spin_unlock(&dentry->d_lock);
3329
3330         dout("lease_release inode %p dentry %p to mds%d\n",
3331              inode, dentry, session->s_mds);
3332         ceph_mdsc_lease_send_msg(session, inode, dentry,
3333                                  CEPH_MDS_LEASE_RELEASE, seq);
3334         ceph_put_mds_session(session);
3335 }
3336
3337 /*
3338  * drop all leases (and dentry refs) in preparation for umount
3339  */
3340 static void drop_leases(struct ceph_mds_client *mdsc)
3341 {
3342         int i;
3343
3344         dout("drop_leases\n");
3345         mutex_lock(&mdsc->mutex);
3346         for (i = 0; i < mdsc->max_sessions; i++) {
3347                 struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
3348                 if (!s)
3349                         continue;
3350                 mutex_unlock(&mdsc->mutex);
3351                 mutex_lock(&s->s_mutex);
3352                 mutex_unlock(&s->s_mutex);
3353                 ceph_put_mds_session(s);
3354                 mutex_lock(&mdsc->mutex);
3355         }
3356         mutex_unlock(&mdsc->mutex);
3357 }
3358
3359
3360
3361 /*
3362  * delayed work -- periodically trim expired leases, renew caps with mds
3363  */
3364 static void schedule_delayed(struct ceph_mds_client *mdsc)
3365 {
3366         int delay = 5;
3367         unsigned hz = round_jiffies_relative(HZ * delay);
3368         schedule_delayed_work(&mdsc->delayed_work, hz);
3369 }
3370
3371 static void delayed_work(struct work_struct *work)
3372 {
3373         int i;
3374         struct ceph_mds_client *mdsc =
3375                 container_of(work, struct ceph_mds_client, delayed_work.work);
3376         int renew_interval;
3377         int renew_caps;
3378
3379         dout("mdsc delayed_work\n");
3380         ceph_check_delayed_caps(mdsc);
3381
3382         mutex_lock(&mdsc->mutex);
3383         renew_interval = mdsc->mdsmap->m_session_timeout >> 2;
3384         renew_caps = time_after_eq(jiffies, HZ*renew_interval +
3385                                    mdsc->last_renew_caps);
3386         if (renew_caps)
3387                 mdsc->last_renew_caps = jiffies;
3388
3389         for (i = 0; i < mdsc->max_sessions; i++) {
3390                 struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
3391                 if (s == NULL)
3392                         continue;
3393                 if (s->s_state == CEPH_MDS_SESSION_CLOSING) {
3394                         dout("resending session close request for mds%d\n",
3395                              s->s_mds);
3396                         request_close_session(mdsc, s);
3397                         ceph_put_mds_session(s);
3398                         continue;
3399                 }
3400                 if (s->s_ttl && time_after(jiffies, s->s_ttl)) {
3401                         if (s->s_state == CEPH_MDS_SESSION_OPEN) {
3402                                 s->s_state = CEPH_MDS_SESSION_HUNG;
3403                                 pr_info("mds%d hung\n", s->s_mds);
3404                         }
3405                 }
3406                 if (s->s_state < CEPH_MDS_SESSION_OPEN) {
3407                         /* this mds is failed or recovering, just wait */
3408                         ceph_put_mds_session(s);
3409                         continue;
3410                 }
3411                 mutex_unlock(&mdsc->mutex);
3412
3413                 mutex_lock(&s->s_mutex);
3414                 if (renew_caps)
3415                         send_renew_caps(mdsc, s);
3416                 else
3417                         ceph_con_keepalive(&s->s_con);
3418                 if (s->s_state == CEPH_MDS_SESSION_OPEN ||
3419                     s->s_state == CEPH_MDS_SESSION_HUNG)
3420                         ceph_send_cap_releases(mdsc, s);
3421                 mutex_unlock(&s->s_mutex);
3422                 ceph_put_mds_session(s);
3423
3424                 mutex_lock(&mdsc->mutex);
3425         }
3426         mutex_unlock(&mdsc->mutex);
3427
3428         schedule_delayed(mdsc);
3429 }
3430
3431 int ceph_mdsc_init(struct ceph_fs_client *fsc)
3432
3433 {
3434         struct ceph_mds_client *mdsc;
3435
3436         mdsc = kzalloc(sizeof(struct ceph_mds_client), GFP_NOFS);
3437         if (!mdsc)
3438                 return -ENOMEM;
3439         mdsc->fsc = fsc;
3440         fsc->mdsc = mdsc;
3441         mutex_init(&mdsc->mutex);
3442         mdsc->mdsmap = kzalloc(sizeof(*mdsc->mdsmap), GFP_NOFS);
3443         if (mdsc->mdsmap == NULL) {
3444                 kfree(mdsc);
3445                 return -ENOMEM;
3446         }
3447
3448         init_completion(&mdsc->safe_umount_waiters);
3449         init_waitqueue_head(&mdsc->session_close_wq);
3450         INIT_LIST_HEAD(&mdsc->waiting_for_map);
3451         mdsc->sessions = NULL;
3452         atomic_set(&mdsc->num_sessions, 0);
3453         mdsc->max_sessions = 0;
3454         mdsc->stopping = 0;
3455         mdsc->last_snap_seq = 0;
3456         init_rwsem(&mdsc->snap_rwsem);
3457         mdsc->snap_realms = RB_ROOT;
3458         INIT_LIST_HEAD(&mdsc->snap_empty);
3459         spin_lock_init(&mdsc->snap_empty_lock);
3460         mdsc->last_tid = 0;
3461         mdsc->oldest_tid = 0;
3462         mdsc->request_tree = RB_ROOT;
3463         INIT_DELAYED_WORK(&mdsc->delayed_work, delayed_work);
3464         mdsc->last_renew_caps = jiffies;
3465         INIT_LIST_HEAD(&mdsc->cap_delay_list);
3466         spin_lock_init(&mdsc->cap_delay_lock);
3467         INIT_LIST_HEAD(&mdsc->snap_flush_list);
3468         spin_lock_init(&mdsc->snap_flush_lock);
3469         mdsc->last_cap_flush_tid = 1;
3470         mdsc->cap_flush_tree = RB_ROOT;
3471         INIT_LIST_HEAD(&mdsc->cap_dirty);
3472         INIT_LIST_HEAD(&mdsc->cap_dirty_migrating);
3473         mdsc->num_cap_flushing = 0;
3474         spin_lock_init(&mdsc->cap_dirty_lock);
3475         init_waitqueue_head(&mdsc->cap_flushing_wq);
3476         spin_lock_init(&mdsc->dentry_lru_lock);
3477         INIT_LIST_HEAD(&mdsc->dentry_lru);
3478
3479         ceph_caps_init(mdsc);
3480         ceph_adjust_min_caps(mdsc, fsc->min_caps);
3481
3482         init_rwsem(&mdsc->pool_perm_rwsem);
3483         mdsc->pool_perm_tree = RB_ROOT;
3484
3485         return 0;
3486 }
3487
3488 /*
3489  * Wait for safe replies on open mds requests.  If we time out, drop
3490  * all requests from the tree to avoid dangling dentry refs.
3491  */
3492 static void wait_requests(struct ceph_mds_client *mdsc)
3493 {
3494         struct ceph_options *opts = mdsc->fsc->client->options;
3495         struct ceph_mds_request *req;
3496
3497         mutex_lock(&mdsc->mutex);
3498         if (__get_oldest_req(mdsc)) {
3499                 mutex_unlock(&mdsc->mutex);
3500
3501                 dout("wait_requests waiting for requests\n");
3502                 wait_for_completion_timeout(&mdsc->safe_umount_waiters,
3503                                     ceph_timeout_jiffies(opts->mount_timeout));
3504
3505                 /* tear down remaining requests */
3506                 mutex_lock(&mdsc->mutex);
3507                 while ((req = __get_oldest_req(mdsc))) {
3508                         dout("wait_requests timed out on tid %llu\n",
3509                              req->r_tid);
3510                         __unregister_request(mdsc, req);
3511                 }
3512         }
3513         mutex_unlock(&mdsc->mutex);
3514         dout("wait_requests done\n");
3515 }
3516
3517 /*
3518  * called before mount is ro, and before dentries are torn down.
3519  * (hmm, does this still race with new lookups?)
3520  */
3521 void ceph_mdsc_pre_umount(struct ceph_mds_client *mdsc)
3522 {
3523         dout("pre_umount\n");
3524         mdsc->stopping = 1;
3525
3526         drop_leases(mdsc);
3527         ceph_flush_dirty_caps(mdsc);
3528         wait_requests(mdsc);
3529
3530         /*
3531          * wait for reply handlers to drop their request refs and
3532          * their inode/dcache refs
3533          */
3534         ceph_msgr_flush();
3535 }
3536
3537 /*
3538  * wait for all write mds requests to flush.
3539  */
3540 static void wait_unsafe_requests(struct ceph_mds_client *mdsc, u64 want_tid)
3541 {
3542         struct ceph_mds_request *req = NULL, *nextreq;
3543         struct rb_node *n;
3544
3545         mutex_lock(&mdsc->mutex);
3546         dout("wait_unsafe_requests want %lld\n", want_tid);
3547 restart:
3548         req = __get_oldest_req(mdsc);
3549         while (req && req->r_tid <= want_tid) {
3550                 /* find next request */
3551                 n = rb_next(&req->r_node);
3552                 if (n)
3553                         nextreq = rb_entry(n, struct ceph_mds_request, r_node);
3554                 else
3555                         nextreq = NULL;
3556                 if (req->r_op != CEPH_MDS_OP_SETFILELOCK &&
3557                     (req->r_op & CEPH_MDS_OP_WRITE)) {
3558                         /* write op */
3559                         ceph_mdsc_get_request(req);
3560                         if (nextreq)
3561                                 ceph_mdsc_get_request(nextreq);
3562                         mutex_unlock(&mdsc->mutex);
3563                         dout("wait_unsafe_requests  wait on %llu (want %llu)\n",
3564                              req->r_tid, want_tid);
3565                         wait_for_completion(&req->r_safe_completion);
3566                         mutex_lock(&mdsc->mutex);
3567                         ceph_mdsc_put_request(req);
3568                         if (!nextreq)
3569                                 break;  /* next dne before, so we're done! */
3570                         if (RB_EMPTY_NODE(&nextreq->r_node)) {
3571                                 /* next request was removed from tree */
3572                                 ceph_mdsc_put_request(nextreq);
3573                                 goto restart;
3574                         }
3575                         ceph_mdsc_put_request(nextreq);  /* won't go away */
3576                 }
3577                 req = nextreq;
3578         }
3579         mutex_unlock(&mdsc->mutex);
3580         dout("wait_unsafe_requests done\n");
3581 }
3582
3583 void ceph_mdsc_sync(struct ceph_mds_client *mdsc)
3584 {
3585         u64 want_tid, want_flush, want_snap;
3586
3587         if (ACCESS_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN)
3588                 return;
3589
3590         dout("sync\n");
3591         mutex_lock(&mdsc->mutex);
3592         want_tid = mdsc->last_tid;
3593         mutex_unlock(&mdsc->mutex);
3594
3595         ceph_flush_dirty_caps(mdsc);
3596         spin_lock(&mdsc->cap_dirty_lock);
3597         want_flush = mdsc->last_cap_flush_tid;
3598         spin_unlock(&mdsc->cap_dirty_lock);
3599
3600         down_read(&mdsc->snap_rwsem);
3601         want_snap = mdsc->last_snap_seq;
3602         up_read(&mdsc->snap_rwsem);
3603
3604         dout("sync want tid %lld flush_seq %lld snap_seq %lld\n",
3605              want_tid, want_flush, want_snap);
3606
3607         wait_unsafe_requests(mdsc, want_tid);
3608         wait_caps_flush(mdsc, want_flush, want_snap);
3609 }
3610
3611 /*
3612  * true if all sessions are closed, or we force unmount
3613  */
3614 static bool done_closing_sessions(struct ceph_mds_client *mdsc)
3615 {
3616         if (ACCESS_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN)
3617                 return true;
3618         return atomic_read(&mdsc->num_sessions) == 0;
3619 }
3620
3621 /*
3622  * called after sb is ro.
3623  */
3624 void ceph_mdsc_close_sessions(struct ceph_mds_client *mdsc)
3625 {
3626         struct ceph_options *opts = mdsc->fsc->client->options;
3627         struct ceph_mds_session *session;
3628         int i;
3629
3630         dout("close_sessions\n");
3631
3632         /* close sessions */
3633         mutex_lock(&mdsc->mutex);
3634         for (i = 0; i < mdsc->max_sessions; i++) {
3635                 session = __ceph_lookup_mds_session(mdsc, i);
3636                 if (!session)
3637                         continue;
3638                 mutex_unlock(&mdsc->mutex);
3639                 mutex_lock(&session->s_mutex);
3640                 __close_session(mdsc, session);
3641                 mutex_unlock(&session->s_mutex);
3642                 ceph_put_mds_session(session);
3643                 mutex_lock(&mdsc->mutex);
3644         }
3645         mutex_unlock(&mdsc->mutex);
3646
3647         dout("waiting for sessions to close\n");
3648         wait_event_timeout(mdsc->session_close_wq, done_closing_sessions(mdsc),
3649                            ceph_timeout_jiffies(opts->mount_timeout));
3650
3651         /* tear down remaining sessions */
3652         mutex_lock(&mdsc->mutex);
3653         for (i = 0; i < mdsc->max_sessions; i++) {
3654                 if (mdsc->sessions[i]) {
3655                         session = get_session(mdsc->sessions[i]);
3656                         __unregister_session(mdsc, session);
3657                         mutex_unlock(&mdsc->mutex);
3658                         mutex_lock(&session->s_mutex);
3659                         remove_session_caps(session);
3660                         mutex_unlock(&session->s_mutex);
3661                         ceph_put_mds_session(session);
3662                         mutex_lock(&mdsc->mutex);
3663                 }
3664         }
3665         WARN_ON(!list_empty(&mdsc->cap_delay_list));
3666         mutex_unlock(&mdsc->mutex);
3667
3668         ceph_cleanup_empty_realms(mdsc);
3669
3670         cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */
3671
3672         dout("stopped\n");
3673 }
3674
3675 void ceph_mdsc_force_umount(struct ceph_mds_client *mdsc)
3676 {
3677         struct ceph_mds_session *session;
3678         int mds;
3679
3680         dout("force umount\n");
3681
3682         mutex_lock(&mdsc->mutex);
3683         for (mds = 0; mds < mdsc->max_sessions; mds++) {
3684                 session = __ceph_lookup_mds_session(mdsc, mds);
3685                 if (!session)
3686                         continue;
3687                 mutex_unlock(&mdsc->mutex);
3688                 mutex_lock(&session->s_mutex);
3689                 __close_session(mdsc, session);
3690                 if (session->s_state == CEPH_MDS_SESSION_CLOSING) {
3691                         cleanup_session_requests(mdsc, session);
3692                         remove_session_caps(session);
3693                 }
3694                 mutex_unlock(&session->s_mutex);
3695                 ceph_put_mds_session(session);
3696                 mutex_lock(&mdsc->mutex);
3697                 kick_requests(mdsc, mds);
3698         }
3699         __wake_requests(mdsc, &mdsc->waiting_for_map);
3700         mutex_unlock(&mdsc->mutex);
3701 }
3702
3703 static void ceph_mdsc_stop(struct ceph_mds_client *mdsc)
3704 {
3705         dout("stop\n");
3706         cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */
3707         if (mdsc->mdsmap)
3708                 ceph_mdsmap_destroy(mdsc->mdsmap);
3709         kfree(mdsc->sessions);
3710         ceph_caps_finalize(mdsc);
3711         ceph_pool_perm_destroy(mdsc);
3712 }
3713
3714 void ceph_mdsc_destroy(struct ceph_fs_client *fsc)
3715 {
3716         struct ceph_mds_client *mdsc = fsc->mdsc;
3717
3718         dout("mdsc_destroy %p\n", mdsc);
3719         ceph_mdsc_stop(mdsc);
3720
3721         /* flush out any connection work with references to us */
3722         ceph_msgr_flush();
3723
3724         fsc->mdsc = NULL;
3725         kfree(mdsc);
3726         dout("mdsc_destroy %p done\n", mdsc);
3727 }
3728
3729
3730 /*
3731  * handle mds map update.
3732  */
3733 void ceph_mdsc_handle_map(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
3734 {
3735         u32 epoch;
3736         u32 maplen;
3737         void *p = msg->front.iov_base;
3738         void *end = p + msg->front.iov_len;
3739         struct ceph_mdsmap *newmap, *oldmap;
3740         struct ceph_fsid fsid;
3741         int err = -EINVAL;
3742
3743         ceph_decode_need(&p, end, sizeof(fsid)+2*sizeof(u32), bad);
3744         ceph_decode_copy(&p, &fsid, sizeof(fsid));
3745         if (ceph_check_fsid(mdsc->fsc->client, &fsid) < 0)
3746                 return;
3747         epoch = ceph_decode_32(&p);
3748         maplen = ceph_decode_32(&p);
3749         dout("handle_map epoch %u len %d\n", epoch, (int)maplen);
3750
3751         /* do we need it? */
3752         ceph_monc_got_mdsmap(&mdsc->fsc->client->monc, epoch);
3753         mutex_lock(&mdsc->mutex);
3754         if (mdsc->mdsmap && epoch <= mdsc->mdsmap->m_epoch) {
3755                 dout("handle_map epoch %u <= our %u\n",
3756                      epoch, mdsc->mdsmap->m_epoch);
3757                 mutex_unlock(&mdsc->mutex);
3758                 return;
3759         }
3760
3761         newmap = ceph_mdsmap_decode(&p, end);
3762         if (IS_ERR(newmap)) {
3763                 err = PTR_ERR(newmap);
3764                 goto bad_unlock;
3765         }
3766
3767         /* swap into place */
3768         if (mdsc->mdsmap) {
3769                 oldmap = mdsc->mdsmap;
3770                 mdsc->mdsmap = newmap;
3771                 check_new_map(mdsc, newmap, oldmap);
3772                 ceph_mdsmap_destroy(oldmap);
3773         } else {
3774                 mdsc->mdsmap = newmap;  /* first mds map */
3775         }
3776         mdsc->fsc->sb->s_maxbytes = mdsc->mdsmap->m_max_file_size;
3777
3778         __wake_requests(mdsc, &mdsc->waiting_for_map);
3779
3780         mutex_unlock(&mdsc->mutex);
3781         schedule_delayed(mdsc);
3782         return;
3783
3784 bad_unlock:
3785         mutex_unlock(&mdsc->mutex);
3786 bad:
3787         pr_err("error decoding mdsmap %d\n", err);
3788         return;
3789 }
3790
3791 static struct ceph_connection *con_get(struct ceph_connection *con)
3792 {
3793         struct ceph_mds_session *s = con->private;
3794
3795         if (get_session(s)) {
3796                 dout("mdsc con_get %p ok (%d)\n", s, atomic_read(&s->s_ref));
3797                 return con;
3798         }
3799         dout("mdsc con_get %p FAIL\n", s);
3800         return NULL;
3801 }
3802
3803 static void con_put(struct ceph_connection *con)
3804 {
3805         struct ceph_mds_session *s = con->private;
3806
3807         dout("mdsc con_put %p (%d)\n", s, atomic_read(&s->s_ref) - 1);
3808         ceph_put_mds_session(s);
3809 }
3810
3811 /*
3812  * if the client is unresponsive for long enough, the mds will kill
3813  * the session entirely.
3814  */
3815 static void peer_reset(struct ceph_connection *con)
3816 {
3817         struct ceph_mds_session *s = con->private;
3818         struct ceph_mds_client *mdsc = s->s_mdsc;
3819
3820         pr_warn("mds%d closed our session\n", s->s_mds);
3821         send_mds_reconnect(mdsc, s);
3822 }
3823
3824 static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
3825 {
3826         struct ceph_mds_session *s = con->private;
3827         struct ceph_mds_client *mdsc = s->s_mdsc;
3828         int type = le16_to_cpu(msg->hdr.type);
3829
3830         mutex_lock(&mdsc->mutex);
3831         if (__verify_registered_session(mdsc, s) < 0) {
3832                 mutex_unlock(&mdsc->mutex);
3833                 goto out;
3834         }
3835         mutex_unlock(&mdsc->mutex);
3836
3837         switch (type) {
3838         case CEPH_MSG_MDS_MAP:
3839                 ceph_mdsc_handle_map(mdsc, msg);
3840                 break;
3841         case CEPH_MSG_CLIENT_SESSION:
3842                 handle_session(s, msg);
3843                 break;
3844         case CEPH_MSG_CLIENT_REPLY:
3845                 handle_reply(s, msg);
3846                 break;
3847         case CEPH_MSG_CLIENT_REQUEST_FORWARD:
3848                 handle_forward(mdsc, s, msg);
3849                 break;
3850         case CEPH_MSG_CLIENT_CAPS:
3851                 ceph_handle_caps(s, msg);
3852                 break;
3853         case CEPH_MSG_CLIENT_SNAP:
3854                 ceph_handle_snap(mdsc, s, msg);
3855                 break;
3856         case CEPH_MSG_CLIENT_LEASE:
3857                 handle_lease(mdsc, s, msg);
3858                 break;
3859
3860         default:
3861                 pr_err("received unknown message type %d %s\n", type,
3862                        ceph_msg_type_name(type));
3863         }
3864 out:
3865         ceph_msg_put(msg);
3866 }
3867
3868 /*
3869  * authentication
3870  */
3871
3872 /*
3873  * Note: returned pointer is the address of a structure that's
3874  * managed separately.  Caller must *not* attempt to free it.
3875  */
3876 static struct ceph_auth_handshake *get_authorizer(struct ceph_connection *con,
3877                                         int *proto, int force_new)
3878 {
3879         struct ceph_mds_session *s = con->private;
3880         struct ceph_mds_client *mdsc = s->s_mdsc;
3881         struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
3882         struct ceph_auth_handshake *auth = &s->s_auth;
3883
3884         if (force_new && auth->authorizer) {
3885                 ceph_auth_destroy_authorizer(ac, auth->authorizer);
3886                 auth->authorizer = NULL;
3887         }
3888         if (!auth->authorizer) {
3889                 int ret = ceph_auth_create_authorizer(ac, CEPH_ENTITY_TYPE_MDS,
3890                                                       auth);
3891                 if (ret)
3892                         return ERR_PTR(ret);
3893         } else {
3894                 int ret = ceph_auth_update_authorizer(ac, CEPH_ENTITY_TYPE_MDS,
3895                                                       auth);
3896                 if (ret)
3897                         return ERR_PTR(ret);
3898         }
3899         *proto = ac->protocol;
3900
3901         return auth;
3902 }
3903
3904
3905 static int verify_authorizer_reply(struct ceph_connection *con, int len)
3906 {
3907         struct ceph_mds_session *s = con->private;
3908         struct ceph_mds_client *mdsc = s->s_mdsc;
3909         struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
3910
3911         return ceph_auth_verify_authorizer_reply(ac, s->s_auth.authorizer, len);
3912 }
3913
3914 static int invalidate_authorizer(struct ceph_connection *con)
3915 {
3916         struct ceph_mds_session *s = con->private;
3917         struct ceph_mds_client *mdsc = s->s_mdsc;
3918         struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
3919
3920         ceph_auth_invalidate_authorizer(ac, CEPH_ENTITY_TYPE_MDS);
3921
3922         return ceph_monc_validate_auth(&mdsc->fsc->client->monc);
3923 }
3924
3925 static struct ceph_msg *mds_alloc_msg(struct ceph_connection *con,
3926                                 struct ceph_msg_header *hdr, int *skip)
3927 {
3928         struct ceph_msg *msg;
3929         int type = (int) le16_to_cpu(hdr->type);
3930         int front_len = (int) le32_to_cpu(hdr->front_len);
3931
3932         if (con->in_msg)
3933                 return con->in_msg;
3934
3935         *skip = 0;
3936         msg = ceph_msg_new(type, front_len, GFP_NOFS, false);
3937         if (!msg) {
3938                 pr_err("unable to allocate msg type %d len %d\n",
3939                        type, front_len);
3940                 return NULL;
3941         }
3942
3943         return msg;
3944 }
3945
3946 static int mds_sign_message(struct ceph_msg *msg)
3947 {
3948        struct ceph_mds_session *s = msg->con->private;
3949        struct ceph_auth_handshake *auth = &s->s_auth;
3950
3951        return ceph_auth_sign_message(auth, msg);
3952 }
3953
3954 static int mds_check_message_signature(struct ceph_msg *msg)
3955 {
3956        struct ceph_mds_session *s = msg->con->private;
3957        struct ceph_auth_handshake *auth = &s->s_auth;
3958
3959        return ceph_auth_check_message_signature(auth, msg);
3960 }
3961
3962 static const struct ceph_connection_operations mds_con_ops = {
3963         .get = con_get,
3964         .put = con_put,
3965         .dispatch = dispatch,
3966         .get_authorizer = get_authorizer,
3967         .verify_authorizer_reply = verify_authorizer_reply,
3968         .invalidate_authorizer = invalidate_authorizer,
3969         .peer_reset = peer_reset,
3970         .alloc_msg = mds_alloc_msg,
3971         .sign_message = mds_sign_message,
3972         .check_message_signature = mds_check_message_signature,
3973 };
3974
3975 /* eof */