I'm getting ThreadSanitizer warnings about data races with how I am using gRPC with async streaming RPCs. I assume I'm using gRPC incorrectly somehow but I am not sure how.
My setup has one thread processing the completion queue (repeatedly calling Next). I then have multiple threads running that may call ServerAsyncWriter::Write on different ServerAsyncWriter instances without synchronizing with CompletionQueue::Next. 2 or more threads may be calling write while another is simultaneously calling Next on the CQ.
I do ensure that the write tag is returned from the CQ for a given ServerAsyncWriter before writing again with that writer. No additional write is invoked for ServerAsyncWriter A until its tag is returned but a new write for ServerAsyncWriter B may be invoked before A's tag is returned.
The output from ThreadSanitizer is below.
WARNING: ThreadSanitizer: data race (pid=100231)
Write of size 1 at 0x7b54002dff2f by thread T4 (mutexes: write M628674345348136664, write M782359682632142904, write M654007093252083640, write M646970218834318416, write M770537733610302832):
#0 memcpy <null> (libtsan.so.0+0x324d5)
#1 grpc_slice_buffer_add <null> (app+0x7770ba)
#2 grpc::internal::CallOpSet<grpc::internal::CallOpSendInitialMetadata, grpc::internal::CallOpSendMessage, grpc::internal::CallOpServerSendStatus, grpc::internal::CallNoOp<4>, grpc::internal::CallNoOp<5>, grpc::internal::CallNoOp<6> >::FillOps(grpc::internal::Call*) <null> (app+0x4c0f01)
#3 grpc_impl::ServerAsyncWriter<pb::Data>::Write(pb::Data const&, void*) <null> (app+0x4cf2bc)
... rest of call stack
Previous write of size 1 at 0x7b54002dff2f by thread T20:
#0 memcpy <null> (libtsan.so.0+0x324d5)
#1 grpc_slice_buffer_add <null> (app+0x777146)
#2 execute_native_thread_routine <null> (app+0x8cc04e)
Location is heap block of size 576 at 0x7b54002dff00 allocated by thread T20:
#0 realloc <null> (libtsan.so.0+0x2b03b)
#1 gpr_realloc <null> (app+0x86de7d)
#2 grpc::internal::CallOpSet<grpc::internal::CallOpSendInitialMetadata, grpc::internal::CallOpSendMessage, grpc::internal::CallOpServerSendStatus, grpc::internal::CallNoOp<4>, grpc::internal::CallNoOp<5>, grpc::internal::CallNoOp<6> >::FillOps(grpc::internal::Call*) <null> (app+0x4c0f01)
#3 grpc_impl::ServerAsyncWriter<pb::ConfigStatus>::Write(pb::ConfigStatus const&, void*) <null> (app+0x4d0d1c)
... rest of call stack
The entry point into the gRPC API is two separate calls (different types, different instances) of Write.
My publishing threads are essentially executing this function:
template<typename TData>
struct Writer
{
void* tag;
grpc::ServerAsyncWriter<TData>* writer;
}
template<typename TData>
void publish(TData d)
{
Writer<TData>* writer = /* lookup existing writer */;
// Real code doesn't actually busy wait but it will not call write until the tag is returned
while (!writeTagHasReturned(writer->tag))
{}
writer->writer->Write(d, writer->tag);
}
Where am I using the API wrong? Is there something unsafe about the following things happening simultaneously?
- Thread 1: writer1->Write(data1, A), wait for A.
- Thread 2: writer2->Write(data2, B), wait for B.
- Thread 3: cq->Next(ok, tag), notify other threads that tag was returned