Put span sink send into its own method for clarity
Separating this call into its own method helps clarify its role on the
critical path, as well as the relationship to the state machine manager.
@@ -208,13 +208,7 @@ func (gs *GRPCStreamingSpanSink) Ingest(ssfSpan *ssf.SSFSpan) error {
ssfSpan.Tags[k] = v
}
- // gRPC indicates that it is safe to send and receive on a stream at the
- // same time, but it is not safe to simultaneously send. Guard against that
- // with a mutex.
- gs.streamMut.Lock()
- err := gs.stream.Send(ssfSpan)
- gs.streamMut.Unlock()
-
+ err := gs.send(ssfSpan)
if err != nil {
atomic.AddUint32(&gs.dropCount, 1)
@@ -255,6 +249,23 @@ func (gs *GRPCStreamingSpanSink) Ingest(ssfSpan *ssf.SSFSpan) error {
return err
}
+// send dispatches a span to the backend over the stream.
+//
+// gRPC rules state that it is safe to send and receive on a stream at the
+// same time, but not safe to simultaneously send. Thus, we serialize
+// sends with a mutex.
+//
+// The same mutex is also used by the gRPC channel state machine manager,
+// maintainStream(), when it recreates the channel. This guarantees that
+// sends will never be attempted on a stream that is in the process of
+// being recreated.
+func (gs *GRPCStreamingSpanSink) send(ssfSpan *ssf.SSFSpan) error {
+ gs.streamMut.Lock()
+ err := gs.stream.Send(ssfSpan)
+ gs.streamMut.Unlock()
+ return err
+}
+
// Flush reports total counts of the number of sent and dropped spans over its
// lifetime.
//
