We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[ENH] --ui-port option for web service can be removed
As no actor pools are created in Mars Worker, the option -p can be adopted as http port, and --ui-port can be merged.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mars/web/__main__.py`
Content:
```
1 #!/usr/bin/env python
2 # -*- coding: utf-8 -*-
3 # Copyright 1999-2018 Alibaba Group Holding Ltd.
4 #
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
8 #
9 #      http://www.apache.org/licenses/LICENSE-2.0
10 #
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16 
17 import gevent.monkey
18 gevent.monkey.patch_all(thread=False)
19 
20 import logging  # noqa: E402
21 import random   # noqa: E402
22 import time     # noqa: E402
23 
24 from ..base_app import BaseApplication   # noqa: E402
25 from ..compat import six                 # noqa: E402
26 from ..errors import StartArgumentError  # noqa: E402
27 
28 logger = logging.getLogger(__name__)
29 
30 
31 class WebApplication(BaseApplication):
32     def __init__(self):
33         super(WebApplication, self).__init__()
34         self.mars_web = None
35         self.require_pool = False
36 
37     def config_args(self, parser):
38         parser.add_argument('--ui-port', help='port of Mars UI')
39 
40     def validate_arguments(self):
41         if not self.args.schedulers and not self.args.kv_store:
42             raise StartArgumentError('Either schedulers or url of kv store is required.')
43 
44     def main_loop(self):
45         try:
46             self.start()
47             while True:
48                 time.sleep(0.1)
49         finally:
50             self.stop()
51 
52     def start(self):
53         from .server import MarsWeb
54         if MarsWeb is None:
55             self.mars_web = None
56             logger.warning('Mars UI cannot be loaded. Please check if necessary components are installed.')
57         else:
58             ui_port = int(self.args.ui_port) if self.args.ui_port else None
59             scheduler_ip = self.args.schedulers or None
60             if isinstance(scheduler_ip, six.string_types):
61                 schedulers = scheduler_ip.split(',')
62                 scheduler_ip = random.choice(schedulers)
63             self.mars_web = MarsWeb(port=ui_port, scheduler_ip=scheduler_ip)
64             self.mars_web.start()
65 
66     def stop(self):
67         if self.mars_web:
68             self.mars_web.stop()
69 
70 
71 main = WebApplication()
72 
73 if __name__ == '__main__':
74     main()
75
```

Path: `mars/base_app.py`
Content:
```
1 # Copyright 1999-2018 Alibaba Group Holding Ltd.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import argparse
16 import logging
17 import os
18 import sys
19 
20 from .actors import create_actor_pool
21 from .config import options
22 from .errors import StartArgumentError
23 from .lib.tblib import pickling_support
24 from .utils import get_next_port
25 
26 pickling_support.install()
27 logger = logging.getLogger(__name__)
28 
29 try:
30     from pytest_cov.embed import cleanup_on_sigterm
31     cleanup_on_sigterm()
32 except ImportError:  # pragma: no cover
33     pass
34 
35 
36 class BaseApplication(object):
37     """
38     :type pool mars.actors.pool.gevent_pool.ActorContext
39     """
40     service_description = ''
41     service_logger = logger
42 
43     def __init__(self):
44         self.args = None
45         self.endpoint = None
46         self.pool = None
47         self.n_process = None
48 
49         self._running = False
50 
51     def __call__(self, argv=None):
52         import json
53 
54         if argv is None:
55             argv = sys.argv[1:]
56         new_argv = []
57         for a in argv:
58             if not a.startswith('-D'):
59                 new_argv.append(a)
60                 continue
61             conf, val = a[2:].split('=', 1)
62             conf_parts = conf.split('.')
63             conf_obj = options
64             for g in conf_parts[:-1]:
65                 conf_obj = getattr(conf_obj, g)
66             try:
67                 setattr(conf_obj, conf_parts[-1], json.loads(val))
68             except:
69                 setattr(conf_obj, conf_parts[-1], val)
70 
71         return self._main(new_argv)
72 
73     def _main(self, argv=None):
74         parser = argparse.ArgumentParser(description=self.service_description)
75         parser.add_argument('-a', '--advertise', help='advertise ip')
76         parser.add_argument('-k', '--kv-store', help='address of kv store service, '
77                                                      'for instance, etcd://localhost:4001')
78         parser.add_argument('-e', '--endpoint', help='endpoint of the service')
79         parser.add_argument('-s', '--schedulers', help='endpoint of scheduler, when single scheduler '
80                                                        'and etcd is not available')
81         parser.add_argument('-H', '--host', help='host of the scheduler service, only available '
82                                                  'when `endpoint` is absent')
83         parser.add_argument('-p', '--port', help='port of the scheduler service, only available '
84                                                  'when `endpoint` is absent')
85         parser.add_argument('--level', help='log level')
86         parser.add_argument('--format', help='log format')
87         parser.add_argument('--log_conf', help='log config file')
88         parser.add_argument('--inspect', help='inspection endpoint')
89         parser.add_argument('--load-modules', nargs='*', help='modules to import')
90         self.config_args(parser)
91         args = parser.parse_args(argv)
92         self.args = args
93 
94         endpoint = args.endpoint
95         host = args.host
96         port = args.port
97         options.kv_store = args.kv_store if args.kv_store else options.kv_store
98 
99         load_modules = []
100         for mod in args.load_modules or ():
101             load_modules.extend(mod.split(','))
102         if not args.load_modules:
103             load_module_str = os.environ.get('MARS_LOAD_MODULES')
104             if load_module_str:
105                 load_modules = load_module_str.split(',')
106         load_modules.append('mars.executor')
107         for m in load_modules:
108             __import__(m, globals(), locals(), [])
109         self.service_logger.info('Modules %s loaded', ','.join(load_modules))
110 
111         self.n_process = 1
112 
113         self.config_service()
114         self.config_logging()
115 
116         if not host:
117             host = args.advertise or '0.0.0.0'
118         if not endpoint and port:
119             endpoint = host + ':' + port
120 
121         try:
122             self.validate_arguments()
123         except StartArgumentError as ex:
124             parser.error('Failed to start application: %s' % ex)
125 
126         if getattr(self, 'require_pool', True):
127             self.endpoint, self.pool = self._try_create_pool(endpoint=endpoint, host=host, port=port)
128         self.service_logger.info('%s started at %s.', self.service_description, self.endpoint)
129         self.main_loop()
130 
131     def config_logging(self):
132         import logging.config
133         log_conf = self.args.log_conf or 'logging.conf'
134 
135         conf_file_paths = [
136             '', os.path.abspath('.'),
137             os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
138         ]
139         log_configured = False
140         for path in conf_file_paths:
141             conf_path = log_conf
142             if path:
143                 conf_path = os.path.join(conf_path)
144             if os.path.exists(conf_path):
145                 logging.config.fileConfig(conf_path, disable_existing_loggers=False)
146                 log_configured = True
147 
148         if not log_configured:
149             if not self.args.level:
150                 level = logging.INFO
151             else:
152                 level = getattr(logging, self.args.level.upper())
153             logging.getLogger('mars').setLevel(level)
154             logging.basicConfig(format=self.args.format)
155 
156     def validate_arguments(self):
157         pass
158 
159     def _try_create_pool(self, endpoint=None, host=None, port=None):
160         pool = None
161         if endpoint:
162             pool = self.create_pool(address=endpoint)
163         else:
164             use_port = None
165             retrial = 5
166             while use_port is None:
167                 use_port = port or get_next_port()
168                 try:
169                     endpoint = '{0}:{1}'.format(host, use_port)
170                     pool = self.create_pool(address=endpoint)
171                     break
172                 except:
173                     retrial -= 1
174                     if retrial == 0:
175                         raise
176 
177                     if port is None:
178                         use_port = None
179                     else:
180                         raise
181         return endpoint, pool
182 
183     def create_pool(self, *args, **kwargs):
184         kwargs.update(dict(n_process=self.n_process, backend='gevent'))
185         return create_actor_pool(*args, **kwargs)
186 
187     def main_loop(self):
188         try:
189             with self.pool:
190                 try:
191                     self.start()
192                     self._running = True
193                     while True:
194                         self.pool.join(1)
195                         stopped = []
196                         for idx, proc in enumerate(self.pool.processes):
197                             if not proc.is_alive():
198                                 stopped.append(idx)
199                         if stopped:
200                             self.handle_process_down(stopped)
201                 finally:
202                     self.stop()
203         finally:
204             self._running = False
205 
206     def handle_process_down(self, proc_indices):
207         """
208         Handle process down event, the default action is to quit
209         the whole application. Applications can inherit this method
210         to do customized process-level failover.
211 
212         :param proc_indices: indices of processes (not pids)
213         """
214         for idx in proc_indices:
215             proc = self.pool.processes[idx]
216             self.service_logger.fatal(
217                 'Process %d exited unpredictably. exitcode=%d', proc.pid, proc.exitcode)
218         raise KeyboardInterrupt
219 
220     def config_service(self):
221         pass
222 
223     def config_args(self, parser):
224         raise NotImplementedError
225 
226     def start(self):
227         raise NotImplementedError
228 
229     def stop(self):
230         raise NotImplementedError
231
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/mars/base_app.py b/mars/base_app.py
--- a/mars/base_app.py
+++ b/mars/base_app.py
@@ -76,11 +76,11 @@
         parser.add_argument('-k', '--kv-store', help='address of kv store service, '
                                                      'for instance, etcd://localhost:4001')
         parser.add_argument('-e', '--endpoint', help='endpoint of the service')
-        parser.add_argument('-s', '--schedulers', help='endpoint of scheduler, when single scheduler '
+        parser.add_argument('-s', '--schedulers', help='endpoint of schedulers, when single scheduler '
                                                        'and etcd is not available')
-        parser.add_argument('-H', '--host', help='host of the scheduler service, only available '
+        parser.add_argument('-H', '--host', help='host of the service, only available '
                                                  'when `endpoint` is absent')
-        parser.add_argument('-p', '--port', help='port of the scheduler service, only available '
+        parser.add_argument('-p', '--port', help='port of the service, only available '
                                                  'when `endpoint` is absent')
         parser.add_argument('--level', help='log level')
         parser.add_argument('--format', help='log format')
@@ -125,7 +125,7 @@
 
         if getattr(self, 'require_pool', True):
             self.endpoint, self.pool = self._try_create_pool(endpoint=endpoint, host=host, port=port)
-        self.service_logger.info('%s started at %s.', self.service_description, self.endpoint)
+            self.service_logger.info('%s started at %s.', self.service_description, self.endpoint)
         self.main_loop()
 
     def config_logging(self):
diff --git a/mars/web/__main__.py b/mars/web/__main__.py
--- a/mars/web/__main__.py
+++ b/mars/web/__main__.py
@@ -17,9 +17,10 @@
 import gevent.monkey
 gevent.monkey.patch_all(thread=False)
 
-import logging  # noqa: E402
-import random   # noqa: E402
-import time     # noqa: E402
+import argparse  # noqa: E402
+import logging   # noqa: E402
+import random    # noqa: E402
+import time      # noqa: E402
 
 from ..base_app import BaseApplication   # noqa: E402
 from ..compat import six                 # noqa: E402
@@ -35,7 +36,7 @@
         self.require_pool = False
 
     def config_args(self, parser):
-        parser.add_argument('--ui-port', help='port of Mars UI')
+        parser.add_argument('--ui-port', help=argparse.SUPPRESS)
 
     def validate_arguments(self):
         if not self.args.schedulers and not self.args.kv_store:
@@ -55,7 +56,8 @@
             self.mars_web = None
             logger.warning('Mars UI cannot be loaded. Please check if necessary components are installed.')
         else:
-            ui_port = int(self.args.ui_port) if self.args.ui_port else None
+            port_arg = self.args.ui_port or self.args.port
+            ui_port = int(port_arg) if port_arg else None
             scheduler_ip = self.args.schedulers or None
             if isinstance(scheduler_ip, six.string_types):
                 schedulers = scheduler_ip.split(',')

{"golden_diff": "diff --git a/mars/base_app.py b/mars/base_app.py\n--- a/mars/base_app.py\n+++ b/mars/base_app.py\n@@ -76,11 +76,11 @@\n         parser.add_argument('-k', '--kv-store', help='address of kv store service, '\n                                                      'for instance, etcd://localhost:4001')\n         parser.add_argument('-e', '--endpoint', help='endpoint of the service')\n-        parser.add_argument('-s', '--schedulers', help='endpoint of scheduler, when single scheduler '\n+        parser.add_argument('-s', '--schedulers', help='endpoint of schedulers, when single scheduler '\n                                                        'and etcd is not available')\n-        parser.add_argument('-H', '--host', help='host of the scheduler service, only available '\n+        parser.add_argument('-H', '--host', help='host of the service, only available '\n                                                  'when `endpoint` is absent')\n-        parser.add_argument('-p', '--port', help='port of the scheduler service, only available '\n+        parser.add_argument('-p', '--port', help='port of the service, only available '\n                                                  'when `endpoint` is absent')\n         parser.add_argument('--level', help='log level')\n         parser.add_argument('--format', help='log format')\n@@ -125,7 +125,7 @@\n \n         if getattr(self, 'require_pool', True):\n             self.endpoint, self.pool = self._try_create_pool(endpoint=endpoint, host=host, port=port)\n-        self.service_logger.info('%s started at %s.', self.service_description, self.endpoint)\n+            self.service_logger.info('%s started at %s.', self.service_description, self.endpoint)\n         self.main_loop()\n \n     def config_logging(self):\ndiff --git a/mars/web/__main__.py b/mars/web/__main__.py\n--- a/mars/web/__main__.py\n+++ b/mars/web/__main__.py\n@@ -17,9 +17,10 @@\n import gevent.monkey\n gevent.monkey.patch_all(thread=False)\n \n-import logging  # noqa: E402\n-import random   # noqa: E402\n-import time     # noqa: E402\n+import argparse  # noqa: E402\n+import logging   # noqa: E402\n+import random    # noqa: E402\n+import time      # noqa: E402\n \n from ..base_app import BaseApplication   # noqa: E402\n from ..compat import six                 # noqa: E402\n@@ -35,7 +36,7 @@\n         self.require_pool = False\n \n     def config_args(self, parser):\n-        parser.add_argument('--ui-port', help='port of Mars UI')\n+        parser.add_argument('--ui-port', help=argparse.SUPPRESS)\n \n     def validate_arguments(self):\n         if not self.args.schedulers and not self.args.kv_store:\n@@ -55,7 +56,8 @@\n             self.mars_web = None\n             logger.warning('Mars UI cannot be loaded. Please check if necessary components are installed.')\n         else:\n-            ui_port = int(self.args.ui_port) if self.args.ui_port else None\n+            port_arg = self.args.ui_port or self.args.port\n+            ui_port = int(port_arg) if port_arg else None\n             scheduler_ip = self.args.schedulers or None\n             if isinstance(scheduler_ip, six.string_types):\n                 schedulers = scheduler_ip.split(',')\n", "issue": "[ENH] --ui-port option for web service can be removed\nAs no actor pools are created in Mars Worker, the option -p can be adopted as http port, and --ui-port can be merged.\n", "before_files": [{"content": "#!/usr/bin/env python\n# -*- coding: utf-8 -*-\n# Copyright 1999-2018 Alibaba Group Holding Ltd.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport gevent.monkey\ngevent.monkey.patch_all(thread=False)\n\nimport logging  # noqa: E402\nimport random   # noqa: E402\nimport time     # noqa: E402\n\nfrom ..base_app import BaseApplication   # noqa: E402\nfrom ..compat import six                 # noqa: E402\nfrom ..errors import StartArgumentError  # noqa: E402\n\nlogger = logging.getLogger(__name__)\n\n\nclass WebApplication(BaseApplication):\n    def __init__(self):\n        super(WebApplication, self).__init__()\n        self.mars_web = None\n        self.require_pool = False\n\n    def config_args(self, parser):\n        parser.add_argument('--ui-port', help='port of Mars UI')\n\n    def validate_arguments(self):\n        if not self.args.schedulers and not self.args.kv_store:\n            raise StartArgumentError('Either schedulers or url of kv store is required.')\n\n    def main_loop(self):\n        try:\n            self.start()\n            while True:\n                time.sleep(0.1)\n        finally:\n            self.stop()\n\n    def start(self):\n        from .server import MarsWeb\n        if MarsWeb is None:\n            self.mars_web = None\n            logger.warning('Mars UI cannot be loaded. Please check if necessary components are installed.')\n        else:\n            ui_port = int(self.args.ui_port) if self.args.ui_port else None\n            scheduler_ip = self.args.schedulers or None\n            if isinstance(scheduler_ip, six.string_types):\n                schedulers = scheduler_ip.split(',')\n                scheduler_ip = random.choice(schedulers)\n            self.mars_web = MarsWeb(port=ui_port, scheduler_ip=scheduler_ip)\n            self.mars_web.start()\n\n    def stop(self):\n        if self.mars_web:\n            self.mars_web.stop()\n\n\nmain = WebApplication()\n\nif __name__ == '__main__':\n    main()\n", "path": "mars/web/__main__.py"}, {"content": "# Copyright 1999-2018 Alibaba Group Holding Ltd.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport argparse\nimport logging\nimport os\nimport sys\n\nfrom .actors import create_actor_pool\nfrom .config import options\nfrom .errors import StartArgumentError\nfrom .lib.tblib import pickling_support\nfrom .utils import get_next_port\n\npickling_support.install()\nlogger = logging.getLogger(__name__)\n\ntry:\n    from pytest_cov.embed import cleanup_on_sigterm\n    cleanup_on_sigterm()\nexcept ImportError:  # pragma: no cover\n    pass\n\n\nclass BaseApplication(object):\n    \"\"\"\n    :type pool mars.actors.pool.gevent_pool.ActorContext\n    \"\"\"\n    service_description = ''\n    service_logger = logger\n\n    def __init__(self):\n        self.args = None\n        self.endpoint = None\n        self.pool = None\n        self.n_process = None\n\n        self._running = False\n\n    def __call__(self, argv=None):\n        import json\n\n        if argv is None:\n            argv = sys.argv[1:]\n        new_argv = []\n        for a in argv:\n            if not a.startswith('-D'):\n                new_argv.append(a)\n                continue\n            conf, val = a[2:].split('=', 1)\n            conf_parts = conf.split('.')\n            conf_obj = options\n            for g in conf_parts[:-1]:\n                conf_obj = getattr(conf_obj, g)\n            try:\n                setattr(conf_obj, conf_parts[-1], json.loads(val))\n            except:\n                setattr(conf_obj, conf_parts[-1], val)\n\n        return self._main(new_argv)\n\n    def _main(self, argv=None):\n        parser = argparse.ArgumentParser(description=self.service_description)\n        parser.add_argument('-a', '--advertise', help='advertise ip')\n        parser.add_argument('-k', '--kv-store', help='address of kv store service, '\n                                                     'for instance, etcd://localhost:4001')\n        parser.add_argument('-e', '--endpoint', help='endpoint of the service')\n        parser.add_argument('-s', '--schedulers', help='endpoint of scheduler, when single scheduler '\n                                                       'and etcd is not available')\n        parser.add_argument('-H', '--host', help='host of the scheduler service, only available '\n                                                 'when `endpoint` is absent')\n        parser.add_argument('-p', '--port', help='port of the scheduler service, only available '\n                                                 'when `endpoint` is absent')\n        parser.add_argument('--level', help='log level')\n        parser.add_argument('--format', help='log format')\n        parser.add_argument('--log_conf', help='log config file')\n        parser.add_argument('--inspect', help='inspection endpoint')\n        parser.add_argument('--load-modules', nargs='*', help='modules to import')\n        self.config_args(parser)\n        args = parser.parse_args(argv)\n        self.args = args\n\n        endpoint = args.endpoint\n        host = args.host\n        port = args.port\n        options.kv_store = args.kv_store if args.kv_store else options.kv_store\n\n        load_modules = []\n        for mod in args.load_modules or ():\n            load_modules.extend(mod.split(','))\n        if not args.load_modules:\n            load_module_str = os.environ.get('MARS_LOAD_MODULES')\n            if load_module_str:\n                load_modules = load_module_str.split(',')\n        load_modules.append('mars.executor')\n        for m in load_modules:\n            __import__(m, globals(), locals(), [])\n        self.service_logger.info('Modules %s loaded', ','.join(load_modules))\n\n        self.n_process = 1\n\n        self.config_service()\n        self.config_logging()\n\n        if not host:\n            host = args.advertise or '0.0.0.0'\n        if not endpoint and port:\n            endpoint = host + ':' + port\n\n        try:\n            self.validate_arguments()\n        except StartArgumentError as ex:\n            parser.error('Failed to start application: %s' % ex)\n\n        if getattr(self, 'require_pool', True):\n            self.endpoint, self.pool = self._try_create_pool(endpoint=endpoint, host=host, port=port)\n        self.service_logger.info('%s started at %s.', self.service_description, self.endpoint)\n        self.main_loop()\n\n    def config_logging(self):\n        import logging.config\n        log_conf = self.args.log_conf or 'logging.conf'\n\n        conf_file_paths = [\n            '', os.path.abspath('.'),\n            os.path.dirname(os.path.dirname(os.path.abspath(__file__)))\n        ]\n        log_configured = False\n        for path in conf_file_paths:\n            conf_path = log_conf\n            if path:\n                conf_path = os.path.join(conf_path)\n            if os.path.exists(conf_path):\n                logging.config.fileConfig(conf_path, disable_existing_loggers=False)\n                log_configured = True\n\n        if not log_configured:\n            if not self.args.level:\n                level = logging.INFO\n            else:\n                level = getattr(logging, self.args.level.upper())\n            logging.getLogger('mars').setLevel(level)\n            logging.basicConfig(format=self.args.format)\n\n    def validate_arguments(self):\n        pass\n\n    def _try_create_pool(self, endpoint=None, host=None, port=None):\n        pool = None\n        if endpoint:\n            pool = self.create_pool(address=endpoint)\n        else:\n            use_port = None\n            retrial = 5\n            while use_port is None:\n                use_port = port or get_next_port()\n                try:\n                    endpoint = '{0}:{1}'.format(host, use_port)\n                    pool = self.create_pool(address=endpoint)\n                    break\n                except:\n                    retrial -= 1\n                    if retrial == 0:\n                        raise\n\n                    if port is None:\n                        use_port = None\n                    else:\n                        raise\n        return endpoint, pool\n\n    def create_pool(self, *args, **kwargs):\n        kwargs.update(dict(n_process=self.n_process, backend='gevent'))\n        return create_actor_pool(*args, **kwargs)\n\n    def main_loop(self):\n        try:\n            with self.pool:\n                try:\n                    self.start()\n                    self._running = True\n                    while True:\n                        self.pool.join(1)\n                        stopped = []\n                        for idx, proc in enumerate(self.pool.processes):\n                            if not proc.is_alive():\n                                stopped.append(idx)\n                        if stopped:\n                            self.handle_process_down(stopped)\n                finally:\n                    self.stop()\n        finally:\n            self._running = False\n\n    def handle_process_down(self, proc_indices):\n        \"\"\"\n        Handle process down event, the default action is to quit\n        the whole application. Applications can inherit this method\n        to do customized process-level failover.\n\n        :param proc_indices: indices of processes (not pids)\n        \"\"\"\n        for idx in proc_indices:\n            proc = self.pool.processes[idx]\n            self.service_logger.fatal(\n                'Process %d exited unpredictably. exitcode=%d', proc.pid, proc.exitcode)\n        raise KeyboardInterrupt\n\n    def config_service(self):\n        pass\n\n    def config_args(self, parser):\n        raise NotImplementedError\n\n    def start(self):\n        raise NotImplementedError\n\n    def stop(self):\n        raise NotImplementedError\n", "path": "mars/base_app.py"}], "after_files": [{"content": "#!/usr/bin/env python\n# -*- coding: utf-8 -*-\n# Copyright 1999-2018 Alibaba Group Holding Ltd.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport gevent.monkey\ngevent.monkey.patch_all(thread=False)\n\nimport argparse  # noqa: E402\nimport logging   # noqa: E402\nimport random    # noqa: E402\nimport time      # noqa: E402\n\nfrom ..base_app import BaseApplication   # noqa: E402\nfrom ..compat import six                 # noqa: E402\nfrom ..errors import StartArgumentError  # noqa: E402\n\nlogger = logging.getLogger(__name__)\n\n\nclass WebApplication(BaseApplication):\n    def __init__(self):\n        super(WebApplication, self).__init__()\n        self.mars_web = None\n        self.require_pool = False\n\n    def config_args(self, parser):\n        parser.add_argument('--ui-port', help=argparse.SUPPRESS)\n\n    def validate_arguments(self):\n        if not self.args.schedulers and not self.args.kv_store:\n            raise StartArgumentError('Either schedulers or url of kv store is required.')\n\n    def main_loop(self):\n        try:\n            self.start()\n            while True:\n                time.sleep(0.1)\n        finally:\n            self.stop()\n\n    def start(self):\n        from .server import MarsWeb\n        if MarsWeb is None:\n            self.mars_web = None\n            logger.warning('Mars UI cannot be loaded. Please check if necessary components are installed.')\n        else:\n            port_arg = self.args.ui_port or self.args.port\n            ui_port = int(port_arg) if port_arg else None\n            scheduler_ip = self.args.schedulers or None\n            if isinstance(scheduler_ip, six.string_types):\n                schedulers = scheduler_ip.split(',')\n                scheduler_ip = random.choice(schedulers)\n            self.mars_web = MarsWeb(port=ui_port, scheduler_ip=scheduler_ip)\n            self.mars_web.start()\n\n    def stop(self):\n        if self.mars_web:\n            self.mars_web.stop()\n\n\nmain = WebApplication()\n\nif __name__ == '__main__':\n    main()\n", "path": "mars/web/__main__.py"}, {"content": "# Copyright 1999-2018 Alibaba Group Holding Ltd.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport argparse\nimport logging\nimport os\nimport sys\n\nfrom .actors import create_actor_pool\nfrom .config import options\nfrom .errors import StartArgumentError\nfrom .lib.tblib import pickling_support\nfrom .utils import get_next_port\n\npickling_support.install()\nlogger = logging.getLogger(__name__)\n\ntry:\n    from pytest_cov.embed import cleanup_on_sigterm\n    cleanup_on_sigterm()\nexcept ImportError:  # pragma: no cover\n    pass\n\n\nclass BaseApplication(object):\n    \"\"\"\n    :type pool mars.actors.pool.gevent_pool.ActorContext\n    \"\"\"\n    service_description = ''\n    service_logger = logger\n\n    def __init__(self):\n        self.args = None\n        self.endpoint = None\n        self.pool = None\n        self.n_process = None\n\n        self._running = False\n\n    def __call__(self, argv=None):\n        import json\n\n        if argv is None:\n            argv = sys.argv[1:]\n        new_argv = []\n        for a in argv:\n            if not a.startswith('-D'):\n                new_argv.append(a)\n                continue\n            conf, val = a[2:].split('=', 1)\n            conf_parts = conf.split('.')\n            conf_obj = options\n            for g in conf_parts[:-1]:\n                conf_obj = getattr(conf_obj, g)\n            try:\n                setattr(conf_obj, conf_parts[-1], json.loads(val))\n            except:\n                setattr(conf_obj, conf_parts[-1], val)\n\n        return self._main(new_argv)\n\n    def _main(self, argv=None):\n        parser = argparse.ArgumentParser(description=self.service_description)\n        parser.add_argument('-a', '--advertise', help='advertise ip')\n        parser.add_argument('-k', '--kv-store', help='address of kv store service, '\n                                                     'for instance, etcd://localhost:4001')\n        parser.add_argument('-e', '--endpoint', help='endpoint of the service')\n        parser.add_argument('-s', '--schedulers', help='endpoint of schedulers, when single scheduler '\n                                                       'and etcd is not available')\n        parser.add_argument('-H', '--host', help='host of the service, only available '\n                                                 'when `endpoint` is absent')\n        parser.add_argument('-p', '--port', help='port of the service, only available '\n                                                 'when `endpoint` is absent')\n        parser.add_argument('--level', help='log level')\n        parser.add_argument('--format', help='log format')\n        parser.add_argument('--log_conf', help='log config file')\n        parser.add_argument('--inspect', help='inspection endpoint')\n        parser.add_argument('--load-modules', nargs='*', help='modules to import')\n        self.config_args(parser)\n        args = parser.parse_args(argv)\n        self.args = args\n\n        endpoint = args.endpoint\n        host = args.host\n        port = args.port\n        options.kv_store = args.kv_store if args.kv_store else options.kv_store\n\n        load_modules = []\n        for mod in args.load_modules or ():\n            load_modules.extend(mod.split(','))\n        if not args.load_modules:\n            load_module_str = os.environ.get('MARS_LOAD_MODULES')\n            if load_module_str:\n                load_modules = load_module_str.split(',')\n        load_modules.append('mars.executor')\n        for m in load_modules:\n            __import__(m, globals(), locals(), [])\n        self.service_logger.info('Modules %s loaded', ','.join(load_modules))\n\n        self.n_process = 1\n\n        self.config_service()\n        self.config_logging()\n\n        if not host:\n            host = args.advertise or '0.0.0.0'\n        if not endpoint and port:\n            endpoint = host + ':' + port\n\n        try:\n            self.validate_arguments()\n        except StartArgumentError as ex:\n            parser.error('Failed to start application: %s' % ex)\n\n        if getattr(self, 'require_pool', True):\n            self.endpoint, self.pool = self._try_create_pool(endpoint=endpoint, host=host, port=port)\n            self.service_logger.info('%s started at %s.', self.service_description, self.endpoint)\n        self.main_loop()\n\n    def config_logging(self):\n        import logging.config\n        log_conf = self.args.log_conf or 'logging.conf'\n\n        conf_file_paths = [\n            '', os.path.abspath('.'),\n            os.path.dirname(os.path.dirname(os.path.abspath(__file__)))\n        ]\n        log_configured = False\n        for path in conf_file_paths:\n            conf_path = log_conf\n            if path:\n                conf_path = os.path.join(conf_path)\n            if os.path.exists(conf_path):\n                logging.config.fileConfig(conf_path, disable_existing_loggers=False)\n                log_configured = True\n\n        if not log_configured:\n            if not self.args.level:\n                level = logging.INFO\n            else:\n                level = getattr(logging, self.args.level.upper())\n            logging.getLogger('mars').setLevel(level)\n            logging.basicConfig(format=self.args.format)\n\n    def validate_arguments(self):\n        pass\n\n    def _try_create_pool(self, endpoint=None, host=None, port=None):\n        pool = None\n        if endpoint:\n            pool = self.create_pool(address=endpoint)\n        else:\n            use_port = None\n            retrial = 5\n            while use_port is None:\n                use_port = port or get_next_port()\n                try:\n                    endpoint = '{0}:{1}'.format(host, use_port)\n                    pool = self.create_pool(address=endpoint)\n                    break\n                except:\n                    retrial -= 1\n                    if retrial == 0:\n                        raise\n\n                    if port is None:\n                        use_port = None\n                    else:\n                        raise\n        return endpoint, pool\n\n    def create_pool(self, *args, **kwargs):\n        kwargs.update(dict(n_process=self.n_process, backend='gevent'))\n        return create_actor_pool(*args, **kwargs)\n\n    def main_loop(self):\n        try:\n            with self.pool:\n                try:\n                    self.start()\n                    self._running = True\n                    while True:\n                        self.pool.join(1)\n                        stopped = []\n                        for idx, proc in enumerate(self.pool.processes):\n                            if not proc.is_alive():\n                                stopped.append(idx)\n                        if stopped:\n                            self.handle_process_down(stopped)\n                finally:\n                    self.stop()\n        finally:\n            self._running = False\n\n    def handle_process_down(self, proc_indices):\n        \"\"\"\n        Handle process down event, the default action is to quit\n        the whole application. Applications can inherit this method\n        to do customized process-level failover.\n\n        :param proc_indices: indices of processes (not pids)\n        \"\"\"\n        for idx in proc_indices:\n            proc = self.pool.processes[idx]\n            self.service_logger.fatal(\n                'Process %d exited unpredictably. exitcode=%d', proc.pid, proc.exitcode)\n        raise KeyboardInterrupt\n\n    def config_service(self):\n        pass\n\n    def config_args(self, parser):\n        raise NotImplementedError\n\n    def start(self):\n        raise NotImplementedError\n\n    def stop(self):\n        raise NotImplementedError\n", "path": "mars/base_app.py"}]}