Context Navigation

rpcio.c @ 4a3db517

4.115

Last change on this file since 4a3db517 was e4d8513, checked in by Daniel Cederman <cederman@…>, on 11/14/14 at 07:58:00

nfs: Add RPCd task affinity config option

Similar to the task priority option, the new CPU affinity
option is first controlled by the RPCI specific rpciodCpuset
option. If that is not set, it uses the global network task config.
If that is also not set, it falls back to not setting the affinity
at all, using all CPUs.

Property mode set to 100644

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

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

Context Navigation

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

Download in other formats: