source: benchmark/readyQ/locality.go@ 94d93510

package main

import (
        "context"
        "flag"
        "fmt"
        "math/rand"
        "os"
        "sync/atomic"
        "time"
        "unsafe"
        "golang.org/x/sync/semaphore"
        "golang.org/x/text/language"
        "golang.org/x/text/message"
)

type GoCtx struct {
        s * semaphore.Weighted
        d unsafe.Pointer
        c context.Context
        id int
}

type Spot struct {
        ptr uintptr
        id int
}

// Main handshake of the code
// Single seat, first thread arriving waits
// Next threads unblocks current one and blocks in its place
// if share == true, exchange data in the process
func (this * Spot) put( ctx * GoCtx, data * [] uint64, share bool) (* [] uint64, bool) {
        new := uintptr(unsafe.Pointer(ctx))
        // old_d := ctx.d
        var raw uintptr
        for true {
                raw = this.ptr
                if raw == uintptr(1) {
                        return nil, true
                }
                if atomic.CompareAndSwapUintptr(&this.ptr, raw, new) {
                        break
                }
        }

        if raw != uintptr(0) {
                var val *GoCtx
                val = (*GoCtx)(unsafe.Pointer(raw))
                if share {
                        // fmt.Printf("[%d] - %d update %d: %p -> %p\n", this.id, ctx.id, val.id, val.d, data)
                        atomic.StorePointer(&val.d, unsafe.Pointer(data))
                }

                // fmt.Printf("[%d] - %d release %d\n", this.id, ctx.id, val.id)
                val.s.Release(1)
        }

        // fmt.Printf("[%d] - %d enter\n", this.id, ctx.id)
        ctx.s.Acquire(ctx.c, 1)
        ret := (* [] uint64)(atomic.LoadPointer(&ctx.d))
        // fmt.Printf("[%d] - %d leave: %p -> %p\n", this.id, ctx.id, ret, old_d)

        return ret, false
}

func (this * Spot) release() {
        val := (*GoCtx)(unsafe.Pointer(atomic.SwapUintptr(&this.ptr, uintptr(1))))
        if val == nil {
                return
        }

        val.s.Release(1)
}

func __xorshift64( state * uint64 ) (uint64) {
        x := *state
        x ^= x << 13
        x ^= x >> 7
        x ^= x << 17
        *state = x
        return x
}

func work(data * [] uint64, size uint64, cnt uint64, state * uint64) {
        for i := uint64(0); i < cnt; i++ {
                (*data)[__xorshift64(state) % size] += 1
        }
}

func local(result chan uint64, start chan struct{}, size uint64, cnt uint64, channels [] Spot, share bool, id int) {
        state := rand.Uint64()
        var my_data [] uint64
        my_data = make([]uint64, size)
        for i := uint64(0); i < size; i++ {
                my_data[i] = 0
        }
        data := &my_data

        sem := semaphore.NewWeighted(1)
        sem.Acquire(context.Background(), 1)
        ctx := GoCtx{sem, unsafe.Pointer(data), context.Background(), id}

        count := uint64(0)
        <- start
        for true {
                work(data, size, cnt, &state)

                i := __xorshift64(&state) % uint64(len(channels))
                var closed bool
                data, closed = channels[i].put(&ctx, data, share)
                count += 1

                if closed { break }
                if  clock_mode && atomic.LoadInt32(&stop) == 1 { break }
                if !clock_mode && count >= stop_count { break }
                if uint64(len(*data)) != size {
                        panic("Data has weird size")
                }
        }

        atomic.AddInt64(&threads_left, -1);
        result <- count
}

func main() {
        work_sizeOpt := flag.Uint64("w", 2    , "Number of words (uint64) per threads")
        countOpt     := flag.Uint64("c", 2    , "Number of words (uint64) to touch")
        shareOpt     := flag.Bool  ("s", false, "Pass the work data to the next thread when blocking")

        defer bench_init()()

        size  := *work_sizeOpt
        cnt   := *countOpt
        share := *shareOpt

        if ! (nthreads > nprocs) {
                fmt.Fprintf(os.Stderr, "Must have more threads than procs\n")
                os.Exit(1)
        }

        barrierStart := make(chan struct{})
        threads_left = int64(nprocs)
        result  := make(chan uint64)
        channels := make([]Spot, nthreads - nprocs)
        for i := range channels {
                channels[i] = Spot{uintptr(0), i}
        }

        for i := 0; i < nthreads; i++ {
                go local(result, barrierStart, size, cnt, channels, share, i)
        }
        fmt.Printf("Starting\n");

        atomic.StoreInt32(&stop, 0)
        start := time.Now()
        close(barrierStart)

        wait(start, true);

        atomic.StoreInt32(&stop, 1)
        end := time.Now()
        delta := end.Sub(start)

        fmt.Printf("\nDone\n")

        for i := range channels {
                channels[i].release()
        }

        global_counter := uint64(0)
        for i := 0; i < nthreads; i++ {
                r := <- result
                global_counter += r
                fmt.Printf("%d\n", r)
        }

        p := message.NewPrinter(language.English)
        p.Printf("Duration (ms)          : %f\n", delta.Seconds());
        p.Printf("Number of processors   : %d\n", nprocs);
        p.Printf("Number of threads      : %d\n", nthreads);
        p.Printf("Work size (64bit words): %d\n", size);
        p.Printf("Total Operations(ops)  : %15d\n", global_counter)
        p.Printf("Ops per second         : %18.2f\n", float64(global_counter) / delta.Seconds())
        p.Printf("ns per ops             : %18.2f\n", float64(delta.Nanoseconds()) / float64(global_counter))
        p.Printf("Ops per threads        : %15d\n", global_counter / uint64(nthreads))
        p.Printf("Ops per procs          : %15d\n", global_counter / uint64(nprocs))
        p.Printf("Ops/sec/procs          : %18.2f\n", (float64(global_counter) / float64(nprocs)) / delta.Seconds())
        p.Printf("ns per ops/procs       : %18.2f\n", float64(delta.Nanoseconds()) / (float64(global_counter) / float64(nprocs)))
}