Whamcloud - gitweb
LU-5396 lov: (and ldlm) make some functions static
[fs/lustre-release.git] / lustre / lov / lov_object.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
18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
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 (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2011, 2013, Intel Corporation.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * Implementation of cl_object for LOV layer.
37  *
38  *   Author: Nikita Danilov <nikita.danilov@sun.com>
39  *   Author: Jinshan Xiong <jinshan.xiong@whamcloud.com>
40  */
41
42 #define DEBUG_SUBSYSTEM S_LOV
43
44 #include "lov_cl_internal.h"
45 #include "lclient.h"
46
47 /** \addtogroup lov
48  *  @{
49  */
50
51 /*****************************************************************************
52  *
53  * Layout operations.
54  *
55  */
56
57 struct lov_layout_operations {
58         int (*llo_init)(const struct lu_env *env, struct lov_device *dev,
59                         struct lov_object *lov,
60                         const struct cl_object_conf *conf,
61                         union lov_layout_state *state);
62         int (*llo_delete)(const struct lu_env *env, struct lov_object *lov,
63                            union lov_layout_state *state);
64         void (*llo_fini)(const struct lu_env *env, struct lov_object *lov,
65                          union lov_layout_state *state);
66         void (*llo_install)(const struct lu_env *env, struct lov_object *lov,
67                             union lov_layout_state *state);
68         int  (*llo_print)(const struct lu_env *env, void *cookie,
69                           lu_printer_t p, const struct lu_object *o);
70         int  (*llo_page_init)(const struct lu_env *env, struct cl_object *obj,
71                               struct cl_page *page, pgoff_t index);
72         int  (*llo_lock_init)(const struct lu_env *env,
73                               struct cl_object *obj, struct cl_lock *lock,
74                               const struct cl_io *io);
75         int  (*llo_io_init)(const struct lu_env *env,
76                             struct cl_object *obj, struct cl_io *io);
77         int  (*llo_getattr)(const struct lu_env *env, struct cl_object *obj,
78                             struct cl_attr *attr);
79 };
80
81 static int lov_layout_wait(const struct lu_env *env, struct lov_object *lov);
82
83 /*****************************************************************************
84  *
85  * Lov object layout operations.
86  *
87  */
88
89 static void lov_install_empty(const struct lu_env *env,
90                               struct lov_object *lov,
91                               union  lov_layout_state *state)
92 {
93         /*
94          * File without objects.
95          */
96 }
97
98 static int lov_init_empty(const struct lu_env *env,
99                           struct lov_device *dev, struct lov_object *lov,
100                           const struct cl_object_conf *conf,
101                           union  lov_layout_state *state)
102 {
103         return 0;
104 }
105
106 static void lov_install_raid0(const struct lu_env *env,
107                               struct lov_object *lov,
108                               union  lov_layout_state *state)
109 {
110 }
111
112 static struct cl_object *lov_sub_find(const struct lu_env *env,
113                                       struct cl_device *dev,
114                                       const struct lu_fid *fid,
115                                       const struct cl_object_conf *conf)
116 {
117         struct lu_object *o;
118
119         ENTRY;
120         o = lu_object_find_at(env, cl2lu_dev(dev), fid, &conf->coc_lu);
121         LASSERT(ergo(!IS_ERR(o), o->lo_dev->ld_type == &lovsub_device_type));
122         RETURN(lu2cl(o));
123 }
124
125 static int lov_init_sub(const struct lu_env *env, struct lov_object *lov,
126                         struct cl_object *stripe, struct lov_layout_raid0 *r0,
127                         int idx)
128 {
129         struct cl_object_header *hdr;
130         struct cl_object_header *subhdr;
131         struct cl_object_header *parent;
132         struct lov_oinfo        *oinfo;
133         int result;
134
135         if (OBD_FAIL_CHECK(OBD_FAIL_LOV_INIT)) {
136                 /* For sanity:test_206.
137                  * Do not leave the object in cache to avoid accessing
138                  * freed memory. This is because osc_object is referring to
139                  * lov_oinfo of lsm_stripe_data which will be freed due to
140                  * this failure. */
141                 cl_object_kill(env, stripe);
142                 cl_object_put(env, stripe);
143                 return -EIO;
144         }
145
146         hdr    = cl_object_header(lov2cl(lov));
147         subhdr = cl_object_header(stripe);
148
149         oinfo = lov->lo_lsm->lsm_oinfo[idx];
150         CDEBUG(D_INODE, DFID"@%p[%d] -> "DFID"@%p: ostid: "DOSTID
151                " idx: %d gen: %d\n",
152                PFID(&subhdr->coh_lu.loh_fid), subhdr, idx,
153                PFID(&hdr->coh_lu.loh_fid), hdr, POSTID(&oinfo->loi_oi),
154                oinfo->loi_ost_idx, oinfo->loi_ost_gen);
155
156         /* reuse ->coh_attr_guard to protect coh_parent change */
157         spin_lock(&subhdr->coh_attr_guard);
158         parent = subhdr->coh_parent;
159         if (parent == NULL) {
160                 subhdr->coh_parent = hdr;
161                 spin_unlock(&subhdr->coh_attr_guard);
162                 subhdr->coh_nesting = hdr->coh_nesting + 1;
163                 lu_object_ref_add(&stripe->co_lu, "lov-parent", lov);
164                 r0->lo_sub[idx] = cl2lovsub(stripe);
165                 r0->lo_sub[idx]->lso_super = lov;
166                 r0->lo_sub[idx]->lso_index = idx;
167                 result = 0;
168         } else {
169                 struct lu_object  *old_obj;
170                 struct lov_object *old_lov;
171                 unsigned int mask = D_INODE;
172
173                 spin_unlock(&subhdr->coh_attr_guard);
174                 old_obj = lu_object_locate(&parent->coh_lu, &lov_device_type);
175                 LASSERT(old_obj != NULL);
176                 old_lov = cl2lov(lu2cl(old_obj));
177                 if (old_lov->lo_layout_invalid) {
178                         /* the object's layout has already changed but isn't
179                          * refreshed */
180                         lu_object_unhash(env, &stripe->co_lu);
181                         result = -EAGAIN;
182                 } else {
183                         mask = D_ERROR;
184                         result = -EIO;
185                 }
186
187                 LU_OBJECT_DEBUG(mask, env, &stripe->co_lu,
188                                 "stripe %d is already owned.\n", idx);
189                 LU_OBJECT_DEBUG(mask, env, old_obj, "owned.\n");
190                 LU_OBJECT_HEADER(mask, env, lov2lu(lov), "try to own.\n");
191                 cl_object_put(env, stripe);
192         }
193         return result;
194 }
195
196 static int lov_page_slice_fixup(struct lov_object *lov,
197                                 struct cl_object *stripe)
198 {
199         struct cl_object_header *hdr = cl_object_header(&lov->lo_cl);
200         struct cl_object *o;
201
202         if (stripe == NULL)
203                 return hdr->coh_page_bufsize - lov->lo_cl.co_slice_off -
204                        cfs_size_round(sizeof(struct lov_page));
205
206         cl_object_for_each(o, stripe)
207                 o->co_slice_off += hdr->coh_page_bufsize;
208
209         return cl_object_header(stripe)->coh_page_bufsize;
210 }
211
212 static int lov_init_raid0(const struct lu_env *env,
213                           struct lov_device *dev, struct lov_object *lov,
214                           const struct cl_object_conf *conf,
215                           union  lov_layout_state *state)
216 {
217         int result;
218         int i;
219
220         struct cl_object        *stripe;
221         struct lov_thread_info  *lti     = lov_env_info(env);
222         struct cl_object_conf   *subconf = &lti->lti_stripe_conf;
223         struct lov_stripe_md    *lsm     = conf->u.coc_md->lsm;
224         struct lu_fid           *ofid    = &lti->lti_fid;
225         struct lov_layout_raid0 *r0      = &state->raid0;
226
227         ENTRY;
228
229         if (lsm->lsm_magic != LOV_MAGIC_V1 && lsm->lsm_magic != LOV_MAGIC_V3) {
230                 dump_lsm(D_ERROR, lsm);
231                 LASSERTF(0, "magic mismatch, expected %d/%d, actual %d.\n",
232                          LOV_MAGIC_V1, LOV_MAGIC_V3, lsm->lsm_magic);
233         }
234
235         LASSERT(lov->lo_lsm == NULL);
236         lov->lo_lsm = lsm_addref(lsm);
237         r0->lo_nr  = lsm->lsm_stripe_count;
238         LASSERT(r0->lo_nr <= lov_targets_nr(dev));
239
240         lov->lo_layout_invalid = true;
241
242         OBD_ALLOC_LARGE(r0->lo_sub, r0->lo_nr * sizeof r0->lo_sub[0]);
243         if (r0->lo_sub != NULL) {
244                 int psz = 0;
245
246                 result = 0;
247                 subconf->coc_inode = conf->coc_inode;
248                 spin_lock_init(&r0->lo_sub_lock);
249                 /*
250                  * Create stripe cl_objects.
251                  */
252                 for (i = 0; i < r0->lo_nr && result == 0; ++i) {
253                         struct cl_device *subdev;
254                         struct lov_oinfo *oinfo = lsm->lsm_oinfo[i];
255                         int ost_idx = oinfo->loi_ost_idx;
256
257                         if (lov_oinfo_is_dummy(oinfo))
258                                 continue;
259
260                         result = ostid_to_fid(ofid, &oinfo->loi_oi,
261                                               oinfo->loi_ost_idx);
262                         if (result != 0)
263                                 GOTO(out, result);
264
265                         subdev = lovsub2cl_dev(dev->ld_target[ost_idx]);
266                         subconf->u.coc_oinfo = oinfo;
267                         LASSERTF(subdev != NULL, "not init ost %d\n", ost_idx);
268                         /* In the function below, .hs_keycmp resolves to
269                          * lu_obj_hop_keycmp() */
270                         /* coverity[overrun-buffer-val] */
271                         stripe = lov_sub_find(env, subdev, ofid, subconf);
272                         if (!IS_ERR(stripe)) {
273                                 result = lov_init_sub(env, lov, stripe, r0, i);
274                                 if (result == -EAGAIN) { /* try again */
275                                         --i;
276                                         result = 0;
277                                         continue;
278                                 }
279                         } else {
280                                 result = PTR_ERR(stripe);
281                         }
282
283                         if (result == 0) {
284                                 int sz = lov_page_slice_fixup(lov, stripe);
285                                 LASSERT(ergo(psz > 0, psz == sz));
286                                 psz = sz;
287                         }
288                 }
289                 if (result == 0)
290                         cl_object_header(&lov->lo_cl)->coh_page_bufsize += psz;
291         } else
292                 result = -ENOMEM;
293 out:
294         RETURN(result);
295 }
296
297 static int lov_init_released(const struct lu_env *env,
298                         struct lov_device *dev, struct lov_object *lov,
299                         const struct cl_object_conf *conf,
300                         union  lov_layout_state *state)
301 {
302         struct lov_stripe_md *lsm = conf->u.coc_md->lsm;
303
304         LASSERT(lsm != NULL);
305         LASSERT(lsm_is_released(lsm));
306         LASSERT(lov->lo_lsm == NULL);
307
308         lov->lo_lsm = lsm_addref(lsm);
309         return 0;
310 }
311
312 static int lov_delete_empty(const struct lu_env *env, struct lov_object *lov,
313                             union lov_layout_state *state)
314 {
315         LASSERT(lov->lo_type == LLT_EMPTY || lov->lo_type == LLT_RELEASED);
316
317         lov_layout_wait(env, lov);
318         return 0;
319 }
320
321 static void lov_subobject_kill(const struct lu_env *env, struct lov_object *lov,
322                                struct lovsub_object *los, int idx)
323 {
324         struct cl_object        *sub;
325         struct lov_layout_raid0 *r0;
326         struct lu_site          *site;
327         struct lu_site_bkt_data *bkt;
328         wait_queue_t          *waiter;
329
330         r0  = &lov->u.raid0;
331         LASSERT(r0->lo_sub[idx] == los);
332
333         sub  = lovsub2cl(los);
334         site = sub->co_lu.lo_dev->ld_site;
335         bkt  = lu_site_bkt_from_fid(site, &sub->co_lu.lo_header->loh_fid);
336
337         cl_object_kill(env, sub);
338         /* release a reference to the sub-object and ... */
339         lu_object_ref_del(&sub->co_lu, "lov-parent", lov);
340         cl_object_put(env, sub);
341
342         /* ... wait until it is actually destroyed---sub-object clears its
343          * ->lo_sub[] slot in lovsub_object_fini() */
344         if (r0->lo_sub[idx] == los) {
345                 waiter = &lov_env_info(env)->lti_waiter;
346                 init_waitqueue_entry_current(waiter);
347                 add_wait_queue(&bkt->lsb_marche_funebre, waiter);
348                 set_current_state(TASK_UNINTERRUPTIBLE);
349                 while (1) {
350                         /* this wait-queue is signaled at the end of
351                          * lu_object_free(). */
352                         set_current_state(TASK_UNINTERRUPTIBLE);
353                         spin_lock(&r0->lo_sub_lock);
354                         if (r0->lo_sub[idx] == los) {
355                                 spin_unlock(&r0->lo_sub_lock);
356                                 waitq_wait(waiter, TASK_UNINTERRUPTIBLE);
357                         } else {
358                                 spin_unlock(&r0->lo_sub_lock);
359                                 set_current_state(TASK_RUNNING);
360                                 break;
361                         }
362                 }
363                 remove_wait_queue(&bkt->lsb_marche_funebre, waiter);
364         }
365         LASSERT(r0->lo_sub[idx] == NULL);
366 }
367
368 static int lov_delete_raid0(const struct lu_env *env, struct lov_object *lov,
369                             union lov_layout_state *state)
370 {
371         struct lov_layout_raid0 *r0 = &state->raid0;
372         struct lov_stripe_md    *lsm = lov->lo_lsm;
373         int i;
374
375         ENTRY;
376
377         dump_lsm(D_INODE, lsm);
378
379         lov_layout_wait(env, lov);
380         if (r0->lo_sub != NULL) {
381                 for (i = 0; i < r0->lo_nr; ++i) {
382                         struct lovsub_object *los = r0->lo_sub[i];
383
384                         if (los != NULL) {
385                                 cl_object_prune(env, &los->lso_cl);
386                                 /*
387                                  * If top-level object is to be evicted from
388                                  * the cache, so are its sub-objects.
389                                  */
390                                 lov_subobject_kill(env, lov, los, i);
391                         }
392                 }
393         }
394         RETURN(0);
395 }
396
397 static void lov_fini_empty(const struct lu_env *env, struct lov_object *lov,
398                            union lov_layout_state *state)
399 {
400         LASSERT(lov->lo_type == LLT_EMPTY || lov->lo_type == LLT_RELEASED);
401 }
402
403 static void lov_fini_raid0(const struct lu_env *env, struct lov_object *lov,
404                            union lov_layout_state *state)
405 {
406         struct lov_layout_raid0 *r0 = &state->raid0;
407         ENTRY;
408
409         if (r0->lo_sub != NULL) {
410                 OBD_FREE_LARGE(r0->lo_sub, r0->lo_nr * sizeof r0->lo_sub[0]);
411                 r0->lo_sub = NULL;
412         }
413
414         dump_lsm(D_INODE, lov->lo_lsm);
415         lov_free_memmd(&lov->lo_lsm);
416
417         EXIT;
418 }
419
420 static void lov_fini_released(const struct lu_env *env, struct lov_object *lov,
421                                 union lov_layout_state *state)
422 {
423         ENTRY;
424         dump_lsm(D_INODE, lov->lo_lsm);
425         lov_free_memmd(&lov->lo_lsm);
426         EXIT;
427 }
428
429 static int lov_print_empty(const struct lu_env *env, void *cookie,
430                            lu_printer_t p, const struct lu_object *o)
431 {
432         (*p)(env, cookie, "empty %d\n", lu2lov(o)->lo_layout_invalid);
433         return 0;
434 }
435
436 static int lov_print_raid0(const struct lu_env *env, void *cookie,
437                            lu_printer_t p, const struct lu_object *o)
438 {
439         struct lov_object       *lov = lu2lov(o);
440         struct lov_layout_raid0 *r0  = lov_r0(lov);
441         struct lov_stripe_md    *lsm = lov->lo_lsm;
442         int                      i;
443
444         (*p)(env, cookie, "stripes: %d, %s, lsm{%p 0x%08X %d %u %u}:\n",
445                 r0->lo_nr, lov->lo_layout_invalid ? "invalid" : "valid", lsm,
446                 lsm->lsm_magic, atomic_read(&lsm->lsm_refc),
447                 lsm->lsm_stripe_count, lsm->lsm_layout_gen);
448         for (i = 0; i < r0->lo_nr; ++i) {
449                 struct lu_object *sub;
450
451                 if (r0->lo_sub[i] != NULL) {
452                         sub = lovsub2lu(r0->lo_sub[i]);
453                         lu_object_print(env, cookie, p, sub);
454                 } else {
455                         (*p)(env, cookie, "sub %d absent\n", i);
456                 }
457         }
458         return 0;
459 }
460
461 static int lov_print_released(const struct lu_env *env, void *cookie,
462                                 lu_printer_t p, const struct lu_object *o)
463 {
464         struct lov_object       *lov = lu2lov(o);
465         struct lov_stripe_md    *lsm = lov->lo_lsm;
466
467         (*p)(env, cookie,
468                 "released: %s, lsm{%p 0x%08X %d %u %u}:\n",
469                 lov->lo_layout_invalid ? "invalid" : "valid", lsm,
470                 lsm->lsm_magic, atomic_read(&lsm->lsm_refc),
471                 lsm->lsm_stripe_count, lsm->lsm_layout_gen);
472         return 0;
473 }
474
475 /**
476  * Implements cl_object_operations::coo_attr_get() method for an object
477  * without stripes (LLT_EMPTY layout type).
478  *
479  * The only attributes this layer is authoritative in this case is
480  * cl_attr::cat_blocks---it's 0.
481  */
482 static int lov_attr_get_empty(const struct lu_env *env, struct cl_object *obj,
483                               struct cl_attr *attr)
484 {
485         attr->cat_blocks = 0;
486         return 0;
487 }
488
489 static int lov_attr_get_raid0(const struct lu_env *env, struct cl_object *obj,
490                               struct cl_attr *attr)
491 {
492         struct lov_object       *lov = cl2lov(obj);
493         struct lov_layout_raid0 *r0 = lov_r0(lov);
494         struct cl_attr          *lov_attr = &r0->lo_attr;
495         int                      result = 0;
496
497         ENTRY;
498
499         /* this is called w/o holding type guard mutex, so it must be inside
500          * an on going IO otherwise lsm may be replaced.
501          * LU-2117: it turns out there exists one exception. For mmaped files,
502          * the lock of those files may be requested in the other file's IO
503          * context, and this function is called in ccc_lock_state(), it will
504          * hit this assertion.
505          * Anyway, it's still okay to call attr_get w/o type guard as layout
506          * can't go if locks exist. */
507         /* LASSERT(atomic_read(&lsm->lsm_refc) > 1); */
508
509         if (!r0->lo_attr_valid) {
510                 struct lov_stripe_md    *lsm = lov->lo_lsm;
511                 struct ost_lvb          *lvb = &lov_env_info(env)->lti_lvb;
512                 __u64                    kms = 0;
513
514                 memset(lvb, 0, sizeof(*lvb));
515                 /* XXX: timestamps can be negative by sanity:test_39m,
516                  * how can it be? */
517                 lvb->lvb_atime = LLONG_MIN;
518                 lvb->lvb_ctime = LLONG_MIN;
519                 lvb->lvb_mtime = LLONG_MIN;
520
521                 /*
522                  * XXX that should be replaced with a loop over sub-objects,
523                  * doing cl_object_attr_get() on them. But for now, let's
524                  * reuse old lov code.
525                  */
526
527                 /*
528                  * XXX take lsm spin-lock to keep lov_merge_lvb_kms()
529                  * happy. It's not needed, because new code uses
530                  * ->coh_attr_guard spin-lock to protect consistency of
531                  * sub-object attributes.
532                  */
533                 lov_stripe_lock(lsm);
534                 result = lov_merge_lvb_kms(lsm, lvb, &kms);
535                 lov_stripe_unlock(lsm);
536                 if (result == 0) {
537                         cl_lvb2attr(lov_attr, lvb);
538                         lov_attr->cat_kms = kms;
539                         r0->lo_attr_valid = 1;
540                 }
541         }
542         if (result == 0) { /* merge results */
543                 attr->cat_blocks = lov_attr->cat_blocks;
544                 attr->cat_size = lov_attr->cat_size;
545                 attr->cat_kms = lov_attr->cat_kms;
546                 if (attr->cat_atime < lov_attr->cat_atime)
547                         attr->cat_atime = lov_attr->cat_atime;
548                 if (attr->cat_ctime < lov_attr->cat_ctime)
549                         attr->cat_ctime = lov_attr->cat_ctime;
550                 if (attr->cat_mtime < lov_attr->cat_mtime)
551                         attr->cat_mtime = lov_attr->cat_mtime;
552         }
553         RETURN(result);
554 }
555
556 const static struct lov_layout_operations lov_dispatch[] = {
557         [LLT_EMPTY] = {
558                 .llo_init      = lov_init_empty,
559                 .llo_delete    = lov_delete_empty,
560                 .llo_fini      = lov_fini_empty,
561                 .llo_install   = lov_install_empty,
562                 .llo_print     = lov_print_empty,
563                 .llo_page_init = lov_page_init_empty,
564                 .llo_lock_init = lov_lock_init_empty,
565                 .llo_io_init   = lov_io_init_empty,
566                 .llo_getattr   = lov_attr_get_empty
567         },
568         [LLT_RAID0] = {
569                 .llo_init      = lov_init_raid0,
570                 .llo_delete    = lov_delete_raid0,
571                 .llo_fini      = lov_fini_raid0,
572                 .llo_install   = lov_install_raid0,
573                 .llo_print     = lov_print_raid0,
574                 .llo_page_init = lov_page_init_raid0,
575                 .llo_lock_init = lov_lock_init_raid0,
576                 .llo_io_init   = lov_io_init_raid0,
577                 .llo_getattr   = lov_attr_get_raid0
578         },
579         [LLT_RELEASED] = {
580                 .llo_init      = lov_init_released,
581                 .llo_delete    = lov_delete_empty,
582                 .llo_fini      = lov_fini_released,
583                 .llo_install   = lov_install_empty,
584                 .llo_print     = lov_print_released,
585                 .llo_page_init = lov_page_init_empty,
586                 .llo_lock_init = lov_lock_init_empty,
587                 .llo_io_init   = lov_io_init_released,
588                 .llo_getattr   = lov_attr_get_empty
589         }
590 };
591
592 /**
593  * Performs a double-dispatch based on the layout type of an object.
594  */
595 #define LOV_2DISPATCH_NOLOCK(obj, op, ...)                              \
596 ({                                                                      \
597         struct lov_object                      *__obj = (obj);          \
598         enum lov_layout_type                    __llt;                  \
599                                                                         \
600         __llt = __obj->lo_type;                                         \
601         LASSERT(0 <= __llt && __llt < ARRAY_SIZE(lov_dispatch));        \
602         lov_dispatch[__llt].op(__VA_ARGS__);                            \
603 })
604
605 /**
606  * Return lov_layout_type associated with a given lsm
607  */
608 static enum lov_layout_type lov_type(struct lov_stripe_md *lsm)
609 {
610         if (lsm == NULL)
611                 return LLT_EMPTY;
612         if (lsm_is_released(lsm))
613                 return LLT_RELEASED;
614         return LLT_RAID0;
615 }
616
617 static inline void lov_conf_freeze(struct lov_object *lov)
618 {
619         if (lov->lo_owner != current)
620                 down_read(&lov->lo_type_guard);
621 }
622
623 static inline void lov_conf_thaw(struct lov_object *lov)
624 {
625         if (lov->lo_owner != current)
626                 up_read(&lov->lo_type_guard);
627 }
628
629 #define LOV_2DISPATCH_MAYLOCK(obj, op, lock, ...)                       \
630 ({                                                                      \
631         struct lov_object                      *__obj = (obj);          \
632         int                                     __lock = !!(lock);      \
633         typeof(lov_dispatch[0].op(__VA_ARGS__)) __result;               \
634                                                                         \
635         if (__lock)                                                     \
636                 lov_conf_freeze(__obj);                                 \
637         __result = LOV_2DISPATCH_NOLOCK(obj, op, __VA_ARGS__);          \
638         if (__lock)                                                     \
639                 lov_conf_thaw(__obj);                                   \
640         __result;                                                       \
641 })
642
643 /**
644  * Performs a locked double-dispatch based on the layout type of an object.
645  */
646 #define LOV_2DISPATCH(obj, op, ...)                     \
647         LOV_2DISPATCH_MAYLOCK(obj, op, 1, __VA_ARGS__)
648
649 #define LOV_2DISPATCH_VOID(obj, op, ...)                                \
650 do {                                                                    \
651         struct lov_object                      *__obj = (obj);          \
652         enum lov_layout_type                    __llt;                  \
653                                                                         \
654         lov_conf_freeze(__obj);                                         \
655         __llt = __obj->lo_type;                                         \
656         LASSERT(0 <= __llt && __llt < ARRAY_SIZE(lov_dispatch));        \
657         lov_dispatch[__llt].op(__VA_ARGS__);                            \
658         lov_conf_thaw(__obj);                                           \
659 } while (0)
660
661 static void lov_conf_lock(struct lov_object *lov)
662 {
663         LASSERT(lov->lo_owner != current);
664         down_write(&lov->lo_type_guard);
665         LASSERT(lov->lo_owner == NULL);
666         lov->lo_owner = current;
667 }
668
669 static void lov_conf_unlock(struct lov_object *lov)
670 {
671         lov->lo_owner = NULL;
672         up_write(&lov->lo_type_guard);
673 }
674
675 static int lov_layout_wait(const struct lu_env *env, struct lov_object *lov)
676 {
677         struct l_wait_info lwi = { 0 };
678         ENTRY;
679
680         while (atomic_read(&lov->lo_active_ios) > 0) {
681                 CDEBUG(D_INODE, "file:"DFID" wait for active IO, now: %d.\n",
682                         PFID(lu_object_fid(lov2lu(lov))),
683                         atomic_read(&lov->lo_active_ios));
684
685                 l_wait_event(lov->lo_waitq,
686                              atomic_read(&lov->lo_active_ios) == 0, &lwi);
687         }
688         RETURN(0);
689 }
690
691 static int lov_layout_change(const struct lu_env *unused,
692                              struct lov_object *lov,
693                              const struct cl_object_conf *conf)
694 {
695         int result;
696         enum lov_layout_type llt = LLT_EMPTY;
697         union lov_layout_state *state = &lov->u;
698         const struct lov_layout_operations *old_ops;
699         const struct lov_layout_operations *new_ops;
700
701         void *cookie;
702         struct lu_env *env;
703         int refcheck;
704         ENTRY;
705
706         LASSERT(0 <= lov->lo_type && lov->lo_type < ARRAY_SIZE(lov_dispatch));
707
708         if (conf->u.coc_md != NULL)
709                 llt = lov_type(conf->u.coc_md->lsm);
710         LASSERT(0 <= llt && llt < ARRAY_SIZE(lov_dispatch));
711
712         cookie = cl_env_reenter();
713         env = cl_env_get(&refcheck);
714         if (IS_ERR(env)) {
715                 cl_env_reexit(cookie);
716                 RETURN(PTR_ERR(env));
717         }
718
719         CDEBUG(D_INODE, DFID" from %s to %s\n",
720                PFID(lu_object_fid(lov2lu(lov))),
721                llt2str(lov->lo_type), llt2str(llt));
722
723         old_ops = &lov_dispatch[lov->lo_type];
724         new_ops = &lov_dispatch[llt];
725
726         result = cl_object_prune(env, &lov->lo_cl);
727         if (result != 0)
728                 GOTO(out, result);
729
730         result = old_ops->llo_delete(env, lov, &lov->u);
731         if (result == 0) {
732                 old_ops->llo_fini(env, lov, &lov->u);
733
734                 LASSERT(atomic_read(&lov->lo_active_ios) == 0);
735
736                 lov->lo_type = LLT_EMPTY;
737                 /* page bufsize fixup */
738                 cl_object_header(&lov->lo_cl)->coh_page_bufsize -=
739                         lov_page_slice_fixup(lov, NULL);
740
741                 result = new_ops->llo_init(env,
742                                         lu2lov_dev(lov->lo_cl.co_lu.lo_dev),
743                                         lov, conf, state);
744                 if (result == 0) {
745                         new_ops->llo_install(env, lov, state);
746                         lov->lo_type = llt;
747                 } else {
748                         new_ops->llo_delete(env, lov, state);
749                         new_ops->llo_fini(env, lov, state);
750                         /* this file becomes an EMPTY file. */
751                 }
752         }
753
754 out:
755         cl_env_put(env, &refcheck);
756         cl_env_reexit(cookie);
757         RETURN(result);
758 }
759
760 /*****************************************************************************
761  *
762  * Lov object operations.
763  *
764  */
765 int lov_object_init(const struct lu_env *env, struct lu_object *obj,
766                     const struct lu_object_conf *conf)
767 {
768         struct lov_device            *dev   = lu2lov_dev(obj->lo_dev);
769         struct lov_object            *lov   = lu2lov(obj);
770         const struct cl_object_conf  *cconf = lu2cl_conf(conf);
771         union  lov_layout_state      *set   = &lov->u;
772         const struct lov_layout_operations *ops;
773         int result;
774
775         ENTRY;
776         init_rwsem(&lov->lo_type_guard);
777         atomic_set(&lov->lo_active_ios, 0);
778         init_waitqueue_head(&lov->lo_waitq);
779
780         cl_object_page_init(lu2cl(obj), sizeof(struct lov_page));
781
782         /* no locking is necessary, as object is being created */
783         lov->lo_type = lov_type(cconf->u.coc_md->lsm);
784         ops = &lov_dispatch[lov->lo_type];
785         result = ops->llo_init(env, dev, lov, cconf, set);
786         if (result == 0)
787                 ops->llo_install(env, lov, set);
788         RETURN(result);
789 }
790
791 static int lov_conf_set(const struct lu_env *env, struct cl_object *obj,
792                         const struct cl_object_conf *conf)
793 {
794         struct lov_stripe_md    *lsm = NULL;
795         struct lov_object       *lov = cl2lov(obj);
796         int                      result = 0;
797         ENTRY;
798
799         lov_conf_lock(lov);
800         if (conf->coc_opc == OBJECT_CONF_INVALIDATE) {
801                 lov->lo_layout_invalid = true;
802                 GOTO(out, result = 0);
803         }
804
805         if (conf->coc_opc == OBJECT_CONF_WAIT) {
806                 if (lov->lo_layout_invalid &&
807                     atomic_read(&lov->lo_active_ios) > 0) {
808                         lov_conf_unlock(lov);
809                         result = lov_layout_wait(env, lov);
810                         lov_conf_lock(lov);
811                 }
812                 GOTO(out, result);
813         }
814
815         LASSERT(conf->coc_opc == OBJECT_CONF_SET);
816
817         if (conf->u.coc_md != NULL)
818                 lsm = conf->u.coc_md->lsm;
819         if ((lsm == NULL && lov->lo_lsm == NULL) ||
820             ((lsm != NULL && lov->lo_lsm != NULL) &&
821              (lov->lo_lsm->lsm_layout_gen == lsm->lsm_layout_gen) &&
822              (lov->lo_lsm->lsm_pattern == lsm->lsm_pattern))) {
823                 /* same version of layout */
824                 lov->lo_layout_invalid = false;
825                 GOTO(out, result = 0);
826         }
827
828         /* will change layout - check if there still exists active IO. */
829         if (atomic_read(&lov->lo_active_ios) > 0) {
830                 lov->lo_layout_invalid = true;
831                 GOTO(out, result = -EBUSY);
832         }
833
834         result = lov_layout_change(env, lov, conf);
835         lov->lo_layout_invalid = result != 0;
836         EXIT;
837
838 out:
839         lov_conf_unlock(lov);
840         CDEBUG(D_INODE, DFID" lo_layout_invalid=%d\n",
841                PFID(lu_object_fid(lov2lu(lov))), lov->lo_layout_invalid);
842         RETURN(result);
843 }
844
845 static void lov_object_delete(const struct lu_env *env, struct lu_object *obj)
846 {
847         struct lov_object *lov = lu2lov(obj);
848
849         ENTRY;
850         LOV_2DISPATCH_VOID(lov, llo_delete, env, lov, &lov->u);
851         EXIT;
852 }
853
854 static void lov_object_free(const struct lu_env *env, struct lu_object *obj)
855 {
856         struct lov_object *lov = lu2lov(obj);
857
858         ENTRY;
859         LOV_2DISPATCH_VOID(lov, llo_fini, env, lov, &lov->u);
860         lu_object_fini(obj);
861         OBD_SLAB_FREE_PTR(lov, lov_object_kmem);
862         EXIT;
863 }
864
865 static int lov_object_print(const struct lu_env *env, void *cookie,
866                             lu_printer_t p, const struct lu_object *o)
867 {
868         return LOV_2DISPATCH_NOLOCK(lu2lov(o), llo_print, env, cookie, p, o);
869 }
870
871 int lov_page_init(const struct lu_env *env, struct cl_object *obj,
872                   struct cl_page *page, pgoff_t index)
873 {
874         return LOV_2DISPATCH_NOLOCK(cl2lov(obj), llo_page_init, env, obj, page,
875                                     index);
876 }
877
878 /**
879  * Implements cl_object_operations::clo_io_init() method for lov
880  * layer. Dispatches to the appropriate layout io initialization method.
881  */
882 int lov_io_init(const struct lu_env *env, struct cl_object *obj,
883                 struct cl_io *io)
884 {
885         CL_IO_SLICE_CLEAN(lov_env_io(env), lis_cl);
886         return LOV_2DISPATCH_MAYLOCK(cl2lov(obj), llo_io_init,
887                                      !io->ci_ignore_layout, env, obj, io);
888 }
889
890 /**
891  * An implementation of cl_object_operations::clo_attr_get() method for lov
892  * layer. For raid0 layout this collects and merges attributes of all
893  * sub-objects.
894  */
895 static int lov_attr_get(const struct lu_env *env, struct cl_object *obj,
896                         struct cl_attr *attr)
897 {
898         /* do not take lock, as this function is called under a
899          * spin-lock. Layout is protected from changing by ongoing IO. */
900         return LOV_2DISPATCH_NOLOCK(cl2lov(obj), llo_getattr, env, obj, attr);
901 }
902
903 static int lov_attr_set(const struct lu_env *env, struct cl_object *obj,
904                         const struct cl_attr *attr, unsigned valid)
905 {
906         /*
907          * No dispatch is required here, as no layout implements this.
908          */
909         return 0;
910 }
911
912 int lov_lock_init(const struct lu_env *env, struct cl_object *obj,
913                   struct cl_lock *lock, const struct cl_io *io)
914 {
915         /* No need to lock because we've taken one refcount of layout.  */
916         return LOV_2DISPATCH_NOLOCK(cl2lov(obj), llo_lock_init, env, obj, lock,
917                                     io);
918 }
919
920 static const struct cl_object_operations lov_ops = {
921         .coo_page_init = lov_page_init,
922         .coo_lock_init = lov_lock_init,
923         .coo_io_init   = lov_io_init,
924         .coo_attr_get  = lov_attr_get,
925         .coo_attr_set  = lov_attr_set,
926         .coo_conf_set  = lov_conf_set
927 };
928
929 static const struct lu_object_operations lov_lu_obj_ops = {
930         .loo_object_init      = lov_object_init,
931         .loo_object_delete    = lov_object_delete,
932         .loo_object_release   = NULL,
933         .loo_object_free      = lov_object_free,
934         .loo_object_print     = lov_object_print,
935         .loo_object_invariant = NULL
936 };
937
938 struct lu_object *lov_object_alloc(const struct lu_env *env,
939                                    const struct lu_object_header *unused,
940                                    struct lu_device *dev)
941 {
942         struct lov_object *lov;
943         struct lu_object  *obj;
944
945         ENTRY;
946         OBD_SLAB_ALLOC_PTR_GFP(lov, lov_object_kmem, GFP_NOFS);
947         if (lov != NULL) {
948                 obj = lov2lu(lov);
949                 lu_object_init(obj, NULL, dev);
950                 lov->lo_cl.co_ops = &lov_ops;
951                 lov->lo_type = -1; /* invalid, to catch uninitialized type */
952                 /*
953                  * object io operation vector (cl_object::co_iop) is installed
954                  * later in lov_object_init(), as different vectors are used
955                  * for object with different layouts.
956                  */
957                 obj->lo_ops = &lov_lu_obj_ops;
958         } else
959                 obj = NULL;
960         RETURN(obj);
961 }
962
963 struct lov_stripe_md *lov_lsm_addref(struct lov_object *lov)
964 {
965         struct lov_stripe_md *lsm = NULL;
966
967         lov_conf_freeze(lov);
968         if (lov->lo_lsm != NULL) {
969                 lsm = lsm_addref(lov->lo_lsm);
970                 CDEBUG(D_INODE, "lsm %p addref %d/%d by %p.\n",
971                         lsm, atomic_read(&lsm->lsm_refc),
972                         lov->lo_layout_invalid, current);
973         }
974         lov_conf_thaw(lov);
975         return lsm;
976 }
977
978 struct lov_stripe_md *lov_lsm_get(struct cl_object *clobj)
979 {
980         struct lu_object *luobj;
981         struct lov_stripe_md *lsm = NULL;
982
983         if (clobj == NULL)
984                 return NULL;
985
986         luobj = lu_object_locate(&cl_object_header(clobj)->coh_lu,
987                                  &lov_device_type);
988         if (luobj != NULL)
989                 lsm = lov_lsm_addref(lu2lov(luobj));
990         return lsm;
991 }
992 EXPORT_SYMBOL(lov_lsm_get);
993
994 void lov_lsm_put(struct cl_object *unused, struct lov_stripe_md *lsm)
995 {
996         if (lsm != NULL)
997                 lov_free_memmd(&lsm);
998 }
999 EXPORT_SYMBOL(lov_lsm_put);
1000
1001 int lov_read_and_clear_async_rc(struct cl_object *clob)
1002 {
1003         struct lu_object *luobj;
1004         int rc = 0;
1005         ENTRY;
1006
1007         luobj = lu_object_locate(&cl_object_header(clob)->coh_lu,
1008                                  &lov_device_type);
1009         if (luobj != NULL) {
1010                 struct lov_object *lov = lu2lov(luobj);
1011
1012                 lov_conf_freeze(lov);
1013                 switch (lov->lo_type) {
1014                 case LLT_RAID0: {
1015                         struct lov_stripe_md *lsm;
1016                         int i;
1017
1018                         lsm = lov->lo_lsm;
1019                         LASSERT(lsm != NULL);
1020                         for (i = 0; i < lsm->lsm_stripe_count; i++) {
1021                                 struct lov_oinfo *loi = lsm->lsm_oinfo[i];
1022
1023                                 if (lov_oinfo_is_dummy(loi))
1024                                         continue;
1025
1026                                 if (loi->loi_ar.ar_rc && !rc)
1027                                         rc = loi->loi_ar.ar_rc;
1028                                 loi->loi_ar.ar_rc = 0;
1029                         }
1030                 }
1031                 case LLT_RELEASED:
1032                 case LLT_EMPTY:
1033                         break;
1034                 default:
1035                         LBUG();
1036                 }
1037                 lov_conf_thaw(lov);
1038         }
1039         RETURN(rc);
1040 }
1041 EXPORT_SYMBOL(lov_read_and_clear_async_rc);
1042
1043 /** @} lov */