.gitignore

esecached
*.6
*.8

esecached.go

package main

import (
    "os";
    "net";
    "fmt";
)

type kvsBackendConn struct {
    command int;
    key []byte;
    data_type int;
    value []byte;
    resp_ch chan<- *kvsBackendConn;
}

func backend(ch <-chan *kvsBackendConn) {
    kvs := make(map[string] []byte);
    for req := range ch {
        switch (req.command) {
        case 0: // get
            value, e := kvs[string(req.key)];
            if e {
                req.resp_ch<- (&kvsBackendConn{command:0, value:value})
            } else {
                req.resp_ch<- (&kvsBackendConn{command:0, value:nil})
            }
            break;
        case 1: //set
            kvs[string(req.key)] = req.value;
            break;
        }
    }
}

func buildResponse(opcode int, status int, data_type int, message_id int, CAS int64, extra []byte, key []byte, value[]byte) ([]byte)
{
    var extra_len int = 0;
    var key_len int = 0;
    var value_len int = 0;

    if extra != nil { extra_len = len(extra); }
    if key != nil  { key_len = len(key); }
    if value != nil  { value_len = len(value); }

    body_len := extra_len + key_len + value_len;

    resp := make([]byte, 24 + body_len);

    resp[0] = 0x81;
    resp[1] = byte(opcode);
    resp[2] = byte(status >> 8);
    resp[3] = byte(status);
    resp[4] = byte(extra_len);
    resp[5] = byte(data_type);
    resp[6] = 0;
    resp[7] = 0;
    resp[8] = byte(body_len >> 24);
    resp[9] = byte(body_len >> 16);
    resp[10] = byte(body_len >> 8);
    resp[11] = byte(body_len);
    resp[12] = byte(message_id >> 24);
    resp[13] = byte(message_id >> 16);
    resp[14] = byte(message_id >> 8);
    resp[15] = byte(message_id);
    resp[16] = byte(CAS >> 56);
    resp[17] = byte(CAS >> 48);
    resp[18] = byte(CAS >> 40);
    resp[19] = byte(CAS >> 32);
    resp[20] = byte(CAS >> 24);
    resp[21] = byte(CAS >> 16);
    resp[22] = byte(CAS >> 8);
    resp[23] = byte(CAS);
    copySlice(resp[24:24 + extra_len], extra);
    copySlice(resp[24 + extra_len:24 + extra_len + key_len], key);
    copySlice(resp[24 + extra_len + key_len:24 + body_len], value);

    return resp;
}

func handleConn(ch net.Conn, backend_ch chan *kvsBackendConn, debug bool) {
    defer ch.Close();

    header := make([]byte, 24);
    recv_ch := make(chan *kvsBackendConn);

    for {
        var l int;
        var e os.Error;
        for {
            l, e = ch.Read(header);
            switch e {
            case nil: break;
            case os.EOF: goto Quit_Loop;
            case os.EAGAIN: continue;
            default:
                error("Error during receiving a header (%s)", e.String());
                goto Quit_Loop;
            }
            break
        }
        if l != len(header) {
            error("Something must've gone wrong (got %d bytes: %s)", l, hexdump(header, l));
            goto Quit_Loop;
        }
        if header[0] != 0x80 {
            error("Invalid magic (%d: %s)", header[0], hexdump(header, l));
            goto Quit_Loop;
        }

        opcode       := int(header[1]);
        key_length   := (int(header[2]) << 8) | int(header[3]);
        extra_length := int(header[4]);
        data_type    := int(header[5]);
        body_length  := (int(header[8]) << 24)
                        | (int(header[9]) << 16)
                        | (int(header[10]) << 8)
                        | int(header[11]);
        message_id   := (int(header[12]) << 24)
                        | (int(header[13]) << 16)
                        | (int(header[14]) << 8)
                        | int(header[15]);
        CAS          := (int64(header[16]) << 56)
                        | (int64(header[17]) << 48)
                        | (int64(header[18]) << 40)
                        | (int64(header[19]) << 32)
                        | (int64(header[20]) << 24)
                        | (int64(header[21]) << 16)
                        | (int64(header[22]) << 8)
                        | int64(header[23]);

        if debug {
            fmt.Fprintf(os.Stderr,
                "Opcode: %d\n"
                "Datatype: %d\n"
                "Total Body length: %d\n"
                "    Key length: %d\n"
                "    Extra length: %d\n"
                "Message ID: %d\n"
                "CAS: %d\n",
                opcode, data_type, body_length, key_length, extra_length,
                message_id, CAS);
        }

        var data []byte;
        if body_length > 0 {
            data = make([]byte, body_length);
            for {
                l, e = ch.Read(data);
                switch e {
                case nil: break;
                case os.EAGAIN: continue;
                default:
                    error("Error during receiving body (%s)", e.String());
                    goto Quit_Loop;
                }
                break
            }
        }

        extra_data := data[0:extra_length];
        key := data[extra_length:extra_length+key_length];
        value := data[extra_length+key_length:body_length];

        if debug {
            fmt.Fprintf(os.Stderr,
                "Extra data:\n    %s\n"
                "Key:\n    %s\n"
                "Value:\n    %s\n",
                hexdump(extra_data, len(extra_data)),
                hexdump(key, len(key)),
                hexdump(value, len(value)));
        }

        var resp []byte;
        switch opcode {
        case 0x00: // get
            backend_ch<- (&kvsBackendConn{ command: 0, key: key, value: nil , resp_ch: recv_ch });
            result := <-recv_ch;
            resp_extra := make([]byte, 4);
            if result != nil {
                resp = buildResponse(opcode, 0, result.data_type, message_id, CAS, resp_extra, nil, result.value)
            } else {
                resp = buildResponse(opcode, 1, 0, message_id, CAS, resp_extra, nil, nil)
            }
            break;
        case 0x0c: // getk
            backend_ch<- (&kvsBackendConn{ command: 0, key: key, value: nil, resp_ch: recv_ch });
            result := <-recv_ch;
            resp_extra := make([]byte, 4);
            if result != nil {
                resp = buildResponse(opcode, 0, result.data_type, message_id, CAS, resp_extra, result.key, result.value)
            } else {
                resp = buildResponse(opcode, 1, 0, message_id, CAS, resp_extra, nil, nil)
            }
            break;
        case 0x01: // set
            backend_ch<- (&kvsBackendConn{ command: 1, key: key, data_type: data_type, value: value });
            resp = buildResponse(opcode, 0, 0, message_id, CAS + 1, nil, nil, nil);
            break;
        case 0x07:
            goto Quit_Loop;
        default:
            error("Unsupported opcode: %d", opcode);
        }
        for {
            l, e = ch.Write(resp);
            switch e {
            case nil: break;
            case os.EAGAIN: continue;
            default:
                error("Error during receiving body (%s)", e.String());
                goto Quit_Loop;
            }
            break
        }
    }

Quit_Loop:
}

func main() {
    if len(os.Args) < 2 {
        fmt.Fprintf(os.Stderr, "usage: %s addr:port\n", os.Args[0]);
        os.Exit(255);
    }

    l, e := net.Listen("tcp", os.Args[1]);
    if e != nil {
        error("An error occurred (%s)", e.String());
        os.Exit(1);
    }

    // blocking channel
    backend_ch := make(chan *kvsBackendConn);

    go backend(backend_ch);

    for {
        ch, e := l.Accept();
        if e != nil {
            error("An error occurred (%s)", e.String());
            os.Exit(1);
        }
        go handleConn(ch, backend_ch, false);
    }
    os.Exit(0)
}

Makefile

GOC = 6g
GOL = 6l

all: esecached

esecached: esecached.6
	$(GOL) -o $@ $^

esecached.6: esecached.go utils.go
	$(GOC) $^

utils.go

package main

import (
    "os";
    "fmt";
)

// stolen from bufio
func copySlice(dst []byte, src []byte) {
    for i := 0; i < len(dst); i++ {
        dst[i] = src[i]
    }
}

func error(format string, v ...) {
    fmt.Fprintf(os.Stderr, os.Args[0] + ": " + format + "\n", v)
}

func hexdump(data []byte, len int)(string) {
    dump := "";
    for i := 0; i < len; i++ {
        dump += fmt.Sprintf("%02x", int(data[i]))
    }
    return dump;
}