projects / GitHub / mt8127 / android_kernel_alcatel_ttab.git / blame
? search:
summary | shortlog | log | commit | commitdiff | tree
blob | blame (incremental) | history | HEAD
ceph: fix parsing of ipv6 addresses
[GitHub/mt8127/android_kernel_alcatel_ttab.git] / fs / ceph / messenger.c
CommitLineData
31b8006e
SW		 1#include "ceph_debug.h"
2
3#include <linux/crc32c.h>
4#include <linux/ctype.h>
5#include <linux/highmem.h>
6#include <linux/inet.h>
7#include <linux/kthread.h>
8#include <linux/net.h>
5a0e3ad6 9#include <linux/slab.h>
31b8006e
SW		 10#include <linux/socket.h>
11#include <linux/string.h>
12#include <net/tcp.h>
13
14#include "super.h"
15#include "messenger.h"
63f2d211 16#include "decode.h"
58bb3b37 17#include "pagelist.h"
31b8006e
SW		 18
19/*
20 * Ceph uses the messenger to exchange ceph_msg messages with other
21 * hosts in the system. The messenger provides ordered and reliable
22 * delivery. We tolerate TCP disconnects by reconnecting (with
23 * exponential backoff) in the case of a fault (disconnection, bad
24 * crc, protocol error). Acks allow sent messages to be discarded by
25 * the sender.
26 */
27
28/* static tag bytes (protocol control messages) */
29static char tag_msg = CEPH_MSGR_TAG_MSG;
30static char tag_ack = CEPH_MSGR_TAG_ACK;
31static char tag_keepalive = CEPH_MSGR_TAG_KEEPALIVE;
32
a6a5349d
SW		 33#ifdef CONFIG_LOCKDEP
34static struct lock_class_key socket_class;
35#endif
36
31b8006e
SW		 37
38static void queue_con(struct ceph_connection *con);
39static void con_work(struct work_struct *);
40static void ceph_fault(struct ceph_connection *con);
41
31b8006e
SW		 42/*
43 * nicely render a sockaddr as a string.
44 */
45#define MAX_ADDR_STR 20
d06dbaf6
SW		 46#define MAX_ADDR_STR_LEN 60
47static char addr_str[MAX_ADDR_STR][MAX_ADDR_STR_LEN];
31b8006e
SW		 48static DEFINE_SPINLOCK(addr_str_lock);
49static int last_addr_str;
50
51const char *pr_addr(const struct sockaddr_storage *ss)
52{
53 int i;
54 char *s;
55 struct sockaddr_in *in4 = (void *)ss;
31b8006e
SW		 56 struct sockaddr_in6 *in6 = (void *)ss;
57
58 spin_lock(&addr_str_lock);
59 i = last_addr_str++;
60 if (last_addr_str == MAX_ADDR_STR)
61 last_addr_str = 0;
62 spin_unlock(&addr_str_lock);
63 s = addr_str[i];
64
65 switch (ss->ss_family) {
66 case AF_INET:
d06dbaf6
SW		 67 snprintf(s, MAX_ADDR_STR_LEN, "%pI4:%u", &in4->sin_addr,
68 (unsigned int)ntohs(in4->sin_port));
31b8006e
SW		 69 break;
70
71 case AF_INET6:
d06dbaf6
SW		 72 snprintf(s, MAX_ADDR_STR_LEN, "[%pI6c]:%u", &in6->sin6_addr,
73 (unsigned int)ntohs(in6->sin6_port));
31b8006e
SW		 74 break;
75
76 default:
77 sprintf(s, "(unknown sockaddr family %d)", (int)ss->ss_family);
78 }
79
80 return s;
81}
82
63f2d211
SW		 83static void encode_my_addr(struct ceph_messenger *msgr)
84{
85 memcpy(&msgr->my_enc_addr, &msgr->inst.addr, sizeof(msgr->my_enc_addr));
86 ceph_encode_addr(&msgr->my_enc_addr);
87}
88
31b8006e
SW		 89/*
90 * work queue for all reading and writing to/from the socket.
91 */
92struct workqueue_struct *ceph_msgr_wq;
93
94int __init ceph_msgr_init(void)
95{
96 ceph_msgr_wq = create_workqueue("ceph-msgr");
97 if (IS_ERR(ceph_msgr_wq)) {
98 int ret = PTR_ERR(ceph_msgr_wq);
99 pr_err("msgr_init failed to create workqueue: %d\n", ret);
100 ceph_msgr_wq = NULL;
101 return ret;
102 }
103 return 0;
104}
105
106void ceph_msgr_exit(void)
107{
108 destroy_workqueue(ceph_msgr_wq);
109}
110
a922d38f
SW		 111void ceph_msgr_flush()
112{
113 flush_workqueue(ceph_msgr_wq);
114}
115
116
31b8006e
SW		 117/*
118 * socket callback functions
119 */
120
121/* data available on socket, or listen socket received a connect */
122static void ceph_data_ready(struct sock *sk, int count_unused)
123{
124 struct ceph_connection *con =
125 (struct ceph_connection *)sk->sk_user_data;
126 if (sk->sk_state != TCP_CLOSE_WAIT) {
127 dout("ceph_data_ready on %p state = %lu, queueing work\n",
128 con, con->state);
129 queue_con(con);
130 }
131}
132
133/* socket has buffer space for writing */
134static void ceph_write_space(struct sock *sk)
135{
136 struct ceph_connection *con =
137 (struct ceph_connection *)sk->sk_user_data;
138
139 /* only queue to workqueue if there is data we want to write. */
140 if (test_bit(WRITE_PENDING, &con->state)) {
141 dout("ceph_write_space %p queueing write work\n", con);
142 queue_con(con);
143 } else {
144 dout("ceph_write_space %p nothing to write\n", con);
145 }
146
147 /* since we have our own write_space, clear the SOCK_NOSPACE flag */
148 clear_bit(SOCK_NOSPACE, &sk->sk_socket->flags);
149}
150
151/* socket's state has changed */
152static void ceph_state_change(struct sock *sk)
153{
154 struct ceph_connection *con =
155 (struct ceph_connection *)sk->sk_user_data;
156
157 dout("ceph_state_change %p state = %lu sk_state = %u\n",
158 con, con->state, sk->sk_state);
159
160 if (test_bit(CLOSED, &con->state))
161 return;
162
163 switch (sk->sk_state) {
164 case TCP_CLOSE:
165 dout("ceph_state_change TCP_CLOSE\n");
166 case TCP_CLOSE_WAIT:
167 dout("ceph_state_change TCP_CLOSE_WAIT\n");
168 if (test_and_set_bit(SOCK_CLOSED, &con->state) == 0) {
169 if (test_bit(CONNECTING, &con->state))
170 con->error_msg = "connection failed";
171 else
172 con->error_msg = "socket closed";
173 queue_con(con);
174 }
175 break;
176 case TCP_ESTABLISHED:
177 dout("ceph_state_change TCP_ESTABLISHED\n");
178 queue_con(con);
179 break;
180 }
181}
182
183/*
184 * set up socket callbacks
185 */
186static void set_sock_callbacks(struct socket *sock,
187 struct ceph_connection *con)
188{
189 struct sock *sk = sock->sk;
190 sk->sk_user_data = (void *)con;
191 sk->sk_data_ready = ceph_data_ready;
192 sk->sk_write_space = ceph_write_space;
193 sk->sk_state_change = ceph_state_change;
194}
195
196
197/*
198 * socket helpers
199 */
200
201/*
202 * initiate connection to a remote socket.
203 */
204static struct socket *ceph_tcp_connect(struct ceph_connection *con)
205{
206 struct sockaddr *paddr = (struct sockaddr *)&con->peer_addr.in_addr;
207 struct socket *sock;
208 int ret;
209
210 BUG_ON(con->sock);
211 ret = sock_create_kern(AF_INET, SOCK_STREAM, IPPROTO_TCP, &sock);
212 if (ret)
213 return ERR_PTR(ret);
214 con->sock = sock;
215 sock->sk->sk_allocation = GFP_NOFS;
216
a6a5349d
SW		 217#ifdef CONFIG_LOCKDEP
218 lockdep_set_class(&sock->sk->sk_lock, &socket_class);
219#endif
220
31b8006e
SW		 221 set_sock_callbacks(sock, con);
222
223 dout("connect %s\n", pr_addr(&con->peer_addr.in_addr));
224
225 ret = sock->ops->connect(sock, paddr, sizeof(*paddr), O_NONBLOCK);
226 if (ret == -EINPROGRESS) {
227 dout("connect %s EINPROGRESS sk_state = %u\n",
228 pr_addr(&con->peer_addr.in_addr),
229 sock->sk->sk_state);
230 ret = 0;
231 }
232 if (ret < 0) {
233 pr_err("connect %s error %d\n",
234 pr_addr(&con->peer_addr.in_addr), ret);
235 sock_release(sock);
236 con->sock = NULL;
237 con->error_msg = "connect error";
238 }
239
240 if (ret < 0)
241 return ERR_PTR(ret);
242 return sock;
243}
244
245static int ceph_tcp_recvmsg(struct socket *sock, void *buf, size_t len)
246{
247 struct kvec iov = {buf, len};
248 struct msghdr msg = { .msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL };
249
250 return kernel_recvmsg(sock, &msg, &iov, 1, len, msg.msg_flags);
251}
252
253/*
254 * write something. @more is true if caller will be sending more data
255 * shortly.
256 */
257static int ceph_tcp_sendmsg(struct socket *sock, struct kvec *iov,
258 size_t kvlen, size_t len, int more)
259{
260 struct msghdr msg = { .msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL };
261
262 if (more)
263 msg.msg_flags |= MSG_MORE;
264 else
265 msg.msg_flags |= MSG_EOR; /* superfluous, but what the hell */
266
267 return kernel_sendmsg(sock, &msg, iov, kvlen, len);
268}
269
270
271/*
272 * Shutdown/close the socket for the given connection.
273 */
274static int con_close_socket(struct ceph_connection *con)
275{
276 int rc;
277
278 dout("con_close_socket on %p sock %p\n", con, con->sock);
279 if (!con->sock)
280 return 0;
281 set_bit(SOCK_CLOSED, &con->state);
282 rc = con->sock->ops->shutdown(con->sock, SHUT_RDWR);
283 sock_release(con->sock);
284 con->sock = NULL;
285 clear_bit(SOCK_CLOSED, &con->state);
286 return rc;
287}
288
289/*
290 * Reset a connection. Discard all incoming and outgoing messages
291 * and clear *_seq state.
292 */
293static void ceph_msg_remove(struct ceph_msg *msg)
294{
295 list_del_init(&msg->list_head);
296 ceph_msg_put(msg);
297}
298static void ceph_msg_remove_list(struct list_head *head)
299{
300 while (!list_empty(head)) {
301 struct ceph_msg *msg = list_first_entry(head, struct ceph_msg,
302 list_head);
303 ceph_msg_remove(msg);
304 }
305}
306
307static void reset_connection(struct ceph_connection *con)
308{
309 /* reset connection, out_queue, msg_ and connect_seq */
310 /* discard existing out_queue and msg_seq */
31b8006e
SW		 311 ceph_msg_remove_list(&con->out_queue);
312 ceph_msg_remove_list(&con->out_sent);
313
cf3e5c40
SW		 314 if (con->in_msg) {
315 ceph_msg_put(con->in_msg);
316 con->in_msg = NULL;
317 }
318
31b8006e
SW		 319 con->connect_seq = 0;
320 con->out_seq = 0;
c86a2930
SW		 321 if (con->out_msg) {
322 ceph_msg_put(con->out_msg);
323 con->out_msg = NULL;
324 }
6f2bc3ff 325 con->out_keepalive_pending = false;
31b8006e 326 con->in_seq = 0;
0e0d5e0c 327 con->in_seq_acked = 0;
31b8006e
SW		 328}
329
330/*
331 * mark a peer down. drop any open connections.
332 */
333void ceph_con_close(struct ceph_connection *con)
334{
335 dout("con_close %p peer %s\n", con, pr_addr(&con->peer_addr.in_addr));
336 set_bit(CLOSED, &con->state); /* in case there's queued work */
337 clear_bit(STANDBY, &con->state); /* avoid connect_seq bump */
1679f876
SW		 338 clear_bit(LOSSYTX, &con->state); /* so we retry next connect */
339 clear_bit(KEEPALIVE_PENDING, &con->state);
340 clear_bit(WRITE_PENDING, &con->state);
ec302645 341 mutex_lock(&con->mutex);
31b8006e 342 reset_connection(con);
6f2bc3ff 343 con->peer_global_seq = 0;
91e45ce3 344 cancel_delayed_work(&con->work);
ec302645 345 mutex_unlock(&con->mutex);
31b8006e
SW		 346 queue_con(con);
347}
348
31b8006e
SW		 349/*
350 * Reopen a closed connection, with a new peer address.
351 */
352void ceph_con_open(struct ceph_connection *con, struct ceph_entity_addr *addr)
353{
354 dout("con_open %p %s\n", con, pr_addr(&addr->in_addr));
355 set_bit(OPENING, &con->state);
356 clear_bit(CLOSED, &con->state);
357 memcpy(&con->peer_addr, addr, sizeof(*addr));
03c677e1 358 con->delay = 0; /* reset backoff memory */
31b8006e
SW		 359 queue_con(con);
360}
361
87b315a5
SW		 362/*
363 * return true if this connection ever successfully opened
364 */
365bool ceph_con_opened(struct ceph_connection *con)
366{
367 return con->connect_seq > 0;
368}
369
31b8006e
SW		 370/*
371 * generic get/put
372 */
373struct ceph_connection *ceph_con_get(struct ceph_connection *con)
374{
375 dout("con_get %p nref = %d -> %d\n", con,
376 atomic_read(&con->nref), atomic_read(&con->nref) + 1);
377 if (atomic_inc_not_zero(&con->nref))
378 return con;
379 return NULL;
380}
381
382void ceph_con_put(struct ceph_connection *con)
383{
384 dout("con_put %p nref = %d -> %d\n", con,
385 atomic_read(&con->nref), atomic_read(&con->nref) - 1);
386 BUG_ON(atomic_read(&con->nref) == 0);
387 if (atomic_dec_and_test(&con->nref)) {
71ececda 388 BUG_ON(con->sock);
31b8006e
SW		 389 kfree(con);
390 }
391}
392
393/*
394 * initialize a new connection.
395 */
396void ceph_con_init(struct ceph_messenger *msgr, struct ceph_connection *con)
397{
398 dout("con_init %p\n", con);
399 memset(con, 0, sizeof(*con));
400 atomic_set(&con->nref, 1);
401 con->msgr = msgr;
ec302645 402 mutex_init(&con->mutex);
31b8006e
SW		 403 INIT_LIST_HEAD(&con->out_queue);
404 INIT_LIST_HEAD(&con->out_sent);
405 INIT_DELAYED_WORK(&con->work, con_work);
406}
407
408
409/*
410 * We maintain a global counter to order connection attempts. Get
411 * a unique seq greater than @gt.
412 */
413static u32 get_global_seq(struct ceph_messenger *msgr, u32 gt)
414{
415 u32 ret;
416
417 spin_lock(&msgr->global_seq_lock);
418 if (msgr->global_seq < gt)
419 msgr->global_seq = gt;
420 ret = ++msgr->global_seq;
421 spin_unlock(&msgr->global_seq_lock);
422 return ret;
423}
424
425
426/*
427 * Prepare footer for currently outgoing message, and finish things
428 * off. Assumes out_kvec* are already valid.. we just add on to the end.
429 */
430static void prepare_write_message_footer(struct ceph_connection *con, int v)
431{
432 struct ceph_msg *m = con->out_msg;
433
434 dout("prepare_write_message_footer %p\n", con);
435 con->out_kvec_is_msg = true;
436 con->out_kvec[v].iov_base = &m->footer;
437 con->out_kvec[v].iov_len = sizeof(m->footer);
438 con->out_kvec_bytes += sizeof(m->footer);
439 con->out_kvec_left++;
440 con->out_more = m->more_to_follow;
c86a2930 441 con->out_msg_done = true;
31b8006e
SW		 442}
443
444/*
445 * Prepare headers for the next outgoing message.
446 */
447static void prepare_write_message(struct ceph_connection *con)
448{
449 struct ceph_msg *m;
450 int v = 0;
451
452 con->out_kvec_bytes = 0;
453 con->out_kvec_is_msg = true;
c86a2930 454 con->out_msg_done = false;
31b8006e
SW		 455
456 /* Sneak an ack in there first? If we can get it into the same
457 * TCP packet that's a good thing. */
458 if (con->in_seq > con->in_seq_acked) {
459 con->in_seq_acked = con->in_seq;
460 con->out_kvec[v].iov_base = &tag_ack;
461 con->out_kvec[v++].iov_len = 1;
462 con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
463 con->out_kvec[v].iov_base = &con->out_temp_ack;
464 con->out_kvec[v++].iov_len = sizeof(con->out_temp_ack);
465 con->out_kvec_bytes = 1 + sizeof(con->out_temp_ack);
466 }
467
31b8006e
SW		 468 m = list_first_entry(&con->out_queue,
469 struct ceph_msg, list_head);
c86a2930 470 con->out_msg = m;
b3d1dbbd 471 if (test_bit(LOSSYTX, &con->state)) {
6c5d1a49
SW		 472 list_del_init(&m->list_head);
473 } else {
b3d1dbbd
SW		 474 /* put message on sent list */
475 ceph_msg_get(m);
476 list_move_tail(&m->list_head, &con->out_sent);
b3d1dbbd 477 }
31b8006e 478
e84346b7
SW		 479 /*
480 * only assign outgoing seq # if we haven't sent this message
481 * yet. if it is requeued, resend with it's original seq.
482 */
483 if (m->needs_out_seq) {
484 m->hdr.seq = cpu_to_le64(++con->out_seq);
485 m->needs_out_seq = false;
486 }
31b8006e
SW		 487
488 dout("prepare_write_message %p seq %lld type %d len %d+%d+%d %d pgs\n",
489 m, con->out_seq, le16_to_cpu(m->hdr.type),
490 le32_to_cpu(m->hdr.front_len), le32_to_cpu(m->hdr.middle_len),
491 le32_to_cpu(m->hdr.data_len),
492 m->nr_pages);
493 BUG_ON(le32_to_cpu(m->hdr.front_len) != m->front.iov_len);
494
495 /* tag + hdr + front + middle */
496 con->out_kvec[v].iov_base = &tag_msg;
497 con->out_kvec[v++].iov_len = 1;
498 con->out_kvec[v].iov_base = &m->hdr;
499 con->out_kvec[v++].iov_len = sizeof(m->hdr);
500 con->out_kvec[v++] = m->front;
501 if (m->middle)
502 con->out_kvec[v++] = m->middle->vec;
503 con->out_kvec_left = v;
504 con->out_kvec_bytes += 1 + sizeof(m->hdr) + m->front.iov_len +
505 (m->middle ? m->middle->vec.iov_len : 0);
506 con->out_kvec_cur = con->out_kvec;
507
508 /* fill in crc (except data pages), footer */
509 con->out_msg->hdr.crc =
510 cpu_to_le32(crc32c(0, (void *)&m->hdr,
511 sizeof(m->hdr) - sizeof(m->hdr.crc)));
512 con->out_msg->footer.flags = CEPH_MSG_FOOTER_COMPLETE;
513 con->out_msg->footer.front_crc =
514 cpu_to_le32(crc32c(0, m->front.iov_base, m->front.iov_len));
515 if (m->middle)
516 con->out_msg->footer.middle_crc =
517 cpu_to_le32(crc32c(0, m->middle->vec.iov_base,
518 m->middle->vec.iov_len));
519 else
520 con->out_msg->footer.middle_crc = 0;
521 con->out_msg->footer.data_crc = 0;
522 dout("prepare_write_message front_crc %u data_crc %u\n",
523 le32_to_cpu(con->out_msg->footer.front_crc),
524 le32_to_cpu(con->out_msg->footer.middle_crc));
525
526 /* is there a data payload? */
527 if (le32_to_cpu(m->hdr.data_len) > 0) {
528 /* initialize page iterator */
529 con->out_msg_pos.page = 0;
530 con->out_msg_pos.page_pos =
531 le16_to_cpu(m->hdr.data_off) & ~PAGE_MASK;
532 con->out_msg_pos.data_pos = 0;
533 con->out_msg_pos.did_page_crc = 0;
534 con->out_more = 1; /* data + footer will follow */
535 } else {
536 /* no, queue up footer too and be done */
537 prepare_write_message_footer(con, v);
538 }
539
540 set_bit(WRITE_PENDING, &con->state);
541}
542
543/*
544 * Prepare an ack.
545 */
546static void prepare_write_ack(struct ceph_connection *con)
547{
548 dout("prepare_write_ack %p %llu -> %llu\n", con,
549 con->in_seq_acked, con->in_seq);
550 con->in_seq_acked = con->in_seq;
551
552 con->out_kvec[0].iov_base = &tag_ack;
553 con->out_kvec[0].iov_len = 1;
554 con->out_temp_ack = cpu_to_le64(con->in_seq_acked);
555 con->out_kvec[1].iov_base = &con->out_temp_ack;
556 con->out_kvec[1].iov_len = sizeof(con->out_temp_ack);
557 con->out_kvec_left = 2;
558 con->out_kvec_bytes = 1 + sizeof(con->out_temp_ack);
559 con->out_kvec_cur = con->out_kvec;
560 con->out_more = 1; /* more will follow.. eventually.. */
561 set_bit(WRITE_PENDING, &con->state);
562}
563
564/*
565 * Prepare to write keepalive byte.
566 */
567static void prepare_write_keepalive(struct ceph_connection *con)
568{
569 dout("prepare_write_keepalive %p\n", con);
570 con->out_kvec[0].iov_base = &tag_keepalive;
571 con->out_kvec[0].iov_len = 1;
572 con->out_kvec_left = 1;
573 con->out_kvec_bytes = 1;
574 con->out_kvec_cur = con->out_kvec;
575 set_bit(WRITE_PENDING, &con->state);
576}
577
578/*
579 * Connection negotiation.
580 */
581
4e7a5dcd
SW		 582static void prepare_connect_authorizer(struct ceph_connection *con)
583{
584 void *auth_buf;
585 int auth_len = 0;
586 int auth_protocol = 0;
587
ec302645 588 mutex_unlock(&con->mutex);
4e7a5dcd
SW		 589 if (con->ops->get_authorizer)
590 con->ops->get_authorizer(con, &auth_buf, &auth_len,
591 &auth_protocol, &con->auth_reply_buf,
592 &con->auth_reply_buf_len,
593 con->auth_retry);
ec302645 594 mutex_lock(&con->mutex);
4e7a5dcd
SW		 595
596 con->out_connect.authorizer_protocol = cpu_to_le32(auth_protocol);
597 con->out_connect.authorizer_len = cpu_to_le32(auth_len);
598
599 con->out_kvec[con->out_kvec_left].iov_base = auth_buf;
600 con->out_kvec[con->out_kvec_left].iov_len = auth_len;
601 con->out_kvec_left++;
602 con->out_kvec_bytes += auth_len;
603}
604
31b8006e
SW		 605/*
606 * We connected to a peer and are saying hello.
607 */
eed0ef2c
SW		 608static void prepare_write_banner(struct ceph_messenger *msgr,
609 struct ceph_connection *con)
31b8006e
SW		 610{
611 int len = strlen(CEPH_BANNER);
eed0ef2c
SW		 612
613 con->out_kvec[0].iov_base = CEPH_BANNER;
614 con->out_kvec[0].iov_len = len;
615 con->out_kvec[1].iov_base = &msgr->my_enc_addr;
616 con->out_kvec[1].iov_len = sizeof(msgr->my_enc_addr);
617 con->out_kvec_left = 2;
618 con->out_kvec_bytes = len + sizeof(msgr->my_enc_addr);
619 con->out_kvec_cur = con->out_kvec;
620 con->out_more = 0;
621 set_bit(WRITE_PENDING, &con->state);
622}
623
624static void prepare_write_connect(struct ceph_messenger *msgr,
625 struct ceph_connection *con,
626 int after_banner)
627{
31b8006e
SW		 628 unsigned global_seq = get_global_seq(con->msgr, 0);
629 int proto;
630
631 switch (con->peer_name.type) {
632 case CEPH_ENTITY_TYPE_MON:
633 proto = CEPH_MONC_PROTOCOL;
634 break;
635 case CEPH_ENTITY_TYPE_OSD:
636 proto = CEPH_OSDC_PROTOCOL;
637 break;
638 case CEPH_ENTITY_TYPE_MDS:
639 proto = CEPH_MDSC_PROTOCOL;
640 break;
641 default:
642 BUG();
643 }
644
645 dout("prepare_write_connect %p cseq=%d gseq=%d proto=%d\n", con,
646 con->connect_seq, global_seq, proto);
4e7a5dcd 647
0cf5537b 648 con->out_connect.features = cpu_to_le64(CEPH_FEATURE_SUPPORTED_CLIENT);
31b8006e
SW		 649 con->out_connect.host_type = cpu_to_le32(CEPH_ENTITY_TYPE_CLIENT);
650 con->out_connect.connect_seq = cpu_to_le32(con->connect_seq);
651 con->out_connect.global_seq = cpu_to_le32(global_seq);
652 con->out_connect.protocol_version = cpu_to_le32(proto);
653 con->out_connect.flags = 0;
31b8006e 654
eed0ef2c
SW		 655 if (!after_banner) {
656 con->out_kvec_left = 0;
657 con->out_kvec_bytes = 0;
658 }
659 con->out_kvec[con->out_kvec_left].iov_base = &con->out_connect;
660 con->out_kvec[con->out_kvec_left].iov_len = sizeof(con->out_connect);
661 con->out_kvec_left++;
662 con->out_kvec_bytes += sizeof(con->out_connect);
31b8006e
SW		 663 con->out_kvec_cur = con->out_kvec;
664 con->out_more = 0;
665 set_bit(WRITE_PENDING, &con->state);
4e7a5dcd
SW		 666
667 prepare_connect_authorizer(con);
31b8006e
SW		 668}
669
670
671/*
672 * write as much of pending kvecs to the socket as we can.
673 * 1 -> done
674 * 0 -> socket full, but more to do
675 * <0 -> error
676 */
677static int write_partial_kvec(struct ceph_connection *con)
678{
679 int ret;
680
681 dout("write_partial_kvec %p %d left\n", con, con->out_kvec_bytes);
682 while (con->out_kvec_bytes > 0) {
683 ret = ceph_tcp_sendmsg(con->sock, con->out_kvec_cur,
684 con->out_kvec_left, con->out_kvec_bytes,
685 con->out_more);
686 if (ret <= 0)
687 goto out;
688 con->out_kvec_bytes -= ret;
689 if (con->out_kvec_bytes == 0)
690 break; /* done */
691 while (ret > 0) {
692 if (ret >= con->out_kvec_cur->iov_len) {
693 ret -= con->out_kvec_cur->iov_len;
694 con->out_kvec_cur++;
695 con->out_kvec_left--;
696 } else {
697 con->out_kvec_cur->iov_len -= ret;
698 con->out_kvec_cur->iov_base += ret;
699 ret = 0;
700 break;
701 }
702 }
703 }
704 con->out_kvec_left = 0;
705 con->out_kvec_is_msg = false;
706 ret = 1;
707out:
708 dout("write_partial_kvec %p %d left in %d kvecs ret = %d\n", con,
709 con->out_kvec_bytes, con->out_kvec_left, ret);
710 return ret; /* done! */
711}
712
713/*
714 * Write as much message data payload as we can. If we finish, queue
715 * up the footer.
716 * 1 -> done, footer is now queued in out_kvec[].
717 * 0 -> socket full, but more to do
718 * <0 -> error
719 */
720static int write_partial_msg_pages(struct ceph_connection *con)
721{
722 struct ceph_msg *msg = con->out_msg;
723 unsigned data_len = le32_to_cpu(msg->hdr.data_len);
724 size_t len;
725 int crc = con->msgr->nocrc;
726 int ret;
727
728 dout("write_partial_msg_pages %p msg %p page %d/%d offset %d\n",
729 con, con->out_msg, con->out_msg_pos.page, con->out_msg->nr_pages,
730 con->out_msg_pos.page_pos);
731
732 while (con->out_msg_pos.page < con->out_msg->nr_pages) {
733 struct page *page = NULL;
734 void *kaddr = NULL;
735
736 /*
737 * if we are calculating the data crc (the default), we need
738 * to map the page. if our pages[] has been revoked, use the
739 * zero page.
740 */
741 if (msg->pages) {
742 page = msg->pages[con->out_msg_pos.page];
743 if (crc)
744 kaddr = kmap(page);
58bb3b37
SW		 745 } else if (msg->pagelist) {
746 page = list_first_entry(&msg->pagelist->head,
747 struct page, lru);
748 if (crc)
749 kaddr = kmap(page);
31b8006e
SW		 750 } else {
751 page = con->msgr->zero_page;
752 if (crc)
753 kaddr = page_address(con->msgr->zero_page);
754 }
755 len = min((int)(PAGE_SIZE - con->out_msg_pos.page_pos),
756 (int)(data_len - con->out_msg_pos.data_pos));
757 if (crc && !con->out_msg_pos.did_page_crc) {
758 void *base = kaddr + con->out_msg_pos.page_pos;
759 u32 tmpcrc = le32_to_cpu(con->out_msg->footer.data_crc);
760
761 BUG_ON(kaddr == NULL);
762 con->out_msg->footer.data_crc =
763 cpu_to_le32(crc32c(tmpcrc, base, len));
764 con->out_msg_pos.did_page_crc = 1;
765 }
766
767 ret = kernel_sendpage(con->sock, page,
768 con->out_msg_pos.page_pos, len,
769 MSG_DONTWAIT | MSG_NOSIGNAL |
770 MSG_MORE);
771
58bb3b37 772 if (crc && (msg->pages || msg->pagelist))
31b8006e
SW		 773 kunmap(page);
774
775 if (ret <= 0)
776 goto out;
777
778 con->out_msg_pos.data_pos += ret;
779 con->out_msg_pos.page_pos += ret;
780 if (ret == len) {
781 con->out_msg_pos.page_pos = 0;
782 con->out_msg_pos.page++;
783 con->out_msg_pos.did_page_crc = 0;
58bb3b37
SW		 784 if (msg->pagelist)
785 list_move_tail(&page->lru,
786 &msg->pagelist->head);
31b8006e
SW		 787 }
788 }
789
790 dout("write_partial_msg_pages %p msg %p done\n", con, msg);
791
792 /* prepare and queue up footer, too */
793 if (!crc)
794 con->out_msg->footer.flags |= CEPH_MSG_FOOTER_NOCRC;
795 con->out_kvec_bytes = 0;
796 con->out_kvec_left = 0;
797 con->out_kvec_cur = con->out_kvec;
798 prepare_write_message_footer(con, 0);
799 ret = 1;
800out:
801 return ret;
802}
803
804/*
805 * write some zeros
806 */
807static int write_partial_skip(struct ceph_connection *con)
808{
809 int ret;
810
811 while (con->out_skip > 0) {
812 struct kvec iov = {
813 .iov_base = page_address(con->msgr->zero_page),
814 .iov_len = min(con->out_skip, (int)PAGE_CACHE_SIZE)
815 };
816
817 ret = ceph_tcp_sendmsg(con->sock, &iov, 1, iov.iov_len, 1);
818 if (ret <= 0)
819 goto out;
820 con->out_skip -= ret;
821 }
822 ret = 1;
823out:
824 return ret;
825}
826
827/*
828 * Prepare to read connection handshake, or an ack.
829 */
eed0ef2c
SW		 830static void prepare_read_banner(struct ceph_connection *con)
831{
832 dout("prepare_read_banner %p\n", con);
833 con->in_base_pos = 0;
834}
835
31b8006e
SW		 836static void prepare_read_connect(struct ceph_connection *con)
837{
838 dout("prepare_read_connect %p\n", con);
839 con->in_base_pos = 0;
840}
841
842static void prepare_read_ack(struct ceph_connection *con)
843{
844 dout("prepare_read_ack %p\n", con);
845 con->in_base_pos = 0;
846}
847
848static void prepare_read_tag(struct ceph_connection *con)
849{
850 dout("prepare_read_tag %p\n", con);
851 con->in_base_pos = 0;
852 con->in_tag = CEPH_MSGR_TAG_READY;
853}
854
855/*
856 * Prepare to read a message.
857 */
858static int prepare_read_message(struct ceph_connection *con)
859{
860 dout("prepare_read_message %p\n", con);
861 BUG_ON(con->in_msg != NULL);
862 con->in_base_pos = 0;
863 con->in_front_crc = con->in_middle_crc = con->in_data_crc = 0;
864 return 0;
865}
866
867
868static int read_partial(struct ceph_connection *con,
869 int *to, int size, void *object)
870{
871 *to += size;
872 while (con->in_base_pos < *to) {
873 int left = *to - con->in_base_pos;
874 int have = size - left;
875 int ret = ceph_tcp_recvmsg(con->sock, object + have, left);
876 if (ret <= 0)
877 return ret;
878 con->in_base_pos += ret;
879 }
880 return 1;
881}
882
883
884/*
885 * Read all or part of the connect-side handshake on a new connection
886 */
eed0ef2c 887static int read_partial_banner(struct ceph_connection *con)
31b8006e
SW		 888{
889 int ret, to = 0;
890
eed0ef2c 891 dout("read_partial_banner %p at %d\n", con, con->in_base_pos);
31b8006e
SW		 892
893 /* peer's banner */
894 ret = read_partial(con, &to, strlen(CEPH_BANNER), con->in_banner);
895 if (ret <= 0)
896 goto out;
897 ret = read_partial(con, &to, sizeof(con->actual_peer_addr),
898 &con->actual_peer_addr);
899 if (ret <= 0)
900 goto out;
901 ret = read_partial(con, &to, sizeof(con->peer_addr_for_me),
902 &con->peer_addr_for_me);
903 if (ret <= 0)
904 goto out;
eed0ef2c
SW		 905out:
906 return ret;
907}
908
909static int read_partial_connect(struct ceph_connection *con)
910{
911 int ret, to = 0;
912
913 dout("read_partial_connect %p at %d\n", con, con->in_base_pos);
914
31b8006e
SW		 915 ret = read_partial(con, &to, sizeof(con->in_reply), &con->in_reply);
916 if (ret <= 0)
917 goto out;
4e7a5dcd
SW		 918 ret = read_partial(con, &to, le32_to_cpu(con->in_reply.authorizer_len),
919 con->auth_reply_buf);
920 if (ret <= 0)
921 goto out;
31b8006e 922
4e7a5dcd
SW		 923 dout("read_partial_connect %p tag %d, con_seq = %u, g_seq = %u\n",
924 con, (int)con->in_reply.tag,
925 le32_to_cpu(con->in_reply.connect_seq),
31b8006e
SW		 926 le32_to_cpu(con->in_reply.global_seq));
927out:
928 return ret;
eed0ef2c 929
31b8006e
SW		 930}
931
932/*
933 * Verify the hello banner looks okay.
934 */
935static int verify_hello(struct ceph_connection *con)
936{
937 if (memcmp(con->in_banner, CEPH_BANNER, strlen(CEPH_BANNER))) {
13e38c8a 938 pr_err("connect to %s got bad banner\n",
31b8006e
SW		 939 pr_addr(&con->peer_addr.in_addr));
940 con->error_msg = "protocol error, bad banner";
941 return -1;
942 }
943 return 0;
944}
945
946static bool addr_is_blank(struct sockaddr_storage *ss)
947{
948 switch (ss->ss_family) {
949 case AF_INET:
950 return ((struct sockaddr_in *)ss)->sin_addr.s_addr == 0;
951 case AF_INET6:
952 return
953 ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[0] == 0 &&
954 ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[1] == 0 &&
955 ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[2] == 0 &&
956 ((struct sockaddr_in6 *)ss)->sin6_addr.s6_addr32[3] == 0;
957 }
958 return false;
959}
960
961static int addr_port(struct sockaddr_storage *ss)
962{
963 switch (ss->ss_family) {
964 case AF_INET:
f28bcfbe 965 return ntohs(((struct sockaddr_in *)ss)->sin_port);
31b8006e 966 case AF_INET6:
f28bcfbe 967 return ntohs(((struct sockaddr_in6 *)ss)->sin6_port);
31b8006e
SW		 968 }
969 return 0;
970}
971
972static void addr_set_port(struct sockaddr_storage *ss, int p)
973{
974 switch (ss->ss_family) {
975 case AF_INET:
976 ((struct sockaddr_in *)ss)->sin_port = htons(p);
977 case AF_INET6:
978 ((struct sockaddr_in6 *)ss)->sin6_port = htons(p);
979 }
980}
981
982/*
983 * Parse an ip[:port] list into an addr array. Use the default
984 * monitor port if a port isn't specified.
985 */
986int ceph_parse_ips(const char *c, const char *end,
987 struct ceph_entity_addr *addr,
988 int max_count, int *count)
989{
990 int i;
991 const char *p = c;
992
993 dout("parse_ips on '%.*s'\n", (int)(end-c), c);
994 for (i = 0; i < max_count; i++) {
995 const char *ipend;
996 struct sockaddr_storage *ss = &addr[i].in_addr;
997 struct sockaddr_in *in4 = (void *)ss;
998 struct sockaddr_in6 *in6 = (void *)ss;
999 int port;
39139f64
SW		 1000 char delim = ',';
1001
1002 if (*p == '[') {
1003 delim = ']';
1004 p++;
1005 }
31b8006e
SW		 1006
1007 memset(ss, 0, sizeof(*ss));
1008 if (in4_pton(p, end - p, (u8 *)&in4->sin_addr.s_addr,
39139f64 1009 delim, &ipend))
31b8006e 1010 ss->ss_family = AF_INET;
39139f64
SW		 1011 else if (in6_pton(p, end - p, (u8 *)&in6->sin6_addr.s6_addr,
1012 delim, &ipend))
31b8006e 1013 ss->ss_family = AF_INET6;
39139f64 1014 else
31b8006e 1015 goto bad;
31b8006e
SW		 1016 p = ipend;
1017
39139f64
SW		 1018 if (delim == ']') {
1019 if (*p != ']') {
1020 dout("missing matching ']'\n");
1021 goto bad;
1022 }
1023 p++;
1024 }
1025
31b8006e
SW		 1026 /* port? */
1027 if (p < end && *p == ':') {
1028 port = 0;
1029 p++;
1030 while (p < end && *p >= '0' && *p <= '9') {
1031 port = (port * 10) + (*p - '0');
1032 p++;
1033 }
1034 if (port > 65535 || port == 0)
1035 goto bad;
1036 } else {
1037 port = CEPH_MON_PORT;
1038 }
1039
1040 addr_set_port(ss, port);
1041
1042 dout("parse_ips got %s\n", pr_addr(ss));
1043
1044 if (p == end)
1045 break;
1046 if (*p != ',')
1047 goto bad;
1048 p++;
1049 }
1050
1051 if (p != end)
1052 goto bad;
1053
1054 if (count)
1055 *count = i + 1;
1056 return 0;
1057
1058bad:
39139f64 1059 pr_err("parse_ips bad ip '%.*s'\n", (int)(end - c), c);
31b8006e
SW		 1060 return -EINVAL;
1061}
1062
eed0ef2c 1063static int process_banner(struct ceph_connection *con)
31b8006e 1064{
eed0ef2c 1065 dout("process_banner on %p\n", con);
31b8006e
SW		 1066
1067 if (verify_hello(con) < 0)
1068 return -1;
1069
63f2d211
SW		 1070 ceph_decode_addr(&con->actual_peer_addr);
1071 ceph_decode_addr(&con->peer_addr_for_me);
1072
31b8006e
SW		 1073 /*
1074 * Make sure the other end is who we wanted. note that the other
1075 * end may not yet know their ip address, so if it's 0.0.0.0, give
1076 * them the benefit of the doubt.
1077 */
103e2d3a
SW		 1078 if (memcmp(&con->peer_addr, &con->actual_peer_addr,
1079 sizeof(con->peer_addr)) != 0 &&
31b8006e
SW		 1080 !(addr_is_blank(&con->actual_peer_addr.in_addr) &&
1081 con->actual_peer_addr.nonce == con->peer_addr.nonce)) {
103e2d3a
SW		 1082 pr_warning("wrong peer, want %s/%lld, got %s/%lld\n",
1083 pr_addr(&con->peer_addr.in_addr),
1084 le64_to_cpu(con->peer_addr.nonce),
1085 pr_addr(&con->actual_peer_addr.in_addr),
1086 le64_to_cpu(con->actual_peer_addr.nonce));
58bb3b37 1087 con->error_msg = "wrong peer at address";
31b8006e
SW		 1088 return -1;
1089 }
1090
1091 /*
1092 * did we learn our address?
1093 */
1094 if (addr_is_blank(&con->msgr->inst.addr.in_addr)) {
1095 int port = addr_port(&con->msgr->inst.addr.in_addr);
1096
1097 memcpy(&con->msgr->inst.addr.in_addr,
1098 &con->peer_addr_for_me.in_addr,
1099 sizeof(con->peer_addr_for_me.in_addr));
1100 addr_set_port(&con->msgr->inst.addr.in_addr, port);
63f2d211 1101 encode_my_addr(con->msgr);
eed0ef2c 1102 dout("process_banner learned my addr is %s\n",
31b8006e
SW		 1103 pr_addr(&con->msgr->inst.addr.in_addr));
1104 }
1105
eed0ef2c
SW		 1106 set_bit(NEGOTIATING, &con->state);
1107 prepare_read_connect(con);
1108 return 0;
1109}
1110
04a419f9
SW		 1111static void fail_protocol(struct ceph_connection *con)
1112{
1113 reset_connection(con);
1114 set_bit(CLOSED, &con->state); /* in case there's queued work */
1115
1116 mutex_unlock(&con->mutex);
1117 if (con->ops->bad_proto)
1118 con->ops->bad_proto(con);
1119 mutex_lock(&con->mutex);
1120}
1121
eed0ef2c
SW		 1122static int process_connect(struct ceph_connection *con)
1123{
dbad185d
SW		 1124 u64 sup_feat = CEPH_FEATURE_SUPPORTED_CLIENT;
1125 u64 req_feat = CEPH_FEATURE_REQUIRED_CLIENT;
04a419f9
SW		 1126 u64 server_feat = le64_to_cpu(con->in_reply.features);
1127
eed0ef2c
SW		 1128 dout("process_connect on %p tag %d\n", con, (int)con->in_tag);
1129
31b8006e 1130 switch (con->in_reply.tag) {
04a419f9
SW		 1131 case CEPH_MSGR_TAG_FEATURES:
1132 pr_err("%s%lld %s feature set mismatch,"
1133 " my %llx < server's %llx, missing %llx\n",
1134 ENTITY_NAME(con->peer_name),
1135 pr_addr(&con->peer_addr.in_addr),
1136 sup_feat, server_feat, server_feat & ~sup_feat);
1137 con->error_msg = "missing required protocol features";
1138 fail_protocol(con);
1139 return -1;
1140
31b8006e 1141 case CEPH_MSGR_TAG_BADPROTOVER:
31b8006e
SW		 1142 pr_err("%s%lld %s protocol version mismatch,"
1143 " my %d != server's %d\n",
1144 ENTITY_NAME(con->peer_name),
1145 pr_addr(&con->peer_addr.in_addr),
1146 le32_to_cpu(con->out_connect.protocol_version),
1147 le32_to_cpu(con->in_reply.protocol_version));
1148 con->error_msg = "protocol version mismatch";
04a419f9 1149 fail_protocol(con);
31b8006e
SW		 1150 return -1;
1151
4e7a5dcd
SW		 1152 case CEPH_MSGR_TAG_BADAUTHORIZER:
1153 con->auth_retry++;
1154 dout("process_connect %p got BADAUTHORIZER attempt %d\n", con,
1155 con->auth_retry);
1156 if (con->auth_retry == 2) {
1157 con->error_msg = "connect authorization failure";
1158 reset_connection(con);
1159 set_bit(CLOSED, &con->state);
1160 return -1;
1161 }
1162 con->auth_retry = 1;
1163 prepare_write_connect(con->msgr, con, 0);
63733a0f 1164 prepare_read_connect(con);
4e7a5dcd 1165 break;
31b8006e
SW		 1166
1167 case CEPH_MSGR_TAG_RESETSESSION:
1168 /*
1169 * If we connected with a large connect_seq but the peer
1170 * has no record of a session with us (no connection, or
1171 * connect_seq == 0), they will send RESETSESION to indicate
1172 * that they must have reset their session, and may have
1173 * dropped messages.
1174 */
1175 dout("process_connect got RESET peer seq %u\n",
1176 le32_to_cpu(con->in_connect.connect_seq));
1177 pr_err("%s%lld %s connection reset\n",
1178 ENTITY_NAME(con->peer_name),
1179 pr_addr(&con->peer_addr.in_addr));
1180 reset_connection(con);
eed0ef2c 1181 prepare_write_connect(con->msgr, con, 0);
31b8006e
SW		 1182 prepare_read_connect(con);
1183
1184 /* Tell ceph about it. */
ec302645 1185 mutex_unlock(&con->mutex);
31b8006e
SW		 1186 pr_info("reset on %s%lld\n", ENTITY_NAME(con->peer_name));
1187 if (con->ops->peer_reset)
1188 con->ops->peer_reset(con);
ec302645 1189 mutex_lock(&con->mutex);
31b8006e
SW		 1190 break;
1191
1192 case CEPH_MSGR_TAG_RETRY_SESSION:
1193 /*
1194 * If we sent a smaller connect_seq than the peer has, try
1195 * again with a larger value.
1196 */
1197 dout("process_connect got RETRY my seq = %u, peer_seq = %u\n",
1198 le32_to_cpu(con->out_connect.connect_seq),
1199 le32_to_cpu(con->in_connect.connect_seq));
1200 con->connect_seq = le32_to_cpu(con->in_connect.connect_seq);
eed0ef2c 1201 prepare_write_connect(con->msgr, con, 0);
31b8006e
SW		 1202 prepare_read_connect(con);
1203 break;
1204
1205 case CEPH_MSGR_TAG_RETRY_GLOBAL:
1206 /*
1207 * If we sent a smaller global_seq than the peer has, try
1208 * again with a larger value.
1209 */
eed0ef2c 1210 dout("process_connect got RETRY_GLOBAL my %u peer_gseq %u\n",
31b8006e
SW		 1211 con->peer_global_seq,
1212 le32_to_cpu(con->in_connect.global_seq));
1213 get_global_seq(con->msgr,
1214 le32_to_cpu(con->in_connect.global_seq));
eed0ef2c 1215 prepare_write_connect(con->msgr, con, 0);
31b8006e
SW		 1216 prepare_read_connect(con);
1217 break;
1218
1219 case CEPH_MSGR_TAG_READY:
04a419f9
SW		 1220 if (req_feat & ~server_feat) {
1221 pr_err("%s%lld %s protocol feature mismatch,"
1222 " my required %llx > server's %llx, need %llx\n",
1223 ENTITY_NAME(con->peer_name),
1224 pr_addr(&con->peer_addr.in_addr),
1225 req_feat, server_feat, req_feat & ~server_feat);
1226 con->error_msg = "missing required protocol features";
1227 fail_protocol(con);
1228 return -1;
1229 }
31b8006e 1230 clear_bit(CONNECTING, &con->state);
31b8006e
SW		 1231 con->peer_global_seq = le32_to_cpu(con->in_reply.global_seq);
1232 con->connect_seq++;
aba558e2 1233 con->peer_features = server_feat;
31b8006e
SW		 1234 dout("process_connect got READY gseq %d cseq %d (%d)\n",
1235 con->peer_global_seq,
1236 le32_to_cpu(con->in_reply.connect_seq),
1237 con->connect_seq);
1238 WARN_ON(con->connect_seq !=
1239 le32_to_cpu(con->in_reply.connect_seq));
92ac41d0
SW		 1240
1241 if (con->in_reply.flags & CEPH_MSG_CONNECT_LOSSY)
1242 set_bit(LOSSYTX, &con->state);
1243
31b8006e
SW		 1244 prepare_read_tag(con);
1245 break;
1246
1247 case CEPH_MSGR_TAG_WAIT:
1248 /*
1249 * If there is a connection race (we are opening
1250 * connections to each other), one of us may just have
1251 * to WAIT. This shouldn't happen if we are the
1252 * client.
1253 */
1254 pr_err("process_connect peer connecting WAIT\n");
1255
1256 default:
1257 pr_err("connect protocol error, will retry\n");
1258 con->error_msg = "protocol error, garbage tag during connect";
1259 return -1;
1260 }
1261 return 0;
1262}
1263
1264
1265/*
1266 * read (part of) an ack
1267 */
1268static int read_partial_ack(struct ceph_connection *con)
1269{
1270 int to = 0;
1271
1272 return read_partial(con, &to, sizeof(con->in_temp_ack),
1273 &con->in_temp_ack);
1274}
1275
1276
1277/*
1278 * We can finally discard anything that's been acked.
1279 */
1280static void process_ack(struct ceph_connection *con)
1281{
1282 struct ceph_msg *m;
1283 u64 ack = le64_to_cpu(con->in_temp_ack);
1284 u64 seq;
1285
31b8006e
SW		 1286 while (!list_empty(&con->out_sent)) {
1287 m = list_first_entry(&con->out_sent, struct ceph_msg,
1288 list_head);
1289 seq = le64_to_cpu(m->hdr.seq);
1290 if (seq > ack)
1291 break;
1292 dout("got ack for seq %llu type %d at %p\n", seq,
1293 le16_to_cpu(m->hdr.type), m);
1294 ceph_msg_remove(m);
1295 }
31b8006e
SW		 1296 prepare_read_tag(con);
1297}
1298
1299
1300
1301
2450418c
YS		 1302static int read_partial_message_section(struct ceph_connection *con,
1303 struct kvec *section, unsigned int sec_len,
1304 u32 *crc)
1305{
1306 int left;
1307 int ret;
1308
1309 BUG_ON(!section);
1310
1311 while (section->iov_len < sec_len) {
1312 BUG_ON(section->iov_base == NULL);
1313 left = sec_len - section->iov_len;
1314 ret = ceph_tcp_recvmsg(con->sock, (char *)section->iov_base +
1315 section->iov_len, left);
1316 if (ret <= 0)
1317 return ret;
1318 section->iov_len += ret;
1319 if (section->iov_len == sec_len)
1320 *crc = crc32c(0, section->iov_base,
1321 section->iov_len);
1322 }
31b8006e 1323
2450418c
YS		 1324 return 1;
1325}
31b8006e 1326
2450418c
YS		 1327static struct ceph_msg *ceph_alloc_msg(struct ceph_connection *con,
1328 struct ceph_msg_header *hdr,
1329 int *skip);
31b8006e
SW		 1330/*
1331 * read (part of) a message.
1332 */
1333static int read_partial_message(struct ceph_connection *con)
1334{
1335 struct ceph_msg *m = con->in_msg;
1336 void *p;
1337 int ret;
9d7f0f13 1338 int to, left;
31b8006e
SW		 1339 unsigned front_len, middle_len, data_len, data_off;
1340 int datacrc = con->msgr->nocrc;
2450418c 1341 int skip;
ae18756b 1342 u64 seq;
31b8006e
SW		 1343
1344 dout("read_partial_message con %p msg %p\n", con, m);
1345
1346 /* header */
1347 while (con->in_base_pos < sizeof(con->in_hdr)) {
1348 left = sizeof(con->in_hdr) - con->in_base_pos;
1349 ret = ceph_tcp_recvmsg(con->sock,
1350 (char *)&con->in_hdr + con->in_base_pos,
1351 left);
1352 if (ret <= 0)
1353 return ret;
1354 con->in_base_pos += ret;
1355 if (con->in_base_pos == sizeof(con->in_hdr)) {
1356 u32 crc = crc32c(0, (void *)&con->in_hdr,
1357 sizeof(con->in_hdr) - sizeof(con->in_hdr.crc));
1358 if (crc != le32_to_cpu(con->in_hdr.crc)) {
1359 pr_err("read_partial_message bad hdr "
1360 " crc %u != expected %u\n",
1361 crc, con->in_hdr.crc);
1362 return -EBADMSG;
1363 }
1364 }
1365 }
31b8006e
SW		 1366 front_len = le32_to_cpu(con->in_hdr.front_len);
1367 if (front_len > CEPH_MSG_MAX_FRONT_LEN)
1368 return -EIO;
1369 middle_len = le32_to_cpu(con->in_hdr.middle_len);
1370 if (middle_len > CEPH_MSG_MAX_DATA_LEN)
1371 return -EIO;
1372 data_len = le32_to_cpu(con->in_hdr.data_len);
1373 if (data_len > CEPH_MSG_MAX_DATA_LEN)
1374 return -EIO;
9d7f0f13 1375 data_off = le16_to_cpu(con->in_hdr.data_off);
31b8006e 1376
ae18756b
SW		 1377 /* verify seq# */
1378 seq = le64_to_cpu(con->in_hdr.seq);
1379 if ((s64)seq - (s64)con->in_seq < 1) {
1380 pr_info("skipping %s%lld %s seq %lld, expected %lld\n",
1381 ENTITY_NAME(con->peer_name),
1382 pr_addr(&con->peer_addr.in_addr),
1383 seq, con->in_seq + 1);
1384 con->in_base_pos = -front_len - middle_len - data_len -
1385 sizeof(m->footer);
1386 con->in_tag = CEPH_MSGR_TAG_READY;
1387 con->in_seq++;
1388 return 0;
1389 } else if ((s64)seq - (s64)con->in_seq > 1) {
1390 pr_err("read_partial_message bad seq %lld expected %lld\n",
1391 seq, con->in_seq + 1);
1392 con->error_msg = "bad message sequence # for incoming message";
1393 return -EBADMSG;
1394 }
1395
31b8006e
SW		 1396 /* allocate message? */
1397 if (!con->in_msg) {
1398 dout("got hdr type %d front %d data %d\n", con->in_hdr.type,
1399 con->in_hdr.front_len, con->in_hdr.data_len);
ae32be31 1400 skip = 0;
2450418c
YS		 1401 con->in_msg = ceph_alloc_msg(con, &con->in_hdr, &skip);
1402 if (skip) {
31b8006e 1403 /* skip this message */
a79832f2 1404 dout("alloc_msg said skip message\n");
ae32be31 1405 BUG_ON(con->in_msg);
31b8006e
SW		 1406 con->in_base_pos = -front_len - middle_len - data_len -
1407 sizeof(m->footer);
1408 con->in_tag = CEPH_MSGR_TAG_READY;
684be25c 1409 con->in_seq++;
31b8006e
SW		 1410 return 0;
1411 }
a79832f2 1412 if (!con->in_msg) {
5b3a4db3
SW		 1413 con->error_msg =
1414 "error allocating memory for incoming message";
a79832f2 1415 return -ENOMEM;
31b8006e
SW		 1416 }
1417 m = con->in_msg;
1418 m->front.iov_len = 0; /* haven't read it yet */
2450418c
YS		 1419 if (m->middle)
1420 m->middle->vec.iov_len = 0;
9d7f0f13
YS		 1421
1422 con->in_msg_pos.page = 0;
1423 con->in_msg_pos.page_pos = data_off & ~PAGE_MASK;
1424 con->in_msg_pos.data_pos = 0;
31b8006e
SW		 1425 }
1426
1427 /* front */
2450418c
YS		 1428 ret = read_partial_message_section(con, &m->front, front_len,
1429 &con->in_front_crc);
1430 if (ret <= 0)
1431 return ret;
31b8006e
SW		 1432
1433 /* middle */
2450418c
YS		 1434 if (m->middle) {
1435 ret = read_partial_message_section(con, &m->middle->vec, middle_len,
1436 &con->in_middle_crc);
31b8006e
SW		 1437 if (ret <= 0)
1438 return ret;
31b8006e
SW		 1439 }
1440
1441 /* (page) data */
31b8006e
SW		 1442 while (con->in_msg_pos.data_pos < data_len) {
1443 left = min((int)(data_len - con->in_msg_pos.data_pos),
1444 (int)(PAGE_SIZE - con->in_msg_pos.page_pos));
1445 BUG_ON(m->pages == NULL);
1446 p = kmap(m->pages[con->in_msg_pos.page]);
1447 ret = ceph_tcp_recvmsg(con->sock, p + con->in_msg_pos.page_pos,
1448 left);
1449 if (ret > 0 && datacrc)
1450 con->in_data_crc =
1451 crc32c(con->in_data_crc,
1452 p + con->in_msg_pos.page_pos, ret);
1453 kunmap(m->pages[con->in_msg_pos.page]);
1454 if (ret <= 0)
1455 return ret;
1456 con->in_msg_pos.data_pos += ret;
1457 con->in_msg_pos.page_pos += ret;
1458 if (con->in_msg_pos.page_pos == PAGE_SIZE) {
1459 con->in_msg_pos.page_pos = 0;
1460 con->in_msg_pos.page++;
1461 }
1462 }
1463
31b8006e
SW		 1464 /* footer */
1465 to = sizeof(m->hdr) + sizeof(m->footer);
1466 while (con->in_base_pos < to) {
1467 left = to - con->in_base_pos;
1468 ret = ceph_tcp_recvmsg(con->sock, (char *)&m->footer +
1469 (con->in_base_pos - sizeof(m->hdr)),
1470 left);
1471 if (ret <= 0)
1472 return ret;
1473 con->in_base_pos += ret;
1474 }
1475 dout("read_partial_message got msg %p %d (%u) + %d (%u) + %d (%u)\n",
1476 m, front_len, m->footer.front_crc, middle_len,
1477 m->footer.middle_crc, data_len, m->footer.data_crc);
1478
1479 /* crc ok? */
1480 if (con->in_front_crc != le32_to_cpu(m->footer.front_crc)) {
1481 pr_err("read_partial_message %p front crc %u != exp. %u\n",
1482 m, con->in_front_crc, m->footer.front_crc);
1483 return -EBADMSG;
1484 }
1485 if (con->in_middle_crc != le32_to_cpu(m->footer.middle_crc)) {
1486 pr_err("read_partial_message %p middle crc %u != exp %u\n",
1487 m, con->in_middle_crc, m->footer.middle_crc);
1488 return -EBADMSG;
1489 }
1490 if (datacrc &&
1491 (m->footer.flags & CEPH_MSG_FOOTER_NOCRC) == 0 &&
1492 con->in_data_crc != le32_to_cpu(m->footer.data_crc)) {
1493 pr_err("read_partial_message %p data crc %u != exp. %u\n", m,
1494 con->in_data_crc, le32_to_cpu(m->footer.data_crc));
1495 return -EBADMSG;
1496 }
1497
1498 return 1; /* done! */
1499}
1500
1501/*
1502 * Process message. This happens in the worker thread. The callback should
1503 * be careful not to do anything that waits on other incoming messages or it
1504 * may deadlock.
1505 */
1506static void process_message(struct ceph_connection *con)
1507{
5e095e8b 1508 struct ceph_msg *msg;
31b8006e 1509
5e095e8b 1510 msg = con->in_msg;
31b8006e
SW		 1511 con->in_msg = NULL;
1512
1513 /* if first message, set peer_name */
1514 if (con->peer_name.type == 0)
dbad185d 1515 con->peer_name = msg->hdr.src;
31b8006e 1516
31b8006e 1517 con->in_seq++;
ec302645 1518 mutex_unlock(&con->mutex);
31b8006e
SW		 1519
1520 dout("===== %p %llu from %s%lld %d=%s len %d+%d (%u %u %u) =====\n",
1521 msg, le64_to_cpu(msg->hdr.seq),
dbad185d 1522 ENTITY_NAME(msg->hdr.src),
31b8006e
SW		 1523 le16_to_cpu(msg->hdr.type),
1524 ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
1525 le32_to_cpu(msg->hdr.front_len),
1526 le32_to_cpu(msg->hdr.data_len),
1527 con->in_front_crc, con->in_middle_crc, con->in_data_crc);
1528 con->ops->dispatch(con, msg);
ec302645
SW		 1529
1530 mutex_lock(&con->mutex);
31b8006e
SW		 1531 prepare_read_tag(con);
1532}
1533
1534
1535/*
1536 * Write something to the socket. Called in a worker thread when the
1537 * socket appears to be writeable and we have something ready to send.
1538 */
1539static int try_write(struct ceph_connection *con)
1540{
1541 struct ceph_messenger *msgr = con->msgr;
1542 int ret = 1;
1543
1544 dout("try_write start %p state %lu nref %d\n", con, con->state,
1545 atomic_read(&con->nref));
1546
31b8006e
SW		 1547more:
1548 dout("try_write out_kvec_bytes %d\n", con->out_kvec_bytes);
1549
1550 /* open the socket first? */
1551 if (con->sock == NULL) {
1552 /*
1553 * if we were STANDBY and are reconnecting _this_
1554 * connection, bump connect_seq now. Always bump
1555 * global_seq.
1556 */
1557 if (test_and_clear_bit(STANDBY, &con->state))
1558 con->connect_seq++;
1559
eed0ef2c
SW		 1560 prepare_write_banner(msgr, con);
1561 prepare_write_connect(msgr, con, 1);
1562 prepare_read_banner(con);
31b8006e 1563 set_bit(CONNECTING, &con->state);
eed0ef2c 1564 clear_bit(NEGOTIATING, &con->state);
31b8006e 1565
cf3e5c40 1566 BUG_ON(con->in_msg);
31b8006e
SW		 1567 con->in_tag = CEPH_MSGR_TAG_READY;
1568 dout("try_write initiating connect on %p new state %lu\n",
1569 con, con->state);
1570 con->sock = ceph_tcp_connect(con);
1571 if (IS_ERR(con->sock)) {
1572 con->sock = NULL;
1573 con->error_msg = "connect error";
1574 ret = -1;
1575 goto out;
1576 }
1577 }
1578
1579more_kvec:
1580 /* kvec data queued? */
1581 if (con->out_skip) {
1582 ret = write_partial_skip(con);
1583 if (ret <= 0)
1584 goto done;
1585 if (ret < 0) {
1586 dout("try_write write_partial_skip err %d\n", ret);
1587 goto done;
1588 }
1589 }
1590 if (con->out_kvec_left) {
1591 ret = write_partial_kvec(con);
1592 if (ret <= 0)
1593 goto done;
31b8006e
SW		 1594 }
1595
1596 /* msg pages? */
1597 if (con->out_msg) {
c86a2930
SW		 1598 if (con->out_msg_done) {
1599 ceph_msg_put(con->out_msg);
1600 con->out_msg = NULL; /* we're done with this one */
1601 goto do_next;
1602 }
1603
31b8006e
SW		 1604 ret = write_partial_msg_pages(con);
1605 if (ret == 1)
1606 goto more_kvec; /* we need to send the footer, too! */
1607 if (ret == 0)
1608 goto done;
1609 if (ret < 0) {
1610 dout("try_write write_partial_msg_pages err %d\n",
1611 ret);
1612 goto done;
1613 }
1614 }
1615
c86a2930 1616do_next:
31b8006e
SW		 1617 if (!test_bit(CONNECTING, &con->state)) {
1618 /* is anything else pending? */
1619 if (!list_empty(&con->out_queue)) {
1620 prepare_write_message(con);
1621 goto more;
1622 }
1623 if (con->in_seq > con->in_seq_acked) {
1624 prepare_write_ack(con);
1625 goto more;
1626 }
1627 if (test_and_clear_bit(KEEPALIVE_PENDING, &con->state)) {
1628 prepare_write_keepalive(con);
1629 goto more;
1630 }
1631 }
1632
1633 /* Nothing to do! */
1634 clear_bit(WRITE_PENDING, &con->state);
1635 dout("try_write nothing else to write.\n");
1636done:
1637 ret = 0;
1638out:
31b8006e
SW		 1639 dout("try_write done on %p\n", con);
1640 return ret;
1641}
1642
1643
1644
1645/*
1646 * Read what we can from the socket.
1647 */
1648static int try_read(struct ceph_connection *con)
1649{
31b8006e
SW		 1650 int ret = -1;
1651
1652 if (!con->sock)
1653 return 0;
1654
1655 if (test_bit(STANDBY, &con->state))
1656 return 0;
1657
1658 dout("try_read start on %p\n", con);
ec302645 1659
31b8006e
SW		 1660more:
1661 dout("try_read tag %d in_base_pos %d\n", (int)con->in_tag,
1662 con->in_base_pos);
1663 if (test_bit(CONNECTING, &con->state)) {
eed0ef2c
SW		 1664 if (!test_bit(NEGOTIATING, &con->state)) {
1665 dout("try_read connecting\n");
1666 ret = read_partial_banner(con);
1667 if (ret <= 0)
1668 goto done;
1669 if (process_banner(con) < 0) {
1670 ret = -1;
1671 goto out;
1672 }
1673 }
31b8006e
SW		 1674 ret = read_partial_connect(con);
1675 if (ret <= 0)
1676 goto done;
1677 if (process_connect(con) < 0) {
1678 ret = -1;
1679 goto out;
1680 }
1681 goto more;
1682 }
1683
1684 if (con->in_base_pos < 0) {
1685 /*
1686 * skipping + discarding content.
1687 *
1688 * FIXME: there must be a better way to do this!
1689 */
1690 static char buf[1024];
1691 int skip = min(1024, -con->in_base_pos);
1692 dout("skipping %d / %d bytes\n", skip, -con->in_base_pos);
1693 ret = ceph_tcp_recvmsg(con->sock, buf, skip);
1694 if (ret <= 0)
1695 goto done;
1696 con->in_base_pos += ret;
1697 if (con->in_base_pos)
1698 goto more;
1699 }
1700 if (con->in_tag == CEPH_MSGR_TAG_READY) {
1701 /*
1702 * what's next?
1703 */
1704 ret = ceph_tcp_recvmsg(con->sock, &con->in_tag, 1);
1705 if (ret <= 0)
1706 goto done;
1707 dout("try_read got tag %d\n", (int)con->in_tag);
1708 switch (con->in_tag) {
1709 case CEPH_MSGR_TAG_MSG:
1710 prepare_read_message(con);
1711 break;
1712 case CEPH_MSGR_TAG_ACK:
1713 prepare_read_ack(con);
1714 break;
1715 case CEPH_MSGR_TAG_CLOSE:
1716 set_bit(CLOSED, &con->state); /* fixme */
1717 goto done;
1718 default:
1719 goto bad_tag;
1720 }
1721 }
1722 if (con->in_tag == CEPH_MSGR_TAG_MSG) {
1723 ret = read_partial_message(con);
1724 if (ret <= 0) {
1725 switch (ret) {
1726 case -EBADMSG:
1727 con->error_msg = "bad crc";
1728 ret = -EIO;
1729 goto out;
1730 case -EIO:
1731 con->error_msg = "io error";
1732 goto out;
1733 default:
1734 goto done;
1735 }
1736 }
1737 if (con->in_tag == CEPH_MSGR_TAG_READY)
1738 goto more;
1739 process_message(con);
1740 goto more;
1741 }
1742 if (con->in_tag == CEPH_MSGR_TAG_ACK) {
1743 ret = read_partial_ack(con);
1744 if (ret <= 0)
1745 goto done;
1746 process_ack(con);
1747 goto more;
1748 }
1749
1750done:
1751 ret = 0;
1752out:
1753 dout("try_read done on %p\n", con);
1754 return ret;
1755
1756bad_tag:
1757 pr_err("try_read bad con->in_tag = %d\n", (int)con->in_tag);
1758 con->error_msg = "protocol error, garbage tag";
1759 ret = -1;
1760 goto out;
1761}
1762
1763
1764/*
1765 * Atomically queue work on a connection. Bump @con reference to
1766 * avoid races with connection teardown.
1767 *
1768 * There is some trickery going on with QUEUED and BUSY because we
1769 * only want a _single_ thread operating on each connection at any
1770 * point in time, but we want to use all available CPUs.
1771 *
1772 * The worker thread only proceeds if it can atomically set BUSY. It
1773 * clears QUEUED and does it's thing. When it thinks it's done, it
1774 * clears BUSY, then rechecks QUEUED.. if it's set again, it loops
1775 * (tries again to set BUSY).
1776 *
1777 * To queue work, we first set QUEUED, _then_ if BUSY isn't set, we
1778 * try to queue work. If that fails (work is already queued, or BUSY)
1779 * we give up (work also already being done or is queued) but leave QUEUED
1780 * set so that the worker thread will loop if necessary.
1781 */
1782static void queue_con(struct ceph_connection *con)
1783{
1784 if (test_bit(DEAD, &con->state)) {
1785 dout("queue_con %p ignoring: DEAD\n",
1786 con);
1787 return;
1788 }
1789
1790 if (!con->ops->get(con)) {
1791 dout("queue_con %p ref count 0\n", con);
1792 return;
1793 }
1794
1795 set_bit(QUEUED, &con->state);
1796 if (test_bit(BUSY, &con->state)) {
1797 dout("queue_con %p - already BUSY\n", con);
1798 con->ops->put(con);
1799 } else if (!queue_work(ceph_msgr_wq, &con->work.work)) {
1800 dout("queue_con %p - already queued\n", con);
1801 con->ops->put(con);
1802 } else {
1803 dout("queue_con %p\n", con);
1804 }
1805}
1806
1807/*
1808 * Do some work on a connection. Drop a connection ref when we're done.
1809 */
1810static void con_work(struct work_struct *work)
1811{
1812 struct ceph_connection *con = container_of(work, struct ceph_connection,
1813 work.work);
1814 int backoff = 0;
1815
1816more:
1817 if (test_and_set_bit(BUSY, &con->state) != 0) {
1818 dout("con_work %p BUSY already set\n", con);
1819 goto out;
1820 }
1821 dout("con_work %p start, clearing QUEUED\n", con);
1822 clear_bit(QUEUED, &con->state);
1823
9dd4658d
SW		 1824 mutex_lock(&con->mutex);
1825
31b8006e
SW		 1826 if (test_bit(CLOSED, &con->state)) { /* e.g. if we are replaced */
1827 dout("con_work CLOSED\n");
1828 con_close_socket(con);
1829 goto done;
1830 }
1831 if (test_and_clear_bit(OPENING, &con->state)) {
1832 /* reopen w/ new peer */
1833 dout("con_work OPENING\n");
1834 con_close_socket(con);
1835 }
1836
1837 if (test_and_clear_bit(SOCK_CLOSED, &con->state) ||
1838 try_read(con) < 0 ||
1839 try_write(con) < 0) {
9dd4658d 1840 mutex_unlock(&con->mutex);
31b8006e
SW		 1841 backoff = 1;
1842 ceph_fault(con); /* error/fault path */
9dd4658d 1843 goto done_unlocked;
31b8006e
SW		 1844 }
1845
1846done:
9dd4658d
SW		 1847 mutex_unlock(&con->mutex);
1848
1849done_unlocked:
31b8006e
SW		 1850 clear_bit(BUSY, &con->state);
1851 dout("con->state=%lu\n", con->state);
1852 if (test_bit(QUEUED, &con->state)) {
e2663ab6 1853 if (!backoff || test_bit(OPENING, &con->state)) {
31b8006e
SW		 1854 dout("con_work %p QUEUED reset, looping\n", con);
1855 goto more;
1856 }
1857 dout("con_work %p QUEUED reset, but just faulted\n", con);
1858 clear_bit(QUEUED, &con->state);
1859 }
1860 dout("con_work %p done\n", con);
1861
1862out:
1863 con->ops->put(con);
1864}
1865
1866
1867/*
1868 * Generic error/fault handler. A retry mechanism is used with
1869 * exponential backoff
1870 */
1871static void ceph_fault(struct ceph_connection *con)
1872{
1873 pr_err("%s%lld %s %s\n", ENTITY_NAME(con->peer_name),
1874 pr_addr(&con->peer_addr.in_addr), con->error_msg);
1875 dout("fault %p state %lu to peer %s\n",
1876 con, con->state, pr_addr(&con->peer_addr.in_addr));
1877
1878 if (test_bit(LOSSYTX, &con->state)) {
1879 dout("fault on LOSSYTX channel\n");
1880 goto out;
1881 }
1882
ec302645 1883 mutex_lock(&con->mutex);
91e45ce3
SW		 1884 if (test_bit(CLOSED, &con->state))
1885 goto out_unlock;
ec302645 1886
31b8006e 1887 con_close_socket(con);
5e095e8b
SW		 1888
1889 if (con->in_msg) {
1890 ceph_msg_put(con->in_msg);
1891 con->in_msg = NULL;
1892 }
31b8006e 1893
e80a52d1
SW		 1894 /* Requeue anything that hasn't been acked */
1895 list_splice_init(&con->out_sent, &con->out_queue);
9bd2e6f8 1896
31b8006e
SW		 1897 /* If there are no messages in the queue, place the connection
1898 * in a STANDBY state (i.e., don't try to reconnect just yet). */
31b8006e
SW		 1899 if (list_empty(&con->out_queue) && !con->out_keepalive_pending) {
1900 dout("fault setting STANDBY\n");
1901 set_bit(STANDBY, &con->state);
e80a52d1
SW		 1902 } else {
1903 /* retry after a delay. */
1904 if (con->delay == 0)
1905 con->delay = BASE_DELAY_INTERVAL;
1906 else if (con->delay < MAX_DELAY_INTERVAL)
1907 con->delay *= 2;
1908 dout("fault queueing %p delay %lu\n", con, con->delay);
1909 con->ops->get(con);
1910 if (queue_delayed_work(ceph_msgr_wq, &con->work,
1911 round_jiffies_relative(con->delay)) == 0)
1912 con->ops->put(con);
31b8006e
SW		 1913 }
1914
91e45ce3
SW		 1915out_unlock:
1916 mutex_unlock(&con->mutex);
31b8006e 1917out:
161fd65a
SW		 1918 /*
1919 * in case we faulted due to authentication, invalidate our
1920 * current tickets so that we can get new ones.
1921 */
1922 if (con->auth_retry && con->ops->invalidate_authorizer) {
1923 dout("calling invalidate_authorizer()\n");
1924 con->ops->invalidate_authorizer(con);
1925 }
1926
31b8006e
SW		 1927 if (con->ops->fault)
1928 con->ops->fault(con);
1929}
1930
1931
1932
1933/*
1934 * create a new messenger instance
1935 */
1936struct ceph_messenger *ceph_messenger_create(struct ceph_entity_addr *myaddr)
1937{
1938 struct ceph_messenger *msgr;
1939
1940 msgr = kzalloc(sizeof(*msgr), GFP_KERNEL);
1941 if (msgr == NULL)
1942 return ERR_PTR(-ENOMEM);
1943
1944 spin_lock_init(&msgr->global_seq_lock);
1945
1946 /* the zero page is needed if a request is "canceled" while the message
1947 * is being written over the socket */
31459fe4 1948 msgr->zero_page = __page_cache_alloc(GFP_KERNEL | __GFP_ZERO);
31b8006e
SW		 1949 if (!msgr->zero_page) {
1950 kfree(msgr);
1951 return ERR_PTR(-ENOMEM);
1952 }
1953 kmap(msgr->zero_page);
1954
1955 if (myaddr)
1956 msgr->inst.addr = *myaddr;
1957
1958 /* select a random nonce */
ac8839d7 1959 msgr->inst.addr.type = 0;
103e2d3a 1960 get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
63f2d211 1961 encode_my_addr(msgr);
31b8006e
SW		 1962
1963 dout("messenger_create %p\n", msgr);
1964 return msgr;
1965}
1966
1967void ceph_messenger_destroy(struct ceph_messenger *msgr)
1968{
1969 dout("destroy %p\n", msgr);
1970 kunmap(msgr->zero_page);
1971 __free_page(msgr->zero_page);
1972 kfree(msgr);
1973 dout("destroyed messenger %p\n", msgr);
1974}
1975
1976/*
1977 * Queue up an outgoing message on the given connection.
1978 */
1979void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg)
1980{
1981 if (test_bit(CLOSED, &con->state)) {
1982 dout("con_send %p closed, dropping %p\n", con, msg);
1983 ceph_msg_put(msg);
1984 return;
1985 }
1986
1987 /* set src+dst */
dbad185d 1988 msg->hdr.src = con->msgr->inst.name;
31b8006e 1989
3ca02ef9
SW		 1990 BUG_ON(msg->front.iov_len != le32_to_cpu(msg->hdr.front_len));
1991
e84346b7
SW		 1992 msg->needs_out_seq = true;
1993
31b8006e 1994 /* queue */
ec302645 1995 mutex_lock(&con->mutex);
31b8006e
SW		 1996 BUG_ON(!list_empty(&msg->list_head));
1997 list_add_tail(&msg->list_head, &con->out_queue);
1998 dout("----- %p to %s%lld %d=%s len %d+%d+%d -----\n", msg,
1999 ENTITY_NAME(con->peer_name), le16_to_cpu(msg->hdr.type),
2000 ceph_msg_type_name(le16_to_cpu(msg->hdr.type)),
2001 le32_to_cpu(msg->hdr.front_len),
2002 le32_to_cpu(msg->hdr.middle_len),
2003 le32_to_cpu(msg->hdr.data_len));
ec302645 2004 mutex_unlock(&con->mutex);
31b8006e
SW		 2005
2006 /* if there wasn't anything waiting to send before, queue
2007 * new work */
2008 if (test_and_set_bit(WRITE_PENDING, &con->state) == 0)
2009 queue_con(con);
2010}
2011
2012/*
2013 * Revoke a message that was previously queued for send
2014 */
2015void ceph_con_revoke(struct ceph_connection *con, struct ceph_msg *msg)
2016{
ec302645 2017 mutex_lock(&con->mutex);
31b8006e 2018 if (!list_empty(&msg->list_head)) {
ed98adad 2019 dout("con_revoke %p msg %p - was on queue\n", con, msg);
31b8006e
SW		 2020 list_del_init(&msg->list_head);
2021 ceph_msg_put(msg);
2022 msg->hdr.seq = 0;
ed98adad
SW		 2023 }
2024 if (con->out_msg == msg) {
2025 dout("con_revoke %p msg %p - was sending\n", con, msg);
2026 con->out_msg = NULL;
31b8006e
SW		 2027 if (con->out_kvec_is_msg) {
2028 con->out_skip = con->out_kvec_bytes;
2029 con->out_kvec_is_msg = false;
2030 }
ed98adad
SW		 2031 ceph_msg_put(msg);
2032 msg->hdr.seq = 0;
31b8006e 2033 }
ec302645 2034 mutex_unlock(&con->mutex);
31b8006e
SW		 2035}
2036
350b1c32 2037/*
0d59ab81 2038 * Revoke a message that we may be reading data into
350b1c32 2039 */
0d59ab81 2040void ceph_con_revoke_message(struct ceph_connection *con, struct ceph_msg *msg)
350b1c32
SW		 2041{
2042 mutex_lock(&con->mutex);
0d59ab81
YS		 2043 if (con->in_msg && con->in_msg == msg) {
2044 unsigned front_len = le32_to_cpu(con->in_hdr.front_len);
2045 unsigned middle_len = le32_to_cpu(con->in_hdr.middle_len);
350b1c32
SW		 2046 unsigned data_len = le32_to_cpu(con->in_hdr.data_len);
2047
2048 /* skip rest of message */
0d59ab81 2049 dout("con_revoke_pages %p msg %p revoked\n", con, msg);
350b1c32
SW		 2050 con->in_base_pos = con->in_base_pos -
2051 sizeof(struct ceph_msg_header) -
0d59ab81
YS		 2052 front_len -
2053 middle_len -
2054 data_len -
350b1c32 2055 sizeof(struct ceph_msg_footer);
350b1c32
SW		 2056 ceph_msg_put(con->in_msg);
2057 con->in_msg = NULL;
2058 con->in_tag = CEPH_MSGR_TAG_READY;
684be25c 2059 con->in_seq++;
350b1c32
SW		 2060 } else {
2061 dout("con_revoke_pages %p msg %p pages %p no-op\n",
0d59ab81 2062 con, con->in_msg, msg);
350b1c32
SW		 2063 }
2064 mutex_unlock(&con->mutex);
2065}
2066
31b8006e
SW		 2067/*
2068 * Queue a keepalive byte to ensure the tcp connection is alive.
2069 */
2070void ceph_con_keepalive(struct ceph_connection *con)
2071{
2072 if (test_and_set_bit(KEEPALIVE_PENDING, &con->state) == 0 &&
2073 test_and_set_bit(WRITE_PENDING, &con->state) == 0)
2074 queue_con(con);
2075}
2076
2077
2078/*
2079 * construct a new message with given type, size
2080 * the new msg has a ref count of 1.
2081 */
34d23762 2082struct ceph_msg *ceph_msg_new(int type, int front_len, gfp_t flags)
31b8006e
SW		 2083{
2084 struct ceph_msg *m;
2085
34d23762 2086 m = kmalloc(sizeof(*m), flags);
31b8006e
SW		 2087 if (m == NULL)
2088 goto out;
c2e552e7 2089 kref_init(&m->kref);
31b8006e
SW		 2090 INIT_LIST_HEAD(&m->list_head);
2091
45c6ceb5 2092 m->hdr.tid = 0;
31b8006e 2093 m->hdr.type = cpu_to_le16(type);
45c6ceb5
SW		 2094 m->hdr.priority = cpu_to_le16(CEPH_MSG_PRIO_DEFAULT);
2095 m->hdr.version = 0;
31b8006e
SW		 2096 m->hdr.front_len = cpu_to_le32(front_len);
2097 m->hdr.middle_len = 0;
bb257664
SW		 2098 m->hdr.data_len = 0;
2099 m->hdr.data_off = 0;
45c6ceb5 2100 m->hdr.reserved = 0;
31b8006e
SW		 2101 m->footer.front_crc = 0;
2102 m->footer.middle_crc = 0;
2103 m->footer.data_crc = 0;
45c6ceb5 2104 m->footer.flags = 0;
31b8006e
SW		 2105 m->front_max = front_len;
2106 m->front_is_vmalloc = false;
2107 m->more_to_follow = false;
2108 m->pool = NULL;
2109
2110 /* front */
2111 if (front_len) {
2112 if (front_len > PAGE_CACHE_SIZE) {
34d23762 2113 m->front.iov_base = __vmalloc(front_len, flags,
31b8006e
SW		 2114 PAGE_KERNEL);
2115 m->front_is_vmalloc = true;
2116 } else {
34d23762 2117 m->front.iov_base = kmalloc(front_len, flags);
31b8006e
SW		 2118 }
2119 if (m->front.iov_base == NULL) {
2120 pr_err("msg_new can't allocate %d bytes\n",
2121 front_len);
2122 goto out2;
2123 }
2124 } else {
2125 m->front.iov_base = NULL;
2126 }
2127 m->front.iov_len = front_len;
2128
2129 /* middle */
2130 m->middle = NULL;
2131
2132 /* data */
bb257664
SW		 2133 m->nr_pages = 0;
2134 m->pages = NULL;
58bb3b37 2135 m->pagelist = NULL;
31b8006e 2136
bb257664 2137 dout("ceph_msg_new %p front %d\n", m, front_len);
31b8006e
SW		 2138 return m;
2139
2140out2:
2141 ceph_msg_put(m);
2142out:
bb257664 2143 pr_err("msg_new can't create type %d front %d\n", type, front_len);
a79832f2 2144 return NULL;
31b8006e
SW		 2145}
2146
31b8006e
SW		 2147/*
2148 * Allocate "middle" portion of a message, if it is needed and wasn't
2149 * allocated by alloc_msg. This allows us to read a small fixed-size
2150 * per-type header in the front and then gracefully fail (i.e.,
2151 * propagate the error to the caller based on info in the front) when
2152 * the middle is too large.
2153 */
2450418c 2154static int ceph_alloc_middle(struct ceph_connection *con, struct ceph_msg *msg)
31b8006e
SW		 2155{
2156 int type = le16_to_cpu(msg->hdr.type);
2157 int middle_len = le32_to_cpu(msg->hdr.middle_len);
2158
2159 dout("alloc_middle %p type %d %s middle_len %d\n", msg, type,
2160 ceph_msg_type_name(type), middle_len);
2161 BUG_ON(!middle_len);
2162 BUG_ON(msg->middle);
2163
b6c1d5b8 2164 msg->middle = ceph_buffer_new(middle_len, GFP_NOFS);
31b8006e
SW		 2165 if (!msg->middle)
2166 return -ENOMEM;
2167 return 0;
2168}
2169
2450418c
YS		 2170/*
2171 * Generic message allocator, for incoming messages.
2172 */
2173static struct ceph_msg *ceph_alloc_msg(struct ceph_connection *con,
2174 struct ceph_msg_header *hdr,
2175 int *skip)
2176{
2177 int type = le16_to_cpu(hdr->type);
2178 int front_len = le32_to_cpu(hdr->front_len);
2179 int middle_len = le32_to_cpu(hdr->middle_len);
2180 struct ceph_msg *msg = NULL;
2181 int ret;
2182
2183 if (con->ops->alloc_msg) {
0547a9b3 2184 mutex_unlock(&con->mutex);
2450418c 2185 msg = con->ops->alloc_msg(con, hdr, skip);
0547a9b3 2186 mutex_lock(&con->mutex);
a79832f2 2187 if (!msg || *skip)
2450418c
YS		 2188 return NULL;
2189 }
2190 if (!msg) {
2191 *skip = 0;
34d23762 2192 msg = ceph_msg_new(type, front_len, GFP_NOFS);
2450418c
YS		 2193 if (!msg) {
2194 pr_err("unable to allocate msg type %d len %d\n",
2195 type, front_len);
a79832f2 2196 return NULL;
2450418c
YS		 2197 }
2198 }
9d7f0f13 2199 memcpy(&msg->hdr, &con->in_hdr, sizeof(con->in_hdr));
2450418c 2200
bb257664 2201 if (middle_len && !msg->middle) {
2450418c 2202 ret = ceph_alloc_middle(con, msg);
2450418c
YS		 2203 if (ret < 0) {
2204 ceph_msg_put(msg);
a79832f2 2205 return NULL;
2450418c
YS		 2206 }
2207 }
9d7f0f13 2208
2450418c
YS		 2209 return msg;
2210}
2211
31b8006e
SW		 2212
2213/*
2214 * Free a generically kmalloc'd message.
2215 */
2216void ceph_msg_kfree(struct ceph_msg *m)
2217{
2218 dout("msg_kfree %p\n", m);
2219 if (m->front_is_vmalloc)
2220 vfree(m->front.iov_base);
2221 else
2222 kfree(m->front.iov_base);
2223 kfree(m);
2224}
2225
2226/*
2227 * Drop a msg ref. Destroy as needed.
2228 */
c2e552e7
SW		 2229void ceph_msg_last_put(struct kref *kref)
2230{
2231 struct ceph_msg *m = container_of(kref, struct ceph_msg, kref);
31b8006e 2232
c2e552e7
SW		 2233 dout("ceph_msg_put last one on %p\n", m);
2234 WARN_ON(!list_empty(&m->list_head));
2235
2236 /* drop middle, data, if any */
2237 if (m->middle) {
2238 ceph_buffer_put(m->middle);
2239 m->middle = NULL;
31b8006e 2240 }
c2e552e7
SW		 2241 m->nr_pages = 0;
2242 m->pages = NULL;
2243
58bb3b37
SW		 2244 if (m->pagelist) {
2245 ceph_pagelist_release(m->pagelist);
2246 kfree(m->pagelist);
2247 m->pagelist = NULL;
2248 }
2249
c2e552e7
SW		 2250 if (m->pool)
2251 ceph_msgpool_put(m->pool, m);
2252 else
2253 ceph_msg_kfree(m);
31b8006e 2254}
9ec7cab1
SW		 2255
2256void ceph_msg_dump(struct ceph_msg *msg)
2257{
2258 pr_debug("msg_dump %p (front_max %d nr_pages %d)\n", msg,
2259 msg->front_max, msg->nr_pages);
2260 print_hex_dump(KERN_DEBUG, "header: ",
2261 DUMP_PREFIX_OFFSET, 16, 1,
2262 &msg->hdr, sizeof(msg->hdr), true);
2263 print_hex_dump(KERN_DEBUG, " front: ",
2264 DUMP_PREFIX_OFFSET, 16, 1,
2265 msg->front.iov_base, msg->front.iov_len, true);
2266 if (msg->middle)
2267 print_hex_dump(KERN_DEBUG, "middle: ",
2268 DUMP_PREFIX_OFFSET, 16, 1,
2269 msg->middle->vec.iov_base,
2270 msg->middle->vec.iov_len, true);
2271 print_hex_dump(KERN_DEBUG, "footer: ",
2272 DUMP_PREFIX_OFFSET, 16, 1,
2273 &msg->footer, sizeof(msg->footer), true);
2274}
kernel source for telekom puls (alcatel/mediatek)