Quantile fiber test

metrics/quantile.lua

@@ -134,9 +134,11 @@ function stream:merge(samples, len)
 	local i = 1
     local r = 0
     for z = 1, len do
-        if i % 1000 == 0 then
-            fiber.yield()
+		if i % 1000 == 0 then
+			fiber.yield()
 		end
+		assert(z - 1 < ffi.sizeof(samples)/ffi.sizeof('double'), z - 1 .. ' out of bound ' .. tostring(ffi.sizeof(samples)/ffi.sizeof('double')))
+
 		local sample = samples[z-1]
 		for j = i, s.l_len do
             local c = s.l[j]
@@ -259,6 +261,8 @@ end
 
 -- Insert inserts v into the stream.
 function quantile.Insert(stream_obj, v)
+	assert(stream_obj.b_len < ffi.sizeof(stream_obj.b)/ffi.sizeof('double'), 'Array index out of bound ' .. stream_obj.b_len .. '  ' .. ffi.sizeof(stream_obj.b)/ffi.sizeof('double'))
+
 	stream_obj.b[stream_obj.b_len] = v
 	stream_obj.b_len = stream_obj.b_len + 1
 	stream_obj.compress_cnt = stream_obj.compress_cnt + 1

test/quantile_fiber_test.lua

@@ -0,0 +1,26 @@
+local quantile = require('metrics.quantile')
+local fiber = require('fiber')
+
+local t = require('luatest')
+local g = t.group('quantile-fiber')
+
+local q = quantile.NewTargeted({[0.5]=0.01, [0.9]=0.01, [0.99]=0.01})
+
+g.test_fiber = function()
+    for _=1,1e6 do quantile.Insert(q, math.random(1)) end
+
+    local fs = {}
+    for _=1,200 do
+        local f = fiber.new(function()
+            for _=1,1e3 do quantile.Insert(q, math.random(1000)) end
+        end)
+        f:set_joinable(true)
+        table.insert(fs, f)
+    end
+
+    for _=1,1e6 do quantile.Insert(q, math.random(1000)) end
+
+    for _, f in ipairs(fs) do
+        f:join()
+    end
+end