net/ceph/mon_client.c

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