Whamcloud - gitweb
5a6a5c1a87e7c7902d78647ab436d8251d5491d7
[fs/lustre-release.git] / lustre / ptlrpc / client.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  *  Copyright (C) 2002 Cluster File Systems, Inc.
5  *
6  *   This file is part of Lustre, http://www.lustre.org.
7  *
8  *   Lustre is free software; you can redistribute it and/or
9  *   modify it under the terms of version 2 of the GNU General Public
10  *   License as published by the Free Software Foundation.
11  *
12  *   Lustre is distributed in the hope that it will be useful,
13  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
14  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15  *   GNU General Public License for more details.
16  *
17  *   You should have received a copy of the GNU General Public License
18  *   along with Lustre; if not, write to the Free Software
19  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
20  *
21  */
22
23 #define EXPORT_SYMTAB
24
25 #include <linux/config.h>
26 #include <linux/module.h>
27 #include <linux/kernel.h>
28 #include <linux/list.h>
29
30 #define DEBUG_SUBSYSTEM S_RPC
31
32 #include <linux/obd_support.h>
33 #include <linux/obd_class.h>
34 #include <linux/lustre_net.h>
35
36 void llite_ha_conn_manage(struct lustre_ha_mgr *mgr, struct ptlrpc_client *cli)
37 {
38         ENTRY;
39         cli->cli_ha_mgr = mgr;
40         spin_lock(&mgr->mgr_lock);
41         list_add(&cli->cli_ha_item, &mgr->mgr_connections_lh); 
42         spin_unlock(&mgr->mgr_lock); 
43         EXIT;
44 }
45
46 void llite_ha_conn_fail(struct ptlrpc_client *cli)
47 {
48         ENTRY;
49         spin_lock(&cli->cli_ha_mgr->mgr_lock);
50         list_del(&cli->cli_ha_item);
51         list_add(&cli->cli_ha_item, &cli->cli_ha_mgr->mgr_troubled_lh); 
52         spin_unlock(&cli->cli_ha_mgr->mgr_lock); 
53         wake_up(&cli->cli_ha_mgr->mgr_waitq);
54         EXIT;
55 }
56
57 void ptlrpc_init_client(struct lustre_ha_mgr *mgr, int req_portal, int rep_portal,
58                           struct ptlrpc_client *cl)
59 {
60         memset(cl, 0, sizeof(*cl));
61         spin_lock_init(&cl->cli_lock);
62         cl->cli_ha_mgr = mgr;
63         if (mgr)
64                 llite_ha_conn_manage(mgr, cl);
65         cl->cli_xid = 1;
66         cl->cli_generation = 1;
67         cl->cli_epoch = 1;
68         cl->cli_bootcount = 0;
69         cl->cli_obd = NULL;
70         cl->cli_request_portal = req_portal;
71         cl->cli_reply_portal = rep_portal;
72         INIT_LIST_HEAD(&cl->cli_sending_head);
73         INIT_LIST_HEAD(&cl->cli_sent_head);
74         sema_init(&cl->cli_rpc_sem, 32);
75 }
76
77 int ptlrpc_connect_client(char *uuid, struct ptlrpc_client *cl,
78                           struct lustre_peer *peer)
79 {
80         int err;
81
82         cl->cli_epoch++;
83         err = kportal_uuid_to_peer(uuid, peer);
84         if (err != 0)
85                 CERROR("cannot find peer %s!\n", uuid);
86
87         return err;
88 }
89
90 struct ptlrpc_bulk_desc *ptlrpc_prep_bulk(struct lustre_peer *peer)
91 {
92         struct ptlrpc_bulk_desc *bulk;
93
94         OBD_ALLOC(bulk, sizeof(*bulk));
95         if (bulk != NULL) {
96                 memcpy(&bulk->b_peer, peer, sizeof(*peer));
97                 init_waitqueue_head(&bulk->b_waitq);
98         }
99
100         return bulk;
101 }
102
103 struct ptlrpc_request *ptlrpc_prep_req(struct ptlrpc_client *cl,
104                                        struct lustre_peer *peer, int opcode,
105                                        int count, int *lengths, char **bufs)
106 {
107         struct ptlrpc_request *request;
108         int rc;
109         ENTRY;
110
111         OBD_ALLOC(request, sizeof(*request));
112         if (!request) {
113                 CERROR("request allocation out of memory\n");
114                 RETURN(NULL);
115         }
116
117         spin_lock(&cl->cli_lock);
118         request->rq_xid = cl->cli_xid++;
119         spin_unlock(&cl->cli_lock);
120
121         rc = lustre_pack_msg(count, lengths, bufs,
122                              &request->rq_reqlen, &request->rq_reqbuf);
123         if (rc) {
124                 CERROR("cannot pack request %d\n", rc);
125                 RETURN(NULL);
126         }
127         request->rq_time = CURRENT_TIME;
128         request->rq_type = PTL_RPC_REQUEST;
129         memcpy(&request->rq_peer, peer, sizeof(*peer));
130         request->rq_reqmsg = (struct lustre_msg *)request->rq_reqbuf;
131         request->rq_reqmsg->opc = HTON__u32(opcode);
132         request->rq_reqmsg->xid = HTON__u32(request->rq_xid);
133         request->rq_reqmsg->type = HTON__u32(request->rq_type);
134         request->rq_client = cl;
135         request->rq_req_portal = cl->cli_request_portal;
136         request->rq_reply_portal = cl->cli_reply_portal;
137
138         RETURN(request);
139 }
140
141 void ptlrpc_free_req(struct ptlrpc_request *request)
142 {
143         if (request == NULL)
144                 return;
145
146         if (request->rq_repbuf != NULL)
147                 OBD_FREE(request->rq_repbuf, request->rq_replen);
148         OBD_FREE(request, sizeof(*request));
149 }
150
151 static int ptlrpc_check_reply(struct ptlrpc_request *req)
152 {
153         int rc = 0;
154
155         schedule_timeout(3 * HZ);  /* 3 second timeout */
156         if (req->rq_repbuf != NULL) {
157                 req->rq_flags = PTL_RPC_REPLY;
158                 GOTO(out, rc = 1);
159         }
160
161         if (CURRENT_TIME - req->rq_time >= 3) { 
162                 CERROR("-- REQ TIMEOUT --\n"); 
163                 if (req->rq_client && req->rq_client->cli_ha_mgr)
164                         llite_ha_conn_fail(req->rq_client); 
165                 return 0;
166         }
167
168         if (sigismember(&(current->pending.signal), SIGKILL) ||
169             sigismember(&(current->pending.signal), SIGTERM) ||
170             sigismember(&(current->pending.signal), SIGINT)) {
171                 req->rq_flags = PTL_RPC_INTR;
172                 GOTO(out, rc = 1);
173         }
174
175  out:
176         return rc;
177 }
178
179 int ptlrpc_check_status(struct ptlrpc_request *req, int err)
180 {
181         ENTRY;
182
183         if (err != 0) {
184                 CERROR("err is %d\n", err);
185                 RETURN(err);
186         }
187
188         if (req == NULL) {
189                 CERROR("req == NULL\n");
190                 RETURN(-ENOMEM);
191         }
192
193         if (req->rq_repmsg == NULL) {
194                 CERROR("req->rq_repmsg == NULL\n");
195                 RETURN(-ENOMEM);
196         }
197
198         if (req->rq_repmsg->status != 0) {
199                 CERROR("req->rq_repmsg->status is %d\n",
200                        req->rq_repmsg->status);
201                 /* XXX: translate this error from net to host */
202                 RETURN(req->rq_repmsg->status);
203         }
204
205         RETURN(0);
206 }
207
208 static void ptlrpc_cleanup_request_buf(struct ptlrpc_request *request)
209 {
210         OBD_FREE(request->rq_reqbuf, request->rq_reqlen);
211         request->rq_reqbuf = NULL;
212         request->rq_reqlen = 0;
213 }
214
215 /* Abort this request and cleanup any resources associated with it. */
216 static int ptlrpc_abort(struct ptlrpc_request *request)
217 {
218         /* First remove the ME for the reply; in theory, this means
219          * that we can tear down the buffer safely. */
220         PtlMEUnlink(request->rq_reply_me_h);
221         OBD_FREE(request->rq_reply_md.start, request->rq_replen);
222         request->rq_repbuf = NULL;
223         request->rq_replen = 0;
224         return 0;
225 }
226
227 int ptlrpc_queue_wait(struct ptlrpc_client *cl, struct ptlrpc_request *req)
228 {
229         int rc = 0;
230         ENTRY;
231
232         init_waitqueue_head(&req->rq_wait_for_rep);
233
234         rc = ptl_send_rpc(req, cl);
235         if (rc) {
236                 CERROR("error %d, opcode %d\n", rc, req->rq_reqmsg->opc);
237                 ptlrpc_cleanup_request_buf(req);
238                 up(&cl->cli_rpc_sem);
239                 RETURN(-rc);
240         }
241
242         CDEBUG(D_OTHER, "-- sleeping\n");
243         wait_event_interruptible(req->rq_wait_for_rep, ptlrpc_check_reply(req));
244         CDEBUG(D_OTHER, "-- done\n");
245         ptlrpc_cleanup_request_buf(req);
246         up(&cl->cli_rpc_sem);
247         if (req->rq_flags == PTL_RPC_INTR) {
248                 /* Clean up the dangling reply buffers */
249                 ptlrpc_abort(req);
250                 GOTO(out, rc = -EINTR);
251         }
252
253         if (req->rq_flags != PTL_RPC_REPLY) {
254                 CERROR("Unknown reason for wakeup\n");
255                 /* XXX Phil - I end up here when I kill obdctl */
256                 ptlrpc_abort(req);
257                 GOTO(out, rc = -EINTR);
258         }
259
260         rc = lustre_unpack_msg(req->rq_repbuf, req->rq_replen);
261         req->rq_repmsg = (struct lustre_msg *)req->rq_repbuf;
262         if (rc) {
263                 CERROR("unpack_rep failed: %d\n", rc);
264                 GOTO(out, rc);
265         }
266         CDEBUG(D_NET, "got rep %d\n", req->rq_repmsg->xid);
267
268         if (req->rq_repmsg->status == 0)
269                 CDEBUG(D_NET, "--> buf %p len %d status %d\n", req->rq_repbuf,
270                        req->rq_replen, req->rq_repmsg->status);
271
272         EXIT;
273  out:
274         return rc;
275 }