Quickly slows down about md5-simd HOT 10 CLOSED

@vitalif I don't think your benchmark shows anything unexpected.

You start a bunch (200) of goroutines that each wait randomly between 0 and 1.5 seconds before they start writing.

I would kind of expect with this setup that the first goroutines that are activated will complete much faster before all 200 are active and competing for compute.

A single "server" can only perform 16 concurrent hash operations, so if more than that is scheduled for a cycle they will have to wait.

You can of course use more servers which will use more cores, but that requires you to manage them and at some point you will run out of memory bandwidth and scaling will stop.

from md5-simd.

Yeah, of course, in the first seconds it slows down from ~0.2 sec to ~1 sec, it's OK, it's just caused by random delays. But in my case then it continues to slow down and reaches 2 sec, 3 sec, 4 sec and so on. Which isn't caused by delays.

Here's the second implementation with a pool:

package main

import (
    "fmt"
    "time"
    "hash"
    "math/rand"
    "github.com/minio/md5-simd"
    "sync"
)

func main() {
    md5Server := md5simd.NewServer()
    p := sync.Pool{
        New: func() interface{} {
            return md5Server.NewHash()
        },
    }
    n := 200
    sem := make(chan int)
    for i := 0; i < n; i++ {
        go func() {
            buf := make([]byte, 1048569)
            for i := 0; i < len(buf); i++ {
                buf[i] = byte(i * 53)
            }
            for i := 0; i < 1000; i++ {
                time.Sleep(time.Duration(1500 * rand.Float32()) * time.Millisecond)
                start := time.Now()
                c := p.Get().(hash.Hash)
                for j := 0; j < 15; j++ {
                    c.Write(buf)
                }
                c.Sum(nil)
                c.Reset()
                p.Put(c)
                dur := time.Now().Sub(start)
                fmt.Printf("%.02f s\n", dur.Seconds())
            }
            sem <- 1
        }()
    }
    for i := 0; i < n; i++ {
        <- sem
    }
}

This one doesn't slow down for me - i.e. it also slows down in the beginning from ~0.2sec to ~1sec and then oscillates between 0.8sec-1.2sec, but does not slow down anymore.

Interestingly my colleague says that on his PC the result is the opposite: reset()-based implementation slows down and close()-based doesn't :D

from md5-simd.

PS About the concurrent hashes - I understand the principle and in fact I think it would be good to implement an internal "2D pool" in the library itself so it can use all cores instead of just 1 :-)

from md5-simd.

Yes, it stabilizes around 0.5s->0.7s for me, so I had to assume you meant the initial fast completions.

You cannot leak hashers, you must call Close to de-register it. If you stuff them in a sync.Pool you are leaking them.

To reuse them add a 16 sized chan hash.Hash. Create 16 clients and put them on the channel.

I can't see anything actionable here.

from md5-simd.

I don't leak them, am I doing anything wrong in the first example?
It doesn't stabilize for me.

from md5-simd.

It reaches 2 then 3 then 4 sec and so on

from md5-simd.

The example with sync.Pool doesn't slow down for me, so my question is why does the first example slow down?))

from md5-simd.

First example doesn't slow down for me. Here is about 5 minutes: out.txt.gz

Is your system overheating?

from md5-simd.

No idea... doesn't seem so. Problem is that it reproduces in pre-production environment too (but with a pool) :) but maybe that's also because of that leak via the pool.

from md5-simd.

Made this which shows more fine grained throughput:

func TestSomething(t *testing.T) {
	var total int64

	md5Server := NewServer()
	n := 2000
	go func() {
		t := time.NewTicker(1 * time.Second)
		lastTime := time.Now()
		for range t.C {
			elapsed := time.Since(lastTime)
			processed := atomic.SwapInt64(&total, 0)
			lastTime = time.Now()
			fmt.Printf("%0.2fGB/s\n", float64(processed)/elapsed.Seconds()/float64(1<<30))
		}
	}()

	var wg sync.WaitGroup
	wg.Add(n)
	for i := 0; i < n; i++ {
		go func() {
			defer wg.Done()
			buf := make([]byte, 1048569)
			for i := 0; i < len(buf); i++ {
				buf[i] = byte(i * 53)
			}
			for i := 0; i < 1000; i++ {
				time.Sleep(time.Duration(1500*rand.Float32()) * time.Millisecond)
				c := md5Server.NewHash()
				for j := 0; j < 15; j++ {
					c.Write(buf)
					atomic.AddInt64(&total, int64(len(buf)))
				}
				c.Sum(nil)
				c.Close()
			}
		}()
	}
	wg.Wait()
}

Again; not seeing anything out of the ordinary.

You are welcome to try to rewrite the synchronization. You can also use BenchmarkAvx512 for less competitive tests to make sure those don't regress.

from md5-simd.