rueian

import ray
import time
from dataclasses import dataclass
from heapq import heappush, heappop

@dataclass
class Meta:
    max: int
    rtt: float = 10
    size: int = 1

rueian

# syntax=docker/dockerfile:experimental

FROM nvidia/cuda:11.4.0-cudnn8-devel-ubuntu20.04

ENV PATH /usr/local/bin:$PATH

# http://bugs.python.org/issue19846
# > At the moment, setting "LANG=C" on a Linux system *fundamentally breaks Python 3*, and that's not OK.
ENV LANG C.UTF-8
ENV DEBIAN_FRONTEND noninteractive

rueian

import os
import sys
import ssl
import json
import socket
import asyncio

import util
async def proxy(node_id):
    dir = os.path.join(os.getcwd(), '.ray-delegate')

rueian

import asyncio
import os
import ssl
import json
import secrets

import util
from ray.util import get_node_ip_address

async def handle_echo(reader, writer):

rueian

package main

import (
	"context"
	"fmt"
	"reflect"
	"time"

	"github.com/rueian/rueidis"
	"github.com/rueian/rueidis/om"

rueian

▶ go test -bench=BenchmarkCond ./ring_test.go
goos: darwin
goarch: arm64
BenchmarkCond/Chan-size-64-parallelism-01-10         	 5995210	       203.6 ns/op
BenchmarkCond/Ring-size-64-parallelism-01-10         	 7178178	       162.0 ns/op
BenchmarkCond/Chan-size-64-parallelism-08-10         	 3785736	       314.9 ns/op
BenchmarkCond/Ring-size-64-parallelism-08-10         	 5421237	       204.3 ns/op
BenchmarkCond/Chan-size-64-parallelism-64-10         	 3847623	       309.3 ns/op
BenchmarkCond/Ring-size-64-parallelism-64-10         	 5169406	       235.1 ns/op
PASS

rueian

var waits = int32(0)
var sleep = int32(0)
var cond = sync.Cond{L: (*emptyLock)(nil)}

func makeRequest(req request, queue *ring) (resp response) {
	waits := atomic.AddInt32(&waits, 1)
	ch := queue.EnqueueRequest(queue)
	if waits == 1 {
		for atomic.LoadInt32(&sleep) != 0 {
			cond.Broadcast()

rueian

package cond

import (
	"runtime"
	"sync"
	"sync/atomic"
	"testing"
)

type Queue interface {

rueian

▶ go test -bench=BenchmarkCond ./ring_test.go
goos: darwin
goarch: arm64
BenchmarkCond/Chan-size-64-parallelism-01-10         	11056395	       110.1 ns/op
BenchmarkCond/Ring-size-64-parallelism-01-10         	11023176	       110.8 ns/op
BenchmarkCond/Chan-size-64-parallelism-08-10         	 4022212	       305.0 ns/op
BenchmarkCond/Ring-size-64-parallelism-08-10         	 5524983	       187.2 ns/op
BenchmarkCond/Chan-size-64-parallelism-64-10         	 3942190	       304.1 ns/op
BenchmarkCond/Ring-size-64-parallelism-64-10         	 6217990	       207.5 ns/op
PASS

rueian

func (r *ring) EnqueueRequest(req request) chan response {
	s := &r.slots[atomic.AddUint64(&r.write, 1)&mask]
	s.cond.L.Lock()
	for s.mark != 0 {
		s.cond.Wait()
	}
	s.req = req
	s.mark = 1
	s.cond.L.Unlock()
	return s.ch