rulinux.net - Форум - Development - Помогите разобраться!

Re:Помогите разобраться!

Wed, 20 Mar 2013 19:36:15 +0400

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

python



import sys

import pyev

import socket

import logging

import traceback

from collections import deque

from functools import partial

from pyev import EV_WRITE, EV_READ, EV_TIMER



class coroutine(object):

    def __new__(cls, *args, **kwds):

        if not hasattr(cls, '_instance'):

            cls.revents = 0

            cls._instance = super(coroutine, cls).__new__(cls)

            cls._instance.loop = pyev.default_loop()

            cls._instance.queue = deque()

        if args or kwds:

            return cls._instance(*args, **kwds)

        else:

            return cls._instance



    def __call__(self, func):

        def decorator(*args, **kwds):

            cont = func(*args, **kwds)

            self.queue.append((cont, (None,)))

            return cont

        return decorator

    

    @classmethod

    def deferred_task(cls, func):

        def decorated(*args, **kwds):

            return partial(func, *args, **kwds)

        return decorated

    

    @staticmethod

    def callback(container, task, watcher, revents):

        for watcher in container[0]:

            watcher.stop()

        container[1] = revents

        coroutine.revents = revents

        container.append(task())

        coroutine.revents = 0



    @classmethod

    def run(cls):

        self = cls()

        while self.queue:

            cont, data = self.queue[0]

            if data == (None,):

                # first call

                self.queue.popleft()

                task, prewatchers = (lambda seq:(seq[0], seq[1:]))(cont.next())

                self.queue.append((cont, (prewatchers, task)))

            elif isinstance(data[0], tuple) and isinstance(data[1], partial):

                # watchers to install

                self.queue.popleft()

                prewatchers, task = data

                self.queue.append((cont, [[], 0]))

                container = self.queue[-1][1]

                for prewatcher in prewatchers:

                    watcher = prewatcher(partial(cls.callback, container, task))

                    container[0].append(watcher)

                    watcher.start()

            elif len(data)==3:

                # result ready

                self.queue.popleft()

                watchers, revents, result = data

                try:

                    task, prewatchers = (lambda seq:(seq[0], seq[1:]))(cont.send(result))

                    self.queue.append((cont, (prewatchers, task)))

                except StopIteration:

                    pass

            else:

                self.queue.rotate(-1)

            # catches events

            self.loop.start(pyev.EVRUN_NOWAIT)





class Timeout(object):

    def __new__(self, seconds):

        return partial(pyev.Timer, seconds, 0, coroutine().loop)

        



class Io(object):

    def __new__(self, fd, *events):

        eventmask = reduce(lambda acc, value: acc|value, events, 0)

        return partial(pyev.Io, fd, eventmask, coroutine().loop)





class EchoServer(object):

    def __init__(self, *address):

        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)

        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

        self.socket.setblocking(0)

        self.socket.bind(address)

        self.socket.listen(128)



    @coroutine.deferred_task

    def accept(self):

        if coroutine.revents & EV_READ:

            return self.socket.accept()

        else:

            return None, coroutine.revents



    @coroutine.deferred_task

    def read(self, socket):

        try:

            if coroutine.revents & EV_READ:

                return socket.recv(1024)

            else:

                return None

        except:

            return sys.exc_info()

            



    @coroutine.deferred_task

    def write(self, socket, bytes):

        try:

            if coroutine.revents & EV_WRITE:

                return socket.send(bytes)

            else:

                return 0

        except:

            return sys.exc_info()

    

    @coroutine

    def start(self):

        while self.socket:

            client = yield self.accept(), Io(self.socket.fileno(), EV_READ), Timeout(3.0)

            if client[0]:

                self.connection_handler(*client)

                connection, address = client

                

    def close(self):

        self.socket.close()

        self.socket = None

    

    @coroutine

    def connection_handler(self, connection, address):

        logging.debug('Connected client: {0}:{1}.'.format(*address))

        connection.setblocking(0)

        try:

            while connection:

                bytes = yield self.read(connection), Io(connection.fileno(), EV_READ), Timeout(15.0)

                if isinstance(bytes, tuple):

                    logging.error('Disconnected client: {0}:{1} by exception.'.format(*address), exc_info=typle)

                    break

                if bytes:

                    while bytes:

                        sent = yield self.write(connection, bytes), Io(connection.fileno(), EV_WRITE), Timeout(15.0)

                        if isinstance(sent, tuple):

                            logging.error('Disconnected client: {0}:{1} by exception.'.format(*address), exc_info=sent)

                            break

                        if sent>0:

                            bytes = bytes[sent:]

                        else:

                            break

                    else:

                        continue

                break

        except:

            logging.exception('Disconnected client: {0}:{1} by exception.'.format(*address))

        logging.debug('Disconnected client: {0}:{1} by timeout.'.format(*address))

        connection.close()



if __name__ == '__main__':

    logging.basicConfig(level=logging.DEBUG)

    echo = EchoServer("127.0.0.1", 7000)

    echo.start()

    try:

        coroutine.run()

    except KeyboardInterrupt:

        echo.close()

Re:Помогите разобраться!

Wed, 20 Mar 2013 13:25:41 +0400

> Меня смущает только один момент:
> Сначала мы читаем, отдаём в менеджер, потом только в следующем цикле эти данные получает сама функция.

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

> А как ты, кстати, хочешь это сделать? В tornado вот полноценный io loop с select/poll.
Если мне еще на пару дней задержат зарплату, то увидишь)))

Re:Помогите разобраться!

Wed, 20 Mar 2013 13:12:13 +0400

>Если ты заметил конкурентное читается стреам тремя тасками.
Ну, если докручивать дальше, то да, что-то выйдет. Просто твой конкретный пример не сильно отличается от трёх генераторов в списке, которые циклом обходятся:)

Коллбэки на завершение тасков добавить явно стоит.

Меня смущает только один момент:

python



try:

    data = cont.send(data[0])

    self.queue.append((cont, (data,)))

except StopIteration:

    pass

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

>Прикрутить туда передачу управления назад в коротину, когда стрим доступен для операции I/O дело техники.
А как ты, кстати, хочешь это сделать? В tornado вот полноценный io loop с select/poll.

>Кстати как ты считаешь? Это все таки continuation или coroutine в чистом виде?
Я не спец, но назвал бы coroutine. Как раз это функция, которая может остановиться и продолжить выполнение с некоторой точки. Точнее, конечно, твой код таки реализует механизм для создания coroutine, а самими coroutine являются функции, завёрнутые в декоратор:) А continuation это хитрое понятие, которое я своими словами описать могу с трудом, что-то вроде функции, которая позволяет переехать в другое место выполнения программы с сохранением контекста. Эдакое goto. Но вещи эти связанные, так как в случае с coroutine вполне себе необходимо прыгать между функцией и остальной программой.

Re:Помогите разобраться!

Wed, 20 Mar 2013 10:42:48 +0400

> Вот более каноничный корутин
Этому короутину надо возвращать управление. Кстати в той статье есть и примеры с диспечером короутин. Но у меня более компактно и красиво на мой взгляд получилось.

Re:Помогите разобраться!

Wed, 20 Mar 2013 10:41:02 +0400

> Не могу сходу представить, зачем такое тебе понадобилось
Асинхронный вебсервер например, нее?

Если ты заметил конкурентное читается стреам тремя тасками. Прикрутить туда передачу управления назад в коротину, когда стрим доступен для операции I/O дело техники.

Ну и потом я считаю шикарно получился класс, как декоратор и менаджер короутин одновременно.

Кстати как ты считаешь? Это все таки continuation или coroutine в чистом виде?

Re:Помогите разобраться!

Wed, 20 Mar 2013 10:34:45 +0400

Алсо.

>Всегда был слаб по части "паттернов программирования" и прочей теоретической чепухи.
И это говорит человек, который постоянно применяет умную фразу Dependency Injection:)

Re:Помогите разобраться!

Wed, 20 Mar 2013 10:30:23 +0400

>При таска по чтению потока, работают одновременно, ну "почти одновременно")))
Да если бы:)

А вообще вы, батенька, извращенец, я даже мозг поломал сначала. Порадовали манипуляции с send, я такое редко встречал.

Наверное это корутин, но он какой-то странный. Вот более каноничный корутин, но там таки между вызовами send идёт какая-то работа, а у тебя всё кучей в run выполняется. Ведь send у тебя вызывается в только run, и между ним ничего не воткнуть - где выгода от возврата выполнения из функции? Не могу сходу представить, зачем такое тебе понадобилось вообще.

Re:Помогите разобраться!

Wed, 20 Mar 2013 06:53:33 +0400

> Еще бы знать что етот твой класс делает.
При таска по чтению потока, работают одновременно, ну "почти одновременно")))

Покойные винды до версии Вынь95, работали на аналогичном принципе. Раньше это называлась "вытесняемая многозадачность". Сейчас этот термин не используется, ибо не вполне кошерен. А современными я не могу дать точное определение.

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

Re:Помогите разобраться!

Wed, 20 Mar 2013 06:45:21 +0400

Еще бы знать что етот твой класс делает. Ибо нихуя не понятно акромя того что ето пистон.

Re:Помогите разобраться!

Wed, 20 Mar 2013 06:44:12 +0400

Какая разница как ты его назвал, это же программа на петоне!

Re:Помогите разобраться!

Wed, 20 Mar 2013 06:43:19 +0400

>> Как вы думаете, шо эта?
> Это программа на петоне!
Эта я понел))) Пояснение для танкистов: Вопрос заключается в том правильно ли я назвал класс?

Re:Помогите разобраться!

Wed, 20 Mar 2013 06:39:39 +0400

> Как вы думаете, шо эта?

Это программа на петоне!

Помогите разобраться!

Wed, 20 Mar 2013 06:29:02 +0400

Вчера на ночь начитался интересных статей)))
С утра сваял интересную на мой взгляд штуку, но не знаю как её назвать. Всегда был слаб по части "паттернов программирования" и прочей теоретической чепухи.

В общем не знаю что это я реализовал continuation или coroutine. Как вы думаете, шо эта? Пока назвал continuation, но не уверен)))

python



# -*- coding: utf8 -*-

from StringIO import StringIO

from collections import deque

from functools import partial



TEXT = u"""Кандидатура помощника президента РФ и бывшего министра

экономического развития Эльвиры Набиуллиной на пост главы Центрального

банка внесена на рассмотрение Государственной Думы. В какие сроки она

будет рассмотрена, не уточняется. Нынешний глава ЦБ Сергей Игнатьев

покинет свой пост в июне текущего года.

"""



class continuation(object):

    def __new__(cls, *args, **kwds):

        if not hasattr(cls, '_instance'):

            cls._instance = super(continuation, cls).__new__(cls)

            cls._instance.queue = deque()

        if args or kwds:

            return cls._instance(*args, **kwds)

        else:

            return cls._instance

    

    def __call__(self, func):

        def decorator(*args, **kwds):

            cont = func(*args, **kwds)

            self.queue.append((cont, (None,)))

            return cont

        return decorator

    

    @classmethod

    def run(cls):

        self = cls()

        while self.queue:

            cont, data = self.queue.popleft()

            if data == (None,):

                data = cont.next()

                self.queue.append((cont, (data,)))

            else:

                try:

                    data = cont.send(data[0])

                    self.queue.append((cont, (data,)))

                except StopIteration:

                    pass





@continuation

def reader(stream, num):

    while True:

        data = yield stream.read(16)

        if data:

            print "#{0}: {1}".format(num, data.encode('utf-8'))

        else:

            break





if __name__ == '__main__':

    stream = StringIO(TEXT)

    reader(stream, 1)

    reader(stream, 2)

    reader(stream, 3)

    continuation.run()