Whamcloud - gitweb
Branch HEAD
[fs/lustre-release.git] / lustre / liblustre / llite_cl.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  *   Copyright (c) 2007 Cluster File Systems, Inc.
5  *   Author: Nikita Danilov <nikita@clusterfs.com>
6  *
7  *   This file is part of Lustre, http://www.lustre.org.
8  *
9  *   Lustre is free software; you can redistribute it and/or
10  *   modify it under the terms of version 2 of the GNU General Public
11  *   License as published by the Free Software Foundation.
12  *
13  *   Lustre is distributed in the hope that it will be useful,
14  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
15  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
16  *   GNU General Public License for more details.
17  *
18  *   You should have received a copy of the GNU General Public License
19  *   along with Lustre; if not, write to the Free Software
20  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
21  */
22
23 #define DEBUG_SUBSYSTEM S_LLITE
24
25 #include <stdlib.h>
26 #include <string.h>
27 #include <assert.h>
28 #include <time.h>
29 #include <sys/types.h>
30 #include <sys/stat.h>
31 #include <fcntl.h>
32 #include <sys/queue.h>
33 #ifndef __CYGWIN__
34 # include <sys/statvfs.h>
35 #else
36 # include <sys/statfs.h>
37 #endif
38
39 #include <sysio.h>
40 #ifdef HAVE_XTIO_H
41 #include <xtio.h>
42 #endif
43 #include <fs.h>
44 #include <mount.h>
45 #include <inode.h>
46 #ifdef HAVE_FILE_H
47 #include <file.h>
48 #endif
49 #include <liblustre.h>
50
51 #include <obd.h>
52 #include <obd_support.h>
53 #include <lustre_fid.h>
54 #include <lustre_lite.h>
55 #include <lustre_dlm.h>
56 #include <lustre_ver.h>
57 #include <lustre_mdc.h>
58 #include <cl_object.h>
59
60 #include "llite_lib.h"
61
62 /*
63  * slp_ prefix stands for "Sysio Library Posix". It corresponds to historical
64  * "llu_" prefix.
65  */
66
67 static int   slp_type_init     (struct lu_device_type *t);
68 static void  slp_type_fini     (struct lu_device_type *t);
69
70 static struct cl_page * slp_page_init(const struct lu_env *env,
71                                      struct cl_object *obj,
72                                      struct cl_page *page, cfs_page_t *vmpage);
73 static int   slp_attr_get     (const struct lu_env *env, struct cl_object *obj,
74                                struct cl_attr *attr);
75
76 static struct lu_device  *slp_device_alloc(const struct lu_env *env,
77                                            struct lu_device_type *t,
78                                            struct lustre_cfg *cfg);
79
80 static int slp_io_init(const struct lu_env *env, struct cl_object *obj,
81                        struct cl_io *io);
82 static struct slp_io *cl2slp_io(const struct lu_env *env,
83                                 const struct cl_io_slice *slice);
84
85
86 static void llu_free_user_page(struct page *page);
87
88 static const struct lu_object_operations      slp_lu_obj_ops;
89 static const struct lu_device_operations      slp_lu_ops;
90 static const struct cl_device_operations      slp_cl_ops;
91 static const struct cl_io_operations          ccc_io_ops;
92 static const struct lu_device_type_operations slp_device_type_ops;
93              //struct lu_device_type            slp_device_type;
94 static const struct cl_page_operations        slp_page_ops;
95 static const struct cl_page_operations        slp_transient_page_ops;
96 static const struct cl_lock_operations        slp_lock_ops;
97
98
99 /*****************************************************************************
100  *
101  * Slp device and device type functions.
102  *
103  */
104
105 void *slp_session_key_init(const struct lu_context *ctx,
106                                   struct lu_context_key *key)
107 {
108         struct slp_session *session;
109
110         OBD_ALLOC_PTR(session);
111         if (session == NULL)
112                 session = ERR_PTR(-ENOMEM);
113         return session;
114 }
115
116 void slp_session_key_fini(const struct lu_context *ctx,
117                                  struct lu_context_key *key, void *data)
118 {
119         struct slp_session *session = data;
120         OBD_FREE_PTR(session);
121 }
122
123 struct lu_context_key slp_session_key = {
124         .lct_tags = LCT_SESSION,
125         .lct_init = slp_session_key_init,
126         .lct_fini = slp_session_key_fini
127 };
128
129 /* type constructor/destructor: slp_type_{init,fini,start,stop}(). */
130 LU_TYPE_INIT_FINI(slp, &ccc_key, &ccc_session_key, &slp_session_key);
131
132 static struct lu_device *slp_device_alloc(const struct lu_env *env,
133                                           struct lu_device_type *t,
134                                           struct lustre_cfg *cfg)
135 {
136         return ccc_device_alloc(env, t, cfg, &slp_lu_ops, &slp_cl_ops);
137 }
138
139 static int slp_lock_init(const struct lu_env *env,
140                          struct cl_object *obj, struct cl_lock *lock,
141                          const struct cl_io *io)
142 {
143         return ccc_lock_init(env, obj, lock, io, &slp_lock_ops);
144 }
145
146 static const struct cl_object_operations slp_ops = {
147         .coo_page_init = slp_page_init,
148         .coo_lock_init = slp_lock_init,
149         .coo_io_init   = slp_io_init,
150         .coo_attr_get  = slp_attr_get,
151         .coo_attr_set  = ccc_attr_set,
152         .coo_conf_set  = ccc_conf_set,
153         .coo_glimpse   = ccc_object_glimpse
154 };
155
156 static int slp_object_print(const struct lu_env *env, void *cookie,
157                             lu_printer_t p, const struct lu_object *o)
158 {
159         struct ccc_object *obj   = lu2ccc(o);
160         struct inode      *inode = obj->cob_inode;
161         struct intnl_stat *st = NULL;
162
163         if (inode)
164                 st = llu_i2stat(inode);
165
166         return (*p)(env, cookie, LUSTRE_SLP_NAME"-object@%p(%p:%lu/%u)",
167                     obj, inode,
168                     st ? (unsigned long)st->st_ino : 0UL,
169                     inode ? (unsigned int)llu_i2info(inode)->lli_st_generation
170                     : 0);
171 }
172
173 static const struct lu_object_operations slp_lu_obj_ops = {
174         .loo_object_init      = ccc_object_init,
175         .loo_object_start     = NULL,
176         .loo_object_delete    = NULL,
177         .loo_object_release   = NULL,
178         .loo_object_free      = ccc_object_free,
179         .loo_object_print     = slp_object_print,
180         .loo_object_invariant = NULL
181 };
182
183 static struct lu_object *slp_object_alloc(const struct lu_env *env,
184                                           const struct lu_object_header *hdr,
185                                           struct lu_device *dev)
186 {
187         return ccc_object_alloc(env, hdr, dev, &slp_ops, &slp_lu_obj_ops);
188 }
189
190 static const struct lu_device_operations slp_lu_ops = {
191         .ldo_object_alloc      = slp_object_alloc
192 };
193
194 static const struct cl_device_operations slp_cl_ops = {
195         .cdo_req_init = ccc_req_init
196 };
197
198 static const struct lu_device_type_operations slp_device_type_ops = {
199         .ldto_init = slp_type_init,
200         .ldto_fini = slp_type_fini,
201
202         .ldto_start = slp_type_start,
203         .ldto_stop  = slp_type_stop,
204
205         .ldto_device_alloc = slp_device_alloc,
206         .ldto_device_free  = ccc_device_free,
207         .ldto_device_init  = ccc_device_init,
208         .ldto_device_fini  = ccc_device_fini
209 };
210
211 struct lu_device_type slp_device_type = {
212         .ldt_tags     = LU_DEVICE_CL,
213         .ldt_name     = LUSTRE_SLP_NAME,
214         .ldt_ops      = &slp_device_type_ops,
215         .ldt_ctx_tags = LCT_CL_THREAD
216 };
217
218 int slp_global_init(void)
219 {
220         int result;
221
222         result = ccc_global_init(&slp_device_type);
223         return result;
224 }
225
226 void slp_global_fini(void)
227 {
228         ccc_global_fini(&slp_device_type);
229 }
230
231 /*****************************************************************************
232  *
233  * Object operations.
234  *
235  */
236
237 static struct cl_page *slp_page_init(const struct lu_env *env,
238                                      struct cl_object *obj,
239                                      struct cl_page *page, cfs_page_t *vmpage)
240 {
241         struct ccc_page *cpg;
242         int result;
243
244         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
245
246         OBD_ALLOC_PTR(cpg);
247         if (cpg != NULL) {
248                 cpg->cpg_page = vmpage;
249
250                 if (page->cp_type == CPT_CACHEABLE) {
251                         LBUG();
252                 } else {
253                         struct ccc_object *clobj = cl2ccc(obj);
254
255                         cl_page_slice_add(page, &cpg->cpg_cl, obj,
256                                           &slp_transient_page_ops);
257                         clobj->cob_transient_pages++;
258                 }
259                 result = 0;
260         } else
261                 result = -ENOMEM;
262         return ERR_PTR(result);
263 }
264
265 static int slp_io_init(const struct lu_env *env, struct cl_object *obj,
266                        struct cl_io *io)
267 {
268         struct ccc_io      *vio   = ccc_env_io(env);
269         int result = 0;
270
271         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
272
273         cl_io_slice_add(io, &vio->cui_cl, obj, &ccc_io_ops);
274         if (io->ci_type == CIT_READ || io->ci_type == CIT_WRITE) {
275                 size_t count;
276
277                 count = io->u.ci_rw.crw_count;
278                 /* "If nbyte is 0, read() will return 0 and have no other
279                  *  results."  -- Single Unix Spec */
280                 if (count == 0)
281                         result = 1;
282                 else {
283                         vio->cui_tot_count = count;
284                         vio->cui_tot_nrsegs = 0;
285                 }
286
287         }
288         return result;
289 }
290
291 static int slp_attr_get(const struct lu_env *env, struct cl_object *obj,
292                         struct cl_attr *attr)
293 {
294         struct inode *inode = ccc_object_inode(obj);
295         struct intnl_stat *st = llu_i2stat(inode);
296
297         attr->cat_size = st->st_size;
298         attr->cat_blocks = st->st_blocks;
299         attr->cat_mtime  = st->st_mtime;
300         attr->cat_atime  = st->st_atime;
301         attr->cat_ctime  = st->st_ctime;
302         /* KMS is not known by this layer */
303         return 0; /* layers below have to fill in the rest */
304 }
305
306 /*****************************************************************************
307  *
308  * Page operations.
309  *
310  */
311
312 static void slp_page_fini_common(struct ccc_page *cp)
313 {
314         cfs_page_t *vmpage = cp->cpg_page;
315
316         LASSERT(vmpage != NULL);
317         llu_free_user_page(vmpage);
318         OBD_FREE_PTR(cp);
319 }
320
321 static void slp_page_completion_common(const struct lu_env *env,
322                                        struct ccc_page *cp, int ioret)
323 {
324         struct cl_sync_io *anchor = cp->cpg_sync_io;
325
326         if (anchor) {
327                 cp->cpg_sync_io  = NULL;
328                 cl_sync_io_note(anchor, ioret);
329         } else {
330                 LBUG();
331         }
332 }
333
334 static void slp_page_completion_read(const struct lu_env *env,
335                                      const struct cl_page_slice *slice,
336                                      int ioret)
337 {
338         struct ccc_page *cp      = cl2ccc_page(slice);
339         ENTRY;
340
341         slp_page_completion_common(env, cp, ioret);
342
343         EXIT;
344 }
345
346 static void slp_page_completion_write_common(const struct lu_env *env,
347                                              const struct cl_page_slice *slice,
348                                              int ioret)
349 {
350         struct ccc_page *cp     = cl2ccc_page(slice);
351
352         if (ioret == 0) {
353                 cp->cpg_write_queued = 0;
354                 /*
355                  * Only ioret == 0, write succeed, then this page could be
356                  * deleted from the pending_writing count.
357                  */
358         }
359         slp_page_completion_common(env, cp, ioret);
360 }
361
362 static int slp_page_is_vmlocked(const struct lu_env *env,
363                                 const struct cl_page_slice *slice)
364 {
365         return -EBUSY;
366 }
367
368 static void slp_transient_page_fini(const struct lu_env *env,
369                                     struct cl_page_slice *slice)
370 {
371         struct ccc_page *cp = cl2ccc_page(slice);
372         struct cl_page *clp = slice->cpl_page;
373         struct ccc_object *clobj = cl2ccc(clp->cp_obj);
374
375         slp_page_fini_common(cp);
376         clobj->cob_transient_pages--;
377 }
378
379
380 static const struct cl_page_operations slp_transient_page_ops = {
381         .cpo_own           = ccc_transient_page_own,
382         .cpo_assume        = ccc_transient_page_assume,
383         .cpo_unassume      = ccc_transient_page_unassume,
384         .cpo_disown        = ccc_transient_page_disown,
385         .cpo_discard       = ccc_transient_page_discard,
386         .cpo_vmpage        = ccc_page_vmpage,
387         .cpo_is_vmlocked   = slp_page_is_vmlocked,
388         .cpo_fini          = slp_transient_page_fini,
389         .cpo_is_under_lock = ccc_page_is_under_lock,
390         .io = {
391                 [CRT_READ] = {
392                         .cpo_completion  = slp_page_completion_read,
393                 },
394                 [CRT_WRITE] = {
395                         .cpo_completion  = slp_page_completion_write_common,
396                 }
397         }
398 };
399
400 /*****************************************************************************
401  *
402  * Lock operations.
403  *
404  */
405
406 static int slp_lock_enqueue(const struct lu_env *env,
407                            const struct cl_lock_slice *slice,
408                            struct cl_io *_, __u32 enqflags)
409 {
410         CLOBINVRNT(env, slice->cls_obj, ccc_object_invariant(slice->cls_obj));
411
412         liblustre_wait_event(0);
413         return 0;
414 }
415
416 static const struct cl_lock_operations slp_lock_ops = {
417         .clo_fini      = ccc_lock_fini,
418         .clo_enqueue   = slp_lock_enqueue,
419         .clo_wait      = ccc_lock_wait,
420         .clo_unuse     = ccc_lock_unuse,
421         .clo_fits_into = ccc_lock_fits_into,
422 };
423
424 /*****************************************************************************
425  *
426  * io operations.
427  *
428  */
429
430 static int slp_io_rw_lock(const struct lu_env *env,
431                              const struct cl_io_slice *ios)
432 {
433         struct cl_io *io = ios->cis_io;
434         loff_t start;
435         loff_t end;
436
437         if (cl_io_is_append(io)) {
438                 start = 0;
439                 end   = OBD_OBJECT_EOF;
440         } else {
441                 start = io->u.ci_wr.wr.crw_pos;
442                 end   = start + io->u.ci_wr.wr.crw_count - 1;
443         }
444         /*
445          * This acquires real DLM lock only in O_APPEND case, because of
446          * the io->ci_lockreq setting in llu_io_init().
447          */
448         LASSERT(ergo(cl_io_is_append(io), io->ci_lockreq == CILR_MANDATORY));
449         LASSERT(ergo(!cl_io_is_append(io), io->ci_lockreq == CILR_NEVER));
450         return ccc_io_one_lock(env, io, 0,
451                                io->ci_type == CIT_READ ? CLM_READ : CLM_WRITE,
452                                start, end);
453
454 }
455
456 static int slp_io_trunc_iter_init(const struct lu_env *env,
457                                   const struct cl_io_slice *ios)
458 {
459         return 0;
460 }
461
462 static int slp_io_trunc_start(const struct lu_env *env,
463                               const struct cl_io_slice *ios)
464 {
465         return 0;
466 }
467
468 static struct page *llu_get_user_page(int index, void *addr, int offset,
469                                       int count)
470 {
471         struct page *page;
472
473         OBD_ALLOC_PTR(page);
474         if (!page)
475                 return NULL;
476         page->index = index;
477         page->addr = addr;
478         page->_offset = offset;
479         page->_count = count;
480
481         CFS_INIT_LIST_HEAD(&page->list);
482         CFS_INIT_LIST_HEAD(&page->_node);
483
484         return page;
485 }
486
487 static void llu_free_user_page(struct page *page)
488 {
489         OBD_FREE_PTR(page);
490 }
491
492 static int llu_queue_pio(const struct lu_env *env, struct cl_io *io,
493                          struct llu_io_group *group,
494                          char *buf, size_t count, loff_t pos)
495 {
496         struct cl_object *obj = io->ci_obj;
497         struct inode *inode = ccc_object_inode(obj);
498         struct intnl_stat *st = llu_i2stat(inode);
499         struct obd_export *exp = llu_i2obdexp(inode);
500         struct page *page;
501         int  rc = 0, npages = 0, ret_bytes = 0;
502         int local_lock;
503         struct cl_page *clp;
504         struct ccc_page *clup;
505         struct cl_2queue *queue;
506         struct cl_sync_io *anchor = &ccc_env_info(env)->cti_sync_io;
507         ENTRY;
508
509         if (!exp)
510                 RETURN(-EINVAL);
511
512         local_lock = group->lig_params->lrp_lock_mode != LCK_NL;
513
514         queue = &io->ci_queue;
515         cl_2queue_init(queue);
516
517
518         /* prepare the pages array */
519         do {
520                 unsigned long index, offset, bytes;
521
522                 offset = (pos & ~CFS_PAGE_MASK);
523                 index = pos >> CFS_PAGE_SHIFT;
524                 bytes = CFS_PAGE_SIZE - offset;
525                 if (bytes > count)
526                         bytes = count;
527
528                 /* prevent read beyond file range */
529                 if (/* local_lock && */
530                     io->ci_type == CIT_READ && pos + bytes >= st->st_size) {
531                         if (pos >= st->st_size)
532                                 break;
533                         bytes = st->st_size - pos;
534                 }
535
536                 /* prepare page for this index */
537                 page = llu_get_user_page(index, buf - offset, offset, bytes);
538                 if (!page) {
539                         rc = -ENOMEM;
540                         break;
541                 }
542
543                 clp = cl_page_find(env, obj,
544                                    cl_index(obj, pos),
545                                    page, CPT_TRANSIENT);
546
547                 if (IS_ERR(clp)) {
548                         rc = PTR_ERR(clp);
549                         break;
550                 }
551
552                 rc = cl_page_own(env, io, clp);
553                 if (rc) {
554                         LASSERT(clp->cp_state == CPS_FREEING);
555                         cl_page_put(env, clp);
556                         break;
557                 }
558
559                 clup = cl2ccc_page(cl_page_at(clp, &slp_device_type));
560                 clup->cpg_sync_io = anchor;
561                 cl_2queue_add(queue, clp);
562
563                 /* drop the reference count for cl_page_find, so that the page
564                  * will be freed in cl_2queue_fini. */
565                 cl_page_put(env, clp);
566
567                 cl_page_clip(env, clp, offset, offset+bytes);
568
569                 npages++;
570                 count -= bytes;
571                 pos += bytes;
572                 buf += bytes;
573
574                 group->lig_rwcount += bytes;
575                 ret_bytes += bytes;
576                 page++;
577         } while (count);
578
579         cl_sync_io_init(anchor, npages);
580         /* printk("Inited anchor with %d pages\n", npages); */
581
582         if (rc == 0) {
583                 enum cl_req_type crt;
584
585                 crt = io->ci_type == CIT_READ ? CRT_READ : CRT_WRITE;
586                 rc = cl_io_submit_rw(env, io, crt, queue, CRP_NORMAL);
587                 if (rc == 0) {
588                         /* If some pages weren't sent for any reason, count
589                          * then as completed, to avoid infinite wait. */
590                         cl_page_list_for_each(clp, &queue->c2_qin) {
591                                 CL_PAGE_DEBUG(D_ERROR, env, clp,
592                                               "not completed\n");
593                                 cl_sync_io_note(anchor, +1);
594                         }
595                         /* wait for the IO to be finished. */
596                         rc = cl_sync_io_wait(env, io, &queue->c2_qout, anchor);
597                 }
598         }
599
600         group->lig_rc = rc;
601
602         cl_2queue_discard(env, io, queue);
603         cl_2queue_disown(env, io, queue);
604         cl_2queue_fini(env, queue);
605
606         RETURN(ret_bytes);
607 }
608
609 static
610 struct llu_io_group * get_io_group(struct inode *inode, int maxpages,
611                                    struct lustre_rw_params *params)
612 {
613         struct llu_io_group *group;
614
615         OBD_ALLOC_PTR(group);
616         if (!group)
617                 return ERR_PTR(-ENOMEM);
618
619         group->lig_params = params;
620
621         return group;
622 }
623
624 static int max_io_pages(ssize_t len, int iovlen)
625 {
626         return (((len + CFS_PAGE_SIZE -1) / CFS_PAGE_SIZE) + 2 + iovlen - 1);
627 }
628
629 void put_io_group(struct llu_io_group *group)
630 {
631         OBD_FREE_PTR(group);
632 }
633
634 static int slp_io_start(const struct lu_env *env, const struct cl_io_slice *ios)
635 {
636         struct ccc_io     *cio   = cl2ccc_io(env, ios);
637         struct cl_io      *io    = ios->cis_io;
638         struct cl_object  *obj   = io->ci_obj;
639         struct inode      *inode = ccc_object_inode(obj);
640         int     err, ret;
641         loff_t  pos;
642         size_t  cnt;
643         struct llu_io_group *iogroup;
644         struct lustre_rw_params p = {0};
645         int iovidx;
646         struct intnl_stat *st = llu_i2stat(inode);
647         struct llu_inode_info *lli = llu_i2info(inode);
648         struct llu_io_session *session = cl2slp_io(env, ios)->sio_session;
649         int write = io->ci_type == CIT_WRITE;
650
651         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
652
653         if (write) {
654                 pos = io->u.ci_wr.wr.crw_pos;
655                 cnt = io->u.ci_wr.wr.crw_count;
656         } else {
657                 pos = io->u.ci_rd.rd.crw_pos;
658                 cnt = io->u.ci_rd.rd.crw_count;
659         }
660         if (io->u.ci_wr.wr_append) {
661                 p.lrp_lock_mode = LCK_PW;
662         } else {
663                 p.lrp_brw_flags = OBD_BRW_SRVLOCK;
664                 p.lrp_lock_mode = LCK_NL;
665         }
666
667         iogroup = get_io_group(inode, max_io_pages(cnt, cio->cui_nrsegs), &p);
668         if (IS_ERR(iogroup))
669                 RETURN(PTR_ERR(iogroup));
670
671         err = ccc_prep_size(env, obj, io, pos + cnt - 1, 0);
672         if (err != 0)
673                 GOTO(out, err);
674
675         CDEBUG(D_INODE,
676                "%s ino %lu, "LPSZ" bytes, offset %lld, i_size %llu\n",
677                write?"Write":"Read", (unsigned long)st->st_ino,
678                cnt, (__u64)pos, (__u64)st->st_size);
679
680         if (write && io->u.ci_wr.wr_append)
681                 pos = io->u.ci_wr.wr.crw_pos = st->st_size; /* XXX? Do we need to change io content too here? */
682                 /* XXX What about if one write syscall writes at 2 different offsets? */
683
684         for (iovidx = 0; iovidx < cio->cui_nrsegs; iovidx++) {
685                 char *buf = (char *) cio->cui_iov[iovidx].iov_base;
686                 size_t count = cio->cui_iov[iovidx].iov_len;
687
688                 if (!count)
689                         continue;
690                 if (cnt < count)
691                         count = cnt;
692                 if (IS_BAD_PTR(buf) || IS_BAD_PTR(buf + count)) {
693                         GOTO(out, err = -EFAULT);
694                 }
695
696                 if (io->ci_type == CIT_READ) {
697                         if (/* local_lock && */ pos >= st->st_size)
698                                 break;
699                 } else if (io->ci_type == CIT_WRITE) {
700                         if (pos >= lli->lli_maxbytes) {
701                                 GOTO(out, err = -EFBIG);
702                         }
703                         if (pos + count >= lli->lli_maxbytes)
704                                 count = lli->lli_maxbytes - pos;
705                 } else {
706                         LBUG();
707                 }
708
709                 ret = llu_queue_pio(env, io, iogroup, buf, count, pos);
710                 if (ret < 0) {
711                         GOTO(out, err = ret);
712                 } else {
713                         io->ci_nob += ret;
714                         pos += ret;
715                         cnt -= ret;
716                         if (io->ci_type == CIT_WRITE) {
717 //                                obd_adjust_kms(exp, lsm, pos, 0); // XXX
718                                 if (pos > st->st_size)
719                                         st->st_size = pos;
720                         }
721                         if (!cnt)
722                                 break;
723                 }
724         }
725         LASSERT(cnt == 0 || io->ci_type == CIT_READ); /* libsysio should guarantee this */
726
727         session->lis_groups[session->lis_ngroups++] = iogroup;
728
729         return 0;
730 out:
731         put_io_group(iogroup);
732         return err;
733 }
734
735 static const struct cl_io_operations ccc_io_ops = {
736         .op = {
737                 [CIT_READ] = {
738                         .cio_fini      = ccc_io_fini,
739                         .cio_lock      = slp_io_rw_lock,
740                         .cio_start     = slp_io_start,
741                         .cio_end       = ccc_io_end
742                 },
743                 [CIT_WRITE] = {
744                         .cio_fini      = ccc_io_fini,
745                         .cio_lock      = slp_io_rw_lock,
746                         .cio_start     = slp_io_start,
747                         .cio_end       = ccc_io_end
748                 },
749                 [CIT_TRUNC] = {
750                         .cio_fini       = ccc_io_fini,
751                         .cio_iter_init  = slp_io_trunc_iter_init,
752                         .cio_start      = slp_io_trunc_start
753                 },
754                 [CIT_MISC] = {
755                         .cio_fini   = ccc_io_fini
756                 }
757         }
758 };
759
760 static struct slp_io *cl2slp_io(const struct lu_env *env,
761                                 const struct cl_io_slice *slice)
762 {
763         /* We call it just for assertion here */
764         cl2ccc_io(env, slice);
765
766         return slp_env_io(env);
767 }
768
769 /*****************************************************************************
770  *
771  * Temporary prototype thing: mirror obd-devices into cl devices.
772  *
773  */
774
775 int cl_sb_init(struct llu_sb_info *sbi)
776 {
777         struct cl_device  *cl;
778         struct lu_env     *env;
779         int rc = 0;
780         int refcheck;
781
782         env = cl_env_get(&refcheck);
783         if (IS_ERR(env))
784                 RETURN(PTR_ERR(env));
785
786         cl = cl_type_setup(env, NULL, &slp_device_type,
787                            sbi->ll_dt_exp->exp_obd->obd_lu_dev);
788         if (IS_ERR(cl))
789                 GOTO(out, rc = PTR_ERR(cl));
790
791         sbi->ll_cl = cl;
792         sbi->ll_site = cl2lu_dev(cl)->ld_site;
793 out:
794         cl_env_put(env, &refcheck);
795         RETURN(rc);
796 }
797
798 int cl_sb_fini(struct llu_sb_info *sbi)
799 {
800         struct lu_env *env;
801         int refcheck;
802
803         ENTRY;
804
805         env = cl_env_get(&refcheck);
806         if (IS_ERR(env))
807                 RETURN(PTR_ERR(env));
808
809         if (sbi->ll_cl != NULL) {
810                 cl_stack_fini(env, sbi->ll_cl);
811                 sbi->ll_cl = NULL;
812                 sbi->ll_site = NULL;
813         }
814         cl_env_put(env, &refcheck);
815         /*
816          * If mount failed (sbi->ll_cl == NULL), and this there are no other
817          * mounts, stop device types manually (this usually happens
818          * automatically when last device is destroyed).
819          */
820         lu_types_stop();
821         cl_env_cache_purge(~0);
822         RETURN(0);
823 }