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 * Open a session with a (new) monitor.
133 static int __open_session(struct ceph_mon_client
*monc
)
138 if (monc
->cur_mon
< 0) {
139 get_random_bytes(&r
, 1);
140 monc
->cur_mon
= r
% monc
->monmap
->num_mon
;
141 dout("open_session num=%d r=%d -> mon%d\n",
142 monc
->monmap
->num_mon
, r
, monc
->cur_mon
);
144 monc
->sub_renew_after
= jiffies
; /* i.e., expired */
145 monc
->want_next_osdmap
= !!monc
->want_next_osdmap
;
147 dout("open_session mon%d opening\n", monc
->cur_mon
);
148 ceph_con_open(&monc
->con
,
149 CEPH_ENTITY_TYPE_MON
, monc
->cur_mon
,
150 &monc
->monmap
->mon_inst
[monc
->cur_mon
].addr
);
152 /* send an initial keepalive to ensure our timestamp is
153 * valid by the time we are in an OPENED state */
154 ceph_con_keepalive(&monc
->con
);
156 /* initiatiate authentication handshake */
157 ret
= ceph_auth_build_hello(monc
->auth
,
158 monc
->m_auth
->front
.iov_base
,
159 monc
->m_auth
->front_alloc_len
);
160 __send_prepared_auth_request(monc
, ret
);
162 dout("open_session mon%d already open\n", monc
->cur_mon
);
167 static bool __sub_expired(struct ceph_mon_client
*monc
)
169 return time_after_eq(jiffies
, monc
->sub_renew_after
);
173 * Reschedule delayed work timer.
175 static void __schedule_delayed(struct ceph_mon_client
*monc
)
177 struct ceph_options
*opt
= monc
->client
->options
;
180 if (monc
->cur_mon
< 0 || __sub_expired(monc
)) {
184 if (opt
->monc_ping_timeout
> 0)
185 delay
= min(delay
, opt
->monc_ping_timeout
/ 3);
187 dout("__schedule_delayed after %lu\n", delay
);
188 schedule_delayed_work(&monc
->delayed_work
,
189 round_jiffies_relative(delay
));
193 * Send subscribe request for mdsmap and/or osdmap.
195 static void __send_subscribe(struct ceph_mon_client
*monc
)
197 dout("__send_subscribe sub_sent=%u exp=%u want_osd=%d\n",
198 (unsigned int)monc
->sub_sent
, __sub_expired(monc
),
199 monc
->want_next_osdmap
);
200 if ((__sub_expired(monc
) && !monc
->sub_sent
) ||
201 monc
->want_next_osdmap
== 1) {
202 struct ceph_msg
*msg
= monc
->m_subscribe
;
203 struct ceph_mon_subscribe_item
*i
;
207 p
= msg
->front
.iov_base
;
208 end
= p
+ msg
->front_alloc_len
;
210 num
= 1 + !!monc
->want_next_osdmap
+ !!monc
->want_mdsmap
;
211 ceph_encode_32(&p
, num
);
213 if (monc
->want_next_osdmap
) {
214 dout("__send_subscribe to 'osdmap' %u\n",
215 (unsigned int)monc
->have_osdmap
);
216 ceph_encode_string(&p
, end
, "osdmap", 6);
218 i
->have
= cpu_to_le64(monc
->have_osdmap
);
221 monc
->want_next_osdmap
= 2; /* requested */
223 if (monc
->want_mdsmap
) {
224 dout("__send_subscribe to 'mdsmap' %u+\n",
225 (unsigned int)monc
->have_mdsmap
);
226 ceph_encode_string(&p
, end
, "mdsmap", 6);
228 i
->have
= cpu_to_le64(monc
->have_mdsmap
);
232 ceph_encode_string(&p
, end
, "monmap", 6);
238 msg
->front
.iov_len
= p
- msg
->front
.iov_base
;
239 msg
->hdr
.front_len
= cpu_to_le32(msg
->front
.iov_len
);
240 ceph_msg_revoke(msg
);
241 ceph_con_send(&monc
->con
, ceph_msg_get(msg
));
243 monc
->sub_sent
= jiffies
| 1; /* never 0 */
247 static void handle_subscribe_ack(struct ceph_mon_client
*monc
,
248 struct ceph_msg
*msg
)
250 unsigned int seconds
;
251 struct ceph_mon_subscribe_ack
*h
= msg
->front
.iov_base
;
253 if (msg
->front
.iov_len
< sizeof(*h
))
255 seconds
= le32_to_cpu(h
->duration
);
257 mutex_lock(&monc
->mutex
);
259 pr_info("mon%d %s session established\n",
261 ceph_pr_addr(&monc
->con
.peer_addr
.in_addr
));
262 monc
->hunting
= false;
264 dout("handle_subscribe_ack after %d seconds\n", seconds
);
265 monc
->sub_renew_after
= monc
->sub_sent
+ (seconds
>> 1)*HZ
- 1;
267 mutex_unlock(&monc
->mutex
);
270 pr_err("got corrupt subscribe-ack msg\n");
275 * Keep track of which maps we have
277 int ceph_monc_got_mdsmap(struct ceph_mon_client
*monc
, u32 got
)
279 mutex_lock(&monc
->mutex
);
280 monc
->have_mdsmap
= got
;
281 mutex_unlock(&monc
->mutex
);
284 EXPORT_SYMBOL(ceph_monc_got_mdsmap
);
286 int ceph_monc_got_osdmap(struct ceph_mon_client
*monc
, u32 got
)
288 mutex_lock(&monc
->mutex
);
289 monc
->have_osdmap
= got
;
290 monc
->want_next_osdmap
= 0;
291 mutex_unlock(&monc
->mutex
);
296 * Register interest in the next osdmap
298 void ceph_monc_request_next_osdmap(struct ceph_mon_client
*monc
)
300 dout("request_next_osdmap have %u\n", monc
->have_osdmap
);
301 mutex_lock(&monc
->mutex
);
302 if (!monc
->want_next_osdmap
)
303 monc
->want_next_osdmap
= 1;
304 if (monc
->want_next_osdmap
< 2)
305 __send_subscribe(monc
);
306 mutex_unlock(&monc
->mutex
);
308 EXPORT_SYMBOL(ceph_monc_request_next_osdmap
);
311 * Wait for an osdmap with a given epoch.
313 * @epoch: epoch to wait for
314 * @timeout: in jiffies, 0 means "wait forever"
316 int ceph_monc_wait_osdmap(struct ceph_mon_client
*monc
, u32 epoch
,
317 unsigned long timeout
)
319 unsigned long started
= jiffies
;
322 mutex_lock(&monc
->mutex
);
323 while (monc
->have_osdmap
< epoch
) {
324 mutex_unlock(&monc
->mutex
);
326 if (timeout
&& time_after_eq(jiffies
, started
+ timeout
))
329 ret
= wait_event_interruptible_timeout(monc
->client
->auth_wq
,
330 monc
->have_osdmap
>= epoch
,
331 ceph_timeout_jiffies(timeout
));
335 mutex_lock(&monc
->mutex
);
338 mutex_unlock(&monc
->mutex
);
341 EXPORT_SYMBOL(ceph_monc_wait_osdmap
);
346 int ceph_monc_open_session(struct ceph_mon_client
*monc
)
348 mutex_lock(&monc
->mutex
);
349 __open_session(monc
);
350 __schedule_delayed(monc
);
351 mutex_unlock(&monc
->mutex
);
354 EXPORT_SYMBOL(ceph_monc_open_session
);
357 * We require the fsid and global_id in order to initialize our
360 static bool have_debugfs_info(struct ceph_mon_client
*monc
)
362 dout("have_debugfs_info fsid %d globalid %lld\n",
363 (int)monc
->client
->have_fsid
, monc
->auth
->global_id
);
364 return monc
->client
->have_fsid
&& monc
->auth
->global_id
> 0;
367 static void ceph_monc_handle_map(struct ceph_mon_client
*monc
,
368 struct ceph_msg
*msg
)
370 struct ceph_client
*client
= monc
->client
;
371 struct ceph_monmap
*monmap
= NULL
, *old
= monc
->monmap
;
373 int had_debugfs_info
, init_debugfs
= 0;
375 mutex_lock(&monc
->mutex
);
377 had_debugfs_info
= have_debugfs_info(monc
);
379 dout("handle_monmap\n");
380 p
= msg
->front
.iov_base
;
381 end
= p
+ msg
->front
.iov_len
;
383 monmap
= ceph_monmap_decode(p
, end
);
384 if (IS_ERR(monmap
)) {
385 pr_err("problem decoding monmap, %d\n",
386 (int)PTR_ERR(monmap
));
390 if (ceph_check_fsid(monc
->client
, &monmap
->fsid
) < 0) {
395 client
->monc
.monmap
= monmap
;
398 if (!client
->have_fsid
) {
399 client
->have_fsid
= true;
400 if (!had_debugfs_info
&& have_debugfs_info(monc
)) {
401 pr_info("client%lld fsid %pU\n",
402 ceph_client_id(monc
->client
),
403 &monc
->client
->fsid
);
406 mutex_unlock(&monc
->mutex
);
410 * do debugfs initialization without mutex to avoid
411 * creating a locking dependency
413 ceph_debugfs_client_init(monc
->client
);
419 mutex_unlock(&monc
->mutex
);
421 wake_up_all(&client
->auth_wq
);
425 * generic requests (currently statfs, mon_get_version)
427 static struct ceph_mon_generic_request
*__lookup_generic_req(
428 struct ceph_mon_client
*monc
, u64 tid
)
430 struct ceph_mon_generic_request
*req
;
431 struct rb_node
*n
= monc
->generic_request_tree
.rb_node
;
434 req
= rb_entry(n
, struct ceph_mon_generic_request
, node
);
437 else if (tid
> req
->tid
)
445 static void __insert_generic_request(struct ceph_mon_client
*monc
,
446 struct ceph_mon_generic_request
*new)
448 struct rb_node
**p
= &monc
->generic_request_tree
.rb_node
;
449 struct rb_node
*parent
= NULL
;
450 struct ceph_mon_generic_request
*req
= NULL
;
454 req
= rb_entry(parent
, struct ceph_mon_generic_request
, node
);
455 if (new->tid
< req
->tid
)
457 else if (new->tid
> req
->tid
)
463 rb_link_node(&new->node
, parent
, p
);
464 rb_insert_color(&new->node
, &monc
->generic_request_tree
);
467 static void release_generic_request(struct kref
*kref
)
469 struct ceph_mon_generic_request
*req
=
470 container_of(kref
, struct ceph_mon_generic_request
, kref
);
473 ceph_msg_put(req
->reply
);
475 ceph_msg_put(req
->request
);
480 static void put_generic_request(struct ceph_mon_generic_request
*req
)
482 kref_put(&req
->kref
, release_generic_request
);
485 static void get_generic_request(struct ceph_mon_generic_request
*req
)
487 kref_get(&req
->kref
);
490 static struct ceph_msg
*get_generic_reply(struct ceph_connection
*con
,
491 struct ceph_msg_header
*hdr
,
494 struct ceph_mon_client
*monc
= con
->private;
495 struct ceph_mon_generic_request
*req
;
496 u64 tid
= le64_to_cpu(hdr
->tid
);
499 mutex_lock(&monc
->mutex
);
500 req
= __lookup_generic_req(monc
, tid
);
502 dout("get_generic_reply %lld dne\n", tid
);
506 dout("get_generic_reply %lld got %p\n", tid
, req
->reply
);
508 m
= ceph_msg_get(req
->reply
);
510 * we don't need to track the connection reading into
511 * this reply because we only have one open connection
515 mutex_unlock(&monc
->mutex
);
519 static int __do_generic_request(struct ceph_mon_client
*monc
, u64 tid
,
520 struct ceph_mon_generic_request
*req
)
524 /* register request */
525 req
->tid
= tid
!= 0 ? tid
: ++monc
->last_tid
;
526 req
->request
->hdr
.tid
= cpu_to_le64(req
->tid
);
527 __insert_generic_request(monc
, req
);
528 monc
->num_generic_requests
++;
529 ceph_con_send(&monc
->con
, ceph_msg_get(req
->request
));
530 mutex_unlock(&monc
->mutex
);
532 err
= wait_for_completion_interruptible(&req
->completion
);
534 mutex_lock(&monc
->mutex
);
535 rb_erase(&req
->node
, &monc
->generic_request_tree
);
536 monc
->num_generic_requests
--;
543 static int do_generic_request(struct ceph_mon_client
*monc
,
544 struct ceph_mon_generic_request
*req
)
548 mutex_lock(&monc
->mutex
);
549 err
= __do_generic_request(monc
, 0, req
);
550 mutex_unlock(&monc
->mutex
);
558 static void handle_statfs_reply(struct ceph_mon_client
*monc
,
559 struct ceph_msg
*msg
)
561 struct ceph_mon_generic_request
*req
;
562 struct ceph_mon_statfs_reply
*reply
= msg
->front
.iov_base
;
563 u64 tid
= le64_to_cpu(msg
->hdr
.tid
);
565 if (msg
->front
.iov_len
!= sizeof(*reply
))
567 dout("handle_statfs_reply %p tid %llu\n", msg
, tid
);
569 mutex_lock(&monc
->mutex
);
570 req
= __lookup_generic_req(monc
, tid
);
572 *(struct ceph_statfs
*)req
->buf
= reply
->st
;
574 get_generic_request(req
);
576 mutex_unlock(&monc
->mutex
);
578 complete_all(&req
->completion
);
579 put_generic_request(req
);
584 pr_err("corrupt statfs reply, tid %llu\n", tid
);
589 * Do a synchronous statfs().
591 int ceph_monc_do_statfs(struct ceph_mon_client
*monc
, struct ceph_statfs
*buf
)
593 struct ceph_mon_generic_request
*req
;
594 struct ceph_mon_statfs
*h
;
597 req
= kzalloc(sizeof(*req
), GFP_NOFS
);
601 kref_init(&req
->kref
);
603 init_completion(&req
->completion
);
606 req
->request
= ceph_msg_new(CEPH_MSG_STATFS
, sizeof(*h
), GFP_NOFS
,
610 req
->reply
= ceph_msg_new(CEPH_MSG_STATFS_REPLY
, 1024, GFP_NOFS
,
615 /* fill out request */
616 h
= req
->request
->front
.iov_base
;
617 h
->monhdr
.have_version
= 0;
618 h
->monhdr
.session_mon
= cpu_to_le16(-1);
619 h
->monhdr
.session_mon_tid
= 0;
620 h
->fsid
= monc
->monmap
->fsid
;
622 err
= do_generic_request(monc
, req
);
625 put_generic_request(req
);
628 EXPORT_SYMBOL(ceph_monc_do_statfs
);
630 static void handle_get_version_reply(struct ceph_mon_client
*monc
,
631 struct ceph_msg
*msg
)
633 struct ceph_mon_generic_request
*req
;
634 u64 tid
= le64_to_cpu(msg
->hdr
.tid
);
635 void *p
= msg
->front
.iov_base
;
636 void *end
= p
+ msg
->front_alloc_len
;
639 dout("%s %p tid %llu\n", __func__
, msg
, tid
);
641 ceph_decode_need(&p
, end
, 2*sizeof(u64
), bad
);
642 handle
= ceph_decode_64(&p
);
643 if (tid
!= 0 && tid
!= handle
)
646 mutex_lock(&monc
->mutex
);
647 req
= __lookup_generic_req(monc
, handle
);
649 *(u64
*)req
->buf
= ceph_decode_64(&p
);
651 get_generic_request(req
);
653 mutex_unlock(&monc
->mutex
);
655 complete_all(&req
->completion
);
656 put_generic_request(req
);
661 pr_err("corrupt mon_get_version reply, tid %llu\n", tid
);
666 * Send MMonGetVersion and wait for the reply.
668 * @what: one of "mdsmap", "osdmap" or "monmap"
670 int ceph_monc_do_get_version(struct ceph_mon_client
*monc
, const char *what
,
673 struct ceph_mon_generic_request
*req
;
678 req
= kzalloc(sizeof(*req
), GFP_NOFS
);
682 kref_init(&req
->kref
);
684 init_completion(&req
->completion
);
686 req
->request
= ceph_msg_new(CEPH_MSG_MON_GET_VERSION
,
687 sizeof(u64
) + sizeof(u32
) + strlen(what
),
694 req
->reply
= ceph_msg_new(CEPH_MSG_MON_GET_VERSION_REPLY
, 1024,
701 p
= req
->request
->front
.iov_base
;
702 end
= p
+ req
->request
->front_alloc_len
;
704 /* fill out request */
705 mutex_lock(&monc
->mutex
);
706 tid
= ++monc
->last_tid
;
707 ceph_encode_64(&p
, tid
); /* handle */
708 ceph_encode_string(&p
, end
, what
, strlen(what
));
710 err
= __do_generic_request(monc
, tid
, req
);
712 mutex_unlock(&monc
->mutex
);
714 put_generic_request(req
);
717 EXPORT_SYMBOL(ceph_monc_do_get_version
);
720 * Resend pending generic requests.
722 static void __resend_generic_request(struct ceph_mon_client
*monc
)
724 struct ceph_mon_generic_request
*req
;
727 for (p
= rb_first(&monc
->generic_request_tree
); p
; p
= rb_next(p
)) {
728 req
= rb_entry(p
, struct ceph_mon_generic_request
, node
);
729 ceph_msg_revoke(req
->request
);
730 ceph_msg_revoke_incoming(req
->reply
);
731 ceph_con_send(&monc
->con
, ceph_msg_get(req
->request
));
736 * Delayed work. If we haven't mounted yet, retry. Otherwise,
737 * renew/retry subscription as needed (in case it is timing out, or we
738 * got an ENOMEM). And keep the monitor connection alive.
740 static void delayed_work(struct work_struct
*work
)
742 struct ceph_mon_client
*monc
=
743 container_of(work
, struct ceph_mon_client
, delayed_work
.work
);
745 dout("monc delayed_work\n");
746 mutex_lock(&monc
->mutex
);
748 __close_session(monc
);
749 __open_session(monc
); /* continue hunting */
751 struct ceph_options
*opt
= monc
->client
->options
;
752 int is_auth
= ceph_auth_is_authenticated(monc
->auth
);
753 if (ceph_con_keepalive_expired(&monc
->con
,
754 opt
->monc_ping_timeout
)) {
755 dout("monc keepalive timeout\n");
757 __close_session(monc
);
758 monc
->hunting
= true;
759 __open_session(monc
);
762 if (!monc
->hunting
) {
763 ceph_con_keepalive(&monc
->con
);
764 __validate_auth(monc
);
768 __send_subscribe(monc
);
770 __schedule_delayed(monc
);
771 mutex_unlock(&monc
->mutex
);
775 * On startup, we build a temporary monmap populated with the IPs
776 * provided by mount(2).
778 static int build_initial_monmap(struct ceph_mon_client
*monc
)
780 struct ceph_options
*opt
= monc
->client
->options
;
781 struct ceph_entity_addr
*mon_addr
= opt
->mon_addr
;
782 int num_mon
= opt
->num_mon
;
785 /* build initial monmap */
786 monc
->monmap
= kzalloc(sizeof(*monc
->monmap
) +
787 num_mon
*sizeof(monc
->monmap
->mon_inst
[0]),
791 for (i
= 0; i
< num_mon
; i
++) {
792 monc
->monmap
->mon_inst
[i
].addr
= mon_addr
[i
];
793 monc
->monmap
->mon_inst
[i
].addr
.nonce
= 0;
794 monc
->monmap
->mon_inst
[i
].name
.type
=
795 CEPH_ENTITY_TYPE_MON
;
796 monc
->monmap
->mon_inst
[i
].name
.num
= cpu_to_le64(i
);
798 monc
->monmap
->num_mon
= num_mon
;
802 int ceph_monc_init(struct ceph_mon_client
*monc
, struct ceph_client
*cl
)
807 memset(monc
, 0, sizeof(*monc
));
810 mutex_init(&monc
->mutex
);
812 err
= build_initial_monmap(monc
);
818 monc
->auth
= ceph_auth_init(cl
->options
->name
,
820 if (IS_ERR(monc
->auth
)) {
821 err
= PTR_ERR(monc
->auth
);
824 monc
->auth
->want_keys
=
825 CEPH_ENTITY_TYPE_AUTH
| CEPH_ENTITY_TYPE_MON
|
826 CEPH_ENTITY_TYPE_OSD
| CEPH_ENTITY_TYPE_MDS
;
830 monc
->m_subscribe_ack
= ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE_ACK
,
831 sizeof(struct ceph_mon_subscribe_ack
),
833 if (!monc
->m_subscribe_ack
)
836 monc
->m_subscribe
= ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE
, 96, GFP_NOFS
,
838 if (!monc
->m_subscribe
)
839 goto out_subscribe_ack
;
841 monc
->m_auth_reply
= ceph_msg_new(CEPH_MSG_AUTH_REPLY
, 4096, GFP_NOFS
,
843 if (!monc
->m_auth_reply
)
846 monc
->m_auth
= ceph_msg_new(CEPH_MSG_AUTH
, 4096, GFP_NOFS
, true);
847 monc
->pending_auth
= 0;
851 ceph_con_init(&monc
->con
, monc
, &mon_con_ops
,
852 &monc
->client
->msgr
);
855 monc
->hunting
= true;
856 monc
->sub_renew_after
= jiffies
;
859 INIT_DELAYED_WORK(&monc
->delayed_work
, delayed_work
);
860 monc
->generic_request_tree
= RB_ROOT
;
861 monc
->num_generic_requests
= 0;
864 monc
->have_mdsmap
= 0;
865 monc
->have_osdmap
= 0;
866 monc
->want_next_osdmap
= 1;
870 ceph_msg_put(monc
->m_auth_reply
);
872 ceph_msg_put(monc
->m_subscribe
);
874 ceph_msg_put(monc
->m_subscribe_ack
);
876 ceph_auth_destroy(monc
->auth
);
882 EXPORT_SYMBOL(ceph_monc_init
);
884 void ceph_monc_stop(struct ceph_mon_client
*monc
)
887 cancel_delayed_work_sync(&monc
->delayed_work
);
889 mutex_lock(&monc
->mutex
);
890 __close_session(monc
);
892 mutex_unlock(&monc
->mutex
);
895 * flush msgr queue before we destroy ourselves to ensure that:
896 * - any work that references our embedded con is finished.
897 * - any osd_client or other work that may reference an authorizer
898 * finishes before we shut down the auth subsystem.
902 ceph_auth_destroy(monc
->auth
);
904 ceph_msg_put(monc
->m_auth
);
905 ceph_msg_put(monc
->m_auth_reply
);
906 ceph_msg_put(monc
->m_subscribe
);
907 ceph_msg_put(monc
->m_subscribe_ack
);
911 EXPORT_SYMBOL(ceph_monc_stop
);
913 static void handle_auth_reply(struct ceph_mon_client
*monc
,
914 struct ceph_msg
*msg
)
918 int had_debugfs_info
, init_debugfs
= 0;
920 mutex_lock(&monc
->mutex
);
921 had_debugfs_info
= have_debugfs_info(monc
);
922 was_auth
= ceph_auth_is_authenticated(monc
->auth
);
923 monc
->pending_auth
= 0;
924 ret
= ceph_handle_auth_reply(monc
->auth
, msg
->front
.iov_base
,
926 monc
->m_auth
->front
.iov_base
,
927 monc
->m_auth
->front_alloc_len
);
929 monc
->client
->auth_err
= ret
;
930 wake_up_all(&monc
->client
->auth_wq
);
931 } else if (ret
> 0) {
932 __send_prepared_auth_request(monc
, ret
);
933 } else if (!was_auth
&& ceph_auth_is_authenticated(monc
->auth
)) {
934 dout("authenticated, starting session\n");
936 monc
->client
->msgr
.inst
.name
.type
= CEPH_ENTITY_TYPE_CLIENT
;
937 monc
->client
->msgr
.inst
.name
.num
=
938 cpu_to_le64(monc
->auth
->global_id
);
940 __send_subscribe(monc
);
941 __resend_generic_request(monc
);
944 if (!had_debugfs_info
&& have_debugfs_info(monc
)) {
945 pr_info("client%lld fsid %pU\n",
946 ceph_client_id(monc
->client
),
947 &monc
->client
->fsid
);
950 mutex_unlock(&monc
->mutex
);
954 * do debugfs initialization without mutex to avoid
955 * creating a locking dependency
957 ceph_debugfs_client_init(monc
->client
);
961 static int __validate_auth(struct ceph_mon_client
*monc
)
965 if (monc
->pending_auth
)
968 ret
= ceph_build_auth(monc
->auth
, monc
->m_auth
->front
.iov_base
,
969 monc
->m_auth
->front_alloc_len
);
971 return ret
; /* either an error, or no need to authenticate */
972 __send_prepared_auth_request(monc
, ret
);
976 int ceph_monc_validate_auth(struct ceph_mon_client
*monc
)
980 mutex_lock(&monc
->mutex
);
981 ret
= __validate_auth(monc
);
982 mutex_unlock(&monc
->mutex
);
985 EXPORT_SYMBOL(ceph_monc_validate_auth
);
988 * handle incoming message
990 static void dispatch(struct ceph_connection
*con
, struct ceph_msg
*msg
)
992 struct ceph_mon_client
*monc
= con
->private;
993 int type
= le16_to_cpu(msg
->hdr
.type
);
999 case CEPH_MSG_AUTH_REPLY
:
1000 handle_auth_reply(monc
, msg
);
1003 case CEPH_MSG_MON_SUBSCRIBE_ACK
:
1004 handle_subscribe_ack(monc
, msg
);
1007 case CEPH_MSG_STATFS_REPLY
:
1008 handle_statfs_reply(monc
, msg
);
1011 case CEPH_MSG_MON_GET_VERSION_REPLY
:
1012 handle_get_version_reply(monc
, msg
);
1015 case CEPH_MSG_MON_MAP
:
1016 ceph_monc_handle_map(monc
, msg
);
1019 case CEPH_MSG_OSD_MAP
:
1020 ceph_osdc_handle_map(&monc
->client
->osdc
, msg
);
1024 /* can the chained handler handle it? */
1025 if (monc
->client
->extra_mon_dispatch
&&
1026 monc
->client
->extra_mon_dispatch(monc
->client
, msg
) == 0)
1029 pr_err("received unknown message type %d %s\n", type
,
1030 ceph_msg_type_name(type
));
1036 * Allocate memory for incoming message
1038 static struct ceph_msg
*mon_alloc_msg(struct ceph_connection
*con
,
1039 struct ceph_msg_header
*hdr
,
1042 struct ceph_mon_client
*monc
= con
->private;
1043 int type
= le16_to_cpu(hdr
->type
);
1044 int front_len
= le32_to_cpu(hdr
->front_len
);
1045 struct ceph_msg
*m
= NULL
;
1050 case CEPH_MSG_MON_SUBSCRIBE_ACK
:
1051 m
= ceph_msg_get(monc
->m_subscribe_ack
);
1053 case CEPH_MSG_STATFS_REPLY
:
1054 return get_generic_reply(con
, hdr
, skip
);
1055 case CEPH_MSG_AUTH_REPLY
:
1056 m
= ceph_msg_get(monc
->m_auth_reply
);
1058 case CEPH_MSG_MON_GET_VERSION_REPLY
:
1059 if (le64_to_cpu(hdr
->tid
) != 0)
1060 return get_generic_reply(con
, hdr
, skip
);
1063 * Older OSDs don't set reply tid even if the orignal
1064 * request had a non-zero tid. Workaround this weirdness
1065 * by falling through to the allocate case.
1067 case CEPH_MSG_MON_MAP
:
1068 case CEPH_MSG_MDS_MAP
:
1069 case CEPH_MSG_OSD_MAP
:
1070 m
= ceph_msg_new(type
, front_len
, GFP_NOFS
, false);
1072 return NULL
; /* ENOMEM--return skip == 0 */
1077 pr_info("alloc_msg unknown type %d\n", type
);
1079 } else if (front_len
> m
->front_alloc_len
) {
1080 pr_warn("mon_alloc_msg front %d > prealloc %d (%u#%llu)\n",
1081 front_len
, m
->front_alloc_len
,
1082 (unsigned int)con
->peer_name
.type
,
1083 le64_to_cpu(con
->peer_name
.num
));
1085 m
= ceph_msg_new(type
, front_len
, GFP_NOFS
, false);
1092 * If the monitor connection resets, pick a new monitor and resubmit
1093 * any pending requests.
1095 static void mon_fault(struct ceph_connection
*con
)
1097 struct ceph_mon_client
*monc
= con
->private;
1102 dout("mon_fault\n");
1103 mutex_lock(&monc
->mutex
);
1108 pr_info("mon%d %s session lost, "
1109 "hunting for new mon\n", monc
->cur_mon
,
1110 ceph_pr_addr(&monc
->con
.peer_addr
.in_addr
));
1112 __close_session(monc
);
1113 if (!monc
->hunting
) {
1115 monc
->hunting
= true;
1116 __open_session(monc
);
1118 /* already hunting, let's wait a bit */
1119 __schedule_delayed(monc
);
1122 mutex_unlock(&monc
->mutex
);
1126 * We can ignore refcounting on the connection struct, as all references
1127 * will come from the messenger workqueue, which is drained prior to
1128 * mon_client destruction.
1130 static struct ceph_connection
*con_get(struct ceph_connection
*con
)
1135 static void con_put(struct ceph_connection
*con
)
1139 static const struct ceph_connection_operations mon_con_ops
= {
1142 .dispatch
= dispatch
,
1144 .alloc_msg
= mon_alloc_msg
,