package profile import ( "context" "demo/deliver" "fmt" "os" "time" ) func shmrecver(ctx context.Context, c deliver.Deliver, index int, ch chan<- bool) { var msg []byte var err error var t int64 var elapse int64 count := 0 for { select { case <-ctx.Done(): return default: msg, err = c.Recv() if err != nil { c.Close() url := "hello" i, err := deliver.NewClientWithError(deliver.Shm, url) for { if err == nil { break } time.Sleep(1 * time.Second) i, err = deliver.NewClientWithError(deliver.Shm, url) fmt.Println("client create failed : ", err) } c = i fmt.Println("recv error : ", err) continue } if ch != nil { ch <- true } if t == 0 { t = time.Now().UnixNano() } elapse = time.Now().UnixNano() - t count++ if elapse > 1e9 { fmt.Printf("NODE-%d: RECEIVED \"%d\" data %s len %d, use \"%d\" ns\n", index, count, string(msg), len(msg), elapse) elapse = 0 count = 0 t = 0 } } // time.Sleep(10 * time.Millisecond) } } func Shmrecv(ctx context.Context, server bool, ipc string, count int) { blocks := 2 if server { s, err := deliver.NewServerWithError(deliver.Shm, ipc, blocks, dLen) for { if err == nil { break } time.Sleep(1 * time.Second) s, err = deliver.NewServerWithError(deliver.Shm, ipc, blocks, dLen) } go shmrecver(ctx, s, 0, nil) } else { recvers(ctx, ipc, count, nil) return chWaiter := make(chan bool, count) cs := recvers(ctx, ipc, count, chWaiter) go func() { waitCount := 0 for { select { case <-ctx.Done(): return case <-chWaiter: waitCount = 0 default: if waitCount > 200*3 { for _, v := range cs { v.Close() } // cs = recvers(ctx, ipc, count, chWaiter) // fmt.Println("restart recievers") // waitCount = 0 // continue } time.Sleep(time.Millisecond * 5) waitCount++ } } }() } } func recvers(ctx context.Context, url string, count int, ch chan<- bool) []deliver.Deliver { for { file := "/dev/shm/" + url exist, _ := pathExists(file) if exist { break } fmt.Println("wait for shm server start") time.Sleep(time.Second) } var cs []deliver.Deliver for i := 0; i < count; i++ { c, err := deliver.NewClientWithError(deliver.Shm, url) for { if err == nil { break } time.Sleep(1 * time.Second) c, err = deliver.NewClientWithError(deliver.Shm, url) fmt.Println(i, " client create failed : ", err) } cs = append(cs, c) go shmrecver(ctx, c, i, ch) } return cs } func pathExists(path string) (bool, error) { _, err := os.Stat(path) if err == nil { return true, nil } if os.IsNotExist(err) { return false, nil } return false, err }