| #!/usr/bin/env python |
| # |
| # Copyright 2007 Google Inc. |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| # |
| |
| |
| """This module adds support for multiple processes in the dev_appserver. |
| |
| Each instance of the application is started as a separate process on a unique |
| port, for state isolation and parallel execution. A load-balancing process is |
| also created for each Backend. An API Server process is launched to handle all |
| memcache and datastore API requests, so that persistent state is shared across |
| all processes. Application and Backend instances forward their memcache and |
| datastore requests to the API Server using the remote_api interface. |
| |
| The base process is considered the Master. It manages all subprocesses, assigns |
| them ports, and issues /_ah/start requests to Backend instances. Ports are |
| either fixed (using the base value of --multiprocess_min_port) or randomly |
| chosen. The Master listens on the --port specified by the user, and forwards |
| all requests to an App Instance process. |
| |
| Each balancer forwards incoming requests to the next free instance, |
| or return with a HTTP 503 error if no free instance is available. |
| """ |
| |
| |
| import BaseHTTPServer |
| import copy |
| import cStringIO |
| import errno |
| import httplib |
| import logging |
| import os |
| import Queue |
| import signal |
| import socket |
| import subprocess |
| import sys |
| import threading |
| import time |
| import weakref |
| |
| from google.appengine.api import backendinfo |
| from google.appengine.api.backends import backends as backends_api |
| from google.appengine.ext.remote_api import remote_api_stub |
| from google.appengine.tools import api_server |
| |
| ARG_ADDRESS = 'address' |
| ARG_PORT = 'port' |
| |
| ARG_BACKENDS = 'backends' |
| ARG_MULTIPROCESS = 'multiprocess' |
| ARG_MULTIPROCESS_MIN_PORT = 'multiprocess_min_port' |
| ARG_MULTIPROCESS_API_PORT = 'multiprocess_api_port' |
| ARG_MULTIPROCESS_API_SERVER = 'multiprocess_api_server' |
| ARG_MULTIPROCESS_APP_INSTANCE_ID = 'multiprocess_app_instance' |
| ARG_MULTIPROCESS_BACKEND_ID = 'multiprocess_backend_id' |
| ARG_MULTIPROCESS_BACKEND_INSTANCE_ID = 'multiprocess_backend_instance_id' |
| ARG_MULTIPROCESS_FRONTEND_PORT = 'multiprocess_frontend_port' |
| |
| |
| |
| API_SERVER_HOST = 'localhost' |
| PATH_DEV_API_SERVER = '/_ah/dev_api_server' |
| |
| BACKEND_MAX_INSTANCES = 20 |
| |
| |
| class Error(Exception): pass |
| |
| |
| def SetThreadName(thread, name): |
| """Sets the name a of thread, including the GlobalProcess name.""" |
| thread.setName('[%s: %s]' % (GlobalProcess().Type(), name)) |
| |
| |
| class StartInstance(threading.Thread): |
| """Thread that periodically attempts to start a backend instance.""" |
| |
| def __init__(self, child): |
| threading.Thread.__init__(self) |
| self.child = child |
| self.setDaemon(True) |
| SetThreadName(self, 'Start %s' % child) |
| |
| def run(self): |
| while True: |
| self.child.SendStartRequest() |
| time.sleep(1) |
| |
| |
| class ChildProcess(object): |
| def __init__(self, |
| host, |
| port, |
| app_instance=None, |
| backend_id=None, |
| instance_id=None, |
| frontend_port=None): |
| """Creates an object representing a child process. |
| |
| Only one of the given args should be provided (except for instance_id when |
| backend_id is specified). |
| |
| Args: |
| app_instance: (int) The process represents the indicated app instance. |
| backend_id: (string) The process represents a backend. |
| instance_id: (int) The process represents the given backend instance. |
| frontend_port: (int) for backends, the frontend port. |
| """ |
| |
| self.app_instance = app_instance |
| self.backend_id = backend_id |
| self.instance_id = instance_id |
| self.process = None |
| self.argv = [] |
| self.started = False |
| self.connection_handler = httplib.HTTPConnection |
| self.SetHostPort(host, port) |
| self.frontend_port = frontend_port |
| |
| def __str__(self): |
| if self.app_instance is not None: |
| string = 'App Instance' |
| elif self.instance_id is not None: |
| string = 'Backend Instance: %s.%d' % (self.backend_id, self.instance_id) |
| elif self.backend_id: |
| string = 'Backend Balancer: %s' % self.backend_id |
| else: |
| string = 'Unknown' |
| return '%s [%s]' % (string, self.Address()) |
| |
| def SetHostPort(self, host, port): |
| """Sets the host and port that this process listens on.""" |
| self.host = host |
| self.port = port |
| if self.backend_id: |
| backends_api._set_dev_port(self.port, |
| self.backend_id, |
| self.instance_id) |
| |
| def Address(self): |
| """Returns the URL for this process.""" |
| return 'http://%s:%d' % self.HostPort() |
| |
| def HostPort(self): |
| """Returns the address of this process as a (host, port) pair.""" |
| return (self.host, self.port) |
| |
| def Start(self, argv, api_port): |
| """Starts the child process. |
| |
| Args: |
| argv: The argv of the parent process. When starting the subprocess, |
| we make a copy of the parent's argv, then modify it in accordance with |
| how the ChildProcess is configured, to represent different processes in |
| the multiprocess dev_appserver. |
| api_port: The port on which the API Server listens. |
| """ |
| self.argv = copy.deepcopy(argv) |
| self.api_port = api_port |
| |
| self.SetFlag('--multiprocess') |
| self.SetFlag('--address', short_flag='-a', value=self.host) |
| self.SetFlag('--port', short_flag='-p', value=self.port) |
| self.SetFlag('--multiprocess_api_port', value=self.api_port) |
| if self.frontend_port is not None: |
| self.SetFlag('--multiprocess_frontend_port', value=self.frontend_port) |
| |
| if self.app_instance is not None: |
| self.SetFlag('--multiprocess_app_instance_id', value=0) |
| if self.backend_id is not None: |
| self.SetFlag('--multiprocess_backend_id', value=self.backend_id) |
| if self.instance_id is not None: |
| self.SetFlag('--multiprocess_backend_instance_id', value=self.instance_id) |
| if self.argv[0].endswith('.py'): |
| |
| |
| |
| self.argv.insert(0, sys.executable) |
| |
| logging.debug('Starting %s with args: %s', self, self.argv) |
| self.process = subprocess.Popen(self.argv) |
| |
| def EnableStartRequests(self): |
| """Starts a thread to periodically send /_ah/start to this instance. |
| |
| We need a thread to do this because we want to restart any resident Backends |
| that have been shutdown, and because a backend instance is not considered |
| to be ready for serving until it has successfully responded to /_ah/start. |
| """ |
| if self.backend_id and self.instance_id is not None: |
| self.start_thread = StartInstance(self) |
| self.start_thread.start() |
| |
| def Connect(self): |
| """Attempts to connect to the child process. |
| |
| Returns: |
| bool: Whether a connection was made. |
| """ |
| logging.debug('Attempting connection to %s', self) |
| sock = None |
| result = True |
| try: |
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| sock.connect(self.HostPort()) |
| except: |
| result = False |
| if sock: |
| sock.close() |
| return result |
| |
| def WaitForConnection(self, timeout_s=30.0, poll_period_s=0.5): |
| """Blocks until the child process has started. |
| |
| This method repeatedly attempts to connect to the process on its HTTP server |
| port. Returns when a connection has been successfully established or the |
| timeout has been reached. |
| |
| Args: |
| timeout_s: Amount of time to wait, in seconds. |
| poll_period_s: Time to wait between connection attempts. |
| """ |
| finish_time = time.time() + timeout_s |
| while time.time() < finish_time: |
| if self.Connect(): |
| return True |
| time.sleep(poll_period_s) |
| logging.info('%s took more than %d seconds to start.', self, timeout_s) |
| return False |
| |
| def SendStartRequest(self): |
| """If the process has not been started, sends a request to /_ah/start.""" |
| if self.started: |
| return |
| |
| try: |
| response = self.SendRequest('GET', '/_ah/start') |
| rc = response.status |
| if (rc >= 200 and rc < 300) or rc == 404: |
| self.started = True |
| except KeyboardInterrupt: |
| pass |
| except Exception, e: |
| logging.error('Failed start request to %s: %s', self, e) |
| |
| def SendRequest(self, command, path, payload=None, headers=None): |
| """Sends an HTTP request to this process. |
| |
| Args: |
| command: The HTTP command (e.g., GET, POST) |
| path: The URL path for the request. |
| headers: A dictionary containing headers as key-value pairs. |
| """ |
| logging.debug('send request: %s %s to %s' % (command, path, self)) |
| connection = self.connection_handler('%s:%d' % self.HostPort()) |
| connection.request(command, path, payload, headers or {}) |
| response = connection.getresponse() |
| return response |
| |
| def SetFlag(self, flag, short_flag=None, value=None): |
| """Add a flag to self.argv, replacing the existing value if set. |
| |
| Args: |
| flag: flag to remove. |
| short_flag: one letter short version of the flag (optional) |
| value: Value of the flag (optional) |
| """ |
| self.RemoveFlag(flag, short_flag=short_flag, has_value=(value is not None)) |
| if value is None: |
| self.argv.append(flag) |
| else: |
| self.argv.append(flag + '=' + str(value)) |
| |
| def RemoveFlag(self, flag, short_flag=None, has_value=False): |
| """Removes an argument from self.argv. |
| |
| Args: |
| flag: flag to remove. |
| short_flag: one letter short version of the flag |
| has_value: True if the next argument after the short flag is the value. |
| """ |
| new_argv = [] |
| index = 0 |
| while index < len(self.argv): |
| value = self.argv[index] |
| index += 1 |
| |
| |
| if flag == value: |
| |
| if has_value: |
| index += 1 |
| continue |
| |
| |
| if has_value and value.startswith(flag + '='): |
| continue |
| |
| |
| if short_flag == value: |
| |
| if has_value: |
| index += 1 |
| continue |
| new_argv.append(value) |
| |
| |
| self.argv = new_argv |
| |
| |
| class DevProcess(object): |
| """Represents a process in the multiprocess dev_appserver.""" |
| |
| TYPE_MASTER = 'Master' |
| TYPE_APP_INSTANCE = 'App Instance' |
| TYPE_BACKEND_BALANCER = 'Backend Balancer' |
| TYPE_BACKEND_INSTANCE = 'Backend Instance' |
| TYPES = frozenset([TYPE_MASTER, |
| TYPE_APP_INSTANCE, |
| TYPE_BACKEND_BALANCER, |
| TYPE_BACKEND_INSTANCE]) |
| |
| def __init__(self): |
| """Creates a DevProcess with a default configuration.""" |
| self.process_type = None |
| |
| |
| self.desc = None |
| |
| |
| self.http_server = None |
| |
| |
| self.app_id = None |
| |
| |
| self.backends = None |
| |
| |
| |
| self.app_instance = None |
| |
| |
| self.backend_id = None |
| |
| |
| self.instance_id = None |
| |
| |
| self.backend_entry = None |
| |
| |
| self.host = None |
| |
| |
| self.port = None |
| |
| |
| self.api_port = None |
| |
| |
| self.multiprocess_min_port = 9000 |
| |
| |
| self.children = [] |
| |
| |
| self.child_app_instance = None |
| |
| |
| self.child_api_server = None |
| |
| |
| self.balance_set = None |
| |
| |
| |
| self.started = False |
| |
| def Init(self, appinfo, backends, options): |
| """Supplies a list of backends for future use. |
| |
| Args: |
| appinfo: An AppInfoExternal object. |
| backends: List of BackendEntry objects. |
| options: Dictionary of command-line options. |
| """ |
| self.backends = backends |
| self.options = options |
| |
| self.app_id = appinfo.application |
| self.host = options[ARG_ADDRESS] |
| self.port = options[ARG_PORT] |
| |
| |
| if ARG_MULTIPROCESS_APP_INSTANCE_ID in options: |
| self.SetType(DevProcess.TYPE_APP_INSTANCE) |
| self.app_instance = options[ARG_MULTIPROCESS_APP_INSTANCE_ID] |
| self.desc = str(self.app_instance) |
| if ARG_MULTIPROCESS_BACKEND_ID in options: |
| self.backend_id = options[ARG_MULTIPROCESS_BACKEND_ID] |
| self.desc = self.backend_id |
| if ARG_MULTIPROCESS_BACKEND_INSTANCE_ID in options: |
| self.SetType(DevProcess.TYPE_BACKEND_INSTANCE) |
| self.instance_id = int(options[ARG_MULTIPROCESS_BACKEND_INSTANCE_ID]) |
| self.desc += '.%d' % self.instance_id |
| else: |
| self.SetType(DevProcess.TYPE_BACKEND_BALANCER) |
| |
| if ARG_MULTIPROCESS_API_PORT in options: |
| self.api_port = int(options[ARG_MULTIPROCESS_API_PORT]) |
| if ARG_MULTIPROCESS_MIN_PORT in options: |
| self.multiprocess_min_port = int(options[ARG_MULTIPROCESS_MIN_PORT]) |
| |
| if self.IsBackend(): |
| self.InitBackendEntry() |
| |
| |
| if not self.Type(): |
| self.SetType(DevProcess.TYPE_MASTER) |
| |
| def InitBackendEntry(self): |
| """Finds the entry for the backend this process represents, if any.""" |
| for backend in self.backends: |
| if backend.name == self.backend_id: |
| self.backend_entry = backend |
| |
| if not self.backend_entry: |
| raise Error('No backend entry found for: ' % self) |
| |
| def HttpServer(self): |
| """Returns the HTTPServer used by this process.""" |
| return self.http_server |
| |
| def Address(self): |
| """Returns the address of this process.""" |
| return 'http://%s:%d' % (self.host, self.port) |
| |
| def SetHttpServer(self, http_server): |
| """Sets the http_server to be used when handling requests. |
| |
| Args: |
| http_server: An HTTPServer that receives requests. |
| """ |
| self.http_server = http_server |
| self.handle_requests = HandleRequestThread() |
| self.handle_requests.start() |
| |
| def StartChildren(self, argv, options): |
| """Starts the set of child processes.""" |
| self.children = [] |
| |
| |
| base_port = self.multiprocess_min_port |
| self.frontend_port = base_port |
| next_port = base_port |
| self.child_app_instance = ChildProcess(self.host, next_port, |
| app_instance=0) |
| self.children.append(self.child_app_instance) |
| next_port += 1 |
| |
| |
| for backend in self.backends: |
| base_port += 100 |
| next_port = base_port |
| |
| |
| for i in xrange(backend.instances): |
| self.children.append(ChildProcess(self.host, next_port, |
| backend_id=backend.name, |
| instance_id=i, |
| frontend_port=self.frontend_port)) |
| next_port += 1 |
| |
| |
| self.children.append(ChildProcess(self.host, base_port + 99, |
| backend_id=backend.name)) |
| |
| |
| |
| base_port += 100 |
| next_port = base_port |
| |
| |
| self.child_api_server = api_server.APIServerProcess( |
| executable=sys.executable, |
| script=os.path.join(os.path.dirname(argv[0]), 'api_server.py'), |
| host=self.host, |
| port=next_port, |
| app_id=self.app_id, |
| application_host=options['address'], |
| application_port=options['port'], |
| application_root=options['root_path'], |
| auto_id_policy=options['auto_id_policy'], |
| blobstore_path=options['blobstore_path'], |
| clear_datastore=options['clear_datastore'], |
| clear_prospective_search=options['clear_prospective_search'], |
| datastore_path=options['datastore_path'], |
| enable_sendmail=options['enable_sendmail'], |
| enable_task_running=not options['disable_task_running'], |
| high_replication=options['high_replication'], |
| logs_path=options['logs_path'], |
| prospective_search_path=options['prospective_search_path'], |
| require_indexes=options['require_indexes'], |
| show_mail_body=options['show_mail_body'], |
| smtp_host=options['smtp_host'], |
| smtp_password=options['smtp_password'], |
| smtp_port=options['smtp_port'], |
| smtp_user=options['smtp_user'], |
| task_retry_seconds=options['task_retry_seconds'], |
| trusted=options['trusted'], |
| use_sqlite=options['use_sqlite'], |
| ) |
| self.child_api_server.Start() |
| |
| |
| if self.multiprocess_min_port == 0: |
| self.AssignPortsRandomly() |
| |
| |
| self.api_port = next_port |
| |
| |
| |
| for child in self.children: |
| child.Start(argv, self.api_port) |
| |
| |
| self.child_api_server.WaitUntilServing() |
| |
| for child in self.children: |
| child.WaitForConnection() |
| |
| |
| |
| |
| message = '\n\nMultiprocess Setup Complete:' |
| message += '\n Remote API Server [%s]' % self.child_api_server.url |
| for child in self.children: |
| message += '\n %s' % child |
| message += '\n' |
| logging.info(message) |
| |
| |
| for child in self.children: |
| child.EnableStartRequests() |
| |
| def AssignPortsRandomly(self): |
| """Acquires a random port for each child process.""" |
| bound = [] |
| for child in self.children: |
| sock = socket.socket() |
| sock.bind(('localhost', 0)) |
| bound.append(sock) |
| child.SetHostPort(self.host, sock.getsockname()[1]) |
| for sock in bound: |
| sock.close() |
| |
| def __str__(self): |
| result = '[%s]' % self.Type() |
| if self.desc: |
| result += ' [%s]' % self.desc |
| return result |
| |
| def SetType(self, process_type): |
| if process_type not in DevProcess.TYPES: |
| raise Error('Unknown process type: %s' % process_type) |
| if self.process_type is not None: |
| raise Error('Process type cannot be set more than once.') |
| self.process_type = process_type |
| |
| def Type(self): |
| return self.process_type |
| |
| def IsDefault(self): |
| """Indicates whether this is the default dev_appserver process.""" |
| return self.Type() is None |
| |
| def IsMaster(self): |
| """Indicates whether this is the master process.""" |
| return self.Type() == DevProcess.TYPE_MASTER |
| |
| def IsSubprocess(self): |
| """Indicates that this is a subprocessess of the dev_appserver.""" |
| return not (self.IsDefault() or self.IsMaster()) |
| |
| def IsAppInstance(self): |
| """Indicates whether this process represents an application instance.""" |
| return self.Type() == DevProcess.TYPE_APP_INSTANCE |
| |
| def IsBackend(self): |
| """Indicates whether this process represents a backend.""" |
| return self.IsBackendBalancer() or self.IsBackendInstance() |
| |
| def IsBackendBalancer(self): |
| """Indicates whether this process represents a backend load balancer.""" |
| return self.Type() == DevProcess.TYPE_BACKEND_BALANCER |
| |
| def IsBackendInstance(self): |
| """Indicates whether this process represents a backend instance.""" |
| return self.Type() == DevProcess.TYPE_BACKEND_INSTANCE |
| |
| def IsBalancer(self): |
| """Indicates whether this process represents a load balancer.""" |
| return self.IsMaster() or self.IsBackendBalancer() |
| |
| def IsInstance(self): |
| """Indicates whether this process represents an instance.""" |
| return self.IsAppInstance() or self.IsBackendInstance() |
| |
| def InitBalanceSet(self): |
| """Construct a list of instances to balance traffic over.""" |
| if self.IsMaster(): |
| self.balance_set = [ self.child_app_instance.port ] |
| |
| if self.IsBackendBalancer(): |
| self.balance_set = [] |
| for instance in xrange(self.backend_entry.instances): |
| port = backends_api._get_dev_port(self.backend_id, instance) |
| self.balance_set.append(port) |
| |
| def GetBalanceSet(self): |
| """Return the set of ports over which this process balances requests.""" |
| return self.balance_set |
| |
| def FailFast(self): |
| """Indicates whether this process has fail-fast behavior.""" |
| if not self.backend_entry: |
| return False |
| if self.backend_entry.failfast: |
| return True |
| |
| return False |
| |
| def PrintStartMessage(self, app_id, host, port): |
| """Print the start message for processes that are started automatically.""" |
| url = 'http://%s:%d' % (host, port) |
| admin_url = '%s/_ah/admin' % url |
| if not self.IsSubprocess(): |
| |
| logging.info('Running application %s on port %d: %s', |
| app_id, port, url) |
| logging.info('Admin console is available at: %s', |
| admin_url) |
| |
| def Children(self): |
| """Returns the children of this process.""" |
| return self.children |
| |
| def MaybeConfigureRemoteDataApis(self): |
| """Set up stubs using remote_api as appropriate. |
| |
| If this is the API server (or is not multiprocess), return False. |
| Otherwise, set up the stubs for data based APIs as remote stubs pointing at |
| the to the API server and return True. |
| """ |
| if self.IsDefault(): |
| return False |
| |
| |
| |
| |
| |
| |
| |
| services = ( |
| 'app_identity_service', |
| 'capability_service', |
| 'datastore_v3', |
| 'mail', |
| 'memcache', |
| 'taskqueue', |
| 'urlfetch', |
| 'xmpp', |
| ) |
| remote_api_stub.ConfigureRemoteApi( |
| self.app_id, PATH_DEV_API_SERVER, lambda: ('', ''), |
| servername='%s:%d' % (API_SERVER_HOST, self.api_port), |
| services=services, use_remote_datastore=False) |
| return True |
| |
| def NewAppInfo(self, appinfo): |
| """Called when a new appinfo is read from disk on each request. |
| |
| The only action we take is to apply backend settings, such as the 'start' |
| directive, which adds a handler for /_ah/start. |
| |
| Args: |
| appinfo: An AppInfoExternal to be used on the next request. |
| """ |
| if self.backends: |
| appinfo.backends = self.backends |
| if self.IsBackend(): |
| appinfo.ApplyBackendSettings(self.backend_id) |
| |
| def UpdateEnv(self, env_dict): |
| """Copies backend port information to the supplied environment dictionary. |
| |
| This information is used by the Backends API to resolve backend and instance |
| addresses in the dev_appserver. |
| |
| User-supplied code has no access to the default environment. This method |
| will copy the environment variables needed for the backends api from the |
| default environment to the environment where user supplied code runs. |
| |
| Args: |
| env_dict: Dictionary with the new environment. |
| """ |
| if self.backend_id: |
| env_dict['BACKEND_ID'] = self.backend_id |
| if self.instance_id is not None: |
| env_dict['INSTANCE_ID'] = str(self.instance_id) |
| |
| for key in os.environ: |
| if key.startswith('BACKEND_PORT'): |
| env_dict[key] = os.environ[key] |
| |
| def ProcessRequest(self, request, client_address): |
| """Handles the SocketServer process_request call. |
| |
| If the request is to a backend the request will be handled by a separate |
| thread. If the backend is busy a 503 response will be sent. |
| |
| If this is a balancer instance each incoming request will be forwarded to |
| its own thread and handled there. |
| |
| If no backends are configured this override has no effect. |
| |
| Args: |
| http_server: The http server handling the request |
| request: the request to process |
| client_address: the client address |
| """ |
| assert not self.IsDefault() |
| |
| if self.IsBalancer(): |
| |
| ForwardRequestThread(request, client_address).start() |
| return |
| |
| assert self.IsAppInstance() or self.IsBackendInstance() |
| |
| if self.handle_requests.Active(): |
| if self.FailFast(): |
| logging.info('respond busy') |
| RespondBusyHandler(request, client_address) |
| return |
| |
| self.handle_requests.Enqueue(request, client_address) |
| |
| def HandleRequest(self, request): |
| """Hook that allows the DevProcess a chance to respond to requests. |
| |
| This hook is invoked just before normal request dispatch occurs in |
| dev_appserver.py. |
| |
| Args: |
| request: The request to be handled. |
| |
| Returns: |
| bool: Indicates whether the request was handled here. If False, normal |
| request handling should proceed. |
| """ |
| if self.IsBackendInstance() and not self.started: |
| if request.path != '/_ah/start': |
| request.send_response(httplib.FORBIDDEN, |
| 'Waiting for start request to finish.') |
| return True |
| |
| |
| return False |
| |
| def RequestComplete(self, request, response): |
| """Invoked when the process has finished handling a request.""" |
| rc = response.status_code |
| |
| if request.path == '/_ah/start': |
| if (rc >= 200 and rc < 300) or rc == 404: |
| self.started = True |
| |
| def UpdateSystemStub(self, system_service_stub): |
| """Copies info about the backends into the system stub.""" |
| if self.IsDefault(): |
| return |
| system_service_stub.set_backend_info(self.backends) |
| |
| |
| class HandleRequestThread(threading.Thread): |
| """Thread for handling HTTP requests. |
| |
| Instances needs to be able to respond with 503 when busy with other requests, |
| therefore requests are accepted in the main thread and forwarded to the |
| serving thread for processing. If the serving thread is busy with other |
| requests and the max pending queue length is reached a 503 error is sent back. |
| |
| Args: |
| http_server: Http server class handling the request. |
| max_pending_requests: The maximum number of pending requests in the queue. |
| """ |
| |
| def __init__(self): |
| threading.Thread.__init__(self) |
| self.setDaemon(True) |
| SetThreadName(self, 'HandleRequestThread') |
| |
| |
| self.active = False |
| |
| |
| self.pending = Queue.Queue() |
| |
| def Active(self): |
| """Indicates whether this thread is busy handling a request.""" |
| return self.active |
| |
| def Enqueue(self, request, client_address): |
| """Adds the indicated request to the pending request queue.""" |
| self.pending.put_nowait((request, client_address)) |
| |
| def run(self): |
| """Takes requests from the queue and handles them.""" |
| while True: |
| request, client_address = self.pending.get() |
| self.active = True |
| |
| |
| try: |
| HandleRequestDirectly(request, client_address) |
| except Exception, e: |
| logging.info('Exception in HandleRequestThread', exc_info=1) |
| finally: |
| self.active = False |
| |
| |
| class RespondBusyHandler(BaseHTTPServer.BaseHTTPRequestHandler): |
| """Handler that always will send back a 503 error.""" |
| |
| def __init__(self, request, client_address): |
| BaseHTTPServer.BaseHTTPRequestHandler.__init__( |
| self, request, client_address, HttpServer()) |
| |
| def handle_one_request(self): |
| """Override.""" |
| self.raw_requestline = self.rfile.readline() |
| if not self.raw_requestline: |
| self.close_connection = 1 |
| return |
| if not self.parse_request(): |
| return |
| self.send_error(httplib.SERVICE_UNAVAILABLE, 'Busy.') |
| |
| |
| class ForwardRequestThread(threading.Thread): |
| """Forwards an incoming request in a separate thread.""" |
| |
| def __init__(self, request, client_address): |
| threading.Thread.__init__(self) |
| self.request = request |
| self.client_address = client_address |
| self.setDaemon(True) |
| SetThreadName(self, 'ForwardRequestThread') |
| |
| def run(self): |
| ForwardRequestHandler(self.request, self.client_address) |
| |
| |
| class ForwardRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler): |
| """Forwards the incoming request to next free backend instance.""" |
| |
| def __init__(self, |
| request, |
| client_address, |
| connection_handler=httplib.HTTPConnection): |
| """Constructor extending BaseHTTPRequestHandler. |
| |
| Args: |
| request: The incoming request. |
| client_address: A (ip, port) tuple with the address of the client. |
| backend: The HTTPServer that received the request. |
| connection_handler: http library to use when balancer the connection to |
| the next available backend instance. Used for dependency injection. |
| """ |
| self.connection_handler = connection_handler |
| BaseHTTPServer.BaseHTTPRequestHandler.__init__(self, |
| request, |
| client_address, |
| HttpServer()) |
| |
| def handle_one_request(self): |
| """Override. Invoked from BaseHTTPRequestHandler constructor.""" |
| self.raw_requestline = self.rfile.readline() |
| if not self.raw_requestline: |
| self.close_connection = 1 |
| return |
| if not self.parse_request(): |
| return |
| |
| process = GlobalProcess() |
| balance_set = process.GetBalanceSet() |
| request_size = int(self.headers.get('content-length', 0)) |
| payload = self.rfile.read(request_size) |
| |
| |
| |
| |
| for port in balance_set: |
| logging.debug('balancer to port %d', port) |
| connection = self.connection_handler(process.host, port=port) |
| |
| |
| connection.response_class = ForwardResponse |
| connection.request(self.command, self.path, payload, dict(self.headers)) |
| try: |
| response = connection.getresponse() |
| except httplib.HTTPException, e: |
| |
| |
| self.send_error(httplib.INTERNAL_SERVER_ERROR, str(e)) |
| return |
| |
| if response.status != httplib.SERVICE_UNAVAILABLE: |
| self.wfile.write(response.data) |
| return |
| |
| |
| self.send_error(httplib.SERVICE_UNAVAILABLE, 'Busy') |
| |
| |
| class ForwardResponse(httplib.HTTPResponse): |
| """Modifies the HTTPResponse class so the raw request data is saved. |
| |
| This class is used by balancer instances when balancer requests to a |
| backend instance. |
| """ |
| |
| def __init__(self, sock, debuglevel=0, strict=0, method=None): |
| httplib.HTTPResponse.__init__(self, sock, debuglevel, strict, method) |
| self.data = self.fp.read() |
| self.fp = cStringIO.StringIO(self.data) |
| |
| |
| |
| |
| |
| _dev_process = DevProcess() |
| |
| |
| def GlobalProcess(): |
| """Returns a global DevProcess object representing the current process.""" |
| return _dev_process |
| |
| |
| def Enabled(): |
| """Indicates whether the dev_appserver is running in multiprocess mode.""" |
| return not GlobalProcess().IsDefault() |
| |
| |
| def HttpServer(): |
| """Returns the HTTPServer used by this process.""" |
| return GlobalProcess().HttpServer() |
| |
| |
| def HandleRequestDirectly(request, client_address): |
| """Handles the indicated request directly, without additional processing.""" |
| BaseHTTPServer.HTTPServer.process_request( |
| HttpServer(), request, client_address) |
| |
| |
| def PosixShutdown(): |
| """Kills a posix process with os.kill.""" |
| dev_process = GlobalProcess() |
| children = dev_process.Children() |
| for term_signal in (signal.SIGTERM, signal.SIGKILL): |
| for child in children: |
| if child.process is None: |
| continue |
| if child.process.returncode is not None: |
| continue |
| pid = child.process.pid |
| try: |
| logging.debug('posix kill %d with signal %d', pid, term_signal) |
| os.kill(pid, term_signal) |
| except OSError, err: |
| logging.error('Error encountered sending pid %d signal %d:%s\n', |
| pid, term_signal, err) |
| break |
| |
| time.sleep(0.2) |
| |
| |
| for child in children: |
| if child.process is None: |
| continue |
| if child.process.returncode is not None: |
| continue |
| try: |
| child.process.wait() |
| except OSError, e: |
| if e.errno != errno.ECHILD: |
| raise e |
| |
| |
| def Shutdown(): |
| """Shut down any child processes started.""" |
| dev_process = GlobalProcess() |
| if not dev_process.IsMaster(): |
| return |
| |
| |
| |
| if os.name == 'nt': |
| |
| import ctypes |
| for child in dev_process.Children(): |
| logging.debug('windows kill ' + str(child.process.pid)) |
| ctypes.windll.kernel32.TerminateProcess(int(child.process._handle), -1) |
| else: |
| PosixShutdown() |
| |
| dev_process.child_api_server.Quit() |
| |
| |
| def SetLogPrefix(prefix): |
| """Adds a prefix to the log handler to identify the process. |
| |
| Args: |
| prefix: The prefix string to append at the beginning of each line. |
| """ |
| formatter = logging.Formatter( |
| str(prefix) + ' [%(filename)s:%(lineno)d] %(levelname)s %(message)s') |
| logging._acquireLock() |
| try: |
| for handler in logging._handlerList: |
| if isinstance(handler, weakref.ref): |
| handler = handler() |
| if handler: |
| handler.setFormatter(formatter) |
| finally: |
| logging._releaseLock() |
| |
| |
| def Init(argv, options, root_path, appinfo): |
| """Enter multiprocess mode, if required. |
| |
| The dev_appserver runs in multiprocess mode if any Backends are configured. |
| The initial process becomes a "master" which acts as a router for the app, and |
| centralized memcache/datastore API server for sharing persistent state. |
| |
| This method works by configuring the global DevProcess object, which is |
| referenced by other files in the dev_appserver when necessary. The DevProcess |
| contains state indicating which role the current process plays in the |
| multiprocess architecture. |
| |
| The master process creates and shuts down subprocesses. A separate process is |
| created to represent an instance of the application, and a separate process is |
| created for each backend (to act as a load balancer) and for each backend |
| instance. |
| |
| On shutdown, the master process kills all subprocesses before exiting. |
| |
| Args: |
| argv: The command line arguments used when starting the main application. |
| options: Parsed dictionary of the command line arguments. |
| root_path: Root directory of the application. |
| appinfo: An AppInfoExternal object representing a parsed app.yaml file. |
| """ |
| if ARG_BACKENDS not in options: |
| return |
| |
| backends_path = os.path.join(root_path, 'backends.yaml') |
| if not os.path.exists(backends_path): |
| backends = [] |
| else: |
| backends_fh = open(backends_path) |
| try: |
| backend_info = backendinfo.LoadBackendInfo(backends_fh.read()) |
| finally: |
| backends_fh.close() |
| backends = backend_info.backends |
| |
| backend_set = set() |
| for backend in backends: |
| if backend.name in backend_set: |
| raise Error('Duplicate backend: %s' % backend.name) |
| if backend.instances is None: |
| backend.instances = 1 |
| elif backend.instances > BACKEND_MAX_INSTANCES: |
| raise Error('Maximum number of instances is %d', BACKEND_MAX_INSTANCES) |
| backend_set.add(backend.name) |
| |
| process = _dev_process |
| process.Init(appinfo, backends, options) |
| |
| if process.IsDefault(): |
| logging.info('Default process') |
| return |
| |
| SetLogPrefix(process) |
| if process.IsMaster(): |
| process.StartChildren(argv, options) |
| |
| process.InitBalanceSet() |
| |
| |
| if process.IsMaster(): |
| options['require_indexes'] = False |
| else: |
| options['require_indexes'] = True |
| |
| options['clear_datastore'] = False |
| options['clear_prospective_search'] = False |
