fs/ceph/mon_client.c

   1 #include "ceph_debug.h"
   2
   3 #include <linux/types.h>
   4 #include <linux/random.h>
   5 #include <linux/sched.h>
   6
   7 #include "mon_client.h"
   8 #include "super.h"
   9 #include "auth.h"
  10 #include "decode.h"
  11
  12 /*
  13  * Interact with Ceph monitor cluster.  Handle requests for new map
  14  * versions, and periodically resend as needed.  Also implement
  15  * statfs() and umount().
  16  *
  17  * A small cluster of Ceph "monitors" are responsible for managing critical
  18  * cluster configuration and state information.  An odd number (e.g., 3, 5)
  19  * of cmon daemons use a modified version of the Paxos part-time parliament
  20  * algorithm to manage the MDS map (mds cluster membership), OSD map, and
  21  * list of clients who have mounted the file system.
  22  *
  23  * We maintain an open, active session with a monitor at all times in order to
  24  * receive timely MDSMap updates.  We periodically send a keepalive byte on the
  25  * TCP socket to ensure we detect a failure.  If the connection does break, we
  26  * randomly hunt for a new monitor.  Once the connection is reestablished, we
  27  * resend any outstanding requests.
  28  */
  29
  30 const static struct ceph_connection_operations mon_con_ops;
  31
  32 static int __validate_auth(struct ceph_mon_client *monc);
  33
  34 /*
  35  * Decode a monmap blob (e.g., during mount).
  36  */
  37 struct ceph_monmap *ceph_monmap_decode(void *p, void *end)
  38 {
  39         struct ceph_monmap *m = NULL;
  40         int i, err = -EINVAL;
  41         struct ceph_fsid fsid;
  42         u32 epoch, num_mon;
  43         u16 version;
  44         u32 len;
  45
  46         ceph_decode_32_safe(&p, end, len, bad);
  47         ceph_decode_need(&p, end, len, bad);
  48
  49         dout("monmap_decode %p %p len %d\n", p, end, (int)(end-p));
  50
  51         ceph_decode_16_safe(&p, end, version, bad);
  52
  53         ceph_decode_need(&p, end, sizeof(fsid) + 2*sizeof(u32), bad);
  54         ceph_decode_copy(&p, &fsid, sizeof(fsid));
  55         epoch = ceph_decode_32(&p);
  56
  57         num_mon = ceph_decode_32(&p);
  58         ceph_decode_need(&p, end, num_mon*sizeof(m->mon_inst[0]), bad);
  59
  60         if (num_mon >= CEPH_MAX_MON)
  61                 goto bad;
  62         m = kmalloc(sizeof(*m) + sizeof(m->mon_inst[0])*num_mon, GFP_NOFS);
  63         if (m == NULL)
  64                 return ERR_PTR(-ENOMEM);
  65         m->fsid = fsid;
  66         m->epoch = epoch;
  67         m->num_mon = num_mon;
  68         ceph_decode_copy(&p, m->mon_inst, num_mon*sizeof(m->mon_inst[0]));
  69         for (i = 0; i < num_mon; i++)
  70                 ceph_decode_addr(&m->mon_inst[i].addr);
  71
  72         dout("monmap_decode epoch %d, num_mon %d\n", m->epoch,
  73              m->num_mon);
  74         for (i = 0; i < m->num_mon; i++)
  75                 dout("monmap_decode  mon%d is %s\n", i,
  76                      pr_addr(&m->mon_inst[i].addr.in_addr));
  77         return m;
  78
  79 bad:
  80         dout("monmap_decode failed with %d\n", err);
  81         kfree(m);
  82         return ERR_PTR(err);
  83 }
  84
  85 /*
  86  * return true if *addr is included in the monmap.
  87  */
  88 int ceph_monmap_contains(struct ceph_monmap *m, struct ceph_entity_addr *addr)
  89 {
  90         int i;
  91
  92         for (i = 0; i < m->num_mon; i++)
  93                 if (memcmp(addr, &m->mon_inst[i].addr, sizeof(*addr)) == 0)
  94                         return 1;
  95         return 0;
  96 }
  97
  98 /*
  99  * Close monitor session, if any.
 100  */
 101 static void __close_session(struct ceph_mon_client *monc)
 102 {
 103         if (monc->con) {
 104                 dout("__close_session closing mon%d\n", monc->cur_mon);
 105                 ceph_con_revoke(monc->con, monc->m_auth);
 106                 ceph_con_close(monc->con);
 107                 monc->cur_mon = -1;
 108                 monc->pending_auth = 0;
 109                 ceph_auth_reset(monc->auth);
 110         }
 111 }
 112
 113 /*
 114  * Open a session with a (new) monitor.
 115  */
 116 static int __open_session(struct ceph_mon_client *monc)
 117 {
 118         char r;
 119         int ret;
 120
 121         if (monc->cur_mon < 0) {
 122                 get_random_bytes(&r, 1);
 123                 monc->cur_mon = r % monc->monmap->num_mon;
 124                 dout("open_session num=%d r=%d -> mon%d\n",
 125                      monc->monmap->num_mon, r, monc->cur_mon);
 126                 monc->sub_sent = 0;
 127                 monc->sub_renew_after = jiffies;  /* i.e., expired */
 128                 monc->want_next_osdmap = !!monc->want_next_osdmap;
 129
 130                 dout("open_session mon%d opening\n", monc->cur_mon);
 131                 monc->con->peer_name.type = CEPH_ENTITY_TYPE_MON;
 132                 monc->con->peer_name.num = cpu_to_le64(monc->cur_mon);
 133                 ceph_con_open(monc->con,
 134                               &monc->monmap->mon_inst[monc->cur_mon].addr);
 135
 136                 /* initiatiate authentication handshake */
 137                 ret = ceph_auth_build_hello(monc->auth,
 138                                             monc->m_auth->front.iov_base,
 139                                             monc->m_auth->front_max);
 140                 monc->m_auth->front.iov_len = ret;
 141                 monc->m_auth->hdr.front_len = cpu_to_le32(ret);
 142                 ceph_msg_get(monc->m_auth);  /* keep our ref */
 143                 ceph_con_send(monc->con, monc->m_auth);
 144         } else {
 145                 dout("open_session mon%d already open\n", monc->cur_mon);
 146         }
 147         return 0;
 148 }
 149
 150 static bool __sub_expired(struct ceph_mon_client *monc)
 151 {
 152         return time_after_eq(jiffies, monc->sub_renew_after);
 153 }
 154
 155 /*
 156  * Reschedule delayed work timer.
 157  */
 158 static void __schedule_delayed(struct ceph_mon_client *monc)
 159 {
 160         unsigned delay;
 161
 162         if (monc->cur_mon < 0 || __sub_expired(monc))
 163                 delay = 10 * HZ;
 164         else
 165                 delay = 20 * HZ;
 166         dout("__schedule_delayed after %u\n", delay);
 167         schedule_delayed_work(&monc->delayed_work, delay);
 168 }
 169
 170 /*
 171  * Send subscribe request for mdsmap and/or osdmap.
 172  */
 173 static void __send_subscribe(struct ceph_mon_client *monc)
 174 {
 175         dout("__send_subscribe sub_sent=%u exp=%u want_osd=%d\n",
 176              (unsigned)monc->sub_sent, __sub_expired(monc),
 177              monc->want_next_osdmap);
 178         if ((__sub_expired(monc) && !monc->sub_sent) ||
 179             monc->want_next_osdmap == 1) {
 180                 struct ceph_msg *msg;
 181                 struct ceph_mon_subscribe_item *i;
 182                 void *p, *end;
 183
 184                 msg = ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE, 96, 0, 0, NULL);
 185                 if (!msg)
 186                         return;
 187
 188                 p = msg->front.iov_base;
 189                 end = p + msg->front.iov_len;
 190
 191                 dout("__send_subscribe to 'mdsmap' %u+\n",
 192                      (unsigned)monc->have_mdsmap);
 193                 if (monc->want_next_osdmap) {
 194                         dout("__send_subscribe to 'osdmap' %u\n",
 195                              (unsigned)monc->have_osdmap);
 196                         ceph_encode_32(&p, 3);
 197                         ceph_encode_string(&p, end, "osdmap", 6);
 198                         i = p;
 199                         i->have = cpu_to_le64(monc->have_osdmap);
 200                         i->onetime = 1;
 201                         p += sizeof(*i);
 202                         monc->want_next_osdmap = 2;  /* requested */
 203                 } else {
 204                         ceph_encode_32(&p, 2);
 205                 }
 206                 ceph_encode_string(&p, end, "mdsmap", 6);
 207                 i = p;
 208                 i->have = cpu_to_le64(monc->have_mdsmap);
 209                 i->onetime = 0;
 210                 p += sizeof(*i);
 211                 ceph_encode_string(&p, end, "monmap", 6);
 212                 i = p;
 213                 i->have = 0;
 214                 i->onetime = 0;
 215                 p += sizeof(*i);
 216
 217                 msg->front.iov_len = p - msg->front.iov_base;
 218                 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
 219                 ceph_con_send(monc->con, msg);
 220
 221                 monc->sub_sent = jiffies | 1;  /* never 0 */
 222         }
 223 }
 224
 225 static void handle_subscribe_ack(struct ceph_mon_client *monc,
 226                                  struct ceph_msg *msg)
 227 {
 228         unsigned seconds;
 229         struct ceph_mon_subscribe_ack *h = msg->front.iov_base;
 230
 231         if (msg->front.iov_len < sizeof(*h))
 232                 goto bad;
 233         seconds = le32_to_cpu(h->duration);
 234
 235         mutex_lock(&monc->mutex);
 236         if (monc->hunting) {
 237                 pr_info("mon%d %s session established\n",
 238                         monc->cur_mon, pr_addr(&monc->con->peer_addr.in_addr));
 239                 monc->hunting = false;
 240         }
 241         dout("handle_subscribe_ack after %d seconds\n", seconds);
 242         monc->sub_renew_after = monc->sub_sent + (seconds >> 1)*HZ - 1;
 243         monc->sub_sent = 0;
 244         mutex_unlock(&monc->mutex);
 245         return;
 246 bad:
 247         pr_err("got corrupt subscribe-ack msg\n");
 248         ceph_msg_dump(msg);
 249 }
 250
 251 /*
 252  * Keep track of which maps we have
 253  */
 254 int ceph_monc_got_mdsmap(struct ceph_mon_client *monc, u32 got)
 255 {
 256         mutex_lock(&monc->mutex);
 257         monc->have_mdsmap = got;
 258         mutex_unlock(&monc->mutex);
 259         return 0;
 260 }
 261
 262 int ceph_monc_got_osdmap(struct ceph_mon_client *monc, u32 got)
 263 {
 264         mutex_lock(&monc->mutex);
 265         monc->have_osdmap = got;
 266         monc->want_next_osdmap = 0;
 267         mutex_unlock(&monc->mutex);
 268         return 0;
 269 }
 270
 271 /*
 272  * Register interest in the next osdmap
 273  */
 274 void ceph_monc_request_next_osdmap(struct ceph_mon_client *monc)
 275 {
 276         dout("request_next_osdmap have %u\n", monc->have_osdmap);
 277         mutex_lock(&monc->mutex);
 278         if (!monc->want_next_osdmap)
 279                 monc->want_next_osdmap = 1;
 280         if (monc->want_next_osdmap < 2)
 281                 __send_subscribe(monc);
 282         mutex_unlock(&monc->mutex);
 283 }
 284
 285 /*
 286  *
 287  */
 288 int ceph_monc_open_session(struct ceph_mon_client *monc)
 289 {
 290         if (!monc->con) {
 291                 monc->con = kmalloc(sizeof(*monc->con), GFP_KERNEL);
 292                 if (!monc->con)
 293                         return -ENOMEM;
 294                 ceph_con_init(monc->client->msgr, monc->con);
 295                 monc->con->private = monc;
 296                 monc->con->ops = &mon_con_ops;
 297         }
 298
 299         mutex_lock(&monc->mutex);
 300         __open_session(monc);
 301         __schedule_delayed(monc);
 302         mutex_unlock(&monc->mutex);
 303         return 0;
 304 }
 305
 306 /*
 307  * The monitor responds with mount ack indicate mount success.  The
 308  * included client ticket allows the client to talk to MDSs and OSDs.
 309  */
 310 static void ceph_monc_handle_map(struct ceph_mon_client *monc,
 311                                  struct ceph_msg *msg)
 312 {
 313         struct ceph_client *client = monc->client;
 314         struct ceph_monmap *monmap = NULL, *old = monc->monmap;
 315         void *p, *end;
 316
 317         mutex_lock(&monc->mutex);
 318
 319         dout("handle_monmap\n");
 320         p = msg->front.iov_base;
 321         end = p + msg->front.iov_len;
 322
 323         monmap = ceph_monmap_decode(p, end);
 324         if (IS_ERR(monmap)) {
 325                 pr_err("problem decoding monmap, %d\n",
 326                        (int)PTR_ERR(monmap));
 327                 goto out;
 328         }
 329
 330         if (ceph_check_fsid(monc->client, &monmap->fsid) < 0) {
 331                 kfree(monmap);
 332                 goto out;
 333         }
 334
 335         client->monc.monmap = monmap;
 336         kfree(old);
 337
 338 out:
 339         mutex_unlock(&monc->mutex);
 340         wake_up(&client->auth_wq);
 341 }
 342
 343 /*
 344  * statfs
 345  */
 346 static struct ceph_mon_statfs_request *__lookup_statfs(
 347         struct ceph_mon_client *monc, u64 tid)
 348 {
 349         struct ceph_mon_statfs_request *req;
 350         struct rb_node *n = monc->statfs_request_tree.rb_node;
 351
 352         while (n) {
 353                 req = rb_entry(n, struct ceph_mon_statfs_request, node);
 354                 if (tid < req->tid)
 355                         n = n->rb_left;
 356                 else if (tid > req->tid)
 357                         n = n->rb_right;
 358                 else
 359                         return req;
 360         }
 361         return NULL;
 362 }
 363
 364 static void __insert_statfs(struct ceph_mon_client *monc,
 365                             struct ceph_mon_statfs_request *new)
 366 {
 367         struct rb_node **p = &monc->statfs_request_tree.rb_node;
 368         struct rb_node *parent = NULL;
 369         struct ceph_mon_statfs_request *req = NULL;
 370
 371         while (*p) {
 372                 parent = *p;
 373                 req = rb_entry(parent, struct ceph_mon_statfs_request, node);
 374                 if (new->tid < req->tid)
 375                         p = &(*p)->rb_left;
 376                 else if (new->tid > req->tid)
 377                         p = &(*p)->rb_right;
 378                 else
 379                         BUG();
 380         }
 381
 382         rb_link_node(&new->node, parent, p);
 383         rb_insert_color(&new->node, &monc->statfs_request_tree);
 384 }
 385
 386 static void handle_statfs_reply(struct ceph_mon_client *monc,
 387                                 struct ceph_msg *msg)
 388 {
 389         struct ceph_mon_statfs_request *req;
 390         struct ceph_mon_statfs_reply *reply = msg->front.iov_base;
 391         u64 tid;
 392
 393         if (msg->front.iov_len != sizeof(*reply))
 394                 goto bad;
 395         tid = le64_to_cpu(msg->hdr.tid);
 396         dout("handle_statfs_reply %p tid %llu\n", msg, tid);
 397
 398         mutex_lock(&monc->mutex);
 399         req = __lookup_statfs(monc, tid);
 400         if (req) {
 401                 *req->buf = reply->st;
 402                 req->result = 0;
 403         }
 404         mutex_unlock(&monc->mutex);
 405         if (req)
 406                 complete(&req->completion);
 407         return;
 408
 409 bad:
 410         pr_err("corrupt statfs reply, no tid\n");
 411         ceph_msg_dump(msg);
 412 }
 413
 414 /*
 415  * (re)send a statfs request
 416  */
 417 static int send_statfs(struct ceph_mon_client *monc,
 418                        struct ceph_mon_statfs_request *req)
 419 {
 420         struct ceph_msg *msg;
 421         struct ceph_mon_statfs *h;
 422
 423         dout("send_statfs tid %llu\n", req->tid);
 424         msg = ceph_msg_new(CEPH_MSG_STATFS, sizeof(*h), 0, 0, NULL);
 425         if (IS_ERR(msg))
 426                 return PTR_ERR(msg);
 427         req->request = msg;
 428         msg->hdr.tid = cpu_to_le64(req->tid);
 429         h = msg->front.iov_base;
 430         h->monhdr.have_version = 0;
 431         h->monhdr.session_mon = cpu_to_le16(-1);
 432         h->monhdr.session_mon_tid = 0;
 433         h->fsid = monc->monmap->fsid;
 434         ceph_con_send(monc->con, msg);
 435         return 0;
 436 }
 437
 438 /*
 439  * Do a synchronous statfs().
 440  */
 441 int ceph_monc_do_statfs(struct ceph_mon_client *monc, struct ceph_statfs *buf)
 442 {
 443         struct ceph_mon_statfs_request req;
 444         int err;
 445
 446         req.buf = buf;
 447         init_completion(&req.completion);
 448
 449         /* allocate memory for reply */
 450         err = ceph_msgpool_resv(&monc->msgpool_statfs_reply, 1);
 451         if (err)
 452                 return err;
 453
 454         /* register request */
 455         mutex_lock(&monc->mutex);
 456         req.tid = ++monc->last_tid;
 457         req.last_attempt = jiffies;
 458         req.delay = BASE_DELAY_INTERVAL;
 459         __insert_statfs(monc, &req);
 460         monc->num_statfs_requests++;
 461         mutex_unlock(&monc->mutex);
 462
 463         /* send request and wait */
 464         err = send_statfs(monc, &req);
 465         if (!err)
 466                 err = wait_for_completion_interruptible(&req.completion);
 467
 468         mutex_lock(&monc->mutex);
 469         rb_erase(&req.node, &monc->statfs_request_tree);
 470         monc->num_statfs_requests--;
 471         ceph_msgpool_resv(&monc->msgpool_statfs_reply, -1);
 472         mutex_unlock(&monc->mutex);
 473
 474         if (!err)
 475                 err = req.result;
 476         return err;
 477 }
 478
 479 /*
 480  * Resend pending statfs requests.
 481  */
 482 static void __resend_statfs(struct ceph_mon_client *monc)
 483 {
 484         struct ceph_mon_statfs_request *req;
 485         struct rb_node *p;
 486
 487         for (p = rb_first(&monc->statfs_request_tree); p; p = rb_next(p)) {
 488                 req = rb_entry(p, struct ceph_mon_statfs_request, node);
 489                 send_statfs(monc, req);
 490         }
 491 }
 492
 493 /*
 494  * Delayed work.  If we haven't mounted yet, retry.  Otherwise,
 495  * renew/retry subscription as needed (in case it is timing out, or we
 496  * got an ENOMEM).  And keep the monitor connection alive.
 497  */
 498 static void delayed_work(struct work_struct *work)
 499 {
 500         struct ceph_mon_client *monc =
 501                 container_of(work, struct ceph_mon_client, delayed_work.work);
 502
 503         dout("monc delayed_work\n");
 504         mutex_lock(&monc->mutex);
 505         if (monc->hunting) {
 506                 __close_session(monc);
 507                 __open_session(monc);  /* continue hunting */
 508         } else {
 509                 ceph_con_keepalive(monc->con);
 510                 mutex_unlock(&monc->mutex);
 511
 512                 __validate_auth(monc);
 513
 514                 mutex_lock(&monc->mutex);
 515                 if (monc->auth->ops->is_authenticated(monc->auth))
 516                         __send_subscribe(monc);
 517         }
 518         __schedule_delayed(monc);
 519         mutex_unlock(&monc->mutex);
 520 }
 521
 522 /*
 523  * On startup, we build a temporary monmap populated with the IPs
 524  * provided by mount(2).
 525  */
 526 static int build_initial_monmap(struct ceph_mon_client *monc)
 527 {
 528         struct ceph_mount_args *args = monc->client->mount_args;
 529         struct ceph_entity_addr *mon_addr = args->mon_addr;
 530         int num_mon = args->num_mon;
 531         int i;
 532
 533         /* build initial monmap */
 534         monc->monmap = kzalloc(sizeof(*monc->monmap) +
 535                                num_mon*sizeof(monc->monmap->mon_inst[0]),
 536                                GFP_KERNEL);
 537         if (!monc->monmap)
 538                 return -ENOMEM;
 539         for (i = 0; i < num_mon; i++) {
 540                 monc->monmap->mon_inst[i].addr = mon_addr[i];
 541                 monc->monmap->mon_inst[i].addr.nonce = 0;
 542                 monc->monmap->mon_inst[i].name.type =
 543                         CEPH_ENTITY_TYPE_MON;
 544                 monc->monmap->mon_inst[i].name.num = cpu_to_le64(i);
 545         }
 546         monc->monmap->num_mon = num_mon;
 547         monc->have_fsid = false;
 548
 549         /* release addr memory */
 550         kfree(args->mon_addr);
 551         args->mon_addr = NULL;
 552         args->num_mon = 0;
 553         return 0;
 554 }
 555
 556 int ceph_monc_init(struct ceph_mon_client *monc, struct ceph_client *cl)
 557 {
 558         int err = 0;
 559
 560         dout("init\n");
 561         memset(monc, 0, sizeof(*monc));
 562         monc->client = cl;
 563         monc->monmap = NULL;
 564         mutex_init(&monc->mutex);
 565
 566         err = build_initial_monmap(monc);
 567         if (err)
 568                 goto out;
 569
 570         monc->con = NULL;
 571
 572         /* authentication */
 573         monc->auth = ceph_auth_init(cl->mount_args->name,
 574                                     cl->mount_args->secret);
 575         if (IS_ERR(monc->auth))
 576                 return PTR_ERR(monc->auth);
 577         monc->auth->want_keys =
 578                 CEPH_ENTITY_TYPE_AUTH | CEPH_ENTITY_TYPE_MON |
 579                 CEPH_ENTITY_TYPE_OSD | CEPH_ENTITY_TYPE_MDS;
 580
 581         /* msg pools */
 582         err = ceph_msgpool_init(&monc->msgpool_subscribe_ack,
 583                                sizeof(struct ceph_mon_subscribe_ack), 1, false);
 584         if (err < 0)
 585                 goto out_monmap;
 586         err = ceph_msgpool_init(&monc->msgpool_statfs_reply,
 587                                 sizeof(struct ceph_mon_statfs_reply), 0, false);
 588         if (err < 0)
 589                 goto out_pool1;
 590         err = ceph_msgpool_init(&monc->msgpool_auth_reply, 4096, 1, false);
 591         if (err < 0)
 592                 goto out_pool2;
 593
 594         monc->m_auth = ceph_msg_new(CEPH_MSG_AUTH, 4096, 0, 0, NULL);
 595         monc->pending_auth = 0;
 596         if (IS_ERR(monc->m_auth)) {
 597                 err = PTR_ERR(monc->m_auth);
 598                 monc->m_auth = NULL;
 599                 goto out_pool3;
 600         }
 601
 602         monc->cur_mon = -1;
 603         monc->hunting = true;
 604         monc->sub_renew_after = jiffies;
 605         monc->sub_sent = 0;
 606
 607         INIT_DELAYED_WORK(&monc->delayed_work, delayed_work);
 608         monc->statfs_request_tree = RB_ROOT;
 609         monc->num_statfs_requests = 0;
 610         monc->last_tid = 0;
 611
 612         monc->have_mdsmap = 0;
 613         monc->have_osdmap = 0;
 614         monc->want_next_osdmap = 1;
 615         return 0;
 616
 617 out_pool3:
 618         ceph_msgpool_destroy(&monc->msgpool_auth_reply);
 619 out_pool2:
 620         ceph_msgpool_destroy(&monc->msgpool_subscribe_ack);
 621 out_pool1:
 622         ceph_msgpool_destroy(&monc->msgpool_statfs_reply);
 623 out_monmap:
 624         kfree(monc->monmap);
 625 out:
 626         return err;
 627 }
 628
 629 void ceph_monc_stop(struct ceph_mon_client *monc)
 630 {
 631         dout("stop\n");
 632         cancel_delayed_work_sync(&monc->delayed_work);
 633
 634         mutex_lock(&monc->mutex);
 635         __close_session(monc);
 636         if (monc->con) {
 637                 monc->con->private = NULL;
 638                 monc->con->ops->put(monc->con);
 639                 monc->con = NULL;
 640         }
 641         mutex_unlock(&monc->mutex);
 642
 643         ceph_auth_destroy(monc->auth);
 644
 645         ceph_msg_put(monc->m_auth);
 646         ceph_msgpool_destroy(&monc->msgpool_subscribe_ack);
 647         ceph_msgpool_destroy(&monc->msgpool_statfs_reply);
 648         ceph_msgpool_destroy(&monc->msgpool_auth_reply);
 649
 650         kfree(monc->monmap);
 651 }
 652
 653 static void __send_prepared_auth_request(struct ceph_mon_client *monc, int len)
 654 {
 655         monc->pending_auth = 1;
 656         monc->m_auth->front.iov_len = len;
 657         monc->m_auth->hdr.front_len = cpu_to_le32(len);
 658         ceph_msg_get(monc->m_auth);  /* keep our ref */
 659         ceph_con_send(monc->con, monc->m_auth);
 660 }
 661
 662
 663 static void handle_auth_reply(struct ceph_mon_client *monc,
 664                               struct ceph_msg *msg)
 665 {
 666         int ret;
 667
 668         mutex_lock(&monc->mutex);
 669         monc->pending_auth = 0;
 670         ret = ceph_handle_auth_reply(monc->auth, msg->front.iov_base,
 671                                      msg->front.iov_len,
 672                                      monc->m_auth->front.iov_base,
 673                                      monc->m_auth->front_max);
 674         if (ret < 0) {
 675                 monc->client->auth_err = ret;
 676                 wake_up(&monc->client->auth_wq);
 677         } else if (ret > 0) {
 678                 __send_prepared_auth_request(monc, ret);
 679         } else if (monc->auth->ops->is_authenticated(monc->auth)) {
 680                 dout("authenticated, starting session\n");
 681
 682                 monc->client->msgr->inst.name.type = CEPH_ENTITY_TYPE_CLIENT;
 683                 monc->client->msgr->inst.name.num = monc->auth->global_id;
 684
 685                 __send_subscribe(monc);
 686                 __resend_statfs(monc);
 687         }
 688         mutex_unlock(&monc->mutex);
 689 }
 690
 691 static int __validate_auth(struct ceph_mon_client *monc)
 692 {
 693         int ret;
 694
 695         if (monc->pending_auth)
 696                 return 0;
 697
 698         ret = ceph_build_auth(monc->auth, monc->m_auth->front.iov_base,
 699                               monc->m_auth->front_max);
 700         if (ret <= 0)
 701                 return ret; /* either an error, or no need to authenticate */
 702         __send_prepared_auth_request(monc, ret);
 703         return 0;
 704 }
 705
 706 int ceph_monc_validate_auth(struct ceph_mon_client *monc)
 707 {
 708         int ret;
 709
 710         mutex_lock(&monc->mutex);
 711         ret = __validate_auth(monc);
 712         mutex_unlock(&monc->mutex);
 713         return ret;
 714 }
 715
 716 /*
 717  * handle incoming message
 718  */
 719 static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
 720 {
 721         struct ceph_mon_client *monc = con->private;
 722         int type = le16_to_cpu(msg->hdr.type);
 723
 724         if (!monc)
 725                 return;
 726
 727         switch (type) {
 728         case CEPH_MSG_AUTH_REPLY:
 729                 handle_auth_reply(monc, msg);
 730                 break;
 731
 732         case CEPH_MSG_MON_SUBSCRIBE_ACK:
 733                 handle_subscribe_ack(monc, msg);
 734                 break;
 735
 736         case CEPH_MSG_STATFS_REPLY:
 737                 handle_statfs_reply(monc, msg);
 738                 break;
 739
 740         case CEPH_MSG_MON_MAP:
 741                 ceph_monc_handle_map(monc, msg);
 742                 break;
 743
 744         case CEPH_MSG_MDS_MAP:
 745                 ceph_mdsc_handle_map(&monc->client->mdsc, msg);
 746                 break;
 747
 748         case CEPH_MSG_OSD_MAP:
 749                 ceph_osdc_handle_map(&monc->client->osdc, msg);
 750                 break;
 751
 752         default:
 753                 pr_err("received unknown message type %d %s\n", type,
 754                        ceph_msg_type_name(type));
 755         }
 756         ceph_msg_put(msg);
 757 }
 758
 759 /*
 760  * Allocate memory for incoming message
 761  */
 762 static struct ceph_msg *mon_alloc_msg(struct ceph_connection *con,
 763                                       struct ceph_msg_header *hdr,
 764                                       int *skip)
 765 {
 766         struct ceph_mon_client *monc = con->private;
 767         int type = le16_to_cpu(hdr->type);
 768         int front_len = le32_to_cpu(hdr->front_len);
 769         struct ceph_msg *m;
 770
 771         *skip = 0;
 772
 773         switch (type) {
 774         case CEPH_MSG_MON_SUBSCRIBE_ACK:
 775                 m = ceph_msgpool_get(&monc->msgpool_subscribe_ack, front_len);
 776                 break;
 777         case CEPH_MSG_STATFS_REPLY:
 778                 m = ceph_msgpool_get(&monc->msgpool_statfs_reply, front_len);
 779                 break;
 780         case CEPH_MSG_AUTH_REPLY:
 781                 m = ceph_msgpool_get(&monc->msgpool_auth_reply, front_len);
 782                 break;
 783         default:
 784                 return NULL;
 785         }
 786
 787         if (!m)
 788                 *skip = 1;
 789
 790         return m;
 791 }
 792
 793 /*
 794  * If the monitor connection resets, pick a new monitor and resubmit
 795  * any pending requests.
 796  */
 797 static void mon_fault(struct ceph_connection *con)
 798 {
 799         struct ceph_mon_client *monc = con->private;
 800
 801         if (!monc)
 802                 return;
 803
 804         dout("mon_fault\n");
 805         mutex_lock(&monc->mutex);
 806         if (!con->private)
 807                 goto out;
 808
 809         if (monc->con && !monc->hunting)
 810                 pr_info("mon%d %s session lost, "
 811                         "hunting for new mon\n", monc->cur_mon,
 812                         pr_addr(&monc->con->peer_addr.in_addr));
 813
 814         __close_session(monc);
 815         if (!monc->hunting) {
 816                 /* start hunting */
 817                 monc->hunting = true;
 818                 __open_session(monc);
 819         } else {
 820                 /* already hunting, let's wait a bit */
 821                 __schedule_delayed(monc);
 822         }
 823 out:
 824         mutex_unlock(&monc->mutex);
 825 }
 826
 827 const static struct ceph_connection_operations mon_con_ops = {
 828         .get = ceph_con_get,
 829         .put = ceph_con_put,
 830         .dispatch = dispatch,
 831         .fault = mon_fault,
 832         .alloc_msg = mon_alloc_msg,
 833 };