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