1// SPDX-License-Identifier: GPL-2.0
   2#include <linux/ceph/ceph_debug.h>
   3
   4#include <linux/module.h>
   5#include <linux/types.h>
   6#include <linux/slab.h>
   7#include <linux/random.h>
   8#include <linux/sched.h>
   9
  10#include <linux/ceph/ceph_features.h>
  11#include <linux/ceph/mon_client.h>
  12#include <linux/ceph/libceph.h>
  13#include <linux/ceph/debugfs.h>
  14#include <linux/ceph/decode.h>
  15#include <linux/ceph/auth.h>
  16
  17/*
  18 * Interact with Ceph monitor cluster.  Handle requests for new map
  19 * versions, and periodically resend as needed.  Also implement
  20 * statfs() and umount().
  21 *
  22 * A small cluster of Ceph "monitors" are responsible for managing critical
  23 * cluster configuration and state information.  An odd number (e.g., 3, 5)
  24 * of cmon daemons use a modified version of the Paxos part-time parliament
  25 * algorithm to manage the MDS map (mds cluster membership), OSD map, and
  26 * list of clients who have mounted the file system.
  27 *
  28 * We maintain an open, active session with a monitor at all times in order to
  29 * receive timely MDSMap updates.  We periodically send a keepalive byte on the
  30 * TCP socket to ensure we detect a failure.  If the connection does break, we
  31 * randomly hunt for a new monitor.  Once the connection is reestablished, we
  32 * resend any outstanding requests.
  33 */
  34
  35static const struct ceph_connection_operations mon_con_ops;
  36
  37static int __validate_auth(struct ceph_mon_client *monc);
  38
  39static int decode_mon_info(void **p, void *end, bool msgr2,
  40			   struct ceph_entity_addr *addr)
  41{
  42	void *mon_info_end;
  43	u32 struct_len;
  44	u8 struct_v;
  45	int ret;
  46
  47	ret = ceph_start_decoding(p, end, 1, "mon_info_t", &struct_v,
  48				  &struct_len);
  49	if (ret)
  50		return ret;
  51
  52	mon_info_end = *p + struct_len;
  53	ceph_decode_skip_string(p, end, e_inval);  /* skip mon name */
  54	ret = ceph_decode_entity_addrvec(p, end, msgr2, addr);
  55	if (ret)
  56		return ret;
  57
  58	*p = mon_info_end;
  59	return 0;
  60
  61e_inval:
  62	return -EINVAL;
  63}
  64
  65/*
  66 * Decode a monmap blob (e.g., during mount).
  67 *
  68 * Assume MonMap v3 (i.e. encoding with MONNAMES and MONENC).
  69 */
  70static struct ceph_monmap *ceph_monmap_decode(void **p, void *end, bool msgr2)
  71{
  72	struct ceph_monmap *monmap = NULL;
  73	struct ceph_fsid fsid;
  74	u32 struct_len;
  75	int blob_len;
  76	int num_mon;
  77	u8 struct_v;
  78	u32 epoch;
  79	int ret;
  80	int i;
  81
  82	ceph_decode_32_safe(p, end, blob_len, e_inval);
  83	ceph_decode_need(p, end, blob_len, e_inval);
  84
  85	ret = ceph_start_decoding(p, end, 6, "monmap", &struct_v, &struct_len);
  86	if (ret)
  87		goto fail;
  88
  89	dout("%s struct_v %d\n", __func__, struct_v);
  90	ceph_decode_copy_safe(p, end, &fsid, sizeof(fsid), e_inval);
  91	ceph_decode_32_safe(p, end, epoch, e_inval);
  92	if (struct_v >= 6) {
  93		u32 feat_struct_len;
  94		u8 feat_struct_v;
  95
  96		*p += sizeof(struct ceph_timespec);  /* skip last_changed */
  97		*p += sizeof(struct ceph_timespec);  /* skip created */
  98
  99		ret = ceph_start_decoding(p, end, 1, "mon_feature_t",
 100					  &feat_struct_v, &feat_struct_len);
 101		if (ret)
 102			goto fail;
 103
 104		*p += feat_struct_len;  /* skip persistent_features */
 105
 106		ret = ceph_start_decoding(p, end, 1, "mon_feature_t",
 107					  &feat_struct_v, &feat_struct_len);
 108		if (ret)
 109			goto fail;
 110
 111		*p += feat_struct_len;  /* skip optional_features */
 112	}
 113	ceph_decode_32_safe(p, end, num_mon, e_inval);
 114
 115	dout("%s fsid %pU epoch %u num_mon %d\n", __func__, &fsid, epoch,
 116	     num_mon);
 117	if (num_mon > CEPH_MAX_MON)
 118		goto e_inval;
 119
 120	monmap = kmalloc(struct_size(monmap, mon_inst, num_mon), GFP_NOIO);
 121	if (!monmap) {
 122		ret = -ENOMEM;
 123		goto fail;
 124	}
 125	monmap->fsid = fsid;
 126	monmap->epoch = epoch;
 127	monmap->num_mon = num_mon;
 128
 129	/* legacy_mon_addr map or mon_info map */
 130	for (i = 0; i < num_mon; i++) {
 131		struct ceph_entity_inst *inst = &monmap->mon_inst[i];
 132
 133		ceph_decode_skip_string(p, end, e_inval);  /* skip mon name */
 134		inst->name.type = CEPH_ENTITY_TYPE_MON;
 135		inst->name.num = cpu_to_le64(i);
 136
 137		if (struct_v >= 6)
 138			ret = decode_mon_info(p, end, msgr2, &inst->addr);
 139		else
 140			ret = ceph_decode_entity_addr(p, end, &inst->addr);
 141		if (ret)
 142			goto fail;
 143
 144		dout("%s mon%d addr %s\n", __func__, i,
 145		     ceph_pr_addr(&inst->addr));
 146	}
 147
 148	return monmap;
 149
 150e_inval:
 151	ret = -EINVAL;
 152fail:
 153	kfree(monmap);
 154	return ERR_PTR(ret);
 155}
 156
 157/*
 158 * return true if *addr is included in the monmap.
 159 */
 160int ceph_monmap_contains(struct ceph_monmap *m, struct ceph_entity_addr *addr)
 161{
 162	int i;
 163
 164	for (i = 0; i < m->num_mon; i++) {
 165		if (ceph_addr_equal_no_type(addr, &m->mon_inst[i].addr))
 166			return 1;
 167	}
 168
 169	return 0;
 170}
 171
 172/*
 173 * Send an auth request.
 174 */
 175static void __send_prepared_auth_request(struct ceph_mon_client *monc, int len)
 176{
 177	monc->pending_auth = 1;
 178	monc->m_auth->front.iov_len = len;
 179	monc->m_auth->hdr.front_len = cpu_to_le32(len);
 180	ceph_msg_revoke(monc->m_auth);
 181	ceph_msg_get(monc->m_auth);  /* keep our ref */
 182	ceph_con_send(&monc->con, monc->m_auth);
 183}
 184
 185/*
 186 * Close monitor session, if any.
 187 */
 188static void __close_session(struct ceph_mon_client *monc)
 189{
 190	dout("__close_session closing mon%d\n", monc->cur_mon);
 191	ceph_msg_revoke(monc->m_auth);
 192	ceph_msg_revoke_incoming(monc->m_auth_reply);
 193	ceph_msg_revoke(monc->m_subscribe);
 194	ceph_msg_revoke_incoming(monc->m_subscribe_ack);
 195	ceph_con_close(&monc->con);
 196
 197	monc->pending_auth = 0;
 198	ceph_auth_reset(monc->auth);
 199}
 200
 201/*
 202 * Pick a new monitor at random and set cur_mon.  If we are repicking
 203 * (i.e. cur_mon is already set), be sure to pick a different one.
 204 */
 205static void pick_new_mon(struct ceph_mon_client *monc)
 206{
 207	int old_mon = monc->cur_mon;
 208
 209	BUG_ON(monc->monmap->num_mon < 1);
 210
 211	if (monc->monmap->num_mon == 1) {
 212		monc->cur_mon = 0;
 213	} else {
 214		int max = monc->monmap->num_mon;
 215		int o = -1;
 216		int n;
 217
 218		if (monc->cur_mon >= 0) {
 219			if (monc->cur_mon < monc->monmap->num_mon)
 220				o = monc->cur_mon;
 221			if (o >= 0)
 222				max--;
 223		}
 224
 225		n = get_random_u32_below(max);
 226		if (o >= 0 && n >= o)
 227			n++;
 228
 229		monc->cur_mon = n;
 230	}
 231
 232	dout("%s mon%d -> mon%d out of %d mons\n", __func__, old_mon,
 233	     monc->cur_mon, monc->monmap->num_mon);
 234}
 235
 236/*
 237 * Open a session with a new monitor.
 238 */
 239static void __open_session(struct ceph_mon_client *monc)
 240{
 241	int ret;
 242
 243	pick_new_mon(monc);
 244
 245	monc->hunting = true;
 246	if (monc->had_a_connection) {
 247		monc->hunt_mult *= CEPH_MONC_HUNT_BACKOFF;
 248		if (monc->hunt_mult > CEPH_MONC_HUNT_MAX_MULT)
 249			monc->hunt_mult = CEPH_MONC_HUNT_MAX_MULT;
 250	}
 251
 252	monc->sub_renew_after = jiffies; /* i.e., expired */
 253	monc->sub_renew_sent = 0;
 254
 255	dout("%s opening mon%d\n", __func__, monc->cur_mon);
 256	ceph_con_open(&monc->con, CEPH_ENTITY_TYPE_MON, monc->cur_mon,
 257		      &monc->monmap->mon_inst[monc->cur_mon].addr);
 258
 259	/*
 260	 * Queue a keepalive to ensure that in case of an early fault
 261	 * the messenger doesn't put us into STANDBY state and instead
 262	 * retries.  This also ensures that our timestamp is valid by
 263	 * the time we finish hunting and delayed_work() checks it.
 264	 */
 265	ceph_con_keepalive(&monc->con);
 266	if (ceph_msgr2(monc->client)) {
 267		monc->pending_auth = 1;
 268		return;
 269	}
 270
 271	/* initiate authentication handshake */
 272	ret = ceph_auth_build_hello(monc->auth,
 273				    monc->m_auth->front.iov_base,
 274				    monc->m_auth->front_alloc_len);
 275	BUG_ON(ret <= 0);
 276	__send_prepared_auth_request(monc, ret);
 277}
 278
 279static void reopen_session(struct ceph_mon_client *monc)
 280{
 281	if (!monc->hunting)
 282		pr_info("mon%d %s session lost, hunting for new mon\n",
 283		    monc->cur_mon, ceph_pr_addr(&monc->con.peer_addr));
 284
 285	__close_session(monc);
 286	__open_session(monc);
 287}
 288
 289void ceph_monc_reopen_session(struct ceph_mon_client *monc)
 290{
 291	mutex_lock(&monc->mutex);
 292	reopen_session(monc);
 293	mutex_unlock(&monc->mutex);
 294}
 295
 296static void un_backoff(struct ceph_mon_client *monc)
 297{
 298	monc->hunt_mult /= 2; /* reduce by 50% */
 299	if (monc->hunt_mult < 1)
 300		monc->hunt_mult = 1;
 301	dout("%s hunt_mult now %d\n", __func__, monc->hunt_mult);
 302}
 303
 304/*
 305 * Reschedule delayed work timer.
 306 */
 307static void __schedule_delayed(struct ceph_mon_client *monc)
 308{
 309	unsigned long delay;
 310
 311	if (monc->hunting)
 312		delay = CEPH_MONC_HUNT_INTERVAL * monc->hunt_mult;
 313	else
 314		delay = CEPH_MONC_PING_INTERVAL;
 315
 316	dout("__schedule_delayed after %lu\n", delay);
 317	mod_delayed_work(system_wq, &monc->delayed_work,
 318			 round_jiffies_relative(delay));
 319}
 320
 321const char *ceph_sub_str[] = {
 322	[CEPH_SUB_MONMAP] = "monmap",
 323	[CEPH_SUB_OSDMAP] = "osdmap",
 324	[CEPH_SUB_FSMAP]  = "fsmap.user",
 325	[CEPH_SUB_MDSMAP] = "mdsmap",
 326};
 327
 328/*
 329 * Send subscribe request for one or more maps, according to
 330 * monc->subs.
 331 */
 332static void __send_subscribe(struct ceph_mon_client *monc)
 333{
 334	struct ceph_msg *msg = monc->m_subscribe;
 335	void *p = msg->front.iov_base;
 336	void *const end = p + msg->front_alloc_len;
 337	int num = 0;
 338	int i;
 339
 340	dout("%s sent %lu\n", __func__, monc->sub_renew_sent);
 341
 342	BUG_ON(monc->cur_mon < 0);
 343
 344	if (!monc->sub_renew_sent)
 345		monc->sub_renew_sent = jiffies | 1; /* never 0 */
 346
 347	msg->hdr.version = cpu_to_le16(2);
 348
 349	for (i = 0; i < ARRAY_SIZE(monc->subs); i++) {
 350		if (monc->subs[i].want)
 351			num++;
 352	}
 353	BUG_ON(num < 1); /* monmap sub is always there */
 354	ceph_encode_32(&p, num);
 355	for (i = 0; i < ARRAY_SIZE(monc->subs); i++) {
 356		char buf[32];
 357		int len;
 358
 359		if (!monc->subs[i].want)
 360			continue;
 361
 362		len = sprintf(buf, "%s", ceph_sub_str[i]);
 363		if (i == CEPH_SUB_MDSMAP &&
 364		    monc->fs_cluster_id != CEPH_FS_CLUSTER_ID_NONE)
 365			len += sprintf(buf + len, ".%d", monc->fs_cluster_id);
 366
 367		dout("%s %s start %llu flags 0x%x\n", __func__, buf,
 368		     le64_to_cpu(monc->subs[i].item.start),
 369		     monc->subs[i].item.flags);
 370		ceph_encode_string(&p, end, buf, len);
 371		memcpy(p, &monc->subs[i].item, sizeof(monc->subs[i].item));
 372		p += sizeof(monc->subs[i].item);
 373	}
 374
 375	BUG_ON(p > end);
 376	msg->front.iov_len = p - msg->front.iov_base;
 377	msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
 378	ceph_msg_revoke(msg);
 379	ceph_con_send(&monc->con, ceph_msg_get(msg));
 380}
 381
 382static void handle_subscribe_ack(struct ceph_mon_client *monc,
 383				 struct ceph_msg *msg)
 384{
 385	unsigned int seconds;
 386	struct ceph_mon_subscribe_ack *h = msg->front.iov_base;
 387
 388	if (msg->front.iov_len < sizeof(*h))
 389		goto bad;
 390	seconds = le32_to_cpu(h->duration);
 391
 392	mutex_lock(&monc->mutex);
 393	if (monc->sub_renew_sent) {
 394		/*
 395		 * This is only needed for legacy (infernalis or older)
 396		 * MONs -- see delayed_work().
 397		 */
 398		monc->sub_renew_after = monc->sub_renew_sent +
 399					    (seconds >> 1) * HZ - 1;
 400		dout("%s sent %lu duration %d renew after %lu\n", __func__,
 401		     monc->sub_renew_sent, seconds, monc->sub_renew_after);
 402		monc->sub_renew_sent = 0;
 403	} else {
 404		dout("%s sent %lu renew after %lu, ignoring\n", __func__,
 405		     monc->sub_renew_sent, monc->sub_renew_after);
 406	}
 407	mutex_unlock(&monc->mutex);
 408	return;
 409bad:
 410	pr_err("got corrupt subscribe-ack msg\n");
 411	ceph_msg_dump(msg);
 412}
 413
 414/*
 415 * Register interest in a map
 416 *
 417 * @sub: one of CEPH_SUB_*
 418 * @epoch: X for "every map since X", or 0 for "just the latest"
 419 */
 420static bool __ceph_monc_want_map(struct ceph_mon_client *monc, int sub,
 421				 u32 epoch, bool continuous)
 422{
 423	__le64 start = cpu_to_le64(epoch);
 424	u8 flags = !continuous ? CEPH_SUBSCRIBE_ONETIME : 0;
 425
 426	dout("%s %s epoch %u continuous %d\n", __func__, ceph_sub_str[sub],
 427	     epoch, continuous);
 428
 429	if (monc->subs[sub].want &&
 430	    monc->subs[sub].item.start == start &&
 431	    monc->subs[sub].item.flags == flags)
 432		return false;
 433
 434	monc->subs[sub].item.start = start;
 435	monc->subs[sub].item.flags = flags;
 436	monc->subs[sub].want = true;
 437
 438	return true;
 439}
 440
 441bool ceph_monc_want_map(struct ceph_mon_client *monc, int sub, u32 epoch,
 442			bool continuous)
 443{
 444	bool need_request;
 445
 446	mutex_lock(&monc->mutex);
 447	need_request = __ceph_monc_want_map(monc, sub, epoch, continuous);
 448	mutex_unlock(&monc->mutex);
 449
 450	return need_request;
 451}
 452EXPORT_SYMBOL(ceph_monc_want_map);
 453
 454/*
 455 * Keep track of which maps we have
 456 *
 457 * @sub: one of CEPH_SUB_*
 458 */
 459static void __ceph_monc_got_map(struct ceph_mon_client *monc, int sub,
 460				u32 epoch)
 461{
 462	dout("%s %s epoch %u\n", __func__, ceph_sub_str[sub], epoch);
 463
 464	if (monc->subs[sub].want) {
 465		if (monc->subs[sub].item.flags & CEPH_SUBSCRIBE_ONETIME)
 466			monc->subs[sub].want = false;
 467		else
 468			monc->subs[sub].item.start = cpu_to_le64(epoch + 1);
 469	}
 470
 471	monc->subs[sub].have = epoch;
 472}
 473
 474void ceph_monc_got_map(struct ceph_mon_client *monc, int sub, u32 epoch)
 475{
 476	mutex_lock(&monc->mutex);
 477	__ceph_monc_got_map(monc, sub, epoch);
 478	mutex_unlock(&monc->mutex);
 479}
 480EXPORT_SYMBOL(ceph_monc_got_map);
 481
 482void ceph_monc_renew_subs(struct ceph_mon_client *monc)
 483{
 484	mutex_lock(&monc->mutex);
 485	__send_subscribe(monc);
 486	mutex_unlock(&monc->mutex);
 487}
 488EXPORT_SYMBOL(ceph_monc_renew_subs);
 489
 490/*
 491 * Wait for an osdmap with a given epoch.
 492 *
 493 * @epoch: epoch to wait for
 494 * @timeout: in jiffies, 0 means "wait forever"
 495 */
 496int ceph_monc_wait_osdmap(struct ceph_mon_client *monc, u32 epoch,
 497			  unsigned long timeout)
 498{
 499	unsigned long started = jiffies;
 500	long ret;
 501
 502	mutex_lock(&monc->mutex);
 503	while (monc->subs[CEPH_SUB_OSDMAP].have < epoch) {
 504		mutex_unlock(&monc->mutex);
 505
 506		if (timeout && time_after_eq(jiffies, started + timeout))
 507			return -ETIMEDOUT;
 508
 509		ret = wait_event_interruptible_timeout(monc->client->auth_wq,
 510				     monc->subs[CEPH_SUB_OSDMAP].have >= epoch,
 511				     ceph_timeout_jiffies(timeout));
 512		if (ret < 0)
 513			return ret;
 514
 515		mutex_lock(&monc->mutex);
 516	}
 517
 518	mutex_unlock(&monc->mutex);
 519	return 0;
 520}
 521EXPORT_SYMBOL(ceph_monc_wait_osdmap);
 522
 523/*
 524 * Open a session with a random monitor.  Request monmap and osdmap,
 525 * which are waited upon in __ceph_open_session().
 526 */
 527int ceph_monc_open_session(struct ceph_mon_client *monc)
 528{
 529	mutex_lock(&monc->mutex);
 530	__ceph_monc_want_map(monc, CEPH_SUB_MONMAP, 0, true);
 531	__ceph_monc_want_map(monc, CEPH_SUB_OSDMAP, 0, false);
 532	__open_session(monc);
 533	__schedule_delayed(monc);
 534	mutex_unlock(&monc->mutex);
 535	return 0;
 536}
 537EXPORT_SYMBOL(ceph_monc_open_session);
 538
 539static void ceph_monc_handle_map(struct ceph_mon_client *monc,
 540				 struct ceph_msg *msg)
 541{
 542	struct ceph_client *client = monc->client;
 543	struct ceph_monmap *monmap;
 544	void *p, *end;
 545
 546	mutex_lock(&monc->mutex);
 547
 548	dout("handle_monmap\n");
 549	p = msg->front.iov_base;
 550	end = p + msg->front.iov_len;
 551
 552	monmap = ceph_monmap_decode(&p, end, ceph_msgr2(client));
 553	if (IS_ERR(monmap)) {
 554		pr_err("problem decoding monmap, %d\n",
 555		       (int)PTR_ERR(monmap));
 556		ceph_msg_dump(msg);
 557		goto out;
 558	}
 559
 560	if (ceph_check_fsid(client, &monmap->fsid) < 0) {
 561		kfree(monmap);
 562		goto out;
 563	}
 564
 565	kfree(monc->monmap);
 566	monc->monmap = monmap;
 567
 568	__ceph_monc_got_map(monc, CEPH_SUB_MONMAP, monc->monmap->epoch);
 569	client->have_fsid = true;
 570
 571out:
 572	mutex_unlock(&monc->mutex);
 573	wake_up_all(&client->auth_wq);
 574}
 575
 576/*
 577 * generic requests (currently statfs, mon_get_version)
 578 */
 579DEFINE_RB_FUNCS(generic_request, struct ceph_mon_generic_request, tid, node)
 580
 581static void release_generic_request(struct kref *kref)
 582{
 583	struct ceph_mon_generic_request *req =
 584		container_of(kref, struct ceph_mon_generic_request, kref);
 585
 586	dout("%s greq %p request %p reply %p\n", __func__, req, req->request,
 587	     req->reply);
 588	WARN_ON(!RB_EMPTY_NODE(&req->node));
 589
 590	if (req->reply)
 591		ceph_msg_put(req->reply);
 592	if (req->request)
 593		ceph_msg_put(req->request);
 594
 595	kfree(req);
 596}
 597
 598static void put_generic_request(struct ceph_mon_generic_request *req)
 599{
 600	if (req)
 601		kref_put(&req->kref, release_generic_request);
 602}
 603
 604static void get_generic_request(struct ceph_mon_generic_request *req)
 605{
 606	kref_get(&req->kref);
 607}
 608
 609static struct ceph_mon_generic_request *
 610alloc_generic_request(struct ceph_mon_client *monc, gfp_t gfp)
 611{
 612	struct ceph_mon_generic_request *req;
 613
 614	req = kzalloc(sizeof(*req), gfp);
 615	if (!req)
 616		return NULL;
 617
 618	req->monc = monc;
 619	kref_init(&req->kref);
 620	RB_CLEAR_NODE(&req->node);
 621	init_completion(&req->completion);
 622
 623	dout("%s greq %p\n", __func__, req);
 624	return req;
 625}
 626
 627static void register_generic_request(struct ceph_mon_generic_request *req)
 628{
 629	struct ceph_mon_client *monc = req->monc;
 630
 631	WARN_ON(req->tid);
 632
 633	get_generic_request(req);
 634	req->tid = ++monc->last_tid;
 635	insert_generic_request(&monc->generic_request_tree, req);
 636}
 637
 638static void send_generic_request(struct ceph_mon_client *monc,
 639				 struct ceph_mon_generic_request *req)
 640{
 641	WARN_ON(!req->tid);
 642
 643	dout("%s greq %p tid %llu\n", __func__, req, req->tid);
 644	req->request->hdr.tid = cpu_to_le64(req->tid);
 645	ceph_con_send(&monc->con, ceph_msg_get(req->request));
 646}
 647
 648static void __finish_generic_request(struct ceph_mon_generic_request *req)
 649{
 650	struct ceph_mon_client *monc = req->monc;
 651
 652	dout("%s greq %p tid %llu\n", __func__, req, req->tid);
 653	erase_generic_request(&monc->generic_request_tree, req);
 654
 655	ceph_msg_revoke(req->request);
 656	ceph_msg_revoke_incoming(req->reply);
 657}
 658
 659static void finish_generic_request(struct ceph_mon_generic_request *req)
 660{
 661	__finish_generic_request(req);
 662	put_generic_request(req);
 663}
 664
 665static void complete_generic_request(struct ceph_mon_generic_request *req)
 666{
 667	if (req->complete_cb)
 668		req->complete_cb(req);
 669	else
 670		complete_all(&req->completion);
 671	put_generic_request(req);
 672}
 673
 674static void cancel_generic_request(struct ceph_mon_generic_request *req)
 675{
 676	struct ceph_mon_client *monc = req->monc;
 677	struct ceph_mon_generic_request *lookup_req;
 678
 679	dout("%s greq %p tid %llu\n", __func__, req, req->tid);
 680
 681	mutex_lock(&monc->mutex);
 682	lookup_req = lookup_generic_request(&monc->generic_request_tree,
 683					    req->tid);
 684	if (lookup_req) {
 685		WARN_ON(lookup_req != req);
 686		finish_generic_request(req);
 687	}
 688
 689	mutex_unlock(&monc->mutex);
 690}
 691
 692static int wait_generic_request(struct ceph_mon_generic_request *req)
 693{
 694	int ret;
 695
 696	dout("%s greq %p tid %llu\n", __func__, req, req->tid);
 697	ret = wait_for_completion_interruptible(&req->completion);
 698	if (ret)
 699		cancel_generic_request(req);
 700	else
 701		ret = req->result; /* completed */
 702
 703	return ret;
 704}
 705
 706static struct ceph_msg *get_generic_reply(struct ceph_connection *con,
 707					 struct ceph_msg_header *hdr,
 708					 int *skip)
 709{
 710	struct ceph_mon_client *monc = con->private;
 711	struct ceph_mon_generic_request *req;
 712	u64 tid = le64_to_cpu(hdr->tid);
 713	struct ceph_msg *m;
 714
 715	mutex_lock(&monc->mutex);
 716	req = lookup_generic_request(&monc->generic_request_tree, tid);
 717	if (!req) {
 718		dout("get_generic_reply %lld dne\n", tid);
 719		*skip = 1;
 720		m = NULL;
 721	} else {
 722		dout("get_generic_reply %lld got %p\n", tid, req->reply);
 723		*skip = 0;
 724		m = ceph_msg_get(req->reply);
 725		/*
 726		 * we don't need to track the connection reading into
 727		 * this reply because we only have one open connection
 728		 * at a time, ever.
 729		 */
 730	}
 731	mutex_unlock(&monc->mutex);
 732	return m;
 733}
 734
 735/*
 736 * statfs
 737 */
 738static void handle_statfs_reply(struct ceph_mon_client *monc,
 739				struct ceph_msg *msg)
 740{
 741	struct ceph_mon_generic_request *req;
 742	struct ceph_mon_statfs_reply *reply = msg->front.iov_base;
 743	u64 tid = le64_to_cpu(msg->hdr.tid);
 744
 745	dout("%s msg %p tid %llu\n", __func__, msg, tid);
 746
 747	if (msg->front.iov_len != sizeof(*reply))
 748		goto bad;
 749
 750	mutex_lock(&monc->mutex);
 751	req = lookup_generic_request(&monc->generic_request_tree, tid);
 752	if (!req) {
 753		mutex_unlock(&monc->mutex);
 754		return;
 755	}
 756
 757	req->result = 0;
 758	*req->u.st = reply->st; /* struct */
 759	__finish_generic_request(req);
 760	mutex_unlock(&monc->mutex);
 761
 762	complete_generic_request(req);
 763	return;
 764
 765bad:
 766	pr_err("corrupt statfs reply, tid %llu\n", tid);
 767	ceph_msg_dump(msg);
 768}
 769
 770/*
 771 * Do a synchronous statfs().
 772 */
 773int ceph_monc_do_statfs(struct ceph_mon_client *monc, u64 data_pool,
 774			struct ceph_statfs *buf)
 775{
 776	struct ceph_mon_generic_request *req;
 777	struct ceph_mon_statfs *h;
 778	int ret = -ENOMEM;
 779
 780	req = alloc_generic_request(monc, GFP_NOFS);
 781	if (!req)
 782		goto out;
 783
 784	req->request = ceph_msg_new(CEPH_MSG_STATFS, sizeof(*h), GFP_NOFS,
 785				    true);
 786	if (!req->request)
 787		goto out;
 788
 789	req->reply = ceph_msg_new(CEPH_MSG_STATFS_REPLY, 64, GFP_NOFS, true);
 790	if (!req->reply)
 791		goto out;
 792
 793	req->u.st = buf;
 794	req->request->hdr.version = cpu_to_le16(2);
 795
 796	mutex_lock(&monc->mutex);
 797	register_generic_request(req);
 798	/* fill out request */
 799	h = req->request->front.iov_base;
 800	h->monhdr.have_version = 0;
 801	h->monhdr.session_mon = cpu_to_le16(-1);
 802	h->monhdr.session_mon_tid = 0;
 803	h->fsid = monc->monmap->fsid;
 804	h->contains_data_pool = (data_pool != CEPH_NOPOOL);
 805	h->data_pool = cpu_to_le64(data_pool);
 806	send_generic_request(monc, req);
 807	mutex_unlock(&monc->mutex);
 808
 809	ret = wait_generic_request(req);
 810out:
 811	put_generic_request(req);
 812	return ret;
 813}
 814EXPORT_SYMBOL(ceph_monc_do_statfs);
 815
 816static void handle_get_version_reply(struct ceph_mon_client *monc,
 817				     struct ceph_msg *msg)
 818{
 819	struct ceph_mon_generic_request *req;
 820	u64 tid = le64_to_cpu(msg->hdr.tid);
 821	void *p = msg->front.iov_base;
 822	void *end = p + msg->front_alloc_len;
 823	u64 handle;
 824
 825	dout("%s msg %p tid %llu\n", __func__, msg, tid);
 826
 827	ceph_decode_need(&p, end, 2*sizeof(u64), bad);
 828	handle = ceph_decode_64(&p);
 829	if (tid != 0 && tid != handle)
 830		goto bad;
 831
 832	mutex_lock(&monc->mutex);
 833	req = lookup_generic_request(&monc->generic_request_tree, handle);
 834	if (!req) {
 835		mutex_unlock(&monc->mutex);
 836		return;
 837	}
 838
 839	req->result = 0;
 840	req->u.newest = ceph_decode_64(&p);
 841	__finish_generic_request(req);
 842	mutex_unlock(&monc->mutex);
 843
 844	complete_generic_request(req);
 845	return;
 846
 847bad:
 848	pr_err("corrupt mon_get_version reply, tid %llu\n", tid);
 849	ceph_msg_dump(msg);
 850}
 851
 852static struct ceph_mon_generic_request *
 853__ceph_monc_get_version(struct ceph_mon_client *monc, const char *what,
 854			ceph_monc_callback_t cb, u64 private_data)
 855{
 856	struct ceph_mon_generic_request *req;
 857
 858	req = alloc_generic_request(monc, GFP_NOIO);
 859	if (!req)
 860		goto err_put_req;
 861
 862	req->request = ceph_msg_new(CEPH_MSG_MON_GET_VERSION,
 863				    sizeof(u64) + sizeof(u32) + strlen(what),
 864				    GFP_NOIO, true);
 865	if (!req->request)
 866		goto err_put_req;
 867
 868	req->reply = ceph_msg_new(CEPH_MSG_MON_GET_VERSION_REPLY, 32, GFP_NOIO,
 869				  true);
 870	if (!req->reply)
 871		goto err_put_req;
 872
 873	req->complete_cb = cb;
 874	req->private_data = private_data;
 875
 876	mutex_lock(&monc->mutex);
 877	register_generic_request(req);
 878	{
 879		void *p = req->request->front.iov_base;
 880		void *const end = p + req->request->front_alloc_len;
 881
 882		ceph_encode_64(&p, req->tid); /* handle */
 883		ceph_encode_string(&p, end, what, strlen(what));
 884		WARN_ON(p != end);
 885	}
 886	send_generic_request(monc, req);
 887	mutex_unlock(&monc->mutex);
 888
 889	return req;
 890
 891err_put_req:
 892	put_generic_request(req);
 893	return ERR_PTR(-ENOMEM);
 894}
 895
 896/*
 897 * Send MMonGetVersion and wait for the reply.
 898 *
 899 * @what: one of "mdsmap", "osdmap" or "monmap"
 900 */
 901int ceph_monc_get_version(struct ceph_mon_client *monc, const char *what,
 902			  u64 *newest)
 903{
 904	struct ceph_mon_generic_request *req;
 905	int ret;
 906
 907	req = __ceph_monc_get_version(monc, what, NULL, 0);
 908	if (IS_ERR(req))
 909		return PTR_ERR(req);
 910
 911	ret = wait_generic_request(req);
 912	if (!ret)
 913		*newest = req->u.newest;
 914
 915	put_generic_request(req);
 916	return ret;
 917}
 918EXPORT_SYMBOL(ceph_monc_get_version);
 919
 920/*
 921 * Send MMonGetVersion,
 922 *
 923 * @what: one of "mdsmap", "osdmap" or "monmap"
 924 */
 925int ceph_monc_get_version_async(struct ceph_mon_client *monc, const char *what,
 926				ceph_monc_callback_t cb, u64 private_data)
 927{
 928	struct ceph_mon_generic_request *req;
 929
 930	req = __ceph_monc_get_version(monc, what, cb, private_data);
 931	if (IS_ERR(req))
 932		return PTR_ERR(req);
 933
 934	put_generic_request(req);
 935	return 0;
 936}
 937EXPORT_SYMBOL(ceph_monc_get_version_async);
 938
 939static void handle_command_ack(struct ceph_mon_client *monc,
 940			       struct ceph_msg *msg)
 941{
 942	struct ceph_mon_generic_request *req;
 943	void *p = msg->front.iov_base;
 944	void *const end = p + msg->front_alloc_len;
 945	u64 tid = le64_to_cpu(msg->hdr.tid);
 946
 947	dout("%s msg %p tid %llu\n", __func__, msg, tid);
 948
 949	ceph_decode_need(&p, end, sizeof(struct ceph_mon_request_header) +
 950							    sizeof(u32), bad);
 951	p += sizeof(struct ceph_mon_request_header);
 952
 953	mutex_lock(&monc->mutex);
 954	req = lookup_generic_request(&monc->generic_request_tree, tid);
 955	if (!req) {
 956		mutex_unlock(&monc->mutex);
 957		return;
 958	}
 959
 960	req->result = ceph_decode_32(&p);
 961	__finish_generic_request(req);
 962	mutex_unlock(&monc->mutex);
 963
 964	complete_generic_request(req);
 965	return;
 966
 967bad:
 968	pr_err("corrupt mon_command ack, tid %llu\n", tid);
 969	ceph_msg_dump(msg);
 970}
 971
 972static __printf(2, 0)
 973int do_mon_command_vargs(struct ceph_mon_client *monc, const char *fmt,
 974			 va_list ap)
 975{
 976	struct ceph_mon_generic_request *req;
 977	struct ceph_mon_command *h;
 978	int ret = -ENOMEM;
 979	int len;
 980
 981	req = alloc_generic_request(monc, GFP_NOIO);
 982	if (!req)
 983		goto out;
 984
 985	req->request = ceph_msg_new(CEPH_MSG_MON_COMMAND, 256, GFP_NOIO, true);
 986	if (!req->request)
 987		goto out;
 988
 989	req->reply = ceph_msg_new(CEPH_MSG_MON_COMMAND_ACK, 512, GFP_NOIO,
 990				  true);
 991	if (!req->reply)
 992		goto out;
 993
 994	mutex_lock(&monc->mutex);
 995	register_generic_request(req);
 996	h = req->request->front.iov_base;
 997	h->monhdr.have_version = 0;
 998	h->monhdr.session_mon = cpu_to_le16(-1);
 999	h->monhdr.session_mon_tid = 0;
1000	h->fsid = monc->monmap->fsid;
1001	h->num_strs = cpu_to_le32(1);
1002	len = vsprintf(h->str, fmt, ap);
1003	h->str_len = cpu_to_le32(len);
1004	send_generic_request(monc, req);
1005	mutex_unlock(&monc->mutex);
1006
1007	ret = wait_generic_request(req);
1008out:
1009	put_generic_request(req);
1010	return ret;
1011}
1012
1013static __printf(2, 3)
1014int do_mon_command(struct ceph_mon_client *monc, const char *fmt, ...)
1015{
1016	va_list ap;
1017	int ret;
1018
1019	va_start(ap, fmt);
1020	ret = do_mon_command_vargs(monc, fmt, ap);
1021	va_end(ap);
1022	return ret;
1023}
1024
1025int ceph_monc_blocklist_add(struct ceph_mon_client *monc,
1026			    struct ceph_entity_addr *client_addr)
1027{
1028	int ret;
1029
1030	ret = do_mon_command(monc,
1031			     "{ \"prefix\": \"osd blocklist\", \
1032				\"blocklistop\": \"add\", \
1033				\"addr\": \"%pISpc/%u\" }",
1034			     &client_addr->in_addr,
1035			     le32_to_cpu(client_addr->nonce));
1036	if (ret == -EINVAL) {
1037		/*
1038		 * The monitor returns EINVAL on an unrecognized command.
1039		 * Try the legacy command -- it is exactly the same except
1040		 * for the name.
1041		 */
1042		ret = do_mon_command(monc,
1043				     "{ \"prefix\": \"osd blacklist\", \
1044					\"blacklistop\": \"add\", \
1045					\"addr\": \"%pISpc/%u\" }",
1046				     &client_addr->in_addr,
1047				     le32_to_cpu(client_addr->nonce));
1048	}
1049	if (ret)
1050		return ret;
1051
1052	/*
1053	 * Make sure we have the osdmap that includes the blocklist
1054	 * entry.  This is needed to ensure that the OSDs pick up the
1055	 * new blocklist before processing any future requests from
1056	 * this client.
1057	 */
1058	return ceph_wait_for_latest_osdmap(monc->client, 0);
1059}
1060EXPORT_SYMBOL(ceph_monc_blocklist_add);
1061
1062/*
1063 * Resend pending generic requests.
1064 */
1065static void __resend_generic_request(struct ceph_mon_client *monc)
1066{
1067	struct ceph_mon_generic_request *req;
1068	struct rb_node *p;
1069
1070	for (p = rb_first(&monc->generic_request_tree); p; p = rb_next(p)) {
1071		req = rb_entry(p, struct ceph_mon_generic_request, node);
1072		ceph_msg_revoke(req->request);
1073		ceph_msg_revoke_incoming(req->reply);
1074		ceph_con_send(&monc->con, ceph_msg_get(req->request));
1075	}
1076}
1077
1078/*
1079 * Delayed work.  If we haven't mounted yet, retry.  Otherwise,
1080 * renew/retry subscription as needed (in case it is timing out, or we
1081 * got an ENOMEM).  And keep the monitor connection alive.
1082 */
1083static void delayed_work(struct work_struct *work)
1084{
1085	struct ceph_mon_client *monc =
1086		container_of(work, struct ceph_mon_client, delayed_work.work);
1087
 
1088	mutex_lock(&monc->mutex);
1089	dout("%s mon%d\n", __func__, monc->cur_mon);
1090	if (monc->cur_mon < 0) {
1091		goto out;
1092	}
1093
1094	if (monc->hunting) {
1095		dout("%s continuing hunt\n", __func__);
1096		reopen_session(monc);
1097	} else {
1098		int is_auth = ceph_auth_is_authenticated(monc->auth);
1099
1100		dout("%s is_authed %d\n", __func__, is_auth);
1101		if (ceph_con_keepalive_expired(&monc->con,
1102					       CEPH_MONC_PING_TIMEOUT)) {
1103			dout("monc keepalive timeout\n");
1104			is_auth = 0;
1105			reopen_session(monc);
1106		}
1107
1108		if (!monc->hunting) {
1109			ceph_con_keepalive(&monc->con);
1110			__validate_auth(monc);
1111			un_backoff(monc);
1112		}
1113
1114		if (is_auth &&
1115		    !(monc->con.peer_features & CEPH_FEATURE_MON_STATEFUL_SUB)) {
1116			unsigned long now = jiffies;
1117
1118			dout("%s renew subs? now %lu renew after %lu\n",
1119			     __func__, now, monc->sub_renew_after);
1120			if (time_after_eq(now, monc->sub_renew_after))
1121				__send_subscribe(monc);
1122		}
1123	}
1124	__schedule_delayed(monc);
1125
1126out:
1127	mutex_unlock(&monc->mutex);
1128}
1129
1130/*
1131 * On startup, we build a temporary monmap populated with the IPs
1132 * provided by mount(2).
1133 */
1134static int build_initial_monmap(struct ceph_mon_client *monc)
1135{
1136	__le32 my_type = ceph_msgr2(monc->client) ?
1137		CEPH_ENTITY_ADDR_TYPE_MSGR2 : CEPH_ENTITY_ADDR_TYPE_LEGACY;
1138	struct ceph_options *opt = monc->client->options;
1139	int num_mon = opt->num_mon;
1140	int i;
1141
1142	/* build initial monmap */
1143	monc->monmap = kzalloc(struct_size(monc->monmap, mon_inst, num_mon),
1144			       GFP_KERNEL);
1145	if (!monc->monmap)
1146		return -ENOMEM;
1147	monc->monmap->num_mon = num_mon;
1148
1149	for (i = 0; i < num_mon; i++) {
1150		struct ceph_entity_inst *inst = &monc->monmap->mon_inst[i];
1151
1152		memcpy(&inst->addr.in_addr, &opt->mon_addr[i].in_addr,
1153		       sizeof(inst->addr.in_addr));
1154		inst->addr.type = my_type;
1155		inst->addr.nonce = 0;
1156		inst->name.type = CEPH_ENTITY_TYPE_MON;
1157		inst->name.num = cpu_to_le64(i);
1158	}
1159	return 0;
1160}
1161
1162int ceph_monc_init(struct ceph_mon_client *monc, struct ceph_client *cl)
1163{
1164	int err;
1165
1166	dout("init\n");
1167	memset(monc, 0, sizeof(*monc));
1168	monc->client = cl;
1169	mutex_init(&monc->mutex);
1170
1171	err = build_initial_monmap(monc);
1172	if (err)
1173		goto out;
1174
1175	/* connection */
1176	/* authentication */
1177	monc->auth = ceph_auth_init(cl->options->name, cl->options->key,
1178				    cl->options->con_modes);
1179	if (IS_ERR(monc->auth)) {
1180		err = PTR_ERR(monc->auth);
1181		goto out_monmap;
1182	}
1183	monc->auth->want_keys =
1184		CEPH_ENTITY_TYPE_AUTH | CEPH_ENTITY_TYPE_MON |
1185		CEPH_ENTITY_TYPE_OSD | CEPH_ENTITY_TYPE_MDS;
1186
1187	/* msgs */
1188	err = -ENOMEM;
1189	monc->m_subscribe_ack = ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE_ACK,
1190				     sizeof(struct ceph_mon_subscribe_ack),
1191				     GFP_KERNEL, true);
1192	if (!monc->m_subscribe_ack)
1193		goto out_auth;
1194
1195	monc->m_subscribe = ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE, 128,
1196					 GFP_KERNEL, true);
1197	if (!monc->m_subscribe)
1198		goto out_subscribe_ack;
1199
1200	monc->m_auth_reply = ceph_msg_new(CEPH_MSG_AUTH_REPLY, 4096,
1201					  GFP_KERNEL, true);
1202	if (!monc->m_auth_reply)
1203		goto out_subscribe;
1204
1205	monc->m_auth = ceph_msg_new(CEPH_MSG_AUTH, 4096, GFP_KERNEL, true);
1206	monc->pending_auth = 0;
1207	if (!monc->m_auth)
1208		goto out_auth_reply;
1209
1210	ceph_con_init(&monc->con, monc, &mon_con_ops,
1211		      &monc->client->msgr);
1212
1213	monc->cur_mon = -1;
1214	monc->had_a_connection = false;
1215	monc->hunt_mult = 1;
1216
1217	INIT_DELAYED_WORK(&monc->delayed_work, delayed_work);
1218	monc->generic_request_tree = RB_ROOT;
1219	monc->last_tid = 0;
1220
1221	monc->fs_cluster_id = CEPH_FS_CLUSTER_ID_NONE;
1222
1223	return 0;
1224
1225out_auth_reply:
1226	ceph_msg_put(monc->m_auth_reply);
1227out_subscribe:
1228	ceph_msg_put(monc->m_subscribe);
1229out_subscribe_ack:
1230	ceph_msg_put(monc->m_subscribe_ack);
1231out_auth:
1232	ceph_auth_destroy(monc->auth);
1233out_monmap:
1234	kfree(monc->monmap);
1235out:
1236	return err;
1237}
1238EXPORT_SYMBOL(ceph_monc_init);
1239
1240void ceph_monc_stop(struct ceph_mon_client *monc)
1241{
1242	dout("stop\n");
 
1243
1244	mutex_lock(&monc->mutex);
1245	__close_session(monc);
1246	monc->hunting = false;
1247	monc->cur_mon = -1;
1248	mutex_unlock(&monc->mutex);
1249
1250	cancel_delayed_work_sync(&monc->delayed_work);
1251
1252	/*
1253	 * flush msgr queue before we destroy ourselves to ensure that:
1254	 *  - any work that references our embedded con is finished.
1255	 *  - any osd_client or other work that may reference an authorizer
1256	 *    finishes before we shut down the auth subsystem.
1257	 */
1258	ceph_msgr_flush();
1259
1260	ceph_auth_destroy(monc->auth);
1261
1262	WARN_ON(!RB_EMPTY_ROOT(&monc->generic_request_tree));
1263
1264	ceph_msg_put(monc->m_auth);
1265	ceph_msg_put(monc->m_auth_reply);
1266	ceph_msg_put(monc->m_subscribe);
1267	ceph_msg_put(monc->m_subscribe_ack);
1268
1269	kfree(monc->monmap);
1270}
1271EXPORT_SYMBOL(ceph_monc_stop);
1272
1273static void finish_hunting(struct ceph_mon_client *monc)
1274{
1275	if (monc->hunting) {
1276		dout("%s found mon%d\n", __func__, monc->cur_mon);
1277		monc->hunting = false;
1278		monc->had_a_connection = true;
1279		un_backoff(monc);
1280		__schedule_delayed(monc);
1281	}
1282}
1283
1284static void finish_auth(struct ceph_mon_client *monc, int auth_err,
1285			bool was_authed)
1286{
1287	dout("%s auth_err %d was_authed %d\n", __func__, auth_err, was_authed);
1288	WARN_ON(auth_err > 0);
1289
1290	monc->pending_auth = 0;
1291	if (auth_err) {
1292		monc->client->auth_err = auth_err;
1293		wake_up_all(&monc->client->auth_wq);
1294		return;
1295	}
1296
1297	if (!was_authed && ceph_auth_is_authenticated(monc->auth)) {
1298		dout("%s authenticated, starting session global_id %llu\n",
1299		     __func__, monc->auth->global_id);
1300
1301		monc->client->msgr.inst.name.type = CEPH_ENTITY_TYPE_CLIENT;
1302		monc->client->msgr.inst.name.num =
1303					cpu_to_le64(monc->auth->global_id);
1304
1305		__send_subscribe(monc);
1306		__resend_generic_request(monc);
1307
1308		pr_info("mon%d %s session established\n", monc->cur_mon,
1309			ceph_pr_addr(&monc->con.peer_addr));
1310	}
1311}
1312
1313static void handle_auth_reply(struct ceph_mon_client *monc,
1314			      struct ceph_msg *msg)
1315{
1316	bool was_authed;
1317	int ret;
1318
1319	mutex_lock(&monc->mutex);
1320	was_authed = ceph_auth_is_authenticated(monc->auth);
1321	ret = ceph_handle_auth_reply(monc->auth, msg->front.iov_base,
1322				     msg->front.iov_len,
1323				     monc->m_auth->front.iov_base,
1324				     monc->m_auth->front_alloc_len);
1325	if (ret > 0) {
1326		__send_prepared_auth_request(monc, ret);
1327	} else {
1328		finish_auth(monc, ret, was_authed);
1329		finish_hunting(monc);
1330	}
1331	mutex_unlock(&monc->mutex);
1332}
1333
1334static int __validate_auth(struct ceph_mon_client *monc)
1335{
1336	int ret;
1337
1338	if (monc->pending_auth)
1339		return 0;
1340
1341	ret = ceph_build_auth(monc->auth, monc->m_auth->front.iov_base,
1342			      monc->m_auth->front_alloc_len);
1343	if (ret <= 0)
1344		return ret; /* either an error, or no need to authenticate */
1345	__send_prepared_auth_request(monc, ret);
1346	return 0;
1347}
1348
1349int ceph_monc_validate_auth(struct ceph_mon_client *monc)
1350{
1351	int ret;
1352
1353	mutex_lock(&monc->mutex);
1354	ret = __validate_auth(monc);
1355	mutex_unlock(&monc->mutex);
1356	return ret;
1357}
1358EXPORT_SYMBOL(ceph_monc_validate_auth);
1359
1360static int mon_get_auth_request(struct ceph_connection *con,
1361				void *buf, int *buf_len,
1362				void **authorizer, int *authorizer_len)
1363{
1364	struct ceph_mon_client *monc = con->private;
1365	int ret;
1366
1367	mutex_lock(&monc->mutex);
1368	ret = ceph_auth_get_request(monc->auth, buf, *buf_len);
1369	mutex_unlock(&monc->mutex);
1370	if (ret < 0)
1371		return ret;
1372
1373	*buf_len = ret;
1374	*authorizer = NULL;
1375	*authorizer_len = 0;
1376	return 0;
1377}
1378
1379static int mon_handle_auth_reply_more(struct ceph_connection *con,
1380				      void *reply, int reply_len,
1381				      void *buf, int *buf_len,
1382				      void **authorizer, int *authorizer_len)
1383{
1384	struct ceph_mon_client *monc = con->private;
1385	int ret;
1386
1387	mutex_lock(&monc->mutex);
1388	ret = ceph_auth_handle_reply_more(monc->auth, reply, reply_len,
1389					  buf, *buf_len);
1390	mutex_unlock(&monc->mutex);
1391	if (ret < 0)
1392		return ret;
1393
1394	*buf_len = ret;
1395	*authorizer = NULL;
1396	*authorizer_len = 0;
1397	return 0;
1398}
1399
1400static int mon_handle_auth_done(struct ceph_connection *con,
1401				u64 global_id, void *reply, int reply_len,
1402				u8 *session_key, int *session_key_len,
1403				u8 *con_secret, int *con_secret_len)
1404{
1405	struct ceph_mon_client *monc = con->private;
1406	bool was_authed;
1407	int ret;
1408
1409	mutex_lock(&monc->mutex);
1410	WARN_ON(!monc->hunting);
1411	was_authed = ceph_auth_is_authenticated(monc->auth);
1412	ret = ceph_auth_handle_reply_done(monc->auth, global_id,
1413					  reply, reply_len,
1414					  session_key, session_key_len,
1415					  con_secret, con_secret_len);
1416	finish_auth(monc, ret, was_authed);
1417	if (!ret)
1418		finish_hunting(monc);
1419	mutex_unlock(&monc->mutex);
1420	return 0;
1421}
1422
1423static int mon_handle_auth_bad_method(struct ceph_connection *con,
1424				      int used_proto, int result,
1425				      const int *allowed_protos, int proto_cnt,
1426				      const int *allowed_modes, int mode_cnt)
1427{
1428	struct ceph_mon_client *monc = con->private;
1429	bool was_authed;
1430
1431	mutex_lock(&monc->mutex);
1432	WARN_ON(!monc->hunting);
1433	was_authed = ceph_auth_is_authenticated(monc->auth);
1434	ceph_auth_handle_bad_method(monc->auth, used_proto, result,
1435				    allowed_protos, proto_cnt,
1436				    allowed_modes, mode_cnt);
1437	finish_auth(monc, -EACCES, was_authed);
1438	mutex_unlock(&monc->mutex);
1439	return 0;
1440}
1441
1442/*
1443 * handle incoming message
1444 */
1445static void mon_dispatch(struct ceph_connection *con, struct ceph_msg *msg)
1446{
1447	struct ceph_mon_client *monc = con->private;
1448	int type = le16_to_cpu(msg->hdr.type);
1449
1450	switch (type) {
1451	case CEPH_MSG_AUTH_REPLY:
1452		handle_auth_reply(monc, msg);
1453		break;
1454
1455	case CEPH_MSG_MON_SUBSCRIBE_ACK:
1456		handle_subscribe_ack(monc, msg);
1457		break;
1458
1459	case CEPH_MSG_STATFS_REPLY:
1460		handle_statfs_reply(monc, msg);
1461		break;
1462
1463	case CEPH_MSG_MON_GET_VERSION_REPLY:
1464		handle_get_version_reply(monc, msg);
1465		break;
1466
1467	case CEPH_MSG_MON_COMMAND_ACK:
1468		handle_command_ack(monc, msg);
1469		break;
1470
1471	case CEPH_MSG_MON_MAP:
1472		ceph_monc_handle_map(monc, msg);
1473		break;
1474
1475	case CEPH_MSG_OSD_MAP:
1476		ceph_osdc_handle_map(&monc->client->osdc, msg);
1477		break;
1478
1479	default:
1480		/* can the chained handler handle it? */
1481		if (monc->client->extra_mon_dispatch &&
1482		    monc->client->extra_mon_dispatch(monc->client, msg) == 0)
1483			break;
1484
1485		pr_err("received unknown message type %d %s\n", type,
1486		       ceph_msg_type_name(type));
1487	}
1488	ceph_msg_put(msg);
1489}
1490
1491/*
1492 * Allocate memory for incoming message
1493 */
1494static struct ceph_msg *mon_alloc_msg(struct ceph_connection *con,
1495				      struct ceph_msg_header *hdr,
1496				      int *skip)
1497{
1498	struct ceph_mon_client *monc = con->private;
1499	int type = le16_to_cpu(hdr->type);
1500	int front_len = le32_to_cpu(hdr->front_len);
1501	struct ceph_msg *m = NULL;
1502
1503	*skip = 0;
1504
1505	switch (type) {
1506	case CEPH_MSG_MON_SUBSCRIBE_ACK:
1507		m = ceph_msg_get(monc->m_subscribe_ack);
1508		break;
1509	case CEPH_MSG_STATFS_REPLY:
1510	case CEPH_MSG_MON_COMMAND_ACK:
1511		return get_generic_reply(con, hdr, skip);
1512	case CEPH_MSG_AUTH_REPLY:
1513		m = ceph_msg_get(monc->m_auth_reply);
1514		break;
1515	case CEPH_MSG_MON_GET_VERSION_REPLY:
1516		if (le64_to_cpu(hdr->tid) != 0)
1517			return get_generic_reply(con, hdr, skip);
1518
1519		/*
1520		 * Older OSDs don't set reply tid even if the original
1521		 * request had a non-zero tid.  Work around this weirdness
1522		 * by allocating a new message.
1523		 */
1524		fallthrough;
1525	case CEPH_MSG_MON_MAP:
1526	case CEPH_MSG_MDS_MAP:
1527	case CEPH_MSG_OSD_MAP:
1528	case CEPH_MSG_FS_MAP_USER:
1529		m = ceph_msg_new(type, front_len, GFP_NOFS, false);
1530		if (!m)
1531			return NULL;	/* ENOMEM--return skip == 0 */
1532		break;
1533	}
1534
1535	if (!m) {
1536		pr_info("alloc_msg unknown type %d\n", type);
1537		*skip = 1;
1538	} else if (front_len > m->front_alloc_len) {
1539		pr_warn("mon_alloc_msg front %d > prealloc %d (%u#%llu)\n",
1540			front_len, m->front_alloc_len,
1541			(unsigned int)con->peer_name.type,
1542			le64_to_cpu(con->peer_name.num));
1543		ceph_msg_put(m);
1544		m = ceph_msg_new(type, front_len, GFP_NOFS, false);
1545	}
1546
1547	return m;
1548}
1549
1550/*
1551 * If the monitor connection resets, pick a new monitor and resubmit
1552 * any pending requests.
1553 */
1554static void mon_fault(struct ceph_connection *con)
1555{
1556	struct ceph_mon_client *monc = con->private;
1557
1558	mutex_lock(&monc->mutex);
1559	dout("%s mon%d\n", __func__, monc->cur_mon);
1560	if (monc->cur_mon >= 0) {
1561		if (!monc->hunting) {
1562			dout("%s hunting for new mon\n", __func__);
1563			reopen_session(monc);
1564			__schedule_delayed(monc);
1565		} else {
1566			dout("%s already hunting\n", __func__);
1567		}
1568	}
1569	mutex_unlock(&monc->mutex);
1570}
1571
1572/*
1573 * We can ignore refcounting on the connection struct, as all references
1574 * will come from the messenger workqueue, which is drained prior to
1575 * mon_client destruction.
1576 */
1577static struct ceph_connection *mon_get_con(struct ceph_connection *con)
1578{
1579	return con;
1580}
1581
1582static void mon_put_con(struct ceph_connection *con)
1583{
1584}
1585
1586static const struct ceph_connection_operations mon_con_ops = {
1587	.get = mon_get_con,
1588	.put = mon_put_con,
1589	.alloc_msg = mon_alloc_msg,
1590	.dispatch = mon_dispatch,
1591	.fault = mon_fault,
1592	.get_auth_request = mon_get_auth_request,
1593	.handle_auth_reply_more = mon_handle_auth_reply_more,
1594	.handle_auth_done = mon_handle_auth_done,
1595	.handle_auth_bad_method = mon_handle_auth_bad_method,
1596};