Whamcloud - gitweb
df05ba9296780948a3d3aa214becc0225a9c60c5
[fs/lustre-release.git] / lustre / lod / lod_pool.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see [sun.com URL with a
18  * copy of GPLv2].
19  *
20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21  * CA 95054 USA or visit www.sun.com if you need additional information or
22  * have any questions.
23  *
24  * GPL HEADER END
25  */
26 /*
27  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
28  * Use is subject to license terms.
29  */
30 /*
31  * This file is part of Lustre, http://www.lustre.org/
32  * Lustre is a trademark of Sun Microsystems, Inc.
33  *
34  * lustre/lod/lod_pool.c
35  *
36  * OST pool methods
37  *
38  * Author: Jacques-Charles LAFOUCRIERE <jc.lafoucriere@cea.fr>
39  * Author: Alex Lyashkov <Alexey.Lyashkov@Sun.COM>
40  * Author: Nathaniel Rutman <Nathan.Rutman@Sun.COM>
41  */
42
43 #define DEBUG_SUBSYSTEM S_LOV
44
45 #include <libcfs/libcfs.h>
46 #include <obd.h>
47 #include "lod_internal.h"
48
49 #define pool_tgt(_p, _i) \
50         OST_TGT(lu2lod_dev((_p)->pool_lobd->obd_lu_dev),(_p)->pool_obds.op_array[_i])
51
52 static void lod_pool_getref(struct pool_desc *pool)
53 {
54         CDEBUG(D_INFO, "pool %p\n", pool);
55         atomic_inc(&pool->pool_refcount);
56 }
57
58 void lod_pool_putref(struct pool_desc *pool)
59 {
60         CDEBUG(D_INFO, "pool %p\n", pool);
61         if (atomic_dec_and_test(&pool->pool_refcount)) {
62                 LASSERT(cfs_hlist_unhashed(&pool->pool_hash));
63                 LASSERT(cfs_list_empty(&pool->pool_list));
64                 LASSERT(pool->pool_proc_entry == NULL);
65                 lod_ost_pool_free(&(pool->pool_rr.lqr_pool));
66                 lod_ost_pool_free(&(pool->pool_obds));
67                 OBD_FREE_PTR(pool);
68                 EXIT;
69         }
70 }
71
72 void lod_pool_putref_locked(struct pool_desc *pool)
73 {
74         CDEBUG(D_INFO, "pool %p\n", pool);
75         LASSERT(cfs_atomic_read(&pool->pool_refcount) > 1);
76
77         cfs_atomic_dec(&pool->pool_refcount);
78 }
79
80
81 /*
82  * hash function using a Rotating Hash algorithm
83  * Knuth, D. The Art of Computer Programming,
84  * Volume 3: Sorting and Searching,
85  * Chapter 6.4.
86  * Addison Wesley, 1973
87  */
88 static __u32 pool_hashfn(cfs_hash_t *hash_body, const void *key, unsigned mask)
89 {
90         int i;
91         __u32 result;
92         char *poolname;
93
94         result = 0;
95         poolname = (char *)key;
96         for (i = 0; i < LOV_MAXPOOLNAME; i++) {
97                 if (poolname[i] == '\0')
98                         break;
99                 result = (result << 4)^(result >> 28) ^  poolname[i];
100         }
101         return (result % mask);
102 }
103
104 static void *pool_key(cfs_hlist_node_t *hnode)
105 {
106         struct pool_desc *pool;
107
108         pool = cfs_hlist_entry(hnode, struct pool_desc, pool_hash);
109         return (pool->pool_name);
110 }
111
112 static int pool_hashkey_keycmp(const void *key, cfs_hlist_node_t *compared_hnode)
113 {
114         char *pool_name;
115         struct pool_desc *pool;
116
117         pool_name = (char *)key;
118         pool = cfs_hlist_entry(compared_hnode, struct pool_desc, pool_hash);
119         return !strncmp(pool_name, pool->pool_name, LOV_MAXPOOLNAME);
120 }
121
122 static void *pool_hashobject(cfs_hlist_node_t *hnode)
123 {
124         return cfs_hlist_entry(hnode, struct pool_desc, pool_hash);
125 }
126
127 static void pool_hashrefcount_get(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
128 {
129         struct pool_desc *pool;
130
131         pool = cfs_hlist_entry(hnode, struct pool_desc, pool_hash);
132         lod_pool_getref(pool);
133 }
134
135 static void pool_hashrefcount_put_locked(cfs_hash_t *hs,
136                 cfs_hlist_node_t *hnode)
137 {
138         struct pool_desc *pool;
139
140         pool = cfs_hlist_entry(hnode, struct pool_desc, pool_hash);
141         lod_pool_putref_locked(pool);
142 }
143
144 cfs_hash_ops_t pool_hash_operations = {
145         .hs_hash        = pool_hashfn,
146         .hs_key         = pool_key,
147         .hs_keycmp      = pool_hashkey_keycmp,
148         .hs_object      = pool_hashobject,
149         .hs_get         = pool_hashrefcount_get,
150         .hs_put_locked  = pool_hashrefcount_put_locked,
151 };
152
153 #ifdef LPROCFS
154 /* ifdef needed for liblustre support */
155 /*
156  * pool /proc seq_file methods
157  */
158 /*
159  * iterator is used to go through the target pool entries
160  * index is the current entry index in the lp_array[] array
161  * index >= pos returned to the seq_file interface
162  * pos is from 0 to (pool->pool_obds.op_count - 1)
163  */
164 #define POOL_IT_MAGIC 0xB001CEA0
165 struct pool_iterator {
166         int magic;
167         struct pool_desc *pool;
168         int idx;        /* from 0 to pool_tgt_size - 1 */
169 };
170
171 static void *pool_proc_next(struct seq_file *s, void *v, loff_t *pos)
172 {
173         struct pool_iterator *iter = (struct pool_iterator *)s->private;
174         int prev_idx;
175
176         LASSERTF(iter->magic == POOL_IT_MAGIC, "%08X", iter->magic);
177
178         /* test if end of file */
179         if (*pos >= pool_tgt_count(iter->pool))
180                 return NULL;
181
182         /* iterate to find a non empty entry */
183         prev_idx = iter->idx;
184         down_read(&pool_tgt_rw_sem(iter->pool));
185         iter->idx++;
186         if (iter->idx == pool_tgt_count(iter->pool)) {
187                 iter->idx = prev_idx; /* we stay on the last entry */
188                 up_read(&pool_tgt_rw_sem(iter->pool));
189                 return NULL;
190         }
191         up_read(&pool_tgt_rw_sem(iter->pool));
192         (*pos)++;
193         /* return != NULL to continue */
194         return iter;
195 }
196
197 static void *pool_proc_start(struct seq_file *s, loff_t *pos)
198 {
199         struct pool_desc *pool = (struct pool_desc *)s->private;
200         struct pool_iterator *iter;
201
202         lod_pool_getref(pool);
203         if ((pool_tgt_count(pool) == 0) ||
204             (*pos >= pool_tgt_count(pool))) {
205                 /* iter is not created, so stop() has no way to
206                  * find pool to dec ref */
207                 lod_pool_putref(pool);
208                 return NULL;
209         }
210
211         OBD_ALLOC_PTR(iter);
212         if (!iter)
213                 return ERR_PTR(-ENOMEM);
214         iter->magic = POOL_IT_MAGIC;
215         iter->pool = pool;
216         iter->idx = 0;
217
218         /* we use seq_file private field to memorized iterator so
219          * we can free it at stop() */
220         /* /!\ do not forget to restore it to pool before freeing it */
221         s->private = iter;
222         if (*pos > 0) {
223                 loff_t i;
224                 void *ptr;
225
226                 i = 0;
227                 do {
228                         ptr = pool_proc_next(s, &iter, &i);
229                 } while ((i < *pos) && (ptr != NULL));
230                 return ptr;
231         }
232         return iter;
233 }
234
235 static void pool_proc_stop(struct seq_file *s, void *v)
236 {
237         struct pool_iterator *iter = (struct pool_iterator *)s->private;
238
239         /* in some cases stop() method is called 2 times, without
240          * calling start() method (see seq_read() from fs/seq_file.c)
241          * we have to free only if s->private is an iterator */
242         if ((iter) && (iter->magic == POOL_IT_MAGIC)) {
243                 /* we restore s->private so next call to pool_proc_start()
244                  * will work */
245                 s->private = iter->pool;
246                 lod_pool_putref(iter->pool);
247                 OBD_FREE_PTR(iter);
248         }
249         return;
250 }
251
252 static int pool_proc_show(struct seq_file *s, void *v)
253 {
254         struct pool_iterator *iter = (struct pool_iterator *)v;
255         struct lod_ost_desc  *osc_desc;
256
257         LASSERTF(iter->magic == POOL_IT_MAGIC, "%08X", iter->magic);
258         LASSERT(iter->pool != NULL);
259         LASSERT(iter->idx <= pool_tgt_count(iter->pool));
260
261         down_read(&pool_tgt_rw_sem(iter->pool));
262         osc_desc = pool_tgt(iter->pool, iter->idx);
263         up_read(&pool_tgt_rw_sem(iter->pool));
264         if (osc_desc)
265                 seq_printf(s, "%s\n", obd_uuid2str(&(osc_desc->ltd_uuid)));
266
267         return 0;
268 }
269
270 static struct seq_operations pool_proc_ops = {
271         .start  = pool_proc_start,
272         .next   = pool_proc_next,
273         .stop   = pool_proc_stop,
274         .show   = pool_proc_show,
275 };
276
277 static int pool_proc_open(struct inode *inode, struct file *file)
278 {
279         int rc;
280
281         rc = seq_open(file, &pool_proc_ops);
282         if (!rc) {
283                 struct seq_file *s = file->private_data;
284                 s->private = PROC_I(inode)->pde->data;
285         }
286         return rc;
287 }
288
289 static struct file_operations pool_proc_operations = {
290         .open           = pool_proc_open,
291         .read           = seq_read,
292         .llseek         = seq_lseek,
293         .release        = seq_release,
294 };
295 #endif /* LPROCFS */
296
297 void lod_dump_pool(int level, struct pool_desc *pool)
298 {
299         int i;
300
301         lod_pool_getref(pool);
302
303         CDEBUG(level, "pool "LOV_POOLNAMEF" has %d members\n",
304                pool->pool_name, pool->pool_obds.op_count);
305         down_read(&pool_tgt_rw_sem(pool));
306
307         for (i = 0; i < pool_tgt_count(pool) ; i++) {
308                 if (!pool_tgt(pool, i) || !(pool_tgt(pool, i))->ltd_exp)
309                         continue;
310                 CDEBUG(level, "pool "LOV_POOLNAMEF"[%d] = %s\n",
311                        pool->pool_name, i,
312                        obd_uuid2str(&((pool_tgt(pool, i))->ltd_uuid)));
313         }
314
315         up_read(&pool_tgt_rw_sem(pool));
316         lod_pool_putref(pool);
317 }
318
319 #define LOD_POOL_INIT_COUNT 2
320 int lod_ost_pool_init(struct ost_pool *op, unsigned int count)
321 {
322         ENTRY;
323
324         if (count == 0)
325                 count = LOD_POOL_INIT_COUNT;
326         op->op_array = NULL;
327         op->op_count = 0;
328         init_rwsem(&op->op_rw_sem);
329         op->op_size = count;
330         OBD_ALLOC(op->op_array, op->op_size * sizeof(op->op_array[0]));
331         if (op->op_array == NULL) {
332                 op->op_size = 0;
333                 RETURN(-ENOMEM);
334         }
335         EXIT;
336         return 0;
337 }
338
339 /* Caller must hold write op_rwlock */
340 int lod_ost_pool_extend(struct ost_pool *op, unsigned int min_count)
341 {
342         __u32 *new;
343         int new_size;
344
345         LASSERT(min_count != 0);
346
347         if (op->op_count < op->op_size)
348                 return 0;
349
350         new_size = max(min_count, 2 * op->op_size);
351         OBD_ALLOC(new, new_size * sizeof(op->op_array[0]));
352         if (new == NULL)
353                 return -ENOMEM;
354
355         /* copy old array to new one */
356         memcpy(new, op->op_array, op->op_size * sizeof(op->op_array[0]));
357         OBD_FREE(op->op_array, op->op_size * sizeof(op->op_array[0]));
358         op->op_array = new;
359         op->op_size = new_size;
360         return 0;
361 }
362
363 int lod_ost_pool_add(struct ost_pool *op, __u32 idx, unsigned int min_count)
364 {
365         int rc = 0, i;
366         ENTRY;
367
368         down_write(&op->op_rw_sem);
369
370         rc = lod_ost_pool_extend(op, min_count);
371         if (rc)
372                 GOTO(out, rc);
373
374         /* search ost in pool array */
375         for (i = 0; i < op->op_count; i++) {
376                 if (op->op_array[i] == idx)
377                         GOTO(out, rc = -EEXIST);
378         }
379         /* ost not found we add it */
380         op->op_array[op->op_count] = idx;
381         op->op_count++;
382         EXIT;
383 out:
384         up_write(&op->op_rw_sem);
385         return rc;
386 }
387
388 int lod_ost_pool_remove(struct ost_pool *op, __u32 idx)
389 {
390         int i;
391         ENTRY;
392
393         down_write(&op->op_rw_sem);
394
395         for (i = 0; i < op->op_count; i++) {
396                 if (op->op_array[i] == idx) {
397                         memmove(&op->op_array[i], &op->op_array[i + 1],
398                                 (op->op_count - i - 1) * sizeof(op->op_array[0]));
399                         op->op_count--;
400                         up_write(&op->op_rw_sem);
401                         EXIT;
402                         return 0;
403                 }
404         }
405
406         up_write(&op->op_rw_sem);
407         RETURN(-EINVAL);
408 }
409
410 int lod_ost_pool_free(struct ost_pool *op)
411 {
412         ENTRY;
413
414         if (op->op_size == 0)
415                 RETURN(0);
416
417         down_write(&op->op_rw_sem);
418
419         OBD_FREE(op->op_array, op->op_size * sizeof(op->op_array[0]));
420         op->op_array = NULL;
421         op->op_count = 0;
422         op->op_size = 0;
423
424         up_write(&op->op_rw_sem);
425         RETURN(0);
426 }
427
428 int lod_pool_new(struct obd_device *obd, char *poolname)
429 {
430         struct lod_device *lod = lu2lod_dev(obd->obd_lu_dev);
431         struct pool_desc  *new_pool;
432         int rc;
433         ENTRY;
434
435         if (strlen(poolname) > LOV_MAXPOOLNAME)
436                 RETURN(-ENAMETOOLONG);
437
438         OBD_ALLOC_PTR(new_pool);
439         if (new_pool == NULL)
440                 RETURN(-ENOMEM);
441
442         strncpy(new_pool->pool_name, poolname, LOV_MAXPOOLNAME);
443         new_pool->pool_name[LOV_MAXPOOLNAME] = '\0';
444         new_pool->pool_lobd = obd;
445         /* ref count init to 1 because when created a pool is always used
446          * up to deletion
447          */
448         atomic_set(&new_pool->pool_refcount, 1);
449         rc = lod_ost_pool_init(&new_pool->pool_obds, 0);
450         if (rc)
451                 GOTO(out_err, rc);
452
453         memset(&(new_pool->pool_rr), 0, sizeof(struct lov_qos_rr));
454         rc = lod_ost_pool_init(&new_pool->pool_rr.lqr_pool, 0);
455         if (rc)
456                 GOTO(out_free_pool_obds, rc);
457
458         INIT_HLIST_NODE(&new_pool->pool_hash);
459
460 #ifdef LPROCFS
461         lod_pool_getref(new_pool);
462         new_pool->pool_proc_entry = lprocfs_add_simple(lod->lod_pool_proc_entry,
463                                                        poolname, NULL, NULL,
464                                                        new_pool,
465                                                        &pool_proc_operations);
466         if (IS_ERR(new_pool->pool_proc_entry)) {
467                 CWARN("Cannot add proc pool entry "LOV_POOLNAMEF"\n", poolname);
468                 new_pool->pool_proc_entry = NULL;
469                 lod_pool_putref(new_pool);
470         }
471         CDEBUG(D_INFO, "pool %p - proc %p\n", new_pool, new_pool->pool_proc_entry);
472 #endif
473
474         spin_lock(&obd->obd_dev_lock);
475         cfs_list_add_tail(&new_pool->pool_list, &lod->lod_pool_list);
476         lod->lod_pool_count++;
477         spin_unlock(&obd->obd_dev_lock);
478
479         /* add to find only when it fully ready  */
480         rc = cfs_hash_add_unique(lod->lod_pools_hash_body, poolname,
481                                  &new_pool->pool_hash);
482         if (rc)
483                 GOTO(out_err, rc = -EEXIST);
484
485         CDEBUG(D_CONFIG, LOV_POOLNAMEF" is pool #%d\n",
486                         poolname, lod->lod_pool_count);
487
488         RETURN(0);
489
490 out_err:
491         spin_lock(&obd->obd_dev_lock);
492         cfs_list_del_init(&new_pool->pool_list);
493         lod->lod_pool_count--;
494         spin_unlock(&obd->obd_dev_lock);
495
496         lprocfs_remove(&new_pool->pool_proc_entry);
497
498         lod_ost_pool_free(&new_pool->pool_rr.lqr_pool);
499 out_free_pool_obds:
500         lod_ost_pool_free(&new_pool->pool_obds);
501         OBD_FREE_PTR(new_pool);
502         return rc;
503 }
504
505 int lod_pool_del(struct obd_device *obd, char *poolname)
506 {
507         struct lod_device *lod = lu2lod_dev(obd->obd_lu_dev);
508         struct pool_desc  *pool;
509         ENTRY;
510
511         /* lookup and kill hash reference */
512         pool = cfs_hash_del_key(lod->lod_pools_hash_body, poolname);
513         if (pool == NULL)
514                 RETURN(-ENOENT);
515
516         if (pool->pool_proc_entry != NULL) {
517                 CDEBUG(D_INFO, "proc entry %p\n", pool->pool_proc_entry);
518                 lprocfs_remove(&pool->pool_proc_entry);
519                 lod_pool_putref(pool);
520         }
521
522         spin_lock(&obd->obd_dev_lock);
523         cfs_list_del_init(&pool->pool_list);
524         lod->lod_pool_count--;
525         spin_unlock(&obd->obd_dev_lock);
526
527         /* release last reference */
528         lod_pool_putref(pool);
529
530         RETURN(0);
531 }
532
533
534 int lod_pool_add(struct obd_device *obd, char *poolname, char *ostname)
535 {
536         struct lod_device *lod = lu2lod_dev(obd->obd_lu_dev);
537         struct obd_uuid    ost_uuid;
538         struct pool_desc  *pool;
539         unsigned int       idx;
540         int                rc = -EINVAL;
541         ENTRY;
542
543         pool = cfs_hash_lookup(lod->lod_pools_hash_body, poolname);
544         if (pool == NULL)
545                 RETURN(-ENOENT);
546
547         obd_str2uuid(&ost_uuid, ostname);
548
549         /* search ost in lod array */
550         lod_getref(lod);
551         lod_foreach_ost(lod, idx) {
552                 if (obd_uuid_equals(&ost_uuid, &OST_TGT(lod, idx)->ltd_uuid)) {
553                         rc = 0;
554                         break;
555                 }
556         }
557
558         if (rc)
559                 GOTO(out, rc);
560
561         rc = lod_ost_pool_add(&pool->pool_obds, idx, lod->lod_osts_size);
562         if (rc)
563                 GOTO(out, rc);
564
565         pool->pool_rr.lqr_dirty = 1;
566
567         CDEBUG(D_CONFIG, "Added %s to "LOV_POOLNAMEF" as member %d\n",
568                         ostname, poolname,  pool_tgt_count(pool));
569
570         EXIT;
571 out:
572         lod_putref(lod);
573         lod_pool_putref(pool);
574         return rc;
575 }
576
577 int lod_pool_remove(struct obd_device *obd, char *poolname, char *ostname)
578 {
579         struct lod_device *lod = lu2lod_dev(obd->obd_lu_dev);
580         struct obd_uuid    ost_uuid;
581         struct pool_desc  *pool;
582         unsigned int       idx;
583         int                rc = -EINVAL;
584         ENTRY;
585
586         pool = cfs_hash_lookup(lod->lod_pools_hash_body, poolname);
587         if (pool == NULL)
588                 RETURN(-ENOENT);
589
590         obd_str2uuid(&ost_uuid, ostname);
591
592         lod_getref(lod);
593         /* search ost in lod array, to get index */
594         cfs_foreach_bit(lod->lod_ost_bitmap, idx) {
595                 if (obd_uuid_equals(&ost_uuid, &OST_TGT(lod, idx)->ltd_uuid)) {
596                         rc = 0;
597                         break;
598                 }
599         }
600
601         /* test if ost found in lod array */
602         if (rc)
603                 GOTO(out, rc);
604
605         lod_ost_pool_remove(&pool->pool_obds, idx);
606
607         pool->pool_rr.lqr_dirty = 1;
608
609         CDEBUG(D_CONFIG, "%s removed from "LOV_POOLNAMEF"\n", ostname,
610                poolname);
611
612         EXIT;
613 out:
614         lod_putref(lod);
615         lod_pool_putref(pool);
616         return rc;
617 }
618
619 int lod_check_index_in_pool(__u32 idx, struct pool_desc *pool)
620 {
621         int i, rc;
622         ENTRY;
623
624         /* caller may no have a ref on pool if it got the pool
625          * without calling lod_find_pool() (e.g. go through the lod pool
626          * list)
627          */
628         lod_pool_getref(pool);
629
630         down_read(&pool_tgt_rw_sem(pool));
631
632         for (i = 0; i < pool_tgt_count(pool); i++) {
633                 if (pool_tgt_array(pool)[i] == idx)
634                         GOTO(out, rc = 0);
635         }
636         rc = -ENOENT;
637         EXIT;
638 out:
639         up_read(&pool_tgt_rw_sem(pool));
640
641         lod_pool_putref(pool);
642         return rc;
643 }
644
645 struct pool_desc *lod_find_pool(struct lod_device *lod, char *poolname)
646 {
647         struct pool_desc *pool;
648
649         pool = NULL;
650         if (poolname[0] != '\0') {
651                 pool = cfs_hash_lookup(lod->lod_pools_hash_body, poolname);
652                 if (pool == NULL)
653                         CWARN("Request for an unknown pool ("LOV_POOLNAMEF")\n",
654                               poolname);
655                 if ((pool != NULL) && (pool_tgt_count(pool) == 0)) {
656                         CWARN("Request for an empty pool ("LOV_POOLNAMEF")\n",
657                               poolname);
658                         /* pool is ignored, so we remove ref on it */
659                         lod_pool_putref(pool);
660                         pool = NULL;
661                 }
662         }
663         return pool;
664 }
665