Скрыть

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

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

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

Скрыть

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

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

Скрыть

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

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

Скрыть

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

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

Скрыть

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

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

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

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

Скрыть

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

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

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

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

Скрыть

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

Алсо.

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

Скрыть

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

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

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

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

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

Скрыть

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

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

Скрыть

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

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

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

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

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:Помогите разобраться!

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


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

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

Скрыть

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

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

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()