better task handling

Author: Francesco Cosentino

README.md

@@ -132,7 +132,12 @@ func main() {
     task2 := worker.Task{
         ID:       uuid.New(),
         Priority: 5,
-        Fn:       func() interface{} { return "Hello, World from Task 2!" },
+        // You can pass in parameters to the function and return a value using a closure
+        Fn: func() interface{} {
+            return func(a int, b int) interface{} {
+                return a + b
+            }(2, 5)
+        },
     }
 
     task3 := worker.Task{

examples/manual/main.go

@@ -0,0 +1,39 @@
+package main
+
+import (
+	"fmt"
+	"log"
+
+	"github.com/google/uuid"
+	"github.com/hyp3rd/go-worker"
+	"github.com/hyp3rd/go-worker/middleware"
+)
+
+func main() {
+	tm := worker.NewTaskManager(1, 10)
+	// Example of using zap logger from uber
+	logger := log.Default()
+
+	// apply middleware in the same order as you want to execute them
+	tm = worker.RegisterMiddleware(tm,
+		// middleware.YourMiddleware,
+		func(next worker.Service) worker.Service {
+			return middleware.NewLoggerMiddleware(next, logger)
+		},
+	)
+
+	task := worker.Task{
+		ID:       uuid.New(),
+		Priority: 1,
+		Fn:       func() interface{} { return "Hello, World from Task!" },
+	}
+
+	// tm.RegisterTask(task)
+
+	res, err := tm.ExecuteTask(task.ID)
+	if err != nil {
+		fmt.Println(err)
+	} else {
+		fmt.Println(res)
+	}
+}

examples/middleware/main.go

@@ -2,7 +2,6 @@ package main
 
 import (
 	"fmt"
-	"log"
 	"time"
 
 	"github.com/google/uuid"
@@ -12,21 +11,29 @@ import (
 
 func main() {
 	tm := worker.NewTaskManager(5, 10)
-	// Example of using zap logger from uber
-	logger := log.Default()
 
 	// apply middleware in the same order as you want to execute them
 	tm = worker.RegisterMiddleware(tm,
 		// middleware.YourMiddleware,
 		func(next worker.Service) worker.Service {
-			return middleware.NewLoggerMiddleware(next, logger)
+			return middleware.NewLoggerMiddleware(next, middleware.DefaultLogger())
 		},
 	)
 
 	task := worker.Task{
 		ID:       uuid.New(),
 		Priority: 1,
-		Fn:       func() interface{} { return "Hello, World from Task 1!" },
+		Fn: func() interface{} {
+			return func(a int, b int) interface{} {
+				return a + b
+			}(2, 5)
+		},
+	}
+
+	// Invalid task, it doesn't have a function
+	task1 := worker.Task{
+		ID:       uuid.New(),
+		Priority: 1,
 	}
 
 	task2 := worker.Task{
@@ -55,7 +62,7 @@ func main() {
 		},
 	}
 
-	tm.RegisterTask(task, task2, task3, task4)
+	tm.RegisterTask(task, task1, task2, task3, task4)
 	tm.Start(5)
 
 	tm.CancelTask(task3.ID)

middleware/logger.go

@@ -1,6 +1,9 @@
 package middleware
 
 import (
+	"log"
+	"os"
+	"runtime"
 	"time"
 
 	"github.com/google/uuid"
@@ -13,6 +16,17 @@ type Logger interface {
 	// Errorf(format string, v ...interface{})
 }
 
+// this is a safeguard, breaking on compile time in case
+// `log.Logger` does not adhere to our `Logger` interface.
+// see https://golang.org/doc/faq#guarantee_satisfies_interface
+var _ Logger = &log.Logger{}
+
+// DefaultLogger returns a `Logger` implementation
+// backed by stdlib's log
+func DefaultLogger() *log.Logger {
+	return log.New(os.Stdout, "", log.LstdFlags)
+}
+
 // loggerMiddleware is a middleware that logs the time it takes to execute the next middleware.
 // Must implement the `worker.Service` interface.
 type loggerMiddleware struct {
@@ -28,21 +42,23 @@ func NewLoggerMiddleware(next worker.Service, logger Logger) worker.Service {
 // RegisterTask registers a new task to the worker
 func (mw *loggerMiddleware) RegisterTask(tasks ...worker.Task) {
 	defer func(begin time.Time) {
+		for _, t := range tasks {
+			mw.logger.Printf("registering task ID %v with priority: %v", t.ID, t.Priority)
+		}
 		mw.logger.Printf("`RegisterTask` took: %s", time.Since(begin))
 	}(time.Now())
 
-	for _, t := range tasks {
-		mw.logger.Printf("registered task ID %v with priority: %v", t.ID, t.Priority)
-	}
-
 	mw.next.RegisterTask(tasks...)
 }
 
 // Start the task manager
 func (mw *loggerMiddleware) Start(numWorkers int) {
 	defer func(begin time.Time) {
-		mw.logger.Printf("`Start` took: %s", time.Since(begin))
+		// var numCPU = runtime.GOMAXPROCS(0)
+		var numCPU = runtime.NumCPU()
+		mw.logger.Printf("the task manager is running on %v CPUs", numCPU)
 		mw.logger.Printf("the task manager started with %v workers", numWorkers)
+		mw.logger.Printf("`Start` took: %s", time.Since(begin))
 	}(time.Now())
 
 	mw.next.Start(numWorkers)
@@ -68,6 +84,21 @@ func (mw *loggerMiddleware) GetTasks() []worker.Task {
 	return mw.next.GetTasks()
 }
 
+// ExecuteTask executes a task given its ID and returns the result
+func (mw *loggerMiddleware) ExecuteTask(id uuid.UUID) (res interface{}, err error) {
+	defer func(begin time.Time) {
+		mw.logger.Printf("`ExecuteTask` took: %s", time.Since(begin))
+	}(time.Now())
+
+	mw.logger.Printf("executing task ID %v", id)
+
+	// if res, err = mw.next.ExecuteTask(id); err != nil {
+	// 	mw.logger.Printf("error while executing task ID %v - %v", id, err)
+	// }
+
+	return mw.next.ExecuteTask(id)
+}
+
 // CancelAll cancels all tasks
 func (mw *loggerMiddleware) CancelAll() {
 	mw.next.CancelAll()

service.go

@@ -16,6 +16,8 @@ type Service interface {
 	GetTask(id uuid.UUID) (task Task, ok bool)
 	// GetTasks gets all tasks
 	GetTasks() []Task
+	// ExecuteTask executes a task given its ID and returns the result
+	ExecuteTask(id uuid.UUID) (interface{}, error)
 	// CancelAll cancels all tasks
 	CancelAll()
 	// CancelTask cancels a task by its ID

task.go

@@ -2,22 +2,70 @@ package worker
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"sync/atomic"
 	"time"
 
 	"github.com/google/uuid"
 )
 
+var (
+	// ErrInvalidTaskID is returned when a task has an invalid ID
+	ErrInvalidTaskID = errors.New("invalid task id")
+	// ErrInvalidTaskFunc is returned when a task has an invalid function
+	ErrInvalidTaskFunc = errors.New("invalid task function")
+	// ErrTaskNotFound is returned when a task is not found
+	ErrTaskNotFound = errors.New("task not found")
+)
+
+// CancelReason is a value used to represent the cancel reason.
+type CancelReason uint8
+
+// CancelReason values
+//   - 1: `ContextDeadlineReached`
+//   - 2: `RateLimited`
+//   - 3: `Cancelled`
+const (
+	// ContextDeadlineReached means the context is past its deadline.
+	ContextDeadlineReached = CancelReason(1)
+	// RateLimited means the number of concurrent tasks per second exceeded the maximum allowed.
+	RateLimited = CancelReason(2)
+	// Cancelled means `CancelTask` was invked and the `Task` was cancelled.
+	Cancelled = CancelReason(3)
+)
+
+// TaskFunc signature of `Task` function
+type TaskFunc func() interface{}
+
 // Task represents a function that can be executed by the task manager
 type Task struct {
-	ID        uuid.UUID          `json:"id"`        // ID is the id of the task
-	Priority  int                `json:"priority"`  // Priority is the priority of the task
-	Fn        func() interface{} `json:"-"`         // Fn is the function that will be executed by the task
-	Ctx       context.Context    `json:"context"`   // Ctx is the context of the task
-	Cancel    context.CancelFunc `json:"-"`         // Cancel is the cancel function of the task
-	Started   atomic.Int64       `json:"started"`   // Started is the time the task started
-	Completed atomic.Int64       `json:"completed"` // Completed is the time the task completed
-	Cancelled atomic.Int64       `json:"cancelled"` // Cancelled is the time the task was cancelled
+	ID           uuid.UUID          `json:"id"`            // ID is the id of the task
+	Priority     int                `json:"priority"`      // Priority is the priority of the task
+	Fn           TaskFunc           `json:"-"`             // Fn is the function that will be executed by the task
+	Ctx          context.Context    `json:"context"`       // Ctx is the context of the task
+	Cancel       context.CancelFunc `json:"-"`             // Cancel is the cancel function of the task
+	Error        atomic.Value       `json:"error"`         // Error is the error of the task
+	Started      atomic.Int64       `json:"started"`       // Started is the time the task started
+	Completed    atomic.Int64       `json:"completed"`     // Completed is the time the task completed
+	Cancelled    atomic.Int64       `json:"cancelled"`     // Cancelled is the time the task was cancelled
+	CancelReason CancelReason       `json:"cancel_reason"` // CancelReason is the reason the task was cancelled
+}
+
+// IsValid returns an error if the task is invalid
+func (t Task) IsValid() (err error) {
+	fmt.Println("validating task")
+	if t.ID == uuid.Nil {
+		err = ErrInvalidTaskID
+		t.Error.Store(err.Error())
+		return
+	}
+	if t.Fn == nil {
+		err = ErrInvalidTaskFunc
+		t.Error.Store(err.Error())
+		return
+	}
+	return
 }
 
 // setStarted handles the start of a task by setting the start time
@@ -42,7 +90,7 @@ type taskHeap []Task
 func (h taskHeap) Len() int { return len(h) }
 
 // Less returns true if the priority of the first task is less than the second task
-func (h taskHeap) Less(i, j int) bool { return h[i].Priority > h[j].Priority }
+func (h taskHeap) Less(i, j int) bool { return h[i].Priority < h[j].Priority }
 
 // Swap swaps the position of two tasks in the heap
 func (h taskHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }