Русский
Регистрация →
← Назад

Рефакторинг Plaza коннектора

Alexander Alexander · 1 мин чтения · 3 354 просмотров
Plaza 2

У нас там ещё и Reflection используется...

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

Какие у кого предложения по структуре?

Комментарии (36)

Вход или Создать аккаунт, Войдите или зарегистрируйтесь, чтобы оставить комментарий

Mikhail Sukhov
Mikhail Sukhov

Alexander: Какие у кого предложения по структуре?

Метаданные (по сути под этим можно считать только наследники от PlazaColumns) нужно отвязать от PlazaTable. Далее, или убрать TableRegistry, или убрать StreamRegistry.

Use cases:

var trader1 = new PlazaTrader();
trader1.FutureTable.Columns.Add(PlazaFututeColumns.Volatility);

var trader2 = new PlazaTrader();
trader2.OptionTable.Columns.Add(PlazaOptionColumns.Volatility);

var trader1 = new PlazaTrader();
trader1.StartExport(trader1.FutureTable);

var trader2 = new PlazaTrader();
trader2.StartExport(trader2.OptionTable);

var trader = new PlazaTrader();

// или
var stream = trader.StartExport(PlazaTableRegistry.Smile);
stream.NewData += 

// или
var stream = trader.StreamRegistry.Smile;
stream.NewData += 
trader.StartExport(trader.StreamRegistry);

Второй вариант мне нравится больше. Соответственно если оставляем реестр стримов, то первые 2 кейса переделываем на использование trader.StreamRegistry

Alexander
Alexander Автор
  1. Как раз уже стал переделывать под второй вариант, мне он тоже больше нравится. PlazaTrader знает только о StreamRegistry, причём именно так:
var stream = trader.StreamRegistry.Smile;
stream.NewData += 
trader.StartExport(trader.StreamRegistry);

ведь не надо забывать о необходимости нескольких подключений - static классов быть не должно.

  1. PlazaTable тоже уже стал удалять из всех метаданных, столкнулся с проблемой рефлектора - GetAllMetadata выкидывает exception.

  2. Необходимо удалить TableRegistry, как тогда будет работать Stream без знаний о текущей таблице?

Mikhail Sukhov
Mikhail Sukhov

Alexander: 2) PlazaTable тоже уже стал удалять из всех метаданных, столкнулся с проблемой рефлектора - GetAllMetadata выкидывает exception.

  1. Необходимо удалить TableRegistry, как тогда будет работать Stream без знаний о текущей таблице?
  1. Через reflection идет инициализация текущей структуры. Тоесть если ты что-то со структурой классов делаешь то и рефлекшен тебе нужно изменить.
  2. Так же как и сейчас, через PlazaStream.Table.
Alexander
Alexander Автор

Mikhail Sukhov: 3. Так же как и сейчас, через PlazaStream.Table.


            streams.Add(Aggregation20FutureStream = new PlazaStream(
                new PlazaTable(PlazaTableSystemName.OrdersAggr, ReplicationStream.Aggr20Future, "Фьючерсы: стакан глубиной 20 котировок")
                ));

?

Mikhail Sukhov
Mikhail Sukhov

Alexander:

Mikhail Sukhov: 3. Так же как и сейчас, через PlazaStream.Table.

        streams.Add(Aggregation20FutureStream = new PlazaStream(
            new PlazaTable(PlazaTableSystemName.OrdersAggr, ReplicationStream.Aggr20Future, "Фьючерсы: стакан глубиной 20 котировок")
            ));

> ?

Да, думаю так и нужно сделать... Может еще проверишь заодно, влияет ли как-нибудь на производительность кол-во созданных стримов (не активных, тоесть данные по ним не текут).
Alexander
Alexander Автор

Текущие таски:

  1. избавиться от перегруженности текущей структуры классов - TableRegistry, ColumnRegistry, StreamRegistry;
  2. убрать зависимость метаданных от Table;
  3. убрать статические классы;
  4. протестировать возможность подключения и получения данных по нескольким логинам.
Mikhail Sukhov
Mikhail Sukhov

Рефакторинг сделал.

Alexander
Alexander Автор

Mikhail Sukhov: Рефакторинг сделал.

Сейчас добавление столбцов происходит после инициализации PlazaStream, в связи с этим возникают проблемы в PlazaRecord.

P.S. fixed. Но данные всё равно не поступают.

FiNick
FiNick

Mikhail Sukhov: Рефакторинг сделал.

Компилил последние исходники, тестил, существующие проблемы не решены: большие задержки при выставлении заявок, загрузка проца под 100% после дисконнекта. Там еще ошибка на SimpleGUI вылетает при подключении(запостил в другой теме).

Alexander
Alexander Автор

FiNick:

Mikhail Sukhov: Рефакторинг сделал.

Компилил последние исходники, тестил, существующие проблемы не решены: большие задержки при выставлении заявок, загрузка проца под 100% после дисконнекта. Там еще ошибка на SimpleGUI вылетает при подключении(запостил в другой теме).

Поправите? Доступ к исходникам у вас вроде есть :)

FiNick
FiNick

Alexander: Поправите? Доступ к исходникам у вас вроде есть :)

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

Mikhail Sukhov
Mikhail Sukhov

FiNick:

Alexander: Поправите? Доступ к исходникам у вас вроде есть :)

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

А с метаданными уже все?

FiNick
FiNick

Mikhail Sukhov: А с метаданными уже все? Метаданные я постепенно дозаполню, это не проблема. Проблема в том что заявки ставятся по 10 секунд. Ведь преимущество плазы это не получение всего многообразия данных с биржи, а максимально быстрая скорость выставления заявок.

Вопрос: это только у меня такая ошибка? у кого-нибудь еще есть проблемы со скоростью? Просто на форуме никто больше не жалуется, очень странно.

Alexander
Alexander Автор

Просьба не забывать добавлять новые файлы в проект (csproj файл), а то не компилируется :)

frontman
frontman

Я жаловался) Ток у меня вообще не приходит ответ от биржи после выставления заявки. Насколько эта проблема связанна с обсуждаемой. Я был бы рад и 10 секундам)

frontman
frontman

Если вы о времени выполнения метода RegisterOrder, то да 10 сек и у меня...

frontman
frontman

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

Alexander
Alexander Автор

Проблема в PollTimeOut в connectionPool. Исследуйте в ту сторону

frontman
frontman
frontman

Смотрите опытным путем установил что событие ITrader.NewOrders вне стратегии срабатывает. Но как только я переношу подписку на это событие в стратегию ничего не работает. Это наводит на мысль что это связанно с классом Strategy как то...

FiNick
FiNick

frontman: Смотрите опытным путем установил что событие ITrader.NewOrders вне стратегии срабатывает. Но как только я переношу подписку на это событие в стратегию ничего не работает. Это наводит на мысль что это связанно с классом Strategy как то...

А как вы регитрируете заявку, strategy.RegisterOrder или plaza.RegisterOrder? Если вторым образом, то события стратегии типа strategy.NewOrders, OrderChanged и т.п. не будут срабатывать. Стратегия не считает такие заявки своими, потому никак не сообщает о них.

frontman
frontman

Так... Теперь установил что если внутри стратегии убрать подписку на событие ITrader.QuotesChanged то новые заявки начинают поступать. Может кто нибудь знает может ли поток данных о стакане как то "забивать" заявки?

frontman
frontman

strategy.RegisterOrder

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

Alexander
Alexander Автор

Выяснил. Задержка, как и писал неоднократно выше, связана с PollTimeOut в PlazaConnectionPool.

У нас заявки посылаются асинхронно, в том же потоке, что делает ProcessMessage. Могут возникать ситуации когда они блокируют. Поставил 100мс, как советуют на форуме РТС.

Далее. Сейчас у нас создаётся 1 подключение для данных и для отправки заявок. Что не оптимально. Предлагаю добавить возможность создавать N (>=1) транзакционных подключений для отправки заявок. В любом случае транзакционное подключение необходимо отделять от потока данных. В этом случае всякая проблема с задержкой заявок уйдёт.

Замечания \ пожелания \ предложения?

P.S. Свой минификс со 100мс залил на codeplex, соберите по исходникам, потестируйте.

Mikhail Sukhov
Mikhail Sukhov

Alexander: Далее. Сейчас у нас создаётся 1 подключение для данных и для отправки заявок. Что не оптимально.

Должно быть раздельно сделано, одно для данных, одно для транзакций. Посмотри, где вызывается класс PlazaConnectionPool.

frontman
frontman

Я думаю даже 100мс скоро много будет) У меня вопрос : а не может ли тот факт что транзакции и получение данных идет в одном потоке мешать выполнению этих самых транзакций. Т.е я говорю не о задержки как писал Александр а о полной блокировки. Например при получении данных о стакане? Т.к. стакан обновляется постоянно, т.е поток входной инфы практически бесконечен?

Alexander
Alexander Автор

frontman: Я думаю даже 100мс скоро много будет) С чего мало? Эти 100мс передаются в ProcessMessage. Для чего они там нужны - см. документацию плазы. Хотите - измените на поменьше, это делайте через свойство StreamTimeOut.

У меня вопрос : а не может ли тот факт что транзакции и получение данных идет в одном потоке мешать выполнению этих самых транзакций. Т.е я говорю не о задержки как писал Александр а о полной блокировки. Например при получении данных о стакане? Т.к. стакан обновляется постоянно, т.е поток входной инфы практически бесконечен?

Так и происходит. Только у нас 2 коннекшена - 1 для данных, 1 - для транзакций. Я по коду проверил, всё должно быть в порядке. Вы протестировали с изменениями? Как результаты?

frontman
frontman

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

frontman
frontman

Александр а вы пробовали одновременно создавать вот такие правила?

this.When(Security.MarketDepthChanged())
                .Do<MarketDepth>(MarketDepthChanged);

this.When(this.StrategyNewOrder())
    .Do<Order>(order => this.AddOrderInfoLog(order, "Выставленна"));
FiNick
FiNick

Повторно подниму эту тему. На сколько я понимаю рефакторинг позволил создавать множество коннекшнов до гейта и соответственно обрабатывать потоки репликации в нескольких тредах. Похоже это не роскошь, а необходимость. Я отсылал в РТС логи, они сделали такие замечания:

По приложенному логу видно, что накапливается очередь сообщений в первые секунды работы. 2011-10-06 16:45:47.234;p2mq-cli;;New message added to recvList. Size: 38 Это приводит к задержкам в получении данных, поэтому предлагается побороться с очередями. Для этого предлагается разбить получение реплики на несколько соединений, работающих в отдельных thread'ах, каждое со своим циклом выборки.

P.S. Начал прикидывать как это можно сделать, пока не получается, все ломается

Alexander
Alexander Автор

FiNick: Повторно подниму эту тему. На сколько я понимаю рефакторинг позволил создавать множество коннекшнов до гейта и соответственно обрабатывать потоки репликации в нескольких тредах. Похоже это не роскошь, а необходимость. Я отсылал в РТС логи, они сделали такие замечания:

По приложенному логу видно, что накапливается очередь сообщений в первые секунды работы. 2011-10-06 16:45:47.234;p2mq-cli;;New message added to recvList. Size: 38 Это приводит к задержкам в получении данных, поэтому предлагается побороться с очередями. Для этого предлагается разбить получение реплики на несколько соединений, работающих в отдельных thread'ах, каждое со своим циклом выборки.

P.S. Начал прикидывать как это можно сделать, пока не получается, все ломается

несколько соединений - connection pool а рефакторинг позволил создавать несколько PlazaTrader

frontman
frontman

Продолжаю свое капание в коде плазы) У меня вызвал некое смущение вот этот метод:

internal void ProcessConnection(IConnectionWrapper connection)
		{
			switch (State)
			{
				case PlazaStreamStates.Closed:
					if (ConfigFileName.IsEmpty())
						throw new InvalidOperationException("ConfigFileName is null or empty");

					NativeStream.TableSet = WrapperHelper.CreateTableSet();
					NativeStream.TableSet.Init(ConfigFileName, Table.ReplicationScheme);

					SetRevision();

					State = PlazaStreamStates.Opened;

					Log.SafeInvoke(this, "Поток открыт.");
					break;
				case PlazaStreamStates.Opened:
					if (NativeStream.State == StreamWrapperStates.Error || NativeStream.State == StreamWrapperStates.Closed)
					{
						Log.SafeInvoke(this, "Состояние потока {0}.".Put(NativeStream.State));

						if (NativeStream.State == StreamWrapperStates.Error)
							NativeStream.Close();

						SetRevision();

						NativeStream.Open(connection);
						ReConnected.SafeInvoke();

						Log.SafeInvoke(this, "Поток переоткрыт.");
					}
					break;
				default:
					throw new ArgumentOutOfRangeException();
			}
		}

Во первых я считаю что его надо разбить на несколько независимых методов. А во вторых с точки зрения логики для меня он непонятен. В первой ветке case PlazaStreamStates.Closed: в лог пишется что поток открыт и статус у потока меняется на Opened, но на самом деле это не так. Поток еще не открыт. Происходит только загрузка схемы для потока. Такая же ситуации ветке case PlazaStreamStates.Opened: В лог пишется "поток переоткрыт" , хотя на самом деле это только его первое открытие(за исключением его реальных переоткрытиях в случае необходимости).

Alexander
Alexander Автор

frontman: Продолжаю свое капание в коде плазы) У меня вызвал некое смущение вот этот метод:

internal void ProcessConnection(IConnectionWrapper connection) { switch (State) { case PlazaStreamStates.Closed: if (ConfigFileName.IsEmpty()) throw new InvalidOperationException("ConfigFileName is null or empty");

				NativeStream.TableSet = WrapperHelper.CreateTableSet();
				NativeStream.TableSet.Init(ConfigFileName, Table.ReplicationScheme);

				SetRevision();

				State = PlazaStreamStates.Opened;

				Log.SafeInvoke(this, "Поток открыт.");
				break;
			case PlazaStreamStates.Opened:
				if (NativeStream.State == StreamWrapperStates.Error || NativeStream.State == StreamWrapperStates.Closed)
				{
					Log.SafeInvoke(this, "Состояние потока {0}.".Put(NativeStream.State));

					if (NativeStream.State == StreamWrapperStates.Error)
						NativeStream.Close();

					SetRevision();

					NativeStream.Open(connection);
					ReConnected.SafeInvoke();

					Log.SafeInvoke(this, "Поток переоткрыт.");
				}
				break;
			default:
				throw new ArgumentOutOfRangeException();
		}
	}

> 
> Во первых я считаю что его надо разбить на несколько независимых методов.
> А во вторых с точки зрения логики для меня он непонятен.
> В первой ветке case PlazaStreamStates.Closed: в лог пишется что поток открыт и статус у потока меняется на Opened, но на самом деле это не так. Поток еще не открыт. Происходит только загрузка схемы для потока.
> Такая же ситуации  ветке case PlazaStreamStates.Opened: В лог пишется "поток переоткрыт" , хотя на самом деле это только его первое открытие(за исключением его реальных переоткрытиях в случае необходимости).


1) на какие методы и зачем? разбивать имеет смысл если этот код может использоваться в других местах. Здесь так?
2) Вас смущает надписать "поток открыт"? ну считайте это "поток открывается"
3) здесь поток переоткрывается, т.к. его статус либо Error, либо Closed

Вы куда-то не туда смотрите. То что сейчас не работает - смена сессий. Прошла неделя с последней нашей беседы на эту тему, вы как-то продвинулись?
frontman
frontman

Alexander Mukhanchikov:

  1. на какие методы и зачем? разбивать имеет смысл если этот код может использоваться в других местах. Здесь так? На методы Initialization, Open, ReOpen. Код разбивают не только в случае его исп в других местах, а еще и с целью логического упрощения. Alexander Mukhanchikov:
  2. Вас смущает надписать "поток открыт"? ну считайте это "поток открывается" Он не открывается, а инициализируеться схема. Поток при этом не обязательно открывать. Alexander Mukhanchikov: Вы куда-то не туда смотрите. То что сейчас не работает - смена сессий. Прошла неделя с последней нашей беседы на эту тему, вы как-то продвинулись? Ну собственно я ответил что нет.
frontman
frontman

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

Насчет смены сесии: скажу честно вопрос пока что отложен, т.к. у меня возникли проблемы с прокачкой данных, что мешает работе робота за которого мне собственно зп платят)) Так что пока что извините...

Alexander
Alexander Автор

frontman: Александр я как раз туда смотрю. И уже много раз вам говорил что вот такое считайте, как бы понятно для вас , но не для тех кто этот код не писал. Все должно быть однозначно. Загрузка схемы - это еще не подключение. Открытие потока и переоткрытие то же разное....

Вопросы я задал. Если хотите от меня реакцию получить - на вопросы надо ответить :)

frontman
frontman

Но это не значит что я отказываюсь от этого вопроса!