Параллельное и распределенное программирование на С++

protected:

ostream* _M_stream;

const char* _M_string; public:

typedef output_iterator_tag iterator_category;

typedef void value_type;

typedef void difference_type;

typedef void pointer;

typedef void reference;

ostream_iterator(ostream& _s) : _M_stream(&_s),_M_string(0) {}

ostream_iterator(ostream& _s, const char* _с): _M_s tream (&_s) , _M_string (_с) { }

ostream_iterator<_Tp>& operator=(const _Tp& _value) {

*_M_stream << _value;

if (_M_string){

*_M_stream << _M_string;

return *this;

}

ostream_iterator<_Tp>& operator* { return *this; }

ostream_iterator<_Tp>& operator++ { return *this; }

ostream_iterator<_Tp>& operator++(int) { return *this; }

};

Конструктор класса ostream_iterator принимает ссылку на объект класса ostream. Класс ostream_iterator находится с классом ostream в отношении агрегирования. Назначение класса istream_iterator прямо противоположно классу ostream_iterator. Он используется с объектами класса istream (а не с объектами класса ostream). Если объекты классов istream_iterator и ostream_iterator связаны с iostream-объектами, которые в свою очередь связаны с файловыми дескрипторами канала, то при каждом инкрементировании итератора типа istream_iterator из канала будут считываться данные, а при каждом инкрементировании итератора типа ostream_iterator в канал будут записываться данные. Чтобы продемонстрировать, как эти компоненты работают вместе, рассмотрим две программы (11.2 и 11.2.1), в которых используются анонимные каналы связи. Про-грамма11.2 представляет родительский процесс, а программа11.2.1— сыновний. В»родительской» части для создания сыновнего процесса используются системные функции fork и execl . При том, что файловые дескрипторы наследуются сыновним процессом, их значения незамедлительно становятся достоянием программы 11.2.1 благодаря вызовуфункции execl .

// Программа 11.2

10 int main(int argc, char *argv[])

11 {

12

13 int Size,Pid,Status,Fdl[2],Fd2[2];

14 pipe(Fdl); pipe(Fd2);

15 strstream Buffer;

16 char Value[50];

17 float Data;

18 vector<float>X(5,2.1221), Y;

19 Buffer « Fdl[0] « ends;

20 Buffer » Value;

21 setenv(«Fdin»,Value,l);

22 Buffer.clear;

23 Buffer « Fd2[l] « ends;

24 Buffer » Value;

25 setenv(«Fdout»,Value,l);

26 Pid = fork;

27 if(Pid != 0){

28 ofstream OPipe;

29 OPipe.attach(Fdl[l] ) ,-

30 ostream_iterator<float> OPtr(OPipe,"\n»);

31 OPipe « X.size « endl;

32 copy(X.begin,X.end,OPtr);

33 OPipe « flush;

34 ifstream IPipe;

35 IPipe.attach(Fd2[0]);

36 IPipe » Size;

37 for(int N = 0; N < Size;N++)

38 {

39 IPi ре » Data;

40 Y.push_back(Data);

41 }

42 wait(&Status);

43 ostream_iterator<float> OPtr2(cout,"\n»);

44 copy(Y.begin,Y.end,OPtr2);

45 OPipe.close;

46 IPipe.close;

47 }

48 else{

49 execl("./programll-2b»,«programll-2b»,NULL);

50 } 51

52 return(0);

53 }

В строках 21 и 25 системнал функция setenv используется для передачи значений файловых дескрипторов сыновнему процессу. Это возможно благодаря тому, что сыновний процесс наслелует среду родительского процесса. Мы можем устанавливать переменные среды в программе с помощью вызова функции setenv . В данном случае мы устанавливаем их следующим образом.

Fdin=filedesc; Fdout=filedesc;

Сыновний процесс затем использует системный вызов getenv( ) для считывания значений переменных Fdin и Fdout. Значение переменной Fdin будет представлять «считывающий конец» канала для сыновнего процесса, а значение переменной Fdout — «записывающий». Использование системных функций setenv и getenv обеспечивает просгую форму межпроцессного взаимодействия (interprocess communication — IPC) между родительским и сыновним процессами. Каналы создаются при выполнении инструкций, приведенных в строке 14. Родительский процесс присоединяется к одному концу канала для операции записи с помощью метода attach (строка29). После присоединения любые данные, помещенные в объект OPipe типа ofstream, будут записаны в канал. Итератор типа ostream_iterator подключается к объекгу OPipe при выполнении следующей инструкции (строка 30):

ostream_iterator<float> OPtr(OPipe,"\n»);

Теперь итератор OPtr ссылается на объект OPipe. После каждой порции помещаемых в канал данных будет вставляться разделитель "\n». С помощью итератора OPtr мы можем поместить в канал любое количество float– значений. При этом мы можем связать с каналом несколько итераторов различных типов. Но в этом случае необходимо, чтобы на «считывающем» конце канала данные извлекались с использованием ите раторов соответствующих типов. При выполнении слелующей инструкции из программы 11.2 в канал сначала помещается количество элементов, подлежащих передаче: OPipe « X.size « endl;

Сами элементы отправляются с использованием одного из стандартных С++-алгоритмов:

copy(X.begin ,X.end ,OPtr) ;

Алгоритм copy копирует содержимое одного контейнера в контейнер, связанный с итератором приемника. Здесь итератором приемника является объект OPtr. Объект OPtr связан с объектом OPipe, поэтому при выполнении алгоритма copy («уместившегося» в одной строке кода) в канал переписывается все содержимое контейнера. Этот пример демонстрирует возможность использования стандартных алгоритмов для организации взаимодействия между различными частями сред параллельного или распределенного программирования. В данном случае алгоритм copy пересылает информацию от одного процесса другому (из одного адресного пространства в другое). Эти процессы выполняются параллельно, и алгоритм copy значительно упрощает взаимодействие между ними. Мы подчеркиваем важность этого подхода, поскольку, если есть хоть какал-то возможность упростить логику параллельной или распределенной программы, ею нужно непременно воспользоваться. Ведь межпроцессное взаимодействие — это один из самых сложных разделов параллельного или распределенного программирования. С++-алгоритмы, библиотека классов iostreamS и итератор типа ostream_iterator как раз и позволяют понизить уровень сложности разработки таких программ. Использование манипулятора flush (в строке 33) гарантирует прохождение данных по каналу.

В программе 11.2.1 сыновний процесс сначала получает количество объектов, принимаемых от канала (в строке 36), а затем для считывания самих объектов использует объект IPipe класса istream.

// Программа 11.2.1

11 class multiplier{

12 float X;

13 public:

14 multiplier(float Value) { X = Value;}

15 float &operator(float Y) { X = (X * Y);return(X);}

16 }; 17

18

19 int main(int argc,char *argv[])

20 {

21 char Value[50] ;

22 int Fd[2] ;

23 float Data;

24 vector<float> X;

25 int NumElements;

26 multiplier N(12.2);

27 strcpy(Value,getenv(«Fdin»));

28 Fd[0] = atoi(Value);

29 strcpy(Value,getenv(«Fdout»));

30 Fd[l] = atoi(Value);

31 ifstream IPipe;

32 ofstream OPipe;

33 IPipe.attach(Fd[0]) ;

34 OPipe.attach(Fd[l]) ;

35 ostream_iterator<float> OPtr(OPipe,"\n»);

36 IPipe » NumElements;

37 for(int N = 0;N < NumElements;N++)

38 {

39 IPipe » Data;

40 X.push_back(Data);

41 }

42 OPipe « X.size « endl;

43 transform(X.begin,X.end,OPtr,N);

44 OPipe « flush;

45 return(0); 46

47 }

Сыновний процесс считывает элементы данных из канала, помещает их в вектор, азатем выполняет математические преобразования над каждым элементом вектора, после чего отправляет их назад родительскому процессу. Математические преобразования (строка43) выполняются с использованием стандартного С++-алгоритма transform и пользовательского класса multiplier. Алгоритм transform применяет к каждому элементу контейнера операцию, а затем результат этой операции помещает в контейнер-приемник. В данном случае контейнером-приемником служит объект Optr, который связан с объектом OPipe. Заголовки, которые необходимо включить в программу 11.2.1, приведены в разделе «Профиль программы 11.2.1».