84724187ec3e52a77b75dc25468c1aa07c4dbaac
[linux-2.6.git] / drivers / staging / dst / dcore.c
1 /*
2  * 2007+ Copyright (c) Evgeniy Polyakov <zbr@ioremap.net>
3  * All rights reserved.
4  *
5  * This program is free software; you can redistribute it and/or modify
6  * it under the terms of the GNU General Public License as published by
7  * the Free Software Foundation; either version 2 of the License, or
8  * (at your option) any later version.
9  *
10  * This program is distributed in the hope that it will be useful,
11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13  * GNU General Public License for more details.
14  */
15
16 #include <linux/module.h>
17 #include <linux/kernel.h>
18 #include <linux/blkdev.h>
19 #include <linux/bio.h>
20 #include <linux/buffer_head.h>
21 #include <linux/connector.h>
22 #include <linux/dst.h>
23 #include <linux/device.h>
24 #include <linux/jhash.h>
25 #include <linux/idr.h>
26 #include <linux/init.h>
27 #include <linux/namei.h>
28 #include <linux/slab.h>
29 #include <linux/socket.h>
30
31 #include <linux/in.h>
32 #include <linux/in6.h>
33
34 #include <net/sock.h>
35
36 static int dst_major;
37
38 static DEFINE_MUTEX(dst_hash_lock);
39 static struct list_head *dst_hashtable;
40 static unsigned int dst_hashtable_size = 128;
41 module_param(dst_hashtable_size, uint, 0644);
42
43 static char dst_name[] = "Dementianting goldfish";
44
45 static DEFINE_IDR(dst_index_idr);
46 static struct cb_id cn_dst_id = { CN_DST_IDX, CN_DST_VAL };
47
48 /*
49  * DST sysfs tree for device called 'storage':
50  *
51  * /sys/bus/dst/devices/storage/
52  * /sys/bus/dst/devices/storage/type : 192.168.4.80:1025
53  * /sys/bus/dst/devices/storage/size : 800
54  * /sys/bus/dst/devices/storage/name : storage
55  */
56
57 static int dst_dev_match(struct device *dev, struct device_driver *drv)
58 {
59         return 1;
60 }
61
62 static struct bus_type dst_dev_bus_type = {
63         .name           = "dst",
64         .match          = &dst_dev_match,
65 };
66
67 static void dst_node_release(struct device *dev)
68 {
69         struct dst_info *info = container_of(dev, struct dst_info, device);
70
71         kfree(info);
72 }
73
74 static struct device dst_node_dev = {
75         .bus            = &dst_dev_bus_type,
76         .release        = &dst_node_release
77 };
78
79 /*
80  * Setting size of the node after it was changed.
81  */
82 static void dst_node_set_size(struct dst_node *n)
83 {
84         struct block_device *bdev;
85
86         set_capacity(n->disk, n->size >> 9);
87
88         bdev = bdget_disk(n->disk, 0);
89         if (bdev) {
90                 mutex_lock(&bdev->bd_inode->i_mutex);
91                 i_size_write(bdev->bd_inode, n->size);
92                 mutex_unlock(&bdev->bd_inode->i_mutex);
93                 bdput(bdev);
94         }
95 }
96
97 /*
98  * Distributed storage request processing function.
99  */
100 static int dst_request(struct request_queue *q, struct bio *bio)
101 {
102         struct dst_node *n = q->queuedata;
103         int err = -EIO;
104
105         if (bio_empty_barrier(bio) && !q->prepare_discard_fn) {
106                 /*
107                  * This is a dirty^Wnice hack, but if we complete this
108                  * operation with -EOPNOTSUPP like intended, XFS
109                  * will stuck and freeze the machine. This may be
110                  * not particulary XFS problem though, but it is the
111                  * only FS which sends empty barrier at umount time
112                  * I worked with.
113                  *
114                  * Empty barriers are not allowed anyway, see 51fd77bd9f512
115                  * for example, although later it was changed to bio_discard()
116                  * only, which does not work in this case.
117                  */
118                 //err = -EOPNOTSUPP;
119                 err = 0;
120                 goto end_io;
121         }
122
123         bio_get(bio);
124
125         return dst_process_bio(n, bio);
126
127 end_io:
128         bio_endio(bio, err);
129         return err;
130 }
131
132 /*
133  * Open/close callbacks for appropriate block device.
134  */
135 static int dst_bdev_open(struct block_device *bdev, fmode_t mode)
136 {
137         struct dst_node *n = bdev->bd_disk->private_data;
138
139         dst_node_get(n);
140         return 0;
141 }
142
143 static int dst_bdev_release(struct gendisk *disk, fmode_t mode)
144 {
145         struct dst_node *n = disk->private_data;
146
147         dst_node_put(n);
148         return 0;
149 }
150
151 static struct block_device_operations dst_blk_ops = {
152         .open           = dst_bdev_open,
153         .release        = dst_bdev_release,
154         .owner          = THIS_MODULE,
155 };
156
157 /*
158  * Block layer binding - disk is created when array is fully configured
159  * by userspace request.
160  */
161 static int dst_node_create_disk(struct dst_node *n)
162 {
163         int err = -ENOMEM;
164         u32 index = 0;
165
166         n->queue = blk_init_queue(NULL, NULL);
167         if (!n->queue)
168                 goto err_out_exit;
169
170         n->queue->queuedata = n;
171         blk_queue_make_request(n->queue, dst_request);
172         blk_queue_max_phys_segments(n->queue, n->max_pages);
173         blk_queue_max_hw_segments(n->queue, n->max_pages);
174
175         err = -ENOMEM;
176         n->disk = alloc_disk(1);
177         if (!n->disk)
178                 goto err_out_free_queue;
179
180         if (!(n->state->permissions & DST_PERM_WRITE)) {
181                 printk(KERN_INFO "DST node %s attached read-only.\n", n->name);
182                 set_disk_ro(n->disk, 1);
183         }
184
185         if (!idr_pre_get(&dst_index_idr, GFP_KERNEL))
186                 goto err_out_put;
187
188         mutex_lock(&dst_hash_lock);
189         err = idr_get_new(&dst_index_idr, NULL, &index);
190         mutex_unlock(&dst_hash_lock);
191         if (err)
192                 goto err_out_put;
193
194         n->disk->major = dst_major;
195         n->disk->first_minor = index;
196         n->disk->fops = &dst_blk_ops;
197         n->disk->queue = n->queue;
198         n->disk->private_data = n;
199         snprintf(n->disk->disk_name, sizeof(n->disk->disk_name), "dst-%s", n->name);
200
201         return 0;
202
203 err_out_put:
204         put_disk(n->disk);
205 err_out_free_queue:
206         blk_cleanup_queue(n->queue);
207 err_out_exit:
208         return err;
209 }
210
211 /*
212  * Sysfs machinery: show device's size.
213  */
214 static ssize_t dst_show_size(struct device *dev,
215                 struct device_attribute *attr, char *buf)
216 {
217         struct dst_info *info = container_of(dev, struct dst_info, device);
218
219         return sprintf(buf, "%llu\n", info->size);
220 }
221
222 /*
223  * Show local exported device.
224  */
225 static ssize_t dst_show_local(struct device *dev,
226                 struct device_attribute *attr, char *buf)
227 {
228         struct dst_info *info = container_of(dev, struct dst_info, device);
229
230         return sprintf(buf, "%s\n", info->local);
231 }
232
233 /*
234  * Shows type of the remote node - device major/minor number
235  * for local nodes and address (af_inet ipv4/ipv6 only) for remote nodes.
236  */
237 static ssize_t dst_show_type(struct device *dev,
238                 struct device_attribute *attr, char *buf)
239 {
240         struct dst_info *info = container_of(dev, struct dst_info, device);
241         int family = info->net.addr.sa_family;
242
243         if (family == AF_INET) {
244                 struct sockaddr_in *sin = (struct sockaddr_in *)&info->net.addr;
245                 return sprintf(buf, "%u.%u.%u.%u:%d\n",
246                         NIPQUAD(sin->sin_addr.s_addr), ntohs(sin->sin_port));
247         } else if (family == AF_INET6) {
248                 struct sockaddr_in6 *sin = (struct sockaddr_in6 *)&info->net.addr;
249                 return sprintf(buf,
250                         "%pi6:%d\n",
251                         &sin->sin6_addr, ntohs(sin->sin6_port));
252         } else {
253                 int i, sz = PAGE_SIZE - 2; /* 0 symbol and '\n' below */
254                 int size, addrlen = info->net.addr.sa_data_len;
255                 unsigned char *a = (unsigned char *)&info->net.addr.sa_data;
256                 char *buf_orig = buf;
257
258                 size = snprintf(buf, sz, "family: %d, addrlen: %u, addr: ",
259                                 family, addrlen);
260                 sz -= size;
261                 buf += size;
262
263                 for (i=0; i<addrlen; ++i) {
264                         if (sz < 3)
265                                 break;
266
267                         size = snprintf(buf, sz, "%02x ", a[i]);
268                         sz -= size;
269                         buf += size;
270                 }
271                 buf += sprintf(buf, "\n");
272
273                 return buf - buf_orig;
274         }
275         return 0;
276 }
277
278 static struct device_attribute dst_node_attrs[] = {
279         __ATTR(size, 0444, dst_show_size, NULL),
280         __ATTR(type, 0444, dst_show_type, NULL),
281         __ATTR(local, 0444, dst_show_local, NULL),
282 };
283
284 static int dst_create_node_attributes(struct dst_node *n)
285 {
286         int err, i;
287
288         for (i=0; i<ARRAY_SIZE(dst_node_attrs); ++i) {
289                 err = device_create_file(&n->info->device,
290                                 &dst_node_attrs[i]);
291                 if (err)
292                         goto err_out_remove_all;
293         }
294         return 0;
295
296 err_out_remove_all:
297         while (--i >= 0)
298                 device_remove_file(&n->info->device,
299                                 &dst_node_attrs[i]);
300
301         return err;
302 }
303
304 static void dst_remove_node_attributes(struct dst_node *n)
305 {
306         int i;
307
308         for (i=0; i<ARRAY_SIZE(dst_node_attrs); ++i)
309                 device_remove_file(&n->info->device,
310                                 &dst_node_attrs[i]);
311 }
312
313 /*
314  * Sysfs cleanup and initialization.
315  * Shows number of useful parameters.
316  */
317 static void dst_node_sysfs_exit(struct dst_node *n)
318 {
319         if (n->info) {
320                 dst_remove_node_attributes(n);
321                 device_unregister(&n->info->device);
322                 n->info = NULL;
323         }
324 }
325
326 static int dst_node_sysfs_init(struct dst_node *n)
327 {
328         int err;
329
330         n->info = kzalloc(sizeof(struct dst_info), GFP_KERNEL);
331         if (!n->info)
332                 return -ENOMEM;
333
334         memcpy(&n->info->device, &dst_node_dev, sizeof(struct device));
335         n->info->size = n->size;
336
337         dev_set_name(&n->info->device, "dst-%s", n->name);
338         err = device_register(&n->info->device);
339         if (err) {
340                 dprintk(KERN_ERR "Failed to register node '%s', err: %d.\n",
341                                 n->name, err);
342                 goto err_out_exit;
343         }
344
345         dst_create_node_attributes(n);
346
347         return 0;
348
349 err_out_exit:
350         kfree(n->info);
351         n->info = NULL;
352         return err;
353 }
354
355 /*
356  * DST node hash tables machinery.
357  */
358 static inline unsigned int dst_hash(char *str, unsigned int size)
359 {
360         return (jhash(str, size, 0) % dst_hashtable_size);
361 }
362
363 static void dst_node_remove(struct dst_node *n)
364 {
365         mutex_lock(&dst_hash_lock);
366         list_del_init(&n->node_entry);
367         mutex_unlock(&dst_hash_lock);
368 }
369
370 static void dst_node_add(struct dst_node *n)
371 {
372         unsigned hash = dst_hash(n->name, sizeof(n->name));
373
374         mutex_lock(&dst_hash_lock);
375         list_add_tail(&n->node_entry, &dst_hashtable[hash]);
376         mutex_unlock(&dst_hash_lock);
377 }
378
379 /*
380  * Cleaning node when it is about to be freed.
381  * There are still users of the socket though,
382  * so connection cleanup should be protected.
383  */
384 static void dst_node_cleanup(struct dst_node *n)
385 {
386         struct dst_state *st = n->state;
387
388         if (!st)
389                 return;
390
391         if (n->queue) {
392                 blk_cleanup_queue(n->queue);
393
394                 mutex_lock(&dst_hash_lock);
395                 idr_remove(&dst_index_idr, n->disk->first_minor);
396                 mutex_unlock(&dst_hash_lock);
397
398                 put_disk(n->disk);
399         }
400
401         if (n->bdev) {
402                 sync_blockdev(n->bdev);
403                 blkdev_put(n->bdev, FMODE_READ|FMODE_WRITE);
404         }
405
406         dst_state_lock(st);
407         st->need_exit = 1;
408         dst_state_exit_connected(st);
409         dst_state_unlock(st);
410
411         wake_up(&st->thread_wait);
412
413         dst_state_put(st);
414         n->state = NULL;
415 }
416
417 /*
418  * Free security attributes attached to given node.
419  */
420 static void dst_security_exit(struct dst_node *n)
421 {
422         struct dst_secure *s, *tmp;
423
424         list_for_each_entry_safe(s, tmp, &n->security_list, sec_entry) {
425                 list_del(&s->sec_entry);
426                 kfree(s);
427         }
428 }
429
430 /*
431  * Free node when there are no more users.
432  * Actually node has to be freed on behalf od userspace process,
433  * since there are number of threads, which are embedded in the
434  * node, so they can not exit and free node from there, that is
435  * why there is a wakeup if reference counter is not equal to zero.
436  */
437 void dst_node_put(struct dst_node *n)
438 {
439         if (unlikely(!n))
440                 return;
441
442         dprintk("%s: n: %p, refcnt: %d.\n",
443                         __func__, n, atomic_read(&n->refcnt));
444
445         if (atomic_dec_and_test(&n->refcnt)) {
446                 dst_node_remove(n);
447                 n->trans_scan_timeout = 0;
448                 dst_node_cleanup(n);
449                 thread_pool_destroy(n->pool);
450                 dst_node_sysfs_exit(n);
451                 dst_node_crypto_exit(n);
452                 dst_security_exit(n);
453                 dst_node_trans_exit(n);
454
455                 kfree(n);
456
457                 dprintk("%s: freed n: %p.\n", __func__, n);
458         } else {
459                 wake_up(&n->wait);
460         }
461 }
462
463 /*
464  * This function finds devices major/minor numbers for given pathname.
465  */
466 static int dst_lookup_device(const char *path, dev_t *dev)
467 {
468         int err;
469         struct nameidata nd;
470         struct inode *inode;
471
472         err = path_lookup(path, LOOKUP_FOLLOW, &nd);
473         if (err)
474                 return err;
475
476         inode = nd.path.dentry->d_inode;
477         if (!inode) {
478                 err = -ENOENT;
479                 goto out;
480         }
481
482         if (!S_ISBLK(inode->i_mode)) {
483                 err = -ENOTBLK;
484                 goto out;
485         }
486
487         *dev = inode->i_rdev;
488
489 out:
490         path_put(&nd.path);
491         return err;
492 }
493
494 /*
495  * Setting up export device: lookup by the name, get its size
496  * and setup listening socket, which will accept clients, which
497  * will submit IO for given storage.
498  */
499 static int dst_setup_export(struct dst_node *n, struct dst_ctl *ctl,
500                 struct dst_export_ctl *le)
501 {
502         int err;
503         dev_t dev = 0; /* gcc likes to scream here */
504
505         snprintf(n->info->local, sizeof(n->info->local), "%s", le->device);
506
507         err = dst_lookup_device(le->device, &dev);
508         if (err)
509                 return err;
510
511         n->bdev = open_by_devnum(dev, FMODE_READ|FMODE_WRITE);
512         if (!n->bdev)
513                 return -ENODEV;
514
515         if (n->size != 0)
516                 n->size = min_t(loff_t, n->bdev->bd_inode->i_size, n->size);
517         else
518                 n->size = n->bdev->bd_inode->i_size;
519
520         n->info->size = n->size;
521         err = dst_node_init_listened(n, le);
522         if (err)
523                 goto err_out_cleanup;
524
525         return 0;
526
527 err_out_cleanup:
528         blkdev_put(n->bdev, FMODE_READ|FMODE_WRITE);
529         n->bdev = NULL;
530
531         return err;
532 }
533
534 /* Empty thread pool callbacks for the network processing threads. */
535 static inline void *dst_thread_network_init(void *data)
536 {
537         dprintk("%s: data: %p.\n", __func__, data);
538         return data;
539 }
540
541 static inline void dst_thread_network_cleanup(void *data)
542 {
543         dprintk("%s: data: %p.\n", __func__, data);
544 }
545
546 /*
547  * Allocate DST node and initialize some of its parameters.
548  */
549 static struct dst_node *dst_alloc_node(struct dst_ctl *ctl,
550                 int (*start)(struct dst_node *),
551                 int num)
552 {
553         struct dst_node *n;
554         int err;
555
556         n = kzalloc(sizeof(struct dst_node), GFP_KERNEL);
557         if (!n)
558                 return NULL;
559
560         INIT_LIST_HEAD(&n->node_entry);
561
562         INIT_LIST_HEAD(&n->security_list);
563         mutex_init(&n->security_lock);
564
565         init_waitqueue_head(&n->wait);
566
567         n->trans_scan_timeout = msecs_to_jiffies(ctl->trans_scan_timeout);
568         if (!n->trans_scan_timeout)
569                 n->trans_scan_timeout = HZ;
570
571         n->trans_max_retries = ctl->trans_max_retries;
572         if (!n->trans_max_retries)
573                 n->trans_max_retries = 10;
574
575         /*
576          * Pretty much arbitrary default numbers.
577          * 32 matches maximum number of pages in bio originated from ext3 (31).
578          */
579         n->max_pages = ctl->max_pages;
580         if (!n->max_pages)
581                 n->max_pages = 32;
582
583         if (n->max_pages > 1024)
584                 n->max_pages = 1024;
585
586         n->start = start;
587         n->size = ctl->size;
588
589         atomic_set(&n->refcnt, 1);
590         atomic_long_set(&n->gen, 0);
591         snprintf(n->name, sizeof(n->name), "%s", ctl->name);
592
593         err = dst_node_sysfs_init(n);
594         if (err)
595                 goto err_out_free;
596
597         n->pool = thread_pool_create(num, n->name, dst_thread_network_init,
598                         dst_thread_network_cleanup, n);
599         if (IS_ERR(n->pool)) {
600                 err = PTR_ERR(n->pool);
601                 goto err_out_sysfs_exit;
602         }
603
604         dprintk("%s: n: %p, name: %s.\n", __func__, n, n->name);
605
606         return n;
607
608 err_out_sysfs_exit:
609         dst_node_sysfs_exit(n);
610 err_out_free:
611         kfree(n);
612         return NULL;
613 }
614
615 /*
616  * Starting a node, connected to the remote server:
617  * register block device and initialize transaction mechanism.
618  * In revers order though.
619  *
620  * It will autonegotiate some parameters with the remote node
621  * and update local if needed.
622  *
623  * Transaction initialization should be the last thing before
624  * starting the node, since transaction should include not only
625  * block IO, but also crypto related data (if any), which are
626  * initialized separately.
627  */
628 static int dst_start_remote(struct dst_node *n)
629 {
630         int err;
631
632         err = dst_node_trans_init(n, sizeof(struct dst_trans));
633         if (err)
634                 return err;
635
636         err = dst_node_create_disk(n);
637         if (err)
638                 return err;
639
640         dst_node_set_size(n);
641         add_disk(n->disk);
642
643         dprintk("DST: started remote node '%s', minor: %d.\n", n->name, n->disk->first_minor);
644
645         return 0;
646 }
647
648 /*
649  * Adding remote node and initialize connection.
650  */
651 static int dst_add_remote(struct dst_node *n, struct dst_ctl *ctl,
652                 void *data, unsigned int size)
653 {
654         int err;
655         struct dst_network_ctl *rctl = data;
656
657         if (n)
658                 return -EEXIST;
659
660         if (size != sizeof(struct dst_network_ctl))
661                 return -EINVAL;
662
663         n = dst_alloc_node(ctl, dst_start_remote, 1);
664         if (!n)
665                 return -ENOMEM;
666
667         memcpy(&n->info->net, rctl, sizeof(struct dst_network_ctl));
668         err = dst_node_init_connected(n, rctl);
669         if (err)
670                 goto err_out_free;
671
672         dst_node_add(n);
673
674         return 0;
675
676 err_out_free:
677         dst_node_put(n);
678         return err;
679 }
680
681 /*
682  * Adding export node: initializing block device and listening socket.
683  */
684 static int dst_add_export(struct dst_node *n, struct dst_ctl *ctl,
685                 void *data, unsigned int size)
686 {
687         int err;
688         struct dst_export_ctl *le = data;
689
690         if (n)
691                 return -EEXIST;
692
693         if (size != sizeof(struct dst_export_ctl))
694                 return -EINVAL;
695
696         n = dst_alloc_node(ctl, dst_start_export, 2);
697         if (!n)
698                 return -EINVAL;
699
700         err = dst_setup_export(n, ctl, le);
701         if (err)
702                 goto err_out_free;
703
704         dst_node_add(n);
705
706         return 0;
707
708 err_out_free:
709         dst_node_put(n);
710         return err;
711 }
712
713 static int dst_node_remove_unload(struct dst_node *n)
714 {
715         printk(KERN_INFO "STOPPED name: '%s', size: %llu.\n",
716                         n->name, n->size);
717
718         if (n->disk)
719                 del_gendisk(n->disk);
720
721         dst_node_remove(n);
722         dst_node_sysfs_exit(n);
723
724         /*
725          * This is not a hack. Really.
726          * Node's reference counter allows to implement fine grained
727          * node freeing, but since all transactions (which hold node's
728          * reference counter) are processed in the dedicated thread,
729          * it is possible that reference will hit zero in that thread,
730          * so we will not be able to exit thread and cleanup the node.
731          *
732          * So, we remove disk, so no new activity is possible, and
733          * wait until all pending transaction are completed (either
734          * in receiving thread or by timeout in workqueue), in this
735          * case reference counter will be less or equal to 2 (once set in
736          * dst_alloc_node() and then in connector message parser;
737          * or when we force module unloading, and connector message
738          * parser does not hold a reference, in this case reference
739          * counter will be equal to 1),
740          * and subsequent dst_node_put() calls will free the node.
741          */
742         dprintk("%s: going to sleep with %d refcnt.\n", __func__, atomic_read(&n->refcnt));
743         wait_event(n->wait, atomic_read(&n->refcnt) <= 2);
744
745         dst_node_put(n);
746         return 0;
747 }
748
749 /*
750  * Remove node from the hash table.
751  */
752 static int dst_del_node(struct dst_node *n, struct dst_ctl *ctl,
753                 void *data, unsigned int size)
754 {
755         if (!n)
756                 return -ENODEV;
757
758         return dst_node_remove_unload(n);
759 }
760
761 /*
762  * Initialize crypto processing for given node.
763  */
764 static int dst_crypto_init(struct dst_node *n, struct dst_ctl *ctl,
765                 void *data, unsigned int size)
766 {
767         struct dst_crypto_ctl *crypto = data;
768
769         if (!n)
770                 return -ENODEV;
771
772         if (size != sizeof(struct dst_crypto_ctl) + crypto->hash_keysize +
773                         crypto->cipher_keysize)
774                 return -EINVAL;
775
776         if (n->trans_cache)
777                 return -EEXIST;
778
779         return dst_node_crypto_init(n, crypto);
780 }
781
782 /*
783  * Security attributes for given node.
784  */
785 static int dst_security_init(struct dst_node *n, struct dst_ctl *ctl,
786                 void *data, unsigned int size)
787 {
788         struct dst_secure *s;
789
790         if (!n)
791                 return -ENODEV;
792
793         if (size != sizeof(struct dst_secure_user))
794                 return -EINVAL;
795
796         s = kmalloc(sizeof(struct dst_secure), GFP_KERNEL);
797         if (!s)
798                 return -ENOMEM;
799
800         memcpy(&s->sec, data, size);
801
802         mutex_lock(&n->security_lock);
803         list_add_tail(&s->sec_entry, &n->security_list);
804         mutex_unlock(&n->security_lock);
805
806         return 0;
807 }
808
809 /*
810  * Kill'em all!
811  */
812 static int dst_start_node(struct dst_node *n, struct dst_ctl *ctl,
813                 void *data, unsigned int size)
814 {
815         int err;
816
817         if (!n)
818                 return -ENODEV;
819
820         if (n->trans_cache)
821                 return 0;
822
823         err = n->start(n);
824         if (err)
825                 return err;
826
827         printk(KERN_INFO "STARTED name: '%s', size: %llu.\n", n->name, n->size);
828         return 0;
829 }
830
831 typedef int (*dst_command_func)(struct dst_node *n, struct dst_ctl *ctl,
832                 void *data, unsigned int size);
833
834 /*
835  * List of userspace commands.
836  */
837 static dst_command_func dst_commands[] = {
838         [DST_ADD_REMOTE] = &dst_add_remote,
839         [DST_ADD_EXPORT] = &dst_add_export,
840         [DST_DEL_NODE] = &dst_del_node,
841         [DST_CRYPTO] = &dst_crypto_init,
842         [DST_SECURITY] = &dst_security_init,
843         [DST_START] = &dst_start_node,
844 };
845
846 /*
847  * Configuration parser.
848  */
849 static void cn_dst_callback(struct cn_msg *msg)
850 {
851         struct dst_ctl *ctl;
852         int err;
853         struct dst_ctl_ack ack;
854         struct dst_node *n = NULL, *tmp;
855         unsigned int hash;
856
857         if (msg->len < sizeof(struct dst_ctl)) {
858                 err = -EBADMSG;
859                 goto out;
860         }
861
862         ctl = (struct dst_ctl *)msg->data;
863
864         if (ctl->cmd >= DST_CMD_MAX) {
865                 err = -EINVAL;
866                 goto out;
867         }
868         hash = dst_hash(ctl->name, sizeof(ctl->name));
869
870         mutex_lock(&dst_hash_lock);
871         list_for_each_entry(tmp, &dst_hashtable[hash], node_entry) {
872                 if (!memcmp(tmp->name, ctl->name, sizeof(tmp->name))) {
873                         n = tmp;
874                         dst_node_get(n);
875                         break;
876                 }
877         }
878         mutex_unlock(&dst_hash_lock);
879
880         err = dst_commands[ctl->cmd](n, ctl, msg->data + sizeof(struct dst_ctl),
881                         msg->len - sizeof(struct dst_ctl));
882
883         dst_node_put(n);
884 out:
885         memcpy(&ack.msg, msg, sizeof(struct cn_msg));
886
887         ack.msg.ack = msg->ack + 1;
888         ack.msg.len = sizeof(struct dst_ctl_ack) - sizeof(struct cn_msg);
889
890         ack.error = err;
891
892         cn_netlink_send(&ack.msg, 0, GFP_KERNEL);
893 }
894
895 /*
896  * Global initialization: sysfs, hash table, block device registration,
897  * connector and various caches.
898  */
899 static int __init dst_sysfs_init(void)
900 {
901         return bus_register(&dst_dev_bus_type);
902 }
903
904 static void dst_sysfs_exit(void)
905 {
906         bus_unregister(&dst_dev_bus_type);
907 }
908
909 static int __init dst_hashtable_init(void)
910 {
911         unsigned int i;
912
913         dst_hashtable = kcalloc(dst_hashtable_size, sizeof(struct list_head),
914                         GFP_KERNEL);
915         if (!dst_hashtable)
916                 return -ENOMEM;
917
918         for (i=0; i<dst_hashtable_size; ++i)
919                 INIT_LIST_HEAD(&dst_hashtable[i]);
920
921         return 0;
922 }
923
924 static void dst_hashtable_exit(void)
925 {
926         unsigned int i;
927         struct dst_node *n, *tmp;
928
929         for (i=0; i<dst_hashtable_size; ++i) {
930                 list_for_each_entry_safe(n, tmp, &dst_hashtable[i], node_entry) {
931                         dst_node_remove_unload(n);
932                 }
933         }
934
935         kfree(dst_hashtable);
936 }
937
938 static int __init dst_sys_init(void)
939 {
940         int err = -ENOMEM;
941
942         err = dst_hashtable_init();
943         if (err)
944                 goto err_out_exit;
945
946         err = dst_export_init();
947         if (err)
948                 goto err_out_hashtable_exit;
949
950         err = register_blkdev(dst_major, DST_NAME);
951         if (err < 0)
952                 goto err_out_export_exit;
953         if (err)
954                 dst_major = err;
955
956         err = dst_sysfs_init();
957         if (err)
958                 goto err_out_unregister;
959
960         err = cn_add_callback(&cn_dst_id, "DST", cn_dst_callback);
961         if (err)
962                 goto err_out_sysfs_exit;
963
964         printk(KERN_INFO "Distributed storage, '%s' release.\n", dst_name);
965
966         return 0;
967
968 err_out_sysfs_exit:
969         dst_sysfs_exit();
970 err_out_unregister:
971         unregister_blkdev(dst_major, DST_NAME);
972 err_out_export_exit:
973         dst_export_exit();
974 err_out_hashtable_exit:
975         dst_hashtable_exit();
976 err_out_exit:
977         return err;
978 }
979
980 static void __exit dst_sys_exit(void)
981 {
982         cn_del_callback(&cn_dst_id);
983         unregister_blkdev(dst_major, DST_NAME);
984         dst_hashtable_exit();
985         dst_sysfs_exit();
986         dst_export_exit();
987 }
988
989 module_init(dst_sys_init);
990 module_exit(dst_sys_exit);
991
992 MODULE_DESCRIPTION("Distributed storage");
993 MODULE_AUTHOR("Evgeniy Polyakov <zbr@ioremap.net>");
994 MODULE_LICENSE("GPL");