How to log pool goroutine panic without blocking on task.Wait()

[sandykellagher]

Hello,
I have a worker pool processing tasks read from a channel

pool := pond.NewPool(s.numWorkers)
for msg := range s.queue {
   pool.Submit(func() { doProcess( msg) })
}

If doProcess should ever fail, eg panic, this is nicely wrapped and handled by the wrapper. But this means there is no log warning message of any kind.

But if the code does a task.Wait() to check for an error (as shown below), then the items being pulled off the channel end up being processed in series rather than in parallel.

pool := pond.NewPool(s.numWorkers)
for msg := range s.queue {
   task := pool.Submit(func() { doProcess( msg) })
   err := task.Wait()
   if err != nil {
    log.Printf("Got error: %v\n", err)
   }
}

I could spawn a goroutine for each pool.Submit but that seems very heavy-handed.

Is there an elegant way to log any task failure that might occur, while still submitting tasks in parallel. For example, a callback function that can be set. Failing that, arguably it'd be preferable not to wrap the panic, so that we get a log message if a code panic occurs.

Thanks
Sandy

[alitto]

Hi @sandykellagher!

At the moment, the library doesn't provide a way to execute an arbitrary function when a task completes or panics, but it could be a nice addition.
In the meantime, you could try creating a function that wraps tasks and logs panics thrown during its execution:

func taskLogger[R any](task func()) func() {
    return func() {
	defer func() {
            if p := recover(); p != nil {
                log.Printf("Task panicked: %v, %s\n", p, debug.Stack())
	    }
	}()

        // Execute task
	task()
	
	log.Println("Task completed")
    }
}

And you could use it as follows:

pool := pond.NewPool(s.numWorkers)

for msg := range s.queue {
    pool.Submit(taskLogger(func() {
       doProcess(msg)
    }))
}

// Stop the pool and wait for tasks to complete
pool.StopAndWait()

Alternatively, if you don't want to stop the pool, you could use a group:

pool := pond.NewPool(s.numWorkers)
group := pool.NewGroup()

for msg := range s.queue {
    group.Submit(taskLogger(func() {
       doProcess(msg)
    }))
}

// Wait for tasks to complete
group.Wait()

Please let me know if this workaround could work for your use case. But I think we could add something to the library to do this without a wrapper.

[sandykellagher]

Hi, thanks for the prompt reply. That's exactly what I was after.

As it turns out, the Ants goroutine worker pool package has an almost identical set of APIs, and its default behaviour is to wrap submitted tasks so as to recover from any panic and log any panic info and stack trace to console.

And I think that is the best behaviour: it's useful to have tasks wrapped so that any panic is not fatal, but equally it's not helpful in development or debug to recover silently from panic events.

[alitto]

100% agree.

Here's another alternative to log panics but also surface them as errors to the caller:

func taskLogger[R any](task func()) func() error {
    return func() (err error) {
	defer func() {
            if p := recover(); p != nil {
                log.Printf("Task panicked: %v, %s\n", p, debug.Stack())

                if e, ok := p.(error); ok {
                    err = fmt.Errorf("Task panicked: %w, %s", e, debug.Stack())
                } else {
                    err = fmt.Errorf("Task panicked: %v, %s", p, debug.Stack())
                }
	    }
	}()

        // Execute task
	task()
	
	log.Println("Task completed")
	
	return
    }
}

And you could use it with SubmitErr instead of Submit like this:

pool := pond.NewPool(s.numWorkers)
group := pool.NewGroup()

for msg := range s.queue {
    group.SubmitErr(taskLogger(func() {
       doProcess(msg)
    }))
}

// Wait for tasks to complete or the first error, whatever occurs first
if err := group.Wait(); err != nil {
    log.Printf("Some tasks failed: %v", err)
}

This is a nice way of keeping control of how panics are logged/handled and its also great for testability.