Здравствуйте, в данной заметке будет затронута тема организации «чистого» завершения для приложений, написанных на языке 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.Sleep(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.Sleep(10 * time.Second)
		// по окончании требуем завершение процесса
		closer.Close()
	}()

	// блокирует, пока не будет отработан выход — по сигналу или через closer.Close
	closer.Hold()
}

func cleanup() {
	fmt.Print("Hang on! I'm closing some DBs, wiping some trails..")
	time.Sleep(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.Sleep(10 * time.Second)
	return errors.New("KAWABANGA!")
}

func cleanup() {
	fmt.Print("Hang on! I'm closing some DBs, wiping some trails...")
	time.Sleep(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.Sleep(10 * time.Second)
	panic("KAWABANGA!")
	return nil
}

func cleanup() {
	fmt.Print("Hang on! I'm closing some DBs, wiping some trails...")
	time.Sleep(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)
}

Удачной вам синхронизации!