Merge pull request #33 from m-lab/sandbox-addcache

implement cached test

Author: yachang

caller.go

@@ -4,7 +4,6 @@ package main
 import (
 	"context"
 	"flag"
-	"fmt"
 	"time"
 
 	"github.com/m-lab/traceroute-caller/connection"
@@ -57,9 +56,7 @@ func main() {
 	if *tcpinfoSocket == "" {
 		connPoller := connectionpoller.New(cache)
 		for ctx.Err() == nil {
-			closedConnections := connPoller.GetClosedConnections()
-			fmt.Printf("length of closed connections: %d\n", len(closedConnections))
-			daemon.TraceAll(closedConnections)
+			connPoller.TraceClosedConnections(&daemon)
 
 			select {
 			case <-time.After(*waitTime):

connectionlistener/connectionlistener_test.go

@@ -28,10 +28,16 @@ type fakeTracer struct {
 	wg  sync.WaitGroup
 }
 
-func (ft *fakeTracer) Trace(conn connection.Connection, t time.Time) {
+func (ft *fakeTracer) Trace(conn connection.Connection, t time.Time) string {
 	log.Println("Tracing", conn)
 	ft.ips = append(ft.ips, conn.RemoteIP)
 	ft.wg.Done()
+	return "Fake test Result"
+}
+
+func (ft *fakeTracer) CreateCacheTest(conn connection.Connection, t time.Time, cachedTest string) {
+	log.Println("Create cached test for: ", conn)
+	return
 }
 
 func TestListener(t *testing.T) {

connectionpoller/connectionpoller.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/m-lab/traceroute-caller/connection"
 	"github.com/m-lab/traceroute-caller/ipcache"
+	"github.com/m-lab/traceroute-caller/scamper"
 )
 
 var (
@@ -136,25 +137,20 @@ type connectionPoller struct {
 // connections which it previously measured to be open, but it can no longer
 // measure to be open.
 type ConnectionPoller interface {
-	GetClosedConnections() []connection.Connection
+	TraceClosedConnections(tracer scamper.Tracer)
 }
 
-// GetClosedConnections returns the list of connections which were previously
-// measured to be open but were not measured to be open this time..
-func (c *connectionPoller) GetClosedConnections() []connection.Connection {
+// TraceClosedConnections send trace for all closed connections.
+func (c *connectionPoller) TraceClosedConnections(tracer scamper.Tracer) {
 	oldConn := c.connectionPool
 	fmt.Printf("old connection size %d\n", len(oldConn))
 	c.connectionPool = c.GetConnections()
 	fmt.Printf("new connection size %d\n", len(c.connectionPool))
-	var closed []connection.Connection
 	for conn := range oldConn {
-		if _, hasConn := c.connectionPool[conn]; !hasConn && !c.recentIPCache.Has(conn.RemoteIP) {
-			closed = append(closed, conn)
-			log.Printf("Try to add " + conn.RemoteIP)
-			c.recentIPCache.Add(conn.RemoteIP)
+		if _, hasConn := c.connectionPool[conn]; !hasConn {
+			go c.recentIPCache.Trace(conn, tracer)
 		}
 	}
-	return closed
 }
 
 // New creates and returns a new ConnectionPoller.

connectionpoller/connectionpoller_test.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"reflect"
 	"testing"
+	"time"
 
 	"github.com/m-lab/traceroute-caller/ipcache"
 
@@ -112,49 +113,51 @@ func TestGetConnectionsWithFakeSS(t *testing.T) {
 	}
 }
 
-func TestConnectionPollerConstruction(t *testing.T) {
-	// The only thing we can verify by default is that the code does not crash.
-	// Which is not nothing, but it's not a lot.
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	cache := ipcache.New(ctx)
-	connPoller := New(cache)
-	connPoller.GetClosedConnections()
+type testTracer struct {
+	calls   int
+	answers []map[connection.Connection]struct{}
+}
+
+func (tt *testTracer) Trace(conn connection.Connection, t time.Time) string {
+	return "Fake Trace test"
+}
+
+func (tt *testTracer) CreateCacheTest(conn connection.Connection, t time.Time, cachedTest string) {
+	return
 }
 
 type testFinder struct {
-	calls   int
-	answers []map[connection.Connection]struct{}
 }
 
 func (tf *testFinder) GetConnections() map[connection.Connection]struct{} {
-	calls := tf.calls
-	tf.calls++
-	return tf.answers[calls]
+	conns := make(map[connection.Connection]struct{})
+	return conns
 }
 
-func TestGetClosedCollection(t *testing.T) {
-	// This setup causes both conn3 and conn2 to disappear, but because conn3 is in
-	// the ipcache, only conn2 should be returned.
+func TestConnectionPollerConstruction(t *testing.T) {
+	// The only thing we can verify by default is that the code does not crash.
+	// Which is not nothing, but it's not a lot.
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	cache := ipcache.New(ctx)
-	connPoller := New(cache).(*connectionPoller)
-	conn1 := connection.Connection{RemoteIP: "1.1.1.1"}
-	conn2 := connection.Connection{RemoteIP: "1.1.1.2"}
-	conn3 := connection.Connection{RemoteIP: "1.1.1.3"}
-	connPoller.recentIPCache.Add(conn3.RemoteIP)
-	connPoller.finder = &testFinder{
-		answers: []map[connection.Connection]struct{}{
-			{conn1: struct{}{}, conn2: struct{}{}, conn3: struct{}{}},
-			{conn1: struct{}{}},
-		},
+	connPoller := &connectionPoller{
+		finder:         &testFinder{},
+		recentIPCache:  cache,
+		connectionPool: make(map[connection.Connection]struct{}),
 	}
-	connPoller.connectionPool = connPoller.GetConnections()
+	conn1 := connection.Connection{
+		RemoteIP:   "1.1.1.2",
+		RemotePort: 5034,
+		LocalIP:    "1.1.1.3",
+		LocalPort:  58790,
+		Cookie:     "10f3d"}
+	connPoller.connectionPool[conn1] = struct{}{}
+	var tt testTracer
+	connPoller.TraceClosedConnections(&tt)
 
-	c := connPoller.GetClosedConnections()
+	time.Sleep(200 * time.Millisecond)
 
-	if len(c) != 1 || c[0] != conn2 {
-		t.Errorf("Wanted %v but got %v", []connection.Connection{conn2}, c)
+	if connPoller.recentIPCache.GetCacheLength() != 1 {
+		t.Errorf("ConnectionPoller not working properly")
 	}
 }

ipcache/ipcache.go

@@ -7,6 +7,9 @@ import (
 	"fmt"
 	"sync"
 	"time"
+
+	"github.com/m-lab/traceroute-caller/connection"
+	"github.com/m-lab/traceroute-caller/scamper"
 )
 
 var (
@@ -17,23 +20,63 @@ var (
 	IPCacheUpdatePeriod = flag.Duration("IPCacheUpdatePeriod", 1*time.Second, "We run the cache eviction loop with this frequency")
 )
 
+type CacheTest struct {
+	timeStamp time.Time
+	data      string
+	done      chan struct{}
+}
+
 // RecentIPCache contains a list of all the IP addresses that we have traced to
 // recently. We keep this list to ensure that we don't traceroute to the same
 // location repeatedly at a high frequency.
 type RecentIPCache struct {
-	cache map[string]time.Time
+	cache map[string]*CacheTest
 	mu    sync.RWMutex
 }
 
+// GetTrace returns test content in []byte given a connection and tracer
+func (rc *RecentIPCache) Trace(conn connection.Connection, sc scamper.Tracer) {
+	ip := conn.RemoteIP
+	rc.mu.Lock()
+	c, ok := rc.cache[ip]
+	if !ok {
+		nc := &CacheTest{
+			timeStamp: time.Now(),
+			done:      make(chan struct{}),
+		}
+		rc.cache[ip] = nc
+		rc.mu.Unlock()
+
+		nc.data = sc.Trace(conn, nc.timeStamp)
+		close(nc.done)
+		return
+	}
+	rc.mu.Unlock()
+
+	<-c.done
+	cachedData := c.data
+	sc.CreateCacheTest(conn, time.Now(), cachedData)
+}
+
+func (rc *RecentIPCache) GetCacheLength() int {
+	return len(rc.cache)
+}
+
+func (rc *RecentIPCache) GetTestContent(ip string) string {
+	rc.mu.RLock()
+	defer rc.mu.RUnlock()
+	c, ok := rc.cache[ip]
+	if ok {
+		return c.data
+	}
+	return ""
+}
+
 // New creates and returns a RecentIPCache. It also starts up a background
 // goroutine that scrubs the cache.
-//
-// TODO(https://github.com/m-lab/traceroute-caller/issues/30) Make this truly threadsafe
 func New(ctx context.Context) *RecentIPCache {
 	m := &RecentIPCache{}
-	m.mu.Lock()
-	m.cache = make(map[string]time.Time)
-	m.mu.Unlock()
+	m.cache = make(map[string]*CacheTest)
 	go func() {
 		ticker := time.NewTicker(*IPCacheUpdatePeriod)
 		defer ticker.Stop()
@@ -42,7 +85,7 @@ func New(ctx context.Context) *RecentIPCache {
 				return
 			}
 			for k, v := range m.cache {
-				if now.Sub(v) > *IPCacheTimeout {
+				if now.Sub(v.timeStamp) > *IPCacheTimeout {
 					fmt.Println("try to delete " + k)
 					m.mu.Lock()
 					delete(m.cache, k)
@@ -69,8 +112,11 @@ func (m *RecentIPCache) Add(ip string) {
 	fmt.Printf("func Add: Now is %d\n", time.Now().Unix())
 	_, present := m.cache[ip]
 	if !present {
-		m.cache[ip] = time.Now()
-		fmt.Printf("just add %s %d\n", ip, m.cache[ip].Unix())
+		m.cache[ip] = &CacheTest{
+			timeStamp: time.Now(),
+			done:      make(chan struct{}),
+		}
+		fmt.Printf("just add %s %d\n", ip, m.cache[ip].timeStamp.Unix())
 	}
 }
 

ipcache/ipcache_test.go

@@ -2,12 +2,67 @@ package ipcache_test
 
 import (
 	"context"
+	"log"
 	"testing"
 	"time"
 
+	"github.com/m-lab/traceroute-caller/connection"
 	"github.com/m-lab/traceroute-caller/ipcache"
 )
 
+type testTracer struct {
+	calls   int
+	answers []map[connection.Connection]struct{}
+}
+
+func (tf *testTracer) Trace(conn connection.Connection, t time.Time) string {
+	log.Println("Create Trace Test")
+	return "Fake Trace test " + conn.RemoteIP
+}
+
+func (tf *testTracer) CreateCacheTest(conn connection.Connection, t time.Time, cachedTest string) {
+	log.Println("Create cached Test " + conn.RemoteIP)
+	return
+}
+
+func TestTrace(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	testCache := ipcache.New(ctx)
+
+	conn1 := connection.Connection{
+		RemoteIP:   "1.1.1.2",
+		RemotePort: 5034,
+		LocalIP:    "1.1.1.3",
+		LocalPort:  58790,
+		Cookie:     "10f3d"}
+	var tt testTracer
+
+	testCache.Trace(conn1, &tt)
+	time.Sleep(200 * time.Millisecond)
+	tmp := testCache.GetTestContent("1.1.1.2")
+	log.Println("!!!" + tmp + "!!!")
+	if tmp != "Fake Trace test 1.1.1.2" {
+		t.Error("cache not trace correctly ")
+	}
+
+	conn2 := connection.Connection{
+		RemoteIP:   "1.1.1.5",
+		RemotePort: 5034,
+		LocalIP:    "1.1.1.7",
+		LocalPort:  58790,
+		Cookie:     "aaaa"}
+
+	testCache.Trace(conn2, &tt)
+	testCache.Trace(conn1, &tt)
+
+	time.Sleep(200 * time.Millisecond)
+
+	if testCache.GetCacheLength() != 2 {
+		t.Error("cache not working correctly ")
+	}
+}
+
 func TestRecentIPCache(t *testing.T) {
 	*ipcache.IPCacheTimeout = 100 * time.Millisecond
 	*ipcache.IPCacheUpdatePeriod = 10 * time.Millisecond