Back to index

lightning-sunbird  0.9+nobinonly
Classes | Defines | Typedefs | Functions | Variables
thruput.c File Reference
#include "prerror.h"
#include "prinrval.h"
#include "prinit.h"
#include "prio.h"
#include "prlock.h"
#include "prmem.h"
#include "prnetdb.h"
#include "prprf.h"
#include "prthread.h"
#include "pprio.h"
#include "plerror.h"
#include "plgetopt.h"

Go to the source code of this file.

Classes

struct  Shared

Defines

#define ADDR_BUFFER   100
#define PORT_NUMBER   51877
#define SAMPLING_INTERVAL   10
#define BUFFER_SIZE   (32 * 1024)

Typedefs

typedef struct Shared Shared

Functions

static PRStatus PrintAddress (const PRNetAddr *address)
static void PR_CALLBACK Clientel (void *arg)
static void Client (const char *server_name)
static void PR_CALLBACK Servette (void *arg)
static void Server (void)
static void Help (void)
PRIntn main (PRIntn argc, char **argv)

Variables

static PRInt32 domain = PR_AF_INET
static PRInt32 protocol = 6
static PRFileDescerr = NULL
static PRIntn concurrency = 1
static PRInt32 xport_buffer = -1
static PRUint32 initial_streams = 1
static PRInt32 buffer_size = BUFFER_SIZE
static PRThreadScope thread_scope = PR_LOCAL_THREAD
static Sharedshared = NULL

Define Documentation

#define ADDR_BUFFER   100

Definition at line 63 of file thruput.c.

#define BUFFER_SIZE   (32 * 1024)

Definition at line 66 of file thruput.c.

#define PORT_NUMBER   51877

Definition at line 64 of file thruput.c.

Definition at line 65 of file thruput.c.


Typedef Documentation

typedef struct Shared Shared

Function Documentation

static void Client ( const char *  server_name) [static]

Definition at line 176 of file thruput.c.

{
    PRStatus rv;
    PRHostEnt host;
    char buffer[PR_NETDB_BUF_SIZE];
    PRIntervalTime dally = PR_SecondsToInterval(60);
    PR_fprintf(err, "Translating the name %s\n", server_name);
    rv = PR_GetHostByName(server_name, buffer, sizeof(buffer), &host);
    if (PR_FAILURE == rv)
        PL_FPrintError(err, "PR_GetHostByName");
    else
    {
        if (PR_EnumerateHostEnt(
            0, &host, PORT_NUMBER, &shared->server_address) < 0)
            PL_FPrintError(err, "PR_EnumerateHostEnt");
        else
        {
            do
            {
                shared->threads += 1;
                (void)PR_CreateThread(
                    PR_USER_THREAD, Clientel, shared,
                    PR_PRIORITY_NORMAL, thread_scope,
                    PR_UNJOINABLE_THREAD, 8 * 1024);
                if (shared->threads == initial_streams)
                {
                    PR_Sleep(dally);
                    initial_streams += 1;
                }
            } while (PR_TRUE);
        }
    }
}

Here is the call graph for this function:

static void PR_CALLBACK Clientel ( void arg) [static]

Definition at line 99 of file thruput.c.

{
    PRStatus rv;
    PRFileDesc *xport;
    PRInt32 bytes, sampled;
    PRIntervalTime now, interval;
    PRBool do_display = PR_FALSE;
    Shared *shared = (Shared*)arg;
    char *buffer = (char*)PR_Malloc(buffer_size);
    PRNetAddr *server_address = &shared->server_address;
    PRIntervalTime connect_timeout = PR_SecondsToInterval(5);
    PRIntervalTime sampling_interval = PR_SecondsToInterval(SAMPLING_INTERVAL);

    PR_fprintf(err, "Client connecting to ");
    (void)PrintAddress(server_address);

    do
    {
        xport = PR_Socket(domain, PR_SOCK_STREAM, protocol);
        if (NULL == xport)
        {
            PL_FPrintError(err, "PR_Socket");
            return;
        }

        if (xport_buffer != -1)
        {
            PRSocketOptionData data;
            data.option = PR_SockOpt_RecvBufferSize;
            data.value.recv_buffer_size = (PRSize)xport_buffer;
            rv = PR_SetSocketOption(xport, &data);
            if (PR_FAILURE == rv)
                PL_FPrintError(err, "PR_SetSocketOption - ignored");
            data.option = PR_SockOpt_SendBufferSize;
            data.value.send_buffer_size = (PRSize)xport_buffer;
            rv = PR_SetSocketOption(xport, &data);
            if (PR_FAILURE == rv)
                PL_FPrintError(err, "PR_SetSocketOption - ignored");
        }

        rv = PR_Connect(xport, server_address, connect_timeout);
        if (PR_FAILURE == rv)
        {
            PL_FPrintError(err, "PR_Connect");
            if (PR_IO_TIMEOUT_ERROR != PR_GetError())
                PR_Sleep(connect_timeout);
            PR_Close(xport);  /* delete it and start over */
        }
    } while (PR_FAILURE == rv);

    do
    {
        bytes = PR_Recv(
            xport, buffer, buffer_size, 0, PR_INTERVAL_NO_TIMEOUT);
        PR_Lock(shared->ml);
        now = PR_IntervalNow();
        shared->sampled += bytes;
        interval = now - shared->timein;
        if (interval > sampling_interval)
        {
            sampled = shared->sampled;
            shared->timein = now;
            shared->sampled = 0;
            do_display = PR_TRUE;
        }
        PR_Unlock(shared->ml);

        if (do_display)
        {
            PRUint32 rate = sampled / PR_IntervalToMilliseconds(interval);
            PR_fprintf(err, "%u streams @ %u Kbytes/sec\n", shared->threads, rate);
            do_display = PR_FALSE;
        }

    } while (bytes > 0);
}  /* Clientel */

Here is the call graph for this function:

Here is the caller graph for this function:

static void Help ( void  ) [static]

Definition at line 307 of file thruput.c.

{
    PR_fprintf(err, "Usage: [-h] [<server>]\n");
    PR_fprintf(err, "\t-s <n>   Initial # of connections        (default: 1)\n");
    PR_fprintf(err, "\t-C <n>   Set 'concurrency'               (default: 1)\n");
    PR_fprintf(err, "\t-b <nK>  Client buffer size              (default: 32k)\n");
    PR_fprintf(err, "\t-B <nK>  Transport recv/send buffer size (default: sys)\n");
    PR_fprintf(err, "\t-G       Use GLOBAL threads              (default: LOCAL)\n");
    PR_fprintf(err, "\t-X       Use XTP transport               (default: TCP)\n");
    PR_fprintf(err, "\t-6       Use IPv6                        (default: IPv4)\n");
    PR_fprintf(err, "\t-h       This message and nothing else\n");
    PR_fprintf(err, "\t<server> DNS name of server\n");
    PR_fprintf(err, "\t\tIf <server> is not specified, this host will be\n");
    PR_fprintf(err, "\t\tthe server and not act as a client.\n");
}  /* Help */
PRIntn main ( PRIntn  argc,
char **  argv 
)

Definition at line 323 of file thruput.c.

{
    PLOptStatus os;
    const char *server_name = NULL;
    PLOptState *opt = PL_CreateOptState(argc, argv, "hGX6C:b:s:B:");

    err = PR_GetSpecialFD(PR_StandardError);

    while (PL_OPT_EOL != (os = PL_GetNextOpt(opt)))
    {
        if (PL_OPT_BAD == os) continue;
        switch (opt->option)
        {
        case 0:  /* Name of server */
            server_name = opt->value;
            break;
        case 'G':  /* Globular threads */
            thread_scope = PR_GLOBAL_THREAD;
            break;
        case 'X':  /* Use XTP as the transport */
            protocol = 36;
            break;
        case '6':  /* Use IPv6 */
            domain = PR_AF_INET6;
            break;
        case 's':  /* initial_streams */
            initial_streams = atoi(opt->value);
            break;
        case 'C':  /* concurrency */
            concurrency = atoi(opt->value);
            break;
        case 'b':  /* buffer size */
            buffer_size = 1024 * atoi(opt->value);
            break;
        case 'B':  /* buffer size */
            xport_buffer = 1024 * atoi(opt->value);
            break;
        case 'h':  /* user wants some guidance */
        default:
            Help();  /* so give him an earful */
            return 2;  /* but not a lot else */
        }
    }
    PL_DestroyOptState(opt);

    shared = PR_NEWZAP(Shared);
    shared->ml = PR_NewLock();

    PR_fprintf(err,
        "This machine is %s\n",
        (NULL == server_name) ? "the SERVER" : "a CLIENT");

    PR_fprintf(err,
        "Transport being used is %s\n",
        (6 == protocol) ? "TCP" : "XTP");

    if (PR_GLOBAL_THREAD == thread_scope)
    {
        if (1 != concurrency)
        {
            PR_fprintf(err, "  **Concurrency > 1 and GLOBAL threads!?!?\n");
            PR_fprintf(err, "  **Ignoring concurrency\n");
            concurrency = 1;
        }
    }

    if (1 != concurrency)
    {
        PR_SetConcurrency(concurrency);
        PR_fprintf(err, "Concurrency set to %u\n", concurrency);
    }

    PR_fprintf(err,
        "All threads will be %s\n",
        (PR_GLOBAL_THREAD == thread_scope) ? "GLOBAL" : "LOCAL");

    PR_fprintf(err, "Client buffer size will be %u\n", buffer_size);
   
    if (-1 != xport_buffer)
    PR_fprintf(
        err, "Transport send & receive buffer size will be %u\n", xport_buffer);
    

    if (NULL == server_name) Server();
    else Client(server_name);

}  /* main */

Here is the call graph for this function:

static PRStatus PrintAddress ( const PRNetAddr address) [static]

Definition at line 88 of file thruput.c.

{
    char buffer[ADDR_BUFFER];
    PRStatus rv = PR_NetAddrToString(address, buffer, sizeof(buffer));
    if (PR_SUCCESS == rv)
        PR_fprintf(err, "%s:%u\n", buffer, PR_ntohs(address->inet.port));
    else PL_FPrintError(err, "PR_NetAddrToString");
    return rv;
}  /* PrintAddress */

Here is the caller graph for this function:

static void Server ( void  ) [static]

Definition at line 262 of file thruput.c.

{
    PRStatus rv;
    PRNetAddr server_address, client_address;
    PRFileDesc *xport = PR_Socket(domain, PR_SOCK_STREAM, protocol);

    if (NULL == xport)
    {
        PL_FPrintError(err, "PR_Socket");
        return;
    }

    rv = PR_InitializeNetAddr(PR_IpAddrAny, PORT_NUMBER, &server_address);
    if (PR_FAILURE == rv) PL_FPrintError(err, "PR_InitializeNetAddr");
    else
    {
        rv = PR_Bind(xport, &server_address);
        if (PR_FAILURE == rv) PL_FPrintError(err, "PR_Bind");
        else
        {
            PRFileDesc *client;
            rv = PR_Listen(xport, 10);
            PR_fprintf(err, "Server listening on ");
            (void)PrintAddress(&server_address);
            do
            {
                client = PR_Accept(
                    xport, &client_address, PR_INTERVAL_NO_TIMEOUT);
                if (NULL == client) PL_FPrintError(err, "PR_Accept");
                else
                {
                    PR_fprintf(err, "Server accepting from ");
                    (void)PrintAddress(&client_address);
                    shared->threads += 1;
                    (void)PR_CreateThread(
                        PR_USER_THREAD, Servette, client,
                        PR_PRIORITY_NORMAL, thread_scope,
                        PR_UNJOINABLE_THREAD, 8 * 1024);
                }
            } while (PR_TRUE);

        }
    }
}  /* Server */

Here is the call graph for this function:

Here is the caller graph for this function:

static void PR_CALLBACK Servette ( void arg) [static]

Definition at line 210 of file thruput.c.

{
    PRInt32 bytes, sampled;
    PRIntervalTime now, interval;
    PRBool do_display = PR_FALSE;
    PRFileDesc *client = (PRFileDesc*)arg;
    char *buffer = (char*)PR_Malloc(buffer_size);
    PRIntervalTime sampling_interval = PR_SecondsToInterval(SAMPLING_INTERVAL);

    if (xport_buffer != -1)
    {
        PRStatus rv;
        PRSocketOptionData data;
        data.option = PR_SockOpt_RecvBufferSize;
        data.value.recv_buffer_size = (PRSize)xport_buffer;
        rv = PR_SetSocketOption(client, &data);
        if (PR_FAILURE == rv)
            PL_FPrintError(err, "PR_SetSocketOption - ignored");
        data.option = PR_SockOpt_SendBufferSize;
        data.value.send_buffer_size = (PRSize)xport_buffer;
        rv = PR_SetSocketOption(client, &data);
        if (PR_FAILURE == rv)
            PL_FPrintError(err, "PR_SetSocketOption - ignored");
    }

    do
    {
        bytes = PR_Send(
            client, buffer, buffer_size, 0, PR_INTERVAL_NO_TIMEOUT);

        PR_Lock(shared->ml);
        now = PR_IntervalNow();
        shared->sampled += bytes;
        interval = now - shared->timein;
        if (interval > sampling_interval)
        {
            sampled = shared->sampled;
            shared->timein = now;
            shared->sampled = 0;
            do_display = PR_TRUE;
        }
        PR_Unlock(shared->ml);

        if (do_display)
        {
            PRUint32 rate = sampled / PR_IntervalToMilliseconds(interval);
            PR_fprintf(err, "%u streams @ %u Kbytes/sec\n", shared->threads, rate);
            do_display = PR_FALSE;
        }
    } while (bytes > 0);
}  /* Servette */

Here is the caller graph for this function:


Variable Documentation

Definition at line 74 of file thruput.c.

PRIntn concurrency = 1 [static]

Definition at line 71 of file thruput.c.

Definition at line 68 of file thruput.c.

PRFileDesc* err = NULL [static]

Definition at line 70 of file thruput.c.

PRUint32 initial_streams = 1 [static]

Definition at line 73 of file thruput.c.

PRInt32 protocol = 6 [static]

Definition at line 69 of file thruput.c.

Shared* shared = NULL [static]

Definition at line 86 of file thruput.c.

Definition at line 75 of file thruput.c.

PRInt32 xport_buffer = -1 [static]

Definition at line 72 of file thruput.c.