servr_ku.c (15074B)
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 ** This server simulates a server running in loopback mode. 9 ** 10 ** The idea is that a single server is created. The server initially creates 11 ** a number of worker threads. Then, with the server running, a number of 12 ** clients are created which start requesting service from the server. 13 ** 14 ** 15 ** Modification History: 16 ** 19-May-97 AGarcia- Converted the test to accomodate the debug_mode flag. 17 ** The debug mode will print all of the printfs associated with this 18 *test. 19 ** The regress mode will be the default mode. Since the regress tool 20 *limits 21 ** the output to a one line status:PASS or FAIL,all of the printf 22 *statements 23 ** have been handled with an if (debug_mode) statement. 24 ** 04-June-97 AGarcia removed the Test_Result function. Regress tool has been 25 *updated to 26 ** recognize the return code from tha main program. 27 ***********************************************************************/ 28 29 /*********************************************************************** 30 ** Includes 31 ***********************************************************************/ 32 /* Used to get the command line option */ 33 #include "plgetopt.h" 34 35 #include "nspr.h" 36 #include "pprthred.h" 37 38 #include <string.h> 39 40 #define PORT 15004 41 #define THREAD_STACKSIZE 0 42 43 static int _iterations = 1000; 44 static int _clients = 1; 45 static int _client_data = 250; 46 static int _server_data = (8 * 1024); 47 48 static PRThreadScope ServerScope, ClientScope; 49 50 #define SERVER "Server" 51 #define MAIN "Main" 52 53 #define SERVER_STATE_STARTUP 0 54 #define SERVER_STATE_READY 1 55 #define SERVER_STATE_DYING 2 56 #define SERVER_STATE_DEAD 4 57 int ServerState; 58 PRLock* ServerStateCVLock; 59 PRCondVar* ServerStateCV; 60 61 #ifdef DEBUGPRINTS 62 # define DPRINTF printf 63 #else 64 # define DPRINTF 65 #endif 66 67 PRIntn failed_already = 0; 68 PRIntn debug_mode; 69 70 static void do_work(void); 71 72 /* --- Server state functions --------------------------------------------- */ 73 void SetServerState(char* waiter, PRInt32 state) { 74 PR_Lock(ServerStateCVLock); 75 ServerState = state; 76 PR_NotifyCondVar(ServerStateCV); 77 78 if (debug_mode) { 79 DPRINTF("\t%s changed state to %d\n", waiter, state); 80 } 81 82 PR_Unlock(ServerStateCVLock); 83 } 84 85 int WaitServerState(char* waiter, PRInt32 state) { 86 PRInt32 rv; 87 88 PR_Lock(ServerStateCVLock); 89 90 if (debug_mode) { 91 DPRINTF("\t%s waiting for state %d\n", waiter, state); 92 } 93 94 while (!(ServerState & state)) { 95 PR_WaitCondVar(ServerStateCV, PR_INTERVAL_NO_TIMEOUT); 96 } 97 rv = ServerState; 98 99 if (debug_mode) 100 DPRINTF("\t%s resuming from wait for state %d; state now %d\n", waiter, 101 state, ServerState); 102 PR_Unlock(ServerStateCVLock); 103 104 return rv; 105 } 106 107 /* --- Server Functions ------------------------------------------- */ 108 109 PRLock* workerThreadsLock; 110 PRInt32 workerThreads; 111 PRInt32 workerThreadsBusy; 112 113 void WorkerThreadFunc(void* _listenSock) { 114 PRFileDesc* listenSock = (PRFileDesc*)_listenSock; 115 PRInt32 bytesRead; 116 PRInt32 bytesWritten; 117 char* dataBuf; 118 char* sendBuf; 119 120 if (debug_mode) 121 DPRINTF("\tServer buffer is %d bytes; %d data, %d netaddrs\n", 122 _client_data + (2 * sizeof(PRNetAddr)) + 32, _client_data, 123 (2 * sizeof(PRNetAddr)) + 32); 124 dataBuf = (char*)PR_MALLOC(_client_data + 2 * sizeof(PRNetAddr) + 32); 125 if (!dataBuf) 126 if (debug_mode) { 127 printf("\tServer could not malloc space!?\n"); 128 } 129 sendBuf = (char*)PR_MALLOC(_server_data * sizeof(char)); 130 if (!sendBuf) 131 if (debug_mode) { 132 printf("\tServer could not malloc space!?\n"); 133 } 134 135 if (debug_mode) { 136 DPRINTF("\tServer worker thread running\n"); 137 } 138 139 while (1) { 140 PRInt32 bytesToRead = _client_data; 141 PRInt32 bytesToWrite = _server_data; 142 PRFileDesc* newSock; 143 PRNetAddr* rAddr; 144 PRInt32 loops = 0; 145 146 loops++; 147 148 if (debug_mode) { 149 DPRINTF("\tServer thread going into accept\n"); 150 } 151 152 bytesRead = PR_AcceptRead(listenSock, &newSock, &rAddr, dataBuf, 153 bytesToRead, PR_INTERVAL_NO_TIMEOUT); 154 155 if (bytesRead < 0) { 156 if (debug_mode) { 157 printf("\tServer error in accept (%d)\n", bytesRead); 158 } 159 continue; 160 } 161 162 if (debug_mode) { 163 DPRINTF("\tServer accepted connection (%d bytes)\n", bytesRead); 164 } 165 166 PR_AtomicIncrement(&workerThreadsBusy); 167 if (workerThreadsBusy == workerThreads) { 168 PR_Lock(workerThreadsLock); 169 if (workerThreadsBusy == workerThreads) { 170 PRThread* WorkerThread; 171 172 WorkerThread = PR_CreateThread( 173 PR_SYSTEM_THREAD, WorkerThreadFunc, listenSock, PR_PRIORITY_NORMAL, 174 ServerScope, PR_UNJOINABLE_THREAD, THREAD_STACKSIZE); 175 176 if (!WorkerThread) { 177 if (debug_mode) { 178 printf("Error creating client thread %d\n", workerThreads); 179 } 180 } else { 181 PR_AtomicIncrement(&workerThreads); 182 if (debug_mode) { 183 DPRINTF("\tServer creates worker (%d)\n", workerThreads); 184 } 185 } 186 } 187 PR_Unlock(workerThreadsLock); 188 } 189 190 bytesToRead -= bytesRead; 191 while (bytesToRead) { 192 bytesRead = 193 PR_Recv(newSock, dataBuf, bytesToRead, 0, PR_INTERVAL_NO_TIMEOUT); 194 if (bytesRead < 0) { 195 if (debug_mode) { 196 printf("\tServer error receiving data (%d)\n", bytesRead); 197 } 198 continue; 199 } 200 if (debug_mode) { 201 DPRINTF("\tServer received %d bytes\n", bytesRead); 202 } 203 } 204 205 bytesWritten = 206 PR_Send(newSock, sendBuf, bytesToWrite, 0, PR_INTERVAL_NO_TIMEOUT); 207 if (bytesWritten != _server_data) { 208 if (debug_mode) 209 printf("\tError sending data to client (%d, %d)\n", bytesWritten, 210 PR_GetOSError()); 211 } else { 212 if (debug_mode) { 213 DPRINTF("\tServer sent %d bytes\n", bytesWritten); 214 } 215 } 216 217 PR_Close(newSock); 218 PR_AtomicDecrement(&workerThreadsBusy); 219 } 220 } 221 222 PRFileDesc* ServerSetup(void) { 223 PRFileDesc* listenSocket; 224 PRSocketOptionData sockOpt; 225 PRNetAddr serverAddr; 226 PRThread* WorkerThread; 227 228 if ((listenSocket = PR_NewTCPSocket()) == NULL) { 229 if (debug_mode) { 230 printf("\tServer error creating listen socket\n"); 231 } else { 232 failed_already = 1; 233 } 234 return NULL; 235 } 236 237 sockOpt.option = PR_SockOpt_Reuseaddr; 238 sockOpt.value.reuse_addr = PR_TRUE; 239 if (PR_SetSocketOption(listenSocket, &sockOpt) == PR_FAILURE) { 240 if (debug_mode) 241 printf("\tServer error setting socket option: OS error %d\n", 242 PR_GetOSError()); 243 else { 244 failed_already = 1; 245 } 246 PR_Close(listenSocket); 247 return NULL; 248 } 249 250 memset(&serverAddr, 0, sizeof(PRNetAddr)); 251 serverAddr.inet.family = PR_AF_INET; 252 serverAddr.inet.port = PR_htons(PORT); 253 serverAddr.inet.ip = PR_htonl(PR_INADDR_ANY); 254 255 if (PR_Bind(listenSocket, &serverAddr) == PR_FAILURE) { 256 if (debug_mode) 257 printf("\tServer error binding to server address: OS error %d\n", 258 PR_GetOSError()); 259 else { 260 failed_already = 1; 261 } 262 PR_Close(listenSocket); 263 return NULL; 264 } 265 266 if (PR_Listen(listenSocket, 128) == PR_FAILURE) { 267 if (debug_mode) { 268 printf("\tServer error listening to server socket\n"); 269 } else { 270 failed_already = 1; 271 } 272 PR_Close(listenSocket); 273 274 return NULL; 275 } 276 277 /* Create Clients */ 278 workerThreads = 0; 279 workerThreadsBusy = 0; 280 281 workerThreadsLock = PR_NewLock(); 282 283 WorkerThread = PR_CreateThread(PR_SYSTEM_THREAD, WorkerThreadFunc, 284 listenSocket, PR_PRIORITY_NORMAL, ServerScope, 285 PR_UNJOINABLE_THREAD, THREAD_STACKSIZE); 286 287 if (!WorkerThread) { 288 if (debug_mode) { 289 printf("error creating working thread\n"); 290 } 291 PR_Close(listenSocket); 292 return NULL; 293 } 294 PR_AtomicIncrement(&workerThreads); 295 if (debug_mode) { 296 DPRINTF("\tServer created primordial worker thread\n"); 297 } 298 299 return listenSocket; 300 } 301 302 /* The main server loop */ 303 void ServerThreadFunc(void* unused) { 304 PRFileDesc* listenSocket; 305 306 /* Do setup */ 307 listenSocket = ServerSetup(); 308 309 if (!listenSocket) { 310 SetServerState(SERVER, SERVER_STATE_DEAD); 311 } else { 312 if (debug_mode) { 313 DPRINTF("\tServer up\n"); 314 } 315 316 /* Tell clients they can start now. */ 317 SetServerState(SERVER, SERVER_STATE_READY); 318 319 /* Now wait for server death signal */ 320 WaitServerState(SERVER, SERVER_STATE_DYING); 321 322 /* Cleanup */ 323 SetServerState(SERVER, SERVER_STATE_DEAD); 324 } 325 } 326 327 /* --- Client Functions ------------------------------------------- */ 328 329 PRInt32 numRequests; 330 PRInt32 numClients; 331 PRMonitor* clientMonitor; 332 333 void ClientThreadFunc(void* unused) { 334 PRNetAddr serverAddr; 335 PRFileDesc* clientSocket; 336 char* sendBuf; 337 char* recvBuf; 338 PRInt32 rv; 339 PRInt32 bytesNeeded; 340 341 sendBuf = (char*)PR_MALLOC(_client_data * sizeof(char)); 342 if (!sendBuf) 343 if (debug_mode) { 344 printf("\tClient could not malloc space!?\n"); 345 } 346 recvBuf = (char*)PR_MALLOC(_server_data * sizeof(char)); 347 if (!recvBuf) 348 if (debug_mode) { 349 printf("\tClient could not malloc space!?\n"); 350 } 351 352 memset(&serverAddr, 0, sizeof(PRNetAddr)); 353 serverAddr.inet.family = PR_AF_INET; 354 serverAddr.inet.port = PR_htons(PORT); 355 serverAddr.inet.ip = PR_htonl(PR_INADDR_LOOPBACK); 356 357 while (numRequests > 0) { 358 if ((numRequests % 10) == 0) 359 if (debug_mode) { 360 printf("."); 361 } 362 if (debug_mode) { 363 DPRINTF("\tClient starting request %d\n", numRequests); 364 } 365 366 clientSocket = PR_NewTCPSocket(); 367 if (!clientSocket) { 368 if (debug_mode) 369 printf("Client error creating socket: OS error %d\n", PR_GetOSError()); 370 continue; 371 } 372 373 if (debug_mode) { 374 DPRINTF("\tClient connecting\n"); 375 } 376 377 rv = PR_Connect(clientSocket, &serverAddr, PR_INTERVAL_NO_TIMEOUT); 378 if (!clientSocket) { 379 if (debug_mode) { 380 printf("\tClient error connecting\n"); 381 } 382 continue; 383 } 384 385 if (debug_mode) { 386 DPRINTF("\tClient connected\n"); 387 } 388 389 rv = 390 PR_Send(clientSocket, sendBuf, _client_data, 0, PR_INTERVAL_NO_TIMEOUT); 391 if (rv != _client_data) { 392 if (debug_mode) { 393 printf("Client error sending data (%d)\n", rv); 394 } 395 PR_Close(clientSocket); 396 continue; 397 } 398 399 if (debug_mode) { 400 DPRINTF("\tClient sent %d bytes\n", rv); 401 } 402 403 bytesNeeded = _server_data; 404 while (bytesNeeded) { 405 rv = PR_Recv(clientSocket, recvBuf, bytesNeeded, 0, 406 PR_INTERVAL_NO_TIMEOUT); 407 if (rv <= 0) { 408 if (debug_mode) 409 printf("Client error receiving data (%d) (%d/%d)\n", rv, 410 (_server_data - bytesNeeded), _server_data); 411 break; 412 } 413 if (debug_mode) { 414 DPRINTF("\tClient received %d bytes; need %d more\n", rv, 415 bytesNeeded - rv); 416 } 417 bytesNeeded -= rv; 418 } 419 420 PR_Close(clientSocket); 421 422 PR_AtomicDecrement(&numRequests); 423 } 424 425 PR_EnterMonitor(clientMonitor); 426 --numClients; 427 PR_Notify(clientMonitor); 428 PR_ExitMonitor(clientMonitor); 429 430 PR_DELETE(sendBuf); 431 PR_DELETE(recvBuf); 432 } 433 434 void RunClients(void) { 435 PRInt32 index; 436 437 numRequests = _iterations; 438 numClients = _clients; 439 clientMonitor = PR_NewMonitor(); 440 441 for (index = 0; index < _clients; index++) { 442 PRThread* clientThread; 443 444 clientThread = PR_CreateThread(PR_USER_THREAD, ClientThreadFunc, NULL, 445 PR_PRIORITY_NORMAL, ClientScope, 446 PR_UNJOINABLE_THREAD, THREAD_STACKSIZE); 447 448 if (!clientThread) { 449 if (debug_mode) { 450 printf("\terror creating client thread %d\n", index); 451 } 452 } else if (debug_mode) { 453 DPRINTF("\tMain created client %d/%d\n", index + 1, _clients); 454 } 455 } 456 457 PR_EnterMonitor(clientMonitor); 458 while (numClients) { 459 PR_Wait(clientMonitor, PR_INTERVAL_NO_TIMEOUT); 460 } 461 PR_ExitMonitor(clientMonitor); 462 } 463 464 /* --- Main Function ---------------------------------------------- */ 465 466 static void do_work() { 467 PRThread* ServerThread; 468 PRInt32 state; 469 470 SetServerState(MAIN, SERVER_STATE_STARTUP); 471 ServerThread = PR_CreateThread(PR_USER_THREAD, ServerThreadFunc, NULL, 472 PR_PRIORITY_NORMAL, ServerScope, 473 PR_JOINABLE_THREAD, THREAD_STACKSIZE); 474 if (!ServerThread) { 475 if (debug_mode) { 476 printf("error creating main server thread\n"); 477 } 478 return; 479 } 480 481 /* Wait for server to be ready */ 482 state = WaitServerState(MAIN, SERVER_STATE_READY | SERVER_STATE_DEAD); 483 484 if (!(state & SERVER_STATE_DEAD)) { 485 /* Run Test Clients */ 486 RunClients(); 487 488 /* Send death signal to server */ 489 SetServerState(MAIN, SERVER_STATE_DYING); 490 } 491 492 PR_JoinThread(ServerThread); 493 } 494 495 static void do_workKU(void) { 496 ServerScope = PR_GLOBAL_THREAD; 497 ClientScope = PR_LOCAL_THREAD; 498 do_work(); 499 } 500 501 static void Measure(void (*func)(void), const char* msg) { 502 PRIntervalTime start, stop; 503 double d; 504 505 start = PR_IntervalNow(); 506 (*func)(); 507 stop = PR_IntervalNow(); 508 509 d = (double)PR_IntervalToMicroseconds(stop - start); 510 511 if (debug_mode) { 512 printf("\n%40s: %6.2f usec\n", msg, d / _iterations); 513 } 514 } 515 516 int main(int argc, char** argv) { 517 /* The command line argument: -d is used to determine if the test is being run 518 in debug mode. The regress tool requires only one line output:PASS or FAIL. 519 All of the printfs associated with this test has been handled with a if 520 (debug_mode) test. Usage: test_name -d 521 */ 522 PLOptStatus os; 523 PLOptState* opt = PL_CreateOptState(argc, argv, "d:"); 524 while (PL_OPT_EOL != (os = PL_GetNextOpt(opt))) { 525 if (PL_OPT_BAD == os) { 526 continue; 527 } 528 switch (opt->option) { 529 case 'd': /* debug mode */ 530 debug_mode = 1; 531 break; 532 default: 533 break; 534 } 535 } 536 PL_DestroyOptState(opt); 537 538 /* main test */ 539 if (debug_mode) { 540 printf("Enter number of iterations: \n"); 541 scanf("%d", &_iterations); 542 printf("Enter number of clients : \n"); 543 scanf("%d", &_clients); 544 printf("Enter size of client data : \n"); 545 scanf("%d", &_client_data); 546 printf("Enter size of server data : \n"); 547 scanf("%d", &_server_data); 548 } else { 549 _iterations = 7; 550 _clients = 7; 551 _client_data = 100; 552 _server_data = 100; 553 } 554 555 if (debug_mode) { 556 printf("\n\n%d iterations with %d client threads.\n", _iterations, 557 _clients); 558 printf("Sending %d bytes of client data and %d bytes of server data\n", 559 _client_data, _server_data); 560 } 561 PR_Init(PR_USER_THREAD, PR_PRIORITY_NORMAL, 0); 562 563 PR_SetThreadRecycleMode(64); 564 565 ServerStateCVLock = PR_NewLock(); 566 ServerStateCV = PR_NewCondVar(ServerStateCVLock); 567 568 Measure(do_workKU, "server loop kernel/user"); 569 570 PR_Cleanup(); 571 if (failed_already) { 572 return 1; 573 } else { 574 return 0; 575 } 576 }