src/rx/rx_pthread.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 /*
  11  * An implementation of the rx socket listener for pthreads (not using select).
  12  * This assumes that multiple read system calls may be extant at any given
  13  * time. Also implements the pthread-specific event handling for rx.
  14  *
  15  * rx_pthread.c is used for the thread safe RX package.
  16  */
  17
  18 #include <afsconfig.h>
  19 #include <afs/param.h>
  20
  21
  22 #include <sys/types.h>
  23 #include <errno.h>
  24 #include <signal.h>
  25 #include <string.h>
  26 #ifndef AFS_NT40_ENV
  27 # include <sys/socket.h>
  28 # include <sys/file.h>
  29 # include <netdb.h>
  30 # include <netinet/in.h>
  31 # include <net/if.h>
  32 # include <sys/ioctl.h>
  33 # include <sys/time.h>
  34 #endif
  35 #include <sys/stat.h>
  36 #include <rx/rx.h>
  37 #include <rx/rx_globals.h>
  38 #include <assert.h>
  39 #include <rx/rx_pthread.h>
  40 #include <rx/rx_clock.h>
  41
  42 /*
  43  * Number of times the event handling thread was signalled because a new
  44  * event was scheduled earlier than the lastest event.
  45  *
  46  * Protected by event_handler_mutex
  47  */
  48 static long rx_pthread_n_event_wakeups;
  49
  50 /* Set rx_pthread_event_rescheduled if event_handler should just try
  51  * again instead of sleeping.
  52  *
  53  * Protected by event_handler_mutex
  54  */
  55 static int rx_pthread_event_rescheduled = 0;
  56
  57 static void *rx_ListenerProc(void *);
  58
  59 /*
  60  * We supply an event handling thread for Rx's event processing.
  61  * The condition variable is used to wakeup the thread whenever a new
  62  * event is scheduled earlier than the previous earliest event.
  63  * This thread is also responsible for keeping time.
  64  */
  65 static pthread_t event_handler_thread;
  66 afs_kcondvar_t rx_event_handler_cond;
  67 afs_kmutex_t event_handler_mutex;
  68 afs_kcondvar_t rx_listener_cond;
  69 afs_kmutex_t listener_mutex;
  70 static int listeners_started = 0;
  71 afs_kmutex_t rx_clock_mutex;
  72 struct clock rxi_clockNow;
  73
  74 /*
  75  * Delay the current thread the specified number of seconds.
  76  */
  77 void
  78 rxi_Delay(int sec)
  79 {
  80     sleep(sec);
  81 }
  82
  83 /*
  84  * Called from rx_Init()
  85  */
  86 void
  87 rxi_InitializeThreadSupport(void)
  88 {
  89         /* listeners_started must only be reset if
  90          * the listener thread terminates */
  91         /* listeners_started = 0; */
  92     clock_GetTime(&rxi_clockNow);
  93 }
  94
  95 static void *
  96 server_entry(void *argp)
  97 {
  98     void (*server_proc) () = (void (*)())argp;
  99     server_proc();
 100     dpf(("rx_pthread.c: server_entry: Server proc returned unexpectedly\n"));
 101     exit(1);
 102     return (void *)0;
 103 }
 104
 105 /*
 106  * Start an Rx server process.
 107  */
 108 void
 109 rxi_StartServerProc(void *(*proc) (void *), int stacksize)
 110 {
 111     pthread_t thread;
 112     pthread_attr_t tattr;
 113     AFS_SIGSET_DECL;
 114
 115     if (pthread_attr_init(&tattr) != 0) {
 116         dpf(("Unable to Create Rx server thread (pthread_attr_init)\n"));
 117         exit(1);
 118     }
 119
 120     if (pthread_attr_setdetachstate(&tattr, PTHREAD_CREATE_DETACHED) != 0) {
 121         dpf
 122             (("Unable to Create Rx server thread (pthread_attr_setdetachstate)\n"));
 123         exit(1);
 124     }
 125
 126     /*
 127      * NOTE: We are ignoring the stack size parameter, for now.
 128      */
 129     AFS_SIGSET_CLEAR();
 130     if (pthread_create(&thread, &tattr, server_entry, (void *)proc) != 0) {
 131         dpf(("Unable to Create Rx server thread\n"));
 132         exit(1);
 133     }
 134     AFS_SIGSET_RESTORE();
 135 }
 136
 137 /*
 138  * The event handling process.
 139  */
 140 static void *
 141 event_handler(void *argp)
 142 {
 143     struct clock rx_pthread_last_event_wait_time = { 0, 0 };
 144     unsigned long rx_pthread_n_event_expired = 0;
 145     unsigned long rx_pthread_n_event_waits = 0;
 146     long rx_pthread_n_event_woken = 0;
 147     unsigned long rx_pthread_n_event_error = 0;
 148     struct timespec rx_pthread_next_event_time = { 0, 0 };
 149     int error;
 150
 151     MUTEX_ENTER(&event_handler_mutex);
 152
 153     for (;;) {
 154         struct clock cv;
 155         struct clock next;
 156
 157         MUTEX_EXIT(&event_handler_mutex);
 158
 159         next.sec = 30;          /* Time to sleep if there are no events scheduled */
 160         next.usec = 0;
 161         clock_GetTime(&cv);
 162         rxevent_RaiseEvents(&next);
 163
 164         MUTEX_ENTER(&event_handler_mutex);
 165         if (rx_pthread_event_rescheduled) {
 166             rx_pthread_event_rescheduled = 0;
 167             continue;
 168         }
 169
 170         clock_Add(&cv, &next);
 171         rx_pthread_next_event_time.tv_sec = cv.sec;
 172         rx_pthread_next_event_time.tv_nsec = cv.usec * 1000;
 173         rx_pthread_n_event_waits++;
 174         error = CV_TIMEDWAIT(&rx_event_handler_cond, &event_handler_mutex, &rx_pthread_next_event_time);
 175         if (error == 0) {
 176             rx_pthread_n_event_woken++;
 177         }
 178 #ifdef AFS_NT40_ENV
 179         else if (error == ETIMEDOUT) {
 180             rx_pthread_n_event_expired++;
 181         } else {
 182             rx_pthread_n_event_error++;
 183         }
 184 #else
 185         else if (errno == ETIMEDOUT) {
 186             rx_pthread_n_event_expired++;
 187         } else {
 188             rx_pthread_n_event_error++;
 189         }
 190 #endif
 191         rx_pthread_event_rescheduled = 0;
 192     }
 193 }
 194
 195
 196 /*
 197  * This routine will get called by the event package whenever a new,
 198  * earlier than others, event is posted. */
 199 void
 200 rxi_ReScheduleEvents(void)
 201 {
 202     MUTEX_ENTER(&event_handler_mutex);
 203     CV_SIGNAL(&rx_event_handler_cond);
 204     rx_pthread_event_rescheduled = 1;
 205     MUTEX_EXIT(&event_handler_mutex);
 206 }
 207
 208
 209 /* Loop to listen on a socket. Return setting *newcallp if this
 210  * thread should become a server thread.  */
 211 static void
 212 rxi_ListenerProc(osi_socket sock, int *tnop, struct rx_call **newcallp)
 213 {
 214     unsigned int host;
 215     u_short port;
 216     struct rx_packet *p = (struct rx_packet *)0;
 217
 218     MUTEX_ENTER(&listener_mutex);
 219     while (!listeners_started) {
 220         CV_WAIT(&rx_listener_cond, &listener_mutex);
 221     }
 222     MUTEX_EXIT(&listener_mutex);
 223
 224     for (;;) {
 225         /*
 226          * Grab a new packet only if necessary (otherwise re-use the old one)
 227          */
 228         if (p) {
 229             rxi_RestoreDataBufs(p);
 230         } else {
 231             if (!(p = rxi_AllocPacket(RX_PACKET_CLASS_RECEIVE))) {
 232                 /* Could this happen with multiple socket listeners? */
 233                 dpf(("rxi_Listener: no packets!"));     /* Shouldn't happen */
 234                 exit(1);
 235             }
 236         }
 237
 238         if (rxi_ReadPacket(sock, p, &host, &port)) {
 239             clock_NewTime();
 240             p = rxi_ReceivePacket(p, sock, host, port, tnop, newcallp);
 241             if (newcallp && *newcallp) {
 242                 if (p)
 243                     rxi_FreePacket(p);
 244                 return;
 245             }
 246         }
 247     }
 248     /* NOTREACHED */
 249 }
 250
 251 /* This is the listener process request loop. The listener process loop
 252  * becomes a server thread when rxi_ListenerProc returns, and stays
 253  * server thread until rxi_ServerProc returns. */
 254 static void *
 255 rx_ListenerProc(void *argp)
 256 {
 257     int threadID;
 258     osi_socket sock = (osi_socket)argp;
 259     struct rx_call *newcall;
 260
 261     while (1) {
 262         newcall = NULL;
 263         threadID = -1;
 264         rxi_ListenerProc(sock, &threadID, &newcall);
 265         /* assert(threadID != -1); */
 266         /* assert(newcall != NULL); */
 267         sock = OSI_NULLSOCKET;
 268         assert(pthread_setspecific(rx_thread_id_key, (void *)threadID) == 0);
 269         rxi_ServerProc(threadID, newcall, &sock);
 270         /* assert(sock != OSI_NULLSOCKET); */
 271     }
 272     /* not reached */
 273 }
 274
 275 /* This is the server process request loop. The server process loop
 276  * becomes a listener thread when rxi_ServerProc returns, and stays
 277  * listener thread until rxi_ListenerProc returns. */
 278 void *
 279 rx_ServerProc(void * dummy)
 280 {
 281     osi_socket sock;
 282     int threadID;
 283     struct rx_call *newcall = NULL;
 284
 285     rxi_MorePackets(rx_maxReceiveWindow + 2);   /* alloc more packets */
 286     MUTEX_ENTER(&rx_quota_mutex);
 287     rxi_dataQuota += rx_initSendWindow; /* Reserve some pkts for hard times */
 288     /* threadID is used for making decisions in GetCall.  Get it by bumping
 289      * number of threads handling incoming calls */
 290     /* Unique thread ID: used for scheduling purposes *and* as index into
 291      * the host hold table (fileserver).
 292      * The previously used rxi_availProcs is unsuitable as it
 293      * will already go up and down as packets arrive while the server
 294      * threads are still initialising! The recently introduced
 295      * rxi_pthread_hinum does not necessarily lead to a server
 296      * thread with id 0, which is not allowed to hop through the
 297      * incoming call queue.
 298      * So either introduce yet another counter or flag the FCFS
 299      * thread... chose the latter.
 300      */
 301     MUTEX_ENTER(&rx_pthread_mutex);
 302     threadID = ++rxi_pthread_hinum;
 303     if (rxi_fcfs_thread_num == 0 && rxi_fcfs_thread_num != threadID)
 304         rxi_fcfs_thread_num = threadID;
 305     MUTEX_EXIT(&rx_pthread_mutex);
 306     ++rxi_availProcs;
 307     MUTEX_EXIT(&rx_quota_mutex);
 308
 309     while (1) {
 310         sock = OSI_NULLSOCKET;
 311         assert(pthread_setspecific(rx_thread_id_key, (void *)threadID) == 0);
 312         rxi_ServerProc(threadID, newcall, &sock);
 313         /* assert(sock != OSI_NULLSOCKET); */
 314         newcall = NULL;
 315         rxi_ListenerProc(sock, &threadID, &newcall);
 316         /* assert(threadID != -1); */
 317         /* assert(newcall != NULL); */
 318     }
 319     /* not reached */
 320 }
 321
 322 /*
 323  * Historically used to start the listener process. We now have multiple
 324  * listener processes (one for each socket); these are started by GetUdpSocket.
 325  *
 326  * The event handling process *is* started here (the old listener used
 327  * to also handle events). The listener threads can't actually start
 328  * listening until rxi_StartListener is called because most of R may not
 329  * be initialized when rxi_Listen is called.
 330  */
 331 void
 332 rxi_StartListener(void)
 333 {
 334     pthread_attr_t tattr;
 335     AFS_SIGSET_DECL;
 336
 337         if (listeners_started)
 338                 return;
 339
 340     if (pthread_attr_init(&tattr) != 0) {
 341         dpf
 342             (("Unable to create Rx event handling thread (pthread_attr_init)\n"));
 343         exit(1);
 344     }
 345
 346     if (pthread_attr_setdetachstate(&tattr, PTHREAD_CREATE_DETACHED) != 0) {
 347         dpf
 348             (("Unable to create Rx event handling thread (pthread_attr_setdetachstate)\n"));
 349         exit(1);
 350     }
 351
 352     AFS_SIGSET_CLEAR();
 353     if (pthread_create(&event_handler_thread, &tattr, event_handler, NULL) !=
 354         0) {
 355         dpf(("Unable to create Rx event handling thread\n"));
 356         exit(1);
 357     }
 358     MUTEX_ENTER(&rx_pthread_mutex);
 359     ++rxi_pthread_hinum;
 360     MUTEX_EXIT(&rx_pthread_mutex);
 361     AFS_SIGSET_RESTORE();
 362
 363     MUTEX_ENTER(&listener_mutex);
 364     CV_BROADCAST(&rx_listener_cond);
 365     listeners_started = 1;
 366     MUTEX_EXIT(&listener_mutex);
 367
 368 }
 369
 370 /*
 371  * Listen on the specified socket.
 372  */
 373 int
 374 rxi_Listen(osi_socket sock)
 375 {
 376     pthread_t thread;
 377     pthread_attr_t tattr;
 378     AFS_SIGSET_DECL;
 379
 380     if (pthread_attr_init(&tattr) != 0) {
 381         dpf
 382             (("Unable to create socket listener thread (pthread_attr_init)\n"));
 383         exit(1);
 384     }
 385
 386     if (pthread_attr_setdetachstate(&tattr, PTHREAD_CREATE_DETACHED) != 0) {
 387         dpf
 388             (("Unable to create socket listener thread (pthread_attr_setdetachstate)\n"));
 389         exit(1);
 390     }
 391
 392     AFS_SIGSET_CLEAR();
 393     if (pthread_create(&thread, &tattr, rx_ListenerProc, (void *)sock) != 0) {
 394         dpf(("Unable to create socket listener thread\n"));
 395         exit(1);
 396     }
 397     MUTEX_ENTER(&rx_pthread_mutex);
 398     ++rxi_pthread_hinum;
 399     MUTEX_EXIT(&rx_pthread_mutex);
 400     AFS_SIGSET_RESTORE();
 401     return 0;
 402 }
 403
 404
 405 /*
 406  * Recvmsg.
 407  *
 408  */
 409 int
 410 rxi_Recvmsg(osi_socket socket, struct msghdr *msg_p, int flags)
 411 {
 412     int ret;
 413 #if defined(HAVE_LINUX_ERRQUEUE_H) && defined(ADAPT_PMTU)
 414     while((rxi_HandleSocketError(socket)) > 0)
 415       ;
 416 #endif
 417     ret = recvmsg(socket, msg_p, flags);
 418     return ret;
 419 }
 420
 421 /*
 422  * Sendmsg.
 423  */
 424 int
 425 rxi_Sendmsg(osi_socket socket, struct msghdr *msg_p, int flags)
 426 {
 427     int ret;
 428     ret = sendmsg(socket, msg_p, flags);
 429 #ifdef AFS_LINUX22_ENV
 430     /* linux unfortunately returns ECONNREFUSED if the target port
 431      * is no longer in use */
 432     /* and EAGAIN if a UDP checksum is incorrect */
 433     if (ret == -1 && errno != ECONNREFUSED && errno != EAGAIN) {
 434 #else
 435     if (ret == -1) {
 436 #endif
 437         dpf(("rxi_sendmsg failed, error %d\n", errno));
 438         fflush(stdout);
 439         return -1;
 440     }
 441     return 0;
 442 }
 443
 444 struct rx_ts_info_t * rx_ts_info_init() {
 445     struct rx_ts_info_t * rx_ts_info;
 446     rx_ts_info = (rx_ts_info_t *) malloc(sizeof(rx_ts_info_t));
 447     assert(rx_ts_info != NULL && pthread_setspecific(rx_ts_info_key, rx_ts_info) == 0);
 448     memset(rx_ts_info, 0, sizeof(rx_ts_info_t));
 449 #ifdef RX_ENABLE_TSFPQ
 450     queue_Init(&rx_ts_info->_FPQ);
 451
 452     MUTEX_ENTER(&rx_packets_mutex);
 453     rx_TSFPQMaxProcs++;
 454     RX_TS_FPQ_COMPUTE_LIMITS;
 455     MUTEX_EXIT(&rx_packets_mutex);
 456 #endif /* RX_ENABLE_TSFPQ */
 457     return rx_ts_info;
 458 }