net/ceph/mon_client.c

   1 #include <linux/ceph/ceph_debug.h>
   2
   3 #include <linux/module.h>
   4 #include <linux/types.h>
   5 #include <linux/slab.h>
   6 #include <linux/random.h>
   7 #include <linux/sched.h>
   8
   9 #include <linux/ceph/mon_client.h>
  10 #include <linux/ceph/libceph.h>
  11 #include <linux/ceph/debugfs.h>
  12 #include <linux/ceph/decode.h>
  13 #include <linux/ceph/auth.h>
  14
  15 /*
  16  * Interact with Ceph monitor cluster.  Handle requests for new map
  17  * versions, and periodically resend as needed.  Also implement
  18  * statfs() and umount().
  19  *
  20  * A small cluster of Ceph "monitors" are responsible for managing critical
  21  * cluster configuration and state information.  An odd number (e.g., 3, 5)
  22  * of cmon daemons use a modified version of the Paxos part-time parliament
  23  * algorithm to manage the MDS map (mds cluster membership), OSD map, and
  24  * list of clients who have mounted the file system.
  25  *
  26  * We maintain an open, active session with a monitor at all times in order to
  27  * receive timely MDSMap updates.  We periodically send a keepalive byte on the
  28  * TCP socket to ensure we detect a failure.  If the connection does break, we
  29  * randomly hunt for a new monitor.  Once the connection is reestablished, we
  30  * resend any outstanding requests.
  31  */
  32
  33 static const struct ceph_connection_operations mon_con_ops;
  34
  35 static int __validate_auth(struct ceph_mon_client *monc);
  36
  37 /*
  38  * Decode a monmap blob (e.g., during mount).
  39  */
  40 struct ceph_monmap *ceph_monmap_decode(void *p, void *end)
  41 {
  42         struct ceph_monmap *m = NULL;
  43         int i, err = -EINVAL;
  44         struct ceph_fsid fsid;
  45         u32 epoch, num_mon;
  46         u16 version;
  47         u32 len;
  48
  49         ceph_decode_32_safe(&p, end, len, bad);
  50         ceph_decode_need(&p, end, len, bad);
  51
  52         dout("monmap_decode %p %p len %d\n", p, end, (int)(end-p));
  53
  54         ceph_decode_16_safe(&p, end, version, bad);
  55
  56         ceph_decode_need(&p, end, sizeof(fsid) + 2*sizeof(u32), bad);
  57         ceph_decode_copy(&p, &fsid, sizeof(fsid));
  58         epoch = ceph_decode_32(&p);
  59
  60         num_mon = ceph_decode_32(&p);
  61         ceph_decode_need(&p, end, num_mon*sizeof(m->mon_inst[0]), bad);
  62
  63         if (num_mon >= CEPH_MAX_MON)
  64                 goto bad;
  65         m = kmalloc(sizeof(*m) + sizeof(m->mon_inst[0])*num_mon, GFP_NOFS);
  66         if (m == NULL)
  67                 return ERR_PTR(-ENOMEM);
  68         m->fsid = fsid;
  69         m->epoch = epoch;
  70         m->num_mon = num_mon;
  71         ceph_decode_copy(&p, m->mon_inst, num_mon*sizeof(m->mon_inst[0]));
  72         for (i = 0; i < num_mon; i++)
  73                 ceph_decode_addr(&m->mon_inst[i].addr);
  74
  75         dout("monmap_decode epoch %d, num_mon %d\n", m->epoch,
  76              m->num_mon);
  77         for (i = 0; i < m->num_mon; i++)
  78                 dout("monmap_decode  mon%d is %s\n", i,
  79                      ceph_pr_addr(&m->mon_inst[i].addr.in_addr));
  80         return m;
  81
  82 bad:
  83         dout("monmap_decode failed with %d\n", err);
  84         kfree(m);
  85         return ERR_PTR(err);
  86 }
  87
  88 /*
  89  * return true if *addr is included in the monmap.
  90  */
  91 int ceph_monmap_contains(struct ceph_monmap *m, struct ceph_entity_addr *addr)
  92 {
  93         int i;
  94
  95         for (i = 0; i < m->num_mon; i++)
  96                 if (memcmp(addr, &m->mon_inst[i].addr, sizeof(*addr)) == 0)
  97                         return 1;
  98         return 0;
  99 }
 100
 101 /*
 102  * Send an auth request.
 103  */
 104 static void __send_prepared_auth_request(struct ceph_mon_client *monc, int len)
 105 {
 106         monc->pending_auth = 1;
 107         monc->m_auth->front.iov_len = len;
 108         monc->m_auth->hdr.front_len = cpu_to_le32(len);
 109         ceph_msg_revoke(monc->m_auth);
 110         ceph_msg_get(monc->m_auth);  /* keep our ref */
 111         ceph_con_send(&monc->con, monc->m_auth);
 112 }
 113
 114 /*
 115  * Close monitor session, if any.
 116  */
 117 static void __close_session(struct ceph_mon_client *monc)
 118 {
 119         dout("__close_session closing mon%d\n", monc->cur_mon);
 120         ceph_msg_revoke(monc->m_auth);
 121         ceph_msg_revoke_incoming(monc->m_auth_reply);
 122         ceph_msg_revoke(monc->m_subscribe);
 123         ceph_msg_revoke_incoming(monc->m_subscribe_ack);
 124         ceph_con_close(&monc->con);
 125
 126         monc->pending_auth = 0;
 127         ceph_auth_reset(monc->auth);
 128 }
 129
 130 /*
 131  * Pick a new monitor at random and set cur_mon.  If we are repicking
 132  * (i.e. cur_mon is already set), be sure to pick a different one.
 133  */
 134 static void pick_new_mon(struct ceph_mon_client *monc)
 135 {
 136         int old_mon = monc->cur_mon;
 137
 138         BUG_ON(monc->monmap->num_mon < 1);
 139
 140         if (monc->monmap->num_mon == 1) {
 141                 monc->cur_mon = 0;
 142         } else {
 143                 int max = monc->monmap->num_mon;
 144                 int o = -1;
 145                 int n;
 146
 147                 if (monc->cur_mon >= 0) {
 148                         if (monc->cur_mon < monc->monmap->num_mon)
 149                                 o = monc->cur_mon;
 150                         if (o >= 0)
 151                                 max--;
 152                 }
 153
 154                 n = prandom_u32() % max;
 155                 if (o >= 0 && n >= o)
 156                         n++;
 157
 158                 monc->cur_mon = n;
 159         }
 160
 161         dout("%s mon%d -> mon%d out of %d mons\n", __func__, old_mon,
 162              monc->cur_mon, monc->monmap->num_mon);
 163 }
 164
 165 /*
 166  * Open a session with a new monitor.
 167  */
 168 static void __open_session(struct ceph_mon_client *monc)
 169 {
 170         int ret;
 171
 172         pick_new_mon(monc);
 173
 174         if (monc->had_a_connection) {
 175                 monc->hunt_mult *= CEPH_MONC_HUNT_BACKOFF;
 176                 if (monc->hunt_mult > CEPH_MONC_HUNT_MAX_MULT)
 177                         monc->hunt_mult = CEPH_MONC_HUNT_MAX_MULT;
 178         }
 179
 180         monc->sub_renew_after = jiffies; /* i.e., expired */
 181         monc->sub_renew_sent = 0;
 182
 183         dout("%s opening mon%d\n", __func__, monc->cur_mon);
 184         ceph_con_open(&monc->con, CEPH_ENTITY_TYPE_MON, monc->cur_mon,
 185                       &monc->monmap->mon_inst[monc->cur_mon].addr);
 186
 187         /*
 188          * send an initial keepalive to ensure our timestamp is valid
 189          * by the time we are in an OPENED state
 190          */
 191         ceph_con_keepalive(&monc->con);
 192
 193         /* initiate authentication handshake */
 194         ret = ceph_auth_build_hello(monc->auth,
 195                                     monc->m_auth->front.iov_base,
 196                                     monc->m_auth->front_alloc_len);
 197         BUG_ON(ret <= 0);
 198         __send_prepared_auth_request(monc, ret);
 199 }
 200
 201 /*
 202  * Reschedule delayed work timer.
 203  */
 204 static void __schedule_delayed(struct ceph_mon_client *monc)
 205 {
 206         unsigned long delay;
 207
 208         if (monc->hunting)
 209                 delay = CEPH_MONC_HUNT_INTERVAL * monc->hunt_mult;
 210         else
 211                 delay = CEPH_MONC_PING_INTERVAL;
 212
 213         dout("__schedule_delayed after %lu\n", delay);
 214         schedule_delayed_work(&monc->delayed_work,
 215                               round_jiffies_relative(delay));
 216 }
 217
 218 const char *ceph_sub_str[] = {
 219         [CEPH_SUB_MDSMAP] = "mdsmap",
 220         [CEPH_SUB_MONMAP] = "monmap",
 221         [CEPH_SUB_OSDMAP] = "osdmap",
 222 };
 223
 224 /*
 225  * Send subscribe request for one or more maps, according to
 226  * monc->subs.
 227  */
 228 static void __send_subscribe(struct ceph_mon_client *monc)
 229 {
 230         struct ceph_msg *msg = monc->m_subscribe;
 231         void *p = msg->front.iov_base;
 232         void *const end = p + msg->front_alloc_len;
 233         int num = 0;
 234         int i;
 235
 236         dout("%s sent %lu\n", __func__, monc->sub_renew_sent);
 237
 238         BUG_ON(monc->cur_mon < 0);
 239
 240         if (!monc->sub_renew_sent)
 241                 monc->sub_renew_sent = jiffies | 1; /* never 0 */
 242
 243         msg->hdr.version = cpu_to_le16(2);
 244
 245         for (i = 0; i < ARRAY_SIZE(monc->subs); i++) {
 246                 if (monc->subs[i].want)
 247                         num++;
 248         }
 249         BUG_ON(num < 1); /* monmap sub is always there */
 250         ceph_encode_32(&p, num);
 251         for (i = 0; i < ARRAY_SIZE(monc->subs); i++) {
 252                 const char *s = ceph_sub_str[i];
 253
 254                 if (!monc->subs[i].want)
 255                         continue;
 256
 257                 dout("%s %s start %llu flags 0x%x\n", __func__, s,
 258                      le64_to_cpu(monc->subs[i].item.start),
 259                      monc->subs[i].item.flags);
 260                 ceph_encode_string(&p, end, s, strlen(s));
 261                 memcpy(p, &monc->subs[i].item, sizeof(monc->subs[i].item));
 262                 p += sizeof(monc->subs[i].item);
 263         }
 264
 265         BUG_ON(p != (end - 35 - (ARRAY_SIZE(monc->subs) - num) * 19));
 266         msg->front.iov_len = p - msg->front.iov_base;
 267         msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
 268         ceph_msg_revoke(msg);
 269         ceph_con_send(&monc->con, ceph_msg_get(msg));
 270 }
 271
 272 static void handle_subscribe_ack(struct ceph_mon_client *monc,
 273                                  struct ceph_msg *msg)
 274 {
 275         unsigned int seconds;
 276         struct ceph_mon_subscribe_ack *h = msg->front.iov_base;
 277
 278         if (msg->front.iov_len < sizeof(*h))
 279                 goto bad;
 280         seconds = le32_to_cpu(h->duration);
 281
 282         mutex_lock(&monc->mutex);
 283         if (monc->sub_renew_sent) {
 284                 monc->sub_renew_after = monc->sub_renew_sent +
 285                                             (seconds >> 1) * HZ - 1;
 286                 dout("%s sent %lu duration %d renew after %lu\n", __func__,
 287                      monc->sub_renew_sent, seconds, monc->sub_renew_after);
 288                 monc->sub_renew_sent = 0;
 289         } else {
 290                 dout("%s sent %lu renew after %lu, ignoring\n", __func__,
 291                      monc->sub_renew_sent, monc->sub_renew_after);
 292         }
 293         mutex_unlock(&monc->mutex);
 294         return;
 295 bad:
 296         pr_err("got corrupt subscribe-ack msg\n");
 297         ceph_msg_dump(msg);
 298 }
 299
 300 /*
 301  * Register interest in a map
 302  *
 303  * @sub: one of CEPH_SUB_*
 304  * @epoch: X for "every map since X", or 0 for "just the latest"
 305  */
 306 static bool __ceph_monc_want_map(struct ceph_mon_client *monc, int sub,
 307                                  u32 epoch, bool continuous)
 308 {
 309         __le64 start = cpu_to_le64(epoch);
 310         u8 flags = !continuous ? CEPH_SUBSCRIBE_ONETIME : 0;
 311
 312         dout("%s %s epoch %u continuous %d\n", __func__, ceph_sub_str[sub],
 313              epoch, continuous);
 314
 315         if (monc->subs[sub].want &&
 316             monc->subs[sub].item.start == start &&
 317             monc->subs[sub].item.flags == flags)
 318                 return false;
 319
 320         monc->subs[sub].item.start = start;
 321         monc->subs[sub].item.flags = flags;
 322         monc->subs[sub].want = true;
 323
 324         return true;
 325 }
 326
 327 bool ceph_monc_want_map(struct ceph_mon_client *monc, int sub, u32 epoch,
 328                         bool continuous)
 329 {
 330         bool need_request;
 331
 332         mutex_lock(&monc->mutex);
 333         need_request = __ceph_monc_want_map(monc, sub, epoch, continuous);
 334         mutex_unlock(&monc->mutex);
 335
 336         return need_request;
 337 }
 338 EXPORT_SYMBOL(ceph_monc_want_map);
 339
 340 /*
 341  * Keep track of which maps we have
 342  *
 343  * @sub: one of CEPH_SUB_*
 344  */
 345 static void __ceph_monc_got_map(struct ceph_mon_client *monc, int sub,
 346                                 u32 epoch)
 347 {
 348         dout("%s %s epoch %u\n", __func__, ceph_sub_str[sub], epoch);
 349
 350         if (monc->subs[sub].want) {
 351                 if (monc->subs[sub].item.flags & CEPH_SUBSCRIBE_ONETIME)
 352                         monc->subs[sub].want = false;
 353                 else
 354                         monc->subs[sub].item.start = cpu_to_le64(epoch + 1);
 355         }
 356
 357         monc->subs[sub].have = epoch;
 358 }
 359
 360 void ceph_monc_got_map(struct ceph_mon_client *monc, int sub, u32 epoch)
 361 {
 362         mutex_lock(&monc->mutex);
 363         __ceph_monc_got_map(monc, sub, epoch);
 364         mutex_unlock(&monc->mutex);
 365 }
 366 EXPORT_SYMBOL(ceph_monc_got_map);
 367
 368 /*
 369  * Register interest in the next osdmap
 370  */
 371 void ceph_monc_request_next_osdmap(struct ceph_mon_client *monc)
 372 {
 373         dout("%s have %u\n", __func__, monc->subs[CEPH_SUB_OSDMAP].have);
 374         mutex_lock(&monc->mutex);
 375         if (__ceph_monc_want_map(monc, CEPH_SUB_OSDMAP,
 376                                  monc->subs[CEPH_SUB_OSDMAP].have + 1, false))
 377                 __send_subscribe(monc);
 378         mutex_unlock(&monc->mutex);
 379 }
 380 EXPORT_SYMBOL(ceph_monc_request_next_osdmap);
 381
 382 /*
 383  * Wait for an osdmap with a given epoch.
 384  *
 385  * @epoch: epoch to wait for
 386  * @timeout: in jiffies, 0 means "wait forever"
 387  */
 388 int ceph_monc_wait_osdmap(struct ceph_mon_client *monc, u32 epoch,
 389                           unsigned long timeout)
 390 {
 391         unsigned long started = jiffies;
 392         long ret;
 393
 394         mutex_lock(&monc->mutex);
 395         while (monc->subs[CEPH_SUB_OSDMAP].have < epoch) {
 396                 mutex_unlock(&monc->mutex);
 397
 398                 if (timeout && time_after_eq(jiffies, started + timeout))
 399                         return -ETIMEDOUT;
 400
 401                 ret = wait_event_interruptible_timeout(monc->client->auth_wq,
 402                                      monc->subs[CEPH_SUB_OSDMAP].have >= epoch,
 403                                      ceph_timeout_jiffies(timeout));
 404                 if (ret < 0)
 405                         return ret;
 406
 407                 mutex_lock(&monc->mutex);
 408         }
 409
 410         mutex_unlock(&monc->mutex);
 411         return 0;
 412 }
 413 EXPORT_SYMBOL(ceph_monc_wait_osdmap);
 414
 415 /*
 416  * Open a session with a random monitor.  Request monmap and osdmap,
 417  * which are waited upon in __ceph_open_session().
 418  */
 419 int ceph_monc_open_session(struct ceph_mon_client *monc)
 420 {
 421         mutex_lock(&monc->mutex);
 422         __ceph_monc_want_map(monc, CEPH_SUB_MONMAP, 0, true);
 423         __ceph_monc_want_map(monc, CEPH_SUB_OSDMAP, 0, false);
 424         __open_session(monc);
 425         __schedule_delayed(monc);
 426         mutex_unlock(&monc->mutex);
 427         return 0;
 428 }
 429 EXPORT_SYMBOL(ceph_monc_open_session);
 430
 431 static void ceph_monc_handle_map(struct ceph_mon_client *monc,
 432                                  struct ceph_msg *msg)
 433 {
 434         struct ceph_client *client = monc->client;
 435         struct ceph_monmap *monmap = NULL, *old = monc->monmap;
 436         void *p, *end;
 437
 438         mutex_lock(&monc->mutex);
 439
 440         dout("handle_monmap\n");
 441         p = msg->front.iov_base;
 442         end = p + msg->front.iov_len;
 443
 444         monmap = ceph_monmap_decode(p, end);
 445         if (IS_ERR(monmap)) {
 446                 pr_err("problem decoding monmap, %d\n",
 447                        (int)PTR_ERR(monmap));
 448                 goto out;
 449         }
 450
 451         if (ceph_check_fsid(monc->client, &monmap->fsid) < 0) {
 452                 kfree(monmap);
 453                 goto out;
 454         }
 455
 456         client->monc.monmap = monmap;
 457         kfree(old);
 458
 459         __ceph_monc_got_map(monc, CEPH_SUB_MONMAP, monc->monmap->epoch);
 460         client->have_fsid = true;
 461
 462 out:
 463         mutex_unlock(&monc->mutex);
 464         wake_up_all(&client->auth_wq);
 465 }
 466
 467 /*
 468  * generic requests (currently statfs, mon_get_version)
 469  */
 470 static struct ceph_mon_generic_request *__lookup_generic_req(
 471         struct ceph_mon_client *monc, u64 tid)
 472 {
 473         struct ceph_mon_generic_request *req;
 474         struct rb_node *n = monc->generic_request_tree.rb_node;
 475
 476         while (n) {
 477                 req = rb_entry(n, struct ceph_mon_generic_request, node);
 478                 if (tid < req->tid)
 479                         n = n->rb_left;
 480                 else if (tid > req->tid)
 481                         n = n->rb_right;
 482                 else
 483                         return req;
 484         }
 485         return NULL;
 486 }
 487
 488 static void __insert_generic_request(struct ceph_mon_client *monc,
 489                             struct ceph_mon_generic_request *new)
 490 {
 491         struct rb_node **p = &monc->generic_request_tree.rb_node;
 492         struct rb_node *parent = NULL;
 493         struct ceph_mon_generic_request *req = NULL;
 494
 495         while (*p) {
 496                 parent = *p;
 497                 req = rb_entry(parent, struct ceph_mon_generic_request, node);
 498                 if (new->tid < req->tid)
 499                         p = &(*p)->rb_left;
 500                 else if (new->tid > req->tid)
 501                         p = &(*p)->rb_right;
 502                 else
 503                         BUG();
 504         }
 505
 506         rb_link_node(&new->node, parent, p);
 507         rb_insert_color(&new->node, &monc->generic_request_tree);
 508 }
 509
 510 static void release_generic_request(struct kref *kref)
 511 {
 512         struct ceph_mon_generic_request *req =
 513                 container_of(kref, struct ceph_mon_generic_request, kref);
 514
 515         if (req->reply)
 516                 ceph_msg_put(req->reply);
 517         if (req->request)
 518                 ceph_msg_put(req->request);
 519
 520         kfree(req);
 521 }
 522
 523 static void put_generic_request(struct ceph_mon_generic_request *req)
 524 {
 525         kref_put(&req->kref, release_generic_request);
 526 }
 527
 528 static void get_generic_request(struct ceph_mon_generic_request *req)
 529 {
 530         kref_get(&req->kref);
 531 }
 532
 533 static struct ceph_msg *get_generic_reply(struct ceph_connection *con,
 534                                          struct ceph_msg_header *hdr,
 535                                          int *skip)
 536 {
 537         struct ceph_mon_client *monc = con->private;
 538         struct ceph_mon_generic_request *req;
 539         u64 tid = le64_to_cpu(hdr->tid);
 540         struct ceph_msg *m;
 541
 542         mutex_lock(&monc->mutex);
 543         req = __lookup_generic_req(monc, tid);
 544         if (!req) {
 545                 dout("get_generic_reply %lld dne\n", tid);
 546                 *skip = 1;
 547                 m = NULL;
 548         } else {
 549                 dout("get_generic_reply %lld got %p\n", tid, req->reply);
 550                 *skip = 0;
 551                 m = ceph_msg_get(req->reply);
 552                 /*
 553                  * we don't need to track the connection reading into
 554                  * this reply because we only have one open connection
 555                  * at a time, ever.
 556                  */
 557         }
 558         mutex_unlock(&monc->mutex);
 559         return m;
 560 }
 561
 562 static int __do_generic_request(struct ceph_mon_client *monc, u64 tid,
 563                                 struct ceph_mon_generic_request *req)
 564 {
 565         int err;
 566
 567         /* register request */
 568         req->tid = tid != 0 ? tid : ++monc->last_tid;
 569         req->request->hdr.tid = cpu_to_le64(req->tid);
 570         __insert_generic_request(monc, req);
 571         monc->num_generic_requests++;
 572         ceph_con_send(&monc->con, ceph_msg_get(req->request));
 573         mutex_unlock(&monc->mutex);
 574
 575         err = wait_for_completion_interruptible(&req->completion);
 576
 577         mutex_lock(&monc->mutex);
 578         rb_erase(&req->node, &monc->generic_request_tree);
 579         monc->num_generic_requests--;
 580
 581         if (!err)
 582                 err = req->result;
 583         return err;
 584 }
 585
 586 static int do_generic_request(struct ceph_mon_client *monc,
 587                               struct ceph_mon_generic_request *req)
 588 {
 589         int err;
 590
 591         mutex_lock(&monc->mutex);
 592         err = __do_generic_request(monc, 0, req);
 593         mutex_unlock(&monc->mutex);
 594
 595         return err;
 596 }
 597
 598 /*
 599  * statfs
 600  */
 601 static void handle_statfs_reply(struct ceph_mon_client *monc,
 602                                 struct ceph_msg *msg)
 603 {
 604         struct ceph_mon_generic_request *req;
 605         struct ceph_mon_statfs_reply *reply = msg->front.iov_base;
 606         u64 tid = le64_to_cpu(msg->hdr.tid);
 607
 608         if (msg->front.iov_len != sizeof(*reply))
 609                 goto bad;
 610         dout("handle_statfs_reply %p tid %llu\n", msg, tid);
 611
 612         mutex_lock(&monc->mutex);
 613         req = __lookup_generic_req(monc, tid);
 614         if (req) {
 615                 *(struct ceph_statfs *)req->buf = reply->st;
 616                 req->result = 0;
 617                 get_generic_request(req);
 618         }
 619         mutex_unlock(&monc->mutex);
 620         if (req) {
 621                 complete_all(&req->completion);
 622                 put_generic_request(req);
 623         }
 624         return;
 625
 626 bad:
 627         pr_err("corrupt statfs reply, tid %llu\n", tid);
 628         ceph_msg_dump(msg);
 629 }
 630
 631 /*
 632  * Do a synchronous statfs().
 633  */
 634 int ceph_monc_do_statfs(struct ceph_mon_client *monc, struct ceph_statfs *buf)
 635 {
 636         struct ceph_mon_generic_request *req;
 637         struct ceph_mon_statfs *h;
 638         int err;
 639
 640         req = kzalloc(sizeof(*req), GFP_NOFS);
 641         if (!req)
 642                 return -ENOMEM;
 643
 644         kref_init(&req->kref);
 645         req->buf = buf;
 646         init_completion(&req->completion);
 647
 648         err = -ENOMEM;
 649         req->request = ceph_msg_new(CEPH_MSG_STATFS, sizeof(*h), GFP_NOFS,
 650                                     true);
 651         if (!req->request)
 652                 goto out;
 653         req->reply = ceph_msg_new(CEPH_MSG_STATFS_REPLY, 1024, GFP_NOFS,
 654                                   true);
 655         if (!req->reply)
 656                 goto out;
 657
 658         /* fill out request */
 659         h = req->request->front.iov_base;
 660         h->monhdr.have_version = 0;
 661         h->monhdr.session_mon = cpu_to_le16(-1);
 662         h->monhdr.session_mon_tid = 0;
 663         h->fsid = monc->monmap->fsid;
 664
 665         err = do_generic_request(monc, req);
 666
 667 out:
 668         put_generic_request(req);
 669         return err;
 670 }
 671 EXPORT_SYMBOL(ceph_monc_do_statfs);
 672
 673 static void handle_get_version_reply(struct ceph_mon_client *monc,
 674                                      struct ceph_msg *msg)
 675 {
 676         struct ceph_mon_generic_request *req;
 677         u64 tid = le64_to_cpu(msg->hdr.tid);
 678         void *p = msg->front.iov_base;
 679         void *end = p + msg->front_alloc_len;
 680         u64 handle;
 681
 682         dout("%s %p tid %llu\n", __func__, msg, tid);
 683
 684         ceph_decode_need(&p, end, 2*sizeof(u64), bad);
 685         handle = ceph_decode_64(&p);
 686         if (tid != 0 && tid != handle)
 687                 goto bad;
 688
 689         mutex_lock(&monc->mutex);
 690         req = __lookup_generic_req(monc, handle);
 691         if (req) {
 692                 *(u64 *)req->buf = ceph_decode_64(&p);
 693                 req->result = 0;
 694                 get_generic_request(req);
 695         }
 696         mutex_unlock(&monc->mutex);
 697         if (req) {
 698                 complete_all(&req->completion);
 699                 put_generic_request(req);
 700         }
 701
 702         return;
 703 bad:
 704         pr_err("corrupt mon_get_version reply, tid %llu\n", tid);
 705         ceph_msg_dump(msg);
 706 }
 707
 708 /*
 709  * Send MMonGetVersion and wait for the reply.
 710  *
 711  * @what: one of "mdsmap", "osdmap" or "monmap"
 712  */
 713 int ceph_monc_do_get_version(struct ceph_mon_client *monc, const char *what,
 714                              u64 *newest)
 715 {
 716         struct ceph_mon_generic_request *req;
 717         void *p, *end;
 718         u64 tid;
 719         int err;
 720
 721         req = kzalloc(sizeof(*req), GFP_NOFS);
 722         if (!req)
 723                 return -ENOMEM;
 724
 725         kref_init(&req->kref);
 726         req->buf = newest;
 727         init_completion(&req->completion);
 728
 729         req->request = ceph_msg_new(CEPH_MSG_MON_GET_VERSION,
 730                                     sizeof(u64) + sizeof(u32) + strlen(what),
 731                                     GFP_NOFS, true);
 732         if (!req->request) {
 733                 err = -ENOMEM;
 734                 goto out;
 735         }
 736
 737         req->reply = ceph_msg_new(CEPH_MSG_MON_GET_VERSION_REPLY, 1024,
 738                                   GFP_NOFS, true);
 739         if (!req->reply) {
 740                 err = -ENOMEM;
 741                 goto out;
 742         }
 743
 744         p = req->request->front.iov_base;
 745         end = p + req->request->front_alloc_len;
 746
 747         /* fill out request */
 748         mutex_lock(&monc->mutex);
 749         tid = ++monc->last_tid;
 750         ceph_encode_64(&p, tid); /* handle */
 751         ceph_encode_string(&p, end, what, strlen(what));
 752
 753         err = __do_generic_request(monc, tid, req);
 754
 755         mutex_unlock(&monc->mutex);
 756 out:
 757         put_generic_request(req);
 758         return err;
 759 }
 760 EXPORT_SYMBOL(ceph_monc_do_get_version);
 761
 762 /*
 763  * Resend pending generic requests.
 764  */
 765 static void __resend_generic_request(struct ceph_mon_client *monc)
 766 {
 767         struct ceph_mon_generic_request *req;
 768         struct rb_node *p;
 769
 770         for (p = rb_first(&monc->generic_request_tree); p; p = rb_next(p)) {
 771                 req = rb_entry(p, struct ceph_mon_generic_request, node);
 772                 ceph_msg_revoke(req->request);
 773                 ceph_msg_revoke_incoming(req->reply);
 774                 ceph_con_send(&monc->con, ceph_msg_get(req->request));
 775         }
 776 }
 777
 778 /*
 779  * Delayed work.  If we haven't mounted yet, retry.  Otherwise,
 780  * renew/retry subscription as needed (in case it is timing out, or we
 781  * got an ENOMEM).  And keep the monitor connection alive.
 782  */
 783 static void delayed_work(struct work_struct *work)
 784 {
 785         struct ceph_mon_client *monc =
 786                 container_of(work, struct ceph_mon_client, delayed_work.work);
 787
 788         dout("monc delayed_work\n");
 789         mutex_lock(&monc->mutex);
 790         if (monc->hunting) {
 791                 __close_session(monc);
 792                 __open_session(monc);  /* continue hunting */
 793         } else {
 794                 int is_auth = ceph_auth_is_authenticated(monc->auth);
 795                 if (ceph_con_keepalive_expired(&monc->con,
 796                                                CEPH_MONC_PING_TIMEOUT)) {
 797                         dout("monc keepalive timeout\n");
 798                         is_auth = 0;
 799                         __close_session(monc);
 800                         monc->hunting = true;
 801                         __open_session(monc);
 802                 }
 803
 804                 if (!monc->hunting) {
 805                         ceph_con_keepalive(&monc->con);
 806                         __validate_auth(monc);
 807                 }
 808
 809                 if (is_auth) {
 810                         unsigned long now = jiffies;
 811
 812                         dout("%s renew subs? now %lu renew after %lu\n",
 813                              __func__, now, monc->sub_renew_after);
 814                         if (time_after_eq(now, monc->sub_renew_after))
 815                                 __send_subscribe(monc);
 816                 }
 817         }
 818         __schedule_delayed(monc);
 819         mutex_unlock(&monc->mutex);
 820 }
 821
 822 /*
 823  * On startup, we build a temporary monmap populated with the IPs
 824  * provided by mount(2).
 825  */
 826 static int build_initial_monmap(struct ceph_mon_client *monc)
 827 {
 828         struct ceph_options *opt = monc->client->options;
 829         struct ceph_entity_addr *mon_addr = opt->mon_addr;
 830         int num_mon = opt->num_mon;
 831         int i;
 832
 833         /* build initial monmap */
 834         monc->monmap = kzalloc(sizeof(*monc->monmap) +
 835                                num_mon*sizeof(monc->monmap->mon_inst[0]),
 836                                GFP_KERNEL);
 837         if (!monc->monmap)
 838                 return -ENOMEM;
 839         for (i = 0; i < num_mon; i++) {
 840                 monc->monmap->mon_inst[i].addr = mon_addr[i];
 841                 monc->monmap->mon_inst[i].addr.nonce = 0;
 842                 monc->monmap->mon_inst[i].name.type =
 843                         CEPH_ENTITY_TYPE_MON;
 844                 monc->monmap->mon_inst[i].name.num = cpu_to_le64(i);
 845         }
 846         monc->monmap->num_mon = num_mon;
 847         return 0;
 848 }
 849
 850 int ceph_monc_init(struct ceph_mon_client *monc, struct ceph_client *cl)
 851 {
 852         int err = 0;
 853
 854         dout("init\n");
 855         memset(monc, 0, sizeof(*monc));
 856         monc->client = cl;
 857         monc->monmap = NULL;
 858         mutex_init(&monc->mutex);
 859
 860         err = build_initial_monmap(monc);
 861         if (err)
 862                 goto out;
 863
 864         /* connection */
 865         /* authentication */
 866         monc->auth = ceph_auth_init(cl->options->name,
 867                                     cl->options->key);
 868         if (IS_ERR(monc->auth)) {
 869                 err = PTR_ERR(monc->auth);
 870                 goto out_monmap;
 871         }
 872         monc->auth->want_keys =
 873                 CEPH_ENTITY_TYPE_AUTH | CEPH_ENTITY_TYPE_MON |
 874                 CEPH_ENTITY_TYPE_OSD | CEPH_ENTITY_TYPE_MDS;
 875
 876         /* msgs */
 877         err = -ENOMEM;
 878         monc->m_subscribe_ack = ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE_ACK,
 879                                      sizeof(struct ceph_mon_subscribe_ack),
 880                                      GFP_NOFS, true);
 881         if (!monc->m_subscribe_ack)
 882                 goto out_auth;
 883
 884         monc->m_subscribe = ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE, 96, GFP_NOFS,
 885                                          true);
 886         if (!monc->m_subscribe)
 887                 goto out_subscribe_ack;
 888
 889         monc->m_auth_reply = ceph_msg_new(CEPH_MSG_AUTH_REPLY, 4096, GFP_NOFS,
 890                                           true);
 891         if (!monc->m_auth_reply)
 892                 goto out_subscribe;
 893
 894         monc->m_auth = ceph_msg_new(CEPH_MSG_AUTH, 4096, GFP_NOFS, true);
 895         monc->pending_auth = 0;
 896         if (!monc->m_auth)
 897                 goto out_auth_reply;
 898
 899         ceph_con_init(&monc->con, monc, &mon_con_ops,
 900                       &monc->client->msgr);
 901
 902         monc->cur_mon = -1;
 903         monc->hunting = true;
 904         monc->sub_renew_after = jiffies;
 905         monc->sub_renew_sent = 0;
 906         monc->had_a_connection = false;
 907         monc->hunt_mult = 1;
 908
 909         INIT_DELAYED_WORK(&monc->delayed_work, delayed_work);
 910         monc->generic_request_tree = RB_ROOT;
 911         monc->num_generic_requests = 0;
 912         monc->last_tid = 0;
 913
 914         return 0;
 915
 916 out_auth_reply:
 917         ceph_msg_put(monc->m_auth_reply);
 918 out_subscribe:
 919         ceph_msg_put(monc->m_subscribe);
 920 out_subscribe_ack:
 921         ceph_msg_put(monc->m_subscribe_ack);
 922 out_auth:
 923         ceph_auth_destroy(monc->auth);
 924 out_monmap:
 925         kfree(monc->monmap);
 926 out:
 927         return err;
 928 }
 929 EXPORT_SYMBOL(ceph_monc_init);
 930
 931 void ceph_monc_stop(struct ceph_mon_client *monc)
 932 {
 933         dout("stop\n");
 934         cancel_delayed_work_sync(&monc->delayed_work);
 935
 936         mutex_lock(&monc->mutex);
 937         __close_session(monc);
 938         monc->cur_mon = -1;
 939         mutex_unlock(&monc->mutex);
 940
 941         /*
 942          * flush msgr queue before we destroy ourselves to ensure that:
 943          *  - any work that references our embedded con is finished.
 944          *  - any osd_client or other work that may reference an authorizer
 945          *    finishes before we shut down the auth subsystem.
 946          */
 947         ceph_msgr_flush();
 948
 949         ceph_auth_destroy(monc->auth);
 950
 951         ceph_msg_put(monc->m_auth);
 952         ceph_msg_put(monc->m_auth_reply);
 953         ceph_msg_put(monc->m_subscribe);
 954         ceph_msg_put(monc->m_subscribe_ack);
 955
 956         kfree(monc->monmap);
 957 }
 958 EXPORT_SYMBOL(ceph_monc_stop);
 959
 960 static void finish_hunting(struct ceph_mon_client *monc)
 961 {
 962         if (monc->hunting) {
 963                 dout("%s found mon%d\n", __func__, monc->cur_mon);
 964                 monc->hunting = false;
 965                 monc->had_a_connection = true;
 966                 monc->hunt_mult /= 2; /* reduce by 50% */
 967                 if (monc->hunt_mult < 1)
 968                         monc->hunt_mult = 1;
 969         }
 970 }
 971
 972 static void handle_auth_reply(struct ceph_mon_client *monc,
 973                               struct ceph_msg *msg)
 974 {
 975         int ret;
 976         int was_auth = 0;
 977
 978         mutex_lock(&monc->mutex);
 979         was_auth = ceph_auth_is_authenticated(monc->auth);
 980         monc->pending_auth = 0;
 981         ret = ceph_handle_auth_reply(monc->auth, msg->front.iov_base,
 982                                      msg->front.iov_len,
 983                                      monc->m_auth->front.iov_base,
 984                                      monc->m_auth->front_alloc_len);
 985         if (ret > 0) {
 986                 __send_prepared_auth_request(monc, ret);
 987                 goto out;
 988         }
 989
 990         finish_hunting(monc);
 991
 992         if (ret < 0) {
 993                 monc->client->auth_err = ret;
 994         } else if (!was_auth && ceph_auth_is_authenticated(monc->auth)) {
 995                 dout("authenticated, starting session\n");
 996
 997                 monc->client->msgr.inst.name.type = CEPH_ENTITY_TYPE_CLIENT;
 998                 monc->client->msgr.inst.name.num =
 999                                         cpu_to_le64(monc->auth->global_id);
1000
1001                 __send_subscribe(monc);
1002                 __resend_generic_request(monc);
1003
1004                 pr_info("mon%d %s session established\n", monc->cur_mon,
1005                         ceph_pr_addr(&monc->con.peer_addr.in_addr));
1006         }
1007
1008 out:
1009         mutex_unlock(&monc->mutex);
1010         if (monc->client->auth_err < 0)
1011                 wake_up_all(&monc->client->auth_wq);
1012 }
1013
1014 static int __validate_auth(struct ceph_mon_client *monc)
1015 {
1016         int ret;
1017
1018         if (monc->pending_auth)
1019                 return 0;
1020
1021         ret = ceph_build_auth(monc->auth, monc->m_auth->front.iov_base,
1022                               monc->m_auth->front_alloc_len);
1023         if (ret <= 0)
1024                 return ret; /* either an error, or no need to authenticate */
1025         __send_prepared_auth_request(monc, ret);
1026         return 0;
1027 }
1028
1029 int ceph_monc_validate_auth(struct ceph_mon_client *monc)
1030 {
1031         int ret;
1032
1033         mutex_lock(&monc->mutex);
1034         ret = __validate_auth(monc);
1035         mutex_unlock(&monc->mutex);
1036         return ret;
1037 }
1038 EXPORT_SYMBOL(ceph_monc_validate_auth);
1039
1040 /*
1041  * handle incoming message
1042  */
1043 static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
1044 {
1045         struct ceph_mon_client *monc = con->private;
1046         int type = le16_to_cpu(msg->hdr.type);
1047
1048         if (!monc)
1049                 return;
1050
1051         switch (type) {
1052         case CEPH_MSG_AUTH_REPLY:
1053                 handle_auth_reply(monc, msg);
1054                 break;
1055
1056         case CEPH_MSG_MON_SUBSCRIBE_ACK:
1057                 handle_subscribe_ack(monc, msg);
1058                 break;
1059
1060         case CEPH_MSG_STATFS_REPLY:
1061                 handle_statfs_reply(monc, msg);
1062                 break;
1063
1064         case CEPH_MSG_MON_GET_VERSION_REPLY:
1065                 handle_get_version_reply(monc, msg);
1066                 break;
1067
1068         case CEPH_MSG_MON_MAP:
1069                 ceph_monc_handle_map(monc, msg);
1070                 break;
1071
1072         case CEPH_MSG_OSD_MAP:
1073                 ceph_osdc_handle_map(&monc->client->osdc, msg);
1074                 break;
1075
1076         default:
1077                 /* can the chained handler handle it? */
1078                 if (monc->client->extra_mon_dispatch &&
1079                     monc->client->extra_mon_dispatch(monc->client, msg) == 0)
1080                         break;
1081
1082                 pr_err("received unknown message type %d %s\n", type,
1083                        ceph_msg_type_name(type));
1084         }
1085         ceph_msg_put(msg);
1086 }
1087
1088 /*
1089  * Allocate memory for incoming message
1090  */
1091 static struct ceph_msg *mon_alloc_msg(struct ceph_connection *con,
1092                                       struct ceph_msg_header *hdr,
1093                                       int *skip)
1094 {
1095         struct ceph_mon_client *monc = con->private;
1096         int type = le16_to_cpu(hdr->type);
1097         int front_len = le32_to_cpu(hdr->front_len);
1098         struct ceph_msg *m = NULL;
1099
1100         *skip = 0;
1101
1102         switch (type) {
1103         case CEPH_MSG_MON_SUBSCRIBE_ACK:
1104                 m = ceph_msg_get(monc->m_subscribe_ack);
1105                 break;
1106         case CEPH_MSG_STATFS_REPLY:
1107                 return get_generic_reply(con, hdr, skip);
1108         case CEPH_MSG_AUTH_REPLY:
1109                 m = ceph_msg_get(monc->m_auth_reply);
1110                 break;
1111         case CEPH_MSG_MON_GET_VERSION_REPLY:
1112                 if (le64_to_cpu(hdr->tid) != 0)
1113                         return get_generic_reply(con, hdr, skip);
1114
1115                 /*
1116                  * Older OSDs don't set reply tid even if the orignal
1117                  * request had a non-zero tid.  Workaround this weirdness
1118                  * by falling through to the allocate case.
1119                  */
1120         case CEPH_MSG_MON_MAP:
1121         case CEPH_MSG_MDS_MAP:
1122         case CEPH_MSG_OSD_MAP:
1123                 m = ceph_msg_new(type, front_len, GFP_NOFS, false);
1124                 if (!m)
1125                         return NULL;    /* ENOMEM--return skip == 0 */
1126                 break;
1127         }
1128
1129         if (!m) {
1130                 pr_info("alloc_msg unknown type %d\n", type);
1131                 *skip = 1;
1132         } else if (front_len > m->front_alloc_len) {
1133                 pr_warn("mon_alloc_msg front %d > prealloc %d (%u#%llu)\n",
1134                         front_len, m->front_alloc_len,
1135                         (unsigned int)con->peer_name.type,
1136                         le64_to_cpu(con->peer_name.num));
1137                 ceph_msg_put(m);
1138                 m = ceph_msg_new(type, front_len, GFP_NOFS, false);
1139         }
1140
1141         return m;
1142 }
1143
1144 /*
1145  * If the monitor connection resets, pick a new monitor and resubmit
1146  * any pending requests.
1147  */
1148 static void mon_fault(struct ceph_connection *con)
1149 {
1150         struct ceph_mon_client *monc = con->private;
1151
1152         if (!monc)
1153                 return;
1154
1155         dout("mon_fault\n");
1156         mutex_lock(&monc->mutex);
1157         if (!con->private)
1158                 goto out;
1159
1160         if (!monc->hunting)
1161                 pr_info("mon%d %s session lost, "
1162                         "hunting for new mon\n", monc->cur_mon,
1163                         ceph_pr_addr(&monc->con.peer_addr.in_addr));
1164
1165         __close_session(monc);
1166         if (!monc->hunting) {
1167                 /* start hunting */
1168                 monc->hunting = true;
1169                 __open_session(monc);
1170         } else {
1171                 /* already hunting, let's wait a bit */
1172                 __schedule_delayed(monc);
1173         }
1174 out:
1175         mutex_unlock(&monc->mutex);
1176 }
1177
1178 /*
1179  * We can ignore refcounting on the connection struct, as all references
1180  * will come from the messenger workqueue, which is drained prior to
1181  * mon_client destruction.
1182  */
1183 static struct ceph_connection *con_get(struct ceph_connection *con)
1184 {
1185         return con;
1186 }
1187
1188 static void con_put(struct ceph_connection *con)
1189 {
1190 }
1191
1192 static const struct ceph_connection_operations mon_con_ops = {
1193         .get = con_get,
1194         .put = con_put,
1195         .dispatch = dispatch,
1196         .fault = mon_fault,
1197         .alloc_msg = mon_alloc_msg,
1198 };