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