src/rx/UKERNEL/rx_knet.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 #include <afsconfig.h>
  11 #include "afs/param.h"
  12
  13
  14 #include "rx/rx_kcommon.h"
  15 #include "rx_atomic.h"
  16 #include "rx_internal.h"
  17
  18 #define SECONDS_TO_SLEEP        0
  19 #define NANO_SECONDS_TO_SLEEP   100000000       /* 100 milliseconds */
  20 #define LOOPS_PER_WAITCHECK     10      /* once per second */
  21
  22 struct usr_socket {
  23     int sock;
  24     short port;
  25 };
  26
  27 unsigned short usr_rx_port = 0;
  28
  29 struct usr_ifnet *usr_ifnet = NULL;
  30 struct usr_in_ifaddr *usr_in_ifaddr = NULL;
  31
  32 void rxk_InitializeSocket(void);
  33 extern int afs_osi_CheckTimedWaits(void);
  34
  35 void
  36 afs_rxevent_daemon(void)
  37 {
  38     struct timespec tv;
  39     struct clock temp;
  40     int i = 0;
  41
  42     AFS_GUNLOCK();
  43     while (1) {
  44         tv.tv_sec = SECONDS_TO_SLEEP;
  45         tv.tv_nsec = NANO_SECONDS_TO_SLEEP;
  46         usr_thread_sleep(&tv);
  47         /*
  48          * Check for shutdown, don't try to stop the listener
  49          */
  50         if (afs_termState == AFSOP_STOP_RXEVENT
  51             || afs_termState == AFSOP_STOP_RXK_LISTENER) {
  52             AFS_GLOCK();
  53             afs_termState = AFSOP_STOP_COMPLETE;
  54             afs_osi_Wakeup(&afs_termState);
  55             return;
  56         }
  57         rxevent_RaiseEvents(&temp);
  58         if (++i >= LOOPS_PER_WAITCHECK) {
  59             i = 0;
  60             afs_osi_CheckTimedWaits();
  61         }
  62     }
  63 }
  64
  65
  66 /* Loop to listen on a socket. Return setting *newcallp if this
  67  * thread should become a server thread.  */
  68 void
  69 rxi_ListenerProc(osi_socket usockp, int *tnop, struct rx_call **newcallp)
  70 {
  71     struct rx_packet *tp;
  72     afs_uint32 host;
  73     u_short port;
  74     int rc;
  75
  76     /*
  77      * Use the rxk_GetPacketProc and rxk_PacketArrivalProc routines
  78      * to allocate rx_packet buffers and pass them to the RX layer
  79      * for processing.
  80      */
  81     while (1) {
  82         /* See if a check for additional packets was issued */
  83         rx_CheckPackets();
  84
  85         tp = rxi_AllocPacket(RX_PACKET_CLASS_RECEIVE);
  86         usr_assert(tp != NULL);
  87         rc = rxi_ReadPacket(usockp, tp, &host, &port);
  88         if (rc != 0) {
  89             tp = rxi_ReceivePacket(tp, usockp, host, port, tnop, newcallp);
  90             if (newcallp && *newcallp) {
  91                 if (tp) {
  92                     rxi_FreePacket(tp);
  93                 }
  94                 return;
  95             }
  96         }
  97         if (tp) {
  98             rxi_FreePacket(tp);
  99         }
 100         if (afs_termState == AFSOP_STOP_RXEVENT) {
 101             afs_termState = AFSOP_STOP_RXK_LISTENER;
 102             afs_osi_Wakeup(&afs_termState);
 103         }
 104     }
 105 }
 106
 107 /* This is the listener process request loop. The listener process loop
 108  * becomes a server thread when rxi_ListenerProc returns, and stays
 109  * server thread until rxi_ServerProc returns. */
 110 void
 111 rxk_Listener(void)
 112 {
 113     int threadID;
 114     osi_socket sock = (osi_socket) rx_socket;
 115     struct rx_call *newcall;
 116     struct usr_socket *usockp;
 117
 118     /*
 119      * Initialize the rx_socket and start the receiver threads
 120      */
 121     rxk_InitializeSocket();
 122
 123     usockp = (struct usr_socket *)rx_socket;
 124     assert(usockp != NULL);
 125
 126     AFS_GUNLOCK();
 127     while (1) {
 128         newcall = NULL;
 129         threadID = -1;
 130         rxi_ListenerProc(sock, &threadID, &newcall);
 131         /* assert(threadID != -1); */
 132         /* assert(newcall != NULL); */
 133         sock = OSI_NULLSOCKET;
 134         rxi_ServerProc(threadID, newcall, &sock);
 135         if (sock == OSI_NULLSOCKET) {
 136             break;
 137         }
 138     }
 139     AFS_GLOCK();
 140 }
 141
 142 /* This is the server process request loop. The server process loop
 143  * becomes a listener thread when rxi_ServerProc returns, and stays
 144  * listener thread until rxi_ListenerProc returns. */
 145 void *
 146 rx_ServerProc(void *unused)
 147 {
 148     osi_socket sock;
 149     int threadID;
 150     struct rx_call *newcall = NULL;
 151
 152     rxi_MorePackets(rx_maxReceiveWindow + 2);   /* alloc more packets */
 153     rxi_dataQuota += rx_initSendWindow; /* Reserve some pkts for hard times */
 154     /* threadID is used for making decisions in GetCall.  Get it by bumping
 155      * number of threads handling incoming calls */
 156     threadID = rxi_availProcs++;
 157
 158     AFS_GUNLOCK();
 159     while (1) {
 160         sock = OSI_NULLSOCKET;
 161         rxi_ServerProc(threadID, newcall, &sock);
 162         if (sock == OSI_NULLSOCKET) {
 163             break;
 164         }
 165         newcall = NULL;
 166         threadID = -1;
 167         rxi_ListenerProc(sock, &threadID, &newcall);
 168         /* assert(threadID != -1); */
 169         /* assert(newcall != NULL); */
 170     }
 171     AFS_GLOCK();
 172     return NULL;
 173 }
 174
 175 /*
 176  * At this point, RX wants a socket, but still has not initialized the
 177  * rx_port variable or the pointers to the packet allocater and arrival
 178  * routines. Allocate the socket buffer here, but don't open it until
 179  * we start the receiver threads.
 180  */
 181 osi_socket *
 182 rxk_NewSocketHost(afs_uint32 ahost, short aport)
 183 {
 184     struct usr_socket *usockp;
 185
 186     usockp = afs_osi_Alloc(sizeof(struct usr_socket));
 187     usr_assert(usockp != NULL);
 188
 189     usockp->sock = -1;
 190
 191     return (osi_socket *)usockp;
 192 }
 193
 194 osi_socket *
 195 rxk_NewSocket(short aport)
 196 {
 197     return rxk_NewSocketHost(htonl(INADDR_ANY), aport);
 198 }
 199
 200 /*
 201  * This routine is called from rxk_Listener. By this time rx_port
 202  * is set to 7001 and rx_socket points to the socket buffer
 203  * we allocated in rxk_NewSocket. Now is the time to bind our
 204  * socket and start the receiver threads.
 205  */
 206 void
 207 rxk_InitializeSocket(void)
 208 {
 209     int rc, sock;
 210 #ifdef AFS_USR_AIX_ENV
 211     unsigned long len, optval, optval0, optlen;
 212 #else /* AFS_USR_AIX_ENV */
 213     socklen_t len, optlen;
 214     int optval, optval0;
 215 #endif /* AFS_USR_AIX_ENV */
 216     struct usr_socket *usockp;
 217     struct sockaddr_in lcladdr;
 218
 219     usr_assert(rx_socket != NULL);
 220     usockp = (struct usr_socket *)rx_socket;
 221
 222     sock = socket(PF_INET, SOCK_DGRAM, 0);
 223     usr_assert(sock >= 0);
 224
 225     memset((void *)&lcladdr, 0, sizeof(struct sockaddr_in));
 226     lcladdr.sin_family = AF_INET;
 227     lcladdr.sin_port = htons(usr_rx_port);
 228     lcladdr.sin_addr.s_addr = INADDR_ANY;
 229     rc = bind(sock, (struct sockaddr *)&lcladdr, sizeof(struct sockaddr_in));
 230     usr_assert(rc >= 0);
 231     len = sizeof(struct sockaddr_in);
 232     rc = getsockname(sock, (struct sockaddr *)&lcladdr, &len);
 233     usr_assert(rc >= 0);
 234 #ifdef AFS_USR_LINUX_ENV
 235     optval0 = 131070;
 236 #else
 237     optval0 = 131072;
 238 #endif
 239     optval = optval0;
 240     rc = setsockopt(sock, SOL_SOCKET, SO_SNDBUF, (void *)&optval,
 241                     sizeof(optval));
 242     usr_assert(rc == 0);
 243     optlen = sizeof(optval);
 244     rc = getsockopt(sock, SOL_SOCKET, SO_SNDBUF, (void *)&optval, &optlen);
 245     usr_assert(rc == 0);
 246     /* usr_assert(optval == optval0); */
 247 #ifdef AFS_USR_LINUX_ENV
 248     optval0 = 131070;
 249 #else
 250     optval0 = 131072;
 251 #endif
 252     optval = optval0;
 253     rc = setsockopt(sock, SOL_SOCKET, SO_RCVBUF, (void *)&optval,
 254                     sizeof(optval));
 255     usr_assert(rc == 0);
 256     optlen = sizeof(optval);
 257     rc = getsockopt(sock, SOL_SOCKET, SO_RCVBUF, (void *)&optval, &optlen);
 258     usr_assert(rc == 0);
 259     /* usr_assert(optval == optval0); */
 260
 261 #ifdef AFS_USR_AIX_ENV
 262     optval = 1;
 263     rc = setsockopt(sock, SOL_SOCKET, SO_CKSUMRECV, (void *)&optval,
 264                     sizeof(optval));
 265     usr_assert(rc == 0);
 266 #endif /* AFS_USR_AIX_ENV */
 267
 268 #ifdef FD_CLOEXEC
 269     fcntl(sock, F_SETFD, FD_CLOEXEC);
 270 #endif
 271
 272     usockp->sock = sock;
 273     usockp->port = lcladdr.sin_port;
 274
 275     /*
 276      * Set the value of rx_port to reflect the address we actually
 277      * are listening on, since the kernel is probably already using 7001.
 278      */
 279     rx_port = usockp->port;
 280 }
 281
 282 int
 283 rxk_FreeSocket(struct usr_socket *sockp)
 284 {
 285     return 0;
 286 }
 287
 288 void
 289 osi_StopListener(void)
 290 {
 291     rxk_FreeSocket((struct usr_socket *)rx_socket);
 292 }
 293
 294 int
 295 osi_NetSend(osi_socket sockp, struct sockaddr_in *addr, struct iovec *iov,
 296             int nio, afs_int32 size, int stack)
 297 {
 298     int rc;
 299     int i;
 300     struct usr_socket *usockp = (struct usr_socket *)sockp;
 301     struct msghdr msg;
 302     struct iovec tmpiov[64];
 303
 304     /*
 305      * The header is in the first iovec
 306      */
 307     usr_assert(nio > 0 && nio <= 64);
 308     for (i = 0; i < nio; i++) {
 309         tmpiov[i].iov_base = iov[i].iov_base;
 310         tmpiov[i].iov_len = iov[i].iov_len;
 311     }
 312
 313     memset(&msg, 0, sizeof(msg));
 314     msg.msg_name = (void *)addr;
 315     msg.msg_namelen = sizeof(struct sockaddr_in);
 316     msg.msg_iov = &tmpiov[0];
 317     msg.msg_iovlen = nio;
 318
 319     rc = sendmsg(usockp->sock, &msg, 0);
 320     if (rc < 0) {
 321         return errno;
 322     }
 323     usr_assert(rc == size);
 324
 325     return 0;
 326 }
 327
 328 void
 329 shutdown_rxkernel(void)
 330 {
 331     rxk_initDone = 0;
 332     rxk_shutdownPorts();
 333 }
 334
 335 void
 336 rx_Finalize(void)
 337 {
 338     usr_assert(0);
 339 }
 340
 341 /*
 342  * Recvmsg.
 343  *
 344  */
 345 int
 346 rxi_Recvmsg(osi_socket socket, struct msghdr *msg_p, int flags)
 347 {
 348     struct usr_socket *usock = (struct usr_socket *)socket;
 349     int ret;
 350     do {
 351         ret = recvmsg(usock->sock, msg_p, flags);
 352     } while (ret == -1 && errno == EAGAIN);
 353     return ret;
 354 }