Whamcloud - gitweb
LU-3534 osp: move RPC pack from declare to execution phase
[fs/lustre-release.git] / lustre / osp / osp_dev.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
19  *
20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21  * CA 95054 USA or visit www.sun.com if you need additional information or
22  * have any questions.
23  *
24  * GPL HEADER END
25  */
26 /*
27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2012, 2014, Intel Corporation.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * lustre/osp/osp_dev.c
37  *
38  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
39  * Author: Mikhail Pershin <mike.pershin@intel.com>
40  * Author: Di Wang <di.wang@intel.com>
41  */
42 /*
43  * The Object Storage Proxy (OSP) module provides an implementation of
44  * the DT API for remote MDTs and OSTs. Every local OSP device (or
45  * object) is a proxy for a remote OSD device (or object). Thus OSP
46  * converts DT operations into RPCs, which are sent to the OUT service
47  * on a remote target, converted back to DT operations, and
48  * executed. Of course there are many ways in which this description
49  * is inaccurate but it's a good enough mental model. OSP is used by
50  * the MDT stack in several ways:
51  *
52  * - OSP devices allocate FIDs for the stripe sub-objects of a striped
53  *   file or directory.
54  *
55  * - OSP objects represent the remote MDT and OST objects that are
56  *   the stripes of a striped object.
57  *
58  * - OSP devices log, send, and track synchronous operations (setattr
59  *   and unlink) to remote targets.
60  *
61  * - OSP objects are the bottom slice of the compound LU object
62  *   representing a remote MDT object: MDT/MDD/LOD/OSP.
63  *
64  * - OSP objects are used by LFSCK to represent remote OST objects
65  *   during the verification of MDT-OST consistency.
66  *
67  * - OSP devices batch idempotent requests (declare_attr_get() and
68  *   declare_xattr_get()) to the remote target and cache their results.
69  *
70  * In addition the OSP layer implements a subset of the OBD device API
71  * to support being a client of a remote target, connecting to other
72  * layers, and FID allocation.
73  */
74
75 #define DEBUG_SUBSYSTEM S_MDS
76
77 #include <obd_class.h>
78 #include <lustre_ioctl.h>
79 #include <lustre_param.h>
80 #include <lustre_log.h>
81
82 #include "osp_internal.h"
83
84 /* Slab for OSP object allocation */
85 struct kmem_cache *osp_object_kmem;
86
87 static struct lu_kmem_descr osp_caches[] = {
88         {
89                 .ckd_cache = &osp_object_kmem,
90                 .ckd_name  = "osp_obj",
91                 .ckd_size  = sizeof(struct osp_object)
92         },
93         {
94                 .ckd_cache = NULL
95         }
96 };
97
98 /**
99  * Implementation of lu_device_operations::ldo_object_alloc
100  *
101  * Allocates an OSP object in memory, whose FID is on the remote target.
102  *
103  * \param[in] env       execution environment
104  * \param[in] hdr       The header of the object stack. If it is NULL, it
105  *                      means the object is not built from top device, i.e.
106  *                      it is a sub-stripe object of striped directory or
107  *                      an OST object.
108  * \param[in] d         OSP device
109  *
110  * \retval object       object being created if the creation succeed.
111  * \retval NULL         NULL if the creation failed.
112  */
113 static struct lu_object *osp_object_alloc(const struct lu_env *env,
114                                           const struct lu_object_header *hdr,
115                                           struct lu_device *d)
116 {
117         struct lu_object_header *h = NULL;
118         struct osp_object       *o;
119         struct lu_object        *l;
120
121         OBD_SLAB_ALLOC_PTR_GFP(o, osp_object_kmem, GFP_NOFS);
122         if (o != NULL) {
123                 l = &o->opo_obj.do_lu;
124
125                 /* If hdr is NULL, it means the object is not built
126                  * from the top dev(MDT/OST), usually it happens when
127                  * building striped object, like data object on MDT or
128                  * striped object for directory */
129                 if (hdr == NULL) {
130                         h = &o->opo_header;
131                         lu_object_header_init(h);
132                         dt_object_init(&o->opo_obj, h, d);
133                         lu_object_add_top(h, l);
134                 } else {
135                         dt_object_init(&o->opo_obj, h, d);
136                 }
137
138                 l->lo_ops = &osp_lu_obj_ops;
139
140                 return l;
141         } else {
142                 return NULL;
143         }
144 }
145
146 /**
147  * Find or create the local object
148  *
149  * Finds or creates the local file referenced by \a reg_id and return the
150  * attributes of the local file.
151  *
152  * \param[in] env       execution environment
153  * \param[in] osp       OSP device
154  * \param[out] attr     attributes of the object
155  * \param[in] reg_id    the local object ID of the file. It will be used
156  *                      to compose a local FID{FID_SEQ_LOCAL_FILE, reg_id, 0}
157  *                      to identify the object.
158  *
159  * \retval object               object(dt_object) found or created
160  * \retval ERR_PTR(errno)       ERR_PTR(errno) if not get the object.
161  */
162 static struct dt_object
163 *osp_find_or_create_local_file(const struct lu_env *env, struct osp_device *osp,
164                                struct lu_attr *attr, __u32 reg_id)
165 {
166         struct osp_thread_info *osi = osp_env_info(env);
167         struct dt_object_format dof = { 0 };
168         struct dt_object       *dto;
169         int                  rc;
170         ENTRY;
171
172         lu_local_obj_fid(&osi->osi_fid, reg_id);
173         attr->la_valid = LA_MODE;
174         attr->la_mode = S_IFREG | 0644;
175         dof.dof_type = DFT_REGULAR;
176         /* Find or create the local object by osi_fid. */
177         dto = dt_find_or_create(env, osp->opd_storage, &osi->osi_fid,
178                                 &dof, attr);
179         if (IS_ERR(dto))
180                 RETURN(dto);
181
182         /* Get attributes of the local object. */
183         rc = dt_attr_get(env, dto, attr);
184         if (rc) {
185                 CERROR("%s: can't be initialized: rc = %d\n",
186                        osp->opd_obd->obd_name, rc);
187                 lu_object_put(env, &dto->do_lu);
188                 RETURN(ERR_PTR(rc));
189         }
190         RETURN(dto);
191 }
192
193 /**
194  * Write data buffer to a local file object.
195  *
196  * \param[in] env       execution environment
197  * \param[in] osp       OSP device
198  * \param[in] dt_obj    object written to
199  * \param[in] buf       buffer containing byte array and length
200  * \param[in] offset    write offset in the object in bytes
201  *
202  * \retval 0            0 if write succeed
203  * \retval -EFAULT      -EFAULT if only part of buffer is written.
204  * \retval negative             other negative errno if write failed.
205  */
206 static int osp_write_local_file(const struct lu_env *env,
207                                 struct osp_device *osp,
208                                 struct dt_object *dt_obj,
209                                 struct lu_buf *buf,
210                                 loff_t offset)
211 {
212         struct thandle *th;
213         int rc;
214
215         th = dt_trans_create(env, osp->opd_storage);
216         if (IS_ERR(th))
217                 RETURN(PTR_ERR(th));
218
219         rc = dt_declare_record_write(env, dt_obj, buf, offset, th);
220         if (rc)
221                 GOTO(out, rc);
222         rc = dt_trans_start_local(env, osp->opd_storage, th);
223         if (rc)
224                 GOTO(out, rc);
225
226         rc = dt_record_write(env, dt_obj, buf, &offset, th);
227 out:
228         dt_trans_stop(env, osp->opd_storage, th);
229         RETURN(rc);
230 }
231
232 /**
233  * Initialize last ID object.
234  *
235  * This function initializes the LAST_ID file, which stores the current last
236  * used id of data objects. The MDT will use the last used id and the last_seq
237  * (\see osp_init_last_seq()) to synchronize the precreate object cache with
238  * OSTs.
239  *
240  * \param[in] env       execution environment
241  * \param[in] osp       OSP device
242  *
243  * \retval 0            0 if initialization succeed
244  * \retval negative     negative errno if initialization failed
245  */
246 static int osp_init_last_objid(const struct lu_env *env, struct osp_device *osp)
247 {
248         struct osp_thread_info  *osi = osp_env_info(env);
249         struct lu_fid           *fid = &osp->opd_last_used_fid;
250         struct dt_object        *dto;
251         int                     rc = -EFAULT;
252         ENTRY;
253
254         dto = osp_find_or_create_local_file(env, osp, &osi->osi_attr,
255                                             MDD_LOV_OBJ_OID);
256         if (IS_ERR(dto))
257                 RETURN(PTR_ERR(dto));
258
259         /* object will be released in device cleanup path */
260         if (osi->osi_attr.la_size >=
261             sizeof(osi->osi_id) * (osp->opd_index + 1)) {
262                 osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_oid,
263                                    osp->opd_index);
264                 rc = dt_record_read(env, dto, &osi->osi_lb, &osi->osi_off);
265                 if (rc != 0 && rc != -EFAULT)
266                         GOTO(out, rc);
267         }
268
269         if (rc == -EFAULT) { /* fresh LAST_ID */
270                 fid->f_oid = 0;
271                 osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_oid,
272                                    osp->opd_index);
273                 rc = osp_write_local_file(env, osp, dto, &osi->osi_lb,
274                                           osi->osi_off);
275                 if (rc != 0)
276                         GOTO(out, rc);
277         }
278         osp->opd_last_used_oid_file = dto;
279         RETURN(0);
280 out:
281         /* object will be released in device cleanup path */
282         CERROR("%s: can't initialize lov_objid: rc = %d\n",
283                osp->opd_obd->obd_name, rc);
284         lu_object_put(env, &dto->do_lu);
285         osp->opd_last_used_oid_file = NULL;
286         RETURN(rc);
287 }
288
289 /**
290  * Initialize last sequence object.
291  *
292  * This function initializes the LAST_SEQ file in the local OSD, which stores
293  * the current last used sequence of data objects. The MDT will use the last
294  * sequence and last id (\see osp_init_last_objid()) to synchronize the
295  * precreate object cache with OSTs.
296  *
297  * \param[in] env       execution environment
298  * \param[in] osp       OSP device
299  *
300  * \retval 0            0 if initialization succeed
301  * \retval negative     negative errno if initialization failed
302  */
303 static int osp_init_last_seq(const struct lu_env *env, struct osp_device *osp)
304 {
305         struct osp_thread_info  *osi = osp_env_info(env);
306         struct lu_fid           *fid = &osp->opd_last_used_fid;
307         struct dt_object        *dto;
308         int                     rc = -EFAULT;
309         ENTRY;
310
311         dto = osp_find_or_create_local_file(env, osp, &osi->osi_attr,
312                                             MDD_LOV_OBJ_OSEQ);
313         if (IS_ERR(dto))
314                 RETURN(PTR_ERR(dto));
315
316         /* object will be released in device cleanup path */
317         if (osi->osi_attr.la_size >=
318             sizeof(osi->osi_id) * (osp->opd_index + 1)) {
319                 osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_seq,
320                                    osp->opd_index);
321                 rc = dt_record_read(env, dto, &osi->osi_lb, &osi->osi_off);
322                 if (rc != 0 && rc != -EFAULT)
323                         GOTO(out, rc);
324         }
325
326         if (rc == -EFAULT) { /* fresh OSP */
327                 fid->f_seq = 0;
328                 osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_seq,
329                                     osp->opd_index);
330                 rc = osp_write_local_file(env, osp, dto, &osi->osi_lb,
331                                           osi->osi_off);
332                 if (rc != 0)
333                         GOTO(out, rc);
334         }
335         osp->opd_last_used_seq_file = dto;
336         RETURN(0);
337 out:
338         /* object will be released in device cleanup path */
339         CERROR("%s: can't initialize lov_seq: rc = %d\n",
340                osp->opd_obd->obd_name, rc);
341         lu_object_put(env, &dto->do_lu);
342         osp->opd_last_used_seq_file = NULL;
343         RETURN(rc);
344 }
345
346 /**
347  * Initialize last OID and sequence object.
348  *
349  * If the MDT is just upgraded to 2.4 from the lower version, where the
350  * LAST_SEQ file does not exist, the file will be created and IDIF sequence
351  * will be written into the file.
352  *
353  * \param[in] env       execution environment
354  * \param[in] osp       OSP device
355  *
356  * \retval 0            0 if initialization succeed
357  * \retval negative     negative error if initialization failed
358  */
359 static int osp_last_used_init(const struct lu_env *env, struct osp_device *osp)
360 {
361         struct osp_thread_info *osi = osp_env_info(env);
362         int                  rc;
363         ENTRY;
364
365         fid_zero(&osp->opd_last_used_fid);
366         rc = osp_init_last_objid(env, osp);
367         if (rc < 0) {
368                 CERROR("%s: Can not get ids %d from old objid!\n",
369                        osp->opd_obd->obd_name, rc);
370                 RETURN(rc);
371         }
372
373         rc = osp_init_last_seq(env, osp);
374         if (rc < 0) {
375                 CERROR("%s: Can not get ids %d from old objid!\n",
376                        osp->opd_obd->obd_name, rc);
377                 GOTO(out, rc);
378         }
379
380         if (fid_oid(&osp->opd_last_used_fid) != 0 &&
381             fid_seq(&osp->opd_last_used_fid) == 0) {
382                 /* Just upgrade from the old version,
383                  * set the seq to be IDIF */
384                 osp->opd_last_used_fid.f_seq =
385                    fid_idif_seq(fid_oid(&osp->opd_last_used_fid),
386                                 osp->opd_index);
387                 osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off,
388                                     &osp->opd_last_used_fid.f_seq,
389                                     osp->opd_index);
390                 rc = osp_write_local_file(env, osp, osp->opd_last_used_seq_file,
391                                           &osi->osi_lb, osi->osi_off);
392                 if (rc) {
393                         CERROR("%s : Can not write seq file: rc = %d\n",
394                                osp->opd_obd->obd_name, rc);
395                         GOTO(out, rc);
396                 }
397         }
398
399         if (!fid_is_zero(&osp->opd_last_used_fid) &&
400                  !fid_is_sane(&osp->opd_last_used_fid)) {
401                 CERROR("%s: Got invalid FID "DFID"\n", osp->opd_obd->obd_name,
402                         PFID(&osp->opd_last_used_fid));
403                 GOTO(out, rc = -EINVAL);
404         }
405
406         CDEBUG(D_INFO, "%s: Init last used fid "DFID"\n",
407                osp->opd_obd->obd_name, PFID(&osp->opd_last_used_fid));
408 out:
409         if (rc != 0) {
410                 if (osp->opd_last_used_oid_file != NULL) {
411                         lu_object_put(env, &osp->opd_last_used_oid_file->do_lu);
412                         osp->opd_last_used_oid_file = NULL;
413                 }
414                 if (osp->opd_last_used_seq_file != NULL) {
415                         lu_object_put(env, &osp->opd_last_used_seq_file->do_lu);
416                         osp->opd_last_used_seq_file = NULL;
417                 }
418         }
419
420         RETURN(rc);
421 }
422
423 /**
424  * Release the last sequence and OID file objects in OSP device.
425  *
426  * \param[in] env       execution environment
427  * \param[in] osp       OSP device
428  */
429 static void osp_last_used_fini(const struct lu_env *env, struct osp_device *osp)
430 {
431         /* release last_used file */
432         if (osp->opd_last_used_oid_file != NULL) {
433                 lu_object_put(env, &osp->opd_last_used_oid_file->do_lu);
434                 osp->opd_last_used_oid_file = NULL;
435         }
436
437         if (osp->opd_last_used_seq_file != NULL) {
438                 lu_object_put(env, &osp->opd_last_used_seq_file->do_lu);
439                 osp->opd_last_used_seq_file = NULL;
440         }
441 }
442
443 /**
444  * Disconnects the connection between OSP and its correspondent MDT or OST, and
445  * the import will be marked as inactive. It will only be called during OSP
446  * cleanup process.
447  *
448  * \param[in] d         OSP device being disconnected
449  *
450  * \retval 0            0 if disconnection succeed
451  * \retval negative     negative errno if disconnection failed
452  */
453 static int osp_disconnect(struct osp_device *d)
454 {
455         struct obd_import *imp;
456         int rc = 0;
457
458         imp = d->opd_obd->u.cli.cl_import;
459
460         /* Mark import deactivated now, so we don't try to reconnect if any
461          * of the cleanup RPCs fails (e.g. ldlm cancel, etc).  We don't
462          * fully deactivate the import, or that would drop all requests. */
463         LASSERT(imp != NULL);
464         spin_lock(&imp->imp_lock);
465         imp->imp_deactive = 1;
466         spin_unlock(&imp->imp_lock);
467
468         ptlrpc_deactivate_import(imp);
469
470         /* Some non-replayable imports (MDS's OSCs) are pinged, so just
471          * delete it regardless.  (It's safe to delete an import that was
472          * never added.) */
473         (void)ptlrpc_pinger_del_import(imp);
474
475         rc = ptlrpc_disconnect_import(imp, 0);
476         if (rc != 0)
477                 CERROR("%s: can't disconnect: rc = %d\n",
478                        d->opd_obd->obd_name, rc);
479
480         ptlrpc_invalidate_import(imp);
481
482         RETURN(rc);
483 }
484
485 /**
486  * Cleanup OSP, which includes disconnect import, cleanup unlink log, stop
487  * precreate threads etc.
488  *
489  * \param[in] env       execution environment.
490  * \param[in] d         OSP device being disconnected.
491  *
492  * \retval 0            0 if cleanup succeed
493  * \retval negative     negative errno if cleanup failed
494  */
495 static int osp_shutdown(const struct lu_env *env, struct osp_device *d)
496 {
497         int                      rc = 0;
498         ENTRY;
499
500         LASSERT(env);
501
502         rc = osp_disconnect(d);
503
504         if (!d->opd_connect_mdt) {
505                 /* stop sync thread */
506                 osp_sync_fini(d);
507
508                 /* stop precreate thread */
509                 osp_precreate_fini(d);
510
511                 /* release last_used file */
512                 osp_last_used_fini(env, d);
513         }
514
515         obd_fid_fini(d->opd_obd);
516
517         RETURN(rc);
518 }
519
520 /**
521  * Implementation of osp_lu_ops::ldo_process_config
522  *
523  * This function processes config log records in OSP layer. It is usually
524  * called from the top layer of MDT stack, and goes through the stack by calling
525  * ldo_process_config of next layer.
526  *
527  * \param[in] env       execution environment
528  * \param[in] dev       lu_device of OSP
529  * \param[in] lcfg      config log
530  *
531  * \retval 0            0 if the config log record is executed correctly.
532  * \retval negative     negative errno if the record execution fails.
533  */
534 static int osp_process_config(const struct lu_env *env,
535                               struct lu_device *dev, struct lustre_cfg *lcfg)
536 {
537         struct osp_device               *d = lu2osp_dev(dev);
538         struct obd_device               *obd = d->opd_obd;
539         int                              rc;
540
541         ENTRY;
542
543         switch (lcfg->lcfg_command) {
544         case LCFG_PRE_CLEANUP:
545                 rc = osp_disconnect(d);
546                 break;
547         case LCFG_CLEANUP:
548                 lu_dev_del_linkage(dev->ld_site, dev);
549                 rc = osp_shutdown(env, d);
550                 break;
551         case LCFG_PARAM:
552                 LASSERT(obd);
553                 rc = class_process_proc_param(PARAM_OSC, obd->obd_vars,
554                                               lcfg, obd);
555                 if (rc > 0)
556                         rc = 0;
557                 if (rc == -ENOSYS) {
558                         /* class_process_proc_param() haven't found matching
559                          * parameter and returned ENOSYS so that layer(s)
560                          * below could use that. But OSP is the bottom, so
561                          * just ignore it */
562                         CERROR("%s: unknown param %s\n",
563                                (char *)lustre_cfg_string(lcfg, 0),
564                                (char *)lustre_cfg_string(lcfg, 1));
565                         rc = 0;
566                 }
567                 break;
568         default:
569                 CERROR("%s: unknown command %u\n",
570                        (char *)lustre_cfg_string(lcfg, 0), lcfg->lcfg_command);
571                 rc = 0;
572                 break;
573         }
574
575         RETURN(rc);
576 }
577
578 /**
579  * Implementation of osp_lu_ops::ldo_recovery_complete
580  *
581  * This function is called after recovery is finished, and OSP layer
582  * will wake up precreate thread here.
583  *
584  * \param[in] env       execution environment
585  * \param[in] dev       lu_device of OSP
586  *
587  * \retval 0            0 unconditionally
588  */
589 static int osp_recovery_complete(const struct lu_env *env,
590                                  struct lu_device *dev)
591 {
592         struct osp_device       *osp = lu2osp_dev(dev);
593
594         ENTRY;
595         osp->opd_recovery_completed = 1;
596
597         if (!osp->opd_connect_mdt && osp->opd_pre != NULL)
598                 wake_up(&osp->opd_pre_waitq);
599
600         RETURN(0);
601 }
602
603 const struct lu_device_operations osp_lu_ops = {
604         .ldo_object_alloc       = osp_object_alloc,
605         .ldo_process_config     = osp_process_config,
606         .ldo_recovery_complete  = osp_recovery_complete,
607 };
608
609 /**
610  * Implementation of dt_device_operations::dt_statfs
611  *
612  * This function provides statfs status (for precreation) from
613  * corresponding OST. Note: this function only retrieves the status
614  * from the OSP device, and the real statfs RPC happens inside
615  * precreate thread (\see osp_statfs_update). Note: OSP for MDT does
616  * not need to retrieve statfs data for now.
617  *
618  * \param[in] env       execution environment.
619  * \param[in] dev       dt_device of OSP.
620  * \param[out] sfs      holds the retrieved statfs data.
621  *
622  * \retval 0            0 statfs data was retrieved successfully or
623  *                      retrieval was not needed
624  * \retval negative     negative errno if get statfs failed.
625  */
626 static int osp_statfs(const struct lu_env *env, struct dt_device *dev,
627                       struct obd_statfs *sfs)
628 {
629         struct osp_device *d = dt2osp_dev(dev);
630
631         ENTRY;
632
633         if (unlikely(d->opd_imp_active == 0))
634                 RETURN(-ENOTCONN);
635
636         if (d->opd_pre == NULL)
637                 RETURN(0);
638
639         /* return recently updated data */
640         *sfs = d->opd_statfs;
641
642         /*
643          * layer above osp (usually lod) can use ffree to estimate
644          * how many objects are available for immediate creation
645          */
646         spin_lock(&d->opd_pre_lock);
647         LASSERTF(fid_seq(&d->opd_pre_last_created_fid) ==
648                  fid_seq(&d->opd_pre_used_fid),
649                  "last_created "DFID", next_fid "DFID"\n",
650                  PFID(&d->opd_pre_last_created_fid),
651                  PFID(&d->opd_pre_used_fid));
652         sfs->os_fprecreated = fid_oid(&d->opd_pre_last_created_fid) -
653                               fid_oid(&d->opd_pre_used_fid);
654         sfs->os_fprecreated -= d->opd_pre_reserved;
655         spin_unlock(&d->opd_pre_lock);
656
657         LASSERT(sfs->os_fprecreated <= OST_MAX_PRECREATE * 2);
658
659         CDEBUG(D_OTHER, "%s: "LPU64" blocks, "LPU64" free, "LPU64" avail, "
660                LPU64" files, "LPU64" free files\n", d->opd_obd->obd_name,
661                sfs->os_blocks, sfs->os_bfree, sfs->os_bavail,
662                sfs->os_files, sfs->os_ffree);
663         RETURN(0);
664 }
665
666 static int osp_sync_timeout(void *data)
667 {
668         return 1;
669 }
670
671 /**
672  * Implementation of dt_device_operations::dt_sync
673  *
674  * This function synchronizes the OSP cache to the remote target. It wakes
675  * up unlink log threads and sends out unlink records to the remote OST.
676  *
677  * \param[in] env       execution environment
678  * \param[in] dev       dt_device of OSP
679  *
680  * \retval 0            0 if synchronization succeeds
681  * \retval negative     negative errno if synchronization fails
682  */
683 static int osp_sync(const struct lu_env *env, struct dt_device *dev)
684 {
685         struct osp_device *d = dt2osp_dev(dev);
686         cfs_time_t         expire;
687         struct l_wait_info lwi = { 0 };
688         unsigned long      id, old;
689         int                rc = 0;
690         unsigned long      start = cfs_time_current();
691         ENTRY;
692
693         /* No Sync between MDTs yet. */
694         if (d->opd_connect_mdt)
695                 RETURN(0);
696
697         if (unlikely(d->opd_imp_active == 0))
698                 RETURN(-ENOTCONN);
699
700         id = d->opd_syn_last_used_id;
701         down_write(&d->opd_async_updates_rwsem);
702
703         CDEBUG(D_OTHER, "%s: async updates %d\n", d->opd_obd->obd_name,
704                atomic_read(&d->opd_async_updates_count));
705
706         /* make sure the connection is fine */
707         expire = cfs_time_shift(obd_timeout);
708         lwi = LWI_TIMEOUT(expire - cfs_time_current(), osp_sync_timeout, d);
709         rc = l_wait_event(d->opd_syn_barrier_waitq,
710                           atomic_read(&d->opd_async_updates_count) == 0,
711                           &lwi);
712         up_write(&d->opd_async_updates_rwsem);
713         if (rc != 0)
714                 GOTO(out, rc);
715
716         CDEBUG(D_CACHE, "%s: id: used %lu, processed %lu\n",
717                d->opd_obd->obd_name, id, d->opd_syn_last_processed_id);
718
719         /* wait till all-in-line are processed */
720         while (d->opd_syn_last_processed_id < id) {
721
722                 old = d->opd_syn_last_processed_id;
723
724                 /* make sure the connection is fine */
725                 expire = cfs_time_shift(obd_timeout);
726                 lwi = LWI_TIMEOUT(expire - cfs_time_current(),
727                                   osp_sync_timeout, d);
728                 l_wait_event(d->opd_syn_barrier_waitq,
729                              d->opd_syn_last_processed_id >= id,
730                              &lwi);
731
732                 if (d->opd_syn_last_processed_id >= id)
733                         break;
734
735                 if (d->opd_syn_last_processed_id != old) {
736                         /* some progress have been made,
737                          * keep trying... */
738                         continue;
739                 }
740
741                 /* no changes and expired, something is wrong */
742                 GOTO(out, rc = -ETIMEDOUT);
743         }
744
745         /* block new processing (barrier>0 - few callers are possible */
746         atomic_inc(&d->opd_syn_barrier);
747
748         CDEBUG(D_CACHE, "%s: %u in flight\n", d->opd_obd->obd_name,
749                d->opd_syn_rpc_in_flight);
750
751         /* wait till all-in-flight are replied, so executed by the target */
752         /* XXX: this is used by LFSCK at the moment, which doesn't require
753          *      all the changes to be committed, but in general it'd be
754          *      better to wait till commit */
755         while (d->opd_syn_rpc_in_flight > 0) {
756
757                 old = d->opd_syn_rpc_in_flight;
758
759                 expire = cfs_time_shift(obd_timeout);
760                 lwi = LWI_TIMEOUT(expire - cfs_time_current(),
761                                   osp_sync_timeout, d);
762                 l_wait_event(d->opd_syn_barrier_waitq,
763                                 d->opd_syn_rpc_in_flight == 0, &lwi);
764
765                 if (d->opd_syn_rpc_in_flight == 0)
766                         break;
767
768                 if (d->opd_syn_rpc_in_flight != old) {
769                         /* some progress have been made */
770                         continue;
771                 }
772
773                 /* no changes and expired, something is wrong */
774                 GOTO(out, rc = -ETIMEDOUT);
775         }
776
777 out:
778         /* resume normal processing (barrier=0) */
779         atomic_dec(&d->opd_syn_barrier);
780         __osp_sync_check_for_work(d);
781
782         CDEBUG(D_CACHE, "%s: done in %lu: rc = %d\n", d->opd_obd->obd_name,
783                cfs_time_current() - start, rc);
784
785         RETURN(rc);
786 }
787
788 const struct dt_device_operations osp_dt_ops = {
789         .dt_statfs       = osp_statfs,
790         .dt_sync         = osp_sync,
791         .dt_trans_create = osp_trans_create,
792         .dt_trans_start  = osp_trans_start,
793         .dt_trans_stop   = osp_trans_stop,
794 };
795
796 /**
797  * Connect OSP to local OSD.
798  *
799  * Locate the local OSD referenced by \a nextdev and connect to it. Sometimes,
800  * OSP needs to access the local OSD to store some information. For example,
801  * during precreate, it needs to update last used OID and sequence file
802  * (LAST_SEQ) in local OSD.
803  *
804  * \param[in] env       execution environment
805  * \param[in] osp       OSP device
806  * \param[in] nextdev   the name of local OSD
807  *
808  * \retval 0            0 connection succeeded
809  * \retval negative     negative errno connection failed
810  */
811 static int osp_connect_to_osd(const struct lu_env *env, struct osp_device *osp,
812                               const char *nextdev)
813 {
814         struct obd_connect_data *data = NULL;
815         struct obd_device       *obd;
816         int                      rc;
817
818         ENTRY;
819
820         LASSERT(osp->opd_storage_exp == NULL);
821
822         OBD_ALLOC_PTR(data);
823         if (data == NULL)
824                 RETURN(-ENOMEM);
825
826         obd = class_name2obd(nextdev);
827         if (obd == NULL) {
828                 CERROR("%s: can't locate next device: %s\n",
829                        osp->opd_obd->obd_name, nextdev);
830                 GOTO(out, rc = -ENOTCONN);
831         }
832
833         rc = obd_connect(env, &osp->opd_storage_exp, obd, &obd->obd_uuid, data,
834                          NULL);
835         if (rc) {
836                 CERROR("%s: cannot connect to next dev %s: rc = %d\n",
837                        osp->opd_obd->obd_name, nextdev, rc);
838                 GOTO(out, rc);
839         }
840
841         osp->opd_dt_dev.dd_lu_dev.ld_site =
842                 osp->opd_storage_exp->exp_obd->obd_lu_dev->ld_site;
843         LASSERT(osp->opd_dt_dev.dd_lu_dev.ld_site);
844         osp->opd_storage = lu2dt_dev(osp->opd_storage_exp->exp_obd->obd_lu_dev);
845
846 out:
847         OBD_FREE_PTR(data);
848         RETURN(rc);
849 }
850
851 /**
852  * Initialize OSP device according to the parameters in the configuration
853  * log \a cfg.
854  *
855  * Reconstruct the local device name from the configuration profile, and
856  * initialize necessary threads and structures according to the OSP type
857  * (MDT or OST).
858  *
859  * Since there is no record in the MDT configuration for the local disk
860  * device, we have to extract this from elsewhere in the profile.
861  * The only information we get at setup is from the OSC records:
862  * setup 0:{fsname}-OSTxxxx-osc[-MDTxxxx] 1:lustre-OST0000_UUID 2:NID
863  *
864  * Note: configs generated by Lustre 1.8 are missing the -MDTxxxx part,
865  * so, we need to reconstruct the name of the underlying OSD from this:
866  * {fsname}-{svname}-osd, for example "lustre-MDT0000-osd".
867  *
868  * \param[in] env       execution environment
869  * \param[in] osp       OSP device
870  * \param[in] ldt       lu device type of OSP
871  * \param[in] cfg       configuration log
872  *
873  * \retval 0            0 if OSP initialization succeeded.
874  * \retval negative     negative errno if OSP initialization failed.
875  */
876 static int osp_init0(const struct lu_env *env, struct osp_device *osp,
877                      struct lu_device_type *ldt, struct lustre_cfg *cfg)
878 {
879         struct obd_device       *obd;
880         struct obd_import       *imp;
881         class_uuid_t            uuid;
882         char                    *src, *tgt, *mdt, *osdname = NULL;
883         int                     rc;
884         long                    idx;
885
886         ENTRY;
887
888         mutex_init(&osp->opd_async_requests_mutex);
889         INIT_LIST_HEAD(&osp->opd_async_updates);
890         init_rwsem(&osp->opd_async_updates_rwsem);
891         atomic_set(&osp->opd_async_updates_count, 0);
892
893         obd = class_name2obd(lustre_cfg_string(cfg, 0));
894         if (obd == NULL) {
895                 CERROR("Cannot find obd with name %s\n",
896                        lustre_cfg_string(cfg, 0));
897                 RETURN(-ENODEV);
898         }
899         osp->opd_obd = obd;
900
901         src = lustre_cfg_string(cfg, 0);
902         if (src == NULL)
903                 RETURN(-EINVAL);
904
905         tgt = strrchr(src, '-');
906         if (tgt == NULL) {
907                 CERROR("%s: invalid target name %s: rc = %d\n",
908                        osp->opd_obd->obd_name, lustre_cfg_string(cfg, 0),
909                        -EINVAL);
910                 RETURN(-EINVAL);
911         }
912
913         if (strncmp(tgt, "-osc", 4) == 0) {
914                 /* Old OSC name fsname-OSTXXXX-osc */
915                 for (tgt--; tgt > src && *tgt != '-'; tgt--)
916                         ;
917                 if (tgt == src) {
918                         CERROR("%s: invalid target name %s: rc = %d\n",
919                                osp->opd_obd->obd_name,
920                                lustre_cfg_string(cfg, 0), -EINVAL);
921                         RETURN(-EINVAL);
922                 }
923
924                 if (strncmp(tgt, "-OST", 4) != 0) {
925                         CERROR("%s: invalid target name %s: rc = %d\n",
926                                osp->opd_obd->obd_name,
927                                lustre_cfg_string(cfg, 0), -EINVAL);
928                         RETURN(-EINVAL);
929                 }
930
931                 idx = simple_strtol(tgt + 4, &mdt, 16);
932                 if (mdt[0] != '-' || idx > INT_MAX || idx < 0) {
933                         CERROR("%s: invalid OST index in '%s': rc = %d\n",
934                                osp->opd_obd->obd_name, src, -EINVAL);
935                         RETURN(-EINVAL);
936                 }
937                 osp->opd_index = idx;
938                 osp->opd_group = 0;
939                 idx = tgt - src;
940         } else {
941                 /* New OSC name fsname-OSTXXXX-osc-MDTXXXX */
942                 if (strncmp(tgt, "-MDT", 4) != 0 &&
943                     strncmp(tgt, "-OST", 4) != 0) {
944                         CERROR("%s: invalid target name %s: rc = %d\n",
945                                osp->opd_obd->obd_name,
946                                lustre_cfg_string(cfg, 0), -EINVAL);
947                         RETURN(-EINVAL);
948                 }
949
950                 idx = simple_strtol(tgt + 4, &mdt, 16);
951                 if (*mdt != '\0' || idx > INT_MAX || idx < 0) {
952                         CERROR("%s: invalid OST index in '%s': rc = %d\n",
953                                osp->opd_obd->obd_name, src, -EINVAL);
954                         RETURN(-EINVAL);
955                 }
956
957                 /* Get MDT index from the name and set it to opd_group,
958                  * which will be used by OSP to connect with OST */
959                 osp->opd_group = idx;
960                 if (tgt - src <= 12) {
961                         CERROR("%s: invalid mdt index from %s: rc =%d\n",
962                                osp->opd_obd->obd_name,
963                                lustre_cfg_string(cfg, 0), -EINVAL);
964                         RETURN(-EINVAL);
965                 }
966
967                 if (strncmp(tgt - 12, "-MDT", 4) == 0)
968                         osp->opd_connect_mdt = 1;
969
970                 idx = simple_strtol(tgt - 8, &mdt, 16);
971                 if (mdt[0] != '-' || idx > INT_MAX || idx < 0) {
972                         CERROR("%s: invalid OST index in '%s': rc =%d\n",
973                                osp->opd_obd->obd_name, src, -EINVAL);
974                         RETURN(-EINVAL);
975                 }
976
977                 osp->opd_index = idx;
978                 idx = tgt - src - 12;
979         }
980         /* check the fsname length, and after this everything else will fit */
981         if (idx > MTI_NAME_MAXLEN) {
982                 CERROR("%s: fsname too long in '%s': rc = %d\n",
983                        osp->opd_obd->obd_name, src, -EINVAL);
984                 RETURN(-EINVAL);
985         }
986
987         OBD_ALLOC(osdname, MAX_OBD_NAME);
988         if (osdname == NULL)
989                 RETURN(-ENOMEM);
990
991         memcpy(osdname, src, idx); /* copy just the fsname part */
992         osdname[idx] = '\0';
993
994         mdt = strstr(mdt, "-MDT");
995         if (mdt == NULL) /* 1.8 configs don't have "-MDT0000" at the end */
996                 strcat(osdname, "-MDT0000");
997         else
998                 strcat(osdname, mdt);
999         strcat(osdname, "-osd");
1000         CDEBUG(D_HA, "%s: connect to %s (%s)\n", obd->obd_name, osdname, src);
1001
1002         if (osp->opd_connect_mdt) {
1003                 struct client_obd *cli = &osp->opd_obd->u.cli;
1004
1005                 OBD_ALLOC(cli->cl_rpc_lock, sizeof(*cli->cl_rpc_lock));
1006                 if (!cli->cl_rpc_lock)
1007                         GOTO(out_fini, rc = -ENOMEM);
1008                 osp_init_rpc_lock(cli->cl_rpc_lock);
1009         }
1010
1011         osp->opd_dt_dev.dd_lu_dev.ld_ops = &osp_lu_ops;
1012         osp->opd_dt_dev.dd_ops = &osp_dt_ops;
1013
1014         obd->obd_lu_dev = &osp->opd_dt_dev.dd_lu_dev;
1015
1016         rc = osp_connect_to_osd(env, osp, osdname);
1017         if (rc)
1018                 GOTO(out_fini, rc);
1019
1020         rc = ptlrpcd_addref();
1021         if (rc)
1022                 GOTO(out_disconnect, rc);
1023
1024         rc = client_obd_setup(obd, cfg);
1025         if (rc) {
1026                 CERROR("%s: can't setup obd: rc = %d\n", osp->opd_obd->obd_name,
1027                        rc);
1028                 GOTO(out_ref, rc);
1029         }
1030
1031         osp_lprocfs_init(osp);
1032
1033         rc = obd_fid_init(osp->opd_obd, NULL, osp->opd_connect_mdt ?
1034                           LUSTRE_SEQ_METADATA : LUSTRE_SEQ_DATA);
1035         if (rc) {
1036                 CERROR("%s: fid init error: rc = %d\n",
1037                        osp->opd_obd->obd_name, rc);
1038                 GOTO(out_proc, rc);
1039         }
1040
1041         if (!osp->opd_connect_mdt) {
1042                 /* Initialize last id from the storage - will be
1043                  * used in orphan cleanup. */
1044                 rc = osp_last_used_init(env, osp);
1045                 if (rc)
1046                         GOTO(out_fid, rc);
1047
1048
1049                 /* Initialize precreation thread, it handles new
1050                  * connections as well. */
1051                 rc = osp_init_precreate(osp);
1052                 if (rc)
1053                         GOTO(out_last_used, rc);
1054
1055                 /*
1056                  * Initialize synhronization mechanism taking
1057                  * care of propogating changes to OST in near
1058                  * transactional manner.
1059                  */
1060                 rc = osp_sync_init(env, osp);
1061                 if (rc < 0)
1062                         GOTO(out_precreat, rc);
1063         }
1064
1065         /*
1066          * Initiate connect to OST
1067          */
1068         ll_generate_random_uuid(uuid);
1069         class_uuid_unparse(uuid, &osp->opd_cluuid);
1070
1071         imp = obd->u.cli.cl_import;
1072
1073         rc = ptlrpc_init_import(imp);
1074         if (rc)
1075                 GOTO(out, rc);
1076         if (osdname)
1077                 OBD_FREE(osdname, MAX_OBD_NAME);
1078         RETURN(0);
1079
1080 out:
1081         if (!osp->opd_connect_mdt)
1082                 /* stop sync thread */
1083                 osp_sync_fini(osp);
1084 out_precreat:
1085         /* stop precreate thread */
1086         if (!osp->opd_connect_mdt)
1087                 osp_precreate_fini(osp);
1088 out_last_used:
1089         if (!osp->opd_connect_mdt)
1090                 osp_last_used_fini(env, osp);
1091 out_fid:
1092         obd_fid_fini(osp->opd_obd);
1093 out_proc:
1094         ptlrpc_lprocfs_unregister_obd(obd);
1095         lprocfs_obd_cleanup(obd);
1096         if (osp->opd_symlink)
1097                 lprocfs_remove(&osp->opd_symlink);
1098         client_obd_cleanup(obd);
1099 out_ref:
1100         ptlrpcd_decref();
1101 out_disconnect:
1102         if (osp->opd_connect_mdt) {
1103                 struct client_obd *cli = &osp->opd_obd->u.cli;
1104                 if (cli->cl_rpc_lock != NULL) {
1105                         OBD_FREE_PTR(cli->cl_rpc_lock);
1106                         cli->cl_rpc_lock = NULL;
1107                 }
1108         }
1109         obd_disconnect(osp->opd_storage_exp);
1110 out_fini:
1111         if (osdname)
1112                 OBD_FREE(osdname, MAX_OBD_NAME);
1113         RETURN(rc);
1114 }
1115
1116 /**
1117  * Implementation of lu_device_type_operations::ldto_device_free
1118  *
1119  * Free the OSP device in memory.  No return value is needed for now,
1120  * so always return NULL to comply with the interface.
1121  *
1122  * \param[in] env       execution environment
1123  * \param[in] lu        lu_device of OSP
1124  *
1125  * \retval NULL         NULL unconditionally
1126  */
1127 static struct lu_device *osp_device_free(const struct lu_env *env,
1128                                          struct lu_device *lu)
1129 {
1130         struct osp_device *osp = lu2osp_dev(lu);
1131
1132         if (atomic_read(&lu->ld_ref) && lu->ld_site) {
1133                 LIBCFS_DEBUG_MSG_DATA_DECL(msgdata, D_ERROR, NULL);
1134                 lu_site_print(env, lu->ld_site, &msgdata, lu_cdebug_printer);
1135         }
1136         dt_device_fini(&osp->opd_dt_dev);
1137         OBD_FREE_PTR(osp);
1138
1139         return NULL;
1140 }
1141
1142 /**
1143  * Implementation of lu_device_type_operations::ldto_device_alloc
1144  *
1145  * This function allocates and initializes OSP device in memory according to
1146  * the config log.
1147  *
1148  * \param[in] env       execution environment
1149  * \param[in] type      device type of OSP
1150  * \param[in] lcfg      config log
1151  *
1152  * \retval pointer              the pointer of allocated OSP if succeed.
1153  * \retval ERR_PTR(errno)       ERR_PTR(errno) if failed.
1154  */
1155 static struct lu_device *osp_device_alloc(const struct lu_env *env,
1156                                           struct lu_device_type *type,
1157                                           struct lustre_cfg *lcfg)
1158 {
1159         struct osp_device *osp;
1160         struct lu_device  *ld;
1161
1162         OBD_ALLOC_PTR(osp);
1163         if (osp == NULL) {
1164                 ld = ERR_PTR(-ENOMEM);
1165         } else {
1166                 int rc;
1167
1168                 ld = osp2lu_dev(osp);
1169                 dt_device_init(&osp->opd_dt_dev, type);
1170                 rc = osp_init0(env, osp, type, lcfg);
1171                 if (rc != 0) {
1172                         osp_device_free(env, ld);
1173                         ld = ERR_PTR(rc);
1174                 }
1175         }
1176         return ld;
1177 }
1178
1179 /**
1180  * Implementation of lu_device_type_operations::ldto_device_fini
1181  *
1182  * This function cleans up the OSP device, i.e. release and free those
1183  * attached items in osp_device.
1184  *
1185  * \param[in] env       execution environment
1186  * \param[in] ld        lu_device of OSP
1187  *
1188  * \retval NULL                 NULL if cleanup succeeded.
1189  * \retval ERR_PTR(errno)       ERR_PTR(errno) if cleanup failed.
1190  */
1191 static struct lu_device *osp_device_fini(const struct lu_env *env,
1192                                          struct lu_device *ld)
1193 {
1194         struct osp_device *osp = lu2osp_dev(ld);
1195         struct obd_import *imp;
1196         int                rc;
1197
1198         ENTRY;
1199
1200         if (osp->opd_async_requests != NULL) {
1201                 dt_update_request_destroy(osp->opd_async_requests);
1202                 osp->opd_async_requests = NULL;
1203         }
1204
1205         if (osp->opd_storage_exp)
1206                 obd_disconnect(osp->opd_storage_exp);
1207
1208         imp = osp->opd_obd->u.cli.cl_import;
1209
1210         if (imp->imp_rq_pool) {
1211                 ptlrpc_free_rq_pool(imp->imp_rq_pool);
1212                 imp->imp_rq_pool = NULL;
1213         }
1214
1215         if (osp->opd_symlink)
1216                 lprocfs_remove(&osp->opd_symlink);
1217
1218         LASSERT(osp->opd_obd);
1219         ptlrpc_lprocfs_unregister_obd(osp->opd_obd);
1220         lprocfs_obd_cleanup(osp->opd_obd);
1221
1222         if (osp->opd_connect_mdt) {
1223                 struct client_obd *cli = &osp->opd_obd->u.cli;
1224                 if (cli->cl_rpc_lock != NULL) {
1225                         OBD_FREE_PTR(cli->cl_rpc_lock);
1226                         cli->cl_rpc_lock = NULL;
1227                 }
1228         }
1229
1230         rc = client_obd_cleanup(osp->opd_obd);
1231         if (rc != 0) {
1232                 ptlrpcd_decref();
1233                 RETURN(ERR_PTR(rc));
1234         }
1235
1236         ptlrpcd_decref();
1237
1238         RETURN(NULL);
1239 }
1240
1241 /**
1242  * Implementation of obd_ops::o_reconnect
1243  *
1244  * This function is empty and does not need to do anything for now.
1245  */
1246 static int osp_reconnect(const struct lu_env *env,
1247                          struct obd_export *exp, struct obd_device *obd,
1248                          struct obd_uuid *cluuid,
1249                          struct obd_connect_data *data,
1250                          void *localdata)
1251 {
1252         return 0;
1253 }
1254
1255 /*
1256  * Implementation of obd_ops::o_connect
1257  *
1258  * Connect OSP to the remote target (MDT or OST). Allocate the
1259  * export and return it to the LOD, which calls this function
1260  * for each OSP to connect it to the remote target. This function
1261  * is currently only called once per OSP.
1262  *
1263  * \param[in] env       execution environment
1264  * \param[out] exp      export connected to OSP
1265  * \param[in] obd       OSP device
1266  * \param[in] cluuid    OSP device client uuid
1267  * \param[in] data      connect_data to be used to connect to the remote
1268  *                      target
1269  * \param[in] localdata necessary for the API interface, but not used in
1270  *                      this function
1271  *
1272  * \retval 0            0 if the connection succeeded.
1273  * \retval negative     negative errno if the connection failed.
1274  */
1275 static int osp_obd_connect(const struct lu_env *env, struct obd_export **exp,
1276                            struct obd_device *obd, struct obd_uuid *cluuid,
1277                            struct obd_connect_data *data, void *localdata)
1278 {
1279         struct osp_device       *osp = lu2osp_dev(obd->obd_lu_dev);
1280         struct obd_connect_data *ocd;
1281         struct obd_import       *imp;
1282         struct lustre_handle     conn;
1283         int                      rc;
1284
1285         ENTRY;
1286
1287         CDEBUG(D_CONFIG, "connect #%d\n", osp->opd_connects);
1288
1289         rc = class_connect(&conn, obd, cluuid);
1290         if (rc)
1291                 RETURN(rc);
1292
1293         *exp = class_conn2export(&conn);
1294         /* Why should there ever be more than 1 connect? */
1295         osp->opd_connects++;
1296         LASSERT(osp->opd_connects == 1);
1297
1298         osp->opd_exp = *exp;
1299
1300         imp = osp->opd_obd->u.cli.cl_import;
1301         imp->imp_dlm_handle = conn;
1302
1303         LASSERT(data != NULL);
1304         LASSERT(data->ocd_connect_flags & OBD_CONNECT_INDEX);
1305         ocd = &imp->imp_connect_data;
1306         *ocd = *data;
1307
1308         imp->imp_connect_flags_orig = ocd->ocd_connect_flags;
1309
1310         ocd->ocd_version = LUSTRE_VERSION_CODE;
1311         ocd->ocd_index = data->ocd_index;
1312         imp->imp_connect_flags_orig = ocd->ocd_connect_flags;
1313
1314         rc = ptlrpc_connect_import(imp);
1315         if (rc) {
1316                 CERROR("%s: can't connect obd: rc = %d\n", obd->obd_name, rc);
1317                 GOTO(out, rc);
1318         }
1319
1320         ptlrpc_pinger_add_import(imp);
1321 out:
1322         RETURN(rc);
1323 }
1324
1325 /**
1326  * Implementation of obd_ops::o_disconnect
1327  *
1328  * Disconnect the export for the OSP.  This is called by LOD to release the
1329  * OSP during cleanup (\see lod_del_device()). The OSP will be released after
1330  * the export is released.
1331  *
1332  * \param[in] exp       export to be disconnected.
1333  *
1334  * \retval 0            0 if disconnection succeed
1335  * \retval negative     negative errno if disconnection failed
1336  */
1337 static int osp_obd_disconnect(struct obd_export *exp)
1338 {
1339         struct obd_device *obd = exp->exp_obd;
1340         struct osp_device *osp = lu2osp_dev(obd->obd_lu_dev);
1341         int                rc;
1342         ENTRY;
1343
1344         /* Only disconnect the underlying layers on the final disconnect. */
1345         LASSERT(osp->opd_connects == 1);
1346         osp->opd_connects--;
1347
1348         rc = class_disconnect(exp);
1349         if (rc) {
1350                 CERROR("%s: class disconnect error: rc = %d\n",
1351                        obd->obd_name, rc);
1352                 RETURN(rc);
1353         }
1354
1355         /* destroy the device */
1356         class_manual_cleanup(obd);
1357
1358         RETURN(rc);
1359 }
1360
1361 /**
1362  * Implementation of obd_ops::o_statfs
1363  *
1364  * Send a RPC to the remote target to get statfs status. This is only used
1365  * in lprocfs helpers by obd_statfs.
1366  *
1367  * \param[in] env       execution environment
1368  * \param[in] exp       connection state from this OSP to the parent (LOD)
1369  *                      device
1370  * \param[out] osfs     hold the statfs result
1371  * \param[in] unused    Not used in this function for now
1372  * \param[in] flags     flags to indicate how OSP will issue the RPC
1373  *
1374  * \retval 0            0 if statfs succeeded.
1375  * \retval negative     negative errno if statfs failed.
1376  */
1377 static int osp_obd_statfs(const struct lu_env *env, struct obd_export *exp,
1378                           struct obd_statfs *osfs, __u64 unused, __u32 flags)
1379 {
1380         struct obd_statfs       *msfs;
1381         struct ptlrpc_request   *req;
1382         struct obd_import       *imp = NULL;
1383         int                      rc;
1384
1385         ENTRY;
1386
1387         /* Since the request might also come from lprocfs, so we need
1388          * sync this with client_disconnect_export Bug15684 */
1389         down_read(&exp->exp_obd->u.cli.cl_sem);
1390         if (exp->exp_obd->u.cli.cl_import)
1391                 imp = class_import_get(exp->exp_obd->u.cli.cl_import);
1392         up_read(&exp->exp_obd->u.cli.cl_sem);
1393         if (!imp)
1394                 RETURN(-ENODEV);
1395
1396         req = ptlrpc_request_alloc(imp, &RQF_OST_STATFS);
1397
1398         class_import_put(imp);
1399
1400         if (req == NULL)
1401                 RETURN(-ENOMEM);
1402
1403         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
1404         if (rc) {
1405                 ptlrpc_request_free(req);
1406                 RETURN(rc);
1407         }
1408         ptlrpc_request_set_replen(req);
1409         req->rq_request_portal = OST_CREATE_PORTAL;
1410         ptlrpc_at_set_req_timeout(req);
1411
1412         if (flags & OBD_STATFS_NODELAY) {
1413                 /* procfs requests not want stat in wait for avoid deadlock */
1414                 req->rq_no_resend = 1;
1415                 req->rq_no_delay = 1;
1416         }
1417
1418         rc = ptlrpc_queue_wait(req);
1419         if (rc)
1420                 GOTO(out, rc);
1421
1422         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
1423         if (msfs == NULL)
1424                 GOTO(out, rc = -EPROTO);
1425
1426         *osfs = *msfs;
1427
1428         EXIT;
1429 out:
1430         ptlrpc_req_finished(req);
1431         return rc;
1432 }
1433
1434 /**
1435  * Prepare fid client.
1436  *
1437  * This function prepares the FID client for the OSP. It will check and assign
1438  * the export (to MDT0) for its FID client, so OSP can allocate super sequence
1439  * or lookup sequence in FLDB of MDT0.
1440  *
1441  * \param[in] osp       OSP device
1442  */
1443 static void osp_prepare_fid_client(struct osp_device *osp)
1444 {
1445         LASSERT(osp->opd_obd->u.cli.cl_seq != NULL);
1446         if (osp->opd_obd->u.cli.cl_seq->lcs_exp != NULL)
1447                 return;
1448
1449         LASSERT(osp->opd_exp != NULL);
1450         osp->opd_obd->u.cli.cl_seq->lcs_exp =
1451                                 class_export_get(osp->opd_exp);
1452 }
1453
1454 /**
1455  * Implementation of obd_ops::o_import_event
1456  *
1457  * This function is called when some related import event happens. It will
1458  * mark the necessary flags according to the event and notify the necessary
1459  * threads (mainly precreate thread).
1460  *
1461  * \param[in] obd       OSP OBD device
1462  * \param[in] imp       import attached from OSP to remote (OST/MDT) service
1463  * \param[in] event     event related to remote service (IMP_EVENT_*)
1464  *
1465  * \retval 0            0 if the event handling succeeded.
1466  * \retval negative     negative errno if the event handling failed.
1467  */
1468 static int osp_import_event(struct obd_device *obd, struct obd_import *imp,
1469                             enum obd_import_event event)
1470 {
1471         struct osp_device *d = lu2osp_dev(obd->obd_lu_dev);
1472
1473         switch (event) {
1474         case IMP_EVENT_DISCON:
1475                 d->opd_got_disconnected = 1;
1476                 d->opd_imp_connected = 0;
1477                 if (d->opd_connect_mdt)
1478                         break;
1479
1480                 if (d->opd_pre != NULL) {
1481                         osp_pre_update_status(d, -ENODEV);
1482                         wake_up(&d->opd_pre_waitq);
1483                 }
1484
1485                 CDEBUG(D_HA, "got disconnected\n");
1486                 break;
1487         case IMP_EVENT_INACTIVE:
1488                 d->opd_imp_active = 0;
1489                 if (d->opd_connect_mdt)
1490                         break;
1491
1492                 if (d->opd_pre != NULL) {
1493                         osp_pre_update_status(d, -ENODEV);
1494                         wake_up(&d->opd_pre_waitq);
1495                 }
1496
1497                 CDEBUG(D_HA, "got inactive\n");
1498                 break;
1499         case IMP_EVENT_ACTIVE:
1500                 d->opd_imp_active = 1;
1501
1502                 osp_prepare_fid_client(d);
1503                 if (d->opd_got_disconnected)
1504                         d->opd_new_connection = 1;
1505                 d->opd_imp_connected = 1;
1506                 d->opd_imp_seen_connected = 1;
1507                 if (d->opd_connect_mdt)
1508                         break;
1509
1510                 if (d->opd_pre != NULL)
1511                         wake_up(&d->opd_pre_waitq);
1512
1513                 __osp_sync_check_for_work(d);
1514                 CDEBUG(D_HA, "got connected\n");
1515                 break;
1516         case IMP_EVENT_INVALIDATE:
1517                 if (obd->obd_namespace == NULL)
1518                         break;
1519                 ldlm_namespace_cleanup(obd->obd_namespace, LDLM_FL_LOCAL_ONLY);
1520                 break;
1521         case IMP_EVENT_OCD:
1522         case IMP_EVENT_DEACTIVATE:
1523         case IMP_EVENT_ACTIVATE:
1524                 break;
1525         default:
1526                 CERROR("%s: unsupported import event: %#x\n",
1527                        obd->obd_name, event);
1528         }
1529         return 0;
1530 }
1531
1532 /**
1533  * Implementation of obd_ops: o_iocontrol
1534  *
1535  * This function is the ioctl handler for OSP. Note: lctl will access the OSP
1536  * directly by ioctl, instead of through the MDS stack.
1537  *
1538  * param[in] cmd        ioctl command.
1539  * param[in] exp        export of this OSP.
1540  * param[in] len        data length of \a karg.
1541  * param[in] karg       input argument which is packed as
1542  *                      obd_ioctl_data
1543  * param[out] uarg      pointer to userspace buffer (must access by
1544  *                      copy_to_user()).
1545  *
1546  * \retval 0            0 if the ioctl handling succeeded.
1547  * \retval negative     negative errno if the ioctl handling failed.
1548  */
1549 static int osp_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
1550                          void *karg, void *uarg)
1551 {
1552         struct obd_device       *obd = exp->exp_obd;
1553         struct osp_device       *d;
1554         struct obd_ioctl_data   *data = karg;
1555         int                      rc = 0;
1556
1557         ENTRY;
1558
1559         LASSERT(obd->obd_lu_dev);
1560         d = lu2osp_dev(obd->obd_lu_dev);
1561         LASSERT(d->opd_dt_dev.dd_ops == &osp_dt_ops);
1562
1563         if (!try_module_get(THIS_MODULE)) {
1564                 CERROR("%s: cannot get module '%s'\n", obd->obd_name,
1565                        module_name(THIS_MODULE));
1566                 return -EINVAL;
1567         }
1568
1569         switch (cmd) {
1570         case OBD_IOC_CLIENT_RECOVER:
1571                 rc = ptlrpc_recover_import(obd->u.cli.cl_import,
1572                                            data->ioc_inlbuf1, 0);
1573                 if (rc > 0)
1574                         rc = 0;
1575                 break;
1576         case IOC_OSC_SET_ACTIVE:
1577                 rc = ptlrpc_set_import_active(obd->u.cli.cl_import,
1578                                               data->ioc_offset);
1579                 break;
1580         case OBD_IOC_PING_TARGET:
1581                 rc = ptlrpc_obd_ping(obd);
1582                 break;
1583         default:
1584                 CERROR("%s: unrecognized ioctl %#x by %s\n", obd->obd_name,
1585                        cmd, current_comm());
1586                 rc = -ENOTTY;
1587         }
1588         module_put(THIS_MODULE);
1589         return rc;
1590 }
1591
1592 /**
1593  * Implementation of obd_ops::o_get_info
1594  *
1595  * Retrieve information by key. Retrieval starts from the top layer
1596  * (MDT) of the MDS stack and traverses the stack by calling the
1597  * obd_get_info() method of the next sub-layer.
1598  *
1599  * \param[in] env       execution environment
1600  * \param[in] exp       export of this OSP
1601  * \param[in] keylen    length of \a key
1602  * \param[in] key       the key
1603  * \param[out] vallen   length of \a val
1604  * \param[out] val      holds the value returned by the key
1605  * \param[in] unused    necessary for the interface but unused
1606  *
1607  * \retval 0            0 if getting information succeeded.
1608  * \retval negative     negative errno if getting information failed.
1609  */
1610 static int osp_obd_get_info(const struct lu_env *env, struct obd_export *exp,
1611                             __u32 keylen, void *key, __u32 *vallen, void *val,
1612                             struct lov_stripe_md *unused)
1613 {
1614         int rc = -EINVAL;
1615
1616         if (KEY_IS(KEY_OSP_CONNECTED)) {
1617                 struct obd_device       *obd = exp->exp_obd;
1618                 struct osp_device       *osp;
1619
1620                 if (!obd->obd_set_up || obd->obd_stopping)
1621                         RETURN(-EAGAIN);
1622
1623                 osp = lu2osp_dev(obd->obd_lu_dev);
1624                 LASSERT(osp);
1625                 /*
1626                  * 1.8/2.0 behaviour is that OST being connected once at least
1627                  * is considered "healthy". and one "healthy" OST is enough to
1628                  * allow lustre clients to connect to MDS
1629                  */
1630                 RETURN(!osp->opd_imp_seen_connected);
1631         }
1632
1633         RETURN(rc);
1634 }
1635
1636 /**
1637  * Implementation of obd_ops: o_fid_alloc
1638  *
1639  * Allocate a FID. There are two cases in which OSP performs
1640  * FID allocation.
1641  *
1642  * 1. FID precreation for data objects, which is done in
1643  *    osp_precreate_fids() instead of this function.
1644  * 2. FID allocation for each sub-stripe of a striped directory.
1645  *    Similar to other FID clients, OSP requests the sequence
1646  *    from its corresponding remote MDT, which in turn requests
1647  *    sequences from the sequence controller (MDT0).
1648  *
1649  * \param[in] env       execution environment
1650  * \param[in] exp       export of the OSP
1651  * \param[out] fid      FID being allocated
1652  * \param[in] unused    necessary for the interface but unused.
1653  *
1654  * \retval 0            0 FID allocated successfully.
1655  * \retval 1            1 FID allocated successfully and new sequence
1656  *                      requested from seq meta server
1657  * \retval negative     negative errno if FID allocation failed.
1658  */
1659 static int osp_fid_alloc(const struct lu_env *env, struct obd_export *exp,
1660                          struct lu_fid *fid, struct md_op_data *unused)
1661 {
1662         struct client_obd       *cli = &exp->exp_obd->u.cli;
1663         struct osp_device       *osp = lu2osp_dev(exp->exp_obd->obd_lu_dev);
1664         struct lu_client_seq    *seq = cli->cl_seq;
1665         ENTRY;
1666
1667         LASSERT(osp->opd_obd->u.cli.cl_seq != NULL);
1668         /* Sigh, fid client is not ready yet */
1669         if (osp->opd_obd->u.cli.cl_seq->lcs_exp == NULL)
1670                 RETURN(-ENOTCONN);
1671
1672         RETURN(seq_client_alloc_fid(env, seq, fid));
1673 }
1674
1675 /* context key constructor/destructor: mdt_key_init, mdt_key_fini */
1676 LU_KEY_INIT_FINI(osp, struct osp_thread_info);
1677 static void osp_key_exit(const struct lu_context *ctx,
1678                          struct lu_context_key *key, void *data)
1679 {
1680         struct osp_thread_info *info = data;
1681
1682         info->osi_attr.la_valid = 0;
1683 }
1684
1685 struct lu_context_key osp_thread_key = {
1686         .lct_tags = LCT_MD_THREAD,
1687         .lct_init = osp_key_init,
1688         .lct_fini = osp_key_fini,
1689         .lct_exit = osp_key_exit
1690 };
1691
1692 /* context key constructor/destructor: mdt_txn_key_init, mdt_txn_key_fini */
1693 LU_KEY_INIT_FINI(osp_txn, struct osp_txn_info);
1694
1695 struct lu_context_key osp_txn_key = {
1696         .lct_tags = LCT_OSP_THREAD | LCT_TX_HANDLE,
1697         .lct_init = osp_txn_key_init,
1698         .lct_fini = osp_txn_key_fini
1699 };
1700 LU_TYPE_INIT_FINI(osp, &osp_thread_key, &osp_txn_key);
1701
1702 static struct lu_device_type_operations osp_device_type_ops = {
1703         .ldto_init           = osp_type_init,
1704         .ldto_fini           = osp_type_fini,
1705
1706         .ldto_start          = osp_type_start,
1707         .ldto_stop           = osp_type_stop,
1708
1709         .ldto_device_alloc   = osp_device_alloc,
1710         .ldto_device_free    = osp_device_free,
1711
1712         .ldto_device_fini    = osp_device_fini
1713 };
1714
1715 static struct lu_device_type osp_device_type = {
1716         .ldt_tags     = LU_DEVICE_DT,
1717         .ldt_name     = LUSTRE_OSP_NAME,
1718         .ldt_ops      = &osp_device_type_ops,
1719         .ldt_ctx_tags = LCT_MD_THREAD | LCT_DT_THREAD,
1720 };
1721
1722 static struct obd_ops osp_obd_device_ops = {
1723         .o_owner        = THIS_MODULE,
1724         .o_add_conn     = client_import_add_conn,
1725         .o_del_conn     = client_import_del_conn,
1726         .o_reconnect    = osp_reconnect,
1727         .o_connect      = osp_obd_connect,
1728         .o_disconnect   = osp_obd_disconnect,
1729         .o_get_info     = osp_obd_get_info,
1730         .o_import_event = osp_import_event,
1731         .o_iocontrol    = osp_iocontrol,
1732         .o_statfs       = osp_obd_statfs,
1733         .o_fid_init     = client_fid_init,
1734         .o_fid_fini     = client_fid_fini,
1735         .o_fid_alloc    = osp_fid_alloc,
1736 };
1737
1738 struct llog_operations osp_mds_ost_orig_logops;
1739
1740 /**
1741  * Initialize OSP module.
1742  *
1743  * Register device types OSP and Light Weight Proxy (LWP) (\see lwp_dev.c)
1744  * in obd_types (\see class_obd.c).  Initialize procfs for the
1745  * the OSP device.  Note: OSP was called OSC before Lustre 2.4,
1746  * so for compatibility it still uses the name "osc" in procfs.
1747  * This is called at module load time.
1748  *
1749  * \retval 0            0 if initialization succeeds.
1750  * \retval negative     negative errno if initialization failed.
1751  */
1752 static int __init osp_mod_init(void)
1753 {
1754         struct obd_type *type;
1755         int rc;
1756
1757         rc = lu_kmem_init(osp_caches);
1758         if (rc)
1759                 return rc;
1760
1761
1762         rc = class_register_type(&osp_obd_device_ops, NULL, true, NULL,
1763                                  LUSTRE_OSP_NAME, &osp_device_type);
1764         if (rc != 0) {
1765                 lu_kmem_fini(osp_caches);
1766                 return rc;
1767         }
1768
1769         rc = class_register_type(&lwp_obd_device_ops, NULL, true, NULL,
1770                                  LUSTRE_LWP_NAME, &lwp_device_type);
1771         if (rc != 0) {
1772                 class_unregister_type(LUSTRE_OSP_NAME);
1773                 lu_kmem_fini(osp_caches);
1774                 return rc;
1775         }
1776
1777         /* Note: add_rec/delcare_add_rec will be only used by catalogs */
1778         osp_mds_ost_orig_logops = llog_osd_ops;
1779         osp_mds_ost_orig_logops.lop_add = llog_cat_add_rec;
1780         osp_mds_ost_orig_logops.lop_declare_add = llog_cat_declare_add_rec;
1781
1782         /* create "osc" entry in procfs for compatibility purposes */
1783         type = class_search_type(LUSTRE_OSC_NAME);
1784         if (type != NULL && type->typ_procroot != NULL)
1785                 return rc;
1786
1787         type = class_search_type(LUSTRE_OSP_NAME);
1788         type->typ_procsym = lprocfs_register("osc", proc_lustre_root,
1789                                              NULL, NULL);
1790         if (IS_ERR(type->typ_procsym)) {
1791                 CERROR("osp: can't create compat entry \"osc\": %d\n",
1792                        (int) PTR_ERR(type->typ_procsym));
1793                 type->typ_procsym = NULL;
1794         }
1795         return rc;
1796 }
1797
1798 /**
1799  * Finalize OSP module.
1800  *
1801  * This callback is called when kernel unloads OSP module from memory, and
1802  * it will deregister OSP and LWP device type from obd_types (\see class_obd.c).
1803  */
1804 static void __exit osp_mod_exit(void)
1805 {
1806         class_unregister_type(LUSTRE_LWP_NAME);
1807         class_unregister_type(LUSTRE_OSP_NAME);
1808         lu_kmem_fini(osp_caches);
1809 }
1810
1811 MODULE_AUTHOR("Intel, Inc. <http://www.intel.com/>");
1812 MODULE_DESCRIPTION("Lustre OST Proxy Device ("LUSTRE_OSP_NAME")");
1813 MODULE_LICENSE("GPL");
1814
1815 cfs_module(osp, LUSTRE_VERSION_STRING, osp_mod_init, osp_mod_exit);