mon_client.c 21.0 KB
Newer Older
S
Sage Weil 已提交
1 2 3
#include "ceph_debug.h"

#include <linux/types.h>
4
#include <linux/slab.h>
S
Sage Weil 已提交
5 6 7 8 9
#include <linux/random.h>
#include <linux/sched.h>

#include "mon_client.h"
#include "super.h"
10
#include "auth.h"
S
Sage Weil 已提交
11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30
#include "decode.h"

/*
 * Interact with Ceph monitor cluster.  Handle requests for new map
 * versions, and periodically resend as needed.  Also implement
 * statfs() and umount().
 *
 * A small cluster of Ceph "monitors" are responsible for managing critical
 * cluster configuration and state information.  An odd number (e.g., 3, 5)
 * of cmon daemons use a modified version of the Paxos part-time parliament
 * algorithm to manage the MDS map (mds cluster membership), OSD map, and
 * list of clients who have mounted the file system.
 *
 * We maintain an open, active session with a monitor at all times in order to
 * receive timely MDSMap updates.  We periodically send a keepalive byte on the
 * TCP socket to ensure we detect a failure.  If the connection does break, we
 * randomly hunt for a new monitor.  Once the connection is reestablished, we
 * resend any outstanding requests.
 */

31
static const struct ceph_connection_operations mon_con_ops;
S
Sage Weil 已提交
32

33 34
static int __validate_auth(struct ceph_mon_client *monc);

S
Sage Weil 已提交
35 36 37 38 39 40 41 42 43 44
/*
 * Decode a monmap blob (e.g., during mount).
 */
struct ceph_monmap *ceph_monmap_decode(void *p, void *end)
{
	struct ceph_monmap *m = NULL;
	int i, err = -EINVAL;
	struct ceph_fsid fsid;
	u32 epoch, num_mon;
	u16 version;
45 46 47 48
	u32 len;

	ceph_decode_32_safe(&p, end, len, bad);
	ceph_decode_need(&p, end, len, bad);
S
Sage Weil 已提交
49 50 51 52 53 54 55

	dout("monmap_decode %p %p len %d\n", p, end, (int)(end-p));

	ceph_decode_16_safe(&p, end, version, bad);

	ceph_decode_need(&p, end, sizeof(fsid) + 2*sizeof(u32), bad);
	ceph_decode_copy(&p, &fsid, sizeof(fsid));
56
	epoch = ceph_decode_32(&p);
S
Sage Weil 已提交
57

58
	num_mon = ceph_decode_32(&p);
S
Sage Weil 已提交
59 60 61 62 63 64 65 66 67 68 69
	ceph_decode_need(&p, end, num_mon*sizeof(m->mon_inst[0]), bad);

	if (num_mon >= CEPH_MAX_MON)
		goto bad;
	m = kmalloc(sizeof(*m) + sizeof(m->mon_inst[0])*num_mon, GFP_NOFS);
	if (m == NULL)
		return ERR_PTR(-ENOMEM);
	m->fsid = fsid;
	m->epoch = epoch;
	m->num_mon = num_mon;
	ceph_decode_copy(&p, m->mon_inst, num_mon*sizeof(m->mon_inst[0]));
70 71
	for (i = 0; i < num_mon; i++)
		ceph_decode_addr(&m->mon_inst[i].addr);
S
Sage Weil 已提交
72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93

	dout("monmap_decode epoch %d, num_mon %d\n", m->epoch,
	     m->num_mon);
	for (i = 0; i < m->num_mon; i++)
		dout("monmap_decode  mon%d is %s\n", i,
		     pr_addr(&m->mon_inst[i].addr.in_addr));
	return m;

bad:
	dout("monmap_decode failed with %d\n", err);
	kfree(m);
	return ERR_PTR(err);
}

/*
 * return true if *addr is included in the monmap.
 */
int ceph_monmap_contains(struct ceph_monmap *m, struct ceph_entity_addr *addr)
{
	int i;

	for (i = 0; i < m->num_mon; i++)
S
Sage Weil 已提交
94
		if (memcmp(addr, &m->mon_inst[i].addr, sizeof(*addr)) == 0)
S
Sage Weil 已提交
95 96 97 98
			return 1;
	return 0;
}

99 100 101 102 103 104 105 106
/*
 * Send an auth request.
 */
static void __send_prepared_auth_request(struct ceph_mon_client *monc, int len)
{
	monc->pending_auth = 1;
	monc->m_auth->front.iov_len = len;
	monc->m_auth->hdr.front_len = cpu_to_le32(len);
107
	ceph_con_revoke(monc->con, monc->m_auth);
108 109 110 111
	ceph_msg_get(monc->m_auth);  /* keep our ref */
	ceph_con_send(monc->con, monc->m_auth);
}

S
Sage Weil 已提交
112 113 114 115 116 117 118
/*
 * Close monitor session, if any.
 */
static void __close_session(struct ceph_mon_client *monc)
{
	if (monc->con) {
		dout("__close_session closing mon%d\n", monc->cur_mon);
119
		ceph_con_revoke(monc->con, monc->m_auth);
S
Sage Weil 已提交
120 121
		ceph_con_close(monc->con);
		monc->cur_mon = -1;
122
		monc->pending_auth = 0;
123
		ceph_auth_reset(monc->auth);
S
Sage Weil 已提交
124 125 126 127 128 129 130 131 132
	}
}

/*
 * Open a session with a (new) monitor.
 */
static int __open_session(struct ceph_mon_client *monc)
{
	char r;
133
	int ret;
S
Sage Weil 已提交
134 135 136 137 138 139 140 141 142 143 144 145 146 147 148

	if (monc->cur_mon < 0) {
		get_random_bytes(&r, 1);
		monc->cur_mon = r % monc->monmap->num_mon;
		dout("open_session num=%d r=%d -> mon%d\n",
		     monc->monmap->num_mon, r, monc->cur_mon);
		monc->sub_sent = 0;
		monc->sub_renew_after = jiffies;  /* i.e., expired */
		monc->want_next_osdmap = !!monc->want_next_osdmap;

		dout("open_session mon%d opening\n", monc->cur_mon);
		monc->con->peer_name.type = CEPH_ENTITY_TYPE_MON;
		monc->con->peer_name.num = cpu_to_le64(monc->cur_mon);
		ceph_con_open(monc->con,
			      &monc->monmap->mon_inst[monc->cur_mon].addr);
149 150 151 152 153

		/* initiatiate authentication handshake */
		ret = ceph_auth_build_hello(monc->auth,
					    monc->m_auth->front.iov_base,
					    monc->m_auth->front_max);
154
		__send_prepared_auth_request(monc, ret);
S
Sage Weil 已提交
155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172
	} else {
		dout("open_session mon%d already open\n", monc->cur_mon);
	}
	return 0;
}

static bool __sub_expired(struct ceph_mon_client *monc)
{
	return time_after_eq(jiffies, monc->sub_renew_after);
}

/*
 * Reschedule delayed work timer.
 */
static void __schedule_delayed(struct ceph_mon_client *monc)
{
	unsigned delay;

173
	if (monc->cur_mon < 0 || __sub_expired(monc))
S
Sage Weil 已提交
174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190
		delay = 10 * HZ;
	else
		delay = 20 * HZ;
	dout("__schedule_delayed after %u\n", delay);
	schedule_delayed_work(&monc->delayed_work, delay);
}

/*
 * Send subscribe request for mdsmap and/or osdmap.
 */
static void __send_subscribe(struct ceph_mon_client *monc)
{
	dout("__send_subscribe sub_sent=%u exp=%u want_osd=%d\n",
	     (unsigned)monc->sub_sent, __sub_expired(monc),
	     monc->want_next_osdmap);
	if ((__sub_expired(monc) && !monc->sub_sent) ||
	    monc->want_next_osdmap == 1) {
191
		struct ceph_msg *msg = monc->m_subscribe;
S
Sage Weil 已提交
192 193 194 195
		struct ceph_mon_subscribe_item *i;
		void *p, *end;

		p = msg->front.iov_base;
196
		end = p + msg->front_max;
S
Sage Weil 已提交
197 198 199 200 201 202

		dout("__send_subscribe to 'mdsmap' %u+\n",
		     (unsigned)monc->have_mdsmap);
		if (monc->want_next_osdmap) {
			dout("__send_subscribe to 'osdmap' %u\n",
			     (unsigned)monc->have_osdmap);
203
			ceph_encode_32(&p, 3);
S
Sage Weil 已提交
204 205 206 207 208 209 210
			ceph_encode_string(&p, end, "osdmap", 6);
			i = p;
			i->have = cpu_to_le64(monc->have_osdmap);
			i->onetime = 1;
			p += sizeof(*i);
			monc->want_next_osdmap = 2;  /* requested */
		} else {
211
			ceph_encode_32(&p, 2);
S
Sage Weil 已提交
212 213 214 215 216 217
		}
		ceph_encode_string(&p, end, "mdsmap", 6);
		i = p;
		i->have = cpu_to_le64(monc->have_mdsmap);
		i->onetime = 0;
		p += sizeof(*i);
218 219 220 221 222
		ceph_encode_string(&p, end, "monmap", 6);
		i = p;
		i->have = 0;
		i->onetime = 0;
		p += sizeof(*i);
S
Sage Weil 已提交
223 224 225

		msg->front.iov_len = p - msg->front.iov_base;
		msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
226 227
		ceph_con_revoke(monc->con, msg);
		ceph_con_send(monc->con, ceph_msg_get(msg));
S
Sage Weil 已提交
228 229 230 231 232 233 234 235 236

		monc->sub_sent = jiffies | 1;  /* never 0 */
	}
}

static void handle_subscribe_ack(struct ceph_mon_client *monc,
				 struct ceph_msg *msg)
{
	unsigned seconds;
237 238 239 240 241
	struct ceph_mon_subscribe_ack *h = msg->front.iov_base;

	if (msg->front.iov_len < sizeof(*h))
		goto bad;
	seconds = le32_to_cpu(h->duration);
S
Sage Weil 已提交
242 243 244 245 246 247 248 249

	mutex_lock(&monc->mutex);
	if (monc->hunting) {
		pr_info("mon%d %s session established\n",
			monc->cur_mon, pr_addr(&monc->con->peer_addr.in_addr));
		monc->hunting = false;
	}
	dout("handle_subscribe_ack after %d seconds\n", seconds);
250
	monc->sub_renew_after = monc->sub_sent + (seconds >> 1)*HZ - 1;
S
Sage Weil 已提交
251 252 253 254 255
	monc->sub_sent = 0;
	mutex_unlock(&monc->mutex);
	return;
bad:
	pr_err("got corrupt subscribe-ack msg\n");
256
	ceph_msg_dump(msg);
S
Sage Weil 已提交
257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292
}

/*
 * Keep track of which maps we have
 */
int ceph_monc_got_mdsmap(struct ceph_mon_client *monc, u32 got)
{
	mutex_lock(&monc->mutex);
	monc->have_mdsmap = got;
	mutex_unlock(&monc->mutex);
	return 0;
}

int ceph_monc_got_osdmap(struct ceph_mon_client *monc, u32 got)
{
	mutex_lock(&monc->mutex);
	monc->have_osdmap = got;
	monc->want_next_osdmap = 0;
	mutex_unlock(&monc->mutex);
	return 0;
}

/*
 * Register interest in the next osdmap
 */
void ceph_monc_request_next_osdmap(struct ceph_mon_client *monc)
{
	dout("request_next_osdmap have %u\n", monc->have_osdmap);
	mutex_lock(&monc->mutex);
	if (!monc->want_next_osdmap)
		monc->want_next_osdmap = 1;
	if (monc->want_next_osdmap < 2)
		__send_subscribe(monc);
	mutex_unlock(&monc->mutex);
}

293
/*
S
Sage Weil 已提交
294
 *
295 296
 */
int ceph_monc_open_session(struct ceph_mon_client *monc)
S
Sage Weil 已提交
297 298 299 300 301 302 303 304 305 306 307
{
	if (!monc->con) {
		monc->con = kmalloc(sizeof(*monc->con), GFP_KERNEL);
		if (!monc->con)
			return -ENOMEM;
		ceph_con_init(monc->client->msgr, monc->con);
		monc->con->private = monc;
		monc->con->ops = &mon_con_ops;
	}

	mutex_lock(&monc->mutex);
308
	__open_session(monc);
S
Sage Weil 已提交
309 310 311 312 313
	__schedule_delayed(monc);
	mutex_unlock(&monc->mutex);
	return 0;
}

314 315 316 317
/*
 * The monitor responds with mount ack indicate mount success.  The
 * included client ticket allows the client to talk to MDSs and OSDs.
 */
318 319
static void ceph_monc_handle_map(struct ceph_mon_client *monc,
				 struct ceph_msg *msg)
320 321 322 323 324 325 326 327 328 329 330 331 332 333 334
{
	struct ceph_client *client = monc->client;
	struct ceph_monmap *monmap = NULL, *old = monc->monmap;
	void *p, *end;

	mutex_lock(&monc->mutex);

	dout("handle_monmap\n");
	p = msg->front.iov_base;
	end = p + msg->front.iov_len;

	monmap = ceph_monmap_decode(p, end);
	if (IS_ERR(monmap)) {
		pr_err("problem decoding monmap, %d\n",
		       (int)PTR_ERR(monmap));
S
Sage Weil 已提交
335
		goto out;
336
	}
337 338

	if (ceph_check_fsid(monc->client, &monmap->fsid) < 0) {
339
		kfree(monmap);
S
Sage Weil 已提交
340
		goto out;
341 342 343 344 345
	}

	client->monc.monmap = monmap;
	kfree(old);

S
Sage Weil 已提交
346
out:
347
	mutex_unlock(&monc->mutex);
348
	wake_up(&client->auth_wq);
349 350
}

S
Sage Weil 已提交
351 352 353
/*
 * statfs
 */
354
static struct ceph_mon_generic_request *__lookup_generic_req(
355 356
	struct ceph_mon_client *monc, u64 tid)
{
357 358
	struct ceph_mon_generic_request *req;
	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 363 364 365 366 367 368 369 370 371
		if (tid < req->tid)
			n = n->rb_left;
		else if (tid > req->tid)
			n = n->rb_right;
		else
			return req;
	}
	return NULL;
}

372 373
static void __insert_generic_request(struct ceph_mon_client *monc,
			    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 380

	while (*p) {
		parent = *p;
381
		req = rb_entry(parent, struct ceph_mon_generic_request, node);
382 383 384 385 386 387 388 389 390
		if (new->tid < req->tid)
			p = &(*p)->rb_left;
		else if (new->tid > req->tid)
			p = &(*p)->rb_right;
		else
			BUG();
	}

	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)
S
Sage Weil 已提交
395
{
396 397
	struct ceph_mon_generic_request *req =
		container_of(kref, struct ceph_mon_generic_request, kref);
S
Sage Weil 已提交
398 399 400 401 402 403 404

	if (req->reply)
		ceph_msg_put(req->reply);
	if (req->request)
		ceph_msg_put(req->request);
}

405
static void put_generic_request(struct ceph_mon_generic_request *req)
S
Sage Weil 已提交
406
{
407
	kref_put(&req->kref, release_generic_request);
S
Sage Weil 已提交
408 409
}

410
static void get_generic_request(struct ceph_mon_generic_request *req)
S
Sage Weil 已提交
411 412 413 414
{
	kref_get(&req->kref);
}

415
static struct ceph_msg *get_generic_reply(struct ceph_connection *con,
S
Sage Weil 已提交
416 417 418 419
					 struct ceph_msg_header *hdr,
					 int *skip)
{
	struct ceph_mon_client *monc = con->private;
420
	struct ceph_mon_generic_request *req;
S
Sage Weil 已提交
421 422 423 424
	u64 tid = le64_to_cpu(hdr->tid);
	struct ceph_msg *m;

	mutex_lock(&monc->mutex);
425
	req = __lookup_generic_req(monc, tid);
S
Sage Weil 已提交
426
	if (!req) {
427
		dout("get_generic_reply %lld dne\n", tid);
S
Sage Weil 已提交
428 429 430
		*skip = 1;
		m = NULL;
	} else {
431
		dout("get_generic_reply %lld got %p\n", tid, req->reply);
S
Sage Weil 已提交
432 433 434 435 436 437 438 439 440 441 442
		m = ceph_msg_get(req->reply);
		/*
		 * we don't need to track the connection reading into
		 * this reply because we only have one open connection
		 * at a time, ever.
		 */
	}
	mutex_unlock(&monc->mutex);
	return m;
}

S
Sage Weil 已提交
443 444 445
static void handle_statfs_reply(struct ceph_mon_client *monc,
				struct ceph_msg *msg)
{
446
	struct ceph_mon_generic_request *req;
S
Sage Weil 已提交
447
	struct ceph_mon_statfs_reply *reply = msg->front.iov_base;
S
Sage Weil 已提交
448
	u64 tid = le64_to_cpu(msg->hdr.tid);
S
Sage Weil 已提交
449 450 451 452 453 454

	if (msg->front.iov_len != sizeof(*reply))
		goto bad;
	dout("handle_statfs_reply %p tid %llu\n", msg, tid);

	mutex_lock(&monc->mutex);
455
	req = __lookup_generic_req(monc, tid);
S
Sage Weil 已提交
456
	if (req) {
457
		*(struct ceph_statfs *)req->buf = reply->st;
S
Sage Weil 已提交
458
		req->result = 0;
459
		get_generic_request(req);
S
Sage Weil 已提交
460 461
	}
	mutex_unlock(&monc->mutex);
S
Sage Weil 已提交
462
	if (req) {
S
Sage Weil 已提交
463
		complete(&req->completion);
464
		put_generic_request(req);
S
Sage Weil 已提交
465
	}
S
Sage Weil 已提交
466 467 468
	return;

bad:
469
	pr_err("corrupt generic reply, no tid\n");
470
	ceph_msg_dump(msg);
S
Sage Weil 已提交
471 472 473
}

/*
S
Sage Weil 已提交
474
 * Do a synchronous statfs().
S
Sage Weil 已提交
475
 */
S
Sage Weil 已提交
476
int ceph_monc_do_statfs(struct ceph_mon_client *monc, struct ceph_statfs *buf)
S
Sage Weil 已提交
477
{
478
	struct ceph_mon_generic_request *req;
S
Sage Weil 已提交
479
	struct ceph_mon_statfs *h;
S
Sage Weil 已提交
480 481
	int err;

J
Julia Lawall 已提交
482
	req = kzalloc(sizeof(*req), GFP_NOFS);
S
Sage Weil 已提交
483 484 485 486 487 488
	if (!req)
		return -ENOMEM;

	kref_init(&req->kref);
	req->buf = buf;
	init_completion(&req->completion);
S
Sage Weil 已提交
489

490
	err = -ENOMEM;
491
	req->request = ceph_msg_new(CEPH_MSG_STATFS, sizeof(*h), GFP_NOFS);
492
	if (!req->request)
S
Sage Weil 已提交
493
		goto out;
494
	req->reply = ceph_msg_new(CEPH_MSG_STATFS_REPLY, 1024, GFP_NOFS);
495
	if (!req->reply)
S
Sage Weil 已提交
496 497 498 499
		goto out;

	/* fill out request */
	h = req->request->front.iov_base;
500 501 502
	h->monhdr.have_version = 0;
	h->monhdr.session_mon = cpu_to_le16(-1);
	h->monhdr.session_mon_tid = 0;
S
Sage Weil 已提交
503 504 505 506
	h->fsid = monc->monmap->fsid;

	/* register request */
	mutex_lock(&monc->mutex);
S
Sage Weil 已提交
507 508
	req->tid = ++monc->last_tid;
	req->request->hdr.tid = cpu_to_le64(req->tid);
509 510
	__insert_generic_request(monc, req);
	monc->num_generic_requests++;
S
Sage Weil 已提交
511 512 513
	mutex_unlock(&monc->mutex);

	/* send request and wait */
S
Sage Weil 已提交
514 515
	ceph_con_send(monc->con, ceph_msg_get(req->request));
	err = wait_for_completion_interruptible(&req->completion);
S
Sage Weil 已提交
516 517

	mutex_lock(&monc->mutex);
518 519
	rb_erase(&req->node, &monc->generic_request_tree);
	monc->num_generic_requests--;
S
Sage Weil 已提交
520 521 522
	mutex_unlock(&monc->mutex);

	if (!err)
S
Sage Weil 已提交
523 524 525
		err = req->result;

out:
526
	kref_put(&req->kref, release_generic_request);
S
Sage Weil 已提交
527 528 529 530 531 532
	return err;
}

/*
 * Resend pending statfs requests.
 */
533
static void __resend_generic_request(struct ceph_mon_client *monc)
S
Sage Weil 已提交
534
{
535
	struct ceph_mon_generic_request *req;
536
	struct rb_node *p;
S
Sage Weil 已提交
537

538 539
	for (p = rb_first(&monc->generic_request_tree); p; p = rb_next(p)) {
		req = rb_entry(p, struct ceph_mon_generic_request, node);
540
		ceph_con_revoke(monc->con, req->request);
S
Sage Weil 已提交
541
		ceph_con_send(monc->con, ceph_msg_get(req->request));
S
Sage Weil 已提交
542 543 544 545 546 547 548 549 550 551 552 553 554 555 556
	}
}

/*
 * Delayed work.  If we haven't mounted yet, retry.  Otherwise,
 * renew/retry subscription as needed (in case it is timing out, or we
 * got an ENOMEM).  And keep the monitor connection alive.
 */
static void delayed_work(struct work_struct *work)
{
	struct ceph_mon_client *monc =
		container_of(work, struct ceph_mon_client, delayed_work.work);

	dout("monc delayed_work\n");
	mutex_lock(&monc->mutex);
557 558 559
	if (monc->hunting) {
		__close_session(monc);
		__open_session(monc);  /* continue hunting */
S
Sage Weil 已提交
560
	} else {
561
		ceph_con_keepalive(monc->con);
562 563 564

		__validate_auth(monc);

565 566
		if (monc->auth->ops->is_authenticated(monc->auth))
			__send_subscribe(monc);
S
Sage Weil 已提交
567 568 569 570 571
	}
	__schedule_delayed(monc);
	mutex_unlock(&monc->mutex);
}

572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596
/*
 * On startup, we build a temporary monmap populated with the IPs
 * provided by mount(2).
 */
static int build_initial_monmap(struct ceph_mon_client *monc)
{
	struct ceph_mount_args *args = monc->client->mount_args;
	struct ceph_entity_addr *mon_addr = args->mon_addr;
	int num_mon = args->num_mon;
	int i;

	/* build initial monmap */
	monc->monmap = kzalloc(sizeof(*monc->monmap) +
			       num_mon*sizeof(monc->monmap->mon_inst[0]),
			       GFP_KERNEL);
	if (!monc->monmap)
		return -ENOMEM;
	for (i = 0; i < num_mon; i++) {
		monc->monmap->mon_inst[i].addr = mon_addr[i];
		monc->monmap->mon_inst[i].addr.nonce = 0;
		monc->monmap->mon_inst[i].name.type =
			CEPH_ENTITY_TYPE_MON;
		monc->monmap->mon_inst[i].name.num = cpu_to_le64(i);
	}
	monc->monmap->num_mon = num_mon;
597
	monc->have_fsid = false;
598 599 600 601 602 603 604 605

	/* release addr memory */
	kfree(args->mon_addr);
	args->mon_addr = NULL;
	args->num_mon = 0;
	return 0;
}

S
Sage Weil 已提交
606 607 608 609 610 611 612 613 614 615
int ceph_monc_init(struct ceph_mon_client *monc, struct ceph_client *cl)
{
	int err = 0;

	dout("init\n");
	memset(monc, 0, sizeof(*monc));
	monc->client = cl;
	monc->monmap = NULL;
	mutex_init(&monc->mutex);

616 617 618 619
	err = build_initial_monmap(monc);
	if (err)
		goto out;

S
Sage Weil 已提交
620 621
	monc->con = NULL;

622 623 624 625 626 627 628 629 630
	/* authentication */
	monc->auth = ceph_auth_init(cl->mount_args->name,
				    cl->mount_args->secret);
	if (IS_ERR(monc->auth))
		return PTR_ERR(monc->auth);
	monc->auth->want_keys =
		CEPH_ENTITY_TYPE_AUTH | CEPH_ENTITY_TYPE_MON |
		CEPH_ENTITY_TYPE_OSD | CEPH_ENTITY_TYPE_MDS;

631
	/* msgs */
632
	err = -ENOMEM;
633
	monc->m_subscribe_ack = ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE_ACK,
634 635
				     sizeof(struct ceph_mon_subscribe_ack),
				     GFP_NOFS);
636
	if (!monc->m_subscribe_ack)
637
		goto out_monmap;
638

639 640 641 642
	monc->m_subscribe = ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE, 96, GFP_NOFS);
	if (!monc->m_subscribe)
		goto out_subscribe_ack;

643
	monc->m_auth_reply = ceph_msg_new(CEPH_MSG_AUTH_REPLY, 4096, GFP_NOFS);
644
	if (!monc->m_auth_reply)
645
		goto out_subscribe;
646

647
	monc->m_auth = ceph_msg_new(CEPH_MSG_AUTH, 4096, GFP_NOFS);
648
	monc->pending_auth = 0;
649
	if (!monc->m_auth)
650
		goto out_auth_reply;
S
Sage Weil 已提交
651 652

	monc->cur_mon = -1;
653
	monc->hunting = true;
S
Sage Weil 已提交
654 655 656 657
	monc->sub_renew_after = jiffies;
	monc->sub_sent = 0;

	INIT_DELAYED_WORK(&monc->delayed_work, delayed_work);
658 659
	monc->generic_request_tree = RB_ROOT;
	monc->num_generic_requests = 0;
S
Sage Weil 已提交
660 661 662 663 664
	monc->last_tid = 0;

	monc->have_mdsmap = 0;
	monc->have_osdmap = 0;
	monc->want_next_osdmap = 1;
665 666
	return 0;

667 668
out_auth_reply:
	ceph_msg_put(monc->m_auth_reply);
669 670
out_subscribe:
	ceph_msg_put(monc->m_subscribe);
671 672
out_subscribe_ack:
	ceph_msg_put(monc->m_subscribe_ack);
673 674
out_monmap:
	kfree(monc->monmap);
S
Sage Weil 已提交
675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692
out:
	return err;
}

void ceph_monc_stop(struct ceph_mon_client *monc)
{
	dout("stop\n");
	cancel_delayed_work_sync(&monc->delayed_work);

	mutex_lock(&monc->mutex);
	__close_session(monc);
	if (monc->con) {
		monc->con->private = NULL;
		monc->con->ops->put(monc->con);
		monc->con = NULL;
	}
	mutex_unlock(&monc->mutex);

693 694 695
	ceph_auth_destroy(monc->auth);

	ceph_msg_put(monc->m_auth);
696
	ceph_msg_put(monc->m_auth_reply);
697
	ceph_msg_put(monc->m_subscribe);
698
	ceph_msg_put(monc->m_subscribe_ack);
S
Sage Weil 已提交
699 700 701 702

	kfree(monc->monmap);
}

703 704 705 706
static void handle_auth_reply(struct ceph_mon_client *monc,
			      struct ceph_msg *msg)
{
	int ret;
707
	int was_auth = 0;
708 709

	mutex_lock(&monc->mutex);
710 711
	if (monc->auth->ops)
		was_auth = monc->auth->ops->is_authenticated(monc->auth);
712
	monc->pending_auth = 0;
713 714 715 716 717
	ret = ceph_handle_auth_reply(monc->auth, msg->front.iov_base,
				     msg->front.iov_len,
				     monc->m_auth->front.iov_base,
				     monc->m_auth->front_max);
	if (ret < 0) {
718 719
		monc->client->auth_err = ret;
		wake_up(&monc->client->auth_wq);
720
	} else if (ret > 0) {
721
		__send_prepared_auth_request(monc, ret);
722
	} else if (!was_auth && monc->auth->ops->is_authenticated(monc->auth)) {
723
		dout("authenticated, starting session\n");
724 725 726 727

		monc->client->msgr->inst.name.type = CEPH_ENTITY_TYPE_CLIENT;
		monc->client->msgr->inst.name.num = monc->auth->global_id;

728
		__send_subscribe(monc);
729
		__resend_generic_request(monc);
730 731 732 733
	}
	mutex_unlock(&monc->mutex);
}

734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758
static int __validate_auth(struct ceph_mon_client *monc)
{
	int ret;

	if (monc->pending_auth)
		return 0;

	ret = ceph_build_auth(monc->auth, monc->m_auth->front.iov_base,
			      monc->m_auth->front_max);
	if (ret <= 0)
		return ret; /* either an error, or no need to authenticate */
	__send_prepared_auth_request(monc, ret);
	return 0;
}

int ceph_monc_validate_auth(struct ceph_mon_client *monc)
{
	int ret;

	mutex_lock(&monc->mutex);
	ret = __validate_auth(monc);
	mutex_unlock(&monc->mutex);
	return ret;
}

S
Sage Weil 已提交
759 760 761 762 763 764 765 766 767 768 769 770
/*
 * handle incoming message
 */
static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
{
	struct ceph_mon_client *monc = con->private;
	int type = le16_to_cpu(msg->hdr.type);

	if (!monc)
		return;

	switch (type) {
771 772
	case CEPH_MSG_AUTH_REPLY:
		handle_auth_reply(monc, msg);
S
Sage Weil 已提交
773 774 775 776 777 778 779 780 781 782
		break;

	case CEPH_MSG_MON_SUBSCRIBE_ACK:
		handle_subscribe_ack(monc, msg);
		break;

	case CEPH_MSG_STATFS_REPLY:
		handle_statfs_reply(monc, msg);
		break;

783 784 785 786
	case CEPH_MSG_MON_MAP:
		ceph_monc_handle_map(monc, msg);
		break;

S
Sage Weil 已提交
787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805
	case CEPH_MSG_MDS_MAP:
		ceph_mdsc_handle_map(&monc->client->mdsc, msg);
		break;

	case CEPH_MSG_OSD_MAP:
		ceph_osdc_handle_map(&monc->client->osdc, msg);
		break;

	default:
		pr_err("received unknown message type %d %s\n", type,
		       ceph_msg_type_name(type));
	}
	ceph_msg_put(msg);
}

/*
 * Allocate memory for incoming message
 */
static struct ceph_msg *mon_alloc_msg(struct ceph_connection *con,
806 807
				      struct ceph_msg_header *hdr,
				      int *skip)
S
Sage Weil 已提交
808 809 810
{
	struct ceph_mon_client *monc = con->private;
	int type = le16_to_cpu(hdr->type);
811
	int front_len = le32_to_cpu(hdr->front_len);
812
	struct ceph_msg *m = NULL;
S
Sage Weil 已提交
813

814
	*skip = 0;
815

S
Sage Weil 已提交
816 817
	switch (type) {
	case CEPH_MSG_MON_SUBSCRIBE_ACK:
818
		m = ceph_msg_get(monc->m_subscribe_ack);
819
		break;
S
Sage Weil 已提交
820
	case CEPH_MSG_STATFS_REPLY:
821
		return get_generic_reply(con, hdr, skip);
822
	case CEPH_MSG_AUTH_REPLY:
823
		m = ceph_msg_get(monc->m_auth_reply);
824
		break;
825 826 827
	case CEPH_MSG_MON_MAP:
	case CEPH_MSG_MDS_MAP:
	case CEPH_MSG_OSD_MAP:
828
		m = ceph_msg_new(type, front_len, GFP_NOFS);
829
		break;
S
Sage Weil 已提交
830
	}
831

832 833
	if (!m) {
		pr_info("alloc_msg unknown type %d\n", type);
834
		*skip = 1;
835
	}
836
	return m;
S
Sage Weil 已提交
837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863
}

/*
 * If the monitor connection resets, pick a new monitor and resubmit
 * any pending requests.
 */
static void mon_fault(struct ceph_connection *con)
{
	struct ceph_mon_client *monc = con->private;

	if (!monc)
		return;

	dout("mon_fault\n");
	mutex_lock(&monc->mutex);
	if (!con->private)
		goto out;

	if (monc->con && !monc->hunting)
		pr_info("mon%d %s session lost, "
			"hunting for new mon\n", monc->cur_mon,
			pr_addr(&monc->con->peer_addr.in_addr));

	__close_session(monc);
	if (!monc->hunting) {
		/* start hunting */
		monc->hunting = true;
864
		__open_session(monc);
S
Sage Weil 已提交
865 866 867 868 869 870 871 872
	} else {
		/* already hunting, let's wait a bit */
		__schedule_delayed(monc);
	}
out:
	mutex_unlock(&monc->mutex);
}

873
static const struct ceph_connection_operations mon_con_ops = {
S
Sage Weil 已提交
874 875 876 877 878 879
	.get = ceph_con_get,
	.put = ceph_con_put,
	.dispatch = dispatch,
	.fault = mon_fault,
	.alloc_msg = mon_alloc_msg,
};