/*
* Copyright 2000, International Business Machines Corporation and others.
* All Rights Reserved.
- *
+ *
* This software has been released under the terms of the IBM Public
* License. For details, see the LICENSE file in the top-level source
* directory or online at http://www.openafs.org/dl/license10.html
#include <afsconfig.h>
#include <afs/param.h>
-RCSID
- ("$Header$");
+#include <roken.h>
-# include <sys/types.h> /* fd_set on older platforms */
-# include <errno.h>
-# include <signal.h>
-#ifdef AFS_NT40_ENV
-# include <winsock2.h>
-#else
-# include <unistd.h> /* select() prototype */
-# include <sys/time.h> /* struct timeval, select() prototype */
-# ifndef FD_SET
-# include <sys/select.h> /* fd_set on newer platforms */
-# endif
-# include <sys/socket.h>
+#ifdef HAVE_SYS_FILE_H
# include <sys/file.h>
-# include <netdb.h>
-# include <sys/stat.h>
-# include <netinet/in.h>
-# include <net/if.h>
-# include <sys/ioctl.h>
-# include <sys/time.h>
#endif
-# include "rx.h"
-# include "rx_globals.h"
-# include <lwp.h>
-#define MAXTHREADNAMELENGTH 64
+#ifndef AFS_PTHREAD_ENV
+
+#include <lwp.h>
-extern int (*registerProgram) ();
-extern int (*swapNameProgram) ();
+#include "rx.h"
+#include "rx_atomic.h"
+#include "rx_globals.h"
+#include "rx_internal.h"
+#include "rx_stats.h"
+#ifdef AFS_NT40_ENV
+#include "rx_xmit_nt.h"
+#endif
+
+#define MAXTHREADNAMELENGTH 64
int debugSelectFailure; /* # of times select failed */
}
PROCESS rx_listenerPid = 0; /* LWP process id of socket listener process */
-static void rx_ListenerProc(void *dummy);
+static void* rx_ListenerProc(void *dummy);
/*
* Delay the current thread the specified number of seconds.
}
void
-rxi_StartServerProc(void (*proc) (void), int stacksize)
+rxi_StartServerProc(void *(*proc) (void *), int stacksize)
{
PROCESS scratchPid;
static int number = 0;
char name[32];
sprintf(name, "srv_%d", ++number);
- LWP_CreateProcess(proc, stacksize, RX_PROCESS_PRIORITY, (void *)0,
+ LWP_CreateProcess(proc, stacksize, RX_PROCESS_PRIORITY, NULL,
"rx_ServerProc", &scratchPid);
if (registerProgram)
(*registerProgram) (scratchPid, name);
/* Priority of listener should be high, so it can keep conns alive */
#define RX_LIST_STACK 24000
LWP_CreateProcess(rx_ListenerProc, RX_LIST_STACK, LWP_MAX_PRIORITY,
- (void *)0, "rx_Listener", &rx_listenerPid);
+ NULL, "rx_Listener", &rx_listenerPid);
if (registerProgram)
(*registerProgram) (rx_listenerPid, "listener");
}
{
afs_uint32 host;
u_short port;
- register struct rx_packet *p = (struct rx_packet *)0;
- int socket;
+ struct rx_packet *p = (struct rx_packet *)0;
+ osi_socket socket;
struct clock cv;
afs_int32 nextPollTime; /* time to next poll FD before sleeping */
int lastPollWorked, doingPoll; /* true iff last poll was useful */
nextPollTime = 0;
code = LWP_CurrentProcess(&pid);
if (code) {
- fprintf(stderr, "rxi_Listener: Can't get my pid.\n");
- exit(1);
+ osi_Panic("rxi_Listener: Can't get my pid.\n");
}
rx_listenerPid = pid;
if (swapNameProgram)
(*swapNameProgram) (pid, "listener", &name[0]);
for (;;) {
+ /* See if a check for additional packets was issued */
+ rx_CheckPackets();
+
/* Grab a new packet only if necessary (otherwise re-use the old one) */
if (p) {
rxi_RestoreDataBufs(p);
tv.tv_usec = cv.usec;
tvp = &tv;
}
- rx_stats.selects++;
+ if (rx_stats_active)
+ rx_atomic_inc(&rx_stats.selects);
*rfds = rx_selectMask;
nextPollTime = clock_Sec() + 4; /* try again in 4 seconds no matter what */
tv.tv_sec = tv.tv_usec = 0; /* make sure we poll */
tvp = &tv;
- code = select(rx_maxSocketNumber + 1, rfds, 0, 0, tvp);
+ code = select((int)(rx_maxSocketNumber + 1), rfds, 0, 0, tvp);
} else {
doingPoll = 0;
- code = IOMGR_Select(rx_maxSocketNumber + 1, rfds, 0, 0, tvp);
+ code = IOMGR_Select((int)(rx_maxSocketNumber + 1), rfds, 0, 0, tvp);
}
lastPollWorked = 0; /* default is that it didn't find anything */
rxi_FreePacket(p);
}
if (swapNameProgram) {
- (*swapNameProgram) (rx_listenerPid, &name, 0);
+ (*swapNameProgram) (rx_listenerPid, name, 0);
rx_listenerPid = 0;
}
return;
rxi_FreePacket(p);
}
if (swapNameProgram) {
- (*swapNameProgram) (rx_listenerPid, &name, 0);
+ (*swapNameProgram) (rx_listenerPid, name, 0);
rx_listenerPid = 0;
}
return;
/* This is the listener process request loop. The listener process loop
* becomes a server thread when rxi_ListenerProc returns, and stays
* server thread until rxi_ServerProc returns. */
-static void
+static void *
rx_ListenerProc(void *dummy)
{
int threadID;
- int sock;
+ osi_socket sock;
struct rx_call *newcall;
fd_set *rfds;
newcall = NULL;
threadID = -1;
rxi_ListenerProc(rfds, &threadID, &newcall);
- /* assert(threadID != -1); */
- /* assert(newcall != NULL); */
+ /* osi_Assert(threadID != -1); */
+ /* osi_Assert(newcall != NULL); */
sock = OSI_NULLSOCKET;
rxi_ServerProc(threadID, newcall, &sock);
- /* assert(sock != OSI_NULLSOCKET); */
+ /* osi_Assert(sock != OSI_NULLSOCKET); */
}
/* not reached */
+ return NULL;
}
/* This is the server process request loop. The server process loop
* becomes a listener thread when rxi_ServerProc returns, and stays
* listener thread until rxi_ListenerProc returns. */
-void
-rx_ServerProc(void)
+void *
+rx_ServerProc(void * unused)
{
- int sock;
+ osi_socket sock;
int threadID;
struct rx_call *newcall = NULL;
fd_set *rfds;
while (1) {
sock = OSI_NULLSOCKET;
rxi_ServerProc(threadID, newcall, &sock);
- /* assert(sock != OSI_NULLSOCKET); */
+ /* osi_Assert(sock != OSI_NULLSOCKET); */
newcall = NULL;
rxi_ListenerProc(rfds, &threadID, &newcall);
- /* assert(threadID != -1); */
- /* assert(newcall != NULL); */
+ /* osi_Assert(threadID != -1); */
+ /* osi_Assert(newcall != NULL); */
}
/* not reached */
+ return NULL;
}
/*
* Put the socket into non-blocking mode so that rx_Listener
* can do a polling read before entering select
*/
-#ifndef AFS_DJGPP_ENV
if (fcntl(sock, F_SETFL, FNDELAY) == -1) {
perror("fcntl");
(osi_Msg "rxi_Listen: unable to set non-blocking mode on socket\n");
return -1;
}
-#else
- if (__djgpp_set_socket_blocking_mode(sock, 1) < 0) {
- perror("__djgpp_set_socket_blocking_mode");
- (osi_Msg "rxi_Listen: unable to set non-blocking mode on socket\n");
- return -1;
- }
-#endif /* AFS_DJGPP_ENV */
if (sock > FD_SETSIZE - 1) {
(osi_Msg "rxi_Listen: socket descriptor > (FD_SETSIZE-1) = %d\n",
* Recvmsg
*/
int
-rxi_Recvmsg(int socket, struct msghdr *msg_p, int flags)
+rxi_Recvmsg(osi_socket socket, struct msghdr *msg_p, int flags)
{
- return recvmsg((int)socket, msg_p, flags);
+ int code;
+ code = recvmsg(socket, msg_p, flags);
+
+#ifdef AFS_RXERRQ_ENV
+ if (code < 0) {
+ while((rxi_HandleSocketError(socket)) > 0)
+ ;
+ }
+#endif
+
+ return code;
}
/*
fd_set *sfds = (fd_set *) 0;
while (sendmsg(socket, msg_p, flags) == -1) {
int err;
- rx_stats.sendSelects++;
+
+#ifdef AFS_NT40_ENV
+ err = WSAGetLastError();
+#else
+ err = errno;
+#endif
+
+ if (rx_stats_active)
+ rx_atomic_inc(&rx_stats.sendSelects);
if (!sfds) {
if (!(sfds = IOMGR_AllocFDSet())) {
(osi_Msg "rx failed to alloc fd_set: ");
perror("rx_sendmsg");
- return 3;
+ return -1;
}
FD_SET(socket, sfds);
}
+#ifdef AFS_RXERRQ_ENV
+ while((rxi_HandleSocketError(socket)) > 0)
+ ;
+#endif
#ifdef AFS_NT40_ENV
- if (errno)
+ if (err)
#elif defined(AFS_LINUX22_ENV)
/* linux unfortunately returns ECONNREFUSED if the target port
* is no longer in use */
/* and EAGAIN if a UDP checksum is incorrect */
- if (errno != EWOULDBLOCK && errno != ENOBUFS && errno != ECONNREFUSED
- && errno != EAGAIN)
+ if (err != EWOULDBLOCK && err != ENOBUFS && err != ECONNREFUSED
+ && err != EAGAIN)
#else
- if (errno != EWOULDBLOCK && errno != ENOBUFS)
+ if (err != EWOULDBLOCK && err != ENOBUFS)
#endif
{
(osi_Msg "rx failed to send packet: ");
perror("rx_sendmsg");
- return 3;
+ if (err > 0)
+ return -err;
+ return -1;
}
- while ((err = select(socket + 1, 0, sfds, 0, 0)) != 1) {
+ while ((err = select(
+#ifdef AFS_NT40_ENV
+ 0,
+#else
+ socket + 1,
+#endif
+ 0, sfds, 0, 0)) != 1) {
if (err >= 0 || errno != EINTR)
osi_Panic("rxi_sendmsg: select error %d.%d", err, errno);
+ FD_ZERO(sfds);
+ FD_SET(socket, sfds);
}
}
if (sfds)
IOMGR_FreeFDSet(sfds);
return 0;
}
+#endif
git://git.openafs.org / openafs.git / blobdiff