Context Navigation

source: rtems-libbsd/rtemsbsd/nfsclient/rpcio.c @ 58c4e1c5

55-freebsd-126-freebsd-12freebsd-9.3

Last change on this file since 58c4e1c5 was 58c4e1c5, checked in by Sebastian Huber <sebastian.huber@…>, on 06/10/16 at 12:11:40
nfsclient: Port to LibBSD
Property mode set to `100644`
File size: 44.3 KB

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

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

Download in other formats: