1 #include <linux/ceph/ceph_debug.h>
3 #include <linux/module.h>
4 #include <linux/types.h>
5 #include <linux/slab.h>
6 #include <linux/random.h>
7 #include <linux/sched.h>
9 #include <linux/ceph/mon_client.h>
10 #include <linux/ceph/libceph.h>
11 #include <linux/ceph/debugfs.h>
12 #include <linux/ceph/decode.h>
13 #include <linux/ceph/auth.h>
16 * Interact with Ceph monitor cluster. Handle requests for new map
17 * versions, and periodically resend as needed. Also implement
18 * statfs() and umount().
20 * A small cluster of Ceph "monitors" are responsible for managing critical
21 * cluster configuration and state information. An odd number (e.g., 3, 5)
22 * of cmon daemons use a modified version of the Paxos part-time parliament
23 * algorithm to manage the MDS map (mds cluster membership), OSD map, and
24 * list of clients who have mounted the file system.
26 * We maintain an open, active session with a monitor at all times in order to
27 * receive timely MDSMap updates. We periodically send a keepalive byte on the
28 * TCP socket to ensure we detect a failure. If the connection does break, we
29 * randomly hunt for a new monitor. Once the connection is reestablished, we
30 * resend any outstanding requests.
33 static const struct ceph_connection_operations mon_con_ops
;
35 static int __validate_auth(struct ceph_mon_client
*monc
);
38 * Decode a monmap blob (e.g., during mount).
40 struct ceph_monmap
*ceph_monmap_decode(void *p
, void *end
)
42 struct ceph_monmap
*m
= NULL
;
44 struct ceph_fsid fsid
;
49 ceph_decode_32_safe(&p
, end
, len
, bad
);
50 ceph_decode_need(&p
, end
, len
, bad
);
52 dout("monmap_decode %p %p len %d\n", p
, end
, (int)(end
-p
));
54 ceph_decode_16_safe(&p
, end
, version
, bad
);
56 ceph_decode_need(&p
, end
, sizeof(fsid
) + 2*sizeof(u32
), bad
);
57 ceph_decode_copy(&p
, &fsid
, sizeof(fsid
));
58 epoch
= ceph_decode_32(&p
);
60 num_mon
= ceph_decode_32(&p
);
61 ceph_decode_need(&p
, end
, num_mon
*sizeof(m
->mon_inst
[0]), bad
);
63 if (num_mon
>= CEPH_MAX_MON
)
65 m
= kmalloc(sizeof(*m
) + sizeof(m
->mon_inst
[0])*num_mon
, GFP_NOFS
);
67 return ERR_PTR(-ENOMEM
);
71 ceph_decode_copy(&p
, m
->mon_inst
, num_mon
*sizeof(m
->mon_inst
[0]));
72 for (i
= 0; i
< num_mon
; i
++)
73 ceph_decode_addr(&m
->mon_inst
[i
].addr
);
75 dout("monmap_decode epoch %d, num_mon %d\n", m
->epoch
,
77 for (i
= 0; i
< m
->num_mon
; i
++)
78 dout("monmap_decode mon%d is %s\n", i
,
79 ceph_pr_addr(&m
->mon_inst
[i
].addr
.in_addr
));
83 dout("monmap_decode failed with %d\n", err
);
89 * return true if *addr is included in the monmap.
91 int ceph_monmap_contains(struct ceph_monmap
*m
, struct ceph_entity_addr
*addr
)
95 for (i
= 0; i
< m
->num_mon
; i
++)
96 if (memcmp(addr
, &m
->mon_inst
[i
].addr
, sizeof(*addr
)) == 0)
102 * Send an auth request.
104 static void __send_prepared_auth_request(struct ceph_mon_client
*monc
, int len
)
106 monc
->pending_auth
= 1;
107 monc
->m_auth
->front
.iov_len
= len
;
108 monc
->m_auth
->hdr
.front_len
= cpu_to_le32(len
);
109 ceph_msg_revoke(monc
->m_auth
);
110 ceph_msg_get(monc
->m_auth
); /* keep our ref */
111 ceph_con_send(&monc
->con
, monc
->m_auth
);
115 * Close monitor session, if any.
117 static void __close_session(struct ceph_mon_client
*monc
)
119 dout("__close_session closing mon%d\n", monc
->cur_mon
);
120 ceph_msg_revoke(monc
->m_auth
);
121 ceph_msg_revoke_incoming(monc
->m_auth_reply
);
122 ceph_msg_revoke(monc
->m_subscribe
);
123 ceph_msg_revoke_incoming(monc
->m_subscribe_ack
);
124 ceph_con_close(&monc
->con
);
126 monc
->pending_auth
= 0;
127 ceph_auth_reset(monc
->auth
);
131 * Pick a new monitor at random and set cur_mon. If we are repicking
132 * (i.e. cur_mon is already set), be sure to pick a different one.
134 static void pick_new_mon(struct ceph_mon_client
*monc
)
136 int old_mon
= monc
->cur_mon
;
138 BUG_ON(monc
->monmap
->num_mon
< 1);
140 if (monc
->monmap
->num_mon
== 1) {
143 int max
= monc
->monmap
->num_mon
;
147 if (monc
->cur_mon
>= 0) {
148 if (monc
->cur_mon
< monc
->monmap
->num_mon
)
154 n
= prandom_u32() % max
;
155 if (o
>= 0 && n
>= o
)
161 dout("%s mon%d -> mon%d out of %d mons\n", __func__
, old_mon
,
162 monc
->cur_mon
, monc
->monmap
->num_mon
);
166 * Open a session with a new monitor.
168 static void __open_session(struct ceph_mon_client
*monc
)
174 monc
->hunting
= true;
175 if (monc
->had_a_connection
) {
176 monc
->hunt_mult
*= CEPH_MONC_HUNT_BACKOFF
;
177 if (monc
->hunt_mult
> CEPH_MONC_HUNT_MAX_MULT
)
178 monc
->hunt_mult
= CEPH_MONC_HUNT_MAX_MULT
;
181 monc
->sub_renew_after
= jiffies
; /* i.e., expired */
182 monc
->sub_renew_sent
= 0;
184 dout("%s opening mon%d\n", __func__
, monc
->cur_mon
);
185 ceph_con_open(&monc
->con
, CEPH_ENTITY_TYPE_MON
, monc
->cur_mon
,
186 &monc
->monmap
->mon_inst
[monc
->cur_mon
].addr
);
189 * send an initial keepalive to ensure our timestamp is valid
190 * by the time we are in an OPENED state
192 ceph_con_keepalive(&monc
->con
);
194 /* initiate authentication handshake */
195 ret
= ceph_auth_build_hello(monc
->auth
,
196 monc
->m_auth
->front
.iov_base
,
197 monc
->m_auth
->front_alloc_len
);
199 __send_prepared_auth_request(monc
, ret
);
202 static void reopen_session(struct ceph_mon_client
*monc
)
205 pr_info("mon%d %s session lost, hunting for new mon\n",
206 monc
->cur_mon
, ceph_pr_addr(&monc
->con
.peer_addr
.in_addr
));
208 __close_session(monc
);
209 __open_session(monc
);
213 * Reschedule delayed work timer.
215 static void __schedule_delayed(struct ceph_mon_client
*monc
)
220 delay
= CEPH_MONC_HUNT_INTERVAL
* monc
->hunt_mult
;
222 delay
= CEPH_MONC_PING_INTERVAL
;
224 dout("__schedule_delayed after %lu\n", delay
);
225 mod_delayed_work(system_wq
, &monc
->delayed_work
,
226 round_jiffies_relative(delay
));
229 const char *ceph_sub_str
[] = {
230 [CEPH_SUB_MDSMAP
] = "mdsmap",
231 [CEPH_SUB_MONMAP
] = "monmap",
232 [CEPH_SUB_OSDMAP
] = "osdmap",
236 * Send subscribe request for one or more maps, according to
239 static void __send_subscribe(struct ceph_mon_client
*monc
)
241 struct ceph_msg
*msg
= monc
->m_subscribe
;
242 void *p
= msg
->front
.iov_base
;
243 void *const end
= p
+ msg
->front_alloc_len
;
247 dout("%s sent %lu\n", __func__
, monc
->sub_renew_sent
);
249 BUG_ON(monc
->cur_mon
< 0);
251 if (!monc
->sub_renew_sent
)
252 monc
->sub_renew_sent
= jiffies
| 1; /* never 0 */
254 msg
->hdr
.version
= cpu_to_le16(2);
256 for (i
= 0; i
< ARRAY_SIZE(monc
->subs
); i
++) {
257 if (monc
->subs
[i
].want
)
260 BUG_ON(num
< 1); /* monmap sub is always there */
261 ceph_encode_32(&p
, num
);
262 for (i
= 0; i
< ARRAY_SIZE(monc
->subs
); i
++) {
263 const char *s
= ceph_sub_str
[i
];
265 if (!monc
->subs
[i
].want
)
268 dout("%s %s start %llu flags 0x%x\n", __func__
, s
,
269 le64_to_cpu(monc
->subs
[i
].item
.start
),
270 monc
->subs
[i
].item
.flags
);
271 ceph_encode_string(&p
, end
, s
, strlen(s
));
272 memcpy(p
, &monc
->subs
[i
].item
, sizeof(monc
->subs
[i
].item
));
273 p
+= sizeof(monc
->subs
[i
].item
);
276 BUG_ON(p
!= (end
- 35 - (ARRAY_SIZE(monc
->subs
) - num
) * 19));
277 msg
->front
.iov_len
= p
- msg
->front
.iov_base
;
278 msg
->hdr
.front_len
= cpu_to_le32(msg
->front
.iov_len
);
279 ceph_msg_revoke(msg
);
280 ceph_con_send(&monc
->con
, ceph_msg_get(msg
));
283 static void handle_subscribe_ack(struct ceph_mon_client
*monc
,
284 struct ceph_msg
*msg
)
286 unsigned int seconds
;
287 struct ceph_mon_subscribe_ack
*h
= msg
->front
.iov_base
;
289 if (msg
->front
.iov_len
< sizeof(*h
))
291 seconds
= le32_to_cpu(h
->duration
);
293 mutex_lock(&monc
->mutex
);
294 if (monc
->sub_renew_sent
) {
295 monc
->sub_renew_after
= monc
->sub_renew_sent
+
296 (seconds
>> 1) * HZ
- 1;
297 dout("%s sent %lu duration %d renew after %lu\n", __func__
,
298 monc
->sub_renew_sent
, seconds
, monc
->sub_renew_after
);
299 monc
->sub_renew_sent
= 0;
301 dout("%s sent %lu renew after %lu, ignoring\n", __func__
,
302 monc
->sub_renew_sent
, monc
->sub_renew_after
);
304 mutex_unlock(&monc
->mutex
);
307 pr_err("got corrupt subscribe-ack msg\n");
312 * Register interest in a map
314 * @sub: one of CEPH_SUB_*
315 * @epoch: X for "every map since X", or 0 for "just the latest"
317 static bool __ceph_monc_want_map(struct ceph_mon_client
*monc
, int sub
,
318 u32 epoch
, bool continuous
)
320 __le64 start
= cpu_to_le64(epoch
);
321 u8 flags
= !continuous
? CEPH_SUBSCRIBE_ONETIME
: 0;
323 dout("%s %s epoch %u continuous %d\n", __func__
, ceph_sub_str
[sub
],
326 if (monc
->subs
[sub
].want
&&
327 monc
->subs
[sub
].item
.start
== start
&&
328 monc
->subs
[sub
].item
.flags
== flags
)
331 monc
->subs
[sub
].item
.start
= start
;
332 monc
->subs
[sub
].item
.flags
= flags
;
333 monc
->subs
[sub
].want
= true;
338 bool ceph_monc_want_map(struct ceph_mon_client
*monc
, int sub
, u32 epoch
,
343 mutex_lock(&monc
->mutex
);
344 need_request
= __ceph_monc_want_map(monc
, sub
, epoch
, continuous
);
345 mutex_unlock(&monc
->mutex
);
349 EXPORT_SYMBOL(ceph_monc_want_map
);
352 * Keep track of which maps we have
354 * @sub: one of CEPH_SUB_*
356 static void __ceph_monc_got_map(struct ceph_mon_client
*monc
, int sub
,
359 dout("%s %s epoch %u\n", __func__
, ceph_sub_str
[sub
], epoch
);
361 if (monc
->subs
[sub
].want
) {
362 if (monc
->subs
[sub
].item
.flags
& CEPH_SUBSCRIBE_ONETIME
)
363 monc
->subs
[sub
].want
= false;
365 monc
->subs
[sub
].item
.start
= cpu_to_le64(epoch
+ 1);
368 monc
->subs
[sub
].have
= epoch
;
371 void ceph_monc_got_map(struct ceph_mon_client
*monc
, int sub
, u32 epoch
)
373 mutex_lock(&monc
->mutex
);
374 __ceph_monc_got_map(monc
, sub
, epoch
);
375 mutex_unlock(&monc
->mutex
);
377 EXPORT_SYMBOL(ceph_monc_got_map
);
380 * Register interest in the next osdmap
382 void ceph_monc_request_next_osdmap(struct ceph_mon_client
*monc
)
384 dout("%s have %u\n", __func__
, monc
->subs
[CEPH_SUB_OSDMAP
].have
);
385 mutex_lock(&monc
->mutex
);
386 if (__ceph_monc_want_map(monc
, CEPH_SUB_OSDMAP
,
387 monc
->subs
[CEPH_SUB_OSDMAP
].have
+ 1, false))
388 __send_subscribe(monc
);
389 mutex_unlock(&monc
->mutex
);
391 EXPORT_SYMBOL(ceph_monc_request_next_osdmap
);
394 * Wait for an osdmap with a given epoch.
396 * @epoch: epoch to wait for
397 * @timeout: in jiffies, 0 means "wait forever"
399 int ceph_monc_wait_osdmap(struct ceph_mon_client
*monc
, u32 epoch
,
400 unsigned long timeout
)
402 unsigned long started
= jiffies
;
405 mutex_lock(&monc
->mutex
);
406 while (monc
->subs
[CEPH_SUB_OSDMAP
].have
< epoch
) {
407 mutex_unlock(&monc
->mutex
);
409 if (timeout
&& time_after_eq(jiffies
, started
+ timeout
))
412 ret
= wait_event_interruptible_timeout(monc
->client
->auth_wq
,
413 monc
->subs
[CEPH_SUB_OSDMAP
].have
>= epoch
,
414 ceph_timeout_jiffies(timeout
));
418 mutex_lock(&monc
->mutex
);
421 mutex_unlock(&monc
->mutex
);
424 EXPORT_SYMBOL(ceph_monc_wait_osdmap
);
427 * Open a session with a random monitor. Request monmap and osdmap,
428 * which are waited upon in __ceph_open_session().
430 int ceph_monc_open_session(struct ceph_mon_client
*monc
)
432 mutex_lock(&monc
->mutex
);
433 __ceph_monc_want_map(monc
, CEPH_SUB_MONMAP
, 0, true);
434 __ceph_monc_want_map(monc
, CEPH_SUB_OSDMAP
, 0, false);
435 __open_session(monc
);
436 __schedule_delayed(monc
);
437 mutex_unlock(&monc
->mutex
);
440 EXPORT_SYMBOL(ceph_monc_open_session
);
442 static void ceph_monc_handle_map(struct ceph_mon_client
*monc
,
443 struct ceph_msg
*msg
)
445 struct ceph_client
*client
= monc
->client
;
446 struct ceph_monmap
*monmap
= NULL
, *old
= monc
->monmap
;
449 mutex_lock(&monc
->mutex
);
451 dout("handle_monmap\n");
452 p
= msg
->front
.iov_base
;
453 end
= p
+ msg
->front
.iov_len
;
455 monmap
= ceph_monmap_decode(p
, end
);
456 if (IS_ERR(monmap
)) {
457 pr_err("problem decoding monmap, %d\n",
458 (int)PTR_ERR(monmap
));
462 if (ceph_check_fsid(monc
->client
, &monmap
->fsid
) < 0) {
467 client
->monc
.monmap
= monmap
;
470 __ceph_monc_got_map(monc
, CEPH_SUB_MONMAP
, monc
->monmap
->epoch
);
471 client
->have_fsid
= true;
474 mutex_unlock(&monc
->mutex
);
475 wake_up_all(&client
->auth_wq
);
479 * generic requests (currently statfs, mon_get_version)
481 static struct ceph_mon_generic_request
*__lookup_generic_req(
482 struct ceph_mon_client
*monc
, u64 tid
)
484 struct ceph_mon_generic_request
*req
;
485 struct rb_node
*n
= monc
->generic_request_tree
.rb_node
;
488 req
= rb_entry(n
, struct ceph_mon_generic_request
, node
);
491 else if (tid
> req
->tid
)
499 static void __insert_generic_request(struct ceph_mon_client
*monc
,
500 struct ceph_mon_generic_request
*new)
502 struct rb_node
**p
= &monc
->generic_request_tree
.rb_node
;
503 struct rb_node
*parent
= NULL
;
504 struct ceph_mon_generic_request
*req
= NULL
;
508 req
= rb_entry(parent
, struct ceph_mon_generic_request
, node
);
509 if (new->tid
< req
->tid
)
511 else if (new->tid
> req
->tid
)
517 rb_link_node(&new->node
, parent
, p
);
518 rb_insert_color(&new->node
, &monc
->generic_request_tree
);
521 static void release_generic_request(struct kref
*kref
)
523 struct ceph_mon_generic_request
*req
=
524 container_of(kref
, struct ceph_mon_generic_request
, kref
);
527 ceph_msg_put(req
->reply
);
529 ceph_msg_put(req
->request
);
534 static void put_generic_request(struct ceph_mon_generic_request
*req
)
536 kref_put(&req
->kref
, release_generic_request
);
539 static void get_generic_request(struct ceph_mon_generic_request
*req
)
541 kref_get(&req
->kref
);
544 static struct ceph_msg
*get_generic_reply(struct ceph_connection
*con
,
545 struct ceph_msg_header
*hdr
,
548 struct ceph_mon_client
*monc
= con
->private;
549 struct ceph_mon_generic_request
*req
;
550 u64 tid
= le64_to_cpu(hdr
->tid
);
553 mutex_lock(&monc
->mutex
);
554 req
= __lookup_generic_req(monc
, tid
);
556 dout("get_generic_reply %lld dne\n", tid
);
560 dout("get_generic_reply %lld got %p\n", tid
, req
->reply
);
562 m
= ceph_msg_get(req
->reply
);
564 * we don't need to track the connection reading into
565 * this reply because we only have one open connection
569 mutex_unlock(&monc
->mutex
);
573 static int __do_generic_request(struct ceph_mon_client
*monc
, u64 tid
,
574 struct ceph_mon_generic_request
*req
)
578 /* register request */
579 req
->tid
= tid
!= 0 ? tid
: ++monc
->last_tid
;
580 req
->request
->hdr
.tid
= cpu_to_le64(req
->tid
);
581 __insert_generic_request(monc
, req
);
582 ceph_con_send(&monc
->con
, ceph_msg_get(req
->request
));
583 mutex_unlock(&monc
->mutex
);
585 err
= wait_for_completion_interruptible(&req
->completion
);
587 mutex_lock(&monc
->mutex
);
588 rb_erase(&req
->node
, &monc
->generic_request_tree
);
595 static int do_generic_request(struct ceph_mon_client
*monc
,
596 struct ceph_mon_generic_request
*req
)
600 mutex_lock(&monc
->mutex
);
601 err
= __do_generic_request(monc
, 0, req
);
602 mutex_unlock(&monc
->mutex
);
610 static void handle_statfs_reply(struct ceph_mon_client
*monc
,
611 struct ceph_msg
*msg
)
613 struct ceph_mon_generic_request
*req
;
614 struct ceph_mon_statfs_reply
*reply
= msg
->front
.iov_base
;
615 u64 tid
= le64_to_cpu(msg
->hdr
.tid
);
617 if (msg
->front
.iov_len
!= sizeof(*reply
))
619 dout("handle_statfs_reply %p tid %llu\n", msg
, tid
);
621 mutex_lock(&monc
->mutex
);
622 req
= __lookup_generic_req(monc
, tid
);
624 *(struct ceph_statfs
*)req
->buf
= reply
->st
;
626 get_generic_request(req
);
628 mutex_unlock(&monc
->mutex
);
630 complete_all(&req
->completion
);
631 put_generic_request(req
);
636 pr_err("corrupt statfs reply, tid %llu\n", tid
);
641 * Do a synchronous statfs().
643 int ceph_monc_do_statfs(struct ceph_mon_client
*monc
, struct ceph_statfs
*buf
)
645 struct ceph_mon_generic_request
*req
;
646 struct ceph_mon_statfs
*h
;
649 req
= kzalloc(sizeof(*req
), GFP_NOFS
);
653 kref_init(&req
->kref
);
655 init_completion(&req
->completion
);
658 req
->request
= ceph_msg_new(CEPH_MSG_STATFS
, sizeof(*h
), GFP_NOFS
,
662 req
->reply
= ceph_msg_new(CEPH_MSG_STATFS_REPLY
, 1024, GFP_NOFS
,
667 /* fill out request */
668 h
= req
->request
->front
.iov_base
;
669 h
->monhdr
.have_version
= 0;
670 h
->monhdr
.session_mon
= cpu_to_le16(-1);
671 h
->monhdr
.session_mon_tid
= 0;
672 h
->fsid
= monc
->monmap
->fsid
;
674 err
= do_generic_request(monc
, req
);
677 put_generic_request(req
);
680 EXPORT_SYMBOL(ceph_monc_do_statfs
);
682 static void handle_get_version_reply(struct ceph_mon_client
*monc
,
683 struct ceph_msg
*msg
)
685 struct ceph_mon_generic_request
*req
;
686 u64 tid
= le64_to_cpu(msg
->hdr
.tid
);
687 void *p
= msg
->front
.iov_base
;
688 void *end
= p
+ msg
->front_alloc_len
;
691 dout("%s %p tid %llu\n", __func__
, msg
, tid
);
693 ceph_decode_need(&p
, end
, 2*sizeof(u64
), bad
);
694 handle
= ceph_decode_64(&p
);
695 if (tid
!= 0 && tid
!= handle
)
698 mutex_lock(&monc
->mutex
);
699 req
= __lookup_generic_req(monc
, handle
);
701 *(u64
*)req
->buf
= ceph_decode_64(&p
);
703 get_generic_request(req
);
705 mutex_unlock(&monc
->mutex
);
707 complete_all(&req
->completion
);
708 put_generic_request(req
);
713 pr_err("corrupt mon_get_version reply, tid %llu\n", tid
);
718 * Send MMonGetVersion and wait for the reply.
720 * @what: one of "mdsmap", "osdmap" or "monmap"
722 int ceph_monc_do_get_version(struct ceph_mon_client
*monc
, const char *what
,
725 struct ceph_mon_generic_request
*req
;
730 req
= kzalloc(sizeof(*req
), GFP_NOFS
);
734 kref_init(&req
->kref
);
736 init_completion(&req
->completion
);
738 req
->request
= ceph_msg_new(CEPH_MSG_MON_GET_VERSION
,
739 sizeof(u64
) + sizeof(u32
) + strlen(what
),
746 req
->reply
= ceph_msg_new(CEPH_MSG_MON_GET_VERSION_REPLY
, 1024,
753 p
= req
->request
->front
.iov_base
;
754 end
= p
+ req
->request
->front_alloc_len
;
756 /* fill out request */
757 mutex_lock(&monc
->mutex
);
758 tid
= ++monc
->last_tid
;
759 ceph_encode_64(&p
, tid
); /* handle */
760 ceph_encode_string(&p
, end
, what
, strlen(what
));
762 err
= __do_generic_request(monc
, tid
, req
);
764 mutex_unlock(&monc
->mutex
);
766 put_generic_request(req
);
769 EXPORT_SYMBOL(ceph_monc_do_get_version
);
772 * Resend pending generic requests.
774 static void __resend_generic_request(struct ceph_mon_client
*monc
)
776 struct ceph_mon_generic_request
*req
;
779 for (p
= rb_first(&monc
->generic_request_tree
); p
; p
= rb_next(p
)) {
780 req
= rb_entry(p
, struct ceph_mon_generic_request
, node
);
781 ceph_msg_revoke(req
->request
);
782 ceph_msg_revoke_incoming(req
->reply
);
783 ceph_con_send(&monc
->con
, ceph_msg_get(req
->request
));
788 * Delayed work. If we haven't mounted yet, retry. Otherwise,
789 * renew/retry subscription as needed (in case it is timing out, or we
790 * got an ENOMEM). And keep the monitor connection alive.
792 static void delayed_work(struct work_struct
*work
)
794 struct ceph_mon_client
*monc
=
795 container_of(work
, struct ceph_mon_client
, delayed_work
.work
);
797 dout("monc delayed_work\n");
798 mutex_lock(&monc
->mutex
);
800 dout("%s continuing hunt\n", __func__
);
801 reopen_session(monc
);
803 int is_auth
= ceph_auth_is_authenticated(monc
->auth
);
804 if (ceph_con_keepalive_expired(&monc
->con
,
805 CEPH_MONC_PING_TIMEOUT
)) {
806 dout("monc keepalive timeout\n");
808 reopen_session(monc
);
811 if (!monc
->hunting
) {
812 ceph_con_keepalive(&monc
->con
);
813 __validate_auth(monc
);
817 unsigned long now
= jiffies
;
819 dout("%s renew subs? now %lu renew after %lu\n",
820 __func__
, now
, monc
->sub_renew_after
);
821 if (time_after_eq(now
, monc
->sub_renew_after
))
822 __send_subscribe(monc
);
825 __schedule_delayed(monc
);
826 mutex_unlock(&monc
->mutex
);
830 * On startup, we build a temporary monmap populated with the IPs
831 * provided by mount(2).
833 static int build_initial_monmap(struct ceph_mon_client
*monc
)
835 struct ceph_options
*opt
= monc
->client
->options
;
836 struct ceph_entity_addr
*mon_addr
= opt
->mon_addr
;
837 int num_mon
= opt
->num_mon
;
840 /* build initial monmap */
841 monc
->monmap
= kzalloc(sizeof(*monc
->monmap
) +
842 num_mon
*sizeof(monc
->monmap
->mon_inst
[0]),
846 for (i
= 0; i
< num_mon
; i
++) {
847 monc
->monmap
->mon_inst
[i
].addr
= mon_addr
[i
];
848 monc
->monmap
->mon_inst
[i
].addr
.nonce
= 0;
849 monc
->monmap
->mon_inst
[i
].name
.type
=
850 CEPH_ENTITY_TYPE_MON
;
851 monc
->monmap
->mon_inst
[i
].name
.num
= cpu_to_le64(i
);
853 monc
->monmap
->num_mon
= num_mon
;
857 int ceph_monc_init(struct ceph_mon_client
*monc
, struct ceph_client
*cl
)
862 memset(monc
, 0, sizeof(*monc
));
865 mutex_init(&monc
->mutex
);
867 err
= build_initial_monmap(monc
);
873 monc
->auth
= ceph_auth_init(cl
->options
->name
,
875 if (IS_ERR(monc
->auth
)) {
876 err
= PTR_ERR(monc
->auth
);
879 monc
->auth
->want_keys
=
880 CEPH_ENTITY_TYPE_AUTH
| CEPH_ENTITY_TYPE_MON
|
881 CEPH_ENTITY_TYPE_OSD
| CEPH_ENTITY_TYPE_MDS
;
885 monc
->m_subscribe_ack
= ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE_ACK
,
886 sizeof(struct ceph_mon_subscribe_ack
),
888 if (!monc
->m_subscribe_ack
)
891 monc
->m_subscribe
= ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE
, 96, GFP_NOFS
,
893 if (!monc
->m_subscribe
)
894 goto out_subscribe_ack
;
896 monc
->m_auth_reply
= ceph_msg_new(CEPH_MSG_AUTH_REPLY
, 4096, GFP_NOFS
,
898 if (!monc
->m_auth_reply
)
901 monc
->m_auth
= ceph_msg_new(CEPH_MSG_AUTH
, 4096, GFP_NOFS
, true);
902 monc
->pending_auth
= 0;
906 ceph_con_init(&monc
->con
, monc
, &mon_con_ops
,
907 &monc
->client
->msgr
);
910 monc
->had_a_connection
= false;
913 INIT_DELAYED_WORK(&monc
->delayed_work
, delayed_work
);
914 monc
->generic_request_tree
= RB_ROOT
;
920 ceph_msg_put(monc
->m_auth_reply
);
922 ceph_msg_put(monc
->m_subscribe
);
924 ceph_msg_put(monc
->m_subscribe_ack
);
926 ceph_auth_destroy(monc
->auth
);
932 EXPORT_SYMBOL(ceph_monc_init
);
934 void ceph_monc_stop(struct ceph_mon_client
*monc
)
937 cancel_delayed_work_sync(&monc
->delayed_work
);
939 mutex_lock(&monc
->mutex
);
940 __close_session(monc
);
942 mutex_unlock(&monc
->mutex
);
945 * flush msgr queue before we destroy ourselves to ensure that:
946 * - any work that references our embedded con is finished.
947 * - any osd_client or other work that may reference an authorizer
948 * finishes before we shut down the auth subsystem.
952 ceph_auth_destroy(monc
->auth
);
954 ceph_msg_put(monc
->m_auth
);
955 ceph_msg_put(monc
->m_auth_reply
);
956 ceph_msg_put(monc
->m_subscribe
);
957 ceph_msg_put(monc
->m_subscribe_ack
);
961 EXPORT_SYMBOL(ceph_monc_stop
);
963 static void finish_hunting(struct ceph_mon_client
*monc
)
966 dout("%s found mon%d\n", __func__
, monc
->cur_mon
);
967 monc
->hunting
= false;
968 monc
->had_a_connection
= true;
969 monc
->hunt_mult
/= 2; /* reduce by 50% */
970 if (monc
->hunt_mult
< 1)
975 static void handle_auth_reply(struct ceph_mon_client
*monc
,
976 struct ceph_msg
*msg
)
981 mutex_lock(&monc
->mutex
);
982 was_auth
= ceph_auth_is_authenticated(monc
->auth
);
983 monc
->pending_auth
= 0;
984 ret
= ceph_handle_auth_reply(monc
->auth
, msg
->front
.iov_base
,
986 monc
->m_auth
->front
.iov_base
,
987 monc
->m_auth
->front_alloc_len
);
989 __send_prepared_auth_request(monc
, ret
);
993 finish_hunting(monc
);
996 monc
->client
->auth_err
= ret
;
997 } else if (!was_auth
&& ceph_auth_is_authenticated(monc
->auth
)) {
998 dout("authenticated, starting session\n");
1000 monc
->client
->msgr
.inst
.name
.type
= CEPH_ENTITY_TYPE_CLIENT
;
1001 monc
->client
->msgr
.inst
.name
.num
=
1002 cpu_to_le64(monc
->auth
->global_id
);
1004 __send_subscribe(monc
);
1005 __resend_generic_request(monc
);
1007 pr_info("mon%d %s session established\n", monc
->cur_mon
,
1008 ceph_pr_addr(&monc
->con
.peer_addr
.in_addr
));
1012 mutex_unlock(&monc
->mutex
);
1013 if (monc
->client
->auth_err
< 0)
1014 wake_up_all(&monc
->client
->auth_wq
);
1017 static int __validate_auth(struct ceph_mon_client
*monc
)
1021 if (monc
->pending_auth
)
1024 ret
= ceph_build_auth(monc
->auth
, monc
->m_auth
->front
.iov_base
,
1025 monc
->m_auth
->front_alloc_len
);
1027 return ret
; /* either an error, or no need to authenticate */
1028 __send_prepared_auth_request(monc
, ret
);
1032 int ceph_monc_validate_auth(struct ceph_mon_client
*monc
)
1036 mutex_lock(&monc
->mutex
);
1037 ret
= __validate_auth(monc
);
1038 mutex_unlock(&monc
->mutex
);
1041 EXPORT_SYMBOL(ceph_monc_validate_auth
);
1044 * handle incoming message
1046 static void dispatch(struct ceph_connection
*con
, struct ceph_msg
*msg
)
1048 struct ceph_mon_client
*monc
= con
->private;
1049 int type
= le16_to_cpu(msg
->hdr
.type
);
1055 case CEPH_MSG_AUTH_REPLY
:
1056 handle_auth_reply(monc
, msg
);
1059 case CEPH_MSG_MON_SUBSCRIBE_ACK
:
1060 handle_subscribe_ack(monc
, msg
);
1063 case CEPH_MSG_STATFS_REPLY
:
1064 handle_statfs_reply(monc
, msg
);
1067 case CEPH_MSG_MON_GET_VERSION_REPLY
:
1068 handle_get_version_reply(monc
, msg
);
1071 case CEPH_MSG_MON_MAP
:
1072 ceph_monc_handle_map(monc
, msg
);
1075 case CEPH_MSG_OSD_MAP
:
1076 ceph_osdc_handle_map(&monc
->client
->osdc
, msg
);
1080 /* can the chained handler handle it? */
1081 if (monc
->client
->extra_mon_dispatch
&&
1082 monc
->client
->extra_mon_dispatch(monc
->client
, msg
) == 0)
1085 pr_err("received unknown message type %d %s\n", type
,
1086 ceph_msg_type_name(type
));
1092 * Allocate memory for incoming message
1094 static struct ceph_msg
*mon_alloc_msg(struct ceph_connection
*con
,
1095 struct ceph_msg_header
*hdr
,
1098 struct ceph_mon_client
*monc
= con
->private;
1099 int type
= le16_to_cpu(hdr
->type
);
1100 int front_len
= le32_to_cpu(hdr
->front_len
);
1101 struct ceph_msg
*m
= NULL
;
1106 case CEPH_MSG_MON_SUBSCRIBE_ACK
:
1107 m
= ceph_msg_get(monc
->m_subscribe_ack
);
1109 case CEPH_MSG_STATFS_REPLY
:
1110 return get_generic_reply(con
, hdr
, skip
);
1111 case CEPH_MSG_AUTH_REPLY
:
1112 m
= ceph_msg_get(monc
->m_auth_reply
);
1114 case CEPH_MSG_MON_GET_VERSION_REPLY
:
1115 if (le64_to_cpu(hdr
->tid
) != 0)
1116 return get_generic_reply(con
, hdr
, skip
);
1119 * Older OSDs don't set reply tid even if the orignal
1120 * request had a non-zero tid. Workaround this weirdness
1121 * by falling through to the allocate case.
1123 case CEPH_MSG_MON_MAP
:
1124 case CEPH_MSG_MDS_MAP
:
1125 case CEPH_MSG_OSD_MAP
:
1126 m
= ceph_msg_new(type
, front_len
, GFP_NOFS
, false);
1128 return NULL
; /* ENOMEM--return skip == 0 */
1133 pr_info("alloc_msg unknown type %d\n", type
);
1135 } else if (front_len
> m
->front_alloc_len
) {
1136 pr_warn("mon_alloc_msg front %d > prealloc %d (%u#%llu)\n",
1137 front_len
, m
->front_alloc_len
,
1138 (unsigned int)con
->peer_name
.type
,
1139 le64_to_cpu(con
->peer_name
.num
));
1141 m
= ceph_msg_new(type
, front_len
, GFP_NOFS
, false);
1148 * If the monitor connection resets, pick a new monitor and resubmit
1149 * any pending requests.
1151 static void mon_fault(struct ceph_connection
*con
)
1153 struct ceph_mon_client
*monc
= con
->private;
1155 mutex_lock(&monc
->mutex
);
1156 dout("%s mon%d\n", __func__
, monc
->cur_mon
);
1157 if (monc
->cur_mon
>= 0) {
1158 if (!monc
->hunting
) {
1159 dout("%s hunting for new mon\n", __func__
);
1160 reopen_session(monc
);
1161 __schedule_delayed(monc
);
1163 dout("%s already hunting\n", __func__
);
1166 mutex_unlock(&monc
->mutex
);
1170 * We can ignore refcounting on the connection struct, as all references
1171 * will come from the messenger workqueue, which is drained prior to
1172 * mon_client destruction.
1174 static struct ceph_connection
*con_get(struct ceph_connection
*con
)
1179 static void con_put(struct ceph_connection
*con
)
1183 static const struct ceph_connection_operations mon_con_ops
= {
1186 .dispatch
= dispatch
,
1188 .alloc_msg
= mon_alloc_msg
,