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