diff --git a/ethdb/mysql/config.go b/ethdb/mysql/config.go new file mode 100644 index 0000000000..b77a77fbb9 --- /dev/null +++ b/ethdb/mysql/config.go @@ -0,0 +1,53 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package mysql + +import ( + "fmt" +) + +type Config struct { + Protocol string `toml:",omitempty"` + Address string `toml:",omitempty"` + Port string `toml:",omitempty"` + User string `toml:",omitempty"` + Password string `toml:",omitempty"` + Database string `toml:",omitempty"` + AllowNativePasswords bool `toml:",omitempty"` +} + +var DefaultConfig = Config{ + Protocol: "tcp", + Address: "localhost", + Password: "my-pw", + Port: "3306", + User: "root", + Database: "db0", + AllowNativePasswords: true, +} + +func (o *Config) String() string { + return fmt.Sprintf( + "%s:%s@%s(%s:%s)/%s?charset=utf8&parseTime=True&loc=Local&allowNativePasswords=%v", + o.User, + o.Password, + o.Protocol, + o.Address, + o.Port, + o.Database, + o.AllowNativePasswords) +} diff --git a/ethdb/mysql/database.go b/ethdb/mysql/database.go new file mode 100644 index 0000000000..4a41ca968f --- /dev/null +++ b/ethdb/mysql/database.go @@ -0,0 +1,199 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package mysql + +import ( + "fmt" + "strings" + "sync/atomic" + "time" + + "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/log" + "github.com/jinzhu/gorm" + _ "github.com/jinzhu/gorm/dialects/mysql" +) + +var ( + ethDB *gorm.DB // share a MySQL instance to avoid too many connections + instanceCount int32 +) + +type database struct { + db *gorm.DB + tableName string +} + +type data struct { + Key []byte `gorm:"unique_index"` + Value []byte `gorm:"type:BLOB"` +} + +// NewDatabase returns a MySQL wrapped object. +func NewDatabase(name string, cfg *Config) (ethdb.Database, error) { + if atomic.CompareAndSwapInt32(&instanceCount, 0, 1) { + // Open db + connectionString := cfg.String() + db, err := gorm.Open("mysql", connectionString) + if err != nil { + return nil, err + } + // Hide the db log + db.LogMode(false) + + ethDB = db + } else { + atomic.AddInt32(&instanceCount, 1) + } + + if !ethDB.HasTable(name) { + if err := ethDB.Table(name).CreateTable(&data{}).Error; err != nil { + if err != nil { + return nil, err + } + } + } + + return &database{ + db: ethDB.Table(name), + tableName: name, + }, nil +} + +// Put puts the given key / value to the queue +func (db *database) Put(key []byte, value []byte) error { + err := putOnMySQL(db.db, db.tableName, key, value) + return err +} + +func (db *database) Has(key []byte) (bool, error) { + if err := db.db.Where(&data{ + Key: key, + }).First(&data{}).Error; gorm.IsRecordNotFoundError(err) { + return false, nil + } else if err != nil { + return false, err + } + return true, nil +} + +// Get returns the given key if it's present. +func (db *database) Get(key []byte) ([]byte, error) { + out := &data{} + err := db.db.Where(&data{ + Key: key, + }).First(out).Error + if err != nil { + return nil, err + } + return out.Value, nil +} + +// Delete deletes the key from the queue and database +func (db *database) Delete(key []byte) error { + err := db.db.Where(&data{ + Key: key, + }).Delete(&data{}).Error + // Hide not found error + if gorm.IsRecordNotFoundError(err) { + return nil + } + return err +} + +func (db *database) Close() { + if atomic.CompareAndSwapInt32(&instanceCount, 1, 0) { + ethDB.Close() + ethDB = nil + } else { + atomic.AddInt32(&instanceCount, -1) + } +} + +// NewBatch create a db transaction to batch insert +func (db *database) NewBatch() ethdb.Batch { + return &batch{ + database: db, + transaction: db.db.Begin(), + } +} + +type batch struct { + *database + transaction *gorm.DB + size int + finished bool +} + +func (b *batch) Put(key, value []byte) (err error) { + defer func() { + // Update size if success, or rollback it + if err == nil { + b.size += len(value) + } else if !b.finished { + b.finished = true + b.transaction.Rollback() + } + }() + return putOnMySQL(b.transaction, b.tableName, key, value) +} + +func (b *batch) Write() (err error) { + // This transaction is finished before. There is no data so ignore it. + if b.finished { + return nil + } + b.finished = true + return b.transaction.Commit().Error +} + +func (b *batch) ValueSize() int { + return b.size +} + +func (b *batch) Reset() { + // Rollback previous transaction + if !b.finished { + b.transaction.Rollback() + } + b.transaction = b.db.Begin() + b.size = 0 + b.finished = false +} + +// putOnMySQL replaces the record if exists, or insert a new one +func putOnMySQL(db *gorm.DB, tableName string, key []byte, value []byte) (err error) { + for count := 0; ; { + err := db.Exec(fmt.Sprintf("REPLACE into %s VALUES(?, ?)", tableName), key, value).Error + if err == nil { + return nil + } + + // Retry if it's transaction deadlock error. + // https://dev.mysql.com/doc/refman/5.7/en/innodb-deadlock-example.html + if strings.Contains(err.Error(), "Error 1213") { + if count == 10 { + return err + } + count++ + log.Debug("Failed to put due to transaction deadlock error", "table", tableName, "count", count, "err", err) + time.Sleep(time.Duration(count*500) * time.Millisecond) + } else { + return err + } + } +}