lockfree.go

package main

import (
        "bytes"
        "errors"
        "flag"
        "log"
        "os"
        "runtime"
        "strconv"
        "sync"
        "time"

        "github.com/spaolacci/murmur3"
)

const KeySize = 35

type keyT []byte

var nullKey = make(keyT, KeySize)

func (k keyT) IsZero() bool {
        return bytes.Equal(k[:], nullKey[:])
}

type MyHash interface {
        Get(keyT) int32
        Set(keyT, int32) (int32, error)
        Keys() []keyT
}

type myHash struct {
        keys   []byte // keySize * Size
        values []int32
        sync.RWMutex
}

func New(size int) MyHash {
        return &myHash{keys: make([]byte, KeySize*size), values: make([]int32, size)}
}

func (hash *myHash) Keys() []keyT {
        hash.RLock()
        defer hash.RUnlock()

        var keys []keyT
        N := len(hash.keys)
        for i := 0; i < N; i += KeySize {
                key := keyT(hash.keys[i : i+KeySize : i+KeySize])
                if !key.IsZero() {
                        keys = append(keys, key)
                }
        }
        return keys
}

func (hash *myHash) Get(key keyT) int32 {
        hash.RLock()
        defer hash.RUnlock()

        N := len(hash.values)
        i := int(murmur3.Sum32(key) % uint32(N))
        for j := i * KeySize; ; i, j = i+1, j+KeySize {
                if i >= N {
                        i, j = 0, 0
                }
                k := hash.keys[j : j+KeySize]
                if bytes.Equal(k, key) {
                        return hash.values[i]
                }
                if bytes.Equal(k, nullKey) {
                        return -1
                }
        }
}

func (hash myHash) Set(key keyT, value int32) (int32, error) {
        seen_zero := 0
        N := len(hash.values)
        i := int(murmur3.Sum32(key) % uint32(N))

        hash.Lock()
        defer hash.Unlock()

        for j := i * KeySize; ; i, j = i+1, j+KeySize {
                if i >= N {
                        i, j = 0, 0
                        seen_zero++
                        if seen_zero > 1 {
                                return int32(-1), errors.New("Hash table full. We looked and looked, couldn't find a spot.")
                        }
                }
                k := hash.keys[j : j+KeySize]
                if bytes.Equal(k, key) {
                        hash.values[i] = value
                        return value, nil
                } else if bytes.Equal(k, nullKey) {
                        copy(k, key)
                        hash.values[i] = value
                        return value, nil
                }
        }
}

var (
        pressure  = flag.Int("pressure", 5, "Amount of write pressure to apply.")
        tablesize = flag.Int("size", 3e5, "Hash Table Size")
        numkeys   = flag.Int("keys", 2.5e5, "Number of Keys. Go larger than tablesize to check full table behaviour.")
        maxprocs  = flag.Int("maxprocs", runtime.NumCPU()*6, "Max threads. Defaults to 6 * core")
)

func main() {
        flag.Parse()
        log.Println("Commencing test with pressure", *pressure, "table size", *tablesize, "keys", *numkeys)
        very_start := time.Now()
        runtime.GOMAXPROCS(*maxprocs)
        myhash := New(*tablesize)
        numthreads := runtime.NumCPU() * 6
        numsets := *numkeys / numthreads
        var wg sync.WaitGroup
        start := time.Now()

        for threadnum := 0; threadnum < numthreads; threadnum++ {
                wg.Add(1)
                //log.Println("goroutine for", threadnum, "with numsets", numsets)
                go func(i int) {
                        mykey := make(keyT, KeySize)
                        for counter := (numsets * i); counter < numsets*(i+1+*pressure); counter++ {
                                copy(mykey, "heyy"+strconv.Itoa(counter))
                                _, err := myhash.Set(mykey, int32(counter))
                                if err != nil {
                                        log.Println("Error setting:", err)
                                        os.Exit(1)
                                }
                        }
                        wg.Done()
                }(threadnum)
        }
        wg.Wait()
        log.Println("Done filling hashtable with ", numsets*numthreads, "items. Took", time.Since(start), "to fill")
        for i := int32(*tablesize / 2); i < int32(*tablesize/2)+5; i++ {
                var mykey keyT
                start = time.Now()
                copy(mykey[:], "heyy"+strconv.Itoa(int(i)))
                _ = myhash.Get(mykey)
                log.Println("Lookup took", time.Since(start))
        }
        start = time.Now()
        log.Println("Commencing Garbage Collection.")
        runtime.GC()
        log.Println("Done. Took", time.Since(start))

        start = time.Now()
        log.Println("Commencing Second Garbage Collection.")
        runtime.GC()
        log.Println("Done. Took", time.Since(start))

        log.Println("All done with test hash containing", len(myhash.Keys()), ". Total runtime", time.Since(very_start))
}