<commit_before>package calcium

import (
	"context"
	"fmt"
	"sync"

	enginetypes "github.com/docker/docker/api/types"
	"github.com/projecteru2/core/types"
	log "github.com/sirupsen/logrus"
)

// ReplaceContainer replace containers with same resource
func (c *Calcium) ReplaceContainer(ctx context.Context, opts *types.DeployOptions) (chan *types.ReplaceContainerMessage, error) {
	oldContainers, err := c.ListContainers(ctx, opts.Name, opts.Entrypoint.Name, opts.Nodename)
	if err != nil {
		return nil, err
	}
	ch := make(chan *types.ReplaceContainerMessage)

	go func() {
		defer close(ch)
		lock, err := c.Lock(ctx, opts.Podname, c.config.LockTimeout)
		if err != nil {
			log.Errorf("[ReplaceContainer] Lock pod failed %v", err)
			return
		}
		defer lock.Unlock(ctx)

		// 并发控制
		step := opts.Count
		wg := sync.WaitGroup{}

		for index, oldContainer := range oldContainers {
			log.Debug("[ReplaceContainer] Replace old container : %s", oldContainer.ID)
			wg.Add(1)
			go func(deployOpts types.DeployOptions, oldContainer *types.Container, index int) {
				defer wg.Done()
				// 使用复制之后的配置
				// 停老的，起新的
				deployOpts.Memory = oldContainer.Memory
				deployOpts.CPUQuota = oldContainer.Quota

				createMessage, err := c.replaceAndRemove(ctx, &deployOpts, oldContainer, index)
				ch <- &types.ReplaceContainerMessage{
					CreateContainerMessage: createMessage,
					OldContainerID:         oldContainer.ID,
					Error:                  err,
				}
				if err != nil {
					log.Errorf("[ReplaceContainer] Replace and remove failed %v, old container restarted", err)
				}
				// 传 opts 的值，产生一次复制
			}(*opts, oldContainer, index)
			if index != 0 && step%index == 0 {
				wg.Wait()
			}
		}
		wg.Wait()
	}()

	return ch, nil
}

func (c *Calcium) replaceAndRemove(
	ctx context.Context,
	opts *types.DeployOptions,
	oldContainer *types.Container,
	index int) (*types.CreateContainerMessage, error) {
	var err error

	// 锁住，防止删除
	lock, err := c.Lock(ctx, fmt.Sprintf("rmcontainer_%s", oldContainer.ID), int(c.config.GlobalTimeout.Seconds()))
	if err != nil {
		return nil, err
	}
	defer lock.Unlock(ctx)

	// 确保得到锁的时候容器没被干掉
	_, err = oldContainer.Inspect(ctx)
	if err != nil {
		return nil, err
	}

	// 预先扣除资源，若成功，老资源会回收，若失败，新资源也会被回收
	err = c.store.UpdateNodeResource(ctx, oldContainer.Podname, oldContainer.Nodename, oldContainer.CPU, oldContainer.Memory, "-")<commit_msg>missing softlimit, add output<commit_after>package calcium

import (
	"context"
	"fmt"
	"sync"

	enginetypes "github.com/docker/docker/api/types"
	"github.com/projecteru2/core/types"
	log "github.com/sirupsen/logrus"
)

// ReplaceContainer replace containers with same resource
func (c *Calcium) ReplaceContainer(ctx context.Context, opts *types.DeployOptions) (chan *types.ReplaceContainerMessage, error) {
	oldContainers, err := c.ListContainers(ctx, opts.Name, opts.Entrypoint.Name, opts.Nodename)
	if err != nil {
		return nil, err
	}
	ch := make(chan *types.ReplaceContainerMessage)

	go func() {
		defer close(ch)
		lock, err := c.Lock(ctx, opts.Podname, c.config.LockTimeout)
		if err != nil {
			log.Errorf("[ReplaceContainer] Lock pod failed %v", err)
			return
		}
		defer lock.Unlock(ctx)

		// 并发控制
		step := opts.Count
		wg := sync.WaitGroup{}

		for index, oldContainer := range oldContainers {
			log.Debugf("[ReplaceContainer] Replace old container %s", oldContainer.ID)
			wg.Add(1)
			go func(deployOpts types.DeployOptions, oldContainer *types.Container, index int) {
				defer wg.Done()
				// 使用复制之后的配置
				// 停老的，起新的
				deployOpts.Memory = oldContainer.Memory
				deployOpts.CPUQuota = oldContainer.Quota
				deployOpts.SoftLimit = oldContainer.SoftLimit

				createMessage, err := c.replaceAndRemove(ctx, &deployOpts, oldContainer, index)
				ch <- &types.ReplaceContainerMessage{
					CreateContainerMessage: createMessage,
					OldContainerID:         oldContainer.ID,
					Error:                  err,
				}
				if err != nil {
					log.Errorf("[ReplaceContainer] Replace and remove failed %v, old container restarted", err)
					return
				}
				log.Infof("[ReplaceContainer] Replace and remove success %s", oldContainer.ID)
				// 传 opts 的值，产生一次复制
			}(*opts, oldContainer, index)
			if (index+1)%step == 0 {
				wg.Wait()
			}
		}
		wg.Wait()
	}()

	return ch, nil
}

func (c *Calcium) replaceAndRemove(
	ctx context.Context,
	opts *types.DeployOptions,
	oldContainer *types.Container,
	index int) (*types.CreateContainerMessage, error) {
	var err error

	// 锁住，防止删除
	lock, err := c.Lock(ctx, fmt.Sprintf("rmcontainer_%s", oldContainer.ID), int(c.config.GlobalTimeout.Seconds()))
	if err != nil {
		return nil, err
	}
	defer lock.Unlock(ctx)

	// 确保得到锁的时候容器没被干掉
	_, err = oldContainer.Inspect(ctx)
	if err != nil {
		return nil, err
	}

	// 预先扣除资源，若成功，老资源会回收，若失败，新资源也会被回收
	err = c.store.UpdateNodeResource(ctx, oldContainer.Podname, oldContainer.Nodename, oldContainer.CPU, oldContainer.Memory, "-")
{'fraction_non_alphanumeric': 0.12754686226568868, 'fraction_numerical': 0.0016299918500407497, 'mean_word_length': 3.7156580211335255, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5378304', 'n_tokens_mistral': 1953, 'n_tokens_neox': 1793, 'n_words': 383}