Whamcloud - gitweb
LU-12616 obclass: fix MDS start/stop race
[fs/lustre-release.git] / lustre / osc / osc_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 OSC 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_OSC
39
40 #include <lustre_obdo.h>
41 #include <lustre_osc.h>
42
43 #include "osc_internal.h"
44
45 /** \addtogroup osc
46  *  @{
47  */
48
49 /*****************************************************************************
50  *
51  * io operations.
52  *
53  */
54
55 static void osc_io_fini(const struct lu_env *env, const struct cl_io_slice *io)
56 {
57 }
58
59 void osc_read_ahead_release(const struct lu_env *env, void *cbdata)
60 {
61         struct ldlm_lock *dlmlock = cbdata;
62         struct lustre_handle lockh;
63
64         ldlm_lock2handle(dlmlock, &lockh);
65         ldlm_lock_decref(&lockh, LCK_PR);
66         LDLM_LOCK_PUT(dlmlock);
67 }
68 EXPORT_SYMBOL(osc_read_ahead_release);
69
70 static int osc_io_read_ahead(const struct lu_env *env,
71                              const struct cl_io_slice *ios,
72                              pgoff_t start, struct cl_read_ahead *ra)
73 {
74         struct osc_object       *osc = cl2osc(ios->cis_obj);
75         struct ldlm_lock        *dlmlock;
76         int                     result = -ENODATA;
77         ENTRY;
78
79         dlmlock = osc_dlmlock_at_pgoff(env, osc, start, 0);
80         if (dlmlock != NULL) {
81                 LASSERT(dlmlock->l_ast_data == osc);
82                 if (dlmlock->l_req_mode != LCK_PR) {
83                         struct lustre_handle lockh;
84                         ldlm_lock2handle(dlmlock, &lockh);
85                         ldlm_lock_addref(&lockh, LCK_PR);
86                         ldlm_lock_decref(&lockh, dlmlock->l_req_mode);
87                 }
88
89                 ra->cra_rpc_size = osc_cli(osc)->cl_max_pages_per_rpc;
90                 ra->cra_end = cl_index(osc2cl(osc),
91                                        dlmlock->l_policy_data.l_extent.end);
92                 ra->cra_release = osc_read_ahead_release;
93                 ra->cra_cbdata = dlmlock;
94                 result = 0;
95         }
96
97         RETURN(result);
98 }
99
100 /**
101  * An implementation of cl_io_operations::cio_io_submit() method for osc
102  * layer. Iterates over pages in the in-queue, prepares each for io by calling
103  * cl_page_prep() and then either submits them through osc_io_submit_page()
104  * or, if page is already submitted, changes osc flags through
105  * osc_set_async_flags().
106  */
107 int osc_io_submit(const struct lu_env *env, const struct cl_io_slice *ios,
108                   enum cl_req_type crt, struct cl_2queue *queue)
109 {
110         struct cl_page    *page;
111         struct cl_page    *tmp;
112         struct client_obd *cli  = NULL;
113         struct osc_object *osc  = NULL; /* to keep gcc happy */
114         struct osc_page   *opg;
115         struct cl_io      *io;
116         struct list_head  list = LIST_HEAD_INIT(list);
117
118         struct cl_page_list *qin      = &queue->c2_qin;
119         struct cl_page_list *qout     = &queue->c2_qout;
120         unsigned int queued = 0;
121         int result = 0;
122         int brw_flags;
123         unsigned int max_pages;
124         unsigned int ppc_bits; /* pages per chunk bits */
125         unsigned int ppc;
126         bool sync_queue = false;
127
128         LASSERT(qin->pl_nr > 0);
129
130         CDEBUG(D_CACHE|D_READA, "%d %d\n", qin->pl_nr, crt);
131
132         osc = cl2osc(ios->cis_obj);
133         cli = osc_cli(osc);
134         max_pages = cli->cl_max_pages_per_rpc;
135         ppc_bits = cli->cl_chunkbits - PAGE_SHIFT;
136         ppc = 1 << ppc_bits;
137
138         brw_flags = osc_io_srvlock(cl2osc_io(env, ios)) ? OBD_BRW_SRVLOCK : 0;
139         brw_flags |= crt == CRT_WRITE ? OBD_BRW_WRITE : OBD_BRW_READ;
140         if (crt == CRT_READ && ios->cis_io->ci_ndelay)
141                 brw_flags |= OBD_BRW_NDELAY;
142
143         /*
144          * NOTE: here @page is a top-level page. This is done to avoid
145          *       creation of sub-page-list.
146          */
147         cl_page_list_for_each_safe(page, tmp, qin) {
148                 struct osc_async_page *oap;
149
150                 /* Top level IO. */
151                 io = page->cp_owner;
152                 LASSERT(io != NULL);
153
154                 opg = osc_cl_page_osc(page, osc);
155                 oap = &opg->ops_oap;
156                 LASSERT(osc == oap->oap_obj);
157
158                 if (!list_empty(&oap->oap_pending_item) ||
159                     !list_empty(&oap->oap_rpc_item)) {
160                         CDEBUG(D_CACHE, "Busy oap %p page %p for submit.\n",
161                                oap, opg);
162                         result = -EBUSY;
163                         break;
164                 }
165
166                 result = cl_page_prep(env, io, page, crt);
167                 if (result != 0) {
168                         LASSERT(result < 0);
169                         if (result != -EALREADY)
170                                 break;
171                         /*
172                          * Handle -EALREADY error: for read case, the page is
173                          * already in UPTODATE state; for write, the page
174                          * is not dirty.
175                          */
176                         result = 0;
177                         continue;
178                 }
179
180                 spin_lock(&oap->oap_lock);
181                 oap->oap_async_flags = ASYNC_URGENT|ASYNC_READY;
182                 oap->oap_async_flags |= ASYNC_COUNT_STABLE;
183                 spin_unlock(&oap->oap_lock);
184
185                 osc_page_submit(env, opg, crt, brw_flags);
186                 list_add_tail(&oap->oap_pending_item, &list);
187
188                 if (page->cp_sync_io != NULL)
189                         cl_page_list_move(qout, qin, page);
190                 else /* async IO */
191                         cl_page_list_del(env, qin, page);
192
193                 queued++;
194                 if (queued == max_pages) {
195                         sync_queue = true;
196                 } else if (crt == CRT_WRITE) {
197                         unsigned int chunks;
198                         unsigned int next_chunks;
199
200                         chunks = (queued + ppc - 1) >> ppc_bits;
201                         /* chunk number if add another page */
202                         next_chunks = (queued + ppc) >> ppc_bits;
203
204                         /* next page will excceed write chunk limit */
205                         if (chunks == osc_max_write_chunks(cli) &&
206                             next_chunks > chunks)
207                                 sync_queue = true;
208                 }
209
210                 if (sync_queue) {
211                         result = osc_queue_sync_pages(env, io, osc, &list,
212                                                       brw_flags);
213                         if (result < 0)
214                                 break;
215                         queued = 0;
216                         sync_queue = false;
217                 }
218         }
219
220         if (queued > 0)
221                 result = osc_queue_sync_pages(env, io, osc, &list, brw_flags);
222
223         /* Update c/mtime for sync write. LU-7310 */
224         if (crt == CRT_WRITE && qout->pl_nr > 0 && result == 0) {
225                 struct cl_object *obj   = ios->cis_obj;
226                 struct cl_attr *attr = &osc_env_info(env)->oti_attr;
227
228                 cl_object_attr_lock(obj);
229                 attr->cat_mtime = attr->cat_ctime = ktime_get_real_seconds();
230                 cl_object_attr_update(env, obj, attr, CAT_MTIME | CAT_CTIME);
231                 cl_object_attr_unlock(obj);
232         }
233
234         CDEBUG(D_INFO, "%d/%d %d\n", qin->pl_nr, qout->pl_nr, result);
235         return qout->pl_nr > 0 ? 0 : result;
236 }
237 EXPORT_SYMBOL(osc_io_submit);
238
239 /**
240  * This is called to update the attributes when modifying a specific page,
241  * both when making new pages and when doing updates to existing cached pages.
242  *
243  * Expand stripe KMS if necessary.
244  */
245 void osc_page_touch_at(const struct lu_env *env, struct cl_object *obj,
246                        pgoff_t idx, size_t to)
247 {
248         struct lov_oinfo  *loi  = cl2osc(obj)->oo_oinfo;
249         struct cl_attr    *attr = &osc_env_info(env)->oti_attr;
250         int valid;
251         __u64 kms;
252
253         ENTRY;
254
255         /* offset within stripe */
256         kms = cl_offset(obj, idx) + to;
257
258         cl_object_attr_lock(obj);
259         CDEBUG(D_INODE, "stripe KMS %sincreasing %llu->%llu %llu\n",
260                kms > loi->loi_kms ? "" : "not ", loi->loi_kms, kms,
261                loi->loi_lvb.lvb_size);
262
263         attr->cat_mtime = attr->cat_ctime = ktime_get_real_seconds();
264         valid = CAT_MTIME | CAT_CTIME;
265         if (kms > loi->loi_kms) {
266                 attr->cat_kms = kms;
267                 valid |= CAT_KMS;
268         }
269         if (kms > loi->loi_lvb.lvb_size) {
270                 attr->cat_size = kms;
271                 valid |= CAT_SIZE;
272         }
273         cl_object_attr_update(env, obj, attr, valid);
274         cl_object_attr_unlock(obj);
275
276         EXIT;
277 }
278
279 int osc_io_commit_async(const struct lu_env *env,
280                         const struct cl_io_slice *ios,
281                         struct cl_page_list *qin, int from, int to,
282                         cl_commit_cbt cb)
283 {
284         struct cl_io    *io = ios->cis_io;
285         struct osc_io   *oio = cl2osc_io(env, ios);
286         struct osc_object *osc = cl2osc(ios->cis_obj);
287         struct cl_page  *page;
288         struct cl_page  *last_page;
289         struct osc_page *opg;
290         int result = 0;
291         ENTRY;
292
293         LASSERT(qin->pl_nr > 0);
294
295         /* Handle partial page cases */
296         last_page = cl_page_list_last(qin);
297         if (oio->oi_lockless) {
298                 page = cl_page_list_first(qin);
299                 if (page == last_page) {
300                         cl_page_clip(env, page, from, to);
301                 } else {
302                         if (from != 0)
303                                 cl_page_clip(env, page, from, PAGE_SIZE);
304                         if (to != PAGE_SIZE)
305                                 cl_page_clip(env, last_page, 0, to);
306                 }
307         }
308
309         while (qin->pl_nr > 0) {
310                 struct osc_async_page *oap;
311
312                 page = cl_page_list_first(qin);
313                 opg = osc_cl_page_osc(page, osc);
314                 oap = &opg->ops_oap;
315
316                 LASSERTF(osc == oap->oap_obj,
317                          "obj mismatch: %p / %p\n", osc, oap->oap_obj);
318
319                 if (!list_empty(&oap->oap_rpc_item)) {
320                         CDEBUG(D_CACHE, "Busy oap %p page %p for submit.\n",
321                                oap, opg);
322                         result = -EBUSY;
323                         break;
324                 }
325
326                 /* The page may be already in dirty cache. */
327                 if (list_empty(&oap->oap_pending_item)) {
328                         result = osc_page_cache_add(env, &opg->ops_cl, io);
329                         if (result != 0)
330                                 break;
331                 }
332
333                 osc_page_touch_at(env, osc2cl(osc), osc_index(opg),
334                                   page == last_page ? to : PAGE_SIZE);
335
336                 cl_page_list_del(env, qin, page);
337
338                 (*cb)(env, io, page);
339                 /* Can't access page any more. Page can be in transfer and
340                  * complete at any time. */
341         }
342
343         /* for sync write, kernel will wait for this page to be flushed before
344          * osc_io_end() is called, so release it earlier.
345          * for mkwrite(), it's known there is no further pages. */
346         if (cl_io_is_sync_write(io) && oio->oi_active != NULL) {
347                 osc_extent_release(env, oio->oi_active);
348                 oio->oi_active = NULL;
349         }
350
351         CDEBUG(D_INFO, "%d %d\n", qin->pl_nr, result);
352         RETURN(result);
353 }
354 EXPORT_SYMBOL(osc_io_commit_async);
355
356 int osc_io_iter_init(const struct lu_env *env, const struct cl_io_slice *ios)
357 {
358         struct osc_object *osc = cl2osc(ios->cis_obj);
359         struct obd_import *imp = osc_cli(osc)->cl_import;
360         struct osc_io *oio = osc_env_io(env);
361         int rc = -EIO;
362
363         spin_lock(&imp->imp_lock);
364         if (likely(!imp->imp_invalid)) {
365                 atomic_inc(&osc->oo_nr_ios);
366                 oio->oi_is_active = 1;
367                 rc = 0;
368         }
369         spin_unlock(&imp->imp_lock);
370
371         if (cfs_capable(CFS_CAP_SYS_RESOURCE))
372                 oio->oi_cap_sys_resource = 1;
373
374         return rc;
375 }
376 EXPORT_SYMBOL(osc_io_iter_init);
377
378 int osc_io_rw_iter_init(const struct lu_env *env,
379                         const struct cl_io_slice *ios)
380 {
381         struct cl_io *io = ios->cis_io;
382         struct osc_io *oio = osc_env_io(env);
383         struct osc_object *osc = cl2osc(ios->cis_obj);
384         unsigned long npages;
385         ENTRY;
386
387         if (cl_io_is_append(io))
388                 RETURN(osc_io_iter_init(env, ios));
389
390         npages = io->u.ci_rw.crw_count >> PAGE_SHIFT;
391         if (io->u.ci_rw.crw_pos & ~PAGE_MASK)
392                 ++npages;
393
394         oio->oi_lru_reserved = osc_lru_reserve(osc_cli(osc), npages);
395
396         RETURN(osc_io_iter_init(env, ios));
397 }
398 EXPORT_SYMBOL(osc_io_rw_iter_init);
399
400 void osc_io_iter_fini(const struct lu_env *env,
401                       const struct cl_io_slice *ios)
402 {
403         struct osc_io *oio = osc_env_io(env);
404
405         if (oio->oi_is_active) {
406                 struct osc_object *osc = cl2osc(ios->cis_obj);
407
408                 oio->oi_is_active = 0;
409                 LASSERT(atomic_read(&osc->oo_nr_ios) > 0);
410                 if (atomic_dec_and_test(&osc->oo_nr_ios))
411                         wake_up_all(&osc->oo_io_waitq);
412         }
413 }
414 EXPORT_SYMBOL(osc_io_iter_fini);
415
416 void osc_io_rw_iter_fini(const struct lu_env *env,
417                          const struct cl_io_slice *ios)
418 {
419         struct osc_io *oio = osc_env_io(env);
420         struct osc_object *osc = cl2osc(ios->cis_obj);
421
422         if (oio->oi_lru_reserved > 0) {
423                 osc_lru_unreserve(osc_cli(osc), oio->oi_lru_reserved);
424                 oio->oi_lru_reserved = 0;
425         }
426         oio->oi_write_osclock = NULL;
427
428         osc_io_iter_fini(env, ios);
429 }
430 EXPORT_SYMBOL(osc_io_rw_iter_fini);
431
432 int osc_io_fault_start(const struct lu_env *env, const struct cl_io_slice *ios)
433 {
434         struct cl_io       *io;
435         struct cl_fault_io *fio;
436         ENTRY;
437
438         io  = ios->cis_io;
439         fio = &io->u.ci_fault;
440         CDEBUG(D_INFO, "%lu %d %zu\n",
441                 fio->ft_index, fio->ft_writable, fio->ft_nob);
442         /*
443          * If mapping is writeable, adjust kms to cover this page,
444          * but do not extend kms beyond actual file size.
445          * See bug 10919.
446          */
447         if (fio->ft_writable)
448                 osc_page_touch_at(env, ios->cis_obj,
449                                   fio->ft_index, fio->ft_nob);
450         RETURN(0);
451 }
452 EXPORT_SYMBOL(osc_io_fault_start);
453
454
455 static int osc_async_upcall(void *a, int rc)
456 {
457         struct osc_async_cbargs *args = a;
458
459         args->opc_rc = rc;
460         complete(&args->opc_sync);
461         return 0;
462 }
463
464 /**
465  * Checks that there are no pages being written in the extent being truncated.
466  */
467 static int trunc_check_cb(const struct lu_env *env, struct cl_io *io,
468                           struct osc_page *ops , void *cbdata)
469 {
470         struct cl_page *page = ops->ops_cl.cpl_page;
471         struct osc_async_page *oap;
472         __u64 start = *(__u64 *)cbdata;
473
474         oap = &ops->ops_oap;
475         if (oap->oap_cmd & OBD_BRW_WRITE &&
476             !list_empty(&oap->oap_pending_item))
477                 CL_PAGE_DEBUG(D_ERROR, env, page, "exists %llu/%s.\n",
478                                 start, current->comm);
479
480         if (PageLocked(page->cp_vmpage))
481                 CDEBUG(D_CACHE, "page %p index %lu locked for %d.\n",
482                        ops, osc_index(ops), oap->oap_cmd & OBD_BRW_RWMASK);
483
484         return CLP_GANG_OKAY;
485 }
486
487 static void osc_trunc_check(const struct lu_env *env, struct cl_io *io,
488                             struct osc_io *oio, __u64 size)
489 {
490         struct cl_object *clob;
491         int     partial;
492         pgoff_t start;
493
494         clob    = oio->oi_cl.cis_obj;
495         start   = cl_index(clob, size);
496         partial = cl_offset(clob, start) < size;
497
498         /*
499          * Complain if there are pages in the truncated region.
500          */
501         osc_page_gang_lookup(env, io, cl2osc(clob),
502                                 start + partial, CL_PAGE_EOF,
503                                 trunc_check_cb, (void *)&size);
504 }
505
506 static int osc_io_setattr_start(const struct lu_env *env,
507                                 const struct cl_io_slice *slice)
508 {
509         struct cl_io            *io     = slice->cis_io;
510         struct osc_io           *oio    = cl2osc_io(env, slice);
511         struct cl_object        *obj    = slice->cis_obj;
512         struct lov_oinfo        *loi    = cl2osc(obj)->oo_oinfo;
513         struct cl_attr          *attr   = &osc_env_info(env)->oti_attr;
514         struct obdo             *oa     = &oio->oi_oa;
515         struct osc_async_cbargs *cbargs = &oio->oi_cbarg;
516         __u64                    size   = io->u.ci_setattr.sa_attr.lvb_size;
517         unsigned int ia_avalid = io->u.ci_setattr.sa_avalid;
518         enum op_xvalid ia_xvalid = io->u.ci_setattr.sa_xvalid;
519         int result = 0;
520
521         ENTRY;
522         /* truncate cache dirty pages first */
523         if (cl_io_is_trunc(io))
524                 result = osc_cache_truncate_start(env, cl2osc(obj), size,
525                                                   &oio->oi_trunc);
526
527         if (result == 0 && oio->oi_lockless == 0) {
528                 cl_object_attr_lock(obj);
529                 result = cl_object_attr_get(env, obj, attr);
530                 if (result == 0) {
531                         struct ost_lvb *lvb = &io->u.ci_setattr.sa_attr;
532                         unsigned int cl_valid = 0;
533
534                         if (ia_avalid & ATTR_SIZE) {
535                                 attr->cat_size = size;
536                                 attr->cat_kms = size;
537                                 cl_valid = (CAT_SIZE | CAT_KMS);
538                         }
539                         if (ia_avalid & ATTR_MTIME_SET) {
540                                 attr->cat_mtime = lvb->lvb_mtime;
541                                 cl_valid |= CAT_MTIME;
542                         }
543                         if (ia_avalid & ATTR_ATIME_SET) {
544                                 attr->cat_atime = lvb->lvb_atime;
545                                 cl_valid |= CAT_ATIME;
546                         }
547                         if (ia_xvalid & OP_XVALID_CTIME_SET) {
548                                 attr->cat_ctime = lvb->lvb_ctime;
549                                 cl_valid |= CAT_CTIME;
550                         }
551                         result = cl_object_attr_update(env, obj, attr,
552                                                        cl_valid);
553                 }
554                 cl_object_attr_unlock(obj);
555         }
556         memset(oa, 0, sizeof(*oa));
557         if (result == 0) {
558                 oa->o_oi = loi->loi_oi;
559                 obdo_set_parent_fid(oa, io->u.ci_setattr.sa_parent_fid);
560                 oa->o_stripe_idx = io->u.ci_setattr.sa_stripe_index;
561                 oa->o_layout = io->u.ci_setattr.sa_layout;
562                 oa->o_valid |= OBD_MD_FLID | OBD_MD_FLGROUP |
563                         OBD_MD_FLOSTLAYOUT;
564                 if (ia_avalid & ATTR_CTIME) {
565                         oa->o_valid |= OBD_MD_FLCTIME;
566                         oa->o_ctime = attr->cat_ctime;
567                 }
568                 if (ia_avalid & ATTR_ATIME) {
569                         oa->o_valid |= OBD_MD_FLATIME;
570                         oa->o_atime = attr->cat_atime;
571                 }
572                 if (ia_avalid & ATTR_MTIME) {
573                         oa->o_valid |= OBD_MD_FLMTIME;
574                         oa->o_mtime = attr->cat_mtime;
575                 }
576                 if (ia_avalid & ATTR_SIZE) {
577                         oa->o_size = size;
578                         oa->o_blocks = OBD_OBJECT_EOF;
579                         oa->o_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
580
581                         if (oio->oi_lockless) {
582                                 oa->o_flags = OBD_FL_SRVLOCK;
583                                 oa->o_valid |= OBD_MD_FLFLAGS;
584                         }
585
586                         if (io->ci_layout_version > 0) {
587                                 /* verify layout version */
588                                 oa->o_valid |= OBD_MD_LAYOUT_VERSION;
589                                 oa->o_layout_version = io->ci_layout_version;
590                         }
591                 } else {
592                         LASSERT(oio->oi_lockless == 0);
593                 }
594
595                 if (ia_xvalid & OP_XVALID_FLAGS) {
596                         oa->o_flags = io->u.ci_setattr.sa_attr_flags;
597                         oa->o_valid |= OBD_MD_FLFLAGS;
598                 }
599
600                 init_completion(&cbargs->opc_sync);
601
602                 if (ia_avalid & ATTR_SIZE)
603                         result = osc_punch_send(osc_export(cl2osc(obj)),
604                                                 oa, osc_async_upcall, cbargs);
605                 else
606                         result = osc_setattr_async(osc_export(cl2osc(obj)),
607                                                    oa, osc_async_upcall,
608                                                    cbargs, PTLRPCD_SET);
609
610                 cbargs->opc_rpc_sent = result == 0;
611         }
612
613         RETURN(result);
614 }
615
616 void osc_io_setattr_end(const struct lu_env *env,
617                         const struct cl_io_slice *slice)
618 {
619         struct cl_io     *io  = slice->cis_io;
620         struct osc_io    *oio = cl2osc_io(env, slice);
621         struct cl_object *obj = slice->cis_obj;
622         struct osc_async_cbargs *cbargs = &oio->oi_cbarg;
623         struct cl_attr  *attr = &osc_env_info(env)->oti_attr;
624         struct obdo *oa = &oio->oi_oa;
625         unsigned int cl_valid = 0;
626         int result = 0;
627
628         if (cbargs->opc_rpc_sent) {
629                 wait_for_completion(&cbargs->opc_sync);
630                 result = io->ci_result = cbargs->opc_rc;
631         }
632
633         if (result == 0) {
634                 if (oio->oi_lockless) {
635                         /* lockless truncate */
636                         struct osc_device *osd = lu2osc_dev(obj->co_lu.lo_dev);
637
638                         LASSERT(cl_io_is_trunc(io));
639                         /* XXX: Need a lock. */
640                         osd->od_stats.os_lockless_truncates++;
641                 }
642         }
643
644         if (cl_io_is_trunc(io)) {
645                 __u64 size = io->u.ci_setattr.sa_attr.lvb_size;
646                 cl_object_attr_lock(obj);
647                 if (oa->o_valid & OBD_MD_FLBLOCKS) {
648                         attr->cat_blocks = oa->o_blocks;
649                         cl_valid |= CAT_BLOCKS;
650                 }
651
652                 cl_object_attr_update(env, obj, attr, cl_valid);
653                 cl_object_attr_unlock(obj);
654                 osc_trunc_check(env, io, oio, size);
655                 osc_cache_truncate_end(env, oio->oi_trunc);
656                 oio->oi_trunc = NULL;
657         }
658 }
659 EXPORT_SYMBOL(osc_io_setattr_end);
660
661 struct osc_data_version_args {
662         struct osc_io *dva_oio;
663 };
664
665 static int
666 osc_data_version_interpret(const struct lu_env *env, struct ptlrpc_request *req,
667                            void *args, int rc)
668 {
669         struct osc_data_version_args *dva = args;
670         struct osc_io *oio = dva->dva_oio;
671         const struct ost_body *body;
672
673         ENTRY;
674         if (rc < 0)
675                 GOTO(out, rc);
676
677         body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
678         if (body == NULL)
679                 GOTO(out, rc = -EPROTO);
680
681         lustre_get_wire_obdo(&req->rq_import->imp_connect_data, &oio->oi_oa,
682                              &body->oa);
683         EXIT;
684 out:
685         oio->oi_cbarg.opc_rc = rc;
686         complete(&oio->oi_cbarg.opc_sync);
687
688         return 0;
689 }
690
691 static int osc_io_data_version_start(const struct lu_env *env,
692                                      const struct cl_io_slice *slice)
693 {
694         struct cl_data_version_io *dv   = &slice->cis_io->u.ci_data_version;
695         struct osc_io           *oio    = cl2osc_io(env, slice);
696         struct obdo             *oa     = &oio->oi_oa;
697         struct osc_async_cbargs *cbargs = &oio->oi_cbarg;
698         struct osc_object       *obj    = cl2osc(slice->cis_obj);
699         struct lov_oinfo        *loi    = obj->oo_oinfo;
700         struct obd_export       *exp    = osc_export(obj);
701         struct ptlrpc_request   *req;
702         struct ost_body         *body;
703         struct osc_data_version_args *dva;
704         int rc;
705
706         ENTRY;
707         memset(oa, 0, sizeof(*oa));
708         oa->o_oi = loi->loi_oi;
709         oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
710
711         if (dv->dv_flags & (LL_DV_RD_FLUSH | LL_DV_WR_FLUSH)) {
712                 oa->o_valid |= OBD_MD_FLFLAGS;
713                 oa->o_flags |= OBD_FL_SRVLOCK;
714                 if (dv->dv_flags & LL_DV_WR_FLUSH)
715                         oa->o_flags |= OBD_FL_FLUSH;
716         }
717
718         init_completion(&cbargs->opc_sync);
719
720         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OST_GETATTR);
721         if (req == NULL)
722                 RETURN(-ENOMEM);
723
724         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_GETATTR);
725         if (rc < 0) {
726                 ptlrpc_request_free(req);
727                 RETURN(rc);
728         }
729
730         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
731         lustre_set_wire_obdo(&req->rq_import->imp_connect_data, &body->oa, oa);
732
733         ptlrpc_request_set_replen(req);
734         req->rq_interpret_reply = osc_data_version_interpret;
735         CLASSERT(sizeof(*dva) <= sizeof(req->rq_async_args));
736         dva = ptlrpc_req_async_args(req);
737         dva->dva_oio = oio;
738
739         ptlrpcd_add_req(req);
740
741         RETURN(0);
742 }
743
744 static void osc_io_data_version_end(const struct lu_env *env,
745                                     const struct cl_io_slice *slice)
746 {
747         struct cl_data_version_io *dv = &slice->cis_io->u.ci_data_version;
748         struct osc_io           *oio    = cl2osc_io(env, slice);
749         struct osc_async_cbargs *cbargs = &oio->oi_cbarg;
750
751         ENTRY;
752         wait_for_completion(&cbargs->opc_sync);
753
754         if (cbargs->opc_rc != 0) {
755                 slice->cis_io->ci_result = cbargs->opc_rc;
756         } else {
757                 slice->cis_io->ci_result = 0;
758                 if (!(oio->oi_oa.o_valid &
759                       (OBD_MD_LAYOUT_VERSION | OBD_MD_FLDATAVERSION)))
760                         slice->cis_io->ci_result = -ENOTSUPP;
761
762                 if (oio->oi_oa.o_valid & OBD_MD_LAYOUT_VERSION)
763                         dv->dv_layout_version = oio->oi_oa.o_layout_version;
764                 if (oio->oi_oa.o_valid & OBD_MD_FLDATAVERSION)
765                         dv->dv_data_version = oio->oi_oa.o_data_version;
766         }
767
768         EXIT;
769 }
770
771 int osc_io_read_start(const struct lu_env *env,
772                       const struct cl_io_slice *slice)
773 {
774         struct cl_object *obj  = slice->cis_obj;
775         struct cl_attr   *attr = &osc_env_info(env)->oti_attr;
776         int rc = 0;
777         ENTRY;
778
779         if (!slice->cis_io->ci_noatime) {
780                 cl_object_attr_lock(obj);
781                 attr->cat_atime = ktime_get_real_seconds();
782                 rc = cl_object_attr_update(env, obj, attr, CAT_ATIME);
783                 cl_object_attr_unlock(obj);
784         }
785
786         RETURN(rc);
787 }
788 EXPORT_SYMBOL(osc_io_read_start);
789
790 int osc_io_write_start(const struct lu_env *env,
791                        const struct cl_io_slice *slice)
792 {
793         struct cl_object *obj   = slice->cis_obj;
794         struct cl_attr   *attr  = &osc_env_info(env)->oti_attr;
795         int rc = 0;
796         ENTRY;
797
798         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_DELAY_SETTIME, 1);
799         cl_object_attr_lock(obj);
800         attr->cat_mtime = attr->cat_ctime = ktime_get_real_seconds();
801         rc = cl_object_attr_update(env, obj, attr, CAT_MTIME | CAT_CTIME);
802         cl_object_attr_unlock(obj);
803
804         RETURN(rc);
805 }
806 EXPORT_SYMBOL(osc_io_write_start);
807
808 int osc_fsync_ost(const struct lu_env *env, struct osc_object *obj,
809                   struct cl_fsync_io *fio)
810 {
811         struct osc_io    *oio   = osc_env_io(env);
812         struct obdo      *oa    = &oio->oi_oa;
813         struct lov_oinfo *loi   = obj->oo_oinfo;
814         struct osc_async_cbargs *cbargs = &oio->oi_cbarg;
815         int rc = 0;
816         ENTRY;
817
818         memset(oa, 0, sizeof(*oa));
819         oa->o_oi = loi->loi_oi;
820         oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
821
822         /* reload size abd blocks for start and end of sync range */
823         oa->o_size = fio->fi_start;
824         oa->o_blocks = fio->fi_end;
825         oa->o_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
826
827         obdo_set_parent_fid(oa, fio->fi_fid);
828
829         init_completion(&cbargs->opc_sync);
830
831         rc = osc_sync_base(obj, oa, osc_async_upcall, cbargs, PTLRPCD_SET);
832         RETURN(rc);
833 }
834 EXPORT_SYMBOL(osc_fsync_ost);
835
836 int osc_io_fsync_start(const struct lu_env *env,
837                        const struct cl_io_slice *slice)
838 {
839         struct cl_io       *io  = slice->cis_io;
840         struct cl_fsync_io *fio = &io->u.ci_fsync;
841         struct cl_object   *obj = slice->cis_obj;
842         struct osc_object  *osc = cl2osc(obj);
843         pgoff_t start  = cl_index(obj, fio->fi_start);
844         pgoff_t end    = cl_index(obj, fio->fi_end);
845         int     result = 0;
846         ENTRY;
847
848         if (fio->fi_end == OBD_OBJECT_EOF)
849                 end = CL_PAGE_EOF;
850
851         result = osc_cache_writeback_range(env, osc, start, end, 0,
852                                            fio->fi_mode == CL_FSYNC_DISCARD);
853         if (result > 0) {
854                 fio->fi_nr_written += result;
855                 result = 0;
856         }
857         if (fio->fi_mode == CL_FSYNC_ALL) {
858                 int rc;
859
860                 /* we have to wait for writeback to finish before we can
861                  * send OST_SYNC RPC. This is bad because it causes extents
862                  * to be written osc by osc. However, we usually start
863                  * writeback before CL_FSYNC_ALL so this won't have any real
864                  * problem. */
865                 rc = osc_cache_wait_range(env, osc, start, end);
866                 if (result == 0)
867                         result = rc;
868                 rc = osc_fsync_ost(env, osc, fio);
869                 if (result == 0)
870                         result = rc;
871         }
872
873         RETURN(result);
874 }
875
876 void osc_io_fsync_end(const struct lu_env *env,
877                       const struct cl_io_slice *slice)
878 {
879         struct cl_fsync_io *fio = &slice->cis_io->u.ci_fsync;
880         struct cl_object   *obj = slice->cis_obj;
881         pgoff_t start = cl_index(obj, fio->fi_start);
882         pgoff_t end   = cl_index(obj, fio->fi_end);
883         int result = 0;
884
885         if (fio->fi_mode == CL_FSYNC_LOCAL) {
886                 result = osc_cache_wait_range(env, cl2osc(obj), start, end);
887         } else if (fio->fi_mode == CL_FSYNC_ALL) {
888                 struct osc_io           *oio    = cl2osc_io(env, slice);
889                 struct osc_async_cbargs *cbargs = &oio->oi_cbarg;
890
891                 wait_for_completion(&cbargs->opc_sync);
892                 if (result == 0)
893                         result = cbargs->opc_rc;
894         }
895         slice->cis_io->ci_result = result;
896 }
897 EXPORT_SYMBOL(osc_io_fsync_end);
898
899 static int osc_io_ladvise_start(const struct lu_env *env,
900                                 const struct cl_io_slice *slice)
901 {
902         int                      result = 0;
903         struct cl_io            *io = slice->cis_io;
904         struct osc_io           *oio = cl2osc_io(env, slice);
905         struct cl_object        *obj = slice->cis_obj;
906         struct lov_oinfo        *loi = cl2osc(obj)->oo_oinfo;
907         struct cl_ladvise_io    *lio = &io->u.ci_ladvise;
908         struct obdo             *oa = &oio->oi_oa;
909         struct osc_async_cbargs *cbargs = &oio->oi_cbarg;
910         struct lu_ladvise       *ladvise;
911         struct ladvise_hdr      *ladvise_hdr;
912         int                      buf_size;
913         int                      num_advise = 1;
914         ENTRY;
915
916         /* TODO: add multiple ladvise support in CLIO */
917         buf_size = offsetof(typeof(*ladvise_hdr), lah_advise[num_advise]);
918         if (osc_env_info(env)->oti_ladvise_buf.lb_len < buf_size)
919                 lu_buf_realloc(&osc_env_info(env)->oti_ladvise_buf, buf_size);
920
921         ladvise_hdr = osc_env_info(env)->oti_ladvise_buf.lb_buf;
922         if (ladvise_hdr == NULL)
923                 RETURN(-ENOMEM);
924
925         memset(ladvise_hdr, 0, buf_size);
926         ladvise_hdr->lah_magic = LADVISE_MAGIC;
927         ladvise_hdr->lah_count = num_advise;
928         ladvise_hdr->lah_flags = lio->li_flags;
929
930         memset(oa, 0, sizeof(*oa));
931         oa->o_oi = loi->loi_oi;
932         oa->o_valid = OBD_MD_FLID;
933         obdo_set_parent_fid(oa, lio->li_fid);
934
935         ladvise = ladvise_hdr->lah_advise;
936         ladvise->lla_start = lio->li_start;
937         ladvise->lla_end = lio->li_end;
938         ladvise->lla_advice = lio->li_advice;
939
940         if (lio->li_flags & LF_ASYNC) {
941                 result = osc_ladvise_base(osc_export(cl2osc(obj)), oa,
942                                           ladvise_hdr, NULL, NULL, NULL);
943         } else {
944                 init_completion(&cbargs->opc_sync);
945                 result = osc_ladvise_base(osc_export(cl2osc(obj)), oa,
946                                           ladvise_hdr, osc_async_upcall,
947                                           cbargs, PTLRPCD_SET);
948                 cbargs->opc_rpc_sent = result == 0;
949         }
950         RETURN(result);
951 }
952
953 static void osc_io_ladvise_end(const struct lu_env *env,
954                                const struct cl_io_slice *slice)
955 {
956         struct cl_io            *io = slice->cis_io;
957         struct osc_io           *oio = cl2osc_io(env, slice);
958         struct osc_async_cbargs *cbargs = &oio->oi_cbarg;
959         int                      result = 0;
960         struct cl_ladvise_io    *lio = &io->u.ci_ladvise;
961
962         if ((!(lio->li_flags & LF_ASYNC)) && cbargs->opc_rpc_sent) {
963                 wait_for_completion(&cbargs->opc_sync);
964                 result = cbargs->opc_rc;
965         }
966         slice->cis_io->ci_result = result;
967 }
968
969 void osc_io_end(const struct lu_env *env, const struct cl_io_slice *slice)
970 {
971         struct osc_io *oio = cl2osc_io(env, slice);
972
973         if (oio->oi_active) {
974                 osc_extent_release(env, oio->oi_active);
975                 oio->oi_active = NULL;
976         }
977 }
978 EXPORT_SYMBOL(osc_io_end);
979
980 static const struct cl_io_operations osc_io_ops = {
981         .op = {
982                 [CIT_READ] = {
983                         .cio_iter_init = osc_io_rw_iter_init,
984                         .cio_iter_fini = osc_io_rw_iter_fini,
985                         .cio_start  = osc_io_read_start,
986                         .cio_fini   = osc_io_fini
987                 },
988                 [CIT_WRITE] = {
989                         .cio_iter_init = osc_io_rw_iter_init,
990                         .cio_iter_fini = osc_io_rw_iter_fini,
991                         .cio_start  = osc_io_write_start,
992                         .cio_end    = osc_io_end,
993                         .cio_fini   = osc_io_fini
994                 },
995                 [CIT_SETATTR] = {
996                         .cio_iter_init = osc_io_iter_init,
997                         .cio_iter_fini = osc_io_iter_fini,
998                         .cio_start  = osc_io_setattr_start,
999                         .cio_end    = osc_io_setattr_end
1000                 },
1001                 [CIT_DATA_VERSION] = {
1002                         .cio_start  = osc_io_data_version_start,
1003                         .cio_end    = osc_io_data_version_end,
1004                 },
1005                 [CIT_FAULT] = {
1006                         .cio_iter_init = osc_io_iter_init,
1007                         .cio_iter_fini = osc_io_iter_fini,
1008                         .cio_start  = osc_io_fault_start,
1009                         .cio_end    = osc_io_end,
1010                         .cio_fini   = osc_io_fini
1011                 },
1012                 [CIT_FSYNC] = {
1013                         .cio_start  = osc_io_fsync_start,
1014                         .cio_end    = osc_io_fsync_end,
1015                         .cio_fini   = osc_io_fini
1016                 },
1017                 [CIT_LADVISE] = {
1018                         .cio_start  = osc_io_ladvise_start,
1019                         .cio_end    = osc_io_ladvise_end,
1020                         .cio_fini   = osc_io_fini
1021                 },
1022                 [CIT_MISC] = {
1023                         .cio_fini   = osc_io_fini
1024                 }
1025         },
1026         .cio_read_ahead             = osc_io_read_ahead,
1027         .cio_submit                 = osc_io_submit,
1028         .cio_commit_async           = osc_io_commit_async
1029 };
1030
1031 /*****************************************************************************
1032  *
1033  * Transfer operations.
1034  *
1035  */
1036
1037 int osc_io_init(const struct lu_env *env,
1038                 struct cl_object *obj, struct cl_io *io)
1039 {
1040         struct osc_io *oio = osc_env_io(env);
1041
1042         CL_IO_SLICE_CLEAN(oio, oi_cl);
1043         cl_io_slice_add(io, &oio->oi_cl, obj, &osc_io_ops);
1044         return 0;
1045 }
1046
1047 /** @} osc */