2 Copyright (C) 2010 by Ronnie Sahlberg <ronniesahlberg@gmail.com>
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.
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.
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/>.
22 #include "aros_compat.h"
26 #include "win32_compat.h"
29 #ifdef HAVE_NETINET_IN_H
30 #include <netinet/in.h>
33 #ifdef HAVE_SYS_SOCKET_H
34 #include <sys/socket.h>
47 #include "libnfs-zdr.h"
49 #include "libnfs-raw.h"
50 #include "libnfs-private.h"
52 void rpc_reset_queue(struct rpc_queue
*q
)
59 * Push to the tail end of the queue
61 void rpc_enqueue(struct rpc_queue
*q
, struct rpc_pdu
*pdu
)
72 * Push to the front/head of the queue
74 void rpc_return_to_queue(struct rpc_queue
*q
, struct rpc_pdu
*pdu
)
82 unsigned int rpc_hash_xid(uint32_t xid
)
84 return (xid
* 7919) % HASHES
;
87 struct 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
)
92 assert(rpc
->magic
== RPC_CONTEXT_MAGIC
);
94 pdu
= malloc(sizeof(struct rpc_pdu
));
96 rpc_set_error(rpc
, "Out of memory: Failed to allocate pdu structure");
99 memset(pdu
, 0, sizeof(struct rpc_pdu
));
100 pdu
->xid
= rpc
->xid
++;
102 pdu
->private_data
= private_data
;
103 pdu
->zdr_decode_fn
= zdr_decode_fn
;
104 pdu
->zdr_decode_bufsize
= zdr_decode_bufsize
;
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 */
111 memset(&msg
, 0, sizeof(struct rpc_msg
));
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
;
121 if (zdr_callmsg(rpc
, &pdu
->zdr
, &msg
) == 0) {
122 rpc_set_error(rpc
, "zdr_callmsg failed with %s",
124 zdr_destroy(&pdu
->zdr
);
132 void rpc_free_pdu(struct rpc_context
*rpc
, struct rpc_pdu
*pdu
)
134 assert(rpc
->magic
== RPC_CONTEXT_MAGIC
);
136 if (pdu
->outdata
.data
!= NULL
) {
137 free(pdu
->outdata
.data
);
138 pdu
->outdata
.data
= NULL
;
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
;
147 zdr_destroy(&pdu
->zdr
);
152 void rpc_set_next_xid(struct rpc_context
*rpc
, uint32_t xid
)
157 int rpc_queue_pdu(struct rpc_context
*rpc
, struct rpc_pdu
*pdu
)
159 int size
, recordmarker
;
161 assert(rpc
->magic
== RPC_CONTEXT_MAGIC
);
163 size
= zdr_getpos(&pdu
->zdr
);
165 /* for udp we dont queue, we just send it straight away */
166 if (rpc
->is_udp
!= 0) {
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
);
176 hash
= rpc_hash_xid(pdu
->xid
);
177 rpc_enqueue(&rpc
->waitpdu
[hash
], pdu
);
181 /* write recordmarker */
182 zdr_setpos(&pdu
->zdr
, 0);
183 recordmarker
= (size
- 4) | 0x80000000;
184 zdr_int(&pdu
->zdr
, &recordmarker
);
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
);
194 memcpy(pdu
->outdata
.data
, rpc
->encodebuf
, pdu
->outdata
.size
);
195 rpc_enqueue(&rpc
->outqueue
, pdu
);
200 int rpc_get_pdu_size(char *buf
)
204 size
= ntohl(*(uint32_t *)buf
);
206 return (size
& 0x7fffffff) + 4;
209 static int rpc_process_reply(struct rpc_context
*rpc
, struct rpc_pdu
*pdu
, ZDR
*zdr
)
213 assert(rpc
->magic
== RPC_CONTEXT_MAGIC
);
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
);
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",
230 memset(pdu
->zdr_decode_buf
, 0, pdu
->zdr_decode_bufsize
);
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
;
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",
240 if (pdu
->zdr_decode_buf
!= NULL
) {
241 free(pdu
->zdr_decode_buf
);
242 pdu
->zdr_decode_buf
= NULL
;
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
);
250 switch (msg
.body
.rbody
.reply
.areply
.stat
) {
252 pdu
->cb(rpc
, RPC_STATUS_SUCCESS
, pdu
->zdr_decode_buf
, pdu
->private_data
);
255 pdu
->cb(rpc
, RPC_STATUS_ERROR
, "Server responded: Program not available", pdu
->private_data
);
258 pdu
->cb(rpc
, RPC_STATUS_ERROR
, "Server responded: Program version mismatch", pdu
->private_data
);
261 pdu
->cb(rpc
, RPC_STATUS_ERROR
, "Server responded: Procedure not available", pdu
->private_data
);
264 pdu
->cb(rpc
, RPC_STATUS_ERROR
, "Server responded: Garbage arguments", pdu
->private_data
);
267 pdu
->cb(rpc
, RPC_STATUS_ERROR
, "Server responded: System Error", pdu
->private_data
);
270 pdu
->cb(rpc
, RPC_STATUS_ERROR
, "Unknown rpc response from server", pdu
->private_data
);
277 int rpc_process_pdu(struct rpc_context
*rpc
, char *buf
, int size
)
279 struct rpc_pdu
*pdu
, *prev_pdu
;
282 int pos
, recordmarker
= 0;
285 char *reasbuf
= NULL
;
287 assert(rpc
->magic
== RPC_CONTEXT_MAGIC
);
289 memset(&zdr
, 0, sizeof(ZDR
));
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");
298 if (!(recordmarker
&0x80000000)) {
300 if (rpc_add_fragment(rpc
, buf
+4, size
-4) != 0) {
301 rpc_set_error(rpc
, "Failed to queue fragment for reassembly.");
309 if (recordmarker
!= 0 && rpc
->fragments
!= NULL
) {
310 struct rpc_fragment
*fragment
;
311 uint32_t total
= size
- 4;
315 for (fragment
= rpc
->fragments
; fragment
; fragment
= fragment
->next
) {
316 total
+= fragment
->size
;
319 reasbuf
= malloc(total
);
320 if (reasbuf
== NULL
) {
321 rpc_set_error(rpc
, "Failed to reassemble PDU");
322 rpc_free_all_fragments(rpc
);
326 for (fragment
= rpc
->fragments
; fragment
; fragment
= fragment
->next
) {
327 memcpy(ptr
, fragment
->data
, fragment
->size
);
328 ptr
+= fragment
->size
;
330 memcpy(ptr
, buf
+ 4, size
- 4);
331 zdrmem_create(&zdr
, reasbuf
, total
, ZDR_DECODE
);
332 rpc_free_all_fragments(rpc
);
335 pos
= zdr_getpos(&zdr
);
336 if (zdr_int(&zdr
, (int *)&xid
) == 0) {
337 rpc_set_error(rpc
, "zdr_int reading xid failed");
339 if (reasbuf
!= NULL
) {
344 zdr_setpos(&zdr
, pos
);
346 /* Look up the transaction in a hash table of our requests */
347 hash
= rpc_hash_xid(xid
);
348 q
= &rpc
->waitpdu
[hash
];
350 /* Follow the hash chain. Linear traverse singly-linked list,
351 * but track previous entry for optimised removal */
353 for (pdu
=q
->head
; pdu
; pdu
=pdu
->next
) {
354 if (pdu
->xid
!= xid
) {
358 if (rpc
->is_udp
== 0 || rpc
->is_broadcast
== 0) {
359 /* Singly-linked but we track head and tail */
364 if (prev_pdu
!= NULL
)
365 prev_pdu
->next
= pdu
->next
;
367 if (rpc_process_reply(rpc
, pdu
, &zdr
) != 0) {
368 rpc_set_error(rpc
, "rpc_procdess_reply failed");
371 if (rpc
->is_udp
== 0 || rpc
->is_broadcast
== 0) {
372 rpc_free_pdu(rpc
, pdu
);
374 if (reasbuf
!= NULL
) {
379 rpc_set_error(rpc
, "No matching pdu found for xid:%d", xid
);
381 if (reasbuf
!= NULL
) {