Skip to content
This repository has been archived by the owner on Feb 1, 2021. It is now read-only.

Commit

Permalink
WIP: Parallel scheduling support for Swarm driver.
Browse files Browse the repository at this point in the history 
Signed-off-by: Andrea Luzzardi <[email protected]>
  • Loading branch information
@aluzzardi
aluzzardi committed Oct 6, 2015
1 parent e234503 commit 6b27eec
Show file tree
Hide file tree
Showing 2 changed files with 111 additions and 41 deletions.
73 changes: 56 additions & 17 deletions cluster/swarm/cluster.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,14 +20,30 @@ import (
"github.com/samalba/dockerclient"
)

type pendingContainer struct {
Config *cluster.ContainerConfig
Name string
Node *node.Node
}

func (p *pendingContainer) ToContainer() *cluster.Container {
return &cluster.Container{
Container: dockerclient.Container{
Names: []string{"/" + p.Name},
},
Config: p.Config,
}
}

// Cluster is exported
type Cluster struct {
sync.RWMutex

eventHandler cluster.EventHandler
engines map[string]*cluster.Engine
scheduler *scheduler.Scheduler
discovery discovery.Discovery
eventHandler cluster.EventHandler
engines map[string]*cluster.Engine
scheduler *scheduler.Scheduler
discovery discovery.Discovery
pendingContainers map[string]*pendingContainer

overcommitRatio float64
TLSConfig *tls.Config
Expand All @@ -38,11 +54,12 @@ func NewCluster(scheduler *scheduler.Scheduler, TLSConfig *tls.Config, discovery
log.WithFields(log.Fields{"name": "swarm"}).Debug("Initializing cluster")

cluster := &Cluster{
engines: make(map[string]*cluster.Engine),
scheduler: scheduler,
TLSConfig: TLSConfig,
discovery: discovery,
overcommitRatio: 0.05,
engines: make(map[string]*cluster.Engine),
scheduler: scheduler,
TLSConfig: TLSConfig,
discovery: discovery,
pendingContainers: make(map[string]*pendingContainer),
overcommitRatio: 0.05,
}

if val, ok := options.Float("swarm.overcommit", ""); ok {
Expand Down Expand Up @@ -102,15 +119,16 @@ func (c *Cluster) CreateContainer(config *cluster.ContainerConfig, name string)

func (c *Cluster) createContainer(config *cluster.ContainerConfig, name string, withSoftImageAffinity bool) (*cluster.Container, error) {
c.scheduler.Lock()
defer c.scheduler.Unlock()

// Ensure the name is available
if cID := c.getIDFromName(name); cID != "" {
c.scheduler.Unlock()
return nil, fmt.Errorf("Conflict, The name %s is already assigned to %s. You have to delete (or rename) that container to be able to assign %s to a container again.", name, cID, name)
}

// Associate a Swarm ID to the container we are creating.
config.SetSwarmID(c.generateUniqueID())
swarmID := c.generateUniqueID()
config.SetSwarmID(swarmID)

configTemp := config
if withSoftImageAffinity {
Expand All @@ -119,15 +137,30 @@ func (c *Cluster) createContainer(config *cluster.ContainerConfig, name string,

n, err := c.scheduler.SelectNodeForContainer(c.listNodes(), configTemp)
if err != nil {
c.scheduler.Unlock()
return nil, err
}
engine, ok := c.engines[n.ID]
if !ok {
c.scheduler.Unlock()
return nil, nil
}

if nn, ok := c.engines[n.ID]; ok {
container, err := nn.Create(config, name, true)
return container, err
c.pendingContainers[swarmID] = &pendingContainer{
Name: name,
Config: config,
Node: n,
}

return nil, nil
c.scheduler.Unlock()

container, err := engine.Create(config, name, true)

c.scheduler.Lock()
delete(c.pendingContainers, swarmID)
c.scheduler.Unlock()

return container, err
}

// RemoveContainer aka Remove a container from the cluster. Containers should
Expand Down Expand Up @@ -571,8 +604,14 @@ func (c *Cluster) listNodes() []*node.Node {
defer c.RUnlock()

out := make([]*node.Node, 0, len(c.engines))
for _, n := range c.engines {
out = append(out, node.NewNode(n))
for _, e := range c.engines {
node := node.NewNode(e)
for _, c := range c.pendingContainers {
if c.Node.ID == node.ID && node.Container(c.Config.SwarmID()) == nil {
node.AddContainer(c.ToContainer())
}
}
out = append(out, node)
}

return out
Expand Down
79 changes: 55 additions & 24 deletions scheduler/filter/dependency_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,24 +14,33 @@ func TestDependencyFilterSimple(t *testing.T) {
f = DependencyFilter{}
nodes = []*node.Node{
{
ID: "node-0-id",
Name: "node-0-name",
Addr: "node-0",
Containers: []*cluster.Container{{Container: dockerclient.Container{Id: "c0"}}},
ID: "node-0-id",
Name: "node-0-name",
Addr: "node-0",
Containers: []*cluster.Container{{
Container: dockerclient.Container{Id: "c0"},
Config: &cluster.ContainerConfig{},
}},
},

{
ID: "node-1-id",
Name: "node-1-name",
Addr: "node-1",
Containers: []*cluster.Container{{Container: dockerclient.Container{Id: "c1"}}},
ID: "node-1-id",
Name: "node-1-name",
Addr: "node-1",
Containers: []*cluster.Container{{
Container: dockerclient.Container{Id: "c1"},
Config: &cluster.ContainerConfig{},
}},
},

{
ID: "node-2-id",
Name: "node-2-name",
Addr: "node-2",
Containers: []*cluster.Container{{Container: dockerclient.Container{Id: "c2"}}},
ID: "node-2-id",
Name: "node-2-name",
Addr: "node-2",
Containers: []*cluster.Container{{
Container: dockerclient.Container{Id: "c2"},
Config: &cluster.ContainerConfig{},
}},
},
}
result []*node.Node
Expand Down Expand Up @@ -109,17 +118,28 @@ func TestDependencyFilterMulti(t *testing.T) {
Name: "node-0-name",
Addr: "node-0",
Containers: []*cluster.Container{
{Container: dockerclient.Container{Id: "c0"}},
{Container: dockerclient.Container{Id: "c1"}},
{
Container: dockerclient.Container{Id: "c0"},
Config: &cluster.ContainerConfig{},
},
{
Container: dockerclient.Container{Id: "c1"},
Config: &cluster.ContainerConfig{},
},
},
},

// nodes[1] has c2
{
ID: "node-1-id",
Name: "node-1-name",
Addr: "node-1",
Containers: []*cluster.Container{{Container: dockerclient.Container{Id: "c2"}}},
ID: "node-1-id",
Name: "node-1-name",
Addr: "node-1",
Containers: []*cluster.Container{
{
Container: dockerclient.Container{Id: "c2"},
Config: &cluster.ContainerConfig{},
},
},
},

// nodes[2] has nothing
Expand Down Expand Up @@ -179,17 +199,28 @@ func TestDependencyFilterChaining(t *testing.T) {
Name: "node-0-name",
Addr: "node-0",
Containers: []*cluster.Container{
{Container: dockerclient.Container{Id: "c0"}},
{Container: dockerclient.Container{Id: "c1"}},
{
Container: dockerclient.Container{Id: "c0"},
Config: &cluster.ContainerConfig{},
},
{
Container: dockerclient.Container{Id: "c1"},
Config: &cluster.ContainerConfig{},
},
},
},

// nodes[1] has c2
{
ID: "node-1-id",
Name: "node-1-name",
Addr: "node-1",
Containers: []*cluster.Container{{Container: dockerclient.Container{Id: "c2"}}},
ID: "node-1-id",
Name: "node-1-name",
Addr: "node-1",
Containers: []*cluster.Container{
{
Container: dockerclient.Container{Id: "c2"},
Config: &cluster.ContainerConfig{},
},
},
},

// nodes[2] has nothing
Expand Down

0 comments on commit 6b27eec

Please sign in to comment.