Whamcloud - gitweb
LU-8569 linkea: linkEA size limitation
[fs/lustre-release.git] / lustre / osp / osp_object.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.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2012, 2016, Intel Corporation.
27  */
28 /*
29  * lustre/osp/osp_object.c
30  *
31  * Lustre OST Proxy Device (OSP) is the agent on the local MDT for the OST
32  * or remote MDT.
33  *
34  * OSP object attributes cache
35  * ---------------------------
36  * OSP object is the stub of the remote OST-object or MDT-object. Both the
37  * attribute and the extended attributes are stored on the peer side remotely.
38  * It is inefficient to send RPC to peer to fetch those attributes when every
39  * get_attr()/get_xattr() called. For a large system, the LFSCK synchronous
40  * mode scanning is prohibitively inefficient.
41  *
42  * So the OSP maintains the OSP object attributes cache to cache some
43  * attributes on the local MDT. The cache is organized against the OSP
44  * object as follows:
45  *
46  * struct osp_xattr_entry {
47  *      struct list_head         oxe_list;
48  *      atomic_t                 oxe_ref;
49  *      void                    *oxe_value;
50  *      int                      oxe_buflen;
51  *      int                      oxe_namelen;
52  *      int                      oxe_vallen;
53  *      unsigned int             oxe_exist:1,
54  *                               oxe_ready:1;
55  *      char                     oxe_buf[0];
56  * };
57  *
58  * struct osp_object {
59  *      ...
60  *      struct lu_attr          opo_attr;
61  *      struct list_head        opo_xattr_list;
62  *      spinlock_t              opo_lock;
63  *      ...
64  * };
65  *
66  * The basic attributes, such as owner/mode/flags, are stored in the
67  * osp_object::opo_attr. The extended attributes will be stored
68  * as osp_xattr_entry. Every extended attribute has an independent
69  * osp_xattr_entry, and all the osp_xattr_entry are linked into the
70  * osp_object::opo_xattr_list. The OSP object attributes cache
71  * is protected by the osp_object::opo_lock.
72  *
73  * Not all OSP objects have an attributes cache because maintaining
74  * the cache requires some resources. Currently, the OSP object
75  * attributes cache will be initialized when the attributes or the
76  * extended attributes are pre-fetched via osp_declare_attr_get()
77  * or osp_declare_xattr_get(). That is usually for LFSCK purpose,
78  * but it also can be shared by others.
79  *
80  *
81  * XXX: NOT prepare out RPC for remote transaction. ((please refer to the
82  *      comment of osp_trans_create() for remote transaction)
83  *
84  * According to our current transaction/dt_object_lock framework (to make
85  * the cross-MDTs modification for DNE1 to be workable), the transaction
86  * sponsor will start the transaction firstly, then try to acquire related
87  * dt_object_lock if needed. Under such rules, if we want to prepare the
88  * OUT RPC in the transaction declare phase, then related attr/xattr
89  * should be known without dt_object_lock. But such condition maybe not
90  * true for some remote transaction case. For example:
91  *
92  * For linkEA repairing (by LFSCK) case, before the LFSCK thread obtained
93  * the dt_object_lock on the target MDT-object, it cannot know whether
94  * the MDT-object has linkEA or not, neither invalid or not.
95  *
96  * Since the LFSCK thread cannot hold dt_object_lock before the remote
97  * transaction start (otherwise there will be some potential deadlock),
98  * it cannot prepare related OUT RPC for repairing during the declare
99  * phase as other normal transactions do.
100  *
101  * To resolve the trouble, we will make OSP to prepare related OUT RPC
102  * after remote transaction started, and trigger the remote updating
103  * (send RPC) when trans_stop. Then the up layer users, such as LFSCK,
104  * can follow the general rule to handle trans_start/dt_object_lock
105  * for repairing linkEA inconsistency without distinguishing remote
106  * MDT-object.
107  *
108  * In fact, above solution for remote transaction should be the normal
109  * model without considering DNE1. The trouble brought by DNE1 will be
110  * resolved in DNE2. At that time, this patch can be removed.
111  *
112  *
113  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
114  * Author: Mikhail Pershin <mike.tappro@intel.com>
115  */
116
117 #define DEBUG_SUBSYSTEM S_MDS
118
119 #include <lustre_obdo.h>
120 #include <lustre_swab.h>
121
122 #include "osp_internal.h"
123
124 static inline __u32 osp_dev2node(struct osp_device *osp)
125 {
126         return osp->opd_storage->dd_lu_dev.ld_site->ld_seq_site->ss_node_id;
127 }
128
129 static inline bool is_ost_obj(struct lu_object *lo)
130 {
131         return !lu2osp_dev(lo->lo_dev)->opd_connect_mdt;
132 }
133
134 /**
135  * Assign FID to the OST object.
136  *
137  * This function will assign the FID to the OST object of a striped file.
138  *
139  * \param[in] env       pointer to the thread context
140  * \param[in] d         pointer to the OSP device
141  * \param[in] o         pointer to the OSP object that the FID will be
142  *                      assigned to
143  */
144 static void osp_object_assign_fid(const struct lu_env *env,
145                                   struct osp_device *d, struct osp_object *o)
146 {
147         struct osp_thread_info *osi = osp_env_info(env);
148
149         LASSERT(fid_is_zero(lu_object_fid(&o->opo_obj.do_lu)));
150         LASSERT(o->opo_reserved);
151         o->opo_reserved = 0;
152
153         osp_precreate_get_fid(env, d, &osi->osi_fid);
154
155         lu_object_assign_fid(env, &o->opo_obj.do_lu, &osi->osi_fid);
156 }
157
158 /**
159  * Find the named extended attribute in the OSP object attributes cache.
160  *
161  * The caller should take the osp_object::opo_lock before calling
162  * this function.
163  *
164  * \param[in] obj       pointer to the OSP object
165  * \param[in] name      the name of the extended attribute
166  * \param[in] namelen   the name length of the extended attribute
167  *
168  * \retval              pointer to the found extended attribute entry
169  * \retval              NULL if the specified extended attribute is not
170  *                      in the cache
171  */
172 static struct osp_xattr_entry *
173 osp_oac_xattr_find_locked(struct osp_object *obj, const char *name,
174                           size_t namelen)
175 {
176         struct osp_xattr_entry *oxe;
177
178         list_for_each_entry(oxe, &obj->opo_xattr_list, oxe_list) {
179                 if (namelen == oxe->oxe_namelen &&
180                     strncmp(name, oxe->oxe_buf, namelen) == 0)
181                         return oxe;
182         }
183
184         return NULL;
185 }
186
187 /**
188  * Find the named extended attribute in the OSP object attributes cache.
189  *
190  * Call osp_oac_xattr_find_locked() with the osp_object::opo_lock held.
191  *
192  * \param[in] obj       pointer to the OSP object
193  * \param[in] name      the name of the extended attribute
194  * \param[in] unlink    true if the extended attribute entry is to be removed
195  *                      from the cache
196  *
197  * \retval              pointer to the found extended attribute entry
198  * \retval              NULL if the specified extended attribute is not
199  *                      in the cache
200  */
201 static struct osp_xattr_entry *osp_oac_xattr_find(struct osp_object *obj,
202                                                   const char *name, bool unlink)
203 {
204         struct osp_xattr_entry *oxe = NULL;
205
206         spin_lock(&obj->opo_lock);
207         oxe = osp_oac_xattr_find_locked(obj, name, strlen(name));
208         if (oxe != NULL) {
209                 if (unlink)
210                         list_del_init(&oxe->oxe_list);
211                 else
212                         atomic_inc(&oxe->oxe_ref);
213         }
214         spin_unlock(&obj->opo_lock);
215
216         return oxe;
217 }
218
219 /**
220  * Find the named extended attribute in the OSP object attributes cache.
221  *
222  * If it is not in the cache, then add an empty entry (that will be
223  * filled later) to cache with the given name.
224  *
225  * \param[in] obj       pointer to the OSP object
226  * \param[in] name      the name of the extended attribute
227  * \param[in] len       the length of the extended attribute value
228  *
229  * \retval              pointer to the found or new-created extended
230  *                      attribute entry
231  * \retval              NULL if the specified extended attribute is not in the
232  *                      cache or fail to add new empty entry to the cache.
233  */
234 static struct osp_xattr_entry *
235 osp_oac_xattr_find_or_add(struct osp_object *obj, const char *name, size_t len)
236 {
237         struct osp_xattr_entry *oxe;
238         struct osp_xattr_entry *tmp     = NULL;
239         size_t                  namelen = strlen(name);
240         size_t                  size    = sizeof(*oxe) + namelen + 1 + len;
241
242         oxe = osp_oac_xattr_find(obj, name, false);
243         if (oxe != NULL)
244                 return oxe;
245
246         OBD_ALLOC(oxe, size);
247         if (unlikely(oxe == NULL))
248                 return NULL;
249
250         INIT_LIST_HEAD(&oxe->oxe_list);
251         oxe->oxe_buflen = size;
252         oxe->oxe_namelen = namelen;
253         memcpy(oxe->oxe_buf, name, namelen);
254         oxe->oxe_value = oxe->oxe_buf + namelen + 1;
255         /* One ref is for the caller, the other is for the entry on the list. */
256         atomic_set(&oxe->oxe_ref, 2);
257
258         spin_lock(&obj->opo_lock);
259         tmp = osp_oac_xattr_find_locked(obj, name, namelen);
260         if (tmp == NULL)
261                 list_add_tail(&oxe->oxe_list, &obj->opo_xattr_list);
262         else
263                 atomic_inc(&tmp->oxe_ref);
264         spin_unlock(&obj->opo_lock);
265
266         if (tmp != NULL) {
267                 OBD_FREE(oxe, size);
268                 oxe = tmp;
269         }
270
271         return oxe;
272 }
273
274 /**
275  * Add the given extended attribute to the OSP object attributes cache.
276  *
277  * If there is an old extended attributed entry with the same name,
278  * remove it from the cache and return it via the parameter \a poxe.
279  *
280  * \param[in] obj       pointer to the OSP object
281  * \param[in,out] poxe  double pointer to the OSP object extended attribute
282  *                      entry: the new extended attribute entry is transferred
283  *                      via such pointer target, and if old the extended
284  *                      attribute entry exists, then it will be returned back
285  *                      via such pointer target.
286  * \param[in] len       the length of the (new) extended attribute value
287  *
288  * \retval              pointer to the new extended attribute entry
289  * \retval              NULL for failure cases.
290  */
291 static struct osp_xattr_entry *
292 osp_oac_xattr_replace(struct osp_object *obj,
293                       struct osp_xattr_entry **poxe, size_t len)
294 {
295         struct osp_xattr_entry *oxe;
296         size_t                  namelen = (*poxe)->oxe_namelen;
297         size_t                  size    = sizeof(*oxe) + namelen + 1 + len;
298
299         OBD_ALLOC(oxe, size);
300         if (unlikely(oxe == NULL))
301                 return NULL;
302
303         INIT_LIST_HEAD(&oxe->oxe_list);
304         oxe->oxe_buflen = size;
305         oxe->oxe_namelen = namelen;
306         memcpy(oxe->oxe_buf, (*poxe)->oxe_buf, namelen);
307         oxe->oxe_value = oxe->oxe_buf + namelen + 1;
308         /* One ref is for the caller, the other is for the entry on the list. */
309         atomic_set(&oxe->oxe_ref, 2);
310
311         spin_lock(&obj->opo_lock);
312         *poxe = osp_oac_xattr_find_locked(obj, oxe->oxe_buf, namelen);
313         LASSERT(*poxe != NULL);
314
315         list_del_init(&(*poxe)->oxe_list);
316         list_add_tail(&oxe->oxe_list, &obj->opo_xattr_list);
317         spin_unlock(&obj->opo_lock);
318
319         return oxe;
320 }
321
322 /**
323  * Release reference from the OSP object extended attribute entry.
324  *
325  * If it is the last reference, then free the entry.
326  *
327  * \param[in] oxe       pointer to the OSP object extended attribute entry.
328  */
329 static inline void osp_oac_xattr_put(struct osp_xattr_entry *oxe)
330 {
331         if (atomic_dec_and_test(&oxe->oxe_ref)) {
332                 LASSERT(list_empty(&oxe->oxe_list));
333
334                 OBD_FREE(oxe, oxe->oxe_buflen);
335         }
336 }
337
338 /**
339  * Parse the OSP object attribute from the RPC reply.
340  *
341  * If the attribute is valid, then it will be added to the OSP object
342  * attributes cache.
343  *
344  * \param[in] env       pointer to the thread context
345  * \param[in] reply     pointer to the RPC reply
346  * \param[in] req       pointer to the RPC request
347  * \param[out] attr     pointer to buffer to hold the output attribute
348  * \param[in] obj       pointer to the OSP object
349  * \param[in] index     the index of the attribute buffer in the reply
350  *
351  * \retval              0 for success
352  * \retval              negative error number on failure
353  */
354 static int osp_get_attr_from_reply(const struct lu_env *env,
355                                    struct object_update_reply *reply,
356                                    struct ptlrpc_request *req,
357                                    struct lu_attr *attr,
358                                    struct osp_object *obj, int index)
359 {
360         struct osp_thread_info  *osi    = osp_env_info(env);
361         struct lu_buf           *rbuf   = &osi->osi_lb2;
362         struct obdo             *lobdo  = &osi->osi_obdo;
363         struct obdo             *wobdo;
364         int                     rc;
365
366         rc = object_update_result_data_get(reply, rbuf, index);
367         if (rc < 0)
368                 return rc;
369
370         wobdo = rbuf->lb_buf;
371         if (rbuf->lb_len != sizeof(*wobdo))
372                 return -EPROTO;
373
374         LASSERT(req != NULL);
375         if (ptlrpc_req_need_swab(req))
376                 lustre_swab_obdo(wobdo);
377
378         lustre_get_wire_obdo(NULL, lobdo, wobdo);
379         spin_lock(&obj->opo_lock);
380         la_from_obdo(&obj->opo_attr, lobdo, lobdo->o_valid);
381         if (attr != NULL)
382                 *attr = obj->opo_attr;
383         spin_unlock(&obj->opo_lock);
384
385         return 0;
386 }
387
388 /**
389  * Interpreter function for getting OSP object attribute asynchronously.
390  *
391  * Called to interpret the result of an async mode RPC for getting the
392  * OSP object attribute.
393  *
394  * \param[in] env       pointer to the thread context
395  * \param[in] reply     pointer to the RPC reply
396  * \param[in] req       pointer to the RPC request
397  * \param[in] obj       pointer to the OSP object
398  * \param[out] data     pointer to buffer to hold the output attribute
399  * \param[in] index     the index of the attribute buffer in the reply
400  * \param[in] rc        the result for handling the RPC
401  *
402  * \retval              0 for success
403  * \retval              negative error number on failure
404  */
405 static int osp_attr_get_interpterer(const struct lu_env *env,
406                                     struct object_update_reply *reply,
407                                     struct ptlrpc_request *req,
408                                     struct osp_object *obj,
409                                     void *data, int index, int rc)
410 {
411         struct lu_attr *attr = data;
412
413         if (rc == 0) {
414                 osp2lu_obj(obj)->lo_header->loh_attr |= LOHA_EXISTS;
415                 obj->opo_non_exist = 0;
416
417                 return osp_get_attr_from_reply(env, reply, req, NULL, obj,
418                                                index);
419         } else {
420                 if (rc == -ENOENT) {
421                         osp2lu_obj(obj)->lo_header->loh_attr &= ~LOHA_EXISTS;
422                         obj->opo_non_exist = 1;
423                 }
424
425                 spin_lock(&obj->opo_lock);
426                 attr->la_valid = 0;
427                 spin_unlock(&obj->opo_lock);
428         }
429
430         return 0;
431 }
432
433 /**
434  * Implement OSP layer dt_object_operations::do_declare_attr_get() interface.
435  *
436  * Declare that the caller will get attribute from the specified OST object.
437  *
438  * This function adds an Object Unified Target (OUT) sub-request to the per-OSP
439  * based shared asynchronous request queue. The osp_attr_get_interpterer()
440  * is registered as the interpreter function to handle the result of this
441  * sub-request.
442  *
443  * \param[in] env       pointer to the thread context
444  * \param[in] dt        pointer to the OSP layer dt_object
445  *
446  * \retval              0 for success
447  * \retval              negative error number on failure
448  */
449 static int osp_declare_attr_get(const struct lu_env *env, struct dt_object *dt)
450 {
451         struct osp_object       *obj    = dt2osp_obj(dt);
452         struct osp_device       *osp    = lu2osp_dev(dt->do_lu.lo_dev);
453         int                      rc     = 0;
454
455         mutex_lock(&osp->opd_async_requests_mutex);
456         rc = osp_insert_async_request(env, OUT_ATTR_GET, obj, 0, NULL, NULL,
457                                       &obj->opo_attr, sizeof(struct obdo),
458                                       osp_attr_get_interpterer);
459         mutex_unlock(&osp->opd_async_requests_mutex);
460
461         return rc;
462 }
463
464 /**
465  * Implement OSP layer dt_object_operations::do_attr_get() interface.
466  *
467  * Get attribute from the specified MDT/OST object.
468  *
469  * If the attribute is in the OSP object attributes cache, then return
470  * the cached attribute directly. Otherwise it will trigger an OUT RPC
471  * to the peer to get the attribute synchronously, if successful, add it
472  * to the OSP attributes cache. (\see lustre/osp/osp_trans.c for OUT RPC.)
473  *
474  * \param[in] env       pointer to the thread context
475  * \param[in] dt        pointer to the OSP layer dt_object
476  * \param[out] attr     pointer to the buffer to hold the output attribute
477  *
478  * \retval              0 for success
479  * \retval              negative error number on failure
480  */
481 int osp_attr_get(const struct lu_env *env, struct dt_object *dt,
482                  struct lu_attr *attr)
483 {
484         struct osp_device               *osp = lu2osp_dev(dt->do_lu.lo_dev);
485         struct osp_object               *obj = dt2osp_obj(dt);
486         struct dt_device                *dev = &osp->opd_dt_dev;
487         struct osp_update_request       *update;
488         struct object_update_reply      *reply;
489         struct ptlrpc_request           *req = NULL;
490         int                             rc = 0;
491         ENTRY;
492
493         if (is_ost_obj(&dt->do_lu) && obj->opo_non_exist)
494                 RETURN(-ENOENT);
495
496         spin_lock(&obj->opo_lock);
497         if (obj->opo_attr.la_valid != 0 && !obj->opo_stale) {
498                 *attr = obj->opo_attr;
499                 spin_unlock(&obj->opo_lock);
500
501                 RETURN(0);
502         }
503         spin_unlock(&obj->opo_lock);
504
505         update = osp_update_request_create(dev);
506         if (IS_ERR(update))
507                 RETURN(PTR_ERR(update));
508
509         rc = osp_update_rpc_pack(env, attr_get, update, OUT_ATTR_GET,
510                                  lu_object_fid(&dt->do_lu));
511         if (rc != 0) {
512                 CERROR("%s: Insert update error "DFID": rc = %d\n",
513                        dev->dd_lu_dev.ld_obd->obd_name,
514                        PFID(lu_object_fid(&dt->do_lu)), rc);
515
516                 GOTO(out, rc);
517         }
518
519         rc = osp_remote_sync(env, osp, update, &req);
520         if (rc != 0) {
521                 if (rc == -ENOENT) {
522                         osp2lu_obj(obj)->lo_header->loh_attr &= ~LOHA_EXISTS;
523                         obj->opo_non_exist = 1;
524                 } else {
525                         CERROR("%s:osp_attr_get update error "DFID": rc = %d\n",
526                                dev->dd_lu_dev.ld_obd->obd_name,
527                                PFID(lu_object_fid(&dt->do_lu)), rc);
528                 }
529
530                 GOTO(out, rc);
531         }
532
533         osp2lu_obj(obj)->lo_header->loh_attr |= LOHA_EXISTS;
534         obj->opo_non_exist = 0;
535         reply = req_capsule_server_sized_get(&req->rq_pill,
536                                              &RMF_OUT_UPDATE_REPLY,
537                                              OUT_UPDATE_REPLY_SIZE);
538         if (reply == NULL || reply->ourp_magic != UPDATE_REPLY_MAGIC)
539                 GOTO(out, rc = -EPROTO);
540
541         rc = osp_get_attr_from_reply(env, reply, req, attr, obj, 0);
542         if (rc != 0)
543                 GOTO(out, rc);
544
545         spin_lock(&obj->opo_lock);
546         obj->opo_stale = 0;
547         spin_unlock(&obj->opo_lock);
548
549         GOTO(out, rc);
550
551 out:
552         if (req != NULL)
553                 ptlrpc_req_finished(req);
554
555         osp_update_request_destroy(env, update);
556
557         return rc;
558 }
559
560 /**
561  * Implement OSP layer dt_object_operations::do_declare_attr_set() interface.
562  *
563  * If the transaction is not remote one, then declare the credits that will
564  * be used for the subsequent llog record for the object's attributes.
565  *
566  * \param[in] env       pointer to the thread context
567  * \param[in] dt        pointer to the OSP layer dt_object
568  * \param[in] attr      pointer to the attribute to be set
569  * \param[in] th        pointer to the transaction handler
570  *
571  * \retval              0 for success
572  * \retval              negative error number on failure
573  */
574 static int osp_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
575                                 const struct lu_attr *attr, struct thandle *th)
576 {
577         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
578         struct osp_object       *o = dt2osp_obj(dt);
579         int                     rc;
580
581         if (is_only_remote_trans(th))
582                 return osp_md_declare_attr_set(env, dt, attr, th);
583         /*
584          * Usually we don't allow server stack to manipulate size
585          * but there is a special case when striping is created
586          * late, after stripeless file got truncated to non-zero.
587          *
588          * In this case we do the following:
589          *
590          * 1) grab id in declare - this can lead to leaked OST objects
591          *    but we don't currently have proper mechanism and the only
592          *    options we have are to do truncate RPC holding transaction
593          *    open (very bad) or to grab id in declare at cost of leaked
594          *    OST object in same very rare unfortunate case (just bad)
595          *    notice 1.6-2.0 do assignment outside of running transaction
596          *    all the time, meaning many more chances for leaked objects.
597          *
598          * 2) send synchronous truncate RPC with just assigned id
599          */
600
601         /* there are few places in MDD code still passing NULL
602          * XXX: to be fixed soon */
603         if (attr == NULL)
604                 RETURN(0);
605
606         if (attr->la_valid & LA_SIZE && attr->la_size > 0 &&
607             fid_is_zero(lu_object_fid(&o->opo_obj.do_lu))) {
608                 LASSERT(!dt_object_exists(dt));
609                 osp_object_assign_fid(env, d, o);
610                 rc = osp_object_truncate(env, dt, attr->la_size);
611                 if (rc != 0)
612                         RETURN(rc);
613         }
614
615         if (!(attr->la_valid & (LA_UID | LA_GID)))
616                 RETURN(0);
617
618         /* track all UID/GID changes via llog */
619         rc = osp_sync_declare_add(env, o, MDS_SETATTR64_REC, th);
620
621         return 0;
622 }
623
624 /**
625  * Implement OSP layer dt_object_operations::do_attr_set() interface.
626  *
627  * Set attribute to the specified OST object.
628  *
629  * If the transaction is a remote one, then add OUT_ATTR_SET sub-request
630  * in the OUT RPC that will be flushed when the remote transaction stop.
631  * Otherwise, it will generate a MDS_SETATTR64_REC record in the llog that
632  * will be handled by a dedicated thread asynchronously.
633  *
634  * If the attribute entry exists in the OSP object attributes cache,
635  * then update the cached attribute according to given attribute.
636  *
637  * \param[in] env       pointer to the thread context
638  * \param[in] dt        pointer to the OSP layer dt_object
639  * \param[in] attr      pointer to the attribute to be set
640  * \param[in] th        pointer to the transaction handler
641  *
642  * \retval              0 for success
643  * \retval              negative error number on failure
644  */
645 static int osp_attr_set(const struct lu_env *env, struct dt_object *dt,
646                         const struct lu_attr *attr, struct thandle *th)
647 {
648         struct osp_object       *o = dt2osp_obj(dt);
649         int                      rc = 0;
650         ENTRY;
651
652         /* we're interested in uid/gid changes only */
653         if (!(attr->la_valid & (LA_UID | LA_GID)))
654                 RETURN(0);
655
656         if (!is_only_remote_trans(th)) {
657                 rc = osp_sync_add(env, o, MDS_SETATTR64_REC, th, attr);
658                 /* XXX: send new uid/gid to OST ASAP? */
659         } else {
660                 struct lu_attr  *la;
661
662                 /* It is for OST-object attr_set directly without updating
663                  * local MDT-object attribute. It is usually used by LFSCK. */
664                 rc = osp_md_attr_set(env, dt, attr, th);
665                 CDEBUG(D_INFO, "(1) set attr "DFID": rc = %d\n",
666                        PFID(&dt->do_lu.lo_header->loh_fid), rc);
667
668                 if (rc != 0)
669                         RETURN(rc);
670
671                 /* Update the OSP object attributes cache. */
672                 la = &o->opo_attr;
673                 spin_lock(&o->opo_lock);
674                 if (attr->la_valid & LA_UID) {
675                         la->la_uid = attr->la_uid;
676                         la->la_valid |= LA_UID;
677                 }
678
679                 if (attr->la_valid & LA_GID) {
680                         la->la_gid = attr->la_gid;
681                         la->la_valid |= LA_GID;
682                 }
683                 spin_unlock(&o->opo_lock);
684         }
685
686         RETURN(rc);
687 }
688
689 /**
690  * Interpreter function for getting OSP object extended attribute asynchronously
691  *
692  * Called to interpret the result of an async mode RPC for getting the
693  * OSP object extended attribute.
694  *
695  * \param[in] env       pointer to the thread context
696  * \param[in] reply     pointer to the RPC reply
697  * \param[in] req       pointer to the RPC request
698  * \param[in] obj       pointer to the OSP object
699  * \param[out] data     pointer to OSP object attributes cache
700  * \param[in] index     the index of the attribute buffer in the reply
701  * \param[in] rc        the result for handling the RPC
702  *
703  * \retval              0 for success
704  * \retval              negative error number on failure
705  */
706 static int osp_xattr_get_interpterer(const struct lu_env *env,
707                                      struct object_update_reply *reply,
708                                      struct ptlrpc_request *req,
709                                      struct osp_object *obj,
710                                      void *data, int index, int rc)
711 {
712         struct osp_xattr_entry  *oxe  = data;
713         struct lu_buf           *rbuf = &osp_env_info(env)->osi_lb2;
714
715         if (rc == 0) {
716                 size_t len = sizeof(*oxe) + oxe->oxe_namelen + 1;
717
718                 rc = object_update_result_data_get(reply, rbuf, index);
719                 if (rc < 0 || rbuf->lb_len > (oxe->oxe_buflen - len)) {
720                         spin_lock(&obj->opo_lock);
721                         oxe->oxe_ready = 0;
722                         spin_unlock(&obj->opo_lock);
723                         osp_oac_xattr_put(oxe);
724
725                         return rc < 0 ? rc : -ERANGE;
726                 }
727
728                 spin_lock(&obj->opo_lock);
729                 oxe->oxe_vallen = rbuf->lb_len;
730                 memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
731                 oxe->oxe_exist = 1;
732                 oxe->oxe_ready = 1;
733                 spin_unlock(&obj->opo_lock);
734         } else if (rc == -ENOENT || rc == -ENODATA) {
735                 spin_lock(&obj->opo_lock);
736                 oxe->oxe_exist = 0;
737                 oxe->oxe_ready = 1;
738                 spin_unlock(&obj->opo_lock);
739         } else {
740                 spin_lock(&obj->opo_lock);
741                 oxe->oxe_ready = 0;
742                 spin_unlock(&obj->opo_lock);
743         }
744
745         osp_oac_xattr_put(oxe);
746
747         return 0;
748 }
749
750 /**
751  * Implement OSP dt_object_operations::do_declare_xattr_get() interface.
752  *
753  * Declare that the caller will get extended attribute from the specified
754  * OST object.
755  *
756  * This function will add an OUT_XATTR_GET sub-request to the per OSP
757  * based shared asynchronous request queue with the interpreter function:
758  * osp_xattr_get_interpterer().
759  *
760  * \param[in] env       pointer to the thread context
761  * \param[in] dt        pointer to the OSP layer dt_object
762  * \param[out] buf      pointer to the lu_buf to hold the extended attribute
763  * \param[in] name      the name for the expected extended attribute
764  *
765  * \retval              0 for success
766  * \retval              negative error number on failure
767  */
768 static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt,
769                                  struct lu_buf *buf, const char *name)
770 {
771         struct osp_object       *obj     = dt2osp_obj(dt);
772         struct osp_device       *osp     = lu2osp_dev(dt->do_lu.lo_dev);
773         struct osp_xattr_entry  *oxe;
774         __u16 namelen;
775         int                      rc      = 0;
776
777         LASSERT(buf != NULL);
778         LASSERT(name != NULL);
779
780         namelen = strlen(name);
781
782         /* If only for xattr size, return directly. */
783         if (unlikely(buf->lb_len == 0))
784                 return 0;
785
786         oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
787         if (oxe == NULL)
788                 return -ENOMEM;
789
790         mutex_lock(&osp->opd_async_requests_mutex);
791         rc = osp_insert_async_request(env, OUT_XATTR_GET, obj, 1,
792                                       &namelen, (const void **)&name,
793                                       oxe, buf->lb_len,
794                                       osp_xattr_get_interpterer);
795         if (rc != 0) {
796                 mutex_unlock(&osp->opd_async_requests_mutex);
797                 osp_oac_xattr_put(oxe);
798         } else {
799                 struct osp_update_request *our;
800                 struct osp_update_request_sub *ours;
801
802                 /* XXX: Currently, we trigger the batched async OUT
803                  *      RPC via dt_declare_xattr_get(). It is not
804                  *      perfect solution, but works well now.
805                  *
806                  *      We will improve it in the future. */
807                 our = osp->opd_async_requests;
808                 ours = osp_current_object_update_request(our);
809                 if (ours != NULL && ours->ours_req != NULL &&
810                     ours->ours_req->ourq_count > 0) {
811                         osp->opd_async_requests = NULL;
812                         mutex_unlock(&osp->opd_async_requests_mutex);
813                         rc = osp_unplug_async_request(env, osp, our);
814                 } else {
815                         mutex_unlock(&osp->opd_async_requests_mutex);
816                 }
817         }
818
819         return rc;
820 }
821
822 /**
823  * Implement OSP layer dt_object_operations::do_xattr_get() interface.
824  *
825  * Get extended attribute from the specified MDT/OST object.
826  *
827  * If the extended attribute is in the OSP object attributes cache, then
828  * return the cached extended attribute directly. Otherwise it will get
829  * the extended attribute synchronously, if successful, add it to the OSP
830  * attributes cache. (\see lustre/osp/osp_trans.c for OUT RPC.)
831  *
832  * There is a race condition: some other thread has added the named extended
833  * attributed entry to the OSP object attributes cache during the current
834  * OUT_XATTR_GET handling. If such case happens, the OSP will replace the
835  * (just) existing extended attribute entry with the new replied one.
836  *
837  * \param[in] env       pointer to the thread context
838  * \param[in] dt        pointer to the OSP layer dt_object
839  * \param[out] buf      pointer to the lu_buf to hold the extended attribute
840  * \param[in] name      the name for the expected extended attribute
841  *
842  * \retval              0 for success
843  * \retval              negative error number on failure
844  */
845 int osp_xattr_get(const struct lu_env *env, struct dt_object *dt,
846                   struct lu_buf *buf, const char *name)
847 {
848         struct osp_device       *osp    = lu2osp_dev(dt->do_lu.lo_dev);
849         struct osp_object       *obj    = dt2osp_obj(dt);
850         struct dt_device        *dev    = &osp->opd_dt_dev;
851         struct lu_buf           *rbuf   = &osp_env_info(env)->osi_lb2;
852         struct osp_update_request *update = NULL;
853         struct ptlrpc_request   *req    = NULL;
854         struct object_update_reply *reply;
855         struct osp_xattr_entry  *oxe    = NULL;
856         const char              *dname  = dt->do_lu.lo_dev->ld_obd->obd_name;
857         int                      rc     = 0;
858         ENTRY;
859
860         LASSERT(buf != NULL);
861         LASSERT(name != NULL);
862
863         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_NETWORK) &&
864             osp->opd_index == cfs_fail_val) {
865                 if (is_ost_obj(&dt->do_lu)) {
866                         if (osp_dev2node(osp) == cfs_fail_val)
867                                 RETURN(-ENOTCONN);
868                 } else {
869                         if (strcmp(name, XATTR_NAME_LINK) == 0)
870                                 RETURN(-ENOTCONN);
871                 }
872         }
873
874         if (unlikely(obj->opo_non_exist))
875                 RETURN(-ENOENT);
876
877         oxe = osp_oac_xattr_find(obj, name, false);
878         if (oxe != NULL) {
879                 spin_lock(&obj->opo_lock);
880                 if (oxe->oxe_ready) {
881                         if (!oxe->oxe_exist)
882                                 GOTO(unlock, rc = -ENODATA);
883
884                         if (buf->lb_buf == NULL)
885                                 GOTO(unlock, rc = oxe->oxe_vallen);
886
887                         if (buf->lb_len < oxe->oxe_vallen)
888                                 GOTO(unlock, rc = -ERANGE);
889
890                         memcpy(buf->lb_buf, oxe->oxe_value,
891                                oxe->oxe_vallen);
892
893                         GOTO(unlock, rc = oxe->oxe_vallen);
894
895 unlock:
896                         spin_unlock(&obj->opo_lock);
897                         osp_oac_xattr_put(oxe);
898
899                         return rc;
900                 }
901                 spin_unlock(&obj->opo_lock);
902         }
903         update = osp_update_request_create(dev);
904         if (IS_ERR(update))
905                 GOTO(out, rc = PTR_ERR(update));
906
907         rc = osp_update_rpc_pack(env, xattr_get, update, OUT_XATTR_GET,
908                                  lu_object_fid(&dt->do_lu), name, buf->lb_len);
909         if (rc != 0) {
910                 CERROR("%s: Insert update error "DFID": rc = %d\n",
911                        dname, PFID(lu_object_fid(&dt->do_lu)), rc);
912                 GOTO(out, rc);
913         }
914
915         rc = osp_remote_sync(env, osp, update, &req);
916         if (rc < 0) {
917                 if (rc == -ENOENT) {
918                         dt->do_lu.lo_header->loh_attr &= ~LOHA_EXISTS;
919                         obj->opo_non_exist = 1;
920                 }
921
922                 if (oxe == NULL)
923                         oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
924
925                 if (oxe == NULL) {
926                         CWARN("%s: Fail to add xattr (%s) to cache for "
927                               DFID" (1): rc = %d\n", dname, name,
928                               PFID(lu_object_fid(&dt->do_lu)), rc);
929
930                         GOTO(out, rc);
931                 }
932
933                 spin_lock(&obj->opo_lock);
934                 if (rc == -ENOENT || rc == -ENODATA) {
935                         oxe->oxe_exist = 0;
936                         oxe->oxe_ready = 1;
937                 } else {
938                         oxe->oxe_ready = 0;
939                 }
940                 spin_unlock(&obj->opo_lock);
941
942                 GOTO(out, rc);
943         }
944
945         reply = req_capsule_server_sized_get(&req->rq_pill,
946                                              &RMF_OUT_UPDATE_REPLY,
947                                              OUT_UPDATE_REPLY_SIZE);
948         if (reply->ourp_magic != UPDATE_REPLY_MAGIC) {
949                 CERROR("%s: Wrong version %x expected %x "DFID": rc = %d\n",
950                        dname, reply->ourp_magic, UPDATE_REPLY_MAGIC,
951                        PFID(lu_object_fid(&dt->do_lu)), -EPROTO);
952
953                 GOTO(out, rc = -EPROTO);
954         }
955
956         rc = object_update_result_data_get(reply, rbuf, 0);
957         if (rc < 0)
958                 GOTO(out, rc);
959
960         if (buf->lb_buf == NULL)
961                 GOTO(out, rc);
962
963         if (unlikely(buf->lb_len < rbuf->lb_len))
964                 GOTO(out, rc = -ERANGE);
965
966         memcpy(buf->lb_buf, rbuf->lb_buf, rbuf->lb_len);
967
968         if (oxe == NULL) {
969                 oxe = osp_oac_xattr_find_or_add(obj, name, rbuf->lb_len);
970                 if (oxe == NULL) {
971                         CWARN("%s: Fail to add xattr (%s) to "
972                               "cache for "DFID" (2): rc = %d\n",
973                               dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
974
975                         GOTO(out, rc);
976                 }
977         }
978
979         if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < rbuf->lb_len) {
980                 struct osp_xattr_entry *old = oxe;
981                 struct osp_xattr_entry *tmp;
982
983                 tmp = osp_oac_xattr_replace(obj, &old, rbuf->lb_len);
984                 osp_oac_xattr_put(oxe);
985                 oxe = tmp;
986                 if (tmp == NULL) {
987                         CWARN("%s: Fail to update xattr (%s) to "
988                               "cache for "DFID": rc = %d\n",
989                               dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
990                         spin_lock(&obj->opo_lock);
991                         old->oxe_ready = 0;
992                         spin_unlock(&obj->opo_lock);
993
994                         GOTO(out, rc);
995                 }
996
997                 /* Drop the ref for entry on list. */
998                 osp_oac_xattr_put(old);
999         }
1000
1001         spin_lock(&obj->opo_lock);
1002         oxe->oxe_vallen = rbuf->lb_len;
1003         memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len);
1004         oxe->oxe_exist = 1;
1005         oxe->oxe_ready = 1;
1006         spin_unlock(&obj->opo_lock);
1007
1008         GOTO(out, rc);
1009
1010 out:
1011         if (req != NULL)
1012                 ptlrpc_req_finished(req);
1013
1014         if (update != NULL && !IS_ERR(update))
1015                 osp_update_request_destroy(env, update);
1016
1017         if (oxe != NULL)
1018                 osp_oac_xattr_put(oxe);
1019
1020         return rc;
1021 }
1022
1023 /**
1024  * Implement OSP layer dt_object_operations::do_declare_xattr_set() interface.
1025  *
1026  * Declare that the caller will set extended attribute to the specified
1027  * MDT/OST object.
1028  *
1029  * If it is non-remote transaction, it will add an OUT_XATTR_SET sub-request
1030  * to the OUT RPC that will be flushed when the transaction start. And if the
1031  * OSP attributes cache is initialized, then check whether the name extended
1032  * attribute entry exists in the cache or not. If yes, replace it; otherwise,
1033  * add the extended attribute to the cache.
1034  *
1035  * \param[in] env       pointer to the thread context
1036  * \param[in] dt        pointer to the OSP layer dt_object
1037  * \param[in] buf       pointer to the lu_buf to hold the extended attribute
1038  * \param[in] name      the name of the extended attribute to be set
1039  * \param[in] flag      to indicate the detailed set operation: LU_XATTR_CREATE
1040  *                      or LU_XATTR_REPLACE or others
1041  * \param[in] th        pointer to the transaction handler
1042  *
1043  * \retval              0 for success
1044  * \retval              negative error number on failure
1045  */
1046 int osp_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
1047                           const struct lu_buf *buf, const char *name,
1048                           int flag, struct thandle *th)
1049 {
1050         return osp_trans_update_request_create(th);
1051 }
1052
1053 /**
1054  * Implement OSP layer dt_object_operations::do_xattr_set() interface.
1055  *
1056  * Set extended attribute to the specified MDT/OST object.
1057  *
1058  * Add an OUT_XATTR_SET sub-request into the OUT RPC that will be flushed in
1059  * the transaction stop. And if the OSP attributes cache is initialized, then
1060  * check whether the name extended attribute entry exists in the cache or not.
1061  * If yes, replace it; otherwise, add the extended attribute to the cache.
1062  *
1063  * \param[in] env       pointer to the thread context
1064  * \param[in] dt        pointer to the OSP layer dt_object
1065  * \param[in] buf       pointer to the lu_buf to hold the extended attribute
1066  * \param[in] name      the name of the extended attribute to be set
1067  * \param[in] fl        to indicate the detailed set operation: LU_XATTR_CREATE
1068  *                      or LU_XATTR_REPLACE or others
1069  * \param[in] th        pointer to the transaction handler
1070  *
1071  * \retval              0 for success
1072  * \retval              negative error number on failure
1073  */
1074 int osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
1075                   const struct lu_buf *buf, const char *name, int fl,
1076                   struct thandle *th)
1077 {
1078         struct osp_object       *o = dt2osp_obj(dt);
1079         struct osp_update_request *update;
1080         struct osp_xattr_entry  *oxe;
1081         int                     rc;
1082         ENTRY;
1083
1084         update = thandle_to_osp_update_request(th);
1085         LASSERT(update != NULL);
1086
1087         CDEBUG(D_INODE, DFID" set xattr '%s' with size %zd\n",
1088                PFID(lu_object_fid(&dt->do_lu)), name, buf->lb_len);
1089
1090         rc = osp_update_rpc_pack(env, xattr_set, update, OUT_XATTR_SET,
1091                                  lu_object_fid(&dt->do_lu), buf, name, fl);
1092         if (rc != 0)
1093                 RETURN(rc);
1094
1095         /* Do not cache linkEA that may be self-adjusted by peers
1096          * under EA overflow case. */
1097         if (strcmp(name, XATTR_NAME_LINK) == 0) {
1098                 oxe = osp_oac_xattr_find(o, name, true);
1099                 if (oxe != NULL)
1100                         osp_oac_xattr_put(oxe);
1101
1102                 RETURN(0);
1103         }
1104
1105         oxe = osp_oac_xattr_find_or_add(o, name, buf->lb_len);
1106         if (oxe == NULL) {
1107                 CWARN("%s: cannot cache xattr '%s' of "DFID"\n",
1108                       dt->do_lu.lo_dev->ld_obd->obd_name,
1109                       name, PFID(lu_object_fid(&dt->do_lu)));
1110
1111                 RETURN(0);
1112         }
1113
1114         if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < buf->lb_len) {
1115                 struct osp_xattr_entry *old = oxe;
1116                 struct osp_xattr_entry *tmp;
1117
1118                 tmp = osp_oac_xattr_replace(o, &old, buf->lb_len);
1119                 osp_oac_xattr_put(oxe);
1120                 oxe = tmp;
1121                 if (tmp == NULL) {
1122                         CWARN("%s: cannot update cached xattr '%s' of "DFID"\n",
1123                               dt->do_lu.lo_dev->ld_obd->obd_name,
1124                               name, PFID(lu_object_fid(&dt->do_lu)));
1125                         spin_lock(&o->opo_lock);
1126                         old->oxe_ready = 0;
1127                         spin_unlock(&o->opo_lock);
1128
1129                         RETURN(0);
1130                 }
1131
1132                 /* Drop the ref for entry on list. */
1133                 osp_oac_xattr_put(old);
1134         }
1135
1136         spin_lock(&o->opo_lock);
1137         oxe->oxe_vallen = buf->lb_len;
1138         memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len);
1139         oxe->oxe_exist = 1;
1140         oxe->oxe_ready = 1;
1141         spin_unlock(&o->opo_lock);
1142         osp_oac_xattr_put(oxe);
1143
1144         RETURN(0);
1145 }
1146
1147 /**
1148  * Implement OSP layer dt_object_operations::do_declare_xattr_del() interface.
1149  *
1150  * Declare that the caller will delete extended attribute on the specified
1151  * MDT/OST object.
1152  *
1153  * If it is non-remote transaction, it will add an OUT_XATTR_DEL sub-request
1154  * to the OUT RPC that will be flushed when the transaction start. And if the
1155  * name extended attribute entry exists in the OSP attributes cache, then remove
1156  * it from the cache.
1157  *
1158  * \param[in] env       pointer to the thread context
1159  * \param[in] dt        pointer to the OSP layer dt_object
1160  * \param[in] name      the name of the extended attribute to be set
1161  * \param[in] th        pointer to the transaction handler
1162  *
1163  * \retval              0 for success
1164  * \retval              negative error number on failure
1165  */
1166 int osp_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
1167                           const char *name, struct thandle *th)
1168 {
1169         return osp_trans_update_request_create(th);
1170 }
1171
1172 /**
1173  * Implement OSP layer dt_object_operations::do_xattr_del() interface.
1174  *
1175  * Delete extended attribute on the specified MDT/OST object.
1176  *
1177  * If it is remote transaction, it will add an OUT_XATTR_DEL sub-request into
1178  * the OUT RPC that will be flushed when the transaction stop. And if the name
1179  * extended attribute entry exists in the OSP attributes cache, then remove it
1180  * from the cache.
1181  *
1182  * \param[in] env       pointer to the thread context
1183  * \param[in] dt        pointer to the OSP layer dt_object
1184  * \param[in] name      the name of the extended attribute to be set
1185  * \param[in] th        pointer to the transaction handler
1186  *
1187  * \retval              0 for success
1188  * \retval              negative error number on failure
1189  */
1190 int osp_xattr_del(const struct lu_env *env, struct dt_object *dt,
1191                   const char *name, struct thandle *th)
1192 {
1193         struct osp_update_request *update;
1194         const struct lu_fid      *fid = lu_object_fid(&dt->do_lu);
1195         struct osp_object        *o     = dt2osp_obj(dt);
1196         struct osp_xattr_entry   *oxe;
1197         int                       rc;
1198
1199         update = thandle_to_osp_update_request(th);
1200         LASSERT(update != NULL);
1201
1202         rc = osp_update_rpc_pack(env, xattr_del, update, OUT_XATTR_DEL,
1203                                  fid, name);
1204         if (rc != 0)
1205                 return rc;
1206
1207         oxe = osp_oac_xattr_find(o, name, true);
1208         if (oxe != NULL)
1209                 /* Drop the ref for entry on list. */
1210                 osp_oac_xattr_put(oxe);
1211
1212         return 0;
1213 }
1214
1215 void osp_obj_invalidate_cache(struct osp_object *obj)
1216 {
1217         struct osp_xattr_entry *oxe;
1218         struct osp_xattr_entry *tmp;
1219
1220         spin_lock(&obj->opo_lock);
1221         list_for_each_entry_safe(oxe, tmp, &obj->opo_xattr_list, oxe_list) {
1222                 oxe->oxe_ready = 0;
1223                 list_del_init(&oxe->oxe_list);
1224                 osp_oac_xattr_put(oxe);
1225         }
1226         obj->opo_attr.la_valid = 0;
1227         spin_unlock(&obj->opo_lock);
1228 }
1229
1230 /**
1231  * Implement OSP layer dt_object_operations::do_invalidate() interface.
1232  *
1233  * Invalidate attributes cached on the specified MDT/OST object.
1234  *
1235  * \param[in] env       pointer to the thread context
1236  * \param[in] dt        pointer to the OSP layer dt_object
1237  *
1238  * \retval              0 for success
1239  * \retval              negative error number on failure
1240  */
1241 int osp_invalidate(const struct lu_env *env, struct dt_object *dt)
1242 {
1243         struct osp_object *obj = dt2osp_obj(dt);
1244         ENTRY;
1245
1246         osp_obj_invalidate_cache(obj);
1247
1248         spin_lock(&obj->opo_lock);
1249         obj->opo_stale = 1;
1250         spin_unlock(&obj->opo_lock);
1251
1252         RETURN(0);
1253 }
1254
1255 /**
1256  * Implement OSP layer dt_object_operations::do_declare_create() interface.
1257  *
1258  * Declare that the caller will create the OST object.
1259  *
1260  * If the transaction is a remote transaction and the FID for the OST-object
1261  * has been assigned already, then handle it as creating (remote) MDT object
1262  * via osp_md_declare_object_create(). This function is usually used for LFSCK
1263  * to re-create the lost OST object. Otherwise, if it is not replay case, the
1264  * OSP will reserve pre-created object for the subsequent create operation;
1265  * if the MDT side cached pre-created objects are less than some threshold,
1266  * then it will wakeup the pre-create thread.
1267  *
1268  * \param[in] env       pointer to the thread context
1269  * \param[in] dt        pointer to the OSP layer dt_object
1270  * \param[in] attr      the attribute for the object to be created
1271  * \param[in] hint      pointer to the hint for creating the object, such as
1272  *                      the parent object
1273  * \param[in] dof       pointer to the dt_object_format for help the creation
1274  * \param[in] th        pointer to the transaction handler
1275  *
1276  * \retval              0 for success
1277  * \retval              negative error number on failure
1278  */
1279 static int osp_declare_object_create(const struct lu_env *env,
1280                                      struct dt_object *dt,
1281                                      struct lu_attr *attr,
1282                                      struct dt_allocation_hint *hint,
1283                                      struct dt_object_format *dof,
1284                                      struct thandle *th)
1285 {
1286         struct osp_thread_info  *osi = osp_env_info(env);
1287         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
1288         struct osp_object       *o = dt2osp_obj(dt);
1289         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1290         struct thandle          *local_th;
1291         int                      rc = 0;
1292
1293         ENTRY;
1294
1295         if (is_only_remote_trans(th) && !fid_is_zero(fid)) {
1296                 LASSERT(fid_is_sane(fid));
1297
1298                 rc = osp_md_declare_object_create(env, dt, attr, hint, dof, th);
1299
1300                 RETURN(rc);
1301         }
1302
1303         /* should happen to non-0 OSP only so that at least one object
1304          * has been already declared in the scenario and LOD should
1305          * cleanup that */
1306         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_CREATE_FAIL) && d->opd_index == 1)
1307                 RETURN(-ENOSPC);
1308
1309         LASSERT(d->opd_last_used_oid_file);
1310
1311         /*
1312          * There can be gaps in precreated ids and record to unlink llog
1313          * XXX: we do not handle gaps yet, implemented before solution
1314          *      was found to be racy, so we disabled that. there is no
1315          *      point in making useless but expensive llog declaration.
1316          */
1317         /* rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th); */
1318
1319         local_th = osp_get_storage_thandle(env, th, d);
1320         if (IS_ERR(local_th))
1321                 RETURN(PTR_ERR(local_th));
1322
1323         if (unlikely(!fid_is_zero(fid))) {
1324                 /* replay case: caller knows fid */
1325                 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
1326                 osi->osi_lb.lb_len = sizeof(osi->osi_id);
1327                 osi->osi_lb.lb_buf = NULL;
1328
1329                 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
1330                                              &osi->osi_lb, osi->osi_off,
1331                                              local_th);
1332                 RETURN(rc);
1333         }
1334
1335         /*
1336          * in declaration we need to reserve object so that we don't block
1337          * awaiting precreation RPC to complete
1338          */
1339         rc = osp_precreate_reserve(env, d);
1340         /*
1341          * we also need to declare update to local "last used id" file for
1342          * recovery if object isn't used for a reason, we need to release
1343          * reservation, this can be made in osd_object_release()
1344          */
1345         if (rc == 0) {
1346                 /* mark id is reserved: in create we don't want to talk
1347                  * to OST */
1348                 LASSERT(o->opo_reserved == 0);
1349                 o->opo_reserved = 1;
1350
1351                 /* common for all OSPs file hystorically */
1352                 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
1353                 osi->osi_lb.lb_len = sizeof(osi->osi_id);
1354                 osi->osi_lb.lb_buf = NULL;
1355                 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
1356                                              &osi->osi_lb, osi->osi_off,
1357                                              local_th);
1358         } else {
1359                 /* not needed in the cache anymore */
1360                 set_bit(LU_OBJECT_HEARD_BANSHEE,
1361                             &dt->do_lu.lo_header->loh_flags);
1362         }
1363         RETURN(rc);
1364 }
1365
1366 /**
1367  * Implement OSP layer dt_object_operations::do_create() interface.
1368  *
1369  * Create the OST object.
1370  *
1371  * If the transaction is a remote transaction and the FID for the OST-object
1372  * has been assigned already, then handle it as handling MDT object via the
1373  * osp_md_object_create(). For other cases, the OSP will assign FID to the
1374  * object to be created, and update last_used Object ID (OID) file.
1375  *
1376  * \param[in] env       pointer to the thread context
1377  * \param[in] dt        pointer to the OSP layer dt_object
1378  * \param[in] attr      the attribute for the object to be created
1379  * \param[in] hint      pointer to the hint for creating the object, such as
1380  *                      the parent object
1381  * \param[in] dof       pointer to the dt_object_format for help the creation
1382  * \param[in] th        pointer to the transaction handler
1383  *
1384  * \retval              0 for success
1385  * \retval              negative error number on failure
1386  */
1387 static int osp_object_create(const struct lu_env *env, struct dt_object *dt,
1388                              struct lu_attr *attr,
1389                              struct dt_allocation_hint *hint,
1390                              struct dt_object_format *dof, struct thandle *th)
1391 {
1392         struct osp_thread_info  *osi = osp_env_info(env);
1393         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
1394         struct osp_object       *o = dt2osp_obj(dt);
1395         int                     rc = 0;
1396         struct lu_fid           *fid = &osi->osi_fid;
1397         struct thandle          *local_th;
1398         ENTRY;
1399
1400         if (is_only_remote_trans(th) &&
1401             !fid_is_zero(lu_object_fid(&dt->do_lu))) {
1402                 LASSERT(fid_is_sane(lu_object_fid(&dt->do_lu)));
1403
1404                 rc = osp_md_object_create(env, dt, attr, hint, dof, th);
1405                 if (rc == 0)
1406                         o->opo_non_exist = 0;
1407
1408                 RETURN(rc);
1409         }
1410
1411         o->opo_non_exist = 0;
1412         if (o->opo_reserved) {
1413                 /* regular case, fid is assigned holding transaction open */
1414                  osp_object_assign_fid(env, d, o);
1415         }
1416
1417         memcpy(fid, lu_object_fid(&dt->do_lu), sizeof(*fid));
1418
1419         LASSERTF(fid_is_sane(fid), "fid for osp_object %p is insane"DFID"!\n",
1420                  o, PFID(fid));
1421
1422         if (!o->opo_reserved) {
1423                 /* special case, id was assigned outside of transaction
1424                  * see comments in osp_declare_attr_set */
1425                 LASSERT(d->opd_pre != NULL);
1426                 spin_lock(&d->opd_pre_lock);
1427                 osp_update_last_fid(d, fid);
1428                 spin_unlock(&d->opd_pre_lock);
1429         }
1430
1431         CDEBUG(D_INODE, "fid for osp_object %p is "DFID"\n", o, PFID(fid));
1432
1433         /* If the precreate ends, it means it will be ready to rollover to
1434          * the new sequence soon, all the creation should be synchronized,
1435          * otherwise during replay, the replay fid will be inconsistent with
1436          * last_used/create fid */
1437         if (osp_precreate_end_seq(env, d) && osp_is_fid_client(d))
1438                 th->th_sync = 1;
1439
1440         local_th = osp_get_storage_thandle(env, th, d);
1441         if (IS_ERR(local_th))
1442                 RETURN(PTR_ERR(local_th));
1443         /*
1444          * it's OK if the import is inactive by this moment - id was created
1445          * by OST earlier, we just need to maintain it consistently on the disk
1446          * once import is reconnected, OSP will claim this and other objects
1447          * used and OST either keep them, if they exist or recreate
1448          */
1449
1450         /* we might have lost precreated objects */
1451         if (unlikely(d->opd_gap_count) > 0) {
1452                 LASSERT(d->opd_pre != NULL);
1453                 spin_lock(&d->opd_pre_lock);
1454                 if (d->opd_gap_count > 0) {
1455                         int count = d->opd_gap_count;
1456
1457                         ostid_set_id(&osi->osi_oi,
1458                                      fid_oid(&d->opd_gap_start_fid));
1459                         d->opd_gap_count = 0;
1460                         spin_unlock(&d->opd_pre_lock);
1461
1462                         CDEBUG(D_HA, "Writing gap "DFID"+%d in llog\n",
1463                                PFID(&d->opd_gap_start_fid), count);
1464                         /* real gap handling is disabled intil ORI-692 will be
1465                          * fixed, now we only report gaps */
1466                 } else {
1467                         spin_unlock(&d->opd_pre_lock);
1468                 }
1469         }
1470
1471         /* Only need update last_used oid file, seq file will only be update
1472          * during seq rollover */
1473         osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off,
1474                            &d->opd_last_used_fid.f_oid, d->opd_index);
1475
1476         rc = dt_record_write(env, d->opd_last_used_oid_file, &osi->osi_lb,
1477                              &osi->osi_off, local_th);
1478
1479         CDEBUG(D_HA, "%s: Wrote last used FID: "DFID", index %d: %d\n",
1480                d->opd_obd->obd_name, PFID(fid), d->opd_index, rc);
1481
1482         RETURN(rc);
1483 }
1484
1485 /**
1486  * Implement OSP layer dt_object_operations::do_declare_destroy() interface.
1487  *
1488  * Declare that the caller will destroy the specified OST object.
1489  *
1490  * The OST object destroy will be handled via llog asynchronously. This
1491  * function will declare the credits for generating MDS_UNLINK64_REC llog.
1492  *
1493  * \param[in] env       pointer to the thread context
1494  * \param[in] dt        pointer to the OSP layer dt_object to be destroyed
1495  * \param[in] th        pointer to the transaction handler
1496  *
1497  * \retval              0 for success
1498  * \retval              negative error number on failure
1499  */
1500 int osp_declare_object_destroy(const struct lu_env *env,
1501                                struct dt_object *dt, struct thandle *th)
1502 {
1503         struct osp_object       *o = dt2osp_obj(dt);
1504         struct osp_device       *osp = lu2osp_dev(dt->do_lu.lo_dev);
1505         int                      rc = 0;
1506
1507         ENTRY;
1508
1509         LASSERT(!osp->opd_connect_mdt);
1510         rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th);
1511
1512         RETURN(rc);
1513 }
1514
1515 /**
1516  * Implement OSP layer dt_object_operations::do_destroy() interface.
1517  *
1518  * Destroy the specified OST object.
1519  *
1520  * The OSP generates a MDS_UNLINK64_REC record in the llog. There
1521  * will be some dedicated thread to handle the llog asynchronously.
1522  *
1523  * It also marks the object as non-cached.
1524  *
1525  * \param[in] env       pointer to the thread context
1526  * \param[in] dt        pointer to the OSP layer dt_object to be destroyed
1527  * \param[in] th        pointer to the transaction handler
1528  *
1529  * \retval              0 for success
1530  * \retval              negative error number on failure
1531  */
1532 static int osp_object_destroy(const struct lu_env *env, struct dt_object *dt,
1533                               struct thandle *th)
1534 {
1535         struct osp_object       *o = dt2osp_obj(dt);
1536         struct osp_device       *osp = lu2osp_dev(dt->do_lu.lo_dev);
1537         int                      rc = 0;
1538
1539         ENTRY;
1540
1541         o->opo_non_exist = 1;
1542
1543         LASSERT(!osp->opd_connect_mdt);
1544         /* once transaction is committed put proper command on
1545          * the queue going to our OST. */
1546         rc = osp_sync_add(env, o, MDS_UNLINK64_REC, th, NULL);
1547         if (rc < 0)
1548                 RETURN(rc);
1549
1550         /* not needed in cache any more */
1551         set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
1552
1553         RETURN(rc);
1554 }
1555
1556 static int osp_orphan_index_lookup(const struct lu_env *env,
1557                                    struct dt_object *dt,
1558                                    struct dt_rec *rec,
1559                                    const struct dt_key *key)
1560 {
1561         return -EOPNOTSUPP;
1562 }
1563
1564 static int osp_orphan_index_declare_insert(const struct lu_env *env,
1565                                            struct dt_object *dt,
1566                                            const struct dt_rec *rec,
1567                                            const struct dt_key *key,
1568                                            struct thandle *handle)
1569 {
1570         return -EOPNOTSUPP;
1571 }
1572
1573 static int osp_orphan_index_insert(const struct lu_env *env,
1574                                    struct dt_object *dt,
1575                                    const struct dt_rec *rec,
1576                                    const struct dt_key *key,
1577                                    struct thandle *handle,
1578                                    int ignore_quota)
1579 {
1580         return -EOPNOTSUPP;
1581 }
1582
1583 static int osp_orphan_index_declare_delete(const struct lu_env *env,
1584                                            struct dt_object *dt,
1585                                            const struct dt_key *key,
1586                                            struct thandle *handle)
1587 {
1588         return -EOPNOTSUPP;
1589 }
1590
1591 static int osp_orphan_index_delete(const struct lu_env *env,
1592                                    struct dt_object *dt,
1593                                    const struct dt_key *key,
1594                                    struct thandle *handle)
1595 {
1596         return -EOPNOTSUPP;
1597 }
1598
1599 /**
1600  * Initialize the OSP layer index iteration.
1601  *
1602  * \param[in] env       pointer to the thread context
1603  * \param[in] dt        pointer to the index object to be iterated
1604  * \param[in] attr      unused
1605  *
1606  * \retval              pointer to the iteration structure
1607  * \retval              negative error number on failure
1608  */
1609 struct dt_it *osp_it_init(const struct lu_env *env, struct dt_object *dt,
1610                           __u32 attr)
1611 {
1612         struct osp_it *it;
1613
1614         OBD_ALLOC_PTR(it);
1615         if (it == NULL)
1616                 return ERR_PTR(-ENOMEM);
1617
1618         it->ooi_pos_ent = -1;
1619         it->ooi_obj = dt;
1620         it->ooi_attr = attr;
1621
1622         return (struct dt_it *)it;
1623 }
1624
1625 /**
1626  * Finalize the OSP layer index iteration.
1627  *
1628  * \param[in] env       pointer to the thread context
1629  * \param[in] di        pointer to the iteration structure
1630  */
1631 void osp_it_fini(const struct lu_env *env, struct dt_it *di)
1632 {
1633         struct osp_it   *it = (struct osp_it *)di;
1634         struct page     **pages = it->ooi_pages;
1635         int             npages = it->ooi_total_npages;
1636         int             i;
1637
1638         if (pages != NULL) {
1639                 for (i = 0; i < npages; i++) {
1640                         if (pages[i] != NULL) {
1641                                 if (pages[i] == it->ooi_cur_page) {
1642                                         kunmap(pages[i]);
1643                                         it->ooi_cur_page = NULL;
1644                                 }
1645                                 __free_page(pages[i]);
1646                         }
1647                 }
1648                 OBD_FREE(pages, npages * sizeof(*pages));
1649         }
1650         OBD_FREE_PTR(it);
1651 }
1652
1653 /**
1654  * Get more records for the iteration from peer.
1655  *
1656  * The new records will be filled in an array of pages. The OSP side
1657  * allows 1MB bulk data to be transferred.
1658  *
1659  * \param[in] env       pointer to the thread context
1660  * \param[in] it        pointer to the iteration structure
1661  *
1662  * \retval              0 for success
1663  * \retval              negative error number on failure
1664  */
1665 static int osp_it_fetch(const struct lu_env *env, struct osp_it *it)
1666 {
1667         struct lu_device         *dev   = it->ooi_obj->do_lu.lo_dev;
1668         struct osp_device        *osp   = lu2osp_dev(dev);
1669         struct page             **pages;
1670         struct ptlrpc_request    *req   = NULL;
1671         struct ptlrpc_bulk_desc  *desc;
1672         struct idx_info          *ii;
1673         int                       npages;
1674         int                       rc;
1675         int                       i;
1676         ENTRY;
1677
1678         /* 1MB bulk */
1679         npages = min_t(unsigned int, OFD_MAX_BRW_SIZE, 1 << 20);
1680         npages /= PAGE_SIZE;
1681
1682         OBD_ALLOC(pages, npages * sizeof(*pages));
1683         if (pages == NULL)
1684                 RETURN(-ENOMEM);
1685
1686         it->ooi_pages = pages;
1687         it->ooi_total_npages = npages;
1688         for (i = 0; i < npages; i++) {
1689                 pages[i] = alloc_page(GFP_NOFS);
1690                 if (pages[i] == NULL)
1691                         RETURN(-ENOMEM);
1692         }
1693
1694         req = ptlrpc_request_alloc(osp->opd_obd->u.cli.cl_import,
1695                                    &RQF_OBD_IDX_READ);
1696         if (req == NULL)
1697                 RETURN(-ENOMEM);
1698
1699         rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, OBD_IDX_READ);
1700         if (rc != 0) {
1701                 ptlrpc_request_free(req);
1702                 RETURN(rc);
1703         }
1704
1705         osp_set_req_replay(osp, req);
1706         req->rq_request_portal = OUT_PORTAL;
1707         ii = req_capsule_client_get(&req->rq_pill, &RMF_IDX_INFO);
1708         memset(ii, 0, sizeof(*ii));
1709         if (fid_is_last_id(lu_object_fid(&it->ooi_obj->do_lu))) {
1710                 /* LFSCK will iterate orphan object[FID_SEQ_LAYOUT_BTREE,
1711                  * ost_index, 0] with LAST_ID FID, so it needs to replace
1712                  * the FID with orphan FID here */
1713                 ii->ii_fid.f_seq = FID_SEQ_LAYOUT_RBTREE;
1714                 ii->ii_fid.f_oid = osp->opd_index;
1715                 ii->ii_fid.f_ver = 0;
1716                 ii->ii_flags = II_FL_NOHASH;
1717                 ii->ii_attrs = osp_dev2node(osp);
1718         } else {
1719                 ii->ii_fid = *lu_object_fid(&it->ooi_obj->do_lu);
1720                 ii->ii_flags = II_FL_NOHASH | II_FL_NOKEY | II_FL_VARKEY |
1721                                II_FL_VARREC;
1722                 ii->ii_attrs = it->ooi_attr;
1723         }
1724         ii->ii_magic = IDX_INFO_MAGIC;
1725         ii->ii_count = npages * LU_PAGE_COUNT;
1726         ii->ii_hash_start = it->ooi_next;
1727
1728         ptlrpc_at_set_req_timeout(req);
1729
1730         desc = ptlrpc_prep_bulk_imp(req, npages, 1,
1731                                     PTLRPC_BULK_PUT_SINK | PTLRPC_BULK_BUF_KIOV,
1732                                     MDS_BULK_PORTAL,
1733                                     &ptlrpc_bulk_kiov_pin_ops);
1734         if (desc == NULL) {
1735                 ptlrpc_request_free(req);
1736                 RETURN(-ENOMEM);
1737         }
1738
1739         for (i = 0; i < npages; i++)
1740                 desc->bd_frag_ops->add_kiov_frag(desc, pages[i], 0,
1741                                                  PAGE_SIZE);
1742
1743         ptlrpc_request_set_replen(req);
1744         rc = ptlrpc_queue_wait(req);
1745         if (rc != 0)
1746                 GOTO(out, rc);
1747
1748         rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk,
1749                                           req->rq_bulk->bd_nob_transferred);
1750         if (rc < 0)
1751                 GOTO(out, rc);
1752         rc = 0;
1753
1754         ii = req_capsule_server_get(&req->rq_pill, &RMF_IDX_INFO);
1755         if (ii->ii_magic != IDX_INFO_MAGIC)
1756                  GOTO(out, rc = -EPROTO);
1757
1758         npages = (ii->ii_count + LU_PAGE_COUNT - 1) >>
1759                  (PAGE_SHIFT - LU_PAGE_SHIFT);
1760         if (npages > it->ooi_total_npages) {
1761                 CERROR("%s: returned more pages than expected, %u > %u\n",
1762                        osp->opd_obd->obd_name, npages, it->ooi_total_npages);
1763                 GOTO(out, rc = -EINVAL);
1764         }
1765
1766         it->ooi_valid_npages = npages;
1767         if (ptlrpc_rep_need_swab(req))
1768                 it->ooi_swab = 1;
1769
1770         it->ooi_next = ii->ii_hash_end;
1771
1772 out:
1773         ptlrpc_req_finished(req);
1774
1775         return rc;
1776 }
1777
1778 /**
1779  * Move the iteration cursor to the next lu_page.
1780  *
1781  * One system page (PAGE_SIZE) may contain multiple lu_page (4KB),
1782  * that depends on the LU_PAGE_COUNT. If it is not the last lu_page
1783  * in current system page, then move the iteration cursor to the next
1784  * lu_page in current system page. Otherwise, if there are more system
1785  * pages in the cache, then move the iteration cursor to the next system
1786  * page. If all the cached records (pages) have been iterated, then fetch
1787  * more records via osp_it_fetch().
1788  *
1789  * \param[in] env       pointer to the thread context
1790  * \param[in] di        pointer to the iteration structure
1791  *
1792  * \retval              positive for end of the directory
1793  * \retval              0 for success
1794  * \retval              negative error number on failure
1795  */
1796 int osp_it_next_page(const struct lu_env *env, struct dt_it *di)
1797 {
1798         struct osp_it           *it = (struct osp_it *)di;
1799         struct lu_idxpage       *idxpage;
1800         struct page             **pages;
1801         int                     rc;
1802         int                     i;
1803         ENTRY;
1804
1805 again2:
1806         idxpage = it->ooi_cur_idxpage;
1807         if (idxpage != NULL) {
1808                 if (idxpage->lip_nr == 0)
1809                         RETURN(1);
1810
1811                 if (it->ooi_pos_ent < idxpage->lip_nr) {
1812                         CDEBUG(D_INFO, "ooi_pos %d nr %d\n",
1813                                (int)it->ooi_pos_ent, (int)idxpage->lip_nr);
1814                         RETURN(0);
1815                 }
1816                 it->ooi_cur_idxpage = NULL;
1817                 it->ooi_pos_lu_page++;
1818
1819 again1:
1820                 if (it->ooi_pos_lu_page < LU_PAGE_COUNT) {
1821                         it->ooi_cur_idxpage = (void *)it->ooi_cur_page +
1822                                          LU_PAGE_SIZE * it->ooi_pos_lu_page;
1823                         if (it->ooi_swab)
1824                                 lustre_swab_lip_header(it->ooi_cur_idxpage);
1825                         if (it->ooi_cur_idxpage->lip_magic != LIP_MAGIC) {
1826                                 struct osp_device *osp =
1827                                         lu2osp_dev(it->ooi_obj->do_lu.lo_dev);
1828
1829                                 CERROR("%s: invalid magic (%x != %x) for page "
1830                                        "%d/%d while read layout orphan index\n",
1831                                        osp->opd_obd->obd_name,
1832                                        it->ooi_cur_idxpage->lip_magic,
1833                                        LIP_MAGIC, it->ooi_pos_page,
1834                                        it->ooi_pos_lu_page);
1835                                 /* Skip this lu_page next time. */
1836                                 it->ooi_pos_ent = idxpage->lip_nr - 1;
1837                                 RETURN(-EINVAL);
1838                         }
1839                         it->ooi_pos_ent = -1;
1840                         goto again2;
1841                 }
1842
1843                 kunmap(it->ooi_cur_page);
1844                 it->ooi_cur_page = NULL;
1845                 it->ooi_pos_page++;
1846
1847 again0:
1848                 pages = it->ooi_pages;
1849                 if (it->ooi_pos_page < it->ooi_valid_npages) {
1850                         it->ooi_cur_page = kmap(pages[it->ooi_pos_page]);
1851                         it->ooi_pos_lu_page = 0;
1852                         goto again1;
1853                 }
1854
1855                 for (i = 0; i < it->ooi_total_npages; i++) {
1856                         if (pages[i] != NULL)
1857                                 __free_page(pages[i]);
1858                 }
1859                 OBD_FREE(pages, it->ooi_total_npages * sizeof(*pages));
1860
1861                 it->ooi_pos_page = 0;
1862                 it->ooi_total_npages = 0;
1863                 it->ooi_valid_npages = 0;
1864                 it->ooi_swab = 0;
1865                 it->ooi_ent = NULL;
1866                 it->ooi_cur_page = NULL;
1867                 it->ooi_cur_idxpage = NULL;
1868                 it->ooi_pages = NULL;
1869         }
1870
1871         if (it->ooi_next == II_END_OFF)
1872                 RETURN(1);
1873
1874         rc = osp_it_fetch(env, it);
1875         if (rc == 0)
1876                 goto again0;
1877
1878         RETURN(rc);
1879 }
1880
1881 /**
1882  * Move the iteration cursor to the next record.
1883  *
1884  * If there are more records in the lu_page, then move the iteration
1885  * cursor to the next record directly. Otherwise, move the iteration
1886  * cursor to the record in the next lu_page via osp_it_next_page()
1887  *
1888  * \param[in] env       pointer to the thread context
1889  * \param[in] di        pointer to the iteration structure
1890  *
1891  * \retval              positive for end of the directory
1892  * \retval              0 for success
1893  * \retval              negative error number on failure
1894  */
1895 static int osp_orphan_it_next(const struct lu_env *env, struct dt_it *di)
1896 {
1897         struct osp_it           *it = (struct osp_it *)di;
1898         struct lu_idxpage       *idxpage;
1899         int                     rc;
1900         ENTRY;
1901
1902 again:
1903         idxpage = it->ooi_cur_idxpage;
1904         if (idxpage != NULL) {
1905                 if (idxpage->lip_nr == 0)
1906                         RETURN(1);
1907
1908                 it->ooi_pos_ent++;
1909                 if (it->ooi_pos_ent < idxpage->lip_nr) {
1910                         it->ooi_ent =
1911                                 (struct lu_orphan_ent *)idxpage->lip_entries +
1912                                                         it->ooi_pos_ent;
1913                         if (it->ooi_swab)
1914                                 lustre_swab_orphan_ent(it->ooi_ent);
1915                         RETURN(0);
1916                 }
1917         }
1918
1919         rc = osp_it_next_page(env, di);
1920         if (rc == 0)
1921                 goto again;
1922
1923         RETURN(rc);
1924 }
1925
1926 int osp_it_get(const struct lu_env *env, struct dt_it *di,
1927                const struct dt_key *key)
1928 {
1929         return 1;
1930 }
1931
1932 void osp_it_put(const struct lu_env *env, struct dt_it *di)
1933 {
1934 }
1935
1936 static struct dt_key *osp_orphan_it_key(const struct lu_env *env,
1937                                         const struct dt_it *di)
1938 {
1939         struct osp_it   *it  = (struct osp_it *)di;
1940         struct lu_orphan_ent    *ent = (struct lu_orphan_ent *)it->ooi_ent;
1941
1942         if (likely(ent != NULL))
1943                 return (struct dt_key *)(&ent->loe_key);
1944
1945         return NULL;
1946 }
1947
1948 static int osp_orphan_it_key_size(const struct lu_env *env,
1949                                   const struct dt_it *di)
1950 {
1951         return sizeof(struct lu_fid);
1952 }
1953
1954 static int osp_orphan_it_rec(const struct lu_env *env, const struct dt_it *di,
1955                              struct dt_rec *rec, __u32 attr)
1956 {
1957         struct osp_it   *it  = (struct osp_it *)di;
1958         struct lu_orphan_ent    *ent = (struct lu_orphan_ent *)it->ooi_ent;
1959
1960         if (likely(ent != NULL)) {
1961                 *(struct lu_orphan_rec *)rec = ent->loe_rec;
1962                 return 0;
1963         }
1964
1965         return -EINVAL;
1966 }
1967
1968 __u64 osp_it_store(const struct lu_env *env, const struct dt_it *di)
1969 {
1970         struct osp_it   *it = (struct osp_it *)di;
1971
1972         return it->ooi_next;
1973 }
1974
1975 /**
1976  * Locate the iteration cursor to the specified position (cookie).
1977  *
1978  * \param[in] env       pointer to the thread context
1979  * \param[in] di        pointer to the iteration structure
1980  * \param[in] hash      the specified position
1981  *
1982  * \retval              positive number for locating to the exactly position
1983  *                      or the next
1984  * \retval              0 for arriving at the end of the iteration
1985  * \retval              negative error number on failure
1986  */
1987 int osp_orphan_it_load(const struct lu_env *env, const struct dt_it *di,
1988                        __u64 hash)
1989 {
1990         struct osp_it   *it     = (struct osp_it *)di;
1991         int              rc;
1992
1993         it->ooi_next = hash;
1994         rc = osp_orphan_it_next(env, (struct dt_it *)di);
1995         if (rc == 1)
1996                 return 0;
1997
1998         if (rc == 0)
1999                 return 1;
2000
2001         return rc;
2002 }
2003
2004 int osp_it_key_rec(const struct lu_env *env, const struct dt_it *di,
2005                    void *key_rec)
2006 {
2007         return 0;
2008 }
2009
2010 static const struct dt_index_operations osp_orphan_index_ops = {
2011         .dio_lookup             = osp_orphan_index_lookup,
2012         .dio_declare_insert     = osp_orphan_index_declare_insert,
2013         .dio_insert             = osp_orphan_index_insert,
2014         .dio_declare_delete     = osp_orphan_index_declare_delete,
2015         .dio_delete             = osp_orphan_index_delete,
2016         .dio_it = {
2017                 .init           = osp_it_init,
2018                 .fini           = osp_it_fini,
2019                 .next           = osp_orphan_it_next,
2020                 .get            = osp_it_get,
2021                 .put            = osp_it_put,
2022                 .key            = osp_orphan_it_key,
2023                 .key_size       = osp_orphan_it_key_size,
2024                 .rec            = osp_orphan_it_rec,
2025                 .store          = osp_it_store,
2026                 .load           = osp_orphan_it_load,
2027                 .key_rec        = osp_it_key_rec,
2028         }
2029 };
2030
2031 /**
2032  * Implement OSP layer dt_object_operations::do_index_try() interface.
2033  *
2034  * Negotiate the index type.
2035  *
2036  * If the target index is an IDIF object, then use osp_orphan_index_ops.
2037  * Otherwise, assign osp_md_index_ops to the dt_object::do_index_ops.
2038  * (\see lustre/include/lustre_fid.h for IDIF.)
2039  *
2040  * \param[in] env       pointer to the thread context
2041  * \param[in] dt        pointer to the OSP layer dt_object
2042  * \param[in] feat      unused
2043  *
2044  * \retval              0 for success
2045  */
2046 static int osp_index_try(const struct lu_env *env,
2047                          struct dt_object *dt,
2048                          const struct dt_index_features *feat)
2049 {
2050         const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
2051
2052         if (fid_is_last_id(fid) && fid_is_idif(fid))
2053                 dt->do_index_ops = &osp_orphan_index_ops;
2054         else
2055                 dt->do_index_ops = &osp_md_index_ops;
2056         return 0;
2057 }
2058
2059 static struct dt_object_operations osp_obj_ops = {
2060         .do_declare_attr_get    = osp_declare_attr_get,
2061         .do_attr_get            = osp_attr_get,
2062         .do_declare_attr_set    = osp_declare_attr_set,
2063         .do_attr_set            = osp_attr_set,
2064         .do_declare_xattr_get   = osp_declare_xattr_get,
2065         .do_xattr_get           = osp_xattr_get,
2066         .do_declare_xattr_set   = osp_declare_xattr_set,
2067         .do_xattr_set           = osp_xattr_set,
2068         .do_declare_create      = osp_declare_object_create,
2069         .do_create              = osp_object_create,
2070         .do_declare_destroy     = osp_declare_object_destroy,
2071         .do_destroy             = osp_object_destroy,
2072         .do_index_try           = osp_index_try,
2073 };
2074
2075 /**
2076  * Implement OSP layer lu_object_operations::loo_object_init() interface.
2077  *
2078  * Initialize the object.
2079  *
2080  * If it is a remote MDT object, then call do_attr_get() to fetch
2081  * the attribute from the peer.
2082  *
2083  * \param[in] env       pointer to the thread context
2084  * \param[in] o         pointer to the OSP layer lu_object
2085  * \param[in] conf      unused
2086  *
2087  * \retval              0 for success
2088  * \retval              negative error number on failure
2089  */
2090 static int osp_object_init(const struct lu_env *env, struct lu_object *o,
2091                            const struct lu_object_conf *conf)
2092 {
2093         struct osp_object       *po = lu2osp_obj(o);
2094         int                     rc = 0;
2095         ENTRY;
2096
2097         spin_lock_init(&po->opo_lock);
2098         o->lo_header->loh_attr |= LOHA_REMOTE;
2099         INIT_LIST_HEAD(&po->opo_xattr_list);
2100         INIT_LIST_HEAD(&po->opo_invalidate_cb_list);
2101
2102         if (is_ost_obj(o)) {
2103                 po->opo_obj.do_ops = &osp_obj_ops;
2104         } else {
2105                 struct lu_attr *la = &osp_env_info(env)->osi_attr;
2106
2107                 po->opo_obj.do_ops = &osp_md_obj_ops;
2108                 po->opo_obj.do_body_ops = &osp_md_body_ops;
2109
2110                 if (conf != NULL && conf->loc_flags & LOC_F_NEW) {
2111                         po->opo_non_exist = 1;
2112                 } else {
2113                         rc = po->opo_obj.do_ops->do_attr_get(env, lu2dt_obj(o),
2114                                                              la);
2115                         if (rc == 0)
2116                                 o->lo_header->loh_attr |=
2117                                         LOHA_EXISTS | (la->la_mode & S_IFMT);
2118                         if (rc == -ENOENT) {
2119                                 po->opo_non_exist = 1;
2120                                 rc = 0;
2121                         }
2122                 }
2123                 init_rwsem(&po->opo_sem);
2124         }
2125         RETURN(rc);
2126 }
2127
2128 /**
2129  * Implement OSP layer lu_object_operations::loo_object_free() interface.
2130  *
2131  * Finalize the object.
2132  *
2133  * If the OSP object has attributes cache, then destroy the cache.
2134  * Free the object finally.
2135  *
2136  * \param[in] env       pointer to the thread context
2137  * \param[in] o         pointer to the OSP layer lu_object
2138  */
2139 static void osp_object_free(const struct lu_env *env, struct lu_object *o)
2140 {
2141         struct osp_object       *obj = lu2osp_obj(o);
2142         struct lu_object_header *h = o->lo_header;
2143         struct osp_xattr_entry *oxe;
2144         struct osp_xattr_entry *tmp;
2145         int                     count;
2146
2147         dt_object_fini(&obj->opo_obj);
2148         lu_object_header_fini(h);
2149         list_for_each_entry_safe(oxe, tmp, &obj->opo_xattr_list, oxe_list) {
2150                 list_del(&oxe->oxe_list);
2151                 count = atomic_read(&oxe->oxe_ref);
2152                 LASSERTF(count == 1,
2153                          "Still has %d users on the xattr entry %.*s\n",
2154                          count-1, (int)oxe->oxe_namelen, oxe->oxe_buf);
2155
2156                 OBD_FREE(oxe, oxe->oxe_buflen);
2157         }
2158         OBD_SLAB_FREE_PTR(obj, osp_object_kmem);
2159 }
2160
2161 /**
2162  * Implement OSP layer lu_object_operations::loo_object_release() interface.
2163  *
2164  * Cleanup (not free) the object.
2165  *
2166  * If it is a reserved object but failed to be created, or it is an OST
2167  * object, then mark the object as non-cached.
2168  *
2169  * \param[in] env       pointer to the thread context
2170  * \param[in] o         pointer to the OSP layer lu_object
2171  */
2172 static void osp_object_release(const struct lu_env *env, struct lu_object *o)
2173 {
2174         struct osp_object       *po = lu2osp_obj(o);
2175         struct osp_device       *d  = lu2osp_dev(o->lo_dev);
2176
2177         ENTRY;
2178
2179         /*
2180          * release reservation if object was declared but not created
2181          * this may require lu_object_put() in LOD
2182          */
2183         if (unlikely(po->opo_reserved)) {
2184                 LASSERT(d->opd_pre != NULL);
2185                 LASSERT(d->opd_pre_reserved > 0);
2186                 spin_lock(&d->opd_pre_lock);
2187                 d->opd_pre_reserved--;
2188                 spin_unlock(&d->opd_pre_lock);
2189
2190                 /* not needed in cache any more */
2191                 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
2192         }
2193
2194         if (is_ost_obj(o))
2195                 /* XXX: Currently, NOT cache OST-object on MDT because:
2196                  *      1. it is not often accessed on MDT.
2197                  *      2. avoid up layer (such as LFSCK) to load too many
2198                  *         once-used OST-objects. */
2199                 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
2200
2201         EXIT;
2202 }
2203
2204 static int osp_object_print(const struct lu_env *env, void *cookie,
2205                             lu_printer_t p, const struct lu_object *l)
2206 {
2207         const struct osp_object *o = lu2osp_obj((struct lu_object *)l);
2208
2209         return (*p)(env, cookie, LUSTRE_OSP_NAME"-object@%p", o);
2210 }
2211
2212 static int osp_object_invariant(const struct lu_object *o)
2213 {
2214         LBUG();
2215 }
2216
2217 struct lu_object_operations osp_lu_obj_ops = {
2218         .loo_object_init        = osp_object_init,
2219         .loo_object_free        = osp_object_free,
2220         .loo_object_release     = osp_object_release,
2221         .loo_object_print       = osp_object_print,
2222         .loo_object_invariant   = osp_object_invariant
2223 };