Whamcloud - gitweb
Branch HEAD
[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(info, ccc_thread_kmem);
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(session, ccc_session_kmem);
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(vrq, ccc_req_kmem);
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(vob, ccc_object_kmem);
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(clk, ccc_lock_kmem);
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        *vio  = 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 (vio->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_fini(const struct lu_env *env, struct cl_lock_slice *slice)
564 {
565         struct ccc_lock *clk = cl2ccc_lock(slice);
566
567         CLOBINVRNT(env, slice->cls_obj, ccc_object_invariant(slice->cls_obj));
568         OBD_SLAB_FREE_PTR(clk, ccc_lock_kmem);
569 }
570
571 int ccc_lock_enqueue(const struct lu_env *env,
572                      const struct cl_lock_slice *slice,
573                      struct cl_io *_, __u32 enqflags)
574 {
575         CLOBINVRNT(env, slice->cls_obj, ccc_object_invariant(slice->cls_obj));
576         return 0;
577 }
578
579 int ccc_lock_unuse(const struct lu_env *env, const struct cl_lock_slice *slice)
580 {
581         CLOBINVRNT(env, slice->cls_obj, ccc_object_invariant(slice->cls_obj));
582         return 0;
583 }
584
585 int ccc_lock_wait(const struct lu_env *env, const struct cl_lock_slice *slice)
586 {
587         CLOBINVRNT(env, slice->cls_obj, ccc_object_invariant(slice->cls_obj));
588         return 0;
589 }
590
591 /**
592  * Implementation of cl_lock_operations::clo_fits_into() methods for ccc
593  * layer. This function is executed every time io finds an existing lock in
594  * the lock cache while creating new lock. This function has to decide whether
595  * cached lock "fits" into io.
596  *
597  * \param slice lock to be checked
598  *
599  * \param io    IO that wants a lock.
600  *
601  * \see lov_lock_fits_into().
602  */
603 int ccc_lock_fits_into(const struct lu_env *env,
604                        const struct cl_lock_slice *slice,
605                        const struct cl_lock_descr *need,
606                        const struct cl_io *io)
607 {
608         const struct cl_lock       *lock  = slice->cls_lock;
609         const struct cl_lock_descr *descr = &lock->cll_descr;
610         const struct ccc_io        *cio   = ccc_env_io(env);
611         int                         result;
612
613         ENTRY;
614         /*
615          * Work around DLM peculiarity: it assumes that glimpse
616          * (LDLM_FL_HAS_INTENT) lock is always LCK_PR, and returns reads lock
617          * when asked for LCK_PW lock with LDLM_FL_HAS_INTENT flag set. Make
618          * sure that glimpse doesn't get CLM_WRITE top-lock, so that it
619          * doesn't enqueue CLM_WRITE sub-locks.
620          */
621         if (cio->cui_glimpse)
622                 result = descr->cld_mode != CLM_WRITE;
623         /*
624          * Also, don't match incomplete write locks for read, otherwise read
625          * would enqueue missing sub-locks in the write mode.
626          *
627          * XXX this is a candidate for generic locking policy, to be moved
628          * into cl_lock_lookup().
629          */
630         else if (need->cld_mode != descr->cld_mode)
631                 result = lock->cll_state >= CLS_ENQUEUED;
632         else
633                 result = 1;
634         RETURN(result);
635 }
636
637 /**
638  * Implements cl_lock_operations::clo_state() method for vvp layer, invoked
639  * whenever lock state changes. Transfers object attributes, that might be
640  * updated as a result of lock acquiring into inode.
641  */
642 void ccc_lock_state(const struct lu_env *env,
643                     const struct cl_lock_slice *slice,
644                     enum cl_lock_state state)
645 {
646         struct cl_lock   *lock;
647         struct cl_object *obj;
648         struct inode     *inode;
649         struct cl_attr   *attr;
650
651         ENTRY;
652         lock = slice->cls_lock;
653
654         /*
655          * Refresh inode attributes when the lock is moving into CLS_HELD
656          * state, and only when this is a result of real enqueue, rather than
657          * of finding lock in the cache.
658          */
659         if (state == CLS_HELD && lock->cll_state < CLS_HELD) {
660                 int rc;
661
662                 obj   = slice->cls_obj;
663                 inode = ccc_object_inode(obj);
664                 attr  = &ccc_env_info(env)->cti_attr;
665
666                 /* vmtruncate()->ll_truncate() first sets the i_size and then
667                  * the kms under both a DLM lock and the
668                  * ll_inode_size_lock().  If we don't get the
669                  * ll_inode_size_lock() here we can match the DLM lock and
670                  * reset i_size from the kms before the truncating path has
671                  * updated the kms.  generic_file_write can then trust the
672                  * stale i_size when doing appending writes and effectively
673                  * cancel the result of the truncate.  Getting the
674                  * ll_inode_size_lock() after the enqueue maintains the DLM
675                  * -> ll_inode_size_lock() acquiring order. */
676                 cl_isize_lock(inode, 0);
677                 cl_object_attr_lock(obj);
678                 rc = cl_object_attr_get(env, obj, attr);
679                 if (rc == 0) {
680                         if (lock->cll_descr.cld_start == 0 &&
681                             lock->cll_descr.cld_end == CL_PAGE_EOF) {
682                                 cl_isize_write(inode, attr->cat_kms);
683                                 CDEBUG(D_INODE, DFID" updating i_size %llu\n",
684                                        PFID(lu_object_fid(&obj->co_lu)),
685                                        (__u64)cl_isize_read(inode));
686                         }
687                         cl_inode_mtime(inode) = attr->cat_mtime;
688                         cl_inode_atime(inode) = attr->cat_atime;
689                         cl_inode_ctime(inode) = attr->cat_ctime;
690                 } else
691                         CL_LOCK_DEBUG(D_ERROR, env, lock, "attr_get: %i\n", rc);
692                 cl_object_attr_unlock(obj);
693                 cl_isize_unlock(inode, 0);
694         }
695         EXIT;
696 }
697
698 /*****************************************************************************
699  *
700  * io operations.
701  *
702  */
703
704 void ccc_io_fini(const struct lu_env *env, const struct cl_io_slice *ios)
705 {
706         struct cl_io *io = ios->cis_io;
707
708         CLOBINVRNT(env, io->ci_obj, ccc_object_invariant(io->ci_obj));
709 }
710
711 int ccc_io_one_lock_index(const struct lu_env *env, struct cl_io *io,
712                           __u32 enqflags, enum cl_lock_mode mode,
713                           pgoff_t start, pgoff_t end)
714 {
715         struct ccc_io          *vio   = ccc_env_io(env);
716         struct cl_lock_descr   *descr = &vio->cui_link.cill_descr;
717         struct cl_object       *obj   = io->ci_obj;
718
719         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
720         ENTRY;
721
722         CDEBUG(D_VFSTRACE, "lock: %i [%lu, %lu]\n", mode, start, end);
723
724         memset(&vio->cui_link, 0, sizeof vio->cui_link);
725         descr->cld_mode  = mode;
726         descr->cld_obj   = obj;
727         descr->cld_start = start;
728         descr->cld_end   = end;
729
730         vio->cui_link.cill_enq_flags = enqflags;
731         cl_io_lock_add(env, io, &vio->cui_link);
732         RETURN(0);
733 }
734
735 int ccc_io_one_lock(const struct lu_env *env, struct cl_io *io,
736                     __u32 enqflags, enum cl_lock_mode mode,
737                     loff_t start, loff_t end)
738 {
739         struct cl_object *obj = io->ci_obj;
740
741         return ccc_io_one_lock_index(env, io, enqflags, mode,
742                                      cl_index(obj, start), cl_index(obj, end));
743 }
744
745 void ccc_io_end(const struct lu_env *env, const struct cl_io_slice *ios)
746 {
747         CLOBINVRNT(env, ios->cis_io->ci_obj,
748                    ccc_object_invariant(ios->cis_io->ci_obj));
749 }
750
751 static void ccc_object_size_lock(struct cl_object *obj, int vfslock)
752 {
753         struct inode *inode = ccc_object_inode(obj);
754
755         if (vfslock)
756                 cl_isize_lock(inode, 0);
757         cl_object_attr_lock(obj);
758 }
759
760 static void ccc_object_size_unlock(struct cl_object *obj, int vfslock)
761 {
762         struct inode *inode = ccc_object_inode(obj);
763
764         cl_object_attr_unlock(obj);
765         if (vfslock)
766                 cl_isize_unlock(inode, 0);
767 }
768
769 /**
770  * Helper function that if necessary adjusts file size (inode->i_size), when
771  * position at the offset \a pos is accessed. File size can be arbitrary stale
772  * on a Lustre client, but client at least knows KMS. If accessed area is
773  * inside [0, KMS], set file size to KMS, otherwise glimpse file size.
774  *
775  * Locking: cl_isize_lock is used to serialize changes to inode size and to
776  * protect consistency between inode size and cl_object
777  * attributes. cl_object_size_lock() protects consistency between cl_attr's of
778  * top-object and sub-objects.
779  *
780  * In page fault path cl_isize_lock cannot be taken, client has to live with
781  * the resulting races.
782  */
783 int ccc_prep_size(const struct lu_env *env, struct cl_object *obj,
784                   struct cl_io *io, loff_t pos, int vfslock)
785 {
786         struct cl_attr *attr  = &ccc_env_info(env)->cti_attr;
787         struct inode   *inode = ccc_object_inode(obj);
788         loff_t kms;
789         int result;
790
791         /*
792          * Consistency guarantees: following possibilities exist for the
793          * relation between region being accessed and real file size at this
794          * moment:
795          *
796          *  (A): the region is completely inside of the file;
797          *
798          *  (B-x): x bytes of region are inside of the file, the rest is
799          *  outside;
800          *
801          *  (C): the region is completely outside of the file.
802          *
803          * This classification is stable under DLM lock already acquired by
804          * the caller, because to change the class, other client has to take
805          * DLM lock conflicting with our lock. Also, any updates to ->i_size
806          * by other threads on this client are serialized by
807          * ll_inode_size_lock(). This guarantees that short reads are handled
808          * correctly in the face of concurrent writes and truncates.
809          */
810         ccc_object_size_lock(obj, vfslock);
811         result = cl_object_attr_get(env, obj, attr);
812         if (result == 0) {
813                 kms = attr->cat_kms;
814                 if (pos > kms) {
815                         /*
816                          * A glimpse is necessary to determine whether we
817                          * return a short read (B) or some zeroes at the end
818                          * of the buffer (C)
819                          */
820                         ccc_object_size_unlock(obj, vfslock);
821                         return cl_glimpse_lock(env, io, inode, obj);
822                 } else {
823                         /*
824                          * region is within kms and, hence, within real file
825                          * size (A). We need to increase i_size to cover the
826                          * read region so that generic_file_read() will do its
827                          * job, but that doesn't mean the kms size is
828                          * _correct_, it is only the _minimum_ size. If
829                          * someone does a stat they will get the correct size
830                          * which will always be >= the kms value here.
831                          * b=11081
832                          */
833                         /*
834                          * XXX in a page fault path, change inode size without
835                          * ll_inode_size_lock() held!  there is a race
836                          * condition with truncate path. (see ll_extent_lock)
837                          */
838                         /*
839                          * XXX i_size_write() is not used because it is not
840                          * safe to take the ll_inode_size_lock() due to a
841                          * potential lock inversion (bug 6077).  And since
842                          * it's not safe to use i_size_write() without a
843                          * covering mutex we do the assignment directly.  It
844                          * is not critical that the size be correct.
845                          */
846                         if (cl_isize_read(inode) < kms) {
847                                 if (vfslock)
848                                         cl_isize_write(inode, kms);
849                                 else
850                                         cl_isize_write_nolock(inode, kms);
851                         }
852                 }
853         }
854         ccc_object_size_unlock(obj, vfslock);
855         return result;
856 }
857
858 /*****************************************************************************
859  *
860  * Transfer operations.
861  *
862  */
863
864 void ccc_req_completion(const struct lu_env *env,
865                         const struct cl_req_slice *slice, int ioret)
866 {
867         struct ccc_req *vrq;
868
869         vrq = cl2ccc_req(slice);
870         OBD_SLAB_FREE_PTR(vrq, ccc_req_kmem);
871 }
872
873 /**
874  * Implementation of struct cl_req_operations::cro_attr_set() for ccc
875  * layer. ccc is responsible for
876  *
877  *    - o_[mac]time
878  *
879  *    - o_mode
880  *
881  *    - o_fid (filled with inode number?!)
882  *
883  *    - o_[ug]id
884  *
885  *    - o_generation
886  *
887  *    - and IO epoch (stored in o_easize),
888  *
889  *  and capability.
890  */
891 void ccc_req_attr_set(const struct lu_env *env,
892                       const struct cl_req_slice *slice,
893                       const struct cl_object *obj,
894                       struct cl_req_attr *attr, obd_valid flags)
895 {
896         struct inode *inode;
897         struct obdo  *oa;
898         obd_flag      valid_flags;
899
900         oa = attr->cra_oa;
901         inode = ccc_object_inode(obj);
902         valid_flags = OBD_MD_FLTYPE|OBD_MD_FLATIME;
903
904         if (flags != (obd_valid)~0ULL)
905                 valid_flags |= OBD_MD_FLMTIME|OBD_MD_FLCTIME|OBD_MD_FLATIME;
906         else {
907                 LASSERT(attr->cra_capa == NULL);
908                 attr->cra_capa = cl_capa_lookup(inode,
909                                                 slice->crs_req->crq_type);
910         }
911
912         if (slice->crs_req->crq_type == CRT_WRITE) {
913                 if (flags & OBD_MD_FLEPOCH) {
914                         oa->o_valid |= OBD_MD_FLEPOCH;
915                         oa->o_easize = cl_i2info(inode)->lli_ioepoch;
916                         valid_flags |= OBD_MD_FLMTIME|OBD_MD_FLCTIME|
917                                 OBD_MD_FLUID|OBD_MD_FLGID|
918                                 OBD_MD_FLFID|OBD_MD_FLGENER;
919                 }
920         }
921         obdo_from_inode(oa, inode, valid_flags & flags);
922 }
923
924 const struct cl_req_operations ccc_req_ops = {
925         .cro_attr_set   = ccc_req_attr_set,
926         .cro_completion = ccc_req_completion
927 };
928
929 /* Setattr helpers */
930 int cl_setattr_do_truncate(struct inode *inode, loff_t size,
931                            struct obd_capa *capa)
932 {
933         struct lu_env *env;
934         struct cl_io  *io;
935         int            result;
936         int            refcheck;
937
938         ENTRY;
939
940         env = cl_env_get(&refcheck);
941         if (IS_ERR(env))
942                 RETURN(PTR_ERR(env));
943
944         io = &ccc_env_info(env)->cti_io;
945         io->ci_obj = cl_i2info(inode)->lli_clob;
946         io->u.ci_truncate.tr_size = size;
947         io->u.ci_truncate.tr_capa = capa;
948         if (cl_io_init(env, io, CIT_TRUNC, io->ci_obj) == 0)
949                 result = cl_io_loop(env, io);
950         else
951                 result = io->ci_result;
952         cl_io_fini(env, io);
953         cl_env_put(env, &refcheck);
954         RETURN(result);
955 }
956
957 int cl_setattr_ost(struct inode *inode, struct obd_capa *capa)
958 {
959         struct cl_inode_info *lli = cl_i2info(inode);
960         struct lov_stripe_md *lsm = lli->lli_smd;
961         int rc;
962         obd_flag flags;
963         struct obd_info oinfo = { { { 0 } } };
964         struct obdo *oa;
965
966         OBDO_ALLOC(oa);
967         if (oa) {
968                 oa->o_id = lsm->lsm_object_id;
969                 oa->o_gr = lsm->lsm_object_gr;
970                 oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
971
972                 flags = OBD_MD_FLTYPE | OBD_MD_FLATIME |
973                         OBD_MD_FLMTIME | OBD_MD_FLCTIME |
974                         OBD_MD_FLFID | OBD_MD_FLGENER |
975                         OBD_MD_FLGROUP;
976
977                 obdo_from_inode(oa, inode, flags);
978
979                 oinfo.oi_oa = oa;
980                 oinfo.oi_md = lsm;
981
982                 /* XXX: this looks unnecessary now. */
983                 rc = obd_setattr_rqset(cl_i2sbi(inode)->ll_dt_exp, &oinfo,
984                                        NULL);
985                 if (rc)
986                         CERROR("obd_setattr_async fails: rc=%d\n", rc);
987                 OBDO_FREE(oa);
988         } else {
989                 rc = -ENOMEM;
990         }
991         return rc;
992 }
993
994
995 /*****************************************************************************
996  *
997  * Type conversions.
998  *
999  */
1000
1001 struct lu_device *ccc2lu_dev(struct ccc_device *vdv)
1002 {
1003         return &vdv->cdv_cl.cd_lu_dev;
1004 }
1005
1006 struct ccc_device *lu2ccc_dev(const struct lu_device *d)
1007 {
1008         return container_of0(d, struct ccc_device, cdv_cl.cd_lu_dev);
1009 }
1010
1011 struct ccc_device *cl2ccc_dev(const struct cl_device *d)
1012 {
1013         return container_of0(d, struct ccc_device, cdv_cl);
1014 }
1015
1016 struct lu_object *ccc2lu(struct ccc_object *vob)
1017 {
1018         return &vob->cob_cl.co_lu;
1019 }
1020
1021 struct ccc_object *lu2ccc(const struct lu_object *obj)
1022 {
1023         return container_of0(obj, struct ccc_object, cob_cl.co_lu);
1024 }
1025
1026 struct ccc_object *cl2ccc(const struct cl_object *obj)
1027 {
1028         return container_of0(obj, struct ccc_object, cob_cl);
1029 }
1030
1031 struct ccc_lock *cl2ccc_lock(const struct cl_lock_slice *slice)
1032 {
1033         return container_of(slice, struct ccc_lock, clk_cl);
1034 }
1035
1036 struct ccc_io *cl2ccc_io(const struct lu_env *env,
1037                          const struct cl_io_slice *slice)
1038 {
1039         struct ccc_io *cio;
1040
1041         cio = container_of(slice, struct ccc_io, cui_cl);
1042         LASSERT(cio == ccc_env_io(env));
1043         return cio;
1044 }
1045
1046 struct ccc_req *cl2ccc_req(const struct cl_req_slice *slice)
1047 {
1048         return container_of0(slice, struct ccc_req, crq_cl);
1049 }
1050
1051 cfs_page_t *cl2vm_page(const struct cl_page_slice *slice)
1052 {
1053         return cl2ccc_page(slice)->cpg_page;
1054 }
1055
1056 /*****************************************************************************
1057  *
1058  * Accessors.
1059  *
1060  */
1061 int ccc_object_invariant(const struct cl_object *obj)
1062 {
1063         struct inode         *inode = ccc_object_inode(obj);
1064         struct cl_inode_info *lli   = cl_i2info(inode);
1065
1066         return (S_ISREG(cl_inode_mode(inode)) ||
1067                 /* i_mode of unlinked inode is zeroed. */
1068                 cl_inode_mode(inode) == 0) && lli->lli_clob == obj;
1069 }
1070
1071 struct inode *ccc_object_inode(const struct cl_object *obj)
1072 {
1073         return cl2ccc(obj)->cob_inode;
1074 }
1075
1076 /**
1077  * Returns a pointer to cl_page associated with \a vmpage, without acquiring
1078  * additional reference to the resulting page. This is an unsafe version of
1079  * cl_vmpage_page() that can only be used under vmpage lock.
1080  */
1081 struct cl_page *ccc_vmpage_page_transient(cfs_page_t *vmpage)
1082 {
1083         KLASSERT(PageLocked(vmpage));
1084         return (struct cl_page *)vmpage->private;
1085 }
1086
1087 /**
1088  * Initializes or updates CLIO part when new meta-data arrives from the
1089  * server.
1090  *
1091  *     - allocates cl_object if necessary,
1092  *     - updated layout, if object was already here.
1093  */
1094 int cl_inode_init(struct inode *inode, struct lustre_md *md)
1095 {
1096         struct lu_env        *env;
1097         struct cl_inode_info *lli;
1098         struct cl_object     *clob;
1099         struct lu_site       *site;
1100         struct lu_fid        *fid;
1101         const struct cl_object_conf conf = {
1102                 .coc_inode = inode,
1103                 .u = {
1104                         .coc_md    = md
1105                 }
1106         };
1107         int result = 0;
1108         int refcheck;
1109
1110         /* LASSERT(inode->i_state & I_NEW); */
1111         LASSERT(md->body->valid & OBD_MD_FLID);
1112
1113         if (!S_ISREG(cl_inode_mode(inode)))
1114                 return 0;
1115
1116         env = cl_env_get(&refcheck);
1117         if (IS_ERR(env))
1118                 return PTR_ERR(env);
1119
1120         site = cl_i2sbi(inode)->ll_site;
1121         lli  = cl_i2info(inode);
1122         fid  = &lli->lli_fid;
1123         LASSERT(fid_is_sane(fid));
1124
1125         if (lli->lli_clob == NULL) {
1126                 clob = cl_object_find(env, lu2cl_dev(site->ls_top_dev),
1127                                       fid, &conf);
1128                 if (!IS_ERR(clob)) {
1129                         /*
1130                          * No locking is necessary, as new inode is
1131                          * locked by I_NEW bit.
1132                          *
1133                          * XXX not true for call from ll_update_inode().
1134                          */
1135                         lli->lli_clob = clob;
1136                         lu_object_ref_add(&clob->co_lu, "inode", inode);
1137                 } else
1138                         result = PTR_ERR(clob);
1139         } else
1140                 result = cl_conf_set(env, lli->lli_clob, &conf);
1141         cl_env_put(env, &refcheck);
1142
1143         if (result != 0)
1144                 CERROR("Failure to initialize cl object "DFID": %d\n",
1145                        PFID(fid), result);
1146         return result;
1147 }
1148
1149 void cl_inode_fini(struct inode *inode)
1150 {
1151         struct lu_env           *env;
1152         struct cl_inode_info    *lli  = cl_i2info(inode);
1153         struct cl_object        *clob = lli->lli_clob;
1154         int refcheck;
1155         int emergency;
1156
1157         if (clob != NULL) {
1158                 struct lu_object_header *head = clob->co_lu.lo_header;
1159                 void                    *cookie;
1160
1161                 cookie = cl_env_reenter();
1162                 env = cl_env_get(&refcheck);
1163                 emergency = IS_ERR(env);
1164                 if (emergency) {
1165                         mutex_lock(&ccc_inode_fini_guard);
1166                         LASSERT(ccc_inode_fini_env != NULL);
1167                         cl_env_implant(ccc_inode_fini_env, &refcheck);
1168                         env = ccc_inode_fini_env;
1169                 }
1170                 /*
1171                  * cl_object cache is a slave to inode cache (which, in turn
1172                  * is a slave to dentry cache), don't keep cl_object in memory
1173                  * when its master is evicted.
1174                  */
1175                 cl_object_kill(env, clob);
1176                 lu_object_ref_del(&clob->co_lu, "inode", inode);
1177                 /* XXX temporary: this is racy */
1178                 LASSERT(atomic_read(&head->loh_ref) == 1);
1179                 cl_object_put(env, clob);
1180                 lli->lli_clob = NULL;
1181                 if (emergency) {
1182                         cl_env_unplant(ccc_inode_fini_env, &refcheck);
1183                         mutex_unlock(&ccc_inode_fini_guard);
1184                 } else
1185                         cl_env_put(env, &refcheck);
1186                 cl_env_reexit(cookie);
1187         }
1188 }