src/libopensrf/osrf_prefork.c

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