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