Merge branch 'for-linus' of git://git.kernel.org/pub/scm/linux/kernel/git/sage/ceph...
[linux-3.10.git] / fs / ceph / super.c
1
2 #include "ceph_debug.h"
3
4 #include <linux/backing-dev.h>
5 #include <linux/fs.h>
6 #include <linux/inet.h>
7 #include <linux/in6.h>
8 #include <linux/module.h>
9 #include <linux/mount.h>
10 #include <linux/parser.h>
11 #include <linux/sched.h>
12 #include <linux/seq_file.h>
13 #include <linux/slab.h>
14 #include <linux/statfs.h>
15 #include <linux/string.h>
16
17 #include "decode.h"
18 #include "super.h"
19 #include "mon_client.h"
20 #include "auth.h"
21
22 /*
23  * Ceph superblock operations
24  *
25  * Handle the basics of mounting, unmounting.
26  */
27
28
29 /*
30  * find filename portion of a path (/foo/bar/baz -> baz)
31  */
32 const char *ceph_file_part(const char *s, int len)
33 {
34         const char *e = s + len;
35
36         while (e != s && *(e-1) != '/')
37                 e--;
38         return e;
39 }
40
41
42 /*
43  * super ops
44  */
45 static void ceph_put_super(struct super_block *s)
46 {
47         struct ceph_client *client = ceph_sb_to_client(s);
48
49         dout("put_super\n");
50         ceph_mdsc_close_sessions(&client->mdsc);
51
52         /*
53          * ensure we release the bdi before put_anon_super releases
54          * the device name.
55          */
56         if (s->s_bdi == &client->backing_dev_info) {
57                 bdi_unregister(&client->backing_dev_info);
58                 s->s_bdi = NULL;
59         }
60
61         return;
62 }
63
64 static int ceph_statfs(struct dentry *dentry, struct kstatfs *buf)
65 {
66         struct ceph_client *client = ceph_inode_to_client(dentry->d_inode);
67         struct ceph_monmap *monmap = client->monc.monmap;
68         struct ceph_statfs st;
69         u64 fsid;
70         int err;
71
72         dout("statfs\n");
73         err = ceph_monc_do_statfs(&client->monc, &st);
74         if (err < 0)
75                 return err;
76
77         /* fill in kstatfs */
78         buf->f_type = CEPH_SUPER_MAGIC;  /* ?? */
79
80         /*
81          * express utilization in terms of large blocks to avoid
82          * overflow on 32-bit machines.
83          */
84         buf->f_bsize = 1 << CEPH_BLOCK_SHIFT;
85         buf->f_blocks = le64_to_cpu(st.kb) >> (CEPH_BLOCK_SHIFT-10);
86         buf->f_bfree = (le64_to_cpu(st.kb) - le64_to_cpu(st.kb_used)) >>
87                 (CEPH_BLOCK_SHIFT-10);
88         buf->f_bavail = le64_to_cpu(st.kb_avail) >> (CEPH_BLOCK_SHIFT-10);
89
90         buf->f_files = le64_to_cpu(st.num_objects);
91         buf->f_ffree = -1;
92         buf->f_namelen = PATH_MAX;
93         buf->f_frsize = PAGE_CACHE_SIZE;
94
95         /* leave fsid little-endian, regardless of host endianness */
96         fsid = *(u64 *)(&monmap->fsid) ^ *((u64 *)&monmap->fsid + 1);
97         buf->f_fsid.val[0] = fsid & 0xffffffff;
98         buf->f_fsid.val[1] = fsid >> 32;
99
100         return 0;
101 }
102
103
104 static int ceph_syncfs(struct super_block *sb, int wait)
105 {
106         dout("sync_fs %d\n", wait);
107         ceph_osdc_sync(&ceph_sb_to_client(sb)->osdc);
108         ceph_mdsc_sync(&ceph_sb_to_client(sb)->mdsc);
109         dout("sync_fs %d done\n", wait);
110         return 0;
111 }
112
113 static int default_congestion_kb(void)
114 {
115         int congestion_kb;
116
117         /*
118          * Copied from NFS
119          *
120          * congestion size, scale with available memory.
121          *
122          *  64MB:    8192k
123          * 128MB:   11585k
124          * 256MB:   16384k
125          * 512MB:   23170k
126          *   1GB:   32768k
127          *   2GB:   46340k
128          *   4GB:   65536k
129          *   8GB:   92681k
130          *  16GB:  131072k
131          *
132          * This allows larger machines to have larger/more transfers.
133          * Limit the default to 256M
134          */
135         congestion_kb = (16*int_sqrt(totalram_pages)) << (PAGE_SHIFT-10);
136         if (congestion_kb > 256*1024)
137                 congestion_kb = 256*1024;
138
139         return congestion_kb;
140 }
141
142 /**
143  * ceph_show_options - Show mount options in /proc/mounts
144  * @m: seq_file to write to
145  * @mnt: mount descriptor
146  */
147 static int ceph_show_options(struct seq_file *m, struct vfsmount *mnt)
148 {
149         struct ceph_client *client = ceph_sb_to_client(mnt->mnt_sb);
150         struct ceph_mount_args *args = client->mount_args;
151
152         if (args->flags & CEPH_OPT_FSID)
153                 seq_printf(m, ",fsidmajor=%llu,fsidminor%llu",
154                            le64_to_cpu(*(__le64 *)&args->fsid.fsid[0]),
155                            le64_to_cpu(*(__le64 *)&args->fsid.fsid[8]));
156         if (args->flags & CEPH_OPT_NOSHARE)
157                 seq_puts(m, ",noshare");
158         if (args->flags & CEPH_OPT_DIRSTAT)
159                 seq_puts(m, ",dirstat");
160         if ((args->flags & CEPH_OPT_RBYTES) == 0)
161                 seq_puts(m, ",norbytes");
162         if (args->flags & CEPH_OPT_NOCRC)
163                 seq_puts(m, ",nocrc");
164         if (args->flags & CEPH_OPT_NOASYNCREADDIR)
165                 seq_puts(m, ",noasyncreaddir");
166
167         if (args->mount_timeout != CEPH_MOUNT_TIMEOUT_DEFAULT)
168                 seq_printf(m, ",mount_timeout=%d", args->mount_timeout);
169         if (args->osd_idle_ttl != CEPH_OSD_IDLE_TTL_DEFAULT)
170                 seq_printf(m, ",osd_idle_ttl=%d", args->osd_idle_ttl);
171         if (args->osd_timeout != CEPH_OSD_TIMEOUT_DEFAULT)
172                 seq_printf(m, ",osdtimeout=%d", args->osd_timeout);
173         if (args->osd_keepalive_timeout != CEPH_OSD_KEEPALIVE_DEFAULT)
174                 seq_printf(m, ",osdkeepalivetimeout=%d",
175                          args->osd_keepalive_timeout);
176         if (args->wsize)
177                 seq_printf(m, ",wsize=%d", args->wsize);
178         if (args->rsize != CEPH_MOUNT_RSIZE_DEFAULT)
179                 seq_printf(m, ",rsize=%d", args->rsize);
180         if (args->congestion_kb != default_congestion_kb())
181                 seq_printf(m, ",write_congestion_kb=%d", args->congestion_kb);
182         if (args->caps_wanted_delay_min != CEPH_CAPS_WANTED_DELAY_MIN_DEFAULT)
183                 seq_printf(m, ",caps_wanted_delay_min=%d",
184                          args->caps_wanted_delay_min);
185         if (args->caps_wanted_delay_max != CEPH_CAPS_WANTED_DELAY_MAX_DEFAULT)
186                 seq_printf(m, ",caps_wanted_delay_max=%d",
187                            args->caps_wanted_delay_max);
188         if (args->cap_release_safety != CEPH_CAP_RELEASE_SAFETY_DEFAULT)
189                 seq_printf(m, ",cap_release_safety=%d",
190                            args->cap_release_safety);
191         if (args->max_readdir != CEPH_MAX_READDIR_DEFAULT)
192                 seq_printf(m, ",readdir_max_entries=%d", args->max_readdir);
193         if (args->max_readdir_bytes != CEPH_MAX_READDIR_BYTES_DEFAULT)
194                 seq_printf(m, ",readdir_max_bytes=%d", args->max_readdir_bytes);
195         if (strcmp(args->snapdir_name, CEPH_SNAPDIRNAME_DEFAULT))
196                 seq_printf(m, ",snapdirname=%s", args->snapdir_name);
197         if (args->name)
198                 seq_printf(m, ",name=%s", args->name);
199         if (args->secret)
200                 seq_puts(m, ",secret=<hidden>");
201         return 0;
202 }
203
204 /*
205  * caches
206  */
207 struct kmem_cache *ceph_inode_cachep;
208 struct kmem_cache *ceph_cap_cachep;
209 struct kmem_cache *ceph_dentry_cachep;
210 struct kmem_cache *ceph_file_cachep;
211
212 static void ceph_inode_init_once(void *foo)
213 {
214         struct ceph_inode_info *ci = foo;
215         inode_init_once(&ci->vfs_inode);
216 }
217
218 static int __init init_caches(void)
219 {
220         ceph_inode_cachep = kmem_cache_create("ceph_inode_info",
221                                       sizeof(struct ceph_inode_info),
222                                       __alignof__(struct ceph_inode_info),
223                                       (SLAB_RECLAIM_ACCOUNT|SLAB_MEM_SPREAD),
224                                       ceph_inode_init_once);
225         if (ceph_inode_cachep == NULL)
226                 return -ENOMEM;
227
228         ceph_cap_cachep = KMEM_CACHE(ceph_cap,
229                                      SLAB_RECLAIM_ACCOUNT|SLAB_MEM_SPREAD);
230         if (ceph_cap_cachep == NULL)
231                 goto bad_cap;
232
233         ceph_dentry_cachep = KMEM_CACHE(ceph_dentry_info,
234                                         SLAB_RECLAIM_ACCOUNT|SLAB_MEM_SPREAD);
235         if (ceph_dentry_cachep == NULL)
236                 goto bad_dentry;
237
238         ceph_file_cachep = KMEM_CACHE(ceph_file_info,
239                                       SLAB_RECLAIM_ACCOUNT|SLAB_MEM_SPREAD);
240         if (ceph_file_cachep == NULL)
241                 goto bad_file;
242
243         return 0;
244
245 bad_file:
246         kmem_cache_destroy(ceph_dentry_cachep);
247 bad_dentry:
248         kmem_cache_destroy(ceph_cap_cachep);
249 bad_cap:
250         kmem_cache_destroy(ceph_inode_cachep);
251         return -ENOMEM;
252 }
253
254 static void destroy_caches(void)
255 {
256         kmem_cache_destroy(ceph_inode_cachep);
257         kmem_cache_destroy(ceph_cap_cachep);
258         kmem_cache_destroy(ceph_dentry_cachep);
259         kmem_cache_destroy(ceph_file_cachep);
260 }
261
262
263 /*
264  * ceph_umount_begin - initiate forced umount.  Tear down down the
265  * mount, skipping steps that may hang while waiting for server(s).
266  */
267 static void ceph_umount_begin(struct super_block *sb)
268 {
269         struct ceph_client *client = ceph_sb_to_client(sb);
270
271         dout("ceph_umount_begin - starting forced umount\n");
272         if (!client)
273                 return;
274         client->mount_state = CEPH_MOUNT_SHUTDOWN;
275         return;
276 }
277
278 static const struct super_operations ceph_super_ops = {
279         .alloc_inode    = ceph_alloc_inode,
280         .destroy_inode  = ceph_destroy_inode,
281         .write_inode    = ceph_write_inode,
282         .sync_fs        = ceph_syncfs,
283         .put_super      = ceph_put_super,
284         .show_options   = ceph_show_options,
285         .statfs         = ceph_statfs,
286         .umount_begin   = ceph_umount_begin,
287 };
288
289
290 const char *ceph_msg_type_name(int type)
291 {
292         switch (type) {
293         case CEPH_MSG_SHUTDOWN: return "shutdown";
294         case CEPH_MSG_PING: return "ping";
295         case CEPH_MSG_AUTH: return "auth";
296         case CEPH_MSG_AUTH_REPLY: return "auth_reply";
297         case CEPH_MSG_MON_MAP: return "mon_map";
298         case CEPH_MSG_MON_GET_MAP: return "mon_get_map";
299         case CEPH_MSG_MON_SUBSCRIBE: return "mon_subscribe";
300         case CEPH_MSG_MON_SUBSCRIBE_ACK: return "mon_subscribe_ack";
301         case CEPH_MSG_STATFS: return "statfs";
302         case CEPH_MSG_STATFS_REPLY: return "statfs_reply";
303         case CEPH_MSG_MDS_MAP: return "mds_map";
304         case CEPH_MSG_CLIENT_SESSION: return "client_session";
305         case CEPH_MSG_CLIENT_RECONNECT: return "client_reconnect";
306         case CEPH_MSG_CLIENT_REQUEST: return "client_request";
307         case CEPH_MSG_CLIENT_REQUEST_FORWARD: return "client_request_forward";
308         case CEPH_MSG_CLIENT_REPLY: return "client_reply";
309         case CEPH_MSG_CLIENT_CAPS: return "client_caps";
310         case CEPH_MSG_CLIENT_CAPRELEASE: return "client_cap_release";
311         case CEPH_MSG_CLIENT_SNAP: return "client_snap";
312         case CEPH_MSG_CLIENT_LEASE: return "client_lease";
313         case CEPH_MSG_OSD_MAP: return "osd_map";
314         case CEPH_MSG_OSD_OP: return "osd_op";
315         case CEPH_MSG_OSD_OPREPLY: return "osd_opreply";
316         default: return "unknown";
317         }
318 }
319
320
321 /*
322  * mount options
323  */
324 enum {
325         Opt_fsidmajor,
326         Opt_fsidminor,
327         Opt_monport,
328         Opt_wsize,
329         Opt_rsize,
330         Opt_osdtimeout,
331         Opt_osdkeepalivetimeout,
332         Opt_mount_timeout,
333         Opt_osd_idle_ttl,
334         Opt_caps_wanted_delay_min,
335         Opt_caps_wanted_delay_max,
336         Opt_cap_release_safety,
337         Opt_readdir_max_entries,
338         Opt_readdir_max_bytes,
339         Opt_congestion_kb,
340         Opt_last_int,
341         /* int args above */
342         Opt_snapdirname,
343         Opt_name,
344         Opt_secret,
345         Opt_last_string,
346         /* string args above */
347         Opt_ip,
348         Opt_noshare,
349         Opt_dirstat,
350         Opt_nodirstat,
351         Opt_rbytes,
352         Opt_norbytes,
353         Opt_nocrc,
354         Opt_noasyncreaddir,
355 };
356
357 static match_table_t arg_tokens = {
358         {Opt_fsidmajor, "fsidmajor=%ld"},
359         {Opt_fsidminor, "fsidminor=%ld"},
360         {Opt_monport, "monport=%d"},
361         {Opt_wsize, "wsize=%d"},
362         {Opt_rsize, "rsize=%d"},
363         {Opt_osdtimeout, "osdtimeout=%d"},
364         {Opt_osdkeepalivetimeout, "osdkeepalive=%d"},
365         {Opt_mount_timeout, "mount_timeout=%d"},
366         {Opt_osd_idle_ttl, "osd_idle_ttl=%d"},
367         {Opt_caps_wanted_delay_min, "caps_wanted_delay_min=%d"},
368         {Opt_caps_wanted_delay_max, "caps_wanted_delay_max=%d"},
369         {Opt_cap_release_safety, "cap_release_safety=%d"},
370         {Opt_readdir_max_entries, "readdir_max_entries=%d"},
371         {Opt_readdir_max_bytes, "readdir_max_bytes=%d"},
372         {Opt_congestion_kb, "write_congestion_kb=%d"},
373         /* int args above */
374         {Opt_snapdirname, "snapdirname=%s"},
375         {Opt_name, "name=%s"},
376         {Opt_secret, "secret=%s"},
377         /* string args above */
378         {Opt_ip, "ip=%s"},
379         {Opt_noshare, "noshare"},
380         {Opt_dirstat, "dirstat"},
381         {Opt_nodirstat, "nodirstat"},
382         {Opt_rbytes, "rbytes"},
383         {Opt_norbytes, "norbytes"},
384         {Opt_nocrc, "nocrc"},
385         {Opt_noasyncreaddir, "noasyncreaddir"},
386         {-1, NULL}
387 };
388
389
390 static struct ceph_mount_args *parse_mount_args(int flags, char *options,
391                                                 const char *dev_name,
392                                                 const char **path)
393 {
394         struct ceph_mount_args *args;
395         const char *c;
396         int err = -ENOMEM;
397         substring_t argstr[MAX_OPT_ARGS];
398
399         args = kzalloc(sizeof(*args), GFP_KERNEL);
400         if (!args)
401                 return ERR_PTR(-ENOMEM);
402         args->mon_addr = kcalloc(CEPH_MAX_MON, sizeof(*args->mon_addr),
403                                  GFP_KERNEL);
404         if (!args->mon_addr)
405                 goto out;
406
407         dout("parse_mount_args %p, dev_name '%s'\n", args, dev_name);
408
409         /* start with defaults */
410         args->sb_flags = flags;
411         args->flags = CEPH_OPT_DEFAULT;
412         args->osd_timeout = CEPH_OSD_TIMEOUT_DEFAULT;
413         args->osd_keepalive_timeout = CEPH_OSD_KEEPALIVE_DEFAULT;
414         args->mount_timeout = CEPH_MOUNT_TIMEOUT_DEFAULT; /* seconds */
415         args->osd_idle_ttl = CEPH_OSD_IDLE_TTL_DEFAULT;   /* seconds */
416         args->caps_wanted_delay_min = CEPH_CAPS_WANTED_DELAY_MIN_DEFAULT;
417         args->caps_wanted_delay_max = CEPH_CAPS_WANTED_DELAY_MAX_DEFAULT;
418         args->rsize = CEPH_MOUNT_RSIZE_DEFAULT;
419         args->snapdir_name = kstrdup(CEPH_SNAPDIRNAME_DEFAULT, GFP_KERNEL);
420         args->cap_release_safety = CEPH_CAP_RELEASE_SAFETY_DEFAULT;
421         args->max_readdir = CEPH_MAX_READDIR_DEFAULT;
422         args->max_readdir_bytes = CEPH_MAX_READDIR_BYTES_DEFAULT;
423         args->congestion_kb = default_congestion_kb();
424
425         /* ip1[:port1][,ip2[:port2]...]:/subdir/in/fs */
426         err = -EINVAL;
427         if (!dev_name)
428                 goto out;
429         *path = strstr(dev_name, ":/");
430         if (*path == NULL) {
431                 pr_err("device name is missing path (no :/ in %s)\n",
432                        dev_name);
433                 goto out;
434         }
435
436         /* get mon ip(s) */
437         err = ceph_parse_ips(dev_name, *path, args->mon_addr,
438                              CEPH_MAX_MON, &args->num_mon);
439         if (err < 0)
440                 goto out;
441
442         /* path on server */
443         *path += 2;
444         dout("server path '%s'\n", *path);
445
446         /* parse mount options */
447         while ((c = strsep(&options, ",")) != NULL) {
448                 int token, intval, ret;
449                 if (!*c)
450                         continue;
451                 err = -EINVAL;
452                 token = match_token((char *)c, arg_tokens, argstr);
453                 if (token < 0) {
454                         pr_err("bad mount option at '%s'\n", c);
455                         goto out;
456                 }
457                 if (token < Opt_last_int) {
458                         ret = match_int(&argstr[0], &intval);
459                         if (ret < 0) {
460                                 pr_err("bad mount option arg (not int) "
461                                        "at '%s'\n", c);
462                                 continue;
463                         }
464                         dout("got int token %d val %d\n", token, intval);
465                 } else if (token > Opt_last_int && token < Opt_last_string) {
466                         dout("got string token %d val %s\n", token,
467                              argstr[0].from);
468                 } else {
469                         dout("got token %d\n", token);
470                 }
471                 switch (token) {
472                 case Opt_fsidmajor:
473                         *(__le64 *)&args->fsid.fsid[0] = cpu_to_le64(intval);
474                         break;
475                 case Opt_fsidminor:
476                         *(__le64 *)&args->fsid.fsid[8] = cpu_to_le64(intval);
477                         break;
478                 case Opt_ip:
479                         err = ceph_parse_ips(argstr[0].from,
480                                              argstr[0].to,
481                                              &args->my_addr,
482                                              1, NULL);
483                         if (err < 0)
484                                 goto out;
485                         args->flags |= CEPH_OPT_MYIP;
486                         break;
487
488                 case Opt_snapdirname:
489                         kfree(args->snapdir_name);
490                         args->snapdir_name = kstrndup(argstr[0].from,
491                                               argstr[0].to-argstr[0].from,
492                                               GFP_KERNEL);
493                         break;
494                 case Opt_name:
495                         args->name = kstrndup(argstr[0].from,
496                                               argstr[0].to-argstr[0].from,
497                                               GFP_KERNEL);
498                         break;
499                 case Opt_secret:
500                         args->secret = kstrndup(argstr[0].from,
501                                                 argstr[0].to-argstr[0].from,
502                                                 GFP_KERNEL);
503                         break;
504
505                         /* misc */
506                 case Opt_wsize:
507                         args->wsize = intval;
508                         break;
509                 case Opt_rsize:
510                         args->rsize = intval;
511                         break;
512                 case Opt_osdtimeout:
513                         args->osd_timeout = intval;
514                         break;
515                 case Opt_osdkeepalivetimeout:
516                         args->osd_keepalive_timeout = intval;
517                         break;
518                 case Opt_mount_timeout:
519                         args->mount_timeout = intval;
520                         break;
521                 case Opt_caps_wanted_delay_min:
522                         args->caps_wanted_delay_min = intval;
523                         break;
524                 case Opt_caps_wanted_delay_max:
525                         args->caps_wanted_delay_max = intval;
526                         break;
527                 case Opt_readdir_max_entries:
528                         args->max_readdir = intval;
529                         break;
530                 case Opt_readdir_max_bytes:
531                         args->max_readdir_bytes = intval;
532                         break;
533                 case Opt_congestion_kb:
534                         args->congestion_kb = intval;
535                         break;
536
537                 case Opt_noshare:
538                         args->flags |= CEPH_OPT_NOSHARE;
539                         break;
540
541                 case Opt_dirstat:
542                         args->flags |= CEPH_OPT_DIRSTAT;
543                         break;
544                 case Opt_nodirstat:
545                         args->flags &= ~CEPH_OPT_DIRSTAT;
546                         break;
547                 case Opt_rbytes:
548                         args->flags |= CEPH_OPT_RBYTES;
549                         break;
550                 case Opt_norbytes:
551                         args->flags &= ~CEPH_OPT_RBYTES;
552                         break;
553                 case Opt_nocrc:
554                         args->flags |= CEPH_OPT_NOCRC;
555                         break;
556                 case Opt_noasyncreaddir:
557                         args->flags |= CEPH_OPT_NOASYNCREADDIR;
558                         break;
559
560                 default:
561                         BUG_ON(token);
562                 }
563         }
564         return args;
565
566 out:
567         kfree(args->mon_addr);
568         kfree(args);
569         return ERR_PTR(err);
570 }
571
572 static void destroy_mount_args(struct ceph_mount_args *args)
573 {
574         dout("destroy_mount_args %p\n", args);
575         kfree(args->snapdir_name);
576         args->snapdir_name = NULL;
577         kfree(args->name);
578         args->name = NULL;
579         kfree(args->secret);
580         args->secret = NULL;
581         kfree(args);
582 }
583
584 /*
585  * create a fresh client instance
586  */
587 static struct ceph_client *ceph_create_client(struct ceph_mount_args *args)
588 {
589         struct ceph_client *client;
590         int err = -ENOMEM;
591
592         client = kzalloc(sizeof(*client), GFP_KERNEL);
593         if (client == NULL)
594                 return ERR_PTR(-ENOMEM);
595
596         mutex_init(&client->mount_mutex);
597
598         init_waitqueue_head(&client->auth_wq);
599
600         client->sb = NULL;
601         client->mount_state = CEPH_MOUNT_MOUNTING;
602         client->mount_args = args;
603
604         client->msgr = NULL;
605
606         client->auth_err = 0;
607         atomic_long_set(&client->writeback_count, 0);
608
609         err = bdi_init(&client->backing_dev_info);
610         if (err < 0)
611                 goto fail;
612
613         err = -ENOMEM;
614         client->wb_wq = create_workqueue("ceph-writeback");
615         if (client->wb_wq == NULL)
616                 goto fail_bdi;
617         client->pg_inv_wq = create_singlethread_workqueue("ceph-pg-invalid");
618         if (client->pg_inv_wq == NULL)
619                 goto fail_wb_wq;
620         client->trunc_wq = create_singlethread_workqueue("ceph-trunc");
621         if (client->trunc_wq == NULL)
622                 goto fail_pg_inv_wq;
623
624         /* set up mempools */
625         err = -ENOMEM;
626         client->wb_pagevec_pool = mempool_create_kmalloc_pool(10,
627                               client->mount_args->wsize >> PAGE_CACHE_SHIFT);
628         if (!client->wb_pagevec_pool)
629                 goto fail_trunc_wq;
630
631         /* caps */
632         client->min_caps = args->max_readdir;
633         ceph_adjust_min_caps(client->min_caps);
634
635         /* subsystems */
636         err = ceph_monc_init(&client->monc, client);
637         if (err < 0)
638                 goto fail_mempool;
639         err = ceph_osdc_init(&client->osdc, client);
640         if (err < 0)
641                 goto fail_monc;
642         err = ceph_mdsc_init(&client->mdsc, client);
643         if (err < 0)
644                 goto fail_osdc;
645         return client;
646
647 fail_osdc:
648         ceph_osdc_stop(&client->osdc);
649 fail_monc:
650         ceph_monc_stop(&client->monc);
651 fail_mempool:
652         mempool_destroy(client->wb_pagevec_pool);
653 fail_trunc_wq:
654         destroy_workqueue(client->trunc_wq);
655 fail_pg_inv_wq:
656         destroy_workqueue(client->pg_inv_wq);
657 fail_wb_wq:
658         destroy_workqueue(client->wb_wq);
659 fail_bdi:
660         bdi_destroy(&client->backing_dev_info);
661 fail:
662         kfree(client);
663         return ERR_PTR(err);
664 }
665
666 static void ceph_destroy_client(struct ceph_client *client)
667 {
668         dout("destroy_client %p\n", client);
669
670         /* unmount */
671         ceph_mdsc_stop(&client->mdsc);
672         ceph_osdc_stop(&client->osdc);
673
674         /*
675          * make sure mds and osd connections close out before destroying
676          * the auth module, which is needed to free those connections'
677          * ceph_authorizers.
678          */
679         ceph_msgr_flush();
680
681         ceph_monc_stop(&client->monc);
682
683         ceph_adjust_min_caps(-client->min_caps);
684
685         ceph_debugfs_client_cleanup(client);
686         destroy_workqueue(client->wb_wq);
687         destroy_workqueue(client->pg_inv_wq);
688         destroy_workqueue(client->trunc_wq);
689
690         bdi_destroy(&client->backing_dev_info);
691
692         if (client->msgr)
693                 ceph_messenger_destroy(client->msgr);
694         mempool_destroy(client->wb_pagevec_pool);
695
696         destroy_mount_args(client->mount_args);
697
698         kfree(client);
699         dout("destroy_client %p done\n", client);
700 }
701
702 /*
703  * Initially learn our fsid, or verify an fsid matches.
704  */
705 int ceph_check_fsid(struct ceph_client *client, struct ceph_fsid *fsid)
706 {
707         if (client->have_fsid) {
708                 if (ceph_fsid_compare(&client->fsid, fsid)) {
709                         pr_err("bad fsid, had " FSID_FORMAT " got " FSID_FORMAT,
710                                PR_FSID(&client->fsid), PR_FSID(fsid));
711                         return -1;
712                 }
713         } else {
714                 pr_info("client%lld fsid " FSID_FORMAT "\n",
715                         client->monc.auth->global_id, PR_FSID(fsid));
716                 memcpy(&client->fsid, fsid, sizeof(*fsid));
717                 ceph_debugfs_client_init(client);
718                 client->have_fsid = true;
719         }
720         return 0;
721 }
722
723 /*
724  * true if we have the mon map (and have thus joined the cluster)
725  */
726 static int have_mon_and_osd_map(struct ceph_client *client)
727 {
728         return client->monc.monmap && client->monc.monmap->epoch &&
729                client->osdc.osdmap && client->osdc.osdmap->epoch;
730 }
731
732 /*
733  * Bootstrap mount by opening the root directory.  Note the mount
734  * @started time from caller, and time out if this takes too long.
735  */
736 static struct dentry *open_root_dentry(struct ceph_client *client,
737                                        const char *path,
738                                        unsigned long started)
739 {
740         struct ceph_mds_client *mdsc = &client->mdsc;
741         struct ceph_mds_request *req = NULL;
742         int err;
743         struct dentry *root;
744
745         /* open dir */
746         dout("open_root_inode opening '%s'\n", path);
747         req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_GETATTR, USE_ANY_MDS);
748         if (IS_ERR(req))
749                 return ERR_CAST(req);
750         req->r_path1 = kstrdup(path, GFP_NOFS);
751         req->r_ino1.ino = CEPH_INO_ROOT;
752         req->r_ino1.snap = CEPH_NOSNAP;
753         req->r_started = started;
754         req->r_timeout = client->mount_args->mount_timeout * HZ;
755         req->r_args.getattr.mask = cpu_to_le32(CEPH_STAT_CAP_INODE);
756         req->r_num_caps = 2;
757         err = ceph_mdsc_do_request(mdsc, NULL, req);
758         if (err == 0) {
759                 dout("open_root_inode success\n");
760                 if (ceph_ino(req->r_target_inode) == CEPH_INO_ROOT &&
761                     client->sb->s_root == NULL)
762                         root = d_alloc_root(req->r_target_inode);
763                 else
764                         root = d_obtain_alias(req->r_target_inode);
765                 req->r_target_inode = NULL;
766                 dout("open_root_inode success, root dentry is %p\n", root);
767         } else {
768                 root = ERR_PTR(err);
769         }
770         ceph_mdsc_put_request(req);
771         return root;
772 }
773
774 /*
775  * mount: join the ceph cluster, and open root directory.
776  */
777 static int ceph_mount(struct ceph_client *client, struct vfsmount *mnt,
778                       const char *path)
779 {
780         struct ceph_entity_addr *myaddr = NULL;
781         int err;
782         unsigned long timeout = client->mount_args->mount_timeout * HZ;
783         unsigned long started = jiffies;  /* note the start time */
784         struct dentry *root;
785
786         dout("mount start\n");
787         mutex_lock(&client->mount_mutex);
788
789         /* initialize the messenger */
790         if (client->msgr == NULL) {
791                 if (ceph_test_opt(client, MYIP))
792                         myaddr = &client->mount_args->my_addr;
793                 client->msgr = ceph_messenger_create(myaddr);
794                 if (IS_ERR(client->msgr)) {
795                         err = PTR_ERR(client->msgr);
796                         client->msgr = NULL;
797                         goto out;
798                 }
799                 client->msgr->nocrc = ceph_test_opt(client, NOCRC);
800         }
801
802         /* open session, and wait for mon, mds, and osd maps */
803         err = ceph_monc_open_session(&client->monc);
804         if (err < 0)
805                 goto out;
806
807         while (!have_mon_and_osd_map(client)) {
808                 err = -EIO;
809                 if (timeout && time_after_eq(jiffies, started + timeout))
810                         goto out;
811
812                 /* wait */
813                 dout("mount waiting for mon_map\n");
814                 err = wait_event_interruptible_timeout(client->auth_wq,
815                        have_mon_and_osd_map(client) || (client->auth_err < 0),
816                        timeout);
817                 if (err == -EINTR || err == -ERESTARTSYS)
818                         goto out;
819                 if (client->auth_err < 0) {
820                         err = client->auth_err;
821                         goto out;
822                 }
823         }
824
825         dout("mount opening root\n");
826         root = open_root_dentry(client, "", started);
827         if (IS_ERR(root)) {
828                 err = PTR_ERR(root);
829                 goto out;
830         }
831         if (client->sb->s_root)
832                 dput(root);
833         else
834                 client->sb->s_root = root;
835
836         if (path[0] == 0) {
837                 dget(root);
838         } else {
839                 dout("mount opening base mountpoint\n");
840                 root = open_root_dentry(client, path, started);
841                 if (IS_ERR(root)) {
842                         err = PTR_ERR(root);
843                         dput(client->sb->s_root);
844                         client->sb->s_root = NULL;
845                         goto out;
846                 }
847         }
848
849         mnt->mnt_root = root;
850         mnt->mnt_sb = client->sb;
851
852         client->mount_state = CEPH_MOUNT_MOUNTED;
853         dout("mount success\n");
854         err = 0;
855
856 out:
857         mutex_unlock(&client->mount_mutex);
858         return err;
859 }
860
861 static int ceph_set_super(struct super_block *s, void *data)
862 {
863         struct ceph_client *client = data;
864         int ret;
865
866         dout("set_super %p data %p\n", s, data);
867
868         s->s_flags = client->mount_args->sb_flags;
869         s->s_maxbytes = 1ULL << 40;  /* temp value until we get mdsmap */
870
871         s->s_fs_info = client;
872         client->sb = s;
873
874         s->s_op = &ceph_super_ops;
875         s->s_export_op = &ceph_export_ops;
876
877         s->s_time_gran = 1000;  /* 1000 ns == 1 us */
878
879         ret = set_anon_super(s, NULL);  /* what is that second arg for? */
880         if (ret != 0)
881                 goto fail;
882
883         return ret;
884
885 fail:
886         s->s_fs_info = NULL;
887         client->sb = NULL;
888         return ret;
889 }
890
891 /*
892  * share superblock if same fs AND options
893  */
894 static int ceph_compare_super(struct super_block *sb, void *data)
895 {
896         struct ceph_client *new = data;
897         struct ceph_mount_args *args = new->mount_args;
898         struct ceph_client *other = ceph_sb_to_client(sb);
899         int i;
900
901         dout("ceph_compare_super %p\n", sb);
902         if (args->flags & CEPH_OPT_FSID) {
903                 if (ceph_fsid_compare(&args->fsid, &other->fsid)) {
904                         dout("fsid doesn't match\n");
905                         return 0;
906                 }
907         } else {
908                 /* do we share (a) monitor? */
909                 for (i = 0; i < new->monc.monmap->num_mon; i++)
910                         if (ceph_monmap_contains(other->monc.monmap,
911                                          &new->monc.monmap->mon_inst[i].addr))
912                                 break;
913                 if (i == new->monc.monmap->num_mon) {
914                         dout("mon ip not part of monmap\n");
915                         return 0;
916                 }
917                 dout("mon ip matches existing sb %p\n", sb);
918         }
919         if (args->sb_flags != other->mount_args->sb_flags) {
920                 dout("flags differ\n");
921                 return 0;
922         }
923         return 1;
924 }
925
926 /*
927  * construct our own bdi so we can control readahead, etc.
928  */
929 static atomic_long_t bdi_seq = ATOMIC_INIT(0);
930
931 static int ceph_register_bdi(struct super_block *sb, struct ceph_client *client)
932 {
933         int err;
934
935         /* set ra_pages based on rsize mount option? */
936         if (client->mount_args->rsize >= PAGE_CACHE_SIZE)
937                 client->backing_dev_info.ra_pages =
938                         (client->mount_args->rsize + PAGE_CACHE_SIZE - 1)
939                         >> PAGE_SHIFT;
940         err = bdi_register(&client->backing_dev_info, NULL, "ceph-%d",
941                            atomic_long_inc_return(&bdi_seq));
942         if (!err)
943                 sb->s_bdi = &client->backing_dev_info;
944         return err;
945 }
946
947 static int ceph_get_sb(struct file_system_type *fs_type,
948                        int flags, const char *dev_name, void *data,
949                        struct vfsmount *mnt)
950 {
951         struct super_block *sb;
952         struct ceph_client *client;
953         int err;
954         int (*compare_super)(struct super_block *, void *) = ceph_compare_super;
955         const char *path = NULL;
956         struct ceph_mount_args *args;
957
958         dout("ceph_get_sb\n");
959         args = parse_mount_args(flags, data, dev_name, &path);
960         if (IS_ERR(args)) {
961                 err = PTR_ERR(args);
962                 goto out_final;
963         }
964
965         /* create client (which we may/may not use) */
966         client = ceph_create_client(args);
967         if (IS_ERR(client)) {
968                 err = PTR_ERR(client);
969                 goto out_final;
970         }
971
972         if (client->mount_args->flags & CEPH_OPT_NOSHARE)
973                 compare_super = NULL;
974         sb = sget(fs_type, compare_super, ceph_set_super, client);
975         if (IS_ERR(sb)) {
976                 err = PTR_ERR(sb);
977                 goto out;
978         }
979
980         if (ceph_sb_to_client(sb) != client) {
981                 ceph_destroy_client(client);
982                 client = ceph_sb_to_client(sb);
983                 dout("get_sb got existing client %p\n", client);
984         } else {
985                 dout("get_sb using new client %p\n", client);
986                 err = ceph_register_bdi(sb, client);
987                 if (err < 0)
988                         goto out_splat;
989         }
990
991         err = ceph_mount(client, mnt, path);
992         if (err < 0)
993                 goto out_splat;
994         dout("root %p inode %p ino %llx.%llx\n", mnt->mnt_root,
995              mnt->mnt_root->d_inode, ceph_vinop(mnt->mnt_root->d_inode));
996         return 0;
997
998 out_splat:
999         ceph_mdsc_close_sessions(&client->mdsc);
1000         deactivate_locked_super(sb);
1001         goto out_final;
1002
1003 out:
1004         ceph_destroy_client(client);
1005 out_final:
1006         dout("ceph_get_sb fail %d\n", err);
1007         return err;
1008 }
1009
1010 static void ceph_kill_sb(struct super_block *s)
1011 {
1012         struct ceph_client *client = ceph_sb_to_client(s);
1013         dout("kill_sb %p\n", s);
1014         ceph_mdsc_pre_umount(&client->mdsc);
1015         kill_anon_super(s);    /* will call put_super after sb is r/o */
1016         ceph_destroy_client(client);
1017 }
1018
1019 static struct file_system_type ceph_fs_type = {
1020         .owner          = THIS_MODULE,
1021         .name           = "ceph",
1022         .get_sb         = ceph_get_sb,
1023         .kill_sb        = ceph_kill_sb,
1024         .fs_flags       = FS_RENAME_DOES_D_MOVE,
1025 };
1026
1027 #define _STRINGIFY(x) #x
1028 #define STRINGIFY(x) _STRINGIFY(x)
1029
1030 static int __init init_ceph(void)
1031 {
1032         int ret = 0;
1033
1034         ret = ceph_debugfs_init();
1035         if (ret < 0)
1036                 goto out;
1037
1038         ret = ceph_msgr_init();
1039         if (ret < 0)
1040                 goto out_debugfs;
1041
1042         ret = init_caches();
1043         if (ret)
1044                 goto out_msgr;
1045
1046         ceph_caps_init();
1047
1048         ret = register_filesystem(&ceph_fs_type);
1049         if (ret)
1050                 goto out_icache;
1051
1052         pr_info("loaded (mon/mds/osd proto %d/%d/%d, osdmap %d/%d %d/%d)\n",
1053                 CEPH_MONC_PROTOCOL, CEPH_MDSC_PROTOCOL, CEPH_OSDC_PROTOCOL,
1054                 CEPH_OSDMAP_VERSION, CEPH_OSDMAP_VERSION_EXT,
1055                 CEPH_OSDMAP_INC_VERSION, CEPH_OSDMAP_INC_VERSION_EXT);
1056         return 0;
1057
1058 out_icache:
1059         destroy_caches();
1060 out_msgr:
1061         ceph_msgr_exit();
1062 out_debugfs:
1063         ceph_debugfs_cleanup();
1064 out:
1065         return ret;
1066 }
1067
1068 static void __exit exit_ceph(void)
1069 {
1070         dout("exit_ceph\n");
1071         unregister_filesystem(&ceph_fs_type);
1072         ceph_caps_finalize();
1073         destroy_caches();
1074         ceph_msgr_exit();
1075         ceph_debugfs_cleanup();
1076 }
1077
1078 module_init(init_ceph);
1079 module_exit(exit_ceph);
1080
1081 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
1082 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
1083 MODULE_AUTHOR("Patience Warnick <patience@newdream.net>");
1084 MODULE_DESCRIPTION("Ceph filesystem for Linux");
1085 MODULE_LICENSE("GPL");