Context Navigation

source: rtems/cpukit/libfs/src/nfsclient/src/rpcio.c @ e7583f67

4.104.115

Last change on this file since e7583f67 was 355b0544, checked in by Chris Johns <chrisj@…>, on 03/27/10 at 04:04:40

2010-03-27 Chris Johns <chrisj@…>

libfs/src/nfsclient/src/cexphelp.c,
libfs/src/nfsclient/src/dirutils.c,
libfs/src/nfsclient/src/nfs.modini.c,
libfs/src/nfsclient/src/nfsTest.c,
libfs/src/nfsclient/src/rpcio.c,
libfs/src/nfsclient/src/rpcio.modini.c,
libfs/src/nfsclient/src/sock_mbuf.c,
libfs/src/nfsclient/src/xdr_mbuf.c,
libfs/src/rfs/rtems-rfs-bitmaps-ut.c,
libfs/src/rfs/rtems-rfs-bitmaps.c,
libfs/src/rfs/rtems-rfs-block.c,
libfs/src/rfs/rtems-rfs-buffer-bdbuf.c,
libfs/src/rfs/rtems-rfs-buffer-devio.c,
libfs/src/rfs/rtems-rfs-buffer.c,
libfs/src/rfs/rtems-rfs-dir-hash.c, libfs/src/rfs/rtems-rfs-dir.c,
libfs/src/rfs/rtems-rfs-file-system.c,
libfs/src/rfs/rtems-rfs-file.c, libfs/src/rfs/rtems-rfs-format.c,
libfs/src/rfs/rtems-rfs-group.c, libfs/src/rfs/rtems-rfs-inode.c,
libfs/src/rfs/rtems-rfs-link.c, libfs/src/rfs/rtems-rfs-mutex.c,
libfs/src/rfs/rtems-rfs-rtems-dev.c,
libfs/src/rfs/rtems-rfs-rtems-dir.c,
libfs/src/rfs/rtems-rfs-rtems-file.c,
libfs/src/rfs/rtems-rfs-rtems-utils.c,
libfs/src/rfs/rtems-rfs-rtems.c, libfs/src/rfs/rtems-rfs-shell.c,
libfs/src/rfs/rtems-rfs-trace.c: Add HAVE_CONFIG_H support to let
files receive configure defines.

Property mode set to 100644

File size: 44.3 KB

Line
1	/* $Id$ */
2
3	/* RPC multiplexor for a multitasking environment */
4
5	/* Author: Till Straumann <strauman@slac.stanford.edu>, 2002 */
6
7	/* This code funnels arbitrary task's UDP/RPC requests
8	* through one socket to arbitrary servers.
9	* The replies are gathered and dispatched to the
10	* requestors.
11	* One task handles all the sending and receiving
12	* work including retries.
13	* It is up to the requestor, however, to do
14	* the XDR encoding of the arguments / decoding
15	* of the results (except for the RPC header which
16	* is handled by the daemon).
17	*/
18
19	/*
20	* Authorship
21	* ----------
22	* This software (NFS-2 client implementation for RTEMS) was created by
23	* Till Straumann <strauman@slac.stanford.edu>, 2002-2007,
24	* Stanford Linear Accelerator Center, Stanford University.
25	*
26	* Acknowledgement of sponsorship
27	* ------------------------------
28	* The NFS-2 client implementation for RTEMS was produced by
29	* the Stanford Linear Accelerator Center, Stanford University,
30	* under Contract DE-AC03-76SFO0515 with the Department of Energy.
31	*
32	* Government disclaimer of liability
33	* ----------------------------------
34	* Neither the United States nor the United States Department of Energy,
35	* nor any of their employees, makes any warranty, express or implied, or
36	* assumes any legal liability or responsibility for the accuracy,
37	* completeness, or usefulness of any data, apparatus, product, or process
38	* disclosed, or represents that its use would not infringe privately owned
39	* rights.
40	*
41	* Stanford disclaimer of liability
42	* --------------------------------
43	* Stanford University makes no representations or warranties, express or
44	* implied, nor assumes any liability for the use of this software.
45	*
46	* Stanford disclaimer of copyright
47	* --------------------------------
48	* Stanford University, owner of the copyright, hereby disclaims its
49	* copyright and all other rights in this software. Hence, anyone may
50	* freely use it for any purpose without restriction.
51	*
52	* Maintenance of notices
53	* ----------------------
54	* In the interest of clarity regarding the origin and status of this
55	* SLAC software, this and all the preceding Stanford University notices
56	* are to remain affixed to any copy or derivative of this software made
57	* or distributed by the recipient and are to be affixed to any copy of
58	* software made or distributed by the recipient that contains a copy or
59	* derivative of this software.
60	*
61	* ------------------ SLAC Software Notices, Set 4 OTT.002a, 2004 FEB 03
62	*/
63
64	#if HAVE_CONFIG_H
65	#include "config.h"
66	#endif
67
68	#include <rtems.h>
69	#include <rtems/error.h>
70	#include <rtems/rtems_bsdnet.h>
71	#include <stdlib.h>
72	#include <time.h>
73	#include <rpc/rpc.h>
74	#include <rpc/pmap_prot.h>
75	#include <errno.h>
76	#include <sys/ioctl.h>
77	#include <assert.h>
78	#include <stdio.h>
79	#include <errno.h>
80	#include <string.h>
81	#include <netinet/in.h>
82	#include <arpa/inet.h>
83
84	#include "rpcio.h"
85
86	/****************************************************************/
87	/* CONFIGURABLE PARAMETERS */
88	/****************************************************************/
89
90	#define MBUF_RX /* If defined: use mbuf XDR stream for
91	* decoding directly out of mbufs
92	* Otherwise, the regular 'recvfrom()'
93	* interface will be used involving an
94	* extra buffer allocation + copy step.
95	*/
96
97	#define MBUF_TX /* If defined: avoid copying data when
98	* sending. Instead, use a wrapper to
99	* 'sosend()' which will point an MBUF
100	* directly to our buffer space.
101	* Note that the BSD stack does not copy
102	* data when fragmenting packets - it
103	* merely uses an mbuf chain pointing
104	* into different areas of the data.
105	*
106	* If undefined, the regular 'sendto()'
107	* interface is used.
108	*/
109
110	#undef REJECT_SERVERIP_MISMATCH
111	/* If defined, RPC replies must come from the server
112	* that was queried. Eric Norum has reported problems
113	* with clustered NFS servers. So we disable this
114	* reducing paranoia...
115	*/
116
117	/* daemon task parameters */
118	#define RPCIOD_STACK 10000
119	#define RPCIOD_PRIO 100 /* fallback priority */
120
121	/* depth of the message queue for sending
122	* RPC requests to the daemon
123	*/
124	#define RPCIOD_QDEPTH 20
125
126	/* Maximum retry limit for retransmission */
127	#define RPCIOD_RETX_CAP_S 3 /* seconds */
128
129	/* Default timeout for RPC calls */
130	#define RPCIOD_DEFAULT_TIMEOUT (&_rpc_default_timeout)
131	static struct timeval _rpc_default_timeout = { 10 /* secs /, 0 / usecs */ };
132
133	/* how many times should we try to resend a failed
134	* transaction with refreshed AUTHs
135	*/
136	#define RPCIOD_REFRESH 2
137
138	/* Events we are using; the RPC_EVENT
139	* MUST NOT be used by any application
140	* thread doing RPC IO (e.g. NFS)
141	*/
142	#define RTEMS_RPC_EVENT RTEMS_EVENT_30 /* THE event used by RPCIO. Every task doing
143	* RPC IO will receive this - hence it is
144	* RESERVED
145	*/
146	#define RPCIOD_RX_EVENT RTEMS_EVENT_1 /* Events the RPCIOD is using/waiting for */
147	#define RPCIOD_TX_EVENT RTEMS_EVENT_2
148	#define RPCIOD_KILL_EVENT RTEMS_EVENT_3 /* send to the daemon to kill it */
149
150	#define LD_XACT_HASH 8 /* ld of the size of the transaction hash table */
151
152
153	/* Debugging Flags */
154
155	/* NOTE: defining DEBUG 0 leaves some 'assert()' paranoia checks
156	* but produces no output
157	*/
158
159	#define DEBUG_TRACE_XACT (1<<0)
160	#define DEBUG_EVENTS (1<<1)
161	#define DEBUG_MALLOC (1<<2)
162	#define DEBUG_TIMEOUT (1<<3)
163	#define DEBUG_PACKLOSS (1<<4) /* This introduces random, artificial packet losses to test retransmission */
164
165	#define DEBUG_PACKLOSS_FRACT (0xffffffff/10)
166
167	/* USE PARENTHESIS WHEN 'or'ing MULTIPLE FLAGS: (DEBUG_XX \| DEBUG_YY) */
168	#define DEBUG (0)
169
170	/****************************************************************/
171	/* END OF CONFIGURABLE SECTION */
172	/****************************************************************/
173
174	/* prevent rollover of our timers by readjusting the epoch on the fly */
175	#if (DEBUG) & DEBUG_TIMEOUT
176	#define RPCIOD_EPOCH_SECS 10
177	#else
178	#define RPCIOD_EPOCH_SECS 10000
179	#endif
180
181	#ifdef DEBUG
182	#define ASSERT(arg) assert(arg)
183	#else
184	#define ASSERT(arg) if (arg)
185	#endif
186
187	/****************************************************************/
188	/* MACROS */
189	/****************************************************************/
190
191
192	#define XACT_HASHS (1<<(LD_XACT_HASH)) /* the hash table size derived from the ld */
193	#define XACT_HASH_MSK ((XACT_HASHS)-1) /* mask to extract the hash index from a RPC-XID */
194
195
196	#define MU_LOCK(mutex) do { \
197	assert( \
198	RTEMS_SUCCESSFUL == \
199	rtems_semaphore_obtain( \
200	(mutex), \
201	RTEMS_WAIT, \
202	RTEMS_NO_TIMEOUT \
203	) ); \
204	} while(0)
205
206	#define MU_UNLOCK(mutex) do { \
207	assert( \
208	RTEMS_SUCCESSFUL == \
209	rtems_semaphore_release( \
210	(mutex) \
211	) ); \
212	} while(0)
213
214	#define MU_CREAT(pmutex) do { \
215	assert( \
216	RTEMS_SUCCESSFUL == \
217	rtems_semaphore_create( \
218	rtems_build_name( \
219	'R','P','C','l' \
220	), \
221	1, \
222	MUTEX_ATTRIBUTES, \
223	0, \
224	(pmutex)) ); \
225	} while (0)
226
227
228	#define MU_DESTROY(mutex) do { \
229	assert( \
230	RTEMS_SUCCESSFUL == \
231	rtems_semaphore_delete( \
232	mutex \
233	) ); \
234	} while (0)
235
236	#define MUTEX_ATTRIBUTES (RTEMS_LOCAL \| \
237	RTEMS_PRIORITY \| \
238	RTEMS_INHERIT_PRIORITY \| \
239	RTEMS_BINARY_SEMAPHORE)
240
241	#define FIRST_ATTEMPT 0x88888888 /* some time that is never reached */
242
243	/****************************************************************/
244	/* TYPE DEFINITIONS */
245	/****************************************************************/
246
247	typedef rtems_interval TimeoutT;
248
249	/* 100000th implementation of a doubly linked list;
250	* since only one thread is looking at these,
251	* we need no locking
252	*/
253	typedef struct ListNodeRec_ {
254	struct ListNodeRec_ next, prev;
255	} ListNodeRec, *ListNode;
256
257
258	/* Structure representing an RPC server */
259	typedef struct RpcUdpServerRec_ {
260	RpcUdpServer next; /* linked list of all servers; protected by hlock */
261	union {
262	struct sockaddr_in sin;
263	struct sockaddr sa;
264	} addr;
265	AUTH *auth;
266	rtems_id authlock; /* must MUTEX the auth object - it's not clear
267	* what is better:
268	* 1 having one (MUTEXed) auth per server
269	* who is shared among all transactions
270	* using that server
271	* 2 maintaining an AUTH per transaction
272	* (there are then other options: manage
273	* XACT pools on a per-server basis instead
274	* of associating a server with a XACT when
275	* sending)
276	* experience will show if the current (1)
277	* approach has to be changed.
278	*/
279	TimeoutT retry_period; /* dynamically adjusted retry period
280	* (based on packet roundtrip time)
281	*/
282	/* STATISTICS */
283	unsigned long retrans; /* how many retries were issued by this server */
284	unsigned long requests; /* how many requests have been sent */
285	unsigned long timeouts; /* how many requests have timed out */
286	unsigned long errors; /* how many errors have occurred (other than timeouts) */
287	char name[20]; /* server's address in IP 'dot' notation */
288	} RpcUdpServerRec;
289
290	typedef union RpcBufU_ {
291	u_long xid;
292	char buf[1];
293	} RpcBufU, *RpcBuf;
294
295	/* RX Buffer implementation; this is either
296	* an MBUF chain (MBUF_RX configuration)
297	* or a buffer allocated from the heap
298	* where recvfrom copies the (encoded) reply
299	* to. The XDR routines the copy/decode
300	* it into the user's data structures.
301	*/
302	#ifdef MBUF_RX
303	typedef struct mbuf * RxBuf; /* an MBUF chain */
304	static void bufFree(struct mbuf **m);
305	#define XID(ibuf) ((mtod((ibuf), u_long )))
306	extern void xdrmbuf_create(XDR , struct mbuf , enum xdr_op);
307	#else
308	typedef RpcBuf RxBuf;
309	#define bufFree(b) do { MY_FREE((b)); (b)=0; } while(0)
310	#define XID(ibuf) ((ibuf)->xid)
311	#endif
312
313	/* A RPC 'transaction' consisting
314	* of server and requestor information,
315	* buffer space and an XDR object
316	* (for encoding arguments).
317	*/
318	typedef struct RpcUdpXactRec_ {
319	ListNodeRec node; /* so we can put XACTs on a list */
320	RpcUdpServer server; /* server this XACT goes to */
321	long lifetime; /* during the lifetime, retry attempts are made */
322	long tolive; /* lifetime timer */
323	struct rpc_err status; /* RPC reply error status */
324	long age; /* age info; needed to manage retransmission */
325	long trip; /* record round trip time in ticks */
326	rtems_id requestor; /* the task waiting for this XACT to complete */
327	RpcUdpXactPool pool; /* if this XACT belong to a pool, this is it */
328	XDR xdrs; /* argument encoder stream */
329	int xdrpos; /* stream position after the (permanent) header */
330	xdrproc_t xres; /* reply decoder proc - TODO needn't be here */
331	caddr_t pres; /* reply decoded obj - TODO needn't be here */
332	#ifndef MBUF_RX
333	int ibufsize; /* size of the ibuf (bytes) */
334	#endif
335	#ifdef MBUF_TX
336	int refcnt; /* mbuf external storage reference count */
337	#endif
338	int obufsize; /* size of the obuf (bytes) */
339	RxBuf ibuf; /* pointer to input buffer assigned by daemon */
340	RpcBufU obuf; /* output buffer (encoded args) APPENDED HERE */
341	} RpcUdpXactRec;
342
343	typedef struct RpcUdpXactPoolRec_ {
344	rtems_id box;
345	int prog;
346	int version;
347	int xactSize;
348	} RpcUdpXactPoolRec;
349
350	/* a global hash table where all 'living' transaction
351	* objects are registered.
352	* A number of bits in a transaction's XID maps 1:1 to
353	* an index in this table. Hence, the XACT matching
354	* an RPC/UDP reply packet can quickly be found
355	* The size of this table imposes a hard limit on the
356	* number of all created transactions in the system.
357	*/
358	static RpcUdpXact xactHashTbl[XACT_HASHS]={0};
359	static u_long xidUpper [XACT_HASHS]={0};
360	static unsigned xidHashSeed = 0 ;
361
362	/* forward declarations */
363	static RpcUdpXact
364	sockRcv(void);
365
366	static void
367	rpcio_daemon(rtems_task_argument);
368
369	#ifdef MBUF_TX
370	ssize_t
371	sendto_nocpy (
372	int s,
373	const void *buf, size_t buflen,
374	int flags,
375	const struct sockaddr *toaddr, int tolen,
376	void *closure,
377	void (*freeproc)(caddr_t, u_int),
378	void (*refproc)(caddr_t, u_int)
379	);
380	static void paranoia_free(caddr_t closure, u_int size);
381	static void paranoia_ref (caddr_t closure, u_int size);
382	#define SENDTO sendto_nocpy
383	#else
384	#define SENDTO sendto
385	#endif
386
387	static RpcUdpServer rpcUdpServers = 0; /* linked list of all servers; protected by llock */
388
389	static int ourSock = -1; /* the socket we are using for communication */
390	static rtems_id rpciod = 0; /* task id of the RPC daemon */
391	static rtems_id msgQ = 0; /* message queue where the daemon picks up
392	* requests
393	*/
394	static rtems_id llock = 0; /* MUTEX protecting the server list */
395	static rtems_id hlock = 0; /* MUTEX protecting the hash table and the list of servers */
396	static rtems_id fini = 0; /* a synchronization semaphore we use during
397	* module cleanup / driver unloading
398	*/
399	static rtems_interval ticksPerSec; /* cached system clock rate (WHO IS ASSUMED NOT
400	* TO CHANGE)
401	*/
402
403	rtems_task_priority rpciodPriority = 0;
404
405	#if (DEBUG) & DEBUG_MALLOC
406	/* malloc wrappers for debugging */
407	static int nibufs = 0;
408
409	static inline void *MY_MALLOC(int s)
410	{
411	if (s) {
412	void *rval;
413	MU_LOCK(hlock);
414	assert(nibufs++ < 2000);
415	MU_UNLOCK(hlock);
416	assert((rval = malloc(s)) != 0);
417	return rval;
418	}
419	return 0;
420	}
421
422	static inline void *MY_CALLOC(int n, int s)
423	{
424	if (s) {
425	void *rval;
426	MU_LOCK(hlock);
427	assert(nibufs++ < 2000);
428	MU_UNLOCK(hlock);
429	assert((rval = calloc(n,s)) != 0);
430	return rval;
431	}
432	return 0;
433	}
434
435
436	static inline void MY_FREE(void *p)
437	{
438	if (p) {
439	MU_LOCK(hlock);
440	nibufs--;
441	MU_UNLOCK(hlock);
442	free(p);
443	}
444	}
445	#else
446	#define MY_MALLOC malloc
447	#define MY_CALLOC calloc
448	#define MY_FREE free
449	#endif
450
451	static inline bool_t
452	locked_marshal(RpcUdpServer s, XDR *xdrs)
453	{
454	bool_t rval;
455	MU_LOCK(s->authlock);
456	rval = AUTH_MARSHALL(s->auth, xdrs);
457	MU_UNLOCK(s->authlock);
458	return rval;
459	}
460
461	/* Locked operations on a server's auth object */
462	static inline bool_t
463	locked_validate(RpcUdpServer s, struct opaque_auth *v)
464	{
465	bool_t rval;
466	MU_LOCK(s->authlock);
467	rval = AUTH_VALIDATE(s->auth, v);
468	MU_UNLOCK(s->authlock);
469	return rval;
470	}
471
472	static inline bool_t
473	locked_refresh(RpcUdpServer s)
474	{
475	bool_t rval;
476	MU_LOCK(s->authlock);
477	rval = AUTH_REFRESH(s->auth);
478	MU_UNLOCK(s->authlock);
479	return rval;
480	}
481
482	/* Create a server object
483	*
484	*/
485	enum clnt_stat
486	rpcUdpServerCreate(
487	struct sockaddr_in *paddr,
488	int prog,
489	int vers,
490	u_long uid,
491	u_long gid,
492	RpcUdpServer *psrv
493	)
494	{
495	RpcUdpServer rval;
496	u_short port;
497	char hname[MAX_MACHINE_NAME + 1];
498	int theuid, thegid;
499	int thegids[NGRPS];
500	gid_t gids[NGROUPS];
501	int len,i;
502	AUTH *auth;
503	enum clnt_stat pmap_err;
504	struct pmap pmaparg;
505
506	if ( gethostname(hname, MAX_MACHINE_NAME) ) {
507	fprintf(stderr,
508	"RPCIO - error: I have no hostname ?? (%s)\n",
509	strerror(errno));
510	return RPC_UNKNOWNHOST;
511	}
512
513	if ( (len = getgroups(NGROUPS, gids) < 0 ) ) {
514	fprintf(stderr,
515	"RPCIO - error: I unable to get group ids (%s)\n",
516	strerror(errno));
517	return RPC_FAILED;
518	}
519
520	if ( len > NGRPS )
521	len = NGRPS;
522
523	for (i=0; i<len; i++)
524	thegids[i] = (int)gids[i];
525
526	theuid = (int) ((RPCIOD_DEFAULT_ID == uid) ? geteuid() : uid);
527	thegid = (int) ((RPCIOD_DEFAULT_ID == gid) ? getegid() : gid);
528
529	if ( !(auth = authunix_create(hname, theuid, thegid, len, thegids)) ) {
530	fprintf(stderr,
531	"RPCIO - error: unable to create RPC AUTH\n");
532	return RPC_FAILED;
533	}
534
535	/* if they specified no port try to ask the portmapper */
536	if (!paddr->sin_port) {
537
538	paddr->sin_port = htons(PMAPPORT);
539
540	pmaparg.pm_prog = prog;
541	pmaparg.pm_vers = vers;
542	pmaparg.pm_prot = IPPROTO_UDP;
543	pmaparg.pm_port = 0; /* not needed or used */
544
545
546	/* dont use non-reentrant pmap_getport ! */
547
548	pmap_err = rpcUdpCallRp(
549	paddr,
550	PMAPPROG,
551	PMAPVERS,
552	PMAPPROC_GETPORT,
553	xdr_pmap,
554	&pmaparg,
555	xdr_u_short,
556	&port,
557	uid,
558	gid,
559	0);
560
561	if ( RPC_SUCCESS != pmap_err ) {
562	paddr->sin_port = 0;
563	return pmap_err;
564	}
565
566	paddr->sin_port = htons(port);
567	}
568
569	if (0==paddr->sin_port) {
570	return RPC_PROGNOTREGISTERED;
571	}
572
573	rval = (RpcUdpServer)MY_MALLOC(sizeof(*rval));
574	memset(rval, 0, sizeof(*rval));
575
576	if (!inet_ntop(AF_INET, &paddr->sin_addr, rval->name, sizeof(rval->name)))
577	sprintf(rval->name,"?.?.?.?");
578	rval->addr.sin = *paddr;
579
580	/* start with a long retransmission interval - it
581	* will be adapted dynamically
582	*/
583	rval->retry_period = RPCIOD_RETX_CAP_S * ticksPerSec;
584
585	rval->auth = auth;
586
587	MU_CREAT( &rval->authlock );
588
589	/* link into list */
590	MU_LOCK( llock );
591	rval->next = rpcUdpServers;
592	rpcUdpServers = rval;
593	MU_UNLOCK( llock );
594
595	*psrv = rval;
596	return RPC_SUCCESS;
597	}
598
599	void
600	rpcUdpServerDestroy(RpcUdpServer s)
601	{
602	RpcUdpServer prev;
603	if (!s)
604	return;
605	/* we should probably verify (but how?) that nobody
606	* (at least: no outstanding XACTs) is using this
607	* server;
608	*/
609
610	/* remove from server list */
611	MU_LOCK(llock);
612	prev = rpcUdpServers;
613	if ( s == prev ) {
614	rpcUdpServers = s->next;
615	} else {
616	for ( ; prev ; prev = prev->next) {
617	if (prev->next == s) {
618	prev->next = s->next;
619	break;
620	}
621	}
622	}
623	MU_UNLOCK(llock);
624
625	/* MUST have found it */
626	assert(prev);
627
628	auth_destroy(s->auth);
629
630	MU_DESTROY(s->authlock);
631	MY_FREE(s);
632	}
633
634	int
635	rpcUdpStats(FILE *f)
636	{
637	RpcUdpServer s;
638
639	if (!f) f = stdout;
640
641	fprintf(f,"RPCIOD statistics:\n");
642
643	MU_LOCK(llock);
644	for (s = rpcUdpServers; s; s=s->next) {
645	fprintf(f,"\nServer -- %s:\n", s->name);
646	fprintf(f," requests sent: %10ld, retransmitted: %10ld\n",
647	s->requests, s->retrans);
648	fprintf(f," timed out: %10ld, send errors: %10ld\n",
649	s->timeouts, s->errors);
650	fprintf(f," current retransmission interval: %dms\n",
651	(unsigned)(s->retry_period * 1000 / ticksPerSec) );
652	}
653	MU_UNLOCK(llock);
654
655	return 0;
656	}
657
658	RpcUdpXact
659	rpcUdpXactCreate(
660	u_long program,
661	u_long version,
662	u_long size
663	)
664	{
665	RpcUdpXact rval=0;
666	struct rpc_msg header;
667	register int i,j;
668
669	if (!size)
670	size = UDPMSGSIZE;
671	/* word align */
672	size = (size + 3) & ~3;
673
674	rval = (RpcUdpXact)MY_CALLOC(1,sizeof(*rval) - sizeof(rval->obuf) + size);
675
676	if (rval) {
677
678	header.rm_xid = 0;
679	header.rm_direction = CALL;
680	header.rm_call.cb_rpcvers = RPC_MSG_VERSION;
681	header.rm_call.cb_prog = program;
682	header.rm_call.cb_vers = version;
683	xdrmem_create(&(rval->xdrs), rval->obuf.buf, size, XDR_ENCODE);
684
685	if (!xdr_callhdr(&(rval->xdrs), &header)) {
686	MY_FREE(rval);
687	return 0;
688	}
689	/* pick a free table slot and initialize the XID */
690	rval->obuf.xid = time(0) ^ (unsigned long)rval;
691	MU_LOCK(hlock);
692	rval->obuf.xid = (xidHashSeed++ ^ ((unsigned long)rval>>10)) & XACT_HASH_MSK;
693	i=j=(rval->obuf.xid & XACT_HASH_MSK);
694	if (msgQ) {
695	/* if there's no message queue, refuse to
696	* give them transactions; we might be in the process to
697	* go away...
698	*/
699	do {
700	i=(i+1) & XACT_HASH_MSK; /* cheap modulo */
701	if (!xactHashTbl[i]) {
702	#if (DEBUG) & DEBUG_TRACE_XACT
703	fprintf(stderr,"RPCIO: entering index %i, val %x\n",i,rval);
704	#endif
705	xactHashTbl[i]=rval;
706	j=-1;
707	break;
708	}
709	} while (i!=j);
710	}
711	MU_UNLOCK(hlock);
712	if (i==j) {
713	XDR_DESTROY(&rval->xdrs);
714	MY_FREE(rval);
715	return 0;
716	}
717	rval->obuf.xid = xidUpper[i] \| i;
718	rval->xdrpos = XDR_GETPOS(&(rval->xdrs));
719	rval->obufsize = size;
720	}
721	return rval;
722	}
723
724	void
725	rpcUdpXactDestroy(RpcUdpXact xact)
726	{
727	int i = xact->obuf.xid & XACT_HASH_MSK;
728
729	#if (DEBUG) & DEBUG_TRACE_XACT
730	fprintf(stderr,"RPCIO: removing index %i, val %x\n",i,xact);
731	#endif
732
733	ASSERT( xactHashTbl[i]==xact );
734
735	MU_LOCK(hlock);
736	xactHashTbl[i]=0;
737	/* remember XID we used last time so we can avoid
738	* reusing the same one (incremented by rpcUdpSend routine)
739	*/
740	xidUpper[i] = xact->obuf.xid & ~XACT_HASH_MSK;
741	MU_UNLOCK(hlock);
742
743	bufFree(&xact->ibuf);
744
745	XDR_DESTROY(&xact->xdrs);
746	MY_FREE(xact);
747	}
748
749
750
751	/* Send a transaction, i.e. enqueue it to the
752	* RPC daemon who will actually send it.
753	*/
754	enum clnt_stat
755	rpcUdpSend(
756	RpcUdpXact xact,
757	RpcUdpServer srvr,
758	struct timeval *timeout,
759	u_long proc,
760	xdrproc_t xres, caddr_t pres,
761	xdrproc_t xargs, caddr_t pargs,
762	...
763	)
764	{
765	register XDR *xdrs;
766	unsigned long ms;
767	va_list ap;
768
769	va_start(ap,pargs);
770
771	if (!timeout)
772	timeout = RPCIOD_DEFAULT_TIMEOUT;
773
774	ms = 1000 * timeout->tv_sec + timeout->tv_usec/1000;
775
776	/* round lifetime to closest # of ticks */
777	xact->lifetime = (ms * ticksPerSec + 500) / 1000;
778	if ( 0 == xact->lifetime )
779	xact->lifetime = 1;
780
781	#if (DEBUG) & DEBUG_TIMEOUT
782	{
783	static int once=0;
784	if (!once++) {
785	fprintf(stderr,
786	"Initial lifetime: %i (ticks)\n",
787	xact->lifetime);
788	}
789	}
790	#endif
791
792	xact->tolive = xact->lifetime;
793
794	xact->xres = xres;
795	xact->pres = pres;
796	xact->server = srvr;
797
798	xdrs = &xact->xdrs;
799	xdrs->x_op = XDR_ENCODE;
800	/* increment transaction ID */
801	xact->obuf.xid += XACT_HASHS;
802	XDR_SETPOS(xdrs, xact->xdrpos);
803	if ( !XDR_PUTLONG(xdrs,(long*)&proc) \|\| !locked_marshal(srvr, xdrs) \|\|
804	!xargs(xdrs, pargs) ) {
805	va_end(ap);
806	return(xact->status.re_status=RPC_CANTENCODEARGS);
807	}
808	while ((xargs=va_arg(ap,xdrproc_t))) {
809	if (!xargs(xdrs, va_arg(ap,caddr_t)))
810	va_end(ap);
811	return(xact->status.re_status=RPC_CANTENCODEARGS);
812	}
813
814	va_end(ap);
815
816	rtems_task_ident(RTEMS_SELF, RTEMS_WHO_AM_I, &xact->requestor);
817	if ( rtems_message_queue_send( msgQ, &xact, sizeof(xact)) ) {
818	return RPC_CANTSEND;
819	}
820	/* wakeup the rpciod */
821	ASSERT( RTEMS_SUCCESSFUL==rtems_event_send(rpciod, RPCIOD_TX_EVENT) );
822
823	return RPC_SUCCESS;
824	}
825
826	/* Block for the RPC reply to an outstanding
827	* transaction.
828	* The caller is woken by the RPC daemon either
829	* upon reception of the reply or on timeout.
830	*/
831	enum clnt_stat
832	rpcUdpRcv(RpcUdpXact xact)
833	{
834	int refresh;
835	XDR reply_xdrs;
836	struct rpc_msg reply_msg;
837	rtems_status_code status;
838	rtems_event_set gotEvents;
839
840	refresh = 0;
841
842	do {
843
844	/* block for the reply */
845	status = rtems_event_receive(
846	RTEMS_RPC_EVENT,
847	RTEMS_WAIT \| RTEMS_EVENT_ANY,
848	RTEMS_NO_TIMEOUT,
849	&gotEvents);
850	ASSERT( status == RTEMS_SUCCESSFUL );
851
852	if (xact->status.re_status) {
853	#ifdef MBUF_RX
854	/* add paranoia */
855	ASSERT( !xact->ibuf );
856	#endif
857	return xact->status.re_status;
858	}
859
860	#ifdef MBUF_RX
861	xdrmbuf_create(&reply_xdrs, xact->ibuf, XDR_DECODE);
862	#else
863	xdrmem_create(&reply_xdrs, xact->ibuf->buf, xact->ibufsize, XDR_DECODE);
864	#endif
865
866	reply_msg.acpted_rply.ar_verf = _null_auth;
867	reply_msg.acpted_rply.ar_results.where = xact->pres;
868	reply_msg.acpted_rply.ar_results.proc = xact->xres;
869
870	if (xdr_replymsg(&reply_xdrs, &reply_msg)) {
871	/* OK */
872	_seterr_reply(&reply_msg, &xact->status);
873	if (RPC_SUCCESS == xact->status.re_status) {
874	if ( !locked_validate(xact->server,
875	&reply_msg.acpted_rply.ar_verf) ) {
876	xact->status.re_status = RPC_AUTHERROR;
877	xact->status.re_why = AUTH_INVALIDRESP;
878	}
879	if (reply_msg.acpted_rply.ar_verf.oa_base) {
880	reply_xdrs.x_op = XDR_FREE;
881	xdr_opaque_auth(&reply_xdrs, &reply_msg.acpted_rply.ar_verf);
882	}
883	refresh = 0;
884	} else {
885	/* should we try to refresh our credentials ? */
886	if ( !refresh ) {
887	/* had never tried before */
888	refresh = RPCIOD_REFRESH;
889	}
890	}
891	} else {
892	reply_xdrs.x_op = XDR_FREE;
893	xdr_replymsg(&reply_xdrs, &reply_msg);
894	xact->status.re_status = RPC_CANTDECODERES;
895	}
896	XDR_DESTROY(&reply_xdrs);
897
898	bufFree(&xact->ibuf);
899
900	#ifndef MBUF_RX
901	xact->ibufsize = 0;
902	#endif
903
904	if (refresh && locked_refresh(xact->server)) {
905	rtems_task_ident(RTEMS_SELF, RTEMS_WHO_AM_I, &xact->requestor);
906	if ( rtems_message_queue_send(msgQ, &xact, sizeof(xact)) ) {
907	return RPC_CANTSEND;
908	}
909	/* wakeup the rpciod */
910	fprintf(stderr,"RPCIO INFO: refreshing my AUTH\n");
911	ASSERT( RTEMS_SUCCESSFUL==rtems_event_send(rpciod, RPCIOD_TX_EVENT) );
912	}
913
914	} while ( 0 && refresh-- > 0 );
915
916	return xact->status.re_status;
917	}
918
919
920	/* On RTEMS, I'm told to avoid select(); this seems to
921	* be more efficient
922	*/
923	static void
924	rxWakeupCB(struct socket *sock, caddr_t arg)
925	{
926	rtems_event_send((rtems_id)arg, RPCIOD_RX_EVENT);
927	}
928
929	int
930	rpcUdpInit(void)
931	{
932	int s;
933	rtems_status_code status;
934	int noblock = 1;
935	struct sockwakeup wkup;
936
937	if (ourSock < 0) {
938	fprintf(stderr,"RTEMS-RPCIOD $Release$, " \
939	"Till Straumann, Stanford/SLAC/SSRL 2002, " \
940	"See LICENSE file for licensing info.\n");
941
942	ourSock=socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
943	if (ourSock>=0) {
944	bindresvport(ourSock,(struct sockaddr_in*)0);
945	s = ioctl(ourSock, FIONBIO, (char*)&noblock);
946	assert( s == 0 );
947	/* assume nobody tampers with the clock !! */
948	ticksPerSec = rtems_clock_get_ticks_per_second();
949	MU_CREAT( &hlock );
950	MU_CREAT( &llock );
951
952	if ( !rpciodPriority ) {
953	/* use configured networking priority */
954	if ( ! (rpciodPriority = rtems_bsdnet_config.network_task_priority) )
955	rpciodPriority = RPCIOD_PRIO; /* fallback value */
956	}
957
958	status = rtems_task_create(
959	rtems_build_name('R','P','C','d'),
960	rpciodPriority,
961	RPCIOD_STACK,
962	RTEMS_DEFAULT_MODES,
963	/* fprintf saves/restores FP registers on PPC :-( */
964	RTEMS_DEFAULT_ATTRIBUTES \| RTEMS_FLOATING_POINT,
965	&rpciod);
966	assert( status == RTEMS_SUCCESSFUL );
967
968	wkup.sw_pfn = rxWakeupCB;
969	wkup.sw_arg = (caddr_t)rpciod;
970	assert( 0==setsockopt(ourSock, SOL_SOCKET, SO_RCVWAKEUP, &wkup, sizeof(wkup)) );
971	status = rtems_message_queue_create(
972	rtems_build_name('R','P','C','q'),
973	RPCIOD_QDEPTH,
974	sizeof(RpcUdpXact),
975	RTEMS_DEFAULT_ATTRIBUTES,
976	&msgQ);
977	assert( status == RTEMS_SUCCESSFUL );
978	status = rtems_task_start( rpciod, rpcio_daemon, 0 );
979	assert( status == RTEMS_SUCCESSFUL );
980
981	} else {
982	return -1;
983	}
984	}
985	return 0;
986	}
987
988	int
989	rpcUdpCleanup(void)
990	{
991	rtems_semaphore_create(
992	rtems_build_name('R','P','C','f'),
993	0,
994	RTEMS_DEFAULT_ATTRIBUTES,
995	0,
996	&fini);
997	rtems_event_send(rpciod, RPCIOD_KILL_EVENT);
998	/* synchronize with daemon */
999	rtems_semaphore_obtain(fini, RTEMS_WAIT, 5*ticksPerSec);
1000	/* if the message queue is still there, something went wrong */
1001	if (!msgQ) {
1002	rtems_task_delete(rpciod);
1003	}
1004	rtems_semaphore_delete(fini);
1005	return (msgQ !=0);
1006	}
1007
1008	/* Another API - simpler but less efficient.
1009	* For each RPCall, a server and a Xact
1010	* are created and destroyed on the fly.
1011	*
1012	* This should be used for infrequent calls
1013	* (e.g. a NFS mount request).
1014	*
1015	* This is roughly compatible with the original
1016	* clnt_call() etc. API - but it uses our
1017	* daemon and is fully reentrant.
1018	*/
1019	enum clnt_stat
1020	rpcUdpClntCreate(
1021	struct sockaddr_in *psaddr,
1022	int prog,
1023	int vers,
1024	u_long uid,
1025	u_long gid,
1026	RpcUdpClnt *pclnt
1027	)
1028	{
1029	RpcUdpXact x;
1030	RpcUdpServer s;
1031	enum clnt_stat err;
1032
1033	if ( RPC_SUCCESS != (err=rpcUdpServerCreate(psaddr, prog, vers, uid, gid, &s)) )
1034	return err;
1035
1036	if ( !(x=rpcUdpXactCreate(prog, vers, UDPMSGSIZE)) ) {
1037	rpcUdpServerDestroy(s);
1038	return RPC_FAILED;
1039	}
1040	/* TODO: could maintain a server cache */
1041
1042	x->server = s;
1043
1044	*pclnt = x;
1045
1046	return RPC_SUCCESS;
1047	}
1048
1049	void
1050	rpcUdpClntDestroy(RpcUdpClnt xact)
1051	{
1052	rpcUdpServerDestroy(xact->server);
1053	rpcUdpXactDestroy(xact);
1054	}
1055
1056	enum clnt_stat
1057	rpcUdpClntCall(
1058	RpcUdpClnt xact,
1059	u_long proc,
1060	XdrProcT xargs,
1061	CaddrT pargs,
1062	XdrProcT xres,
1063	CaddrT pres,
1064	struct timeval *timeout
1065	)
1066	{
1067	enum clnt_stat stat;
1068
1069	if ( (stat = rpcUdpSend(xact, xact->server, timeout, proc,
1070	xres, pres,
1071	xargs, pargs,
1072	0)) ) {
1073	fprintf(stderr,"RPCIO Send failed: %i\n",stat);
1074	return stat;
1075	}
1076	return rpcUdpRcv(xact);
1077	}
1078
1079	/* a yet simpler interface */
1080	enum clnt_stat
1081	rpcUdpCallRp(
1082	struct sockaddr_in *psrvr,
1083	u_long prog,
1084	u_long vers,
1085	u_long proc,
1086	XdrProcT xargs,
1087	CaddrT pargs,
1088	XdrProcT xres,
1089	CaddrT pres,
1090	u_long uid, /* RPCIO_DEFAULT_ID picks default */
1091	u_long gid, /* RPCIO_DEFAULT_ID picks default */
1092	struct timeval timeout / NULL picks default */
1093	)
1094	{
1095	RpcUdpClnt clp;
1096	enum clnt_stat stat;
1097
1098	stat = rpcUdpClntCreate(
1099	psrvr,
1100	prog,
1101	vers,
1102	uid,
1103	gid,
1104	&clp);
1105
1106	if ( RPC_SUCCESS != stat )
1107	return stat;
1108
1109	stat = rpcUdpClntCall(
1110	clp,
1111	proc,
1112	xargs, pargs,
1113	xres, pres,
1114	timeout);
1115
1116	rpcUdpClntDestroy(clp);
1117
1118	return stat;
1119	}
1120
1121	/* linked list primitives */
1122	static void
1123	nodeXtract(ListNode n)
1124	{
1125	if (n->prev)
1126	n->prev->next = n->next;
1127	if (n->next)
1128	n->next->prev = n->prev;
1129	n->next = n->prev = 0;
1130	}
1131
1132	static void
1133	nodeAppend(ListNode l, ListNode n)
1134	{
1135	if ( (n->next = l->next) )
1136	n->next->prev = n;
1137	l->next = n;
1138	n->prev = l;
1139
1140	}
1141
1142	/* this code does the work */
1143	static void
1144	rpcio_daemon(rtems_task_argument arg)
1145	{
1146	rtems_status_code stat;
1147	RpcUdpXact xact;
1148	RpcUdpServer srv;
1149	rtems_interval next_retrans, then, unow;
1150	long now; /* need to do signed comparison with age! */
1151	rtems_event_set events;
1152	ListNode newList;
1153	size_t size;
1154	rtems_id q = 0;
1155	ListNodeRec listHead = {0, 0};
1156	unsigned long epoch = RPCIOD_EPOCH_SECS * ticksPerSec;
1157	unsigned long max_period = RPCIOD_RETX_CAP_S * ticksPerSec;
1158	rtems_status_code status;
1159
1160
1161	then = rtems_clock_get_ticks_since_boot();
1162
1163	for (next_retrans = epoch;;) {
1164
1165	if ( RTEMS_SUCCESSFUL !=
1166	(stat = rtems_event_receive(
1167	RPCIOD_RX_EVENT \| RPCIOD_TX_EVENT \| RPCIOD_KILL_EVENT,
1168	RTEMS_WAIT \| RTEMS_EVENT_ANY,
1169	next_retrans,
1170	&events)) ) {
1171	ASSERT( RTEMS_TIMEOUT == stat );
1172	events = 0;
1173	}
1174
1175	if (events & RPCIOD_KILL_EVENT) {
1176	int i;
1177
1178	#if (DEBUG) & DEBUG_EVENTS
1179	fprintf(stderr,"RPCIO: got KILL event\n");
1180	#endif
1181
1182	MU_LOCK(hlock);
1183	for (i=XACT_HASHS-1; i>=0; i--) {
1184	if (xactHashTbl[i]) {
1185	break;
1186	}
1187	}
1188	if (i<0) {
1189	/* prevent them from creating and enqueueing more messages */
1190	q=msgQ;
1191	/* messages queued after we executed this assignment will fail */
1192	msgQ=0;
1193	}
1194	MU_UNLOCK(hlock);
1195	if (i>=0) {
1196	fprintf(stderr,"RPCIO There are still transactions circulating; I refuse to go away\n");
1197	fprintf(stderr,"(1st in slot %i)\n",i);
1198	rtems_semaphore_release(fini);
1199	} else {
1200	break;
1201	}
1202	}
1203
1204	unow = rtems_clock_get_ticks_since_boot();
1205
1206	/* measure everything relative to then to protect against
1207	* rollover
1208	*/
1209	now = unow - then;
1210
1211	/* NOTE: we don't lock the hash table while we are operating
1212	* on transactions; the paradigm is that we 'own' a particular
1213	* transaction (and hence it's hash table slot) from the
1214	* time the xact was put into the message queue until we
1215	* wake up the requestor.
1216	*/
1217
1218	if (RPCIOD_RX_EVENT & events) {
1219
1220	#if (DEBUG) & DEBUG_EVENTS
1221	fprintf(stderr,"RPCIO: got RX event\n");
1222	#endif
1223
1224	while ((xact=sockRcv())) {
1225
1226	/* extract from the retransmission list */
1227	nodeXtract(&xact->node);
1228
1229	/* change the ID - there might already be
1230	* a retransmission on the way. When it's
1231	* reply arrives we must not find it's ID
1232	* in the hashtable
1233	*/
1234	xact->obuf.xid += XACT_HASHS;
1235
1236	xact->status.re_status = RPC_SUCCESS;
1237
1238	/* calculate roundtrip ticks */
1239	xact->trip = now - xact->trip;
1240
1241	srv = xact->server;
1242
1243	/* adjust the server's retry period */
1244	{
1245	register TimeoutT rtry = srv->retry_period;
1246	register TimeoutT trip = xact->trip;
1247
1248	ASSERT( trip >= 0 );
1249
1250	if ( 0==trip )
1251	trip = 1;
1252
1253	/* retry_new = 0.75retry_old + 0.25 8 * roundrip */
1254	rtry = (3*rtry + (trip << 3)) >> 2;
1255
1256	if ( rtry > max_period )
1257	rtry = max_period;
1258
1259	srv->retry_period = rtry;
1260	}
1261
1262	/* wakeup requestor */
1263	rtems_event_send(xact->requestor, RTEMS_RPC_EVENT);
1264	}
1265	}
1266
1267	if (RPCIOD_TX_EVENT & events) {
1268
1269	#if (DEBUG) & DEBUG_EVENTS
1270	fprintf(stderr,"RPCIO: got TX event\n");
1271	#endif
1272
1273	while (RTEMS_SUCCESSFUL == rtems_message_queue_receive(
1274	msgQ,
1275	&xact,
1276	&size,
1277	RTEMS_NO_WAIT,
1278	RTEMS_NO_TIMEOUT)) {
1279	/* put to the head of timeout q */
1280	nodeAppend(&listHead, &xact->node);
1281
1282	xact->age = now;
1283	xact->trip = FIRST_ATTEMPT;
1284	}
1285	}
1286
1287
1288	/* work the timeout q */
1289	newList = 0;
1290	for ( xact=(RpcUdpXact)listHead.next;
1291	xact && xact->age <= now;
1292	xact=(RpcUdpXact)listHead.next ) {
1293
1294	/* extract from the list */
1295	nodeXtract(&xact->node);
1296
1297	srv = xact->server;
1298
1299	if (xact->tolive < 0) {
1300	/* this one timed out */
1301	xact->status.re_errno = ETIMEDOUT;
1302	xact->status.re_status = RPC_TIMEDOUT;
1303
1304	srv->timeouts++;
1305
1306	/* Change the ID - there might still be
1307	* a reply on the way. When it arrives we
1308	* must not find it's ID in the hash table
1309	*
1310	* Thanks to Steven Johnson for hunting this
1311	* one down.
1312	*/
1313	xact->obuf.xid += XACT_HASHS;
1314
1315	#if (DEBUG) & DEBUG_TIMEOUT
1316	fprintf(stderr,"RPCIO XACT timed out; waking up requestor\n");
1317	#endif
1318	if ( rtems_event_send(xact->requestor, RTEMS_RPC_EVENT) ) {
1319	rtems_panic("RPCIO PANIC file %s line: %i, requestor id was 0x%08x",
1320	__FILE__,
1321	__LINE__,
1322	xact->requestor);
1323	}
1324
1325	} else {
1326	int len;
1327
1328	len = (int)XDR_GETPOS(&xact->xdrs);
1329
1330	#ifdef MBUF_TX
1331	xact->refcnt = 1; /* sendto itself */
1332	#endif
1333	if ( len != SENDTO( ourSock,
1334	xact->obuf.buf,
1335	len,
1336	0,
1337	&srv->addr.sa,
1338	sizeof(srv->addr.sin)
1339	#ifdef MBUF_TX
1340	, xact,
1341	paranoia_free,
1342	paranoia_ref
1343	#endif
1344	) ) {
1345
1346	xact->status.re_errno = errno;
1347	xact->status.re_status = RPC_CANTSEND;
1348	srv->errors++;
1349
1350	/* wakeup requestor */
1351	fprintf(stderr,"RPCIO: SEND failure\n");
1352	status = rtems_event_send(xact->requestor, RTEMS_RPC_EVENT);
1353	assert( status == RTEMS_SUCCESSFUL );
1354
1355	} else {
1356	/* send successful; calculate retransmission time
1357	* and enqueue to temporary list
1358	*/
1359	if (FIRST_ATTEMPT != xact->trip) {
1360	#if (DEBUG) & DEBUG_TIMEOUT
1361	fprintf(stderr,
1362	"timed out; tolive is %i (ticks), retry period is %i (ticks)\n",
1363	xact->tolive,
1364	srv->retry_period);
1365	#endif
1366	/* this is a real retry; we backup
1367	* the server's retry interval
1368	*/
1369	if ( srv->retry_period < max_period ) {
1370
1371	/* If multiple transactions for this server
1372	* fail (e.g. because it died) this will
1373	* back-off very agressively (doubling
1374	* the retransmission period for every
1375	* timed out transaction up to the CAP limit)
1376	* which is desirable - single packet failure
1377	* is treated more gracefully by this algorithm.
1378	*/
1379
1380	srv->retry_period<<=1;
1381	#if (DEBUG) & DEBUG_TIMEOUT
1382	fprintf(stderr,
1383	"adjusted to; retry period %i\n",
1384	srv->retry_period);
1385	#endif
1386	} else {
1387	/* never wait longer than RPCIOD_RETX_CAP_S seconds */
1388	fprintf(stderr,
1389	"RPCIO: server '%s' not responding - still trying\n",
1390	srv->name);
1391	}
1392	if ( 0 == ++srv->retrans % 1000) {
1393	fprintf(stderr,
1394	"RPCIO - statistics: already %li retries to server %s\n",
1395	srv->retrans,
1396	srv->name);
1397	}
1398	} else {
1399	srv->requests++;
1400	}
1401	xact->trip = now;
1402	{
1403	long capped_period = srv->retry_period;
1404	if ( xact->lifetime < capped_period )
1405	capped_period = xact->lifetime;
1406	xact->age = now + capped_period;
1407	xact->tolive -= capped_period;
1408	}
1409	/* enqueue to the list of newly sent transactions */
1410	xact->node.next = newList;
1411	newList = &xact->node;
1412	#if (DEBUG) & DEBUG_TIMEOUT
1413	fprintf(stderr,
1414	"XACT (0x%08x) age is 0x%x, now: 0x%x\n",
1415	xact,
1416	xact->age,
1417	now);
1418	#endif
1419	}
1420	}
1421	}
1422
1423	/* insert the newly sent transactions into the
1424	* sorted retransmission list
1425	*/
1426	for (; (xact = (RpcUdpXact)newList); ) {
1427	register ListNode p,n;
1428	newList = newList->next;
1429	for ( p=&listHead; (n=p->next) && xact->age > ((RpcUdpXact)n)->age; p=n )
1430	/* nothing else to do */;
1431	nodeAppend(p, &xact->node);
1432	}
1433
1434	if (now > epoch) {
1435	/* every now and then, readjust the epoch */
1436	register ListNode n;
1437	then += now;
1438	for (n=listHead.next; n; n=n->next) {
1439	/* readjust outstanding time intervals subject to the
1440	* condition that the 'absolute' time must remain
1441	* the same. 'age' and 'trip' are measured with
1442	* respect to 'then' - hence:
1443	*
1444	* abs_age == old_age + old_then == new_age + new_then
1445	*
1446	* ==> new_age = old_age + old_then - new_then == old_age - 'now'
1447	*/
1448	((RpcUdpXact)n)->age -= now;
1449	((RpcUdpXact)n)->trip -= now;
1450	#if (DEBUG) & DEBUG_TIMEOUT
1451	fprintf(stderr,
1452	"readjusted XACT (0x%08x); age is 0x%x, trip: 0x%x now: 0x%x\n",
1453	(RpcUdpXact)n,
1454	((RpcUdpXact)n)->trip,
1455	((RpcUdpXact)n)->age,
1456	now);
1457	#endif
1458	}
1459	now = 0;
1460	}
1461
1462	next_retrans = listHead.next ?
1463	((RpcUdpXact)listHead.next)->age - now :
1464	epoch; /* make sure we don't miss updating the epoch */
1465	#if (DEBUG) & DEBUG_TIMEOUT
1466	fprintf(stderr,"RPCIO: next timeout is %x\n",next_retrans);
1467	#endif
1468	}
1469	/* close our socket; shut down the receiver */
1470	close(ourSock);
1471
1472	#if 0 /* if we get here, no transactions exist, hence there can be none
1473	* in the queue whatsoever
1474	*/
1475	/* flush the message queue */
1476	while (RTEMS_SUCCESSFUL == rtems_message_queue_receive(
1477	q,
1478	&xact,
1479	&size,
1480	RTEMS_NO_WAIT,
1481	RTEMS_NO_TIMEOUT)) {
1482	/* TODO enque xact */
1483	}
1484
1485	/* flush all outstanding transactions */
1486
1487	for (xact=((RpcUdpXact)listHead.next); xact; xact=((RpcUdpXact)xact->node.next)) {
1488	xact->status.re_status = RPC_TIMEDOUT;
1489	rtems_event_send(xact->requestor, RTEMS_RPC_EVENT);
1490	}
1491	#endif
1492
1493	rtems_message_queue_delete(q);
1494
1495	MU_DESTROY(hlock);
1496
1497	fprintf(stderr,"RPC daemon exited...\n");
1498
1499	rtems_semaphore_release(fini);
1500	rtems_task_suspend(RTEMS_SELF);
1501	}
1502
1503
1504	/* support for transaction 'pools'. A number of XACT objects
1505	* is always kept around. The initial number is 0 but it
1506	* is allowed to grow up to a maximum.
1507	* If the need grows beyond the maximum, behavior depends:
1508	* Users can either block until a transaction becomes available,
1509	* they can create a new XACT on the fly or get an error
1510	* if no free XACT is available from the pool.
1511	*/
1512
1513	RpcUdpXactPool
1514	rpcUdpXactPoolCreate(
1515	int prog, int version,
1516	int xactsize, int poolsize)
1517	{
1518	RpcUdpXactPool rval = MY_MALLOC(sizeof(*rval));
1519	rtems_status_code status;
1520
1521	ASSERT( rval );
1522	status = rtems_message_queue_create(
1523	rtems_build_name('R','P','C','p'),
1524	poolsize,
1525	sizeof(RpcUdpXact),
1526	RTEMS_DEFAULT_ATTRIBUTES,
1527	&rval->box);
1528	assert( status == RTEMS_SUCCESSFUL );
1529
1530	rval->prog = prog;
1531	rval->version = version;
1532	rval->xactSize = xactsize;
1533	return rval;
1534	}
1535
1536	void
1537	rpcUdpXactPoolDestroy(RpcUdpXactPool pool)
1538	{
1539	RpcUdpXact xact;
1540
1541	while ((xact = rpcUdpXactPoolGet(pool, XactGetFail))) {
1542	rpcUdpXactDestroy(xact);
1543	}
1544	rtems_message_queue_delete(pool->box);
1545	MY_FREE(pool);
1546	}
1547
1548	RpcUdpXact
1549	rpcUdpXactPoolGet(RpcUdpXactPool pool, XactPoolGetMode mode)
1550	{
1551	RpcUdpXact xact = 0;
1552	size_t size;
1553
1554	if (RTEMS_SUCCESSFUL != rtems_message_queue_receive(
1555	pool->box,
1556	&xact,
1557	&size,
1558	XactGetWait == mode ?
1559	RTEMS_WAIT : RTEMS_NO_WAIT,
1560	RTEMS_NO_TIMEOUT)) {
1561
1562	/* nothing found in box; should we create a new one ? */
1563
1564	xact = (XactGetCreate == mode) ?
1565	rpcUdpXactCreate(
1566	pool->prog,
1567	pool->version,
1568	pool->xactSize) : 0 ;
1569	if (xact)
1570	xact->pool = pool;
1571
1572	}
1573	return xact;
1574	}
1575
1576	void
1577	rpcUdpXactPoolPut(RpcUdpXact xact)
1578	{
1579	RpcUdpXactPool pool;
1580
1581	pool = xact->pool;
1582	ASSERT( pool );
1583
1584	if (RTEMS_SUCCESSFUL != rtems_message_queue_send(
1585	pool->box,
1586	&xact,
1587	sizeof(xact)))
1588	rpcUdpXactDestroy(xact);
1589	}
1590
1591	#ifdef MBUF_RX
1592
1593	/* WORKAROUND: include sys/mbuf.h (or other bsdnet headers) only
1594	* _after_ using malloc()/free() & friends because
1595	* the RTEMS/BSDNET headers redefine those :-(
1596	*/
1597
1598	#define _KERNEL
1599	#include <sys/mbuf.h>
1600
1601	ssize_t
1602	recv_mbuf_from(int s, struct mbuf *ppm, long len, struct sockaddr fromaddr, int *fromlen);
1603
1604	static void
1605	bufFree(struct mbuf **m)
1606	{
1607	if (*m) {
1608	rtems_bsdnet_semaphore_obtain();
1609	m_freem(*m);
1610	rtems_bsdnet_semaphore_release();
1611	*m = 0;
1612	}
1613	}
1614	#endif
1615
1616	#ifdef MBUF_TX
1617	static void
1618	paranoia_free(caddr_t closure, u_int size)
1619	{
1620	#if (DEBUG)
1621	RpcUdpXact xact = (RpcUdpXact)closure;
1622	int len = (int)XDR_GETPOS(&xact->xdrs);
1623
1624	ASSERT( --xact->refcnt >= 0 && size == len );
1625	#endif
1626	}
1627
1628	static void
1629	paranoia_ref (caddr_t closure, u_int size)
1630	{
1631	#if (DEBUG)
1632	RpcUdpXact xact = (RpcUdpXact)closure;
1633	int len = (int)XDR_GETPOS(&xact->xdrs);
1634	ASSERT( size == len );
1635	xact->refcnt++;
1636	#endif
1637	}
1638	#endif
1639
1640	/* receive from a socket and find
1641	* the transaction corresponding to the
1642	* transaction ID received in the server
1643	* reply.
1644	*
1645	* The semantics of the 'pibuf' pointer are
1646	* as follows:
1647	*
1648	* MBUF_RX:
1649	*
1650	*/
1651
1652	#define RPCIOD_RXBUFSZ UDPMSGSIZE
1653
1654	static RpcUdpXact
1655	sockRcv(void)
1656	{
1657	int len,i;
1658	u_long xid;
1659	union {
1660	struct sockaddr_in sin;
1661	struct sockaddr sa;
1662	} fromAddr;
1663	int fromLen = sizeof(fromAddr.sin);
1664	RxBuf ibuf = 0;
1665	RpcUdpXact xact = 0;
1666
1667	do {
1668
1669	/* rcv_mbuf() and recvfrom() differ in that the
1670	* former allocates buffers and passes them back
1671	* to us whereas the latter requires us to provide
1672	* buffer space.
1673	* Hence, in the first case whe have to make sure
1674	* no old buffer is leaked - in the second case,
1675	* we might well re-use an old buffer but must
1676	* make sure we have one allocated
1677	*/
1678	#ifdef MBUF_RX
1679	if (ibuf)
1680	bufFree(&ibuf);
1681
1682	len = recv_mbuf_from(
1683	ourSock,
1684	&ibuf,
1685	RPCIOD_RXBUFSZ,
1686	&fromAddr.sa,
1687	&fromLen);
1688	#else
1689	if ( !ibuf )
1690	ibuf = (RpcBuf)MY_MALLOC(RPCIOD_RXBUFSZ);
1691	if ( !ibuf )
1692	goto cleanup; /* no memory - drop this message */
1693
1694	len = recvfrom(ourSock,
1695	ibuf->buf,
1696	RPCIOD_RXBUFSZ,
1697	0,
1698	&fromAddr.sa,
1699	&fromLen);
1700	#endif
1701
1702	if (len <= 0) {
1703	if (EAGAIN != errno)
1704	fprintf(stderr,"RECV failed: %s\n",strerror(errno));
1705	goto cleanup;
1706	}
1707
1708	#if (DEBUG) & DEBUG_PACKLOSS
1709	if ( (unsigned)rand() < DEBUG_PACKLOSS_FRACT ) {
1710	/* lose packets once in a while */
1711	static int xxx = 0;
1712	if ( ++xxx % 16 == 0 )
1713	fprintf(stderr,"DEBUG: dropped %i packets, so far...\n",xxx);
1714	if ( ibuf )
1715	bufFree( &ibuf );
1716	continue;
1717	}
1718	#endif
1719
1720	i = (xid=XID(ibuf)) & XACT_HASH_MSK;
1721
1722	if ( !(xact=xactHashTbl[i]) \|\|
1723	xact->obuf.xid != xid \|\|
1724	#ifdef REJECT_SERVERIP_MISMATCH
1725	xact->server->addr.sin.sin_addr.s_addr != fromAddr.sin.sin_addr.s_addr \|\|
1726	#endif
1727	xact->server->addr.sin.sin_port != fromAddr.sin.sin_port ) {
1728
1729	if (xact) {
1730	if (
1731	#ifdef REJECT_SERVERIP_MISMATCH
1732	xact->server->addr.sin.sin_addr.s_addr == fromAddr.sin.sin_addr.s_addr &&
1733	#endif
1734	xact->server->addr.sin.sin_port == fromAddr.sin.sin_port &&
1735	( xact->obuf.xid == xid + XACT_HASHS \|\|
1736	xact->obuf.xid == xid + 2*XACT_HASHS )
1737	) {
1738	#ifndef DEBUG /* don't complain if it's just a late arrival of a retry */
1739	fprintf(stderr,"RPCIO - FYI sockRcv(): dropping late/redundant retry answer\n");
1740	#endif
1741	} else {
1742	fprintf(stderr,"RPCIO WARNING sockRcv(): transaction mismatch\n");
1743	fprintf(stderr,"xact: xid 0x%08lx -- got 0x%08lx\n",
1744	xact->obuf.xid, xid);
1745	fprintf(stderr,"xact: addr 0x%08lx -- got 0x%08lx\n",
1746	xact->server->addr.sin.sin_addr.s_addr,
1747	fromAddr.sin.sin_addr.s_addr);
1748	fprintf(stderr,"xact: port 0x%08x -- got 0x%08x\n",
1749	xact->server->addr.sin.sin_port,
1750	fromAddr.sin.sin_port);
1751	}
1752	} else {
1753	fprintf(stderr,
1754	"RPCIO WARNING sockRcv(): got xid 0x%08lx but its slot is empty\n",
1755	xid);
1756	}
1757	/* forget about this one and try again */
1758	xact = 0;
1759	}
1760
1761	} while ( !xact );
1762
1763	xact->ibuf = ibuf;
1764	#ifndef MBUF_RX
1765	xact->ibufsize = RPCIOD_RXBUFSZ;
1766	#endif
1767
1768	return xact;
1769
1770	cleanup:
1771
1772	bufFree(&ibuf);
1773
1774	return 0;
1775	}
1776
1777
1778	#include <rtems/rtems_bsdnet_internal.h>
1779	/* double check the event configuration; should probably globally
1780	* manage system events!!
1781	* We do this at the end of the file for the same reason we had
1782	* included mbuf.h only a couple of lines above - see comment up
1783	* there...
1784	*/
1785	#if RTEMS_RPC_EVENT & SOSLEEP_EVENT & SBWAIT_EVENT & NETISR_EVENTS
1786	#error ILLEGAL EVENT CONFIGURATION
1787	#endif

Note: See TracBrowser for help on using the repository browser.

Download in other formats: