Ну так у вас в /builder/inserter/inserter.go:143 паника, смотрите, что там могло выйти за пределы массива

flows_replace := make([]string, len(flow_fields))
 for i := range s.flowsForCount {
   flows_replace[i] = fmt.Sprintf("$%v", i+1)
   }

var fmsg flow.FlowMessage


 ts := time.Unix(int64(fmsg.TimeFlowStart), 0)
   srcip := net.IP(fmsg.SrcAddr)
   dstip := net.IP(fmsg.DstAddr)
   srcipstr := srcip.String()
   dstipstr := dstip.String()
   if srcipstr == "<nil>" {
     srcipstr = "0.0.0.0"
   }
   if dstipstr == "<nil>" {
     dstipstr = "0.0.0.0"
   }

   extract := []interface{}{
     "NOW()",
     ts,
     fmsg.Type,
     fmsg.SamplingRate,
     srcipstr,
     dstipstr,
     fmsg.Bytes,
     fmsg.Packets,
     fmsg.SrcPort,
     fmsg.DstPort,
     fmsg.Etype,
     fmsg.Proto,
     fmsg.SrcAS,
     fmsg.DstAS,
   }
   s.flowsForCount = append(s.flowsForCount, extract)
   s.flows = append(s.flows, extract)


 flows_replace := make([]string, len(flow_fields))
 for i := range s.flowsForCount {
   flows_replace[i] = fmt.Sprintf("$%v", i+1)
   }

s.flowsForCount = append(s.flowsForCount, extract)
Возможно у вас тут flowsForCount не пустой изначально, поэтому после append его длина больше, чем длина flows_replace

А почему не
for i :=0;i< len(flows_replace);i++{
   flows_replace[i] = fmt.Sprintf("$%v", i+1)
   }
?

Добрый день, есть сотни воркеров и тысячи (скажем 100 тыс) задач которые нужно выполнить.  Сейчас задачи складывают в очередь и воркеры их выполняют.

Очередь реализована как

lim = semaphore.NewWeighted(int64(workers))

--- // код воркера // ----
lim.Acquire(task.ctx, 1)
тут код воркера , а потом релиз семафора.

контекст задачи - это task.ctx = context.WithTimeout(.....)
----

все работает как нужно, вот только  проблема в том, что задачи идут не в FIFO. и получается что задача которая зашла одной из первых может протухнуть так и не дождавшись выполнения.

можно реализовать на каналах, но тогда не будет работать контекст.

Какую либу можно посмотреть чтобы и FIFO было и контексты работали?

А чем канал не подходит?

запульнул я в канал 100 тыс задач, у каждой из них разный таймаут выполнения.
например задача с индексом 50000 протухла по таймауту, мне нужно об этом узнать как можно скорее. А как это сделать в случае с каналами?

ведь пока 49999 задач впереди нее не обработаются (предположим у них огромный таймаут), я об этом не узнаю

у тебя размер s.flowsForCount больше flows_replace

Так два канала, один с задачами, второй с ответами от задач

можно пояснить? не улавливаю мысль, точнее как это поможет узнать что задача протухла?

/usr/local/go/src/net/http/server.go:3013 +0x39b
2021/06/03 15:13:38 http: panic serving 194.36.150.113:46598: runtime error: index out of range [4] with length 4
goroutine 38 [running]:
net/http.(*conn).serve.func1(0xc000136320)
 /usr/local/go/src/net/http/server.go:1824 +0x153
panic(0x9b6220, 0xc0000c6720)
 /usr/local/go/src/runtime/panic.go:971 +0x499
main.(*state).Stage2Handler(0xc00011a200, 0xaa7af0, 0xc000166620, 0xc000122c00)
 /builder/inserter/inserter.go:152 +0xbaf
net/http.HandlerFunc.ServeHTTP(0xc000091b50, 0xaa7af0, 0xc000166620, 0xc000122c00)
 /usr/local/go/src/net/http/server.go:2069 +0x44
github.com/go-chi/chi.(*Mux).routeHTTP(0xc0000ad320, 0xaa7af0, 0xc000166620, 0xc000122c00)
 /go/pkg/mod/github.com/go-chi/chi@v1.5.4/mux.go:436 +0x28b
net/http.HandlerFunc.ServeHTTP(0xc000091b40, 0xaa7af0, 0xc000166620, 0xc000122c00)
 /usr/local/go/src/net/http/server.go:2069 +0x44
github.com/go-chi/chi.(*Mux).ServeHTTP(0xc0000ad320, 0xaa7af0, 0xc000166620, 0xc000122b00)
 /go/pkg/mod/github.com/go-chi/chi@v1.5.4/mux.go:87 +0x331
net/http.serverHandler.ServeHTTP(0xc000166000, 0xaa7af0, 0xc000166620, 0xc000122b00)
 /usr/local/go/src/net/http/server.go:2887 +0xa3
net/http.(*conn).serve(0xc000136320, 0xaa8ef8, 0xc0000d5280)
 /usr/local/go/src/net/http/server.go:1952 +0x8cd
created by net/http.(*Server).Serve
 /usr/local/go/src/net/http/server.go:3013 +0x39b
2021/06/03 15:13:43 http: panic serving 194.36.150.113:46640: runtime error: index out of range [4] with length 4
goroutine 40 [running]:
net/http.(*conn).serve.func1(0xc0001363c0)
 /usr/local/go/src/net/http/server.go:1824 +0x153
panic(0x9b6220, 0xc0000c6840)
 /usr/local/go/src/runtime/panic.go:971 +0x499
main.(*state).Stage2Handler(0xc00011a200, 0xaa7af0, 0xc0001668c0, 0xc000122f00)
 /builder/inserter/inserter.go:152 +0xbaf
net/http.HandlerFunc.ServeHTTP(0xc000091b50, 0xaa7af0, 0xc0001668c0, 0xc000122f00)
 /usr/local/go/src/net/http/server.go:2069 +0x44
github.com/go-chi/chi.(*Mux).routeHTTP(0xc0000ad320, 0xaa7af0, 0xc0001668c0, 0xc000122f00)
 /go/pkg/mod/github.com/go-chi/chi@v1.5.4/mux.go:436 +0x28b
net/http.HandlerFunc.ServeHTTP(0xc000091b40, 0xaa7af0, 0xc0001668c0, 0xc000122f00)
 /usr/local/go/src/net/http/server.go:2069 +0x44
github.com/go-chi/chi.(*Mux).ServeHTTP(0xc0000ad320, 0xaa7af0, 0xc0001668c0, 0xc000122e00)
 /go/pkg/mod/github.com/go-chi/chi@v1.5.4/mux.go:87 +0x331
net/http.serverHandler.ServeHTTP(0xc000166000, 0xaa7af0, 0xc0001668c0, 0xc000122e00)
 /usr/local/go/src/net/http/server.go:2887 +0xa3
net/http.(*conn).serve(0xc0001363c0, 0xaa8ef8, 0xc0000d5400)
 /usr/local/go/src/net/http/server.go:1952 +0x8cd
created by net/http.(*Server).Serve
 /usr/local/go/src/net/http/server.go:3013 +0x39b
time="2021-06-03T15:13:45Z" level=info msg="Processed 0 r

Теперь смотрите, что у вас в 152ой строке

тебе ошибка index out of range непонятна? Вопрос сейчас в чём?

У вас i + 1 выходит за пределы. Вот допустим длинна 4 и когда берется индекс 3, то [3 + 1] = [4], а 4-го индекса не существует

Ничем. Проблема в том, что у вас задачи протухают. И этот факт должен отслеживать продьюсер этих задач

поясните пожалуйста.

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

Кто тут должен отслеживать?

мемкеш 😳

что-то вы странное говорите или я вас не понимаю. memcache это просто хранилище, как оно что-то может отслеживать?

но по сути мы уходим от вопроса, вопрос был в том, есть ли какая либа которая реализует семафоры c поддержкой контекстов и гарантирует FIFO?  semaphore.Weighted - все есть, за исключением FIFO

Выкинь мемкэш, юзай редис, там очереди есть да и в скорости не уступает