rx: Make statistics interface use Atomics
[openafs.git] / src / rx / rx_lwp.c
1 /*
2  * Copyright 2000, International Business Machines Corporation and others.
3  * All Rights Reserved.
4  *
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
8  */
9
10 /* rx_user.c contains routines specific to the user space UNIX implementation of rx */
11
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 */
17 #ifndef _WIN32
18 #define FD_SETSIZE 65536
19 #endif
20
21 #include <afsconfig.h>
22 #include <afs/param.h>
23
24
25 # include <sys/types.h>         /* fd_set on older platforms */
26 # include <errno.h>
27 # include <signal.h>
28 #ifdef AFS_NT40_ENV
29 # include <winsock2.h>
30 #else
31 # include <unistd.h>            /* select() prototype */
32 # include <sys/time.h>          /* struct timeval, select() prototype */
33 # ifndef FD_SET
34 #  include <sys/select.h>       /* fd_set on newer platforms */
35 # endif
36 # include <sys/socket.h>
37 # include <sys/file.h>
38 # include <netdb.h>
39 # include <sys/stat.h>
40 # include <netinet/in.h>
41 # include <net/if.h>
42 # include <sys/ioctl.h>
43 # include <sys/time.h>
44 #endif
45 # include <assert.h>
46 # include "rx.h"
47 # include "rx_atomic.h"
48 # include "rx_globals.h"
49 # include "rx_stats.h"
50 # include <lwp.h>
51
52 #define MAXTHREADNAMELENGTH 64
53
54 int debugSelectFailure;         /* # of times select failed */
55
56 /*
57  * Sleep on the unique wait channel provided.
58  */
59 void
60 rxi_Sleep(void *addr)
61 {
62     LWP_WaitProcess(addr);
63 }
64
65 /*
66  * Wakeup any threads on the channel provided.
67  * They may be woken up spuriously, and must check any conditions.
68  */
69 void
70 rxi_Wakeup(void *addr)
71 {
72     LWP_NoYieldSignal(addr);
73 }
74
75 PROCESS rx_listenerPid = 0;     /* LWP process id of socket listener process */
76 static void* rx_ListenerProc(void *dummy);
77
78 /*
79  * Delay the current thread the specified number of seconds.
80  */
81 void
82 rxi_Delay(int sec)
83 {
84     IOMGR_Sleep(sec);
85 }
86
87 static int quitListening = 0;
88
89 /* This routine will kill the listener thread, if it exists. */
90 void
91 rxi_StopListener(void)
92 {
93     quitListening = 1;
94     rxi_ReScheduleEvents();
95 }
96
97 /* This routine will get called by the event package whenever a new,
98    earlier than others, event is posted.  If the Listener process
99    is blocked in selects, this will unblock it.  It also can be called
100    to force a new trip through the rxi_Listener select loop when the set
101    of file descriptors it should be listening to changes... */
102 void
103 rxi_ReScheduleEvents(void)
104 {
105     if (rx_listenerPid)
106         IOMGR_Cancel(rx_listenerPid);
107 }
108
109 void
110 rxi_InitializeThreadSupport(void)
111 {
112     PROCESS junk;
113
114     LWP_InitializeProcessSupport(LWP_NORMAL_PRIORITY, &junk);
115     IOMGR_Initialize();
116     FD_ZERO(&rx_selectMask);
117 }
118
119 void
120 rxi_StartServerProc(void *(*proc) (void *), int stacksize)
121 {
122     PROCESS scratchPid;
123     static int number = 0;
124     char name[32];
125
126     sprintf(name, "srv_%d", ++number);
127     LWP_CreateProcess(proc, stacksize, RX_PROCESS_PRIORITY, NULL,
128                       "rx_ServerProc", &scratchPid);
129     if (registerProgram)
130         (*registerProgram) (scratchPid, name);
131 }
132
133 void
134 rxi_StartListener(void)
135 {
136     /* Priority of listener should be high, so it can keep conns alive */
137 #define RX_LIST_STACK   24000
138     LWP_CreateProcess(rx_ListenerProc, RX_LIST_STACK, LWP_MAX_PRIORITY,
139                       NULL, "rx_Listener", &rx_listenerPid);
140     if (registerProgram)
141         (*registerProgram) (rx_listenerPid, "listener");
142 }
143
144 /* The main loop which listens to the net for datagrams, and handles timeouts
145    and retransmissions, etc.  It also is responsible for scheduling the
146    execution of pending events (in conjunction with event.c).
147
148    Note interaction of nextPollTime and lastPollWorked.  The idea is that if rx is not
149    keeping up with the incoming stream of packets (because there are threads that
150    are interfering with its running sufficiently often), rx does a polling select
151    before doing a real IOMGR_Select system call.  Doing a real select means that
152    we don't have to let other processes run before processing more packets.
153
154    So, our algorithm is that if the last poll on the file descriptor found useful data, or
155    we're at the time nextPollTime (which is advanced so that it occurs every 3 or 4 seconds),
156    then we try the polling select before the IOMGR_Select.  If we eventually catch up
157    (which we can tell by the polling select returning no input packets ready), then we
158    don't do a polling select again until several seconds later (via nextPollTime mechanism).
159    */
160
161 static void
162 rxi_ListenerProc(fd_set * rfds, int *tnop, struct rx_call **newcallp)
163 {
164     afs_uint32 host;
165     u_short port;
166     struct rx_packet *p = (struct rx_packet *)0;
167     osi_socket socket;
168     struct clock cv;
169     afs_int32 nextPollTime;     /* time to next poll FD before sleeping */
170     int lastPollWorked, doingPoll;      /* true iff last poll was useful */
171     struct timeval tv, *tvp;
172     int code;
173 #ifdef AFS_NT40_ENV
174     int i;
175 #endif
176     PROCESS pid;
177     char name[MAXTHREADNAMELENGTH] = "srv_0";
178
179     clock_NewTime();
180     lastPollWorked = 0;
181     nextPollTime = 0;
182     code = LWP_CurrentProcess(&pid);
183     if (code) {
184         fprintf(stderr, "rxi_Listener: Can't get my pid.\n");
185         assert(0);
186     }
187     rx_listenerPid = pid;
188     if (swapNameProgram)
189         (*swapNameProgram) (pid, "listener", &name[0]);
190
191     for (;;) {
192         /* See if a check for additional packets was issued */
193         rx_CheckPackets();
194
195         /* Grab a new packet only if necessary (otherwise re-use the old one) */
196         if (p) {
197             rxi_RestoreDataBufs(p);
198         } else {
199             if (!(p = rxi_AllocPacket(RX_PACKET_CLASS_RECEIVE)))
200                 osi_Panic("rxi_ListenerProc: no packets!");     /* Shouldn't happen */
201         }
202         /* Wait for the next event time or a packet to arrive. */
203         /* event_RaiseEvents schedules any events whose time has come and
204          * then atomically computes the time to the next event, guaranteeing
205          * that this is positive.  If there is no next event, it returns 0 */
206         clock_NewTime();
207         if (!rxevent_RaiseEvents(&cv))
208             tvp = NULL;
209         else {
210             /* It's important to copy cv to tv, because the 4.3 documentation
211              * for select threatens that *tv may be updated after a select, in
212              * future editions of the system, to indicate how much of the time
213              * period has elapsed.  So we shouldn't rely on tv not being altered. */
214             tv.tv_sec = cv.sec; /* Time to next event */
215             tv.tv_usec = cv.usec;
216             tvp = &tv;
217         }
218         if (rx_stats_active)
219             rx_atomic_inc(&rx_stats.selects);
220
221         *rfds = rx_selectMask;
222
223         if (lastPollWorked || nextPollTime < clock_Sec()) {
224             /* we're catching up, or haven't tried to for a few seconds */
225             doingPoll = 1;
226             nextPollTime = clock_Sec() + 4;     /* try again in 4 seconds no matter what */
227             tv.tv_sec = tv.tv_usec = 0; /* make sure we poll */
228             tvp = &tv;
229             code = select((int)(rx_maxSocketNumber + 1), rfds, 0, 0, tvp);
230         } else {
231             doingPoll = 0;
232             code = IOMGR_Select((int)(rx_maxSocketNumber + 1), rfds, 0, 0, tvp);
233         }
234         lastPollWorked = 0;     /* default is that it didn't find anything */
235
236         if (quitListening) {
237             quitListening = 0;
238             LWP_DestroyProcess(pid);
239         }
240
241         switch (code) {
242         case 0:
243             /* Timer interrupt:
244              * If it was a timer interrupt then we can assume that
245              * the time has advanced by roughly the value of the
246              * previous timeout, and that there is now at least
247              * one pending event.
248              */
249             clock_NewTime();
250             break;
251         case -1:
252             /* select or IOMGR_Select returned failure */
253             debugSelectFailure++;       /* update debugging counter */
254             clock_NewTime();
255             break;
256         case -2:
257             /* IOMGR_Cancel:
258              * IOMGR_Cancel is invoked whenever a new event is
259              * posted that is earlier than any existing events.
260              * So we re-evaluate the time, and then go back to
261              * reschedule events
262              */
263             clock_NewTime();
264             break;
265
266         default:
267             /* Packets have arrived, presumably:
268              * If it wasn't a timer interrupt, then no event should have
269              * timed out yet (well some event may have, but only just...), so
270              * we don't bother looking to see if any have timed out, but just
271              * go directly to reading the data packets
272              */
273             clock_NewTime();
274             if (doingPoll)
275                 lastPollWorked = 1;
276 #ifdef AFS_NT40_ENV
277             for (i = 0; p && i < rfds->fd_count; i++) {
278                 socket = rfds->fd_array[i];
279                 if (rxi_ReadPacket(socket, p, &host, &port)) {
280                     *newcallp = NULL;
281                     p = rxi_ReceivePacket(p, socket, host, port, tnop,
282                                           newcallp);
283                     if (newcallp && *newcallp) {
284                         if (p) {
285                             rxi_FreePacket(p);
286                         }
287                         if (swapNameProgram) {
288                             (*swapNameProgram) (rx_listenerPid, name, 0);
289                             rx_listenerPid = 0;
290                         }
291                         return;
292                     }
293                 }
294             }
295 #else
296             for (socket = rx_minSocketNumber;
297                  p && socket <= rx_maxSocketNumber; socket++) {
298                 if (!FD_ISSET(socket, rfds))
299                     continue;
300                 if (rxi_ReadPacket(socket, p, &host, &port)) {
301                     p = rxi_ReceivePacket(p, socket, host, port, tnop,
302                                           newcallp);
303                     if (newcallp && *newcallp) {
304                         if (p) {
305                             rxi_FreePacket(p);
306                         }
307                         if (swapNameProgram) {
308                             (*swapNameProgram) (rx_listenerPid, name, 0);
309                             rx_listenerPid = 0;
310                         }
311                         return;
312                     }
313                 }
314             }
315 #endif
316             break;
317         }
318     }
319     /* NOTREACHED */
320 }
321
322 /* This is the listener process request loop. The listener process loop
323  * becomes a server thread when rxi_ListenerProc returns, and stays
324  * server thread until rxi_ServerProc returns. */
325 static void *
326 rx_ListenerProc(void *dummy)
327 {
328     int threadID;
329     osi_socket sock;
330     struct rx_call *newcall;
331     fd_set *rfds;
332
333     if (!(rfds = IOMGR_AllocFDSet())) {
334         osi_Panic("rx_ListenerProc: no fd_sets!\n");
335     }
336
337     while (1) {
338         newcall = NULL;
339         threadID = -1;
340         rxi_ListenerProc(rfds, &threadID, &newcall);
341         /* assert(threadID != -1); */
342         /* assert(newcall != NULL); */
343         sock = OSI_NULLSOCKET;
344         rxi_ServerProc(threadID, newcall, &sock);
345         /* assert(sock != OSI_NULLSOCKET); */
346     }
347     /* not reached */
348     return NULL;
349 }
350
351 /* This is the server process request loop. The server process loop
352  * becomes a listener thread when rxi_ServerProc returns, and stays
353  * listener thread until rxi_ListenerProc returns. */
354 void *
355 rx_ServerProc(void * unused)
356 {
357     osi_socket sock;
358     int threadID;
359     struct rx_call *newcall = NULL;
360     fd_set *rfds;
361
362     if (!(rfds = IOMGR_AllocFDSet())) {
363         osi_Panic("rxi_ListenerProc: no fd_sets!\n");
364     }
365
366     rxi_MorePackets(rx_maxReceiveWindow + 2);   /* alloc more packets */
367     rxi_dataQuota += rx_initSendWindow; /* Reserve some pkts for hard times */
368     /* threadID is used for making decisions in GetCall.  Get it by bumping
369      * number of threads handling incoming calls */
370     threadID = rxi_availProcs++;
371
372     while (1) {
373         sock = OSI_NULLSOCKET;
374         rxi_ServerProc(threadID, newcall, &sock);
375         /* assert(sock != OSI_NULLSOCKET); */
376         newcall = NULL;
377         rxi_ListenerProc(rfds, &threadID, &newcall);
378         /* assert(threadID != -1); */
379         /* assert(newcall != NULL); */
380     }
381     /* not reached */
382     return NULL;
383 }
384
385 /*
386  * Called from GetUDPSocket.
387  * Called from a single thread at startup.
388  * Returns 0 on success; -1 on failure.
389  */
390 int
391 rxi_Listen(osi_socket sock)
392 {
393 #ifndef AFS_NT40_ENV
394     /*
395      * Put the socket into non-blocking mode so that rx_Listener
396      * can do a polling read before entering select
397      */
398     if (fcntl(sock, F_SETFL, FNDELAY) == -1) {
399         perror("fcntl");
400         (osi_Msg "rxi_Listen: unable to set non-blocking mode on socket\n");
401         return -1;
402     }
403
404     if (sock > FD_SETSIZE - 1) {
405         (osi_Msg "rxi_Listen: socket descriptor > (FD_SETSIZE-1) = %d\n",
406          FD_SETSIZE - 1);
407         return -1;
408     }
409 #endif
410
411     FD_SET(sock, &rx_selectMask);
412     if (sock > rx_maxSocketNumber)
413         rx_maxSocketNumber = sock;
414     if (sock < rx_minSocketNumber)
415         rx_minSocketNumber = sock;
416     return 0;
417 }
418
419 /*
420  * Recvmsg
421  */
422 int
423 rxi_Recvmsg(osi_socket socket, struct msghdr *msg_p, int flags)
424 {
425 #if defined(HAVE_LINUX_ERRQUEUE_H) && defined(ADAPT_PMTU)
426     while((rxi_HandleSocketError(socket)) > 0)
427         ;
428 #endif
429     return recvmsg(socket, msg_p, flags);
430 }
431
432 /*
433  * Simulate a blocking sendmsg on the non-blocking socket.
434  * It's non blocking because it was set that way for recvmsg.
435  */
436 int
437 rxi_Sendmsg(osi_socket socket, struct msghdr *msg_p, int flags)
438 {
439     fd_set *sfds = (fd_set *) 0;
440     while (sendmsg(socket, msg_p, flags) == -1) {
441         int err;
442         if (rx_stats_active)
443             rx_atomic_inc(&rx_stats.sendSelects);
444
445         if (!sfds) {
446             if (!(sfds = IOMGR_AllocFDSet())) {
447                 (osi_Msg "rx failed to alloc fd_set: ");
448                 perror("rx_sendmsg");
449                 return -1;
450             }
451             FD_SET(socket, sfds);
452         }
453 #if defined(HAVE_LINUX_ERRQUEUE_H) && defined(ADAPT_PMTU)
454         while((rxi_HandleSocketError(socket)) > 0)
455           ;
456 #endif
457 #ifdef AFS_NT40_ENV
458         if (WSAGetLastError())
459 #elif defined(AFS_LINUX22_ENV)
460         /* linux unfortunately returns ECONNREFUSED if the target port
461          * is no longer in use */
462         /* and EAGAIN if a UDP checksum is incorrect */
463         if (errno != EWOULDBLOCK && errno != ENOBUFS && errno != ECONNREFUSED
464             && errno != EAGAIN)
465 #else
466         if (errno != EWOULDBLOCK && errno != ENOBUFS)
467 #endif
468         {
469             (osi_Msg "rx failed to send packet: ");
470             perror("rx_sendmsg");
471 #ifndef AFS_NT40_ENV
472             if (errno > 0)
473               return -errno;
474 #else
475             if (WSAGetLastError() > 0)
476               return -WSAGetLastError();
477 #endif
478             return -1;
479         }
480         while ((err = select(socket + 1, 0, sfds, 0, 0)) != 1) {
481             if (err >= 0 || errno != EINTR)
482                 osi_Panic("rxi_sendmsg: select error %d.%d", err, errno);
483         }
484     }
485     if (sfds)
486         IOMGR_FreeFDSet(sfds);
487     return 0;
488 }