Whamcloud - gitweb
LU-6245 libcfs: remove wrappers for timer functions
[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         CDEBUG(D_HA, "Invalidate osp_object "DFID"\n",
1247                PFID(lu_object_fid(&dt->do_lu)));
1248         osp_obj_invalidate_cache(obj);
1249
1250         spin_lock(&obj->opo_lock);
1251         obj->opo_stale = 1;
1252         spin_unlock(&obj->opo_lock);
1253
1254         RETURN(0);
1255 }
1256
1257 /**
1258  * Implement OSP layer dt_object_operations::do_declare_create() interface.
1259  *
1260  * Declare that the caller will create the OST object.
1261  *
1262  * If the transaction is a remote transaction and the FID for the OST-object
1263  * has been assigned already, then handle it as creating (remote) MDT object
1264  * via osp_md_declare_object_create(). This function is usually used for LFSCK
1265  * to re-create the lost OST object. Otherwise, if it is not replay case, the
1266  * OSP will reserve pre-created object for the subsequent create operation;
1267  * if the MDT side cached pre-created objects are less than some threshold,
1268  * then it will wakeup the pre-create thread.
1269  *
1270  * \param[in] env       pointer to the thread context
1271  * \param[in] dt        pointer to the OSP layer dt_object
1272  * \param[in] attr      the attribute for the object to be created
1273  * \param[in] hint      pointer to the hint for creating the object, such as
1274  *                      the parent object
1275  * \param[in] dof       pointer to the dt_object_format for help the creation
1276  * \param[in] th        pointer to the transaction handler
1277  *
1278  * \retval              0 for success
1279  * \retval              negative error number on failure
1280  */
1281 static int osp_declare_object_create(const struct lu_env *env,
1282                                      struct dt_object *dt,
1283                                      struct lu_attr *attr,
1284                                      struct dt_allocation_hint *hint,
1285                                      struct dt_object_format *dof,
1286                                      struct thandle *th)
1287 {
1288         struct osp_thread_info  *osi = osp_env_info(env);
1289         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
1290         struct osp_object       *o = dt2osp_obj(dt);
1291         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1292         struct thandle          *local_th;
1293         int                      rc = 0;
1294
1295         ENTRY;
1296
1297         if (is_only_remote_trans(th) && !fid_is_zero(fid)) {
1298                 LASSERT(fid_is_sane(fid));
1299
1300                 rc = osp_md_declare_object_create(env, dt, attr, hint, dof, th);
1301
1302                 RETURN(rc);
1303         }
1304
1305         /* should happen to non-0 OSP only so that at least one object
1306          * has been already declared in the scenario and LOD should
1307          * cleanup that */
1308         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_CREATE_FAIL) && d->opd_index == 1)
1309                 RETURN(-ENOSPC);
1310
1311         LASSERT(d->opd_last_used_oid_file);
1312
1313         /*
1314          * There can be gaps in precreated ids and record to unlink llog
1315          * XXX: we do not handle gaps yet, implemented before solution
1316          *      was found to be racy, so we disabled that. there is no
1317          *      point in making useless but expensive llog declaration.
1318          */
1319         /* rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th); */
1320
1321         local_th = osp_get_storage_thandle(env, th, d);
1322         if (IS_ERR(local_th))
1323                 RETURN(PTR_ERR(local_th));
1324
1325         if (unlikely(!fid_is_zero(fid))) {
1326                 /* replay case: caller knows fid */
1327                 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
1328                 osi->osi_lb.lb_len = sizeof(osi->osi_id);
1329                 osi->osi_lb.lb_buf = NULL;
1330
1331                 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
1332                                              &osi->osi_lb, osi->osi_off,
1333                                              local_th);
1334                 RETURN(rc);
1335         }
1336
1337         /*
1338          * in declaration we need to reserve object so that we don't block
1339          * awaiting precreation RPC to complete
1340          */
1341         rc = osp_precreate_reserve(env, d);
1342         /*
1343          * we also need to declare update to local "last used id" file for
1344          * recovery if object isn't used for a reason, we need to release
1345          * reservation, this can be made in osd_object_release()
1346          */
1347         if (rc == 0) {
1348                 /* mark id is reserved: in create we don't want to talk
1349                  * to OST */
1350                 LASSERT(o->opo_reserved == 0);
1351                 o->opo_reserved = 1;
1352
1353                 /* common for all OSPs file hystorically */
1354                 osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
1355                 osi->osi_lb.lb_len = sizeof(osi->osi_id);
1356                 osi->osi_lb.lb_buf = NULL;
1357                 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
1358                                              &osi->osi_lb, osi->osi_off,
1359                                              local_th);
1360         } else {
1361                 /* not needed in the cache anymore */
1362                 set_bit(LU_OBJECT_HEARD_BANSHEE,
1363                             &dt->do_lu.lo_header->loh_flags);
1364         }
1365         RETURN(rc);
1366 }
1367
1368 /**
1369  * Implement OSP layer dt_object_operations::do_create() interface.
1370  *
1371  * Create the OST object.
1372  *
1373  * If the transaction is a remote transaction and the FID for the OST-object
1374  * has been assigned already, then handle it as handling MDT object via the
1375  * osp_md_object_create(). For other cases, the OSP will assign FID to the
1376  * object to be created, and update last_used Object ID (OID) file.
1377  *
1378  * \param[in] env       pointer to the thread context
1379  * \param[in] dt        pointer to the OSP layer dt_object
1380  * \param[in] attr      the attribute for the object to be created
1381  * \param[in] hint      pointer to the hint for creating the object, such as
1382  *                      the parent object
1383  * \param[in] dof       pointer to the dt_object_format for help the creation
1384  * \param[in] th        pointer to the transaction handler
1385  *
1386  * \retval              0 for success
1387  * \retval              negative error number on failure
1388  */
1389 static int osp_object_create(const struct lu_env *env, struct dt_object *dt,
1390                              struct lu_attr *attr,
1391                              struct dt_allocation_hint *hint,
1392                              struct dt_object_format *dof, struct thandle *th)
1393 {
1394         struct osp_thread_info  *osi = osp_env_info(env);
1395         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
1396         struct osp_object       *o = dt2osp_obj(dt);
1397         int                     rc = 0;
1398         struct lu_fid           *fid = &osi->osi_fid;
1399         struct thandle          *local_th;
1400         ENTRY;
1401
1402         if (is_only_remote_trans(th) &&
1403             !fid_is_zero(lu_object_fid(&dt->do_lu))) {
1404                 LASSERT(fid_is_sane(lu_object_fid(&dt->do_lu)));
1405
1406                 rc = osp_md_object_create(env, dt, attr, hint, dof, th);
1407                 if (rc == 0)
1408                         o->opo_non_exist = 0;
1409
1410                 RETURN(rc);
1411         }
1412
1413         o->opo_non_exist = 0;
1414         if (o->opo_reserved) {
1415                 /* regular case, fid is assigned holding transaction open */
1416                  osp_object_assign_fid(env, d, o);
1417         }
1418
1419         memcpy(fid, lu_object_fid(&dt->do_lu), sizeof(*fid));
1420
1421         LASSERTF(fid_is_sane(fid), "fid for osp_object %p is insane"DFID"!\n",
1422                  o, PFID(fid));
1423
1424         if (!o->opo_reserved) {
1425                 /* special case, id was assigned outside of transaction
1426                  * see comments in osp_declare_attr_set */
1427                 LASSERT(d->opd_pre != NULL);
1428                 spin_lock(&d->opd_pre_lock);
1429                 osp_update_last_fid(d, fid);
1430                 spin_unlock(&d->opd_pre_lock);
1431         }
1432
1433         CDEBUG(D_INODE, "fid for osp_object %p is "DFID"\n", o, PFID(fid));
1434
1435         /* If the precreate ends, it means it will be ready to rollover to
1436          * the new sequence soon, all the creation should be synchronized,
1437          * otherwise during replay, the replay fid will be inconsistent with
1438          * last_used/create fid */
1439         if (osp_precreate_end_seq(env, d) && osp_is_fid_client(d))
1440                 th->th_sync = 1;
1441
1442         local_th = osp_get_storage_thandle(env, th, d);
1443         if (IS_ERR(local_th))
1444                 RETURN(PTR_ERR(local_th));
1445         /*
1446          * it's OK if the import is inactive by this moment - id was created
1447          * by OST earlier, we just need to maintain it consistently on the disk
1448          * once import is reconnected, OSP will claim this and other objects
1449          * used and OST either keep them, if they exist or recreate
1450          */
1451
1452         /* we might have lost precreated objects */
1453         if (unlikely(d->opd_gap_count) > 0) {
1454                 LASSERT(d->opd_pre != NULL);
1455                 spin_lock(&d->opd_pre_lock);
1456                 if (d->opd_gap_count > 0) {
1457                         int count = d->opd_gap_count;
1458
1459                         ostid_set_id(&osi->osi_oi,
1460                                      fid_oid(&d->opd_gap_start_fid));
1461                         d->opd_gap_count = 0;
1462                         spin_unlock(&d->opd_pre_lock);
1463
1464                         CDEBUG(D_HA, "Writing gap "DFID"+%d in llog\n",
1465                                PFID(&d->opd_gap_start_fid), count);
1466                         /* real gap handling is disabled intil ORI-692 will be
1467                          * fixed, now we only report gaps */
1468                 } else {
1469                         spin_unlock(&d->opd_pre_lock);
1470                 }
1471         }
1472
1473         /* Only need update last_used oid file, seq file will only be update
1474          * during seq rollover */
1475         osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off,
1476                            &d->opd_last_used_fid.f_oid, d->opd_index);
1477
1478         rc = dt_record_write(env, d->opd_last_used_oid_file, &osi->osi_lb,
1479                              &osi->osi_off, local_th);
1480
1481         CDEBUG(D_HA, "%s: Wrote last used FID: "DFID", index %d: %d\n",
1482                d->opd_obd->obd_name, PFID(fid), d->opd_index, rc);
1483
1484         RETURN(rc);
1485 }
1486
1487 /**
1488  * Implement OSP layer dt_object_operations::do_declare_destroy() interface.
1489  *
1490  * Declare that the caller will destroy the specified OST object.
1491  *
1492  * The OST object destroy will be handled via llog asynchronously. This
1493  * function will declare the credits for generating MDS_UNLINK64_REC llog.
1494  *
1495  * \param[in] env       pointer to the thread context
1496  * \param[in] dt        pointer to the OSP layer dt_object to be destroyed
1497  * \param[in] th        pointer to the transaction handler
1498  *
1499  * \retval              0 for success
1500  * \retval              negative error number on failure
1501  */
1502 int osp_declare_object_destroy(const struct lu_env *env,
1503                                struct dt_object *dt, struct thandle *th)
1504 {
1505         struct osp_object       *o = dt2osp_obj(dt);
1506         struct osp_device       *osp = lu2osp_dev(dt->do_lu.lo_dev);
1507         int                      rc = 0;
1508
1509         ENTRY;
1510
1511         LASSERT(!osp->opd_connect_mdt);
1512         rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th);
1513
1514         RETURN(rc);
1515 }
1516
1517 /**
1518  * Implement OSP layer dt_object_operations::do_destroy() interface.
1519  *
1520  * Destroy the specified OST object.
1521  *
1522  * The OSP generates a MDS_UNLINK64_REC record in the llog. There
1523  * will be some dedicated thread to handle the llog asynchronously.
1524  *
1525  * It also marks the object as non-cached.
1526  *
1527  * \param[in] env       pointer to the thread context
1528  * \param[in] dt        pointer to the OSP layer dt_object to be destroyed
1529  * \param[in] th        pointer to the transaction handler
1530  *
1531  * \retval              0 for success
1532  * \retval              negative error number on failure
1533  */
1534 static int osp_object_destroy(const struct lu_env *env, struct dt_object *dt,
1535                               struct thandle *th)
1536 {
1537         struct osp_object       *o = dt2osp_obj(dt);
1538         struct osp_device       *osp = lu2osp_dev(dt->do_lu.lo_dev);
1539         int                      rc = 0;
1540
1541         ENTRY;
1542
1543         o->opo_non_exist = 1;
1544
1545         LASSERT(!osp->opd_connect_mdt);
1546         /* once transaction is committed put proper command on
1547          * the queue going to our OST. */
1548         rc = osp_sync_add(env, o, MDS_UNLINK64_REC, th, NULL);
1549         if (rc < 0)
1550                 RETURN(rc);
1551
1552         /* not needed in cache any more */
1553         set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
1554
1555         RETURN(rc);
1556 }
1557
1558 static int osp_orphan_index_lookup(const struct lu_env *env,
1559                                    struct dt_object *dt,
1560                                    struct dt_rec *rec,
1561                                    const struct dt_key *key)
1562 {
1563         return -EOPNOTSUPP;
1564 }
1565
1566 static int osp_orphan_index_declare_insert(const struct lu_env *env,
1567                                            struct dt_object *dt,
1568                                            const struct dt_rec *rec,
1569                                            const struct dt_key *key,
1570                                            struct thandle *handle)
1571 {
1572         return -EOPNOTSUPP;
1573 }
1574
1575 static int osp_orphan_index_insert(const struct lu_env *env,
1576                                    struct dt_object *dt,
1577                                    const struct dt_rec *rec,
1578                                    const struct dt_key *key,
1579                                    struct thandle *handle,
1580                                    int ignore_quota)
1581 {
1582         return -EOPNOTSUPP;
1583 }
1584
1585 static int osp_orphan_index_declare_delete(const struct lu_env *env,
1586                                            struct dt_object *dt,
1587                                            const struct dt_key *key,
1588                                            struct thandle *handle)
1589 {
1590         return -EOPNOTSUPP;
1591 }
1592
1593 static int osp_orphan_index_delete(const struct lu_env *env,
1594                                    struct dt_object *dt,
1595                                    const struct dt_key *key,
1596                                    struct thandle *handle)
1597 {
1598         return -EOPNOTSUPP;
1599 }
1600
1601 /**
1602  * Initialize the OSP layer index iteration.
1603  *
1604  * \param[in] env       pointer to the thread context
1605  * \param[in] dt        pointer to the index object to be iterated
1606  * \param[in] attr      unused
1607  *
1608  * \retval              pointer to the iteration structure
1609  * \retval              negative error number on failure
1610  */
1611 struct dt_it *osp_it_init(const struct lu_env *env, struct dt_object *dt,
1612                           __u32 attr)
1613 {
1614         struct osp_it *it;
1615
1616         OBD_ALLOC_PTR(it);
1617         if (it == NULL)
1618                 return ERR_PTR(-ENOMEM);
1619
1620         it->ooi_pos_ent = -1;
1621         it->ooi_obj = dt;
1622         it->ooi_attr = attr;
1623
1624         return (struct dt_it *)it;
1625 }
1626
1627 /**
1628  * Finalize the OSP layer index iteration.
1629  *
1630  * \param[in] env       pointer to the thread context
1631  * \param[in] di        pointer to the iteration structure
1632  */
1633 void osp_it_fini(const struct lu_env *env, struct dt_it *di)
1634 {
1635         struct osp_it   *it = (struct osp_it *)di;
1636         struct page     **pages = it->ooi_pages;
1637         int             npages = it->ooi_total_npages;
1638         int             i;
1639
1640         if (pages != NULL) {
1641                 for (i = 0; i < npages; i++) {
1642                         if (pages[i] != NULL) {
1643                                 if (pages[i] == it->ooi_cur_page) {
1644                                         kunmap(pages[i]);
1645                                         it->ooi_cur_page = NULL;
1646                                 }
1647                                 __free_page(pages[i]);
1648                         }
1649                 }
1650                 OBD_FREE(pages, npages * sizeof(*pages));
1651         }
1652         OBD_FREE_PTR(it);
1653 }
1654
1655 /**
1656  * Get more records for the iteration from peer.
1657  *
1658  * The new records will be filled in an array of pages. The OSP side
1659  * allows 1MB bulk data to be transferred.
1660  *
1661  * \param[in] env       pointer to the thread context
1662  * \param[in] it        pointer to the iteration structure
1663  *
1664  * \retval              0 for success
1665  * \retval              negative error number on failure
1666  */
1667 static int osp_it_fetch(const struct lu_env *env, struct osp_it *it)
1668 {
1669         struct lu_device         *dev   = it->ooi_obj->do_lu.lo_dev;
1670         struct osp_device        *osp   = lu2osp_dev(dev);
1671         struct page             **pages;
1672         struct ptlrpc_request    *req   = NULL;
1673         struct ptlrpc_bulk_desc  *desc;
1674         struct idx_info          *ii;
1675         int                       npages;
1676         int                       rc;
1677         int                       i;
1678         ENTRY;
1679
1680         /* 1MB bulk */
1681         npages = min_t(unsigned int, OFD_MAX_BRW_SIZE, 1 << 20);
1682         npages /= PAGE_SIZE;
1683
1684         OBD_ALLOC(pages, npages * sizeof(*pages));
1685         if (pages == NULL)
1686                 RETURN(-ENOMEM);
1687
1688         it->ooi_pages = pages;
1689         it->ooi_total_npages = npages;
1690         for (i = 0; i < npages; i++) {
1691                 pages[i] = alloc_page(GFP_NOFS);
1692                 if (pages[i] == NULL)
1693                         RETURN(-ENOMEM);
1694         }
1695
1696         req = ptlrpc_request_alloc(osp->opd_obd->u.cli.cl_import,
1697                                    &RQF_OBD_IDX_READ);
1698         if (req == NULL)
1699                 RETURN(-ENOMEM);
1700
1701         rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, OBD_IDX_READ);
1702         if (rc != 0) {
1703                 ptlrpc_request_free(req);
1704                 RETURN(rc);
1705         }
1706
1707         osp_set_req_replay(osp, req);
1708         req->rq_request_portal = OUT_PORTAL;
1709         ii = req_capsule_client_get(&req->rq_pill, &RMF_IDX_INFO);
1710         memset(ii, 0, sizeof(*ii));
1711         if (fid_is_last_id(lu_object_fid(&it->ooi_obj->do_lu))) {
1712                 /* LFSCK will iterate orphan object[FID_SEQ_LAYOUT_BTREE,
1713                  * ost_index, 0] with LAST_ID FID, so it needs to replace
1714                  * the FID with orphan FID here */
1715                 ii->ii_fid.f_seq = FID_SEQ_LAYOUT_RBTREE;
1716                 ii->ii_fid.f_oid = osp->opd_index;
1717                 ii->ii_fid.f_ver = 0;
1718                 ii->ii_flags = II_FL_NOHASH;
1719                 ii->ii_attrs = osp_dev2node(osp);
1720         } else {
1721                 ii->ii_fid = *lu_object_fid(&it->ooi_obj->do_lu);
1722                 ii->ii_flags = II_FL_NOHASH | II_FL_NOKEY | II_FL_VARKEY |
1723                                II_FL_VARREC;
1724                 ii->ii_attrs = it->ooi_attr;
1725         }
1726         ii->ii_magic = IDX_INFO_MAGIC;
1727         ii->ii_count = npages * LU_PAGE_COUNT;
1728         ii->ii_hash_start = it->ooi_next;
1729
1730         ptlrpc_at_set_req_timeout(req);
1731
1732         desc = ptlrpc_prep_bulk_imp(req, npages, 1,
1733                                     PTLRPC_BULK_PUT_SINK | PTLRPC_BULK_BUF_KIOV,
1734                                     MDS_BULK_PORTAL,
1735                                     &ptlrpc_bulk_kiov_pin_ops);
1736         if (desc == NULL) {
1737                 ptlrpc_request_free(req);
1738                 RETURN(-ENOMEM);
1739         }
1740
1741         for (i = 0; i < npages; i++)
1742                 desc->bd_frag_ops->add_kiov_frag(desc, pages[i], 0,
1743                                                  PAGE_SIZE);
1744
1745         ptlrpc_request_set_replen(req);
1746         rc = ptlrpc_queue_wait(req);
1747         if (rc != 0)
1748                 GOTO(out, rc);
1749
1750         rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk,
1751                                           req->rq_bulk->bd_nob_transferred);
1752         if (rc < 0)
1753                 GOTO(out, rc);
1754         rc = 0;
1755
1756         ii = req_capsule_server_get(&req->rq_pill, &RMF_IDX_INFO);
1757         if (ii->ii_magic != IDX_INFO_MAGIC)
1758                  GOTO(out, rc = -EPROTO);
1759
1760         npages = (ii->ii_count + LU_PAGE_COUNT - 1) >>
1761                  (PAGE_SHIFT - LU_PAGE_SHIFT);
1762         if (npages > it->ooi_total_npages) {
1763                 CERROR("%s: returned more pages than expected, %u > %u\n",
1764                        osp->opd_obd->obd_name, npages, it->ooi_total_npages);
1765                 GOTO(out, rc = -EINVAL);
1766         }
1767
1768         it->ooi_valid_npages = npages;
1769         if (ptlrpc_rep_need_swab(req))
1770                 it->ooi_swab = 1;
1771
1772         it->ooi_next = ii->ii_hash_end;
1773
1774 out:
1775         ptlrpc_req_finished(req);
1776
1777         return rc;
1778 }
1779
1780 /**
1781  * Move the iteration cursor to the next lu_page.
1782  *
1783  * One system page (PAGE_SIZE) may contain multiple lu_page (4KB),
1784  * that depends on the LU_PAGE_COUNT. If it is not the last lu_page
1785  * in current system page, then move the iteration cursor to the next
1786  * lu_page in current system page. Otherwise, if there are more system
1787  * pages in the cache, then move the iteration cursor to the next system
1788  * page. If all the cached records (pages) have been iterated, then fetch
1789  * more records via osp_it_fetch().
1790  *
1791  * \param[in] env       pointer to the thread context
1792  * \param[in] di        pointer to the iteration structure
1793  *
1794  * \retval              positive for end of the directory
1795  * \retval              0 for success
1796  * \retval              negative error number on failure
1797  */
1798 int osp_it_next_page(const struct lu_env *env, struct dt_it *di)
1799 {
1800         struct osp_it           *it = (struct osp_it *)di;
1801         struct lu_idxpage       *idxpage;
1802         struct page             **pages;
1803         int                     rc;
1804         int                     i;
1805         ENTRY;
1806
1807 again2:
1808         idxpage = it->ooi_cur_idxpage;
1809         if (idxpage != NULL) {
1810                 if (idxpage->lip_nr == 0)
1811                         RETURN(1);
1812
1813                 if (it->ooi_pos_ent < idxpage->lip_nr) {
1814                         CDEBUG(D_INFO, "ooi_pos %d nr %d\n",
1815                                (int)it->ooi_pos_ent, (int)idxpage->lip_nr);
1816                         RETURN(0);
1817                 }
1818                 it->ooi_cur_idxpage = NULL;
1819                 it->ooi_pos_lu_page++;
1820
1821 again1:
1822                 if (it->ooi_pos_lu_page < LU_PAGE_COUNT) {
1823                         it->ooi_cur_idxpage = (void *)it->ooi_cur_page +
1824                                          LU_PAGE_SIZE * it->ooi_pos_lu_page;
1825                         if (it->ooi_swab)
1826                                 lustre_swab_lip_header(it->ooi_cur_idxpage);
1827                         if (it->ooi_cur_idxpage->lip_magic != LIP_MAGIC) {
1828                                 struct osp_device *osp =
1829                                         lu2osp_dev(it->ooi_obj->do_lu.lo_dev);
1830
1831                                 CERROR("%s: invalid magic (%x != %x) for page "
1832                                        "%d/%d while read layout orphan index\n",
1833                                        osp->opd_obd->obd_name,
1834                                        it->ooi_cur_idxpage->lip_magic,
1835                                        LIP_MAGIC, it->ooi_pos_page,
1836                                        it->ooi_pos_lu_page);
1837                                 /* Skip this lu_page next time. */
1838                                 it->ooi_pos_ent = idxpage->lip_nr - 1;
1839                                 RETURN(-EINVAL);
1840                         }
1841                         it->ooi_pos_ent = -1;
1842                         goto again2;
1843                 }
1844
1845                 kunmap(it->ooi_cur_page);
1846                 it->ooi_cur_page = NULL;
1847                 it->ooi_pos_page++;
1848
1849 again0:
1850                 pages = it->ooi_pages;
1851                 if (it->ooi_pos_page < it->ooi_valid_npages) {
1852                         it->ooi_cur_page = kmap(pages[it->ooi_pos_page]);
1853                         it->ooi_pos_lu_page = 0;
1854                         goto again1;
1855                 }
1856
1857                 for (i = 0; i < it->ooi_total_npages; i++) {
1858                         if (pages[i] != NULL)
1859                                 __free_page(pages[i]);
1860                 }
1861                 OBD_FREE(pages, it->ooi_total_npages * sizeof(*pages));
1862
1863                 it->ooi_pos_page = 0;
1864                 it->ooi_total_npages = 0;
1865                 it->ooi_valid_npages = 0;
1866                 it->ooi_swab = 0;
1867                 it->ooi_ent = NULL;
1868                 it->ooi_cur_page = NULL;
1869                 it->ooi_cur_idxpage = NULL;
1870                 it->ooi_pages = NULL;
1871         }
1872
1873         if (it->ooi_next == II_END_OFF)
1874                 RETURN(1);
1875
1876         rc = osp_it_fetch(env, it);
1877         if (rc == 0)
1878                 goto again0;
1879
1880         RETURN(rc);
1881 }
1882
1883 /**
1884  * Move the iteration cursor to the next record.
1885  *
1886  * If there are more records in the lu_page, then move the iteration
1887  * cursor to the next record directly. Otherwise, move the iteration
1888  * cursor to the record in the next lu_page via osp_it_next_page()
1889  *
1890  * \param[in] env       pointer to the thread context
1891  * \param[in] di        pointer to the iteration structure
1892  *
1893  * \retval              positive for end of the directory
1894  * \retval              0 for success
1895  * \retval              negative error number on failure
1896  */
1897 static int osp_orphan_it_next(const struct lu_env *env, struct dt_it *di)
1898 {
1899         struct osp_it           *it = (struct osp_it *)di;
1900         struct lu_idxpage       *idxpage;
1901         int                     rc;
1902         ENTRY;
1903
1904 again:
1905         idxpage = it->ooi_cur_idxpage;
1906         if (idxpage != NULL) {
1907                 if (idxpage->lip_nr == 0)
1908                         RETURN(1);
1909
1910                 it->ooi_pos_ent++;
1911                 if (it->ooi_pos_ent < idxpage->lip_nr) {
1912                         it->ooi_ent =
1913                                 (struct lu_orphan_ent *)idxpage->lip_entries +
1914                                                         it->ooi_pos_ent;
1915                         if (it->ooi_swab)
1916                                 lustre_swab_orphan_ent(it->ooi_ent);
1917                         RETURN(0);
1918                 }
1919         }
1920
1921         rc = osp_it_next_page(env, di);
1922         if (rc == 0)
1923                 goto again;
1924
1925         RETURN(rc);
1926 }
1927
1928 int osp_it_get(const struct lu_env *env, struct dt_it *di,
1929                const struct dt_key *key)
1930 {
1931         return 1;
1932 }
1933
1934 void osp_it_put(const struct lu_env *env, struct dt_it *di)
1935 {
1936 }
1937
1938 static struct dt_key *osp_orphan_it_key(const struct lu_env *env,
1939                                         const struct dt_it *di)
1940 {
1941         struct osp_it   *it  = (struct osp_it *)di;
1942         struct lu_orphan_ent    *ent = (struct lu_orphan_ent *)it->ooi_ent;
1943
1944         if (likely(ent != NULL))
1945                 return (struct dt_key *)(&ent->loe_key);
1946
1947         return NULL;
1948 }
1949
1950 static int osp_orphan_it_key_size(const struct lu_env *env,
1951                                   const struct dt_it *di)
1952 {
1953         return sizeof(struct lu_fid);
1954 }
1955
1956 static int osp_orphan_it_rec(const struct lu_env *env, const struct dt_it *di,
1957                              struct dt_rec *rec, __u32 attr)
1958 {
1959         struct osp_it   *it  = (struct osp_it *)di;
1960         struct lu_orphan_ent    *ent = (struct lu_orphan_ent *)it->ooi_ent;
1961
1962         if (likely(ent != NULL)) {
1963                 *(struct lu_orphan_rec *)rec = ent->loe_rec;
1964                 return 0;
1965         }
1966
1967         return -EINVAL;
1968 }
1969
1970 __u64 osp_it_store(const struct lu_env *env, const struct dt_it *di)
1971 {
1972         struct osp_it   *it = (struct osp_it *)di;
1973
1974         return it->ooi_next;
1975 }
1976
1977 /**
1978  * Locate the iteration cursor to the specified position (cookie).
1979  *
1980  * \param[in] env       pointer to the thread context
1981  * \param[in] di        pointer to the iteration structure
1982  * \param[in] hash      the specified position
1983  *
1984  * \retval              positive number for locating to the exactly position
1985  *                      or the next
1986  * \retval              0 for arriving at the end of the iteration
1987  * \retval              negative error number on failure
1988  */
1989 int osp_orphan_it_load(const struct lu_env *env, const struct dt_it *di,
1990                        __u64 hash)
1991 {
1992         struct osp_it   *it     = (struct osp_it *)di;
1993         int              rc;
1994
1995         it->ooi_next = hash;
1996         rc = osp_orphan_it_next(env, (struct dt_it *)di);
1997         if (rc == 1)
1998                 return 0;
1999
2000         if (rc == 0)
2001                 return 1;
2002
2003         return rc;
2004 }
2005
2006 int osp_it_key_rec(const struct lu_env *env, const struct dt_it *di,
2007                    void *key_rec)
2008 {
2009         return 0;
2010 }
2011
2012 static const struct dt_index_operations osp_orphan_index_ops = {
2013         .dio_lookup             = osp_orphan_index_lookup,
2014         .dio_declare_insert     = osp_orphan_index_declare_insert,
2015         .dio_insert             = osp_orphan_index_insert,
2016         .dio_declare_delete     = osp_orphan_index_declare_delete,
2017         .dio_delete             = osp_orphan_index_delete,
2018         .dio_it = {
2019                 .init           = osp_it_init,
2020                 .fini           = osp_it_fini,
2021                 .next           = osp_orphan_it_next,
2022                 .get            = osp_it_get,
2023                 .put            = osp_it_put,
2024                 .key            = osp_orphan_it_key,
2025                 .key_size       = osp_orphan_it_key_size,
2026                 .rec            = osp_orphan_it_rec,
2027                 .store          = osp_it_store,
2028                 .load           = osp_orphan_it_load,
2029                 .key_rec        = osp_it_key_rec,
2030         }
2031 };
2032
2033 /**
2034  * Implement OSP layer dt_object_operations::do_index_try() interface.
2035  *
2036  * Negotiate the index type.
2037  *
2038  * If the target index is an IDIF object, then use osp_orphan_index_ops.
2039  * Otherwise, assign osp_md_index_ops to the dt_object::do_index_ops.
2040  * (\see lustre/include/lustre_fid.h for IDIF.)
2041  *
2042  * \param[in] env       pointer to the thread context
2043  * \param[in] dt        pointer to the OSP layer dt_object
2044  * \param[in] feat      unused
2045  *
2046  * \retval              0 for success
2047  */
2048 static int osp_index_try(const struct lu_env *env,
2049                          struct dt_object *dt,
2050                          const struct dt_index_features *feat)
2051 {
2052         const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
2053
2054         if (fid_is_last_id(fid) && fid_is_idif(fid))
2055                 dt->do_index_ops = &osp_orphan_index_ops;
2056         else
2057                 dt->do_index_ops = &osp_md_index_ops;
2058         return 0;
2059 }
2060
2061 static struct dt_object_operations osp_obj_ops = {
2062         .do_declare_attr_get    = osp_declare_attr_get,
2063         .do_attr_get            = osp_attr_get,
2064         .do_declare_attr_set    = osp_declare_attr_set,
2065         .do_attr_set            = osp_attr_set,
2066         .do_declare_xattr_get   = osp_declare_xattr_get,
2067         .do_xattr_get           = osp_xattr_get,
2068         .do_declare_xattr_set   = osp_declare_xattr_set,
2069         .do_xattr_set           = osp_xattr_set,
2070         .do_declare_create      = osp_declare_object_create,
2071         .do_create              = osp_object_create,
2072         .do_declare_destroy     = osp_declare_object_destroy,
2073         .do_destroy             = osp_object_destroy,
2074         .do_index_try           = osp_index_try,
2075 };
2076
2077 /**
2078  * Implement OSP layer lu_object_operations::loo_object_init() interface.
2079  *
2080  * Initialize the object.
2081  *
2082  * If it is a remote MDT object, then call do_attr_get() to fetch
2083  * the attribute from the peer.
2084  *
2085  * \param[in] env       pointer to the thread context
2086  * \param[in] o         pointer to the OSP layer lu_object
2087  * \param[in] conf      unused
2088  *
2089  * \retval              0 for success
2090  * \retval              negative error number on failure
2091  */
2092 static int osp_object_init(const struct lu_env *env, struct lu_object *o,
2093                            const struct lu_object_conf *conf)
2094 {
2095         struct osp_object       *po = lu2osp_obj(o);
2096         int                     rc = 0;
2097         ENTRY;
2098
2099         spin_lock_init(&po->opo_lock);
2100         o->lo_header->loh_attr |= LOHA_REMOTE;
2101         INIT_LIST_HEAD(&po->opo_xattr_list);
2102         INIT_LIST_HEAD(&po->opo_invalidate_cb_list);
2103
2104         if (is_ost_obj(o)) {
2105                 po->opo_obj.do_ops = &osp_obj_ops;
2106         } else {
2107                 struct lu_attr *la = &osp_env_info(env)->osi_attr;
2108
2109                 po->opo_obj.do_ops = &osp_md_obj_ops;
2110                 po->opo_obj.do_body_ops = &osp_md_body_ops;
2111
2112                 if (conf != NULL && conf->loc_flags & LOC_F_NEW) {
2113                         po->opo_non_exist = 1;
2114                 } else {
2115                         rc = po->opo_obj.do_ops->do_attr_get(env, lu2dt_obj(o),
2116                                                              la);
2117                         if (rc == 0)
2118                                 o->lo_header->loh_attr |=
2119                                         LOHA_EXISTS | (la->la_mode & S_IFMT);
2120                         if (rc == -ENOENT) {
2121                                 po->opo_non_exist = 1;
2122                                 rc = 0;
2123                         }
2124                 }
2125                 init_rwsem(&po->opo_sem);
2126         }
2127         RETURN(rc);
2128 }
2129
2130 /**
2131  * Implement OSP layer lu_object_operations::loo_object_free() interface.
2132  *
2133  * Finalize the object.
2134  *
2135  * If the OSP object has attributes cache, then destroy the cache.
2136  * Free the object finally.
2137  *
2138  * \param[in] env       pointer to the thread context
2139  * \param[in] o         pointer to the OSP layer lu_object
2140  */
2141 static void osp_object_free(const struct lu_env *env, struct lu_object *o)
2142 {
2143         struct osp_object       *obj = lu2osp_obj(o);
2144         struct lu_object_header *h = o->lo_header;
2145         struct osp_xattr_entry *oxe;
2146         struct osp_xattr_entry *tmp;
2147         int                     count;
2148
2149         dt_object_fini(&obj->opo_obj);
2150         lu_object_header_fini(h);
2151         list_for_each_entry_safe(oxe, tmp, &obj->opo_xattr_list, oxe_list) {
2152                 list_del(&oxe->oxe_list);
2153                 count = atomic_read(&oxe->oxe_ref);
2154                 LASSERTF(count == 1,
2155                          "Still has %d users on the xattr entry %.*s\n",
2156                          count-1, (int)oxe->oxe_namelen, oxe->oxe_buf);
2157
2158                 OBD_FREE(oxe, oxe->oxe_buflen);
2159         }
2160         OBD_SLAB_FREE_PTR(obj, osp_object_kmem);
2161 }
2162
2163 /**
2164  * Implement OSP layer lu_object_operations::loo_object_release() interface.
2165  *
2166  * Cleanup (not free) the object.
2167  *
2168  * If it is a reserved object but failed to be created, or it is an OST
2169  * object, then mark the object as non-cached.
2170  *
2171  * \param[in] env       pointer to the thread context
2172  * \param[in] o         pointer to the OSP layer lu_object
2173  */
2174 static void osp_object_release(const struct lu_env *env, struct lu_object *o)
2175 {
2176         struct osp_object       *po = lu2osp_obj(o);
2177         struct osp_device       *d  = lu2osp_dev(o->lo_dev);
2178
2179         ENTRY;
2180
2181         /*
2182          * release reservation if object was declared but not created
2183          * this may require lu_object_put() in LOD
2184          */
2185         if (unlikely(po->opo_reserved)) {
2186                 LASSERT(d->opd_pre != NULL);
2187                 LASSERT(d->opd_pre_reserved > 0);
2188                 spin_lock(&d->opd_pre_lock);
2189                 d->opd_pre_reserved--;
2190                 spin_unlock(&d->opd_pre_lock);
2191
2192                 /* not needed in cache any more */
2193                 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
2194         }
2195
2196         if (is_ost_obj(o))
2197                 /* XXX: Currently, NOT cache OST-object on MDT because:
2198                  *      1. it is not often accessed on MDT.
2199                  *      2. avoid up layer (such as LFSCK) to load too many
2200                  *         once-used OST-objects. */
2201                 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
2202
2203         EXIT;
2204 }
2205
2206 static int osp_object_print(const struct lu_env *env, void *cookie,
2207                             lu_printer_t p, const struct lu_object *l)
2208 {
2209         const struct osp_object *o = lu2osp_obj((struct lu_object *)l);
2210
2211         return (*p)(env, cookie, LUSTRE_OSP_NAME"-object@%p", o);
2212 }
2213
2214 static int osp_object_invariant(const struct lu_object *o)
2215 {
2216         LBUG();
2217 }
2218
2219 struct lu_object_operations osp_lu_obj_ops = {
2220         .loo_object_init        = osp_object_init,
2221         .loo_object_free        = osp_object_free,
2222         .loo_object_release     = osp_object_release,
2223         .loo_object_print       = osp_object_print,
2224         .loo_object_invariant   = osp_object_invariant
2225 };