Здравствуйте, в данной заметке будет затронута тема организации «чистого» завершения для приложений, написанных на языке Go.
Чистым выходом я называю наличие гарантий того, что в момент завершения процесса (по сигналу или по любым иным причинам кроме system failure), будут выполнены определённые процедуры и выход будет отложен до окончания их выполнения. Далее я приведу несколько типичных примеров, расскажу о стандартном подходе, а также продемонстрирую свой пакет для упрощённого применения этого подхода в ваших программах и сервисах.
TL;DR: github.com/xlab/closer 
1. Введение
Итак, наверняка вы замечали хоть раз, как какой-нибудь сервер или утилита ловит ваш кручёный Ctrl^C и, дико извиняясь конечно, просит подождать, пока она порешает дела, которые никак нельзя отложить. Хорошо написанные программы завершают дела и выходят, плохие же впадают в deadlock и сдаются только при виде SIGKILL. Точнее, о SIGKILL программа узнать не успевает, подробно процесс описан здесь: SIGTERM vs. SIGKILL и Unix Signal.
При переходе на Go в качестве основного языка разработки и после продолжительного использования последнего для написания различных сервисов мне стало ясно, что добавлять обработку сигналов нужно буквально в каждый сервис. В основном из-за того, что в Go многопоточность является примитивом языка. Внутри одного процесса могут одновременно работать, к примеру, следующие потоки:
- Connection pool клиентов БД;
- Consumer для pub/sub очереди;
- Publisher для pub/sub очереди;
- N потоков собственно воркеров;
- Кэш в памяти;
- Открытые файлы логов;
Здесь нет ничего сверхъестественного (извините, если обидел), тем более на деле это представляет собой несколько сущностей, которые делают свою работу в фоне (go-рутины), и общаются между собой через go-каналы (типизированные очереди). Обычный такой сервис микросервисной архитектуры.
И с запуском всё предельно просто: сначала стартуем пул клиентов БД, если не стартовал — выходим с ошибкой. Затем инициализируем кэш в памяти. Затем запускаем publisher, если не стартовал — выходим с ошибкой. Затем открываем файлы — например логи. Затем запускаем воркеров, да побольше, которые будут потреблять данные через consumer, писать в БД и что-то держать в кэше, а результаты складывать в publisher. Ах да, ещё события обработки будут писаться в логи, не обязательно из тех же потоков. И, наконец, активируем всё это открыв поток данных consumer, а если не отрылся — выходим.
Инициализация происходит последовательно, в один поток, в случае ошибки на одном этапе откатывать уже выполненные этапы инициализации не обязательно, так как система находится в нулевом положении всё это время, пока не откроем поток данных. И вот открыли поток данных, а через 5 минут нам срочно потребовалось выйти, завершить всё, да так, чтобы красиво и чисто.
Зачем? А потому что не все результаты из буферизированного канала могли успеть быть полученными процессом записи в БД, да и те, что были считаны из канала, могли не успеть дойти до БД по сети. И не все объекты могли успеть опубликоваться в pub/sub очередь. Не все воркеры могли успеть сдать свои результаты в соответствующие каналы. Потребление очереди воркерами могло быть также буферизировано, а значит, небольшая часть объектов могла оказаться считанной с сервера pub/sub очереди, но ещё не обработанной воркерами. Кэш в памяти, например, должен быть сдамплен на диск в момент завершения программы, а ещё все буферы с данными логов должны быть очищены в соответствующие файлы. Всё это перечислено здесь с целью показать, что любой примитивный сервис с несколькими фоновыми задачами обречён иметь способ надёжного отслеживания выхода приложения. И вовсе не ради красивого уведомления «Bye bye...» в консоли, а как жизненно необходимый механизм синхронизации многопоточного комбайна.
2. Немного практики
В Go имеется хороший инструмент — defer, это выражение, будучи применённым к функции, добавит её в специальный список. Функции из этого списка будут выполнены в обратном порядке перед возвратом из текущей функции. Такой механизм иной раз упрощает работу с мьютексами и прочими ресурсами, которые нужно освободить при возврате. Эффект defer действует даже если случается паника (=исключение), то есть, определённый в deferred-функции код получает гарантию быть выполненным, а сами исключения таким способом могут быть пойманы и обработаны.
func Checked() {
defer func() {
// проверка, была ли паника
if x := recover(); x != nil {
// можно написать в лог, а также пробросить исключение наверх
}
}()
// что-нибудь делаем, случается паника
}
Но есть один злостный антипаттерн, почему-то зачастую defer начинают использовать в функции main. Например:
func main() {
defer doCleanup()
// немного псевдоработы
fmt.Println("10 seconds to go...")
<-time.Tick(10 * time.Second)
}
Код отлично отработает в случае обычного возврата и даже паники, но люди забыли о том, что defer не сработает в случае получения процессом сигнала на завершение (выполняется syscall exit, из документации Go: «The program terminates immediately; deferred functions are not run.»).
Чтобы грамотно обработать подобную ситуацию, сигналы следует ловить вручную «подписавшись» на нужные типы сигналов. Распространённая практика (судя по ответам на StackOverflow) заключается в использовании signal.Notify, паттерн выглядит примерно так:
sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan,
syscall.SIGHUP,
syscall.SIGINT,
syscall.SIGTERM,
syscall.SIGQUIT)
go func() {
s := <-sigChan
// поймали один из
}()
Для скрытия лишних деталей реализации и был придуман пакет xlab/closer, о нём пойдёт речь дальше.
3. Closer
Итак, пакет closer берёт на себя обязанность отслеживать сигналы, позволяет привязать функции и автоматически выполнит их в обратном порядке при завершении. Пакет потокобезопасен, тем самым избавляя пользователя от необходимости думать о возможных здесь состояниях гонки при вызове closer.Close из нескольких потоков одновременно. API на данный момент состоит из 5 функций: Init, Bind, Checked, Hold и Close. Init позволяет пользователю переопределить список сигналов и другие опции, использование остальных функций рассмотрим на примерах.
Стандартный список сигналов: syscall.SIGINT, syscall.SIGHUP, syscall.SIGTERM, syscall.SIGABRT.
Пример обычный
func main() {
closer.Bind(cleanup)
go func() {
// делаем работу в отдельном потоке
fmt.Println("10 seconds to go...")
<-time.Tick(10 * time.Second)
// по окончании требуем завершение процесса
closer.Close()
}()
// блокирует, пока не будет отработан выход — по сигналу или через closer.Close
closer.Hold()
}
func cleanup() {
fmt.Print("Hang on! I'm closing some DBs, wiping some trails..")
<-time.Tick(3 * time.Second)
fmt.Println(" Done.")
}
Пример с ошибкой
Функция closer.Checked позволяет делать проверку на ошибки и ловить исключения. Здесь код возврата будет отличен от нуля, причём обработкой выхода занимается по-прежнему пакет closer.
func main() {
closer.Bind(cleanup)
closer.Checked(run, true)
}
func run() error {
fmt.Println("Will throw an error in 10 seconds...")
<-time.Tick(10 * time.Second)
return errors.New("KAWABANGA!")
}
func cleanup() {
fmt.Print("Hang on! I'm closing some DBs, wiping some trails...")
<-time.Tick(3 * time.Second)
fmt.Println(" Done.")
}
Пример с паникой (исключением)
func main() {
closer.Bind(cleanup)
closer.Checked(run, true)
}
func run() error {
fmt.Println("Will panic in 10 seconds...")
<-time.Tick(10 * time.Second)
panic("KAWABANGA!")
return nil
}
func cleanup() {
fmt.Print("Hang on! I'm closing some DBs, wiping some trails...")
<-time.Tick(3 * time.Second)
fmt.Println(" Done.")
}
Таблица соответствия кодов завершения:
Событие | Код завершения
------------- | -------------
error = nil | 0 (успех)
error != nil | 1 (ошибка)
panic | 1 (ошибка)
Заключение
Таким образом, вне зависимости от первопричины завершения процесса, ваше приложение на Go отработает необходимую процедуру «чистого» завершения. В Go принято для каждой сущности, требующей такой процедуры, писать метод Close, который бы финализировал все внутренние процессы этой сущности. Значит, завершение вышеописанного сервиса из второй части данной статьи будет заключаться в вызове метода Close() для всех созданных сущностей, в обратном порядке.
Сначала закрывается поток данных consumer очереди pub/sub, новых задач в систему поступать не будет, затем система дождётся, пока все воркеры отработают и завершатся, только после этого будет синхронизирован с диском кэш, закрыт канал записи в БД, закрыт канал publisher, синхронизированы и закрыты файлы логов, и, наконец, будут закрыты подключения к БД и сам publisher. На словах звучит достаточно серьёзно, но на деле же достаточно лишь грамотно написать метод Close каждой сущности и в main при инициализации использовать closer.Bind. Эскиз main для наглядности:
func main() {
defer closer.Close()
pool, _ := xxx.NewPool()
closer.Bind(pool.Close)
pub, _ := yyy.NewPublisher()
closer.Bind(function(){
pub.Stop()
<-pub.StopChan
})
wChan := make(chan string, BUFFER_SIZE)
workers, _ := zzz.NewWorkgroup(pool, pub, wChan)
closer.Bind(workers.Close)
sub, _ := yyy.NewConsumer()
closer.Bind(sub.Stop)
// блокирующий вызов (иначе используйте closer.Hold)
sub.Consume(wChan)
}
Удачной вам синхронизации!
Автор: Xlab
