Test concurrency
@@ -2,6 +2,9 @@ package com.datadoghq.trace.writer
import spock.lang.Specification
+import java.util.concurrent.Phaser
+import java.util.concurrent.atomic.AtomicInteger
+
class WriterQueueTest extends Specification {
def "instantiate a empty queue throws an exception"() {
@@ -78,4 +81,103 @@ class WriterQueueTest extends Specification {
}
+ def "check concurrency on writes"() {
+ setup:
+
+ def phaser_1 = new Phaser()
+ def phaser_2 = new Phaser()
+ def queue = new WriterQueue<Integer>(capacity)
+ def insertionCount = new AtomicInteger(0)
+
+ phaser_1.register() // global start
+ phaser_2.register() // global stop
+
+ numberThreads.times {
+ phaser_1.register()
+ Thread.start {
+ phaser_2.register()
+ phaser_1.arriveAndAwaitAdvance()
+ numberInsertionsPerThread.times {
+ queue.add(1)
+ insertionCount.getAndIncrement()
+ }
+ phaser_2.arriveAndAwaitAdvance()
+ }
+ }
+
+ when:
+ phaser_1.arriveAndAwaitAdvance() // allow threads to start
+ phaser_2.arriveAndAwaitAdvance() // wait till the job is not finished
+
+ then:
+ queue.size() == capacity
+ insertionCount.get() == numberInsertionsPerThread * numberThreads
+
+ where:
+ capacity = 100
+ numberThreads << [1, 10, 100]
+ numberInsertionsPerThread = 100
+
+ }
+
+
+ def "check concurrency on writes and reads"() {
+ setup:
+ def phaser_1 = new Phaser()
+ def phaser_2 = new Phaser()
+ def queue = new WriterQueue<Integer>(capacity)
+ def insertionCount = new AtomicInteger(0)
+ def droppedCount = new AtomicInteger(0)
+ def getCount = new AtomicInteger(0)
+ def numberElements = new AtomicInteger(0)
+
+ phaser_1.register() // global start
+ phaser_2.register() // global stop
+
+ // writes
+ numberThreadsWrites.times {
+ phaser_1.register()
+ Thread.start {
+ phaser_2.register()
+ phaser_1.arriveAndAwaitAdvance()
+ numberInsertionsPerThread.times {
+ queue.add(1) != null ? droppedCount.getAndIncrement() : null
+ insertionCount.getAndIncrement()
+ }
+ phaser_2.arriveAndAwaitAdvance()
+ }
+ }
+
+ // reads
+ numberThreadsReads.times {
+ phaser_1.register()
+ Thread.start {
+ phaser_2.register()
+ phaser_1.arriveAndAwaitAdvance()
+ numberGetsPerThread.times {
+ numberElements.getAndAdd(queue.getAll().size())
+ getCount.getAndIncrement()
+ }
+ phaser_2.arriveAndAwaitAdvance()
+ }
+ }
+
+ when:
+ phaser_1.arriveAndAwaitAdvance() // allow threads to start
+ phaser_2.arriveAndAwaitAdvance() // wait till the job is not finished
+
+ then:
+ insertionCount.get() == numberInsertionsPerThread * numberThreadsWrites
+ getCount.get() == numberGetsPerThread * numberThreadsReads
+ insertionCount.get() == numberElements + queue.size() + droppedCount
+
+ where:
+ capacity = 100
+ numberThreadsWrites << [1, 10, 100]
+ numberThreadsReads << [1, 5, 10]
+ numberInsertionsPerThread = 100
+ numberGetsPerThread = 5
+
+ }
+
}
