socket.c

Netscape NSPR库源码

 *    Server Thread
 *    Bind an address to a socket and listen for incoming connections
 *    Create worker threads to service clients
 */
static void
TransmitFile_Server(void *arg)
{
    PRThread **t = NULL;  /* an array of PRThread pointers */
    Server_Param *sp = (Server_Param *) arg;
    Serve_Client_Param *scp;
    PRFileDesc *sockfd = NULL, *newsockfd;
    PRNetAddr netaddr;
    PRInt32 i;

    t = (PRThread**)PR_MALLOC(num_transmitfile_clients * sizeof(PRThread *));
    if (t == NULL) {
        fprintf(stderr, "prsocket_test: run out of memory\n");
        failed_already=1;
        goto exit;
    }
    /*
     * Create a tcp socket
     */
    if ((sockfd = PR_OpenTCPSocket(PR_AF_INET)) == NULL) {
        fprintf(stderr,"prsocket_test: PR_OpenTCPSocket failed\n");
        failed_already=1;
        goto exit;
    }
    memset(&netaddr, 0 , sizeof(netaddr));
    netaddr.inet.family = PR_AF_INET;
    netaddr.inet.port = PR_htons(TCP_SERVER_PORT);
    netaddr.inet.ip = PR_htonl(PR_INADDR_ANY);
    /*
     * try a few times to bind server's address, if addresses are in
     * use
     */
    i = 0;
    while (PR_Bind(sockfd, &netaddr) < 0) {
        if (PR_GetError() == PR_ADDRESS_IN_USE_ERROR) {
            netaddr.inet.port += 2;
            if (i++ < SERVER_MAX_BIND_COUNT)
                continue;
        }
        fprintf(stderr,"prsocket_test: ERROR - PR_Bind failed\n");
        failed_already=1;
        perror("PR_Bind");
        goto exit;
    }

    if (PR_Listen(sockfd, 32) < 0) {
        fprintf(stderr,"prsocket_test: ERROR - PR_Listen failed\n");
        failed_already=1;
        goto exit;
    }

    if (PR_GetSockName(sockfd, &netaddr) < 0) {
        fprintf(stderr,
            "prsocket_test: ERROR - PR_GetSockName failed\n");
        failed_already=1;
        goto exit;
    }

    DPRINTF(("TCP_Server: PR_BIND netaddr.inet.ip = 0x%lx, netaddr.inet.port = %d\n",
        netaddr.inet.ip, netaddr.inet.port));
    tcp_server_addr.inet.family = netaddr.inet.family;
    tcp_server_addr.inet.port = netaddr.inet.port;
    tcp_server_addr.inet.ip = netaddr.inet.ip;

    /*
     * Wake up parent thread because server address is bound and made
     * available in the global variable 'tcp_server_addr'
     */
    PR_PostSem(sp->addr_sem);

    for (i = 0; i < num_transmitfile_clients ; i++) {
        /* test both null and non-null 'addr' argument to PR_Accept */
        PRNetAddr *addrp = (i%2 ? &netaddr: NULL);

        if ((newsockfd = PR_Accept(sockfd, addrp,
            PR_INTERVAL_NO_TIMEOUT)) == NULL) {
            fprintf(stderr,
                "prsocket_test: ERROR - PR_Accept failed\n");
            failed_already=1;
            goto exit;
        }
        /* test both regular and emulated PR_SendFile */
        if (i%2) {
            PRFileDesc *layer = PR_CreateIOLayerStub(
                emuSendFileIdentity, &emuSendFileMethods);
            if (layer == NULL) {
                fprintf(stderr,
                    "prsocket_test: ERROR - PR_CreateIOLayerStub failed\n");
                failed_already=1;
                goto exit;
            }
            if (PR_PushIOLayer(newsockfd, PR_TOP_IO_LAYER, layer)
                    == PR_FAILURE) {
                fprintf(stderr,
                    "prsocket_test: ERROR - PR_PushIOLayer failed\n");
                failed_already=1;
                goto exit;
            }
        }
        scp = PR_NEW(Serve_Client_Param);
        if (scp == NULL) {
            fprintf(stderr,"prsocket_test: PR_NEW failed\n");
            failed_already=1;
            goto exit;
        }

        /*
         * Start a Serve_Client thread for each incoming connection
         */
        scp->sockfd = newsockfd;
        scp->datalen = sp->datalen;

        t[i] = PR_CreateThread(PR_USER_THREAD,
            Serve_TransmitFile_Client, (void *)scp, 
            PR_PRIORITY_NORMAL,
            PR_LOCAL_THREAD,
            PR_JOINABLE_THREAD,
            0);
        if (t[i] == NULL) {
            fprintf(stderr,
                "prsocket_test: PR_CreateThread failed\n");
            failed_already=1;
            goto exit;
        }
        DPRINTF(("TransmitFile_Server: Created Serve_TransmitFile_Client = 0x%lx\n", t));
    }

    /*
     * Wait for all the worker threads to end, so that we know
     * they are no longer using the small and large file fd's.
     */

    for (i = 0; i < num_transmitfile_clients; i++) {
        PR_JoinThread(t[i]);
    }

exit:
    if (t) {
        PR_DELETE(t);
    }
    if (sockfd) {
        PR_Close(sockfd);
    }

    /*
     * Decrement exit_counter and notify parent thread
     */

    PR_EnterMonitor(sp->exit_mon);
    --(*sp->exit_counter);
    PR_Notify(sp->exit_mon);
    PR_ExitMonitor(sp->exit_mon);
    DPRINTF(("TransmitFile_Server [0x%lx] exiting\n", PR_GetCurrentThread()));
}

/*
 * Socket_Misc_Test    - test miscellaneous functions 
 *    
 */
static PRInt32
Socket_Misc_Test(void)
{
    PRIntn i, rv = 0, bytes, count, len;
    PRThread *t;
    PRSemaphore *server_sem;
    Server_Param *sparamp;
    Client_Param *cparamp;
    PRMonitor *mon2;
    PRInt32    datalen;

    /*
 * We deliberately pick a buffer size that is not a nice multiple
 * of 1024.
 */
#define TRANSMITFILE_BUF_SIZE    (4 * 1024 - 11)

    typedef struct {
        char    data[TRANSMITFILE_BUF_SIZE];
    } file_buf;
    file_buf *buf = NULL;

    /*
     * create file(s) to be transmitted
     */
    if ((PR_MkDir(TEST_DIR, 0777)) < 0) {
        printf("prsocket_test failed to create dir %s\n",TEST_DIR);
        failed_already=1;
        return -1;
    }

    small_file_fd = PR_Open(SMALL_FILE_NAME, PR_RDWR | PR_CREATE_FILE,0777);

    if (small_file_fd == NULL) {
        fprintf(stderr,"prsocket_test failed to create/open file %s\n",
            SMALL_FILE_NAME);
        failed_already=1;
        rv = -1;
        goto done;
    }
    buf = PR_NEW(file_buf);
    if (buf == NULL) {
        fprintf(stderr,"prsocket_test failed to allocate buffer\n");
        failed_already=1;
        rv = -1;
        goto done;
    }
    /*
     * fill in random data
     */
    for (i = 0; i < TRANSMITFILE_BUF_SIZE; i++) {
        buf->data[i] = i;
    }
    count = 0;
    do {
        len = (SMALL_FILE_SIZE - count) > TRANSMITFILE_BUF_SIZE ?
            TRANSMITFILE_BUF_SIZE : (SMALL_FILE_SIZE - count);
        bytes = PR_Write(small_file_fd, buf->data, len);
        if (bytes <= 0) {
            fprintf(stderr,
                "prsocket_test failed to write to file %s\n",
                SMALL_FILE_NAME);
            failed_already=1;
            rv = -1;
            goto done;
        }
        count += bytes;
    } while (count < SMALL_FILE_SIZE);
#ifdef XP_UNIX
    /*
     * map the small file; used in checking for data corruption
     */
    small_file_addr = mmap(0, SMALL_FILE_SIZE, PROT_READ,
        MAP_SHARED, small_file_fd->secret->md.osfd, 0);
    if (small_file_addr == (void *) -1) {
        fprintf(stderr,"prsocket_test failed to mmap file %s\n",
            SMALL_FILE_NAME);
        failed_already=1;
        rv = -1;
        goto done;
    }
#endif
    /*
     * header for small file
     */
    small_file_header = PR_MALLOC(SMALL_FILE_HEADER_SIZE);
    if (small_file_header == NULL) {
        fprintf(stderr,"prsocket_test failed to malloc header file\n");
        failed_already=1;
        rv = -1;
        goto done;
    }
    memset(small_file_header, (int) PR_IntervalNow(),
        SMALL_FILE_HEADER_SIZE);
    /*
     * trailer for small file
     */
    small_file_trailer = PR_MALLOC(SMALL_FILE_TRAILER_SIZE);
    if (small_file_trailer == NULL) {
        fprintf(stderr,"prsocket_test failed to malloc header trailer\n");
        failed_already=1;
        rv = -1;
        goto done;
    }
    memset(small_file_trailer, (int) PR_IntervalNow(),
        SMALL_FILE_TRAILER_SIZE);
    /*
     * setup large file
     */
    large_file_fd = PR_Open(LARGE_FILE_NAME, PR_RDWR | PR_CREATE_FILE,0777);

    if (large_file_fd == NULL) {
        fprintf(stderr,"prsocket_test failed to create/open file %s\n",
            LARGE_FILE_NAME);
        failed_already=1;
        rv = -1;
        goto done;
    }
    /*
     * fill in random data
     */
    for (i = 0; i < TRANSMITFILE_BUF_SIZE; i++) {
        buf->data[i] = i;
    }
    count = 0;
    do {
        len = (LARGE_FILE_SIZE - count) > TRANSMITFILE_BUF_SIZE ?
            TRANSMITFILE_BUF_SIZE : (LARGE_FILE_SIZE - count);
        bytes = PR_Write(large_file_fd, buf->data, len);
        if (bytes <= 0) {
            fprintf(stderr,
                "prsocket_test failed to write to file %s: (%ld, %ld)\n",
                LARGE_FILE_NAME,
                PR_GetError(), PR_GetOSError());
            failed_already=1;
            rv = -1;
            goto done;
        }
        count += bytes;
    } while (count < LARGE_FILE_SIZE);
#ifdef XP_UNIX
    /*
     * map the large file; used in checking for data corruption
     */
    large_file_addr = mmap(0, LARGE_FILE_SIZE, PROT_READ,
        MAP_SHARED, large_file_fd->secret->md.osfd, 0);
    if (large_file_addr == (void *) -1) {
        fprintf(stderr,"prsocket_test failed to mmap file %s\n",
            LARGE_FILE_NAME);
        failed_already=1;
        rv = -1;
        goto done;
    }
#endif
    /*
     * header for large file
     */
    large_file_header = PR_MALLOC(LARGE_FILE_HEADER_SIZE);
    if (large_file_header == NULL) {
        fprintf(stderr,"prsocket_test failed to malloc header file\n");
        failed_already=1;
        rv = -1;
        goto done;
    }
    memset(large_file_header, (int) PR_IntervalNow(),
        LARGE_FILE_HEADER_SIZE);
    /*
     * trailer for large file
     */
    large_file_trailer = PR_MALLOC(LARGE_FILE_TRAILER_SIZE);
    if (large_file_trailer == NULL) {
        fprintf(stderr,"prsocket_test failed to malloc header trailer\n");
        failed_already=1;
        rv = -1;
        goto done;
    }
    memset(large_file_trailer, (int) PR_IntervalNow(),
        LARGE_FILE_TRAILER_SIZE);

    datalen = tcp_mesg_size;
    thread_count = 0;
    /*
     * start the server thread
     */
    sparamp = PR_NEW(Server_Param);
    if (sparamp == NULL) {
        fprintf(stderr,"prsocket_test: PR_NEW failed\n");
        failed_already=1;
        rv = -1;
        goto done;
    }
    server_sem = PR_NewSem(0);
    if (server_sem == NULL) {
        fprintf(stderr,"prsocket_test: PR_NewSem failed\n");
        failed_already=1;
        rv = -1;
        goto done;
    }
    mon2 = PR_NewMonitor();
    if (mon2 == NULL) {
        fprintf(stderr,"prsocket_test: PR_NewMonitor failed\n");
        failed_already=1;
        rv = -1;
        goto done;
    }
    PR_EnterMonitor(mon2);

    sparamp->addr_sem = server_sem;
    sparamp->exit_mon = mon2;
    sparamp->exit_counter = &thread_count;
    sparamp->datalen = datalen;
    t = PR_CreateThread(PR_USER_THREAD,
        TransmitFile_Server, (void *)sparamp, 
        PR_PRIORITY_NORMAL,
        PR_LOCAL_THREAD,
        PR_UNJOINABLE_THREAD,
        0);
    if (t == NULL) {
        fprintf(stderr,"prsocket_test: PR_CreateThread failed\n");
        failed_already=1;
        rv = -1;
        goto done;
    }
    DPRINTF(("Created TCP server = 0x%x\n", t));
    thread_count++;

    /*
     * wait till the server address is setup
     */
    PR_WaitSem(server_sem);

    /*
     * Now start a bunch of client threads
     */

    cparamp = PR_NEW(Client_Param);
    if (cparamp == NULL) {
        fprintf(stderr,"prsocket_test: PR_NEW failed\n");
        failed_already=1;
        rv = -1;
        goto done;
    }
    cparamp->server_addr = tcp_server_addr;
    cparamp->server_addr.inet.ip = PR_htonl(PR_INADDR_LOOPBACK);
    cparamp->exit_mon = mon2;
    cparamp->exit_counter = &thread_count;
    cparamp->datalen = datalen;
    for (i = 0; i < num_transmitfile_clients; i++) {
        t = create_new_thread(PR_USER_THREAD,
            TransmitFile_Client, (void *) cparamp,
            PR_PRIORITY_NORMAL,
            PR_LOCAL_THREAD,
            PR_UNJOINABLE_THREAD,
            0, i);
        if (t == NULL) {
            fprintf(stderr,"prsocket_test: PR_CreateThread failed\n");
            rv = -1;
            failed_already=1;
            goto done;
        }
        DPRINTF(("Created TransmitFile client = 0x%lx\n", t));
        thread_count++;
    }
    /* Wait for server and client threads to exit */
    while (thread_count) {
        PR_Wait(mon2, PR_INTERVAL_NO_TIMEOUT);
        DPRINTF(("Socket_Misc_Test - thread_count  = %d\n", t