Skip to content

Revert "move cgroup pool creation logic to init (#412)" #446

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
m0mosenpai wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Revert "move cgroup pool creation logic to init (#412)" #446

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 0 additions & 5 deletions go/common/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -492,8 +492,3 @@ func GetOlPath(ctx *cli.Context) (string, error) {
}
return filepath.Abs(olPath)
}

// CgroupPoolPath returns the cgroup pool root path for the given OL directory.
func CgroupPoolPath(olPath string) string {
return filepath.Join("/sys/fs/cgroup", filepath.Base(olPath)+"-sandboxes")
}
16 changes: 3 additions & 13 deletions go/worker/commands.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@ import (

"github.com/open-lambda/open-lambda/go/common"
"github.com/open-lambda/open-lambda/go/worker/event"
"github.com/open-lambda/open-lambda/go/worker/sandbox/cgroups"

"github.com/urfave/cli/v2"
)
Expand All @@ -40,27 +39,18 @@ func udsGet(requestPath string) (*http.Response, error) {

// initCmd corresponds to the "init" command of the admin tool.
func initCmd(ctx *cli.Context) error {
if os.Getuid() != 0 {
return fmt.Errorf("'ol worker init' must be run with sudo")
}

olPath, err := common.GetOlPath(ctx)
if err != nil {
return fmt.Errorf("init failed to get OL path: %w", err)
return err
}

if err := common.LoadDefaults(olPath); err != nil {
return fmt.Errorf("init failed to load config defaults: %w", err)
return err
}

if err := initOLDir(olPath, ctx.String("image"), ctx.Bool("newbase")); err != nil {
return fmt.Errorf("init failed to create OL directory: %w", err)
}

if err := cgroups.InitPoolRoot(common.CgroupPoolPath(olPath)); err != nil {
return fmt.Errorf("init failed to create cgroup pool: %w", err)
return err
}

fmt.Printf("\nYou may optionally modify the defaults here: %s\n\n",
filepath.Join(olPath, "config.json"))
fmt.Printf("Next start a worker using the \"ol worker up\" command.\n")
Expand Down
14 changes: 10 additions & 4 deletions go/worker/helpers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@ import (
"github.com/open-lambda/open-lambda/go/worker/embedded"
)


func initOLBaseDir(baseDir string, dockerBaseImage string) error {
if dockerBaseImage == "" {
dockerBaseImage = "ol-wasm"
Expand Down Expand Up @@ -280,8 +279,8 @@ func runningToStoppedClean() error {
// It cleans up cgroups and mounts associated with the OpenLambda instance at `olPath`.
// Returns errors encountered during cleanup operations.
func stoppedDirtyToStoppedClean(olPath string) error {
// Clean up child cgroups, preserving the pool root
cgRoot := common.CgroupPoolPath(olPath)
// Clean up cgroups associated with sandboxes
cgRoot := filepath.Join("/sys", "fs", "cgroup", filepath.Base(olPath)+"-sandboxes")
fmt.Printf("Attempting to clean up cgroups at %s\n", cgRoot)

cgroupErrorCount := 0
Expand All @@ -303,6 +302,7 @@ func stoppedDirtyToStoppedClean(olPath string) error {
}
kill := filepath.Join(cgRoot, "cgroup.kill")
if err := os.WriteFile(kill, []byte(fmt.Sprintf("%d", 1)), os.ModeAppend); err != nil {
// Print an error if killing processes in the cgroup fails.
fmt.Printf("Could not kill processes in cgroup: %s\n", err.Error())
cgroupErrorCount += 1
}
Expand All @@ -311,11 +311,17 @@ func stoppedDirtyToStoppedClean(olPath string) error {
cg := filepath.Join(cgRoot, file.Name())
fmt.Printf("Attempting to remove %s\n", cg)
if err := syscall.Rmdir(cg); err != nil {
fmt.Printf("could not remove cgroup: %s\n", err.Error())
// Print an error if removing a cgroup fails.
fmt.Printf("could not remove cgroup: %s", err.Error())
cgroupErrorCount += 1
}
}
}
if err := syscall.Rmdir(cgRoot); err != nil {
// Print an error if removing the cgroup root directory fails.
fmt.Printf("could not remove cgroup root: %s", err.Error())
cgroupErrorCount += 1
}
}

sandboxErrorCount := 0
Expand Down
6 changes: 3 additions & 3 deletions go/worker/sandbox/cgroups/cgroup.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,9 +78,9 @@ func (cg *CgroupImpl) Destroy() {
}
}

// GroupPath returns the path to this cgroup directory.
// GroupPath returns the path to the Cgroup pool for OpenLambda
func (cg *CgroupImpl) GroupPath() string {
return fmt.Sprintf("%s/%s", cg.pool.poolPath, cg.name)
return fmt.Sprintf("%s/%s", cg.pool.GroupPath(), cg.name)
}

func (cg *CgroupImpl) MemoryEvents() map[string]int64 {
Expand All @@ -107,7 +107,7 @@ func (cg *CgroupImpl) MemoryEvents() map[string]int64 {

// ResourcePath returns the path to a specific resource in this cgroup
func (cg *CgroupImpl) ResourcePath(resource string) string {
return fmt.Sprintf("%s/%s/%s", cg.pool.poolPath, cg.name, resource)
return fmt.Sprintf("%s/%s/%s", cg.pool.GroupPath(), cg.name, resource)
}

func (cg *CgroupImpl) TryWriteInt(resource string, val int64) error {
Expand Down
77 changes: 38 additions & 39 deletions go/worker/sandbox/cgroups/pool.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,13 @@ package cgroups

import (
"fmt"
"io/ioutil"
"log/slog"
"os"
"path/filepath"
"strconv"
"path"
"strings"
"syscall"
"time"

"github.com/open-lambda/open-lambda/go/common"
)
Expand All @@ -18,56 +19,35 @@ const CGROUP_RESERVE = 16

type CgroupPool struct {
Name string
poolPath string
ready chan *CgroupImpl
recycled chan *CgroupImpl
quit chan chan bool
nextID int
}

// InitPoolRoot creates the cgroup pool root directory and enables controllers.
func InitPoolRoot(poolPath string) error {

if err := os.MkdirAll(poolPath, 0700); err != nil {
return fmt.Errorf("failed to create cgroup pool root %s: %w", poolPath, err)
}

ctrlPath := filepath.Join(poolPath, "cgroup.subtree_control")
if err := os.WriteFile(ctrlPath, []byte("+pids +io +memory +cpu"), os.ModeAppend); err != nil {
return fmt.Errorf("failed to enable controllers at %s: %w", ctrlPath, err)
}

uidStr := os.Getenv("SUDO_UID")
if uidStr == "" {
return fmt.Errorf("SUDO_UID not set; worker must be run with sudo")
}
uid, err := strconv.Atoi(uidStr)
if err != nil {
return fmt.Errorf("invalid SUDO_UID value %q: %w", uidStr, err)
}
if err := os.Chown(poolPath, uid, uid); err != nil {
return fmt.Errorf("failed to chown cgroup pool root: %w", err)
}

fmt.Printf("\tCreated cgroup pool root at %s\n", poolPath)
return nil
}

func NewCgroupPool(name string, poolPath string) (*CgroupPool, error) {
// NewCgroupPool creates a new CgroupPool with the specified name.
func NewCgroupPool(name string) (*CgroupPool, error) {
pool := &CgroupPool{
Name: name,
poolPath: poolPath,
Name: path.Base(path.Dir(common.Conf.Worker_dir)) + "-" + name,
ready: make(chan *CgroupImpl, CGROUP_RESERVE),
recycled: make(chan *CgroupImpl, CGROUP_RESERVE),
quit: make(chan chan bool),
nextID: 0,
}

if st, err := os.Stat(poolPath); err != nil || !st.IsDir() {
return nil, fmt.Errorf("cgroup pool root %s does not exist.", poolPath)
// create cgroup
groupPath := pool.GroupPath()
pool.printf("create %s", groupPath)
if err := syscall.Mkdir(groupPath, 0700); err != nil {
return nil, fmt.Errorf("Mkdir %s: %s", groupPath, err)
}

// Make controllers available to child groups
rpath := fmt.Sprintf("%s/cgroup.subtree_control", groupPath)
if err := ioutil.WriteFile(rpath, []byte("+pids +io +memory +cpu"), os.ModeAppend); err != nil {
panic(fmt.Sprintf("Error writing to %s: %v", rpath, err))
}

pool.printf("reusing pool root %s", poolPath)
go pool.cgTask()
return pool, nil
}
Expand Down Expand Up @@ -156,14 +136,28 @@ Empty:
done <- true
}

// Destroy drains all child cgroups but preserves the pool root.
// Destroy this entire cgroup pool
func (pool *CgroupPool) Destroy() {
// signal cgTask, then wait for it to finish
ch := make(chan bool)
pool.quit <- ch
<-ch

pool.printf("destroyed all child cgroups, pool root preserved")
// Destroy cgroup for this entire pool
gpath := pool.GroupPath()
pool.printf("Destroying cgroup pool with path \"%s\"", gpath)
for i := 100; i >= 0; i-- {
if err := syscall.Rmdir(gpath); err != nil {
if i == 0 {
panic(fmt.Errorf("Rmdir %s: %s", gpath, err))
}

pool.printf("cgroup pool Rmdir failed, trying again in 5ms")
time.Sleep(5 * time.Millisecond)
} else {
break
}
}
}

// GetCg retrieves a cgroup from the pool, setting its memory limit and CPU percentage.
Expand All @@ -184,3 +178,8 @@ func (pool *CgroupPool) GetCg(memLimitMB int, moveMemCharge bool, cpuPercent int

return cg
}

// GroupPath returns the path to the Cgroup pool for OpenLambda
func (pool *CgroupPool) GroupPath() string {
return fmt.Sprintf("/sys/fs/cgroup/%s", pool.Name)
}
3 changes: 1 addition & 2 deletions go/worker/sandbox/sockPool.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,7 @@ type SOCKPool struct {

// NewSOCKPool creates a SOCKPool.
func NewSOCKPool(name string, mem *MemPool) (cf *SOCKPool, err error) {
olPath := filepath.Dir(common.Conf.Worker_dir)
cgPool, err := cgroups.NewCgroupPool(name, common.CgroupPoolPath(olPath))
cgPool, err := cgroups.NewCgroupPool(name)
if err != nil {
return nil, err
}
Expand Down
Loading