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
;
48 ceph_decode_32_safe(&p
, end
, len
, bad
);
49 ceph_decode_need(&p
, end
, len
, bad
);
51 dout("monmap_decode %p %p len %d\n", p
, end
, (int)(end
-p
));
52 p
+= sizeof(u16
); /* skip version */
54 ceph_decode_need(&p
, end
, sizeof(fsid
) + 2*sizeof(u32
), bad
);
55 ceph_decode_copy(&p
, &fsid
, sizeof(fsid
));
56 epoch
= ceph_decode_32(&p
);
58 num_mon
= ceph_decode_32(&p
);
59 ceph_decode_need(&p
, end
, num_mon
*sizeof(m
->mon_inst
[0]), bad
);
61 if (num_mon
>= CEPH_MAX_MON
)
63 m
= kmalloc(sizeof(*m
) + sizeof(m
->mon_inst
[0])*num_mon
, GFP_NOFS
);
65 return ERR_PTR(-ENOMEM
);
69 ceph_decode_copy(&p
, m
->mon_inst
, num_mon
*sizeof(m
->mon_inst
[0]));
70 for (i
= 0; i
< num_mon
; i
++)
71 ceph_decode_addr(&m
->mon_inst
[i
].addr
);
73 dout("monmap_decode epoch %d, num_mon %d\n", m
->epoch
,
75 for (i
= 0; i
< m
->num_mon
; i
++)
76 dout("monmap_decode mon%d is %s\n", i
,
77 ceph_pr_addr(&m
->mon_inst
[i
].addr
.in_addr
));
81 dout("monmap_decode failed with %d\n", err
);
87 * return true if *addr is included in the monmap.
89 int ceph_monmap_contains(struct ceph_monmap
*m
, struct ceph_entity_addr
*addr
)
93 for (i
= 0; i
< m
->num_mon
; i
++)
94 if (memcmp(addr
, &m
->mon_inst
[i
].addr
, sizeof(*addr
)) == 0)
100 * Send an auth request.
102 static void __send_prepared_auth_request(struct ceph_mon_client
*monc
, int len
)
104 monc
->pending_auth
= 1;
105 monc
->m_auth
->front
.iov_len
= len
;
106 monc
->m_auth
->hdr
.front_len
= cpu_to_le32(len
);
107 ceph_msg_revoke(monc
->m_auth
);
108 ceph_msg_get(monc
->m_auth
); /* keep our ref */
109 ceph_con_send(&monc
->con
, monc
->m_auth
);
113 * Close monitor session, if any.
115 static void __close_session(struct ceph_mon_client
*monc
)
117 dout("__close_session closing mon%d\n", monc
->cur_mon
);
118 ceph_msg_revoke(monc
->m_auth
);
119 ceph_msg_revoke_incoming(monc
->m_auth_reply
);
120 ceph_msg_revoke(monc
->m_subscribe
);
121 ceph_msg_revoke_incoming(monc
->m_subscribe_ack
);
122 ceph_con_close(&monc
->con
);
124 monc
->pending_auth
= 0;
125 ceph_auth_reset(monc
->auth
);
129 * Pick a new monitor at random and set cur_mon. If we are repicking
130 * (i.e. cur_mon is already set), be sure to pick a different one.
132 static void pick_new_mon(struct ceph_mon_client
*monc
)
134 int old_mon
= monc
->cur_mon
;
136 BUG_ON(monc
->monmap
->num_mon
< 1);
138 if (monc
->monmap
->num_mon
== 1) {
141 int max
= monc
->monmap
->num_mon
;
145 if (monc
->cur_mon
>= 0) {
146 if (monc
->cur_mon
< monc
->monmap
->num_mon
)
152 n
= prandom_u32() % max
;
153 if (o
>= 0 && n
>= o
)
159 dout("%s mon%d -> mon%d out of %d mons\n", __func__
, old_mon
,
160 monc
->cur_mon
, monc
->monmap
->num_mon
);
164 * Open a session with a new monitor.
166 static void __open_session(struct ceph_mon_client
*monc
)
172 monc
->hunting
= true;
173 if (monc
->had_a_connection
) {
174 monc
->hunt_mult
*= CEPH_MONC_HUNT_BACKOFF
;
175 if (monc
->hunt_mult
> CEPH_MONC_HUNT_MAX_MULT
)
176 monc
->hunt_mult
= CEPH_MONC_HUNT_MAX_MULT
;
179 monc
->sub_renew_after
= jiffies
; /* i.e., expired */
180 monc
->sub_renew_sent
= 0;
182 dout("%s opening mon%d\n", __func__
, monc
->cur_mon
);
183 ceph_con_open(&monc
->con
, CEPH_ENTITY_TYPE_MON
, monc
->cur_mon
,
184 &monc
->monmap
->mon_inst
[monc
->cur_mon
].addr
);
187 * send an initial keepalive to ensure our timestamp is valid
188 * by the time we are in an OPENED state
190 ceph_con_keepalive(&monc
->con
);
192 /* initiate authentication handshake */
193 ret
= ceph_auth_build_hello(monc
->auth
,
194 monc
->m_auth
->front
.iov_base
,
195 monc
->m_auth
->front_alloc_len
);
197 __send_prepared_auth_request(monc
, ret
);
200 static void reopen_session(struct ceph_mon_client
*monc
)
203 pr_info("mon%d %s session lost, hunting for new mon\n",
204 monc
->cur_mon
, ceph_pr_addr(&monc
->con
.peer_addr
.in_addr
));
206 __close_session(monc
);
207 __open_session(monc
);
211 * Reschedule delayed work timer.
213 static void __schedule_delayed(struct ceph_mon_client
*monc
)
218 delay
= CEPH_MONC_HUNT_INTERVAL
* monc
->hunt_mult
;
220 delay
= CEPH_MONC_PING_INTERVAL
;
222 dout("__schedule_delayed after %lu\n", delay
);
223 mod_delayed_work(system_wq
, &monc
->delayed_work
,
224 round_jiffies_relative(delay
));
227 const char *ceph_sub_str
[] = {
228 [CEPH_SUB_MONMAP
] = "monmap",
229 [CEPH_SUB_OSDMAP
] = "osdmap",
230 [CEPH_SUB_FSMAP
] = "fsmap.user",
231 [CEPH_SUB_MDSMAP
] = "mdsmap",
235 * Send subscribe request for one or more maps, according to
238 static void __send_subscribe(struct ceph_mon_client
*monc
)
240 struct ceph_msg
*msg
= monc
->m_subscribe
;
241 void *p
= msg
->front
.iov_base
;
242 void *const end
= p
+ msg
->front_alloc_len
;
246 dout("%s sent %lu\n", __func__
, monc
->sub_renew_sent
);
248 BUG_ON(monc
->cur_mon
< 0);
250 if (!monc
->sub_renew_sent
)
251 monc
->sub_renew_sent
= jiffies
| 1; /* never 0 */
253 msg
->hdr
.version
= cpu_to_le16(2);
255 for (i
= 0; i
< ARRAY_SIZE(monc
->subs
); i
++) {
256 if (monc
->subs
[i
].want
)
259 BUG_ON(num
< 1); /* monmap sub is always there */
260 ceph_encode_32(&p
, num
);
261 for (i
= 0; i
< ARRAY_SIZE(monc
->subs
); i
++) {
265 if (!monc
->subs
[i
].want
)
268 len
= sprintf(buf
, "%s", ceph_sub_str
[i
]);
269 if (i
== CEPH_SUB_MDSMAP
&&
270 monc
->fs_cluster_id
!= CEPH_FS_CLUSTER_ID_NONE
)
271 len
+= sprintf(buf
+ len
, ".%d", monc
->fs_cluster_id
);
273 dout("%s %s start %llu flags 0x%x\n", __func__
, buf
,
274 le64_to_cpu(monc
->subs
[i
].item
.start
),
275 monc
->subs
[i
].item
.flags
);
276 ceph_encode_string(&p
, end
, buf
, len
);
277 memcpy(p
, &monc
->subs
[i
].item
, sizeof(monc
->subs
[i
].item
));
278 p
+= sizeof(monc
->subs
[i
].item
);
282 msg
->front
.iov_len
= p
- msg
->front
.iov_base
;
283 msg
->hdr
.front_len
= cpu_to_le32(msg
->front
.iov_len
);
284 ceph_msg_revoke(msg
);
285 ceph_con_send(&monc
->con
, ceph_msg_get(msg
));
288 static void handle_subscribe_ack(struct ceph_mon_client
*monc
,
289 struct ceph_msg
*msg
)
291 unsigned int seconds
;
292 struct ceph_mon_subscribe_ack
*h
= msg
->front
.iov_base
;
294 if (msg
->front
.iov_len
< sizeof(*h
))
296 seconds
= le32_to_cpu(h
->duration
);
298 mutex_lock(&monc
->mutex
);
299 if (monc
->sub_renew_sent
) {
300 monc
->sub_renew_after
= monc
->sub_renew_sent
+
301 (seconds
>> 1) * HZ
- 1;
302 dout("%s sent %lu duration %d renew after %lu\n", __func__
,
303 monc
->sub_renew_sent
, seconds
, monc
->sub_renew_after
);
304 monc
->sub_renew_sent
= 0;
306 dout("%s sent %lu renew after %lu, ignoring\n", __func__
,
307 monc
->sub_renew_sent
, monc
->sub_renew_after
);
309 mutex_unlock(&monc
->mutex
);
312 pr_err("got corrupt subscribe-ack msg\n");
317 * Register interest in a map
319 * @sub: one of CEPH_SUB_*
320 * @epoch: X for "every map since X", or 0 for "just the latest"
322 static bool __ceph_monc_want_map(struct ceph_mon_client
*monc
, int sub
,
323 u32 epoch
, bool continuous
)
325 __le64 start
= cpu_to_le64(epoch
);
326 u8 flags
= !continuous
? CEPH_SUBSCRIBE_ONETIME
: 0;
328 dout("%s %s epoch %u continuous %d\n", __func__
, ceph_sub_str
[sub
],
331 if (monc
->subs
[sub
].want
&&
332 monc
->subs
[sub
].item
.start
== start
&&
333 monc
->subs
[sub
].item
.flags
== flags
)
336 monc
->subs
[sub
].item
.start
= start
;
337 monc
->subs
[sub
].item
.flags
= flags
;
338 monc
->subs
[sub
].want
= true;
343 bool ceph_monc_want_map(struct ceph_mon_client
*monc
, int sub
, u32 epoch
,
348 mutex_lock(&monc
->mutex
);
349 need_request
= __ceph_monc_want_map(monc
, sub
, epoch
, continuous
);
350 mutex_unlock(&monc
->mutex
);
354 EXPORT_SYMBOL(ceph_monc_want_map
);
357 * Keep track of which maps we have
359 * @sub: one of CEPH_SUB_*
361 static void __ceph_monc_got_map(struct ceph_mon_client
*monc
, int sub
,
364 dout("%s %s epoch %u\n", __func__
, ceph_sub_str
[sub
], epoch
);
366 if (monc
->subs
[sub
].want
) {
367 if (monc
->subs
[sub
].item
.flags
& CEPH_SUBSCRIBE_ONETIME
)
368 monc
->subs
[sub
].want
= false;
370 monc
->subs
[sub
].item
.start
= cpu_to_le64(epoch
+ 1);
373 monc
->subs
[sub
].have
= epoch
;
376 void ceph_monc_got_map(struct ceph_mon_client
*monc
, int sub
, u32 epoch
)
378 mutex_lock(&monc
->mutex
);
379 __ceph_monc_got_map(monc
, sub
, epoch
);
380 mutex_unlock(&monc
->mutex
);
382 EXPORT_SYMBOL(ceph_monc_got_map
);
384 void ceph_monc_renew_subs(struct ceph_mon_client
*monc
)
386 mutex_lock(&monc
->mutex
);
387 __send_subscribe(monc
);
388 mutex_unlock(&monc
->mutex
);
390 EXPORT_SYMBOL(ceph_monc_renew_subs
);
393 * Wait for an osdmap with a given epoch.
395 * @epoch: epoch to wait for
396 * @timeout: in jiffies, 0 means "wait forever"
398 int ceph_monc_wait_osdmap(struct ceph_mon_client
*monc
, u32 epoch
,
399 unsigned long timeout
)
401 unsigned long started
= jiffies
;
404 mutex_lock(&monc
->mutex
);
405 while (monc
->subs
[CEPH_SUB_OSDMAP
].have
< epoch
) {
406 mutex_unlock(&monc
->mutex
);
408 if (timeout
&& time_after_eq(jiffies
, started
+ timeout
))
411 ret
= wait_event_interruptible_timeout(monc
->client
->auth_wq
,
412 monc
->subs
[CEPH_SUB_OSDMAP
].have
>= epoch
,
413 ceph_timeout_jiffies(timeout
));
417 mutex_lock(&monc
->mutex
);
420 mutex_unlock(&monc
->mutex
);
423 EXPORT_SYMBOL(ceph_monc_wait_osdmap
);
426 * Open a session with a random monitor. Request monmap and osdmap,
427 * which are waited upon in __ceph_open_session().
429 int ceph_monc_open_session(struct ceph_mon_client
*monc
)
431 mutex_lock(&monc
->mutex
);
432 __ceph_monc_want_map(monc
, CEPH_SUB_MONMAP
, 0, true);
433 __ceph_monc_want_map(monc
, CEPH_SUB_OSDMAP
, 0, false);
434 __open_session(monc
);
435 __schedule_delayed(monc
);
436 mutex_unlock(&monc
->mutex
);
439 EXPORT_SYMBOL(ceph_monc_open_session
);
441 static void ceph_monc_handle_map(struct ceph_mon_client
*monc
,
442 struct ceph_msg
*msg
)
444 struct ceph_client
*client
= monc
->client
;
445 struct ceph_monmap
*monmap
= NULL
, *old
= monc
->monmap
;
448 mutex_lock(&monc
->mutex
);
450 dout("handle_monmap\n");
451 p
= msg
->front
.iov_base
;
452 end
= p
+ msg
->front
.iov_len
;
454 monmap
= ceph_monmap_decode(p
, end
);
455 if (IS_ERR(monmap
)) {
456 pr_err("problem decoding monmap, %d\n",
457 (int)PTR_ERR(monmap
));
461 if (ceph_check_fsid(monc
->client
, &monmap
->fsid
) < 0) {
466 client
->monc
.monmap
= monmap
;
469 __ceph_monc_got_map(monc
, CEPH_SUB_MONMAP
, monc
->monmap
->epoch
);
470 client
->have_fsid
= true;
473 mutex_unlock(&monc
->mutex
);
474 wake_up_all(&client
->auth_wq
);
478 * generic requests (currently statfs, mon_get_version)
480 DEFINE_RB_FUNCS(generic_request
, struct ceph_mon_generic_request
, tid
, node
)
482 static void release_generic_request(struct kref
*kref
)
484 struct ceph_mon_generic_request
*req
=
485 container_of(kref
, struct ceph_mon_generic_request
, kref
);
487 dout("%s greq %p request %p reply %p\n", __func__
, req
, req
->request
,
489 WARN_ON(!RB_EMPTY_NODE(&req
->node
));
492 ceph_msg_put(req
->reply
);
494 ceph_msg_put(req
->request
);
499 static void put_generic_request(struct ceph_mon_generic_request
*req
)
502 kref_put(&req
->kref
, release_generic_request
);
505 static void get_generic_request(struct ceph_mon_generic_request
*req
)
507 kref_get(&req
->kref
);
510 static struct ceph_mon_generic_request
*
511 alloc_generic_request(struct ceph_mon_client
*monc
, gfp_t gfp
)
513 struct ceph_mon_generic_request
*req
;
515 req
= kzalloc(sizeof(*req
), gfp
);
520 kref_init(&req
->kref
);
521 RB_CLEAR_NODE(&req
->node
);
522 init_completion(&req
->completion
);
524 dout("%s greq %p\n", __func__
, req
);
528 static void register_generic_request(struct ceph_mon_generic_request
*req
)
530 struct ceph_mon_client
*monc
= req
->monc
;
534 get_generic_request(req
);
535 req
->tid
= ++monc
->last_tid
;
536 insert_generic_request(&monc
->generic_request_tree
, req
);
539 static void send_generic_request(struct ceph_mon_client
*monc
,
540 struct ceph_mon_generic_request
*req
)
544 dout("%s greq %p tid %llu\n", __func__
, req
, req
->tid
);
545 req
->request
->hdr
.tid
= cpu_to_le64(req
->tid
);
546 ceph_con_send(&monc
->con
, ceph_msg_get(req
->request
));
549 static void __finish_generic_request(struct ceph_mon_generic_request
*req
)
551 struct ceph_mon_client
*monc
= req
->monc
;
553 dout("%s greq %p tid %llu\n", __func__
, req
, req
->tid
);
554 erase_generic_request(&monc
->generic_request_tree
, req
);
556 ceph_msg_revoke(req
->request
);
557 ceph_msg_revoke_incoming(req
->reply
);
560 static void finish_generic_request(struct ceph_mon_generic_request
*req
)
562 __finish_generic_request(req
);
563 put_generic_request(req
);
566 static void complete_generic_request(struct ceph_mon_generic_request
*req
)
568 if (req
->complete_cb
)
569 req
->complete_cb(req
);
571 complete_all(&req
->completion
);
572 put_generic_request(req
);
575 static void cancel_generic_request(struct ceph_mon_generic_request
*req
)
577 struct ceph_mon_client
*monc
= req
->monc
;
578 struct ceph_mon_generic_request
*lookup_req
;
580 dout("%s greq %p tid %llu\n", __func__
, req
, req
->tid
);
582 mutex_lock(&monc
->mutex
);
583 lookup_req
= lookup_generic_request(&monc
->generic_request_tree
,
586 WARN_ON(lookup_req
!= req
);
587 finish_generic_request(req
);
590 mutex_unlock(&monc
->mutex
);
593 static int wait_generic_request(struct ceph_mon_generic_request
*req
)
597 dout("%s greq %p tid %llu\n", __func__
, req
, req
->tid
);
598 ret
= wait_for_completion_interruptible(&req
->completion
);
600 cancel_generic_request(req
);
602 ret
= req
->result
; /* completed */
607 static struct ceph_msg
*get_generic_reply(struct ceph_connection
*con
,
608 struct ceph_msg_header
*hdr
,
611 struct ceph_mon_client
*monc
= con
->private;
612 struct ceph_mon_generic_request
*req
;
613 u64 tid
= le64_to_cpu(hdr
->tid
);
616 mutex_lock(&monc
->mutex
);
617 req
= lookup_generic_request(&monc
->generic_request_tree
, tid
);
619 dout("get_generic_reply %lld dne\n", tid
);
623 dout("get_generic_reply %lld got %p\n", tid
, req
->reply
);
625 m
= ceph_msg_get(req
->reply
);
627 * we don't need to track the connection reading into
628 * this reply because we only have one open connection
632 mutex_unlock(&monc
->mutex
);
639 static void handle_statfs_reply(struct ceph_mon_client
*monc
,
640 struct ceph_msg
*msg
)
642 struct ceph_mon_generic_request
*req
;
643 struct ceph_mon_statfs_reply
*reply
= msg
->front
.iov_base
;
644 u64 tid
= le64_to_cpu(msg
->hdr
.tid
);
646 dout("%s msg %p tid %llu\n", __func__
, msg
, tid
);
648 if (msg
->front
.iov_len
!= sizeof(*reply
))
651 mutex_lock(&monc
->mutex
);
652 req
= lookup_generic_request(&monc
->generic_request_tree
, tid
);
654 mutex_unlock(&monc
->mutex
);
659 *req
->u
.st
= reply
->st
; /* struct */
660 __finish_generic_request(req
);
661 mutex_unlock(&monc
->mutex
);
663 complete_generic_request(req
);
667 pr_err("corrupt statfs reply, tid %llu\n", tid
);
672 * Do a synchronous statfs().
674 int ceph_monc_do_statfs(struct ceph_mon_client
*monc
, struct ceph_statfs
*buf
)
676 struct ceph_mon_generic_request
*req
;
677 struct ceph_mon_statfs
*h
;
680 req
= alloc_generic_request(monc
, GFP_NOFS
);
684 req
->request
= ceph_msg_new(CEPH_MSG_STATFS
, sizeof(*h
), GFP_NOFS
,
689 req
->reply
= ceph_msg_new(CEPH_MSG_STATFS_REPLY
, 64, GFP_NOFS
, true);
695 mutex_lock(&monc
->mutex
);
696 register_generic_request(req
);
697 /* fill out request */
698 h
= req
->request
->front
.iov_base
;
699 h
->monhdr
.have_version
= 0;
700 h
->monhdr
.session_mon
= cpu_to_le16(-1);
701 h
->monhdr
.session_mon_tid
= 0;
702 h
->fsid
= monc
->monmap
->fsid
;
703 send_generic_request(monc
, req
);
704 mutex_unlock(&monc
->mutex
);
706 ret
= wait_generic_request(req
);
708 put_generic_request(req
);
711 EXPORT_SYMBOL(ceph_monc_do_statfs
);
713 static void handle_get_version_reply(struct ceph_mon_client
*monc
,
714 struct ceph_msg
*msg
)
716 struct ceph_mon_generic_request
*req
;
717 u64 tid
= le64_to_cpu(msg
->hdr
.tid
);
718 void *p
= msg
->front
.iov_base
;
719 void *end
= p
+ msg
->front_alloc_len
;
722 dout("%s msg %p tid %llu\n", __func__
, msg
, tid
);
724 ceph_decode_need(&p
, end
, 2*sizeof(u64
), bad
);
725 handle
= ceph_decode_64(&p
);
726 if (tid
!= 0 && tid
!= handle
)
729 mutex_lock(&monc
->mutex
);
730 req
= lookup_generic_request(&monc
->generic_request_tree
, handle
);
732 mutex_unlock(&monc
->mutex
);
737 req
->u
.newest
= ceph_decode_64(&p
);
738 __finish_generic_request(req
);
739 mutex_unlock(&monc
->mutex
);
741 complete_generic_request(req
);
745 pr_err("corrupt mon_get_version reply, tid %llu\n", tid
);
749 static struct ceph_mon_generic_request
*
750 __ceph_monc_get_version(struct ceph_mon_client
*monc
, const char *what
,
751 ceph_monc_callback_t cb
, u64 private_data
)
753 struct ceph_mon_generic_request
*req
;
755 req
= alloc_generic_request(monc
, GFP_NOIO
);
759 req
->request
= ceph_msg_new(CEPH_MSG_MON_GET_VERSION
,
760 sizeof(u64
) + sizeof(u32
) + strlen(what
),
765 req
->reply
= ceph_msg_new(CEPH_MSG_MON_GET_VERSION_REPLY
, 32, GFP_NOIO
,
770 req
->complete_cb
= cb
;
771 req
->private_data
= private_data
;
773 mutex_lock(&monc
->mutex
);
774 register_generic_request(req
);
776 void *p
= req
->request
->front
.iov_base
;
777 void *const end
= p
+ req
->request
->front_alloc_len
;
779 ceph_encode_64(&p
, req
->tid
); /* handle */
780 ceph_encode_string(&p
, end
, what
, strlen(what
));
783 send_generic_request(monc
, req
);
784 mutex_unlock(&monc
->mutex
);
789 put_generic_request(req
);
790 return ERR_PTR(-ENOMEM
);
794 * Send MMonGetVersion and wait for the reply.
796 * @what: one of "mdsmap", "osdmap" or "monmap"
798 int ceph_monc_get_version(struct ceph_mon_client
*monc
, const char *what
,
801 struct ceph_mon_generic_request
*req
;
804 req
= __ceph_monc_get_version(monc
, what
, NULL
, 0);
808 ret
= wait_generic_request(req
);
810 *newest
= req
->u
.newest
;
812 put_generic_request(req
);
815 EXPORT_SYMBOL(ceph_monc_get_version
);
818 * Send MMonGetVersion,
820 * @what: one of "mdsmap", "osdmap" or "monmap"
822 int ceph_monc_get_version_async(struct ceph_mon_client
*monc
, const char *what
,
823 ceph_monc_callback_t cb
, u64 private_data
)
825 struct ceph_mon_generic_request
*req
;
827 req
= __ceph_monc_get_version(monc
, what
, cb
, private_data
);
831 put_generic_request(req
);
834 EXPORT_SYMBOL(ceph_monc_get_version_async
);
836 static void handle_command_ack(struct ceph_mon_client
*monc
,
837 struct ceph_msg
*msg
)
839 struct ceph_mon_generic_request
*req
;
840 void *p
= msg
->front
.iov_base
;
841 void *const end
= p
+ msg
->front_alloc_len
;
842 u64 tid
= le64_to_cpu(msg
->hdr
.tid
);
844 dout("%s msg %p tid %llu\n", __func__
, msg
, tid
);
846 ceph_decode_need(&p
, end
, sizeof(struct ceph_mon_request_header
) +
848 p
+= sizeof(struct ceph_mon_request_header
);
850 mutex_lock(&monc
->mutex
);
851 req
= lookup_generic_request(&monc
->generic_request_tree
, tid
);
853 mutex_unlock(&monc
->mutex
);
857 req
->result
= ceph_decode_32(&p
);
858 __finish_generic_request(req
);
859 mutex_unlock(&monc
->mutex
);
861 complete_generic_request(req
);
865 pr_err("corrupt mon_command ack, tid %llu\n", tid
);
869 int ceph_monc_blacklist_add(struct ceph_mon_client
*monc
,
870 struct ceph_entity_addr
*client_addr
)
872 struct ceph_mon_generic_request
*req
;
873 struct ceph_mon_command
*h
;
877 req
= alloc_generic_request(monc
, GFP_NOIO
);
881 req
->request
= ceph_msg_new(CEPH_MSG_MON_COMMAND
, 256, GFP_NOIO
, true);
885 req
->reply
= ceph_msg_new(CEPH_MSG_MON_COMMAND_ACK
, 512, GFP_NOIO
,
890 mutex_lock(&monc
->mutex
);
891 register_generic_request(req
);
892 h
= req
->request
->front
.iov_base
;
893 h
->monhdr
.have_version
= 0;
894 h
->monhdr
.session_mon
= cpu_to_le16(-1);
895 h
->monhdr
.session_mon_tid
= 0;
896 h
->fsid
= monc
->monmap
->fsid
;
897 h
->num_strs
= cpu_to_le32(1);
898 len
= sprintf(h
->str
, "{ \"prefix\": \"osd blacklist\", \
899 \"blacklistop\": \"add\", \
900 \"addr\": \"%pISpc/%u\" }",
901 &client_addr
->in_addr
, le32_to_cpu(client_addr
->nonce
));
902 h
->str_len
= cpu_to_le32(len
);
903 send_generic_request(monc
, req
);
904 mutex_unlock(&monc
->mutex
);
906 ret
= wait_generic_request(req
);
908 put_generic_request(req
);
911 EXPORT_SYMBOL(ceph_monc_blacklist_add
);
914 * Resend pending generic requests.
916 static void __resend_generic_request(struct ceph_mon_client
*monc
)
918 struct ceph_mon_generic_request
*req
;
921 for (p
= rb_first(&monc
->generic_request_tree
); p
; p
= rb_next(p
)) {
922 req
= rb_entry(p
, struct ceph_mon_generic_request
, node
);
923 ceph_msg_revoke(req
->request
);
924 ceph_msg_revoke_incoming(req
->reply
);
925 ceph_con_send(&monc
->con
, ceph_msg_get(req
->request
));
930 * Delayed work. If we haven't mounted yet, retry. Otherwise,
931 * renew/retry subscription as needed (in case it is timing out, or we
932 * got an ENOMEM). And keep the monitor connection alive.
934 static void delayed_work(struct work_struct
*work
)
936 struct ceph_mon_client
*monc
=
937 container_of(work
, struct ceph_mon_client
, delayed_work
.work
);
939 dout("monc delayed_work\n");
940 mutex_lock(&monc
->mutex
);
942 dout("%s continuing hunt\n", __func__
);
943 reopen_session(monc
);
945 int is_auth
= ceph_auth_is_authenticated(monc
->auth
);
946 if (ceph_con_keepalive_expired(&monc
->con
,
947 CEPH_MONC_PING_TIMEOUT
)) {
948 dout("monc keepalive timeout\n");
950 reopen_session(monc
);
953 if (!monc
->hunting
) {
954 ceph_con_keepalive(&monc
->con
);
955 __validate_auth(monc
);
959 unsigned long now
= jiffies
;
961 dout("%s renew subs? now %lu renew after %lu\n",
962 __func__
, now
, monc
->sub_renew_after
);
963 if (time_after_eq(now
, monc
->sub_renew_after
))
964 __send_subscribe(monc
);
967 __schedule_delayed(monc
);
968 mutex_unlock(&monc
->mutex
);
972 * On startup, we build a temporary monmap populated with the IPs
973 * provided by mount(2).
975 static int build_initial_monmap(struct ceph_mon_client
*monc
)
977 struct ceph_options
*opt
= monc
->client
->options
;
978 struct ceph_entity_addr
*mon_addr
= opt
->mon_addr
;
979 int num_mon
= opt
->num_mon
;
982 /* build initial monmap */
983 monc
->monmap
= kzalloc(sizeof(*monc
->monmap
) +
984 num_mon
*sizeof(monc
->monmap
->mon_inst
[0]),
988 for (i
= 0; i
< num_mon
; i
++) {
989 monc
->monmap
->mon_inst
[i
].addr
= mon_addr
[i
];
990 monc
->monmap
->mon_inst
[i
].addr
.nonce
= 0;
991 monc
->monmap
->mon_inst
[i
].name
.type
=
992 CEPH_ENTITY_TYPE_MON
;
993 monc
->monmap
->mon_inst
[i
].name
.num
= cpu_to_le64(i
);
995 monc
->monmap
->num_mon
= num_mon
;
999 int ceph_monc_init(struct ceph_mon_client
*monc
, struct ceph_client
*cl
)
1004 memset(monc
, 0, sizeof(*monc
));
1006 monc
->monmap
= NULL
;
1007 mutex_init(&monc
->mutex
);
1009 err
= build_initial_monmap(monc
);
1014 /* authentication */
1015 monc
->auth
= ceph_auth_init(cl
->options
->name
,
1017 if (IS_ERR(monc
->auth
)) {
1018 err
= PTR_ERR(monc
->auth
);
1021 monc
->auth
->want_keys
=
1022 CEPH_ENTITY_TYPE_AUTH
| CEPH_ENTITY_TYPE_MON
|
1023 CEPH_ENTITY_TYPE_OSD
| CEPH_ENTITY_TYPE_MDS
;
1027 monc
->m_subscribe_ack
= ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE_ACK
,
1028 sizeof(struct ceph_mon_subscribe_ack
),
1030 if (!monc
->m_subscribe_ack
)
1033 monc
->m_subscribe
= ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE
, 128,
1035 if (!monc
->m_subscribe
)
1036 goto out_subscribe_ack
;
1038 monc
->m_auth_reply
= ceph_msg_new(CEPH_MSG_AUTH_REPLY
, 4096,
1040 if (!monc
->m_auth_reply
)
1043 monc
->m_auth
= ceph_msg_new(CEPH_MSG_AUTH
, 4096, GFP_KERNEL
, true);
1044 monc
->pending_auth
= 0;
1046 goto out_auth_reply
;
1048 ceph_con_init(&monc
->con
, monc
, &mon_con_ops
,
1049 &monc
->client
->msgr
);
1052 monc
->had_a_connection
= false;
1053 monc
->hunt_mult
= 1;
1055 INIT_DELAYED_WORK(&monc
->delayed_work
, delayed_work
);
1056 monc
->generic_request_tree
= RB_ROOT
;
1059 monc
->fs_cluster_id
= CEPH_FS_CLUSTER_ID_NONE
;
1064 ceph_msg_put(monc
->m_auth_reply
);
1066 ceph_msg_put(monc
->m_subscribe
);
1068 ceph_msg_put(monc
->m_subscribe_ack
);
1070 ceph_auth_destroy(monc
->auth
);
1072 kfree(monc
->monmap
);
1076 EXPORT_SYMBOL(ceph_monc_init
);
1078 void ceph_monc_stop(struct ceph_mon_client
*monc
)
1081 cancel_delayed_work_sync(&monc
->delayed_work
);
1083 mutex_lock(&monc
->mutex
);
1084 __close_session(monc
);
1086 mutex_unlock(&monc
->mutex
);
1089 * flush msgr queue before we destroy ourselves to ensure that:
1090 * - any work that references our embedded con is finished.
1091 * - any osd_client or other work that may reference an authorizer
1092 * finishes before we shut down the auth subsystem.
1096 ceph_auth_destroy(monc
->auth
);
1098 WARN_ON(!RB_EMPTY_ROOT(&monc
->generic_request_tree
));
1100 ceph_msg_put(monc
->m_auth
);
1101 ceph_msg_put(monc
->m_auth_reply
);
1102 ceph_msg_put(monc
->m_subscribe
);
1103 ceph_msg_put(monc
->m_subscribe_ack
);
1105 kfree(monc
->monmap
);
1107 EXPORT_SYMBOL(ceph_monc_stop
);
1109 static void finish_hunting(struct ceph_mon_client
*monc
)
1111 if (monc
->hunting
) {
1112 dout("%s found mon%d\n", __func__
, monc
->cur_mon
);
1113 monc
->hunting
= false;
1114 monc
->had_a_connection
= true;
1115 monc
->hunt_mult
/= 2; /* reduce by 50% */
1116 if (monc
->hunt_mult
< 1)
1117 monc
->hunt_mult
= 1;
1121 static void handle_auth_reply(struct ceph_mon_client
*monc
,
1122 struct ceph_msg
*msg
)
1127 mutex_lock(&monc
->mutex
);
1128 was_auth
= ceph_auth_is_authenticated(monc
->auth
);
1129 monc
->pending_auth
= 0;
1130 ret
= ceph_handle_auth_reply(monc
->auth
, msg
->front
.iov_base
,
1132 monc
->m_auth
->front
.iov_base
,
1133 monc
->m_auth
->front_alloc_len
);
1135 __send_prepared_auth_request(monc
, ret
);
1139 finish_hunting(monc
);
1142 monc
->client
->auth_err
= ret
;
1143 } else if (!was_auth
&& ceph_auth_is_authenticated(monc
->auth
)) {
1144 dout("authenticated, starting session\n");
1146 monc
->client
->msgr
.inst
.name
.type
= CEPH_ENTITY_TYPE_CLIENT
;
1147 monc
->client
->msgr
.inst
.name
.num
=
1148 cpu_to_le64(monc
->auth
->global_id
);
1150 __send_subscribe(monc
);
1151 __resend_generic_request(monc
);
1153 pr_info("mon%d %s session established\n", monc
->cur_mon
,
1154 ceph_pr_addr(&monc
->con
.peer_addr
.in_addr
));
1158 mutex_unlock(&monc
->mutex
);
1159 if (monc
->client
->auth_err
< 0)
1160 wake_up_all(&monc
->client
->auth_wq
);
1163 static int __validate_auth(struct ceph_mon_client
*monc
)
1167 if (monc
->pending_auth
)
1170 ret
= ceph_build_auth(monc
->auth
, monc
->m_auth
->front
.iov_base
,
1171 monc
->m_auth
->front_alloc_len
);
1173 return ret
; /* either an error, or no need to authenticate */
1174 __send_prepared_auth_request(monc
, ret
);
1178 int ceph_monc_validate_auth(struct ceph_mon_client
*monc
)
1182 mutex_lock(&monc
->mutex
);
1183 ret
= __validate_auth(monc
);
1184 mutex_unlock(&monc
->mutex
);
1187 EXPORT_SYMBOL(ceph_monc_validate_auth
);
1190 * handle incoming message
1192 static void dispatch(struct ceph_connection
*con
, struct ceph_msg
*msg
)
1194 struct ceph_mon_client
*monc
= con
->private;
1195 int type
= le16_to_cpu(msg
->hdr
.type
);
1201 case CEPH_MSG_AUTH_REPLY
:
1202 handle_auth_reply(monc
, msg
);
1205 case CEPH_MSG_MON_SUBSCRIBE_ACK
:
1206 handle_subscribe_ack(monc
, msg
);
1209 case CEPH_MSG_STATFS_REPLY
:
1210 handle_statfs_reply(monc
, msg
);
1213 case CEPH_MSG_MON_GET_VERSION_REPLY
:
1214 handle_get_version_reply(monc
, msg
);
1217 case CEPH_MSG_MON_COMMAND_ACK
:
1218 handle_command_ack(monc
, msg
);
1221 case CEPH_MSG_MON_MAP
:
1222 ceph_monc_handle_map(monc
, msg
);
1225 case CEPH_MSG_OSD_MAP
:
1226 ceph_osdc_handle_map(&monc
->client
->osdc
, msg
);
1230 /* can the chained handler handle it? */
1231 if (monc
->client
->extra_mon_dispatch
&&
1232 monc
->client
->extra_mon_dispatch(monc
->client
, msg
) == 0)
1235 pr_err("received unknown message type %d %s\n", type
,
1236 ceph_msg_type_name(type
));
1242 * Allocate memory for incoming message
1244 static struct ceph_msg
*mon_alloc_msg(struct ceph_connection
*con
,
1245 struct ceph_msg_header
*hdr
,
1248 struct ceph_mon_client
*monc
= con
->private;
1249 int type
= le16_to_cpu(hdr
->type
);
1250 int front_len
= le32_to_cpu(hdr
->front_len
);
1251 struct ceph_msg
*m
= NULL
;
1256 case CEPH_MSG_MON_SUBSCRIBE_ACK
:
1257 m
= ceph_msg_get(monc
->m_subscribe_ack
);
1259 case CEPH_MSG_STATFS_REPLY
:
1260 case CEPH_MSG_MON_COMMAND_ACK
:
1261 return get_generic_reply(con
, hdr
, skip
);
1262 case CEPH_MSG_AUTH_REPLY
:
1263 m
= ceph_msg_get(monc
->m_auth_reply
);
1265 case CEPH_MSG_MON_GET_VERSION_REPLY
:
1266 if (le64_to_cpu(hdr
->tid
) != 0)
1267 return get_generic_reply(con
, hdr
, skip
);
1270 * Older OSDs don't set reply tid even if the orignal
1271 * request had a non-zero tid. Workaround this weirdness
1272 * by falling through to the allocate case.
1274 case CEPH_MSG_MON_MAP
:
1275 case CEPH_MSG_MDS_MAP
:
1276 case CEPH_MSG_OSD_MAP
:
1277 case CEPH_MSG_FS_MAP_USER
:
1278 m
= ceph_msg_new(type
, front_len
, GFP_NOFS
, false);
1280 return NULL
; /* ENOMEM--return skip == 0 */
1285 pr_info("alloc_msg unknown type %d\n", type
);
1287 } else if (front_len
> m
->front_alloc_len
) {
1288 pr_warn("mon_alloc_msg front %d > prealloc %d (%u#%llu)\n",
1289 front_len
, m
->front_alloc_len
,
1290 (unsigned int)con
->peer_name
.type
,
1291 le64_to_cpu(con
->peer_name
.num
));
1293 m
= ceph_msg_new(type
, front_len
, GFP_NOFS
, false);
1300 * If the monitor connection resets, pick a new monitor and resubmit
1301 * any pending requests.
1303 static void mon_fault(struct ceph_connection
*con
)
1305 struct ceph_mon_client
*monc
= con
->private;
1307 mutex_lock(&monc
->mutex
);
1308 dout("%s mon%d\n", __func__
, monc
->cur_mon
);
1309 if (monc
->cur_mon
>= 0) {
1310 if (!monc
->hunting
) {
1311 dout("%s hunting for new mon\n", __func__
);
1312 reopen_session(monc
);
1313 __schedule_delayed(monc
);
1315 dout("%s already hunting\n", __func__
);
1318 mutex_unlock(&monc
->mutex
);
1322 * We can ignore refcounting on the connection struct, as all references
1323 * will come from the messenger workqueue, which is drained prior to
1324 * mon_client destruction.
1326 static struct ceph_connection
*con_get(struct ceph_connection
*con
)
1331 static void con_put(struct ceph_connection
*con
)
1335 static const struct ceph_connection_operations mon_con_ops
= {
1338 .dispatch
= dispatch
,
1340 .alloc_msg
= mon_alloc_msg
,