Whamcloud - gitweb
LU-5099 api: transfer object type via dt_insert API
[fs/lustre-release.git] / lustre / mdd / mdd_orphans.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
19  *
20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21  * CA 95054 USA or visit www.sun.com if you need additional information or
22  * have any questions.
23  *
24  * GPL HEADER END
25  */
26 /*
27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2011, 2013, Intel Corporation.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * lustre/mdd/mdd_orphans.c
37  *
38  * Orphan handling code
39  *
40  * Author: Mike Pershin <tappro@clusterfs.com>
41  *         Pravin B Shelar <pravin.shelar@sun.com>
42  */
43
44 #define DEBUG_SUBSYSTEM S_MDS
45
46 #include <obd.h>
47 #include <obd_class.h>
48 #include <lustre_ver.h>
49 #include <obd_support.h>
50 #include <lustre_fid.h>
51 #include "mdd_internal.h"
52
53 const char orph_index_name[] = "PENDING";
54 static const char dotdot[] = "..";
55
56 enum {
57         ORPH_OP_UNLINK,
58         ORPH_OP_TRUNCATE
59 };
60
61 #define ORPHAN_FILE_NAME_FORMAT         "%016llx:%08x:%08x:%2x"
62 #define ORPHAN_FILE_NAME_FORMAT_18      "%llx:%08x"
63
64 static struct dt_key* orph_key_fill(const struct lu_env *env,
65                                     const struct lu_fid *lf, __u32 op)
66 {
67         char *key = mdd_env_info(env)->mti_key;
68         int rc;
69
70         LASSERT(key);
71         rc = snprintf(key, NAME_MAX + 1, ORPHAN_FILE_NAME_FORMAT,
72                       (long long unsigned int)fid_seq(lf),
73                       fid_oid(lf), fid_ver(lf), op);
74         if (rc > 0)
75                 return (struct dt_key*) key;
76         else
77                 return ERR_PTR(rc);
78 }
79
80 static struct dt_key* orph_key_fill_18(const struct lu_env *env,
81                                        const struct lu_fid *lf)
82 {
83         char *key = mdd_env_info(env)->mti_key;
84         int rc;
85
86         LASSERT(key);
87         rc = snprintf(key, NAME_MAX + 1, ORPHAN_FILE_NAME_FORMAT_18,
88                       (unsigned long long)fid_seq(lf), fid_oid(lf));
89         if (rc > 0)
90                 return (struct dt_key*) key;
91         else
92                 return ERR_PTR(rc);
93 }
94
95 static inline void mdd_orphan_write_lock(const struct lu_env *env,
96                                     struct mdd_device *mdd)
97 {
98
99         struct dt_object        *dor    = mdd->mdd_orphans;
100         dor->do_ops->do_write_lock(env, dor, MOR_TGT_ORPHAN);
101 }
102
103 static inline void mdd_orphan_write_unlock(const struct lu_env *env,
104                                            struct mdd_device *mdd)
105 {
106
107         struct dt_object        *dor    = mdd->mdd_orphans;
108         dor->do_ops->do_write_unlock(env, dor);
109 }
110
111 static inline int mdd_orphan_insert_obj(const struct lu_env *env,
112                                         struct mdd_device *mdd,
113                                         struct mdd_object *obj,
114                                         __u32 op, struct thandle *th)
115 {
116         struct dt_insert_rec    *rec    = &mdd_env_info(env)->mti_dt_rec;
117         struct dt_object        *dor    = mdd->mdd_orphans;
118         const struct lu_fid     *lf     = mdo2fid(obj);
119         struct dt_key           *key    = orph_key_fill(env, lf, op);
120
121         rec->rec_fid = lf;
122         rec->rec_type = mdd_object_type(obj);
123
124         return  dor->do_index_ops->dio_insert(env, dor,
125                                               (const struct dt_rec *)rec,
126                                               key, th, BYPASS_CAPA, 1);
127 }
128
129 static inline int mdd_orphan_delete_obj(const struct lu_env *env,
130                                         struct mdd_device  *mdd ,
131                                         struct dt_key *key,
132                                         struct thandle *th)
133 {
134         struct dt_object        *dor    = mdd->mdd_orphans;
135
136         return  dor->do_index_ops->dio_delete(env, dor,
137                                               key, th,
138                                               BYPASS_CAPA);
139 }
140
141 static inline void mdd_orphan_ref_add(const struct lu_env *env,
142                                  struct mdd_device *mdd,
143                                  struct thandle *th)
144 {
145         struct dt_object        *dor    = mdd->mdd_orphans;
146         dor->do_ops->do_ref_add(env, dor, th);
147 }
148
149 static inline void mdd_orphan_ref_del(const struct lu_env *env,
150                                  struct mdd_device *mdd,
151                                  struct thandle *th)
152 {
153         struct dt_object        *dor    = mdd->mdd_orphans;
154         dor->do_ops->do_ref_del(env, dor, th);
155 }
156
157
158 int orph_declare_index_insert(const struct lu_env *env,
159                               struct mdd_object *obj,
160                               umode_t mode, struct thandle *th)
161 {
162         struct dt_insert_rec    *rec = &mdd_env_info(env)->mti_dt_rec;
163         struct mdd_device       *mdd = mdo2mdd(&obj->mod_obj);
164         struct dt_key           *key;
165         int                     rc;
166
167         key = orph_key_fill(env, mdo2fid(obj), ORPH_OP_UNLINK);
168
169         rec->rec_fid = mdo2fid(obj);
170         rec->rec_type = mode;
171         rc = dt_declare_insert(env, mdd->mdd_orphans,
172                                (const struct dt_rec *)rec, key, th);
173         if (rc != 0)
174                 return rc;
175
176         rc = mdo_declare_ref_add(env, obj, th);
177         if (rc)
178                 return rc;
179
180         if (!S_ISDIR(mode))
181                 return 0;
182
183         rc = mdo_declare_ref_add(env, obj, th);
184         if (rc)
185                 return rc;
186
187         rc = dt_declare_ref_add(env, mdd->mdd_orphans, th);
188         if (rc)
189                 return rc;
190
191         rc = mdo_declare_index_delete(env, obj, dotdot, th);
192         if (rc)
193                 return rc;
194
195         rc = mdo_declare_index_insert(env, obj,
196                                       lu_object_fid(&mdd->mdd_orphans->do_lu),
197                                       S_IFDIR, dotdot, th);
198
199         return rc;
200 }
201
202 static int orph_index_insert(const struct lu_env *env,
203                              struct mdd_object *obj,
204                              __u32 op, struct thandle *th)
205 {
206         struct mdd_device       *mdd    = mdo2mdd(&obj->mod_obj);
207         struct dt_object        *dor    = mdd->mdd_orphans;
208         const struct lu_fid     *lf_dor = lu_object_fid(&dor->do_lu);
209         struct dt_object        *next   = mdd_object_child(obj);
210         struct dt_insert_rec    *rec    = &mdd_env_info(env)->mti_dt_rec;
211         int                      rc;
212         ENTRY;
213
214         LASSERT(mdd_write_locked(env, obj) != 0);
215         LASSERT(!(obj->mod_flags & ORPHAN_OBJ));
216
217         mdd_orphan_write_lock(env, mdd);
218
219         rc = mdd_orphan_insert_obj(env, mdd, obj, op, th);
220         if (rc)
221                 GOTO(out, rc);
222
223         mdo_ref_add(env, obj, th);
224         if (!S_ISDIR(mdd_object_type(obj)))
225                 GOTO(out, rc = 0);
226
227         mdo_ref_add(env, obj, th);
228         mdd_orphan_ref_add(env, mdd, th);
229
230         /* try best to fixup directory, dont return errors
231          * from here */
232         if (!dt_try_as_dir(env, next))
233                 GOTO(out, rc = 0);
234         next->do_index_ops->dio_delete(env, next,
235                                        (const struct dt_key *)dotdot,
236                                        th, BYPASS_CAPA);
237
238         rec->rec_fid = lf_dor;
239         rec->rec_type = S_IFDIR;
240         next->do_index_ops->dio_insert(env, next, (const struct dt_rec *)rec,
241                                        (const struct dt_key *)dotdot,
242                                        th, BYPASS_CAPA, 1);
243
244 out:
245         if (rc == 0)
246                 obj->mod_flags |= ORPHAN_OBJ;
247
248         mdd_orphan_write_unlock(env, mdd);
249
250         RETURN(rc);
251 }
252
253 int orph_declare_index_delete(const struct lu_env *env,
254                               struct mdd_object *obj,
255
256                               struct thandle *th)
257 {
258         struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
259         struct dt_key     *key;
260         int                rc;
261
262         key = orph_key_fill(env, mdo2fid(obj), ORPH_OP_UNLINK);
263
264         rc = dt_declare_delete(env, mdd->mdd_orphans, key, th);
265         if (rc)
266                 return rc;
267
268         rc = mdo_declare_ref_del(env, obj, th);
269         if (rc)
270                 return rc;
271
272         if (S_ISDIR(mdd_object_type(obj))) {
273                 rc = mdo_declare_ref_del(env, obj, th);
274                 if (rc)
275                         return rc;
276
277                 rc = dt_declare_ref_del(env, mdd->mdd_orphans, th);
278         }
279
280         return rc;
281 }
282
283 static int orph_index_delete(const struct lu_env *env,
284                              struct mdd_object *obj,
285                              __u32 op,
286                              struct thandle *th)
287 {
288         struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
289         struct dt_object *dor = mdd->mdd_orphans;
290         struct dt_key *key;
291         int rc;
292
293         ENTRY;
294
295         LASSERT(mdd_write_locked(env, obj) != 0);
296         LASSERT(obj->mod_flags & ORPHAN_OBJ);
297         LASSERT(obj->mod_count == 0);
298
299         LASSERT(dor);
300
301         key = orph_key_fill(env, mdo2fid(obj), op);
302         mdd_orphan_write_lock(env, mdd);
303
304         rc = mdd_orphan_delete_obj(env, mdd, key, th);
305
306         if (rc == -ENOENT) {
307                 key = orph_key_fill_18(env, mdo2fid(obj));
308                 rc = mdd_orphan_delete_obj(env, mdd, key, th);
309         }
310
311         if (!rc) {
312                 /* lov objects will be destroyed by caller */
313                 mdo_ref_del(env, obj, th);
314                 if (S_ISDIR(mdd_object_type(obj))) {
315                         mdo_ref_del(env, obj, th);
316                         mdd_orphan_ref_del(env, mdd, th);
317                 }
318                 obj->mod_flags &= ~ORPHAN_OBJ;
319         } else {
320                 CERROR("could not delete object: rc = %d\n",rc);
321         }
322
323         mdd_orphan_write_unlock(env, mdd);
324         RETURN(rc);
325 }
326
327
328 static int orphan_object_destroy(const struct lu_env *env,
329                                  struct mdd_object *obj,
330                                  struct dt_key *key)
331 {
332         struct thandle *th = NULL;
333         struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
334         int rc = 0;
335         ENTRY;
336
337         th = mdd_trans_create(env, mdd);
338         if (IS_ERR(th)) {
339                 CERROR("Cannot get thandle\n");
340                 RETURN(PTR_ERR(th));
341         }
342
343         rc = orph_declare_index_delete(env, obj, th);
344         if (rc)
345                 GOTO(stop, rc);
346
347         rc = mdo_declare_destroy(env, obj, th);
348         if (rc)
349                 GOTO(stop, rc);
350
351         rc = mdd_trans_start(env, mdd, th);
352         if (rc)
353                 GOTO(stop, rc);
354
355         mdd_write_lock(env, obj, MOR_TGT_CHILD);
356         if (likely(obj->mod_count == 0)) {
357                 mdd_orphan_write_lock(env, mdd);
358                 rc = mdd_orphan_delete_obj(env, mdd, key, th);
359                 if (rc == 0) {
360                         mdo_ref_del(env, obj, th);
361                         if (S_ISDIR(mdd_object_type(obj))) {
362                                 mdo_ref_del(env, obj, th);
363                                 mdd_orphan_ref_del(env, mdd, th);
364                         }
365                         rc = mdo_destroy(env, obj, th);
366                 } else
367                         CERROR("could not delete object: rc = %d\n",rc);
368                 mdd_orphan_write_unlock(env, mdd);
369         }
370         mdd_write_unlock(env, obj);
371
372 stop:
373         mdd_trans_stop(env, mdd, 0, th);
374
375         RETURN(rc);
376 }
377
378 /**
379  * Delete unused orphan with FID \a lf from PENDING directory
380  *
381  * \param mdd  MDD device finishing recovery
382  * \param lf   FID of file or directory to delete
383  * \param key  cookie for this entry in index iterator
384  *
385  * \retval 0   success
386  * \retval -ve error
387  */
388 static int orph_key_test_and_del(const struct lu_env *env,
389                                  struct mdd_device *mdd,
390                                  struct lu_fid *lf,
391                                  struct dt_key *key)
392 {
393         struct mdd_object *mdo;
394         int rc;
395
396         mdo = mdd_object_find(env, mdd, lf);
397
398         if (IS_ERR(mdo))
399                 return PTR_ERR(mdo);
400
401         rc = -EBUSY;
402         if (mdo->mod_count == 0) {
403                 CDEBUG(D_HA, "Found orphan "DFID", delete it\n", PFID(lf));
404                 rc = orphan_object_destroy(env, mdo, key);
405                 if (rc) /* so replay-single.sh test_37 works */
406                         CERROR("%s: error unlinking orphan "DFID" from "
407                                "PENDING: rc = %d\n",
408                                mdd2obd_dev(mdd)->obd_name, PFID(lf), rc);
409         } else {
410                 mdd_write_lock(env, mdo, MOR_TGT_CHILD);
411                 if (likely(mdo->mod_count > 0)) {
412                         CDEBUG(D_HA, "Found orphan "DFID" count %d, skip it\n",
413                                PFID(lf), mdo->mod_count);
414                         mdo->mod_flags |= ORPHAN_OBJ;
415                 }
416                 mdd_write_unlock(env, mdo);
417         }
418
419         mdd_object_put(env, mdo);
420         return rc;
421 }
422
423 /**
424  * delete unreferenced files and directories in the PENDING directory
425  *
426  * Files that remain in PENDING after client->MDS recovery has completed
427  * have to be referenced (opened) by some client during recovery, or they
428  * will be deleted here (for clients that did not complete recovery).
429  *
430  * \param thread  info about orphan cleanup thread
431  *
432  * \retval 0   success
433  * \retval -ve error
434  */
435 static int orph_index_iterate(const struct lu_env *env,
436                               struct mdd_generic_thread *thread)
437 {
438         struct mdd_device *mdd = (struct mdd_device *)thread->mgt_data;
439         struct dt_object *dor = mdd->mdd_orphans;
440         struct lu_dirent *ent = &mdd_env_info(env)->mti_ent;
441         const struct dt_it_ops *iops;
442         struct dt_it     *it;
443         struct lu_fid     fid;
444         int               key_sz = 0;
445         int               rc;
446         __u64             cookie;
447         ENTRY;
448
449         iops = &dor->do_index_ops->dio_it;
450         it = iops->init(env, dor, LUDA_64BITHASH, BYPASS_CAPA);
451         if (IS_ERR(it)) {
452                 rc = PTR_ERR(it);
453                 CERROR("%s: cannot clean PENDING: rc = %d\n",
454                        mdd2obd_dev(mdd)->obd_name, rc);
455                 GOTO(out, rc);
456         }
457
458         rc = iops->load(env, it, 0);
459         if (rc < 0)
460                 GOTO(out_put, rc);
461         if (rc == 0) {
462                 CERROR("%s: error loading iterator to clean PENDING\n",
463                        mdd2obd_dev(mdd)->obd_name);
464                 /* Index contains no zero key? */
465                 GOTO(out_put, rc = -EIO);
466         }
467
468         do {
469                 if (thread->mgt_abort)
470                         break;
471
472                 key_sz = iops->key_size(env, it);
473                 /* filter out "." and ".." entries from PENDING dir. */
474                 if (key_sz < 8)
475                         goto next;
476
477                 rc = iops->rec(env, it, (struct dt_rec *)ent, LUDA_64BITHASH);
478                 if (rc != 0) {
479                         CERROR("%s: fail to get FID for orphan it: rc = %d\n",
480                                mdd2obd_dev(mdd)->obd_name, rc);
481                         goto next;
482                 }
483
484                 fid_le_to_cpu(&fid, &ent->lde_fid);
485                 if (!fid_is_sane(&fid)) {
486                         CERROR("%s: bad FID "DFID" cleaning PENDING\n",
487                                mdd2obd_dev(mdd)->obd_name, PFID(&fid));
488                         goto next;
489                 }
490
491                 /* kill orphan object */
492                 cookie = iops->store(env, it);
493                 iops->put(env, it);
494                 rc = orph_key_test_and_del(env, mdd, &fid,
495                                            (struct dt_key *)ent->lde_name);
496
497                 /* after index delete reset iterator */
498                 if (rc == 0)
499                         rc = iops->get(env, it, (const void *)"");
500                 else
501                         rc = iops->load(env, it, cookie);
502 next:
503                 rc = iops->next(env, it);
504         } while (rc == 0);
505
506         GOTO(out_put, rc = 0);
507 out_put:
508         iops->put(env, it);
509         iops->fini(env, it);
510
511 out:
512         return rc;
513 }
514
515 /**
516  * open the PENDING directory for device \a mdd
517  *
518  * The PENDING directory persistently tracks files and directories that were
519  * unlinked from the namespace (nlink == 0) but are still held open by clients.
520  * Those inodes shouldn't be deleted if the MDS crashes, because the clients
521  * would not be able to recover and reopen those files.  Instead, these inodes
522  * are linked into the PENDING directory on disk, and only deleted if all
523  * clients close them, or the MDS finishes client recovery without any client
524  * reopening them (i.e. former clients didn't join recovery).
525  *  \param d   mdd device being started.
526  *
527  *  \retval 0  success
528  *  \retval  -ve index operation error.
529  *
530  */
531 int orph_index_init(const struct lu_env *env, struct mdd_device *mdd)
532 {
533         struct lu_fid            fid;
534         struct dt_object        *d;
535         int                      rc = 0;
536
537         ENTRY;
538
539         /* create PENDING dir */
540         fid_zero(&fid);
541         rc = mdd_local_file_create(env, mdd, &mdd->mdd_local_root_fid,
542                                    orph_index_name, S_IFDIR | S_IRUGO |
543                                    S_IWUSR | S_IXUGO, &fid);
544         if (rc < 0)
545                 RETURN(rc);
546
547         d = dt_locate(env, mdd->mdd_child, &fid);
548         if (IS_ERR(d))
549                 RETURN(PTR_ERR(d));
550         LASSERT(lu_object_exists(&d->do_lu));
551         if (!dt_try_as_dir(env, d)) {
552                 CERROR("%s: \"%s\" is not an index: rc = %d\n",
553                        mdd2obd_dev(mdd)->obd_name, orph_index_name, rc);
554                 lu_object_put(env, &d->do_lu);
555                 RETURN(-ENOTDIR);
556         }
557         mdd->mdd_orphans = d;
558         RETURN(0);
559 }
560
561 void orph_index_fini(const struct lu_env *env, struct mdd_device *mdd)
562 {
563         ENTRY;
564         if (mdd->mdd_orphans != NULL) {
565                 lu_object_put(env, &mdd->mdd_orphans->do_lu);
566                 mdd->mdd_orphans = NULL;
567         }
568         EXIT;
569 }
570
571 static int __mdd_orphan_cleanup(void *args)
572 {
573         struct mdd_generic_thread *thread = (struct mdd_generic_thread *)args;
574         struct lu_env             *env = NULL;
575         int                        rc;
576         ENTRY;
577
578         complete(&thread->mgt_started);
579
580         OBD_ALLOC_PTR(env);
581         if (env == NULL)
582                 GOTO(out, rc = -ENOMEM);
583
584         rc = lu_env_init(env, LCT_MD_THREAD);
585         if (rc)
586                 GOTO(out, rc);
587
588         rc = orph_index_iterate(env, thread);
589
590         lu_env_fini(env);
591         GOTO(out, rc);
592 out:
593         if (env)
594                 OBD_FREE_PTR(env);
595         complete(&thread->mgt_finished);
596         return rc;
597 }
598
599 /**
600  *  Iterate orphan index to cleanup orphan objects after recovery is done.
601  *  \param d   mdd device in recovery.
602  */
603 int mdd_orphan_cleanup(const struct lu_env *env, struct mdd_device *d)
604 {
605         int rc = -ENOMEM;
606         char *name = NULL;
607
608         OBD_ALLOC(name, MTI_NAME_MAXLEN);
609         if (name == NULL)
610                 goto out;
611
612         snprintf(name, MTI_NAME_MAXLEN, "orph_cleanup_%s",
613                  mdd2obd_dev(d)->obd_name);
614
615         rc = mdd_generic_thread_start(&d->mdd_orph_cleanup_thread,
616                                       __mdd_orphan_cleanup, (void *)d, name);
617 out:
618         if (rc)
619                 CERROR("%s: start orphan cleanup thread failed:%d\n",
620                        mdd2obd_dev(d)->obd_name, rc);
621         if (name)
622                 OBD_FREE(name, MTI_NAME_MAXLEN);
623         return rc;
624 }
625
626 /**
627  *  add an orphan \a obj to the orphan index.
628  *  \param obj file or directory.
629  *  \param th  transaction for index insert.
630  *
631  *  \pre obj nlink == 0 && obj->mod_count != 0
632  *
633  *  \retval 0  success
634  *  \retval  -ve index operation error.
635  */
636 int __mdd_orphan_add(const struct lu_env *env,
637                      struct mdd_object *obj, struct thandle *th)
638 {
639         return orph_index_insert(env, obj, ORPH_OP_UNLINK, th);
640 }
641
642 /**
643  *  delete an orphan \a obj from orphan index.
644  *  \param obj file or directory.
645  *  \param th  transaction for index deletion and object destruction.
646  *
647  *  \pre obj->mod_count == 0 && ORPHAN_OBJ is set for obj.
648  *
649  *  \retval 0  success
650  *  \retval  -ve index operation error.
651  */
652 int __mdd_orphan_del(const struct lu_env *env,
653                      struct mdd_object *obj, struct thandle *th)
654 {
655         return orph_index_delete(env, obj, ORPH_OP_UNLINK, th);
656 }