fs/ceph/mon_client.c

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