OOM for 3K websockets about gaio HOT 7 OPEN

@xtaci any idea, why? looking at the profiler it shows somehow this list https://github.com/xtaci/gaio/blob/master/watcher.go#L563 is getting larger and larger and eventually runs out even though I have 8GB machine.

from gaio.

lemme look into this, somewhat wired.

from gaio.

@xtaci Please let me know if you need any more info. I can provide that pretty quickly since I spent few days tweaking it. Also, I have one question regarding async io call backs. It appears to me(although not 100% sure) that an aiocb object is created for every write request (ignoring reads for the sake of this discussion) if so, I am guessing what might be happening is that if we have slow clients the number of callbacks are piling up in the list pointed out in my previous comment and eventually running out of 8GB memory. so I tried changing write to writeTimeout of 100ms in the above code and that still didn't work. it is still causing OOM. Now the question I wanted to ask is, what happens if my server send a ton of write requests to all 3K sockets? would ii block at some point? if not, how to prevent OOM?

from gaio.

could you show me a code snippet which can re-appear the OOM

from gaio.

clients

aiocb is allocated via pool, and will be recycled

func init() {
    aiocbPool.New = func() interface{} {
        return new(aiocb)
    }
}

from gaio.

@xtaci its the same as pasted above. let me know if you need more clarification ? I am also pasting below. this is very similar to your push server example, except I am using map of connections instead of can array. OOM happens when there are 1K+ connections with the code below. On a high level the code is just trying broadcast each message from a message queue to all the connected clients concurrently.

        clients := map[string]net.Conn{}
	go func() {
		for {
			select {
			case res := <-chIO: // receive IO events from watcher
				if res.Error != nil {
					log.Error().Msgf("Error receiving IO event from watcher: %v", res.Error)
					delete(clients, res.Conn.RemoteAddr().String())
					err = w.Free(res.Conn)
					if err != nil {
						log.Error().Msgf("error freeing connection: %v", err)
					}
					continue
				}
			case feed := <-out:
				f := ws.NewTextFrame(feed)
				bts := CompileHeader(f.Header)
				for index, conn := range clients {
					if conn != nil {
						err = w.Write(nil, conn, bts)
						if err != nil {
							if errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNRESET) {
								delete(clients, index)
							} else {
								log.Error().Msgf("unable to write header: %v", err)
							}
						}
						err = w.Write(nil, conn, f.Payload)
						if err != nil {
							if errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNRESET) {
								delete(clients, index)
							} else {
								log.Error().Msgf("unable to write payload: %v", err)
							}
						}
					}
				}
			case conn := <-chConn: // receive new connection events
				clients[conn.RemoteAddr().String()] = conn
			}
		}
	}()

@xtaci let me know if there are specific parts of this code you want me to explain?

from gaio.

func init() {
    aiocbPool.New = func() interface{} {
        return new(aiocb)
    }
}

@xtaci what is the pool size here? and what happens if the pool is full? Does the event loop suspend its thread?

from gaio.