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