Реализация взаимодействия между клиентскими терминалами MetaTrader 5 при помощи именованных каналов (Named Pipes)

Введение

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

Естественное решение было в использовании отдельных файлов на жестком диске - один терминал будет писать данные в файл, другой будет читать их. Этот метод, будучи приемлемым для отправки одиночных сообщений, не является эффективным для потоков котировок.

Потом я ознакомился с хорошей статьей "MetaTrader 5. Экспорт котировок в .NET приложение, используя WCF сервисы", в которой рассказывается о том, как экспортировать котировки в приложения .NET при помощи технологии WCF. А когда я приблизился к завершению, появилась другая статья - "Использование WinInet.dll для обмена данными между терминалами через Интернет".

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

В процессе поиска в Сети я нашел рекомендацию, что для такой связи можно использовать именованные каналы (именованные конвейеры, named pipes) и внимательно прочитал спецификацию MSDN о межпроцессном взаимодействии с использованием именованных каналов.

Я обнаружил, что именованные каналы могут использоваться как на одном компьютере, так и на различных компьютерах локальной сети (intranet), поэтому я решил использовать этот подход.

Данная статья познакомит вас с осуществлением связи между процессами посредством именованных каналов. Предложен класс CNamedPipes, реализующий возможность использования именованных каналов.

В статье также приведен тиковый индикатор для тестирования связи между терминалами MetaTrader 5 и измерения общей пропускной способности системы.

1. Связь между процессами при помощи именованных каналов

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

Представьте себе трубу, которая соединяет два процесса, в нашем случае это терминалы MetaTrader, которые обмениваются данными.

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

Для того чтобы взаимодействовать при помощи каналов, серверный процесс должен настроить канал с известным именем. Имя канала - строка, которая должна иметь вид: "\\servername\pipe\pipename". Если каналы используются на одном и том же компьютере, имя сервера может быть опущено, но вместо него нужно поставить точку: "\\.\pipe\pipename". Клиент, который пытается соединиться с каналом, должен знать его имя.

Для того чтобы различать терминалы, для имен каналов я использую обозначения вида: "\\.\pipe\mt[account_number]", где account_number-номер счета, но способ именования каналов может быть и другой.

2. Реализация класса CNamedPipes

Я начну с краткого описания низкоуровневого механизма создания и соединения с именованным каналом. В операционных системах Windows все функции управления каналами доступны через библиотеку kernel32.dll. Функция CreateNamedPipe() создает именованный канал на стороне сервера.

После того как канал создан, сервер вызывает функцию ConnectNamedPipe() для ожидания клиентских соединений.  Если соединение успешно, ConnectNamedPipe() возвращает ненулевое значение. Впрочем, не исключено, что клиент будет успешно подключен после вызова CreateNamedPipe() и до вызова функции ConnectNamedPipe(). В этом случае ConnectNamedPipe() вернет нулевое значение, а функция GetLastError() вернет ошибку 535(0X217) : ERROR_PIPE_CONNECTED.

Запись и чтение из канала осуществляется с помощью тех же функций, что и функции доступа к файлам:

BOOL WINAPI ReadFile(
  __in         HANDLE hFile,
  __out        LPVOID lpBuffer,
  __in         DWORD nNumberOfBytesToRead,
  __out_opt    LPDWORD lpNumberOfBytesRead,
  __inout_opt  LPOVERLAPPED lpOverlapped
);

BOOL WINAPI WriteFile(
  __in         HANDLE hFile,
  __in         LPCVOID lpBuffer,
  __in         DWORD nNumberOfBytesToWrite,
  __out_opt    LPDWORD lpNumberOfBytesWritten,
  __inout_opt  LPOVERLAPPED lpOverlapped
);

Научившись работать с именованными каналами, я создал класс CNamedPipes для того, чтобы скрыть основной низкоуровневый код. Теперь достаточно поместить файл CNamedPipes.mqh в соответствующий каталог (\Include\), включить его в исходный код, и объявить объект CNamedPipe. Класс, разработанный мной, предоставляет несколько основных методов для обработки именованных каналов:

Create(), Connect(), Disconnect(), Open(), Close(), WriteUnicode(), ReadUnicode(), WriteANSI(), ReadANSI(), WriteTick(), ReadTick()

Этот класс может быть в дальнейшем расширен в соответствии с дополнительными требованиями.

Метод Create() пытается создать канал с заданным именем. Для упрощения соединения между терминалами используется входной параметр 'account' - это номер клиентского счета, который будет использовать канал. Если номер счета не указан, метод пытается открыть канал с текущим номером клиентского счета. Функция Create() возвращает true в случае успешного создания канала.

//+------------------------------------------------------------------+
// Создает новый именованный канал (Named Pipe)
// Параметры:  account - номер счета терминала  
// Возвращает: true - если успешно, иначе false 
//+------------------------------------------------------------------+
bool CNamedPipe::Create(int account=0)
  {
   if(account==0)
      pipeNumber=IntegerToString(AccountInfoInteger(ACCOUNT_LOGIN));
   else
      pipeNumber=IntegerToString(account);

   string fullPipeName=pipeNamePrefix+pipeNumber;

   hPipe=CreateNamedPipeW(fullPipeName,
                          (int)GENERIC_READ|GENERIC_WRITE|(ENUM_PIPE_ACCESS)PIPE_ACCESS_DUPLEX,
                          (ENUM_PIPE_MODE)PIPE_TYPE_RW_BYTE,PIPE_UNLIMITED_INSTANCES,
                          BufferSize*sizeof(ushort),BufferSize*sizeof(ushort),0,NULL);

   if(hPipe==INVALID_HANDLE_VALUE) return false;
   else
      return true;

  }

Метод Connect() ожидает подключения клиента к каналу. В случае успешного подключения к каналу возвращает true.

//+------------------------------------------------------------------+
// Ожидает подключение клиента к каналу
// Возвращает true в случае успешного подключения, иначе false.
//+------------------------------------------------------------------+
bool CNamedPipe::Connect(void)
  {
   if(ConnectNamedPipe(hPipe,NULL)==false)
      return(kernel32::GetLastError()==ERROR_PIPE_CONNECTED);
   else return true;
  }

Метод Disconnect() разрывает соединение сервера с каналом.

//+------------------------------------------------------------------+
// Разрывает соединение сервера с каналом
// Возвращает true если успешно, иначе false    
//+------------------------------------------------------------------+
bool CNamedPipe::Disconnect(void)
  {
   return DisconnectNamedPipe(hPipe);
  }

Метод Open() должен быть использован клиентом, он пытается открыть созданный ранее канал. Возвращает true, если открытие канала было успешно. Возвращает false, если по какой-то причине он не может присоединиться к созданному каналу в течение 5 секунд (таймаут) или если открытие канала не было успешным.

//+------------------------------------------------------------------+
// Открывает канал, созданный ранее
// Параметры: account - номер счета терминала  
// Возвращает: true - если успешно, иначе false 
//+------------------------------------------------------------------+
bool CNamedPipe::Open(int account=0)
  {
   if(account==0)
      pipeName=IntegerToString(AccountInfoInteger(ACCOUNT_LOGIN));
   else
      pipeName=IntegerToString(account);

   string fullPipeName=pipeNamePrefix+pipeName;

   if(hPipe==INVALID_HANDLE_VALUE)
     {
      if(WaitNamedPipeW(fullPipeName,5000)==0)
        {
         Print("Канал "+fullPipeName+" не доступен...");
         return false;
        }

      hPipe=CreateFileW(fullPipeName,GENERIC_READ|GENERIC_WRITE,0,NULL,OPEN_EXISTING,0,NULL);
      if(hPipe==INVALID_HANDLE_VALUE)
        {
         Print("Ошибка открытия канала");
         return false;
        }

     }
   return true;
  }

Метод Close() закрывает хэндл канала.

//+------------------------------------------------------------------+
// Закрывает хэндл канала
// Возвращает 0 если успешно, иначе ненулевое значение  
//+------------------------------------------------------------------+
int CNamedPipe::Close(void)
  {
   return CloseHandle(hPipe);
  }

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

В языке MQL5 строковая переменная хранится в формате Unicode, поэтому естественно было предоставлять Unicode методы, однако, поскольку в MQL5 есть метод UnicodeToANSI, я также реализовал работу с ANSI-строками. Последние два метода осуществляют пересылку и получение объекта MqlTick через именованный канал.

Метод WriteUnicode() записывает сообщение, содержащее символы в формате Unicode. Поскольку каждый символ состоит из двух байт, сообщение отсылается в канал как массив данных типа ushort.

//+------------------------------------------------------------------+
//Записывает строку формата Unicode в канал 
//Параметры: message - строка для отправки  
//Возвращает:количество байт, записанных в канал
//+------------------------------------------------------------------+
int CNamedPipe::WriteUnicode(string message)
  {
   int ushortsToWrite, bytesWritten;
   ushort UNICODEarray[];
   ushortsToWrite = StringToShortArray(message, UNICODEarray);
   WriteFile(hPipe,ushortsToWrite,sizeof(int),bytesWritten,0);
   WriteFile(hPipe,UNICODEarray,ushortsToWrite*sizeof(ushort),bytesWritten,0);
   return bytesWritten;
  }

Метод ReadUnicode() получает массив данных типа ushort и возвращает объект типа string.

//+------------------------------------------------------------------+
//Читает строку формата Unicode из канала 
//Возвращает: строка в формате Unicode (string в MQL5) 
//+------------------------------------------------------------------+
string CNamedPipe::ReadUnicode(void)
  {
   string ret;
   ushort UNICODEarray[STR_SIZE*sizeof(uint)];
   int bytesRead, ushortsToRead;
 
   ReadFile(hPipe,ushortsToRead,sizeof(int),bytesRead,0);
   ReadFile(hPipe,UNICODEarray,ushortsToRead*sizeof(ushort),bytesRead,0);
   if(bytesRead!=0)
      ret = ShortArrayToString(UNICODEarray);
   
   return ret;
  }

Метод WriteANSI() записывает массив данных типа uchar в канал.

//+------------------------------------------------------------------+
//Записывает строку формата ANSI в канал 
//Параметры: message  - строка для отправки  
//Возвращает:количество байт, записанных в канал
//+------------------------------------------------------------------+
int CNamedPipe::WriteANSI(string message)
  {
   int bytesToWrite, bytesWritten;
   uchar ANSIarray[];
   bytesToWrite = StringToCharArray(message, ANSIarray);
   WriteFile(hPipe,bytesToWrite,sizeof(int),bytesWritten,0);
   WriteFile(hPipe,ANSIarray,bytesToWrite,bytesWritten,0);
   return bytesWritten;
  }

Метод ReadANSI() читает массив данных типа uchar из канала и возвращает объект типа string.

//+------------------------------------------------------------------+
//Читает строку формата ANSI из канала 
//Возвращает: строка в формате Unicode (string в MQL5) 
//+------------------------------------------------------------------+
string CNamedPipe::ReadANSI(void)
  {
   string ret;
   uchar ANSIarray[STR_SIZE];
   int bytesRead, bytesToRead;
 
   ReadFile(hPipe,bytesToRead,sizeof(int),bytesRead,0);
   ReadFile(hPipe,ANSIarray,bytesToRead,bytesRead,0);
   if(bytesRead!=0)
      ret = CharArrayToString(ANSIarray);
   
   return ret;
  }

Метод WriteTick() записывает один объект типа MqlTick в канал.

//+------------------------------------------------------------------+
//Записывает MqlTick в канал. 
//Параметры: outgoing  - MqlTick для отправки  
//Возвращает: true если тик был записан корректно, иначе false 
//+------------------------------------------------------------------+
int CNamedPipe::WriteTick(MqlTick &outgoing)
  {
   int bytesWritten;

   WriteFile(hPipe,outgoing,MQLTICK_SIZE,bytesWritten,0);

   return bytesWritten;
  }

Метод ReadTick() читает один объект типа MqlTick из канала. Если канал пуст, он возвращает 0, иначе должен вернуть количество байт объекта типа MqlTick.

//+------------------------------------------------------------------+
// Читает MqlTick из канала
// Возвращает true если данные были успешно прочитаны, иначе false
//+------------------------------------------------------------------+
int CNamedPipe::ReadTick(MqlTick &incoming)
  {
   int bytesRead;

   ReadFile(hPipe,incoming,MQLTICK_SIZE,bytesRead,NULL);

   return bytesRead;
  }
//+------------------------------------------------------------------+

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

3. Серверный скрипт для получения котировок

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

//+------------------------------------------------------------------+
//|                                          SpeedTestPipeServer.mq5 |
//|                                      Copyright 2010, Investeo.pl |
//|                                                http:/Investeo.pl |
//+------------------------------------------------------------------+
#property copyright "Copyright 2010, Investeo.pl"
#property link      "http:/Investeo.pl"
#property version   "1.00"

#property script_show_inputs
#include <CNamedPipes.mqh>

input int account=0;
bool tickReceived;
uint start,stop;

CNamedPipe pipe;
//+------------------------------------------------------------------+
//| Script program start function                                    |
//+------------------------------------------------------------------+
void OnStart()
  {
   int i=0;
   if(pipe.Create(account)==true)
      if(pipe.Connect()==true)
         Print("Канал подключен");

   do
     {
      tickReceived=pipe.ReadTick();
      if(i==0) start=GetTickCount();
      if(tickReceived==false)
        {
         if(kernel32::GetLastError()==ERROR_BROKEN_PIPE)
           {
            Print("Клиент отсоединился от канала "+pipe.GetPipeName());
            pipe.Disconnect();
            break;
           }
        }
      else i++;
     }
   while(tickReceived==true);
   stop=GetTickCount();

   if(i>0)
     {
      Print(IntegerToString(i)+" тиков получено.");
      i=0;
     };
   
   pipe.Close();
   Print("Сервер: Прием занял "+IntegerToString(stop-start)+" [ms]");

  }
//+------------------------------------------------------------------+

4. Простой индикатор для отправки котировок

Индикатор для отправки котировок внутри OnInit() функции открывает канал и посылает одиночный MqlTick каждый раз при вызове метода OnCalculate().

//+------------------------------------------------------------------+
//|                                        SendTickPipeIndicator.mq5 |
//|                                      Copyright 2010, Investeo.pl |
//|                                                http:/Investeo.pl |
//+------------------------------------------------------------------+
#property copyright "Copyright 2010, Investeo.pl"
#property link      "http:/Investeo.pl"
#property version   "1.00"
#property indicator_chart_window

#include <CNamedPipes.mqh>

CNamedPipe pipe;
int ctx;

//+------------------------------------------------------------------+
//| Custom indicator initialization function                         |
//+------------------------------------------------------------------+
int OnInit()
  {
 
   while (!pipe.Open(AccountInfoInteger(ACCOUNT_LOGIN)))
   {
      Print("Канал не создан, повторная попытка через 5 секунд...");
      if (GlobalVariableCheck("gvar1")==true) break;
   }
   
   ctx = 0;
   return(0);
  }
//+------------------------------------------------------------------+
//| Custom indicator iteration function                              |
//+------------------------------------------------------------------+
int OnCalculate(const int rates_total,
              const int prev_calculated,
              const datetime& time[],
              const double& open[],
              const double& high[],
              const double& low[],
              const double& close[],
              const long& tick_volume[],
              const long& volume[],
              const int& spread[])
  {
   ctx++;
   MqlTick outgoing;
   SymbolInfoTick(Symbol(), outgoing);
   pipe.WriteTick(outgoing);
   Print(IntegerToString(ctx)+" тиков отправлено серверу.");
   return(rates_total);
  }
//+------------------------------------------------------------------+

5. Запускаем на одном терминале тиковые индикаторы от нескольких поставщиков котировок

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

Котировки Bid и Ask посылаются в виде строки, они  разделяются точкой с запятой, например '1.20223;120225'. Соответствующий индикатор обрабатывает пользовательское событие внутри OnChartEvent() и показывает тиковый график.

//+------------------------------------------------------------------+
//|                                   NamedPipeServerBroadcaster.mq5 |
//|                                      Copyright 2010, Investeo.pl |
//|                                                http:/Investeo.pl |
//+------------------------------------------------------------------+
#property copyright "Copyright 2010, Investeo.pl"
#property link      "http:/Investeo.pl"
#property version   "1.00"
#property script_show_inputs
#include <CNamedPipes.mqh>

input int account = 0;

CNamedPipe pipe;
//+------------------------------------------------------------------+
//| Script program start function                                    |
//+------------------------------------------------------------------+
void OnStart()
  {
   bool tickReceived;
   int i=0;

   if(pipe.Create(account)==true)
      while(GlobalVariableCheck("gvar0")==false)
        {
         if(pipe.Connect()==true)
            Print("Канал подключен");
            i=0;
         while(true)
           {
            do
              {
               tickReceived=pipe.ReadTick();
               if(tickReceived==false)
                 {
                  if(kernel32::GetLastError()==ERROR_BROKEN_PIPE)
                    {
                     Print("Клиент отсоединился от канала "+pipe.GetPipeName());
                     pipe.Disconnect();
                     break;
                    }
                  } else  {
                   i++; Print(IntegerToString(i)+" Тиков получено сервером.");
                  string bidask=DoubleToString(pipe.incoming.bid)+";"+DoubleToString(pipe.incoming.ask);
                  long currChart=ChartFirst(); int chart=0;
                  while(chart<100) // Разумеется, открытых графиков у нас меньше чем 100
                    {
                     EventChartCustom(currChart,6666,0,(double)account,bidask);
                     currChart=ChartNext(currChart); // Получили новый график из предыдущего
                     if(currChart==0) break;         // Достигнут конец списка графиков
                     chart++;// не забываем увелививать счетчик
                    }
                     if(GlobalVariableCheck("gvar0")==true || (kernel32::GetLastError()==ERROR_BROKEN_PIPE)) break;
              
                 }
              }
            while(tickReceived==true);
            if(i>0)
              {
               Print(IntegerToString(i)+" тиков получено.");
               i=0;
              };
            if(GlobalVariableCheck("gvar0")==true || (kernel32::GetLastError()==ERROR_BROKEN_PIPE)) break;
            Sleep(100);
           }

        }


  pipe.Close(); 
  }
//+------------------------------------------------------------------+

Для показа тиков я выбрал тиковый индикатор, опубликованный в MQLmagazine, но вместо метода OnCalculate() я реализовал обработку внутри OnChartEvent() и добавил инструкции проверки условий. Котировка принимается для обработки только в случае, если параметр dparam равен номеру канала, а id события равно CHARTEVENT_CUSTOM+6666:

void OnChartEvent(const int id,
                const long &lparam,
                const double &dparam,
                const string &sparam)
  {
  if (dparam==(double)incomingPipe)
   if(id>CHARTEVENT_CUSTOM)
     {
      if(id==CHARTEVENT_CUSTOM+6666)
        {
        // обработка приходящего тика
        }
     } else
        {
         // обработка пользовательских событий 
        }
  }

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

Рисунок 1. Получение котировок через именованный канал

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

Рисунок 2. Видеоролик, объясняющий установку индикаторов

6. Тестирование пропускной способности системы

Поскольку каналы используют коллективную память (shared memory), то связь очень быстрая. Я провел испытания по отправке 100 000 и 1 000 000 тиков за раз между двумя терминалами MetaTrader 5.

Скрипт отсылки тиков использует функцию WriteTick() и измеряет промежуток затраченного времени при помощи функции GetTickCount().

   Print("Отправка...");
   uint start = GetTickCount();
   for (int i=0;i<100000;i++)
      pipe.WriteTick(outgoing);
   uint stop = GetTickCount();
   Print("Отправка заняла " + IntegerToString(stop-start) + " [ms]");
   pipe.Close();

Сервер читает поступающие котировки. Промежуток времени измеряется между временем прихода первой котировки и временем отсоединения клиента:

//+------------------------------------------------------------------+
//|                                          SpeedTestPipeServer.mq5 |
//|                                      Copyright 2010, Investeo.pl |
//|                                                http:/Investeo.pl |
//+------------------------------------------------------------------+
#property copyright "Copyright 2010, Investeo.pl"
#property link      "http:/Investeo.pl"
#property version   "1.00"

#property script_show_inputs
#include <CNamedPipes.mqh>

input int account=0;
bool tickReceived;
uint start,stop;

CNamedPipe pipe;
//+------------------------------------------------------------------+
//| Script program start function                                    |
//+------------------------------------------------------------------+
void OnStart()
  {
   int i=0;
   if(pipe.Create(account)==true)
      if(pipe.Connect()==true)
         Print("Канал подключен");

   do
     {
      tickReceived=pipe.ReadTick();
      if(i==0) start=GetTickCount();
      if(tickReceived==false)
        {
         if(kernel32::GetLastError()==ERROR_BROKEN_PIPE)
           {
            Print("Клиент отсоединился от канала "+pipe.GetPipeName());
            pipe.Disconnect();
            break;
           }
        }
      else i++;
     }
   while(tickReceived==true);
   stop=GetTickCount();

   if(i>0)
     {
      Print(IntegerToString(i)+" тиков получено.");
      i=0;
     };
   
   pipe.Close();
   Print("Сервер: Прием занял "+IntegerToString(stop-start)+" [ms]");

  }
//+------------------------------------------------------------------+

Результаты 10 запусков были следующие:

Запуск	Кол-во котировок	Время отсылки  [ms]	Время получения  [ms]
1	100000	624	624
2	100000	702	702
3	100000	687	687
4	100000	592	608
5	100000	624	624
6	1000000	5616	5616
7	1000000	5788	5788
8	1000000	5928	5913
9	1000000	5772	5756
10	1000000	5710	5710

Таблица 1 Измерения пропускной способности

Средняя скорость отсылки 1 000 000 котировок была около 170 000 тиков/сек на ноутбуке с Windows Vista с 2.0GHz T4200 CPU и 3Gb памяти.

Выводы

Представлен метод взаимодействия между терминалами MetaTrader 5 при помощи именованных каналов. Метод оказался достаточным для отправки котировок в реальном времени между терминалами. 

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