fs/ceph/mds_client.c

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