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