libceph: always preallocate mon connection
[linux-2.6.git] / net / ceph / mon_client.c
1 #include <linux/ceph/ceph_debug.h>
2
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>
8
9 #include <linux/ceph/mon_client.h>
10 #include <linux/ceph/libceph.h>
11 #include <linux/ceph/decode.h>
12
13 #include <linux/ceph/auth.h>
14
15 /*
16  * Interact with Ceph monitor cluster.  Handle requests for new map
17  * versions, and periodically resend as needed.  Also implement
18  * statfs() and umount().
19  *
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.
25  *
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.
31  */
32
33 static const struct ceph_connection_operations mon_con_ops;
34
35 static int __validate_auth(struct ceph_mon_client *monc);
36
37 /*
38  * Decode a monmap blob (e.g., during mount).
39  */
40 struct ceph_monmap *ceph_monmap_decode(void *p, void *end)
41 {
42         struct ceph_monmap *m = NULL;
43         int i, err = -EINVAL;
44         struct ceph_fsid fsid;
45         u32 epoch, num_mon;
46         u16 version;
47         u32 len;
48
49         ceph_decode_32_safe(&p, end, len, bad);
50         ceph_decode_need(&p, end, len, bad);
51
52         dout("monmap_decode %p %p len %d\n", p, end, (int)(end-p));
53
54         ceph_decode_16_safe(&p, end, version, bad);
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         ceph_decode_need(&p, end, num_mon*sizeof(m->mon_inst[0]), bad);
62
63         if (num_mon >= CEPH_MAX_MON)
64                 goto bad;
65         m = kmalloc(sizeof(*m) + sizeof(m->mon_inst[0])*num_mon, GFP_NOFS);
66         if (m == NULL)
67                 return ERR_PTR(-ENOMEM);
68         m->fsid = fsid;
69         m->epoch = epoch;
70         m->num_mon = num_mon;
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);
74
75         dout("monmap_decode epoch %d, num_mon %d\n", m->epoch,
76              m->num_mon);
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));
80         return m;
81
82 bad:
83         dout("monmap_decode failed with %d\n", err);
84         kfree(m);
85         return ERR_PTR(err);
86 }
87
88 /*
89  * return true if *addr is included in the monmap.
90  */
91 int ceph_monmap_contains(struct ceph_monmap *m, struct ceph_entity_addr *addr)
92 {
93         int i;
94
95         for (i = 0; i < m->num_mon; i++)
96                 if (memcmp(addr, &m->mon_inst[i].addr, sizeof(*addr)) == 0)
97                         return 1;
98         return 0;
99 }
100
101 /*
102  * Send an auth request.
103  */
104 static void __send_prepared_auth_request(struct ceph_mon_client *monc, int len)
105 {
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_con_revoke(monc->con, monc->m_auth);
110         ceph_msg_get(monc->m_auth);  /* keep our ref */
111         ceph_con_send(monc->con, monc->m_auth);
112 }
113
114 /*
115  * Close monitor session, if any.
116  */
117 static void __close_session(struct ceph_mon_client *monc)
118 {
119         dout("__close_session closing mon%d\n", monc->cur_mon);
120         ceph_con_revoke(monc->con, monc->m_auth);
121         ceph_con_close(monc->con);
122         monc->cur_mon = -1;
123         monc->pending_auth = 0;
124         ceph_auth_reset(monc->auth);
125 }
126
127 /*
128  * Open a session with a (new) monitor.
129  */
130 static int __open_session(struct ceph_mon_client *monc)
131 {
132         char r;
133         int ret;
134
135         if (monc->cur_mon < 0) {
136                 get_random_bytes(&r, 1);
137                 monc->cur_mon = r % monc->monmap->num_mon;
138                 dout("open_session num=%d r=%d -> mon%d\n",
139                      monc->monmap->num_mon, r, monc->cur_mon);
140                 monc->sub_sent = 0;
141                 monc->sub_renew_after = jiffies;  /* i.e., expired */
142                 monc->want_next_osdmap = !!monc->want_next_osdmap;
143
144                 dout("open_session mon%d opening\n", monc->cur_mon);
145                 monc->con->peer_name.type = CEPH_ENTITY_TYPE_MON;
146                 monc->con->peer_name.num = cpu_to_le64(monc->cur_mon);
147                 ceph_con_open(monc->con,
148                               &monc->monmap->mon_inst[monc->cur_mon].addr);
149
150                 /* initiatiate authentication handshake */
151                 ret = ceph_auth_build_hello(monc->auth,
152                                             monc->m_auth->front.iov_base,
153                                             monc->m_auth->front_max);
154                 __send_prepared_auth_request(monc, ret);
155         } else {
156                 dout("open_session mon%d already open\n", monc->cur_mon);
157         }
158         return 0;
159 }
160
161 static bool __sub_expired(struct ceph_mon_client *monc)
162 {
163         return time_after_eq(jiffies, monc->sub_renew_after);
164 }
165
166 /*
167  * Reschedule delayed work timer.
168  */
169 static void __schedule_delayed(struct ceph_mon_client *monc)
170 {
171         unsigned delay;
172
173         if (monc->cur_mon < 0 || __sub_expired(monc))
174                 delay = 10 * HZ;
175         else
176                 delay = 20 * HZ;
177         dout("__schedule_delayed after %u\n", delay);
178         schedule_delayed_work(&monc->delayed_work, delay);
179 }
180
181 /*
182  * Send subscribe request for mdsmap and/or osdmap.
183  */
184 static void __send_subscribe(struct ceph_mon_client *monc)
185 {
186         dout("__send_subscribe sub_sent=%u exp=%u want_osd=%d\n",
187              (unsigned)monc->sub_sent, __sub_expired(monc),
188              monc->want_next_osdmap);
189         if ((__sub_expired(monc) && !monc->sub_sent) ||
190             monc->want_next_osdmap == 1) {
191                 struct ceph_msg *msg = monc->m_subscribe;
192                 struct ceph_mon_subscribe_item *i;
193                 void *p, *end;
194                 int num;
195
196                 p = msg->front.iov_base;
197                 end = p + msg->front_max;
198
199                 num = 1 + !!monc->want_next_osdmap + !!monc->want_mdsmap;
200                 ceph_encode_32(&p, num);
201
202                 if (monc->want_next_osdmap) {
203                         dout("__send_subscribe to 'osdmap' %u\n",
204                              (unsigned)monc->have_osdmap);
205                         ceph_encode_string(&p, end, "osdmap", 6);
206                         i = p;
207                         i->have = cpu_to_le64(monc->have_osdmap);
208                         i->onetime = 1;
209                         p += sizeof(*i);
210                         monc->want_next_osdmap = 2;  /* requested */
211                 }
212                 if (monc->want_mdsmap) {
213                         dout("__send_subscribe to 'mdsmap' %u+\n",
214                              (unsigned)monc->have_mdsmap);
215                         ceph_encode_string(&p, end, "mdsmap", 6);
216                         i = p;
217                         i->have = cpu_to_le64(monc->have_mdsmap);
218                         i->onetime = 0;
219                         p += sizeof(*i);
220                 }
221                 ceph_encode_string(&p, end, "monmap", 6);
222                 i = p;
223                 i->have = 0;
224                 i->onetime = 0;
225                 p += sizeof(*i);
226
227                 msg->front.iov_len = p - msg->front.iov_base;
228                 msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
229                 ceph_con_revoke(monc->con, msg);
230                 ceph_con_send(monc->con, ceph_msg_get(msg));
231
232                 monc->sub_sent = jiffies | 1;  /* never 0 */
233         }
234 }
235
236 static void handle_subscribe_ack(struct ceph_mon_client *monc,
237                                  struct ceph_msg *msg)
238 {
239         unsigned seconds;
240         struct ceph_mon_subscribe_ack *h = msg->front.iov_base;
241
242         if (msg->front.iov_len < sizeof(*h))
243                 goto bad;
244         seconds = le32_to_cpu(h->duration);
245
246         mutex_lock(&monc->mutex);
247         if (monc->hunting) {
248                 pr_info("mon%d %s session established\n",
249                         monc->cur_mon,
250                         ceph_pr_addr(&monc->con->peer_addr.in_addr));
251                 monc->hunting = false;
252         }
253         dout("handle_subscribe_ack after %d seconds\n", seconds);
254         monc->sub_renew_after = monc->sub_sent + (seconds >> 1)*HZ - 1;
255         monc->sub_sent = 0;
256         mutex_unlock(&monc->mutex);
257         return;
258 bad:
259         pr_err("got corrupt subscribe-ack msg\n");
260         ceph_msg_dump(msg);
261 }
262
263 /*
264  * Keep track of which maps we have
265  */
266 int ceph_monc_got_mdsmap(struct ceph_mon_client *monc, u32 got)
267 {
268         mutex_lock(&monc->mutex);
269         monc->have_mdsmap = got;
270         mutex_unlock(&monc->mutex);
271         return 0;
272 }
273 EXPORT_SYMBOL(ceph_monc_got_mdsmap);
274
275 int ceph_monc_got_osdmap(struct ceph_mon_client *monc, u32 got)
276 {
277         mutex_lock(&monc->mutex);
278         monc->have_osdmap = got;
279         monc->want_next_osdmap = 0;
280         mutex_unlock(&monc->mutex);
281         return 0;
282 }
283
284 /*
285  * Register interest in the next osdmap
286  */
287 void ceph_monc_request_next_osdmap(struct ceph_mon_client *monc)
288 {
289         dout("request_next_osdmap have %u\n", monc->have_osdmap);
290         mutex_lock(&monc->mutex);
291         if (!monc->want_next_osdmap)
292                 monc->want_next_osdmap = 1;
293         if (monc->want_next_osdmap < 2)
294                 __send_subscribe(monc);
295         mutex_unlock(&monc->mutex);
296 }
297
298 /*
299  *
300  */
301 int ceph_monc_open_session(struct ceph_mon_client *monc)
302 {
303         mutex_lock(&monc->mutex);
304         __open_session(monc);
305         __schedule_delayed(monc);
306         mutex_unlock(&monc->mutex);
307         return 0;
308 }
309 EXPORT_SYMBOL(ceph_monc_open_session);
310
311 /*
312  * The monitor responds with mount ack indicate mount success.  The
313  * included client ticket allows the client to talk to MDSs and OSDs.
314  */
315 static void ceph_monc_handle_map(struct ceph_mon_client *monc,
316                                  struct ceph_msg *msg)
317 {
318         struct ceph_client *client = monc->client;
319         struct ceph_monmap *monmap = NULL, *old = monc->monmap;
320         void *p, *end;
321
322         mutex_lock(&monc->mutex);
323
324         dout("handle_monmap\n");
325         p = msg->front.iov_base;
326         end = p + msg->front.iov_len;
327
328         monmap = ceph_monmap_decode(p, end);
329         if (IS_ERR(monmap)) {
330                 pr_err("problem decoding monmap, %d\n",
331                        (int)PTR_ERR(monmap));
332                 goto out;
333         }
334
335         if (ceph_check_fsid(monc->client, &monmap->fsid) < 0) {
336                 kfree(monmap);
337                 goto out;
338         }
339
340         client->monc.monmap = monmap;
341         kfree(old);
342
343 out:
344         mutex_unlock(&monc->mutex);
345         wake_up_all(&client->auth_wq);
346 }
347
348 /*
349  * generic requests (e.g., statfs, poolop)
350  */
351 static struct ceph_mon_generic_request *__lookup_generic_req(
352         struct ceph_mon_client *monc, u64 tid)
353 {
354         struct ceph_mon_generic_request *req;
355         struct rb_node *n = monc->generic_request_tree.rb_node;
356
357         while (n) {
358                 req = rb_entry(n, struct ceph_mon_generic_request, node);
359                 if (tid < req->tid)
360                         n = n->rb_left;
361                 else if (tid > req->tid)
362                         n = n->rb_right;
363                 else
364                         return req;
365         }
366         return NULL;
367 }
368
369 static void __insert_generic_request(struct ceph_mon_client *monc,
370                             struct ceph_mon_generic_request *new)
371 {
372         struct rb_node **p = &monc->generic_request_tree.rb_node;
373         struct rb_node *parent = NULL;
374         struct ceph_mon_generic_request *req = NULL;
375
376         while (*p) {
377                 parent = *p;
378                 req = rb_entry(parent, struct ceph_mon_generic_request, node);
379                 if (new->tid < req->tid)
380                         p = &(*p)->rb_left;
381                 else if (new->tid > req->tid)
382                         p = &(*p)->rb_right;
383                 else
384                         BUG();
385         }
386
387         rb_link_node(&new->node, parent, p);
388         rb_insert_color(&new->node, &monc->generic_request_tree);
389 }
390
391 static void release_generic_request(struct kref *kref)
392 {
393         struct ceph_mon_generic_request *req =
394                 container_of(kref, struct ceph_mon_generic_request, kref);
395
396         if (req->reply)
397                 ceph_msg_put(req->reply);
398         if (req->request)
399                 ceph_msg_put(req->request);
400
401         kfree(req);
402 }
403
404 static void put_generic_request(struct ceph_mon_generic_request *req)
405 {
406         kref_put(&req->kref, release_generic_request);
407 }
408
409 static void get_generic_request(struct ceph_mon_generic_request *req)
410 {
411         kref_get(&req->kref);
412 }
413
414 static struct ceph_msg *get_generic_reply(struct ceph_connection *con,
415                                          struct ceph_msg_header *hdr,
416                                          int *skip)
417 {
418         struct ceph_mon_client *monc = con->private;
419         struct ceph_mon_generic_request *req;
420         u64 tid = le64_to_cpu(hdr->tid);
421         struct ceph_msg *m;
422
423         mutex_lock(&monc->mutex);
424         req = __lookup_generic_req(monc, tid);
425         if (!req) {
426                 dout("get_generic_reply %lld dne\n", tid);
427                 *skip = 1;
428                 m = NULL;
429         } else {
430                 dout("get_generic_reply %lld got %p\n", tid, req->reply);
431                 m = ceph_msg_get(req->reply);
432                 /*
433                  * we don't need to track the connection reading into
434                  * this reply because we only have one open connection
435                  * at a time, ever.
436                  */
437         }
438         mutex_unlock(&monc->mutex);
439         return m;
440 }
441
442 static int do_generic_request(struct ceph_mon_client *monc,
443                               struct ceph_mon_generic_request *req)
444 {
445         int err;
446
447         /* register request */
448         mutex_lock(&monc->mutex);
449         req->tid = ++monc->last_tid;
450         req->request->hdr.tid = cpu_to_le64(req->tid);
451         __insert_generic_request(monc, req);
452         monc->num_generic_requests++;
453         ceph_con_send(monc->con, ceph_msg_get(req->request));
454         mutex_unlock(&monc->mutex);
455
456         err = wait_for_completion_interruptible(&req->completion);
457
458         mutex_lock(&monc->mutex);
459         rb_erase(&req->node, &monc->generic_request_tree);
460         monc->num_generic_requests--;
461         mutex_unlock(&monc->mutex);
462
463         if (!err)
464                 err = req->result;
465         return err;
466 }
467
468 /*
469  * statfs
470  */
471 static void handle_statfs_reply(struct ceph_mon_client *monc,
472                                 struct ceph_msg *msg)
473 {
474         struct ceph_mon_generic_request *req;
475         struct ceph_mon_statfs_reply *reply = msg->front.iov_base;
476         u64 tid = le64_to_cpu(msg->hdr.tid);
477
478         if (msg->front.iov_len != sizeof(*reply))
479                 goto bad;
480         dout("handle_statfs_reply %p tid %llu\n", msg, tid);
481
482         mutex_lock(&monc->mutex);
483         req = __lookup_generic_req(monc, tid);
484         if (req) {
485                 *(struct ceph_statfs *)req->buf = reply->st;
486                 req->result = 0;
487                 get_generic_request(req);
488         }
489         mutex_unlock(&monc->mutex);
490         if (req) {
491                 complete_all(&req->completion);
492                 put_generic_request(req);
493         }
494         return;
495
496 bad:
497         pr_err("corrupt generic reply, tid %llu\n", tid);
498         ceph_msg_dump(msg);
499 }
500
501 /*
502  * Do a synchronous statfs().
503  */
504 int ceph_monc_do_statfs(struct ceph_mon_client *monc, struct ceph_statfs *buf)
505 {
506         struct ceph_mon_generic_request *req;
507         struct ceph_mon_statfs *h;
508         int err;
509
510         req = kzalloc(sizeof(*req), GFP_NOFS);
511         if (!req)
512                 return -ENOMEM;
513
514         kref_init(&req->kref);
515         req->buf = buf;
516         req->buf_len = sizeof(*buf);
517         init_completion(&req->completion);
518
519         err = -ENOMEM;
520         req->request = ceph_msg_new(CEPH_MSG_STATFS, sizeof(*h), GFP_NOFS);
521         if (!req->request)
522                 goto out;
523         req->reply = ceph_msg_new(CEPH_MSG_STATFS_REPLY, 1024, GFP_NOFS);
524         if (!req->reply)
525                 goto out;
526
527         /* fill out request */
528         h = req->request->front.iov_base;
529         h->monhdr.have_version = 0;
530         h->monhdr.session_mon = cpu_to_le16(-1);
531         h->monhdr.session_mon_tid = 0;
532         h->fsid = monc->monmap->fsid;
533
534         err = do_generic_request(monc, req);
535
536 out:
537         kref_put(&req->kref, release_generic_request);
538         return err;
539 }
540 EXPORT_SYMBOL(ceph_monc_do_statfs);
541
542 /*
543  * pool ops
544  */
545 static int get_poolop_reply_buf(const char *src, size_t src_len,
546                                 char *dst, size_t dst_len)
547 {
548         u32 buf_len;
549
550         if (src_len != sizeof(u32) + dst_len)
551                 return -EINVAL;
552
553         buf_len = le32_to_cpu(*(u32 *)src);
554         if (buf_len != dst_len)
555                 return -EINVAL;
556
557         memcpy(dst, src + sizeof(u32), dst_len);
558         return 0;
559 }
560
561 static void handle_poolop_reply(struct ceph_mon_client *monc,
562                                 struct ceph_msg *msg)
563 {
564         struct ceph_mon_generic_request *req;
565         struct ceph_mon_poolop_reply *reply = msg->front.iov_base;
566         u64 tid = le64_to_cpu(msg->hdr.tid);
567
568         if (msg->front.iov_len < sizeof(*reply))
569                 goto bad;
570         dout("handle_poolop_reply %p tid %llu\n", msg, tid);
571
572         mutex_lock(&monc->mutex);
573         req = __lookup_generic_req(monc, tid);
574         if (req) {
575                 if (req->buf_len &&
576                     get_poolop_reply_buf(msg->front.iov_base + sizeof(*reply),
577                                      msg->front.iov_len - sizeof(*reply),
578                                      req->buf, req->buf_len) < 0) {
579                         mutex_unlock(&monc->mutex);
580                         goto bad;
581                 }
582                 req->result = le32_to_cpu(reply->reply_code);
583                 get_generic_request(req);
584         }
585         mutex_unlock(&monc->mutex);
586         if (req) {
587                 complete(&req->completion);
588                 put_generic_request(req);
589         }
590         return;
591
592 bad:
593         pr_err("corrupt generic reply, tid %llu\n", tid);
594         ceph_msg_dump(msg);
595 }
596
597 /*
598  * Do a synchronous pool op.
599  */
600 int ceph_monc_do_poolop(struct ceph_mon_client *monc, u32 op,
601                         u32 pool, u64 snapid,
602                         char *buf, int len)
603 {
604         struct ceph_mon_generic_request *req;
605         struct ceph_mon_poolop *h;
606         int err;
607
608         req = kzalloc(sizeof(*req), GFP_NOFS);
609         if (!req)
610                 return -ENOMEM;
611
612         kref_init(&req->kref);
613         req->buf = buf;
614         req->buf_len = len;
615         init_completion(&req->completion);
616
617         err = -ENOMEM;
618         req->request = ceph_msg_new(CEPH_MSG_POOLOP, sizeof(*h), GFP_NOFS);
619         if (!req->request)
620                 goto out;
621         req->reply = ceph_msg_new(CEPH_MSG_POOLOP_REPLY, 1024, GFP_NOFS);
622         if (!req->reply)
623                 goto out;
624
625         /* fill out request */
626         req->request->hdr.version = cpu_to_le16(2);
627         h = req->request->front.iov_base;
628         h->monhdr.have_version = 0;
629         h->monhdr.session_mon = cpu_to_le16(-1);
630         h->monhdr.session_mon_tid = 0;
631         h->fsid = monc->monmap->fsid;
632         h->pool = cpu_to_le32(pool);
633         h->op = cpu_to_le32(op);
634         h->auid = 0;
635         h->snapid = cpu_to_le64(snapid);
636         h->name_len = 0;
637
638         err = do_generic_request(monc, req);
639
640 out:
641         kref_put(&req->kref, release_generic_request);
642         return err;
643 }
644
645 int ceph_monc_create_snapid(struct ceph_mon_client *monc,
646                             u32 pool, u64 *snapid)
647 {
648         return ceph_monc_do_poolop(monc,  POOL_OP_CREATE_UNMANAGED_SNAP,
649                                    pool, 0, (char *)snapid, sizeof(*snapid));
650
651 }
652 EXPORT_SYMBOL(ceph_monc_create_snapid);
653
654 int ceph_monc_delete_snapid(struct ceph_mon_client *monc,
655                             u32 pool, u64 snapid)
656 {
657         return ceph_monc_do_poolop(monc,  POOL_OP_CREATE_UNMANAGED_SNAP,
658                                    pool, snapid, 0, 0);
659
660 }
661
662 /*
663  * Resend pending generic requests.
664  */
665 static void __resend_generic_request(struct ceph_mon_client *monc)
666 {
667         struct ceph_mon_generic_request *req;
668         struct rb_node *p;
669
670         for (p = rb_first(&monc->generic_request_tree); p; p = rb_next(p)) {
671                 req = rb_entry(p, struct ceph_mon_generic_request, node);
672                 ceph_con_revoke(monc->con, req->request);
673                 ceph_con_send(monc->con, ceph_msg_get(req->request));
674         }
675 }
676
677 /*
678  * Delayed work.  If we haven't mounted yet, retry.  Otherwise,
679  * renew/retry subscription as needed (in case it is timing out, or we
680  * got an ENOMEM).  And keep the monitor connection alive.
681  */
682 static void delayed_work(struct work_struct *work)
683 {
684         struct ceph_mon_client *monc =
685                 container_of(work, struct ceph_mon_client, delayed_work.work);
686
687         dout("monc delayed_work\n");
688         mutex_lock(&monc->mutex);
689         if (monc->hunting) {
690                 __close_session(monc);
691                 __open_session(monc);  /* continue hunting */
692         } else {
693                 ceph_con_keepalive(monc->con);
694
695                 __validate_auth(monc);
696
697                 if (monc->auth->ops->is_authenticated(monc->auth))
698                         __send_subscribe(monc);
699         }
700         __schedule_delayed(monc);
701         mutex_unlock(&monc->mutex);
702 }
703
704 /*
705  * On startup, we build a temporary monmap populated with the IPs
706  * provided by mount(2).
707  */
708 static int build_initial_monmap(struct ceph_mon_client *monc)
709 {
710         struct ceph_options *opt = monc->client->options;
711         struct ceph_entity_addr *mon_addr = opt->mon_addr;
712         int num_mon = opt->num_mon;
713         int i;
714
715         /* build initial monmap */
716         monc->monmap = kzalloc(sizeof(*monc->monmap) +
717                                num_mon*sizeof(monc->monmap->mon_inst[0]),
718                                GFP_KERNEL);
719         if (!monc->monmap)
720                 return -ENOMEM;
721         for (i = 0; i < num_mon; i++) {
722                 monc->monmap->mon_inst[i].addr = mon_addr[i];
723                 monc->monmap->mon_inst[i].addr.nonce = 0;
724                 monc->monmap->mon_inst[i].name.type =
725                         CEPH_ENTITY_TYPE_MON;
726                 monc->monmap->mon_inst[i].name.num = cpu_to_le64(i);
727         }
728         monc->monmap->num_mon = num_mon;
729         monc->have_fsid = false;
730         return 0;
731 }
732
733 int ceph_monc_init(struct ceph_mon_client *monc, struct ceph_client *cl)
734 {
735         int err = 0;
736
737         dout("init\n");
738         memset(monc, 0, sizeof(*monc));
739         monc->client = cl;
740         monc->monmap = NULL;
741         mutex_init(&monc->mutex);
742
743         err = build_initial_monmap(monc);
744         if (err)
745                 goto out;
746
747         /* connection */
748         monc->con = kmalloc(sizeof(*monc->con), GFP_KERNEL);
749         if (!monc->con)
750                 goto out_monmap;
751         ceph_con_init(monc->client->msgr, monc->con);
752         monc->con->private = monc;
753         monc->con->ops = &mon_con_ops;
754
755         /* authentication */
756         monc->auth = ceph_auth_init(cl->options->name,
757                                     cl->options->key);
758         if (IS_ERR(monc->auth))
759                 return PTR_ERR(monc->auth);
760         monc->auth->want_keys =
761                 CEPH_ENTITY_TYPE_AUTH | CEPH_ENTITY_TYPE_MON |
762                 CEPH_ENTITY_TYPE_OSD | CEPH_ENTITY_TYPE_MDS;
763
764         /* msgs */
765         err = -ENOMEM;
766         monc->m_subscribe_ack = ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE_ACK,
767                                      sizeof(struct ceph_mon_subscribe_ack),
768                                      GFP_NOFS);
769         if (!monc->m_subscribe_ack)
770                 goto out_con;
771
772         monc->m_subscribe = ceph_msg_new(CEPH_MSG_MON_SUBSCRIBE, 96, GFP_NOFS);
773         if (!monc->m_subscribe)
774                 goto out_subscribe_ack;
775
776         monc->m_auth_reply = ceph_msg_new(CEPH_MSG_AUTH_REPLY, 4096, GFP_NOFS);
777         if (!monc->m_auth_reply)
778                 goto out_subscribe;
779
780         monc->m_auth = ceph_msg_new(CEPH_MSG_AUTH, 4096, GFP_NOFS);
781         monc->pending_auth = 0;
782         if (!monc->m_auth)
783                 goto out_auth_reply;
784
785         monc->cur_mon = -1;
786         monc->hunting = true;
787         monc->sub_renew_after = jiffies;
788         monc->sub_sent = 0;
789
790         INIT_DELAYED_WORK(&monc->delayed_work, delayed_work);
791         monc->generic_request_tree = RB_ROOT;
792         monc->num_generic_requests = 0;
793         monc->last_tid = 0;
794
795         monc->have_mdsmap = 0;
796         monc->have_osdmap = 0;
797         monc->want_next_osdmap = 1;
798         return 0;
799
800 out_auth_reply:
801         ceph_msg_put(monc->m_auth_reply);
802 out_subscribe:
803         ceph_msg_put(monc->m_subscribe);
804 out_subscribe_ack:
805         ceph_msg_put(monc->m_subscribe_ack);
806 out_con:
807         monc->con->ops->put(monc->con);
808 out_monmap:
809         kfree(monc->monmap);
810 out:
811         return err;
812 }
813 EXPORT_SYMBOL(ceph_monc_init);
814
815 void ceph_monc_stop(struct ceph_mon_client *monc)
816 {
817         dout("stop\n");
818         cancel_delayed_work_sync(&monc->delayed_work);
819
820         mutex_lock(&monc->mutex);
821         __close_session(monc);
822
823         monc->con->private = NULL;
824         monc->con->ops->put(monc->con);
825         monc->con = NULL;
826
827         mutex_unlock(&monc->mutex);
828
829         ceph_auth_destroy(monc->auth);
830
831         ceph_msg_put(monc->m_auth);
832         ceph_msg_put(monc->m_auth_reply);
833         ceph_msg_put(monc->m_subscribe);
834         ceph_msg_put(monc->m_subscribe_ack);
835
836         kfree(monc->monmap);
837 }
838 EXPORT_SYMBOL(ceph_monc_stop);
839
840 static void handle_auth_reply(struct ceph_mon_client *monc,
841                               struct ceph_msg *msg)
842 {
843         int ret;
844         int was_auth = 0;
845
846         mutex_lock(&monc->mutex);
847         if (monc->auth->ops)
848                 was_auth = monc->auth->ops->is_authenticated(monc->auth);
849         monc->pending_auth = 0;
850         ret = ceph_handle_auth_reply(monc->auth, msg->front.iov_base,
851                                      msg->front.iov_len,
852                                      monc->m_auth->front.iov_base,
853                                      monc->m_auth->front_max);
854         if (ret < 0) {
855                 monc->client->auth_err = ret;
856                 wake_up_all(&monc->client->auth_wq);
857         } else if (ret > 0) {
858                 __send_prepared_auth_request(monc, ret);
859         } else if (!was_auth && monc->auth->ops->is_authenticated(monc->auth)) {
860                 dout("authenticated, starting session\n");
861
862                 monc->client->msgr->inst.name.type = CEPH_ENTITY_TYPE_CLIENT;
863                 monc->client->msgr->inst.name.num =
864                                         cpu_to_le64(monc->auth->global_id);
865
866                 __send_subscribe(monc);
867                 __resend_generic_request(monc);
868         }
869         mutex_unlock(&monc->mutex);
870 }
871
872 static int __validate_auth(struct ceph_mon_client *monc)
873 {
874         int ret;
875
876         if (monc->pending_auth)
877                 return 0;
878
879         ret = ceph_build_auth(monc->auth, monc->m_auth->front.iov_base,
880                               monc->m_auth->front_max);
881         if (ret <= 0)
882                 return ret; /* either an error, or no need to authenticate */
883         __send_prepared_auth_request(monc, ret);
884         return 0;
885 }
886
887 int ceph_monc_validate_auth(struct ceph_mon_client *monc)
888 {
889         int ret;
890
891         mutex_lock(&monc->mutex);
892         ret = __validate_auth(monc);
893         mutex_unlock(&monc->mutex);
894         return ret;
895 }
896 EXPORT_SYMBOL(ceph_monc_validate_auth);
897
898 /*
899  * handle incoming message
900  */
901 static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
902 {
903         struct ceph_mon_client *monc = con->private;
904         int type = le16_to_cpu(msg->hdr.type);
905
906         if (!monc)
907                 return;
908
909         switch (type) {
910         case CEPH_MSG_AUTH_REPLY:
911                 handle_auth_reply(monc, msg);
912                 break;
913
914         case CEPH_MSG_MON_SUBSCRIBE_ACK:
915                 handle_subscribe_ack(monc, msg);
916                 break;
917
918         case CEPH_MSG_STATFS_REPLY:
919                 handle_statfs_reply(monc, msg);
920                 break;
921
922         case CEPH_MSG_POOLOP_REPLY:
923                 handle_poolop_reply(monc, msg);
924                 break;
925
926         case CEPH_MSG_MON_MAP:
927                 ceph_monc_handle_map(monc, msg);
928                 break;
929
930         case CEPH_MSG_OSD_MAP:
931                 ceph_osdc_handle_map(&monc->client->osdc, msg);
932                 break;
933
934         default:
935                 /* can the chained handler handle it? */
936                 if (monc->client->extra_mon_dispatch &&
937                     monc->client->extra_mon_dispatch(monc->client, msg) == 0)
938                         break;
939                         
940                 pr_err("received unknown message type %d %s\n", type,
941                        ceph_msg_type_name(type));
942         }
943         ceph_msg_put(msg);
944 }
945
946 /*
947  * Allocate memory for incoming message
948  */
949 static struct ceph_msg *mon_alloc_msg(struct ceph_connection *con,
950                                       struct ceph_msg_header *hdr,
951                                       int *skip)
952 {
953         struct ceph_mon_client *monc = con->private;
954         int type = le16_to_cpu(hdr->type);
955         int front_len = le32_to_cpu(hdr->front_len);
956         struct ceph_msg *m = NULL;
957
958         *skip = 0;
959
960         switch (type) {
961         case CEPH_MSG_MON_SUBSCRIBE_ACK:
962                 m = ceph_msg_get(monc->m_subscribe_ack);
963                 break;
964         case CEPH_MSG_POOLOP_REPLY:
965         case CEPH_MSG_STATFS_REPLY:
966                 return get_generic_reply(con, hdr, skip);
967         case CEPH_MSG_AUTH_REPLY:
968                 m = ceph_msg_get(monc->m_auth_reply);
969                 break;
970         case CEPH_MSG_MON_MAP:
971         case CEPH_MSG_MDS_MAP:
972         case CEPH_MSG_OSD_MAP:
973                 m = ceph_msg_new(type, front_len, GFP_NOFS);
974                 break;
975         }
976
977         if (!m) {
978                 pr_info("alloc_msg unknown type %d\n", type);
979                 *skip = 1;
980         }
981         return m;
982 }
983
984 /*
985  * If the monitor connection resets, pick a new monitor and resubmit
986  * any pending requests.
987  */
988 static void mon_fault(struct ceph_connection *con)
989 {
990         struct ceph_mon_client *monc = con->private;
991
992         if (!monc)
993                 return;
994
995         dout("mon_fault\n");
996         mutex_lock(&monc->mutex);
997         if (!con->private)
998                 goto out;
999
1000         if (!monc->hunting)
1001                 pr_info("mon%d %s session lost, "
1002                         "hunting for new mon\n", monc->cur_mon,
1003                         ceph_pr_addr(&monc->con->peer_addr.in_addr));
1004
1005         __close_session(monc);
1006         if (!monc->hunting) {
1007                 /* start hunting */
1008                 monc->hunting = true;
1009                 __open_session(monc);
1010         } else {
1011                 /* already hunting, let's wait a bit */
1012                 __schedule_delayed(monc);
1013         }
1014 out:
1015         mutex_unlock(&monc->mutex);
1016 }
1017
1018 static const struct ceph_connection_operations mon_con_ops = {
1019         .get = ceph_con_get,
1020         .put = ceph_con_put,
1021         .dispatch = dispatch,
1022         .fault = mon_fault,
1023         .alloc_msg = mon_alloc_msg,
1024 };