?

а я уже запарился конкурентный код писать..

боюсь, будут рейсы

рейсы по wg?

не

нам нужен двухпозиционный триггер

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

и сняться с нее, когда скажут

это, собственно, означает обратную связь

мы из горутины должны проинформировать управляющую процедуру, что вся обработка закончена, и мы готовы к новой итерации

так?

переписал свой пример

хотя, фигня всёравно выходит.

но я тоже поужинал и теперь уже не помошник.

короче..

package main import ( "bufio" "fmt" "log" "os" "sync" "time" ) type Dam struct { lock sync.Mutex queue []chan struct{} } func NewDam() *Dam { return &Dam{ lock: sync.Mutex{}, queue: make([]chan struct{}, 0), } } func (dam *Dam) Await(timeout time.Duration) error { newChan := make(chan struct{}) dam.lock.Lock() dam.queue = append(dam.queue, newChan) dam.lock.Unlock() select { case <-newChan: return nil case <-time.After(timeout): return fmt.Errorf("timeout!") } } func (dam *Dam) Flush() int { dam.lock.Lock() defer dam.lock.Unlock() totalCount := len(dam.queue) for _, waitChan := range dam.queue { close(waitChan) } // Reset queue dam.queue = make([]chan struct{}, 0) return totalCount } func main() { dam := NewDam() go func() { // Flush the dam every 3 seconds for { time.Sleep(3 * time.Second) totalFlushed := dam.Flush() log.Println("Dam flushed! accumulated:", totalFlushed) } }() scanner := bufio.NewScanner(os.Stdin) for scanner.Scan() { go func() { creationTime := time.Now() fmt.Println("Added", creationTime) if err := dam.Await(time.Second * 2); err != nil { fmt.Println("timed out", creationTime) return } fmt.Println("Finished!", creationTime) }() } if err := scanner.Err(); err != nil { panic(err) } }

реализация дамбы

думаю теперь как убрать слайс

слайс с локом - это очень похоже на канал. ?

слайс с локом - это очень похоже на канал. ?

https://play.golang.org/p/G1-6obuQVcD но это не работает..

Await блокирует главный поток, не даёт flush вызвать пока не затаймаутит

из-за совместного лока поверх канала

без лока здесь тоже нельзя поскольку происходит мутация канала которая должна быть синхронизирована, нельзя ресетить канал, пока кто-то его прослушивает

я вот забыл спросить - это все зачем?

я вот забыл спросить - это все зачем?

для реализация автоконекта в webwire

эмм

а подробнее?

там довольно сложный механизм автоматического восстановления соединения, который значительно упрощает работу с сокетами. как только клиент понимает что сервер ушёл - он запускает горутину по восстановлению соединения, которая при завершении "флашит дамбу" а дамбда это по сути queue запросов от клиента. Т.е. user code пишет в очередь запросов с помощью client.Request(timeout) которые либо таймаутят выходя с очереди дамбы, либо пробиваются до сервера когда дамба флашится (когда горутине по восстановлению связи удаётся подсоединиться)

приложение например выполнило 4 запроса на сервер, но вдруг, прям до этого, сервер пошёл на перезагрузку.. timeout у запросов скажем 60 секунд. Если в течении 60 секунд сервер перезапускается - то все запросы встреливают, если нет то возвращают timeout ошибку

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

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

в смысле синхронно, блокирующе?

ага

тогда придётся самому ручками писать перезапуска... при каждой ошибке прописывать reconnect логику

чего-то я не понимаю

чего я хочу предотвратить.. webwire client даёт гарантию что соединение будет восстановлено автоматически, тебе как пользователю библиотеки об этом даже думать не надо

зачем вообще такой странный механизм, с дамбой?

но

у нас есть буферизованный канал - это очередь запросов

зачем вообще такой странный механизм, с дамбой?

чтоб запросы, которые аккумулировались во время отсуствия сервера - выстрелили как только "дамба прорвётся" т.е. соединение восстановится

у нас есть соединение, которое, если порвалось, занимается восстановлением, а не чтением из канала

а как восстановилось - читает из канала

почему нельзя сделать так?

у нас есть буферизованный канал - это очередь запросов

это не совсем то, буферизированный канал выстреливает тогда, когда набирается определённое колво элементов, а тут дамба пробивается когда соединение восстанавливается и колво конкурентных запросов изначально неизвестно

а?!

что значит "канал выстреливает"?

пишет

в receiver

как нейрон

ох

есть соединение

так?

так

есть горутина читатель и горутина писатель

так?

нет

а как?

множество горутин могут одновременно делать запросы на сервер: https://github.com/qbeon/webwire-go/blob/master/examples/echo/client/client.go#L45 webwire больше похож на дуплексный HTTP, нежели на традиционные сокеты, сокет он абстрагирует

а это надо зачем-то?

почему не сделать одну горутину, которая делает запросы, и не засылать ей запросы каналом?

т.е. следственно мы можем параллельно замультиплексить несаколько запросов: for reqData := range requestsToMake { go func() { reply, err := client.Request("givemesomething", reqData) switch err := err.(type) { case nil: // got a reply! case webwire.ReqErr: // got an error reply case webwire.ReqTimeoutErr: // request timed out } }() }

никакой параллельности на одном соединение не будет палюбэ

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

еще раз

никакой параллельности на одном соединение не будет палюбэ

multiplexing, т.е. псевдопараллельность.

как HTTP2

ERROR: S client not available

это означает, что запускать более одной горутины-писателя не имеет смысла

правильно?

Там zmq рекламировали, как консервативную штуку

Тогда уж наследника юзать

http://nanomsg.org/

И на Go полная pure имплементация

https://github.com/go-mangos/mangos

а это надо зачем-то?

по сути это замена HTTP2 которая умеет в обе стороны... с HTTP2 нам пришлось бы посылать запросы по HTTP, а сигналы от сервера получать по websocket, потому-что ни у HTTP1.1 ни HTTP2 нет обратной связи, собственно почему сокеты и появились на свет... но это два отдельных канала, их нужно отдельно аутентифицировать и много ещё нюансов, которые мне все дико надоели и я решил свершить мини-революцию, реализовать абстракцию над вебсокетами, которая даст аутентификацию, сессии, автовосстановление как соединения так и сессии, request-reply, graceful shutdown и прочее другое..

собственно получилось https://github.com/qbeon/webwire-go

а... ну и конечно-же оффициальная поддержка JavaScript чтоб всё как из одной литой формы

короче чуя я надо FAQ писать

я не понимаю, зачем там асинхронность

все остальное я понимаю хорошо

а вот асинхронность - нет

Там zmq рекламировали, как консервативную штуку

zmq это про другое, nanomsg и mangos тоже

я не понимаю, зачем там асинхронность

мм, всмысле?

Нет, они об одном и том же

Просто различается громодзкость решения

в webwire другой набор фич реализован, в README я надеюсь всё описал)

еще раз - зачем асинхронность?

самый плизкий конкурент это socket.io но сокет.ио не обладает нужным набором фич и оффициальной поддержкой Go ну и многое другое..

еще раз - зачем асинхронность?

асинхронность в чём конкретно? не понимаю вопрос))

в чем угодно

еще центрифуга похожа

"asynchronous duplex messaging library" - вот почему asynchronous?

асинхронность в чём конкретно? не понимаю вопрос))

Попробую объяснить ?

Там нет стримов

Всегда request-reply

"asynchronous duplex messaging library" - вот почему asynchronous?

ну во первых multiplexing.. мы не ждём синхронно пока получим ответ на запрос А, а посылаем запрос Б и получаем возможно даже ответ на Б ранее А это не синхронная коммуникация

самый плизкий конкурент это socket.io но сокет.ио не обладает нужным набором фич и оффициальной поддержкой Go ну и многое другое..

чем отличается от центрифуги?

чем отличается от центрифуги?

ссылочку плиз) не слышал о "центрифуге"