Whamcloud - gitweb
b161c11c328faff479e3e335b25f0b477bca90bd
[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        *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 ccc 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 start, size_t count, int vfslock,
785                   int *exceed)
786 {
787         struct cl_attr *attr  = &ccc_env_info(env)->cti_attr;
788         struct inode   *inode = ccc_object_inode(obj);
789         loff_t          pos   = start + count - 1;
790         loff_t kms;
791         int result;
792
793         /*
794          * Consistency guarantees: following possibilities exist for the
795          * relation between region being accessed and real file size at this
796          * moment:
797          *
798          *  (A): the region is completely inside of the file;
799          *
800          *  (B-x): x bytes of region are inside of the file, the rest is
801          *  outside;
802          *
803          *  (C): the region is completely outside of the file.
804          *
805          * This classification is stable under DLM lock already acquired by
806          * the caller, because to change the class, other client has to take
807          * DLM lock conflicting with our lock. Also, any updates to ->i_size
808          * by other threads on this client are serialized by
809          * ll_inode_size_lock(). This guarantees that short reads are handled
810          * correctly in the face of concurrent writes and truncates.
811          */
812         ccc_object_size_lock(obj, vfslock);
813         result = cl_object_attr_get(env, obj, attr);
814         if (result == 0) {
815                 kms = attr->cat_kms;
816                 if (pos > kms) {
817                         /*
818                          * A glimpse is necessary to determine whether we
819                          * return a short read (B) or some zeroes at the end
820                          * of the buffer (C)
821                          */
822                         ccc_object_size_unlock(obj, vfslock);
823                         result = cl_glimpse_lock(env, io, inode, obj);
824                         if (result == 0 && exceed != NULL) {
825                                 /* If objective page index exceed end-of-file
826                                  * page index, return directly. Do not expect
827                                  * kernel will check such case correctly.
828                                  * linux-2.6.18-128.1.1 miss to do that.
829                                  * --bug 17336 */
830                                 size_t size = cl_isize_read(inode);
831                                 unsigned long cur_index = start >> CFS_PAGE_SHIFT;
832
833                                 if ((size == 0 && cur_index != 0) ||
834                                     (((size - 1) >> CFS_PAGE_SHIFT) < cur_index))
835                                 *exceed = 1;
836                         }
837                         return result;
838                 } else {
839                         /*
840                          * region is within kms and, hence, within real file
841                          * size (A). We need to increase i_size to cover the
842                          * read region so that generic_file_read() will do its
843                          * job, but that doesn't mean the kms size is
844                          * _correct_, it is only the _minimum_ size. If
845                          * someone does a stat they will get the correct size
846                          * which will always be >= the kms value here.
847                          * b=11081
848                          */
849                         /*
850                          * XXX in a page fault path, change inode size without
851                          * ll_inode_size_lock() held!  there is a race
852                          * condition with truncate path. (see ll_extent_lock)
853                          */
854                         /*
855                          * XXX i_size_write() is not used because it is not
856                          * safe to take the ll_inode_size_lock() due to a
857                          * potential lock inversion (bug 6077).  And since
858                          * it's not safe to use i_size_write() without a
859                          * covering mutex we do the assignment directly.  It
860                          * is not critical that the size be correct.
861                          */
862                         if (cl_isize_read(inode) < kms) {
863                                 if (vfslock)
864                                         cl_isize_write(inode, kms);
865                                 else
866                                         cl_isize_write_nolock(inode, kms);
867                         }
868                 }
869         }
870         ccc_object_size_unlock(obj, vfslock);
871         return result;
872 }
873
874 /*****************************************************************************
875  *
876  * Transfer operations.
877  *
878  */
879
880 void ccc_req_completion(const struct lu_env *env,
881                         const struct cl_req_slice *slice, int ioret)
882 {
883         struct ccc_req *vrq;
884
885         vrq = cl2ccc_req(slice);
886         OBD_SLAB_FREE_PTR(vrq, ccc_req_kmem);
887 }
888
889 /**
890  * Implementation of struct cl_req_operations::cro_attr_set() for ccc
891  * layer. ccc is responsible for
892  *
893  *    - o_[mac]time
894  *
895  *    - o_mode
896  *
897  *    - o_fid (filled with inode number?!)
898  *
899  *    - o_[ug]id
900  *
901  *    - o_generation
902  *
903  *    - and IO epoch (stored in o_easize),
904  *
905  *  and capability.
906  */
907 void ccc_req_attr_set(const struct lu_env *env,
908                       const struct cl_req_slice *slice,
909                       const struct cl_object *obj,
910                       struct cl_req_attr *attr, obd_valid flags)
911 {
912         struct inode *inode;
913         struct obdo  *oa;
914         obd_flag      valid_flags;
915
916         oa = attr->cra_oa;
917         inode = ccc_object_inode(obj);
918         valid_flags = OBD_MD_FLTYPE|OBD_MD_FLATIME;
919
920         if (flags != (obd_valid)~0ULL)
921                 valid_flags |= OBD_MD_FLMTIME|OBD_MD_FLCTIME|OBD_MD_FLATIME;
922         else {
923                 LASSERT(attr->cra_capa == NULL);
924                 attr->cra_capa = cl_capa_lookup(inode,
925                                                 slice->crs_req->crq_type);
926         }
927
928         if (slice->crs_req->crq_type == CRT_WRITE) {
929                 if (flags & OBD_MD_FLEPOCH) {
930                         oa->o_valid |= OBD_MD_FLEPOCH;
931                         oa->o_easize = cl_i2info(inode)->lli_ioepoch;
932                         valid_flags |= OBD_MD_FLMTIME|OBD_MD_FLCTIME|
933                                 OBD_MD_FLUID|OBD_MD_FLGID|
934                                 OBD_MD_FLFID|OBD_MD_FLGENER;
935                 }
936         }
937         obdo_from_inode(oa, inode, valid_flags & flags);
938 }
939
940 const struct cl_req_operations ccc_req_ops = {
941         .cro_attr_set   = ccc_req_attr_set,
942         .cro_completion = ccc_req_completion
943 };
944
945 /* Setattr helpers */
946 int cl_setattr_do_truncate(struct inode *inode, loff_t size,
947                            struct obd_capa *capa)
948 {
949         struct lu_env *env;
950         struct cl_io  *io;
951         int            result;
952         int            refcheck;
953
954         ENTRY;
955
956         env = cl_env_get(&refcheck);
957         if (IS_ERR(env))
958                 RETURN(PTR_ERR(env));
959
960         io = &ccc_env_info(env)->cti_io;
961         io->ci_obj = cl_i2info(inode)->lli_clob;
962         io->u.ci_truncate.tr_size = size;
963         io->u.ci_truncate.tr_capa = capa;
964         if (cl_io_init(env, io, CIT_TRUNC, io->ci_obj) == 0)
965                 result = cl_io_loop(env, io);
966         else
967                 result = io->ci_result;
968         cl_io_fini(env, io);
969         cl_env_put(env, &refcheck);
970         RETURN(result);
971 }
972
973 int cl_setattr_ost(struct inode *inode, struct obd_capa *capa)
974 {
975         struct cl_inode_info *lli = cl_i2info(inode);
976         struct lov_stripe_md *lsm = lli->lli_smd;
977         int rc;
978         obd_flag flags;
979         struct obd_info oinfo = { { { 0 } } };
980         struct obdo *oa;
981
982         OBDO_ALLOC(oa);
983         if (oa) {
984                 oa->o_id = lsm->lsm_object_id;
985                 oa->o_gr = lsm->lsm_object_gr;
986                 oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
987
988                 flags = OBD_MD_FLTYPE | OBD_MD_FLATIME |
989                         OBD_MD_FLMTIME | OBD_MD_FLCTIME |
990                         OBD_MD_FLFID | OBD_MD_FLGENER |
991                         OBD_MD_FLGROUP;
992
993                 obdo_from_inode(oa, inode, flags);
994
995                 oinfo.oi_oa = oa;
996                 oinfo.oi_md = lsm;
997
998                 /* XXX: this looks unnecessary now. */
999                 rc = obd_setattr_rqset(cl_i2sbi(inode)->ll_dt_exp, &oinfo,
1000                                        NULL);
1001                 if (rc)
1002                         CERROR("obd_setattr_async fails: rc=%d\n", rc);
1003                 OBDO_FREE(oa);
1004         } else {
1005                 rc = -ENOMEM;
1006         }
1007         return rc;
1008 }
1009
1010
1011 /*****************************************************************************
1012  *
1013  * Type conversions.
1014  *
1015  */
1016
1017 struct lu_device *ccc2lu_dev(struct ccc_device *vdv)
1018 {
1019         return &vdv->cdv_cl.cd_lu_dev;
1020 }
1021
1022 struct ccc_device *lu2ccc_dev(const struct lu_device *d)
1023 {
1024         return container_of0(d, struct ccc_device, cdv_cl.cd_lu_dev);
1025 }
1026
1027 struct ccc_device *cl2ccc_dev(const struct cl_device *d)
1028 {
1029         return container_of0(d, struct ccc_device, cdv_cl);
1030 }
1031
1032 struct lu_object *ccc2lu(struct ccc_object *vob)
1033 {
1034         return &vob->cob_cl.co_lu;
1035 }
1036
1037 struct ccc_object *lu2ccc(const struct lu_object *obj)
1038 {
1039         return container_of0(obj, struct ccc_object, cob_cl.co_lu);
1040 }
1041
1042 struct ccc_object *cl2ccc(const struct cl_object *obj)
1043 {
1044         return container_of0(obj, struct ccc_object, cob_cl);
1045 }
1046
1047 struct ccc_lock *cl2ccc_lock(const struct cl_lock_slice *slice)
1048 {
1049         return container_of(slice, struct ccc_lock, clk_cl);
1050 }
1051
1052 struct ccc_io *cl2ccc_io(const struct lu_env *env,
1053                          const struct cl_io_slice *slice)
1054 {
1055         struct ccc_io *cio;
1056
1057         cio = container_of(slice, struct ccc_io, cui_cl);
1058         LASSERT(cio == ccc_env_io(env));
1059         return cio;
1060 }
1061
1062 struct ccc_req *cl2ccc_req(const struct cl_req_slice *slice)
1063 {
1064         return container_of0(slice, struct ccc_req, crq_cl);
1065 }
1066
1067 cfs_page_t *cl2vm_page(const struct cl_page_slice *slice)
1068 {
1069         return cl2ccc_page(slice)->cpg_page;
1070 }
1071
1072 /*****************************************************************************
1073  *
1074  * Accessors.
1075  *
1076  */
1077 int ccc_object_invariant(const struct cl_object *obj)
1078 {
1079         struct inode         *inode = ccc_object_inode(obj);
1080         struct cl_inode_info *lli   = cl_i2info(inode);
1081
1082         return (S_ISREG(cl_inode_mode(inode)) ||
1083                 /* i_mode of unlinked inode is zeroed. */
1084                 cl_inode_mode(inode) == 0) && lli->lli_clob == obj;
1085 }
1086
1087 struct inode *ccc_object_inode(const struct cl_object *obj)
1088 {
1089         return cl2ccc(obj)->cob_inode;
1090 }
1091
1092 /**
1093  * Returns a pointer to cl_page associated with \a vmpage, without acquiring
1094  * additional reference to the resulting page. This is an unsafe version of
1095  * cl_vmpage_page() that can only be used under vmpage lock.
1096  */
1097 struct cl_page *ccc_vmpage_page_transient(cfs_page_t *vmpage)
1098 {
1099         KLASSERT(PageLocked(vmpage));
1100         return (struct cl_page *)vmpage->private;
1101 }
1102
1103 /**
1104  * Initializes or updates CLIO part when new meta-data arrives from the
1105  * server.
1106  *
1107  *     - allocates cl_object if necessary,
1108  *     - updated layout, if object was already here.
1109  */
1110 int cl_inode_init(struct inode *inode, struct lustre_md *md)
1111 {
1112         struct lu_env        *env;
1113         struct cl_inode_info *lli;
1114         struct cl_object     *clob;
1115         struct lu_site       *site;
1116         struct lu_fid        *fid;
1117         const struct cl_object_conf conf = {
1118                 .coc_inode = inode,
1119                 .u = {
1120                         .coc_md    = md
1121                 }
1122         };
1123         int result = 0;
1124         int refcheck;
1125
1126         /* LASSERT(inode->i_state & I_NEW); */
1127         LASSERT(md->body->valid & OBD_MD_FLID);
1128
1129         if (!S_ISREG(cl_inode_mode(inode)))
1130                 return 0;
1131
1132         env = cl_env_get(&refcheck);
1133         if (IS_ERR(env))
1134                 return PTR_ERR(env);
1135
1136         site = cl_i2sbi(inode)->ll_site;
1137         lli  = cl_i2info(inode);
1138         fid  = &lli->lli_fid;
1139         LASSERT(fid_is_sane(fid));
1140
1141         if (lli->lli_clob == NULL) {
1142                 clob = cl_object_find(env, lu2cl_dev(site->ls_top_dev),
1143                                       fid, &conf);
1144                 if (!IS_ERR(clob)) {
1145                         /*
1146                          * No locking is necessary, as new inode is
1147                          * locked by I_NEW bit.
1148                          *
1149                          * XXX not true for call from ll_update_inode().
1150                          */
1151                         lli->lli_clob = clob;
1152                         lu_object_ref_add(&clob->co_lu, "inode", inode);
1153                 } else
1154                         result = PTR_ERR(clob);
1155         } else
1156                 result = cl_conf_set(env, lli->lli_clob, &conf);
1157         cl_env_put(env, &refcheck);
1158
1159         if (result != 0)
1160                 CERROR("Failure to initialize cl object "DFID": %d\n",
1161                        PFID(fid), result);
1162         return result;
1163 }
1164
1165 void cl_inode_fini(struct inode *inode)
1166 {
1167         struct lu_env           *env;
1168         struct cl_inode_info    *lli  = cl_i2info(inode);
1169         struct cl_object        *clob = lli->lli_clob;
1170         int refcheck;
1171         int emergency;
1172
1173         if (clob != NULL) {
1174                 struct lu_object_header *head = clob->co_lu.lo_header;
1175                 void                    *cookie;
1176
1177                 cookie = cl_env_reenter();
1178                 env = cl_env_get(&refcheck);
1179                 emergency = IS_ERR(env);
1180                 if (emergency) {
1181                         mutex_lock(&ccc_inode_fini_guard);
1182                         LASSERT(ccc_inode_fini_env != NULL);
1183                         cl_env_implant(ccc_inode_fini_env, &refcheck);
1184                         env = ccc_inode_fini_env;
1185                 }
1186                 /*
1187                  * cl_object cache is a slave to inode cache (which, in turn
1188                  * is a slave to dentry cache), don't keep cl_object in memory
1189                  * when its master is evicted.
1190                  */
1191                 cl_object_kill(env, clob);
1192                 lu_object_ref_del(&clob->co_lu, "inode", inode);
1193                 /* XXX temporary: this is racy */
1194                 LASSERT(atomic_read(&head->loh_ref) == 1);
1195                 cl_object_put(env, clob);
1196                 lli->lli_clob = NULL;
1197                 if (emergency) {
1198                         cl_env_unplant(ccc_inode_fini_env, &refcheck);
1199                         mutex_unlock(&ccc_inode_fini_guard);
1200                 } else
1201                         cl_env_put(env, &refcheck);
1202                 cl_env_reexit(cookie);
1203         }
1204 }
1205
1206 /**
1207  * return IF_* type for given lu_dirent entry.
1208  * IF_* flag shld be converted to particular OS file type in
1209  * platform llite module.
1210  */
1211 __u16 ll_dirent_type_get(struct lu_dirent *ent)
1212 {
1213         __u16 type = 0;
1214         struct luda_type *lt;
1215         int len = 0;
1216
1217         if (le32_to_cpu(ent->lde_attrs) & LUDA_TYPE) {
1218                 const unsigned align = sizeof(struct luda_type) - 1;
1219
1220                 len = le16_to_cpu(ent->lde_namelen);
1221                 len = (len + align) & ~align;
1222                 lt = (void *) ent->lde_name + len;
1223                 type = CFS_IFTODT(le16_to_cpu(lt->lt_type));
1224         }
1225         return type;
1226 }