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         if (req->r_inode) {
1632                 struct ceph_cap *cap =
1633                         ceph_get_cap_for_mds(ceph_inode(req->r_inode), mds);
1634
1635                 if (cap)
1636                         req->r_sent_on_mseq = cap->mseq;
1637                 else
1638                         req->r_sent_on_mseq = -1;
1639         }
1640         dout("prepare_send_request %p tid %lld %s (attempt %d)\n", req,
1641              req->r_tid, ceph_mds_op_name(req->r_op), req->r_attempts);
1642
1643         if (req->r_got_unsafe) {
1644                 /*
1645                  * Replay.  Do not regenerate message (and rebuild
1646                  * paths, etc.); just use the original message.
1647                  * Rebuilding paths will break for renames because
1648                  * d_move mangles the src name.
1649                  */
1650                 msg = req->r_request;
1651                 rhead = msg->front.iov_base;
1652
1653                 flags = le32_to_cpu(rhead->flags);
1654                 flags |= CEPH_MDS_FLAG_REPLAY;
1655                 rhead->flags = cpu_to_le32(flags);
1656
1657                 if (req->r_target_inode)
1658                         rhead->ino = cpu_to_le64(ceph_ino(req->r_target_inode));
1659
1660                 rhead->num_retry = req->r_attempts - 1;
1661
1662                 /* remove cap/dentry releases from message */
1663                 rhead->num_releases = 0;
1664                 msg->hdr.front_len = cpu_to_le32(req->r_request_release_offset);
1665                 msg->front.iov_len = req->r_request_release_offset;
1666                 return 0;
1667         }
1668
1669         if (req->r_request) {
1670                 ceph_msg_put(req->r_request);
1671                 req->r_request = NULL;
1672         }
1673         msg = create_request_message(mdsc, req, mds);
1674         if (IS_ERR(msg)) {
1675                 req->r_err = PTR_ERR(msg);
1676                 complete_request(mdsc, req);
1677                 return PTR_ERR(msg);
1678         }
1679         req->r_request = msg;
1680
1681         rhead = msg->front.iov_base;
1682         rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
1683         if (req->r_got_unsafe)
1684                 flags |= CEPH_MDS_FLAG_REPLAY;
1685         if (req->r_locked_dir)
1686                 flags |= CEPH_MDS_FLAG_WANT_DENTRY;
1687         rhead->flags = cpu_to_le32(flags);
1688         rhead->num_fwd = req->r_num_fwd;
1689         rhead->num_retry = req->r_attempts - 1;
1690         rhead->ino = 0;
1691
1692         dout(" r_locked_dir = %p\n", req->r_locked_dir);
1693         return 0;
1694 }
1695
1696 /*
1697  * send request, or put it on the appropriate wait list.
1698  */
1699 static int __do_request(struct ceph_mds_client *mdsc,
1700                         struct ceph_mds_request *req)
1701 {
1702         struct ceph_mds_session *session = NULL;
1703         int mds = -1;
1704         int err = -EAGAIN;
1705
1706         if (req->r_err || req->r_got_result)
1707                 goto out;
1708
1709         if (req->r_timeout &&
1710             time_after_eq(jiffies, req->r_started + req->r_timeout)) {
1711                 dout("do_request timed out\n");
1712                 err = -EIO;
1713                 goto finish;
1714         }
1715
1716         mds = __choose_mds(mdsc, req);
1717         if (mds < 0 ||
1718             ceph_mdsmap_get_state(mdsc->mdsmap, mds) < CEPH_MDS_STATE_ACTIVE) {
1719                 dout("do_request no mds or not active, waiting for map\n");
1720                 list_add(&req->r_wait, &mdsc->waiting_for_map);
1721                 goto out;
1722         }
1723
1724         /* get, open session */
1725         session = __ceph_lookup_mds_session(mdsc, mds);
1726         if (!session) {
1727                 session = register_session(mdsc, mds);
1728                 if (IS_ERR(session)) {
1729                         err = PTR_ERR(session);
1730                         goto finish;
1731                 }
1732         }
1733         dout("do_request mds%d session %p state %s\n", mds, session,
1734              session_state_name(session->s_state));
1735         if (session->s_state != CEPH_MDS_SESSION_OPEN &&
1736             session->s_state != CEPH_MDS_SESSION_HUNG) {
1737                 if (session->s_state == CEPH_MDS_SESSION_NEW ||
1738                     session->s_state == CEPH_MDS_SESSION_CLOSING)
1739                         __open_session(mdsc, session);
1740                 list_add(&req->r_wait, &session->s_waiting);
1741                 goto out_session;
1742         }
1743
1744         /* send request */
1745         req->r_session = get_session(session);
1746         req->r_resend_mds = -1;   /* forget any previous mds hint */
1747
1748         if (req->r_request_started == 0)   /* note request start time */
1749                 req->r_request_started = jiffies;
1750
1751         err = __prepare_send_request(mdsc, req, mds);
1752         if (!err) {
1753                 ceph_msg_get(req->r_request);
1754                 ceph_con_send(&session->s_con, req->r_request);
1755         }
1756
1757 out_session:
1758         ceph_put_mds_session(session);
1759 out:
1760         return err;
1761
1762 finish:
1763         req->r_err = err;
1764         complete_request(mdsc, req);
1765         goto out;
1766 }
1767
1768 /*
1769  * called under mdsc->mutex
1770  */
1771 static void __wake_requests(struct ceph_mds_client *mdsc,
1772                             struct list_head *head)
1773 {
1774         struct ceph_mds_request *req, *nreq;
1775
1776         list_for_each_entry_safe(req, nreq, head, r_wait) {
1777                 list_del_init(&req->r_wait);
1778                 __do_request(mdsc, req);
1779         }
1780 }
1781
1782 /*
1783  * Wake up threads with requests pending for @mds, so that they can
1784  * resubmit their requests to a possibly different mds.
1785  */
1786 static void kick_requests(struct ceph_mds_client *mdsc, int mds)
1787 {
1788         struct ceph_mds_request *req;
1789         struct rb_node *p;
1790
1791         dout("kick_requests mds%d\n", mds);
1792         for (p = rb_first(&mdsc->request_tree); p; p = rb_next(p)) {
1793                 req = rb_entry(p, struct ceph_mds_request, r_node);
1794                 if (req->r_got_unsafe)
1795                         continue;
1796                 if (req->r_session &&
1797                     req->r_session->s_mds == mds) {
1798                         dout(" kicking tid %llu\n", req->r_tid);
1799                         put_request_session(req);
1800                         __do_request(mdsc, req);
1801                 }
1802         }
1803 }
1804
1805 void ceph_mdsc_submit_request(struct ceph_mds_client *mdsc,
1806                               struct ceph_mds_request *req)
1807 {
1808         dout("submit_request on %p\n", req);
1809         mutex_lock(&mdsc->mutex);
1810         __register_request(mdsc, req, NULL);
1811         __do_request(mdsc, req);
1812         mutex_unlock(&mdsc->mutex);
1813 }
1814
1815 /*
1816  * Synchrously perform an mds request.  Take care of all of the
1817  * session setup, forwarding, retry details.
1818  */
1819 int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
1820                          struct inode *dir,
1821                          struct ceph_mds_request *req)
1822 {
1823         int err;
1824
1825         dout("do_request on %p\n", req);
1826
1827         /* take CAP_PIN refs for r_inode, r_locked_dir, r_old_dentry */
1828         if (req->r_inode)
1829                 ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
1830         if (req->r_locked_dir)
1831                 ceph_get_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN);
1832         if (req->r_old_dentry)
1833                 ceph_get_cap_refs(
1834                         ceph_inode(req->r_old_dentry->d_parent->d_inode),
1835                         CEPH_CAP_PIN);
1836
1837         /* issue */
1838         mutex_lock(&mdsc->mutex);
1839         __register_request(mdsc, req, dir);
1840         __do_request(mdsc, req);
1841
1842         if (req->r_err) {
1843                 err = req->r_err;
1844                 __unregister_request(mdsc, req);
1845                 dout("do_request early error %d\n", err);
1846                 goto out;
1847         }
1848
1849         /* wait */
1850         mutex_unlock(&mdsc->mutex);
1851         dout("do_request waiting\n");
1852         if (req->r_timeout) {
1853                 err = (long)wait_for_completion_killable_timeout(
1854                         &req->r_completion, req->r_timeout);
1855                 if (err == 0)
1856                         err = -EIO;
1857         } else {
1858                 err = wait_for_completion_killable(&req->r_completion);
1859         }
1860         dout("do_request waited, got %d\n", err);
1861         mutex_lock(&mdsc->mutex);
1862
1863         /* only abort if we didn't race with a real reply */
1864         if (req->r_got_result) {
1865                 err = le32_to_cpu(req->r_reply_info.head->result);
1866         } else if (err < 0) {
1867                 dout("aborted request %lld with %d\n", req->r_tid, err);
1868
1869                 /*
1870                  * ensure we aren't running concurrently with
1871                  * ceph_fill_trace or ceph_readdir_prepopulate, which
1872                  * rely on locks (dir mutex) held by our caller.
1873                  */
1874                 mutex_lock(&req->r_fill_mutex);
1875                 req->r_err = err;
1876                 req->r_aborted = true;
1877                 mutex_unlock(&req->r_fill_mutex);
1878
1879                 if (req->r_locked_dir &&
1880                     (req->r_op & CEPH_MDS_OP_WRITE))
1881                         ceph_invalidate_dir_request(req);
1882         } else {
1883                 err = req->r_err;
1884         }
1885
1886 out:
1887         mutex_unlock(&mdsc->mutex);
1888         dout("do_request %p done, result %d\n", req, err);
1889         return err;
1890 }
1891
1892 /*
1893  * Invalidate dir I_COMPLETE, dentry lease state on an aborted MDS
1894  * namespace request.
1895  */
1896 void ceph_invalidate_dir_request(struct ceph_mds_request *req)
1897 {
1898         struct inode *inode = req->r_locked_dir;
1899         struct ceph_inode_info *ci = ceph_inode(inode);
1900
1901         dout("invalidate_dir_request %p (I_COMPLETE, lease(s))\n", inode);
1902         spin_lock(&inode->i_lock);
1903         ci->i_ceph_flags &= ~CEPH_I_COMPLETE;
1904         ci->i_release_count++;
1905         spin_unlock(&inode->i_lock);
1906
1907         if (req->r_dentry)
1908                 ceph_invalidate_dentry_lease(req->r_dentry);
1909         if (req->r_old_dentry)
1910                 ceph_invalidate_dentry_lease(req->r_old_dentry);
1911 }
1912
1913 /*
1914  * Handle mds reply.
1915  *
1916  * We take the session mutex and parse and process the reply immediately.
1917  * This preserves the logical ordering of replies, capabilities, etc., sent
1918  * by the MDS as they are applied to our local cache.
1919  */
1920 static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
1921 {
1922         struct ceph_mds_client *mdsc = session->s_mdsc;
1923         struct ceph_mds_request *req;
1924         struct ceph_mds_reply_head *head = msg->front.iov_base;
1925         struct ceph_mds_reply_info_parsed *rinfo;  /* parsed reply info */
1926         u64 tid;
1927         int err, result;
1928         int mds = session->s_mds;
1929
1930         if (msg->front.iov_len < sizeof(*head)) {
1931                 pr_err("mdsc_handle_reply got corrupt (short) reply\n");
1932                 ceph_msg_dump(msg);
1933                 return;
1934         }
1935
1936         /* get request, session */
1937         tid = le64_to_cpu(msg->hdr.tid);
1938         mutex_lock(&mdsc->mutex);
1939         req = __lookup_request(mdsc, tid);
1940         if (!req) {
1941                 dout("handle_reply on unknown tid %llu\n", tid);
1942                 mutex_unlock(&mdsc->mutex);
1943                 return;
1944         }
1945         dout("handle_reply %p\n", req);
1946
1947         /* correct session? */
1948         if (req->r_session != session) {
1949                 pr_err("mdsc_handle_reply got %llu on session mds%d"
1950                        " not mds%d\n", tid, session->s_mds,
1951                        req->r_session ? req->r_session->s_mds : -1);
1952                 mutex_unlock(&mdsc->mutex);
1953                 goto out;
1954         }
1955
1956         /* dup? */
1957         if ((req->r_got_unsafe && !head->safe) ||
1958             (req->r_got_safe && head->safe)) {
1959                 pr_warning("got a dup %s reply on %llu from mds%d\n",
1960                            head->safe ? "safe" : "unsafe", tid, mds);
1961                 mutex_unlock(&mdsc->mutex);
1962                 goto out;
1963         }
1964         if (req->r_got_safe && !head->safe) {
1965                 pr_warning("got unsafe after safe on %llu from mds%d\n",
1966                            tid, mds);
1967                 mutex_unlock(&mdsc->mutex);
1968                 goto out;
1969         }
1970
1971         result = le32_to_cpu(head->result);
1972
1973         /*
1974          * Handle an ESTALE
1975          * if we're not talking to the authority, send to them
1976          * if the authority has changed while we weren't looking,
1977          * send to new authority
1978          * Otherwise we just have to return an ESTALE
1979          */
1980         if (result == -ESTALE) {
1981                 dout("got ESTALE on request %llu", req->r_tid);
1982                 if (!req->r_inode) ; //do nothing; not an authority problem
1983                 else if (req->r_direct_mode != USE_AUTH_MDS) {
1984                         dout("not using auth, setting for that now");
1985                         req->r_direct_mode = USE_AUTH_MDS;
1986                         __do_request(mdsc, req);
1987                         mutex_unlock(&mdsc->mutex);
1988                         goto out;
1989                 } else  {
1990                         struct ceph_inode_info *ci = ceph_inode(req->r_inode);
1991                         struct ceph_cap *cap =
1992                                 ceph_get_cap_for_mds(ci, req->r_mds);;
1993
1994                         dout("already using auth");
1995                         if ((!cap || cap != ci->i_auth_cap) ||
1996                             (cap->mseq != req->r_sent_on_mseq)) {
1997                                 dout("but cap changed, so resending");
1998                                 __do_request(mdsc, req);
1999                                 mutex_unlock(&mdsc->mutex);
2000                                 goto out;
2001                         }
2002                 }
2003                 dout("have to return ESTALE on request %llu", req->r_tid);
2004         }
2005
2006
2007         if (head->safe) {
2008                 req->r_got_safe = true;
2009                 __unregister_request(mdsc, req);
2010                 complete_all(&req->r_safe_completion);
2011
2012                 if (req->r_got_unsafe) {
2013                         /*
2014                          * We already handled the unsafe response, now do the
2015                          * cleanup.  No need to examine the response; the MDS
2016                          * doesn't include any result info in the safe
2017                          * response.  And even if it did, there is nothing
2018                          * useful we could do with a revised return value.
2019                          */
2020                         dout("got safe reply %llu, mds%d\n", tid, mds);
2021                         list_del_init(&req->r_unsafe_item);
2022
2023                         /* last unsafe request during umount? */
2024                         if (mdsc->stopping && !__get_oldest_req(mdsc))
2025                                 complete_all(&mdsc->safe_umount_waiters);
2026                         mutex_unlock(&mdsc->mutex);
2027                         goto out;
2028                 }
2029         } else {
2030                 req->r_got_unsafe = true;
2031                 list_add_tail(&req->r_unsafe_item, &req->r_session->s_unsafe);
2032         }
2033
2034         dout("handle_reply tid %lld result %d\n", tid, result);
2035         rinfo = &req->r_reply_info;
2036         err = parse_reply_info(msg, rinfo);
2037         mutex_unlock(&mdsc->mutex);
2038
2039         mutex_lock(&session->s_mutex);
2040         if (err < 0) {
2041                 pr_err("mdsc_handle_reply got corrupt reply mds%d\n", mds);
2042                 ceph_msg_dump(msg);
2043                 goto out_err;
2044         }
2045
2046         /* snap trace */
2047         if (rinfo->snapblob_len) {
2048                 down_write(&mdsc->snap_rwsem);
2049                 ceph_update_snap_trace(mdsc, rinfo->snapblob,
2050                                rinfo->snapblob + rinfo->snapblob_len,
2051                                le32_to_cpu(head->op) == CEPH_MDS_OP_RMSNAP);
2052                 downgrade_write(&mdsc->snap_rwsem);
2053         } else {
2054                 down_read(&mdsc->snap_rwsem);
2055         }
2056
2057         /* insert trace into our cache */
2058         mutex_lock(&req->r_fill_mutex);
2059         err = ceph_fill_trace(mdsc->client->sb, req, req->r_session);
2060         if (err == 0) {
2061                 if (result == 0 && rinfo->dir_nr)
2062                         ceph_readdir_prepopulate(req, req->r_session);
2063                 ceph_unreserve_caps(mdsc, &req->r_caps_reservation);
2064         }
2065         mutex_unlock(&req->r_fill_mutex);
2066
2067         up_read(&mdsc->snap_rwsem);
2068 out_err:
2069         mutex_lock(&mdsc->mutex);
2070         if (!req->r_aborted) {
2071                 if (err) {
2072                         req->r_err = err;
2073                 } else {
2074                         req->r_reply = msg;
2075                         ceph_msg_get(msg);
2076                         req->r_got_result = true;
2077                 }
2078         } else {
2079                 dout("reply arrived after request %lld was aborted\n", tid);
2080         }
2081         mutex_unlock(&mdsc->mutex);
2082
2083         ceph_add_cap_releases(mdsc, req->r_session);
2084         mutex_unlock(&session->s_mutex);
2085
2086         /* kick calling process */
2087         complete_request(mdsc, req);
2088 out:
2089         ceph_mdsc_put_request(req);
2090         return;
2091 }
2092
2093
2094
2095 /*
2096  * handle mds notification that our request has been forwarded.
2097  */
2098 static void handle_forward(struct ceph_mds_client *mdsc,
2099                            struct ceph_mds_session *session,
2100                            struct ceph_msg *msg)
2101 {
2102         struct ceph_mds_request *req;
2103         u64 tid = le64_to_cpu(msg->hdr.tid);
2104         u32 next_mds;
2105         u32 fwd_seq;
2106         int err = -EINVAL;
2107         void *p = msg->front.iov_base;
2108         void *end = p + msg->front.iov_len;
2109
2110         ceph_decode_need(&p, end, 2*sizeof(u32), bad);
2111         next_mds = ceph_decode_32(&p);
2112         fwd_seq = ceph_decode_32(&p);
2113
2114         mutex_lock(&mdsc->mutex);
2115         req = __lookup_request(mdsc, tid);
2116         if (!req) {
2117                 dout("forward tid %llu to mds%d - req dne\n", tid, next_mds);
2118                 goto out;  /* dup reply? */
2119         }
2120
2121         if (req->r_aborted) {
2122                 dout("forward tid %llu aborted, unregistering\n", tid);
2123                 __unregister_request(mdsc, req);
2124         } else if (fwd_seq <= req->r_num_fwd) {
2125                 dout("forward tid %llu to mds%d - old seq %d <= %d\n",
2126                      tid, next_mds, req->r_num_fwd, fwd_seq);
2127         } else {
2128                 /* resend. forward race not possible; mds would drop */
2129                 dout("forward tid %llu to mds%d (we resend)\n", tid, next_mds);
2130                 BUG_ON(req->r_err);
2131                 BUG_ON(req->r_got_result);
2132                 req->r_num_fwd = fwd_seq;
2133                 req->r_resend_mds = next_mds;
2134                 put_request_session(req);
2135                 __do_request(mdsc, req);
2136         }
2137         ceph_mdsc_put_request(req);
2138 out:
2139         mutex_unlock(&mdsc->mutex);
2140         return;
2141
2142 bad:
2143         pr_err("mdsc_handle_forward decode error err=%d\n", err);
2144 }
2145
2146 /*
2147  * handle a mds session control message
2148  */
2149 static void handle_session(struct ceph_mds_session *session,
2150                            struct ceph_msg *msg)
2151 {
2152         struct ceph_mds_client *mdsc = session->s_mdsc;
2153         u32 op;
2154         u64 seq;
2155         int mds = session->s_mds;
2156         struct ceph_mds_session_head *h = msg->front.iov_base;
2157         int wake = 0;
2158
2159         /* decode */
2160         if (msg->front.iov_len != sizeof(*h))
2161                 goto bad;
2162         op = le32_to_cpu(h->op);
2163         seq = le64_to_cpu(h->seq);
2164
2165         mutex_lock(&mdsc->mutex);
2166         if (op == CEPH_SESSION_CLOSE)
2167                 __unregister_session(mdsc, session);
2168         /* FIXME: this ttl calculation is generous */
2169         session->s_ttl = jiffies + HZ*mdsc->mdsmap->m_session_autoclose;
2170         mutex_unlock(&mdsc->mutex);
2171
2172         mutex_lock(&session->s_mutex);
2173
2174         dout("handle_session mds%d %s %p state %s seq %llu\n",
2175              mds, ceph_session_op_name(op), session,
2176              session_state_name(session->s_state), seq);
2177
2178         if (session->s_state == CEPH_MDS_SESSION_HUNG) {
2179                 session->s_state = CEPH_MDS_SESSION_OPEN;
2180                 pr_info("mds%d came back\n", session->s_mds);
2181         }
2182
2183         switch (op) {
2184         case CEPH_SESSION_OPEN:
2185                 if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
2186                         pr_info("mds%d reconnect success\n", session->s_mds);
2187                 session->s_state = CEPH_MDS_SESSION_OPEN;
2188                 renewed_caps(mdsc, session, 0);
2189                 wake = 1;
2190                 if (mdsc->stopping)
2191                         __close_session(mdsc, session);
2192                 break;
2193
2194         case CEPH_SESSION_RENEWCAPS:
2195                 if (session->s_renew_seq == seq)
2196                         renewed_caps(mdsc, session, 1);
2197                 break;
2198
2199         case CEPH_SESSION_CLOSE:
2200                 if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
2201                         pr_info("mds%d reconnect denied\n", session->s_mds);
2202                 remove_session_caps(session);
2203                 wake = 1; /* for good measure */
2204                 complete_all(&mdsc->session_close_waiters);
2205                 kick_requests(mdsc, mds);
2206                 break;
2207
2208         case CEPH_SESSION_STALE:
2209                 pr_info("mds%d caps went stale, renewing\n",
2210                         session->s_mds);
2211                 spin_lock(&session->s_cap_lock);
2212                 session->s_cap_gen++;
2213                 session->s_cap_ttl = 0;
2214                 spin_unlock(&session->s_cap_lock);
2215                 send_renew_caps(mdsc, session);
2216                 break;
2217
2218         case CEPH_SESSION_RECALL_STATE:
2219                 trim_caps(mdsc, session, le32_to_cpu(h->max_caps));
2220                 break;
2221
2222         default:
2223                 pr_err("mdsc_handle_session bad op %d mds%d\n", op, mds);
2224                 WARN_ON(1);
2225         }
2226
2227         mutex_unlock(&session->s_mutex);
2228         if (wake) {
2229                 mutex_lock(&mdsc->mutex);
2230                 __wake_requests(mdsc, &session->s_waiting);
2231                 mutex_unlock(&mdsc->mutex);
2232         }
2233         return;
2234
2235 bad:
2236         pr_err("mdsc_handle_session corrupt message mds%d len %d\n", mds,
2237                (int)msg->front.iov_len);
2238         ceph_msg_dump(msg);
2239         return;
2240 }
2241
2242
2243 /*
2244  * called under session->mutex.
2245  */
2246 static void replay_unsafe_requests(struct ceph_mds_client *mdsc,
2247                                    struct ceph_mds_session *session)
2248 {
2249         struct ceph_mds_request *req, *nreq;
2250         int err;
2251
2252         dout("replay_unsafe_requests mds%d\n", session->s_mds);
2253
2254         mutex_lock(&mdsc->mutex);
2255         list_for_each_entry_safe(req, nreq, &session->s_unsafe, r_unsafe_item) {
2256                 err = __prepare_send_request(mdsc, req, session->s_mds);
2257                 if (!err) {
2258                         ceph_msg_get(req->r_request);
2259                         ceph_con_send(&session->s_con, req->r_request);
2260                 }
2261         }
2262         mutex_unlock(&mdsc->mutex);
2263 }
2264
2265 /*
2266  * Encode information about a cap for a reconnect with the MDS.
2267  */
2268 static int encode_caps_cb(struct inode *inode, struct ceph_cap *cap,
2269                           void *arg)
2270 {
2271         struct ceph_mds_cap_reconnect rec;
2272         struct ceph_inode_info *ci;
2273         struct ceph_pagelist *pagelist = arg;
2274         char *path;
2275         int pathlen, err;
2276         u64 pathbase;
2277         struct dentry *dentry;
2278
2279         ci = cap->ci;
2280
2281         dout(" adding %p ino %llx.%llx cap %p %lld %s\n",
2282              inode, ceph_vinop(inode), cap, cap->cap_id,
2283              ceph_cap_string(cap->issued));
2284         err = ceph_pagelist_encode_64(pagelist, ceph_ino(inode));
2285         if (err)
2286                 return err;
2287
2288         dentry = d_find_alias(inode);
2289         if (dentry) {
2290                 path = ceph_mdsc_build_path(dentry, &pathlen, &pathbase, 0);
2291                 if (IS_ERR(path)) {
2292                         err = PTR_ERR(path);
2293                         BUG_ON(err);
2294                 }
2295         } else {
2296                 path = NULL;
2297                 pathlen = 0;
2298         }
2299         err = ceph_pagelist_encode_string(pagelist, path, pathlen);
2300         if (err)
2301                 goto out;
2302
2303         spin_lock(&inode->i_lock);
2304         cap->seq = 0;        /* reset cap seq */
2305         cap->issue_seq = 0;  /* and issue_seq */
2306         rec.cap_id = cpu_to_le64(cap->cap_id);
2307         rec.pathbase = cpu_to_le64(pathbase);
2308         rec.wanted = cpu_to_le32(__ceph_caps_wanted(ci));
2309         rec.issued = cpu_to_le32(cap->issued);
2310         rec.size = cpu_to_le64(inode->i_size);
2311         ceph_encode_timespec(&rec.mtime, &inode->i_mtime);
2312         ceph_encode_timespec(&rec.atime, &inode->i_atime);
2313         rec.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
2314         spin_unlock(&inode->i_lock);
2315
2316         err = ceph_pagelist_append(pagelist, &rec, sizeof(rec));
2317
2318 out:
2319         kfree(path);
2320         dput(dentry);
2321         return err;
2322 }
2323
2324
2325 /*
2326  * If an MDS fails and recovers, clients need to reconnect in order to
2327  * reestablish shared state.  This includes all caps issued through
2328  * this session _and_ the snap_realm hierarchy.  Because it's not
2329  * clear which snap realms the mds cares about, we send everything we
2330  * know about.. that ensures we'll then get any new info the
2331  * recovering MDS might have.
2332  *
2333  * This is a relatively heavyweight operation, but it's rare.
2334  *
2335  * called with mdsc->mutex held.
2336  */
2337 static void send_mds_reconnect(struct ceph_mds_client *mdsc,
2338                                struct ceph_mds_session *session)
2339 {
2340         struct ceph_msg *reply;
2341         struct rb_node *p;
2342         int mds = session->s_mds;
2343         int err = -ENOMEM;
2344         struct ceph_pagelist *pagelist;
2345
2346         pr_info("mds%d reconnect start\n", mds);
2347
2348         pagelist = kmalloc(sizeof(*pagelist), GFP_NOFS);
2349         if (!pagelist)
2350                 goto fail_nopagelist;
2351         ceph_pagelist_init(pagelist);
2352
2353         reply = ceph_msg_new(CEPH_MSG_CLIENT_RECONNECT, 0, GFP_NOFS);
2354         if (!reply)
2355                 goto fail_nomsg;
2356
2357         mutex_lock(&session->s_mutex);
2358         session->s_state = CEPH_MDS_SESSION_RECONNECTING;
2359         session->s_seq = 0;
2360
2361         ceph_con_open(&session->s_con,
2362                       ceph_mdsmap_get_addr(mdsc->mdsmap, mds));
2363
2364         /* replay unsafe requests */
2365         replay_unsafe_requests(mdsc, session);
2366
2367         down_read(&mdsc->snap_rwsem);
2368
2369         dout("session %p state %s\n", session,
2370              session_state_name(session->s_state));
2371
2372         /* drop old cap expires; we're about to reestablish that state */
2373         discard_cap_releases(mdsc, session);
2374
2375         /* traverse this session's caps */
2376         err = ceph_pagelist_encode_32(pagelist, session->s_nr_caps);
2377         if (err)
2378                 goto fail;
2379         err = iterate_session_caps(session, encode_caps_cb, pagelist);
2380         if (err < 0)
2381                 goto fail;
2382
2383         /*
2384          * snaprealms.  we provide mds with the ino, seq (version), and
2385          * parent for all of our realms.  If the mds has any newer info,
2386          * it will tell us.
2387          */
2388         for (p = rb_first(&mdsc->snap_realms); p; p = rb_next(p)) {
2389                 struct ceph_snap_realm *realm =
2390                         rb_entry(p, struct ceph_snap_realm, node);
2391                 struct ceph_mds_snaprealm_reconnect sr_rec;
2392
2393                 dout(" adding snap realm %llx seq %lld parent %llx\n",
2394                      realm->ino, realm->seq, realm->parent_ino);
2395                 sr_rec.ino = cpu_to_le64(realm->ino);
2396                 sr_rec.seq = cpu_to_le64(realm->seq);
2397                 sr_rec.parent = cpu_to_le64(realm->parent_ino);
2398                 err = ceph_pagelist_append(pagelist, &sr_rec, sizeof(sr_rec));
2399                 if (err)
2400                         goto fail;
2401         }
2402
2403         reply->pagelist = pagelist;
2404         reply->hdr.data_len = cpu_to_le32(pagelist->length);
2405         reply->nr_pages = calc_pages_for(0, pagelist->length);
2406         ceph_con_send(&session->s_con, reply);
2407
2408         mutex_unlock(&session->s_mutex);
2409
2410         mutex_lock(&mdsc->mutex);
2411         __wake_requests(mdsc, &session->s_waiting);
2412         mutex_unlock(&mdsc->mutex);
2413
2414         up_read(&mdsc->snap_rwsem);
2415         return;
2416
2417 fail:
2418         ceph_msg_put(reply);
2419         up_read(&mdsc->snap_rwsem);
2420         mutex_unlock(&session->s_mutex);
2421 fail_nomsg:
2422         ceph_pagelist_release(pagelist);
2423         kfree(pagelist);
2424 fail_nopagelist:
2425         pr_err("error %d preparing reconnect for mds%d\n", err, mds);
2426         return;
2427 }
2428
2429
2430 /*
2431  * compare old and new mdsmaps, kicking requests
2432  * and closing out old connections as necessary
2433  *
2434  * called under mdsc->mutex.
2435  */
2436 static void check_new_map(struct ceph_mds_client *mdsc,
2437                           struct ceph_mdsmap *newmap,
2438                           struct ceph_mdsmap *oldmap)
2439 {
2440         int i;
2441         int oldstate, newstate;
2442         struct ceph_mds_session *s;
2443
2444         dout("check_new_map new %u old %u\n",
2445              newmap->m_epoch, oldmap->m_epoch);
2446
2447         for (i = 0; i < oldmap->m_max_mds && i < mdsc->max_sessions; i++) {
2448                 if (mdsc->sessions[i] == NULL)
2449                         continue;
2450                 s = mdsc->sessions[i];
2451                 oldstate = ceph_mdsmap_get_state(oldmap, i);
2452                 newstate = ceph_mdsmap_get_state(newmap, i);
2453
2454                 dout("check_new_map mds%d state %s%s -> %s%s (session %s)\n",
2455                      i, ceph_mds_state_name(oldstate),
2456                      ceph_mdsmap_is_laggy(oldmap, i) ? " (laggy)" : "",
2457                      ceph_mds_state_name(newstate),
2458                      ceph_mdsmap_is_laggy(newmap, i) ? " (laggy)" : "",
2459                      session_state_name(s->s_state));
2460
2461                 if (memcmp(ceph_mdsmap_get_addr(oldmap, i),
2462                            ceph_mdsmap_get_addr(newmap, i),
2463                            sizeof(struct ceph_entity_addr))) {
2464                         if (s->s_state == CEPH_MDS_SESSION_OPENING) {
2465                                 /* the session never opened, just close it
2466                                  * out now */
2467                                 __wake_requests(mdsc, &s->s_waiting);
2468                                 __unregister_session(mdsc, s);
2469                         } else {
2470                                 /* just close it */
2471                                 mutex_unlock(&mdsc->mutex);
2472                                 mutex_lock(&s->s_mutex);
2473                                 mutex_lock(&mdsc->mutex);
2474                                 ceph_con_close(&s->s_con);
2475                                 mutex_unlock(&s->s_mutex);
2476                                 s->s_state = CEPH_MDS_SESSION_RESTARTING;
2477                         }
2478
2479                         /* kick any requests waiting on the recovering mds */
2480                         kick_requests(mdsc, i);
2481                 } else if (oldstate == newstate) {
2482                         continue;  /* nothing new with this mds */
2483                 }
2484
2485                 /*
2486                  * send reconnect?
2487                  */
2488                 if (s->s_state == CEPH_MDS_SESSION_RESTARTING &&
2489                     newstate >= CEPH_MDS_STATE_RECONNECT) {
2490                         mutex_unlock(&mdsc->mutex);
2491                         send_mds_reconnect(mdsc, s);
2492                         mutex_lock(&mdsc->mutex);
2493                 }
2494
2495                 /*
2496                  * kick request on any mds that has gone active.
2497                  */
2498                 if (oldstate < CEPH_MDS_STATE_ACTIVE &&
2499                     newstate >= CEPH_MDS_STATE_ACTIVE) {
2500                         if (oldstate != CEPH_MDS_STATE_CREATING &&
2501                             oldstate != CEPH_MDS_STATE_STARTING)
2502                                 pr_info("mds%d recovery completed\n", s->s_mds);
2503                         kick_requests(mdsc, i);
2504                         ceph_kick_flushing_caps(mdsc, s);
2505                         wake_up_session_caps(s, 1);
2506                 }
2507         }
2508
2509         for (i = 0; i < newmap->m_max_mds && i < mdsc->max_sessions; i++) {
2510                 s = mdsc->sessions[i];
2511                 if (!s)
2512                         continue;
2513                 if (!ceph_mdsmap_is_laggy(newmap, i))
2514                         continue;
2515                 if (s->s_state == CEPH_MDS_SESSION_OPEN ||
2516                     s->s_state == CEPH_MDS_SESSION_HUNG ||
2517                     s->s_state == CEPH_MDS_SESSION_CLOSING) {
2518                         dout(" connecting to export targets of laggy mds%d\n",
2519                              i);
2520                         __open_export_target_sessions(mdsc, s);
2521                 }
2522         }
2523 }
2524
2525
2526
2527 /*
2528  * leases
2529  */
2530
2531 /*
2532  * caller must hold session s_mutex, dentry->d_lock
2533  */
2534 void __ceph_mdsc_drop_dentry_lease(struct dentry *dentry)
2535 {
2536         struct ceph_dentry_info *di = ceph_dentry(dentry);
2537
2538         ceph_put_mds_session(di->lease_session);
2539         di->lease_session = NULL;
2540 }
2541
2542 static void handle_lease(struct ceph_mds_client *mdsc,
2543                          struct ceph_mds_session *session,
2544                          struct ceph_msg *msg)
2545 {
2546         struct super_block *sb = mdsc->client->sb;
2547         struct inode *inode;
2548         struct ceph_inode_info *ci;
2549         struct dentry *parent, *dentry;
2550         struct ceph_dentry_info *di;
2551         int mds = session->s_mds;
2552         struct ceph_mds_lease *h = msg->front.iov_base;
2553         u32 seq;
2554         struct ceph_vino vino;
2555         int mask;
2556         struct qstr dname;
2557         int release = 0;
2558
2559         dout("handle_lease from mds%d\n", mds);
2560
2561         /* decode */
2562         if (msg->front.iov_len < sizeof(*h) + sizeof(u32))
2563                 goto bad;
2564         vino.ino = le64_to_cpu(h->ino);
2565         vino.snap = CEPH_NOSNAP;
2566         mask = le16_to_cpu(h->mask);
2567         seq = le32_to_cpu(h->seq);
2568         dname.name = (void *)h + sizeof(*h) + sizeof(u32);
2569         dname.len = msg->front.iov_len - sizeof(*h) - sizeof(u32);
2570         if (dname.len != get_unaligned_le32(h+1))
2571                 goto bad;
2572
2573         mutex_lock(&session->s_mutex);
2574         session->s_seq++;
2575
2576         /* lookup inode */
2577         inode = ceph_find_inode(sb, vino);
2578         dout("handle_lease %s, mask %d, ino %llx %p %.*s\n",
2579              ceph_lease_op_name(h->action), mask, vino.ino, inode,
2580              dname.len, dname.name);
2581         if (inode == NULL) {
2582                 dout("handle_lease no inode %llx\n", vino.ino);
2583                 goto release;
2584         }
2585         ci = ceph_inode(inode);
2586
2587         /* dentry */
2588         parent = d_find_alias(inode);
2589         if (!parent) {
2590                 dout("no parent dentry on inode %p\n", inode);
2591                 WARN_ON(1);
2592                 goto release;  /* hrm... */
2593         }
2594         dname.hash = full_name_hash(dname.name, dname.len);
2595         dentry = d_lookup(parent, &dname);
2596         dput(parent);
2597         if (!dentry)
2598                 goto release;
2599
2600         spin_lock(&dentry->d_lock);
2601         di = ceph_dentry(dentry);
2602         switch (h->action) {
2603         case CEPH_MDS_LEASE_REVOKE:
2604                 if (di && di->lease_session == session) {
2605                         if (ceph_seq_cmp(di->lease_seq, seq) > 0)
2606                                 h->seq = cpu_to_le32(di->lease_seq);
2607                         __ceph_mdsc_drop_dentry_lease(dentry);
2608                 }
2609                 release = 1;
2610                 break;
2611
2612         case CEPH_MDS_LEASE_RENEW:
2613                 if (di && di->lease_session == session &&
2614                     di->lease_gen == session->s_cap_gen &&
2615                     di->lease_renew_from &&
2616                     di->lease_renew_after == 0) {
2617                         unsigned long duration =
2618                                 le32_to_cpu(h->duration_ms) * HZ / 1000;
2619
2620                         di->lease_seq = seq;
2621                         dentry->d_time = di->lease_renew_from + duration;
2622                         di->lease_renew_after = di->lease_renew_from +
2623                                 (duration >> 1);
2624                         di->lease_renew_from = 0;
2625                 }
2626                 break;
2627         }
2628         spin_unlock(&dentry->d_lock);
2629         dput(dentry);
2630
2631         if (!release)
2632                 goto out;
2633
2634 release:
2635         /* let's just reuse the same message */
2636         h->action = CEPH_MDS_LEASE_REVOKE_ACK;
2637         ceph_msg_get(msg);
2638         ceph_con_send(&session->s_con, msg);
2639
2640 out:
2641         iput(inode);
2642         mutex_unlock(&session->s_mutex);
2643         return;
2644
2645 bad:
2646         pr_err("corrupt lease message\n");
2647         ceph_msg_dump(msg);
2648 }
2649
2650 void ceph_mdsc_lease_send_msg(struct ceph_mds_session *session,
2651                               struct inode *inode,
2652                               struct dentry *dentry, char action,
2653                               u32 seq)
2654 {
2655         struct ceph_msg *msg;
2656         struct ceph_mds_lease *lease;
2657         int len = sizeof(*lease) + sizeof(u32);
2658         int dnamelen = 0;
2659
2660         dout("lease_send_msg inode %p dentry %p %s to mds%d\n",
2661              inode, dentry, ceph_lease_op_name(action), session->s_mds);
2662         dnamelen = dentry->d_name.len;
2663         len += dnamelen;
2664
2665         msg = ceph_msg_new(CEPH_MSG_CLIENT_LEASE, len, GFP_NOFS);
2666         if (!msg)
2667                 return;
2668         lease = msg->front.iov_base;
2669         lease->action = action;
2670         lease->mask = cpu_to_le16(1);
2671         lease->ino = cpu_to_le64(ceph_vino(inode).ino);
2672         lease->first = lease->last = cpu_to_le64(ceph_vino(inode).snap);
2673         lease->seq = cpu_to_le32(seq);
2674         put_unaligned_le32(dnamelen, lease + 1);
2675         memcpy((void *)(lease + 1) + 4, dentry->d_name.name, dnamelen);
2676
2677         /*
2678          * if this is a preemptive lease RELEASE, no need to
2679          * flush request stream, since the actual request will
2680          * soon follow.
2681          */
2682         msg->more_to_follow = (action == CEPH_MDS_LEASE_RELEASE);
2683
2684         ceph_con_send(&session->s_con, msg);
2685 }
2686
2687 /*
2688  * Preemptively release a lease we expect to invalidate anyway.
2689  * Pass @inode always, @dentry is optional.
2690  */
2691 void ceph_mdsc_lease_release(struct ceph_mds_client *mdsc, struct inode *inode,
2692                              struct dentry *dentry, int mask)
2693 {
2694         struct ceph_dentry_info *di;
2695         struct ceph_mds_session *session;
2696         u32 seq;
2697
2698         BUG_ON(inode == NULL);
2699         BUG_ON(dentry == NULL);
2700         BUG_ON(mask == 0);
2701
2702         /* is dentry lease valid? */
2703         spin_lock(&dentry->d_lock);
2704         di = ceph_dentry(dentry);
2705         if (!di || !di->lease_session ||
2706             di->lease_session->s_mds < 0 ||
2707             di->lease_gen != di->lease_session->s_cap_gen ||
2708             !time_before(jiffies, dentry->d_time)) {
2709                 dout("lease_release inode %p dentry %p -- "
2710                      "no lease on %d\n",
2711                      inode, dentry, mask);
2712                 spin_unlock(&dentry->d_lock);
2713                 return;
2714         }
2715
2716         /* we do have a lease on this dentry; note mds and seq */
2717         session = ceph_get_mds_session(di->lease_session);
2718         seq = di->lease_seq;
2719         __ceph_mdsc_drop_dentry_lease(dentry);
2720         spin_unlock(&dentry->d_lock);
2721
2722         dout("lease_release inode %p dentry %p mask %d to mds%d\n",
2723              inode, dentry, mask, session->s_mds);
2724         ceph_mdsc_lease_send_msg(session, inode, dentry,
2725                                  CEPH_MDS_LEASE_RELEASE, seq);
2726         ceph_put_mds_session(session);
2727 }
2728
2729 /*
2730  * drop all leases (and dentry refs) in preparation for umount
2731  */
2732 static void drop_leases(struct ceph_mds_client *mdsc)
2733 {
2734         int i;
2735
2736         dout("drop_leases\n");
2737         mutex_lock(&mdsc->mutex);
2738         for (i = 0; i < mdsc->max_sessions; i++) {
2739                 struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
2740                 if (!s)
2741                         continue;
2742                 mutex_unlock(&mdsc->mutex);
2743                 mutex_lock(&s->s_mutex);
2744                 mutex_unlock(&s->s_mutex);
2745                 ceph_put_mds_session(s);
2746                 mutex_lock(&mdsc->mutex);
2747         }
2748         mutex_unlock(&mdsc->mutex);
2749 }
2750
2751
2752
2753 /*
2754  * delayed work -- periodically trim expired leases, renew caps with mds
2755  */
2756 static void schedule_delayed(struct ceph_mds_client *mdsc)
2757 {
2758         int delay = 5;
2759         unsigned hz = round_jiffies_relative(HZ * delay);
2760         schedule_delayed_work(&mdsc->delayed_work, hz);
2761 }
2762
2763 static void delayed_work(struct work_struct *work)
2764 {
2765         int i;
2766         struct ceph_mds_client *mdsc =
2767                 container_of(work, struct ceph_mds_client, delayed_work.work);
2768         int renew_interval;
2769         int renew_caps;
2770
2771         dout("mdsc delayed_work\n");
2772         ceph_check_delayed_caps(mdsc);
2773
2774         mutex_lock(&mdsc->mutex);
2775         renew_interval = mdsc->mdsmap->m_session_timeout >> 2;
2776         renew_caps = time_after_eq(jiffies, HZ*renew_interval +
2777                                    mdsc->last_renew_caps);
2778         if (renew_caps)
2779                 mdsc->last_renew_caps = jiffies;
2780
2781         for (i = 0; i < mdsc->max_sessions; i++) {
2782                 struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
2783                 if (s == NULL)
2784                         continue;
2785                 if (s->s_state == CEPH_MDS_SESSION_CLOSING) {
2786                         dout("resending session close request for mds%d\n",
2787                              s->s_mds);
2788                         request_close_session(mdsc, s);
2789                         ceph_put_mds_session(s);
2790                         continue;
2791                 }
2792                 if (s->s_ttl && time_after(jiffies, s->s_ttl)) {
2793                         if (s->s_state == CEPH_MDS_SESSION_OPEN) {
2794                                 s->s_state = CEPH_MDS_SESSION_HUNG;
2795                                 pr_info("mds%d hung\n", s->s_mds);
2796                         }
2797                 }
2798                 if (s->s_state < CEPH_MDS_SESSION_OPEN) {
2799                         /* this mds is failed or recovering, just wait */
2800                         ceph_put_mds_session(s);
2801                         continue;
2802                 }
2803                 mutex_unlock(&mdsc->mutex);
2804
2805                 mutex_lock(&s->s_mutex);
2806                 if (renew_caps)
2807                         send_renew_caps(mdsc, s);
2808                 else
2809                         ceph_con_keepalive(&s->s_con);
2810                 ceph_add_cap_releases(mdsc, s);
2811                 if (s->s_state == CEPH_MDS_SESSION_OPEN ||
2812                     s->s_state == CEPH_MDS_SESSION_HUNG)
2813                         ceph_send_cap_releases(mdsc, s);
2814                 mutex_unlock(&s->s_mutex);
2815                 ceph_put_mds_session(s);
2816
2817                 mutex_lock(&mdsc->mutex);
2818         }
2819         mutex_unlock(&mdsc->mutex);
2820
2821         schedule_delayed(mdsc);
2822 }
2823
2824
2825 int ceph_mdsc_init(struct ceph_mds_client *mdsc, struct ceph_client *client)
2826 {
2827         mdsc->client = client;
2828         mutex_init(&mdsc->mutex);
2829         mdsc->mdsmap = kzalloc(sizeof(*mdsc->mdsmap), GFP_NOFS);
2830         if (mdsc->mdsmap == NULL)
2831                 return -ENOMEM;
2832
2833         init_completion(&mdsc->safe_umount_waiters);
2834         init_completion(&mdsc->session_close_waiters);
2835         INIT_LIST_HEAD(&mdsc->waiting_for_map);
2836         mdsc->sessions = NULL;
2837         mdsc->max_sessions = 0;
2838         mdsc->stopping = 0;
2839         init_rwsem(&mdsc->snap_rwsem);
2840         mdsc->snap_realms = RB_ROOT;
2841         INIT_LIST_HEAD(&mdsc->snap_empty);
2842         spin_lock_init(&mdsc->snap_empty_lock);
2843         mdsc->last_tid = 0;
2844         mdsc->request_tree = RB_ROOT;
2845         INIT_DELAYED_WORK(&mdsc->delayed_work, delayed_work);
2846         mdsc->last_renew_caps = jiffies;
2847         INIT_LIST_HEAD(&mdsc->cap_delay_list);
2848         spin_lock_init(&mdsc->cap_delay_lock);
2849         INIT_LIST_HEAD(&mdsc->snap_flush_list);
2850         spin_lock_init(&mdsc->snap_flush_lock);
2851         mdsc->cap_flush_seq = 0;
2852         INIT_LIST_HEAD(&mdsc->cap_dirty);
2853         mdsc->num_cap_flushing = 0;
2854         spin_lock_init(&mdsc->cap_dirty_lock);
2855         init_waitqueue_head(&mdsc->cap_flushing_wq);
2856         spin_lock_init(&mdsc->dentry_lru_lock);
2857         INIT_LIST_HEAD(&mdsc->dentry_lru);
2858
2859         ceph_caps_init(mdsc);
2860         ceph_adjust_min_caps(mdsc, client->min_caps);
2861
2862         return 0;
2863 }
2864
2865 /*
2866  * Wait for safe replies on open mds requests.  If we time out, drop
2867  * all requests from the tree to avoid dangling dentry refs.
2868  */
2869 static void wait_requests(struct ceph_mds_client *mdsc)
2870 {
2871         struct ceph_mds_request *req;
2872         struct ceph_client *client = mdsc->client;
2873
2874         mutex_lock(&mdsc->mutex);
2875         if (__get_oldest_req(mdsc)) {
2876                 mutex_unlock(&mdsc->mutex);
2877
2878                 dout("wait_requests waiting for requests\n");
2879                 wait_for_completion_timeout(&mdsc->safe_umount_waiters,
2880                                     client->mount_args->mount_timeout * HZ);
2881
2882                 /* tear down remaining requests */
2883                 mutex_lock(&mdsc->mutex);
2884                 while ((req = __get_oldest_req(mdsc))) {
2885                         dout("wait_requests timed out on tid %llu\n",
2886                              req->r_tid);
2887                         __unregister_request(mdsc, req);
2888                 }
2889         }
2890         mutex_unlock(&mdsc->mutex);
2891         dout("wait_requests done\n");
2892 }
2893
2894 /*
2895  * called before mount is ro, and before dentries are torn down.
2896  * (hmm, does this still race with new lookups?)
2897  */
2898 void ceph_mdsc_pre_umount(struct ceph_mds_client *mdsc)
2899 {
2900         dout("pre_umount\n");
2901         mdsc->stopping = 1;
2902
2903         drop_leases(mdsc);
2904         ceph_flush_dirty_caps(mdsc);
2905         wait_requests(mdsc);
2906
2907         /*
2908          * wait for reply handlers to drop their request refs and
2909          * their inode/dcache refs
2910          */
2911         ceph_msgr_flush();
2912 }
2913
2914 /*
2915  * wait for all write mds requests to flush.
2916  */
2917 static void wait_unsafe_requests(struct ceph_mds_client *mdsc, u64 want_tid)
2918 {
2919         struct ceph_mds_request *req = NULL, *nextreq;
2920         struct rb_node *n;
2921
2922         mutex_lock(&mdsc->mutex);
2923         dout("wait_unsafe_requests want %lld\n", want_tid);
2924 restart:
2925         req = __get_oldest_req(mdsc);
2926         while (req && req->r_tid <= want_tid) {
2927                 /* find next request */
2928                 n = rb_next(&req->r_node);
2929                 if (n)
2930                         nextreq = rb_entry(n, struct ceph_mds_request, r_node);
2931                 else
2932                         nextreq = NULL;
2933                 if ((req->r_op & CEPH_MDS_OP_WRITE)) {
2934                         /* write op */
2935                         ceph_mdsc_get_request(req);
2936                         if (nextreq)
2937                                 ceph_mdsc_get_request(nextreq);
2938                         mutex_unlock(&mdsc->mutex);
2939                         dout("wait_unsafe_requests  wait on %llu (want %llu)\n",
2940                              req->r_tid, want_tid);
2941                         wait_for_completion(&req->r_safe_completion);
2942                         mutex_lock(&mdsc->mutex);
2943                         ceph_mdsc_put_request(req);
2944                         if (!nextreq)
2945                                 break;  /* next dne before, so we're done! */
2946                         if (RB_EMPTY_NODE(&nextreq->r_node)) {
2947                                 /* next request was removed from tree */
2948                                 ceph_mdsc_put_request(nextreq);
2949                                 goto restart;
2950                         }
2951                         ceph_mdsc_put_request(nextreq);  /* won't go away */
2952                 }
2953                 req = nextreq;
2954         }
2955         mutex_unlock(&mdsc->mutex);
2956         dout("wait_unsafe_requests done\n");
2957 }
2958
2959 void ceph_mdsc_sync(struct ceph_mds_client *mdsc)
2960 {
2961         u64 want_tid, want_flush;
2962
2963         if (mdsc->client->mount_state == CEPH_MOUNT_SHUTDOWN)
2964                 return;
2965
2966         dout("sync\n");
2967         mutex_lock(&mdsc->mutex);
2968         want_tid = mdsc->last_tid;
2969         want_flush = mdsc->cap_flush_seq;
2970         mutex_unlock(&mdsc->mutex);
2971         dout("sync want tid %lld flush_seq %lld\n", want_tid, want_flush);
2972
2973         ceph_flush_dirty_caps(mdsc);
2974
2975         wait_unsafe_requests(mdsc, want_tid);
2976         wait_event(mdsc->cap_flushing_wq, check_cap_flush(mdsc, want_flush));
2977 }
2978
2979
2980 /*
2981  * called after sb is ro.
2982  */
2983 void ceph_mdsc_close_sessions(struct ceph_mds_client *mdsc)
2984 {
2985         struct ceph_mds_session *session;
2986         int i;
2987         int n;
2988         struct ceph_client *client = mdsc->client;
2989         unsigned long started, timeout = client->mount_args->mount_timeout * HZ;
2990
2991         dout("close_sessions\n");
2992
2993         mutex_lock(&mdsc->mutex);
2994
2995         /* close sessions */
2996         started = jiffies;
2997         while (time_before(jiffies, started + timeout)) {
2998                 dout("closing sessions\n");
2999                 n = 0;
3000                 for (i = 0; i < mdsc->max_sessions; i++) {
3001                         session = __ceph_lookup_mds_session(mdsc, i);
3002                         if (!session)
3003                                 continue;
3004                         mutex_unlock(&mdsc->mutex);
3005                         mutex_lock(&session->s_mutex);
3006                         __close_session(mdsc, session);
3007                         mutex_unlock(&session->s_mutex);
3008                         ceph_put_mds_session(session);
3009                         mutex_lock(&mdsc->mutex);
3010                         n++;
3011                 }
3012                 if (n == 0)
3013                         break;
3014
3015                 if (client->mount_state == CEPH_MOUNT_SHUTDOWN)
3016                         break;
3017
3018                 dout("waiting for sessions to close\n");
3019                 mutex_unlock(&mdsc->mutex);
3020                 wait_for_completion_timeout(&mdsc->session_close_waiters,
3021                                             timeout);
3022                 mutex_lock(&mdsc->mutex);
3023         }
3024
3025         /* tear down remaining sessions */
3026         for (i = 0; i < mdsc->max_sessions; i++) {
3027                 if (mdsc->sessions[i]) {
3028                         session = get_session(mdsc->sessions[i]);
3029                         __unregister_session(mdsc, session);
3030                         mutex_unlock(&mdsc->mutex);
3031                         mutex_lock(&session->s_mutex);
3032                         remove_session_caps(session);
3033                         mutex_unlock(&session->s_mutex);
3034                         ceph_put_mds_session(session);
3035                         mutex_lock(&mdsc->mutex);
3036                 }
3037         }
3038
3039         WARN_ON(!list_empty(&mdsc->cap_delay_list));
3040
3041         mutex_unlock(&mdsc->mutex);
3042
3043         ceph_cleanup_empty_realms(mdsc);
3044
3045         cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */
3046
3047         dout("stopped\n");
3048 }
3049
3050 void ceph_mdsc_stop(struct ceph_mds_client *mdsc)
3051 {
3052         dout("stop\n");
3053         cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */
3054         if (mdsc->mdsmap)
3055                 ceph_mdsmap_destroy(mdsc->mdsmap);
3056         kfree(mdsc->sessions);
3057         ceph_caps_finalize(mdsc);
3058 }
3059
3060
3061 /*
3062  * handle mds map update.
3063  */
3064 void ceph_mdsc_handle_map(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
3065 {
3066         u32 epoch;
3067         u32 maplen;
3068         void *p = msg->front.iov_base;
3069         void *end = p + msg->front.iov_len;
3070         struct ceph_mdsmap *newmap, *oldmap;
3071         struct ceph_fsid fsid;
3072         int err = -EINVAL;
3073
3074         ceph_decode_need(&p, end, sizeof(fsid)+2*sizeof(u32), bad);
3075         ceph_decode_copy(&p, &fsid, sizeof(fsid));
3076         if (ceph_check_fsid(mdsc->client, &fsid) < 0)
3077                 return;
3078         epoch = ceph_decode_32(&p);
3079         maplen = ceph_decode_32(&p);
3080         dout("handle_map epoch %u len %d\n", epoch, (int)maplen);
3081
3082         /* do we need it? */
3083         ceph_monc_got_mdsmap(&mdsc->client->monc, epoch);
3084         mutex_lock(&mdsc->mutex);
3085         if (mdsc->mdsmap && epoch <= mdsc->mdsmap->m_epoch) {
3086                 dout("handle_map epoch %u <= our %u\n",
3087                      epoch, mdsc->mdsmap->m_epoch);
3088                 mutex_unlock(&mdsc->mutex);
3089                 return;
3090         }
3091
3092         newmap = ceph_mdsmap_decode(&p, end);
3093         if (IS_ERR(newmap)) {
3094                 err = PTR_ERR(newmap);
3095                 goto bad_unlock;
3096         }
3097
3098         /* swap into place */
3099         if (mdsc->mdsmap) {
3100                 oldmap = mdsc->mdsmap;
3101                 mdsc->mdsmap = newmap;
3102                 check_new_map(mdsc, newmap, oldmap);
3103                 ceph_mdsmap_destroy(oldmap);
3104         } else {
3105                 mdsc->mdsmap = newmap;  /* first mds map */
3106         }
3107         mdsc->client->sb->s_maxbytes = mdsc->mdsmap->m_max_file_size;
3108
3109         __wake_requests(mdsc, &mdsc->waiting_for_map);
3110
3111         mutex_unlock(&mdsc->mutex);
3112         schedule_delayed(mdsc);
3113         return;
3114
3115 bad_unlock:
3116         mutex_unlock(&mdsc->mutex);
3117 bad:
3118         pr_err("error decoding mdsmap %d\n", err);
3119         return;
3120 }
3121
3122 static struct ceph_connection *con_get(struct ceph_connection *con)
3123 {
3124         struct ceph_mds_session *s = con->private;
3125
3126         if (get_session(s)) {
3127                 dout("mdsc con_get %p ok (%d)\n", s, atomic_read(&s->s_ref));
3128                 return con;
3129         }
3130         dout("mdsc con_get %p FAIL\n", s);
3131         return NULL;
3132 }
3133
3134 static void con_put(struct ceph_connection *con)
3135 {
3136         struct ceph_mds_session *s = con->private;
3137
3138         ceph_put_mds_session(s);
3139         dout("mdsc con_put %p (%d)\n", s, atomic_read(&s->s_ref));
3140 }
3141
3142 /*
3143  * if the client is unresponsive for long enough, the mds will kill
3144  * the session entirely.
3145  */
3146 static void peer_reset(struct ceph_connection *con)
3147 {
3148         struct ceph_mds_session *s = con->private;
3149         struct ceph_mds_client *mdsc = s->s_mdsc;
3150
3151         pr_warning("mds%d closed our session\n", s->s_mds);
3152         send_mds_reconnect(mdsc, s);
3153 }
3154
3155 static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
3156 {
3157         struct ceph_mds_session *s = con->private;
3158         struct ceph_mds_client *mdsc = s->s_mdsc;
3159         int type = le16_to_cpu(msg->hdr.type);
3160
3161         mutex_lock(&mdsc->mutex);
3162         if (__verify_registered_session(mdsc, s) < 0) {
3163                 mutex_unlock(&mdsc->mutex);
3164                 goto out;
3165         }
3166         mutex_unlock(&mdsc->mutex);
3167
3168         switch (type) {
3169         case CEPH_MSG_MDS_MAP:
3170                 ceph_mdsc_handle_map(mdsc, msg);
3171                 break;
3172         case CEPH_MSG_CLIENT_SESSION:
3173                 handle_session(s, msg);
3174                 break;
3175         case CEPH_MSG_CLIENT_REPLY:
3176                 handle_reply(s, msg);
3177                 break;
3178         case CEPH_MSG_CLIENT_REQUEST_FORWARD:
3179                 handle_forward(mdsc, s, msg);
3180                 break;
3181         case CEPH_MSG_CLIENT_CAPS:
3182                 ceph_handle_caps(s, msg);
3183                 break;
3184         case CEPH_MSG_CLIENT_SNAP:
3185                 ceph_handle_snap(mdsc, s, msg);
3186                 break;
3187         case CEPH_MSG_CLIENT_LEASE:
3188                 handle_lease(mdsc, s, msg);
3189                 break;
3190
3191         default:
3192                 pr_err("received unknown message type %d %s\n", type,
3193                        ceph_msg_type_name(type));
3194         }
3195 out:
3196         ceph_msg_put(msg);
3197 }
3198
3199 /*
3200  * authentication
3201  */
3202 static int get_authorizer(struct ceph_connection *con,
3203                           void **buf, int *len, int *proto,
3204                           void **reply_buf, int *reply_len, int force_new)
3205 {
3206         struct ceph_mds_session *s = con->private;
3207         struct ceph_mds_client *mdsc = s->s_mdsc;
3208         struct ceph_auth_client *ac = mdsc->client->monc.auth;
3209         int ret = 0;
3210
3211         if (force_new && s->s_authorizer) {
3212                 ac->ops->destroy_authorizer(ac, s->s_authorizer);
3213                 s->s_authorizer = NULL;
3214         }
3215         if (s->s_authorizer == NULL) {
3216                 if (ac->ops->create_authorizer) {
3217                         ret = ac->ops->create_authorizer(
3218                                 ac, CEPH_ENTITY_TYPE_MDS,
3219                                 &s->s_authorizer,
3220                                 &s->s_authorizer_buf,
3221                                 &s->s_authorizer_buf_len,
3222                                 &s->s_authorizer_reply_buf,
3223                                 &s->s_authorizer_reply_buf_len);
3224                         if (ret)
3225                                 return ret;
3226                 }
3227         }
3228
3229         *proto = ac->protocol;
3230         *buf = s->s_authorizer_buf;
3231         *len = s->s_authorizer_buf_len;
3232         *reply_buf = s->s_authorizer_reply_buf;
3233         *reply_len = s->s_authorizer_reply_buf_len;
3234         return 0;
3235 }
3236
3237
3238 static int verify_authorizer_reply(struct ceph_connection *con, int len)
3239 {
3240         struct ceph_mds_session *s = con->private;
3241         struct ceph_mds_client *mdsc = s->s_mdsc;
3242         struct ceph_auth_client *ac = mdsc->client->monc.auth;
3243
3244         return ac->ops->verify_authorizer_reply(ac, s->s_authorizer, len);
3245 }
3246
3247 static int invalidate_authorizer(struct ceph_connection *con)
3248 {
3249         struct ceph_mds_session *s = con->private;
3250         struct ceph_mds_client *mdsc = s->s_mdsc;
3251         struct ceph_auth_client *ac = mdsc->client->monc.auth;
3252
3253         if (ac->ops->invalidate_authorizer)
3254                 ac->ops->invalidate_authorizer(ac, CEPH_ENTITY_TYPE_MDS);
3255
3256         return ceph_monc_validate_auth(&mdsc->client->monc);
3257 }
3258
3259 static const struct ceph_connection_operations mds_con_ops = {
3260         .get = con_get,
3261         .put = con_put,
3262         .dispatch = dispatch,
3263         .get_authorizer = get_authorizer,
3264         .verify_authorizer_reply = verify_authorizer_reply,
3265         .invalidate_authorizer = invalidate_authorizer,
3266         .peer_reset = peer_reset,
3267 };
3268
3269
3270
3271
3272 /* eof */