X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Flov%2Flov_io.c;h=8322798d467e1007a8d0e468c9de2e793d7d8adb;hp=4ab7d2515389ba5344d89db590331ed6ebb9089c;hb=101f527e687941024311ab42f4c5843421448234;hpb=89f659febd4220cc517a181f8ac6cc52235b76ca;ds=sidebyside

diff --git a/lustre/lov/lov_io.c b/lustre/lov/lov_io.c
index 4ab7d25..8322798 100644
--- a/lustre/lov/lov_io.c
+++ b/lustre/lov/lov_io.c
@@ -1,6 +1,4 @@
-/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
- * vim:expandtab:shiftwidth=8:tabstop=8:
- *
+/*
  * GPL HEADER START
  *
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
@@ -26,8 +24,10 @@
  * GPL HEADER END
  */
 /*
- * Copyright 2008 Sun Microsystems, Inc.  All rights reserved.
+ * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
+ *
+ * Copyright (c) 2011, 2013, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -36,32 +36,24 @@
  * Implementation of cl_io for LOV layer.
  *
  *   Author: Nikita Danilov <nikita.danilov@sun.com>
+ *   Author: Jinshan Xiong <jinshan.xiong@whamcloud.com>
  */
 
 #define DEBUG_SUBSYSTEM S_LOV
 
 #include "lov_cl_internal.h"
 
-/** \addtogroup lov lov @{ */
+/** \addtogroup lov
+ *  @{
+ */
 
-static void lov_sub_enter(struct lov_io_sub *sub)
+static inline void lov_sub_enter(struct lov_io_sub *sub)
 {
-        ENTRY;
-        if (sub->sub_reenter++ == 0) {
-                sub->sub_cookie = cl_env_reenter();
-                cl_env_implant(sub->sub_env, &sub->sub_refcheck2);
-        }
-        EXIT;
+        sub->sub_reenter++;
 }
-
-static void lov_sub_exit(struct lov_io_sub *sub)
+static inline void lov_sub_exit(struct lov_io_sub *sub)
 {
-        ENTRY;
-        if (--sub->sub_reenter == 0) {
-                cl_env_unplant(sub->sub_env, &sub->sub_refcheck2);
-                cl_env_reexit(sub->sub_cookie);
-        }
-        EXIT;
+        sub->sub_reenter--;
 }
 
 static void lov_io_sub_fini(const struct lu_env *env, struct lov_io *lio,
@@ -93,16 +85,20 @@ static void lov_io_sub_fini(const struct lu_env *env, struct lov_io *lio,
 static void lov_io_sub_inherit(struct cl_io *io, struct lov_io *lio,
                                int stripe, loff_t start, loff_t end)
 {
-        struct lov_stripe_md *lsm    = lov_r0(lio->lis_object)->lo_lsm;
-        struct cl_io         *parent = lio->lis_cl.cis_io;
+	struct lov_stripe_md *lsm    = lio->lis_object->lo_lsm;
+	struct cl_io         *parent = lio->lis_cl.cis_io;
 
         switch(io->ci_type) {
-        case CIT_TRUNC: {
-                size_t new_size = parent->u.ci_truncate.tr_size;
-
-                new_size = lov_size_to_stripe(lsm, new_size, stripe);
-                io->u.ci_truncate.tr_capa = parent->u.ci_truncate.tr_capa;
-                io->u.ci_truncate.tr_size = new_size;
+        case CIT_SETATTR: {
+                io->u.ci_setattr.sa_attr = parent->u.ci_setattr.sa_attr;
+                io->u.ci_setattr.sa_valid = parent->u.ci_setattr.sa_valid;
+                io->u.ci_setattr.sa_capa = parent->u.ci_setattr.sa_capa;
+                if (cl_io_is_trunc(io)) {
+                        loff_t new_size = parent->u.ci_setattr.sa_attr.lvb_size;
+
+                        new_size = lov_size_to_stripe(lsm, new_size, stripe);
+                        io->u.ci_setattr.sa_attr.lvb_size = new_size;
+                }
                 break;
         }
         case CIT_FAULT: {
@@ -114,10 +110,23 @@ static void lov_io_sub_inherit(struct cl_io *io, struct lov_io *lio,
                 io->u.ci_fault.ft_index = cl_index(obj, off);
                 break;
         }
-        case CIT_READ:
-        case CIT_WRITE: {
-                io->u.ci_rw.crw_pos = start;
-                io->u.ci_rw.crw_count = end - start;
+	case CIT_FSYNC: {
+		io->u.ci_fsync.fi_start = start;
+		io->u.ci_fsync.fi_end = end;
+		io->u.ci_fsync.fi_capa = parent->u.ci_fsync.fi_capa;
+		io->u.ci_fsync.fi_fid = parent->u.ci_fsync.fi_fid;
+		io->u.ci_fsync.fi_mode = parent->u.ci_fsync.fi_mode;
+		break;
+	}
+	case CIT_READ:
+	case CIT_WRITE: {
+		io->u.ci_wr.wr_sync = cl_io_is_sync_write(parent);
+                if (cl_io_is_append(parent)) {
+                        io->u.ci_wr.wr_append = 1;
+                } else {
+                        io->u.ci_rw.crw_pos = start;
+                        io->u.ci_rw.crw_count = end - start;
+                }
                 break;
         }
         default:
@@ -146,17 +155,8 @@ static int lov_io_sub_init(const struct lu_env *env, struct lov_io *lio,
         sub->sub_io_initialized = 0;
         sub->sub_borrowed = 0;
 
-        /*
-         * First sub-io. Use ->lis_single_subio and current environment, to
-         * avoid dynamic allocation.
-         */
-        if (lio->lis_active_subios == 0) {
-                sub->sub_io = &lio->lis_single_subio;
-                lio->lis_single_subio_index = stripe;
-                sub->sub_env = cl_env_get(&sub->sub_refcheck);
-                LASSERT(sub->sub_env == env);
-        } else if (lio->lis_mem_frozen) {
-                LASSERT(mutex_is_locked(&ld->ld_mutex));
+        if (lio->lis_mem_frozen) {
+		LASSERT(mutex_is_locked(&ld->ld_mutex));
                 sub->sub_io  = &ld->ld_emrg[stripe]->emrg_subio;
                 sub->sub_env = ld->ld_emrg[stripe]->emrg_env;
                 sub->sub_borrowed = 1;
@@ -167,12 +167,23 @@ static int lov_io_sub_init(const struct lu_env *env, struct lov_io *lio,
                 cookie = cl_env_reenter();
                 sub->sub_env = cl_env_get(&sub->sub_refcheck);
                 cl_env_reexit(cookie);
-
-                OBD_ALLOC_PTR(sub->sub_io);
                 if (IS_ERR(sub->sub_env))
                         result = PTR_ERR(sub->sub_env);
-                else if (sub->sub_io == NULL)
-                        result = -ENOMEM;
+
+                if (result == 0) {
+                        /*
+                         * First sub-io. Use ->lis_single_subio to
+                         * avoid dynamic allocation.
+                         */
+                        if (lio->lis_active_subios == 0) {
+                                sub->sub_io = &lio->lis_single_subio;
+                                lio->lis_single_subio_index = stripe;
+                        } else {
+                                OBD_ALLOC_PTR(sub->sub_io);
+                                if (sub->sub_io == NULL)
+                                        result = -ENOMEM;
+                        }
+                }
         }
 
         if (result == 0) {
@@ -185,6 +196,7 @@ static int lov_io_sub_init(const struct lu_env *env, struct lov_io *lio,
                 sub_io->ci_parent  = io;
                 sub_io->ci_lockreq = io->ci_lockreq;
                 sub_io->ci_type    = io->ci_type;
+                sub_io->ci_no_srvlock = io->ci_no_srvlock;
 
                 lov_sub_enter(sub);
                 result = cl_io_sub_init(sub->sub_env, sub_io,
@@ -248,9 +260,9 @@ static int lov_page_stripe(const struct cl_page *page)
 struct lov_io_sub *lov_page_subio(const struct lu_env *env, struct lov_io *lio,
                                   const struct cl_page_slice *slice)
 {
-        struct lov_stripe_md *lsm  = lov_r0(lio->lis_object)->lo_lsm;
-        struct cl_page       *page = slice->cpl_page;
-        int stripe;
+	struct lov_stripe_md *lsm  = lio->lis_object->lo_lsm;
+	struct cl_page       *page = slice->cpl_page;
+	int stripe;
 
         LASSERT(lio->lis_cl.cis_io != NULL);
         LASSERT(cl2lov(slice->cpl_obj) == lio->lis_object);
@@ -266,9 +278,8 @@ struct lov_io_sub *lov_page_subio(const struct lu_env *env, struct lov_io *lio,
 static int lov_io_subio_init(const struct lu_env *env, struct lov_io *lio,
                              struct cl_io *io)
 {
-        struct lov_object    *lov = lio->lis_object;
-        struct lov_stripe_md *lsm = lov_r0(lov)->lo_lsm;
-        int result;
+	struct lov_stripe_md *lsm = lio->lis_object->lo_lsm;
+	int result;
 
         LASSERT(lio->lis_object != NULL);
         ENTRY;
@@ -277,8 +288,8 @@ static int lov_io_subio_init(const struct lu_env *env, struct lov_io *lio,
          * Need to be optimized, we can't afford to allocate a piece of memory
          * when writing a page. -jay
          */
-        OBD_ALLOC(lio->lis_subs,
-                  lsm->lsm_stripe_count * sizeof lio->lis_subs[0]);
+        OBD_ALLOC_LARGE(lio->lis_subs,
+                        lsm->lsm_stripe_count * sizeof lio->lis_subs[0]);
         if (lio->lis_subs != NULL) {
                 lio->lis_nr_subios = lio->lis_stripe_count;
                 lio->lis_single_subio_index = -1;
@@ -292,14 +303,13 @@ static int lov_io_subio_init(const struct lu_env *env, struct lov_io *lio,
 static void lov_io_slice_init(struct lov_io *lio,
                               struct lov_object *obj, struct cl_io *io)
 {
-        struct lov_stripe_md *lsm = lov_r0(obj)->lo_lsm;
+	ENTRY;
 
-        LASSERT(lsm != NULL);
-        ENTRY;
+	io->ci_result = 0;
+	lio->lis_object = obj;
 
-        io->ci_result = 0;
-        lio->lis_object = obj;
-        lio->lis_stripe_count = lsm->lsm_stripe_count;
+	LASSERT(obj->lo_lsm != NULL);
+	lio->lis_stripe_count = obj->lo_lsm->lsm_stripe_count;
 
         switch (io->ci_type) {
         case CIT_READ:
@@ -314,8 +324,11 @@ static void lov_io_slice_init(struct lov_io *lio,
                 }
                 break;
 
-        case CIT_TRUNC:
-                lio->lis_pos = io->u.ci_truncate.tr_size;
+        case CIT_SETATTR:
+                if (cl_io_is_trunc(io))
+                        lio->lis_pos = io->u.ci_setattr.sa_attr.lvb_size;
+                else
+                        lio->lis_pos = 0;
                 lio->lis_endpos = OBD_OBJECT_EOF;
                 break;
 
@@ -326,6 +339,12 @@ static void lov_io_slice_init(struct lov_io *lio,
                 break;
         }
 
+	case CIT_FSYNC: {
+		lio->lis_pos = io->u.ci_fsync.fi_start;
+		lio->lis_endpos = io->u.ci_fsync.fi_end;
+		break;
+	}
+
         case CIT_MISC:
                 lio->lis_pos = 0;
                 lio->lis_endpos = OBD_OBJECT_EOF;
@@ -340,18 +359,23 @@ static void lov_io_slice_init(struct lov_io *lio,
 
 static void lov_io_fini(const struct lu_env *env, const struct cl_io_slice *ios)
 {
-        struct lov_io *lio = cl2lov_io(env, ios);
-        int i;
+	struct lov_io *lio = cl2lov_io(env, ios);
+	struct lov_object *lov = cl2lov(ios->cis_obj);
+	int i;
 
         ENTRY;
         if (lio->lis_subs != NULL) {
                 for (i = 0; i < lio->lis_nr_subios; i++)
                         lov_io_sub_fini(env, lio, &lio->lis_subs[i]);
-                OBD_FREE(lio->lis_subs,
+                OBD_FREE_LARGE(lio->lis_subs,
                          lio->lis_nr_subios * sizeof lio->lis_subs[0]);
                 lio->lis_nr_subios = 0;
         }
-        EXIT;
+
+	LASSERT(cfs_atomic_read(&lov->lo_active_ios) > 0);
+	if (cfs_atomic_dec_and_test(&lov->lo_active_ios))
+		cfs_waitq_broadcast(&lov->lo_waitq);
+	EXIT;
 }
 
 static obd_off lov_offset_mod(obd_off val, int delta)
@@ -364,8 +388,8 @@ static obd_off lov_offset_mod(obd_off val, int delta)
 static int lov_io_iter_init(const struct lu_env *env,
                             const struct cl_io_slice *ios)
 {
-        struct lov_io        *lio = cl2lov_io(env, ios);
-        struct lov_stripe_md *lsm = lov_r0(lio->lis_object)->lo_lsm;
+	struct lov_io        *lio = cl2lov_io(env, ios);
+	struct lov_stripe_md *lsm = lio->lis_object->lo_lsm;
         struct lov_io_sub    *sub;
         obd_off endpos;
         obd_off start;
@@ -387,13 +411,13 @@ static int lov_io_iter_init(const struct lu_env *env,
                                            start, end);
                         rc = cl_io_iter_init(sub->sub_env, sub->sub_io);
                         lov_sub_put(sub);
-                        CDEBUG(D_VFSTRACE, "shrink: %i [%llu, %llu)\n",
+                        CDEBUG(D_VFSTRACE, "shrink: %d ["LPU64", "LPU64")\n",
                                stripe, start, end);
                 } else
                         rc = PTR_ERR(sub);
 
                 if (!rc)
-                        list_add_tail(&sub->sub_linkage, &lio->lis_active);
+                        cfs_list_add_tail(&sub->sub_linkage, &lio->lis_active);
                 else
                         break;
         }
@@ -403,12 +427,12 @@ static int lov_io_iter_init(const struct lu_env *env,
 static int lov_io_rw_iter_init(const struct lu_env *env,
                                const struct cl_io_slice *ios)
 {
-        struct lov_io        *lio = cl2lov_io(env, ios);
-        struct cl_io         *io  = ios->cis_io;
-        struct lov_stripe_md *lsm = lov_r0(cl2lov(ios->cis_obj))->lo_lsm;
+	struct lov_io        *lio = cl2lov_io(env, ios);
+	struct cl_io         *io  = ios->cis_io;
+	struct lov_stripe_md *lsm = lio->lis_object->lo_lsm;
         loff_t start = io->u.ci_rw.crw_pos;
         loff_t next;
-        int ssize = lsm->lsm_stripe_size;
+        unsigned long ssize = lsm->lsm_stripe_size;
 
         LASSERT(io->ci_type == CIT_READ || io->ci_type == CIT_WRITE);
         ENTRY;
@@ -416,42 +440,46 @@ static int lov_io_rw_iter_init(const struct lu_env *env,
         /* fast path for common case. */
         if (lio->lis_nr_subios != 1 && !cl_io_is_append(io)) {
 
-                do_div(start, ssize);
-                next = (start + 1) * ssize;
-                if (next <= start * ssize)
-                        next = ~0ull;
+		lov_do_div64(start, ssize);
+		next = (start + 1) * ssize;
+		if (next <= start * ssize)
+			next = ~0ull;
 
                 io->ci_continue = next < lio->lis_io_endpos;
                 io->u.ci_rw.crw_count = min_t(loff_t, lio->lis_io_endpos,
                                               next) - io->u.ci_rw.crw_pos;
                 lio->lis_pos    = io->u.ci_rw.crw_pos;
                 lio->lis_endpos = io->u.ci_rw.crw_pos + io->u.ci_rw.crw_count;
-                CDEBUG(D_VFSTRACE, "stripe: %llu chunk: [%llu, %llu) %llu\n",
-                       (__u64)start, lio->lis_pos, lio->lis_endpos,
-                       (__u64)lio->lis_io_endpos);
-        }
-        /*
-         * XXX The following call should be optimized: we know, that
-         * [lio->lis_pos, lio->lis_endpos) intersects with exactly one stripe.
-         */
-        RETURN(lov_io_iter_init(env, ios));
+		CDEBUG(D_VFSTRACE, "stripe: "LPU64" chunk: ["LPU64", "LPU64") "
+		       LPU64"\n", (__u64)start, lio->lis_pos, lio->lis_endpos,
+		       (__u64)lio->lis_io_endpos);
+	}
+	/*
+	 * XXX The following call should be optimized: we know, that
+	 * [lio->lis_pos, lio->lis_endpos) intersects with exactly one stripe.
+	 */
+	RETURN(lov_io_iter_init(env, ios));
 }
 
 static int lov_io_call(const struct lu_env *env, struct lov_io *lio,
                        int (*iofunc)(const struct lu_env *, struct cl_io *))
 {
+	struct cl_io *parent = lio->lis_cl.cis_io;
         struct lov_io_sub *sub;
         int rc = 0;
 
         ENTRY;
-        list_for_each_entry(sub, &lio->lis_active, sub_linkage) {
+        cfs_list_for_each_entry(sub, &lio->lis_active, sub_linkage) {
                 lov_sub_enter(sub);
                 rc = iofunc(sub->sub_env, sub->sub_io);
                 lov_sub_exit(sub);
                 if (rc)
                         break;
-        }
-        RETURN(rc);
+
+		if (parent->ci_result == 0)
+			parent->ci_result = sub->sub_io->ci_result;
+	}
+	RETURN(rc);
 }
 
 static int lov_io_lock(const struct lu_env *env, const struct cl_io_slice *ios)
@@ -510,8 +538,8 @@ static void lov_io_iter_fini(const struct lu_env *env,
         ENTRY;
         rc = lov_io_call(env, lio, lov_io_iter_fini_wrapper);
         LASSERT(rc == 0);
-        while (!list_empty(&lio->lis_active))
-                list_del_init(lio->lis_active.next);
+        while (!cfs_list_empty(&lio->lis_active))
+                cfs_list_del_init(lio->lis_active.next);
         EXIT;
 }
 
@@ -551,7 +579,7 @@ static struct cl_page_list *lov_io_submit_qin(struct lov_device *ld,
  */
 static int lov_io_submit(const struct lu_env *env,
                          const struct cl_io_slice *ios,
-                         enum cl_req_type crt, struct cl_2queue *queue)
+			 enum cl_req_type crt, struct cl_2queue *queue)
 {
         struct lov_io          *lio = cl2lov_io(env, ios);
         struct lov_object      *obj = lio->lis_object;
@@ -581,15 +609,16 @@ static int lov_io_submit(const struct lu_env *env,
                 sub = lov_sub_get(env, lio, idx);
                 LASSERT(!IS_ERR(sub));
                 LASSERT(sub->sub_io == &lio->lis_single_subio);
-                rc = cl_io_submit_rw(sub->sub_env, sub->sub_io, crt, queue);
+                rc = cl_io_submit_rw(sub->sub_env, sub->sub_io,
+				     crt, queue);
                 lov_sub_put(sub);
                 RETURN(rc);
         }
 
         LASSERT(lio->lis_subs != NULL);
         if (alloc) {
-                OBD_ALLOC(stripes_qin,
-                          sizeof(*stripes_qin) * lio->lis_nr_subios);
+                OBD_ALLOC_LARGE(stripes_qin,
+                                sizeof(*stripes_qin) * lio->lis_nr_subios);
                 if (stripes_qin == NULL)
                         RETURN(-ENOMEM);
 
@@ -601,7 +630,7 @@ static int lov_io_submit(const struct lu_env *env,
                  * In order to not make things worse, even don't try to
                  * allocate the memory with __GFP_NOWARN. -jay
                  */
-                mutex_lock(&ld->ld_mutex);
+		mutex_lock(&ld->ld_mutex);
                 lio->lis_mem_frozen = 1;
         }
 
@@ -615,14 +644,14 @@ static int lov_io_submit(const struct lu_env *env,
                 struct lov_io_sub   *sub;
                 struct cl_page_list *sub_qin = QIN(stripe);
 
-                if (list_empty(&sub_qin->pl_pages))
+                if (cfs_list_empty(&sub_qin->pl_pages))
                         continue;
 
                 cl_page_list_splice(sub_qin, &cl2q->c2_qin);
                 sub = lov_sub_get(env, lio, stripe);
                 if (!IS_ERR(sub)) {
                         rc = cl_io_submit_rw(sub->sub_env, sub->sub_io,
-                                             crt, cl2q);
+					     crt, cl2q);
                         lov_sub_put(sub);
                 } else
                         rc = PTR_ERR(sub);
@@ -635,14 +664,14 @@ static int lov_io_submit(const struct lu_env *env,
         for (stripe = 0; stripe < lio->lis_nr_subios; stripe++) {
                 struct cl_page_list *sub_qin = QIN(stripe);
 
-                if (list_empty(&sub_qin->pl_pages))
+                if (cfs_list_empty(&sub_qin->pl_pages))
                         continue;
 
                 cl_page_list_splice(sub_qin, qin);
         }
 
         if (alloc) {
-                OBD_FREE(stripes_qin,
+                OBD_FREE_LARGE(stripes_qin,
                          sizeof(*stripes_qin) * lio->lis_nr_subios);
         } else {
                 int i;
@@ -654,7 +683,7 @@ static int lov_io_submit(const struct lu_env *env,
                                 lov_io_sub_fini(env, lio, &lio->lis_subs[i]);
                 }
                 lio->lis_mem_frozen = 0;
-                mutex_unlock(&ld->ld_mutex);
+		mutex_unlock(&ld->ld_mutex);
         }
 
         RETURN(rc);
@@ -719,6 +748,28 @@ static int lov_io_fault_start(const struct lu_env *env,
         RETURN(lov_io_start(env, ios));
 }
 
+static void lov_io_fsync_end(const struct lu_env *env,
+			     const struct cl_io_slice *ios)
+{
+	struct lov_io *lio = cl2lov_io(env, ios);
+	struct lov_io_sub *sub;
+	unsigned int *written = &ios->cis_io->u.ci_fsync.fi_nr_written;
+	ENTRY;
+
+	*written = 0;
+	cfs_list_for_each_entry(sub, &lio->lis_active, sub_linkage) {
+		struct cl_io *subio = sub->sub_io;
+
+		lov_sub_enter(sub);
+		lov_io_end_wrapper(sub->sub_env, subio);
+		lov_sub_exit(sub);
+
+		if (subio->ci_result == 0)
+			*written += subio->u.ci_fsync.fi_nr_written;
+	}
+	RETURN_EXIT;
+}
+
 static const struct cl_io_operations lov_io_ops = {
         .op = {
                 [CIT_READ] = {
@@ -739,7 +790,7 @@ static const struct cl_io_operations lov_io_ops = {
                         .cio_start     = lov_io_start,
                         .cio_end       = lov_io_end
                 },
-                [CIT_TRUNC] = {
+                [CIT_SETATTR] = {
                         .cio_fini      = lov_io_fini,
                         .cio_iter_init = lov_io_iter_init,
                         .cio_iter_fini = lov_io_iter_fini,
@@ -757,6 +808,15 @@ static const struct cl_io_operations lov_io_ops = {
                         .cio_start     = lov_io_fault_start,
                         .cio_end       = lov_io_end
                 },
+		[CIT_FSYNC] = {
+			.cio_fini      = lov_io_fini,
+			.cio_iter_init = lov_io_iter_init,
+			.cio_iter_fini = lov_io_iter_fini,
+			.cio_lock      = lov_io_lock,
+			.cio_unlock    = lov_io_unlock,
+			.cio_start     = lov_io_start,
+			.cio_end       = lov_io_fsync_end
+		},
                 [CIT_MISC] = {
                         .cio_fini   = lov_io_fini
                 }
@@ -782,7 +842,11 @@ static const struct cl_io_operations lov_io_ops = {
 static void lov_empty_io_fini(const struct lu_env *env,
                               const struct cl_io_slice *ios)
 {
+	struct lov_object *lov = cl2lov(ios->cis_obj);
         ENTRY;
+
+	if (cfs_atomic_dec_and_test(&lov->lo_active_ios))
+		cfs_waitq_broadcast(&lov->lo_waitq);
         EXIT;
 }
 
@@ -800,8 +864,8 @@ static void lov_empty_impossible(const struct lu_env *env,
 static const struct cl_io_operations lov_empty_io_ops = {
         .op = {
                 [CIT_READ] = {
-#if 0
                         .cio_fini       = lov_empty_io_fini,
+#if 0
                         .cio_iter_init  = LOV_EMPTY_IMPOSSIBLE,
                         .cio_lock       = LOV_EMPTY_IMPOSSIBLE,
                         .cio_start      = LOV_EMPTY_IMPOSSIBLE,
@@ -815,7 +879,7 @@ static const struct cl_io_operations lov_empty_io_ops = {
                         .cio_start     = LOV_EMPTY_IMPOSSIBLE,
                         .cio_end       = LOV_EMPTY_IMPOSSIBLE
                 },
-                [CIT_TRUNC] = {
+                [CIT_SETATTR] = {
                         .cio_fini      = lov_empty_io_fini,
                         .cio_iter_init = LOV_EMPTY_IMPOSSIBLE,
                         .cio_lock      = LOV_EMPTY_IMPOSSIBLE,
@@ -829,6 +893,9 @@ static const struct cl_io_operations lov_empty_io_ops = {
                         .cio_start     = LOV_EMPTY_IMPOSSIBLE,
                         .cio_end       = LOV_EMPTY_IMPOSSIBLE
                 },
+		[CIT_FSYNC] = {
+			.cio_fini   = lov_empty_io_fini
+		},
                 [CIT_MISC] = {
                         .cio_fini   = lov_empty_io_fini
                 }
@@ -854,10 +921,11 @@ int lov_io_init_raid0(const struct lu_env *env, struct cl_object *obj,
         CFS_INIT_LIST_HEAD(&lio->lis_active);
         lov_io_slice_init(lio, lov, io);
         if (io->ci_result == 0) {
-                LASSERT(lov_r0(lov)->lo_lsm != NULL);
                 io->ci_result = lov_io_subio_init(env, lio, io);
-                if (io->ci_result == 0)
+                if (io->ci_result == 0) {
                         cl_io_slice_add(io, &lio->lis_cl, obj, &lov_io_ops);
+			cfs_atomic_inc(&lov->lo_active_ios);
+		}
         }
         RETURN(io->ci_result);
 }
@@ -865,19 +933,24 @@ int lov_io_init_raid0(const struct lu_env *env, struct cl_object *obj,
 int lov_io_init_empty(const struct lu_env *env, struct cl_object *obj,
                       struct cl_io *io)
 {
-        struct lov_io *lio = lov_env_io(env);
-        int result;
-
-        ENTRY;
-        switch (io->ci_type) {
-        default:
-                LBUG();
-        case CIT_MISC:
-        case CIT_READ:
-                result = 0;
-                break;
+	struct lov_object *lov = cl2lov(obj);
+	struct lov_io *lio = lov_env_io(env);
+	int result;
+	ENTRY;
+
+	lio->lis_object = lov;
+	switch (io->ci_type) {
+	default:
+		LBUG();
+	case CIT_MISC:
+	case CIT_READ:
+		result = 0;
+		break;
+	case CIT_FSYNC:
+	case CIT_SETATTR:
+		result = +1;
+		break;
         case CIT_WRITE:
-        case CIT_TRUNC:
                 result = -EBADF;
                 break;
         case CIT_FAULT:
@@ -886,10 +959,47 @@ int lov_io_init_empty(const struct lu_env *env, struct cl_object *obj,
                        PFID(lu_object_fid(&obj->co_lu)));
                 break;
         }
-        if (result == 0)
+        if (result == 0) {
                 cl_io_slice_add(io, &lio->lis_cl, obj, &lov_empty_io_ops);
-        io->ci_result = result;
-        RETURN(result != 0);
+		cfs_atomic_inc(&lov->lo_active_ios);
+	}
+
+	io->ci_result = result < 0 ? result : 0;
+	RETURN(result != 0);
+}
+
+int lov_io_init_released(const struct lu_env *env, struct cl_object *obj,
+			struct cl_io *io)
+{
+	struct lov_object *lov = cl2lov(obj);
+	struct lov_io *lio = lov_env_io(env);
+	int result;
+	ENTRY;
+
+	LASSERT(lov->lo_lsm != NULL);
+	lio->lis_object = lov;
+
+	switch (io->ci_type) {
+	default:
+		LASSERTF(0, "invalid type %d\n", io->ci_type);
+	case CIT_MISC:
+	case CIT_FSYNC:
+		result = +1;
+		break;
+	case CIT_SETATTR:
+	case CIT_READ:
+	case CIT_WRITE:
+	case CIT_FAULT:
+		/* TODO: need to restore the file. */
+		result = -EBADF;
+		break;
+	}
+	if (result == 0) {
+		cl_io_slice_add(io, &lio->lis_cl, obj, &lov_empty_io_ops);
+		cfs_atomic_inc(&lov->lo_active_ios);
+	}
+
+	io->ci_result = result < 0 ? result : 0;
+	RETURN(result != 0);
 }
-
 /** @} lov */