Многопоточное программирование на C++

 if (*it==ch_rng("-s")) {

    isS=true; 

    if (!*++it) throw newExitMsgException(mp, _FLINE_, usage, 1);

} 

 numThr=atoi(*it);

 if ( !(numThr>=1 && numThr<=100) ) {

    throw newExitMsgException(mp, _FLINE_, "num_threads must be in [1, 100]",

      1);

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

 } 

 if (!*++it) throw newExitMsgException(mp, _FLINE_, usage, 1);

*word=*it; 

 if (!*++it) throw newExitMsgException(mp, _FLINE_, usage, 1);

*mask=*it; 

 if (*++it) throw newExitMsgException(mp, _FLINE_, usage, 1);

} 

void MainTask::proc(mem_pool& mp, const dq_vec& dqv, void*, task_opers&)

{

data_queue& dq=*dqv[0];

 for (;;) {

Этот объемлющий цикл вокруг цикла обработки сообщений  встречается только в программе mtftext, т.к. только в ней у пользователя есть возможность не прерывать работу в случае обнаружения ошибок.

     shException exc(mp, 0);

     try {

         for (MsgIO mio(mp, dq); ; ) {

             sh_ptr<Msg> msg=mio.read();

             if (!msg.get()) break;

Отсутствие прочитанного сообщения может означать как  исчерпание всех сообщений, так и is_intr() состояние очереди.

             switch (msg->getType()) {

                    case Msg::FindFiles: {

                         doFindFiles(mp, dq, msg->to<FindFilesMsg>());

                         break;

                    }

                    case Msg::ScanFile: {

                         doScanFile(mp, dq, msg->to<ScanFileMsg>());

                         break;

                    }

             }

         } 

         return;

     }

     catch (shException she) { exc=she; }

     catch (...) { exc=recatchException(mp, _FLINE_); }

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

     file(mp, fd::err).write(text_buf(toTextAll(exc))+'\n');

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

     if (exitOnErr) {

        dq.set_intr(true);

        break;

     }

}

} 

void MainTask::doFindFiles(mem_pool& mp, data_queue& dq, const FindFilesMsg&

  msg)

{

MsgIO mio(mp, dq);

 sh_dir shd=new_dir(mp, msg.dirName); 

 for (dir::entry dent(mp); shd->find_next(dent); ) {

     if (dent.name=="." || dent.name=="..")

        continue; 

     sh_text fname=shd->full_name(dent); 

     if (dent.isdir) {

        FindFilesMsg m(fname);

        mio.write(m); 

        continue;

     } 

     if (fileMchr.match(dent.name)) {

        ScanFileMsg m(fname);

        mio.write(m);

     }

}

} 

void MainTask::doScanFile(mem_pool& mp, data_queue&, const ScanFileMsg& msg)

{

 file out(mp, fd::out);

Создаем файл для  вывода, привязанный к дескриптору stdout.

 file fin(mp, msg.fileName, file::rdo, 0);

 buf_reader br(mp, fin, 64*1024); 

 sh_text line(nt(mp));

 for (int num=1; br.read_line(line); num++) {

     if (line->find(srchPatt)!=line->end())

        out.write(text_buf(mp)+msg.fileName+':'+num+':'+line+'\n');

}

} 

}  // namespace ::mtftext 

int main(int argc, char** argv)

{

 using namespace ders;

 using namespace mtftext; 

 mem_pool mp;

 file err(mp, fd::err);

 file out(mp, fd::out); 

 shException exc(mp, 0);

 try {

     CmdLineParser clp(mp);

     clp.parse(argc, argv); 

     MainTask mt(clp.isS, clp.word, clp.mask); 

     sh_data_queue dq=new_data_queue(mp); 

     MsgIO mio(mp, *dq);

     FindFilesMsg m(nt(mp, ""));

     mio.write(m);

Записываем в  очередь первое сообщение, предписывающее искать файлы в текущей директории.

     sh_thread_pool tp=(clp.numThr>1) ? new_thread_pool(mp, clp.numThr-1) :

       new_thread_pool(mp);

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

Специально отмечу, что возможность однопоточной отладки логики работы многопоточных приложений трудно переоценить! Благодаря существованию отдельной однопоточной реализации thread_pool, практически вся разработка может проходить в комфортной и предсказуемой однопоточной среде!

     tp->exec(mt, dq_vec(1, dq.get()));

Запускаем одновременное  выполнение функции MainTask::proc() всеми потоками thread_pool-а + вызвавшим exec() потоком функции main(). Именно поэтому в new_thread_pool() передается значение numThr-1.

     return (dq->is_intr()) ? 2 : 0;

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

 }

 catch (shException she) { exc=she; }

 catch (...) { exc=recatchException(mp, _FLINE_); } 

ExitMsgException* em=exc->is<ExitMsgException>();

 if (em) {

    if (em->message->size())

       (em->exitCode ? err : out).write(text_buf(em->message)+'\n');

    return em->exitCode;

}

Типичный блок обработки исключения ExitMsgException.

 err.write(text_buf(toTextAll(exc))+'\n');

 return 2;

} 

Проверка  исходного кода

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

все строки файла  соответствуют DOS или UNIX формату 

строки не содержат символов табуляции 

длина строки не превышает 80 символов

отсутствуют пробелы  в конце строки

Параметры командной  строки: mtcksrc [-nt2] d|u mask[,mask2...]

Примеры использования: 
mtcksrc.exe u *.h,*.hpp,*.c,*.cpp 
mtcksrc.exe -nt1 d *.?pp

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

void MainTask::doCheckFile(mem_pool& mp, data_queue&, const CheckFileMsg& msg)

{

 text_buf tout(mp); 

 file fin(mp, msg.fileName, file::rdo, 0);

 buf_reader br(mp, fin, 64*1024); 

 sh_text line(nt(mp));

 for (int num=1; br.read_line(line, false); num++) {

     assert(line->size()>0);

     if (line->back()=='\n') {

        if (line->ends("\r\n")) {

           if (!dos) tout+msg.fileName+':'+num+":dos\n";

           line->uninitialized_resize(line->size()-2);

        }

        else {

             if (dos) tout+msg.fileName+':'+num+":unix\n";

             line->uninitialized_resize(line->size()-1);

        }

     } 

     if (line->find('\t')!=line->end()) tout+msg.fileName+':'+num+":tab\n"; 

     if (line->size()>80) tout+msg.fileName+':'+num+":long "+line->size()+'\n'; 

     if (line->size()>0 && line->back()==' ')

        tout+msg.fileName+':'+num+":ending space\n";

} 

 if (tout.size()) file(mp, fd::out).write(tout);

}

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

Ну, а теперь самое время проверить ваши собственные  программы -- из-за пресловутого Copy/Paste DOS фрагменты не так уж и редко встречаются в UNIX файлах...  

Удаление  файлов

Программа предназначена  для рекурсивного удаления файлов по маскам.

Внимание! Удаление файлов зачастую приводит к безвозвратной потере данных, не запускайте эту программу из праздного любопытства!

Параметры командной  строки: mtdel [-nt2] mask[,mask2...]

Примеры использования: 
mtdel.exe *.obj,*.exe,*.res,out.???\* 
mtdel.exe -nt1 *.bak

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

Вывод итоговой статистики, т.е. общего количества удаленных  файлов и их совокупного размера.

К несчастью, решение данной простой задачи не вызовет никакого труда даже у едва знакомого с многопоточным программированием кодировщика: всего-то и нужно, что создать один глобальный счетчик, да защитить его mutex-ом!