Whamcloud - gitweb
LU-6655 ptlrpc: skip delayed replay requests
[fs/lustre-release.git] / lustre / ldlm / ldlm_lib.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, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2010, 2017, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  * Lustre is a trademark of Sun Microsystems, Inc.
31  */
32
33 /**
34  * This file deals with various client/target related logic including recovery.
35  *
36  * TODO: This code more logically belongs in the ptlrpc module than in ldlm and
37  * should be moved.
38  */
39
40 #define DEBUG_SUBSYSTEM S_LDLM
41
42 #include <linux/kthread.h>
43 #include <libcfs/libcfs.h>
44 #include <obd.h>
45 #include <obd_class.h>
46 #include <lustre_dlm.h>
47 #include <lustre_net.h>
48 #include <lustre_sec.h>
49 #include "ldlm_internal.h"
50
51 /* @priority: If non-zero, move the selected connection to the list head.
52  * @create: If zero, only search in existing connections.
53  */
54 static int import_set_conn(struct obd_import *imp, struct obd_uuid *uuid,
55                            int priority, int create)
56 {
57         struct ptlrpc_connection *ptlrpc_conn;
58         struct obd_import_conn *imp_conn = NULL, *item;
59         lnet_nid_t nid4refnet = LNET_NID_ANY;
60         int rc = 0;
61         ENTRY;
62
63         if (!create && !priority) {
64                 CDEBUG(D_HA, "Nothing to do\n");
65                 RETURN(-EINVAL);
66         }
67
68         if (imp->imp_connection &&
69             imp->imp_connection->c_remote_uuid.uuid[0] == 0)
70                 /* nid4refnet is used to restrict network connections */
71                 nid4refnet = imp->imp_connection->c_self;
72         ptlrpc_conn = ptlrpc_uuid_to_connection(uuid, nid4refnet);
73         if (!ptlrpc_conn) {
74                 CDEBUG(D_HA, "can't find connection %s\n", uuid->uuid);
75                 RETURN(-ENOENT);
76         }
77
78         if (create) {
79                 OBD_ALLOC(imp_conn, sizeof(*imp_conn));
80                 if (!imp_conn)
81                         GOTO(out_put, rc = -ENOMEM);
82         }
83
84         spin_lock(&imp->imp_lock);
85         list_for_each_entry(item, &imp->imp_conn_list, oic_item) {
86                 if (obd_uuid_equals(uuid, &item->oic_uuid)) {
87                         if (priority) {
88                                 list_del(&item->oic_item);
89                                 list_add(&item->oic_item,
90                                          &imp->imp_conn_list);
91                                 item->oic_last_attempt = 0;
92                         }
93                         CDEBUG(D_HA, "imp %p@%s: found existing conn %s%s\n",
94                                imp, imp->imp_obd->obd_name, uuid->uuid,
95                                (priority ? ", moved to head" : ""));
96                         spin_unlock(&imp->imp_lock);
97                         GOTO(out_free, rc = 0);
98                 }
99         }
100         /* No existing import connection found for \a uuid. */
101         if (create) {
102                 imp_conn->oic_conn = ptlrpc_conn;
103                 imp_conn->oic_uuid = *uuid;
104                 imp_conn->oic_last_attempt = 0;
105                 if (priority)
106                         list_add(&imp_conn->oic_item, &imp->imp_conn_list);
107                 else
108                         list_add_tail(&imp_conn->oic_item,
109                                       &imp->imp_conn_list);
110                 CDEBUG(D_HA, "imp %p@%s: add connection %s at %s\n",
111                        imp, imp->imp_obd->obd_name, uuid->uuid,
112                        (priority ? "head" : "tail"));
113         } else {
114                 spin_unlock(&imp->imp_lock);
115                 GOTO(out_free, rc = -ENOENT);
116         }
117
118         spin_unlock(&imp->imp_lock);
119         RETURN(0);
120 out_free:
121         if (imp_conn)
122                 OBD_FREE(imp_conn, sizeof(*imp_conn));
123 out_put:
124         ptlrpc_connection_put(ptlrpc_conn);
125         RETURN(rc);
126 }
127
128 int import_set_conn_priority(struct obd_import *imp, struct obd_uuid *uuid)
129 {
130         return import_set_conn(imp, uuid, 1, 0);
131 }
132
133 int client_import_add_conn(struct obd_import *imp, struct obd_uuid *uuid,
134                            int priority)
135 {
136         return import_set_conn(imp, uuid, priority, 1);
137 }
138 EXPORT_SYMBOL(client_import_add_conn);
139
140 int client_import_del_conn(struct obd_import *imp, struct obd_uuid *uuid)
141 {
142         struct obd_import_conn *imp_conn;
143         struct obd_export *dlmexp;
144         int rc = -ENOENT;
145         ENTRY;
146
147         spin_lock(&imp->imp_lock);
148         if (list_empty(&imp->imp_conn_list)) {
149                 LASSERT(!imp->imp_connection);
150                 GOTO(out, rc);
151         }
152
153         list_for_each_entry(imp_conn, &imp->imp_conn_list, oic_item) {
154                 if (!obd_uuid_equals(uuid, &imp_conn->oic_uuid))
155                         continue;
156                 LASSERT(imp_conn->oic_conn);
157
158                 if (imp_conn == imp->imp_conn_current) {
159                         LASSERT(imp_conn->oic_conn == imp->imp_connection);
160
161                         if (imp->imp_state != LUSTRE_IMP_CLOSED &&
162                             imp->imp_state != LUSTRE_IMP_DISCON) {
163                                 CERROR("can't remove current connection\n");
164                                 GOTO(out, rc = -EBUSY);
165                         }
166
167                         ptlrpc_connection_put(imp->imp_connection);
168                         imp->imp_connection = NULL;
169
170                         dlmexp = class_conn2export(&imp->imp_dlm_handle);
171                         if (dlmexp && dlmexp->exp_connection) {
172                                 LASSERT(dlmexp->exp_connection ==
173                                         imp_conn->oic_conn);
174                                 ptlrpc_connection_put(dlmexp->exp_connection);
175                                 dlmexp->exp_connection = NULL;
176                         }
177
178                         if (dlmexp != NULL)
179                                 class_export_put(dlmexp);
180                 }
181
182                 list_del(&imp_conn->oic_item);
183                 ptlrpc_connection_put(imp_conn->oic_conn);
184                 OBD_FREE(imp_conn, sizeof(*imp_conn));
185                 CDEBUG(D_HA, "imp %p@%s: remove connection %s\n",
186                        imp, imp->imp_obd->obd_name, uuid->uuid);
187                 rc = 0;
188                 break;
189         }
190 out:
191         spin_unlock(&imp->imp_lock);
192         if (rc == -ENOENT)
193                 CERROR("connection %s not found\n", uuid->uuid);
194         RETURN(rc);
195 }
196 EXPORT_SYMBOL(client_import_del_conn);
197
198 /**
199  * Find conn UUID by peer NID. \a peer is a server NID. This function is used
200  * to find a conn uuid of \a imp which can reach \a peer.
201  */
202 int client_import_find_conn(struct obd_import *imp, lnet_nid_t peer,
203                             struct obd_uuid *uuid)
204 {
205         struct obd_import_conn *conn;
206         int rc = -ENOENT;
207         ENTRY;
208
209         spin_lock(&imp->imp_lock);
210         list_for_each_entry(conn, &imp->imp_conn_list, oic_item) {
211                 /* Check if conn UUID does have this peer NID. */
212                 if (class_check_uuid(&conn->oic_uuid, peer)) {
213                         *uuid = conn->oic_uuid;
214                         rc = 0;
215                         break;
216                 }
217         }
218         spin_unlock(&imp->imp_lock);
219         RETURN(rc);
220 }
221 EXPORT_SYMBOL(client_import_find_conn);
222
223 void client_destroy_import(struct obd_import *imp)
224 {
225         /* Drop security policy instance after all RPCs have finished/aborted
226          * to let all busy contexts be released. */
227         class_import_get(imp);
228         class_destroy_import(imp);
229         sptlrpc_import_sec_put(imp);
230         class_import_put(imp);
231 }
232 EXPORT_SYMBOL(client_destroy_import);
233
234 /**
235  * Check whether or not the OSC is on MDT.
236  * In the config log,
237  * osc on MDT
238  *      setup 0:{fsname}-OSTxxxx-osc[-MDTxxxx] 1:lustre-OST0000_UUID 2:NID
239  * osc on client
240  *      setup 0:{fsname}-OSTxxxx-osc 1:lustre-OST0000_UUID 2:NID
241  *
242  **/
243 static int osc_on_mdt(char *obdname)
244 {
245         char *ptr;
246
247         ptr = strrchr(obdname, '-');
248         if (ptr == NULL)
249                 return 0;
250
251         if (strncmp(ptr + 1, "MDT", 3) == 0)
252                 return 1;
253
254         return 0;
255 }
256
257 /* Configure an RPC client OBD device.
258  *
259  * lcfg parameters:
260  * 1 - client UUID
261  * 2 - server UUID
262  * 3 - inactive-on-startup
263  * 4 - restrictive net
264  */
265 int client_obd_setup(struct obd_device *obddev, struct lustre_cfg *lcfg)
266 {
267         struct client_obd *cli = &obddev->u.cli;
268         struct obd_import *imp;
269         struct obd_uuid server_uuid;
270         int rq_portal, rp_portal, connect_op;
271         char *name = obddev->obd_type->typ_name;
272         enum ldlm_ns_type ns_type = LDLM_NS_TYPE_UNKNOWN;
273         char *cli_name = lustre_cfg_buf(lcfg, 0);
274         struct ptlrpc_connection fake_conn = { .c_self = 0,
275                                                .c_remote_uuid.uuid[0] = 0 };
276         int rc;
277         ENTRY;
278
279         /* In a more perfect world, we would hang a ptlrpc_client off of
280          * obd_type and just use the values from there. */
281         if (!strcmp(name, LUSTRE_OSC_NAME)) {
282                 rq_portal = OST_REQUEST_PORTAL;
283                 rp_portal = OSC_REPLY_PORTAL;
284                 connect_op = OST_CONNECT;
285                 cli->cl_sp_me = LUSTRE_SP_CLI;
286                 cli->cl_sp_to = LUSTRE_SP_OST;
287                 ns_type = LDLM_NS_TYPE_OSC;
288         } else if (!strcmp(name, LUSTRE_MDC_NAME) ||
289                    !strcmp(name, LUSTRE_LWP_NAME)) {
290                 rq_portal = MDS_REQUEST_PORTAL;
291                 rp_portal = MDC_REPLY_PORTAL;
292                 connect_op = MDS_CONNECT;
293                 if (is_lwp_on_ost(cli_name))
294                         cli->cl_sp_me = LUSTRE_SP_OST;
295                 else if (is_lwp_on_mdt(cli_name))
296                         cli->cl_sp_me = LUSTRE_SP_MDT;
297                 else
298                         cli->cl_sp_me = LUSTRE_SP_CLI;
299                 cli->cl_sp_to = LUSTRE_SP_MDT;
300                 ns_type = LDLM_NS_TYPE_MDC;
301         } else if (!strcmp(name, LUSTRE_OSP_NAME)) {
302                 if (strstr(lustre_cfg_buf(lcfg, 1), "OST") == NULL) {
303                         /* OSP_on_MDT for other MDTs */
304                         connect_op = MDS_CONNECT;
305                         cli->cl_sp_to = LUSTRE_SP_MDT;
306                         ns_type = LDLM_NS_TYPE_MDC;
307                         rq_portal = OUT_PORTAL;
308                 } else {
309                         /* OSP on MDT for OST */
310                         connect_op = OST_CONNECT;
311                         cli->cl_sp_to = LUSTRE_SP_OST;
312                         ns_type = LDLM_NS_TYPE_OSC;
313                         rq_portal = OST_REQUEST_PORTAL;
314                 }
315                 rp_portal = OSC_REPLY_PORTAL;
316                 cli->cl_sp_me = LUSTRE_SP_MDT;
317         } else if (!strcmp(name, LUSTRE_MGC_NAME)) {
318                 rq_portal = MGS_REQUEST_PORTAL;
319                 rp_portal = MGC_REPLY_PORTAL;
320                 connect_op = MGS_CONNECT;
321                 cli->cl_sp_me = LUSTRE_SP_MGC;
322                 cli->cl_sp_to = LUSTRE_SP_MGS;
323                 cli->cl_flvr_mgc.sf_rpc = SPTLRPC_FLVR_INVALID;
324                 ns_type = LDLM_NS_TYPE_MGC;
325         } else {
326                 CERROR("unknown client OBD type \"%s\", can't setup\n",
327                        name);
328                 RETURN(-EINVAL);
329         }
330
331         if (LUSTRE_CFG_BUFLEN(lcfg, 1) < 1) {
332                 CERROR("requires a TARGET UUID\n");
333                 RETURN(-EINVAL);
334         }
335
336         if (LUSTRE_CFG_BUFLEN(lcfg, 1) > 37) {
337                 CERROR("client UUID must be less than 38 characters\n");
338                 RETURN(-EINVAL);
339         }
340
341         if (LUSTRE_CFG_BUFLEN(lcfg, 2) < 1) {
342                 CERROR("setup requires a SERVER UUID\n");
343                 RETURN(-EINVAL);
344         }
345
346         if (LUSTRE_CFG_BUFLEN(lcfg, 2) > 37) {
347                 CERROR("target UUID must be less than 38 characters\n");
348                 RETURN(-EINVAL);
349         }
350
351         init_rwsem(&cli->cl_sem);
352         mutex_init(&cli->cl_mgc_mutex);
353         cli->cl_seq = NULL;
354         init_rwsem(&cli->cl_seq_rwsem);
355         cli->cl_conn_count = 0;
356         memcpy(server_uuid.uuid, lustre_cfg_buf(lcfg, 2),
357                min_t(unsigned int, LUSTRE_CFG_BUFLEN(lcfg, 2),
358                      sizeof(server_uuid)));
359
360         cli->cl_dirty_pages = 0;
361         cli->cl_avail_grant = 0;
362         /* FIXME: Should limit this for the sum of all cl_dirty_max_pages. */
363         /* cl_dirty_max_pages may be changed at connect time in
364          * ptlrpc_connect_interpret(). */
365         client_adjust_max_dirty(cli);
366         INIT_LIST_HEAD(&cli->cl_cache_waiters);
367         INIT_LIST_HEAD(&cli->cl_loi_ready_list);
368         INIT_LIST_HEAD(&cli->cl_loi_hp_ready_list);
369         INIT_LIST_HEAD(&cli->cl_loi_write_list);
370         INIT_LIST_HEAD(&cli->cl_loi_read_list);
371         spin_lock_init(&cli->cl_loi_list_lock);
372         atomic_set(&cli->cl_pending_w_pages, 0);
373         atomic_set(&cli->cl_pending_r_pages, 0);
374         cli->cl_r_in_flight = 0;
375         cli->cl_w_in_flight = 0;
376
377         spin_lock_init(&cli->cl_read_rpc_hist.oh_lock);
378         spin_lock_init(&cli->cl_write_rpc_hist.oh_lock);
379         spin_lock_init(&cli->cl_read_page_hist.oh_lock);
380         spin_lock_init(&cli->cl_write_page_hist.oh_lock);
381         spin_lock_init(&cli->cl_read_offset_hist.oh_lock);
382         spin_lock_init(&cli->cl_write_offset_hist.oh_lock);
383
384         /* lru for osc. */
385         INIT_LIST_HEAD(&cli->cl_lru_osc);
386         atomic_set(&cli->cl_lru_shrinkers, 0);
387         atomic_long_set(&cli->cl_lru_busy, 0);
388         atomic_long_set(&cli->cl_lru_in_list, 0);
389         INIT_LIST_HEAD(&cli->cl_lru_list);
390         spin_lock_init(&cli->cl_lru_list_lock);
391         atomic_long_set(&cli->cl_unstable_count, 0);
392         INIT_LIST_HEAD(&cli->cl_shrink_list);
393
394         INIT_LIST_HEAD(&cli->cl_flight_waiters);
395         cli->cl_rpcs_in_flight = 0;
396
397         init_waitqueue_head(&cli->cl_destroy_waitq);
398         atomic_set(&cli->cl_destroy_in_flight, 0);
399 #ifdef ENABLE_CHECKSUM
400         /* Turn on checksumming by default. */
401         cli->cl_checksum = 1;
402         /*
403          * The supported checksum types will be worked out at connect time
404          * Set cl_chksum* to CRC32 for now to avoid returning screwed info
405          * through procfs.
406          */
407         cli->cl_cksum_type = cli->cl_supp_cksum_types = OBD_CKSUM_CRC32;
408 #endif
409         atomic_set(&cli->cl_resends, OSC_DEFAULT_RESENDS);
410
411         /* Set it to possible maximum size. It may be reduced by ocd_brw_size
412          * from OFD after connecting. */
413         cli->cl_max_pages_per_rpc = PTLRPC_MAX_BRW_PAGES;
414
415         cli->cl_short_io_bytes = OBD_MAX_SHORT_IO_BYTES;
416
417         /* set cl_chunkbits default value to PAGE_SHIFT,
418          * it will be updated at OSC connection time. */
419         cli->cl_chunkbits = PAGE_SHIFT;
420
421         if (!strcmp(name, LUSTRE_MDC_NAME)) {
422                 cli->cl_max_rpcs_in_flight = OBD_MAX_RIF_DEFAULT;
423         } else if (totalram_pages >> (20 - PAGE_SHIFT) <= 128 /* MB */) {
424                 cli->cl_max_rpcs_in_flight = 2;
425         } else if (totalram_pages >> (20 - PAGE_SHIFT) <= 256 /* MB */) {
426                 cli->cl_max_rpcs_in_flight = 3;
427         } else if (totalram_pages >> (20 - PAGE_SHIFT) <= 512 /* MB */) {
428                 cli->cl_max_rpcs_in_flight = 4;
429         } else {
430                 if (osc_on_mdt(obddev->obd_name))
431                         cli->cl_max_rpcs_in_flight = OBD_MAX_RIF_MAX;
432                 else
433                         cli->cl_max_rpcs_in_flight = OBD_MAX_RIF_DEFAULT;
434         }
435
436         spin_lock_init(&cli->cl_mod_rpcs_lock);
437         spin_lock_init(&cli->cl_mod_rpcs_hist.oh_lock);
438         cli->cl_max_mod_rpcs_in_flight = 0;
439         cli->cl_mod_rpcs_in_flight = 0;
440         cli->cl_close_rpcs_in_flight = 0;
441         init_waitqueue_head(&cli->cl_mod_rpcs_waitq);
442         cli->cl_mod_tag_bitmap = NULL;
443
444         INIT_LIST_HEAD(&cli->cl_chg_dev_linkage);
445
446         if (connect_op == MDS_CONNECT) {
447                 cli->cl_max_mod_rpcs_in_flight = cli->cl_max_rpcs_in_flight - 1;
448                 OBD_ALLOC(cli->cl_mod_tag_bitmap,
449                           BITS_TO_LONGS(OBD_MAX_RIF_MAX) * sizeof(long));
450                 if (cli->cl_mod_tag_bitmap == NULL)
451                         GOTO(err, rc = -ENOMEM);
452         }
453
454         rc = ldlm_get_ref();
455         if (rc) {
456                 CERROR("ldlm_get_ref failed: %d\n", rc);
457                 GOTO(err, rc);
458         }
459
460         ptlrpc_init_client(rq_portal, rp_portal, name,
461                            &obddev->obd_ldlm_client);
462
463         imp = class_new_import(obddev);
464         if (imp == NULL)
465                 GOTO(err_ldlm, rc = -ENOENT);
466         imp->imp_client = &obddev->obd_ldlm_client;
467         imp->imp_connect_op = connect_op;
468         memcpy(cli->cl_target_uuid.uuid, lustre_cfg_buf(lcfg, 1),
469                LUSTRE_CFG_BUFLEN(lcfg, 1));
470         class_import_put(imp);
471
472         if (lustre_cfg_buf(lcfg, 4)) {
473                 __u32 refnet = libcfs_str2net(lustre_cfg_string(lcfg, 4));
474
475                 if (refnet == LNET_NIDNET(LNET_NID_ANY)) {
476                         rc = -EINVAL;
477                         CERROR("%s: bad mount option 'network=%s': rc = %d\n",
478                                obddev->obd_name, lustre_cfg_string(lcfg, 4),
479                                rc);
480                         GOTO(err_import, rc);
481                 }
482                 fake_conn.c_self = LNET_MKNID(refnet, 0);
483                 imp->imp_connection = &fake_conn;
484         }
485
486         rc = client_import_add_conn(imp, &server_uuid, 1);
487         if (rc) {
488                 CERROR("can't add initial connection\n");
489                 GOTO(err_import, rc);
490         }
491         imp->imp_connection = NULL;
492
493         cli->cl_import = imp;
494         /* cli->cl_max_mds_easize updated by mdc_init_ea_size() */
495         cli->cl_max_mds_easize = sizeof(struct lov_mds_md_v3);
496
497         if (LUSTRE_CFG_BUFLEN(lcfg, 3) > 0) {
498                 if (!strcmp(lustre_cfg_string(lcfg, 3), "inactive")) {
499                         CDEBUG(D_HA, "marking %s %s->%s as inactive\n",
500                                name, obddev->obd_name,
501                                cli->cl_target_uuid.uuid);
502                         spin_lock(&imp->imp_lock);
503                         imp->imp_deactive = 1;
504                         spin_unlock(&imp->imp_lock);
505                 }
506         }
507
508         obddev->obd_namespace = ldlm_namespace_new(obddev, obddev->obd_name,
509                                                    LDLM_NAMESPACE_CLIENT,
510                                                    LDLM_NAMESPACE_GREEDY,
511                                                    ns_type);
512         if (obddev->obd_namespace == NULL) {
513                 CERROR("Unable to create client namespace - %s\n",
514                        obddev->obd_name);
515                 GOTO(err_import, rc = -ENOMEM);
516         }
517
518         RETURN(rc);
519
520 err_import:
521         class_destroy_import(imp);
522 err_ldlm:
523         ldlm_put_ref();
524 err:
525         if (cli->cl_mod_tag_bitmap != NULL)
526                 OBD_FREE(cli->cl_mod_tag_bitmap,
527                          BITS_TO_LONGS(OBD_MAX_RIF_MAX) * sizeof(long));
528         cli->cl_mod_tag_bitmap = NULL;
529         RETURN(rc);
530
531 }
532 EXPORT_SYMBOL(client_obd_setup);
533
534 int client_obd_cleanup(struct obd_device *obddev)
535 {
536         struct client_obd *cli = &obddev->u.cli;
537         ENTRY;
538
539         ldlm_namespace_free_post(obddev->obd_namespace);
540         obddev->obd_namespace = NULL;
541
542         obd_cleanup_client_import(obddev);
543         LASSERT(obddev->u.cli.cl_import == NULL);
544
545         ldlm_put_ref();
546
547         if (cli->cl_mod_tag_bitmap != NULL)
548                 OBD_FREE(cli->cl_mod_tag_bitmap,
549                          BITS_TO_LONGS(OBD_MAX_RIF_MAX) * sizeof(long));
550         cli->cl_mod_tag_bitmap = NULL;
551
552         RETURN(0);
553 }
554 EXPORT_SYMBOL(client_obd_cleanup);
555
556 /* ->o_connect() method for client side (OSC and MDC and MGC) */
557 int client_connect_import(const struct lu_env *env,
558                           struct obd_export **exp,
559                           struct obd_device *obd, struct obd_uuid *cluuid,
560                           struct obd_connect_data *data, void *localdata)
561 {
562         struct client_obd       *cli    = &obd->u.cli;
563         struct obd_import       *imp    = cli->cl_import;
564         struct obd_connect_data *ocd;
565         struct lustre_handle    conn    = { 0 };
566         int                     rc;
567         ENTRY;
568
569         *exp = NULL;
570         down_write(&cli->cl_sem);
571         if (cli->cl_conn_count > 0)
572                 GOTO(out_sem, rc = -EALREADY);
573
574         rc = class_connect(&conn, obd, cluuid);
575         if (rc)
576                 GOTO(out_sem, rc);
577
578         cli->cl_conn_count++;
579         *exp = class_conn2export(&conn);
580
581         LASSERT(obd->obd_namespace);
582
583         imp->imp_dlm_handle = conn;
584         rc = ptlrpc_init_import(imp);
585         if (rc != 0)
586                 GOTO(out_ldlm, rc);
587
588         ocd = &imp->imp_connect_data;
589         if (data) {
590                 *ocd = *data;
591                 imp->imp_connect_flags_orig = data->ocd_connect_flags;
592                 imp->imp_connect_flags2_orig = data->ocd_connect_flags2;
593         }
594
595         rc = ptlrpc_connect_import(imp);
596         if (rc != 0) {
597                 LASSERT(imp->imp_state == LUSTRE_IMP_DISCON);
598                 GOTO(out_ldlm, rc);
599         }
600         LASSERT(*exp != NULL && (*exp)->exp_connection);
601
602         if (data) {
603                 LASSERTF((ocd->ocd_connect_flags & data->ocd_connect_flags) ==
604                          ocd->ocd_connect_flags, "old %#llx, new %#llx\n",
605                          data->ocd_connect_flags, ocd->ocd_connect_flags);
606                 data->ocd_connect_flags = ocd->ocd_connect_flags;
607                 data->ocd_connect_flags2 = ocd->ocd_connect_flags2;
608         }
609
610         ptlrpc_pinger_add_import(imp);
611
612         EXIT;
613
614         if (rc) {
615 out_ldlm:
616                 cli->cl_conn_count--;
617                 class_disconnect(*exp);
618                 *exp = NULL;
619         }
620 out_sem:
621         up_write(&cli->cl_sem);
622
623         return rc;
624 }
625 EXPORT_SYMBOL(client_connect_import);
626
627 int client_disconnect_export(struct obd_export *exp)
628 {
629         struct obd_device *obd = class_exp2obd(exp);
630         struct client_obd *cli;
631         struct obd_import *imp;
632         int rc = 0, err;
633         ENTRY;
634
635         if (!obd) {
636                 CERROR("invalid export for disconnect: exp %p cookie %#llx\n",
637                        exp, exp ? exp->exp_handle.h_cookie : -1);
638                 RETURN(-EINVAL);
639         }
640
641         cli = &obd->u.cli;
642         imp = cli->cl_import;
643
644         down_write(&cli->cl_sem);
645         CDEBUG(D_INFO, "disconnect %s - %zu\n", obd->obd_name,
646                 cli->cl_conn_count);
647
648         if (cli->cl_conn_count == 0) {
649                 CERROR("disconnecting disconnected device (%s)\n",
650                        obd->obd_name);
651                 GOTO(out_disconnect, rc = -EINVAL);
652         }
653
654         cli->cl_conn_count--;
655         if (cli->cl_conn_count != 0)
656                 GOTO(out_disconnect, rc = 0);
657
658         /* Mark import deactivated now, so we don't try to reconnect if any
659          * of the cleanup RPCs fails (e.g. LDLM cancel, etc).  We don't
660          * fully deactivate the import, or that would drop all requests. */
661         spin_lock(&imp->imp_lock);
662         imp->imp_deactive = 1;
663         spin_unlock(&imp->imp_lock);
664
665         /* Some non-replayable imports (MDS's OSCs) are pinged, so just
666          * delete it regardless.  (It's safe to delete an import that was
667          * never added.) */
668         (void)ptlrpc_pinger_del_import(imp);
669
670         if (obd->obd_namespace != NULL) {
671                 /* obd_force == local only */
672                 ldlm_cli_cancel_unused(obd->obd_namespace, NULL,
673                                        obd->obd_force ? LCF_LOCAL : 0, NULL);
674                 ldlm_namespace_free_prior(obd->obd_namespace, imp, obd->obd_force);
675         }
676
677         /* There's no need to hold sem while disconnecting an import,
678          * and it may actually cause deadlock in GSS. */
679         up_write(&cli->cl_sem);
680         rc = ptlrpc_disconnect_import(imp, 0);
681         down_write(&cli->cl_sem);
682
683         ptlrpc_invalidate_import(imp);
684
685         EXIT;
686
687 out_disconnect:
688         /* Use server style - class_disconnect should be always called for
689          * o_disconnect. */
690         err = class_disconnect(exp);
691         if (!rc && err)
692                 rc = err;
693
694         up_write(&cli->cl_sem);
695
696         RETURN(rc);
697 }
698 EXPORT_SYMBOL(client_disconnect_export);
699
700 #ifdef HAVE_SERVER_SUPPORT
701 int server_disconnect_export(struct obd_export *exp)
702 {
703         int rc;
704         ENTRY;
705
706         /* Disconnect early so that clients can't keep using export. */
707         rc = class_disconnect(exp);
708         /* Close import to avoid sending any requests. */
709         if (exp->exp_imp_reverse)
710                 ptlrpc_cleanup_imp(exp->exp_imp_reverse);
711
712         ldlm_bl_thread_wakeup();
713
714         /* complete all outstanding replies */
715         spin_lock(&exp->exp_lock);
716         while (!list_empty(&exp->exp_outstanding_replies)) {
717                 struct ptlrpc_reply_state *rs =
718                         list_entry(exp->exp_outstanding_replies.next,
719                                        struct ptlrpc_reply_state, rs_exp_list);
720                 struct ptlrpc_service_part *svcpt = rs->rs_svcpt;
721
722                 spin_lock(&svcpt->scp_rep_lock);
723
724                 list_del_init(&rs->rs_exp_list);
725
726                 spin_lock(&rs->rs_lock);
727                 /* clear rs_convert_lock to make sure rs is handled and put */
728                 rs->rs_convert_lock = 0;
729                 ptlrpc_schedule_difficult_reply(rs);
730                 spin_unlock(&rs->rs_lock);
731
732                 spin_unlock(&svcpt->scp_rep_lock);
733         }
734         spin_unlock(&exp->exp_lock);
735
736         RETURN(rc);
737 }
738 EXPORT_SYMBOL(server_disconnect_export);
739
740 /* --------------------------------------------------------------------------
741  * from old lib/target.c
742  * -------------------------------------------------------------------------- */
743
744 static int target_handle_reconnect(struct lustre_handle *conn,
745                                    struct obd_export *exp,
746                                    struct obd_uuid *cluuid)
747 {
748         struct obd_device *target;
749         struct lustre_handle *hdl;
750         time64_t deadline;
751         time64_t timeout;
752         time64_t now;
753         int rc = 0;
754
755         ENTRY;
756         hdl = &exp->exp_imp_reverse->imp_remote_handle;
757         if (!exp->exp_connection || !lustre_handle_is_used(hdl)) {
758                 conn->cookie = exp->exp_handle.h_cookie;
759                 CDEBUG(D_HA, "connect export for UUID '%s' at %p,"
760                        " cookie %#llx\n", cluuid->uuid, exp, conn->cookie);
761                 RETURN(0);
762         }
763
764         target = exp->exp_obd;
765
766         /* Might be a re-connect after a partition. */
767         if (memcmp(&conn->cookie, &hdl->cookie, sizeof conn->cookie)) {
768                 LCONSOLE_WARN("%s: already connected client %s (at %s) "
769                               "with handle %#llx. Rejecting client "
770                               "with the same UUID trying to reconnect "
771                               "with handle %#llx\n", target->obd_name,
772                               obd_uuid2str(&exp->exp_client_uuid),
773                               obd_export_nid2str(exp),
774                               hdl->cookie, conn->cookie);
775                 memset(conn, 0, sizeof *conn);
776                 /* target_handle_connect() treats EALREADY and
777                  * -EALREADY differently.  -EALREADY is an error
778                  * (same UUID, different handle). */
779                 RETURN(-EALREADY);
780         }
781
782         if (!target->obd_recovering) {
783                 LCONSOLE_WARN("%s: Client %s (at %s) reconnecting\n",
784                         target->obd_name, obd_uuid2str(&exp->exp_client_uuid),
785                         obd_export_nid2str(exp));
786                 GOTO(out_already, rc);
787         }
788
789         now = ktime_get_seconds();
790         deadline = cfs_duration_sec(target->obd_recovery_timer.expires);
791         if (now < deadline) {
792                 struct target_distribute_txn_data *tdtd;
793                 int size = 0;
794                 int count = 0;
795                 char *buf = NULL;
796
797                 timeout = deadline - now;
798                 tdtd = class_exp2tgt(exp)->lut_tdtd;
799                 if (tdtd && tdtd->tdtd_show_update_logs_retrievers)
800                         buf = tdtd->tdtd_show_update_logs_retrievers(
801                                 tdtd->tdtd_show_retrievers_cbdata,
802                                 &size, &count);
803
804                 if (count > 0)
805                         LCONSOLE_WARN("%s: Recovery already passed deadline "
806                                       "%lld:%.02lld. It is due to DNE recovery "
807                                       "failed/stuck on the %d MDT(s):%s. "
808                                       "Please wait until all MDTs recovered "
809                                       "or abort the recovery by force.\n",
810                                       target->obd_name, timeout / 60,
811                                       timeout % 60, count,
812                                       buf ? buf : "unknown (not enough RAM)");
813                 else
814                         LCONSOLE_WARN("%s: Recovery already passed deadline "
815                                       "%lld:%.02lld. If you do not want to wait "
816                                       "more, please abort the recovery by "
817                                       "force.\n", target->obd_name,
818                                       timeout / 60, timeout % 60);
819
820                 if (buf != NULL)
821                         OBD_FREE(buf, size);
822         } else {
823                 timeout = now - deadline;
824                 LCONSOLE_WARN("%s: Recovery already passed deadline"
825                         " %lld:%.02lld, It is most likely due to DNE"
826                         " recovery is failed or stuck, please wait a"
827                         " few more minutes or abort the recovery.\n",
828                         target->obd_name, timeout / 60, timeout % 60);
829         }
830
831 out_already:
832         conn->cookie = exp->exp_handle.h_cookie;
833         /* target_handle_connect() treats EALREADY and
834          * -EALREADY differently.  EALREADY means we are
835          * doing a valid reconnect from the same client. */
836         RETURN(EALREADY);
837 }
838
839 static void
840 check_and_start_recovery_timer(struct obd_device *obd,
841                                struct ptlrpc_request *req, int new_client);
842
843 /**
844  * update flags for import during reconnect process
845  */
846 static int rev_import_flags_update(struct obd_import *revimp,
847                                    struct ptlrpc_request *req)
848 {
849         int rc;
850         struct obd_connect_data *data;
851
852         data = req_capsule_client_get(&req->rq_pill, &RMF_CONNECT_DATA);
853
854         if (data->ocd_connect_flags & OBD_CONNECT_AT)
855                 revimp->imp_msghdr_flags |= MSGHDR_AT_SUPPORT;
856         else
857                 revimp->imp_msghdr_flags &= ~MSGHDR_AT_SUPPORT;
858
859         revimp->imp_msghdr_flags |= MSGHDR_CKSUM_INCOMPAT18;
860
861         rc = sptlrpc_import_sec_adapt(revimp, req->rq_svc_ctx, &req->rq_flvr);
862         if (rc) {
863                 CERROR("%s: cannot get reverse import %s security: rc = %d\n",
864                         revimp->imp_client->cli_name,
865                         libcfs_id2str(req->rq_peer), rc);
866                 return rc;
867         }
868
869         return 0;
870 }
871
872 /**
873  * Allocate a new reverse import for an export.
874  *
875  * \retval -errno in case error hit
876  * \retval 0 if reverse import correctly init
877  **/
878 int rev_import_init(struct obd_export *export)
879 {
880         struct obd_device *obd = export->exp_obd;
881         struct obd_import *revimp;
882
883         LASSERT(export->exp_imp_reverse == NULL);
884
885         revimp = class_new_import(obd);
886         if (revimp == NULL)
887                 return -ENOMEM;
888
889         revimp->imp_remote_handle.cookie = 0ULL;
890         revimp->imp_client = &obd->obd_ldlm_client;
891         revimp->imp_dlm_fake = 1;
892
893         /* it is safe to connect import in new state as no sends possible */
894         spin_lock(&export->exp_lock);
895         export->exp_imp_reverse = revimp;
896         spin_unlock(&export->exp_lock);
897         class_import_put(revimp);
898
899         return 0;
900 }
901 EXPORT_SYMBOL(rev_import_init);
902
903 /**
904  * Handle reconnect for an export.
905  *
906  * \param exp export to handle reconnect process
907  * \param req client reconnect request
908  *
909  * \retval -rc in case securitfy flavor can't be changed
910  * \retval 0 in case none problems
911  */
912 static int rev_import_reconnect(struct obd_export *exp,
913                                 struct ptlrpc_request *req)
914 {
915         struct obd_import *revimp = exp->exp_imp_reverse;
916         struct lustre_handle *lh;
917         int rc;
918
919         /* avoid sending a request until import flags are changed */
920         ptlrpc_import_enter_resend(revimp);
921
922         if (revimp->imp_connection != NULL)
923                 ptlrpc_connection_put(revimp->imp_connection);
924
925         /*
926          * client from recovery don't have a handle so we need to take from
927          * request. it may produce situation when wrong client connected
928          * to recovery as we trust a client uuid
929          */
930         lh = req_capsule_client_get(&req->rq_pill, &RMF_CONN);
931         revimp->imp_remote_handle = *lh;
932
933         /* unknown versions will be caught in
934          * ptlrpc_handle_server_req_in->lustre_unpack_msg() */
935         revimp->imp_msg_magic = req->rq_reqmsg->lm_magic;
936
937         revimp->imp_connection = ptlrpc_connection_addref(exp->exp_connection);
938
939         rc = rev_import_flags_update(revimp, req);
940         if (rc != 0) {
941                 /* it is safe to still be in RECOVERY phase as we are not able
942                  * to setup correct security flavor so requests are not able to
943                  * be delivered correctly */
944                 return rc;
945         }
946
947         /* resend all rpc's via new connection */
948         return ptlrpc_import_recovery_state_machine(revimp);
949 }
950
951 int target_handle_connect(struct ptlrpc_request *req)
952 {
953         struct obd_device *target = NULL;
954         struct obd_export *export = NULL;
955         /* connect handle - filled from target_handle_reconnect in
956          * reconnect case */
957         struct lustre_handle conn;
958         struct lustre_handle *tmp;
959         struct obd_uuid cluuid;
960         char *str;
961         int rc = 0;
962         char *target_start;
963         int target_len;
964         bool     mds_conn = false, lw_client = false, initial_conn = false;
965         bool     mds_mds_conn = false;
966         bool     new_mds_mds_conn = false;
967         struct obd_connect_data *data, *tmpdata;
968         int size, tmpsize;
969         lnet_nid_t *client_nid = NULL;
970         ENTRY;
971
972         OBD_RACE(OBD_FAIL_TGT_CONN_RACE);
973
974         str = req_capsule_client_get(&req->rq_pill, &RMF_TGTUUID);
975         if (str == NULL) {
976                 DEBUG_REQ(D_ERROR, req, "bad target UUID for connect");
977                 GOTO(out, rc = -EINVAL);
978         }
979
980         target = class_dev_by_str(str);
981         if (!target) {
982                 deuuidify(str, NULL, &target_start, &target_len);
983                 LCONSOLE_ERROR_MSG(0x137, "%s: not available for connect "
984                                    "from %s (no target). If you are running "
985                                    "an HA pair check that the target is "
986                                    "mounted on the other server.\n", str,
987                                    libcfs_nid2str(req->rq_peer.nid));
988                 GOTO(out, rc = -ENODEV);
989         }
990
991         spin_lock(&target->obd_dev_lock);
992
993         target->obd_conn_inprogress++;
994
995         if (target->obd_stopping || !target->obd_set_up) {
996                 spin_unlock(&target->obd_dev_lock);
997
998                 deuuidify(str, NULL, &target_start, &target_len);
999                 LCONSOLE_INFO("%.*s: Not available for connect from %s (%s)\n",
1000                               target_len, target_start,
1001                               libcfs_nid2str(req->rq_peer.nid),
1002                               (target->obd_stopping ?
1003                                "stopping" : "not set up"));
1004                 GOTO(out, rc = -ENODEV);
1005         }
1006
1007         if (target->obd_no_conn) {
1008                 spin_unlock(&target->obd_dev_lock);
1009
1010                 CDEBUG(D_INFO, "%s: Temporarily refusing client connection "
1011                                "from %s\n", target->obd_name,
1012                                libcfs_nid2str(req->rq_peer.nid));
1013                 GOTO(out, rc = -EAGAIN);
1014         }
1015
1016         spin_unlock(&target->obd_dev_lock);
1017
1018         str = req_capsule_client_get(&req->rq_pill, &RMF_CLUUID);
1019         if (str == NULL) {
1020                 DEBUG_REQ(D_ERROR, req, "bad client UUID for connect");
1021                 GOTO(out, rc = -EINVAL);
1022         }
1023
1024         obd_str2uuid(&cluuid, str);
1025
1026         tmp = req_capsule_client_get(&req->rq_pill, &RMF_CONN);
1027         if (tmp == NULL)
1028                 GOTO(out, rc = -EPROTO);
1029
1030         conn = *tmp;
1031
1032         size = req_capsule_get_size(&req->rq_pill, &RMF_CONNECT_DATA,
1033                                     RCL_CLIENT);
1034         data = req_capsule_client_get(&req->rq_pill, &RMF_CONNECT_DATA);
1035         if (!data)
1036                 GOTO(out, rc = -EPROTO);
1037
1038         rc = req_capsule_server_pack(&req->rq_pill);
1039         if (rc)
1040                 GOTO(out, rc);
1041
1042 #if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(3, 0, 53, 0)
1043         /* Don't allow clients to connect that are using old 1.8 format
1044          * protocol conventions (LUSTRE_MSG_MAGIC_v1, !MSGHDR_CKSUM_INCOMPAT18,
1045          * ldlm_flock_policy_wire format, MDT_ATTR_xTIME_SET, etc).  The
1046          * FULL20 flag should be set on all connections since 2.0, but no
1047          * longer affects behaviour.
1048          *
1049          * Later this check will be disabled and the flag can be retired
1050          * completely once interop with 3.0 is no longer needed.
1051          */
1052         if (!(data->ocd_connect_flags & OBD_CONNECT_FULL20))
1053                 GOTO(out, rc = -EPROTO);
1054
1055         /* Don't allow liblustre clients to connect.
1056          * - testing was disabled in v2_2_50_0-61-g6a75d65
1057          * - building was disabled in v2_5_58_0-28-g7277179
1058          * - client code was deleted in v2_6_50_0-101-gcdfbc72,
1059          * - clients were refused connect for version difference > 0.0.1.32  */
1060         if (lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_LIBCLIENT) {
1061                 DEBUG_REQ(D_WARNING, req, "Refusing libclient connection");
1062                 GOTO(out, rc = -EPROTO);
1063         }
1064 #endif
1065
1066         /* Note: lw_client is needed in MDS-MDS failover during update log
1067          * processing, so we needs to allow lw_client to be connected at
1068          * anytime, instead of only the initial connection */
1069         lw_client = (data->ocd_connect_flags & OBD_CONNECT_LIGHTWEIGHT) != 0;
1070
1071         if (lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_INITIAL) {
1072                 initial_conn = true;
1073                 mds_conn = (data->ocd_connect_flags & OBD_CONNECT_MDS) != 0;
1074                 mds_mds_conn = (data->ocd_connect_flags &
1075                                 OBD_CONNECT_MDS_MDS) != 0;
1076
1077                 /* OBD_CONNECT_MNE_SWAB is defined as OBD_CONNECT_MDS_MDS
1078                  * for Imperative Recovery connection from MGC to MGS.
1079                  *
1080                  * Via check OBD_CONNECT_FID, we can distinguish whether
1081                  * the OBD_CONNECT_MDS_MDS/OBD_CONNECT_MNE_SWAB is from
1082                  * MGC or MDT, since MGC does not use OBD_CONNECT_FID.
1083                  */
1084                 if (!lw_client &&
1085                     (data->ocd_connect_flags & OBD_CONNECT_MDS_MDS) &&
1086                     (data->ocd_connect_flags & OBD_CONNECT_FID) &&
1087                     (data->ocd_connect_flags & OBD_CONNECT_VERSION)) {
1088                         __u32 major = OBD_OCD_VERSION_MAJOR(data->ocd_version);
1089                         __u32 minor = OBD_OCD_VERSION_MINOR(data->ocd_version);
1090                         __u32 patch = OBD_OCD_VERSION_PATCH(data->ocd_version);
1091
1092                         /* We do not support the MDT-MDT interoperations with
1093                          * different version MDT because of protocol changes. */
1094                         if (unlikely(major != LUSTRE_MAJOR ||
1095                                      minor != LUSTRE_MINOR ||
1096                                      abs(patch - LUSTRE_PATCH) > 3)) {
1097                                 LCONSOLE_WARN("%s (%u.%u.%u.%u) refused the "
1098                                         "connection from different version MDT "
1099                                         "(%d.%d.%d.%d) %s %s\n",
1100                                         target->obd_name, LUSTRE_MAJOR,
1101                                         LUSTRE_MINOR, LUSTRE_PATCH, LUSTRE_FIX,
1102                                         major, minor, patch,
1103                                         OBD_OCD_VERSION_FIX(data->ocd_version),
1104                                         libcfs_nid2str(req->rq_peer.nid), str);
1105
1106                                 GOTO(out, rc = -EPROTO);
1107                         }
1108                 }
1109         }
1110
1111         /* lctl gets a backstage, all-access pass. */
1112         if (obd_uuid_equals(&cluuid, &target->obd_uuid))
1113                 goto dont_check_exports;
1114
1115         export = cfs_hash_lookup(target->obd_uuid_hash, &cluuid);
1116         if (!export)
1117                 goto no_export;
1118
1119         /* We've found an export in the hash. */
1120
1121         spin_lock(&export->exp_lock);
1122
1123         if (export->exp_connecting) { /* bug 9635, et. al. */
1124                 spin_unlock(&export->exp_lock);
1125                 LCONSOLE_WARN("%s: Export %p already connecting from %s\n",
1126                               export->exp_obd->obd_name, export,
1127                               libcfs_nid2str(req->rq_peer.nid));
1128                 class_export_put(export);
1129                 export = NULL;
1130                 rc = -EALREADY;
1131         } else if ((mds_conn || (lw_client && initial_conn) ||
1132                    data->ocd_connect_flags & OBD_CONNECT_MDS_MDS) &&
1133                    export->exp_connection != NULL) {
1134                 spin_unlock(&export->exp_lock);
1135                 if (req->rq_peer.nid != export->exp_connection->c_peer.nid) {
1136                         /* MDS or LWP reconnected after failover. */
1137                         LCONSOLE_WARN("%s: Received %s connection from "
1138                             "%s, removing former export from %s\n",
1139                             target->obd_name, mds_conn ? "MDS" : "LWP",
1140                             libcfs_nid2str(req->rq_peer.nid),
1141                             libcfs_nid2str(export->exp_connection->c_peer.nid));
1142                 } else {
1143                         /* New MDS connection from the same NID. */
1144                         LCONSOLE_WARN("%s: Received new %s connection from "
1145                                 "%s, removing former export from same NID\n",
1146                                 target->obd_name, mds_conn ? "MDS" : "LWP",
1147                                 libcfs_nid2str(req->rq_peer.nid));
1148                 }
1149
1150                 if (req->rq_peer.nid == export->exp_connection->c_peer.nid &&
1151                     data->ocd_connect_flags & OBD_CONNECT_MDS_MDS) {
1152                         /* Because exports between MDTs will always be
1153                          * kept, let's do not fail such export if they
1154                          * come from the same NID, otherwise it might
1155                          * cause eviction between MDTs, which might
1156                          * cause namespace inconsistency */
1157                         spin_lock(&export->exp_lock);
1158                         export->exp_connecting = 1;
1159                         spin_unlock(&export->exp_lock);
1160                         conn.cookie = export->exp_handle.h_cookie;
1161                         rc = EALREADY;
1162                 } else {
1163                         class_fail_export(export);
1164                         class_export_put(export);
1165                         export = NULL;
1166                         rc = 0;
1167                 }
1168         } else if (export->exp_connection != NULL && initial_conn &&
1169                    req->rq_peer.nid != export->exp_connection->c_peer.nid) {
1170                 spin_unlock(&export->exp_lock);
1171                 /* In MDS failover we have static UUID but NID can change. */
1172                 LCONSOLE_WARN("%s: Client %s seen on new nid %s when "
1173                               "existing nid %s is already connected\n",
1174                               target->obd_name, cluuid.uuid,
1175                               libcfs_nid2str(req->rq_peer.nid),
1176                               libcfs_nid2str(
1177                                       export->exp_connection->c_peer.nid));
1178                 rc = -EALREADY;
1179                 class_export_put(export);
1180                 export = NULL;
1181         } else {
1182                 export->exp_connecting = 1;
1183                 spin_unlock(&export->exp_lock);
1184                 LASSERT(export->exp_obd == target);
1185
1186                 rc = target_handle_reconnect(&conn, export, &cluuid);
1187         }
1188
1189         /* If we found an export, we already unlocked. */
1190         if (!export) {
1191 no_export:
1192                 OBD_FAIL_TIMEOUT(OBD_FAIL_TGT_DELAY_CONNECT, 2 * obd_timeout);
1193         } else if (req->rq_export == NULL &&
1194                    atomic_read(&export->exp_rpc_count) > 0) {
1195                 LCONSOLE_WARN("%s: Client %s (at %s) refused connection, "
1196                               "still busy with %d references\n",
1197                               target->obd_name, cluuid.uuid,
1198                               libcfs_nid2str(req->rq_peer.nid),
1199                               atomic_read(&export->exp_refcount));
1200                 GOTO(out, rc = -EBUSY);
1201         } else if (lustre_msg_get_conn_cnt(req->rq_reqmsg) == 1) {
1202                 if (!strstr(cluuid.uuid, "mdt"))
1203                         LCONSOLE_WARN("%s: Rejecting reconnect from the "
1204                                       "known client %s (at %s) because it "
1205                                       "is indicating it is a new client",
1206                                       target->obd_name, cluuid.uuid,
1207                                       libcfs_nid2str(req->rq_peer.nid));
1208                 GOTO(out, rc = -EALREADY);
1209         } else {
1210                 OBD_FAIL_TIMEOUT(OBD_FAIL_TGT_DELAY_RECONNECT, 2 * obd_timeout);
1211         }
1212
1213         if (rc < 0) {
1214                 GOTO(out, rc);
1215         }
1216
1217         CDEBUG(D_HA, "%s: connection from %s@%s %st%llu exp %p cur %lld last %lld\n",
1218                target->obd_name, cluuid.uuid, libcfs_nid2str(req->rq_peer.nid),
1219                target->obd_recovering ? "recovering/" : "", data->ocd_transno,
1220                export, ktime_get_real_seconds(),
1221                export ? export->exp_last_request_time : 0);
1222
1223         /* If this is the first time a client connects, reset the recovery
1224          * timer. Discard lightweight connections which might be local. */
1225         if (!lw_client && rc == 0 && target->obd_recovering)
1226                 check_and_start_recovery_timer(target, req, export == NULL);
1227
1228         /* We want to handle EALREADY but *not* -EALREADY from
1229          * target_handle_reconnect(), return reconnection state in a flag. */
1230         if (rc == EALREADY) {
1231                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_RECONNECT);
1232                 rc = 0;
1233         } else {
1234                 LASSERT(rc == 0);
1235         }
1236
1237         /* Tell the client if we support replayable requests. */
1238         if (target->obd_replayable)
1239                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_REPLAYABLE);
1240         client_nid = &req->rq_peer.nid;
1241
1242         if (export == NULL) {
1243                 /* allow lightweight connections during recovery */
1244                 /* allow "new" MDT to be connected during recovery, since we
1245                  * need retrieve recovery update records from it */
1246                 if (target->obd_recovering && !lw_client && !mds_mds_conn) {
1247                         time64_t t;
1248                         int c; /* connected */
1249                         int i; /* in progress */
1250                         int k; /* known */
1251                         int s; /* stale/evicted */
1252
1253                         c = atomic_read(&target->obd_connected_clients);
1254                         i = atomic_read(&target->obd_lock_replay_clients);
1255                         k = target->obd_max_recoverable_clients;
1256                         s = target->obd_stale_clients;
1257                         t = target->obd_recovery_timer.expires;
1258                         t = cfs_duration_sec(target->obd_recovery_timer.expires);
1259                         t -= ktime_get_seconds();
1260                         LCONSOLE_WARN("%s: Denying connection for new client %s"
1261                                       "(at %s), waiting for %d known clients "
1262                                       "(%d recovered, %d in progress, and %d "
1263                                       "evicted) to recover in %lld:%.02lld\n",
1264                                       target->obd_name, cluuid.uuid,
1265                                       libcfs_nid2str(req->rq_peer.nid), k,
1266                                       c - i, i, s, t / 60, t % 60);
1267                         rc = -EBUSY;
1268                 } else {
1269 dont_check_exports:
1270                         rc = obd_connect(req->rq_svc_thread->t_env,
1271                                          &export, target, &cluuid, data,
1272                                          client_nid);
1273                         if (mds_conn && OBD_FAIL_CHECK(OBD_FAIL_TGT_RCVG_FLAG))
1274                                 lustre_msg_add_op_flags(req->rq_repmsg,
1275                                                         MSG_CONNECT_RECOVERING);
1276                         if (rc == 0) {
1277                                 conn.cookie = export->exp_handle.h_cookie;
1278                                 rc = rev_import_init(export);
1279                         }
1280
1281                         if (mds_mds_conn)
1282                                 new_mds_mds_conn = true;
1283                 }
1284         } else {
1285                 rc = obd_reconnect(req->rq_svc_thread->t_env,
1286                                    export, target, &cluuid, data, client_nid);
1287         }
1288         if (rc)
1289                 GOTO(out, rc);
1290
1291         LASSERT(target->u.obt.obt_magic == OBT_MAGIC);
1292         data->ocd_instance = target->u.obt.obt_instance;
1293
1294         /* Return only the parts of obd_connect_data that we understand, so the
1295          * client knows that we don't understand the rest. */
1296         if (data) {
1297                 tmpsize = req_capsule_get_size(&req->rq_pill, &RMF_CONNECT_DATA,
1298                                                RCL_SERVER);
1299                 tmpdata = req_capsule_server_get(&req->rq_pill,
1300                                                  &RMF_CONNECT_DATA);
1301                 /* Don't use struct assignment here, because the client reply
1302                  * buffer may be smaller/larger than the local struct
1303                  * obd_connect_data. */
1304                 memcpy(tmpdata, data, min(tmpsize, size));
1305         }
1306
1307         /* If the client and the server are the same node, we will already
1308          * have an export that really points to the client's DLM export,
1309          * because we have a shared handles table.
1310          *
1311          * XXX this will go away when shaver stops sending the "connect" handle
1312          * in the real "remote handle" field of the request --phik 24 Apr 2003
1313          */
1314         ptlrpc_request_change_export(req, export);
1315
1316         spin_lock(&export->exp_lock);
1317         if (export->exp_conn_cnt >= lustre_msg_get_conn_cnt(req->rq_reqmsg)) {
1318                 spin_unlock(&export->exp_lock);
1319                 CDEBUG(D_RPCTRACE, "%s: %s already connected at greater "
1320                        "or equal conn_cnt: %d >= %d\n",
1321                        cluuid.uuid, libcfs_nid2str(req->rq_peer.nid),
1322                        export->exp_conn_cnt,
1323                        lustre_msg_get_conn_cnt(req->rq_reqmsg));
1324
1325                 GOTO(out, rc = -EALREADY);
1326         }
1327         LASSERT(lustre_msg_get_conn_cnt(req->rq_reqmsg) > 0);
1328         export->exp_conn_cnt = lustre_msg_get_conn_cnt(req->rq_reqmsg);
1329         spin_unlock(&export->exp_lock);
1330
1331         if (export->exp_connection != NULL) {
1332                 /* Check to see if connection came from another NID. */
1333                 if ((export->exp_connection->c_peer.nid != req->rq_peer.nid) &&
1334                     !hlist_unhashed(&export->exp_nid_hash))
1335                         cfs_hash_del(export->exp_obd->obd_nid_hash,
1336                                      &export->exp_connection->c_peer.nid,
1337                                      &export->exp_nid_hash);
1338
1339                 ptlrpc_connection_put(export->exp_connection);
1340         }
1341
1342         export->exp_connection = ptlrpc_connection_get(req->rq_peer,
1343                                                        req->rq_self,
1344                                                        &cluuid);
1345         if (hlist_unhashed(&export->exp_nid_hash))
1346                 cfs_hash_add(export->exp_obd->obd_nid_hash,
1347                              &export->exp_connection->c_peer.nid,
1348                              &export->exp_nid_hash);
1349
1350         lustre_msg_set_handle(req->rq_repmsg, &conn);
1351
1352         rc = rev_import_reconnect(export, req);
1353         if (rc != 0)
1354                 GOTO(out, rc);
1355
1356         if (target->obd_recovering && !export->exp_in_recovery && !lw_client) {
1357                 int has_transno;
1358                 __u64 transno = data->ocd_transno;
1359
1360                 spin_lock(&export->exp_lock);
1361                 /* possible race with class_disconnect_stale_exports,
1362                  * export may be already in the eviction process */
1363                 if (export->exp_failed) {
1364                         spin_unlock(&export->exp_lock);
1365                         GOTO(out, rc = -ENODEV);
1366                 }
1367                 export->exp_in_recovery = 1;
1368                 export->exp_req_replay_needed = 1;
1369                 export->exp_lock_replay_needed = 1;
1370                 spin_unlock(&export->exp_lock);
1371
1372                 has_transno = !!(lustre_msg_get_op_flags(req->rq_reqmsg) &
1373                                  MSG_CONNECT_TRANSNO);
1374                 if (has_transno && transno == 0)
1375                         CWARN("Connect with zero transno!\n");
1376
1377                 if (has_transno && transno > 0 &&
1378                     transno < target->obd_next_recovery_transno &&
1379                     transno > target->obd_last_committed) {
1380                         /* Another way is to use cmpxchg() to be lock-free. */
1381                         spin_lock(&target->obd_recovery_task_lock);
1382                         if (transno < target->obd_next_recovery_transno)
1383                                 target->obd_next_recovery_transno = transno;
1384                         spin_unlock(&target->obd_recovery_task_lock);
1385                 }
1386
1387                 atomic_inc(&target->obd_req_replay_clients);
1388                 atomic_inc(&target->obd_lock_replay_clients);
1389                 /* Note: MDS-MDS connection is allowed to be connected during
1390                  * recovery, no matter if the exports needs to be recoveried.
1391                  * Because we need retrieve updates logs from all other MDTs.
1392                  * So if the MDS-MDS export is new, obd_max_recoverable_clients
1393                  * also needs to be increased to match other recovery checking
1394                  * condition. */
1395                 if (new_mds_mds_conn)
1396                         target->obd_max_recoverable_clients++;
1397                 if (atomic_inc_return(&target->obd_connected_clients) ==
1398                     target->obd_max_recoverable_clients)
1399                         wake_up(&target->obd_next_transno_waitq);
1400         }
1401
1402         /* Tell the client we're in recovery, when client is involved in it. */
1403         if (target->obd_recovering && !lw_client)
1404                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_RECOVERING);
1405
1406 out:
1407         if (export) {
1408                 spin_lock(&export->exp_lock);
1409                 export->exp_connecting = 0;
1410                 spin_unlock(&export->exp_lock);
1411
1412                 class_export_put(export);
1413         }
1414         if (target != NULL) {
1415                 spin_lock(&target->obd_dev_lock);
1416                 target->obd_conn_inprogress--;
1417                 spin_unlock(&target->obd_dev_lock);
1418                 class_decref(target, "find", current);
1419         }
1420         req->rq_status = rc;
1421         RETURN(rc);
1422 }
1423
1424 int target_handle_disconnect(struct ptlrpc_request *req)
1425 {
1426         int rc;
1427         ENTRY;
1428
1429         rc = req_capsule_server_pack(&req->rq_pill);
1430         if (rc)
1431                 RETURN(rc);
1432
1433         /* Keep the rq_export around so we can send the reply. */
1434         req->rq_status = obd_disconnect(class_export_get(req->rq_export));
1435
1436         RETURN(0);
1437 }
1438
1439 void target_destroy_export(struct obd_export *exp)
1440 {
1441         struct obd_import       *imp = NULL;
1442         /* exports created from last_rcvd data, and "fake"
1443            exports created by lctl don't have an import */
1444         spin_lock(&exp->exp_lock);
1445         if (exp->exp_imp_reverse != NULL) {
1446                 imp = exp->exp_imp_reverse;
1447                 exp->exp_imp_reverse = NULL;
1448         }
1449         spin_unlock(&exp->exp_lock);
1450         if (imp != NULL)
1451                 client_destroy_import(imp);
1452
1453         LASSERT_ATOMIC_ZERO(&exp->exp_locks_count);
1454         LASSERT_ATOMIC_ZERO(&exp->exp_rpc_count);
1455         LASSERT_ATOMIC_ZERO(&exp->exp_cb_count);
1456         LASSERT_ATOMIC_ZERO(&exp->exp_replay_count);
1457 }
1458 EXPORT_SYMBOL(target_destroy_export);
1459
1460 /*
1461  * Recovery functions
1462  */
1463 static void target_request_copy_get(struct ptlrpc_request *req)
1464 {
1465         class_export_rpc_inc(req->rq_export);
1466         LASSERT(list_empty(&req->rq_list));
1467         INIT_LIST_HEAD(&req->rq_replay_list);
1468
1469         /* Increase refcount to keep request in queue. */
1470         atomic_inc(&req->rq_refcount);
1471         /* Let export know it has replays to be handled. */
1472         atomic_inc(&req->rq_export->exp_replay_count);
1473 }
1474
1475 static void target_request_copy_put(struct ptlrpc_request *req)
1476 {
1477         LASSERT(list_empty(&req->rq_replay_list));
1478         LASSERT_ATOMIC_POS(&req->rq_export->exp_replay_count);
1479
1480         atomic_dec(&req->rq_export->exp_replay_count);
1481         class_export_rpc_dec(req->rq_export);
1482         ptlrpc_server_drop_request(req);
1483 }
1484
1485 static int target_exp_enqueue_req_replay(struct ptlrpc_request *req)
1486 {
1487         __u64                  transno = lustre_msg_get_transno(req->rq_reqmsg);
1488         struct obd_export     *exp = req->rq_export;
1489         struct ptlrpc_request *reqiter;
1490         struct ptlrpc_request *dup_req = NULL;
1491         int                    dup = 0;
1492
1493         LASSERT(exp);
1494
1495         spin_lock(&exp->exp_lock);
1496         list_for_each_entry(reqiter, &exp->exp_req_replay_queue,
1497                                 rq_replay_list) {
1498                 if (lustre_msg_get_transno(reqiter->rq_reqmsg) == transno) {
1499                         dup_req = reqiter;
1500                         dup = 1;
1501                         break;
1502                 }
1503         }
1504
1505         if (dup) {
1506                 /* We expect it with RESENT and REPLAY flags. */
1507                 if ((lustre_msg_get_flags(req->rq_reqmsg) &
1508                      (MSG_RESENT | MSG_REPLAY)) != (MSG_RESENT | MSG_REPLAY))
1509                         CERROR("invalid flags %x of resent replay\n",
1510                                lustre_msg_get_flags(req->rq_reqmsg));
1511
1512                 if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) {
1513                         __u32 new_conn;
1514
1515                         new_conn = lustre_msg_get_conn_cnt(req->rq_reqmsg);
1516                         if (new_conn >
1517                             lustre_msg_get_conn_cnt(dup_req->rq_reqmsg))
1518                                 lustre_msg_set_conn_cnt(dup_req->rq_reqmsg,
1519                                                         new_conn);
1520                 }
1521         } else {
1522                 list_add_tail(&req->rq_replay_list,
1523                                   &exp->exp_req_replay_queue);
1524         }
1525
1526         spin_unlock(&exp->exp_lock);
1527         return dup;
1528 }
1529
1530 static void target_exp_dequeue_req_replay(struct ptlrpc_request *req)
1531 {
1532         LASSERT(!list_empty(&req->rq_replay_list));
1533         LASSERT(req->rq_export);
1534
1535         spin_lock(&req->rq_export->exp_lock);
1536         list_del_init(&req->rq_replay_list);
1537         spin_unlock(&req->rq_export->exp_lock);
1538 }
1539
1540 static void target_finish_recovery(struct lu_target *lut)
1541 {
1542         struct obd_device *obd = lut->lut_obd;
1543         ENTRY;
1544
1545         /* Only log a recovery message when recovery has occurred. */
1546         if (obd->obd_recovery_start) {
1547                 time64_t now = ktime_get_real_seconds();
1548                 time64_t elapsed_time;
1549
1550                 elapsed_time = max_t(time64_t, now - obd->obd_recovery_start, 1);
1551                 LCONSOLE_INFO("%s: Recovery over after %lld:%.02lld, of %d clients "
1552                         "%d recovered and %d %s evicted.\n", obd->obd_name,
1553                         (s64)elapsed_time / 60, (s64)elapsed_time % 60,
1554                         obd->obd_max_recoverable_clients,
1555                         atomic_read(&obd->obd_connected_clients),
1556                         obd->obd_stale_clients,
1557                         obd->obd_stale_clients == 1 ? "was" : "were");
1558         }
1559
1560         ldlm_reprocess_recovery_done(obd->obd_namespace);
1561         spin_lock(&obd->obd_recovery_task_lock);
1562         if (!list_empty(&obd->obd_req_replay_queue) ||
1563             !list_empty(&obd->obd_lock_replay_queue) ||
1564             !list_empty(&obd->obd_final_req_queue)) {
1565                 CERROR("%s: Recovery queues ( %s%s%s) are not empty\n",
1566                        obd->obd_name,
1567                        list_empty(&obd->obd_req_replay_queue) ? "" : "req ",
1568                        list_empty(&obd->obd_lock_replay_queue) ? \
1569                                "" : "lock ",
1570                        list_empty(&obd->obd_final_req_queue) ? \
1571                                "" : "final ");
1572                 spin_unlock(&obd->obd_recovery_task_lock);
1573                 LBUG();
1574         }
1575         spin_unlock(&obd->obd_recovery_task_lock);
1576
1577         obd->obd_recovery_end = ktime_get_real_seconds();
1578
1579         /* When recovery finished, cleanup orphans on MDS and OST. */
1580         if (obd->obd_type && OBP(obd, postrecov)) {
1581                 int rc = OBP(obd, postrecov)(obd);
1582
1583                 if (rc < 0)
1584                         LCONSOLE_WARN("%s: Post recovery failed, rc %d\n",
1585                                       obd->obd_name, rc);
1586         }
1587         EXIT;
1588 }
1589
1590 static void abort_req_replay_queue(struct obd_device *obd)
1591 {
1592         struct ptlrpc_request *req, *n;
1593         struct list_head abort_list;
1594
1595         INIT_LIST_HEAD(&abort_list);
1596         spin_lock(&obd->obd_recovery_task_lock);
1597         list_splice_init(&obd->obd_req_replay_queue, &abort_list);
1598         spin_unlock(&obd->obd_recovery_task_lock);
1599         list_for_each_entry_safe(req, n, &abort_list, rq_list) {
1600                 DEBUG_REQ(D_WARNING, req, "aborted:");
1601                 req->rq_status = -ENOTCONN;
1602                 if (ptlrpc_error(req)) {
1603                         DEBUG_REQ(D_ERROR, req,
1604                                   "failed abort_req_reply; skipping");
1605                 }
1606                 target_exp_dequeue_req_replay(req);
1607                 target_request_copy_put(req);
1608         }
1609 }
1610
1611 static void abort_lock_replay_queue(struct obd_device *obd)
1612 {
1613         struct ptlrpc_request *req, *n;
1614         struct list_head abort_list;
1615
1616         INIT_LIST_HEAD(&abort_list);
1617         spin_lock(&obd->obd_recovery_task_lock);
1618         list_splice_init(&obd->obd_lock_replay_queue, &abort_list);
1619         spin_unlock(&obd->obd_recovery_task_lock);
1620         list_for_each_entry_safe(req, n, &abort_list, rq_list) {
1621                 DEBUG_REQ(D_ERROR, req, "aborted:");
1622                 req->rq_status = -ENOTCONN;
1623                 if (ptlrpc_error(req)) {
1624                         DEBUG_REQ(D_ERROR, req,
1625                                   "failed abort_lock_reply; skipping");
1626                 }
1627                 target_request_copy_put(req);
1628         }
1629 }
1630
1631 /* Called from a cleanup function if the device is being cleaned up
1632    forcefully.  The exports should all have been disconnected already,
1633    the only thing left to do is
1634      - clear the recovery flags
1635      - cancel the timer
1636      - free queued requests and replies, but don't send replies
1637    Because the obd_stopping flag is set, no new requests should be received.
1638
1639 */
1640 void target_cleanup_recovery(struct obd_device *obd)
1641 {
1642         struct ptlrpc_request *req, *n;
1643         struct list_head clean_list;
1644
1645         INIT_LIST_HEAD(&clean_list);
1646         spin_lock(&obd->obd_dev_lock);
1647         if (!obd->obd_recovering) {
1648                 spin_unlock(&obd->obd_dev_lock);
1649                 EXIT;
1650                 return;
1651         }
1652         obd->obd_recovering = obd->obd_abort_recovery = 0;
1653         spin_unlock(&obd->obd_dev_lock);
1654
1655         spin_lock(&obd->obd_recovery_task_lock);
1656         target_cancel_recovery_timer(obd);
1657         list_splice_init(&obd->obd_req_replay_queue, &clean_list);
1658         spin_unlock(&obd->obd_recovery_task_lock);
1659
1660         list_for_each_entry_safe(req, n, &clean_list, rq_list) {
1661                 LASSERT(req->rq_reply_state == NULL);
1662                 target_exp_dequeue_req_replay(req);
1663                 target_request_copy_put(req);
1664         }
1665
1666         spin_lock(&obd->obd_recovery_task_lock);
1667         list_splice_init(&obd->obd_lock_replay_queue, &clean_list);
1668         list_splice_init(&obd->obd_final_req_queue, &clean_list);
1669         spin_unlock(&obd->obd_recovery_task_lock);
1670
1671         list_for_each_entry_safe(req, n, &clean_list, rq_list) {
1672                 LASSERT(req->rq_reply_state == NULL);
1673                 target_request_copy_put(req);
1674         }
1675
1676         EXIT;
1677 }
1678 EXPORT_SYMBOL(target_cleanup_recovery);
1679
1680 /* obd_recovery_task_lock should be held */
1681 void target_cancel_recovery_timer(struct obd_device *obd)
1682 {
1683         CDEBUG(D_HA, "%s: cancel recovery timer\n", obd->obd_name);
1684         del_timer(&obd->obd_recovery_timer);
1685 }
1686
1687 static void target_start_recovery_timer(struct obd_device *obd)
1688 {
1689         if (obd->obd_recovery_start != 0)
1690                 return;
1691
1692         spin_lock(&obd->obd_dev_lock);
1693         if (!obd->obd_recovering || obd->obd_abort_recovery) {
1694                 spin_unlock(&obd->obd_dev_lock);
1695                 return;
1696         }
1697
1698         LASSERT(obd->obd_recovery_timeout != 0);
1699
1700         if (obd->obd_recovery_start != 0) {
1701                 spin_unlock(&obd->obd_dev_lock);
1702                 return;
1703         }
1704
1705         mod_timer(&obd->obd_recovery_timer,
1706                   jiffies + cfs_time_seconds(obd->obd_recovery_timeout));
1707         obd->obd_recovery_start = ktime_get_real_seconds();
1708         spin_unlock(&obd->obd_dev_lock);
1709
1710         LCONSOLE_WARN("%s: Will be in recovery for at least %llu:%02llu, or until %d client%s reconnect%s\n",
1711                       obd->obd_name,
1712                       obd->obd_recovery_timeout / 60,
1713                       obd->obd_recovery_timeout % 60,
1714                       obd->obd_max_recoverable_clients,
1715                       (obd->obd_max_recoverable_clients == 1) ? "" : "s",
1716                       (obd->obd_max_recoverable_clients == 1) ? "s": "");
1717 }
1718
1719 /**
1720  * extend recovery window.
1721  *
1722  * if @extend is true, extend recovery window to have @drt remaining at least;
1723  * otherwise, make sure the recovery timeout value is not less than @drt.
1724  */
1725 static void extend_recovery_timer(struct obd_device *obd, time64_t drt,
1726                                   bool extend)
1727 {
1728         time64_t now;
1729         time64_t end;
1730         time64_t left;
1731         time64_t to;
1732
1733         spin_lock(&obd->obd_dev_lock);
1734         if (!obd->obd_recovering || obd->obd_abort_recovery) {
1735                 spin_unlock(&obd->obd_dev_lock);
1736                 return;
1737         }
1738         LASSERT(obd->obd_recovery_start != 0);
1739
1740         now = ktime_get_real_seconds();
1741         to = obd->obd_recovery_timeout;
1742         end = obd->obd_recovery_start + to;
1743         left = end - now;
1744
1745         if (extend && (drt > left)) {
1746                 to += drt - left;
1747         } else if (!extend && (drt > to)) {
1748                 to = drt;
1749         }
1750
1751         if (to > obd->obd_recovery_time_hard) {
1752                 to = obd->obd_recovery_time_hard;
1753                 CWARN("%s: extended recovery timer reaching hard limit: %lld, extend: %d\n",
1754                       obd->obd_name, to, extend);
1755         }
1756
1757         if (obd->obd_recovery_timeout < to) {
1758                 obd->obd_recovery_timeout = to;
1759                 end = obd->obd_recovery_start + to;
1760                 mod_timer(&obd->obd_recovery_timer,
1761                           jiffies + cfs_time_seconds(end - now));
1762         }
1763         spin_unlock(&obd->obd_dev_lock);
1764
1765         CDEBUG(D_HA, "%s: recovery timer will expire in %lld seconds\n",
1766                 obd->obd_name, (s64)(end - now));
1767 }
1768
1769 /* Reset the timer with each new client connection */
1770 /*
1771  * This timer is actually reconnect_timer, which is for making sure
1772  * the total recovery window is at least as big as my reconnect
1773  * attempt timing. So the initial recovery time_out will be set to
1774  * OBD_RECOVERY_FACTOR * obd_timeout. If the timeout coming
1775  * from client is bigger than this, then the recovery time_out will
1776  * be extended to make sure the client could be reconnected, in the
1777  * process, the timeout from the new client should be ignored.
1778  */
1779
1780 static void
1781 check_and_start_recovery_timer(struct obd_device *obd,
1782                                struct ptlrpc_request *req,
1783                                int new_client)
1784 {
1785         time64_t service_time = lustre_msg_get_service_time(req->rq_reqmsg);
1786         struct obd_device_target *obt = &obd->u.obt;
1787
1788         if (!new_client && service_time)
1789                 /* Teach server about old server's estimates, as first guess
1790                  * at how long new requests will take. */
1791                 at_measured(&req->rq_rqbd->rqbd_svcpt->scp_at_estimate,
1792                             service_time);
1793
1794         target_start_recovery_timer(obd);
1795
1796         /* Convert the service time to RPC timeout,
1797          * and reuse service_time to limit stack usage.
1798          */
1799         service_time = at_est2timeout(service_time);
1800
1801         if (OBD_FAIL_CHECK(OBD_FAIL_TGT_SLUGGISH_NET) &&
1802             service_time < at_extra)
1803                 service_time = at_extra;
1804
1805         /* We expect other clients to timeout within service_time, then try
1806          * to reconnect, then try the failover server.  The max delay between
1807          * connect attempts is SWITCH_MAX + SWITCH_INC + INITIAL. */
1808         service_time += 2 * INITIAL_CONNECT_TIMEOUT;
1809
1810         LASSERT(obt->obt_magic == OBT_MAGIC);
1811         service_time += 2 * (CONNECTION_SWITCH_MAX + CONNECTION_SWITCH_INC);
1812         if (service_time > obd->obd_recovery_timeout && !new_client)
1813                 extend_recovery_timer(obd, service_time, false);
1814 }
1815
1816 /** Health checking routines */
1817 static inline int exp_connect_healthy(struct obd_export *exp)
1818 {
1819         return (exp->exp_in_recovery);
1820 }
1821
1822 /** if export done req_replay or has replay in queue */
1823 static inline int exp_req_replay_healthy(struct obd_export *exp)
1824 {
1825         return (!exp->exp_req_replay_needed ||
1826                 atomic_read(&exp->exp_replay_count) > 0);
1827 }
1828
1829
1830 static inline int exp_req_replay_healthy_or_from_mdt(struct obd_export *exp)
1831 {
1832         return (exp_connect_flags(exp) & OBD_CONNECT_MDS_MDS) ||
1833                exp_req_replay_healthy(exp);
1834 }
1835
1836 /** if export done lock_replay or has replay in queue */
1837 static inline int exp_lock_replay_healthy(struct obd_export *exp)
1838 {
1839         return (!exp->exp_lock_replay_needed ||
1840                 atomic_read(&exp->exp_replay_count) > 0);
1841 }
1842
1843 static inline int exp_vbr_healthy(struct obd_export *exp)
1844 {
1845         return (!exp->exp_vbr_failed);
1846 }
1847
1848 static inline int exp_finished(struct obd_export *exp)
1849 {
1850         return (exp->exp_in_recovery && !exp->exp_lock_replay_needed);
1851 }
1852
1853 static inline int exp_finished_or_from_mdt(struct obd_export *exp)
1854 {
1855         return (exp_connect_flags(exp) & OBD_CONNECT_MDS_MDS) ||
1856                 exp_finished(exp);
1857 }
1858
1859 static int check_for_next_transno(struct lu_target *lut)
1860 {
1861         struct ptlrpc_request *req = NULL;
1862         struct obd_device *obd = lut->lut_obd;
1863         struct target_distribute_txn_data *tdtd = lut->lut_tdtd;
1864         int wake_up = 0, connected, completed, queue_len;
1865         __u64 req_transno = 0;
1866         __u64 update_transno = 0;
1867         __u64 next_transno = 0;
1868         ENTRY;
1869
1870         spin_lock(&obd->obd_recovery_task_lock);
1871         if (!list_empty(&obd->obd_req_replay_queue)) {
1872                 req = list_entry(obd->obd_req_replay_queue.next,
1873                                      struct ptlrpc_request, rq_list);
1874                 req_transno = lustre_msg_get_transno(req->rq_reqmsg);
1875         }
1876
1877         if (tdtd != NULL)
1878                 update_transno = distribute_txn_get_next_transno(tdtd);
1879
1880         connected = atomic_read(&obd->obd_connected_clients);
1881         completed = connected - atomic_read(&obd->obd_req_replay_clients);
1882         queue_len = obd->obd_requests_queued_for_recovery;
1883         next_transno = obd->obd_next_recovery_transno;
1884
1885         CDEBUG(D_HA, "max: %d, connected: %d, completed: %d, queue_len: %d, "
1886                "req_transno: %llu, next_transno: %llu\n",
1887                obd->obd_max_recoverable_clients, connected, completed,
1888                queue_len, req_transno, next_transno);
1889
1890         if (obd->obd_abort_recovery) {
1891                 CDEBUG(D_HA, "waking for aborted recovery\n");
1892                 wake_up = 1;
1893         } else if (obd->obd_recovery_expired) {
1894                 CDEBUG(D_HA, "waking for expired recovery\n");
1895                 wake_up = 1;
1896         } else if (tdtd != NULL && req != NULL &&
1897                    is_req_replayed_by_update(req)) {
1898                 LASSERTF(req_transno < next_transno, "req_transno %llu"
1899                          "next_transno%llu\n", req_transno, next_transno);
1900                 CDEBUG(D_HA, "waking for duplicate req (%llu)\n",
1901                        req_transno);
1902                 wake_up = 1;
1903         } else if (req_transno == next_transno ||
1904                    (update_transno != 0 && update_transno <= next_transno)) {
1905                 CDEBUG(D_HA, "waking for next (%lld)\n", next_transno);
1906                 wake_up = 1;
1907         } else if (queue_len > 0 &&
1908                    queue_len == atomic_read(&obd->obd_req_replay_clients)) {
1909                 /** handle gaps occured due to lost reply or VBR */
1910                 LASSERTF(req_transno >= next_transno,
1911                          "req_transno: %llu, next_transno: %llu\n",
1912                          req_transno, next_transno);
1913                 CDEBUG(D_HA,
1914                        "%s: waking for gap in transno, VBR is %s (skip: "
1915                        "%lld, ql: %d, comp: %d, conn: %d, next: %lld"
1916                        ", next_update %lld last_committed: %lld)\n",
1917                        obd->obd_name, obd->obd_version_recov ? "ON" : "OFF",
1918                        next_transno, queue_len, completed, connected,
1919                        req_transno, update_transno, obd->obd_last_committed);
1920                 obd->obd_next_recovery_transno = req_transno;
1921                 wake_up = 1;
1922         } else if (atomic_read(&obd->obd_req_replay_clients) == 0) {
1923                 CDEBUG(D_HA, "waking for completed recovery\n");
1924                 wake_up = 1;
1925         } else if (OBD_FAIL_CHECK(OBD_FAIL_MDS_RECOVERY_ACCEPTS_GAPS)) {
1926                 CDEBUG(D_HA, "accepting transno gaps is explicitly allowed"
1927                        " by fail_lock, waking up (%lld)\n", next_transno);
1928                 obd->obd_next_recovery_transno = req_transno;
1929                 wake_up = 1;
1930         }
1931         spin_unlock(&obd->obd_recovery_task_lock);
1932         return wake_up;
1933 }
1934
1935 static int check_for_next_lock(struct lu_target *lut)
1936 {
1937         struct obd_device *obd = lut->lut_obd;
1938         int wake_up = 0;
1939
1940         spin_lock(&obd->obd_recovery_task_lock);
1941         if (!list_empty(&obd->obd_lock_replay_queue)) {
1942                 CDEBUG(D_HA, "waking for next lock\n");
1943                 wake_up = 1;
1944         } else if (atomic_read(&obd->obd_lock_replay_clients) == 0) {
1945                 CDEBUG(D_HA, "waking for completed lock replay\n");
1946                 wake_up = 1;
1947         } else if (obd->obd_abort_recovery) {
1948                 CDEBUG(D_HA, "waking for aborted recovery\n");
1949                 wake_up = 1;
1950         } else if (obd->obd_recovery_expired) {
1951                 CDEBUG(D_HA, "waking for expired recovery\n");
1952                 wake_up = 1;
1953         }
1954         spin_unlock(&obd->obd_recovery_task_lock);
1955
1956         return wake_up;
1957 }
1958
1959 /**
1960  * wait for recovery events,
1961  * check its status with help of check_routine
1962  * evict dead clients via health_check
1963  */
1964 static int target_recovery_overseer(struct lu_target *lut,
1965                                     int (*check_routine)(struct lu_target *),
1966                                     int (*health_check)(struct obd_export *))
1967 {
1968         struct obd_device       *obd = lut->lut_obd;
1969         struct target_distribute_txn_data *tdtd;
1970         time64_t last = 0;
1971         time64_t now;
1972 repeat:
1973         if (obd->obd_recovering && obd->obd_recovery_start == 0) {
1974                 now = ktime_get_seconds();
1975                 if (now - last > 600) {
1976                         LCONSOLE_INFO("%s: in recovery but waiting for "
1977                                       "the first client to connect\n",
1978                                       obd->obd_name);
1979                         last = now;
1980                 }
1981         }
1982         if (obd->obd_recovery_start != 0 && ktime_get_real_seconds() >=
1983               (obd->obd_recovery_start + obd->obd_recovery_time_hard)) {
1984                 __u64 next_update_transno = 0;
1985
1986                 /* Only abort the recovery if there are no update recovery
1987                  * left in the queue */
1988                 spin_lock(&obd->obd_recovery_task_lock);
1989                 if (lut->lut_tdtd != NULL) {
1990                         next_update_transno =
1991                                 distribute_txn_get_next_transno(lut->lut_tdtd);
1992
1993                         tdtd = lut->lut_tdtd;
1994                         /* If next_update_transno == 0, it probably because
1995                          * updatelog retrieve threads did not get any records
1996                          * yet, let's wait those threads stopped */
1997                         if (next_update_transno == 0) {
1998                                 struct l_wait_info lwi = { 0 };
1999
2000                                 l_wait_event(tdtd->tdtd_recovery_threads_waitq,
2001                                        atomic_read(
2002                                        &tdtd->tdtd_recovery_threads_count) == 0,
2003                                        &lwi);
2004
2005                                 next_update_transno =
2006                                         distribute_txn_get_next_transno(
2007                                                                 lut->lut_tdtd);
2008                         }
2009                 }
2010
2011                 if (next_update_transno != 0 && !obd->obd_abort_recovery) {
2012                         obd->obd_next_recovery_transno = next_update_transno;
2013                         spin_unlock(&obd->obd_recovery_task_lock);
2014                         /* Disconnect unfinished exports from clients, and
2015                          * keep connection from MDT to make sure the update
2016                          * recovery will still keep trying until some one
2017                          * manually abort the recovery */
2018                         class_disconnect_stale_exports(obd,
2019                                                 exp_finished_or_from_mdt);
2020                         /* Abort all of replay and replay lock req from
2021                          * clients */
2022                         abort_req_replay_queue(obd);
2023                         abort_lock_replay_queue(obd);
2024                         CDEBUG(D_HA, "%s: there are still update replay (%#llx"
2025                                ")in the queue.\n", obd->obd_name,
2026                                next_update_transno);
2027                 } else {
2028                         obd->obd_abort_recovery = 1;
2029                         spin_unlock(&obd->obd_recovery_task_lock);
2030                         CWARN("%s recovery is aborted by hard timeout\n",
2031                               obd->obd_name);
2032                 }
2033         }
2034
2035         while (wait_event_timeout(obd->obd_next_transno_waitq,
2036                                   check_routine(lut),
2037                                   msecs_to_jiffies(60 * MSEC_PER_SEC)) == 0)
2038                 /* wait indefinitely for event, but don't trigger watchdog */;
2039
2040         if (obd->obd_abort_recovery) {
2041                 CWARN("recovery is aborted, evict exports in recovery\n");
2042                 if (lut->lut_tdtd != NULL) {
2043                         struct l_wait_info lwi = { 0 };
2044
2045                         tdtd = lut->lut_tdtd;
2046                         /* Let's wait all of the update log recovery thread
2047                          * finished */
2048                         l_wait_event(tdtd->tdtd_recovery_threads_waitq,
2049                          atomic_read(&tdtd->tdtd_recovery_threads_count) == 0,
2050                              &lwi);
2051                         /* Then abort the update recovery list */
2052                         dtrq_list_destroy(lut->lut_tdtd);
2053                 }
2054
2055                 /** evict exports which didn't finish recovery yet */
2056                 class_disconnect_stale_exports(obd, exp_finished);
2057                 return 1;
2058         } else if (obd->obd_recovery_expired) {
2059                 obd->obd_recovery_expired = 0;
2060                 /** If some clients died being recovered, evict them */
2061                 LCONSOLE_WARN("%s: recovery is timed out, "
2062                               "evict stale exports\n", obd->obd_name);
2063                 /** evict cexports with no replay in queue, they are stalled */
2064                 class_disconnect_stale_exports(obd, health_check);
2065
2066                 /** continue with VBR */
2067                 spin_lock(&obd->obd_dev_lock);
2068                 obd->obd_version_recov = 1;
2069                 spin_unlock(&obd->obd_dev_lock);
2070                 /**
2071                  * reset timer, recovery will proceed with versions now,
2072                  * timeout is set just to handle reconnection delays
2073                  */
2074                 extend_recovery_timer(obd, RECONNECT_DELAY_MAX, true);
2075                 /** Wait for recovery events again, after evicting bad clients */
2076                 goto repeat;
2077         }
2078         return 0;
2079 }
2080
2081 static struct ptlrpc_request *target_next_replay_lock(struct lu_target *lut)
2082 {
2083         struct obd_device       *obd = lut->lut_obd;
2084         struct ptlrpc_request *req = NULL;
2085
2086         CDEBUG(D_HA, "Waiting for lock\n");
2087         if (target_recovery_overseer(lut, check_for_next_lock,
2088                                      exp_lock_replay_healthy))
2089                 abort_lock_replay_queue(obd);
2090
2091         spin_lock(&obd->obd_recovery_task_lock);
2092         if (!list_empty(&obd->obd_lock_replay_queue)) {
2093                 req = list_entry(obd->obd_lock_replay_queue.next,
2094                                      struct ptlrpc_request, rq_list);
2095                 list_del_init(&req->rq_list);
2096                 spin_unlock(&obd->obd_recovery_task_lock);
2097         } else {
2098                 spin_unlock(&obd->obd_recovery_task_lock);
2099                 LASSERT(list_empty(&obd->obd_lock_replay_queue));
2100                 LASSERT(atomic_read(&obd->obd_lock_replay_clients) == 0);
2101                 /** evict exports failed VBR */
2102                 class_disconnect_stale_exports(obd, exp_vbr_healthy);
2103         }
2104         return req;
2105 }
2106
2107 static struct ptlrpc_request *target_next_final_ping(struct obd_device *obd)
2108 {
2109         struct ptlrpc_request *req = NULL;
2110
2111         spin_lock(&obd->obd_recovery_task_lock);
2112         if (!list_empty(&obd->obd_final_req_queue)) {
2113                 req = list_entry(obd->obd_final_req_queue.next,
2114                                      struct ptlrpc_request, rq_list);
2115                 list_del_init(&req->rq_list);
2116                 spin_unlock(&obd->obd_recovery_task_lock);
2117                 if (req->rq_export->exp_in_recovery) {
2118                         spin_lock(&req->rq_export->exp_lock);
2119                         req->rq_export->exp_in_recovery = 0;
2120                         spin_unlock(&req->rq_export->exp_lock);
2121                 }
2122         } else {
2123                 spin_unlock(&obd->obd_recovery_task_lock);
2124         }
2125         return req;
2126 }
2127
2128 static void handle_recovery_req(struct ptlrpc_thread *thread,
2129                                 struct ptlrpc_request *req,
2130                                 svc_handler_t handler)
2131 {
2132         ENTRY;
2133
2134         /**
2135          * export can be evicted during recovery, no need to handle replays for
2136          * it after that, discard such request silently
2137          */
2138         if (req->rq_export->exp_disconnected)
2139                 RETURN_EXIT;
2140
2141         req->rq_session.lc_thread = thread;
2142         req->rq_svc_thread = thread;
2143         req->rq_svc_thread->t_env->le_ses = &req->rq_session;
2144
2145         /* thread context */
2146         lu_context_enter(&thread->t_env->le_ctx);
2147         (void)handler(req);
2148         lu_context_exit(&thread->t_env->le_ctx);
2149
2150         /* don't reset timer for final stage */
2151         if (!exp_finished(req->rq_export)) {
2152                 time64_t to = obd_timeout;
2153
2154                 /**
2155                  * Add request timeout to the recovery time so next request from
2156                  * this client may come in recovery time
2157                  */
2158                 if (!AT_OFF) {
2159                         struct ptlrpc_service_part *svcpt;
2160
2161                         svcpt = req->rq_rqbd->rqbd_svcpt;
2162                         /* If the server sent early reply for this request,
2163                          * the client will recalculate the timeout according to
2164                          * current server estimate service time, so we will
2165                          * use the maxium timeout here for waiting the client
2166                          * sending the next req */
2167                         to = max((int)at_est2timeout(
2168                                  at_get(&svcpt->scp_at_estimate)),
2169                                  (int)lustre_msg_get_timeout(req->rq_reqmsg));
2170                         /* Add 2 net_latency, one for balance rq_deadline
2171                          * (see ptl_send_rpc), one for resend the req to server,
2172                          * Note: client will pack net_latency in replay req
2173                          * (see ptlrpc_replay_req) */
2174                         to += 2 * lustre_msg_get_service_time(req->rq_reqmsg);
2175                 }
2176                 extend_recovery_timer(class_exp2obd(req->rq_export), to, true);
2177         }
2178         EXIT;
2179 }
2180
2181 /** Checking routines for recovery */
2182 static int check_for_recovery_ready(struct lu_target *lut)
2183 {
2184         struct obd_device *obd = lut->lut_obd;
2185         unsigned int clnts = atomic_read(&obd->obd_connected_clients);
2186
2187         CDEBUG(D_HA, "connected %d stale %d max_recoverable_clients %d"
2188                " abort %d expired %d\n", clnts, obd->obd_stale_clients,
2189                obd->obd_max_recoverable_clients, obd->obd_abort_recovery,
2190                obd->obd_recovery_expired);
2191
2192         if (!obd->obd_abort_recovery && !obd->obd_recovery_expired) {
2193                 LASSERT(clnts <= obd->obd_max_recoverable_clients);
2194                 if (clnts + obd->obd_stale_clients <
2195                     obd->obd_max_recoverable_clients)
2196                         return 0;
2197         }
2198
2199         if (lut->lut_tdtd != NULL) {
2200                 if (!lut->lut_tdtd->tdtd_replay_ready &&
2201                     !obd->obd_abort_recovery) {
2202                         /* Let's extend recovery timer, in case the recovery
2203                          * timer expired, and some clients got evicted */
2204                         extend_recovery_timer(obd, obd->obd_recovery_timeout,
2205                                               true);
2206                         CDEBUG(D_HA, "%s update recovery is not ready, extend recovery %llu\n",
2207                                obd->obd_name, obd->obd_recovery_timeout);
2208                         return 0;
2209                 }
2210         }
2211
2212         return 1;
2213 }
2214
2215 enum {
2216         REQUEST_RECOVERY = 1,
2217         UPDATE_RECOVERY = 2,
2218 };
2219
2220 static __u64 get_next_replay_req_transno(struct obd_device *obd)
2221 {
2222         __u64 transno = 0;
2223
2224         if (!list_empty(&obd->obd_req_replay_queue)) {
2225                 struct ptlrpc_request *req;
2226
2227                 req = list_entry(obd->obd_req_replay_queue.next,
2228                                  struct ptlrpc_request, rq_list);
2229                 transno = lustre_msg_get_transno(req->rq_reqmsg);
2230         }
2231
2232         return transno;
2233 }
2234
2235 static __u64 get_next_transno(struct lu_target *lut, int *type)
2236 {
2237         struct obd_device *obd = lut->lut_obd;
2238         struct target_distribute_txn_data *tdtd = lut->lut_tdtd;
2239         __u64 transno = 0;
2240         __u64 update_transno;
2241         ENTRY;
2242
2243         transno = get_next_replay_req_transno(obd);
2244         if (type != NULL)
2245                 *type = REQUEST_RECOVERY;
2246
2247         if (tdtd == NULL)
2248                 RETURN(transno);
2249
2250         update_transno = distribute_txn_get_next_transno(tdtd);
2251         if (transno == 0 || (transno >= update_transno &&
2252                              update_transno != 0)) {
2253                 transno = update_transno;
2254                 if (type != NULL)
2255                         *type = UPDATE_RECOVERY;
2256         }
2257
2258         RETURN(transno);
2259 }
2260
2261 /**
2262  * drop duplicate replay request
2263  *
2264  * Because the operation has been replayed by update recovery, the request
2265  * with the same transno will be dropped and also notify the client to send
2266  * next replay request.
2267  *
2268  * \param[in] env       execution environment
2269  * \param[in] obd       failover obd device
2270  * \param[in] req       request to be dropped
2271  */
2272 static void drop_duplicate_replay_req(struct lu_env *env,
2273                                       struct obd_device *obd,
2274                                       struct ptlrpc_request *req)
2275 {
2276         DEBUG_REQ(D_HA, req, "remove t%lld from %s because of duplicate"
2277                   " update records are found.\n",
2278                   lustre_msg_get_transno(req->rq_reqmsg),
2279                   libcfs_nid2str(req->rq_peer.nid));
2280
2281         /* Right now, only for MDS reint operation update replay and
2282          * normal request replay can have the same transno */
2283         if (lustre_msg_get_opc(req->rq_reqmsg) == MDS_REINT) {
2284                 req_capsule_set(&req->rq_pill, &RQF_MDS_REINT);
2285                 req->rq_status = req_capsule_server_pack(&req->rq_pill);
2286                 if (likely(req->rq_export))
2287                         target_committed_to_req(req);
2288                 lustre_msg_set_transno(req->rq_repmsg, req->rq_transno);
2289                 target_send_reply(req, req->rq_status, 0);
2290         } else {
2291                 DEBUG_REQ(D_ERROR, req, "wrong opc" "from %s\n",
2292                 libcfs_nid2str(req->rq_peer.nid));
2293         }
2294         target_exp_dequeue_req_replay(req);
2295         target_request_copy_put(req);
2296         obd->obd_replayed_requests++;
2297 }
2298
2299 static void replay_request_or_update(struct lu_env *env,
2300                                      struct lu_target *lut,
2301                                      struct target_recovery_data *trd,
2302                                      struct ptlrpc_thread *thread)
2303 {
2304         struct obd_device *obd = lut->lut_obd;
2305         struct ptlrpc_request *req = NULL;
2306         int                     type;
2307         __u64                   transno;
2308         ENTRY;
2309
2310         CDEBUG(D_HA, "Waiting for transno %lld\n",
2311                obd->obd_next_recovery_transno);
2312
2313         /* Replay all of request and update by transno */
2314         do {
2315                 struct target_distribute_txn_data *tdtd = lut->lut_tdtd;
2316
2317                 CFS_FAIL_TIMEOUT(OBD_FAIL_TGT_REPLAY_DELAY2, cfs_fail_val);
2318
2319                 /** It is needed to extend recovery window above
2320                  *  recovery_time_soft. Extending is possible only in the
2321                  *  end of recovery window (see more details in
2322                  *  handle_recovery_req()).
2323                  */
2324                 CFS_FAIL_TIMEOUT_MS(OBD_FAIL_TGT_REPLAY_DELAY, 300);
2325
2326                 if (target_recovery_overseer(lut, check_for_next_transno,
2327                                         exp_req_replay_healthy_or_from_mdt)) {
2328                         abort_req_replay_queue(obd);
2329                         abort_lock_replay_queue(obd);
2330                         goto abort;
2331                 }
2332
2333                 spin_lock(&obd->obd_recovery_task_lock);
2334                 transno = get_next_transno(lut, &type);
2335                 if (type == REQUEST_RECOVERY && transno != 0) {
2336                         /* Drop replay request from client side, if the
2337                          * replay has been executed by update with the
2338                          * same transno */
2339                         req = list_entry(obd->obd_req_replay_queue.next,
2340                                         struct ptlrpc_request, rq_list);
2341
2342                         list_del_init(&req->rq_list);
2343                         obd->obd_requests_queued_for_recovery--;
2344                         spin_unlock(&obd->obd_recovery_task_lock);
2345
2346                         /* Let's check if the request has been redone by
2347                          * update replay */
2348                         if (is_req_replayed_by_update(req)) {
2349                                 struct distribute_txn_replay_req *dtrq;
2350
2351                                 dtrq = distribute_txn_lookup_finish_list(tdtd,
2352                                                                    req->rq_xid);
2353                                 LASSERT(dtrq != NULL);
2354                                 spin_lock(&tdtd->tdtd_replay_list_lock);
2355                                 list_del_init(&dtrq->dtrq_list);
2356                                 spin_unlock(&tdtd->tdtd_replay_list_lock);
2357                                 dtrq_destroy(dtrq);
2358
2359                                 drop_duplicate_replay_req(env, obd, req);
2360
2361                                 continue;
2362                         }
2363
2364                         LASSERT(trd->trd_processing_task == current_pid());
2365                         DEBUG_REQ(D_HA, req, "processing t%lld from %s",
2366                                   lustre_msg_get_transno(req->rq_reqmsg),
2367                                   libcfs_nid2str(req->rq_peer.nid));
2368
2369                         handle_recovery_req(thread, req,
2370                                             trd->trd_recovery_handler);
2371                         /**
2372                          * bz18031: increase next_recovery_transno before
2373                          * target_request_copy_put() will drop exp_rpc reference
2374                          */
2375                         spin_lock(&obd->obd_recovery_task_lock);
2376                         obd->obd_next_recovery_transno++;
2377                         spin_unlock(&obd->obd_recovery_task_lock);
2378                         target_exp_dequeue_req_replay(req);
2379                         target_request_copy_put(req);
2380                         obd->obd_replayed_requests++;
2381                 } else if (type == UPDATE_RECOVERY && transno != 0) {
2382                         struct distribute_txn_replay_req *dtrq;
2383                         int rc;
2384
2385                         spin_unlock(&obd->obd_recovery_task_lock);
2386
2387                         LASSERT(tdtd != NULL);
2388                         dtrq = distribute_txn_get_next_req(tdtd);
2389                         lu_context_enter(&thread->t_env->le_ctx);
2390                         rc = tdtd->tdtd_replay_handler(env, tdtd, dtrq);
2391                         lu_context_exit(&thread->t_env->le_ctx);
2392                         extend_recovery_timer(obd, obd_timeout, true);
2393
2394                         if (rc == 0 && dtrq->dtrq_xid != 0) {
2395                                 CDEBUG(D_HA, "Move x%llu t%llu"
2396                                        " to finish list\n", dtrq->dtrq_xid,
2397                                        dtrq->dtrq_master_transno);
2398
2399                                 /* Add it to the replay finish list */
2400                                 spin_lock(&tdtd->tdtd_replay_list_lock);
2401                                 list_add(&dtrq->dtrq_list,
2402                                          &tdtd->tdtd_replay_finish_list);
2403                                 spin_unlock(&tdtd->tdtd_replay_list_lock);
2404
2405                                 spin_lock(&obd->obd_recovery_task_lock);
2406                                 if (transno == obd->obd_next_recovery_transno)
2407                                         obd->obd_next_recovery_transno++;
2408                                 else if (transno >
2409                                          obd->obd_next_recovery_transno)
2410                                         obd->obd_next_recovery_transno =
2411                                                                 transno + 1;
2412                                 spin_unlock(&obd->obd_recovery_task_lock);
2413                         } else {
2414                                 dtrq_destroy(dtrq);
2415                         }
2416                 } else {
2417                         spin_unlock(&obd->obd_recovery_task_lock);
2418 abort:
2419                         LASSERT(list_empty(&obd->obd_req_replay_queue));
2420                         LASSERT(atomic_read(&obd->obd_req_replay_clients) == 0);
2421                         /** evict exports failed VBR */
2422                         class_disconnect_stale_exports(obd, exp_vbr_healthy);
2423                         break;
2424                 }
2425         } while (1);
2426 }
2427
2428 static int target_recovery_thread(void *arg)
2429 {
2430         struct lu_target *lut = arg;
2431         struct obd_device *obd = lut->lut_obd;
2432         struct ptlrpc_request *req;
2433         struct target_recovery_data *trd = &obd->obd_recovery_data;
2434         unsigned long delta;
2435         struct lu_env *env;
2436         struct ptlrpc_thread *thread = NULL;
2437         int rc = 0;
2438         ENTRY;
2439
2440         unshare_fs_struct();
2441         OBD_ALLOC_PTR(thread);
2442         if (thread == NULL)
2443                 RETURN(-ENOMEM);
2444
2445         OBD_ALLOC_PTR(env);
2446         if (env == NULL) {
2447                 OBD_FREE_PTR(thread);
2448                 RETURN(-ENOMEM);
2449         }
2450
2451         rc = lu_context_init(&env->le_ctx, LCT_MD_THREAD | LCT_DT_THREAD);
2452         if (rc) {
2453                 OBD_FREE_PTR(thread);
2454                 OBD_FREE_PTR(env);
2455                 RETURN(rc);
2456         }
2457
2458         thread->t_env = env;
2459         thread->t_id = -1; /* force filter_iobuf_get/put to use local buffers */
2460         env->le_ctx.lc_thread = thread;
2461         tgt_io_thread_init(thread); /* init thread_big_cache for IO requests */
2462         thread->t_watchdog = NULL;
2463
2464         CDEBUG(D_HA, "%s: started recovery thread pid %d\n", obd->obd_name,
2465                current_pid());
2466         trd->trd_processing_task = current_pid();
2467
2468         spin_lock(&obd->obd_dev_lock);
2469         obd->obd_recovering = 1;
2470         spin_unlock(&obd->obd_dev_lock);
2471         complete(&trd->trd_starting);
2472
2473         /* first of all, we have to know the first transno to replay */
2474         if (target_recovery_overseer(lut, check_for_recovery_ready,
2475                                      exp_connect_healthy)) {
2476                 abort_req_replay_queue(obd);
2477                 abort_lock_replay_queue(obd);
2478                 if (lut->lut_tdtd != NULL)
2479                         dtrq_list_destroy(lut->lut_tdtd);
2480         }
2481
2482         /* next stage: replay requests or update */
2483         delta = jiffies;
2484         CDEBUG(D_INFO, "1: request replay stage - %d clients from t%llu\n",
2485                atomic_read(&obd->obd_req_replay_clients),
2486                obd->obd_next_recovery_transno);
2487         replay_request_or_update(env, lut, trd, thread);
2488
2489         /**
2490          * The second stage: replay locks
2491          */
2492         CDEBUG(D_INFO, "2: lock replay stage - %d clients\n",
2493                atomic_read(&obd->obd_lock_replay_clients));
2494         while ((req = target_next_replay_lock(lut))) {
2495                 LASSERT(trd->trd_processing_task == current_pid());
2496                 DEBUG_REQ(D_HA, req, "processing lock from %s: ",
2497                           libcfs_nid2str(req->rq_peer.nid));
2498                 handle_recovery_req(thread, req,
2499                                     trd->trd_recovery_handler);
2500                 target_request_copy_put(req);
2501                 obd->obd_replayed_locks++;
2502         }
2503
2504         /**
2505          * The third stage: reply on final pings, at this moment all clients
2506          * must have request in final queue
2507          */
2508         CFS_FAIL_TIMEOUT(OBD_FAIL_TGT_REPLAY_RECONNECT, cfs_fail_val);
2509         CDEBUG(D_INFO, "3: final stage - process recovery completion pings\n");
2510         /** Update server last boot epoch */
2511         tgt_boot_epoch_update(lut);
2512         /* We drop recoverying flag to forward all new requests
2513          * to regular mds_handle() since now */
2514         spin_lock(&obd->obd_dev_lock);
2515         obd->obd_recovering = obd->obd_abort_recovery = 0;
2516         spin_unlock(&obd->obd_dev_lock);
2517         spin_lock(&obd->obd_recovery_task_lock);
2518         target_cancel_recovery_timer(obd);
2519         spin_unlock(&obd->obd_recovery_task_lock);
2520         while ((req = target_next_final_ping(obd))) {
2521                 LASSERT(trd->trd_processing_task == current_pid());
2522                 DEBUG_REQ(D_HA, req, "processing final ping from %s: ",
2523                           libcfs_nid2str(req->rq_peer.nid));
2524                 handle_recovery_req(thread, req,
2525                                     trd->trd_recovery_handler);
2526                 /* Because the waiting client can not send ping to server,
2527                  * so we need refresh the last_request_time, to avoid the
2528                  * export is being evicted */
2529                 ptlrpc_update_export_timer(req->rq_export, 0);
2530                 target_request_copy_put(req);
2531         }
2532
2533         delta = jiffies_to_msecs(jiffies - delta) / MSEC_PER_SEC;
2534         CDEBUG(D_INFO,"4: recovery completed in %lus - %d/%d reqs/locks\n",
2535               delta, obd->obd_replayed_requests, obd->obd_replayed_locks);
2536         if (delta > OBD_RECOVERY_TIME_SOFT) {
2537                 CWARN("too long recovery - read logs\n");
2538                 libcfs_debug_dumplog();
2539         }
2540
2541         target_finish_recovery(lut);
2542
2543         lu_context_fini(&env->le_ctx);
2544         trd->trd_processing_task = 0;
2545         complete(&trd->trd_finishing);
2546
2547         tgt_io_thread_done(thread);
2548         OBD_FREE_PTR(thread);
2549         OBD_FREE_PTR(env);
2550         RETURN(rc);
2551 }
2552
2553 static int target_start_recovery_thread(struct lu_target *lut,
2554                                         svc_handler_t handler)
2555 {
2556         struct obd_device *obd = lut->lut_obd;
2557         int rc = 0;
2558         struct target_recovery_data *trd = &obd->obd_recovery_data;
2559         int index;
2560
2561         memset(trd, 0, sizeof(*trd));
2562         init_completion(&trd->trd_starting);
2563         init_completion(&trd->trd_finishing);
2564         trd->trd_recovery_handler = handler;
2565
2566         rc = server_name2index(obd->obd_name, &index, NULL);
2567         if (rc < 0)
2568                 return rc;
2569
2570         if (!IS_ERR(kthread_run(target_recovery_thread,
2571                                 lut, "tgt_recover_%d", index))) {
2572                 wait_for_completion(&trd->trd_starting);
2573                 LASSERT(obd->obd_recovering != 0);
2574         } else {
2575                 rc = -ECHILD;
2576         }
2577
2578         return rc;
2579 }
2580
2581 void target_stop_recovery_thread(struct obd_device *obd)
2582 {
2583         if (obd->obd_recovery_data.trd_processing_task > 0) {
2584                 struct target_recovery_data *trd = &obd->obd_recovery_data;
2585                 /** recovery can be done but postrecovery is not yet */
2586                 spin_lock(&obd->obd_dev_lock);
2587                 if (obd->obd_recovering) {
2588                         CERROR("%s: Aborting recovery\n", obd->obd_name);
2589                         obd->obd_abort_recovery = 1;
2590                         wake_up(&obd->obd_next_transno_waitq);
2591                 }
2592                 spin_unlock(&obd->obd_dev_lock);
2593                 wait_for_completion(&trd->trd_finishing);
2594         }
2595 }
2596 EXPORT_SYMBOL(target_stop_recovery_thread);
2597
2598 void target_recovery_fini(struct obd_device *obd)
2599 {
2600         class_disconnect_exports(obd);
2601         target_stop_recovery_thread(obd);
2602         target_cleanup_recovery(obd);
2603 }
2604 EXPORT_SYMBOL(target_recovery_fini);
2605
2606 static void target_recovery_expired(unsigned long castmeharder)
2607 {
2608         struct obd_device *obd = (struct obd_device *)castmeharder;
2609         CDEBUG(D_HA, "%s: recovery timed out; %d clients are still in recovery"
2610                " after %llus (%d clients connected)\n",
2611                obd->obd_name, atomic_read(&obd->obd_lock_replay_clients),
2612                (s64)(ktime_get_real_seconds() - obd->obd_recovery_start),
2613                atomic_read(&obd->obd_connected_clients));
2614
2615         obd->obd_recovery_expired = 1;
2616         wake_up(&obd->obd_next_transno_waitq);
2617 }
2618
2619 void target_recovery_init(struct lu_target *lut, svc_handler_t handler)
2620 {
2621         struct obd_device *obd = lut->lut_obd;
2622
2623         if (lut->lut_bottom->dd_rdonly)
2624                 return;
2625
2626         if (obd->obd_max_recoverable_clients == 0) {
2627                 /** Update server last boot epoch */
2628                 tgt_boot_epoch_update(lut);
2629                 return;
2630         }
2631
2632         CDEBUG(D_HA, "RECOVERY: service %s, %d recoverable clients, "
2633                "last_transno %llu\n", obd->obd_name,
2634                obd->obd_max_recoverable_clients, obd->obd_last_committed);
2635         LASSERT(obd->obd_stopping == 0);
2636         obd->obd_next_recovery_transno = obd->obd_last_committed + 1;
2637         obd->obd_recovery_start = 0;
2638         obd->obd_recovery_end = 0;
2639
2640         setup_timer(&obd->obd_recovery_timer, target_recovery_expired,
2641                     (unsigned long)obd);
2642         target_start_recovery_thread(lut, handler);
2643 }
2644 EXPORT_SYMBOL(target_recovery_init);
2645
2646 static int target_process_req_flags(struct obd_device *obd,
2647                                     struct ptlrpc_request *req)
2648 {
2649         struct obd_export *exp = req->rq_export;
2650         LASSERT(exp != NULL);
2651         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REQ_REPLAY_DONE) {
2652                 /* client declares he's ready to replay locks */
2653                 spin_lock(&exp->exp_lock);
2654                 if (exp->exp_req_replay_needed) {
2655                         exp->exp_req_replay_needed = 0;
2656                         spin_unlock(&exp->exp_lock);
2657
2658                         LASSERT_ATOMIC_POS(&obd->obd_req_replay_clients);
2659                         atomic_dec(&obd->obd_req_replay_clients);
2660                 } else {
2661                         spin_unlock(&exp->exp_lock);
2662                 }
2663         }
2664         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_LOCK_REPLAY_DONE) {
2665                 /* client declares he's ready to complete recovery
2666                  * so, we put the request on th final queue */
2667                 spin_lock(&exp->exp_lock);
2668                 if (exp->exp_lock_replay_needed) {
2669                         exp->exp_lock_replay_needed = 0;
2670                         spin_unlock(&exp->exp_lock);
2671
2672                         LASSERT_ATOMIC_POS(&obd->obd_lock_replay_clients);
2673                         atomic_dec(&obd->obd_lock_replay_clients);
2674                 } else {
2675                         spin_unlock(&exp->exp_lock);
2676                 }
2677         }
2678         return 0;
2679 }
2680
2681 int target_queue_recovery_request(struct ptlrpc_request *req,
2682                                   struct obd_device *obd)
2683 {
2684         __u64 transno = lustre_msg_get_transno(req->rq_reqmsg);
2685         struct ptlrpc_request *reqiter;
2686         int inserted = 0;
2687         ENTRY;
2688
2689         if (obd->obd_recovery_data.trd_processing_task == current_pid()) {
2690                 /* Processing the queue right now, don't re-add. */
2691                 RETURN(1);
2692         }
2693
2694         target_process_req_flags(obd, req);
2695
2696         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_LOCK_REPLAY_DONE) {
2697                 if (unlikely(OBD_FAIL_CHECK(OBD_FAIL_TGT_RECOVERY_REQ_RACE))) {
2698                         if (cfs_fail_val == 1) {
2699                                 cfs_race_state = 1;
2700                                 cfs_fail_val = 0;
2701                                 wake_up(&cfs_race_waitq);
2702
2703                                 set_current_state(TASK_INTERRUPTIBLE);
2704                                 schedule_timeout(cfs_time_seconds(1));
2705                         }
2706                 }
2707
2708                 /* client declares he's ready to complete recovery
2709                  * so, we put the request on th final queue */
2710                 target_request_copy_get(req);
2711                 DEBUG_REQ(D_HA, req, "queue final req");
2712                 wake_up(&obd->obd_next_transno_waitq);
2713                 spin_lock(&obd->obd_recovery_task_lock);
2714                 if (obd->obd_recovering) {
2715                         struct ptlrpc_request *tmp;
2716                         struct ptlrpc_request *duplicate = NULL;
2717
2718                         if (likely(!req->rq_export->exp_replay_done)) {
2719                                 req->rq_export->exp_replay_done = 1;
2720                                 list_add_tail(&req->rq_list,
2721                                               &obd->obd_final_req_queue);
2722                                 spin_unlock(&obd->obd_recovery_task_lock);
2723                                 RETURN(0);
2724                         }
2725
2726                         /* XXX O(n), but only happens if final ping is
2727                          * timed out, probably reorganize the list as
2728                          * a hash list later */
2729                         list_for_each_entry_safe(reqiter, tmp,
2730                                                  &obd->obd_final_req_queue,
2731                                                  rq_list) {
2732                                 if (reqiter->rq_export == req->rq_export) {
2733                                         list_del_init(&reqiter->rq_list);
2734                                         duplicate = reqiter;
2735                                         break;
2736                                 }
2737                         }
2738
2739                         list_add_tail(&req->rq_list,
2740                                       &obd->obd_final_req_queue);
2741                         req->rq_export->exp_replay_done = 1;
2742                         spin_unlock(&obd->obd_recovery_task_lock);
2743
2744                         if (duplicate != NULL) {
2745                                 DEBUG_REQ(D_HA, duplicate,
2746                                           "put prev final req\n");
2747                                 target_request_copy_put(duplicate);
2748                         }
2749                         RETURN(0);
2750                 } else {
2751                         spin_unlock(&obd->obd_recovery_task_lock);
2752                         target_request_copy_put(req);
2753                         RETURN(obd->obd_stopping ? -ENOTCONN : 1);
2754                 }
2755         }
2756         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REQ_REPLAY_DONE) {
2757                 /* client declares he's ready to replay locks */
2758                 target_request_copy_get(req);
2759                 DEBUG_REQ(D_HA, req, "queue lock replay req");
2760                 wake_up(&obd->obd_next_transno_waitq);
2761                 spin_lock(&obd->obd_recovery_task_lock);
2762                 LASSERT(obd->obd_recovering);
2763                 /* usually due to recovery abort */
2764                 if (!req->rq_export->exp_in_recovery) {
2765                         spin_unlock(&obd->obd_recovery_task_lock);
2766                         target_request_copy_put(req);
2767                         RETURN(-ENOTCONN);
2768                 }
2769                 LASSERT(req->rq_export->exp_lock_replay_needed);
2770                 list_add_tail(&req->rq_list, &obd->obd_lock_replay_queue);
2771                 spin_unlock(&obd->obd_recovery_task_lock);
2772                 RETURN(0);
2773         }
2774
2775         /* CAVEAT EMPTOR: The incoming request message has been swabbed
2776          * (i.e. buflens etc are in my own byte order), but type-dependent
2777          * buffers (eg mdt_body, ost_body etc) have NOT been swabbed. */
2778
2779         if (!transno) {
2780                 INIT_LIST_HEAD(&req->rq_list);
2781                 DEBUG_REQ(D_HA, req, "not queueing");
2782                 RETURN(1);
2783         }
2784
2785         /* If we're processing the queue, we want don't want to queue this
2786          * message.
2787          *
2788          * Also, if this request has a transno less than the one we're waiting
2789          * for, we should process it now.  It could (and currently always will)
2790          * be an open request for a descriptor that was opened some time ago.
2791          *
2792          * Also, a resent, replayed request that has already been
2793          * handled will pass through here and be processed immediately.
2794          */
2795         CDEBUG(D_HA, "Next recovery transno: %llu"
2796                ", current: %llu, replaying\n",
2797                obd->obd_next_recovery_transno, transno);
2798
2799         /* If the request has been replayed by update replay, then sends this
2800          * request to the recovery thread (replay_request_or_update()), where
2801          * it will be handled */
2802         spin_lock(&obd->obd_recovery_task_lock);
2803         if (transno < obd->obd_next_recovery_transno &&
2804             !is_req_replayed_by_update(req)) {
2805                 /* Processing the queue right now, don't re-add. */
2806                 LASSERT(list_empty(&req->rq_list));
2807                 spin_unlock(&obd->obd_recovery_task_lock);
2808                 RETURN(1);
2809         }
2810         spin_unlock(&obd->obd_recovery_task_lock);
2811
2812         if (OBD_FAIL_CHECK(OBD_FAIL_TGT_REPLAY_DROP))
2813                 RETURN(0);
2814
2815         target_request_copy_get(req);
2816         if (!req->rq_export->exp_in_recovery) {
2817                 target_request_copy_put(req);
2818                 RETURN(-ENOTCONN);
2819         }
2820         LASSERT(req->rq_export->exp_req_replay_needed);
2821
2822         if (target_exp_enqueue_req_replay(req)) {
2823                 DEBUG_REQ(D_ERROR, req, "dropping resent queued req");
2824                 target_request_copy_put(req);
2825                 RETURN(0);
2826         }
2827
2828         /* XXX O(n^2) */
2829         spin_lock(&obd->obd_recovery_task_lock);
2830         LASSERT(obd->obd_recovering);
2831         list_for_each_entry(reqiter, &obd->obd_req_replay_queue, rq_list) {
2832                 if (lustre_msg_get_transno(reqiter->rq_reqmsg) > transno) {
2833                         list_add_tail(&req->rq_list, &reqiter->rq_list);
2834                         inserted = 1;
2835                         goto added;
2836                 }
2837
2838                 if (unlikely(lustre_msg_get_transno(reqiter->rq_reqmsg) ==
2839                              transno)) {
2840                         DEBUG_REQ(D_ERROR, req, "dropping replay: transno "
2841                                   "has been claimed by another client");
2842                         spin_unlock(&obd->obd_recovery_task_lock);
2843                         target_exp_dequeue_req_replay(req);
2844                         target_request_copy_put(req);
2845                         RETURN(0);
2846                 }
2847         }
2848 added:
2849         if (!inserted)
2850                 list_add_tail(&req->rq_list, &obd->obd_req_replay_queue);
2851
2852         obd->obd_requests_queued_for_recovery++;
2853         spin_unlock(&obd->obd_recovery_task_lock);
2854         wake_up(&obd->obd_next_transno_waitq);
2855         RETURN(0);
2856 }
2857
2858 int target_handle_ping(struct ptlrpc_request *req)
2859 {
2860         obd_ping(req->rq_svc_thread->t_env, req->rq_export);
2861         return req_capsule_server_pack(&req->rq_pill);
2862 }
2863
2864 void target_committed_to_req(struct ptlrpc_request *req)
2865 {
2866         struct obd_export *exp = req->rq_export;
2867
2868         if (!exp->exp_obd->obd_no_transno && req->rq_repmsg != NULL)
2869                 lustre_msg_set_last_committed(req->rq_repmsg,
2870                                               exp->exp_last_committed);
2871         else
2872                 DEBUG_REQ(D_IOCTL, req, "not sending last_committed update (%d/"
2873                           "%d)", exp->exp_obd->obd_no_transno,
2874                           req->rq_repmsg == NULL);
2875
2876         CDEBUG(D_INFO, "last_committed %llu, transno %llu, xid %llu\n",
2877                exp->exp_last_committed, req->rq_transno, req->rq_xid);
2878 }
2879
2880 #endif /* HAVE_SERVER_SUPPORT */
2881
2882 /**
2883  * Packs current SLV and Limit into \a req.
2884  */
2885 int target_pack_pool_reply(struct ptlrpc_request *req)
2886 {
2887         struct obd_device *obd;
2888         ENTRY;
2889
2890         /* Check that we still have all structures alive as this may
2891          * be some late RPC at shutdown time. */
2892         if (unlikely(!req->rq_export || !req->rq_export->exp_obd ||
2893                      !exp_connect_lru_resize(req->rq_export))) {
2894                 lustre_msg_set_slv(req->rq_repmsg, 0);
2895                 lustre_msg_set_limit(req->rq_repmsg, 0);
2896                 RETURN(0);
2897         }
2898
2899         /* OBD is alive here as export is alive, which we checked above. */
2900         obd = req->rq_export->exp_obd;
2901
2902         read_lock(&obd->obd_pool_lock);
2903         lustre_msg_set_slv(req->rq_repmsg, obd->obd_pool_slv);
2904         lustre_msg_set_limit(req->rq_repmsg, obd->obd_pool_limit);
2905         read_unlock(&obd->obd_pool_lock);
2906
2907         RETURN(0);
2908 }
2909
2910 static int target_send_reply_msg(struct ptlrpc_request *req,
2911                                  int rc, int fail_id)
2912 {
2913         if (OBD_FAIL_CHECK_ORSET(fail_id & ~OBD_FAIL_ONCE, OBD_FAIL_ONCE)) {
2914                 DEBUG_REQ(D_ERROR, req, "dropping reply");
2915                 return -ECOMM;
2916         }
2917         /* We can have a null rq_reqmsg in the event of bad signature or
2918          * no context when unwrapping */
2919         if (req->rq_reqmsg &&
2920             unlikely(lustre_msg_get_opc(req->rq_reqmsg) == MDS_REINT &&
2921             OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_MULTI_NET_REP)))
2922                 return -ECOMM;
2923
2924         if (unlikely(rc)) {
2925                 DEBUG_REQ(D_NET, req, "processing error (%d)", rc);
2926                 req->rq_status = rc;
2927                 return ptlrpc_send_error(req, 1);
2928         } else {
2929                 DEBUG_REQ(D_NET, req, "sending reply");
2930         }
2931
2932         return ptlrpc_send_reply(req, PTLRPC_REPLY_MAYBE_DIFFICULT);
2933 }
2934
2935 void target_send_reply(struct ptlrpc_request *req, int rc, int fail_id)
2936 {
2937         struct ptlrpc_service_part *svcpt;
2938         int                        netrc;
2939         struct ptlrpc_reply_state *rs;
2940         struct obd_export         *exp;
2941         ENTRY;
2942
2943         if (req->rq_no_reply) {
2944                 EXIT;
2945                 return;
2946         }
2947
2948         svcpt = req->rq_rqbd->rqbd_svcpt;
2949         rs = req->rq_reply_state;
2950         if (rs == NULL || !rs->rs_difficult) {
2951                 /* no notifiers */
2952                 target_send_reply_msg (req, rc, fail_id);
2953                 EXIT;
2954                 return;
2955         }
2956
2957         /* must be an export if locks saved */
2958         LASSERT(req->rq_export != NULL);
2959         /* req/reply consistent */
2960         LASSERT(rs->rs_svcpt == svcpt);
2961
2962         /* "fresh" reply */
2963         LASSERT(!rs->rs_scheduled);
2964         LASSERT(!rs->rs_scheduled_ever);
2965         LASSERT(!rs->rs_handled);
2966         LASSERT(!rs->rs_on_net);
2967         LASSERT(rs->rs_export == NULL);
2968         LASSERT(list_empty(&rs->rs_obd_list));
2969         LASSERT(list_empty(&rs->rs_exp_list));
2970
2971         exp = class_export_get(req->rq_export);
2972
2973         /* disable reply scheduling while I'm setting up */
2974         rs->rs_scheduled = 1;
2975         rs->rs_on_net    = 1;
2976         rs->rs_xid       = req->rq_xid;
2977         rs->rs_transno   = req->rq_transno;
2978         rs->rs_export    = exp;
2979         rs->rs_opc       = lustre_msg_get_opc(req->rq_reqmsg);
2980
2981         spin_lock(&exp->exp_uncommitted_replies_lock);
2982         CDEBUG(D_NET, "rs transno = %llu, last committed = %llu\n",
2983                rs->rs_transno, exp->exp_last_committed);
2984         if (rs->rs_transno > exp->exp_last_committed) {
2985                 /* not committed already */
2986                 list_add_tail(&rs->rs_obd_list,
2987                                   &exp->exp_uncommitted_replies);
2988         }
2989         spin_unlock(&exp->exp_uncommitted_replies_lock);
2990
2991         spin_lock(&exp->exp_lock);
2992         list_add_tail(&rs->rs_exp_list, &exp->exp_outstanding_replies);
2993         spin_unlock(&exp->exp_lock);
2994
2995         netrc = target_send_reply_msg(req, rc, fail_id);
2996
2997         spin_lock(&svcpt->scp_rep_lock);
2998
2999         atomic_inc(&svcpt->scp_nreps_difficult);
3000
3001         if (netrc != 0) {
3002                 /* error sending: reply is off the net.  Also we need +1
3003                  * reply ref until ptlrpc_handle_rs() is done
3004                  * with the reply state (if the send was successful, there
3005                  * would have been +1 ref for the net, which
3006                  * reply_out_callback leaves alone) */
3007                 rs->rs_on_net = 0;
3008                 ptlrpc_rs_addref(rs);
3009         }
3010
3011         spin_lock(&rs->rs_lock);
3012         if (rs->rs_transno <= exp->exp_last_committed ||
3013             (!rs->rs_on_net && !rs->rs_no_ack) ||
3014             list_empty(&rs->rs_exp_list) ||     /* completed already */
3015             list_empty(&rs->rs_obd_list)) {
3016                 CDEBUG(D_HA, "Schedule reply immediately\n");
3017                 ptlrpc_dispatch_difficult_reply(rs);
3018         } else {
3019                 list_add(&rs->rs_list, &svcpt->scp_rep_active);
3020                 rs->rs_scheduled = 0;   /* allow notifier to schedule */
3021         }
3022         spin_unlock(&rs->rs_lock);
3023         spin_unlock(&svcpt->scp_rep_lock);
3024         EXIT;
3025 }
3026
3027 enum ldlm_mode lck_compat_array[] = {
3028         [LCK_EX]    = LCK_COMPAT_EX,
3029         [LCK_PW]    = LCK_COMPAT_PW,
3030         [LCK_PR]    = LCK_COMPAT_PR,
3031         [LCK_CW]    = LCK_COMPAT_CW,
3032         [LCK_CR]    = LCK_COMPAT_CR,
3033         [LCK_NL]    = LCK_COMPAT_NL,
3034         [LCK_GROUP] = LCK_COMPAT_GROUP,
3035         [LCK_COS]   = LCK_COMPAT_COS,
3036 };
3037
3038 /**
3039  * Rather arbitrary mapping from LDLM error codes to errno values. This should
3040  * not escape to the user level.
3041  */
3042 int ldlm_error2errno(enum ldlm_error error)
3043 {
3044         int result;
3045
3046         switch (error) {
3047         case ELDLM_OK:
3048         case ELDLM_LOCK_MATCHED:
3049                 result = 0;
3050                 break;
3051         case ELDLM_LOCK_CHANGED:
3052                 result = -ESTALE;
3053                 break;
3054         case ELDLM_LOCK_ABORTED:
3055                 result = -ENAVAIL;
3056                 break;
3057         case ELDLM_LOCK_REPLACED:
3058                 result = -ESRCH;
3059                 break;
3060         case ELDLM_NO_LOCK_DATA:
3061                 result = -ENOENT;
3062                 break;
3063         case ELDLM_NAMESPACE_EXISTS:
3064                 result = -EEXIST;
3065                 break;
3066         case ELDLM_BAD_NAMESPACE:
3067                 result = -EBADF;
3068                 break;
3069         default:
3070                 if (((int)error) < 0) { /* cast to signed type */
3071                         result = error; /* as ldlm_error can be unsigned */
3072                 } else {
3073                         CERROR("Invalid DLM result code: %d\n", error);
3074                         result = -EPROTO;
3075                 }
3076         }
3077         return result;
3078 }
3079 EXPORT_SYMBOL(ldlm_error2errno);
3080
3081 /**
3082  * Dual to ldlm_error2errno(): maps errno values back to enum ldlm_error.
3083  */
3084 enum ldlm_error ldlm_errno2error(int err_no)
3085 {
3086         int error;
3087
3088         switch (err_no) {
3089         case 0:
3090                 error = ELDLM_OK;
3091                 break;
3092         case -ESTALE:
3093                 error = ELDLM_LOCK_CHANGED;
3094                 break;
3095         case -ENAVAIL:
3096                 error = ELDLM_LOCK_ABORTED;
3097                 break;
3098         case -ESRCH:
3099                 error = ELDLM_LOCK_REPLACED;
3100                 break;
3101         case -ENOENT:
3102                 error = ELDLM_NO_LOCK_DATA;
3103                 break;
3104         case -EEXIST:
3105                 error = ELDLM_NAMESPACE_EXISTS;
3106                 break;
3107         case -EBADF:
3108                 error = ELDLM_BAD_NAMESPACE;
3109                 break;
3110         default:
3111                 error = err_no;
3112         }
3113         return error;
3114 }
3115
3116 #if LUSTRE_TRACKS_LOCK_EXP_REFS
3117 void ldlm_dump_export_locks(struct obd_export *exp)
3118 {
3119         spin_lock(&exp->exp_locks_list_guard);
3120         if (!list_empty(&exp->exp_locks_list)) {
3121                 struct ldlm_lock *lock;
3122
3123                 CERROR("dumping locks for export %p,"
3124                        "ignore if the unmount doesn't hang\n", exp);
3125                 list_for_each_entry(lock, &exp->exp_locks_list,
3126                                         l_exp_refs_link)
3127                         LDLM_ERROR(lock, "lock:");
3128         }
3129         spin_unlock(&exp->exp_locks_list_guard);
3130 }
3131 #endif
3132
3133 #ifdef HAVE_SERVER_SUPPORT
3134 static int target_bulk_timeout(void *data)
3135 {
3136         ENTRY;
3137         /* We don't fail the connection here, because having the export
3138          * killed makes the (vital) call to commitrw very sad.
3139          */
3140         RETURN(1);
3141 }
3142
3143 static inline const char *bulk2type(struct ptlrpc_request *req)
3144 {
3145         if (req->rq_bulk_read)
3146                 return "READ";
3147         if (req->rq_bulk_write)
3148                 return "WRITE";
3149         return "UNKNOWN";
3150 }
3151
3152 int target_bulk_io(struct obd_export *exp, struct ptlrpc_bulk_desc *desc,
3153                    struct l_wait_info *lwi)
3154 {
3155         struct ptlrpc_request *req = desc->bd_req;
3156         time64_t start = ktime_get_real_seconds();
3157         time64_t deadline;
3158         int rc = 0;
3159
3160         ENTRY;
3161
3162         /* If there is eviction in progress, wait for it to finish. */
3163         if (unlikely(atomic_read(&exp->exp_obd->obd_evict_inprogress))) {
3164                 *lwi = LWI_INTR(NULL, NULL);
3165                 rc = l_wait_event(exp->exp_obd->obd_evict_inprogress_waitq,
3166                                   !atomic_read(&exp->exp_obd->
3167                                                    obd_evict_inprogress),
3168                                   lwi);
3169         }
3170
3171         /* Check if client was evicted or reconnected already. */
3172         if (exp->exp_failed ||