Initial commit

Author: pbanos

Diff for: README.md

@@ -0,0 +1,39 @@
+# Ingram Micro Senior Backend Developer test
+
+This task challenges you to implement in Go an in-memory queue satisfying a specific interface. Our intention behind this task is to have something to talk about in an interview, so it does not matter if you get stuck and cannot complete it: just deliver whatever you got.
+
+## Setup
+
+- [Install Go](https://golang.org/doc/install#install) if you do not have it already
+- Clone this repository outside your GOPATH (because it is using [go modules](https://blog.golang.org/using-go-modules))
+
+## Develop
+
+You should write your code in the queue package, modifying the worker.go file to complete the New method and maybe in other .go files. You should probably leave the interfaces.go file as it is.
+
+As specified below, the delivery format is a git repository. It would be nice that the git history showed your progress.
+
+### Run your code
+
+The main.go file in the top-level directory of this repo will use your implementation to approximate pi. Before delivering your task you should try your best to make it work. You can run it with:
+
+```Bash
+go run main.go
+```
+
+### Tips
+
+* Channels are not a silver bullet in go
+* You will probably need to get familiar with the [context package](https://golang.org/pkg/context/). The following blog post can be an introductory read: http://p.agnihotry.com/post/understanding_the_context_package_in_golang/
+
+### Bonus tracks
+
+If you really want to impress us by going the extra mile you could do one of the following:
+
+- Add tests for your code
+- Dockerize the running of the main.go file
+- Implement a persisted (file, redis or something else -based) implementation of the queue, maybe in another package, that satisfies the interface.
+
+## Deliver
+
+Create a public github repository and push your code there. Then send us an email with a link to it.

Diff for: go.mod

@@ -0,0 +1,3 @@
+module github.com/ingrammicro/backend-test
+
+go 1.12

Diff for: main.go

@@ -0,0 +1,138 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"math/big"
+	"math/rand"
+	"time"
+
+	"github.com/ingrammicro/backend-test/queue"
+)
+
+// piComputeData holds both the input and output of a pi processing job
+// Total (both input and output) is the number of random points to pick
+// in a [0,1)x[0,1) square.
+// InCircle (only output) is the number of the randomly picked points that
+// where inside the circle of radius 1 cented in (0,0).
+type piComputeData struct {
+	InCircle uint64 `json:"i"`
+	Total    uint64 `json:"t"`
+}
+
+// piProcessor is a processor that can work out pi processing jobs
+type piProcessor struct{}
+
+// Process processes a pi processing job. To do so, it extracts piComputeData from
+// the given job, computes it and stores it back into the job. It returns an error
+// if any of the three operations fail.
+func (pp piProcessor) Process(ctx context.Context, j queue.JobProcessingAccess) error {
+	pcd := &piComputeData{}
+	err := j.GetData(pcd)
+	if err != nil {
+		return err
+	}
+	err = pcd.Compute(ctx)
+	if err != nil {
+		return err
+	}
+	err = j.SetData(ctx, pcd)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// Compute picks a Total number of points in the [0,1)x[0,1) square
+// and checks for each of one if they are inside the circle of radius 1 cented in (0,0).
+// Specifically, given a (x,y) point, it checks whether x²+y² <= 1.
+// It updates InCircle with the number of points that were inside.
+func (pcd *piComputeData) Compute(ctx context.Context) error {
+	r := rand.New(rand.NewSource(time.Now().UTC().UnixNano()))
+	for i := uint64(0); i < pcd.Total; i++ {
+		x, y := r.Float64(), r.Float64()
+		if (x*x)+(y*y) <= 1 {
+			pcd.InCircle++
+		}
+	}
+	return nil
+}
+
+func (pcd *piComputeData) String() string {
+	return fmt.Sprintf("%d/%d", pcd.InCircle, pcd.Total)
+}
+
+// Marshal encodes the piComputeData into the returned byte slice
+// as JSON.
+func (pcd *piComputeData) Marshal() ([]byte, error) {
+	buf := &bytes.Buffer{}
+	err := json.NewEncoder(buf).Encode(pcd)
+	if err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
+// Unmarshal decodes the JSON in the given byte slice
+// and fills the piComputeData with it.
+func (pcd *piComputeData) Unmarshal(b []byte) error {
+	return json.NewDecoder(bytes.NewReader(b)).Decode(pcd)
+}
+
+// main pushes numberOfJobs pi processing jobs (each computing a million points),
+// starts 10 workers, waits for all the jobs to be processed and then aggregates
+// the results of the jobs to approximate pi. Finally, it prints the approximation
+// and exits orderly.
+func main() {
+	const numberOfJobs = 10000
+	ctx, cancelCtx := context.WithCancel(context.Background())
+	defer cancelCtx()
+	client, worker := queue.New(piProcessor{})
+	log.Printf("Pushing %d pi processing jobs...", numberOfJobs)
+	for i := 0; i < numberOfJobs; i++ {
+		client.CreateJob(ctx, fmt.Sprintf("j-%d", i), &piComputeData{Total: 1000000})
+	}
+	log.Print("Starting 10 workers...")
+	workerStopped := make(chan struct{})
+	go func() {
+		worker.Run(ctx, 10)
+		close(workerStopped)
+	}()
+	log.Print("Waiting for results and aggregating them...")
+	result := &big.Rat{}
+	for i := 0; i < numberOfJobs; i++ {
+		jobID := fmt.Sprintf("j-%d", i)
+		for {
+			job, err := client.GetJob(ctx, jobID)
+			if err != nil {
+				log.Fatal(err)
+			}
+			if job == nil {
+				log.Fatalf("Job %q could not be found", jobID)
+			}
+			state := job.State()
+			if state == queue.Failed {
+				log.Fatal(job.Error())
+			}
+			if state == queue.Finished {
+				var partialResult piComputeData
+				err = job.GetData(&partialResult)
+				if err != nil {
+					log.Fatal(err)
+				}
+				result = result.Add(result, big.NewRat(4*int64(partialResult.InCircle), int64(partialResult.Total)))
+				break
+			}
+			time.Sleep(5 * time.Second) // Wait a bit for the job to finish
+		}
+	}
+	cancelCtx()
+	result = result.Mul(result, big.NewRat(1, numberOfJobs))
+	log.Printf("Result is %+v = %s", result, result.FloatString(20))
+	log.Printf("Preparing to exit...")
+	<-workerStopped
+	log.Printf("Exiting")
+}

Diff for: queue/interface.go

@@ -0,0 +1,106 @@
+package queue
+
+import (
+	"context"
+)
+
+// State represents the state of a job
+type State string
+
+// MarshalUnmarshaler wraps the Marshal and Unmarshal methods.
+// Job payloads must implement this interface that allows
+// converting them into a slice of bytes and back.
+//
+// The Marshal method marshals the object into a stream of bytes
+// and returns the slice of bytes of an error.
+//
+// The Unmarshal method unmarshals an object from a stream of bytes
+// into itself and may return an error.
+//
+// Implementations should ensure the unmarshaling of the slice of bytes
+// provided by the marshaling of an object provides essentially the same
+// object.
+//
+// This interface has already an implementation by us in the main.go file.
+type MarshalUnmarshaler interface {
+	Marshal() ([]byte, error)
+	Unmarshal([]byte) error
+}
+
+// Job is the interface that wraps the methods
+// use to read data from a job obtained from a client.
+//
+// The ID methods provide the ID of the job.
+//
+// The GetData method unmarshals into data the payload of the job.
+//
+// The State method returns the State of the job.
+//
+// The Error method returns a string describing the error with which a job failed.
+type Job interface {
+	ID() string
+	GetData(data MarshalUnmarshaler) error
+	State() State
+	Error() string
+}
+
+// JobProcessingAccess is just the Job interface with an extra method
+// that allows setting the data payload of the job. It is meant to be
+// used by job processors launched by workers, which will need to store
+// their results there.
+//
+// The SetData method takes some data and updates the
+// job's payload data with it. Implementations should use the context
+// argument to allow canceling or expiration of the SetData operation,
+// and return an error in that case or if the payload data update cannot
+// be performed.
+type JobProcessingAccess interface {
+	Job
+	SetData(ctx context.Context, data MarshalUnmarshaler) error
+}
+
+// A Processor defines the worker's job execution.
+// It returns an error:
+//  * If the error is not nil, the job is marked as Failed.
+//  * If the error is nil the job is marked as finished
+//    (successfully).
+//
+// This interface has already an implementation by us in the main.go file.
+type Processor interface {
+	Process(ctx context.Context, j JobProcessingAccess) error
+}
+
+// Worker is an interface that wraps the Run method, which
+// allows processing jobs in a queue.
+//
+// Implementations should attempt to process as many jobs as
+// the given worker integer simultaneously using a Processor.
+// They should also use the given context to allow users to timeout
+// or cancel processing, returning only after all workers have stopped.
+type Worker interface {
+	Run(ctx context.Context, workers int) error
+}
+
+// Client is an interface that allows pushing jobs into a queue
+// and querying their state and results.
+//
+// Implementations of GetJob should return
+//  * a nil job and a nil error when the job is not found
+//  * the job and a nil error when the job is found
+//  * a nil job and an error, when some error prevents the retrieval
+//    of the job
+type Client interface {
+	CreateJob(ctx context.Context, id string, initialData MarshalUnmarshaler) error
+	GetJob(ctx context.Context, id string) (Job, error)
+}
+
+const (
+	// Queued but not processing yet
+	Queued State = "queued"
+	// Processing - The worker got active
+	Processing State = "processing"
+	// Failed by worker or from outside (timeout) - See Jobs's Error field
+	Failed State = "failed"
+	// Finished successfully
+	Finished State = "finished"
+)

Diff for: queue/worker.go

@@ -0,0 +1,10 @@
+package queue
+
+// New takes a processor and returns both
+// a client and a worker. The client allows
+// pushing jobs to the queue (with CreateJob)
+// and the worker can run those jobs using
+// the given Processor
+func New(p Processor) (Client, Worker) {
+	return nil, nil
+}