2 * Copyright 2000, International Business Machines Corporation and others.
5 * This software has been released under the terms of the IBM Public
6 * License. For details, see the LICENSE file in the top-level source
7 * directory or online at http://www.openafs.org/dl/license10.html
10 /* rx_user.c contains routines specific to the user space UNIX implementation of rx */
12 /* This controls the size of an fd_set; it must be defined early before
13 * the system headers define that type and the macros that operate on it.
14 * Its value should be as large as the maximum file descriptor limit we
15 * are likely to run into on any platform. Right now, that is 65536
16 * which is the default hard fd limit on Solaris 9 */
18 #define FD_SETSIZE 65536
21 #include <afsconfig.h>
22 #include <afs/param.h>
26 #ifdef HAVE_SYS_FILE_H
27 # include <sys/file.h>
30 #ifndef AFS_PTHREAD_ENV
35 #include "rx_atomic.h"
36 #include "rx_globals.h"
37 #include "rx_internal.h"
40 #include "rx_xmit_nt.h"
43 #define MAXTHREADNAMELENGTH 64
45 int debugSelectFailure
; /* # of times select failed */
48 * Sleep on the unique wait channel provided.
53 LWP_WaitProcess(addr
);
57 * Wakeup any threads on the channel provided.
58 * They may be woken up spuriously, and must check any conditions.
61 rxi_Wakeup(void *addr
)
63 LWP_NoYieldSignal(addr
);
66 PROCESS rx_listenerPid
= 0; /* LWP process id of socket listener process */
67 static void* rx_ListenerProc(void *dummy
);
70 * Delay the current thread the specified number of seconds.
78 static int quitListening
= 0;
80 /* This routine will kill the listener thread, if it exists. */
82 rxi_StopListener(void)
85 rxi_ReScheduleEvents();
88 /* This routine will get called by the event package whenever a new,
89 earlier than others, event is posted. If the Listener process
90 is blocked in selects, this will unblock it. It also can be called
91 to force a new trip through the rxi_Listener select loop when the set
92 of file descriptors it should be listening to changes... */
94 rxi_ReScheduleEvents(void)
97 IOMGR_Cancel(rx_listenerPid
);
101 rxi_InitializeThreadSupport(void)
105 LWP_InitializeProcessSupport(LWP_NORMAL_PRIORITY
, &junk
);
107 FD_ZERO(&rx_selectMask
);
111 rxi_StartServerProc(void *(*proc
) (void *), int stacksize
)
114 static int number
= 0;
117 sprintf(name
, "srv_%d", ++number
);
118 LWP_CreateProcess(proc
, stacksize
, RX_PROCESS_PRIORITY
, NULL
,
119 "rx_ServerProc", &scratchPid
);
121 (*registerProgram
) (scratchPid
, name
);
125 rxi_StartListener(void)
127 /* Priority of listener should be high, so it can keep conns alive */
128 #define RX_LIST_STACK 24000
129 LWP_CreateProcess(rx_ListenerProc
, RX_LIST_STACK
, LWP_MAX_PRIORITY
,
130 NULL
, "rx_Listener", &rx_listenerPid
);
132 (*registerProgram
) (rx_listenerPid
, "listener");
135 /* The main loop which listens to the net for datagrams, and handles timeouts
136 and retransmissions, etc. It also is responsible for scheduling the
137 execution of pending events (in conjunction with event.c).
139 Note interaction of nextPollTime and lastPollWorked. The idea is that if rx is not
140 keeping up with the incoming stream of packets (because there are threads that
141 are interfering with its running sufficiently often), rx does a polling select
142 before doing a real IOMGR_Select system call. Doing a real select means that
143 we don't have to let other processes run before processing more packets.
145 So, our algorithm is that if the last poll on the file descriptor found useful data, or
146 we're at the time nextPollTime (which is advanced so that it occurs every 3 or 4 seconds),
147 then we try the polling select before the IOMGR_Select. If we eventually catch up
148 (which we can tell by the polling select returning no input packets ready), then we
149 don't do a polling select again until several seconds later (via nextPollTime mechanism).
153 rxi_ListenerProc(fd_set
* rfds
, int *tnop
, struct rx_call
**newcallp
)
157 struct rx_packet
*p
= (struct rx_packet
*)0;
160 afs_int32 nextPollTime
; /* time to next poll FD before sleeping */
161 int lastPollWorked
, doingPoll
; /* true iff last poll was useful */
162 struct timeval tv
, *tvp
;
168 char name
[MAXTHREADNAMELENGTH
] = "srv_0";
173 code
= LWP_CurrentProcess(&pid
);
175 osi_Panic("rxi_Listener: Can't get my pid.\n");
177 rx_listenerPid
= pid
;
179 (*swapNameProgram
) (pid
, "listener", &name
[0]);
182 /* See if a check for additional packets was issued */
185 /* Grab a new packet only if necessary (otherwise re-use the old one) */
187 rxi_RestoreDataBufs(p
);
189 if (!(p
= rxi_AllocPacket(RX_PACKET_CLASS_RECEIVE
)))
190 osi_Panic("rxi_ListenerProc: no packets!"); /* Shouldn't happen */
192 /* Wait for the next event time or a packet to arrive. */
193 /* event_RaiseEvents schedules any events whose time has come and
194 * then atomically computes the time to the next event, guaranteeing
195 * that this is positive. If there is no next event, it returns 0 */
197 if (!rxevent_RaiseEvents(&cv
))
200 /* It's important to copy cv to tv, because the 4.3 documentation
201 * for select threatens that *tv may be updated after a select, in
202 * future editions of the system, to indicate how much of the time
203 * period has elapsed. So we shouldn't rely on tv not being altered. */
204 tv
.tv_sec
= cv
.sec
; /* Time to next event */
205 tv
.tv_usec
= cv
.usec
;
209 rx_atomic_inc(&rx_stats
.selects
);
211 *rfds
= rx_selectMask
;
213 if (lastPollWorked
|| nextPollTime
< clock_Sec()) {
214 /* we're catching up, or haven't tried to for a few seconds */
216 nextPollTime
= clock_Sec() + 4; /* try again in 4 seconds no matter what */
217 tv
.tv_sec
= tv
.tv_usec
= 0; /* make sure we poll */
219 code
= select((int)(rx_maxSocketNumber
+ 1), rfds
, 0, 0, tvp
);
222 code
= IOMGR_Select((int)(rx_maxSocketNumber
+ 1), rfds
, 0, 0, tvp
);
224 lastPollWorked
= 0; /* default is that it didn't find anything */
228 LWP_DestroyProcess(pid
);
234 * If it was a timer interrupt then we can assume that
235 * the time has advanced by roughly the value of the
236 * previous timeout, and that there is now at least
242 /* select or IOMGR_Select returned failure */
243 debugSelectFailure
++; /* update debugging counter */
248 * IOMGR_Cancel is invoked whenever a new event is
249 * posted that is earlier than any existing events.
250 * So we re-evaluate the time, and then go back to
257 /* Packets have arrived, presumably:
258 * If it wasn't a timer interrupt, then no event should have
259 * timed out yet (well some event may have, but only just...), so
260 * we don't bother looking to see if any have timed out, but just
261 * go directly to reading the data packets
267 for (i
= 0; p
&& i
< rfds
->fd_count
; i
++) {
268 socket
= rfds
->fd_array
[i
];
269 if (rxi_ReadPacket(socket
, p
, &host
, &port
)) {
271 p
= rxi_ReceivePacket(p
, socket
, host
, port
, tnop
,
273 if (newcallp
&& *newcallp
) {
277 if (swapNameProgram
) {
278 (*swapNameProgram
) (rx_listenerPid
, name
, 0);
286 for (socket
= rx_minSocketNumber
;
287 p
&& socket
<= rx_maxSocketNumber
; socket
++) {
288 if (!FD_ISSET(socket
, rfds
))
290 if (rxi_ReadPacket(socket
, p
, &host
, &port
)) {
291 p
= rxi_ReceivePacket(p
, socket
, host
, port
, tnop
,
293 if (newcallp
&& *newcallp
) {
297 if (swapNameProgram
) {
298 (*swapNameProgram
) (rx_listenerPid
, name
, 0);
312 /* This is the listener process request loop. The listener process loop
313 * becomes a server thread when rxi_ListenerProc returns, and stays
314 * server thread until rxi_ServerProc returns. */
316 rx_ListenerProc(void *dummy
)
320 struct rx_call
*newcall
;
323 if (!(rfds
= IOMGR_AllocFDSet())) {
324 osi_Panic("rx_ListenerProc: no fd_sets!\n");
330 rxi_ListenerProc(rfds
, &threadID
, &newcall
);
331 /* osi_Assert(threadID != -1); */
332 /* osi_Assert(newcall != NULL); */
333 sock
= OSI_NULLSOCKET
;
334 rxi_ServerProc(threadID
, newcall
, &sock
);
335 /* osi_Assert(sock != OSI_NULLSOCKET); */
337 AFS_UNREACHED(return(NULL
));
340 /* This is the server process request loop. The server process loop
341 * becomes a listener thread when rxi_ServerProc returns, and stays
342 * listener thread until rxi_ListenerProc returns. */
344 rx_ServerProc(void * unused
)
348 struct rx_call
*newcall
= NULL
;
351 if (!(rfds
= IOMGR_AllocFDSet())) {
352 osi_Panic("rxi_ListenerProc: no fd_sets!\n");
355 rxi_MorePackets(rx_maxReceiveWindow
+ 2); /* alloc more packets */
356 rxi_dataQuota
+= rx_initSendWindow
; /* Reserve some pkts for hard times */
357 /* threadID is used for making decisions in GetCall. Get it by bumping
358 * number of threads handling incoming calls */
359 threadID
= rxi_availProcs
++;
362 sock
= OSI_NULLSOCKET
;
363 rxi_ServerProc(threadID
, newcall
, &sock
);
364 /* osi_Assert(sock != OSI_NULLSOCKET); */
366 rxi_ListenerProc(rfds
, &threadID
, &newcall
);
367 /* osi_Assert(threadID != -1); */
368 /* osi_Assert(newcall != NULL); */
370 AFS_UNREACHED(return(NULL
));
374 * Called from GetUDPSocket.
375 * Called from a single thread at startup.
376 * Returns 0 on success; -1 on failure.
379 rxi_Listen(osi_socket sock
)
383 * Put the socket into non-blocking mode so that rx_Listener
384 * can do a polling read before entering select
386 if (fcntl(sock
, F_SETFL
, FNDELAY
) == -1) {
388 (osi_Msg
"rxi_Listen: unable to set non-blocking mode on socket\n");
392 if (sock
> FD_SETSIZE
- 1) {
393 (osi_Msg
"rxi_Listen: socket descriptor > (FD_SETSIZE-1) = %d\n",
399 FD_SET(sock
, &rx_selectMask
);
400 if (sock
> rx_maxSocketNumber
)
401 rx_maxSocketNumber
= sock
;
402 if (sock
< rx_minSocketNumber
)
403 rx_minSocketNumber
= sock
;
411 rxi_Recvmsg(osi_socket socket
, struct msghdr
*msg_p
, int flags
)
414 code
= recvmsg(socket
, msg_p
, flags
);
416 #ifdef AFS_RXERRQ_ENV
418 while((rxi_HandleSocketError(socket
)) > 0)
427 * Simulate a blocking sendmsg on the non-blocking socket.
428 * It's non blocking because it was set that way for recvmsg.
431 rxi_Sendmsg(osi_socket socket
, struct msghdr
*msg_p
, int flags
)
433 fd_set
*sfds
= (fd_set
*) 0;
434 while (sendmsg(socket
, msg_p
, flags
) == -1) {
438 err
= WSAGetLastError();
444 rx_atomic_inc(&rx_stats
.sendSelects
);
447 if (!(sfds
= IOMGR_AllocFDSet())) {
448 (osi_Msg
"rx failed to alloc fd_set: ");
449 perror("rx_sendmsg");
452 FD_SET(socket
, sfds
);
454 #ifdef AFS_RXERRQ_ENV
455 while((rxi_HandleSocketError(socket
)) > 0)
460 #elif defined(AFS_LINUX22_ENV)
461 /* linux unfortunately returns ECONNREFUSED if the target port
462 * is no longer in use */
463 /* and EAGAIN if a UDP checksum is incorrect */
464 if (err
!= EWOULDBLOCK
&& err
!= ENOBUFS
&& err
!= ECONNREFUSED
467 if (err
!= EWOULDBLOCK
&& err
!= ENOBUFS
)
470 (osi_Msg
"rx failed to send packet: ");
471 perror("rx_sendmsg");
476 while ((err
= select(
482 0, sfds
, 0, 0)) != 1) {
483 if (err
>= 0 || errno
!= EINTR
)
484 osi_Panic("rxi_sendmsg: select error %d.%d", err
, errno
);
486 FD_SET(socket
, sfds
);
490 IOMGR_FreeFDSet(sfds
);