Improving threading safety for telemetry #250
Conversation
| @@ -113,8 +121,8 @@ object Telemetry: Subscriber { | |||
| * Called automatically when Telemetry.enable is set to true and when configuration data is received from Segment. | |||
| */ | |||
| fun start() { | |||
| if (!enable || started || sampleRate == 0.0) return | |||
| started = true | |||
| if (!enable || started.get() || sampleRate == 0.0) return | |||
There was a problem hiding this comment.
if we need atomic on started, likely we need it on enable
There was a problem hiding this comment.
Started could maybe get away without as well, but enable will in most cases not ever change and there are several checks where if it happens to pass by one check, it will avoid sending later. But mainly I thought it not worth changing the interface and documentation just yet, since enable is the only exposed API.
| var sampleRate: Double | ||
| get() = _sampleRate | ||
| set(value) { | ||
| synchronized(this) { |
There was a problem hiding this comment.
we should try our best to avoid using synchronized since our API is coroutine-based. people could use our API inside a coroutine which causes a suspend function to block.
| m.value = (m.value / sampleRate).roundToInt() | ||
| sendQueue.add(m) | ||
| } | ||
| val sendQueue: MutableList<RemoteMetric> |
There was a problem hiding this comment.
instead of snapshot the whole queue, snapshot the size of the queue and dequeue just for that size should work and that avoid of using synchronized
There was a problem hiding this comment.
That's what the code was doing originally, can probably just make the queueBytes atomic and leave it at that.
| /** | ||
| * Starts the telemetry if it is enabled and not already started, and the sample rate is greater than 0. | ||
| * Called automatically when Telemetry.enable is set to true and when configuration data is received from Segment. | ||
| */ | ||
| fun start() { | ||
| if (!enable || started.get() || sampleRate == 0.0) return | ||
| if (!enable || started.get() || sampleRate.get() == 0.0) return |
There was a problem hiding this comment.
should enable be atomic too?
| @@ -334,7 +351,7 @@ object Telemetry: Subscriber { | |||
| private suspend fun systemUpdate(system: com.segment.analytics.kotlin.core.System) { | |||
| system.settings?.let { settings -> | |||
| settings.metrics["sampleRate"]?.jsonPrimitive?.double?.let { | |||
| sampleRate = it | |||
| sampleRate.set(it) | |||
There was a problem hiding this comment.
if use sovran, sampleRate does not have to be atomic, since sovran serialize the task in the same thread. same thing applies to enable and started
No description provided.