Changeset 2dd0689 for benchmark/readyQ

Timestamp:

Dec 16, 2020, 2:37:31 PM (3 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

ADT, arm-eh, ast-experimental, enum, forall-pointer-decay, jacob/cs343-translation, master, new-ast-unique-expr, pthread-emulation, qualifiedEnum

Children:

fd84538

Parents:

72b1800

Message:

Fix implementation of locality to properly use spots.

File:

• benchmark/readyQ/locality.go (modified) (6 diffs)

benchmark/readyQ/locality.go

@@ -2 +2 @@
 
 import (
+        "context"
         "flag"
         "fmt"
@@ -8 +9 @@
         "sync/atomic"
         "time"
+        "unsafe"
+        "golang.org/x/sync/semaphore"
         "golang.org/x/text/language"
         "golang.org/x/text/message"
 )
 
-func handshake(stop chan struct {}, c chan [] uint64, data [] uint64, share bool) (bool, [] uint64) {
-        var s [] uint64 = data
-        if !share {
-                s = nil
+type GoCtx struct {
+        s * semaphore.Weighted
+        d * [] uint64
+}
+
+type Spot struct {
+        ptr uintptr
+}
+
+func (this * Spot) put( s * semaphore.Weighted, data [] uint64, share bool) ([] uint64) {
+        ctx := GoCtx{s, &data}
+
+        var raw uintptr
+        for true {
+                raw = this.ptr
+                if raw == uintptr(1) {
+                        return nil
+                }
+                if atomic.CompareAndSwapUintptr(&this.ptr, raw, uintptr(unsafe.Pointer(&ctx))) {
+                        break
+                }
         }
 
-        // send the data
-        select {
-        case <- stop:
-                return true, nil
-        case c <- s:
+        if raw != uintptr(0) {
+                val := (*GoCtx)(unsafe.Pointer(raw))
+                if share {
+                        val.d = &data
+                }
+
+                val.s.Release(1)
         }
 
-        // get the new data chunk
-        select {
-        case <- stop:
-                return true, nil
-        case n := <- c:
-                if share {
-                        return false, n
-                }
-                return false, data
-        }
+        ctx.s.Acquire(context.Background(), 1)
+
+        return *ctx.d
 }
 
-func local(result chan uint64, start chan struct{}, stop chan struct{}, size uint64, cnt uint64, channels []chan [] uint64, chan_cnt uint64, share bool) {
+func (this * Spot) release() {
+        val := (*GoCtx)(unsafe.Pointer(atomic.SwapUintptr(&this.ptr, uintptr(1))))
+        if val == nil {
+                return
+        }
+
+        val.s.Release(1)
+}
+
+
+
+func local(result chan uint64, start chan struct{}, size uint64, cnt uint64, channels [] Spot, share bool) {
+        lrand := rand.New(rand.NewSource(rand.Int63()))
         var data [] uint64
         data = make([]uint64, size)
@@ -43 +70 @@
                 data[i] = 0
         }
+
+        sem := semaphore.NewWeighted(1)
+        sem.Acquire(context.Background(), 1)
+
         count := uint64(0)
         <- start
         for true {
                 for i := uint64(0); i < cnt; i++ {
-                        data[rand.Uint64() % size] += 1
+                        data[lrand.Uint64() % size] += 1
                 }
 
-                i := rand.Uint64() % chan_cnt
-                var closed bool
-                closed, data = handshake(stop, channels[i], data, share)
+                i := lrand.Int() % len(channels)
+                data = channels[i].put(sem, 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")
+                }
         }
 
@@ -64 +97 @@
 
 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")
 
-        bench_init()
+        defer bench_init()()
 
         size  := *work_sizeOpt
@@ -81 +113 @@
 
         barrierStart := make(chan struct{})
-        barrierStop  := make(chan struct{})
         threads_left = int64(nthreads)
         result  := make(chan uint64)
-        channels := make([]chan [] uint64, nthreads - nprocs)
+        channels := make([]Spot, nthreads - nprocs)
         for i := range channels {
-                channels[i] = make(chan [] uint64, 1)
+                channels[i] = Spot{uintptr(0)}
         }
 
         for i := 0; i < nthreads; i++ {
-                go local(result, barrierStart, barrierStop, size, cnt, channels, uint64(nthreads - nprocs), share)
+                go local(result, barrierStart, size, cnt, channels, share)
         }
         fmt.Printf("Starting\n");
 
+        atomic.StoreInt32(&stop, 0)
         start := time.Now()
         close(barrierStart)
@@ -99 +131 @@
         wait(start, true);
 
-        close(barrierStop)
+        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)