Optimisations to the pdu queues
[deb_libnfs.git] / lib / pdu.c
CommitLineData
84004dbf
RS
1/*
2 Copyright (C) 2010 by Ronnie Sahlberg <ronniesahlberg@gmail.com>
3
4 This program is free software; you can redistribute it and/or modify
5 it under the terms of the GNU Lesser General Public License as published by
6 the Free Software Foundation; either version 2.1 of the License, or
7 (at your option) any later version.
8
9 This program is distributed in the hope that it will be useful,
10 but WITHOUT ANY WARRANTY; without even the implied warranty of
11 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 GNU Lesser General Public License for more details.
13
14 You should have received a copy of the GNU Lesser General Public License
15 along with this program; if not, see <http://www.gnu.org/licenses/>.
16*/
72897005
RS
17#ifdef HAVE_CONFIG_H
18#include "config.h"
19#endif
20
108c622a
RS
21#ifdef AROS
22#include "aros_compat.h"
23#endif
24
a8a1b858
M
25#ifdef WIN32
26#include "win32_compat.h"
bff8fe46 27#endif
6874f61e 28
72897005
RS
29#ifdef HAVE_NETINET_IN_H
30#include <netinet/in.h>
31#endif
32
7057e733
RS
33#ifdef HAVE_SYS_SOCKET_H
34#include <sys/socket.h>
35#endif
36
bff8fe46
RS
37#ifdef HAVE_STRINGS_H
38#include <strings.h>
39#endif
40
6874f61e 41#include <stdio.h>
98f5fee8 42#include <stdlib.h>
763cd6e3 43#include <string.h>
f3a75078 44#include <assert.h>
a669857d 45#include <errno.h>
84004dbf 46#include "slist.h"
763cd6e3 47#include "libnfs-zdr.h"
84004dbf
RS
48#include "libnfs.h"
49#include "libnfs-raw.h"
50#include "libnfs-private.h"
51
aec45c62
MH
52void rpc_reset_queue(struct rpc_queue *q)
53{
54 q->head = NULL;
55 q->tail = NULL;
56}
57
58/*
59 * Push to the tail end of the queue
60 */
61void rpc_enqueue(struct rpc_queue *q, struct rpc_pdu *pdu)
62{
63 if (q->head == NULL)
64 q->head = pdu;
65 else
66 q->tail->next = pdu;
67 q->tail = pdu;
68}
69
70/*
71 * Push to the front/head of the queue
72 */
73void rpc_return_to_queue(struct rpc_queue *q, struct rpc_pdu *pdu)
74{
75 pdu->next = q->head;
76 q->head = pdu;
77 if (q->tail == NULL)
78 q->tail = pdu;
79}
80
763cd6e3 81struct rpc_pdu *rpc_allocate_pdu(struct rpc_context *rpc, int program, int version, int procedure, rpc_cb cb, void *private_data, zdrproc_t zdr_decode_fn, int zdr_decode_bufsize)
84004dbf
RS
82{
83 struct rpc_pdu *pdu;
84 struct rpc_msg msg;
85
f3a75078 86 assert(rpc->magic == RPC_CONTEXT_MAGIC);
84004dbf
RS
87
88 pdu = malloc(sizeof(struct rpc_pdu));
89 if (pdu == NULL) {
1896d37b 90 rpc_set_error(rpc, "Out of memory: Failed to allocate pdu structure");
84004dbf
RS
91 return NULL;
92 }
ea98629a 93 memset(pdu, 0, sizeof(struct rpc_pdu));
84004dbf
RS
94 pdu->xid = rpc->xid++;
95 pdu->cb = cb;
96 pdu->private_data = private_data;
763cd6e3
RS
97 pdu->zdr_decode_fn = zdr_decode_fn;
98 pdu->zdr_decode_bufsize = zdr_decode_bufsize;
84004dbf 99
763cd6e3 100 zdrmem_create(&pdu->zdr, rpc->encodebuf, rpc->encodebuflen, ZDR_ENCODE);
a1992412 101 if (rpc->is_udp == 0) {
763cd6e3 102 zdr_setpos(&pdu->zdr, 4); /* skip past the record marker */
a1992412 103 }
84004dbf 104
ea98629a 105 memset(&msg, 0, sizeof(struct rpc_msg));
aab6538b
RS
106 msg.xid = pdu->xid;
107 msg.direction = CALL;
108 msg.body.cbody.rpcvers = RPC_MSG_VERSION;
109 msg.body.cbody.prog = program;
110 msg.body.cbody.vers = version;
111 msg.body.cbody.proc = procedure;
112 msg.body.cbody.cred = rpc->auth->ah_cred;
113 msg.body.cbody.verf = rpc->auth->ah_verf;
84004dbf 114
f0cb8042
RS
115 if (zdr_callmsg(rpc, &pdu->zdr, &msg) == 0) {
116 rpc_set_error(rpc, "zdr_callmsg failed with %s",
117 rpc_get_error(rpc));
763cd6e3 118 zdr_destroy(&pdu->zdr);
84004dbf
RS
119 free(pdu);
120 return NULL;
121 }
122
123 return pdu;
124}
125
f3a75078 126void rpc_free_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu)
84004dbf 127{
f3a75078
RS
128 assert(rpc->magic == RPC_CONTEXT_MAGIC);
129
84004dbf
RS
130 if (pdu->outdata.data != NULL) {
131 free(pdu->outdata.data);
132 pdu->outdata.data = NULL;
133 }
134
763cd6e3
RS
135 if (pdu->zdr_decode_buf != NULL) {
136 zdr_free(pdu->zdr_decode_fn, pdu->zdr_decode_buf);
137 free(pdu->zdr_decode_buf);
138 pdu->zdr_decode_buf = NULL;
84004dbf
RS
139 }
140
763cd6e3 141 zdr_destroy(&pdu->zdr);
df5af25f 142
84004dbf
RS
143 free(pdu);
144}
145
3b943d2f
RS
146void rpc_set_next_xid(struct rpc_context *rpc, uint32_t xid)
147{
148 rpc->xid = xid;
149}
84004dbf
RS
150
151int rpc_queue_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu)
152{
153 int size, recordmarker;
154
f3a75078
RS
155 assert(rpc->magic == RPC_CONTEXT_MAGIC);
156
763cd6e3 157 size = zdr_getpos(&pdu->zdr);
84004dbf 158
a669857d
RS
159 /* for udp we dont queue, we just send it straight away */
160 if (rpc->is_udp != 0) {
763cd6e3 161// XXX add a rpc->udp_dest_sock_size and get rid of sys/socket.h and netinet/in.h
a669857d
RS
162 if (sendto(rpc->fd, rpc->encodebuf, size, MSG_DONTWAIT, rpc->udp_dest, sizeof(struct sockaddr_in)) < 0) {
163 rpc_set_error(rpc, "Sendto failed with errno %s", strerror(errno));
164 rpc_free_pdu(rpc, pdu);
165 return -1;
166 }
aec45c62 167 rpc_enqueue(&rpc->waitpdu, pdu);
a669857d
RS
168 return 0;
169 }
170
84004dbf 171 /* write recordmarker */
763cd6e3 172 zdr_setpos(&pdu->zdr, 0);
84004dbf 173 recordmarker = (size - 4) | 0x80000000;
763cd6e3 174 zdr_int(&pdu->zdr, &recordmarker);
84004dbf
RS
175
176 pdu->outdata.size = size;
177 pdu->outdata.data = malloc(pdu->outdata.size);
178 if (pdu->outdata.data == NULL) {
179 rpc_set_error(rpc, "Out of memory. Failed to allocate buffer for pdu\n");
180 rpc_free_pdu(rpc, pdu);
181 return -1;
182 }
183
184 memcpy(pdu->outdata.data, rpc->encodebuf, pdu->outdata.size);
aec45c62 185 rpc_enqueue(&rpc->outqueue, pdu);
84004dbf
RS
186
187 return 0;
188}
189
190int rpc_get_pdu_size(char *buf)
191{
192 uint32_t size;
193
194 size = ntohl(*(uint32_t *)buf);
195
84004dbf
RS
196 return (size & 0x7fffffff) + 4;
197}
198
763cd6e3 199static int rpc_process_reply(struct rpc_context *rpc, struct rpc_pdu *pdu, ZDR *zdr)
84004dbf
RS
200{
201 struct rpc_msg msg;
202
f3a75078
RS
203 assert(rpc->magic == RPC_CONTEXT_MAGIC);
204
ea98629a 205 memset(&msg, 0, sizeof(struct rpc_msg));
aab6538b 206 msg.body.rbody.reply.areply.verf = _null_auth;
763cd6e3
RS
207 if (pdu->zdr_decode_bufsize > 0) {
208 if (pdu->zdr_decode_buf != NULL) {
209 free(pdu->zdr_decode_buf);
1b9917b8 210 }
763cd6e3
RS
211 pdu->zdr_decode_buf = malloc(pdu->zdr_decode_bufsize);
212 if (pdu->zdr_decode_buf == NULL) {
f0cb8042
RS
213 rpc_set_error(rpc, "Failed to allocate memory for "
214 "zdr_encode_buf in rpc_process_reply");
215 pdu->cb(rpc, RPC_STATUS_ERROR, "Failed to allocate "
216 "buffer for decoding of ZDR reply",
217 pdu->private_data);
84004dbf
RS
218 return 0;
219 }
763cd6e3 220 memset(pdu->zdr_decode_buf, 0, pdu->zdr_decode_bufsize);
84004dbf 221 }
aab6538b
RS
222 msg.body.rbody.reply.areply.reply_data.results.where = pdu->zdr_decode_buf;
223 msg.body.rbody.reply.areply.reply_data.results.proc = pdu->zdr_decode_fn;
84004dbf 224
f0cb8042
RS
225 if (zdr_replymsg(rpc, zdr, &msg) == 0) {
226 rpc_set_error(rpc, "zdr_replymsg failed in rpc_process_reply: "
227 "%s", rpc_get_error(rpc));
228 pdu->cb(rpc, RPC_STATUS_ERROR, "Message rejected by server",
229 pdu->private_data);
763cd6e3
RS
230 if (pdu->zdr_decode_buf != NULL) {
231 free(pdu->zdr_decode_buf);
232 pdu->zdr_decode_buf = NULL;
84004dbf
RS
233 }
234 return 0;
235 }
aab6538b 236 if (msg.body.rbody.stat != MSG_ACCEPTED) {
84004dbf
RS
237 pdu->cb(rpc, RPC_STATUS_ERROR, "RPC Packet not accepted by the server", pdu->private_data);
238 return 0;
239 }
aab6538b 240 switch (msg.body.rbody.reply.areply.stat) {
84004dbf 241 case SUCCESS:
763cd6e3 242 pdu->cb(rpc, RPC_STATUS_SUCCESS, pdu->zdr_decode_buf, pdu->private_data);
84004dbf
RS
243 break;
244 case PROG_UNAVAIL:
245 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: Program not available", pdu->private_data);
246 break;
247 case PROG_MISMATCH:
248 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: Program version mismatch", pdu->private_data);
249 break;
250 case PROC_UNAVAIL:
251 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: Procedure not available", pdu->private_data);
252 break;
253 case GARBAGE_ARGS:
254 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: Garbage arguments", pdu->private_data);
255 break;
256 case SYSTEM_ERR:
257 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: System Error", pdu->private_data);
258 break;
259 default:
260 pdu->cb(rpc, RPC_STATUS_ERROR, "Unknown rpc response from server", pdu->private_data);
261 break;
262 }
263
264 return 0;
265}
266
267int rpc_process_pdu(struct rpc_context *rpc, char *buf, int size)
268{
aec45c62 269 struct rpc_pdu *pdu, *prev_pdu;
763cd6e3 270 ZDR zdr;
d678b73e 271 int pos, recordmarker = 0;
b93082da 272 uint32_t xid;
d678b73e 273 char *reasbuf = NULL;
84004dbf 274
f3a75078
RS
275 assert(rpc->magic == RPC_CONTEXT_MAGIC);
276
763cd6e3 277 memset(&zdr, 0, sizeof(ZDR));
84004dbf 278
763cd6e3 279 zdrmem_create(&zdr, buf, size, ZDR_DECODE);
a669857d 280 if (rpc->is_udp == 0) {
763cd6e3
RS
281 if (zdr_int(&zdr, &recordmarker) == 0) {
282 rpc_set_error(rpc, "zdr_int reading recordmarker failed");
283 zdr_destroy(&zdr);
a669857d
RS
284 return -1;
285 }
d678b73e 286 if (!(recordmarker&0x80000000)) {
763cd6e3 287 zdr_destroy(&zdr);
d678b73e
RS
288 if (rpc_add_fragment(rpc, buf+4, size-4) != 0) {
289 rpc_set_error(rpc, "Failed to queue fragment for reassembly.");
290 return -1;
291 }
292 return 0;
293 }
84004dbf 294 }
d678b73e
RS
295
296 /* reassembly */
297 if (recordmarker != 0 && rpc->fragments != NULL) {
298 struct rpc_fragment *fragment;
574095b9 299 uint32_t total = size - 4;
d678b73e
RS
300 char *ptr;
301
763cd6e3 302 zdr_destroy(&zdr);
d678b73e
RS
303 for (fragment = rpc->fragments; fragment; fragment = fragment->next) {
304 total += fragment->size;
305 }
306
307 reasbuf = malloc(total);
308 if (reasbuf == NULL) {
309 rpc_set_error(rpc, "Failed to reassemble PDU");
310 rpc_free_all_fragments(rpc);
311 return -1;
312 }
313 ptr = reasbuf;
314 for (fragment = rpc->fragments; fragment; fragment = fragment->next) {
315 memcpy(ptr, fragment->data, fragment->size);
316 ptr += fragment->size;
317 }
318 memcpy(ptr, buf + 4, size - 4);
763cd6e3 319 zdrmem_create(&zdr, reasbuf, total, ZDR_DECODE);
d678b73e
RS
320 rpc_free_all_fragments(rpc);
321 }
322
763cd6e3
RS
323 pos = zdr_getpos(&zdr);
324 if (zdr_int(&zdr, (int *)&xid) == 0) {
325 rpc_set_error(rpc, "zdr_int reading xid failed");
326 zdr_destroy(&zdr);
d678b73e
RS
327 if (reasbuf != NULL) {
328 free(reasbuf);
329 }
84004dbf
RS
330 return -1;
331 }
763cd6e3 332 zdr_setpos(&zdr, pos);
84004dbf 333
aec45c62
MH
334 /* Linear traverse singly-linked list, but track previous
335 * entry for optimised removal */
336 prev_pdu = NULL;
337 for (pdu=rpc->waitpdu.head; pdu; pdu=pdu->next) {
84004dbf 338 if (pdu->xid != xid) {
aec45c62 339 prev_pdu = pdu;
84004dbf
RS
340 continue;
341 }
a669857d 342 if (rpc->is_udp == 0 || rpc->is_broadcast == 0) {
aec45c62
MH
343 /* Singly-linked but we track head and tail */
344 if (pdu == rpc->waitpdu.head)
345 rpc->waitpdu.head = pdu->next;
346 if (pdu == rpc->waitpdu.tail)
347 rpc->waitpdu.tail = prev_pdu;
348 if (prev_pdu != NULL)
349 prev_pdu->next = pdu->next;
a669857d 350 }
763cd6e3 351 if (rpc_process_reply(rpc, pdu, &zdr) != 0) {
1896d37b 352 rpc_set_error(rpc, "rpc_procdess_reply failed");
84004dbf 353 }
763cd6e3 354 zdr_destroy(&zdr);
a669857d
RS
355 if (rpc->is_udp == 0 || rpc->is_broadcast == 0) {
356 rpc_free_pdu(rpc, pdu);
357 }
d678b73e
RS
358 if (reasbuf != NULL) {
359 free(reasbuf);
360 }
84004dbf
RS
361 return 0;
362 }
1896d37b 363 rpc_set_error(rpc, "No matching pdu found for xid:%d", xid);
763cd6e3 364 zdr_destroy(&zdr);
d678b73e
RS
365 if (reasbuf != NULL) {
366 free(reasbuf);
367 }
84004dbf
RS
368 return -1;
369}
370