unlimited buffered channel

chan.go

package utils

import (
  "log"
  "reflect"
)

func NewChan(in, out interface{}) {
  inValue := reflect.ValueOf(in)
  outValue := reflect.ValueOf(out)
  if inValue.Kind() != reflect.Chan || outValue.Kind() != reflect.Chan {
    log.Fatal("NewChan: argument is not a chan")
  }
  go func() {
    defer outValue.Close()
    headNode := new(element)
    headNode.next = headNode
    head := headNode
    tail := headNode
    for {
      if tail != head {
        chosen, v, ok := reflect.Select([]reflect.SelectCase{reflect.SelectCase{
          Dir: reflect.SelectSend,
          Chan: outValue,
          Send: tail.value,
        }, reflect.SelectCase{
          Dir: reflect.SelectRecv,
          Chan: inValue,
        }})
        if chosen == 0 { // out
          tail = tail.next
        } else {
          if !ok { return }
          e := &element{value: v}
          if head == tail {
            tail = e
          }
          e.next = head
          head.next.next = e
          head.next = e
        }
      } else {
        _, v, ok := reflect.Select([]reflect.SelectCase{reflect.SelectCase{
          Dir: reflect.SelectRecv,
          Chan: inValue,
        }})
        if !ok { return }
        e := &element{value: v}
        if head == tail {
          tail = e
        }
        e.next = head
        head.next.next = e
        head.next = e
      }
    }
  }()
}

type element struct {
  value reflect.Value
  next *element
}

chan_test.go

package utils

import (
  "testing"
  "fmt"
  "runtime"
)

func TestChan(t *testing.T) {
  for ci := 0; ci < 10; ci++ {
    in := make(chan string)
    out := make(chan string)
    NewChan(in, out)
    n := 1000
    var memStats runtime.MemStats
    for i := 0; i < n; i++ {
      in <- fmt.Sprintf("%d", i)
    }
    for i := 0; i < n; i++ {
      s := <-out
      if s != fmt.Sprintf("%d", i) {
        t.Fail()
      }
      runtime.GC()
      if i % 100 == 0 {
        runtime.ReadMemStats(&memStats)
        fmt.Printf("%v %d\n", memStats.Alloc, runtime.NumGoroutine())
      }
    }
    close(in)
    runtime.GC()
    _, ok := <-out
    if ok { t.Fail() }
  }
  fmt.Printf("check whether memory or goroutine is leaking.\n")
}

func BenchmarkChan(b *testing.B) {
  in := make(chan int)
  out := make(chan int)
  NewChan(in, out)
  b.ResetTimer()
  for i := 0; i < b.N; i++ {
    in <- 5
    <-out
  }
}