Good practice по алгоритмизации в условиях асинхронности - Архитектура программного обеспечения

Здравствуйте, sergunok, Вы писали:

G>>TPL с Extras (с помощью итераторов), async в F# (монада).
S>Где зрить про Extras?
http://blogs.msdn.com/b/pfxteam/archive/2010/04/04/9990342.aspx
http://code.msdn.microsoft.com/ParExtSamples

G>>Абсолютно не вариант писать коллбеки, посмотри Мейера, он там детально про это рассказывает.
S>В Twisted'е привык

Только через некоторое время стал пользоваться его же "сахаром" с yield.
Ну суть в том что колбеки контекст теряют (читай не работают замыкания).

G>>Надо мудрствовать, тебе нужна функция IQueryable<T1> -> IQueryable<T2>, где T1 входные события, T2 — выходное состояние, возможно после выполнения некоторых действий. Оставайся в монаде и будет счастье.
S>Кстати, что есть монада в данном контексте?

IQueryable<T>

Здравствуйте, sergunok, Вы писали:

S>>> Причем реакция может быть "сложной", до своего завершения требовать прихода нескольких различных событий и отсылки ответных воздействий. Пример — реализация большого объема акций. На бирже такая заявка может не реализоваться за раз и будут приходить цепочки ответов, "реализовано N акций". Такая сложные реакция сама может внутри себя содержать свою очередь событий. Однако присутствовать в основной очереди реакций просто как отдельная реакция.
G>>Видимо наоборот из нескольких событий "продано N акций" будет получаться событие "продан пакет акций". По факту никаких очередей не будет, буду асинхронные коллекции.
S>Про асинхронные коллекции — да.

S>В примере имею в виду, что действие может быть не атомарным, а многошаговым и высокоуровневым. Типа продавать до победного по рыночной цене.. И в основной очереди действий, например, захочется оперировать именно такими действиями. А каждое из них будет декомпозироваться на более мелкие шажки.
Во, ты почувствовал Силу, называется она функциональное программирование.

Здравствуйте, Undying, Вы писали:

U>Здравствуйте, Undying, Вы писали:

U>>Хотя нет, я не прав, проблема, что у меня внутри асинхронного метода не будет ссылки на Task. Т.е. если пошаманить работать будет, а вот красивого решения пока что не вижу.

U>Изящное решение все же есть. Добавляем еще один вид Step:

U>

U>class ServiceStep : WaitStep
U>{
U>  public ServiceStep() : base(TimeSpan.Zero) {}
U>  public void SetServiceInfo(LabeledThread thread, Task task);
U>  public void ForceTask()
U>  {
U>    thread.ForceTask(task);
U>  }
U>}
U>

U>Т.е. получив в качестве результата такой Step, LabeledThread вызывает у него SetServiceInfo и передает ему ссылку на себя и на таску.
То есть таки придется переписать чуть менее чем весь клиентский код.

U>Соответственно получаем возможность с легкостью комбинировать событийную и временную модель пробуждения потоков, используя их сильные стороны.
1)В чем сильная сторона пробуждения по времени?
2)Почему бы не сделать System.Threading.Timer для "времнной модели", а не делать ожидание в потоке исполнителе?
3)Как исполнитель будет узнавать о том что событие таки наступило?

Пытаюсь подступить к RX со стороны биржевого симулятора.

На базе которого можно будет потом гонять какие-то свои примерчики стратегий на RX.

Но что-то тяжко идет..

Во-первых, почему-то не вижу элементов в OrderChanged, во-вторых не работает Delay.

В чем может быть дело?

    public enum OrderStatus
    {
        Sent,
        Matched,
        Canceled
    }

    class Order
    {
        public double Price;
        public int Volume;
        public OrderStatus Status;
    }
        
    class Trader
    {
        
        public IObservable<double> Ticks {get; private set; }
        public IObservable<Order> OrderChanged { get; private set; }

        private Action<Order> orderChanged;

        public Trader()
        {
            var rnd = new Random();
            this.Ticks = Observable.Defer(
                () => Observable.Return(Math.Round(1500.0 + rnd.NextDouble() * 10, 2)))
                //.Delay(TimeSpan.FromMilliSeconds(100 * rnd.NextDouble())))
        .Repeat();

            var subj = new Subject<Order>();
            orderChanged += subj.OnNext;

            this.OrderChanged = Observable.Defer(() => subj);
            //this.OrderChanged = Observable.Defer(() => subj).Throttle(TimeSpan.FromMilliseconds(500 * rnd.NextDouble()));
        }

        public void RegisterOrder(Order order)
        {
            order.Status = OrderStatus.Matched;
            orderChanged(order);
        }

        public void CancelOrder(Order order)
        {
            order.Status = OrderStatus.Canceled;
            orderChanged(order);
        }      
    }

    class Program
    {
        static void Main(string[] args)
        {
            var trader = new Trader();
            using (trader.Ticks.Subscribe(
            tick => 
            {
                Console.WriteLine("Tick:" +  tick);
                if (tick > 1505) 
                    trader.RegisterOrder(new Order()); 
            },
            ex => Console.WriteLine(ex),
            () => Console.WriteLine("Completed")
            ))
            {
            }

            using (trader.OrderChanged.Subscribe(
                        order =>
                        {
                            Console.WriteLine("OrderChanged:", order.Status);
                        },
                        ex => Console.WriteLine(ex),
                        () => Console.WriteLine("Completed")
                        ))
            {
            }

            Console.ReadKey();
        }
    }

Здравствуйте, sergunok, Вы писали:

S>Пытаюсь подступить к RX со стороны биржевого симулятора.

S>На базе которого можно будет потом гонять какие-то свои примерчики стратегий на RX.

S>Но что-то тяжко идет..

S>Во-первых, почему-то не вижу элементов в OrderChanged, во-вторых не работает Delay.

S>В чем может быть дело?

S>    public enum OrderStatus
S>    {
S>        Sent,
S>        Matched,
S>        Canceled
S>    }

S>    class Order
S>    {
S>        public double Price;
S>        public int Volume;
S>        public OrderStatus Status;
S>    }
        
S>    class Trader
S>    {
        
S>        public IObservable<double> Ticks {get; private set; }
S>        public IObservable<Order> OrderChanged { get; private set; }

S>        private Action<Order> orderChanged;

S>        public Trader()
S>        {
S>            var rnd = new Random();
S>            this.Ticks = Observable.Defer(
S>                () => Observable.Return(Math.Round(1500.0 + rnd.NextDouble() * 10, 2)))
S>                //.Delay(TimeSpan.FromMilliSeconds(100 * rnd.NextDouble())))
S>        .Repeat();//Используй GenerateWithTime

S>            var subj = new Subject<Order>();
S>            orderChanged += subj.OnNext;

S>            this.OrderChanged = Observable.Defer(() => subj); //Defer тут не нужен
S>            //this.OrderChanged = Observable.Defer(() => subj).Throttle(TimeSpan.FromMilliseconds(500 * rnd.NextDouble()));
S>        }

S>        public void RegisterOrder(Order order)
S>        {
S>            order.Status = OrderStatus.Matched;
S>            orderChanged(order);
S>        }

S>        public void CancelOrder(Order order)
S>        {
S>            order.Status = OrderStatus.Canceled;
S>            orderChanged(order);
S>        }      
S>    }

S>    class Program
S>    {
S>        static void Main(string[] args)
S>        {
S>            var trader = new Trader();
S>            using (trader.Ticks.Subscribe(
S>            tick => 
S>            {
S>                Console.WriteLine("Tick:" +  tick);
S>                if (tick > 1505) 
S>                    trader.RegisterOrder(new Order()); 
S>            },
S>            ex => Console.WriteLine(ex),
S>            () => Console.WriteLine("Completed")
S>            ))
S>            {
S>            }

S>            using (trader.OrderChanged.Subscribe(
S>                        order =>
S>                        {
S>                            Console.WriteLine("OrderChanged:", order.Status);
S>                        },
S>                        ex => Console.WriteLine(ex),
S>                        () => Console.WriteLine("Completed")
S>                        ))
S>            {
S>            }

S>            Console.ReadKey();
S>        }
S>    }
S>

Пару комментов выделил.

Кроме того:
1)Не надо иметь состояние (order), лучше его передавать в IObservable
2)Не надо делать subscribe, надо использовать комбинаторы и находиться в монаде до последнего, то есть передавать IObservable пока это возможно.

Здравствуйте, sergunok, Вы писали:

Кстати

S>using (trader.Ticks.Subscribe(...))
S>{
S>}

Подписка сразу же снимается по завершении блока using, не стоит так делать.

Здравствуйте, gandjustas, Вы писали:

U>>Соответственно получаем возможность с легкостью комбинировать событийную и временную модель пробуждения потоков, используя их сильные стороны.
G>1)В чем сильная сторона пробуждения по времени?

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

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

G>2)Почему бы не сделать System.Threading.Timer для "времнной модели", а не делать ожидание в потоке исполнителе?

А смысл? TaskPulling создает для своей работы указанное число потоков, поставить задачи в эти потоки можно только в виде Task, соответственно если Task'ов требующих выполнения в текущий момент нет, то делать ничего не надо и поток временно засыпает.

G>3)Как исполнитель будет узнавать о том что событие таки наступило?

Что тебе непонятно в этом коде: http://rsdn.ru/forum/design/3931808.1.aspx ?

СПАСИБО!

Основная проблема была именно из-за using..

Но что самое интересное с using но без Delay все работало, т.е. подписка не заканчивалась.

G>Кроме того:
G>1)Не надо иметь состояние (order), лучше его передавать в IObservable
Это ты про что? Менять order.State в Do?

G>2)Не надо делать subscribe, надо использовать комбинаторы и находиться в монаде до последнего, то есть передавать IObservable пока это возможно.
Поясни plz.

Кстати, RegisterOrder и CancelOrder реализован через дополнительный event orderChanged и Subject. Есть другой способ пополнения IObservable? (без введения event'а)

Здравствуйте, Undying, Вы писали:

U>Здравствуйте, gandjustas, Вы писали:

U>>>Соответственно получаем возможность с легкостью комбинировать событийную и временную модель пробуждения потоков, используя их сильные стороны.
G>>1)В чем сильная сторона пробуждения по времени?

U>Во-первых, по коду сразу же видна последовательность выполнения действий, соответственно хоть код и является асинхронным понимать его столь же легко, как и синхронный. В чисто событийной модели последовательность действий не очевидна, соответственно и понимать его сложно и баги связанные с неопределенностью последовательности вызова событий очень сложно недопускать и отлавливать.
Если используется yield, то последовательность и так видна, коллбек нужен только для продолжения.

G>>2)Почему бы не сделать System.Threading.Timer для "времнной модели", а не делать ожидание в потоке исполнителе?

U>А смысл? TaskPulling создает для своей работы указанное число потоков, поставить задачи в эти потоки можно только в виде Task, соответственно если Task'ов требующих выполнения в текущий момент нет, то делать ничего не надо и поток временно засыпает.
Я уже писал что при ожидании по времени есть некоторый предел, связаный с точностью системного таймера, для колбеков — нет, поток будет спать ровно столько сколько нужно.

G>>3)Как исполнитель будет узнавать о том что событие таки наступило?

U>Что тебе непонятно в этом коде: http://rsdn.ru/forum/design/3931808.1.aspx ?
Где будет вызываться SetServiceInfo для ServiceTask?

Здравствуйте, sergunok, Вы писали:

S>СПАСИБО!

S>Основная проблема была именно из-за using..

S>Но что самое интересное с using но без Delay все работало, т.е. подписка не заканчивалась.

G>>Кроме того:
G>>1)Не надо иметь состояние (order), лучше его передавать в IObservable
S>Это ты про что? Менять order.State в Do?
Вообще не надо иметь изменяемую переменную order

G>>2)Не надо делать subscribe, надо использовать комбинаторы и находиться в монаде до последнего, то есть передавать IObservable пока это возможно.
S>Поясни plz.
Ну просто не писать Subscribe и использовать комбинаторы.

S>Кстати, RegisterOrder и CancelOrder реализован через дополнительный event orderChanged и Subject. Есть другой способ пополнения IObservable? (без введения event'а)
Конечно, RegisterOrder и CancelOrder сами могут возвращать IObservable.

Здравствуйте, gandjustas, Вы писали:

U>>А смысл? TaskPulling создает для своей работы указанное число потоков, поставить задачи в эти потоки можно только в виде Task, соответственно если Task'ов требующих выполнения в текущий момент нет, то делать ничего не надо и поток временно засыпает.

Ничего не понял. Как твой исходный вопрос этой ветки обсуждения:

G>2)Почему бы не сделать System.Threading.Timer для "времнной модели", а не делать ожидание в потоке исполнителе?

Связан с твоим дальнейшим ответом:

G>Я уже писал что при ожидании по времени есть некоторый предел, связаный с точностью системного таймера, для колбеков — нет, поток будет спать ровно столько сколько нужно.

???

U>>Что тебе непонятно в этом коде: http://rsdn.ru/forum/design/3931808.1.aspx ?
G>Где будет вызываться SetServiceInfo для ServiceTask?

Под ServiceTask ты имеешь в виду ServiceStep? SetServiceInfo будет вызываться в LabeledThread.ExecuteStep.

G>>>2)Не надо делать subscribe, надо использовать комбинаторы и находиться в монаде до последнего, то есть передавать IObservable пока это возможно.
S>>Поясни plz.
G>Ну просто не писать Subscribe и использовать комбинаторы.
т.е. использовать extension методы RX и LINQ над IObserver?

S>>Кстати, RegisterOrder и CancelOrder реализован через дополнительный event orderChanged и Subject. Есть другой способ пополнения IObservable? (без введения event'а)
G>Конечно, RegisterOrder и CancelOrder сами могут возвращать IObservable.
А как примерно будет выглядеть метод RegisterOrder с таким типом возвращаемого значения да еще и при условии, что исполнение заявки может происходить частями (т.е. будет несколько раз модифицироваться RealizedVolume в Order)?

    public class Order
    {
        public int Id;
        public double Price;
        public int Volume;
        public int RealizedVolume;
        public OrderStatus Status;

        public override string ToString()
        {
            return string.Format("Order Id={0} Status={1}", Id, Status);
        }
    }

Здравствуйте, Undying, Вы писали:

U>Здравствуйте, gandjustas, Вы писали:

U>>>А смысл? TaskPulling создает для своей работы указанное число потоков, поставить задачи в эти потоки можно только в виде Task, соответственно если Task'ов требующих выполнения в текущий момент нет, то делать ничего не надо и поток временно засыпает.

U>Ничего не понял. Как твой исходный вопрос этой ветки обсуждения:

G>>2)Почему бы не сделать System.Threading.Timer для "времнной модели", а не делать ожидание в потоке исполнителе?

U>Связан с твоим дальнейшим ответом:

G>>Я уже писал что при ожидании по времени есть некоторый предел, связаный с точностью системного таймера, для колбеков — нет, поток будет спать ровно столько сколько нужно.

U>???
Никак.
Я просто не вижу смысла ждать определенное время. В 99.9% случаев надо продолжать выполнение после какого-то события, а тот 0.1% где действительно надо ждать некоторое время можно реализовать таймером.

U>>>Что тебе непонятно в этом коде: http://rsdn.ru/forum/design/3931808.1.aspx ?
G>>Где будет вызываться SetServiceInfo для ServiceTask?

U>Под ServiceTask ты имеешь в виду ServiceStep?
Да.

U>SetServiceInfo будет вызываться в LabeledThread.ExecuteStep.
Уже понял, все равно придется править весь клиентский код для такого поведения.

ЗЫ. Кстати так и не увидел копирования потоков на TaskPull

Здравствуйте, gandjustas, Вы писали:

G>Я просто не вижу смысла ждать определенное время. В 99.9% случаев надо продолжать выполнение после какого-то события, а тот 0.1% где действительно надо ждать некоторое время можно реализовать таймером.

А, к примеру, отваливание по таймауту как делать?

U>>SetServiceInfo будет вызываться в LabeledThread.ExecuteStep.
G>Уже понял, все равно придется править весь клиентский код для такого поведения.

Логично. Если хочешь использовать только временную модель, то о ForceTask ничего знать не надо. Если хочешь использовать еще и событийную, то надо будет добавить три строчки кода в каждую таску. В чем проблема-то?

G>ЗЫ. Кстати так и не увидел копирования потоков на TaskPull

Сейчас отвечу.

Здравствуйте, gandjustas, Вы писали:

U>>Чтения блока естественно выполняется синхронно, а как оно еще может выполняться?
G>Асинхронно.

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

G>>>Надо использовать BeginRead\EndRead, BeginWrite\EndWrite.
U>>Зачем? В чем будет преимущество?
G>В том что поток на время чтения не будет блокироваться.

Он и так не блокируется. Можешь поставить чтение из нескольких файлов в один поток и оно будет выполняться параллельно, а не последовательно.

Здравствуйте, sergunok, Вы писали:

G>>>>2)Не надо делать subscribe, надо использовать комбинаторы и находиться в монаде до последнего, то есть передавать IObservable пока это возможно.
S>>>Поясни plz.
G>>Ну просто не писать Subscribe и использовать комбинаторы.
S>т.е. использовать extension методы RX и LINQ над IObserver?
над IObservable

S>>>Кстати, RegisterOrder и CancelOrder реализован через дополнительный event orderChanged и Subject. Есть другой способ пополнения IObservable? (без введения event'а)
G>>Конечно, RegisterOrder и CancelOrder сами могут возвращать IObservable.
S>А как примерно будет выглядеть метод RegisterOrder с таким типом возвращаемого значения да еще и при условии, что исполнение заявки может происходить частями (т.е. будет несколько раз модифицироваться RealizedVolume в Order)?
IObservable<RealizedVolume> например

Посмотри мои примеры, там ровно одни Subscribe, и ниче.

Здравствуйте, Undying, Вы писали:

U>Здравствуйте, gandjustas, Вы писали:

G>>Я просто не вижу смысла ждать определенное время. В 99.9% случаев надо продолжать выполнение после какого-то события, а тот 0.1% где действительно надо ждать некоторое время можно реализовать таймером.

U>А, к примеру, отваливание по таймауту как делать?

1)Отваливание по таймауту обычно уже реализовано в асинхронном методе
2)Как уже писал: ждать события завершения или таймера.

U>>>SetServiceInfo будет вызываться в LabeledThread.ExecuteStep.
G>>Уже понял, все равно придется править весь клиентский код для такого поведения.

U>Логично. Если хочешь использовать только временную модель, то о ForceTask ничего знать не надо. Если хочешь использовать еще и событийную, то надо будет добавить три строчки кода в каждую таску. В чем проблема-то?

В том что переписывать надо.

Здравствуйте, Undying, Вы писали:

U>Здравствуйте, gandjustas, Вы писали:

U>>>Чтения блока естественно выполняется синхронно, а как оно еще может выполняться?
G>>Асинхронно.

U>Что значит асинхронно? В смысле в другом потоке?
Нет, для чтения с диска например вообще не нужны потоки, вот Stream и позволяет вызывать асинхронную операцию, которая для большинства стримов отобразится на асинхронную операцию в ОС, которая в свою очередь вообще не трогает потоки до завершения.

U>Ну так поставь таску на чтение файла в другой поток, а не в основной и будет тебе счастье.
Это не то.

G>>>>Надо использовать BeginRead\EndRead, BeginWrite\EndWrite.
U>>>Зачем? В чем будет преимущество?
G>>В том что поток на время чтения не будет блокироваться.

U>Он и так не блокируется.
Блокируется пока выполняется Read и Write.

U>Можешь поставить чтение из нескольких файлов в один поток и оно будет выполняться параллельно, а не последовательно.
нет, нужно именно один Stream

Ты не в курсе как работают асинхронные операции?

Здравствуйте, gandjustas, Вы писали:

U>>А, к примеру, отваливание по таймауту как делать?
G>1)Отваливание по таймауту обычно уже реализовано в асинхронном методе

А если не реализовано? То ждать пока реализуют?

G>2)Как уже писал: ждать события завершения или таймера.

Какого таймера? И когда должен вызываться таймер, если операция у нас состоит из трех асинхронных шагов, у первого из которых таймаут 15 секунд, у второго — 5 минут, у третьего — 1 секунда?

U>>Логично. Если хочешь использовать только временную модель, то о ForceTask ничего знать не надо. Если хочешь использовать еще и событийную, то надо будет добавить три строчки кода в каждую таску. В чем проблема-то?

G>В том что переписывать надо.

А в альтернативных вариантах информация по какому событию надо завершать выполнение шага будет проброшена магически? Или для этого тоже придется код писать?

Здравствуйте, Undying, Вы писали:

U>Здравствуйте, gandjustas, Вы писали:

U>>>А, к примеру, отваливание по таймауту как делать?
G>>1)Отваливание по таймауту обычно уже реализовано в асинхронном методе

U>А если не реализовано? То ждать пока реализуют?

G>>2)Как уже писал: ждать события завершения или таймера.

U>Какого таймера? И когда должен вызываться таймер, если операция у нас состоит из трех асинхронных шагов, у первого из которых таймаут 15 секунд, у второго — 5 минут, у третьего — 1 секунда?
Ну сделать 3 таймера, проблема?

U>>>Логично. Если хочешь использовать только временную модель, то о ForceTask ничего знать не надо. Если хочешь использовать еще и событийную, то надо будет добавить три строчки кода в каждую таску. В чем проблема-то?

G>>В том что переписывать надо.

U>А в альтернативных вариантах информация по какому событию надо завершать выполнение шага будет проброшена магически? Или для этого тоже придется код писать?
Ну я же приводил пример с чтением файла http://rsdn.ru/forum/design/3931077.aspx
Явного ожидания нет, тем не менее между вызовом асинхронного чтения и завершением операции ни один поток блокирован не будет.

От:	gandjustas	http://blog.gandjustas.ru/
Дата:	24.08.10 18:38
Оценка:

От:	gandjustas	http://blog.gandjustas.ru/
Дата:	24.08.10 18:40
Оценка:

От:	gandjustas	http://blog.gandjustas.ru/
Дата:	24.08.10 18:49
Оценка:

	От:	sergunok
	Дата:	24.08.10 21:36
	Оценка:

От:	gandjustas	http://blog.gandjustas.ru/
Дата:	24.08.10 21:50
Оценка:	2 (1)

	От:	Undying
	Дата:	25.08.10 03:48
	Оценка:

	От:	Undying
	Дата:	25.08.10 05:39
	Оценка:

	От:	Undying
	Дата:	25.08.10 05:59
	Оценка:

	От:	Undying
	Дата:	25.08.10 06:01
	Оценка: