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