servr_ku.c

Netscape NSPR库源码

/* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* 
 * The contents of this file are subject to the Mozilla Public
 * License Version 1.1 (the "License"); you may not use this file
 * except in compliance with the License. You may obtain a copy of
 * the License at http://www.mozilla.org/MPL/
 * 
 * Software distributed under the License is distributed on an "AS
 * IS" basis, WITHOUT WARRANTY OF ANY KIND, either express or
 * implied. See the License for the specific language governing
 * rights and limitations under the License.
 * 
 * The Original Code is the Netscape Portable Runtime (NSPR).
 * 
 * The Initial Developer of the Original Code is Netscape
 * Communications Corporation.  Portions created by Netscape are 
 * Copyright (C) 1998-2000 Netscape Communications Corporation.  All
 * Rights Reserved.
 * 
 * Contributor(s):
 * 
 * Alternatively, the contents of this file may be used under the
 * terms of the GNU General Public License Version 2 or later (the
 * "GPL"), in which case the provisions of the GPL are applicable 
 * instead of those above.  If you wish to allow use of your 
 * version of this file only under the terms of the GPL and not to
 * allow others to use your version of this file under the MPL,
 * indicate your decision by deleting the provisions above and
 * replace them with the notice and other provisions required by
 * the GPL.  If you do not delete the provisions above, a recipient
 * may use your version of this file under either the MPL or the
 * GPL.
 */

/***********************************************************************
**
** This server simulates a server running in loopback mode.
**
** The idea is that a single server is created.  The server initially creates
** a number of worker threads.  Then, with the server running, a number of 
** clients are created which start requesting service from the server.
**
**
** Modification History:
** 19-May-97 AGarcia- Converted the test to accomodate the debug_mode flag.
**	         The debug mode will print all of the printfs associated with this test.
**			 The regress mode will be the default mode. Since the regress tool limits
**           the output to a one line status:PASS or FAIL,all of the printf statements
**			 have been handled with an if (debug_mode) statement.
** 04-June-97 AGarcia removed the Test_Result function. Regress tool has been updated to
**			recognize the return code from tha main program.
***********************************************************************/

/***********************************************************************
** Includes
***********************************************************************/
/* Used to get the command line option */
#include "plgetopt.h"

#include "nspr.h"
#include "pprthred.h"

#include <string.h>

#define PORT 15004
#define STACKSIZE 0

static int _iterations = 1000;
static int _clients = 1;
static int _client_data = 250;
static int _server_data = (8*1024);

static PRThreadScope ServerScope, ClientScope;

#define SERVER "Server"
#define MAIN   "Main"

#define SERVER_STATE_STARTUP 0
#define SERVER_STATE_READY   1
#define SERVER_STATE_DYING   2
#define SERVER_STATE_DEAD    4
int       ServerState;
PRLock    *ServerStateCVLock;
PRCondVar *ServerStateCV;

#ifdef DEBUGPRINTS
#define DPRINTF printf
#else
#define DPRINTF
#endif

PRIntn failed_already=0;
PRIntn debug_mode;

static void do_work(void);

/* --- Server state functions --------------------------------------------- */
void
SetServerState(char *waiter, PRInt32 state)
{
    PR_Lock(ServerStateCVLock);
    ServerState = state;
    PR_NotifyCondVar(ServerStateCV);

	if (debug_mode) DPRINTF("\t%s changed state to %d\n", waiter, state);

    PR_Unlock(ServerStateCVLock);
}

int
WaitServerState(char *waiter, PRInt32 state)
{
    PRInt32 rv;

    PR_Lock(ServerStateCVLock);

    if (debug_mode) DPRINTF("\t%s waiting for state %d\n", waiter, state);

    while(!(ServerState & state))
        PR_WaitCondVar(ServerStateCV, PR_INTERVAL_NO_TIMEOUT);
    rv = ServerState;

    if (debug_mode) DPRINTF("\t%s resuming from wait for state %d; state now %d\n", 
        waiter, state, ServerState);
    PR_Unlock(ServerStateCVLock);

    return rv;
}

/* --- Server Functions ------------------------------------------- */

PRLock *workerThreadsLock;
PRInt32 workerThreads;
PRInt32 workerThreadsBusy;

void
WorkerThreadFunc(void *_listenSock)
{
    PRFileDesc *listenSock = (PRFileDesc *)_listenSock;
    PRInt32 bytesRead;
    PRInt32 bytesWritten;
    char *dataBuf;
    char *sendBuf;

    if (debug_mode) DPRINTF("\tServer buffer is %d bytes; %d data, %d netaddrs\n",
            _client_data+(2*sizeof(PRNetAddr))+32, _client_data, (2*sizeof(PRNetAddr))+32);
    dataBuf = (char *)PR_MALLOC(_client_data + 2*sizeof(PRNetAddr) + 32);
    if (!dataBuf)
        if (debug_mode) printf("\tServer could not malloc space!?\n");
    sendBuf = (char *)PR_MALLOC(_server_data *sizeof(char));
    if (!sendBuf)
        if (debug_mode) printf("\tServer could not malloc space!?\n");

    if (debug_mode) DPRINTF("\tServer worker thread running\n");

    while(1) {
        PRInt32 bytesToRead = _client_data;
        PRInt32 bytesToWrite = _server_data;
        PRFileDesc *newSock;
        PRNetAddr *rAddr;
        PRInt32 loops = 0;

        loops++;

        if (debug_mode) DPRINTF("\tServer thread going into accept\n");

        bytesRead = PR_AcceptRead(listenSock, 
                                  &newSock,
                                  &rAddr,
                                  dataBuf,
                                  bytesToRead,
                                  PR_INTERVAL_NO_TIMEOUT);

        if (bytesRead < 0) {
            if (debug_mode) printf("\tServer error in accept (%d)\n", bytesRead);
            continue;
        }

        if (debug_mode) DPRINTF("\tServer accepted connection (%d bytes)\n", bytesRead);
        
        PR_AtomicIncrement(&workerThreadsBusy);
        if (workerThreadsBusy == workerThreads) {

            PR_Lock(workerThreadsLock);
            if (workerThreadsBusy == workerThreads) {
                PRThread *WorkerThread;

                WorkerThread = PR_CreateThread(
                                  PR_SYSTEM_THREAD,
                                  WorkerThreadFunc,
                                  listenSock,
                                  PR_PRIORITY_NORMAL,
                                  ServerScope,
                                  PR_UNJOINABLE_THREAD,
                                  STACKSIZE);

                if (!WorkerThread)
                    if (debug_mode) printf("Error creating client thread %d\n", workerThreads);
                else {
                    PR_AtomicIncrement(&workerThreads);
                    if (debug_mode) DPRINTF("\tServer creates worker (%d)\n", workerThreads);
                }
            }
            PR_Unlock(workerThreadsLock);
        }
 
        bytesToRead -= bytesRead;
        while (bytesToRead) {
            bytesRead = PR_Recv(newSock, 
                                dataBuf, 
                                bytesToRead, 
                                0, 
                                PR_INTERVAL_NO_TIMEOUT);
            if (bytesRead < 0) {
                if (debug_mode) printf("\tServer error receiving data (%d)\n", bytesRead);
                continue;
            }
            if (debug_mode) DPRINTF("\tServer received %d bytes\n", bytesRead);
        }

        bytesWritten = PR_Send(newSock,
                               sendBuf, 
                               bytesToWrite, 
                               0, 
                               PR_INTERVAL_NO_TIMEOUT);
        if (bytesWritten != _server_data)
            if (debug_mode) printf("\tError sending data to client (%d, %d)\n", 
                bytesWritten, PR_GetOSError());
        else
            if (debug_mode) DPRINTF("\tServer sent %d bytes\n", bytesWritten);

        PR_Close(newSock);
        PR_AtomicDecrement(&workerThreadsBusy);
    }
}

PRFileDesc *
ServerSetup(void)
{
    PRFileDesc *listenSocket;
    PRSocketOptionData sockOpt;
    PRNetAddr serverAddr;
    PRThread *WorkerThread;

    if ( (listenSocket = PR_NewTCPSocket()) == NULL) {
        if (debug_mode) printf("\tServer error creating listen socket\n");
		else failed_already=1;
        return NULL;
    }

    sockOpt.option = PR_SockOpt_Reuseaddr;
    sockOpt.value.reuse_addr = PR_TRUE;
    if ( PR_SetSocketOption(listenSocket, &sockOpt) == PR_FAILURE) {
        if (debug_mode) printf("\tServer error setting socket option: OS error %d\n",
                PR_GetOSError());
		else failed_already=1;
        PR_Close(listenSocket);
        return NULL;
    }

    memset(&serverAddr, 0, sizeof(PRNetAddr));
    serverAddr.inet.family = PR_AF_INET;
    serverAddr.inet.port = PR_htons(PORT);
    serverAddr.inet.ip = PR_htonl(PR_INADDR_ANY);

    if ( PR_Bind(listenSocket, &serverAddr) == PR_FAILURE) {
        if (debug_mode) printf("\tServer error binding to server address: OS error %d\n",
                PR_GetOSError());
		else failed_already=1;
        PR_Close(listenSocket);
        return NULL;
    }

    if ( PR_Listen(listenSocket, 128) == PR_FAILURE) {
        if (debug_mode) printf("\tServer error listening to server socket\n");
		else failed_already=1;
        PR_Close(listenSocket);

        return NULL;
    }

    /* Create Clients */
    workerThreads = 0;
    workerThreadsBusy = 0;

    workerThreadsLock = PR_NewLock();

    WorkerThread = PR_CreateThread(
                      PR_SYSTEM_THREAD,
                      WorkerThreadFunc,
                      listenSocket,
                      PR_PRIORITY_NORMAL,
                      ServerScope,
                      PR_UNJOINABLE_THREAD,
                      STACKSIZE);