Подпроцессы. Мультизадачность.

Мультизадачность может быть обеспечена 3 техниками.

1. Многопоточность.

2. Многопроцессовость.

3. Асинхронные вызовы.

Каждыйте раз, когда вы запускаете программу Python из командной строки, например:

python myprogramm.py

Операционная система создает новый процесс, в котором происходит работа вашей программы.

Любая программа оперирует тремя потоками: поток ввода (input), поток вывода (output) и поток ошибок (error).

Обычно input - клавиатура output и error - окно терминала.

Иногда, в программе требуется запустить другую программу так же как из командной строки, предоставив ему отдельный подпроцесс, работающий независимо от основной программы.

Для этого можно использовать встроенный модуль subprocess

Сигнатура его вызова следующая:

subprocess.call(args, *, stdin=None, stdout=None, stderr=None, shell=False)

Как видим, при вызове мы можем определить поведение с 3-мя потоками.

Первым параметром мы передаем команду, которую хотим выполнить.

Это может быть список строк, из которой будет сформирована команда. Обычно первым указывают исполняемую программу, а затем аргументы.

Последним параметром мы указываем будет ли наша команда выполнятся в контексте коммандного интерпретатора shell.

Если он установле в True, то Python перед запуском вашей команды запустит интерпретатор shell и в нем уже выполнит команду.

При таком подходе вам станут доступными все команды shell и переменные окружения, что делает запуск не совсем безопасным.

Примеры вызова:

from subprocess import call

subprocess.call(['ls','l'],shell=True)

subprocess.call(['pwd'],shell=True)

Пример уязвимости:

filename = input("What file would you like to display?\n")
What file would you like to display?
on_existent; rm -rf / 
call("cat " + filename, shell=True) # Это может очень плохо закончиться!

Пример перехвата вывода и сохранения в файл.

from subprocess import call
file = open('output.txt','w')
call('ls',stdout=file,shell=True)
print 'done!'

subprocess.check_call(args, *, stdin=None, stdout=None, stderr=None, shell=False, timeout=None)

Выполняет команду, описанную args. Ожидает завершения команды, а затем завершается, если код возврата 0, или поднимает исключение CalledProcessError, объект которого возвращает код завершения атрибутом returncode.

subprocess.check_output(args, *, input=None, stdin=None, stderr=None, shell=False, universal_newlines=False, timeout=None)

Позволяет забрать вывод команды в строку.

c = subprocess.check_output(['pwd'],shell=True)

Поднимает исключение CalledProcessError, если код возврата ненулевой.

Библиотека subprocess оперирует системными вызовами os.fork() и os.execve() и это значит то, что каждый процесс получит свое адресное пространство и копии всех данных родителя.

Поэтому чтобы обеспечить передачу данных между процессами необходимо использовать дополнительные механизмы, такие как сокеты или каналы.

Popen класс.

Служит для создания процесса и управления его потоком выполнения.

>>>p = args = ['ls']
>>> p = subprocess.Popen(args)
>>> chdir_ex.py  os_ex.py  out_in_file.py  output.txt
p
<subprocess.Popen object at 0x7f5248d8f290>

С помощью этого конструктора мы тоже запускаем команду в новом процессе, не блокируя выполнение вызывающей программы.

Нам возвращается объект и мы можем взаимодействовать с этим объектом в момент выполнения в нем нашей команды.

В отличие от call, Popen не блокирует выполнение вызывающей программы и она продолжает свое выполнение.

Такая работа, при которой одна программа порождает другую не прерываясь, называется асинхронной.

Методы класса Popen.

Popen.poll() - если процесс завершил работу - вернёт код возврата, в ином случае None.

Popen.wait(timeout=None) - ожидает завершения работы процесса и возвращает код возврата. Если в течение timeout процесс не завершился, поднимется исключение TimeoutExpired (которое можно перехватить, после чего сделать ещё раз wait).

Popen.communicate(input=None, timeout=None) - взаимодействовует с процессом: посылает данные, содержащиеся в input в stdin процесса, ожидает завершения работы процесса, возвращает кортеж данных потока вывода и ошибок. При этом в Popen необходимо задать значение PIPE для stdin (если вы хотите посылать в stdin), stdout, stderr (если вы хотите прочитать вывод дочернего процесса).

Если в течение timeout процесс не завершился, поднимется исключение TimeoutExpired (которое можно перехватить, после чего сделать ещё раз communicate, либо убить дочерний процесс).

Popen.send_signal(signal) - посылает сигнал signal.

Popen.terminate() - останавливает дочерний процесс.

Popen.kill() - убивает дочерний процесс.

Multiprocessing

Эта библиотека служит для запуска функций внутри вашей программы.

При этом становится возможным их запуск не внутри отдельного процесса, а внутри потока. Что делает возможным обмен данными между потоками т.к. они запускаются в едином адресном пространстве процесса.

Асинхронное программирование с использованием asyncio.

(Распаралеливание)[http://toly.github.io/blog/2014/02/13/parallelism-in-one-line/]

Допустим у нас есть следующий словарь.

store = {
    'place1' : [],
    'place2' : []
}

Заполним список по ключу place1 100 словарями.

store = {
    'place1' : [],
    'place2' : []
}

for i in range(100):
    store['place1'].append({'name':  'brick %s' %i})

print(store)

Вывод

{‘place2’: [], ‘place1’: [{‘name’: ‘brick 0’}, {‘name’: ‘brick 1’}, {‘name’: ‘brick 2’},…

Теперь стоит задача перенести все объекты кирпичей из place1 в place2.

Можно определить функцию для этого и 100 раз ее вызвать.

store = {
    'place1' : [],
    'place2' : []
}

for i in range(100):
    store['place1'].append({'name':  'brick %s' %i})

def move(store):
    brick = store['place1'].pop(0)
    store['place2'].append(brick)

for i in range(100):
    move(store)

print(store)

При этом мы в функции выталкиваем первый кирпич из одного списка.

brick = store['place1'].pop(0)

и далее заталкиваем во второй.

store['place2'].append(brick)

Предположим, что этот процесс у нас занимает много времени, добавим в него задержку в 1 сек.

import time
def move():
    time.sleep(1)
    brick = store['place1'].pop(0)
    print('Moving %s' % brick['name'])
    store['place2'].append(brick)


shell>python3 ex.py 
Moving brick 0
Moving brick 1
Moving brick 2

Получается, что весь процесс у нас займет 100 сек.

Как можно его ускорить?

Для этого нужно запустить эту функцию в несколько потоков, или асинхронно.

Для реализации асинхронного запуска необходимо иметь цикл событий.

В этом цикле система будет контролировать выполнение одновременно-выполняющихся задач, предоставляя механизм переключения между ними.

Таким образом после выполнения своей работы, функция будет передавать управления циклу событий, который в свою очередь, будет запускать другие асинхронные задачи.

Библиотека Asyncio позволяем разбивать ваш код на процедуры (функции), которые определять как корутины, что даёт возможность управлять ими как пожелаете, включая и одновременное выполнение.

За переключение контекста в asyncio отвечает await, который передаёт управление обратно в event loop (цикл событий), а тот в свою очередь — к другой корутине.

Выражение после await должно представлять специальный awaitable объект.

Существует 3 варианта таких объектов.

1. Другая сопрограмма.

2. Сопрограмма на основе генератора.

3. Специальный объект, у которого реализован магический метод await, возвращающий итератор.

Мы можем воспользоваться функцией asyncio.sleep(1), которая возвратит awaitable объект, эмулируя задержку.

Добавим функцию цикл от 0 до 50 и обьявим ее как корутину с помощью async.

async def move(store):
    for i in range(50):
        brick = store['place1'].pop(0)
        print('Moving %s' % brick['name'])
        store['place2'].append(brick)
        await asyncio.sleep(1)

Теперь необходимо создать список из 2-х таких функций.

tasks = [
    ioloop.create_task(move(store)),
    ioloop.create_task(move(store))
]

Корутины могут быть запущены только из другой корутины, или обёрнуты в задачу с помощью create_task. Мы воспользовались ioloop.create_task() для того, обёрнуть корутину в задачу.

После того, как у нас оказались 2 задачи, объединим их, используя wait

asyncio.wait(tasks)

И, наконец, отправим на выполнение в цикл событий через run_until_complete

ioloop.run_until_complete(asyncio.wait(tasks))

Полный код примера.

store = {
    'place1' : [],
    'place2' : []
}
import asyncio

for i in range(100):
    store['place1'].append({'name':  'brick %s' %i})

async def move(store):
    for i in range(50):
        brick = store['place1'].pop(0)
        print('Moving %s' % brick['name'])
        store['place2'].append(brick)
        await asyncio.sleep(1)


ioloop = asyncio.get_event_loop()

tasks = [
    ioloop.create_task(move(store)),
    ioloop.create_task(move(store))
]

ioloop.run_until_complete(asyncio.wait(tasks))
ioloop.close()

print(store)

Мы можем вызывать одни корутины из других, связывая их в цепочки.

Например, если мы хотим логировать процесс в файл в другой корутине.

async def logger(brick):
    with open('log.txt', 'a') as f:
        f.write('Log: moving %s' % brick['name'])
    await asyncio.sleep(1)


async def move(store):
    for i in range(50):
        brick = store['place1'].pop(0)
        print('Moving %s' % brick['name'])
        store['place2'].append(brick)
        await logger(brick)

Или создать класс с методом await, который работает как генератор с функцией yield.

class Loger(object):
    def __init__(self, brick):
        self.brick = brick
    def __await__(self):
        with open('log.txt', 'a') as f:
            f.write('Log: moving %s' % self.brick['name'])
        yield


for i in range(100):
    store['place1'].append({'name':  'brick %s' %i})

import time
async def move(store):
    for i in range(50):
        brick = store['place1'].pop(0)
        print('Moving %s' % brick['name'])
        store['place2'].append(brick)
        await Loger(brick)

Создание репозитория GIT на сервере.

Для создания репозитория необходимо залогиниться на сервере по ssh.

Установить git.

apt install git

Затем создадим нового пользователя.

adduser git

Зайдем под этим пользователем.

su git

Перейдем в его домашнюю директорию.

cd

Создадим папку.

mkdir repo.git

Зайдем внутрь папки.

cd repo.git

Запустим создание репозитория с флагом –bare что будет означать что наш репозиторий не будет содержать исходников а только файлы для контроля версий. git init –bare

Теперь локально можно создать репозиторий.

git init

Добавить удаленный репозиторий в него.

git remote add origin ssh://git@domainname/home/git/repo.git

И залить файлы.

git add --all
git commit -m 'init'
git push --set-upstream origin master

Ответить на вопрос о добавлении отпечатка сервера.

The authenticity of host 'dima.webmonstr.com (188.120.241.104)' can't be established.
ECDSA key fingerprint is SHA256:fQQiO1wiRMkF9jsH7Qk3Qhi1Z1hA1MnYbp6bm+ZHPRs.
Are you sure you want to continue connecting (yes/no)?

Клонировать репозиторий можно командой.

git clone ssh://git@domainname/home/git/repo.git

Оператор with - менеджер контекста.

Рассмотрим такой пример кода.

f = open(filename)
try:
    do something
finally:
    f.close()

В начале мы открываем файл для чтения и в конструкции try совершаем с ним операции.

Для того, чтоб закрыть файл и освободить ресурс не зависимо от того, возникнет ли исключение или нет мы закрываем файл блоке finally.

Однако, если мы производим это много раз то такая конструкция может стать утомительной.

Более удобным подходом будет создание библиотечной функции, которая открывала и закрывала файл.

Для упрощения записи используется оператор with.

def controlled_execution(callback):
    f = open(filename)
    try:
        callback(thing)
    finally:
        f.close()

def my_function(thing):
    do something

controlled_execution(my_function)

Вместо функции, удобней использоваьть объект.

class СontrolledExecution:
    def __enter__(self):
        f = open(filename)
        return f
    def __exit__(self, type, value, traceback):
        tear f.close()

with СontrolledExecution() as thing:
     some code with thing

Оператор with задействует внутренний метод enter класса СontrolledExecution, который называется “хранителем контекста”.

Все что он возвращает будет присвоено переменной после ключевого слова as внутри блока.

Затем Python выполнит все что находится в блоке with и не зависимо от того что произойдет при выполнении в конце задействует метод exit. если исключения не было ему передается None.

Как дополнительный бонус, метод exit может перехватить исключение и информацию об ошибке.

def __exit__(self, type, value, traceback):
    return isinstance(value, TypeError)

Функция open поддерживает менеджер контекста и можно открыть файл гораздо проще.

with open("x.txt") as f:
    data = f.read()
    do something with data

В этом примере файл будет закрыт в независимости от результата внутри with.

Существует возможноть использовать сразу несколько операций.

with A() as a, B() as b:
    suite

Что будет эквивалентно записи.

with A() as a:
    with B() as b:
        suite

Домашнее задание. Асинхронное программирование.

Есть такой словарь с url адресами и таймаутами.

db = [ {‘url’: ‘http://google.com’, ‘timeout’: 3 }, {‘url’: ‘http://webmonstr.com’, ‘timeout’: 2 }, … ]

Необходимо создать консольную программу, которая в асинхронном режиме совершает GET запросы по заданным адресам и получает код возврата.

Использовать можно библиотеку requests.

requests.get(url)

Запросы совершать через заданный для каждого url таймаут.

Результаты писать в лог файл, используя менеджер контекста with.

Примерный формат лога.

[
    {
        'url': 'http://google.com', result: [
            {'time': '12.00:00', 'result': 200},
            {'time': '12.00:10', 'result': 400}
        ]
    } ... {} 
]