Philip O'Toole
2 years ago
committed by
GitHub
commit
bcd32493c6
@ -0,0 +1,22 @@
|
|||||||
|
# Queued Writes API
|
||||||
|
> :warning: **This functionality was introduced in version 7.5. It does not exist in earlier releases.**
|
||||||
|
|
||||||
|
rqlite exposes a special API, which will queue up write-requests and execute them in bulk. This allows clients to send multiple distinct requests to a rqlite node, and have rqlite automatically do the batching and bulk insert for the client, without the client doing any extra work. This functionality is best illustrated by an example, showing two requests being queued.
|
||||||
|
```bash
|
||||||
|
curl -XPOST 'localhost:4001/db/execute/queue/_default' -H "Content-Type: application/json" -d '[
|
||||||
|
["INSERT INTO foo(name) VALUES(?)", "fiona"],
|
||||||
|
["INSERT INTO foo(name) VALUES(?)", "sinead"]
|
||||||
|
]'
|
||||||
|
curl -XPOST 'localhost:4001/db/execute/queue/_default' -H "Content-Type: application/json" -d '[
|
||||||
|
["INSERT INTO foo(name) VALUES(?)", "declan"]
|
||||||
|
]'
|
||||||
|
```
|
||||||
|
rqlite will merge these requests, and execute them as though they had been both contained in a single request. For the same reason that using the [Bulk API](https://github.com/rqlite/rqlite/blob/master/DOC/BULK.md) results in much higher write performance, using the _Queued Writes_ API will also result in much higher write performance.
|
||||||
|
|
||||||
|
The behaviour of the queue rqlite uses to batch the requests is configurable at rqlite launch time. Pass `-h` to `rqlited` to see the queue defaults, and list all configuration options.
|
||||||
|
|
||||||
|
## Caveats
|
||||||
|
Because the API returns immediately after queuing the requests **but before the data is commited to the SQLite database** there is a risk of data loss in the event the node crashes before queued data is persisted.
|
||||||
|
|
||||||
|
Like most databases there is a trade-off to be made between write-performance and durability. In addition, when the API returns `HTTP 200 OK`, that simply acknowledges that the data has been queued correctly. It does not indicate that the SQL statements will actually be applied successfully to the database. Be sure to check the node's logs if you have any concerns about failed queued writes.
|
||||||
|
|
||||||
@ -0,0 +1,120 @@
|
|||||||
|
package queue
|
||||||
|
|
||||||
|
import (
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/rqlite/rqlite/command"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Queue is a batching queue with a timeout.
|
||||||
|
type Queue struct {
|
||||||
|
maxSize int
|
||||||
|
batchSize int
|
||||||
|
timeout time.Duration
|
||||||
|
|
||||||
|
batchCh chan *command.Statement
|
||||||
|
sendCh chan []*command.Statement
|
||||||
|
C <-chan []*command.Statement
|
||||||
|
|
||||||
|
done chan struct{}
|
||||||
|
closed chan struct{}
|
||||||
|
flush chan struct{}
|
||||||
|
|
||||||
|
// Whitebox unit-testing
|
||||||
|
numTimeouts int
|
||||||
|
}
|
||||||
|
|
||||||
|
// New returns a instance of a Queue
|
||||||
|
func New(maxSize, batchSize int, t time.Duration) *Queue {
|
||||||
|
q := &Queue{
|
||||||
|
maxSize: maxSize,
|
||||||
|
batchSize: batchSize,
|
||||||
|
timeout: t,
|
||||||
|
batchCh: make(chan *command.Statement, maxSize),
|
||||||
|
sendCh: make(chan []*command.Statement, maxSize),
|
||||||
|
done: make(chan struct{}),
|
||||||
|
closed: make(chan struct{}),
|
||||||
|
flush: make(chan struct{}),
|
||||||
|
}
|
||||||
|
|
||||||
|
q.C = q.sendCh
|
||||||
|
go q.run()
|
||||||
|
return q
|
||||||
|
}
|
||||||
|
|
||||||
|
// Write queues a request.
|
||||||
|
func (q *Queue) Write(stmt *command.Statement) error {
|
||||||
|
if stmt == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
q.batchCh <- stmt
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Flush flushes the queue
|
||||||
|
func (q *Queue) Flush() error {
|
||||||
|
q.flush <- struct{}{}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Close closes the queue. A closed queue should not be used.
|
||||||
|
func (q *Queue) Close() error {
|
||||||
|
select {
|
||||||
|
case <-q.done:
|
||||||
|
default:
|
||||||
|
close(q.done)
|
||||||
|
<-q.closed
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Depth returns the number of queue requests
|
||||||
|
func (q *Queue) Depth() int {
|
||||||
|
return len(q.batchCh)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Stats returns stats on this queue.
|
||||||
|
func (q *Queue) Stats() (map[string]interface{}, error) {
|
||||||
|
return map[string]interface{}{
|
||||||
|
"max_size": q.maxSize,
|
||||||
|
"batch_size": q.batchSize,
|
||||||
|
"timeout": q.timeout,
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (q *Queue) run() {
|
||||||
|
defer close(q.closed)
|
||||||
|
var stmts []*command.Statement
|
||||||
|
timer := time.NewTimer(q.timeout)
|
||||||
|
timer.Stop()
|
||||||
|
|
||||||
|
writeFn := func() {
|
||||||
|
newStmts := make([]*command.Statement, len(stmts))
|
||||||
|
copy(newStmts, stmts)
|
||||||
|
q.sendCh <- newStmts
|
||||||
|
|
||||||
|
stmts = nil
|
||||||
|
timer.Stop()
|
||||||
|
}
|
||||||
|
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case s := <-q.batchCh:
|
||||||
|
stmts = append(stmts, s)
|
||||||
|
if len(stmts) == 1 {
|
||||||
|
timer.Reset(q.timeout)
|
||||||
|
}
|
||||||
|
if len(stmts) >= q.batchSize {
|
||||||
|
writeFn()
|
||||||
|
}
|
||||||
|
case <-timer.C:
|
||||||
|
q.numTimeouts++
|
||||||
|
writeFn()
|
||||||
|
case <-q.flush:
|
||||||
|
writeFn()
|
||||||
|
case <-q.done:
|
||||||
|
timer.Stop()
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -0,0 +1,206 @@
|
|||||||
|
package queue
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/rqlite/rqlite/command"
|
||||||
|
)
|
||||||
|
|
||||||
|
var testStmt = &command.Statement{
|
||||||
|
Sql: "SELECT * FROM foo",
|
||||||
|
}
|
||||||
|
|
||||||
|
func Test_NewQueue(t *testing.T) {
|
||||||
|
q := New(1, 1, 100*time.Millisecond)
|
||||||
|
if q == nil {
|
||||||
|
t.Fatalf("failed to create new Queue")
|
||||||
|
}
|
||||||
|
defer q.Close()
|
||||||
|
}
|
||||||
|
|
||||||
|
func Test_NewQueueWriteNil(t *testing.T) {
|
||||||
|
q := New(1, 1, 60*time.Second)
|
||||||
|
defer q.Close()
|
||||||
|
|
||||||
|
if err := q.Write(nil); err != nil {
|
||||||
|
t.Fatalf("failing to write nil: %s", err.Error())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func Test_NewQueueWriteBatchSizeSingle(t *testing.T) {
|
||||||
|
q := New(1024, 1, 60*time.Second)
|
||||||
|
defer q.Close()
|
||||||
|
|
||||||
|
if err := q.Write(testStmt); err != nil {
|
||||||
|
t.Fatalf("failed to write: %s", err.Error())
|
||||||
|
}
|
||||||
|
|
||||||
|
select {
|
||||||
|
case stmts := <-q.C:
|
||||||
|
if len(stmts) != 1 {
|
||||||
|
t.Fatalf("received wrong length slice")
|
||||||
|
}
|
||||||
|
if stmts[0].Sql != "SELECT * FROM foo" {
|
||||||
|
t.Fatalf("received wrong SQL")
|
||||||
|
}
|
||||||
|
case <-time.NewTimer(5 * time.Second).C:
|
||||||
|
t.Fatalf("timed out waiting for statement")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func Test_NewQueueWriteBatchSizeMulti(t *testing.T) {
|
||||||
|
q := New(1024, 5, 60*time.Second)
|
||||||
|
defer q.Close()
|
||||||
|
|
||||||
|
// Write a batch size and wait for it.
|
||||||
|
for i := 0; i < 5; i++ {
|
||||||
|
if err := q.Write(testStmt); err != nil {
|
||||||
|
t.Fatalf("failed to write: %s", err.Error())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case stmts := <-q.C:
|
||||||
|
if len(stmts) != 5 {
|
||||||
|
t.Fatalf("received wrong length slice")
|
||||||
|
}
|
||||||
|
if q.numTimeouts != 0 {
|
||||||
|
t.Fatalf("queue timeout expired?")
|
||||||
|
}
|
||||||
|
case <-time.NewTimer(5 * time.Second).C:
|
||||||
|
t.Fatalf("timed out waiting for first statements")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Write one more than a batch size, should still get a batch.
|
||||||
|
for i := 0; i < 6; i++ {
|
||||||
|
if err := q.Write(testStmt); err != nil {
|
||||||
|
t.Fatalf("failed to write: %s", err.Error())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case stmts := <-q.C:
|
||||||
|
if len(stmts) < 5 {
|
||||||
|
t.Fatalf("received too-short slice")
|
||||||
|
}
|
||||||
|
if q.numTimeouts != 0 {
|
||||||
|
t.Fatalf("queue timeout expired?")
|
||||||
|
}
|
||||||
|
case <-time.NewTimer(5 * time.Second).C:
|
||||||
|
t.Fatalf("timed out waiting for second statements")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func Test_NewQueueWriteTimeout(t *testing.T) {
|
||||||
|
q := New(1024, 10, 1*time.Second)
|
||||||
|
defer q.Close()
|
||||||
|
|
||||||
|
if err := q.Write(testStmt); err != nil {
|
||||||
|
t.Fatalf("failed to write: %s", err.Error())
|
||||||
|
}
|
||||||
|
|
||||||
|
select {
|
||||||
|
case stmts := <-q.C:
|
||||||
|
if len(stmts) != 1 {
|
||||||
|
t.Fatalf("received wrong length slice")
|
||||||
|
}
|
||||||
|
if stmts[0].Sql != "SELECT * FROM foo" {
|
||||||
|
t.Fatalf("received wrong SQL")
|
||||||
|
}
|
||||||
|
if q.numTimeouts != 1 {
|
||||||
|
t.Fatalf("queue timeout didn't expire")
|
||||||
|
}
|
||||||
|
case <-time.NewTimer(5 * time.Second).C:
|
||||||
|
t.Fatalf("timed out waiting for statement")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Test_NewQueueWriteTimeoutMulti ensures that timer expiring
|
||||||
|
// twice in a row works fine.
|
||||||
|
func Test_NewQueueWriteTimeoutMulti(t *testing.T) {
|
||||||
|
q := New(1024, 10, 1*time.Second)
|
||||||
|
defer q.Close()
|
||||||
|
|
||||||
|
if err := q.Write(testStmt); err != nil {
|
||||||
|
t.Fatalf("failed to write: %s", err.Error())
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case stmts := <-q.C:
|
||||||
|
if len(stmts) != 1 {
|
||||||
|
t.Fatalf("received wrong length slice")
|
||||||
|
}
|
||||||
|
if stmts[0].Sql != "SELECT * FROM foo" {
|
||||||
|
t.Fatalf("received wrong SQL")
|
||||||
|
}
|
||||||
|
if q.numTimeouts != 1 {
|
||||||
|
t.Fatalf("queue timeout didn't expire")
|
||||||
|
}
|
||||||
|
case <-time.NewTimer(5 * time.Second).C:
|
||||||
|
t.Fatalf("timed out waiting for first statement")
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := q.Write(testStmt); err != nil {
|
||||||
|
t.Fatalf("failed to write: %s", err.Error())
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case stmts := <-q.C:
|
||||||
|
if len(stmts) != 1 {
|
||||||
|
t.Fatalf("received wrong length slice")
|
||||||
|
}
|
||||||
|
if stmts[0].Sql != "SELECT * FROM foo" {
|
||||||
|
t.Fatalf("received wrong SQL")
|
||||||
|
}
|
||||||
|
if q.numTimeouts != 2 {
|
||||||
|
t.Fatalf("queue timeout didn't expire")
|
||||||
|
}
|
||||||
|
case <-time.NewTimer(5 * time.Second).C:
|
||||||
|
t.Fatalf("timed out waiting for second statement")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Test_NewQueueWriteTimeoutBatch ensures that timer expiring
|
||||||
|
// followed by a batch, works fine.
|
||||||
|
func Test_NewQueueWriteTimeoutBatch(t *testing.T) {
|
||||||
|
q := New(1024, 2, 1*time.Second)
|
||||||
|
defer q.Close()
|
||||||
|
|
||||||
|
if err := q.Write(testStmt); err != nil {
|
||||||
|
t.Fatalf("failed to write: %s", err.Error())
|
||||||
|
}
|
||||||
|
|
||||||
|
select {
|
||||||
|
case stmts := <-q.C:
|
||||||
|
if len(stmts) != 1 {
|
||||||
|
t.Fatalf("received wrong length slice")
|
||||||
|
}
|
||||||
|
if stmts[0].Sql != "SELECT * FROM foo" {
|
||||||
|
t.Fatalf("received wrong SQL")
|
||||||
|
}
|
||||||
|
if q.numTimeouts != 1 {
|
||||||
|
t.Fatalf("queue timeout didn't expire")
|
||||||
|
}
|
||||||
|
case <-time.NewTimer(5 * time.Second).C:
|
||||||
|
t.Fatalf("timed out waiting for statement")
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := q.Write(testStmt); err != nil {
|
||||||
|
t.Fatalf("failed to write: %s", err.Error())
|
||||||
|
}
|
||||||
|
if err := q.Write(testStmt); err != nil {
|
||||||
|
t.Fatalf("failed to write: %s", err.Error())
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case stmts := <-q.C:
|
||||||
|
// Should happen before the timeout expires.
|
||||||
|
if len(stmts) != 2 {
|
||||||
|
t.Fatalf("received wrong length slice")
|
||||||
|
}
|
||||||
|
if stmts[0].Sql != "SELECT * FROM foo" {
|
||||||
|
t.Fatalf("received wrong SQL")
|
||||||
|
}
|
||||||
|
if q.numTimeouts != 1 {
|
||||||
|
t.Fatalf("queue timeout expired?")
|
||||||
|
}
|
||||||
|
case <-time.NewTimer(5 * time.Second).C:
|
||||||
|
t.Fatalf("timed out waiting for statement")
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Reference in New Issue