1
2#include <linux/ceph/ceph_debug.h>
3
4#include <linux/module.h>
5#include <linux/types.h>
6#include <linux/slab.h>
7#include <linux/random.h>
8#include <linux/sched.h>
9
10#include <linux/ceph/ceph_features.h>
11#include <linux/ceph/mon_client.h>
12#include <linux/ceph/libceph.h>
13#include <linux/ceph/debugfs.h>
14#include <linux/ceph/decode.h>
15#include <linux/ceph/auth.h>
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35static const struct ceph_connection_operations mon_con_ops;
36
37static int __validate_auth(struct ceph_mon_client *monc);
38
39
40
41
42static struct ceph_monmap *ceph_monmap_decode(void *p, void *end)
43{
44 struct ceph_monmap *m = NULL;
45 int i, err = -EINVAL;
46 struct ceph_fsid fsid;
47 u32 epoch, num_mon;
48 u32 len;
49
50 ceph_decode_32_safe(&p, end, len, bad);
51 ceph_decode_need(&p, end, len, bad);
52
53 dout("monmap_decode %p %p len %d (%d)\n", p, end, len, (int)(end-p));
54 p += sizeof(u16);
55
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);
59
60 num_mon = ceph_decode_32(&p);
61
62 if (num_mon > CEPH_MAX_MON)
63 goto bad;
64 m = kmalloc(struct_size(m, mon_inst, num_mon), GFP_NOFS);
65 if (m == NULL)
66 return ERR_PTR(-ENOMEM);
67 m->fsid = fsid;
68 m->epoch = epoch;
69 m->num_mon = num_mon;
70 for (i = 0; i < num_mon; ++i) {
71 struct ceph_entity_inst *inst = &m->mon_inst[i];
72
73
74 ceph_decode_copy_safe(&p, end, &inst->name,
75 sizeof(inst->name), bad);
76 err = ceph_decode_entity_addr(&p, end, &inst->addr);
77 if (err)
78 goto bad;
79 }
80 dout("monmap_decode epoch %d, num_mon %d\n", m->epoch,
81 m->num_mon);
82 for (i = 0; i < m->num_mon; i++)
83 dout("monmap_decode mon%d is %s\n", i,
84 ceph_pr_addr(&m->mon_inst[i].addr));
85 return m;
86bad:
87 dout("monmap_decode failed with %d\n", err);
88 kfree(m);
89 return ERR_PTR(err);
90}
91
92
93
94
95int ceph_monmap_contains(struct ceph_monmap *m, struct ceph_entity_addr *addr)
96{
97 int i;
98
99 for (i = 0; i < m->num_mon; i++)
100 if (memcmp(addr, &m->mon_inst[i].addr, sizeof(*addr)) == 0)
101 return 1;
102 return 0;
103}
104
105
106
107
108static void __send_prepared_auth_request(struct ceph_mon_client *monc, int len)
109{
110 monc->pending_auth = 1;
111 monc->m_auth->front.iov_len = len;
112 monc->m_auth->hdr.front_len = cpu_to_le32(len);
113 ceph_msg_revoke(monc->m_auth);
114 ceph_msg_get(monc->m_auth);
115 ceph_con_send(&monc->con, monc->m_auth);
116}
117
118
119
120
121static void __close_session(struct ceph_mon_client *monc)
122{
123 dout("__close_session closing mon%d\n", monc->cur_mon);
124 ceph_msg_revoke(monc->m_auth);
125 ceph_msg_revoke_incoming(monc->m_auth_reply);
126 ceph_msg_revoke(monc->m_subscribe);
127 ceph_msg_revoke_incoming(monc->m_subscribe_ack);
128 ceph_con_close(&monc->con);
129
130 monc->pending_auth = 0;
131 ceph_auth_reset(monc->auth);
132}
133
134
135
136
137
138static void pick_new_mon(struct ceph_mon_client *monc)
139{
140 int old_mon = monc->cur_mon;
141
142 BUG_ON(monc->monmap->num_mon < 1);
143
144 if (monc->monmap->num_mon == 1) {
145 monc->cur_mon = 0;
146 } else {
147 int max = monc->monmap->num_mon;
148 int o = -1;
149 int n;
150
151 if (monc->cur_mon >= 0) {
152 if (monc->cur_mon < monc->monmap->num_mon)
153 o = monc->cur_mon;
154 if (o >= 0)
155 max--;
156 }
157
158 n = prandom_u32() % max;
159 if (o >= 0 && n >= o)
160 n++;
161
162 monc->cur_mon = n;
163 }
164
165 dout("%s mon%d -> mon%d out of %d mons\n", __func__, old_mon,
166 monc->cur_mon, monc->monmap->num_mon);
167}
168
169
170
171
172static void __open_session(struct ceph_mon_client *monc)
173{
174 int ret;
175
176 pick_new_mon(monc);
177
178 monc->hunting = true;
179 if (monc->had_a_connection) {
180 monc->hunt_mult *= CEPH_MONC_HUNT_BACKOFF;
181 if (monc->hunt_mult > CEPH_MONC_HUNT_MAX_MULT)
182 monc->hunt_mult = CEPH_MONC_HUNT_MAX_MULT;
183 }
184
185 monc->sub_renew_after = jiffies;
186 monc->sub_renew_sent = 0;
187
188 dout("%s opening mon%d\n", __func__, monc->cur_mon);
189 ceph_con_open(&monc->con, CEPH_ENTITY_TYPE_MON, monc->cur_mon,
190 &monc->monmap->mon_inst[monc->cur_mon].addr);
191
192
193
194
195
196 ceph_con_keepalive(&monc->con);
197
198
199 ret = ceph_auth_build_hello(monc->auth,
200 monc->m_auth->front.iov_base,
201 monc->m_auth->front_alloc_len);
202 BUG_ON(ret <= 0);
203 __send_prepared_auth_request(monc, ret);
204}
205
206static void reopen_session(struct ceph_mon_client *monc)
207{
208 if (!monc->hunting)
209 pr_info("mon%d %s session lost, hunting for new mon\n",
210 monc->cur_mon, ceph_pr_addr(&monc->con.peer_addr));
211
212 __close_session(monc);
213 __open_session(monc);
214}
215
216static void un_backoff(struct ceph_mon_client *monc)
217{
218 monc->hunt_mult /= 2;
219 if (monc->hunt_mult < 1)
220 monc->hunt_mult = 1;
221 dout("%s hunt_mult now %d\n", __func__, monc->hunt_mult);
222}
223
224
225
226
227static void __schedule_delayed(struct ceph_mon_client *monc)
228{
229 unsigned long delay;
230
231 if (monc->hunting)
232 delay = CEPH_MONC_HUNT_INTERVAL * monc->hunt_mult;
233 else
234 delay = CEPH_MONC_PING_INTERVAL;
235
236 dout("__schedule_delayed after %lu\n", delay);
237 mod_delayed_work(system_wq, &monc->delayed_work,
238 round_jiffies_relative(delay));
239}
240
241const char *ceph_sub_str[] = {
242 [CEPH_SUB_MONMAP] = "monmap",
243 [CEPH_SUB_OSDMAP] = "osdmap",
244 [CEPH_SUB_FSMAP] = "fsmap.user",
245 [CEPH_SUB_MDSMAP] = "mdsmap",
246};
247
248
249
250
251
252static void __send_subscribe(struct ceph_mon_client *monc)
253{
254 struct ceph_msg *msg = monc->m_subscribe;
255 void *p = msg->front.iov_base;
256 void *const end = p + msg->front_alloc_len;
257 int num = 0;
258 int i;
259
260 dout("%s sent %lu\n", __func__, monc->sub_renew_sent);
261
262 BUG_ON(monc->cur_mon < 0);
263
264 if (!monc->sub_renew_sent)
265 monc->sub_renew_sent = jiffies | 1;
266
267 msg->hdr.version = cpu_to_le16(2);
268
269 for (i = 0; i < ARRAY_SIZE(monc->subs); i++) {
270 if (monc->subs[i].want)
271 num++;
272 }
273 BUG_ON(num < 1);
274 ceph_encode_32(&p, num);
275 for (i = 0; i < ARRAY_SIZE(monc->subs); i++) {
276 char buf[32];
277 int len;
278
279 if (!monc->subs[i].want)
280 continue;
281
282 len = sprintf(buf, "%s", ceph_sub_str[i]);
283 if (i == CEPH_SUB_MDSMAP &&
284 monc->fs_cluster_id != CEPH_FS_CLUSTER_ID_NONE)
285 len += sprintf(buf + len, ".%d", monc->fs_cluster_id);
286
287 dout("%s %s start %llu flags 0x%x\n", __func__, buf,
288 le64_to_cpu(monc->subs[i].item.start),
289 monc->subs[i].item.flags);
290 ceph_encode_string(&p, end, buf, len);
291 memcpy(p, &monc->subs[i].item, sizeof(monc->subs[i].item));
292 p += sizeof(monc->subs[i].item);
293 }
294
295 BUG_ON(p > end);
296 msg->front.iov_len = p - msg->front.iov_base;
297 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
298 ceph_msg_revoke(msg);
299 ceph_con_send(&monc->con, ceph_msg_get(msg));
300}
301
302static void handle_subscribe_ack(struct ceph_mon_client *monc,
303 struct ceph_msg *msg)
304{
305 unsigned int seconds;
306 struct ceph_mon_subscribe_ack *h = msg->front.iov_base;
307
308 if (msg->front.iov_len < sizeof(*h))
309 goto bad;
310 seconds = le32_to_cpu(h->duration);
311
312 mutex_lock(&monc->mutex);
313 if (monc->sub_renew_sent) {
314
315
316
317
318 monc->sub_renew_after = monc->sub_renew_sent +
319 (seconds >> 1) * HZ - 1;
320 dout("%s sent %lu duration %d renew after %lu\n", __func__,
321 monc->sub_renew_sent, seconds, monc->sub_renew_after);
322 monc->sub_renew_sent = 0;
323 } else {
324 dout("%s sent %lu renew after %lu, ignoring\n", __func__,
325 monc->sub_renew_sent, monc->sub_renew_after);
326 }
327 mutex_unlock(&monc->mutex);
328 return;
329bad:
330 pr_err("got corrupt subscribe-ack msg\n");
331 ceph_msg_dump(msg);
332}
333
334
335
336
337
338
339
340static bool __ceph_monc_want_map(struct ceph_mon_client *monc, int sub,
341 u32 epoch, bool continuous)
342{
343 __le64 start = cpu_to_le64(epoch);
344 u8 flags = !continuous ? CEPH_SUBSCRIBE_ONETIME : 0;
345
346 dout("%s %s epoch %u continuous %d\n", __func__, ceph_sub_str[sub],
347 epoch, continuous);
348
349 if (monc->subs[sub].want &&
350 monc->subs[sub].item.start == start &&
351 monc->subs[sub].item.flags == flags)
352 return false;
353
354 monc->subs[sub].item.start = start;
355 monc->subs[sub].item.flags = flags;
356 monc->subs[sub].want = true;
357
358 return true;
359}
360
361bool ceph_monc_want_map(struct ceph_mon_client *monc, int sub, u32 epoch,
362 bool continuous)
363{
364 bool need_request;
365
366 mutex_lock(&monc->mutex);
367 need_request = __ceph_monc_want_map(monc, sub, epoch, continuous);
368 mutex_unlock(&monc->mutex);
369
370 return need_request;
371}
372EXPORT_SYMBOL(ceph_monc_want_map);
373
374
375
376
377
378
379static void __ceph_monc_got_map(struct ceph_mon_client *monc, int sub,
380 u32 epoch)
381{
382 dout("%s %s epoch %u\n", __func__, ceph_sub_str[sub], epoch);
383
384 if (monc->subs[sub].want) {
385 if (monc->subs[sub].item.flags & CEPH_SUBSCRIBE_ONETIME)
386 monc->subs[sub].want = false;
387 else
388 monc->subs[sub].item.start = cpu_to_le64(epoch + 1);
389 }
390
391 monc->subs[sub].have = epoch;
392}
393
394void ceph_monc_got_map(struct ceph_mon_client *monc, int sub, u32 epoch)
395{
396 mutex_lock(&monc->mutex);
397 __ceph_monc_got_map(monc, sub, epoch);
398 mutex_unlock(&monc->mutex);
399}
400EXPORT_SYMBOL(ceph_monc_got_map);
401
402void ceph_monc_renew_subs(struct ceph_mon_client *monc)
403{
404 mutex_lock(&monc->mutex);
405 __send_subscribe(monc);
406 mutex_unlock(&monc->mutex);
407}
408EXPORT_SYMBOL(ceph_monc_renew_subs);
409
410
411
412
413
414
415
416int ceph_monc_wait_osdmap(struct ceph_mon_client *monc, u32 epoch,
417 unsigned long timeout)
418{
419 unsigned long started = jiffies;
420 long ret;
421
422 mutex_lock(&monc->mutex);
423 while (monc->subs[CEPH_SUB_OSDMAP].have < epoch) {
424 mutex_unlock(&monc->mutex);
425
426 if (timeout && time_after_eq(jiffies, started + timeout))
427 return -ETIMEDOUT;
428
429 ret = wait_event_interruptible_timeout(monc->client->auth_wq,
430 monc->subs[CEPH_SUB_OSDMAP].have >= epoch,
431 ceph_timeout_jiffies(timeout));
432 if (ret < 0)
433 return ret;
434
435 mutex_lock(&monc->mutex);
436 }
437
438 mutex_unlock(&monc->mutex);
439 return 0;
440}
441EXPORT_SYMBOL(ceph_monc_wait_osdmap);
442
443
444
445
446
447int ceph_monc_open_session(struct ceph_mon_client *monc)
448{
449 mutex_lock(&monc->mutex);
450 __ceph_monc_want_map(monc, CEPH_SUB_MONMAP, 0, true);
451 __ceph_monc_want_map(monc, CEPH_SUB_OSDMAP, 0, false);
452 __open_session(monc);
453 __schedule_delayed(monc);
454 mutex_unlock(&monc->mutex);
455 return 0;
456}
457EXPORT_SYMBOL(ceph_monc_open_session);
458
459static void ceph_monc_handle_map(struct ceph_mon_client *monc,
460 struct ceph_msg *msg)
461{
462 struct ceph_client *client = monc->client;
463 struct ceph_monmap *monmap = NULL, *old = monc->monmap;
464 void *p, *end;
465
466 mutex_lock(&monc->mutex);
467
468 dout("handle_monmap\n");
469 p = msg->front.iov_base;
470 end = p + msg->front.iov_len;
471
472 monmap = ceph_monmap_decode(p, end);
473 if (IS_ERR(monmap)) {
474 pr_err("problem decoding monmap, %d\n",
475 (int)PTR_ERR(monmap));
476 ceph_msg_dump(msg);
477 goto out;
478 }
479
480 if (ceph_check_fsid(monc->client, &monmap->fsid) < 0) {
481 kfree(monmap);
482 goto out;
483 }
484
485 client->monc.monmap = monmap;
486 kfree(old);
487
488 __ceph_monc_got_map(monc, CEPH_SUB_MONMAP, monc->monmap->epoch);
489 client->have_fsid = true;
490
491out:
492 mutex_unlock(&monc->mutex);
493 wake_up_all(&client->auth_wq);
494}
495
496
497
498
499DEFINE_RB_FUNCS(generic_request, struct ceph_mon_generic_request, tid, node)
500
501static void release_generic_request(struct kref *kref)
502{
503 struct ceph_mon_generic_request *req =
504 container_of(kref, struct ceph_mon_generic_request, kref);
505
506 dout("%s greq %p request %p reply %p\n", __func__, req, req->request,
507 req->reply);
508 WARN_ON(!RB_EMPTY_NODE(&req->node));
509
510 if (req->reply)
511 ceph_msg_put(req->reply);
512 if (req->request)
513 ceph_msg_put(req->request);
514
515 kfree(req);
516}
517
518static void put_generic_request(struct ceph_mon_generic_request *req)
519{
520 if (req)
521 kref_put(&req->kref, release_generic_request);
522}
523
524static void get_generic_request(struct ceph_mon_generic_request *req)
525{
526 kref_get(&req->kref);
527}
528
529static struct ceph_mon_generic_request *
530alloc_generic_request(struct ceph_mon_client *monc, gfp_t gfp)
531{
532 struct ceph_mon_generic_request *req;
533
534 req = kzalloc(sizeof(*req), gfp);
535 if (!req)
536 return NULL;
537
538 req->monc = monc;
539 kref_init(&req->kref);
540 RB_CLEAR_NODE(&req->node);
541 init_completion(&req->completion);
542
543 dout("%s greq %p\n", __func__, req);
544 return req;
545}
546
547static void register_generic_request(struct ceph_mon_generic_request *req)
548{
549 struct ceph_mon_client *monc = req->monc;
550
551 WARN_ON(req->tid);
552
553 get_generic_request(req);
554 req->tid = ++monc->last_tid;
555 insert_generic_request(&monc->generic_request_tree, req);
556}
557
558static void send_generic_request(struct ceph_mon_client *monc,
559 struct ceph_mon_generic_request *req)
560{
561 WARN_ON(!req->tid);
562
563 dout("%s greq %p tid %llu\n", __func__, req, req->tid);
564 req->request->hdr.tid = cpu_to_le64(req->tid);
565 ceph_con_send(&monc->con, ceph_msg_get(req->request));
566}
567
568static void __finish_generic_request(struct ceph_mon_generic_request *req)
569{
570 struct ceph_mon_client *monc = req->monc;
571
572 dout("%s greq %p tid %llu\n", __func__, req, req->tid);
573 erase_generic_request(&monc->generic_request_tree, req);
574
575 ceph_msg_revoke(req->request);
576 ceph_msg_revoke_incoming(req->reply);
577}
578
579static void finish_generic_request(struct ceph_mon_generic_request *req)
580{
581 __finish_generic_request(req);
582 put_generic_request(req);
583}
584
585static void complete_generic_request(struct ceph_mon_generic_request *req)
586{
587 if (req->complete_cb)
588 req->complete_cb(req);
589 else
590 complete_all(&req->completion);
591 put_generic_request(req);
592}
593
594static void cancel_generic_request(struct ceph_mon_generic_request *req)
595{
596 struct ceph_mon_client *monc = req->monc;
597 struct ceph_mon_generic_request *lookup_req;
598
599 dout("%s greq %p tid %llu\n", __func__, req, req->tid);
600
601 mutex_lock(&monc->mutex);
602 lookup_req = lookup_generic_request(&monc->generic_request_tree,
603 req->tid);
604 if (lookup_req) {
605 WARN_ON(lookup_req != req);
606 finish_generic_request(req);
607 }
608
609 mutex_unlock(&monc->mutex);
610}
611
612static int wait_generic_request(struct ceph_mon_generic_request *req)
613{
614 int ret;
615
616 dout("%s greq %p tid %llu\n", __func__, req, req->tid);
617 ret = wait_for_completion_interruptible(&req->completion);
618 if (ret)
619 cancel_generic_request(req);
620 else
621 ret = req->result;
622
623 return ret;
624}
625
626static struct ceph_msg *get_generic_reply(struct ceph_connection *con,
627 struct ceph_msg_header *hdr,
628 int *skip)
629{
630 struct ceph_mon_client *monc = con->private;
631 struct ceph_mon_generic_request *req;
632 u64 tid = le64_to_cpu(hdr->tid);
633 struct ceph_msg *m;
634
635 mutex_lock(&monc->mutex);
636 req = lookup_generic_request(&monc->generic_request_tree, tid);
637 if (!req) {
638 dout("get_generic_reply %lld dne\n", tid);
639 *skip = 1;
640 m = NULL;
641 } else {
642 dout("get_generic_reply %lld got %p\n", tid, req->reply);
643 *skip = 0;
644 m = ceph_msg_get(req->reply);
645
646
647
648
649
650 }
651 mutex_unlock(&monc->mutex);
652 return m;
653}
654
655
656
657
658static void handle_statfs_reply(struct ceph_mon_client *monc,
659 struct ceph_msg *msg)
660{
661 struct ceph_mon_generic_request *req;
662 struct ceph_mon_statfs_reply *reply = msg->front.iov_base;
663 u64 tid = le64_to_cpu(msg->hdr.tid);
664
665 dout("%s msg %p tid %llu\n", __func__, msg, tid);
666
667 if (msg->front.iov_len != sizeof(*reply))
668 goto bad;
669
670 mutex_lock(&monc->mutex);
671 req = lookup_generic_request(&monc->generic_request_tree, tid);
672 if (!req) {
673 mutex_unlock(&monc->mutex);
674 return;
675 }
676
677 req->result = 0;
678 *req->u.st = reply->st;
679 __finish_generic_request(req);
680 mutex_unlock(&monc->mutex);
681
682 complete_generic_request(req);
683 return;
684
685bad:
686 pr_err("corrupt statfs reply, tid %llu\n", tid);
687 ceph_msg_dump(msg);
688}
689
690
691
692
693int ceph_monc_do_statfs(struct ceph_mon_client *monc, u64 data_pool,
694 struct ceph_statfs *buf)
695{
696 struct ceph_mon_generic_request *req;
697 struct ceph_mon_statfs *h;
698 int ret = -ENOMEM;
699
700 req = alloc_generic_request(monc, GFP_NOFS);
701 if (!req)
702 goto out;
703
704 req->request = ceph_msg_new(CEPH_MSG_STATFS, sizeof(*h), GFP_NOFS,
705 true);
706 if (!req->request)
707 goto out;
708
709 req->reply = ceph_msg_new(CEPH_MSG_STATFS_REPLY, 64, GFP_NOFS, true);
710 if (!req->reply)
711 goto out;
712
713 req->u.st = buf;
714 req->request->hdr.version = cpu_to_le16(2);
715
716 mutex_lock(&monc->mutex);
717 register_generic_request(req);
718
719 h = req->request->front.iov_base;
720 h->monhdr.have_version = 0;
721 h->monhdr.session_mon = cpu_to_le16(-1);
722 h->monhdr.session_mon_tid = 0;
723 h->fsid = monc->monmap->fsid;
724 h->contains_data_pool = (data_pool != CEPH_NOPOOL);
725 h->data_pool = cpu_to_le64(data_pool);
726 send_generic_request(monc, req);
727 mutex_unlock(&monc->mutex);
728
729 ret = wait_generic_request(req);
730out:
731 put_generic_request(req);
732 return ret;
733}
734EXPORT_SYMBOL(ceph_monc_do_statfs);
735
736static void handle_get_version_reply(struct ceph_mon_client *monc,
737 struct ceph_msg *msg)
738{
739 struct ceph_mon_generic_request *req;
740 u64 tid = le64_to_cpu(msg->hdr.tid);
741 void *p = msg->front.iov_base;
742 void *end = p + msg->front_alloc_len;
743 u64 handle;
744
745 dout("%s msg %p tid %llu\n", __func__, msg, tid);
746
747 ceph_decode_need(&p, end, 2*sizeof(u64), bad);
748 handle = ceph_decode_64(&p);
749 if (tid != 0 && tid != handle)
750 goto bad;
751
752 mutex_lock(&monc->mutex);
753 req = lookup_generic_request(&monc->generic_request_tree, handle);
754 if (!req) {
755 mutex_unlock(&monc->mutex);
756 return;
757 }
758
759 req->result = 0;
760 req->u.newest = ceph_decode_64(&p);
761 __finish_generic_request(req);
762 mutex_unlock(&monc->mutex);
763
764 complete_generic_request(req);
765 return;
766
767bad:
768 pr_err("corrupt mon_get_version reply, tid %llu\n", tid);
769 ceph_msg_dump(msg);
770}
771
772static struct ceph_mon_generic_request *
773__ceph_monc_get_version(struct ceph_mon_client *monc, const char *what,
774 ceph_monc_callback_t cb, u64 private_data)
775{
776 struct ceph_mon_generic_request *req;
777
778 req = alloc_generic_request(monc, GFP_NOIO);
779 if (!req)
780 goto err_put_req;
781
782 req->request = ceph_msg_new(CEPH_MSG_MON_GET_VERSION,
783 sizeof(u64) + sizeof(u32) + strlen(what),
784 GFP_NOIO, true);
785 if (!req->request)
786 goto err_put_req;
787
788 req->reply = ceph_msg_new(CEPH_MSG_MON_GET_VERSION_REPLY, 32, GFP_NOIO,
789 true);
790 if (!req->reply)
791 goto err_put_req;
792
793 req->complete_cb = cb;
794 req->private_data = private_data;
795
796 mutex_lock(&monc->mutex);
797 register_generic_request(req);
798 {
799 void *p = req->request->front.iov_base;
800 void *const end = p + req->request->front_alloc_len;
801
802 ceph_encode_64(&p, req->tid);
803 ceph_encode_string(&p, end, what, strlen(what));
804 WARN_ON(p != end);
805 }
806 send_generic_request(monc, req);
807 mutex_unlock(&monc->mutex);
808
809 return req;
810
811err_put_req:
812 put_generic_request(req);
813 return ERR_PTR(-ENOMEM);
814}
815
816
817
818
819
820
821int ceph_monc_get_version(struct ceph_mon_client *monc, const char *what,
822 u64 *newest)
823{
824 struct ceph_mon_generic_request *req;
825 int ret;
826
827 req = __ceph_monc_get_version(monc, what, NULL, 0);
828 if (IS_ERR(req))
829 return PTR_ERR(req);
830
831 ret = wait_generic_request(req);
832 if (!ret)
833 *newest = req->u.newest;
834
835 put_generic_request(req);
836 return ret;
837}
838EXPORT_SYMBOL(ceph_monc_get_version);
839
840
841
842
843
844
845int ceph_monc_get_version_async(struct ceph_mon_client *monc, const char *what,
846 ceph_monc_callback_t cb, u64 private_data)
847{
848 struct ceph_mon_generic_request *req;
849
850 req = __ceph_monc_get_version(monc, what, cb, private_data);
851 if (IS_ERR(req))
852 return PTR_ERR(req);
853
854 put_generic_request(req);
855 return 0;
856}
857EXPORT_SYMBOL(ceph_monc_get_version_async);
858
859static void handle_command_ack(struct ceph_mon_client *monc,
860 struct ceph_msg *msg)
861{
862 struct ceph_mon_generic_request *req;
863 void *p = msg->front.iov_base;
864 void *const end = p + msg->front_alloc_len;
865 u64 tid = le64_to_cpu(msg->hdr.tid);
866
867 dout("%s msg %p tid %llu\n", __func__, msg, tid);
868
869 ceph_decode_need(&p, end, sizeof(struct ceph_mon_request_header) +
870 sizeof(u32), bad);
871 p += sizeof(struct ceph_mon_request_header);
872
873 mutex_lock(&monc->mutex);
874 req = lookup_generic_request(&monc->generic_request_tree, tid);
875 if (!req) {
876 mutex_unlock(&monc->mutex);
877 return;
878 }
879
880 req->result = ceph_decode_32(&p);
881 __finish_generic_request(req);
882 mutex_unlock(&monc->mutex);
883
884 complete_generic_request(req);
885 return;
886
887bad:
888 pr_err("corrupt mon_command ack, tid %llu\n", tid);
889 ceph_msg_dump(msg);
890}
891
892int ceph_monc_blacklist_add(struct ceph_mon_client *monc,
893 struct ceph_entity_addr *client_addr)
894{
895 struct ceph_mon_generic_request *req;
896 struct ceph_mon_command *h;
897 int ret = -ENOMEM;
898 int len;
899
900 req = alloc_generic_request(monc, GFP_NOIO);
901 if (!req)
902 goto out;
903
904 req->request = ceph_msg_new(CEPH_MSG_MON_COMMAND, 256, GFP_NOIO, true);
905 if (!req->request)
906 goto out;
907
908 req->reply = ceph_msg_new(CEPH_MSG_MON_COMMAND_ACK, 512, GFP_NOIO,
909 true);
910 if (!req->reply)
911 goto out;
912
913 mutex_lock(&monc->mutex);
914 register_generic_request(req);
915 h = req->request->front.iov_base;
916 h->monhdr.have_version = 0;
917 h->monhdr.session_mon = cpu_to_le16(-1);
918 h->monhdr.session_mon_tid = 0;
919 h->fsid = monc->monmap->fsid;
920 h->num_strs = cpu_to_le32(1);
921 len = sprintf(h->str, "{ \"prefix\": \"osd blacklist\", \
922 \"blacklistop\": \"add\", \
923 \"addr\": \"%pISpc/%u\" }",
924 &client_addr->in_addr, le32_to_cpu(client_addr->nonce));
925 h->str_len = cpu_to_le32(len);
926 send_generic_request(monc, req);
927 mutex_unlock(&monc->mutex);
928
929 ret = wait_generic_request(req);
930 if (!ret)
931
932
933
934
935
936
937 ret = ceph_wait_for_latest_osdmap(monc->client, 0);
938
939out:
940 put_generic_request(req);
941 return ret;
942}
943EXPORT_SYMBOL(ceph_monc_blacklist_add);
944
945
946
947
948static void __resend_generic_request(struct ceph_mon_client *monc)
949{
950 struct ceph_mon_generic_request *req;
951 struct rb_node *p;
952
953 for (p = rb_first(&monc->generic_request_tree); p; p = rb_next(p)) {
954 req = rb_entry(p, struct ceph_mon_generic_request, node);
955 ceph_msg_revoke(req->request);
956 ceph_msg_revoke_incoming(req->reply);
957 ceph_con_send(&monc->con, ceph_msg_get(req->request));
958 }
959}
960
961
962
963
964
965
966static void delayed_work(struct work_struct *work)
967{
968 struct ceph_mon_client *monc =
969 container_of(work, struct ceph_mon_client, delayed_work.work);
970
971 dout("monc delayed_work\n");
972 mutex_lock(&monc->mutex);
973 if (monc->hunting) {
974 dout("%s continuing hunt\n", __func__);
975 reopen_session(monc);
976 } else {
977 int is_auth = ceph_auth_is_authenticated(monc->auth);
978 if (ceph_con_keepalive_expired(&monc->con,
979 CEPH_MONC_PING_TIMEOUT)) {
980 dout("monc keepalive timeout\n");
981 is_auth = 0;
982 reopen_session(monc);
983 }
984
985 if (!monc->hunting) {
986 ceph_con_keepalive(&monc->con);
987 __validate_auth(monc);
988 un_backoff(monc);
989 }
990
991 if (is_auth &&
992 !(monc->con.peer_features & CEPH_FEATURE_MON_STATEFUL_SUB)) {
993 unsigned long now = jiffies;
994
995 dout("%s renew subs? now %lu renew after %lu\n",
996 __func__, now, monc->sub_renew_after);
997 if (time_after_eq(now, monc->sub_renew_after))
998 __send_subscribe(monc);
999 }
1000 }
1001 __schedule_delayed(monc);
1002 mutex_unlock(&monc->mutex);
1003}
1004
1005
1006
1007
1008
1009static int build_initial_monmap(struct ceph_mon_client *monc)
1010{
1011 struct ceph_options *opt = monc->client->options;
1012 struct ceph_entity_addr *mon_addr = opt->mon_addr;
1013 int num_mon = opt->num_mon;
1014 int i;
1015
1016
1017 monc->monmap = kzalloc(struct_size(monc->monmap, mon_inst, num_mon),
1018 GFP_KERNEL);
1019 if (!monc->monmap)
1020 return -ENOMEM;
1021 for (i = 0; i < num_mon; i++) {
1022 monc->monmap->mon_inst[i].addr = mon_addr[i];
1023 monc->monmap->mon_inst[i].addr.nonce = 0;
1024 monc->monmap->mon_inst[i].name.type =
1025 CEPH_ENTITY_TYPE_MON;
1026 monc->monmap->mon_inst[i].name.num = cpu_to_le64(i);
1027 }
1028 monc->monmap->num_mon = num_mon;
1029 return 0;
1030}
1031
1032int ceph_monc_init(struct ceph_mon_client *monc, struct ceph_client *cl)
1033{
1034 int err = 0;
1035
1036 dout("init\n");
1037 memset(monc, 0, sizeof(*monc));
1038 monc->client = cl;
1039 monc->monmap = NULL;
1040 mutex_init(&monc->mutex);
1041
1042 err = build_initial_monmap(monc);
1043 if (err)
1044 goto out;
1045
1046
1047
1048 monc->auth = ceph_auth_init(cl->options->name,
1049 cl->options->key);
1050 if (IS_ERR(monc->auth)) {
1051 err = PTR_ERR(monc->auth);
1052 goto out_monmap;
1053 }
1054 monc->auth->want_keys =
1055 CEPH_ENTITY_TYPE_AUTH | CEPH_ENTITY_TYPE_MON |
1056 CEPH_ENTITY_TYPE_OSD | CEPH_ENTITY_TYPE_MDS;
1057
1058
1059 err = -ENOMEM;
1060 monc->m_subscribe_ack = ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE_ACK,
1061 sizeof(struct ceph_mon_subscribe_ack),
1062 GFP_KERNEL, true);
1063 if (!monc->m_subscribe_ack)
1064 goto out_auth;
1065
1066 monc->m_subscribe = ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE, 128,
1067 GFP_KERNEL, true);
1068 if (!monc->m_subscribe)
1069 goto out_subscribe_ack;
1070
1071 monc->m_auth_reply = ceph_msg_new(CEPH_MSG_AUTH_REPLY, 4096,
1072 GFP_KERNEL, true);
1073 if (!monc->m_auth_reply)
1074 goto out_subscribe;
1075
1076 monc->m_auth = ceph_msg_new(CEPH_MSG_AUTH, 4096, GFP_KERNEL, true);
1077 monc->pending_auth = 0;
1078 if (!monc->m_auth)
1079 goto out_auth_reply;
1080
1081 ceph_con_init(&monc->con, monc, &mon_con_ops,
1082 &monc->client->msgr);
1083
1084 monc->cur_mon = -1;
1085 monc->had_a_connection = false;
1086 monc->hunt_mult = 1;
1087
1088 INIT_DELAYED_WORK(&monc->delayed_work, delayed_work);
1089 monc->generic_request_tree = RB_ROOT;
1090 monc->last_tid = 0;
1091
1092 monc->fs_cluster_id = CEPH_FS_CLUSTER_ID_NONE;
1093
1094 return 0;
1095
1096out_auth_reply:
1097 ceph_msg_put(monc->m_auth_reply);
1098out_subscribe:
1099 ceph_msg_put(monc->m_subscribe);
1100out_subscribe_ack:
1101 ceph_msg_put(monc->m_subscribe_ack);
1102out_auth:
1103 ceph_auth_destroy(monc->auth);
1104out_monmap:
1105 kfree(monc->monmap);
1106out:
1107 return err;
1108}
1109EXPORT_SYMBOL(ceph_monc_init);
1110
1111void ceph_monc_stop(struct ceph_mon_client *monc)
1112{
1113 dout("stop\n");
1114 cancel_delayed_work_sync(&monc->delayed_work);
1115
1116 mutex_lock(&monc->mutex);
1117 __close_session(monc);
1118 monc->cur_mon = -1;
1119 mutex_unlock(&monc->mutex);
1120
1121
1122
1123
1124
1125
1126
1127 ceph_msgr_flush();
1128
1129 ceph_auth_destroy(monc->auth);
1130
1131 WARN_ON(!RB_EMPTY_ROOT(&monc->generic_request_tree));
1132
1133 ceph_msg_put(monc->m_auth);
1134 ceph_msg_put(monc->m_auth_reply);
1135 ceph_msg_put(monc->m_subscribe);
1136 ceph_msg_put(monc->m_subscribe_ack);
1137
1138 kfree(monc->monmap);
1139}
1140EXPORT_SYMBOL(ceph_monc_stop);
1141
1142static void finish_hunting(struct ceph_mon_client *monc)
1143{
1144 if (monc->hunting) {
1145 dout("%s found mon%d\n", __func__, monc->cur_mon);
1146 monc->hunting = false;
1147 monc->had_a_connection = true;
1148 un_backoff(monc);
1149 __schedule_delayed(monc);
1150 }
1151}
1152
1153static void handle_auth_reply(struct ceph_mon_client *monc,
1154 struct ceph_msg *msg)
1155{
1156 int ret;
1157 int was_auth = 0;
1158
1159 mutex_lock(&monc->mutex);
1160 was_auth = ceph_auth_is_authenticated(monc->auth);
1161 monc->pending_auth = 0;
1162 ret = ceph_handle_auth_reply(monc->auth, msg->front.iov_base,
1163 msg->front.iov_len,
1164 monc->m_auth->front.iov_base,
1165 monc->m_auth->front_alloc_len);
1166 if (ret > 0) {
1167 __send_prepared_auth_request(monc, ret);
1168 goto out;
1169 }
1170
1171 finish_hunting(monc);
1172
1173 if (ret < 0) {
1174 monc->client->auth_err = ret;
1175 } else if (!was_auth && ceph_auth_is_authenticated(monc->auth)) {
1176 dout("authenticated, starting session\n");
1177
1178 monc->client->msgr.inst.name.type = CEPH_ENTITY_TYPE_CLIENT;
1179 monc->client->msgr.inst.name.num =
1180 cpu_to_le64(monc->auth->global_id);
1181
1182 __send_subscribe(monc);
1183 __resend_generic_request(monc);
1184
1185 pr_info("mon%d %s session established\n", monc->cur_mon,
1186 ceph_pr_addr(&monc->con.peer_addr));
1187 }
1188
1189out:
1190 mutex_unlock(&monc->mutex);
1191 if (monc->client->auth_err < 0)
1192 wake_up_all(&monc->client->auth_wq);
1193}
1194
1195static int __validate_auth(struct ceph_mon_client *monc)
1196{
1197 int ret;
1198
1199 if (monc->pending_auth)
1200 return 0;
1201
1202 ret = ceph_build_auth(monc->auth, monc->m_auth->front.iov_base,
1203 monc->m_auth->front_alloc_len);
1204 if (ret <= 0)
1205 return ret;
1206 __send_prepared_auth_request(monc, ret);
1207 return 0;
1208}
1209
1210int ceph_monc_validate_auth(struct ceph_mon_client *monc)
1211{
1212 int ret;
1213
1214 mutex_lock(&monc->mutex);
1215 ret = __validate_auth(monc);
1216 mutex_unlock(&monc->mutex);
1217 return ret;
1218}
1219EXPORT_SYMBOL(ceph_monc_validate_auth);
1220
1221
1222
1223
1224static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
1225{
1226 struct ceph_mon_client *monc = con->private;
1227 int type = le16_to_cpu(msg->hdr.type);
1228
1229 if (!monc)
1230 return;
1231
1232 switch (type) {
1233 case CEPH_MSG_AUTH_REPLY:
1234 handle_auth_reply(monc, msg);
1235 break;
1236
1237 case CEPH_MSG_MON_SUBSCRIBE_ACK:
1238 handle_subscribe_ack(monc, msg);
1239 break;
1240
1241 case CEPH_MSG_STATFS_REPLY:
1242 handle_statfs_reply(monc, msg);
1243 break;
1244
1245 case CEPH_MSG_MON_GET_VERSION_REPLY:
1246 handle_get_version_reply(monc, msg);
1247 break;
1248
1249 case CEPH_MSG_MON_COMMAND_ACK:
1250 handle_command_ack(monc, msg);
1251 break;
1252
1253 case CEPH_MSG_MON_MAP:
1254 ceph_monc_handle_map(monc, msg);
1255 break;
1256
1257 case CEPH_MSG_OSD_MAP:
1258 ceph_osdc_handle_map(&monc->client->osdc, msg);
1259 break;
1260
1261 default:
1262
1263 if (monc->client->extra_mon_dispatch &&
1264 monc->client->extra_mon_dispatch(monc->client, msg) == 0)
1265 break;
1266
1267 pr_err("received unknown message type %d %s\n", type,
1268 ceph_msg_type_name(type));
1269 }
1270 ceph_msg_put(msg);
1271}
1272
1273
1274
1275
1276static struct ceph_msg *mon_alloc_msg(struct ceph_connection *con,
1277 struct ceph_msg_header *hdr,
1278 int *skip)
1279{
1280 struct ceph_mon_client *monc = con->private;
1281 int type = le16_to_cpu(hdr->type);
1282 int front_len = le32_to_cpu(hdr->front_len);
1283 struct ceph_msg *m = NULL;
1284
1285 *skip = 0;
1286
1287 switch (type) {
1288 case CEPH_MSG_MON_SUBSCRIBE_ACK:
1289 m = ceph_msg_get(monc->m_subscribe_ack);
1290 break;
1291 case CEPH_MSG_STATFS_REPLY:
1292 case CEPH_MSG_MON_COMMAND_ACK:
1293 return get_generic_reply(con, hdr, skip);
1294 case CEPH_MSG_AUTH_REPLY:
1295 m = ceph_msg_get(monc->m_auth_reply);
1296 break;
1297 case CEPH_MSG_MON_GET_VERSION_REPLY:
1298 if (le64_to_cpu(hdr->tid) != 0)
1299 return get_generic_reply(con, hdr, skip);
1300
1301
1302
1303
1304
1305
1306
1307 case CEPH_MSG_MON_MAP:
1308 case CEPH_MSG_MDS_MAP:
1309 case CEPH_MSG_OSD_MAP:
1310 case CEPH_MSG_FS_MAP_USER:
1311 m = ceph_msg_new(type, front_len, GFP_NOFS, false);
1312 if (!m)
1313 return NULL;
1314 break;
1315 }
1316
1317 if (!m) {
1318 pr_info("alloc_msg unknown type %d\n", type);
1319 *skip = 1;
1320 } else if (front_len > m->front_alloc_len) {
1321 pr_warn("mon_alloc_msg front %d > prealloc %d (%u#%llu)\n",
1322 front_len, m->front_alloc_len,
1323 (unsigned int)con->peer_name.type,
1324 le64_to_cpu(con->peer_name.num));
1325 ceph_msg_put(m);
1326 m = ceph_msg_new(type, front_len, GFP_NOFS, false);
1327 }
1328
1329 return m;
1330}
1331
1332
1333
1334
1335
1336static void mon_fault(struct ceph_connection *con)
1337{
1338 struct ceph_mon_client *monc = con->private;
1339
1340 mutex_lock(&monc->mutex);
1341 dout("%s mon%d\n", __func__, monc->cur_mon);
1342 if (monc->cur_mon >= 0) {
1343 if (!monc->hunting) {
1344 dout("%s hunting for new mon\n", __func__);
1345 reopen_session(monc);
1346 __schedule_delayed(monc);
1347 } else {
1348 dout("%s already hunting\n", __func__);
1349 }
1350 }
1351 mutex_unlock(&monc->mutex);
1352}
1353
1354
1355
1356
1357
1358
1359static struct ceph_connection *con_get(struct ceph_connection *con)
1360{
1361 return con;
1362}
1363
1364static void con_put(struct ceph_connection *con)
1365{
1366}
1367
1368static const struct ceph_connection_operations mon_con_ops = {
1369 .get = con_get,
1370 .put = con_put,
1371 .dispatch = dispatch,
1372 .fault = mon_fault,
1373 .alloc_msg = mon_alloc_msg,
1374};
1375