freecache.go

package main

import (
	"errors"
	"flag"
	"fmt"
	"log"
	"net"
	"net/textproto"
	"os"
	"os/signal"
	"runtime"
	"runtime/debug"
	"strconv"
	"strings"
	"sync"
	"syscall"

	"github.com/coocood/freecache"
)

var (
	mu       = &sync.RWMutex{}
	kv       = make(map[string][]byte)
	cache    *freecache.Cache
	errProto = errors.New("ERROR")
	listen   = flag.String("l", "", "Interface to listen on. Default to all addresses.")
	network  = flag.String("n", "tcp", "Network to listen on (tcp,tcp4,tcp6,unix). unix not tested! Default is tcp")
	port     = flag.Int("p", 11211, "TCP port number to listen on (default: 11211)")
	threads  = flag.Int("t", runtime.NumCPU(), fmt.Sprintf("number of threads to use (default: %d)", runtime.NumCPU()))
)

func main() {
	flag.Parse()
	cacheSize := 100 * 1024 * 1024

	cache = freecache.NewCache(cacheSize)
	debug.SetGCPercent(20)
	runtime.GOMAXPROCS(*threads)
	address := fmt.Sprintf("%s:%d", *listen, *port)

	listener, err := net.Listen(*network, address)
	if err != nil {
		log.Fatalf("failed to serve: %s", err.Error())
		return
	}
	//go onKill(listener)
	defer listener.Close()
	fmt.Printf("Server is listening on %s %s", *network, address)
	for {

		conn, err := listener.Accept()
		if err != nil {
			fmt.Println(err)
			conn.Close()
			continue
		}
		go handleConnection(textproto.NewConn(conn)) // запускаем горутину для обработки сокета
	}
}

// обработка мемкеш протокола по заветам Ильича - Фитцпатрика
// https://github.com/memcached/memcached/blob/master/doc/protocol.txt
func handleConnection(c *textproto.Conn) {
	defer c.Close()

	for {
		var err error
		// read one line (ended with \n or \r\n)
		line, err := c.ReadLine()
		// check protocol error
		if err != nil {
			if err.Error() != "EOF" {
				//network error and so on
			}
			break
		}
		if line == "" {
			// strange
			continue
		}
		// len args always more zero
		args := strings.Split(line, " ")

		// для кривых клиентов
		cmd := strings.ToLower(args[0])
		if cmd == "gets" {
			cmd = "get" // similar
		}
		switch cmd {
		case "set":
			// norepy is optional, but always last param
			noreply := false
			if args[len(args)-1] == "noreply" {
				noreply = true
			}
			if len(args) < 5 || args[1] == "" {
				// вот тут я странно пукнул насчет ключа пустого, но еще странней обнаружить там значение однажды
				err = errProto
				break
			}
			bytes, err := strconv.Atoi(args[4])
			if err != nil || bytes == 0 {
				//log.Println(bytes, err)
				err = errProto
				break
			}
			b := make([]byte, bytes)
			// тут читаем напрямик, так и быстрей и безопасно для байтов
			n, err := c.R.Read(b)

			if err != nil || n != bytes {
				//log.Println(n != bytes, err)
				err = errProto
				break
			}
			// не забудем про /r/n
			crlf := make([]byte, 2)
			_, err = c.R.Read(crlf)
			err = cache.Set([]byte(args[1]), b, 0)
			//mu.Lock()
			//kv[args[1]] = b
			//mu.Unlock()
			if !noreply {
				err = c.PrintfLine("STORED")
			}

		case "get":
			if len(args) < 2 {
				err = errProto
				break
			}
			for _, arg := range args[1:] {
				b, err := cache.Get([]byte(arg))
				//mu.RLock()
				//b, ok := kv[arg]
				//mu.RUnlock()
				if err == nil {
					// If some of the keys appearing in a retrieval request are not sent back
					// by the server in the item list this means that the server does not
					// hold items with such keys

					fmt.Fprintf(c.W, "VALUE %s 0 %d\r\n%s\r\n", arg, len(b), b)
					//err = c.PrintfLine("VALUE %s 0 %d\r\n%s\r\nEND", args[1], len(b), b)
				}
			}
			fmt.Fprintf(c.W, "END\r\n")
			// отцы копят в буфере, потом зараз кидают в трубу, для быстродействия полезно это
			// стандартный протокол шлет построчно - медленней в 2 раза будет
			err = c.W.Flush()

		case "delete":
			if len(args) < 2 {
				err = errProto
				break
			}
			ok := cache.Del([]byte(args[1]))
			//mu.RLock()
			//_, ok := kv[args[1]]
			//mu.RUnlock()
			if ok {
				//mu.Lock()
				//delete(kv, args[1])
				//mu.Unlock()
				c.Writer.PrintfLine("DELETED")

			} else {
				c.Writer.PrintfLine("NOT_FOUND")
			}

		case "close":
			err = errors.New("CLOSE")

		default:
			if len([]byte(cmd)) > 0 && !isASCIILetter([]byte(cmd)[0]) {
				// This is SPARTA!
				err = errors.New("CLOSE")
			} else {
				err = errProto
			}
			log.Println("default", line, cmd, len(line))

		}
		// если у нас ошибка в цикле - это повод соскочить с сокета
		if err != nil {
			if err == errProto {
				if c.PrintfLine("ERROR") != nil {
					// не удалось записать в сокет
					break
				}
				err = nil
				// можно считать кол-во ошибок и спрыгивать - если там мусор
				//break
			} else {
				if err.Error() != "CLOSE" {
					fmt.Println(err)
				}
				//Ошибка на сервере
				break
			}
		}
	}
}

func isASCIILetter(b byte) bool {
	b |= 0x20 // make lower case
	return 'a' <= b && b <= 'z'
}

func onKill(l net.Listener) {
	signalChan := make(chan os.Signal, 1) //https://go101.org/article/panic-and-recover-use-cases.html
	//SIGHUP: Process restart/reload (example: nginx, sshd, apache)? syscall.SIGUSR2?
	signal.Notify(signalChan, os.Interrupt, os.Kill, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT, syscall.SIGPIPE) //syscall.SIGINT, syscall.SIGTERM, syscall.SIGILL,

	q := <-signalChan
	if q == syscall.SIGPIPE {
		return
	}
	fmt.Printf("\nShutdown signal received, exiting...\n")
	l.Close()
	os.Exit(0)
}