socket.c: we need netinet/in.h on some platforms
[deb_libnfs.git] / lib / pdu.c
... / ...
CommitLineData
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*/
17#ifdef HAVE_CONFIG_H
18#include "config.h"
19#endif
20
21#ifdef AROS
22#include "aros_compat.h"
23#endif
24
25#ifdef WIN32
26#include "win32_compat.h"
27#endif
28
29#ifdef HAVE_NETINET_IN_H
30#include <netinet/in.h>
31#endif
32
33#ifdef HAVE_SYS_SOCKET_H
34#include <sys/socket.h>
35#endif
36
37#ifdef HAVE_STRINGS_H
38#include <strings.h>
39#endif
40
41#include <stdio.h>
42#include <stdlib.h>
43#include <string.h>
44#include <assert.h>
45#include <errno.h>
46#include "slist.h"
47#include "libnfs-zdr.h"
48#include "libnfs.h"
49#include "libnfs-raw.h"
50#include "libnfs-private.h"
51
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 pdu->next = NULL;
69}
70
71/*
72 * Push to the front/head of the queue
73 */
74void rpc_return_to_queue(struct rpc_queue *q, struct rpc_pdu *pdu)
75{
76 pdu->next = q->head;
77 q->head = pdu;
78 if (q->tail == NULL)
79 q->tail = pdu;
80}
81
82unsigned int rpc_hash_xid(uint32_t xid)
83{
84 return (xid * 7919) % HASHES;
85}
86
87struct 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)
88{
89 struct rpc_pdu *pdu;
90 struct rpc_msg msg;
91
92 assert(rpc->magic == RPC_CONTEXT_MAGIC);
93
94 pdu = malloc(sizeof(struct rpc_pdu));
95 if (pdu == NULL) {
96 rpc_set_error(rpc, "Out of memory: Failed to allocate pdu structure");
97 return NULL;
98 }
99 memset(pdu, 0, sizeof(struct rpc_pdu));
100 pdu->xid = rpc->xid++;
101 pdu->cb = cb;
102 pdu->private_data = private_data;
103 pdu->zdr_decode_fn = zdr_decode_fn;
104 pdu->zdr_decode_bufsize = zdr_decode_bufsize;
105
106 zdrmem_create(&pdu->zdr, rpc->encodebuf, rpc->encodebuflen, ZDR_ENCODE);
107 if (rpc->is_udp == 0) {
108 zdr_setpos(&pdu->zdr, 4); /* skip past the record marker */
109 }
110
111 memset(&msg, 0, sizeof(struct rpc_msg));
112 msg.xid = pdu->xid;
113 msg.direction = CALL;
114 msg.body.cbody.rpcvers = RPC_MSG_VERSION;
115 msg.body.cbody.prog = program;
116 msg.body.cbody.vers = version;
117 msg.body.cbody.proc = procedure;
118 msg.body.cbody.cred = rpc->auth->ah_cred;
119 msg.body.cbody.verf = rpc->auth->ah_verf;
120
121 if (zdr_callmsg(rpc, &pdu->zdr, &msg) == 0) {
122 rpc_set_error(rpc, "zdr_callmsg failed with %s",
123 rpc_get_error(rpc));
124 zdr_destroy(&pdu->zdr);
125 free(pdu);
126 return NULL;
127 }
128
129 return pdu;
130}
131
132void rpc_free_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu)
133{
134 assert(rpc->magic == RPC_CONTEXT_MAGIC);
135
136 if (pdu->outdata.data != NULL) {
137 free(pdu->outdata.data);
138 pdu->outdata.data = NULL;
139 }
140
141 if (pdu->zdr_decode_buf != NULL) {
142 zdr_free(pdu->zdr_decode_fn, pdu->zdr_decode_buf);
143 free(pdu->zdr_decode_buf);
144 pdu->zdr_decode_buf = NULL;
145 }
146
147 zdr_destroy(&pdu->zdr);
148
149 free(pdu);
150}
151
152void rpc_set_next_xid(struct rpc_context *rpc, uint32_t xid)
153{
154 rpc->xid = xid;
155}
156
157int rpc_queue_pdu(struct rpc_context *rpc, struct rpc_pdu *pdu)
158{
159 int size, recordmarker;
160
161 assert(rpc->magic == RPC_CONTEXT_MAGIC);
162
163 size = zdr_getpos(&pdu->zdr);
164
165 /* for udp we dont queue, we just send it straight away */
166 if (rpc->is_udp != 0) {
167 unsigned int hash;
168
169// XXX add a rpc->udp_dest_sock_size and get rid of sys/socket.h and netinet/in.h
170 if (sendto(rpc->fd, rpc->encodebuf, size, MSG_DONTWAIT, rpc->udp_dest, sizeof(struct sockaddr_in)) < 0) {
171 rpc_set_error(rpc, "Sendto failed with errno %s", strerror(errno));
172 rpc_free_pdu(rpc, pdu);
173 return -1;
174 }
175
176 hash = rpc_hash_xid(pdu->xid);
177 rpc_enqueue(&rpc->waitpdu[hash], pdu);
178 return 0;
179 }
180
181 /* write recordmarker */
182 zdr_setpos(&pdu->zdr, 0);
183 recordmarker = (size - 4) | 0x80000000;
184 zdr_int(&pdu->zdr, &recordmarker);
185
186 pdu->outdata.size = size;
187 pdu->outdata.data = malloc(pdu->outdata.size);
188 if (pdu->outdata.data == NULL) {
189 rpc_set_error(rpc, "Out of memory. Failed to allocate buffer for pdu\n");
190 rpc_free_pdu(rpc, pdu);
191 return -1;
192 }
193
194 memcpy(pdu->outdata.data, rpc->encodebuf, pdu->outdata.size);
195 rpc_enqueue(&rpc->outqueue, pdu);
196
197 return 0;
198}
199
200int rpc_get_pdu_size(char *buf)
201{
202 uint32_t size;
203
204 size = ntohl(*(uint32_t *)buf);
205
206 return (size & 0x7fffffff) + 4;
207}
208
209static int rpc_process_reply(struct rpc_context *rpc, struct rpc_pdu *pdu, ZDR *zdr)
210{
211 struct rpc_msg msg;
212
213 assert(rpc->magic == RPC_CONTEXT_MAGIC);
214
215 memset(&msg, 0, sizeof(struct rpc_msg));
216 msg.body.rbody.reply.areply.verf = _null_auth;
217 if (pdu->zdr_decode_bufsize > 0) {
218 if (pdu->zdr_decode_buf != NULL) {
219 free(pdu->zdr_decode_buf);
220 }
221 pdu->zdr_decode_buf = malloc(pdu->zdr_decode_bufsize);
222 if (pdu->zdr_decode_buf == NULL) {
223 rpc_set_error(rpc, "Failed to allocate memory for "
224 "zdr_encode_buf in rpc_process_reply");
225 pdu->cb(rpc, RPC_STATUS_ERROR, "Failed to allocate "
226 "buffer for decoding of ZDR reply",
227 pdu->private_data);
228 return 0;
229 }
230 memset(pdu->zdr_decode_buf, 0, pdu->zdr_decode_bufsize);
231 }
232 msg.body.rbody.reply.areply.reply_data.results.where = pdu->zdr_decode_buf;
233 msg.body.rbody.reply.areply.reply_data.results.proc = pdu->zdr_decode_fn;
234
235 if (zdr_replymsg(rpc, zdr, &msg) == 0) {
236 rpc_set_error(rpc, "zdr_replymsg failed in rpc_process_reply: "
237 "%s", rpc_get_error(rpc));
238 pdu->cb(rpc, RPC_STATUS_ERROR, "Message rejected by server",
239 pdu->private_data);
240 if (pdu->zdr_decode_buf != NULL) {
241 free(pdu->zdr_decode_buf);
242 pdu->zdr_decode_buf = NULL;
243 }
244 return 0;
245 }
246 if (msg.body.rbody.stat != MSG_ACCEPTED) {
247 pdu->cb(rpc, RPC_STATUS_ERROR, "RPC Packet not accepted by the server", pdu->private_data);
248 return 0;
249 }
250 switch (msg.body.rbody.reply.areply.stat) {
251 case SUCCESS:
252 pdu->cb(rpc, RPC_STATUS_SUCCESS, pdu->zdr_decode_buf, pdu->private_data);
253 break;
254 case PROG_UNAVAIL:
255 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: Program not available", pdu->private_data);
256 break;
257 case PROG_MISMATCH:
258 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: Program version mismatch", pdu->private_data);
259 break;
260 case PROC_UNAVAIL:
261 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: Procedure not available", pdu->private_data);
262 break;
263 case GARBAGE_ARGS:
264 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: Garbage arguments", pdu->private_data);
265 break;
266 case SYSTEM_ERR:
267 pdu->cb(rpc, RPC_STATUS_ERROR, "Server responded: System Error", pdu->private_data);
268 break;
269 default:
270 pdu->cb(rpc, RPC_STATUS_ERROR, "Unknown rpc response from server", pdu->private_data);
271 break;
272 }
273
274 return 0;
275}
276
277int rpc_process_pdu(struct rpc_context *rpc, char *buf, int size)
278{
279 struct rpc_pdu *pdu, *prev_pdu;
280 struct rpc_queue *q;
281 ZDR zdr;
282 int pos, recordmarker = 0;
283 unsigned int hash;
284 uint32_t xid;
285 char *reasbuf = NULL;
286
287 assert(rpc->magic == RPC_CONTEXT_MAGIC);
288
289 memset(&zdr, 0, sizeof(ZDR));
290
291 zdrmem_create(&zdr, buf, size, ZDR_DECODE);
292 if (rpc->is_udp == 0) {
293 if (zdr_int(&zdr, &recordmarker) == 0) {
294 rpc_set_error(rpc, "zdr_int reading recordmarker failed");
295 zdr_destroy(&zdr);
296 return -1;
297 }
298 if (!(recordmarker&0x80000000)) {
299 zdr_destroy(&zdr);
300 if (rpc_add_fragment(rpc, buf+4, size-4) != 0) {
301 rpc_set_error(rpc, "Failed to queue fragment for reassembly.");
302 return -1;
303 }
304 return 0;
305 }
306 }
307
308 /* reassembly */
309 if (recordmarker != 0 && rpc->fragments != NULL) {
310 struct rpc_fragment *fragment;
311 uint32_t total = size - 4;
312 char *ptr;
313
314 zdr_destroy(&zdr);
315 for (fragment = rpc->fragments; fragment; fragment = fragment->next) {
316 total += fragment->size;
317 }
318
319 reasbuf = malloc(total);
320 if (reasbuf == NULL) {
321 rpc_set_error(rpc, "Failed to reassemble PDU");
322 rpc_free_all_fragments(rpc);
323 return -1;
324 }
325 ptr = reasbuf;
326 for (fragment = rpc->fragments; fragment; fragment = fragment->next) {
327 memcpy(ptr, fragment->data, fragment->size);
328 ptr += fragment->size;
329 }
330 memcpy(ptr, buf + 4, size - 4);
331 zdrmem_create(&zdr, reasbuf, total, ZDR_DECODE);
332 rpc_free_all_fragments(rpc);
333 }
334
335 pos = zdr_getpos(&zdr);
336 if (zdr_int(&zdr, (int *)&xid) == 0) {
337 rpc_set_error(rpc, "zdr_int reading xid failed");
338 zdr_destroy(&zdr);
339 if (reasbuf != NULL) {
340 free(reasbuf);
341 }
342 return -1;
343 }
344 zdr_setpos(&zdr, pos);
345
346 /* Look up the transaction in a hash table of our requests */
347 hash = rpc_hash_xid(xid);
348 q = &rpc->waitpdu[hash];
349
350 /* Follow the hash chain. Linear traverse singly-linked list,
351 * but track previous entry for optimised removal */
352 prev_pdu = NULL;
353 for (pdu=q->head; pdu; pdu=pdu->next) {
354 if (pdu->xid != xid) {
355 prev_pdu = pdu;
356 continue;
357 }
358 if (rpc->is_udp == 0 || rpc->is_broadcast == 0) {
359 /* Singly-linked but we track head and tail */
360 if (pdu == q->head)
361 q->head = pdu->next;
362 if (pdu == q->tail)
363 q->tail = prev_pdu;
364 if (prev_pdu != NULL)
365 prev_pdu->next = pdu->next;
366 }
367 if (rpc_process_reply(rpc, pdu, &zdr) != 0) {
368 rpc_set_error(rpc, "rpc_procdess_reply failed");
369 }
370 zdr_destroy(&zdr);
371 if (rpc->is_udp == 0 || rpc->is_broadcast == 0) {
372 rpc_free_pdu(rpc, pdu);
373 }
374 if (reasbuf != NULL) {
375 free(reasbuf);
376 }
377 return 0;
378 }
379 rpc_set_error(rpc, "No matching pdu found for xid:%d", xid);
380 zdr_destroy(&zdr);
381 if (reasbuf != NULL) {
382 free(reasbuf);
383 }
384 return -1;
385}
386