Whamcloud - gitweb
b=18721
[fs/lustre-release.git] / lustre / lclient / lcommon_cl.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * GPL HEADER START
5  *
6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
7  *
8  * This program is free software; you can redistribute it and/or modify
9  * it under the terms of the GNU General Public License version 2 only,
10  * as published by the Free Software Foundation.
11  *
12  * This program is distributed in the hope that it will be useful, but
13  * WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
15  * General Public License version 2 for more details (a copy is included
16  * in the LICENSE file that accompanied this code).
17  *
18  * You should have received a copy of the GNU General Public License
19  * version 2 along with this program; If not, see
20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
21  *
22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
23  * CA 95054 USA or visit www.sun.com if you need additional information or
24  * have any questions.
25  *
26  * GPL HEADER END
27  */
28 /*
29  * Copyright 2008 Sun Microsystems, Inc.  All rights reserved.
30  * Use is subject to license terms.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * cl code shared between vvp and liblustre (and other Lustre clients in the
37  * future).
38  *
39  *   Author: Nikita Danilov <nikita.danilov@sun.com>
40  */
41
42 #define DEBUG_SUBSYSTEM S_LLITE
43
44 #ifdef __KERNEL__
45 # include <libcfs/libcfs.h>
46 # include <linux/fs.h>
47 # include <linux/sched.h>
48 # include <linux/mm.h>
49 # include <linux/smp_lock.h>
50 # include <linux/quotaops.h>
51 # include <linux/highmem.h>
52 # include <linux/pagemap.h>
53 # include <linux/rbtree.h>
54 #else /* __KERNEL__ */
55 #include <stdlib.h>
56 #include <string.h>
57 #include <assert.h>
58 #include <time.h>
59 #include <sys/types.h>
60 #include <sys/stat.h>
61 #include <sys/queue.h>
62 #include <fcntl.h>
63 # include <sysio.h>
64 # ifdef HAVE_XTIO_H
65 #  include <xtio.h>
66 # endif
67 # include <fs.h>
68 # include <mount.h>
69 # include <inode.h>
70 # ifdef HAVE_FILE_H
71 #  include <file.h>
72 # endif
73 # include <liblustre.h>
74 #endif
75
76 #include <obd.h>
77 #include <obd_support.h>
78 #include <lustre_fid.h>
79 #include <lustre_lite.h>
80 #include <lustre_dlm.h>
81 #include <lustre_ver.h>
82 #include <lustre_mdc.h>
83 #include <cl_object.h>
84
85 #include <lclient.h>
86
87 #ifdef __KERNEL__
88 #include "../llite/llite_internal.h"
89 #else
90 #include "../liblustre/llite_lib.h"
91 #endif
92
93 const struct cl_req_operations ccc_req_ops;
94
95 /*
96  * ccc_ prefix stands for "Common Client Code".
97  */
98
99 static cfs_mem_cache_t *ccc_lock_kmem;
100 static cfs_mem_cache_t *ccc_object_kmem;
101 static cfs_mem_cache_t *ccc_thread_kmem;
102 static cfs_mem_cache_t *ccc_session_kmem;
103 static cfs_mem_cache_t *ccc_req_kmem;
104
105 static struct lu_kmem_descr ccc_caches[] = {
106         {
107                 .ckd_cache = &ccc_lock_kmem,
108                 .ckd_name  = "ccc_lock_kmem",
109                 .ckd_size  = sizeof (struct ccc_lock)
110         },
111         {
112                 .ckd_cache = &ccc_object_kmem,
113                 .ckd_name  = "ccc_object_kmem",
114                 .ckd_size  = sizeof (struct ccc_object)
115         },
116         {
117                 .ckd_cache = &ccc_thread_kmem,
118                 .ckd_name  = "ccc_thread_kmem",
119                 .ckd_size  = sizeof (struct ccc_thread_info),
120         },
121         {
122                 .ckd_cache = &ccc_session_kmem,
123                 .ckd_name  = "ccc_session_kmem",
124                 .ckd_size  = sizeof (struct ccc_session)
125         },
126         {
127                 .ckd_cache = &ccc_req_kmem,
128                 .ckd_name  = "ccc_req_kmem",
129                 .ckd_size  = sizeof (struct ccc_req)
130         },
131         {
132                 .ckd_cache = NULL
133         }
134 };
135
136 /*****************************************************************************
137  *
138  * Vvp device and device type functions.
139  *
140  */
141
142 void *ccc_key_init(const struct lu_context *ctx,
143                           struct lu_context_key *key)
144 {
145         struct ccc_thread_info *info;
146
147         OBD_SLAB_ALLOC_PTR_GFP(info, ccc_thread_kmem, CFS_ALLOC_IO);
148         if (info == NULL)
149                 info = ERR_PTR(-ENOMEM);
150         return info;
151 }
152
153 void ccc_key_fini(const struct lu_context *ctx,
154                          struct lu_context_key *key, void *data)
155 {
156         struct ccc_thread_info *info = data;
157         OBD_SLAB_FREE_PTR(info, ccc_thread_kmem);
158 }
159
160 void *ccc_session_key_init(const struct lu_context *ctx,
161                                   struct lu_context_key *key)
162 {
163         struct ccc_session *session;
164
165         OBD_SLAB_ALLOC_PTR_GFP(session, ccc_session_kmem, CFS_ALLOC_IO);
166         if (session == NULL)
167                 session = ERR_PTR(-ENOMEM);
168         return session;
169 }
170
171 void ccc_session_key_fini(const struct lu_context *ctx,
172                                  struct lu_context_key *key, void *data)
173 {
174         struct ccc_session *session = data;
175         OBD_SLAB_FREE_PTR(session, ccc_session_kmem);
176 }
177
178 struct lu_context_key ccc_key = {
179         .lct_tags = LCT_CL_THREAD,
180         .lct_init = ccc_key_init,
181         .lct_fini = ccc_key_fini
182 };
183
184 struct lu_context_key ccc_session_key = {
185         .lct_tags = LCT_SESSION,
186         .lct_init = ccc_session_key_init,
187         .lct_fini = ccc_session_key_fini
188 };
189
190
191 /* type constructor/destructor: ccc_type_{init,fini,start,stop}(). */
192 // LU_TYPE_INIT_FINI(ccc, &ccc_key, &ccc_session_key);
193
194 int ccc_device_init(const struct lu_env *env, struct lu_device *d,
195                            const char *name, struct lu_device *next)
196 {
197         struct ccc_device  *vdv;
198         int rc;
199         ENTRY;
200
201         vdv = lu2ccc_dev(d);
202         vdv->cdv_next = lu2cl_dev(next);
203
204         LASSERT(d->ld_site != NULL && next->ld_type != NULL);
205         next->ld_site = d->ld_site;
206         rc = next->ld_type->ldt_ops->ldto_device_init(
207                         env, next, next->ld_type->ldt_name, NULL);
208         if (rc == 0) {
209                 lu_device_get(next);
210                 lu_ref_add(&next->ld_reference, "lu-stack", &lu_site_init);
211         }
212         RETURN(rc);
213 }
214
215 struct lu_device *ccc_device_fini(const struct lu_env *env,
216                                          struct lu_device *d)
217 {
218         return cl2lu_dev(lu2ccc_dev(d)->cdv_next);
219 }
220
221 struct lu_device *ccc_device_alloc(const struct lu_env *env,
222                                    struct lu_device_type *t,
223                                    struct lustre_cfg *cfg,
224                                    const struct lu_device_operations *luops,
225                                    const struct cl_device_operations *clops)
226 {
227         struct ccc_device *vdv;
228         struct lu_device  *lud;
229         struct cl_site    *site;
230         int rc;
231         ENTRY;
232
233         OBD_ALLOC_PTR(vdv);
234         if (vdv == NULL)
235                 RETURN(ERR_PTR(-ENOMEM));
236
237         lud = &vdv->cdv_cl.cd_lu_dev;
238         cl_device_init(&vdv->cdv_cl, t);
239         ccc2lu_dev(vdv)->ld_ops = luops;
240         vdv->cdv_cl.cd_ops = clops;
241
242         OBD_ALLOC_PTR(site);
243         if (site != NULL) {
244                 rc = cl_site_init(site, &vdv->cdv_cl);
245                 if (rc == 0)
246                         rc = lu_site_init_finish(&site->cs_lu);
247                 else {
248                         LASSERT(lud->ld_site == NULL);
249                         CERROR("Cannot init lu_site, rc %d.\n", rc);
250                         OBD_FREE_PTR(site);
251                 }
252         } else
253                 rc = -ENOMEM;
254         if (rc != 0) {
255                 ccc_device_free(env, lud);
256                 lud = ERR_PTR(rc);
257         }
258         RETURN(lud);
259 }
260
261 struct lu_device *ccc_device_free(const struct lu_env *env,
262                                          struct lu_device *d)
263 {
264         struct ccc_device *vdv  = lu2ccc_dev(d);
265         struct cl_site    *site = lu2cl_site(d->ld_site);
266         struct lu_device  *next = cl2lu_dev(vdv->cdv_next);
267
268         if (d->ld_site != NULL) {
269                 cl_site_fini(site);
270                 OBD_FREE_PTR(site);
271         }
272         cl_device_fini(lu2cl_dev(d));
273         OBD_FREE_PTR(vdv);
274         return next;
275 }
276
277 int ccc_req_init(const struct lu_env *env, struct cl_device *dev,
278                         struct cl_req *req)
279 {
280         struct ccc_req *vrq;
281         int result;
282
283         OBD_SLAB_ALLOC_PTR_GFP(vrq, ccc_req_kmem, CFS_ALLOC_IO);
284         if (vrq != NULL) {
285                 cl_req_slice_add(req, &vrq->crq_cl, dev, &ccc_req_ops);
286                 result = 0;
287         } else
288                 result = -ENOMEM;
289         return result;
290 }
291
292 /**
293  * An `emergency' environment used by ccc_inode_fini() when cl_env_get()
294  * fails. Access to this environment is serialized by ccc_inode_fini_guard
295  * mutex.
296  */
297 static struct lu_env *ccc_inode_fini_env = NULL;
298
299 /**
300  * A mutex serializing calls to slp_inode_fini() under extreme memory
301  * pressure, when environments cannot be allocated.
302  */
303 static DEFINE_MUTEX(ccc_inode_fini_guard);
304 static int dummy_refcheck;
305
306 int ccc_global_init(struct lu_device_type *device_type)
307 {
308         int result;
309
310         result = lu_kmem_init(ccc_caches);
311         if (result == 0) {
312                 result = lu_device_type_init(device_type);
313                 ccc_inode_fini_env = cl_env_alloc(&dummy_refcheck,
314                                                   LCT_REMEMBER|LCT_NOREF);
315                 if (IS_ERR(ccc_inode_fini_env))
316                         result = PTR_ERR(ccc_inode_fini_env);
317                 else
318                         ccc_inode_fini_env->le_ctx.lc_cookie = 0x4;
319         }
320         return result;
321 }
322
323 void ccc_global_fini(struct lu_device_type *device_type)
324 {
325         if (ccc_inode_fini_env != NULL) {
326                 cl_env_put(ccc_inode_fini_env, &dummy_refcheck);
327                 ccc_inode_fini_env = NULL;
328         }
329         lu_device_type_fini(device_type);
330         lu_kmem_fini(ccc_caches);
331 }
332
333 /*****************************************************************************
334  *
335  * Object operations.
336  *
337  */
338
339 struct lu_object *ccc_object_alloc(const struct lu_env *env,
340                                    const struct lu_object_header *_,
341                                    struct lu_device *dev,
342                                    const struct cl_object_operations *clops,
343                                    const struct lu_object_operations *luops)
344 {
345         struct ccc_object *vob;
346         struct lu_object  *obj;
347
348         OBD_SLAB_ALLOC_PTR_GFP(vob, ccc_object_kmem, CFS_ALLOC_IO);
349         if (vob != NULL) {
350                 struct cl_object_header *hdr;
351
352                 obj = ccc2lu(vob);
353                 hdr = &vob->cob_header;
354                 cl_object_header_init(hdr);
355                 lu_object_init(obj, &hdr->coh_lu, dev);
356                 lu_object_add_top(&hdr->coh_lu, obj);
357
358                 vob->cob_cl.co_ops = clops;
359                 obj->lo_ops = luops;
360         } else
361                 obj = NULL;
362         return obj;
363 }
364
365 int ccc_object_init0(const struct lu_env *env,
366                             struct ccc_object *vob,
367                             const struct cl_object_conf *conf)
368 {
369         vob->cob_inode = conf->coc_inode;
370         vob->cob_transient_pages = 0;
371         return 0;
372 }
373
374 int ccc_object_init(const struct lu_env *env, struct lu_object *obj,
375                            const struct lu_object_conf *conf)
376 {
377         struct ccc_device *dev = lu2ccc_dev(obj->lo_dev);
378         struct ccc_object *vob = lu2ccc(obj);
379         struct lu_object  *below;
380         struct lu_device  *under;
381         int result;
382
383         under = &dev->cdv_next->cd_lu_dev;
384         below = under->ld_ops->ldo_object_alloc(env, obj->lo_header, under);
385         if (below != NULL) {
386                 const struct cl_object_conf *cconf;
387
388                 cconf = lu2cl_conf(conf);
389                 CFS_INIT_LIST_HEAD(&vob->cob_pending_list);
390                 lu_object_add(obj, below);
391                 result = ccc_object_init0(env, vob, cconf);
392         } else
393                 result = -ENOMEM;
394         return result;
395 }
396
397 void ccc_object_free(const struct lu_env *env, struct lu_object *obj)
398 {
399         struct ccc_object *vob = lu2ccc(obj);
400
401         lu_object_fini(obj);
402         lu_object_header_fini(obj->lo_header);
403         OBD_SLAB_FREE_PTR(vob, ccc_object_kmem);
404 }
405
406 int ccc_lock_init(const struct lu_env *env,
407                   struct cl_object *obj, struct cl_lock *lock,
408                   const struct cl_io *_,
409                   const struct cl_lock_operations *lkops)
410 {
411         struct ccc_lock *clk;
412         int result;
413
414         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
415
416         OBD_SLAB_ALLOC_PTR_GFP(clk, ccc_lock_kmem, CFS_ALLOC_IO);
417         if (clk != NULL) {
418                 cl_lock_slice_add(lock, &clk->clk_cl, obj, lkops);
419                 result = 0;
420         } else
421                 result = -ENOMEM;
422         return result;
423 }
424
425 int ccc_attr_set(const struct lu_env *env, struct cl_object *obj,
426                  const struct cl_attr *attr, unsigned valid)
427 {
428         return 0;
429 }
430
431 int ccc_object_glimpse(const struct lu_env *env,
432                        const struct cl_object *obj, struct ost_lvb *lvb)
433 {
434         struct inode *inode = ccc_object_inode(obj);
435
436         ENTRY;
437         lvb->lvb_mtime = cl_inode_mtime(inode);
438         lvb->lvb_atime = cl_inode_atime(inode);
439         lvb->lvb_ctime = cl_inode_ctime(inode);
440         RETURN(0);
441 }
442
443
444
445 int ccc_conf_set(const struct lu_env *env, struct cl_object *obj,
446                         const struct cl_object_conf *conf)
447 {
448         /* TODO: destroy all pages attached to this object. */
449         return 0;
450 }
451
452 /*****************************************************************************
453  *
454  * Page operations.
455  *
456  */
457
458 cfs_page_t *ccc_page_vmpage(const struct lu_env *env,
459                             const struct cl_page_slice *slice)
460 {
461         return cl2vm_page(slice);
462 }
463
464 int ccc_page_is_under_lock(const struct lu_env *env,
465                            const struct cl_page_slice *slice,
466                            struct cl_io *io)
467 {
468         struct ccc_io        *cio  = ccc_env_io(env);
469         struct cl_lock_descr *desc = &ccc_env_info(env)->cti_descr;
470         struct cl_page       *page = slice->cpl_page;
471
472         int result;
473
474         ENTRY;
475
476         if (io->ci_type == CIT_READ || io->ci_type == CIT_WRITE ||
477             io->ci_type == CIT_FAULT) {
478                 if (cio->cui_fd->fd_flags & LL_FILE_GROUP_LOCKED)
479                         result = -EBUSY;
480                 else {
481                         desc->cld_start = page->cp_index;
482                         desc->cld_end   = page->cp_index;
483                         desc->cld_obj   = page->cp_obj;
484                         desc->cld_mode  = CLM_READ;
485                         result = cl_queue_match(&io->ci_lockset.cls_done,
486                                                 desc) ? -EBUSY : 0;
487                 }
488         } else
489                 result = 0;
490         RETURN(result);
491 }
492
493 int ccc_fail(const struct lu_env *env, const struct cl_page_slice *slice)
494 {
495         /*
496          * Cached read?
497          */
498         LBUG();
499         return 0;
500 }
501
502 void ccc_transient_page_verify(const struct cl_page *page)
503 {
504 }
505
506 void ccc_transient_page_own(const struct lu_env *env,
507                                    const struct cl_page_slice *slice,
508                                    struct cl_io *_)
509 {
510         ccc_transient_page_verify(slice->cpl_page);
511 }
512
513 void ccc_transient_page_assume(const struct lu_env *env,
514                                       const struct cl_page_slice *slice,
515                                       struct cl_io *_)
516 {
517         ccc_transient_page_verify(slice->cpl_page);
518 }
519
520 void ccc_transient_page_unassume(const struct lu_env *env,
521                                         const struct cl_page_slice *slice,
522                                         struct cl_io *_)
523 {
524         ccc_transient_page_verify(slice->cpl_page);
525 }
526
527 void ccc_transient_page_disown(const struct lu_env *env,
528                                       const struct cl_page_slice *slice,
529                                       struct cl_io *_)
530 {
531         ccc_transient_page_verify(slice->cpl_page);
532 }
533
534 void ccc_transient_page_discard(const struct lu_env *env,
535                                        const struct cl_page_slice *slice,
536                                        struct cl_io *_)
537 {
538         struct cl_page *page = slice->cpl_page;
539
540         ccc_transient_page_verify(slice->cpl_page);
541
542         /*
543          * For transient pages, remove it from the radix tree.
544          */
545         cl_page_delete(env, page);
546 }
547
548 int ccc_transient_page_prep(const struct lu_env *env,
549                                    const struct cl_page_slice *slice,
550                                    struct cl_io *_)
551 {
552         ENTRY;
553         /* transient page should always be sent. */
554         RETURN(0);
555 }
556
557 /*****************************************************************************
558  *
559  * Lock operations.
560  *
561  */
562
563 void ccc_lock_delete(const struct lu_env *env,
564                      const struct cl_lock_slice *slice)
565 {
566         CLOBINVRNT(env, slice->cls_obj, ccc_object_invariant(slice->cls_obj));
567 }
568
569 void ccc_lock_fini(const struct lu_env *env, struct cl_lock_slice *slice)
570 {
571         struct ccc_lock *clk = cl2ccc_lock(slice);
572         OBD_SLAB_FREE_PTR(clk, ccc_lock_kmem);
573 }
574
575 int ccc_lock_enqueue(const struct lu_env *env,
576                      const struct cl_lock_slice *slice,
577                      struct cl_io *_, __u32 enqflags)
578 {
579         CLOBINVRNT(env, slice->cls_obj, ccc_object_invariant(slice->cls_obj));
580         return 0;
581 }
582
583 int ccc_lock_unuse(const struct lu_env *env, const struct cl_lock_slice *slice)
584 {
585         CLOBINVRNT(env, slice->cls_obj, ccc_object_invariant(slice->cls_obj));
586         return 0;
587 }
588
589 int ccc_lock_wait(const struct lu_env *env, const struct cl_lock_slice *slice)
590 {
591         CLOBINVRNT(env, slice->cls_obj, ccc_object_invariant(slice->cls_obj));
592         return 0;
593 }
594
595 /**
596  * Implementation of cl_lock_operations::clo_fits_into() methods for ccc
597  * layer. This function is executed every time io finds an existing lock in
598  * the lock cache while creating new lock. This function has to decide whether
599  * cached lock "fits" into io.
600  *
601  * \param slice lock to be checked
602  *
603  * \param io    IO that wants a lock.
604  *
605  * \see lov_lock_fits_into().
606  */
607 int ccc_lock_fits_into(const struct lu_env *env,
608                        const struct cl_lock_slice *slice,
609                        const struct cl_lock_descr *need,
610                        const struct cl_io *io)
611 {
612         const struct cl_lock       *lock  = slice->cls_lock;
613         const struct cl_lock_descr *descr = &lock->cll_descr;
614         const struct ccc_io        *cio   = ccc_env_io(env);
615         int                         result;
616
617         ENTRY;
618         /*
619          * Work around DLM peculiarity: it assumes that glimpse
620          * (LDLM_FL_HAS_INTENT) lock is always LCK_PR, and returns reads lock
621          * when asked for LCK_PW lock with LDLM_FL_HAS_INTENT flag set. Make
622          * sure that glimpse doesn't get CLM_WRITE top-lock, so that it
623          * doesn't enqueue CLM_WRITE sub-locks.
624          */
625         if (cio->cui_glimpse)
626                 result = descr->cld_mode != CLM_WRITE;
627         /*
628          * Also, don't match incomplete write locks for read, otherwise read
629          * would enqueue missing sub-locks in the write mode.
630          *
631          * XXX this is a candidate for generic locking policy, to be moved
632          * into cl_lock_lookup().
633          */
634         else if (need->cld_mode != descr->cld_mode)
635                 result = lock->cll_state >= CLS_ENQUEUED;
636         else
637                 result = 1;
638         RETURN(result);
639 }
640
641 /**
642  * Implements cl_lock_operations::clo_state() method for ccc layer, invoked
643  * whenever lock state changes. Transfers object attributes, that might be
644  * updated as a result of lock acquiring into inode.
645  */
646 void ccc_lock_state(const struct lu_env *env,
647                     const struct cl_lock_slice *slice,
648                     enum cl_lock_state state)
649 {
650         struct cl_lock   *lock;
651         struct cl_object *obj;
652         struct inode     *inode;
653         struct cl_attr   *attr;
654
655         ENTRY;
656         lock = slice->cls_lock;
657
658         /*
659          * Refresh inode attributes when the lock is moving into CLS_HELD
660          * state, and only when this is a result of real enqueue, rather than
661          * of finding lock in the cache.
662          */
663         if (state == CLS_HELD && lock->cll_state < CLS_HELD) {
664                 int rc;
665
666                 obj   = slice->cls_obj;
667                 inode = ccc_object_inode(obj);
668                 attr  = &ccc_env_info(env)->cti_attr;
669
670                 /* vmtruncate()->ll_truncate() first sets the i_size and then
671                  * the kms under both a DLM lock and the
672                  * ll_inode_size_lock().  If we don't get the
673                  * ll_inode_size_lock() here we can match the DLM lock and
674                  * reset i_size from the kms before the truncating path has
675                  * updated the kms.  generic_file_write can then trust the
676                  * stale i_size when doing appending writes and effectively
677                  * cancel the result of the truncate.  Getting the
678                  * ll_inode_size_lock() after the enqueue maintains the DLM
679                  * -> ll_inode_size_lock() acquiring order. */
680                 cl_isize_lock(inode, 0);
681                 cl_object_attr_lock(obj);
682                 rc = cl_object_attr_get(env, obj, attr);
683                 if (rc == 0) {
684                         if (lock->cll_descr.cld_start == 0 &&
685                             lock->cll_descr.cld_end == CL_PAGE_EOF) {
686                                 cl_isize_write(inode, attr->cat_kms);
687                                 CDEBUG(D_INODE, DFID" updating i_size %llu\n",
688                                        PFID(lu_object_fid(&obj->co_lu)),
689                                        (__u64)cl_isize_read(inode));
690                         }
691                         cl_inode_mtime(inode) = attr->cat_mtime;
692                         cl_inode_atime(inode) = attr->cat_atime;
693                         cl_inode_ctime(inode) = attr->cat_ctime;
694                 } else
695                         CL_LOCK_DEBUG(D_ERROR, env, lock, "attr_get: %i\n", rc);
696                 cl_object_attr_unlock(obj);
697                 cl_isize_unlock(inode, 0);
698         }
699         EXIT;
700 }
701
702 /*****************************************************************************
703  *
704  * io operations.
705  *
706  */
707
708 void ccc_io_fini(const struct lu_env *env, const struct cl_io_slice *ios)
709 {
710         struct cl_io *io = ios->cis_io;
711
712         CLOBINVRNT(env, io->ci_obj, ccc_object_invariant(io->ci_obj));
713 }
714
715 int ccc_io_one_lock_index(const struct lu_env *env, struct cl_io *io,
716                           __u32 enqflags, enum cl_lock_mode mode,
717                           pgoff_t start, pgoff_t end)
718 {
719         struct ccc_io          *cio   = ccc_env_io(env);
720         struct cl_lock_descr   *descr = &cio->cui_link.cill_descr;
721         struct cl_object       *obj   = io->ci_obj;
722
723         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
724         ENTRY;
725
726         CDEBUG(D_VFSTRACE, "lock: %i [%lu, %lu]\n", mode, start, end);
727
728         memset(&cio->cui_link, 0, sizeof cio->cui_link);
729         descr->cld_mode  = mode;
730         descr->cld_obj   = obj;
731         descr->cld_start = start;
732         descr->cld_end   = end;
733
734         cio->cui_link.cill_enq_flags = enqflags;
735         cl_io_lock_add(env, io, &cio->cui_link);
736         RETURN(0);
737 }
738
739 void ccc_io_update_iov(const struct lu_env *env,
740                        struct ccc_io *cio, struct cl_io *io)
741 {
742         int i;
743         size_t size = io->u.ci_rw.crw_count;
744
745         cio->cui_iov_olen = 0;
746         if (cl_io_is_sendfile(io) || size == cio->cui_tot_count)
747                 return;
748
749         if (cio->cui_tot_nrsegs == 0)
750                 cio->cui_tot_nrsegs =  cio->cui_nrsegs;
751
752         for (i = 0; i < cio->cui_tot_nrsegs; i++) {
753                 struct iovec *iv = &cio->cui_iov[i];
754
755                 if (iv->iov_len < size)
756                         size -= iv->iov_len;
757                 else {
758                         if (iv->iov_len > size) {
759                                 cio->cui_iov_olen = iv->iov_len;
760                                 iv->iov_len = size;
761                         }
762                         break;
763                 }
764         }
765
766         cio->cui_nrsegs = i + 1;
767 }
768
769 int ccc_io_one_lock(const struct lu_env *env, struct cl_io *io,
770                     __u32 enqflags, enum cl_lock_mode mode,
771                     loff_t start, loff_t end)
772 {
773         struct cl_object *obj = io->ci_obj;
774         return ccc_io_one_lock_index(env, io, enqflags, mode,
775                                      cl_index(obj, start), cl_index(obj, end));
776 }
777
778 void ccc_io_end(const struct lu_env *env, const struct cl_io_slice *ios)
779 {
780         CLOBINVRNT(env, ios->cis_io->ci_obj,
781                    ccc_object_invariant(ios->cis_io->ci_obj));
782 }
783
784 void ccc_io_advance(const struct lu_env *env,
785                     const struct cl_io_slice *ios,
786                     size_t nob)
787 {
788         struct ccc_io    *cio = cl2ccc_io(env, ios);
789         struct cl_io     *io  = ios->cis_io;
790         struct cl_object *obj = ios->cis_io->ci_obj;
791
792         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
793
794         if (!cl_io_is_sendfile(io) && io->ci_continue) {
795                 /* update the iov */
796                 LASSERT(cio->cui_tot_nrsegs >= cio->cui_nrsegs);
797                 LASSERT(cio->cui_tot_count  >= nob);
798
799                 cio->cui_iov        += cio->cui_nrsegs;
800                 cio->cui_tot_nrsegs -= cio->cui_nrsegs;
801                 cio->cui_tot_count  -= nob;
802
803                 if (cio->cui_iov_olen) {
804                         struct iovec *iv;
805
806                         cio->cui_iov--;
807                         cio->cui_tot_nrsegs++;
808                         iv = &cio->cui_iov[0];
809                         iv->iov_base += iv->iov_len;
810                         LASSERT(cio->cui_iov_olen > iv->iov_len);
811                         iv->iov_len = cio->cui_iov_olen - iv->iov_len;
812                 }
813         }
814 }
815
816 static void ccc_object_size_lock(struct cl_object *obj, int vfslock)
817 {
818         struct inode *inode = ccc_object_inode(obj);
819
820         if (vfslock)
821                 cl_isize_lock(inode, 0);
822         cl_object_attr_lock(obj);
823 }
824
825 static void ccc_object_size_unlock(struct cl_object *obj, int vfslock)
826 {
827         struct inode *inode = ccc_object_inode(obj);
828
829         cl_object_attr_unlock(obj);
830         if (vfslock)
831                 cl_isize_unlock(inode, 0);
832 }
833
834 /**
835  * Helper function that if necessary adjusts file size (inode->i_size), when
836  * position at the offset \a pos is accessed. File size can be arbitrary stale
837  * on a Lustre client, but client at least knows KMS. If accessed area is
838  * inside [0, KMS], set file size to KMS, otherwise glimpse file size.
839  *
840  * Locking: cl_isize_lock is used to serialize changes to inode size and to
841  * protect consistency between inode size and cl_object
842  * attributes. cl_object_size_lock() protects consistency between cl_attr's of
843  * top-object and sub-objects.
844  *
845  * In page fault path cl_isize_lock cannot be taken, client has to live with
846  * the resulting races.
847  */
848 int ccc_prep_size(const struct lu_env *env, struct cl_object *obj,
849                   struct cl_io *io, loff_t start, size_t count, int vfslock,
850                   int *exceed)
851 {
852         struct cl_attr *attr  = &ccc_env_info(env)->cti_attr;
853         struct inode   *inode = ccc_object_inode(obj);
854         loff_t          pos   = start + count - 1;
855         loff_t kms;
856         int result;
857
858         /*
859          * Consistency guarantees: following possibilities exist for the
860          * relation between region being accessed and real file size at this
861          * moment:
862          *
863          *  (A): the region is completely inside of the file;
864          *
865          *  (B-x): x bytes of region are inside of the file, the rest is
866          *  outside;
867          *
868          *  (C): the region is completely outside of the file.
869          *
870          * This classification is stable under DLM lock already acquired by
871          * the caller, because to change the class, other client has to take
872          * DLM lock conflicting with our lock. Also, any updates to ->i_size
873          * by other threads on this client are serialized by
874          * ll_inode_size_lock(). This guarantees that short reads are handled
875          * correctly in the face of concurrent writes and truncates.
876          */
877         ccc_object_size_lock(obj, vfslock);
878         result = cl_object_attr_get(env, obj, attr);
879         if (result == 0) {
880                 kms = attr->cat_kms;
881                 if (pos > kms) {
882                         /*
883                          * A glimpse is necessary to determine whether we
884                          * return a short read (B) or some zeroes at the end
885                          * of the buffer (C)
886                          */
887                         ccc_object_size_unlock(obj, vfslock);
888                         result = cl_glimpse_lock(env, io, inode, obj);
889                         if (result == 0 && exceed != NULL) {
890                                 /* If objective page index exceed end-of-file
891                                  * page index, return directly. Do not expect
892                                  * kernel will check such case correctly.
893                                  * linux-2.6.18-128.1.1 miss to do that.
894                                  * --bug 17336 */
895                                 loff_t size = cl_isize_read(inode);
896                                 unsigned long cur_index = start >> CFS_PAGE_SHIFT;
897
898                                 if ((size == 0 && cur_index != 0) ||
899                                     (((size - 1) >> CFS_PAGE_SHIFT) < cur_index))
900                                 *exceed = 1;
901                         }
902                         return result;
903                 } else {
904                         /*
905                          * region is within kms and, hence, within real file
906                          * size (A). We need to increase i_size to cover the
907                          * read region so that generic_file_read() will do its
908                          * job, but that doesn't mean the kms size is
909                          * _correct_, it is only the _minimum_ size. If
910                          * someone does a stat they will get the correct size
911                          * which will always be >= the kms value here.
912                          * b=11081
913                          */
914                         /*
915                          * XXX in a page fault path, change inode size without
916                          * ll_inode_size_lock() held!  there is a race
917                          * condition with truncate path. (see ll_extent_lock)
918                          */
919                         /*
920                          * XXX i_size_write() is not used because it is not
921                          * safe to take the ll_inode_size_lock() due to a
922                          * potential lock inversion (bug 6077).  And since
923                          * it's not safe to use i_size_write() without a
924                          * covering mutex we do the assignment directly.  It
925                          * is not critical that the size be correct.
926                          */
927                         if (cl_isize_read(inode) < kms) {
928                                 if (vfslock)
929                                         cl_isize_write(inode, kms);
930                                 else
931                                         cl_isize_write_nolock(inode, kms);
932                         }
933                 }
934         }
935         ccc_object_size_unlock(obj, vfslock);
936         return result;
937 }
938
939 /*****************************************************************************
940  *
941  * Transfer operations.
942  *
943  */
944
945 void ccc_req_completion(const struct lu_env *env,
946                         const struct cl_req_slice *slice, int ioret)
947 {
948         struct ccc_req *vrq;
949
950         vrq = cl2ccc_req(slice);
951         OBD_SLAB_FREE_PTR(vrq, ccc_req_kmem);
952 }
953
954 /**
955  * Implementation of struct cl_req_operations::cro_attr_set() for ccc
956  * layer. ccc is responsible for
957  *
958  *    - o_[mac]time
959  *
960  *    - o_mode
961  *
962  *    - o_fid (filled with inode number?!)
963  *
964  *    - o_[ug]id
965  *
966  *    - o_generation
967  *
968  *    - and IO epoch (stored in o_easize),
969  *
970  *  and capability.
971  */
972 void ccc_req_attr_set(const struct lu_env *env,
973                       const struct cl_req_slice *slice,
974                       const struct cl_object *obj,
975                       struct cl_req_attr *attr, obd_valid flags)
976 {
977         struct inode *inode;
978         struct obdo  *oa;
979         obd_flag      valid_flags;
980
981         oa = attr->cra_oa;
982         inode = ccc_object_inode(obj);
983         valid_flags = OBD_MD_FLTYPE|OBD_MD_FLATIME;
984
985         if (flags != (obd_valid)~0ULL)
986                 valid_flags |= OBD_MD_FLMTIME|OBD_MD_FLCTIME|OBD_MD_FLATIME;
987         else {
988                 LASSERT(attr->cra_capa == NULL);
989                 attr->cra_capa = cl_capa_lookup(inode,
990                                                 slice->crs_req->crq_type);
991         }
992
993         if (slice->crs_req->crq_type == CRT_WRITE) {
994                 if (flags & OBD_MD_FLEPOCH) {
995                         oa->o_valid |= OBD_MD_FLEPOCH;
996                         oa->o_easize = cl_i2info(inode)->lli_ioepoch;
997                         valid_flags |= OBD_MD_FLMTIME|OBD_MD_FLCTIME|
998                                 OBD_MD_FLUID|OBD_MD_FLGID|
999                                 OBD_MD_FLFID|OBD_MD_FLGENER;
1000                 }
1001         }
1002         obdo_from_inode(oa, inode, valid_flags & flags);
1003 }
1004
1005 const struct cl_req_operations ccc_req_ops = {
1006         .cro_attr_set   = ccc_req_attr_set,
1007         .cro_completion = ccc_req_completion
1008 };
1009
1010 /* Setattr helpers */
1011 int cl_setattr_do_truncate(struct inode *inode, loff_t size,
1012                            struct obd_capa *capa)
1013 {
1014         struct lu_env *env;
1015         struct cl_io  *io;
1016         int            result;
1017         int            refcheck;
1018
1019         ENTRY;
1020
1021         env = cl_env_get(&refcheck);
1022         if (IS_ERR(env))
1023                 RETURN(PTR_ERR(env));
1024
1025         io = &ccc_env_info(env)->cti_io;
1026         io->ci_obj = cl_i2info(inode)->lli_clob;
1027         io->u.ci_truncate.tr_size = size;
1028         io->u.ci_truncate.tr_capa = capa;
1029         if (cl_io_init(env, io, CIT_TRUNC, io->ci_obj) == 0)
1030                 result = cl_io_loop(env, io);
1031         else
1032                 result = io->ci_result;
1033         cl_io_fini(env, io);
1034         cl_env_put(env, &refcheck);
1035         RETURN(result);
1036 }
1037
1038 int cl_setattr_ost(struct inode *inode, struct obd_capa *capa)
1039 {
1040         struct cl_inode_info *lli = cl_i2info(inode);
1041         struct lov_stripe_md *lsm = lli->lli_smd;
1042         int rc;
1043         obd_flag flags;
1044         struct obd_info oinfo = { { { 0 } } };
1045         struct obdo *oa;
1046
1047         OBDO_ALLOC(oa);
1048         if (oa) {
1049                 oa->o_id = lsm->lsm_object_id;
1050                 oa->o_gr = lsm->lsm_object_gr;
1051                 oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
1052
1053                 flags = OBD_MD_FLTYPE | OBD_MD_FLATIME |
1054                         OBD_MD_FLMTIME | OBD_MD_FLCTIME |
1055                         OBD_MD_FLFID | OBD_MD_FLGENER |
1056                         OBD_MD_FLGROUP;
1057
1058                 obdo_from_inode(oa, inode, flags);
1059
1060                 oinfo.oi_oa = oa;
1061                 oinfo.oi_md = lsm;
1062
1063                 /* XXX: this looks unnecessary now. */
1064                 rc = obd_setattr_rqset(cl_i2sbi(inode)->ll_dt_exp, &oinfo,
1065                                        NULL);
1066                 if (rc)
1067                         CERROR("obd_setattr_async fails: rc=%d\n", rc);
1068                 OBDO_FREE(oa);
1069         } else {
1070                 rc = -ENOMEM;
1071         }
1072         return rc;
1073 }
1074
1075
1076 /*****************************************************************************
1077  *
1078  * Type conversions.
1079  *
1080  */
1081
1082 struct lu_device *ccc2lu_dev(struct ccc_device *vdv)
1083 {
1084         return &vdv->cdv_cl.cd_lu_dev;
1085 }
1086
1087 struct ccc_device *lu2ccc_dev(const struct lu_device *d)
1088 {
1089         return container_of0(d, struct ccc_device, cdv_cl.cd_lu_dev);
1090 }
1091
1092 struct ccc_device *cl2ccc_dev(const struct cl_device *d)
1093 {
1094         return container_of0(d, struct ccc_device, cdv_cl);
1095 }
1096
1097 struct lu_object *ccc2lu(struct ccc_object *vob)
1098 {
1099         return &vob->cob_cl.co_lu;
1100 }
1101
1102 struct ccc_object *lu2ccc(const struct lu_object *obj)
1103 {
1104         return container_of0(obj, struct ccc_object, cob_cl.co_lu);
1105 }
1106
1107 struct ccc_object *cl2ccc(const struct cl_object *obj)
1108 {
1109         return container_of0(obj, struct ccc_object, cob_cl);
1110 }
1111
1112 struct ccc_lock *cl2ccc_lock(const struct cl_lock_slice *slice)
1113 {
1114         return container_of(slice, struct ccc_lock, clk_cl);
1115 }
1116
1117 struct ccc_io *cl2ccc_io(const struct lu_env *env,
1118                          const struct cl_io_slice *slice)
1119 {
1120         struct ccc_io *cio;
1121
1122         cio = container_of(slice, struct ccc_io, cui_cl);
1123         LASSERT(cio == ccc_env_io(env));
1124         return cio;
1125 }
1126
1127 struct ccc_req *cl2ccc_req(const struct cl_req_slice *slice)
1128 {
1129         return container_of0(slice, struct ccc_req, crq_cl);
1130 }
1131
1132 cfs_page_t *cl2vm_page(const struct cl_page_slice *slice)
1133 {
1134         return cl2ccc_page(slice)->cpg_page;
1135 }
1136
1137 /*****************************************************************************
1138  *
1139  * Accessors.
1140  *
1141  */
1142 int ccc_object_invariant(const struct cl_object *obj)
1143 {
1144         struct inode         *inode = ccc_object_inode(obj);
1145         struct cl_inode_info *lli   = cl_i2info(inode);
1146
1147         return (S_ISREG(cl_inode_mode(inode)) ||
1148                 /* i_mode of unlinked inode is zeroed. */
1149                 cl_inode_mode(inode) == 0) && lli->lli_clob == obj;
1150 }
1151
1152 struct inode *ccc_object_inode(const struct cl_object *obj)
1153 {
1154         return cl2ccc(obj)->cob_inode;
1155 }
1156
1157 /**
1158  * Returns a pointer to cl_page associated with \a vmpage, without acquiring
1159  * additional reference to the resulting page. This is an unsafe version of
1160  * cl_vmpage_page() that can only be used under vmpage lock.
1161  */
1162 struct cl_page *ccc_vmpage_page_transient(cfs_page_t *vmpage)
1163 {
1164         KLASSERT(PageLocked(vmpage));
1165         return (struct cl_page *)vmpage->private;
1166 }
1167
1168 /**
1169  * Initializes or updates CLIO part when new meta-data arrives from the
1170  * server.
1171  *
1172  *     - allocates cl_object if necessary,
1173  *     - updated layout, if object was already here.
1174  */
1175 int cl_inode_init(struct inode *inode, struct lustre_md *md)
1176 {
1177         struct lu_env        *env;
1178         struct cl_inode_info *lli;
1179         struct cl_object     *clob;
1180         struct lu_site       *site;
1181         struct lu_fid        *fid;
1182         const struct cl_object_conf conf = {
1183                 .coc_inode = inode,
1184                 .u = {
1185                         .coc_md    = md
1186                 }
1187         };
1188         int result = 0;
1189         int refcheck;
1190
1191         /* LASSERT(inode->i_state & I_NEW); */
1192         LASSERT(md->body->valid & OBD_MD_FLID);
1193
1194         if (!S_ISREG(cl_inode_mode(inode)))
1195                 return 0;
1196
1197         env = cl_env_get(&refcheck);
1198         if (IS_ERR(env))
1199                 return PTR_ERR(env);
1200
1201         site = cl_i2sbi(inode)->ll_site;
1202         lli  = cl_i2info(inode);
1203         fid  = &lli->lli_fid;
1204         LASSERT(fid_is_sane(fid));
1205
1206         if (lli->lli_clob == NULL) {
1207                 clob = cl_object_find(env, lu2cl_dev(site->ls_top_dev),
1208                                       fid, &conf);
1209                 if (!IS_ERR(clob)) {
1210                         /*
1211                          * No locking is necessary, as new inode is
1212                          * locked by I_NEW bit.
1213                          *
1214                          * XXX not true for call from ll_update_inode().
1215                          */
1216                         lli->lli_clob = clob;
1217                         lu_object_ref_add(&clob->co_lu, "inode", inode);
1218                 } else
1219                         result = PTR_ERR(clob);
1220         } else
1221                 result = cl_conf_set(env, lli->lli_clob, &conf);
1222         cl_env_put(env, &refcheck);
1223
1224         if (result != 0)
1225                 CERROR("Failure to initialize cl object "DFID": %d\n",
1226                        PFID(fid), result);
1227         return result;
1228 }
1229
1230 void cl_inode_fini(struct inode *inode)
1231 {
1232         struct lu_env           *env;
1233         struct cl_inode_info    *lli  = cl_i2info(inode);
1234         struct cl_object        *clob = lli->lli_clob;
1235         int refcheck;
1236         int emergency;
1237
1238         if (clob != NULL) {
1239                 void                    *cookie;
1240
1241                 cookie = cl_env_reenter();
1242                 env = cl_env_get(&refcheck);
1243                 emergency = IS_ERR(env);
1244                 if (emergency) {
1245                         mutex_lock(&ccc_inode_fini_guard);
1246                         LASSERT(ccc_inode_fini_env != NULL);
1247                         cl_env_implant(ccc_inode_fini_env, &refcheck);
1248                         env = ccc_inode_fini_env;
1249                 }
1250                 /*
1251                  * cl_object cache is a slave to inode cache (which, in turn
1252                  * is a slave to dentry cache), don't keep cl_object in memory
1253                  * when its master is evicted.
1254                  */
1255                 cl_object_kill(env, clob);
1256                 lu_object_ref_del(&clob->co_lu, "inode", inode);
1257                 cl_object_put(env, clob);
1258                 lli->lli_clob = NULL;
1259                 if (emergency) {
1260                         cl_env_unplant(ccc_inode_fini_env, &refcheck);
1261                         mutex_unlock(&ccc_inode_fini_guard);
1262                 } else
1263                         cl_env_put(env, &refcheck);
1264                 cl_env_reexit(cookie);
1265         }
1266 }
1267
1268 /**
1269  * return IF_* type for given lu_dirent entry.
1270  * IF_* flag shld be converted to particular OS file type in
1271  * platform llite module.
1272  */
1273 __u16 ll_dirent_type_get(struct lu_dirent *ent)
1274 {
1275         __u16 type = 0;
1276         struct luda_type *lt;
1277         int len = 0;
1278
1279         if (le32_to_cpu(ent->lde_attrs) & LUDA_TYPE) {
1280                 const unsigned align = sizeof(struct luda_type) - 1;
1281
1282                 len = le16_to_cpu(ent->lde_namelen);
1283                 len = (len + align) & ~align;
1284                 lt = (void *) ent->lde_name + len;
1285                 type = CFS_IFTODT(le16_to_cpu(lt->lt_type));
1286         }
1287         return type;
1288 }