TestMu Conference is Back! Future of Testing + AI LambdaTest G2 Announcement

How to use Close method of event Package

Best Testkube code snippet using event.Close

reader.go

Source:reader.go

...11	"github.com/cilium/ebpf/internal"12	"github.com/cilium/ebpf/internal/unix"13)14var (15	ErrClosed = errors.New("perf reader was closed")16	errEOR    = errors.New("end of ring")17)18// perfEventHeader must match 'struct perf_event_header` in <linux/perf_event.h>.19type perfEventHeader struct {20	Type uint3221	Misc uint1622	Size uint1623}24func addToEpoll(epollfd, fd int, cpu int) error {25	if int64(cpu) > math.MaxInt32 {26		return fmt.Errorf("unsupported CPU number: %d", cpu)27	}28	// The representation of EpollEvent isn't entirely accurate.29	// Pad is fully useable, not just padding. Hence we stuff the30	// CPU in there, which allows us to use a slice to access31	// the correct perf ring.32	event := unix.EpollEvent{33		Events: unix.EPOLLIN,34		Fd:     int32(fd),35		Pad:    int32(cpu),36	}37	if err := unix.EpollCtl(epollfd, unix.EPOLL_CTL_ADD, fd, &event); err != nil {38		return fmt.Errorf("can't add fd to epoll: %v", err)39	}40	return nil41}42func cpuForEvent(event *unix.EpollEvent) int {43	return int(event.Pad)44}45// Record contains either a sample or a counter of the46// number of lost samples.47type Record struct {48	// The CPU this record was generated on.49	CPU int50	// The data submitted via bpf_perf_event_output.51	// Due to a kernel bug, this can contain between 0 and 7 bytes of trailing52	// garbage from the ring depending on the input sample's length.53	RawSample []byte54	// The number of samples which could not be output, since55	// the ring buffer was full.56	LostSamples uint6457}58// NB: Has to be preceded by a call to ring.loadHead.59func readRecordFromRing(ring *perfEventRing) (Record, error) {60	defer ring.writeTail()61	return readRecord(ring, ring.cpu)62}63func readRecord(rd io.Reader, cpu int) (Record, error) {64	var header perfEventHeader65	err := binary.Read(rd, internal.NativeEndian, &header)66	if err == io.EOF {67		return Record{}, errEOR68	}69	if err != nil {70		return Record{}, fmt.Errorf("can't read event header: %v", err)71	}72	switch header.Type {73	case unix.PERF_RECORD_LOST:74		lost, err := readLostRecords(rd)75		return Record{CPU: cpu, LostSamples: lost}, err76	case unix.PERF_RECORD_SAMPLE:77		sample, err := readRawSample(rd)78		return Record{CPU: cpu, RawSample: sample}, err79	default:80		return Record{}, &unknownEventError{header.Type}81	}82}83func readLostRecords(rd io.Reader) (uint64, error) {84	// lostHeader must match 'struct perf_event_lost in kernel sources.85	var lostHeader struct {86		ID   uint6487		Lost uint6488	}89	err := binary.Read(rd, internal.NativeEndian, &lostHeader)90	if err != nil {91		return 0, fmt.Errorf("can't read lost records header: %v", err)92	}93	return lostHeader.Lost, nil94}95func readRawSample(rd io.Reader) ([]byte, error) {96	// This must match 'struct perf_event_sample in kernel sources.97	var size uint3298	if err := binary.Read(rd, internal.NativeEndian, &size); err != nil {99		return nil, fmt.Errorf("can't read sample size: %v", err)100	}101	data := make([]byte, int(size))102	if _, err := io.ReadFull(rd, data); err != nil {103		return nil, fmt.Errorf("can't read sample: %v", err)104	}105	return data, nil106}107// Reader allows reading bpf_perf_event_output108// from user space.109type Reader struct {110	// mu protects read/write access to the Reader structure with the111	// exception of 'pauseFds', which is protected by 'pauseMu'.112	// If locking both 'mu' and 'pauseMu', 'mu' must be locked first.113	mu sync.Mutex114	// Closing a PERF_EVENT_ARRAY removes all event fds115	// stored in it, so we keep a reference alive.116	array *ebpf.Map117	rings []*perfEventRing118	epollFd     int119	epollEvents []unix.EpollEvent120	epollRings  []*perfEventRing121	// Eventfds for closing122	closeFd int123	// Ensure we only close once124	closeOnce sync.Once125	// pauseFds are a copy of the fds in 'rings', protected by 'pauseMu'.126	// These allow Pause/Resume to be executed independently of any ongoing127	// Read calls, which would otherwise need to be interrupted.128	pauseMu  sync.Mutex129	pauseFds []int130}131// ReaderOptions control the behaviour of the user132// space reader.133type ReaderOptions struct {134	// The number of written bytes required in any per CPU buffer before135	// Read will process data. Must be smaller than PerCPUBuffer.136	// The default is to start processing as soon as data is available.137	Watermark int138}139// NewReader creates a new reader with default options.140//141// array must be a PerfEventArray. perCPUBuffer gives the size of the142// per CPU buffer in bytes. It is rounded up to the nearest multiple143// of the current page size.144func NewReader(array *ebpf.Map, perCPUBuffer int) (*Reader, error) {145	return NewReaderWithOptions(array, perCPUBuffer, ReaderOptions{})146}147// NewReaderWithOptions creates a new reader with the given options.148func NewReaderWithOptions(array *ebpf.Map, perCPUBuffer int, opts ReaderOptions) (pr *Reader, err error) {149	if perCPUBuffer < 1 {150		return nil, errors.New("perCPUBuffer must be larger than 0")151	}152	epollFd, err := unix.EpollCreate1(unix.EPOLL_CLOEXEC)153	if err != nil {154		return nil, fmt.Errorf("can't create epoll fd: %v", err)155	}156	var (157		fds      = []int{epollFd}158		nCPU     = int(array.MaxEntries())159		rings    = make([]*perfEventRing, 0, nCPU)160		pauseFds = make([]int, 0, nCPU)161	)162	defer func() {163		if err != nil {164			for _, fd := range fds {165				unix.Close(fd)166			}167			for _, ring := range rings {168				if ring != nil {169					ring.Close()170				}171			}172		}173	}()174	// bpf_perf_event_output checks which CPU an event is enabled on,175	// but doesn't allow using a wildcard like -1 to specify "all CPUs".176	// Hence we have to create a ring for each CPU.177	for i := 0; i < nCPU; i++ {178		ring, err := newPerfEventRing(i, perCPUBuffer, opts.Watermark)179		if errors.Is(err, unix.ENODEV) {180			// The requested CPU is currently offline, skip it.181			rings = append(rings, nil)182			pauseFds = append(pauseFds, -1)183			continue184		}185		if err != nil {186			return nil, fmt.Errorf("failed to create perf ring for CPU %d: %v", i, err)187		}188		rings = append(rings, ring)189		pauseFds = append(pauseFds, ring.fd)190		if err := addToEpoll(epollFd, ring.fd, len(rings)-1); err != nil {191			return nil, err192		}193	}194	closeFd, err := unix.Eventfd(0, unix.O_CLOEXEC|unix.O_NONBLOCK)195	if err != nil {196		return nil, err197	}198	fds = append(fds, closeFd)199	if err := addToEpoll(epollFd, closeFd, -1); err != nil {200		return nil, err201	}202	array, err = array.Clone()203	if err != nil {204		return nil, err205	}206	pr = &Reader{207		array:   array,208		rings:   rings,209		epollFd: epollFd,210		// Allocate extra event for closeFd211		epollEvents: make([]unix.EpollEvent, len(rings)+1),212		epollRings:  make([]*perfEventRing, 0, len(rings)),213		closeFd:     closeFd,214		pauseFds:    pauseFds,215	}216	if err = pr.Resume(); err != nil {217		return nil, err218	}219	runtime.SetFinalizer(pr, (*Reader).Close)220	return pr, nil221}222// Close frees resources used by the reader.223//224// It interrupts calls to Read.225//226// Calls to perf_event_output from eBPF programs will return227// ENOENT after calling this method.228func (pr *Reader) Close() error {229	var err error230	pr.closeOnce.Do(func() {231		runtime.SetFinalizer(pr, nil)232		// Interrupt Read() via the event fd.233		var value [8]byte234		internal.NativeEndian.PutUint64(value[:], 1)235		_, err = unix.Write(pr.closeFd, value[:])236		if err != nil {237			err = fmt.Errorf("can't write event fd: %v", err)238			return239		}240		// Acquire the locks. This ensures that Read, Pause and Resume241		// aren't running.242		pr.mu.Lock()243		defer pr.mu.Unlock()244		pr.pauseMu.Lock()245		defer pr.pauseMu.Unlock()246		unix.Close(pr.epollFd)247		unix.Close(pr.closeFd)248		pr.epollFd, pr.closeFd = -1, -1249		// Close rings250		for _, ring := range pr.rings {251			if ring != nil {252				ring.Close()253			}254		}255		pr.rings = nil256		pr.pauseFds = nil257		pr.array.Close()258	})259	if err != nil {260		return fmt.Errorf("close PerfReader: %w", err)261	}262	return nil263}264// Read the next record from the perf ring buffer.265//266// The function blocks until there are at least Watermark bytes in one267// of the per CPU buffers. Records from buffers below the Watermark268// are not returned.269//270// Records can contain between 0 and 7 bytes of trailing garbage from the ring271// depending on the input sample's length.272//273// Calling Close interrupts the function.274func (pr *Reader) Read() (Record, error) {275	pr.mu.Lock()276	defer pr.mu.Unlock()277	if pr.epollFd == -1 {278		return Record{}, fmt.Errorf("%w", ErrClosed)279	}280	for {281		if len(pr.epollRings) == 0 {282			nEvents, err := unix.EpollWait(pr.epollFd, pr.epollEvents, -1)283			if temp, ok := err.(temporaryError); ok && temp.Temporary() {284				// Retry the syscall if we we're interrupted, see https://github.com/golang/go/issues/20400285				continue286			}287			if err != nil {288				return Record{}, err289			}290			for _, event := range pr.epollEvents[:nEvents] {291				if int(event.Fd) == pr.closeFd {292					return Record{}, fmt.Errorf("%w", ErrClosed)293				}294				ring := pr.rings[cpuForEvent(&event)]295				pr.epollRings = append(pr.epollRings, ring)296				// Read the current head pointer now, not every time297				// we read a record. This prevents a single fast producer298				// from keeping the reader busy.299				ring.loadHead()300			}301		}302		// Start at the last available event. The order in which we303		// process them doesn't matter, and starting at the back allows304		// resizing epollRings to keep track of processed rings.305		record, err := readRecordFromRing(pr.epollRings[len(pr.epollRings)-1])306		if err == errEOR {307			// We've emptied the current ring buffer, process308			// the next one.309			pr.epollRings = pr.epollRings[:len(pr.epollRings)-1]310			continue311		}312		return record, err313	}314}315// Pause stops all notifications from this Reader.316//317// While the Reader is paused, any attempts to write to the event buffer from318// BPF programs will return -ENOENT.319//320// Subsequent calls to Read will block until a call to Resume.321func (pr *Reader) Pause() error {322	pr.pauseMu.Lock()323	defer pr.pauseMu.Unlock()324	if pr.pauseFds == nil {325		return fmt.Errorf("%w", ErrClosed)326	}327	for i := range pr.pauseFds {328		if err := pr.array.Delete(uint32(i)); err != nil && !errors.Is(err, ebpf.ErrKeyNotExist) {329			return fmt.Errorf("could't delete event fd for CPU %d: %w", i, err)330		}331	}332	return nil333}334// Resume allows this perf reader to emit notifications.335//336// Subsequent calls to Read will block until the next event notification.337func (pr *Reader) Resume() error {338	pr.pauseMu.Lock()339	defer pr.pauseMu.Unlock()340	if pr.pauseFds == nil {341		return fmt.Errorf("%w", ErrClosed)342	}343	for i, fd := range pr.pauseFds {344		if fd == -1 {345			continue346		}347		if err := pr.array.Put(uint32(i), uint32(fd)); err != nil {348			return fmt.Errorf("couldn't put event fd %d for CPU %d: %w", fd, i, err)349		}350	}351	return nil352}353type temporaryError interface {354	Temporary() bool355}356// IsClosed returns true if the error occurred because357// a Reader was closed.358//359// Deprecated: use errors.Is(err, ErrClosed) instead.360func IsClosed(err error) bool {361	return errors.Is(err, ErrClosed)362}363type unknownEventError struct {364	eventType uint32365}366func (uev *unknownEventError) Error() string {367	return fmt.Sprintf("unknown event type: %d", uev.eventType)368}369// IsUnknownEvent returns true if the error occurred370// because an unknown event was submitted to the perf event ring.371func IsUnknownEvent(err error) bool {372	var uee *unknownEventError373	return errors.As(err, &uee)374}...

event.go

Source:event.go

...7	"time"8)9var (10	ErrEvtWaitTimeout = errors.New("event wait time out")11	ErrEvtClosed      = errors.New("event closed")12)13type Event struct {14	C         chan byte15	chanClose chan byte16}17func NewEvent() *Event {18	return &Event{19		C:         make(chan byte, 1),20		chanClose: make(chan byte),21	}22}23// Send event.24func (e *Event) Send() error {25	select {26	case <-e.chanClose:27		return ErrEvtClosed28	default:29		select {30		case e.C <- 1:31		default:32		}33	}34	return nil35	// if e.bClosed {36	// 	return ErrEvtClosed37	// }38	// if len(e.C) == 0 {39	// 	e.C <- 140	// }41	// return nil42}43// Wait event.44func (e *Event) Wait() error {45	select {46	case <-e.chanClose:47		return ErrEvtClosed48	case <-e.C:49	}50	return nil51	// _, ok := <-e.C52	// if ok {53	// 	return nil54	// }55	// return ErrEvtClosed56}57// Close the event.58func (e *Event) Close() {59	select {60	case <-e.chanClose:61		return62	default:63		close(e.chanClose)64	}65	// if !e.bClosed {66	// 	e.bClosed = true67	// 	close(e.C)68	// }69}70// Wait event until timeout.71// @param timeoutMSec, timeout after millisecond. 0 mean check at once, only can use in one waiter scene.72// @return error, ErrEvtWaitTimeout mean timeout.73func (e *Event) WaitUntilTimeout(timeoutMSec uint32) error {74	if timeoutMSec == 0 {75		select {76		case <-e.C:77			return nil78		default:79			return ErrEvtWaitTimeout80		}81	}82	var err error = nil83	t := time.NewTimer(time.Millisecond * time.Duration(timeoutMSec))84	select {85	case <-e.chanClose:86		t.Stop()87		err = ErrEvtClosed88	case <-e.C:89		t.Stop()90	case <-t.C:91		err = ErrEvtWaitTimeout92	}93	return err94}...

Using AI Code Generation

1import (2func main() {3	err := sdl.Init(sdl.INIT_EVERYTHING)4	if err != nil {5		fmt.Println(err)6	}7	defer sdl.Quit()8	window, err := sdl.CreateWindow("Tutorial 2", sdl.WINDOWPOS_UNDEFINED, sdl.WINDOWPOS_UNDEFINED, 800, 600, sdl.WINDOW_SHOWN)9	if err != nil {10		fmt.Println(err)11	}12	defer window.Destroy()13	surface, err := window.GetSurface()14	if err != nil {15		fmt.Println(err)16	}17	rect := sdl.Rect{0, 0, 200, 200}18	surface.FillRect(&rect, 0xffff0000)19	window.UpdateSurface()20	for running {21		for event := sdl.PollEvent(); event != nil; event = sdl.PollEvent() {22			switch t := event.(type) {23				fmt.Println("Quit")24			}25		}26	}27}

Using AI Code Generation

1import (2func main() {3gtk.Init(nil)4win, _ := gtk.WindowNew(gtk.WINDOW_TOPLEVEL)5win.SetTitle("Hello, gotk3!")6win.Connect("destroy", func() {7gtk.MainQuit()8})9win.ShowAll()10gtk.Main()11}12import (13func main() {14gtk.Init(nil)15win, _ := gtk.WindowNew(gtk.WINDOW_TOPLEVEL)16win.SetTitle("Hello, gotk3!")17win.Connect("destroy", func() {18gtk.MainQuit()19})20win.ShowAll()21gtk.Main()22}23import (24func main() {25gtk.Init(nil)26win, _ := gtk.WindowNew(gtk.WINDOW_TOPLEVEL)27win.SetTitle("Hello, gotk3!")28win.Connect("destroy", func() {29gtk.MainQuit()30})31win.ShowAll()32gtk.Main()33}34import (35func main() {36gtk.Init(nil)37win, _ := gtk.WindowNew(gtk.WINDOW_TOPLEVEL)38win.SetTitle("Hello, gotk3!")39win.Connect("destroy", func() {40gtk.MainQuit()41})42win.ShowAll()43gtk.Main()44}45import (46func main() {47gtk.Init(nil)48win, _ := gtk.WindowNew(gtk.WINDOW_TOPLEVEL)49win.SetTitle("Hello, gotk3!")50win.Connect("destroy", func() {51gtk.MainQuit()52})53win.ShowAll()54gtk.Main()55}56import (57func main() {58gtk.Init(nil)59win, _ := gtk.WindowNew(gtk.WINDOW_TOP

Using AI Code Generation

1import (2func loop() {3    giu.SingleWindow("Hello").Layout(4        giu.Button("Click me").OnClick(func() {5            fmt.Println("Clicked")6        }),7}8func main() {9    w := giu.NewMasterWindow("Hello", 400, 200, 0)10    w.Main(loop)11}12import (13func loop() {14    giu.SingleWindow("Hello").Layout(15        giu.Button("Click me").OnClick(func() {16            fmt.Println("Clicked")17        }),18}19func main() {20    w := giu.NewMasterWindow("Hello", 400, 200, 0)21    w.Main(loop)22}23import (24func loop() {25    giu.SingleWindow("Hello").Layout(26        giu.Button("Click me").OnClick(func() {27            fmt.Println("Clicked")28        }),29}30func main() {31    w := giu.NewMasterWindow("Hello", 400, 200, 0)32    w.Main(loop)33}34import (35func loop() {36    giu.SingleWindow("Hello").Layout(37        giu.Button("Click me").OnClick(func() {38            fmt.Println("Clicked")39        }),40}41func main() {42    w := giu.NewMasterWindow("Hello", 400, 200, 0)43    w.Main(loop)44}45import (

Using AI Code Generation

1import (2type Event struct {3}4func NewEvent(name string) *Event {5    return &Event{name: name, time: time.Now()}6}7func (e *Event) Close() {8    fmt.Printf("Event %s took %s9", e.name, time.Since(e.time))10}11func main() {12    e := NewEvent("my event")13    defer e.Close()14    time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)15}

Using AI Code Generation

1import (2type Event struct {3}4func (e *Event) Close() {5	fmt.Println("closing")6}7func main() {8	size := int64(0)9	event := Event{path, name, size, hash, time, eventType}10	defer event.Close()11	fmt.Println("doing something")12}13import (14type Event struct {15}16func (e *Event) Close() {17	fmt.Println("closing")18}19func main() {

Using AI Code Generation

1import (2func main() {3	robotgo.EventEnd()4	fmt.Println("EventEnd")5}6import (7func main() {8	robotgo.EventEnd()9	robotgo.EventEnd()10	fmt.Println("EventEnd")11}12import (13func main() {14	robotgo.EventEnd()15	robotgo.EventEnd()16	fmt.Println("EventEnd")17}18import (19func main() {20	robotgo.EventEnd()21	robotgo.EventEnd()22	fmt.Println("EventEnd")23}24import (25func main() {26	robotgo.EventEnd()27	robotgo.EventEnd()28	fmt.Println("EventEnd")29}30import (31func main() {32	robotgo.EventEnd()33	robotgo.EventEnd()34	fmt.Println("EventEnd")35}36import (37func main() {38	robotgo.EventEnd()39	robotgo.EventEnd()40	fmt.Println("EventEnd")41}42import (43func main() {44	robotgo.EventEnd()45	robotgo.EventEnd()

Using AI Code Generation

1func main() {2    e := event.New()3    e.On("greet", func(s string) {4        fmt.Println("Hello", s)5    })6    e.Fire("greet", "John")7    e.Close()8}

Using AI Code Generation

1import (2func main() {3	event := NewEvent()4	ch := make(chan int)5	event.Register(ch)6	event.Trigger(1)7	time.Sleep(3 * time.Second)8	event.Close()9	time.Sleep(1 * time.Second)10	event.Trigger(2)11	time.Sleep(1 * time.Second)12}

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.