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,
 577                                       mdt_coordinator_cb, &hsd);
 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);
 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         mutex_init(&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);
1705
1706 out_cdt_state:
1707         /* Enable coordinator, unless the coordinator was stopping. */
1708         set_cdt_state(cdt, old_state, NULL);
1709         lu_context_exit(&session);
1710         lu_context_fini(&session);
1711 out_env:
1712         lu_env_fini(&env);
1713
1714         RETURN(rc);
1715 }
1716
1717 /**
1718  * check if a request is compatible with file status
1719  * \param hai [IN] request description
1720  * \param archive_id [IN] request archive id
1721  * \param rq_flags [IN] request flags
1722  * \param hsm [IN] file HSM metadata
1723  * \retval boolean
1724  */
1725 bool mdt_hsm_is_action_compat(const struct hsm_action_item *hai,
1726                               u32 archive_id, u64 rq_flags,
1727                               const struct md_hsm *hsm)
1728 {
1729         int      is_compat = false;
1730         int      hsm_flags;
1731         ENTRY;
1732
1733         hsm_flags = hsm->mh_flags;
1734         switch (hai->hai_action) {
1735         case HSMA_ARCHIVE:
1736                 if (!(hsm_flags & HS_NOARCHIVE) &&
1737                     (hsm_flags & HS_DIRTY || !(hsm_flags & HS_ARCHIVED)))
1738                         is_compat = true;
1739
1740                 if (hsm_flags & HS_EXISTS &&
1741                     archive_id != 0 &&
1742                     archive_id != hsm->mh_arch_id)
1743                         is_compat = false;
1744
1745                 break;
1746         case HSMA_RESTORE:
1747                 if (!(hsm_flags & HS_DIRTY) && (hsm_flags & HS_RELEASED) &&
1748                     hsm_flags & HS_ARCHIVED && !(hsm_flags & HS_LOST))
1749                         is_compat = true;
1750                 break;
1751         case HSMA_REMOVE:
1752                 if (!(hsm_flags & HS_RELEASED) &&
1753                     (hsm_flags & (HS_ARCHIVED | HS_EXISTS)))
1754                         is_compat = true;
1755                 break;
1756         case HSMA_CANCEL:
1757                 is_compat = true;
1758                 break;
1759         }
1760         CDEBUG(D_HSM, "fid="DFID" action=%s flags=%#llx"
1761                       " extent=%#llx-%#llx hsm_flags=%.8X %s\n",
1762                       PFID(&hai->hai_fid),
1763                       hsm_copytool_action2name(hai->hai_action), rq_flags,
1764                       hai->hai_extent.offset, hai->hai_extent.length,
1765                       hsm->mh_flags,
1766                       (is_compat ? "compatible" : "uncompatible"));
1767
1768         RETURN(is_compat);
1769 }
1770
1771 /*
1772  * /proc interface used to get/set HSM behaviour (cdt->cdt_policy)
1773  */
1774 static const struct {
1775         __u64            bit;
1776         char            *name;
1777         char            *nickname;
1778 } hsm_policy_names[] = {
1779         { CDT_NONBLOCKING_RESTORE,      "NonBlockingRestore",   "NBR"},
1780         { CDT_NORETRY_ACTION,           "NoRetryAction",        "NRA"},
1781         { 0 },
1782 };
1783
1784 /**
1785  * convert a policy name to a bit
1786  * \param name [IN] policy name
1787  * \retval 0 unknown
1788  * \retval   policy bit
1789  */
1790 static __u64 hsm_policy_str2bit(const char *name)
1791 {
1792         int      i;
1793
1794         for (i = 0; hsm_policy_names[i].bit != 0; i++)
1795                 if (strcmp(hsm_policy_names[i].nickname, name) == 0 ||
1796                     strcmp(hsm_policy_names[i].name, name) == 0)
1797                         return hsm_policy_names[i].bit;
1798         return 0;
1799 }
1800
1801 /**
1802  * convert a policy bit field to a string
1803  * \param mask [IN] policy bit field
1804  * \param hexa [IN] print mask before bit names
1805  * \param buffer [OUT] string
1806  * \param count [IN] size of buffer
1807  */
1808 static void hsm_policy_bit2str(struct seq_file *m, const __u64 mask,
1809                                 const bool hexa)
1810 {
1811         int      i, j;
1812         __u64    bit;
1813         ENTRY;
1814
1815         if (hexa)
1816                 seq_printf(m, "(%#llx) ", mask);
1817
1818         for (i = 0; i < CDT_POLICY_SHIFT_COUNT; i++) {
1819                 bit = (1ULL << i);
1820
1821                 for (j = 0; hsm_policy_names[j].bit != 0; j++) {
1822                         if (hsm_policy_names[j].bit == bit)
1823                                 break;
1824                 }
1825                 if (bit & mask)
1826                         seq_printf(m, "[%s] ", hsm_policy_names[j].name);
1827                 else
1828                         seq_printf(m, "%s ", hsm_policy_names[j].name);
1829         }
1830         /* remove last ' ' */
1831         m->count--;
1832         seq_putc(m, '\n');
1833 }
1834
1835 /* methods to read/write HSM policy flags */
1836 static int mdt_hsm_policy_seq_show(struct seq_file *m, void *data)
1837 {
1838         struct mdt_device       *mdt = m->private;
1839         struct coordinator      *cdt = &mdt->mdt_coordinator;
1840         ENTRY;
1841
1842         hsm_policy_bit2str(m, cdt->cdt_policy, false);
1843         RETURN(0);
1844 }
1845
1846 static ssize_t
1847 mdt_hsm_policy_seq_write(struct file *file, const char __user *buffer,
1848                          size_t count, loff_t *off)
1849 {
1850         struct seq_file         *m = file->private_data;
1851         struct mdt_device       *mdt = m->private;
1852         struct coordinator      *cdt = &mdt->mdt_coordinator;
1853         char                    *start, *token, sign;
1854         char                    *buf;
1855         __u64                    policy;
1856         __u64                    add_mask, remove_mask, set_mask;
1857         int                      rc;
1858         ENTRY;
1859
1860         if (count + 1 > PAGE_SIZE)
1861                 RETURN(-EINVAL);
1862
1863         OBD_ALLOC(buf, count + 1);
1864         if (buf == NULL)
1865                 RETURN(-ENOMEM);
1866
1867         if (copy_from_user(buf, buffer, count))
1868                 GOTO(out, rc = -EFAULT);
1869
1870         buf[count] = '\0';
1871
1872         start = buf;
1873         CDEBUG(D_HSM, "%s: receive new policy: '%s'\n", mdt_obd_name(mdt),
1874                start);
1875
1876         add_mask = remove_mask = set_mask = 0;
1877         do {
1878                 token = strsep(&start, "\n ");
1879                 sign = *token;
1880
1881                 if (sign == '\0')
1882                         continue;
1883
1884                 if (sign == '-' || sign == '+')
1885                         token++;
1886
1887                 policy = hsm_policy_str2bit(token);
1888                 if (policy == 0) {
1889                         CWARN("%s: '%s' is unknown, "
1890                               "supported policies are:\n", mdt_obd_name(mdt),
1891                               token);
1892                         hsm_policy_bit2str(m, 0, false);
1893                         GOTO(out, rc = -EINVAL);
1894                 }
1895                 switch (sign) {
1896                 case '-':
1897                         remove_mask |= policy;
1898                         break;
1899                 case '+':
1900                         add_mask |= policy;
1901                         break;
1902                 default:
1903                         set_mask |= policy;
1904                         break;
1905                 }
1906
1907         } while (start != NULL);
1908
1909         CDEBUG(D_HSM, "%s: new policy: rm=%#llx add=%#llx set=%#llx\n",
1910                mdt_obd_name(mdt), remove_mask, add_mask, set_mask);
1911
1912         /* if no sign in all string, it is a clear and set
1913          * if some sign found, all unsigned are converted
1914          * to add
1915          * P1 P2 = set to P1 and P2
1916          * P1 -P2 = add P1 clear P2 same as +P1 -P2
1917          */
1918         if (remove_mask == 0 && add_mask == 0) {
1919                 cdt->cdt_policy = set_mask;
1920         } else {
1921                 cdt->cdt_policy |= set_mask | add_mask;
1922                 cdt->cdt_policy &= ~remove_mask;
1923         }
1924
1925         GOTO(out, rc = count);
1926
1927 out:
1928         OBD_FREE(buf, count + 1);
1929         RETURN(rc);
1930 }
1931 LPROC_SEQ_FOPS(mdt_hsm_policy);
1932
1933 #define GENERATE_PROC_METHOD(VAR)                                       \
1934 static int mdt_hsm_##VAR##_seq_show(struct seq_file *m, void *data)     \
1935 {                                                                       \
1936         struct mdt_device       *mdt = m->private;                      \
1937         struct coordinator      *cdt = &mdt->mdt_coordinator;           \
1938         ENTRY;                                                          \
1939                                                                         \
1940         seq_printf(m, "%llu\n", (__u64)cdt->VAR);                       \
1941         RETURN(0);                                                      \
1942 }                                                                       \
1943 static ssize_t                                                          \
1944 mdt_hsm_##VAR##_seq_write(struct file *file, const char __user *buffer, \
1945                           size_t count, loff_t *off)                    \
1946                                                                         \
1947 {                                                                       \
1948         struct seq_file         *m = file->private_data;                \
1949         struct mdt_device       *mdt = m->private;                      \
1950         struct coordinator      *cdt = &mdt->mdt_coordinator;           \
1951         __s64                    val;                                   \
1952         int                      rc;                                    \
1953         ENTRY;                                                          \
1954                                                                         \
1955         rc = lprocfs_str_to_s64(buffer, count, &val);                   \
1956         if (rc)                                                         \
1957                 RETURN(rc);                                             \
1958         if (val > 0 && val < INT_MAX) {                                 \
1959                 cdt->VAR = val;                                         \
1960                 RETURN(count);                                          \
1961         }                                                               \
1962         RETURN(-EINVAL);                                                \
1963 }                                                                       \
1964
1965 GENERATE_PROC_METHOD(cdt_loop_period)
1966 GENERATE_PROC_METHOD(cdt_grace_delay)
1967 GENERATE_PROC_METHOD(cdt_active_req_timeout)
1968 GENERATE_PROC_METHOD(cdt_max_requests)
1969 GENERATE_PROC_METHOD(cdt_default_archive_id)
1970
1971 /*
1972  * procfs write method for MDT/hsm_control
1973  * proc entry is in mdt directory so data is mdt obd_device pointer
1974  */
1975 #define CDT_ENABLE_CMD   "enabled"
1976 #define CDT_STOP_CMD     "shutdown"
1977 #define CDT_DISABLE_CMD  "disabled"
1978 #define CDT_PURGE_CMD    "purge"
1979 #define CDT_HELP_CMD     "help"
1980 #define CDT_MAX_CMD_LEN  10
1981
1982 ssize_t
1983 mdt_hsm_cdt_control_seq_write(struct file *file, const char __user *buffer,
1984                               size_t count, loff_t *off)
1985 {
1986         struct seq_file         *m = file->private_data;
1987         struct obd_device       *obd = m->private;
1988         struct mdt_device       *mdt = mdt_dev(obd->obd_lu_dev);
1989         struct coordinator      *cdt = &(mdt->mdt_coordinator);
1990         int                      rc, usage = 0;
1991         char                     kernbuf[CDT_MAX_CMD_LEN];
1992         ENTRY;
1993
1994         if (count == 0 || count >= sizeof(kernbuf))
1995                 RETURN(-EINVAL);
1996
1997         if (copy_from_user(kernbuf, buffer, count))
1998                 RETURN(-EFAULT);
1999         kernbuf[count] = 0;
2000
2001         if (kernbuf[count - 1] == '\n')
2002                 kernbuf[count - 1] = 0;
2003
2004         rc = 0;
2005         if (strcmp(kernbuf, CDT_ENABLE_CMD) == 0) {
2006                 if (cdt->cdt_state == CDT_DISABLE) {
2007                         rc = set_cdt_state(cdt, CDT_RUNNING, NULL);
2008                         mdt_hsm_cdt_wakeup(mdt);
2009                 } else {
2010                         rc = mdt_hsm_cdt_start(mdt);
2011                 }
2012         } else if (strcmp(kernbuf, CDT_STOP_CMD) == 0) {
2013                 if ((cdt->cdt_state == CDT_STOPPING) ||
2014                     (cdt->cdt_state == CDT_STOPPED)) {
2015                         CERROR("%s: Coordinator already stopped\n",
2016                                mdt_obd_name(mdt));
2017                         rc = -EALREADY;
2018                 } else {
2019                         rc = mdt_hsm_cdt_stop(mdt);
2020                         mdt_hsm_cdt_wakeup(mdt);
2021                 }
2022         } else if (strcmp(kernbuf, CDT_DISABLE_CMD) == 0) {
2023                 if ((cdt->cdt_state == CDT_STOPPING) ||
2024                     (cdt->cdt_state == CDT_STOPPED)) {
2025                         CERROR("%s: Coordinator is stopped\n",
2026                                mdt_obd_name(mdt));
2027                         rc = -EINVAL;
2028                 } else {
2029                         rc = set_cdt_state(cdt, CDT_DISABLE, NULL);
2030                 }
2031         } else if (strcmp(kernbuf, CDT_PURGE_CMD) == 0) {
2032                 rc = hsm_cancel_all_actions(mdt);
2033         } else if (strcmp(kernbuf, CDT_HELP_CMD) == 0) {
2034                 usage = 1;
2035         } else {
2036                 usage = 1;
2037                 rc = -EINVAL;
2038         }
2039
2040         if (usage == 1)
2041                 CERROR("%s: Valid coordinator control commands are: "
2042                        "%s %s %s %s %s\n", mdt_obd_name(mdt),
2043                        CDT_ENABLE_CMD, CDT_STOP_CMD, CDT_DISABLE_CMD,
2044                        CDT_PURGE_CMD, CDT_HELP_CMD);
2045
2046         if (rc)
2047                 RETURN(rc);
2048
2049         RETURN(count);
2050 }
2051
2052 int mdt_hsm_cdt_control_seq_show(struct seq_file *m, void *data)
2053 {
2054         struct obd_device       *obd = m->private;
2055         struct coordinator      *cdt;
2056         ENTRY;
2057
2058         cdt = &(mdt_dev(obd->obd_lu_dev)->mdt_coordinator);
2059
2060         seq_printf(m, "%s\n", cdt_mdt_state2str(cdt->cdt_state));
2061
2062         RETURN(0);
2063 }
2064
2065 static int
2066 mdt_hsm_request_mask_show(struct seq_file *m, __u64 mask)
2067 {
2068         bool first = true;
2069         int i;
2070         ENTRY;
2071
2072         for (i = 0; i < 8 * sizeof(mask); i++) {
2073                 if (mask & (1UL << i)) {
2074                         seq_printf(m, "%s%s", first ? "" : " ",
2075                                    hsm_copytool_action2name(i));
2076                         first = false;
2077                 }
2078         }
2079         seq_putc(m, '\n');
2080
2081         RETURN(0);
2082 }
2083
2084 static int
2085 mdt_hsm_user_request_mask_seq_show(struct seq_file *m, void *data)
2086 {
2087         struct mdt_device *mdt = m->private;
2088         struct coordinator *cdt = &mdt->mdt_coordinator;
2089
2090         return mdt_hsm_request_mask_show(m, cdt->cdt_user_request_mask);
2091 }
2092
2093 static int
2094 mdt_hsm_group_request_mask_seq_show(struct seq_file *m, void *data)
2095 {
2096         struct mdt_device *mdt = m->private;
2097         struct coordinator *cdt = &mdt->mdt_coordinator;
2098
2099         return mdt_hsm_request_mask_show(m, cdt->cdt_group_request_mask);
2100 }
2101
2102 static int
2103 mdt_hsm_other_request_mask_seq_show(struct seq_file *m, void *data)
2104 {
2105         struct mdt_device *mdt = m->private;
2106         struct coordinator *cdt = &mdt->mdt_coordinator;
2107
2108         return mdt_hsm_request_mask_show(m, cdt->cdt_other_request_mask);
2109 }
2110
2111 static inline enum hsm_copytool_action
2112 hsm_copytool_name2action(const char *name)
2113 {
2114         if (strcasecmp(name, "NOOP") == 0)
2115                 return HSMA_NONE;
2116         else if (strcasecmp(name, "ARCHIVE") == 0)
2117                 return HSMA_ARCHIVE;
2118         else if (strcasecmp(name, "RESTORE") == 0)
2119                 return HSMA_RESTORE;
2120         else if (strcasecmp(name, "REMOVE") == 0)
2121                 return HSMA_REMOVE;
2122         else if (strcasecmp(name, "CANCEL") == 0)
2123                 return HSMA_CANCEL;
2124         else
2125                 return -1;
2126 }
2127
2128 static ssize_t
2129 mdt_write_hsm_request_mask(struct file *file, const char __user *user_buf,
2130                             size_t user_count, __u64 *mask)
2131 {
2132         char *buf, *pos, *name;
2133         size_t buf_size;
2134         __u64 new_mask = 0;
2135         int rc;
2136         ENTRY;
2137
2138         if (!(user_count < 4096))
2139                 RETURN(-ENOMEM);
2140
2141         buf_size = user_count + 1;
2142
2143         OBD_ALLOC(buf, buf_size);
2144         if (buf == NULL)
2145                 RETURN(-ENOMEM);
2146
2147         if (copy_from_user(buf, user_buf, buf_size - 1))
2148                 GOTO(out, rc = -EFAULT);
2149
2150         buf[buf_size - 1] = '\0';
2151
2152         pos = buf;
2153         while ((name = strsep(&pos, " \t\v\n")) != NULL) {
2154                 int action;
2155
2156                 if (*name == '\0')
2157                         continue;
2158
2159                 action = hsm_copytool_name2action(name);
2160                 if (action < 0)
2161                         GOTO(out, rc = -EINVAL);
2162
2163                 new_mask |= (1UL << action);
2164         }
2165
2166         *mask = new_mask;
2167         rc = user_count;
2168 out:
2169         OBD_FREE(buf, buf_size);
2170
2171         RETURN(rc);
2172 }
2173
2174 static ssize_t
2175 mdt_hsm_user_request_mask_seq_write(struct file *file, const char __user *buf,
2176                                         size_t count, loff_t *off)
2177 {
2178         struct seq_file         *m = file->private_data;
2179         struct mdt_device       *mdt = m->private;
2180         struct coordinator *cdt = &mdt->mdt_coordinator;
2181
2182         return mdt_write_hsm_request_mask(file, buf, count,
2183                                            &cdt->cdt_user_request_mask);
2184 }
2185
2186 static ssize_t
2187 mdt_hsm_group_request_mask_seq_write(struct file *file, const char __user *buf,
2188                                         size_t count, loff_t *off)
2189 {
2190         struct seq_file         *m = file->private_data;
2191         struct mdt_device       *mdt = m->private;
2192         struct coordinator      *cdt = &mdt->mdt_coordinator;
2193
2194         return mdt_write_hsm_request_mask(file, buf, count,
2195                                            &cdt->cdt_group_request_mask);
2196 }
2197
2198 static ssize_t
2199 mdt_hsm_other_request_mask_seq_write(struct file *file, const char __user *buf,
2200                                         size_t count, loff_t *off)
2201 {
2202         struct seq_file         *m = file->private_data;
2203         struct mdt_device       *mdt = m->private;
2204         struct coordinator      *cdt = &mdt->mdt_coordinator;
2205
2206         return mdt_write_hsm_request_mask(file, buf, count,
2207                                            &cdt->cdt_other_request_mask);
2208 }
2209
2210 LPROC_SEQ_FOPS(mdt_hsm_cdt_loop_period);
2211 LPROC_SEQ_FOPS(mdt_hsm_cdt_grace_delay);
2212 LPROC_SEQ_FOPS(mdt_hsm_cdt_active_req_timeout);
2213 LPROC_SEQ_FOPS(mdt_hsm_cdt_max_requests);
2214 LPROC_SEQ_FOPS(mdt_hsm_cdt_default_archive_id);
2215 LPROC_SEQ_FOPS(mdt_hsm_user_request_mask);
2216 LPROC_SEQ_FOPS(mdt_hsm_group_request_mask);
2217 LPROC_SEQ_FOPS(mdt_hsm_other_request_mask);
2218
2219 static struct lprocfs_vars lprocfs_mdt_hsm_vars[] = {
2220         { .name =       "agents",
2221           .fops =       &mdt_hsm_agent_fops                     },
2222         { .name =       "actions",
2223           .fops =       &mdt_hsm_actions_fops,
2224           .proc_mode =  0444                                    },
2225         { .name =       "default_archive_id",
2226           .fops =       &mdt_hsm_cdt_default_archive_id_fops    },
2227         { .name =       "grace_delay",
2228           .fops =       &mdt_hsm_cdt_grace_delay_fops           },
2229         { .name =       "loop_period",
2230           .fops =       &mdt_hsm_cdt_loop_period_fops           },
2231         { .name =       "max_requests",
2232           .fops =       &mdt_hsm_cdt_max_requests_fops          },
2233         { .name =       "policy",
2234           .fops =       &mdt_hsm_policy_fops                    },
2235         { .name =       "active_request_timeout",
2236           .fops =       &mdt_hsm_cdt_active_req_timeout_fops    },
2237         { .name =       "active_requests",
2238           .fops =       &mdt_hsm_active_requests_fops           },
2239         { .name =       "user_request_mask",
2240           .fops =       &mdt_hsm_user_request_mask_fops,        },
2241         { .name =       "group_request_mask",
2242           .fops =       &mdt_hsm_group_request_mask_fops,       },
2243         { .name =       "other_request_mask",
2244           .fops =       &mdt_hsm_other_request_mask_fops,       },
2245         { 0 }
2246 };