time.After() memory leaks?

[Jingcheng Zhang]

The following code leaks memory whenever I connect to poster(), which pushes memory to all clients of pusher().

It doesn't collect garbage even when I connect to gc(), which runs runtime.GC().

If I remove the case time.After(), no memory leaks.

Anybody knows why?

package main

import (
"flag"
"net"
"time"
"runtime"
)

var pusherPort, posterPort, gcPort string

func main() {
flag.StringVar(&pusherPort, "p", "6666", "pusher port")
flag.StringVar(&posterPort, "a", "6667", "poster port")
flag.StringVar(&gcPort, "g", "6668", "gc port")
flag.Parse()
go pusher()
go poster()
go gc()
select {}
}

func pusher() {
l, err := net.Listen("tcp", ":" + pusherPort)
if err != nil {
panic(err.Error())
}
for {
conn, err := l.Accept()
if err != nil {
panic(err.Error())
}
server := newServer(conn)
servers[server] = server
go server.serve()
}
}

func poster() {
l, err := net.Listen("tcp", ":" + posterPort)
if err != nil {
panic(err.Error())
}
s := "haha"
for {
conn, err := l.Accept()
if err != nil {
panic(err.Error())
}
for _, server := range servers {
server.push(s)
}
conn.Close()
}
}

func gc() {
l, err := net.Listen("tcp", ":" + gcPort)
if err != nil {
panic(err.Error())
}
for {
conn, err := l.Accept()
if err != nil {
panic(err.Error())
}
runtime.GC()
conn.Close()
}
}




var servers = map[*server]*server{}

type server struct {
conn net.Conn
queue chan string
}

func newServer(conn net.Conn) *server {
return &server{conn : conn, queue : make(chan string, 4)}
}

func (this *server) serve() {
defer func () {
this.conn.Close()
delete(servers, this)
}()
for {
select {
case s := <- this.queue:
_, err := this.conn.Write([]byte(s))
if err != nil {
return
}
case <-time.After(5 * time.Minute):
break
}
}
}

func (this *server) push(s string) {
this.queue <- s
}

--
Best regards,
Jingcheng Zhang
Beijing, P.R.China

[John Beisley]

Every time you call time.After, it allocates a new time.Timer object.
There's nowhere that you clear this up. I would suggest that you use
time.NewTimer instead so that you can call the underlying Stop()
method on the timer object to let it be cleaned up.

As an additional note, you don't appear to have any protection around
the servers map. I suggest that you put a mutex or some other
protection around it (a sync.RWMutex might work nicely) so that it's
not read/written from multiple goroutines at the same time.

[roger peppe]

On 19 June 2012 09:50, Jingcheng Zhang <dio...@gmail.com> wrote:

> The following code leaks memory whenever I connect to poster(), which pushes
> memory to all clients of pusher().
> It doesn't collect garbage even when I connect to gc(), which runs
> runtime.GC().
>
> If I remove the case time.After(), no memory leaks.
> Anybody knows why?

Each time you leave a time.After even waiting to fire, there's
a small amount of heap allocated until it does fire.
That means you'll end up with 300 * events/second
items allocated in your serve loop, which could be
very large if you're accepting a high rate of connections.

I recommend using time.NewTimer and stopping the
timer when the timer does not time out.

Another problem your code has is that you're
concurrently sharing access to the servers map.

Here's an alternate phrasing of the functionality in
your code that is perhaps a little more idiomatic.

package main

import (
"flag"

"log"
"net"
"time"
)

func main() {
pusherPort := flag.String("p", "6666", "pusher port")
posterPort := flag.String("a", "6667", "poster port")
flag.Parse()

pusher, err := accepter(*pusherPort)
if err != nil {
log.Fatal(err)
}
poster, err := accepter(*posterPort)
if err != nil {
log.Fatal(err)
}
done := make(chan *server)
servers := make(map[*server]*server)
for {
select {
case conn, ok := <-pusher:
if !ok {
return
}
server := newServer(conn, done)

servers[server] = server
go server.serve()

case conn, ok := <-poster:
if !ok {
return

}
for _, server := range servers {

server.push("haha")
}
conn.Close()
case server := <-done:
delete(servers, server)
server.close()
}
}
}

func accepter(port string) (<-chan net.Conn, error) {
l, err := net.Listen("tcp", ":"+port)
if err != nil {
return nil, err
}
conns := make(chan net.Conn)
go func() {
defer l.Close()
defer close(conns)

for {
conn, err := l.Accept()
if err != nil {

log.Printf("accept %v: %q", port, err)
return
}
conns <- conn
}
}()
return conns, nil

}

type server struct {
conn net.Conn
queue chan string

done chan<- *server
}

func newServer(conn net.Conn, done chan<- *server) *server {
return &server{
conn: conn,
queue: make(chan string, 4),
done: done,
}
}

func (srv *server) serve() {
defer func() {
srv.done <- srv
}()
for {
t := time.NewTimer(5 * time.Minute)
select {
case s := <-srv.queue:
t.Stop()
_, err := srv.conn.Write([]byte(s))

if err != nil {
return
}
case <-time.After(5 * time.Minute):

return

}
}
}

func (this *server) push(s string) {
this.queue <- s
}

func (srv *server) close() error {
return srv.conn.Close()
}

[Jingcheng Zhang]

Hello John,

Thanks for you reply. I was not aware of cleaning the Timer object, I thought it will clean itself when it returns.

For the servers map, I will pass it between goroutines through channel, in favour of the rule "Don't communicate by shared memory. Instead, share memory by communicating". By the way, if there are a lot of goroutines, is this way efficient enough? 

[Jingcheng Zhang]

Thank you, your modifications look complicated, I'll learn different thoughts from it after I fixed my memory leak issue. 

[John Asmuth]

On Tuesday, June 19, 2012 12:16:05 PM UTC-4, Jingcheng Zhang wrote:

Thanks for you reply. I was not aware of cleaning the Timer object, I thought it will clean itself when it returns.

It does get cleaned up eventually, once its time has ellapsed. If you stop waiting for it, it has yet to ellapse, and will continue to occupy heap space.