Whamcloud - gitweb
LU-4423 lov: use correct env in lov_io_data_version_end()
[fs/lustre-release.git] / lustre / lov / lov_io.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.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2011, 2017, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  * Lustre is a trademark of Sun Microsystems, Inc.
31  *
32  * Implementation of cl_io for LOV layer.
33  *
34  *   Author: Nikita Danilov <nikita.danilov@sun.com>
35  *   Author: Jinshan Xiong <jinshan.xiong@whamcloud.com>
36  */
37
38 #define DEBUG_SUBSYSTEM S_LOV
39
40 #include "lov_cl_internal.h"
41
42 /** \addtogroup lov
43  *  @{
44  */
45
46 static inline struct lov_io_sub *lov_sub_alloc(struct lov_io *lio, int index)
47 {
48         struct lov_io_sub *sub;
49
50         if (lio->lis_nr_subios == 0) {
51                 LASSERT(lio->lis_single_subio_index == -1);
52                 sub = &lio->lis_single_subio;
53                 lio->lis_single_subio_index = index;
54                 memset(sub, 0, sizeof(*sub));
55         } else {
56                 OBD_ALLOC_PTR(sub);
57         }
58
59         if (sub != NULL) {
60                 INIT_LIST_HEAD(&sub->sub_list);
61                 INIT_LIST_HEAD(&sub->sub_linkage);
62                 sub->sub_subio_index = index;
63         }
64
65         return sub;
66 }
67
68 static inline void lov_sub_free(struct lov_io *lio, struct lov_io_sub *sub)
69 {
70         if (sub->sub_subio_index == lio->lis_single_subio_index) {
71                 LASSERT(sub == &lio->lis_single_subio);
72                 lio->lis_single_subio_index = -1;
73         } else {
74                 OBD_FREE_PTR(sub);
75         }
76 }
77
78 static void lov_io_sub_fini(const struct lu_env *env, struct lov_io *lio,
79                             struct lov_io_sub *sub)
80 {
81         ENTRY;
82
83         cl_io_fini(sub->sub_env, &sub->sub_io);
84
85         if (sub->sub_env != NULL && !IS_ERR(sub->sub_env)) {
86                 cl_env_put(sub->sub_env, &sub->sub_refcheck);
87                 sub->sub_env = NULL;
88         }
89         EXIT;
90 }
91
92 static inline bool
93 is_index_within_mirror(struct lov_object *lov, int index, int mirror_index)
94 {
95         struct lov_layout_composite *comp = &lov->u.composite;
96         struct lov_mirror_entry *lre = &comp->lo_mirrors[mirror_index];
97
98         return (index >= lre->lre_start && index <= lre->lre_end);
99 }
100
101 static int lov_io_sub_init(const struct lu_env *env, struct lov_io *lio,
102                            struct lov_io_sub *sub)
103 {
104         struct lov_object *lov = lio->lis_object;
105         struct cl_io *sub_io;
106         struct cl_object *sub_obj;
107         struct cl_io *io = lio->lis_cl.cis_io;
108         int index = lov_comp_entry(sub->sub_subio_index);
109         int stripe = lov_comp_stripe(sub->sub_subio_index);
110         int result = 0;
111         LASSERT(sub->sub_env == NULL);
112         ENTRY;
113
114         if (unlikely(!lov_r0(lov, index)->lo_sub ||
115                      !lov_r0(lov, index)->lo_sub[stripe]))
116                 RETURN(-EIO);
117
118         LASSERTF(is_index_within_mirror(lov, index, lio->lis_mirror_index),
119                  DFID "iot = %d, index = %d, mirror = %d\n",
120                  PFID(lu_object_fid(lov2lu(lov))), io->ci_type, index,
121                  lio->lis_mirror_index);
122
123         /* obtain new environment */
124         sub->sub_env = cl_env_get(&sub->sub_refcheck);
125         if (IS_ERR(sub->sub_env))
126                 result = PTR_ERR(sub->sub_env);
127
128         sub_obj = lovsub2cl(lov_r0(lov, index)->lo_sub[stripe]);
129         sub_io  = &sub->sub_io;
130
131         sub_io->ci_obj    = sub_obj;
132         sub_io->ci_result = 0;
133
134         sub_io->ci_parent  = io;
135         sub_io->ci_lockreq = io->ci_lockreq;
136         sub_io->ci_type    = io->ci_type;
137         sub_io->ci_no_srvlock = io->ci_no_srvlock;
138         sub_io->ci_noatime = io->ci_noatime;
139         sub_io->ci_pio = io->ci_pio;
140         sub_io->ci_lock_no_expand = io->ci_lock_no_expand;
141         sub_io->ci_ndelay = io->ci_ndelay;
142         sub_io->ci_layout_version = io->ci_layout_version;
143
144         result = cl_io_sub_init(sub->sub_env, sub_io, io->ci_type, sub_obj);
145
146         if (result < 0)
147                 lov_io_sub_fini(env, lio, sub);
148
149         RETURN(result);
150 }
151
152 struct lov_io_sub *lov_sub_get(const struct lu_env *env,
153                                struct lov_io *lio, int index)
154 {
155         struct lov_io_sub *sub;
156         int rc = 0;
157
158         ENTRY;
159
160         list_for_each_entry(sub, &lio->lis_subios, sub_list) {
161                 if (sub->sub_subio_index == index) {
162                         rc = 1;
163                         break;
164                 }
165         }
166
167         if (rc == 0) {
168                 sub = lov_sub_alloc(lio, index);
169                 if (sub == NULL)
170                         GOTO(out, rc = -ENOMEM);
171
172                 rc = lov_io_sub_init(env, lio, sub);
173                 if (rc < 0) {
174                         lov_sub_free(lio, sub);
175                         GOTO(out, rc);
176                 }
177
178                 list_add_tail(&sub->sub_list, &lio->lis_subios);
179                 lio->lis_nr_subios++;
180         }
181 out:
182         if (rc < 0)
183                 sub = ERR_PTR(rc);
184         RETURN(sub);
185 }
186
187 /*****************************************************************************
188  *
189  * Lov io operations.
190  *
191  */
192
193 int lov_page_index(const struct cl_page *page)
194 {
195         const struct cl_page_slice *slice;
196         ENTRY;
197
198         slice = cl_page_at(page, &lov_device_type);
199         LASSERT(slice != NULL);
200         LASSERT(slice->cpl_obj != NULL);
201
202         RETURN(cl2lov_page(slice)->lps_index);
203 }
204
205 static int lov_io_subio_init(const struct lu_env *env, struct lov_io *lio,
206                              struct cl_io *io)
207 {
208         ENTRY;
209
210         LASSERT(lio->lis_object != NULL);
211
212         INIT_LIST_HEAD(&lio->lis_subios);
213         lio->lis_single_subio_index = -1;
214         lio->lis_nr_subios = 0;
215
216         RETURN(0);
217 }
218
219 /**
220  * Decide if it will need write intent RPC
221  */
222 static int lov_io_mirror_write_intent(struct lov_io *lio,
223         struct lov_object *obj, struct cl_io *io)
224 {
225         struct lov_layout_composite *comp = &obj->u.composite;
226         struct lu_extent *ext = &io->ci_write_intent;
227         struct lov_mirror_entry *lre;
228         struct lov_mirror_entry *primary;
229         struct lov_layout_entry *lle;
230         size_t count = 0;
231         ENTRY;
232
233         *ext = (typeof(*ext)) { lio->lis_pos, lio->lis_endpos };
234         io->ci_need_write_intent = 0;
235
236         if (!(io->ci_type == CIT_WRITE || cl_io_is_trunc(io) ||
237               cl_io_is_mkwrite(io)))
238                 RETURN(0);
239
240         /* FLR: check if it needs to send a write intent RPC to server.
241          * Writing to sync_pending file needs write intent RPC to change
242          * the file state back to write_pending, so that the layout version
243          * can be increased when the state changes to sync_pending at a later
244          * time. Otherwise there exists a chance that an evicted client may
245          * dirty the file data while resync client is working on it.
246          * Designated I/O is allowed for resync workload.
247          */
248         if (lov_flr_state(obj) == LCM_FL_RDONLY ||
249             (lov_flr_state(obj) == LCM_FL_SYNC_PENDING &&
250              io->ci_designated_mirror == 0)) {
251                 io->ci_need_write_intent = 1;
252                 RETURN(0);
253         }
254
255         LASSERT((lov_flr_state(obj) == LCM_FL_WRITE_PENDING));
256         LASSERT(comp->lo_preferred_mirror >= 0);
257
258         /* need to iterate all components to see if there are
259          * multiple components covering the writing component */
260         primary = &comp->lo_mirrors[comp->lo_preferred_mirror];
261         LASSERT(!primary->lre_stale);
262         lov_foreach_mirror_layout_entry(obj, lle, primary) {
263                 LASSERT(lle->lle_valid);
264                 if (!lu_extent_is_overlapped(ext, lle->lle_extent))
265                         continue;
266
267                 ext->e_start = MIN(ext->e_start, lle->lle_extent->e_start);
268                 ext->e_end = MAX(ext->e_end, lle->lle_extent->e_end);
269                 ++count;
270         }
271         if (count == 0) {
272                 CERROR(DFID ": cannot find any valid components covering "
273                        "file extent "DEXT", mirror: %d\n",
274                        PFID(lu_object_fid(lov2lu(obj))), PEXT(ext),
275                        primary->lre_mirror_id);
276                 RETURN(-EIO);
277         }
278
279         count = 0;
280         lov_foreach_mirror_entry(obj, lre) {
281                 if (lre == primary)
282                         continue;
283
284                 lov_foreach_mirror_layout_entry(obj, lle, lre) {
285                         if (!lle->lle_valid)
286                                 continue;
287
288                         if (lu_extent_is_overlapped(ext, lle->lle_extent)) {
289                                 ++count;
290                                 break;
291                         }
292                 }
293         }
294
295         CDEBUG(D_VFSTRACE, DFID "there are %zd components to be staled to "
296                "modify file extent "DEXT", iot: %d\n",
297                PFID(lu_object_fid(lov2lu(obj))), count, PEXT(ext), io->ci_type);
298
299         io->ci_need_write_intent = count > 0;
300
301         RETURN(0);
302 }
303
304 static int lov_io_mirror_init(struct lov_io *lio, struct lov_object *obj,
305                                struct cl_io *io)
306 {
307         struct lov_layout_composite *comp = &obj->u.composite;
308         int index;
309         int i;
310         int result;
311         ENTRY;
312
313         if (!lov_is_flr(obj)) {
314                 LASSERT(comp->lo_preferred_mirror == 0);
315                 lio->lis_mirror_index = comp->lo_preferred_mirror;
316                 io->ci_ndelay = 0;
317                 RETURN(0);
318         }
319
320         /* transfer the layout version for verification */
321         if (io->ci_layout_version == 0)
322                 io->ci_layout_version = obj->lo_lsm->lsm_layout_gen;
323
324         /* find the corresponding mirror for designated mirror IO */
325         if (io->ci_designated_mirror > 0) {
326                 struct lov_mirror_entry *entry;
327
328                 LASSERT(!io->ci_ndelay);
329
330                 CDEBUG(D_LAYOUT, "designated I/O mirror state: %d\n",
331                       lov_flr_state(obj));
332
333                 if ((cl_io_is_trunc(io) || io->ci_type == CIT_WRITE) &&
334                     (io->ci_layout_version != obj->lo_lsm->lsm_layout_gen)) {
335                         /* For resync I/O, the ci_layout_version was the layout
336                          * version when resync starts. If it doesn't match the
337                          * current object layout version, it means the layout
338                          * has been changed */
339                         RETURN(-ESTALE);
340                 }
341
342                 io->ci_layout_version |= LU_LAYOUT_RESYNC;
343
344                 index = 0;
345                 lio->lis_mirror_index = -1;
346                 lov_foreach_mirror_entry(obj, entry) {
347                         if (entry->lre_mirror_id ==
348                             io->ci_designated_mirror) {
349                                 lio->lis_mirror_index = index;
350                                 break;
351                         }
352
353                         index++;
354                 }
355
356                 RETURN(lio->lis_mirror_index < 0 ? -EINVAL : 0);
357         }
358
359         result = lov_io_mirror_write_intent(lio, obj, io);
360         if (result)
361                 RETURN(result);
362
363         if (io->ci_need_write_intent) {
364                 CDEBUG(D_VFSTRACE, DFID " need write intent for [%llu, %llu)\n",
365                        PFID(lu_object_fid(lov2lu(obj))),
366                        lio->lis_pos, lio->lis_endpos);
367
368                 if (cl_io_is_trunc(io)) {
369                         /**
370                          * for truncate, we uses [size, EOF) to judge whether
371                          * a write intent needs to be send, but we need to
372                          * restore the write extent to [0, size).
373                          */
374                         io->ci_write_intent.e_start = 0;
375                         io->ci_write_intent.e_end =
376                                         io->u.ci_setattr.sa_attr.lvb_size;
377                 }
378                 /* stop cl_io_init() loop */
379                 RETURN(1);
380         }
381
382         if (io->ci_ndelay_tried == 0 || /* first time to try */
383             /* reset the mirror index if layout has changed */
384             lio->lis_mirror_layout_gen != obj->lo_lsm->lsm_layout_gen) {
385                 lio->lis_mirror_layout_gen = obj->lo_lsm->lsm_layout_gen;
386                 index = lio->lis_mirror_index = comp->lo_preferred_mirror;
387         } else {
388                 index = lio->lis_mirror_index;
389                 LASSERT(index >= 0);
390
391                 /* move mirror index to the next one */
392                 index = (index + 1) % comp->lo_mirror_count;
393         }
394
395         for (i = 0; i < comp->lo_mirror_count; i++) {
396                 struct lu_extent ext = { .e_start = lio->lis_pos,
397                                          .e_end   = lio->lis_pos + 1 };
398                 struct lov_mirror_entry *lre;
399                 struct lov_layout_entry *lle;
400                 bool found = false;
401
402                 lre = &comp->lo_mirrors[(index + i) % comp->lo_mirror_count];
403                 if (!lre->lre_valid)
404                         continue;
405
406                 lov_foreach_mirror_layout_entry(obj, lle, lre) {
407                         if (!lle->lle_valid)
408                                 continue;
409
410                         if (lu_extent_is_overlapped(&ext, lle->lle_extent)) {
411                                 found = true;
412                                 break;
413                         }
414                 }
415
416                 if (found) {
417                         index = (index + i) % comp->lo_mirror_count;
418                         break;
419                 }
420         }
421         if (i == comp->lo_mirror_count) {
422                 CERROR(DFID": failed to find a component covering "
423                        "I/O region at %llu\n",
424                        PFID(lu_object_fid(lov2lu(obj))), lio->lis_pos);
425
426                 dump_lsm(D_ERROR, obj->lo_lsm);
427
428                 RETURN(-EIO);
429         }
430
431         CDEBUG(D_VFSTRACE, DFID ": flr state: %d, move mirror from %d to %d, "
432                "have retried: %d, mirror count: %d\n",
433                PFID(lu_object_fid(lov2lu(obj))), lov_flr_state(obj),
434                lio->lis_mirror_index, index, io->ci_ndelay_tried,
435                comp->lo_mirror_count);
436
437         lio->lis_mirror_index = index;
438
439         /* FLR: if all mirrors have been tried once, most likely the network
440          * of this client has been partitioned. We should relinquish CPU for
441          * a while before trying again. */
442         ++io->ci_ndelay_tried;
443         if (io->ci_ndelay && io->ci_ndelay_tried >= comp->lo_mirror_count) {
444                 set_current_state(TASK_INTERRUPTIBLE);
445                 schedule_timeout(msecs_to_jiffies(MSEC_PER_SEC)); /* 10ms */
446                 if (signal_pending(current))
447                         RETURN(-EINTR);
448
449                 /* reset retry counter */
450                 io->ci_ndelay_tried = 1;
451         }
452
453         CDEBUG(D_VFSTRACE, "use %sdelayed RPC state for this IO\n",
454                io->ci_ndelay ? "non-" : "");
455
456         RETURN(0);
457 }
458
459 static int lov_io_slice_init(struct lov_io *lio,
460                              struct lov_object *obj, struct cl_io *io)
461 {
462         int index;
463         int result = 0;
464         ENTRY;
465
466         io->ci_result = 0;
467         lio->lis_object = obj;
468
469         LASSERT(obj->lo_lsm != NULL);
470
471         switch (io->ci_type) {
472         case CIT_READ:
473         case CIT_WRITE:
474                 lio->lis_pos = io->u.ci_rw.rw_range.cir_pos;
475                 lio->lis_endpos = lio->lis_pos + io->u.ci_rw.rw_range.cir_count;
476                 lio->lis_io_endpos = lio->lis_endpos;
477                 if (cl_io_is_append(io)) {
478                         LASSERT(io->ci_type == CIT_WRITE);
479
480                         /* If there is LOV EA hole, then we may cannot locate
481                          * the current file-tail exactly. */
482                         if (unlikely(obj->lo_lsm->lsm_entries[0]->lsme_pattern &
483                                      LOV_PATTERN_F_HOLE))
484                                 GOTO(out, result = -EIO);
485
486                         lio->lis_pos = 0;
487                         lio->lis_endpos = OBD_OBJECT_EOF;
488                 }
489                 break;
490
491         case CIT_SETATTR:
492                 if (cl_io_is_trunc(io))
493                         lio->lis_pos = io->u.ci_setattr.sa_attr.lvb_size;
494                 else
495                         lio->lis_pos = 0;
496                 lio->lis_endpos = OBD_OBJECT_EOF;
497                 break;
498
499         case CIT_DATA_VERSION:
500                 lio->lis_pos = 0;
501                 lio->lis_endpos = OBD_OBJECT_EOF;
502                 break;
503
504         case CIT_FAULT: {
505                 pgoff_t index = io->u.ci_fault.ft_index;
506                 lio->lis_pos = cl_offset(io->ci_obj, index);
507                 lio->lis_endpos = cl_offset(io->ci_obj, index + 1);
508                 break;
509         }
510
511         case CIT_FSYNC: {
512                 lio->lis_pos = io->u.ci_fsync.fi_start;
513                 lio->lis_endpos = io->u.ci_fsync.fi_end;
514                 break;
515         }
516
517         case CIT_LADVISE: {
518                 lio->lis_pos = io->u.ci_ladvise.li_start;
519                 lio->lis_endpos = io->u.ci_ladvise.li_end;
520                 break;
521         }
522
523         case CIT_GLIMPSE:
524                 lio->lis_pos = 0;
525                 lio->lis_endpos = OBD_OBJECT_EOF;
526
527                 if (lov_flr_state(obj) == LCM_FL_RDONLY &&
528                     !OBD_FAIL_CHECK(OBD_FAIL_FLR_GLIMPSE_IMMUTABLE))
529                         /* SoM is accurate, no need glimpse */
530                         GOTO(out, result = 1);
531                 break;
532
533         case CIT_MISC:
534                 lio->lis_pos = 0;
535                 lio->lis_endpos = OBD_OBJECT_EOF;
536                 break;
537
538         default:
539                 LBUG();
540         }
541
542         result = lov_io_mirror_init(lio, obj, io);
543         if (result)
544                 GOTO(out, result);
545
546         /* check if it needs to instantiate layout */
547         if (!(io->ci_type == CIT_WRITE || cl_io_is_mkwrite(io) ||
548               (cl_io_is_trunc(io) && io->u.ci_setattr.sa_attr.lvb_size > 0)))
549                 GOTO(out, result = 0);
550
551         /* for truncate, it only needs to instantiate the components
552          * before the truncated size. */
553         if (cl_io_is_trunc(io)) {
554                 io->ci_write_intent.e_start = 0;
555                 io->ci_write_intent.e_end = io->u.ci_setattr.sa_attr.lvb_size;
556         } else {
557                 io->ci_write_intent.e_start = lio->lis_pos;
558                 io->ci_write_intent.e_end = lio->lis_endpos;
559         }
560
561         index = 0;
562         lov_foreach_io_layout(index, lio, &io->ci_write_intent) {
563                 if (!lsm_entry_inited(obj->lo_lsm, index)) {
564                         io->ci_need_write_intent = 1;
565                         break;
566                 }
567         }
568
569         if (io->ci_need_write_intent && io->ci_designated_mirror > 0) {
570                 /* REINT_SYNC RPC has already tried to instantiate all of the
571                  * components involved, obviously it didn't succeed. Skip this
572                  * mirror for now. The server won't be able to figure out
573                  * which mirror it should instantiate components */
574                 CERROR(DFID": trying to instantiate components for designated "
575                        "I/O, file state: %d\n",
576                        PFID(lu_object_fid(lov2lu(obj))), lov_flr_state(obj));
577
578                 io->ci_need_write_intent = 0;
579                 GOTO(out, result = -EIO);
580         }
581
582         if (io->ci_need_write_intent)
583                 GOTO(out, result = 1);
584
585         EXIT;
586
587 out:
588         return result;
589 }
590
591 static void lov_io_fini(const struct lu_env *env, const struct cl_io_slice *ios)
592 {
593         struct lov_io *lio = cl2lov_io(env, ios);
594         struct lov_object *lov = cl2lov(ios->cis_obj);
595
596         ENTRY;
597
598         LASSERT(list_empty(&lio->lis_active));
599
600         while (!list_empty(&lio->lis_subios)) {
601                 struct lov_io_sub *sub = list_entry(lio->lis_subios.next,
602                                                     struct lov_io_sub,
603                                                     sub_list);
604
605                 list_del_init(&sub->sub_list);
606                 lio->lis_nr_subios--;
607
608                 lov_io_sub_fini(env, lio, sub);
609                 lov_sub_free(lio, sub);
610         }
611         LASSERT(lio->lis_nr_subios == 0);
612
613         LASSERT(atomic_read(&lov->lo_active_ios) > 0);
614         if (atomic_dec_and_test(&lov->lo_active_ios))
615                 wake_up_all(&lov->lo_waitq);
616         EXIT;
617 }
618
619 static void lov_io_sub_inherit(struct lov_io_sub *sub, struct lov_io *lio,
620                                loff_t start, loff_t end)
621 {
622         struct cl_io *io = &sub->sub_io;
623         struct lov_stripe_md *lsm = lio->lis_object->lo_lsm;
624         struct cl_io *parent = lio->lis_cl.cis_io;
625         int index = lov_comp_entry(sub->sub_subio_index);
626         int stripe = lov_comp_stripe(sub->sub_subio_index);
627
628         io->ci_pio = parent->ci_pio;
629         switch (io->ci_type) {
630         case CIT_SETATTR: {
631                 io->u.ci_setattr.sa_attr = parent->u.ci_setattr.sa_attr;
632                 io->u.ci_setattr.sa_attr_flags =
633                         parent->u.ci_setattr.sa_attr_flags;
634                 io->u.ci_setattr.sa_valid = parent->u.ci_setattr.sa_valid;
635                 io->u.ci_setattr.sa_stripe_index = stripe;
636                 io->u.ci_setattr.sa_parent_fid =
637                                         parent->u.ci_setattr.sa_parent_fid;
638                 if (cl_io_is_trunc(io)) {
639                         loff_t new_size = parent->u.ci_setattr.sa_attr.lvb_size;
640
641                         new_size = lov_size_to_stripe(lsm, index, new_size,
642                                                       stripe);
643                         io->u.ci_setattr.sa_attr.lvb_size = new_size;
644                 }
645                 lov_lsm2layout(lsm, lsm->lsm_entries[index],
646                                &io->u.ci_setattr.sa_layout);
647                 break;
648         }
649         case CIT_DATA_VERSION: {
650                 io->u.ci_data_version.dv_data_version = 0;
651                 io->u.ci_data_version.dv_flags =
652                         parent->u.ci_data_version.dv_flags;
653                 break;
654         }
655         case CIT_FAULT: {
656                 struct cl_object *obj = parent->ci_obj;
657                 loff_t off = cl_offset(obj, parent->u.ci_fault.ft_index);
658
659                 io->u.ci_fault = parent->u.ci_fault;
660                 off = lov_size_to_stripe(lsm, index, off, stripe);
661                 io->u.ci_fault.ft_index = cl_index(obj, off);
662                 break;
663         }
664         case CIT_FSYNC: {
665                 io->u.ci_fsync.fi_start = start;
666                 io->u.ci_fsync.fi_end = end;
667                 io->u.ci_fsync.fi_fid = parent->u.ci_fsync.fi_fid;
668                 io->u.ci_fsync.fi_mode = parent->u.ci_fsync.fi_mode;
669                 break;
670         }
671         case CIT_READ:
672         case CIT_WRITE: {
673                 io->u.ci_rw.rw_ptask = parent->u.ci_rw.rw_ptask;
674                 io->u.ci_rw.rw_iter = parent->u.ci_rw.rw_iter;
675                 io->u.ci_rw.rw_iocb = parent->u.ci_rw.rw_iocb;
676                 io->u.ci_rw.rw_file = parent->u.ci_rw.rw_file;
677                 io->u.ci_rw.rw_sync = parent->u.ci_rw.rw_sync;
678                 if (cl_io_is_append(parent)) {
679                         io->u.ci_rw.rw_append = 1;
680                 } else {
681                         io->u.ci_rw.rw_range.cir_pos = start;
682                         io->u.ci_rw.rw_range.cir_count = end - start;
683                 }
684                 break;
685         }
686         case CIT_LADVISE: {
687                 io->u.ci_ladvise.li_start = start;
688                 io->u.ci_ladvise.li_end = end;
689                 io->u.ci_ladvise.li_fid = parent->u.ci_ladvise.li_fid;
690                 io->u.ci_ladvise.li_advice = parent->u.ci_ladvise.li_advice;
691                 io->u.ci_ladvise.li_flags = parent->u.ci_ladvise.li_flags;
692                 break;
693         }
694         case CIT_GLIMPSE:
695         case CIT_MISC:
696         default:
697                 break;
698         }
699 }
700
701 static loff_t lov_offset_mod(loff_t val, int delta)
702 {
703         if (val != OBD_OBJECT_EOF)
704                 val += delta;
705         return val;
706 }
707
708 static int lov_io_iter_init(const struct lu_env *env,
709                             const struct cl_io_slice *ios)
710 {
711         struct lov_io        *lio = cl2lov_io(env, ios);
712         struct lov_stripe_md *lsm = lio->lis_object->lo_lsm;
713         struct lov_io_sub    *sub;
714         struct lu_extent ext;
715         int index;
716         int rc = 0;
717
718         ENTRY;
719
720         ext.e_start = lio->lis_pos;
721         ext.e_end = lio->lis_endpos;
722
723         lov_foreach_io_layout(index, lio, &ext) {
724                 struct lov_layout_entry *le = lov_entry(lio->lis_object, index);
725                 struct lov_layout_raid0 *r0 = &le->lle_raid0;
726                 u64 start;
727                 u64 end;
728                 int stripe;
729
730                 CDEBUG(D_VFSTRACE, "component[%d] flags %#x\n",
731                        index, lsm->lsm_entries[index]->lsme_flags);
732                 if (!lsm_entry_inited(lsm, index)) {
733                         /* Read from uninitialized components should return
734                          * zero filled pages. */
735                         continue;
736                 }
737
738                 if (!le->lle_valid && !ios->cis_io->ci_designated_mirror) {
739                         CERROR("I/O to invalid component: %d, mirror: %d\n",
740                                index, lio->lis_mirror_index);
741                         RETURN(-EIO);
742                 }
743
744                 for (stripe = 0; stripe < r0->lo_nr; stripe++) {
745                         if (!lov_stripe_intersects(lsm, index, stripe,
746                                                    &ext, &start, &end))
747                                 continue;
748
749                         if (unlikely(r0->lo_sub[stripe] == NULL)) {
750                                 if (ios->cis_io->ci_type == CIT_READ ||
751                                     ios->cis_io->ci_type == CIT_WRITE ||
752                                     ios->cis_io->ci_type == CIT_FAULT)
753                                         RETURN(-EIO);
754
755                                 continue;
756                         }
757
758                         end = lov_offset_mod(end, 1);
759                         sub = lov_sub_get(env, lio,
760                                           lov_comp_index(index, stripe));
761                         if (IS_ERR(sub)) {
762                                 rc = PTR_ERR(sub);
763                                 break;
764                         }
765
766                         lov_io_sub_inherit(sub, lio, start, end);
767                         rc = cl_io_iter_init(sub->sub_env, &sub->sub_io);
768                         if (rc != 0)
769                                 cl_io_iter_fini(sub->sub_env, &sub->sub_io);
770                         if (rc != 0)
771                                 break;
772
773                         CDEBUG(D_VFSTRACE,
774                                 "shrink stripe: {%d, %d} range: [%llu, %llu)\n",
775                                 index, stripe, start, end);
776
777                         list_add_tail(&sub->sub_linkage, &lio->lis_active);
778                 }
779                 if (rc != 0)
780                         break;
781         }
782         RETURN(rc);
783 }
784
785 static int lov_io_rw_iter_init(const struct lu_env *env,
786                                const struct cl_io_slice *ios)
787 {
788         struct cl_io *io = ios->cis_io;
789         struct lov_io *lio = cl2lov_io(env, ios);
790         struct lov_stripe_md_entry *lse;
791         struct cl_io_range *range = &io->u.ci_rw.rw_range;
792         loff_t start = range->cir_pos;
793         loff_t next;
794         int index;
795
796         LASSERT(io->ci_type == CIT_READ || io->ci_type == CIT_WRITE);
797         ENTRY;
798
799         if (cl_io_is_append(io))
800                 RETURN(lov_io_iter_init(env, ios));
801
802         index = lov_io_layout_at(lio, range->cir_pos);
803         if (index < 0) { /* non-existing layout component */
804                 if (io->ci_type == CIT_READ) {
805                         /* TODO: it needs to detect the next component and
806                          * then set the next pos */
807                         io->ci_continue = 0;
808                         /* execute it in main thread */
809                         io->ci_pio = 0;
810
811                         RETURN(lov_io_iter_init(env, ios));
812                 }
813
814                 RETURN(-ENODATA);
815         }
816
817         if (!lov_entry(lio->lis_object, index)->lle_valid &&
818             !io->ci_designated_mirror)
819                 RETURN(io->ci_type == CIT_READ ? -EAGAIN : -EIO);
820
821         lse = lov_lse(lio->lis_object, index);
822
823         next = MAX_LFS_FILESIZE;
824         if (lse->lsme_stripe_count > 1) {
825                 unsigned long ssize = lse->lsme_stripe_size;
826
827                 lov_do_div64(start, ssize);
828                 next = (start + 1) * ssize;
829                 if (next <= start * ssize)
830                         next = MAX_LFS_FILESIZE;
831         }
832
833         LASSERTF(range->cir_pos >= lse->lsme_extent.e_start,
834                  "pos %lld, [%lld, %lld)\n", range->cir_pos,
835                  lse->lsme_extent.e_start, lse->lsme_extent.e_end);
836         next = min_t(__u64, next, lse->lsme_extent.e_end);
837         next = min_t(loff_t, next, lio->lis_io_endpos);
838
839         io->ci_continue  = next < lio->lis_io_endpos;
840         range->cir_count = next - range->cir_pos;
841         lio->lis_pos     = range->cir_pos;
842         lio->lis_endpos  = range->cir_pos + range->cir_count;
843         CDEBUG(D_VFSTRACE,
844                "stripe: {%d, %llu} range: [%llu, %llu) end: %llu, count: %zd\n",
845                index, start, lio->lis_pos, lio->lis_endpos,
846                lio->lis_io_endpos, range->cir_count);
847
848         if (!io->ci_continue) {
849                 /* the last piece of IO, execute it in main thread */
850                 io->ci_pio = 0;
851         }
852
853         if (io->ci_pio)
854                 RETURN(0);
855
856         /*
857          * XXX The following call should be optimized: we know, that
858          * [lio->lis_pos, lio->lis_endpos) intersects with exactly one stripe.
859          */
860         RETURN(lov_io_iter_init(env, ios));
861 }
862
863 static int lov_io_setattr_iter_init(const struct lu_env *env,
864                                     const struct cl_io_slice *ios)
865 {
866         struct lov_io *lio = cl2lov_io(env, ios);
867         struct cl_io *io = ios->cis_io;
868         int index;
869         ENTRY;
870
871         if (cl_io_is_trunc(io) && lio->lis_pos > 0) {
872                 index = lov_io_layout_at(lio, lio->lis_pos - 1);
873                 /* no entry found for such offset */
874                 if (index < 0)
875                         RETURN(io->ci_result = -ENODATA);
876         }
877
878         RETURN(lov_io_iter_init(env, ios));
879 }
880
881 static int lov_io_call(const struct lu_env *env, struct lov_io *lio,
882                        int (*iofunc)(const struct lu_env *, struct cl_io *))
883 {
884         struct cl_io *parent = lio->lis_cl.cis_io;
885         struct lov_io_sub *sub;
886         int rc = 0;
887
888         ENTRY;
889         list_for_each_entry(sub, &lio->lis_active, sub_linkage) {
890                 rc = iofunc(sub->sub_env, &sub->sub_io);
891                 if (rc)
892                         break;
893
894                 if (parent->ci_result == 0)
895                         parent->ci_result = sub->sub_io.ci_result;
896         }
897         RETURN(rc);
898 }
899
900 static int lov_io_lock(const struct lu_env *env, const struct cl_io_slice *ios)
901 {
902         ENTRY;
903         RETURN(lov_io_call(env, cl2lov_io(env, ios), cl_io_lock));
904 }
905
906 static int lov_io_start(const struct lu_env *env, const struct cl_io_slice *ios)
907 {
908         ENTRY;
909         RETURN(lov_io_call(env, cl2lov_io(env, ios), cl_io_start));
910 }
911
912 static int lov_io_end_wrapper(const struct lu_env *env, struct cl_io *io)
913 {
914         ENTRY;
915         /*
916          * It's possible that lov_io_start() wasn't called against this
917          * sub-io, either because previous sub-io failed, or upper layer
918          * completed IO.
919          */
920         if (io->ci_state == CIS_IO_GOING)
921                 cl_io_end(env, io);
922         else
923                 io->ci_state = CIS_IO_FINISHED;
924         RETURN(0);
925 }
926
927 static int lov_io_iter_fini_wrapper(const struct lu_env *env, struct cl_io *io)
928 {
929         cl_io_iter_fini(env, io);
930         RETURN(0);
931 }
932
933 static int lov_io_unlock_wrapper(const struct lu_env *env, struct cl_io *io)
934 {
935         cl_io_unlock(env, io);
936         RETURN(0);
937 }
938
939 static void lov_io_end(const struct lu_env *env, const struct cl_io_slice *ios)
940 {
941         int rc;
942
943         rc = lov_io_call(env, cl2lov_io(env, ios), lov_io_end_wrapper);
944         LASSERT(rc == 0);
945 }
946
947 static void
948 lov_io_data_version_end(const struct lu_env *env, const struct cl_io_slice *ios)
949 {
950         struct lov_io *lio = cl2lov_io(env, ios);
951         struct cl_io *parent = lio->lis_cl.cis_io;
952         struct cl_data_version_io *pdv = &parent->u.ci_data_version;
953         struct lov_io_sub *sub;
954
955         ENTRY;
956         list_for_each_entry(sub, &lio->lis_active, sub_linkage) {
957                 struct cl_data_version_io *sdv = &sub->sub_io.u.ci_data_version;
958
959                 lov_io_end_wrapper(sub->sub_env, &sub->sub_io);
960
961                 pdv->dv_data_version += sdv->dv_data_version;
962                 if (pdv->dv_layout_version > sdv->dv_layout_version)
963                         pdv->dv_layout_version = sdv->dv_layout_version;
964
965                 if (parent->ci_result == 0)
966                         parent->ci_result = sub->sub_io.ci_result;
967         }
968
969         EXIT;
970 }
971
972 static void lov_io_iter_fini(const struct lu_env *env,
973                              const struct cl_io_slice *ios)
974 {
975         struct lov_io *lio = cl2lov_io(env, ios);
976         int rc;
977
978         ENTRY;
979         rc = lov_io_call(env, lio, lov_io_iter_fini_wrapper);
980         LASSERT(rc == 0);
981         while (!list_empty(&lio->lis_active))
982                 list_del_init(lio->lis_active.next);
983         EXIT;
984 }
985
986 static void lov_io_unlock(const struct lu_env *env,
987                           const struct cl_io_slice *ios)
988 {
989         int rc;
990
991         ENTRY;
992         rc = lov_io_call(env, cl2lov_io(env, ios), lov_io_unlock_wrapper);
993         LASSERT(rc == 0);
994         EXIT;
995 }
996
997 static int lov_io_read_ahead(const struct lu_env *env,
998                              const struct cl_io_slice *ios,
999                              pgoff_t start, struct cl_read_ahead *ra)
1000 {
1001         struct lov_io           *lio = cl2lov_io(env, ios);
1002         struct lov_object       *loo = lio->lis_object;
1003         struct cl_object        *obj = lov2cl(loo);
1004         struct lov_layout_raid0 *r0;
1005         struct lov_io_sub       *sub;
1006         loff_t                   offset;
1007         loff_t                   suboff;
1008         pgoff_t                  ra_end;
1009         unsigned int             pps; /* pages per stripe */
1010         int                      stripe;
1011         int                      index;
1012         int                      rc;
1013         ENTRY;
1014
1015         offset = cl_offset(obj, start);
1016         index = lov_io_layout_at(lio, offset);
1017         if (index < 0 || !lsm_entry_inited(loo->lo_lsm, index))
1018                 RETURN(-ENODATA);
1019
1020         /* avoid readahead to expand to stale components */
1021         if (!lov_entry(loo, index)->lle_valid)
1022                 RETURN(-EIO);
1023
1024         stripe = lov_stripe_number(loo->lo_lsm, index, offset);
1025
1026         r0 = lov_r0(loo, index);
1027         if (unlikely(r0->lo_sub[stripe] == NULL))
1028                 RETURN(-EIO);
1029
1030         sub = lov_sub_get(env, lio, lov_comp_index(index, stripe));
1031         if (IS_ERR(sub))
1032                 RETURN(PTR_ERR(sub));
1033
1034         lov_stripe_offset(loo->lo_lsm, index, offset, stripe, &suboff);
1035         rc = cl_io_read_ahead(sub->sub_env, &sub->sub_io,
1036                               cl_index(lovsub2cl(r0->lo_sub[stripe]), suboff),
1037                               ra);
1038
1039         CDEBUG(D_READA, DFID " cra_end = %lu, stripes = %d, rc = %d\n",
1040                PFID(lu_object_fid(lov2lu(loo))), ra->cra_end, r0->lo_nr, rc);
1041         if (rc != 0)
1042                 RETURN(rc);
1043
1044         /**
1045          * Adjust the stripe index by layout of comp. ra->cra_end is the
1046          * maximum page index covered by an underlying DLM lock.
1047          * This function converts cra_end from stripe level to file level, and
1048          * make sure it's not beyond stripe and component boundary.
1049          */
1050
1051         /* cra_end is stripe level, convert it into file level */
1052         ra_end = ra->cra_end;
1053         if (ra_end != CL_PAGE_EOF)
1054                 ra->cra_end = lov_stripe_pgoff(loo->lo_lsm, index,
1055                                                ra_end, stripe);
1056
1057         /* boundary of current component */
1058         ra_end = cl_index(obj, (loff_t)lov_io_extent(lio, index)->e_end);
1059         if (ra_end != CL_PAGE_EOF && ra->cra_end >= ra_end)
1060                 ra->cra_end = ra_end - 1;
1061
1062         if (r0->lo_nr == 1) /* single stripe file */
1063                 RETURN(0);
1064
1065         pps = lov_lse(loo, index)->lsme_stripe_size >> PAGE_SHIFT;
1066
1067         CDEBUG(D_READA, DFID " max_index = %lu, pps = %u, index = %u, "
1068                "stripe_size = %u, stripe no = %u, start index = %lu\n",
1069                PFID(lu_object_fid(lov2lu(loo))), ra->cra_end, pps, index,
1070                lov_lse(loo, index)->lsme_stripe_size, stripe, start);
1071
1072         /* never exceed the end of the stripe */
1073         ra->cra_end = min_t(pgoff_t,
1074                             ra->cra_end, start + pps - start % pps - 1);
1075         RETURN(0);
1076 }
1077
1078 /**
1079  * lov implementation of cl_operations::cio_submit() method. It takes a list
1080  * of pages in \a queue, splits it into per-stripe sub-lists, invokes
1081  * cl_io_submit() on underlying devices to submit sub-lists, and then splices
1082  * everything back.
1083  *
1084  * Major complication of this function is a need to handle memory cleansing:
1085  * cl_io_submit() is called to write out pages as a part of VM memory
1086  * reclamation, and hence it may not fail due to memory shortages (system
1087  * dead-locks otherwise). To deal with this, some resources (sub-lists,
1088  * sub-environment, etc.) are allocated per-device on "startup" (i.e., in a
1089  * not-memory cleansing context), and in case of memory shortage, these
1090  * pre-allocated resources are used by lov_io_submit() under
1091  * lov_device::ld_mutex mutex.
1092  */
1093 static int lov_io_submit(const struct lu_env *env,
1094                          const struct cl_io_slice *ios,
1095                          enum cl_req_type crt, struct cl_2queue *queue)
1096 {
1097         struct cl_page_list     *qin = &queue->c2_qin;
1098         struct lov_io           *lio = cl2lov_io(env, ios);
1099         struct lov_io_sub       *sub;
1100         struct cl_page_list     *plist = &lov_env_info(env)->lti_plist;
1101         struct cl_page          *page;
1102         int index;
1103         int rc = 0;
1104         ENTRY;
1105
1106         cl_page_list_init(plist);
1107         while (qin->pl_nr > 0) {
1108                 struct cl_2queue  *cl2q = &lov_env_info(env)->lti_cl2q;
1109
1110                 page = cl_page_list_first(qin);
1111                 if (lov_page_is_empty(page)) {
1112                         cl_page_list_move(&queue->c2_qout, qin, page);
1113
1114                         /* it could only be mirror read to get here therefore
1115                          * the pages will be transient. We don't care about
1116                          * the return code of cl_page_prep() at all. */
1117                         (void) cl_page_prep(env, ios->cis_io, page, crt);
1118                         cl_page_completion(env, page, crt, 0);
1119                         continue;
1120                 }
1121
1122                 cl_2queue_init(cl2q);
1123                 cl_page_list_move(&cl2q->c2_qin, qin, page);
1124
1125                 index = lov_page_index(page);
1126                 while (qin->pl_nr > 0) {
1127                         page = cl_page_list_first(qin);
1128                         if (index != lov_page_index(page))
1129                                 break;
1130
1131                         cl_page_list_move(&cl2q->c2_qin, qin, page);
1132                 }
1133
1134                 sub = lov_sub_get(env, lio, index);
1135                 if (!IS_ERR(sub)) {
1136                         rc = cl_io_submit_rw(sub->sub_env, &sub->sub_io,
1137                                              crt, cl2q);
1138                 } else {
1139                         rc = PTR_ERR(sub);
1140                 }
1141
1142                 cl_page_list_splice(&cl2q->c2_qin, plist);
1143                 cl_page_list_splice(&cl2q->c2_qout, &queue->c2_qout);
1144                 cl_2queue_fini(env, cl2q);
1145
1146                 if (rc != 0)
1147                         break;
1148         }
1149
1150         cl_page_list_splice(plist, qin);
1151         cl_page_list_fini(env, plist);
1152
1153         RETURN(rc);
1154 }
1155
1156 static int lov_io_commit_async(const struct lu_env *env,
1157                                const struct cl_io_slice *ios,
1158                                struct cl_page_list *queue, int from, int to,
1159                                cl_commit_cbt cb)
1160 {
1161         struct cl_page_list *plist = &lov_env_info(env)->lti_plist;
1162         struct lov_io     *lio = cl2lov_io(env, ios);
1163         struct lov_io_sub *sub;
1164         struct cl_page *page;
1165         int rc = 0;
1166         ENTRY;
1167
1168         if (lio->lis_nr_subios == 1) {
1169                 int idx = lio->lis_single_subio_index;
1170
1171                 LASSERT(!lov_page_is_empty(cl_page_list_first(queue)));
1172
1173                 sub = lov_sub_get(env, lio, idx);
1174                 LASSERT(!IS_ERR(sub));
1175                 LASSERT(sub == &lio->lis_single_subio);
1176                 rc = cl_io_commit_async(sub->sub_env, &sub->sub_io, queue,
1177                                         from, to, cb);
1178                 RETURN(rc);
1179         }
1180
1181         cl_page_list_init(plist);
1182         while (queue->pl_nr > 0) {
1183                 int stripe_to = to;
1184                 int index;
1185
1186                 LASSERT(plist->pl_nr == 0);
1187                 page = cl_page_list_first(queue);
1188                 LASSERT(!lov_page_is_empty(page));
1189
1190                 cl_page_list_move(plist, queue, page);
1191
1192                 index = lov_page_index(page);
1193                 while (queue->pl_nr > 0) {
1194                         page = cl_page_list_first(queue);
1195                         if (index != lov_page_index(page))
1196                                 break;
1197
1198                         cl_page_list_move(plist, queue, page);
1199                 }
1200
1201                 if (queue->pl_nr > 0) /* still has more pages */
1202                         stripe_to = PAGE_SIZE;
1203
1204                 sub = lov_sub_get(env, lio, index);
1205                 if (!IS_ERR(sub)) {
1206                         rc = cl_io_commit_async(sub->sub_env, &sub->sub_io,
1207                                                 plist, from, stripe_to, cb);
1208                 } else {
1209                         rc = PTR_ERR(sub);
1210                         break;
1211                 }
1212
1213                 if (plist->pl_nr > 0) /* short write */
1214                         break;
1215
1216                 from = 0;
1217         }
1218
1219         /* for error case, add the page back into the qin list */
1220         LASSERT(ergo(rc == 0, plist->pl_nr == 0));
1221         while (plist->pl_nr > 0) {
1222                 /* error occurred, add the uncommitted pages back into queue */
1223                 page = cl_page_list_last(plist);
1224                 cl_page_list_move_head(queue, plist, page);
1225         }
1226
1227         RETURN(rc);
1228 }
1229
1230 static int lov_io_fault_start(const struct lu_env *env,
1231                               const struct cl_io_slice *ios)
1232 {
1233         struct cl_fault_io *fio;
1234         struct lov_io      *lio;
1235         struct lov_io_sub  *sub;
1236
1237         ENTRY;
1238
1239         fio = &ios->cis_io->u.ci_fault;
1240         lio = cl2lov_io(env, ios);
1241         sub = lov_sub_get(env, lio, lov_page_index(fio->ft_page));
1242         sub->sub_io.u.ci_fault.ft_nob = fio->ft_nob;
1243
1244         RETURN(lov_io_start(env, ios));
1245 }
1246
1247 static void lov_io_fsync_end(const struct lu_env *env,
1248                              const struct cl_io_slice *ios)
1249 {
1250         struct lov_io *lio = cl2lov_io(env, ios);
1251         struct lov_io_sub *sub;
1252         unsigned int *written = &ios->cis_io->u.ci_fsync.fi_nr_written;
1253         ENTRY;
1254
1255         *written = 0;
1256         list_for_each_entry(sub, &lio->lis_active, sub_linkage) {
1257                 struct cl_io *subio = &sub->sub_io;
1258
1259                 lov_io_end_wrapper(sub->sub_env, subio);
1260
1261                 if (subio->ci_result == 0)
1262                         *written += subio->u.ci_fsync.fi_nr_written;
1263         }
1264         RETURN_EXIT;
1265 }
1266
1267 static const struct cl_io_operations lov_io_ops = {
1268         .op = {
1269                 [CIT_READ] = {
1270                         .cio_fini      = lov_io_fini,
1271                         .cio_iter_init = lov_io_rw_iter_init,
1272                         .cio_iter_fini = lov_io_iter_fini,
1273                         .cio_lock      = lov_io_lock,
1274                         .cio_unlock    = lov_io_unlock,
1275                         .cio_start     = lov_io_start,
1276                         .cio_end       = lov_io_end
1277                 },
1278                 [CIT_WRITE] = {
1279                         .cio_fini      = lov_io_fini,
1280                         .cio_iter_init = lov_io_rw_iter_init,
1281                         .cio_iter_fini = lov_io_iter_fini,
1282                         .cio_lock      = lov_io_lock,
1283                         .cio_unlock    = lov_io_unlock,
1284                         .cio_start     = lov_io_start,
1285                         .cio_end       = lov_io_end
1286                 },
1287                 [CIT_SETATTR] = {
1288                         .cio_fini      = lov_io_fini,
1289                         .cio_iter_init = lov_io_setattr_iter_init,
1290                         .cio_iter_fini = lov_io_iter_fini,
1291                         .cio_lock      = lov_io_lock,
1292                         .cio_unlock    = lov_io_unlock,
1293                         .cio_start     = lov_io_start,
1294                         .cio_end       = lov_io_end
1295                 },
1296                 [CIT_DATA_VERSION] = {
1297                         .cio_fini       = lov_io_fini,
1298                         .cio_iter_init  = lov_io_iter_init,
1299                         .cio_iter_fini  = lov_io_iter_fini,
1300                         .cio_lock       = lov_io_lock,
1301                         .cio_unlock     = lov_io_unlock,
1302                         .cio_start      = lov_io_start,
1303                         .cio_end        = lov_io_data_version_end,
1304                 },
1305                 [CIT_FAULT] = {
1306                         .cio_fini      = lov_io_fini,
1307                         .cio_iter_init = lov_io_iter_init,
1308                         .cio_iter_fini = lov_io_iter_fini,
1309                         .cio_lock      = lov_io_lock,
1310                         .cio_unlock    = lov_io_unlock,
1311                         .cio_start     = lov_io_fault_start,
1312                         .cio_end       = lov_io_end
1313                 },
1314                 [CIT_FSYNC] = {
1315                         .cio_fini      = lov_io_fini,
1316                         .cio_iter_init = lov_io_iter_init,
1317                         .cio_iter_fini = lov_io_iter_fini,
1318                         .cio_lock      = lov_io_lock,
1319                         .cio_unlock    = lov_io_unlock,
1320                         .cio_start     = lov_io_start,
1321                         .cio_end       = lov_io_fsync_end
1322                 },
1323                 [CIT_LADVISE] = {
1324                         .cio_fini      = lov_io_fini,
1325                         .cio_iter_init = lov_io_iter_init,
1326                         .cio_iter_fini = lov_io_iter_fini,
1327                         .cio_lock      = lov_io_lock,
1328                         .cio_unlock    = lov_io_unlock,
1329                         .cio_start     = lov_io_start,
1330                         .cio_end       = lov_io_end
1331                 },
1332                 [CIT_GLIMPSE] = {
1333                         .cio_fini      = lov_io_fini,
1334                 },
1335                 [CIT_MISC] = {
1336                         .cio_fini      = lov_io_fini
1337                 }
1338         },
1339         .cio_read_ahead                = lov_io_read_ahead,
1340         .cio_submit                    = lov_io_submit,
1341         .cio_commit_async              = lov_io_commit_async,
1342 };
1343
1344 /*****************************************************************************
1345  *
1346  * Empty lov io operations.
1347  *
1348  */
1349
1350 static void lov_empty_io_fini(const struct lu_env *env,
1351                               const struct cl_io_slice *ios)
1352 {
1353         struct lov_object *lov = cl2lov(ios->cis_obj);
1354         ENTRY;
1355
1356         if (atomic_dec_and_test(&lov->lo_active_ios))
1357                 wake_up_all(&lov->lo_waitq);
1358         EXIT;
1359 }
1360
1361 static int lov_empty_io_submit(const struct lu_env *env,
1362                                const struct cl_io_slice *ios,
1363                                enum cl_req_type crt, struct cl_2queue *queue)
1364 {
1365         return -EBADF;
1366 }
1367
1368 static void lov_empty_impossible(const struct lu_env *env,
1369                                  struct cl_io_slice *ios)
1370 {
1371         LBUG();
1372 }
1373
1374 #define LOV_EMPTY_IMPOSSIBLE ((void *)lov_empty_impossible)
1375
1376 /**
1377  * An io operation vector for files without stripes.
1378  */
1379 static const struct cl_io_operations lov_empty_io_ops = {
1380         .op = {
1381                 [CIT_READ] = {
1382                         .cio_fini       = lov_empty_io_fini,
1383 #if 0
1384                         .cio_iter_init  = LOV_EMPTY_IMPOSSIBLE,
1385                         .cio_lock       = LOV_EMPTY_IMPOSSIBLE,
1386                         .cio_start      = LOV_EMPTY_IMPOSSIBLE,
1387                         .cio_end        = LOV_EMPTY_IMPOSSIBLE
1388 #endif
1389                 },
1390                 [CIT_WRITE] = {
1391                         .cio_fini      = lov_empty_io_fini,
1392                         .cio_iter_init = LOV_EMPTY_IMPOSSIBLE,
1393                         .cio_lock      = LOV_EMPTY_IMPOSSIBLE,
1394                         .cio_start     = LOV_EMPTY_IMPOSSIBLE,
1395                         .cio_end       = LOV_EMPTY_IMPOSSIBLE
1396                 },
1397                 [CIT_SETATTR] = {
1398                         .cio_fini      = lov_empty_io_fini,
1399                         .cio_iter_init = LOV_EMPTY_IMPOSSIBLE,
1400                         .cio_lock      = LOV_EMPTY_IMPOSSIBLE,
1401                         .cio_start     = LOV_EMPTY_IMPOSSIBLE,
1402                         .cio_end       = LOV_EMPTY_IMPOSSIBLE
1403                 },
1404                 [CIT_FAULT] = {
1405                         .cio_fini      = lov_empty_io_fini,
1406                         .cio_iter_init = LOV_EMPTY_IMPOSSIBLE,
1407                         .cio_lock      = LOV_EMPTY_IMPOSSIBLE,
1408                         .cio_start     = LOV_EMPTY_IMPOSSIBLE,
1409                         .cio_end       = LOV_EMPTY_IMPOSSIBLE
1410                 },
1411                 [CIT_FSYNC] = {
1412                         .cio_fini      = lov_empty_io_fini
1413                 },
1414                 [CIT_LADVISE] = {
1415                         .cio_fini   = lov_empty_io_fini
1416                 },
1417                 [CIT_GLIMPSE] = {
1418                         .cio_fini      = lov_empty_io_fini
1419                 },
1420                 [CIT_MISC] = {
1421                         .cio_fini      = lov_empty_io_fini
1422                 }
1423         },
1424         .cio_submit                    = lov_empty_io_submit,
1425         .cio_commit_async              = LOV_EMPTY_IMPOSSIBLE
1426 };
1427
1428 int lov_io_init_composite(const struct lu_env *env, struct cl_object *obj,
1429                           struct cl_io *io)
1430 {
1431         struct lov_io       *lio = lov_env_io(env);
1432         struct lov_object   *lov = cl2lov(obj);
1433         int result;
1434         ENTRY;
1435
1436         INIT_LIST_HEAD(&lio->lis_active);
1437         result = lov_io_slice_init(lio, lov, io);
1438         if (result)
1439                 GOTO(out, result);
1440
1441         result = lov_io_subio_init(env, lio, io);
1442         if (!result) {
1443                 cl_io_slice_add(io, &lio->lis_cl, obj, &lov_io_ops);
1444                 atomic_inc(&lov->lo_active_ios);
1445         }
1446         EXIT;
1447 out:
1448         io->ci_result = result < 0 ? result : 0;
1449         return result;
1450 }
1451
1452 int lov_io_init_empty(const struct lu_env *env, struct cl_object *obj,
1453                       struct cl_io *io)
1454 {
1455         struct lov_object *lov = cl2lov(obj);
1456         struct lov_io *lio = lov_env_io(env);
1457         int result;
1458         ENTRY;
1459
1460         lio->lis_object = lov;
1461         switch (io->ci_type) {
1462         default:
1463                 LBUG();
1464         case CIT_MISC:
1465         case CIT_GLIMPSE:
1466         case CIT_READ:
1467                 result = 0;
1468                 break;
1469         case CIT_FSYNC:
1470         case CIT_LADVISE:
1471         case CIT_SETATTR:
1472         case CIT_DATA_VERSION:
1473                 result = +1;
1474                 break;
1475         case CIT_WRITE:
1476                 result = -EBADF;
1477                 break;
1478         case CIT_FAULT:
1479                 result = -EFAULT;
1480                 CERROR("Page fault on a file without stripes: "DFID"\n",
1481                        PFID(lu_object_fid(&obj->co_lu)));
1482                 break;
1483         }
1484         if (result == 0) {
1485                 cl_io_slice_add(io, &lio->lis_cl, obj, &lov_empty_io_ops);
1486                 atomic_inc(&lov->lo_active_ios);
1487         }
1488
1489         io->ci_result = result < 0 ? result : 0;
1490         RETURN(result);
1491 }
1492
1493 int lov_io_init_released(const struct lu_env *env, struct cl_object *obj,
1494                         struct cl_io *io)
1495 {
1496         struct lov_object *lov = cl2lov(obj);
1497         struct lov_io *lio = lov_env_io(env);
1498         int result;
1499         ENTRY;
1500
1501         LASSERT(lov->lo_lsm != NULL);
1502         lio->lis_object = lov;
1503
1504         switch (io->ci_type) {
1505         default:
1506                 LASSERTF(0, "invalid type %d\n", io->ci_type);
1507                 result = -EOPNOTSUPP;
1508                 break;
1509         case CIT_GLIMPSE:
1510         case CIT_MISC:
1511         case CIT_FSYNC:
1512         case CIT_LADVISE:
1513         case CIT_DATA_VERSION:
1514                 result = 1;
1515                 break;
1516         case CIT_SETATTR:
1517                 /* the truncate to 0 is managed by MDT:
1518                  * - in open, for open O_TRUNC
1519                  * - in setattr, for truncate
1520                  */
1521                 /* the truncate is for size > 0 so triggers a restore */
1522                 if (cl_io_is_trunc(io)) {
1523                         io->ci_restore_needed = 1;
1524                         result = -ENODATA;
1525                 } else
1526                         result = 1;
1527                 break;
1528         case CIT_READ:
1529         case CIT_WRITE:
1530         case CIT_FAULT:
1531                 io->ci_restore_needed = 1;
1532                 result = -ENODATA;
1533                 break;
1534         }
1535
1536         if (result == 0) {
1537                 cl_io_slice_add(io, &lio->lis_cl, obj, &lov_empty_io_ops);
1538                 atomic_inc(&lov->lo_active_ios);
1539         }
1540
1541         io->ci_result = result < 0 ? result : 0;
1542         RETURN(result);
1543 }
1544
1545 /**
1546  * Return the index in composite:lo_entries by the file offset
1547  */
1548 int lov_io_layout_at(struct lov_io *lio, __u64 offset)
1549 {
1550         struct lov_object *lov = lio->lis_object;
1551         struct lov_layout_composite *comp = &lov->u.composite;
1552         int start_index = 0;
1553         int end_index = comp->lo_entry_count - 1;
1554         int i;
1555
1556         LASSERT(lov->lo_type == LLT_COMP);
1557
1558         /* This is actual file offset so nothing can cover eof. */
1559         if (offset == LUSTRE_EOF)
1560                 return -1;
1561
1562         if (lov_is_flr(lov)) {
1563                 struct lov_mirror_entry *lre;
1564
1565                 LASSERT(lio->lis_mirror_index >= 0);
1566
1567                 lre = &comp->lo_mirrors[lio->lis_mirror_index];
1568                 start_index = lre->lre_start;
1569                 end_index = lre->lre_end;
1570         }
1571
1572         for (i = start_index; i <= end_index; i++) {
1573                 struct lov_layout_entry *lle = lov_entry(lov, i);
1574
1575                 if ((offset >= lle->lle_extent->e_start &&
1576                      offset < lle->lle_extent->e_end) ||
1577                     (offset == OBD_OBJECT_EOF &&
1578                      lle->lle_extent->e_end == OBD_OBJECT_EOF))
1579                         return i;
1580         }
1581
1582         return -1;
1583 }
1584
1585 /** @} lov */