A G-Counter Implementation in Go (NOT TESTED)

gcounter.go

// Copyright (c) 2015, Kohei YOSHIDA <license@yosida95.com>. All rights reserved.

// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:

//     * Redistributions of source code must retain the above copyright
//       notice, this list of conditions and the following disclaimer.
//     * Redistributions in binary form must reproduce the above copyright
//       notice, this list of conditions and the following disclaimer in the
//       documentation and/or other materials provided with the distribution.
//     * Neither the name of the copyright holder nor the names of its
//       contributors may be used to endorse or promote products derived from
//       this software without specific prior written permission.

// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

package main

import (
	"strconv"
	"sync"
	"time"

	"github.com/garyburd/redigo/redis"
)

type GCounter interface {
	Update(string, int64) (int64, error)
	Query(string) (int64, error)
	Merge(string) int64
}

type cluster struct {
	sync.Mutex
	conns []redis.Conn
}

func NewCluster(addrs ...string) (ret GCounter, err error) {
	var wg sync.WaitGroup

	c := &cluster{
		conns: make([]redis.Conn, 0, len(addrs)),
	}

	for _, addr := range addrs {
		wg.Add(1)

		go func(addr string) {
			defer wg.Done()

			if err != nil {
				return
			}

			conn, errc := redis.DialTimeout("tcp", addr, time.Second, time.Second, time.Second)
			c.Lock()
			defer c.Unlock()

			if errc != nil {
				err = errc
				return
			}
			c.conns = append(c.conns, conn)
		}(addr)
	}
	wg.Wait()

	ret = c
	return
}

func (c *cluster) Update(name string, delta int64) (i int64, err error) {
	c.Lock()
	defer c.Unlock()

	var (
		mu     sync.Mutex
		wg     sync.WaitGroup
		numErr int
	)

	for _, conn := range c.conns {
		wg.Add(1)
		go func(conn redis.Conn) {
			defer wg.Done()
			ic, errc := redis.Int64(conn.Do("INCRBY", name, delta))
			mu.Lock()
			defer mu.Unlock()

			if errc != nil {
				err = errc
				numErr++
				return
			}

			if i < ic {
				i = ic
			}
		}(conn)
	}
	wg.Wait()

	if numErr < len(c.conns) {
		err = nil
	}
	return
}

func (c *cluster) Query(name string) (i int64, err error) {
	c.Lock()
	defer c.Unlock()

	for _, conn := range c.conns {
		var (
			repstr string
			rep    int64
		)

		repstr, err = redis.String(conn.Do("GET", name))
		if err != nil {
			continue
		}

		rep, err = strconv.ParseInt(repstr, 10, 64)
		if err != nil {
			continue
		}

		i = rep
		return
	}

	return
}

func (c *cluster) Merge(name string) (i int64) {
	c.Lock()
	defer c.Unlock()

	var (
		mu   sync.Mutex
		wg   sync.WaitGroup
		reps = make(map[redis.Conn]int64)
	)

	for _, conn := range c.conns {
		wg.Add(1)
		go func(conn redis.Conn) {
			defer wg.Done()

			repstr, err := redis.String(conn.Do("GET", name))
			mu.Lock()
			defer mu.Unlock()

			if err != nil {
				return
			}

			repi, err := strconv.ParseInt(repstr, 10, 64)
			if err != nil {
				return
			}

			reps[conn] = repi
			if i < repi {
				i = repi
			}
			return
		}(conn)
	}
	wg.Wait()

	return
}

type counter struct {
	sync.Mutex
	conns []GCounter
}

func (c *counter) Update(name string, delta int64) (i int64, err error) {
	c.Lock()
	defer c.Unlock()

	for _, conn := range c.conns {
		if i, err = conn.Update(name, delta); err == nil {
			break
		}
	}

	return
}

func (c *counter) Query(name string) (rep int64, successful, failure int) {
	c.Lock()
	defer c.Unlock()

	var (
		mu sync.Mutex
		wg sync.WaitGroup
	)

	for _, conn := range c.conns {
		wg.Add(1)
		go func(conn GCounter) {
			defer wg.Done()

			i, err := conn.Query(name)
			mu.Lock()
			defer mu.Unlock()

			if err != nil {
				failure++
				return
			}

			rep += i
			successful++
			return
		}(conn)
	}
	wg.Wait()

	return
}

func (c *counter) Merge(name string) (i int64) {
	c.Lock()
	defer c.Unlock()

	var (
		mu sync.Locker
		wg sync.WaitGroup
	)

	for _, conn := range c.conns {
		wg.Add(1)
		go func(conn GCounter) {
			defer wg.Done()

			rep := conn.Merge(name)
			mu.Lock()
			defer mu.Unlock()

			i += rep
		}(conn)
	}

	return
}