oyd
/
minio
6
1
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Remove probe and tasker. (#1733)

Browse Source 
Fixes #1717
master
Harshavardhana 9 years ago committed by Anand Babu (AB) Periasamy
parent b48b2e7f7c
commit 64b0976e1b
11 changed files with 27 additions and 772 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 30
      logger.go
  2. 5
      main.go
  3. 217
      pkg/probe/probe.go
  4. 64
      pkg/probe/probe_test.go
  5. 43
      pkg/probe/wrapper.go
  6. 44
      pkg/tasker/commands.go
  7. 56
      pkg/tasker/handle.go
  8. 35
      pkg/tasker/status.go
  9. 103
      pkg/tasker/task.go
  10. 164
      pkg/tasker/taskctl.go
  11. 38
      pkg/tasker/taskctl_test.go

30
logger.go
Unescape View File

@ -17,11 +17,14 @@
package main
import (
"os"
"reflect"
"runtime"
"runtime/debug"
"strconv"
"github.com/Sirupsen/logrus"
"github.com/minio/minio/pkg/probe"
"github.com/dustin/go-humanize"
)
type fields map[string]interface{}
@ -42,12 +45,33 @@ type logger struct {
// Add new loggers here.
}
// getSysInfo returns useful system statistics.
func getSysInfo() map[string]string {
host, err := os.Hostname()
if err != nil {
host = ""
}
memstats := &runtime.MemStats{}
runtime.ReadMemStats(memstats)
return map[string]string{
"host.name": host,
"host.os": runtime.GOOS,
"host.arch": runtime.GOARCH,
"host.lang": runtime.Version(),
"host.cpus": strconv.Itoa(runtime.NumCPU()),
"mem.used": humanize.Bytes(memstats.Alloc),
"mem.total": humanize.Bytes(memstats.Sys),
"mem.heap.used": humanize.Bytes(memstats.HeapAlloc),
"mem.heap.total": humanize.Bytes(memstats.HeapSys),
}
}
// errorIf synonymous with fatalIf but doesn't exit on error != nil
func errorIf(err error, msg string, data ...interface{}) {
if err == nil {
return
}
sysInfo := probe.GetSysInfo()
sysInfo := getSysInfo()
fields := logrus.Fields{
"cause": err.Error(),
"type": reflect.TypeOf(err),
@ -65,7 +89,7 @@ func fatalIf(err error, msg string, data ...interface{}) {
if err == nil {
return
}
sysInfo := probe.GetSysInfo()
sysInfo := getSysInfo()
fields := logrus.Fields{
"cause": err.Error(),
"type": reflect.TypeOf(err),

5
main.go
Unescape View File

@ -24,7 +24,6 @@ import (
"github.com/minio/cli"
"github.com/minio/mc/pkg/console"
"github.com/minio/minio/pkg/probe"
"github.com/pkg/profile"
)
@ -154,10 +153,6 @@ func mustGetProfilePath() string {
}
func main() {
probe.Init() // Set project's root source path.
probe.SetAppInfo("Release-Tag", minioReleaseTag)
probe.SetAppInfo("Commit-ID", minioShortCommitID)
app := registerApp()
app.Before = func(c *cli.Context) error {
// Sets new config folder.

217
pkg/probe/probe.go
Unescape View File

@ -1,217 +0,0 @@
/*
* Minimalist Object Storage, (C) 2015 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses)/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// Package probe implements a simple mechanism to trace and return errors in large programs.
package probe
import (
"fmt"
"os"
"path/filepath"
"runtime"
"strconv"
"strings"
"sync"
"github.com/dustin/go-humanize"
)
var (
// Root path to the project's source.
rootPath string
// App specific info to be included reporting.
appInfo map[string]string
)
// Init initializes probe. It is typically called once from the main()
// function or at least from any source file placed at the top level
// source directory.
func Init() {
// Root path is automatically determined from the calling function's source file location.
// Catch the calling function's source file path.
_, file, _, _ := runtime.Caller(1)
// Save the directory alone.
rootPath = filepath.Dir(file)
appInfo = make(map[string]string)
}
// SetAppInfo sets app speific key:value to report additionally during call trace dump.
// Eg. SetAppInfo("ReleaseTag", "RELEASE_42_0")
// SetAppInfo("Version", "42.0")
// SetAppInfo("Commit", "00611fb")
func SetAppInfo(key, value string) {
appInfo[key] = value
}
// GetSysInfo returns useful system statistics.
func GetSysInfo() map[string]string {
host, err := os.Hostname()
if err != nil {
host = ""
}
memstats := &runtime.MemStats{}
runtime.ReadMemStats(memstats)
return map[string]string{
"host.name": host,
"host.os": runtime.GOOS,
"host.arch": runtime.GOARCH,
"host.lang": runtime.Version(),
"host.cpus": strconv.Itoa(runtime.NumCPU()),
"mem.used": humanize.Bytes(memstats.Alloc),
"mem.total": humanize.Bytes(memstats.Sys),
"mem.heap.used": humanize.Bytes(memstats.HeapAlloc),
"mem.heap.total": humanize.Bytes(memstats.HeapSys),
}
}
// TracePoint container for individual trace entries in overall call trace
type TracePoint struct {
Line int `json:"line,omitempty"`
Filename string `json:"file,omitempty"`
Function string `json:"func,omitempty"`
Env map[string][]string `json:"env,omitempty"`
}
// Error implements tracing error functionality.
type Error struct {
lock sync.RWMutex
Cause error `json:"cause,omitempty"`
CallTrace []TracePoint `json:"trace,omitempty"`
SysInfo map[string]string `json:"sysinfo,omitempty"`
}
// NewError function instantiates an error probe for tracing.
// Default ``error`` (golang's error interface) is injected in
// only once. Rest of the time, you trace the return path with
// ``probe.Trace`` and finally handling them at top level
//
// Following dummy code talks about how one can pass up the
// errors and put them in CallTrace.
//
// func sendError() *probe.Error {
// return probe.NewError(errors.New("Help Needed"))
// }
// func recvError() *probe.Error {
// return sendError().Trace()
// }
// if err := recvError(); err != nil {
// log.Fatalln(err.Trace())
// }
//
func NewError(e error) *Error {
if e == nil {
return nil
}
Err := Error{lock: sync.RWMutex{}, Cause: e, CallTrace: []TracePoint{}, SysInfo: GetSysInfo()}
return Err.trace() // Skip NewError and only instead register the NewError's caller.
}
// Trace records the point at which it is invoked.
// Stack traces are important for debugging purposes.
func (e *Error) Trace(fields ...string) *Error {
if e == nil {
return nil
}
e.lock.Lock()
defer e.lock.Unlock()
return e.trace(fields...)
}
// trace records caller's caller. It is intended for probe's own
// internal use. Take a look at probe.NewError for example.
func (e *Error) trace(fields ...string) *Error {
if e == nil {
return nil
}
pc, file, line, _ := runtime.Caller(2)
function := runtime.FuncForPC(pc).Name()
_, function = filepath.Split(function)
file = strings.TrimPrefix(file, rootPath+string(os.PathSeparator)) // trims project's root path.
tp := TracePoint{}
if len(fields) > 0 {
tp = TracePoint{Line: line, Filename: file, Function: function, Env: map[string][]string{"Tags": fields}}
} else {
tp = TracePoint{Line: line, Filename: file, Function: function}
}
e.CallTrace = append(e.CallTrace, tp)
return e
}
// Untrace erases last known trace entry.
func (e *Error) Untrace() *Error {
if e == nil {
return nil
}
e.lock.Lock()
defer e.lock.Unlock()
l := len(e.CallTrace)
if l == 0 {
return nil
}
e.CallTrace = e.CallTrace[:l-1]
return e
}
// ToGoError returns original error message.
func (e *Error) ToGoError() error {
if e == nil || e.Cause == nil {
return nil
}
return e.Cause
}
// String returns error message.
func (e *Error) String() string {
if e == nil || e.Cause == nil {
return "<nil>"
}
e.lock.RLock()
defer e.lock.RUnlock()
if e.Cause != nil {
str := e.Cause.Error()
callLen := len(e.CallTrace)
for i := callLen - 1; i >= 0; i-- {
if len(e.CallTrace[i].Env) > 0 {
str += fmt.Sprintf("\n (%d) %s:%d %s(..) Tags: [%s]",
i, e.CallTrace[i].Filename, e.CallTrace[i].Line, e.CallTrace[i].Function, strings.Join(e.CallTrace[i].Env["Tags"], ", "))
} else {
str += fmt.Sprintf("\n (%d) %s:%d %s(..)",
i, e.CallTrace[i].Filename, e.CallTrace[i].Line, e.CallTrace[i].Function)
}
}
str += "\n "
for key, value := range appInfo {
str += key + ":" + value + " | "
}
str += "Host:" + e.SysInfo["host.name"] + " | "
str += "OS:" + e.SysInfo["host.os"] + " | "
str += "Arch:" + e.SysInfo["host.arch"] + " | "
str += "Lang:" + e.SysInfo["host.lang"] + " | "
str += "Mem:" + e.SysInfo["mem.used"] + "/" + e.SysInfo["mem.total"] + " | "
str += "Heap:" + e.SysInfo["mem.heap.used"] + "/" + e.SysInfo["mem.heap.total"]
return str
}
return "<nil>"
}

64
pkg/probe/probe_test.go
Unescape View File

@ -1,64 +0,0 @@
/*
* Minio Cloud Storage, (C) 2015 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package probe_test
import (
"os"
"testing"
"github.com/minio/minio/pkg/probe"
. "gopkg.in/check.v1"
)
func Test(t *testing.T) { TestingT(t) }
type MySuite struct{}
var _ = Suite(&MySuite{})
func testDummy0() *probe.Error {
_, e := os.Stat("this-file-cannot-exit")
return probe.NewError(e)
}
func testDummy1() *probe.Error {
return testDummy0().Trace("DummyTag1")
}
func testDummy2() *probe.Error {
return testDummy1().Trace("DummyTag2")
}
func (s *MySuite) TestProbe(c *C) {
probe.Init() // Set project's root source path.
probe.SetAppInfo("Commit-ID", "7390cc957239")
es := testDummy2().Trace("TopOfStack")
// Uncomment the following Println to visually test probe call trace.
// fmt.Println("Expecting a simulated error here.", es)
c.Assert(es, Not(Equals), nil)
newES := es.Trace()
c.Assert(newES, Not(Equals), nil)
}
func (s *MySuite) TestWrappedError(c *C) {
_, e := os.Stat("this-file-cannot-exit")
es := probe.NewError(e) // *probe.Error
e = probe.WrapError(es) // *probe.WrappedError
_, ok := probe.UnwrapError(e)
c.Assert(ok, Equals, true)
}

43
pkg/probe/wrapper.go
Unescape View File

@ -1,43 +0,0 @@
/*
* Minimalist Object Storage, (C) 2015 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses)/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// Package probe implements a simple mechanism to trace and return errors in large programs.
package probe
// wrappedError implements a container for *probe.Error.
type wrappedError struct {
err *Error
}
// WrapError function wraps a *probe.Error into a 'error' compatible duck type.
func WrapError(err *Error) error {
return &wrappedError{err: err}
}
// UnwrapError tries to convert generic 'error' into typed *probe.Error and returns true, false otherwise.
func UnwrapError(err error) (*Error, bool) {
switch e := err.(type) {
case *wrappedError:
return e.err, true
default:
return nil, false
}
}
// Error interface method.
func (w *wrappedError) Error() string {
return w.err.String()
}

44
pkg/tasker/commands.go
Unescape View File

@ -1,44 +0,0 @@
/*
* Minio Cloud Storage, (C) 2015 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package tasker
// Command is number that uniquely identifies a command function.
type Command uint8
// Enumerate the task commands.
const (
// CmdNOOP does nothing. It is a default placeholder. Uninitialized variable of this type will point to NOOP command by default.
CmdNOOP Command = iota
// CmdSignalEnd gracefully ends current task. Never ending tasks (loop over) or Batched jobs will not take the next iteration,
// but may finish the current state to completion.
CmdSignalEnd
// CmdSignalAbort ends the current task at hand immediately. It may still cleanup dangling issues quickly.
CmdSignalAbort
// CmdSignalSuspend suspends the current task.
CmdSignalSuspend
// CmdSignalResume resumes a suspended task.
CmdSignalResume
// CmdPriorityLow is optimized to conserve resources and complete the task at a slow pace. This option is ideal for batch processed tasks.
CmdPriorityLow
// CmdPriorityMedium is the default priority. It is a balanced option between resources and speed.
CmdPriorityMedium
// CmdPriorityHigh is optimized for speed. This option is ideal for short lived tasks (like meta-data related) that are latency sensitive. Use this option wisely.
CmdPriorityHigh
// CmdPrioritySuper is an exclusive priority. All tasks with priority lower than Super (including High) are paused
// temporarily until this task completes. Anytime you consider using this priority level, please seek for approval.
CmdPrioritySuper
)

56
pkg/tasker/handle.go
Unescape View File

@ -1,56 +0,0 @@
/*
* Minio Cloud Storage, (C) 2015 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package tasker
import "github.com/minio/minio/pkg/probe"
// Handle as the name suggests is a handle (self reference) to its
// own task structure. Task has limited privileges over itself. Only the
// task controller (TaskCtl) can manage the task by sending commands to
// the task over channels.
type Handle struct {
this taskRef
cmdCh <-chan Command // Channel to receive commands from TaskCtl.
statusCh chan<- status // Channel to send completion status and error (if any) to TaskCtl.
closeCh chan<- taskRef // Channel to notify the TaskCtl about ending this task.
}
// Listen returns a channel to receive commands.
func (t Handle) Listen() <-chan Command {
return t.cmdCh
}
// StatusDone acknowledges successful completion of a command.
func (t Handle) StatusDone() {
t.statusCh <- status{code: statusDone, err: nil}
}
// StatusBusy rejects a command with busy status.
func (t Handle) StatusBusy() {
t.statusCh <- status{code: statusBusy, err: nil}
}
// StatusFail returns failure status.
func (t Handle) StatusFail(err *probe.Error) {
t.statusCh <- status{code: statusFail, err: err}
}
// Close notifies the TaskCtl about the end of this Task. Owner of the
// task must invoke Close() when it is done performing its job.
func (t Handle) Close() {
t.closeCh <- t.this
}

35
pkg/tasker/status.go
Unescape View File

@ -1,35 +0,0 @@
/*
* Minio Cloud Storage, (C) 2015 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package tasker
import "github.com/minio/minio/pkg/probe"
// StatusCode denotes the completion status of a command.
type statusCode int8
// Enumerate task return status codes.
const (
statusDone statusCode = iota
statusBusy
statusFail
)
// Status returns the completion status and error (if any) of a command.
type status struct {
code statusCode // Completion code.
err *probe.Error // Error if any.
}

103
pkg/tasker/task.go
Unescape View File

@ -1,103 +0,0 @@
/*
* Minio Cloud Storage, (C) 2015 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package tasker
import (
"container/list"
"sync"
)
// NOTE: Task is a private entity. It is created and managed by TaskCtl
// entirely. Only TaskCtl and Handle objects are exposed outside.
// taskRef is a unique reference ID to a task. It is assigned by the
// TaskCtl during the creation of a task. All tasfRef variables are
// named "this".
type taskRef *list.Element
// Task is an abstract concept built on top of Go routines and
// channels. Tasks themselves are expected to co-operate and comply with
// the TaskCtl commands.
type task struct {
mutex *sync.Mutex
this taskRef // Refence to task entry in the TaskCtl's task list.
name string // Free form name.
priority Command // Current priority.
cmdCh chan Command // Channel to receive commands from TaskCtl.
statusCh chan status // Channel to send completion status and error (if any) to TaskCtl.
closeCh chan taskRef // Channel to notify the TaskCtl about ending this task.
}
// NewTask creates a new task structure and returns a handle to
// it. Only the task controller has access to the task structure. The
// caller routine only receives a handle to its task structure. Task
// handle is like a reference to task self. Caller is expected to listen
// for commands from the task controller and comply with it co-operatively.
// this: Task reference is unique identifier assigned by the TaskCtl.
// name: Free form name of the task. Eg. "Late Night Disk Scrubber".
func newTask(name string) task {
return task{
// this: Is set by the TaskCtl's NewTask function.
mutex: &sync.Mutex{},
name: name,
priority: CmdPriorityMedium,
cmdCh: make(chan Command),
statusCh: make(chan status),
closeCh: make(chan taskRef),
}
}
// getHandle returns a handle to the task. Handle has limited access to the task structure and it is safe to be exposed.
func (t task) getHandle() Handle {
t.mutex.Lock()
defer t.mutex.Unlock()
// Make a handle with limited access to channels (only send or receive).
return Handle{
cmdCh: t.cmdCh,
statusCh: t.statusCh,
closeCh: t.closeCh,
}
}
// command method sends a command code to the task and returns its completion status.
func (t task) command(cmd Command) status {
t.mutex.Lock()
defer t.mutex.Unlock()
t.cmdCh <- cmd
return <-t.statusCh
}
// close releases all the resources held by this task.
func (t task) close() {
t.mutex.Lock()
defer t.mutex.Unlock()
// Task can be ended in 2 ways.
// 1) Calling application invokes Handle.Close().
// 2) TaskCtl.Shutdown() ending the task's life.
// In either case, task.close() is invoked only via the
// TaskCtl. Handle.Close() only sends a message to the TaskCtl to
// initiate a close call.
close(t.cmdCh)
close(t.statusCh)
close(t.closeCh)
}

164
pkg/tasker/taskctl.go
Unescape View File

@ -1,164 +0,0 @@
/*
* Minio Cloud Storage, (C) 2015 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package tasker
import (
"container/list"
"sync"
)
// TaskCtl (Task Controller) is a framework to create and manage
// tasks.
type TaskCtl struct {
mutex *sync.Mutex // Lock
// List of tasks managed by this task controller.
tasks *list.List
}
// New creates a new TaskCtl to create and control a collection of tasks.
// Single application can create multiple task controllers to manage different set of tasks separately.
func New(name string) *TaskCtl {
return &TaskCtl{
mutex: &sync.Mutex{},
tasks: list.New(),
}
}
// NewTask creates a new task structure and returns a handle to it. Only the task controller
// has access to the task structure. The caller routine only receives a handle to its task structure.
// Task handle is like a reference to task self. Caller is expected to listen for commands from
// the task controller and comply with it co-operatively.
func (tc *TaskCtl) NewTask(name string) Handle {
tc.mutex.Lock()
defer tc.mutex.Unlock()
// Create a new task.
tsk := newTask(name)
// Register this task in the TaskCtl's tasklist and save the reference.
tsk.this = tc.tasks.PushBack(tsk)
// Free task from the tasklist upon close call.
go func() {
// Release the tasks resources upon return of this function.
defer tsk.close()
// Will be notified here upon task's end of life.
this := <-tsk.closeCh
tc.mutex.Lock()
defer tc.mutex.Unlock()
// Release the task structure from the task list.
tc.tasks.Remove(this)
}()
// Return a handle to this task.
return tsk.getHandle()
}
// Shutdown ends all tasks, including the suspended ones.
func (tc *TaskCtl) Shutdown() {
tc.mutex.Lock()
defer tc.mutex.Unlock()
var wg sync.WaitGroup
// End all tasks.
for e := tc.tasks.Front(); e != nil; e = e.Next() {
wg.Add(1)
thisTask := e.Value.(task) // Make a local copy for go routine.
// End tasks in background. Flow of events from here is as follows: thisTask.handle.Close() -> tc.NewTask() -> this.task.close().
go func() {
thisTask.getHandle().Close()
wg.Done()
}()
}
wg.Wait() // Wait for all tasks to end gracefully.
// Reset the task pool.
tc.tasks = nil
}
// Suspend puts all tasks to sleep.
func (tc *TaskCtl) Suspend() bool {
tc.mutex.Lock()
defer tc.mutex.Unlock()
var wg sync.WaitGroup
// If any one of the task fails to suspend, this flag will set to false.
statusAll := make([]status, tc.tasks.Len())
// Suspend all tasks.
i := 0
for e := tc.tasks.Front(); e != nil; e = e.Next() {
wg.Add(1)
locTask := e.Value.(task) // Make a local copy for go routine.
locI := i // local i
// Suspend a task in background.
go func(locI int) {
defer wg.Done()
statusAll[locI] = locTask.command(CmdSignalSuspend)
}(locI)
i++
}
wg.Wait() // Wait for all tasks to suspend gracefully.
for _, st := range statusAll {
if st.code != statusDone {
return false
}
}
return true
}
// Resume wakes up all suspended task from sleep.
func (tc *TaskCtl) Resume() bool {
tc.mutex.Lock()
defer tc.mutex.Unlock()
var wg sync.WaitGroup
// If any one of the task fails to suspend, this flag will set to false.
statusAll := make([]status, tc.tasks.Len())
i := 0
// Resume all suspended tasks.
for e := tc.tasks.Front(); e != nil; e = e.Next() {
wg.Add(1)
locTask := e.Value.(task) // Make a local copy for go routine.
locI := i // local i
// Resume a task in background.
go func(locI int) {
defer wg.Done()
statusAll[locI] = locTask.command(CmdSignalResume)
}(locI)
i++
}
wg.Wait() // Wait for all tasks to resume.
for _, st := range statusAll {
if st.code != statusDone {
return false
}
}
return true
}

38
pkg/tasker/taskctl_test.go
Unescape View File

@ -1,38 +0,0 @@
/*
* Quick - Quick key value store for config files and persistent state files
*
* Minio Client (C) 2015 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package tasker_test
import (
"testing"
"github.com/minio/minio/pkg/tasker"
. "gopkg.in/check.v1"
)
func Test(t *testing.T) { TestingT(t) }
type MySuite struct{}
var _ = Suite(&MySuite{})
func (s *MySuite) TestCheckData(c *C) {
testTasks := tasker.New("Test Task")
testTasks.Shutdown()
// c.Assert(err, Not(IsNil))
}
Write Preview
Loading…
Cancel
Save