docker for server

Author: lightforever

.dockerignore

@@ -3,4 +3,4 @@ dist
 *.egg-ingo
 examples
 .git
-mlcomp/server
+mlcomp/server/front/node_modules

.gitignore

@@ -110,4 +110,5 @@ Pipfile
 mlcomp/server/front/node_modules
 examples/**/data
 examples/**/log
-dist
+dist
+docker/*.conf

docker/Readme.md

@@ -2,7 +2,17 @@
 
 Worker:
 
+docker build -f docker/worker-dev -t mlcomp-worker .
 
-docker run --net=host -it mlcomp-worker /bin/bash 
+docker run --net=host -v /opt/mlcomp/:/opt/mlcomp -it mlcomp-worker /bin/bash 
 
-PYTHONPATH=../ python __main__.py worker 0
+PYTHONPATH=../ python __main__.py worker 0
+
+
+Server:
+
+docker build -f docker/server-dev -t mlcomp-server .
+
+docker run --net=host -p 4201:4201 -it mlcomp-server /bin/bash 
+
+PYTHONPATH=../ python __main__.py start-server

docker/server-dev

@@ -1,19 +1,31 @@
-FROM ubuntu:16.04
+FROM python:3.6
 
-RUN export DEBIAN_FRONTEND=noninteractive \
-    && apt-get update -qq && apt-get upgrade -qq \
-    && apt-get install -y --no-install-recommends \
-        python3 \
-        python3-pip \
-        python3-setuptools \
-        supervisor \
-    && BUILD_DEPS='build-essential python3-dev git' \
-    && apt-get install -y --no-install-recommends ${BUILD_DEPS}
+RUN pip install --no-cache-dir \
+    numpy>=1.16.3 \
+    click>=7.0 \
+    typing \
+    apscheduler>=3.6.0 \
+    sqlalchemy>=1.3.4 \
+    celery>=4.3.0 \
+    flask>=1.0.2 \
+    requests \
+    flask_cors>=3.0.6 \
+    sqlalchemy_serializer==1.1.1 \
+    psycopg2-binary>=2.8.2 \
+    matplotlib \
+    scikit-learn>=0.21.2 \
+    Pillow \
+    PyYAML \
+    pathspec>=0.5.9 \
+    psutil>=5.6.2 \
+    GPUtil==1.4.0
 
 # Set the locale
 ENV LANG C.UTF-8
 ENV LC_ALL C.UTF-8
 
 # Copy source files
 COPY mlcomp /app/mlcomp
-WORKDIR /app/mlcomp
+WORKDIR /app/mlcomp
+
+CMD ["python", "__main__.py", "start-server"]

docker/worker-dev

@@ -17,13 +17,14 @@ RUN apt-get update && apt-get install -y \
     libpq-dev \
 	libturbojpeg \
 	software-properties-common \
+	supervisor \
 	&& apt-get clean \
 	&& rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/*
 
-RUN pip install setuptools>=41.0.1 \
-    torchvision>=0.3.0 \
+RUN pip install --no-cache-dir  \
+    setuptools>=41.0.1 \
+    torchvision>=0.2.0 \
     numpy>=1.16.3 \
-    torch==1.1.0 \
     click>=7.0 \
     psutil>=5.6.2 \
     GPUtil==1.4.0 \
@@ -52,4 +53,10 @@ ENV LC_ALL C.UTF-8
 # Copy source files
 COPY mlcomp /app/mlcomp
 WORKDIR /app/mlcomp
+COPY docker/worker.py /app/mlcomp/worker.py
+COPY docker/worker.sh /app/mlcomp/worker.sh
+
+
+RUN chmod 777 worker.sh
+CMD [",/worker.sh"]
 

docker/worker.py

@@ -0,0 +1,18 @@
+import os
+from multiprocessing import cpu_count
+
+pr = os.getenv('CPU', cpu_count())
+text = [
+    '[supervisord]',
+    'nodaemon=true',
+    ''
+]
+for p in range(pr):
+    text.append(f'[program:worker{p}]')
+    text.append(f'command=python __main__.py worker {p}')
+    text.append('autostart=true')
+    text.append('autorestart=true')
+    text.append('')
+
+with open('supervisord.conf', 'w') as f:
+    f.writelines('\n'.join(text))

docker/worker.sh

@@ -0,0 +1,4 @@
+#!/usr/bin/env bash
+
+python worker.py
+PYTHONPATH=../ supervisord -c supervisord.conf

examples/catalyst/config.yml

@@ -2,7 +2,7 @@ info:
   name: catalyst
   project: first
   folder: examples/catalyst
-  data_folder: /f/data/first
+  data_folder: catalyst
 executors:
   dummy1:
     type: Catalyst

examples/catalyst/experiment.py

@@ -38,15 +38,17 @@ def get_datasets(self, stage: str, **kwargs):
             transform=Experiment.get_transforms(stage=stage, mode="valid")
         )
 
-        # while True:
-        #     pass
-        #raise Exception('assd')
+        while True:
+            continue
 
-        trainset.data = trainset.data[:32]
-        trainset.targets = np.clip(trainset.targets[:32], 0, 1)
+        trainset.train_data = trainset.train_data[:32]
+        trainset.train_labels = np.clip(trainset.train_labels[:32], 0, 1)
 
-        testset.data = trainset.data[:32]
-        testset.targets = np.clip(trainset.targets[:32], 0, 1)
+        testset.train_data = trainset.train_data[:32]
+        testset.train_labels = np.clip(trainset.train_labels[:32], 0, 1)
+
+        testset.test_data = testset.test_data[:32]
+        testset.test_labels = np.clip(testset.test_labels[:32], 0, 1)
 
         datasets["train"] = trainset
         datasets["valid"] = testset

examples/catalyst/model.py

@@ -21,4 +21,4 @@ def forward(self, x):
         x = F.relu(self.fc1(x))
         x = F.relu(self.fc2(x))
         x = self.fc3(x)
-        return x
+        return x

mlcomp/__main__.py

@@ -3,17 +3,16 @@
 import os
 from mlcomp.task.storage import Storage
 from mlcomp.utils.config import load_ordered_yaml
-from mlcomp.task.executors import Executor
 from mlcomp.task.app import app
 import socket
-from multiprocessing import cpu_count, Process
-import torch
-
+from multiprocessing import cpu_count
 from mlcomp.utils.misc import dict_func
 import psutil
 import GPUtil
 import numpy as np
 from mlcomp.task.tasks import execute_by_id
+from mlcomp.utils.schedule import start_schedule
+
 
 @click.group()
 def main():
@@ -46,28 +45,43 @@ def worker_usage():
 @main.command()
 @click.argument('number', type=int)
 def worker(number):
+    docker_img = os.getenv('DOCKER_IMG', 'default')
+    argv = [
+        'worker',
+        '--loglevel=INFO',
+        '-P=solo',
+        f'-n={number}',
+        '-O fair',
+        '-c=1',
+        '--prefetch-multiplier=1',
+        '-Q',
+        f'{socket.gethostname()}_{docker_img}'
+    ]
+    app.worker_main(argv)
+
+@main.command()
+def worker_supervisor():
     provider = ComputerProvider()
     tot_m, used_m, free_m = map(int, os.popen('free -t -m').readlines()[-1].split()[1:])
-
-    computer = Computer(name=socket.gethostname(), gpu=torch.cuda.device_count(), cpu=cpu_count(), memory=tot_m)
+    computer = Computer(name=socket.gethostname(), gpu=len(GPUtil.getGPUs()), cpu=cpu_count(), memory=tot_m)
     provider.create_or_update(computer, 'name')
 
-    # start_schedule([(worker_usage, 60)])
+    start_schedule([(worker_usage, 60)])
 
+    docker_img = os.getenv('DOCKER_IMG', 'default')
     argv = [
         'worker',
         '--loglevel=INFO',
         '-P=solo',
-        f'-n={number}',
+        f'-n=1',
         '-O fair',
         '-c=1',
         '--prefetch-multiplier=1',
         '-Q',
-        socket.gethostname()
+        f'{socket.gethostname()}_{docker_img}_supervisor'
     ]
     app.worker_main(argv)
 
-
 @main.command()
 def start_server():
     from mlcomp.server.back.app import start_server as _start_server
@@ -101,7 +115,8 @@ def _dag(config: str, debug: bool = False):
 
     folder = os.path.join(os.getcwd(), info['folder'])
     project = ProjectProvider().by_name(info['project']).id
-    dag = dag_provider.add(Dag(config=config_text, project=project, name=info['name']))
+    dag = dag_provider.add(Dag(config=config_text, project=project,
+                               name=info['name'], docker_img=info.get('docker_img')))
     storage.upload(folder, dag)
 
     created = OrderedDict()
@@ -112,8 +127,6 @@ def _dag(config: str, debug: bool = False):
                 for d in v['depends']:
                     if d not in executors:
                         raise Exception(f'Executor {k} depend on {d} which does not exist')
-                    if not Executor.is_registered(executors[d]['type']):
-                        raise Exception(f'Executor {d} has not been registered')
 
                     valid = valid and d in created
             if valid:
@@ -166,8 +179,6 @@ def execute(config: str):
                 for d in v['depends']:
                     if d not in executors:
                         raise Exception(f'Executor {k} depend on {d} which does not exist')
-                    if not Executor.is_registered(executors[d]['type']):
-                        raise Exception(f'Executor {d} has not been registered')
 
                     valid = valid and d in created
             if valid:

mlcomp/db/models/dag.py

@@ -11,4 +11,4 @@ class Dag(Base):
     name = sa.Column(sa.String)
     tasks = relationship('Task', lazy='noload')
     project_rel = relationship('Project', lazy='noload')
-
+    docker_img = sa.Column(sa.String)

mlcomp/db/models/task.py

@@ -24,6 +24,7 @@ class Task(Base):
     dag_rel = relationship('Dag', lazy='noload')
     debug = sa.Column(sa.Boolean, default=False)
     pid = sa.Column(sa.Integer)
+    replay = sa.Column(sa.Boolean)
 
 class TaskDependence(Base):
     __tablename__ = 'task_dependencies'

mlcomp/db/providers/base.py

@@ -52,6 +52,7 @@ def create_or_update(self, obj: Base, field: str):
 
     def update(self):
         self.session.update()
+        self.session.commit()
 
     @property
     def session(self):

mlcomp/db/providers/task.py

@@ -49,7 +49,7 @@ def change_status(self, task, status: TaskStatus):
             task.finished = now()
 
         task.status = status.value
-        self.session.update()
+        self.update()
 
     def by_status(self, status: TaskStatus):
         return self.query(Task).filter(Task.status == status.value).options(joinedload(Task.dag_rel)).all()

mlcomp/server/back/app.py

@@ -166,7 +166,8 @@ def tasks():
 @requires_auth
 def task_stop():
     data = request_data()
-    status = celery_tasks.stop(data['id'])
+    task = TaskProvider().by_id(data['id'], joinedload(Task.dag_rel))
+    status = celery_tasks.stop(task)
     return json.dumps({'success': True, 'status': to_snake(TaskStatus(status).name)})
 
 
@@ -178,7 +179,8 @@ def dag_stop():
     id = int(data['id'])
     dag = provider.by_id(id, joined_load=['tasks'])
     for t in dag.tasks:
-        celery_tasks.stop(t.id)
+        t.dag_rel = dag
+        celery_tasks.stop(t)
     return json.dumps({'success': True, 'dag': provider.get({'id': id})['data'][0]})
 
 
@@ -297,6 +299,7 @@ def all_exception_handler(error):
     if type(error) == ProgrammingError:
         Session.cleanup()
 
+    logger.error(error)
     return str(error), 500
 
 

mlcomp/server/back/supervisor.py

@@ -1,7 +1,7 @@
 import time
 from mlcomp.utils.logging import logger
 import traceback
-from mlcomp.task.tasks import execute
+from mlcomp.task.tasks import execute, queue_list
 from mlcomp.db.providers import *
 from mlcomp.utils.schedule import start_schedule
 
@@ -11,7 +11,10 @@ def supervisor():
     computer_provider = ComputerProvider()
 
     try:
-        time.sleep(1)
+        queues = queue_list()
+        if len(queues)==0:
+            return
+
         not_ran_tasks = provider.by_status(TaskStatus.NotRan)
         not_ran_tasks = [task for task in not_ran_tasks if not task.debug]
         logger.info(f'Found {len(not_ran_tasks)} not ran tasks')
@@ -39,12 +42,16 @@ def supervisor():
                 if task.computer is not None and task.computer != computer.name:
                     continue
 
-                r = execute.apply_async((task.id,), queue=computer['name'])
+                queue = f'{computer["name"]}_{task.dag_rel.docker_img or "default"}'
+                if queue not in queues:
+                    continue
+
+                r = execute.apply_async((task.id,), queue=queue)
                 task.status = TaskStatus.Queued.value
                 task.computer_assigned = computer['name']
                 task.celery_id = r.id
 
-                provider.session.update()
+                provider.update()
 
                 computer['gpu'] -= task.gpu
                 computer['cpu'] -= task.cpu

mlcomp/task/app.py

@@ -1,8 +1,8 @@
 from __future__ import absolute_import, unicode_literals
 from celery import Celery
 
-app = Celery('task',
+app = Celery('mlcomp',
              broker='amqp://',
              backend='amqp://',
              include=['mlcomp.task.tasks']
-             )
+             )