Whamcloud - gitweb
LU-14742 socklnd: detect link state to set fatal error on ni
[fs/lustre-release.git] / lustre / obdclass / scrub.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) 2017, Intel Corporation.
24  */
25 /*
26  * lustre/obdclass/scrub.c
27  *
28  * The OI scrub is used for checking and (re)building Object Index files
29  * that are usually backend special. Here are some general scrub related
30  * functions that can be shared by different backends for OI scrub.
31  *
32  * Author: Fan Yong <fan.yong@intel.com>
33  */
34
35 #define DEBUG_SUBSYSTEM S_LFSCK
36
37 #include <linux/kthread.h>
38 #include <lustre_scrub.h>
39 #include <lustre_lib.h>
40 #include <lustre_fid.h>
41
42 static inline struct dt_device *scrub_obj2dev(struct dt_object *obj)
43 {
44         return container_of_safe(obj->do_lu.lo_dev, struct dt_device,
45                                  dd_lu_dev);
46 }
47
48 static void scrub_file_to_cpu(struct scrub_file *des, struct scrub_file *src)
49 {
50         uuid_copy(&des->sf_uuid, &src->sf_uuid);
51         des->sf_flags   = le64_to_cpu(src->sf_flags);
52         des->sf_magic   = le32_to_cpu(src->sf_magic);
53         des->sf_status  = le16_to_cpu(src->sf_status);
54         des->sf_param   = le16_to_cpu(src->sf_param);
55         des->sf_time_last_complete      =
56                                 le64_to_cpu(src->sf_time_last_complete);
57         des->sf_time_latest_start       =
58                                 le64_to_cpu(src->sf_time_latest_start);
59         des->sf_time_last_checkpoint    =
60                                 le64_to_cpu(src->sf_time_last_checkpoint);
61         des->sf_pos_latest_start        =
62                                 le64_to_cpu(src->sf_pos_latest_start);
63         des->sf_pos_last_checkpoint     =
64                                 le64_to_cpu(src->sf_pos_last_checkpoint);
65         des->sf_pos_first_inconsistent  =
66                                 le64_to_cpu(src->sf_pos_first_inconsistent);
67         des->sf_items_checked           =
68                                 le64_to_cpu(src->sf_items_checked);
69         des->sf_items_updated           =
70                                 le64_to_cpu(src->sf_items_updated);
71         des->sf_items_failed            =
72                                 le64_to_cpu(src->sf_items_failed);
73         des->sf_items_updated_prior     =
74                                 le64_to_cpu(src->sf_items_updated_prior);
75         des->sf_run_time        = le32_to_cpu(src->sf_run_time);
76         des->sf_success_count   = le32_to_cpu(src->sf_success_count);
77         des->sf_oi_count        = le16_to_cpu(src->sf_oi_count);
78         des->sf_internal_flags  = le16_to_cpu(src->sf_internal_flags);
79         memcpy(des->sf_oi_bitmap, src->sf_oi_bitmap, SCRUB_OI_BITMAP_SIZE);
80 }
81
82 static void scrub_file_to_le(struct scrub_file *des, struct scrub_file *src)
83 {
84         uuid_copy(&des->sf_uuid, &src->sf_uuid);
85         des->sf_flags   = cpu_to_le64(src->sf_flags);
86         des->sf_magic   = cpu_to_le32(src->sf_magic);
87         des->sf_status  = cpu_to_le16(src->sf_status);
88         des->sf_param   = cpu_to_le16(src->sf_param);
89         des->sf_time_last_complete      =
90                                 cpu_to_le64(src->sf_time_last_complete);
91         des->sf_time_latest_start       =
92                                 cpu_to_le64(src->sf_time_latest_start);
93         des->sf_time_last_checkpoint    =
94                                 cpu_to_le64(src->sf_time_last_checkpoint);
95         des->sf_pos_latest_start        =
96                                 cpu_to_le64(src->sf_pos_latest_start);
97         des->sf_pos_last_checkpoint     =
98                                 cpu_to_le64(src->sf_pos_last_checkpoint);
99         des->sf_pos_first_inconsistent  =
100                                 cpu_to_le64(src->sf_pos_first_inconsistent);
101         des->sf_items_checked           =
102                                 cpu_to_le64(src->sf_items_checked);
103         des->sf_items_updated           =
104                                 cpu_to_le64(src->sf_items_updated);
105         des->sf_items_failed            =
106                                 cpu_to_le64(src->sf_items_failed);
107         des->sf_items_updated_prior     =
108                                 cpu_to_le64(src->sf_items_updated_prior);
109         des->sf_run_time        = cpu_to_le32(src->sf_run_time);
110         des->sf_success_count   = cpu_to_le32(src->sf_success_count);
111         des->sf_oi_count        = cpu_to_le16(src->sf_oi_count);
112         des->sf_internal_flags  = cpu_to_le16(src->sf_internal_flags);
113         memcpy(des->sf_oi_bitmap, src->sf_oi_bitmap, SCRUB_OI_BITMAP_SIZE);
114 }
115
116 void scrub_file_init(struct lustre_scrub *scrub, uuid_t uuid)
117 {
118         struct scrub_file *sf = &scrub->os_file;
119
120         memset(sf, 0, sizeof(*sf));
121         uuid_copy(&sf->sf_uuid, &uuid);
122         sf->sf_magic = SCRUB_MAGIC_V1;
123         sf->sf_status = SS_INIT;
124 }
125 EXPORT_SYMBOL(scrub_file_init);
126
127 void scrub_file_reset(struct lustre_scrub *scrub, uuid_t uuid, u64 flags)
128 {
129         struct scrub_file *sf = &scrub->os_file;
130
131         CDEBUG(D_LFSCK, "%s: reset OI scrub file, old flags = "
132                "%#llx, add flags = %#llx\n",
133                scrub->os_name, sf->sf_flags, flags);
134
135         uuid_copy(&sf->sf_uuid, &uuid);
136         sf->sf_status = SS_INIT;
137         sf->sf_flags |= flags;
138         sf->sf_flags &= ~SF_AUTO;
139         sf->sf_run_time = 0;
140         sf->sf_time_latest_start = 0;
141         sf->sf_time_last_checkpoint = 0;
142         sf->sf_pos_latest_start = 0;
143         sf->sf_pos_last_checkpoint = 0;
144         sf->sf_pos_first_inconsistent = 0;
145         sf->sf_items_checked = 0;
146         sf->sf_items_updated = 0;
147         sf->sf_items_failed = 0;
148         sf->sf_items_noscrub = 0;
149         sf->sf_items_igif = 0;
150         if (!scrub->os_in_join)
151                 sf->sf_items_updated_prior = 0;
152 }
153 EXPORT_SYMBOL(scrub_file_reset);
154
155 int scrub_file_load(const struct lu_env *env, struct lustre_scrub *scrub)
156 {
157         struct scrub_file *sf = &scrub->os_file;
158         struct lu_buf buf = {
159                 .lb_buf = &scrub->os_file_disk,
160                 .lb_len = sizeof(scrub->os_file_disk)
161         };
162         loff_t pos = 0;
163         int rc;
164
165         rc = dt_read(env, scrub->os_obj, &buf, &pos);
166         /* failure */
167         if (rc < 0) {
168                 CERROR("%s: fail to load scrub file: rc = %d\n",
169                        scrub->os_name, rc);
170                 return rc;
171         }
172
173         /* empty */
174         if (!rc)
175                 return -ENOENT;
176
177         /* corrupted */
178         if (rc < buf.lb_len) {
179                 CDEBUG(D_LFSCK, "%s: fail to load scrub file, "
180                        "expected = %d: rc = %d\n",
181                        scrub->os_name, (int)buf.lb_len, rc);
182                 return -EFAULT;
183         }
184
185         scrub_file_to_cpu(sf, &scrub->os_file_disk);
186         if (sf->sf_magic != SCRUB_MAGIC_V1) {
187                 CDEBUG(D_LFSCK, "%s: invalid scrub magic 0x%x != 0x%x\n",
188                        scrub->os_name, sf->sf_magic, SCRUB_MAGIC_V1);
189                 return -EFAULT;
190         }
191
192         return 0;
193 }
194 EXPORT_SYMBOL(scrub_file_load);
195
196 int scrub_file_store(const struct lu_env *env, struct lustre_scrub *scrub)
197 {
198         struct scrub_file *sf = &scrub->os_file_disk;
199         struct dt_object *obj = scrub->os_obj;
200         struct dt_device *dev = scrub_obj2dev(obj);
201         struct lu_buf buf = {
202                 .lb_buf = sf,
203                 .lb_len = sizeof(*sf)
204         };
205         struct thandle *th;
206         loff_t pos = 0;
207         int rc;
208         ENTRY;
209
210         /* Skip store under rdonly mode. */
211         if (dev->dd_rdonly)
212                 RETURN(0);
213
214         scrub_file_to_le(sf, &scrub->os_file);
215         th = dt_trans_create(env, dev);
216         if (IS_ERR(th))
217                 GOTO(log, rc = PTR_ERR(th));
218
219         rc = dt_declare_record_write(env, obj, &buf, pos, th);
220         if (rc)
221                 GOTO(stop, rc);
222
223         rc = dt_trans_start_local(env, dev, th);
224         if (rc)
225                 GOTO(stop, rc);
226
227         rc = dt_record_write(env, obj, &buf, &pos, th);
228
229         GOTO(stop, rc);
230
231 stop:
232         dt_trans_stop(env, dev, th);
233
234 log:
235         if (rc)
236                 CERROR("%s: store scrub file: rc = %d\n",
237                        scrub->os_name, rc);
238         else
239                 CDEBUG(D_LFSCK, "%s: store scrub file: rc = %d\n",
240                        scrub->os_name, rc);
241
242         scrub->os_time_last_checkpoint = ktime_get_seconds();
243         scrub->os_time_next_checkpoint = scrub->os_time_last_checkpoint +
244                                          SCRUB_CHECKPOINT_INTERVAL;
245         return rc;
246 }
247 EXPORT_SYMBOL(scrub_file_store);
248
249 int scrub_checkpoint(const struct lu_env *env, struct lustre_scrub *scrub)
250 {
251         struct scrub_file *sf = &scrub->os_file;
252         time64_t now = ktime_get_seconds();
253         int rc;
254
255         if (likely(now < scrub->os_time_next_checkpoint ||
256                    scrub->os_new_checked == 0))
257                 return 0;
258
259         CDEBUG(D_LFSCK, "%s: OI scrub checkpoint at pos %llu\n",
260                scrub->os_name, scrub->os_pos_current);
261
262         down_write(&scrub->os_rwsem);
263         sf->sf_items_checked += scrub->os_new_checked;
264         scrub->os_new_checked = 0;
265         sf->sf_pos_last_checkpoint = scrub->os_pos_current;
266         sf->sf_time_last_checkpoint = ktime_get_real_seconds();
267         sf->sf_run_time += now - scrub->os_time_last_checkpoint;
268         rc = scrub_file_store(env, scrub);
269         up_write(&scrub->os_rwsem);
270
271         return rc;
272 }
273 EXPORT_SYMBOL(scrub_checkpoint);
274
275 int scrub_start(int (*threadfn)(void *data), struct lustre_scrub *scrub,
276                 void *data, __u32 flags)
277 {
278         struct task_struct *task;
279         int rc;
280         ENTRY;
281
282         if (scrub->os_task)
283                 RETURN(-EALREADY);
284
285         if (scrub->os_file.sf_status == SS_COMPLETED) {
286                 if (!(flags & SS_SET_FAILOUT))
287                         flags |= SS_CLEAR_FAILOUT;
288
289                 if (!(flags & SS_SET_DRYRUN))
290                         flags |= SS_CLEAR_DRYRUN;
291
292                 flags |= SS_RESET;
293         }
294
295         task = kthread_create(threadfn, data, "OI_scrub");
296         if (IS_ERR(task)) {
297                 rc = PTR_ERR(task);
298                 CERROR("%s: cannot start iteration thread: rc = %d\n",
299                        scrub->os_name, rc);
300                 RETURN(rc);
301         }
302         spin_lock(&scrub->os_lock);
303         if (scrub->os_task) {
304                 /* Lost a race */
305                 spin_unlock(&scrub->os_lock);
306                 kthread_stop(task);
307                 RETURN(-EALREADY);
308         }
309         scrub->os_start_flags = flags;
310         scrub->os_task = task;
311         wake_up_process(task);
312         spin_unlock(&scrub->os_lock);
313         wait_var_event(scrub, scrub->os_running || !scrub->os_task);
314
315         RETURN(0);
316 }
317 EXPORT_SYMBOL(scrub_start);
318
319 void scrub_stop(struct lustre_scrub *scrub)
320 {
321         struct task_struct *task;
322
323         spin_lock(&scrub->os_lock);
324         scrub->os_running = 0;
325         spin_unlock(&scrub->os_lock);
326         task = xchg(&scrub->os_task, NULL);
327         if (task)
328                 kthread_stop(task);
329 }
330 EXPORT_SYMBOL(scrub_stop);
331
332 const char *const scrub_status_names[] = {
333         "init",
334         "scanning",
335         "completed",
336         "failed",
337         "stopped",
338         "paused",
339         "crashed",
340         NULL
341 };
342
343 const char *const scrub_flags_names[] = {
344         "recreated",
345         "inconsistent",
346         "auto",
347         "upgrade",
348         NULL
349 };
350
351 const char *const scrub_param_names[] = {
352         "failout",
353         "dryrun",
354         NULL
355 };
356
357 static void scrub_bits_dump(struct seq_file *m, int bits,
358                             const char *const names[],
359                             const char *prefix)
360 {
361         int flag;
362         int i;
363
364         seq_printf(m, "%s:%c", prefix, bits != 0 ? ' ' : '\n');
365
366         for (i = 0, flag = 1; bits != 0; i++, flag = BIT(i)) {
367                 if (flag & bits) {
368                         bits &= ~flag;
369                         seq_printf(m, "%s%c", names[i],
370                                    bits != 0 ? ',' : '\n');
371                 }
372         }
373 }
374
375 static void scrub_time_dump(struct seq_file *m, time64_t time,
376                             const char *prefix)
377 {
378         if (time != 0)
379                 seq_printf(m, "%s: %llu seconds\n", prefix,
380                            ktime_get_real_seconds() - time);
381         else
382                 seq_printf(m, "%s: N/A\n", prefix);
383 }
384
385 static void scrub_pos_dump(struct seq_file *m, __u64 pos, const char *prefix)
386 {
387         if (pos != 0)
388                 seq_printf(m, "%s: %llu\n", prefix, pos);
389         else
390                 seq_printf(m, "%s: N/A\n", prefix);
391 }
392
393 void scrub_dump(struct seq_file *m, struct lustre_scrub *scrub)
394 {
395         struct scrub_file *sf = &scrub->os_file;
396         u64 checked;
397         s64 speed;
398
399         down_read(&scrub->os_rwsem);
400         seq_printf(m, "name: OI_scrub\n"
401                    "magic: 0x%x\n"
402                    "oi_files: %d\n"
403                    "status: %s\n",
404                    sf->sf_magic, (int)sf->sf_oi_count,
405                    scrub_status_names[sf->sf_status]);
406
407         scrub_bits_dump(m, sf->sf_flags, scrub_flags_names, "flags");
408
409         scrub_bits_dump(m, sf->sf_param, scrub_param_names, "param");
410
411         scrub_time_dump(m, sf->sf_time_last_complete,
412                         "time_since_last_completed");
413
414         scrub_time_dump(m, sf->sf_time_latest_start,
415                         "time_since_latest_start");
416
417         scrub_time_dump(m, sf->sf_time_last_checkpoint,
418                         "time_since_last_checkpoint");
419
420         scrub_pos_dump(m, sf->sf_pos_latest_start,
421                         "latest_start_position");
422
423         scrub_pos_dump(m, sf->sf_pos_last_checkpoint,
424                         "last_checkpoint_position");
425
426         scrub_pos_dump(m, sf->sf_pos_first_inconsistent,
427                         "first_failure_position");
428
429         checked = sf->sf_items_checked + scrub->os_new_checked;
430         seq_printf(m, "checked: %llu\n"
431                    "%s: %llu\n"
432                    "failed: %llu\n"
433                    "prior_%s: %llu\n"
434                    "noscrub: %llu\n"
435                    "igif: %llu\n"
436                    "success_count: %u\n",
437                    checked,
438                    sf->sf_param & SP_DRYRUN ? "inconsistent" : "updated",
439                    sf->sf_items_updated, sf->sf_items_failed,
440                    sf->sf_param & SP_DRYRUN ? "inconsistent" : "updated",
441                    sf->sf_items_updated_prior, sf->sf_items_noscrub,
442                    sf->sf_items_igif, sf->sf_success_count);
443
444         speed = checked;
445         if (scrub->os_running) {
446                 s64 new_checked = scrub->os_new_checked;
447                 time64_t duration;
448                 time64_t rtime;
449
450                 /* Since the time resolution is in seconds for new system
451                  * or small devices it ismore likely that duration will be
452                  * zero which will lead to inaccurate results.
453                  */
454                 duration = ktime_get_seconds() -
455                            scrub->os_time_last_checkpoint;
456                 if (duration != 0)
457                         new_checked = div_s64(new_checked, duration);
458
459                 rtime = sf->sf_run_time + duration;
460                 if (rtime != 0)
461                         speed = div_s64(speed, rtime);
462
463                 seq_printf(m, "run_time: %lld seconds\n"
464                            "average_speed: %lld objects/sec\n"
465                            "real_time_speed: %lld objects/sec\n"
466                            "current_position: %llu\n"
467                            "scrub_in_prior: %s\n"
468                            "scrub_full_speed: %s\n"
469                            "partial_scan: %s\n",
470                            rtime, speed, new_checked,
471                            scrub->os_pos_current,
472                            scrub->os_in_prior ? "yes" : "no",
473                            scrub->os_full_speed ? "yes" : "no",
474                            scrub->os_partial_scan ? "yes" : "no");
475         } else {
476                 if (sf->sf_run_time != 0)
477                         speed = div_s64(speed, sf->sf_run_time);
478                 seq_printf(m, "run_time: %d seconds\n"
479                            "average_speed: %lld objects/sec\n"
480                            "real_time_speed: N/A\n"
481                            "current_position: N/A\n",
482                            sf->sf_run_time, speed);
483         }
484
485         up_read(&scrub->os_rwsem);
486 }
487 EXPORT_SYMBOL(scrub_dump);
488
489 int lustre_liru_new(struct list_head *head, const struct lu_fid *pfid,
490                     const struct lu_fid *cfid, __u64 child,
491                     const char *name, int namelen)
492 {
493         struct lustre_index_restore_unit *liru;
494         int len = sizeof(*liru) + namelen + 1;
495
496         OBD_ALLOC(liru, len);
497         if (!liru)
498                 return -ENOMEM;
499
500         INIT_LIST_HEAD(&liru->liru_link);
501         liru->liru_pfid = *pfid;
502         liru->liru_cfid = *cfid;
503         liru->liru_clid = child;
504         liru->liru_len = len;
505         memcpy(liru->liru_name, name, namelen);
506         liru->liru_name[namelen] = 0;
507         list_add_tail(&liru->liru_link, head);
508
509         return 0;
510 }
511 EXPORT_SYMBOL(lustre_liru_new);
512
513 int lustre_index_register(struct dt_device *dev, const char *devname,
514                           struct list_head *head, spinlock_t *lock, int *guard,
515                           const struct lu_fid *fid,
516                           __u32 keysize, __u32 recsize)
517 {
518         struct lustre_index_backup_unit *libu, *pos;
519         int rc = 0;
520         ENTRY;
521
522         if (dev->dd_rdonly || *guard)
523                 RETURN(1);
524
525         OBD_ALLOC_PTR(libu);
526         if (!libu)
527                 RETURN(-ENOMEM);
528
529         INIT_LIST_HEAD(&libu->libu_link);
530         libu->libu_keysize = keysize;
531         libu->libu_recsize = recsize;
532         libu->libu_fid = *fid;
533
534         spin_lock(lock);
535         if (unlikely(*guard)) {
536                 spin_unlock(lock);
537                 OBD_FREE_PTR(libu);
538
539                 RETURN(1);
540         }
541
542         list_for_each_entry_reverse(pos, head, libu_link) {
543                 rc = lu_fid_cmp(&pos->libu_fid, fid);
544                 if (rc < 0) {
545                         list_add(&libu->libu_link, &pos->libu_link);
546                         spin_unlock(lock);
547
548                         RETURN(0);
549                 }
550
551                 if (!rc) {
552                         /* Registered already. But the former registered one
553                          * has different keysize/recsize. It may because that
554                          * the former values are from disk and corrupted, then
555                          * replace it with new values. */
556                         if (unlikely(keysize != pos->libu_keysize ||
557                                      recsize != pos->libu_recsize)) {
558                                 CWARN("%s: the index "DFID" has registered "
559                                       "with %u/%u, may be invalid, replace "
560                                       "with %u/%u\n",
561                                       devname, PFID(fid), pos->libu_keysize,
562                                       pos->libu_recsize, keysize, recsize);
563
564                                 pos->libu_keysize = keysize;
565                                 pos->libu_recsize = recsize;
566                         } else {
567                                 rc = 1;
568                         }
569
570                         spin_unlock(lock);
571                         OBD_FREE_PTR(libu);
572
573                         RETURN(rc);
574                 }
575         }
576
577         list_add(&libu->libu_link, head);
578         spin_unlock(lock);
579
580         RETURN(0);
581 }
582 EXPORT_SYMBOL(lustre_index_register);
583
584 static void lustre_index_degister(struct list_head *head, spinlock_t *lock,
585                                   const struct lu_fid *fid)
586 {
587         struct lustre_index_backup_unit *libu;
588         int rc = -ENOENT;
589
590         spin_lock(lock);
591         list_for_each_entry_reverse(libu, head, libu_link) {
592                 rc = lu_fid_cmp(&libu->libu_fid, fid);
593                 /* NOT registered. */
594                 if (rc < 0)
595                         break;
596
597                 if (!rc) {
598                         list_del(&libu->libu_link);
599                         break;
600                 }
601         }
602         spin_unlock(lock);
603
604         if (!rc)
605                 OBD_FREE_PTR(libu);
606 }
607
608 static void
609 lustre_index_backup_make_header(struct lustre_index_backup_header *header,
610                                 __u32 keysize, __u32 recsize,
611                                 const struct lu_fid *fid, __u32 count)
612 {
613         memset(header, 0, sizeof(*header));
614         header->libh_magic = cpu_to_le32(INDEX_BACKUP_MAGIC_V1);
615         header->libh_count = cpu_to_le32(count);
616         header->libh_keysize = cpu_to_le32(keysize);
617         header->libh_recsize = cpu_to_le32(recsize);
618         fid_cpu_to_le(&header->libh_owner, fid);
619 }
620
621 static int lustre_index_backup_body(const struct lu_env *env,
622                                     struct dt_object *obj, loff_t *pos,
623                                     void *buf, int bufsize)
624 {
625         struct dt_device *dev = lu2dt_dev(obj->do_lu.lo_dev);
626         struct thandle *th;
627         struct lu_buf lbuf = {
628                 .lb_buf = buf,
629                 .lb_len = bufsize
630         };
631         int rc;
632         ENTRY;
633
634         th = dt_trans_create(env, dev);
635         if (IS_ERR(th))
636                 RETURN(PTR_ERR(th));
637
638         rc = dt_declare_record_write(env, obj, &lbuf, *pos, th);
639         if (rc)
640                 GOTO(stop, rc);
641
642         rc = dt_trans_start_local(env, dev, th);
643         if (rc)
644                 GOTO(stop, rc);
645
646         rc = dt_record_write(env, obj, &lbuf, pos, th);
647
648         GOTO(stop, rc);
649
650 stop:
651         dt_trans_stop(env, dev, th);
652         return rc;
653 }
654
655 static int lustre_index_backup_header(const struct lu_env *env,
656                                       struct dt_object *obj,
657                                       const struct lu_fid *tgt_fid,
658                                       __u32 keysize, __u32 recsize,
659                                       void *buf, int bufsize, int count)
660 {
661         struct dt_device *dev = lu2dt_dev(obj->do_lu.lo_dev);
662         struct lustre_index_backup_header *header = buf;
663         struct lu_attr *la = buf;
664         struct thandle *th;
665         struct lu_buf lbuf = {
666                 .lb_buf = header,
667                 .lb_len = sizeof(*header)
668         };
669         loff_t size = sizeof(*header) + (keysize + recsize) * count;
670         loff_t pos = 0;
671         int rc;
672         bool punch = false;
673         ENTRY;
674
675         LASSERT(sizeof(*la) <= bufsize);
676         LASSERT(sizeof(*header) <= bufsize);
677
678         rc = dt_attr_get(env, obj, la);
679         if (rc)
680                 RETURN(rc);
681
682         if (la->la_size > size)
683                 punch = true;
684
685         lustre_index_backup_make_header(header, keysize, recsize,
686                                         tgt_fid, count);
687         th = dt_trans_create(env, dev);
688         if (IS_ERR(th))
689                 RETURN(PTR_ERR(th));
690
691         rc = dt_declare_record_write(env, obj, &lbuf, pos, th);
692         if (rc)
693                 GOTO(stop, rc);
694
695         if (punch) {
696                 rc = dt_declare_punch(env, obj, size, OBD_OBJECT_EOF, th);
697                 if (rc)
698                         GOTO(stop, rc);
699         }
700
701         rc = dt_trans_start_local(env, dev, th);
702         if (rc)
703                 GOTO(stop, rc);
704
705         rc = dt_record_write(env, obj, &lbuf, &pos, th);
706         if (!rc && punch)
707                 rc = dt_punch(env, obj, size, OBD_OBJECT_EOF, th);
708
709         GOTO(stop, rc);
710
711 stop:
712         dt_trans_stop(env, dev, th);
713         return rc;
714 }
715
716 static int lustre_index_update_lma(const struct lu_env *env,
717                                    struct dt_object *obj,
718                                    void *buf, int bufsize)
719 {
720         struct dt_device *dev = lu2dt_dev(obj->do_lu.lo_dev);
721         struct lustre_mdt_attrs *lma = buf;
722         struct lu_buf lbuf = {
723                 .lb_buf = lma,
724                 .lb_len = sizeof(struct lustre_ost_attrs)
725         };
726         struct thandle *th;
727         int fl = LU_XATTR_REPLACE;
728         int rc;
729         ENTRY;
730
731         LASSERT(bufsize >= lbuf.lb_len);
732
733         rc = dt_xattr_get(env, obj, &lbuf, XATTR_NAME_LMA);
734         if (unlikely(rc == -ENODATA)) {
735                 fl = LU_XATTR_CREATE;
736                 lustre_lma_init(lma, lu_object_fid(&obj->do_lu),
737                                 LMAC_IDX_BACKUP, 0);
738                 rc = sizeof(*lma);
739         } else if (rc < sizeof(*lma)) {
740                 RETURN(rc < 0 ? rc : -EFAULT);
741         } else {
742                 lustre_lma_swab(lma);
743                 if (lma->lma_compat & LMAC_IDX_BACKUP)
744                         RETURN(0);
745
746                 lma->lma_compat |= LMAC_IDX_BACKUP;
747         }
748
749         lustre_lma_swab(lma);
750         lbuf.lb_len = rc;
751         th = dt_trans_create(env, dev);
752         if (IS_ERR(th))
753                 RETURN(rc);
754
755         rc = dt_declare_xattr_set(env, obj, &lbuf, XATTR_NAME_LMA, fl, th);
756         if (rc)
757                 GOTO(stop, rc);
758
759         rc = dt_trans_start_local(env, dev, th);
760         if (rc)
761                 GOTO(stop, rc);
762
763         rc = dt_xattr_set(env, obj, &lbuf, XATTR_NAME_LMA, fl, th);
764
765         GOTO(stop, rc);
766
767 stop:
768         dt_trans_stop(env, dev, th);
769         return rc;
770 }
771
772 static int lustre_index_backup_one(const struct lu_env *env,
773                                    struct local_oid_storage *los,
774                                    struct dt_object *parent,
775                                    struct lustre_index_backup_unit *libu,
776                                    char *buf, int bufsize)
777 {
778         struct dt_device *dev = scrub_obj2dev(parent);
779         struct dt_object *tgt_obj = NULL;
780         struct dt_object *bak_obj = NULL;
781         const struct dt_it_ops *iops;
782         struct dt_it *di;
783         loff_t pos = sizeof(struct lustre_index_backup_header);
784         int count = 0;
785         int size = 0;
786         int rc;
787         ENTRY;
788
789         tgt_obj = lu2dt(lu_object_find_slice(env, &dev->dd_lu_dev,
790                                              &libu->libu_fid, NULL));
791         if (IS_ERR_OR_NULL(tgt_obj))
792                 GOTO(out, rc = tgt_obj ? PTR_ERR(tgt_obj) : -ENOENT);
793
794         if (!dt_object_exists(tgt_obj))
795                 GOTO(out, rc = 0);
796
797         if (!tgt_obj->do_index_ops) {
798                 struct dt_index_features feat;
799
800                 feat.dif_flags = DT_IND_UPDATE;
801                 feat.dif_keysize_min = libu->libu_keysize;
802                 feat.dif_keysize_max = libu->libu_keysize;
803                 feat.dif_recsize_min = libu->libu_recsize;
804                 feat.dif_recsize_max = libu->libu_recsize;
805                 feat.dif_ptrsize = 4;
806                 rc = tgt_obj->do_ops->do_index_try(env, tgt_obj, &feat);
807                 if (rc)
808                         GOTO(out, rc);
809         }
810
811         lustre_fid2lbx(buf, &libu->libu_fid, bufsize);
812         bak_obj = local_file_find_or_create(env, los, parent, buf,
813                                             S_IFREG | S_IRUGO | S_IWUSR);
814         if (IS_ERR_OR_NULL(bak_obj))
815                 GOTO(out, rc = bak_obj ? PTR_ERR(bak_obj) : -ENOENT);
816
817         iops = &tgt_obj->do_index_ops->dio_it;
818         di = iops->init(env, tgt_obj, 0);
819         if (IS_ERR(di))
820                 GOTO(out, rc = PTR_ERR(di));
821
822         rc = iops->load(env, di, 0);
823         if (!rc)
824                 rc = iops->next(env, di);
825         else if (rc > 0)
826                 rc = 0;
827
828         while (!rc) {
829                 void *key;
830                 void *rec;
831
832                 key = iops->key(env, di);
833                 memcpy(&buf[size], key, libu->libu_keysize);
834                 size += libu->libu_keysize;
835                 rec = &buf[size];
836                 rc = iops->rec(env, di, rec, 0);
837                 if (rc)
838                         GOTO(fini, rc);
839
840                 size += libu->libu_recsize;
841                 count++;
842                 if (size + libu->libu_keysize + libu->libu_recsize > bufsize) {
843                         rc = lustre_index_backup_body(env, bak_obj, &pos,
844                                                       buf, size);
845                         if (rc)
846                                 GOTO(fini, rc);
847
848                         size = 0;
849                 }
850
851                 rc = iops->next(env, di);
852         }
853
854         if (rc >= 0 && size > 0)
855                 rc = lustre_index_backup_body(env, bak_obj, &pos, buf, size);
856
857         if (rc < 0)
858                 GOTO(fini, rc);
859
860         rc = lustre_index_backup_header(env, bak_obj, &libu->libu_fid,
861                                         libu->libu_keysize, libu->libu_recsize,
862                                         buf, bufsize, count);
863         if (!rc)
864                 rc = lustre_index_update_lma(env, tgt_obj, buf, bufsize);
865
866         if (!rc && OBD_FAIL_CHECK(OBD_FAIL_OSD_INDEX_CRASH)) {
867                 LASSERT(bufsize >= 512);
868
869                 pos = 0;
870                 memset(buf, 0, 512);
871                 lustre_index_backup_body(env, tgt_obj, &pos, buf, 512);
872         }
873
874         GOTO(fini, rc);
875
876 fini:
877         iops->fini(env, di);
878 out:
879         if (!IS_ERR_OR_NULL(tgt_obj))
880                 dt_object_put_nocache(env, tgt_obj);
881         if (!IS_ERR_OR_NULL(bak_obj))
882                 dt_object_put_nocache(env, bak_obj);
883         return rc;
884 }
885
886 void lustre_index_backup(const struct lu_env *env, struct dt_device *dev,
887                          const char *devname, struct list_head *head,
888                          spinlock_t *lock, int *guard, bool backup)
889 {
890         struct lustre_index_backup_unit *libu;
891         struct local_oid_storage *los = NULL;
892         struct dt_object *parent = NULL;
893         char *buf = NULL;
894         struct lu_fid fid;
895         int rc;
896         ENTRY;
897
898         if (dev->dd_rdonly || *guard)
899                 RETURN_EXIT;
900
901         spin_lock(lock);
902         *guard = 1;
903         spin_unlock(lock);
904
905         if (list_empty(head))
906                 RETURN_EXIT;
907
908         /* Handle kinds of failures during mount process. */
909         if (!dev->dd_lu_dev.ld_site || !dev->dd_lu_dev.ld_site->ls_top_dev)
910                 backup = false;
911
912         if (backup) {
913                 OBD_ALLOC_LARGE(buf, INDEX_BACKUP_BUFSIZE);
914                 if (!buf) {
915                         backup = false;
916                         goto scan;
917                 }
918
919                 lu_local_obj_fid(&fid, INDEX_BACKUP_OID);
920                 parent = lu2dt(lu_object_find_slice(env, &dev->dd_lu_dev,
921                                                     &fid, NULL));
922                 if (IS_ERR_OR_NULL(parent)) {
923                         CERROR("%s: failed to locate backup dir: rc = %ld\n",
924                                devname, parent ? PTR_ERR(parent) : -ENOENT);
925                         backup = false;
926                         goto scan;
927                 }
928
929                 lu_local_name_obj_fid(&fid, 1);
930                 rc = local_oid_storage_init(env, dev, &fid, &los);
931                 if (rc) {
932                         CERROR("%s: failed to init local storage: rc = %d\n",
933                                devname, rc);
934                         backup = false;
935                 }
936         }
937
938 scan:
939         spin_lock(lock);
940         while (!list_empty(head)) {
941                 libu = list_entry(head->next,
942                                   struct lustre_index_backup_unit, libu_link);
943                 list_del_init(&libu->libu_link);
944                 spin_unlock(lock);
945
946                 if (backup) {
947                         rc = lustre_index_backup_one(env, los, parent, libu,
948                                                      buf, INDEX_BACKUP_BUFSIZE);
949                         CDEBUG(D_WARNING, "%s: backup index "DFID": rc = %d\n",
950                                devname, PFID(&libu->libu_fid), rc);
951                 }
952
953                 OBD_FREE_PTR(libu);
954                 spin_lock(lock);
955         }
956         spin_unlock(lock);
957
958         if (los)
959                 local_oid_storage_fini(env, los);
960         if (parent)
961                 dt_object_put_nocache(env, parent);
962         if (buf)
963                 OBD_FREE_LARGE(buf, INDEX_BACKUP_BUFSIZE);
964
965         EXIT;
966 }
967 EXPORT_SYMBOL(lustre_index_backup);
968
969 int lustre_index_restore(const struct lu_env *env, struct dt_device *dev,
970                          const struct lu_fid *parent_fid,
971                          const struct lu_fid *tgt_fid,
972                          const struct lu_fid *bak_fid, const char *name,
973                          struct list_head *head, spinlock_t *lock,
974                          char *buf, int bufsize)
975 {
976         struct dt_object *parent_obj = NULL;
977         struct dt_object *tgt_obj = NULL;
978         struct dt_object *bak_obj = NULL;
979         struct lustre_index_backup_header *header;
980         struct dt_index_features *feat;
981         struct dt_object_format *dof;
982         struct lu_attr *la;
983         struct thandle *th;
984         struct lu_object_conf conf;
985         struct dt_insert_rec ent;
986         struct lu_buf lbuf;
987         struct lu_fid tfid;
988         loff_t pos = 0;
989         __u32 keysize;
990         __u32 recsize;
991         __u32 pairsize;
992         int count;
993         int rc;
994         bool registered = false;
995         ENTRY;
996
997         LASSERT(bufsize >= sizeof(*la) + sizeof(*dof) +
998                 sizeof(*feat) + sizeof(*header));
999
1000         memset(buf, 0, bufsize);
1001         la = (struct lu_attr *)buf;
1002         dof = (void *)la + sizeof(*la);
1003         feat = (void *)dof + sizeof(*dof);
1004         header = (void *)feat + sizeof(*feat);
1005         lbuf.lb_buf = header;
1006         lbuf.lb_len = sizeof(*header);
1007
1008         tgt_obj = lu2dt(lu_object_find_slice(env, &dev->dd_lu_dev,
1009                                              tgt_fid, NULL));
1010         if (IS_ERR_OR_NULL(tgt_obj))
1011                 GOTO(out, rc = tgt_obj ? PTR_ERR(tgt_obj) : -ENOENT);
1012
1013         bak_obj = lu2dt(lu_object_find_slice(env, &dev->dd_lu_dev,
1014                                              bak_fid, NULL));
1015         if (IS_ERR_OR_NULL(bak_obj))
1016                 GOTO(out, rc = bak_obj ? PTR_ERR(bak_obj) : -ENOENT);
1017
1018         if (!dt_object_exists(bak_obj))
1019                 GOTO(out, rc = -ENOENT);
1020
1021         parent_obj = lu2dt(lu_object_find_slice(env, &dev->dd_lu_dev,
1022                                                 parent_fid, NULL));
1023         if (IS_ERR_OR_NULL(parent_obj))
1024                 GOTO(out, rc = parent_obj ? PTR_ERR(parent_obj) : -ENOENT);
1025
1026         LASSERT(dt_object_exists(parent_obj));
1027
1028         if (unlikely(!dt_try_as_dir(env, parent_obj)))
1029                 GOTO(out, rc = -ENOTDIR);
1030
1031         rc = dt_attr_get(env, tgt_obj, la);
1032         if (rc)
1033                 GOTO(out, rc);
1034
1035         rc = dt_record_read(env, bak_obj, &lbuf, &pos);
1036         if (rc)
1037                 GOTO(out, rc);
1038
1039         if (le32_to_cpu(header->libh_magic) != INDEX_BACKUP_MAGIC_V1)
1040                 GOTO(out, rc = -EINVAL);
1041
1042         fid_le_to_cpu(&tfid, &header->libh_owner);
1043         if (unlikely(!lu_fid_eq(tgt_fid, &tfid)))
1044                 GOTO(out, rc = -EINVAL);
1045
1046         keysize = le32_to_cpu(header->libh_keysize);
1047         recsize = le32_to_cpu(header->libh_recsize);
1048         pairsize = keysize + recsize;
1049
1050         memset(feat, 0, sizeof(*feat));
1051         feat->dif_flags = DT_IND_UPDATE;
1052         feat->dif_keysize_min = feat->dif_keysize_max = keysize;
1053         feat->dif_recsize_min = feat->dif_recsize_max = recsize;
1054         feat->dif_ptrsize = 4;
1055
1056         /* T1: remove old name entry and destroy old index. */
1057         th = dt_trans_create(env, dev);
1058         if (IS_ERR(th))
1059                 GOTO(out, rc = PTR_ERR(th));
1060
1061         rc = dt_declare_delete(env, parent_obj,
1062                                (const struct dt_key *)name, th);
1063         if (rc)
1064                 GOTO(stop, rc);
1065
1066         rc = dt_declare_ref_del(env, tgt_obj, th);
1067         if (rc)
1068                 GOTO(stop, rc);
1069
1070         rc = dt_declare_destroy(env, tgt_obj, th);
1071         if (rc)
1072                 GOTO(stop, rc);
1073
1074         rc = dt_trans_start_local(env, dev, th);
1075         if (rc)
1076                 GOTO(stop, rc);
1077
1078         rc = dt_delete(env, parent_obj, (const struct dt_key *)name, th);
1079         if (rc)
1080                 GOTO(stop, rc);
1081
1082         dt_write_lock(env, tgt_obj, 0);
1083         rc = dt_ref_del(env, tgt_obj, th);
1084         if (rc == 0) {
1085                 if (S_ISDIR(tgt_obj->do_lu.lo_header->loh_attr))
1086                         dt_ref_del(env, tgt_obj, th);
1087                 rc = dt_destroy(env, tgt_obj, th);
1088         }
1089         dt_write_unlock(env, tgt_obj);
1090         dt_trans_stop(env, dev, th);
1091         if (rc)
1092                 GOTO(out, rc);
1093
1094         la->la_valid = LA_MODE | LA_UID | LA_GID;
1095         conf.loc_flags = LOC_F_NEW;
1096         dof->u.dof_idx.di_feat = feat;
1097         dof->dof_type = DFT_INDEX;
1098         ent.rec_type = S_IFREG;
1099         ent.rec_fid = tgt_fid;
1100
1101         /* Drop cache before re-create it. */
1102         dt_object_put_nocache(env, tgt_obj);
1103         tgt_obj = lu2dt(lu_object_find_slice(env, &dev->dd_lu_dev,
1104                                              tgt_fid, &conf));
1105         if (IS_ERR_OR_NULL(tgt_obj))
1106                 GOTO(out, rc = tgt_obj ? PTR_ERR(tgt_obj) : -ENOENT);
1107
1108         LASSERT(!dt_object_exists(tgt_obj));
1109
1110         /* T2: create new index and insert new name entry. */
1111         th = dt_trans_create(env, dev);
1112         if (IS_ERR(th))
1113                 GOTO(out, rc = PTR_ERR(th));
1114
1115         rc = dt_declare_create(env, tgt_obj, la, NULL, dof, th);
1116         if (rc)
1117                 GOTO(stop, rc);
1118
1119         rc = dt_declare_insert(env, parent_obj, (const struct dt_rec *)&ent,
1120                                (const struct dt_key *)name, th);
1121         if (rc)
1122                 GOTO(stop, rc);
1123
1124         rc = dt_trans_start_local(env, dev, th);
1125         if (rc)
1126                 GOTO(stop, rc);
1127
1128         dt_write_lock(env, tgt_obj, 0);
1129         rc = dt_create(env, tgt_obj, la, NULL, dof, th);
1130         dt_write_unlock(env, tgt_obj);
1131         if (rc)
1132                 GOTO(stop, rc);
1133
1134         rc = dt_insert(env, parent_obj, (const struct dt_rec *)&ent,
1135                        (const struct dt_key *)name, th);
1136         dt_trans_stop(env, dev, th);
1137         /* Some index name may has been inserted by OSD
1138          * automatically when create the index object. */
1139         if (unlikely(rc == -EEXIST))
1140                 rc = 0;
1141         if (rc)
1142                 GOTO(out, rc);
1143
1144         /* The new index will register via index_try. */
1145         rc = tgt_obj->do_ops->do_index_try(env, tgt_obj, feat);
1146         if (rc)
1147                 GOTO(out, rc);
1148
1149         registered = true;
1150         count = le32_to_cpu(header->libh_count);
1151         while (!rc && count > 0) {
1152                 int size = pairsize * count;
1153                 int items = count;
1154                 int i;
1155
1156                 if (size > bufsize) {
1157                         items = bufsize / pairsize;
1158                         size = pairsize * items;
1159                 }
1160
1161                 lbuf.lb_buf = buf;
1162                 lbuf.lb_len = size;
1163                 rc = dt_record_read(env, bak_obj, &lbuf, &pos);
1164                 for (i = 0; i < items && !rc; i++) {
1165                         void *key = &buf[i * pairsize];
1166                         void *rec = &buf[i * pairsize + keysize];
1167
1168                         /* Tn: restore the records. */
1169                         th = dt_trans_create(env, dev);
1170                         if (!th)
1171                                 GOTO(out, rc = -ENOMEM);
1172
1173                         rc = dt_declare_insert(env, tgt_obj, rec, key, th);
1174                         if (rc)
1175                                 GOTO(stop, rc);
1176
1177                         rc = dt_trans_start_local(env, dev, th);
1178                         if (rc)
1179                                 GOTO(stop, rc);
1180
1181                         rc = dt_insert(env, tgt_obj, rec, key, th);
1182                         if (unlikely(rc == -EEXIST))
1183                                 rc = 0;
1184
1185                         dt_trans_stop(env, dev, th);
1186                 }
1187
1188                 count -= items;
1189         }
1190
1191         GOTO(out, rc);
1192
1193 stop:
1194         dt_trans_stop(env, dev, th);
1195         if (rc && registered)
1196                 /* Degister the index to avoid overwriting the backup. */
1197                 lustre_index_degister(head, lock, tgt_fid);
1198
1199 out:
1200         if (!IS_ERR_OR_NULL(tgt_obj))
1201                 dt_object_put_nocache(env, tgt_obj);
1202         if (!IS_ERR_OR_NULL(bak_obj))
1203                 dt_object_put_nocache(env, bak_obj);
1204         if (!IS_ERR_OR_NULL(parent_obj))
1205                 dt_object_put_nocache(env, parent_obj);
1206         return rc;
1207 }
1208 EXPORT_SYMBOL(lustre_index_restore);