LU-1030 clio: reimplement ll_fsync in clio way

[fs/lustre-release.git] / lustre / obdclass / cl_io.c

diff --git a/lustre/obdclass/cl_io.c b/lustre/obdclass/cl_io.c
index 64ef712..4ac5087 100644 (file)
--- a/lustre/obdclass/cl_io.c
+++ b/lustre/obdclass/cl_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.
@@ -28,6 +26,8 @@
 /*
  * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
+ *
+ * Copyright (c) 2011, 2012, Whamcloud, Inc.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -39,9 +39,6 @@
  */
 
 #define DEBUG_SUBSYSTEM S_CLASS
-#ifndef EXPORT_SYMTAB
-# define EXPORT_SYMTAB
-#endif
 
 #include <obd_class.h>
 #include <obd_support.h>
@@ -233,13 +230,41 @@ cl_lock_descr_fid(const struct cl_lock_descr *descr)
         return lu_object_fid(&descr->cld_obj->co_lu);
 }
 
-static int cl_lock_descr_cmp(const struct cl_lock_descr *d0,
-                             const struct cl_lock_descr *d1)
+static int cl_lock_descr_sort(const struct cl_lock_descr *d0,
+                              const struct cl_lock_descr *d1)
 {
         return lu_fid_cmp(cl_lock_descr_fid(d0), cl_lock_descr_fid(d1)) ?:
                 __diff_normalize(d0->cld_start, d1->cld_start);
 }
 
+static int cl_lock_descr_cmp(const struct cl_lock_descr *d0,
+                             const struct cl_lock_descr *d1)
+{
+        int ret;
+
+        ret = lu_fid_cmp(cl_lock_descr_fid(d0), cl_lock_descr_fid(d1));
+        if (ret)
+                return ret;
+        if (d0->cld_end < d1->cld_start)
+                return -1;
+        if (d0->cld_start > d0->cld_end)
+                return 1;
+        return 0;
+}
+
+static void cl_lock_descr_merge(struct cl_lock_descr *d0,
+                                const struct cl_lock_descr *d1)
+{
+        d0->cld_start = min(d0->cld_start, d1->cld_start);
+        d0->cld_end = max(d0->cld_end, d1->cld_end);
+
+        if (d1->cld_mode == CLM_WRITE && d0->cld_mode != CLM_WRITE)
+                d0->cld_mode = CLM_WRITE;
+
+        if (d1->cld_mode == CLM_GROUP && d0->cld_mode != CLM_GROUP)
+                d0->cld_mode = CLM_GROUP;
+}
+
 /*
  * Sort locks in lexicographical order of their (fid, start-offset) pairs.
  */
@@ -261,7 +286,7 @@ static void cl_io_locks_sort(struct cl_io *io)
                                              &io->ci_lockset.cls_todo,
                                              cill_linkage) {
                         if (prev != NULL) {
-                                switch (cl_lock_descr_cmp(&prev->cill_descr,
+                                switch (cl_lock_descr_sort(&prev->cill_descr,
                                                           &curr->cill_descr)) {
                                 case 0:
                                         /*
@@ -303,16 +328,41 @@ int cl_queue_match(const cfs_list_t *queue,
                if (cl_lock_descr_match(&scan->cill_descr, need))
                        RETURN(+1);
        }
-       return 0;
+       RETURN(0);
 }
 EXPORT_SYMBOL(cl_queue_match);
 
+static int cl_queue_merge(const cfs_list_t *queue,
+                          const struct cl_lock_descr *need)
+{
+       struct cl_io_lock_link *scan;
+
+       ENTRY;
+       cfs_list_for_each_entry(scan, queue, cill_linkage) {
+               if (cl_lock_descr_cmp(&scan->cill_descr, need))
+                       continue;
+               cl_lock_descr_merge(&scan->cill_descr, need);
+               CDEBUG(D_VFSTRACE, "lock: %d: [%lu, %lu]\n",
+                      scan->cill_descr.cld_mode, scan->cill_descr.cld_start,
+                      scan->cill_descr.cld_end);
+               RETURN(+1);
+       }
+       RETURN(0);
+
+}
+
 static int cl_lockset_match(const struct cl_lockset *set,
-                            const struct cl_lock_descr *need, int all_queues)
+                            const struct cl_lock_descr *need)
+{
+        return cl_queue_match(&set->cls_curr, need) ||
+               cl_queue_match(&set->cls_done, need);
+}
+
+static int cl_lockset_merge(const struct cl_lockset *set,
+                            const struct cl_lock_descr *need)
 {
-        return (all_queues ? cl_queue_match(&set->cls_todo, need) : 0) ||
-                cl_queue_match(&set->cls_curr, need) ||
-                cl_queue_match(&set->cls_done, need);
+        return cl_queue_merge(&set->cls_todo, need) ||
+               cl_lockset_match(set, need);
 }
 
 static int cl_lockset_lock_one(const struct lu_env *env,
@@ -324,7 +374,13 @@ static int cl_lockset_lock_one(const struct lu_env *env,
 
         ENTRY;
 
-        lock = cl_lock_request(env, io, &link->cill_descr, "io", io);
+        if (io->ci_lockreq == CILR_PEEK) {
+                lock = cl_lock_peek(env, io, &link->cill_descr, "io", io);
+                if (lock == NULL)
+                        lock = ERR_PTR(-ENODATA);
+        } else
+                lock = cl_lock_request(env, io, &link->cill_descr, "io", io);
+
         if (!IS_ERR(lock)) {
                 link->cill_lock = lock;
                 cfs_list_move(&link->cill_linkage, &set->cls_curr);
@@ -367,7 +423,7 @@ static int cl_lockset_lock(const struct lu_env *env, struct cl_io *io,
         ENTRY;
         result = 0;
         cfs_list_for_each_entry_safe(link, temp, &set->cls_todo, cill_linkage) {
-                if (!cl_lockset_match(set, &link->cill_descr, 0)) {
+                if (!cl_lockset_match(set, &link->cill_descr)) {
                         /* XXX some locking to guarantee that locks aren't
                          * expanded in between. */
                         result = cl_lockset_lock_one(env, io, set, link);
@@ -555,7 +611,7 @@ int cl_io_lock_add(const struct lu_env *env, struct cl_io *io,
         int result;
 
         ENTRY;
-        if (cl_lockset_match(&io->ci_lockset, &link->cill_descr, 1))
+        if (cl_lockset_merge(&io->ci_lockset, &link->cill_descr))
                 result = +1;
         else {
                 cfs_list_add(&link->cill_linkage, &io->ci_lockset.cls_todo);
@@ -977,7 +1033,9 @@ int cl_io_loop(const struct lu_env *env, struct cl_io *io)
                 }
                 cl_io_iter_fini(env, io);
         } while (result == 0 && io->ci_continue);
-        RETURN(result < 0 ? result : 0);
+       if (result == 0)
+               result = io->ci_result;
+       RETURN(result < 0 ? result : 0);
 }
 EXPORT_SYMBOL(cl_io_loop);
 
@@ -1490,10 +1548,12 @@ void cl_req_page_add(const struct lu_env *env,
         ENTRY;
         page = cl_page_top(page);
 
-        LINVRNT(cl_page_is_vmlocked(env, page));
         LASSERT(cfs_list_empty(&page->cp_flight));
         LASSERT(page->cp_req == NULL);
 
+        CL_PAGE_DEBUG(D_PAGE, env, page, "req %p, %d, %u\n",
+                      req, req->crq_type, req->crq_nrpages);
+
         cfs_list_add_tail(&page->cp_flight, &req->crq_pages);
         ++req->crq_nrpages;
         page->cp_req = req;
@@ -1522,7 +1582,6 @@ void cl_req_page_done(const struct lu_env *env, struct cl_page *page)
         ENTRY;
         page = cl_page_top(page);
 
-        LINVRNT(cl_page_is_vmlocked(env, page));
         LASSERT(!cfs_list_empty(&page->cp_flight));
         LASSERT(req->crq_nrpages > 0);