+++ /dev/null
-/* threaded_queue.cc
-
- Copyright 2001, 2002 Red Hat Inc.
-
- Written by Robert Collins <rbtcollins@hotmail.com>
-
-This file is part of Cygwin.
-
-This software is a copyrighted work licensed under the terms of the
-Cygwin license. Please consult the file "CYGWIN_LICENSE" for
-details. */
-
-#include "woutsup.h"
-
-#include <assert.h>
-#include <errno.h>
-#include <stdio.h>
-#include <unistd.h>
-#include <sys/types.h>
-#include <stdlib.h>
-#include "threaded_queue.h"
-
-/*****************************************************************************/
-
-/* queue_request */
-
-queue_request::~queue_request ()
-{}
-
-/*****************************************************************************/
-
-/* threaded_queue */
-
-threaded_queue::threaded_queue (const size_t initial_workers)
- : _workers_count (0),
- _running (false),
- _submitters_head (NULL),
- _requests_count (0),
- _requests_head (NULL),
- _requests_sem (NULL)
-{
- InitializeCriticalSection (&_queue_lock);
-
- // This semaphore's count is the number of requests on the queue.
- // The maximum count (129792) is calculated as MAXIMUM_WAIT_OBJECTS
- // multiplied by max. threads per process (2028?), which is (a few)
- // more requests than could ever be pending with the current design.
-
- _requests_sem = CreateSemaphore (NULL, // SECURITY_ATTRIBUTES
- 0, // Initial count
- 129792, // Maximum count
- NULL); // Anonymous
-
- if (!_requests_sem)
- {
- system_printf (("failed to create the request queue semaphore, "
- "error = %lu"),
- GetLastError ());
- abort ();
- }
-
- create_workers (initial_workers);
-}
-
-threaded_queue::~threaded_queue ()
-{
- if (_running)
- stop ();
-
- debug_printf ("deleting all pending queue requests");
- queue_request *reqptr = _requests_head;
- while (reqptr)
- {
- queue_request *const ptr = reqptr;
- reqptr = reqptr->_next;
- safe_delete (ptr);
- }
-
- DeleteCriticalSection (&_queue_lock);
- if (_requests_sem)
- (void) CloseHandle (_requests_sem);
-}
-
-/* FIXME: return success or failure rather than quitting */
-void
-threaded_queue::add_submission_loop (queue_submission_loop *const submitter)
-{
- assert (this);
- assert (submitter);
- assert (submitter->_queue == this);
- assert (!submitter->_next);
-
- submitter->_next =
- TInterlockedExchangePointer (&_submitters_head, submitter);
-
- if (_running)
- submitter->start ();
-}
-
-bool
-threaded_queue::start ()
-{
- EnterCriticalSection (&_queue_lock);
- const bool was_running = _running;
- _running = true;
- queue_submission_loop *loopptr = _submitters_head;
- LeaveCriticalSection (&_queue_lock);
-
- if (!was_running)
- {
- debug_printf ("starting all queue submission loops");
-
- while (loopptr)
- {
- queue_submission_loop *const ptr = loopptr;
- loopptr = loopptr->_next;
- ptr->start ();
- }
- }
-
- return was_running;
-}
-
-bool
-threaded_queue::stop ()
-{
- EnterCriticalSection (&_queue_lock);
- const bool was_running = _running;
- _running = false;
- queue_submission_loop *loopptr = _submitters_head;
- LeaveCriticalSection (&_queue_lock);
-
- if (was_running)
- {
- debug_printf ("stopping all queue submission loops");
- while (loopptr)
- {
- queue_submission_loop *const ptr = loopptr;
- loopptr = loopptr->_next;
- ptr->stop ();
- }
-
- ReleaseSemaphore (_requests_sem, _workers_count, NULL);
- while (_workers_count)
- {
- debug_printf (("waiting for worker threads to terminate: "
- "%lu still running"),
- _workers_count);
- Sleep (1000);
- }
- debug_printf ("all worker threads have terminated");
- }
-
- return was_running;
-}
-
-/* FIXME: return success or failure */
-void
-threaded_queue::add (queue_request *const therequest)
-{
- assert (this);
- assert (therequest);
- assert (!therequest->_next);
-
- if (!_workers_count)
- {
- system_printf ("warning: no worker threads to handle request!");
- // FIXME: And then what?
- }
-
- EnterCriticalSection (&_queue_lock);
- if (!_requests_head)
- _requests_head = therequest;
- else
- {
- /* Add to the queue end. */
- queue_request *reqptr = _requests_head;
- for (; reqptr->_next; reqptr = reqptr->_next)
- {}
- assert (reqptr);
- assert (!reqptr->_next);
- reqptr->_next = therequest;
- }
-
- _requests_count += 1;
- assert (_requests_count > 0);
- LeaveCriticalSection (&_queue_lock);
-
- (void) ReleaseSemaphore (_requests_sem, 1, NULL);
-}
-
-/*static*/ DWORD WINAPI
-threaded_queue::start_routine (const LPVOID lpParam)
-{
- class threaded_queue *const queue = (class threaded_queue *) lpParam;
- assert (queue);
-
- queue->worker_loop ();
-
- const long count = InterlockedDecrement (&queue->_workers_count);
- assert (count >= 0);
-
- if (queue->_running)
- debug_printf ("worker loop has exited; thread about to terminate");
-
- return 0;
-}
-
-/* Called from the constructor: so no need to be thread-safe until the
- * worker threads start to be created; thus the interlocked increment
- * of the `_workers_count' field.
- */
-
-void
-threaded_queue::create_workers (const size_t initial_workers)
-{
- assert (initial_workers > 0);
-
- for (unsigned int i = 0; i != initial_workers; i++)
- {
- const long count = InterlockedIncrement (&_workers_count);
- assert (count > 0);
-
- DWORD tid;
- const HANDLE hThread =
- CreateThread (NULL, 0, start_routine, this, 0, &tid);
-
- if (!hThread)
- {
- system_printf ("failed to create thread, error = %lu",
- GetLastError ());
- abort ();
- }
-
- (void) CloseHandle (hThread);
- }
-}
-
-void
-threaded_queue::worker_loop ()
-{
- while (true)
- {
- const DWORD rc = WaitForSingleObject (_requests_sem, INFINITE);
- if (rc == WAIT_FAILED)
- {
- system_printf ("wait for request semaphore failed, error = %lu",
- GetLastError ());
- return;
- }
- assert (rc == WAIT_OBJECT_0);
-
- EnterCriticalSection (&_queue_lock);
- if (!_running)
- {
- LeaveCriticalSection (&_queue_lock);
- return;
- }
-
- assert (_requests_head);
- queue_request *const reqptr = _requests_head;
- _requests_head = reqptr->_next;
-
- _requests_count -= 1;
- assert (_requests_count >= 0);
- LeaveCriticalSection (&_queue_lock);
-
- assert (reqptr);
- reqptr->process ();
- safe_delete (reqptr);
- }
-}
-
-/*****************************************************************************/
-
-/* queue_submission_loop */
-
-queue_submission_loop::queue_submission_loop (threaded_queue *const queue,
- const bool ninterruptible)
- : _running (false),
- _interrupt_event (NULL),
- _queue (queue),
- _interruptible (ninterruptible),
- _hThread (NULL),
- _tid (0),
- _next (NULL)
-{
- if (_interruptible)
- {
- // verbose: debug_printf ("creating an interruptible processing thread");
-
- _interrupt_event = CreateEvent (NULL, // SECURITY_ATTRIBUTES
- FALSE, // Auto-reset
- FALSE, // Initially non-signalled
- NULL); // Anonymous
-
- if (!_interrupt_event)
- {
- system_printf ("failed to create interrupt event, error = %lu",
- GetLastError ());
- abort ();
- }
- }
-}
-
-queue_submission_loop::~queue_submission_loop ()
-{
- if (_running)
- stop ();
- if (_interrupt_event)
- (void) CloseHandle (_interrupt_event);
- if (_hThread)
- (void) CloseHandle (_hThread);
-}
-
-bool
-queue_submission_loop::start ()
-{
- assert (this);
- assert (!_hThread);
-
- const bool was_running = _running;
-
- if (!was_running)
- {
- _running = true;
-
- _hThread = CreateThread (NULL, 0, start_routine, this, 0, &_tid);
- if (!_hThread)
- {
- system_printf ("failed to create thread, error = %lu",
- GetLastError ());
- abort ();
- }
- }
-
- return was_running;
-}
-
-bool
-queue_submission_loop::stop ()
-{
- assert (this);
- assert (_hThread && _hThread != INVALID_HANDLE_VALUE);
-
- const bool was_running = _running;
-
- if (_running)
- {
- _running = false;
-
- if (_interruptible)
- {
- assert (_interrupt_event
- && _interrupt_event != INVALID_HANDLE_VALUE);
-
- SetEvent (_interrupt_event);
-
- if (WaitForSingleObject (_hThread, 1000) == WAIT_TIMEOUT)
- {
- system_printf (("request loop thread %lu failed to shutdown "
- "when asked politely: about to get heavy"),
- _tid);
-
- if (!TerminateThread (_hThread, 0))
- {
- system_printf (("failed to kill request loop thread %lu"
- ", error = %lu"),
- _tid, GetLastError ());
- abort ();
- }
- }
- }
- else
- {
- // FIXME: could wait to see if the request loop notices that
- // the submission loop is no longer running and shuts down
- // voluntarily.
-
- debug_printf ("killing request loop thread %lu", _tid);
-
- if (!TerminateThread (_hThread, 0))
- system_printf (("failed to kill request loop thread %lu"
- ", error = %lu"),
- _tid, GetLastError ());
- }
- }
-
- return was_running;
-}
-
-/*static*/ DWORD WINAPI
-queue_submission_loop::start_routine (const LPVOID lpParam)
-{
- class queue_submission_loop *const submission_loop =
- (class queue_submission_loop *) lpParam;
- assert (submission_loop);
-
- submission_loop->request_loop ();
-
- debug_printf ("submission loop has exited; thread about to terminate");
-
- submission_loop->stop ();
-
- return 0;
-}
-
-/*****************************************************************************/
OSDN Git Service
