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