michael@0: /* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 2 -*- */ michael@0: /* This Source Code Form is subject to the terms of the Mozilla Public michael@0: * License, v. 2.0. If a copy of the MPL was not distributed with this michael@0: * file, You can obtain one at http://mozilla.org/MPL/2.0/. */ michael@0: michael@0: /*********************************************************************** michael@0: ** michael@0: ** Name: thrpool.c michael@0: ** michael@0: ** Description: Test threadpool functionality. michael@0: ** michael@0: ** Modification History: michael@0: */ michael@0: #include "primpl.h" michael@0: michael@0: #include "plgetopt.h" michael@0: michael@0: #include michael@0: #include michael@0: #include michael@0: #ifdef XP_UNIX michael@0: #include michael@0: #endif michael@0: #if defined(_PR_PTHREADS) && !defined(_PR_DCETHREADS) michael@0: #include michael@0: #endif michael@0: michael@0: /* for getcwd */ michael@0: #if defined(XP_UNIX) || defined (XP_OS2) || defined(XP_BEOS) michael@0: #include michael@0: #elif defined(XP_PC) michael@0: #include michael@0: #endif michael@0: michael@0: #ifdef WIN32 michael@0: #include michael@0: #endif michael@0: michael@0: static int _debug_on = 0; michael@0: static char *program_name = NULL; michael@0: static void serve_client_write(void *arg); michael@0: michael@0: #include "obsolete/prsem.h" michael@0: michael@0: #ifdef XP_PC michael@0: #define mode_t int michael@0: #endif michael@0: michael@0: #define DPRINTF(arg) if (_debug_on) printf arg michael@0: michael@0: michael@0: #define BUF_DATA_SIZE (2 * 1024) michael@0: #define TCP_MESG_SIZE 1024 michael@0: #define NUM_TCP_CLIENTS 10 /* for a listen queue depth of 5 */ michael@0: michael@0: michael@0: #define NUM_TCP_CONNECTIONS_PER_CLIENT 10 michael@0: #define NUM_TCP_MESGS_PER_CONNECTION 10 michael@0: #define TCP_SERVER_PORT 10000 michael@0: #define SERVER_MAX_BIND_COUNT 100 michael@0: michael@0: #ifdef WINCE michael@0: char *getcwd(char *buf, size_t size) michael@0: { michael@0: wchar_t wpath[MAX_PATH]; michael@0: _wgetcwd(wpath, MAX_PATH); michael@0: WideCharToMultiByte(CP_ACP, 0, wpath, -1, buf, size, 0, 0); michael@0: } michael@0: michael@0: #define perror(s) michael@0: #endif michael@0: michael@0: static PRInt32 num_tcp_clients = NUM_TCP_CLIENTS; michael@0: static PRInt32 num_tcp_connections_per_client = NUM_TCP_CONNECTIONS_PER_CLIENT; michael@0: static PRInt32 tcp_mesg_size = TCP_MESG_SIZE; michael@0: static PRInt32 num_tcp_mesgs_per_connection = NUM_TCP_MESGS_PER_CONNECTION; michael@0: static void TCP_Server_Accept(void *arg); michael@0: michael@0: michael@0: int failed_already=0; michael@0: typedef struct buffer { michael@0: char data[BUF_DATA_SIZE]; michael@0: } buffer; michael@0: michael@0: michael@0: typedef struct Server_Param { michael@0: PRJobIoDesc iod; /* socket to read from/write to */ michael@0: PRInt32 datalen; /* bytes of data transfered in each read/write */ michael@0: PRNetAddr netaddr; michael@0: PRMonitor *exit_mon; /* monitor to signal on exit */ michael@0: PRInt32 *job_counterp; /* counter to decrement, before exit */ michael@0: PRInt32 conn_counter; /* counter to decrement, before exit */ michael@0: PRThreadPool *tp; michael@0: } Server_Param; michael@0: michael@0: typedef struct Serve_Client_Param { michael@0: PRJobIoDesc iod; /* socket to read from/write to */ michael@0: PRInt32 datalen; /* bytes of data transfered in each read/write */ michael@0: PRMonitor *exit_mon; /* monitor to signal on exit */ michael@0: PRInt32 *job_counterp; /* counter to decrement, before exit */ michael@0: PRThreadPool *tp; michael@0: } Serve_Client_Param; michael@0: michael@0: typedef struct Session { michael@0: PRJobIoDesc iod; /* socket to read from/write to */ michael@0: buffer *in_buf; michael@0: PRInt32 bytes; michael@0: PRInt32 msg_num; michael@0: PRInt32 bytes_read; michael@0: PRMonitor *exit_mon; /* monitor to signal on exit */ michael@0: PRInt32 *job_counterp; /* counter to decrement, before exit */ michael@0: PRThreadPool *tp; michael@0: } Session; michael@0: michael@0: static void michael@0: serve_client_read(void *arg) michael@0: { michael@0: Session *sp = (Session *) arg; michael@0: int rem; michael@0: int bytes; michael@0: int offset; michael@0: PRFileDesc *sockfd; michael@0: char *buf; michael@0: PRJob *jobp; michael@0: michael@0: PRIntervalTime timeout = PR_INTERVAL_NO_TIMEOUT; michael@0: michael@0: sockfd = sp->iod.socket; michael@0: buf = sp->in_buf->data; michael@0: michael@0: PR_ASSERT(sp->msg_num < num_tcp_mesgs_per_connection); michael@0: PR_ASSERT(sp->bytes_read < sp->bytes); michael@0: michael@0: offset = sp->bytes_read; michael@0: rem = sp->bytes - offset; michael@0: bytes = PR_Recv(sockfd, buf + offset, rem, 0, timeout); michael@0: if (bytes < 0) { michael@0: return; michael@0: } michael@0: sp->bytes_read += bytes; michael@0: sp->iod.timeout = PR_SecondsToInterval(60); michael@0: if (sp->bytes_read < sp->bytes) { michael@0: jobp = PR_QueueJob_Read(sp->tp, &sp->iod, serve_client_read, sp, michael@0: PR_FALSE); michael@0: PR_ASSERT(NULL != jobp); michael@0: return; michael@0: } michael@0: PR_ASSERT(sp->bytes_read == sp->bytes); michael@0: DPRINTF(("serve_client: read complete, msg(%d) \n", sp->msg_num)); michael@0: michael@0: sp->iod.timeout = PR_SecondsToInterval(60); michael@0: jobp = PR_QueueJob_Write(sp->tp, &sp->iod, serve_client_write, sp, michael@0: PR_FALSE); michael@0: PR_ASSERT(NULL != jobp); michael@0: michael@0: return; michael@0: } michael@0: michael@0: static void michael@0: serve_client_write(void *arg) michael@0: { michael@0: Session *sp = (Session *) arg; michael@0: int bytes; michael@0: PRFileDesc *sockfd; michael@0: char *buf; michael@0: PRJob *jobp; michael@0: michael@0: sockfd = sp->iod.socket; michael@0: buf = sp->in_buf->data; michael@0: michael@0: PR_ASSERT(sp->msg_num < num_tcp_mesgs_per_connection); michael@0: michael@0: bytes = PR_Send(sockfd, buf, sp->bytes, 0, PR_INTERVAL_NO_TIMEOUT); michael@0: PR_ASSERT(bytes == sp->bytes); michael@0: michael@0: if (bytes < 0) { michael@0: return; michael@0: } michael@0: DPRINTF(("serve_client: write complete, msg(%d) \n", sp->msg_num)); michael@0: sp->msg_num++; michael@0: if (sp->msg_num < num_tcp_mesgs_per_connection) { michael@0: sp->bytes_read = 0; michael@0: sp->iod.timeout = PR_SecondsToInterval(60); michael@0: jobp = PR_QueueJob_Read(sp->tp, &sp->iod, serve_client_read, sp, michael@0: PR_FALSE); michael@0: PR_ASSERT(NULL != jobp); michael@0: return; michael@0: } michael@0: michael@0: DPRINTF(("serve_client: read/write complete, msg(%d) \n", sp->msg_num)); michael@0: if (PR_Shutdown(sockfd, PR_SHUTDOWN_BOTH) < 0) { michael@0: fprintf(stderr,"%s: ERROR - PR_Shutdown\n", program_name); michael@0: } michael@0: michael@0: PR_Close(sockfd); michael@0: PR_EnterMonitor(sp->exit_mon); michael@0: --(*sp->job_counterp); michael@0: PR_Notify(sp->exit_mon); michael@0: PR_ExitMonitor(sp->exit_mon); michael@0: michael@0: PR_DELETE(sp->in_buf); michael@0: PR_DELETE(sp); michael@0: michael@0: return; michael@0: } michael@0: michael@0: /* michael@0: * Serve_Client michael@0: * Thread, started by the server, for serving a client connection. michael@0: * Reads data from socket and writes it back, unmodified, and michael@0: * closes the socket michael@0: */ michael@0: static void PR_CALLBACK michael@0: Serve_Client(void *arg) michael@0: { michael@0: Serve_Client_Param *scp = (Serve_Client_Param *) arg; michael@0: buffer *in_buf; michael@0: Session *sp; michael@0: PRJob *jobp; michael@0: michael@0: sp = PR_NEW(Session); michael@0: sp->iod = scp->iod; michael@0: michael@0: in_buf = PR_NEW(buffer); michael@0: if (in_buf == NULL) { michael@0: fprintf(stderr,"%s: failed to alloc buffer struct\n",program_name); michael@0: failed_already=1; michael@0: return; michael@0: } michael@0: michael@0: sp->in_buf = in_buf; michael@0: sp->bytes = scp->datalen; michael@0: sp->msg_num = 0; michael@0: sp->bytes_read = 0; michael@0: sp->tp = scp->tp; michael@0: sp->exit_mon = scp->exit_mon; michael@0: sp->job_counterp = scp->job_counterp; michael@0: michael@0: sp->iod.timeout = PR_SecondsToInterval(60); michael@0: jobp = PR_QueueJob_Read(sp->tp, &sp->iod, serve_client_read, sp, michael@0: PR_FALSE); michael@0: PR_ASSERT(NULL != jobp); michael@0: PR_DELETE(scp); michael@0: } michael@0: michael@0: static void michael@0: print_stats(void *arg) michael@0: { michael@0: Server_Param *sp = (Server_Param *) arg; michael@0: PRThreadPool *tp = sp->tp; michael@0: PRInt32 counter; michael@0: PRJob *jobp; michael@0: michael@0: PR_EnterMonitor(sp->exit_mon); michael@0: counter = (*sp->job_counterp); michael@0: PR_ExitMonitor(sp->exit_mon); michael@0: michael@0: printf("PRINT_STATS: #client connections = %d\n",counter); michael@0: michael@0: michael@0: jobp = PR_QueueJob_Timer(tp, PR_MillisecondsToInterval(500), michael@0: print_stats, sp, PR_FALSE); michael@0: michael@0: PR_ASSERT(NULL != jobp); michael@0: } michael@0: michael@0: static int job_counter = 0; michael@0: /* michael@0: * TCP Server michael@0: * Server binds an address to a socket, starts a client process and michael@0: * listens for incoming connections. michael@0: * Each client connects to the server and sends a chunk of data michael@0: * Starts a Serve_Client job for each incoming connection, to read michael@0: * the data from the client and send it back to the client, unmodified. michael@0: * Each client checks that data received from server is same as the michael@0: * data it sent to the server. michael@0: * Finally, the threadpool is shutdown michael@0: */ michael@0: static void PR_CALLBACK michael@0: TCP_Server(void *arg) michael@0: { michael@0: PRThreadPool *tp = (PRThreadPool *) arg; michael@0: Server_Param *sp; michael@0: PRFileDesc *sockfd; michael@0: PRNetAddr netaddr; michael@0: PRMonitor *sc_mon; michael@0: PRJob *jobp; michael@0: int i; michael@0: PRStatus rval; michael@0: michael@0: /* michael@0: * Create a tcp socket michael@0: */ michael@0: if ((sockfd = PR_NewTCPSocket()) == NULL) { michael@0: fprintf(stderr,"%s: PR_NewTCPSocket failed\n", program_name); michael@0: return; michael@0: } michael@0: memset(&netaddr, 0 , sizeof(netaddr)); michael@0: netaddr.inet.family = PR_AF_INET; michael@0: netaddr.inet.port = PR_htons(TCP_SERVER_PORT); michael@0: netaddr.inet.ip = PR_htonl(PR_INADDR_ANY); michael@0: /* michael@0: * try a few times to bind server's address, if addresses are in michael@0: * use michael@0: */ michael@0: i = 0; michael@0: while (PR_Bind(sockfd, &netaddr) < 0) { michael@0: if (PR_GetError() == PR_ADDRESS_IN_USE_ERROR) { michael@0: netaddr.inet.port += 2; michael@0: if (i++ < SERVER_MAX_BIND_COUNT) michael@0: continue; michael@0: } michael@0: fprintf(stderr,"%s: ERROR - PR_Bind failed\n", program_name); michael@0: perror("PR_Bind"); michael@0: failed_already=1; michael@0: return; michael@0: } michael@0: michael@0: if (PR_Listen(sockfd, 32) < 0) { michael@0: fprintf(stderr,"%s: ERROR - PR_Listen failed\n", program_name); michael@0: failed_already=1; michael@0: return; michael@0: } michael@0: michael@0: if (PR_GetSockName(sockfd, &netaddr) < 0) { michael@0: fprintf(stderr,"%s: ERROR - PR_GetSockName failed\n", program_name); michael@0: failed_already=1; michael@0: return; michael@0: } michael@0: michael@0: DPRINTF(( michael@0: "TCP_Server: PR_BIND netaddr.inet.ip = 0x%lx, netaddr.inet.port = %d\n", michael@0: netaddr.inet.ip, netaddr.inet.port)); michael@0: michael@0: sp = PR_NEW(Server_Param); michael@0: if (sp == NULL) { michael@0: fprintf(stderr,"%s: PR_NEW failed\n", program_name); michael@0: failed_already=1; michael@0: return; michael@0: } michael@0: sp->iod.socket = sockfd; michael@0: sp->iod.timeout = PR_SecondsToInterval(60); michael@0: sp->datalen = tcp_mesg_size; michael@0: sp->exit_mon = sc_mon; michael@0: sp->job_counterp = &job_counter; michael@0: sp->conn_counter = 0; michael@0: sp->tp = tp; michael@0: sp->netaddr = netaddr; michael@0: michael@0: /* create and cancel an io job */ michael@0: jobp = PR_QueueJob_Accept(tp, &sp->iod, TCP_Server_Accept, sp, michael@0: PR_FALSE); michael@0: PR_ASSERT(NULL != jobp); michael@0: rval = PR_CancelJob(jobp); michael@0: PR_ASSERT(PR_SUCCESS == rval); michael@0: michael@0: /* michael@0: * create the client process michael@0: */ michael@0: { michael@0: #define MAX_ARGS 4 michael@0: char *argv[MAX_ARGS + 1]; michael@0: int index = 0; michael@0: char port[32]; michael@0: char path[1024 + sizeof("/thrpool_client")]; michael@0: michael@0: getcwd(path, sizeof(path)); michael@0: michael@0: (void)strcat(path, "/thrpool_client"); michael@0: #ifdef XP_PC michael@0: (void)strcat(path, ".exe"); michael@0: #endif michael@0: argv[index++] = path; michael@0: sprintf(port,"%d",PR_ntohs(netaddr.inet.port)); michael@0: if (_debug_on) michael@0: { michael@0: argv[index++] = "-d"; michael@0: argv[index++] = "-p"; michael@0: argv[index++] = port; michael@0: argv[index++] = NULL; michael@0: } else { michael@0: argv[index++] = "-p"; michael@0: argv[index++] = port; michael@0: argv[index++] = NULL; michael@0: } michael@0: PR_ASSERT(MAX_ARGS >= (index - 1)); michael@0: michael@0: DPRINTF(("creating client process %s ...\n", path)); michael@0: if (PR_FAILURE == PR_CreateProcessDetached(path, argv, NULL, NULL)) { michael@0: fprintf(stderr, michael@0: "thrpool_server: ERROR - PR_CreateProcessDetached failed\n"); michael@0: failed_already=1; michael@0: return; michael@0: } michael@0: } michael@0: michael@0: sc_mon = PR_NewMonitor(); michael@0: if (sc_mon == NULL) { michael@0: fprintf(stderr,"%s: PR_NewMonitor failed\n", program_name); michael@0: failed_already=1; michael@0: return; michael@0: } michael@0: michael@0: sp->iod.socket = sockfd; michael@0: sp->iod.timeout = PR_SecondsToInterval(60); michael@0: sp->datalen = tcp_mesg_size; michael@0: sp->exit_mon = sc_mon; michael@0: sp->job_counterp = &job_counter; michael@0: sp->conn_counter = 0; michael@0: sp->tp = tp; michael@0: sp->netaddr = netaddr; michael@0: michael@0: /* create and cancel a timer job */ michael@0: jobp = PR_QueueJob_Timer(tp, PR_MillisecondsToInterval(5000), michael@0: print_stats, sp, PR_FALSE); michael@0: PR_ASSERT(NULL != jobp); michael@0: rval = PR_CancelJob(jobp); michael@0: PR_ASSERT(PR_SUCCESS == rval); michael@0: michael@0: DPRINTF(("TCP_Server: Accepting connections \n")); michael@0: michael@0: jobp = PR_QueueJob_Accept(tp, &sp->iod, TCP_Server_Accept, sp, michael@0: PR_FALSE); michael@0: PR_ASSERT(NULL != jobp); michael@0: return; michael@0: } michael@0: michael@0: static void michael@0: TCP_Server_Accept(void *arg) michael@0: { michael@0: Server_Param *sp = (Server_Param *) arg; michael@0: PRThreadPool *tp = sp->tp; michael@0: Serve_Client_Param *scp; michael@0: PRFileDesc *newsockfd; michael@0: PRJob *jobp; michael@0: michael@0: if ((newsockfd = PR_Accept(sp->iod.socket, &sp->netaddr, michael@0: PR_INTERVAL_NO_TIMEOUT)) == NULL) { michael@0: fprintf(stderr,"%s: ERROR - PR_Accept failed\n", program_name); michael@0: failed_already=1; michael@0: goto exit; michael@0: } michael@0: scp = PR_NEW(Serve_Client_Param); michael@0: if (scp == NULL) { michael@0: fprintf(stderr,"%s: PR_NEW failed\n", program_name); michael@0: failed_already=1; michael@0: goto exit; michael@0: } michael@0: michael@0: /* michael@0: * Start a Serve_Client job for each incoming connection michael@0: */ michael@0: scp->iod.socket = newsockfd; michael@0: scp->iod.timeout = PR_SecondsToInterval(60); michael@0: scp->datalen = tcp_mesg_size; michael@0: scp->exit_mon = sp->exit_mon; michael@0: scp->job_counterp = sp->job_counterp; michael@0: scp->tp = sp->tp; michael@0: michael@0: PR_EnterMonitor(sp->exit_mon); michael@0: (*sp->job_counterp)++; michael@0: PR_ExitMonitor(sp->exit_mon); michael@0: jobp = PR_QueueJob(tp, Serve_Client, scp, michael@0: PR_FALSE); michael@0: michael@0: PR_ASSERT(NULL != jobp); michael@0: DPRINTF(("TCP_Server: Created Serve_Client = 0x%lx\n", jobp)); michael@0: michael@0: /* michael@0: * single-threaded update; no lock needed michael@0: */ michael@0: sp->conn_counter++; michael@0: if (sp->conn_counter < michael@0: (num_tcp_clients * num_tcp_connections_per_client)) { michael@0: jobp = PR_QueueJob_Accept(tp, &sp->iod, TCP_Server_Accept, sp, michael@0: PR_FALSE); michael@0: PR_ASSERT(NULL != jobp); michael@0: return; michael@0: } michael@0: jobp = PR_QueueJob_Timer(tp, PR_MillisecondsToInterval(500), michael@0: print_stats, sp, PR_FALSE); michael@0: michael@0: PR_ASSERT(NULL != jobp); michael@0: DPRINTF(("TCP_Server: Created print_stats timer job = 0x%lx\n", jobp)); michael@0: michael@0: exit: michael@0: PR_EnterMonitor(sp->exit_mon); michael@0: /* Wait for server jobs to finish */ michael@0: while (0 != *sp->job_counterp) { michael@0: PR_Wait(sp->exit_mon, PR_INTERVAL_NO_TIMEOUT); michael@0: DPRINTF(("TCP_Server: conn_counter = %d\n", michael@0: *sp->job_counterp)); michael@0: } michael@0: michael@0: PR_ExitMonitor(sp->exit_mon); michael@0: if (sp->iod.socket) { michael@0: PR_Close(sp->iod.socket); michael@0: } michael@0: PR_DestroyMonitor(sp->exit_mon); michael@0: printf("%30s","TCP_Socket_Client_Server_Test:"); michael@0: printf("%2ld Server %2ld Clients %2ld connections_per_client\n",1l, michael@0: num_tcp_clients, num_tcp_connections_per_client); michael@0: printf("%30s %2ld messages_per_connection %4ld bytes_per_message\n",":", michael@0: num_tcp_mesgs_per_connection, tcp_mesg_size); michael@0: michael@0: DPRINTF(("%s: calling PR_ShutdownThreadPool\n", program_name)); michael@0: PR_ShutdownThreadPool(sp->tp); michael@0: PR_DELETE(sp); michael@0: } michael@0: michael@0: /************************************************************************/ michael@0: michael@0: #define DEFAULT_INITIAL_THREADS 4 michael@0: #define DEFAULT_MAX_THREADS 100 michael@0: #define DEFAULT_STACKSIZE (512 * 1024) michael@0: michael@0: int main(int argc, char **argv) michael@0: { michael@0: PRInt32 initial_threads = DEFAULT_INITIAL_THREADS; michael@0: PRInt32 max_threads = DEFAULT_MAX_THREADS; michael@0: PRInt32 stacksize = DEFAULT_STACKSIZE; michael@0: PRThreadPool *tp = NULL; michael@0: PRStatus rv; michael@0: PRJob *jobp; michael@0: michael@0: /* michael@0: * -d debug mode michael@0: */ michael@0: PLOptStatus os; michael@0: PLOptState *opt; michael@0: michael@0: program_name = argv[0]; michael@0: opt = PL_CreateOptState(argc, argv, "d"); michael@0: while (PL_OPT_EOL != (os = PL_GetNextOpt(opt))) michael@0: { michael@0: if (PL_OPT_BAD == os) continue; michael@0: switch (opt->option) michael@0: { michael@0: case 'd': /* debug mode */ michael@0: _debug_on = 1; michael@0: break; michael@0: default: michael@0: break; michael@0: } michael@0: } michael@0: PL_DestroyOptState(opt); michael@0: michael@0: PR_Init(PR_USER_THREAD, PR_PRIORITY_NORMAL, 0); michael@0: PR_STDIO_INIT(); michael@0: michael@0: PR_SetConcurrency(4); michael@0: michael@0: tp = PR_CreateThreadPool(initial_threads, max_threads, stacksize); michael@0: if (NULL == tp) { michael@0: printf("PR_CreateThreadPool failed\n"); michael@0: failed_already=1; michael@0: goto done; michael@0: } michael@0: jobp = PR_QueueJob(tp, TCP_Server, tp, PR_TRUE); michael@0: rv = PR_JoinJob(jobp); michael@0: PR_ASSERT(PR_SUCCESS == rv); michael@0: michael@0: DPRINTF(("%s: calling PR_JoinThreadPool\n", program_name)); michael@0: rv = PR_JoinThreadPool(tp); michael@0: PR_ASSERT(PR_SUCCESS == rv); michael@0: DPRINTF(("%s: returning from PR_JoinThreadPool\n", program_name)); michael@0: michael@0: done: michael@0: PR_Cleanup(); michael@0: if (failed_already) return 1; michael@0: else return 0; michael@0: }