forked from pingcap/tiflow
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[resource manager]: add separated resouce manager interface (pingcap#44)
* [resource manager]: add separated resouce manager interface * fix merge conflicts * address comments * fix data race
- Loading branch information
Showing
9 changed files
with
171 additions
and
128 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,108 @@ | ||
| package resource | ||
|
|
||
| import ( | ||
| "sync" | ||
|
|
||
| "github.com/hanfei1991/microcosm/model" | ||
| "github.com/hanfei1991/microcosm/pb" | ||
| "github.com/hanfei1991/microcosm/pkg/errors" | ||
| "github.com/pingcap/log" | ||
| "go.uber.org/zap" | ||
| ) | ||
|
|
||
| // CapRescMgr implements ResourceMgr interface, and it uses node capacity as | ||
| // alloction algorithm | ||
| type CapRescMgr struct { | ||
| mu sync.Mutex | ||
| executors map[model.ExecutorID]*ExecutorResource | ||
| } | ||
|
|
||
| func NewCapRescMgr() *CapRescMgr { | ||
| return &CapRescMgr{ | ||
| executors: make(map[model.ExecutorID]*ExecutorResource), | ||
| } | ||
| } | ||
|
|
||
| // Register implements RescMgr.Register | ||
| func (m *CapRescMgr) Register(id model.ExecutorID, capacity RescUnit) { | ||
| m.mu.Lock() | ||
| defer m.mu.Unlock() | ||
| m.executors[id] = &ExecutorResource{ | ||
| ID: id, | ||
| Capacity: capacity, | ||
| } | ||
| log.L().Info("executor resource is registered", | ||
| zap.String("executor-id", string(id)), zap.Int("capacity", int(capacity))) | ||
| } | ||
|
|
||
| // Unregister implements RescMgr.Unregister | ||
| func (m *CapRescMgr) Unregister(id model.ExecutorID) { | ||
| m.mu.Lock() | ||
| defer m.mu.Unlock() | ||
| delete(m.executors, id) | ||
| log.L().Info("executor resource is unregistered", | ||
| zap.String("executor-id", string(id))) | ||
| } | ||
|
|
||
| // Allocate implements RescMgr.Allocate | ||
| func (m *CapRescMgr) Allocate(tasks []*pb.ScheduleTask) (bool, *pb.TaskSchedulerResponse) { | ||
| return m.allocateTasksWithNaiveStrategy(tasks) | ||
| } | ||
|
|
||
| // Update implements RescMgr.Update | ||
| func (m *CapRescMgr) Update(id model.ExecutorID, use RescUnit, status model.ExecutorStatus) error { | ||
| m.mu.Lock() | ||
| defer m.mu.Unlock() | ||
| exec, ok := m.executors[id] | ||
| if !ok { | ||
| return errors.ErrUnknownExecutorID.GenWithStackByArgs(id) | ||
| } | ||
| exec.Used = use | ||
| exec.Status = status | ||
| return nil | ||
| } | ||
|
|
||
| // getAvailableResource returns resources that are available | ||
| func (m *CapRescMgr) getAvailableResource() []*ExecutorResource { | ||
| res := make([]*ExecutorResource, 0) | ||
| for _, exec := range m.executors { | ||
| if exec.Status == model.Running && | ||
| exec.Capacity > exec.Reserved && exec.Capacity > exec.Used { | ||
| res = append(res, exec) | ||
| } | ||
| } | ||
| return res | ||
| } | ||
|
|
||
| func (m *CapRescMgr) allocateTasksWithNaiveStrategy( | ||
| tasks []*pb.ScheduleTask, | ||
| ) (bool, *pb.TaskSchedulerResponse) { | ||
| m.mu.Lock() | ||
| defer m.mu.Unlock() | ||
| result := make(map[int32]*pb.ScheduleResult) | ||
| resources := m.getAvailableResource() | ||
| var idx int = 0 | ||
| for _, task := range tasks { | ||
| originalIdx := idx | ||
| for { | ||
| exec := resources[idx] | ||
| used := exec.Used | ||
| if exec.Reserved > used { | ||
| used = exec.Reserved | ||
| } | ||
| rest := exec.Capacity - used | ||
| if rest >= RescUnit(task.Cost) { | ||
| result[task.GetTask().Id] = &pb.ScheduleResult{ | ||
| ExecutorId: string(exec.ID), | ||
| } | ||
| exec.Reserved = exec.Reserved + RescUnit(task.GetCost()) | ||
| break | ||
| } | ||
| idx = (idx + 1) % len(resources) | ||
| if idx == originalIdx { | ||
| return false, nil | ||
| } | ||
| } | ||
| } | ||
| return true, &pb.TaskSchedulerResponse{Schedule: result} | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| package resource | ||
|
|
||
| import ( | ||
| "github.com/hanfei1991/microcosm/model" | ||
| "github.com/hanfei1991/microcosm/pb" | ||
| ) | ||
|
|
||
| // RescMgr manages the resources of the clusters. | ||
| type RescMgr interface { | ||
| // Register registers new executor, it is called when an executor joins | ||
| Register(id model.ExecutorID, capacity RescUnit) | ||
|
|
||
| // Unregister is called when an executor exits | ||
| Unregister(id model.ExecutorID) | ||
|
|
||
| // Allocate allocates executor resources to given tasks | ||
| Allocate(tasks []*pb.ScheduleTask) (bool, *pb.TaskSchedulerResponse) | ||
|
|
||
| // Update updates executor resource usage and running status | ||
| Update(id model.ExecutorID, use RescUnit, status model.ExecutorStatus) error | ||
| } | ||
|
|
||
| // RescUnit is the min unit of resource that we count. | ||
| type RescUnit int | ||
|
|
||
| type ExecutorResource struct { | ||
| ID model.ExecutorID | ||
| Status model.ExecutorStatus | ||
|
|
||
| Capacity RescUnit | ||
| Reserved RescUnit | ||
| Used RescUnit | ||
| } |
Oops, something went wrong.