1 /* 2 * work_thread.c - threads implementation for blocking worker child. 3 */ 4 #include <config.h> 5 #include "ntp_workimpl.h" 6 7 #ifdef WORK_THREAD 8 9 #include <stdio.h> 10 #include <ctype.h> 11 #include <signal.h> 12 #ifndef SYS_WINNT 13 #include <pthread.h> 14 #endif 15 16 #include "ntp_stdlib.h" 17 #include "ntp_malloc.h" 18 #include "ntp_syslog.h" 19 #include "ntpd.h" 20 #include "ntp_io.h" 21 #include "ntp_assert.h" 22 #include "ntp_unixtime.h" 23 #include "timespecops.h" 24 #include "ntp_worker.h" 25 26 #define CHILD_EXIT_REQ ((blocking_pipe_header *)(intptr_t)-1) 27 #define CHILD_GONE_RESP CHILD_EXIT_REQ 28 /* Queue size increments: 29 * The request queue grows a bit faster than the response queue -- the 30 * deamon can push requests and pull results faster on avarage than the 31 * worker can process requests and push results... If this really pays 32 * off is debatable. 33 */ 34 #define WORKITEMS_ALLOC_INC 16 35 #define RESPONSES_ALLOC_INC 4 36 37 /* Fiddle with min/max stack sizes. 64kB minimum seems to work, so we 38 * set the maximum to 256kB. If the minimum goes below the 39 * system-defined minimum stack size, we have to adjust accordingly. 40 */ 41 #ifndef THREAD_MINSTACKSIZE 42 # define THREAD_MINSTACKSIZE (64U * 1024) 43 #endif 44 #ifndef __sun 45 #if defined(PTHREAD_STACK_MIN) && THREAD_MINSTACKSIZE < PTHREAD_STACK_MIN 46 # undef THREAD_MINSTACKSIZE 47 # define THREAD_MINSTACKSIZE PTHREAD_STACK_MIN 48 #endif 49 #endif 50 51 #ifndef THREAD_MAXSTACKSIZE 52 # define THREAD_MAXSTACKSIZE (256U * 1024) 53 #endif 54 #if THREAD_MAXSTACKSIZE < THREAD_MINSTACKSIZE 55 # undef THREAD_MAXSTACKSIZE 56 # define THREAD_MAXSTACKSIZE THREAD_MINSTACKSIZE 57 #endif 58 59 60 #ifdef SYS_WINNT 61 62 # define thread_exit(c) _endthreadex(c) 63 # define tickle_sem(sh) ReleaseSemaphore((sh->shnd), 1, NULL) 64 u_int WINAPI blocking_thread(void *); 65 static BOOL same_os_sema(const sem_ref obj, void * osobj); 66 67 #else 68 69 # define thread_exit(c) pthread_exit((void*)(size_t)(c)) 70 # define tickle_sem sem_post 71 void * blocking_thread(void *); 72 static void block_thread_signals(sigset_t *); 73 74 #endif 75 76 #ifdef WORK_PIPE 77 addremove_io_fd_func addremove_io_fd; 78 #else 79 addremove_io_semaphore_func addremove_io_semaphore; 80 #endif 81 82 static void start_blocking_thread(blocking_child *); 83 static void start_blocking_thread_internal(blocking_child *); 84 static void prepare_child_sems(blocking_child *); 85 static int wait_for_sem(sem_ref, struct timespec *); 86 static int ensure_workitems_empty_slot(blocking_child *); 87 static int ensure_workresp_empty_slot(blocking_child *); 88 static int queue_req_pointer(blocking_child *, blocking_pipe_header *); 89 static void cleanup_after_child(blocking_child *); 90 91 static sema_type worker_mmutex; 92 static sem_ref worker_memlock; 93 94 /* -------------------------------------------------------------------- 95 * locking the global worker state table (and other global stuff) 96 */ 97 void 98 worker_global_lock( 99 int inOrOut) 100 { 101 if (worker_memlock) { 102 if (inOrOut) 103 wait_for_sem(worker_memlock, NULL); 104 else 105 tickle_sem(worker_memlock); 106 } 107 } 108 109 /* -------------------------------------------------------------------- 110 * implementation isolation wrapper 111 */ 112 void 113 exit_worker( 114 int exitcode 115 ) 116 { 117 thread_exit(exitcode); /* see #define thread_exit */ 118 } 119 120 /* -------------------------------------------------------------------- 121 * sleep for a given time or until the wakup semaphore is tickled. 122 */ 123 int 124 worker_sleep( 125 blocking_child * c, 126 time_t seconds 127 ) 128 { 129 struct timespec until; 130 int rc; 131 132 # ifdef HAVE_CLOCK_GETTIME 133 if (0 != clock_gettime(CLOCK_REALTIME, &until)) { 134 msyslog(LOG_ERR, "worker_sleep: clock_gettime() failed: %m"); 135 return -1; 136 } 137 # else 138 if (0 != getclock(TIMEOFDAY, &until)) { 139 msyslog(LOG_ERR, "worker_sleep: getclock() failed: %m"); 140 return -1; 141 } 142 # endif 143 until.tv_sec += seconds; 144 rc = wait_for_sem(c->wake_scheduled_sleep, &until); 145 if (0 == rc) 146 return -1; 147 if (-1 == rc && ETIMEDOUT == errno) 148 return 0; 149 msyslog(LOG_ERR, "worker_sleep: sem_timedwait: %m"); 150 return -1; 151 } 152 153 154 /* -------------------------------------------------------------------- 155 * Wake up a worker that takes a nap. 156 */ 157 void 158 interrupt_worker_sleep(void) 159 { 160 u_int idx; 161 blocking_child * c; 162 163 for (idx = 0; idx < blocking_children_alloc; idx++) { 164 c = blocking_children[idx]; 165 if (NULL == c || NULL == c->wake_scheduled_sleep) 166 continue; 167 tickle_sem(c->wake_scheduled_sleep); 168 } 169 } 170 171 /* -------------------------------------------------------------------- 172 * Make sure there is an empty slot at the head of the request 173 * queue. Tell if the queue is currently empty. 174 */ 175 static int 176 ensure_workitems_empty_slot( 177 blocking_child *c 178 ) 179 { 180 /* 181 ** !!! PRECONDITION: caller holds access lock! 182 ** 183 ** This simply tries to increase the size of the buffer if it 184 ** becomes full. The resize operation does *not* maintain the 185 ** order of requests, but that should be irrelevant since the 186 ** processing is considered asynchronous anyway. 187 ** 188 ** Return if the buffer is currently empty. 189 */ 190 191 static const size_t each = 192 sizeof(blocking_children[0]->workitems[0]); 193 194 size_t new_alloc; 195 size_t slots_used; 196 size_t sidx; 197 198 slots_used = c->head_workitem - c->tail_workitem; 199 if (slots_used >= c->workitems_alloc) { 200 new_alloc = c->workitems_alloc + WORKITEMS_ALLOC_INC; 201 c->workitems = erealloc(c->workitems, new_alloc * each); 202 for (sidx = c->workitems_alloc; sidx < new_alloc; ++sidx) 203 c->workitems[sidx] = NULL; 204 c->tail_workitem = 0; 205 c->head_workitem = c->workitems_alloc; 206 c->workitems_alloc = new_alloc; 207 } 208 INSIST(NULL == c->workitems[c->head_workitem % c->workitems_alloc]); 209 return (0 == slots_used); 210 } 211 212 /* -------------------------------------------------------------------- 213 * Make sure there is an empty slot at the head of the response 214 * queue. Tell if the queue is currently empty. 215 */ 216 static int 217 ensure_workresp_empty_slot( 218 blocking_child *c 219 ) 220 { 221 /* 222 ** !!! PRECONDITION: caller holds access lock! 223 ** 224 ** Works like the companion function above. 225 */ 226 227 static const size_t each = 228 sizeof(blocking_children[0]->responses[0]); 229 230 size_t new_alloc; 231 size_t slots_used; 232 size_t sidx; 233 234 slots_used = c->head_response - c->tail_response; 235 if (slots_used >= c->responses_alloc) { 236 new_alloc = c->responses_alloc + RESPONSES_ALLOC_INC; 237 c->responses = erealloc(c->responses, new_alloc * each); 238 for (sidx = c->responses_alloc; sidx < new_alloc; ++sidx) 239 c->responses[sidx] = NULL; 240 c->tail_response = 0; 241 c->head_response = c->responses_alloc; 242 c->responses_alloc = new_alloc; 243 } 244 INSIST(NULL == c->responses[c->head_response % c->responses_alloc]); 245 return (0 == slots_used); 246 } 247 248 249 /* -------------------------------------------------------------------- 250 * queue_req_pointer() - append a work item or idle exit request to 251 * blocking_workitems[]. Employ proper locking. 252 */ 253 static int 254 queue_req_pointer( 255 blocking_child * c, 256 blocking_pipe_header * hdr 257 ) 258 { 259 size_t qhead; 260 261 /* >>>> ACCESS LOCKING STARTS >>>> */ 262 wait_for_sem(c->accesslock, NULL); 263 ensure_workitems_empty_slot(c); 264 qhead = c->head_workitem; 265 c->workitems[qhead % c->workitems_alloc] = hdr; 266 c->head_workitem = 1 + qhead; 267 tickle_sem(c->accesslock); 268 /* <<<< ACCESS LOCKING ENDS <<<< */ 269 270 /* queue consumer wake-up notification */ 271 tickle_sem(c->workitems_pending); 272 273 return 0; 274 } 275 276 /* -------------------------------------------------------------------- 277 * API function to make sure a worker is running, a proper private copy 278 * of the data is made, the data eneterd into the queue and the worker 279 * is signalled. 280 */ 281 int 282 send_blocking_req_internal( 283 blocking_child * c, 284 blocking_pipe_header * hdr, 285 void * data 286 ) 287 { 288 blocking_pipe_header * threadcopy; 289 size_t payload_octets; 290 291 REQUIRE(hdr != NULL); 292 REQUIRE(data != NULL); 293 DEBUG_REQUIRE(BLOCKING_REQ_MAGIC == hdr->magic_sig); 294 295 if (hdr->octets <= sizeof(*hdr)) 296 return 1; /* failure */ 297 payload_octets = hdr->octets - sizeof(*hdr); 298 299 if (NULL == c->thread_ref) 300 start_blocking_thread(c); 301 threadcopy = emalloc(hdr->octets); 302 memcpy(threadcopy, hdr, sizeof(*hdr)); 303 memcpy((char *)threadcopy + sizeof(*hdr), data, payload_octets); 304 305 return queue_req_pointer(c, threadcopy); 306 } 307 308 /* -------------------------------------------------------------------- 309 * Wait for the 'incoming queue no longer empty' signal, lock the shared 310 * structure and dequeue an item. 311 */ 312 blocking_pipe_header * 313 receive_blocking_req_internal( 314 blocking_child * c 315 ) 316 { 317 blocking_pipe_header * req; 318 size_t qhead, qtail; 319 320 req = NULL; 321 do { 322 /* wait for tickle from the producer side */ 323 wait_for_sem(c->workitems_pending, NULL); 324 325 /* >>>> ACCESS LOCKING STARTS >>>> */ 326 wait_for_sem(c->accesslock, NULL); 327 qhead = c->head_workitem; 328 do { 329 qtail = c->tail_workitem; 330 if (qhead == qtail) 331 break; 332 c->tail_workitem = qtail + 1; 333 qtail %= c->workitems_alloc; 334 req = c->workitems[qtail]; 335 c->workitems[qtail] = NULL; 336 } while (NULL == req); 337 tickle_sem(c->accesslock); 338 /* <<<< ACCESS LOCKING ENDS <<<< */ 339 340 } while (NULL == req); 341 342 INSIST(NULL != req); 343 if (CHILD_EXIT_REQ == req) { /* idled out */ 344 send_blocking_resp_internal(c, CHILD_GONE_RESP); 345 req = NULL; 346 } 347 348 return req; 349 } 350 351 /* -------------------------------------------------------------------- 352 * Push a response into the return queue and eventually tickle the 353 * receiver. 354 */ 355 int 356 send_blocking_resp_internal( 357 blocking_child * c, 358 blocking_pipe_header * resp 359 ) 360 { 361 size_t qhead; 362 int empty; 363 364 /* >>>> ACCESS LOCKING STARTS >>>> */ 365 wait_for_sem(c->accesslock, NULL); 366 empty = ensure_workresp_empty_slot(c); 367 qhead = c->head_response; 368 c->responses[qhead % c->responses_alloc] = resp; 369 c->head_response = 1 + qhead; 370 tickle_sem(c->accesslock); 371 /* <<<< ACCESS LOCKING ENDS <<<< */ 372 373 /* queue consumer wake-up notification */ 374 if (empty) 375 { 376 # ifdef WORK_PIPE 377 write(c->resp_write_pipe, "", 1); 378 # else 379 tickle_sem(c->responses_pending); 380 # endif 381 } 382 return 0; 383 } 384 385 386 #ifndef WORK_PIPE 387 388 /* -------------------------------------------------------------------- 389 * Check if a (Windows-)hanndle to a semaphore is actually the same we 390 * are using inside the sema wrapper. 391 */ 392 static BOOL 393 same_os_sema( 394 const sem_ref obj, 395 void* osh 396 ) 397 { 398 return obj && osh && (obj->shnd == (HANDLE)osh); 399 } 400 401 /* -------------------------------------------------------------------- 402 * Find the shared context that associates to an OS handle and make sure 403 * the data is dequeued and processed. 404 */ 405 void 406 handle_blocking_resp_sem( 407 void * context 408 ) 409 { 410 blocking_child * c; 411 u_int idx; 412 413 c = NULL; 414 for (idx = 0; idx < blocking_children_alloc; idx++) { 415 c = blocking_children[idx]; 416 if (c != NULL && 417 c->thread_ref != NULL && 418 same_os_sema(c->responses_pending, context)) 419 break; 420 } 421 if (idx < blocking_children_alloc) 422 process_blocking_resp(c); 423 } 424 #endif /* !WORK_PIPE */ 425 426 /* -------------------------------------------------------------------- 427 * Fetch the next response from the return queue. In case of signalling 428 * via pipe, make sure the pipe is flushed, too. 429 */ 430 blocking_pipe_header * 431 receive_blocking_resp_internal( 432 blocking_child * c 433 ) 434 { 435 blocking_pipe_header * removed; 436 size_t qhead, qtail, slot; 437 438 #ifdef WORK_PIPE 439 int rc; 440 char scratch[32]; 441 442 do 443 rc = read(c->resp_read_pipe, scratch, sizeof(scratch)); 444 while (-1 == rc && EINTR == errno); 445 #endif 446 447 /* >>>> ACCESS LOCKING STARTS >>>> */ 448 wait_for_sem(c->accesslock, NULL); 449 qhead = c->head_response; 450 qtail = c->tail_response; 451 for (removed = NULL; !removed && (qhead != qtail); ++qtail) { 452 slot = qtail % c->responses_alloc; 453 removed = c->responses[slot]; 454 c->responses[slot] = NULL; 455 } 456 c->tail_response = qtail; 457 tickle_sem(c->accesslock); 458 /* <<<< ACCESS LOCKING ENDS <<<< */ 459 460 if (NULL != removed) { 461 DEBUG_ENSURE(CHILD_GONE_RESP == removed || 462 BLOCKING_RESP_MAGIC == removed->magic_sig); 463 } 464 if (CHILD_GONE_RESP == removed) { 465 cleanup_after_child(c); 466 removed = NULL; 467 } 468 469 return removed; 470 } 471 472 /* -------------------------------------------------------------------- 473 * Light up a new worker. 474 */ 475 static void 476 start_blocking_thread( 477 blocking_child * c 478 ) 479 { 480 481 DEBUG_INSIST(!c->reusable); 482 483 prepare_child_sems(c); 484 start_blocking_thread_internal(c); 485 } 486 487 /* -------------------------------------------------------------------- 488 * Create a worker thread. There are several differences between POSIX 489 * and Windows, of course -- most notably the Windows thread is no 490 * detached thread, and we keep the handle around until we want to get 491 * rid of the thread. The notification scheme also differs: Windows 492 * makes use of semaphores in both directions, POSIX uses a pipe for 493 * integration with 'select()' or alike. 494 */ 495 static void 496 start_blocking_thread_internal( 497 blocking_child * c 498 ) 499 #ifdef SYS_WINNT 500 { 501 BOOL resumed; 502 503 c->thread_ref = NULL; 504 (*addremove_io_semaphore)(c->responses_pending->shnd, FALSE); 505 c->thr_table[0].thnd = 506 (HANDLE)_beginthreadex( 507 NULL, 508 0, 509 &blocking_thread, 510 c, 511 CREATE_SUSPENDED, 512 NULL); 513 514 if (NULL == c->thr_table[0].thnd) { 515 msyslog(LOG_ERR, "start blocking thread failed: %m"); 516 exit(-1); 517 } 518 /* remember the thread priority is only within the process class */ 519 if (!SetThreadPriority(c->thr_table[0].thnd, 520 THREAD_PRIORITY_BELOW_NORMAL)) 521 msyslog(LOG_ERR, "Error lowering blocking thread priority: %m"); 522 523 resumed = ResumeThread(c->thr_table[0].thnd); 524 DEBUG_INSIST(resumed); 525 c->thread_ref = &c->thr_table[0]; 526 } 527 #else /* pthreads start_blocking_thread_internal() follows */ 528 { 529 # ifdef NEED_PTHREAD_INIT 530 static int pthread_init_called; 531 # endif 532 pthread_attr_t thr_attr; 533 int rc; 534 int pipe_ends[2]; /* read then write */ 535 int is_pipe; 536 int flags; 537 size_t ostacksize; 538 size_t nstacksize; 539 sigset_t saved_sig_mask; 540 541 c->thread_ref = NULL; 542 543 # ifdef NEED_PTHREAD_INIT 544 /* 545 * from lib/isc/unix/app.c: 546 * BSDI 3.1 seg faults in pthread_sigmask() if we don't do this. 547 */ 548 if (!pthread_init_called) { 549 pthread_init(); 550 pthread_init_called = TRUE; 551 } 552 # endif 553 554 rc = pipe_socketpair(&pipe_ends[0], &is_pipe); 555 if (0 != rc) { 556 msyslog(LOG_ERR, "start_blocking_thread: pipe_socketpair() %m"); 557 exit(1); 558 } 559 c->resp_read_pipe = move_fd(pipe_ends[0]); 560 c->resp_write_pipe = move_fd(pipe_ends[1]); 561 c->ispipe = is_pipe; 562 flags = fcntl(c->resp_read_pipe, F_GETFL, 0); 563 if (-1 == flags) { 564 msyslog(LOG_ERR, "start_blocking_thread: fcntl(F_GETFL) %m"); 565 exit(1); 566 } 567 rc = fcntl(c->resp_read_pipe, F_SETFL, O_NONBLOCK | flags); 568 if (-1 == rc) { 569 msyslog(LOG_ERR, 570 "start_blocking_thread: fcntl(F_SETFL, O_NONBLOCK) %m"); 571 exit(1); 572 } 573 (*addremove_io_fd)(c->resp_read_pipe, c->ispipe, FALSE); 574 pthread_attr_init(&thr_attr); 575 pthread_attr_setdetachstate(&thr_attr, PTHREAD_CREATE_DETACHED); 576 #if defined(HAVE_PTHREAD_ATTR_GETSTACKSIZE) && \ 577 defined(HAVE_PTHREAD_ATTR_SETSTACKSIZE) 578 rc = pthread_attr_getstacksize(&thr_attr, &ostacksize); 579 if (0 != rc) { 580 msyslog(LOG_ERR, 581 "start_blocking_thread: pthread_attr_getstacksize() -> %s", 582 strerror(rc)); 583 } else { 584 if (ostacksize < THREAD_MINSTACKSIZE) 585 nstacksize = THREAD_MINSTACKSIZE; 586 else if (ostacksize > THREAD_MAXSTACKSIZE) 587 nstacksize = THREAD_MAXSTACKSIZE; 588 else 589 nstacksize = ostacksize; 590 if (nstacksize != ostacksize) 591 rc = pthread_attr_setstacksize(&thr_attr, nstacksize); 592 if (0 != rc) 593 msyslog(LOG_ERR, 594 "start_blocking_thread: pthread_attr_setstacksize(0x%lx -> 0x%lx) -> %s", 595 (u_long)ostacksize, (u_long)nstacksize, 596 strerror(rc)); 597 } 598 #else 599 UNUSED_ARG(nstacksize); 600 UNUSED_ARG(ostacksize); 601 #endif 602 #if defined(PTHREAD_SCOPE_SYSTEM) && defined(NEED_PTHREAD_SCOPE_SYSTEM) 603 pthread_attr_setscope(&thr_attr, PTHREAD_SCOPE_SYSTEM); 604 #endif 605 c->thread_ref = emalloc_zero(sizeof(*c->thread_ref)); 606 block_thread_signals(&saved_sig_mask); 607 rc = pthread_create(&c->thr_table[0], &thr_attr, 608 &blocking_thread, c); 609 pthread_sigmask(SIG_SETMASK, &saved_sig_mask, NULL); 610 pthread_attr_destroy(&thr_attr); 611 if (0 != rc) { 612 msyslog(LOG_ERR, "start_blocking_thread: pthread_create() -> %s", 613 strerror(rc)); 614 exit(1); 615 } 616 c->thread_ref = &c->thr_table[0]; 617 } 618 #endif 619 620 /* -------------------------------------------------------------------- 621 * block_thread_signals() 622 * 623 * Temporarily block signals used by ntpd main thread, so that signal 624 * mask inherited by child threads leaves them blocked. Returns prior 625 * active signal mask via pmask, to be restored by the main thread 626 * after pthread_create(). 627 */ 628 #ifndef SYS_WINNT 629 void 630 block_thread_signals( 631 sigset_t * pmask 632 ) 633 { 634 sigset_t block; 635 636 sigemptyset(&block); 637 # ifdef HAVE_SIGNALED_IO 638 # ifdef SIGIO 639 sigaddset(&block, SIGIO); 640 # endif 641 # ifdef SIGPOLL 642 sigaddset(&block, SIGPOLL); 643 # endif 644 # endif /* HAVE_SIGNALED_IO */ 645 sigaddset(&block, SIGALRM); 646 sigaddset(&block, MOREDEBUGSIG); 647 sigaddset(&block, LESSDEBUGSIG); 648 # ifdef SIGDIE1 649 sigaddset(&block, SIGDIE1); 650 # endif 651 # ifdef SIGDIE2 652 sigaddset(&block, SIGDIE2); 653 # endif 654 # ifdef SIGDIE3 655 sigaddset(&block, SIGDIE3); 656 # endif 657 # ifdef SIGDIE4 658 sigaddset(&block, SIGDIE4); 659 # endif 660 # ifdef SIGBUS 661 sigaddset(&block, SIGBUS); 662 # endif 663 sigemptyset(pmask); 664 pthread_sigmask(SIG_BLOCK, &block, pmask); 665 } 666 #endif /* !SYS_WINNT */ 667 668 669 /* -------------------------------------------------------------------- 670 * Create & destroy semaphores. This is sufficiently different between 671 * POSIX and Windows to warrant wrapper functions and close enough to 672 * use the concept of synchronization via semaphore for all platforms. 673 */ 674 static sem_ref 675 create_sema( 676 sema_type* semptr, 677 u_int inival, 678 u_int maxval) 679 { 680 #ifdef SYS_WINNT 681 682 long svini, svmax; 683 if (NULL != semptr) { 684 svini = (inival < LONG_MAX) 685 ? (long)inival : LONG_MAX; 686 svmax = (maxval < LONG_MAX && maxval > 0) 687 ? (long)maxval : LONG_MAX; 688 semptr->shnd = CreateSemaphore(NULL, svini, svmax, NULL); 689 if (NULL == semptr->shnd) 690 semptr = NULL; 691 } 692 693 #else 694 695 (void)maxval; 696 if (semptr && sem_init(semptr, FALSE, inival)) 697 semptr = NULL; 698 699 #endif 700 701 return semptr; 702 } 703 704 /* ------------------------------------------------------------------ */ 705 static sem_ref 706 delete_sema( 707 sem_ref obj) 708 { 709 710 # ifdef SYS_WINNT 711 712 if (obj) { 713 if (obj->shnd) 714 CloseHandle(obj->shnd); 715 obj->shnd = NULL; 716 } 717 718 # else 719 720 if (obj) 721 sem_destroy(obj); 722 723 # endif 724 725 return NULL; 726 } 727 728 /* -------------------------------------------------------------------- 729 * prepare_child_sems() 730 * 731 * create sync & access semaphores 732 * 733 * All semaphores are cleared, only the access semaphore has 1 unit. 734 * Childs wait on 'workitems_pending', then grabs 'sema_access' 735 * and dequeues jobs. When done, 'sema_access' is given one unit back. 736 * 737 * The producer grabs 'sema_access', manages the queue, restores 738 * 'sema_access' and puts one unit into 'workitems_pending'. 739 * 740 * The story goes the same for the response queue. 741 */ 742 static void 743 prepare_child_sems( 744 blocking_child *c 745 ) 746 { 747 if (NULL == worker_memlock) 748 worker_memlock = create_sema(&worker_mmutex, 1, 1); 749 750 c->accesslock = create_sema(&c->sem_table[0], 1, 1); 751 c->workitems_pending = create_sema(&c->sem_table[1], 0, 0); 752 c->wake_scheduled_sleep = create_sema(&c->sem_table[2], 0, 1); 753 # ifndef WORK_PIPE 754 c->responses_pending = create_sema(&c->sem_table[3], 0, 0); 755 # endif 756 } 757 758 /* -------------------------------------------------------------------- 759 * wait for semaphore. Where the wait can be interrupted, it will 760 * internally resume -- When this function returns, there is either no 761 * semaphore at all, a timeout occurred, or the caller could 762 * successfully take a token from the semaphore. 763 * 764 * For untimed wait, not checking the result of this function at all is 765 * definitely an option. 766 */ 767 static int 768 wait_for_sem( 769 sem_ref sem, 770 struct timespec * timeout /* wall-clock */ 771 ) 772 #ifdef SYS_WINNT 773 { 774 struct timespec now; 775 struct timespec delta; 776 DWORD msec; 777 DWORD rc; 778 779 if (!(sem && sem->shnd)) { 780 errno = EINVAL; 781 return -1; 782 } 783 784 if (NULL == timeout) { 785 msec = INFINITE; 786 } else { 787 getclock(TIMEOFDAY, &now); 788 delta = sub_tspec(*timeout, now); 789 if (delta.tv_sec < 0) { 790 msec = 0; 791 } else if ((delta.tv_sec + 1) >= (MAXDWORD / 1000)) { 792 msec = INFINITE; 793 } else { 794 msec = 1000 * (DWORD)delta.tv_sec; 795 msec += delta.tv_nsec / (1000 * 1000); 796 } 797 } 798 rc = WaitForSingleObject(sem->shnd, msec); 799 if (WAIT_OBJECT_0 == rc) 800 return 0; 801 if (WAIT_TIMEOUT == rc) { 802 errno = ETIMEDOUT; 803 return -1; 804 } 805 msyslog(LOG_ERR, "WaitForSingleObject unexpected 0x%x", rc); 806 errno = EFAULT; 807 return -1; 808 } 809 #else /* pthreads wait_for_sem() follows */ 810 { 811 int rc = -1; 812 813 if (sem) do { 814 if (NULL == timeout) 815 rc = sem_wait(sem); 816 else 817 rc = sem_timedwait(sem, timeout); 818 } while (rc == -1 && errno == EINTR); 819 else 820 errno = EINVAL; 821 822 return rc; 823 } 824 #endif 825 826 /* -------------------------------------------------------------------- 827 * blocking_thread - thread functions have WINAPI (aka 'stdcall') 828 * calling conventions under Windows and POSIX-defined signature 829 * otherwise. 830 */ 831 #ifdef SYS_WINNT 832 u_int WINAPI 833 #else 834 void * 835 #endif 836 blocking_thread( 837 void * ThreadArg 838 ) 839 { 840 blocking_child *c; 841 842 c = ThreadArg; 843 exit_worker(blocking_child_common(c)); 844 845 /* NOTREACHED */ 846 return 0; 847 } 848 849 /* -------------------------------------------------------------------- 850 * req_child_exit() runs in the parent. 851 * 852 * This function is called from from the idle timer, too, and possibly 853 * without a thread being there any longer. Since we have folded up our 854 * tent in that case and all the semaphores are already gone, we simply 855 * ignore this request in this case. 856 * 857 * Since the existence of the semaphores is controlled exclusively by 858 * the parent, there's no risk of data race here. 859 */ 860 int 861 req_child_exit( 862 blocking_child *c 863 ) 864 { 865 return (c->accesslock) 866 ? queue_req_pointer(c, CHILD_EXIT_REQ) 867 : 0; 868 } 869 870 /* -------------------------------------------------------------------- 871 * cleanup_after_child() runs in parent. 872 */ 873 static void 874 cleanup_after_child( 875 blocking_child * c 876 ) 877 { 878 DEBUG_INSIST(!c->reusable); 879 880 # ifdef SYS_WINNT 881 /* The thread was not created in detached state, so we better 882 * clean up. 883 */ 884 if (c->thread_ref && c->thread_ref->thnd) { 885 WaitForSingleObject(c->thread_ref->thnd, INFINITE); 886 INSIST(CloseHandle(c->thread_ref->thnd)); 887 c->thread_ref->thnd = NULL; 888 } 889 # endif 890 c->thread_ref = NULL; 891 892 /* remove semaphores and (if signalling vi IO) pipes */ 893 894 c->accesslock = delete_sema(c->accesslock); 895 c->workitems_pending = delete_sema(c->workitems_pending); 896 c->wake_scheduled_sleep = delete_sema(c->wake_scheduled_sleep); 897 898 # ifdef WORK_PIPE 899 DEBUG_INSIST(-1 != c->resp_read_pipe); 900 DEBUG_INSIST(-1 != c->resp_write_pipe); 901 (*addremove_io_fd)(c->resp_read_pipe, c->ispipe, TRUE); 902 close(c->resp_write_pipe); 903 close(c->resp_read_pipe); 904 c->resp_write_pipe = -1; 905 c->resp_read_pipe = -1; 906 # else 907 DEBUG_INSIST(NULL != c->responses_pending); 908 (*addremove_io_semaphore)(c->responses_pending->shnd, TRUE); 909 c->responses_pending = delete_sema(c->responses_pending); 910 # endif 911 912 /* Is it necessary to check if there are pending requests and 913 * responses? If so, and if there are, what to do with them? 914 */ 915 916 /* re-init buffer index sequencers */ 917 c->head_workitem = 0; 918 c->tail_workitem = 0; 919 c->head_response = 0; 920 c->tail_response = 0; 921 922 c->reusable = TRUE; 923 } 924 925 926 #else /* !WORK_THREAD follows */ 927 char work_thread_nonempty_compilation_unit; 928 #endif 929