1 """An ISAPI extension base class implemented using a thread-pool.""" 2 3 # $Id$ 4 5 import sys 6 import threading 7 import time 8 import traceback 9 10 from pywintypes import OVERLAPPED 11 from win32event import INFINITE 12 from win32file import ( 13 CloseHandle, 14 CreateIoCompletionPort, 15 GetQueuedCompletionStatus, 16 PostQueuedCompletionStatus, 17 ) 18 from win32security import SetThreadToken 19 20 import isapi.simple 21 from isapi import ExtensionError, isapicon 22 23 ISAPI_REQUEST = 1 24 ISAPI_SHUTDOWN = 2 25 26 27 class WorkerThread(threading.Thread): 28 def __init__(self, extension, io_req_port): 29 self.running = False 30 self.io_req_port = io_req_port 31 self.extension = extension 32 threading.Thread.__init__(self) 33 # We wait 15 seconds for a thread to terminate, but if it fails to, 34 # we don't want the process to hang at exit waiting for it... 35 self.setDaemon(True) 36 37 def run(self): 38 self.running = True 39 while self.running: 40 errCode, bytes, key, overlapped = GetQueuedCompletionStatus( 41 self.io_req_port, INFINITE 42 ) 43 if key == ISAPI_SHUTDOWN and overlapped is None: 44 break 45 46 # Let the parent extension handle the command. 47 dispatcher = self.extension.dispatch_map.get(key) 48 if dispatcher is None: 49 raise RuntimeError(f"Bad request '{key}'") 50 51 dispatcher(errCode, bytes, key, overlapped) 52 53 def call_handler(self, cblock): 54 self.extension.Dispatch(cblock) 55 56 57 # A generic thread-pool based extension, using IO Completion Ports. 58 # Sub-classes can override one method to implement a simple extension, or 59 # may leverage the CompletionPort to queue their own requests, and implement a 60 # fully asynch extension. 61 class ThreadPoolExtension(isapi.simple.SimpleExtension): 62 "Base class for an ISAPI extension based around a thread-pool" 63 64 max_workers = 20 65 worker_shutdown_wait = 15000 # 15 seconds for workers to quit... 66 67 def __init__(self): 68 self.workers = [] 69 # extensible dispatch map, for sub-classes that need to post their 70 # own requests to the completion port. 71 # Each of these functions is called with the result of 72 # GetQueuedCompletionStatus for our port. 73 self.dispatch_map = { 74 ISAPI_REQUEST: self.DispatchConnection, 75 } 76 77 def GetExtensionVersion(self, vi): 78 isapi.simple.SimpleExtension.GetExtensionVersion(self, vi) 79 # As per Q192800, the CompletionPort should be created with the number 80 # of processors, even if the number of worker threads is much larger. 81 # Passing 0 means the system picks the number. 82 self.io_req_port = CreateIoCompletionPort(-1, None, 0, 0) 83 # start up the workers 84 self.workers = [] 85 for i in range(self.max_workers): 86 worker = WorkerThread(self, self.io_req_port) 87 worker.start() 88 self.workers.append(worker) 89 90 def HttpExtensionProc(self, control_block): 91 overlapped = OVERLAPPED() 92 overlapped.object = control_block 93 PostQueuedCompletionStatus(self.io_req_port, 0, ISAPI_REQUEST, overlapped) 94 return isapicon.HSE_STATUS_PENDING 95 96 def TerminateExtension(self, status): 97 for worker in self.workers: 98 worker.running = False 99 for worker in self.workers: 100 PostQueuedCompletionStatus(self.io_req_port, 0, ISAPI_SHUTDOWN, None) 101 # wait for them to terminate - pity we aren't using 'native' threads 102 # as then we could do a smart wait - but now we need to poll.... 103 end_time = time.time() + self.worker_shutdown_wait / 1000 104 alive = self.workers 105 while alive: 106 if time.time() > end_time: 107 # xxx - might be nice to log something here. 108 break 109 time.sleep(0.2) 110 alive = [w for w in alive if w.is_alive()] 111 self.dispatch_map = {} # break circles 112 CloseHandle(self.io_req_port) 113 114 # This is the one operation the base class supports - a simple 115 # Connection request. We setup the thread-token, and dispatch to the 116 # sub-class's 'Dispatch' method. 117 def DispatchConnection(self, errCode, bytes, key, overlapped): 118 control_block = overlapped.object 119 # setup the correct user for this request 120 hRequestToken = control_block.GetImpersonationToken() 121 SetThreadToken(None, hRequestToken) 122 try: 123 try: 124 self.Dispatch(control_block) 125 except: 126 self.HandleDispatchError(control_block) 127 finally: 128 # reset the security context 129 SetThreadToken(None, None) 130 131 def Dispatch(self, ecb): 132 """Overridden by the sub-class to handle connection requests. 133 134 This class creates a thread-pool using a Windows completion port, 135 and dispatches requests via this port. Sub-classes can generally 136 implement each connection request using blocking reads and writes, and 137 the thread-pool will still provide decent response to the end user. 138 139 The sub-class can set a max_workers attribute (default is 20). Note 140 that this generally does *not* mean 20 threads will all be concurrently 141 running, via the magic of Windows completion ports. 142 143 There is no default implementation - sub-classes must implement this. 144 """ 145 raise NotImplementedError("sub-classes should override Dispatch") 146 147 def HandleDispatchError(self, ecb): 148 """Handles errors in the Dispatch method. 149 150 When a Dispatch method call fails, this method is called to handle 151 the exception. The default implementation formats the traceback 152 in the browser. 153 """ 154 ecb.HttpStatusCode = isapicon.HSE_STATUS_ERROR 155 # control_block.LogData = "we failed!" 156 exc_typ, exc_val, exc_tb = sys.exc_info() 157 limit = None 158 try: 159 try: 160 import html 161 162 ecb.SendResponseHeaders( 163 "200 OK", "Content-type: text/html\r\n\r\n", False 164 ) 165 print(file=ecb) 166 print("<H3>Traceback (most recent call last):</H3>", file=ecb) 167 list = traceback.format_tb( 168 exc_tb, limit 169 ) + traceback.format_exception_only(exc_typ, exc_val) 170 bold = list.pop() 171 print( 172 "<PRE>{}<B>{}</B></PRE>".format( 173 html.escape("".join(list)), 174 html.escape(bold), 175 ), 176 file=ecb, 177 ) 178 except ExtensionError: 179 # The client disconnected without reading the error body - 180 # it's probably not a real browser at the other end, ignore it. 181 pass 182 except: 183 print("FAILED to render the error message!") 184 traceback.print_exc() 185 print("ORIGINAL extension error:") 186 traceback.print_exception(exc_typ, exc_val, exc_tb) 187 finally: 188 # holding tracebacks in a local of a frame that may itself be 189 # part of a traceback used to be evil and cause leaks! 190 exc_tb = None 191 ecb.DoneWithSession() 192