pb.Message.Marshal (and possibly others) not thread-safe

[srijs]

Hi.

I'm running into a crash when load testing a database POC based on the etcd/raft library. It seems like there is an error in the message serialization logic (is it autogenerated from the proto file?).

The crash is reproducible (after varyings amount of load). I will try to isolate the error a bit more to provide a small reproducing test case. Posting this now in case any body has an idea where to look specifically.

Stack trace:

panic: runtime error: index out of range

goroutine 4 [running]:
github.com/coreos/etcd/raft/raftpb.encodeVarintRaft(0xc8208b0c00, 0xbae, 0xbae, 0xbae, 0xf, 0x0)
        /Users/sreis/go/src/github.com/coreos/etcd/raft/raftpb/raft.pb.go:555 +0x70
github.com/coreos/etcd/raft/raftpb.(*Message).MarshalTo(0xc820ad7d00, 0xc8208b0c00, 0xbae, 0xbae, 0xbae, 0x0, 0x0)
        /Users/sreis/go/src/github.com/coreos/etcd/raft/raftpb/raft.pb.go:401 +0x374
github.com/coreos/etcd/raft/raftpb.(*Message).Marshal(0xc820ad7d00, 0xc8208b0c00, 0xbae, 0xbae, 0x0, 0x0)
        /Users/sreis/go/src/github.com/coreos/etcd/raft/raftpb/raft.pb.go:367 +0xa0
main.(*Transport).writeRaftMsg(0xc820074300, 0xc820ad7d00, 0x0, 0x0)
        /Users/sreis/Repositories/folddb2/shard/transport.go:85 +0x42
main.(*Transport).runWriteLoop(0xc820074300, 0x794570, 0xc820078660, 0x0, 0x0)
        /Users/sreis/Repositories/folddb2/shard/transport.go:111 +0x168
main.(*Transport).Run.func2(0xc820074300, 0x794570, 0xc820078660, 0xc82000e140)
        /Users/sreis/Repositories/folddb2/shard/transport.go:147 +0x35
created by main.(*Transport).Run
        /Users/sreis/Repositories/folddb2/shard/transport.go:149 +0xc3

goroutine 1 [chan send]:
main.(*State).send(0xc82007cb40, 0xc82025c800, 0x8, 0x8)
        /Users/sreis/Repositories/folddb2/shard/state.go:49 +0x106
main.(*State).run(0xc82007cb40, 0x794570, 0xc820078660, 0x0, 0x0)
        /Users/sreis/Repositories/folddb2/shard/state.go:112 +0x894
main.single(0xc820096080, 0xc820053e30, 0x3, 0x3, 0x0, 0x0)
        /Users/sreis/Repositories/folddb2/shard/main.go:32 +0x695
main.main()
        /Users/sreis/Repositories/folddb2/shard/main.go:77 +0x492

goroutine 17 [syscall, 2 minutes, locked to thread]:
runtime.goexit()
        /usr/local/go/src/runtime/asm_amd64.s:1696 +0x1

goroutine 20 [syscall, 2 minutes]:
os/signal.loop()
        /usr/local/go/src/os/signal/signal_unix.go:22 +0x18
created by os/signal.init.1
        /usr/local/go/src/os/signal/signal_unix.go:28 +0x37

goroutine 21 [select, 2 minutes, locked to thread]:
runtime.gopark(0x294690, 0xc82002bf28, 0x235c50, 0x6, 0x2f918, 0x2)
        /usr/local/go/src/runtime/proc.go:185 +0x163
runtime.selectgoImpl(0xc82002bf28, 0x0, 0x18)
        /usr/local/go/src/runtime/select.go:392 +0xa64
runtime.selectgo(0xc82002bf28)
        /usr/local/go/src/runtime/select.go:212 +0x12
runtime.ensureSigM.func1()
        /usr/local/go/src/runtime/signal1_unix.go:227 +0x323
runtime.goexit()
        /usr/local/go/src/runtime/asm_amd64.s:1696 +0x1

goroutine 22 [chan receive, 2 minutes]:
main.main.func1(0xc820076120, 0xc820078890)
        /Users/sreis/Repositories/folddb2/shard/main.go:52 +0x68
created by main.main
        /Users/sreis/Repositories/folddb2/shard/main.go:59 +0x1e4

goroutine 23 [semacquire, 2 minutes]:
sync.runtime_Semacquire(0xc82000e14c)
        /usr/local/go/src/runtime/sema.go:43 +0x26
sync.(*WaitGroup).Wait(0xc82000e140)
        /usr/local/go/src/sync/waitgroup.go:126 +0xb4
main.(*Transport).Run(0xc820074300, 0x794570, 0xc820078660)
        /Users/sreis/Repositories/folddb2/shard/transport.go:151 +0xd1
created by main.single
        /Users/sreis/Repositories/folddb2/shard/main.go:25 +0x59d

goroutine 3 [chan send]:
main.(*Transport).runReadLoop(0xc820074300, 0x794570, 0xc820078660, 0x0, 0x0)
        /Users/sreis/Repositories/folddb2/shard/transport.go:78 +0x1b9
main.(*Transport).Run.func1(0xc820074300, 0x794570, 0xc820078660, 0xc82000e140)
        /Users/sreis/Repositories/folddb2/shard/transport.go:142 +0x35
created by main.(*Transport).Run
        /Users/sreis/Repositories/folddb2/shard/transport.go:144 +0x83

goroutine 25 [select]:
github.com/coreos/etcd/raft.(*node).run(0xc82008e5a0, 0xc8200b20e0)
        /Users/sreis/go/src/github.com/coreos/etcd/raft/node.go:275 +0x1383
created by github.com/coreos/etcd/raft.StartNode
        /Users/sreis/go/src/github.com/coreos/etcd/raft/node.go:178 +0x73d

goroutine 26 [select]:
main.(*State).runStatusBroadcasting(0xc82007cb40, 0x794570, 0xc820078660, 0x0, 0x0)
        /Users/sreis/Repositories/folddb2/shard/state.go:61 +0x151
created by main.(*State).run
        /Users/sreis/Repositories/folddb2/shard/state.go:72 +0x82

[srijs]

Because I suspected a race condition, I instrumented the function that calls Marshal to print the message sizes before and after the crash. It turns out, when calling Size on the message before mashalling it returns a smaller number than when called after the panic.

Code:

func (t *Transport) writeRaftMsg(raftMsg *raftpb.Message) error {
    mSize := raftMsg.Size()
    defer func() {
        if r := recover(); r != nil {
            log.Println("Recovered: ", r)
            log.Println("Message Sizes: ", mSize, raftMsg.Size())
            log.Println("Message: ", raftMsg)
        }
    }()
    mData, err := raftMsg.Marshal()
    if err != nil {
        return err
    }
    return WriteMessage(t.w, &Message{MessageTypeRaft, ShardNodeId{t.id.shardId, raftMsg.To}, mData})
}

(Truncated) Output:

2016/01/26 22:30:18 Recovered:  runtime error: index out of range
2016/01/26 22:30:18 Message Sizes:  1069 4673
2016/01/26 22:30:18 Message:  type:MsgApp to:2 from:1 term:2 logTerm:2 in....

[srijs]

So, my hypothesis so far: When Marshal gets called, it calls Size, which returns 1069. Marshal uses this to allocate the buffer. At some point in while it is marshalling, before it started marshalling the entries, the entries are appended to. When it finally reaches the entries, it iterates through them and tries to write them to the buffer. But since the number entries has grown since then, it needs more space in the buffer than it has, and runs into the index out of range error.

I should probably mention at this point that messages are sent asynchronously in my case, not blocking the main raft node loop. But imo it should not cause that race condition in any case.

[srijs]

Okay, I now pulled all message serialization logic into the main node loop, and only perform the sending of the messages outside. Running the load test several times I was not able to reproduce the crash with this setup. My conclusion would be that marshalling the messages is not thread-safe and needs to run in the same thread as the main raft loop.

Where do we want to go from here? Do we just extend the documentation to clarify this restriction, or is it possible to fix the race condition by either copying or locking (both of which could negatively impact the performance of etcd)?

[heyitsanthony]

You're right the Marshal protobuf code doesn't have any synchronization; a message has to be wrapped in a mutex to work as a shared structure. I believe the reasoning for not generating any locking code is it's expensive and not strictly necessary. etcd avoids having to lock while marshaling a message by either implicitly passing the message ownership or by copying (e.g., raft/node.go:Node.Ready() will send copies) so that only one thread may access the message at a time. It's definitely safe to marshal messages outside the raft event loop.

Have you tried using the golang race detector? https://golang.org/doc/articles/race_detector.html

[srijs]

Thanks for the pointer to the data race detector, this is definitely helpful!

I'm still thinking about how to prevent others from running into the same problem. I'm considering whether add something like this to raft/doc.go would help:

Note: Marshalling messages is not thread-safe; it is important that you
make sure that no new entries are persisted while marshalling.
The easiest way to achieve this is to serialise the messages directly inside
your main raft loop.

Would you accept a PR with that change?

[xiang90]

@srijs I am fine with adding it. But please find a good place for it. The doc is long enough :)