Си

- Добрый день, я на интервью на позицию старшего разработчика.

- Здравствуйте, давайте начнем с небольшого теста, пока я ваше CV смотрю. Напишите программу, которая выводила бы числа от 1 до, скажем, миллиарда, притом если число кратно трем, то вместо числа выводится Fizz, если кратно пяти, то Buzz, а если и трем, и пяти, то FizzBuzz.

Серьезно, FizzBuzz? Задачка для начальной школы, на сениорскую позицию? Ну ладно.

Я достаю свой верный лаптоп, и пишу такой код:

#include <stdio.h>#define LIMIT 1000000000int main(void) {    for (int i = 1; i <= LIMIT; i++) {        if (0 == i % 3) {            if (0 == i % 5) {                printf("FizzBuzz\n");            } else {                printf("Fizz\n");            }        } else if (0 == i % 5) {            printf("Buzz\n");        } else {            printf("%d\n", i);        }    }    return 0;}

Запускаю программу, она себе бежит, но не так чтобы сильно быстро, через 3 с чем-то минуты, после первого миллиона, я ее прерываю и быстренько высчитываю, что весь процесс займет больше двух суток. Да, наверно надо было включить буферизацию, это бы несколько ускорило, но это не спасет, лучше просто перенаправить вывод в файл, что я и делаю, и через 41.5 секунду у меня есть красивенький файл на 7.5 гигов.

- Вам не кажется, что можно побыстрее? - спрашивает интервьюер.

- Да ладно, основное время занимает I/O, 7.5 гигов записать не шутка, даже на SSD.

- А давайте перенаправим вывод в /dev/null.

- Без проблем.

Через минуту:

- Как это 39.5 секунд? То есть весь I/O занимает 2 секунды, а все остальное время мой код?

- Да, так получается. Это не самая медленная реализация, на каждой итерации два сравнения и один printf, я часто вижу вариант с тремя сравнениями и двумя printfами. Для джуниора, я бы сказал, это даже хорошо. А вот для сениора ...

Это было больно, но, пожалуй, заслужено. Ладно, я тебе покажу, кто тут джуниор.

- Сейчас сделаю побыстрее.

- Попробуйте. Только объясняйте, что вы делаете.

- Видите, что у нас тут есть паттерн каждые 3*5, то есть 15 итераций цикла логика полностью повторяется. Тогда можно переделать цикл:

    for (i = 1; i < LIMIT - 15; i += 15) {        printf( "%d\n"          // 1                "%d\n"          // 2                "Fizz\n"        // 3                "%d\n"          // 4                "Buzz\n"        // 5                "Fizz\n"        // 6                "%d\n"          // 7                "%d\n"          // 8                "Fizz\n"        // 9                "Buzz\n"        // 10                "%d\n"          // 11                "Fizz\n"        // 12                "%d\n"          // 13                "%d\n"          // 14                "FizzBuzz\n",   // 15                i, i+1, i+3, i+6, i+7, i+10, i+12, i+13);    }

- Если раньше на каждые 15 чисел у нас приходилось 15 сравнений переменной цикла и два ifа в теле цикла, то есть в общей сложности 45 сравнений, а каждое сравнение это потенциальная проблема с branch predictionом, то теперь одно. Да и вызовов printfа стало в 15 раз меньше. Одна только проблема цикл не дойдет ровно до миллиарда, а только до 999999990 (макс число, кратное 15 и меньшее миллиарда), так что оставим старый цикл, но только для обработки хвоста, то есть последних 10 значений (это практически не влияет на производительность).

После всех изменений получился такой код.

- И что у нас со временем получается?

- Если вывод в файл, то 22.5 секунды, если в /dev/null 20.2

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

- Я думаю, что это не предел.

- В самом деле? А что тут можно еще оптимизировать?

- Я уменьшил количество вызовов printfа в 15 раз, но при этом сами эти printfы стали тяжелее. Да и вообще printf сам по себе тяжелый, из-за своей мощности это ведь фактически виртуальная машина со своим языком, полным по Тьюрингу, на нем даже крестики-нолики писали. В данной ситуации используется лишь небольшая часть возможностей printf, так что можно его заменить на что-то свое, более легкое:

#define NUM cur += myitoa(num++, cur)#define FIZZ do { memcpy(cur, "Fizz\n", 5); cur += 5; num++; } while (0)#define BUZZ do { memcpy(cur, "Buzz\n", 5); cur += 5; num++; } while (0)#define FIZZBUZZ do { memcpy(cur, "FizzBuzz\n", 9); cur += 9; } while (0)void print(int num) {    static char wrkbuf[CHUNK_SIZE];    char *cur = wrkbuf;    NUM;    NUM;    FIZZ;    NUM;    BUZZ;    FIZZ;    NUM;    NUM;    FIZZ;    BUZZ;    NUM;    FIZZ;    NUM;    NUM;    FIZZBUZZ;    fwrite(wrkbuf, cur - wrkbuf, 1, stdout);}

- Можно, конечно, использовать уже готовую itoa, но это нестандартная функция, не везде есть, да и она слишком универсальная, поскольку поддерживает разные системы счисления, а у нас только десятичная система - упрощаем все, что можно. Ну и, конечно, в главном цикле просто вызываем print(i) вместо длинного printfа.

Получается такой код.

Я подхожу к доске и рисую табличку с результатами запусков:

- В принципе на вывод в файл можно особо не смотреть там какое-то время съедается на I/O, и оно плавает, так что лучше ориентироваться на время без I/O.

Я стираю ту часть, где про вывод в файл.

- Итого ускорение в 4 с половиной раза.

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

- Я знаю, как можно еще ускорить.

- Серьезно?

- Абсолютно. Я до этого использовал чисто технические способы ускорения, а ведь можно еще и алгоритмически улучшить. Смотрите, что будет напечатано, когда мы вызываем, например, print(150000001) и следующий за ним print(150000016):

150000001\n150000002\nFizz\n150000004\nBuzz\nFizz\n150000007\n150000008\nFizz\nBuzz\n150000011\nFizz\n150000013\n150000014\nFizzBuzz\n150000016\n150000017\nFizz\n150000019\nBuzz\nFizz\n150000022\n150000023\nFizz\nBuzz\n150000026\nFizz\n150000028\n150000029\nFizzBuzz\n       ^^         ^^               ^^                     ^^         ^^                     ^^               ^^         ^^ 

- Отличия всего в 16 байтах, а программа всю строку пересобирает с нуля. Можно просто менять байты на месте. Правда, заранее неизвестно, сколько десятичных разрядов надо поменять, так что это потребуется вычислить сравнить два буфера, и определить, где они отличаются. Это, пожалуй, тяжеловатая задача, но у нас есть, я делаю театральную паузу векторные инструкции и интринсики для них!

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

Открываю Intelовскую страницу с интринсиками и нахожу там нужные векторные функции для работы с 16-байтными векторами. У меня тут максимум 10-байтные, но их можно добить нулями до 16, не проблема. И да, 16-байтные вектора это SSE инструкции, никакой AVX-512 тут не нужен, мой 4-летний мобильный проц это точно потянет.

Получаю такой кусок с жирными и вкусными интринсиками:

unsigned int diff = 0xFFFF & ~_mm_movemask_epi8(_mm_cmpeq_epi8(                                  _mm_load_si128((__m128i const *)prev_first_number),                                  _mm_load_si128((__m128i const *)last_number)));unsigned int diff_pos = 16 - _tzcnt_u32(diff);   // number of changed digits

Быстрая проверка flags в /proc/cpuinfo нужные для выбранных мной интринсиков SSE2 (еще со времен Pentium 4) и BMI1 (появился в Haswellах) в CPU есть, все должно работать.

Запускаю тот код, что получился, смотрю уже только вывод в /dev/null и обновляю табличку:

Еще почти в 2 раза ускорились! А по сравнению с начальным вариантов так вообще почти в 9. Жаль, до 10 раз не дотянул.

- Ну все, наверно теперь уже хватит. Это уже вполне по-сениорски.

Во взгляде интервьюера читается облегчение.

- Скорее всего, мы вплотную подошли к пределу того, что можно выжать из однопоточной программы, - говорю я, медленно подвисая к концу фразы. Как же я об этом раньше не подумал! - Я ведь еще многопоточность не попробовал!

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

- Я буду выделять каждому рабочему потоку кусок числового поля, и этот поток будет возвращать готовый буфер для своего куска, а главный поток будет печатать эти буферы, соблюдая порядок:

for (int j = 0; j < THREAD_COUNT; j++) {        thread_pool[j].start_num = i;        thread_pool[j].count = NUMS_PER_THREAD;        thread_pool[j].buf = malloc(BUFFER_SIZE);        pthread_create(&thread_pool[j].id, NULL, worker, (void *)&thread_pool[j]);        i += NUMS_PER_THREAD;    }    int active_threads = THREAD_COUNT;    int max = LIMIT / 15 * 15;    for (int j = 0; active_threads; j = (j+1) % THREAD_COUNT) {        pthread_join(thread_pool[j].id, NULL);        fwrite(thread_pool[j].buf, thread_pool[j].buflen, 1, stdout);        if (max - i > NUMS_PER_THREAD) {            thread_pool[j].start_num = i;            pthread_create(&thread_pool[j].id, NULL, worker, (void *)&thread_pool[j]);            i += NUMS_PER_THREAD;        } else if (max > i) {            thread_pool[j].start_num = i;            thread_pool[j].count = max - i + 1;            pthread_create(&thread_pool[j].id, NULL, worker, (void *)&thread_pool[j]);            i += max - i + 1;        } else {            free(thread_pool[j].buf);            active_threads--;        }    } 

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

Проведя несколько замеров, я остановился на кусках по 3 миллиона чисел удобное число, кратное 15, и результат хороший.

Получился такой код.

Запускаю, и обновляю данный в табличке:

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

Я отвернулся от доски и увидел, что в переговорке я один. Через стеклянную стенку переговорки я разглядел интервьюера, который что-то быстро говорил секретарю, показывая пальцем в мою сторону. Он что, вызывает охрану? Похоже, что мне тут больше не рады.

Я быстро закрыл лаптоп и покинул офис. Почему-то мне так и не перезвонили.

Все тесты делались на Dell Latitude 7480 с i7-7600U 2.8 Ghz, 16 Gb памяти, SSD и OpenSUSE Leap 15.1 с kernelом 4.12.14, каждый тест не менее 10 раз, выбиралось наименьшее значение. При компиляции использовались флаги -O3 -march=native -pthread

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

Приветствую, в этой статье вы узнаете: как сделать датчик движения с использованием ИК диода и ИК приемника на STM32 с минимальным использованием ядра (т.е. с максимальной загрузкой периферии) на регистрах, используя таймеры.

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

Оглавление:

Общая картина

Схемы устройств

Работа ИК диода и ИК приемника

Код для передатчика ИК сигнала

Код для приемника ИК сигнала

Полный код

Дополнительно

Заключение

Общая картина

Используются два устройства: передатчик ИК сигнала и приемник ИК сигнала.

Общий принцип работы следующий: передатчик ИК сигнала излучает сигнал инфракрасного диапазона длин волн, а приемник ИК сигнала их принимает. Между этими устройствами образуется луч, пересечение которого каким-либо объектом фиксируется приемным устройством.

В качестве ИК диода используется TSAL6200, в качестве ИК приемника TSOP4856.

TSAL6200	TSOP4856

Схемы устройств

Опустим цепи питания и другие элементы реальных устройств и рассмотрим только необходимые.

Передатчик ИК сигнала состоит из микроконтроллера STM32L151C8T6, полевого транзистора 2N7002, резистора 1 Ом и ИК диода TSAL6200. Ниже представлена электрическая схема передатчика.

Транзистор необходим для усиления тока, протекающего через ИК диод, так как выходной ток с пина МК ограничен (выходной ток с пина МК STM32L151C8T6 не более 25 мА, максимальный постоянный ток через ИК диод TSAL6200 100 мА).

Выбор транзистора на ваше усмотрение. Здесь выбран транзистор 2N7002, потому что он дешевый и его характеристик достаточно для моего использования. Стоит выбирать транзистор с меньшим пороговым напряжением затвора (Gate Threshold Voltage), так как на затвор вы сможете подать напряжение не более напряжения питания без дополнительных цепей, в нашем случае 3.3 В.

Точный расчет величины сопротивления резистора является трудным, конечно, существуют специальные формулы, но я предлагаю подобрать резистор опытным путем. Так в моем случае используется резистор 1 Ом, амплитуда тока цепи составляет 20 мА, дальность связи при таком токе достигает 14 м в зависимости от конфигурации передаваемого сообщения (количество импульсов в пачке, скважность, об этом ниже). Если вам требуется большая дальность, следует или подобрать резистор поменьше, или увеличить напряжение питания для диода, или выбрать транзистор с меньшим пороговым напряжением и меньшим падением напряжения на сток-истоке.

Приемник ИК сигнала состоит из микроконтроллера STM32L151C8T6, ИК приемника TSOP4856, резистора 100 Ом и конденсатора 0.1 мкФ. Ниже представлена электрическая схема приемника.

Работа ИК диода и ИК приемника

При протекании тока через диод TSAL6200 излучается ИК сигнал с длиной волны 940 нм. Это излучение необходимо промодулировать меандром с частотой 56 кГц (приемник настроен на данную частоту, см. даташит). Передавать следует пачки таких импульсов с определенной скважностью.

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

Остается только запрограммировать работу передатчика и приемника.

Код передатчика ИК сигнала

Будем использовать таймеры для формирования пачек импульсов с выводом сигнала на конкретный пин. Таким образом, ядро МК только инициализирует эти таймеры, а дальше может заниматься какой-нибудь другой работой.

Выбор пина должен быть обоснован. Необходимо выбирать такой пин, к которому может быть подключен какой-либо канал (кроме четвертого) любого General-purpose таймера. В нашем случае выбран пин PB6, к нему подключается первый канал TIM4. Приемник же будет подключен к пину PB7, второй канал TIM4.

Будем использовать два таймера: TIM4 будет генерировать меандр с частотой 56 кГц, TIM2 будет управлять таймером TIM4, т.е. создавать пачки из меандра. TIM2 будет работать в режиме Master, TIM4 Slave. Почему именно таймер TIM2? Он выбран исходя из таблицы даташита подключений таймеров друг к другу.

Приступим к написанию функции инициализации таймеров Tim_Init_Transmitter(). Код написан для пустого проекта. Сначала объявляем функцию, в теле main её инициализируем, а после мэйна прописываем уже саму функцию.

#include "main.h"void Timer_Init_Transmitter(void);int main(void){   RCC->ICSCR |= RCC_ICSCR_MSIRANGE_6;   // MSI 4.194 MHz enable   Timer_Init_Transmitter();   while(1)   {   }}void Timer_Init_Transmitter(void){}

RCC->ICSCR |= RCCICSCRMSIRANGE_6 - эта строчка просто устанавливает частоту 4.194 МГц МК. У вас может быть любая другая частота.

Настраиваем пин. Включаем тактирование, включаем альтернативную функцию, выбираем скорость пина повыше (это не является необходимым), выбираем какую именно альтернативную функцию хотим подключить к пину. Под спойлерами описано все подробно.

В даташите находим регистр RCC_AHBENR. Нужно прописать 1 в поле GPIOBEN.

Для этого выбираем в библиотеке CMSIS строчку RCC_AHBENR_GPIOBEN, которая ставит данный бит в единицу.

И изменяем регистр AHBENR следующим образом:

RCC->AHBENR |= RCC_AHBENR_GPIOBEN; //GPIO port B clock enable

В даташите находим регистр GPIOx_MODER. Нужно прописать 10 в поле MODER6 (для пина PB6).

Для этого выбираем в библиотеке CMSIS строчку GPIO_MODER_MODER6_1, которая ставит второй бит в единицу.

В дальнейшем все такие обозначения будут браться аналогично.

И изменяем регистр MODER следующим образом:

GPIOB->MODER |= GPIO_MODER_MODER6_1;//Alternative function mode enable

Включение высокой скорости для пина делается аналогично включению альтернативной функции.

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

Нам нужна альтернативная функция для TIM4 это AF2.

Стоит отметить, что существует два регистра для выбора альтернативных функций: нижний GPIOx_AFRL для пинов с номерами от 0 до 7 и верхний GPIOx_AFRH для пинов с номерами от 8 до 15. Однако, в библиотеке CMSIS определен сдвоенный регистр AFR[2], прописав в скобки 0, мы выбираем нижний регистр, прописав 1, выбираем верхний регистр.

В даташите находим регистр GPIOx_AFRL.

Нам нужно прописать 0010 в поле AFRL6, для этого нам нужно число в шестнадцатеричной форме 0x2000000, 2 седьмая справа для пина PB6, потому что первые шесть цифр для пинов с номерами от 0 до 5.

И изменяем регистр AFRL следующим образом:

GPIOB->AFR[0] |= 0x2000000; //Pin PB6 TIM4 alternative function AF2 enable

Получили следующий код для настройки пина внутри нашей функции:

void Timer_Init_Transmitter (void){   //Settings for GPIO PB6   RCC->AHBENR |= RCC_AHBENR_GPIOBEN;             //GPIO port B clock enable   GPIOB->MODER |= GPIO_MODER_MODER6_1;           //Alternative function mode enable   GPIOB->OSPEEDR |= GPIO_OSPEEDER_OSPEEDR6_1;  //High speed   GPIOB->AFR[0] |= 0x2000000;                    //Pin PB6 TIM4 alternative function AF2 enable}

Теперь перейдем к настройке TIM4.

Нам нужно выбрать прескейлер PSC (делитель частоты), рассчитать значения для регистров CCR1 (длительность импульса) и ARR (период импульсов), выбрать режим работы таймера ШИМ, выбрать входной управляющий сигнал от таймера TIM2, выбрать режим для Slave и подать выходной сигнал таймера на пин PB6.

Включение тактирования осуществляется аналогично включению тактирования GPIO.

Выберем значение преселлектора входной частоты равное 0, в этом случае частота тактирования таймера будет равна частоте МК.

И изменяем регистр PSC следующим образом:

TIM4->PSC = 0; //Prescaler value

Перейдем к расчету ARR.

Для расчета ARR нам нужно поделить тактовую частоту таймера 4.194 МГц (у вас своя) на 56 кГц. Получаем 74,89, округляем до целого. Я округлил до 75. И вписываем в регистр ARR:

TIM4->ARR = 75 //Auto-reload value

Осталось рассчитать CCR1.

Так как у нас простой меандр, то в регистр CCR1 при режиме ШИМ нужно указать половину от значения ARR:

TIM4->CCR1 = 37;//Capture/Compare 1 value

Так как у нас первый канал, то используем регистр CCMR1. Нас интересует в этом регистре поле OC1M. Режим ШИМ позволяет настраивать длительность импульса при фиксированном периоде.

Выберем PMW mode 1, вписав две единицы во второй и третий биты, т.е. 110:

TIM4->CCMR1 |= TIM_CCMR1_OC1M_1 | TIM_CCMR1_OC1M_2; //Output compare PMW mode 1

В регистр CCER в поле CC1E нужно вписать 1, включив подачу выходного сигнала на пин.

Вписываем следующую строчку:

TIM4->CCER |= TIM_CCER_CC1E; //OC3 signal is output on the corresponding pin

Для управления TIM4 таймером TIM2 нужно выбрать входной сигнал ITR1. Для этого нужно вписать в поле TS регистра TIMx_SMCR 001. Также выберем режим для Slave, вписав 101 в поле SMS. В этом режиме пока входной сигнал ITR1 будет высоким, то TIM4 будет работать и выдавать меандр на пин, как только ITR1 станет низким, TIM4 выключается.

Для этого вписываем:

TIM4->SMCR |= TIM_SMCR_TS_0;//choosing ITR1

TIM4->SMCR |= TIM_SMCR_SMS_0 | TIM_SMCR_SMS_2; //Gated Mode

Добавим к уже существующему коду, и получим:

void Timer_Init_Transmitter (void){   //Settings for GPIO PB6   RCC->AHBENR |= RCC_AHBENR_GPIOBEN;             //GPIO port B clock enable   GPIOB->MODER |= GPIO_MODER_MODER6_1;           //Alternative function mode enable   GPIOB->OSPEEDR |= GPIO_OSPEEDER_OSPEEDR6_1;  //High speed   GPIOB->AFR[0] |= 0x2000000;                    //Pin PB6 TIM4 alternative function AF2 enable   //Settings for TIM4 - Slave   RCC->APB1ENR |= RCC_APB1ENR_TIM4EN;            //TIM4 clock enable   TIM4->PSC = 0;                                 //Prescaler value   TIM4->ARR = 75;                                //Auto-reload value   TIM4->CCR1 = 37;                               //Capture/Compare 1 value   TIM4->CCMR1 |= TIM_CCMR1_OC1M_1 | TIM_CCMR1_OC1M_2; //Output compare PMW mode 1    enable   TIM4->CCER |= TIM_CCER_CC1E;                   //OC3 signal is output on the corresponding output pin   TIM4->SMCR |= TIM_SMCR_TS_0;                   //choosing ITR1   TIM4->SMCR |= TIM_SMCR_SMS_0 | TIM_SMCR_SMS_2; //Gated Mode   TIM4->CR1 |= TIM_CR1_CEN;                      //TIM4 enable}

Включение TIM4 можно производить, а можно и нет, так как управляющий сигнал с TIM2, всё равно включит его.

Теперь перейдем к настройке TIM2.

Необходимо подать тактирование на таймер, рассчитать PSC, CCR1 и ARR, выбрать режим работы таймера ШИМ, выбрать какой сигнал будет выходным (входным управляющим для TIM4) и включить таймер.

Включение тактирования аналогично, как и для таймера TIM4.

RCC->APB1ENR |= RCC_APB1ENR_TIM2EN;//TIM2 clock enable

Так как по даташиту на диод в пачке следует использовать не менее 10 импульсов (на самом деле можно и меньше, но качество работы может быть не очень хорошим), то выберем тактирование таймера TIM2 в десять раз меньше, чем TIM4.

Как видим в формуле для расчета частоты тактирования таймера в знаменателе уже есть +1, т.е. нам нужно записать в регистр PSC 9, чтобы получить частоту в 10 раз меньшую, чем частоту МК.

TIM2->PSC = 9;//Prescaler value

Теперь рассчитаем значение для CCR1: отправлять будем 10 импульсов, следовательно, нам нужно взять значение ARR для TIM4 (период одного импульса, это 75) и умножить на 10, т.е. получаем 750, однако, у нас есть прескейлер, который делит частоту на 10, т.е. нам нужно поделить 750 на 10, в итоге получаем снова 75 (как и в TIM4, но уже с другим прескейлером). Запишем это значение в регистр CCR1 таймера TIM2.

TIM2->CCR1 = 75;//Capture/Compare 1 value

Перейдём к расчету ARR: тут всё просто, допустим, хотим стрелять пачками со скважностью 11.2, при этом период излучения пачек будет около 2 мс (из расчета, что 1 мс это 4194000/1000 = 4194 тика, и умножаем на 2, получаем округлённо 8400, а с прескейлером 10, получаем 840 тиков), умножаем длительность пачки 75 на 11.2 и получаем 840, как видите, значения совпадают. Запишем это в регистр ARR.

TIM2->ARR = 840;//Auto-reload value

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

Режим работы таймера TIM2 точно такой же, как и у TIM4 - ШИМ.

TIM2->CCMR1 |= TIM_CCMR1_OC1M_1 | TIM_CCMR1_OC1M_2; //Output compare PMW mode 1 enable

Найдем регистр TIMx_CR2.

Нам нужно подать сигнал с первого канала (мы же используем CCR1), на выход с таймера TIM2. Этим сигналом является OC1REF. Ниже на картинке можно найти комбинацию битов для этого сигнала 100.

В поле MMS необходимо вписать 1 в третий бит.

TIM2->CR2 |= TIM_CR2_MMS_2;//OC1REF signal is used as trigger output (TRGO)

Теперь осталось включить TIM2, записав следующую строчку:

TIM2->CR1 |= TIM_CR1_CEN;//TIM2 enable

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

Добавим эти строчки к предыдущим и получим:

void Timer_Init_Transmitter (void){   //Settings for GPIO PB6   RCC->AHBENR |= RCC_AHBENR_GPIOBEN;             //GPIO port B clock enable   GPIOB->MODER |= GPIO_MODER_MODER6_1;           //Alternative function mode enable   GPIOB->OSPEEDR |= GPIO_OSPEEDER_OSPEEDR6_1;  //High speed   GPIOB->AFR[0] |= 0x2000000;                    //Pin PB6 TIM4 alternative function AF2 enable   //Settings for TIM4 - Slave   RCC->APB1ENR |= RCC_APB1ENR_TIM4EN;            //TIM4 clock enable   TIM4->PSC = 0;                                 //Prescaler value   TIM4->ARR = 75;                                //Auto-reload value   TIM4->CCR1 = 37;                               //Capture/Compare 1 value   TIM4->CCMR1 |= TIM_CCMR1_OC1M_1 | TIM_CCMR1_OC1M_2; //Output compare PMW mode 1 enable   TIM4->CCER |= TIM_CCER_CC1E;                   //OC3 signal is output on the corresponding output pin   TIM4->SMCR &= ~TIM_SMCR_TS;                    //clear bits   TIM4->SMCR |= TIM_SMCR_TS_0;                   //choosing ITR1   TIM4->SMCR &= ~TIM_SMCR_SMS;                   //clear bits   TIM4->SMCR |= TIM_SMCR_SMS_0 | TIM_SMCR_SMS_2; //Gated Mode   TIM4->CR1 |= TIM_CR1_CEN;                      //TIM4 enable   //Settings for TIM2 - Master   RCC->APB1ENR |= RCC_APB1ENR_TIM2EN;            //TIM2 clock enable   TIM2->PSC = 9;                                 //Prescaler value   TIM2->ARR = 840;                               //Auto-reload value   TIM2->CCR1 = 75;                               //Capture/Compare 1 value   TIM2->CCMR1 |= TIM_CCMR1_OC1M_1 | TIM_CCMR1_OC1M_2; //Output compare PMW mode 1 enable   TIM2->CR2 |= TIM_CR2_MMS_2;                    //OC1REF signal is used as trigger output (TRGO)   TIM2->CR1 |= TIM_CR1_CEN;                      //TIM2 enable}

Теперь у нас ИК диод будет излучать пачки по 10 импульсов, у которых частота 56 кГц, со скважностью 11.2, т.е. с периодом следования пачек в 2 мс. Заметим, что момент пересечения каким-либо объектом внутри периода пачек не определен, т.е. мы можем судить о пересечении ИК луча только по отсутствию следующей пачки. Таким образом, погрешность измерения момента времени пересечения луча составляем 2 мс.

Код приемника ИК сигнала

Для приема ИК сигнала снова будем использовать таймеры, а точнее всего один. В микроконтроллерах STM32 таймеры могут быть не только Master или Slave, а могут быть Master/Slave, т.е. они могут управлять сами собой.

В разделе статьи Код передатчика ИК сигнала мы уже выбрали пин PB7 для приемника ИК сигнала. Схему подключения см. выше. К этому пину подключается второй канал TIM4.

Ниже на картинке представлена схема устройства таймера.

Из схемы видно, что у каждого канала есть вход и выход, у таймера есть входной управляющий сигнал TRGI, также показаны тактирование и устройство Trigger controller.

Принцип работы для нашего проекта заключается в следующем: пусть наш таймер будет просто считать время, допустим, равное 5 периодам излучаемых пачек ИК сигнала, следовательно, в ARR будет число, равное 5 периодам пачек, т.е. 840 * 5 = 4200. Настроим таймер так, чтобы при приеме каждой пачки таймер перезагружался и начинал считать заново. При достижении таймером числа в регистре ARR таймер выдает прерывание, которое означает, что на вход уже как 5 периодов не приходит пачка, следовательно, луч пересекается каким-то объектом. Вот и вся работа таймера. Осталось настроить TIM4.

Создадим новый проект и функцию инициализации таймера:

#iclude "main.h"void Timer_Init_Receiver(void);int main(void){   RCC->ICSCR |= RCC_ICSCR_MSIRANGE_6;    // MSI 4.194 MHz enable   Timer_Init_Receiver();   while(1)   {   }}void Timer_Init_Receiver(void){}

Необходимо настроить пин PB7: включить тактирование порта B, настроить альтернативную функцию пина, настроить скорость и выбрать какую конкретную альтернативную функцию подключить к пину. Всё, как и в предыдущем случае.

void Timer_Init_Receiver(void){//Settings for GPIO PB7RCC->AHBENR |= RCC_AHBENR_GPIOBEN;          // GPIO port B clock enableGPIOB->MODER |= GPIO_MODER_MODER7_1;        // Alternative function mode enableGPIOB->OSPEEDR |= GPIO_OSPEEDER_OSPEEDR7_1;// High speedGPIOB->PUPDR |= GPIO_PUPDR_PUPDR7_0;        // pull-up PB7GPIOB->AFR[0] |= 0x20000000;                // Pin PB7 TIM4 alternative function AF2 enable}

Все настройки аналогичны, как и для передатчика, только альтернативная функция немного отличается, 2 нужно вписать уже на восьмое место справа. Еще необходимо подтянуть пин к плюсу прописав строчку:

GPIOB->PUPDR |= GPIO_PUPDR_PUPDR7_0;

Теперь настроим TIM4.

Включим тактирование таймера. Так как у нас второй канал, то будем использовать CCR2. Прескейлер приравняем к 9, чтобы частота тактирования таймера была как у TIM2 передатчика. Для приемника можно приравнять CCR2 и ARR. Как уже рассчитали ранее, вписываем в эти регистры 4200.

RCC->APB1ENR |= RCC_APB1ENR_TIM4EN;// TIM4 clock enable

TIM4->PSC = 9;// Prescaler value

TIM4->ARR = 4200;// Auto-reload value

TIM4->CCR2 = 4200;// Capture/Compare 2 value

Далее настроим режим работы таймера.

Нам нужно, чтобы он просто считал, и всё. Для этого в даташите находим TIMx_CCMR1 и в поле OC2M вписываем 000, что соответствует Frozen mode. Изменим регистр, обнулив все биты:

TIM4->CCMR1 &= ~TIM_CCMR1_OC2M;// Frozen mode enable

Настроим канал на выход, обнулив биты поля CC2S (более подробно написано в даташите), впишем нули в эти биты:

TIM4->CCMR1 &= ~TIM_CCMR1_CC2S; // Output mode

Теперь обратимся к картинке со схемой устройства таймера (см. выше). Нам необходимо, чтобы входной сигнал с пина PB7 был управляющим для таймера TIM4, таким сигналом является TI2FP2. Проследите путь этого сигнала от TIMx_CH2 до TRGI. Чтобы выбрать данный сигнал в качестве управляющего, необходимо в регистр TIMx_SMCR в поле TS вписать 110. Сразу же выберем режим для Slave: Reset mode, вписав 100 в поле SMS. Для этого пропишем строчки:

TIM4->SMCR |= TIM_SMCR_TS_1 | TIM_SMCR_TS_2; // Choosing TI2FP2

TIM4->SMCR |= TIM_SMCR_SMS_2;// Reset mode

Теперь нам нужно определиться, что будет являться триггером обновления таймера: передний или задний фронт входных импульсов (на самом деле не особо важно, но лучше выбрать по переднему фронту, в нашем случае, это фронт спада напряжения). Для этого нужно использовать пару битовых полей регистра CCER: CC2P и CC2NP, они работают в паре, более подробно написано о них в даташите.

Для нашего случая вписываем 1 в CC2P и 0 в CC2NP. Делаем это так:

TIM4->CCER &= ~TIM_CCER_CC2NP;// This bit is used in conjunction with CC2P.

TIM4->CCER |= TIM_CCER_CC2P; // Inverted/falling edge

Разрешим прерывание по переполнению. Для этого в регистре TIMx_DIER в поле CC2IE впишем 1.

Для этого впишем строчку:

TIM4->DIER |= TIM_DIER_CC2IE; // Capture/Compare 2 interrupt enable

Включаем таймер строчкой:

TIM4->CR1 |= TIM_CR1_CEN;// TIM4 enable

Не забываем разрешить глобальное прерывание:

NVIC_EnableIRQ(TIM4_IRQn); // TIM4 global Interrupt enable

Наша функция инициализации таймера выглядит следующим образом:

void Timer_Init_Receiver(void){//Settings for GPIO PB7RCC->AHBENR |= RCC_AHBENR_GPIOBEN;           // GPIO port B clock enableGPIOB->MODER |= GPIO_MODER_MODER7_1;         // Alternative function mode enableGPIOB->OSPEEDR |= GPIO_OSPEEDER_OSPEEDR7_1; // High speedGPIOB->PUPDR |= GPIO_PUPDR_PUPDR7_0;         // pull-up PB7GPIOB->AFR[0] |= 0x20000000;                 // Pin PB7 TIM4 alternative function AF2 enable//Settings for TIM4RCC->APB1ENR |= RCC_APB1ENR_TIM4EN;          // TIM4 clock enableTIM4->PSC = 9;                                 // Prescaler valueTIM4->ARR = 4200;                              // Auto-reload valueTIM4->CCR2 = 4200;                             // Capture/Compare 2 valueTIM4->CCMR1 &= ~TIM_CCMR1_OC2M;              // Frozen mode enableTIM4->CCMR1 &= ~TIM_CCMR1_CC2S;              // Output modeTIM4->CCER &= ~TIM_CCER_CC2NP;               // This bit is used in conjunction with CC2P.TIM4->CCER |= TIM_CCER_CC2P;                 // Inverted/falling edgeTIM4->SMCR |= TIM_SMCR_TS_1 | TIM_SMCR_TS_2; // Choosing TI2FP2TIM4->SMCR |= TIM_SMCR_SMS_2;                // Reset modeTIM4->DIER |= TIM_DIER_CC2IE;                // Capture/Compare 2 interrupt enableTIM4->CR1 |= TIM_CR1_CEN;                    // TIM4 enableNVIC_EnableIRQ(TIM4_IRQn);                   // TIM4 global Interrupt enable}

Теперь напишем обработчик прерываний:

Первое что нужно сделать, это снять флаг в регистре статуса TIMx_SR:

TIM4->SR &= ~TIM_SR_CC2IF;

Далее можем делать, что хотим. Хоть ставить какой-нибудь флаг, хоть зажигать диод. Допустим будем зажигать светодиод. Настроим пин PB15 на выход, прописав в мэйне:

GPIOB->MODER |= GPIO_MODER_MODER15_0;// PB15 output mode

Наш обработчик прерываний будет выглядеть следующим образом:

void TIM4_IRQHandler(void){TIM4->SR &= ~TIM_SR_CC2IF;GPIOB->ODR |= GPIO_ODR_ODR_15;  // Led red on}

Готово! Теперь при пересечении ИК луча каким-либо объектом, на приемнике будет зажигаться светодиод.

Полный код

void Timer_Init_Transmitter(void);int main(void){   RCC->ICSCR |= RCC_ICSCR_MSIRANGE_6;             // MSI 4.194 MHz enable   Timer_Init_Transmitter();   while(1)   {   }}void Timer_Init_Transmitter(void){  //Settings for GPIO PB6   RCC->AHBENR |= RCC_AHBENR_GPIOBEN;             //GPIO port B clock enable   GPIOB->MODER |= GPIO_MODER_MODER6_1;           //Alternative function mode enable   GPIOB->OSPEEDR |= GPIO_OSPEEDER_OSPEEDR6_1;  //High speed   GPIOB->AFR[0] |= 0x2000000;                    //Pin PB6 TIM4 alternative function AF2 enable   //Settings for TIM4 - Slave   RCC->APB1ENR |= RCC_APB1ENR_TIM4EN;            //TIM4 clock enable   TIM4->PSC = 0;                                 //Prescaler value   TIM4->ARR = 75;                                //Auto-reload value   TIM4->CCR1 = 37;                               //Capture/Compare 1 value   TIM4->CCMR1 |= TIM_CCMR1_OC1M_1 | TIM_CCMR1_OC1M_2; //Output compare PMW mode 1 enable   TIM4->CCER |= TIM_CCER_CC1E;                   //OC3 signal is output on the corresponding output pin   TIM4->SMCR &= ~TIM_SMCR_TS;                    //clear bits   TIM4->SMCR |= TIM_SMCR_TS_0;                   //choosing ITR1   TIM4->SMCR &= ~TIM_SMCR_SMS;                   //clear bits   TIM4->SMCR |= TIM_SMCR_SMS_0 | TIM_SMCR_SMS_2; //Gated Mode   TIM4->CR1 |= TIM_CR1_CEN;                      //TIM4 enable   //Settings for TIM2 - Master   RCC->APB1ENR |= RCC_APB1ENR_TIM2EN;            //TIM2 clock enable   TIM2->PSC = 9;                                 //Prescaler value   TIM2->ARR = 840;                               //Auto-reload value   TIM2->CCR1 = 75;                               //Capture/Compare 1 value   TIM2->CCMR1 |= TIM_CCMR1_OC1M_1 | TIM_CCMR1_OC1M_2; //Output compare PMW mode 1 enable   TIM2->CR2 |= TIM_CR2_MMS_2;                    //OC1REF signal is used as trigger output (TRGO)   TIM2->CR1 |= TIM_CR1_CEN;                      //TIM2 enable}

#include main.hvoid Timer_Init_Receiver(void);int main(void){   RCC->ICSCR |= RCC_ICSCR_MSIRANGE_6;         // MSI 4.194 MHz enable  GPIOB->MODER |= GPIO_MODER_MODER15_0;        // PB15 output mode   Timer_Init_Receiver();   while(1)   {   }}void Timer_Init_Receiver(void){//Settings for GPIO PB7RCC->AHBENR |= RCC_AHBENR_GPIOBEN;           // GPIO port B clock enableGPIOB->MODER |= GPIO_MODER_MODER7_1;         // Alternative function mode enableGPIOB->OSPEEDR |= GPIO_OSPEEDER_OSPEEDR7_1; // High speedGPIOB->PUPDR |= GPIO_PUPDR_PUPDR7_0;         // pull-up PB7GPIOB->AFR[0] |= 0x20000000;                 // Pin PB7 TIM4 alternative function AF2 enable//Settings for TIM4RCC->APB1ENR |= RCC_APB1ENR_TIM4EN;          // TIM4 clock enableTIM4->PSC = 9;                                 // Prescaler valueTIM4->ARR = 4200;                              // Auto-reload valueTIM4->CCR2 = 4200;                             // Capture/Compare 2 valueTIM4->CCMR1 &= ~TIM_CCMR1_OC2M;              // Frozen mode enableTIM4->CCMR1 &= ~TIM_CCMR1_CC2S;              // Output modeTIM4->CCER &= ~TIM_CCER_CC2NP;               // This bit is used in conjunction with CC2P.TIM4->CCER |= TIM_CCER_CC2P;                 // Inverted/falling edgeTIM4->SMCR |= TIM_SMCR_TS_1 | TIM_SMCR_TS_2; // Choosing TI2FP2TIM4->SMCR |= TIM_SMCR_SMS_2;                // Reset modeTIM4->DIER |= TIM_DIER_CC2IE;                // Capture/Compare 2 interrupt enableTIM4->CR1 |= TIM_CR1_CEN;                    // TIM4 enableNVIC_EnableIRQ(TIM4_IRQn);                   // TIM4 global Interrupt enable}void TIM4_IRQHandler(void){TIM4->SR &= ~TIM_SR_CC2IF;GPIOB->ODR |= GPIO_ODR_ODR_15;               // Led red on}

Дополнительно

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

Нужно добавить глобальную переменную, например, написав между "инклюдами" и мэйном такую строчку:

int StatusDiode = 0;// 0 - diode is off, 1 - diode is on

Эта переменная необходима для запоминания статуса светодиода: выключен или включен.

Далее в функции инициализации таймера нужно изменить строчку с разрешением прерывания по переполнению на разрешение прерывания по триггеру, такой строчкой:

TIM4->DIER |= TIM_DIER_TIE;// Trigger interrupt enable

И последнее: изменим обработчик прерываний.

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

Если диод был включен, делаем все наоборот: снимаем соответствующий флаг, выключаем диод, выключаем разрешение прерывания по переполнению, включаем разрешение прерывания по триггеру и изменяем статус диода.

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

Получим такой обработчик прерываний:

void TIM4_IRQHandler(void){if (StatusDiode == 0){    TIM4->SR &= ~TIM_SR_TIF;    GPIOB->ODR |= GPIO_ODR_ODR_15;         // Led red on    TIM4->DIER &= ~TIM_DIER_TIE;           // Trigger interrupt disable    TIM4->DIER |= TIM_DIER_CC2IE;          // Capture/Compare 2 interrupt enable    TIM4->CNT = 0;    StatusDiode = 1;}else{    TIM4->SR &= ~TIM_SR_CC2IF;    GPIOB->ODR &= ~GPIO_ODR_ODR_15;        // Led red off    TIM4->DIER &= ~TIM_DIER_CC2IE;         // Capture/Compare 2 interrupt disable    TIM4->DIER |= TIM_DIER_TIE;            // Trigger interrupt enable    StatusDiode = 0;}}

#include main.hvoid Timer_Init_Receiver(void);int main(void){   RCC->ICSCR |= RCC_ICSCR_MSIRANGE_6;      // MSI 4.194 MHz enable  GPIOB->MODER |= GPIO_MODER_MODER15_0;     // PB15 output mode   Timer_Init_Receiver();   while(1)   {   }}void Timer_Init_Receiver(void){//Settings for GPIO PB7RCC->AHBENR |= RCC_AHBENR_GPIOBEN;           // GPIO port B clock enableGPIOB->MODER |= GPIO_MODER_MODER7_1;         // Alternative function mode enableGPIOB->OSPEEDR |= GPIO_OSPEEDER_OSPEEDR7_1; // High speedGPIOB->PUPDR |= GPIO_PUPDR_PUPDR7_0;         // pull-up PB7GPIOB->AFR[0] |= 0x20000000;                 // Pin PB7 TIM4 alternative function AF2 enable//Settings for TIM4RCC->APB1ENR |= RCC_APB1ENR_TIM4EN;          // TIM4 clock enableTIM4->PSC = 9;                                 // Prescaler valueTIM4->ARR = 4200;                              // Auto-reload valueTIM4->CCR2 = 4200;                             // Capture/Compare 2 valueTIM4->CCMR1 &= ~TIM_CCMR1_OC2M;              // Frozen mode enableTIM4->CCMR1 &= ~TIM_CCMR1_CC2S;              // Output modeTIM4->CCER &= ~TIM_CCER_CC2NP;               // This bit is used in conjunction with CC2P.TIM4->CCER |= TIM_CCER_CC2P;                 // Inverted/falling edgeTIM4->SMCR |= TIM_SMCR_TS_1 | TIM_SMCR_TS_2; // Choosing TI2FP2TIM4->SMCR |= TIM_SMCR_SMS_2;                // Reset modeTIM4->DIER |= TIM_DIER_TIE;                  // Trigger interrupt enableTIM4->CR1 |= TIM_CR1_CEN;                    // TIM4 enableNVIC_EnableIRQ(TIM4_IRQn);                   // TIM4 global Interrupt enable}void TIM4_IRQHandler(void){if (StatusDiode == 0){    TIM4->SR &= ~TIM_SR_TIF;    GPIOB->ODR |= GPIO_ODR_ODR_15;           // Led red on    TIM4->DIER &= ~TIM_DIER_TIE;             // Trigger interrupt disable    TIM4->DIER |= TIM_DIER_CC2IE;            // Capture/Compare 2 interrupt enable    TIM4->CNT = 0;    StatusDiode = 1;}else{    TIM4->SR &= ~TIM_SR_CC2IF;    GPIOB->ODR &= ~GPIO_ODR_ODR_15;          // Led red off    TIM4->DIER &= ~TIM_DIER_CC2IE;           // Capture/Compare 2 interrupt disable    TIM4->DIER |= TIM_DIER_TIE;              // Trigger interrupt enable    StatusDiode = 0;}}

Заключение

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

Введение

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

Вообще говоря, сама задача разбиения строк на подстроки, каждая из которых отделена в исходной строке определённым символом, является довольно распространённой. Очень часто необходимо извлечь из строки слова, разделённые пробелами. Конечно, в стандартной библиотеке языка Си уже есть функция strtok (заголовочный файл <string.h>), но она имеет свои побочные эффекты, перечисленные ниже.

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

2. Так как функция модифицирует строку, то при передачи ей строчного литерала, будет получено SEGV, поскольку память для таких литеральных строк выделяется в сегменте кода, доступного только для чтения.

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

4. Она не учитывает экранирование символов разделителей.

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

1. Не менять оригинальную строку, в которой ищутся лексемы.

2. Для каждой найденной лексемы создавать новую строку.

3. Сохранять свою текущую позицию, а именно - указатель на подстроку, которая ещё не разбиралась.

4. Иметь однородную последовательность вызовов.

5. Иметь возможность экранировать символы разделители, при сложных лексемах.

Основные шаги при разделении строк.

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

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

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

Разработка функции.

Приступим к разработке. Для начала определим заголовочный файл "str_utils.h", содержащий все прототипы необходимых функций. Реализации функций положим в файл "str_utils.c".

Начнём с функции нахождения символов в строке. Библиотечная функция strchr могла бы решить эту задачу. Но проблема в том, что она не допускает в качестве аргумента строки, в которой надо искать символ, значения NULL. При попытке компилировать с флагами -Wall -Werror, файл с таким аргументом не скомпилируется. Хотя, такую ситуацию можно было бы обработать, вернув NULL. Поэтому был определён свой вариант данной функции с именем contains_symbol. Её прототип выглядит следующим образом:

size_t contains_symbol(char *src, char symbol);

Её реализация определена следующим образом (файл "str_utils.c"):

size_t contains_symbol(char *src, char symbol){  size_t pos = 1;  if(symbols == NULL)    return -1;  while(*symbols != '\0'){    if(*symbols++ == symbol)      return pos;    pos++;  }  return 0;}

Данная функция возвращает позицию символа в строке, увеличенную на единицу. Она не учитывает нулевой символ. Если символ не был найден, функция вернёт 0, либо -1, если ей передали NULL. Её удобно использовать в цикле while, при проверке текущего символа строки на его наличие в другой строке.

Для инкапсуляции работы с памятью был определён отдельный заголовочный файл "mem.h", содержащий следующие прототипы:

void *alloc_mem(size_t nbytes);void *calloc_mem(size_t nelems, size_t elem_size);#define alloc_str(x) ((char *) alloc_mem(x + 1))

Соответствующие функции реализованы в отдельном файле "mem.c":

#include <string.h>#include <stdlib.h>void *alloc_mem(size_t nbytes){  char *buf = (char *)malloc(nbytes);  if(buf != NULL){    memset(buf, '\0', nbytes);    return buf;  }  exit(-1);}void *calloc_mem(size_t nelems, size_t elem_size){  void *buf = calloc(nelems, elem_size);  if(buf != NULL){    return buf;  }  exit(-1);}

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

Функция обрезки разделителей строки trim_separators выглядит следующим образом:

/* trims symbols from separators at src string *//* returns new trimmed string */char *trim_separators(char *src, char *separators);

char *trim_separators(char *src, char *separators){  if(src == NULL || separators == NULL)    return NULL;  char *sp = src;  while(contains_symbol(separators, *sp)) sp++;    /* if it contains only symbols from separators => NULL */  if(sp - s == strlen(s)) return NULL;    char *sp2 = s + strlen(s) - 1; /* last char at src */  while(contains_symbol(separators, *sp2)) sp2--;    /* if it contains only symbols from separators => NULL */  if(sp2 < s) return NULL;    size_t sz = 0;  if(sp2 - sp == 0 && *sp == '\0') return NULL; /* zero byte is not a character */  else if(sp2 - sp == 0){    sz = 1;  }  else{    sz = (sp2 - sp) + 1;  }  char *res = alloc_mem(sz);  memcpy(res, sp, sz);/* copy all chars except last zero byte */  return res;}

В начале мы проверяем на NULL аргументы функции. Если они нулевые, то возвращаем NULL.

Далее, через указатель sp, проходим строку слева направо, пока мы встречаем символы из строки separators. Если мы прошли всю строку, значит она целиком и полностью состоит из сепараторов, следовательно надо удалить все символы, или же просто вернуть NULL.

char *sp = src;while(contains_symbol(separators, *sp)) sp++;  /* if it contains only symbols from separators => NULL */if(sp - s == strlen(s)) return NULL;

Аналогично, далее через указатель sp2, проходим строку справа налево, проверяя, находится ли текущий символ в массиве separators. Если это не так, то мы прерываем цикл, а указатели будут содержать ссылку на первые символы, не являющимися разделителями. Если мы опять прошли всю строку, значит снова придётся удалять всю строку, следовательно, возвращаем NULL.

char *sp2 = s + strlen(s) - 1; /* last char at src */while(contains_symbol(separators, *sp2)) sp2--;  /* if it contains only symbols from separators => NULL */if(sp2 < s) return NULL;

Наконец, вычисляем длину строки. Если указатели ссылаются на одно и то же место, то в строке был лишь один символ, не являющийся разделителем, а потому размер результата будет равным 1 байту (один лишний байт для нулевого символа учтён в макросе alloc_str). Если же этот единственный символ является нулевым (маркером конца), то возвращаем NULL. Иначе берём разницу между адресами указателями, прибавляем к ней единицу, и получаем длину новой строки. Затем мы просто выделяем память для новой строки и копируем в неё строку, начинающуюся с указателя sp.

Теперь, объединим работу выше написанных функции, в единую функцию get_token().

Код функции get_token дан ниже:

char *get_token(char *src, char *delims, char **next){  if(src == NULL || delims == NULL)    return NULL;  char *delims_p = delims;  /* the end of lexem (points to symbol that follows right after lexem */  char *src_p = trim_separators(src, delims);  /* the begining of the lexem */  char *lex_begin = src_p;  if(src_p == NULL){    *next = NULL;    return NULL;  }    /* flag that indicates reaching of delimeter */  int flag = 0;  while(*src_p != '\0'){    flag = 0;    while(*delims_p != '\0'){      if(*delims_p == *src_p){        flag = 1;        break;      }      delims_p++;    }    if(flag == 1)      break;    delims_p = delims;    src_p++;  }    /* now src_p points to the symbol right after lexem */  /* compute lexem size and reset pointers (from trimmed to the original src) */  char *offset;  size_t tok_size;  offset = (src + strspn(src, delims));  tok_size = (src_p - lex_begin);  free(lex_begin);  lex_begin = offset;  src_p = offset + tok_size;  if(*src_p == '\0')    *next = NULL;  else    *next = src_p;    /* result token */  char *res = alloc_str(tok_size);  memcpy(res, lex_begin, tok_size);  return res;}

В ней используется функция обрезки trim_separators(). Функция обрезки возвращает новую строку, и далее сканирование ведётся по ней. В цикле лишь проверяется, не равен ли текущий символ какому-либо символу разделителю из массива символов delims, и если равен, то выйти из цикла. Указатель src_p проходит по сканируемой строке. После цикла он будет указывать на символ, следующий за лексемой (конец лексемы). А начало лексемы сохраняется в указателе lex_begin, который изначально указывает на начало обрезанной, сканируемой строки. После обнаружения границ лексемы, вычисляется её размер (её число символом), а затем сканируемая строка удаляется из динамической кучи. Затем указатели переустанавливаются на позиции в оригинальной строке (первый аргумент функции get_token()), а часть строки, которая ещё не была разобрана, присваивается в качестве содержимого двойному указателю next. Обратите внимание, что next является ссылкой на другой указатель (в данном случае, на указатель строки). Двойной указатель позволяет менять значение переменной типа char *, записывая новый адрес в next. Для первого вызова данной функции, next должен хранить адрес переменной указателя, которая указывает на строку и хранит адрес первой ячейки строки. Однако, при работе с двойным указателем возможна серьёзная и незаметная ошибка, если в качестве начального значения next передать адрес переменной, которая непосредственно указывает на строку, а не адрес переменной копии, которая содержит копию адреса строки. В следующем разделе подробно описана данная ситуация, и показан пример работы данной функции.

Пример работы get_token().

Ниже дан простой рабочий пример функции get_token(). Оригинальная строка с лексемами хранится в указателе test, копия адреса строки (копия переменной test) хранится в переменной copytest. Указатель tok хранит текущую распознанную лексему, а next - сканируемую часть строки. Данная программа разделяет строку test по пробелу и символу табуляции на подстроки, и выводит их. Также она выводит саму строку test до и после работы функции. Как можно убедиться по выводу, оригинальная строка не меняется.

#include <stdio.h>#include <stdlib.h>#include <string.h>#include "mem.h"#include "str_utils.h"int main(int argc, char **argv){  char *test = "  They have  a    cat.\n \0";  char *copytest = test;  char **next = &copytest; /* has side effect on copytest */  char *tok = NULL;    printf("src:%s\n", test);  printf("copytest:%s\n", copytest);  while(*next != NULL){    tok = get_token(*next, " \t\0", next);    if(tok == NULL)      break;    printf("%s\n", tok);    free(tok);  }  printf("src after:%s\n", test);  printf("copytest after:%s\n", copytest);  return 0;}  

Вывод данной программы:

src:  They have  a    cat.copytest:  They have  a    cat.Theyhaveacat.src after:  They have  a    cat.copytest:(null)

Обратите внимание, что в цикле есть дополнительная проверка на NULL указателя tok. Дело в том, что при получении последнего слова в строке (а именно "cat.\n"), указатель next будет указывать на подстроку, состоящую лишь из одних пробелов (плюс нулевой символ). Функция trim_separators()для таких строк возвращает NULL, так как по логике придётся урезать все символы в строке. В итоге get_token() также вернёт NULL, поскольку уже ничего не осталось для сканирования. Поэтому переменная tok сохранит значение NULL, на последнем шаге.

Теперь снова по поводу двойного указателя next. Как вы могли заметить, в вышеприведённом коде ему передаётся адрес переменной copytest, а не переменной test. Дело в том, что мы можем нечаянно затереть значение переменной test (именно переменной, а не самой строки). Для примера, изменим код следующим образом. Передадим адрес test в указатель next. В итоге мы получим следующий вывод.

src:  They have  a    cat.copytest:  They have  a    cat.Theyhaveacat.src after:(null)copytest:  They have  a    cat.

Как видите, сама строка не меняется, но изменилось значение переменной test. Теперь она содержит NULL, поскольку на последнем шаге, функция присваивает ей соответствующее значение. Отсюда следует, что операции вида:

*next = addr;*next = NULL;

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

Модификация функции get_token(). Экранирование разделителей.

Функция get_token() умеет возвращать новые подстроки (токены) из исходной строки, не меняя её. Однако она совершенно не умеет их экранировать, в случае, когда лексемы представляют собой более сложные объекты. Например, а что если лексема содержит символ, который мы выбрали в качестве разделителя?

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

1233,"John Cenna","Male",4.22,"2004, 2005, 2006",11234,"John Doe","Male",4.24,"2001, 2004, 2007",01235,"Maria Laws","Female",4.23,"2003, 2006, 2008",1

Данные значения формируют следующую таблицу:

Заметим, что в данном случае, значения отделяются друг от друга запятой, причём сам разделитель (запятая) не учитывается, когда он находится в кавычках. Это значит, что сам разделитель дополнительно экранирован (escaped) двойными кавычками.

Для таких особых случаев, была определена новая функция get_token_escaped(), которая в качестве дополнительного параметра принимает массив (строку) символов, экранирующих разделители. Символы разделители не должны пересекаться с данным массивом. Т.е. один и тот же символ должен быть либо управляющим, либо разделяющим но не одним и тем же одновременно. Если же это так, то функция будет возвращать NULL. Для контроля начала и конца экранируемой последовательности, была заведена отдельная переменная neflag. Переменная neflag указывает, не является ли текущий символ частью экранируемой последовательности (neflag => not escaped flag). Она равна 0, когда символ должен быть экранируем, и 1, когда символ не экранируется. В цикле сканирования, сначала текущий символ ищется среди экранирующих (escaped). Если он найден и он равен предыдущему управляющему символу, то устанавливаем соответствующий флаг neflag равным 0, так как была найдена пара - начала и конец экранируемой последовательности. Если же это другой экранирующий символ, не равный предыдущему, то если мы уже находимся в экранируемой последовательности, то ничего не надо делать (продолжаем поиск, в надежде, что мы отыщем нужную пару (пред. символ)). Иначе, если мы нашли его впервые, то запоминаем его в переменной esym, и сбрасываем флаг neflag в 0. Символ разделитель будет учтён, если он был обнаружен (flag == 1) и он не является частью экранируемой последовательности (neflag == 1).

Ниже дан код данной процедуры.

char *get_token_escaped(char *src, char *delims, char *escaped, char **next){  if(src == NULL || delims == NULL || escaped == NULL)    return NULL;  char *delims_p = delims;  char *escaped_p = escaped;  /* the end of lexem (points to symbol that follows right after lexem */  char *src_p = trim_separators(src, delims);  /* the begining of the lexem */  char *lex_begin = src_p;  if(src_p == NULL){    *next = NULL;    return NULL;  }    /* check that (delims INTERSECTION escaped) IS NULL. */  /* IF NOT => return NULL */  int err = 0;  while(*delims_p != '\0'){    while(*escaped_p != '\0' && (err = (*escaped_p == *delims_p) ) != 1) escaped_p++;    escaped_p = escaped;    if(err){      return NULL;    }    delims_p++;  }  delims_p = delims;    /* flag that indicates reaching of delimeter */  int flag = 0;  /* flag that indicates that we are not in escaped sequence */  int neflag = 1;  /* previously saved escape character (i.e. the begining of the escaped seq.) */  char *esym = (char)0;  while(*src_p != '\0'){    flag = 0;    while(*escaped_p != '\0'){      if(*src_p == *escaped_p && *src_p == esym){        neflag = 1;        esym = (char)0;        break;      }      else if(*src_p == *escaped_p && neflag){        neflag = 0;        esym = *escaped_p;        break;      }      escaped_p++;    }    while(*delims_p != '\0'){      if(*delims_p == *src_p){        flag = 1;        break;      }      delims_p++;    }    if(flag && neflag)      break;    delims_p = delims;    escaped_p = escaped;    src_p++;  }    /* now src_p points to the symbol right after lexem */  /* compute lexem size and reset pointers (from trimmed to the original src) */  char *offset;  size_t tok_size;  offset = (src + strspn(src, delims));  tok_size = (src_p - lex_begin);  free(lex_begin);  lex_begin = offset;  src_p = offset + tok_size;  if(*src_p == '\0')    *next = NULL;  else    *next = src_p;    /* result token */  char *res = alloc_str(tok_size);  memcpy(res, lex_begin, tok_size);  return res;}

Пример её использования дан ниже:

#include <stdio.h>#include <stdlib.h>#include <string.h>#include "mem.h"#include "str_utils.h"int main(int argc, char **argv){  char *test = "  They have  \"cats dogs mice  \"\n \0";  char *copytest = test;  char **next = &copytest; /* has side effect on copytest */  char *tok = NULL;    printf("src:%s\n", test);  printf("copytest:%s\n", copytest);  while(*next != NULL){    tok = get_token_escaped(*next, " \t\0", "\"\0", next);    if(tok == NULL)      break;    printf("%s\n", tok);    free(tok);  }  printf("src after:%s\n", test);  printf("copytest after:%s\n", copytest);  return 0;}

Вывод:

src:  They have  "cats dogs mice  "copytest:  They have  "cats dogs mice  "Theyhave"cats dogs mice  "src after:  They have  "cats dogs mice  "copytest after:(null)

Заключение

Разработанная функция get_token() позволяет извлекать лексемы в отдельные новые строки. Она не меняет исходной строки, и имеет одинаковую последовательность вызовов. Из недостатков, она использует двойной указатель для сохранения текущей позиции сканера. Чтобы не затирать значение переменной, адрес которой содержит двойной указатель next, необходимо передавать адрес другой переменной, являющейся копией исходной (см. код выше). Функция также не умет экранировать символы разделители. Эту работу делает другая функция - get_token_escaped().

Функцию get_token_escaped() можно использовать при работе с CSV файлами. Однако ей должны передаваться непересекающиеся множества символов разделителей, и экранирующих символов. Иначе будет неоднозначность между ними. Функция не умеет пока анализировать такие неоднозначности, поэтому в таких случаях она вернёт NULL. Кроме того, она не допускает вложенных экранируемых последовательностей. Т.е. если был встречен экранируемый символ, всё что следует за ним, включительно до его клона (такого же символа, но в следующей позиции), считается как одна экранируемая последовательность.

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

Макросы - один из моих самых любимых инструментов в языках С и С++. Умные люди и умные книжки советуют по максимуму избегать использования макросов, по возможности заменяя их шаблонами, константами и inline-функциями, и на то есть веские основания. С помощью макросов можно создавать не только изящный код, но и плодить не менее изящные баги, которые потом будет очень сложно отловить и пофиксить. Но если соблюдать ряд несложных правил при работе с макросами, они становятся мощным оружием, которое не стреляет по твоим собственным коленям. Но сперва давай разберемся, что вообще такое макросы в С и С++?

Что есть макросы?

В языках С и С++ есть такой механизм, как препроцессор. Он обрабатывает исходный код программы ДО того, как она будет скомпилированна. У перпроцессора есть свои директивы, такие как #include, #pragma, #if и тд. Но нам интересна только директива #define.

В языке Си довольно распространенной практикой является объявление глобальных констант с помощью директивы #define:

#define PI 3.14159

А потом, на этапе препроцессинга все использования PI будут заменены указанным значением:

double area = 2 * PI * r * r;

После препроцессинга, который по сути является банальной подстановкой, это выражение превратится в:

double area = 2 * 3.14159 * r * r;

PI - макрос, в самом простом его исполнении. Естественно, макросы в таком виде не работают как переменные. Им нельзя присваивать новое значение или использовать их адрес.

// Так нельзя:PI = 3; // после препроцессинга: 3.14159 = 3int *x = Π    // после препроцессинга: int *x = &3.14159

О макросах важно понимать, что область видимости у них такая же, как у нестатических функций в языке Си, то есть они видны везде, куда их "заинклюдили". Однако в отличии от функций, объявление макроса можно отменить:

#undef PI

После этой строчки обращаться к PI будет уже нельзя.

Макросы с параметрами

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

#define MAX(a, b) a >= b ? a : b

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

#define SWAP(type, a, b) type tmp = a; a = b; b = tmp;

Поскольку мы первым параметром передаем тип, данный макрос будет работать с переменными любого типа:

SWAP(int, num1, num2)SWAP(float, num1, num2)

Макросы так же можно записывать в несколько строк, но тогда каждая строка, кроме последней, должна заканчиваться символом '\':

#define SWAP(type, a, b) type tmp = a; \ a = b; \b = tmp;

Параметр макроса можно превратить в строку, добавив перед ним знак '#':

#define PRINT_VAL(val) printf("Value of %s is %d" #val, val);int = 5;PRINT_VAL(x)  // -> Value of x is 5

А еще параметр можно приклеить к чему-то еще, чтобы получился новый идентификатор. Для этого между параметром и тем, с чем пы его склеиваем, нужно поставить '##':

#define PRINT_VAL (number) printf("%d", value_##number);int value_one = 10, value_two = 20;PRINT_VAL(one)  // -> 10PRINT_VAL(two)  // -> 20

Техника безопасности при работе с макросами

Есть несколько основных правил, которые нужно соблюдать при работе с макросами.

1. Параметрами макросов не должны быть выражения и вызовы функций.

Ранее я уже объявлял макрос MAX. Но что получится, если попытаться вызвать его вот так:

int x = 1, y = 5;int max = MAX(++x, --y);

Со стороны все выглядит нормально, но вот что получится в результате макроподстановки:

int max = ++x >= --y ? ++x : --y;

В итоге переменная max будет равна не 4, как мы ожидали, а 3. Потом можно уйму времени потратить, отлавливая эту ошибку. Так что в качестве аргумента макроса нужно всегда передавать уже конечное значение, а не какое-то выражение или вызов функции. Иначе выражение или функция будут вычислены столько раз, сколько используется этот параметр в теле макроса.

2. Все аргументы макроса и сам макрос должны быть заключены в скобки.

Это правило я уже нарушил при написании макроса MAX. Что получится, если мы захотим использовать этот макрос в составе какого-то математического выражения?

int result = 5 + MAX(1, 4);

По логике, переменная result должна будет иметь значение 9, однако вот что мы получаем в результате макроподстановки:

int result = 5 + 1 > 4 ? 1 : 4;

И переменная result внезапно примет значение 1. Чтобы такого не происходило, макрос MAX должен быть объявлен следующим образом:

#define MAX(a, b) ((a) >= (b) ? (a) : (b))

В таком случае все действия произойдут в нужном порядке.

3. Многострочные макросы должны иметь свою область видимости.

Например у нас есть макрос, который вызывает две функции:

#define MACRO() doSomething(); \ doSomethinElse();

А теперь попробуем использовать этот макрос в таком контексте:

if (some_condition) MACRO()

После макроподстановки мы увидим вот такую картину:

if (some_condition) doSomething();doSomethinElse();

Нетрудно заметить, что под действие if попадет только первая функция, а вторая будет вызываться всегда. Именно для того, чтобы избежать подобных багов, у макросов должна быть объявлена своя область видимости. Для удобства в этих целях принято использовать цикл do {} while (0); .

#define MACRO() do { \ doSomething(); \           doSomethinElse(); \         } while(0)

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

Еще немного примеров

В языке Си при помощи макросов можно эффективно избавляться от дублирования кода. Банальный пример - объявим несколько функций сложения для работы с разными типами данных:

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

DEF_SUM(int)DEF_SUM(float)DEF_SUM(double)int main() {     sum_int(1, 2);  sum_float(2.4, 6,3); sum_double(1.43434, 2,546656);}

Таким образом у нас получился аналог шаблонов из С++. Но стоит сразу обратить внимание, что данный способ не подойдет для типов, название которых состоит более чем из одного слова, например long long или unsigned short, потому что не получится нормально склеить название функции (sum_##type). Для этого сперва придется объявить для них новый тип, состоящий из одного слова.

Но все же, если есть возможность обходится без макросов, лучше обходится без макросов. Даже если ты прекрасно усвоил, как они работают, и понимаешь их силу, это никак не страхует тебя от того, что ты допустишь какую-то глупую ошибку при объявлении макроса, а потом 20 минут потратишь на ее поиски. Так что не стоит оборачивать в макросы вообще все, что только можно.

Мой блог в Телеграме

В этом цикле статей речь идет о параллельном программировании с использованием MPI.

• Часть 1. MPI - Введение и первая программа.

• Часть 2. MPI - Учимся следить за процессами.

• Часть 3. MPI - Как процессы обшаются? Сообщения типа точка-точка.

В прошлой статье мы обсудили как распределяется работа между процессами, зачем нужно знать какой процесс выполняется на конкретном потоке и как фиксировать время выполнение работы программы. Что дальше?
Иногда нам требуется остановить один/все процессы чтобы они подождали какого-либо действия системы, пользователя, пересылать одни локальные данные с процесса в другой процесс, ведь, напомню, они работают с независимой памятью и изменение переменной в одном потоке не приведет к изменению переменной с тем же именем в другом процессе, более формально процессы работают в непересекающихся множествах адресов памяти, а изменение данных в одном из них никак не повлияетс на другое множество(собственно из-за того, что они непересекающиеся).

Предисловие

Практически все программы с использованием технологии MPI используют не только средства для порождения процессов и их завершения, но и одну из самых важных частей, заложенных в названии самой технологии (Message Passing Interface), конечно же явная посылка сообщений между процессами. Описание этих процедур начнем, пожалуй с операций типа точка-точка.

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

Работает это так: один из процессов, назовем его P1, какого то коммуникатора C должен указать явный номер процесса P2, который также должен быть под коммуникатором С, и с помощью одной из процедур передать ему данные D, но на самом деле не обязательно нужно знать номер процесса, но это мы обсудим далее.

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

Оба типа процедур мы затронем подробно, а пока начнем с процедур с блокировкой.

Передаем и принимаем сообщения

Наконец приступим к практической части. Для передачи сообщений используется процедура MPI_Send. Эта процедура осуществляет передачу сообщения с блокировкой. Синтаксис у нее следующий:

int MPI_Send(void* buf, int count, MPI_Datatype datatype, int dest,  int msgtag, MPI_Comm comm);

Что тут есть что:
buf - ссылка на адрес по которому лежат данные, которые мы пересылаем. В случае массивов ссылка на первый элемент.
count - количество элементов в этом массиве, если отправляем просто переменную, то пишем 1.
datatype - тут уже чутка посложнее, у MPI есть свои переопределенные типы данных которые существуют в С++. Их таблицу я приведу чуть дальше.
dest - номер процесса кому отправляем сообщения.
msgtag - ID сообщения (любое целое число)
comm - Коммуникатор в котором находится процесс которому мы отправляем сообщение.

А вот как называются основные стандартные типы данных С++ определенные в MPI_Datatype:

Аналогичным способом указываются и другие типы данных определенные в стандартной библиотеке С/С++ - MPI_[Через _ в верхнем регистре пишем тип так как он назван в С]. Еще один пример для закрепления понимания, есть тип беззнаковых 32 битных целых чисел, назван он uint32_t, чтобы получить этот тип данных переопределенным в MPI необходимо написать следующую конструкцию: MPI_UINT32_T. То есть все вполне логично и легко, верхний регистр, вместо пробелов знаки андерскора и в начале пишем MPI.

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

Теперь поговорим о приеме этих сообщений. Для этого в MPI определена процедура MPI_Recv. Она осуществляет, соответственно, блокирующий прием данных. Синтаксис выглядит вот так:

int MPI_Recv(void* buf, int count, MPI_Datatype datatype, int source, int tag, MPI_Comm comm, MPI_Status* status);

Что тут есть что:
buf - ссылка на адрес по которому будут сохранены передаваемые данные.
count - максимальное количество принимаемых элементов.
datatype - тип данных переопределенный в MPI(по аналогии с Send).
source - номер процесса который отправил сообщение.
tag - ID сообщения которое мы принимаем (любое целое число)
comm - Коммуникатор в котором находится процесс от которого получаем сообщение.
status - структура, определенная в MPI которая хранит информацию о пересылке и статус ее завершения.

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

MPI_Status status;MPI_Recv(&buffer, 1, MPI_Float, MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status)tag = status.MPI_SOURCEsource = status.MPI_SOURCE

Здесь представлен кусочек возможного кода в процессе который принимает данные не более одного float элемента от любого процесса с любым тегом сообщения. Чтобы узнать какой процесс прислал это сообщение и с каким тэгом нужно собственно воспользоваться структурой MPI_Status.

Заметим появление констант MPI_ANY_SOURCE и MPI_ANY_TAG, они явно указывают, что можно принимать сообщения от любого процесса с любым тэгом.

Дабы закрепить знания об этих двух процедурах приведу пример как это выглядит на практике, данная программа определяет простые числа на заданном интервале:

#include <stdio.h>#include <time.h>#include "mpi.h"#define RETURN return 0#define FIRST_THREAD 0int* get_interval(int, int, int*);inline void print_simple_range(int, int);void wait(int);int main(int argc, char **argv){// инициализируем необходимые переменныеint thread, thread_size, processor_name_length;int* thread_range, interval;double cpu_time_start, cpu_time_fini;char* processor_name = new char[MPI_MAX_PROCESSOR_NAME * sizeof(char)];MPI_Status status;interval = new int[2];// Инициализируем работу MPIMPI_Init(&argc, &argv);// Получаем имя физического процессораMPI_Get_processor_name(processor_name, &processor_name_length);// Получаем номер конкретного процесса на котором запущена программаMPI_Comm_rank(MPI_COMM_WORLD, &thread);// Получаем количество запущенных процессовMPI_Comm_size(MPI_COMM_WORLD, &thread_size);// Если это первый процесс, то выполняем следующий участок кодаif(thread == FIRST_THREAD){// Выводим информацию о запускеprintf("----- Programm information -----\n");printf(">>> Processor: %s\n", processor_name);printf(">>> Num threads: %d\n", thread_size);printf(">>> Input the interval: ");// Просим пользователья ввести интервал на котором будут вычисленияscanf("%d %d", &interval[0], &interval[1]);// Каждому процессу отправляем полученный интервал с тегом сообщения 0. for (int to_thread = 1; to_thread < thread_size; to_thread++)       MPI_Send(&interval, 2, MPI_INT, to_thread, 0, MPI_COMM_WORLD);// Начинаем считать время выполненияcpu_time_start = MPI_Wtime();}// Если процесс не первый, тогда ожидаем получения данныхelse     MPI_Recv(&interval, 2, MPI_INT, MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status);// Все процессы запрашивают свой интервалrange = get_interval(thread, thread_size, interval);// После чего отправляют полученный интервал в функцию которая производит вычисленияprint_simple_range(range[0], range[1]);// Последний процесс фиксирует время завершения, ожидает 1 секунду и выводит результатif(thread == thread_size - 1){cpu_time_fini = MPI_Wtime();wait(1);printf("CPU Time: %lf ms\n", (cpu_time_fini - cpu_time_start) * 1000);}MPI_Finalize();RETURN;}int* get_interval(int proc, int size, int interval){// Функция для рассчета интервала каждого процессаint* range = new int[2];int interval_size = (interval[1] - interval[0]) / size;range[0] = interval[0] + interval_size * proc;range[1] = interval[0] + interval_size * (proc + 1);range[1] = range[1] == interval[1] - 1 ? interval[1] : range[1];return range;}inline void print_simple_range(int ibeg, int iend){// Прострейшая реализация определения простого числаbool res;for(int i = ibeg; i <= iend; i++){res = true;while(res){res = false;for(int j = 2; j < i; j++) if(i % j == 0) res = true;if(res) break;}res = not res;if(res) printf("Simple value ---> %d\n", i);}}void wait(int seconds) { // Функция ожидающая в течение seconds секундclock_t endwait;endwait = clock () + seconds * CLOCKS_PER_SEC ;while (clock() < endwait) {};}

Поясню основные моменты и идею. Здесь передача сообщений задействована дабы остановить все процессы кроме первого до того момента пока пользователь не введет необходимые данные. Как только мы вводим требуемый интервал для вычисления, то процессы получают эту информацию от того потока, который занимался ее сбором и начинают свою работу независимо. В конце последний процесс фиксирует время вычислений, ожидает одну секунду(дабы остальные уж точно завершили за это время свою работу) и выводит результат рассчетов времени. Почему же именно последний?
Нужно учитывать тот факт, что интервал который приходит от пользователя разбивается на N равных частей, а значит последнему процессу достанется часть с самыми большими числами, вследствие вычисления займут наибольшее время именно в нем, а значит и программа в худшем случае отработает именно за это время.

Делаем прием данных более гибким

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

Первая процедура которую мы обсудим следующая:

int MPI_Get_count(MPI_Status* status, MPI_Datatype datatype, int* count);

По структуре status процедура определяет сколько данных типа datatype передано соответствующим сообщением и записывает результат по адресу count.

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

Также есть еще одна процедура MPI_Get_elements. По синтаксису они отличаются лишь названиями, но назначение слегка разное. Если в сообщении передаются данные не базового типа, а типа который является производным от базовых(то есть составлен с помощью базовых типов), то нам вернет не количество этих данных, а именно количество данных базового типа. Однако в случае если передаются данные базового типа, то функции вернут одинаковые значения.

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

Хорошо, мы можем принимать на вход какие либо данные и не знать сколько их точно поступает. В таком случае нужно рационально выделять какой-то объем памяти для их сохранения(буферизации). Возникает логичный вопрос о том какой объем выделять, в этом нам поможет процедура MPI_Probe. Она позволяет получить информацию о сообщении которое ожидает в очереди на прием не получая самого сообщения. Синтаксис ее выглядит следующим образом:

int MPI_Probe(int source, int tag, MPI_Comm comm, MPI_Status* status);

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

Теперь на очень простом примере соединим эти процедуры вместе:

#include <iostream>#include "mpi.h"using namespace std;void show_arr(int* arr, int size){for(int i=0; i < size; i++) cout << arr[i] << " ";cout << endl;}int main(int argc, char **argv){int size, rank;MPI_Init(&argc, &argv);MPI_Comm_size(MPI_COMM_WORLD, &size);MPI_Comm_rank(MPI_COMM_WORLD, &rank);if(rank == 0){int* arr = new int[size];for(int i=0; i < size; i++) arr[i] = i;for(int i=1; i < size; i++) MPI_Send(arr, i, MPI_INT, i, 5, MPI_COMM_WORLD);}else{int count;MPI_Status status;MPI_Probe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status);MPI_Get_count(&status, MPI_INT, &count);int* buf = new int[count];MPI_Recv(buf, count, MPI_INT, MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status);cout << "Process:" << rank << " || Count: " << count << " || Array: ";show_arr(buf, count);}MPI_Finalize();return 0;}

Что тут происходит?
В данной программе первый процесс создает массив размером равным количеству процессов и заполняет его номерами процессов по очереди. Потом соответствующему процессу он отправляет такое число элементов этого массива, какой номер у этого процесса. Напрмер: процесс 1 получит 1 элемент, процесс 2 получит 2 элемента этого массива и так далее.

Следующие же процессы должны принять это сообщение и для этого смотрят в очередь сообщений, видят это сообщение, собирают информацию в структуру status, после чего получают длину переданного сообщения и создают буфер для его сохранения, после чего просто вы водят то что получили. Для 5 запущенных процессов результат будет вот таким:

Process:1 || Count: 1 || Array: 0 Process:2 || Count: 2 || Array: 0 1 Process:4 || Count: 4 || Array: 0 1 2 3 Process:3 || Count: 3 || Array: 0 1 2 

Собственно 4 результата потому что нулевой процесс занимается отправкой этих сообщений.

И еще несколько типов процедур посылки

Отметим, что вызов и возврат из процедуры MPI_Send далеко не гарантирует того, что сообщение покинуло данный процесс, было получено процессом, которому оно отправлено. Гарантия дается только на то, что мы после этой процедуры можем использовать передаваемые данные в своих целях не опасаясь того что они как то изменятся в отправленном сообщении.

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

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

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

И еще одна процедура - MPI_Rsend. Она осуществляет передачу сообщения по готовности. Такой процедурой нужно пользоваться аккуратно, так как она требует чтобы процесс-получатель мыл уже готов принять это сообщение. Для того чтобы она выполнилась корректно необходимо заранее позаботиться о синхронизации процессов, либо явно знать, что этот процесс будет находиться в стадии ожидании получения сообщения к моменту вызова Rsend.

Резюме

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

На этом пока все, приятного отдыха хабравчане :)

В этом цикле статей речь идет о параллельном программировании с использованием MPI.

• Часть 1. MPI - Введение и первая программа.

• Часть 2. MPI - Учимся следить за процессами.

• Часть 3. MPI - Как процессы общаются? Сообщения типа точка-точка.

В прошлой статье мы обсудили как распределяется работа между процессами, зачем нужно знать какой процесс выполняется на конкретном потоке и как фиксировать время выполнение работы программы. Что дальше?
Иногда нам требуется остановить один/все процессы чтобы они подождали какого-либо действия системы, пользователя, пересылать одни локальные данные с процесса в другой процесс, ведь, напомню, они работают с независимой памятью и изменение переменной в одном потоке не приведет к изменению переменной с тем же именем в другом процессе, более формально процессы работают в непересекающихся множествах адресов памяти, а изменение данных в одном из них никак не повлияет на другое множество(собственно из-за того, что они непересекающиеся).

Предисловие

Практически все программы с применением технологии MPI используют не только средства для порождения процессов и их завершения, но и одну из самых важных частей, заложенных в названии самой технологии (Message Passing Interface), конечно же явная посылка сообщений между процессами. Описание этих процедур начнем, пожалуй, с операций типа точка-точка.

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

Работает это так: один из процессов, назовем его P1, какого-то коммуникатора C должен указать явный номер процесса P2, который также должен быть под коммуникатором С, и с помощью одной из процедур передать ему данные D, но на самом деле не обязательно нужно знать номер процесса, но это мы обсудим далее.

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

Оба типа процедур мы затронем подробно, а пока начнем с процедур с блокировкой.

Передаем и принимаем сообщения

Наконец приступим к практической части. Для передачи сообщений используется процедура MPI_Send. Эта процедура осуществляет передачу сообщения с блокировкой. Синтаксис у нее следующий:

int MPI_Send(void* buf, int count, MPI_Datatype datatype, int dest,  int msgtag, MPI_Comm comm);

Что тут есть что:
buf - ссылка на адрес по которому лежат данные, которые мы пересылаем. В случае массивов ссылка на первый элемент.
count - количество элементов в этом массиве, если отправляем просто переменную, то пишем 1.
datatype - тут уже чутка посложнее, у MPI есть свои переопределенные типы данных которые существуют в С++. Их таблицу я приведу чуть дальше.
dest - номер процесса кому отправляем сообщения.
msgtag - ID сообщения (любое целое число)
comm - Коммуникатор в котором находится процесс которому мы отправляем сообщение.

А вот как называются основные стандартные типы данных С++ определенные в MPI_Datatype:

Аналогичным способом указываются и другие типы данных определенные в стандартной библиотеке С/С++ - MPI_[Через _ в верхнем регистре пишем тип так как он назван в С]. Еще один пример для закрепления понимания, есть тип беззнаковых 32 битных целых чисел, назван он uint32_t, чтобы получить этот тип данных переопределенным в MPI необходимо написать следующую конструкцию: MPI_UINT32_T. То есть все вполне логично и легко, верхний регистр, вместо пробелов знаки андерскора и в начале пишем MPI.

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

Теперь поговорим о приеме этих сообщений. Для этого в MPI определена процедура MPI_Recv. Она осуществляет, соответственно, блокирующий прием данных. Синтаксис выглядит вот так:

int MPI_Recv(void* buf, int count, MPI_Datatype datatype, int source, int tag, MPI_Comm comm, MPI_Status* status);

Что тут есть что:
buf - ссылка на адрес по которому будут сохранены передаваемые данные.
count - максимальное количество принимаемых элементов.
datatype - тип данных переопределенный в MPI(по аналогии с Send).
source - номер процесса который отправил сообщение.
tag - ID сообщения которое мы принимаем (любое целое число)
comm - Коммуникатор в котором находится процесс от которого получаем сообщение.
status - структура, определенная в MPI которая хранит информацию о пересылке и статус ее завершения.

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

MPI_Status status;MPI_Recv(&buffer, 1, MPI_Float, MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status)tag = status.MPI_SOURCEsource = status.MPI_SOURCE

Здесь представлен кусочек возможного кода в процессе который принимает данные не более одного float элемента от любого процесса с любым тегом сообщения. Чтобы узнать какой процесс прислал это сообщение и с каким тэгом нужно собственно воспользоваться структурой MPI_Status.

Заметим появление констант MPI_ANY_SOURCE и MPI_ANY_TAG, они явно указывают, что можно принимать сообщения от любого процесса с любым тэгом.

Дабы закрепить знания об этих двух процедурах приведу пример как это выглядит на практике, данная программа определяет простые числа на заданном интервале:

#include <stdio.h>#include <time.h>#include "mpi.h"#define RETURN return 0#define FIRST_THREAD 0int* get_interval(int, int, int*);inline void print_simple_range(int, int);void wait(int);int main(int argc, char **argv){// инициализируем необходимые переменныеint thread, thread_size, processor_name_length;int* thread_range, interval;double cpu_time_start, cpu_time_fini;char* processor_name = new char[MPI_MAX_PROCESSOR_NAME * sizeof(char)];MPI_Status status;interval = new int[2];// Инициализируем работу MPIMPI_Init(&argc, &argv);// Получаем имя физического процессораMPI_Get_processor_name(processor_name, &processor_name_length);// Получаем номер конкретного процесса на котором запущена программаMPI_Comm_rank(MPI_COMM_WORLD, &thread);// Получаем количество запущенных процессовMPI_Comm_size(MPI_COMM_WORLD, &thread_size);// Если это первый процесс, то выполняем следующий участок кодаif(thread == FIRST_THREAD){// Выводим информацию о запускеprintf("----- Programm information -----\n");printf(">>> Processor: %s\n", processor_name);printf(">>> Num threads: %d\n", thread_size);printf(">>> Input the interval: ");// Просим пользователья ввести интервал на котором будут вычисленияscanf("%d %d", &interval[0], &interval[1]);// Каждому процессу отправляем полученный интервал с тегом сообщения 0. for (int to_thread = 1; to_thread < thread_size; to_thread++)       MPI_Send(&interval, 2, MPI_INT, to_thread, 0, MPI_COMM_WORLD);// Начинаем считать время выполненияcpu_time_start = MPI_Wtime();}// Если процесс не первый, тогда ожидаем получения данныхelse     MPI_Recv(&interval, 2, MPI_INT, MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status);// Все процессы запрашивают свой интервалrange = get_interval(thread, thread_size, interval);// После чего отправляют полученный интервал в функцию которая производит вычисленияprint_simple_range(range[0], range[1]);// Последний процесс фиксирует время завершения, ожидает 1 секунду и выводит результатif(thread == thread_size - 1){cpu_time_fini = MPI_Wtime();wait(1);printf("CPU Time: %lf ms\n", (cpu_time_fini - cpu_time_start) * 1000);}MPI_Finalize();RETURN;}int* get_interval(int proc, int size, int interval){// Функция для рассчета интервала каждого процессаint* range = new int[2];int interval_size = (interval[1] - interval[0]) / size;range[0] = interval[0] + interval_size * proc;range[1] = interval[0] + interval_size * (proc + 1);range[1] = range[1] == interval[1] - 1 ? interval[1] : range[1];return range;}inline void print_simple_range(int ibeg, int iend){// Прострейшая реализация определения простого числаbool res;for(int i = ibeg; i <= iend; i++){res = true;while(res){res = false;for(int j = 2; j < i; j++) if(i % j == 0) res = true;if(res) break;}res = not res;if(res) printf("Simple value ---> %d\n", i);}}void wait(int seconds) { // Функция ожидающая в течение seconds секундclock_t endwait;endwait = clock () + seconds * CLOCKS_PER_SEC ;while (clock() < endwait) {};}

Поясню основные моменты и идею. Здесь передача сообщений задействована дабы остановить все процессы кроме первого до того момента пока пользователь не введет необходимые данные. Как только мы вводим требуемый интервал для вычисления, то процессы получают эту информацию от того потока, который занимался ее сбором и начинают свою работу независимо. В конце последний процесс фиксирует время вычислений, ожидает одну секунду(дабы остальные уж точно завершили за это время свою работу) и выводит результат расчета времени. Почему же именно последний?
Нужно учитывать тот факт, что интервал который приходит от пользователя разбивается на N равных частей, а значит последнему процессу достанется часть с самыми большими числами, вследствие вычисления займут наибольшее время именно в нем, а значит и программа в худшем случае отработает именно за это время.

Делаем прием данных более гибким

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

Первая процедура которую мы обсудим следующая:

int MPI_Get_count(MPI_Status* status, MPI_Datatype datatype, int* count);

По структуре status процедура определяет сколько данных типа datatype передано соответствующим сообщением и записывает результат по адресу count.

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

Также есть еще одна процедура MPI_Get_elements. По синтаксису они отличаются лишь названиями, но назначение слегка разное. Если в сообщении передаются данные не базового типа, а типа который является производным от базовых(то есть составлен с помощью базовых типов), то нам вернет не количество этих данных, а именно количество данных базового типа. Однако в случае если передаются данные базового типа, то функции вернут одинаковые значения.

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

Хорошо, мы можем принимать на вход какие либо данные и не знать сколько их точно поступает. В таком случае нужно рационально выделять какой-то объем памяти для их сохранения(буферизации). Возникает логичный вопрос о том какой объем выделять, в этом нам поможет процедура MPI_Probe. Она позволяет получить информацию о сообщении которое ожидает в очереди на прием не получая самого сообщения. Синтаксис ее выглядит следующим образом:

int MPI_Probe(int source, int tag, MPI_Comm comm, MPI_Status* status);

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

Теперь на очень простом примере соединим эти процедуры вместе:

#include <iostream>#include "mpi.h"using namespace std;void show_arr(int* arr, int size){for(int i=0; i < size; i++) cout << arr[i] << " ";cout << endl;}int main(int argc, char **argv){int size, rank;MPI_Init(&argc, &argv);MPI_Comm_size(MPI_COMM_WORLD, &size);MPI_Comm_rank(MPI_COMM_WORLD, &rank);if(rank == 0){int* arr = new int[size];for(int i=0; i < size; i++) arr[i] = i;for(int i=1; i < size; i++) MPI_Send(arr, i, MPI_INT, i, 5, MPI_COMM_WORLD);}else{int count;MPI_Status status;MPI_Probe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status);MPI_Get_count(&status, MPI_INT, &count);int* buf = new int[count];MPI_Recv(buf, count, MPI_INT, MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status);cout << "Process:" << rank << " || Count: " << count << " || Array: ";show_arr(buf, count);}MPI_Finalize();return 0;}

Что тут происходит?
В данной программе первый процесс создает массив размером равным количеству процессов и заполняет его номерами процессов по очереди. Потом соответствующему процессу он отправляет такое число элементов этого массива, какой номер у этого процесса. Напрмер: процесс 1 получит 1 элемент, процесс 2 получит 2 элемента этого массива и так далее.

Следующие же процессы должны принять это сообщение и для этого смотрят в очередь сообщений, видят это сообщение, собирают информацию в структуру status, после чего получают длину переданного сообщения и создают буфер для его сохранения, после чего просто выводят то, что получили. Для 5 запущенных процессов результат будет вот таким:

Process:1 || Count: 1 || Array: 0 Process:2 || Count: 2 || Array: 0 1 Process:4 || Count: 4 || Array: 0 1 2 3 Process:3 || Count: 3 || Array: 0 1 2 

Собственно 4 результата потому что нулевой процесс занимается отправкой этих сообщений.

И еще несколько типов процедур посылки

Отметим, что вызов и возврат из процедуры MPI_Send далеко не гарантирует того, что сообщение покинуло данный процесс, было получено процессом, которому оно отправлено. Гарантия дается только на то, что мы после этой процедуры можем использовать передаваемые данные в своих целях не опасаясь того, что они как-то изменятся в отправленном сообщении.

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

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

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

И еще одна процедура - MPI_Rsend. Она осуществляет передачу сообщения по готовности. Такой процедурой нужно пользоваться аккуратно, так как она требует чтобы процесс-получатель мыл уже готов принять это сообщение. Для того чтобы она выполнилась корректно необходимо заранее позаботиться о синхронизации процессов, либо явно знать, что этот процесс будет находиться в стадии ожидании получения сообщения к моменту вызова Rsend.

Резюме

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

На этом пока все, приятного отдыха, хабравчане.