src/libopensrf/osrf_prefork.c

   1 #include <signal.h>
   2 #include "opensrf/osrf_prefork.h"
   3 #include "opensrf/osrf_app_session.h"
   4 #include "opensrf/osrf_application.h"
   5
   6 #define READ_BUFSIZE 1024
   7 #define ABS_MAX_CHILDREN 256
   8
   9 typedef struct {
  10         int max_requests;
  11         int min_children;
  12         int max_children;
  13         int fd;               /**< Unused. */
  14         int data_to_child;    /**< Unused. */
  15         int data_to_parent;   /**< Unused. */
  16         int current_num_children;
  17         int keepalive;        /**< keepalive time for stateful sessions. */
  18         char* appname;
  19         struct prefork_child_struct* first_child;
  20         transport_client* connection;
  21 } prefork_simple;
  22
  23 struct prefork_child_struct {
  24         pid_t pid;
  25         int read_data_fd;
  26         int write_data_fd;
  27         int read_status_fd;
  28         int write_status_fd;
  29         int min_children;
  30         int available;
  31         int max_requests;
  32         char* appname;
  33         int keepalive;
  34         struct prefork_child_struct* next;
  35 };
  36
  37 typedef struct prefork_child_struct prefork_child;
  38
  39 static prefork_simple* prefork_simple_init( transport_client* client,
  40         int max_requests, int min_children, int max_children );
  41 static prefork_child* launch_child( prefork_simple* forker );
  42 static void prefork_launch_children( prefork_simple* forker );
  43 static void prefork_run(prefork_simple* forker);
  44 static void add_prefork_child( prefork_simple* forker, prefork_child* child );
  45
  46 //static prefork_child* find_prefork_child( prefork_simple* forker, pid_t pid );
  47
  48 static void del_prefork_child( prefork_simple* forker, pid_t pid );
  49 static void check_children( prefork_simple* forker, int forever );
  50 static void prefork_child_process_request(prefork_child*, char* data);
  51 static int prefork_child_init_hook(prefork_child*);
  52 static prefork_child* prefork_child_init(
  53                 int max_requests, int read_data_fd, int write_data_fd,
  54                 int read_status_fd, int write_status_fd );
  55
  56 /* listens on the 'data_to_child' fd and wait for incoming data */
  57 static void prefork_child_wait( prefork_child* child );
  58 static int prefork_free( prefork_simple* );
  59 static int prefork_child_free( prefork_child* );
  60 static void osrf_prefork_register_routers( const char* appname );
  61 static void osrf_prefork_child_exit( prefork_child* );
  62
  63
  64 /* true if we just deleted a child.  This will allow us to make sure we're
  65         not trying to use freed memory */
  66 static sig_atomic_t child_dead;
  67
  68 static void sigchld_handler( int sig );
  69
  70 int osrf_prefork_run(const char* appname) {
  71
  72         if(!appname) {
  73                 osrfLogError( OSRF_LOG_MARK, "osrf_prefork_run requires an appname to run!");
  74                 return -1;
  75         }
  76
  77         set_proc_title( "OpenSRF Listener [%s]", appname );
  78
  79         int maxr = 1000;
  80         int maxc = 10;
  81         int minc = 3;
  82         int kalive = 5;
  83
  84         osrfLogInfo( OSRF_LOG_MARK, "Loading config in osrf_forker for app %s", appname);
  85
  86         char* max_req      = osrf_settings_host_value("/apps/%s/unix_config/max_requests", appname);
  87         char* min_children = osrf_settings_host_value("/apps/%s/unix_config/min_children", appname);
  88         char* max_children = osrf_settings_host_value("/apps/%s/unix_config/max_children", appname);
  89         char* keepalive    = osrf_settings_host_value("/apps/%s/keepalive", appname);
  90
  91         if(!keepalive) osrfLogWarning( OSRF_LOG_MARK, "Keepalive is not defined, assuming %d", kalive);
  92         else kalive = atoi(keepalive);
  93
  94         if(!max_req) osrfLogWarning( OSRF_LOG_MARK, "Max requests not defined, assuming %d", maxr);
  95         else maxr = atoi(max_req);
  96
  97         if(!min_children) osrfLogWarning( OSRF_LOG_MARK,
  98                         "Min children not defined, assuming %d", minc);
  99         else minc = atoi(min_children);
 100
 101         if(!max_children) osrfLogWarning( OSRF_LOG_MARK,
 102                         "Max children not defined, assuming %d", maxc);
 103         else maxc = atoi(max_children);
 104
 105         free(keepalive);
 106         free(max_req);
 107         free(min_children);
 108         free(max_children);
 109         /* --------------------------------------------------- */
 110
 111         char* resc = va_list_to_string("%s_listener", appname);
 112
 113         if(!osrfSystemBootstrapClientResc( NULL, NULL, resc )) {
 114                 osrfLogError( OSRF_LOG_MARK, "Unable to bootstrap client for osrf_prefork_run()");
 115                 free(resc);
 116                 return -1;
 117         }
 118
 119         free(resc);
 120
 121         prefork_simple* forker = prefork_simple_init(
 122                 osrfSystemGetTransportClient(), maxr, minc, maxc);
 123
 124         if(forker == NULL) {
 125                 osrfLogError( OSRF_LOG_MARK, "osrf_prefork_run() failed to create prefork_simple object");
 126                 return -1;
 127         }
 128
 129         forker->appname   = strdup(appname);
 130         forker->keepalive = kalive;
 131
 132         prefork_launch_children(forker);
 133
 134         osrf_prefork_register_routers(appname);
 135
 136         osrfLogInfo( OSRF_LOG_MARK, "Launching osrf_forker for app %s", appname);
 137         prefork_run(forker);
 138
 139         osrfLogWarning( OSRF_LOG_MARK, "prefork_run() retuned - how??");
 140         prefork_free(forker);
 141         return 0;
 142
 143 }
 144
 145 /* sends the "register" packet to the specified router */
 146 static void osrf_prefork_send_router_registration(const char* appname, const char* routerName, const char* routerDomain) {
 147         transport_client* client = osrfSystemGetTransportClient();
 148         char* jid = va_list_to_string( "%s@%s/router", routerName, routerDomain );
 149         osrfLogInfo( OSRF_LOG_MARK, "%s registering with router %s", appname, jid );
 150         transport_message* msg = message_init("registering", NULL, NULL, jid, NULL );
 151         message_set_router_info( msg, NULL, NULL, appname, "register", 0 );
 152         client_send_message( client, msg );
 153         message_free( msg );
 154         free(jid);
 155 }
 156
 157 /** parses a single "complex" router configuration chunk */
 158 static void osrf_prefork_parse_router_chunk(const char* appname, jsonObject* routerChunk) {
 159
 160         const char* routerName = jsonObjectGetString(jsonObjectGetKeyConst(routerChunk, "name"));
 161         const char* domain = jsonObjectGetString(jsonObjectGetKeyConst(routerChunk, "domain"));
 162         const jsonObject* services = jsonObjectGetKeyConst(routerChunk, "services");
 163         osrfLogDebug(OSRF_LOG_MARK, "found router config with domain %s and name %s",
 164                         routerName, domain);
 165
 166         if( services && services->type == JSON_HASH ) {
 167                 osrfLogDebug(OSRF_LOG_MARK, "investigating router information...");
 168                 const jsonObject* service_obj = jsonObjectGetKeyConst(services, "service");
 169                 if( !service_obj )
 170                         ;    // do nothing (shouldn't happen)
 171                 else if( JSON_ARRAY == service_obj->type ) {
 172                         int j;
 173                         for(j = 0; j < service_obj->size; j++ ) {
 174                                 const char* service = jsonObjectGetString(jsonObjectGetIndex(service_obj, j));
 175                                 if( service && !strcmp( appname, service ))
 176                                         osrf_prefork_send_router_registration(appname, routerName, domain);
 177                         }
 178                 }
 179                 else if( JSON_STRING == service_obj->type ) {
 180                         if( !strcmp(appname, jsonObjectGetString( service_obj )) )
 181                                 osrf_prefork_send_router_registration(appname, routerName, domain);
 182                 }
 183         } else {
 184                 osrf_prefork_send_router_registration(appname, routerName, domain);
 185         }
 186 }
 187
 188 static void osrf_prefork_register_routers( const char* appname ) {
 189
 190         jsonObject* routerInfo = osrfConfigGetValueObject(NULL, "/routers/router");
 191
 192         int i;
 193         for(i = 0; i < routerInfo->size; i++) {
 194                 jsonObject* routerChunk = jsonObjectGetIndex(routerInfo, i);
 195
 196                 if(routerChunk->type == JSON_STRING) {
 197                         /* this accomodates simple router configs */
 198                         char* routerName = osrfConfigGetValue( NULL, "/router_name" );
 199                         char* domain = osrfConfigGetValue(NULL, "/routers/router");
 200                         osrfLogDebug(OSRF_LOG_MARK, "found simple router settings with router name %s",
 201                                         routerName);
 202                         osrf_prefork_send_router_registration(appname, routerName, domain);
 203
 204                 } else {
 205                         osrf_prefork_parse_router_chunk(appname, routerChunk);
 206                 }
 207         }
 208 }
 209
 210 static int prefork_child_init_hook(prefork_child* child) {
 211
 212         if(!child) return -1;
 213         osrfLogDebug( OSRF_LOG_MARK, "Child init hook for child %d", child->pid);
 214
 215         osrfSystemInitCache();
 216         char* resc = va_list_to_string("%s_drone",child->appname);
 217
 218         /* if we're a source-client, tell the logger now that we're a new process*/
 219         char* isclient = osrfConfigGetValue(NULL, "/client");
 220         if( isclient && !strcasecmp(isclient,"true") )
 221                 osrfLogSetIsClient(1);
 222         free(isclient);
 223
 224         /* we want to remove traces of our parents socket connection
 225         * so we can have our own */
 226         osrfSystemIgnoreTransportClient();
 227
 228         if(!osrfSystemBootstrapClientResc( NULL, NULL, resc)) {
 229                 osrfLogError( OSRF_LOG_MARK, "Unable to bootstrap client for osrf_prefork_run()");
 230                 free(resc);
 231                 return -1;
 232         }
 233
 234         free(resc);
 235
 236         if( ! osrfAppRunChildInit(child->appname) ) {
 237                 osrfLogDebug(OSRF_LOG_MARK, "Prefork child_init succeeded\n");
 238         } else {
 239                 osrfLogError(OSRF_LOG_MARK, "Prefork child_init failed\n");
 240                 return -1;
 241         }
 242
 243         set_proc_title( "OpenSRF Drone [%s]", child->appname );
 244         return 0;
 245 }
 246
 247 static void prefork_child_process_request(prefork_child* child, char* data) {
 248         if( !child ) return;
 249
 250         transport_client* client = osrfSystemGetTransportClient();
 251
 252         if(!client_connected(client)) {
 253                 osrfSystemIgnoreTransportClient();
 254                 osrfLogWarning(OSRF_LOG_MARK, "Reconnecting child to opensrf after disconnect...");
 255                 if(!osrf_system_bootstrap_client(NULL, NULL)) {
 256                         osrfLogError( OSRF_LOG_MARK,
 257                                 "Unable to bootstrap client in prefork_child_process_request()");
 258                         sleep(1);
 259                         osrf_prefork_child_exit(child);
 260                 }
 261         }
 262
 263         /* construct the message from the xml */
 264         transport_message* msg = new_message_from_xml( data );
 265
 266         osrfAppSession* session = osrf_stack_transport_handler(msg, child->appname);
 267         if(!session) return;
 268
 269         if( session->stateless && session->state != OSRF_SESSION_CONNECTED ) {
 270                 osrfAppSessionFree( session );
 271                 return;
 272         }
 273
 274         osrfLogDebug( OSRF_LOG_MARK, "Entering keepalive loop for session %s", session->session_id );
 275         int keepalive = child->keepalive;
 276         int retval;
 277         int recvd;
 278         time_t start;
 279         time_t end;
 280
 281         while(1) {
 282
 283                 osrfLogDebug(OSRF_LOG_MARK,
 284                                 "osrf_prefork calling queue_wait [%d] in keepalive loop", keepalive);
 285                 start   = time(NULL);
 286                 retval  = osrf_app_session_queue_wait(session, keepalive, &recvd);
 287                 end     = time(NULL);
 288
 289                 osrfLogDebug(OSRF_LOG_MARK, "Data received == %d", recvd);
 290
 291                 if(retval) {
 292                         osrfLogError(OSRF_LOG_MARK, "queue-wait returned non-success %d", retval);
 293                         break;
 294                 }
 295
 296                 /* see if the client disconnected from us */
 297                 if(session->state != OSRF_SESSION_CONNECTED) break;
 298
 299                 /* if no data was reveived within the timeout interval */
 300                 if( !recvd && (end - start) >= keepalive ) {
 301                         osrfLogInfo(OSRF_LOG_MARK,
 302                                         "No request was received in %d seconds, exiting stateful session", keepalive);
 303                         osrfAppSessionStatus(
 304                                         session,
 305                                         OSRF_STATUS_TIMEOUT,
 306                                         "osrfConnectStatus",
 307                                         0, "Disconnected on timeout" );
 308
 309                         break;
 310                 }
 311         }
 312
 313         osrfLogDebug( OSRF_LOG_MARK, "Exiting keepalive loop for session %s", session->session_id );
 314         osrfAppSessionFree( session );
 315         return;
 316 }
 317
 318
 319 static prefork_simple*  prefork_simple_init( transport_client* client,
 320                 int max_requests, int min_children, int max_children ) {
 321
 322         if( min_children > max_children ) {
 323                 osrfLogError( OSRF_LOG_MARK,  "min_children (%d) is greater "
 324                                 "than max_children (%d)", min_children, max_children );
 325                 return NULL;
 326         }
 327
 328         if( max_children > ABS_MAX_CHILDREN ) {
 329                 osrfLogError( OSRF_LOG_MARK,  "max_children (%d) is greater than ABS_MAX_CHILDREN (%d)",
 330                                 max_children, ABS_MAX_CHILDREN );
 331                 return NULL;
 332         }
 333
 334         osrfLogInfo(OSRF_LOG_MARK, "Prefork launching child with max_request=%d,"
 335                 "min_children=%d, max_children=%d", max_requests, min_children, max_children );
 336
 337         /* flesh out the struct */
 338         prefork_simple* prefork = safe_malloc(sizeof(prefork_simple));
 339         prefork->max_requests = max_requests;
 340         prefork->min_children = min_children;
 341         prefork->max_children = max_children;
 342         prefork->fd           = 0;
 343         prefork->data_to_child = 0;
 344         prefork->data_to_parent = 0;
 345         prefork->current_num_children = 0;
 346         prefork->keepalive    = 0;
 347         prefork->appname      = NULL;
 348         prefork->first_child  = NULL;
 349         prefork->connection   = client;
 350
 351         return prefork;
 352 }
 353
 354 static prefork_child* launch_child( prefork_simple* forker ) {
 355
 356         pid_t pid;
 357         int data_fd[2];
 358         int status_fd[2];
 359
 360         /* Set up the data pipes and add the child struct to the parent */
 361         if( pipe(data_fd) < 0 ) { /* build the data pipe*/
 362                 osrfLogError( OSRF_LOG_MARK,  "Pipe making error" );
 363                 return NULL;
 364         }
 365
 366         if( pipe(status_fd) < 0 ) {/* build the status pipe */
 367                 osrfLogError( OSRF_LOG_MARK,  "Pipe making error" );
 368                 return NULL;
 369         }
 370
 371         osrfLogInternal( OSRF_LOG_MARK,  "Pipes: %d %d %d %d",
 372                         data_fd[0], data_fd[1], status_fd[0], status_fd[1] );
 373         prefork_child* child = prefork_child_init( forker->max_requests, data_fd[0],
 374                         data_fd[1], status_fd[0], status_fd[1] );
 375
 376         child->appname = strdup(forker->appname);
 377         child->keepalive = forker->keepalive;
 378
 379
 380         add_prefork_child( forker, child );
 381
 382         if( (pid=fork()) < 0 ) {
 383                 osrfLogError( OSRF_LOG_MARK,  "Forking Error" );
 384                 return NULL;
 385         }
 386
 387         if( pid > 0 ) {  /* parent */
 388
 389                 signal(SIGCHLD, sigchld_handler);
 390                 (forker->current_num_children)++;
 391                 child->pid = pid;
 392
 393                 osrfLogDebug( OSRF_LOG_MARK,  "Parent launched %d", pid );
 394                 /* *no* child pipe FD's can be closed or the parent will re-use fd's that
 395                         the children are currently using */
 396                 return child;
 397         }
 398
 399         else { /* child */
 400
 401                 osrfLogInternal( OSRF_LOG_MARK,
 402                                 "I am  new child with read_data_fd = %d and write_status_fd = %d",
 403                                 child->read_data_fd, child->write_status_fd );
 404
 405                 child->pid = getpid();
 406                 close( child->write_data_fd );
 407                 close( child->read_status_fd );
 408
 409                 /* do the initing */
 410                 if( prefork_child_init_hook(child) == -1 ) {
 411                         osrfLogError(OSRF_LOG_MARK,
 412                                 "Forker child going away because we could not connect to OpenSRF...");
 413                         osrf_prefork_child_exit(child);
 414                 }
 415
 416                 prefork_child_wait( child );
 417                 osrf_prefork_child_exit(child); /* just to be sure */
 418         }
 419         return NULL;
 420 }
 421
 422 static void osrf_prefork_child_exit(prefork_child* child) {
 423         osrfAppRunExitCode();
 424         exit(0);
 425 }
 426
 427 static void prefork_launch_children( prefork_simple* forker ) {
 428         if(!forker) return;
 429         int c = 0;
 430         while( c++ < forker->min_children )
 431                 launch_child( forker );
 432 }
 433
 434
 435 static void sigchld_handler( int sig ) {
 436         signal(SIGCHLD, sigchld_handler);
 437         child_dead = 1;
 438 }
 439
 440
 441 void reap_children( prefork_simple* forker ) {
 442
 443         pid_t child_pid;
 444         int status;
 445
 446         while( (child_pid=waitpid(-1,&status,WNOHANG)) > 0)
 447                 del_prefork_child( forker, child_pid );
 448
 449         /* replenish */
 450         while( forker->current_num_children < forker->min_children )
 451                 launch_child( forker );
 452
 453         child_dead = 0;
 454 }
 455
 456 static void prefork_run(prefork_simple* forker) {
 457
 458         if( forker->first_child == NULL )
 459                 return;
 460
 461         transport_message* cur_msg = NULL;
 462
 463
 464         while(1) {
 465
 466                 if( forker->first_child == NULL ) {/* no more children */
 467                         osrfLogWarning( OSRF_LOG_MARK, "No more children..." );
 468                         return;
 469                 }
 470
 471                 osrfLogDebug( OSRF_LOG_MARK, "Forker going into wait for data...");
 472                 cur_msg = client_recv( forker->connection, -1 );
 473
 474                 //fprintf(stderr, "Got Data %f\n", get_timestamp_millis() );
 475
 476                 if( cur_msg == NULL ) continue;
 477
 478                 int honored = 0;     /* true if we've serviced the request */
 479                 int no_recheck = 0;
 480
 481                 while( ! honored ) {
 482
 483                         if(!no_recheck) check_children( forker, 0 );
 484                         no_recheck = 0;
 485
 486                         osrfLogDebug( OSRF_LOG_MARK,  "Server received inbound data" );
 487                         int k;
 488                         prefork_child* cur_child = forker->first_child;
 489
 490                         /* Look for an available child */
 491                         for( k = 0; k < forker->current_num_children; k++ ) {
 492
 493                                 osrfLogInternal( OSRF_LOG_MARK,
 494                                                 "Searching for available child. cur_child->pid = %d", cur_child->pid );
 495                                 osrfLogInternal( OSRF_LOG_MARK, "Current num children %d and loop %d",
 496                                                 forker->current_num_children, k);
 497
 498                                 if( cur_child->available ) {
 499                                         osrfLogDebug( OSRF_LOG_MARK,  "forker sending data to %d", cur_child->pid );
 500
 501                                         message_prepare_xml( cur_msg );
 502                                         char* data = cur_msg->msg_xml;
 503                                         if( ! data || strlen(data) < 1 ) break;
 504
 505                                         cur_child->available = 0;
 506                                         osrfLogInternal( OSRF_LOG_MARK, "Writing to child fd %d",
 507                                                         cur_child->write_data_fd );
 508
 509                                         int written = 0;
 510                                         if( (written = write( cur_child->write_data_fd, data, strlen(data) + 1 )) < 0 ) {
 511                                                 osrfLogWarning( OSRF_LOG_MARK, "Write returned error %d", errno);
 512                                                 cur_child = cur_child->next;
 513                                                 continue;
 514                                         }
 515
 516                                         forker->first_child = cur_child->next;
 517                                         honored = 1;
 518                                         break;
 519                                 } else
 520                                         cur_child = cur_child->next;
 521                         }
 522
 523                         /* if none available, add a new child if we can */
 524                         if( ! honored ) {
 525                                 osrfLogDebug( OSRF_LOG_MARK, "Not enough children, attempting to add...");
 526
 527                                 if( forker->current_num_children < forker->max_children ) {
 528                                         osrfLogDebug( OSRF_LOG_MARK,  "Launching new child with current_num = %d",
 529                                                         forker->current_num_children );
 530
 531                                         prefork_child* new_child = launch_child( forker );
 532                                         if( new_child ) {
 533
 534                                                 message_prepare_xml( cur_msg );
 535                                                 char* data = cur_msg->msg_xml;
 536
 537                                                 if( data ) {
 538                                                         int len = strlen(data);
 539
 540                                                         if( len > 0 ) {
 541                                                                 new_child->available = 0;
 542                                                                 osrfLogDebug( OSRF_LOG_MARK, "Writing to new child fd %d : pid %d",
 543                                                                                 new_child->write_data_fd, new_child->pid );
 544
 545                                                                 if( write( new_child->write_data_fd, data, len + 1 ) >= 0 ) {
 546                                                                         forker->first_child = new_child->next;
 547                                                                         honored = 1;
 548                                                                 }
 549                                                         }
 550                                                 }
 551                                         }
 552
 553                                 }
 554                         }
 555
 556                         if( !honored ) {
 557                                 osrfLogWarning( OSRF_LOG_MARK,  "No children available, waiting...");
 558
 559                                 check_children( forker, 1 );  /* non-poll version */
 560                                 /* tell the loop no to call check_children again, since we're calling it now */
 561                                 no_recheck = 1;
 562                         }
 563
 564                         if( child_dead )
 565                                 reap_children(forker);
 566
 567
 568                         //fprintf(stderr, "Parent done with request %f\n", get_timestamp_millis() );
 569
 570                 } // honored?
 571
 572                 message_free( cur_msg );
 573
 574         } /* top level listen loop */
 575
 576 }
 577
 578
 579 /** XXX Add a flag which tells select() to wait forever on children
 580         in the best case, this will be faster than calling usleep(x), and
 581         in the worst case it won't be slower and will do less logging...
 582 */
 583 static void check_children( prefork_simple* forker, int forever ) {
 584
 585         //check_begin:
 586
 587         int select_ret;
 588         fd_set read_set;
 589         FD_ZERO(&read_set);
 590         int max_fd = 0;
 591         int n;
 592
 593
 594         if( child_dead )
 595                 reap_children(forker);
 596
 597         prefork_child* cur_child = forker->first_child;
 598
 599         int i;
 600         for( i = 0; i!= forker->current_num_children; i++ ) {
 601
 602                 if( cur_child->read_status_fd > max_fd )
 603                         max_fd = cur_child->read_status_fd;
 604                 FD_SET( cur_child->read_status_fd, &read_set );
 605                 cur_child = cur_child->next;
 606         }
 607
 608         FD_CLR(0,&read_set); /* just to be sure */
 609
 610         if( forever ) {
 611                 osrfLogWarning(OSRF_LOG_MARK,
 612                                 "We have no children available - waiting for one to show up...");
 613
 614                 if( (select_ret=select( max_fd + 1 , &read_set, NULL, NULL, NULL)) == -1 ) {
 615                         osrfLogWarning( OSRF_LOG_MARK,  "Select returned error %d on check_children", errno );
 616                 }
 617                 osrfLogInfo(OSRF_LOG_MARK,
 618                                 "select() completed after waiting on children to become available");
 619
 620         } else {
 621
 622                 struct timeval tv;
 623                 tv.tv_sec   = 0;
 624                 tv.tv_usec  = 0;
 625
 626                 if( (select_ret=select( max_fd + 1 , &read_set, NULL, NULL, &tv)) == -1 ) {
 627                         osrfLogWarning( OSRF_LOG_MARK,  "Select returned error %d on check_children", errno );
 628                 }
 629         }
 630
 631         if( select_ret == 0 )
 632                 return;
 633
 634         /* see if one of a child has told us it's done */
 635         cur_child = forker->first_child;
 636         int j;
 637         int num_handled = 0;
 638         for( j = 0; j!= forker->current_num_children && num_handled < select_ret ; j++ ) {
 639
 640                 if( FD_ISSET( cur_child->read_status_fd, &read_set ) ) {
 641                         //printf( "Server received status from a child %d\n", cur_child->pid );
 642                         osrfLogDebug( OSRF_LOG_MARK,
 643                                         "Server received status from a child %d", cur_child->pid );
 644
 645                         num_handled++;
 646
 647                         /* now suck off the data */
 648                         char buf[64];
 649                         osrf_clearbuf( buf, sizeof(buf) );
 650                         if( (n=read(cur_child->read_status_fd, buf, sizeof(buf) - 1)) < 0 ) {
 651                                 osrfLogWarning( OSRF_LOG_MARK,
 652                                                 "Read error after select in child status read with errno %d", errno);
 653                         }
 654                         else {
 655                                 buf[n] = '\0';
 656                                 osrfLogDebug( OSRF_LOG_MARK,  "Read %d bytes from status buffer: %s", n, buf );
 657                         }
 658                         cur_child->available = 1;
 659                 }
 660                 cur_child = cur_child->next;
 661         }
 662
 663 }
 664
 665
 666 static void prefork_child_wait( prefork_child* child ) {
 667
 668         int i,n;
 669         growing_buffer* gbuf = buffer_init( READ_BUFSIZE );
 670         char buf[READ_BUFSIZE];
 671         osrf_clearbuf( buf, sizeof(buf) );
 672
 673         for( i = 0; i < child->max_requests; i++ ) {
 674
 675                 n = -1;
 676                 int gotdata = 0;
 677                 clr_fl(child->read_data_fd, O_NONBLOCK );
 678
 679                 while( (n=read(child->read_data_fd, buf, READ_BUFSIZE-1)) > 0 ) {
 680                         buf[n] = '\0';
 681                         osrfLogDebug(OSRF_LOG_MARK, "Prefork child read %d bytes of data", n);
 682                         if(!gotdata)
 683                                 set_fl(child->read_data_fd, O_NONBLOCK );
 684                         buffer_add( gbuf, buf );
 685                         osrf_clearbuf( buf, sizeof(buf) );
 686                         gotdata = 1;
 687                 }
 688
 689                 if( errno == EAGAIN ) n = 0;
 690
 691                 if( errno == EPIPE ) {
 692                         osrfLogDebug(OSRF_LOG_MARK, "C child attempted read on broken pipe, exiting...");
 693                         break;
 694                 }
 695
 696                 if( n < 0 ) {
 697                         osrfLogWarning( OSRF_LOG_MARK,
 698                                         "Prefork child read returned error with errno %d", errno );
 699                         break;
 700
 701                 } else if( gotdata ) {
 702                         osrfLogDebug(OSRF_LOG_MARK, "Prefork child got a request.. processing..");
 703                         prefork_child_process_request(child, gbuf->buf);
 704                         buffer_reset( gbuf );
 705                 }
 706
 707                 if( i < child->max_requests - 1 )
 708                         write( child->write_status_fd, "available" /*less than 64 bytes*/, 9 );
 709         }
 710
 711         buffer_free(gbuf);
 712
 713         osrfLogDebug( OSRF_LOG_MARK, "Child with max-requests=%d, num-served=%d exiting...[%ld]",
 714                         child->max_requests, i, (long) getpid() );
 715
 716         osrf_prefork_child_exit(child); /* just to be sure */
 717 }
 718
 719
 720 static void add_prefork_child( prefork_simple* forker, prefork_child* child ) {
 721
 722         if( forker->first_child == NULL ) {
 723                 forker->first_child = child;
 724                 child->next = child;
 725                 return;
 726         }
 727
 728         /* we put the child in as the last because, regardless,
 729                 we have to do the DLL splice dance, and this is the
 730                 simplest way */
 731
 732         prefork_child* start_child = forker->first_child;
 733         while(1) {
 734                 if( forker->first_child->next == start_child )
 735                         break;
 736                 forker->first_child = forker->first_child->next;
 737         }
 738
 739         /* here we know that forker->first_child is the last element
 740                 in the list and start_child is the first.  Insert the
 741                 new child between them*/
 742
 743         forker->first_child->next = child;
 744         child->next = start_child;
 745         return;
 746 }
 747
 748 //static prefork_child* find_prefork_child( prefork_simple* forker, pid_t pid ) {
 749 //
 750 //  if( forker->first_child == NULL ) { return NULL; }
 751 //  prefork_child* start_child = forker->first_child;
 752 //  do {
 753 //      if( forker->first_child->pid == pid )
 754 //          return forker->first_child;
 755 //  } while( (forker->first_child = forker->first_child->next) != start_child );
 756 //
 757 //  return NULL;
 758 //}
 759
 760
 761 static void del_prefork_child( prefork_simple* forker, pid_t pid ) {
 762
 763         if( forker->first_child == NULL ) { return; }
 764
 765         (forker->current_num_children)--;
 766         osrfLogDebug( OSRF_LOG_MARK, "Deleting Child: %d", pid );
 767
 768         prefork_child* start_child = forker->first_child; /* starting point */
 769         prefork_child* cur_child   = start_child; /* current pointer */
 770         prefork_child* prev_child  = start_child; /* the trailing pointer */
 771
 772         /* special case where there is only one in the list */
 773         if( start_child == start_child->next ) {
 774                 if( start_child->pid == pid ) {
 775                         forker->first_child = NULL;
 776
 777                         close( start_child->read_data_fd );
 778                         close( start_child->write_data_fd );
 779                         close( start_child->read_status_fd );
 780                         close( start_child->write_status_fd );
 781
 782                         prefork_child_free( start_child );
 783                 }
 784                 return;
 785         }
 786
 787
 788         /* special case where the first item in the list needs to be removed */
 789         if( start_child->pid == pid ) {
 790
 791                 /* find the last one so we can remove the start_child */
 792                 do {
 793                         prev_child = cur_child;
 794                         cur_child = cur_child->next;
 795                 } while( cur_child != start_child );
 796
 797                 /* now cur_child == start_child */
 798                 prev_child->next = cur_child->next;
 799                 forker->first_child = prev_child;
 800
 801                 close( cur_child->read_data_fd );
 802                 close( cur_child->write_data_fd );
 803                 close( cur_child->read_status_fd );
 804                 close( cur_child->write_status_fd );
 805
 806                 prefork_child_free( cur_child );
 807                 return;
 808         }
 809
 810         do {
 811                 prev_child = cur_child;
 812                 cur_child = cur_child->next;
 813
 814                 if( cur_child->pid == pid ) {
 815                         prev_child->next = cur_child->next;
 816
 817                         close( cur_child->read_data_fd );
 818                         close( cur_child->write_data_fd );
 819                         close( cur_child->read_status_fd );
 820                         close( cur_child->write_status_fd );
 821
 822                         prefork_child_free( cur_child );
 823                         return;
 824                 }
 825
 826         } while(cur_child != start_child);
 827 }
 828
 829
 830
 831
 832 static prefork_child* prefork_child_init(
 833         int max_requests, int read_data_fd, int write_data_fd,
 834         int read_status_fd, int write_status_fd ) {
 835
 836         prefork_child* child = (prefork_child*) safe_malloc(sizeof(prefork_child));
 837         child->pid              = 0;
 838         child->max_requests     = max_requests;
 839         child->read_data_fd     = read_data_fd;
 840         child->write_data_fd    = write_data_fd;
 841         child->read_status_fd   = read_status_fd;
 842         child->write_status_fd  = write_status_fd;
 843         child->min_children     = 0;
 844         child->available        = 1;
 845         child->appname          = NULL;
 846         child->keepalive        = 0;
 847         child->next             = NULL;
 848
 849         return child;
 850 }
 851
 852
 853 static int prefork_free( prefork_simple* prefork ) {
 854
 855         while( prefork->first_child != NULL ) {
 856                 osrfLogInfo( OSRF_LOG_MARK, "Killing children and sleeping 1 to reap..." );
 857                 kill( 0, SIGKILL );
 858                 sleep(1);
 859         }
 860
 861         client_free(prefork->connection);
 862         free(prefork->appname);
 863         free( prefork );
 864         return 1;
 865 }
 866
 867 static int prefork_child_free( prefork_child* child ) {
 868         free(child->appname);
 869         close(child->read_data_fd);
 870         close(child->write_status_fd);
 871         free( child );
 872         return 1;
 873 }