codekingpro/portable-devtools

1"""An ISAPI extension base class implemented using a thread-pool."""2# $Id$3 4import sys5import threading6import time7import traceback8 9from pywintypes import OVERLAPPED10from win32event import INFINITE11from win32file import (12    CloseHandle,13    CreateIoCompletionPort,14    GetQueuedCompletionStatus,15    PostQueuedCompletionStatus,16)17from win32security import SetThreadToken18 19import isapi.simple20from isapi import ExtensionError, isapicon21 22ISAPI_REQUEST = 123ISAPI_SHUTDOWN = 224 25 26class WorkerThread(threading.Thread):27    def __init__(self, extension, io_req_port):28        self.running = False29        self.io_req_port = io_req_port30        self.extension = extension31        threading.Thread.__init__(self)32        # We wait 15 seconds for a thread to terminate, but if it fails to,33        # we don't want the process to hang at exit waiting for it...34        self.setDaemon(True)35 36    def run(self):37        self.running = True38        while self.running:39            errCode, bytes, key, overlapped = GetQueuedCompletionStatus(40                self.io_req_port, INFINITE41            )42            if key == ISAPI_SHUTDOWN and overlapped is None:43                break44 45            # Let the parent extension handle the command.46            dispatcher = self.extension.dispatch_map.get(key)47            if dispatcher is None:48                raise RuntimeError("Bad request '%s'" % (key,))49 50            dispatcher(errCode, bytes, key, overlapped)51 52    def call_handler(self, cblock):53        self.extension.Dispatch(cblock)54 55 56# A generic thread-pool based extension, using IO Completion Ports.57# Sub-classes can override one method to implement a simple extension, or58# may leverage the CompletionPort to queue their own requests, and implement a59# fully asynch extension.60class ThreadPoolExtension(isapi.simple.SimpleExtension):61    "Base class for an ISAPI extension based around a thread-pool"62    max_workers = 2063    worker_shutdown_wait = 15000  # 15 seconds for workers to quit...64 65    def __init__(self):66        self.workers = []67        # extensible dispatch map, for sub-classes that need to post their68        # own requests to the completion port.69        # Each of these functions is called with the result of70        # GetQueuedCompletionStatus for our port.71        self.dispatch_map = {72            ISAPI_REQUEST: self.DispatchConnection,73        }74 75    def GetExtensionVersion(self, vi):76        isapi.simple.SimpleExtension.GetExtensionVersion(self, vi)77        # As per Q192800, the CompletionPort should be created with the number78        # of processors, even if the number of worker threads is much larger.79        # Passing 0 means the system picks the number.80        self.io_req_port = CreateIoCompletionPort(-1, None, 0, 0)81        # start up the workers82        self.workers = []83        for i in range(self.max_workers):84            worker = WorkerThread(self, self.io_req_port)85            worker.start()86            self.workers.append(worker)87 88    def HttpExtensionProc(self, control_block):89        overlapped = OVERLAPPED()90        overlapped.object = control_block91        PostQueuedCompletionStatus(self.io_req_port, 0, ISAPI_REQUEST, overlapped)92        return isapicon.HSE_STATUS_PENDING93 94    def TerminateExtension(self, status):95        for worker in self.workers:96            worker.running = False97        for worker in self.workers:98            PostQueuedCompletionStatus(self.io_req_port, 0, ISAPI_SHUTDOWN, None)99        # wait for them to terminate - pity we aren't using 'native' threads100        # as then we could do a smart wait - but now we need to poll....101        end_time = time.time() + self.worker_shutdown_wait / 1000102        alive = self.workers103        while alive:104            if time.time() > end_time:105                # xxx - might be nice to log something here.106                break107            time.sleep(0.2)108            alive = [w for w in alive if w.is_alive()]109        self.dispatch_map = {}  # break circles110        CloseHandle(self.io_req_port)111 112    # This is the one operation the base class supports - a simple113    # Connection request.  We setup the thread-token, and dispatch to the114    # sub-class's 'Dispatch' method.115    def DispatchConnection(self, errCode, bytes, key, overlapped):116        control_block = overlapped.object117        # setup the correct user for this request118        hRequestToken = control_block.GetImpersonationToken()119        SetThreadToken(None, hRequestToken)120        try:121            try:122                self.Dispatch(control_block)123            except:124                self.HandleDispatchError(control_block)125        finally:126            # reset the security context127            SetThreadToken(None, None)128 129    def Dispatch(self, ecb):130        """Overridden by the sub-class to handle connection requests.131 132        This class creates a thread-pool using a Windows completion port,133        and dispatches requests via this port.  Sub-classes can generally134        implement each connection request using blocking reads and writes, and135        the thread-pool will still provide decent response to the end user.136 137        The sub-class can set a max_workers attribute (default is 20).  Note138        that this generally does *not* mean 20 threads will all be concurrently139        running, via the magic of Windows completion ports.140 141        There is no default implementation - sub-classes must implement this.142        """143        raise NotImplementedError("sub-classes should override Dispatch")144 145    def HandleDispatchError(self, ecb):146        """Handles errors in the Dispatch method.147 148        When a Dispatch method call fails, this method is called to handle149        the exception.  The default implementation formats the traceback150        in the browser.151        """152        ecb.HttpStatusCode = isapicon.HSE_STATUS_ERROR153        # control_block.LogData = "we failed!"154        exc_typ, exc_val, exc_tb = sys.exc_info()155        limit = None156        try:157            try:158                import cgi159 160                ecb.SendResponseHeaders(161                    "200 OK", "Content-type: text/html\r\n\r\n", False162                )163                print(file=ecb)164                print("<H3>Traceback (most recent call last):</H3>", file=ecb)165                list = traceback.format_tb(166                    exc_tb, limit167                ) + traceback.format_exception_only(exc_typ, exc_val)168                print(169                    "<PRE>%s<B>%s</B></PRE>"170                    % (171                        cgi.escape("".join(list[:-1])),172                        cgi.escape(list[-1]),173                    ),174                    file=ecb,175                )176            except ExtensionError:177                # The client disconnected without reading the error body -178                # its probably not a real browser at the other end, ignore it.179                pass180            except:181                print("FAILED to render the error message!")182                traceback.print_exc()183                print("ORIGINAL extension error:")184                traceback.print_exception(exc_typ, exc_val, exc_tb)185        finally:186            # holding tracebacks in a local of a frame that may itself be187            # part of a traceback used to be evil and cause leaks!188            exc_tb = None189            ecb.DoneWithSession()190