Од­на из самых слож­ных и запутан­ных задач при раз­работ­ке мно­гопо­точ­ных при­ложе­ний — орга­низо­вать сов­мес­тную работу горутин. Тебе нуж­но выс­тро­ить пра­виль­ный пай­плайн обра­бот­ки дан­ных, акку­рат­но зак­рыть каналы и дож­дать­ся завер­шения горутин — даже если что‑то пош­ло не так.

В зак­лючитель­ной статье о кон­курен­тнос­ти в Go раз­берем инс­трук­цию select, сиг­наль­ные каналы и инс­тру­мен­ты пакета context. С их помощью удоб­но обме­нивать­ся сооб­щени­ями, управлять дед­лай­нами и отме­нять запущен­ные опе­рации. В качес­тве прак­тичес­кого при­мера реали­зуем прос­той механизм кре­кин­га хеша MD5: будем под­бирать пароль из сло­варя, пока его хеш не сов­падет с нуж­ным.

Сло­вари мож­но взять, нап­ример, в ре­пози­тории SecLists. Заранее выбери любое сло­во не из пер­вых строк сло­варя и запиши его MD5-хеш. Нап­ример, для пароля netadmin он будет таким:

07ded64f812afb8da181014a9d087728

Это при­годит­ся для экспе­римен­тов с кодом.

Соз­дай новый модуль:

mkdir brutemd5 && cd $_
go mod init brutemd5
touch main.go

Вставь в main.go такой код:

package main
import (
"bufio"
"crypto/md5"
"encoding/hex"
"fmt"
"os"
"runtime"
"strings"
"sync"
)
type PipelineConfig struct {
JobCh chan string
ResultCh chan string
DoneCh <-chan struct{}
ErrorCh chan<- error
SrcFileName string
TargetHash *[md5.Size]byte
}
// Функция produce генерирует задания для обработки, считывая значения из cfg.SrcFileName, и помещает их в канал cfg.JobCh; завершается по окончании чтения файла
func produce(cfg *PipelineConfig) {
file, err := os.Open(cfg.SrcFileName)
if err != nil {
cfg.ErrorCh <- fmt.Errorf("opening %s: %w", cfg.SrcFileName, err)
return
}
defer file.Close()
scanner := bufio.NewScanner(file)
for scanner.Scan() {
s := strings.TrimSpace(scanner.Text())
if s == "" {
continue
}
select {
case <-cfg.DoneCh:
return
case cfg.JobCh <- s:
}
}
if err := scanner.Err(); err != nil {
cfg.ErrorCh <- fmt.Errorf("reading %s: %w", cfg.SrcFileName, err)
}
}
// Функция worker читает слова из cfg.JobCh, вычисляет MD5-хеш и сравнивает его с cfg.TargetHash; при совпадении помещает слово в cfg.ResultCh и завершается; при отсутствии совпадений завершается
func worker(cfg *PipelineConfig) {
for {
select {
case job, ok := <-cfg.JobCh:
if !ok {
return
}
if md5.Sum([]byte(job)) == *cfg.TargetHash {
cfg.ResultCh <- job
return
}
case <-cfg.DoneCh:
return
}
}
}
// Функция collect ожидает чтения либо закрытия cfg.ResultCh, выводит значение в консоль и завершается
func collect(cfg *PipelineConfig) {
pwd, ok := <-cfg.ResultCh
if !ok {
// Канал закрыт, данные не поступили
fmt.Println("\rNo match found!")
} else {
// Принято сообщение из канала
fmt.Println("\rMatched successfully: ", pwd)
}
}
func main() {
// TODO
}

Здесь мы опре­дели­ли горути­ны, из которых соберем кон­вей­ер обра­бот­ки. Что­бы не переда­вать в фун­кции кучу аргу­мен­тов, исполь­зуем струк­туру PipelineConfig: в ней хра­ним все парамет­ры кон­вей­ера. Канал JobCh нужен для переда­чи сооб­щений, которые надо обра­ботать, то есть слов, счи­тан­ных из сло­варя. В ResultCh отпра­вим сло­во, чей хеш сов­пал с иско­мым, если такое сло­во най­дет­ся.

В ErrorCh отправ­ляем сооб­щения об ошиб­ках. DoneCh — сиг­наль­ный канал: через него переда­дим всем горути­нам коман­ду сра­зу завер­шить работу. Стро­ковая перемен­ная SrcFileName хра­нит имя фай­ла сло­варя, из которо­го мы чита­ем дан­ные, а TargetHash — ука­затель на хеш, с которым ищем сов­падение.

Фун­кция produce() генери­рует задания для даль­нейшей обра­бот­ки: чита­ет стро­ки из спис­ка, валиди­рует их и отправ­ляет в канал. Сама обра­бот­ка — вычис­ление хешей — в задачи этой фун­кции не вхо­дит и на ее код не вли­яет.

Фун­кция worker() обра­баты­вает задания: вычис­ляет хеш, срав­нива­ет его с целевым зна­чени­ем и, если они сов­пада­ют, отправ­ляет сло­во в канал резуль­татов и завер­шает­ся. Обра­ти вни­мание: ког­да ты парал­лель­но запус­каешь нес­коль­ко вор­керов, сов­падение най­дет один из них и завер­шится, но осталь­ные об этом не узна­ют и про­дол­жат грин­дить дан­ные, пока те не закон­чатся. Поэто­му нам нужен отдель­ный сиг­нал завер­шения, что­бы передать всем горути­нам «стоп».

Фун­кция collect() чита­ет резуль­таты и выводит их в кон­соль. Здесь мы ждем толь­ко один резуль­тат: так устро­ена задача. Поэто­му цик­ла нет — прос­то ждем пер­вое сооб­щение или зак­рытие канала, смот­ря что слу­чит­ся рань­ше.

info

О чте­нии из канала мы говори­ли в пре­дыду­щей статье, но вкрат­це напом­ню, если ты вдруг забыл: val, ok := <-ch бло­киру­ет текущую горути­ну, пока из канала ch не при­дет зна­чение val. Если канал зак­роют, val получит zero value для типа канала, а ok ста­нет false. Мы исполь­зуем про­вер­ку if !ok, что­бы раз­делить две ситу­ации — получе­ние сооб­щения и зак­рытие канала — и выпол­нить нуж­ные дей­ствия.

Как и в пре­дыду­щей статье, собира­ем кон­вей­ер обра­бот­ки из отдель­ных горутин: одна генери­рует задания, пул горутин парал­лель­но обра­баты­вает их, а еще одна собира­ет резуль­таты:

func main() {
const srcFileName = "dict.txt"
var maxWorkers = runtime.GOMAXPROCS(0)
// Целевой хеш — аргумент запуска
if len(os.Args) <= 1 {
fmt.Fprintf(os.Stderr, "Target MD5 hash not specified\n")
os.Exit(1)
}
hashStr := strings.TrimSpace(os.Args[1])
// Валидация и преобразование
if len(hashStr) != 32 {
fmt.Fprintf(os.Stderr, "MD5 hash must be 32 characters long\n")
os.Exit(1)
}
var hashBytes [md5.Size]byte
_, err := hex.Decode(hashBytes[:], []byte(hashStr))
if err != nil {
fmt.Fprintf(os.Stderr, "%v", err)
os.Exit(1)
}
// Канал с заданиями
jobCh := make(chan string, maxWorkers)
// Канал с результатами
resultCh := make(chan string)
// Канал с ошибками
errCh := make(chan error, 3)
// Канал для сигнализации о завершении
doneCh := make(chan struct{})
conf := &PipelineConfig{
JobCh: jobCh,
ResultCh: resultCh,
DoneCh: doneCh,
ErrorCh: errCh,
SrcFileName: srcFileName,
TargetHash: &hashBytes,
}
// Группа конвейера обработки
var pipelineWg sync.WaitGroup
// Группа пула воркеров
var workerWg sync.WaitGroup
// Запускаем конвейер обработки
pipelineWg.Go(func() {
produce(conf)
close(jobCh)
})
pipelineWg.Go(func() {
collect(conf)
close(doneCh)
})
// Запускаем обработчики
for range maxWorkers {
workerWg.Go(func() {
worker(conf)
})
}
// Закрываем канал результатов по завершении обработчиков
go func() {
workerWg.Wait()
close(resultCh)
}()
// Закрываем канал ошибок по завершении конвейера
go func() {
pipelineWg.Wait()
close(errCh)
}()
// Сбор ошибок
hadError := false
for err := range errCh {
if err != nil {
fmt.Fprintf(os.Stderr, "error %v\n", err)
hadError = true
}
}
if hadError {
os.Exit(1)
}
}

Что­бы сэконо­мить мес­то, я не буду заново раз­бирать, как работа­ет кон­вей­ер. Если нуж­но — вер­нись к пре­дыду­щей статье. Крат­ко напом­ню пос­ледова­тель­ность запус­ка: мы ини­циали­зиро­вали нуж­ные каналы и струк­туру с нас­трой­ками.

  • Го­рути­на produce (для наг­ляднос­ти я буду называть горути­ны по име­нам соот­ветс­тву­ющих фун­кций) запус­кает­ся и работа­ет, пока не про­чита­ет файл srcFileName.
  • Го­рути­на collect запус­кает­ся и работа­ет, пока не зак­роет­ся канал resultCh или пока она не получит резуль­тат вычис­лений.
  • Го­рути­ны worker запус­кают­ся и работа­ют, пока не зак­роет­ся канал jobCh.

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

Еще в одной отдель­ной горути­не ждем завер­шения кон­вей­ера и зак­рыва­ем канал оши­бок.

Тем вре­менем поток выпол­нения идет даль­ше: main() заходит в бес­конеч­ный цикл чте­ния из канала оши­бок errCh и оста­ется там, пока канал не зак­роет­ся. Пос­ле это­го по зна­чению фла­га hadError выбира­ется код воз­вра­та.

И пос­ледова­тель­ность оста­нов­ки:

  • Пос­ле чте­ния фай­ла produce завер­шает­ся, а канал jobCh зак­рыва­ется.
  • Цикл чте­ния из jobCh внут­ри worker завер­шает­ся, ког­да канал зак­рыва­ется. Пос­ле это­го worker() воз­вра­щает управле­ние. В отдель­ной горути­не мы ждем, пока завер­шатся все вор­керы, и зак­рыва­ем канал resultCh.
  • Чте­ние из resultCh внут­ри collect тоже завер­шает­ся: либо мы получи­ли иско­мое зна­чение, либо все вор­керы отра­бота­ли и канал resultCh зак­рылся. Пос­ле воз­вра­та из collect() зак­рыва­ем канал doneCh — это самый важ­ный момент, но о нем чуть поз­же.

В отдель­ной горути­не ждем, пока оста­новит­ся кон­вей­ер, то есть завер­шатся produce и collect, а затем зак­рыва­ем канал errCh.

Пос­ле зак­рытия errCh глав­ная горути­на выходит из цик­ла чте­ния.

 

Сигнальный канал и select

Раз­берем код фун­кции worker() под­робнее:

func worker(cfg *PipelineConfig) {
for {
select {
case job, ok := <-cfg.JobCh:
if !ok {
return
}
// ...вычисление хеша...
case <-cfg.DoneCh:
return
}
}
}

Продолжение доступно только участникам

Материалы из последних выпусков становятся доступны по отдельности только через два месяца после публикации. Чтобы продолжить чтение, необходимо стать участником сообщества «Xakep.ru».

Присоединяйся к сообществу «Xakep.ru»!

Членство в сообществе в течение указанного срока откроет тебе доступ ко ВСЕМ материалам «Хакера», позволит скачивать выпуски в PDF, отключит рекламу на сайте и увеличит личную накопительную скидку! Подробнее

  • Подпишись на наc в Telegram!

    Только важные новости и лучшие статьи

    Подписаться

    • 1 комментарий
    Подписаться
    Уведомить о
    1 Комментарий
    Старые
    Новые Популярные
    Межтекстовые Отзывы
    Посмотреть все комментарии