How to use newWorkQueue method of main Package

Best Syzkaller code snippet using main.newWorkQueue

work_queue_test.go

Source:work_queue_test.go

1package main2import (3	"container/list"4	"runtime"5	"testing"6	"time"7)8type fatalfer interface {9	Fatalf(string, ...interface{})10}11func makeTestWorkList(ary []int) *list.List {12	l := list.New()13	for _, n := range ary {14		l.PushBack(n)15	}16	return l17}18func expectChannelEmpty(t fatalfer, c <-chan interface{}) {19	select {20	case item, ok := <-c:21		if ok {22			t.Fatalf("Received value (%+v) from channel that we expected to be empty", item)23		}24	default:25	}26}27func expectChannelNotEmpty(t fatalfer, c <-chan interface{}) interface{} {28	select {29	case item, ok := <-c:30		if !ok {31			t.Fatalf("expected data on a closed channel")32		}33		return item34	case <-time.After(time.Second):35		t.Fatalf("expected data on an empty channel")36		return nil37	}38}39func expectChannelClosedWithin(t fatalfer, timeout time.Duration, c <-chan interface{}) {40	select {41	case received, ok := <-c:42		if ok {43			t.Fatalf("Expected channel to be closed, but received %+v instead", received)44		}45	case <-time.After(timeout):46		t.Fatalf("Expected channel to be closed, but it is still open after %v", timeout)47	}48}49func doWorkItems(t fatalfer, q *WorkQueue, expected []int) {50	for i := range expected {51		actual, ok := <-q.NextItem52		if !ok {53			t.Fatalf("Expected %+v but channel was closed after receiving %+v as expected.", expected, expected[:i])54		}55		q.DoneItem <- struct{}{}56		if actual.(int) != expected[i] {57			t.Fatalf("Expected %+v but received %+v after receiving %+v as expected.", expected[i], actual, expected[:i])58		}59	}60}61func expectEqualWithin(t fatalfer, timeout time.Duration, expect interface{}, f func() interface{}) {62	ok := make(chan struct{})63	giveup := false64	go func() {65		for f() != expect && !giveup {66			time.Sleep(time.Millisecond)67		}68		close(ok)69	}()70	select {71	case <-ok:72	case <-time.After(timeout):73		giveup = true74		_, file, line, _ := runtime.Caller(1)75		t.Fatalf("Still getting %+v, timed out waiting for %+v\n%s:%d", f(), expect, file, line)76	}77}78func expectQueued(t fatalfer, b *WorkQueue, expectQueued int) {79	if l := b.Status().Queued; l != expectQueued {80		t.Fatalf("Got Queued==%d, expected %d", l, expectQueued)81	}82}83func TestWorkQueueDoneness(t *testing.T) {84	b := NewWorkQueue()85	defer b.Close()86	b.ReplaceQueue(makeTestWorkList([]int{1, 2, 3}))87	expectQueued(t, b, 3)88	gate := make(chan struct{})89	go func() {90		<-gate91		for range b.NextItem {92			<-gate93			time.Sleep(time.Millisecond)94			b.DoneItem <- struct{}{}95		}96	}()97	expectEqualWithin(t, time.Second, 0, func() interface{} { return b.Status().InProgress })98	b.ReplaceQueue(makeTestWorkList([]int{4, 5, 6}))99	for i := 1; i <= 3; i++ {100		gate <- struct{}{}101		expectEqualWithin(t, time.Second, 3-i, func() interface{} { return b.Status().Queued })102		expectEqualWithin(t, time.Second, 1, func() interface{} { return b.Status().InProgress })103	}104	close(gate)105	expectEqualWithin(t, time.Second, 0, func() interface{} { return b.Status().InProgress })106	expectChannelEmpty(t, b.NextItem)107}108// Create a WorkQueue, generate a list for it, and instantiate a worker.109func TestWorkQueueReadWrite(t *testing.T) {110	var input = []int{1, 1, 2, 3, 5, 8, 13, 21, 34}111	b := NewWorkQueue()112	expectQueued(t, b, 0)113	b.ReplaceQueue(makeTestWorkList(input))114	expectQueued(t, b, len(input))115	doWorkItems(t, b, input)116	expectChannelEmpty(t, b.NextItem)117	b.Close()118}119// Start a worker before the list has any input.120func TestWorkQueueEarlyRead(t *testing.T) {121	var input = []int{1, 1, 2, 3, 5, 8, 13, 21, 34}122	b := NewWorkQueue()123	defer b.Close()124	// First, demonstrate that nothing is available on the NextItem125	// channel.126	expectChannelEmpty(t, b.NextItem)127	// Start a reader in a goroutine. The reader will block until the128	// block work list has been initialized.129	//130	done := make(chan int)131	go func() {132		doWorkItems(t, b, input)133		done <- 1134	}()135	// Feed the blocklist a new worklist, and wait for the worker to136	// finish.137	b.ReplaceQueue(makeTestWorkList(input))138	<-done139	expectQueued(t, b, 0)140}141// After Close(), NextItem closes, work finishes, then stats return zero.142func TestWorkQueueClose(t *testing.T) {143	b := NewWorkQueue()144	input := []int{1, 2, 3, 4, 5, 6, 7, 8}145	mark := make(chan struct{})146	go func() {147		<-b.NextItem148		mark <- struct{}{}149		<-mark150		b.DoneItem <- struct{}{}151	}()152	b.ReplaceQueue(makeTestWorkList(input))153	// Wait for worker to take item 1154	<-mark155	b.Close()156	expectEqualWithin(t, time.Second, 1, func() interface{} { return b.Status().InProgress })157	// Tell worker to report done158	mark <- struct{}{}159	expectEqualWithin(t, time.Second, 0, func() interface{} { return b.Status().InProgress })160	expectChannelClosedWithin(t, time.Second, b.NextItem)161}162// Show that a reader may block when the manager's list is exhausted,163// and that the reader resumes automatically when new data is164// available.165func TestWorkQueueReaderBlocks(t *testing.T) {166	var (167		inputBeforeBlock = []int{1, 2, 3, 4, 5}168		inputAfterBlock  = []int{6, 7, 8, 9, 10}169	)170	b := NewWorkQueue()171	defer b.Close()172	sendmore := make(chan int)173	done := make(chan int)174	go func() {175		doWorkItems(t, b, inputBeforeBlock)176		// Confirm that the channel is empty, so a subsequent read177		// on it will block.178		expectChannelEmpty(t, b.NextItem)179		// Signal that we're ready for more input.180		sendmore <- 1181		doWorkItems(t, b, inputAfterBlock)182		done <- 1183	}()184	// Write a slice of the first five elements and wait for the185	// reader to signal that it's ready for us to send more input.186	b.ReplaceQueue(makeTestWorkList(inputBeforeBlock))187	<-sendmore188	b.ReplaceQueue(makeTestWorkList(inputAfterBlock))189	// Wait for the reader to complete.190	<-done191}192// Replace one active work list with another.193func TestWorkQueueReplaceQueue(t *testing.T) {194	var firstInput = []int{1, 1, 2, 3, 5, 8, 13, 21, 34}195	var replaceInput = []int{1, 4, 9, 16, 25, 36, 49, 64, 81}196	b := NewWorkQueue()197	b.ReplaceQueue(makeTestWorkList(firstInput))198	// Read just the first five elements from the work list.199	// Confirm that the channel is not empty.200	doWorkItems(t, b, firstInput[0:5])201	expectChannelNotEmpty(t, b.NextItem)202	// Replace the work list and read five more elements.203	// The old list should have been discarded and all new204	// elements come from the new list.205	b.ReplaceQueue(makeTestWorkList(replaceInput))206	doWorkItems(t, b, replaceInput[0:5])207	b.Close()208}...

queue_node.go

Source:queue_node.go

...73type WorkQueue struct {74	tasks Queue75	wg *sync.WaitGroup76}77func newWorkQueue(s *sync.WaitGroup) *WorkQueue {78	return &WorkQueue{wg: s}79}80func (w *WorkQueue) addTask(c *ServerCall) {81	w.tasks.enqueue(c)82}83func (w *WorkQueue) run(ctx context.Context, proxy *httputil.ReverseProxy) {84	for {85		time.Sleep(time.Second / time.Duration(4))86		if !w.tasks.empty(){87			serverCall, err := w.tasks.dequeue(ctx)88			if err == nil {89				proxy.ServeHTTP(90					serverCall.response,91					serverCall.request,92				)93				serverCall.wg.Done()94			}95		}96	}97}98func newDirector(port string) func(*http.Request) {99	origin, _ := url.Parse("http://localhost" + port)100	director := func(req *http.Request) {101		req.Header.Add("X-Forwarded-Host", req.Host)102		req.Header.Add("X-Origin-Host", origin.Host)103		req.URL.Scheme = "http"104		req.URL.Host = origin.Host105	}106	return director107}108type Router struct {}109func (r *Router) ServeHTTP(response http.ResponseWriter, request *http.Request){110	var wg sync.WaitGroup111	wg.Add(1)112	worker.addTask(&ServerCall{response, request, &wg})113	wg.Wait()114}115var worker *WorkQueue116func main(){117	ctx := context.TODO()118	var wg sync.WaitGroup119	worker = newWorkQueue(&wg)120	proxy1 := &httputil.ReverseProxy{Director: newDirector(":8000")}121	proxy2 := &httputil.ReverseProxy{Director: newDirector(":9000")}122	proxy3 := &httputil.ReverseProxy{Director: newDirector(":10000")}123	go worker.run(ctx, proxy1)124	go worker.run(ctx, proxy2)125	go worker.run(ctx, proxy3)126	go worker.run(ctx, proxy3)127	go worker.run(ctx, proxy1)128	go worker.run(ctx, proxy2)129	go worker.run(ctx, proxy3)130	go worker.run(ctx, proxy3)131	router := &Router{}132	log.Fatal(http.ListenAndServe(":5000", router))133}...

work_queue.go

Source:work_queue.go

1package main2import (3	"log"4	"os"5	"path/filepath"6	"strings"7	"sync"8)9// JobType is an enumeration of the different types of jobs we handle10type JobType int11const (12	Unknown JobType = iota13	XMLFix14	PDFFix15	FileCopy16)17func (jt JobType) String() string {18	switch jt {19	case XMLFix:20		return "XML fix"21	case PDFFix:22		return "PDF fix"23	case FileCopy:24		return "File copy"25	}26	return "Unknown"27}28// FileRequest holds the source and destination paths for a file which needs to29// be processed, and the type of processing it needs30type Job struct {31	SourcePath string32	DestPath   string33	Type       JobType34	Failures   int35}36// The WorkQueue holds the workers and allows adding jobs and stopping the job37// collection process38type WorkQueue struct {39	workers []*Worker40	queue   chan *Job41	wg      *sync.WaitGroup42}43// NewWorkQueue creates n workers and starts them listening for jobs44func NewWorkQueue(ctx *FixContext, n int) *WorkQueue {45	var q = &WorkQueue{46		workers: make([]*Worker, n),47		queue:   make(chan *Job, 100000),48		wg:      new(sync.WaitGroup),49	}50	for i := 0; i < n; i++ {51		q.workers[i] = &Worker{52			ID:       i,53			queue:    q.queue,54			wg:       q.wg,55			badLCCN:  []byte(ctx.BadLCCN),56			goodLCCN: []byte(ctx.GoodLCCN),57		}58		go q.workers[i].Start()59	}60	return q61}62func (q *WorkQueue) Add(sourcePath, destDir, baseName string) {63	// Create the destination directory if it doesn't exist64	var err = os.MkdirAll(destDir, 0755)65	if err != nil {66		log.Printf("ERROR: could not create %q: %s", destDir, err)67		return68	}69	var ext = strings.ToLower(filepath.Ext(baseName)[1:])70	var destFile = filepath.Join(destDir, baseName)71	var job = &Job{SourcePath: sourcePath, DestPath: destFile}72	switch ext {73	case "xml":74		// For XML, we do NOT want to change the OCR encoding, but we need to nab75		// batch description XMLs76		if len(baseName) > 10 || baseName[:5] == "batch" {77			job.Type = XMLFix78		} else {79			job.Type = FileCopy80		}81	case "pdf":82		job.Type = PDFFix83	default:84		job.Type = FileCopy85	}86	log.Printf("INFO: queueing job for %q (destination %q, type %s)", sourcePath, destDir, job.Type)87	q.queue <- job88}89// Wait blocks until the queue is empty and all workers have quit90func (q *WorkQueue) Wait() {91	for _, w := range q.workers {92		w.Done()93	}94	q.wg.Wait()95}...

newWorkQueue

Using AI Code Generation

1func main() {2    queue := newWorkQueue()3    queue.add(1)4    queue.add(2)5    queue.add(3)6    queue.add(4)7    queue.add(5)8    queue.add(6)9    fmt.Println(queue.get())10    fmt.Println(queue.get())11    fmt.Println(

newWorkQueue

Using AI Code Generation

1import (2func main() {3    wq := newWorkQueue()4    wq.addTask("Task1")5    wq.addTask("Task2")6    wq.addTask("Task3")7    wq.addTask("Task4")8    wq.addTask("Task5")9    wq.addTask("Task6")10    wq.addTask("Task7")11    wq.addTask("Task8")12    wq.addTask("Task9")13    wq.addTask("Task10")14    wq.addTask("Task11")15    wq.addTask("Task12")16    wq.addTask("Task13")17    wq.addTask("Task14")18    wq.addTask("Task15")19    wq.addTask("Task16")20    wq.addTask("Task17")21    wq.addTask("Task18")22    wq.addTask("Task19")23    wq.addTask("Task20")24    wq.addTask("Task21")25    wq.addTask("Task22")26    wq.addTask("Task23")27    wq.addTask("Task24")28    wq.addTask("Task25")29    wq.addTask("Task26")30    wq.addTask("Task27")31    wq.addTask("Task28")32    wq.addTask("Task29")33    wq.addTask("Task30")34    wq.addTask("Task31")35    wq.addTask("Task32")36    wq.addTask("Task33")37    wq.addTask("Task34")38    wq.addTask("Task35")39    wq.addTask("Task36")40    wq.addTask("Task37")41    wq.addTask("Task38")42    wq.addTask("Task39")43    wq.addTask("Task40")44    wq.addTask("Task41")45    wq.addTask("Task42")46    wq.addTask("Task43")47    wq.addTask("Task44")48    wq.addTask("Task45")49    wq.addTask("Task46")50    wq.addTask("Task47")51    wq.addTask("Task48")52    wq.addTask("Task49")53    wq.addTask("Task50")54    wq.addTask("Task51")55    wq.addTask("Task52")56    wq.addTask("Task53

newWorkQueue

Using AI Code Generation

1import (2func main() {3	wq := newWorkQueue(10)4	wq.addWork(20)5	wq.addWork(30)6	wq.addWork(40)7	wq.addWork(50)8	wq.addWork(60)9	wq.addWork(70)10	wq.addWork(80)11	wq.addWork(90)12	wq.addWork(100)13	wq.addWork(110)14	wq.addWork(120)15	wq.addWork(130)16	wq.addWork(140)17	wq.addWork(150)18	wq.addWork(160)19	wq.addWork(170)20	wq.addWork(180)21	wq.addWork(190)22	wq.addWork(200)23	wq.addWork(210)24	wq.addWork(220)25	wq.addWork(230)26	wq.addWork(240)27	wq.addWork(250)28	wq.addWork(260)29	wq.addWork(270)30	wq.addWork(280)31	wq.addWork(290)32	wq.addWork(300)33	wq.addWork(310)34	wq.addWork(320)35	wq.addWork(330)36	wq.addWork(340)37	wq.addWork(350)38	wq.addWork(360)39	wq.addWork(370)40	wq.addWork(380)41	wq.addWork(390)42	wq.addWork(400)43	wq.addWork(410)44	wq.addWork(420)45	wq.addWork(430)46	wq.addWork(440)47	wq.addWork(450)48	wq.addWork(460)49	wq.addWork(470)50	wq.addWork(480)51	wq.addWork(490)52	wq.addWork(500)53	wq.addWork(510)54	wq.addWork(520)55	wq.addWork(530)56	wq.addWork(540)57	wq.addWork(550)58	wq.addWork(560)59	wq.addWork(570)60	wq.addWork(580)61	wq.addWork(590)62	wq.addWork(600)63	wq.addWork(610)64	wq.addWork(620)65	wq.addWork(630)66	wq.addWork(640)67	wq.addWork(650)68	wq.addWork(660)69	wq.addWork(670)70	wq.addWork(680)71	wq.addWork(690)

newWorkQueue

Using AI Code Generation

1import (2func main() {3    fmt.Println("Hello, playground")4    wq := newWorkQueue(100)5    fmt.Println(wq)6    wq.addWork("Hello")7    wq.addWork("World")8    fmt.Println(wq.getWork())9    fmt.Println(wq.getWork())10}11import (12func main() {13    fmt.Println("Hello, playground")14    wq := newWorkQueue(100)15    fmt.Println(wq)16    wq.addWork("Hello")17    wq.addWork("World")18    fmt.Println(wq.getWork())19    fmt.Println(wq.getWork())20}21import (22func main() {23    fmt.Println("Hello, playground")24    wq := newWorkQueue(100)25    fmt.Println(wq)26    wq.addWork("Hello")27    wq.addWork("World")28    fmt.Println(wq.getWork())29    fmt.Println(wq.getWork())30}31import (32func main() {33    fmt.Println("Hello, playground")34    wq := newWorkQueue(100)35    fmt.Println(wq)36    wq.addWork("Hello")37    wq.addWork("World")38    fmt.Println(wq.getWork())39    fmt.Println(wq.getWork())40}41import (42func main() {43    fmt.Println("Hello, playground")44    wq := newWorkQueue(100)45    fmt.Println(wq)46    wq.addWork("Hello")47    wq.addWork("World")48    fmt.Println(wq.getWork())49    fmt.Println(wq.getWork())50}51import (52func main() {53    fmt.Println("Hello, playground")54    wq := newWorkQueue(100)55    fmt.Println(wq)56    wq.addWork("Hello")

newWorkQueue

Using AI Code Generation

1func main() {2    var workQueue = newWorkQueue()3}4import (5type workQueue struct {6    work chan func()7}8func (wq *workQueue) start() {9    go func() {10        for {11            select {12                f()13            }14        }15    }()16}17func (wq *workQueue) stop() {18}19func (wq *workQueue) add(f func()) {20}21func main() {22    var wq = workQueue{23        work: make(chan func()),24        done: make(chan bool),25    }26    wq.start()27    wq.add(func() {28        fmt.Println("Hello, world!")29    })30    wq.add(func() {31        time.Sleep(time.Second)32        fmt.Println("Bye, world!")33    })34    time.Sleep(time.Second * 2)35    wq.stop()36}37import (38type workQueue struct {39    work chan func()40}41func (wq *workQueue) start() {42    go func() {43        for {44            select {45                f()46            }47        }48    }()49}50func (wq *workQueue) stop() {51}52func (wq *workQueue) add(f func()) {53}54func main() {55    var wq = workQueue{56        work: make(chan func()),57        done: make(chan bool),58    }59    wq.start()60    wq.add(func() {61        fmt.Println("Hello, world!")62    })63    wq.add(func() {64        time.Sleep(time.Second)65        fmt.Println("Bye, world!")66    })67    time.Sleep(time.Second * 2)68    wq.stop()69}70import (71type workQueue struct {72    work chan func()73}74func (wq *workQueue) start() {

newWorkQueue

Using AI Code Generation

1import (2func main() {3    fmt.Println("main started")4    wq := newWorkQueue()5    wq.addWork(1)6    wq.addWork(2)7    wq.addWork(3)8    wq.addWork(4)9    wq.addWork(5)10    wq.addWork(6)11    wq.addWork(7)12    wq.addWork(8)13    wq.addWork(9)14    wq.addWork(10)15    wq.addWork(11)16    wq.addWork(12)17    wq.addWork(13)18    wq.addWork(14)19    wq.addWork(15)20    wq.addWork(16)21    wq.addWork(17)22    wq.addWork(18)23    wq.addWork(19)24    wq.addWork(20)25    wq.addWork(21)26    wq.addWork(22)27    wq.addWork(23)28    wq.addWork(24)29    wq.addWork(25)30    wq.addWork(26)31    wq.addWork(

newWorkQueue

Using AI Code Generation

1import (2func main() {3	fmt.Println("hello")4	queue := newWorkQueue()5	queue.AddWork("A")6	queue.AddWork("B")7	queue.AddWork("C")8	queue.StartProcessing()9	queue.AddWork("D")10	queue.AddWork("E")11	queue.AddWork("F")12	queue.StopProcessing()13	queue.AddWork("G")14	queue.AddWork("H")15	queue.AddWork("I")16	queue.StartProcessing()17	queue.StopProcessing()18}19import (20type workQueue struct {21}22func newWorkQueue() *workQueue {23	queue := workQueue{24		work: make(chan string),25		wg:   sync.WaitGroup{},26		stop: make(chan bool),27	}28}29func (queue *workQueue) AddWork(work string) {30}31func (queue *workQueue) StartProcessing() {32	queue.wg.Add(1)33	go queue.process()34}35func (queue *workQueue) StopProcessing() {36	queue.wg.Wait()37}38func (queue *workQueue) process() {39	defer queue.wg.Done()40	for {41		select {

newWorkQueue

Using AI Code Generation

1import (2func main() {3	wq := queues.NewWorkQueue()4	job := queues.NewJob("some_task", 1)5	wq.Add(job)6	j := wq.Next()7	fmt.Println(j.Name)8}9import (10func main() {11	wq := queues.NewWorkQueue()12	job := queues.NewJob("some_task", 1)13	wq.Add(job)14	j := wq.Next()15	fmt.Println(j.Name)16}17import (18func main() {19	wq := queues.NewWorkQueue()20	job := queues.NewJob("some_task", 1)21	wq.Add(job)22	j := wq.Next()23	fmt.Println(j.Name)24}25import (26func main() {27	wq := queues.NewWorkQueue()28	job := queues.NewJob("some_task", 1)29	wq.Add(job)30	j := wq.Next()31	fmt.Println(j.Name)32}

newWorkQueue

Using AI Code Generation

1import (2func main() {3	workQueue := main.NewWorkQueue()4	workQueue.AddWork("work1")5	workQueue.AddWork("work2")6	work := workQueue.GetWork()7	fmt.Println(work)8	work = workQueue.GetWork()9	fmt.Println(work)10}

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.