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