Привет Хабр! В последнее время я очень сильно увлекся разработкой на языке GO. Изящный и выразительный язык программирования. Мне давно хотелось сделать что-нибудь полезное. По специфике своей работы мне приходится работать с медицинскими архивами DICOM-изображений PACS.
Github: github.com/Loafter/dtools
Версия Linux-amd64: github.com/Loafter/dtools/releases/download/1.0/dcmjsser
Я решил, что пришло время создать свой dicom-клиент с (блэкджеком..) веб-интерфейсом, который может выполнять следующие стандартные операции:
- Dicom ping;
- Cкачивание исследований;
- Загрузка исследования,
- А также поиск по реквизитам
(c-echo, c-move,c-store,c-find соотвествено).
В качестве dicom-библиотеки была выбрана библиотека GrassRoot SDK. Наш клиент будет распараллеливать задачи. Язык go для этого хорошо адаптирован
Похожий сценарий работы был описан habrahabr.ru/post/198150.
Наш сценарий несколько отличается:
У нас есть некий балансировщик задач, который получает задания dicom-сервиса, проверяет возможность выполнения и асинхронно их выполняет. Для того чтобы не было ситуации, когда параллельно выполняется 1000 задач, мы реализуем очередь задач таким образом, чтобы были активные задачи и те, которые находятся в спящем состоянии. По умолчанию только 10 задач будут активными. В противном случае мы могли бы обойтись без баллансировщика вообще, тупо параллельно выполнить 1000 задач параллельно без какого либо контроля.
Весь код балансировщика находится в файле job_ballancer.go.
В начале идет описание интерфейсов обработчиков. В случае если работа была выполнена успешно, в случае если вернулась ошибка и сам процесс обработки задачи.
type JobDispatcher interface {
Dispatch(interface{}) (interface{}, error)
}
type ErrDispatcher interface {
DispatchError(FaJob) error
}
type CompDispatcher interface {
DispatchSuccess(CompJob) error
}
Когда мы создаем экземпляр диспетчера, мы его инициализируем соответствующими обработчиками.
srv.jbBal.Init(&srv.dDisp, srv, srv)
//Сама структура балансировщика
type JobBallancer struct {
jChan chan interface{} //канал в который мы передаем задания
acJob map[string]Job //список активных работ
slJob map[string]Job //список неактивных работ
errDisp ErrDispatcher //обработчик работ завершившихся error
jobDisp JobDispatcher //обработчик заданий
compDisp CompDispatcher //обработчик успешно выполненных работ
JbDone sync.WaitGroup //ожидаем завершения всех работ
aJobC int //количество параллельных (активных)
}
//инициализация балансировщика
func (jbal *JobBallancer) Init(jdis JobDispatcher, cmd CompDispatcher, erd ErrDispatcher) {
jbal.errDisp = erd
jbal.jobDisp = jdis
jbal.compDisp = cmd
jbal.acJob = make(map[string]Job)
jbal.slJob = make(map[string]Job)
jbal.aJobC = 10
jbal.jChan = make(chan interface{})
go jbal.takeJob() //запускаем поток в котором осуществляем //балансировку работ
log.Println("info: job ballancer inited")
}
добавление новой работы в очередь. Операция асинхронная, т.е. работа отсылается в канал, а функция takeJob подбирает ее от туда.
func (jbal *JobBallancer) PushJob(jdat interface{}) error {
if jbal.checkInit() {
return errors.New("error: JobChan is not inited")
}
uid := genUid()
job := Job{JobId: uid, Data: jdat}
jbal.jChan <- job
return nil
}
func (jbal *JobBallancer) takeJob() {
for {
//извлекаем работу из канала
recivedTask := <-jbal.jChan
log.Println("info: job taken")
switch job := recivedTask.(type) {
case TermJob:
//если мы получаем сигнал на завершение выходим из функции
log.Println("info: recive terminate dispatch singal")
return
case Job:
//обычная обработка (если все слоты активных работ заняты то работа записывается в список не активных работ)
if len(jbal.acJob) < jbal.aJobC {
jbal.JbDone.Add(1)
jbal.addActiveJob(job)
go jbal.startJob(job)
log.Println("info: normal dispatch")
} else {
jbal.addSleepJob(job)
jbal.JbDone.Add(1)
log.Println("info: attend maximum active job")
}
case CompJob:
//работа завершилась успехом
if err := jbal.compDisp.DispatchSuccess(job); err != nil {
log.Println("error: failed dispatch success" + job.Job.JobId)
}
//удачно завершившуюся работу можно удалить из списка
jbal.removeJob(job.Job.JobId)
jbal.JbDone.Done()
jbal.resumeJobs()
case FaJob:
//работа завершилась ошибкой
if err := jbal.errDisp.DispatchError(job); err != nil {
log.Println("error: failed dispatch error" + job.Job.JobId)
}
//завершившуюся работу можно удалить из списка
jbal.removeJob(job.Job.JobId)
jbal.JbDone.Done()
jbal.resumeJobs()
default:
log.Fatalln("error: unknown job type")
jbal.JbDone.Done()
}
}
}
//функция удаления работы
func (jbal *JobBallancer) removeJob(jid string) error {
if _, isFind := jbal.acJob[jid]; isFind {
delete(jbal.acJob, jid)
} else {
return errors.New("error: can't remove job because job with id not found")
}
return nil
}
//функция позволяющая правильно завершить работу балансировщика, в случае если есть работы которые не завершены, функция будет ожидать их завершения
func (jbal *JobBallancer) TerminateTakeJob() error {
if jbal.checkInit() {
return errors.New("error: is not inited")
}
jbal.JbDone.Wait()
jbal.jChan <- TermJob{}
close(jbal.jChan)
if len(jbal.acJob) > 0 {
return errors.New("error: list job is not empty")
}
log.Println("info: greacefully terminate take job")
return nil
}
Остальные вспомогательные функции мы не будем рассматривать. полный код можно посмотреть
github.com/Loafter/dtools/blob/master/dcmjsser/job_ballancer.go
Не смотря, что код не сложный и я долго обдумывал его. Но все равно для проверки надежности я реализовал нагрузочный тест на десятки задач:
testJobDispatcher := TestJobDispatcher{}
testErrorDispatcher := TestErrorDispatcher{}
testSuccessDispatcher := TestCompletedDispatcher{}
jobBallancer := JobBallancer{}
jobBallancer.Init(&testJobDispatcher, &testSuccessDispatcher, &testErrorDispatcher)
for i := 0; i < 40; i++ {
jobBallancer.PushJob("data: " + strconv.Itoa(i))
}
jobBallancer.TerminateTakeJob()
Он отработал нормально. Все задачи были выполнены, а функция TerminateTakeJob завершилась тогда, когда все задачи были выполненны. Для контроля отработаных задач используется объект синхронизации sync.WaitGroup JbDone, который ведет подсчет количества выполненных работ. Как я уже отмечал выше, код балансировщика является универсальным и для того чтобы наш балансировщик работал по-другому, нам достаточно проинстанцировать его соотвествующими обработчиками.
Как и в прошлой своей поделке) (http://habrahabr.ru/post/247727/) интерефейс приложения я реализовал в виде веб-интерфейса.
Для теста я использовал публичный dicom-архив 213.165.94.158:11112. С него можно скачивать исследования, если есть прямой айпи и если на стороне клиента открыт порт 11112. Так же я проверил работу на свободном dicom-архиве dcm4che sourceforge.net/projects/cdmedicpacsweb/files/latest/download?source=files.
Мне удалось собрать рабочую версию для Linux, к сожалению собрать под Widows мне не удалось. Библиотека grassroot успешно собралась, но ошибка возникает при линковке самого приложения.
cmd/ld: Malformed PE file: Unexpected flags for PE section.
Об этой ошибке много написано тут: github.com/golang/go/issues/4069.
К сожалению я не настолько знаком с тонкостями сборки и поэтому получилась версия только под Linux. Может «хабра-эфект» сдвинет с мертвой точки и эту проблему. Для пользователей Windows, которые хотят проверить и посмотреть как это работает, я подготовил виртуальную машину на базе CoreOS (https://yadi.sk/d/y81KC-tyfar6A). В демо машине наш dicom-клиент работает как systemd-сервис.
При наличии желания, можно например реализовать сервис, который выкачивает исследования с различных dicom-узлов, и выкладывает в zip-архиве для скачивания. Для управления сервисом можно использовать json-сообщения, так же, как делает наш GUI.
А можно поступить, как поступил я: прикрутить в наше приложение какой-нибудь веб-просмотровщик на базе html5.
