messenger.c 85.1 KB
Newer Older
1
#include <linux/ceph/ceph_debug.h>
Sage Weil's avatar
Sage Weil committed
2 3 4 5 6 7 8

#include <linux/crc32c.h>
#include <linux/ctype.h>
#include <linux/highmem.h>
#include <linux/inet.h>
#include <linux/kthread.h>
#include <linux/net.h>
9
#include <linux/nsproxy.h>
10
#include <linux/slab.h>
Sage Weil's avatar
Sage Weil committed
11 12
#include <linux/socket.h>
#include <linux/string.h>
13
#ifdef	CONFIG_BLOCK
14
#include <linux/bio.h>
15
#endif	/* CONFIG_BLOCK */
Noah Watkins's avatar
Noah Watkins committed
16
#include <linux/dns_resolver.h>
Sage Weil's avatar
Sage Weil committed
17 18
#include <net/tcp.h>

Ilya Dryomov's avatar
Ilya Dryomov committed
19
#include <linux/ceph/ceph_features.h>
20 21 22 23
#include <linux/ceph/libceph.h>
#include <linux/ceph/messenger.h>
#include <linux/ceph/decode.h>
#include <linux/ceph/pagelist.h>
24
#include <linux/export.h>
Sage Weil's avatar
Sage Weil committed
25 26 27 28 29 30 31 32 33 34

/*
 * Ceph uses the messenger to exchange ceph_msg messages with other
 * hosts in the system.  The messenger provides ordered and reliable
 * delivery.  We tolerate TCP disconnects by reconnecting (with
 * exponential backoff) in the case of a fault (disconnection, bad
 * crc, protocol error).  Acks allow sent messages to be discarded by
 * the sender.
 */

Alex Elder's avatar
Alex Elder committed
35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53
/*
 * We track the state of the socket on a given connection using
 * values defined below.  The transition to a new socket state is
 * handled by a function which verifies we aren't coming from an
 * unexpected state.
 *
 *      --------
 *      | NEW* |  transient initial state
 *      --------
 *          | con_sock_state_init()
 *          v
 *      ----------
 *      | CLOSED |  initialized, but no socket (and no
 *      ----------  TCP connection)
 *       ^      \
 *       |       \ con_sock_state_connecting()
 *       |        ----------------------
 *       |                              \
 *       + con_sock_state_closed()       \
54 55 56 57 58 59 60 61 62 63 64
 *       |+---------------------------    \
 *       | \                          \    \
 *       |  -----------                \    \
 *       |  | CLOSING |  socket event;  \    \
 *       |  -----------  await close     \    \
 *       |       ^                        \   |
 *       |       |                         \  |
 *       |       + con_sock_state_closing() \ |
 *       |      / \                         | |
 *       |     /   ---------------          | |
 *       |    /                   \         v v
Alex Elder's avatar
Alex Elder committed
65 66 67 68 69 70 71 72 73 74 75
 *       |   /                    --------------
 *       |  /    -----------------| CONNECTING |  socket created, TCP
 *       |  |   /                 --------------  connect initiated
 *       |  |   | con_sock_state_connected()
 *       |  |   v
 *      -------------
 *      | CONNECTED |  TCP connection established
 *      -------------
 *
 * State values for ceph_connection->sock_state; NEW is assumed to be 0.
 */
76 77 78 79 80 81 82

#define CON_SOCK_STATE_NEW		0	/* -> CLOSED */
#define CON_SOCK_STATE_CLOSED		1	/* -> CONNECTING */
#define CON_SOCK_STATE_CONNECTING	2	/* -> CONNECTED or -> CLOSING */
#define CON_SOCK_STATE_CONNECTED	3	/* -> CLOSING or -> CLOSED */
#define CON_SOCK_STATE_CLOSING		4	/* -> CLOSED */

83 84 85 86 87 88 89 90 91 92
/*
 * connection states
 */
#define CON_STATE_CLOSED        1  /* -> PREOPEN */
#define CON_STATE_PREOPEN       2  /* -> CONNECTING, CLOSED */
#define CON_STATE_CONNECTING    3  /* -> NEGOTIATING, CLOSED */
#define CON_STATE_NEGOTIATING   4  /* -> OPEN, CLOSED */
#define CON_STATE_OPEN          5  /* -> STANDBY, CLOSED */
#define CON_STATE_STANDBY       6  /* -> PREOPEN, CLOSED */

Sage Weil's avatar
Sage Weil committed
93 94 95 96 97 98 99 100 101
/*
 * ceph_connection flag bits
 */
#define CON_FLAG_LOSSYTX           0  /* we can close channel or drop
				       * messages on errors */
#define CON_FLAG_KEEPALIVE_PENDING 1  /* we need to send a keepalive */
#define CON_FLAG_WRITE_PENDING	   2  /* we have data ready to send */
#define CON_FLAG_SOCK_CLOSED	   3  /* socket state changed to closed */
#define CON_FLAG_BACKOFF           4  /* need to retry queuing delayed work */
102

103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153
static bool con_flag_valid(unsigned long con_flag)
{
	switch (con_flag) {
	case CON_FLAG_LOSSYTX:
	case CON_FLAG_KEEPALIVE_PENDING:
	case CON_FLAG_WRITE_PENDING:
	case CON_FLAG_SOCK_CLOSED:
	case CON_FLAG_BACKOFF:
		return true;
	default:
		return false;
	}
}

static void con_flag_clear(struct ceph_connection *con, unsigned long con_flag)
{
	BUG_ON(!con_flag_valid(con_flag));

	clear_bit(con_flag, &con->flags);
}

static void con_flag_set(struct ceph_connection *con, unsigned long con_flag)
{
	BUG_ON(!con_flag_valid(con_flag));

	set_bit(con_flag, &con->flags);
}

static bool con_flag_test(struct ceph_connection *con, unsigned long con_flag)
{
	BUG_ON(!con_flag_valid(con_flag));

	return test_bit(con_flag, &con->flags);
}

static bool con_flag_test_and_clear(struct ceph_connection *con,
					unsigned long con_flag)
{
	BUG_ON(!con_flag_valid(con_flag));

	return test_and_clear_bit(con_flag, &con->flags);
}

static bool con_flag_test_and_set(struct ceph_connection *con,
					unsigned long con_flag)
{
	BUG_ON(!con_flag_valid(con_flag));

	return test_and_set_bit(con_flag, &con->flags);
}

154 155 156
/* Slab caches for frequently-allocated structures */

static struct kmem_cache	*ceph_msg_cache;
157
static struct kmem_cache	*ceph_msg_data_cache;
158

Sage Weil's avatar
Sage Weil committed
159 160 161 162
/* static tag bytes (protocol control messages) */
static char tag_msg = CEPH_MSGR_TAG_MSG;
static char tag_ack = CEPH_MSGR_TAG_ACK;
static char tag_keepalive = CEPH_MSGR_TAG_KEEPALIVE;
163
static char tag_keepalive2 = CEPH_MSGR_TAG_KEEPALIVE2;
Sage Weil's avatar
Sage Weil committed
164

165 166 167 168
#ifdef CONFIG_LOCKDEP
static struct lock_class_key socket_class;
#endif

Alex Elder's avatar
Alex Elder committed
169 170 171 172 173
/*
 * When skipping (ignoring) a block of input we read it into a "skip
 * buffer," which is this many bytes in size.
 */
#define SKIP_BUF_SIZE	1024
Sage Weil's avatar
Sage Weil committed
174 175

static void queue_con(struct ceph_connection *con);
176
static void cancel_con(struct ceph_connection *con);
177
static void ceph_con_workfn(struct work_struct *);
178
static void con_fault(struct ceph_connection *con);
Sage Weil's avatar
Sage Weil committed
179 180

/*
181 182
 * Nicely render a sockaddr as a string.  An array of formatted
 * strings is used, to approximate reentrancy.
Sage Weil's avatar
Sage Weil committed
183
 */
184 185 186 187 188 189 190
#define ADDR_STR_COUNT_LOG	5	/* log2(# address strings in array) */
#define ADDR_STR_COUNT		(1 << ADDR_STR_COUNT_LOG)
#define ADDR_STR_COUNT_MASK	(ADDR_STR_COUNT - 1)
#define MAX_ADDR_STR_LEN	64	/* 54 is enough */

static char addr_str[ADDR_STR_COUNT][MAX_ADDR_STR_LEN];
static atomic_t addr_str_seq = ATOMIC_INIT(0);
Sage Weil's avatar
Sage Weil committed
191

192 193
static struct page *zero_page;		/* used in certain error cases */

194
const char *ceph_pr_addr(const struct sockaddr_storage *ss)
Sage Weil's avatar
Sage Weil committed
195 196 197
{
	int i;
	char *s;
198 199
	struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
	struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
Sage Weil's avatar
Sage Weil committed
200

201
	i = atomic_inc_return(&addr_str_seq) & ADDR_STR_COUNT_MASK;
Sage Weil's avatar
Sage Weil committed
202 203 204 205
	s = addr_str[i];

	switch (ss->ss_family) {
	case AF_INET:
206 207
		snprintf(s, MAX_ADDR_STR_LEN, "%pI4:%hu", &in4->sin_addr,
			 ntohs(in4->sin_port));
Sage Weil's avatar
Sage Weil committed
208 209 210
		break;

	case AF_INET6:
211 212
		snprintf(s, MAX_ADDR_STR_LEN, "[%pI6c]:%hu", &in6->sin6_addr,
			 ntohs(in6->sin6_port));
Sage Weil's avatar
Sage Weil committed
213 214 215
		break;

	default:
Alex Elder's avatar
Alex Elder committed
216 217
		snprintf(s, MAX_ADDR_STR_LEN, "(unknown sockaddr family %hu)",
			 ss->ss_family);
Sage Weil's avatar
Sage Weil committed
218 219 220 221
	}

	return s;
}
222
EXPORT_SYMBOL(ceph_pr_addr);
Sage Weil's avatar
Sage Weil committed
223

224 225 226 227 228 229
static void encode_my_addr(struct ceph_messenger *msgr)
{
	memcpy(&msgr->my_enc_addr, &msgr->inst.addr, sizeof(msgr->my_enc_addr));
	ceph_encode_addr(&msgr->my_enc_addr);
}

Sage Weil's avatar
Sage Weil committed
230 231 232
/*
 * work queue for all reading and writing to/from the socket.
 */
233
static struct workqueue_struct *ceph_msgr_wq;
Sage Weil's avatar
Sage Weil committed
234

235 236 237
static int ceph_msgr_slab_init(void)
{
	BUG_ON(ceph_msg_cache);
Geliang Tang's avatar
Geliang Tang committed
238
	ceph_msg_cache = KMEM_CACHE(ceph_msg, 0);
239 240 241 242
	if (!ceph_msg_cache)
		return -ENOMEM;

	BUG_ON(ceph_msg_data_cache);
Geliang Tang's avatar
Geliang Tang committed
243
	ceph_msg_data_cache = KMEM_CACHE(ceph_msg_data, 0);
244 245 246 247 248 249 250
	if (ceph_msg_data_cache)
		return 0;

	kmem_cache_destroy(ceph_msg_cache);
	ceph_msg_cache = NULL;

	return -ENOMEM;
251 252 253 254
}

static void ceph_msgr_slab_exit(void)
{
255 256 257 258
	BUG_ON(!ceph_msg_data_cache);
	kmem_cache_destroy(ceph_msg_data_cache);
	ceph_msg_data_cache = NULL;

259 260 261 262 263
	BUG_ON(!ceph_msg_cache);
	kmem_cache_destroy(ceph_msg_cache);
	ceph_msg_cache = NULL;
}

264
static void _ceph_msgr_exit(void)
265
{
Alex Elder's avatar
Alex Elder committed
266
	if (ceph_msgr_wq) {
267
		destroy_workqueue(ceph_msgr_wq);
Alex Elder's avatar
Alex Elder committed
268 269
		ceph_msgr_wq = NULL;
	}
270 271 272 273

	BUG_ON(zero_page == NULL);
	page_cache_release(zero_page);
	zero_page = NULL;
274 275

	ceph_msgr_slab_exit();
276 277
}

278
int ceph_msgr_init(void)
Sage Weil's avatar
Sage Weil committed
279
{
280 281 282
	if (ceph_msgr_slab_init())
		return -ENOMEM;

283 284 285 286
	BUG_ON(zero_page != NULL);
	zero_page = ZERO_PAGE(0);
	page_cache_get(zero_page);

287 288 289 290 291
	/*
	 * The number of active work items is limited by the number of
	 * connections, so leave @max_active at default.
	 */
	ceph_msgr_wq = alloc_workqueue("ceph-msgr", WQ_MEM_RECLAIM, 0);
292 293
	if (ceph_msgr_wq)
		return 0;
294

295 296
	pr_err("msgr_init failed to create workqueue\n");
	_ceph_msgr_exit();
297

298
	return -ENOMEM;
Sage Weil's avatar
Sage Weil committed
299
}
300
EXPORT_SYMBOL(ceph_msgr_init);
Sage Weil's avatar
Sage Weil committed
301 302 303

void ceph_msgr_exit(void)
{
304 305
	BUG_ON(ceph_msgr_wq == NULL);

306
	_ceph_msgr_exit();
Sage Weil's avatar
Sage Weil committed
307
}
308
EXPORT_SYMBOL(ceph_msgr_exit);
Sage Weil's avatar
Sage Weil committed
309

Yehuda Sadeh's avatar
Yehuda Sadeh committed
310
void ceph_msgr_flush(void)
311 312 313
{
	flush_workqueue(ceph_msgr_wq);
}
314
EXPORT_SYMBOL(ceph_msgr_flush);
315

316 317 318 319 320 321 322 323 324
/* Connection socket state transition functions */

static void con_sock_state_init(struct ceph_connection *con)
{
	int old_state;

	old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CLOSED);
	if (WARN_ON(old_state != CON_SOCK_STATE_NEW))
		printk("%s: unexpected old state %d\n", __func__, old_state);
325 326
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CLOSED);
327 328 329 330 331 332 333 334 335
}

static void con_sock_state_connecting(struct ceph_connection *con)
{
	int old_state;

	old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CONNECTING);
	if (WARN_ON(old_state != CON_SOCK_STATE_CLOSED))
		printk("%s: unexpected old state %d\n", __func__, old_state);
336 337
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CONNECTING);
338 339 340 341 342 343 344 345 346
}

static void con_sock_state_connected(struct ceph_connection *con)
{
	int old_state;

	old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CONNECTED);
	if (WARN_ON(old_state != CON_SOCK_STATE_CONNECTING))
		printk("%s: unexpected old state %d\n", __func__, old_state);
347 348
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CONNECTED);
349 350 351 352 353 354 355 356 357 358 359
}

static void con_sock_state_closing(struct ceph_connection *con)
{
	int old_state;

	old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CLOSING);
	if (WARN_ON(old_state != CON_SOCK_STATE_CONNECTING &&
			old_state != CON_SOCK_STATE_CONNECTED &&
			old_state != CON_SOCK_STATE_CLOSING))
		printk("%s: unexpected old state %d\n", __func__, old_state);
360 361
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CLOSING);
362 363 364 365 366 367 368 369
}

static void con_sock_state_closed(struct ceph_connection *con)
{
	int old_state;

	old_state = atomic_xchg(&con->sock_state, CON_SOCK_STATE_CLOSED);
	if (WARN_ON(old_state != CON_SOCK_STATE_CONNECTED &&
370
		    old_state != CON_SOCK_STATE_CLOSING &&
371 372
		    old_state != CON_SOCK_STATE_CONNECTING &&
		    old_state != CON_SOCK_STATE_CLOSED))
373
		printk("%s: unexpected old state %d\n", __func__, old_state);
374 375
	dout("%s con %p sock %d -> %d\n", __func__, con, old_state,
	     CON_SOCK_STATE_CLOSED);
376
}
377

Sage Weil's avatar
Sage Weil committed
378 379 380 381 382
/*
 * socket callback functions
 */

/* data available on socket, or listen socket received a connect */
383
static void ceph_sock_data_ready(struct sock *sk)
Sage Weil's avatar
Sage Weil committed
384
{
385
	struct ceph_connection *con = sk->sk_user_data;
386 387 388
	if (atomic_read(&con->msgr->stopping)) {
		return;
	}
389

Sage Weil's avatar
Sage Weil committed
390
	if (sk->sk_state != TCP_CLOSE_WAIT) {
Alex Elder's avatar
Alex Elder committed
391
		dout("%s on %p state = %lu, queueing work\n", __func__,
Sage Weil's avatar
Sage Weil committed
392 393 394 395 396 397
		     con, con->state);
		queue_con(con);
	}
}

/* socket has buffer space for writing */
Alex Elder's avatar
Alex Elder committed
398
static void ceph_sock_write_space(struct sock *sk)
Sage Weil's avatar
Sage Weil committed
399
{
Alex Elder's avatar
Alex Elder committed
400
	struct ceph_connection *con = sk->sk_user_data;
Sage Weil's avatar
Sage Weil committed
401

402 403
	/* only queue to workqueue if there is data we want to write,
	 * and there is sufficient space in the socket buffer to accept
Alex Elder's avatar
Alex Elder committed
404
	 * more data.  clear SOCK_NOSPACE so that ceph_sock_write_space()
405 406 407 408
	 * doesn't get called again until try_write() fills the socket
	 * buffer. See net/ipv4/tcp_input.c:tcp_check_space()
	 * and net/core/stream.c:sk_stream_write_space().
	 */
409
	if (con_flag_test(con, CON_FLAG_WRITE_PENDING)) {
410
		if (sk_stream_is_writeable(sk)) {
Alex Elder's avatar
Alex Elder committed
411
			dout("%s %p queueing write work\n", __func__, con);
412 413 414
			clear_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
			queue_con(con);
		}
Sage Weil's avatar
Sage Weil committed
415
	} else {
Alex Elder's avatar
Alex Elder committed
416
		dout("%s %p nothing to write\n", __func__, con);
Sage Weil's avatar
Sage Weil committed
417 418 419 420
	}
}

/* socket's state has changed */
Alex Elder's avatar
Alex Elder committed
421
static void ceph_sock_state_change(struct sock *sk)
Sage Weil's avatar
Sage Weil committed
422
{
423
	struct ceph_connection *con = sk->sk_user_data;
Sage Weil's avatar
Sage Weil committed
424

Alex Elder's avatar
Alex Elder committed
425
	dout("%s %p state = %lu sk_state = %u\n", __func__,
Sage Weil's avatar
Sage Weil committed
426 427 428 429
	     con, con->state, sk->sk_state);

	switch (sk->sk_state) {
	case TCP_CLOSE:
Alex Elder's avatar
Alex Elder committed
430
		dout("%s TCP_CLOSE\n", __func__);
Sage Weil's avatar
Sage Weil committed
431
	case TCP_CLOSE_WAIT:
Alex Elder's avatar
Alex Elder committed
432
		dout("%s TCP_CLOSE_WAIT\n", __func__);
433
		con_sock_state_closing(con);
434
		con_flag_set(con, CON_FLAG_SOCK_CLOSED);
435
		queue_con(con);
Sage Weil's avatar
Sage Weil committed
436 437
		break;
	case TCP_ESTABLISHED:
Alex Elder's avatar
Alex Elder committed
438
		dout("%s TCP_ESTABLISHED\n", __func__);
439
		con_sock_state_connected(con);
Sage Weil's avatar
Sage Weil committed
440 441
		queue_con(con);
		break;
Alex Elder's avatar
Alex Elder committed
442 443
	default:	/* Everything else is uninteresting */
		break;
Sage Weil's avatar
Sage Weil committed
444 445 446 447 448 449 450 451 452 453
	}
}

/*
 * set up socket callbacks
 */
static void set_sock_callbacks(struct socket *sock,
			       struct ceph_connection *con)
{
	struct sock *sk = sock->sk;
454
	sk->sk_user_data = con;
Alex Elder's avatar
Alex Elder committed
455 456 457
	sk->sk_data_ready = ceph_sock_data_ready;
	sk->sk_write_space = ceph_sock_write_space;
	sk->sk_state_change = ceph_sock_state_change;
Sage Weil's avatar
Sage Weil committed
458 459 460 461 462 463 464 465 466 467
}


/*
 * socket helpers
 */

/*
 * initiate connection to a remote socket.
 */
468
static int ceph_tcp_connect(struct ceph_connection *con)
Sage Weil's avatar
Sage Weil committed
469
{
Sage Weil's avatar
Sage Weil committed
470
	struct sockaddr_storage *paddr = &con->peer_addr.in_addr;
Sage Weil's avatar
Sage Weil committed
471 472 473 474
	struct socket *sock;
	int ret;

	BUG_ON(con->sock);
475
	ret = sock_create_kern(read_pnet(&con->msgr->net), paddr->ss_family,
476
			       SOCK_STREAM, IPPROTO_TCP, &sock);
Sage Weil's avatar
Sage Weil committed
477
	if (ret)
478
		return ret;
479
	sock->sk->sk_allocation = GFP_NOFS;
Sage Weil's avatar
Sage Weil committed
480

481 482 483 484
#ifdef CONFIG_LOCKDEP
	lockdep_set_class(&sock->sk->sk_lock, &socket_class);
#endif

Sage Weil's avatar
Sage Weil committed
485 486
	set_sock_callbacks(sock, con);

487
	dout("connect %s\n", ceph_pr_addr(&con->peer_addr.in_addr));
Sage Weil's avatar
Sage Weil committed
488

489
	con_sock_state_connecting(con);
Sage Weil's avatar
Sage Weil committed
490 491
	ret = sock->ops->connect(sock, (struct sockaddr *)paddr, sizeof(*paddr),
				 O_NONBLOCK);
Sage Weil's avatar
Sage Weil committed
492 493
	if (ret == -EINPROGRESS) {
		dout("connect %s EINPROGRESS sk_state = %u\n",
494
		     ceph_pr_addr(&con->peer_addr.in_addr),
Sage Weil's avatar
Sage Weil committed
495
		     sock->sk->sk_state);
496
	} else if (ret < 0) {
Sage Weil's avatar
Sage Weil committed
497
		pr_err("connect %s error %d\n",
498
		       ceph_pr_addr(&con->peer_addr.in_addr), ret);
Sage Weil's avatar
Sage Weil committed
499
		sock_release(sock);
500
		return ret;
501
	}
502

503
	if (ceph_test_opt(from_msgr(con->msgr), TCP_NODELAY)) {
504 505 506 507 508 509 510 511 512
		int optval = 1;

		ret = kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY,
					(char *)&optval, sizeof(optval));
		if (ret)
			pr_err("kernel_setsockopt(TCP_NODELAY) failed: %d",
			       ret);
	}

513
	con->sock = sock;
514
	return 0;
Sage Weil's avatar
Sage Weil committed
515 516 517 518 519 520
}

static int ceph_tcp_recvmsg(struct socket *sock, void *buf, size_t len)
{
	struct kvec iov = {buf, len};
	struct msghdr msg = { .msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL };
521
	int r;
Sage Weil's avatar
Sage Weil committed
522

523 524 525 526
	r = kernel_recvmsg(sock, &msg, &iov, 1, len, msg.msg_flags);
	if (r == -EAGAIN)
		r = 0;
	return r;
Sage Weil's avatar
Sage Weil committed
527 528
}

529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544
static int ceph_tcp_recvpage(struct socket *sock, struct page *page,
		     int page_offset, size_t length)
{
	void *kaddr;
	int ret;

	BUG_ON(page_offset + length > PAGE_SIZE);

	kaddr = kmap(page);
	BUG_ON(!kaddr);
	ret = ceph_tcp_recvmsg(sock, kaddr + page_offset, length);
	kunmap(page);

	return ret;
}

Sage Weil's avatar
Sage Weil committed
545 546 547 548 549 550 551 552
/*
 * write something.  @more is true if caller will be sending more data
 * shortly.
 */
static int ceph_tcp_sendmsg(struct socket *sock, struct kvec *iov,
		     size_t kvlen, size_t len, int more)
{
	struct msghdr msg = { .msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL };
553
	int r;
Sage Weil's avatar
Sage Weil committed
554 555 556 557 558 559

	if (more)
		msg.msg_flags |= MSG_MORE;
	else
		msg.msg_flags |= MSG_EOR;  /* superfluous, but what the hell */

560 561 562 563
	r = kernel_sendmsg(sock, &msg, iov, kvlen, len);
	if (r == -EAGAIN)
		r = 0;
	return r;
Sage Weil's avatar
Sage Weil committed
564 565
}

566
static int __ceph_tcp_sendpage(struct socket *sock, struct page *page,
567
		     int offset, size_t size, bool more)
568 569 570 571 572 573 574 575 576 577 578
{
	int flags = MSG_DONTWAIT | MSG_NOSIGNAL | (more ? MSG_MORE : MSG_EOR);
	int ret;

	ret = kernel_sendpage(sock, page, offset, size, flags);
	if (ret == -EAGAIN)
		ret = 0;

	return ret;
}

579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596
static int ceph_tcp_sendpage(struct socket *sock, struct page *page,
		     int offset, size_t size, bool more)
{
	int ret;
	struct kvec iov;

	/* sendpage cannot properly handle pages with page_count == 0,
	 * we need to fallback to sendmsg if that's the case */
	if (page_count(page) >= 1)
		return __ceph_tcp_sendpage(sock, page, offset, size, more);

	iov.iov_base = kmap(page) + offset;
	iov.iov_len = size;
	ret = ceph_tcp_sendmsg(sock, &iov, 1, size, more);
	kunmap(page);

	return ret;
}
Sage Weil's avatar
Sage Weil committed
597 598 599 600 601 602

/*
 * Shutdown/close the socket for the given connection.
 */
static int con_close_socket(struct ceph_connection *con)
{
603
	int rc = 0;
Sage Weil's avatar
Sage Weil committed
604 605

	dout("con_close_socket on %p sock %p\n", con, con->sock);
606 607 608 609 610
	if (con->sock) {
		rc = con->sock->ops->shutdown(con->sock, SHUT_RDWR);
		sock_release(con->sock);
		con->sock = NULL;
	}
611 612

	/*
Sage Weil's avatar
Sage Weil committed
613
	 * Forcibly clear the SOCK_CLOSED flag.  It gets set
614 615 616 617
	 * independent of the connection mutex, and we could have
	 * received a socket close event before we had the chance to
	 * shut the socket down.
	 */
618
	con_flag_clear(con, CON_FLAG_SOCK_CLOSED);
619

620
	con_sock_state_closed(con);
Sage Weil's avatar
Sage Weil committed
621 622 623 624 625 626 627 628 629 630
	return rc;
}

/*
 * Reset a connection.  Discard all incoming and outgoing messages
 * and clear *_seq state.
 */
static void ceph_msg_remove(struct ceph_msg *msg)
{
	list_del_init(&msg->list_head);
631

Sage Weil's avatar
Sage Weil committed
632 633 634 635 636 637 638 639 640 641 642 643 644 645 646
	ceph_msg_put(msg);
}
static void ceph_msg_remove_list(struct list_head *head)
{
	while (!list_empty(head)) {
		struct ceph_msg *msg = list_first_entry(head, struct ceph_msg,
							list_head);
		ceph_msg_remove(msg);
	}
}

static void reset_connection(struct ceph_connection *con)
{
	/* reset connection, out_queue, msg_ and connect_seq */
	/* discard existing out_queue and msg_seq */
647
	dout("reset_connection %p\n", con);
Sage Weil's avatar
Sage Weil committed
648 649 650
	ceph_msg_remove_list(&con->out_queue);
	ceph_msg_remove_list(&con->out_sent);

651
	if (con->in_msg) {
652
		BUG_ON(con->in_msg->con != con);
653 654 655 656
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
	}

Sage Weil's avatar
Sage Weil committed
657 658
	con->connect_seq = 0;
	con->out_seq = 0;
659
	if (con->out_msg) {
660
		BUG_ON(con->out_msg->con != con);
661 662 663
		ceph_msg_put(con->out_msg);
		con->out_msg = NULL;
	}
Sage Weil's avatar
Sage Weil committed
664
	con->in_seq = 0;
665
	con->in_seq_acked = 0;
Ilya Dryomov's avatar
Ilya Dryomov committed
666 667

	con->out_skip = 0;
Sage Weil's avatar
Sage Weil committed
668 669 670 671 672 673 674
}

/*
 * mark a peer down.  drop any open connections.
 */
void ceph_con_close(struct ceph_connection *con)
{
675
	mutex_lock(&con->mutex);
676 677
	dout("con_close %p peer %s\n", con,
	     ceph_pr_addr(&con->peer_addr.in_addr));
678
	con->state = CON_STATE_CLOSED;
679

680 681 682 683
	con_flag_clear(con, CON_FLAG_LOSSYTX);	/* so we retry next connect */
	con_flag_clear(con, CON_FLAG_KEEPALIVE_PENDING);
	con_flag_clear(con, CON_FLAG_WRITE_PENDING);
	con_flag_clear(con, CON_FLAG_BACKOFF);
684

Sage Weil's avatar
Sage Weil committed
685
	reset_connection(con);
Sage Weil's avatar
Sage Weil committed
686
	con->peer_global_seq = 0;
687
	cancel_con(con);
688
	con_close_socket(con);
689
	mutex_unlock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
690
}
691
EXPORT_SYMBOL(ceph_con_close);
Sage Weil's avatar
Sage Weil committed
692 693 694 695

/*
 * Reopen a closed connection, with a new peer address.
 */
696 697 698
void ceph_con_open(struct ceph_connection *con,
		   __u8 entity_type, __u64 entity_num,
		   struct ceph_entity_addr *addr)
Sage Weil's avatar
Sage Weil committed
699
{
700
	mutex_lock(&con->mutex);
701
	dout("con_open %p %s\n", con, ceph_pr_addr(&addr->in_addr));
702

703
	WARN_ON(con->state != CON_STATE_CLOSED);
704
	con->state = CON_STATE_PREOPEN;
705

706 707 708
	con->peer_name.type = (__u8) entity_type;
	con->peer_name.num = cpu_to_le64(entity_num);

Sage Weil's avatar
Sage Weil committed
709
	memcpy(&con->peer_addr, addr, sizeof(*addr));
710
	con->delay = 0;      /* reset backoff memory */
711
	mutex_unlock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
712 713
	queue_con(con);
}
714
EXPORT_SYMBOL(ceph_con_open);
Sage Weil's avatar
Sage Weil committed
715

716 717 718 719 720 721 722 723
/*
 * return true if this connection ever successfully opened
 */
bool ceph_con_opened(struct ceph_connection *con)
{
	return con->connect_seq > 0;
}

Sage Weil's avatar
Sage Weil committed
724 725 726
/*
 * initialize a new connection.
 */
727 728
void ceph_con_init(struct ceph_connection *con, void *private,
	const struct ceph_connection_operations *ops,
729
	struct ceph_messenger *msgr)
Sage Weil's avatar
Sage Weil committed
730 731 732
{
	dout("con_init %p\n", con);
	memset(con, 0, sizeof(*con));
733 734
	con->private = private;
	con->ops = ops;
Sage Weil's avatar
Sage Weil committed
735
	con->msgr = msgr;
736 737 738

	con_sock_state_init(con);

739
	mutex_init(&con->mutex);
Sage Weil's avatar
Sage Weil committed
740 741
	INIT_LIST_HEAD(&con->out_queue);
	INIT_LIST_HEAD(&con->out_sent);
742
	INIT_DELAYED_WORK(&con->work, ceph_con_workfn);
743

744
	con->state = CON_STATE_CLOSED;
Sage Weil's avatar
Sage Weil committed
745
}
746
EXPORT_SYMBOL(ceph_con_init);
Sage Weil's avatar
Sage Weil committed
747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764


/*
 * We maintain a global counter to order connection attempts.  Get
 * a unique seq greater than @gt.
 */
static u32 get_global_seq(struct ceph_messenger *msgr, u32 gt)
{
	u32 ret;

	spin_lock(&msgr->global_seq_lock);
	if (msgr->global_seq < gt)
		msgr->global_seq = gt;
	ret = ++msgr->global_seq;
	spin_unlock(&msgr->global_seq_lock);
	return ret;
}

765
static void con_out_kvec_reset(struct ceph_connection *con)
766
{
Ilya Dryomov's avatar
Ilya Dryomov committed
767 768
	BUG_ON(con->out_skip);

769 770 771 772 773
	con->out_kvec_left = 0;
	con->out_kvec_bytes = 0;
	con->out_kvec_cur = &con->out_kvec[0];
}

774
static void con_out_kvec_add(struct ceph_connection *con,
775 776
				size_t size, void *data)
{
Ilya Dryomov's avatar
Ilya Dryomov committed
777
	int index = con->out_kvec_left;
778

Ilya Dryomov's avatar
Ilya Dryomov committed
779
	BUG_ON(con->out_skip);
780 781 782 783 784 785 786
	BUG_ON(index >= ARRAY_SIZE(con->out_kvec));

	con->out_kvec[index].iov_len = size;
	con->out_kvec[index].iov_base = data;
	con->out_kvec_left++;
	con->out_kvec_bytes += size;
}
Sage Weil's avatar
Sage Weil committed
787

Ilya Dryomov's avatar
Ilya Dryomov committed
788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808
/*
 * Chop off a kvec from the end.  Return residual number of bytes for
 * that kvec, i.e. how many bytes would have been written if the kvec
 * hadn't been nuked.
 */
static int con_out_kvec_skip(struct ceph_connection *con)
{
	int off = con->out_kvec_cur - con->out_kvec;
	int skip = 0;

	if (con->out_kvec_bytes > 0) {
		skip = con->out_kvec[off + con->out_kvec_left - 1].iov_len;
		BUG_ON(con->out_kvec_bytes < skip);
		BUG_ON(!con->out_kvec_left);
		con->out_kvec_bytes -= skip;
		con->out_kvec_left--;
	}

	return skip;
}

809
#ifdef CONFIG_BLOCK
810 811 812 813 814 815

/*
 * For a bio data item, a piece is whatever remains of the next
 * entry in the current bio iovec, or the first entry in the next
 * bio in the list.
 */
816
static void ceph_msg_data_bio_cursor_init(struct ceph_msg_data_cursor *cursor,
817
					size_t length)
818
{
819
	struct ceph_msg_data *data = cursor->data;
820 821 822 823 824 825 826
	struct bio *bio;

	BUG_ON(data->type != CEPH_MSG_DATA_BIO);

	bio = data->bio;
	BUG_ON(!bio);

827
	cursor->resid = min(length, data->bio_length);
828
	cursor->bio = bio;
829 830 831
	cursor->bvec_iter = bio->bi_iter;
	cursor->last_piece =
		cursor->resid <= bio_iter_len(bio, cursor->bvec_iter);
832 833
}

834
static struct page *ceph_msg_data_bio_next(struct ceph_msg_data_cursor *cursor,
835 836 837
						size_t *page_offset,
						size_t *length)
{
838
	struct ceph_msg_data *data = cursor->data;
839
	struct bio *bio;
840
	struct bio_vec bio_vec;
841 842 843 844 845 846

	BUG_ON(data->type != CEPH_MSG_DATA_BIO);

	bio = cursor->bio;
	BUG_ON(!bio);

847
	bio_vec = bio_iter_iovec(bio, cursor->bvec_iter);
848

849
	*page_offset = (size_t) bio_vec.bv_offset;
850
	BUG_ON(*page_offset >= PAGE_SIZE);
851 852 853
	if (cursor->last_piece) /* pagelist offset is always 0 */
		*length = cursor->resid;
	else
854
		*length = (size_t) bio_vec.bv_len;
855
	BUG_ON(*length > cursor->resid);
856
	BUG_ON(*page_offset + *length > PAGE_SIZE);
857

858
	return bio_vec.bv_page;
859 860
}

861 862
static bool ceph_msg_data_bio_advance(struct ceph_msg_data_cursor *cursor,
					size_t bytes)
863 864
{
	struct bio *bio;
865
	struct bio_vec bio_vec;
866

867
	BUG_ON(cursor->data->type != CEPH_MSG_DATA_BIO);
868 869 870 871

	bio = cursor->bio;
	BUG_ON(!bio);

872
	bio_vec = bio_iter_iovec(bio, cursor->bvec_iter);
873 874 875

	/* Advance the cursor offset */

876 877
	BUG_ON(cursor->resid < bytes);
	cursor->resid -= bytes;
878 879 880 881

	bio_advance_iter(bio, &cursor->bvec_iter, bytes);

	if (bytes < bio_vec.bv_len)
882 883 884 885
		return false;	/* more bytes to process in this segment */

	/* Move on to the next segment, and possibly the next bio */

886
	if (!cursor->bvec_iter.bi_size) {
887
		bio = bio->bi_next;
888 889 890 891 892 893
		cursor->bio = bio;
		if (bio)
			cursor->bvec_iter = bio->bi_iter;
		else
			memset(&cursor->bvec_iter, 0,
			       sizeof(cursor->bvec_iter));
894 895
	}

896 897 898 899
	if (!cursor->last_piece) {
		BUG_ON(!cursor->resid);
		BUG_ON(!bio);
		/* A short read is OK, so use <= rather than == */
900
		if (cursor->resid <= bio_iter_len(bio, cursor->bvec_iter))
901
			cursor->last_piece = true;
902
	}
903 904 905

	return true;
}
906
#endif /* CONFIG_BLOCK */
907

908 909 910 911
/*
 * For a page array, a piece comes from the first page in the array
 * that has not already been fully consumed.
 */
912
static void ceph_msg_data_pages_cursor_init(struct ceph_msg_data_cursor *cursor,
913
					size_t length)
914
{
915
	struct ceph_msg_data *data = cursor->data;
916 917 918 919 920 921 922
	int page_count;

	BUG_ON(data->type != CEPH_MSG_DATA_PAGES);

	BUG_ON(!data->pages);
	BUG_ON(!data->length);

923
	cursor->resid = min(length, data->length);
924 925 926
	page_count = calc_pages_for(data->alignment, (u64)data->length);
	cursor->page_offset = data->alignment & ~PAGE_MASK;
	cursor->page_index = 0;
927 928 929
	BUG_ON(page_count > (int)USHRT_MAX);
	cursor->page_count = (unsigned short)page_count;
	BUG_ON(length > SIZE_MAX - cursor->page_offset);
930
	cursor->last_piece = cursor->page_offset + cursor->resid <= PAGE_SIZE;
931 932
}

933 934 935
static struct page *
ceph_msg_data_pages_next(struct ceph_msg_data_cursor *cursor,
					size_t *page_offset, size_t *length)
936
{
937
	struct ceph_msg_data *data = cursor->data;
938 939 940 941 942 943 944

	BUG_ON(data->type != CEPH_MSG_DATA_PAGES);

	BUG_ON(cursor->page_index >= cursor->page_count);
	BUG_ON(cursor->page_offset >= PAGE_SIZE);

	*page_offset = cursor->page_offset;
945
	if (cursor->last_piece)
946
		*length = cursor->resid;
947
	else
948 949 950 951 952
		*length = PAGE_SIZE - *page_offset;

	return data->pages[cursor->page_index];
}

953
static bool ceph_msg_data_pages_advance(struct ceph_msg_data_cursor *cursor,
954 955
						size_t bytes)
{
956
	BUG_ON(cursor->data->type != CEPH_MSG_DATA_PAGES);
957 958 959 960 961 962

	BUG_ON(cursor->page_offset + bytes > PAGE_SIZE);

	/* Advance the cursor page offset */

	cursor->resid -= bytes;
963 964
	cursor->page_offset = (cursor->page_offset + bytes) & ~PAGE_MASK;
	if (!bytes || cursor->page_offset)
965 966
		return false;	/* more bytes to process in the current page */

967 968 969
	if (!cursor->resid)
		return false;   /* no more data */

970
	/* Move on to the next page; offset is already at 0 */
971 972 973

	BUG_ON(cursor->page_index >= cursor->page_count);
	cursor->page_index++;
974
	cursor->last_piece = cursor->resid <= PAGE_SIZE;
975 976 977 978

	return true;
}

979
/*
980 981
 * For a pagelist, a piece is whatever remains to be consumed in the
 * first page in the list, or the front of the next page.
982
 */
983 984
static void
ceph_msg_data_pagelist_cursor_init(struct ceph_msg_data_cursor *cursor,
985
					size_t length)
986
{
987
	struct ceph_msg_data *data = cursor->data;
988 989 990
	struct ceph_pagelist *pagelist;
	struct page *page;

991
	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);
992 993 994

	pagelist = data->pagelist;
	BUG_ON(!pagelist);
995 996

	if (!length)
997 998 999 1000 1001
		return;		/* pagelist can be assigned but empty */

	BUG_ON(list_empty(&pagelist->head));
	page = list_first_entry(&pagelist->head, struct page, lru);

1002
	cursor->resid = min(length, pagelist->length);
1003 1004
	cursor->page = page;
	cursor->offset = 0;
Alex Elder's avatar
Alex Elder committed
1005
	cursor->last_piece = cursor->resid <= PAGE_SIZE;
1006 1007
}

1008 1009 1010
static struct page *
ceph_msg_data_pagelist_next(struct ceph_msg_data_cursor *cursor,
				size_t *page_offset, size_t *length)
1011
{
1012
	struct ceph_msg_data *data = cursor->data;
1013 1014 1015 1016 1017 1018 1019 1020
	struct ceph_pagelist *pagelist;

	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);

	pagelist = data->pagelist;
	BUG_ON(!pagelist);

	BUG_ON(!cursor->page);
1021
	BUG_ON(cursor->offset + cursor->resid != pagelist->length);
1022

1023
	/* offset of first page in pagelist is always 0 */
1024
	*page_offset = cursor->offset & ~PAGE_MASK;
1025
	if (cursor->last_piece)
1026 1027 1028
		*length = cursor->resid;
	else
		*length = PAGE_SIZE - *page_offset;
1029

1030
	return cursor->page;
1031 1032
}

1033
static bool ceph_msg_data_pagelist_advance(struct ceph_msg_data_cursor *cursor,
1034
						size_t bytes)
1035
{
1036
	struct ceph_msg_data *data = cursor->data;
1037 1038 1039 1040 1041 1042
	struct ceph_pagelist *pagelist;

	BUG_ON(data->type != CEPH_MSG_DATA_PAGELIST);

	pagelist = data->pagelist;
	BUG_ON(!pagelist);
1043 1044

	BUG_ON(cursor->offset + cursor->resid != pagelist->length);
1045 1046 1047 1048
	BUG_ON((cursor->offset & ~PAGE_MASK) + bytes > PAGE_SIZE);

	/* Advance the cursor offset */

1049
	cursor->resid -= bytes;
1050
	cursor->offset += bytes;
1051
	/* offset of first page in pagelist is always 0 */
1052 1053 1054
	if (!bytes || cursor->offset & ~PAGE_MASK)
		return false;	/* more bytes to process in the current page */

1055 1056 1057
	if (!cursor->resid)
		return false;   /* no more data */

1058 1059 1060
	/* Move on to the next page */

	BUG_ON(list_is_last(&cursor->page->lru, &pagelist->head));
1061
	cursor->page = list_next_entry(cursor->page, lru);
1062
	cursor->last_piece = cursor->resid <= PAGE_SIZE;
1063 1064 1065 1066

	return true;
}

1067 1068 1069 1070 1071 1072 1073 1074
/*
 * Message data is handled (sent or received) in pieces, where each
 * piece resides on a single page.  The network layer might not
 * consume an entire piece at once.  A data item's cursor keeps
 * track of which piece is next to process and how much remains to
 * be processed in that piece.  It also tracks whether the current
 * piece is the last one in the data item.
 */
1075
static void __ceph_msg_data_cursor_init(struct ceph_msg_data_cursor *cursor)
1076
{
1077
	size_t length = cursor->total_resid;
1078 1079

	switch (cursor->data->type) {
1080
	case CEPH_MSG_DATA_PAGELIST:
1081
		ceph_msg_data_pagelist_cursor_init(cursor, length);
1082
		break;
1083
	case CEPH_MSG_DATA_PAGES:
1084
		ceph_msg_data_pages_cursor_init(cursor, length);
1085
		break;
1086 1087
#ifdef CONFIG_BLOCK
	case CEPH_MSG_DATA_BIO:
1088
		ceph_msg_data_bio_cursor_init(cursor, length);
1089
		break;
1090
#endif /* CONFIG_BLOCK */
1091
	case CEPH_MSG_DATA_NONE:
1092 1093 1094 1095
	default:
		/* BUG(); */
		break;
	}
1096
	cursor->need_crc = true;
1097 1098
}

1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115
static void ceph_msg_data_cursor_init(struct ceph_msg *msg, size_t length)
{
	struct ceph_msg_data_cursor *cursor = &msg->cursor;
	struct ceph_msg_data *data;

	BUG_ON(!length);
	BUG_ON(length > msg->data_length);
	BUG_ON(list_empty(&msg->data));

	cursor->data_head = &msg->data;
	cursor->total_resid = length;
	data = list_first_entry(&msg->data, struct ceph_msg_data, links);
	cursor->data = data;

	__ceph_msg_data_cursor_init(cursor);
}

1116 1117 1118 1119 1120
/*
 * Return the page containing the next piece to process for a given
 * data item, and supply the page offset and length of that piece.
 * Indicate whether this is the last piece in this data item.
 */
1121 1122
static struct page *ceph_msg_data_next(struct ceph_msg_data_cursor *cursor,
					size_t *page_offset, size_t *length,
1123 1124 1125 1126
					bool *last_piece)
{
	struct page *page;

1127
	switch (cursor->data->type) {
1128
	case CEPH_MSG_DATA_PAGELIST:
1129
		page = ceph_msg_data_pagelist_next(cursor, page_offset, length);
1130
		break;
1131
	case CEPH_MSG_DATA_PAGES:
1132
		page = ceph_msg_data_pages_next(cursor, page_offset, length);
1133
		break;
1134 1135
#ifdef CONFIG_BLOCK
	case CEPH_MSG_DATA_BIO:
1136
		page = ceph_msg_data_bio_next(cursor, page_offset, length);
1137
		break;
1138
#endif /* CONFIG_BLOCK */
1139
	case CEPH_MSG_DATA_NONE:
1140 1141 1142 1143 1144 1145 1146 1147
	default:
		page = NULL;
		break;
	}
	BUG_ON(!page);
	BUG_ON(*page_offset + *length > PAGE_SIZE);
	BUG_ON(!*length);
	if (last_piece)
1148
		*last_piece = cursor->last_piece;
1149 1150 1151 1152 1153 1154 1155 1156

	return page;
}

/*
 * Returns true if the result moves the cursor on to the next piece
 * of the data item.
 */
1157 1158
static bool ceph_msg_data_advance(struct ceph_msg_data_cursor *cursor,
				size_t bytes)
1159 1160 1161
{
	bool new_piece;

1162
	BUG_ON(bytes > cursor->resid);
1163
	switch (cursor->data->type) {
1164
	case CEPH_MSG_DATA_PAGELIST:
1165
		new_piece = ceph_msg_data_pagelist_advance(cursor, bytes);
1166
		break;
1167
	case CEPH_MSG_DATA_PAGES:
1168
		new_piece = ceph_msg_data_pages_advance(cursor, bytes);
1169
		break;
1170 1171
#ifdef CONFIG_BLOCK
	case CEPH_MSG_DATA_BIO:
1172
		new_piece = ceph_msg_data_bio_advance(cursor, bytes);
1173
		break;
1174
#endif /* CONFIG_BLOCK */
1175
	case CEPH_MSG_DATA_NONE:
1176 1177 1178 1179
	default:
		BUG();
		break;
	}
1180
	cursor->total_resid -= bytes;
1181

1182 1183 1184
	if (!cursor->resid && cursor->total_resid) {
		WARN_ON(!cursor->last_piece);
		BUG_ON(list_is_last(&cursor->data->links, cursor->data_head));
1185
		cursor->data = list_next_entry(cursor->data, links);
1186
		__ceph_msg_data_cursor_init(cursor);
Alex Elder's avatar
Alex Elder committed
1187
		new_piece = true;
1188
	}
Alex Elder's avatar
Alex Elder committed
1189
	cursor->need_crc = new_piece;
1190

1191 1192 1193
	return new_piece;
}

1194 1195 1196 1197 1198 1199 1200
static size_t sizeof_footer(struct ceph_connection *con)
{
	return (con->peer_features & CEPH_FEATURE_MSG_AUTH) ?
	    sizeof(struct ceph_msg_footer) :
	    sizeof(struct ceph_msg_footer_old);
}

1201
static void prepare_message_data(struct ceph_msg *msg, u32 data_len)
1202 1203
{
	BUG_ON(!msg);
1204
	BUG_ON(!data_len);
1205

1206
	/* Initialize data cursor */
1207

1208
	ceph_msg_data_cursor_init(msg, (size_t)data_len);
1209 1210
}

Sage Weil's avatar
Sage Weil committed
1211 1212 1213 1214
/*
 * Prepare footer for currently outgoing message, and finish things
 * off.  Assumes out_kvec* are already valid.. we just add on to the end.
 */
1215
static void prepare_write_message_footer(struct ceph_connection *con)
Sage Weil's avatar
Sage Weil committed
1216 1217 1218
{
	struct ceph_msg *m = con->out_msg;

1219 1220
	m->footer.flags |= CEPH_MSG_FOOTER_COMPLETE;

Sage Weil's avatar
Sage Weil committed
1221
	dout("prepare_write_message_footer %p\n", con);
1222
	con_out_kvec_add(con, sizeof_footer(con), &m->footer);
1223 1224
	if (con->peer_features & CEPH_FEATURE_MSG_AUTH) {
		if (con->ops->sign_message)
1225
			con->ops->sign_message(m);
1226 1227 1228 1229 1230
		else
			m->footer.sig = 0;
	} else {
		m->old_footer.flags = m->footer.flags;
	}
Sage Weil's avatar
Sage Weil committed
1231
	con->out_more = m->more_to_follow;
1232
	con->out_msg_done = true;
Sage Weil's avatar
Sage Weil committed
1233 1234 1235 1236 1237 1238 1239 1240
}

/*
 * Prepare headers for the next outgoing message.
 */
static void prepare_write_message(struct ceph_connection *con)
{
	struct ceph_msg *m;
1241
	u32 crc;
Sage Weil's avatar
Sage Weil committed
1242

1243
	con_out_kvec_reset(con);
1244
	con->out_msg_done = false;
Sage Weil's avatar
Sage Weil committed
1245 1246 1247 1248 1249

	/* Sneak an ack in there first?  If we can get it into the same
	 * TCP packet that's a good thing. */
	if (con->in_seq > con->in_seq_acked) {
		con->in_seq_acked = con->in_seq;
1250
		con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
Sage Weil's avatar
Sage Weil committed
1251
		con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
1252
		con_out_kvec_add(con, sizeof (con->out_temp_ack),
1253
			&con->out_temp_ack);
Sage Weil's avatar
Sage Weil committed
1254 1255
	}

1256
	BUG_ON(list_empty(&con->out_queue));
1257
	m = list_first_entry(&con->out_queue, struct ceph_msg, list_head);
1258
	con->out_msg = m;
1259
	BUG_ON(m->con != con);
1260 1261 1262 1263

	/* put message on sent list */
	ceph_msg_get(m);
	list_move_tail(&m->list_head, &con->out_sent);
Sage Weil's avatar
Sage Weil committed
1264

1265 1266 1267 1268 1269 1270 1271 1272
	/*
	 * only assign outgoing seq # if we haven't sent this message
	 * yet.  if it is requeued, resend with it's original seq.
	 */
	if (m->needs_out_seq) {
		m->hdr.seq = cpu_to_le64(++con->out_seq);
		m->needs_out_seq = false;
	}
1273
	WARN_ON(m->data_length != le32_to_cpu(m->hdr.data_len));
Sage Weil's avatar
Sage Weil committed
1274

1275
	dout("prepare_write_message %p seq %lld type %d len %d+%d+%zd\n",
Sage Weil's avatar
Sage Weil committed
1276 1277
	     m, con->out_seq, le16_to_cpu(m->hdr.type),
	     le32_to_cpu(m->hdr.front_len), le32_to_cpu(m->hdr.middle_len),
1278
	     m->data_length);
Sage Weil's avatar
Sage Weil committed
1279 1280 1281
	BUG_ON(le32_to_cpu(m->hdr.front_len) != m->front.iov_len);

	/* tag + hdr + front + middle */
1282
	con_out_kvec_add(con, sizeof (tag_msg), &tag_msg);
Ilya Dryomov's avatar
Ilya Dryomov committed
1283
	con_out_kvec_add(con, sizeof(con->out_hdr), &con->out_hdr);
1284
	con_out_kvec_add(con, m->front.iov_len, m->front.iov_base);
1285

Sage Weil's avatar
Sage Weil committed
1286
	if (m->middle)
1287
		con_out_kvec_add(con, m->middle->vec.iov_len,
1288
			m->middle->vec.iov_base);
Sage Weil's avatar
Sage Weil committed
1289

Ilya Dryomov's avatar
Ilya Dryomov committed
1290
	/* fill in hdr crc and finalize hdr */
1291 1292
	crc = crc32c(0, &m->hdr, offsetof(struct ceph_msg_header, crc));
	con->out_msg->hdr.crc = cpu_to_le32(crc);
Ilya Dryomov's avatar
Ilya Dryomov committed
1293
	memcpy(&con->out_hdr, &con->out_msg->hdr, sizeof(con->out_hdr));
1294

Ilya Dryomov's avatar
Ilya Dryomov committed
1295
	/* fill in front and middle crc, footer */
1296 1297 1298 1299 1300 1301 1302
	crc = crc32c(0, m->front.iov_base, m->front.iov_len);
	con->out_msg->footer.front_crc = cpu_to_le32(crc);
	if (m->middle) {
		crc = crc32c(0, m->middle->vec.iov_base,
				m->middle->vec.iov_len);
		con->out_msg->footer.middle_crc = cpu_to_le32(crc);
	} else
Sage Weil's avatar
Sage Weil committed
1303
		con->out_msg->footer.middle_crc = 0;
1304
	dout("%s front_crc %u middle_crc %u\n", __func__,
Sage Weil's avatar
Sage Weil committed
1305 1306
	     le32_to_cpu(con->out_msg->footer.front_crc),
	     le32_to_cpu(con->out_msg->footer.middle_crc));
Ilya Dryomov's avatar
Ilya Dryomov committed
1307
	con->out_msg->footer.flags = 0;
Sage Weil's avatar
Sage Weil committed
1308 1309

	/* is there a data payload? */
1310
	con->out_msg->footer.data_crc = 0;
1311 1312
	if (m->data_length) {
		prepare_message_data(con->out_msg, m->data_length);
1313 1314
		con->out_more = 1;  /* data + footer will follow */
	} else {
Sage Weil's avatar
Sage Weil committed
1315
		/* no, queue up footer too and be done */
1316
		prepare_write_message_footer(con);
1317
	}
Sage Weil's avatar
Sage Weil committed
1318

1319
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
Sage Weil's avatar
Sage Weil committed
1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330
}

/*
 * Prepare an ack.
 */
static void prepare_write_ack(struct ceph_connection *con)
{
	dout("prepare_write_ack %p %llu -> %llu\n", con,
	     con->in_seq_acked, con->in_seq);
	con->in_seq_acked = con->in_seq;

1331
	con_out_kvec_reset(con);
1332

1333
	con_out_kvec_add(con, sizeof (tag_ack), &tag_ack);
1334

Sage Weil's avatar
Sage Weil committed
1335
	con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
1336
	con_out_kvec_add(con, sizeof (con->out_temp_ack),
1337 1338
				&con->out_temp_ack);

Sage Weil's avatar
Sage Weil committed
1339
	con->out_more = 1;  /* more will follow.. eventually.. */
1340
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
Sage Weil's avatar
Sage Weil committed
1341 1342
}

1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360
/*
 * Prepare to share the seq during handshake
 */
static void prepare_write_seq(struct ceph_connection *con)
{
	dout("prepare_write_seq %p %llu -> %llu\n", con,
	     con->in_seq_acked, con->in_seq);
	con->in_seq_acked = con->in_seq;

	con_out_kvec_reset(con);

	con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
	con_out_kvec_add(con, sizeof (con->out_temp_ack),
			 &con->out_temp_ack);

	con_flag_set(con, CON_FLAG_WRITE_PENDING);
}

Sage Weil's avatar
Sage Weil committed
1361 1362 1363 1364 1365 1366
/*
 * Prepare to write keepalive byte.
 */
static void prepare_write_keepalive(struct ceph_connection *con)
{
	dout("prepare_write_keepalive %p\n", con);
1367
	con_out_kvec_reset(con);
1368
	if (con->peer_features & CEPH_FEATURE_MSGR_KEEPALIVE2) {
1369 1370
		struct timespec now = CURRENT_TIME;

1371
		con_out_kvec_add(con, sizeof(tag_keepalive2), &tag_keepalive2);
1372 1373 1374
		ceph_encode_timespec(&con->out_temp_keepalive2, &now);
		con_out_kvec_add(con, sizeof(con->out_temp_keepalive2),
				 &con->out_temp_keepalive2);
1375 1376 1377
	} else {
		con_out_kvec_add(con, sizeof(tag_keepalive), &tag_keepalive);
	}
1378
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
Sage Weil's avatar
Sage Weil committed
1379 1380 1381 1382 1383 1384
}

/*
 * Connection negotiation.
 */

1385 1386
static struct ceph_auth_handshake *get_connect_authorizer(struct ceph_connection *con,
						int *auth_proto)
1387
{
1388
	struct ceph_auth_handshake *auth;
1389 1390 1391 1392

	if (!con->ops->get_authorizer) {
		con->out_connect.authorizer_protocol = CEPH_AUTH_UNKNOWN;
		con->out_connect.authorizer_len = 0;
1393
		return NULL;
1394 1395 1396
	}

	/* Can't hold the mutex while getting authorizer */
1397
	mutex_unlock(&con->mutex);
1398
	auth = con->ops->get_authorizer(con, auth_proto, con->auth_retry);
1399
	mutex_lock(&con->mutex);
1400

1401
	if (IS_ERR(auth))
1402
		return auth;
1403
	if (con->state != CON_STATE_NEGOTIATING)
1404
		return ERR_PTR(-EAGAIN);
1405

1406 1407
	con->auth_reply_buf = auth->authorizer_reply_buf;
	con->auth_reply_buf_len = auth->authorizer_reply_buf_len;
1408
	return auth;
1409 1410
}

Sage Weil's avatar
Sage Weil committed
1411 1412 1413
/*
 * We connected to a peer and are saying hello.
 */
1414
static void prepare_write_banner(struct ceph_connection *con)
Sage Weil's avatar
Sage Weil committed
1415
{
1416 1417
	con_out_kvec_add(con, strlen(CEPH_BANNER), CEPH_BANNER);
	con_out_kvec_add(con, sizeof (con->msgr->my_enc_addr),
1418
					&con->msgr->my_enc_addr);
1419 1420

	con->out_more = 0;
1421
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
1422 1423
}

1424
static int prepare_write_connect(struct ceph_connection *con)
1425
{
1426
	unsigned int global_seq = get_global_seq(con->msgr, 0);
Sage Weil's avatar
Sage Weil committed
1427
	int proto;
1428
	int auth_proto;
1429
	struct ceph_auth_handshake *auth;
Sage Weil's avatar
Sage Weil committed
1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446

	switch (con->peer_name.type) {
	case CEPH_ENTITY_TYPE_MON:
		proto = CEPH_MONC_PROTOCOL;
		break;
	case CEPH_ENTITY_TYPE_OSD:
		proto = CEPH_OSDC_PROTOCOL;
		break;
	case CEPH_ENTITY_TYPE_MDS:
		proto = CEPH_MDSC_PROTOCOL;
		break;
	default:
		BUG();
	}

	dout("prepare_write_connect %p cseq=%d gseq=%d proto=%d\n", con,
	     con->connect_seq, global_seq, proto);
1447

1448 1449
	con->out_connect.features =
	    cpu_to_le64(from_msgr(con->msgr)->supported_features);
Sage Weil's avatar
Sage Weil committed
1450 1451 1452 1453 1454 1455
	con->out_connect.host_type = cpu_to_le32(CEPH_ENTITY_TYPE_CLIENT);
	con->out_connect.connect_seq = cpu_to_le32(con->connect_seq);
	con->out_connect.global_seq = cpu_to_le32(global_seq);
	con->out_connect.protocol_version = cpu_to_le32(proto);
	con->out_connect.flags = 0;

1456 1457
	auth_proto = CEPH_AUTH_UNKNOWN;
	auth = get_connect_authorizer(con, &auth_proto);
1458 1459
	if (IS_ERR(auth))
		return PTR_ERR(auth);
1460

1461
	con->out_connect.authorizer_protocol = cpu_to_le32(auth_proto);
1462 1463 1464
	con->out_connect.authorizer_len = auth ?
		cpu_to_le32(auth->authorizer_buf_len) : 0;

1465
	con_out_kvec_add(con, sizeof (con->out_connect),
1466 1467
					&con->out_connect);
	if (auth && auth->authorizer_buf_len)
1468
		con_out_kvec_add(con, auth->authorizer_buf_len,
1469
					auth->authorizer_buf);
1470

Sage Weil's avatar
Sage Weil committed
1471
	con->out_more = 0;
1472
	con_flag_set(con, CON_FLAG_WRITE_PENDING);
1473

1474
	return 0;
Sage Weil's avatar
Sage Weil committed
1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496
}

/*
 * write as much of pending kvecs to the socket as we can.
 *  1 -> done
 *  0 -> socket full, but more to do
 * <0 -> error
 */
static int write_partial_kvec(struct ceph_connection *con)
{
	int ret;

	dout("write_partial_kvec %p %d left\n", con, con->out_kvec_bytes);
	while (con->out_kvec_bytes > 0) {
		ret = ceph_tcp_sendmsg(con->sock, con->out_kvec_cur,
				       con->out_kvec_left, con->out_kvec_bytes,
				       con->out_more);
		if (ret <= 0)
			goto out;
		con->out_kvec_bytes -= ret;
		if (con->out_kvec_bytes == 0)
			break;            /* done */
1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508

		/* account for full iov entries consumed */
		while (ret >= con->out_kvec_cur->iov_len) {
			BUG_ON(!con->out_kvec_left);
			ret -= con->out_kvec_cur->iov_len;
			con->out_kvec_cur++;
			con->out_kvec_left--;
		}
		/* and for a partially-consumed entry */
		if (ret) {
			con->out_kvec_cur->iov_len -= ret;
			con->out_kvec_cur->iov_base += ret;
Sage Weil's avatar
Sage Weil committed
1509 1510 1511 1512 1513 1514 1515 1516 1517 1518
		}
	}
	con->out_kvec_left = 0;
	ret = 1;
out:
	dout("write_partial_kvec %p %d left in %d kvecs ret = %d\n", con,
	     con->out_kvec_bytes, con->out_kvec_left, ret);
	return ret;  /* done! */
}

1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531
static u32 ceph_crc32c_page(u32 crc, struct page *page,
				unsigned int page_offset,
				unsigned int length)
{
	char *kaddr;

	kaddr = kmap(page);
	BUG_ON(kaddr == NULL);
	crc = crc32c(crc, kaddr + page_offset, length);
	kunmap(page);

	return crc;
}
Sage Weil's avatar
Sage Weil committed
1532 1533 1534 1535 1536 1537 1538
/*
 * Write as much message data payload as we can.  If we finish, queue
 * up the footer.
 *  1 -> done, footer is now queued in out_kvec[].
 *  0 -> socket full, but more to do
 * <0 -> error
 */
1539
static int write_partial_message_data(struct ceph_connection *con)
Sage Weil's avatar
Sage Weil committed
1540 1541
{
	struct ceph_msg *msg = con->out_msg;
1542
	struct ceph_msg_data_cursor *cursor = &msg->cursor;
1543
	bool do_datacrc = !ceph_test_opt(from_msgr(con->msgr), NOCRC);
1544
	u32 crc;
Sage Weil's avatar
Sage Weil committed
1545

1546
	dout("%s %p msg %p\n", __func__, con, msg);
Sage Weil's avatar
Sage Weil committed
1547

1548
	if (list_empty(&msg->data))
1549 1550
		return -EINVAL;

1551 1552 1553 1554 1555 1556 1557 1558
	/*
	 * Iterate through each page that contains data to be
	 * written, and send as much as possible for each.
	 *
	 * If we are calculating the data crc (the default), we will
	 * need to map the page.  If we have no pages, they have
	 * been revoked, so use the zero page.
	 */
1559
	crc = do_datacrc ? le32_to_cpu(msg->footer.data_crc) : 0;
1560
	while (cursor->resid) {
1561
		struct page *page;
1562 1563
		size_t page_offset;
		size_t length;
1564
		bool last_piece;
1565
		bool need_crc;
1566
		int ret;
1567

1568 1569
		page = ceph_msg_data_next(cursor, &page_offset, &length,
					  &last_piece);
1570
		ret = ceph_tcp_sendpage(con->sock, page, page_offset,
1571
					length, !last_piece);
1572 1573 1574
		if (ret <= 0) {
			if (do_datacrc)
				msg->footer.data_crc = cpu_to_le32(crc);
Sage Weil's avatar
Sage Weil committed
1575

1576 1577
			return ret;
		}
1578 1579
		if (do_datacrc && cursor->need_crc)
			crc = ceph_crc32c_page(crc, page, page_offset, length);
1580
		need_crc = ceph_msg_data_advance(cursor, (size_t)ret);
Sage Weil's avatar
Sage Weil committed
1581 1582
	}

1583
	dout("%s %p msg %p done\n", __func__, con, msg);
Sage Weil's avatar
Sage Weil committed
1584 1585

	/* prepare and queue up footer, too */
1586 1587 1588
	if (do_datacrc)
		msg->footer.data_crc = cpu_to_le32(crc);
	else
1589
		msg->footer.flags |= CEPH_MSG_FOOTER_NOCRC;
1590
	con_out_kvec_reset(con);
1591
	prepare_write_message_footer(con);
1592 1593

	return 1;	/* must return > 0 to indicate success */
Sage Weil's avatar
Sage Weil committed
1594 1595 1596 1597 1598 1599 1600 1601 1602
}

/*
 * write some zeros
 */
static int write_partial_skip(struct ceph_connection *con)
{
	int ret;

Ilya Dryomov's avatar
Ilya Dryomov committed
1603
	dout("%s %p %d left\n", __func__, con, con->out_skip);
Sage Weil's avatar
Sage Weil committed
1604
	while (con->out_skip > 0) {
1605
		size_t size = min(con->out_skip, (int) PAGE_CACHE_SIZE);
Sage Weil's avatar
Sage Weil committed
1606

1607
		ret = ceph_tcp_sendpage(con->sock, zero_page, 0, size, true);
Sage Weil's avatar
Sage Weil committed
1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619
		if (ret <= 0)
			goto out;
		con->out_skip -= ret;
	}
	ret = 1;
out:
	return ret;
}

/*
 * Prepare to read connection handshake, or an ack.
 */
1620 1621 1622 1623 1624 1625
static void prepare_read_banner(struct ceph_connection *con)
{
	dout("prepare_read_banner %p\n", con);
	con->in_base_pos = 0;
}

Sage Weil's avatar
Sage Weil committed
1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637
static void prepare_read_connect(struct ceph_connection *con)
{
	dout("prepare_read_connect %p\n", con);
	con->in_base_pos = 0;
}

static void prepare_read_ack(struct ceph_connection *con)
{
	dout("prepare_read_ack %p\n", con);
	con->in_base_pos = 0;
}

1638 1639 1640 1641 1642 1643 1644
static void prepare_read_seq(struct ceph_connection *con)
{
	dout("prepare_read_seq %p\n", con);
	con->in_base_pos = 0;
	con->in_tag = CEPH_MSGR_TAG_SEQ;
}

Sage Weil's avatar
Sage Weil committed
1645 1646 1647 1648 1649 1650 1651
static void prepare_read_tag(struct ceph_connection *con)
{
	dout("prepare_read_tag %p\n", con);
	con->in_base_pos = 0;
	con->in_tag = CEPH_MSGR_TAG_READY;
}

1652 1653 1654 1655 1656 1657
static void prepare_read_keepalive_ack(struct ceph_connection *con)
{
	dout("prepare_read_keepalive_ack %p\n", con);
	con->in_base_pos = 0;
}

Sage Weil's avatar
Sage Weil committed
1658 1659 1660 1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671
/*
 * Prepare to read a message.
 */
static int prepare_read_message(struct ceph_connection *con)
{
	dout("prepare_read_message %p\n", con);
	BUG_ON(con->in_msg != NULL);
	con->in_base_pos = 0;
	con->in_front_crc = con->in_middle_crc = con->in_data_crc = 0;
	return 0;
}


static int read_partial(struct ceph_connection *con,
1672
			int end, int size, void *object)
Sage Weil's avatar
Sage Weil committed
1673
{
1674 1675
	while (con->in_base_pos < end) {
		int left = end - con->in_base_pos;
Sage Weil's avatar
Sage Weil committed
1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688
		int have = size - left;
		int ret = ceph_tcp_recvmsg(con->sock, object + have, left);
		if (ret <= 0)
			return ret;
		con->in_base_pos += ret;
	}
	return 1;
}


/*
 * Read all or part of the connect-side handshake on a new connection
 */
1689
static int read_partial_banner(struct ceph_connection *con)
Sage Weil's avatar
Sage Weil committed
1690
{
1691 1692 1693
	int size;
	int end;
	int ret;
Sage Weil's avatar
Sage Weil committed
1694

1695
	dout("read_partial_banner %p at %d\n", con, con->in_base_pos);
Sage Weil's avatar
Sage Weil committed
1696 1697

	/* peer's banner */
1698 1699 1700
	size = strlen(CEPH_BANNER);
	end = size;
	ret = read_partial(con, end, size, con->in_banner);
Sage Weil's avatar
Sage Weil committed
1701 1702
	if (ret <= 0)
		goto out;
1703 1704 1705 1706

	size = sizeof (con->actual_peer_addr);
	end += size;
	ret = read_partial(con, end, size, &con->actual_peer_addr);
Sage Weil's avatar
Sage Weil committed
1707 1708
	if (ret <= 0)
		goto out;
1709 1710 1711 1712

	size = sizeof (con->peer_addr_for_me);
	end += size;
	ret = read_partial(con, end, size, &con->peer_addr_for_me);
Sage Weil's avatar
Sage Weil committed
1713 1714
	if (ret <= 0)
		goto out;
1715

1716 1717 1718 1719 1720 1721
out:
	return ret;
}

static int read_partial_connect(struct ceph_connection *con)
{
1722 1723 1724
	int size;
	int end;
	int ret;
1725 1726 1727

	dout("read_partial_connect %p at %d\n", con, con->in_base_pos);

1728 1729 1730
	size = sizeof (con->in_reply);
	end = size;
	ret = read_partial(con, end, size, &con->in_reply);
Sage Weil's avatar
Sage Weil committed
1731 1732
	if (ret <= 0)
		goto out;
1733 1734 1735 1736

	size = le32_to_cpu(con->in_reply.authorizer_len);
	end += size;
	ret = read_partial(con, end, size, con->auth_reply_buf);
1737 1738
	if (ret <= 0)
		goto out;
Sage Weil's avatar
Sage Weil committed
1739

1740 1741 1742
	dout("read_partial_connect %p tag %d, con_seq = %u, g_seq = %u\n",
	     con, (int)con->in_reply.tag,
	     le32_to_cpu(con->in_reply.connect_seq),
Sage Weil's avatar
Sage Weil committed
1743 1744 1745
	     le32_to_cpu(con->in_reply.global_seq));
out:
	return ret;
1746

Sage Weil's avatar
Sage Weil committed
1747 1748 1749 1750 1751 1752 1753 1754
}

/*
 * Verify the hello banner looks okay.
 */
static int verify_hello(struct ceph_connection *con)
{
	if (memcmp(con->in_banner, CEPH_BANNER, strlen(CEPH_BANNER))) {
1755
		pr_err("connect to %s got bad banner\n",
1756
		       ceph_pr_addr(&con->peer_addr.in_addr));
Sage Weil's avatar
Sage Weil committed
1757 1758 1759 1760 1761 1762 1763 1764
		con->error_msg = "protocol error, bad banner";
		return -1;
	}
	return 0;
}

static bool addr_is_blank(struct sockaddr_storage *ss)
{
1765 1766 1767
	struct in_addr *addr = &((struct sockaddr_in *)ss)->sin_addr;
	struct in6_addr *addr6 = &((struct sockaddr_in6 *)ss)->sin6_addr;

Sage Weil's avatar
Sage Weil committed
1768 1769
	switch (ss->ss_family) {
	case AF_INET:
1770
		return addr->s_addr == htonl(INADDR_ANY);
Sage Weil's avatar
Sage Weil committed
1771
	case AF_INET6:
1772 1773 1774
		return ipv6_addr_any(addr6);
	default:
		return true;
Sage Weil's avatar
Sage Weil committed
1775 1776 1777 1778 1779 1780 1781
	}
}

static int addr_port(struct sockaddr_storage *ss)
{
	switch (ss->ss_family) {
	case AF_INET:
Sage Weil's avatar
Sage Weil committed
1782
		return ntohs(((struct sockaddr_in *)ss)->sin_port);
Sage Weil's avatar
Sage Weil committed
1783
	case AF_INET6:
Sage Weil's avatar
Sage Weil committed
1784
		return ntohs(((struct sockaddr_in6 *)ss)->sin6_port);
Sage Weil's avatar
Sage Weil committed
1785 1786 1787 1788 1789 1790 1791 1792 1793
	}
	return 0;
}

static void addr_set_port(struct sockaddr_storage *ss, int p)
{
	switch (ss->ss_family) {
	case AF_INET:
		((struct sockaddr_in *)ss)->sin_port = htons(p);
1794
		break;
Sage Weil's avatar
Sage Weil committed
1795 1796
	case AF_INET6:
		((struct sockaddr_in6 *)ss)->sin6_port = htons(p);
1797
		break;
Sage Weil's avatar
Sage Weil committed
1798 1799 1800
	}
}

Noah Watkins's avatar
Noah Watkins committed
1801 1802 1803 1804 1805 1806
/*
 * Unlike other *_pton function semantics, zero indicates success.
 */
static int ceph_pton(const char *str, size_t len, struct sockaddr_storage *ss,
		char delim, const char **ipend)
{
1807 1808
	struct sockaddr_in *in4 = (struct sockaddr_in *) ss;
	struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) ss;
Noah Watkins's avatar
Noah Watkins committed
1809 1810 1811 1812 1813 1814 1815 1816 1817 1818 1819 1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848 1849 1850 1851 1852 1853 1854 1855 1856 1857 1858 1859 1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878 1879 1880 1881 1882 1883 1884 1885 1886 1887 1888 1889 1890 1891 1892 1893 1894 1895

	memset(ss, 0, sizeof(*ss));

	if (in4_pton(str, len, (u8 *)&in4->sin_addr.s_addr, delim, ipend)) {
		ss->ss_family = AF_INET;
		return 0;
	}

	if (in6_pton(str, len, (u8 *)&in6->sin6_addr.s6_addr, delim, ipend)) {
		ss->ss_family = AF_INET6;
		return 0;
	}

	return -EINVAL;
}

/*
 * Extract hostname string and resolve using kernel DNS facility.
 */
#ifdef CONFIG_CEPH_LIB_USE_DNS_RESOLVER
static int ceph_dns_resolve_name(const char *name, size_t namelen,
		struct sockaddr_storage *ss, char delim, const char **ipend)
{
	const char *end, *delim_p;
	char *colon_p, *ip_addr = NULL;
	int ip_len, ret;

	/*
	 * The end of the hostname occurs immediately preceding the delimiter or
	 * the port marker (':') where the delimiter takes precedence.
	 */
	delim_p = memchr(name, delim, namelen);
	colon_p = memchr(name, ':', namelen);

	if (delim_p && colon_p)
		end = delim_p < colon_p ? delim_p : colon_p;
	else if (!delim_p && colon_p)
		end = colon_p;
	else {
		end = delim_p;
		if (!end) /* case: hostname:/ */
			end = name + namelen;
	}

	if (end <= name)
		return -EINVAL;

	/* do dns_resolve upcall */
	ip_len = dns_query(NULL, name, end - name, NULL, &ip_addr, NULL);
	if (ip_len > 0)
		ret = ceph_pton(ip_addr, ip_len, ss, -1, NULL);
	else
		ret = -ESRCH;

	kfree(ip_addr);

	*ipend = end;

	pr_info("resolve '%.*s' (ret=%d): %s\n", (int)(end - name), name,
			ret, ret ? "failed" : ceph_pr_addr(ss));

	return ret;
}
#else
static inline int ceph_dns_resolve_name(const char *name, size_t namelen,
		struct sockaddr_storage *ss, char delim, const char **ipend)
{
	return -EINVAL;
}
#endif

/*
 * Parse a server name (IP or hostname). If a valid IP address is not found
 * then try to extract a hostname to resolve using userspace DNS upcall.
 */
static int ceph_parse_server_name(const char *name, size_t namelen,
			struct sockaddr_storage *ss, char delim, const char **ipend)
{
	int ret;

	ret = ceph_pton(name, namelen, ss, delim, ipend);
	if (ret)
		ret = ceph_dns_resolve_name(name, namelen, ss, delim, ipend);

	return ret;
}

Sage Weil's avatar
Sage Weil committed
1896 1897 1898 1899 1900 1901 1902 1903
/*
 * Parse an ip[:port] list into an addr array.  Use the default
 * monitor port if a port isn't specified.
 */
int ceph_parse_ips(const char *c, const char *end,
		   struct ceph_entity_addr *addr,
		   int max_count, int *count)
{
Noah Watkins's avatar
Noah Watkins committed
1904
	int i, ret = -EINVAL;
Sage Weil's avatar
Sage Weil committed
1905 1906 1907 1908 1909 1910 1911
	const char *p = c;

	dout("parse_ips on '%.*s'\n", (int)(end-c), c);
	for (i = 0; i < max_count; i++) {
		const char *ipend;
		struct sockaddr_storage *ss = &addr[i].in_addr;
		int port;
1912 1913 1914 1915 1916 1917
		char delim = ',';

		if (*p == '[') {
			delim = ']';
			p++;
		}
Sage Weil's avatar
Sage Weil committed
1918

Noah Watkins's avatar
Noah Watkins committed
1919 1920
		ret = ceph_parse_server_name(p, end - p, ss, delim, &ipend);
		if (ret)
Sage Weil's avatar
Sage Weil committed
1921
			goto bad;
Noah Watkins's avatar
Noah Watkins committed
1922 1923
		ret = -EINVAL;

Sage Weil's avatar
Sage Weil committed
1924 1925
		p = ipend;

1926 1927 1928 1929 1930 1931 1932 1933
		if (delim == ']') {
			if (*p != ']') {
				dout("missing matching ']'\n");
				goto bad;
			}
			p++;
		}

Sage Weil's avatar
Sage Weil committed
1934 1935 1936 1937 1938 1939 1940 1941
		/* port? */
		if (p < end && *p == ':') {
			port = 0;
			p++;
			while (p < end && *p >= '0' && *p <= '9') {
				port = (port * 10) + (*p - '0');
				p++;
			}
1942 1943 1944
			if (port == 0)
				port = CEPH_MON_PORT;
			else if (port > 65535)
Sage Weil's avatar
Sage Weil committed
1945 1946 1947 1948 1949 1950 1951
				goto bad;
		} else {
			port = CEPH_MON_PORT;
		}

		addr_set_port(ss, port);

1952
		dout("parse_ips got %s\n", ceph_pr_addr(ss));
Sage Weil's avatar
Sage Weil committed
1953 1954 1955 1956 1957 1958 1959 1960 1961 1962 1963 1964 1965 1966 1967 1968

		if (p == end)
			break;
		if (*p != ',')
			goto bad;
		p++;
	}

	if (p != end)
		goto bad;

	if (count)
		*count = i + 1;
	return 0;

bad:
1969
	pr_err("parse_ips bad ip '%.*s'\n", (int)(end - c), c);
Noah Watkins's avatar
Noah Watkins committed
1970
	return ret;
Sage Weil's avatar
Sage Weil committed
1971
}
1972
EXPORT_SYMBOL(ceph_parse_ips);
Sage Weil's avatar
Sage Weil committed
1973

1974
static int process_banner(struct ceph_connection *con)
Sage Weil's avatar
Sage Weil committed
1975
{
1976
	dout("process_banner on %p\n", con);
Sage Weil's avatar
Sage Weil committed
1977 1978 1979 1980

	if (verify_hello(con) < 0)
		return -1;

1981 1982 1983
	ceph_decode_addr(&con->actual_peer_addr);
	ceph_decode_addr(&con->peer_addr_for_me);

Sage Weil's avatar
Sage Weil committed
1984 1985 1986 1987 1988
	/*
	 * Make sure the other end is who we wanted.  note that the other
	 * end may not yet know their ip address, so if it's 0.0.0.0, give
	 * them the benefit of the doubt.
	 */
Sage Weil's avatar
Sage Weil committed
1989 1990
	if (memcmp(&con->peer_addr, &con->actual_peer_addr,
		   sizeof(con->peer_addr)) != 0 &&
Sage Weil's avatar
Sage Weil committed
1991 1992
	    !(addr_is_blank(&con->actual_peer_addr.in_addr) &&
	      con->actual_peer_addr.nonce == con->peer_addr.nonce)) {
1993 1994 1995 1996 1997
		pr_warn("wrong peer, want %s/%d, got %s/%d\n",
			ceph_pr_addr(&con->peer_addr.in_addr),
			(int)le32_to_cpu(con->peer_addr.nonce),
			ceph_pr_addr(&con->actual_peer_addr.in_addr),
			(int)le32_to_cpu(con->actual_peer_addr.nonce));
1998
		con->error_msg = "wrong peer at address";
Sage Weil's avatar
Sage Weil committed
1999 2000 2001 2002 2003 2004 2005 2006 2007 2008 2009 2010 2011
		return -1;
	}

	/*
	 * did we learn our address?
	 */
	if (addr_is_blank(&con->msgr->inst.addr.in_addr)) {
		int port = addr_port(&con->msgr->inst.addr.in_addr);

		memcpy(&con->msgr->inst.addr.in_addr,
		       &con->peer_addr_for_me.in_addr,
		       sizeof(con->peer_addr_for_me.in_addr));
		addr_set_port(&con->msgr->inst.addr.in_addr, port);
2012
		encode_my_addr(con->msgr);
2013
		dout("process_banner learned my addr is %s\n",
2014
		     ceph_pr_addr(&con->msgr->inst.addr.in_addr));
Sage Weil's avatar
Sage Weil committed
2015 2016
	}

2017 2018 2019 2020 2021
	return 0;
}

static int process_connect(struct ceph_connection *con)
{
2022 2023
	u64 sup_feat = from_msgr(con->msgr)->supported_features;
	u64 req_feat = from_msgr(con->msgr)->required_features;
Ilya Dryomov's avatar
Ilya Dryomov committed
2024 2025
	u64 server_feat = ceph_sanitize_features(
				le64_to_cpu(con->in_reply.features));
2026
	int ret;
2027

2028 2029
	dout("process_connect on %p tag %d\n", con, (int)con->in_tag);

Sage Weil's avatar
Sage Weil committed
2030
	switch (con->in_reply.tag) {
2031 2032 2033 2034
	case CEPH_MSGR_TAG_FEATURES:
		pr_err("%s%lld %s feature set mismatch,"
		       " my %llx < server's %llx, missing %llx\n",
		       ENTITY_NAME(con->peer_name),
2035
		       ceph_pr_addr(&con->peer_addr.in_addr),
2036 2037
		       sup_feat, server_feat, server_feat & ~sup_feat);
		con->error_msg = "missing required protocol features";
2038
		reset_connection(con);
2039 2040
		return -1;

Sage Weil's avatar
Sage Weil committed
2041 2042 2043 2044
	case CEPH_MSGR_TAG_BADPROTOVER:
		pr_err("%s%lld %s protocol version mismatch,"
		       " my %d != server's %d\n",
		       ENTITY_NAME(con->peer_name),
2045
		       ceph_pr_addr(&con->peer_addr.in_addr),
Sage Weil's avatar
Sage Weil committed
2046 2047 2048
		       le32_to_cpu(con->out_connect.protocol_version),
		       le32_to_cpu(con->in_reply.protocol_version));
		con->error_msg = "protocol version mismatch";
2049
		reset_connection(con);
Sage Weil's avatar
Sage Weil committed
2050 2051
		return -1;

2052 2053 2054 2055 2056 2057 2058 2059
	case CEPH_MSGR_TAG_BADAUTHORIZER:
		con->auth_retry++;
		dout("process_connect %p got BADAUTHORIZER attempt %d\n", con,
		     con->auth_retry);
		if (con->auth_retry == 2) {
			con->error_msg = "connect authorization failure";
			return -1;
		}
2060
		con_out_kvec_reset(con);
2061
		ret = prepare_write_connect(con);
2062 2063
		if (ret < 0)
			return ret;
Sage Weil's avatar
Sage Weil committed
2064
		prepare_read_connect(con);
2065
		break;
Sage Weil's avatar
Sage Weil committed
2066 2067 2068 2069 2070 2071 2072 2073 2074 2075

	case CEPH_MSGR_TAG_RESETSESSION:
		/*
		 * If we connected with a large connect_seq but the peer
		 * has no record of a session with us (no connection, or
		 * connect_seq == 0), they will send RESETSESION to indicate
		 * that they must have reset their session, and may have
		 * dropped messages.
		 */
		dout("process_connect got RESET peer seq %u\n",
Sage Weil's avatar
Sage Weil committed
2076
		     le32_to_cpu(con->in_reply.connect_seq));
Sage Weil's avatar
Sage Weil committed
2077 2078
		pr_err("%s%lld %s connection reset\n",
		       ENTITY_NAME(con->peer_name),
2079
		       ceph_pr_addr(&con->peer_addr.in_addr));
Sage Weil's avatar
Sage Weil committed
2080
		reset_connection(con);
2081
		con_out_kvec_reset(con);
2082 2083 2084
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
Sage Weil's avatar
Sage Weil committed
2085 2086 2087
		prepare_read_connect(con);

		/* Tell ceph about it. */
2088
		mutex_unlock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
2089 2090 2091
		pr_info("reset on %s%lld\n", ENTITY_NAME(con->peer_name));
		if (con->ops->peer_reset)
			con->ops->peer_reset(con);
2092
		mutex_lock(&con->mutex);
2093
		if (con->state != CON_STATE_NEGOTIATING)
2094
			return -EAGAIN;
Sage Weil's avatar
Sage Weil committed
2095 2096 2097 2098 2099 2100 2101
		break;

	case CEPH_MSGR_TAG_RETRY_SESSION:
		/*
		 * If we sent a smaller connect_seq than the peer has, try
		 * again with a larger value.
		 */
Sage Weil's avatar
Sage Weil committed
2102
		dout("process_connect got RETRY_SESSION my seq %u, peer %u\n",
Sage Weil's avatar
Sage Weil committed
2103
		     le32_to_cpu(con->out_connect.connect_seq),
Sage Weil's avatar
Sage Weil committed
2104 2105
		     le32_to_cpu(con->in_reply.connect_seq));
		con->connect_seq = le32_to_cpu(con->in_reply.connect_seq);
2106
		con_out_kvec_reset(con);
2107 2108 2109
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
Sage Weil's avatar
Sage Weil committed
2110 2111 2112 2113 2114 2115 2116 2117
		prepare_read_connect(con);
		break;

	case CEPH_MSGR_TAG_RETRY_GLOBAL:
		/*
		 * If we sent a smaller global_seq than the peer has, try
		 * again with a larger value.
		 */
2118
		dout("process_connect got RETRY_GLOBAL my %u peer_gseq %u\n",
Sage Weil's avatar
Sage Weil committed
2119
		     con->peer_global_seq,
Sage Weil's avatar
Sage Weil committed
2120
		     le32_to_cpu(con->in_reply.global_seq));
Sage Weil's avatar
Sage Weil committed
2121
		get_global_seq(con->msgr,
Sage Weil's avatar
Sage Weil committed
2122
			       le32_to_cpu(con->in_reply.global_seq));
2123
		con_out_kvec_reset(con);
2124 2125 2126
		ret = prepare_write_connect(con);
		if (ret < 0)
			return ret;
Sage Weil's avatar
Sage Weil committed
2127 2128 2129
		prepare_read_connect(con);
		break;

2130
	case CEPH_MSGR_TAG_SEQ:
Sage Weil's avatar
Sage Weil committed
2131
	case CEPH_MSGR_TAG_READY:
2132 2133 2134 2135
		if (req_feat & ~server_feat) {
			pr_err("%s%lld %s protocol feature mismatch,"
			       " my required %llx > server's %llx, need %llx\n",
			       ENTITY_NAME(con->peer_name),
2136
			       ceph_pr_addr(&con->peer_addr.in_addr),
2137 2138
			       req_feat, server_feat, req_feat & ~server_feat);
			con->error_msg = "missing required protocol features";
2139
			reset_connection(con);
2140 2141
			return -1;
		}
2142

2143
		WARN_ON(con->state != CON_STATE_NEGOTIATING);
2144
		con->state = CON_STATE_OPEN;
2145
		con->auth_retry = 0;    /* we authenticated; clear flag */
Sage Weil's avatar
Sage Weil committed
2146 2147
		con->peer_global_seq = le32_to_cpu(con->in_reply.global_seq);
		con->connect_seq++;
2148
		con->peer_features = server_feat;
Sage Weil's avatar
Sage Weil committed
2149 2150 2151 2152 2153 2154
		dout("process_connect got READY gseq %d cseq %d (%d)\n",
		     con->peer_global_seq,
		     le32_to_cpu(con->in_reply.connect_seq),
		     con->connect_seq);
		WARN_ON(con->connect_seq !=
			le32_to_cpu(con->in_reply.connect_seq));
2155 2156

		if (con->in_reply.flags & CEPH_MSG_CONNECT_LOSSY)
2157
			con_flag_set(con, CON_FLAG_LOSSYTX);
2158

2159
		con->delay = 0;      /* reset backoff memory */
2160

2161 2162 2163 2164 2165 2166
		if (con->in_reply.tag == CEPH_MSGR_TAG_SEQ) {
			prepare_write_seq(con);
			prepare_read_seq(con);
		} else {
			prepare_read_tag(con);
		}
Sage Weil's avatar
Sage Weil committed
2167 2168 2169 2170 2171 2172 2173 2174 2175
		break;

	case CEPH_MSGR_TAG_WAIT:
		/*
		 * If there is a connection race (we are opening
		 * connections to each other), one of us may just have
		 * to WAIT.  This shouldn't happen if we are the
		 * client.
		 */
Sage Weil's avatar
Sage Weil committed
2176 2177
		con->error_msg = "protocol error, got WAIT as client";
		return -1;
Sage Weil's avatar
Sage Weil committed
2178 2179 2180 2181 2182 2183 2184 2185 2186 2187 2188 2189 2190 2191

	default:
		con->error_msg = "protocol error, garbage tag during connect";
		return -1;
	}
	return 0;
}


/*
 * read (part of) an ack
 */
static int read_partial_ack(struct ceph_connection *con)
{
2192 2193
	int size = sizeof (con->in_temp_ack);
	int end = size;
Sage Weil's avatar
Sage Weil committed
2194

2195
	return read_partial(con, end, size, &con->in_temp_ack);
Sage Weil's avatar
Sage Weil committed
2196 2197 2198 2199 2200 2201 2202 2203 2204 2205 2206 2207 2208 2209 2210 2211 2212 2213 2214
}

/*
 * We can finally discard anything that's been acked.
 */
static void process_ack(struct ceph_connection *con)
{
	struct ceph_msg *m;
	u64 ack = le64_to_cpu(con->in_temp_ack);
	u64 seq;

	while (!list_empty(&con->out_sent)) {
		m = list_first_entry(&con->out_sent, struct ceph_msg,
				     list_head);
		seq = le64_to_cpu(m->hdr.seq);
		if (seq > ack)
			break;
		dout("got ack for seq %llu type %d at %p\n", seq,
		     le16_to_cpu(m->hdr.type), m);
2215
		m->ack_stamp = jiffies;
Sage Weil's avatar
Sage Weil committed
2216 2217 2218 2219 2220 2221
		ceph_msg_remove(m);
	}
	prepare_read_tag(con);
}


2222
static int read_partial_message_section(struct ceph_connection *con,
Sage Weil's avatar
Sage Weil committed
2223 2224
					struct kvec *section,
					unsigned int sec_len, u32 *crc)
2225
{
2226
	int ret, left;
2227 2228 2229 2230 2231 2232 2233 2234 2235 2236 2237 2238

	BUG_ON(!section);

	while (section->iov_len < sec_len) {
		BUG_ON(section->iov_base == NULL);
		left = sec_len - section->iov_len;
		ret = ceph_tcp_recvmsg(con->sock, (char *)section->iov_base +
				       section->iov_len, left);
		if (ret <= 0)
			return ret;
		section->iov_len += ret;
	}
2239 2240
	if (section->iov_len == sec_len)
		*crc = crc32c(0, section->iov_base, section->iov_len);
Sage Weil's avatar
Sage Weil committed
2241

2242 2243
	return 1;
}
Sage Weil's avatar
Sage Weil committed
2244

2245 2246 2247
static int read_partial_msg_data(struct ceph_connection *con)
{
	struct ceph_msg *msg = con->in_msg;
2248
	struct ceph_msg_data_cursor *cursor = &msg->cursor;
2249
	bool do_datacrc = !ceph_test_opt(from_msgr(con->msgr), NOCRC);
Alex Elder's avatar
Alex Elder committed
2250 2251 2252
	struct page *page;
	size_t page_offset;
	size_t length;
2253
	u32 crc = 0;
2254 2255 2256
	int ret;

	BUG_ON(!msg);
2257
	if (list_empty(&msg->data))
2258
		return -EIO;
2259

2260 2261
	if (do_datacrc)
		crc = con->in_data_crc;
2262
	while (cursor->resid) {
2263
		page = ceph_msg_data_next(cursor, &page_offset, &length, NULL);
Alex Elder's avatar
Alex Elder committed
2264
		ret = ceph_tcp_recvpage(con->sock, page, page_offset, length);
2265 2266 2267 2268
		if (ret <= 0) {
			if (do_datacrc)
				con->in_data_crc = crc;

Alex Elder's avatar
Alex Elder committed
2269
			return ret;
2270
		}
Alex Elder's avatar
Alex Elder committed
2271 2272

		if (do_datacrc)
2273
			crc = ceph_crc32c_page(crc, page, page_offset, ret);
2274
		(void) ceph_msg_data_advance(cursor, (size_t)ret);
2275
	}
2276 2277
	if (do_datacrc)
		con->in_data_crc = crc;
2278 2279 2280 2281

	return 1;	/* must return > 0 to indicate success */
}

Sage Weil's avatar
Sage Weil committed
2282 2283 2284
/*
 * read (part of) a message.
 */
Alex Elder's avatar
Alex Elder committed
2285 2286
static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip);

Sage Weil's avatar
Sage Weil committed
2287 2288 2289
static int read_partial_message(struct ceph_connection *con)
{
	struct ceph_msg *m = con->in_msg;
2290 2291
	int size;
	int end;
Sage Weil's avatar
Sage Weil committed
2292
	int ret;
2293
	unsigned int front_len, middle_len, data_len;
2294
	bool do_datacrc = !ceph_test_opt(from_msgr(con->msgr), NOCRC);
2295
	bool need_sign = (con->peer_features & CEPH_FEATURE_MSG_AUTH);
2296
	u64 seq;
2297
	u32 crc;
Sage Weil's avatar
Sage Weil committed
2298 2299 2300 2301

	dout("read_partial_message con %p msg %p\n", con, m);

	/* header */
2302 2303 2304
	size = sizeof (con->in_hdr);
	end = size;
	ret = read_partial(con, end, size, &con->in_hdr);
2305 2306
	if (ret <= 0)
		return ret;
2307 2308 2309

	crc = crc32c(0, &con->in_hdr, offsetof(struct ceph_msg_header, crc));
	if (cpu_to_le32(crc) != con->in_hdr.crc) {
2310
		pr_err("read_partial_message bad hdr crc %u != expected %u\n",
2311 2312 2313 2314
		       crc, con->in_hdr.crc);
		return -EBADMSG;
	}

Sage Weil's avatar
Sage Weil committed
2315 2316 2317 2318
	front_len = le32_to_cpu(con->in_hdr.front_len);
	if (front_len > CEPH_MSG_MAX_FRONT_LEN)
		return -EIO;
	middle_len = le32_to_cpu(con->in_hdr.middle_len);
2319
	if (middle_len > CEPH_MSG_MAX_MIDDLE_LEN)
Sage Weil's avatar
Sage Weil committed
2320 2321 2322 2323 2324
		return -EIO;
	data_len = le32_to_cpu(con->in_hdr.data_len);
	if (data_len > CEPH_MSG_MAX_DATA_LEN)
		return -EIO;

2325 2326 2327
	/* verify seq# */
	seq = le64_to_cpu(con->in_hdr.seq);
	if ((s64)seq - (s64)con->in_seq < 1) {
2328
		pr_info("skipping %s%lld %s seq %lld expected %lld\n",
2329
			ENTITY_NAME(con->peer_name),
2330
			ceph_pr_addr(&con->peer_addr.in_addr),
2331 2332
			seq, con->in_seq + 1);
		con->in_base_pos = -front_len - middle_len - data_len -
2333
			sizeof_footer(con);
2334
		con->in_tag = CEPH_MSGR_TAG_READY;
2335
		return 1;
2336 2337 2338 2339
	} else if ((s64)seq - (s64)con->in_seq > 1) {
		pr_err("read_partial_message bad seq %lld expected %lld\n",
		       seq, con->in_seq + 1);
		con->error_msg = "bad message sequence # for incoming message";
2340
		return -EBADE;
2341 2342
	}

Sage Weil's avatar
Sage Weil committed
2343 2344
	/* allocate message? */
	if (!con->in_msg) {
2345 2346
		int skip = 0;

Sage Weil's avatar
Sage Weil committed
2347
		dout("got hdr type %d front %d data %d\n", con->in_hdr.type,
2348
		     front_len, data_len);
2349 2350 2351
		ret = ceph_con_in_msg_alloc(con, &skip);
		if (ret < 0)
			return ret;
2352 2353

		BUG_ON(!con->in_msg ^ skip);
2354
		if (skip) {
Sage Weil's avatar
Sage Weil committed
2355
			/* skip this message */
2356
			dout("alloc_msg said skip message\n");
Sage Weil's avatar
Sage Weil committed
2357
			con->in_base_pos = -front_len - middle_len - data_len -
2358
				sizeof_footer(con);
Sage Weil's avatar
Sage Weil committed
2359
			con->in_tag = CEPH_MSGR_TAG_READY;
2360
			con->in_seq++;
2361
			return 1;
Sage Weil's avatar
Sage Weil committed
2362
		}
2363

2364
		BUG_ON(!con->in_msg);
2365
		BUG_ON(con->in_msg->con != con);
Sage Weil's avatar
Sage Weil committed
2366 2367
		m = con->in_msg;
		m->front.iov_len = 0;    /* haven't read it yet */
2368 2369
		if (m->middle)
			m->middle->vec.iov_len = 0;
2370

2371
		/* prepare for data payload, if any */
2372

2373
		if (data_len)
2374
			prepare_message_data(con->in_msg, data_len);
Sage Weil's avatar
Sage Weil committed
2375 2376 2377
	}

	/* front */
2378 2379 2380 2381
	ret = read_partial_message_section(con, &m->front, front_len,
					   &con->in_front_crc);
	if (ret <= 0)
		return ret;
Sage Weil's avatar
Sage Weil committed
2382 2383

	/* middle */
2384
	if (m->middle) {
Sage Weil's avatar
Sage Weil committed
2385 2386
		ret = read_partial_message_section(con, &m->middle->vec,
						   middle_len,
2387
						   &con->in_middle_crc);
Sage Weil's avatar
Sage Weil committed
2388 2389 2390 2391 2392
		if (ret <= 0)
			return ret;
	}

	/* (page) data */
2393 2394 2395 2396
	if (data_len) {
		ret = read_partial_msg_data(con);
		if (ret <= 0)
			return ret;
Sage Weil's avatar
Sage Weil committed
2397 2398 2399
	}

	/* footer */
2400
	size = sizeof_footer(con);
2401 2402
	end += size;
	ret = read_partial(con, end, size, &m->footer);
2403 2404 2405
	if (ret <= 0)
		return ret;

2406 2407 2408 2409 2410
	if (!need_sign) {
		m->footer.flags = m->old_footer.flags;
		m->footer.sig = 0;
	}

Sage Weil's avatar
Sage Weil committed
2411 2412 2413 2414 2415 2416 2417 2418 2419 2420 2421 2422 2423 2424 2425
	dout("read_partial_message got msg %p %d (%u) + %d (%u) + %d (%u)\n",
	     m, front_len, m->footer.front_crc, middle_len,
	     m->footer.middle_crc, data_len, m->footer.data_crc);

	/* crc ok? */
	if (con->in_front_crc != le32_to_cpu(m->footer.front_crc)) {
		pr_err("read_partial_message %p front crc %u != exp. %u\n",
		       m, con->in_front_crc, m->footer.front_crc);
		return -EBADMSG;
	}
	if (con->in_middle_crc != le32_to_cpu(m->footer.middle_crc)) {
		pr_err("read_partial_message %p middle crc %u != exp %u\n",
		       m, con->in_middle_crc, m->footer.middle_crc);
		return -EBADMSG;
	}
2426
	if (do_datacrc &&
Sage Weil's avatar
Sage Weil committed
2427 2428 2429 2430 2431 2432 2433
	    (m->footer.flags & CEPH_MSG_FOOTER_NOCRC) == 0 &&
	    con->in_data_crc != le32_to_cpu(m->footer.data_crc)) {
		pr_err("read_partial_message %p data crc %u != exp. %u\n", m,
		       con->in_data_crc, le32_to_cpu(m->footer.data_crc));
		return -EBADMSG;
	}

2434
	if (need_sign && con->ops->check_message_signature &&
2435
	    con->ops->check_message_signature(m)) {
2436 2437 2438 2439
		pr_err("read_partial_message %p signature check failed\n", m);
		return -EBADMSG;
	}

Sage Weil's avatar
Sage Weil committed
2440 2441 2442 2443 2444 2445 2446 2447 2448 2449
	return 1; /* done! */
}

/*
 * Process message.  This happens in the worker thread.  The callback should
 * be careful not to do anything that waits on other incoming messages or it
 * may deadlock.
 */
static void process_message(struct ceph_connection *con)
{
2450
	struct ceph_msg *msg = con->in_msg;
Sage Weil's avatar
Sage Weil committed
2451

2452
	BUG_ON(con->in_msg->con != con);
Sage Weil's avatar
Sage Weil committed
2453 2454 2455 2456
	con->in_msg = NULL;

	/* if first message, set peer_name */
	if (con->peer_name.type == 0)
2457
		con->peer_name = msg->hdr.src;
Sage Weil's avatar
Sage Weil committed
2458 2459

	con->in_seq++;
2460
	mutex_unlock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
2461 2462 2463

	dout("===== %p %llu from %s%lld %d=%s len %d+%d (%u %u %u) =====\n",
	     msg, le64_to_cpu(msg->hdr.seq),
2464
	     ENTITY_NAME(msg->hdr.src),
Sage Weil's avatar
Sage Weil committed
2465 2466 2467 2468 2469 2470
	     le16_to_cpu(msg->hdr.type),
	     ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
	     le32_to_cpu(msg->hdr.front_len),
	     le32_to_cpu(msg->hdr.data_len),
	     con->in_front_crc, con->in_middle_crc, con->in_data_crc);
	con->ops->dispatch(con, msg);
2471 2472

	mutex_lock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
2473 2474
}

2475 2476 2477 2478 2479 2480 2481 2482 2483 2484 2485
static int read_keepalive_ack(struct ceph_connection *con)
{
	struct ceph_timespec ceph_ts;
	size_t size = sizeof(ceph_ts);
	int ret = read_partial(con, size, size, &ceph_ts);
	if (ret <= 0)
		return ret;
	ceph_decode_timespec(&con->last_keepalive_ack, &ceph_ts);
	prepare_read_tag(con);
	return 1;
}
Sage Weil's avatar
Sage Weil committed
2486 2487 2488 2489 2490 2491 2492 2493 2494

/*
 * Write something to the socket.  Called in a worker thread when the
 * socket appears to be writeable and we have something ready to send.
 */
static int try_write(struct ceph_connection *con)
{
	int ret = 1;

2495
	dout("try_write start %p state %lu\n", con, con->state);
Sage Weil's avatar
Sage Weil committed
2496 2497 2498 2499 2500

more:
	dout("try_write out_kvec_bytes %d\n", con->out_kvec_bytes);

	/* open the socket first? */
2501 2502 2503
	if (con->state == CON_STATE_PREOPEN) {
		BUG_ON(con->sock);
		con->state = CON_STATE_CONNECTING;
2504

2505
		con_out_kvec_reset(con);
2506
		prepare_write_banner(con);
2507
		prepare_read_banner(con);
Sage Weil's avatar
Sage Weil committed
2508

2509
		BUG_ON(con->in_msg);
Sage Weil's avatar
Sage Weil committed
2510 2511 2512
		con->in_tag = CEPH_MSGR_TAG_READY;
		dout("try_write initiating connect on %p new state %lu\n",
		     con, con->state);
2513 2514
		ret = ceph_tcp_connect(con);
		if (ret < 0) {
Sage Weil's avatar
Sage Weil committed
2515 2516 2517 2518 2519 2520 2521
			con->error_msg = "connect error";
			goto out;
		}
	}

more_kvec:
	/* kvec data queued? */
Ilya Dryomov's avatar
Ilya Dryomov committed
2522 2523
	if (con->out_kvec_left) {
		ret = write_partial_kvec(con);
Sage Weil's avatar
Sage Weil committed
2524
		if (ret <= 0)
2525
			goto out;
Sage Weil's avatar
Sage Weil committed
2526
	}
Ilya Dryomov's avatar
Ilya Dryomov committed
2527 2528
	if (con->out_skip) {
		ret = write_partial_skip(con);
Sage Weil's avatar
Sage Weil committed
2529
		if (ret <= 0)
2530
			goto out;
Sage Weil's avatar
Sage Weil committed
2531 2532 2533 2534
	}

	/* msg pages? */
	if (con->out_msg) {
2535 2536 2537 2538 2539 2540
		if (con->out_msg_done) {
			ceph_msg_put(con->out_msg);
			con->out_msg = NULL;   /* we're done with this one */
			goto do_next;
		}

2541
		ret = write_partial_message_data(con);
Sage Weil's avatar
Sage Weil committed
2542 2543 2544
		if (ret == 1)
			goto more_kvec;  /* we need to send the footer, too! */
		if (ret == 0)
2545
			goto out;
Sage Weil's avatar
Sage Weil committed
2546
		if (ret < 0) {
2547
			dout("try_write write_partial_message_data err %d\n",
Sage Weil's avatar
Sage Weil committed
2548
			     ret);
2549
			goto out;
Sage Weil's avatar
Sage Weil committed
2550 2551 2552
		}
	}

2553
do_next:
2554
	if (con->state == CON_STATE_OPEN) {
2555 2556 2557 2558
		if (con_flag_test_and_clear(con, CON_FLAG_KEEPALIVE_PENDING)) {
			prepare_write_keepalive(con);
			goto more;
		}
Sage Weil's avatar
Sage Weil committed
2559 2560 2561 2562 2563 2564 2565 2566 2567 2568 2569 2570
		/* is anything else pending? */
		if (!list_empty(&con->out_queue)) {
			prepare_write_message(con);
			goto more;
		}
		if (con->in_seq > con->in_seq_acked) {
			prepare_write_ack(con);
			goto more;
		}
	}

	/* Nothing to do! */
2571
	con_flag_clear(con, CON_FLAG_WRITE_PENDING);
Sage Weil's avatar
Sage Weil committed
2572 2573 2574
	dout("try_write nothing else to write.\n");
	ret = 0;
out:
2575
	dout("try_write done on %p ret %d\n", con, ret);
Sage Weil's avatar
Sage Weil committed
2576 2577 2578 2579 2580 2581 2582 2583 2584 2585 2586 2587
	return ret;
}



/*
 * Read what we can from the socket.
 */
static int try_read(struct ceph_connection *con)
{
	int ret = -1;

2588 2589 2590 2591 2592
more:
	dout("try_read start on %p state %lu\n", con, con->state);
	if (con->state != CON_STATE_CONNECTING &&
	    con->state != CON_STATE_NEGOTIATING &&
	    con->state != CON_STATE_OPEN)
Sage Weil's avatar
Sage Weil committed
2593 2594
		return 0;

2595
	BUG_ON(!con->sock);
2596

Sage Weil's avatar
Sage Weil committed
2597 2598
	dout("try_read tag %d in_base_pos %d\n", (int)con->in_tag,
	     con->in_base_pos);
2599

2600
	if (con->state == CON_STATE_CONNECTING) {
2601 2602 2603
		dout("try_read connecting\n");
		ret = read_partial_banner(con);
		if (ret <= 0)
2604
			goto out;
2605 2606 2607 2608
		ret = process_banner(con);
		if (ret < 0)
			goto out;

2609
		con->state = CON_STATE_NEGOTIATING;
2610

2611 2612 2613 2614 2615
		/*
		 * Received banner is good, exchange connection info.
		 * Do not reset out_kvec, as sending our banner raced
		 * with receiving peer banner after connect completed.
		 */
2616 2617 2618 2619 2620 2621
		ret = prepare_write_connect(con);
		if (ret < 0)
			goto out;
		prepare_read_connect(con);

		/* Send connection info before awaiting response */
2622 2623 2624
		goto out;
	}

2625
	if (con->state == CON_STATE_NEGOTIATING) {
2626
		dout("try_read negotiating\n");
Sage Weil's avatar
Sage Weil committed
2627 2628 2629
		ret = read_partial_connect(con);
		if (ret <= 0)
			goto out;
2630 2631 2632
		ret = process_connect(con);
		if (ret < 0)
			goto out;
Sage Weil's avatar
Sage Weil committed
2633 2634 2635
		goto more;
	}

2636
	WARN_ON(con->state != CON_STATE_OPEN);
2637

Sage Weil's avatar
Sage Weil committed
2638 2639 2640 2641 2642 2643
	if (con->in_base_pos < 0) {
		/*
		 * skipping + discarding content.
		 *
		 * FIXME: there must be a better way to do this!
		 */
Alex Elder's avatar
Alex Elder committed
2644 2645 2646
		static char buf[SKIP_BUF_SIZE];
		int skip = min((int) sizeof (buf), -con->in_base_pos);

Sage Weil's avatar
Sage Weil committed
2647 2648 2649
		dout("skipping %d / %d bytes\n", skip, -con->in_base_pos);
		ret = ceph_tcp_recvmsg(con->sock, buf, skip);
		if (ret <= 0)
2650
			goto out;
Sage Weil's avatar
Sage Weil committed
2651 2652 2653 2654 2655 2656 2657 2658 2659 2660
		con->in_base_pos += ret;
		if (con->in_base_pos)
			goto more;
	}
	if (con->in_tag == CEPH_MSGR_TAG_READY) {
		/*
		 * what's next?
		 */
		ret = ceph_tcp_recvmsg(con->sock, &con->in_tag, 1);
		if (ret <= 0)
2661
			goto out;
Sage Weil's avatar
Sage Weil committed
2662 2663 2664 2665 2666 2667 2668 2669
		dout("try_read got tag %d\n", (int)con->in_tag);
		switch (con->in_tag) {
		case CEPH_MSGR_TAG_MSG:
			prepare_read_message(con);
			break;
		case CEPH_MSGR_TAG_ACK:
			prepare_read_ack(con);
			break;
2670 2671 2672
		case CEPH_MSGR_TAG_KEEPALIVE2_ACK:
			prepare_read_keepalive_ack(con);
			break;
Sage Weil's avatar
Sage Weil committed
2673
		case CEPH_MSGR_TAG_CLOSE:
2674 2675
			con_close_socket(con);
			con->state = CON_STATE_CLOSED;
2676
			goto out;
Sage Weil's avatar
Sage Weil committed
2677 2678 2679 2680 2681 2682 2683 2684 2685
		default:
			goto bad_tag;
		}
	}
	if (con->in_tag == CEPH_MSGR_TAG_MSG) {
		ret = read_partial_message(con);
		if (ret <= 0) {
			switch (ret) {
			case -EBADMSG:
2686
				con->error_msg = "bad crc/signature";
2687 2688
				/* fall through */
			case -EBADE:
Sage Weil's avatar
Sage Weil committed
2689
				ret = -EIO;
2690
				break;
Sage Weil's avatar
Sage Weil committed
2691 2692
			case -EIO:
				con->error_msg = "io error";
2693
				break;
Sage Weil's avatar
Sage Weil committed
2694
			}
2695
			goto out;
Sage Weil's avatar
Sage Weil committed
2696 2697 2698 2699
		}
		if (con->in_tag == CEPH_MSGR_TAG_READY)
			goto more;
		process_message(con);
2700 2701
		if (con->state == CON_STATE_OPEN)
			prepare_read_tag(con);
Sage Weil's avatar
Sage Weil committed
2702 2703
		goto more;
	}
2704 2705 2706 2707 2708 2709
	if (con->in_tag == CEPH_MSGR_TAG_ACK ||
	    con->in_tag == CEPH_MSGR_TAG_SEQ) {
		/*
		 * the final handshake seq exchange is semantically
		 * equivalent to an ACK
		 */
Sage Weil's avatar
Sage Weil committed
2710 2711
		ret = read_partial_ack(con);
		if (ret <= 0)
2712
			goto out;
Sage Weil's avatar
Sage Weil committed
2713 2714 2715
		process_ack(con);
		goto more;
	}
2716 2717 2718 2719 2720 2721
	if (con->in_tag == CEPH_MSGR_TAG_KEEPALIVE2_ACK) {
		ret = read_keepalive_ack(con);
		if (ret <= 0)
			goto out;
		goto more;
	}
Sage Weil's avatar
Sage Weil committed
2722 2723

out:
2724
	dout("try_read done on %p ret %d\n", con, ret);
Sage Weil's avatar
Sage Weil committed
2725 2726 2727 2728 2729 2730 2731 2732 2733 2734 2735
	return ret;

bad_tag:
	pr_err("try_read bad con->in_tag = %d\n", (int)con->in_tag);
	con->error_msg = "protocol error, garbage tag";
	ret = -1;
	goto out;
}


/*
2736 2737 2738
 * Atomically queue work on a connection after the specified delay.
 * Bump @con reference to avoid races with connection teardown.
 * Returns 0 if work was queued, or an error code otherwise.
Sage Weil's avatar
Sage Weil committed
2739
 */
2740
static int queue_con_delay(struct ceph_connection *con, unsigned long delay)
Sage Weil's avatar
Sage Weil committed
2741 2742
{
	if (!con->ops->get(con)) {
2743 2744
		dout("%s %p ref count 0\n", __func__, con);
		return -ENOENT;
Sage Weil's avatar
Sage Weil committed
2745 2746
	}

2747 2748
	if (!queue_delayed_work(ceph_msgr_wq, &con->work, delay)) {
		dout("%s %p - already queued\n", __func__, con);
Sage Weil's avatar
Sage Weil committed
2749
		con->ops->put(con);
2750
		return -EBUSY;
Sage Weil's avatar
Sage Weil committed
2751
	}
2752 2753 2754 2755 2756 2757 2758 2759

	dout("%s %p %lu\n", __func__, con, delay);
	return 0;
}

static void queue_con(struct ceph_connection *con)
{
	(void) queue_con_delay(con, 0);
Sage Weil's avatar
Sage Weil committed
2760 2761
}

2762 2763 2764 2765 2766 2767 2768 2769
static void cancel_con(struct ceph_connection *con)
{
	if (cancel_delayed_work(&con->work)) {
		dout("%s %p\n", __func__, con);
		con->ops->put(con);
	}
}

2770 2771
static bool con_sock_closed(struct ceph_connection *con)
{
2772
	if (!con_flag_test_and_clear(con, CON_FLAG_SOCK_CLOSED))
2773 2774 2775 2776 2777 2778 2779 2780 2781 2782 2783 2784 2785 2786 2787
		return false;

#define CASE(x)								\
	case CON_STATE_ ## x:						\
		con->error_msg = "socket closed (con state " #x ")";	\
		break;

	switch (con->state) {
	CASE(CLOSED);
	CASE(PREOPEN);
	CASE(CONNECTING);
	CASE(NEGOTIATING);
	CASE(OPEN);
	CASE(STANDBY);
	default:
2788
		pr_warn("%s con %p unrecognized state %lu\n",
2789 2790 2791 2792 2793 2794 2795 2796 2797 2798
			__func__, con, con->state);
		con->error_msg = "unrecognized con state";
		BUG();
		break;
	}
#undef CASE

	return true;
}

2799 2800 2801 2802 2803 2804 2805 2806 2807 2808 2809 2810 2811 2812 2813 2814 2815 2816
static bool con_backoff(struct ceph_connection *con)
{
	int ret;

	if (!con_flag_test_and_clear(con, CON_FLAG_BACKOFF))
		return false;

	ret = queue_con_delay(con, round_jiffies_relative(con->delay));
	if (ret) {
		dout("%s: con %p FAILED to back off %lu\n", __func__,
			con, con->delay);
		BUG_ON(ret == -ENOENT);
		con_flag_set(con, CON_FLAG_BACKOFF);
	}

	return true;
}

2817 2818 2819 2820
/* Finish fault handling; con->mutex must *not* be held here */

static void con_fault_finish(struct ceph_connection *con)
{
2821 2822
	dout("%s %p\n", __func__, con);

2823 2824 2825 2826
	/*
	 * in case we faulted due to authentication, invalidate our
	 * current tickets so that we can get new ones.
	 */
2827 2828 2829 2830 2831
	if (con->auth_retry) {
		dout("auth_retry %d, invalidating\n", con->auth_retry);
		if (con->ops->invalidate_authorizer)
			con->ops->invalidate_authorizer(con);
		con->auth_retry = 0;
2832 2833 2834 2835 2836 2837
	}

	if (con->ops->fault)
		con->ops->fault(con);
}

Sage Weil's avatar
Sage Weil committed
2838 2839 2840
/*
 * Do some work on a connection.  Drop a connection ref when we're done.
 */
2841
static void ceph_con_workfn(struct work_struct *work)
Sage Weil's avatar
Sage Weil committed
2842 2843 2844
{
	struct ceph_connection *con = container_of(work, struct ceph_connection,
						   work.work);
2845
	bool fault;
Sage Weil's avatar
Sage Weil committed
2846

Sage Weil's avatar
Sage Weil committed
2847
	mutex_lock(&con->mutex);
2848 2849
	while (true) {
		int ret;
Sage Weil's avatar
Sage Weil committed
2850

2851 2852 2853 2854 2855 2856 2857 2858 2859 2860 2861 2862 2863 2864 2865 2866 2867 2868 2869 2870 2871
		if ((fault = con_sock_closed(con))) {
			dout("%s: con %p SOCK_CLOSED\n", __func__, con);
			break;
		}
		if (con_backoff(con)) {
			dout("%s: con %p BACKOFF\n", __func__, con);
			break;
		}
		if (con->state == CON_STATE_STANDBY) {
			dout("%s: con %p STANDBY\n", __func__, con);
			break;
		}
		if (con->state == CON_STATE_CLOSED) {
			dout("%s: con %p CLOSED\n", __func__, con);
			BUG_ON(con->sock);
			break;
		}
		if (con->state == CON_STATE_PREOPEN) {
			dout("%s: con %p PREOPEN\n", __func__, con);
			BUG_ON(con->sock);
		}
2872

2873 2874 2875 2876
		ret = try_read(con);
		if (ret < 0) {
			if (ret == -EAGAIN)
				continue;
2877 2878
			if (!con->error_msg)
				con->error_msg = "socket error on read";
2879 2880 2881 2882 2883 2884 2885 2886
			fault = true;
			break;
		}

		ret = try_write(con);
		if (ret < 0) {
			if (ret == -EAGAIN)
				continue;
2887 2888
			if (!con->error_msg)
				con->error_msg = "socket error on write";
2889 2890 2891 2892
			fault = true;
		}

		break;	/* If we make it to here, we're done */
2893
	}
2894 2895
	if (fault)
		con_fault(con);
Sage Weil's avatar
Sage Weil committed
2896
	mutex_unlock(&con->mutex);
2897

2898 2899 2900 2901
	if (fault)
		con_fault_finish(con);

	con->ops->put(con);
Sage Weil's avatar
Sage Weil committed
2902 2903 2904 2905 2906 2907
}

/*
 * Generic error/fault handler.  A retry mechanism is used with
 * exponential backoff
 */
2908
static void con_fault(struct ceph_connection *con)
Sage Weil's avatar
Sage Weil committed
2909 2910
{
	dout("fault %p state %lu to peer %s\n",
2911
	     con, con->state, ceph_pr_addr(&con->peer_addr.in_addr));
Sage Weil's avatar
Sage Weil committed
2912

2913 2914 2915 2916
	pr_warn("%s%lld %s %s\n", ENTITY_NAME(con->peer_name),
		ceph_pr_addr(&con->peer_addr.in_addr), con->error_msg);
	con->error_msg = NULL;

2917
	WARN_ON(con->state != CON_STATE_CONNECTING &&
2918 2919
	       con->state != CON_STATE_NEGOTIATING &&
	       con->state != CON_STATE_OPEN);
2920

Sage Weil's avatar
Sage Weil committed
2921
	con_close_socket(con);
Sage Weil's avatar
Sage Weil committed
2922

2923
	if (con_flag_test(con, CON_FLAG_LOSSYTX)) {
2924 2925
		dout("fault on LOSSYTX channel, marking CLOSED\n");
		con->state = CON_STATE_CLOSED;
2926
		return;
2927 2928
	}

Sage Weil's avatar
Sage Weil committed
2929
	if (con->in_msg) {
2930
		BUG_ON(con->in_msg->con != con);
Sage Weil's avatar
Sage Weil committed
2931 2932 2933
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
	}
Sage Weil's avatar
Sage Weil committed
2934

2935 2936
	/* Requeue anything that hasn't been acked */
	list_splice_init(&con->out_sent, &con->out_queue);
2937

Sage Weil's avatar
Sage Weil committed
2938 2939 2940
	/* If there are no messages queued or keepalive pending, place
	 * the connection in a STANDBY state */
	if (list_empty(&con->out_queue) &&
2941
	    !con_flag_test(con, CON_FLAG_KEEPALIVE_PENDING)) {
Sage Weil's avatar
Sage Weil committed
2942
		dout("fault %p setting STANDBY clearing WRITE_PENDING\n", con);
2943
		con_flag_clear(con, CON_FLAG_WRITE_PENDING);
2944
		con->state = CON_STATE_STANDBY;
2945 2946
	} else {
		/* retry after a delay. */
2947
		con->state = CON_STATE_PREOPEN;
2948 2949 2950 2951
		if (con->delay == 0)
			con->delay = BASE_DELAY_INTERVAL;
		else if (con->delay < MAX_DELAY_INTERVAL)
			con->delay *= 2;
2952
		con_flag_set(con, CON_FLAG_BACKOFF);
2953
		queue_con(con);
Sage Weil's avatar
Sage Weil committed
2954 2955 2956 2957 2958 2959
	}
}



/*
2960
 * initialize a new messenger instance
Sage Weil's avatar
Sage Weil committed
2961
 */
2962
void ceph_messenger_init(struct ceph_messenger *msgr,
2963
			 struct ceph_entity_addr *myaddr)
Sage Weil's avatar
Sage Weil committed
2964 2965 2966 2967 2968 2969 2970
{
	spin_lock_init(&msgr->global_seq_lock);

	if (myaddr)
		msgr->inst.addr = *myaddr;

	/* select a random nonce */
2971
	msgr->inst.addr.type = 0;
Sage Weil's avatar
Sage Weil committed
2972
	get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
2973
	encode_my_addr(msgr);
Sage Weil's avatar
Sage Weil committed
2974

2975
	atomic_set(&msgr->stopping, 0);
2976
	write_pnet(&msgr->net, get_net(current->nsproxy->net_ns));
Sage Weil's avatar
Sage Weil committed
2977

2978
	dout("%s %p\n", __func__, msgr);
Sage Weil's avatar
Sage Weil committed
2979
}
2980
EXPORT_SYMBOL(ceph_messenger_init);
Sage Weil's avatar
Sage Weil committed
2981

2982 2983 2984 2985 2986 2987
void ceph_messenger_fini(struct ceph_messenger *msgr)
{
	put_net(read_pnet(&msgr->net));
}
EXPORT_SYMBOL(ceph_messenger_fini);

2988 2989 2990 2991 2992 2993 2994 2995 2996
static void msg_con_set(struct ceph_msg *msg, struct ceph_connection *con)
{
	if (msg->con)
		msg->con->ops->put(msg->con);

	msg->con = con ? con->ops->get(con) : NULL;
	BUG_ON(msg->con != con);
}

Sage Weil's avatar
Sage Weil committed
2997 2998 2999
static void clear_standby(struct ceph_connection *con)
{
	/* come back from STANDBY? */
3000
	if (con->state == CON_STATE_STANDBY) {
Sage Weil's avatar
Sage Weil committed
3001
		dout("clear_standby %p and ++connect_seq\n", con);
3002
		con->state = CON_STATE_PREOPEN;
Sage Weil's avatar
Sage Weil committed
3003
		con->connect_seq++;
3004 3005
		WARN_ON(con_flag_test(con, CON_FLAG_WRITE_PENDING));
		WARN_ON(con_flag_test(con, CON_FLAG_KEEPALIVE_PENDING));
Sage Weil's avatar
Sage Weil committed
3006 3007 3008
	}
}

Sage Weil's avatar
Sage Weil committed
3009 3010 3011 3012 3013 3014
/*
 * Queue up an outgoing message on the given connection.
 */
void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg)
{
	/* set src+dst */
3015
	msg->hdr.src = con->msgr->inst.name;
3016
	BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len));
3017 3018
	msg->needs_out_seq = true;

3019
	mutex_lock(&con->mutex);
3020

3021
	if (con->state == CON_STATE_CLOSED) {
3022 3023 3024 3025 3026 3027
		dout("con_send %p closed, dropping %p\n", con, msg);
		ceph_msg_put(msg);
		mutex_unlock(&con->mutex);
		return;
	}

3028
	msg_con_set(msg, con);
3029

Sage Weil's avatar
Sage Weil committed
3030 3031 3032 3033 3034 3035 3036 3037
	BUG_ON(!list_empty(&msg->list_head));
	list_add_tail(&msg->list_head, &con->out_queue);
	dout("----- %p to %s%lld %d=%s len %d+%d+%d -----\n", msg,
	     ENTITY_NAME(con->peer_name), le16_to_cpu(msg->hdr.type),
	     ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
	     le32_to_cpu(msg->hdr.front_len),
	     le32_to_cpu(msg->hdr.middle_len),
	     le32_to_cpu(msg->hdr.data_len));
3038 3039

	clear_standby(con);
3040
	mutex_unlock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
3041 3042 3043

	/* if there wasn't anything waiting to send before, queue
	 * new work */
3044
	if (con_flag_test_and_set(con, CON_FLAG_WRITE_PENDING) == 0)
Sage Weil's avatar
Sage Weil committed
3045 3046
		queue_con(con);
}
3047
EXPORT_SYMBOL(ceph_con_send);
Sage Weil's avatar
Sage Weil committed
3048 3049 3050 3051

/*
 * Revoke a message that was previously queued for send
 */
3052
void ceph_msg_revoke(struct ceph_msg *msg)
Sage Weil's avatar
Sage Weil committed
3053
{
3054 3055
	struct ceph_connection *con = msg->con;

3056 3057
	if (!con) {
		dout("%s msg %p null con\n", __func__, msg);
3058
		return;		/* Message not in our possession */
3059
	}
3060

3061
	mutex_lock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
3062
	if (!list_empty(&msg->list_head)) {
3063
		dout("%s %p msg %p - was on queue\n", __func__, con, msg);
Sage Weil's avatar
Sage Weil committed
3064 3065
		list_del_init(&msg->list_head);
		msg->hdr.seq = 0;
3066

Sage Weil's avatar
Sage Weil committed
3067
		ceph_msg_put(msg);
Sage Weil's avatar
Sage Weil committed
3068 3069
	}
	if (con->out_msg == msg) {
Ilya Dryomov's avatar
Ilya Dryomov committed
3070 3071 3072 3073 3074 3075
		BUG_ON(con->out_skip);
		/* footer */
		if (con->out_msg_done) {
			con->out_skip += con_out_kvec_skip(con);
		} else {
			BUG_ON(!msg->data_length);
3076
			con->out_skip += sizeof_footer(con);
Sage Weil's avatar
Sage Weil committed
3077
		}
Ilya Dryomov's avatar
Ilya Dryomov committed
3078 3079 3080 3081 3082 3083 3084 3085 3086
		/* data, middle, front */
		if (msg->data_length)
			con->out_skip += msg->cursor.total_resid;
		if (msg->middle)
			con->out_skip += con_out_kvec_skip(con);
		con->out_skip += con_out_kvec_skip(con);

		dout("%s %p msg %p - was sending, will write %d skip %d\n",
		     __func__, con, msg, con->out_kvec_bytes, con->out_skip);
Sage Weil's avatar
Sage Weil committed
3087
		msg->hdr.seq = 0;
Ilya Dryomov's avatar
Ilya Dryomov committed
3088
		con->out_msg = NULL;
3089
		ceph_msg_put(msg);
Sage Weil's avatar
Sage Weil committed
3090
	}
Ilya Dryomov's avatar
Ilya Dryomov committed
3091

3092
	mutex_unlock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
3093 3094
}

3095
/*
3096
 * Revoke a message that we may be reading data into
3097
 */
3098
void ceph_msg_revoke_incoming(struct ceph_msg *msg)
3099
{
3100
	struct ceph_connection *con = msg->con;
3101

3102
	if (!con) {
3103 3104 3105 3106
		dout("%s msg %p null con\n", __func__, msg);
		return;		/* Message not in our possession */
	}

3107
	mutex_lock(&con->mutex);
3108
	if (con->in_msg == msg) {
3109 3110 3111
		unsigned int front_len = le32_to_cpu(con->in_hdr.front_len);
		unsigned int middle_len = le32_to_cpu(con->in_hdr.middle_len);
		unsigned int data_len = le32_to_cpu(con->in_hdr.data_len);
3112 3113

		/* skip rest of message */
3114 3115
		dout("%s %p msg %p revoked\n", __func__, con, msg);
		con->in_base_pos = con->in_base_pos -
3116
				sizeof(struct ceph_msg_header) -
3117 3118 3119
				front_len -
				middle_len -
				data_len -
3120 3121 3122 3123
				sizeof(struct ceph_msg_footer);
		ceph_msg_put(con->in_msg);
		con->in_msg = NULL;
		con->in_tag = CEPH_MSGR_TAG_READY;
3124
		con->in_seq++;
3125
	} else {
3126 3127
		dout("%s %p in_msg %p msg %p no-op\n",
		     __func__, con, con->in_msg, msg);
3128 3129 3130 3131
	}
	mutex_unlock(&con->mutex);
}

Sage Weil's avatar
Sage Weil committed
3132 3133 3134 3135 3136
/*
 * Queue a keepalive byte to ensure the tcp connection is alive.
 */
void ceph_con_keepalive(struct ceph_connection *con)
{
Sage Weil's avatar
Sage Weil committed
3137
	dout("con_keepalive %p\n", con);
3138
	mutex_lock(&con->mutex);
Sage Weil's avatar
Sage Weil committed
3139
	clear_standby(con);
3140
	mutex_unlock(&con->mutex);
3141 3142
	if (con_flag_test_and_set(con, CON_FLAG_KEEPALIVE_PENDING) == 0 &&
	    con_flag_test_and_set(con, CON_FLAG_WRITE_PENDING) == 0)
Sage Weil's avatar
Sage Weil committed
3143 3144
		queue_con(con);
}
3145
EXPORT_SYMBOL(ceph_con_keepalive);
Sage Weil's avatar
Sage Weil committed
3146

3147 3148 3149 3150 3151 3152 3153 3154 3155 3156 3157 3158 3159 3160
bool ceph_con_keepalive_expired(struct ceph_connection *con,
			       unsigned long interval)
{
	if (interval > 0 &&
	    (con->peer_features & CEPH_FEATURE_MSGR_KEEPALIVE2)) {
		struct timespec now = CURRENT_TIME;
		struct timespec ts;
		jiffies_to_timespec(interval, &ts);
		ts = timespec_add(con->last_keepalive_ack, ts);
		return timespec_compare(&now, &ts) >= 0;
	}
	return false;
}

3161
static struct ceph_msg_data *ceph_msg_data_create(enum ceph_msg_data_type type)
Alex Elder's avatar
Alex Elder committed
3162
{
3163 3164 3165 3166 3167
	struct ceph_msg_data *data;

	if (WARN_ON(!ceph_msg_data_type_valid(type)))
		return NULL;

3168
	data = kmem_cache_zalloc(ceph_msg_data_cache, GFP_NOFS);
3169 3170
	if (data)
		data->type = type;
3171
	INIT_LIST_HEAD(&data->links);
3172 3173 3174 3175 3176 3177 3178 3179 3180

	return data;
}

static void ceph_msg_data_destroy(struct ceph_msg_data *data)
{
	if (!data)
		return;

3181
	WARN_ON(!list_empty(&data->links));
3182
	if (data->type == CEPH_MSG_DATA_PAGELIST)
3183
		ceph_pagelist_release(data->pagelist);
3184
	kmem_cache_free(ceph_msg_data_cache, data);
Alex Elder's avatar
Alex Elder committed
3185 3186
}

3187
void ceph_msg_data_add_pages(struct ceph_msg *msg, struct page **pages,
3188
		size_t length, size_t alignment)
3189
{
3190 3191
	struct ceph_msg_data *data;

3192 3193
	BUG_ON(!pages);
	BUG_ON(!length);
3194 3195 3196 3197 3198 3199

	data = ceph_msg_data_create(CEPH_MSG_DATA_PAGES);
	BUG_ON(!data);
	data->pages = pages;
	data->length = length;
	data->alignment = alignment & ~PAGE_MASK;
3200

3201 3202
	list_add_tail(&data->links, &msg->data);
	msg->data_length += length;
3203
}
3204
EXPORT_SYMBOL(ceph_msg_data_add_pages);
Sage Weil's avatar
Sage Weil committed
3205

3206
void ceph_msg_data_add_pagelist(struct ceph_msg *msg,
3207 3208
				struct ceph_pagelist *pagelist)
{
3209 3210
	struct ceph_msg_data *data;

3211 3212
	BUG_ON(!pagelist);
	BUG_ON(!pagelist->length);
3213

3214 3215 3216 3217
	data = ceph_msg_data_create(CEPH_MSG_DATA_PAGELIST);
	BUG_ON(!data);
	data->pagelist = pagelist;

3218 3219
	list_add_tail(&data->links, &msg->data);
	msg->data_length += pagelist->length;
3220
}
3221
EXPORT_SYMBOL(ceph_msg_data_add_pagelist);
3222

3223
#ifdef	CONFIG_BLOCK
3224
void ceph_msg_data_add_bio(struct ceph_msg *msg, struct bio *bio,
3225
		size_t length)
3226
{
3227 3228
	struct ceph_msg_data *data;

3229
	BUG_ON(!bio);
3230

3231 3232 3233
	data = ceph_msg_data_create(CEPH_MSG_DATA_BIO);
	BUG_ON(!data);
	data->bio = bio;
Alex Elder's avatar
Alex Elder committed
3234
	data->bio_length = length;
3235

3236 3237
	list_add_tail(&data->links, &msg->data);
	msg->data_length += length;
3238
}
3239
EXPORT_SYMBOL(ceph_msg_data_add_bio);
3240
#endif	/* CONFIG_BLOCK */
3241

Sage Weil's avatar
Sage Weil committed
3242 3243 3244 3245
/*
 * construct a new message with given type, size
 * the new msg has a ref count of 1.
 */
3246 3247
struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags,
			      bool can_fail)
Sage Weil's avatar
Sage Weil committed
3248 3249 3250
{
	struct ceph_msg *m;

3251
	m = kmem_cache_zalloc(ceph_msg_cache, flags);
Sage Weil's avatar
Sage Weil committed
3252 3253 3254 3255
	if (m == NULL)
		goto out;

	m->hdr.type = cpu_to_le16(type);
3256
	m->hdr.priority = cpu_to_le16(CEPH_MSG_PRIO_DEFAULT);
Sage Weil's avatar
Sage Weil committed
3257
	m->hdr.front_len = cpu_to_le32(front_len);
3258

3259 3260
	INIT_LIST_HEAD(&m->list_head);
	kref_init(&m->kref);
3261
	INIT_LIST_HEAD(&m->data);
3262

Sage Weil's avatar
Sage Weil committed
3263 3264
	/* front */
	if (front_len) {
3265
		m->front.iov_base = ceph_kvmalloc(front_len, flags);
Sage Weil's avatar
Sage Weil committed
3266
		if (m->front.iov_base == NULL) {
3267
			dout("ceph_msg_new can't allocate %d bytes\n",
Sage Weil's avatar
Sage Weil committed
3268 3269 3270 3271 3272 3273
			     front_len);
			goto out2;
		}
	} else {
		m->front.iov_base = NULL;
	}
3274
	m->front_alloc_len = m->front.iov_len = front_len;
Sage Weil's avatar
Sage Weil committed
3275

Sage Weil's avatar
Sage Weil committed
3276
	dout("ceph_msg_new %p front %d\n", m, front_len);
Sage Weil's avatar
Sage Weil committed
3277 3278 3279 3280 3281
	return m;

out2:
	ceph_msg_put(m);
out:
3282 3283 3284
	if (!can_fail) {
		pr_err("msg_new can't create type %d front %d\n", type,
		       front_len);
3285
		WARN_ON(1);
3286 3287 3288 3289
	} else {
		dout("msg_new can't create type %d front %d\n", type,
		     front_len);
	}
3290
	return NULL;
Sage Weil's avatar
Sage Weil committed
3291
}
3292
EXPORT_SYMBOL(ceph_msg_new);
Sage Weil's avatar
Sage Weil committed
3293 3294 3295 3296 3297 3298 3299 3300

/*
 * Allocate "middle" portion of a message, if it is needed and wasn't
 * allocated by alloc_msg.  This allows us to read a small fixed-size
 * per-type header in the front and then gracefully fail (i.e.,
 * propagate the error to the caller based on info in the front) when
 * the middle is too large.
 */
3301
static int ceph_alloc_middle(struct ceph_connection *con, struct ceph_msg *msg)
Sage Weil's avatar
Sage Weil committed
3302 3303 3304 3305 3306 3307 3308 3309 3310
{
	int type = le16_to_cpu(msg->hdr.type);
	int middle_len = le32_to_cpu(msg->hdr.middle_len);

	dout("alloc_middle %p type %d %s middle_len %d\n", msg, type,
	     ceph_msg_type_name(type), middle_len);
	BUG_ON(!middle_len);
	BUG_ON(msg->middle);

3311
	msg->middle = ceph_buffer_new(middle_len, GFP_NOFS);
Sage Weil's avatar
Sage Weil committed
3312 3313 3314 3315 3316
	if (!msg->middle)
		return -ENOMEM;
	return 0;
}

3317
/*
Alex Elder's avatar
Alex Elder committed
3318 3319 3320 3321
 * Allocate a message for receiving an incoming message on a
 * connection, and save the result in con->in_msg.  Uses the
 * connection's private alloc_msg op if available.
 *
3322 3323 3324 3325 3326 3327 3328 3329 3330
 * Returns 0 on success, or a negative error code.
 *
 * On success, if we set *skip = 1:
 *  - the next message should be skipped and ignored.
 *  - con->in_msg == NULL
 * or if we set *skip = 0:
 *  - con->in_msg is non-null.
 * On error (ENOMEM, EAGAIN, ...),
 *  - con->in_msg == NULL
3331
 */
3332
static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip)
3333
{
3334
	struct ceph_msg_header *hdr = &con->in_hdr;
3335
	int middle_len = le32_to_cpu(hdr->middle_len);
3336
	struct ceph_msg *msg;
3337
	int ret = 0;
3338

Alex Elder's avatar
Alex Elder committed
3339
	BUG_ON(con->in_msg != NULL);
3340
	BUG_ON(!con->ops->alloc_msg);
3341

3342 3343 3344 3345 3346
	mutex_unlock(&con->mutex);
	msg = con->ops->alloc_msg(con, hdr, skip);
	mutex_lock(&con->mutex);
	if (con->state != CON_STATE_OPEN) {
		if (msg)
3347
			ceph_msg_put(msg);
3348 3349
		return -EAGAIN;
	}
3350 3351
	if (msg) {
		BUG_ON(*skip);
3352
		msg_con_set(msg, con);
3353 3354 3355 3356 3357 3358 3359 3360 3361 3362
		con->in_msg = msg;
	} else {
		/*
		 * Null message pointer means either we should skip
		 * this message or we couldn't allocate memory.  The
		 * former is not an error.
		 */
		if (*skip)
			return 0;

3363
		con->error_msg = "error allocating memory for incoming message";
3364
		return -ENOMEM;
3365
	}
Alex Elder's avatar
Alex Elder committed
3366
	memcpy(&con->in_msg->hdr, &con->in_hdr, sizeof(con->in_hdr));
3367

Alex Elder's avatar
Alex Elder committed
3368 3369
	if (middle_len && !con->in_msg->middle) {
		ret = ceph_alloc_middle(con, con->in_msg);
3370
		if (ret < 0) {
Alex Elder's avatar
Alex Elder committed
3371 3372
			ceph_msg_put(con->in_msg);
			con->in_msg = NULL;
3373 3374
		}
	}
3375

3376
	return ret;
3377 3378
}

Sage Weil's avatar
Sage Weil committed
3379 3380 3381 3382

/*
 * Free a generically kmalloc'd message.
 */
3383
static void ceph_msg_free(struct ceph_msg *m)
Sage Weil's avatar
Sage Weil committed
3384
{
3385
	dout("%s %p\n", __func__, m);
Ilya Dryomov's avatar
Ilya Dryomov committed
3386
	kvfree(m->front.iov_base);
3387
	kmem_cache_free(ceph_msg_cache, m);
Sage Weil's avatar
Sage Weil committed
3388 3389
}

3390
static void ceph_msg_release(struct kref *kref)
Sage Weil's avatar
Sage Weil committed
3391 3392
{
	struct ceph_msg *m = container_of(kref, struct ceph_msg, kref);
3393
	struct ceph_msg_data *data, *next;
Sage Weil's avatar
Sage Weil committed
3394

3395
	dout("%s %p\n", __func__, m);
Sage Weil's avatar
Sage Weil committed
3396 3397
	WARN_ON(!list_empty(&m->list_head));

3398 3399
	msg_con_set(m, NULL);

Sage Weil's avatar
Sage Weil committed
3400 3401 3402 3403
	/* drop middle, data, if any */
	if (m->middle) {
		ceph_buffer_put(m->middle);
		m->middle = NULL;
Sage Weil's avatar
Sage Weil committed
3404
	}
3405

3406 3407
	list_for_each_entry_safe(data, next, &m->data, links) {
		list_del_init(&data->links);
3408 3409
		ceph_msg_data_destroy(data);
	}
3410
	m->data_length = 0;
3411

Sage Weil's avatar
Sage Weil committed
3412 3413 3414
	if (m->pool)
		ceph_msgpool_put(m->pool, m);
	else
3415 3416 3417 3418 3419 3420 3421 3422 3423 3424 3425 3426 3427 3428 3429 3430 3431
		ceph_msg_free(m);
}

struct ceph_msg *ceph_msg_get(struct ceph_msg *msg)
{
	dout("%s %p (was %d)\n", __func__, msg,
	     atomic_read(&msg->kref.refcount));
	kref_get(&msg->kref);
	return msg;
}
EXPORT_SYMBOL(ceph_msg_get);

void ceph_msg_put(struct ceph_msg *msg)
{
	dout("%s %p (was %d)\n", __func__, msg,
	     atomic_read(&msg->kref.refcount));
	kref_put(&msg->kref, ceph_msg_release);
Sage Weil's avatar
Sage Weil committed
3432
}
3433
EXPORT_SYMBOL(ceph_msg_put);
3434 3435 3436

void ceph_msg_dump(struct ceph_msg *msg)
{
3437 3438
	pr_debug("msg_dump %p (front_alloc_len %d length %zd)\n", msg,
		 msg->front_alloc_len, msg->data_length);
3439 3440 3441 3442 3443 3444 3445 3446 3447 3448 3449 3450 3451 3452 3453
	print_hex_dump(KERN_DEBUG, "header: ",
		       DUMP_PREFIX_OFFSET, 16, 1,
		       &msg->hdr, sizeof(msg->hdr), true);
	print_hex_dump(KERN_DEBUG, " front: ",
		       DUMP_PREFIX_OFFSET, 16, 1,
		       msg->front.iov_base, msg->front.iov_len, true);
	if (msg->middle)
		print_hex_dump(KERN_DEBUG, "middle: ",
			       DUMP_PREFIX_OFFSET, 16, 1,
			       msg->middle->vec.iov_base,
			       msg->middle->vec.iov_len, true);
	print_hex_dump(KERN_DEBUG, "footer: ",
		       DUMP_PREFIX_OFFSET, 16, 1,
		       &msg->footer, sizeof(msg->footer), true);
}
3454
EXPORT_SYMBOL(ceph_msg_dump);