Merge pull request #32 from runcom/inmemory-store

use an in memory store for containers

oci/history.go

@@ -0,0 +1,31 @@
+package oci
+
+import "sort"
+
+// History is a convenience type for storing a list of containers,
+// sorted by creation date in descendant order.
+type History []*Container
+
+// Len returns the number of containers in the history.
+func (history *History) Len() int {
+	return len(*history)
+}
+
+// Less compares two containers and returns true if the second one
+// was created before the first one.
+func (history *History) Less(i, j int) bool {
+	containers := *history
+	// FIXME: state access should be serialized
+	return containers[j].state.Created.Before(containers[i].state.Created)
+}
+
+// Swap switches containers i and j positions in the history.
+func (history *History) Swap(i, j int) {
+	containers := *history
+	containers[i], containers[j] = containers[j], containers[i]
+}
+
+// sort orders the history by creation date in descendant order.
+func (history *History) sort() {
+	sort.Sort(history)
+}

oci/memory_store.go

@@ -0,0 +1,92 @@
+package oci
+
+import "sync"
+
+// memoryStore implements a Store in memory.
+type memoryStore struct {
+	s map[string]*Container
+	sync.RWMutex
+}
+
+// NewMemoryStore initializes a new memory store.
+func NewMemoryStore() Store {
+	return &memoryStore{
+		s: make(map[string]*Container),
+	}
+}
+
+// Add appends a new container to the memory store.
+// It overrides the id if it existed before.
+func (c *memoryStore) Add(id string, cont *Container) {
+	c.Lock()
+	c.s[id] = cont
+	c.Unlock()
+}
+
+// Get returns a container from the store by id.
+func (c *memoryStore) Get(id string) *Container {
+	c.RLock()
+	res := c.s[id]
+	c.RUnlock()
+	return res
+}
+
+// Delete removes a container from the store by id.
+func (c *memoryStore) Delete(id string) {
+	c.Lock()
+	delete(c.s, id)
+	c.Unlock()
+}
+
+// List returns a sorted list of containers from the store.
+// The containers are ordered by creation date.
+func (c *memoryStore) List() []*Container {
+	containers := History(c.all())
+	containers.sort()
+	return containers
+}
+
+// Size returns the number of containers in the store.
+func (c *memoryStore) Size() int {
+	c.RLock()
+	defer c.RUnlock()
+	return len(c.s)
+}
+
+// First returns the first container found in the store by a given filter.
+func (c *memoryStore) First(filter StoreFilter) *Container {
+	for _, cont := range c.all() {
+		if filter(cont) {
+			return cont
+		}
+	}
+	return nil
+}
+
+// ApplyAll calls the reducer function with every container in the store.
+// This operation is asyncronous in the memory store.
+// NOTE: Modifications to the store MUST NOT be done by the StoreReducer.
+func (c *memoryStore) ApplyAll(apply StoreReducer) {
+	wg := new(sync.WaitGroup)
+	for _, cont := range c.all() {
+		wg.Add(1)
+		go func(container *Container) {
+			apply(container)
+			wg.Done()
+		}(cont)
+	}
+
+	wg.Wait()
+}
+
+func (c *memoryStore) all() []*Container {
+	c.RLock()
+	containers := make([]*Container, 0, len(c.s))
+	for _, cont := range c.s {
+		containers = append(containers, cont)
+	}
+	c.RUnlock()
+	return containers
+}
+
+var _ Store = &memoryStore{}

oci/store.go

@@ -0,0 +1,28 @@
+package oci
+
+// StoreFilter defines a function to filter
+// container in the store.
+type StoreFilter func(*Container) bool
+
+// StoreReducer defines a function to
+// manipulate containers in the store
+type StoreReducer func(*Container)
+
+// Store defines an interface that
+// any container store must implement.
+type Store interface {
+	// Add appends a new container to the store.
+	Add(string, *Container)
+	// Get returns a container from the store by the identifier it was stored with.
+	Get(string) *Container
+	// Delete removes a container from the store by the identifier it was stored with.
+	Delete(string)
+	// List returns a list of containers from the store.
+	List() []*Container
+	// Size returns the number of containers in the store.
+	Size() int
+	// First returns the first container found in the store by a given filter.
+	First(StoreFilter) *Container
+	// ApplyAll calls the reducer function with every container in the store.
+	ApplyAll(StoreReducer)
+}

server/runtime.go

@@ -107,7 +107,7 @@ func (s *Server) CreatePodSandbox(ctx context.Context, req *pb.CreatePodSandboxR
 		name:       name,
 		logDir:     logDir,
 		labels:     labels,
-		containers: make(map[string]*oci.Container),
+		containers: oci.NewMemoryStore(),
 	})
 
 	annotations := req.GetConfig().GetAnnotations()
@@ -197,26 +197,22 @@ func (s *Server) StopPodSandbox(ctx context.Context, req *pb.StopPodSandboxReque
 	}
 
 	podInfraContainer := *sbName + "-infra"
-	sb.containersLock.Lock()
-	for _, c := range sb.containers {
+	containersList := sb.containers.List()
+	for _, c := range containersList {
 		if podInfraContainer == c.Name() {
 			podNamespace := ""
 			netnsPath, err := c.NetNsPath()
 			if err != nil {
-				sb.containersLock.Unlock()
 				return nil, err
 			}
 			if err := s.netPlugin.TearDownPod(netnsPath, podNamespace, *sbName, podInfraContainer); err != nil {
-				sb.containersLock.Unlock()
 				return nil, fmt.Errorf("failed to destroy network for container %s in sandbox %s: %v", c.Name(), *sbName, err)
 			}
 		}
 		if err := s.runtime.StopContainer(c); err != nil {
-			sb.containersLock.Unlock()
 			return nil, fmt.Errorf("failed to stop container %s in sandbox %s: %v", c.Name(), *sbName, err)
 		}
 	}
-	sb.containersLock.Unlock()
 
 	return &pb.StopPodSandboxResponse{}, nil
 }
@@ -236,23 +232,19 @@ func (s *Server) RemovePodSandbox(ctx context.Context, req *pb.RemovePodSandboxR
 	podInfraContainer := *sbName + "-infra"
 
 	// Delete all the containers in the sandbox
-	sb.containersLock.Lock()
-	for _, c := range sb.containers {
+	containersList := sb.containers.List()
+	for _, c := range containersList {
 		if err := s.runtime.DeleteContainer(c); err != nil {
-			sb.containersLock.Unlock()
 			return nil, fmt.Errorf("failed to delete container %s in sandbox %s: %v", c.Name(), *sbName, err)
 		}
 		if podInfraContainer == c.Name() {
-			sb.containersLock.Unlock()
 			continue
 		}
 		containerDir := filepath.Join(s.runtime.ContainerDir(), c.Name())
 		if err := os.RemoveAll(containerDir); err != nil {
-			sb.containersLock.Unlock()
 			return nil, fmt.Errorf("failed to remove container %s directory: %v", c.Name(), err)
 		}
 	}
-	sb.containersLock.Unlock()
 
 	// Remove the files related to the sandbox
 	podSandboxDir := filepath.Join(s.sandboxDir, *sbName)
@@ -539,7 +531,7 @@ func (s *Server) CreateContainer(ctx context.Context, req *pb.CreateContainerReq
 
 	// Join the namespace paths for the pod sandbox container.
 	podContainerName := podSandboxId + "-infra"
-	podInfraContainer := s.state.containers[podContainerName]
+	podInfraContainer := s.state.containers.Get(podContainerName)
 	podInfraState := s.runtime.ContainerStatus(podInfraContainer)
 
 	logrus.Infof("pod container state %v", podInfraState)
@@ -590,7 +582,7 @@ func (s *Server) StartContainer(ctx context.Context, req *pb.StartContainerReque
 	if *containerName == "" {
 		return nil, fmt.Errorf("container ID should not be empty")
 	}
-	c := s.state.containers[*containerName]
+	c := s.state.containers.Get(*containerName)
 	if c == nil {
 		return nil, fmt.Errorf("specified container not found: %s", *containerName)
 	}
@@ -609,7 +601,7 @@ func (s *Server) StopContainer(ctx context.Context, req *pb.StopContainerRequest
 	if *containerName == "" {
 		return nil, fmt.Errorf("container ID should not be empty")
 	}
-	c := s.state.containers[*containerName]
+	c := s.state.containers.Get(*containerName)
 	if c == nil {
 		return nil, fmt.Errorf("specified container not found: %s", *containerName)
 	}
@@ -629,7 +621,7 @@ func (s *Server) RemoveContainer(ctx context.Context, req *pb.RemoveContainerReq
 	if *containerName == "" {
 		return nil, fmt.Errorf("container ID should not be empty")
 	}
-	c := s.state.containers[*containerName]
+	c := s.state.containers.Get(*containerName)
 	if c == nil {
 		return nil, fmt.Errorf("specified container not found: %s", *containerName)
 	}
@@ -660,7 +652,7 @@ func (s *Server) ContainerStatus(ctx context.Context, req *pb.ContainerStatusReq
 	if *containerName == "" {
 		return nil, fmt.Errorf("container ID should not be empty")
 	}
-	c := s.state.containers[*containerName]
+	c := s.state.containers.Get(*containerName)
 
 	if c == nil {
 		return nil, fmt.Errorf("specified container not found: %s", *containerName)

server/server.go

@@ -43,7 +43,7 @@ func New(runtimePath, sandboxDir, containerDir string) (*Server, error) {
 		return nil, err
 	}
 	sandboxes := make(map[string]*sandbox)
-	containers := make(map[string]*oci.Container)
+	containers := oci.NewMemoryStore()
 	netPlugin, err := ocicni.InitCNI("")
 	if err != nil {
 		return nil, err
@@ -61,34 +61,26 @@ func New(runtimePath, sandboxDir, containerDir string) (*Server, error) {
 
 type serverState struct {
 	sandboxes  map[string]*sandbox
-	containers map[string]*oci.Container
+	containers oci.Store
 }
 
 type sandbox struct {
 	name       string
 	logDir     string
 	labels     map[string]string
-	containersLock sync.Mutex
-	containers     map[string]*oci.Container
+	containers oci.Store
 }
 
 func (s *sandbox) addContainer(c *oci.Container) {
-	s.containersLock.Lock()
-	s.containers[c.Name()] = c
-	s.containersLock.Unlock()
+	s.containers.Add(c.Name(), c)
 }
 
 func (s *sandbox) getContainer(name string) *oci.Container {
-	s.containersLock.Lock()
-	c := s.containers[name]
-	s.containersLock.Unlock()
-	return c
+	return s.containers.Get(name)
 }
 
 func (s *sandbox) removeContainer(c *oci.Container) {
-	s.containersLock.Lock()
-	delete(s.containers, c.Name())
-	s.containersLock.Unlock()
+	s.containers.Delete(c.Name())
 }
 
 func (s *Server) addSandbox(sb *sandbox) {
@@ -115,13 +107,13 @@ func (s *Server) addContainer(c *oci.Container) {
 	s.stateLock.Lock()
 	sandbox := s.state.sandboxes[c.Sandbox()]
 	sandbox.addContainer(c)
-	s.state.containers[c.Name()] = c
+	s.state.containers.Add(c.Name(), c)
 	s.stateLock.Unlock()
 }
 
 func (s *Server) getContainer(name string) *oci.Container {
 	s.stateLock.Lock()
-	c := s.state.containers[name]
+	c := s.state.containers.Get(name)
 	s.stateLock.Unlock()
 	return c
 }
@@ -130,6 +122,6 @@ func (s *Server) removeContainer(c *oci.Container) {
 	s.stateLock.Lock()
 	sandbox := s.state.sandboxes[c.Sandbox()]
 	sandbox.removeContainer(c)
-	delete(s.state.containers, c.Name())
+	s.state.containers.Delete(c.Name())
 	s.stateLock.Unlock()
 }