Whamcloud - gitweb
LU-10030 hsm: make changelog flag argument an enum
[fs/lustre-release.git] / lustre / mdt / mdt_coordinator.c
index 55c4cc5..19fd0f5 100644 (file)
@@ -23,7 +23,7 @@
  * Copyright (c) 2011, 2012 Commissariat a l'energie atomique et aux energies
  *                          alternatives
  *
- * Copyright (c) 2013, 2016, Intel Corporation.
+ * Copyright (c) 2013, 2017, Intel Corporation.
  * Use is subject to license terms.
  */
 /*
@@ -101,21 +101,21 @@ void mdt_hsm_dump_hal(int level, const char *prefix,
        struct hsm_action_item  *hai;
        char                     buf[12];
 
-       CDEBUG(level, "%s: HAL header: version %X count %d compound %#llx"
+       CDEBUG(level, "%s: HAL header: version %X count %d"
                      " archive_id %d flags %#llx\n",
               prefix, hal->hal_version, hal->hal_count,
-              hal->hal_compound_id, hal->hal_archive_id, hal->hal_flags);
+              hal->hal_archive_id, hal->hal_flags);
 
        hai = hai_first(hal);
        for (i = 0; i < hal->hal_count; i++) {
                sz = hai->hai_len - sizeof(*hai);
                CDEBUG(level, "%s %d: fid="DFID" dfid="DFID
-                      " compound/cookie=%#llx/%#llx"
+                      " cookie=%#llx"
                       " action=%s extent=%#llx-%#llx gid=%#llx"
                       " datalen=%d data=[%s]\n",
                       prefix, i,
                       PFID(&hai->hai_fid), PFID(&hai->hai_dfid),
-                      hal->hal_compound_id, hai->hai_cookie,
+                      hai->hai_cookie,
                       hsm_copytool_action2name(hai->hai_action),
                       hai->hai_extent.offset,
                       hai->hai_extent.length,
@@ -136,14 +136,262 @@ struct hsm_scan_request {
 };
 
 struct hsm_scan_data {
-       struct mdt_thread_info          *mti;
-       char                             fs_name[MTI_NAME_MAXLEN+1];
-       /* request to be send to agents */
-       int                              max_requests;  /** vector size */
-       int                              request_cnt;   /** used count */
-       struct hsm_scan_request         *request;
+       struct mdt_thread_info  *hsd_mti;
+       char                     hsd_fsname[MTI_NAME_MAXLEN + 1];
+       /* are we scanning the logs for housekeeping, or just looking
+        * for new work?
+        */
+       bool                     hsd_housekeeping;
+       bool                     hsd_one_restore;
+       int                      hsd_action_count;
+       int                      hsd_request_len; /* array alloc len */
+       int                      hsd_request_count; /* array used count */
+       struct hsm_scan_request *hsd_request;
 };
 
+static int mdt_cdt_waiting_cb(const struct lu_env *env,
+                             struct mdt_device *mdt,
+                             struct llog_handle *llh,
+                             struct llog_agent_req_rec *larr,
+                             struct hsm_scan_data *hsd)
+{
+       struct coordinator *cdt = &mdt->mdt_coordinator;
+       struct hsm_scan_request *request;
+       struct hsm_action_item *hai;
+       size_t hai_size;
+       u32 archive_id;
+       int i;
+
+       /* Are agents full? */
+       if (hsd->hsd_action_count + atomic_read(&cdt->cdt_request_count) >=
+           cdt->cdt_max_requests) {
+               /* We cannot send any more request
+                *
+                *                     *** SPECIAL CASE ***
+                *
+                * Restore requests are too important not to schedule at least
+                * one, everytime we can.
+                */
+               if (larr->arr_hai.hai_action != HSMA_RESTORE ||
+                   hsd->hsd_one_restore)
+                       RETURN(hsd->hsd_housekeeping ? 0 : LLOG_PROC_BREAK);
+       }
+
+       hai_size = cfs_size_round(larr->arr_hai.hai_len);
+       archive_id = larr->arr_archive_id;
+
+       /* Can we add this action to one of the existing HALs in hsd. */
+       request = NULL;
+       for (i = 0; i < hsd->hsd_request_count; i++) {
+               if (hsd->hsd_request[i].hal->hal_archive_id == archive_id &&
+                   hsd->hsd_request[i].hal_used_sz + hai_size <=
+                   LDLM_MAXREQSIZE) {
+                       request = &hsd->hsd_request[i];
+                       break;
+               }
+       }
+
+       /* Are we trying to force-schedule a request? */
+       if (hsd->hsd_action_count + atomic_read(&cdt->cdt_request_count) >=
+           cdt->cdt_max_requests) {
+               /* Is there really no compatible hsm_scan_request? */
+               if (!request) {
+                       for (i -= 1; i >= 0; i--) {
+                               if (hsd->hsd_request[i].hal->hal_archive_id ==
+                                   archive_id) {
+                                       request = &hsd->hsd_request[i];
+                                       break;
+                               }
+                       }
+               }
+
+               /* Make room for the hai */
+               if (request) {
+                       /* Discard the last hai until there is enough space */
+                       do {
+                               request->hal->hal_count--;
+
+                               hai = hai_first(request->hal);
+                               for (i = 0; i < request->hal->hal_count; i++)
+                                       hai = hai_next(hai);
+                               request->hal_used_sz -=
+                                       cfs_size_round(hai->hai_len);
+                               hsd->hsd_action_count--;
+                       } while (request->hal_used_sz + hai_size >
+                                LDLM_MAXREQSIZE);
+               } else if (hsd->hsd_housekeeping) {
+                       struct hsm_scan_request *tmp;
+
+                       /* Discard the (whole) last hal */
+                       hsd->hsd_request_count--;
+                       tmp = &hsd->hsd_request[hsd->hsd_request_count];
+                       hsd->hsd_action_count -= tmp->hal->hal_count;
+                       OBD_FREE(tmp->hal, tmp->hal_sz);
+               } else {
+                       /* Bailing out, this code path is too hot */
+                       RETURN(LLOG_PROC_BREAK);
+
+               }
+       }
+
+       if (!request) {
+               struct hsm_action_list *hal;
+
+               LASSERT(hsd->hsd_request_count < hsd->hsd_request_len);
+               request = &hsd->hsd_request[hsd->hsd_request_count];
+
+               /* allocates hai vector size just needs to be large
+                * enough */
+               request->hal_sz = sizeof(*request->hal) +
+                       cfs_size_round(MTI_NAME_MAXLEN + 1) + 2 * hai_size;
+               OBD_ALLOC_LARGE(hal, request->hal_sz);
+               if (!hal)
+                       RETURN(-ENOMEM);
+
+               hal->hal_version = HAL_VERSION;
+               strlcpy(hal->hal_fsname, hsd->hsd_fsname, MTI_NAME_MAXLEN + 1);
+               hal->hal_archive_id = larr->arr_archive_id;
+               hal->hal_flags = larr->arr_flags;
+               hal->hal_count = 0;
+               request->hal_used_sz = hal_size(hal);
+               request->hal = hal;
+               hsd->hsd_request_count++;
+       } else if (request->hal_sz < request->hal_used_sz + hai_size) {
+               /* Not enough room, need an extension */
+               void *hal_buffer;
+               int sz;
+
+               sz = min_t(int, 2 * request->hal_sz, LDLM_MAXREQSIZE);
+               LASSERT(request->hal_used_sz + hai_size < sz);
+
+               OBD_ALLOC_LARGE(hal_buffer, sz);
+               if (!hal_buffer)
+                       RETURN(-ENOMEM);
+
+               memcpy(hal_buffer, request->hal, request->hal_used_sz);
+               OBD_FREE_LARGE(request->hal, request->hal_sz);
+               request->hal = hal_buffer;
+               request->hal_sz = sz;
+       }
+
+       hai = hai_first(request->hal);
+       for (i = 0; i < request->hal->hal_count; i++)
+               hai = hai_next(hai);
+
+       memcpy(hai, &larr->arr_hai, larr->arr_hai.hai_len);
+
+       request->hal_used_sz += hai_size;
+       request->hal->hal_count++;
+
+       hsd->hsd_action_count++;
+
+       switch (hai->hai_action) {
+       case HSMA_CANCEL:
+               break;
+       case HSMA_RESTORE:
+               hsd->hsd_one_restore = true;
+               /* Intentional fallthrough */
+       default:
+               cdt_agent_record_hash_add(cdt, hai->hai_cookie,
+                                         llh->lgh_hdr->llh_cat_idx,
+                                         larr->arr_hdr.lrh_index);
+       }
+
+       RETURN(0);
+}
+
+static int mdt_cdt_started_cb(const struct lu_env *env,
+                             struct mdt_device *mdt,
+                             struct llog_handle *llh,
+                             struct llog_agent_req_rec *larr,
+                             struct hsm_scan_data *hsd)
+{
+       struct coordinator *cdt = &mdt->mdt_coordinator;
+       struct hsm_action_item *hai = &larr->arr_hai;
+       struct cdt_agent_req *car;
+       time64_t now = ktime_get_real_seconds();
+       time64_t last;
+       enum changelog_rec_flags clf_flags;
+       int rc;
+
+       if (!hsd->hsd_housekeeping)
+               RETURN(0);
+
+       /* we search for a running request
+        * error may happen if coordinator crashes or stopped
+        * with running request
+        */
+       car = mdt_cdt_find_request(cdt, hai->hai_cookie);
+       if (car == NULL) {
+               last = larr->arr_req_change;
+       } else {
+               last = car->car_req_update;
+       }
+
+       /* test if request too long, if yes cancel it
+        * the same way the copy tool acknowledge a cancel request */
+       if (now <= last + cdt->cdt_active_req_timeout)
+               GOTO(out_car, rc = 0);
+
+       dump_llog_agent_req_rec("request timed out, start cleaning", larr);
+
+       if (car != NULL) {
+               car->car_req_update = now;
+               mdt_hsm_agent_update_statistics(cdt, 0, 1, 0, &car->car_uuid);
+               /* Remove car from memory list (LU-9075) */
+               mdt_cdt_remove_request(cdt, hai->hai_cookie);
+       }
+
+       /* Emit a changelog record for the failed action.*/
+       clf_flags = 0;
+       hsm_set_cl_error(&clf_flags, ECANCELED);
+
+       switch (hai->hai_action) {
+       case HSMA_ARCHIVE:
+               hsm_set_cl_event(&clf_flags, HE_ARCHIVE);
+               break;
+       case HSMA_RESTORE:
+               hsm_set_cl_event(&clf_flags, HE_RESTORE);
+               break;
+       case HSMA_REMOVE:
+               hsm_set_cl_event(&clf_flags, HE_REMOVE);
+               break;
+       case HSMA_CANCEL:
+               hsm_set_cl_event(&clf_flags, HE_CANCEL);
+               break;
+       default:
+               /* Unknown record type, skip changelog. */
+               clf_flags = 0;
+               break;
+       }
+
+       if (clf_flags != 0)
+               mo_changelog(env, CL_HSM, clf_flags, mdt->mdt_child,
+                            &hai->hai_fid);
+
+       if (hai->hai_action == HSMA_RESTORE)
+               cdt_restore_handle_del(hsd->hsd_mti, cdt, &hai->hai_fid);
+
+       larr->arr_status = ARS_CANCELED;
+       larr->arr_req_change = now;
+       rc = llog_write(hsd->hsd_mti->mti_env, llh, &larr->arr_hdr,
+                       larr->arr_hdr.lrh_index);
+       if (rc < 0) {
+               CERROR("%s: cannot update agent log: rc = %d\n",
+                      mdt_obd_name(mdt), rc);
+               rc = LLOG_DEL_RECORD;
+       }
+
+       /* ct has completed a request, so a slot is available,
+        * signal the coordinator to find new work */
+       mdt_hsm_cdt_event(cdt);
+out_car:
+       if (car != NULL)
+               mdt_cdt_put_request(car);
+
+       RETURN(rc);
+}
+
 /**
  *  llog_cat_process() callback, used to:
  *  - find waiting request and start action
@@ -160,194 +408,32 @@ static int mdt_coordinator_cb(const struct lu_env *env,
                              struct llog_rec_hdr *hdr,
                              void *data)
 {
-       struct llog_agent_req_rec       *larr;
-       struct hsm_scan_data            *hsd;
-       struct hsm_action_item          *hai;
-       struct mdt_device               *mdt;
-       struct coordinator              *cdt;
-       int                              rc;
+       struct llog_agent_req_rec *larr = (struct llog_agent_req_rec *)hdr;
+       struct hsm_scan_data *hsd = data;
+       struct mdt_device *mdt = hsd->hsd_mti->mti_mdt;
+       struct coordinator *cdt = &mdt->mdt_coordinator;
        ENTRY;
 
-       hsd = data;
-       mdt = hsd->mti->mti_mdt;
-       cdt = &mdt->mdt_coordinator;
-
        larr = (struct llog_agent_req_rec *)hdr;
        dump_llog_agent_req_rec("mdt_coordinator_cb(): ", larr);
        switch (larr->arr_status) {
-       case ARS_WAITING: {
-               int i;
-               struct hsm_scan_request *request;
-
-               /* Are agents full? */
-               if (atomic_read(&cdt->cdt_request_count) >=
-                   cdt->cdt_max_requests)
-                       break;
-
-               /* first search whether the request is found in the
-                * list we have built. */
-               request = NULL;
-               for (i = 0; i < hsd->request_cnt; i++) {
-                       if (hsd->request[i].hal->hal_compound_id ==
-                           larr->arr_compound_id) {
-                               request = &hsd->request[i];
-                               break;
-                       }
-               }
-
-               if (!request) {
-                       struct hsm_action_list *hal;
-
-                       if (hsd->request_cnt == hsd->max_requests)
-                               /* Unknown request and no more room
-                                * for a new request. Continue to scan
-                                * to find other entries for already
-                                * existing requests.
-                                */
-                               RETURN(0);
-
-                       request = &hsd->request[hsd->request_cnt];
-
-                       /* allocates hai vector size just needs to be large
-                        * enough */
-                       request->hal_sz =
-                               sizeof(*request->hal) +
-                               cfs_size_round(MTI_NAME_MAXLEN+1) +
-                               2 * cfs_size_round(larr->arr_hai.hai_len);
-                       OBD_ALLOC(hal, request->hal_sz);
-                       if (!hal)
-                               RETURN(-ENOMEM);
-                       hal->hal_version = HAL_VERSION;
-                       strlcpy(hal->hal_fsname, hsd->fs_name,
-                               MTI_NAME_MAXLEN + 1);
-                       hal->hal_compound_id = larr->arr_compound_id;
-                       hal->hal_archive_id = larr->arr_archive_id;
-                       hal->hal_flags = larr->arr_flags;
-                       hal->hal_count = 0;
-                       request->hal_used_sz = hal_size(hal);
-                       request->hal = hal;
-                       hsd->request_cnt++;
-                       hai = hai_first(hal);
-               } else {
-                       /* request is known */
-                       /* we check if record archive num is the same as the
-                        * known request, if not we will serve it in multiple
-                        * time because we do not know if the agent can serve
-                        * multiple backend
-                        * a use case is a compound made of multiple restore
-                        * where the files are not archived in the same backend
-                        */
-                       if (larr->arr_archive_id !=
-                           request->hal->hal_archive_id)
-                               RETURN(0);
-
-                       if (request->hal_sz <
-                           request->hal_used_sz +
-                           cfs_size_round(larr->arr_hai.hai_len)) {
-                               /* Not enough room, need an extension */
-                               void *hal_buffer;
-                               int sz;
-
-                               sz = 2 * request->hal_sz;
-                               OBD_ALLOC(hal_buffer, sz);
-                               if (!hal_buffer)
-                                       RETURN(-ENOMEM);
-                               memcpy(hal_buffer, request->hal,
-                                      request->hal_used_sz);
-                               OBD_FREE(request->hal,
-                                        request->hal_sz);
-                               request->hal = hal_buffer;
-                               request->hal_sz = sz;
-                       }
-                       hai = hai_first(request->hal);
-                       for (i = 0; i < request->hal->hal_count; i++)
-                               hai = hai_next(hai);
-               }
-               memcpy(hai, &larr->arr_hai, larr->arr_hai.hai_len);
-               hai->hai_cookie = larr->arr_hai.hai_cookie;
-               hai->hai_gid = larr->arr_hai.hai_gid;
-
-               request->hal_used_sz += cfs_size_round(hai->hai_len);
-               request->hal->hal_count++;
-               break;
-       }
-       case ARS_STARTED: {
-               struct hsm_progress_kernel pgs;
-               struct cdt_agent_req *car;
-               cfs_time_t now = cfs_time_current_sec();
-               cfs_time_t last;
-
-               /* we search for a running request
-                * error may happen if coordinator crashes or stopped
-                * with running request
-                */
-               car = mdt_cdt_find_request(cdt, larr->arr_hai.hai_cookie);
-               if (car == NULL) {
-                       last = larr->arr_req_change;
-               } else {
-                       last = car->car_req_update;
-                       mdt_cdt_put_request(car);
-               }
-
-               /* test if request too long, if yes cancel it
-                * the same way the copy tool acknowledge a cancel request */
-               if (now <= last + cdt->cdt_active_req_timeout)
+       case ARS_WAITING:
+               RETURN(mdt_cdt_waiting_cb(env, mdt, llh, larr, hsd));
+       case ARS_STARTED:
+               RETURN(mdt_cdt_started_cb(env, mdt, llh, larr, hsd));
+       default:
+               if (!hsd->hsd_housekeeping)
                        RETURN(0);
 
-               dump_llog_agent_req_rec("request timed out, start cleaning",
-                                       larr);
-               /* a too old cancel request just needs to be removed
-                * this can happen, if copy tool does not support
-                * cancel for other requests, we have to remove the
-                * running request and notify the copytool */
-               pgs.hpk_fid = larr->arr_hai.hai_fid;
-               pgs.hpk_cookie = larr->arr_hai.hai_cookie;
-               pgs.hpk_extent = larr->arr_hai.hai_extent;
-               pgs.hpk_flags = HP_FLAG_COMPLETED;
-               pgs.hpk_errval = ENOSYS;
-               pgs.hpk_data_version = 0;
-
-               /* update request state, but do not record in llog, to
-                * avoid deadlock on cdt_llog_lock */
-               rc = mdt_hsm_update_request_state(hsd->mti, &pgs, 0);
-               if (rc)
-                       CERROR("%s: cannot cleanup timed out request: "
-                              DFID" for cookie %#llx action=%s\n",
-                              mdt_obd_name(mdt),
-                              PFID(&pgs.hpk_fid), pgs.hpk_cookie,
-                              hsm_copytool_action2name(
-                                      larr->arr_hai.hai_action));
-
-               if (rc == -ENOENT) {
-                       /* The request no longer exists, forget
-                        * about it, and do not send a cancel request
-                        * to the client, for which an error will be
-                        * sent back, leading to an endless cycle of
-                        * cancellation. */
+               if ((larr->arr_req_change + cdt->cdt_grace_delay) <
+                   ktime_get_real_seconds()) {
+                       cdt_agent_record_hash_del(cdt,
+                                                 larr->arr_hai.hai_cookie);
                        RETURN(LLOG_DEL_RECORD);
                }
 
-               /* XXX A cancel request cannot be cancelled. */
-               if (larr->arr_hai.hai_action == HSMA_CANCEL)
-                       RETURN(0);
-
-               larr->arr_status = ARS_CANCELED;
-               larr->arr_req_change = now;
-               rc = llog_write(hsd->mti->mti_env, llh, hdr, hdr->lrh_index);
-               if (rc < 0)
-                       CERROR("%s: cannot update agent log: rc = %d\n",
-                              mdt_obd_name(mdt), rc);
-               break;
-       }
-       case ARS_FAILED:
-       case ARS_CANCELED:
-       case ARS_SUCCEED:
-               if ((larr->arr_req_change + cdt->cdt_grace_delay) <
-                   cfs_time_current_sec())
-                       RETURN(LLOG_DEL_RECORD);
-               break;
+               RETURN(0);
        }
-       RETURN(0);
 }
 
 /**
@@ -424,13 +510,17 @@ static void mdt_hsm_cdt_cleanup(struct mdt_device *mdt)
        down_write(&cdt->cdt_agent_lock);
        list_for_each_entry_safe(ha, tmp2, &cdt->cdt_agents, ha_list) {
                list_del(&ha->ha_list);
+               if (ha->ha_archive_cnt != 0)
+                       OBD_FREE(ha->ha_archive_id, ha->ha_archive_cnt *
+                                sizeof(*ha->ha_archive_id));
                OBD_FREE_PTR(ha);
        }
        up_write(&cdt->cdt_agent_lock);
 
        cdt_mti = lu_context_key_get(&cdt->cdt_env.le_ctx, &mdt_thread_key);
        mutex_lock(&cdt->cdt_restore_lock);
-       list_for_each_entry_safe(crh, tmp3, &cdt->cdt_restore_hdl, crh_list) {
+       list_for_each_entry_safe(crh, tmp3, &cdt->cdt_restore_handle_list,
+                                crh_list) {
                list_del(&crh->crh_list);
                /* give back layout lock */
                mdt_object_unlock(cdt_mti, NULL, &crh->crh_lh, 1);
@@ -447,10 +537,10 @@ static void mdt_hsm_cdt_cleanup(struct mdt_device *mdt)
 static bool cdt_transition[CDT_STATES_COUNT][CDT_STATES_COUNT] = {
        /* from -> to:    stopped init   running disable stopping */
        /* stopped */   { true,   true,  false,  false,  false },
-       /* init */      { true,   false, true,   false,  true },
+       /* init */      { true,   false, true,   false,  false },
        /* running */   { false,  false, true,   true,   true },
        /* disable */   { false,  false, true,   true,   true },
-       /* stopping */  { true,   false, false,  false,  true }
+       /* stopping */  { true,   false, false,  false,  false }
 };
 
 /**
@@ -460,24 +550,18 @@ static bool cdt_transition[CDT_STATES_COUNT][CDT_STATES_COUNT] = {
  * Returns 0 on success, with old_state set if not NULL, or -EINVAL if
  * the transition was not possible.
  */
-static int set_cdt_state(struct coordinator *cdt, enum cdt_states new_state,
-                        enum cdt_states *old_state)
+static int set_cdt_state_locked(struct coordinator *cdt,
+                               enum cdt_states new_state)
 {
        int rc;
        enum cdt_states state;
 
-       spin_lock(&cdt->cdt_state_lock);
-
        state = cdt->cdt_state;
 
        if (cdt_transition[state][new_state]) {
                cdt->cdt_state = new_state;
-               spin_unlock(&cdt->cdt_state_lock);
-               if (old_state)
-                       *old_state = state;
                rc = 0;
        } else {
-               spin_unlock(&cdt->cdt_state_lock);
                CDEBUG(D_HSM,
                       "unexpected coordinator transition, from=%s, to=%s\n",
                       cdt_mdt_state2str(state), cdt_mdt_state2str(new_state));
@@ -487,6 +571,19 @@ static int set_cdt_state(struct coordinator *cdt, enum cdt_states new_state,
        return rc;
 }
 
+static int set_cdt_state(struct coordinator *cdt, enum cdt_states new_state)
+{
+       int rc;
+
+       mutex_lock(&cdt->cdt_state_lock);
+       rc = set_cdt_state_locked(cdt, new_state);
+       mutex_unlock(&cdt->cdt_state_lock);
+
+       return rc;
+}
+
+
+
 /**
  * coordinator thread
  * \param data [IN] obd device
@@ -499,80 +596,106 @@ static int mdt_coordinator(void *data)
        struct mdt_device       *mdt = mti->mti_mdt;
        struct coordinator      *cdt = &mdt->mdt_coordinator;
        struct hsm_scan_data     hsd = { NULL };
-       int                      rc = 0;
-       int                      request_sz;
+       time64_t                 last_housekeeping = 0;
+       size_t request_sz = 0;
+       int rc;
        ENTRY;
 
        CDEBUG(D_HSM, "%s: coordinator thread starting, pid=%d\n",
               mdt_obd_name(mdt), current_pid());
 
-       /* we use a copy of cdt_max_requests in the cb, so if cdt_max_requests
-        * increases due to a change from /proc we do not overflow the
-        * hsd.request[] vector
-        */
-       hsd.max_requests = cdt->cdt_max_requests;
-       request_sz = hsd.max_requests * sizeof(*hsd.request);
-       OBD_ALLOC_LARGE(hsd.request, request_sz);
-       if (!hsd.request)
-               GOTO(out, rc = -ENOMEM);
+       hsd.hsd_mti = mti;
+       obd_uuid2fsname(hsd.hsd_fsname, mdt_obd_name(mdt),
+                       sizeof(hsd.hsd_fsname));
 
-       hsd.mti = mti;
-       obd_uuid2fsname(hsd.fs_name, mdt_obd_name(mdt), MTI_NAME_MAXLEN);
-
-       set_cdt_state(cdt, CDT_RUNNING, NULL);
+       set_cdt_state(cdt, CDT_RUNNING);
 
        /* Inform mdt_hsm_cdt_start(). */
        wake_up_all(&cdt->cdt_waitq);
 
        while (1) {
-               struct l_wait_info lwi;
                int i;
+               int update_idx = 0;
+               int updates_sz;
+               int updates_cnt;
+               struct hsm_record_update *updates;
+
+               /* Limit execution of the expensive requests traversal
+                * to at most one second. This prevents repeatedly
+                * locking/unlocking the catalog for each request
+                * and preventing other HSM operations from happening
+                */
+               wait_event_interruptible_timeout(cdt->cdt_waitq,
+                                                kthread_should_stop() ||
+                                                cdt->cdt_wakeup_coordinator,
+                                                cfs_time_seconds(1));
 
-               lwi = LWI_TIMEOUT(cfs_time_seconds(cdt->cdt_loop_period),
-                                 NULL, NULL);
-               l_wait_event(cdt->cdt_waitq,
-                            cdt->cdt_event || (cdt->cdt_state == CDT_STOPPING),
-                            &lwi);
-
+               cdt->cdt_wakeup_coordinator = false;
                CDEBUG(D_HSM, "coordinator resumes\n");
 
-               if (cdt->cdt_state == CDT_STOPPING) {
+               if (kthread_should_stop()) {
+                       CDEBUG(D_HSM, "Coordinator stops\n");
                        rc = 0;
                        break;
                }
 
-               cdt->cdt_event = false;
-
                /* if coordinator is suspended continue to wait */
                if (cdt->cdt_state == CDT_DISABLE) {
                        CDEBUG(D_HSM, "disable state, coordinator sleeps\n");
                        continue;
                }
 
+               /* If no event, and no housekeeping to do, continue to
+                * wait. */
+               if (last_housekeeping + cdt->cdt_loop_period <=
+                   ktime_get_real_seconds()) {
+                       last_housekeeping = ktime_get_real_seconds();
+                       hsd.hsd_housekeeping = true;
+               } else if (cdt->cdt_event) {
+                       hsd.hsd_housekeeping = false;
+               } else {
+                       continue;
+               }
+
+               cdt->cdt_event = false;
+
                CDEBUG(D_HSM, "coordinator starts reading llog\n");
 
-               if (hsd.max_requests != cdt->cdt_max_requests) {
+               if (hsd.hsd_request_len != cdt->cdt_max_requests) {
                        /* cdt_max_requests has changed,
                         * we need to allocate a new buffer
                         */
-                       OBD_FREE_LARGE(hsd.request, request_sz);
-                       hsd.max_requests = cdt->cdt_max_requests;
-                       request_sz = hsd.max_requests * sizeof(*hsd.request);
-                       OBD_ALLOC_LARGE(hsd.request, request_sz);
-                       if (!hsd.request) {
-                               rc = -ENOMEM;
-                               break;
+                       struct hsm_scan_request *tmp = NULL;
+                       int max_requests = cdt->cdt_max_requests;
+                       OBD_ALLOC_LARGE(tmp, max_requests *
+                                       sizeof(struct hsm_scan_request));
+                       if (!tmp) {
+                               CERROR("Failed to resize request buffer, "
+                                      "keeping it at %d\n",
+                                      hsd.hsd_request_len);
+                       } else {
+                               if (hsd.hsd_request != NULL)
+                                       OBD_FREE_LARGE(hsd.hsd_request,
+                                                      request_sz);
+
+                               hsd.hsd_request_len = max_requests;
+                               request_sz = hsd.hsd_request_len *
+                                       sizeof(struct hsm_scan_request);
+                               hsd.hsd_request = tmp;
                        }
                }
 
-               hsd.request_cnt = 0;
+               hsd.hsd_action_count = 0;
+               hsd.hsd_request_count = 0;
+               hsd.hsd_one_restore = false;
 
                rc = cdt_llog_process(mti->mti_env, mdt, mdt_coordinator_cb,
-                                     &hsd, WRITE);
+                                     &hsd, 0, 0, WRITE);
                if (rc < 0)
                        goto clean_cb_alloc;
 
-               CDEBUG(D_HSM, "found %d requests to send\n", hsd.request_cnt);
+               CDEBUG(D_HSM, "found %d requests to send\n",
+                      hsd.hsd_request_count);
 
                if (list_empty(&cdt->cdt_agents)) {
                        CDEBUG(D_HSM, "no agent available, "
@@ -580,14 +703,32 @@ static int mdt_coordinator(void *data)
                        goto clean_cb_alloc;
                }
 
+               /* Compute how many HAI we have in all the requests */
+               updates_cnt = 0;
+               for (i = 0; i < hsd.hsd_request_count; i++) {
+                       const struct hsm_scan_request *request =
+                               &hsd.hsd_request[i];
+
+                       updates_cnt += request->hal->hal_count;
+               }
+
+               /* Allocate a temporary array to store the cookies to
+                * update, and their status. */
+               updates_sz = updates_cnt * sizeof(*updates);
+               OBD_ALLOC_LARGE(updates, updates_sz);
+               if (updates == NULL) {
+                       CERROR("%s: Cannot allocate memory (%d o) "
+                              "for %d updates\n",
+                              mdt_obd_name(mdt), updates_sz, updates_cnt);
+                       continue;
+               }
+
                /* here hsd contains a list of requests to be started */
-               for (i = 0; i < hsd.request_cnt; i++) {
-                       struct hsm_scan_request *request = &hsd.request[i];
+               for (i = 0; i < hsd.hsd_request_count; i++) {
+                       struct hsm_scan_request *request = &hsd.hsd_request[i];
                        struct hsm_action_list  *hal = request->hal;
                        struct hsm_action_item  *hai;
-                       __u64                   *cookies;
-                       int                      sz, j;
-                       enum agent_req_status    status;
+                       int                      j;
 
                        /* still room for work ? */
                        if (atomic_read(&cdt->cdt_request_count) >=
@@ -599,54 +740,46 @@ static int mdt_coordinator(void *data)
                         * if the copy tool failed to do the request
                         * it has to use hsm_progress
                         */
-                       status = (rc ? ARS_WAITING : ARS_STARTED);
 
                        /* set up cookie vector to set records status
                         * after copy tools start or failed
                         */
-                       sz = hal->hal_count * sizeof(__u64);
-                       OBD_ALLOC(cookies, sz);
-                       if (cookies == NULL)
-                               continue;
-
                        hai = hai_first(hal);
                        for (j = 0; j < hal->hal_count; j++) {
-                               cookies[j] = hai->hai_cookie;
+                               updates[update_idx].cookie = hai->hai_cookie;
+                               updates[update_idx].status =
+                                       (rc ? ARS_WAITING : ARS_STARTED);
                                hai = hai_next(hai);
+                               update_idx++;
                        }
+               }
 
-                       rc = mdt_agent_record_update(mti->mti_env, mdt, cookies,
-                                                    hal->hal_count, status);
+               if (update_idx) {
+                       rc = mdt_agent_record_update(mti->mti_env, mdt,
+                                                    updates, update_idx);
                        if (rc)
                                CERROR("%s: mdt_agent_record_update() failed, "
-                                      "rc=%d, cannot update status to %s "
+                                      "rc=%d, cannot update records "
                                       "for %d cookies\n",
-                                      mdt_obd_name(mdt), rc,
-                                      agent_req_status2name(status),
-                                      hal->hal_count);
-
-                       OBD_FREE(cookies, sz);
+                                      mdt_obd_name(mdt), rc, update_idx);
                }
+
+               OBD_FREE_LARGE(updates, updates_sz);
+
 clean_cb_alloc:
                /* free hal allocated by callback */
-               for (i = 0; i < hsd.request_cnt; i++) {
-                       struct hsm_scan_request *request = &hsd.request[i];
+               for (i = 0; i < hsd.hsd_request_count; i++) {
+                       struct hsm_scan_request *request = &hsd.hsd_request[i];
 
-                       OBD_FREE(request->hal, request->hal_sz);
+                       OBD_FREE_LARGE(request->hal, request->hal_sz);
                }
        }
-       EXIT;
-out:
-       set_cdt_state(cdt, CDT_STOPPING, NULL);
 
-       if (hsd.request)
-               OBD_FREE_LARGE(hsd.request, request_sz);
+       if (hsd.hsd_request != NULL)
+               OBD_FREE_LARGE(hsd.hsd_request, request_sz);
 
        mdt_hsm_cdt_cleanup(mdt);
 
-       set_cdt_state(cdt, CDT_STOPPED, NULL);
-       wake_up_all(&cdt->cdt_waitq);
-
        if (rc != 0)
                CERROR("%s: coordinator thread exiting, process=%d, rc=%d\n",
                       mdt_obd_name(mdt), current_pid(), rc);
@@ -655,6 +788,56 @@ out:
                              " no error\n",
                       mdt_obd_name(mdt), current_pid());
 
+       RETURN(rc);
+}
+
+int cdt_restore_handle_add(struct mdt_thread_info *mti, struct coordinator *cdt,
+                          const struct lu_fid *fid,
+                          const struct hsm_extent *he)
+{
+       struct cdt_restore_handle *crh;
+       struct mdt_object *obj;
+       int rc;
+       ENTRY;
+
+       OBD_SLAB_ALLOC_PTR(crh, mdt_hsm_cdt_kmem);
+       if (crh == NULL)
+               RETURN(-ENOMEM);
+
+       crh->crh_fid = *fid;
+       /* in V1 all file is restored
+        * crh->extent.start = he->offset;
+        * crh->extent.end = he->offset + he->length;
+        */
+       crh->crh_extent.start = 0;
+       crh->crh_extent.end = he->length;
+       /* get the layout lock */
+       mdt_lock_reg_init(&crh->crh_lh, LCK_EX);
+       obj = mdt_object_find_lock(mti, &crh->crh_fid, &crh->crh_lh,
+                                  MDS_INODELOCK_LAYOUT);
+       if (IS_ERR(obj))
+               GOTO(out_crh, rc = PTR_ERR(obj));
+
+       /* We do not keep a reference on the object during the restore
+        * which can be very long. */
+       mdt_object_put(mti->mti_env, obj);
+
+       mutex_lock(&cdt->cdt_restore_lock);
+       if (unlikely(cdt->cdt_state == CDT_STOPPED ||
+                    cdt->cdt_state == CDT_STOPPING)) {
+               mutex_unlock(&cdt->cdt_restore_lock);
+               GOTO(out_lh, rc = -EAGAIN);
+       }
+
+       list_add_tail(&crh->crh_list, &cdt->cdt_restore_handle_list);
+       mutex_unlock(&cdt->cdt_restore_lock);
+
+       RETURN(0);
+out_lh:
+       mdt_object_unlock(mti, NULL, &crh->crh_lh, 1);
+out_crh:
+       OBD_SLAB_FREE_PTR(crh, mdt_hsm_cdt_kmem);
+
        return rc;
 }
 
@@ -666,19 +849,41 @@ out:
  * \retval cdt_restore_handle found
  * \retval NULL not found
  */
-struct cdt_restore_handle *mdt_hsm_restore_hdl_find(struct coordinator *cdt,
-                                                      const struct lu_fid *fid)
+struct cdt_restore_handle *cdt_restore_handle_find(struct coordinator *cdt,
+                                                  const struct lu_fid *fid)
 {
-       struct cdt_restore_handle       *crh;
+       struct cdt_restore_handle *crh;
        ENTRY;
 
-       list_for_each_entry(crh, &cdt->cdt_restore_hdl, crh_list) {
+       list_for_each_entry(crh, &cdt->cdt_restore_handle_list, crh_list) {
                if (lu_fid_eq(&crh->crh_fid, fid))
                        RETURN(crh);
        }
+
        RETURN(NULL);
 }
 
+void cdt_restore_handle_del(struct mdt_thread_info *mti,
+                           struct coordinator *cdt, const struct lu_fid *fid)
+{
+       struct cdt_restore_handle *crh;
+
+       /* give back layout lock */
+       mutex_lock(&cdt->cdt_restore_lock);
+       crh = cdt_restore_handle_find(cdt, fid);
+       if (crh != NULL)
+               list_del(&crh->crh_list);
+       mutex_unlock(&cdt->cdt_restore_lock);
+
+       if (crh == NULL)
+               return;
+
+       /* XXX We pass a NULL object since the restore handle does not
+        * keep a reference on the object being restored. */
+       mdt_object_unlock(mti, NULL, &crh->crh_lh, 1);
+       OBD_SLAB_FREE_PTR(crh, mdt_hsm_cdt_kmem);
+}
+
 /**
  * data passed to llog_cat_process() callback
  * to scan requests and take actions
@@ -703,11 +908,9 @@ static int hsm_restore_cb(const struct lu_env *env,
 {
        struct llog_agent_req_rec       *larr;
        struct hsm_restore_data         *hrd;
-       struct cdt_restore_handle       *crh;
        struct hsm_action_item          *hai;
        struct mdt_thread_info          *mti;
        struct coordinator              *cdt;
-       struct mdt_object               *child;
        int rc;
        ENTRY;
 
@@ -732,39 +935,13 @@ static int hsm_restore_cb(const struct lu_env *env,
         * when being re-started */
        if (larr->arr_status == ARS_STARTED) {
                larr->arr_status = ARS_WAITING;
-               larr->arr_req_change = cfs_time_current_sec();
+               larr->arr_req_change = ktime_get_real_seconds();
                rc = llog_write(env, llh, hdr, hdr->lrh_index);
                if (rc != 0)
                        GOTO(out, rc);
        }
 
-       OBD_SLAB_ALLOC_PTR(crh, mdt_hsm_cdt_kmem);
-       if (crh == NULL)
-               RETURN(-ENOMEM);
-
-       crh->crh_fid = hai->hai_fid;
-       /* in V1 all file is restored
-       crh->extent.start = hai->hai_extent.offset;
-       crh->extent.end = hai->hai_extent.offset + hai->hai_extent.length;
-       */
-       crh->crh_extent.start = 0;
-       crh->crh_extent.end = hai->hai_extent.length;
-       /* get the layout lock */
-       mdt_lock_reg_init(&crh->crh_lh, LCK_EX);
-       child = mdt_object_find_lock(mti, &crh->crh_fid, &crh->crh_lh,
-                                    MDS_INODELOCK_LAYOUT);
-       if (IS_ERR(child))
-               GOTO(out, rc = PTR_ERR(child));
-
-       rc = 0;
-       /* we choose to not keep a reference
-        * on the object during the restore time which can be very long */
-       mdt_object_put(mti->mti_env, child);
-
-       mutex_lock(&cdt->cdt_restore_lock);
-       list_add_tail(&crh->crh_list, &cdt->cdt_restore_hdl);
-       mutex_unlock(&cdt->cdt_restore_lock);
-
+       rc = cdt_restore_handle_add(mti, cdt, &hai->hai_fid, &hai->hai_extent);
 out:
        RETURN(rc);
 }
@@ -782,8 +959,8 @@ static int mdt_hsm_pending_restore(struct mdt_thread_info *mti)
 
        hrd.hrd_mti = mti;
 
-       rc = cdt_llog_process(mti->mti_env, mti->mti_mdt,
-                             hsm_restore_cb, &hrd, WRITE);
+       rc = cdt_llog_process(mti->mti_env, mti->mti_mdt, hsm_restore_cb, &hrd,
+                             0, 0, WRITE);
 
        RETURN(rc);
 }
@@ -807,27 +984,8 @@ static int hsm_init_ucred(struct lu_ucred *uc)
        uc->uc_umask = 0777;
        uc->uc_ginfo = NULL;
        uc->uc_identity = NULL;
-
-       RETURN(0);
-}
-
-/**
- * wake up coordinator thread
- * \param mdt [IN] device
- * \retval 0 success
- * \retval -ve failure
- */
-int mdt_hsm_cdt_wakeup(struct mdt_device *mdt)
-{
-       struct coordinator      *cdt = &mdt->mdt_coordinator;
-       ENTRY;
-
-       if (cdt->cdt_state == CDT_STOPPED)
-               RETURN(-ESRCH);
-
-       /* wake up coordinator */
-       cdt->cdt_event = true;
-       wake_up_all(&cdt->cdt_waitq);
+       /* always record internal HSM activity if also enabled globally */
+       uc->uc_enable_audit = 1;
 
        RETURN(0);
 }
@@ -850,12 +1008,12 @@ int mdt_hsm_cdt_init(struct mdt_device *mdt)
        init_rwsem(&cdt->cdt_agent_lock);
        init_rwsem(&cdt->cdt_request_lock);
        mutex_init(&cdt->cdt_restore_lock);
-       spin_lock_init(&cdt->cdt_state_lock);
-       cdt->cdt_state = CDT_STOPPED;
+       mutex_init(&cdt->cdt_state_lock);
+       set_cdt_state(cdt, CDT_STOPPED);
 
        INIT_LIST_HEAD(&cdt->cdt_request_list);
        INIT_LIST_HEAD(&cdt->cdt_agents);
-       INIT_LIST_HEAD(&cdt->cdt_restore_hdl);
+       INIT_LIST_HEAD(&cdt->cdt_restore_handle_list);
 
        cdt->cdt_request_cookie_hash = cfs_hash_create("REQUEST_COOKIE_HASH",
                                                       CFS_HASH_BITS_MIN,
@@ -869,9 +1027,21 @@ int mdt_hsm_cdt_init(struct mdt_device *mdt)
        if (cdt->cdt_request_cookie_hash == NULL)
                RETURN(-ENOMEM);
 
+       cdt->cdt_agent_record_hash = cfs_hash_create("AGENT_RECORD_HASH",
+                                                    CFS_HASH_BITS_MIN,
+                                                    CFS_HASH_BITS_MAX,
+                                                    CFS_HASH_BKT_BITS,
+                                                    0 /* extra bytes */,
+                                                    CFS_HASH_MIN_THETA,
+                                                    CFS_HASH_MAX_THETA,
+                                                    &cdt_agent_record_hash_ops,
+                                                    CFS_HASH_DEFAULT);
+       if (cdt->cdt_agent_record_hash == NULL)
+               GOTO(out_request_cookie_hash, rc = -ENOMEM);
+
        rc = lu_env_init(&cdt->cdt_env, LCT_MD_THREAD);
        if (rc < 0)
-               GOTO(out_request_cookie_hash, rc);
+               GOTO(out_agent_record_hash, rc);
 
        /* for mdt_ucred(), lu_ucred stored in lu_ucred_key */
        rc = lu_context_init(&cdt->cdt_session, LCT_SERVER_SESSION);
@@ -898,10 +1068,16 @@ int mdt_hsm_cdt_init(struct mdt_device *mdt)
        cdt->cdt_policy = CDT_DEFAULT_POLICY;
        cdt->cdt_active_req_timeout = 3600;
 
+       /* by default do not remove archives on last unlink */
+       cdt->cdt_remove_archive_on_last_unlink = false;
+
        RETURN(0);
 
 out_env:
        lu_env_fini(&cdt->cdt_env);
+out_agent_record_hash:
+       cfs_hash_putref(cdt->cdt_agent_record_hash);
+       cdt->cdt_agent_record_hash = NULL;
 out_request_cookie_hash:
        cfs_hash_putref(cdt->cdt_request_cookie_hash);
        cdt->cdt_request_cookie_hash = NULL;
@@ -923,6 +1099,9 @@ int  mdt_hsm_cdt_fini(struct mdt_device *mdt)
 
        lu_env_fini(&cdt->cdt_env);
 
+       cfs_hash_putref(cdt->cdt_agent_record_hash);
+       cdt->cdt_agent_record_hash = NULL;
+
        cfs_hash_putref(cdt->cdt_request_cookie_hash);
        cdt->cdt_request_cookie_hash = NULL;
 
@@ -938,9 +1117,9 @@ int  mdt_hsm_cdt_fini(struct mdt_device *mdt)
 static int mdt_hsm_cdt_start(struct mdt_device *mdt)
 {
        struct coordinator      *cdt = &mdt->mdt_coordinator;
+       struct mdt_thread_info *cdt_mti;
        int                      rc;
        void                    *ptr;
-       struct mdt_thread_info  *cdt_mti;
        struct task_struct      *task;
        ENTRY;
 
@@ -949,7 +1128,7 @@ static int mdt_hsm_cdt_start(struct mdt_device *mdt)
         */
        ptr = dump_requests;
 
-       rc = set_cdt_state(cdt, CDT_INIT, NULL);
+       rc = set_cdt_state(cdt, CDT_INIT);
        if (rc) {
                CERROR("%s: Coordinator already started or stopping\n",
                       mdt_obd_name(mdt));
@@ -959,11 +1138,13 @@ static int mdt_hsm_cdt_start(struct mdt_device *mdt)
        CLASSERT(1 << (CDT_POLICY_SHIFT_COUNT - 1) == CDT_POLICY_LAST);
        cdt->cdt_policy = CDT_DEFAULT_POLICY;
 
-       atomic_set(&cdt->cdt_compound_id, cfs_time_current_sec());
        /* just need to be larger than previous one */
        /* cdt_last_cookie is protected by cdt_llog_lock */
-       cdt->cdt_last_cookie = cfs_time_current_sec();
+       cdt->cdt_last_cookie = ktime_get_real_seconds();
        atomic_set(&cdt->cdt_request_count, 0);
+       atomic_set(&cdt->cdt_archive_count, 0);
+       atomic_set(&cdt->cdt_restore_count, 0);
+       atomic_set(&cdt->cdt_remove_count, 0);
        cdt->cdt_user_request_mask = (1UL << HSMA_RESTORE);
        cdt->cdt_group_request_mask = (1UL << HSMA_RESTORE);
        cdt->cdt_other_request_mask = (1UL << HSMA_RESTORE);
@@ -985,22 +1166,16 @@ static int mdt_hsm_cdt_start(struct mdt_device *mdt)
        task = kthread_run(mdt_coordinator, cdt_mti, "hsm_cdtr");
        if (IS_ERR(task)) {
                rc = PTR_ERR(task);
-               set_cdt_state(cdt, CDT_STOPPED, NULL);
+               set_cdt_state(cdt, CDT_STOPPED);
                CERROR("%s: error starting coordinator thread: %d\n",
                       mdt_obd_name(mdt), rc);
        } else {
+               cdt->cdt_task = task;
                wait_event(cdt->cdt_waitq,
                           cdt->cdt_state != CDT_INIT);
-               if (cdt->cdt_state == CDT_RUNNING) {
-                       CDEBUG(D_HSM, "%s: coordinator thread started\n",
-                              mdt_obd_name(mdt));
-                       rc = 0;
-               } else {
-                       CDEBUG(D_HSM,
-                              "%s: coordinator thread failed to start\n",
-                              mdt_obd_name(mdt));
-                       rc = -EINVAL;
-               }
+               CDEBUG(D_HSM, "%s: coordinator thread started\n",
+                      mdt_obd_name(mdt));
+               rc = 0;
        }
 
        RETURN(rc);
@@ -1017,12 +1192,12 @@ int mdt_hsm_cdt_stop(struct mdt_device *mdt)
 
        ENTRY;
        /* stop coordinator thread */
-       rc = set_cdt_state(cdt, CDT_STOPPING, NULL);
-       if (rc != 0)
-               RETURN(rc);
-
-       wake_up_all(&cdt->cdt_waitq);
-       wait_event(cdt->cdt_waitq, cdt->cdt_state != CDT_STOPPING);
+       rc = set_cdt_state(cdt, CDT_STOPPING);
+       if (rc == 0) {
+               kthread_stop(cdt->cdt_task);
+               cdt->cdt_task = NULL;
+               set_cdt_state(cdt, CDT_STOPPED);
+       }
 
        RETURN(rc);
 }
@@ -1082,9 +1257,13 @@ int mdt_hsm_add_hal(struct mdt_thread_info *mti,
                 * it will be done when updating the request status
                 */
                if (hai->hai_action == HSMA_CANCEL) {
+                       struct hsm_record_update update = {
+                               .cookie = hai->hai_cookie,
+                               .status = ARS_CANCELED,
+                       };
+
                        rc = mdt_agent_record_update(mti->mti_env, mti->mti_mdt,
-                                                    &hai->hai_cookie,
-                                                    1, ARS_CANCELED);
+                                                    &update, 1);
                        if (rc) {
                                CERROR("%s: mdt_agent_record_update() failed, "
                                       "rc=%d, cannot update status to %s "
@@ -1120,8 +1299,7 @@ int mdt_hsm_add_hal(struct mdt_thread_info *mti,
                                GOTO(out, rc);
                }
 
-               car = mdt_cdt_alloc_request(hal->hal_compound_id,
-                                           hal->hal_archive_id, hal->hal_flags,
+               car = mdt_cdt_alloc_request(hal->hal_archive_id, hal->hal_flags,
                                            uuid, hai);
                if (IS_ERR(car))
                        GOTO(out, rc = PTR_ERR(car));
@@ -1184,7 +1362,15 @@ static int hsm_swap_layouts(struct mdt_thread_info *mti,
                                     mdt_object_child(obj),
                                     mdt_object_child(dobj),
                                     SWAP_LAYOUTS_MDS_HSM);
-
+       if (rc == 0) {
+               rc = mdt_lsom_downgrade(mti, obj);
+               if (rc)
+                       CDEBUG(D_INODE,
+                              "%s: File fid="DFID" SOM "
+                              "downgrade failed, rc = %d\n",
+                              mdt_obd_name(mti->mti_mdt),
+                              PFID(mdt_object_fid(obj)), rc);
+       }
 out_dobj:
        mdt_object_unlock_put(mti, dobj, dlh, 1);
 out:
@@ -1195,7 +1381,6 @@ out:
  * update status of a completed request
  * \param mti [IN] context
  * \param pgs [IN] progress of the copy tool
- * \param update_record [IN] update llog record
  * \retval 0 success
  * \retval -ve failure
  */
@@ -1204,16 +1389,17 @@ static int hsm_cdt_request_completed(struct mdt_thread_info *mti,
                                     const struct cdt_agent_req *car,
                                     enum agent_req_status *status)
 {
-       const struct lu_env     *env = mti->mti_env;
-       struct mdt_device       *mdt = mti->mti_mdt;
-       struct coordinator      *cdt = &mdt->mdt_coordinator;
-       struct mdt_object       *obj = NULL;
-       int                      cl_flags = 0, rc = 0;
-       struct md_hsm            mh;
-       bool                     is_mh_changed;
-       bool                     need_changelog = true;
-       ENTRY;
+       const struct lu_env *env = mti->mti_env;
+       struct mdt_device *mdt = mti->mti_mdt;
+       struct coordinator *cdt = &mdt->mdt_coordinator;
+       struct mdt_object *obj = NULL;
+       enum changelog_rec_flags clf_flags = 0;
+       struct md_hsm mh;
+       bool is_mh_changed;
+       bool need_changelog = true;
+       int rc = 0;
 
+       ENTRY;
        /* default is to retry */
        *status = ARS_WAITING;
 
@@ -1263,25 +1449,24 @@ static int hsm_cdt_request_completed(struct mdt_thread_info *mti,
                               mdt_obd_name(mdt),
                               pgs->hpk_cookie, PFID(&pgs->hpk_fid),
                               pgs->hpk_errval);
-                       hsm_set_cl_error(&cl_flags,
-                                        CLF_HSM_ERROVERFLOW);
+                       hsm_set_cl_error(&clf_flags, CLF_HSM_ERROVERFLOW);
                        rc = -EINVAL;
                } else {
-                       hsm_set_cl_error(&cl_flags, pgs->hpk_errval);
+                       hsm_set_cl_error(&clf_flags, pgs->hpk_errval);
                }
 
                switch (car->car_hai->hai_action) {
                case HSMA_ARCHIVE:
-                       hsm_set_cl_event(&cl_flags, HE_ARCHIVE);
+                       hsm_set_cl_event(&clf_flags, HE_ARCHIVE);
                        break;
                case HSMA_RESTORE:
-                       hsm_set_cl_event(&cl_flags, HE_RESTORE);
+                       hsm_set_cl_event(&clf_flags, HE_RESTORE);
                        break;
                case HSMA_REMOVE:
-                       hsm_set_cl_event(&cl_flags, HE_REMOVE);
+                       hsm_set_cl_event(&clf_flags, HE_REMOVE);
                        break;
                case HSMA_CANCEL:
-                       hsm_set_cl_event(&cl_flags, HE_CANCEL);
+                       hsm_set_cl_event(&clf_flags, HE_CANCEL);
                        CERROR("%s: Failed request %#llx on "DFID
                               " cannot be a CANCEL\n",
                               mdt_obd_name(mdt),
@@ -1301,7 +1486,7 @@ static int hsm_cdt_request_completed(struct mdt_thread_info *mti,
                *status = ARS_SUCCEED;
                switch (car->car_hai->hai_action) {
                case HSMA_ARCHIVE:
-                       hsm_set_cl_event(&cl_flags, HE_ARCHIVE);
+                       hsm_set_cl_event(&clf_flags, HE_ARCHIVE);
                        /* set ARCHIVE keep EXIST and clear LOST and
                         * DIRTY */
                        mh.mh_arch_ver = pgs->hpk_data_version;
@@ -1310,7 +1495,7 @@ static int hsm_cdt_request_completed(struct mdt_thread_info *mti,
                        is_mh_changed = true;
                        break;
                case HSMA_RESTORE:
-                       hsm_set_cl_event(&cl_flags, HE_RESTORE);
+                       hsm_set_cl_event(&clf_flags, HE_RESTORE);
 
                        /* do not clear RELEASED and DIRTY here
                         * this will occur in hsm_swap_layouts()
@@ -1322,13 +1507,13 @@ static int hsm_cdt_request_completed(struct mdt_thread_info *mti,
                        is_mh_changed = true;
                        break;
                case HSMA_REMOVE:
-                       hsm_set_cl_event(&cl_flags, HE_REMOVE);
+                       hsm_set_cl_event(&clf_flags, HE_REMOVE);
                        /* clear ARCHIVED EXISTS and LOST */
                        mh.mh_flags &= ~(HS_ARCHIVED | HS_EXISTS | HS_LOST);
                        is_mh_changed = true;
                        break;
                case HSMA_CANCEL:
-                       hsm_set_cl_event(&cl_flags, HE_CANCEL);
+                       hsm_set_cl_event(&clf_flags, HE_CANCEL);
                        CERROR("%s: Successful request %#llx on "DFID" cannot be a CANCEL\n",
                               mdt_obd_name(mdt),
                               pgs->hpk_cookie,
@@ -1346,9 +1531,11 @@ static int hsm_cdt_request_completed(struct mdt_thread_info *mti,
 
        /* rc != 0 means error when analysing action, it may come from
         * a crasy CT no need to manage DIRTY
+        * and if mdt_hsm_get_md_hsm() has returned an error, mh has not been
+        * filled
         */
-       if (rc == 0)
-               hsm_set_cl_flags(&cl_flags,
+       if (rc == 0 && !IS_ERR(obj))
+               hsm_set_cl_flags(&clf_flags,
                                 mh.mh_flags & HS_DIRTY ? CLF_HSM_DIRTY : 0);
 
        /* unlock is done later, after layout lock management */
@@ -1359,8 +1546,6 @@ static int hsm_cdt_request_completed(struct mdt_thread_info *mti,
         * if no retry will be attempted and if object is still alive,
         * in other cases we just unlock the object */
        if (car->car_hai->hai_action == HSMA_RESTORE) {
-               struct cdt_restore_handle       *crh;
-
                /* restore in data FID done, we swap the layouts
                 * only if restore is successful */
                if (pgs->hpk_errval == 0 && !IS_ERR(obj)) {
@@ -1379,25 +1564,11 @@ static int hsm_cdt_request_completed(struct mdt_thread_info *mti,
                /* restore special case, need to create ChangeLog record
                 * before to give back layout lock to avoid concurrent
                 * file updater to post out of order ChangeLog */
-               mo_changelog(env, CL_HSM, cl_flags, mdt->mdt_child,
+               mo_changelog(env, CL_HSM, clf_flags, mdt->mdt_child,
                             &car->car_hai->hai_fid);
                need_changelog = false;
 
-               /* give back layout lock */
-               mutex_lock(&cdt->cdt_restore_lock);
-               crh = mdt_hsm_restore_hdl_find(cdt, &car->car_hai->hai_fid);
-               if (crh != NULL)
-                       list_del(&crh->crh_list);
-               mutex_unlock(&cdt->cdt_restore_lock);
-               /* Just give back layout lock, we keep the reference
-                * which is given back later with the lock for HSM
-                * flags.
-                * XXX obj may be invalid so we do not pass it. */
-               if (crh != NULL)
-                       mdt_object_unlock(mti, NULL, &crh->crh_lh, 1);
-
-               if (crh != NULL)
-                       OBD_SLAB_FREE_PTR(crh, mdt_hsm_cdt_kmem);
+               cdt_restore_handle_del(mti, cdt, &car->car_hai->hai_fid);
        }
 
        GOTO(out, rc);
@@ -1405,7 +1576,7 @@ static int hsm_cdt_request_completed(struct mdt_thread_info *mti,
 out:
        /* always add a ChangeLog record */
        if (need_changelog)
-               mo_changelog(env, CL_HSM, cl_flags, mdt->mdt_child,
+               mo_changelog(env, CL_HSM, clf_flags, mdt->mdt_child,
                             &car->car_hai->hai_fid);
 
        if (!IS_ERR(obj))
@@ -1418,13 +1589,11 @@ out:
  * update status of a request
  * \param mti [IN] context
  * \param pgs [IN] progress of the copy tool
- * \param update_record [IN] update llog record
  * \retval 0 success
  * \retval -ve failure
  */
 int mdt_hsm_update_request_state(struct mdt_thread_info *mti,
-                                struct hsm_progress_kernel *pgs,
-                                const int update_record)
+                                struct hsm_progress_kernel *pgs)
 {
        struct mdt_device       *mdt = mti->mti_mdt;
        struct coordinator      *cdt = &mdt->mdt_coordinator;
@@ -1493,40 +1662,39 @@ int mdt_hsm_update_request_state(struct mdt_thread_info *mti,
        hsm_init_ucred(mdt_ucred(mti));
 
        if (pgs->hpk_flags & HP_FLAG_COMPLETED) {
-               enum agent_req_status    status;
+               enum agent_req_status status;
+               struct hsm_record_update update;
+               int rc1;
 
                rc = hsm_cdt_request_completed(mti, pgs, car, &status);
 
-               CDEBUG(D_HSM, "%s record: fid="DFID" cookie=%#llx action=%s "
+               CDEBUG(D_HSM, "updating record: fid="DFID" cookie=%#llx action=%s "
                              "status=%s\n",
-                      update_record ? "Updating" : "Not updating",
                       PFID(&pgs->hpk_fid), pgs->hpk_cookie,
                       hsm_copytool_action2name(car->car_hai->hai_action),
                       agent_req_status2name(status));
 
                /* update record first (LU-9075) */
-               if (update_record) {
-                       int rc1;
-
-                       rc1 = mdt_agent_record_update(mti->mti_env, mdt,
-                                                    &pgs->hpk_cookie, 1,
-                                                    status);
-                       if (rc1)
-                               CERROR("%s: mdt_agent_record_update() failed,"
-                                      " rc=%d, cannot update status to %s"
-                                      " for cookie %#llx\n",
-                                      mdt_obd_name(mdt), rc1,
-                                      agent_req_status2name(status),
-                                      pgs->hpk_cookie);
-                       rc = (rc != 0 ? rc : rc1);
-               }
+               update.cookie = pgs->hpk_cookie;
+               update.status = status;
+
+               rc1 = mdt_agent_record_update(mti->mti_env, mdt,
+                                             &update, 1);
+               if (rc1)
+                       CERROR("%s: mdt_agent_record_update() failed,"
+                              " rc=%d, cannot update status to %s"
+                              " for cookie %#llx\n",
+                              mdt_obd_name(mdt), rc1,
+                              agent_req_status2name(status),
+                              pgs->hpk_cookie);
+               rc = (rc != 0 ? rc : rc1);
 
                /* then remove request from memory list (LU-9075) */
                mdt_cdt_remove_request(cdt, pgs->hpk_cookie);
 
-               /* ct has completed a request, so a slot is available, wakeup
-                * cdt to find new work */
-               mdt_hsm_cdt_wakeup(mdt);
+               /* ct has completed a request, so a slot is available,
+                * signal the coordinator to find new work */
+               mdt_hsm_cdt_event(cdt);
        } else {
                /* if copytool send a progress on a canceled request
                 * we inform copytool it should stop
@@ -1576,7 +1744,7 @@ static int mdt_cancel_all_cb(const struct lu_env *env,
        if (larr->arr_status == ARS_WAITING ||
            larr->arr_status == ARS_STARTED) {
                larr->arr_status = ARS_CANCELED;
-               larr->arr_req_change = cfs_time_current_sec();
+               larr->arr_req_change = ktime_get_real_seconds();
                rc = llog_write(env, llh, hdr, hdr->lrh_index);
        }
 
@@ -1621,10 +1789,13 @@ static int hsm_cancel_all_actions(struct mdt_device *mdt)
 
        hsm_init_ucred(mdt_ucred(mti));
 
+       mutex_lock(&cdt->cdt_state_lock);
+       old_state = cdt->cdt_state;
+
        /* disable coordinator */
-       rc = set_cdt_state(cdt, CDT_DISABLE, &old_state);
+       rc = set_cdt_state_locked(cdt, CDT_DISABLE);
        if (rc)
-               RETURN(rc);
+               GOTO(out_cdt_state_unlock, rc);
 
        /* send cancel to all running requests */
        down_read(&cdt->cdt_request_lock);
@@ -1664,7 +1835,6 @@ static int hsm_cancel_all_actions(struct mdt_device *mdt)
                obd_uuid2fsname(hal->hal_fsname, mdt_obd_name(mdt),
                                MTI_NAME_MAXLEN);
                hal->hal_fsname[MTI_NAME_MAXLEN] = '\0';
-               hal->hal_compound_id = car->car_compound_id;
                hal->hal_archive_id = car->car_archive_id;
                hal->hal_flags = car->car_flags;
                hal->hal_count = 0;
@@ -1694,11 +1864,14 @@ static int hsm_cancel_all_actions(struct mdt_device *mdt)
        /* cancel all on-disk records */
        hcad.mdt = mdt;
 
-       rc = cdt_llog_process(mti->mti_env, mti->mti_mdt,
-                             mdt_cancel_all_cb, &hcad, WRITE);
+       rc = cdt_llog_process(mti->mti_env, mti->mti_mdt, mdt_cancel_all_cb,
+                             &hcad, 0, 0, WRITE);
 out_cdt_state:
        /* Enable coordinator, unless the coordinator was stopping. */
-       set_cdt_state(cdt, old_state, NULL);
+       set_cdt_state_locked(cdt, old_state);
+out_cdt_state_unlock:
+       mutex_unlock(&cdt->cdt_state_lock);
+
        lu_context_exit(&session);
        lu_context_fini(&session);
 out_env:
@@ -1941,14 +2114,15 @@ mdt_hsm_##VAR##_seq_write(struct file *file, const char __user *buffer, \
        struct seq_file         *m = file->private_data;                \
        struct mdt_device       *mdt = m->private;                      \
        struct coordinator      *cdt = &mdt->mdt_coordinator;           \
-       __s64                    val;                                   \
-       int                      rc;                                    \
-       ENTRY;                                                          \
+       unsigned int val;                                               \
+       int rc;                                                         \
                                                                        \
-       rc = lprocfs_str_to_s64(buffer, count, &val);                   \
+       ENTRY;                                                          \
+       rc = kstrtouint_from_user(buffer, count, 0, &val);              \
        if (rc)                                                         \
                RETURN(rc);                                             \
-       if (val > 0 && val < INT_MAX) {                                 \
+                                                                       \
+       if (val !=  0) {                                                \
                cdt->VAR = val;                                         \
                RETURN(count);                                          \
        }                                                               \
@@ -1997,8 +2171,9 @@ mdt_hsm_cdt_control_seq_write(struct file *file, const char __user *buffer,
        rc = 0;
        if (strcmp(kernbuf, CDT_ENABLE_CMD) == 0) {
                if (cdt->cdt_state == CDT_DISABLE) {
-                       rc = set_cdt_state(cdt, CDT_RUNNING, NULL);
-                       mdt_hsm_cdt_wakeup(mdt);
+                       rc = set_cdt_state(cdt, CDT_RUNNING);
+                       mdt_hsm_cdt_event(cdt);
+                       wake_up(&cdt->cdt_waitq);
                } else {
                        rc = mdt_hsm_cdt_start(mdt);
                }
@@ -2010,7 +2185,6 @@ mdt_hsm_cdt_control_seq_write(struct file *file, const char __user *buffer,
                        rc = -EALREADY;
                } else {
                        rc = mdt_hsm_cdt_stop(mdt);
-                       mdt_hsm_cdt_wakeup(mdt);
                }
        } else if (strcmp(kernbuf, CDT_DISABLE_CMD) == 0) {
                if ((cdt->cdt_state == CDT_STOPPING) ||
@@ -2019,7 +2193,7 @@ mdt_hsm_cdt_control_seq_write(struct file *file, const char __user *buffer,
                               mdt_obd_name(mdt));
                        rc = -EINVAL;
                } else {
-                       rc = set_cdt_state(cdt, CDT_DISABLE, NULL);
+                       rc = set_cdt_state(cdt, CDT_DISABLE);
                }
        } else if (strcmp(kernbuf, CDT_PURGE_CMD) == 0) {
                rc = hsm_cancel_all_actions(mdt);
@@ -2200,6 +2374,36 @@ mdt_hsm_other_request_mask_seq_write(struct file *file, const char __user *buf,
                                           &cdt->cdt_other_request_mask);
 }
 
+static int mdt_hsm_cdt_raolu_seq_show(struct seq_file *m, void *data)
+{
+       struct mdt_device *mdt = m->private;
+       struct coordinator *cdt = &mdt->mdt_coordinator;
+       ENTRY;
+
+       seq_printf(m, "%d\n", (int)cdt->cdt_remove_archive_on_last_unlink);
+       RETURN(0);
+}
+
+static ssize_t
+mdt_hsm_cdt_raolu_seq_write(struct file *file, const char __user *buffer,
+                           size_t count, loff_t *off)
+
+{
+       struct seq_file *m = file->private_data;
+       struct mdt_device *mdt = m->private;
+       struct coordinator *cdt = &mdt->mdt_coordinator;
+       bool val;
+       int rc;
+
+       ENTRY;
+       rc = kstrtobool_from_user(buffer, count, &val);
+       if (rc < 0)
+               RETURN(rc);
+
+       cdt->cdt_remove_archive_on_last_unlink = val;
+       RETURN(count);
+}
+
 LPROC_SEQ_FOPS(mdt_hsm_cdt_loop_period);
 LPROC_SEQ_FOPS(mdt_hsm_cdt_grace_delay);
 LPROC_SEQ_FOPS(mdt_hsm_cdt_active_req_timeout);
@@ -2208,6 +2412,42 @@ LPROC_SEQ_FOPS(mdt_hsm_cdt_default_archive_id);
 LPROC_SEQ_FOPS(mdt_hsm_user_request_mask);
 LPROC_SEQ_FOPS(mdt_hsm_group_request_mask);
 LPROC_SEQ_FOPS(mdt_hsm_other_request_mask);
+LPROC_SEQ_FOPS(mdt_hsm_cdt_raolu);
+
+/* Read-only proc files for request counters */
+static int mdt_hsm_cdt_archive_count_seq_show(struct seq_file *m, void *data)
+{
+       struct mdt_device *mdt = m->private;
+       struct coordinator *cdt = &mdt->mdt_coordinator;
+       ENTRY;
+
+       seq_printf(m, "%d\n", atomic_read(&cdt->cdt_archive_count));
+       RETURN(0);
+}
+
+static int mdt_hsm_cdt_restore_count_seq_show(struct seq_file *m, void *data)
+{
+       struct mdt_device *mdt = m->private;
+       struct coordinator *cdt = &mdt->mdt_coordinator;
+       ENTRY;
+
+       seq_printf(m, "%d\n", atomic_read(&cdt->cdt_restore_count));
+       RETURN(0);
+}
+
+static int mdt_hsm_cdt_remove_count_seq_show(struct seq_file *m, void *data)
+{
+       struct mdt_device *mdt = m->private;
+       struct coordinator *cdt = &mdt->mdt_coordinator;
+       ENTRY;
+
+       seq_printf(m, "%d\n", atomic_read(&cdt->cdt_remove_count));
+       RETURN(0);
+}
+
+LPROC_SEQ_FOPS_RO(mdt_hsm_cdt_archive_count);
+LPROC_SEQ_FOPS_RO(mdt_hsm_cdt_restore_count);
+LPROC_SEQ_FOPS_RO(mdt_hsm_cdt_remove_count);
 
 static struct lprocfs_vars lprocfs_mdt_hsm_vars[] = {
        { .name =       "agents",
@@ -2235,5 +2475,13 @@ static struct lprocfs_vars lprocfs_mdt_hsm_vars[] = {
          .fops =       &mdt_hsm_group_request_mask_fops,       },
        { .name =       "other_request_mask",
          .fops =       &mdt_hsm_other_request_mask_fops,       },
+       { .name =       "remove_archive_on_last_unlink",
+         .fops =       &mdt_hsm_cdt_raolu_fops,                },
+       { .name =       "archive_count",
+         .fops =       &mdt_hsm_cdt_archive_count_fops,        },
+       { .name =       "restore_count",
+         .fops =       &mdt_hsm_cdt_restore_count_fops,        },
+       { .name =       "remove_count",
+         .fops =       &mdt_hsm_cdt_remove_count_fops,         },
        { 0 }
 };