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