nsprpub/pr/tests/thrpool_server.c

   1 /* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
   2 /* This Source Code Form is subject to the terms of the Mozilla Public
   3  * License, v. 2.0. If a copy of the MPL was not distributed with this
   4  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
   5
   6 /***********************************************************************
   7 **
   8 ** Name: thrpool.c
   9 **
  10 ** Description: Test threadpool functionality.
  11 **
  12 ** Modification History:
  13 */
  14 #include "primpl.h"
  15
  16 #include "plgetopt.h"
  17
  18 #include <stdio.h>
  19 #include <string.h>
  20 #include <errno.h>
  21 #ifdef XP_UNIX
  22 #include <sys/mman.h>
  23 #endif
  24 #if defined(_PR_PTHREADS)
  25 #include <pthread.h>
  26 #endif
  27
  28 /* for getcwd */
  29 #if defined(XP_UNIX) || defined (XP_OS2)
  30 #include <unistd.h>
  31 #elif defined(XP_PC)
  32 #include <direct.h>
  33 #endif
  34
  35 #ifdef WIN32
  36 #include <process.h>
  37 #endif
  38
  39 static int _debug_on = 0;
  40 static char *program_name = NULL;
  41 static void serve_client_write(void *arg);
  42
  43 #include "obsolete/prsem.h"
  44
  45 #ifdef XP_PC
  46 #define mode_t int
  47 #endif
  48
  49 #define DPRINTF(arg) if (_debug_on) printf arg
  50
  51
  52 #define BUF_DATA_SIZE    (2 * 1024)
  53 #define TCP_MESG_SIZE    1024
  54 #define NUM_TCP_CLIENTS  10 /* for a listen queue depth of 5 */
  55
  56
  57 #define NUM_TCP_CONNECTIONS_PER_CLIENT  10
  58 #define NUM_TCP_MESGS_PER_CONNECTION    10
  59 #define TCP_SERVER_PORT                 10000
  60 #define SERVER_MAX_BIND_COUNT           100
  61
  62 #ifdef WINCE
  63 char *getcwd(char *buf, size_t size)
  64 {
  65     wchar_t wpath[MAX_PATH];
  66     _wgetcwd(wpath, MAX_PATH);
  67     WideCharToMultiByte(CP_ACP, 0, wpath, -1, buf, size, 0, 0);
  68 }
  69
  70 #define perror(s)
  71 #endif
  72
  73 static PRInt32 num_tcp_clients = NUM_TCP_CLIENTS;
  74 static PRInt32 num_tcp_connections_per_client = NUM_TCP_CONNECTIONS_PER_CLIENT;
  75 static PRInt32 tcp_mesg_size = TCP_MESG_SIZE;
  76 static PRInt32 num_tcp_mesgs_per_connection = NUM_TCP_MESGS_PER_CONNECTION;
  77 static void TCP_Server_Accept(void *arg);
  78
  79
  80 int failed_already=0;
  81 typedef struct buffer {
  82     char    data[BUF_DATA_SIZE];
  83 } buffer;
  84
  85
  86 typedef struct Server_Param {
  87     PRJobIoDesc iod;    /* socket to read from/write to    */
  88     PRInt32     datalen;    /* bytes of data transfered in each read/write */
  89     PRNetAddr   netaddr;
  90     PRMonitor   *exit_mon;    /* monitor to signal on exit            */
  91     PRInt32     *job_counterp;    /* counter to decrement, before exit        */
  92     PRInt32     conn_counter;    /* counter to decrement, before exit        */
  93     PRThreadPool *tp;
  94 } Server_Param;
  95
  96 typedef struct Serve_Client_Param {
  97     PRJobIoDesc iod;    /* socket to read from/write to    */
  98     PRInt32    datalen;    /* bytes of data transfered in each read/write */
  99     PRMonitor *exit_mon;    /* monitor to signal on exit            */
 100     PRInt32 *job_counterp;    /* counter to decrement, before exit        */
 101     PRThreadPool *tp;
 102 } Serve_Client_Param;
 103
 104 typedef struct Session {
 105     PRJobIoDesc iod;    /* socket to read from/write to    */
 106     buffer  *in_buf;
 107     PRInt32 bytes;
 108     PRInt32 msg_num;
 109     PRInt32 bytes_read;
 110     PRMonitor *exit_mon;    /* monitor to signal on exit            */
 111     PRInt32 *job_counterp;    /* counter to decrement, before exit        */
 112     PRThreadPool *tp;
 113 } Session;
 114
 115 static void
 116 serve_client_read(void *arg)
 117 {
 118     Session *sp = (Session *) arg;
 119     int rem;
 120     int bytes;
 121     int offset;
 122     PRFileDesc *sockfd;
 123     char *buf;
 124     PRJob *jobp;
 125
 126     PRIntervalTime timeout = PR_INTERVAL_NO_TIMEOUT;
 127
 128     sockfd = sp->iod.socket;
 129     buf = sp->in_buf->data;
 130
 131     PR_ASSERT(sp->msg_num < num_tcp_mesgs_per_connection);
 132     PR_ASSERT(sp->bytes_read < sp->bytes);
 133
 134     offset = sp->bytes_read;
 135     rem = sp->bytes - offset;
 136     bytes = PR_Recv(sockfd, buf + offset, rem, 0, timeout);
 137     if (bytes < 0) {
 138         return;
 139     }
 140     sp->bytes_read += bytes;
 141     sp->iod.timeout = PR_SecondsToInterval(60);
 142     if (sp->bytes_read <  sp->bytes) {
 143         jobp = PR_QueueJob_Read(sp->tp, &sp->iod, serve_client_read, sp,
 144                                 PR_FALSE);
 145         PR_ASSERT(NULL != jobp);
 146         return;
 147     }
 148     PR_ASSERT(sp->bytes_read == sp->bytes);
 149     DPRINTF(("serve_client: read complete, msg(%d) \n", sp->msg_num));
 150
 151     sp->iod.timeout = PR_SecondsToInterval(60);
 152     jobp = PR_QueueJob_Write(sp->tp, &sp->iod, serve_client_write, sp,
 153                              PR_FALSE);
 154     PR_ASSERT(NULL != jobp);
 155
 156     return;
 157 }
 158
 159 static void
 160 serve_client_write(void *arg)
 161 {
 162     Session *sp = (Session *) arg;
 163     int bytes;
 164     PRFileDesc *sockfd;
 165     char *buf;
 166     PRJob *jobp;
 167
 168     sockfd = sp->iod.socket;
 169     buf = sp->in_buf->data;
 170
 171     PR_ASSERT(sp->msg_num < num_tcp_mesgs_per_connection);
 172
 173     bytes = PR_Send(sockfd, buf, sp->bytes, 0, PR_INTERVAL_NO_TIMEOUT);
 174     PR_ASSERT(bytes == sp->bytes);
 175
 176     if (bytes < 0) {
 177         return;
 178     }
 179     DPRINTF(("serve_client: write complete, msg(%d) \n", sp->msg_num));
 180     sp->msg_num++;
 181     if (sp->msg_num < num_tcp_mesgs_per_connection) {
 182         sp->bytes_read = 0;
 183         sp->iod.timeout = PR_SecondsToInterval(60);
 184         jobp = PR_QueueJob_Read(sp->tp, &sp->iod, serve_client_read, sp,
 185                                 PR_FALSE);
 186         PR_ASSERT(NULL != jobp);
 187         return;
 188     }
 189
 190     DPRINTF(("serve_client: read/write complete, msg(%d) \n", sp->msg_num));
 191     if (PR_Shutdown(sockfd, PR_SHUTDOWN_BOTH) < 0) {
 192         fprintf(stderr,"%s: ERROR - PR_Shutdown\n", program_name);
 193     }
 194
 195     PR_Close(sockfd);
 196     PR_EnterMonitor(sp->exit_mon);
 197     --(*sp->job_counterp);
 198     PR_Notify(sp->exit_mon);
 199     PR_ExitMonitor(sp->exit_mon);
 200
 201     PR_DELETE(sp->in_buf);
 202     PR_DELETE(sp);
 203
 204     return;
 205 }
 206
 207 /*
 208  * Serve_Client
 209  *    Thread, started by the server, for serving a client connection.
 210  *    Reads data from socket and writes it back, unmodified, and
 211  *    closes the socket
 212  */
 213 static void PR_CALLBACK
 214 Serve_Client(void *arg)
 215 {
 216     Serve_Client_Param *scp = (Serve_Client_Param *) arg;
 217     buffer *in_buf;
 218     Session *sp;
 219     PRJob *jobp;
 220
 221     sp = PR_NEW(Session);
 222     sp->iod = scp->iod;
 223
 224     in_buf = PR_NEW(buffer);
 225     if (in_buf == NULL) {
 226         fprintf(stderr,"%s: failed to alloc buffer struct\n",program_name);
 227         failed_already=1;
 228         return;
 229     }
 230
 231     sp->in_buf = in_buf;
 232     sp->bytes = scp->datalen;
 233     sp->msg_num = 0;
 234     sp->bytes_read = 0;
 235     sp->tp = scp->tp;
 236     sp->exit_mon = scp->exit_mon;
 237     sp->job_counterp = scp->job_counterp;
 238
 239     sp->iod.timeout = PR_SecondsToInterval(60);
 240     jobp = PR_QueueJob_Read(sp->tp, &sp->iod, serve_client_read, sp,
 241                             PR_FALSE);
 242     PR_ASSERT(NULL != jobp);
 243     PR_DELETE(scp);
 244 }
 245
 246 static void
 247 print_stats(void *arg)
 248 {
 249     Server_Param *sp = (Server_Param *) arg;
 250     PRThreadPool *tp = sp->tp;
 251     PRInt32 counter;
 252     PRJob *jobp;
 253
 254     PR_EnterMonitor(sp->exit_mon);
 255     counter = (*sp->job_counterp);
 256     PR_ExitMonitor(sp->exit_mon);
 257
 258     printf("PRINT_STATS: #client connections = %d\n",counter);
 259
 260
 261     jobp = PR_QueueJob_Timer(tp, PR_MillisecondsToInterval(500),
 262                              print_stats, sp, PR_FALSE);
 263
 264     PR_ASSERT(NULL != jobp);
 265 }
 266
 267 static int job_counter = 0;
 268 /*
 269  * TCP Server
 270  *    Server binds an address to a socket, starts a client process and
 271  *    listens for incoming connections.
 272  *    Each client connects to the server and sends a chunk of data
 273  *    Starts a Serve_Client job for each incoming connection, to read
 274  *    the data from the client and send it back to the client, unmodified.
 275  *    Each client checks that data received from server is same as the
 276  *    data it sent to the server.
 277  *    Finally, the threadpool is shutdown
 278  */
 279 static void PR_CALLBACK
 280 TCP_Server(void *arg)
 281 {
 282     PRThreadPool *tp = (PRThreadPool *) arg;
 283     Server_Param *sp;
 284     PRFileDesc *sockfd;
 285     PRNetAddr netaddr;
 286     PRMonitor *sc_mon;
 287     PRJob *jobp;
 288     int i;
 289     PRStatus rval;
 290
 291     /*
 292      * Create a tcp socket
 293      */
 294     if ((sockfd = PR_NewTCPSocket()) == NULL) {
 295         fprintf(stderr,"%s: PR_NewTCPSocket failed\n", program_name);
 296         return;
 297     }
 298     memset(&netaddr, 0, sizeof(netaddr));
 299     netaddr.inet.family = PR_AF_INET;
 300     netaddr.inet.port = PR_htons(TCP_SERVER_PORT);
 301     netaddr.inet.ip = PR_htonl(PR_INADDR_ANY);
 302     /*
 303      * try a few times to bind server's address, if addresses are in
 304      * use
 305      */
 306     i = 0;
 307     while (PR_Bind(sockfd, &netaddr) < 0) {
 308         if (PR_GetError() == PR_ADDRESS_IN_USE_ERROR) {
 309             netaddr.inet.port += 2;
 310             if (i++ < SERVER_MAX_BIND_COUNT) {
 311                 continue;
 312             }
 313         }
 314         fprintf(stderr,"%s: ERROR - PR_Bind failed\n", program_name);
 315         perror("PR_Bind");
 316         failed_already=1;
 317         return;
 318     }
 319
 320     if (PR_Listen(sockfd, 32) < 0) {
 321         fprintf(stderr,"%s: ERROR - PR_Listen failed\n", program_name);
 322         failed_already=1;
 323         return;
 324     }
 325
 326     if (PR_GetSockName(sockfd, &netaddr) < 0) {
 327         fprintf(stderr,"%s: ERROR - PR_GetSockName failed\n", program_name);
 328         failed_already=1;
 329         return;
 330     }
 331
 332     DPRINTF((
 333                 "TCP_Server: PR_BIND netaddr.inet.ip = 0x%lx, netaddr.inet.port = %d\n",
 334                 netaddr.inet.ip, netaddr.inet.port));
 335
 336     sp = PR_NEW(Server_Param);
 337     if (sp == NULL) {
 338         fprintf(stderr,"%s: PR_NEW failed\n", program_name);
 339         failed_already=1;
 340         return;
 341     }
 342     sp->iod.socket = sockfd;
 343     sp->iod.timeout = PR_SecondsToInterval(60);
 344     sp->datalen = tcp_mesg_size;
 345     sp->exit_mon = sc_mon;
 346     sp->job_counterp = &job_counter;
 347     sp->conn_counter = 0;
 348     sp->tp = tp;
 349     sp->netaddr = netaddr;
 350
 351     /* create and cancel an io job */
 352     jobp = PR_QueueJob_Accept(tp, &sp->iod, TCP_Server_Accept, sp,
 353                               PR_FALSE);
 354     PR_ASSERT(NULL != jobp);
 355     rval = PR_CancelJob(jobp);
 356     PR_ASSERT(PR_SUCCESS == rval);
 357
 358     /*
 359      * create the client process
 360      */
 361     {
 362 #define MAX_ARGS 4
 363         char *argv[MAX_ARGS + 1];
 364         int index = 0;
 365         char port[32];
 366         char path[1024 + sizeof("/thrpool_client")];
 367
 368         getcwd(path, sizeof(path));
 369
 370         (void)strcat(path, "/thrpool_client");
 371 #ifdef XP_PC
 372         (void)strcat(path, ".exe");
 373 #endif
 374         argv[index++] = path;
 375         sprintf(port,"%d",PR_ntohs(netaddr.inet.port));
 376         if (_debug_on)
 377         {
 378             argv[index++] = "-d";
 379             argv[index++] = "-p";
 380             argv[index++] = port;
 381             argv[index++] = NULL;
 382         } else {
 383             argv[index++] = "-p";
 384             argv[index++] = port;
 385             argv[index++] = NULL;
 386         }
 387         PR_ASSERT(MAX_ARGS >= (index - 1));
 388
 389         DPRINTF(("creating client process %s ...\n", path));
 390         if (PR_FAILURE == PR_CreateProcessDetached(path, argv, NULL, NULL)) {
 391             fprintf(stderr,
 392                     "thrpool_server: ERROR - PR_CreateProcessDetached failed\n");
 393             failed_already=1;
 394             return;
 395         }
 396     }
 397
 398     sc_mon = PR_NewMonitor();
 399     if (sc_mon == NULL) {
 400         fprintf(stderr,"%s: PR_NewMonitor failed\n", program_name);
 401         failed_already=1;
 402         return;
 403     }
 404
 405     sp->iod.socket = sockfd;
 406     sp->iod.timeout = PR_SecondsToInterval(60);
 407     sp->datalen = tcp_mesg_size;
 408     sp->exit_mon = sc_mon;
 409     sp->job_counterp = &job_counter;
 410     sp->conn_counter = 0;
 411     sp->tp = tp;
 412     sp->netaddr = netaddr;
 413
 414     /* create and cancel a timer job */
 415     jobp = PR_QueueJob_Timer(tp, PR_MillisecondsToInterval(5000),
 416                              print_stats, sp, PR_FALSE);
 417     PR_ASSERT(NULL != jobp);
 418     rval = PR_CancelJob(jobp);
 419     PR_ASSERT(PR_SUCCESS == rval);
 420
 421     DPRINTF(("TCP_Server: Accepting connections \n"));
 422
 423     jobp = PR_QueueJob_Accept(tp, &sp->iod, TCP_Server_Accept, sp,
 424                               PR_FALSE);
 425     PR_ASSERT(NULL != jobp);
 426     return;
 427 }
 428
 429 static void
 430 TCP_Server_Accept(void *arg)
 431 {
 432     Server_Param *sp = (Server_Param *) arg;
 433     PRThreadPool *tp = sp->tp;
 434     Serve_Client_Param *scp;
 435     PRFileDesc *newsockfd;
 436     PRJob *jobp;
 437
 438     if ((newsockfd = PR_Accept(sp->iod.socket, &sp->netaddr,
 439                                PR_INTERVAL_NO_TIMEOUT)) == NULL) {
 440         fprintf(stderr,"%s: ERROR - PR_Accept failed\n", program_name);
 441         failed_already=1;
 442         goto exit;
 443     }
 444     scp = PR_NEW(Serve_Client_Param);
 445     if (scp == NULL) {
 446         fprintf(stderr,"%s: PR_NEW failed\n", program_name);
 447         failed_already=1;
 448         goto exit;
 449     }
 450
 451     /*
 452      * Start a Serve_Client job for each incoming connection
 453      */
 454     scp->iod.socket = newsockfd;
 455     scp->iod.timeout = PR_SecondsToInterval(60);
 456     scp->datalen = tcp_mesg_size;
 457     scp->exit_mon = sp->exit_mon;
 458     scp->job_counterp = sp->job_counterp;
 459     scp->tp = sp->tp;
 460
 461     PR_EnterMonitor(sp->exit_mon);
 462     (*sp->job_counterp)++;
 463     PR_ExitMonitor(sp->exit_mon);
 464     jobp = PR_QueueJob(tp, Serve_Client, scp,
 465                        PR_FALSE);
 466
 467     PR_ASSERT(NULL != jobp);
 468     DPRINTF(("TCP_Server: Created Serve_Client = 0x%lx\n", jobp));
 469
 470     /*
 471      * single-threaded update; no lock needed
 472      */
 473     sp->conn_counter++;
 474     if (sp->conn_counter <
 475         (num_tcp_clients * num_tcp_connections_per_client)) {
 476         jobp = PR_QueueJob_Accept(tp, &sp->iod, TCP_Server_Accept, sp,
 477                                   PR_FALSE);
 478         PR_ASSERT(NULL != jobp);
 479         return;
 480     }
 481     jobp = PR_QueueJob_Timer(tp, PR_MillisecondsToInterval(500),
 482                              print_stats, sp, PR_FALSE);
 483
 484     PR_ASSERT(NULL != jobp);
 485     DPRINTF(("TCP_Server: Created print_stats timer job = 0x%lx\n", jobp));
 486
 487 exit:
 488     PR_EnterMonitor(sp->exit_mon);
 489     /* Wait for server jobs to finish */
 490     while (0 != *sp->job_counterp) {
 491         PR_Wait(sp->exit_mon, PR_INTERVAL_NO_TIMEOUT);
 492         DPRINTF(("TCP_Server: conn_counter = %d\n",
 493                  *sp->job_counterp));
 494     }
 495
 496     PR_ExitMonitor(sp->exit_mon);
 497     if (sp->iod.socket) {
 498         PR_Close(sp->iod.socket);
 499     }
 500     PR_DestroyMonitor(sp->exit_mon);
 501     printf("%30s","TCP_Socket_Client_Server_Test:");
 502     printf("%2ld Server %2ld Clients %2ld connections_per_client\n",1l,
 503            num_tcp_clients, num_tcp_connections_per_client);
 504     printf("%30s %2ld messages_per_connection %4ld bytes_per_message\n",":",
 505            num_tcp_mesgs_per_connection, tcp_mesg_size);
 506
 507     DPRINTF(("%s: calling PR_ShutdownThreadPool\n", program_name));
 508     PR_ShutdownThreadPool(sp->tp);
 509     PR_DELETE(sp);
 510 }
 511
 512 /************************************************************************/
 513
 514 #define DEFAULT_INITIAL_THREADS     4
 515 #define DEFAULT_MAX_THREADS         100
 516 #define DEFAULT_STACKSIZE           (512 * 1024)
 517
 518 int main(int argc, char **argv)
 519 {
 520     PRInt32 initial_threads = DEFAULT_INITIAL_THREADS;
 521     PRInt32 max_threads = DEFAULT_MAX_THREADS;
 522     PRInt32 stacksize = DEFAULT_STACKSIZE;
 523     PRThreadPool *tp = NULL;
 524     PRStatus rv;
 525     PRJob *jobp;
 526
 527     /*
 528      * -d           debug mode
 529      */
 530     PLOptStatus os;
 531     PLOptState *opt;
 532
 533     program_name = argv[0];
 534     opt = PL_CreateOptState(argc, argv, "d");
 535     while (PL_OPT_EOL != (os = PL_GetNextOpt(opt)))
 536     {
 537         if (PL_OPT_BAD == os) {
 538             continue;
 539         }
 540         switch (opt->option)
 541         {
 542             case 'd':  /* debug mode */
 543                 _debug_on = 1;
 544                 break;
 545             default:
 546                 break;
 547         }
 548     }
 549     PL_DestroyOptState(opt);
 550
 551     PR_Init(PR_USER_THREAD, PR_PRIORITY_NORMAL, 0);
 552     PR_STDIO_INIT();
 553
 554     PR_SetConcurrency(4);
 555
 556     tp = PR_CreateThreadPool(initial_threads, max_threads, stacksize);
 557     if (NULL == tp) {
 558         printf("PR_CreateThreadPool failed\n");
 559         failed_already=1;
 560         goto done;
 561     }
 562     jobp = PR_QueueJob(tp, TCP_Server, tp, PR_TRUE);
 563     rv = PR_JoinJob(jobp);
 564     PR_ASSERT(PR_SUCCESS == rv);
 565
 566     DPRINTF(("%s: calling PR_JoinThreadPool\n", program_name));
 567     rv = PR_JoinThreadPool(tp);
 568     PR_ASSERT(PR_SUCCESS == rv);
 569     DPRINTF(("%s: returning from PR_JoinThreadPool\n", program_name));
 570
 571 done:
 572     PR_Cleanup();
 573     if (failed_already) {
 574         return 1;
 575     }
 576     else {
 577         return 0;
 578     }
 579 }