пятница, 29 мая 2015 г.

Стрелки (arrows) и бесточечная (pointfree) нотация в haskell на конкретном примере

Стрелка — это модель обобщения вычислительного процесса, когда на вход вычисления может подаваться сразу несколько входных данных (например, в виде кортежа), а сам процесс вычисления строится с помощью комбинаторов типа first, second, (***), (&&&) и (>>>). Бесточечная нотация — это когда при записи функции в выражении (в том числе в ее определении), аргументы опускаются и, соответственно, не связываются (bind) в выражении. Поскольку стрелка абстрагирует входные данные, она выглядит хорошим кандидатом для применения в бесточечной нотации. Рассмотрим применение стрелки и бесточечной нотации на простом примере. Я взял его из очередного твита в 1HaskellADay. Задача простейшая — вывести на экран счет от целого числа from до целого числа to без использования явных циклов. Явные циклы в haskell, да и вообще в функциональном программировании, очень не приветствуются, поскольку относятся к миру императивного программирования. Поэтому данная задача выглядит простой и, в общем-то, таковой и является. Так как я собираюсь привести несколько решений, имеет смысл написать осто́вную функцию count, которая будет принимать основную функцию вычисления счета, передавать ей входные значения from и to, и выводить счет на экран.
count :: (Int -> Int -> [Int]) -> Int -> Int -> IO ()
count f from to = mapM_ print $ f from to
Здесь можно добавить проверку на то, что from меньше to, но я этого не сделал для простоты. Самую очевидную реализацию функции f я назвал cheating, поскольку в ней используется синтаксический сахар, предоставленный самим языком как раз для создания непрерывно возрастающего списка от некоторого числа m до некоторого числа n.
cheating from to = [from .. to]
Совершенно очевидное и незамысловатое решение. Его можно проверить в ghci или в функции main с помощью вызова
count cheating 1 10
Теперь простое решение без использования сахара haskell.
simple from to = takeWhile (<= to) . iterate succ $ from
Тоже несложно. Функция iterate succ строит бесконечный непрерывно увеличивающийся список, который начинается со значения from. Этот список трассируется функцией takeWhile до тех пор, пока выполняется предикат в скобках, то есть пока очередное значение в списке не превысит to. Обе функции cheating и simple просты и хороши. Но я бы хотел видеть их в бесточечной нотации, то есть без явного указания и связывания аргументов from и to. К сожалению, это невозможно, поскольку в первом случае синтаксис построения списка-диапазона требует указания обоих значений, а во втором случае значение to используется внутри предиката в самом центре тела функции. И здесь на помощь приходят стрелки! Для их использования необходимо импортировать модуль Control.Arrow.
import Control.Arrow
Я приведу тело функции arrowedPointfree, а ниже объясню, что в нем происходит.
arrowedPointfree = curry $ iterate succ *** (>=) >>> uncurry (flip takeWhile)
Видите, никаких from и to. Они неявно присутствуют в определении функции, но не связаны внутри ее тела. Оба аргумента склеиваются функцией curry в кортеж и передаются в вычислительный процесс, собранный из функций и стрелок, который начинается за символом $. Следуя традиции графического изображения стрелочных вычислений (см. здесь), я представлю (в меру своих художественных способностей) этот процесс вычисления на картинке.
Итак, кортеж (from, to) поступает на вход стрелочного комбинатора (***). Первый элемент кортежа (from) поступает на вход функции iterate succ, формируя бесконечный возрастающий список [from ..], второй (to) — на вход функции (>=), формируя новую функцию (>=) to. Заметьте, в функции simple мы использовали сечение (<= to). Наша новая функция эквивалентна сечению (to >=), что соответствует предикату из simple. На выходе комбинатора (***) новый кортеж ([from ..], (>=) to) поступает на оператор композиции (>>>), который передает вычисление в функцию uncurry. Функция uncurry декомпонует кортеж в последовательность двух аргументов (я изобразил это сужением вычислительного конвейера) и вызывает функцию flip takeWhile. Функция flip нужна для того, чтобы переставить аргументы [from ..] и (>=) to местами, сформировав takeWhile ((>=) to) [from ..]. Дело сделано! Не находите это решение красивым? Лично я нахожу. Это несмотря на то, что функции cheating и simple выглядят и проще, и понятнее. Но здесь нам удалось абстрагироваться от входных данных и сосредоточиться на потоке вычислений. Да, поначалу стрелки выглядят сложными, но только до тех пор, пока процесс вычисления не представлен графически. Есть еще один аспект. Часто в погоне за возможностью выразить код в бесточечном стиле (pointfree), программисты вынужденно усложняют понимание исходного кода, превращая его в pointless (бессмысленный). Поначалу я случайно назвал функцию arrowedPointfree arrowedPointless, и не сразу оценил всю глубину смысловой нагрузки этой оговорки. В этой статье приводится пример подобного усложнения (по мнению автора) и также обыгрываются синонимы pointfree и pointless. С другой стороны, невозможность записать выражение в бесточечном стиле может свидетельствовать об излишней монолитности функции. Попробуйте записать определение нашей осто́вной функции count в бесточечном стиле. Например,
count :: (Int -> Int -> [Int]) -> Int -> Int -> IO ()
count = mapM_ print
Этот вариант просто не скомпилируется: ghc пожалуется на то, что в функцию map передается слишком много аргументов. Максимальный уровень бесточечности, который можно выжать в определении функции count с такой сигнатурой, заключается в возможности опустить последний аргумент to, соединив mapM_ print и f from оператором (.).
count :: (Int -> Int -> [Int]) -> Int -> Int -> IO ()
count f from = mapM_ print . f from
Почему мы не можем выразить определение count в бесточечном стиле? Потому что мы намеренно усложнили ее определение, фактически объединив две разные функции в одну. Но в общем случае, зачем функции count знать о какой-то другой функции, принимающей какие-то аргументы и возвращающей список [Int]? Ей достаточно знать последний факт, то есть то, что она работает со списком целых чисел. Исходя из этого, определение count можно переписать в виде
count :: [Int] -> IO ()
count = mapM_ print
Вот и бесточечная нотация! Глядя на такое простое определение, мы можем заключить, что функция count нам вообще не нужна, это же просто mapM_ print, которую можно комбинировать напрямую в месте вызова. Например,
mapM_ print $ arrowedPointfree 1 10
Правда, в этом варианте пропало слово счет (count), но это уже вопрос правильного именования функций: при отсутствии осто́вной функции все функции счета следует переименовать.

пятница, 15 мая 2015 г.

Правильное нестрогое сопоставление юникодных строк в C++

Я имею в виду поиск алгоритма из набора существующих библиотек, который посчитает, что строки Семён зна́ет и семен знает равны, и сможет найти строку Знает в первой из них. Никакие strncasecmp() и iequals() в таком сравнении строк не помогут, я уже молчу о поиске подстроки — представьте, как вы будете искать подстроку с возможно измененными символами возможно другой длины в байтах! Вообще говоря, приведенный выше тип сравнения сравнением не является. В самом деле, игнорирование умляутов и преобразование произвольных букв из набора Unicode в разные регистры должны предполагать некоторую эвристику. Я не зря не использовал термин сравнение в заголовке статьи. Потому что это сопоставление, или, по-латински, коллация (collation). Объект, выполняющий коллацию, называется коллатором. Обычно существует несколько уровней коллации, начиная от самого либерального (первичный или primary) и заканчивая самым жестким, требующим полной идентичности всех символов (identical). Я не в курсе, кто изобрел эту концепцию, но она точно присутствует в библиотеке интернационализации ICU (см. документацию к классу icu::Collator). Библиотека boost::locale тоже предоставляет такой интерфейс (см. здесь), но это и не удивительно, поскольку libboost_locale, будучи слинкована с библиотеками libicuuc и libicui18n, по-видимому, просто оборачивает алгоритмы ICU. Давайте реализуем предложенное в начале статьи сравнение строк с помощью boost::locale.
#include <boost/locale.hpp>
#include <iostream>

using namespace  boost::locale;


namespace
{
    generator                 gen;
    std::locale               loc( gen( "" ) );
    const collator< char > &  col( std::use_facet< collator < char > >( loc ) );
}


int  main( void )
{
    std::string  a( "Семён зна́ет" );
    std::string  b( "семен знает" );

    bool  eq( col.compare( collator_base::primary, a, b ) == 0 );

    std::cout << a << " and " << b << " are " <<
            ( eq ? "identical" : "different" ) << " on primary comparison" <<
            std::endl;

    eq = col.compare( collator_base::secondary, a, b ) == 0;

    std::cout << a << " and " << b << " are " <<
            ( eq ? "identical" : "different" ) << " on secondary comparison" <<
            std::endl;

    std::string  c( "приём!" );

    std::cout << to_upper( c, loc ) << std::endl;
}
Внутри анонимного пространства имен определены базовые объекты boost::locale: генератор локалей gen, локаль loc, инициализируемая системной локалью (да, я предполагаю, что системной локалью является UTF-8), и ссылка на коллатор col. Внутри функции main() мы сравниваем строки с Семёном, используя первичный и вторичный уровни коллации в предположении, что в первом случае строки должны оказаться равными, а во втором нет за счет замены ё на е и снятия ударения над а. В самом низу мы также проверим, что функция boost::locale::to_upper() правильно переводит строку приём! в верхний регистр. Поместим данный исходный код в файл test.cc и скомпилируем.
g++ -Wall -o test test.cc -lboost_locale
Теперь запустим программу на выполнение.
./test
Семён зна́ет and семен знает are identical on primary comparison
Семён зна́ет and семен знает are different on secondary comparison
ПРИЁМ!
Работает! Теперь давайте искать подстроки в строке Семён зна́ет. Идею я взял из этого комментария. Вверху программы добавим новые инклюды.
#include <vector>
#include <string>
#include <algorithm>
#include <iterator>
Ниже определим тело функции find().
bool  find( const std::vector< std::string > &  p, const std::string &  s,
            collator_base::level_type  level = collator_base::primary )
{
    std::size_t  pos( 0 );
    std::string  snorm( col.transform( level, s ) ); 

    std::size_t  ssize( snorm.size() - 1 );
    std::size_t  len( ssize );

    for ( std::vector< std::string >::const_iterator  k( p.begin() );
          k != p.end(); ++k )
    {
        std::string  pnorm( col.transform( level, *k ) );
        std::size_t  psize( pnorm.size() - 1 );

        if ( psize > len )
            return false;

        std::string  scur( std::string( snorm, pos, len ) );
        std::size_t  scur_pos( scur.find( pnorm.c_str(), 0, psize ) );

        if ( scur_pos == std::string::npos )
            return false;

        std::size_t  shift( scur_pos + psize );

        pos += shift;
        len -= shift;
    }

    return true;
}
На вход find() подается ссылка на вектор искомых подстрок p, строка s, в которой будет осуществляться поиск, а также уровень коллации level. Внутри функции find() из строки s с помощью метода transform() коллатора col формируется новая бинарная строка snorm, внутри которой в цикле по всем элементам вектора p формируются соответствующие бинарные строки pnorm, которые последовательно ищутся в строке snorm обычным методом std::string::find(). Согласно автору приведенного комментария, строки snorm и pnorm содержат в конце ненужные нулевые символы, поэтому все сравнения производятся с подстроками snorm и pnorm без последнего символа (для этого объявлены дли́ны ssize и psize). Внизу функции main() добавляем код, тестирующий функцию find().
    std::vector< std::string >  parts;

    parts.push_back( "емен" );
    parts.push_back( "Зн" );
    parts.push_back( "ет" );

    bool  in( find( parts, a ) );

    std::cout << a << ( in ? " contains " : " does not contain " );
    std::copy( parts.begin(), parts.end(),
               std::ostream_iterator< std::string >( std::cout, " " ) );
    std::cout << "on primary comparison" << std::endl;
Вектор искомых подстрок parts состоит из трех элементов: емен, Зн и ет. Они должны быть найдены в строке a при использовании первичной коллации. Я не стал добавлять проверку вторичной коллации, потому что … Скомпилируем и запустим программу.
./test
Семён зна́ет and семен знает are identical on primary comparison
Семён зна́ет and семен знает are different on secondary comparison
ПРИЁМ!
Семён зна́ет does not contain емен Зн ет on primary comparison
Потому что наша find() не работает! Если вы думаете, что я где-то ошибся, попробуйте вставить поиск подстроки емен в строке Семён зна́ет в исходный код оригинального примера. Этот поиск не сработает. В этом треде автор вопроса утверждает, что после трансформации строки методом transform() на ее хвосте появляется не одиночный нулевой символ, а сразу несколько разных, в том числе не нулевых, символов. Главным результатом является один из ответов на этот вопрос, в котором утверждается, что transform() не гарантирует возможность поиска подстрок, являясь лишь инструментом для сопоставления строк. Получается, boost::locale не предоставляет возможности для нестрого поиска подстрок внутри юникодной строки. Что же тогда делать? А давайте обратимся к ICU, на которой, как мы увидели, основан boost::locale. Добавляем новые инклюды
#include <unicode/unistr.h>
#include <unicode/stsearch.h>
#include <unicode/coll.h>
, объявляем использование нового пространства имен
using namespace  icu;
, пишем новую функцию find_icu().
bool  find_icu( const std::vector< std::string > &  p, const std::string &  s,
                Collator::ECollationStrength  strength = Collator::PRIMARY )
{
    int32_t        pos( 0 );
    UnicodeString  su( UnicodeString::fromUTF8( StringPiece( s ) ) );

    for ( std::vector< std::string >::const_iterator  k( p.begin() );
          k != p.end(); ++k )
    {
        UnicodeString  pu( UnicodeString::fromUTF8( StringPiece( *k ) ) );
        UErrorCode     status( U_ZERO_ERROR );

        StringSearch   it( pu, UnicodeString( su, pos ), Locale::getDefault(),
                           NULL, status );

        it.getCollator()->setStrength( strength );

        int32_t        scur_pos( it.first( status ) );

        if ( scur_pos == USEARCH_DONE )
            return false;

        pos += scur_pos + it.getMatchedLength();
    }

    return true;
}
Здесь много новых типов, но логика не отличается от логики функции find(). Вместо строки snorm здесь мы создаем юникодную строку su, в цикле по элементам вектора p вместо строк pnorm создаются юникодные строки pu. Собственно поиск осуществляется с помощью метода first() итератора it типа StringSearch. Вместо длины строки pu к значению pos прибавляется значение, возвращаемое методом getMatchedLength() итератора it. Уровень коллации в ICU называется строгостью (strength), поэтому мы заменили название переменной level на strength. Вставим полноценную проверку поиска подстрок функцией find_icu() в main().
    in = find_icu( parts, a );

    std::cout << a << ( in ? " contains " : " does not contain " );
    std::copy( parts.begin(), parts.end(),
               std::ostream_iterator< std::string >( std::cout, " " ) );
    std::cout << "on ICU primary comparison" << std::endl;

    in = find_icu( parts, a, Collator::SECONDARY );

    std::cout << a << ( in ? " contains " : " does not contain " );
    std::copy( parts.begin(), parts.end(),
               std::ostream_iterator< std::string >( std::cout, " " ) );
    std::cout << "on ICU secondary comparison" << std::endl;
Компилируем программу
g++ -Wall -o test test.cc -lboost_locale -licuuc -licui18n
(заметили линковку новых библиотек?), и запускаем ее на выполнение.
./test
Семён зна́ет and семен знает are identical on primary comparison
Семён зна́ет and семен знает are different on secondary comparison
ПРИЁМ!
Семён зна́ет does not contain емен Зн ет on primary comparison
Семён зна́ет contains емен Зн ет on ICU primary comparison
Семён зна́ет does not contain емен Зн ет on ICU secondary comparison
Вот теперь все работает! Исходный код тестовой программы можно скачать отсюда.

среда, 1 апреля 2015 г.

Как я доказал математическую теорему с помощью haskell

В твиттере есть замечательный ресурс 1HaskellADay. Каждый день (или почти каждый) там предлагается решить какую-нибудь интересную задачу на языке haskell. Недавно решал такую задачку. Нужно найти непрерывную последовательность из 17 целых чисел, для каждого из которых существует другое число из этой же последовательности, такое, что эти два числа не являются взаимно простыми. Взаимно простыми (co-prime) называются такие два числа, у которых наибольший общий делитель (НОД или по-английски greatest common divisor (GCD)) больше 1. Так, числа 5 и 9 являются взаимно простыми, а все четные числа — нет, поскольку их НОД как минимум равен 2. Не взаимно простые числа называются по-английски co-composed (по крайней мере в этой задаче). Решение в haskell выглядит очень лаконично.
import Data.List

coComposedSet :: Integer -> [[Integer]]
coComposedSet s =
    [r | a <- [1 .. ], let r = [a .. a + s - 1],
     (== s) $ genericLength $ group [x | x <- r, y <- r, y /= x, gcd x y > 1]]
Функция coComposedSet состоит из двух строк и представляет собой генератор таких последовательностей длиной s на всем множестве целых чисел (поэтому типом множества выбран Integer, который не имеет ограничений). Чтобы получить первые 10 последовательностей длиной 17, нужно передать генератор функции take 10.
main :: IO ()
main = print $ take 10 $ coComposedSet 17
Считается эта штука достаточно быстро. Если скомпилировать программу с флагом -O2, то на моей машине Intel Core i5 время расчета составит чуть больше четырех секунд.
ghc --make -O2 coComposedSet.hs
[1 of 1] Compiling Main             ( coComposedSet.hs, coComposedSet.o )
Linking coComposedSet ...
time ./coComposedSet 
[[2184,2185,2186,2187,2188,2189,2190,2191,2192,2193,2194,2195,2196,2197,2198,2199,2200],
[27830,27831,27832,27833,27834,27835,27836,27837,27838,27839,27840,27841,27842,27843,27844,27845,27846],
[32214,32215,32216,32217,32218,32219,32220,32221,32222,32223,32224,32225,32226,32227,32228,32229,32230],
[57860,57861,57862,57863,57864,57865,57866,57867,57868,57869,57870,57871,57872,57873,57874,57875,57876],
[62244,62245,62246,62247,62248,62249,62250,62251,62252,62253,62254,62255,62256,62257,62258,62259,62260],
[87890,87891,87892,87893,87894,87895,87896,87897,87898,87899,87900,87901,87902,87903,87904,87905,87906],
[92274,92275,92276,92277,92278,92279,92280,92281,92282,92283,92284,92285,92286,92287,92288,92289,92290],
[117920,117921,117922,117923,117924,117925,117926,117927,117928,117929,117930,117931,117932,117933,117934,117935,117936],
[122304,122305,122306,122307,122308,122309,122310,122311,122312,122313,122314,122315,122316,122317,122318,122319,122320],
[147950,147951,147952,147953,147954,147955,147956,147957,147958,147959,147960,147961,147962,147963,147964,147965,147966]]

real    0m4.161s
user    0m4.147s
sys     0m0.013s
Я перенес все подсписки в отдельные строки для удобочитаемости, программа выводит всё в одну строку. А теперь переходим к теореме. Автор задачи попросил доказать, что не существует последовательностей длиной меньше 17, которые удовлетворяют условиям задачи. С короткими последовательностями длиной от 2 до 5 всё просто. Я приведу теоретические доказательства для этих случаев.
  • Длина 2. Легко показать, что все соседние числа являются взаимно простыми. В самом деле, пусть k является НОД двух соседних чисел n и n + 1. Тогда kl=nk \cdot l = n km=n+1k \cdot m = n + 1 , где l и m — произведения остальных делителей чисел n и n + 1. Отсюда следует, что k(ml)=1k \cdot (m - l) = 1 , а это значит, что k может быть равным только 1, то есть числа n и n + 1 — взаимно простые.
  • Длина 3. Число в середине последовательности является соседним для обоих крайних чисел, а значит взаимно простым с ними.
  • Длина 4. Возможны две зеркальные ситуации: последовательность начинается с четного числа или с нечетного. В обоих случаях нечетное число, находящееся между двумя четными, не может быть не взаимно простым ни с этими четными числами, так как они для него являются соседними, ни с крайним нечетным числом, поскольку для двух последовательных нечетных чисел должно выполняться равенство k(ml)=2k \cdot (m - l) = 2 , то есть k как НОД должен быть равен 2, однако это противоречит условию, что эти два числа нечетные.
  • Длина 5. Возможны два случая.
    • Четные числа по краям и в середине, нечетные - второе и четвертое. В этом случае, следуя ограничениям, выведенным в предыдущих случаях, второе число может быть не взаимно простым только с пятым, а четвертое — с первым. При этом их НОД должен удовлетворять условию k(ml)=3k \cdot (m - l) = 3 , следовательно он равен 3 для обоих нечетных чисел — второго и четвертого. Это противоречит нашему выводу о том, что два последовательных нечетных числа могут быть только взаимно простыми.
    • Нечетные числа по краям и в середине, четные - второе и четвертое. Среднее нечетное число является всегда взаимно простым с другими четырьмя — это также следует из уже доказанных свойств соседних чисел.
Случай с последовательностью из 16 чисел я теоретически доказывать не хочу, на мой взгляд это сложно, хотя, возможно, я упускаю какие-то простые факты, которые можно было бы использовать в доказательстве. Но я не буду этого делать еще по одной простой причине — доказательство можно легко провести с помощью простой программы на haskell! Идея такова. Внутри любой непрерывной последовательности из s чисел не существует числа, являющегося взаимно простым по отношению к остальным числам последовательности только в том случае, если ни одна из всевозможных комбинаций НОД-соседних чисел внутри заданной последовательности, включающая 2 и более элементов для всех НОД, соответствующих простым числам меньшим s, не задействует все элементы последовательности одновременно. НОД-соседними числами я называю идущие подряд четные числа, идущие подряд числа, кратные трем и так далее. Так, в случае последовательности длиной 16 нужно доказать, что ни одна из взаимных комбинаций НОД-соседних чисел для НОД равных 2, 3, 5, 7, 11 и 13 длиной от двух элементов не покрывает все 16 чисел последовательности одновременно. Следующая функция coComposedLayouts возвращает всевозможные комбинации НОД-соседних чисел внутри произвольной последовательности длиной s для заданного значения НОД m.
-- get all possible layouts of numbers with common factor m inside
-- an arbitrary contiguous range of s numbers; each layout must contain
-- at least 2 elements
coComposedLayouts :: Int -> Int -> [[Int]]
coComposedLayouts m s =
    filter ((> 1) . length . take 2) $
    map (takeWhile (<= s) . iterate (+ m)) [1 .. m]
Загрузим исходную программу (файл coComposedSet.hs) в ghci и проверим, как она работает.
:l coComposedSet.hs
[1 of 1] Compiling Main             ( coComposedSet.hs, interpreted )
Ok, modules loaded: Main.
coComposedLayouts 2 16
[[1,3,5,7,9,11,13,15],[2,4,6,8,10,12,14,16]]
coComposedLayouts 3 16
[[1,4,7,10,13,16],[2,5,8,11,14],[3,6,9,12,15]]
Числа внутри подсписков — индексы элементов внутри произвольной непрерывной последовательности чисел (от 1 до s), которые будут заняты НОД-соседними числами. В первом тесте мы проверили возможные варианты расположения четных чисел внутри последовательности длиной 16, во втором — варианты расположения чисел, кратных трем в такой же последовательности чисел. Выглядит разумно. А теперь напишем функцию coComposedPrimeLayouts13, которая соберет все возможные комбинации НОД-соседних чисел для всех простых НОД от 2 до 13, сортируя все элементы внутри каждой комбинации.
-- get union of all possible combinations of numbers with common prime factors
-- 2, 3, 5, 7, 11 and 13 inside an arbitrary contiguous range of s numbers
coComposedPrimeLayouts13 :: Int -> [[Int]]
coComposedPrimeLayouts13 s =
    map (sort . concat)
    [a:b:c:d:e:[f] | a <- coComposedLayouts 2  s, b <- coComposedLayouts 3  s,
                     c <- coComposedLayouts 5  s, d <- coComposedLayouts 7  s,
                     e <- coComposedLayouts 11 s, f <- coComposedLayouts 13 s]
Выглядит не очень красиво (в идеале хотелось бы обобщенную реализацию для всех простых НОД меньших s), но работает правильно: можете проверить в ghci, но вывод будет очень длинный! А вот и функция, которая находит, существует ли последовательность длиной s, удовлетворяющая условию задачи.
-- test if an arbitrary contiguous range of s numbers may have full co-composed
-- set of numbers, maximum tested prime layout is 13: this is enough for ranges
-- up to 17 numbers
hasCoComposedSet13 :: Int -> Bool
hasCoComposedSet13 s =
    elem s $ map (length . group) $ coComposedPrimeLayouts13 s
С помощью функции hasCoComposedSet13 можно доказать, что на всем множестве целых чисел не существует непрерывной последовательности длиной меньше 17, в которой для каждого числа найдется хотя бы одно другое число, не взаимно простое с исходным. А также доказать, что имеются последовательности длиной 17 чисел, для которых это условие удовлетворяется: мы находили такие последовательности с помощью функции coComposedSet 17.
map hasCoComposedSet13 [2 .. 17]
[False,False,False,False,False,False,False,False,False,False,False,False,False,False,False,True]
Теорема доказана! Исходный код программы можно найти здесь. Комментарии в строках 28–43 не верны, я не удалил их по историческим причинам. Update. Написал обобщенные версии функций coComposedPrimeLayouts и hasCoComposedSet, которые вычисляют список простых чисел, укладывающихся в заданную длину s последовательности чисел и, следовательно, подходящих для проверки существования последовательностей с искомыми свойствами произвольной длины, во время выполнения программы.
-- generalized coComposedPrimeLayouts13 with run-time calculation of
-- prime factors, requires import Data.Numbers.Primes
coComposedPrimeLayouts :: Int -> [[Int]]
coComposedPrimeLayouts s =
    map (sort . concat) $ mapM (`coComposedLayouts` s) $ takeWhile (< s) primes

--- same as hasCoComposedSet13 but uses generalized coComposedPrimeLayouts
hasCoComposedSet :: Int -> Bool
hasCoComposedSet s =
    elem s $ map (length . group) $ coComposedPrimeLayouts s
Для использования функции primes нужно подключить модуль Data.Numbers.Primes.

понедельник, 16 марта 2015 г.

Мастеринг связанных процессов в linux

Это развитие темы, поднятой в этой статье. Напомню, в ней был представлен способ гарантированного перезапуска сбоящего приложения, основанный на простой модели процессов мастер + воркер. Единственной задачей главного процесса (мастера) был немедленный перезапуск дочернего процесса (воркера) в случае завершения последнего в результате посылки ядром сигнала SIGSEGV. Новая задача будет сформулирована по-другому. Пусть у нас имеется два разных приложения. Нужно гарантировать, во-первых, что оба приложения будут выполняться одновременно, во-вторых, что перезапуск одного из приложений (после нормального завершения или получения сигнала) будет приводить к перезапуску второго приложения, и в-третьих, собственно перезапуск приложений в случае нормального завершения или получения заданных сигналов (пусть это будет SIGSEGV для определенности). Первые два условия означают, что гарантируется уникальность пар экземпляров двух приложений в любой момент времени: под связанностью процессов в заголовке статьи я подразумевал именно это. Условие связанности процессов может быть востребовано в случае, если один из них играет роль бэкенда, хранящего авторизационную информацию клиента, и общающегося с другим процессом — фронтэндом, непосредственно обслуживающим соединение с клиентом, через транспорт, не предоставляющий гарантий сохранения экземпляров взаимодействующих процессов, например TCP или UNIX-сокеты. Если условие связанности не будет выполняться, то перезапуск бэкенда приведет к утере авторизационной информации, в то время как клиентские сессии на фронтэнде останутся невредимы. Перезапуск фронтэнда в этом случае позволил бы перезагрузить переставшие быть валидными клиентские сессии. Эта проблема, на первый взгляд, кажется немного надуманной, но все же может возникнуть в реальности. Например, вам может понадобиться разработать приложение-бэкенд к сервису slapd, общающееся с последним через механизм slapd-sock. В этом случае slapd будет являться фронтэндом вашего приложения, который будет обязан перезапускать клиентские сессии (в рамках нашей задачи — “перезапускаться” сам) в случае завершения или падения бэкенда. В реальности перезапуском обеих частей нашего сервиса, как и прежде, будет заниматься мастер-процесс. Ниже я привожу исходный код соответствующей реализации, построчные комментарии ниже. Многие части полностью соответствуют коду из оригинальной статьи: их я комментировать не стану. Название файла с исходным кодом — main2.cpp.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
#include <unistd.h>
#include <sys/wait.h>
#include <sys/prctl.h>
#include <sys/time.h>
#include <string.h>
#include <signal.h>
#include <stdlib.h>
#include <errno.h>
#include <iostream>
#include <iomanip>

#ifdef MAXCYCLES
#define LOOPSTOPCOND      i < MAXCYCLES
#else
#define LOOPSTOPCOND
#endif

#ifndef PDEATHQUITSIGNAL
#define PDEATHQUITSIGNAL  SIGTERM
#endif

#ifndef CHILDLIFETIME
#define CHILDLIFETIME     4
#endif

static int  pid0( 0 );
static int  pid1( 0 );
static int  pid2( 0 );

static struct timeval  start;


static inline std::ostream &  tprint( std::ostream &  out = std::cout,
                                      const std::string &  delim = " | " )
{
    struct timeval  tv;
    gettimeofday( &tv, NULL );
    int  ms( ( tv.tv_sec - start.tv_sec ) * 1000 +
             ( tv.tv_usec - start.tv_usec ) / 1000 );
    return out << std::setw( 7 ) << float( ms ) / 1000 << delim;
}


static void  inth( int  sig )
{
    if ( getpid() != pid0 )
        exit( 0 );

    tprint() << "Master terminated by signal " << sig << std::endl;
    if ( pid1 > 0 )
    {
        tprint() << "Sending signal " << sig << " to worker 1" << std::endl;
        kill( pid1, sig );
        waitpid( pid1, NULL, 0 );
    }
    if ( pid2 > 0 )
    {
        tprint() << "Sending signal " << sig << " to worker 2" << std::endl;
        kill( pid2, sig );
        waitpid( pid2, NULL, 0 );
    }
    exit( 0 );
}


static void  setinth( void ( *handler )( int ) )
{
    struct sigaction  act;
    memset( &act, 0, sizeof( act ) );
    act.sa_handler = handler;

    int  ints[] = { SIGINT, SIGQUIT, SIGTERM, SIGHUP, 0 };

    for ( int *  s( ints ); *s != 0; ++s )
        sigaction( *s, &act, NULL );
}


int  main( int  argc, char **  argv )
{
    std::cout.precision( 3 );
    std::cout.setf( std::ios::fixed );

    gettimeofday( &start, NULL );

    pid0 = getpid();
    tprint() << "Master: " << pid0 << std::endl;

    setinth( inth );

    for ( int  i( 0 ); LOOPSTOPCOND; ++i )
    {
        if ( ( pid1 = fork() ) == 0 )    /* Worker process 1 */
        {
            pid2 = 0;
            if ( prctl( PR_SET_PDEATHSIG, PDEATHQUITSIGNAL ) == -1 )
            {
                tprint() << "Worker 1: failed to set parent death signal, "
                        "exiting" << std::endl;
                return 1;
            }
            setinth( SIG_DFL );

            tprint() << "(cycle " << i << ") Worker 1: " << getpid() <<
                    std::endl;
            sleep( CHILDLIFETIME );

            break;
        }

        if ( pid1 < 0 )
        {
            tprint() << "Failed to fork a worker 1 process, exiting" <<
                    std::endl;
            return 1;
        }

        if ( ( pid2 = fork() ) == 0 )    /* Worker process 2 */
        {
            pid1 = 0;
            if ( prctl( PR_SET_PDEATHSIG, PDEATHQUITSIGNAL ) == -1 )
            {
                tprint() << "Worker 2: failed to set parent death signal, "
                        "exiting" << std::endl;
                return 1;
            }

            tprint() << "(cycle " << i << ") Worker 2: " << getpid() <<
                    std::endl;

            char * const  cmd[] = { ( char * )"test_slapd",
                                    ( char * )"-d",
                                    ( char * )"0",
                                    ( char * )"-h",
                                    ( char * )"ldap://localhost:3333/",
                                    ( char * )"-f",
                                    ( char * )"nullslapd.conf",
                                    NULL };
            execve( "/usr/sbin/slapd", cmd, NULL );

            tprint() << "Failed to exec slapd process, exiting" << std::endl;
            return 1;
        }

        if ( pid2 < 0 )
        {
            tprint() << "Failed to fork a worker 2 process, exiting" <<
                    std::endl;
            return 1;
        }

        bool       respawn( false );
        siginfo_t  siginfo;

        if ( waitid( P_ALL, 0, &siginfo, WEXITED | WSTOPPED ) == -1 )
        {
            tprint() << "waitid() error '" << strerror( errno ) <<
                    "', exiting" << std::endl;
            return 1;
        }

        int        cpid( siginfo.si_pid == pid1 ? 1 :
                         ( siginfo.si_pid == pid2 ? 2 : -1 ) );

        if ( cpid == -1 )
        {
            tprint() << "Bad child pid " << siginfo.si_pid <<
                    ", exiting" << std::endl;
            return 1;
        }

        int *      ppid( cpid == 1 ? &pid1 : &pid2 );

        *ppid = 0;

        tprint() << "Worker " << cpid;

        if ( siginfo.si_code == CLD_KILLED || siginfo.si_code == CLD_DUMPED )
        {
            int  sig( siginfo.si_status );

            std::cout << " was signaled " << sig << std::endl;

            switch ( sig )
            {
            case SIGSEGV:
                respawn = true;
            default:
                break;
            }
        }
        else
        {
            std::cout << " exited with status " << siginfo.si_status <<
                    std::endl;
            respawn = true;
        }

        cpid = cpid == 1 ? 2 : 1;

        tprint() << "Sending quit signal to worker " << cpid << std::endl;

        ppid = cpid == 1 ? &pid1 : &pid2;
        kill( *ppid, PDEATHQUITSIGNAL );
        waitpid( *ppid, NULL, 0 );
        *ppid = 0;

        if ( ! respawn )
            break;
    }

    return 0;
}
В строках 26–28 объявлены глобальные переменные pid0, pid1 и pid2, которые в дальнейшем, в функции main(), будут инициализированы значениями PID мастер-процесса, воркера-бэкенда и воркера-фронтэнда соответственно. Их необходимо сделать глобальными, поскольку обработчик сигнала inth(), о котором речь пойдет ниже, нуждается в доступе к ним. Если вам не хочется засорять глобальное пространство имен, то поместите объявления этих переменных внутрь анонимного namespace — все же таки на C++ пишем! В строке 30 объявлена еще одна глобальная переменная start, которая будет инициализирована в функции main() текущим значением времени. Она объявлена глобальной, поскольку к ней требуется доступ из функции tprint(), расположенной в строках 33–41. Функция tprint() очень полезна, она выводит в выходной поток (предположительно std::cout или std::cerr) время, прошедшее с начала старта программы. В строках 44–63 определена функция inth()обработчик прерывания мастер-процесса. Эта функция посылает тот же сигнал прерывания sig, которым был прерван мастер-процесс обоим дочерним процессам. Но предварительно она проверяет, что вызвавший ее процесс является мастером, сравнивая вызов getpid() с pid0. Ниже вы увидите, что воркер-бэкенд устанавливает все сигналы, которые обрабатываются в inth() в значение по умолчанию SIG_DFL, а воркер-фронтэнд вызывает execve(), которая в конечном итоге делает то же самое. Спрашивается, зачем тогда нужна эта проверка? Как известно, новый процесс после вызова fork() наследует обработчики сигналов родителя, соответственно существует очень короткий промежуток времени между рождением процесса и установкой его собственных обработчиков сигналов, в течение которого, если этот новый процесс будет прерван, в нем будет вызван родительский обработчик inth(), а это очень плохо. Поэтому проверка на pid0 в обработчике inth() необходима. Собственно обработчик прерывания настраивается в функции setinth(), определенной в строках 66–76. Этот код оформлен в виде отдельной функции, поскольку нашему воркеру-бэкенду понадобится вернуть обработчики прерываний в исходные значения SIG_DFL. Переходим к функции main(). В строках 81–82 настраивается форматирование потока cout для вывода времени. В строке 84 инициализируется значение глобальной переменной start, которая будет использоваться для вычисления времени, прошедшего с начала старта программы, в функции tprint(). В строке 89 настраиваются обработчики прерывания мастер-процесса. В строках 93–109 внутри цикла for, перезапускающего воркер-процессы (см. оригинальную статью), находится код воркера-бэкенда. Его задача простая — установить сигнал смерти родителя с помощью вызова функции prctl(), восстановить обработчики прерывания по умолчанию с помощью вызова setinth(), вывести сообщение о своем старте и просто заснуть на время, определенное в секундах в макросе CHILDLIFETIME, который по умолчанию равен 4 и может быть задан во время компиляции. Волшебный вызов prctl() нужен для гарантированного завершения воркера в случае смерти мастера. Зачем, спросите вы. Ведь мы и так посылаем сигнал прерывания воркерам из обработчика inth(). Верно, но если мастер будет убит сигналом SIGKILL, этот обработчик вызван не будет, и воркеры перейдут процессу init. Данный вызов prctl() гарантирует посылку заданного сигнала, который настраивается нашим макросом PDEATHQUITSIGNAL, в случае смерти родителя, даже если тот был убит сигналом SIGKILL. В строках 111–116 — банальная проверка на правильную отработку fork(). Далее, в строках 118–143 идет код воркера-фронтэнда, который запускает экземпляр исполняемого файла /usr/sbin/slapd с помощью вызова execve(). Перед этим, как и в случае с воркером-бэкендом, устанавливается сигнал смерти родителя и выводится сообщение о старте. Переустановка обработчиков сигналов не требуется, поскольку вызов execve() устанавливает обработчики в значения по умолчанию. Функция execve() принимает список строк cmd, который будет передан как массив строк argv в функцию main() нового исполняемого кода. Если execve() будет выполнен успешно, то код, следующий за ним (строки 141–142), выполняться не будет, другими словами в этих строках находится код, отвечающий за обработку ошибки execve(). Итак, в массиве cmd находится список опций командной строки исполняемого файла /usr/sbin/slapd. Первый элемент — это имя процесса. Если бы мы запустили slapd из командной строки оболочки, оно бы соответствовало slapd, в нашем случае оно будет test_slapd. Остальные опции подобраны таким образом, чтобы slapd можно было запустить без отрыва от терминала (-d 0) обычному пользователю (-h ldap://localhost:3333/ -f nullslapd.conf). Пустой файл nullslapd.conf необходимо предварительно создать в текущей директории. Есть одна интересная тонкость. Если в опции cmd добавить -u <user>, то сигнал смерти родителя сбросится. То есть в случае посылки мастеру сигнала SIGKILL процесс test_slapd не завершится, а поменяет родителя на процесс init, а это не то, что мы ожидаем. Это связано с тем, что опция -u приводит к вызовам setgid() и setuid() внутри кода slapd, а это приводит к сбросу сигнала смерти родителя (см. man prctl). Единственный способ предотвратить это — пропатчить исходный код slapd. В строках 145–150 — банальная проверка на правильную обработку fork() для воркера-фронтэнда. Обратите внимание, что между двумя воркерами нет никакого взаимодействия: данный пример просто не рассчитан на такие подробности. Зато ниже идет код, который будет обрабатывать завершение одного из воркеров вследствие нормального выхода или прерывания сигналом. Главную работу выполняет функция waitid(), которая ожидает завершения любого из потомков мастер-процесса (строки 155–160). Переменная cpid инициализируется значением 1, если был завершен воркер-бэкенд (воркер 1), или 2, если был завершен воркер-фронтэнд (воркер 2). После определения завершившегося процесса соответствующей глобальной переменной pid1 или pid2 присваивается значение 0 для того, чтобы не возникло проблем в обработчике прерывания мастер-процесса inth(). В строках 178–197 идет обработка информации о завершившемся процессе подобная той, которая была в оригинальной статье. Только на этот раз мы присваиваем переменной respawn значение true и в том случае, если процесс завершился нормально. Кроме этого, макросы WIFSIGNALED и WTERMSIG не работают правильно с waitid(), поэтому вместо них производится прямая проверка полей si_code и si_status переменной siginfo. В строках 199–206 мы идентифицируем второй воркер, посылаем ему сигнал, установленный в макросе PDEATHQUITSIGNAL, ожидаем его завершения и присваиваем соответствующей глобальной переменной pid1 или pid2 значение 0. Понятно, зачем первый воркер-бэкенд просто завершает свою работу после заданного времени? Я хочу протестировать, что второй воркер-фронтэнд получит сигнал завершения и оба процесса будут перезапущены. Соберем программу test2 с числом перезапусков 4
g++ -g -DMAXCYCLES=4 -o test2 main2.cpp
, и запустим ее без прерываний.
./test2
  0.000 | Master: 28165
  0.000 | (cycle 0) Worker 2: 28167
  0.002 | (cycle 0) Worker 1: 28166
  4.002 | Worker 1 exited with status 0
  4.002 | Sending quit signal to worker 2
  4.007 | (cycle 1) Worker 1: 28221
  4.007 | (cycle 1) Worker 2: 28222
  8.007 | Worker 1 exited with status 0
  8.007 | Sending quit signal to worker 2
  8.011 | (cycle 2) Worker 1: 28276
  8.012 | (cycle 2) Worker 2: 28277
 12.012 | Worker 1 exited with status 0
 12.012 | Sending quit signal to worker 2
 12.018 | (cycle 3) Worker 1: 28331
 12.019 | (cycle 3) Worker 2: 28332
 16.019 | Worker 1 exited with status 0
 16.019 | Sending quit signal to worker 2
Все верно. Каждые четыре секунды воркер-бэкенд завершал работу, мастер посылал сигнал прерывания фронтэнду и перезапускал их обоих. Давайте на каком-либо этапе прервем мастер-процесс.
./test2
  0.000 | Master: 325
  0.000 | (cycle 0) Worker 1: 326
  0.001 | (cycle 0) Worker 2: 327
  4.001 | Worker 1 exited with status 0
  4.001 | Sending quit signal to worker 2
  4.008 | (cycle 1) Worker 1: 383
  4.009 | (cycle 1) Worker 2: 384
^C  5.465 | Master terminated by signal 2
  5.465 | Sending signal 2 to worker 1
  5.465 | Sending signal 2 to worker 2
Работает. А теперь давайте запустим test2, перейдем во второй терминал, узнаем PID воркера-бэкенда, и пошлем ему сигнал SIGSEGV. Только сначала пересоберем test2 с другим значением CHILDLIFETIME, а то я не буду успевать переключаться между терминалами.
g++ -g -DMAXCYCLES=4 -DCHILDLIFETIME=10 -o test2 main2.cpp
./test2
  0.000 | Master: 9834
  0.000 | (cycle 0) Worker 1: 9835
  0.001 | (cycle 0) Worker 2: 9836
 10.001 | Worker 1 exited with status 0
 10.001 | Sending quit signal to worker 2
 10.005 | (cycle 1) Worker 1: 9973
 10.005 | (cycle 1) Worker 2: 9974
Во втором терминале быстро, как только появилась запись о старте cycle 1, вводим
ps -ef | grep [t]est
lyokha    9834 27454  0 22:47 pts/4    00:00:00 ./test2
lyokha    9973  9834  0 22:48 pts/4    00:00:00 ./test2
lyokha    9974  9834  1 22:48 pts/4    00:00:00 test_slapd -d 0 -h ldap://localhost:3333/ -f nullslapd.conf
kill -SEGV 9973
Возвращаемся в первый терминал и смотрим остаток вывода test2.
 18.545 | Worker 1 was signaled 11
 18.545 | Sending quit signal to worker 2
 18.551 | (cycle 2) Worker 2: 10122
 18.554 | (cycle 2) Worker 1: 10121
 28.555 | Worker 1 exited with status 0
 28.555 | Sending quit signal to worker 2
 28.567 | (cycle 3) Worker 2: 10272
 28.570 | (cycle 3) Worker 1: 10271
 38.571 | Worker 1 exited with status 0
 38.571 | Sending quit signal to worker 2
Все верно. Ручные переключение во второй терминал, определение PID воркера-бэкенда и посылка ему сигнала заняли 8.5 секунд, так что четыре секунды мне бы явно не хватило. Можно еще поиграть разными способами. Например, послать сигнал прерывания или сигнал SIGKILL одному из воркеров, или убить мастер-процесс сигналом SIGKILL. В обоих случаях и мастер, и оба воркера должны благополучно завершиться.