Process.start() method does not call Process._notify() about pymesos HOT 7 CLOSED

import logging
import os
import signal
import socket
import sys
import threading
import typing
import time
import uuid

import pymesos

logging.basicConfig(level=logging.DEBUG)

TASK_CPU = 4
TASK_MEM = 32
EXECUTOR_CPUS = 0.1
EXECUTOR_MEM = 32


class TestScheduler(pymesos.Scheduler):

    def __init__(self, executor: typing.Dict):
        self._executor: typing.Dict = executor

    def registered(self, driver, frameworkId, masterInfo):
        logging.debug(
            f'Registered (framework: {frameworkId}, master info: {masterInfo}).')

    def resourceOffers(self, driver, offers):

        filters = {'refuse_seconds': 5}

        for offer in offers:
            cpus = self.getResource(offer['resources'], 'cpus')
            mem = self.getResource(offer['resources'], 'mem')
            if cpus < TASK_CPU or mem < TASK_MEM:
                continue

            task_id = str(uuid.uuid4())
            task = {
                'task_id': {
                    'value': task_id
                },
                'agent_id': {
                    'value': offer['agent_id']['value']
                },
                'name': f'task {task_id}',
                'executor': self._executor,
                'data': pymesos.encode_data(f'Hello from task {task_id}!'.encode())
            }

            task['resources'] = [
                {
                    'name': 'cpus',
                    'type': 'SCALAR',
                    'scalar': {'value': TASK_CPU}
                },
                {
                    'name': 'mem',
                    'type': 'SCALAR',
                    'scalar': {'value': TASK_MEM}
                }
            ]

            driver.launchTasks(offer['id'], [task], filters)

    def getResource(self, res, name):
        resource = list(filter(lambda x: x['name'] == name, res))[0]
        return resource.get('scalar', {'value': 0.0}).get('value', 0.0)

    def statusUpdate(self, driver, update):
        logging.debug(
            f'Status update TID {update["task_id"]["value"]} {update["state"]}')


def main(master):

    executor = {
        'executor_id': {
            'value': str(uuid.uuid4())
        },
        'name': 'TestExecutor',
        'command': {
            'value': f'python3 /opt/python/executor.py'
        },
        'resources': [
            {
                'name': 'mem',
                'type': 'SCALAR',
                'scalar': {
                    'value': EXECUTOR_MEM
                }
            },
            {
                'name': 'cpus',
                'type': 'SCALAR',
                'scalar': {
                    'value': EXECUTOR_CPUS
                }
            }
        ]
    }

    driver = pymesos.MesosSchedulerDriver(
        sched=TestScheduler(executor=executor),
        framework={
            'user': 'test_user',
            'name': 'test_name'
        },
        master_uri=master
    )

    def signal_handler(signal, frame):
        driver.stop()

    def run_driver_thread():
        driver.run()

    driver_thread = threading.Thread(target=run_driver_thread, args=())
    driver_thread.start()

    logging.info('Scheduler running, Ctrl+C to quit.')
    signal.signal(signal.SIGINT, signal_handler)

    while driver_thread.is_alive():
        time.sleep(1)


if __name__ == '__main__':
    main(master='127.0.0.1:5050')

import logging
import sys
import time
import threading
import typing

import pymesos

LOG = logging.getLogger(__name__)
LOG.setLevel(logging.DEBUG)
LOG.addHandler(logging.FileHandler('task.log'))


class TestExecutor(pymesos.Executor):
    def registered(self, driver, executorInfo, frameworkInfo, slaveInfo):
        LOG.debug(
            f'Executor regitered (driver: {driver}, executorInfo: {executorInfo}, frameworkInfo: {frameworkInfo}, slaveInfo:  {slaveInfo}).')

    def launchTask(self, driver, task):
        def run_task(task: typing.Dict):
            update = {
                'task_id': {
                    'value': task['task_id']['value'],
                },
                'state': 'TASK_RUNNING',
                'timestmp': time.time(),
            }

            driver.sendStatusUpdate(update)

            LOG.info(pymesos.decode_data(task['data']))
            time.sleep(30)

            update = {
                'task_id': {
                    'value': task['task_id']['value'],
                },
                'state': 'TASK_FINISHED',
                'timestmp': time.time(),
            }

            driver.sendStatusUpdate(update)

        thread = threading.Thread(target=run_task, args=(task,))
        thread.start()


if __name__ == '__main__':

    driver = pymesos.MesosExecutorDriver(TestExecutor())
    driver.run()

class TestMesosExecutorDriver(pymesos.MesosExecutorDriver):
    
    def start(self):
        super().start()
        self._notify()

if __name__ == '__main__':

    driver = TestMesosExecutorDriver(TestExecutor())
    driver.run()