Context Navigation

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

4.104.115

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