// GetWork requests one or more work units to perform.  The work unit
// attempts are associated with workerID, which need not have been
// previously registered.  If there is no work to do, may return
// neither work nor an error.
//
// Each work unit is returned as a cborrpc.PythonTuple holding the
// work spec name, work unit key as a byte slice, and work unit data
// dictionary.  If options does not contain "max_jobs" or if that
// value is 1, returns a tuple or nil, otherwise returns a slice of
// tuples (maybe 1 or none).
func (jobs *JobServer) GetWork(workerID string, options map[string]interface{}) (interface{}, string, error) {
	// This is the Big Kahuna.  The Python Coordinate server tries
	// to be extra clever with its return value, returning None if
	// there is no work, a concrete value if one work unit was
	// requested, and a list if more than one was requested, and
	// this same rule is enforced in the client code.  So, this will
	// return either exactly one PythonTuple or a list of PythonTuple.
	var (
		attempts  []coordinate.Attempt
		err       error
		gwOptions GetWorkOptions
		worker    coordinate.Worker
	)
	err = decode(&gwOptions, options)
	if err == nil {
		worker, err = jobs.Namespace.Worker(workerID)
	}
	if err == nil {
		if gwOptions.MaxJobs < 1 {
			gwOptions.MaxJobs = 1
		}
		req := coordinate.AttemptRequest{
			NumberOfWorkUnits: gwOptions.MaxJobs,
		}
		attempts, err = worker.RequestAttempts(req)
	}
	if err != nil {
		return nil, "", err
	}
	// successful return
	if gwOptions.MaxJobs == 1 {
		if len(attempts) == 0 {
			tuple := cborrpc.PythonTuple{
				Items: []interface{}{nil, nil, nil},
			}
			return tuple, "", nil
		}
		if len(attempts) == 1 {
			tuple, err := getWorkTuple(attempts[0])
			if err != nil {
				return nil, "", err
			}
			return tuple, "", nil
		}
	}
	result := make([]cborrpc.PythonTuple, len(attempts))
	for i, attempt := range attempts {
		tuple, err := getWorkTuple(attempt)
		if err != nil {
			return nil, "", err
		}
		result[i] = tuple
	}
	return result, "", nil
}

// TestChainingExpiry tests that, if an attempt finishes but is no
// longer the active attempt, then its successor work units will not
// be created.
func (s *Suite) TestChainingExpiry(c *check.C) {
	var (
		one, two coordinate.WorkSpec
		err      error
		worker   coordinate.Worker
		unit     coordinate.WorkUnit
		attempts []coordinate.Attempt
	)

	one, err = s.Namespace.SetWorkSpec(map[string]interface{}{
		"name": "one",
		"then": "two",
	})
	c.Assert(err, check.IsNil)

	two, err = s.Namespace.SetWorkSpec(map[string]interface{}{
		"name":     "two",
		"disabled": true,
	})
	c.Assert(err, check.IsNil)

	worker, err = s.Namespace.Worker("worker")
	c.Assert(err, check.IsNil)

	// Create and perform a work unit, with no output
	unit, err = one.AddWorkUnit("a", map[string]interface{}{}, 0.0)
	c.Assert(err, check.IsNil)

	attempts, err = worker.RequestAttempts(coordinate.AttemptRequest{})
	c.Assert(err, check.IsNil)
	c.Assert(attempts, check.HasLen, 1)
	attempt := attempts[0]

	// But wait!  We got preempted
	err = unit.ClearActiveAttempt()
	c.Assert(err, check.IsNil)
	attempts, err = worker.RequestAttempts(coordinate.AttemptRequest{})
	c.Assert(err, check.IsNil)
	c.Assert(attempts, check.HasLen, 1)

	// Now, let the original attempt finish, trying to generate
	// more outputs
	err = attempt.Finish(map[string]interface{}{
		"output": []string{"unit"},
	})
	c.Assert(err, check.IsNil)

	// Since attempt is no longer active, this shouldn't generate
	// new outputs
	units, err := two.WorkUnits(coordinate.WorkUnitQuery{})
	c.Assert(err, check.IsNil)
	c.Check(units, check.HasLen, 0)
}

// makeWorkUnits creates a handful of work units within a work spec.
// These have keys "available", "pending", "finished", "failed",
// "expired", and "retryable", and wind up in the corresponding
// states.
func makeWorkUnits(spec coordinate.WorkSpec, worker coordinate.Worker) (map[string]coordinate.WorkUnit, error) {
	result := map[string]coordinate.WorkUnit{
		"available": nil,
		"pending":   nil,
		"finished":  nil,
		"failed":    nil,
		"expired":   nil,
		"retryable": nil,
	}
	for key := range result {
		unit, err := spec.AddWorkUnit(key, map[string]interface{}{}, 0)
		if err != nil {
			return nil, err
		}
		result[key] = unit

		// Run the workflow
		if key == "available" {
			continue
		}
		attempt, err := worker.MakeAttempt(unit, time.Duration(0))
		if err != nil {
			return nil, err
		}
		switch key {
		case "pending":
			{
			} // leave it running
		case "finished":
			err = attempt.Finish(nil)
		case "failed":
			err = attempt.Fail(nil)
		case "expired":
			err = attempt.Expire(nil)
		case "retryable":
			err = attempt.Retry(nil)
		}
		if err != nil {
			return nil, err
		}
	}
	return result, nil
}

// checkWorkUnitOrder verifies that getting all of the work possible
// retrieves work units in a specific order.
func checkWorkUnitOrder(
	c *check.C,
	worker coordinate.Worker,
	spec coordinate.WorkSpec,
	unitNames ...string,
) {
	var processedUnits []string
	for {
		attempts, err := worker.RequestAttempts(coordinate.AttemptRequest{})
		c.Assert(err, check.IsNil)
		if len(attempts) == 0 {
			break
		}
		c.Assert(attempts, check.HasLen, 1)
		attempt := attempts[0]
		c.Check(attempt.WorkUnit().WorkSpec().Name(), check.Equals, spec.Name())
		processedUnits = append(processedUnits, attempt.WorkUnit().Name())
		err = attempt.Finish(nil)
		c.Assert(err, check.IsNil)
	}

	c.Check(processedUnits, check.DeepEquals, unitNames)
}

// TestWorkerAncestry does basic tests on worker parents and children.
func (s *Suite) TestWorkerAncestry(c *check.C) {
	var (
		err                   error
		parent, child, worker coordinate.Worker
		kids                  []coordinate.Worker
	)

	// start in the middle
	parent, err = s.Namespace.Worker("parent")
	c.Assert(err, check.IsNil)

	worker, err = parent.Parent()
	c.Assert(err, check.IsNil)
	c.Check(worker, check.IsNil)
	kids, err = parent.Children()
	c.Assert(err, check.IsNil)
	c.Check(kids, check.HasLen, 0)

	// Create a child
	child, err = s.Namespace.Worker("child")
	c.Assert(err, check.IsNil)
	err = child.SetParent(parent)
	c.Assert(err, check.IsNil)

	// this should update the parent metadata
	worker, err = parent.Parent()
	c.Assert(err, check.IsNil)
	c.Check(worker, check.IsNil)
	kids, err = parent.Children()
	c.Assert(err, check.IsNil)
	c.Check(kids, check.HasLen, 1)
	if len(kids) > 0 {
		c.Check(kids[0].Name(), check.Equals, "child")
	}

	// and also the child metadata
	worker, err = child.Parent()
	c.Assert(err, check.IsNil)
	c.Check(worker, check.NotNil)
	if worker != nil {
		c.Check(worker.Name(), check.Equals, "parent")
	}
	kids, err = child.Children()
	c.Assert(err, check.IsNil)
	c.Check(kids, check.HasLen, 0)
}

// TestWorkerAdoption hands a child worker to a new parent.
func (s *Suite) TestWorkerAdoption(c *check.C) {
	var (
		err                                 error
		child, oldParent, newParent, worker coordinate.Worker
		kids                                []coordinate.Worker
	)

	// Create the worker objects
	child, err = s.Namespace.Worker("child")
	c.Assert(err, check.IsNil)
	oldParent, err = s.Namespace.Worker("old")
	c.Assert(err, check.IsNil)
	newParent, err = s.Namespace.Worker("new")
	c.Assert(err, check.IsNil)

	// Set up the original ancestry
	err = child.SetParent(oldParent)
	c.Assert(err, check.IsNil)

	// Move it to the new parent
	err = child.SetParent(newParent)
	c.Assert(err, check.IsNil)

	// Checks
	worker, err = child.Parent()
	c.Assert(err, check.IsNil)
	c.Check(worker, check.NotNil)
	if worker != nil {
		c.Check(worker.Name(), check.Equals, "new")
	}
	kids, err = child.Children()
	c.Assert(err, check.IsNil)
	c.Check(kids, check.HasLen, 0)

	worker, err = oldParent.Parent()
	c.Assert(err, check.IsNil)
	c.Check(worker, check.IsNil)
	kids, err = oldParent.Children()
	c.Assert(err, check.IsNil)
	c.Check(kids, check.HasLen, 0)

	worker, err = newParent.Parent()
	c.Assert(err, check.IsNil)
	c.Check(worker, check.IsNil)
	kids, err = newParent.Children()
	c.Assert(err, check.IsNil)
	c.Check(kids, check.HasLen, 1)
	if len(kids) > 0 {
		c.Check(kids[0].Name(), check.Equals, "child")
	}
}

// TestChainingTwoStep separately renews an attempt to insert an output
// key, then finishes the work unit; it should still chain.
func (s *Suite) TestChainingTwoStep(c *check.C) {
	var (
		one, two coordinate.WorkSpec
		worker   coordinate.Worker
		attempts []coordinate.Attempt
		units    map[string]coordinate.WorkUnit
		unit     coordinate.WorkUnit
		data     map[string]interface{}
		priority float64
		ok       bool
		err      error
	)

	one, err = s.Namespace.SetWorkSpec(map[string]interface{}{
		"name": "one",
		"then": "two",
	})
	c.Assert(err, check.IsNil)

	two, err = s.Namespace.SetWorkSpec(map[string]interface{}{
		"name": "two",
	})
	c.Assert(err, check.IsNil)

	worker, err = s.Namespace.Worker("worker")
	c.Assert(err, check.IsNil)

	_, err = one.AddWorkUnit("a", map[string]interface{}{}, 0.0)
	c.Assert(err, check.IsNil)

	attempts, err = worker.RequestAttempts(coordinate.AttemptRequest{})
	c.Assert(err, check.IsNil)
	c.Assert(attempts, check.HasLen, 1)

	err = attempts[0].Renew(time.Duration(900)*time.Second,
		map[string]interface{}{
			"output": []interface{}{
				[]byte{1, 2, 3, 4},
				cborrpc.PythonTuple{Items: []interface{}{
					[]byte{1, 2, 3, 4},
					map[interface{}]interface{}{},
					map[interface{}]interface{}{
						"priority": 0,
					},
				}},
			},
		})
	c.Assert(err, check.IsNil)

	err = attempts[0].Finish(nil)

	units, err = two.WorkUnits(coordinate.WorkUnitQuery{})
	c.Assert(err, check.IsNil)
	c.Check(units, HasKeys, []string{"\x01\x02\x03\x04"})
	if unit, ok = units["\x01\x02\x03\x04"]; ok {
		data, err = unit.Data()
		c.Assert(err, check.IsNil)
		c.Check(data, check.DeepEquals, map[string]interface{}{})

		priority, err = unit.Priority()
		c.Assert(err, check.IsNil)
		c.Check(priority, check.Equals, 0.0)
	}
}

// TestChainingMixed uses a combination of strings and tuples in its
// "output" data.
func (s *Suite) TestChainingMixed(c *check.C) {
	var (
		one, two coordinate.WorkSpec
		worker   coordinate.Worker
		attempts []coordinate.Attempt
		units    map[string]coordinate.WorkUnit
		unit     coordinate.WorkUnit
		data     map[string]interface{}
		priority float64
		ok       bool
		err      error
	)

	one, err = s.Namespace.SetWorkSpec(map[string]interface{}{
		"name": "one",
		"then": "two",
	})
	c.Assert(err, check.IsNil)

	two, err = s.Namespace.SetWorkSpec(map[string]interface{}{
		"name": "two",
	})
	c.Assert(err, check.IsNil)

	worker, err = s.Namespace.Worker("worker")
	c.Assert(err, check.IsNil)

	_, err = one.AddWorkUnit("a", map[string]interface{}{}, 0.0)
	c.Assert(err, check.IsNil)

	attempts, err = worker.RequestAttempts(coordinate.AttemptRequest{})
	c.Assert(err, check.IsNil)
	c.Assert(attempts, check.HasLen, 1)

	err = attempts[0].Finish(map[string]interface{}{
		"output": []interface{}{
			"key",
			cborrpc.PythonTuple{Items: []interface{}{
				"key",
				map[string]interface{}{
					"data": "x",
				},
				map[string]interface{}{
					"priority": 10.0,
				},
			}},
		},
	})
	c.Assert(err, check.IsNil)

	units, err = two.WorkUnits(coordinate.WorkUnitQuery{})
	c.Assert(err, check.IsNil)
	c.Check(units, HasKeys, []string{"key"})
	if unit, ok = units["key"]; ok {
		data, err = unit.Data()
		c.Assert(err, check.IsNil)
		c.Check(data, check.DeepEquals, map[string]interface{}{"data": "x"})

		priority, err = unit.Priority()
		c.Assert(err, check.IsNil)
		c.Check(priority, check.Equals, 10.0)
	}
}

// TestWorkUnitChaining tests that completing work units in one work spec
// will cause work units to appear in another, if so configured.
func (s *Suite) TestWorkUnitChaining(c *check.C) {
	var (
		err      error
		worker   coordinate.Worker
		one, two coordinate.WorkSpec
		units    map[string]coordinate.WorkUnit
		attempts []coordinate.Attempt
		data     map[string]interface{}
		ok       bool
	)

	one, err = s.Namespace.SetWorkSpec(map[string]interface{}{
		"name": "one",
		"then": "two",
	})
	c.Assert(err, check.IsNil)

	two, err = s.Namespace.SetWorkSpec(map[string]interface{}{
		"name":     "two",
		"disabled": true,
	})
	c.Assert(err, check.IsNil)

	worker, err = s.Namespace.Worker("worker")
	c.Assert(err, check.IsNil)

	// Create and perform a work unit, with no output
	_, err = one.AddWorkUnit("a", map[string]interface{}{}, 0.0)
	c.Assert(err, check.IsNil)

	attempts, err = worker.RequestAttempts(coordinate.AttemptRequest{})
	c.Assert(err, check.IsNil)
	c.Assert(attempts, check.HasLen, 1)

	err = attempts[0].Finish(nil)
	c.Assert(err, check.IsNil)

	units, err = two.WorkUnits(coordinate.WorkUnitQuery{})
	c.Assert(err, check.IsNil)
	c.Check(units, HasKeys, []string{})

	// Create and perform a work unit, with a map output
	_, err = one.AddWorkUnit("b", map[string]interface{}{}, 0.0)
	c.Assert(err, check.IsNil)

	attempts, err = worker.RequestAttempts(coordinate.AttemptRequest{})
	c.Assert(err, check.IsNil)
	c.Assert(attempts, check.HasLen, 1)

	err = attempts[0].Finish(map[string]interface{}{
		"output": map[string]interface{}{
			"two_b": map[string]interface{}{"k": "v"},
		},
	})
	c.Assert(err, check.IsNil)

	units, err = two.WorkUnits(coordinate.WorkUnitQuery{})
	c.Assert(err, check.IsNil)
	c.Check(units, HasKeys, []string{"two_b"})

	if _, ok = units["two_b"]; ok {
		data, err = units["two_b"].Data()
		c.Assert(err, check.IsNil)
		c.Check(data, check.DeepEquals, map[string]interface{}{"k": "v"})
	}

	// Create and perform a work unit, with a slice output
	_, err = one.AddWorkUnit("c", map[string]interface{}{}, 0.0)
	c.Assert(err, check.IsNil)

	attempts, err = worker.RequestAttempts(coordinate.AttemptRequest{})
	c.Assert(err, check.IsNil)
	c.Assert(attempts, check.HasLen, 1)

	err = attempts[0].Finish(map[string]interface{}{
		"output": []string{"two_c", "two_cc"},
	})
	c.Assert(err, check.IsNil)

	units, err = two.WorkUnits(coordinate.WorkUnitQuery{})
	c.Assert(err, check.IsNil)
	c.Check(units, HasKeys, []string{"two_b", "two_c", "two_cc"})

	if _, ok = units["two_c"]; ok {
		data, err = units["two_c"].Data()
		c.Assert(err, check.IsNil)
		c.Check(data, check.DeepEquals, map[string]interface{}{})
	}

	// Put the output in the original work unit data
	_, err = one.AddWorkUnit("d", map[string]interface{}{
		"output": []string{"two_d"},
	}, 0.0)
	c.Assert(err, check.IsNil)
	attempts, err = worker.RequestAttempts(coordinate.AttemptRequest{})
	c.Assert(err, check.IsNil)
	c.Assert(attempts, check.HasLen, 1)

//.........这里部分代码省略.........

// TestAttemptLifetime validates a basic attempt lifetime.
func (s *Suite) TestAttemptLifetime(c *check.C) {
	var (
		err               error
		data              map[string]interface{}
		attempt, attempt2 coordinate.Attempt
		aStatus           coordinate.AttemptStatus
		spec              coordinate.WorkSpec
		unit              coordinate.WorkUnit
		worker            coordinate.Worker
		uStatus           coordinate.WorkUnitStatus
	)
	spec, worker = s.makeWorkSpecAndWorker(c)

	// Create a work unit
	unit, err = spec.AddWorkUnit("a", map[string]interface{}{}, 0.0)
	c.Assert(err, check.IsNil)

	// The work unit should be "available"
	uStatus, err = unit.Status()
	c.Assert(err, check.IsNil)
	c.Check(uStatus, check.Equals, coordinate.AvailableUnit)

	// The work unit data should be defined but empty
	data, err = unit.Data()
	c.Assert(err, check.IsNil)
	c.Check(data, check.HasLen, 0)

	// Get an attempt for it
	attempts, err := worker.RequestAttempts(coordinate.AttemptRequest{})
	c.Assert(err, check.IsNil)
	c.Assert(attempts, check.HasLen, 1)
	attempt = attempts[0]

	// The work unit should be "pending"
	uStatus, err = unit.Status()
	c.Assert(err, check.IsNil)
	c.Check(uStatus, check.Equals, coordinate.PendingUnit)

	// The attempt should be "pending" too
	aStatus, err = attempt.Status()
	c.Assert(err, check.IsNil)
	c.Check(aStatus, check.Equals, coordinate.Pending)

	// The active attempt for the unit should match this
	attempt2, err = unit.ActiveAttempt()
	c.Assert(err, check.IsNil)
	c.Check(attempt2, AttemptMatches, attempt)

	// There should be one active attempt for the worker and it should
	// also match
	attempts, err = worker.ActiveAttempts()
	c.Assert(err, check.IsNil)
	c.Check(attempts, check.HasLen, 1)
	if len(attempts) > 0 {
		c.Check(attempts[0], AttemptMatches, attempt)
	}

	// The work unit data should (still) be defined but empty
	data, err = unit.Data()
	c.Assert(err, check.IsNil)
	c.Check(data, check.HasLen, 0)

	// Now finish the attempt with some updated data
	err = attempt.Finish(map[string]interface{}{
		"outputs": []string{"yes"},
	})
	c.Assert(err, check.IsNil)

	// The unit should report "finished"
	uStatus, err = unit.Status()
	c.Assert(err, check.IsNil)
	c.Check(uStatus, check.Equals, coordinate.FinishedUnit)

	// The attempt should report "finished"
	aStatus, err = attempt.Status()
	c.Assert(err, check.IsNil)
	c.Check(aStatus, check.Equals, coordinate.Finished)

	// The attempt should still be the active attempt for the unit
	attempt2, err = unit.ActiveAttempt()
	c.Assert(err, check.IsNil)
	c.Check(attempt2, AttemptMatches, attempt)

	// The attempt should not be in the active attempt list for the worker
	attempts, err = worker.ActiveAttempts()
	c.Assert(err, check.IsNil)
	c.Check(attempts, check.HasLen, 0)

	// Both the unit and the worker should have one archived attempt
	attempts, err = unit.Attempts()
	c.Assert(err, check.IsNil)
	c.Check(attempts, check.HasLen, 1)
	if len(attempts) > 0 {
		c.Check(attempts[0], AttemptMatches, attempt)
	}

	attempts, err = worker.AllAttempts()
	c.Assert(err, check.IsNil)
	c.Check(attempts, check.HasLen, 1)
//.........这里部分代码省略.........