Whamcloud - gitweb
if client_disconnect_export was called without force flag set,
[fs/lustre-release.git] / lustre / ldlm / ldlm_lib.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  *  Copyright (c) 2003 Cluster File Systems, Inc.
5  *
6  *   This file is part of the Lustre file system, http://www.lustre.org
7  *   Lustre is a trademark of Cluster File Systems, Inc.
8  *
9  *   You may have signed or agreed to another license before downloading
10  *   this software.  If so, you are bound by the terms and conditions
11  *   of that agreement, and the following does not apply to you.  See the
12  *   LICENSE file included with this distribution for more information.
13  *
14  *   If you did not agree to a different license, then this copy of Lustre
15  *   is open source software; you can redistribute it and/or modify it
16  *   under the terms of version 2 of the GNU General Public License as
17  *   published by the Free Software Foundation.
18  *
19  *   In either case, Lustre is distributed in the hope that it will be
20  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
21  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
22  *   license text for more details.
23  */
24
25 #ifndef EXPORT_SYMTAB
26 # define EXPORT_SYMTAB
27 #endif
28 #define DEBUG_SUBSYSTEM S_LDLM
29
30 #ifdef __KERNEL__
31 # include <libcfs/libcfs.h>
32 #else
33 # include <liblustre.h>
34 #endif
35 #include <obd.h>
36 #include <lustre_mds.h>
37 #include <lustre_dlm.h>
38 #include <lustre_net.h>
39 #include <lustre_sec.h>
40 #include "ldlm_internal.h"
41
42
43 /* @priority: if non-zero, move the selected to the list head
44  * @create: if zero, only search in existed connections
45  */
46 static int import_set_conn(struct obd_import *imp, struct obd_uuid *uuid,
47                            int priority, int create)
48 {
49         struct ptlrpc_connection *ptlrpc_conn;
50         struct obd_import_conn *imp_conn = NULL, *item;
51         int rc = 0;
52         ENTRY;
53
54         if (!create && !priority) {
55                 CDEBUG(D_HA, "Nothing to do\n");
56                 RETURN(-EINVAL);
57         }
58
59         ptlrpc_conn = ptlrpc_uuid_to_connection(uuid);
60         if (!ptlrpc_conn) {
61                 CDEBUG(D_HA, "can't find connection %s\n", uuid->uuid);
62                 RETURN (-ENOENT);
63         }
64
65         if (create) {
66                 OBD_ALLOC(imp_conn, sizeof(*imp_conn));
67                 if (!imp_conn) {
68                         GOTO(out_put, rc = -ENOMEM);
69                 }
70         }
71
72         spin_lock(&imp->imp_lock);
73         list_for_each_entry(item, &imp->imp_conn_list, oic_item) {
74                 if (obd_uuid_equals(uuid, &item->oic_uuid)) {
75                         if (priority) {
76                                 list_del(&item->oic_item);
77                                 list_add(&item->oic_item, &imp->imp_conn_list);
78                                 item->oic_last_attempt = 0;
79                         }
80                         CDEBUG(D_HA, "imp %p@%s: found existing conn %s%s\n",
81                                imp, imp->imp_obd->obd_name, uuid->uuid,
82                                (priority ? ", moved to head" : ""));
83                         spin_unlock(&imp->imp_lock);
84                         GOTO(out_free, rc = 0);
85                 }
86         }
87         /* not found */
88         if (create) {
89                 imp_conn->oic_conn = ptlrpc_conn;
90                 imp_conn->oic_uuid = *uuid;
91                 item->oic_last_attempt = 0;
92                 if (priority)
93                         list_add(&imp_conn->oic_item, &imp->imp_conn_list);
94                 else
95                         list_add_tail(&imp_conn->oic_item, &imp->imp_conn_list);
96                 CDEBUG(D_HA, "imp %p@%s: add connection %s at %s\n",
97                        imp, imp->imp_obd->obd_name, uuid->uuid,
98                        (priority ? "head" : "tail"));
99         } else {
100                 spin_unlock(&imp->imp_lock);
101                 GOTO(out_free, rc = -ENOENT);
102         }
103
104         spin_unlock(&imp->imp_lock);
105         RETURN(0);
106 out_free:
107         if (imp_conn)
108                 OBD_FREE(imp_conn, sizeof(*imp_conn));
109 out_put:
110         ptlrpc_put_connection(ptlrpc_conn);
111         RETURN(rc);
112 }
113
114 int import_set_conn_priority(struct obd_import *imp, struct obd_uuid *uuid)
115 {
116         return import_set_conn(imp, uuid, 1, 0);
117 }
118
119 int client_import_add_conn(struct obd_import *imp, struct obd_uuid *uuid,
120                            int priority)
121 {
122         return import_set_conn(imp, uuid, priority, 1);
123 }
124
125 int client_import_del_conn(struct obd_import *imp, struct obd_uuid *uuid)
126 {
127         struct obd_import_conn *imp_conn;
128         struct obd_export *dlmexp;
129         int rc = -ENOENT;
130         ENTRY;
131
132         spin_lock(&imp->imp_lock);
133         if (list_empty(&imp->imp_conn_list)) {
134                 LASSERT(!imp->imp_connection);
135                 GOTO(out, rc);
136         }
137
138         list_for_each_entry(imp_conn, &imp->imp_conn_list, oic_item) {
139                 if (!obd_uuid_equals(uuid, &imp_conn->oic_uuid))
140                         continue;
141                 LASSERT(imp_conn->oic_conn);
142
143                 /* is current conn? */
144                 if (imp_conn == imp->imp_conn_current) {
145                         LASSERT(imp_conn->oic_conn == imp->imp_connection);
146
147                         if (imp->imp_state != LUSTRE_IMP_CLOSED &&
148                             imp->imp_state != LUSTRE_IMP_DISCON) {
149                                 CERROR("can't remove current connection\n");
150                                 GOTO(out, rc = -EBUSY);
151                         }
152
153                         ptlrpc_put_connection(imp->imp_connection);
154                         imp->imp_connection = NULL;
155
156                         dlmexp = class_conn2export(&imp->imp_dlm_handle);
157                         if (dlmexp && dlmexp->exp_connection) {
158                                 LASSERT(dlmexp->exp_connection ==
159                                         imp_conn->oic_conn);
160                                 ptlrpc_put_connection(dlmexp->exp_connection);
161                                 dlmexp->exp_connection = NULL;
162                         }
163                 }
164
165                 list_del(&imp_conn->oic_item);
166                 ptlrpc_put_connection(imp_conn->oic_conn);
167                 OBD_FREE(imp_conn, sizeof(*imp_conn));
168                 CDEBUG(D_HA, "imp %p@%s: remove connection %s\n",
169                        imp, imp->imp_obd->obd_name, uuid->uuid);
170                 rc = 0;
171                 break;
172         }
173 out:
174         spin_unlock(&imp->imp_lock);
175         if (rc == -ENOENT)
176                 CERROR("connection %s not found\n", uuid->uuid);
177         RETURN(rc);
178 }
179
180 static void destroy_import(struct obd_import *imp)
181 {
182         /* drop security policy instance after all rpc finished/aborted
183          * to let all busy credentials be released.
184          */
185         class_import_get(imp);
186         class_destroy_import(imp);
187         sptlrpc_import_put_sec(imp);
188         class_import_put(imp);
189 }
190
191 /* configure an RPC client OBD device
192  *
193  * lcfg parameters:
194  * 1 - client UUID
195  * 2 - server UUID
196  * 3 - inactive-on-startup
197  */
198 int client_obd_setup(struct obd_device *obddev, struct lustre_cfg *lcfg)
199 {
200         struct client_obd *cli = &obddev->u.cli;
201         struct obd_import *imp;
202         struct obd_uuid server_uuid;
203         int rq_portal, rp_portal, connect_op;
204         char *name = obddev->obd_type->typ_name;
205         int rc;
206         ENTRY;
207
208         /* In a more perfect world, we would hang a ptlrpc_client off of
209          * obd_type and just use the values from there. */
210         if (!strcmp(name, LUSTRE_OSC_NAME)) {
211                 rq_portal = OST_REQUEST_PORTAL;
212                 rp_portal = OSC_REPLY_PORTAL;
213                 connect_op = OST_CONNECT;
214         } else if (!strcmp(name, LUSTRE_MDC_NAME)) {
215                 rq_portal = MDS_REQUEST_PORTAL;
216                 rp_portal = MDC_REPLY_PORTAL;
217                 connect_op = MDS_CONNECT;
218         } else if (!strcmp(name, LUSTRE_MGC_NAME)) {
219                 rq_portal = MGS_REQUEST_PORTAL;
220                 rp_portal = MGC_REPLY_PORTAL;
221                 connect_op = MGS_CONNECT;
222         } else {
223                 CERROR("unknown client OBD type \"%s\", can't setup\n",
224                        name);
225                 RETURN(-EINVAL);
226         }
227
228         if (LUSTRE_CFG_BUFLEN(lcfg, 1) < 1) {
229                 CERROR("requires a TARGET UUID\n");
230                 RETURN(-EINVAL);
231         }
232
233         if (LUSTRE_CFG_BUFLEN(lcfg, 1) > 37) {
234                 CERROR("client UUID must be less than 38 characters\n");
235                 RETURN(-EINVAL);
236         }
237
238         if (LUSTRE_CFG_BUFLEN(lcfg, 2) < 1) {
239                 CERROR("setup requires a SERVER UUID\n");
240                 RETURN(-EINVAL);
241         }
242
243         if (LUSTRE_CFG_BUFLEN(lcfg, 2) > 37) {
244                 CERROR("target UUID must be less than 38 characters\n");
245                 RETURN(-EINVAL);
246         }
247
248         sema_init(&cli->cl_sem, 1);
249         sema_init(&cli->cl_mgc_sem, 1);
250         cli->cl_sec_conf.sfc_rpc_flavor = SPTLRPC_FLVR_NULL;
251         cli->cl_sec_conf.sfc_bulk_csum = BULK_CSUM_ALG_NULL;
252         cli->cl_sec_conf.sfc_bulk_priv = BULK_PRIV_ALG_NULL;
253         cli->cl_sec_conf.sfc_flags = 0;
254         cli->cl_conn_count = 0;
255         memcpy(server_uuid.uuid, lustre_cfg_buf(lcfg, 2),
256                min_t(unsigned int, LUSTRE_CFG_BUFLEN(lcfg, 2),
257                      sizeof(server_uuid)));
258
259         cli->cl_dirty = 0;
260         cli->cl_avail_grant = 0;
261         /* FIXME: should limit this for the sum of all cl_dirty_max */
262         cli->cl_dirty_max = OSC_MAX_DIRTY_DEFAULT * 1024 * 1024;
263         if (cli->cl_dirty_max >> CFS_PAGE_SHIFT > num_physpages / 8)
264                 cli->cl_dirty_max = num_physpages << (CFS_PAGE_SHIFT - 3);
265         CFS_INIT_LIST_HEAD(&cli->cl_cache_waiters);
266         CFS_INIT_LIST_HEAD(&cli->cl_loi_ready_list);
267         CFS_INIT_LIST_HEAD(&cli->cl_loi_write_list);
268         CFS_INIT_LIST_HEAD(&cli->cl_loi_read_list);
269         client_obd_list_lock_init(&cli->cl_loi_list_lock);
270         cli->cl_r_in_flight = 0;
271         cli->cl_w_in_flight = 0;
272
273         spin_lock_init(&cli->cl_read_rpc_hist.oh_lock);
274         spin_lock_init(&cli->cl_write_rpc_hist.oh_lock);
275         spin_lock_init(&cli->cl_read_page_hist.oh_lock);
276         spin_lock_init(&cli->cl_write_page_hist.oh_lock);
277         spin_lock_init(&cli->cl_read_offset_hist.oh_lock);
278         spin_lock_init(&cli->cl_write_offset_hist.oh_lock);
279 #ifdef ENABLE_CHECKSUM
280         cli->cl_checksum = 1;
281 #endif
282         atomic_set(&cli->cl_resends, OSC_DEFAULT_RESENDS);
283
284         /* This value may be changed at connect time in
285            ptlrpc_connect_interpret. */
286         cli->cl_max_pages_per_rpc = min((int)PTLRPC_MAX_BRW_PAGES,
287                                         (int)(1024 * 1024 >> CFS_PAGE_SHIFT));
288
289         if (!strcmp(name, LUSTRE_MDC_NAME)) {
290                 cli->cl_max_rpcs_in_flight = MDC_MAX_RIF_DEFAULT;
291         } else if (num_physpages >> (20 - CFS_PAGE_SHIFT) <= 128 /* MB */) {
292                 cli->cl_max_rpcs_in_flight = 2;
293         } else if (num_physpages >> (20 - CFS_PAGE_SHIFT) <= 256 /* MB */) {
294                 cli->cl_max_rpcs_in_flight = 3;
295         } else if (num_physpages >> (20 - CFS_PAGE_SHIFT) <= 512 /* MB */) {
296                 cli->cl_max_rpcs_in_flight = 4;
297         } else {
298                 cli->cl_max_rpcs_in_flight = OSC_MAX_RIF_DEFAULT;
299         }
300
301         rc = ldlm_get_ref();
302         if (rc) {
303                 CERROR("ldlm_get_ref failed: %d\n", rc);
304                 GOTO(err, rc);
305         }
306
307         ptlrpc_init_client(rq_portal, rp_portal, name,
308                            &obddev->obd_ldlm_client);
309
310         imp = class_new_import(obddev);
311         if (imp == NULL)
312                 GOTO(err_ldlm, rc = -ENOENT);
313         imp->imp_client = &obddev->obd_ldlm_client;
314         imp->imp_connect_op = connect_op;
315         imp->imp_initial_recov = 1;
316         imp->imp_initial_recov_bk = 0;
317         CFS_INIT_LIST_HEAD(&imp->imp_pinger_chain);
318         memcpy(cli->cl_target_uuid.uuid, lustre_cfg_buf(lcfg, 1),
319                LUSTRE_CFG_BUFLEN(lcfg, 1));
320         class_import_put(imp);
321
322         rc = client_import_add_conn(imp, &server_uuid, 1);
323         if (rc) {
324                 CERROR("can't add initial connection\n");
325                 GOTO(err_import, rc);
326         }
327
328         cli->cl_import = imp;
329         /* cli->cl_max_mds_{easize,cookiesize} updated by mdc_init_ea_size() */
330         cli->cl_max_mds_easize = sizeof(struct lov_mds_md);
331         cli->cl_max_mds_cookiesize = sizeof(struct llog_cookie);
332
333         if (LUSTRE_CFG_BUFLEN(lcfg, 3) > 0) {
334                 if (!strcmp(lustre_cfg_string(lcfg, 3), "inactive")) {
335                         CDEBUG(D_HA, "marking %s %s->%s as inactive\n",
336                                name, obddev->obd_name,
337                                cli->cl_target_uuid.uuid);
338                         spin_lock(&imp->imp_lock);
339                         imp->imp_invalid = 1;
340                         spin_unlock(&imp->imp_lock);
341                 }
342         }
343
344         cli->cl_qchk_stat = CL_NOT_QUOTACHECKED;
345
346         RETURN(rc);
347
348 err_import:
349         class_destroy_import(imp);
350 err_ldlm:
351         ldlm_put_ref(0);
352 err:
353         RETURN(rc);
354
355 }
356
357 int client_obd_cleanup(struct obd_device *obddev)
358 {
359         ENTRY;
360         ldlm_put_ref(obddev->obd_force);
361         RETURN(0);
362 }
363
364 /* ->o_connect() method for client side (OSC and MDC and MGC) */
365 int client_connect_import(const struct lu_env *env,
366                           struct lustre_handle *dlm_handle,
367                           struct obd_device *obd, struct obd_uuid *cluuid,
368                           struct obd_connect_data *data)
369 {
370         struct client_obd *cli = &obd->u.cli;
371         struct obd_import *imp = cli->cl_import;
372         struct obd_export *exp;
373         struct obd_connect_data *ocd;
374         struct ldlm_namespace *to_be_freed = NULL;
375         int rc;
376         ENTRY;
377
378         mutex_down(&cli->cl_sem);
379         rc = class_connect(dlm_handle, obd, cluuid);
380         if (rc)
381                 GOTO(out_sem, rc);
382
383         cli->cl_conn_count++;
384         if (cli->cl_conn_count > 1)
385                 GOTO(out_sem, rc);
386         exp = class_conn2export(dlm_handle);
387
388         if (obd->obd_namespace != NULL)
389                 CERROR("already have namespace!\n");
390         obd->obd_namespace = ldlm_namespace_new(obd->obd_name,
391                                                 LDLM_NAMESPACE_CLIENT,
392                                                 LDLM_NAMESPACE_GREEDY);
393         if (obd->obd_namespace == NULL)
394                 GOTO(out_disco, rc = -ENOMEM);
395
396         imp->imp_dlm_handle = *dlm_handle;
397         rc = ptlrpc_init_import(imp);
398         if (rc != 0)
399                 GOTO(out_ldlm, rc);
400
401         rc = sptlrpc_import_get_sec(imp, NULL, cli->cl_sec_conf.sfc_rpc_flavor,
402                                     cli->cl_sec_conf.sfc_flags);
403         if (rc)
404                 GOTO(out_ldlm, rc);
405
406         ocd = &imp->imp_connect_data;
407         if (data) {
408                 *ocd = *data;
409                 imp->imp_connect_flags_orig = data->ocd_connect_flags;
410         }
411
412         rc = ptlrpc_connect_import(imp, NULL);
413         if (rc != 0) {
414                 LASSERT (imp->imp_state == LUSTRE_IMP_DISCON);
415                 GOTO(out_ldlm, rc);
416         }
417         LASSERT(exp->exp_connection);
418
419         if (data) {
420                 LASSERT((ocd->ocd_connect_flags & data->ocd_connect_flags) ==
421                         ocd->ocd_connect_flags);
422                 data->ocd_connect_flags = ocd->ocd_connect_flags;
423         }
424
425         ptlrpc_pinger_add_import(imp);
426
427         EXIT;
428
429         if (rc) {
430 out_ldlm:
431                 ldlm_namespace_free_prior(obd->obd_namespace);
432                 to_be_freed = obd->obd_namespace;
433                 obd->obd_namespace = NULL;
434 out_disco:
435                 cli->cl_conn_count--;
436                 class_disconnect(exp);
437         } else {
438                 class_export_put(exp);
439         }
440 out_sem:
441         mutex_up(&cli->cl_sem);
442         if (to_be_freed)
443                 ldlm_namespace_free_post(to_be_freed, 1);
444
445         return rc;
446 }
447
448 int client_disconnect_export(struct obd_export *exp)
449 {
450         struct obd_device *obd = class_exp2obd(exp);
451         struct client_obd *cli;
452         struct obd_import *imp;
453         int rc = 0, err;
454         struct ldlm_namespace *to_be_freed = NULL;
455         ENTRY;
456
457         if (!obd) {
458                 CERROR("invalid export for disconnect: exp %p cookie "LPX64"\n",
459                        exp, exp ? exp->exp_handle.h_cookie : -1);
460                 RETURN(-EINVAL);
461         }
462
463         cli = &obd->u.cli;
464         imp = cli->cl_import;
465
466         mutex_down(&cli->cl_sem);
467         if (!cli->cl_conn_count) {
468                 CERROR("disconnecting disconnected device (%s)\n",
469                        obd->obd_name);
470                 GOTO(out_sem, rc = -EINVAL);
471         }
472
473         cli->cl_conn_count--;
474         if (cli->cl_conn_count)
475                 GOTO(out_no_disconnect, rc = 0);
476
477         /* Mark import deactivated now, so we don't try to reconnect if any
478          * of the cleanup RPCs fails (e.g. ldlm cancel, etc).  We don't
479          * fully deactivate the import, or that would drop all requests. */
480         spin_lock(&imp->imp_lock);
481         imp->imp_deactive = 1;
482         spin_unlock(&imp->imp_lock);
483         
484         /* Some non-replayable imports (MDS's OSCs) are pinged, so just
485          * delete it regardless.  (It's safe to delete an import that was
486          * never added.) */
487         (void)ptlrpc_pinger_del_import(imp);
488
489         if (obd->obd_namespace != NULL) {
490                 /* obd_force == local only */
491                 ldlm_cli_cancel_unused(obd->obd_namespace, NULL,
492                                        obd->obd_force ? LDLM_FL_LOCAL_ONLY:0,
493                                        NULL);
494                 ldlm_namespace_free_prior(obd->obd_namespace);
495                 to_be_freed = obd->obd_namespace;
496         }
497
498         rc = ptlrpc_disconnect_import(imp, 0);
499
500         ptlrpc_invalidate_import(imp);
501         /* set obd_namespace to NULL only after invalidate, because we can have
502          * some connect requests in flight, and his need store a connect flags
503          * in obd_namespace. bug 14260 */
504         obd->obd_namespace = NULL;
505
506         ptlrpc_free_rq_pool(imp->imp_rq_pool);
507         destroy_import(imp);
508         cli->cl_import = NULL;
509
510         EXIT;
511  out_no_disconnect:
512         err = class_disconnect(exp);
513         if (!rc && err)
514                 rc = err;
515  out_sem:
516         mutex_up(&cli->cl_sem);
517         if (to_be_freed)
518                 ldlm_namespace_free_post(to_be_freed, obd->obd_force);
519
520         RETURN(rc);
521 }
522
523 /* --------------------------------------------------------------------------
524  * from old lib/target.c
525  * -------------------------------------------------------------------------- */
526
527 int target_handle_reconnect(struct lustre_handle *conn, struct obd_export *exp,
528                             struct obd_uuid *cluuid, int initial_conn)
529 {
530         ENTRY;
531         if (exp->exp_connection && exp->exp_imp_reverse && !initial_conn) {
532                 struct lustre_handle *hdl;
533                 hdl = &exp->exp_imp_reverse->imp_remote_handle;
534                 /* Might be a re-connect after a partition. */
535                 if (!memcmp(&conn->cookie, &hdl->cookie, sizeof conn->cookie)) {
536                         CWARN("%s: %s reconnecting\n", exp->exp_obd->obd_name,
537                               cluuid->uuid);
538                         conn->cookie = exp->exp_handle.h_cookie;
539                         /* target_handle_connect() treats EALREADY and
540                          * -EALREADY differently.  EALREADY means we are
541                          * doing a valid reconnect from the same client. */
542                         RETURN(EALREADY);
543                 } else {
544                         CERROR("%s reconnecting from %s, "
545                                "handle mismatch (ours "LPX64", theirs "
546                                LPX64")\n", cluuid->uuid,
547                                exp->exp_connection->c_remote_uuid.uuid,
548                                hdl->cookie, conn->cookie);
549                         memset(conn, 0, sizeof *conn);
550                         /* target_handle_connect() treats EALREADY and
551                          * -EALREADY differently.  -EALREADY is an error
552                          * (same UUID, different handle). */
553                         RETURN(-EALREADY);
554                 }
555         }
556
557         conn->cookie = exp->exp_handle.h_cookie;
558         CDEBUG(D_HA, "connect export for UUID '%s' at %p, cookie "LPX64"\n",
559                cluuid->uuid, exp, conn->cookie);
560         RETURN(0);
561 }
562
563 void target_client_add_cb(struct obd_device *obd, __u64 transno, void *cb_data,
564                           int error)
565 {
566         struct obd_export *exp = cb_data;
567
568         CDEBUG(D_RPCTRACE, "%s: committing for initial connect of %s\n",
569                obd->obd_name, exp->exp_client_uuid.uuid);
570
571         spin_lock(&exp->exp_lock);
572         exp->exp_need_sync = 0;
573         spin_unlock(&exp->exp_lock);
574 }
575 EXPORT_SYMBOL(target_client_add_cb);
576
577 int target_handle_connect(struct ptlrpc_request *req)
578 {
579         struct obd_device *target, *targref = NULL;
580         struct obd_export *export = NULL;
581         struct obd_import *revimp;
582         struct lustre_handle conn;
583         struct obd_uuid tgtuuid;
584         struct obd_uuid cluuid;
585         struct obd_uuid remote_uuid;
586         char *str, *tmp;
587         int rc = 0;
588         int initial_conn = 0;
589         struct obd_connect_data *data;
590         int size[2] = { sizeof(struct ptlrpc_body), sizeof(*data) };
591         ENTRY;
592
593         OBD_RACE(OBD_FAIL_TGT_CONN_RACE);
594
595         lustre_set_req_swabbed(req, REQ_REC_OFF);
596         str = lustre_msg_string(req->rq_reqmsg, REQ_REC_OFF, sizeof(tgtuuid)-1);
597         if (str == NULL) {
598                 DEBUG_REQ(D_ERROR, req, "bad target UUID for connect");
599                 GOTO(out, rc = -EINVAL);
600         }
601
602         obd_str2uuid (&tgtuuid, str);
603         target = class_uuid2obd(&tgtuuid);
604         if (!target)
605                 target = class_name2obd(str);
606
607         if (!target || target->obd_stopping || !target->obd_set_up) {
608                 LCONSOLE_ERROR_MSG(0x137, "UUID '%s' is not available "
609                                    " for connect (%s)\n", str,
610                                    !target ? "no target" :
611                                    (target->obd_stopping ? "stopping" :
612                                    "not set up"));
613                 GOTO(out, rc = -ENODEV);
614         }
615
616         if (target->obd_no_conn) {
617                 LCONSOLE_WARN("%s: temporarily refusing client connection "
618                               "from %s\n", target->obd_name, 
619                               libcfs_nid2str(req->rq_peer.nid));
620                 GOTO(out, rc = -EAGAIN);
621         }
622
623         /* Make sure the target isn't cleaned up while we're here. Yes,
624            there's still a race between the above check and our incref here.
625            Really, class_uuid2obd should take the ref. */
626         targref = class_incref(target);
627
628         lustre_set_req_swabbed(req, REQ_REC_OFF + 1);
629         str = lustre_msg_string(req->rq_reqmsg, REQ_REC_OFF + 1,
630                                 sizeof(cluuid) - 1);
631         if (str == NULL) {
632                 DEBUG_REQ(D_ERROR, req, "bad client UUID for connect");
633                 GOTO(out, rc = -EINVAL);
634         }
635
636         obd_str2uuid (&cluuid, str);
637
638         /* XXX extract a nettype and format accordingly */
639         switch (sizeof(lnet_nid_t)) {
640                 /* NB the casts only avoid compiler warnings */
641         case 8:
642                 snprintf(remote_uuid.uuid, sizeof remote_uuid,
643                          "NET_"LPX64"_UUID", (__u64)req->rq_peer.nid);
644                 break;
645         case 4:
646                 snprintf(remote_uuid.uuid, sizeof remote_uuid,
647                          "NET_%x_UUID", (__u32)req->rq_peer.nid);
648                 break;
649         default:
650                 LBUG();
651         }
652
653         tmp = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF + 2, sizeof conn);
654         if (tmp == NULL)
655                 GOTO(out, rc = -EPROTO);
656
657         memcpy(&conn, tmp, sizeof conn);
658
659         data = lustre_swab_reqbuf(req, REQ_REC_OFF + 3, sizeof(*data),
660                                   lustre_swab_connect);
661
662         if (!data)
663                 GOTO(out, rc = -EPROTO);
664
665         rc = lustre_pack_reply(req, 2, size, NULL);
666         if (rc)
667                 GOTO(out, rc);
668
669         if (lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_LIBCLIENT) {
670                 if (!data) {
671                         DEBUG_REQ(D_WARNING, req, "Refusing old (unversioned) "
672                                   "libclient connection attempt");
673                         GOTO(out, rc = -EPROTO);
674                 } else if (data->ocd_version < LUSTRE_VERSION_CODE -
675                                                LUSTRE_VERSION_ALLOWED_OFFSET ||
676                            data->ocd_version > LUSTRE_VERSION_CODE +
677                                                LUSTRE_VERSION_ALLOWED_OFFSET) {
678                         DEBUG_REQ(D_WARNING, req, "Refusing %s (%d.%d.%d.%d) "
679                                   "libclient connection attempt",
680                                   data->ocd_version < LUSTRE_VERSION_CODE ?
681                                   "old" : "new",
682                                   OBD_OCD_VERSION_MAJOR(data->ocd_version),
683                                   OBD_OCD_VERSION_MINOR(data->ocd_version),
684                                   OBD_OCD_VERSION_PATCH(data->ocd_version),
685                                   OBD_OCD_VERSION_FIX(data->ocd_version));
686                         data = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
687                                               offsetof(typeof(*data),
688                                                        ocd_version) +
689                                               sizeof(data->ocd_version));
690                         if (data) {
691                                 data->ocd_connect_flags = OBD_CONNECT_VERSION;
692                                 data->ocd_version = LUSTRE_VERSION_CODE;
693                         }
694                         GOTO(out, rc = -EPROTO);
695                 }
696         }
697
698         if (lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_INITIAL)
699                 initial_conn = 1;
700
701         /* lctl gets a backstage, all-access pass. */
702         if (obd_uuid_equals(&cluuid, &target->obd_uuid))
703                 goto dont_check_exports;
704
705         spin_lock(&target->obd_dev_lock);
706         export = lustre_hash_get_object_by_key(target->obd_uuid_hash_body, &cluuid);
707
708         if (export != NULL && export->exp_connecting) { /* bug 9635, et. al. */
709                 CWARN("%s: exp %p already connecting\n",
710                       export->exp_obd->obd_name, export);
711                 class_export_put(export);
712                 export = NULL;
713                 rc = -EALREADY;
714         } else if (export != NULL && export->exp_connection != NULL &&
715                    req->rq_peer.nid != export->exp_connection->c_peer.nid) {
716                 /* make darn sure this is coming from the same peer
717                  * if the UUIDs matched */
718                   CWARN("%s: cookie %s seen on new NID %s when "
719                           "existing NID %s is already connected\n",
720                         target->obd_name, cluuid.uuid,
721                   libcfs_nid2str(req->rq_peer.nid),
722                   libcfs_nid2str(export->exp_connection->c_peer.nid));
723                   class_export_put(export);
724                   export = NULL;
725                   rc = -EALREADY;
726         } else if (export != NULL) {
727                 spin_lock(&export->exp_lock);
728                 export->exp_connecting = 1;
729                 spin_unlock(&export->exp_lock);
730                 class_export_put(export);
731                 spin_unlock(&target->obd_dev_lock);
732                 LASSERT(export->exp_obd == target);
733
734                 rc = target_handle_reconnect(&conn, export, &cluuid, initial_conn);
735         }
736
737         /* If we found an export, we already unlocked. */
738         if (!export) {
739                 spin_unlock(&target->obd_dev_lock);
740                 OBD_FAIL_TIMEOUT(OBD_FAIL_TGT_DELAY_CONNECT, 2 * obd_timeout);
741         } else if (req->rq_export == NULL &&
742                    atomic_read(&export->exp_rpc_count) > 0) {
743                 CWARN("%s: refuse connection from %s/%s to 0x%p/%d\n",
744                       target->obd_name, cluuid.uuid,
745                       libcfs_nid2str(req->rq_peer.nid),
746                       export, atomic_read(&export->exp_refcount));
747                 GOTO(out, rc = -EBUSY);
748         } else if (req->rq_export != NULL &&
749                    (atomic_read(&export->exp_rpc_count) > 1)) {
750                 CWARN("%s: refuse reconnection from %s@%s to 0x%p/%d\n",
751                       target->obd_name, cluuid.uuid,
752                       libcfs_nid2str(req->rq_peer.nid),
753                       export, atomic_read(&export->exp_rpc_count));
754                 GOTO(out, rc = -EBUSY);
755         } else if (lustre_msg_get_conn_cnt(req->rq_reqmsg) == 1 &&
756                    !initial_conn) {
757                 CERROR("%s: NID %s (%s) reconnected with 1 conn_cnt; "
758                        "cookies not random?\n", target->obd_name,
759                        libcfs_nid2str(req->rq_peer.nid), cluuid.uuid);
760                 GOTO(out, rc = -EALREADY);
761         } else {
762                 OBD_FAIL_TIMEOUT(OBD_FAIL_TGT_DELAY_RECONNECT, 2 * obd_timeout);
763                 if (req->rq_export == NULL && initial_conn)
764                        export->exp_last_request_time = 
765                                max(export->exp_last_request_time,
766                                    (time_t)CURRENT_SECONDS);
767         }
768
769         /* We want to handle EALREADY but *not* -EALREADY from
770          * target_handle_reconnect(), return reconnection state in a flag */
771         if (rc == EALREADY) {
772                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_RECONNECT);
773                 rc = 0;
774         } else if (rc) {
775                 GOTO(out, rc);
776         }
777         /* Tell the client if we're in recovery. */
778         /* If this is the first client, start the recovery timer */
779         CWARN("%s: connection from %s@%s %st"LPU64" exp %p cur %ld last %ld\n",
780                target->obd_name, cluuid.uuid, libcfs_nid2str(req->rq_peer.nid),
781               target->obd_recovering ? "recovering/" : "", data->ocd_transno,
782               export, (long)CURRENT_SECONDS, 
783               export ? (long)export->exp_last_request_time : 0);
784
785
786         if (target->obd_recovering) {
787                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_RECOVERING);
788                 target_start_recovery_timer(target);
789         }
790
791         /* Tell the client if we support replayable requests */
792         if (target->obd_replayable)
793                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_REPLAYABLE);
794
795         if (export == NULL) {
796                 if (target->obd_recovering) {
797                         CERROR("%s: denying connection for new client %s (%s): "
798                                "%d clients in recovery for %lds\n",
799                                target->obd_name,
800                                libcfs_nid2str(req->rq_peer.nid), cluuid.uuid,
801                                target->obd_recoverable_clients,
802                                cfs_duration_sec(cfs_time_sub(cfs_timer_deadline(&target->obd_recovery_timer),
803                                                              cfs_time_current())));
804                         rc = -EBUSY;
805                 } else {
806 dont_check_exports:
807                         rc = obd_connect(req->rq_svc_thread->t_env,
808                                          &conn, target, &cluuid, data);
809                 }
810         } else {
811                 rc = obd_reconnect(export, target, &cluuid, data);
812         }
813         if (rc)
814                 GOTO(out, rc);
815         /* Return only the parts of obd_connect_data that we understand, so the
816          * client knows that we don't understand the rest. */
817         if (data)
818                 memcpy(lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
819                                       sizeof(*data)),
820                        data, sizeof(*data));
821
822         /* If all else goes well, this is our RPC return code. */
823         req->rq_status = 0;
824
825         lustre_msg_set_handle(req->rq_repmsg, &conn);
826
827         /* ownership of this export ref transfers to the request AFTER we
828          * drop any previous reference the request had, but we don't want
829          * that to go to zero before we get our new export reference. */
830         export = class_conn2export(&conn);
831         if (!export) {
832                 DEBUG_REQ(D_ERROR, req, "Missing export!");
833                 GOTO(out, rc = -ENODEV);
834         }
835
836         /* If the client and the server are the same node, we will already
837          * have an export that really points to the client's DLM export,
838          * because we have a shared handles table.
839          *
840          * XXX this will go away when shaver stops sending the "connect" handle
841          * in the real "remote handle" field of the request --phik 24 Apr 2003
842          */
843         if (req->rq_export != NULL)
844                 class_export_put(req->rq_export);
845
846         req->rq_export = export;
847
848         spin_lock(&export->exp_lock);
849         if (initial_conn) {
850                 lustre_msg_set_conn_cnt(req->rq_repmsg, export->exp_conn_cnt + 1);
851         } else if (export->exp_conn_cnt >= lustre_msg_get_conn_cnt(req->rq_reqmsg)) {
852                 spin_unlock(&export->exp_lock);
853                 CERROR("%s: %s already connected at higher conn_cnt: %d > %d\n",
854                        cluuid.uuid, libcfs_nid2str(req->rq_peer.nid),
855                        export->exp_conn_cnt,
856                        lustre_msg_get_conn_cnt(req->rq_reqmsg));
857
858                 GOTO(out, rc = -EALREADY);
859         }
860         export->exp_conn_cnt = lustre_msg_get_conn_cnt(req->rq_reqmsg);
861
862         /* request from liblustre?  Don't evict it for not pinging. */
863         if (lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_LIBCLIENT) {
864                 export->exp_libclient = 1;
865                 spin_unlock(&export->exp_lock);
866                 
867                 spin_lock(&target->obd_dev_lock);
868                 list_del_init(&export->exp_obd_chain_timed);
869                 spin_unlock(&target->obd_dev_lock);
870         } else {
871                 spin_unlock(&export->exp_lock);
872         }
873
874         if (export->exp_connection != NULL)
875                 ptlrpc_put_connection(export->exp_connection);
876         export->exp_connection = ptlrpc_get_connection(req->rq_peer,
877                                                        req->rq_self,
878                                                        &remote_uuid);
879
880         spin_lock(&target->obd_dev_lock);
881         /* Export might be hashed already, e.g. if this is reconnect */
882         if (hlist_unhashed(&export->exp_nid_hash))
883                 lustre_hash_additem(export->exp_obd->obd_nid_hash_body, 
884                                     &export->exp_connection->c_peer.nid, 
885                                     &export->exp_nid_hash);
886         spin_unlock(&target->obd_dev_lock);
887
888         spin_lock_bh(&target->obd_processing_task_lock);
889         if (target->obd_recovering && !export->exp_in_recovery) {
890                 spin_lock(&export->exp_lock);
891                 export->exp_in_recovery = 1;
892                 export->exp_req_replay_needed = 1;
893                 export->exp_lock_replay_needed = 1;                
894                 spin_unlock(&export->exp_lock);
895                 if ((lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_TRANSNO)
896                      && data->ocd_transno < target->obd_next_recovery_transno)
897                         target->obd_next_recovery_transno = data->ocd_transno;
898                 target->obd_connected_clients++;
899                 /* each connected client is counted as recoverable */
900                 target->obd_recoverable_clients++;
901                 atomic_inc(&target->obd_req_replay_clients);
902                 atomic_inc(&target->obd_lock_replay_clients);
903                 if (target->obd_connected_clients == 
904                     target->obd_max_recoverable_clients)
905                         wake_up(&target->obd_next_transno_waitq);
906         }
907         spin_unlock_bh(&target->obd_processing_task_lock);
908         memcpy(&conn,
909                lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF + 2, sizeof conn),
910                sizeof conn);
911
912         if (export->exp_imp_reverse != NULL) {
913                 /* destroyed import can be still referenced in ctxt */
914                 obd_set_info_async(export, strlen(KEY_REVIMP_UPD), 
915                                    KEY_REVIMP_UPD, 0, NULL, NULL);
916                 destroy_import(export->exp_imp_reverse);
917         }
918
919         /* for the rest part, we return -ENOTCONN in case of errors
920          * in order to let client initialize connection again.
921          */
922         revimp = export->exp_imp_reverse = class_new_import(target);
923         if (!revimp) {
924                 CERROR("fail to alloc new reverse import.\n");
925                 GOTO(out, rc = -ENOTCONN);
926         }
927
928         revimp->imp_connection = ptlrpc_connection_addref(export->exp_connection);
929         revimp->imp_client = &export->exp_obd->obd_ldlm_client;
930         revimp->imp_remote_handle = conn;
931         revimp->imp_dlm_fake = 1;
932         revimp->imp_state = LUSTRE_IMP_FULL;
933
934         if (lustre_msg_get_op_flags(req->rq_reqmsg) & MSG_CONNECT_NEXT_VER) {
935                 revimp->imp_msg_magic = LUSTRE_MSG_MAGIC_V2;
936                 lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_NEXT_VER);
937         }
938
939         rc = sptlrpc_import_get_sec(revimp, req->rq_svc_ctx,
940                                     req->rq_sec_flavor, 0);
941         if (rc) {
942                 CERROR("Failed to get sec for reverse import: %d\n", rc);
943                 export->exp_imp_reverse = NULL;
944                 class_destroy_import(revimp);
945         }
946
947         class_import_put(revimp);
948 out:
949         if (export) {
950                 spin_lock(&export->exp_lock);
951                 export->exp_connecting = 0;
952                 spin_unlock(&export->exp_lock);
953         }
954         if (targref)
955                 class_decref(targref);
956         if (rc)
957                 req->rq_status = rc;
958         RETURN(rc);
959 }
960
961 int target_handle_disconnect(struct ptlrpc_request *req)
962 {
963         int rc;
964         ENTRY;
965
966         rc = lustre_pack_reply(req, 1, NULL, NULL);
967         if (rc)
968                 RETURN(rc);
969
970         /* keep the rq_export around so we can send the reply */
971         req->rq_status = obd_disconnect(class_export_get(req->rq_export));
972         
973         RETURN(0);
974 }
975
976 void target_destroy_export(struct obd_export *exp)
977 {
978         /* exports created from last_rcvd data, and "fake"
979            exports created by lctl don't have an import */
980         if (exp->exp_imp_reverse != NULL)
981                 destroy_import(exp->exp_imp_reverse);
982
983         /* We cancel locks at disconnect time, but this will catch any locks
984          * granted in a race with recovery-induced disconnect. */
985         if (exp->exp_obd->obd_namespace != NULL)
986                 ldlm_cancel_locks_for_export(exp);
987 }
988
989 /*
990  * Recovery functions
991  */
992
993 struct ptlrpc_request *ptlrpc_clone_req( struct ptlrpc_request *orig_req)
994 {
995         struct ptlrpc_request *copy_req;
996         struct lustre_msg *copy_reqmsg;
997         struct ptlrpc_user_desc *udesc = NULL;
998
999         OBD_ALLOC_PTR(copy_req);
1000         if (!copy_req)
1001                 return NULL;
1002         OBD_ALLOC(copy_reqmsg, orig_req->rq_reqlen);
1003         if (!copy_reqmsg){
1004                 OBD_FREE_PTR(copy_req);
1005                 return NULL;
1006         }
1007
1008         if (orig_req->rq_user_desc) {
1009                 int ngroups = orig_req->rq_user_desc->pud_ngroups;
1010
1011                 OBD_ALLOC(udesc, sptlrpc_user_desc_size(ngroups));
1012                 if (!udesc) {
1013                         OBD_FREE(copy_reqmsg, orig_req->rq_reqlen);
1014                         OBD_FREE_PTR(copy_req);
1015                         return NULL;
1016                 }
1017                 memcpy(udesc, orig_req->rq_user_desc,
1018                        sptlrpc_user_desc_size(ngroups));
1019         }
1020
1021         *copy_req = *orig_req;
1022         memcpy(copy_reqmsg, orig_req->rq_reqmsg, orig_req->rq_reqlen);
1023         copy_req->rq_reqmsg = copy_reqmsg;
1024         copy_req->rq_user_desc = udesc;
1025
1026         class_export_get(copy_req->rq_export);
1027         CFS_INIT_LIST_HEAD(&copy_req->rq_list);
1028         sptlrpc_svc_ctx_addref(copy_req);
1029
1030         if (copy_req->rq_reply_state) {
1031                 /* the copied req takes over the reply state */
1032                 orig_req->rq_reply_state = NULL;
1033                 /* to catch further access */
1034                 orig_req->rq_repmsg = NULL;
1035                 orig_req->rq_replen = 0;
1036         }
1037
1038         return copy_req;
1039 }
1040
1041 void ptlrpc_free_clone( struct ptlrpc_request *req)
1042 {
1043         if (req->rq_reply_state) {
1044                 ptlrpc_rs_decref(req->rq_reply_state);
1045                 req->rq_reply_state = NULL;
1046         }
1047
1048         sptlrpc_svc_ctx_decref(req);
1049         class_export_put(req->rq_export);
1050         list_del(&req->rq_list);
1051
1052         if (req->rq_user_desc) {
1053                 int ngroups = req->rq_user_desc->pud_ngroups;
1054                 OBD_FREE(req->rq_user_desc, sptlrpc_user_desc_size(ngroups));
1055         }
1056         OBD_FREE(req->rq_reqmsg, req->rq_reqlen);
1057         OBD_FREE_PTR(req);
1058 }
1059
1060 #ifdef __KERNEL__
1061 static void target_finish_recovery(struct obd_device *obd)
1062 {
1063         ENTRY;
1064         LCONSOLE_INFO("%s: sending delayed replies to recovered clients\n",
1065                       obd->obd_name);
1066
1067         ldlm_reprocess_all_ns(obd->obd_namespace);
1068
1069         /* when recovery finished, cleanup orphans on mds and ost */
1070         if (OBT(obd) && OBP(obd, postrecov)) {
1071                 int rc = OBP(obd, postrecov)(obd);
1072                 LCONSOLE_WARN("%s: recovery %s: rc %d\n", obd->obd_name,
1073                               rc < 0 ? "failed" : "complete", rc);
1074         }
1075
1076         obd->obd_recovery_end = CURRENT_SECONDS;
1077         EXIT;
1078 }
1079
1080 static void abort_req_replay_queue(struct obd_device *obd)
1081 {
1082         struct ptlrpc_request *req, *n;
1083
1084         list_for_each_entry_safe(req, n, &obd->obd_req_replay_queue, rq_list) {
1085                 DEBUG_REQ(D_WARNING, req, "aborted:");
1086                 req->rq_status = -ENOTCONN;
1087                 if (ptlrpc_error(req)) {
1088                         DEBUG_REQ(D_ERROR, req,
1089                                   "failed abort_req_reply; skipping");
1090                 }
1091                 ptlrpc_free_clone(req);
1092         }
1093 }
1094
1095 static void abort_lock_replay_queue(struct obd_device *obd)
1096 {
1097         struct ptlrpc_request *req, *n;
1098
1099         list_for_each_entry_safe(req, n, &obd->obd_lock_replay_queue, rq_list){
1100                 DEBUG_REQ(D_ERROR, req, "aborted:");
1101                 req->rq_status = -ENOTCONN;
1102                 if (ptlrpc_error(req)) { 
1103                         DEBUG_REQ(D_ERROR, req,
1104                                   "failed abort_lock_reply; skipping");
1105                 }
1106                 ptlrpc_free_clone(req);
1107         }
1108 }
1109 #endif
1110
1111 /* Called from a cleanup function if the device is being cleaned up
1112    forcefully.  The exports should all have been disconnected already,
1113    the only thing left to do is
1114      - clear the recovery flags
1115      - cancel the timer
1116      - free queued requests and replies, but don't send replies
1117    Because the obd_stopping flag is set, no new requests should be received.
1118
1119 */
1120 void target_cleanup_recovery(struct obd_device *obd)
1121 {
1122         struct ptlrpc_request *req, *n;
1123         ENTRY;
1124
1125         LASSERT(obd->obd_stopping);
1126
1127         spin_lock_bh(&obd->obd_processing_task_lock);
1128         if (!obd->obd_recovering) {
1129                 spin_unlock_bh(&obd->obd_processing_task_lock);
1130                 EXIT;
1131                 return;
1132         }
1133         obd->obd_recovering = obd->obd_abort_recovery = 0;
1134         target_cancel_recovery_timer(obd);
1135         spin_unlock_bh(&obd->obd_processing_task_lock);
1136
1137         list_for_each_entry_safe(req, n, &obd->obd_req_replay_queue, rq_list) {
1138                 LASSERT (req->rq_reply_state == 0);
1139                 ptlrpc_free_clone(req);
1140         }
1141         list_for_each_entry_safe(req, n, &obd->obd_lock_replay_queue, rq_list){
1142                 LASSERT (req->rq_reply_state == 0);
1143                 ptlrpc_free_clone(req);
1144         }
1145         list_for_each_entry_safe(req, n, &obd->obd_final_req_queue, rq_list) {
1146                 LASSERT (req->rq_reply_state == 0);
1147                 ptlrpc_free_clone(req);
1148         }
1149
1150         EXIT;
1151 }
1152
1153 static void target_recovery_expired(unsigned long castmeharder)
1154 {
1155         struct obd_device *obd = (struct obd_device *)castmeharder;
1156         CERROR("%s: recovery timed out, aborting\n", obd->obd_name);
1157         spin_lock_bh(&obd->obd_processing_task_lock);
1158         if (obd->obd_recovering)
1159                 obd->obd_abort_recovery = 1;
1160         cfs_waitq_signal(&obd->obd_next_transno_waitq);
1161         spin_unlock_bh(&obd->obd_processing_task_lock);
1162 }
1163
1164
1165 /* obd_processing_task_lock should be held */
1166 void target_cancel_recovery_timer(struct obd_device *obd)
1167 {
1168         CDEBUG(D_HA, "%s: cancel recovery timer\n", obd->obd_name);
1169         cfs_timer_disarm(&obd->obd_recovery_timer);
1170 }
1171
1172 static void reset_recovery_timer(struct obd_device *obd)
1173 {
1174         spin_lock_bh(&obd->obd_processing_task_lock);
1175         if (!obd->obd_recovering) {
1176                 spin_unlock_bh(&obd->obd_processing_task_lock);
1177                 return;
1178         }
1179         cfs_timer_arm(&obd->obd_recovery_timer,
1180                       cfs_time_shift(OBD_RECOVERY_TIMEOUT));
1181         spin_unlock_bh(&obd->obd_processing_task_lock);
1182         CDEBUG(D_HA, "%s: timer will expire in %u seconds\n", obd->obd_name,
1183                OBD_RECOVERY_TIMEOUT);
1184         /* Only used for lprocfs_status */
1185         obd->obd_recovery_end = CURRENT_SECONDS + OBD_RECOVERY_TIMEOUT;
1186 }
1187
1188
1189 /* Only start it the first time called */
1190 void target_start_recovery_timer(struct obd_device *obd)
1191 {
1192         spin_lock_bh(&obd->obd_processing_task_lock);
1193         if (obd->obd_recovery_handler
1194             || timer_pending((struct timer_list *)&obd->obd_recovery_timer)) {
1195                 spin_unlock_bh(&obd->obd_processing_task_lock);
1196                 return;
1197         }
1198         CWARN("%s: starting recovery timer (%us)\n", obd->obd_name,
1199               OBD_RECOVERY_TIMEOUT);
1200         cfs_timer_init(&obd->obd_recovery_timer, target_recovery_expired, obd);
1201         spin_unlock_bh(&obd->obd_processing_task_lock);
1202
1203         reset_recovery_timer(obd);
1204 }
1205
1206 #ifdef __KERNEL__
1207 static int check_for_next_transno(struct obd_device *obd)
1208 {
1209         struct ptlrpc_request *req = NULL;
1210         int wake_up = 0, connected, completed, queue_len, max;
1211         __u64 next_transno, req_transno;
1212         ENTRY;
1213         spin_lock_bh(&obd->obd_processing_task_lock);
1214
1215         if (!list_empty(&obd->obd_req_replay_queue)) {
1216                 req = list_entry(obd->obd_req_replay_queue.next,
1217                                  struct ptlrpc_request, rq_list);
1218                 req_transno = lustre_msg_get_transno(req->rq_reqmsg);
1219         } else {
1220                 req_transno = 0;
1221         }
1222
1223         max = obd->obd_max_recoverable_clients;
1224         connected = obd->obd_connected_clients;
1225         completed = connected - obd->obd_recoverable_clients;
1226         queue_len = obd->obd_requests_queued_for_recovery;
1227         next_transno = obd->obd_next_recovery_transno;
1228
1229         CDEBUG(D_HA, "max: %d, connected: %d, completed: %d, queue_len: %d, "
1230                "req_transno: "LPU64", next_transno: "LPU64"\n",
1231                max, connected, completed, queue_len, req_transno, next_transno);
1232
1233         if (obd->obd_abort_recovery) {
1234                 CDEBUG(D_HA, "waking for aborted recovery\n");
1235                 wake_up = 1;
1236         } else if (atomic_read(&obd->obd_req_replay_clients) == 0) {
1237                 CDEBUG(D_HA, "waking for completed recovery\n");
1238                 wake_up = 1;
1239         } else if (req_transno == next_transno) {
1240                 CDEBUG(D_HA, "waking for next ("LPD64")\n", next_transno);
1241                 wake_up = 1;
1242         } else if (queue_len + completed == max) {
1243                 /* handle gaps occured due to lost reply. It is allowed gaps
1244                  * because all clients are connected and there will be resend
1245                  * for missed transaction */
1246                 LASSERTF(req_transno >= next_transno,
1247                          "req_transno: "LPU64", next_transno: "LPU64"\n",
1248                          req_transno, next_transno);
1249
1250                 CDEBUG(req_transno > obd->obd_last_committed ? D_ERROR : D_HA,
1251                        "waking for skipped transno (skip: "LPD64
1252                        ", ql: %d, comp: %d, conn: %d, next: "LPD64")\n",
1253                        next_transno, queue_len, completed, connected, req_transno);
1254                 obd->obd_next_recovery_transno = req_transno;
1255                 wake_up = 1;
1256         } else if (queue_len == atomic_read(&obd->obd_req_replay_clients)) {
1257                 /* some clients haven't connected in time, but we can try
1258                  * to replay requests that demand on already committed ones
1259                  * also, we can replay first non-committed transation */
1260                 LASSERT(req_transno != 0);
1261                 if (req_transno == obd->obd_last_committed + 1) {
1262                         obd->obd_next_recovery_transno = req_transno;
1263                 } else if (req_transno > obd->obd_last_committed) {
1264                         /* can't continue recovery: have no needed transno */
1265                         obd->obd_abort_recovery = 1;
1266                         CDEBUG(D_ERROR, "abort due to missed clients. max: %d, "
1267                                "connected: %d, completed: %d, queue_len: %d, "
1268                                "req_transno: "LPU64", next_transno: "LPU64"\n",
1269                                max, connected, completed, queue_len,
1270                                req_transno, next_transno);
1271                 }
1272                 wake_up = 1;
1273         }
1274
1275         spin_unlock_bh(&obd->obd_processing_task_lock);
1276         return wake_up;
1277 }
1278
1279 static struct ptlrpc_request *target_next_replay_req(struct obd_device *obd)
1280 {
1281         struct l_wait_info lwi = { 0 };
1282         struct ptlrpc_request *req;
1283
1284         CDEBUG(D_HA, "Waiting for transno "LPD64"\n",
1285                obd->obd_next_recovery_transno);
1286         l_wait_event(obd->obd_next_transno_waitq,
1287                      check_for_next_transno(obd), &lwi);
1288
1289         spin_lock_bh(&obd->obd_processing_task_lock);
1290         if (obd->obd_abort_recovery) {
1291                 req = NULL;
1292         } else if (!list_empty(&obd->obd_req_replay_queue)) {
1293                 req = list_entry(obd->obd_req_replay_queue.next,
1294                                  struct ptlrpc_request, rq_list);
1295                 list_del_init(&req->rq_list);
1296                 obd->obd_requests_queued_for_recovery--;
1297         } else {
1298                 req = NULL;
1299         }
1300         spin_unlock_bh(&obd->obd_processing_task_lock);
1301         RETURN(req);
1302 }
1303
1304 static int check_for_next_lock(struct obd_device *obd)
1305 {
1306         struct ptlrpc_request *req = NULL;
1307         int wake_up = 0;
1308
1309         spin_lock_bh(&obd->obd_processing_task_lock);
1310         if (!list_empty(&obd->obd_lock_replay_queue)) {
1311                 req = list_entry(obd->obd_lock_replay_queue.next,
1312                                  struct ptlrpc_request, rq_list);
1313                 CDEBUG(D_HA, "waking for next lock\n");
1314                 wake_up = 1;
1315         } else if (atomic_read(&obd->obd_lock_replay_clients) == 0) {
1316                 CDEBUG(D_HA, "waking for completed lock replay\n");
1317                 wake_up = 1;
1318         } else if (obd->obd_abort_recovery) {
1319                 CDEBUG(D_HA, "waking for aborted recovery\n");
1320                 wake_up = 1;
1321         }
1322         spin_unlock_bh(&obd->obd_processing_task_lock);
1323
1324         return wake_up;
1325 }
1326
1327 static struct ptlrpc_request *target_next_replay_lock(struct obd_device *obd)
1328 {
1329         struct l_wait_info lwi = { 0 };
1330         struct ptlrpc_request *req;
1331
1332         CDEBUG(D_HA, "Waiting for lock\n");
1333         l_wait_event(obd->obd_next_transno_waitq,
1334                      check_for_next_lock(obd), &lwi);
1335
1336         spin_lock_bh(&obd->obd_processing_task_lock);
1337         if (obd->obd_abort_recovery) {
1338                 req = NULL;
1339         } else if (!list_empty(&obd->obd_lock_replay_queue)) {
1340                 req = list_entry(obd->obd_lock_replay_queue.next,
1341                                  struct ptlrpc_request, rq_list);
1342                 list_del_init(&req->rq_list);
1343         } else {
1344                 req = NULL;
1345         }
1346         spin_unlock_bh(&obd->obd_processing_task_lock);
1347         return req;
1348 }
1349
1350 static struct ptlrpc_request *target_next_final_ping(struct obd_device *obd)
1351 {
1352         struct ptlrpc_request *req;
1353
1354         spin_lock_bh(&obd->obd_processing_task_lock);
1355         if (!list_empty(&obd->obd_final_req_queue)) {
1356                 req = list_entry(obd->obd_final_req_queue.next,
1357                                  struct ptlrpc_request, rq_list);
1358                 list_del_init(&req->rq_list);
1359         } else {
1360                 req = NULL;
1361         }
1362         spin_unlock_bh(&obd->obd_processing_task_lock);
1363         return req;
1364 }
1365
1366 static inline int req_replay_done(struct obd_export *exp)
1367 {
1368         return (exp->exp_req_replay_needed == 0);
1369 }
1370
1371 static inline int lock_replay_done(struct obd_export *exp)
1372 {
1373         return (exp->exp_lock_replay_needed == 0);
1374 }
1375
1376 static inline int connect_done(struct obd_export *exp)
1377 {
1378         return (exp->exp_in_recovery != 0);
1379 }
1380
1381 static int check_for_clients(struct obd_device *obd)
1382 {
1383         if (obd->obd_abort_recovery)
1384                 return 1;
1385         LASSERT(obd->obd_connected_clients <= obd->obd_max_recoverable_clients);
1386         if (obd->obd_no_conn == 0 &&
1387             obd->obd_connected_clients == obd->obd_max_recoverable_clients)
1388                 return 1;
1389         return 0;
1390 }
1391
1392 static int handle_recovery_req(struct ptlrpc_thread *thread,
1393                                struct ptlrpc_request *req,
1394                                svc_handler_t handler)
1395 {
1396         int rc;
1397         ENTRY;
1398
1399         rc = lu_context_init(&req->rq_session, LCT_SESSION);
1400         if (rc) {
1401                 CERROR("Failure to initialize session: %d\n", rc);
1402                 return rc;
1403         }
1404         req->rq_session.lc_thread = thread;
1405         lu_context_enter(&req->rq_session);
1406         req->rq_svc_thread = thread;
1407         req->rq_svc_thread->t_env->le_ses = &req->rq_session;
1408
1409         (void)handler(req);
1410
1411         lu_context_exit(&req->rq_session);
1412         lu_context_fini(&req->rq_session);
1413         /* don't reset timer for final stage */
1414         if (!req_replay_done(req->rq_export) ||
1415             !lock_replay_done(req->rq_export))
1416                 reset_recovery_timer(class_exp2obd(req->rq_export));
1417         ptlrpc_free_clone(req);
1418         RETURN(0);
1419 }
1420
1421 static int target_recovery_thread(void *arg)
1422 {
1423         struct obd_device *obd = arg;
1424         struct ptlrpc_request *req;
1425         struct target_recovery_data *trd = &obd->obd_recovery_data;
1426         struct l_wait_info lwi = { 0 };
1427         unsigned long delta;
1428         unsigned long flags;
1429         struct lu_env env;
1430         struct ptlrpc_thread fake_svc_thread, *thread = &fake_svc_thread;
1431         __u32 recov_ctx_tags = LCT_MD_THREAD;
1432         int rc = 0;
1433         ENTRY;
1434
1435         cfs_daemonize("tgt_recov");
1436
1437         SIGNAL_MASK_LOCK(current, flags);
1438         sigfillset(&current->blocked);
1439         RECALC_SIGPENDING;
1440         SIGNAL_MASK_UNLOCK(current, flags);
1441
1442         rc = lu_context_init(&env.le_ctx, recov_ctx_tags);
1443         if (rc)
1444                 RETURN(rc);
1445
1446         thread->t_env = &env;
1447         env.le_ctx.lc_thread = thread;
1448
1449         CERROR("%s: started recovery thread pid %d\n", obd->obd_name,
1450                current->pid);
1451         trd->trd_processing_task = current->pid;
1452
1453         obd->obd_recovering = 1;
1454         complete(&trd->trd_starting);
1455
1456         /* first of all, we have to know the first transno to replay */
1457         obd->obd_abort_recovery = 0;
1458         l_wait_event(obd->obd_next_transno_waitq,
1459                      check_for_clients(obd), &lwi);
1460
1461         spin_lock_bh(&obd->obd_processing_task_lock);
1462         target_cancel_recovery_timer(obd);
1463         spin_unlock_bh(&obd->obd_processing_task_lock);
1464
1465         /* If some clients haven't connected in time, evict them */
1466         if (obd->obd_abort_recovery) {
1467                 CWARN("Some clients haven't connect in time (%d/%d),"
1468                        "evict them\n", obd->obd_connected_clients,
1469                        obd->obd_max_recoverable_clients);
1470                 obd->obd_abort_recovery = obd->obd_stopping;
1471                 class_disconnect_stale_exports(obd, connect_done);
1472         }
1473         /* next stage: replay requests */
1474         delta = jiffies;
1475         obd->obd_req_replaying = 1;
1476         CDEBUG(D_INFO, "1: request replay stage - %d clients from t"LPU64"\n",
1477               atomic_read(&obd->obd_req_replay_clients),
1478               obd->obd_next_recovery_transno);
1479         while ((req = target_next_replay_req(obd))) {
1480                 LASSERT(trd->trd_processing_task == current->pid);
1481                 DEBUG_REQ(D_HA, req, "processing t"LPD64" from %s",
1482                           lustre_msg_get_transno(req->rq_reqmsg),
1483                           libcfs_nid2str(req->rq_peer.nid));
1484
1485                 handle_recovery_req(thread, req,
1486                                     trd->trd_recovery_handler);
1487                 obd->obd_replayed_requests++;
1488                 spin_lock_bh(&obd->obd_processing_task_lock);
1489                 obd->obd_next_recovery_transno++;
1490                 spin_unlock_bh(&obd->obd_processing_task_lock);
1491         }
1492
1493         spin_lock_bh(&obd->obd_processing_task_lock);
1494         target_cancel_recovery_timer(obd);
1495         spin_unlock_bh(&obd->obd_processing_task_lock);
1496
1497         /* If some clients haven't replayed requests in time, evict them */
1498         if (obd->obd_abort_recovery) {
1499                 CDEBUG(D_ERROR, "req replay timed out, aborting ...\n");
1500                 obd->obd_abort_recovery = obd->obd_stopping;
1501                 class_disconnect_stale_exports(obd, req_replay_done);
1502                 abort_req_replay_queue(obd);
1503         }
1504         /* The second stage: replay locks */
1505         CDEBUG(D_INFO, "2: lock replay stage - %d clients\n",
1506                atomic_read(&obd->obd_lock_replay_clients));
1507         while ((req = target_next_replay_lock(obd))) {
1508                 LASSERT(trd->trd_processing_task == current->pid);
1509                 DEBUG_REQ(D_HA|D_WARNING, req, "processing lock from %s: ",
1510                           libcfs_nid2str(req->rq_peer.nid));
1511                 handle_recovery_req(thread, req,
1512                                     trd->trd_recovery_handler);
1513                 obd->obd_replayed_locks++;
1514         }
1515
1516         spin_lock_bh(&obd->obd_processing_task_lock);
1517         target_cancel_recovery_timer(obd);
1518         spin_unlock_bh(&obd->obd_processing_task_lock);
1519         /* If some clients haven't replayed requests in time, evict them */
1520         if (obd->obd_abort_recovery) {
1521                 int stale;
1522                 CERROR("lock replay timed out, aborting ...\n");
1523                 obd->obd_abort_recovery = obd->obd_stopping;
1524                 stale = class_disconnect_stale_exports(obd, lock_replay_done);
1525                 abort_lock_replay_queue(obd);
1526         }
1527
1528         /* We drop recoverying flag to forward all new requests
1529          * to regular mds_handle() since now */
1530         spin_lock_bh(&obd->obd_processing_task_lock);
1531         obd->obd_recovering = obd->obd_abort_recovery = 0;
1532         spin_unlock_bh(&obd->obd_processing_task_lock);
1533         /* The third stage: reply on final pings */
1534         CDEBUG(D_INFO, "3: final stage - process recovery completion pings\n");
1535         while ((req = target_next_final_ping(obd))) {
1536                 LASSERT(trd->trd_processing_task == current->pid);
1537                 DEBUG_REQ(D_HA, req, "processing final ping from %s: ",
1538                           libcfs_nid2str(req->rq_peer.nid));
1539                 handle_recovery_req(thread, req,
1540                                     trd->trd_recovery_handler);
1541         }
1542
1543         delta = (jiffies - delta) / HZ;
1544         CDEBUG(D_INFO,"4: recovery completed in %lus - %d/%d reqs/locks\n",
1545               delta, obd->obd_replayed_requests, obd->obd_replayed_locks);
1546         LASSERT(atomic_read(&obd->obd_req_replay_clients) == 0);
1547         LASSERT(atomic_read(&obd->obd_lock_replay_clients) == 0);
1548         if (delta > obd_timeout * 2) {
1549                 CWARN("too long recovery - read logs\n");
1550                 libcfs_debug_dumplog();
1551         }
1552
1553         target_finish_recovery(obd);
1554
1555         lu_env_fini(&env);
1556         trd->trd_processing_task = 0;
1557         complete(&trd->trd_finishing);
1558         RETURN(rc);
1559 }
1560
1561 int target_start_recovery_thread(struct obd_device *obd, svc_handler_t handler)
1562 {
1563         int rc = 0;
1564         struct target_recovery_data *trd = &obd->obd_recovery_data;
1565
1566         memset(trd, 0, sizeof(*trd));
1567         init_completion(&trd->trd_starting);
1568         init_completion(&trd->trd_finishing);
1569         trd->trd_recovery_handler = handler;
1570
1571         if (kernel_thread(target_recovery_thread, obd, 0) > 0) {
1572                 wait_for_completion(&trd->trd_starting);
1573                 LASSERT(obd->obd_recovering != 0);
1574         } else
1575                 rc = -ECHILD;
1576
1577         return rc;
1578 }
1579
1580 void target_stop_recovery_thread(struct obd_device *obd)
1581 {
1582         spin_lock_bh(&obd->obd_processing_task_lock);
1583         if (obd->obd_recovery_data.trd_processing_task > 0) {
1584                 struct target_recovery_data *trd = &obd->obd_recovery_data;
1585                 CERROR("%s: Aborting recovery\n", obd->obd_name);
1586                 obd->obd_abort_recovery = 1;
1587                 wake_up(&obd->obd_next_transno_waitq);
1588                 spin_unlock_bh(&obd->obd_processing_task_lock);
1589                 wait_for_completion(&trd->trd_finishing);
1590         } else {
1591                 spin_unlock_bh(&obd->obd_processing_task_lock);
1592         }
1593 }
1594
1595 void target_recovery_fini(struct obd_device *obd)
1596 {
1597         class_disconnect_exports(obd);
1598         target_stop_recovery_thread(obd);
1599         target_cleanup_recovery(obd);
1600 }
1601 EXPORT_SYMBOL(target_recovery_fini);
1602
1603 void target_recovery_init(struct obd_device *obd, svc_handler_t handler)
1604 {
1605         if (obd->obd_max_recoverable_clients == 0)
1606                 return;
1607         
1608         CWARN("RECOVERY: service %s, %d recoverable clients, "
1609               "last_transno "LPU64"\n", obd->obd_name,
1610               obd->obd_max_recoverable_clients, obd->obd_last_committed);
1611         obd->obd_next_recovery_transno = obd->obd_last_committed + 1;
1612         target_start_recovery_thread(obd, handler);
1613         obd->obd_recovery_start = CURRENT_SECONDS;
1614         /* Only used for lprocfs_status */
1615         obd->obd_recovery_end = obd->obd_recovery_start + OBD_RECOVERY_TIMEOUT;
1616 }
1617 EXPORT_SYMBOL(target_recovery_init);
1618
1619 #endif
1620
1621 int target_process_req_flags(struct obd_device *obd, struct ptlrpc_request *req)
1622 {
1623         struct obd_export *exp = req->rq_export;
1624         LASSERT(exp != NULL);
1625         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REQ_REPLAY_DONE) {
1626                 /* client declares he's ready to replay locks */
1627                 spin_lock_bh(&obd->obd_processing_task_lock);
1628                 if (exp->exp_req_replay_needed) {
1629                         LASSERT(atomic_read(&obd->obd_req_replay_clients) > 0);
1630                         spin_lock(&exp->exp_lock);
1631                         exp->exp_req_replay_needed = 0;
1632                         spin_unlock(&exp->exp_lock);
1633                         atomic_dec(&obd->obd_req_replay_clients);
1634                         LASSERT(obd->obd_recoverable_clients > 0);
1635                         obd->obd_recoverable_clients--;
1636                         if (atomic_read(&obd->obd_req_replay_clients) == 0)
1637                                 CDEBUG(D_HA, "all clients have replayed reqs\n");
1638                         wake_up(&obd->obd_next_transno_waitq);
1639                 }
1640                 spin_unlock_bh(&obd->obd_processing_task_lock);
1641         }
1642         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_LOCK_REPLAY_DONE) {
1643                 /* client declares he's ready to complete recovery
1644                  * so, we put the request on th final queue */
1645                 spin_lock_bh(&obd->obd_processing_task_lock);
1646                 if (exp->exp_lock_replay_needed) {
1647                         LASSERT(atomic_read(&obd->obd_lock_replay_clients) > 0);
1648                         spin_lock(&exp->exp_lock);
1649                         exp->exp_lock_replay_needed = 0;
1650                         spin_unlock(&exp->exp_lock);
1651                         atomic_dec(&obd->obd_lock_replay_clients);
1652                         if (atomic_read(&obd->obd_lock_replay_clients) == 0)
1653                                 CDEBUG(D_HA, "all clients have replayed locks\n");
1654                         wake_up(&obd->obd_next_transno_waitq);
1655                 }
1656                 spin_unlock_bh(&obd->obd_processing_task_lock);
1657         }
1658
1659         return 0;
1660 }
1661
1662 int target_queue_recovery_request(struct ptlrpc_request *req,
1663                                   struct obd_device *obd)
1664 {
1665         struct list_head *tmp;
1666         int inserted = 0;
1667         __u64 transno = lustre_msg_get_transno(req->rq_reqmsg);
1668
1669         ENTRY;
1670
1671         if (obd->obd_recovery_data.trd_processing_task == cfs_curproc_pid()) {
1672                 /* Processing the queue right now, don't re-add. */
1673                 RETURN(1);
1674         }
1675
1676         target_process_req_flags(obd, req);
1677
1678         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_LOCK_REPLAY_DONE) {
1679                 /* client declares he's ready to complete recovery
1680                  * so, we put the request on th final queue */
1681                 req = ptlrpc_clone_req(req);
1682                 if (req == NULL)
1683                         RETURN(-ENOMEM);
1684                 DEBUG_REQ(D_HA, req, "queue final req");
1685                 spin_lock_bh(&obd->obd_processing_task_lock);
1686                 if (obd->obd_recovering)
1687                         list_add_tail(&req->rq_list, &obd->obd_final_req_queue);
1688                 else {
1689                         spin_unlock_bh(&obd->obd_processing_task_lock);
1690                         ptlrpc_free_clone(req);
1691                         if (obd->obd_stopping) {
1692                                 RETURN(-ENOTCONN);
1693                         } else {
1694                                 RETURN(1);
1695                         }
1696                 }
1697                 spin_unlock_bh(&obd->obd_processing_task_lock);
1698                 RETURN(0);
1699         }
1700         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REQ_REPLAY_DONE) {
1701                 /* client declares he's ready to replay locks */
1702                 req = ptlrpc_clone_req(req);
1703                 if (req == NULL)
1704                         RETURN(-ENOMEM);
1705                 DEBUG_REQ(D_HA, req, "queue lock replay req");
1706                 spin_lock_bh(&obd->obd_processing_task_lock);
1707                 LASSERT(obd->obd_recovering);
1708                 /* usually due to recovery abort */
1709                 if (!req->rq_export->exp_in_recovery) {
1710                         spin_unlock_bh(&obd->obd_processing_task_lock);
1711                         ptlrpc_free_clone(req);
1712                         RETURN(-ENOTCONN);
1713                 }
1714                 LASSERT(req->rq_export->exp_lock_replay_needed);
1715                 list_add_tail(&req->rq_list, &obd->obd_lock_replay_queue);
1716                 spin_unlock_bh(&obd->obd_processing_task_lock);
1717                 wake_up(&obd->obd_next_transno_waitq);
1718                 RETURN(0);
1719         }
1720
1721         /* CAVEAT EMPTOR: The incoming request message has been swabbed
1722          * (i.e. buflens etc are in my own byte order), but type-dependent
1723          * buffers (eg mds_body, ost_body etc) have NOT been swabbed. */
1724
1725         if (!transno) {
1726                 CFS_INIT_LIST_HEAD(&req->rq_list);
1727                 DEBUG_REQ(D_HA, req, "not queueing");
1728                 RETURN(1);
1729         }
1730
1731         spin_lock_bh(&obd->obd_processing_task_lock);
1732
1733         /* If we're processing the queue, we want don't want to queue this
1734          * message.
1735          *
1736          * Also, if this request has a transno less than the one we're waiting
1737          * for, we should process it now.  It could (and currently always will)
1738          * be an open request for a descriptor that was opened some time ago.
1739          *
1740          * Also, a resent, replayed request that has already been
1741          * handled will pass through here and be processed immediately.
1742          */
1743         CWARN("Next recovery transno: "LPU64", current: "LPU64", replaying: %i\n",
1744               obd->obd_next_recovery_transno, transno, obd->obd_req_replaying);
1745         if (transno < obd->obd_next_recovery_transno && obd->obd_req_replaying) {
1746                 /* Processing the queue right now, don't re-add. */
1747                 LASSERT(list_empty(&req->rq_list));
1748                 spin_unlock_bh(&obd->obd_processing_task_lock);
1749                 RETURN(1);
1750         }
1751         spin_unlock_bh(&obd->obd_processing_task_lock);
1752
1753         /* A resent, replayed request that is still on the queue; just drop it.
1754            The queued request will handle this. */
1755         if ((lustre_msg_get_flags(req->rq_reqmsg) & (MSG_RESENT|MSG_REPLAY)) ==
1756             (MSG_RESENT | MSG_REPLAY)) {
1757                 DEBUG_REQ(D_ERROR, req, "dropping resent queued req");
1758                 RETURN(0);
1759         }
1760
1761         req = ptlrpc_clone_req(req);
1762         if (req == NULL)
1763                 RETURN(-ENOMEM);
1764
1765         spin_lock_bh(&obd->obd_processing_task_lock);
1766         LASSERT(obd->obd_recovering);
1767         if (!req->rq_export->exp_in_recovery) {
1768                 spin_unlock_bh(&obd->obd_processing_task_lock);
1769                 ptlrpc_free_clone(req);
1770                 RETURN(-ENOTCONN);
1771         }
1772         LASSERT(req->rq_export->exp_req_replay_needed);
1773
1774         /* XXX O(n^2) */
1775         list_for_each(tmp, &obd->obd_req_replay_queue) {
1776                 struct ptlrpc_request *reqiter =
1777                         list_entry(tmp, struct ptlrpc_request, rq_list);
1778
1779                 if (lustre_msg_get_transno(reqiter->rq_reqmsg) > transno) {
1780                         list_add_tail(&req->rq_list, &reqiter->rq_list);
1781                         inserted = 1;
1782                         break;
1783                 }
1784         }
1785
1786         if (!inserted)
1787                 list_add_tail(&req->rq_list, &obd->obd_req_replay_queue);
1788
1789         obd->obd_requests_queued_for_recovery++;
1790         wake_up(&obd->obd_next_transno_waitq);
1791         spin_unlock_bh(&obd->obd_processing_task_lock);
1792         RETURN(0);
1793
1794 }
1795
1796 struct obd_device * target_req2obd(struct ptlrpc_request *req)
1797 {
1798         return req->rq_export->exp_obd;
1799 }
1800
1801 static inline struct ldlm_pool *ldlm_exp2pl(struct obd_export *exp)
1802 {
1803         LASSERT(exp != NULL);
1804         return &exp->exp_obd->obd_namespace->ns_pool;
1805 }
1806
1807 int target_pack_pool_reply(struct ptlrpc_request *req)
1808 {
1809         struct ldlm_pool *pl;
1810         ENTRY;
1811    
1812         if (!req->rq_export || !exp_connect_lru_resize(req->rq_export)) {
1813                 lustre_msg_set_slv(req->rq_repmsg, 0);
1814                 lustre_msg_set_limit(req->rq_repmsg, 0);
1815                 RETURN(0);
1816         }
1817  
1818         pl = ldlm_exp2pl(req->rq_export);
1819
1820         spin_lock(&pl->pl_lock);
1821         LASSERT(ldlm_pool_get_slv(pl) != 0 && ldlm_pool_get_limit(pl) != 0);
1822         lustre_msg_set_slv(req->rq_repmsg, ldlm_pool_get_slv(pl));
1823         lustre_msg_set_limit(req->rq_repmsg, ldlm_pool_get_limit(pl));
1824         spin_unlock(&pl->pl_lock);
1825
1826         RETURN(0);
1827 }
1828
1829 int target_send_reply_msg(struct ptlrpc_request *req, int rc, int fail_id)
1830 {
1831         if (OBD_FAIL_CHECK(fail_id | OBD_FAIL_ONCE)) {
1832                 obd_fail_loc |= OBD_FAIL_ONCE | OBD_FAILED;
1833                 DEBUG_REQ(D_ERROR, req, "dropping reply");
1834                 return (-ECOMM);
1835         }
1836
1837         if (unlikely(rc)) {
1838                 DEBUG_REQ(D_ERROR, req, "processing error (%d)", rc);
1839                 req->rq_status = rc;
1840                 return (ptlrpc_error(req));
1841         } else {
1842                 DEBUG_REQ(D_NET, req, "sending reply");
1843         }
1844
1845         return (ptlrpc_send_reply(req, 1));
1846 }
1847
1848 void target_send_reply(struct ptlrpc_request *req, int rc, int fail_id)
1849 {
1850         int                        netrc;
1851         struct ptlrpc_reply_state *rs;
1852         struct obd_device         *obd;
1853         struct obd_export         *exp;
1854         struct ptlrpc_service     *svc;
1855
1856         if (req->rq_no_reply)
1857                 return;
1858
1859         svc = req->rq_rqbd->rqbd_service;
1860         rs = req->rq_reply_state;
1861         if (rs == NULL || !rs->rs_difficult) {
1862                 /* no notifiers */
1863                 target_send_reply_msg (req, rc, fail_id);
1864                 return;
1865         }
1866
1867         /* must be an export if locks saved */
1868         LASSERT (req->rq_export != NULL);
1869         /* req/reply consistent */
1870         LASSERT (rs->rs_service == svc);
1871
1872         /* "fresh" reply */
1873         LASSERT (!rs->rs_scheduled);
1874         LASSERT (!rs->rs_scheduled_ever);
1875         LASSERT (!rs->rs_handled);
1876         LASSERT (!rs->rs_on_net);
1877         LASSERT (rs->rs_export == NULL);
1878         LASSERT (list_empty(&rs->rs_obd_list));
1879         LASSERT (list_empty(&rs->rs_exp_list));
1880
1881         exp = class_export_get (req->rq_export);
1882         obd = exp->exp_obd;
1883
1884         /* disable reply scheduling onto srv_reply_queue while I'm setting up */
1885         rs->rs_scheduled = 1;
1886         rs->rs_on_net    = 1;
1887         rs->rs_xid       = req->rq_xid;
1888         rs->rs_transno   = req->rq_transno;
1889         rs->rs_export    = exp;
1890
1891         spin_lock(&obd->obd_uncommitted_replies_lock);
1892
1893         if (rs->rs_transno > obd->obd_last_committed) {
1894                 /* not committed already */
1895                 list_add_tail (&rs->rs_obd_list,
1896                                &obd->obd_uncommitted_replies);
1897         }
1898
1899         spin_unlock (&obd->obd_uncommitted_replies_lock);
1900         spin_lock (&exp->exp_lock);
1901
1902         list_add_tail (&rs->rs_exp_list, &exp->exp_outstanding_replies);
1903
1904         spin_unlock(&exp->exp_lock);
1905
1906         netrc = target_send_reply_msg (req, rc, fail_id);
1907
1908         spin_lock(&svc->srv_lock);
1909
1910         svc->srv_n_difficult_replies++;
1911
1912         if (netrc != 0) {
1913                 /* error sending: reply is off the net.  Also we need +1
1914                  * reply ref until ptlrpc_server_handle_reply() is done
1915                  * with the reply state (if the send was successful, there
1916                  * would have been +1 ref for the net, which
1917                  * reply_out_callback leaves alone) */
1918                 rs->rs_on_net = 0;
1919                 ptlrpc_rs_addref(rs);
1920                 atomic_inc (&svc->srv_outstanding_replies);
1921         }
1922
1923         if (!rs->rs_on_net ||                   /* some notifier */
1924             list_empty(&rs->rs_exp_list) ||     /* completed already */
1925             list_empty(&rs->rs_obd_list)) {
1926                 list_add_tail (&rs->rs_list, &svc->srv_reply_queue);
1927                 cfs_waitq_signal (&svc->srv_waitq);
1928         } else {
1929                 list_add (&rs->rs_list, &svc->srv_active_replies);
1930                 rs->rs_scheduled = 0;           /* allow notifier to schedule */
1931         }
1932
1933         spin_unlock(&svc->srv_lock);
1934 }
1935
1936 int target_handle_ping(struct ptlrpc_request *req)
1937 {
1938         obd_ping(req->rq_export);
1939         return lustre_pack_reply(req, 1, NULL, NULL);
1940 }
1941
1942 void target_committed_to_req(struct ptlrpc_request *req)
1943 {
1944         struct obd_device *obd;
1945
1946         if (req == NULL || req->rq_export == NULL)
1947                 return;
1948
1949         obd = req->rq_export->exp_obd;
1950         if (obd == NULL)
1951                 return;
1952
1953         if (!obd->obd_no_transno && req->rq_repmsg != NULL)
1954                 lustre_msg_set_last_committed(req->rq_repmsg,
1955                                               obd->obd_last_committed);
1956         else
1957                 DEBUG_REQ(D_IOCTL, req, "not sending last_committed update");
1958
1959         CDEBUG(D_INFO, "last_committed "LPU64", transno "LPU64", xid "LPU64"\n",
1960                obd->obd_last_committed, req->rq_transno, req->rq_xid);
1961 }
1962
1963 EXPORT_SYMBOL(target_committed_to_req);
1964
1965 #ifdef HAVE_QUOTA_SUPPORT
1966 int target_handle_qc_callback(struct ptlrpc_request *req)
1967 {
1968         struct obd_quotactl *oqctl;
1969         struct client_obd *cli = &req->rq_export->exp_obd->u.cli;
1970
1971         oqctl = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*oqctl),
1972                                    lustre_swab_obd_quotactl);
1973         if (oqctl == NULL) {
1974                 CERROR("Can't unpack obd_quotactl\n");
1975                 RETURN(-EPROTO);
1976         }
1977
1978         cli->cl_qchk_stat = oqctl->qc_stat;
1979
1980         return 0;
1981 }
1982
1983 int target_handle_dqacq_callback(struct ptlrpc_request *req)
1984 {
1985 #ifdef __KERNEL__
1986         struct obd_device *obd = req->rq_export->exp_obd;
1987         struct obd_device *master_obd;
1988         struct lustre_quota_ctxt *qctxt;
1989         struct qunit_data *qdata;
1990         void* rep;
1991         struct qunit_data_old *qdata_old;
1992         int rc = 0;
1993         int repsize[2] = { sizeof(struct ptlrpc_body),
1994                            sizeof(struct qunit_data) };
1995         ENTRY;
1996
1997         rc = lustre_pack_reply(req, 2, repsize, NULL);
1998         if (rc)
1999                 RETURN(rc);
2000
2001         LASSERT(req->rq_export);
2002
2003         /* fixed for bug10707 */
2004         if ((req->rq_export->exp_connect_flags & OBD_CONNECT_QUOTA64) &&
2005             !OBD_FAIL_CHECK(OBD_FAIL_QUOTA_QD_COUNT_32BIT)) {
2006                 CDEBUG(D_QUOTA, "qd_count is 64bit!\n");
2007                 rep = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
2008                                      sizeof(struct qunit_data));
2009                 LASSERT(rep);
2010                 qdata = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*qdata),
2011                                            lustre_swab_qdata);
2012         } else {
2013                 CDEBUG(D_QUOTA, "qd_count is 32bit!\n");
2014                 rep = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
2015                                      sizeof(struct qunit_data_old));
2016                 LASSERT(rep);
2017                 qdata_old = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*qdata_old),
2018                                                lustre_swab_qdata_old);
2019                 qdata = lustre_quota_old_to_new(qdata_old);
2020         }
2021
2022         if (qdata == NULL) {
2023                 CERROR("Can't unpack qunit_data\n");
2024                 RETURN(-EPROTO);
2025         }
2026
2027         /* we use the observer */
2028         LASSERT(obd->obd_observer && obd->obd_observer->obd_observer);
2029         master_obd = obd->obd_observer->obd_observer;
2030         qctxt = &master_obd->u.obt.obt_qctxt;
2031
2032         LASSERT(qctxt->lqc_handler);
2033         rc = qctxt->lqc_handler(master_obd, qdata,
2034                                 lustre_msg_get_opc(req->rq_reqmsg));
2035         if (rc && rc != -EDQUOT)
2036                 CDEBUG(rc == -EBUSY  ? D_QUOTA : D_ERROR,
2037                        "dqacq failed! (rc:%d)\n", rc);
2038
2039         /* the qd_count might be changed in lqc_handler */
2040         if ((req->rq_export->exp_connect_flags & OBD_CONNECT_QUOTA64) &&
2041             !OBD_FAIL_CHECK(OBD_FAIL_QUOTA_QD_COUNT_32BIT)) {
2042                 memcpy(rep,qdata,sizeof(*qdata));
2043         } else {
2044                 qdata_old = lustre_quota_new_to_old(qdata);
2045                 memcpy(rep,qdata_old,sizeof(*qdata_old));
2046         }
2047         req->rq_status = rc;
2048         rc = ptlrpc_reply(req);
2049
2050         RETURN(rc);
2051 #else
2052         return 0;
2053 #endif /* !__KERNEL__ */
2054 }
2055 #endif /* HAVE_QUOTA_SUPPORT */
2056
2057 ldlm_mode_t lck_compat_array[] = {
2058         [LCK_EX] LCK_COMPAT_EX,
2059         [LCK_PW] LCK_COMPAT_PW,
2060         [LCK_PR] LCK_COMPAT_PR,
2061         [LCK_CW] LCK_COMPAT_CW,
2062         [LCK_CR] LCK_COMPAT_CR,
2063         [LCK_NL] LCK_COMPAT_NL,
2064         [LCK_GROUP] LCK_COMPAT_GROUP
2065 };