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