Whamcloud - gitweb
f212c25198a33e9a51118f9d45b3287d9e9bb735
[fs/lustre-release.git] / lustre / mdt / mdt_coordinator.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,
11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13  * GNU General Public License version 2 for more details.  A copy is
14  * included in the COPYING file that accompanied this code.
15  *
16  * You should have received a copy of the GNU General Public License
17  * along with this program; if not, write to the Free Software
18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2011, 2012 Commissariat a l'energie atomique et aux energies
24  *                          alternatives
25  *
26  * Copyright (c) 2013, 2017, Intel Corporation.
27  * Use is subject to license terms.
28  */
29 /*
30  * lustre/mdt/mdt_coordinator.c
31  *
32  * Lustre HSM Coordinator
33  *
34  * Author: Jacques-Charles Lafoucriere <jacques-charles.lafoucriere@cea.fr>
35  * Author: Aurelien Degremont <aurelien.degremont@cea.fr>
36  * Author: Thomas Leibovici <thomas.leibovici@cea.fr>
37  */
38
39 #define DEBUG_SUBSYSTEM S_MDS
40
41 #include <linux/kthread.h>
42 #include <linux/kernel.h>
43 #include <obd_support.h>
44 #include <lustre_export.h>
45 #include <obd.h>
46 #include <lprocfs_status.h>
47 #include <lustre_log.h>
48 #include <lustre_kernelcomm.h>
49 #include "mdt_internal.h"
50
51 /**
52  * get obj and HSM attributes on a fid
53  * \param mti [IN] context
54  * \param fid [IN] object fid
55  * \param hsm [OUT] HSM meta data
56  * \retval obj or error (-ENOENT if not found)
57  */
58 struct mdt_object *mdt_hsm_get_md_hsm(struct mdt_thread_info *mti,
59                                       const struct lu_fid *fid,
60                                       struct md_hsm *hsm)
61 {
62         struct md_attr          *ma;
63         struct mdt_object       *obj;
64         int                      rc;
65         ENTRY;
66
67         ma = &mti->mti_attr;
68         ma->ma_need = MA_HSM;
69         ma->ma_valid = 0;
70
71         /* find object by FID */
72         obj = mdt_object_find(mti->mti_env, mti->mti_mdt, fid);
73         if (IS_ERR(obj))
74                 RETURN(obj);
75
76         if (!mdt_object_exists(obj)) {
77                 /* no more object */
78                 mdt_object_put(mti->mti_env, obj);
79                 RETURN(ERR_PTR(-ENOENT));
80         }
81
82         rc = mdt_attr_get_complex(mti, obj, ma);
83         if (rc) {
84                 mdt_object_put(mti->mti_env, obj);
85                 RETURN(ERR_PTR(rc));
86         }
87
88         if (ma->ma_valid & MA_HSM)
89                 *hsm = ma->ma_hsm;
90         else
91                 memset(hsm, 0, sizeof(*hsm));
92         ma->ma_valid = 0;
93         RETURN(obj);
94 }
95
96 void mdt_hsm_dump_hal(int level, const char *prefix,
97                       struct hsm_action_list *hal)
98 {
99         int                      i, sz;
100         struct hsm_action_item  *hai;
101         char                     buf[12];
102
103         CDEBUG(level, "%s: HAL header: version %X count %d"
104                       " archive_id %d flags %#llx\n",
105                prefix, hal->hal_version, hal->hal_count,
106                hal->hal_archive_id, hal->hal_flags);
107
108         hai = hai_first(hal);
109         for (i = 0; i < hal->hal_count; i++) {
110                 sz = hai->hai_len - sizeof(*hai);
111                 CDEBUG(level, "%s %d: fid="DFID" dfid="DFID
112                        " cookie=%#llx"
113                        " action=%s extent=%#llx-%#llx gid=%#llx"
114                        " datalen=%d data=[%s]\n",
115                        prefix, i,
116                        PFID(&hai->hai_fid), PFID(&hai->hai_dfid),
117                        hai->hai_cookie,
118                        hsm_copytool_action2name(hai->hai_action),
119                        hai->hai_extent.offset,
120                        hai->hai_extent.length,
121                        hai->hai_gid, sz,
122                        hai_dump_data_field(hai, buf, sizeof(buf)));
123                 hai = hai_next(hai);
124         }
125 }
126
127 /**
128  * data passed to llog_cat_process() callback
129  * to scan requests and take actions
130  */
131 struct hsm_scan_request {
132         int                      hal_sz;
133         int                      hal_used_sz;
134         struct hsm_action_list  *hal;
135 };
136
137 struct hsm_scan_data {
138         struct mdt_thread_info  *hsd_mti;
139         char                     hsd_fsname[MTI_NAME_MAXLEN + 1];
140         /* are we scanning the logs for housekeeping, or just looking
141          * for new work?
142          */
143         bool                     hsd_housekeeping;
144         bool                     hsd_one_restore;
145         u32                      hsd_start_cat_idx;
146         u32                      hsd_start_rec_idx;
147         int                      hsd_action_count;
148         int                      hsd_request_len; /* array alloc len */
149         int                      hsd_request_count; /* array used count */
150         struct hsm_scan_request *hsd_request;
151 };
152
153 static int mdt_cdt_waiting_cb(const struct lu_env *env,
154                               struct mdt_device *mdt,
155                               struct llog_handle *llh,
156                               struct llog_agent_req_rec *larr,
157                               struct hsm_scan_data *hsd)
158 {
159         struct coordinator *cdt = &mdt->mdt_coordinator;
160         struct hsm_scan_request *request;
161         struct hsm_action_item *hai;
162         size_t hai_size;
163         u32 archive_id;
164         bool wrapped;
165         int i;
166
167         /* Are agents full? */
168         if (atomic_read(&cdt->cdt_request_count) >= cdt->cdt_max_requests)
169                 RETURN(hsd->hsd_housekeeping ? 0 : LLOG_PROC_BREAK);
170
171         if (hsd->hsd_action_count + atomic_read(&cdt->cdt_request_count) >=
172             cdt->cdt_max_requests) {
173                 /* We cannot send any more request
174                  *
175                  *                     *** SPECIAL CASE ***
176                  *
177                  * Restore requests are too important not to schedule at least
178                  * one, everytime we can.
179                  */
180                 if (larr->arr_hai.hai_action != HSMA_RESTORE ||
181                     hsd->hsd_one_restore)
182                         RETURN(hsd->hsd_housekeeping ? 0 : LLOG_PROC_BREAK);
183         }
184
185         hai_size = cfs_size_round(larr->arr_hai.hai_len);
186         archive_id = larr->arr_archive_id;
187
188         /* Can we add this action to one of the existing HALs in hsd. */
189         request = NULL;
190         for (i = 0; i < hsd->hsd_request_count; i++) {
191                 if (hsd->hsd_request[i].hal->hal_archive_id == archive_id &&
192                     hsd->hsd_request[i].hal_used_sz + hai_size <=
193                     LDLM_MAXREQSIZE) {
194                         request = &hsd->hsd_request[i];
195                         break;
196                 }
197         }
198
199         /* Are we trying to force-schedule a request? */
200         if (hsd->hsd_action_count + atomic_read(&cdt->cdt_request_count) >=
201             cdt->cdt_max_requests) {
202                 /* Is there really no compatible hsm_scan_request? */
203                 if (!request) {
204                         for (i -= 1; i >= 0; i--) {
205                                 if (hsd->hsd_request[i].hal->hal_archive_id ==
206                                     archive_id) {
207                                         request = &hsd->hsd_request[i];
208                                         break;
209                                 }
210                         }
211                 }
212
213                 /* Make room for the hai */
214                 if (request) {
215                         /* Discard the last hai until there is enough space */
216                         do {
217                                 request->hal->hal_count--;
218
219                                 hai = hai_first(request->hal);
220                                 for (i = 0; i < request->hal->hal_count; i++)
221                                         hai = hai_next(hai);
222                                 request->hal_used_sz -=
223                                         cfs_size_round(hai->hai_len);
224                                 hsd->hsd_action_count--;
225                         } while (request->hal_used_sz + hai_size >
226                                  LDLM_MAXREQSIZE);
227                 } else if (hsd->hsd_housekeeping) {
228                         struct hsm_scan_request *tmp;
229
230                         /* Discard the (whole) last hal */
231                         hsd->hsd_request_count--;
232                         LASSERT(hsd->hsd_request_count >= 0);
233                         tmp = &hsd->hsd_request[hsd->hsd_request_count];
234                         hsd->hsd_action_count -= tmp->hal->hal_count;
235                         LASSERT(hsd->hsd_action_count >= 0);
236                         OBD_FREE(tmp->hal, tmp->hal_sz);
237                 } else {
238                         /* Bailing out, this code path is too hot */
239                         RETURN(LLOG_PROC_BREAK);
240
241                 }
242         }
243
244         if (!request) {
245                 struct hsm_action_list *hal;
246
247                 LASSERT(hsd->hsd_request_count < hsd->hsd_request_len);
248                 request = &hsd->hsd_request[hsd->hsd_request_count];
249
250                 /* allocates hai vector size just needs to be large
251                  * enough */
252                 request->hal_sz = sizeof(*request->hal) +
253                         cfs_size_round(MTI_NAME_MAXLEN + 1) + 2 * hai_size;
254                 OBD_ALLOC_LARGE(hal, request->hal_sz);
255                 if (!hal)
256                         RETURN(-ENOMEM);
257
258                 hal->hal_version = HAL_VERSION;
259                 strlcpy(hal->hal_fsname, hsd->hsd_fsname, MTI_NAME_MAXLEN + 1);
260                 hal->hal_archive_id = larr->arr_archive_id;
261                 hal->hal_flags = larr->arr_flags;
262                 hal->hal_count = 0;
263                 request->hal_used_sz = hal_size(hal);
264                 request->hal = hal;
265                 hsd->hsd_request_count++;
266         } else if (request->hal_sz < request->hal_used_sz + hai_size) {
267                 /* Not enough room, need an extension */
268                 void *hal_buffer;
269                 int sz;
270
271                 sz = min_t(int, 2 * request->hal_sz, LDLM_MAXREQSIZE);
272                 LASSERT(request->hal_used_sz + hai_size < sz);
273
274                 OBD_ALLOC_LARGE(hal_buffer, sz);
275                 if (!hal_buffer)
276                         RETURN(-ENOMEM);
277
278                 memcpy(hal_buffer, request->hal, request->hal_used_sz);
279                 OBD_FREE_LARGE(request->hal, request->hal_sz);
280                 request->hal = hal_buffer;
281                 request->hal_sz = sz;
282         }
283
284         hai = hai_first(request->hal);
285         for (i = 0; i < request->hal->hal_count; i++)
286                 hai = hai_next(hai);
287
288         memcpy(hai, &larr->arr_hai, larr->arr_hai.hai_len);
289
290         request->hal_used_sz += hai_size;
291         request->hal->hal_count++;
292
293         hsd->hsd_action_count++;
294
295         switch (hai->hai_action) {
296         case HSMA_CANCEL:
297                 break;
298         case HSMA_RESTORE:
299                 hsd->hsd_one_restore = true;
300                 fallthrough;
301         default:
302                 cdt_agent_record_hash_add(cdt, hai->hai_cookie,
303                                           llh->lgh_hdr->llh_cat_idx,
304                                           larr->arr_hdr.lrh_index);
305         }
306
307         wrapped = llh->lgh_hdr->llh_cat_idx >= llh->lgh_last_idx &&
308                   llh->lgh_hdr->llh_count > 1;
309         if ((!wrapped && llh->lgh_hdr->llh_cat_idx > hsd->hsd_start_cat_idx) ||
310             (wrapped && llh->lgh_hdr->llh_cat_idx < hsd->hsd_start_cat_idx) ||
311             (llh->lgh_hdr->llh_cat_idx == hsd->hsd_start_cat_idx &&
312              larr->arr_hdr.lrh_index > hsd->hsd_start_rec_idx)) {
313                 hsd->hsd_start_cat_idx = llh->lgh_hdr->llh_cat_idx;
314                 hsd->hsd_start_rec_idx = larr->arr_hdr.lrh_index;
315         }
316
317         RETURN(0);
318 }
319
320 static int mdt_cdt_started_cb(const struct lu_env *env,
321                               struct mdt_device *mdt,
322                               struct llog_handle *llh,
323                               struct llog_agent_req_rec *larr,
324                               struct hsm_scan_data *hsd)
325 {
326         struct coordinator *cdt = &mdt->mdt_coordinator;
327         struct hsm_action_item *hai = &larr->arr_hai;
328         struct cdt_agent_req *car;
329         time64_t now = ktime_get_real_seconds();
330         time64_t last;
331         enum changelog_rec_flags clf_flags;
332         int rc;
333
334         if (!hsd->hsd_housekeeping)
335                 RETURN(0);
336
337         /* we search for a running request
338          * error may happen if coordinator crashes or stopped
339          * with running request
340          */
341         car = mdt_cdt_find_request(cdt, hai->hai_cookie);
342         if (car == NULL) {
343                 last = larr->arr_req_change;
344         } else {
345                 last = car->car_req_update;
346         }
347
348         /* test if request too long, if yes cancel it
349          * the same way the copy tool acknowledge a cancel request */
350         if (now <= last + cdt->cdt_active_req_timeout)
351                 GOTO(out_car, rc = 0);
352
353         dump_llog_agent_req_rec("request timed out, start cleaning", larr);
354
355         if (car != NULL) {
356                 car->car_req_update = now;
357                 mdt_hsm_agent_update_statistics(cdt, 0, 1, 0, &car->car_uuid);
358                 /* Remove car from memory list (LU-9075) */
359                 mdt_cdt_remove_request(cdt, hai->hai_cookie);
360         }
361
362         /* Emit a changelog record for the failed action.*/
363         clf_flags = 0;
364         hsm_set_cl_error(&clf_flags, ECANCELED);
365
366         switch (hai->hai_action) {
367         case HSMA_ARCHIVE:
368                 hsm_set_cl_event(&clf_flags, HE_ARCHIVE);
369                 break;
370         case HSMA_RESTORE:
371                 hsm_set_cl_event(&clf_flags, HE_RESTORE);
372                 break;
373         case HSMA_REMOVE:
374                 hsm_set_cl_event(&clf_flags, HE_REMOVE);
375                 break;
376         case HSMA_CANCEL:
377                 hsm_set_cl_event(&clf_flags, HE_CANCEL);
378                 break;
379         default:
380                 /* Unknown record type, skip changelog. */
381                 clf_flags = 0;
382                 break;
383         }
384
385         if (clf_flags != 0)
386                 mo_changelog(env, CL_HSM, clf_flags, mdt->mdt_child,
387                              &hai->hai_fid);
388
389         if (hai->hai_action == HSMA_RESTORE)
390                 cdt_restore_handle_del(hsd->hsd_mti, cdt, &hai->hai_fid);
391
392         larr->arr_status = ARS_CANCELED;
393         larr->arr_req_change = now;
394         rc = llog_write(hsd->hsd_mti->mti_env, llh, &larr->arr_hdr,
395                         larr->arr_hdr.lrh_index);
396         if (rc < 0) {
397                 CERROR("%s: cannot update agent log: rc = %d\n",
398                        mdt_obd_name(mdt), rc);
399                 rc = LLOG_DEL_RECORD;
400         }
401
402         /* ct has completed a request, so a slot is available,
403          * signal the coordinator to find new work */
404         mdt_hsm_cdt_event(cdt);
405 out_car:
406         if (car != NULL)
407                 mdt_cdt_put_request(car);
408
409         RETURN(rc);
410 }
411
412 /**
413  *  llog_cat_process() callback, used to:
414  *  - find waiting request and start action
415  *  - purge canceled and done requests
416  * \param env [IN] environment
417  * \param llh [IN] llog handle
418  * \param hdr [IN] llog record
419  * \param data [IN/OUT] cb data = struct hsm_scan_data
420  * \retval 0 success
421  * \retval -ve failure
422  */
423 static int mdt_coordinator_cb(const struct lu_env *env,
424                               struct llog_handle *llh,
425                               struct llog_rec_hdr *hdr,
426                               void *data)
427 {
428         struct llog_agent_req_rec *larr = (struct llog_agent_req_rec *)hdr;
429         struct hsm_scan_data *hsd = data;
430         struct mdt_device *mdt = hsd->hsd_mti->mti_mdt;
431         struct coordinator *cdt = &mdt->mdt_coordinator;
432         ENTRY;
433
434         larr = (struct llog_agent_req_rec *)hdr;
435         dump_llog_agent_req_rec("mdt_coordinator_cb(): ", larr);
436         switch (larr->arr_status) {
437         case ARS_WAITING:
438                 RETURN(mdt_cdt_waiting_cb(env, mdt, llh, larr, hsd));
439         case ARS_STARTED:
440                 RETURN(mdt_cdt_started_cb(env, mdt, llh, larr, hsd));
441         default:
442                 if (!hsd->hsd_housekeeping)
443                         RETURN(0);
444
445                 if ((larr->arr_req_change + cdt->cdt_grace_delay) <
446                     ktime_get_real_seconds()) {
447                         cdt_agent_record_hash_del(cdt,
448                                                   larr->arr_hai.hai_cookie);
449                         RETURN(LLOG_DEL_RECORD);
450                 }
451
452                 RETURN(0);
453         }
454 }
455
456 /* Release the ressource used by the coordinator. Called when the
457  * coordinator is stopping. */
458 static void mdt_hsm_cdt_cleanup(struct mdt_device *mdt)
459 {
460         struct coordinator              *cdt = &mdt->mdt_coordinator;
461         struct cdt_agent_req            *car, *tmp1;
462         struct hsm_agent                *ha, *tmp2;
463         struct cdt_restore_handle       *crh, *tmp3;
464         struct mdt_thread_info          *cdt_mti;
465
466         /* start cleaning */
467         down_write(&cdt->cdt_request_lock);
468         list_for_each_entry_safe(car, tmp1, &cdt->cdt_request_list,
469                                  car_request_list) {
470                 cfs_hash_del(cdt->cdt_request_cookie_hash,
471                              &car->car_hai->hai_cookie,
472                              &car->car_cookie_hash);
473                 list_del(&car->car_request_list);
474                 mdt_cdt_put_request(car);
475         }
476         up_write(&cdt->cdt_request_lock);
477
478         down_write(&cdt->cdt_agent_lock);
479         list_for_each_entry_safe(ha, tmp2, &cdt->cdt_agents, ha_list) {
480                 list_del(&ha->ha_list);
481                 if (ha->ha_archive_cnt != 0)
482                         OBD_FREE_PTR_ARRAY(ha->ha_archive_id,
483                                            ha->ha_archive_cnt);
484                 OBD_FREE_PTR(ha);
485         }
486         up_write(&cdt->cdt_agent_lock);
487
488         cdt_mti = lu_context_key_get(&cdt->cdt_env.le_ctx, &mdt_thread_key);
489         mutex_lock(&cdt->cdt_restore_lock);
490         list_for_each_entry_safe(crh, tmp3, &cdt->cdt_restore_handle_list,
491                                  crh_list) {
492                 /* not locked yet, cleanup by cdt_restore_handle_add() */
493                 if (crh->crh_lh.mlh_type == MDT_NUL_LOCK)
494                         continue;
495                 list_del(&crh->crh_list);
496                 /* give back layout lock */
497                 mdt_object_unlock(cdt_mti, NULL, &crh->crh_lh, 1);
498                 OBD_SLAB_FREE_PTR(crh, mdt_hsm_cdt_kmem);
499         }
500         mutex_unlock(&cdt->cdt_restore_lock);
501 }
502
503 /*
504  * Coordinator state transition table, indexed on enum cdt_states, taking
505  * from and to states. For instance since CDT_INIT to CDT_RUNNING is a
506  * valid transition, cdt_transition[CDT_INIT][CDT_RUNNING] is true.
507  */
508 static bool cdt_transition[CDT_STATES_COUNT][CDT_STATES_COUNT] = {
509         /* from -> to:    stopped init   running disable stopping */
510         /* stopped */   { true,   true,  false,  false,  false },
511         /* init */      { true,   false, true,   false,  false },
512         /* running */   { false,  false, true,   true,   true },
513         /* disable */   { false,  false, true,   true,   true },
514         /* stopping */  { true,   false, false,  false,  false }
515 };
516
517 /**
518  * Change coordinator thread state
519  * Some combinations are not valid, so catch them here.
520  *
521  * Returns 0 on success, with old_state set if not NULL, or -EINVAL if
522  * the transition was not possible.
523  */
524 static int set_cdt_state_locked(struct coordinator *cdt,
525                                 enum cdt_states new_state)
526 {
527         int rc;
528         enum cdt_states state;
529
530         state = cdt->cdt_state;
531
532         if (cdt_transition[state][new_state]) {
533                 cdt->cdt_state = new_state;
534                 rc = 0;
535         } else {
536                 CDEBUG(D_HSM,
537                        "unexpected coordinator transition, from=%s, to=%s\n",
538                        cdt_mdt_state2str(state), cdt_mdt_state2str(new_state));
539                 rc = -EINVAL;
540         }
541
542         return rc;
543 }
544
545 static int set_cdt_state(struct coordinator *cdt, enum cdt_states new_state)
546 {
547         int rc;
548
549         mutex_lock(&cdt->cdt_state_lock);
550         rc = set_cdt_state_locked(cdt, new_state);
551         mutex_unlock(&cdt->cdt_state_lock);
552
553         return rc;
554 }
555
556 static int mdt_hsm_pending_restore(struct mdt_thread_info *mti);
557
558 static void cdt_start_pending_restore(struct mdt_device *mdt,
559                                       struct coordinator *cdt)
560 {
561         struct mdt_thread_info *cdt_mti;
562         unsigned int i = 0;
563         int rc;
564
565         /* wait until MDD initialize hsm actions llog */
566         while (!test_bit(MDT_FL_CFGLOG, &mdt->mdt_state) && i < obd_timeout) {
567                 schedule_timeout_interruptible(cfs_time_seconds(1));
568                 i++;
569         }
570         if (!test_bit(MDT_FL_CFGLOG, &mdt->mdt_state))
571                 CWARN("%s: trying to init HSM before MDD\n", mdt_obd_name(mdt));
572
573         /* set up list of started restore requests */
574         cdt_mti = lu_context_key_get(&cdt->cdt_env.le_ctx, &mdt_thread_key);
575         rc = mdt_hsm_pending_restore(cdt_mti);
576         if (rc)
577                 CERROR("%s: cannot take the layout locks needed for registered restore: %d\n",
578                        mdt_obd_name(mdt), rc);
579
580 }
581
582 /**
583  * coordinator thread
584  * \param data [IN] obd device
585  * \retval 0 success
586  * \retval -ve failure
587  */
588 static int mdt_coordinator(void *data)
589 {
590         struct mdt_thread_info  *mti = data;
591         struct mdt_device       *mdt = mti->mti_mdt;
592         struct coordinator      *cdt = &mdt->mdt_coordinator;
593         struct hsm_scan_data     hsd = { NULL };
594         time64_t                 last_housekeeping = 0;
595         size_t request_sz = 0;
596         int rc;
597         ENTRY;
598
599         CDEBUG(D_HSM, "%s: coordinator thread starting, pid=%d\n",
600                mdt_obd_name(mdt), current->pid);
601
602         hsd.hsd_mti = mti;
603         obd_uuid2fsname(hsd.hsd_fsname, mdt_obd_name(mdt),
604                         sizeof(hsd.hsd_fsname));
605
606         set_cdt_state(cdt, CDT_RUNNING);
607
608         /* Inform mdt_hsm_cdt_start(). */
609         wake_up(&cdt->cdt_waitq);
610         cdt_start_pending_restore(mdt, cdt);
611
612         while (1) {
613                 int i;
614                 int update_idx = 0;
615                 int updates_sz;
616                 int updates_cnt;
617                 u32 start_cat_idx;
618                 u32 start_rec_idx;
619                 struct hsm_record_update *updates;
620
621                 /* Limit execution of the expensive requests traversal
622                  * to at most one second. This prevents repeatedly
623                  * locking/unlocking the catalog for each request
624                  * and preventing other HSM operations from happening
625                  */
626                 wait_event_interruptible_timeout(cdt->cdt_waitq,
627                                                  kthread_should_stop() ||
628                                                  cdt->cdt_wakeup_coordinator,
629                                                  cfs_time_seconds(1));
630
631                 cdt->cdt_wakeup_coordinator = false;
632                 CDEBUG(D_HSM, "coordinator resumes\n");
633
634                 if (kthread_should_stop()) {
635                         CDEBUG(D_HSM, "Coordinator stops\n");
636                         rc = 0;
637                         break;
638                 }
639
640                 /* if coordinator is suspended continue to wait */
641                 if (cdt->cdt_state == CDT_DISABLE) {
642                         CDEBUG(D_HSM, "disable state, coordinator sleeps\n");
643                         continue;
644                 }
645
646                 /* If no event, and no housekeeping to do, continue to
647                  * wait. */
648                 if (last_housekeeping + cdt->cdt_loop_period <=
649                     ktime_get_real_seconds()) {
650                         last_housekeeping = ktime_get_real_seconds();
651                         hsd.hsd_housekeeping = true;
652                         start_cat_idx = 0;
653                         start_rec_idx = 0;
654                 } else if (cdt->cdt_event) {
655                         hsd.hsd_housekeeping = false;
656                         start_cat_idx = hsd.hsd_start_cat_idx;
657                         start_rec_idx = hsd.hsd_start_rec_idx;
658                 } else {
659                         continue;
660                 }
661
662                 cdt->cdt_event = false;
663
664                 CDEBUG(D_HSM, "coordinator starts reading llog\n");
665
666                 if (hsd.hsd_request_len != cdt->cdt_max_requests) {
667                         /* cdt_max_requests has changed,
668                          * we need to allocate a new buffer
669                          */
670                         struct hsm_scan_request *tmp = NULL;
671                         int max_requests = cdt->cdt_max_requests;
672                         OBD_ALLOC_LARGE(tmp, max_requests *
673                                         sizeof(struct hsm_scan_request));
674                         if (!tmp) {
675                                 CERROR("Failed to resize request buffer, "
676                                        "keeping it at %d\n",
677                                        hsd.hsd_request_len);
678                         } else {
679                                 if (hsd.hsd_request != NULL)
680                                         OBD_FREE_LARGE(hsd.hsd_request,
681                                                        request_sz);
682
683                                 hsd.hsd_request_len = max_requests;
684                                 request_sz = hsd.hsd_request_len *
685                                         sizeof(struct hsm_scan_request);
686                                 hsd.hsd_request = tmp;
687                         }
688                 }
689
690                 hsd.hsd_action_count = 0;
691                 hsd.hsd_request_count = 0;
692                 hsd.hsd_one_restore = false;
693
694                 rc = cdt_llog_process(mti->mti_env, mdt, mdt_coordinator_cb,
695                                       &hsd, start_cat_idx, start_rec_idx,
696                                       WRITE);
697                 if (rc < 0)
698                         goto clean_cb_alloc;
699
700                 CDEBUG(D_HSM, "found %d requests to send\n",
701                        hsd.hsd_request_count);
702
703                 if (list_empty(&cdt->cdt_agents)) {
704                         CDEBUG(D_HSM, "no agent available, "
705                                       "coordinator sleeps\n");
706                         /* reset HSM scanning index range. */
707                         hsd.hsd_start_cat_idx = start_cat_idx;
708                         hsd.hsd_start_rec_idx = start_rec_idx;
709                         goto clean_cb_alloc;
710                 }
711
712                 /* Compute how many HAI we have in all the requests */
713                 updates_cnt = 0;
714                 for (i = 0; i < hsd.hsd_request_count; i++) {
715                         const struct hsm_scan_request *request =
716                                 &hsd.hsd_request[i];
717
718                         updates_cnt += request->hal->hal_count;
719                 }
720
721                 /* Allocate a temporary array to store the cookies to
722                  * update, and their status. */
723                 updates_sz = updates_cnt * sizeof(*updates);
724                 OBD_ALLOC_LARGE(updates, updates_sz);
725                 if (updates == NULL) {
726                         CERROR("%s: Cannot allocate memory (%d bytes) "
727                                "for %d updates. Too many HSM requests?\n",
728                                mdt_obd_name(mdt), updates_sz, updates_cnt);
729                         goto clean_cb_alloc;
730                 }
731
732                 /* here hsd contains a list of requests to be started */
733                 for (i = 0; i < hsd.hsd_request_count; i++) {
734                         struct hsm_scan_request *request = &hsd.hsd_request[i];
735                         struct hsm_action_list  *hal = request->hal;
736                         struct hsm_action_item  *hai;
737                         int                      j;
738
739                         /* still room for work ? */
740                         if (atomic_read(&cdt->cdt_request_count) >=
741                             cdt->cdt_max_requests)
742                                 break;
743
744                         rc = mdt_hsm_agent_send(mti, hal, 0);
745                         /* if failure, we suppose it is temporary
746                          * if the copy tool failed to do the request
747                          * it has to use hsm_progress
748                          */
749
750                         /* set up cookie vector to set records status
751                          * after copy tools start or failed
752                          */
753                         hai = hai_first(hal);
754                         for (j = 0; j < hal->hal_count; j++) {
755                                 updates[update_idx].cookie = hai->hai_cookie;
756                                 updates[update_idx].status =
757                                         (rc ? ARS_WAITING : ARS_STARTED);
758                                 hai = hai_next(hai);
759                                 update_idx++;
760                         }
761
762                         /* TODO: narrow down the HSM action range that already
763                          * scanned accroding to the cookies when a failure
764                          * occurs.
765                          */
766                         if (rc) {
767                                 hsd.hsd_start_cat_idx = start_cat_idx;
768                                 hsd.hsd_start_rec_idx = start_rec_idx;
769                         }
770                 }
771
772                 if (update_idx) {
773                         rc = mdt_agent_record_update(mti, updates, update_idx);
774                         if (rc)
775                                 CERROR("%s: mdt_agent_record_update() failed, "
776                                        "rc=%d, cannot update records "
777                                        "for %d cookies\n",
778                                        mdt_obd_name(mdt), rc, update_idx);
779                 }
780
781                 OBD_FREE_LARGE(updates, updates_sz);
782
783 clean_cb_alloc:
784                 /* free hal allocated by callback */
785                 for (i = 0; i < hsd.hsd_request_count; i++) {
786                         struct hsm_scan_request *request = &hsd.hsd_request[i];
787
788                         OBD_FREE_LARGE(request->hal, request->hal_sz);
789                 }
790         }
791
792         if (hsd.hsd_request != NULL)
793                 OBD_FREE_LARGE(hsd.hsd_request, request_sz);
794
795         mdt_hsm_cdt_cleanup(mdt);
796
797         if (rc != 0)
798                 CERROR("%s: coordinator thread exiting, process=%d, rc=%d\n",
799                        mdt_obd_name(mdt), current->pid, rc);
800         else
801                 CDEBUG(D_HSM, "%s: coordinator thread exiting, process=%d,"
802                               " no error\n",
803                        mdt_obd_name(mdt), current->pid);
804
805         RETURN(rc);
806 }
807
808 /**
809  * register a new HSM restore handle for a file and take EX lock on the layout
810  * \param mti [IN] thread info
811  * \param cdt [IN] coordinator
812  * \param fid [IN] fid of the file to restore
813  * \param he  [IN] HSM extent
814  * \retval 0 success
815  * \retval 1 restore handle already exists for the fid
816  * \retval -ve failure
817  */
818 int cdt_restore_handle_add(struct mdt_thread_info *mti, struct coordinator *cdt,
819                            const struct lu_fid *fid,
820                            const struct hsm_extent *he)
821 {
822         struct mdt_lock_handle lh = { 0 };
823         struct cdt_restore_handle *crh;
824         struct mdt_object *obj;
825         int rc;
826         ENTRY;
827
828         OBD_SLAB_ALLOC_PTR(crh, mdt_hsm_cdt_kmem);
829         if (crh == NULL)
830                 RETURN(-ENOMEM);
831
832         crh->crh_fid = *fid;
833         /* in V1 all file is restored
834          * crh->extent.start = he->offset;
835          * crh->extent.end = he->offset + he->length;
836          */
837         crh->crh_extent.start = 0;
838         crh->crh_extent.end = he->length;
839         crh->crh_lh.mlh_type = MDT_NUL_LOCK;
840
841         mutex_lock(&cdt->cdt_restore_lock);
842         if (cdt_restore_handle_find(cdt, fid) != NULL)
843                 GOTO(out_crl, rc = 1);
844
845         if (unlikely(cdt->cdt_state == CDT_STOPPED ||
846                      cdt->cdt_state == CDT_STOPPING))
847                 GOTO(out_crl, rc = -EAGAIN);
848
849         list_add_tail(&crh->crh_list, &cdt->cdt_restore_handle_list);
850         mutex_unlock(&cdt->cdt_restore_lock);
851
852         /* get the layout lock */
853         mdt_lock_reg_init(&lh, LCK_EX);
854         obj = mdt_object_find_lock(mti, &crh->crh_fid, &lh,
855                                    MDS_INODELOCK_LAYOUT);
856         if (IS_ERR(obj)) {
857                 mutex_lock(&cdt->cdt_restore_lock);
858                 GOTO(out_ldel, rc = PTR_ERR(obj));
859         }
860
861         /* We do not keep a reference on the object during the restore
862          * which can be very long.
863          */
864         mdt_object_put(mti->mti_env, obj);
865
866         mutex_lock(&cdt->cdt_restore_lock);
867         if (unlikely(cdt->cdt_state == CDT_STOPPED ||
868                      cdt->cdt_state == CDT_STOPPING))
869                 GOTO(out_lh, rc = -EAGAIN);
870
871         crh->crh_lh = lh;
872         mutex_unlock(&cdt->cdt_restore_lock);
873
874         RETURN(0);
875 out_lh:
876         mdt_object_unlock(mti, NULL, &crh->crh_lh, 1);
877 out_ldel:
878         list_del(&crh->crh_list);
879 out_crl:
880         mutex_unlock(&cdt->cdt_restore_lock);
881         OBD_SLAB_FREE_PTR(crh, mdt_hsm_cdt_kmem);
882
883         return rc;
884 }
885
886 /**
887  * lookup a restore handle by FID
888  * caller needs to hold cdt_restore_lock
889  * \param cdt [IN] coordinator
890  * \param fid [IN] FID
891  * \retval cdt_restore_handle found
892  * \retval NULL not found
893  */
894 struct cdt_restore_handle *cdt_restore_handle_find(struct coordinator *cdt,
895                                                    const struct lu_fid *fid)
896 {
897         struct cdt_restore_handle *crh;
898         ENTRY;
899
900         list_for_each_entry(crh, &cdt->cdt_restore_handle_list, crh_list) {
901                 if (lu_fid_eq(&crh->crh_fid, fid))
902                         RETURN(crh);
903         }
904
905         RETURN(NULL);
906 }
907
908 void cdt_restore_handle_del(struct mdt_thread_info *mti,
909                             struct coordinator *cdt, const struct lu_fid *fid)
910 {
911         struct cdt_restore_handle *crh;
912
913         /* give back layout lock */
914         mutex_lock(&cdt->cdt_restore_lock);
915         crh = cdt_restore_handle_find(cdt, fid);
916         if (crh != NULL)
917                 list_del(&crh->crh_list);
918         mutex_unlock(&cdt->cdt_restore_lock);
919
920         if (crh == NULL)
921                 return;
922
923         /* XXX We pass a NULL object since the restore handle does not
924          * keep a reference on the object being restored. */
925         mdt_object_unlock(mti, NULL, &crh->crh_lh, 1);
926         OBD_SLAB_FREE_PTR(crh, mdt_hsm_cdt_kmem);
927 }
928
929 /**
930  * data passed to llog_cat_process() callback
931  * to scan requests and take actions
932  */
933 struct hsm_restore_data {
934         struct mdt_thread_info  *hrd_mti;
935 };
936
937 /**
938  *  llog_cat_process() callback, used to:
939  *  - find restore request and allocate the restore handle
940  * \param env [IN] environment
941  * \param llh [IN] llog handle
942  * \param hdr [IN] llog record
943  * \param data [IN/OUT] cb data = struct hsm_restore_data
944  * \retval 0 success
945  * \retval -ve failure
946  */
947 static int hsm_restore_cb(const struct lu_env *env,
948                           struct llog_handle *llh,
949                           struct llog_rec_hdr *hdr, void *data)
950 {
951         struct llog_agent_req_rec       *larr;
952         struct hsm_restore_data         *hrd;
953         struct hsm_action_item          *hai;
954         struct mdt_thread_info          *mti;
955         struct coordinator              *cdt;
956         int rc;
957         ENTRY;
958
959         hrd = data;
960         mti = hrd->hrd_mti;
961         cdt = &mti->mti_mdt->mdt_coordinator;
962
963         larr = (struct llog_agent_req_rec *)hdr;
964         hai = &larr->arr_hai;
965         if (hai->hai_cookie >= cdt->cdt_last_cookie) {
966                 /* update the cookie to avoid collision */
967                 cdt->cdt_last_cookie = hai->hai_cookie + 1;
968         }
969
970         if (hai->hai_action != HSMA_RESTORE ||
971             agent_req_in_final_state(larr->arr_status))
972                 RETURN(0);
973
974         /* restore request not in a final state */
975
976         /* force replay of restore requests left in started state from previous
977          * CDT context, to be canceled later if finally found to be incompatible
978          * when being re-started */
979         if (larr->arr_status == ARS_STARTED) {
980                 larr->arr_status = ARS_WAITING;
981                 larr->arr_req_change = ktime_get_real_seconds();
982                 rc = llog_write(env, llh, hdr, hdr->lrh_index);
983                 if (rc != 0)
984                         GOTO(out, rc);
985         }
986
987         rc = cdt_restore_handle_add(mti, cdt, &hai->hai_fid, &hai->hai_extent);
988         if (rc == 1)
989                 rc = 0;
990 out:
991         RETURN(rc);
992 }
993
994 /**
995  * restore coordinator state at startup
996  * the goal is to take a layout lock for each registered restore request
997  * \param mti [IN] context
998  */
999 static int mdt_hsm_pending_restore(struct mdt_thread_info *mti)
1000 {
1001         struct hsm_restore_data  hrd;
1002         int                      rc;
1003         ENTRY;
1004
1005         hrd.hrd_mti = mti;
1006
1007         rc = cdt_llog_process(mti->mti_env, mti->mti_mdt, hsm_restore_cb, &hrd,
1008                               0, 0, WRITE);
1009
1010         RETURN(rc);
1011 }
1012
1013 int hsm_init_ucred(struct lu_ucred *uc)
1014 {
1015         ENTRY;
1016         uc->uc_valid = UCRED_OLD;
1017         uc->uc_o_uid = 0;
1018         uc->uc_o_gid = 0;
1019         uc->uc_o_fsuid = 0;
1020         uc->uc_o_fsgid = 0;
1021         uc->uc_uid = 0;
1022         uc->uc_gid = 0;
1023         uc->uc_fsuid = 0;
1024         uc->uc_fsgid = 0;
1025         uc->uc_suppgids[0] = -1;
1026         uc->uc_suppgids[1] = -1;
1027         uc->uc_cap = cap_combine(CAP_FS_SET, CAP_NFSD_SET);
1028         uc->uc_umask = 0777;
1029         uc->uc_ginfo = NULL;
1030         uc->uc_identity = NULL;
1031         /* always record internal HSM activity if also enabled globally */
1032         uc->uc_enable_audit = 1;
1033         /* do not let rbac interfere with HSM internal processing */
1034         uc->uc_rbac_file_perms = 1;
1035         uc->uc_rbac_dne_ops = 1;
1036         uc->uc_rbac_quota_ops = 1;
1037         uc->uc_rbac_byfid_ops = 1;
1038         uc->uc_rbac_chlg_ops = 1;
1039
1040         RETURN(0);
1041 }
1042
1043 /**
1044  * initialize coordinator struct
1045  * \param mdt [IN] device
1046  * \retval 0 success
1047  * \retval -ve failure
1048  */
1049 int mdt_hsm_cdt_init(struct mdt_device *mdt)
1050 {
1051         struct coordinator      *cdt = &mdt->mdt_coordinator;
1052         struct mdt_thread_info  *cdt_mti = NULL;
1053         int                      rc;
1054         ENTRY;
1055
1056         init_waitqueue_head(&cdt->cdt_waitq);
1057         init_rwsem(&cdt->cdt_llog_lock);
1058         init_rwsem(&cdt->cdt_agent_lock);
1059         init_rwsem(&cdt->cdt_request_lock);
1060         mutex_init(&cdt->cdt_restore_lock);
1061         mutex_init(&cdt->cdt_state_lock);
1062         set_cdt_state(cdt, CDT_STOPPED);
1063
1064         INIT_LIST_HEAD(&cdt->cdt_request_list);
1065         INIT_LIST_HEAD(&cdt->cdt_agents);
1066         INIT_LIST_HEAD(&cdt->cdt_restore_handle_list);
1067
1068         cdt->cdt_request_cookie_hash = cfs_hash_create("REQUEST_COOKIE_HASH",
1069                                                        CFS_HASH_BITS_MIN,
1070                                                        CFS_HASH_BITS_MAX,
1071                                                        CFS_HASH_BKT_BITS,
1072                                                        0 /* extra bytes */,
1073                                                        CFS_HASH_MIN_THETA,
1074                                                        CFS_HASH_MAX_THETA,
1075                                                 &cdt_request_cookie_hash_ops,
1076                                                        CFS_HASH_DEFAULT);
1077         if (cdt->cdt_request_cookie_hash == NULL)
1078                 RETURN(-ENOMEM);
1079
1080         cdt->cdt_agent_record_hash = cfs_hash_create("AGENT_RECORD_HASH",
1081                                                      CFS_HASH_BITS_MIN,
1082                                                      CFS_HASH_BITS_MAX,
1083                                                      CFS_HASH_BKT_BITS,
1084                                                      0 /* extra bytes */,
1085                                                      CFS_HASH_MIN_THETA,
1086                                                      CFS_HASH_MAX_THETA,
1087                                                      &cdt_agent_record_hash_ops,
1088                                                      CFS_HASH_DEFAULT);
1089         if (cdt->cdt_agent_record_hash == NULL)
1090                 GOTO(out_request_cookie_hash, rc = -ENOMEM);
1091
1092         rc = lu_env_init(&cdt->cdt_env, LCT_MD_THREAD);
1093         if (rc < 0)
1094                 GOTO(out_agent_record_hash, rc);
1095
1096         /* for mdt_ucred(), lu_ucred stored in lu_ucred_key */
1097         rc = lu_context_init(&cdt->cdt_session, LCT_SERVER_SESSION);
1098         if (rc < 0)
1099                 GOTO(out_env, rc);
1100
1101         lu_context_enter(&cdt->cdt_session);
1102         cdt->cdt_env.le_ses = &cdt->cdt_session;
1103
1104         cdt_mti = lu_context_key_get(&cdt->cdt_env.le_ctx, &mdt_thread_key);
1105         LASSERT(cdt_mti != NULL);
1106
1107         cdt_mti->mti_env = &cdt->cdt_env;
1108         cdt_mti->mti_mdt = mdt;
1109
1110         hsm_init_ucred(mdt_ucred(cdt_mti));
1111
1112         /* default values for sysfs tunnables
1113          * can be override by MGS conf */
1114         cdt->cdt_default_archive_id = 1;
1115         cdt->cdt_grace_delay = 60;
1116         cdt->cdt_loop_period = 10;
1117         cdt->cdt_max_requests = 3;
1118         cdt->cdt_policy = CDT_DEFAULT_POLICY;
1119         cdt->cdt_active_req_timeout = 3600;
1120
1121         /* by default do not remove archives on last unlink */
1122         cdt->cdt_remove_archive_on_last_unlink = false;
1123
1124         RETURN(0);
1125
1126 out_env:
1127         lu_env_fini(&cdt->cdt_env);
1128 out_agent_record_hash:
1129         cfs_hash_putref(cdt->cdt_agent_record_hash);
1130         cdt->cdt_agent_record_hash = NULL;
1131 out_request_cookie_hash:
1132         cfs_hash_putref(cdt->cdt_request_cookie_hash);
1133         cdt->cdt_request_cookie_hash = NULL;
1134
1135         return rc;
1136 }
1137
1138 /**
1139  * free a coordinator thread
1140  * \param mdt [IN] device
1141  */
1142 int  mdt_hsm_cdt_fini(struct mdt_device *mdt)
1143 {
1144         struct coordinator *cdt = &mdt->mdt_coordinator;
1145         ENTRY;
1146
1147         lu_context_exit(cdt->cdt_env.le_ses);
1148         lu_context_fini(cdt->cdt_env.le_ses);
1149
1150         lu_env_fini(&cdt->cdt_env);
1151
1152         cfs_hash_putref(cdt->cdt_agent_record_hash);
1153         cdt->cdt_agent_record_hash = NULL;
1154
1155         cfs_hash_putref(cdt->cdt_request_cookie_hash);
1156         cdt->cdt_request_cookie_hash = NULL;
1157
1158         RETURN(0);
1159 }
1160
1161 /**
1162  * start a coordinator thread
1163  * \param mdt [IN] device
1164  * \retval 0 success
1165  * \retval -ve failure
1166  */
1167 static int mdt_hsm_cdt_start(struct mdt_device *mdt)
1168 {
1169         struct coordinator *cdt = &mdt->mdt_coordinator;
1170         struct mdt_thread_info *cdt_mti;
1171         int rc;
1172         void *ptr;
1173         struct task_struct *task;
1174         ENTRY;
1175
1176         /* functions defined but not yet used
1177          * this avoid compilation warning
1178          */
1179         ptr = dump_requests;
1180
1181         rc = set_cdt_state(cdt, CDT_INIT);
1182         if (rc) {
1183                 CERROR("%s: Coordinator already started or stopping\n",
1184                        mdt_obd_name(mdt));
1185                 RETURN(-EALREADY);
1186         }
1187
1188         BUILD_BUG_ON(BIT(CDT_POLICY_SHIFT_COUNT - 1) != CDT_POLICY_LAST);
1189         cdt->cdt_policy = CDT_DEFAULT_POLICY;
1190
1191         /* just need to be larger than previous one */
1192         /* cdt_last_cookie is protected by cdt_llog_lock */
1193         cdt->cdt_last_cookie = ktime_get_real_seconds();
1194         atomic_set(&cdt->cdt_request_count, 0);
1195         atomic_set(&cdt->cdt_archive_count, 0);
1196         atomic_set(&cdt->cdt_restore_count, 0);
1197         atomic_set(&cdt->cdt_remove_count, 0);
1198         cdt->cdt_user_request_mask = (1UL << HSMA_RESTORE);
1199         cdt->cdt_group_request_mask = (1UL << HSMA_RESTORE);
1200         cdt->cdt_other_request_mask = (1UL << HSMA_RESTORE);
1201
1202         /* to avoid deadlock when start is made through sysfs
1203          * sysfs entries are created by the coordinator thread
1204          */
1205         if (mdt->mdt_bottom->dd_rdonly)
1206                 RETURN(0);
1207
1208         cdt_mti = lu_context_key_get(&cdt->cdt_env.le_ctx, &mdt_thread_key);
1209         task = kthread_run(mdt_coordinator, cdt_mti, "hsm_cdtr");
1210         if (IS_ERR(task)) {
1211                 rc = PTR_ERR(task);
1212                 set_cdt_state(cdt, CDT_STOPPED);
1213                 CERROR("%s: error starting coordinator thread: %d\n",
1214                        mdt_obd_name(mdt), rc);
1215         } else {
1216                 cdt->cdt_task = task;
1217                 wait_event(cdt->cdt_waitq,
1218                            cdt->cdt_state != CDT_INIT);
1219                 CDEBUG(D_HSM, "%s: coordinator thread started\n",
1220                        mdt_obd_name(mdt));
1221                 rc = 0;
1222         }
1223
1224         RETURN(rc);
1225 }
1226
1227 /**
1228  * stop a coordinator thread
1229  * \param mdt [IN] device
1230  */
1231 int mdt_hsm_cdt_stop(struct mdt_device *mdt)
1232 {
1233         struct coordinator *cdt = &mdt->mdt_coordinator;
1234         int rc;
1235
1236         ENTRY;
1237         /* stop coordinator thread */
1238         rc = set_cdt_state(cdt, CDT_STOPPING);
1239         if (rc == 0) {
1240                 kthread_stop(cdt->cdt_task);
1241                 cdt->cdt_task = NULL;
1242                 set_cdt_state(cdt, CDT_STOPPED);
1243         }
1244
1245         RETURN(rc);
1246 }
1247
1248 static int mdt_hsm_set_exists(struct mdt_thread_info *mti,
1249                               const struct lu_fid *fid,
1250                               u32 archive_id)
1251 {
1252         struct mdt_object *obj;
1253         struct md_hsm mh;
1254         int rc;
1255
1256         obj = mdt_hsm_get_md_hsm(mti, fid, &mh);
1257         if (IS_ERR(obj))
1258                 GOTO(out, rc = PTR_ERR(obj));
1259
1260         if (mh.mh_flags & HS_EXISTS &&
1261             mh.mh_arch_id == archive_id)
1262                 GOTO(out_obj, rc = 0);
1263
1264         mh.mh_flags |= HS_EXISTS;
1265         mh.mh_arch_id = archive_id;
1266         rc = mdt_hsm_attr_set(mti, obj, &mh);
1267
1268 out_obj:
1269         mdt_object_put(mti->mti_env, obj);
1270 out:
1271         return rc;
1272 }
1273
1274 /**
1275  * register all requests from an hal in the memory list
1276  * \param mti [IN] context
1277  * \param hal [IN] request
1278  * \param uuid [OUT] in case of CANCEL, the uuid of the agent
1279  *  which is running the CT
1280  * \retval 0 success
1281  * \retval -ve failure
1282  */
1283 int mdt_hsm_add_hal(struct mdt_thread_info *mti,
1284                     struct hsm_action_list *hal, struct obd_uuid *uuid)
1285 {
1286         struct mdt_device       *mdt = mti->mti_mdt;
1287         struct coordinator      *cdt = &mdt->mdt_coordinator;
1288         struct hsm_action_item  *hai;
1289         int                      rc = 0, i;
1290         ENTRY;
1291
1292         /* register request in memory list */
1293         hai = hai_first(hal);
1294         for (i = 0; i < hal->hal_count; i++, hai = hai_next(hai)) {
1295                 struct cdt_agent_req *car;
1296
1297                 /* in case of a cancel request, we first mark the ondisk
1298                  * record of the request we want to stop as canceled
1299                  * this does not change the cancel record
1300                  * it will be done when updating the request status
1301                  */
1302                 if (hai->hai_action == HSMA_CANCEL) {
1303                         struct hsm_record_update update = {
1304                                 .cookie = hai->hai_cookie,
1305                                 .status = ARS_CANCELED,
1306                         };
1307
1308                         rc = mdt_agent_record_update(mti, &update, 1);
1309                         if (rc) {
1310                                 CERROR("%s: mdt_agent_record_update() failed, "
1311                                        "rc=%d, cannot update status to %s "
1312                                        "for cookie %#llx\n",
1313                                        mdt_obd_name(mdt), rc,
1314                                        agent_req_status2name(ARS_CANCELED),
1315                                        hai->hai_cookie);
1316                                 GOTO(out, rc);
1317                         }
1318
1319                         /* find the running request to set it canceled */
1320                         car = mdt_cdt_find_request(cdt, hai->hai_cookie);
1321                         if (car != NULL) {
1322                                 car->car_canceled = 1;
1323                                 /* uuid has to be changed to the one running the
1324                                 * request to cancel */
1325                                 *uuid = car->car_uuid;
1326                                 mdt_cdt_put_request(car);
1327                         }
1328                         /* no need to memorize cancel request
1329                          * this also avoid a deadlock when we receive
1330                          * a purge all requests command
1331                          */
1332                         continue;
1333                 }
1334
1335                 if (hai->hai_action == HSMA_ARCHIVE) {
1336                         rc = mdt_hsm_set_exists(mti, &hai->hai_fid,
1337                                                 hal->hal_archive_id);
1338                         if (rc == -ENOENT)
1339                                 continue;
1340                         else if (rc < 0)
1341                                 GOTO(out, rc);
1342                 }
1343
1344                 car = mdt_cdt_alloc_request(hal->hal_archive_id, hal->hal_flags,
1345                                             uuid, hai);
1346                 if (IS_ERR(car))
1347                         GOTO(out, rc = PTR_ERR(car));
1348
1349                 rc = mdt_cdt_add_request(cdt, car);
1350                 if (rc != 0)
1351                         mdt_cdt_free_request(car);
1352         }
1353 out:
1354         RETURN(rc);
1355 }
1356
1357 /**
1358  * swap layouts between 2 fids
1359  * \param mti [IN] context
1360  * \param obj [IN]
1361  * \param dfid [IN]
1362  * \param mh_common [IN] MD HSM
1363  */
1364 static int hsm_swap_layouts(struct mdt_thread_info *mti,
1365                             struct mdt_object *obj, const struct lu_fid *dfid,
1366                             struct md_hsm *mh_common)
1367 {
1368         struct mdt_object       *dobj;
1369         struct mdt_lock_handle  *dlh;
1370         int                      rc;
1371         ENTRY;
1372
1373         if (!mdt_object_exists(obj))
1374                 GOTO(out, rc = -ENOENT);
1375
1376         /* we already have layout lock on obj so take only
1377          * on dfid */
1378         dlh = &mti->mti_lh[MDT_LH_OLD];
1379         mdt_lock_reg_init(dlh, LCK_EX);
1380         dobj = mdt_object_find_lock(mti, dfid, dlh, MDS_INODELOCK_LAYOUT);
1381         if (IS_ERR(dobj))
1382                 GOTO(out, rc = PTR_ERR(dobj));
1383
1384         /* if copy tool closes the volatile before sending the final
1385          * progress through llapi_hsm_copy_end(), all the objects
1386          * are removed and mdd_swap_layout LBUG */
1387         if (!mdt_object_exists(dobj)) {
1388                 CERROR("%s: Copytool has closed volatile file "DFID"\n",
1389                        mdt_obd_name(mti->mti_mdt), PFID(dfid));
1390                 GOTO(out_dobj, rc = -ENOENT);
1391         }
1392         /* Since we only handle restores here, unconditionally use
1393          * SWAP_LAYOUTS_MDS_HSM flag to ensure original layout will
1394          * be preserved in case of failure during swap_layout and not
1395          * leave a file in an intermediate but incoherent state.
1396          * But need to setup HSM xattr of data FID before, reuse
1397          * mti and mh presets for FID in hsm_cdt_request_completed(),
1398          * only need to clear RELEASED and DIRTY.
1399          */
1400         mh_common->mh_flags &= ~(HS_RELEASED | HS_DIRTY);
1401         rc = mdt_hsm_attr_set(mti, dobj, mh_common);
1402         if (rc == 0)
1403                 rc = mo_swap_layouts(mti->mti_env,
1404                                      mdt_object_child(obj),
1405                                      mdt_object_child(dobj),
1406                                      SWAP_LAYOUTS_MDS_HSM);
1407         if (rc == 0) {
1408                 rc = mdt_lsom_downgrade(mti, obj);
1409                 if (rc)
1410                         CDEBUG(D_INODE,
1411                                "%s: File fid="DFID" SOM "
1412                                "downgrade failed, rc = %d\n",
1413                                mdt_obd_name(mti->mti_mdt),
1414                                PFID(mdt_object_fid(obj)), rc);
1415         }
1416 out_dobj:
1417         mdt_object_unlock_put(mti, dobj, dlh, 1);
1418 out:
1419         RETURN(rc);
1420 }
1421
1422 /**
1423  * update status of a completed request
1424  * \param mti [IN] context
1425  * \param pgs [IN] progress of the copy tool
1426  * \retval 0 success
1427  * \retval -ve failure
1428  */
1429 static int hsm_cdt_request_completed(struct mdt_thread_info *mti,
1430                                      struct hsm_progress_kernel *pgs,
1431                                      const struct cdt_agent_req *car,
1432                                      enum agent_req_status *status)
1433 {
1434         const struct lu_env *env = mti->mti_env;
1435         struct mdt_device *mdt = mti->mti_mdt;
1436         struct coordinator *cdt = &mdt->mdt_coordinator;
1437         struct mdt_object *obj = NULL;
1438         enum changelog_rec_flags clf_flags = 0;
1439         struct md_hsm mh;
1440         bool is_mh_changed;
1441         bool need_changelog = true;
1442         int rc = 0;
1443
1444         ENTRY;
1445         /* default is to retry */
1446         *status = ARS_WAITING;
1447
1448         /* find object by FID, mdt_hsm_get_md_hsm() returns obj or err
1449          * if error/removed continue anyway to get correct reporting done */
1450         obj = mdt_hsm_get_md_hsm(mti, &car->car_hai->hai_fid, &mh);
1451         /* we will update MD HSM only if needed */
1452         is_mh_changed = false;
1453
1454         /* no need to change mh->mh_arch_id
1455          * mdt_hsm_get_md_hsm() got it from disk and it is still valid
1456          */
1457         if (pgs->hpk_errval != 0) {
1458                 switch (pgs->hpk_errval) {
1459                 case ENOSYS:
1460                         /* the copy tool does not support cancel
1461                          * so the cancel request is failed
1462                          * As we cannot distinguish a cancel progress
1463                          * from another action progress (they have the
1464                          * same cookie), we suppose here the CT returns
1465                          * ENOSYS only if does not support cancel
1466                          */
1467                         /* this can also happen when cdt calls it to
1468                          * for a timed out request */
1469                         *status = ARS_FAILED;
1470                         /* to have a cancel event in changelog */
1471                         pgs->hpk_errval = ECANCELED;
1472                         break;
1473                 case ECANCELED:
1474                         /* the request record has already been set to
1475                          * ARS_CANCELED, this set the cancel request
1476                          * to ARS_SUCCEED */
1477                         *status = ARS_SUCCEED;
1478                         break;
1479                 default:
1480                         /* retry only if current policy or requested, and
1481                          * object is not on error/removed */
1482                         *status = (cdt->cdt_policy & CDT_NORETRY_ACTION ||
1483                                    !(pgs->hpk_flags & HP_FLAG_RETRY) ||
1484                                    IS_ERR(obj)) ? ARS_FAILED : ARS_WAITING;
1485                         break;
1486                 }
1487
1488                 rc = hsm_set_cl_error(&clf_flags, pgs->hpk_errval);
1489                 if (rc == -EOVERFLOW) {
1490                         CERROR("%s: Request %#llx on "DFID" failed, error code %d too large\n",
1491                                mdt_obd_name(mdt), pgs->hpk_cookie,
1492                                PFID(&pgs->hpk_fid), (int)abs(pgs->hpk_errval));
1493                         rc = 0;
1494                 }
1495
1496                 switch (car->car_hai->hai_action) {
1497                 case HSMA_ARCHIVE:
1498                         hsm_set_cl_event(&clf_flags, HE_ARCHIVE);
1499                         break;
1500                 case HSMA_RESTORE:
1501                         hsm_set_cl_event(&clf_flags, HE_RESTORE);
1502                         break;
1503                 case HSMA_REMOVE:
1504                         hsm_set_cl_event(&clf_flags, HE_REMOVE);
1505                         break;
1506                 case HSMA_CANCEL:
1507                         hsm_set_cl_event(&clf_flags, HE_CANCEL);
1508                         CERROR("%s: Failed request %#llx on "DFID
1509                                " cannot be a CANCEL\n",
1510                                mdt_obd_name(mdt),
1511                                pgs->hpk_cookie,
1512                                PFID(&pgs->hpk_fid));
1513                         break;
1514                 default:
1515                         CERROR("%s: Failed request %#llx on "DFID
1516                                " %d is an unknown action\n",
1517                                mdt_obd_name(mdt),
1518                                pgs->hpk_cookie, PFID(&pgs->hpk_fid),
1519                                car->car_hai->hai_action);
1520                         rc = -EINVAL;
1521                         break;
1522                 }
1523         } else {
1524                 *status = ARS_SUCCEED;
1525                 switch (car->car_hai->hai_action) {
1526                 case HSMA_ARCHIVE:
1527                         hsm_set_cl_event(&clf_flags, HE_ARCHIVE);
1528                         /* set ARCHIVE keep EXIST and clear LOST and
1529                          * DIRTY */
1530                         mh.mh_arch_ver = pgs->hpk_data_version;
1531                         mh.mh_flags |= HS_ARCHIVED;
1532                         mh.mh_flags &= ~(HS_LOST|HS_DIRTY);
1533                         is_mh_changed = true;
1534                         break;
1535                 case HSMA_RESTORE:
1536                         hsm_set_cl_event(&clf_flags, HE_RESTORE);
1537
1538                         /* do not clear RELEASED and DIRTY here
1539                          * this will occur in hsm_swap_layouts()
1540                          */
1541
1542                         /* Restoring has changed the file version on
1543                          * disk. */
1544                         mh.mh_arch_ver = pgs->hpk_data_version;
1545                         is_mh_changed = true;
1546                         break;
1547                 case HSMA_REMOVE:
1548                         hsm_set_cl_event(&clf_flags, HE_REMOVE);
1549                         /* clear ARCHIVED EXISTS and LOST */
1550                         mh.mh_flags &= ~(HS_ARCHIVED | HS_EXISTS | HS_LOST);
1551                         is_mh_changed = true;
1552                         break;
1553                 case HSMA_CANCEL:
1554                         hsm_set_cl_event(&clf_flags, HE_CANCEL);
1555                         CERROR("%s: Successful request %#llx on "DFID" cannot be a CANCEL\n",
1556                                mdt_obd_name(mdt),
1557                                pgs->hpk_cookie,
1558                                PFID(&pgs->hpk_fid));
1559                         break;
1560                 default:
1561                         CERROR("%s: Successful request %#llx on "DFID" %d is an unknown action\n",
1562                                mdt_obd_name(mdt),
1563                                pgs->hpk_cookie, PFID(&pgs->hpk_fid),
1564                                car->car_hai->hai_action);
1565                         rc = -EINVAL;
1566                         break;
1567                 }
1568         }
1569
1570         /* rc != 0 means error when analysing action, it may come from
1571          * a crasy CT no need to manage DIRTY
1572          * and if mdt_hsm_get_md_hsm() has returned an error, mh has not been
1573          * filled
1574          */
1575         if (rc == 0 && !IS_ERR(obj))
1576                 hsm_set_cl_flags(&clf_flags,
1577                                  mh.mh_flags & HS_DIRTY ? CLF_HSM_DIRTY : 0);
1578
1579         /* unlock is done later, after layout lock management */
1580         if (is_mh_changed && !IS_ERR(obj))
1581                 rc = mdt_hsm_attr_set(mti, obj, &mh);
1582
1583         /* we give back layout lock only if restore was successful or
1584          * if no retry will be attempted and if object is still alive,
1585          * in other cases we just unlock the object */
1586         if (car->car_hai->hai_action == HSMA_RESTORE) {
1587                 struct mdt_lock_handle *lh;
1588
1589                 /* restore in data FID done, we swap the layouts
1590                  * only if restore is successful */
1591                 if (pgs->hpk_errval == 0 && !IS_ERR(obj)) {
1592                         rc = hsm_swap_layouts(mti, obj, &car->car_hai->hai_dfid,
1593                                               &mh);
1594                         if (rc) {
1595                                 if (cdt->cdt_policy & CDT_NORETRY_ACTION)
1596                                         *status = ARS_FAILED;
1597                                 pgs->hpk_errval = -rc;
1598                                 hsm_set_cl_error(&clf_flags, pgs->hpk_errval);
1599                         }
1600                 }
1601                 /* we have to retry, so keep layout lock */
1602                 if (*status == ARS_WAITING)
1603                         GOTO(out, rc);
1604
1605                 /* restore special case, need to create ChangeLog record
1606                  * before to give back layout lock to avoid concurrent
1607                  * file updater to post out of order ChangeLog */
1608                 mo_changelog(env, CL_HSM, clf_flags, mdt->mdt_child,
1609                              &car->car_hai->hai_fid);
1610                 need_changelog = false;
1611
1612                 cdt_restore_handle_del(mti, cdt, &car->car_hai->hai_fid);
1613                 if (!IS_ERR_OR_NULL(obj)) {
1614                         /* flush UPDATE lock so attributes are upadated */
1615                         lh = &mti->mti_lh[MDT_LH_OLD];
1616                         mdt_lock_reg_init(lh, LCK_EX);
1617                         mdt_object_lock(mti, obj, lh, MDS_INODELOCK_UPDATE);
1618                         mdt_object_unlock(mti, obj, lh, 1);
1619                 }
1620         }
1621
1622         GOTO(out, rc);
1623
1624 out:
1625         /* always add a ChangeLog record */
1626         if (need_changelog)
1627                 mo_changelog(env, CL_HSM, clf_flags, mdt->mdt_child,
1628                              &car->car_hai->hai_fid);
1629
1630         if (!IS_ERR(obj))
1631                 mdt_object_put(mti->mti_env, obj);
1632
1633         RETURN(rc);
1634 }
1635
1636 /**
1637  * update status of a request
1638  * \param mti [IN] context
1639  * \param pgs [IN] progress of the copy tool
1640  * \retval 0 success
1641  * \retval -ve failure
1642  */
1643 int mdt_hsm_update_request_state(struct mdt_thread_info *mti,
1644                                  struct hsm_progress_kernel *pgs)
1645 {
1646         struct mdt_device       *mdt = mti->mti_mdt;
1647         struct coordinator      *cdt = &mdt->mdt_coordinator;
1648         struct cdt_agent_req    *car;
1649         int                      rc = 0;
1650         ENTRY;
1651
1652         /* no coordinator started, so we cannot serve requests */
1653         if (cdt->cdt_state == CDT_STOPPED)
1654                 RETURN(-EAGAIN);
1655
1656         /* first do sanity checks */
1657         car = mdt_cdt_update_request(cdt, pgs);
1658         if (IS_ERR(car)) {
1659                 CERROR("%s: Cannot find running request for cookie %#llx"
1660                        " on fid="DFID"\n",
1661                        mdt_obd_name(mdt),
1662                        pgs->hpk_cookie, PFID(&pgs->hpk_fid));
1663
1664                 RETURN(PTR_ERR(car));
1665         }
1666
1667         CDEBUG(D_HSM, "Progress received for fid="DFID" cookie=%#llx"
1668                       " action=%s flags=%d err=%d fid="DFID" dfid="DFID"\n",
1669                       PFID(&pgs->hpk_fid), pgs->hpk_cookie,
1670                       hsm_copytool_action2name(car->car_hai->hai_action),
1671                       pgs->hpk_flags, pgs->hpk_errval,
1672                       PFID(&car->car_hai->hai_fid),
1673                       PFID(&car->car_hai->hai_dfid));
1674
1675         /* progress is done on FID or data FID depending of the action and
1676          * of the copy progress */
1677         /* for restore progress is used to send back the data FID to cdt */
1678         if (car->car_hai->hai_action == HSMA_RESTORE &&
1679             lu_fid_eq(&car->car_hai->hai_fid, &car->car_hai->hai_dfid))
1680                 car->car_hai->hai_dfid = pgs->hpk_fid;
1681
1682         if ((car->car_hai->hai_action == HSMA_RESTORE ||
1683              car->car_hai->hai_action == HSMA_ARCHIVE) &&
1684             (!lu_fid_eq(&pgs->hpk_fid, &car->car_hai->hai_dfid) &&
1685              !lu_fid_eq(&pgs->hpk_fid, &car->car_hai->hai_fid))) {
1686                 CERROR("%s: Progress on "DFID" for cookie %#llx"
1687                        " does not match request FID "DFID" nor data FID "
1688                        DFID"\n",
1689                        mdt_obd_name(mdt),
1690                        PFID(&pgs->hpk_fid), pgs->hpk_cookie,
1691                        PFID(&car->car_hai->hai_fid),
1692                        PFID(&car->car_hai->hai_dfid));
1693                 GOTO(out, rc = -EINVAL);
1694         }
1695
1696         if (pgs->hpk_errval != 0 && !(pgs->hpk_flags & HP_FLAG_COMPLETED)) {
1697                 CERROR("%s: Progress on "DFID" for cookie %#llx action=%s"
1698                        " is not coherent (err=%d and not completed"
1699                        " (flags=%d))\n",
1700                        mdt_obd_name(mdt),
1701                        PFID(&pgs->hpk_fid), pgs->hpk_cookie,
1702                        hsm_copytool_action2name(car->car_hai->hai_action),
1703                        pgs->hpk_errval, pgs->hpk_flags);
1704                 GOTO(out, rc = -EINVAL);
1705         }
1706
1707         /* now progress is valid */
1708
1709         /* we use a root like ucred */
1710         hsm_init_ucred(mdt_ucred(mti));
1711
1712         if (pgs->hpk_flags & HP_FLAG_COMPLETED) {
1713                 enum agent_req_status status;
1714                 struct hsm_record_update update;
1715                 int rc1;
1716
1717                 rc = hsm_cdt_request_completed(mti, pgs, car, &status);
1718
1719                 CDEBUG(D_HSM, "updating record: fid="DFID" cookie=%#llx action=%s "
1720                               "status=%s\n",
1721                        PFID(&pgs->hpk_fid), pgs->hpk_cookie,
1722                        hsm_copytool_action2name(car->car_hai->hai_action),
1723                        agent_req_status2name(status));
1724
1725                 /* update record first (LU-9075) */
1726                 update.cookie = pgs->hpk_cookie;
1727                 update.status = status;
1728
1729                 rc1 = mdt_agent_record_update(mti, &update, 1);
1730                 if (rc1)
1731                         CERROR("%s: mdt_agent_record_update() failed,"
1732                                " rc=%d, cannot update status to %s"
1733                                " for cookie %#llx\n",
1734                                mdt_obd_name(mdt), rc1,
1735                                agent_req_status2name(status),
1736                                pgs->hpk_cookie);
1737                 rc = (rc != 0 ? rc : rc1);
1738
1739                 /* then remove request from memory list (LU-9075) */
1740                 mdt_cdt_remove_request(cdt, pgs->hpk_cookie);
1741
1742                 /* ct has completed a request, so a slot is available,
1743                  * signal the coordinator to find new work */
1744                 mdt_hsm_cdt_event(cdt);
1745         } else {
1746                 /* if copytool send a progress on a canceled request
1747                  * we inform copytool it should stop
1748                  */
1749                 if (car->car_canceled == 1)
1750                         rc = -ECANCELED;
1751         }
1752         GOTO(out, rc);
1753
1754 out:
1755         /* remove ref got from mdt_cdt_update_request() */
1756         mdt_cdt_put_request(car);
1757
1758         return rc;
1759 }
1760
1761
1762 /**
1763  *  llog_cat_process() callback, used to:
1764  *  - purge all requests
1765  * \param env [IN] environment
1766  * \param llh [IN] llog handle
1767  * \param hdr [IN] llog record
1768  * \param data [IN] cb data = struct mdt_thread_info
1769  * \retval 0 success
1770  * \retval -ve failure
1771  */
1772 static int mdt_cancel_all_cb(const struct lu_env *env,
1773                              struct llog_handle *llh,
1774                              struct llog_rec_hdr *hdr, void *data)
1775 {
1776         struct llog_agent_req_rec *larr = (struct llog_agent_req_rec *)hdr;
1777         struct hsm_action_item *hai = &larr->arr_hai;
1778         struct mdt_thread_info  *mti = data;
1779         struct coordinator *cdt = &mti->mti_mdt->mdt_coordinator;
1780         int rc;
1781         ENTRY;
1782
1783         if (larr->arr_status != ARS_WAITING &&
1784             larr->arr_status != ARS_STARTED)
1785                 RETURN(0);
1786
1787         /* Unlock the EX layout lock */
1788         if (hai->hai_action == HSMA_RESTORE)
1789                 cdt_restore_handle_del(mti, cdt, &hai->hai_fid);
1790
1791         larr->arr_status = ARS_CANCELED;
1792         larr->arr_req_change = ktime_get_real_seconds();
1793         rc = llog_write(env, llh, hdr, hdr->lrh_index);
1794         if (rc < 0) {
1795                 CERROR("%s: cannot update agent log: rc = %d\n",
1796                        mdt_obd_name(mti->mti_mdt), rc);
1797                 rc = LLOG_DEL_RECORD;
1798         }
1799
1800         RETURN(rc);
1801 }
1802
1803 /**
1804  * cancel all actions
1805  * \param obd [IN] MDT device
1806  */
1807 static int hsm_cancel_all_actions(struct mdt_device *mdt)
1808 {
1809         struct lu_env                    env;
1810         struct lu_context                session;
1811         struct mdt_thread_info          *mti;
1812         struct coordinator              *cdt = &mdt->mdt_coordinator;
1813         struct cdt_agent_req            *car;
1814         struct hsm_action_list          *hal = NULL;
1815         struct hsm_action_item          *hai;
1816         int                              hal_sz = 0, hal_len, rc;
1817         enum cdt_states                  old_state;
1818         ENTRY;
1819
1820         rc = lu_env_init(&env, LCT_MD_THREAD);
1821         if (rc < 0)
1822                 RETURN(rc);
1823
1824         /* for mdt_ucred(), lu_ucred stored in lu_ucred_key */
1825         rc = lu_context_init(&session, LCT_SERVER_SESSION);
1826         if (rc < 0)
1827                 GOTO(out_env, rc);
1828
1829         lu_context_enter(&session);
1830         env.le_ses = &session;
1831
1832         mti = lu_context_key_get(&env.le_ctx, &mdt_thread_key);
1833         LASSERT(mti != NULL);
1834
1835         mti->mti_env = &env;
1836         mti->mti_mdt = mdt;
1837
1838         hsm_init_ucred(mdt_ucred(mti));
1839
1840         mutex_lock(&cdt->cdt_state_lock);
1841         old_state = cdt->cdt_state;
1842
1843         /* disable coordinator */
1844         rc = set_cdt_state_locked(cdt, CDT_DISABLE);
1845         if (rc)
1846                 GOTO(out_cdt_state_unlock, rc);
1847
1848         /* send cancel to all running requests */
1849         down_read(&cdt->cdt_request_lock);
1850         list_for_each_entry(car, &cdt->cdt_request_list, car_request_list) {
1851                 mdt_cdt_get_request(car);
1852                 /* request is not yet removed from list, it will be done
1853                  * when copytool will return progress
1854                  */
1855
1856                 if (car->car_hai->hai_action == HSMA_CANCEL) {
1857                         mdt_cdt_put_request(car);
1858                         continue;
1859                 }
1860
1861                 /* needed size */
1862                 hal_len = sizeof(*hal) + cfs_size_round(MTI_NAME_MAXLEN + 1) +
1863                           cfs_size_round(car->car_hai->hai_len);
1864
1865                 if (hal_len > hal_sz && hal_sz > 0) {
1866                         /* not enough room, free old buffer */
1867                         OBD_FREE(hal, hal_sz);
1868                         hal = NULL;
1869                 }
1870
1871                 /* empty buffer, allocate one */
1872                 if (hal == NULL) {
1873                         hal_sz = hal_len;
1874                         OBD_ALLOC(hal, hal_sz);
1875                         if (hal == NULL) {
1876                                 mdt_cdt_put_request(car);
1877                                 up_read(&cdt->cdt_request_lock);
1878                                 GOTO(out_cdt_state, rc = -ENOMEM);
1879                         }
1880                 }
1881
1882                 hal->hal_version = HAL_VERSION;
1883                 obd_uuid2fsname(hal->hal_fsname, mdt_obd_name(mdt),
1884                                 MTI_NAME_MAXLEN);
1885                 hal->hal_fsname[MTI_NAME_MAXLEN] = '\0';
1886                 hal->hal_archive_id = car->car_archive_id;
1887                 hal->hal_flags = car->car_flags;
1888                 hal->hal_count = 0;
1889
1890                 hai = hai_first(hal);
1891                 memcpy(hai, car->car_hai, car->car_hai->hai_len);
1892                 hai->hai_action = HSMA_CANCEL;
1893                 hal->hal_count = 1;
1894
1895                 /* it is possible to safely call mdt_hsm_agent_send()
1896                  * (ie without a deadlock on cdt_request_lock), because the
1897                  * write lock is taken only if we are not in purge mode
1898                  * (mdt_hsm_agent_send() does not call mdt_cdt_add_request()
1899                  *   nor mdt_cdt_remove_request())
1900                  */
1901                 /* no conflict with cdt thread because cdt is disable and we
1902                  * have the request lock */
1903                 mdt_hsm_agent_send(mti, hal, 1);
1904
1905                 mdt_cdt_put_request(car);
1906         }
1907         up_read(&cdt->cdt_request_lock);
1908
1909         if (hal != NULL)
1910                 OBD_FREE(hal, hal_sz);
1911
1912         /* cancel all on-disk records */
1913         rc = cdt_llog_process(mti->mti_env, mti->mti_mdt, mdt_cancel_all_cb,
1914                               (void *)mti, 0, 0, WRITE);
1915 out_cdt_state:
1916         /* Enable coordinator, unless the coordinator was stopping. */
1917         set_cdt_state_locked(cdt, old_state);
1918 out_cdt_state_unlock:
1919         mutex_unlock(&cdt->cdt_state_lock);
1920
1921         lu_context_exit(&session);
1922         lu_context_fini(&session);
1923 out_env:
1924         lu_env_fini(&env);
1925
1926         RETURN(rc);
1927 }
1928
1929 /**
1930  * check if a request is compatible with file status
1931  * \param hai [IN] request description
1932  * \param archive_id [IN] request archive id
1933  * \param rq_flags [IN] request flags
1934  * \param hsm [IN] file HSM metadata
1935  * \retval boolean
1936  */
1937 bool mdt_hsm_is_action_compat(const struct hsm_action_item *hai,
1938                               u32 archive_id, u64 rq_flags,
1939                               const struct md_hsm *hsm)
1940 {
1941         int      is_compat = false;
1942         int      hsm_flags;
1943         ENTRY;
1944
1945         hsm_flags = hsm->mh_flags;
1946         switch (hai->hai_action) {
1947         case HSMA_ARCHIVE:
1948                 if (!(hsm_flags & HS_NOARCHIVE) &&
1949                     (hsm_flags & HS_DIRTY || !(hsm_flags & HS_ARCHIVED)))
1950                         is_compat = true;
1951
1952                 if (hsm_flags & HS_EXISTS &&
1953                     archive_id != 0 &&
1954                     archive_id != hsm->mh_arch_id)
1955                         is_compat = false;
1956
1957                 break;
1958         case HSMA_RESTORE:
1959                 if (!(hsm_flags & HS_DIRTY) && (hsm_flags & HS_RELEASED) &&
1960                     hsm_flags & HS_ARCHIVED && !(hsm_flags & HS_LOST))
1961                         is_compat = true;
1962                 break;
1963         case HSMA_REMOVE:
1964                 if (!(hsm_flags & HS_RELEASED) &&
1965                     (hsm_flags & (HS_ARCHIVED | HS_EXISTS)))
1966                         is_compat = true;
1967                 break;
1968         case HSMA_CANCEL:
1969                 is_compat = true;
1970                 break;
1971         }
1972         CDEBUG(D_HSM, "fid="DFID" action=%s flags=%#llx"
1973                       " extent=%#llx-%#llx hsm_flags=%.8X %s\n",
1974                       PFID(&hai->hai_fid),
1975                       hsm_copytool_action2name(hai->hai_action), rq_flags,
1976                       hai->hai_extent.offset, hai->hai_extent.length,
1977                       hsm->mh_flags,
1978                       (is_compat ? "compatible" : "uncompatible"));
1979
1980         RETURN(is_compat);
1981 }
1982
1983 /*
1984  * sysfs interface used to get/set HSM behaviour (cdt->cdt_policy)
1985  */
1986 static const struct {
1987         __u64            bit;
1988         char            *name;
1989         char            *nickname;
1990 } hsm_policy_names[] = {
1991         { CDT_NONBLOCKING_RESTORE,      "NonBlockingRestore",   "NBR"},
1992         { CDT_NORETRY_ACTION,           "NoRetryAction",        "NRA"},
1993         { 0 },
1994 };
1995
1996 /**
1997  * convert a policy name to a bit
1998  * \param name [IN] policy name
1999  * \retval 0 unknown
2000  * \retval   policy bit
2001  */
2002 static __u64 hsm_policy_str2bit(const char *name)
2003 {
2004         int      i;
2005
2006         for (i = 0; hsm_policy_names[i].bit != 0; i++)
2007                 if (strcmp(hsm_policy_names[i].nickname, name) == 0 ||
2008                     strcmp(hsm_policy_names[i].name, name) == 0)
2009                         return hsm_policy_names[i].bit;
2010         return 0;
2011 }
2012
2013 /**
2014  * convert a policy bit field to a string
2015  * \param mask [IN] policy bit field
2016  * \param hexa [IN] print mask before bit names
2017  * \param buffer [OUT] string
2018  * \param count [IN] size of buffer
2019  */
2020 static void hsm_policy_bit2str(struct seq_file *m, const __u64 mask,
2021                                 const bool hexa)
2022 {
2023         int      i, j;
2024         __u64    bit;
2025         ENTRY;
2026
2027         if (hexa)
2028                 seq_printf(m, "(%#llx) ", mask);
2029
2030         for (i = 0; i < CDT_POLICY_SHIFT_COUNT; i++) {
2031                 bit = (1ULL << i);
2032
2033                 for (j = 0; hsm_policy_names[j].bit != 0; j++) {
2034                         if (hsm_policy_names[j].bit == bit)
2035                                 break;
2036                 }
2037                 if (bit & mask)
2038                         seq_printf(m, "[%s] ", hsm_policy_names[j].name);
2039                 else
2040                         seq_printf(m, "%s ", hsm_policy_names[j].name);
2041         }
2042         /* remove last ' ' */
2043         m->count--;
2044         seq_putc(m, '\n');
2045 }
2046
2047 /* methods to read/write HSM policy flags */
2048 static int mdt_hsm_policy_seq_show(struct seq_file *m, void *data)
2049 {
2050         struct mdt_device       *mdt = m->private;
2051         struct coordinator      *cdt = &mdt->mdt_coordinator;
2052         ENTRY;
2053
2054         hsm_policy_bit2str(m, cdt->cdt_policy, false);
2055         RETURN(0);
2056 }
2057
2058 static ssize_t
2059 mdt_hsm_policy_seq_write(struct file *file, const char __user *buffer,
2060                          size_t count, loff_t *off)
2061 {
2062         struct seq_file         *m = file->private_data;
2063         struct mdt_device       *mdt = m->private;
2064         struct coordinator      *cdt = &mdt->mdt_coordinator;
2065         char                    *start, *token, sign;
2066         char                    *buf;
2067         __u64                    policy;
2068         __u64                    add_mask, remove_mask, set_mask;
2069         int                      rc;
2070         ENTRY;
2071
2072         if (count + 1 > PAGE_SIZE)
2073                 RETURN(-EINVAL);
2074
2075         OBD_ALLOC(buf, count + 1);
2076         if (buf == NULL)
2077                 RETURN(-ENOMEM);
2078
2079         if (copy_from_user(buf, buffer, count))
2080                 GOTO(out, rc = -EFAULT);
2081
2082         buf[count] = '\0';
2083
2084         start = buf;
2085         CDEBUG(D_HSM, "%s: receive new policy: '%s'\n", mdt_obd_name(mdt),
2086                start);
2087
2088         add_mask = remove_mask = set_mask = 0;
2089         do {
2090                 token = strsep(&start, "\n ");
2091                 sign = *token;
2092
2093                 if (sign == '\0')
2094                         continue;
2095
2096                 if (sign == '-' || sign == '+')
2097                         token++;
2098
2099                 policy = hsm_policy_str2bit(token);
2100                 if (policy == 0) {
2101                         CWARN("%s: '%s' is unknown, "
2102                               "supported policies are:\n", mdt_obd_name(mdt),
2103                               token);
2104                         hsm_policy_bit2str(m, 0, false);
2105                         GOTO(out, rc = -EINVAL);
2106                 }
2107                 switch (sign) {
2108                 case '-':
2109                         remove_mask |= policy;
2110                         break;
2111                 case '+':
2112                         add_mask |= policy;
2113                         break;
2114                 default:
2115                         set_mask |= policy;
2116                         break;
2117                 }
2118
2119         } while (start != NULL);
2120
2121         CDEBUG(D_HSM, "%s: new policy: rm=%#llx add=%#llx set=%#llx\n",
2122                mdt_obd_name(mdt), remove_mask, add_mask, set_mask);
2123
2124         /* if no sign in all string, it is a clear and set
2125          * if some sign found, all unsigned are converted
2126          * to add
2127          * P1 P2 = set to P1 and P2
2128          * P1 -P2 = add P1 clear P2 same as +P1 -P2
2129          */
2130         if (remove_mask == 0 && add_mask == 0) {
2131                 cdt->cdt_policy = set_mask;
2132         } else {
2133                 cdt->cdt_policy |= set_mask | add_mask;
2134                 cdt->cdt_policy &= ~remove_mask;
2135         }
2136
2137         GOTO(out, rc = count);
2138
2139 out:
2140         OBD_FREE(buf, count + 1);
2141         RETURN(rc);
2142 }
2143 LDEBUGFS_SEQ_FOPS(mdt_hsm_policy);
2144
2145 ssize_t loop_period_show(struct kobject *kobj, struct attribute *attr,
2146                          char *buf)
2147 {
2148         struct coordinator *cdt = container_of(kobj, struct coordinator,
2149                                                cdt_hsm_kobj);
2150
2151         return scnprintf(buf, PAGE_SIZE, "%u\n", cdt->cdt_loop_period);
2152 }
2153
2154 ssize_t loop_period_store(struct kobject *kobj, struct attribute *attr,
2155                           const char *buffer, size_t count)
2156 {
2157         struct coordinator *cdt = container_of(kobj, struct coordinator,
2158                                                cdt_hsm_kobj);
2159         unsigned int val;
2160         int rc;
2161
2162         rc = kstrtouint(buffer, 0, &val);
2163         if (rc)
2164                 return rc;
2165
2166         if (val != 0)
2167                 cdt->cdt_loop_period = val;
2168
2169         return val ? count : -EINVAL;
2170 }
2171 LUSTRE_RW_ATTR(loop_period);
2172
2173 ssize_t grace_delay_show(struct kobject *kobj, struct attribute *attr,
2174                          char *buf)
2175 {
2176         struct coordinator *cdt = container_of(kobj, struct coordinator,
2177                                                cdt_hsm_kobj);
2178
2179         return scnprintf(buf, PAGE_SIZE, "%u\n", cdt->cdt_grace_delay);
2180 }
2181
2182 ssize_t grace_delay_store(struct kobject *kobj, struct attribute *attr,
2183                           const char *buffer, size_t count)
2184 {
2185         struct coordinator *cdt = container_of(kobj, struct coordinator,
2186                                                cdt_hsm_kobj);
2187         unsigned int val;
2188         int rc;
2189
2190         rc = kstrtouint(buffer, 0, &val);
2191         if (rc)
2192                 return rc;
2193
2194         if (val != 0)
2195                 cdt->cdt_grace_delay = val;
2196
2197         return val ? count : -EINVAL;
2198 }
2199 LUSTRE_RW_ATTR(grace_delay);
2200
2201 ssize_t active_request_timeout_show(struct kobject *kobj,
2202                                     struct attribute *attr,
2203                                     char *buf)
2204 {
2205         struct coordinator *cdt = container_of(kobj, struct coordinator,
2206                                                cdt_hsm_kobj);
2207
2208         return scnprintf(buf, PAGE_SIZE, "%d\n", cdt->cdt_active_req_timeout);
2209 }
2210
2211 ssize_t active_request_timeout_store(struct kobject *kobj,
2212                                      struct attribute *attr,
2213                                      const char *buffer, size_t count)
2214 {
2215         struct coordinator *cdt = container_of(kobj, struct coordinator,
2216                                                cdt_hsm_kobj);
2217         unsigned int val;
2218         int rc;
2219
2220         rc = kstrtouint(buffer, 0, &val);
2221         if (rc)
2222                 return rc;
2223
2224         if (val != 0)
2225                 cdt->cdt_active_req_timeout = val;
2226
2227         return val ? count : -EINVAL;
2228 }
2229 LUSTRE_RW_ATTR(active_request_timeout);
2230
2231 ssize_t max_requests_show(struct kobject *kobj, struct attribute *attr,
2232                           char *buf)
2233 {
2234         struct coordinator *cdt = container_of(kobj, struct coordinator,
2235                                                cdt_hsm_kobj);
2236
2237         return scnprintf(buf, PAGE_SIZE, "%llu\n", cdt->cdt_max_requests);
2238 }
2239
2240 ssize_t max_requests_store(struct kobject *kobj, struct attribute *attr,
2241                            const char *buffer, size_t count)
2242 {
2243         struct coordinator *cdt = container_of(kobj, struct coordinator,
2244                                                cdt_hsm_kobj);
2245         unsigned long long val;
2246         int rc;
2247
2248         rc = kstrtoull(buffer, 0, &val);
2249         if (rc)
2250                 return rc;
2251
2252         if (val != 0)
2253                 cdt->cdt_max_requests = val;
2254
2255         return val ? count : -EINVAL;
2256 }
2257 LUSTRE_RW_ATTR(max_requests);
2258
2259 ssize_t default_archive_id_show(struct kobject *kobj, struct attribute *attr,
2260                                 char *buf)
2261 {
2262         struct coordinator *cdt = container_of(kobj, struct coordinator,
2263                                                cdt_hsm_kobj);
2264
2265         return scnprintf(buf, PAGE_SIZE, "%u\n", cdt->cdt_default_archive_id);
2266 }
2267
2268 ssize_t default_archive_id_store(struct kobject *kobj, struct attribute *attr,
2269                                  const char *buffer, size_t count)
2270 {
2271         struct coordinator *cdt = container_of(kobj, struct coordinator,
2272                                                cdt_hsm_kobj);
2273         unsigned int val;
2274         int rc;
2275
2276         rc = kstrtouint(buffer, 0, &val);
2277         if (rc)
2278                 return rc;
2279
2280         if (val != 0)
2281                 cdt->cdt_default_archive_id = val;
2282
2283         return val ? count : -EINVAL;
2284 }
2285 LUSTRE_RW_ATTR(default_archive_id);
2286
2287 /*
2288  * procfs write method for MDT/hsm_control
2289  * proc entry is in mdt directory so data is mdt obd_device pointer
2290  */
2291 #define CDT_ENABLE_CMD   "enabled"
2292 #define CDT_STOP_CMD     "shutdown"
2293 #define CDT_DISABLE_CMD  "disabled"
2294 #define CDT_PURGE_CMD    "purge"
2295 #define CDT_HELP_CMD     "help"
2296 #define CDT_MAX_CMD_LEN  10
2297
2298 ssize_t hsm_control_store(struct kobject *kobj, struct attribute *attr,
2299                           const char *buffer, size_t count)
2300 {
2301         struct obd_device *obd = container_of(kobj, struct obd_device,
2302                                               obd_kset.kobj);
2303         struct mdt_device *mdt = mdt_dev(obd->obd_lu_dev);
2304         struct coordinator *cdt = &(mdt->mdt_coordinator);
2305         int usage = 0;
2306         int rc = 0;
2307
2308         if (count == 0 || count >= CDT_MAX_CMD_LEN)
2309                 return -EINVAL;
2310
2311         if (strncmp(buffer, CDT_ENABLE_CMD, strlen(CDT_ENABLE_CMD)) == 0) {
2312                 if (cdt->cdt_state == CDT_DISABLE) {
2313                         rc = set_cdt_state(cdt, CDT_RUNNING);
2314                         mdt_hsm_cdt_event(cdt);
2315                         wake_up(&cdt->cdt_waitq);
2316                 } else if (cdt->cdt_state == CDT_RUNNING) {
2317                         rc = 0;
2318                 } else {
2319                         rc = mdt_hsm_cdt_start(mdt);
2320                 }
2321         } else if (strncmp(buffer, CDT_STOP_CMD, strlen(CDT_STOP_CMD)) == 0) {
2322                 if (cdt->cdt_state == CDT_STOPPING) {
2323                         CERROR("%s: Coordinator is already stopping\n",
2324                                mdt_obd_name(mdt));
2325                         rc = -EALREADY;
2326                 } else if (cdt->cdt_state == CDT_STOPPED) {
2327                         rc = 0;
2328                 } else {
2329                         rc = mdt_hsm_cdt_stop(mdt);
2330                 }
2331         } else if (strncmp(buffer, CDT_DISABLE_CMD,
2332                            strlen(CDT_DISABLE_CMD)) == 0) {
2333                 if ((cdt->cdt_state == CDT_STOPPING) ||
2334                     (cdt->cdt_state == CDT_STOPPED)) {
2335                         /* exit gracefully if coordinator is being stopped
2336                          * or stopped already.
2337                          */
2338                         rc = 0;
2339                 } else {
2340                         rc = set_cdt_state(cdt, CDT_DISABLE);
2341                 }
2342         } else if (strncmp(buffer, CDT_PURGE_CMD,
2343                            strlen(CDT_PURGE_CMD)) == 0) {
2344                 rc = hsm_cancel_all_actions(mdt);
2345         } else if (strncmp(buffer, CDT_HELP_CMD,
2346                            strlen(CDT_HELP_CMD)) == 0) {
2347                 usage = 1;
2348         } else {
2349                 usage = 1;
2350                 rc = -EINVAL;
2351         }
2352
2353         if (usage == 1)
2354                 CERROR("%s: Valid coordinator control commands are: "
2355                        "%s %s %s %s %s\n", mdt_obd_name(mdt),
2356                        CDT_ENABLE_CMD, CDT_STOP_CMD, CDT_DISABLE_CMD,
2357                        CDT_PURGE_CMD, CDT_HELP_CMD);
2358
2359         if (rc)
2360                 RETURN(rc);
2361
2362         RETURN(count);
2363 }
2364
2365 ssize_t hsm_control_show(struct kobject *kobj, struct attribute *attr,
2366                          char *buf)
2367 {
2368         struct obd_device *obd = container_of(kobj, struct obd_device,
2369                                               obd_kset.kobj);
2370         struct coordinator *cdt;
2371
2372         cdt = &(mdt_dev(obd->obd_lu_dev)->mdt_coordinator);
2373
2374         return scnprintf(buf, PAGE_SIZE, "%s\n",
2375                          cdt_mdt_state2str(cdt->cdt_state));
2376 }
2377
2378 static int
2379 mdt_hsm_request_mask_show(struct seq_file *m, __u64 mask)
2380 {
2381         bool first = true;
2382         int i;
2383         ENTRY;
2384
2385         for (i = 0; i < 8 * sizeof(mask); i++) {
2386                 if (mask & (1UL << i)) {
2387                         seq_printf(m, "%s%s", first ? "" : " ",
2388                                    hsm_copytool_action2name(i));
2389                         first = false;
2390                 }
2391         }
2392         seq_putc(m, '\n');
2393
2394         RETURN(0);
2395 }
2396
2397 static int
2398 mdt_hsm_user_request_mask_seq_show(struct seq_file *m, void *data)
2399 {
2400         struct mdt_device *mdt = m->private;
2401         struct coordinator *cdt = &mdt->mdt_coordinator;
2402
2403         return mdt_hsm_request_mask_show(m, cdt->cdt_user_request_mask);
2404 }
2405
2406 static int
2407 mdt_hsm_group_request_mask_seq_show(struct seq_file *m, void *data)
2408 {
2409         struct mdt_device *mdt = m->private;
2410         struct coordinator *cdt = &mdt->mdt_coordinator;
2411
2412         return mdt_hsm_request_mask_show(m, cdt->cdt_group_request_mask);
2413 }
2414
2415 static int
2416 mdt_hsm_other_request_mask_seq_show(struct seq_file *m, void *data)
2417 {
2418         struct mdt_device *mdt = m->private;
2419         struct coordinator *cdt = &mdt->mdt_coordinator;
2420
2421         return mdt_hsm_request_mask_show(m, cdt->cdt_other_request_mask);
2422 }
2423
2424 static inline enum hsm_copytool_action
2425 hsm_copytool_name2action(const char *name)
2426 {
2427         if (strcasecmp(name, "NOOP") == 0)
2428                 return HSMA_NONE;
2429         else if (strcasecmp(name, "ARCHIVE") == 0)
2430                 return HSMA_ARCHIVE;
2431         else if (strcasecmp(name, "RESTORE") == 0)
2432                 return HSMA_RESTORE;
2433         else if (strcasecmp(name, "REMOVE") == 0)
2434                 return HSMA_REMOVE;
2435         else if (strcasecmp(name, "CANCEL") == 0)
2436                 return HSMA_CANCEL;
2437         else
2438                 return -1;
2439 }
2440
2441 static ssize_t
2442 mdt_write_hsm_request_mask(struct file *file, const char __user *user_buf,
2443                             size_t user_count, __u64 *mask)
2444 {
2445         char *buf, *pos, *name;
2446         size_t buf_size;
2447         __u64 new_mask = 0;
2448         int rc;
2449         ENTRY;
2450
2451         if (!(user_count < 4096))
2452                 RETURN(-ENOMEM);
2453
2454         buf_size = user_count + 1;
2455
2456         OBD_ALLOC(buf, buf_size);
2457         if (buf == NULL)
2458                 RETURN(-ENOMEM);
2459
2460         if (copy_from_user(buf, user_buf, buf_size - 1))
2461                 GOTO(out, rc = -EFAULT);
2462
2463         buf[buf_size - 1] = '\0';
2464
2465         pos = buf;
2466         while ((name = strsep(&pos, " \t\v\n")) != NULL) {
2467                 int action;
2468
2469                 if (*name == '\0')
2470                         continue;
2471
2472                 action = hsm_copytool_name2action(name);
2473                 if (action < 0)
2474                         GOTO(out, rc = -EINVAL);
2475
2476                 new_mask |= (1UL << action);
2477         }
2478
2479         *mask = new_mask;
2480         rc = user_count;
2481 out:
2482         OBD_FREE(buf, buf_size);
2483
2484         RETURN(rc);
2485 }
2486
2487 static ssize_t
2488 mdt_hsm_user_request_mask_seq_write(struct file *file, const char __user *buf,
2489                                         size_t count, loff_t *off)
2490 {
2491         struct seq_file         *m = file->private_data;
2492         struct mdt_device       *mdt = m->private;
2493         struct coordinator *cdt = &mdt->mdt_coordinator;
2494
2495         return mdt_write_hsm_request_mask(file, buf, count,
2496                                            &cdt->cdt_user_request_mask);
2497 }
2498
2499 static ssize_t
2500 mdt_hsm_group_request_mask_seq_write(struct file *file, const char __user *buf,
2501                                         size_t count, loff_t *off)
2502 {
2503         struct seq_file         *m = file->private_data;
2504         struct mdt_device       *mdt = m->private;
2505         struct coordinator      *cdt = &mdt->mdt_coordinator;
2506
2507         return mdt_write_hsm_request_mask(file, buf, count,
2508                                            &cdt->cdt_group_request_mask);
2509 }
2510
2511 static ssize_t
2512 mdt_hsm_other_request_mask_seq_write(struct file *file, const char __user *buf,
2513                                         size_t count, loff_t *off)
2514 {
2515         struct seq_file         *m = file->private_data;
2516         struct mdt_device       *mdt = m->private;
2517         struct coordinator      *cdt = &mdt->mdt_coordinator;
2518
2519         return mdt_write_hsm_request_mask(file, buf, count,
2520                                            &cdt->cdt_other_request_mask);
2521 }
2522
2523 static ssize_t remove_archive_on_last_unlink_show(struct kobject *kobj,
2524                                                   struct attribute *attr,
2525                                                   char *buf)
2526 {
2527         struct coordinator *cdt = container_of(kobj, struct coordinator,
2528                                                cdt_hsm_kobj);
2529
2530         return scnprintf(buf, PAGE_SIZE, "%u\n",
2531                          cdt->cdt_remove_archive_on_last_unlink);
2532 }
2533
2534 static ssize_t remove_archive_on_last_unlink_store(struct kobject *kobj,
2535                                                    struct attribute *attr,
2536                                                    const char *buffer,
2537                                                    size_t count)
2538 {
2539         struct coordinator *cdt = container_of(kobj, struct coordinator,
2540                                                cdt_hsm_kobj);
2541         bool val;
2542         int rc;
2543
2544         rc = kstrtobool(buffer, &val);
2545         if (rc < 0)
2546                 return rc;
2547
2548         cdt->cdt_remove_archive_on_last_unlink = val;
2549         return count;
2550 }
2551 LUSTRE_RW_ATTR(remove_archive_on_last_unlink);
2552
2553 LDEBUGFS_SEQ_FOPS(mdt_hsm_user_request_mask);
2554 LDEBUGFS_SEQ_FOPS(mdt_hsm_group_request_mask);
2555 LDEBUGFS_SEQ_FOPS(mdt_hsm_other_request_mask);
2556
2557 /* Read-only sysfs files for request counters */
2558 static ssize_t archive_count_show(struct kobject *kobj, struct attribute *attr,
2559                                   char *buf)
2560 {
2561         struct coordinator *cdt = container_of(kobj, struct coordinator,
2562                                                cdt_hsm_kobj);
2563
2564         return scnprintf(buf, PAGE_SIZE, "%d\n",
2565                          atomic_read(&cdt->cdt_archive_count));
2566 }
2567 LUSTRE_RO_ATTR(archive_count);
2568
2569 static ssize_t restore_count_show(struct kobject *kobj, struct attribute *attr,
2570                                   char *buf)
2571 {
2572         struct coordinator *cdt = container_of(kobj, struct coordinator,
2573                                                cdt_hsm_kobj);
2574
2575         return scnprintf(buf, PAGE_SIZE, "%d\n",
2576                          atomic_read(&cdt->cdt_restore_count));
2577 }
2578 LUSTRE_RO_ATTR(restore_count);
2579
2580 static ssize_t remove_count_show(struct kobject *kobj, struct attribute *attr,
2581                                  char *buf)
2582 {
2583         struct coordinator *cdt = container_of(kobj, struct coordinator,
2584                                                cdt_hsm_kobj);
2585
2586         return scnprintf(buf, PAGE_SIZE, "%d\n",
2587                          atomic_read(&cdt->cdt_remove_count));
2588 }
2589 LUSTRE_RO_ATTR(remove_count);
2590
2591 static struct ldebugfs_vars ldebugfs_mdt_hsm_vars[] = {
2592         { .name =       "agents",
2593           .fops =       &mdt_hsm_agent_fops                     },
2594         { .name =       "actions",
2595           .fops =       &mdt_hsm_actions_fops,
2596           .proc_mode =  0444                                    },
2597         { .name =       "policy",
2598           .fops =       &mdt_hsm_policy_fops                    },
2599         { .name =       "active_requests",
2600           .fops =       &mdt_hsm_active_requests_fops           },
2601         { .name =       "user_request_mask",
2602           .fops =       &mdt_hsm_user_request_mask_fops,        },
2603         { .name =       "group_request_mask",
2604           .fops =       &mdt_hsm_group_request_mask_fops,       },
2605         { .name =       "other_request_mask",
2606           .fops =       &mdt_hsm_other_request_mask_fops,       },
2607         { 0 }
2608 };
2609
2610 static struct attribute *hsm_attrs[] = {
2611         &lustre_attr_loop_period.attr,
2612         &lustre_attr_grace_delay.attr,
2613         &lustre_attr_active_request_timeout.attr,
2614         &lustre_attr_max_requests.attr,
2615         &lustre_attr_default_archive_id.attr,
2616         &lustre_attr_remove_archive_on_last_unlink.attr,
2617         &lustre_attr_archive_count.attr,
2618         &lustre_attr_restore_count.attr,
2619         &lustre_attr_remove_count.attr,
2620         NULL,
2621 };
2622
2623 KOBJ_ATTRIBUTE_GROUPS(hsm); /* creates hsm_groups from hsm_attrs */
2624
2625 static void hsm_kobj_release(struct kobject *kobj)
2626 {
2627         struct coordinator *cdt = container_of(kobj, struct coordinator,
2628                                                cdt_hsm_kobj);
2629
2630         debugfs_remove_recursive(cdt->cdt_debugfs_dir);
2631         cdt->cdt_debugfs_dir = NULL;
2632
2633         complete(&cdt->cdt_kobj_unregister);
2634 }
2635
2636 static struct kobj_type hsm_ktype = {
2637         .default_groups = KOBJ_ATTR_GROUPS(hsm),
2638         .sysfs_ops      = &lustre_sysfs_ops,
2639         .release        = hsm_kobj_release,
2640 };
2641
2642 /**
2643  * create sysfs entries for coordinator
2644  * \param mdt [IN]
2645  * \retval 0 success
2646  * \retval -ve failure
2647  */
2648 int hsm_cdt_tunables_init(struct mdt_device *mdt)
2649 {
2650         struct coordinator *cdt = &mdt->mdt_coordinator;
2651         struct obd_device *obd = mdt2obd_dev(mdt);
2652         int rc;
2653
2654         init_completion(&cdt->cdt_kobj_unregister);
2655         rc = kobject_init_and_add(&cdt->cdt_hsm_kobj, &hsm_ktype,
2656                                   &obd->obd_kset.kobj, "%s", "hsm");
2657         if (rc) {
2658                 kobject_put(&cdt->cdt_hsm_kobj);
2659                 return rc;
2660         }
2661
2662         /* init debugfs entries, failure is not critical */
2663         cdt->cdt_debugfs_dir = debugfs_create_dir("hsm",
2664                                                   obd->obd_debugfs_entry);
2665         ldebugfs_add_vars(cdt->cdt_debugfs_dir, ldebugfs_mdt_hsm_vars, mdt);
2666
2667         return 0;
2668 }
2669
2670 /**
2671  * remove sysfs entries for coordinator
2672  *
2673  * @mdt
2674  */
2675 void hsm_cdt_tunables_fini(struct mdt_device *mdt)
2676 {
2677         struct coordinator *cdt = &mdt->mdt_coordinator;
2678
2679         kobject_put(&cdt->cdt_hsm_kobj);
2680         wait_for_completion(&cdt->cdt_kobj_unregister);
2681 }