We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Close learn process gracefuly
`close_api_gracefully` only closes the API, but not any ongoing learn processes, which are left orphaned and consuming loads of memory.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mindsdb/interfaces/native/learn_process.py`
Content:
```
1 import os
2 import torch.multiprocessing as mp
3 
4 from mindsdb.interfaces.database.database import DatabaseWrapper
5 from mindsdb.utilities.os_specific import get_mp_context
6 from mindsdb.interfaces.storage.db import session, Predictor
7 from mindsdb.interfaces.storage.fs import FsSotre
8 from mindsdb.utilities.config import Config
9 
10 
11 ctx = mp.get_context('spawn')
12 
13 
14 class LearnProcess(ctx.Process):
15     daemon = True
16 
17     def __init__(self, *args):
18         super(LearnProcess, self).__init__(args=args)
19 
20     def run(self):
21         '''
22         running at subprocess due to
23         ValueError: signal only works in main thread
24 
25         this is work for celery worker here?
26         '''
27         import mindsdb_native
28 
29         fs_store = FsSotre()
30         config = Config()
31         company_id = os.environ.get('MINDSDB_COMPANY_ID', None)
32         name, from_data, to_predict, kwargs, datasource_id = self._args
33 
34         mdb = mindsdb_native.Predictor(name=name, run_env={'trigger': 'mindsdb'})
35 
36         predictor_record = Predictor.query.filter_by(company_id=company_id, name=name).first()
37         predictor_record.datasource_id = datasource_id
38         predictor_record.to_predict = to_predict
39         predictor_record.version = mindsdb_native.__version__
40         predictor_record.data = {
41             'name': name,
42             'status': 'training'
43         }
44         #predictor_record.datasource_id = ... <-- can be done once `learn` is passed a datasource name
45         session.commit()
46 
47         to_predict = to_predict if isinstance(to_predict, list) else [to_predict]
48         data_source = getattr(mindsdb_native, from_data['class'])(*from_data['args'], **from_data['kwargs'])
49         try:
50             mdb.learn(
51                 from_data=data_source,
52                 to_predict=to_predict,
53                 **kwargs
54             )
55         except Exception as e:
56             pass
57 
58         fs_store.put(name, f'predictor_{company_id}_{predictor_record.id}', config['paths']['predictors'])
59 
60         model_data = mindsdb_native.F.get_model_data(name)
61 
62         predictor_record = Predictor.query.filter_by(company_id=company_id, name=name).first()
63         predictor_record.data = model_data
64         session.commit()
65 
66         DatabaseWrapper().register_predictors([model_data])
67
```

Path: `mindsdb/utilities/ps.py`
Content:
```
1 import sys
2 import time
3 from collections import namedtuple
4 import psutil
5 
6 
7 def net_connections():
8     """Cross-platform psutil.net_connections like interface"""
9     if sys.platform.lower().startswith('linux'):
10         return psutil.net_connections()
11 
12     all_connections = []
13     Pconn = None
14     for p in psutil.process_iter(['pid']):
15         try:
16             process = psutil.Process(p.pid)
17             connections = process.connections()
18             if connections:
19                 for conn in connections:
20                     # Adding pid to the returned instance
21                     # for consistency with psutil.net_connections()
22                     if Pconn is None:
23                         fields = list(conn._fields)
24                         fields.append('pid')
25                         _conn = namedtuple('Pconn', fields)
26                     for attr in conn._fields:
27                         setattr(_conn, attr, getattr(conn, attr))
28                     _conn.pid = p.pid
29                     all_connections.append(_conn)
30 
31         except (psutil.AccessDenied, psutil.ZombieProcess, psutil.NoSuchProcess):
32             pass
33     return all_connections
34 
35 
36 def is_port_in_use(port_num):
37     """Check does any of child process uses specified port."""
38     parent_process = psutil.Process()
39     child_pids = [x.pid for x in parent_process.children(recursive=True)]
40     conns = net_connections()
41     portsinuse = [x.laddr[1] for x in conns if x.pid in child_pids and x.status == 'LISTEN']
42     portsinuse.sort()
43     return int(port_num) in portsinuse
44 
45 
46 def wait_func_is_true(func, timeout, *args, **kwargs):
47     start_time = time.time()
48 
49     result = func(*args, **kwargs)
50     while result is False and (time.time() - start_time) < timeout:
51         time.sleep(2)
52         result = func(*args, **kwargs)
53 
54     return result
55 
56 
57 def wait_port(port_num, timeout):
58     return wait_func_is_true(func=is_port_in_use, timeout=timeout, port_num=port_num)
59 
60 
61 def get_listen_ports(pid):
62     try:
63         p = psutil.Process(pid)
64         cons = p.connections()
65         cons = [x.laddr.port for x in cons]
66     except Exception:
67         return []
68     return cons
69 
70 
71 def is_pid_listen_port(pid, port):
72     ports = get_listen_ports(pid)
73     return int(port) in ports
74
```

Path: `mindsdb/__main__.py`
Content:
```
1 import atexit
2 import traceback
3 import sys
4 import os
5 import time
6 import asyncio
7 import datetime
8 import platform
9 
10 import torch.multiprocessing as mp
11 
12 from mindsdb.utilities.config import Config
13 from mindsdb.utilities.os_specific import get_mp_context
14 from mindsdb.interfaces.native.native import NativeInterface
15 from mindsdb.interfaces.custom.custom_models import CustomModels
16 from mindsdb.api.http.start import start as start_http
17 from mindsdb.api.mysql.start import start as start_mysql
18 from mindsdb.api.mongo.start import start as start_mongo
19 from mindsdb.utilities.ps import is_pid_listen_port
20 from mindsdb.interfaces.database.database import DatabaseWrapper
21 from mindsdb.utilities.functions import args_parse, get_all_models_meta_data
22 from mindsdb.utilities.log import log
23 
24 
25 def close_api_gracefully(apis):
26     try:
27         for api in apis.values():
28             process = api['process']
29             sys.stdout.flush()
30             process.terminate()
31             process.join()
32             sys.stdout.flush()
33     except KeyboardInterrupt:
34         sys.exit(0)
35 
36 
37 if __name__ == '__main__':
38     mp.freeze_support()
39     args = args_parse()
40     config = Config()
41 
42     if args.verbose is True:
43         config.set(['log', 'level', 'console'], 'DEBUG')
44 
45     os.environ['DEFAULT_LOG_LEVEL'] = config['log']['level']['console']
46     os.environ['LIGHTWOOD_LOG_LEVEL'] = config['log']['level']['console']
47     config.set(['mindsdb_last_started_at'], str(datetime.datetime.now()))
48     
49     from lightwood.__about__ import __version__ as lightwood_version
50     from mindsdb_native.__about__ import __version__ as mindsdb_native_version
51     from mindsdb.__about__ import __version__ as mindsdb_version
52     print('Versions:')
53     print(f' - lightwood {lightwood_version}')
54     print(f' - MindsDB_native {mindsdb_native_version}')
55     print(f' - MindsDB {mindsdb_version}')
56 
57     print(f'Configuration file:\n   {config.config_path}')
58     print(f"Storage path:\n   {config.paths['root']}")
59 
60     if args.api is None:
61         api_arr = ['http', 'mysql']
62     else:
63         api_arr = args.api.split(',')
64 
65     apis = {
66         api: {
67             'port': config['api'][api]['port'],
68             'process': None,
69             'started': False
70         } for api in api_arr
71     }
72 
73     for api_name in apis.keys():
74         if api_name not in config['api']:
75             print(f"Trying run '{api_name}' API, but is no config for this api.")
76             print(f"Please, fill config['api']['{api_name}']")
77             sys.exit(0)
78 
79     start_functions = {
80         'http': start_http,
81         'mysql': start_mysql,
82         'mongodb': start_mongo
83     }
84 
85     mdb = NativeInterface()
86     cst = CustomModels()
87 
88     model_data_arr = get_all_models_meta_data(mdb, cst)
89 
90     dbw = DatabaseWrapper()
91     for db_alias in config['integrations']:
92         dbw.setup_integration(db_alias)
93     dbw.register_predictors(model_data_arr)
94 
95     for broken_name in [name for name, connected in dbw.check_connections().items() if connected is False]:
96         log.error(f'Error failed to integrate with database aliased: {broken_name}')
97 
98     ctx = mp.get_context('spawn')
99     # Switch to this once the native interface has it's own thread :/
100     # ctx = mp.get_context(get_mp_context())
101 
102 
103     for api_name, api_data in apis.items():
104         print(f'{api_name} API: starting...')
105         try:
106             if api_name == 'http':
107                 p = ctx.Process(target=start_functions[api_name], args=(args.verbose,args.no_studio))
108             else:
109                 p = ctx.Process(target=start_functions[api_name], args=(args.verbose,))
110             p.start()
111             api_data['process'] = p
112         except Exception as e:
113             log.error(f'Failed to start {api_name} API with exception {e}\n{traceback.format_exc()}')
114             close_api_gracefully(apis)
115             raise e
116 
117     atexit.register(close_api_gracefully, apis=apis)
118 
119     async def wait_api_start(api_name, pid, port):
120         timeout = 60
121         start_time = time.time()
122         started = is_pid_listen_port(pid, port)
123         while (time.time() - start_time) < timeout and started is False:
124             await asyncio.sleep(0.5)
125             started = is_pid_listen_port(pid, port)
126         return api_name, port, started
127 
128     async def wait_apis_start():
129         futures = [
130             wait_api_start(api_name, api_data['process'].pid, api_data['port'])
131             for api_name, api_data in apis.items()
132         ]
133         for i, future in enumerate(asyncio.as_completed(futures)):
134             api_name, port, started = await future
135             if started:
136                 print(f"{api_name} API: started on {port}")
137             else:
138                 log.error(f"ERROR: {api_name} API cant start on {port}")
139 
140     ioloop = asyncio.get_event_loop()
141     ioloop.run_until_complete(wait_apis_start())
142     ioloop.close()
143 
144     try:
145         for api_data in apis.values():
146             api_data['process'].join()
147     except KeyboardInterrupt:
148         print('Closing app...')
149
```
--- 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/mindsdb/__main__.py b/mindsdb/__main__.py
--- a/mindsdb/__main__.py
+++ b/mindsdb/__main__.py
@@ -5,7 +5,7 @@
 import time
 import asyncio
 import datetime
-import platform
+import signal
 
 import torch.multiprocessing as mp
 
@@ -16,7 +16,7 @@
 from mindsdb.api.http.start import start as start_http
 from mindsdb.api.mysql.start import start as start_mysql
 from mindsdb.api.mongo.start import start as start_mongo
-from mindsdb.utilities.ps import is_pid_listen_port
+from mindsdb.utilities.ps import is_pid_listen_port, get_child_pids
 from mindsdb.interfaces.database.database import DatabaseWrapper
 from mindsdb.utilities.functions import args_parse, get_all_models_meta_data
 from mindsdb.utilities.log import log
@@ -26,6 +26,9 @@
     try:
         for api in apis.values():
             process = api['process']
+            childs = get_child_pids(process.pid)
+            for p in childs:
+                os.kill(p, signal.SIGTERM)
             sys.stdout.flush()
             process.terminate()
             process.join()
diff --git a/mindsdb/interfaces/native/learn_process.py b/mindsdb/interfaces/native/learn_process.py
--- a/mindsdb/interfaces/native/learn_process.py
+++ b/mindsdb/interfaces/native/learn_process.py
@@ -26,8 +26,8 @@
         '''
         import mindsdb_native
 
-        fs_store = FsSotre()
         config = Config()
+        fs_store = FsSotre()
         company_id = os.environ.get('MINDSDB_COMPANY_ID', None)
         name, from_data, to_predict, kwargs, datasource_id = self._args
 
diff --git a/mindsdb/utilities/ps.py b/mindsdb/utilities/ps.py
--- a/mindsdb/utilities/ps.py
+++ b/mindsdb/utilities/ps.py
@@ -4,6 +4,11 @@
 import psutil
 
 
+def get_child_pids(pid):
+    p = psutil.Process(pid=pid)
+    return p.children(recursive=True)
+
+
 def net_connections():
     """Cross-platform psutil.net_connections like interface"""
     if sys.platform.lower().startswith('linux'):

{"golden_diff": "diff --git a/mindsdb/__main__.py b/mindsdb/__main__.py\n--- a/mindsdb/__main__.py\n+++ b/mindsdb/__main__.py\n@@ -5,7 +5,7 @@\n import time\n import asyncio\n import datetime\n-import platform\n+import signal\n \n import torch.multiprocessing as mp\n \n@@ -16,7 +16,7 @@\n from mindsdb.api.http.start import start as start_http\n from mindsdb.api.mysql.start import start as start_mysql\n from mindsdb.api.mongo.start import start as start_mongo\n-from mindsdb.utilities.ps import is_pid_listen_port\n+from mindsdb.utilities.ps import is_pid_listen_port, get_child_pids\n from mindsdb.interfaces.database.database import DatabaseWrapper\n from mindsdb.utilities.functions import args_parse, get_all_models_meta_data\n from mindsdb.utilities.log import log\n@@ -26,6 +26,9 @@\n     try:\n         for api in apis.values():\n             process = api['process']\n+            childs = get_child_pids(process.pid)\n+            for p in childs:\n+                os.kill(p, signal.SIGTERM)\n             sys.stdout.flush()\n             process.terminate()\n             process.join()\ndiff --git a/mindsdb/interfaces/native/learn_process.py b/mindsdb/interfaces/native/learn_process.py\n--- a/mindsdb/interfaces/native/learn_process.py\n+++ b/mindsdb/interfaces/native/learn_process.py\n@@ -26,8 +26,8 @@\n         '''\n         import mindsdb_native\n \n-        fs_store = FsSotre()\n         config = Config()\n+        fs_store = FsSotre()\n         company_id = os.environ.get('MINDSDB_COMPANY_ID', None)\n         name, from_data, to_predict, kwargs, datasource_id = self._args\n \ndiff --git a/mindsdb/utilities/ps.py b/mindsdb/utilities/ps.py\n--- a/mindsdb/utilities/ps.py\n+++ b/mindsdb/utilities/ps.py\n@@ -4,6 +4,11 @@\n import psutil\n \n \n+def get_child_pids(pid):\n+    p = psutil.Process(pid=pid)\n+    return p.children(recursive=True)\n+\n+\n def net_connections():\n     \"\"\"Cross-platform psutil.net_connections like interface\"\"\"\n     if sys.platform.lower().startswith('linux'):\n", "issue": "Close learn process gracefuly\n`close_api_gracefully` only closes the API, but not any ongoing learn processes, which are left orphaned and consuming loads of memory.\n", "before_files": [{"content": "import os\nimport torch.multiprocessing as mp\n\nfrom mindsdb.interfaces.database.database import DatabaseWrapper\nfrom mindsdb.utilities.os_specific import get_mp_context\nfrom mindsdb.interfaces.storage.db import session, Predictor\nfrom mindsdb.interfaces.storage.fs import FsSotre\nfrom mindsdb.utilities.config import Config\n\n\nctx = mp.get_context('spawn')\n\n\nclass LearnProcess(ctx.Process):\n    daemon = True\n\n    def __init__(self, *args):\n        super(LearnProcess, self).__init__(args=args)\n\n    def run(self):\n        '''\n        running at subprocess due to\n        ValueError: signal only works in main thread\n\n        this is work for celery worker here?\n        '''\n        import mindsdb_native\n\n        fs_store = FsSotre()\n        config = Config()\n        company_id = os.environ.get('MINDSDB_COMPANY_ID', None)\n        name, from_data, to_predict, kwargs, datasource_id = self._args\n\n        mdb = mindsdb_native.Predictor(name=name, run_env={'trigger': 'mindsdb'})\n\n        predictor_record = Predictor.query.filter_by(company_id=company_id, name=name).first()\n        predictor_record.datasource_id = datasource_id\n        predictor_record.to_predict = to_predict\n        predictor_record.version = mindsdb_native.__version__\n        predictor_record.data = {\n            'name': name,\n            'status': 'training'\n        }\n        #predictor_record.datasource_id = ... <-- can be done once `learn` is passed a datasource name\n        session.commit()\n\n        to_predict = to_predict if isinstance(to_predict, list) else [to_predict]\n        data_source = getattr(mindsdb_native, from_data['class'])(*from_data['args'], **from_data['kwargs'])\n        try:\n            mdb.learn(\n                from_data=data_source,\n                to_predict=to_predict,\n                **kwargs\n            )\n        except Exception as e:\n            pass\n\n        fs_store.put(name, f'predictor_{company_id}_{predictor_record.id}', config['paths']['predictors'])\n\n        model_data = mindsdb_native.F.get_model_data(name)\n\n        predictor_record = Predictor.query.filter_by(company_id=company_id, name=name).first()\n        predictor_record.data = model_data\n        session.commit()\n\n        DatabaseWrapper().register_predictors([model_data])\n", "path": "mindsdb/interfaces/native/learn_process.py"}, {"content": "import sys\nimport time\nfrom collections import namedtuple\nimport psutil\n\n\ndef net_connections():\n    \"\"\"Cross-platform psutil.net_connections like interface\"\"\"\n    if sys.platform.lower().startswith('linux'):\n        return psutil.net_connections()\n\n    all_connections = []\n    Pconn = None\n    for p in psutil.process_iter(['pid']):\n        try:\n            process = psutil.Process(p.pid)\n            connections = process.connections()\n            if connections:\n                for conn in connections:\n                    # Adding pid to the returned instance\n                    # for consistency with psutil.net_connections()\n                    if Pconn is None:\n                        fields = list(conn._fields)\n                        fields.append('pid')\n                        _conn = namedtuple('Pconn', fields)\n                    for attr in conn._fields:\n                        setattr(_conn, attr, getattr(conn, attr))\n                    _conn.pid = p.pid\n                    all_connections.append(_conn)\n\n        except (psutil.AccessDenied, psutil.ZombieProcess, psutil.NoSuchProcess):\n            pass\n    return all_connections\n\n\ndef is_port_in_use(port_num):\n    \"\"\"Check does any of child process uses specified port.\"\"\"\n    parent_process = psutil.Process()\n    child_pids = [x.pid for x in parent_process.children(recursive=True)]\n    conns = net_connections()\n    portsinuse = [x.laddr[1] for x in conns if x.pid in child_pids and x.status == 'LISTEN']\n    portsinuse.sort()\n    return int(port_num) in portsinuse\n\n\ndef wait_func_is_true(func, timeout, *args, **kwargs):\n    start_time = time.time()\n\n    result = func(*args, **kwargs)\n    while result is False and (time.time() - start_time) < timeout:\n        time.sleep(2)\n        result = func(*args, **kwargs)\n\n    return result\n\n\ndef wait_port(port_num, timeout):\n    return wait_func_is_true(func=is_port_in_use, timeout=timeout, port_num=port_num)\n\n\ndef get_listen_ports(pid):\n    try:\n        p = psutil.Process(pid)\n        cons = p.connections()\n        cons = [x.laddr.port for x in cons]\n    except Exception:\n        return []\n    return cons\n\n\ndef is_pid_listen_port(pid, port):\n    ports = get_listen_ports(pid)\n    return int(port) in ports\n", "path": "mindsdb/utilities/ps.py"}, {"content": "import atexit\nimport traceback\nimport sys\nimport os\nimport time\nimport asyncio\nimport datetime\nimport platform\n\nimport torch.multiprocessing as mp\n\nfrom mindsdb.utilities.config import Config\nfrom mindsdb.utilities.os_specific import get_mp_context\nfrom mindsdb.interfaces.native.native import NativeInterface\nfrom mindsdb.interfaces.custom.custom_models import CustomModels\nfrom mindsdb.api.http.start import start as start_http\nfrom mindsdb.api.mysql.start import start as start_mysql\nfrom mindsdb.api.mongo.start import start as start_mongo\nfrom mindsdb.utilities.ps import is_pid_listen_port\nfrom mindsdb.interfaces.database.database import DatabaseWrapper\nfrom mindsdb.utilities.functions import args_parse, get_all_models_meta_data\nfrom mindsdb.utilities.log import log\n\n\ndef close_api_gracefully(apis):\n    try:\n        for api in apis.values():\n            process = api['process']\n            sys.stdout.flush()\n            process.terminate()\n            process.join()\n            sys.stdout.flush()\n    except KeyboardInterrupt:\n        sys.exit(0)\n\n\nif __name__ == '__main__':\n    mp.freeze_support()\n    args = args_parse()\n    config = Config()\n\n    if args.verbose is True:\n        config.set(['log', 'level', 'console'], 'DEBUG')\n\n    os.environ['DEFAULT_LOG_LEVEL'] = config['log']['level']['console']\n    os.environ['LIGHTWOOD_LOG_LEVEL'] = config['log']['level']['console']\n    config.set(['mindsdb_last_started_at'], str(datetime.datetime.now()))\n    \n    from lightwood.__about__ import __version__ as lightwood_version\n    from mindsdb_native.__about__ import __version__ as mindsdb_native_version\n    from mindsdb.__about__ import __version__ as mindsdb_version\n    print('Versions:')\n    print(f' - lightwood {lightwood_version}')\n    print(f' - MindsDB_native {mindsdb_native_version}')\n    print(f' - MindsDB {mindsdb_version}')\n\n    print(f'Configuration file:\\n   {config.config_path}')\n    print(f\"Storage path:\\n   {config.paths['root']}\")\n\n    if args.api is None:\n        api_arr = ['http', 'mysql']\n    else:\n        api_arr = args.api.split(',')\n\n    apis = {\n        api: {\n            'port': config['api'][api]['port'],\n            'process': None,\n            'started': False\n        } for api in api_arr\n    }\n\n    for api_name in apis.keys():\n        if api_name not in config['api']:\n            print(f\"Trying run '{api_name}' API, but is no config for this api.\")\n            print(f\"Please, fill config['api']['{api_name}']\")\n            sys.exit(0)\n\n    start_functions = {\n        'http': start_http,\n        'mysql': start_mysql,\n        'mongodb': start_mongo\n    }\n\n    mdb = NativeInterface()\n    cst = CustomModels()\n\n    model_data_arr = get_all_models_meta_data(mdb, cst)\n\n    dbw = DatabaseWrapper()\n    for db_alias in config['integrations']:\n        dbw.setup_integration(db_alias)\n    dbw.register_predictors(model_data_arr)\n\n    for broken_name in [name for name, connected in dbw.check_connections().items() if connected is False]:\n        log.error(f'Error failed to integrate with database aliased: {broken_name}')\n\n    ctx = mp.get_context('spawn')\n    # Switch to this once the native interface has it's own thread :/\n    # ctx = mp.get_context(get_mp_context())\n\n\n    for api_name, api_data in apis.items():\n        print(f'{api_name} API: starting...')\n        try:\n            if api_name == 'http':\n                p = ctx.Process(target=start_functions[api_name], args=(args.verbose,args.no_studio))\n            else:\n                p = ctx.Process(target=start_functions[api_name], args=(args.verbose,))\n            p.start()\n            api_data['process'] = p\n        except Exception as e:\n            log.error(f'Failed to start {api_name} API with exception {e}\\n{traceback.format_exc()}')\n            close_api_gracefully(apis)\n            raise e\n\n    atexit.register(close_api_gracefully, apis=apis)\n\n    async def wait_api_start(api_name, pid, port):\n        timeout = 60\n        start_time = time.time()\n        started = is_pid_listen_port(pid, port)\n        while (time.time() - start_time) < timeout and started is False:\n            await asyncio.sleep(0.5)\n            started = is_pid_listen_port(pid, port)\n        return api_name, port, started\n\n    async def wait_apis_start():\n        futures = [\n            wait_api_start(api_name, api_data['process'].pid, api_data['port'])\n            for api_name, api_data in apis.items()\n        ]\n        for i, future in enumerate(asyncio.as_completed(futures)):\n            api_name, port, started = await future\n            if started:\n                print(f\"{api_name} API: started on {port}\")\n            else:\n                log.error(f\"ERROR: {api_name} API cant start on {port}\")\n\n    ioloop = asyncio.get_event_loop()\n    ioloop.run_until_complete(wait_apis_start())\n    ioloop.close()\n\n    try:\n        for api_data in apis.values():\n            api_data['process'].join()\n    except KeyboardInterrupt:\n        print('Closing app...')\n", "path": "mindsdb/__main__.py"}], "after_files": [{"content": "import os\nimport torch.multiprocessing as mp\n\nfrom mindsdb.interfaces.database.database import DatabaseWrapper\nfrom mindsdb.utilities.os_specific import get_mp_context\nfrom mindsdb.interfaces.storage.db import session, Predictor\nfrom mindsdb.interfaces.storage.fs import FsSotre\nfrom mindsdb.utilities.config import Config\n\n\nctx = mp.get_context('spawn')\n\n\nclass LearnProcess(ctx.Process):\n    daemon = True\n\n    def __init__(self, *args):\n        super(LearnProcess, self).__init__(args=args)\n\n    def run(self):\n        '''\n        running at subprocess due to\n        ValueError: signal only works in main thread\n\n        this is work for celery worker here?\n        '''\n        import mindsdb_native\n\n        config = Config()\n        fs_store = FsSotre()\n        company_id = os.environ.get('MINDSDB_COMPANY_ID', None)\n        name, from_data, to_predict, kwargs, datasource_id = self._args\n\n        mdb = mindsdb_native.Predictor(name=name, run_env={'trigger': 'mindsdb'})\n\n        predictor_record = Predictor.query.filter_by(company_id=company_id, name=name).first()\n        predictor_record.datasource_id = datasource_id\n        predictor_record.to_predict = to_predict\n        predictor_record.version = mindsdb_native.__version__\n        predictor_record.data = {\n            'name': name,\n            'status': 'training'\n        }\n        #predictor_record.datasource_id = ... <-- can be done once `learn` is passed a datasource name\n        session.commit()\n\n        to_predict = to_predict if isinstance(to_predict, list) else [to_predict]\n        data_source = getattr(mindsdb_native, from_data['class'])(*from_data['args'], **from_data['kwargs'])\n        try:\n            mdb.learn(\n                from_data=data_source,\n                to_predict=to_predict,\n                **kwargs\n            )\n        except Exception as e:\n            pass\n\n        fs_store.put(name, f'predictor_{company_id}_{predictor_record.id}', config['paths']['predictors'])\n\n        model_data = mindsdb_native.F.get_model_data(name)\n\n        predictor_record = Predictor.query.filter_by(company_id=company_id, name=name).first()\n        predictor_record.data = model_data\n        session.commit()\n\n        DatabaseWrapper().register_predictors([model_data])\n", "path": "mindsdb/interfaces/native/learn_process.py"}, {"content": "import sys\nimport time\nfrom collections import namedtuple\nimport psutil\n\n\ndef get_child_pids(pid):\n    p = psutil.Process(pid=pid)\n    return p.children(recursive=True)\n\n\ndef net_connections():\n    \"\"\"Cross-platform psutil.net_connections like interface\"\"\"\n    if sys.platform.lower().startswith('linux'):\n        return psutil.net_connections()\n\n    all_connections = []\n    Pconn = None\n    for p in psutil.process_iter(['pid']):\n        try:\n            process = psutil.Process(p.pid)\n            connections = process.connections()\n            if connections:\n                for conn in connections:\n                    # Adding pid to the returned instance\n                    # for consistency with psutil.net_connections()\n                    if Pconn is None:\n                        fields = list(conn._fields)\n                        fields.append('pid')\n                        _conn = namedtuple('Pconn', fields)\n                    for attr in conn._fields:\n                        setattr(_conn, attr, getattr(conn, attr))\n                    _conn.pid = p.pid\n                    all_connections.append(_conn)\n\n        except (psutil.AccessDenied, psutil.ZombieProcess, psutil.NoSuchProcess):\n            pass\n    return all_connections\n\n\ndef is_port_in_use(port_num):\n    \"\"\"Check does any of child process uses specified port.\"\"\"\n    parent_process = psutil.Process()\n    child_pids = [x.pid for x in parent_process.children(recursive=True)]\n    conns = net_connections()\n    portsinuse = [x.laddr[1] for x in conns if x.pid in child_pids and x.status == 'LISTEN']\n    portsinuse.sort()\n    return int(port_num) in portsinuse\n\n\ndef wait_func_is_true(func, timeout, *args, **kwargs):\n    start_time = time.time()\n\n    result = func(*args, **kwargs)\n    while result is False and (time.time() - start_time) < timeout:\n        time.sleep(2)\n        result = func(*args, **kwargs)\n\n    return result\n\n\ndef wait_port(port_num, timeout):\n    return wait_func_is_true(func=is_port_in_use, timeout=timeout, port_num=port_num)\n\n\ndef get_listen_ports(pid):\n    try:\n        p = psutil.Process(pid)\n        cons = p.connections()\n        cons = [x.laddr.port for x in cons]\n    except Exception:\n        return []\n    return cons\n\n\ndef is_pid_listen_port(pid, port):\n    ports = get_listen_ports(pid)\n    return int(port) in ports\n", "path": "mindsdb/utilities/ps.py"}, {"content": "import atexit\nimport traceback\nimport sys\nimport os\nimport time\nimport asyncio\nimport datetime\nimport signal\n\nimport torch.multiprocessing as mp\n\nfrom mindsdb.utilities.config import Config\nfrom mindsdb.utilities.os_specific import get_mp_context\nfrom mindsdb.interfaces.native.native import NativeInterface\nfrom mindsdb.interfaces.custom.custom_models import CustomModels\nfrom mindsdb.api.http.start import start as start_http\nfrom mindsdb.api.mysql.start import start as start_mysql\nfrom mindsdb.api.mongo.start import start as start_mongo\nfrom mindsdb.utilities.ps import is_pid_listen_port, get_child_pids\nfrom mindsdb.interfaces.database.database import DatabaseWrapper\nfrom mindsdb.utilities.functions import args_parse, get_all_models_meta_data\nfrom mindsdb.utilities.log import log\n\n\ndef close_api_gracefully(apis):\n    try:\n        for api in apis.values():\n            process = api['process']\n            childs = get_child_pids(process.pid)\n            for p in childs:\n                os.kill(p, signal.SIGTERM)\n            sys.stdout.flush()\n            process.terminate()\n            process.join()\n            sys.stdout.flush()\n    except KeyboardInterrupt:\n        sys.exit(0)\n\n\nif __name__ == '__main__':\n    mp.freeze_support()\n    args = args_parse()\n    config = Config()\n\n    if args.verbose is True:\n        config.set(['log', 'level', 'console'], 'DEBUG')\n\n    os.environ['DEFAULT_LOG_LEVEL'] = config['log']['level']['console']\n    os.environ['LIGHTWOOD_LOG_LEVEL'] = config['log']['level']['console']\n    config.set(['mindsdb_last_started_at'], str(datetime.datetime.now()))\n    \n    from lightwood.__about__ import __version__ as lightwood_version\n    from mindsdb_native.__about__ import __version__ as mindsdb_native_version\n    from mindsdb.__about__ import __version__ as mindsdb_version\n    print('Versions:')\n    print(f' - lightwood {lightwood_version}')\n    print(f' - MindsDB_native {mindsdb_native_version}')\n    print(f' - MindsDB {mindsdb_version}')\n\n    print(f'Configuration file:\\n   {config.config_path}')\n    print(f\"Storage path:\\n   {config.paths['root']}\")\n\n    if args.api is None:\n        api_arr = ['http', 'mysql']\n    else:\n        api_arr = args.api.split(',')\n\n    apis = {\n        api: {\n            'port': config['api'][api]['port'],\n            'process': None,\n            'started': False\n        } for api in api_arr\n    }\n\n    for api_name in apis.keys():\n        if api_name not in config['api']:\n            print(f\"Trying run '{api_name}' API, but is no config for this api.\")\n            print(f\"Please, fill config['api']['{api_name}']\")\n            sys.exit(0)\n\n    start_functions = {\n        'http': start_http,\n        'mysql': start_mysql,\n        'mongodb': start_mongo\n    }\n\n    mdb = NativeInterface()\n    cst = CustomModels()\n\n    model_data_arr = get_all_models_meta_data(mdb, cst)\n\n    dbw = DatabaseWrapper()\n    for db_alias in config['integrations']:\n        dbw.setup_integration(db_alias)\n    dbw.register_predictors(model_data_arr)\n\n    for broken_name in [name for name, connected in dbw.check_connections().items() if connected is False]:\n        log.error(f'Error failed to integrate with database aliased: {broken_name}')\n\n    ctx = mp.get_context('spawn')\n    # Switch to this once the native interface has it's own thread :/\n    # ctx = mp.get_context(get_mp_context())\n\n\n    for api_name, api_data in apis.items():\n        print(f'{api_name} API: starting...')\n        try:\n            if api_name == 'http':\n                p = ctx.Process(target=start_functions[api_name], args=(args.verbose,args.no_studio))\n            else:\n                p = ctx.Process(target=start_functions[api_name], args=(args.verbose,))\n            p.start()\n            api_data['process'] = p\n        except Exception as e:\n            log.error(f'Failed to start {api_name} API with exception {e}\\n{traceback.format_exc()}')\n            close_api_gracefully(apis)\n            raise e\n\n    atexit.register(close_api_gracefully, apis=apis)\n\n    async def wait_api_start(api_name, pid, port):\n        timeout = 60\n        start_time = time.time()\n        started = is_pid_listen_port(pid, port)\n        while (time.time() - start_time) < timeout and started is False:\n            await asyncio.sleep(0.5)\n            started = is_pid_listen_port(pid, port)\n        return api_name, port, started\n\n    async def wait_apis_start():\n        futures = [\n            wait_api_start(api_name, api_data['process'].pid, api_data['port'])\n            for api_name, api_data in apis.items()\n        ]\n        for i, future in enumerate(asyncio.as_completed(futures)):\n            api_name, port, started = await future\n            if started:\n                print(f\"{api_name} API: started on {port}\")\n            else:\n                log.error(f\"ERROR: {api_name} API cant start on {port}\")\n\n    ioloop = asyncio.get_event_loop()\n    ioloop.run_until_complete(wait_apis_start())\n    ioloop.close()\n\n    try:\n        for api_data in apis.values():\n            api_data['process'].join()\n    except KeyboardInterrupt:\n        print('Closing app...')\n", "path": "mindsdb/__main__.py"}]}